Add simple parallel executor with sim prestate and pass TestEIP155Transition test

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 {

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
+}

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})
 }

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)
 }
 

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()