perso/syncthing-arm
1
0
Fork 0
Code Issues Pull Requests Releases Wiki Activity

We should pass around db.Instance instead of leveldb.DB

Browse Source 
We're going to need the db.Instance to keep some state, and for that to
work we need the same one passed around everywhere. Hence this moves the
leveldb-specific file opening stuff into the db package and exports the
dbInstance type.
This commit is contained in:
Jakob Borg
2015-10-31 12:31:25 +01:00
parent 313485e406
commit 2a4fc28318
20 changed files with 145 additions and 241 deletions
Download Patch File Download Diff File Expand all files Collapse all files

8
lib/db/blockmap.go
View File

@@ -29,11 +29,11 @@ var blockFinder *BlockFinder
const maxBatchSize = 256 << 10
type BlockMap struct {
db *leveldb.DB
db *Instance
folder string
}
func NewBlockMap(db *leveldb.DB, folder string) *BlockMap {
func NewBlockMap(db *Instance, folder string) *BlockMap {
return &BlockMap{
db: db,
folder: folder,
@@ -146,10 +146,10 @@ func (m *BlockMap) blockKeyInto(o, hash []byte, file string) []byte {
}
type BlockFinder struct {
db *leveldb.DB
db *Instance
}
func NewBlockFinder(db *leveldb.DB) *BlockFinder {
func NewBlockFinder(db *Instance) *BlockFinder {
if blockFinder != nil {
return blockFinder
}

12
lib/db/blockmap_test.go
View File

@@ -10,9 +10,6 @@ import (
"testing"
"github.com/syncthing/syncthing/lib/protocol"
"github.com/syndtr/goleveldb/leveldb"
"github.com/syndtr/goleveldb/leveldb/storage"
)
func genBlocks(n int) []protocol.BlockInfo {
@@ -50,17 +47,14 @@ func init() {
}
}
func setup() (*leveldb.DB, *BlockFinder) {
func setup() (*Instance, *BlockFinder) {
// Setup
db, err := leveldb.Open(storage.NewMemStorage(), nil)
if err != nil {
panic(err)
}
db := OpenMemory()
return db, NewBlockFinder(db)
}
func dbEmpty(db *leveldb.DB) bool {
func dbEmpty(db *Instance) bool {
iter := db.NewIterator(nil, nil)
defer iter.Release()
if iter.Next() {

101
lib/db/leveldb_dbinstance.go
View File

@@ -8,27 +8,64 @@ package db
import (
"bytes"
"os"
"sort"
"strings"
"github.com/syncthing/syncthing/lib/protocol"
"github.com/syndtr/goleveldb/leveldb"
"github.com/syndtr/goleveldb/leveldb/errors"
"github.com/syndtr/goleveldb/leveldb/iterator"
"github.com/syndtr/goleveldb/leveldb/opt"
"github.com/syndtr/goleveldb/leveldb/storage"
"github.com/syndtr/goleveldb/leveldb/util"
)
type deletionHandler func(t readWriteTransaction, folder, device, name []byte, dbi iterator.Iterator) int64
type dbInstance struct {
type Instance struct {
*leveldb.DB
}
func newDBInstance(db *leveldb.DB) *dbInstance {
return &dbInstance{
func Open(file string) (*Instance, error) {
opts := &opt.Options{
OpenFilesCacheCapacity: 100,
WriteBuffer: 4 << 20,
}
db, err := leveldb.OpenFile(file, opts)
if leveldbIsCorrupted(err) {
db, err = leveldb.RecoverFile(file, opts)
}
if leveldbIsCorrupted(err) {
// The database is corrupted, and we've tried to recover it but it
// didn't work. At this point there isn't much to do beyond dropping
// the database and reindexing...
l.Infoln("Database corruption detected, unable to recover. Reinitializing...")
if err := os.RemoveAll(file); err != nil {
return nil, err
}
db, err = leveldb.OpenFile(file, opts)
}
if err != nil {
return nil, err
}
return newDBInstance(db), nil
}
func OpenMemory() *Instance {
db, _ := leveldb.Open(storage.NewMemStorage(), nil)
return newDBInstance(db)
}
func newDBInstance(db *leveldb.DB) *Instance {
return &Instance{
DB: db,
}
}
func (db *dbInstance) 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) int64 {
sort.Sort(fileList(fs)) // sort list on name, same as in the database
start := db.deviceKey(folder, device, nil) // before all folder/device files
@@ -126,7 +163,7 @@ func (db *dbInstance) genericReplace(folder, device []byte, fs []protocol.FileIn
return maxLocalVer
}
func (db *dbInstance) replace(folder, device []byte, fs []protocol.FileInfo, localSize, globalSize *sizeTracker) int64 {
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 {
// Database has a file that we are missing. Remove it.
@@ -137,7 +174,7 @@ func (db *dbInstance) replace(folder, device []byte, fs []protocol.FileInfo, loc
})
}
func (db *dbInstance) updateFiles(folder, device []byte, fs []protocol.FileInfo, localSize, globalSize *sizeTracker) int64 {
func (db *Instance) updateFiles(folder, device []byte, fs []protocol.FileInfo, localSize, globalSize *sizeTracker) int64 {
t := db.newReadWriteTransaction()
defer t.close()
@@ -195,7 +232,7 @@ func (db *dbInstance) updateFiles(folder, device []byte, fs []protocol.FileInfo,
return maxLocalVer
}
func (db *dbInstance) withHave(folder, device []byte, truncate bool, fn Iterator) {
func (db *Instance) withHave(folder, device []byte, truncate bool, fn Iterator) {
start := db.deviceKey(folder, device, nil) // before all folder/device files
limit := db.deviceKey(folder, device, []byte{0xff, 0xff, 0xff, 0xff}) // after all folder/device files
@@ -216,7 +253,7 @@ func (db *dbInstance) withHave(folder, device []byte, truncate bool, fn Iterator
}
}
func (db *dbInstance) withAllFolderTruncated(folder []byte, fn func(device []byte, f FileInfoTruncated) bool) {
func (db *Instance) withAllFolderTruncated(folder []byte, fn func(device []byte, f FileInfoTruncated) bool) {
start := db.deviceKey(folder, nil, nil) // before all folder/device files
limit := db.deviceKey(folder, protocol.LocalDeviceID[:], []byte{0xff, 0xff, 0xff, 0xff}) // after all folder/device files
@@ -249,11 +286,11 @@ func (db *dbInstance) withAllFolderTruncated(folder []byte, fn func(device []byt
}
}
func (db *dbInstance) getFile(folder, device, file []byte) (protocol.FileInfo, bool) {
func (db *Instance) getFile(folder, device, file []byte) (protocol.FileInfo, bool) {
return getFile(db, db.deviceKey(folder, device, file))
}
func (db *dbInstance) getGlobal(folder, file []byte, truncate bool) (FileIntf, bool) {
func (db *Instance) getGlobal(folder, file []byte, truncate bool) (FileIntf, bool) {
k := db.globalKey(folder, file)
t := db.newReadOnlyTransaction()
@@ -290,7 +327,7 @@ func (db *dbInstance) getGlobal(folder, file []byte, truncate bool) (FileIntf, b
return fi, true
}
func (db *dbInstance) withGlobal(folder, prefix []byte, truncate bool, fn Iterator) {
func (db *Instance) withGlobal(folder, prefix []byte, truncate bool, fn Iterator) {
t := db.newReadOnlyTransaction()
defer t.close()
@@ -333,7 +370,7 @@ func (db *dbInstance) withGlobal(folder, prefix []byte, truncate bool, fn Iterat
}
}
func (db *dbInstance) availability(folder, file []byte) []protocol.DeviceID {
func (db *Instance) availability(folder, file []byte) []protocol.DeviceID {
k := db.globalKey(folder, file)
bs, err := db.Get(k, nil)
if err == leveldb.ErrNotFound {
@@ -361,7 +398,7 @@ func (db *dbInstance) availability(folder, file []byte) []protocol.DeviceID {
return devices
}
func (db *dbInstance) withNeed(folder, device []byte, truncate bool, fn Iterator) {
func (db *Instance) withNeed(folder, device []byte, truncate bool, fn Iterator) {
start := db.globalKey(folder, nil)
limit := db.globalKey(folder, []byte{0xff, 0xff, 0xff, 0xff})
@@ -452,7 +489,7 @@ nextFile:
}
}
func (db *dbInstance) listFolders() []string {
func (db *Instance) ListFolders() []string {
t := db.newReadOnlyTransaction()
defer t.close()
@@ -476,7 +513,7 @@ func (db *dbInstance) listFolders() []string {
return folders
}
func (db *dbInstance) dropFolder(folder []byte) {
func (db *Instance) dropFolder(folder []byte) {
t := db.newReadOnlyTransaction()
defer t.close()
@@ -501,7 +538,7 @@ func (db *dbInstance) dropFolder(folder []byte) {
dbi.Release()
}
func (db *dbInstance) checkGlobals(folder []byte, globalSize *sizeTracker) {
func (db *Instance) checkGlobals(folder []byte, globalSize *sizeTracker) {
t := db.newReadWriteTransaction()
defer t.close()
@@ -560,11 +597,11 @@ func (db *dbInstance) checkGlobals(folder []byte, globalSize *sizeTracker) {
// folder (64 bytes)
// device (32 bytes)
// name (variable size)
func (db *dbInstance) deviceKey(folder, device, file []byte) []byte {
func (db *Instance) deviceKey(folder, device, file []byte) []byte {
return db.deviceKeyInto(nil, folder, device, file)
}
func (db *dbInstance) deviceKeyInto(k []byte, folder, device, file []byte) []byte {
func (db *Instance) deviceKeyInto(k []byte, folder, device, file []byte) []byte {
reqLen := 1 + 64 + 32 + len(file)
if len(k) < reqLen {
k = make([]byte, reqLen)
@@ -579,11 +616,11 @@ func (db *dbInstance) deviceKeyInto(k []byte, folder, device, file []byte) []byt
return k[:reqLen]
}
func (db *dbInstance) deviceKeyName(key []byte) []byte {
func (db *Instance) deviceKeyName(key []byte) []byte {
return key[1+64+32:]
}
func (db *dbInstance) deviceKeyFolder(key []byte) []byte {
func (db *Instance) deviceKeyFolder(key []byte) []byte {
folder := key[1 : 1+64]
izero := bytes.IndexByte(folder, 0)
if izero < 0 {
@@ -592,7 +629,7 @@ func (db *dbInstance) deviceKeyFolder(key []byte) []byte {
return folder[:izero]
}
func (db *dbInstance) deviceKeyDevice(key []byte) []byte {
func (db *Instance) deviceKeyDevice(key []byte) []byte {
return key[1+64 : 1+64+32]
}
@@ -600,7 +637,7 @@ func (db *dbInstance) deviceKeyDevice(key []byte) []byte {
// keyTypeGlobal (1 byte)
// folder (64 bytes)
// name (variable size)
func (db *dbInstance) globalKey(folder, file []byte) []byte {
func (db *Instance) globalKey(folder, file []byte) []byte {
k := make([]byte, 1+64+len(file))
k[0] = KeyTypeGlobal
if len(folder) > 64 {
@@ -611,11 +648,11 @@ func (db *dbInstance) globalKey(folder, file []byte) []byte {
return k
}
func (db *dbInstance) globalKeyName(key []byte) []byte {
func (db *Instance) globalKeyName(key []byte) []byte {
return key[1+64:]
}
func (db *dbInstance) globalKeyFolder(key []byte) []byte {
func (db *Instance) globalKeyFolder(key []byte) []byte {
folder := key[1 : 1+64]
izero := bytes.IndexByte(folder, 0)
if izero < 0 {
@@ -635,3 +672,19 @@ func unmarshalTrunc(bs []byte, truncate bool) (FileIntf, error) {
err := tf.UnmarshalXDR(bs)
return tf, err
}
// A "better" version of leveldb's errors.IsCorrupted.
func leveldbIsCorrupted(err error) bool {
switch {
case err == nil:
return false
case errors.IsCorrupted(err):
return true
case strings.Contains(err.Error(), "corrupted"):
return true
}
return false
}

4
lib/db/leveldb_test.go
View File

@@ -16,7 +16,7 @@ func TestDeviceKey(t *testing.T) {
dev := []byte("device67890123456789012345678901")
name := []byte("name")
db := &dbInstance{}
db := &Instance{}
key := db.deviceKey(fld, dev, name)
@@ -38,7 +38,7 @@ func TestGlobalKey(t *testing.T) {
fld := []byte("folder6789012345678901234567890123456789012345678901234567890123")
name := []byte("name")
db := &dbInstance{}
db := &Instance{}
key := db.globalKey(fld, name)

6
lib/db/leveldb_transactions.go
View File

@@ -16,10 +16,10 @@ import (
// A readOnlyTransaction represents a database snapshot.
type readOnlyTransaction struct {
*leveldb.Snapshot
db *dbInstance
db *Instance
}
func (db *dbInstance) newReadOnlyTransaction() readOnlyTransaction {
func (db *Instance) newReadOnlyTransaction() readOnlyTransaction {
snap, err := db.GetSnapshot()
if err != nil {
panic(err)
@@ -46,7 +46,7 @@ type readWriteTransaction struct {
*leveldb.Batch
}
func (db *dbInstance) newReadWriteTransaction() readWriteTransaction {
func (db *Instance) newReadWriteTransaction() readWriteTransaction {
t := db.newReadOnlyTransaction()
return readWriteTransaction{
readOnlyTransaction: t,

4
lib/db/namespaced.go
View File

@@ -17,13 +17,13 @@ import (
// NamespacedKV is a simple key-value store using a specific namespace within
// a leveldb.
type NamespacedKV struct {
db *leveldb.DB
db *Instance
prefix []byte
}
// NewNamespacedKV returns a new NamespacedKV that lives in the namespace
// specified by the prefix.
func NewNamespacedKV(db *leveldb.DB, prefix string) *NamespacedKV {
func NewNamespacedKV(db *Instance, prefix string) *NamespacedKV {
return &NamespacedKV{
db: db,
prefix: []byte(prefix),

23
lib/db/namespaced_test.go
View File

@@ -9,16 +9,10 @@ package db
import (
"testing"
"time"
"github.com/syndtr/goleveldb/leveldb"
"github.com/syndtr/goleveldb/leveldb/storage"
)
func TestNamespacedInt(t *testing.T) {
ldb, err := leveldb.Open(storage.NewMemStorage(), nil)
if err != nil {
t.Fatal(err)
}
ldb := OpenMemory()
n1 := NewNamespacedKV(ldb, "foo")
n2 := NewNamespacedKV(ldb, "bar")
@@ -53,10 +47,7 @@ func TestNamespacedInt(t *testing.T) {
}
func TestNamespacedTime(t *testing.T) {
ldb, err := leveldb.Open(storage.NewMemStorage(), nil)
if err != nil {
t.Fatal(err)
}
ldb := OpenMemory()
n1 := NewNamespacedKV(ldb, "foo")
@@ -73,10 +64,7 @@ func TestNamespacedTime(t *testing.T) {
}
func TestNamespacedString(t *testing.T) {
ldb, err := leveldb.Open(storage.NewMemStorage(), nil)
if err != nil {
t.Fatal(err)
}
ldb := OpenMemory()
n1 := NewNamespacedKV(ldb, "foo")
@@ -92,10 +80,7 @@ func TestNamespacedString(t *testing.T) {
}
func TestNamespacedReset(t *testing.T) {
ldb, err := leveldb.Open(storage.NewMemStorage(), nil)
if err != nil {
t.Fatal(err)
}
ldb := OpenMemory()
n1 := NewNamespacedKV(ldb, "foo")

18
lib/db/set.go
View File

@@ -18,14 +18,13 @@ import (
"github.com/syncthing/syncthing/lib/osutil"
"github.com/syncthing/syncthing/lib/protocol"
"github.com/syncthing/syncthing/lib/sync"
"github.com/syndtr/goleveldb/leveldb"
)
type FileSet struct {
localVersion map[protocol.DeviceID]int64
mutex sync.Mutex
folder string
db *dbInstance
db *Instance
blockmap *BlockMap
localSize sizeTracker
globalSize sizeTracker
@@ -93,11 +92,11 @@ func (s *sizeTracker) Size() (files, deleted int, bytes int64) {
return s.files, s.deleted, s.bytes
}
func NewFileSet(folder string, db *leveldb.DB) *FileSet {
func NewFileSet(folder string, db *Instance) *FileSet {
var s = FileSet{
localVersion: make(map[protocol.DeviceID]int64),
folder: folder,
db: newDBInstance(db),
db: db,
blockmap: NewBlockMap(db, folder),
mutex: sync.NewMutex(),
}
@@ -239,17 +238,10 @@ 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 {
i := newDBInstance(db)
return i.listFolders()
}
// DropFolder clears out all information related to the given folder from the
// database.
func DropFolder(db *leveldb.DB, folder string) {
i := newDBInstance(db)
i.dropFolder([]byte(folder))
func DropFolder(db *Instance, folder string) {
db.dropFolder([]byte(folder))
bm := &BlockMap{
db: db,
folder: folder,

57
lib/db/set_test.go
View File

@@ -15,8 +15,6 @@ import (
"github.com/syncthing/syncthing/lib/db"
"github.com/syncthing/syncthing/lib/protocol"
"github.com/syndtr/goleveldb/leveldb"
"github.com/syndtr/goleveldb/leveldb/storage"
)
var remoteDevice0, remoteDevice1 protocol.DeviceID
@@ -96,11 +94,7 @@ func (l fileList) String() string {
}
func TestGlobalSet(t *testing.T) {
ldb, err := leveldb.Open(storage.NewMemStorage(), nil)
if err != nil {
t.Fatal(err)
}
ldb := db.OpenMemory()
m := db.NewFileSet("test", ldb)
@@ -303,10 +297,7 @@ func TestGlobalSet(t *testing.T) {
}
func TestNeedWithInvalid(t *testing.T) {
ldb, err := leveldb.Open(storage.NewMemStorage(), nil)
if err != nil {
t.Fatal(err)
}
ldb := db.OpenMemory()
s := db.NewFileSet("test", ldb)
@@ -343,10 +334,7 @@ func TestNeedWithInvalid(t *testing.T) {
}
func TestUpdateToInvalid(t *testing.T) {
ldb, err := leveldb.Open(storage.NewMemStorage(), nil)
if err != nil {
t.Fatal(err)
}
ldb := db.OpenMemory()
s := db.NewFileSet("test", ldb)
@@ -378,10 +366,7 @@ func TestUpdateToInvalid(t *testing.T) {
}
func TestInvalidAvailability(t *testing.T) {
ldb, err := leveldb.Open(storage.NewMemStorage(), nil)
if err != nil {
t.Fatal(err)
}
ldb := db.OpenMemory()
s := db.NewFileSet("test", ldb)
@@ -419,10 +404,7 @@ func TestInvalidAvailability(t *testing.T) {
}
func TestGlobalReset(t *testing.T) {
ldb, err := leveldb.Open(storage.NewMemStorage(), nil)
if err != nil {
t.Fatal(err)
}
ldb := db.OpenMemory()
m := db.NewFileSet("test", ldb)
@@ -460,10 +442,7 @@ func TestGlobalReset(t *testing.T) {
}
func TestNeed(t *testing.T) {
ldb, err := leveldb.Open(storage.NewMemStorage(), nil)
if err != nil {
t.Fatal(err)
}
ldb := db.OpenMemory()
m := db.NewFileSet("test", ldb)
@@ -501,10 +480,7 @@ func TestNeed(t *testing.T) {
}
func TestLocalVersion(t *testing.T) {
ldb, err := leveldb.Open(storage.NewMemStorage(), nil)
if err != nil {
t.Fatal(err)
}
ldb := db.OpenMemory()
m := db.NewFileSet("test", ldb)
@@ -534,10 +510,7 @@ func TestLocalVersion(t *testing.T) {
}
func TestListDropFolder(t *testing.T) {
ldb, err := leveldb.Open(storage.NewMemStorage(), nil)
if err != nil {
t.Fatal(err)
}
ldb := db.OpenMemory()
s0 := db.NewFileSet("test0", ldb)
local1 := []protocol.FileInfo{
@@ -558,7 +531,7 @@ func TestListDropFolder(t *testing.T) {
// Check that we have both folders and their data is in the global list
expectedFolderList := []string{"test0", "test1"}
if actualFolderList := db.ListFolders(ldb); !reflect.DeepEqual(actualFolderList, expectedFolderList) {
if actualFolderList := ldb.ListFolders(); !reflect.DeepEqual(actualFolderList, expectedFolderList) {
t.Fatalf("FolderList mismatch\nE: %v\nA: %v", expectedFolderList, actualFolderList)
}
if l := len(globalList(s0)); l != 3 {
@@ -573,7 +546,7 @@ func TestListDropFolder(t *testing.T) {
db.DropFolder(ldb, "test1")
expectedFolderList = []string{"test0"}
if actualFolderList := db.ListFolders(ldb); !reflect.DeepEqual(actualFolderList, expectedFolderList) {
if actualFolderList := ldb.ListFolders(); !reflect.DeepEqual(actualFolderList, expectedFolderList) {
t.Fatalf("FolderList mismatch\nE: %v\nA: %v", expectedFolderList, actualFolderList)
}
if l := len(globalList(s0)); l != 3 {
@@ -585,10 +558,7 @@ func TestListDropFolder(t *testing.T) {
}
func TestGlobalNeedWithInvalid(t *testing.T) {
ldb, err := leveldb.Open(storage.NewMemStorage(), nil)
if err != nil {
t.Fatal(err)
}
ldb := db.OpenMemory()
s := db.NewFileSet("test1", ldb)
@@ -625,10 +595,7 @@ func TestGlobalNeedWithInvalid(t *testing.T) {
}
func TestLongPath(t *testing.T) {
ldb, err := leveldb.Open(storage.NewMemStorage(), nil)
if err != nil {
t.Fatal(err)
}
ldb := db.OpenMemory()
s := db.NewFileSet("test", ldb)

4
lib/db/virtualmtime.go
View File

@@ -9,8 +9,6 @@ package db
import (
"fmt"
"time"
"github.com/syndtr/goleveldb/leveldb"
)
// This type encapsulates a repository of mtimes for platforms where file mtimes
@@ -25,7 +23,7 @@ type VirtualMtimeRepo struct {
ns *NamespacedKV
}
func NewVirtualMtimeRepo(ldb *leveldb.DB, folder string) *VirtualMtimeRepo {
func NewVirtualMtimeRepo(ldb *Instance, folder string) *VirtualMtimeRepo {
prefix := string(KeyTypeVirtualMtime) + folder
return &VirtualMtimeRepo{

8
lib/db/virtualmtime_test.go
View File

@@ -9,16 +9,10 @@ package db
import (
"testing"
"time"
"github.com/syndtr/goleveldb/leveldb"
"github.com/syndtr/goleveldb/leveldb/storage"
)
func TestVirtualMtimeRepo(t *testing.T) {
ldb, err := leveldb.Open(storage.NewMemStorage(), nil)
if err != nil {
t.Fatal(err)
}
ldb := OpenMemory()
// A few repos so we can ensure they don't pollute each other
repo1 := NewVirtualMtimeRepo(ldb, "folder1")