diff --git a/core/parallel_state_processor.go b/core/parallel_state_processor.go index c28531c709..6d2b9f9662 100644 --- a/core/parallel_state_processor.go +++ b/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 @@ -61,35 +71,50 @@ func (p *ParallelStateProcessor) Process(block *types.Block, statedb *state.Stat 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) + receipts = make(types.Receipts, len(block.Transactions())) + header = block.Header() + blockHash = block.Hash() + blockNumber = block.Number() + allLogs []*types.Log + preStatedb = statedb.Copy() + preStateProvider PreStateProvider + 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 := postStates[i] + cleanStatedb, err := preStateProvider.PrestateAtIndex(i) + if err != nil { + return nil, err + } + i := i - - wg.Add(1) - go func() { - defer wg.Done() - + workers.Go(func() error { 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 - 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) - } + // if i == len(block.Transactions())-1 { + // *statedb = *cleanStatedb + // } + return nil + }) } - wg.Wait() + err := workers.Wait() + if err != nil { + return nil, err + } // 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 +} diff --git a/core/state/journal.go b/core/state/journal.go index d69e0d731a..87b376715f 100644 --- a/core/state/journal.go +++ b/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. diff --git a/core/state/state_object.go b/core/state/state_object.go index e6650418d1..96263bcaa3 100644 --- a/core/state/state_object.go +++ b/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 diff --git a/core/state/statedb.go b/core/state/statedb.go index 75f2b88c48..127cf46f7a 100644 --- a/core/state/statedb.go +++ b/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) } }