parallel merging poststate while executing txs

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 {

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()
+		header    = block.Header()
-		gp     = new(GasPool).AddGas(block.GasLimit())
+		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
+		// // EIP-7002
-		if err := ProcessWithdrawalQueue(&requests, evm); err != nil {
+		// if err := ProcessWithdrawalQueue(&requests, evm); err != nil {
-			return nil, err
+		// 	return nil, err
-		}
+		// }
-		// EIP-7251
+		// // EIP-7251
-		if err := ProcessConsolidationQueue(&requests, evm); err != nil {
+		// if err := ProcessConsolidationQueue(&requests, evm); err != nil {
-			return nil, err
+		// 	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())
+	// 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
 
 	return &ProcessResult{
 		Receipts: receipts,

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
+			// codeHash
-			}
+			if acctVal.Code != nil {
-			if acctVal.Root == (common.Hash{}) {
+				codeHash = crypto.Keccak256(acctVal.Code)
-				acctVal.Root = acct.Root
 			}
+
 			return &types.StateAccount{
-				Nonce:    acctVal.Nonce,
+				Nonce:    nonce,
-				Balance:  acctVal.Balance,
+				Balance:  balance,
-				Root:     acctVal.Root,
+				Root:     acct.Root,
-				CodeHash: acctVal.CodeHash,
+				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)
 	}

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)
+			acctBal, err := s.reader.AccountBAL(addr)
-			obj := newObject(s, addr, acct)
+			var obj *stateObject
-			accounts <- obj
 			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)
+			fmt.Println(s.originalRoot, obj.addrHash, obj.origin.Root)
-			panic("fail to create trie for:")
+			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 {

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
+	if obj != nil {
-}
+		return &obj.data, nil
-
+	}
-func (s *StateDB) AccountBAL(addr common.Address) (*types.StateAccount, error) {
+	return nil, fmt.Errorf("readerWithBal account %v, not exist", addr)
-	panic("AccountBAL not implemented for statedb")
 }
 
 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")
 }

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