add parallel exe with simulated prestate

core/parallel_state_processor.go

@@ -3,7 +3,7 @@ package core
 import (
 	"fmt"
 	"math/big"
-	"sync"
+	"runtime"
 
 	"github.com/ethereum/go-ethereum/common"
 	"github.com/ethereum/go-ethereum/consensus/misc"
@@ -11,8 +11,18 @@ import (
 	"github.com/ethereum/go-ethereum/core/types"
 	"github.com/ethereum/go-ethereum/core/vm"
 	"github.com/ethereum/go-ethereum/params"
+	"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
@@ -66,30 +76,45 @@ func (p *ParallelStateProcessor) executeParallel(block *types.Block, statedb *st
 		blockHash        = block.Hash()
 		blockNumber      = block.Number()
 		allLogs          []*types.Log
-		wg            sync.WaitGroup
 		preStatedb       = statedb.Copy()
-		sequentialEvm = vm.NewEVM(*blockContext, preStatedb, p.config, cfg)
+		preStateProvider PreStateProvider
-		seqUsedGas    = new(uint64)
+		workers          errgroup.Group
 	)
+	workers.SetLimit(runtime.NumCPU() / 2)
 	// Fetch prestate for each tx
 
+	// todo: handle gp with RW lock
+	switch preStateType {
+	case BALPreState:
+		panic("unimplemented")
+	case SeqPreState:
+		{
+			gpcp := *gp
+			preStateProvider = &SequentialPrestateProvider{
+				statedb: preStatedb,
+				block:   block,
+				gp:      &gpcp,
+				signer:  signer,
+				usedGas: new(uint64),
+				evm:     vm.NewEVM(*blockContext, preStatedb, p.config, cfg),
+			}
+		}
+	}
+
 	// 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, err := preStateProvider.PrestateAtIndex(i)
-		cleanStatedb := postStates[i]
+		if err != nil {
+			return nil, err
+		}
+
 		i := i
-
+		workers.Go(func() error {
-		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)
+				return err
 			}
 			cleanStatedb.SetTxContext(tx.Hash(), i)
 
@@ -97,26 +122,22 @@ func (p *ParallelStateProcessor) executeParallel(block *types.Block, statedb *st
 
 			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)
+				return err
 			}
 			receipts[i] = receipt
 			postEntries[i] = entries
-		}()
 
-		// execute the transaction again to simulate the state changes
+			// if i == len(block.Transactions())-1 {
-		msg, err := TransactionToMessage(tx, signer, header.BaseFee)
+			// 	*statedb = *cleanStatedb
+			// }
+			return nil
+		})
+	}
+
+	err := workers.Wait()
 	if err != nil {
-			return nil, fmt.Errorf("could not apply tx to msg %d [%v]: %w", i, tx.Hash().Hex(), err)
+		return nil, 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
@@ -196,3 +217,45 @@ 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
+}

core/state/journal.go

@@ -74,6 +74,34 @@ func newJournal() *journal {
 	}
 }
 
+func PrintJournal(entries []JournalEntry) {
+	for _, entry := range entries {
+		if entry == nil {
+			continue
+		}
+		switch v := entry.(type) {
+		case createObjectChange:
+			fmt.Println("createObjectChange")
+		case createContractChange:
+			fmt.Println("createContractChange", v)
+		case selfDestructChange:
+			fmt.Println("selfDestructChange")
+		case balanceChange:
+			fmt.Println("balanceChange", v)
+		case nonceChange:
+			fmt.Println("nonceChange", v)
+		case storageChange:
+			fmt.Println("storageChange", v)
+		case codeChange:
+			fmt.Println("codeChange")
+		case refundChange:
+			fmt.Println("refundChange")
+		default:
+
+		}
+	}
+}
+
 // reset clears the journal, after this operation the journal can be used anew.
 // It is semantically similar to calling 'newJournal', but the underlying slices
 // can be reused.
@@ -108,6 +136,21 @@ func (j *journal) revertToSnapshot(revid int, s *StateDB) {
 	j.validRevisions = j.validRevisions[:idx]
 }
 
+func (j *journal) revertToSnapshotWithoutRevertState(revid int, s *StateDB) {
+	// Find the snapshot in the stack of valid snapshots.
+	idx := sort.Search(len(j.validRevisions), func(i int) bool {
+		return j.validRevisions[i].id >= revid
+	})
+	if idx == len(j.validRevisions) || j.validRevisions[idx].id != revid {
+		panic(fmt.Errorf("revision id %v cannot be reverted", revid))
+	}
+	snapshot := j.validRevisions[idx].journalIndex
+
+	// Replay the journal to undo journal changes and remove invalidated snapshots
+	j.revertWithoutRevertState(s, snapshot)
+	j.validRevisions = j.validRevisions[:idx]
+}
+
 // append inserts a new modification entry to the end of the change journal.
 func (j *journal) append(entry journalEntry) {
 	j.entries = append(j.entries, entry)
@@ -133,6 +176,18 @@ func (j *journal) revert(statedb *StateDB, snapshot int) {
 	j.entries = j.entries[:snapshot]
 }
 
+func (j *journal) revertWithoutRevertState(statedb *StateDB, snapshot int) {
+	for i := len(j.entries) - 1; i >= snapshot; i-- {
+		// Drop any dirty tracking induced by the change
+		if addr := j.entries[i].dirtied(); addr != nil {
+			if j.dirties[*addr]--; j.dirties[*addr] == 0 {
+				delete(j.dirties, *addr)
+			}
+		}
+	}
+	j.entries = j.entries[:snapshot]
+}
+
 // dirty explicitly sets an address to dirty, even if the change entries would
 // otherwise suggest it as clean. This method is an ugly hack to handle the RIPEMD
 // precompile consensus exception.

core/state/state_object.go

@@ -115,6 +115,7 @@ func (s *stateObject) markSelfdestructed() {
 
 func (s *stateObject) touch() {
 	s.db.journal.touchChange(s.address)
+	s.db.updateJournal.touchChange(s.address)
 }
 
 // getTrie returns the associated storage trie. The trie will be opened if it's

core/state/statedb.go

@@ -18,9 +18,11 @@
 package state
 
 import (
+	"encoding/json"
 	"errors"
 	"fmt"
 	"maps"
+	"os"
 	"slices"
 	"sync"
 	"sync/atomic"
@@ -159,6 +161,84 @@ type StateDB struct {
 	StorageDeleted atomic.Int64 // Number of storage slots deleted during the state transition
 }
 
+type BALType int
+
+const (
+	OnlyKey BALType = iota
+	WithAddrKeySlotV
+	WithAllKV
+	BalKeyConstruction
+	BalKeyValConstruction
+	BalPreblockKeysPostValues
+)
+
+type BALs struct {
+	Pre  map[uint64]types.AccessList     `json:"pre"`
+	Post map[uint64]map[int]TxPostValues `json:"post"`
+}
+
+type AcctPostValues struct {
+	Nonce     uint64                      `json:"nonce"`
+	Balance   *uint256.Int                `json:"balance"`
+	Code      []byte                      `json:"code"`
+	StorageKV map[common.Hash]common.Hash `json:"storageKV"`
+	Destruct  bool                        `json:"destruct"`
+}
+
+// For acccount destruct or storage clearing corresponding values would be 0
+type TxPostValues map[common.Address]*AcctPostValues
+
+var (
+	AllBlockBal         = BALs{}
+	AllBlockAccessLists = map[uint64]types.AccessList{}
+	// Blocknumber => TxIndex => TxPostValues
+	AllBlockTxPostValues = map[uint64]map[int]TxPostValues{}
+)
+
+const (
+	balType = BalPreblockKeysPostValues
+	// balType = OnlyKey
+)
+
+func init() {
+	// load the access lists for all blocks
+	println("Importing BAL:", balType)
+	var (
+		fileName string
+	)
+	switch balType {
+	case OnlyKey:
+		{
+			println("bal onlykey")
+			fileName = "access_lists.2000.json"
+		}
+	case WithAddrKeySlotV:
+	case WithAllKV:
+	case BalKeyConstruction:
+		{
+		}
+	case BalPreblockKeysPostValues:
+		{
+			println("bal preblock keys post values")
+			fileName = "access_lists_kpostv.json"
+			data, err := os.ReadFile(fileName)
+			if err != nil {
+				log.Error("Failed to load access lists", "err", err)
+				return
+			}
+			if err := json.Unmarshal(data, &AllBlockBal); err != nil {
+				panic("Failed to unmarshal access lists")
+			}
+
+			AllBlockAccessLists = AllBlockBal.Pre
+			AllBlockTxPostValues = AllBlockBal.Post
+		}
+	}
+
+	println("Imported BAL", fileName)
+	println("Access lists loaded:", len(AllBlockAccessLists), "blocks with pre-block-keys, ", len(AllBlockTxPostValues), "blocks with post values")
+}
+
 // New creates a new state from a given trie.
 func New(root common.Hash, db Database) (*StateDB, error) {
 	reader, err := db.Reader(root)
@@ -245,6 +325,7 @@ func (s *StateDB) Error() error {
 
 func (s *StateDB) AddLog(log *types.Log) {
 	s.journal.logChange(s.thash)
+	s.updateJournal.logChange(s.thash)
 
 	log.TxHash = s.thash
 	log.TxIndex = uint(s.txIndex)
@@ -288,6 +369,7 @@ func (s *StateDB) Preimages() map[common.Hash][]byte {
 func (s *StateDB) AddRefund(gas uint64) {
 	s.journal.refundChange(s.refund)
 	s.refund += gas
+	s.updateJournal.refundChange(s.refund)
 }
 
 func (s *StateDB) JournalEntriesCopy() []JournalEntry {
@@ -320,16 +402,23 @@ func (s *StateDB) MergeState(entries []JournalEntry) {
 			}
 		case codeChange:
 			s.SetCode(v.account, v.prevCode)
+		case refundChange:
+			s.setRefund(v.prev)
 		default:
 
 		}
 	}
 }
 
+func (s *StateDB) setRefund(gas uint64) {
+	s.refund = gas
+}
+
 // SubRefund removes gas from the refund counter.
 // This method will panic if the refund counter goes below zero
 func (s *StateDB) SubRefund(gas uint64) {
 	s.journal.refundChange(s.refund)
+	s.updateJournal.refundChange(s.refund - gas)
 	if gas > s.refund {
 		panic(fmt.Sprintf("Refund counter below zero (gas: %d > refund: %d)", gas, s.refund))
 	}
@@ -581,6 +670,7 @@ func (s *StateDB) SetTransientState(addr common.Address, key, value common.Hash)
 		return
 	}
 	s.journal.transientStateChange(addr, key, prev)
+	s.updateJournal.transientStateChange(addr, key, prev)
 	s.setTransientState(addr, key, value)
 }
 
@@ -764,12 +854,15 @@ func (s *StateDB) Copy() *StateDB {
 
 // Snapshot returns an identifier for the current revision of the state.
 func (s *StateDB) Snapshot() int {
+	s.updateJournal.snapshot()
 	return s.journal.snapshot()
 }
 
 // RevertToSnapshot reverts all state changes made since the given revision.
 func (s *StateDB) RevertToSnapshot(revid int) {
 	s.journal.revertToSnapshot(revid, s)
+	s.updateJournal.revertToSnapshotWithoutRevertState(revid, s)
+
 }
 
 // GetRefund returns the current value of the refund counter.
@@ -981,6 +1074,7 @@ func (s *StateDB) SetTxContext(thash common.Hash, ti int) {
 
 func (s *StateDB) clearJournalAndRefund() {
 	s.journal.reset()
+	s.updateJournal.reset()
 	s.refund = 0
 }
 
@@ -1429,6 +1523,7 @@ func (s *StateDB) Prepare(rules params.Rules, sender, coinbase common.Address, d
 func (s *StateDB) AddAddressToAccessList(addr common.Address) {
 	if s.accessList.AddAddress(addr) {
 		s.journal.accessListAddAccount(addr)
+		s.updateJournal.accessListAddAccount(addr)
 	}
 }
 
@@ -1441,9 +1536,11 @@ func (s *StateDB) AddSlotToAccessList(addr common.Address, slot common.Hash) {
 		// to the access list (via call-variant, create, etc).
 		// Better safe than sorry, though
 		s.journal.accessListAddAccount(addr)
+		s.updateJournal.accessListAddAccount(addr)
 	}
 	if slotMod {
 		s.journal.accessListAddSlot(addr, slot)
+		s.updateJournal.accessListAddSlot(addr, slot)
 	}
 }