core/state, trie, light: improve commit paralellization

core/state/database.go

@@ -83,6 +83,8 @@ type Trie interface {
 	// and external (for account tries) references.
 	Commit(onleaf trie.LeafCallback) (common.Hash, error)
 
+	CommitTo(onleaf trie.LeafCallback, dbi *trie.DbInserter) (common.Hash, error)
+
 	// NodeIterator returns an iterator that returns nodes of the trie. Iteration
 	// starts at the key after the given start key.
 	NodeIterator(startKey []byte) trie.NodeIterator

core/state/state_object.go

@@ -19,6 +19,7 @@ package state
 import (
 	"bytes"
 	"fmt"
+	"github.com/ethereum/go-ethereum/trie"
 	"io"
 	"math/big"
 	"time"
@@ -272,7 +273,7 @@ func (s *stateObject) finalise() {
 }
 
 // updateTrie writes cached storage modifications into the object's storage trie.
-func (s *stateObject) updateTrie(db Database) Trie {
+func (s *stateObject) updateTrie(tr Trie) Trie {
 	// Make sure all dirty slots are finalized into the pending storage area
 	s.finalise()
 
@@ -281,7 +282,6 @@ func (s *stateObject) updateTrie(db Database) Trie {
 		defer func(start time.Time) { s.db.StorageUpdates += time.Since(start) }(time.Now())
 	}
 	// Insert all the pending updates into the trie
-	tr := s.getTrie(db)
 	for key, value := range s.pendingStorage {
 		// Skip noop changes, persist actual changes
 		if value == s.originStorage[key] {
@@ -304,8 +304,8 @@ func (s *stateObject) updateTrie(db Database) Trie {
 }
 
 // UpdateRoot sets the trie root to the current root hash of
-func (s *stateObject) updateRoot(db Database) {
-	s.updateTrie(db)
+func (s *stateObject) updateRoot(tr Trie) {
+	s.updateTrie(tr)
 
 	// Track the amount of time wasted on hashing the storge trie
 	if metrics.EnabledExpensive {
@@ -316,8 +316,8 @@ func (s *stateObject) updateRoot(db Database) {
 
 // CommitTrie the storage trie of the object to db.
 // This updates the trie root.
-func (s *stateObject) CommitTrie(db Database) error {
-	s.updateTrie(db)
+func (s *stateObject) CommitTrie(tr Trie) error {
+	s.updateTrie(tr)
 	if s.dbErr != nil {
 		return s.dbErr
 	}
@@ -332,6 +332,22 @@ func (s *stateObject) CommitTrie(db Database) error {
 	return err
 }
 
+func (s *stateObject) CommitTrieTo(tr Trie, inserter *trie.DbInserter) error {
+	s.updateTrie(tr)
+	if s.dbErr != nil {
+		return s.dbErr
+	}
+	// Track the amount of time wasted on committing the storge trie
+	if metrics.EnabledExpensive {
+		defer func(start time.Time) { s.db.StorageCommits += time.Since(start) }(time.Now())
+	}
+	root, err := s.trie.CommitTo(nil, inserter)
+	if err == nil {
+		s.data.Root = root
+	}
+	return err
+}
+
 // AddBalance removes amount from c's balance.
 // It is used to add funds to the destination account of a transfer.
 func (s *stateObject) AddBalance(amount *big.Int) {

core/state/statedb.go

@@ -18,6 +18,7 @@
 package state
 
 import (
+	"bytes"
 	"errors"
 	"fmt"
 	"math/big"
@@ -330,7 +331,7 @@ func (s *StateDB) StorageTrie(addr common.Address) Trie {
 		return nil
 	}
 	cpy := stateObject.deepCopy(s)
-	return cpy.updateTrie(s.db)
+	return cpy.updateTrie(cpy.getTrie(s.db))
 }
 
 func (s *StateDB) HasSuicided(addr common.Address) bool {
@@ -694,7 +695,7 @@ func (s *StateDB) IntermediateRoot(deleteEmptyObjects bool) common.Hash {
 		if obj.deleted {
 			s.deleteStateObject(obj)
 		} else {
-			obj.updateRoot(s.db)
+			obj.updateRoot(obj.getTrie(s.db))
 			s.updateStateObject(obj)
 		}
 	}
@@ -730,20 +731,30 @@ func (s *StateDB) Commit(deleteEmptyObjects bool) (common.Hash, error) {
 	// Finalize any pending changes and merge everything into the tries
 	s.IntermediateRoot(deleteEmptyObjects)
 
+	// The commit phase. We start by committing the account storage tries
+	//
+	// Start the dedicated inserter
+	dbi := trie.StartDBInserter(s.db.TrieDB())
+
 	// Commit objects to the trie, measuring the elapsed time
 	for addr := range s.stateObjectsDirty {
 		if obj := s.stateObjects[addr]; !obj.deleted {
 			// Write any contract code associated with the state object
 			if obj.code != nil && obj.dirtyCode {
-				s.db.TrieDB().InsertBlob(common.BytesToHash(obj.CodeHash()), obj.code)
+				dbi.InsertBlob(obj.code, common.BytesToHash(obj.CodeHash()))
+				//s.db.TrieDB().InsertBlob(common.BytesToHash(obj.CodeHash()), obj.code)
 				obj.dirtyCode = false
 			}
 			// Write any storage changes in the state object to its storage trie
-			if err := obj.CommitTrie(s.db); err != nil {
+			if err := obj.CommitTrieTo(obj.getTrie(s.db), dbi); err != nil {
 				return common.Hash{}, err
 			}
 		}
 	}
+	// Wait for storage update to punch through
+	//dbi.WaitForEmpty()
+	// .. or not
+
 	if len(s.stateObjectsDirty) > 0 {
 		s.stateObjectsDirty = make(map[common.Address]struct{})
 	}
@@ -754,17 +765,21 @@ func (s *StateDB) Commit(deleteEmptyObjects bool) (common.Hash, error) {
 	// The onleaf func is called _serially_, so we can reuse the same account
 	// for unmarshalling every time.
 	var account Account
-	return s.trie.Commit(func(leaf []byte, parent common.Hash) error {
+	var trieDb = s.db.TrieDB()
+	h, e := s.trie.CommitTo(func(leaf []byte, parent common.Hash) error {
 		if err := rlp.DecodeBytes(leaf, &account); err != nil {
 			return nil
 		}
 		if account.Root != emptyRoot {
-			s.db.TrieDB().Reference(account.Root, parent)
+			trieDb.Reference(account.Root, parent)
 		}
-		code := common.BytesToHash(account.CodeHash)
-		if code != emptyCode {
-			s.db.TrieDB().Reference(code, parent)
+		if !bytes.Equal(emptyCodeHash, account.CodeHash) {
+			trieDb.Reference(common.BytesToHash(account.CodeHash), parent)
 		}
 		return nil
-	})
+	}, dbi)
+	// Close it, and wait for empty
+	dbi.Close()
+
+	return h, e
 }

light/trie.go

@@ -129,6 +129,13 @@ func (t *odrTrie) Commit(onleaf trie.LeafCallback) (common.Hash, error) {
 	return t.trie.Commit(onleaf)
 }
 
+func (t *odrTrie) CommitTo(onleaf trie.LeafCallback, dbi *trie.DbInserter) (common.Hash, error) {
+	if t.trie == nil {
+		return t.id.Root, nil
+	}
+	return t.trie.Commit(onleaf)
+}
+
 func (t *odrTrie) Hash() common.Hash {
 	if t.trie == nil {
 		return t.id.Root

trie/pure_committer.go

@@ -19,6 +19,7 @@ package trie
 import (
 	"errors"
 	"fmt"
+	"math/rand"
 	"sync"
 
 	"github.com/ethereum/go-ethereum/common"
@@ -32,6 +33,7 @@ type Leaf struct {
 	hash   common.Hash // hash of rlp data
 	node   node        // the node to commit
 	vnodes bool        // set to true if the node (possibly) contains a valueNode
+	onleaf LeafCallback
 }
 
 type committer struct {
@@ -52,10 +54,11 @@ var committerPool = sync.Pool{
 	},
 }
 
-func newCommitter(onleaf LeafCallback) *committer {
+func newCommitter(onleaf LeafCallback, leafCh chan *Leaf) *committer {
 	h := committerPool.Get().(*committer)
 	h.onleaf = onleaf
-	if onleaf != nil {
+	h.leafCh = leafCh
+	if onleaf != nil && leafCh == nil {
 		h.leafCh = make(chan *Leaf, 200) // arbitrary number
 	}
 	return h
@@ -182,6 +185,7 @@ func (h *committer) store(n node, db *Database, force bool, hasVnodeChildren boo
 			hash:   common.BytesToHash(hash),
 			node:   n,
 			vnodes: hasVnodeChildren,
+			onleaf: h.onleaf,
 		}
 	} else if db != nil {
 		// No leaf-callback used, but there's still a database. Do serial
@@ -275,7 +279,7 @@ have to do the waitgroup-wait between each trie commit.
 The code below is a rough sketch, it needs to be integrated nicely without causing
 dependency cycles between state, core and trie.
 
-
+**/
 
 type DbInserter struct {
 	inputCh  chan *Leaf // This is where input to database is sent
@@ -293,7 +297,7 @@ func (dbi *DbInserter) run() {
 			size      = item.size
 			n         = item.node
 			hasVnodes = item.vnodes
-			onleaf    = item.onLeaf
+			onleaf    = item.onleaf
 		)
 		if size < 0 {
 			// This is an end-marker object.
@@ -334,14 +338,14 @@ func (dbi *DbInserter) Insert(leaf *Leaf) {
 // channel has been handled
 func (dbi *DbInserter) WaitForEmpty() {
 	// Send an arbitrary id there
-	checksum := rand.Uint32()
-	dbi.inputCh <- &trie.Leaf{
+	checksum := rand.Int()
+	dbi.inputCh <- &Leaf{
 		size: -checksum,
 	}
 	// And wait for it to come back
 	for {
 		select {
-		case retval <- dbi.reportCh:
+		case retval := <-dbi.reportCh:
 			if retval == checksum {
 				return
 			}
@@ -351,7 +355,7 @@ func (dbi *DbInserter) WaitForEmpty() {
 }
 
 func (dbi *DbInserter) InsertBlob(blob []byte, blobHash common.Hash) {
-	dbi.inputCh <- &trie.Leaf{
+	dbi.inputCh <- &Leaf{
 		size:   len(blob),
 		hash:   blobHash,
 		node:   rawNode(blob),
@@ -366,8 +370,7 @@ func StartDBInserter(db *Database) *DbInserter {
 		reportCh: make(chan int),
 		db:       db,
 	}
+	dbi.wg.Add(1)
 	go dbi.run()
+	return dbi
 }
-
-
-*/

trie/secure_trie.go

@@ -161,6 +161,20 @@ func (t *SecureTrie) Commit(onleaf LeafCallback) (root common.Hash, err error) {
 	return t.trie.Commit(onleaf)
 }
 
+func (t *SecureTrie) CommitTo(onleaf LeafCallback, inserter *DbInserter ) (root common.Hash, err error) {
+	// Write all the pre-images to the actual disk database
+	if len(t.getSecKeyCache()) > 0 {
+		t.trie.db.lock.Lock()
+		for hk, key := range t.secKeyCache {
+			t.trie.db.insertPreimage(common.BytesToHash([]byte(hk)), key)
+		}
+		t.trie.db.lock.Unlock()
+
+		t.secKeyCache = make(map[string][]byte)
+	}
+	return t.trie.CommitTo(onleaf, inserter)
+}
+
 // Hash returns the root hash of SecureTrie. It does not write to the
 // database and can be used even if the trie doesn't have one.
 func (t *SecureTrie) Hash() common.Hash {

trie/trie.go

@@ -52,7 +52,7 @@ type Trie struct {
 	dirtyCount int
 	// And leafs to hash
 	unhashedCount int
-	batchMode bool
+	batchMode     bool
 }
 
 // newFlag returns the cache flag value for a newly created node.
@@ -187,10 +187,10 @@ func (t *Trie) TryUpdate(key, value []byte) error {
 	return nil
 }
 
-func (t *Trie) batchStart(){
+func (t *Trie) batchStart() {
 	t.batchMode = true
 }
-func (t *Trie) batchEnd(){
+func (t *Trie) batchEnd() {
 	t.batchMode = false
 }
 
@@ -240,14 +240,14 @@ func (t *Trie) insert(n node, prefix, key []byte, value node) (bool, node, error
 		// When we modify a node, we only copy it in case it is an old committed
 		// node.
 		// If the node is "new", we just update in place.
-		if t.batchMode{
-			if h, dirty := n.cache(); !dirty || h != nil{
+		if t.batchMode {
+			if h, dirty := n.cache(); !dirty || h != nil {
 				// This node is either not dirty, or already hashed. We copy it
 				n = n.copy()
-			}else{
+			} else {
 				// No copy
 			}
-		}else{
+		} else {
 			n = n.copy()
 		}
 		n.flags = t.newFlag()
@@ -468,7 +468,7 @@ func (t *Trie) Commit(onleaf LeafCallback) (root common.Hash, err error) {
 		return emptyRoot, nil
 	}
 	rootHash := t.Hash()
-	h := newCommitter(onleaf)
+	h := newCommitter(onleaf, nil)
 	defer returnCommitterToPool(h)
 	var wg sync.WaitGroup
 	if onleaf != nil {
@@ -487,6 +487,24 @@ func (t *Trie) Commit(onleaf LeafCallback) (root common.Hash, err error) {
 	return rootHash, nil
 }
 
+func (t *Trie) CommitTo(onleaf LeafCallback, dbi *DbInserter) (root common.Hash, err error) {
+	if t.db == nil {
+		panic("commit called on trie with nil database")
+	}
+	if t.root == nil {
+		return emptyRoot, nil
+	}
+	rootHash := t.Hash()
+	h := newCommitter(onleaf, dbi.inputCh)
+	h.leafCh = dbi.inputCh
+	_, err = h.commit(t.root, t.db, true)
+	if err != nil {
+		return common.Hash{}, err
+	}
+	t.dirtyCount = 0
+	return rootHash, nil
+}
+
 // oldHashRoot is the old implementation of hashRoot, which uses the regular hasher
 func (t *Trie) oldHashRoot(db *Database, onleaf LeafCallback) (node, node, error) {
 	if t.root == nil {

trie/trie_test.go

@@ -507,10 +507,12 @@ func benchUpdate(b *testing.B, e binary.ByteOrder) *Trie {
 	trie := newEmpty()
 	k := make([]byte, 32)
 	b.ReportAllocs()
+	trie.batchStart()
 	for i := 0; i < b.N; i++ {
 		e.PutUint64(k, uint64(i))
 		trie.Update(k, k)
 	}
+	trie.batchEnd()
 	return trie
 }