[experiment] Trie batch update

core/state/database.go

@@ -69,7 +69,8 @@ type Trie interface {
 	// by the caller while they are stored in the trie. If a node was not found in the
 	// database, a trie.MissingNodeError is returned.
 	TryUpdate(key, value []byte) error
-
+	BatchStart()
+	BatchEnd()
 	// TryDelete removes any existing value for key from the trie. If a node was not
 	// found in the database, a trie.MissingNodeError is returned.
 	TryDelete(key []byte) error
@@ -82,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 {
@@ -688,16 +689,17 @@ func (s *StateDB) Finalise(deleteEmptyObjects bool) {
 func (s *StateDB) IntermediateRoot(deleteEmptyObjects bool) common.Hash {
 	// Finalise all the dirty storage states and write them into the tries
 	s.Finalise(deleteEmptyObjects)
-
+	s.trie.BatchStart()
 	for addr := range s.stateObjectsPending {
 		obj := s.stateObjects[addr]
 		if obj.deleted {
 			s.deleteStateObject(obj)
 		} else {
-			obj.updateRoot(s.db)
+			obj.updateRoot(obj.getTrie(s.db))
 			s.updateStateObject(obj)
 		}
 	}
+	s.trie.BatchEnd()
 	if len(s.stateObjectsPending) > 0 {
 		s.stateObjectsPending = make(map[common.Address]struct{})
 	}
@@ -729,39 +731,55 @@ 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{})
 	}
 	// Write the account trie changes, measuing the amount of wasted time
 	if metrics.EnabledExpensive {
 		defer func(start time.Time) { s.AccountCommits += time.Since(start) }(time.Now())
 	}
-	return s.trie.Commit(func(leaf []byte, parent common.Hash) error {
-		var account Account
+	// The onleaf func is called _serially_, so we can reuse the same account
+	// for unmarshalling every time.
+	var account Account
+	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

@@ -112,6 +112,9 @@ func (t *odrTrie) TryUpdate(key, value []byte) error {
 	})
 }
 
+func (t *odrTrie) BatchStart() {}
+func (t *odrTrie) BatchEnd()   {}
+
 func (t *odrTrie) TryDelete(key []byte) error {
 	key = crypto.Keccak256(key)
 	return t.do(key, func() error {
@@ -126,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/database.go

@@ -314,24 +314,24 @@ func (db *Database) InsertBlob(hash common.Hash, blob []byte) {
 	db.lock.Lock()
 	defer db.lock.Unlock()
 
-	db.insert(hash, blob, rawNode(blob))
+	db.insert(hash, len(blob), rawNode(blob))
 }
 
 // insert inserts a collapsed trie node into the memory database. This method is
 // a more generic version of InsertBlob, supporting both raw blob insertions as
-// well ex trie node insertions. The blob must always be specified to allow proper
+// well ex trie node insertions. The blob size must be specified to allow proper
 // size tracking.
-func (db *Database) insert(hash common.Hash, blob []byte, node node) {
+func (db *Database) insert(hash common.Hash, size int, node node) {
 	// If the node's already cached, skip
 	if _, ok := db.dirties[hash]; ok {
 		return
 	}
-	memcacheDirtyWriteMeter.Mark(int64(len(blob)))
+	memcacheDirtyWriteMeter.Mark(int64(size))
 
 	// Create the cached entry for this node
 	entry := &cachedNode{
 		node:      simplifyNode(node),
-		size:      uint16(len(blob)),
+		size:      uint16(size),
 		flushPrev: db.newest,
 	}
 	for _, child := range entry.childs() {

trie/hasher.go

@@ -67,6 +67,7 @@ func newHasher(onleaf LeafCallback) *hasher {
 }
 
 func returnHasherToPool(h *hasher) {
+	h.onleaf = nil
 	hasherPool.Put(h)
 }
 
@@ -184,7 +185,7 @@ func (h *hasher) store(n node, db *Database, force bool) (node, error) {
 		hash := common.BytesToHash(hash)
 
 		db.lock.Lock()
-		db.insert(hash, h.tmp, n)
+		db.insert(hash, len(h.tmp), n)
 		db.lock.Unlock()
 
 		// Track external references from account->storage trie