diff --git a/backend/jottacloud/jottacloud.go b/backend/jottacloud/jottacloud.go index aad41a422..6af70b99b 100644 --- a/backend/jottacloud/jottacloud.go +++ b/backend/jottacloud/jottacloud.go @@ -34,6 +34,7 @@ import ( "github.com/rclone/rclone/fs/hash" "github.com/rclone/rclone/fs/list" "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/rest" @@ -1924,8 +1925,9 @@ func readMD5(in io.Reader, size, threshold int64) (md5sum string, out io.Reader, // nothing to clean up by default cleanup = func() {} - // don't cache small files on disk to reduce wear of the disk - if size > threshold { + // don't cache small files on disk to reduce wear of the disk, but + // spool unknown sized streams there as they can't be bounded in memory + if size > threshold || size < 0 { var tempFile *os.File // create the cache file @@ -1955,15 +1957,15 @@ func readMD5(in io.Reader, size, threshold int64) (md5sum string, out io.Reader, // replace the already read source with a reader of our cached file out = tempFile } else { - // that's a small file, just read it into memory - var inData []byte - inData, err = io.ReadAll(teeReader) - if err != nil { + // that's a small file, just read it into memory from the global pool + rw := multipart.NewRW() + cleanup = func() { + _ = rw.Close() + } + if _, err = io.Copy(rw, teeReader); err != nil { return } - - // set the reader to our read memory block - out = bytes.NewReader(inData) + out = rw } return hex.EncodeToString(md5Hasher.Sum(nil)), out, cleanup, nil } diff --git a/backend/jottacloud/jottacloud_internal_test.go b/backend/jottacloud/jottacloud_internal_test.go index db9d6a157..1889400c1 100644 --- a/backend/jottacloud/jottacloud_internal_test.go +++ b/backend/jottacloud/jottacloud_internal_test.go @@ -5,12 +5,14 @@ import ( "crypto/md5" "fmt" "io" + "os" "testing" "time" "github.com/rclone/rclone/fs" "github.com/rclone/rclone/fstest" "github.com/rclone/rclone/fstest/fstests" + "github.com/rclone/rclone/lib/pool" "github.com/rclone/rclone/lib/random" "github.com/rclone/rclone/lib/readers" "github.com/stretchr/testify/assert" @@ -28,11 +30,16 @@ func TestReadMD5(t *testing.T) { wantMD5 := fmt.Sprintf("%x", hasher.Sum(nil)) for _, threshold := range []int64{512, 1024, 10 * 1024, 20 * 1024} { t.Run(fmt.Sprintf("%d", threshold), func(t *testing.T) { + inUse := pool.Global().InUse() in := readers.NewPatternReader(size) gotMD5, out, cleanup, err := readMD5(in, size, threshold) - defer cleanup() require.NoError(t, err) assert.Equal(t, wantMD5, gotMD5) + if size > threshold { + assert.IsType(t, (*os.File)(nil), out, "big files should be spooled to disk") + } else { + assert.IsType(t, (*pool.RW)(nil), out, "small files should be buffered in the pool") + } // check md5hash of out hasher := md5.New() @@ -41,10 +48,36 @@ func TestReadMD5(t *testing.T) { assert.Equal(t, n, size) outMD5 := fmt.Sprintf("%x", hasher.Sum(nil)) assert.Equal(t, wantMD5, outMD5) + + // check cleanup returns the buffer to the pool + cleanup() + assert.Equal(t, inUse, pool.Global().InUse(), "pool buffers leaked") }) } }) } + + // Unknown sized streams can't be bounded in memory so must be spooled to disk + t.Run("UnknownSize", func(t *testing.T) { + const size = 1024 + inUse := pool.Global().InUse() + hasher := md5.New() + _, err := io.Copy(hasher, readers.NewPatternReader(size)) + require.NoError(t, err) + wantMD5 := fmt.Sprintf("%x", hasher.Sum(nil)) + + gotMD5, out, cleanup, err := readMD5(readers.NewPatternReader(size), -1, 10*size) + require.NoError(t, err) + assert.Equal(t, wantMD5, gotMD5) + assert.IsType(t, (*os.File)(nil), out) + hasher.Reset() + n, err := io.Copy(hasher, out) + require.NoError(t, err) + assert.Equal(t, int64(size), n) + assert.Equal(t, wantMD5, fmt.Sprintf("%x", hasher.Sum(nil))) + cleanup() + assert.Equal(t, inUse, pool.Global().InUse()) + }) } func (f *Fs) InternalTestMetadata(t *testing.T) {