From 4722b94d1acb98b3bd4ee9c5af68e6d2f7ccc783 Mon Sep 17 00:00:00 2001 From: Splainte Date: Fri, 28 Aug 2026 22:21:16 +0200 Subject: [PATCH] sftp: implement multi-thread uploads - fixes #8185 The sftp backend could be used as the source of a multi-thread copy - each chunk of a download is read on its own connection - but not as the destination, because it did not implement OpenWriterAt. Uploading a single large file was therefore limited to one connection while downloading the same file was not, capping upload throughput well below the link speed on high latency connections. This implements OpenWriterAt for the sftp backend using lib/filepool. A small pool of open write handles, each backed by its own connection, lets the core write the chunks of a large file concurrently over several connections. The file is created and truncated once up front so every chunk offset is valid before the concurrent writes start. Multi-thread uploads are off by default and turned on with the new --sftp-multithread-upload flag, since many sftp servers only accept sequential writes and would fail large uploads otherwise. Even when enabled they fall back to a single connection when --sftp-disable-concurrent-writes is set (a server that can't take out-of-order packets on one handle won't take several handles either) or when --sftp-connections caps the pool (the per-file fan-out would otherwise deadlock waiting on it). The OpenSSH test server (TestSFTPOpenssh) enables --sftp-multithread-upload so the feature is exercised in CI against a real server. Tested against OpenSSH: a single large upload over a high latency link went from ~12 MB/s to ~90 MB/s. Co-Authored-By: Claude Opus 4.8 Signed-off-by: Splainte --- backend/sftp/sftp.go | 42 +++++++ backend/sftp/writerat.go | 137 +++++++++++++++++++++++ fstest/testserver/init.d/TestSFTPOpenssh | 3 + 3 files changed, 182 insertions(+) create mode 100644 backend/sftp/writerat.go 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} }