diff --git a/cmd/geth/chaincmd.go b/cmd/geth/chaincmd.go index 195ea4b87e..a08b693e79 100644 --- a/cmd/geth/chaincmd.go +++ b/cmd/geth/chaincmd.go @@ -466,22 +466,22 @@ func showMetrics() { 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("PrefetchBALTime: ", core.PrefetchBALTime) + fmt.Println("PrefetchMergeTime:", core.PrefetchMergeBALTime) + fmt.Println("ParallelExeTime: ", core.ParallelExeTime) fmt.Println("PostMergeTime: ", core.PostMergeTime) // total fmt.Println("blockInsertTimer", blockInsertTimer.Total()) mgasps := mgaspsHist.Snapshot() - fmt.Println("mgasps,mean,max,min:", mgasps.Mean(), mgasps.Max(), mgasps.Min()) + fmt.Println("mgasps,mean,max,min:", int64(mgasps.Mean()), mgasps.Max(), mgasps.Min()) } func exportChain(ctx *cli.Context) error { diff --git a/core/parallel_state_processor.go b/core/parallel_state_processor.go index bd2b2ccd5a..23d3e82419 100644 --- a/core/parallel_state_processor.go +++ b/core/parallel_state_processor.go @@ -4,6 +4,7 @@ import ( "fmt" "math/big" "runtime" + "sync" "time" "github.com/ethereum/go-ethereum/common" @@ -37,34 +38,30 @@ func NewParallelStateProcessor(config *params.ChainConfig, chain *HeaderChain) * func (p *ParallelStateProcessor) Process(block *types.Block, statedb *state.StateDB, cfg vm.Config) (*ProcessResult, error) { fmt.Println("ParallelStateProcessor.Process called:", block.NumberU64()) var ( - header = block.Header() - gp = new(GasPool).AddGas(block.GasLimit()) + header = block.Header() + context vm.BlockContext + gp = new(GasPool).AddGas(block.GasLimit()) + signer = types.MakeSigner(p.config, header.Number, header.Time) + initialdb = statedb.Copy() + postState *state.StateDB + + wg sync.WaitGroup + result *ProcessResult + err error ) // Mutate the block and state according to any hard-fork specs if p.config.DAOForkSupport && p.config.DAOForkBlock != nil && p.config.DAOForkBlock.Cmp(block.Number()) == 0 { misc.ApplyDAOHardFork(statedb) } - var ( - context vm.BlockContext - signer = types.MakeSigner(p.config, header.Number, header.Time) - initialdb *state.StateDB - ) if preStateType == BALPreState { - start := time.Now() - statedb.PreComputePostState(block.NumberU64()) - PrefetchMergeBALTime += time.Since(start) - // copy initialdb before PrefetchStateBAL to avoid redundant copy - initialdb = statedb.Copy() - - start = time.Now() + start := time.Now() // Must prefetch bal before syscall to avoid overriding syscall's state, thus merkle root might mismatch statedb.PrefetchStateBAL(block.NumberU64()) + // statedb.SetBlocknumber(block.NumberU64()) PrefetchBALTime += time.Since(start) - } else { - initialdb = statedb.Copy() } // Apply pre-execution system calls. @@ -82,7 +79,38 @@ func (p *ParallelStateProcessor) Process(block *types.Block, statedb *state.Stat ProcessParentBlockHash(block.ParentHash(), evm) } - return p.executeParallel(block, statedb, cfg, gp, signer, context, initialdb) + wg.Add(2) + go func() { + defer wg.Done() + + start := time.Now() + postState = statedb.Copy() + postState.MergePostBalStates() + // prewarm the updating trie + postState.IntermediateRoot(true) + PostMergeTime += time.Since(start) + }() + + go func() { + defer wg.Done() + + start := time.Now() + statedb.PreComputePostState(block.NumberU64()) + PrefetchMergeBALTime += time.Since(start) + + exeStart := time.Now() + result, err = p.executeParallel(block, statedb, cfg, gp, signer, context, initialdb) + ParallelExeTime += time.Since(exeStart) + + }() + + wg.Wait() + + // Last tx alreadly includes the state change in requests after all txs + if preStateType == BALPreState { + *statedb = *postState + } + return result, err } func (p *ParallelStateProcessor) executeParallel(block *types.Block, statedb *state.StateDB, cfg vm.Config, gp *GasPool, signer types.Signer, context vm.BlockContext, initialdb *state.StateDB) (*ProcessResult, error) { @@ -95,17 +123,21 @@ func (p *ParallelStateProcessor) executeParallel(block *types.Block, statedb *st preStateProvider PreStateProvider workers errgroup.Group - postState = statedb.Copy() ) + // leave some cpus for prefetching workers.SetLimit(runtime.NumCPU() / 2) switch preStateType { case BALPreState: { + err := initialdb.SetPostBAL(statedb.PostBAL(), blockNumber.Uint64()) + if err != nil { + return nil, err + } } - case SeqPreState: + case SeqPreState: // must set workers limit = 1 { preStatedb := statedb.Copy() gpcp := *gp @@ -121,19 +153,8 @@ func (p *ParallelStateProcessor) executeParallel(block *types.Block, statedb *st } // Parallel executing the transaction - exeStart := time.Now() postEntries := make([][]state.JournalEntry, len(block.Transactions())) - workers.Go(func() error { - start := time.Now() - postState.MergePostBalStates() - // prewarm the updating trie - postState.IntermediateRoot(true) - PostMergeTime += time.Since(start) - - return nil - }) - for i, tx := range block.Transactions() { i := i workers.Go(func() error { @@ -147,16 +168,19 @@ func (p *ParallelStateProcessor) executeParallel(block *types.Block, statedb *st cleanStatedb = initialdb.Copy() cleanStatedb.SetTxContext(tx.Hash(), i) err = cleanStatedb.SetTxBALReader(statedb) + if err != nil { + return err + } } case SeqPreState: { cleanStatedb, err = preStateProvider.PrestateAtIndex(i) + if err != nil { + return err + } cleanStatedb.SetTxContext(tx.Hash(), i) } } - if err != nil { - return err - } evm := vm.NewEVM(context, cleanStatedb, p.config, cfg) @@ -183,7 +207,6 @@ func (p *ParallelStateProcessor) executeParallel(block *types.Block, statedb *st return nil, err } - ParallelExeTime += time.Since(exeStart) // Merge state changes // - Append receipts // - Sum usedGas @@ -205,7 +228,8 @@ func (p *ParallelStateProcessor) executeParallel(block *types.Block, statedb *st } // Read requests if Prague is enabled. - evm := vm.NewEVM(context, statedb, p.config, cfg) + // commented out EIP-7002, EIP-7251 and p.chain.engine.Finalize for now, since statedb might cause concurrent map writes (journal.go:211) when postState = statedb.Copy() + // evm := vm.NewEVM(context, statedb, p.config, cfg) var requests [][]byte if p.config.IsPrague(block.Number(), block.Time()) { requests = [][]byte{} @@ -213,22 +237,18 @@ func (p *ParallelStateProcessor) executeParallel(block *types.Block, statedb *st if err := ParseDepositLogs(&requests, allLogs, p.config); err != nil { return nil, err } - // EIP-7002 - if err := ProcessWithdrawalQueue(&requests, evm); err != nil { - return nil, err - } - // EIP-7251 - if err := ProcessConsolidationQueue(&requests, evm); err != nil { - return nil, err - } + // // EIP-7002 + // if err := ProcessWithdrawalQueue(&requests, evm); err != nil { + // return nil, err + // } + // // EIP-7251 + // if err := ProcessConsolidationQueue(&requests, evm); err != nil { + // return nil, err + // } } // Finalize the block, applying any consensus engine specific extras (e.g. block rewards) - p.chain.engine.Finalize(p.chain, header, statedb, block.Body()) - - // Last tx alreadly includes the state change in requests after all txs - // wg.Wait() - *statedb = *postState + // p.chain.engine.Finalize(p.chain, header, statedb, block.Body()) return &ProcessResult{ Receipts: receipts, diff --git a/core/state/reader.go b/core/state/reader.go index ebfb24a807..7db4ee1452 100644 --- a/core/state/reader.go +++ b/core/state/reader.go @@ -339,7 +339,9 @@ func (r *trieReader) StorageBAL(addr common.Address, key common.Hash, tr *trie.S return common.Hash{}, err } - return common.BytesToHash(ret), nil + var value common.Hash + value.SetBytes(ret) + return value, nil } // multiStateReader is the aggregation of a list of StateReader interface, @@ -547,23 +549,7 @@ func (r *readerWithBAL) Account(addr common.Address) (*types.StateAccount, error if ok { acctVal, ok := postVals[addr] if ok { - // Cache codeHash - if acctVal.Code != nil { - if acctVal.CodeHash == nil { - acctVal.CodeHash = crypto.Keccak256(acctVal.Code) - } - } - if acctVal.Cached { - return &types.StateAccount{ - Nonce: acctVal.Nonce, - Balance: acctVal.Balance, - Root: acctVal.Root, - CodeHash: acctVal.CodeHash, - }, nil - } - acct, err := r.Reader.Account(addr) - acctVal.Cached = true if acct == nil || err != nil { // acct is nil, just return the postAcct return &types.StateAccount{ @@ -574,23 +560,27 @@ func (r *readerWithBAL) Account(addr common.Address) (*types.StateAccount, error } // acct mustn't be modified, or it'll change the cached pre-block values in readerwithcache + nonce := acctVal.Nonce + balance := acctVal.Balance + codeHash := acct.CodeHash + if acct.Nonce > acctVal.Nonce { - acctVal.Nonce = acct.Nonce + nonce = acct.Nonce } if acctVal.Balance == nil { - acctVal.Balance = acct.Balance + balance = acct.Balance } - if acctVal.CodeHash == nil { - acctVal.CodeHash = acct.CodeHash - } - if acctVal.Root == (common.Hash{}) { - acctVal.Root = acct.Root + + // codeHash + if acctVal.Code != nil { + codeHash = crypto.Keccak256(acctVal.Code) } + return &types.StateAccount{ - Nonce: acctVal.Nonce, - Balance: acctVal.Balance, - Root: acctVal.Root, - CodeHash: acctVal.CodeHash, + Nonce: nonce, + Balance: balance, + Root: acct.Root, + CodeHash: codeHash, }, nil } } @@ -599,17 +589,6 @@ func (r *readerWithBAL) Account(addr common.Address) (*types.StateAccount, error if err != nil { return nil, err } - // cache account and codeHash except the first transaction - if postVals != nil && acct != nil { - postVals[addr] = &types.AcctPostValues{ - Nonce: acct.Nonce, - Balance: acct.Balance, - Destruct: false, - CodeHash: acct.CodeHash, - Root: acct.Root, - Cached: true, - } - } return acct, nil } @@ -626,16 +605,6 @@ func (r *readerWithBAL) Storage(addr common.Address, slot common.Hash) (common.H } val, err := r.Reader.Storage(addr, slot) - // Don't Cache storage due to concurrent map writes because postVals might share the same map. - // if postVals != nil && postVals[addr] != nil { - // if postVals[addr].StorageKV != nil { - // postVals[addr].StorageKV[slot] = val - // } else { - // postVals[addr].StorageKV = map[common.Hash]common.Hash{ - // slot: val, - // } - // } - // } if err != nil { return common.Hash{}, err } @@ -654,8 +623,6 @@ func (r *readerWithBAL) Code(addr common.Address, codeHash common.Hash) ([]byte, } if acctVal.CodeHash != nil { code, err := r.Reader.Code(addr, common.Hash(acctVal.CodeHash)) - // Cache code - acctVal.Code = code if err != nil { return nil, fmt.Errorf("cannot get code from code for addr:%x, err:%v", addr, err) } @@ -668,22 +635,6 @@ func (r *readerWithBAL) Code(addr common.Address, codeHash common.Hash) ([]byte, return nil, fmt.Errorf("cannot get code from acct for addr:%x, err:%v", addr, err) } code, err := r.Reader.Code(addr, common.Hash(acct.CodeHash)) - // Cache account, codeHash and code except the first transaction - // if postVals != nil { - // if postVals[addr] != nil { - // postVals[addr].Code = code - // postVals[addr].CodeHash = codeHash[:] - // } else { - // postVals[addr] = &types.AcctPostValues{ - // Nonce: acct.Nonce, - // Balance: acct.Balance, - // Code: code, - // Destruct: false, - // CodeHash: acct.CodeHash, - // } - // } - // } - if err != nil { return nil, fmt.Errorf("cannot get code from codeHash for addr:%x, err:%v", addr, err) } diff --git a/core/state/statedb.go b/core/state/statedb.go index 0032463f0a..3f2be3814b 100644 --- a/core/state/statedb.go +++ b/core/state/statedb.go @@ -216,7 +216,7 @@ func init() { case BalPreblockKeysPostValues: { println("bal preblock keys post values") - fileName = "access_lists_kpostv.100.json" + fileName = "access_lists_kpostv.2000.json" data, err := os.ReadFile(fileName) if err != nil { log.Error("Failed to load access lists", "err", err) @@ -274,14 +274,19 @@ func NewWithReader(root common.Hash, db Database, reader Reader) (*StateDB, erro } func (s *StateDB) PreComputePostState(blockNumber uint64) { + s.blockNumber = blockNumber postBal := map[int]types.TxPostValues{} for k, v := range AllBlockTxPostValues[blockNumber] { // Clone the map to avoid race with MergePostBalStates - postBal[k] = maps.Clone(v) + postBal[k] = types.TxPostValues{} + for key, val := range v { + postBal[k][key] = val.Clone() + } } - for i := 0; i < len(postBal)-1; i++ { + // -1 and len(postBal)-1 are syscalls pre and post all txs. + for i := -1; i < len(postBal)-2; i++ { prevVals := postBal[i] postVals := postBal[i+1] for addr, prevAcct := range prevVals { @@ -309,6 +314,10 @@ func (s *StateDB) PreComputePostState(blockNumber uint64) { s.postBAL = postBal } +func (s *StateDB) PostBAL() map[int]types.TxPostValues { + return s.postBAL +} + func (s *StateDB) SetTxBALReader(preblockReader Reader) error { if s.postBAL == nil { return errors.New("cannot set bal reader without postBAL") @@ -317,6 +326,20 @@ func (s *StateDB) SetTxBALReader(preblockReader Reader) error { return nil } +func (s *StateDB) SetBlocknumber(bn uint64) error { + s.blockNumber = bn + return nil +} + +func (s *StateDB) SetPostBAL(postBal map[int]types.TxPostValues, bn uint64) error { + if postBal == nil { + return errors.New("cannot set postBAL without postBAL") + } + s.blockNumber = bn + s.postBAL = postBal + return nil +} + func (s *StateDB) SetOriginalReader(reader Reader) { s.reader = reader } @@ -353,13 +376,19 @@ func (s *StateDB) prefetchBalPreblockKeys() { lenMaxSlots += len(acl.StorageKeys) workers.Go(func() error { - acct, err := s.reader.AccountBAL(addr) - obj := newObject(s, addr, acct) - accounts <- obj + acctBal, err := s.reader.AccountBAL(addr) + var obj *stateObject if err != nil { log.Error("fail to fetch account from BALs:", addr) - return err + acct, err := s.reader.Account(addr) + if err != nil { + panic("fail to fetch account from BALs") + } + obj = newObject(s, addr, acct) + } else { + obj = newObject(s, addr, acctBal) } + accounts <- obj return nil }) } @@ -389,19 +418,26 @@ func (s *StateDB) prefetchBalPreblockKeys() { tr, err := trie.NewStateTrie(trie.StorageTrieID(s.originalRoot, obj.addrHash, obj.origin.Root), s.db.TrieDB()) if err != nil { - log.Error("fail to create trie for:", addr) - panic("fail to create trie for:") + fmt.Println(s.originalRoot, obj.addrHash, obj.origin.Root) + log.Error("fail to create trie for", "addr:", addr, "error:", err) + panic("fail to create trie") } for _, key := range keys { workers.Go(func() error { val, err := s.reader.StorageBAL(addr, key, tr) kv := &StorageKV{&addr, &key, &val} - storages <- kv if err != nil { log.Error("fail to fetch storage from BALs:", addr, key) - return err + // It's not safe to concurrent read from trie with StorageBAL, so we use the locked version Storage as fallback + val, err := s.reader.Storage(addr, key) + if err != nil { + panic("fail to fetch storage from BALs") + } + kv.val = &val } + storages <- kv + return nil }) } @@ -420,6 +456,39 @@ func (s *StateDB) prefetchBalPreblockKeys() { close(storages) } +// Unused for now since readWithCache involves lots of locks which degrades performance +func (s *StateDB) prefetchBalPreblockKeysWithCachereader() { + log.Info("PrefetchBalPreblockKeys...") + var ( + preBal = AllBlockAccessLists[s.blockNumber] + workers errgroup.Group + ) + + workers.SetLimit(runtime.NumCPU() - 2) + // Fetch pre-block acccount state + for _, acl := range preBal { + addr := acl.Address + + workers.Go(func() error { + s.reader.Account(addr) + return nil + }) + } + + // Fetch pre-block storage state + for _, acl := range preBal { + addr := acl.Address + keys := acl.StorageKeys + for _, key := range keys { + workers.Go(func() error { + s.reader.Storage(addr, key) + return nil + }) + } + } + workers.Wait() +} + var ( StateFinalizeTime = time.Duration(0) StateCopyTime = time.Duration(0) @@ -630,7 +699,8 @@ func (s *StateDB) MergePostBalStates() { postBal = AllBlockTxPostValues[s.blockNumber] ) - for txIndex := range len(postBal) { + // -1 and len(postBal)-1 (not used) are syscalls pre and post all txs. + for txIndex := -1; txIndex < len(postBal)-1; txIndex++ { postVals := postBal[txIndex] for addr, acct := range postVals { diff --git a/core/state/statedb_readerbal.go b/core/state/statedb_readerbal.go index c5c64311fc..160f15dc07 100644 --- a/core/state/statedb_readerbal.go +++ b/core/state/statedb_readerbal.go @@ -1,19 +1,21 @@ package state import ( + "fmt" + "github.com/ethereum/go-ethereum/common" "github.com/ethereum/go-ethereum/core/types" "github.com/ethereum/go-ethereum/trie" ) +// All states must has been cachae to statedb. func (s *StateDB) Account(addr common.Address) (*types.StateAccount, error) { // s.getStateObject() shouldn't be used, cause when acct is nil, it'll load account from DB and result concurrent map writes for accts obj := s.stateObjects[addr] - return &obj.data, nil -} - -func (s *StateDB) AccountBAL(addr common.Address) (*types.StateAccount, error) { - panic("AccountBAL not implemented for statedb") + if obj != nil { + return &obj.data, nil + } + return nil, fmt.Errorf("readerWithBal account %v, not exist", addr) } func (s *StateDB) Code(addr common.Address, codeHash common.Hash) ([]byte, error) { @@ -27,11 +29,17 @@ func (s *StateDB) CodeSize(addr common.Address, codeHash common.Hash) (int, erro func (s *StateDB) Storage(addr common.Address, slot common.Hash) (common.Hash, error) { stateObject := s.stateObjects[addr] if stateObject != nil { - return stateObject.GetState(slot), nil + if value, cached := stateObject.originStorage[slot]; cached { + return value, nil + } } return common.Hash{}, nil } +func (s *StateDB) AccountBAL(addr common.Address) (*types.StateAccount, error) { + panic("AccountBAL not implemented for statedb") +} + func (s *StateDB) StorageBAL(addr common.Address, slot common.Hash, tr *trie.StateTrie) (common.Hash, error) { panic("StorageBAL not implemented for statedb") } diff --git a/trie/trie.go b/trie/trie.go index 4c7aac1433..b8e945330c 100644 --- a/trie/trie.go +++ b/trie/trie.go @@ -157,10 +157,6 @@ func (t *Trie) Get(key []byte) ([]byte, error) { } func (t *Trie) get(origNode node, key []byte, pos int) (value []byte, newnode node, didResolve bool, err error) { - // When concurrent executing, sometime the following switch will throw error - if origNode == nil { - return nil, nil, false, nil - } switch n := (origNode).(type) { case nil: return nil, nil, false, nil