Revert "Share libkv store handles across datastore handles"
This commit is contained in:
parent
305ab4a6ad
commit
284c9cd0f5
|
@ -5,7 +5,6 @@ import (
|
||||||
"log"
|
"log"
|
||||||
"reflect"
|
"reflect"
|
||||||
"strings"
|
"strings"
|
||||||
"sync"
|
|
||||||
|
|
||||||
"github.com/docker/libkv"
|
"github.com/docker/libkv"
|
||||||
"github.com/docker/libkv/store"
|
"github.com/docker/libkv/store"
|
||||||
|
@ -56,7 +55,6 @@ type datastore struct {
|
||||||
scope string
|
scope string
|
||||||
store store.Store
|
store store.Store
|
||||||
cache *cache
|
cache *cache
|
||||||
cfg ScopeCfg
|
|
||||||
}
|
}
|
||||||
|
|
||||||
// KVObject is Key/Value interface used by objects to be part of the DataStore
|
// KVObject is Key/Value interface used by objects to be part of the DataStore
|
||||||
|
@ -105,12 +103,6 @@ type ScopeClientCfg struct {
|
||||||
Config *store.Config
|
Config *store.Config
|
||||||
}
|
}
|
||||||
|
|
||||||
type storeTableData struct {
|
|
||||||
refCnt int
|
|
||||||
store store.Store
|
|
||||||
once sync.Once
|
|
||||||
}
|
|
||||||
|
|
||||||
const (
|
const (
|
||||||
// LocalScope indicates to store the KV object in local datastore such as boltdb
|
// LocalScope indicates to store the KV object in local datastore such as boltdb
|
||||||
LocalScope = "local"
|
LocalScope = "local"
|
||||||
|
@ -128,8 +120,6 @@ const (
|
||||||
|
|
||||||
var (
|
var (
|
||||||
defaultScopes = makeDefaultScopes()
|
defaultScopes = makeDefaultScopes()
|
||||||
storeLock sync.Mutex
|
|
||||||
storeTable = make(map[ScopeCfg]*storeTableData)
|
|
||||||
)
|
)
|
||||||
|
|
||||||
func makeDefaultScopes() map[string]*ScopeCfg {
|
func makeDefaultScopes() map[string]*ScopeCfg {
|
||||||
|
@ -198,7 +188,11 @@ func ParseKey(key string) ([]string, error) {
|
||||||
}
|
}
|
||||||
|
|
||||||
// newClient used to connect to KV Store
|
// newClient used to connect to KV Store
|
||||||
func newClient(scope string, kv string, addrs string, config *store.Config, cached bool) (*datastore, error) {
|
func newClient(scope string, kv string, addrs string, config *store.Config, cached bool) (DataStore, error) {
|
||||||
|
if cached && scope != LocalScope {
|
||||||
|
return nil, fmt.Errorf("caching supported only for scope %s", LocalScope)
|
||||||
|
}
|
||||||
|
|
||||||
if config == nil {
|
if config == nil {
|
||||||
config = &store.Config{}
|
config = &store.Config{}
|
||||||
}
|
}
|
||||||
|
@ -217,76 +211,24 @@ func newClient(scope string, kv string, addrs string, config *store.Config, cach
|
||||||
|
|
||||||
// NewDataStore creates a new instance of LibKV data store
|
// NewDataStore creates a new instance of LibKV data store
|
||||||
func NewDataStore(scope string, cfg *ScopeCfg) (DataStore, error) {
|
func NewDataStore(scope string, cfg *ScopeCfg) (DataStore, error) {
|
||||||
var (
|
if cfg == nil || cfg.Client.Provider == "" || cfg.Client.Address == "" {
|
||||||
err error
|
c, ok := defaultScopes[scope]
|
||||||
ds *datastore
|
if !ok || c.Client.Provider == "" || c.Client.Address == "" {
|
||||||
)
|
return nil, fmt.Errorf("unexpected scope %s without configuration passed", scope)
|
||||||
|
|
||||||
if !cfg.IsValid() {
|
|
||||||
return nil, fmt.Errorf("invalid datastore configuration passed for scope %s", scope)
|
|
||||||
}
|
|
||||||
|
|
||||||
storeLock.Lock()
|
|
||||||
sdata, ok := storeTable[*cfg]
|
|
||||||
if ok {
|
|
||||||
sdata.refCnt++
|
|
||||||
// If sdata already has a store nothing to do. Just
|
|
||||||
// create a datastore handle using it and return with
|
|
||||||
// that.
|
|
||||||
if sdata.store != nil {
|
|
||||||
storeLock.Unlock()
|
|
||||||
return &datastore{scope: scope, cfg: *cfg, store: sdata.store}, nil
|
|
||||||
}
|
|
||||||
} else {
|
|
||||||
// If sdata is not present create one and add ito
|
|
||||||
// storeTable while holding the lock.
|
|
||||||
sdata = &storeTableData{refCnt: 1}
|
|
||||||
storeTable[*cfg] = sdata
|
|
||||||
}
|
|
||||||
storeLock.Unlock()
|
|
||||||
|
|
||||||
// We come here either because:
|
|
||||||
//
|
|
||||||
// 1. We just created the store table data OR
|
|
||||||
// 2. We picked up the store table data from table but store was not initialized.
|
|
||||||
//
|
|
||||||
// In both cases the once function will ensure the store
|
|
||||||
// initialization happens exactly once
|
|
||||||
sdata.once.Do(func() {
|
|
||||||
ds, err = newClient(scope, cfg.Client.Provider, cfg.Client.Address, cfg.Client.Config, scope == LocalScope)
|
|
||||||
if err != nil {
|
|
||||||
return
|
|
||||||
}
|
}
|
||||||
|
|
||||||
ds.cfg = *cfg
|
cfg = c
|
||||||
sdata.store = ds.store
|
|
||||||
})
|
|
||||||
|
|
||||||
if err != nil {
|
|
||||||
return nil, err
|
|
||||||
}
|
}
|
||||||
|
|
||||||
return ds, nil
|
var cached bool
|
||||||
|
if scope == LocalScope {
|
||||||
|
cached = true
|
||||||
|
}
|
||||||
|
|
||||||
|
return newClient(scope, cfg.Client.Provider, cfg.Client.Address, cfg.Client.Config, cached)
|
||||||
}
|
}
|
||||||
|
|
||||||
func (ds *datastore) Close() {
|
func (ds *datastore) Close() {
|
||||||
storeLock.Lock()
|
|
||||||
sdata := storeTable[ds.cfg]
|
|
||||||
|
|
||||||
if sdata == nil {
|
|
||||||
storeLock.Unlock()
|
|
||||||
return
|
|
||||||
}
|
|
||||||
|
|
||||||
sdata.refCnt--
|
|
||||||
if sdata.refCnt > 0 {
|
|
||||||
storeLock.Unlock()
|
|
||||||
return
|
|
||||||
}
|
|
||||||
|
|
||||||
delete(storeTable, ds.cfg)
|
|
||||||
storeLock.Unlock()
|
|
||||||
|
|
||||||
ds.store.Close()
|
ds.store.Close()
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|
|
@ -42,17 +42,9 @@ func TestBoltdbBackend(t *testing.T) {
|
||||||
|
|
||||||
func testLocalBackend(t *testing.T, provider, url string, storeConfig *store.Config) {
|
func testLocalBackend(t *testing.T, provider, url string, storeConfig *store.Config) {
|
||||||
cfgOptions := []config.Option{}
|
cfgOptions := []config.Option{}
|
||||||
if provider != "" {
|
cfgOptions = append(cfgOptions, config.OptionLocalKVProvider(provider))
|
||||||
cfgOptions = append(cfgOptions, config.OptionLocalKVProvider(provider))
|
cfgOptions = append(cfgOptions, config.OptionLocalKVProviderURL(url))
|
||||||
}
|
cfgOptions = append(cfgOptions, config.OptionLocalKVProviderConfig(storeConfig))
|
||||||
|
|
||||||
if url != "" {
|
|
||||||
cfgOptions = append(cfgOptions, config.OptionLocalKVProviderURL(url))
|
|
||||||
}
|
|
||||||
|
|
||||||
if storeConfig != nil {
|
|
||||||
cfgOptions = append(cfgOptions, config.OptionLocalKVProviderConfig(storeConfig))
|
|
||||||
}
|
|
||||||
|
|
||||||
driverOptions := options.Generic{}
|
driverOptions := options.Generic{}
|
||||||
genericOption := make(map[string]interface{})
|
genericOption := make(map[string]interface{})
|
||||||
|
@ -78,7 +70,7 @@ func testLocalBackend(t *testing.T, provider, url string, storeConfig *store.Con
|
||||||
if exists, err := store.Exists(datastore.Key([]string{datastore.EndpointKeyPrefix, string(nw.ID()), string(ep.ID())}...)); !exists || err != nil {
|
if exists, err := store.Exists(datastore.Key([]string{datastore.EndpointKeyPrefix, string(nw.ID()), string(ep.ID())}...)); !exists || err != nil {
|
||||||
t.Fatalf("Endpoint key should have been created.")
|
t.Fatalf("Endpoint key should have been created.")
|
||||||
}
|
}
|
||||||
ctrl.(*controller).getStore(datastore.LocalScope).Close()
|
store.Close()
|
||||||
|
|
||||||
// test restore of local store
|
// test restore of local store
|
||||||
ctrl, err = New(cfgOptions...)
|
ctrl, err = New(cfgOptions...)
|
||||||
|
|
Loading…
Reference in New Issue