diff --git a/lib/db/leveldb.go b/lib/db/leveldb.go index a4923875..151f3ff3 100644 --- a/lib/db/leveldb.go +++ b/lib/db/leveldb.go @@ -169,7 +169,7 @@ func globalKeyFolder(key []byte) []byte { type deletionHandler func(db dbReader, batch dbWriter, folder, device, name []byte, dbi iterator.Iterator) int64 -func ldbGenericReplace(db *leveldb.DB, folder, device []byte, fs []protocol.FileInfo, deleteFn deletionHandler) int64 { +func ldbGenericReplace(db *leveldb.DB, folder, device []byte, fs []protocol.FileInfo, localSize, globalSize *sizeTracker, deleteFn deletionHandler) int64 { runtime.GC() sort.Sort(fileList(fs)) // sort list on name, same as in the database @@ -196,6 +196,7 @@ func ldbGenericReplace(db *leveldb.DB, folder, device []byte, fs []protocol.File fsi := 0 var maxLocalVer int64 + isLocalDevice := bytes.Equal(device, protocol.LocalDeviceID[:]) for { var newName, oldName []byte moreFs := fsi < len(fs) @@ -223,10 +224,13 @@ func ldbGenericReplace(db *leveldb.DB, folder, device []byte, fs []protocol.File if lv := ldbInsert(batch, folder, device, fs[fsi]); lv > maxLocalVer { maxLocalVer = lv } + if isLocalDevice { + localSize.addFile(fs[fsi]) + } if fs[fsi].IsInvalid() { - ldbRemoveFromGlobal(snap, batch, folder, device, newName) + ldbRemoveFromGlobal(snap, batch, folder, device, newName, globalSize) } else { - ldbUpdateGlobal(snap, batch, folder, device, fs[fsi]) + ldbUpdateGlobal(snap, batch, folder, device, fs[fsi], globalSize) } fsi++ @@ -242,10 +246,14 @@ func ldbGenericReplace(db *leveldb.DB, folder, device []byte, fs []protocol.File if lv := ldbInsert(batch, folder, device, fs[fsi]); lv > maxLocalVer { maxLocalVer = lv } + if isLocalDevice { + localSize.removeFile(ef) + localSize.addFile(fs[fsi]) + } if fs[fsi].IsInvalid() { - ldbRemoveFromGlobal(snap, batch, folder, device, newName) + ldbRemoveFromGlobal(snap, batch, folder, device, newName, globalSize) } else { - ldbUpdateGlobal(snap, batch, folder, device, fs[fsi]) + ldbUpdateGlobal(snap, batch, folder, device, fs[fsi], globalSize) } } else { l.Debugln("generic replace; equal - ignore") @@ -285,19 +293,19 @@ func ldbGenericReplace(db *leveldb.DB, folder, device []byte, fs []protocol.File return maxLocalVer } -func ldbReplace(db *leveldb.DB, folder, device []byte, fs []protocol.FileInfo) int64 { +func ldbReplace(db *leveldb.DB, folder, device []byte, fs []protocol.FileInfo, localSize, globalSize *sizeTracker) int64 { // TODO: Return the remaining maxLocalVer? - return ldbGenericReplace(db, folder, device, fs, func(db dbReader, batch dbWriter, folder, device, name []byte, dbi iterator.Iterator) int64 { + return ldbGenericReplace(db, folder, device, fs, localSize, globalSize, func(db dbReader, batch dbWriter, folder, device, name []byte, dbi iterator.Iterator) int64 { // Database has a file that we are missing. Remove it. l.Debugf("delete; folder=%q device=%v name=%q", folder, protocol.DeviceIDFromBytes(device), name) - ldbRemoveFromGlobal(db, batch, folder, device, name) + ldbRemoveFromGlobal(db, batch, folder, device, name, globalSize) l.Debugf("batch.Delete %p %x", batch, dbi.Key()) batch.Delete(dbi.Key()) return 0 }) } -func ldbUpdate(db *leveldb.DB, folder, device []byte, fs []protocol.FileInfo) int64 { +func ldbUpdate(db *leveldb.DB, folder, device []byte, fs []protocol.FileInfo, localSize, globalSize *sizeTracker) int64 { runtime.GC() batch := new(leveldb.Batch) @@ -314,19 +322,24 @@ func ldbUpdate(db *leveldb.DB, folder, device []byte, fs []protocol.FileInfo) in var maxLocalVer int64 var fk []byte + isLocalDevice := bytes.Equal(device, protocol.LocalDeviceID[:]) for _, f := range fs { name := []byte(f.Name) fk = deviceKeyInto(fk[:cap(fk)], folder, device, name) l.Debugf("snap.Get %p %x", snap, fk) bs, err := snap.Get(fk, nil) if err == leveldb.ErrNotFound { + if isLocalDevice { + localSize.addFile(f) + } + if lv := ldbInsert(batch, folder, device, f); lv > maxLocalVer { maxLocalVer = lv } if f.IsInvalid() { - ldbRemoveFromGlobal(snap, batch, folder, device, name) + ldbRemoveFromGlobal(snap, batch, folder, device, name, globalSize) } else { - ldbUpdateGlobal(snap, batch, folder, device, f) + ldbUpdateGlobal(snap, batch, folder, device, f, globalSize) } continue } @@ -339,13 +352,18 @@ func ldbUpdate(db *leveldb.DB, folder, device []byte, fs []protocol.FileInfo) in // Flags might change without the version being bumped when we set the // invalid flag on an existing file. if !ef.Version.Equal(f.Version) || ef.Flags != f.Flags { + if isLocalDevice { + localSize.removeFile(ef) + localSize.addFile(f) + } + if lv := ldbInsert(batch, folder, device, f); lv > maxLocalVer { maxLocalVer = lv } if f.IsInvalid() { - ldbRemoveFromGlobal(snap, batch, folder, device, name) + ldbRemoveFromGlobal(snap, batch, folder, device, name, globalSize) } else { - ldbUpdateGlobal(snap, batch, folder, device, f) + ldbUpdateGlobal(snap, batch, folder, device, f, globalSize) } } @@ -390,7 +408,7 @@ func ldbInsert(batch dbWriter, folder, device []byte, file protocol.FileInfo) in // ldbUpdateGlobal adds this device+version to the version list for the given // file. If the device is already present in the list, the version is updated. // If the file does not have an entry in the global list, it is created. -func ldbUpdateGlobal(db dbReader, batch dbWriter, folder, device []byte, file protocol.FileInfo) bool { +func ldbUpdateGlobal(db dbReader, batch dbWriter, folder, device []byte, file protocol.FileInfo, globalSize *sizeTracker) bool { l.Debugf("update global; folder=%q device=%v file=%q version=%d", folder, protocol.DeviceIDFromBytes(device), file.Name, file.Version) name := []byte(file.Name) gk := globalKey(folder, name) @@ -400,7 +418,8 @@ func ldbUpdateGlobal(db dbReader, batch dbWriter, folder, device []byte, file pr } var fl versionList - + var oldFile protocol.FileInfo + var hasOldFile bool // Remove the device from the current version list if svl != nil { err = fl.UnmarshalXDR(svl) @@ -414,6 +433,13 @@ func ldbUpdateGlobal(db dbReader, batch dbWriter, folder, device []byte, file pr // No need to do anything return false } + + if i == 0 { + // Keep the current newest file around so we can subtract it from + // the globalSize if we replace it. + oldFile, hasOldFile = ldbGet(db, folder, fl.versions[0].device, name) + } + fl.versions = append(fl.versions[:i], fl.versions[i+1:]...) break } @@ -425,6 +451,7 @@ func ldbUpdateGlobal(db dbReader, batch dbWriter, folder, device []byte, file pr version: file.Version, } + insertedAt := -1 // Find a position in the list to insert this file. The file at the front // of the list is the newer, the "global". for i := range fl.versions { @@ -433,6 +460,7 @@ func ldbUpdateGlobal(db dbReader, batch dbWriter, folder, device []byte, file pr // The version at this point in the list is equal to or lesser // ("older") than us. We insert ourselves in front of it. fl.versions = insertVersion(fl.versions, i, nv) + insertedAt = i goto done case protocol.ConcurrentLesser, protocol.ConcurrentGreater: @@ -448,6 +476,7 @@ func ldbUpdateGlobal(db dbReader, batch dbWriter, folder, device []byte, file pr } if file.WinsConflict(of) { fl.versions = insertVersion(fl.versions, i, nv) + insertedAt = i goto done } } @@ -455,8 +484,28 @@ func ldbUpdateGlobal(db dbReader, batch dbWriter, folder, device []byte, file pr // We didn't find a position for an insert above, so append to the end. fl.versions = append(fl.versions, nv) + insertedAt = len(fl.versions) - 1 done: + if insertedAt == 0 { + // We just inserted a new newest version. Fixup the global size + // calculation. + if !file.Version.Equal(oldFile.Version) { + globalSize.addFile(file) + if hasOldFile { + // We have the old file that was removed at the head of the list. + globalSize.removeFile(oldFile) + } else if len(fl.versions) > 1 { + // The previous newest version is now at index 1, grab it from there. + oldFile, ok := ldbGet(db, folder, fl.versions[1].device, name) + if !ok { + panic("file referenced in version list does not exist") + } + globalSize.removeFile(oldFile) + } + } + } + l.Debugf("batch.Put %p %x", batch, gk) l.Debugf("new global after update: %v", fl) batch.Put(gk, fl.MustMarshalXDR()) @@ -474,7 +523,7 @@ func insertVersion(vl []fileVersion, i int, v fileVersion) []fileVersion { // ldbRemoveFromGlobal removes the device from the global version list for the // given file. If the version list is empty after this, the file entry is // removed entirely. -func ldbRemoveFromGlobal(db dbReader, batch dbWriter, folder, device, file []byte) { +func ldbRemoveFromGlobal(db dbReader, batch dbWriter, folder, device, file []byte, globalSize *sizeTracker) { l.Debugf("remove from global; folder=%q device=%v file=%q", folder, protocol.DeviceIDFromBytes(device), file) gk := globalKey(folder, file) @@ -491,8 +540,17 @@ func ldbRemoveFromGlobal(db dbReader, batch dbWriter, folder, device, file []byt panic(err) } + removed := false for i := range fl.versions { if bytes.Compare(fl.versions[i].device, device) == 0 { + if i == 0 && globalSize != nil { + f, ok := ldbGet(db, folder, device, file) + if !ok { + panic("removing nonexistent file") + } + globalSize.removeFile(f) + removed = true + } fl.versions = append(fl.versions[:i], fl.versions[i+1:]...) break } @@ -505,6 +563,13 @@ func ldbRemoveFromGlobal(db dbReader, batch dbWriter, folder, device, file []byt l.Debugf("batch.Put %p %x", batch, gk) l.Debugf("new global after remove: %v", fl) batch.Put(gk, fl.MustMarshalXDR()) + if removed { + f, ok := ldbGet(db, folder, fl.versions[0].device, file) + if !ok { + panic("new global is nonexistent file") + } + globalSize.addFile(f) + } } } @@ -565,7 +630,7 @@ func ldbWithAllFolderTruncated(db *leveldb.DB, folder []byte, fn func(device []b case "", ".", "..", "/": // A few obviously invalid filenames l.Infof("Dropping invalid filename %q from database", f.Name) batch := new(leveldb.Batch) - ldbRemoveFromGlobal(db, batch, folder, device, nil) + ldbRemoveFromGlobal(db, batch, folder, device, nil, nil) batch.Delete(dbi.Key()) db.Write(batch, nil) continue @@ -900,7 +965,7 @@ func unmarshalTrunc(bs []byte, truncate bool) (FileIntf, error) { return tf, err } -func ldbCheckGlobals(db *leveldb.DB, folder []byte) { +func ldbCheckGlobals(db *leveldb.DB, folder []byte, globalSize *sizeTracker) { defer runtime.GC() snap, err := db.GetSnapshot() @@ -937,8 +1002,9 @@ func ldbCheckGlobals(db *leveldb.DB, folder []byte) { name := globalKeyName(gk) var newVL versionList - for _, version := range vl.versions { + for i, version := range vl.versions { fk = deviceKeyInto(fk[:cap(fk)], folder, version.device, name) + l.Debugf("snap.Get %p %x", snap, fk) _, err := snap.Get(fk, nil) if err == leveldb.ErrNotFound { @@ -948,6 +1014,14 @@ func ldbCheckGlobals(db *leveldb.DB, folder []byte) { panic(err) } newVL.versions = append(newVL.versions, version) + + if i == 0 { + fi, ok := ldbGet(snap, folder, version.device, name) + if !ok { + panic("nonexistent global master file") + } + globalSize.addFile(fi) + } } if len(newVL.versions) != len(vl.versions) { diff --git a/lib/db/set.go b/lib/db/set.go index 140b1611..bdc4e33d 100644 --- a/lib/db/set.go +++ b/lib/db/set.go @@ -13,6 +13,8 @@ package db import ( + stdsync "sync" + "github.com/syncthing/syncthing/lib/osutil" "github.com/syncthing/syncthing/lib/protocol" "github.com/syncthing/syncthing/lib/sync" @@ -25,6 +27,8 @@ type FileSet struct { folder string db *leveldb.DB blockmap *BlockMap + localSize sizeTracker + globalSize sizeTracker } // FileIntf is the set of methods implemented by both protocol.FileInfo and @@ -43,6 +47,52 @@ type FileIntf interface { // continue iteration, false to stop. type Iterator func(f FileIntf) bool +type sizeTracker struct { + files int + deleted int + bytes int64 + mut stdsync.Mutex +} + +func (s *sizeTracker) addFile(f FileIntf) { + if f.IsInvalid() { + return + } + + s.mut.Lock() + if f.IsDeleted() { + s.deleted++ + } else { + s.files++ + } + s.bytes += f.Size() + s.mut.Unlock() +} + +func (s *sizeTracker) removeFile(f FileIntf) { + if f.IsInvalid() { + return + } + + s.mut.Lock() + if f.IsDeleted() { + s.deleted-- + } else { + s.files-- + } + s.bytes -= f.Size() + if s.deleted < 0 || s.files < 0 { + panic("bug: removed more than added") + } + s.mut.Unlock() +} + +func (s *sizeTracker) Size() (files, deleted int, bytes int64) { + s.mut.Lock() + defer s.mut.Unlock() + return s.files, s.deleted, s.bytes +} + func NewFileSet(folder string, db *leveldb.DB) *FileSet { var s = FileSet{ localVersion: make(map[protocol.DeviceID]int64), @@ -52,7 +102,7 @@ func NewFileSet(folder string, db *leveldb.DB) *FileSet { mutex: sync.NewMutex(), } - ldbCheckGlobals(db, []byte(folder)) + ldbCheckGlobals(db, []byte(folder), &s.globalSize) var deviceID protocol.DeviceID ldbWithAllFolderTruncated(db, []byte(folder), func(device []byte, f FileInfoTruncated) bool { @@ -60,6 +110,9 @@ func NewFileSet(folder string, db *leveldb.DB) *FileSet { if f.LocalVersion > s.localVersion[deviceID] { s.localVersion[deviceID] = f.LocalVersion } + if deviceID == protocol.LocalDeviceID { + s.localSize.addFile(f) + } return true }) l.Debugf("loaded localVersion for %q: %#v", folder, s.localVersion) @@ -73,7 +126,7 @@ func (s *FileSet) Replace(device protocol.DeviceID, fs []protocol.FileInfo) { normalizeFilenames(fs) s.mutex.Lock() defer s.mutex.Unlock() - s.localVersion[device] = ldbReplace(s.db, []byte(s.folder), device[:], fs) + s.localVersion[device] = ldbReplace(s.db, []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 @@ -102,7 +155,7 @@ func (s *FileSet) Update(device protocol.DeviceID, fs []protocol.FileInfo) { s.blockmap.Discard(discards) s.blockmap.Update(updates) } - if lv := ldbUpdate(s.db, []byte(s.folder), device[:], fs); lv > s.localVersion[device] { + if lv := ldbUpdate(s.db, []byte(s.folder), device[:], fs, &s.localSize, &s.globalSize); lv > s.localVersion[device] { s.localVersion[device] = lv } } @@ -178,6 +231,14 @@ func (s *FileSet) LocalVersion(device protocol.DeviceID) int64 { return s.localVersion[device] } +func (s *FileSet) LocalSize() (files, deleted int, bytes int64) { + return s.localSize.Size() +} + +func (s *FileSet) GlobalSize() (files, deleted int, bytes int64) { + return s.globalSize.Size() +} + // ListFolders returns the folder IDs seen in the database. func ListFolders(db *leveldb.DB) []string { return ldbListFolders(db) diff --git a/lib/db/set_test.go b/lib/db/set_test.go index abe355c4..64e12603 100644 --- a/lib/db/set_test.go +++ b/lib/db/set_test.go @@ -173,6 +173,29 @@ func TestGlobalSet(t *testing.T) { t.Errorf("Global incorrect;\n A: %v !=\n E: %v", g, expectedGlobal) } + globalFiles, globalDeleted, globalBytes := 0, 0, int64(0) + for _, f := range g { + if f.IsInvalid() { + continue + } + if f.IsDeleted() { + globalDeleted++ + } else { + globalFiles++ + } + globalBytes += f.Size() + } + gsFiles, gsDeleted, gsBytes := m.GlobalSize() + if gsFiles != globalFiles { + t.Errorf("Incorrect GlobalSize files; %d != %d", gsFiles, globalFiles) + } + if gsDeleted != globalDeleted { + t.Errorf("Incorrect GlobalSize deleted; %d != %d", gsDeleted, globalDeleted) + } + if gsBytes != globalBytes { + t.Errorf("Incorrect GlobalSize bytes; %d != %d", gsBytes, globalBytes) + } + h := fileList(haveList(m, protocol.LocalDeviceID)) sort.Sort(h) @@ -180,6 +203,29 @@ func TestGlobalSet(t *testing.T) { t.Errorf("Have incorrect;\n A: %v !=\n E: %v", h, localTot) } + haveFiles, haveDeleted, haveBytes := 0, 0, int64(0) + for _, f := range h { + if f.IsInvalid() { + continue + } + if f.IsDeleted() { + haveDeleted++ + } else { + haveFiles++ + } + haveBytes += f.Size() + } + lsFiles, lsDeleted, lsBytes := m.LocalSize() + if lsFiles != haveFiles { + t.Errorf("Incorrect LocalSize files; %d != %d", lsFiles, haveFiles) + } + if lsDeleted != haveDeleted { + t.Errorf("Incorrect LocalSize deleted; %d != %d", lsDeleted, haveDeleted) + } + if lsBytes != haveBytes { + t.Errorf("Incorrect LocalSize bytes; %d != %d", lsBytes, haveBytes) + } + h = fileList(haveList(m, remoteDevice0)) sort.Sort(h) diff --git a/lib/model/model.go b/lib/model/model.go index b6b12c17..2e54a9ab 100644 --- a/lib/model/model.go +++ b/lib/model/model.go @@ -329,8 +329,6 @@ func (m *Model) FolderStatistics() map[string]stats.FolderStatistics { // Completion returns the completion status, in percent, for the given device // and folder. func (m *Model) Completion(device protocol.DeviceID, folder string) float64 { - var tot int64 - m.fmut.RLock() rf, ok := m.folderFiles[folder] m.fmut.RUnlock() @@ -338,29 +336,22 @@ func (m *Model) Completion(device protocol.DeviceID, folder string) float64 { return 0 // Folder doesn't exist, so we hardly have any of it } - rf.WithGlobalTruncated(func(f db.FileIntf) bool { - if !f.IsDeleted() { - tot += f.Size() - } - return true - }) - + _, _, tot := rf.GlobalSize() if tot == 0 { return 100 // Folder is empty, so we have all of it } var need int64 rf.WithNeedTruncated(device, func(f db.FileIntf) bool { - if !f.IsDeleted() { - need += f.Size() - } + need += f.Size() return true }) - res := 100 * (1 - float64(need)/float64(tot)) - l.Debugf("%v Completion(%s, %q): %f (%d / %d)", m, device, folder, res, need, tot) + needRatio := float64(need) / float64(tot) + completionPct := 100 * (1 - needRatio) + l.Debugf("%v Completion(%s, %q): %f (%d / %d = %f)", m, device, folder, completionPct, need, tot, needRatio) - return res + return completionPct } func sizeOf(fs []protocol.FileInfo) (files, deleted int, bytes int64) { @@ -389,13 +380,7 @@ func (m *Model) GlobalSize(folder string) (nfiles, deleted int, bytes int64) { m.fmut.RLock() defer m.fmut.RUnlock() if rf, ok := m.folderFiles[folder]; ok { - rf.WithGlobalTruncated(func(f db.FileIntf) bool { - fs, de, by := sizeOfFile(f) - nfiles += fs - deleted += de - bytes += by - return true - }) + nfiles, deleted, bytes = rf.GlobalSize() } return } @@ -406,16 +391,7 @@ func (m *Model) LocalSize(folder string) (nfiles, deleted int, bytes int64) { m.fmut.RLock() defer m.fmut.RUnlock() if rf, ok := m.folderFiles[folder]; ok { - rf.WithHaveTruncated(protocol.LocalDeviceID, func(f db.FileIntf) bool { - if f.IsInvalid() { - return true - } - fs, de, by := sizeOfFile(f) - nfiles += fs - deleted += de - bytes += by - return true - }) + nfiles, deleted, bytes = rf.LocalSize() } return }