add pre-block fetcher and pass test in vanilla block processor

core/paralle_state_provider.go

@@ -0,0 +1,60 @@
+package core
+
+import (
+	"fmt"
+
+	"github.com/ethereum/go-ethereum/core/state"
+	"github.com/ethereum/go-ethereum/core/types"
+	"github.com/ethereum/go-ethereum/core/vm"
+)
+
+type PreStateType int
+
+const (
+	BALPreState PreStateType = iota
+	SeqPreState              // Only for debug
+)
+
+const preStateType = SeqPreState
+
+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/parallel_state_processor.go

@@ -14,15 +14,6 @@ import (
 	"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
@@ -127,9 +118,6 @@ func (p *ParallelStateProcessor) executeParallel(block *types.Block, statedb *st
 			receipts[i] = receipt
 			postEntries[i] = entries
 
-			// if i == len(block.Transactions())-1 {
-			// 	*statedb = *cleanStatedb
-			// }
 			return nil
 		})
 	}
@@ -217,45 +205,3 @@ 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/statedb.go

@@ -159,6 +159,9 @@ type StateDB struct {
 	StorageLoaded  int          // Number of storage slots retrieved from the database during the state transition
 	StorageUpdated atomic.Int64 // Number of storage slots updated during the state transition
 	StorageDeleted atomic.Int64 // Number of storage slots deleted during the state transition
+
+	// The block number context for BALs
+	blockNumber uint64
 }
 
 type BALType int
@@ -276,6 +279,94 @@ func NewWithReader(root common.Hash, db Database, reader Reader) (*StateDB, erro
 	return sdb, nil
 }
 
+func (s *StateDB) PrefetchStateBAL(blockNumber uint64) {
+	s.blockNumber = blockNumber
+	switch balType {
+	case BalPreblockKeysPostValues:
+		s.PrefetchBalPreblockKeys()
+	}
+}
+
+func (s *StateDB) PrefetchBalPreblockKeys() {
+	log.Info("PrefetchBalPreblockKeys...")
+	type StorageKV struct {
+		addr *common.Address
+		key  *common.Hash
+		val  *common.Hash
+	}
+	var (
+		preBal   = AllBlockAccessLists[s.blockNumber]
+		lenSlots = 0
+		workers  errgroup.Group
+	)
+
+	workers.SetLimit(25)
+	// Fetch pre-block acccount state
+	accounts := make(chan *stateObject, len(preBal))
+	for _, acl := range preBal {
+		addr := acl.Address
+		lenSlots += len(acl.StorageKeys)
+
+		workers.Go(func() error {
+			acct, err := s.reader.Account(addr)
+			obj := newObject(s, addr, acct)
+			accounts <- obj
+			if err != nil {
+				log.Error("fail to fetch account from BALs:", addr)
+				return err
+			}
+			return nil
+		})
+	}
+	workers.Wait()
+
+	for range len(preBal) {
+		obj := <-accounts
+		// must set it first to avoid accounts read later in storageBAL fetching
+		s.setStateObject(obj)
+	}
+
+	// Fetch pre-block storage state
+	storages := make(chan *StorageKV, lenSlots)
+	for _, acl := range preBal {
+		addr := acl.Address
+		keys := acl.StorageKeys
+		for _, key := range keys {
+			workers.Go(func() error {
+				val, err := s.reader.Storage(addr, key)
+				kv := &StorageKV{&addr, &key, &val}
+				storages <- kv
+				if err != nil {
+					log.Error("fail to fetch storage from BALs:", addr, key)
+					return err
+				}
+				return nil
+			})
+		}
+	}
+	workers.Wait()
+
+	for range lenSlots {
+		kv := <-storages
+		account := s.getStateObject(*kv.addr)
+		if account != nil {
+			account.originStorage[*kv.key] = *kv.val
+			s.setStateObject(account)
+		}
+	}
+}
+
+func (s *StateDB) PreStateAtTxIndex(index int) *StateDB {
+	// 1. Fetch all pre-block state
+	// 2. Merge with post-state
+	if balType != BalPreblockKeysPostValues {
+		panic("PreStateAtTxIndex is only supported with BalPreblockKeysPostValues")
+	}
+
+	// Merge with BALs post state
+	return nil
+}
+
 // StartPrefetcher initializes a new trie prefetcher to pull in nodes from the
 // state trie concurrently while the state is mutated so that when we reach the
 // commit phase, most of the needed data is already hot.