Merge pull request #84 from mavenugo/b-ts

Cherry-pick 2b31300cef639e8c54aaaf52a6f57d1cf11883a3 into master (Synchronize boltbd libkv apis)

libnetwork/internal/kvstore/boltdb/boltdb.go

@@ -6,6 +6,7 @@ import (
 	"errors"
 	"os"
 	"path/filepath"
+	"sync"
 	"sync/atomic"
 	"time"
 
@@ -41,6 +42,7 @@ type BoltDB struct {
 	// PersistConnection flag provides an option to override ths behavior.
 	// ie: open the connection in New and use it till Close is called.
 	PersistConnection bool
+	sync.Mutex
 }
 
 const (
@@ -128,6 +130,8 @@ func (b *BoltDB) Get(key string) (*store.KVPair, error) {
 		db  *bolt.DB
 		err error
 	)
+	b.Lock()
+	defer b.Unlock()
 
 	if db, err = b.getDBhandle(); err != nil {
 		return nil, err
@@ -167,6 +171,9 @@ func (b *BoltDB) Put(key string, value []byte, opts *store.WriteOptions) error {
 		db      *bolt.DB
 		err     error
 	)
+	b.Lock()
+	defer b.Unlock()
+
 	dbval := make([]byte, libkvmetadatalen)
 
 	if db, err = b.getDBhandle(); err != nil {
@@ -199,6 +206,9 @@ func (b *BoltDB) Delete(key string) error {
 		db  *bolt.DB
 		err error
 	)
+	b.Lock()
+	defer b.Unlock()
+
 	if db, err = b.getDBhandle(); err != nil {
 		return err
 	}
@@ -222,6 +232,8 @@ func (b *BoltDB) Exists(key string) (bool, error) {
 		db  *bolt.DB
 		err error
 	)
+	b.Lock()
+	defer b.Unlock()
 
 	if db, err = b.getDBhandle(); err != nil {
 		return false, err
@@ -251,6 +263,9 @@ func (b *BoltDB) List(keyPrefix string) ([]*store.KVPair, error) {
 		db  *bolt.DB
 		err error
 	)
+	b.Lock()
+	defer b.Unlock()
+
 	kv := []*store.KVPair{}
 
 	if db, err = b.getDBhandle(); err != nil {
@@ -297,6 +312,8 @@ func (b *BoltDB) AtomicDelete(key string, previous *store.KVPair) (bool, error)
 		db  *bolt.DB
 		err error
 	)
+	b.Lock()
+	defer b.Unlock()
 
 	if previous == nil {
 		return false, store.ErrPreviousNotSpecified
@@ -335,6 +352,9 @@ func (b *BoltDB) AtomicPut(key string, value []byte, previous *store.KVPair, opt
 		db      *bolt.DB
 		err     error
 	)
+	b.Lock()
+	defer b.Unlock()
+
 	dbval := make([]byte, libkvmetadatalen)
 
 	if db, err = b.getDBhandle(); err != nil {
@@ -389,6 +409,9 @@ func (b *BoltDB) AtomicPut(key string, value []byte, previous *store.KVPair, opt
 
 // Close the db connection to the BoltDB
 func (b *BoltDB) Close() {
+	b.Lock()
+	defer b.Unlock()
+
 	if !b.PersistConnection {
 		b.reset()
 	} else {
@@ -403,6 +426,9 @@ func (b *BoltDB) DeleteTree(keyPrefix string) error {
 		db  *bolt.DB
 		err error
 	)
+	b.Lock()
+	defer b.Unlock()
+
 	if db, err = b.getDBhandle(); err != nil {
 		return err
 	}