From 8bfb58e30d386d4111e59ed0398b628c5cdc6a63 Mon Sep 17 00:00:00 2001 From: Martin Holst Swende Date: Wed, 1 Jan 2020 23:31:39 +0100 Subject: [PATCH] core/state, trie, light: improve commit paralellization --- core/state/database.go | 2 ++ core/state/state_object.go | 28 ++++++++++++++++++++++------ core/state/statedb.go | 35 +++++++++++++++++++++++++---------- light/trie.go | 7 +++++++ trie/pure_committer.go | 25 ++++++++++++++----------- trie/secure_trie.go | 14 ++++++++++++++ trie/trie.go | 34 ++++++++++++++++++++++++++-------- trie/trie_test.go | 2 ++ 8 files changed, 112 insertions(+), 35 deletions(-) diff --git a/core/state/database.go b/core/state/database.go index c3d4b62410..ccdf6cd715 100644 --- a/core/state/database.go +++ b/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 diff --git a/core/state/state_object.go b/core/state/state_object.go index 8680de021f..93174edefd 100644 --- a/core/state/state_object.go +++ b/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) { diff --git a/core/state/statedb.go b/core/state/statedb.go index f54fd929b2..7511b26b4e 100644 --- a/core/state/statedb.go +++ b/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 } diff --git a/light/trie.go b/light/trie.go index 4579cc2571..8a47435ee5 100644 --- a/light/trie.go +++ b/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 diff --git a/trie/pure_committer.go b/trie/pure_committer.go index 30404b217d..dad9d566d2 100644 --- a/trie/pure_committer.go +++ b/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 } - - -*/ diff --git a/trie/secure_trie.go b/trie/secure_trie.go index 4e385c3593..96bf81581b 100644 --- a/trie/secure_trie.go +++ b/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 { diff --git a/trie/trie.go b/trie/trie.go index 1c27bab5e5..bfd3d7733f 100644 --- a/trie/trie.go +++ b/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 { diff --git a/trie/trie_test.go b/trie/trie_test.go index b0759b4438..fc3a98c707 100644 --- a/trie/trie_test.go +++ b/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 }