From ff44a796a191a3717640de179ea685d7c12d694b Mon Sep 17 00:00:00 2001 From: Jonny Rhea <5555162+jrhea@users.noreply.github.com> Date: Fri, 24 Jul 2026 02:43:46 -0500 Subject: [PATCH] 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 | 7 +++++-- core/blockchain_test.go | 2 +- core/state_prefetcher.go | 40 +++++++++++++++++++++++++++++++++++++--- core/state_processor.go | 7 ++++++- core/stateless.go | 2 +- core/types.go | 4 ++-- eth/state_accessor.go | 2 +- 7 files changed, 53 insertions(+), 11 deletions(-) diff --git a/core/blockchain.go b/core/blockchain.go index 798d3f1ae6..0c495966a5 100644 --- a/core/blockchain.go +++ b/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) diff --git a/core/blockchain_test.go b/core/blockchain_test.go index 4530895b9d..18b9f837f4 100644 --- a/core/blockchain_test.go +++ b/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 diff --git a/core/state_prefetcher.go b/core/state_prefetcher.go index c635481730..93181f5451 100644 --- a/core/state_prefetcher.go +++ b/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 +} diff --git a/core/state_processor.go b/core/state_processor.go index 27009cb49a..ea677d42d3 100644 --- a/core/state_processor.go +++ b/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) diff --git a/core/stateless.go b/core/stateless.go index f8a3f81b12..3cbdc549d7 100644 --- a/core/stateless.go +++ b/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 } diff --git a/core/types.go b/core/types.go index f6f15101e0..76354f81dd 100644 --- a/core/types.go +++ b/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. diff --git a/eth/state_accessor.go b/eth/state_accessor.go index 7bcd83d581..ebf6053921 100644 --- a/eth/state_accessor.go +++ b/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) }