Martchus
/
syncthing
1
0
Fork 0
Code Issues Pull Requests Releases Wiki Activity

Merge pull request #2415 from calmh/dbkeys 

Add database and transaction instances
This commit is contained in:
Audrius Butkevicius 2015-10-28 21:50:29 +00:00
parent a52811dfa3 fd7b8ec77e
commit eab1d6782b
2 changed files with 173 additions and 231 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

362
lib/db/leveldb.go
View File

@ -89,9 +89,18 @@ type dbReader interface {
Get([]byte, *opt.ReadOptions) ([]byte, error) Get([]byte, *opt.ReadOptions) ([]byte, error)
} }
type dbWriter interface { type dbInstance struct {
Put([]byte, []byte) *leveldb.DB
Delete([]byte) }
type readOnlyTransaction struct {
*leveldb.Snapshot
db *leveldb.DB
}
type readWriteTransaction struct {
readOnlyTransaction
*leveldb.Batch
} }
// Flush batches to disk when they contain this many records. // Flush batches to disk when they contain this many records.
@ -166,27 +175,57 @@ func globalKeyFolder(key []byte) []byte {
return folder[:izero] return folder[:izero]
} }
type deletionHandler func(db dbReader, batch dbWriter, folder, device, name []byte, dbi iterator.Iterator) int64 type deletionHandler func(t readWriteTransaction, folder, device, name []byte, dbi iterator.Iterator) int64
func ldbGenericReplace(db *leveldb.DB, folder, device []byte, fs []protocol.FileInfo, localSize, globalSize *sizeTracker, deleteFn deletionHandler) int64 { func (db *dbInstance) newReadOnlyTransaction() readOnlyTransaction {
snap, err := db.GetSnapshot()
if err != nil {
panic(err)
}
return readOnlyTransaction{
Snapshot: snap,
db: db.DB,
}
}
func (db *dbInstance) newReadWriteTransaction() readWriteTransaction {
t := db.newReadOnlyTransaction()
return readWriteTransaction{
readOnlyTransaction: t,
Batch: new(leveldb.Batch),
}
}
func (t readOnlyTransaction) close() {
t.Release()
}
func (t readWriteTransaction) close() {
if err := t.db.Write(t.Batch, nil); err != nil {
panic(err)
}
t.readOnlyTransaction.close()
}
func (t readWriteTransaction) checkFlush() {
if t.Batch.Len() > batchFlushSize {
if err := t.db.Write(t.Batch, nil); err != nil {
panic(err)
}
t.Batch.Reset()
}
}
func (db *dbInstance) genericReplace(folder, device []byte, fs []protocol.FileInfo, localSize, globalSize *sizeTracker, deleteFn deletionHandler) int64 {
sort.Sort(fileList(fs)) // sort list on name, same as in the database sort.Sort(fileList(fs)) // sort list on name, same as in the database
start := deviceKey(folder, device, nil) // before all folder/device files start := deviceKey(folder, device, nil) // before all folder/device files
limit := deviceKey(folder, device, []byte{0xff, 0xff, 0xff, 0xff}) // after all folder/device files limit := deviceKey(folder, device, []byte{0xff, 0xff, 0xff, 0xff}) // after all folder/device files
batch := new(leveldb.Batch) t := db.newReadWriteTransaction()
l.Debugf("new batch %p", batch) defer t.close()
snap, err := db.GetSnapshot()
if err != nil {
panic(err)
}
l.Debugf("created snapshot %p", snap)
defer func() {
l.Debugf("close snapshot %p", snap)
snap.Release()
}()
dbi := snap.NewIterator(&util.Range{Start: start, Limit: limit}, nil) dbi := t.NewIterator(&util.Range{Start: start, Limit: limit}, nil)
defer dbi.Release() defer dbi.Release()
moreDb := dbi.Next() moreDb := dbi.Next()
@ -218,16 +257,16 @@ func ldbGenericReplace(db *leveldb.DB, folder, device []byte, fs []protocol.File
case moreFs && (!moreDb || cmp == -1): case moreFs && (!moreDb || cmp == -1):
l.Debugln("generic replace; missing - insert") l.Debugln("generic replace; missing - insert")
// Database is missing this file. Insert it. // Database is missing this file. Insert it.
if lv := ldbInsert(batch, folder, device, fs[fsi]); lv > maxLocalVer { if lv := t.insertFile(folder, device, fs[fsi]); lv > maxLocalVer {
maxLocalVer = lv maxLocalVer = lv
} }
if isLocalDevice { if isLocalDevice {
localSize.addFile(fs[fsi]) localSize.addFile(fs[fsi])
} }
if fs[fsi].IsInvalid() { if fs[fsi].IsInvalid() {
ldbRemoveFromGlobal(snap, batch, folder, device, newName, globalSize) t.removeFromGlobal(folder, device, newName, globalSize)
} else { } else {
ldbUpdateGlobal(snap, batch, folder, device, fs[fsi], globalSize) t.updateGlobal(folder, device, fs[fsi], globalSize)
} }
fsi++ fsi++
@ -240,7 +279,7 @@ func ldbGenericReplace(db *leveldb.DB, folder, device []byte, fs []protocol.File
ef.UnmarshalXDR(dbi.Value()) ef.UnmarshalXDR(dbi.Value())
if !fs[fsi].Version.Equal(ef.Version) || fs[fsi].Flags != ef.Flags { if !fs[fsi].Version.Equal(ef.Version) || fs[fsi].Flags != ef.Flags {
l.Debugln("generic replace; differs - insert") l.Debugln("generic replace; differs - insert")
if lv := ldbInsert(batch, folder, device, fs[fsi]); lv > maxLocalVer { if lv := t.insertFile(folder, device, fs[fsi]); lv > maxLocalVer {
maxLocalVer = lv maxLocalVer = lv
} }
if isLocalDevice { if isLocalDevice {
@ -248,9 +287,9 @@ func ldbGenericReplace(db *leveldb.DB, folder, device []byte, fs []protocol.File
localSize.addFile(fs[fsi]) localSize.addFile(fs[fsi])
} }
if fs[fsi].IsInvalid() { if fs[fsi].IsInvalid() {
ldbRemoveFromGlobal(snap, batch, folder, device, newName, globalSize) t.removeFromGlobal(folder, device, newName, globalSize)
} else { } else {
ldbUpdateGlobal(snap, batch, folder, device, fs[fsi], globalSize) t.updateGlobal(folder, device, fs[fsi], globalSize)
} }
} else { } else {
l.Debugln("generic replace; equal - ignore") l.Debugln("generic replace; equal - ignore")
@ -261,7 +300,7 @@ func ldbGenericReplace(db *leveldb.DB, folder, device []byte, fs []protocol.File
case moreDb && (!moreFs || cmp == 1): case moreDb && (!moreFs || cmp == 1):
l.Debugln("generic replace; exists - remove") l.Debugln("generic replace; exists - remove")
if lv := deleteFn(snap, batch, folder, device, oldName, dbi); lv > maxLocalVer { if lv := deleteFn(t, folder, device, oldName, dbi); lv > maxLocalVer {
maxLocalVer = lv maxLocalVer = lv
} }
moreDb = dbi.Next() moreDb = dbi.Next()
@ -269,51 +308,26 @@ func ldbGenericReplace(db *leveldb.DB, folder, device []byte, fs []protocol.File
// Write out and reuse the batch every few records, to avoid the batch // Write out and reuse the batch every few records, to avoid the batch
// growing too large and thus allocating unnecessarily much memory. // growing too large and thus allocating unnecessarily much memory.
if batch.Len() > batchFlushSize { t.checkFlush()
l.Debugf("db.Write %p", batch)
err = db.Write(batch, nil)
if err != nil {
panic(err)
}
batch.Reset()
}
}
l.Debugf("db.Write %p", batch)
err = db.Write(batch, nil)
if err != nil {
panic(err)
} }
return maxLocalVer return maxLocalVer
} }
func ldbReplace(db *leveldb.DB, folder, device []byte, fs []protocol.FileInfo, localSize, globalSize *sizeTracker) int64 { func (db *dbInstance) replace(folder, device []byte, fs []protocol.FileInfo, localSize, globalSize *sizeTracker) int64 {
// TODO: Return the remaining maxLocalVer? // TODO: Return the remaining maxLocalVer?
return ldbGenericReplace(db, folder, device, fs, localSize, globalSize, func(db dbReader, batch dbWriter, folder, device, name []byte, dbi iterator.Iterator) int64 { return db.genericReplace(folder, device, fs, localSize, globalSize, func(t readWriteTransaction, folder, device, name []byte, dbi iterator.Iterator) int64 {
// Database has a file that we are missing. Remove it. // Database has a file that we are missing. Remove it.
l.Debugf("delete; folder=%q device=%v name=%q", folder, protocol.DeviceIDFromBytes(device), name) l.Debugf("delete; folder=%q device=%v name=%q", folder, protocol.DeviceIDFromBytes(device), name)
ldbRemoveFromGlobal(db, batch, folder, device, name, globalSize) t.removeFromGlobal(folder, device, name, globalSize)
l.Debugf("batch.Delete %p %x", batch, dbi.Key()) t.Delete(dbi.Key())
batch.Delete(dbi.Key())
return 0 return 0
}) })
} }
func ldbUpdate(db *leveldb.DB, folder, device []byte, fs []protocol.FileInfo, localSize, globalSize *sizeTracker) int64 { func (db *dbInstance) updateFiles(folder, device []byte, fs []protocol.FileInfo, localSize, globalSize *sizeTracker) int64 {
batch := new(leveldb.Batch) t := db.newReadWriteTransaction()
l.Debugf("new batch %p", batch) defer t.close()
snap, err := db.GetSnapshot()
if err != nil {
panic(err)
}
l.Debugf("created snapshot %p", snap)
defer func() {
l.Debugf("close snapshot %p", snap)
snap.Release()
}()
var maxLocalVer int64 var maxLocalVer int64
var fk []byte var fk []byte
@ -321,20 +335,19 @@ func ldbUpdate(db *leveldb.DB, folder, device []byte, fs []protocol.FileInfo, lo
for _, f := range fs { for _, f := range fs {
name := []byte(f.Name) name := []byte(f.Name)
fk = deviceKeyInto(fk[:cap(fk)], folder, device, name) fk = deviceKeyInto(fk[:cap(fk)], folder, device, name)
l.Debugf("snap.Get %p %x", snap, fk) bs, err := t.Get(fk, nil)
bs, err := snap.Get(fk, nil)
if err == leveldb.ErrNotFound { if err == leveldb.ErrNotFound {
if isLocalDevice { if isLocalDevice {
localSize.addFile(f) localSize.addFile(f)
} }
if lv := ldbInsert(batch, folder, device, f); lv > maxLocalVer { if lv := t.insertFile(folder, device, f); lv > maxLocalVer {
maxLocalVer = lv maxLocalVer = lv
} }
if f.IsInvalid() { if f.IsInvalid() {
ldbRemoveFromGlobal(snap, batch, folder, device, name, globalSize) t.removeFromGlobal(folder, device, name, globalSize)
} else { } else {
ldbUpdateGlobal(snap, batch, folder, device, f, globalSize) t.updateGlobal(folder, device, f, globalSize)
} }
continue continue
} }
@ -352,40 +365,25 @@ func ldbUpdate(db *leveldb.DB, folder, device []byte, fs []protocol.FileInfo, lo
localSize.addFile(f) localSize.addFile(f)
} }
if lv := ldbInsert(batch, folder, device, f); lv > maxLocalVer { if lv := t.insertFile(folder, device, f); lv > maxLocalVer {
maxLocalVer = lv maxLocalVer = lv
} }
if f.IsInvalid() { if f.IsInvalid() {
ldbRemoveFromGlobal(snap, batch, folder, device, name, globalSize) t.removeFromGlobal(folder, device, name, globalSize)
} else { } else {
ldbUpdateGlobal(snap, batch, folder, device, f, globalSize) t.updateGlobal(folder, device, f, globalSize)
} }
} }
// Write out and reuse the batch every few records, to avoid the batch // Write out and reuse the batch every few records, to avoid the batch
// growing too large and thus allocating unnecessarily much memory. // growing too large and thus allocating unnecessarily much memory.
if batch.Len() > batchFlushSize { t.checkFlush()
l.Debugf("db.Write %p", batch)
err = db.Write(batch, nil)
if err != nil {
panic(err)
}
batch.Reset()
}
}
l.Debugf("db.Write %p", batch)
err = db.Write(batch, nil)
if err != nil {
panic(err)
} }
return maxLocalVer return maxLocalVer
} }
func ldbInsert(batch dbWriter, folder, device []byte, file protocol.FileInfo) int64 { func (t readWriteTransaction) insertFile(folder, device []byte, file protocol.FileInfo) int64 {
l.Debugf("insert; folder=%q device=%v %v", folder, protocol.DeviceIDFromBytes(device), file) l.Debugf("insert; folder=%q device=%v %v", folder, protocol.DeviceIDFromBytes(device), file)
if file.LocalVersion == 0 { if file.LocalVersion == 0 {
@ -394,20 +392,19 @@ func ldbInsert(batch dbWriter, folder, device []byte, file protocol.FileInfo) in
name := []byte(file.Name) name := []byte(file.Name)
nk := deviceKey(folder, device, name) nk := deviceKey(folder, device, name)
l.Debugf("batch.Put %p %x", batch, nk) t.Put(nk, file.MustMarshalXDR())
batch.Put(nk, file.MustMarshalXDR())
return file.LocalVersion return file.LocalVersion
} }
// ldbUpdateGlobal adds this device+version to the version list for the given // updateGlobal 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. // 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. // 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, globalSize *sizeTracker) bool { func (t readWriteTransaction) updateGlobal(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) l.Debugf("update global; folder=%q device=%v file=%q version=%d", folder, protocol.DeviceIDFromBytes(device), file.Name, file.Version)
name := []byte(file.Name) name := []byte(file.Name)
gk := globalKey(folder, name) gk := globalKey(folder, name)
svl, err := db.Get(gk, nil) svl, err := t.Get(gk, nil)
if err != nil && err != leveldb.ErrNotFound { if err != nil && err != leveldb.ErrNotFound {
panic(err) panic(err)
} }
@ -432,7 +429,7 @@ func ldbUpdateGlobal(db dbReader, batch dbWriter, folder, device []byte, file pr
if i == 0 { if i == 0 {
// Keep the current newest file around so we can subtract it from // Keep the current newest file around so we can subtract it from
// the globalSize if we replace it. // the globalSize if we replace it.
oldFile, hasOldFile = ldbGet(db, folder, fl.versions[0].device, name) oldFile, hasOldFile = t.getFile(folder, fl.versions[0].device, name)
} }
fl.versions = append(fl.versions[:i], fl.versions[i+1:]...) fl.versions = append(fl.versions[:i], fl.versions[i+1:]...)
@ -465,7 +462,7 @@ func ldbUpdateGlobal(db dbReader, batch dbWriter, folder, device []byte, file pr
// "Greater" in the condition above is just based on the device // "Greater" in the condition above is just based on the device
// IDs in the version vector, which is not the only thing we use // IDs in the version vector, which is not the only thing we use
// to determine the winner.) // to determine the winner.)
of, ok := ldbGet(db, folder, fl.versions[i].device, name) of, ok := t.getFile(folder, fl.versions[i].device, name)
if !ok { if !ok {
panic("file referenced in version list does not exist") panic("file referenced in version list does not exist")
} }
@ -492,7 +489,7 @@ done:
globalSize.removeFile(oldFile) globalSize.removeFile(oldFile)
} else if len(fl.versions) > 1 { } else if len(fl.versions) > 1 {
// The previous newest version is now at index 1, grab it from there. // The previous newest version is now at index 1, grab it from there.
oldFile, ok := ldbGet(db, folder, fl.versions[1].device, name) oldFile, ok := t.getFile(folder, fl.versions[1].device, name)
if !ok { if !ok {
panic("file referenced in version list does not exist") panic("file referenced in version list does not exist")
} }
@ -501,9 +498,8 @@ done:
} }
} }
l.Debugf("batch.Put %p %x", batch, gk)
l.Debugf("new global after update: %v", fl) l.Debugf("new global after update: %v", fl)
batch.Put(gk, fl.MustMarshalXDR()) t.Put(gk, fl.MustMarshalXDR())
return true return true
} }
@ -518,11 +514,11 @@ func insertVersion(vl []fileVersion, i int, v fileVersion) []fileVersion {
// ldbRemoveFromGlobal removes the device from the global version list for the // 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 // given file. If the version list is empty after this, the file entry is
// removed entirely. // removed entirely.
func ldbRemoveFromGlobal(db dbReader, batch dbWriter, folder, device, file []byte, globalSize *sizeTracker) { func (t readWriteTransaction) removeFromGlobal(folder, device, file []byte, globalSize *sizeTracker) {
l.Debugf("remove from global; folder=%q device=%v file=%q", folder, protocol.DeviceIDFromBytes(device), file) l.Debugf("remove from global; folder=%q device=%v file=%q", folder, protocol.DeviceIDFromBytes(device), file)
gk := globalKey(folder, file) gk := globalKey(folder, file)
svl, err := db.Get(gk, nil) svl, err := t.Get(gk, nil)
if err != nil { if err != nil {
// We might be called to "remove" a global version that doesn't exist // We might be called to "remove" a global version that doesn't exist
// if the first update for the file is already marked invalid. // if the first update for the file is already marked invalid.
@ -539,7 +535,7 @@ func ldbRemoveFromGlobal(db dbReader, batch dbWriter, folder, device, file []byt
for i := range fl.versions { for i := range fl.versions {
if bytes.Compare(fl.versions[i].device, device) == 0 { if bytes.Compare(fl.versions[i].device, device) == 0 {
if i == 0 && globalSize != nil { if i == 0 && globalSize != nil {
f, ok := ldbGet(db, folder, device, file) f, ok := t.getFile(folder, device, file)
if !ok { if !ok {
panic("removing nonexistent file") panic("removing nonexistent file")
} }
@ -552,14 +548,12 @@ func ldbRemoveFromGlobal(db dbReader, batch dbWriter, folder, device, file []byt
} }
if len(fl.versions) == 0 { if len(fl.versions) == 0 {
l.Debugf("batch.Delete %p %x", batch, gk) t.Delete(gk)
batch.Delete(gk)
} else { } else {
l.Debugf("batch.Put %p %x", batch, gk)
l.Debugf("new global after remove: %v", fl) l.Debugf("new global after remove: %v", fl)
batch.Put(gk, fl.MustMarshalXDR()) t.Put(gk, fl.MustMarshalXDR())
if removed { if removed {
f, ok := ldbGet(db, folder, fl.versions[0].device, file) f, ok := t.getFile(folder, fl.versions[0].device, file)
if !ok { if !ok {
panic("new global is nonexistent file") panic("new global is nonexistent file")
} }
@ -568,20 +562,14 @@ func ldbRemoveFromGlobal(db dbReader, batch dbWriter, folder, device, file []byt
} }
} }
func ldbWithHave(db *leveldb.DB, folder, device []byte, truncate bool, fn Iterator) { func (db *dbInstance) withHave(folder, device []byte, truncate bool, fn Iterator) {
start := deviceKey(folder, device, nil) // before all folder/device files start := deviceKey(folder, device, nil) // before all folder/device files
limit := deviceKey(folder, device, []byte{0xff, 0xff, 0xff, 0xff}) // after all folder/device files limit := deviceKey(folder, device, []byte{0xff, 0xff, 0xff, 0xff}) // after all folder/device files
snap, err := db.GetSnapshot()
if err != nil {
panic(err)
}
l.Debugf("created snapshot %p", snap)
defer func() {
l.Debugf("close snapshot %p", snap)
snap.Release()
}()
dbi := snap.NewIterator(&util.Range{Start: start, Limit: limit}, nil) t := db.newReadOnlyTransaction()
defer t.close()
dbi := t.NewIterator(&util.Range{Start: start, Limit: limit}, nil)
defer dbi.Release() defer dbi.Release()
for dbi.Next() { for dbi.Next() {
@ -595,20 +583,14 @@ func ldbWithHave(db *leveldb.DB, folder, device []byte, truncate bool, fn Iterat
} }
} }
func ldbWithAllFolderTruncated(db *leveldb.DB, folder []byte, fn func(device []byte, f FileInfoTruncated) bool) { func (db *dbInstance) withAllFolderTruncated(folder []byte, fn func(device []byte, f FileInfoTruncated) bool) {
start := deviceKey(folder, nil, nil) // before all folder/device files start := deviceKey(folder, nil, nil) // before all folder/device files
limit := deviceKey(folder, protocol.LocalDeviceID[:], []byte{0xff, 0xff, 0xff, 0xff}) // after all folder/device files limit := deviceKey(folder, protocol.LocalDeviceID[:], []byte{0xff, 0xff, 0xff, 0xff}) // after all folder/device files
snap, err := db.GetSnapshot()
if err != nil {
panic(err)
}
l.Debugf("created snapshot %p", snap)
defer func() {
l.Debugf("close snapshot %p", snap)
snap.Release()
}()
dbi := snap.NewIterator(&util.Range{Start: start, Limit: limit}, nil) t := db.newReadWriteTransaction()
defer t.close()
dbi := t.NewIterator(&util.Range{Start: start, Limit: limit}, nil)
defer dbi.Release() defer dbi.Release()
for dbi.Next() { for dbi.Next() {
@ -622,10 +604,9 @@ func ldbWithAllFolderTruncated(db *leveldb.DB, folder []byte, fn func(device []b
switch f.Name { switch f.Name {
case "", ".", "..", "/": // A few obviously invalid filenames case "", ".", "..", "/": // A few obviously invalid filenames
l.Infof("Dropping invalid filename %q from database", f.Name) l.Infof("Dropping invalid filename %q from database", f.Name)
batch := new(leveldb.Batch) t.removeFromGlobal(folder, device, nil, nil)
ldbRemoveFromGlobal(db, batch, folder, device, nil, nil) t.Delete(dbi.Key())
batch.Delete(dbi.Key()) t.checkFlush()
db.Write(batch, nil)
continue continue
} }
@ -635,7 +616,15 @@ func ldbWithAllFolderTruncated(db *leveldb.DB, folder []byte, fn func(device []b
} }
} }
func ldbGet(db dbReader, folder, device, file []byte) (protocol.FileInfo, bool) { func (db *dbInstance) getFile(folder, device, file []byte) (protocol.FileInfo, bool) {
return ldbGetFile(db, folder, device, file)
}
func (t readOnlyTransaction) getFile(folder, device, file []byte) (protocol.FileInfo, bool) {
return ldbGetFile(t, folder, device, file)
}
func ldbGetFile(db dbReader, folder, device, file []byte) (protocol.FileInfo, bool) {
nk := deviceKey(folder, device, file) nk := deviceKey(folder, device, file)
bs, err := db.Get(nk, nil) bs, err := db.Get(nk, nil)
if err == leveldb.ErrNotFound { if err == leveldb.ErrNotFound {
@ -653,20 +642,13 @@ func ldbGet(db dbReader, folder, device, file []byte) (protocol.FileInfo, bool)
return f, true return f, true
} }
func ldbGetGlobal(db *leveldb.DB, folder, file []byte, truncate bool) (FileIntf, bool) { func (db *dbInstance) getGlobal(folder, file []byte, truncate bool) (FileIntf, bool) {
k := globalKey(folder, file) k := globalKey(folder, file)
snap, err := db.GetSnapshot()
if err != nil {
panic(err)
}
l.Debugf("created snapshot %p", snap)
defer func() {
l.Debugf("close snapshot %p", snap)
snap.Release()
}()
l.Debugf("snap.Get %p %x", snap, k) t := db.newReadOnlyTransaction()
bs, err := snap.Get(k, nil) defer t.close()
bs, err := t.Get(k, nil)
if err == leveldb.ErrNotFound { if err == leveldb.ErrNotFound {
return nil, false return nil, false
} }
@ -685,8 +667,7 @@ func ldbGetGlobal(db *leveldb.DB, folder, file []byte, truncate bool) (FileIntf,
} }
k = deviceKey(folder, vl.versions[0].device, file) k = deviceKey(folder, vl.versions[0].device, file)
l.Debugf("snap.Get %p %x", snap, k) bs, err = t.Get(k, nil)
bs, err = snap.Get(k, nil)
if err != nil { if err != nil {
panic(err) panic(err)
} }
@ -698,18 +679,11 @@ func ldbGetGlobal(db *leveldb.DB, folder, file []byte, truncate bool) (FileIntf,
return fi, true return fi, true
} }
func ldbWithGlobal(db *leveldb.DB, folder, prefix []byte, truncate bool, fn Iterator) { func (db *dbInstance) withGlobal(folder, prefix []byte, truncate bool, fn Iterator) {
snap, err := db.GetSnapshot() t := db.newReadOnlyTransaction()
if err != nil { defer t.close()
panic(err)
}
l.Debugf("created snapshot %p", snap)
defer func() {
l.Debugf("close snapshot %p", snap)
snap.Release()
}()
dbi := snap.NewIterator(util.BytesPrefix(globalKey(folder, prefix)), nil) dbi := t.NewIterator(util.BytesPrefix(globalKey(folder, prefix)), nil)
defer dbi.Release() defer dbi.Release()
var fk []byte var fk []byte
@ -725,8 +699,7 @@ func ldbWithGlobal(db *leveldb.DB, folder, prefix []byte, truncate bool, fn Iter
} }
name := globalKeyName(dbi.Key()) name := globalKeyName(dbi.Key())
fk = deviceKeyInto(fk[:cap(fk)], folder, vl.versions[0].device, name) fk = deviceKeyInto(fk[:cap(fk)], folder, vl.versions[0].device, name)
l.Debugf("snap.Get %p %x", snap, fk) bs, err := t.Get(fk, nil)
bs, err := snap.Get(fk, nil)
if err != nil { if err != nil {
l.Debugf("folder: %q (%x)", folder, folder) l.Debugf("folder: %q (%x)", folder, folder)
l.Debugf("key: %q (%x)", dbi.Key(), dbi.Key()) l.Debugf("key: %q (%x)", dbi.Key(), dbi.Key())
@ -749,7 +722,7 @@ func ldbWithGlobal(db *leveldb.DB, folder, prefix []byte, truncate bool, fn Iter
} }
} }
func ldbAvailability(db *leveldb.DB, folder, file []byte) []protocol.DeviceID { func (db *dbInstance) availability(folder, file []byte) []protocol.DeviceID {
k := globalKey(folder, file) k := globalKey(folder, file)
bs, err := db.Get(k, nil) bs, err := db.Get(k, nil)
if err == leveldb.ErrNotFound { if err == leveldb.ErrNotFound {
@ -777,20 +750,14 @@ func ldbAvailability(db *leveldb.DB, folder, file []byte) []protocol.DeviceID {
return devices return devices
} }
func ldbWithNeed(db *leveldb.DB, folder, device []byte, truncate bool, fn Iterator) { func (db *dbInstance) withNeed(folder, device []byte, truncate bool, fn Iterator) {
start := globalKey(folder, nil) start := globalKey(folder, nil)
limit := globalKey(folder, []byte{0xff, 0xff, 0xff, 0xff}) limit := globalKey(folder, []byte{0xff, 0xff, 0xff, 0xff})
snap, err := db.GetSnapshot()
if err != nil {
panic(err)
}
l.Debugf("created snapshot %p", snap)
defer func() {
l.Debugf("close snapshot %p", snap)
snap.Release()
}()
dbi := snap.NewIterator(&util.Range{Start: start, Limit: limit}, nil) t := db.newReadOnlyTransaction()
defer t.close()
dbi := t.NewIterator(&util.Range{Start: start, Limit: limit}, nil)
defer dbi.Release() defer dbi.Release()
var fk []byte var fk []byte
@ -832,8 +799,7 @@ nextFile:
continue nextFile continue nextFile
} }
fk = deviceKeyInto(fk[:cap(fk)], folder, vl.versions[i].device, name) fk = deviceKeyInto(fk[:cap(fk)], folder, vl.versions[i].device, name)
l.Debugf("snap.Get %p %x", snap, fk) bs, err := t.Get(fk, nil)
bs, err := snap.Get(fk, nil)
if err != nil { if err != nil {
var id protocol.DeviceID var id protocol.DeviceID
copy(id[:], device) copy(id[:], device)
@ -875,18 +841,11 @@ nextFile:
} }
} }
func ldbListFolders(db *leveldb.DB) []string { func (db *dbInstance) listFolders() []string {
snap, err := db.GetSnapshot() t := db.newReadOnlyTransaction()
if err != nil { defer t.close()
panic(err)
}
l.Debugf("created snapshot %p", snap)
defer func() {
l.Debugf("close snapshot %p", snap)
snap.Release()
}()
dbi := snap.NewIterator(util.BytesPrefix([]byte{KeyTypeGlobal}), nil) dbi := t.NewIterator(util.BytesPrefix([]byte{KeyTypeGlobal}), nil)
defer dbi.Release() defer dbi.Release()
folderExists := make(map[string]bool) folderExists := make(map[string]bool)
@ -906,19 +865,12 @@ func ldbListFolders(db *leveldb.DB) []string {
return folders return folders
} }
func ldbDropFolder(db *leveldb.DB, folder []byte) { func (db *dbInstance) dropFolder(folder []byte) {
snap, err := db.GetSnapshot() t := db.newReadOnlyTransaction()
if err != nil { defer t.close()
panic(err)
}
l.Debugf("created snapshot %p", snap)
defer func() {
l.Debugf("close snapshot %p", snap)
snap.Release()
}()
// Remove all items related to the given folder from the device->file bucket // Remove all items related to the given folder from the device->file bucket
dbi := snap.NewIterator(util.BytesPrefix([]byte{KeyTypeDevice}), nil) dbi := t.NewIterator(util.BytesPrefix([]byte{KeyTypeDevice}), nil)
for dbi.Next() { for dbi.Next() {
itemFolder := deviceKeyFolder(dbi.Key()) itemFolder := deviceKeyFolder(dbi.Key())
if bytes.Compare(folder, itemFolder) == 0 { if bytes.Compare(folder, itemFolder) == 0 {
@ -928,7 +880,7 @@ func ldbDropFolder(db *leveldb.DB, folder []byte) {
dbi.Release() dbi.Release()
// Remove all items related to the given folder from the global bucket // Remove all items related to the given folder from the global bucket
dbi = snap.NewIterator(util.BytesPrefix([]byte{KeyTypeGlobal}), nil) dbi = t.NewIterator(util.BytesPrefix([]byte{KeyTypeGlobal}), nil)
for dbi.Next() { for dbi.Next() {
itemFolder := globalKeyFolder(dbi.Key()) itemFolder := globalKeyFolder(dbi.Key())
if bytes.Compare(folder, itemFolder) == 0 { if bytes.Compare(folder, itemFolder) == 0 {
@ -950,25 +902,15 @@ func unmarshalTrunc(bs []byte, truncate bool) (FileIntf, error) {
return tf, err return tf, err
} }
func ldbCheckGlobals(db *leveldb.DB, folder []byte, globalSize *sizeTracker) { func (db *dbInstance) checkGlobals(folder []byte, globalSize *sizeTracker) {
snap, err := db.GetSnapshot() t := db.newReadWriteTransaction()
if err != nil { defer t.close()
panic(err)
}
l.Debugf("created snapshot %p", snap)
defer func() {
l.Debugf("close snapshot %p", snap)
snap.Release()
}()
start := globalKey(folder, nil) start := globalKey(folder, nil)
limit := globalKey(folder, []byte{0xff, 0xff, 0xff, 0xff}) limit := globalKey(folder, []byte{0xff, 0xff, 0xff, 0xff})
dbi := snap.NewIterator(&util.Range{Start: start, Limit: limit}, nil) dbi := t.NewIterator(&util.Range{Start: start, Limit: limit}, nil)
defer dbi.Release() defer dbi.Release()
batch := new(leveldb.Batch)
l.Debugf("new batch %p", batch)
var fk []byte var fk []byte
for dbi.Next() { for dbi.Next() {
gk := dbi.Key() gk := dbi.Key()
@ -988,8 +930,7 @@ func ldbCheckGlobals(db *leveldb.DB, folder []byte, globalSize *sizeTracker) {
for i, version := range vl.versions { for i, version := range vl.versions {
fk = deviceKeyInto(fk[:cap(fk)], folder, version.device, name) fk = deviceKeyInto(fk[:cap(fk)], folder, version.device, name)
l.Debugf("snap.Get %p %x", snap, fk) _, err := t.Get(fk, nil)
_, err := snap.Get(fk, nil)
if err == leveldb.ErrNotFound { if err == leveldb.ErrNotFound {
continue continue
} }
@ -999,7 +940,7 @@ func ldbCheckGlobals(db *leveldb.DB, folder []byte, globalSize *sizeTracker) {
newVL.versions = append(newVL.versions, version) newVL.versions = append(newVL.versions, version)
if i == 0 { if i == 0 {
fi, ok := ldbGet(snap, folder, version.device, name) fi, ok := t.getFile(folder, version.device, name)
if !ok { if !ok {
panic("nonexistent global master file") panic("nonexistent global master file")
} }
@ -1008,10 +949,9 @@ func ldbCheckGlobals(db *leveldb.DB, folder []byte, globalSize *sizeTracker) {
} }
if len(newVL.versions) != len(vl.versions) { if len(newVL.versions) != len(vl.versions) {
l.Infof("db repair: rewriting global version list for %x %x", gk[1:1+64], gk[1+64:]) t.Put(dbi.Key(), newVL.MustMarshalXDR())
batch.Put(dbi.Key(), newVL.MustMarshalXDR()) t.checkFlush()
} }
} }
l.Debugf("db check completed for %q", folder) l.Debugf("db check completed for %q", folder)
db.Write(batch, nil)
} }

42
lib/db/set.go
View File

@ -25,7 +25,7 @@ type FileSet struct {
localVersion map[protocol.DeviceID]int64 localVersion map[protocol.DeviceID]int64
mutex sync.Mutex mutex sync.Mutex
folder string folder string
db *leveldb.DB db *dbInstance
blockmap *BlockMap blockmap *BlockMap
localSize sizeTracker localSize sizeTracker
globalSize sizeTracker globalSize sizeTracker
@ -97,15 +97,15 @@ func NewFileSet(folder string, db *leveldb.DB) *FileSet {
var s = FileSet{ var s = FileSet{
localVersion: make(map[protocol.DeviceID]int64), localVersion: make(map[protocol.DeviceID]int64),
folder: folder, folder: folder,
db: db, db: &dbInstance{db},
blockmap: NewBlockMap(db, folder), blockmap: NewBlockMap(db, folder),
mutex: sync.NewMutex(), mutex: sync.NewMutex(),
} }
ldbCheckGlobals(db, []byte(folder), &s.globalSize) s.db.checkGlobals([]byte(folder), &s.globalSize)
var deviceID protocol.DeviceID var deviceID protocol.DeviceID
ldbWithAllFolderTruncated(db, []byte(folder), func(device []byte, f FileInfoTruncated) bool { s.db.withAllFolderTruncated([]byte(folder), func(device []byte, f FileInfoTruncated) bool {
copy(deviceID[:], device) copy(deviceID[:], device)
if f.LocalVersion > s.localVersion[deviceID] { if f.LocalVersion > s.localVersion[deviceID] {
s.localVersion[deviceID] = f.LocalVersion s.localVersion[deviceID] = f.LocalVersion
@ -126,7 +126,7 @@ func (s *FileSet) Replace(device protocol.DeviceID, fs []protocol.FileInfo) {
normalizeFilenames(fs) normalizeFilenames(fs)
s.mutex.Lock() s.mutex.Lock()
defer s.mutex.Unlock() defer s.mutex.Unlock()
s.localVersion[device] = ldbReplace(s.db, []byte(s.folder), device[:], fs, &s.localSize, &s.globalSize) s.localVersion[device] = s.db.replace([]byte(s.folder), device[:], fs, &s.localSize, &s.globalSize)
if len(fs) == 0 { if len(fs) == 0 {
// Reset the local version if all files were removed. // Reset the local version if all files were removed.
s.localVersion[device] = 0 s.localVersion[device] = 0
@ -146,7 +146,7 @@ func (s *FileSet) Update(device protocol.DeviceID, fs []protocol.FileInfo) {
discards := make([]protocol.FileInfo, 0, len(fs)) discards := make([]protocol.FileInfo, 0, len(fs))
updates := make([]protocol.FileInfo, 0, len(fs)) updates := make([]protocol.FileInfo, 0, len(fs))
for _, newFile := range fs { for _, newFile := range fs {
existingFile, ok := ldbGet(s.db, []byte(s.folder), device[:], []byte(newFile.Name)) existingFile, ok := s.db.getFile([]byte(s.folder), device[:], []byte(newFile.Name))
if !ok || !existingFile.Version.Equal(newFile.Version) { if !ok || !existingFile.Version.Equal(newFile.Version) {
discards = append(discards, existingFile) discards = append(discards, existingFile)
updates = append(updates, newFile) updates = append(updates, newFile)
@ -155,54 +155,54 @@ func (s *FileSet) Update(device protocol.DeviceID, fs []protocol.FileInfo) {
s.blockmap.Discard(discards) s.blockmap.Discard(discards)
s.blockmap.Update(updates) s.blockmap.Update(updates)
} }
if lv := ldbUpdate(s.db, []byte(s.folder), device[:], fs, &s.localSize, &s.globalSize); lv > s.localVersion[device] { if lv := s.db.updateFiles([]byte(s.folder), device[:], fs, &s.localSize, &s.globalSize); lv > s.localVersion[device] {
s.localVersion[device] = lv s.localVersion[device] = lv
} }
} }
func (s *FileSet) WithNeed(device protocol.DeviceID, fn Iterator) { func (s *FileSet) WithNeed(device protocol.DeviceID, fn Iterator) {
l.Debugf("%s WithNeed(%v)", s.folder, device) l.Debugf("%s WithNeed(%v)", s.folder, device)
ldbWithNeed(s.db, []byte(s.folder), device[:], false, nativeFileIterator(fn)) s.db.withNeed([]byte(s.folder), device[:], false, nativeFileIterator(fn))
} }
func (s *FileSet) WithNeedTruncated(device protocol.DeviceID, fn Iterator) { func (s *FileSet) WithNeedTruncated(device protocol.DeviceID, fn Iterator) {
l.Debugf("%s WithNeedTruncated(%v)", s.folder, device) l.Debugf("%s WithNeedTruncated(%v)", s.folder, device)
ldbWithNeed(s.db, []byte(s.folder), device[:], true, nativeFileIterator(fn)) s.db.withNeed([]byte(s.folder), device[:], true, nativeFileIterator(fn))
} }
func (s *FileSet) WithHave(device protocol.DeviceID, fn Iterator) { func (s *FileSet) WithHave(device protocol.DeviceID, fn Iterator) {
l.Debugf("%s WithHave(%v)", s.folder, device) l.Debugf("%s WithHave(%v)", s.folder, device)
ldbWithHave(s.db, []byte(s.folder), device[:], false, nativeFileIterator(fn)) s.db.withHave([]byte(s.folder), device[:], false, nativeFileIterator(fn))
} }
func (s *FileSet) WithHaveTruncated(device protocol.DeviceID, fn Iterator) { func (s *FileSet) WithHaveTruncated(device protocol.DeviceID, fn Iterator) {
l.Debugf("%s WithHaveTruncated(%v)", s.folder, device) l.Debugf("%s WithHaveTruncated(%v)", s.folder, device)
ldbWithHave(s.db, []byte(s.folder), device[:], true, nativeFileIterator(fn)) s.db.withHave([]byte(s.folder), device[:], true, nativeFileIterator(fn))
} }
func (s *FileSet) WithGlobal(fn Iterator) { func (s *FileSet) WithGlobal(fn Iterator) {
l.Debugf("%s WithGlobal()", s.folder) l.Debugf("%s WithGlobal()", s.folder)
ldbWithGlobal(s.db, []byte(s.folder), nil, false, nativeFileIterator(fn)) s.db.withGlobal([]byte(s.folder), nil, false, nativeFileIterator(fn))
} }
func (s *FileSet) WithGlobalTruncated(fn Iterator) { func (s *FileSet) WithGlobalTruncated(fn Iterator) {
l.Debugf("%s WithGlobalTruncated()", s.folder) l.Debugf("%s WithGlobalTruncated()", s.folder)
ldbWithGlobal(s.db, []byte(s.folder), nil, true, nativeFileIterator(fn)) s.db.withGlobal([]byte(s.folder), nil, true, nativeFileIterator(fn))
} }
func (s *FileSet) WithPrefixedGlobalTruncated(prefix string, fn Iterator) { func (s *FileSet) WithPrefixedGlobalTruncated(prefix string, fn Iterator) {
l.Debugf("%s WithPrefixedGlobalTruncated()", s.folder, prefix) l.Debugf("%s WithPrefixedGlobalTruncated()", s.folder, prefix)
ldbWithGlobal(s.db, []byte(s.folder), []byte(osutil.NormalizedFilename(prefix)), true, nativeFileIterator(fn)) s.db.withGlobal([]byte(s.folder), []byte(osutil.NormalizedFilename(prefix)), true, nativeFileIterator(fn))
} }
func (s *FileSet) Get(device protocol.DeviceID, file string) (protocol.FileInfo, bool) { func (s *FileSet) Get(device protocol.DeviceID, file string) (protocol.FileInfo, bool) {
f, ok := ldbGet(s.db, []byte(s.folder), device[:], []byte(osutil.NormalizedFilename(file))) f, ok := s.db.getFile([]byte(s.folder), device[:], []byte(osutil.NormalizedFilename(file)))
f.Name = osutil.NativeFilename(f.Name) f.Name = osutil.NativeFilename(f.Name)
return f, ok return f, ok
} }
func (s *FileSet) GetGlobal(file string) (protocol.FileInfo, bool) { func (s *FileSet) GetGlobal(file string) (protocol.FileInfo, bool) {
fi, ok := ldbGetGlobal(s.db, []byte(s.folder), []byte(osutil.NormalizedFilename(file)), false) fi, ok := s.db.getGlobal([]byte(s.folder), []byte(osutil.NormalizedFilename(file)), false)
if !ok { if !ok {
return protocol.FileInfo{}, false return protocol.FileInfo{}, false
} }
@ -212,7 +212,7 @@ func (s *FileSet) GetGlobal(file string) (protocol.FileInfo, bool) {
} }
func (s *FileSet) GetGlobalTruncated(file string) (FileInfoTruncated, bool) { func (s *FileSet) GetGlobalTruncated(file string) (FileInfoTruncated, bool) {
fi, ok := ldbGetGlobal(s.db, []byte(s.folder), []byte(osutil.NormalizedFilename(file)), true) fi, ok := s.db.getGlobal([]byte(s.folder), []byte(osutil.NormalizedFilename(file)), true)
if !ok { if !ok {
return FileInfoTruncated{}, false return FileInfoTruncated{}, false
} }
@ -222,7 +222,7 @@ func (s *FileSet) GetGlobalTruncated(file string) (FileInfoTruncated, bool) {
} }
func (s *FileSet) Availability(file string) []protocol.DeviceID { func (s *FileSet) Availability(file string) []protocol.DeviceID {
return ldbAvailability(s.db, []byte(s.folder), []byte(osutil.NormalizedFilename(file))) return s.db.availability([]byte(s.folder), []byte(osutil.NormalizedFilename(file)))
} }
func (s *FileSet) LocalVersion(device protocol.DeviceID) int64 { func (s *FileSet) LocalVersion(device protocol.DeviceID) int64 {
@ -241,13 +241,15 @@ func (s *FileSet) GlobalSize() (files, deleted int, bytes int64) {
// ListFolders returns the folder IDs seen in the database. // ListFolders returns the folder IDs seen in the database.
func ListFolders(db *leveldb.DB) []string { func ListFolders(db *leveldb.DB) []string {
return ldbListFolders(db) i := &dbInstance{db}
return i.listFolders()
} }
// DropFolder clears out all information related to the given folder from the // DropFolder clears out all information related to the given folder from the
// database. // database.
func DropFolder(db *leveldb.DB, folder string) { func DropFolder(db *leveldb.DB, folder string) {
ldbDropFolder(db, []byte(folder)) i := &dbInstance{db}
i.dropFolder([]byte(folder))
bm := &BlockMap{ bm := &BlockMap{
db: db, db: db,
folder: folder, folder: folder,