From a182fcbce8b7baa987c4819f3b76986517281cf5 Mon Sep 17 00:00:00 2001 From: Teddy Tennant Date: Sat, 8 Aug 2026 16:05:20 -0400 Subject: [PATCH] operations: speed up checksum by hashing files in parallel - fixes #9727 rclone checksum and rclone check --checkfile hashed one file at a time unless --download was in use. The non-download branch of CheckSum called Object.Hash synchronously from inside the listing callback, so --checkers had no effect there; only the --download branch used the checker token pool. Both branches now do their work in the same goroutine bounded by --checkers, so --checkers means the same thing with and without --download. Checking a sum file against 2.6 GiB of local files with --checkers 8 went from 1.70s to 0.31s here. As a result the "hashing" checking transfer now covers the hashing work in both branches, rather than being finished as soon as the download goroutine had been started. --- fs/operations/check.go | 23 +++---- fs/operations/check_test.go | 118 ++++++++++++++++++++++++++++++++++++ 2 files changed, 128 insertions(+), 13 deletions(-) diff --git a/fs/operations/check.go b/fs/operations/check.go index d4d1eb3fd..683c2626f 100644 --- a/fs/operations/check.go +++ b/fs/operations/check.go @@ -481,24 +481,16 @@ func (c *checkMarch) checkSum(ctx context.Context, obj fs.Object, download bool, return } - var err error tr := accounting.Stats(ctx).NewCheckingTransfer(obj, "hashing") - defer tr.Done(ctx, err) if !sumFound { - err = errors.New("sum not found") + err := errors.New("sum not found") _ = fs.CountError(ctx, err) fs.Errorf(obj, "%v", err) c.differences.Add(1) c.srcFilesMissing.Add(1) c.report(obj, c.opt.MissingOnSrc, '-') - return - } - - if !download { - var objHash string - objHash, err = obj.Hash(ctx, hashType) - c.matchSum(ctx, sumHash, objHash, obj, err, hashType) + tr.Done(ctx, nil) return } @@ -512,16 +504,21 @@ func (c *checkMarch) checkSum(ctx context.Context, obj fs.Object, download bool, ) defer func() { c.matchSum(ctx, sumHash, objHash, obj, err, hashType) + tr.Done(ctx, nil) <-c.tokens // get the token back to free up a slot c.wg.Done() }() + if !download { + objHash, err = obj.Hash(ctx, hashType) + return + } if in, err = Open(ctx, obj); err != nil { return } - tr := accounting.Stats(ctx).NewTransfer(obj, nil) - in = tr.Account(ctx, in).WithBuffer() // account and buffer the transfer + dlTr := accounting.Stats(ctx).NewTransfer(obj, nil) + in = dlTr.Account(ctx, in).WithBuffer() // account and buffer the transfer defer func() { - tr.Done(ctx, nil) // will close the stream + dlTr.Done(ctx, nil) // will close the stream }() hashVals, err2 := hash.StreamTypes(in, hash.NewHashSet(hashType)) if err2 != nil { diff --git a/fs/operations/check_test.go b/fs/operations/check_test.go index e6c309277..f843a8cd3 100644 --- a/fs/operations/check_test.go +++ b/fs/operations/check_test.go @@ -8,7 +8,9 @@ import ( "io" "sort" "strings" + "sync" "testing" + "time" "github.com/rclone/rclone/cmd/bisync/bilib" "github.com/rclone/rclone/fs" @@ -16,6 +18,8 @@ import ( "github.com/rclone/rclone/fs/hash" "github.com/rclone/rclone/fs/operations" "github.com/rclone/rclone/fstest" + "github.com/rclone/rclone/fstest/mockfs" + "github.com/rclone/rclone/fstest/mockobject" "github.com/rclone/rclone/lib/readers" "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" @@ -541,6 +545,120 @@ func TestCheckSumDownload(t *testing.T) { testCheckSum(t, true) } +// hashProbe records how many hashing operations are running at once. Each +// operation blocks until want of them are in flight, or until the deadline +// expires so a serial implementation finishes instead of deadlocking. +type hashProbe struct { + mu sync.Mutex + inFlight int + max int + want int + once sync.Once + reached chan struct{} + timedOut chan struct{} +} + +func newHashProbe(t *testing.T, want int, deadline time.Duration) *hashProbe { + p := &hashProbe{ + want: want, + reached: make(chan struct{}), + timedOut: make(chan struct{}), + } + timer := time.AfterFunc(deadline, func() { close(p.timedOut) }) + t.Cleanup(func() { timer.Stop() }) + return p +} + +func (p *hashProbe) enter() { + p.mu.Lock() + p.inFlight++ + p.max = max(p.max, p.inFlight) + full := p.inFlight >= p.want + p.mu.Unlock() + if full { + p.once.Do(func() { close(p.reached) }) + } + select { + case <-p.reached: + case <-p.timedOut: + } +} + +func (p *hashProbe) leave() { + p.mu.Lock() + p.inFlight-- + p.mu.Unlock() +} + +func (p *hashProbe) maxInFlight() int { + p.mu.Lock() + defer p.mu.Unlock() + return p.max +} + +// probeObject is an object whose hashing is instrumented by a hashProbe +type probeObject struct { + *mockobject.ContentMockObject + probe *hashProbe +} + +func (o *probeObject) Hash(ctx context.Context, ht hash.Type) (string, error) { + o.probe.enter() + defer o.probe.leave() + return o.ContentMockObject.Hash(ctx, ht) +} + +func (o *probeObject) Open(ctx context.Context, options ...fs.OpenOption) (io.ReadCloser, error) { + o.probe.enter() + defer o.probe.leave() + return o.ContentMockObject.Open(ctx, options...) +} + +func testCheckSumConcurrency(t *testing.T, download bool) { + const ( + checkers = 2 + numFiles = 4 + sumFile = "test.sum" + ) + hashType := hash.MD5 + + ctx, ci := fs.AddConfig(context.Background()) + ci.Checkers = checkers + probe := newHashProbe(t, checkers, 10*time.Second) + + f, err := mockfs.NewFs(ctx, "checkSum", "checkSum", nil) + require.NoError(t, err) + fsrc := f.(*mockfs.Fs) + fsrc.SetHashes(hash.NewHashSet(hashType)) + + sums := &bytes.Buffer{} + for i := range numFiles { + remote := fmt.Sprintf("file%d", i) + o := mockobject.New(remote).WithContent([]byte("contents of "+remote), mockobject.SeekModeNone) + sum, err := o.Hash(ctx, hashType) + require.NoError(t, err) + _, _ = fmt.Fprintf(sums, "%s %s\n", sum, remote) + fsrc.AddObject(&probeObject{ContentMockObject: o, probe: probe}) + } + + f, err = mockfs.NewFs(ctx, "sums", "sums", nil) + require.NoError(t, err) + fsum := f.(*mockfs.Fs) + fsum.AddObject(mockobject.New(sumFile).WithContent(sums.Bytes(), mockobject.SeekModeNone)) + + opt := operations.CheckOpt{Combined: new(bytes.Buffer)} + require.NoError(t, operations.CheckSum(ctx, fsrc, fsum, sumFile, hashType, &opt, download)) + assert.Equal(t, checkers, probe.maxInFlight(), "hashing was not run --checkers at a time") +} + +func TestCheckSumConcurrency(t *testing.T) { + testCheckSumConcurrency(t, false) +} + +func TestCheckSumDownloadConcurrency(t *testing.T) { + testCheckSumConcurrency(t, true) +} + func TestApplyTransforms(t *testing.T) { var ( hashType = hash.MD5