Skip to content

Commit

Permalink
Trie prefetch on state pretch (bnb-chain#996)
Browse files Browse the repository at this point in the history
* 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.
  • Loading branch information
setunapo authored and brilliant-lx committed Aug 1, 2022
1 parent df3e1be commit 77c8372
Show file tree
Hide file tree
Showing 5 changed files with 31 additions and 26 deletions.
4 changes: 2 additions & 2 deletions core/blockchain.go
Expand Up @@ -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
Expand Down
30 changes: 20 additions & 10 deletions core/state/statedb.go
Expand Up @@ -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
}
Expand Down Expand Up @@ -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,
Expand Down Expand Up @@ -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
Expand Down
12 changes: 2 additions & 10 deletions core/state/trie_prefetcher.go
Expand Up @@ -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
}
}
}
Expand Down Expand Up @@ -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...)
Expand All @@ -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) {
Expand Down
9 changes: 6 additions & 3 deletions core/state_prefetcher.go
Expand Up @@ -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)
Expand Down Expand Up @@ -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)
Expand Down Expand Up @@ -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)
}

}
2 changes: 1 addition & 1 deletion miner/worker.go
Expand Up @@ -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 {
Expand Down

0 comments on commit 77c8372

Please sign in to comment.