Merge pull request #563 from jbenet/hotfix/fsrepo-leveldb-multi-open

hotfix(FSRepo) allow multiple goroutines to call 

repo/fsrepo/assert/assert.go

+package assert
+
+import "testing"
+
+func Nil(err error, t *testing.T, msgs ...string) {
+	if err != nil {
+		t.Fatal(msgs, "error:", err)
+	}
+}
+
+func True(v bool, t *testing.T, msgs ...string) {
+	if !v {
+		t.Fatal(msgs)
+	}
+}
+
+func Err(err error, t *testing.T, msgs ...string) {
+	if err == nil {
+		t.Fatal(msgs, "error:", err)
+	}
+}

repo/fsrepo/component/datastore.go

 package component
 
 import (
+	"errors"
+	"sync"
+
 	datastore "github.com/jbenet/go-ipfs/Godeps/_workspace/src/github.com/jbenet/go-datastore"
 	levelds "github.com/jbenet/go-ipfs/Godeps/_workspace/src/github.com/jbenet/go-datastore/leveldb"
 	ldbopts "github.com/jbenet/go-ipfs/Godeps/_workspace/src/github.com/syndtr/goleveldb/leveldb/opt"
 	config "github.com/jbenet/go-ipfs/repo/config"
+	counter "github.com/jbenet/go-ipfs/repo/fsrepo/counter"
 	dir "github.com/jbenet/go-ipfs/repo/fsrepo/dir"
 	util "github.com/jbenet/go-ipfs/util"
 	ds2 "github.com/jbenet/go-ipfs/util/datastore2"
 	debugerror "github.com/jbenet/go-ipfs/util/debugerror"
 )
 
-var _ Component = &DatastoreComponent{}
-var _ Initializer = InitDatastoreComponent
-var _ InitializationChecker = DatastoreComponentIsInitialized
+var (
+	_ Component             = &DatastoreComponent{}
+	_ Initializer           = InitDatastoreComponent
+	_ InitializationChecker = DatastoreComponentIsInitialized
+
+	dsLock         sync.Mutex // protects openersCounter and datastores
+	openersCounter *counter.Openers
+	datastores     map[string]ds2.ThreadSafeDatastoreCloser
+)
+
+func init() {
+	openersCounter = counter.NewOpenersCounter()
+	datastores = make(map[string]ds2.ThreadSafeDatastoreCloser)
+}
 
 func InitDatastoreComponent(path string, conf *config.Config) error {
 	// The actual datastore contents are initialized lazily when Opened.
@@ -41,24 +56,57 @@ func DatastoreComponentIsInitialized(path string) bool {
 }
 
 // DatastoreComponent abstracts the datastore component of the FSRepo.
-// NB: create with makeDatastoreComponent function.
 type DatastoreComponent struct {
-	path string
-	ds   ds2.ThreadSafeDatastoreCloser
+	path string                        // required
+	ds   ds2.ThreadSafeDatastoreCloser // assigned when repo is opened
 }
 
+func (dsc *DatastoreComponent) SetPath(p string)                         { dsc.path = p }
+func (dsc *DatastoreComponent) Datastore() datastore.ThreadSafeDatastore { return dsc.ds }
+
 // Open returns an error if the config file is not present.
 func (dsc *DatastoreComponent) Open() error {
-	ds, err := levelds.NewDatastore(dsc.path, &levelds.Options{
-		Compression: ldbopts.NoCompression,
-	})
-	if err != nil {
-		return err
+
+	dsLock.Lock()
+	defer dsLock.Unlock()
+
+	// if no other goroutines have the datastore Open, initialize it and assign
+	// it to the package-scoped map for the goroutines that follow.
+	if openersCounter.NumOpeners(dsc.path) == 0 {
+		ds, err := levelds.NewDatastore(dsc.path, &levelds.Options{
+			Compression: ldbopts.NoCompression,
+		})
+		if err != nil {
+			return errors.New("unable to open leveldb datastore")
+		}
+		datastores[dsc.path] = ds
+	}
+
+	// get the datastore from the package-scoped map and record self as an
+	// opener.
+	ds, dsIsPresent := datastores[dsc.path]
+	if !dsIsPresent {
+		// This indicates a programmer error has occurred.
+		return errors.New("datastore should be available, but it isn't")
 	}
 	dsc.ds = ds
+	openersCounter.AddOpener(dsc.path) // only after success
 	return nil
 }
 
-func (dsc *DatastoreComponent) Close() error                             { return dsc.ds.Close() }
-func (dsc *DatastoreComponent) SetPath(p string)                         { dsc.path = p }
-func (dsc *DatastoreComponent) Datastore() datastore.ThreadSafeDatastore { return dsc.ds }
+func (dsc *DatastoreComponent) Close() error {
+
+	dsLock.Lock()
+	defer dsLock.Unlock()
+
+	// decrement the Opener count. if this goroutine is the last, also close
+	// the underlying datastore (and remove its reference from the map)
+
+	openersCounter.RemoveOpener(dsc.path)
+
+	if openersCounter.NumOpeners(dsc.path) == 0 {
+		delete(datastores, dsc.path) // remove the reference
+		return dsc.ds.Close()
+	}
+	return nil
+}

repo/fsrepo/component/datastore_test.go

+package component
+
+import (
+	"io/ioutil"
+	"path/filepath"
+	"testing"
+
+	"github.com/jbenet/go-ipfs/repo/fsrepo/assert"
+)
+
+// swap arg order
+func testRepoPath(t *testing.T, path ...string) string {
+	name, err := ioutil.TempDir("", filepath.Join(path...))
+	if err != nil {
+		t.Fatal(err)
+	}
+	return name
+}
+
+func TestOpenMoreThanOnceInSameProcess(t *testing.T) {
+	t.Parallel()
+	path := testRepoPath(t)
+	dsc1 := DatastoreComponent{path: path}
+	dsc2 := DatastoreComponent{path: path}
+	assert.Nil(dsc1.Open(), t, "first repo should open successfully")
+	assert.Nil(dsc2.Open(), t, "second repo should open successfully")
+
+	assert.Nil(dsc1.Close(), t)
+	assert.Nil(dsc2.Close(), t)
+}

repo/fsrepo/fsrepo_test.go

@@ -7,6 +7,7 @@ import (
 
 	datastore "github.com/jbenet/go-ipfs/Godeps/_workspace/src/github.com/jbenet/go-datastore"
 	"github.com/jbenet/go-ipfs/repo/config"
+	"github.com/jbenet/go-ipfs/repo/fsrepo/assert"
 )
 
 // swap arg order
@@ -22,35 +23,35 @@ func TestInitIdempotence(t *testing.T) {
 	t.Parallel()
 	path := testRepoPath("", t)
 	for i := 0; i < 10; i++ {
-		AssertNil(Init(path, &config.Config{}), t, "multiple calls to init should succeed")
+		assert.Nil(Init(path, &config.Config{}), t, "multiple calls to init should succeed")
 	}
 }
 
 func TestRemove(t *testing.T) {
 	t.Parallel()
 	path := testRepoPath("foo", t)
-	AssertNil(Remove(path), t, "should be able to remove after closed")
+	assert.Nil(Remove(path), t, "should be able to remove after closed")
 }
 
 func TestCannotRemoveIfOpen(t *testing.T) {
 	t.Parallel()
 	path := testRepoPath("TestCannotRemoveIfOpen", t)
-	AssertNil(Init(path, &config.Config{}), t, "should initialize successfully")
+	assert.Nil(Init(path, &config.Config{}), t, "should initialize successfully")
 	r := At(path)
-	AssertNil(r.Open(), t)
-	AssertErr(Remove(path), t, "should not be able to remove while open")
-	AssertNil(r.Close(), t)
-	AssertNil(Remove(path), t, "should be able to remove after closed")
+	assert.Nil(r.Open(), t)
+	assert.Err(Remove(path), t, "should not be able to remove while open")
+	assert.Nil(r.Close(), t)
+	assert.Nil(Remove(path), t, "should be able to remove after closed")
 }
 
 func TestCannotBeReopened(t *testing.T) {
 	t.Parallel()
 	path := testRepoPath("", t)
-	AssertNil(Init(path, &config.Config{}), t)
+	assert.Nil(Init(path, &config.Config{}), t)
 	r := At(path)
-	AssertNil(r.Open(), t)
-	AssertNil(r.Close(), t)
-	AssertErr(r.Open(), t, "shouldn't be possible to re-open the repo")
+	assert.Nil(r.Open(), t)
+	assert.Nil(r.Close(), t)
+	assert.Err(r.Open(), t, "shouldn't be possible to re-open the repo")
 
 	// mutable state is the enemy. Take Close() as an opportunity to reduce
 	// entropy. Callers ought to start fresh with a new handle by calling `At`.
@@ -62,83 +63,79 @@ func TestCanManageReposIndependently(t *testing.T) {
 	pathB := testRepoPath("b", t)
 
 	t.Log("initialize two repos")
-	AssertNil(Init(pathA, &config.Config{}), t, "a", "should initialize successfully")
-	AssertNil(Init(pathB, &config.Config{}), t, "b", "should initialize successfully")
+	assert.Nil(Init(pathA, &config.Config{}), t, "a", "should initialize successfully")
+	assert.Nil(Init(pathB, &config.Config{}), t, "b", "should initialize successfully")
 
 	t.Log("ensure repos initialized")
-	Assert(IsInitialized(pathA), t, "a should be initialized")
-	Assert(IsInitialized(pathB), t, "b should be initialized")
+	assert.True(IsInitialized(pathA), t, "a should be initialized")
+	assert.True(IsInitialized(pathB), t, "b should be initialized")
 
 	t.Log("open the two repos")
 	repoA := At(pathA)
 	repoB := At(pathB)
-	AssertNil(repoA.Open(), t, "a")
-	AssertNil(repoB.Open(), t, "b")
+	assert.Nil(repoA.Open(), t, "a")
+	assert.Nil(repoB.Open(), t, "b")
 
 	t.Log("close and remove b while a is open")
-	AssertNil(repoB.Close(), t, "close b")
-	AssertNil(Remove(pathB), t, "remove b")
+	assert.Nil(repoB.Close(), t, "close b")
+	assert.Nil(Remove(pathB), t, "remove b")
 
 	t.Log("close and remove a")
-	AssertNil(repoA.Close(), t)
-	AssertNil(Remove(pathA), t)
+	assert.Nil(repoA.Close(), t)
+	assert.Nil(Remove(pathA), t)
 }
 
 func TestDatastoreGetNotAllowedAfterClose(t *testing.T) {
 	t.Parallel()
 	path := testRepoPath("test", t)
 
-	Assert(!IsInitialized(path), t, "should NOT be initialized")
-	AssertNil(Init(path, &config.Config{}), t, "should initialize successfully")
+	assert.True(!IsInitialized(path), t, "should NOT be initialized")
+	assert.Nil(Init(path, &config.Config{}), t, "should initialize successfully")
 	r := At(path)
-	AssertNil(r.Open(), t, "should open successfully")
+	assert.Nil(r.Open(), t, "should open successfully")
 
 	k := "key"
 	data := []byte(k)
-	AssertNil(r.Datastore().Put(datastore.NewKey(k), data), t, "Put should be successful")
+	assert.Nil(r.Datastore().Put(datastore.NewKey(k), data), t, "Put should be successful")
 
-	AssertNil(r.Close(), t)
+	assert.Nil(r.Close(), t)
 	_, err := r.Datastore().Get(datastore.NewKey(k))
-	AssertErr(err, t, "after closer, Get should be fail")
+	assert.Err(err, t, "after closer, Get should be fail")
 }
 
 func TestDatastorePersistsFromRepoToRepo(t *testing.T) {
 	t.Parallel()
 	path := testRepoPath("test", t)
 
-	AssertNil(Init(path, &config.Config{}), t)
+	assert.Nil(Init(path, &config.Config{}), t)
 	r1 := At(path)
-	AssertNil(r1.Open(), t)
+	assert.Nil(r1.Open(), t)
 
 	k := "key"
 	expected := []byte(k)
-	AssertNil(r1.Datastore().Put(datastore.NewKey(k), expected), t, "using first repo, Put should be successful")
-	AssertNil(r1.Close(), t)
+	assert.Nil(r1.Datastore().Put(datastore.NewKey(k), expected), t, "using first repo, Put should be successful")
+	assert.Nil(r1.Close(), t)
 
 	r2 := At(path)
-	AssertNil(r2.Open(), t)
+	assert.Nil(r2.Open(), t)
 	v, err := r2.Datastore().Get(datastore.NewKey(k))
-	AssertNil(err, t, "using second repo, Get should be successful")
+	assert.Nil(err, t, "using second repo, Get should be successful")
 	actual, ok := v.([]byte)
-	Assert(ok, t, "value should be the []byte from r1's Put")
-	AssertNil(r2.Close(), t)
-	Assert(bytes.Compare(expected, actual) == 0, t, "data should match")
+	assert.True(ok, t, "value should be the []byte from r1's Put")
+	assert.Nil(r2.Close(), t)
+	assert.True(bytes.Compare(expected, actual) == 0, t, "data should match")
 }
 
-func AssertNil(err error, t *testing.T, msgs ...string) {
-	if err != nil {
-		t.Fatal(msgs, "error:", err)
-	}
-}
+func TestOpenMoreThanOnceInSameProcess(t *testing.T) {
+	t.Parallel()
+	path := testRepoPath("", t)
+	assert.Nil(Init(path, &config.Config{}), t)
 
-func Assert(v bool, t *testing.T, msgs ...string) {
-	if !v {
-		t.Fatal(msgs)
-	}
-}
+	r1 := At(path)
+	r2 := At(path)
+	assert.Nil(r1.Open(), t, "first repo should open successfully")
+	assert.Nil(r2.Open(), t, "second repo should open successfully")
 
-func AssertErr(err error, t *testing.T, msgs ...string) {
-	if err == nil {
-		t.Fatal(msgs, "error:", err)
-	}
+	assert.Nil(r1.Close(), t)
+	assert.Nil(r2.Close(), t)
 }