syncthing/internal/files/leveldb.go

1016 lines
24 KiB
Go
Raw Normal View History

2014-11-16 20:13:20 +00:00
// Copyright (C) 2014 The Syncthing Authors.
2014-09-29 19:43:32 +00:00
//
// This program is free software: you can redistribute it and/or modify it
// under the terms of the GNU General Public License as published by the Free
// Software Foundation, either version 3 of the License, or (at your option)
// any later version.
//
// This program is distributed in the hope that it will be useful, but WITHOUT
// ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
// FITNESS FOR A PARTICULAR PURPOSE. See the GNU General Public License for
// more details.
//
// You should have received a copy of the GNU General Public License along
// with this program. If not, see <http://www.gnu.org/licenses/>.
2014-09-04 06:31:38 +00:00
//go:generate -command genxdr go run ../../Godeps/_workspace/src/github.com/calmh/xdr/cmd/genxdr/main.go
//go:generate genxdr -o leveldb_xdr.go leveldb.go
2014-07-06 12:46:48 +00:00
package files
import (
"bytes"
"fmt"
"runtime"
2014-07-06 12:46:48 +00:00
"sort"
"sync"
2014-07-06 12:46:48 +00:00
2014-09-22 19:42:11 +00:00
"github.com/syncthing/syncthing/internal/lamport"
"github.com/syncthing/syncthing/internal/protocol"
2014-07-06 12:46:48 +00:00
"github.com/syndtr/goleveldb/leveldb"
"github.com/syndtr/goleveldb/leveldb/iterator"
"github.com/syndtr/goleveldb/leveldb/opt"
"github.com/syndtr/goleveldb/leveldb/util"
)
var (
clockTick uint64
clockMut sync.Mutex
)
func clock(v uint64) uint64 {
clockMut.Lock()
defer clockMut.Unlock()
if v > clockTick {
clockTick = v + 1
} else {
clockTick++
}
return clockTick
}
2014-07-06 12:46:48 +00:00
const (
keyTypeDevice = iota
2014-07-06 12:46:48 +00:00
keyTypeGlobal
2014-10-06 20:57:33 +00:00
keyTypeBlock
2014-07-06 12:46:48 +00:00
)
type fileVersion struct {
version uint64
2014-09-28 11:05:25 +00:00
device []byte
2014-07-06 12:46:48 +00:00
}
type versionList struct {
versions []fileVersion
}
func (l versionList) String() string {
var b bytes.Buffer
var id protocol.DeviceID
b.WriteString("{")
for i, v := range l.versions {
if i > 0 {
b.WriteString(", ")
}
copy(id[:], v.device)
fmt.Fprintf(&b, "{%d, %v}", v.version, id)
}
b.WriteString("}")
return b.String()
}
2014-07-12 21:06:48 +00:00
type fileList []protocol.FileInfo
2014-07-06 12:46:48 +00:00
func (l fileList) Len() int {
return len(l)
}
func (l fileList) Swap(a, b int) {
l[a], l[b] = l[b], l[a]
}
func (l fileList) Less(a, b int) bool {
return l[a].Name < l[b].Name
}
type dbReader interface {
Get([]byte, *opt.ReadOptions) ([]byte, error)
}
type dbWriter interface {
Put([]byte, []byte)
Delete([]byte)
}
// deviceKey returns a byte slice encoding the following information:
// keyTypeDevice (1 byte)
// folder (64 bytes)
// device (32 bytes)
// name (variable size)
func deviceKey(folder, device, file []byte) []byte {
2014-07-06 12:46:48 +00:00
k := make([]byte, 1+64+32+len(file))
k[0] = keyTypeDevice
if len(folder) > 64 {
panic("folder name too long")
}
copy(k[1:], []byte(folder))
copy(k[1+64:], device[:])
2014-07-06 12:46:48 +00:00
copy(k[1+64+32:], []byte(file))
return k
}
func deviceKeyName(key []byte) []byte {
2014-07-06 12:46:48 +00:00
return key[1+64+32:]
}
func deviceKeyFolder(key []byte) []byte {
folder := key[1 : 1+64]
izero := bytes.IndexByte(folder, 0)
if izero < 0 {
return folder
}
return folder[:izero]
2014-07-17 09:00:54 +00:00
}
func deviceKeyDevice(key []byte) []byte {
2014-07-17 09:00:54 +00:00
return key[1+64 : 1+64+32]
}
2014-07-06 12:46:48 +00:00
// globalKey returns a byte slice encoding the following information:
// keyTypeGlobal (1 byte)
// folder (64 bytes)
// name (variable size)
func globalKey(folder, file []byte) []byte {
k := make([]byte, 1+64+len(file))
k[0] = keyTypeGlobal
if len(folder) > 64 {
panic("folder name too long")
}
copy(k[1:], []byte(folder))
copy(k[1+64:], []byte(file))
return k
}
2014-07-06 12:46:48 +00:00
func globalKeyName(key []byte) []byte {
return key[1+64:]
}
func globalKeyFolder(key []byte) []byte {
folder := key[1 : 1+64]
izero := bytes.IndexByte(folder, 0)
if izero < 0 {
return folder
}
return folder[:izero]
}
type deletionHandler func(db dbReader, batch dbWriter, folder, device, name []byte, dbi iterator.Iterator) uint64
2014-07-06 12:46:48 +00:00
func ldbGenericReplace(db *leveldb.DB, folder, device []byte, fs []protocol.FileInfo, deleteFn deletionHandler) uint64 {
runtime.GC()
sort.Sort(fileList(fs)) // sort list on name, same as in the database
2014-07-06 12:46:48 +00:00
start := deviceKey(folder, device, nil) // before all folder/device files
limit := deviceKey(folder, device, []byte{0xff, 0xff, 0xff, 0xff}) // after all folder/device files
2014-07-06 12:46:48 +00:00
batch := new(leveldb.Batch)
if debugDB {
l.Debugf("new batch %p", batch)
}
2014-07-06 12:46:48 +00:00
snap, err := db.GetSnapshot()
if err != nil {
panic(err)
}
if debugDB {
l.Debugf("created snapshot %p", snap)
}
defer func() {
if debugDB {
l.Debugf("close snapshot %p", snap)
}
snap.Release()
}()
2014-07-06 12:46:48 +00:00
dbi := snap.NewIterator(&util.Range{Start: start, Limit: limit}, nil)
defer dbi.Release()
moreDb := dbi.Next()
fsi := 0
var maxLocalVer uint64
2014-07-06 12:46:48 +00:00
for {
var newName, oldName []byte
moreFs := fsi < len(fs)
if !moreDb && !moreFs {
break
}
if moreFs {
newName = []byte(fs[fsi].Name)
}
if moreDb {
oldName = deviceKeyName(dbi.Key())
2014-07-06 12:46:48 +00:00
}
cmp := bytes.Compare(newName, oldName)
if debugDB {
l.Debugf("generic replace; folder=%q device=%v moreFs=%v moreDb=%v cmp=%d newName=%q oldName=%q", folder, protocol.DeviceIDFromBytes(device), moreFs, moreDb, cmp, newName, oldName)
2014-07-06 12:46:48 +00:00
}
switch {
case moreFs && (!moreDb || cmp == -1):
if debugDB {
l.Debugln("generic replace; missing - insert")
}
// Database is missing this file. Insert it.
if lv := ldbInsert(batch, folder, device, fs[fsi]); lv > maxLocalVer {
maxLocalVer = lv
}
if fs[fsi].IsInvalid() {
ldbRemoveFromGlobal(snap, batch, folder, device, newName)
} else {
ldbUpdateGlobal(snap, batch, folder, device, newName, fs[fsi].Version)
}
2014-07-06 12:46:48 +00:00
fsi++
case moreFs && moreDb && cmp == 0:
// File exists on both sides - compare versions. We might get an
// update with the same version and different flags if a device has
// marked a file as invalid, so handle that too.
if debugDB {
l.Debugln("generic replace; exists - compare")
}
var ef FileInfoTruncated
2014-07-06 12:46:48 +00:00
ef.UnmarshalXDR(dbi.Value())
if fs[fsi].Version > ef.Version ||
(fs[fsi].Version == ef.Version && fs[fsi].Flags != ef.Flags) {
if debugDB {
l.Debugln("generic replace; differs - insert")
}
if lv := ldbInsert(batch, folder, device, fs[fsi]); lv > maxLocalVer {
maxLocalVer = lv
}
if fs[fsi].IsInvalid() {
ldbRemoveFromGlobal(snap, batch, folder, device, newName)
} else {
ldbUpdateGlobal(snap, batch, folder, device, newName, fs[fsi].Version)
}
} else if debugDB {
l.Debugln("generic replace; equal - ignore")
2014-07-06 12:46:48 +00:00
}
2014-07-06 12:46:48 +00:00
fsi++
moreDb = dbi.Next()
case moreDb && (!moreFs || cmp == 1):
if debugDB {
l.Debugln("generic replace; exists - remove")
}
if lv := deleteFn(snap, batch, folder, device, oldName, dbi); lv > maxLocalVer {
maxLocalVer = lv
2014-07-06 12:46:48 +00:00
}
moreDb = dbi.Next()
}
}
if debugDB {
l.Debugf("db.Write %p", batch)
}
2014-07-06 12:46:48 +00:00
err = db.Write(batch, nil)
if err != nil {
panic(err)
}
return maxLocalVer
2014-07-06 12:46:48 +00:00
}
func ldbReplace(db *leveldb.DB, folder, device []byte, fs []protocol.FileInfo) uint64 {
// TODO: Return the remaining maxLocalVer?
return ldbGenericReplace(db, folder, device, fs, func(db dbReader, batch dbWriter, folder, device, name []byte, dbi iterator.Iterator) uint64 {
// Database has a file that we are missing. Remove it.
if debugDB {
l.Debugf("delete; folder=%q device=%v name=%q", folder, protocol.DeviceIDFromBytes(device), name)
2014-07-06 12:46:48 +00:00
}
ldbRemoveFromGlobal(db, batch, folder, device, name)
if debugDB {
l.Debugf("batch.Delete %p %x", batch, dbi.Key())
}
2014-07-15 15:54:00 +00:00
batch.Delete(dbi.Key())
return 0
2014-07-06 12:46:48 +00:00
})
}
func ldbReplaceWithDelete(db *leveldb.DB, folder, device []byte, fs []protocol.FileInfo) uint64 {
return ldbGenericReplace(db, folder, device, fs, func(db dbReader, batch dbWriter, folder, device, name []byte, dbi iterator.Iterator) uint64 {
var tf FileInfoTruncated
err := tf.UnmarshalXDR(dbi.Value())
2014-07-06 12:46:48 +00:00
if err != nil {
panic(err)
}
if !tf.IsDeleted() {
if debugDB {
l.Debugf("mark deleted; folder=%q device=%v name=%q", folder, protocol.DeviceIDFromBytes(device), name)
2014-07-06 12:46:48 +00:00
}
ts := clock(tf.LocalVersion)
f := protocol.FileInfo{
Name: tf.Name,
Version: lamport.Default.Tick(tf.Version),
LocalVersion: ts,
Flags: tf.Flags | protocol.FlagDeleted,
Modified: tf.Modified,
}
2014-10-21 06:40:05 +00:00
bs, _ := f.MarshalXDR()
if debugDB {
l.Debugf("batch.Put %p %x", batch, dbi.Key())
}
2014-10-21 06:40:05 +00:00
batch.Put(dbi.Key(), bs)
ldbUpdateGlobal(db, batch, folder, device, deviceKeyName(dbi.Key()), f.Version)
return ts
2014-07-06 12:46:48 +00:00
}
return 0
2014-07-06 12:46:48 +00:00
})
}
func ldbUpdate(db *leveldb.DB, folder, device []byte, fs []protocol.FileInfo) uint64 {
runtime.GC()
2014-07-06 12:46:48 +00:00
batch := new(leveldb.Batch)
if debugDB {
l.Debugf("new batch %p", batch)
}
2014-07-06 12:46:48 +00:00
snap, err := db.GetSnapshot()
if err != nil {
panic(err)
}
if debugDB {
l.Debugf("created snapshot %p", snap)
}
defer func() {
if debugDB {
l.Debugf("close snapshot %p", snap)
}
snap.Release()
}()
2014-07-06 12:46:48 +00:00
var maxLocalVer uint64
2014-07-06 12:46:48 +00:00
for _, f := range fs {
name := []byte(f.Name)
fk := deviceKey(folder, device, name)
if debugDB {
l.Debugf("snap.Get %p %x", snap, fk)
}
2014-07-06 12:46:48 +00:00
bs, err := snap.Get(fk, nil)
if err == leveldb.ErrNotFound {
if lv := ldbInsert(batch, folder, device, f); lv > maxLocalVer {
maxLocalVer = lv
}
if f.IsInvalid() {
ldbRemoveFromGlobal(snap, batch, folder, device, name)
} else {
ldbUpdateGlobal(snap, batch, folder, device, name, f.Version)
}
2014-07-06 12:46:48 +00:00
continue
}
var ef FileInfoTruncated
2014-07-06 12:46:48 +00:00
err = ef.UnmarshalXDR(bs)
if err != nil {
panic(err)
}
// Flags might change without the version being bumped when we set the
// invalid flag on an existing file.
if ef.Version != f.Version || ef.Flags != f.Flags {
if lv := ldbInsert(batch, folder, device, f); lv > maxLocalVer {
maxLocalVer = lv
}
if f.IsInvalid() {
ldbRemoveFromGlobal(snap, batch, folder, device, name)
} else {
ldbUpdateGlobal(snap, batch, folder, device, name, f.Version)
}
2014-07-06 12:46:48 +00:00
}
}
if debugDB {
l.Debugf("db.Write %p", batch)
}
2014-07-06 12:46:48 +00:00
err = db.Write(batch, nil)
if err != nil {
panic(err)
}
return maxLocalVer
2014-07-06 12:46:48 +00:00
}
func ldbInsert(batch dbWriter, folder, device []byte, file protocol.FileInfo) uint64 {
if debugDB {
l.Debugf("insert; folder=%q device=%v %v", folder, protocol.DeviceIDFromBytes(device), file)
2014-07-06 12:46:48 +00:00
}
if file.LocalVersion == 0 {
file.LocalVersion = clock(0)
}
name := []byte(file.Name)
nk := deviceKey(folder, device, name)
if debugDB {
l.Debugf("batch.Put %p %x", batch, nk)
}
2014-10-21 06:40:05 +00:00
batch.Put(nk, file.MustMarshalXDR())
return file.LocalVersion
2014-07-06 12:46:48 +00:00
}
// 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.
2014-07-06 12:46:48 +00:00
// If the file does not have an entry in the global list, it is created.
func ldbUpdateGlobal(db dbReader, batch dbWriter, folder, device, file []byte, version uint64) bool {
if debugDB {
l.Debugf("update global; folder=%q device=%v file=%q version=%d", folder, protocol.DeviceIDFromBytes(device), file, version)
2014-07-06 12:46:48 +00:00
}
gk := globalKey(folder, file)
2014-07-06 12:46:48 +00:00
svl, err := db.Get(gk, nil)
if err != nil && err != leveldb.ErrNotFound {
panic(err)
}
var fl versionList
nv := fileVersion{
2014-09-28 11:05:25 +00:00
device: device,
2014-07-06 12:46:48 +00:00
version: version,
}
if svl != nil {
err = fl.UnmarshalXDR(svl)
if err != nil {
panic(err)
}
for i := range fl.versions {
if bytes.Compare(fl.versions[i].device, device) == 0 {
2014-07-06 12:46:48 +00:00
if fl.versions[i].version == version {
// No need to do anything
return false
}
fl.versions = append(fl.versions[:i], fl.versions[i+1:]...)
break
}
}
}
for i := range fl.versions {
if fl.versions[i].version <= version {
t := append(fl.versions, fileVersion{})
copy(t[i+1:], t[i:])
t[i] = nv
fl.versions = t
goto done
}
}
fl.versions = append(fl.versions, nv)
done:
if debugDB {
l.Debugf("batch.Put %p %x", batch, gk)
l.Debugf("new global after update: %v", fl)
}
2014-10-21 06:40:05 +00:00
batch.Put(gk, fl.MustMarshalXDR())
2014-07-06 12:46:48 +00:00
return true
}
// ldbRemoveFromGlobal removes the device from the global version list for the
2014-07-06 12:46:48 +00:00
// 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) {
if debugDB {
l.Debugf("remove from global; folder=%q device=%v file=%q", folder, protocol.DeviceIDFromBytes(device), file)
2014-07-06 12:46:48 +00:00
}
gk := globalKey(folder, file)
2014-07-06 12:46:48 +00:00
svl, err := db.Get(gk, nil)
if err != nil {
// We might be called to "remove" a global version that doesn't exist
// if the first update for the file is already marked invalid.
return
2014-07-06 12:46:48 +00:00
}
var fl versionList
err = fl.UnmarshalXDR(svl)
if err != nil {
panic(err)
}
for i := range fl.versions {
if bytes.Compare(fl.versions[i].device, device) == 0 {
2014-07-06 12:46:48 +00:00
fl.versions = append(fl.versions[:i], fl.versions[i+1:]...)
break
}
}
if len(fl.versions) == 0 {
if debugDB {
l.Debugf("batch.Delete %p %x", batch, gk)
}
2014-07-06 12:46:48 +00:00
batch.Delete(gk)
} else {
if debugDB {
l.Debugf("batch.Put %p %x", batch, gk)
l.Debugf("new global after remove: %v", fl)
}
2014-10-21 06:40:05 +00:00
batch.Put(gk, fl.MustMarshalXDR())
2014-07-06 12:46:48 +00:00
}
}
func ldbWithHave(db *leveldb.DB, folder, device []byte, truncate bool, fn Iterator) {
start := deviceKey(folder, device, nil) // before all folder/device files
limit := deviceKey(folder, device, []byte{0xff, 0xff, 0xff, 0xff}) // after all folder/device files
2014-07-06 12:46:48 +00:00
snap, err := db.GetSnapshot()
if err != nil {
panic(err)
}
if debugDB {
l.Debugf("created snapshot %p", snap)
}
defer func() {
if debugDB {
l.Debugf("close snapshot %p", snap)
}
snap.Release()
}()
2014-07-06 12:46:48 +00:00
dbi := snap.NewIterator(&util.Range{Start: start, Limit: limit}, nil)
defer dbi.Release()
for dbi.Next() {
f, err := unmarshalTrunc(dbi.Value(), truncate)
2014-07-06 12:46:48 +00:00
if err != nil {
panic(err)
}
if cont := fn(f); !cont {
return
}
}
}
func ldbWithAllFolderTruncated(db *leveldb.DB, folder []byte, fn func(device []byte, f FileInfoTruncated) bool) {
runtime.GC()
2014-09-28 11:05:25 +00:00
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
2014-07-17 09:00:54 +00:00
snap, err := db.GetSnapshot()
if err != nil {
panic(err)
}
if debugDB {
l.Debugf("created snapshot %p", snap)
}
defer func() {
if debugDB {
l.Debugf("close snapshot %p", snap)
}
snap.Release()
}()
2014-07-17 09:00:54 +00:00
dbi := snap.NewIterator(&util.Range{Start: start, Limit: limit}, nil)
defer dbi.Release()
for dbi.Next() {
device := deviceKeyDevice(dbi.Key())
var f FileInfoTruncated
2014-07-17 09:00:54 +00:00
err := f.UnmarshalXDR(dbi.Value())
if err != nil {
panic(err)
}
if cont := fn(device, f); !cont {
2014-07-17 09:00:54 +00:00
return
}
}
}
func ldbGet(db *leveldb.DB, folder, device, file []byte) (protocol.FileInfo, bool) {
nk := deviceKey(folder, device, file)
2014-07-06 12:46:48 +00:00
bs, err := db.Get(nk, nil)
if err == leveldb.ErrNotFound {
return protocol.FileInfo{}, false
2014-07-06 12:46:48 +00:00
}
if err != nil {
panic(err)
}
2014-07-12 21:06:48 +00:00
var f protocol.FileInfo
2014-07-06 12:46:48 +00:00
err = f.UnmarshalXDR(bs)
if err != nil {
panic(err)
}
return f, true
2014-07-06 12:46:48 +00:00
}
2015-01-09 07:41:02 +00:00
func ldbGetGlobal(db *leveldb.DB, folder, file []byte, truncate bool) (FileIntf, bool) {
k := globalKey(folder, file)
2014-07-06 12:46:48 +00:00
snap, err := db.GetSnapshot()
if err != nil {
panic(err)
}
if debugDB {
l.Debugf("created snapshot %p", snap)
}
defer func() {
if debugDB {
l.Debugf("close snapshot %p", snap)
}
snap.Release()
}()
2014-07-06 12:46:48 +00:00
if debugDB {
l.Debugf("snap.Get %p %x", snap, k)
}
2014-07-06 12:46:48 +00:00
bs, err := snap.Get(k, nil)
if err == leveldb.ErrNotFound {
2015-01-09 07:41:02 +00:00
return nil, false
2014-07-06 12:46:48 +00:00
}
if err != nil {
panic(err)
}
var vl versionList
err = vl.UnmarshalXDR(bs)
if err != nil {
panic(err)
}
if len(vl.versions) == 0 {
l.Debugln(k)
panic("no versions?")
}
k = deviceKey(folder, vl.versions[0].device, file)
if debugDB {
l.Debugf("snap.Get %p %x", snap, k)
}
2014-07-06 12:46:48 +00:00
bs, err = snap.Get(k, nil)
if err != nil {
panic(err)
}
2015-01-09 07:41:02 +00:00
fi, err := unmarshalTrunc(bs, truncate)
2014-07-06 12:46:48 +00:00
if err != nil {
panic(err)
}
2015-01-09 07:41:02 +00:00
return fi, true
2014-07-06 12:46:48 +00:00
}
func ldbWithGlobal(db *leveldb.DB, folder []byte, truncate bool, fn Iterator) {
runtime.GC()
start := globalKey(folder, nil)
limit := globalKey(folder, []byte{0xff, 0xff, 0xff, 0xff})
2014-07-06 12:46:48 +00:00
snap, err := db.GetSnapshot()
if err != nil {
panic(err)
}
if debugDB {
l.Debugf("created snapshot %p", snap)
}
defer func() {
if debugDB {
l.Debugf("close snapshot %p", snap)
}
snap.Release()
}()
2014-07-06 12:46:48 +00:00
dbi := snap.NewIterator(&util.Range{Start: start, Limit: limit}, nil)
defer dbi.Release()
for dbi.Next() {
var vl versionList
err := vl.UnmarshalXDR(dbi.Value())
if err != nil {
panic(err)
}
if len(vl.versions) == 0 {
l.Debugln(dbi.Key())
panic("no versions?")
}
name := globalKeyName(dbi.Key())
fk := deviceKey(folder, vl.versions[0].device, name)
if debugDB {
l.Debugf("snap.Get %p %x", snap, fk)
}
2014-07-06 12:46:48 +00:00
bs, err := snap.Get(fk, nil)
if err != nil {
l.Debugf("folder: %q (%x)", folder, folder)
l.Debugf("key: %q (%x)", dbi.Key(), dbi.Key())
l.Debugf("vl: %v", vl)
l.Debugf("vl.versions[0].device: %x", vl.versions[0].device)
l.Debugf("name: %q (%x)", name, name)
l.Debugf("fk: %q", fk)
l.Debugf("fk: %x %x %x", fk[1:1+64], fk[1+64:1+64+32], fk[1+64+32:])
2014-07-06 12:46:48 +00:00
panic(err)
}
f, err := unmarshalTrunc(bs, truncate)
2014-07-06 12:46:48 +00:00
if err != nil {
panic(err)
}
if cont := fn(f); !cont {
return
}
}
}
func ldbAvailability(db *leveldb.DB, folder, file []byte) []protocol.DeviceID {
k := globalKey(folder, file)
2014-07-06 12:46:48 +00:00
bs, err := db.Get(k, nil)
if err == leveldb.ErrNotFound {
return nil
}
if err != nil {
panic(err)
}
var vl versionList
err = vl.UnmarshalXDR(bs)
if err != nil {
panic(err)
}
var devices []protocol.DeviceID
2014-07-06 12:46:48 +00:00
for _, v := range vl.versions {
if v.version != vl.versions[0].version {
break
}
n := protocol.DeviceIDFromBytes(v.device)
devices = append(devices, n)
2014-07-06 12:46:48 +00:00
}
return devices
2014-07-06 12:46:48 +00:00
}
func ldbWithNeed(db *leveldb.DB, folder, device []byte, truncate bool, fn Iterator) {
runtime.GC()
start := globalKey(folder, nil)
limit := globalKey(folder, []byte{0xff, 0xff, 0xff, 0xff})
2014-07-06 12:46:48 +00:00
snap, err := db.GetSnapshot()
if err != nil {
panic(err)
}
if debugDB {
l.Debugf("created snapshot %p", snap)
}
defer func() {
if debugDB {
l.Debugf("close snapshot %p", snap)
}
snap.Release()
}()
2014-07-06 12:46:48 +00:00
dbi := snap.NewIterator(&util.Range{Start: start, Limit: limit}, nil)
defer dbi.Release()
outer:
2014-07-06 12:46:48 +00:00
for dbi.Next() {
var vl versionList
err := vl.UnmarshalXDR(dbi.Value())
if err != nil {
panic(err)
}
if len(vl.versions) == 0 {
l.Debugln(dbi.Key())
panic("no versions?")
}
have := false // If we have the file, any version
need := false // If we have a lower version of the file
var haveVersion uint64
for _, v := range vl.versions {
if bytes.Compare(v.device, device) == 0 {
2014-07-06 12:46:48 +00:00
have = true
haveVersion = v.version
need = v.version < vl.versions[0].version
break
}
}
if need || !have {
name := globalKeyName(dbi.Key())
needVersion := vl.versions[0].version
inner:
for i := range vl.versions {
if vl.versions[i].version != needVersion {
// We haven't found a valid copy of the file with the needed version.
continue outer
}
fk := deviceKey(folder, vl.versions[i].device, name)
if debugDB {
l.Debugf("snap.Get %p %x", snap, fk)
}
bs, err := snap.Get(fk, nil)
if err != nil {
var id protocol.DeviceID
copy(id[:], device)
l.Debugf("device: %v", id)
l.Debugf("need: %v, have: %v", need, have)
l.Debugf("key: %q (%x)", dbi.Key(), dbi.Key())
l.Debugf("vl: %v", vl)
l.Debugf("i: %v", i)
l.Debugf("fk: %q (%x)", fk, fk)
l.Debugf("name: %q (%x)", name, name)
panic(err)
}
2014-07-06 12:46:48 +00:00
gf, err := unmarshalTrunc(bs, truncate)
if err != nil {
panic(err)
}
2014-07-06 12:46:48 +00:00
if gf.IsInvalid() {
// The file is marked invalid for whatever reason, don't use it.
continue inner
}
2014-07-06 12:46:48 +00:00
if gf.IsDeleted() && !have {
// We don't need deleted files that we don't have
continue outer
}
if debugDB {
l.Debugf("need folder=%q device=%v name=%q need=%v have=%v haveV=%d globalV=%d", folder, protocol.DeviceIDFromBytes(device), name, need, have, haveVersion, vl.versions[0].version)
}
if cont := fn(gf); !cont {
return
}
// This file is handled, no need to look further in the version list
continue outer
2014-07-06 12:46:48 +00:00
}
}
}
}
func ldbListFolders(db *leveldb.DB) []string {
runtime.GC()
start := []byte{keyTypeGlobal}
limit := []byte{keyTypeGlobal + 1}
snap, err := db.GetSnapshot()
if err != nil {
panic(err)
}
if debugDB {
l.Debugf("created snapshot %p", snap)
}
defer func() {
if debugDB {
l.Debugf("close snapshot %p", snap)
}
snap.Release()
}()
dbi := snap.NewIterator(&util.Range{Start: start, Limit: limit}, nil)
defer dbi.Release()
folderExists := make(map[string]bool)
for dbi.Next() {
folder := string(globalKeyFolder(dbi.Key()))
if !folderExists[folder] {
folderExists[folder] = true
}
}
folders := make([]string, 0, len(folderExists))
for k := range folderExists {
folders = append(folders, k)
}
sort.Strings(folders)
return folders
}
func ldbDropFolder(db *leveldb.DB, folder []byte) {
runtime.GC()
snap, err := db.GetSnapshot()
if err != nil {
panic(err)
}
if debugDB {
l.Debugf("created snapshot %p", snap)
}
defer func() {
if debugDB {
l.Debugf("close snapshot %p", snap)
}
snap.Release()
}()
// Remove all items related to the given folder from the device->file bucket
start := []byte{keyTypeDevice}
limit := []byte{keyTypeDevice + 1}
dbi := snap.NewIterator(&util.Range{Start: start, Limit: limit}, nil)
for dbi.Next() {
itemFolder := deviceKeyFolder(dbi.Key())
if bytes.Compare(folder, itemFolder) == 0 {
db.Delete(dbi.Key(), nil)
}
}
dbi.Release()
// Remove all items related to the given folder from the global bucket
start = []byte{keyTypeGlobal}
limit = []byte{keyTypeGlobal + 1}
dbi = snap.NewIterator(&util.Range{Start: start, Limit: limit}, nil)
for dbi.Next() {
itemFolder := globalKeyFolder(dbi.Key())
if bytes.Compare(folder, itemFolder) == 0 {
db.Delete(dbi.Key(), nil)
}
}
dbi.Release()
}
func unmarshalTrunc(bs []byte, truncate bool) (FileIntf, error) {
if truncate {
var tf FileInfoTruncated
err := tf.UnmarshalXDR(bs)
return tf, err
} else {
var tf protocol.FileInfo
err := tf.UnmarshalXDR(bs)
return tf, err
}
}
func ldbCheckGlobals(db *leveldb.DB, folder []byte) {
defer runtime.GC()
snap, err := db.GetSnapshot()
if err != nil {
panic(err)
}
if debugDB {
l.Debugf("created snapshot %p", snap)
}
defer func() {
if debugDB {
l.Debugf("close snapshot %p", snap)
}
snap.Release()
}()
start := globalKey(folder, nil)
limit := globalKey(folder, []byte{0xff, 0xff, 0xff, 0xff})
dbi := snap.NewIterator(&util.Range{Start: start, Limit: limit}, nil)
defer dbi.Release()
batch := new(leveldb.Batch)
if debugDB {
l.Debugf("new batch %p", batch)
}
for dbi.Next() {
gk := dbi.Key()
var vl versionList
err := vl.UnmarshalXDR(dbi.Value())
if err != nil {
panic(err)
}
// Check the global version list for consistency. An issue in previous
// versions of goleveldb could result in reordered writes so that
// there are global entries pointing to no longer existing files. Here
// we find those and clear them out.
name := globalKeyName(gk)
var newVL versionList
for _, version := range vl.versions {
fk := deviceKey(folder, version.device, name)
if debugDB {
l.Debugf("snap.Get %p %x", snap, fk)
}
_, err := snap.Get(fk, nil)
if err == leveldb.ErrNotFound {
continue
}
if err != nil {
panic(err)
}
newVL.versions = append(newVL.versions, version)
}
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:])
batch.Put(dbi.Key(), newVL.MustMarshalXDR())
}
}
if debugDB {
l.Infoln("db check completed for %q", folder)
}
db.Write(batch, nil)
}