cmd/syncthing, lib/db, lib/model, lib/protocol: Implement delta indexes (fixes #438)

GitHub-Pull-Request: https://github.com/syncthing/syncthing/pull/3427
2016-07-23 12:46:31 +00:00
parent 8ab6b60778
commit 47fa4b0a2c
11 changed files with 428 additions and 222 deletions
--- a/lib/db/leveldb.go
+++ b/lib/db/leveldb.go
@@ -11,27 +11,10 @@ import (
 	"fmt"

 	"github.com/syncthing/syncthing/lib/protocol"
-	"github.com/syncthing/syncthing/lib/sync"
 	"github.com/syndtr/goleveldb/leveldb"
 	"github.com/syndtr/goleveldb/leveldb/opt"
 )

-var (
-	clockTick int64
-	clockMut  = sync.NewMutex()
-)
-
-func clock(v int64) int64 {
-	clockMut.Lock()
-	defer clockMut.Unlock()
-	if v > clockTick {
-		clockTick = v + 1
-	} else {
-		clockTick++
-	}
-	return clockTick
-}
-
 const (
 	KeyTypeDevice = iota
 	KeyTypeGlobal
@@ -41,6 +24,7 @@ const (
 	KeyTypeVirtualMtime
 	KeyTypeFolderIdx
 	KeyTypeDeviceIdx
+	KeyTypeIndexID
 )

 func (l VersionList) String() string {
--- a/lib/db/leveldb_dbinstance.go
+++ b/lib/db/leveldb_dbinstance.go
@@ -24,7 +24,7 @@ import (
 	"github.com/syndtr/goleveldb/leveldb/util"
 )

-type deletionHandler func(t readWriteTransaction, folder, device, name []byte, dbi iterator.Iterator) int64
+type deletionHandler func(t readWriteTransaction, folder, device, name []byte, dbi iterator.Iterator)

 type Instance struct {
 	committed int64 // this must be the first attribute in the struct to ensure 64 bit alignment on 32 bit plaforms
@@ -86,7 +86,7 @@ func (db *Instance) Committed() int64 {
 	return atomic.LoadInt64(&db.committed)
 }

-func (db *Instance) genericReplace(folder, device []byte, fs []protocol.FileInfo, localSize, globalSize *sizeTracker, deleteFn deletionHandler) int64 {
+func (db *Instance) genericReplace(folder, device []byte, fs []protocol.FileInfo, localSize, globalSize *sizeTracker, deleteFn deletionHandler) {
 	sort.Sort(fileList(fs)) // sort list on name, same as in the database

 	t := db.newReadWriteTransaction()
@@ -97,7 +97,6 @@ func (db *Instance) genericReplace(folder, device []byte, fs []protocol.FileInfo

 	moreDb := dbi.Next()
 	fsi := 0
-	var maxLocalVer int64

 	isLocalDevice := bytes.Equal(device, protocol.LocalDeviceID[:])
 	for {
@@ -124,9 +123,7 @@ func (db *Instance) genericReplace(folder, device []byte, fs []protocol.FileInfo
 		case moreFs && (!moreDb || cmp == -1):
 			l.Debugln("generic replace; missing - insert")
 			// Database is missing this file. Insert it.
-			if lv := t.insertFile(folder, device, fs[fsi]); lv > maxLocalVer {
-				maxLocalVer = lv
-			}
+			t.insertFile(folder, device, fs[fsi])
 			if isLocalDevice {
 				localSize.addFile(fs[fsi])
 			}
@@ -146,9 +143,7 @@ func (db *Instance) genericReplace(folder, device []byte, fs []protocol.FileInfo
 			ef.Unmarshal(dbi.Value())
 			if !fs[fsi].Version.Equal(ef.Version) || fs[fsi].Invalid != ef.Invalid {
 				l.Debugln("generic replace; differs - insert")
-				if lv := t.insertFile(folder, device, fs[fsi]); lv > maxLocalVer {
-					maxLocalVer = lv
-				}
+				t.insertFile(folder, device, fs[fsi])
 				if isLocalDevice {
 					localSize.removeFile(ef)
 					localSize.addFile(fs[fsi])
@@ -167,9 +162,7 @@ func (db *Instance) genericReplace(folder, device []byte, fs []protocol.FileInfo

 		case moreDb && (!moreFs || cmp == 1):
 			l.Debugln("generic replace; exists - remove")
-			if lv := deleteFn(t, folder, device, oldName, dbi); lv > maxLocalVer {
-				maxLocalVer = lv
-			}
+			deleteFn(t, folder, device, oldName, dbi)
 			moreDb = dbi.Next()
 		}

@@ -177,26 +170,21 @@ func (db *Instance) genericReplace(folder, device []byte, fs []protocol.FileInfo
 		// growing too large and thus allocating unnecessarily much memory.
 		t.checkFlush()
 	}
-
-	return maxLocalVer
 }

-func (db *Instance) replace(folder, device []byte, fs []protocol.FileInfo, localSize, globalSize *sizeTracker) int64 {
-	// TODO: Return the remaining maxLocalVer?
-	return db.genericReplace(folder, device, fs, localSize, globalSize, func(t readWriteTransaction, folder, device, name []byte, dbi iterator.Iterator) int64 {
+func (db *Instance) replace(folder, device []byte, fs []protocol.FileInfo, localSize, globalSize *sizeTracker) {
+	db.genericReplace(folder, device, fs, localSize, globalSize, func(t readWriteTransaction, folder, device, name []byte, dbi iterator.Iterator) {
 		// Database has a file that we are missing. Remove it.
 		l.Debugf("delete; folder=%q device=%v name=%q", folder, protocol.DeviceIDFromBytes(device), name)
 		t.removeFromGlobal(folder, device, name, globalSize)
 		t.Delete(dbi.Key())
-		return 0
 	})
 }

-func (db *Instance) updateFiles(folder, device []byte, fs []protocol.FileInfo, localSize, globalSize *sizeTracker) int64 {
+func (db *Instance) updateFiles(folder, device []byte, fs []protocol.FileInfo, localSize, globalSize *sizeTracker) {
 	t := db.newReadWriteTransaction()
 	defer t.close()

-	var maxLocalVer int64
 	var fk []byte
 	isLocalDevice := bytes.Equal(device, protocol.LocalDeviceID[:])
 	for _, f := range fs {
@@ -208,9 +196,7 @@ func (db *Instance) updateFiles(folder, device []byte, fs []protocol.FileInfo, l
 				localSize.addFile(f)
 			}

-			if lv := t.insertFile(folder, device, f); lv > maxLocalVer {
-				maxLocalVer = lv
-			}
+			t.insertFile(folder, device, f)
 			if f.IsInvalid() {
 				t.removeFromGlobal(folder, device, name, globalSize)
 			} else {
@@ -231,9 +217,7 @@ func (db *Instance) updateFiles(folder, device []byte, fs []protocol.FileInfo, l
 				localSize.addFile(f)
 			}

-			if lv := t.insertFile(folder, device, f); lv > maxLocalVer {
-				maxLocalVer = lv
-			}
+			t.insertFile(folder, device, f)
 			if f.IsInvalid() {
 				t.removeFromGlobal(folder, device, name, globalSize)
 			} else {
@@ -245,8 +229,6 @@ func (db *Instance) updateFiles(folder, device []byte, fs []protocol.FileInfo, l
 		// growing too large and thus allocating unnecessarily much memory.
 		t.checkFlush()
 	}
-
-	return maxLocalVer
 }

 func (db *Instance) withHave(folder, device, prefix []byte, truncate bool, fn Iterator) {
@@ -699,6 +681,37 @@ func (db *Instance) globalKeyFolder(key []byte) []byte {
 	return folder
 }

+func (db *Instance) getIndexID(device, folder []byte) protocol.IndexID {
+	key := db.indexIDKey(device, folder)
+	cur, err := db.Get(key, nil)
+	if err != nil {
+		return 0
+	}
+
+	var id protocol.IndexID
+	if err := id.Unmarshal(cur); err != nil {
+		return 0
+	}
+
+	return id
+}
+
+func (db *Instance) setIndexID(device, folder []byte, id protocol.IndexID) {
+	key := db.indexIDKey(device, folder)
+	bs, _ := id.Marshal() // marshalling can't fail
+	if err := db.Put(key, bs, nil); err != nil {
+		panic("storing index ID: " + err.Error())
+	}
+}
+
+func (db *Instance) indexIDKey(device, folder []byte) []byte {
+	k := make([]byte, keyPrefixLen+keyDeviceLen+keyFolderLen)
+	k[0] = KeyTypeIndexID
+	binary.BigEndian.PutUint32(k[keyPrefixLen:], db.deviceIdx.ID(device))
+	binary.BigEndian.PutUint32(k[keyPrefixLen+keyDeviceLen:], db.folderIdx.ID(folder))
+	return k
+}
+
 func unmarshalTrunc(bs []byte, truncate bool) (FileIntf, error) {
 	if truncate {
 		var tf FileInfoTruncated
--- a/lib/db/leveldb_transactions.go
+++ b/lib/db/leveldb_transactions.go
@@ -74,18 +74,12 @@ func (t readWriteTransaction) flush() {
 	atomic.AddInt64(&t.db.committed, int64(t.Batch.Len()))
 }

-func (t readWriteTransaction) insertFile(folder, device []byte, file protocol.FileInfo) int64 {
+func (t readWriteTransaction) insertFile(folder, device []byte, file protocol.FileInfo) {
 	l.Debugf("insert; folder=%q device=%v %v", folder, protocol.DeviceIDFromBytes(device), file)

-	if file.LocalVersion == 0 {
-		file.LocalVersion = clock(0)
-	}
-
 	name := []byte(file.Name)
 	nk := t.db.deviceKey(folder, device, name)
 	t.Put(nk, mustMarshal(&file))
-
-	return file.LocalVersion
 }

 // updateGlobal adds this device+version to the version list for the given
--- a/lib/db/set.go
+++ b/lib/db/set.go
@@ -14,6 +14,7 @@ package db

 import (
 	stdsync "sync"
+	"sync/atomic"

 	"github.com/syncthing/syncthing/lib/osutil"
 	"github.com/syncthing/syncthing/lib/protocol"
@@ -21,13 +22,15 @@ import (
 )

 type FileSet struct {
-	localVersion map[protocol.DeviceID]int64
-	mutex        sync.Mutex
+	localVersion int64 // Our local version counter
 	folder       string
 	db           *Instance
 	blockmap     *BlockMap
 	localSize    sizeTracker
 	globalSize   sizeTracker
+
+	remoteLocalVersion map[protocol.DeviceID]int64 // Highest seen local versions for other devices
+	rlvMutex           sync.Mutex                  // protects remoteLocalVersion
 }

 // FileIntf is the set of methods implemented by both protocol.FileInfo and
@@ -95,11 +98,11 @@ func (s *sizeTracker) Size() (files, deleted int, bytes int64) {

 func NewFileSet(folder string, db *Instance) *FileSet {
 	var s = FileSet{
-		localVersion: make(map[protocol.DeviceID]int64),
-		folder:       folder,
-		db:           db,
-		blockmap:     NewBlockMap(db, db.folderIdx.ID([]byte(folder))),
-		mutex:        sync.NewMutex(),
+		remoteLocalVersion: make(map[protocol.DeviceID]int64),
+		folder:             folder,
+		db:                 db,
+		blockmap:           NewBlockMap(db, db.folderIdx.ID([]byte(folder))),
+		rlvMutex:           sync.NewMutex(),
 	}

 	s.db.checkGlobals([]byte(folder), &s.globalSize)
@@ -107,16 +110,17 @@ func NewFileSet(folder string, db *Instance) *FileSet {
 	var deviceID protocol.DeviceID
 	s.db.withAllFolderTruncated([]byte(folder), func(device []byte, f FileInfoTruncated) bool {
 		copy(deviceID[:], device)
-		if f.LocalVersion > s.localVersion[deviceID] {
-			s.localVersion[deviceID] = f.LocalVersion
-		}
 		if deviceID == protocol.LocalDeviceID {
+			if f.LocalVersion > s.localVersion {
+				s.localVersion = f.LocalVersion
+			}
 			s.localSize.addFile(f)
+		} else if f.LocalVersion > s.remoteLocalVersion[deviceID] {
+			s.remoteLocalVersion[deviceID] = f.LocalVersion
 		}
 		return true
 	})
 	l.Debugf("loaded localVersion for %q: %#v", folder, s.localVersion)
-	clock(s.localVersion[protocol.LocalDeviceID])

 	return &s
 }
@@ -124,13 +128,23 @@ func NewFileSet(folder string, db *Instance) *FileSet {
 func (s *FileSet) Replace(device protocol.DeviceID, fs []protocol.FileInfo) {
 	l.Debugf("%s Replace(%v, [%d])", s.folder, device, len(fs))
 	normalizeFilenames(fs)
-	s.mutex.Lock()
-	defer s.mutex.Unlock()
-	s.localVersion[device] = s.db.replace([]byte(s.folder), device[:], fs, &s.localSize, &s.globalSize)
-	if len(fs) == 0 {
-		// Reset the local version if all files were removed.
-		s.localVersion[device] = 0
+	if device == protocol.LocalDeviceID {
+		if len(fs) == 0 {
+			s.localVersion = 0
+		} else {
+			// Always overwrite LocalVersion on updated files to ensure
+			// correct ordering. The caller is supposed to leave it set to
+			// zero anyhow.
+			for i := range fs {
+				fs[i].LocalVersion = atomic.AddInt64(&s.localVersion, 1)
+			}
+		}
+	} else {
+		s.rlvMutex.Lock()
+		s.remoteLocalVersion[device] = maxLocalVersion(fs)
+		s.rlvMutex.Unlock()
 	}
+	s.db.replace([]byte(s.folder), device[:], fs, &s.localSize, &s.globalSize)
 	if device == protocol.LocalDeviceID {
 		s.blockmap.Drop()
 		s.blockmap.Add(fs)
@@ -140,12 +154,11 @@ func (s *FileSet) Replace(device protocol.DeviceID, fs []protocol.FileInfo) {
 func (s *FileSet) Update(device protocol.DeviceID, fs []protocol.FileInfo) {
 	l.Debugf("%s Update(%v, [%d])", s.folder, device, len(fs))
 	normalizeFilenames(fs)
-	s.mutex.Lock()
-	defer s.mutex.Unlock()
 	if device == protocol.LocalDeviceID {
 		discards := make([]protocol.FileInfo, 0, len(fs))
 		updates := make([]protocol.FileInfo, 0, len(fs))
-		for _, newFile := range fs {
+		for i, newFile := range fs {
+			fs[i].LocalVersion = atomic.AddInt64(&s.localVersion, 1)
 			existingFile, ok := s.db.getFile([]byte(s.folder), device[:], []byte(newFile.Name))
 			if !ok || !existingFile.Version.Equal(newFile.Version) {
 				discards = append(discards, existingFile)
@@ -154,10 +167,12 @@ func (s *FileSet) Update(device protocol.DeviceID, fs []protocol.FileInfo) {
 		}
 		s.blockmap.Discard(discards)
 		s.blockmap.Update(updates)
+	} else {
+		s.rlvMutex.Lock()
+		s.remoteLocalVersion[device] = maxLocalVersion(fs)
+		s.rlvMutex.Unlock()
 	}
-	if lv := s.db.updateFiles([]byte(s.folder), device[:], fs, &s.localSize, &s.globalSize); lv > s.localVersion[device] {
-		s.localVersion[device] = lv
-	}
+	s.db.updateFiles([]byte(s.folder), device[:], fs, &s.localSize, &s.globalSize)
 }

 func (s *FileSet) WithNeed(device protocol.DeviceID, fn Iterator) {
@@ -230,9 +245,13 @@ func (s *FileSet) Availability(file string) []protocol.DeviceID {
 }

 func (s *FileSet) LocalVersion(device protocol.DeviceID) int64 {
-	s.mutex.Lock()
-	defer s.mutex.Unlock()
-	return s.localVersion[device]
+	if device == protocol.LocalDeviceID {
+		return atomic.LoadInt64(&s.localVersion)
+	}
+
+	s.rlvMutex.Lock()
+	defer s.rlvMutex.Unlock()
+	return s.remoteLocalVersion[device]
 }

 func (s *FileSet) LocalSize() (files, deleted int, bytes int64) {
@@ -243,6 +262,37 @@ func (s *FileSet) GlobalSize() (files, deleted int, bytes int64) {
 	return s.globalSize.Size()
 }

+func (s *FileSet) IndexID(device protocol.DeviceID) protocol.IndexID {
+	id := s.db.getIndexID(device[:], []byte(s.folder))
+	if id == 0 && device == protocol.LocalDeviceID {
+		// No index ID set yet. We create one now.
+		id = protocol.NewIndexID()
+		s.db.setIndexID(device[:], []byte(s.folder), id)
+	}
+	return id
+}
+
+func (s *FileSet) SetIndexID(device protocol.DeviceID, id protocol.IndexID) {
+	if device == protocol.LocalDeviceID {
+		panic("do not explicitly set index ID for local device")
+	}
+	s.db.setIndexID(device[:], []byte(s.folder), id)
+}
+
+// maxLocalVersion returns the highest of the LocalVersion numbers found in
+// the given slice of FileInfos. This should really be the LocalVersion of
+// the last item, but Syncthing v0.14.0 and other implementations may not
+// implement update sorting....
+func maxLocalVersion(fs []protocol.FileInfo) int64 {
+	var max int64
+	for _, f := range fs {
+		if f.LocalVersion > max {
+			max = f.LocalVersion
+		}
+	}
+	return max
+}
+
 // DropFolder clears out all information related to the given folder from the
 // database.
 func DropFolder(db *Instance, folder string) {
--- a/lib/db/set_test.go
+++ b/lib/db/set_test.go
@@ -100,11 +100,11 @@ func TestGlobalSet(t *testing.T) {
 	m := db.NewFileSet("test", ldb)

 	local0 := fileList{
-		protocol.FileInfo{Name: "a", Version: protocol.Vector{Counters: []protocol.Counter{{ID: myID, Value: 1000}}}, Blocks: genBlocks(1)},
-		protocol.FileInfo{Name: "b", Version: protocol.Vector{Counters: []protocol.Counter{{ID: myID, Value: 1000}}}, Blocks: genBlocks(2)},
-		protocol.FileInfo{Name: "c", Version: protocol.Vector{Counters: []protocol.Counter{{ID: myID, Value: 1000}}}, Blocks: genBlocks(3)},
-		protocol.FileInfo{Name: "d", Version: protocol.Vector{Counters: []protocol.Counter{{ID: myID, Value: 1000}}}, Blocks: genBlocks(4)},
-		protocol.FileInfo{Name: "z", Version: protocol.Vector{Counters: []protocol.Counter{{ID: myID, Value: 1000}}}, Blocks: genBlocks(8)},
+		protocol.FileInfo{Name: "a", LocalVersion: 1, Version: protocol.Vector{Counters: []protocol.Counter{{ID: myID, Value: 1000}}}, Blocks: genBlocks(1)},
+		protocol.FileInfo{Name: "b", LocalVersion: 2, Version: protocol.Vector{Counters: []protocol.Counter{{ID: myID, Value: 1000}}}, Blocks: genBlocks(2)},
+		protocol.FileInfo{Name: "c", LocalVersion: 3, Version: protocol.Vector{Counters: []protocol.Counter{{ID: myID, Value: 1000}}}, Blocks: genBlocks(3)},
+		protocol.FileInfo{Name: "d", LocalVersion: 4, Version: protocol.Vector{Counters: []protocol.Counter{{ID: myID, Value: 1000}}}, Blocks: genBlocks(4)},
+		protocol.FileInfo{Name: "z", LocalVersion: 5, Version: protocol.Vector{Counters: []protocol.Counter{{ID: myID, Value: 1000}}}, Blocks: genBlocks(8)},
 	}
 	local1 := fileList{
 		protocol.FileInfo{Name: "a", Version: protocol.Vector{Counters: []protocol.Counter{{ID: myID, Value: 1000}}}, Blocks: genBlocks(1)},
@@ -687,3 +687,35 @@ func BenchmarkUpdateOneFile(b *testing.B) {

 	b.ReportAllocs()
 }
+
+func TestIndexID(t *testing.T) {
+	ldb := db.OpenMemory()
+
+	s := db.NewFileSet("test", ldb)
+
+	// The Index ID for some random device is zero by default.
+	id := s.IndexID(remoteDevice0)
+	if id != 0 {
+		t.Errorf("index ID for remote device should default to zero, not %d", id)
+	}
+
+	// The Index ID for someone else should be settable
+	s.SetIndexID(remoteDevice0, 42)
+	id = s.IndexID(remoteDevice0)
+	if id != 42 {
+		t.Errorf("index ID for remote device should be remembered; got %d, expected %d", id, 42)
+	}
+
+	// Our own index ID should be generated randomly.
+	id = s.IndexID(protocol.LocalDeviceID)
+	if id == 0 {
+		t.Errorf("index ID for local device should be random, not zero")
+	}
+	t.Logf("random index ID is 0x%016x", id)
+
+	// But of course always the same after that.
+	again := s.IndexID(protocol.LocalDeviceID)
+	if again != id {
+		t.Errorf("index ID changed; %d != %d", again, id)
+	}
+}