From 77c8372cc424d517ff43b288d073ae380d37c1c5 Mon Sep 17 00:00:00 2001 From: setunapo <98502954+setunapo@users.noreply.github.com> Date: Tue, 26 Jul 2022 22:23:55 +0800 Subject: [PATCH] Trie prefetch on state pretch (#996) * feature: do trie prefetch on state prefetch Currently, state prefetch just pre execute the transactions and discard the results. It is helpful to increase the snapshot cache hit rate. It would be more helpful, if it can do trie prefetch at the same time, since the it will preload the trie node and build the trie tree in advance. This patch is to implement it, by reusing the main trie prefetch and doing finalize after transaction is executed. * some code improvements for trie prefetch ** increase pendingSize before dispatch tasks ** use throwaway StateDB for TriePrefetchInAdvance and remove the prefetcherLock ** remove the necessary drain operation in trie prefetch mainloop, trie prefetcher won't be used after close. --- core/blockchain.go | 4 ++-- core/state/statedb.go | 30 ++++++++++++++++++++---------- core/state/trie_prefetcher.go | 12 ++---------- core/state_prefetcher.go | 9 ++++++--- miner/worker.go | 2 +- 5 files changed, 31 insertions(+), 26 deletions(-) diff --git a/core/blockchain.go b/core/blockchain.go index 2a509f919..60bee8576 100644 --- a/core/blockchain.go +++ b/core/blockchain.go @@ -1892,13 +1892,13 @@ func (bc *BlockChain) insertChain(chain types.Blocks, verifySeals, setHead bool) // do Prefetch in a separate goroutine to avoid blocking the critical path // 1.do state prefetch for snapshot cache - throwaway := statedb.Copy() + throwaway := statedb.CopyDoPrefetch() go bc.prefetcher.Prefetch(block, throwaway, &bc.vmConfig, interruptCh) // 2.do trie prefetch for MPT trie node cache // it is for the big state trie tree, prefetch based on transaction's From/To address. // trie prefetcher is thread safe now, ok to prefetch in a separate routine - go statedb.TriePrefetchInAdvance(block, signer) + go throwaway.TriePrefetchInAdvance(block, signer) } //Process block using the parent state as reference point diff --git a/core/state/statedb.go b/core/state/statedb.go index ad2638722..ec3d80e7f 100644 --- a/core/state/statedb.go +++ b/core/state/statedb.go @@ -237,10 +237,9 @@ func (s *StateDB) StopPrefetcher() { } func (s *StateDB) TriePrefetchInAdvance(block *types.Block, signer types.Signer) { - s.prefetcherLock.Lock() - prefetcher := s.prefetcher // s.prefetcher could be resetted to nil - s.prefetcherLock.Unlock() - + // s is a temporary throw away StateDB, s.prefetcher won't be resetted to nil + // so no need to add lock for s.prefetcher + prefetcher := s.prefetcher if prefetcher == nil { return } @@ -805,6 +804,17 @@ func (db *StateDB) ForEachStorage(addr common.Address, cb func(key, value common // Copy creates a deep, independent copy of the state. // Snapshots of the copied state cannot be applied to the copy. func (s *StateDB) Copy() *StateDB { + return s.copyInternal(false) +} + +// It is mainly for state prefetcher to do trie prefetch right now. +func (s *StateDB) CopyDoPrefetch() *StateDB { + return s.copyInternal(true) +} + +// If doPrefetch is true, it tries to reuse the prefetcher, the copied StateDB will do active trie prefetch. +// otherwise, just do inactive copy trie prefetcher. +func (s *StateDB) copyInternal(doPrefetch bool) *StateDB { // Copy all the basic fields, initialize the memory ones state := &StateDB{ db: s.db, @@ -871,12 +881,12 @@ func (s *StateDB) Copy() *StateDB { state.accessList = s.accessList.Copy() } - // If there's a prefetcher running, make an inactive copy of it that can - // only access data but does not actively preload (since the user will not - // know that they need to explicitly terminate an active copy). - prefetcher := s.prefetcher - if prefetcher != nil { - state.prefetcher = prefetcher.copy() + state.prefetcher = s.prefetcher + if s.prefetcher != nil && !doPrefetch { + // If there's a prefetcher running, make an inactive copy of it that can + // only access data but does not actively preload (since the user will not + // know that they need to explicitly terminate an active copy). + state.prefetcher = state.prefetcher.copy() } if s.snaps != nil { // In order for the miner to be able to use and make additions diff --git a/core/state/trie_prefetcher.go b/core/state/trie_prefetcher.go index 9004f489e..dd22f9d08 100644 --- a/core/state/trie_prefetcher.go +++ b/core/state/trie_prefetcher.go @@ -172,15 +172,7 @@ func (p *triePrefetcher) mainLoop() { p.fetchersMutex.Lock() p.fetchers = nil p.fetchersMutex.Unlock() - - // drain all the channels before quit the loop - for { - select { - case <-p.prefetchChan: - default: - return - } - } + return } } } @@ -362,6 +354,7 @@ func newSubfetcher(db Database, root common.Hash, accountHash common.Hash) *subf // schedule adds a batch of trie keys to the queue to prefetch. func (sf *subfetcher) schedule(keys [][]byte) { + atomic.AddUint32(&sf.pendingSize, uint32(len(keys))) // Append the tasks to the current queue sf.lock.Lock() sf.tasks = append(sf.tasks, keys...) @@ -371,7 +364,6 @@ func (sf *subfetcher) schedule(keys [][]byte) { case sf.wake <- struct{}{}: default: } - atomic.AddUint32(&sf.pendingSize, uint32(len(keys))) } func (sf *subfetcher) scheduleParallel(keys [][]byte) { diff --git a/core/state_prefetcher.go b/core/state_prefetcher.go index d39f3e4e2..4dffe2234 100644 --- a/core/state_prefetcher.go +++ b/core/state_prefetcher.go @@ -58,7 +58,7 @@ func (p *statePrefetcher) Prefetch(block *types.Block, statedb *state.StateDB, c // No need to execute the first batch, since the main processor will do it. for i := 0; i < prefetchThread; i++ { go func() { - newStatedb := statedb.Copy() + newStatedb := statedb.CopyDoPrefetch() newStatedb.EnableWriteOnSharedStorage() gaspool := new(GasPool).AddGas(block.GasLimit()) blockContext := NewEVMBlockContext(header, p.bc, nil) @@ -100,7 +100,7 @@ func (p *statePrefetcher) PrefetchMining(txs *types.TransactionsByPriceAndNonce, for i := 0; i < prefetchThread; i++ { go func(startCh <-chan *types.Transaction, stopCh <-chan struct{}) { idx := 0 - newStatedb := statedb.Copy() + newStatedb := statedb.CopyDoPrefetch() newStatedb.EnableWriteOnSharedStorage() gaspool := new(GasPool).AddGas(gasLimit) blockContext := NewEVMBlockContext(header, p.bc, nil) @@ -153,5 +153,8 @@ func precacheTransaction(msg types.Message, config *params.ChainConfig, gaspool // Update the evm with the new transaction context. evm.Reset(NewEVMTxContext(msg), statedb) // Add addresses to access list if applicable - ApplyMessage(evm, msg, gaspool) + if _, err := ApplyMessage(evm, msg, gaspool); err == nil { + statedb.Finalise(true) + } + } diff --git a/miner/worker.go b/miner/worker.go index 0658dbd92..7a7eaf14c 100644 --- a/miner/worker.go +++ b/miner/worker.go @@ -900,7 +900,7 @@ func (w *worker) commitTransactions(env *environment, txs *types.TransactionsByP txsPrefetch := txs.Copy() tx := txsPrefetch.Peek() txCurr := &tx - w.prefetcher.PrefetchMining(txsPrefetch, env.header, env.gasPool.Gas(), env.state.Copy(), *w.chain.GetVMConfig(), interruptCh, txCurr) + w.prefetcher.PrefetchMining(txsPrefetch, env.header, env.gasPool.Gas(), env.state.CopyDoPrefetch(), *w.chain.GetVMConfig(), interruptCh, txCurr) LOOP: for {