From 4fc3d374dd8e0fba468f103c1ca01b2c1d8b4645 Mon Sep 17 00:00:00 2001 From: Po Date: Wed, 4 Jun 2025 06:27:19 +0200 Subject: [PATCH] seq poststate --- cmd/geth/chaincmd.go | 12 ++ core/paralle_state_provider.go | 2 +- core/parallel_state_processor.go | 55 ++++++--- core/state/journal.go | 6 +- core/state/state_object.go | 21 ++++ core/state/statedb.go | 184 +++++++++++++++++++++++++++++-- 6 files changed, 251 insertions(+), 29 deletions(-) diff --git a/cmd/geth/chaincmd.go b/cmd/geth/chaincmd.go index 70093585b8..69a5ae298c 100644 --- a/cmd/geth/chaincmd.go +++ b/cmd/geth/chaincmd.go @@ -464,6 +464,18 @@ func showMetrics() { fmt.Println("triedbCommitTimer", triedbCommitTimer.Total()) fmt.Println("blockWriteTimer", blockWriteTimer.Total()) + // Parallel + fmt.Println("PrefetchBALTime: ", core.PrefetchBALTime) + fmt.Println("PrefetchMergeTime:", core.PrefetchMergeBALTime) + fmt.Println("StateSetTime: ", state.StateSetTime) + fmt.Println("StateLoadTime: ", state.StateLoadTime) + fmt.Println("StateFinaliTime: ", state.StateFinalizeTime) + fmt.Println("StateCopyTime: ", state.StateCopyTime) + fmt.Println("StateCopyNewTime: ", state.StateNewTime) + fmt.Println("StateDeepCpTime: ", state.StateDeepCpTime) + fmt.Println("ParalleleExeTime: ", core.ParallelExeTime) + fmt.Println("PostMergeTime: ", core.PostMergeTime) + // total fmt.Println("blockInsertTimer", blockInsertTimer.Total()) } diff --git a/core/paralle_state_provider.go b/core/paralle_state_provider.go index 022547c02d..50566b37a9 100644 --- a/core/paralle_state_provider.go +++ b/core/paralle_state_provider.go @@ -15,7 +15,7 @@ const ( SeqPreState // Only for debug ) -const preStateType = SeqPreState +const preStateType = BALPreState type PreStateProvider interface { PrestateAtIndex(i int) (*state.StateDB, error) diff --git a/core/parallel_state_processor.go b/core/parallel_state_processor.go index 66b8f87ece..621948d1f8 100644 --- a/core/parallel_state_processor.go +++ b/core/parallel_state_processor.go @@ -4,6 +4,7 @@ import ( "fmt" "math/big" "runtime" + "time" "github.com/ethereum/go-ethereum/common" "github.com/ethereum/go-ethereum/consensus/misc" @@ -14,6 +15,13 @@ import ( "golang.org/x/sync/errgroup" ) +var ( + PrefetchBALTime = time.Duration(0) + PrefetchMergeBALTime = time.Duration(0) + ParallelExeTime = time.Duration(0) + PostMergeTime = time.Duration(0) +) + type ParallelStateProcessor struct { config *params.ChainConfig // Chain configuration options chain *HeaderChain // Canonical header chain @@ -42,6 +50,12 @@ func (p *ParallelStateProcessor) Process(block *types.Block, statedb *state.Stat signer = types.MakeSigner(p.config, header.Number, header.Time) ) + if preStateType == BALPreState { + start := time.Now() + statedb.PrefetchStateBAL(block.NumberU64()) + PrefetchBALTime += time.Since(start) + } + // Apply pre-execution system calls. var tracingStateDB = vm.StateDB(statedb) if hooks := cfg.Tracer; hooks != nil { @@ -57,29 +71,36 @@ func (p *ParallelStateProcessor) Process(block *types.Block, statedb *state.Stat ProcessParentBlockHash(block.ParentHash(), evm) } - return p.executeParallel(block, statedb, &context, cfg, gp, signer) + return p.executeParallel(block, statedb, cfg, gp, signer, context) } -func (p *ParallelStateProcessor) executeParallel(block *types.Block, statedb *state.StateDB, blockContext *vm.BlockContext, cfg vm.Config, gp *GasPool, signer types.Signer) (*ProcessResult, error) { +func (p *ParallelStateProcessor) executeParallel(block *types.Block, statedb *state.StateDB, cfg vm.Config, gp *GasPool, signer types.Signer, context vm.BlockContext) (*ProcessResult, error) { var ( - receipts = make(types.Receipts, len(block.Transactions())) - header = block.Header() - blockHash = block.Hash() - blockNumber = block.Number() - allLogs []*types.Log - preStatedb = statedb.Copy() + receipts = make(types.Receipts, len(block.Transactions())) + header = block.Header() + blockHash = block.Hash() + blockNumber = block.Number() + allLogs []*types.Log + preStateProvider PreStateProvider workers errgroup.Group ) - workers.SetLimit(runtime.NumCPU() / 2) + workers.SetLimit(runtime.NumCPU() - 6) // Fetch prestate for each tx // todo: handle gp with RW lock switch preStateType { case BALPreState: - panic("unimplemented") + { + start := time.Now() + statedb.MergePostBal() + PrefetchMergeBALTime += time.Since(start) + preStateProvider = statedb + } + case SeqPreState: { + preStatedb := statedb.Copy() gpcp := *gp preStateProvider = &SequentialPrestateProvider{ statedb: preStatedb, @@ -87,12 +108,13 @@ func (p *ParallelStateProcessor) executeParallel(block *types.Block, statedb *st gp: &gpcp, signer: signer, usedGas: new(uint64), - evm: vm.NewEVM(*blockContext, preStatedb, p.config, cfg), + evm: vm.NewEVM(context, preStatedb, p.config, cfg), } } } // Parallel executing the transaction + exeStart := time.Now() postEntries := make([][]state.JournalEntry, len(block.Transactions())) for i, tx := range block.Transactions() { cleanStatedb, err := preStateProvider.PrestateAtIndex(i) @@ -101,6 +123,7 @@ func (p *ParallelStateProcessor) executeParallel(block *types.Block, statedb *st } i := i + gpcp := *gp workers.Go(func() error { usedGas := new(uint64) msg, err := TransactionToMessage(tx, signer, header.BaseFee) @@ -109,9 +132,9 @@ func (p *ParallelStateProcessor) executeParallel(block *types.Block, statedb *st } cleanStatedb.SetTxContext(tx.Hash(), i) - evm := vm.NewEVM(*blockContext, cleanStatedb, p.config, cfg) + evm := vm.NewEVM(context, cleanStatedb, p.config, cfg) - receipt, entries, err := ApplyTransactionWithParallelEVM(msg, gp, cleanStatedb, blockNumber, blockHash, tx, usedGas, evm) + receipt, entries, err := ApplyTransactionWithParallelEVM(msg, &gpcp, cleanStatedb, blockNumber, blockHash, tx, usedGas, evm) if err != nil { return err } @@ -126,12 +149,15 @@ func (p *ParallelStateProcessor) executeParallel(block *types.Block, statedb *st if err != nil { return nil, err } + ParallelExeTime += time.Since(exeStart) // Merge state changes // - Append receipts // - Sum usedGas // - Collect state state changes: simple overwrite // - Ommit preimages for now usedGas := uint64(0) + + start := time.Now() for i, receipt := range receipts { if receipt == nil { continue // Skip nil receipts @@ -141,9 +167,10 @@ func (p *ParallelStateProcessor) executeParallel(block *types.Block, statedb *st allLogs = append(allLogs, receipt.Logs...) statedb.MergeState(postEntries[i]) } + PostMergeTime += time.Since(start) // Read requests if Prague is enabled. - evm := vm.NewEVM(*blockContext, statedb, p.config, cfg) + evm := vm.NewEVM(context, statedb, p.config, cfg) var requests [][]byte if p.config.IsPrague(block.Number(), block.Time()) { requests = [][]byte{} diff --git a/core/state/journal.go b/core/state/journal.go index 87b376715f..54434bc7ba 100644 --- a/core/state/journal.go +++ b/core/state/journal.go @@ -87,11 +87,11 @@ func PrintJournal(entries []JournalEntry) { case selfDestructChange: fmt.Println("selfDestructChange") case balanceChange: - fmt.Println("balanceChange", v) + fmt.Println("balanceChange", v.account, v.prev) case nonceChange: - fmt.Println("nonceChange", v) + fmt.Println("nonceChange", v.account, v.prev) case storageChange: - fmt.Println("storageChange", v) + fmt.Println("storageChange", v.account, v.key, v.prevvalue) case codeChange: fmt.Println("codeChange") case refundChange: diff --git a/core/state/state_object.go b/core/state/state_object.go index 96263bcaa3..02e92bf601 100644 --- a/core/state/state_object.go +++ b/core/state/state_object.go @@ -503,6 +503,27 @@ func (s *stateObject) deepCopy(db *StateDB) *stateObject { return obj } +func (s *stateObject) simpleCopy(db *StateDB) *stateObject { + obj := &stateObject{ + db: db, + address: s.address, + addrHash: s.addrHash, + origin: s.origin, + data: s.data, + code: s.code, + originStorage: s.originStorage.Copy(), + pendingStorage: make(Storage), + dirtyStorage: make(Storage), + uncommittedStorage: make(Storage), + dirtyCode: s.dirtyCode, + selfDestructed: s.selfDestructed, + newContract: s.newContract, + } + + obj.trie = s.trie + return obj +} + // // Attribute accessors // diff --git a/core/state/statedb.go b/core/state/statedb.go index 82edf4a5bb..015b3e0fe2 100644 --- a/core/state/statedb.go +++ b/core/state/statedb.go @@ -162,6 +162,8 @@ type StateDB struct { // The block number context for BALs blockNumber uint64 + // postState after appling tx + postStates map[int]*StateDB } type BALType int @@ -223,7 +225,7 @@ func init() { case BalPreblockKeysPostValues: { println("bal preblock keys post values") - fileName = "access_lists_kpostv.json" + fileName = "access_lists_kpostv.100.json" data, err := os.ReadFile(fileName) if err != nil { log.Error("Failed to load access lists", "err", err) @@ -272,6 +274,7 @@ func NewWithReader(root common.Hash, db Database, reader Reader) (*StateDB, erro updateJournal: newJournal(), accessList: newAccessList(), transientStorage: newTransientStorage(), + postStates: make(map[int]*StateDB), } if db.TrieDB().IsVerkle() { sdb.accessEvents = NewAccessEvents(db.PointCache()) @@ -283,11 +286,13 @@ func (s *StateDB) PrefetchStateBAL(blockNumber uint64) { s.blockNumber = blockNumber switch balType { case BalPreblockKeysPostValues: - s.PrefetchBalPreblockKeys() + { + s.prefetchBalPreblockKeys() + } } } -func (s *StateDB) PrefetchBalPreblockKeys() { +func (s *StateDB) prefetchBalPreblockKeys() { log.Info("PrefetchBalPreblockKeys...") type StorageKV struct { addr *common.Address @@ -356,15 +361,121 @@ func (s *StateDB) PrefetchBalPreblockKeys() { } } -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") - } +var ( + StateFinalizeTime = time.Duration(0) + StateCopyTime = time.Duration(0) + StateNewTime = time.Duration(0) + StateDeepCpTime = time.Duration(0) + StateSetTime = time.Duration(0) + StateLoadTime = time.Duration(0) +) - // Merge with BALs post state - return nil +func (s *StateDB) MergePostBal() { + if balType != BalPreblockKeysPostValues { + panic("MergePostBal is only supported with BalPreblockKeysPostValues") + } + start := time.Now() + var ( + postBal = AllBlockTxPostValues[s.blockNumber] + ) + StateLoadTime += time.Since(start) + + postState := s.Copy() + postState.prefetcher = nil + for txIndex := range len(postBal) { + postVals := postBal[txIndex] + start := time.Now() + for addr, acct := range postVals { + account := postState.getStateObject(addr) + if account == nil { + return + } + + if acct.Destruct { + account.markSelfdestructed() + continue + } + + account.setNonce(acct.Nonce) + if acct.Balance != nil { + account.setBalance(acct.Balance) + } + if acct.Code != nil { + account.setCode(crypto.Keccak256Hash(acct.Code), acct.Code) + } + maps.Copy(account.originStorage, acct.StorageKV) + postState.setStateObject(account) + } + StateSetTime += time.Since(start) + + start = time.Now() + s.postStates[txIndex] = postState.Copy() + StateCopyTime += time.Since(start) + } +} + +func (s *StateDB) MergePostBalBuggy() { + if balType != BalPreblockKeysPostValues { + panic("MergePostBal is only supported with BalPreblockKeysPostValues") + } + start := time.Now() + var ( + postBal = AllBlockTxPostValues[s.blockNumber] + ) + StateLoadTime += time.Since(start) + + postState := s.Copy() + postState.prefetcher = nil + for txIndex := range len(postBal) { + postVals := postBal[txIndex] + start := time.Now() + for addr, acct := range postVals { + account := postState.getStateObject(addr) + if account == nil { + return + } + + if acct.Destruct { + account.markSelfdestructed() + continue + } + + account.SetNonce(acct.Nonce) + if acct.Balance != nil { + account.SetBalance(acct.Balance) + } + if acct.Code != nil { + account.SetCode(crypto.Keccak256Hash(acct.Code), acct.Code) + } + // Will cause failure if postState.Finalise is not called. + for k, v := range acct.StorageKV { + account.SetState(k, v) + } + } + StateSetTime += time.Since(start) + + start = time.Now() + // postState.Finalise(true) + StateFinalizeTime += time.Since(start) + + start = time.Now() + s.postStates[txIndex] = postState.Copy() + StateCopyTime += time.Since(start) + } +} + +func (s *StateDB) PrestateAtIndex(txIndex int) (*StateDB, error) { + if balType != BalPreblockKeysPostValues { + return nil, fmt.Errorf("PreStateAtTxIndex is only supported with BalPreblockKeysPostValues") + } + if txIndex == 0 { + return s.Copy(), nil + } + state, ok := s.postStates[txIndex-1] + if !ok { + return nil, fmt.Errorf("PreState at txIndex: %d doesn't exists, PrefetchStateBAL must be called first", txIndex) + } + return state, nil } // StartPrefetcher initializes a new trie prefetcher to pull in nodes from the @@ -943,6 +1054,57 @@ func (s *StateDB) Copy() *StateDB { return state } +func (s *StateDB) CopyState() *StateDB { + // Copy all the basic fields, initialize the memory ones + start := time.Now() + state := &StateDB{ + db: s.db, + trie: s.trie, + reader: s.reader, + originalRoot: s.originalRoot, + stateObjects: make(map[common.Address]*stateObject, len(s.stateObjects)), + stateObjectsDestruct: make(map[common.Address]*stateObject, len(s.stateObjectsDestruct)), + mutations: make(map[common.Address]*mutation), + logs: make(map[common.Hash][]*types.Log), + preimages: make(map[common.Hash][]byte), + journal: newJournal(), + updateJournal: newJournal(), + accessList: newAccessList(), + transientStorage: newTransientStorage(), + } + StateNewTime += time.Since(start) + // if s.witness != nil { + // state.witness = s.witness.Copy() + // } + // if s.accessEvents != nil { + // state.accessEvents = s.accessEvents.Copy() + // } + // Deep copy cached state objects. + start = time.Now() + for addr, obj := range s.stateObjects { + state.stateObjects[addr] = obj.simpleCopy(state) + } + // Deep copy destructed state objects. + for addr, obj := range s.stateObjectsDestruct { + state.stateObjectsDestruct[addr] = obj.simpleCopy(state) + } + StateDeepCpTime += time.Since(start) + // Deep copy the object state markers. + // for addr, op := range s.mutations { + // state.mutations[addr] = op.copy() + // } + // Deep copy the logs occurred in the scope of block + // for hash, logs := range s.logs { + // cpy := make([]*types.Log, len(logs)) + // for i, l := range logs { + // cpy[i] = new(types.Log) + // *cpy[i] = *l + // } + // state.logs[hash] = cpy + // } + return state +} + // Snapshot returns an identifier for the current revision of the state. func (s *StateDB) Snapshot() int { s.updateJournal.snapshot()