diff --git a/core/blockchain.go b/core/blockchain.go index 5b8be1b9be..8177957dc4 100644 --- a/core/blockchain.go +++ b/core/blockchain.go @@ -342,7 +342,8 @@ func NewBlockChain(db ethdb.Database, cacheConfig *CacheConfig, genesis *Genesis bc.statedb = state.NewDatabase(bc.triedb, nil) bc.validator = NewBlockValidator(chainConfig, bc) bc.prefetcher = newStatePrefetcher(chainConfig, bc.hc) - bc.processor = NewStateProcessor(chainConfig, bc.hc) + // bc.processor = NewStateProcessor(chainConfig, bc.hc) + bc.processor = NewParallelStateProcessor(chainConfig, bc.hc) genesisHeader := bc.GetHeaderByNumber(0) if genesisHeader == nil { @@ -1899,6 +1900,7 @@ func (bc *BlockChain) processBlock(parentRoot common.Hash, block *types.Block, s ) defer interrupt.Store(true) // terminate the prefetch at the end + bc.cacheConfig.TrieCleanNoPrefetch = true if bc.cacheConfig.TrieCleanNoPrefetch { statedb, err = state.New(parentRoot, bc.statedb) if err != nil { diff --git a/core/parallel_state_processor.go b/core/parallel_state_processor.go new file mode 100644 index 0000000000..c28531c709 --- /dev/null +++ b/core/parallel_state_processor.go @@ -0,0 +1,198 @@ +package core + +import ( + "fmt" + "math/big" + "sync" + + "github.com/ethereum/go-ethereum/common" + "github.com/ethereum/go-ethereum/consensus/misc" + "github.com/ethereum/go-ethereum/core/state" + "github.com/ethereum/go-ethereum/core/types" + "github.com/ethereum/go-ethereum/core/vm" + "github.com/ethereum/go-ethereum/params" +) + +type ParallelStateProcessor struct { + config *params.ChainConfig // Chain configuration options + chain *HeaderChain // Canonical header chain +} + +func NewParallelStateProcessor(config *params.ChainConfig, chain *HeaderChain) *ParallelStateProcessor { + return &ParallelStateProcessor{ + config: config, + chain: chain, + } +} + +func (p *ParallelStateProcessor) Process(block *types.Block, statedb *state.StateDB, cfg vm.Config) (*ProcessResult, error) { + fmt.Println("ParallelStateProcessor.Process called") + var ( + header = block.Header() + gp = new(GasPool).AddGas(block.GasLimit()) + ) + + // 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) + ) + + // Apply pre-execution system calls. + var tracingStateDB = vm.StateDB(statedb) + if hooks := cfg.Tracer; hooks != nil { + tracingStateDB = state.NewHookedState(statedb, hooks) + } + context = NewEVMBlockContext(header, p.chain, nil) + evm := vm.NewEVM(context, tracingStateDB, p.config, cfg) + + if beaconRoot := block.BeaconRoot(); beaconRoot != nil { + ProcessBeaconBlockRoot(*beaconRoot, evm) + } + if p.config.IsPrague(block.Number(), block.Time()) || p.config.IsVerkle(block.Number(), block.Time()) { + ProcessParentBlockHash(block.ParentHash(), evm) + } + + return p.executeParallel(block, statedb, &context, cfg, gp, signer) +} + +func (p *ParallelStateProcessor) executeParallel(block *types.Block, statedb *state.StateDB, blockContext *vm.BlockContext, cfg vm.Config, gp *GasPool, signer types.Signer) (*ProcessResult, error) { + var ( + receipts = make(types.Receipts, len(block.Transactions())) + header = block.Header() + blockHash = block.Hash() + blockNumber = block.Number() + allLogs []*types.Log + wg sync.WaitGroup + preStatedb = statedb.Copy() + sequentialEvm = vm.NewEVM(*blockContext, preStatedb, p.config, cfg) + seqUsedGas = new(uint64) + ) + // Fetch prestate for each tx + + // Parallel executing the transaction + postStates := make([]*state.StateDB, len(block.Transactions())) + postEntries := make([][]state.JournalEntry, len(block.Transactions())) + for i, tx := range block.Transactions() { + postStates[i] = preStatedb.Copy() // Copy the state for each transaction + cleanStatedb := postStates[i] + i := i + + wg.Add(1) + go func() { + defer wg.Done() + + usedGas := new(uint64) + msg, err := TransactionToMessage(tx, signer, header.BaseFee) + // todo: handle error: break all routines and return error + if err != nil { + fmt.Printf("could not apply tx %d [%v]: %v", i, tx.Hash().Hex(), err) + } + cleanStatedb.SetTxContext(tx.Hash(), i) + + evm := vm.NewEVM(*blockContext, cleanStatedb, p.config, cfg) + + receipt, entries, err := ApplyTransactionWithParallelEVM(msg, gp, cleanStatedb, blockNumber, blockHash, tx, usedGas, evm) + if err != nil { + fmt.Printf("could not apply parallel tx %d [%v]: %v", i, tx.Hash().Hex(), err) + } + receipts[i] = receipt + postEntries[i] = entries + }() + + // 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 to msg %d [%v]: %w", i, tx.Hash().Hex(), err) + } + preStatedb.SetTxContext(tx.Hash(), i) + + _, err = ApplyTransactionWithEVM(msg, gp, preStatedb, blockNumber, blockHash, tx, seqUsedGas, sequentialEvm) + if err != nil { + return nil, fmt.Errorf("could not apply tx %d [%v]: %w", i, tx.Hash().Hex(), err) + } + } + + wg.Wait() + // Merge state changes + // - Append receipts + // - Sum usedGas + // - Collect state state changes: simple overwrite + // - Ommit preimages for now + usedGas := uint64(0) + for i, receipt := range receipts { + if receipt == nil { + continue // Skip nil receipts + } + receipt.CumulativeGasUsed = usedGas + receipt.GasUsed + usedGas += receipt.GasUsed + allLogs = append(allLogs, receipt.Logs...) + statedb.MergeState(postEntries[i]) + } + + // Read requests if Prague is enabled. + evm := vm.NewEVM(*blockContext, statedb, p.config, cfg) + var requests [][]byte + if p.config.IsPrague(block.Number(), block.Time()) { + requests = [][]byte{} + // EIP-6110 + 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 + } + } + + // Finalize the block, applying any consensus engine specific extras (e.g. block rewards) + p.chain.engine.Finalize(p.chain, header, statedb, block.Body()) + + return &ProcessResult{ + Receipts: receipts, + Requests: requests, + Logs: allLogs, + GasUsed: usedGas, + }, nil +} + +func ApplyTransactionWithParallelEVM(msg *Message, gp *GasPool, statedb *state.StateDB, blockNumber *big.Int, blockHash common.Hash, tx *types.Transaction, usedGas *uint64, evm *vm.EVM) (receipt *types.Receipt, entries []state.JournalEntry, err error) { + if hooks := evm.Config.Tracer; hooks != nil { + if hooks.OnTxStart != nil { + hooks.OnTxStart(evm.GetVMContext(), tx, msg.From) + } + if hooks.OnTxEnd != nil { + defer func() { hooks.OnTxEnd(receipt, err) }() + } + } + // Apply the transaction to the current state (included in the env). + result, err := ApplyMessage(evm, msg, gp) + if err != nil { + return nil, nil, err + } + // copy changed state + entries = statedb.JournalEntriesCopy() + // Update the state with pending changes. + var root []byte + if evm.ChainConfig().IsByzantium(blockNumber) { + evm.StateDB.Finalise(true) + } else { + root = statedb.IntermediateRoot(evm.ChainConfig().IsEIP158(blockNumber)).Bytes() + } + *usedGas += result.UsedGas + + // Merge the tx-local access event into the "block-local" one, in order to collect + // all values, so that the witness can be built. + if statedb.GetTrie().IsVerkle() { + statedb.AccessEvents().Merge(evm.AccessEvents) + } + + return MakeReceipt(evm, result, statedb, blockNumber, blockHash, tx, *usedGas, root), entries, nil +} diff --git a/core/state/journal.go b/core/state/journal.go index 13332dbd79..d69e0d731a 100644 --- a/core/state/journal.go +++ b/core/state/journal.go @@ -45,6 +45,17 @@ type journalEntry interface { copy() journalEntry } +type JournalEntry interface { + // revert undoes the changes introduced by this journal entry. + revert(*StateDB) + + // dirtied returns the Ethereum address modified by this journal entry. + dirtied() *common.Address + + // copy returns a deep-copied journal entry. + copy() journalEntry +} + // journal contains the list of state modifications applied since the last state // commit. These are tracked to be able to be reverted in the case of an execution // exception or request for reversal. @@ -148,6 +159,14 @@ func (j *journal) copy() *journal { } } +func (j *journal) copyEntries() []JournalEntry { + entries := make([]JournalEntry, j.length()) + for i := range j.length() { + entries[i] = j.entries[i].copy() + } + return entries +} + func (j *journal) logChange(txHash common.Hash) { j.append(addLogChange{txhash: txHash}) } diff --git a/core/state/state_object.go b/core/state/state_object.go index a6979bd361..e6650418d1 100644 --- a/core/state/state_object.go +++ b/core/state/state_object.go @@ -218,6 +218,7 @@ func (s *stateObject) SetState(key, value common.Hash) common.Hash { } // New value is different, update and journal the change s.db.journal.storageChange(s.address, key, prev, origin) + s.db.updateJournal.storageChange(s.address, key, value, prev) s.setState(key, value, origin) return prev } @@ -470,6 +471,7 @@ func (s *stateObject) AddBalance(amount *uint256.Int) uint256.Int { func (s *stateObject) SetBalance(amount *uint256.Int) uint256.Int { prev := *s.data.Balance s.db.journal.balanceChange(s.address, s.data.Balance) + s.db.updateJournal.balanceChange(s.address, amount) s.setBalance(amount) return prev } @@ -551,6 +553,7 @@ func (s *stateObject) CodeSize() int { func (s *stateObject) SetCode(codeHash common.Hash, code []byte) (prev []byte) { prev = slices.Clone(s.code) s.db.journal.setCode(s.address, prev) + s.db.updateJournal.setCode(s.address, code) s.setCode(codeHash, code) return prev } @@ -563,6 +566,7 @@ func (s *stateObject) setCode(codeHash common.Hash, code []byte) { func (s *stateObject) SetNonce(nonce uint64) { s.db.journal.nonceChange(s.address, s.data.Nonce) + s.db.updateJournal.nonceChange(s.address, nonce) s.setNonce(nonce) } diff --git a/core/state/statedb.go b/core/state/statedb.go index 2453d67f3e..75f2b88c48 100644 --- a/core/state/statedb.go +++ b/core/state/statedb.go @@ -134,6 +134,8 @@ type StateDB struct { // Journal of state modifications. This is the backbone of // Snapshot and RevertToSnapshot. journal *journal + // + updateJournal *journal // State witness if cross validation is needed witness *stateless.Witness @@ -184,6 +186,7 @@ func NewWithReader(root common.Hash, db Database, reader Reader) (*StateDB, erro logs: make(map[common.Hash][]*types.Log), preimages: make(map[common.Hash][]byte), journal: newJournal(), + updateJournal: newJournal(), accessList: newAccessList(), transientStorage: newTransientStorage(), } @@ -287,6 +290,42 @@ func (s *StateDB) AddRefund(gas uint64) { s.refund += gas } +func (s *StateDB) JournalEntriesCopy() []JournalEntry { + return s.updateJournal.copyEntries() +} + +func (s *StateDB) MergeState(entries []JournalEntry) { + for _, entry := range entries { + if entry == nil { + continue + } + switch v := entry.(type) { + case createObjectChange: + s.CreateAccount(v.account) + case createContractChange: + s.CreateContract(v.account) + case selfDestructChange: + s.SelfDestruct(v.account) + case balanceChange: + { + s.SetBalance(v.account, v.prev, tracing.BalanceChangeUnspecified) + + } + case nonceChange: + s.SetNonce(v.account, v.prev, tracing.NonceChangeUnspecified) + case storageChange: + { + obj := s.getOrNewStateObject(v.account) + obj.SetState(v.key, v.prevvalue) + } + case codeChange: + s.SetCode(v.account, v.prevCode) + default: + + } + } +} + // SubRefund removes gas from the refund counter. // This method will panic if the refund counter goes below zero func (s *StateDB) SubRefund(gas uint64) { @@ -516,6 +555,7 @@ func (s *StateDB) SelfDestruct(addr common.Address) uint256.Int { // for journalling a second time. if !stateObject.selfDestructed { s.journal.destruct(addr) + s.updateJournal.destruct(addr) stateObject.markSelfdestructed() } return prevBalance @@ -633,6 +673,7 @@ func (s *StateDB) getOrNewStateObject(addr common.Address) *stateObject { func (s *StateDB) createObject(addr common.Address) *stateObject { obj := newObject(s, addr, nil) s.journal.createObject(addr) + s.updateJournal.createObject(addr) s.setStateObject(obj) return obj } @@ -655,6 +696,7 @@ func (s *StateDB) CreateContract(addr common.Address) { if !obj.newContract { obj.newContract = true s.journal.createContract(addr) + s.updateJournal.createContract(addr) } } @@ -687,6 +729,8 @@ func (s *StateDB) Copy() *StateDB { accessList: s.accessList.Copy(), transientStorage: s.transientStorage.Copy(), journal: s.journal.copy(), + // The update journal is not copies to avoid duplicated updates in later transactions. + updateJournal: newJournal(), } if s.witness != nil { state.witness = s.witness.Copy()