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 <noreply@anthropic.com>
Signed-off-by: Splainte <r.wycke@hotmail.fr>

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{}

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
+}

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}
 }