diff --git a/core/blockchain.go b/core/blockchain.go index 7027a17ddd..efc71cba34 100644 --- a/core/blockchain.go +++ b/core/blockchain.go @@ -2128,6 +2128,104 @@ type ExecuteConfig struct { EnableWitnessStats bool } +// useBALExecution reports whether the block will be executed through the +// BAL-driven parallel processor. +func (bc *BlockChain) useBALExecution(block *types.Block, wantWitness bool) bool { + return supportsParallelExecution(block, bc.chainConfig, wantWitness, bc.cfg.VmConfig.Tracer != nil, bc.cfg.VmConfig.DisableParallelExecution) +} + +// setupExecutionState builds the state instance that block execution reads from +// and writes to. +// +// - BAL-driven parallel execution (Amsterdam blocks carrying an access list): +// a single reader(the underlying state reader wrapped with a shared cache +// and an access-list-hint prefetcher) feeds both the canonical state and +// every per-transaction state built on top of it. +// +// - Sequential execution with prefetching: the main processor and a +// speculative whole-block prefetcher share one cached reader. +// +// - No prefetching: a plain reader, with a no-op cleanup. +func (bc *BlockChain) setupExecutionState(parentRoot common.Hash, block *types.Block, config ExecuteConfig, interrupt *atomic.Bool, execIndex *atomic.Int64) (*state.StateDB, func(*blockProcessingResult), error) { + noop := func(*blockProcessingResult) {} + + var sdb state.Database + if bc.chainConfig.IsUBT(block.Number(), block.Time()) { + sdb = state.NewUBTDatabase(bc.triedb, bc.codedb) + } else { + sdb = state.NewMPTDatabase(bc.triedb, bc.codedb).WithSnapshot(bc.snaps) + } + type prewarmReader interface { + // ReadersWithCacheStats creates a pair of state readers that share the + // same underlying state reader and internal state cache, while maintaining + // separate statistics respectively. + ReadersWithCacheStats(stateRoot common.Hash) (state.Reader, state.Reader, error) + } + wantWitness := config.StatelessSelfValidation || config.MakeWitness + + switch warmer, ok := sdb.(prewarmReader); { + case bc.useBALExecution(block, wantWitness): + base, err := sdb.Reader(parentRoot) + if err != nil { + return nil, nil, err + } + reader, stop := state.NewBlockExecutionReader(base, prefetchHint(block.AccessList()), runtime.NumCPU()) + statedb, err := state.NewWithReader(parentRoot, sdb, reader) + if err != nil { + stop() + return nil, nil, err + } + return statedb, func(*blockProcessingResult) { stop() }, nil + + case bc.cfg.NoPrefetch || !ok: + statedb, err := state.New(parentRoot, sdb) + if err != nil { + return nil, nil, err + } + return statedb, noop, nil + + default: + // The main processor and the speculative prefetcher share the same reader + // with a local cache for mitigating the overhead of state access. + prefetch, process, err := warmer.ReadersWithCacheStats(parentRoot) + if err != nil { + return nil, nil, err + } + throwaway, err := state.NewWithReader(parentRoot, sdb, prefetch) + if err != nil { + return nil, nil, err + } + statedb, err := state.NewWithReader(parentRoot, sdb, process) + if err != nil { + return nil, nil, err + } + go func(start time.Time) { + // Disable tracing for prefetcher executions. + vmCfg := bc.cfg.VmConfig + vmCfg.Tracer = nil + bc.prefetcher.Prefetch(block, throwaway, bc.jumpDestCache, bc.precompileCache.PrefetchView(), vmCfg, interrupt, execIndex) + + blockPrefetchExecuteTimer.Update(time.Since(start)) + if interrupt.Load() { + blockPrefetchInterruptMeter.Mark(1) + } + }(time.Now()) + + return statedb, func(result *blockProcessingResult) { + // Upload the statistics of reader at the end. + if result == nil { + return + } + if stater, ok := prefetch.(state.ReaderStater); ok { + result.stats.StatePrefetchCacheStats = stater.GetStats() + } + if stater, ok := process.(state.ReaderStater); ok { + result.stats.StateReadCacheStats = stater.GetStats() + } + }, nil + } +} + // ProcessBlock executes and validates the given block. If there was no error // it writes the block and associated state to database. func (bc *BlockChain) ProcessBlock(ctx context.Context, parentRoot common.Hash, block *types.Block, config ExecuteConfig) (result *blockProcessingResult, blockEndErr error) { @@ -2137,75 +2235,17 @@ func (bc *BlockChain) ProcessBlock(ctx context.Context, parentRoot common.Hash, 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) - } else { - sdb = state.NewMPTDatabase(bc.triedb, bc.codedb).WithSnapshot(bc.snaps) - } - // If prefetching is enabled, run that against the current state to pre-cache - // transactions and probabilistically some of the account/storage trie nodes. - // - // Note: the main processor and prefetcher share the same reader with a local - // cache for mitigating the overhead of state access. - type prewarmReader interface { - // ReadersWithCacheStats creates a pair of state readers that share the - // same underlying state reader and internal state cache, while maintaining - // separate statistics respectively. - ReadersWithCacheStats(stateRoot common.Hash) (state.Reader, state.Reader, error) - } - warmer, ok := sdb.(prewarmReader) - - if bc.cfg.NoPrefetch || !ok { - statedb, err = state.New(parentRoot, sdb) - if err != nil { - return nil, err - } - } else { - // If prefetching is enabled, run that against the current state to pre-cache - // transactions and probabilistically some of the account/storage trie nodes. - // - // Note: the main processor and prefetcher share the same reader with a local - // cache for mitigating the overhead of state access. - prefetch, process, err := warmer.ReadersWithCacheStats(parentRoot) - if err != nil { - return nil, err - } - throwaway, err := state.NewWithReader(parentRoot, sdb, prefetch) - if err != nil { - return nil, err - } - statedb, err = state.NewWithReader(parentRoot, sdb, process) - if err != nil { - return nil, err - } - // Upload the statistics of reader at the end - defer func() { - if result != nil { - if stater, ok := prefetch.(state.ReaderStater); ok { - result.stats.StatePrefetchCacheStats = stater.GetStats() - } - if stater, ok := process.(state.ReaderStater); ok { - result.stats.StateReadCacheStats = stater.GetStats() - } - } - }() - go func(start time.Time, throwaway *state.StateDB, block *types.Block) { - // Disable tracing for prefetcher executions. - vmCfg := bc.cfg.VmConfig - vmCfg.Tracer = nil - bc.prefetcher.Prefetch(block, throwaway, bc.jumpDestCache, bc.precompileCache.PrefetchView(), vmCfg, &interrupt, &execIndex) - - blockPrefetchExecuteTimer.Update(time.Since(start)) - if interrupt.Load() { - blockPrefetchInterruptMeter.Mark(1) - } - }(time.Now(), throwaway, block) + // Set up the state reader feeding execution, along with a cleanup to run once + // processing is complete (stop the prefetcher, upload reader statistics). + statedb, cleanup, err := bc.setupExecutionState(parentRoot, block, config, &interrupt, &execIndex) + if err != nil { + return nil, err } + defer func() { cleanup(result) }() // If we are past Byzantium, enable prefetching to pull in trie node paths // while processing transactions. Before Byzantium the prefetcher is mostly @@ -2220,7 +2260,11 @@ func (bc *BlockChain) ProcessBlock(ctx context.Context, parentRoot common.Hash, if err != nil { return nil, err } + defer witness.ReportMetrics(block.NumberU64()) } + // The prefetcher warms trie node paths in the background. + // - Sequential execution feeds it from the EVM as it touches state; + // - BAL-driven parallel execution feeds it from the block access list; statedb.StartPrefetcher("chain", witness) defer statedb.StopPrefetcher() } @@ -2347,10 +2391,6 @@ func (bc *BlockChain) ProcessBlock(ctx context.Context, parentRoot common.Hash, stats.DatabaseCommit = statedb.DatabaseCommits // Database commits are complete, we can mark them stats.BlockWrite = time.Since(wstart) - max(statedb.AccountCommits, statedb.StorageCommits) /* concurrent */ - statedb.DatabaseCommits } - // Report the collected witness statistics - if witness != nil { - witness.ReportMetrics(block.NumberU64()) - } elapsed := time.Since(startTime) + 1 // prevent zero division stats.TotalTime = elapsed stats.MgasPerSecond = float64(res.GasUsed) * 1000 / float64(elapsed) diff --git a/core/chain_makers.go b/core/chain_makers.go index f02e0341bb..de0b585d12 100644 --- a/core/chain_makers.go +++ b/core/chain_makers.go @@ -97,11 +97,10 @@ func (b *BlockGen) Difficulty() *big.Int { } // SetParentBeaconRoot sets the parent beacon root field of the generated -// block. +// block. The corresponding EIP-4788 system call is applied later, during block +// finalization, so that generation mirrors the real block processor. func (b *BlockGen) SetParentBeaconRoot(root common.Hash) { b.header.ParentBeaconRoot = &root - blockContext := NewEVMBlockContext(b.header, b.cm, &b.header.Coinbase) - ProcessBeaconBlockRoot(root, vm.NewEVM(blockContext, b.statedb, b.cm.config, vm.Config{}), b.bal) } // addTx adds a transaction to the generated block. If no coinbase has @@ -404,6 +403,22 @@ func GenerateChain(config *params.ChainConfig, parent *types.Block, engine conse gen(i, b) } + // EIP-4788: process the parent beacon block root as a pre-execution + // system call. + // + // It is applied after the gen callback so an explicit SetParentBeaconRoot + // is honored; ProcessBeaconBlockRoot pins the write to block-access index 0, + // so it is recorded as pre-execution regardless of this ordering. + // + // TODO(rjl493456442) rework the chain maker, replacing the individual calls + // with PreExecution. + if b.header.ParentBeaconRoot != nil { + blockContext := NewEVMBlockContext(b.header, cm, &b.header.Coinbase) + blockContext.Random = &common.Hash{} // enable post-merge instruction set + evm := vm.NewEVM(blockContext, statedb, cm.config, vm.Config{}) + ProcessBeaconBlockRoot(*b.header.ParentBeaconRoot, evm, b.bal) + } + requests, bal := b.collectRequests(false) if requests != nil { reqHash := types.CalcRequestsHash(requests) diff --git a/core/eip7928_test.go b/core/eip7928_test.go index 816161a34f..c70eed081e 100644 --- a/core/eip7928_test.go +++ b/core/eip7928_test.go @@ -18,18 +18,23 @@ package core import ( "bytes" + "context" "crypto/ecdsa" "maps" "math/big" "testing" "github.com/ethereum/go-ethereum/common" + "github.com/ethereum/go-ethereum/consensus" "github.com/ethereum/go-ethereum/consensus/beacon" "github.com/ethereum/go-ethereum/consensus/ethash" + "github.com/ethereum/go-ethereum/core/rawdb" "github.com/ethereum/go-ethereum/core/types" "github.com/ethereum/go-ethereum/core/types/bal" + "github.com/ethereum/go-ethereum/core/vm" "github.com/ethereum/go-ethereum/crypto" "github.com/ethereum/go-ethereum/params" + "github.com/ethereum/go-ethereum/trie" "github.com/holiman/uint256" ) @@ -90,9 +95,94 @@ func (e *balTestEnv) run(t *testing.T, gen func(*BlockGen)) (*bal.BlockAccessLis if blocks[0].AccessList() == nil { t.Fatal("expected non-nil block access list") } + assertParallelEquiv(t, e.gspec, engine, blocks[0]) + return blocks[0].AccessList(), receipts[0] } +// assertParallelEquiv re-executes a sequentially-generated block through both +// the BAL-driven parallel processor and the sequential processor and asserts +// they agree. +// +// Two independent properties are checked: +// +// - Parallel execution reproduces the committed block: it reconstructs the +// block's state root from the block-level access list and its receipts and +// gas from re-execution. +// +// - The parallel and sequential processors rebuild the identical access list +// and agree on gas, receipts and requests. This is the property that would +// break if parallel execution diverged from sequential. +func assertParallelEquiv(t *testing.T, gspec *Genesis, engine consensus.Engine, block *types.Block) { + t.Helper() + if block.AccessList() == nil { + return // not a parallel-eligible block + } + bc, err := NewBlockChain(rawdb.NewMemoryDatabase(), gspec, engine, nil) + if err != nil { + t.Fatalf("new blockchain: %v", err) + } + defer bc.Stop() + + // Parallel path (default for Amsterdam blocks carrying an access list). + parState, err := bc.State() + if err != nil { + t.Fatalf("state: %v", err) + } + parRes, err := NewStateProcessor(bc).Process(context.Background(), block, parState, nil, nil, vm.Config{}, nil) + if err != nil { + t.Fatalf("parallel process: %v", err) + } + parRoot := parState.IntermediateRoot(gspec.Config.IsEIP158(block.Number())) + + // Sequential path, forced explicitly via DisableParallelExecution. + seqState, err := bc.State() + if err != nil { + t.Fatalf("state: %v", err) + } + seqRes, err := NewStateProcessor(bc).Process(context.Background(), block, seqState, nil, nil, vm.Config{DisableParallelExecution: true}, nil) + if err != nil { + t.Fatalf("sequential process: %v", err) + } + + // Parallel execution must reconstruct the committed block. + if parRoot != block.Root() { + t.Fatalf("parallel state root %x != committed %x", parRoot, block.Root()) + } + if parRes.GasUsed != block.GasUsed() { + t.Fatalf("parallel gas used %d != committed %d", parRes.GasUsed, block.GasUsed()) + } + if got := types.DeriveSha(parRes.Receipts, trie.NewStackTrie(nil)); got != block.ReceiptHash() { + t.Fatalf("parallel receipt root %x != committed %x", got, block.ReceiptHash()) + } + if p, s := parRes.Bal.ToEncodingObj().Hash(), *block.BlockAccessListHash(); p != s { + t.Fatalf("parallel access list hash %x != committed %x", p, s) + } + if parRes.Requests == nil { + t.Fatalf("parallel requests is nil") + } + if p, s := types.CalcRequestsHash(parRes.Requests), *block.RequestsHash(); p != s { + t.Fatalf("parallel requests hash %x != committed %x", p, s) + } + + // Parallel and sequential must agree on every re-executed output. + if p, s := parRes.Bal.ToEncodingObj().Hash(), seqRes.Bal.ToEncodingObj().Hash(); p != s { + t.Fatalf("rebuilt access list hash: parallel %x != sequential %x", p, s) + } + if parRes.GasUsed != seqRes.GasUsed { + t.Fatalf("gas used: parallel %d != sequential %d", parRes.GasUsed, seqRes.GasUsed) + } + if p, s := types.DeriveSha(parRes.Receipts, trie.NewStackTrie(nil)), types.DeriveSha(seqRes.Receipts, trie.NewStackTrie(nil)); p != s { + t.Fatalf("receipt root: parallel %x != sequential %x", p, s) + } + if seqRes.Requests == nil { + t.Fatalf("seqentual requests is nil") + } + if p, s := types.CalcRequestsHash(parRes.Requests), types.CalcRequestsHash(seqRes.Requests); p != s { + t.Fatalf("requests hash: parallel %x != sequential %x", p, s) + } +} + // --- assertion helpers --- func findAccount(b *bal.BlockAccessList, addr common.Address) *bal.AccountAccess { diff --git a/core/eip8037_test.go b/core/eip8037_test.go index 1aab804ae5..1d22c2d641 100644 --- a/core/eip8037_test.go +++ b/core/eip8037_test.go @@ -21,6 +21,7 @@ package core import ( + "context" "errors" "math/big" "testing" @@ -28,6 +29,7 @@ import ( "github.com/ethereum/go-ethereum/common" "github.com/ethereum/go-ethereum/consensus/beacon" "github.com/ethereum/go-ethereum/consensus/ethash" + "github.com/ethereum/go-ethereum/core/rawdb" "github.com/ethereum/go-ethereum/core/state" "github.com/ethereum/go-ethereum/core/tracing" "github.com/ethereum/go-ethereum/core/types" @@ -968,3 +970,49 @@ func TestSystemCallNotCountedInBlock(t *testing.T) { t.Fatalf("block gas used = %d, want 0 (system calls excluded)", blocks[0].GasUsed()) } } + +func TestParallelReservationOverflowRejected(t *testing.T) { + env := newBALTestEnv(nil) + env.gspec.GasLimit = 30_000_000 + engine := beacon.New(ethash.NewFaker()) + + // A single self-transfer with a 5,000,000 gas limit but only ~21,000 of + // actual usage (recipient exists, no new state). + to := env.from + _, blocks, _ := GenerateChainWithGenesis(env.gspec, engine, 1, func(_ int, b *BlockGen) { + b.AddTx(env.tx(0, &to, big.NewInt(1), 5_000_000, 0, nil)) + }) + valid := blocks[0] + + bc, err := NewBlockChain(rawdb.NewMemoryDatabase(), env.gspec, engine, nil) + if err != nil { + t.Fatalf("new blockchain: %v", err) + } + defer bc.Stop() + + // The block as built (30M limit, well above the 5M reservation) is accepted: + // the reservation check must not over-reject valid blocks. + statedb, err := bc.State() + if err != nil { + t.Fatalf("state: %v", err) + } + if _, err := NewStateProcessor(bc).Process(context.Background(), valid, statedb, nil, nil, vm.Config{}, nil); err != nil { + t.Fatalf("valid block rejected by parallel processor: %v", err) + } + + // Lower the block gas limit below the transaction's worst-case reservation + // (5,000,000) while keeping it above the actual usage (~21,000). The + // transaction can no longer be admitted, so the block is invalid. + hdr := valid.Header() + hdr.GasLimit = 100_000 + invalid := valid.WithSeal(hdr) + + statedb, err = bc.State() + if err != nil { + t.Fatalf("state: %v", err) + } + _, err = NewStateProcessor(bc).Process(context.Background(), invalid, statedb, nil, nil, vm.Config{}, nil) + if !errors.Is(err, ErrGasLimitReached) { + t.Fatalf("parallel processor accepted a reservation-overflow block (err = %v), want ErrGasLimitReached", err) + } +} diff --git a/core/gaspool.go b/core/gaspool.go index 83420c7640..dbd0c1a49b 100644 --- a/core/gaspool.go +++ b/core/gaspool.go @@ -109,6 +109,19 @@ func (gp *GasPool) CumulativeUsed() uint64 { return gp.cumulativeUsed } +// CumulativeRegular returns the cumulative regular-dimension gas consumed +// (EIP-8037). It is used to derive the block gas used when transactions are +// charged against independent pools during parallel execution. +func (gp *GasPool) CumulativeRegular() uint64 { + return gp.cumulativeRegular +} + +// CumulativeState returns the cumulative state-dimension gas consumed +// (EIP-8037). See CumulativeRegular for the rationale. +func (gp *GasPool) CumulativeState() uint64 { + return gp.cumulativeState +} + // Used returns the amount of consumed gas. func (gp *GasPool) Used() uint64 { // After 8037, return max(sum_regular, sum_state) diff --git a/core/state/reader.go b/core/state/reader.go index be07cec0f9..473b2a69a9 100644 --- a/core/state/reader.go +++ b/core/state/reader.go @@ -382,30 +382,44 @@ func (r *multiStateReader) Storage(addr common.Address, slot common.Hash) (commo return common.Hash{}, errors.Join(errs...) } +const stateReaderCacheBuckets = 64 + // stateReaderWithCache is a wrapper around StateReader that maintains additional // state caches to support concurrent state access. type stateReaderWithCache struct { StateReader - // Previously resolved state entries. - accounts map[common.Address]*types.StateAccount - accountLock sync.RWMutex + // Account buckets are selected by account address. This reader is typically + // used in scenarios requiring concurrent access to accounts; multiple buckets + // reduce lock contention. + accountBuckets [stateReaderCacheBuckets]struct { + lock sync.RWMutex + accounts map[common.Address]*types.StateAccount + } - // List of storage buckets, each of which is thread-safe. - // This reader is typically used in scenarios requiring concurrent - // access to storage. Using multiple buckets helps mitigate - // the overhead caused by locking. - storageBuckets [16]struct { + // Storage buckets are selected by both account address and storage key. This + // avoids serializing accesses to distinct slots of the same account. + storageBuckets [stateReaderCacheBuckets]struct { lock sync.RWMutex storages map[common.Address]map[common.Hash]common.Hash } } +func accountCacheBucket(addr common.Address) int { + return int(addr[0] & (stateReaderCacheBuckets - 1)) +} + +func storageCacheBucket(addr common.Address, slot common.Hash) int { + return int((addr[0] ^ slot[0] ^ slot[len(slot)-1]) & (stateReaderCacheBuckets - 1)) +} + // newStateReaderWithCache constructs the state reader with local cache. func newStateReaderWithCache(sr StateReader) *stateReaderWithCache { r := &stateReaderWithCache{ StateReader: sr, - accounts: make(map[common.Address]*types.StateAccount), + } + for i := range r.accountBuckets { + r.accountBuckets[i].accounts = make(map[common.Address]*types.StateAccount) } for i := range r.storageBuckets { r.storageBuckets[i].storages = make(map[common.Address]map[common.Hash]common.Hash) @@ -419,10 +433,12 @@ func newStateReaderWithCache(sr StateReader) *stateReaderWithCache { // // An error will be returned if the state is corrupted in the underlying reader. func (r *stateReaderWithCache) account(addr common.Address) (*types.StateAccount, bool, error) { + bucket := &r.accountBuckets[accountCacheBucket(addr)] + // Try to resolve the requested account in the local cache - r.accountLock.RLock() - acct, ok := r.accounts[addr] - r.accountLock.RUnlock() + bucket.lock.RLock() + acct, ok := bucket.accounts[addr] + bucket.lock.RUnlock() if ok { return acct, true, nil } @@ -431,9 +447,9 @@ func (r *stateReaderWithCache) account(addr common.Address) (*types.StateAccount if err != nil { return nil, false, err } - r.accountLock.Lock() - r.accounts[addr] = acct - r.accountLock.Unlock() + bucket.lock.Lock() + bucket.accounts[addr] = acct + bucket.lock.Unlock() return acct, false, nil } @@ -453,7 +469,7 @@ func (r *stateReaderWithCache) storage(addr common.Address, slot common.Hash) (c var ( value common.Hash ok bool - bucket = &r.storageBuckets[addr[0]&0x0f] + bucket = &r.storageBuckets[storageCacheBucket(addr, slot)] ) // Try to resolve the requested storage slot in the local cache bucket.lock.RLock() diff --git a/core/state/reader_eip_7928.go b/core/state/reader_eip_7928.go index ff315ac5eb..8dcf063bb5 100644 --- a/core/state/reader_eip_7928.go +++ b/core/state/reader_eip_7928.go @@ -22,6 +22,7 @@ import ( "github.com/ethereum/go-ethereum/common" "github.com/ethereum/go-ethereum/core/types" "github.com/ethereum/go-ethereum/core/types/bal" + "github.com/ethereum/go-ethereum/crypto" ) // The EIP27928 reader utilizes a hierarchical architecture to optimize state @@ -86,7 +87,6 @@ type prefetchStateReader struct { closeOnce sync.Once } -// nolint:unused func newPrefetchStateReader(reader StateReader, accessList map[common.Address][]common.Hash, nThreads int) *prefetchStateReader { tasks := make([]*fetchTask, 0, len(accessList)) for addr, slots := range accessList { @@ -193,13 +193,28 @@ func (r *prefetchStateReader) process(start, limit int) { } } +// NewBlockExecutionReader wraps base with a shared, concurrency-safe cache so +// that any state resolved once, whether by the background prefetcher or by +// transaction execution, is not fetched from the underlying reader again. +func NewBlockExecutionReader(base Reader, prefetch map[common.Address][]common.Hash, threads int) (Reader, func()) { + var ( + cache = newStateReaderWithCache(base) + stop = func() {} + ) + if len(prefetch) > 0 && threads > 0 { + pf := newPrefetchStateReader(cache, prefetch, threads) + stop = pf.Close + } + return newReader(base, newStateReaderWithStats(cache)), stop +} + // ReaderWithBlockLevelAccessList provides state access that reflects the // pre-transition state combined with the mutations made by transactions // prior to TxIndex. type ReaderWithBlockLevelAccessList struct { Reader - AccessList *bal.ConstructionBlockAccessList - TxIndex int + lookup *bal.Lookup + txIndex uint32 } // NewReaderWithBlockLevelAccessList constructs a reader for accessing states @@ -209,39 +224,85 @@ type ReaderWithBlockLevelAccessList struct { // - 0 for pre‑execution system contract calls. // - 1 … n for transactions (in block order). // - n + 1 for post‑execution system contract calls. -func NewReaderWithBlockLevelAccessList(base Reader, accessList *bal.ConstructionBlockAccessList, txIndex int) *ReaderWithBlockLevelAccessList { +func NewReaderWithBlockLevelAccessList(base Reader, lookup *bal.Lookup, txIndex int) *ReaderWithBlockLevelAccessList { return &ReaderWithBlockLevelAccessList{ - Reader: base, - AccessList: accessList, - TxIndex: txIndex, + Reader: base, + lookup: lookup, + txIndex: uint32(txIndex), } } // Account implements Reader, returning the account with the specific address. +// +// The returned account reflects the pre-transition state overlaid with all +// mutations made by call frames prior to the reader's TxIndex. func (r *ReaderWithBlockLevelAccessList) Account(addr common.Address) (*types.StateAccount, error) { - panic("implement me") + base, err := r.Reader.Account(addr) + if err != nil { + return nil, err + } + balance, nonce, code, hasBalance, hasNonce, hasCode := r.lookup.AccountChanges(addr, r.txIndex) + + // No mutation precedes the current call frame, return the base account as is. + if !hasBalance && !hasNonce && !hasCode { + return base, nil + } + // Overlay the mutations on top of a copy of the base account. The base + // account must not be mutated in place: with a shared cache in front of the + // underlying reader, the same instance is handed to concurrent readers. + account := types.NewEmptyStateAccount() + if base != nil { + account = base.Copy() + } + if hasBalance { + account.Balance = balance.Clone() + } + if hasNonce { + account.Nonce = nonce + } + if hasCode { + if len(code) == 0 { + account.CodeHash = types.EmptyCodeHash.Bytes() + } else { + account.CodeHash = crypto.Keccak256(code) + } + } + return account, nil } // Storage implements Reader, returning the storage slot with the specific // address and slot key. func (r *ReaderWithBlockLevelAccessList) Storage(addr common.Address, slot common.Hash) (common.Hash, error) { - panic("implement me") + if value, ok := r.lookup.Storage(addr, slot, r.txIndex); ok { + return value, nil + } + return r.Reader.Storage(addr, slot) } // Has implements Reader, returning the flag indicating whether the contract // code with specified address and hash exists or not. func (r *ReaderWithBlockLevelAccessList) Has(addr common.Address, codeHash common.Hash) bool { - panic("implement me") + if _, ok := r.lookup.Code(addr, r.txIndex); ok { + return true + } + return r.Reader.Has(addr, codeHash) } // Code implements Reader, returning the contract code with specified address -// and hash. -func (r *ReaderWithBlockLevelAccessList) Code(addr common.Address, codeHash common.Hash) ([]byte, error) { - panic("implement me") +// and hash. Code created earlier in the block (and therefore absent from the +// pre-transition state) is served directly from the access list. +func (r *ReaderWithBlockLevelAccessList) Code(addr common.Address, codeHash common.Hash) []byte { + if code, ok := r.lookup.Code(addr, r.txIndex); ok { + return code + } + return r.Reader.Code(addr, codeHash) } // CodeSize implements Reader, returning the contract code size with specified // address and hash. -func (r *ReaderWithBlockLevelAccessList) CodeSize(addr common.Address, codeHash common.Hash) (int, error) { - panic("implement me") +func (r *ReaderWithBlockLevelAccessList) CodeSize(addr common.Address, codeHash common.Hash) int { + if code, ok := r.lookup.Code(addr, r.txIndex); ok { + return len(code) + } + return r.Reader.CodeSize(addr, codeHash) } diff --git a/core/state/statedb_eip_7928.go b/core/state/statedb_eip_7928.go new file mode 100644 index 0000000000..4a643071c1 --- /dev/null +++ b/core/state/statedb_eip_7928.go @@ -0,0 +1,237 @@ +// Copyright 2026 The go-ethereum Authors +// This file is part of the go-ethereum library. +// +// The go-ethereum library is free software: you can redistribute it and/or modify +// it under the terms of the GNU Lesser General Public License as published by +// the Free Software Foundation, either version 3 of the License, or +// (at your option) any later version. +// +// The go-ethereum library is distributed in the hope that it will be useful, +// but WITHOUT ANY WARRANTY; without even the implied warranty of +// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +// GNU Lesser General Public License for more details. +// +// You should have received a copy of the GNU Lesser General Public License +// along with the go-ethereum library. If not, see . + +package state + +import ( + "fmt" + "runtime" + "slices" + "sync" + "sync/atomic" + "time" + + "github.com/ethereum/go-ethereum/common" + "github.com/ethereum/go-ethereum/core/types" + "github.com/ethereum/go-ethereum/core/types/bal" + "github.com/ethereum/go-ethereum/crypto" + "github.com/ethereum/go-ethereum/log" + "golang.org/x/sync/errgroup" +) + +// ApplyBlockAccessList installs the post-state recorded in a block access list +// directly into the state, without executing any transactions. +func (s *StateDB) ApplyBlockAccessList(list *bal.BlockAccessList) error { + if list == nil { + return nil + } + return s.applyBlockAccessList(*list, runtime.GOMAXPROCS(0)) +} + +type balSlot struct { + key common.Hash + value common.Hash +} + +type balAccount struct { + access *bal.AccountAccess + slots []balSlot + obj *stateObject // nil if the account ends up untouched +} + +// hasMetadataChange reports whether the account has a metadata mutation. +func (a *balAccount) hasMetadataChange() bool { + return len(a.access.BalanceChanges) != 0 || len(a.access.NonceChanges) != 0 || len(a.access.CodeChanges) != 0 +} + +// mutated extends the mutation check with storage. +func (a *balAccount) mutated() bool { + return a.hasMetadataChange() || len(a.slots) != 0 +} + +// balApplyContext holds the shared context through the concurrent workers. +type balApplyContext struct { + accountReads atomic.Int64 + storageReads atomic.Int64 + prefetchMu sync.Mutex +} + +func (s *StateDB) applyBlockAccessList(list bal.BlockAccessList, threads int) error { + var ( + accounts = make([]*balAccount, 0, len(list)) + addresses = make([]common.Address, 0, len(list)) + ) + for i := range list { + access := &(list)[i] + entry := &balAccount{access: access} + + for j := range access.StorageChanges { + change := &access.StorageChanges[j] + if n := len(change.SlotChanges); n > 0 { + entry.slots = append(entry.slots, balSlot{ + key: change.Slot.Bytes32(), + value: change.SlotChanges[n-1].PostValue.Bytes32(), + }) + } + } + // Skip the read-only account. It is a cheap validation by checking + // purely with the access list. Whether the account is truly mutated + // is only known once its pre-state value is read. + if !entry.mutated() { + continue + } + accounts = append(accounts, entry) + addresses = append(addresses, access.Address) + } + // Schedule background warming of the account trie for every mutated account. + if s.prefetcher != nil && len(addresses) > 0 { + if err := s.prefetcher.prefetch(common.Hash{}, s.originalRoot, common.Address{}, addresses, nil, false); err != nil { + log.Error("Failed to prefetch account trie", "err", err) + } + } + // Process the accounts by applying the final value of mutated fields. + var ba balApplyContext + if err := parallelBALApply(len(accounts), threads, func(i int) error { + return s.prepareBALAccount(accounts[i], &ba) + }); err != nil { + return err + } + var storageLoaded int + for _, entry := range accounts { + storageLoaded += len(entry.slots) + + obj := entry.obj + if obj == nil { + continue + } + if obj.empty() { + s.markDelete(obj.address) + s.stateObjectsDestruct[obj.address] = obj + } else { + s.markUpdate(obj.address) + s.setStateObject(obj) + } + } + s.AccountLoaded += len(addresses) + s.AccountReads += time.Duration(ba.accountReads.Load()) + s.StorageLoaded += storageLoaded + s.StorageReads += time.Duration(ba.storageReads.Load()) + return nil +} + +// prepareBALAccount loads one account's pre-state, schedules warming of the +// trie nodes required to hash its mutations, and builds the resulting state +// object. +func (s *StateDB) prepareBALAccount(entry *balAccount, ba *balApplyContext) error { + // Resolve the account object from the database. + var ( + addr = entry.access.Address + start = time.Now() + ) + account, err := s.reader.Account(addr) + ba.accountReads.Add(int64(time.Since(start))) + if err != nil { + return fmt.Errorf("load account %x: %w", addr, err) + } + obj := newObject(s, addr, account) + + // Apply the final value of each mutated field. + if n := len(entry.access.BalanceChanges); n > 0 { + obj.setBalance(entry.access.BalanceChanges[n-1].PostBalance.Clone()) + } + if n := len(entry.access.NonceChanges); n > 0 { + obj.setNonce(entry.access.NonceChanges[n-1].PostNonce) + } + if n := len(entry.access.CodeChanges); n > 0 { + code := entry.access.CodeChanges[n-1].NewCode + obj.setCode(crypto.Keccak256Hash(code), slices.Clone(code)) + } + if err := s.applyBALStorage(obj, entry.slots, ba); err != nil { + return err + } + // Drop accounts whose writes all reverted to their original value. + if !entry.hasMetadataChange() && len(obj.pendingStorage) == 0 { + return nil + } + entry.obj = obj + return nil +} + +// applyBALStorage schedules warming of the account's storage trie and stages the +// writes that actually change a slot's value. The storage trie itself is left +// unopened on the object; IntermediateRoot pulls the warmed trie back from the +// prefetcher (or opens it lazily if prefetching is disabled). +func (s *StateDB) applyBALStorage(obj *stateObject, slots []balSlot, ba *balApplyContext) error { + if len(slots) == 0 { + return nil + } + addr := obj.address + + // Schedule background warming of the storage trie paths to the mutated slots. + if obj.data.Root != types.EmptyRootHash && s.prefetcher != nil { + keys := make([]common.Hash, len(slots)) + for i := range slots { + keys[i] = slots[i].key + } + ba.prefetchMu.Lock() + s.prefetcher.prefetch(obj.addrHash(), obj.data.Root, addr, nil, keys, false) + ba.prefetchMu.Unlock() + } + // Stage the writes that differ from the slot's pre-state value. + for i := range slots { + start := time.Now() + origin, err := s.reader.Storage(addr, slots[i].key) + ba.storageReads.Add(int64(time.Since(start))) + if err != nil { + return fmt.Errorf("load storage %x/%x: %w", addr, slots[i].key, err) + } + if slots[i].value == origin { + continue // slot ended the block at its original value + } + obj.originStorage[slots[i].key] = origin + obj.pendingStorage[slots[i].key] = slots[i].value + obj.uncommittedStorage[slots[i].key] = origin + } + return nil +} + +// parallelBALApply invokes apply for every index in [0, tasks) across at most +// workers goroutines, returning the first error reported by any of them. +func parallelBALApply(tasks, workers int, apply func(int) error) error { + if tasks == 0 { + return nil + } + workers = min(max(workers, 1), tasks) + + var ( + next atomic.Uint64 + group errgroup.Group + ) + for range workers { + group.Go(func() error { + for { + i := int(next.Add(1)) - 1 + if i >= tasks { + return nil + } + if err := apply(i); err != nil { + return err + } + } + }) + } + return group.Wait() +} diff --git a/core/state/statedb_eip_7928_test.go b/core/state/statedb_eip_7928_test.go new file mode 100644 index 0000000000..d96aad8fde --- /dev/null +++ b/core/state/statedb_eip_7928_test.go @@ -0,0 +1,161 @@ +// Copyright 2026 The go-ethereum Authors +// This file is part of the go-ethereum library. +// +// The go-ethereum library is free software: you can redistribute it and/or modify +// it under the terms of the GNU Lesser General Public License as published by +// the Free Software Foundation, either version 3 of the License, or +// (at your option) any later version. +// +// The go-ethereum library is distributed in the hope that it will be useful, +// but WITHOUT ANY WARRANTY; without even the implied warranty of +// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +// GNU Lesser General Public License for more details. +// +// You should have received a copy of the GNU Lesser General Public License +// along with the go-ethereum library. If not, see . + +package state + +import ( + "testing" + + "github.com/ethereum/go-ethereum/common" + "github.com/ethereum/go-ethereum/core/tracing" + "github.com/ethereum/go-ethereum/core/types" + "github.com/ethereum/go-ethereum/core/types/bal" + "github.com/holiman/uint256" +) + +// TestApplyBlockAccessListConcurrentPrefetch stresses the concurrent storage +// prefetch scheduling with many contracts (each carrying a non-empty storage +// trie), so that several BAL workers schedule prefetches into the shared, +// non-thread-safe prefetcher at once. Run with -race to detect unguarded +// concurrent access. The resulting root must still match the sequential one. +func TestApplyBlockAccessListConcurrentPrefetch(t *testing.T) { + const n = 64 + code := []byte{0x60, 0x00, 0x60, 0x00} + addrOf := func(i int) common.Address { + return common.BigToAddress(uint256.NewInt(uint64(0x1000 + i)).ToBig()) + } + slot := common.HexToHash("0x01") + + // Base state: n contracts, each with a pre-existing storage slot so its + // storage root is non-empty. + db := NewDatabaseForTesting() + base, _ := New(types.EmptyRootHash, db) + for i := range n { + addr := addrOf(i) + base.SetBalance(addr, uint256.NewInt(100), tracing.BalanceChangeUnspecified) + base.SetCode(addr, code, tracing.CodeChangeUnspecified) + base.SetState(addr, slot, common.HexToHash("0xaa")) + } + root0, err := base.Commit(0, false, false) + if err != nil { + t.Fatalf("commit base: %v", err) + } + + mutate := func(s *StateDB) { + for i := range n { + addr := addrOf(i) + s.SetBalance(addr, uint256.NewInt(uint64(200+i)), tracing.BalanceChangeUnspecified) + s.SetState(addr, slot, common.BigToHash(uint256.NewInt(uint64(i+1)).ToBig())) + } + } + seq, _ := New(root0, db) + mutate(seq) + wantRoot := seq.IntermediateRoot(true) + + cb := bal.NewConstructionBlockAccessList() + for i := range n { + addr := addrOf(i) + cb.BalanceChange(0, addr, uint256.NewInt(uint64(200+i))) + cb.StorageWrite(0, addr, slot, common.BigToHash(uint256.NewInt(uint64(i+1)).ToBig())) + } + balState, _ := New(root0, db) + balState.StartPrefetcher("test", nil) + if err := balState.ApplyBlockAccessList(cb.ToEncodingObj()); err != nil { + balState.StopPrefetcher() + t.Fatalf("apply block access list: %v", err) + } + gotRoot := balState.IntermediateRoot(true) + balState.StopPrefetcher() + + if gotRoot != wantRoot { + t.Fatalf("BAL apply root = %x, want %x", gotRoot, wantRoot) + } +} + +// TestApplyBlockAccessListMatchesSequential checks that installing a block's +// post-state through ApplyBlockAccessList yields exactly the same state root as +// applying the same mutations one by one. The BAL path warms both the account +// trie and the storage tries through the prefetcher and pulls them back at +// IntermediateRoot, so this also exercises that machinery. Run with -race to +// catch data races in the concurrent prefetch scheduling. +func TestApplyBlockAccessListMatchesSequential(t *testing.T) { + var ( + existing = common.HexToAddress("0x1111") + contract = common.HexToAddress("0x2222") + fresh = common.HexToAddress("0x3333") + + slotA = common.HexToHash("0x01") + slotB = common.HexToHash("0x02") + code = []byte{0x60, 0x00, 0x60, 0x00} + ) + + // Build a base state with a plain account and a contract that already has + // some storage (so its storage root is non-empty and the storage-trie + // prefetch path is exercised). + db := NewDatabaseForTesting() + base, _ := New(types.EmptyRootHash, db) + base.SetBalance(existing, uint256.NewInt(1000), tracing.BalanceChangeUnspecified) + base.SetNonce(existing, 1, tracing.NonceChangeUnspecified) + base.SetBalance(contract, uint256.NewInt(50), tracing.BalanceChangeUnspecified) + base.SetCode(contract, code, tracing.CodeChangeUnspecified) + base.SetState(contract, slotA, common.HexToHash("0xaa")) + base.SetState(contract, slotB, common.HexToHash("0xbb")) + root0, err := base.Commit(0, false, false) + if err != nil { + t.Fatalf("commit base: %v", err) + } + + // mutate applies the block's post-state via the ordinary setters. + mutate := func(s *StateDB) { + s.SetBalance(existing, uint256.NewInt(1234), tracing.BalanceChangeUnspecified) + s.SetNonce(existing, 2, tracing.NonceChangeUnspecified) + s.SetState(contract, slotA, common.HexToHash("0xcc")) // changed + s.SetState(contract, slotB, common.HexToHash("0xbb")) // unchanged, must be a no-op + s.SetBalance(fresh, uint256.NewInt(7), tracing.BalanceChangeUnspecified) + s.SetNonce(fresh, 1, tracing.NonceChangeUnspecified) + } + + // Sequential reference root. + seq, _ := New(root0, db) + mutate(seq) + wantRoot := seq.IntermediateRoot(true) + if wantRoot == root0 { + t.Fatal("mutations did not change the state root") + } + + // Same post-state expressed as a block access list. + cb := bal.NewConstructionBlockAccessList() + cb.BalanceChange(0, existing, uint256.NewInt(1234)) + cb.NonceChange(existing, 0, 2) + cb.StorageWrite(0, contract, slotA, common.HexToHash("0xcc")) + cb.StorageWrite(0, contract, slotB, common.HexToHash("0xbb")) // no-op write + cb.BalanceChange(0, fresh, uint256.NewInt(7)) + cb.NonceChange(fresh, 0, 1) + list := cb.ToEncodingObj() + + balState, _ := New(root0, db) + balState.StartPrefetcher("test", nil) + if err := balState.ApplyBlockAccessList(list); err != nil { + balState.StopPrefetcher() + t.Fatalf("apply block access list: %v", err) + } + gotRoot := balState.IntermediateRoot(true) + balState.StopPrefetcher() + + if gotRoot != wantRoot { + t.Fatalf("BAL apply root = %x, want %x", gotRoot, wantRoot) + } +} diff --git a/core/state_processor.go b/core/state_processor.go index c962c3dc5e..a68cd1091c 100644 --- a/core/state_processor.go +++ b/core/state_processor.go @@ -65,6 +65,9 @@ func (p *StateProcessor) chainConfig() *params.ChainConfig { // 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, precompileCache *vm.PrecompileCache, cfg vm.Config, execIndex *atomic.Int64) (*ProcessResult, error) { + if supportsParallelExecution(block, p.chainConfig(), statedb.Witness() != nil, cfg.Tracer != nil, cfg.DisableParallelExecution) { + return p.processParallel(ctx, block, statedb, jumpDestCache, cfg) + } var ( config = p.chainConfig() receipts = make(types.Receipts, 0, len(block.Transactions())) diff --git a/core/state_processor_parallel.go b/core/state_processor_parallel.go new file mode 100644 index 0000000000..fd627184ef --- /dev/null +++ b/core/state_processor_parallel.go @@ -0,0 +1,362 @@ +// Copyright 2026 The go-ethereum Authors +// This file is part of the go-ethereum library. +// +// The go-ethereum library is free software: you can redistribute it and/or modify +// it under the terms of the GNU Lesser General Public License as published by +// the Free Software Foundation, either version 3 of the License, or +// (at your option) any later version. +// +// The go-ethereum library is distributed in the hope that it will be useful, +// but WITHOUT ANY WARRANTY; without even the implied warranty of +// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +// GNU Lesser General Public License for more details. +// +// You should have received a copy of the GNU Lesser General Public License +// along with the go-ethereum library. If not, see . + +package core + +import ( + "context" + "fmt" + "runtime" + "sync/atomic" + "time" + + "github.com/ethereum/go-ethereum/common" + "github.com/ethereum/go-ethereum/core/state" + "github.com/ethereum/go-ethereum/core/types" + "github.com/ethereum/go-ethereum/core/types/bal" + "github.com/ethereum/go-ethereum/core/vm" + "github.com/ethereum/go-ethereum/log" + "github.com/ethereum/go-ethereum/metrics" + "github.com/ethereum/go-ethereum/params" + "golang.org/x/sync/errgroup" +) + +// Per-phase timers for BAL-driven parallel block execution. +var ( + parallelSystemExecTimer = metrics.NewRegisteredResettingTimer("chain/execution/parallel/system", nil) + parallelTxExecTimer = metrics.NewRegisteredResettingTimer("chain/execution/parallel/transactions", nil) + parallelStateHashTimer = metrics.NewRegisteredResettingTimer("chain/execution/parallel/statehash", nil) + parallelTotalTimer = metrics.NewRegisteredResettingTimer("chain/execution/parallel/total", nil) + parallelAccountCacheHitMeter = metrics.NewRegisteredMeter("chain/execution/parallel/reads/account/cache/hit", nil) + parallelAccountCacheMissMeter = metrics.NewRegisteredMeter("chain/execution/parallel/reads/account/cache/miss", nil) + parallelStorageCacheHitMeter = metrics.NewRegisteredMeter("chain/execution/parallel/reads/storage/cache/hit", nil) + parallelStorageCacheMissMeter = metrics.NewRegisteredMeter("chain/execution/parallel/reads/storage/cache/miss", nil) +) + +// supportsParallelExecution reports whether the block can be executed using the +// BAL-driven parallel processor. +func supportsParallelExecution(block *types.Block, config *params.ChainConfig, wantWitness bool, wantTrace bool, disableParallel bool) bool { + // Parallel execution explicitly disabled via config (e.g. by tests that + // want to force the sequential path). + if disableParallel { + return false + } + // No tracer is attached (tracing requires the strict sequential + // ordering of state operations that parallel execution does not + // preserve). + if wantTrace { + return false + } + // No witness is being collected (witness building must observe + // every state access alongside the proof). + if wantWitness { + return false + } + // Disable the parallel execution if either the Amsterdam hasn't been + // activated, or the accessList is not accessible. + return block.AccessList() != nil && config.IsAmsterdam(block.Number(), block.Time()) +} + +// txExecResult holds the per-transaction outcome of parallel execution. +type txExecResult struct { + receipt *types.Receipt + accessList *bal.ConstructionBlockAccessList + + // regular and state are the EIP-8037 per-transaction + // gas contributions to the two block-inclusion dimensions. + regular uint64 + state uint64 +} + +// processParallel executes the block's transactions concurrently using the +// block-level access list. +func (p *StateProcessor) processParallel(ctx context.Context, block *types.Block, statedb *state.StateDB, jumpDestCache vm.JumpDestCache, cfg vm.Config) (*ProcessResult, error) { + var ( + config = p.chainConfig() + header = block.Header() + txs = block.Transactions() + start = time.Now() + + signer = types.MakeSigner(config, header.Number, header.Time) + context = NewEVMBlockContext(header, p.chain, nil) + postIndex = uint32(len(txs) + 1) + db = statedb.Database() + + accessList = block.AccessList() + lookup = accessList.Lookup() + + // blockAccessList is the access list rebuilt from the actual execution. + blockAccessList = bal.NewConstructionBlockAccessList() + ) + + // Resolve the parent state root, the point all execution reads from. + parent := p.chain.GetHeader(block.ParentHash(), block.NumberU64()-1) + if parent == nil { + return nil, fmt.Errorf("parent header %x not found", block.ParentHash()) + } + parentRoot := parent.Root + + // The base reader: the underlying state reader wrapped with a shared + // cache and an access-list-hint prefetcher. This reader is shared by + // all tx-executors. + base := statedb.Reader() + + // Stats + var ( + systemExec time.Duration + txExec time.Duration + stateApply time.Duration + stateHash time.Duration + ) + // Post-execution state root, computed concurrently with execution. + var wg errgroup.Group + wg.Go(func() error { + start := time.Now() + if err := statedb.ApplyBlockAccessList(accessList); err != nil { + return err + } + stateApply = time.Since(start) + + start = time.Now() + statedb.IntermediateRoot(config.IsEIP158(header.Number)) + stateHash = time.Since(start) + return statedb.Error() + }) + // Ensure the root goroutine has stopped mutating the canonical state before + // returning on any path, including the error paths below. Wait is idempotent, + // so the explicit join on the happy path remains valid. + defer func() { _ = wg.Wait() }() + + // Pre-execution system calls, replayed against an ephemeral access-list + // state at block-access index 0, to contribute their entries to the rebuilt + // access list. + // + // TODO(rjl493456442) both the pre/post execution can be performed alongside + // the transaction execution. Measure the overhead before making the changes. + preStart := time.Now() + preState, err := newAccessListState(db, parentRoot, base, lookup, 0) + if err != nil { + return nil, err + } + preEVM := vm.NewEVM(context, preState, config, cfg) + if jumpDestCache != nil { + preEVM.SetJumpDestCache(jumpDestCache) + } + blockAccessList.Merge(PreExecution(ctx, block.BeaconRoot(), parent, config, preEVM, header.Number, header.Time)) + preEVM.Release() + systemExec += time.Since(preStart) + + // Execute the transactions concurrently. Each transaction runs against its + // own ephemeral state instance, whose reads are served from the block-level + // access list overlaid on the parent state. + txStart := time.Now() + results, err := p.executeTransactionsParallel(block, parentRoot, db, base, lookup, context, signer, jumpDestCache, cfg) + if err != nil { + return nil, err + } + txExec = time.Since(txStart) + + // Gather the per-transaction results in block order and charge their gas into + // a single block-level gas pool, exactly as sequential execution does. + var ( + receipts = make(types.Receipts, 0, len(txs)) + allLogs []*types.Log + gp = NewGasPool(block.GasLimit()) + logIndex uint + ) + for i := range txs { + receipt := results[i].receipt + gasLimit := txs[i].Gas() + if err := gp.CheckGasAmsterdam(min(gasLimit, params.MaxTxGas), gasLimit); err != nil { + return nil, fmt.Errorf("could not apply tx %d [%v]: %w", i, txs[i].Hash().Hex(), err) + } + if err := gp.ChargeGasAmsterdam(results[i].regular, results[i].state, receipt.GasUsed); err != nil { + return nil, fmt.Errorf("could not apply tx %d [%v]: %w", i, txs[i].Hash().Hex(), err) + } + // Correct the receipt object with block-level fields + receipt.CumulativeGasUsed = gp.CumulativeUsed() + for _, lg := range receipt.Logs { + lg.Index = logIndex + logIndex++ + } + receipts = append(receipts, receipt) + allLogs = append(allLogs, receipt.Logs...) + blockAccessList.Merge(results[i].accessList) + } + // Post-execution system calls against an ephemeral access-list state at + // index n+1. + postStart := time.Now() + postState, err := newAccessListState(db, parentRoot, base, lookup, int(postIndex)) + if err != nil { + return nil, err + } + postEVM := vm.NewEVM(context, postState, config, cfg) + if jumpDestCache != nil { + postEVM.SetJumpDestCache(jumpDestCache) + } + requests, postBAL, err := PostExecution(ctx, config, header.Number, header.Time, allLogs, postEVM, postIndex) + postEVM.Release() + if err != nil { + return nil, err + } + blockAccessList.Merge(postBAL) + p.chain.Engine().Finalize(p.chain, header, postState, block.Body(), postIndex, blockAccessList) + systemExec += time.Since(postStart) + + // Join the concurrent root computation. + if err := wg.Wait(); err != nil { + return nil, err + } + parallelSystemExecTimer.Update(systemExec) + parallelTxExecTimer.Update(txExec) + parallelStateHashTimer.Update(stateHash) + parallelTotalTimer.UpdateSince(start) + + log.Debug("Parallel block execution", "number", header.Number, "txs", len(txs), + "system", common.PrettyDuration(systemExec), "txexec", common.PrettyDuration(txExec), + "stateapply", common.PrettyDuration(stateApply), "statehash", common.PrettyDuration(stateHash), + "elapsed", common.PrettyDuration(time.Since(start)), + ) + return &ProcessResult{ + Receipts: receipts, + Requests: requests, + Logs: allLogs, + GasUsed: gp.Used(), + Bal: blockAccessList, + }, nil +} + +// newAccessListState constructs an ephemeral state, reading through base, whose +// view reflects the mutations recorded in the access list for all block-access +// indices below index. +func newAccessListState(db state.Database, parentRoot common.Hash, base state.Reader, lookup *bal.Lookup, index int) (*state.StateDB, error) { + return state.NewWithReader(parentRoot, db, state.NewReaderWithBlockLevelAccessList(base, lookup, index)) +} + +// executeTransactionsParallel applies all transactions to independent, +// access-list-backed state instances using a pool of workers, and returns +// the per-transaction results in block order. +func (p *StateProcessor) executeTransactionsParallel(block *types.Block, parentRoot common.Hash, db state.Database, base state.Reader, lookup *bal.Lookup, context vm.BlockContext, signer types.Signer, jumpDestCache vm.JumpDestCache, cfg vm.Config) ([]txExecResult, error) { + var ( + config = p.chainConfig() + header = block.Header() + blockHash = block.Hash() + blockNumber = block.Number() + txs = block.Transactions() + results = make([]txExecResult, len(txs)) + ) + workers := runtime.GOMAXPROCS(0) + if workers > len(txs) { + workers = len(txs) + } + var ( + cursor atomic.Int64 + group errgroup.Group + ) + for w := 0; w < workers; w++ { + group.Go(func() error { + evm := vm.NewEVM(context, nil, config, cfg) + if jumpDestCache != nil { + evm.SetJumpDestCache(jumpDestCache) + } + defer evm.Release() + + for { + i := int(cursor.Add(1)) - 1 + if i >= len(txs) { + return nil + } + tx := txs[i] + msg, err := TransactionToMessage(tx, signer, header.BaseFee) + if err != nil { + return fmt.Errorf("could not apply tx %d [%v]: %w", i, tx.Hash().Hex(), err) + } + + // Construct the dedicated pre-tx state with the BAL overlay wrapped. + reader := state.NewReaderWithBlockLevelAccessList(base, lookup, i+1) + sdb, err := state.NewWithReader(parentRoot, db, reader) + if err != nil { + return err + } + sdb.SetTxContext(tx.Hash(), i, uint32(i+1)) + evm.SetStateDB(sdb) + + // A transaction-local gas pool, sized to the transaction's own gas + // limit: enough to let the state transition run to completion. + gp := NewGasPool(msg.GasLimit) + receipt, accessList, err := ApplyTransactionWithEVM(msg, gp, sdb, blockNumber, blockHash, context.Time, tx, evm) + if err != nil { + return fmt.Errorf("could not apply tx %d [%v]: %w", i, tx.Hash().Hex(), err) + } + results[i] = txExecResult{ + receipt: receipt, + accessList: accessList, + regular: gp.CumulativeRegular(), + state: gp.CumulativeState(), + } + } + }) + } + if err := group.Wait(); err != nil { + return nil, err + } + reportParallelReadStats(block, base) + return results, nil +} + +// reportParallelReadStats reports the state read statistics. TODO(rjl) integrate +// it into blockchain stats. +func reportParallelReadStats(block *types.Block, reader state.Reader) { + stater, ok := reader.(state.ReaderStater) + if !ok { + return + } + var ( + stats = stater.GetStats().StateStats + accountHit = stats.AccountCacheHit + accountMiss = stats.AccountCacheMiss + storageHit = stats.StorageCacheHit + storageMiss = stats.StorageCacheMiss + ) + parallelAccountCacheHitMeter.Mark(accountHit) + parallelAccountCacheMissMeter.Mark(accountMiss) + parallelStorageCacheHitMeter.Mark(storageHit) + parallelStorageCacheMissMeter.Mark(storageMiss) + + log.Debug("Parallel execution read statistics", "number", block.Number(), + "account.hit", accountHit, "account.miss", accountMiss, + "account.hitrate", stats.AccountCacheHitRate(), + "storage.hit", storageHit, "storage.miss", storageMiss, + "storage.hitrate", stats.StorageCacheHitRate()) +} + +// prefetchHint returns a set of storage slots alongside their account address +// for batch reading. +func prefetchHint(list *bal.BlockAccessList) map[common.Address][]common.Hash { + hint := make(map[common.Address][]common.Hash, len(*list)) + for i := range *list { + acc := &(*list)[i] + slots := make([]common.Hash, 0, len(acc.StorageReads)+len(acc.StorageChanges)) + for _, slot := range acc.StorageReads { + slots = append(slots, slot.Bytes32()) + } + for j := range acc.StorageChanges { + slots = append(slots, acc.StorageChanges[j].Slot.Bytes32()) + } + hint[acc.Address] = slots + } + return hint +} diff --git a/core/types/bal/bal_lookup.go b/core/types/bal/bal_lookup.go new file mode 100644 index 0000000000..77b9014a33 --- /dev/null +++ b/core/types/bal/bal_lookup.go @@ -0,0 +1,127 @@ +// Copyright 2026 The go-ethereum Authors +// This file is part of the go-ethereum library. +// +// The go-ethereum library is free software: you can redistribute it and/or modify +// it under the terms of the GNU Lesser General Public License as published by +// the Free Software Foundation, either version 3 of the License, or +// (at your option) any later version. +// +// The go-ethereum library is distributed in the hope that it will be useful, +// but WITHOUT ANY WARRANTY; without even the implied warranty of +// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +// GNU Lesser General Public License for more details. +// +// You should have received a copy of the GNU Lesser General Public License +// along with the go-ethereum library. If not, see . + +package bal + +import ( + "sort" + + "github.com/ethereum/go-ethereum/common" + "github.com/holiman/uint256" +) + +// accountLookup references an account's per-index mutations. The slices are the +// ones from the encoded access list, which the spec requires to be sorted +// ascending (and unique) by block-access index, so they can be binary-searched +// directly without copying. +type accountLookup struct { + balances []encodingBalanceChange + nonces []encodingAccountNonce + codes []encodingCodeChange + storage map[common.Hash][]encodingStorageWrite +} + +// Lookup is a read-optimized, index-addressable view over a block access list. +type Lookup struct { + accounts map[common.Address]*accountLookup +} + +// Lookup builds a Lookup over the access list. The returned view aliases the +// receiver's slices, so the access list must not be mutated while it is in use. +func (e *BlockAccessList) Lookup() *Lookup { + l := &Lookup{ + accounts: make(map[common.Address]*accountLookup, len(*e)), + } + for i := range *e { + acc := &(*e)[i] + al := &accountLookup{ + balances: acc.BalanceChanges, + nonces: acc.NonceChanges, + codes: acc.CodeChanges, + storage: make(map[common.Hash][]encodingStorageWrite, len(acc.StorageChanges)), + } + for j := range acc.StorageChanges { + sc := &acc.StorageChanges[j] + al.storage[sc.Slot.Bytes32()] = sc.SlotChanges + } + l.accounts[acc.Address] = al + } + return l +} + +// searchLatest returns the entry with the highest block-access index strictly +// below limit, relying on entries being sorted ascending by that index. +func searchLatest[E any](entries []E, limit uint32, index func(E) uint32) (E, bool) { + i := sort.Search(len(entries), func(i int) bool { + return index(entries[i]) >= limit + }) + // All entries satisfy the condition (index >= limit) + if i == 0 { + var zero E + return zero, false + } + return entries[i-1], true +} + +// AccountChanges returns the account field values observed at block-access index +// limit (i.e. the latest mutation recorded strictly before limit). Each boolean +// reports whether the corresponding field was mutated before limit. +func (l *Lookup) AccountChanges(addr common.Address, limit uint32) (balance *uint256.Int, nonce uint64, code []byte, hasBalance, hasNonce, hasCode bool) { + acc, ok := l.accounts[addr] + if !ok { + return nil, 0, nil, false, false, false + } + if e, ok := searchLatest(acc.balances, limit, func(e encodingBalanceChange) uint32 { return e.BlockAccessIndex }); ok { + balance, hasBalance = e.PostBalance, true + } + if e, ok := searchLatest(acc.nonces, limit, func(e encodingAccountNonce) uint32 { return e.BlockAccessIndex }); ok { + nonce, hasNonce = e.PostNonce, true + } + if e, ok := searchLatest(acc.codes, limit, func(e encodingCodeChange) uint32 { return e.BlockAccessIndex }); ok { + code, hasCode = e.NewCode, true + } + return balance, nonce, code, hasBalance, hasNonce, hasCode +} + +// Code returns the contract code observed at block-access index limit, and +// whether the code was set before limit. +func (l *Lookup) Code(addr common.Address, limit uint32) ([]byte, bool) { + acc, ok := l.accounts[addr] + if !ok { + return nil, false + } + if e, ok := searchLatest(acc.codes, limit, func(e encodingCodeChange) uint32 { return e.BlockAccessIndex }); ok { + return e.NewCode, true + } + return nil, false +} + +// Storage returns the value of the storage slot observed at block-access index +// limit, and whether the slot was written before limit. +func (l *Lookup) Storage(addr common.Address, slot common.Hash, limit uint32) (common.Hash, bool) { + acc, ok := l.accounts[addr] + if !ok { + return common.Hash{}, false + } + writes, ok := acc.storage[slot] + if !ok { + return common.Hash{}, false + } + if e, ok := searchLatest(writes, limit, func(e encodingStorageWrite) uint32 { return e.BlockAccessIndex }); ok { + return e.PostValue.Bytes32(), true + } + return common.Hash{}, false +} diff --git a/core/vm/evm.go b/core/vm/evm.go index b94215277f..6cb0516e91 100644 --- a/core/vm/evm.go +++ b/core/vm/evm.go @@ -225,6 +225,11 @@ func (evm *EVM) SetPrecompileCache(cache *PrecompileCache) { evm.precompileCache = cache } +// SetStateDB configures the state for interaction. +func (evm *EVM) SetStateDB(statedb *state.StateDB) { + evm.StateDB = statedb +} + // SetTxContext resets the EVM with a new transaction context. // This is not threadsafe and should only be done very cautiously. func (evm *EVM) SetTxContext(txCtx TxContext) { diff --git a/core/vm/interpreter.go b/core/vm/interpreter.go index c2dfe3769c..8a82e9602d 100644 --- a/core/vm/interpreter.go +++ b/core/vm/interpreter.go @@ -29,9 +29,10 @@ import ( type Config struct { Tracer *tracing.Hooks - NoBaseFee bool // Forces the EIP-1559 baseFee to 0 (needed for 0 price calls) - EnablePreimageRecording bool // Enables recording of SHA3/keccak preimages - ExtraEips []int // Additional EIPS that are to be enabled + NoBaseFee bool // Forces the EIP-1559 baseFee to 0 (needed for 0 price calls) + EnablePreimageRecording bool // Enables recording of SHA3/keccak preimages + ExtraEips []int // Additional EIPS that are to be enabled + DisableParallelExecution bool // Disable parallel block processing } // ScopeContext contains the things that are per-call, such as stack and memory,