diff --git a/backend/sftp/sftp.go b/backend/sftp/sftp.go index 6d639ee6f..03c40eaaa 100644 --- a/backend/sftp/sftp.go +++ b/backend/sftp/sftp.go @@ -390,6 +390,28 @@ Normally rclone uses concurrent writes to upload files. This improves the performance greatly, especially for distant servers. This option disables concurrent writes should that be necessary. +`, + Advanced: true, + }, { + Name: "multithread_upload", + Default: false, + Help: `Set this to use multi-thread (multi-connection) uploads. + +Normally rclone uploads a single large file over one SFTP connection. +With this set rclone uploads large files over several SFTP connections +at once (one per chunk), which greatly improves throughput on high +latency links, the same way multi-thread downloads already work. Each +connection writes a different, non-overlapping byte range of the file, +so this requires a server that allows several handles to the same file +with writes at arbitrary offsets, which OpenSSH does. + +This is off by default because many SFTP servers only accept sequential +writes (for example some object storage backed SFTP gateways) and would +fail large uploads with "truncate failed" or "invalid offset" style +errors. + +This is ignored (uploads stay single-threaded) when +--sftp-disable-concurrent-writes or --sftp-connections is in use. `, Advanced: true, }, { @@ -648,6 +670,7 @@ type Options struct { UseFstat bool `config:"use_fstat"` DisableConcurrentReads bool `config:"disable_concurrent_reads"` DisableConcurrentWrites bool `config:"disable_concurrent_writes"` + MultithreadUpload bool `config:"multithread_upload"` IdleTimeout fs.Duration `config:"idle_timeout"` ChunkSize fs.SizeSuffix `config:"chunk_size"` Concurrency int `config:"concurrency"` @@ -1594,6 +1617,24 @@ func NewFsWithConnection(ctx context.Context, f *Fs, name string, root string, m // Disable server side copy unless --sftp-copy-is-hardlink is set f.features.Copy = nil } + // Multi-thread uploads open one write handle per connection. They are off + // by default (many SFTP servers only take sequential writes) and turned on + // with --sftp-multithread-upload. Even when asked for, fall back to + // single-connection uploads when concurrent writes are disabled (a server + // that can't take out-of-order packets on one handle won't take several + // handles writing at arbitrary offsets either) or when connections are + // capped (the per-file fan-out would deadlock waiting on the pool, and a + // user limiting connections doesn't want it anyway). + switch { + case !opt.MultithreadUpload: + f.features.OpenWriterAt = nil + case opt.DisableConcurrentWrites: + fs.Logf(f, "Disabling multi-thread uploads because --sftp-disable-concurrent-writes is set") + f.features.OpenWriterAt = nil + case opt.Connections > 0: + fs.Logf(f, "Disabling multi-thread uploads because --sftp-connections is set") + f.features.OpenWriterAt = nil + } // Make a connection and pool it to return errors early c, err := f.getSftpConnection(ctx) if err != nil { @@ -2845,6 +2886,7 @@ var ( _ fs.PutStreamer = &Fs{} _ fs.Mover = &Fs{} _ fs.Copier = &Fs{} + _ fs.OpenWriterAter = &Fs{} _ fs.DirMover = &Fs{} _ fs.DirSetModTimer = &Fs{} _ fs.Abouter = &Fs{} diff --git a/backend/sftp/writerat.go b/backend/sftp/writerat.go new file mode 100644 index 000000000..36d923fc8 --- /dev/null +++ b/backend/sftp/writerat.go @@ -0,0 +1,137 @@ +//go:build !plan9 + +package sftp + +import ( + "context" + "errors" + "fmt" + "os" + "sync" + + "github.com/pkg/sftp" + "github.com/rclone/rclone/fs" + "github.com/rclone/rclone/lib/filepool" +) + +// poolFile is a pooled write handle together with the connection it lives on. +type poolFile struct { + file *sftp.File + c *conn +} + +// openPoolFile opens a fresh write handle on its own connection for path. The +// file must already exist: the handle opens O_WRONLY only, so it never races +// another handle to create or truncate. +func (f *Fs) openPoolFile(path string) func(context.Context) (*poolFile, error) { + return func(ctx context.Context) (*poolFile, error) { + c, err := f.getSftpConnection(ctx) + if err != nil { + return nil, err + } + file, err := c.sftpClient.OpenFile(path, os.O_WRONLY) + if err != nil { + f.putSftpConnection(&c, err) + return nil, err + } + return &poolFile{file: file, c: c}, nil + } +} + +// releasePoolFile closes a pooled handle and returns its connection. +func (f *Fs) releasePoolFile(pf *poolFile, err error) error { + closeErr := pf.file.Close() + if err == nil { + err = closeErr + } + f.putSftpConnection(&pf.c, err) + return closeErr +} + +// sftpWriterAt is the fs.WriterAtCloser used by the core's multi-thread copy. +// WriteAt is called concurrently at non-overlapping offsets, each borrowing its +// own handle from the pool. +type sftpWriterAt struct { + fs *Fs + pool *filepool.Pool[*poolFile] + closeMu sync.Mutex + closed bool + wg sync.WaitGroup +} + +// WriteAt writes p at offset off using a handle borrowed from the pool. +func (w *sftpWriterAt) WriteAt(p []byte, off int64) (int, error) { + w.closeMu.Lock() + if w.closed { + w.closeMu.Unlock() + return 0, errors.New("sftp: WriteAt on closed writer") + } + w.wg.Add(1) + w.closeMu.Unlock() + defer w.wg.Done() + + pf, err := w.pool.Get() + if err != nil { + return 0, err + } + n, writeErr := pf.file.WriteAt(p, off) + w.pool.Put(pf, writeErr) + if writeErr != nil { + return n, fmt.Errorf("failed to write at offset %d: %w", off, writeErr) + } + return n, nil +} + +// Close waits for outstanding writes then closes every pooled handle. +func (w *sftpWriterAt) Close() error { + w.closeMu.Lock() + if w.closed { + w.closeMu.Unlock() + return nil + } + w.closed = true + w.closeMu.Unlock() + + w.wg.Wait() + err := w.pool.Drain() + w.fs.removeSession() + return err +} + +// OpenWriterAt opens remote for random-access writes, truncating any existing +// object, and pre-sizes it to size (if known) so every chunk offset is valid. +// +// The file is created and truncated once here, on a single connection, so the +// pooled handles open O_WRONLY and never race to truncate each other's data. +func (f *Fs) OpenWriterAt(ctx context.Context, remote string, size int64) (fs.WriterAtCloser, error) { + err := f.mkParentDir(ctx, remote) + if err != nil { + return nil, fmt.Errorf("OpenWriterAt: %w", err) + } + path := f.remotePath(remote) + + c, err := f.getSftpConnection(ctx) + if err != nil { + return nil, fmt.Errorf("OpenWriterAt: %w", err) + } + file, err := c.sftpClient.OpenFile(path, os.O_WRONLY|os.O_CREATE|os.O_TRUNC) + if err != nil { + f.putSftpConnection(&c, err) + return nil, fmt.Errorf("OpenWriterAt: create failed: %w", err) + } + if size > 0 { + if truncErr := file.Truncate(size); truncErr != nil { + _ = file.Close() + f.putSftpConnection(&c, truncErr) + return nil, fmt.Errorf("OpenWriterAt: truncate failed (the server may not support multi-thread uploads; try without --sftp-multithread-upload): %w", truncErr) + } + } + if closeErr := file.Close(); closeErr != nil { + f.putSftpConnection(&c, closeErr) + return nil, fmt.Errorf("OpenWriterAt: close failed: %w", closeErr) + } + f.putSftpConnection(&c, nil) + + f.addSession() + return &sftpWriterAt{fs: f, pool: filepool.New(ctx, f.openPoolFile(path), f.releasePoolFile)}, nil +} diff --git a/fstest/testserver/init.d/TestSFTPOpenssh b/fstest/testserver/init.d/TestSFTPOpenssh index 91a9c9a1f..b856ee855 100755 --- a/fstest/testserver/init.d/TestSFTPOpenssh +++ b/fstest/testserver/init.d/TestSFTPOpenssh @@ -20,6 +20,9 @@ start() { echo user=$USER echo pass=$(rclone obscure $PASS) echo copy_is_hardlink=true + # OpenSSH supports concurrent writes at arbitrary offsets, so exercise + # multi-thread uploads here (off by default) against a real server. + echo multithread_upload=true echo _connect=127.0.0.1:${PORT} }