From ab3a8135828c2990be6a80053103adae707e30a0 Mon Sep 17 00:00:00 2001 From: Po Date: Sun, 13 Jul 2025 14:58:58 +0200 Subject: [PATCH] perf: parallel fetching trie --- cmd/geth/chaincmd.go | 3 + core/parallel_state_processor.go | 13 +++- core/state/state_object.go | 9 ++- core/state/statedb.go | 117 +++++++++++++++++++++++-------- trie/trie.go | 10 +++ 5 files changed, 118 insertions(+), 34 deletions(-) diff --git a/cmd/geth/chaincmd.go b/cmd/geth/chaincmd.go index a08b693e79..48aa0e4648 100644 --- a/cmd/geth/chaincmd.go +++ b/cmd/geth/chaincmd.go @@ -47,6 +47,7 @@ import ( "github.com/ethereum/go-ethereum/log" "github.com/ethereum/go-ethereum/metrics" "github.com/ethereum/go-ethereum/params" + "github.com/ethereum/go-ethereum/trie" "github.com/urfave/cli/v2" ) @@ -450,6 +451,7 @@ func showMetrics() { // execution fmt.Println("prefetchTimer", blockPrefetchExecuteTimer.Total()) fmt.Println("accountReadTimer", accountReadTimer.Total()) + fmt.Println("ResolveTimer", trie.ResolveTime) fmt.Println("storageReadTimer", storageReadTimer.Total()) fmt.Println("blockExecutionTimer", blockExecutionTimer.Total()) @@ -476,6 +478,7 @@ func showMetrics() { fmt.Println("PrefetchMergeTime:", core.PrefetchMergeBALTime) fmt.Println("ParallelExeTime: ", core.ParallelExeTime) fmt.Println("PostMergeTime: ", core.PostMergeTime) + fmt.Println("PrefetchTrieTime: ", core.PrefetchTrieTimer) // total fmt.Println("blockInsertTimer", blockInsertTimer.Total()) diff --git a/core/parallel_state_processor.go b/core/parallel_state_processor.go index a070f4b8e6..aa08530be1 100644 --- a/core/parallel_state_processor.go +++ b/core/parallel_state_processor.go @@ -21,6 +21,7 @@ var ( PrefetchMergeBALTime = time.Duration(0) ParallelExeTime = time.Duration(0) PostMergeTime = time.Duration(0) + PrefetchTrieTimer = time.Duration(0) ) type ParallelStateProcessor struct { @@ -84,11 +85,17 @@ func (p *ParallelStateProcessor) Process(block *types.Block, statedb *state.Stat defer wg.Done() start := time.Now() - postState = statedb.Copy() + postState = statedb.CopyState() + // Stop prefetcher cause we'll directly fetch tries in parallel + postState.StopPrefetcher() + postState.MergePostBalStates() - // prewarm the updating trie - postState.IntermediateRoot(true) PostMergeTime += time.Since(start) + + // Prewarm the updating trie + start = time.Now() + postState.PrefetchTrie() + PrefetchTrieTimer += time.Since(start) }() go func() { diff --git a/core/state/state_object.go b/core/state/state_object.go index 02e92bf601..5b7b57b3bc 100644 --- a/core/state/state_object.go +++ b/core/state/state_object.go @@ -144,6 +144,9 @@ func (s *stateObject) getTrie() (Trie, error) { func (s *stateObject) getPrefetchedTrie() Trie { // If there's nothing to meaningfully return, let the user figure it out by // pulling the trie from disk. + if s.trie != nil { + return s.trie + } if (s.data.Root == types.EmptyRootHash && !s.db.db.TrieDB().IsVerkle()) || s.db.prefetcher == nil { return nil } @@ -512,9 +515,9 @@ func (s *stateObject) simpleCopy(db *StateDB) *stateObject { data: s.data, code: s.code, originStorage: s.originStorage.Copy(), - pendingStorage: make(Storage), - dirtyStorage: make(Storage), - uncommittedStorage: make(Storage), + pendingStorage: s.pendingStorage.Copy(), + dirtyStorage: s.dirtyStorage.Copy(), + uncommittedStorage: s.uncommittedStorage.Copy(), dirtyCode: s.dirtyCode, selfDestructed: s.selfDestructed, newContract: s.newContract, diff --git a/core/state/statedb.go b/core/state/statedb.go index 98fe52c58b..e9ffba5689 100644 --- a/core/state/statedb.go +++ b/core/state/statedb.go @@ -732,6 +732,54 @@ func (s *StateDB) MergePostBalStates() { } } +func (s *StateDB) PrefetchTrie() { + var workers errgroup.Group + + for addr := range s.journal.dirties { + addr := addr + // s.trie.GetAccount(addr) + workers.Go(func() error { + _, err := s.trie.GetAccount(addr) + if err != nil { + fmt.Println("error fetching account:", err) + } + return err + }) + + workers.Go(func() error { + obj, exist := s.stateObjects[addr] + if exist { + tr, err := s.db.OpenStorageTrie(s.originalRoot, addr, obj.data.Root, nil) + if err != nil { + log.Warn("Trie prefetcher failed opening storage trie", "root", obj.data.Root, "err", err) + return err + } + obj.trie = tr + + if tr == nil { + return nil + } + for slot := range obj.dirtyStorage { + slot := slot + // tr.GetStorage(addr, slot[:]) + workers.Go(func() error { + _, err := tr.GetStorage(addr, slot[:]) + return err + }) + } + } + return nil + }) + } + workers.Wait() + + obj, exist := s.stateObjects[params.BeaconRootsAddress] + if exist { + tr, _ := s.db.OpenStorageTrie(s.originalRoot, params.BeaconRootsAddress, obj.data.Root, nil) + obj.trie = tr + } +} + func (s *StateDB) setRefund(gas uint64) { s.refund = gas } @@ -1179,31 +1227,45 @@ func (s *StateDB) Copy() *StateDB { func (s *StateDB) CopyState() *StateDB { // Copy all the basic fields, initialize the memory ones - start := time.Now() state := &StateDB{ db: s.db, + prefetcher: s.prefetcher, trie: s.trie, reader: s.reader, originalRoot: s.originalRoot, stateObjects: make(map[common.Address]*stateObject, len(s.stateObjects)), stateObjectsDestruct: make(map[common.Address]*stateObject, len(s.stateObjectsDestruct)), - mutations: make(map[common.Address]*mutation), - logs: make(map[common.Hash][]*types.Log), - preimages: make(map[common.Hash][]byte), - journal: newJournal(), - updateJournal: newJournal(), - accessList: newAccessList(), - transientStorage: newTransientStorage(), + mutations: make(map[common.Address]*mutation, len(s.mutations)), + dbErr: s.dbErr, + refund: s.refund, + thash: s.thash, + txIndex: s.txIndex, + logs: make(map[common.Hash][]*types.Log, len(s.logs)), + logSize: s.logSize, + preimages: maps.Clone(s.preimages), + + // Do we need to copy the access list and transient storage? + // In practice: No. At the start of a transaction, these two lists are empty. + // In practice, we only ever copy state _between_ transactions/blocks, never + // in the middle of a transaction. However, it doesn't cost us much to copy + // empty lists, so we do it anyway to not blow up if we ever decide copy them + // in the middle of a transaction. + accessList: s.accessList.Copy(), + transientStorage: s.transientStorage.Copy(), + journal: s.journal.copy(), + // The update journal is not copies to avoid duplicated updates in later transactions. + blockNumber: s.blockNumber, + updateJournal: newJournal(), + postSnapshot: s.postSnapshot, + postVals: s.postVals, + } + if s.witness != nil { + state.witness = s.witness.Copy() + } + if s.accessEvents != nil { + state.accessEvents = s.accessEvents.Copy() } - StateNewTime += time.Since(start) - // if s.witness != nil { - // state.witness = s.witness.Copy() - // } - // if s.accessEvents != nil { - // state.accessEvents = s.accessEvents.Copy() - // } // Deep copy cached state objects. - start = time.Now() for addr, obj := range s.stateObjects { state.stateObjects[addr] = obj.simpleCopy(state) } @@ -1211,20 +1273,19 @@ func (s *StateDB) CopyState() *StateDB { for addr, obj := range s.stateObjectsDestruct { state.stateObjectsDestruct[addr] = obj.simpleCopy(state) } - StateDeepCpTime += time.Since(start) // Deep copy the object state markers. - // for addr, op := range s.mutations { - // state.mutations[addr] = op.copy() - // } + for addr, op := range s.mutations { + state.mutations[addr] = op.copy() + } // Deep copy the logs occurred in the scope of block - // for hash, logs := range s.logs { - // cpy := make([]*types.Log, len(logs)) - // for i, l := range logs { - // cpy[i] = new(types.Log) - // *cpy[i] = *l - // } - // state.logs[hash] = cpy - // } + for hash, logs := range s.logs { + cpy := make([]*types.Log, len(logs)) + for i, l := range logs { + cpy[i] = new(types.Log) + *cpy[i] = *l + } + state.logs[hash] = cpy + } return state } diff --git a/trie/trie.go b/trie/trie.go index b8e945330c..81b619efd3 100644 --- a/trie/trie.go +++ b/trie/trie.go @@ -21,6 +21,7 @@ import ( "bytes" "errors" "fmt" + "time" "github.com/ethereum/go-ethereum/common" "github.com/ethereum/go-ethereum/core/types" @@ -324,6 +325,8 @@ func (t *Trie) update(key, value []byte) error { return nil } +var ResolveTime = time.Duration(0) + func (t *Trie) insert(n node, prefix, key []byte, value node) (bool, node, error) { if len(key) == 0 { if v, ok := n.(valueNode); ok { @@ -387,7 +390,9 @@ func (t *Trie) insert(n node, prefix, key []byte, value node) (bool, node, error // We've hit a part of the trie that isn't loaded yet. Load // the node and insert into it. This leaves all child nodes on // the path to the value in the trie. + start := time.Now() rn, err := t.resolveAndTrack(n, prefix) + ResolveTime += time.Since(start) if err != nil { return false, nil, err } @@ -517,7 +522,10 @@ func (t *Trie) delete(n node, prefix, key []byte) (bool, node, error) { // shortNode{..., shortNode{...}}. Since the entry // might not be loaded yet, resolve it just for this // check. + start := time.Now() cnode, err := t.resolve(n.Children[pos], append(prefix, byte(pos))) + ResolveTime += time.Since(start) + if err != nil { return false, nil, err } @@ -548,7 +556,9 @@ func (t *Trie) delete(n node, prefix, key []byte) (bool, node, error) { // We've hit a part of the trie that isn't loaded yet. Load // the node and delete from it. This leaves all child nodes on // the path to the value in the trie. + start := time.Now() rn, err := t.resolveAndTrack(n, prefix) + ResolveTime += time.Since(start) if err != nil { return false, nil, err }