seq poststate

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

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)

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()))
+		receipts    = make(types.Receipts, len(block.Transactions()))
-		header           = block.Header()
+		header      = block.Header()
-		blockHash        = block.Hash()
+		blockHash   = block.Hash()
-		blockNumber      = block.Number()
+		blockNumber = block.Number()
-		allLogs          []*types.Log
+		allLogs     []*types.Log
-		preStatedb       = statedb.Copy()
+
 		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{}

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:

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
 //

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 {
+var (
-	// 1. Fetch all pre-block state
+	StateFinalizeTime = time.Duration(0)
-	// 2. Merge with post-state
+	StateCopyTime     = time.Duration(0)
-	if balType != BalPreblockKeysPostValues {
+	StateNewTime      = time.Duration(0)
-		panic("PreStateAtTxIndex is only supported with BalPreblockKeysPostValues")
+	StateDeepCpTime   = time.Duration(0)
-	}
+	StateSetTime      = time.Duration(0)
+	StateLoadTime     = time.Duration(0)
+)
 
-	// Merge with BALs post state
+func (s *StateDB) MergePostBal() {
-	return nil
+	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()