From bed539369cbb33745808464000c0db5bf0a69fff Mon Sep 17 00:00:00 2001 From: Po Date: Fri, 30 May 2025 07:47:13 +0200 Subject: [PATCH] add pre-block fetcher and pass test in vanilla block processor --- core/paralle_state_provider.go | 60 +++++++++++++++++++++ core/parallel_state_processor.go | 54 ------------------- core/state/statedb.go | 91 ++++++++++++++++++++++++++++++++ 3 files changed, 151 insertions(+), 54 deletions(-) create mode 100644 core/paralle_state_provider.go diff --git a/core/paralle_state_provider.go b/core/paralle_state_provider.go new file mode 100644 index 0000000000..022547c02d --- /dev/null +++ b/core/paralle_state_provider.go @@ -0,0 +1,60 @@ +package core + +import ( + "fmt" + + "github.com/ethereum/go-ethereum/core/state" + "github.com/ethereum/go-ethereum/core/types" + "github.com/ethereum/go-ethereum/core/vm" +) + +type PreStateType int + +const ( + BALPreState PreStateType = iota + SeqPreState // Only for debug +) + +const preStateType = SeqPreState + +type PreStateProvider interface { + PrestateAtIndex(i int) (*state.StateDB, error) +} + +type SequentialPrestateProvider struct { + statedb *state.StateDB + block *types.Block + gp *GasPool + signer types.Signer + // contex + usedGas *uint64 + evm *vm.EVM +} + +func (s *SequentialPrestateProvider) PrestateAtIndex(i int) (*state.StateDB, error) { + if i < 0 || i > len(s.block.Transactions()) { + return nil, fmt.Errorf("tx index %d out of range [0, %d)", i, len(s.block.Transactions())) + } + if i == 0 { + return s.statedb.Copy(), nil // first transaction uses the original state + } + i = i - 1 + tx := s.block.Transactions()[i] + signer := s.signer + header := s.block.Header() + statedb := s.statedb + blockHash := s.block.Hash() + blockNumber := s.block.Number() + // execute the transaction again to simulate the state changes + 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) + } + statedb.SetTxContext(tx.Hash(), i) + + _, err = ApplyTransactionWithEVM(msg, s.gp, statedb, blockNumber, blockHash, tx, s.usedGas, s.evm) + if err != nil { + return nil, fmt.Errorf("could not apply tx %d [%v]: %w", i, tx.Hash().Hex(), err) + } + return statedb.Copy(), nil +} diff --git a/core/parallel_state_processor.go b/core/parallel_state_processor.go index 6d2b9f9662..66b8f87ece 100644 --- a/core/parallel_state_processor.go +++ b/core/parallel_state_processor.go @@ -14,15 +14,6 @@ import ( "golang.org/x/sync/errgroup" ) -type PreStateType int - -const ( - BALPreState PreStateType = iota - SeqPreState // Only for debug -) - -const preStateType = SeqPreState - type ParallelStateProcessor struct { config *params.ChainConfig // Chain configuration options chain *HeaderChain // Canonical header chain @@ -127,9 +118,6 @@ func (p *ParallelStateProcessor) executeParallel(block *types.Block, statedb *st receipts[i] = receipt postEntries[i] = entries - // if i == len(block.Transactions())-1 { - // *statedb = *cleanStatedb - // } return nil }) } @@ -217,45 +205,3 @@ func ApplyTransactionWithParallelEVM(msg *Message, gp *GasPool, statedb *state.S return MakeReceipt(evm, result, statedb, blockNumber, blockHash, tx, *usedGas, root), entries, nil } - -type PreStateProvider interface { - PrestateAtIndex(i int) (*state.StateDB, error) -} - -type SequentialPrestateProvider struct { - statedb *state.StateDB - block *types.Block - gp *GasPool - signer types.Signer - // contex - usedGas *uint64 - evm *vm.EVM -} - -func (s *SequentialPrestateProvider) PrestateAtIndex(i int) (*state.StateDB, error) { - if i < 0 || i > len(s.block.Transactions()) { - return nil, fmt.Errorf("tx index %d out of range [0, %d)", i, len(s.block.Transactions())) - } - if i == 0 { - return s.statedb.Copy(), nil // first transaction uses the original state - } - i = i - 1 - tx := s.block.Transactions()[i] - signer := s.signer - header := s.block.Header() - statedb := s.statedb - blockHash := s.block.Hash() - blockNumber := s.block.Number() - // execute the transaction again to simulate the state changes - 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) - } - statedb.SetTxContext(tx.Hash(), i) - - _, err = ApplyTransactionWithEVM(msg, s.gp, statedb, blockNumber, blockHash, tx, s.usedGas, s.evm) - if err != nil { - return nil, fmt.Errorf("could not apply tx %d [%v]: %w", i, tx.Hash().Hex(), err) - } - return statedb.Copy(), nil -} diff --git a/core/state/statedb.go b/core/state/statedb.go index 127cf46f7a..82edf4a5bb 100644 --- a/core/state/statedb.go +++ b/core/state/statedb.go @@ -159,6 +159,9 @@ type StateDB struct { StorageLoaded int // Number of storage slots retrieved from the database during the state transition StorageUpdated atomic.Int64 // Number of storage slots updated during the state transition StorageDeleted atomic.Int64 // Number of storage slots deleted during the state transition + + // The block number context for BALs + blockNumber uint64 } type BALType int @@ -276,6 +279,94 @@ func NewWithReader(root common.Hash, db Database, reader Reader) (*StateDB, erro return sdb, nil } +func (s *StateDB) PrefetchStateBAL(blockNumber uint64) { + s.blockNumber = blockNumber + switch balType { + case BalPreblockKeysPostValues: + s.PrefetchBalPreblockKeys() + } +} + +func (s *StateDB) PrefetchBalPreblockKeys() { + log.Info("PrefetchBalPreblockKeys...") + type StorageKV struct { + addr *common.Address + key *common.Hash + val *common.Hash + } + var ( + preBal = AllBlockAccessLists[s.blockNumber] + lenSlots = 0 + workers errgroup.Group + ) + + workers.SetLimit(25) + // Fetch pre-block acccount state + accounts := make(chan *stateObject, len(preBal)) + for _, acl := range preBal { + addr := acl.Address + lenSlots += len(acl.StorageKeys) + + workers.Go(func() error { + acct, err := s.reader.Account(addr) + obj := newObject(s, addr, acct) + accounts <- obj + if err != nil { + log.Error("fail to fetch account from BALs:", addr) + return err + } + return nil + }) + } + workers.Wait() + + for range len(preBal) { + obj := <-accounts + // must set it first to avoid accounts read later in storageBAL fetching + s.setStateObject(obj) + } + + // Fetch pre-block storage state + storages := make(chan *StorageKV, lenSlots) + for _, acl := range preBal { + addr := acl.Address + keys := acl.StorageKeys + for _, key := range keys { + workers.Go(func() error { + val, err := s.reader.Storage(addr, key) + kv := &StorageKV{&addr, &key, &val} + storages <- kv + if err != nil { + log.Error("fail to fetch storage from BALs:", addr, key) + return err + } + return nil + }) + } + } + workers.Wait() + + for range lenSlots { + kv := <-storages + account := s.getStateObject(*kv.addr) + if account != nil { + account.originStorage[*kv.key] = *kv.val + s.setStateObject(account) + } + } +} + +func (s *StateDB) PreStateAtTxIndex(index int) *StateDB { + // 1. Fetch all pre-block state + // 2. Merge with post-state + if balType != BalPreblockKeysPostValues { + panic("PreStateAtTxIndex is only supported with BalPreblockKeysPostValues") + } + + // Merge with BALs post state + return nil +} + // StartPrefetcher initializes a new trie prefetcher to pull in nodes from the // state trie concurrently while the state is mutated so that when we reach the // commit phase, most of the needed data is already hot.