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.
This commit is contained in:
Teddy Tennant
committed by
Nick Craig-Wood
Nick Craig-Wood
parent
fc348fcb0b
commit
a182fcbce8
+10
-13
@@ -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 {
|
||||
|
||||
@@ -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
|
||||
|
||||
Reference in New Issue
Block a user