core: coordinate the state prefetcher with block processing (#35404)

This PR coordinates the prefetcher with the main tx executor. Block
processing publishes the index of the transaction it is executing,
prefetch workers skip anything already reached and transactions above 1M
gas are promoted to the front of the prefetch queue while the rest keeps
block order.

core/blockchain.go

@@ -113,6 +113,7 @@ var (
 	blockPrefetchInterruptMeter  = metrics.NewRegisteredMeter("chain/prefetch/interrupts", nil)
 	blockPrefetchTxsInvalidMeter = metrics.NewRegisteredMeter("chain/prefetch/txs/invalid", nil)
 	blockPrefetchTxsValidMeter   = metrics.NewRegisteredMeter("chain/prefetch/txs/valid", nil)
+	blockPrefetchTxsSkippedMeter = metrics.NewRegisteredMeter("chain/prefetch/txs/skipped", nil)
 
 	errInsertionInterrupted = errors.New("insertion is interrupted")
 	errChainStopped         = errors.New("blockchain is stopped")
@@ -2133,9 +2134,11 @@ func (bc *BlockChain) ProcessBlock(ctx context.Context, parentRoot common.Hash,
 		startTime = time.Now()
 		statedb   *state.StateDB
 		interrupt atomic.Bool
+		execIndex atomic.Int64
 		sdb       state.Database
 	)
 	defer interrupt.Store(true) // terminate the prefetch at the end
+	execIndex.Store(-1)         // no transaction executed yet
 
 	if bc.chainConfig.IsUBT(block.Number(), block.Time()) {
 		sdb = state.NewUBTDatabase(bc.triedb, bc.codedb)
@@ -2193,7 +2196,7 @@ func (bc *BlockChain) ProcessBlock(ctx context.Context, parentRoot common.Hash,
 			// Disable tracing for prefetcher executions.
 			vmCfg := bc.cfg.VmConfig
 			vmCfg.Tracer = nil
-			bc.prefetcher.Prefetch(block, throwaway, bc.jumpDestCache, vmCfg, &interrupt)
+			bc.prefetcher.Prefetch(block, throwaway, bc.jumpDestCache, vmCfg, &interrupt, &execIndex)
 
 			blockPrefetchExecuteTimer.Update(time.Since(start))
 			if interrupt.Load() {
@@ -2239,7 +2242,7 @@ func (bc *BlockChain) ProcessBlock(ctx context.Context, parentRoot common.Hash,
 	// Process block using the parent state as reference point
 	pstart := time.Now()
 	pctx, _, spanEnd := telemetry.StartSpan(ctx, "bc.processor.Process")
-	res, err := bc.processor.Process(pctx, block, statedb, bc.jumpDestCache, bc.cfg.VmConfig)
+	res, err := bc.processor.Process(pctx, block, statedb, bc.jumpDestCache, bc.cfg.VmConfig, &execIndex)
 	spanEnd(&err)
 	if err != nil {
 		bc.reportBadBlock(block, res, err)

core/blockchain_test.go

@@ -160,7 +160,7 @@ func testBlockChainImport(chain types.Blocks, blockchain *BlockChain) error {
 		if err != nil {
 			return err
 		}
-		res, err := blockchain.processor.Process(context.Background(), block, statedb, nil, vm.Config{})
+		res, err := blockchain.processor.Process(context.Background(), block, statedb, nil, vm.Config{}, nil)
 		if err != nil {
 			blockchain.reportBadBlock(block, res, err)
 			return err

core/state_prefetcher.go

@@ -19,6 +19,7 @@ package core
 import (
 	"bytes"
 	"runtime"
+	"sort"
 	"sync/atomic"
 
 	"github.com/ethereum/go-ethereum/common"
@@ -49,24 +50,33 @@ func newStatePrefetcher(config *params.ChainConfig, chain *HeaderChain) *statePr
 // Prefetch processes the state changes according to the Ethereum rules by running
 // the transaction messages using the statedb, but any changes are discarded. The
 // only goal is to warm the state caches.
-func (p *statePrefetcher) Prefetch(block *types.Block, statedb *state.StateDB, jumpDestCache vm.JumpDestCache, cfg vm.Config, interrupt *atomic.Bool) {
+func (p *statePrefetcher) Prefetch(block *types.Block, statedb *state.StateDB, jumpDestCache vm.JumpDestCache, cfg vm.Config, interrupt *atomic.Bool, execIndex *atomic.Int64) {
 	var (
 		fails   atomic.Int64
+		skips   atomic.Int64
 		header  = block.Header()
 		signer  = types.MakeSigner(p.config, header.Number, header.Time)
 		workers errgroup.Group
 		reader  = statedb.Reader()
+		txs     = block.Transactions()
 	)
 	workers.SetLimit(max(1, 4*runtime.NumCPU()/5)) // Aggressively run the prefetching
 
 	// Iterate over and process the individual transactions
-	for i, tx := range block.Transactions() {
+	for _, n := range prefetchOrder(txs) {
+		i, tx := n, txs[n]
 		stateCpy := statedb.Copy() // closure
 		workers.Go(func() error {
 			// If block precaching was interrupted, abort
 			if interrupt != nil && interrupt.Load() {
 				return nil
 			}
+			// Skip transactions the main pass has already reached, warming
+			// them up can not help anymore.
+			if execIndex != nil && execIndex.Load() >= int64(i) {
+				skips.Add(1)
+				return nil
+			}
 			// Preload the touched accounts and storage slots in advance
 			sender, err := types.Sender(signer, tx)
 			if err != nil {
@@ -120,7 +130,31 @@ func (p *statePrefetcher) Prefetch(block *types.Block, statedb *state.StateDB, j
 	}
 	workers.Wait()
 
-	blockPrefetchTxsValidMeter.Mark(int64(len(block.Transactions())) - fails.Load())
+	blockPrefetchTxsValidMeter.Mark(int64(len(txs)) - fails.Load() - skips.Load())
 	blockPrefetchTxsInvalidMeter.Mark(fails.Load())
+	blockPrefetchTxsSkippedMeter.Mark(skips.Load())
 	return
 }
+
+// prefetchPromoteGas is the gas limit above which a transaction is promoted
+// to the front of the prefetch queue. Below it the worker pool keeps up with
+// the main pass in block order anyway.
+const prefetchPromoteGas = 1_000_000
+
+// prefetchOrder returns the submission order of the block transactions.
+// Heavy transactions go first, giving them a head start over the main
+// pass, while the rest stays in block order.
+func prefetchOrder(txs types.Transactions) []int {
+	order := make([]int, len(txs))
+	for i := range order {
+		order[i] = i
+	}
+	sort.SliceStable(order, func(a, b int) bool {
+		gasA, gasB := txs[order[a]].Gas(), txs[order[b]].Gas()
+		if gasA < prefetchPromoteGas && gasB < prefetchPromoteGas {
+			return false // regular transactions keep block order
+		}
+		return gasA > gasB // heavier transactions go first
+	})
+	return order
+}

core/state_processor.go

@@ -20,6 +20,7 @@ import (
 	"context"
 	"fmt"
 	"math/big"
+	"sync/atomic"
 
 	"github.com/ethereum/go-ethereum/common"
 	"github.com/ethereum/go-ethereum/consensus"
@@ -63,7 +64,7 @@ func (p *StateProcessor) chainConfig() *params.ChainConfig {
 // Process returns the receipts and logs accumulated during the process and
 // returns the amount of gas that was used in the process. If any of the
 // transactions failed to execute due to insufficient gas it will return an error.
-func (p *StateProcessor) Process(ctx context.Context, block *types.Block, statedb *state.StateDB, jumpDestCache vm.JumpDestCache, cfg vm.Config) (*ProcessResult, error) {
+func (p *StateProcessor) Process(ctx context.Context, block *types.Block, statedb *state.StateDB, jumpDestCache vm.JumpDestCache, cfg vm.Config, execIndex *atomic.Int64) (*ProcessResult, error) {
 	var (
 		config      = p.chainConfig()
 		receipts    = make(types.Receipts, 0, len(block.Transactions()))
@@ -101,6 +102,10 @@ func (p *StateProcessor) Process(ctx context.Context, block *types.Block, stated
 
 	// Iterate over and process the individual transactions
 	for i, tx := range block.Transactions() {
+		// Publish the progress, letting the prefetcher skip caught up work.
+		if execIndex != nil {
+			execIndex.Store(int64(i))
+		}
 		msg, err := TransactionToMessage(tx, signer, header.BaseFee)
 		if err != nil {
 			return nil, fmt.Errorf("could not apply tx %d [%v]: %w", i, tx.Hash().Hex(), err)

core/stateless.go

@@ -68,7 +68,7 @@ func ExecuteStateless(ctx context.Context, config *params.ChainConfig, vmconfig
 	validator := NewBlockValidator(config, nil) // No chain, we only validate the state, not the block
 
 	// Run the stateless blocks processing and self-validate certain fields
-	res, err := processor.Process(ctx, block, db, nil, vmconfig)
+	res, err := processor.Process(ctx, block, db, nil, vmconfig, nil)
 	if err != nil {
 		return common.Hash{}, common.Hash{}, err
 	}

core/types.go

@@ -42,7 +42,7 @@ type Prefetcher interface {
 	// Prefetch processes the state changes according to the Ethereum rules by running
 	// the transaction messages using the statedb, but any changes are discarded. The
 	// only goal is to pre-cache transaction signatures and state trie nodes.
-	Prefetch(block *types.Block, statedb *state.StateDB, jumpDestCache vm.JumpDestCache, cfg vm.Config, interrupt *atomic.Bool)
+	Prefetch(block *types.Block, statedb *state.StateDB, jumpDestCache vm.JumpDestCache, cfg vm.Config, interrupt *atomic.Bool, execIndex *atomic.Int64)
 }
 
 // Processor is an interface for processing blocks using a given initial state.
@@ -50,7 +50,7 @@ type Processor interface {
 	// Process processes the state changes according to the Ethereum rules by running
 	// the transaction messages using the statedb and applying any rewards to both
 	// the processor (coinbase) and any included uncles.
-	Process(ctx context.Context, block *types.Block, statedb *state.StateDB, jumpDestCache vm.JumpDestCache, cfg vm.Config) (*ProcessResult, error)
+	Process(ctx context.Context, block *types.Block, statedb *state.StateDB, jumpDestCache vm.JumpDestCache, cfg vm.Config, execIndex *atomic.Int64) (*ProcessResult, error)
 }
 
 // ProcessResult contains the values computed by Process.

eth/state_accessor.go

@@ -151,7 +151,7 @@ func (eth *Ethereum) hashState(ctx context.Context, block *types.Block, base *st
 		if current = eth.blockchain.GetBlockByNumber(next); current == nil {
 			return nil, nil, fmt.Errorf("block #%d not found", next)
 		}
-		_, err := eth.blockchain.Processor().Process(ctx, current, statedb, nil, vm.Config{})
+		_, err := eth.blockchain.Processor().Process(ctx, current, statedb, nil, vm.Config{}, nil)
 		if err != nil {
 			return nil, nil, fmt.Errorf("processing block %d failed: %v", current.NumberU64(), err)
 		}