chore: remove weak hashing which does not pull its weight (#10005)
We've had weak/rolling hashing in the code for quite a while. It was a popular request for a while, based on the belief that rsync does this and we should too. However, the benefit is quite small; we save on average about 0.8% of transferred blocks over the population as a whole: <img width="974" alt="Screenshot 2025-03-28 at 17 09 02" src="https://github.com/user-attachments/assets/bbe10dea-f85e-4043-9823-7cef1220b4a2" /> This would be fine if the cost was comparably low, however the downside of attempting rolling hash matching is that we (by default) do a complete file read on the destination in order to look for matches before we starting pulling blocks for the file. For any larger file this means a sometimes long, I/O-intensive pause before the file starts syncing, for usually no benefit. I propose we simply rip off the bandaid and save the effort.
This commit is contained in:
Jakob Borg
@@ -75,7 +75,7 @@ func (f *fakeConnection) DownloadProgress(_ context.Context, dp *protocol.Downlo
|
||||
|
||||
func (f *fakeConnection) addFileLocked(name string, flags uint32, ftype protocol.FileInfoType, data []byte, version protocol.Vector, localFlags uint32) {
|
||||
blockSize := protocol.BlockSize(int64(len(data)))
|
||||
blocks, _ := scanner.Blocks(context.TODO(), bytes.NewReader(data), blockSize, int64(len(data)), nil, true)
|
||||
blocks, _ := scanner.Blocks(context.TODO(), bytes.NewReader(data), blockSize, int64(len(data)), nil)
|
||||
|
||||
file := protocol.FileInfo{
|
||||
Name: name,
|
||||
|
||||
@@ -560,7 +560,7 @@ func setupKnownFiles(t *testing.T, ffs fs.Filesystem, data []byte) []protocol.Fi
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
blocks, _ := scanner.Blocks(context.TODO(), bytes.NewReader(data), protocol.BlockSize(int64(len(data))), int64(len(data)), nil, true)
|
||||
blocks, _ := scanner.Blocks(context.TODO(), bytes.NewReader(data), protocol.BlockSize(int64(len(data))), int64(len(data)), nil)
|
||||
knownFiles := []protocol.FileInfo{
|
||||
{
|
||||
Name: "knownDir",
|
||||
|
||||
+40
-116
@@ -31,7 +31,6 @@ import (
|
||||
"github.com/syncthing/syncthing/lib/semaphore"
|
||||
"github.com/syncthing/syncthing/lib/sync"
|
||||
"github.com/syncthing/syncthing/lib/versioner"
|
||||
"github.com/syncthing/syncthing/lib/weakhash"
|
||||
)
|
||||
|
||||
var (
|
||||
@@ -1136,12 +1135,12 @@ func (f *sendReceiveFolder) handleFile(file protocol.FileInfo, snap *db.Snapshot
|
||||
func (f *sendReceiveFolder) reuseBlocks(blocks []protocol.BlockInfo, reused []int, file protocol.FileInfo, tempName string) ([]protocol.BlockInfo, []int) {
|
||||
// Check for an old temporary file which might have some blocks we could
|
||||
// reuse.
|
||||
tempBlocks, err := scanner.HashFile(f.ctx, f.ID, f.mtimefs, tempName, file.BlockSize(), nil, false)
|
||||
tempBlocks, err := scanner.HashFile(f.ctx, f.ID, f.mtimefs, tempName, file.BlockSize(), nil)
|
||||
if err != nil {
|
||||
var caseErr *fs.ErrCaseConflict
|
||||
if errors.As(err, &caseErr) {
|
||||
if rerr := f.mtimefs.Rename(caseErr.Real, tempName); rerr == nil {
|
||||
tempBlocks, err = scanner.HashFile(f.ctx, f.ID, f.mtimefs, tempName, file.BlockSize(), nil, false)
|
||||
tempBlocks, err = scanner.HashFile(f.ctx, f.ID, f.mtimefs, tempName, file.BlockSize(), nil)
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -1310,8 +1309,6 @@ func (f *sendReceiveFolder) copierRoutine(in <-chan copyBlocksState, pullChan ch
|
||||
f.model.progressEmitter.Register(state.sharedPullerState)
|
||||
}
|
||||
|
||||
weakHashFinder, file := f.initWeakHashFinder(state)
|
||||
|
||||
blocks:
|
||||
for _, block := range state.blocks {
|
||||
select {
|
||||
@@ -1336,75 +1333,49 @@ func (f *sendReceiveFolder) copierRoutine(in <-chan copyBlocksState, pullChan ch
|
||||
|
||||
buf = protocol.BufferPool.Upgrade(buf, int(block.Size))
|
||||
|
||||
var found bool
|
||||
if f.Type != config.FolderTypeReceiveEncrypted {
|
||||
found, err = weakHashFinder.Iterate(block.WeakHash, buf, func(offset int64) bool {
|
||||
if f.verifyBuffer(buf, block) != nil {
|
||||
return true
|
||||
}
|
||||
|
||||
err = f.limitedWriteAt(dstFd, buf, block.Offset)
|
||||
if err != nil {
|
||||
state.fail(fmt.Errorf("dst write: %w", err))
|
||||
}
|
||||
if offset == block.Offset {
|
||||
state.copiedFromOrigin(block.Size)
|
||||
} else {
|
||||
state.copiedFromOriginShifted(block.Size)
|
||||
}
|
||||
|
||||
return false
|
||||
})
|
||||
found := f.model.finder.Iterate(folders, block.Hash, func(folder, path string, index int32) bool {
|
||||
ffs := folderFilesystems[folder]
|
||||
fd, err := ffs.Open(path)
|
||||
if err != nil {
|
||||
l.Debugln("weak hasher iter", err)
|
||||
return false
|
||||
}
|
||||
}
|
||||
defer fd.Close()
|
||||
|
||||
if !found {
|
||||
found = f.model.finder.Iterate(folders, block.Hash, func(folder, path string, index int32) bool {
|
||||
ffs := folderFilesystems[folder]
|
||||
fd, err := ffs.Open(path)
|
||||
if err != nil {
|
||||
srcOffset := int64(state.file.BlockSize()) * int64(index)
|
||||
_, err = fd.ReadAt(buf, srcOffset)
|
||||
if err != nil {
|
||||
return false
|
||||
}
|
||||
|
||||
// Hash is not SHA256 as it's an encrypted hash token. In that
|
||||
// case we can't verify the block integrity so we'll take it on
|
||||
// trust. (The other side can and will verify.)
|
||||
if f.Type != config.FolderTypeReceiveEncrypted {
|
||||
if err := f.verifyBuffer(buf, block); err != nil {
|
||||
l.Debugln("Finder failed to verify buffer", err)
|
||||
return false
|
||||
}
|
||||
defer fd.Close()
|
||||
}
|
||||
|
||||
srcOffset := int64(state.file.BlockSize()) * int64(index)
|
||||
_, err = fd.ReadAt(buf, srcOffset)
|
||||
if err != nil {
|
||||
return false
|
||||
}
|
||||
|
||||
// Hash is not SHA256 as it's an encrypted hash token. In that
|
||||
// case we can't verify the block integrity so we'll take it on
|
||||
// trust. (The other side can and will verify.)
|
||||
if f.Type != config.FolderTypeReceiveEncrypted {
|
||||
if err := f.verifyBuffer(buf, block); err != nil {
|
||||
l.Debugln("Finder failed to verify buffer", err)
|
||||
return false
|
||||
}
|
||||
}
|
||||
|
||||
if f.CopyRangeMethod != config.CopyRangeMethodStandard {
|
||||
err = f.withLimiter(func() error {
|
||||
dstFd.mut.Lock()
|
||||
defer dstFd.mut.Unlock()
|
||||
return fs.CopyRange(f.CopyRangeMethod.ToFS(), fd, dstFd.fd, srcOffset, block.Offset, int64(block.Size))
|
||||
})
|
||||
} else {
|
||||
err = f.limitedWriteAt(dstFd, buf, block.Offset)
|
||||
}
|
||||
if err != nil {
|
||||
state.fail(fmt.Errorf("dst write: %w", err))
|
||||
}
|
||||
if path == state.file.Name {
|
||||
state.copiedFromOrigin(block.Size)
|
||||
} else {
|
||||
state.copiedFromElsewhere(block.Size)
|
||||
}
|
||||
return true
|
||||
})
|
||||
}
|
||||
if f.CopyRangeMethod != config.CopyRangeMethodStandard {
|
||||
err = f.withLimiter(func() error {
|
||||
dstFd.mut.Lock()
|
||||
defer dstFd.mut.Unlock()
|
||||
return fs.CopyRange(f.CopyRangeMethod.ToFS(), fd, dstFd.fd, srcOffset, block.Offset, int64(block.Size))
|
||||
})
|
||||
} else {
|
||||
err = f.limitedWriteAt(dstFd, buf, block.Offset)
|
||||
}
|
||||
if err != nil {
|
||||
state.fail(fmt.Errorf("dst write: %w", err))
|
||||
}
|
||||
if path == state.file.Name {
|
||||
state.copiedFromOrigin(block.Size)
|
||||
} else {
|
||||
state.copiedFromElsewhere(block.Size)
|
||||
}
|
||||
return true
|
||||
})
|
||||
|
||||
if state.failed() != nil {
|
||||
break
|
||||
@@ -1421,58 +1392,11 @@ func (f *sendReceiveFolder) copierRoutine(in <-chan copyBlocksState, pullChan ch
|
||||
state.copyDone(block)
|
||||
}
|
||||
}
|
||||
if file != nil {
|
||||
// os.File used to return invalid argument if nil.
|
||||
// fs.File panics as it's an interface.
|
||||
file.Close()
|
||||
}
|
||||
|
||||
out <- state.sharedPullerState
|
||||
}
|
||||
}
|
||||
|
||||
func (f *sendReceiveFolder) initWeakHashFinder(state copyBlocksState) (*weakhash.Finder, fs.File) {
|
||||
if f.Type == config.FolderTypeReceiveEncrypted {
|
||||
l.Debugln("not weak hashing due to folder type", f.Type)
|
||||
return nil, nil
|
||||
}
|
||||
|
||||
blocksPercentChanged := 0
|
||||
if tot := len(state.file.Blocks); tot > 0 {
|
||||
blocksPercentChanged = (tot - state.have) * 100 / tot
|
||||
}
|
||||
|
||||
if blocksPercentChanged < f.WeakHashThresholdPct {
|
||||
l.Debugf("not weak hashing %s. not enough changed %.02f < %d", state.file.Name, blocksPercentChanged, f.WeakHashThresholdPct)
|
||||
return nil, nil
|
||||
}
|
||||
|
||||
hashesToFind := make([]uint32, 0, len(state.blocks))
|
||||
for _, block := range state.blocks {
|
||||
if block.WeakHash != 0 {
|
||||
hashesToFind = append(hashesToFind, block.WeakHash)
|
||||
}
|
||||
}
|
||||
|
||||
if len(hashesToFind) == 0 {
|
||||
l.Debugf("not weak hashing %s. file did not contain any weak hashes", state.file.Name)
|
||||
return nil, nil
|
||||
}
|
||||
|
||||
file, err := f.mtimefs.Open(state.file.Name)
|
||||
if err != nil {
|
||||
l.Debugln("weak hasher", err)
|
||||
return nil, nil
|
||||
}
|
||||
|
||||
weakHashFinder, err := weakhash.NewFinder(f.ctx, file, state.file.BlockSize(), hashesToFind)
|
||||
if err != nil {
|
||||
l.Debugln("weak hasher", err)
|
||||
return nil, file
|
||||
}
|
||||
return weakHashFinder, file
|
||||
}
|
||||
|
||||
func (*sendReceiveFolder) verifyBuffer(buf []byte, block protocol.BlockInfo) error {
|
||||
if len(buf) != int(block.Size) {
|
||||
return fmt.Errorf("length mismatch %d != %d", len(buf), block.Size)
|
||||
@@ -1574,7 +1498,7 @@ loop:
|
||||
activity.using(selected)
|
||||
var buf []byte
|
||||
blockNo := int(state.block.Offset / int64(state.file.BlockSize()))
|
||||
buf, lastError = f.model.RequestGlobal(f.ctx, selected.ID, f.folderID, state.file.Name, blockNo, state.block.Offset, int(state.block.Size), state.block.Hash, state.block.WeakHash, selected.FromTemporary)
|
||||
buf, lastError = f.model.RequestGlobal(f.ctx, selected.ID, f.folderID, state.file.Name, blockNo, state.block.Offset, int(state.block.Size), state.block.Hash, selected.FromTemporary)
|
||||
activity.done(selected)
|
||||
if lastError != nil {
|
||||
l.Debugln("request:", f.folderID, state.file.Name, state.block.Offset, state.block.Size, selected.ID.Short(), "returned error:", lastError)
|
||||
|
||||
@@ -25,7 +25,6 @@ import (
|
||||
"github.com/syncthing/syncthing/lib/fs"
|
||||
"github.com/syncthing/syncthing/lib/ignore"
|
||||
"github.com/syncthing/syncthing/lib/protocol"
|
||||
"github.com/syncthing/syncthing/lib/rand"
|
||||
"github.com/syncthing/syncthing/lib/scanner"
|
||||
"github.com/syncthing/syncthing/lib/sync"
|
||||
)
|
||||
@@ -300,7 +299,7 @@ func TestCopierFinder(t *testing.T) {
|
||||
}
|
||||
|
||||
// Verify that the fetched blocks have actually been written to the temp file
|
||||
blks, err := scanner.HashFile(context.TODO(), f.ID, f.Filesystem(nil), tempFile, protocol.MinBlockSize, nil, false)
|
||||
blks, err := scanner.HashFile(context.TODO(), f.ID, f.Filesystem(nil), tempFile, protocol.MinBlockSize, nil)
|
||||
if err != nil {
|
||||
t.Log(err)
|
||||
}
|
||||
@@ -312,128 +311,6 @@ func TestCopierFinder(t *testing.T) {
|
||||
}
|
||||
}
|
||||
|
||||
func TestWeakHash(t *testing.T) {
|
||||
// Setup the model/pull environment
|
||||
_, fo, wcfgCancel := setupSendReceiveFolder(t)
|
||||
defer wcfgCancel()
|
||||
ffs := fo.Filesystem(nil)
|
||||
|
||||
tempFile := fs.TempName("weakhash")
|
||||
var shift int64 = 10
|
||||
var size int64 = 1 << 20
|
||||
expectBlocks := int(size / protocol.MinBlockSize)
|
||||
expectPulls := int(shift / protocol.MinBlockSize)
|
||||
if shift > 0 {
|
||||
expectPulls++
|
||||
}
|
||||
|
||||
f, err := ffs.Create("weakhash")
|
||||
must(t, err)
|
||||
defer f.Close()
|
||||
_, err = io.CopyN(f, rand.Reader, size)
|
||||
if err != nil {
|
||||
t.Error(err)
|
||||
}
|
||||
info, err := f.Stat()
|
||||
if err != nil {
|
||||
t.Error(err)
|
||||
}
|
||||
|
||||
// Create two files, second file has `shifted` bytes random prefix, yet
|
||||
// both are of the same length, for example:
|
||||
// File 1: abcdefgh
|
||||
// File 2: xyabcdef
|
||||
f.Seek(0, io.SeekStart)
|
||||
existing, err := scanner.Blocks(context.TODO(), f, protocol.MinBlockSize, size, nil, true)
|
||||
if err != nil {
|
||||
t.Error(err)
|
||||
}
|
||||
|
||||
f.Seek(0, io.SeekStart)
|
||||
remainder := io.LimitReader(f, size-shift)
|
||||
prefix := io.LimitReader(rand.Reader, shift)
|
||||
nf := io.MultiReader(prefix, remainder)
|
||||
desired, err := scanner.Blocks(context.TODO(), nf, protocol.MinBlockSize, size, nil, true)
|
||||
if err != nil {
|
||||
t.Error(err)
|
||||
}
|
||||
|
||||
existingFile := protocol.FileInfo{
|
||||
Name: "weakhash",
|
||||
Blocks: existing,
|
||||
Size: size,
|
||||
ModifiedS: info.ModTime().Unix(),
|
||||
ModifiedNs: int32(info.ModTime().Nanosecond()),
|
||||
}
|
||||
desiredFile := protocol.FileInfo{
|
||||
Name: "weakhash",
|
||||
Size: size,
|
||||
Blocks: desired,
|
||||
ModifiedS: info.ModTime().Unix() + 1,
|
||||
}
|
||||
|
||||
fo.updateLocalsFromScanning([]protocol.FileInfo{existingFile})
|
||||
|
||||
copyChan := make(chan copyBlocksState)
|
||||
pullChan := make(chan pullBlockState, expectBlocks)
|
||||
finisherChan := make(chan *sharedPullerState, 1)
|
||||
|
||||
// Run a single fetcher routine
|
||||
go fo.copierRoutine(copyChan, pullChan, finisherChan)
|
||||
defer close(copyChan)
|
||||
|
||||
// Test 1 - no weak hashing, file gets fully repulled (`expectBlocks` pulls).
|
||||
fo.WeakHashThresholdPct = 101
|
||||
fo.handleFile(desiredFile, fsetSnapshot(t, fo.fset), copyChan)
|
||||
|
||||
var pulls []pullBlockState
|
||||
timeout := time.After(10 * time.Second)
|
||||
for len(pulls) < expectBlocks {
|
||||
select {
|
||||
case pull := <-pullChan:
|
||||
pulls = append(pulls, pull)
|
||||
case <-timeout:
|
||||
t.Fatalf("timed out, got %d pulls expected %d", len(pulls), expectPulls)
|
||||
}
|
||||
}
|
||||
finish := <-finisherChan
|
||||
|
||||
select {
|
||||
case <-pullChan:
|
||||
t.Fatal("Pull channel has data to be read")
|
||||
case <-finisherChan:
|
||||
t.Fatal("Finisher channel has data to be read")
|
||||
default:
|
||||
}
|
||||
|
||||
cleanupSharedPullerState(finish)
|
||||
if err := ffs.Remove(tempFile); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
|
||||
// Test 2 - using weak hash, expectPulls blocks pulled.
|
||||
fo.WeakHashThresholdPct = -1
|
||||
fo.handleFile(desiredFile, fsetSnapshot(t, fo.fset), copyChan)
|
||||
|
||||
pulls = pulls[:0]
|
||||
for len(pulls) < expectPulls {
|
||||
select {
|
||||
case pull := <-pullChan:
|
||||
pulls = append(pulls, pull)
|
||||
case <-time.After(10 * time.Second):
|
||||
t.Fatalf("timed out, got %d pulls expected %d", len(pulls), expectPulls)
|
||||
}
|
||||
}
|
||||
|
||||
finish = <-finisherChan
|
||||
cleanupSharedPullerState(finish)
|
||||
|
||||
expectShifted := expectBlocks - expectPulls
|
||||
if finish.copyOriginShifted != expectShifted {
|
||||
t.Errorf("did not copy %d shifted", expectShifted)
|
||||
}
|
||||
}
|
||||
|
||||
// Test that updating a file removes its old blocks from the blockmap
|
||||
func TestCopierCleanup(t *testing.T) {
|
||||
iterFn := func(folder, file string, index int32) bool {
|
||||
@@ -709,8 +586,8 @@ func TestIssue3164(t *testing.T) {
|
||||
|
||||
func TestDiff(t *testing.T) {
|
||||
for i, test := range diffTestData {
|
||||
a, _ := scanner.Blocks(context.TODO(), bytes.NewBufferString(test.a), test.s, -1, nil, false)
|
||||
b, _ := scanner.Blocks(context.TODO(), bytes.NewBufferString(test.b), test.s, -1, nil, false)
|
||||
a, _ := scanner.Blocks(context.TODO(), bytes.NewBufferString(test.a), test.s, -1, nil)
|
||||
b, _ := scanner.Blocks(context.TODO(), bytes.NewBufferString(test.b), test.s, -1, nil)
|
||||
_, d := blockDiff(a, b)
|
||||
if len(d) != len(test.d) {
|
||||
t.Fatalf("Incorrect length for diff %d; %d != %d", i, len(d), len(test.d))
|
||||
@@ -730,8 +607,8 @@ func TestDiff(t *testing.T) {
|
||||
func BenchmarkDiff(b *testing.B) {
|
||||
testCases := make([]struct{ a, b []protocol.BlockInfo }, 0, len(diffTestData))
|
||||
for _, test := range diffTestData {
|
||||
a, _ := scanner.Blocks(context.TODO(), bytes.NewBufferString(test.a), test.s, -1, nil, false)
|
||||
b, _ := scanner.Blocks(context.TODO(), bytes.NewBufferString(test.b), test.s, -1, nil, false)
|
||||
a, _ := scanner.Blocks(context.TODO(), bytes.NewBufferString(test.a), test.s, -1, nil)
|
||||
b, _ := scanner.Blocks(context.TODO(), bytes.NewBufferString(test.b), test.s, -1, nil)
|
||||
testCases = append(testCases, struct{ a, b []protocol.BlockInfo }{a, b})
|
||||
}
|
||||
b.ReportAllocs()
|
||||
|
||||
@@ -55,16 +55,15 @@ var (
|
||||
Namespace: "syncthing",
|
||||
Subsystem: "model",
|
||||
Name: "folder_processed_bytes_total",
|
||||
Help: "Total amount of data processed during folder syncing, per folder ID and data source (network/local_origin/local_other/local_shifted/skipped)",
|
||||
Help: "Total amount of data processed during folder syncing, per folder ID and data source (network/local_origin/local_other/skipped)",
|
||||
}, []string{"folder", "source"})
|
||||
)
|
||||
|
||||
const (
|
||||
metricSourceNetwork = "network" // from the network
|
||||
metricSourceLocalOrigin = "local_origin" // from the existing version of the local file
|
||||
metricSourceLocalOther = "local_other" // from a different local file
|
||||
metricSourceLocalShifted = "local_shifted" // from the existing version of the local file, rolling hash shifted
|
||||
metricSourceSkipped = "skipped" // block of all zeroes, invented out of thin air
|
||||
metricSourceNetwork = "network" // from the network
|
||||
metricSourceLocalOrigin = "local_origin" // from the existing version of the local file
|
||||
metricSourceLocalOther = "local_other" // from a different local file
|
||||
metricSourceSkipped = "skipped" // block of all zeroes, invented out of thin air
|
||||
|
||||
metricScopeGlobal = "global"
|
||||
metricScopeLocal = "local"
|
||||
@@ -88,6 +87,5 @@ func registerFolderMetrics(folderID string) {
|
||||
metricFolderProcessedBytesTotal.WithLabelValues(folderID, metricSourceNetwork)
|
||||
metricFolderProcessedBytesTotal.WithLabelValues(folderID, metricSourceLocalOrigin)
|
||||
metricFolderProcessedBytesTotal.WithLabelValues(folderID, metricSourceLocalOther)
|
||||
metricFolderProcessedBytesTotal.WithLabelValues(folderID, metricSourceLocalShifted)
|
||||
metricFolderProcessedBytesTotal.WithLabelValues(folderID, metricSourceSkipped)
|
||||
}
|
||||
|
||||
+26
-28
@@ -435,19 +435,18 @@ type Model struct {
|
||||
result1 protocol.RequestResponse
|
||||
result2 error
|
||||
}
|
||||
RequestGlobalStub func(context.Context, protocol.DeviceID, string, string, int, int64, int, []byte, uint32, bool) ([]byte, error)
|
||||
RequestGlobalStub func(context.Context, protocol.DeviceID, string, string, int, int64, int, []byte, bool) ([]byte, error)
|
||||
requestGlobalMutex sync.RWMutex
|
||||
requestGlobalArgsForCall []struct {
|
||||
arg1 context.Context
|
||||
arg2 protocol.DeviceID
|
||||
arg3 string
|
||||
arg4 string
|
||||
arg5 int
|
||||
arg6 int64
|
||||
arg7 int
|
||||
arg8 []byte
|
||||
arg9 uint32
|
||||
arg10 bool
|
||||
arg1 context.Context
|
||||
arg2 protocol.DeviceID
|
||||
arg3 string
|
||||
arg4 string
|
||||
arg5 int
|
||||
arg6 int64
|
||||
arg7 int
|
||||
arg8 []byte
|
||||
arg9 bool
|
||||
}
|
||||
requestGlobalReturns struct {
|
||||
result1 []byte
|
||||
@@ -2580,7 +2579,7 @@ func (fake *Model) RequestReturnsOnCall(i int, result1 protocol.RequestResponse,
|
||||
}{result1, result2}
|
||||
}
|
||||
|
||||
func (fake *Model) RequestGlobal(arg1 context.Context, arg2 protocol.DeviceID, arg3 string, arg4 string, arg5 int, arg6 int64, arg7 int, arg8 []byte, arg9 uint32, arg10 bool) ([]byte, error) {
|
||||
func (fake *Model) RequestGlobal(arg1 context.Context, arg2 protocol.DeviceID, arg3 string, arg4 string, arg5 int, arg6 int64, arg7 int, arg8 []byte, arg9 bool) ([]byte, error) {
|
||||
var arg8Copy []byte
|
||||
if arg8 != nil {
|
||||
arg8Copy = make([]byte, len(arg8))
|
||||
@@ -2589,23 +2588,22 @@ func (fake *Model) RequestGlobal(arg1 context.Context, arg2 protocol.DeviceID, a
|
||||
fake.requestGlobalMutex.Lock()
|
||||
ret, specificReturn := fake.requestGlobalReturnsOnCall[len(fake.requestGlobalArgsForCall)]
|
||||
fake.requestGlobalArgsForCall = append(fake.requestGlobalArgsForCall, struct {
|
||||
arg1 context.Context
|
||||
arg2 protocol.DeviceID
|
||||
arg3 string
|
||||
arg4 string
|
||||
arg5 int
|
||||
arg6 int64
|
||||
arg7 int
|
||||
arg8 []byte
|
||||
arg9 uint32
|
||||
arg10 bool
|
||||
}{arg1, arg2, arg3, arg4, arg5, arg6, arg7, arg8Copy, arg9, arg10})
|
||||
arg1 context.Context
|
||||
arg2 protocol.DeviceID
|
||||
arg3 string
|
||||
arg4 string
|
||||
arg5 int
|
||||
arg6 int64
|
||||
arg7 int
|
||||
arg8 []byte
|
||||
arg9 bool
|
||||
}{arg1, arg2, arg3, arg4, arg5, arg6, arg7, arg8Copy, arg9})
|
||||
stub := fake.RequestGlobalStub
|
||||
fakeReturns := fake.requestGlobalReturns
|
||||
fake.recordInvocation("RequestGlobal", []interface{}{arg1, arg2, arg3, arg4, arg5, arg6, arg7, arg8Copy, arg9, arg10})
|
||||
fake.recordInvocation("RequestGlobal", []interface{}{arg1, arg2, arg3, arg4, arg5, arg6, arg7, arg8Copy, arg9})
|
||||
fake.requestGlobalMutex.Unlock()
|
||||
if stub != nil {
|
||||
return stub(arg1, arg2, arg3, arg4, arg5, arg6, arg7, arg8, arg9, arg10)
|
||||
return stub(arg1, arg2, arg3, arg4, arg5, arg6, arg7, arg8, arg9)
|
||||
}
|
||||
if specificReturn {
|
||||
return ret.result1, ret.result2
|
||||
@@ -2619,17 +2617,17 @@ func (fake *Model) RequestGlobalCallCount() int {
|
||||
return len(fake.requestGlobalArgsForCall)
|
||||
}
|
||||
|
||||
func (fake *Model) RequestGlobalCalls(stub func(context.Context, protocol.DeviceID, string, string, int, int64, int, []byte, uint32, bool) ([]byte, error)) {
|
||||
func (fake *Model) RequestGlobalCalls(stub func(context.Context, protocol.DeviceID, string, string, int, int64, int, []byte, bool) ([]byte, error)) {
|
||||
fake.requestGlobalMutex.Lock()
|
||||
defer fake.requestGlobalMutex.Unlock()
|
||||
fake.RequestGlobalStub = stub
|
||||
}
|
||||
|
||||
func (fake *Model) RequestGlobalArgsForCall(i int) (context.Context, protocol.DeviceID, string, string, int, int64, int, []byte, uint32, bool) {
|
||||
func (fake *Model) RequestGlobalArgsForCall(i int) (context.Context, protocol.DeviceID, string, string, int, int64, int, []byte, bool) {
|
||||
fake.requestGlobalMutex.RLock()
|
||||
defer fake.requestGlobalMutex.RUnlock()
|
||||
argsForCall := fake.requestGlobalArgsForCall[i]
|
||||
return argsForCall.arg1, argsForCall.arg2, argsForCall.arg3, argsForCall.arg4, argsForCall.arg5, argsForCall.arg6, argsForCall.arg7, argsForCall.arg8, argsForCall.arg9, argsForCall.arg10
|
||||
return argsForCall.arg1, argsForCall.arg2, argsForCall.arg3, argsForCall.arg4, argsForCall.arg5, argsForCall.arg6, argsForCall.arg7, argsForCall.arg8, argsForCall.arg9
|
||||
}
|
||||
|
||||
func (fake *Model) RequestGlobalReturns(result1 []byte, result2 error) {
|
||||
|
||||
+8
-14
@@ -118,7 +118,7 @@ type Model interface {
|
||||
|
||||
GlobalDirectoryTree(folder, prefix string, levels int, dirsOnly bool) ([]*TreeEntry, error)
|
||||
|
||||
RequestGlobal(ctx context.Context, deviceID protocol.DeviceID, folder, name string, blockNo int, offset int64, size int, hash []byte, weakHash uint32, fromTemporary bool) ([]byte, error)
|
||||
RequestGlobal(ctx context.Context, deviceID protocol.DeviceID, folder, name string, blockNo int, offset int64, size int, hash []byte, fromTemporary bool) ([]byte, error)
|
||||
}
|
||||
|
||||
type model struct {
|
||||
@@ -606,8 +606,6 @@ func (m *model) UsageReportingStats(report *contract.Report, version int, previe
|
||||
report.BlockStats.Pulled = v
|
||||
case "copyOrigin":
|
||||
report.BlockStats.CopyOrigin = v
|
||||
case "copyOriginShifted":
|
||||
report.BlockStats.CopyOriginShifted = v
|
||||
case "copyElsewhere":
|
||||
report.BlockStats.CopyElsewhere = v
|
||||
}
|
||||
@@ -2036,7 +2034,7 @@ func (m *model) Request(conn protocol.Connection, req *protocol.Request) (out pr
|
||||
return nil, protocol.ErrNoSuchFile
|
||||
}
|
||||
_, err := readOffsetIntoBuf(folderFs, tempFn, req.Offset, res.data)
|
||||
if err == nil && scanner.Validate(res.data, req.Hash, req.WeakHash) {
|
||||
if err == nil && scanner.Validate(res.data, req.Hash) {
|
||||
return res, nil
|
||||
}
|
||||
// Fall through to reading from a non-temp file, just in case the temp
|
||||
@@ -2065,8 +2063,8 @@ func (m *model) Request(conn protocol.Connection, req *protocol.Request) (out pr
|
||||
return nil, protocol.ErrGeneric
|
||||
}
|
||||
|
||||
if folderCfg.Type != config.FolderTypeReceiveEncrypted && len(req.Hash) > 0 && !scanner.Validate(res.data[:n], req.Hash, req.WeakHash) {
|
||||
m.recheckFile(deviceID, req.Folder, req.Name, req.Offset, req.Hash, req.WeakHash)
|
||||
if folderCfg.Type != config.FolderTypeReceiveEncrypted && len(req.Hash) > 0 && !scanner.Validate(res.data[:n], req.Hash) {
|
||||
m.recheckFile(deviceID, req.Folder, req.Name, req.Offset, req.Hash)
|
||||
l.Debugf("%v REQ(in) failed validating data: %s: %q / %q o=%d s=%d", m, deviceID.Short(), req.Folder, req.Name, req.Offset, req.Size)
|
||||
return nil, protocol.ErrNoSuchFile
|
||||
}
|
||||
@@ -2092,7 +2090,7 @@ func newLimitedRequestResponse(size int, limiters ...*semaphore.Semaphore) *requ
|
||||
return res
|
||||
}
|
||||
|
||||
func (m *model) recheckFile(deviceID protocol.DeviceID, folder, name string, offset int64, hash []byte, weakHash uint32) {
|
||||
func (m *model) recheckFile(deviceID protocol.DeviceID, folder, name string, offset int64, hash []byte) {
|
||||
cf, ok, err := m.CurrentFolderFile(folder, name)
|
||||
if err != nil {
|
||||
l.Debugf("%v recheckFile: %s: %q / %q: current file error: %v", m, deviceID, folder, name, err)
|
||||
@@ -2121,10 +2119,6 @@ func (m *model) recheckFile(deviceID protocol.DeviceID, folder, name string, off
|
||||
l.Debugf("%v recheckFile: %s: %q / %q i=%d: hash mismatch %x != %x", m, deviceID, folder, name, blockIndex, block.Hash, hash)
|
||||
return
|
||||
}
|
||||
if weakHash != 0 && block.WeakHash != weakHash {
|
||||
l.Debugf("%v recheckFile: %s: %q / %q i=%d: weak hash mismatch %v != %v", m, deviceID, folder, name, blockIndex, block.WeakHash, weakHash)
|
||||
return
|
||||
}
|
||||
|
||||
// The hashes provided part of the request match what we expect to find according
|
||||
// to what we have in the database, yet the content we've read off the filesystem doesn't
|
||||
@@ -2462,14 +2456,14 @@ func (m *model) deviceDidCloseRLocked(deviceID protocol.DeviceID, duration time.
|
||||
}
|
||||
}
|
||||
|
||||
func (m *model) RequestGlobal(ctx context.Context, deviceID protocol.DeviceID, folder, name string, blockNo int, offset int64, size int, hash []byte, weakHash uint32, fromTemporary bool) ([]byte, error) {
|
||||
func (m *model) RequestGlobal(ctx context.Context, deviceID protocol.DeviceID, folder, name string, blockNo int, offset int64, size int, hash []byte, fromTemporary bool) ([]byte, error) {
|
||||
conn, connOK := m.requestConnectionForDevice(deviceID)
|
||||
if !connOK {
|
||||
return nil, fmt.Errorf("requestGlobal: no connection to device: %s", deviceID.Short())
|
||||
}
|
||||
|
||||
l.Debugf("%v REQ(out): %s (%s): %q / %q b=%d o=%d s=%d h=%x wh=%x ft=%t", m, deviceID.Short(), conn, folder, name, blockNo, offset, size, hash, weakHash, fromTemporary)
|
||||
return conn.Request(ctx, &protocol.Request{Folder: folder, Name: name, BlockNo: blockNo, Offset: offset, Size: size, Hash: hash, WeakHash: weakHash, FromTemporary: fromTemporary})
|
||||
l.Debugf("%v REQ(out): %s (%s): %q / %q b=%d o=%d s=%d h=%x ft=%t", m, deviceID.Short(), conn, folder, name, blockNo, offset, size, hash, fromTemporary)
|
||||
return conn.Request(ctx, &protocol.Request{Folder: folder, Name: name, BlockNo: blockNo, Offset: offset, Size: size, Hash: hash, FromTemporary: fromTemporary})
|
||||
}
|
||||
|
||||
// requestConnectionForDevice returns a connection to the given device, to
|
||||
|
||||
@@ -222,7 +222,7 @@ func BenchmarkRequestOut(b *testing.B) {
|
||||
|
||||
b.ResetTimer()
|
||||
for i := 0; i < b.N; i++ {
|
||||
data, err := m.RequestGlobal(context.Background(), device1, "default", files[i%n].Name, 0, 0, 32, nil, 0, false)
|
||||
data, err := m.RequestGlobal(context.Background(), device1, "default", files[i%n].Name, 0, 0, 32, nil, false)
|
||||
if err != nil {
|
||||
b.Error(err)
|
||||
}
|
||||
@@ -2819,9 +2819,9 @@ func TestIssue5002(t *testing.T) {
|
||||
}
|
||||
blockSize := int32(file.BlockSize())
|
||||
|
||||
m.recheckFile(protocol.LocalDeviceID, "default", "foo", file.Size-int64(blockSize), []byte{1, 2, 3, 4}, 0)
|
||||
m.recheckFile(protocol.LocalDeviceID, "default", "foo", file.Size, []byte{1, 2, 3, 4}, 0) // panic
|
||||
m.recheckFile(protocol.LocalDeviceID, "default", "foo", file.Size+int64(blockSize), []byte{1, 2, 3, 4}, 0)
|
||||
m.recheckFile(protocol.LocalDeviceID, "default", "foo", file.Size-int64(blockSize), []byte{1, 2, 3, 4})
|
||||
m.recheckFile(protocol.LocalDeviceID, "default", "foo", file.Size, []byte{1, 2, 3, 4}) // panic
|
||||
m.recheckFile(protocol.LocalDeviceID, "default", "foo", file.Size+int64(blockSize), []byte{1, 2, 3, 4})
|
||||
}
|
||||
|
||||
func TestParentOfUnignored(t *testing.T) {
|
||||
|
||||
@@ -436,11 +436,8 @@ func TestRescanIfHaveInvalidContent(t *testing.T) {
|
||||
}
|
||||
|
||||
f := checkReceived(<-received)
|
||||
if f.Blocks[0].WeakHash != 103547413 {
|
||||
t.Fatalf("unexpected weak hash: %d != 103547413", f.Blocks[0].WeakHash)
|
||||
}
|
||||
|
||||
res, err := m.Request(device1Conn, &protocol.Request{Folder: "default", Name: "foo", Size: len(payload), Hash: f.Blocks[0].Hash, WeakHash: f.Blocks[0].WeakHash})
|
||||
res, err := m.Request(device1Conn, &protocol.Request{Folder: "default", Name: "foo", Size: len(payload), Hash: f.Blocks[0].Hash})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
@@ -454,17 +451,14 @@ func TestRescanIfHaveInvalidContent(t *testing.T) {
|
||||
|
||||
writeFile(t, tfs, "foo", payload)
|
||||
|
||||
_, err = m.Request(device1Conn, &protocol.Request{Folder: "default", Name: "foo", Size: len(payload), Hash: f.Blocks[0].Hash, WeakHash: f.Blocks[0].WeakHash})
|
||||
_, err = m.Request(device1Conn, &protocol.Request{Folder: "default", Name: "foo", Size: len(payload), Hash: f.Blocks[0].Hash})
|
||||
if err == nil {
|
||||
t.Fatalf("expected failure")
|
||||
}
|
||||
|
||||
select {
|
||||
case fs := <-received:
|
||||
f := checkReceived(fs)
|
||||
if f.Blocks[0].WeakHash != 41943361 {
|
||||
t.Fatalf("unexpected weak hash: %d != 41943361", f.Blocks[0].WeakHash)
|
||||
}
|
||||
checkReceived(fs)
|
||||
case <-time.After(time.Second):
|
||||
t.Fatalf("timed out")
|
||||
}
|
||||
|
||||
@@ -285,15 +285,6 @@ func (s *sharedPullerState) skippedSparseBlock(bytes int) {
|
||||
metricFolderProcessedBytesTotal.WithLabelValues(s.folder, metricSourceSkipped).Add(float64(bytes))
|
||||
}
|
||||
|
||||
func (s *sharedPullerState) copiedFromOriginShifted(bytes int) {
|
||||
s.mut.Lock()
|
||||
s.copyOrigin++
|
||||
s.copyOriginShifted++
|
||||
s.updated = time.Now()
|
||||
s.mut.Unlock()
|
||||
metricFolderProcessedBytesTotal.WithLabelValues(s.folder, metricSourceLocalShifted).Add(float64(bytes))
|
||||
}
|
||||
|
||||
func (s *sharedPullerState) pullStarted() {
|
||||
s.mut.Lock()
|
||||
s.copyTotal--
|
||||
|
||||
Reference in New Issue
Block a user