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) }