Merge pull request #1419 from hyperledger/concurrency

Make RWTree concurrency safe

Author: Casey Kuhlman

.gitattributes

@@ -0,0 +1,2 @@
+*.pb.go linguist-generated
+yarn.lock linguist-generated

rpc/rpctransact/transact_server.go

@@ -2,7 +2,6 @@ package rpctransact
 
 import (
 	"fmt"
-	"sync"
 	"time"
 
 	"github.com/hyperledger/burrow/logging"
@@ -27,7 +26,6 @@ type transactServer struct {
 	transactor *execution.Transactor
 	txCodec    txs.Codec
 	logger     *logging.Logger
-	lock       *sync.Mutex
 }
 
 func NewTransactServer(state acmstate.Reader, blockchain bcm.BlockchainInfo, transactor *execution.Transactor,
@@ -38,7 +36,6 @@ func NewTransactServer(state acmstate.Reader, blockchain bcm.BlockchainInfo, tra
 		transactor: transactor,
 		txCodec:    txCodec,
 		logger:     logger.WithScope("NewTransactServer()"),
-		lock:       &sync.Mutex{},
 	}
 }
 
@@ -104,14 +101,10 @@ func (ts *transactServer) CallTxSim(ctx context.Context, param *payload.CallTx)
 	if param.Address == nil {
 		return nil, fmt.Errorf("CallSim requires a non-nil address from which to retrieve code")
 	}
-	ts.lock.Lock()
-	defer ts.lock.Unlock()
 	return execution.CallSim(ts.state, ts.blockchain, param.Input.Address, *param.Address, param.Data, ts.logger)
 }
 
 func (ts *transactServer) CallCodeSim(ctx context.Context, param *CallCodeParam) (*exec.TxExecution, error) {
-	ts.lock.Lock()
-	defer ts.lock.Unlock()
 	return execution.CallCodeSim(ts.state, ts.blockchain, param.FromAddress, param.FromAddress, param.Code, param.Data,
 		ts.logger)
 }

storage/key_value_store.go

@@ -0,0 +1,41 @@
+package storage
+
+import (
+	"bytes"
+)
+
+// KeyOrder maps []byte{} -> -1, []byte(nil) -> 1, and everything else to 0. This encodes the assumptions of the
+// KVIterator domain endpoints
+func KeyOrder(key []byte) int {
+	if key == nil {
+		// Sup
+		return 1
+	}
+	if len(key) == 0 {
+		// Inf
+		return -1
+	}
+	// Normal key
+	return 0
+}
+
+// Sorts the keys as if they were compared lexicographically with their KeyOrder prepended
+func CompareKeys(k1, k2 []byte) int {
+	ko1 := KeyOrder(k1)
+	ko2 := KeyOrder(k2)
+	if ko1 < ko2 {
+		return -1
+	}
+	if ko1 > ko2 {
+		return 1
+	}
+	return bytes.Compare(k1, k2)
+}
+
+// NormaliseDomain encodes the assumption that when nil is used as a lower bound is interpreted as low rather than high
+func NormaliseDomain(low, high []byte) ([]byte, []byte) {
+	if len(low) == 0 {
+		low = []byte{}
+	}
+	return low, high
+}

storage/kvstore_test.go renamed to storage/key_value_store_test.go

@@ -2,9 +2,10 @@ package storage
 
 import (
 	"fmt"
+	"sync"
+	"sync/atomic"
 
 	"github.com/cosmos/iavl"
-
 	dbm "github.com/tendermint/tm-db"
 	"github.com/xlab/treeprint"
 )
@@ -13,26 +14,40 @@ import (
 // recently saved version of the tree - which provides immutable access. Writes are routed to a working tree that is
 // mutable. On save the working tree is saved to DB, frozen, and replaces the previous immutable read tree.
 type RWTree struct {
+	// Synchronise 'write side access' - i.e. to the write tree and updated flag (which only read-locks this mutex)
+	sync.RWMutex
 	// Working tree accumulating writes
 	tree *MutableTree
 	// Read-only tree serving previous state
-	*ImmutableTree
+	readTree atomic.Value // *ImmutableTree
 	// Have any writes occurred since last save
 	updated bool
 }
 
+var _ KVCallbackIterableReader = &RWTree{}
+var _ Versioned = &RWTree{}
+
 // Creates a concurrency safe version of an IAVL tree whereby reads are routed to the last saved tree.
 // Writes must be serialised (as they are within a commit for example).
 func NewRWTree(db dbm.DB, cacheSize int) (*RWTree, error) {
 	tree, err := NewMutableTree(db, cacheSize)
-	return &RWTree{
-		tree:          tree,
-		ImmutableTree: &ImmutableTree{iavl.NewImmutableTree(db, cacheSize)},
-	}, err
+	if err != nil {
+		return nil, err
+	}
+	readTree := &ImmutableTree{iavl.NewImmutableTree(db, cacheSize)}
+	rwt := &RWTree{
+		tree: tree,
+	}
+	rwt.readTree.Store(readTree)
+	return rwt, nil
 }
 
+// Write-side write methods - of the mutable tree - synchronised by write-lock of RWMutex
+
 // Tries to load the execution state from DB, returns nil with no error if no state found
 func (rwt *RWTree) Load(version int64, overwriting bool) error {
+	rwt.Lock()
+	defer rwt.Unlock()
 	const errHeader = "RWTree.Load():"
 	if version <= 0 {
 		return fmt.Errorf("%s trying to load from non-positive version %d", errHeader, version)
@@ -42,52 +57,92 @@ func (rwt *RWTree) Load(version int64, overwriting bool) error {
 		return fmt.Errorf("%s loading version %d: %v", errHeader, version, err)
 	}
 	// Set readTree at commit point == tree
-	rwt.ImmutableTree, err = rwt.tree.GetImmutable(version)
+	readTree, err := rwt.tree.GetImmutable(version)
 	if err != nil {
 		return fmt.Errorf("%s loading version %d: %v", errHeader, version, err)
 	}
+	rwt.readTree.Store(readTree)
+	rwt.updated = false
+
 	return nil
 }
 
 // Save the current write tree making writes accessible from read tree.
 func (rwt *RWTree) Save() ([]byte, int64, error) {
+	rwt.Lock()
+	defer rwt.Unlock()
 	// save state at a new version may still be orphaned before we save the version against the hash
 	hash, version, err := rwt.tree.SaveVersion()
 	if err != nil {
 		return nil, 0, fmt.Errorf("could not save RWTree: %v", err)
 	}
 	// Take an immutable reference to the tree we just saved for querying
-	rwt.ImmutableTree, err = rwt.tree.GetImmutable(version)
+	readTree, err := rwt.tree.GetImmutable(version)
 	if err != nil {
-		return nil, 0, fmt.Errorf("RWTree.Save() could not obtain ImmutableTree read tree: %v", err)
+		return nil, 0, fmt.Errorf("RWTree.Save() could not obtain immutable read tree: %v", err)
 	}
+	rwt.readTree.Store(readTree)
 	rwt.updated = false
 	return hash, version, nil
 }
 
 func (rwt *RWTree) Set(key, value []byte) bool {
+	rwt.Lock()
+	defer rwt.Unlock()
 	rwt.updated = true
 	return rwt.tree.Set(key, value)
 }
 
 func (rwt *RWTree) Delete(key []byte) ([]byte, bool) {
+	rwt.Lock()
+	defer rwt.Unlock()
 	rwt.updated = true
 	return rwt.tree.Remove(key)
 }
 
+// Write-side read methods - of the mutable tree - synchronised by read-lock of RWMutex
+
 // Returns true if there have been any writes since last save
 func (rwt *RWTree) Updated() bool {
+	rwt.RLock()
+	defer rwt.RUnlock()
 	return rwt.updated
 }
 
 func (rwt *RWTree) GetImmutable(version int64) (*ImmutableTree, error) {
+	rwt.RLock()
+	defer rwt.RUnlock()
 	return rwt.tree.GetImmutable(version)
 }
 
 func (rwt *RWTree) IterateWriteTree(start, end []byte, ascending bool, fn func(key []byte, value []byte) error) error {
+	rwt.RLock()
+	defer rwt.RUnlock()
 	return rwt.tree.IterateWriteTree(start, end, ascending, fn)
 }
 
+// Read-side read methods - of the immutable read tree (previous saved state) - synchronised solely via atomic.Value
+
+func (rwt *RWTree) Hash() []byte {
+	return rwt.readTree.Load().(*ImmutableTree).Hash()
+}
+
+func (rwt *RWTree) Version() int64 {
+	return rwt.readTree.Load().(*ImmutableTree).Version()
+}
+
+func (rwt *RWTree) Get(key []byte) ([]byte, error) {
+	return rwt.readTree.Load().(*ImmutableTree).Get(key)
+}
+
+func (rwt *RWTree) Has(key []byte) (bool, error) {
+	return rwt.readTree.Load().(*ImmutableTree).Has(key)
+}
+
+func (rwt *RWTree) Iterate(low, high []byte, ascending bool, fn func(key []byte, value []byte) error) error {
+	return rwt.readTree.Load().(*ImmutableTree).Iterate(low, high, ascending, fn)
+}
+
 // Tree printing
 
 func (rwt *RWTree) Dump() string {

storage/rwtree.go

@@ -2,8 +2,10 @@ package storage
 
 import (
 	"fmt"
+	"math/rand"
 	"runtime/debug"
 	"testing"
+	"time"
 
 	"github.com/stretchr/testify/require"
 
@@ -151,58 +153,68 @@ func capturePanic(f func() error) (err error) {
 	return
 }
 
-// TODO: fix
-//func TestConcurrentReadWriteSave(t *testing.T) {
-//	rwt, err := NewRWTree(dbm.NewMemDB(), 100)
-//	require.NoError(t, err)
-//	n := 100
-//
-//	doneCh := make(chan struct{})
-//	errCh := make(chan interface{})
-//
-//	go func() {
-//		for b := byte(0); true; b++ {
-//			val := []byte{b}
-//			go func() {
-//				err := capturePanic(func() error {
-//					rwt.Set(val, val)
-//					return nil
-//				})
-//				if err != nil {
-//					errCh <- err
-//				}
-//			}()
-//			go func() {
-//				err := capturePanic(func() error {
-//					_, err := rwt.Get(val)
-//					return err
-//				})
-//				if err != nil {
-//					errCh <- err
-//				}
-//			}()
-//
-//			select {
-//			case <-doneCh:
-//				return
-//			default:
-//			}
-//		}
-//	}()
-//
-//	for i := 0; i < n/10; i++ {
-//		time.Sleep(time.Duration(rand.Int63n(100)) * time.Millisecond)
-//		err := capturePanic(func() error {
-//			_, _, err := rwt.Save()
-//			return err
-//		})
-//		if err != nil {
-//			break
-//		}
-//	}
-//	close(doneCh)
-//
-//	for err := range errCh {
-//		t.Fatal(err)
-//	}
-//}
+func TestConcurrentReadWriteSave(t *testing.T) {
+	rwt, err := NewRWTree(dbm.NewMemDB(), 100)
+	require.NoError(t, err)
+	n := 100
+
+	doneCh := make(chan struct{})
+	errCh := make(chan interface{})
+
+	// Saturate with concurrent reads and writes
+	var spin func()
+	spin = func() {
+		for i := 0; i < n; i++ {
+			val := []byte{byte(i)}
+			go func() {
+				err := capturePanic(func() error {
+					rwt.Set(val, val)
+					return nil
+				})
+				if err != nil {
+					errCh <- err
+				}
+			}()
+			go func() {
+				err := capturePanic(func() error {
+					_, err := rwt.Get(val)
+					return err
+				})
+				if err != nil {
+					errCh <- err
+				}
+			}()
+
+		}
+		select {
+		case <-doneCh:
+			return
+		default:
+			// Avoid starvation
+			time.Sleep(time.Millisecond)
+			spin()
+		}
+	}
+
+	// let's
+	go spin()
+
+	// Ensure Save() is safe with concurrent read/writes
+	for i := 0; i < n/10; i++ {
+		time.Sleep(time.Duration(rand.Int63n(100)) * time.Millisecond)
+		err := capturePanic(func() error {
+			_, _, err := rwt.Save()
+			return err
+		})
+		if err != nil {
+			break
+		}
+	}
+	close(doneCh)
+
+	select {
+	case err := <-errCh:
+		t.Fatal(err)
+	default:
+	}
+}