f-denkena/rclone
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

dropbox: reuse upload chunk buffers via the global memory pool

Browse Source 
Each chunked upload allocated a fresh chunk-sized buffer (48 MiB by
default), so bulk transfers of many files churned allocations and GC.

Buffer chunks with multipart.NewRW instead so chunk memory is reused
across uploads and is part of rclone's central memory management. The
pool.RW is seekable, so the existing IncorrectOffset recovery which
skips already-received bytes on retry carries over unchanged, and the
"chunk received OK" check now compares against the bytes actually
buffered so a short final chunk is recognised too.

The Dropbox SDK wraps the request body in io.NopCloser, so the transport
never closes the pool buffer. The upload loop closes it after every
chunk, on error paths included.

The FsPutRetry integration test covers the retry of a failed upload
request and checks the buffers are returned to the pool.
This commit is contained in:
Nick Craig-Wood
2026-09-01 14:21:52 +01:00
parent cdabcc7cc4
commit 738eadf8df
1 changed files with 17 additions and 10 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/dropbox/dropbox.go
+17 -10
View File
@@ -51,6 +51,7 @@ import (
"github.com/rclone/rclone/fs/operations"
"github.com/rclone/rclone/lib/batcher"
"github.com/rclone/rclone/lib/encoder"
"github.com/rclone/rclone/lib/multipart"
"github.com/rclone/rclone/lib/oauthutil"
"github.com/rclone/rclone/lib/pacer"
"github.com/rclone/rclone/lib/readers"
@@ -2079,11 +2080,6 @@ func (o *Object) uploadChunked(ctx context.Context, in0 io.Reader, commitInfo *f
// write chunks
in := readers.NewCountingReader(in0)
bufSize := chunkSize
if size >= 0 && size < bufSize {
bufSize = size
}
buf := make([]byte, int(bufSize))
cursor := files.UploadSessionCursor{
SessionId: res.SessionId,
Offset: 0,
@@ -2103,14 +2099,21 @@ func (o *Object) uploadChunked(ctx context.Context, in0 io.Reader, commitInfo *f
fs.Debugf(o, "Uploading chunk %d/%d", currentChunk, chunks)
}
chunk := readers.NewRepeatableLimitReaderBuffer(in, buf, chunkSize)
// Buffer the chunk in memory from the global pool for retries
rw := multipart.NewRW()
var n int64
n, err = io.CopyN(rw, in, chunkSize)
if err != nil && err != io.EOF {
_ = rw.Close()
return nil, err
}
skip := int64(0)
err = o.fs.pacer.Call(func() (bool, error) {
// seek to the start in case this is a retry
if _, err = chunk.Seek(skip, io.SeekStart); err != nil {
if _, err = rw.Seek(skip, io.SeekStart); err != nil {
return false, err
}
err = o.fs.srv.UploadSessionAppendV2Context(ctx, &appendArg, chunk)
err = o.fs.srv.UploadSessionAppendV2Context(ctx, &appendArg, rw)
// after session is started, we retry everything
if err != nil {
// Check for incorrect offset error and retry with new offset
@@ -2122,10 +2125,10 @@ func (o *Object) uploadChunked(ctx context.Context, in0 io.Reader, commitInfo *f
what := fmt.Sprintf("incorrect offset error received: sent %d, need %d, skip %d", cursor.Offset, correctOffset, skip)
if skip < 0 {
return false, fmt.Errorf("can't seek backwards to correct offset: %s", what)
} else if skip == chunkSize {
} else if skip == n {
fs.Debugf(o, "%s: chunk received OK - continuing", what)
return false, nil
} else if skip > chunkSize {
} else if skip > n {
// This error should never happen
return false, fmt.Errorf("can't seek forwards by more than a chunk to correct offset: %s", what)
}
@@ -2142,9 +2145,13 @@ func (o *Object) uploadChunked(ctx context.Context, in0 io.Reader, commitInfo *f
}
return err != nil, err
})
closeErr := rw.Close()
if err != nil {
return nil, err
}
if closeErr != nil {
return nil, closeErr
}
if size >= 0 {
// Check for sources which truncate early
expected := min(uint64(currentChunk)*uint64(chunkSize), uint64(size))