prefetch snap data

core/blockchain.go

@@ -90,6 +90,7 @@ const (
  maxFutureBlocks = 256
  maxTimeFutureBlocks = 30
  maxBeyondBlocks = 2048
+ prefetchTxNumber = 100
 
  diffLayerFreezerRecheckInterval = 3 * time.Second
  diffLayerPruneRecheckInterval = 1 * time.Second // The interval to prune unverified diff layers
@@ -2071,12 +2072,11 @@ func (bc *BlockChain) insertChain(chain types.Blocks, verifySeals bool) (int, er
  // Enable prefetching to pull in trie node paths while processing transactions
  statedb.StartPrefetcher("chain")
  var followupInterrupt uint32
- if followup, err := it.peek(); followup != nil && err == nil {
+ if len(block.Transactions()) >= prefetchTxNumber {
  throwaway, _ := state.New(parent.Root, bc.stateCache, bc.snaps)
-
  go func(start time.Time, followup *types.Block, throwaway *state.StateDB, interrupt *uint32) {
  bc.prefetcher.Prefetch(followup, throwaway, bc.vmConfig, &followupInterrupt)
- }(time.Now(), followup, throwaway, &followupInterrupt)
+ }(time.Now(), block, throwaway, &followupInterrupt)
  }
  //Process block using the parent state as reference point
  substart := time.Now()

core/state_prefetcher.go

@@ -17,6 +17,7 @@
 package core
 
 import (
+ "runtime"
  "sync/atomic"
 
  "github.com/ethereum/go-ethereum/consensus"
@@ -49,28 +50,43 @@ func NewStatePrefetcher(config *params.ChainConfig, bc *BlockChain, engine conse
 // only goal is to pre-cache transaction signatures and state trie nodes.
 func (p *statePrefetcher) Prefetch(block *types.Block, statedb *state.StateDB, cfg vm.Config, interrupt *uint32) {
  var (
- header = block.Header()
- gaspool = new(GasPool).AddGas(block.GasLimit())
- blockContext = NewEVMBlockContext(header, p.bc, nil)
- evm = vm.NewEVM(blockContext, vm.TxContext{}, statedb, p.config, cfg)
- signer = types.MakeSigner(p.config, header.Number)
+ header = block.Header()
+ signer = types.MakeSigner(p.config, header.Number)
  )
- // Iterate over and process the individual transactions
- for i, tx := range block.Transactions() {
- // If block precaching was interrupted, abort
- if interrupt != nil && atomic.LoadUint32(interrupt) == 1 {
- return
- }
- // Convert the transaction into an executable message and pre-cache its sender
- msg, err := tx.AsMessage(signer)
- if err != nil {
- return // Also invalid block, bail out
- }
- statedb.Prepare(tx.Hash(), block.Hash(), i)
- if err := precacheTransaction(msg, p.config, gaspool, statedb, header, evm); err != nil {
- return // Ugh, something went horribly wrong, bail out
+ transactions := block.Transactions()
+ threads := runtime.NumCPU()
+ batch := len(transactions) / (threads + 1)
+
+ for i := 1; i <= threads; i++ {
+ start := i * batch
+ end := (i + 1) * batch
+ if i == threads {
+ end = len(transactions)
  }
+ go func(start, end int) {
+ newStatedb := statedb.Copy()
+ gaspool := new(GasPool).AddGas(block.GasLimit())
+ blockContext := NewEVMBlockContext(header, p.bc, nil)
+ evm := vm.NewEVM(blockContext, vm.TxContext{}, statedb, p.config, cfg)
+ // Iterate over and process the individual transactions
+ for i, tx := range transactions[start:end] {
+ // If block precaching was interrupted, abort
+ if interrupt != nil && atomic.LoadUint32(interrupt) == 1 {
+ return
+ }
+ // Convert the transaction into an executable message and pre-cache its sender
+ msg, err := tx.AsMessage(signer)
+ if err != nil {
+ return // Also invalid block, bail out
+ }
+ newStatedb.Prepare(tx.Hash(), block.Hash(), i)
+ if err := precacheTransaction(msg, p.config, gaspool, newStatedb, header, evm); err != nil {
+ return // Ugh, something went horribly wrong, bail out
+ }
+ }
+ }(start, end)
  }
+
 }
 
 // precacheTransaction attempts to apply a transaction to the given state database