Merge pull request ethereum#137 from rauljordan/shard-storage

Use Persistent Key-Val DB for Shard Node Collation Storage

sharding/database/database.go

@@ -0,0 +1,23 @@
+package database
+
+import (
+	"path/filepath"
+
+	"github.com/ethereum/go-ethereum/ethdb"
+)
+
+// ShardBackend defines an interface for a shardDB's necessary method
+// signatures.
+type ShardBackend interface {
+	Get(k []byte) ([]byte, error)
+	Has(k []byte) (bool, error)
+	Put(k []byte, val []byte) error
+	Delete(k []byte) error
+}
+
+// NewShardDB initializes a shardDB that writes to local disk.
+func NewShardDB(dataDir string, name string) (ShardBackend, error) {
+	// Uses default cache and handles values.
+	// TODO: allow these arguments to be set based on cli context.
+	return ethdb.NewLDBDatabase(filepath.Join(dataDir, name), 16, 16)
+}

sharding/database/database_test.go

@@ -0,0 +1,92 @@
+package database
+
+import (
+	"strconv"
+	"testing"
+)
+
+var db ShardBackend
+
+func init() {
+	shardDB, err := NewShardDB("/tmp/datadir", "shardchaindata")
+	if err != nil {
+		panic(err)
+	}
+	db = shardDB
+}
+
+// Testing the concurrency of the shardDB with multiple processes attempting to write.
+func Test_DBConcurrent(t *testing.T) {
+	for i := 0; i < 100; i++ {
+		go func(val string) {
+			if err := db.Put([]byte("ralph merkle"), []byte(val)); err != nil {
+				t.Errorf("could not save value in db: %v", err)
+			}
+		}(strconv.Itoa(i))
+	}
+}
+
+func Test_DBPut(t *testing.T) {
+	if err := db.Put([]byte("ralph merkle"), []byte{1, 2, 3}); err != nil {
+		t.Errorf("could not save value in db: %v", err)
+	}
+}
+
+func Test_DBHas(t *testing.T) {
+	key := []byte("ralph merkle")
+
+	if err := db.Put(key, []byte{1, 2, 3}); err != nil {
+		t.Fatalf("could not save value in db: %v", err)
+	}
+
+	has, err := db.Has(key)
+	if err != nil {
+		t.Errorf("could not check if db has key: %v", err)
+	}
+	if !has {
+		t.Errorf("db should have key: %v", key)
+	}
+
+	key2 := []byte{}
+	has2, err := db.Has(key2)
+	if err != nil {
+		t.Errorf("could not check if db has key: %v", err)
+	}
+	if has2 {
+		t.Errorf("db should not have non-existent key: %v", key2)
+	}
+}
+
+func Test_DBGet(t *testing.T) {
+	key := []byte("ralph merkle")
+
+	if err := db.Put(key, []byte{1, 2, 3}); err != nil {
+		t.Fatalf("could not save value in db: %v", err)
+	}
+
+	val, err := db.Get(key)
+	if err != nil {
+		t.Errorf("get failed: %v", err)
+	}
+	if len(val) == 0 {
+		t.Errorf("no value stored for key")
+	}
+
+	key2 := []byte{}
+	val2, err := db.Get(key2)
+	if len(val2) != 0 {
+		t.Errorf("non-existent key should not have a value. key=%v, value=%v", key2, val2)
+	}
+}
+
+func Test_DBDelete(t *testing.T) {
+	key := []byte("ralph merkle")
+
+	if err := db.Put(key, []byte{1, 2, 3}); err != nil {
+		t.Fatalf("could not save value in db: %v", err)
+	}
+
+	if err := db.Delete(key); err != nil {
+		t.Errorf("could not delete key: %v", key)
+	}
+}

sharding/database/inmemory.go

@@ -12,48 +12,48 @@ import (
 
 // ShardKV is an in-memory mapping of hashes to RLP encoded values.
 type ShardKV struct {
-	kv   map[common.Hash]*[]byte
+	kv   map[common.Hash][]byte
 	lock sync.RWMutex
 }
 
 // NewShardKV initializes a keyval store in memory.
 func NewShardKV() *ShardKV {
-	return &ShardKV{kv: make(map[common.Hash]*[]byte)}
+	return &ShardKV{kv: make(map[common.Hash][]byte)}
 }
 
 // Get fetches a val from the mappping by key.
-func (sb *ShardKV) Get(k common.Hash) (*[]byte, error) {
+func (sb *ShardKV) Get(k []byte) ([]byte, error) {
 	sb.lock.RLock()
 	defer sb.lock.RUnlock()
-	v, ok := sb.kv[k]
+	v, ok := sb.kv[common.BytesToHash(k)]
 	if !ok {
-		return nil, fmt.Errorf("key not found: %v", k)
+		return []byte{}, fmt.Errorf("key not found: %v", k)
 	}
 	return v, nil
 }
 
 // Has checks if the key exists in the mapping.
-func (sb *ShardKV) Has(k common.Hash) bool {
+func (sb *ShardKV) Has(k []byte) (bool, error) {
 	sb.lock.RLock()
 	defer sb.lock.RUnlock()
-	v := sb.kv[k]
-	return v != nil
+	v := sb.kv[common.BytesToHash(k)]
+	return v != nil, nil
 }
 
 // Put updates a key's value in the mapping.
-func (sb *ShardKV) Put(k common.Hash, v []byte) error {
+func (sb *ShardKV) Put(k []byte, v []byte) error {
 	sb.lock.Lock()
 	defer sb.lock.Unlock()
 	// there is no error in a simple setting of a value in a go map.
-	sb.kv[k] = &v
+	sb.kv[common.BytesToHash(k)] = v
 	return nil
 }
 
 // Delete removes the key and value from the mapping.
-func (sb *ShardKV) Delete(k common.Hash) error {
+func (sb *ShardKV) Delete(k []byte) error {
 	sb.lock.Lock()
 	defer sb.lock.Unlock()
 	// There is no return value for deleting a simple key in a go map.
-	delete(sb.kv, k)
+	delete(sb.kv, common.BytesToHash(k))
 	return nil
 }

sharding/database/inmemory_test.go

@@ -2,73 +2,77 @@ package database
 
 import (
 	"testing"
-
-	"github.com/ethereum/go-ethereum/common"
-	"github.com/ethereum/go-ethereum/sharding"
 )
 
 // Verifies that ShardKV implements the ShardBackend interface.
-var _ = sharding.ShardBackend(&ShardKV{})
+var _ = ShardBackend(&ShardKV{})
 
 func Test_ShardKVPut(t *testing.T) {
 	kv := NewShardKV()
-	hash := common.BytesToHash([]byte("ralph merkle"))
 
-	if err := kv.Put(hash, []byte{1, 2, 3}); err != nil {
+	if err := kv.Put([]byte("ralph merkle"), []byte{1, 2, 3}); err != nil {
 		t.Errorf("could not save value in kv store: %v", err)
 	}
 }
 
 func Test_ShardKVHas(t *testing.T) {
 	kv := NewShardKV()
-	hash := common.BytesToHash([]byte("ralph merkle"))
+	key := []byte("ralph merkle")
 
-	if err := kv.Put(hash, []byte{1, 2, 3}); err != nil {
+	if err := kv.Put(key, []byte{1, 2, 3}); err != nil {
 		t.Fatalf("could not save value in kv store: %v", err)
 	}
 
-	if !kv.Has(hash) {
-		t.Errorf("kv store does not have hash: %v", hash)
+	has, err := kv.Has(key)
+	if err != nil {
+		t.Errorf("could not check if kv store has key: %v", err)
+	}
+	if !has {
+		t.Errorf("kv store should have key: %v", key)
 	}
 
-	hash2 := common.BytesToHash([]byte{})
-	if kv.Has(hash2) {
-		t.Errorf("kv store should not contain unset key: %v", hash2)
+	key2 := []byte{}
+	has2, err := kv.Has(key2)
+	if err != nil {
+		t.Errorf("could not check if kv store has key: %v", err)
+	}
+	if has2 {
+		t.Errorf("kv store should not have non-existent key: %v", key2)
 	}
 }
 
 func Test_ShardKVGet(t *testing.T) {
 	kv := NewShardKV()
-	hash := common.BytesToHash([]byte("ralph merkle"))
+	key := []byte("ralph merkle")
 
-	if err := kv.Put(hash, []byte{1, 2, 3}); err != nil {
+	if err := kv.Put(key, []byte{1, 2, 3}); err != nil {
 		t.Fatalf("could not save value in kv store: %v", err)
 	}
 
-	val, err := kv.Get(hash)
+	val, err := kv.Get(key)
 	if err != nil {
 		t.Errorf("get failed: %v", err)
 	}
-	if val == nil {
+	if len(val) == 0 {
 		t.Errorf("no value stored for key")
 	}
 
-	hash2 := common.BytesToHash([]byte{})
-	val2, err := kv.Get(hash2)
-	if val2 != nil {
-		t.Errorf("non-existent key should not have a value. key=%v, value=%v", hash2, val2)
+	key2 := []byte{}
+	val2, err := kv.Get(key2)
+	if len(val2) != 0 {
+		t.Errorf("non-existent key should not have a value. key=%v, value=%v", key2, val2)
 	}
 }
 
 func Test_ShardKVDelete(t *testing.T) {
 	kv := NewShardKV()
-	hash := common.BytesToHash([]byte("ralph merkle"))
+	key := []byte("ralph merkle")
 
-	if err := kv.Put(hash, []byte{1, 2, 3}); err != nil {
+	if err := kv.Put(key, []byte{1, 2, 3}); err != nil {
 		t.Fatalf("could not save value in kv store: %v", err)
 	}
 
-	if err := kv.Delete(hash); err != nil {
-		t.Errorf("could not delete key: %v", hash)
+	if err := kv.Delete(key); err != nil {
+		t.Errorf("could not delete key: %v", key)
 	}
 }

sharding/node/node.go

@@ -44,6 +44,7 @@ type Node interface {
 	SMCCaller() *contracts.SMCCaller
 	SMCTransactor() *contracts.SMCTransactor
 	DepositFlagSet() bool
+	DataDirFlag() string
 }
 
 // General node for a sharding-enabled system.
@@ -213,6 +214,11 @@ func (n *shardingNode) DepositFlagSet() bool {
 	return n.ctx.GlobalBool(utils.DepositFlag.Name)
 }
 
+// DataDirFlag returns the datadir flag as a string.
+func (n *shardingNode) DataDirFlag() string {
+	return n.ctx.GlobalString(utils.DataDirFlag.Name)
+}
+
 // Client to interact with a geth node via JSON-RPC.
 func (n *shardingNode) ethereumClient() *ethclient.Client {
 	return n.client

sharding/notary/service.go

@@ -2,19 +2,28 @@ package notary
 
 import (
 	"github.com/ethereum/go-ethereum/log"
+	"github.com/ethereum/go-ethereum/sharding/database"
 	"github.com/ethereum/go-ethereum/sharding/node"
 )
 
 // Notary holds functionality required to run a collation notary
 // in a sharded system. Must satisfy the Service interface defined in
 // sharding/service.go.
 type Notary struct {
-	node node.Node
+	node    node.Node
+	shardDB database.ShardBackend
 }
 
 // NewNotary creates a new notary instance.
 func NewNotary(node node.Node) (*Notary, error) {
-	return &Notary{node}, nil
+	// Initializes a shardDB that writes to disk at /path/to/datadir/shardchaindata.
+	// This DB can be used by the Notary service to create Shard struct
+	// instances.
+	shardDB, err := database.NewShardDB(node.DataDirFlag(), "shardchaindata")
+	if err != nil {
+		return nil, err
+	}
+	return &Notary{node, shardDB}, nil
 }
 
 // Start the main routine for a notary.

sharding/notary/service_test.go

@@ -63,6 +63,10 @@ func (m *mockNode) DepositFlagSet() bool {
 	return m.DepositFlag
 }
 
+func (m *mockNode) DataDirFlag() string {
+	return "/tmp/datadir"
+}
+
 // Unused mockClient methods.
 func (m *mockNode) Start() error {
 	m.t.Fatal("Start called")

sharding/proposer/service.go

@@ -2,21 +2,28 @@ package proposer
 
 import (
 	"github.com/ethereum/go-ethereum/log"
+	"github.com/ethereum/go-ethereum/sharding/database"
 	"github.com/ethereum/go-ethereum/sharding/node"
 )
 
 // Proposer holds functionality required to run a collation proposer
 // in a sharded system. Must satisfy the Service interface defined in
 // sharding/service.go.
 type Proposer struct {
-	node node.Node
+	node    node.Node
+	shardDB database.ShardBackend
 }
 
 // NewProposer creates a struct instance. It is initialized and
 // registered as a service upon start of a sharding node.
 // Has access to the public methods of this node.
 func NewProposer(node node.Node) (*Proposer, error) {
-	return &Proposer{node}, nil
+	// Initializes a shardchaindata directory persistent db.
+	shardDB, err := database.NewShardDB(node.DataDirFlag(), "shardchaindata")
+	if err != nil {
+		return nil, err
+	}
+	return &Proposer{node, shardDB}, nil
 }
 
 // Start the main loop for proposing collations.