diff --git a/cmd/geth/chaincmd.go b/cmd/geth/chaincmd.go index 6589fb35d0..f078c5ab9c 100644 --- a/cmd/geth/chaincmd.go +++ b/cmd/geth/chaincmd.go @@ -439,7 +439,7 @@ func showMetrics() { blockWriteTimer := metrics.GetOrRegisterResettingTimer("chain/write", nil) blockPrefetchExecuteTimer := metrics.GetOrRegisterResettingTimer("chain/prefetch/executes", nil) - mgaspsHist := metrics.GetOrRegisterHistogram("chain/execution/mgasps", nil, metrics.NewUniformSample(2000)) + chainMgaspsMeter := metrics.GetOrRegisterResettingTimer("chain/mgasps", nil) // not important fmt.Println("accountReadSingleTimer", accountReadSingleTimer.Total()) @@ -475,14 +475,16 @@ func showMetrics() { fmt.Println("PrefetchBALtime: ", core.PrefetchBALTime) fmt.Println("PrefetchMergeTime:", core.PrefetchMergeBALTime) fmt.Println("ParallelExeTime: ", core.ParallelExeTime) - fmt.Println("PostMergeTime: ", core.PostMergeTime) fmt.Println("PrefetchTrieWallTime: ", core.PrefetchTrieTimer) // fmt.Println("PrefetchTrieCPUtime: ", state.PrefetchTrieCPUTime) + // Sync state time + fmt.Println("PrefetchChTime: ", core.PrefetchChTime) + // total fmt.Println("blockInsertTimer", blockInsertTimer.Total()) - mgasps := mgaspsHist.Snapshot() + mgasps := chainMgaspsMeter.Snapshot() fmt.Println("mgasps,mean,max,min:", int64(mgasps.Mean()), mgasps.Max(), mgasps.Min()) } diff --git a/cmd/utils/cmd.go b/cmd/utils/cmd.go index 6931dd332c..1671175858 100644 --- a/cmd/utils/cmd.go +++ b/cmd/utils/cmd.go @@ -52,7 +52,7 @@ import ( ) const ( - importBatchSize = 1 + importBatchSize = 10 ) // ErrImportInterrupted is returned when the user interrupts the import process. diff --git a/core/block_validator.go b/core/block_validator.go index 591e472bc1..d074ff96a1 100644 --- a/core/block_validator.go +++ b/core/block_validator.go @@ -20,7 +20,6 @@ import ( "errors" "fmt" - "github.com/ethereum/go-ethereum/consensus" "github.com/ethereum/go-ethereum/core/state" "github.com/ethereum/go-ethereum/core/types" "github.com/ethereum/go-ethereum/params" @@ -108,12 +107,13 @@ func (v *BlockValidator) ValidateBody(block *types.Block) error { } // Ancestor block must be known. - if !v.bc.HasBlockAndState(block.ParentHash(), block.NumberU64()-1) { - if !v.bc.HasBlock(block.ParentHash(), block.NumberU64()-1) { - return consensus.ErrUnknownAncestor - } - return consensus.ErrPrunedAncestor - } + // Don't check intermediate blocks for N-blocks + // if !v.bc.HasBlockAndState(block.ParentHash(), block.NumberU64()-1) { + // if !v.bc.HasBlock(block.ParentHash(), block.NumberU64()-1) { + // return consensus.ErrUnknownAncestor + // } + // return consensus.ErrPrunedAncestor + // } return nil } @@ -148,19 +148,20 @@ func (v *BlockValidator) ValidateState(block *types.Block, statedb *state.StateD return fmt.Errorf("invalid receipt root hash (remote: %x local: %x)", header.ReceiptHash, receiptSha) } // Validate the parsed requests match the expected header value. - if header.RequestsHash != nil { - reqhash := types.CalcRequestsHash(res.Requests) - if reqhash != *header.RequestsHash { - return fmt.Errorf("invalid requests hash (remote: %x local: %x)", *header.RequestsHash, reqhash) - } - } else if res.Requests != nil { - return errors.New("block has requests before prague fork") - } + // if header.RequestsHash != nil { + // reqhash := types.CalcRequestsHash(res.Requests) + // if reqhash != *header.RequestsHash { + // return fmt.Errorf("invalid requests hash (remote: %x local: %x)", *header.RequestsHash, reqhash) + // } + // } else if res.Requests != nil { + // return errors.New("block has requests before prague fork") + // } // Validate the state root against the received state root and throw // an error if they don't match. - if root := statedb.IntermediateRoot(v.config.IsEIP158(header.Number)); header.Root != root { - return fmt.Errorf("invalid merkle root (remote: %x local: %x) dberr: %w", header.Root, root, statedb.Error()) - } + // Commented for N-blocks + // if root := statedb.IntermediateRoot(v.config.IsEIP158(header.Number)); header.Root != root { + // return fmt.Errorf("invalid merkle root (remote: %x local: %x) dberr: %w", header.Root, root, statedb.Error()) + // } return nil } diff --git a/core/blockchain.go b/core/blockchain.go index cbf56f4690..1d3852ad59 100644 --- a/core/blockchain.go +++ b/core/blockchain.go @@ -108,8 +108,6 @@ var ( blockPrefetchTxsInvalidMeter = metrics.NewRegisteredMeter("chain/prefetch/txs/invalid", nil) blockPrefetchTxsValidMeter = metrics.NewRegisteredMeter("chain/prefetch/txs/valid", nil) - mgaspsHist = metrics.NewRegisteredHistogram("chain/execution/mgasps", nil, metrics.NewUniformSample(2000)) - errInsertionInterrupted = errors.New("insertion is interrupted") errChainStopped = errors.New("blockchain is stopped") errInvalidOldChain = errors.New("invalid old chain") @@ -1712,7 +1710,8 @@ func (bc *BlockChain) InsertChain(chain types.Blocks) (int, error) { } defer bc.chainmu.Unlock() - _, n, err := bc.insertChain(chain, true, false) // No witness collection for mass inserts (would get super large) + // _, n, err := bc.insertChain(chain, true, false) // No witness collection for mass inserts (would get super large) + _, n, err := bc.insertChainN(chain, true, false) return n, err } @@ -1946,6 +1945,9 @@ type blockProcessingResult struct { procTime time.Duration status WriteStatus witness *stateless.Witness + // For merged N-blocks + receipts types.Receipts + logs []*types.Log } // processBlock executes and validates the given block. If there was no error diff --git a/core/blockchainN.go b/core/blockchainN.go index 4ce6521986..78d8a917f4 100644 --- a/core/blockchainN.go +++ b/core/blockchainN.go @@ -20,6 +20,7 @@ package core import ( "errors" "fmt" + "sync" "sync/atomic" "time" @@ -35,6 +36,13 @@ import ( "github.com/ethereum/go-ethereum/log" ) +var ( + PrefetchBALTime = time.Duration(0) + PostMergeTime = time.Duration(0) + PrefetchTrieTimer = time.Duration(0) + PrefetchChTime = time.Duration(0) +) + func (bc *BlockChain) insertChainN(chain types.Blocks, setHead bool, makeWitness bool) (*stateless.Witness, int, error) { // If the chain is terminating, don't even bother starting up. if bc.insertStopped() { @@ -133,125 +141,228 @@ func (bc *BlockChain) insertChainN(chain types.Blocks, setHead bool, makeWitness return nil, it.index, err } // Track the singleton witness from this chain insertion (if any) - var witness *stateless.Witness + var ( + witness *stateless.Witness + wg sync.WaitGroup + allReceipts types.Receipts + allLogs []*types.Log + totalGasUsed uint64 + headerTime time.Duration + ) - for ; block != nil && err == nil || errors.Is(err, ErrKnownBlock); block, err = it.next() { - // If the chain is terminating, stop processing blocks - if bc.insertStopped() { - log.Debug("Abort during block processing") - break + // All blocks share the same stateDB to simulate commiting after processing multiple blocks + startBlock := block + endBlock := chain[len(chain)-1] + parent := it.previous() + if parent == nil { + parent = bc.GetHeader(startBlock.ParentHash(), startBlock.NumberU64()-1) + } + statedb, err := state.New(parent.Root, bc.statedb) + if err != nil { + log.Crit("failed to initailzied state", "error", err, "root:", parent.Root.Hex()) + } + // Prefetch pre-N-blocks state with merged pre-block BALs for N-blocks + // Here pre-block BALs are not merged, but we skipped allready fetch state to simulate merged operations. + prefetchStart := time.Now() + for _, blk := range chain { + statedb.PrefetchStateBAL(blk.NumberU64()) + } + PrefetchBALTime += time.Since(prefetchStart) + + // pre-block state for block N + prestateCh := make(chan *state.StateDB, len(chain)) + + wg.Add(2) + // process post-N-blocks state with merged post-BALs for N-blocks + go func() { + defer wg.Done() + + mstart := time.Now() + // Stop prefetcher cause we'll directly fetch tries in parallel + statedb.StopPrefetcher() + + for _, blk := range chain { + // We don't need to worry about blockNumber is not changed, because it'll be set during process block + prestateCh <- statedb.CopyState() + statedb.MergePostBalStates(blk.NumberU64()) } - // If the block is known (in the middle of the chain), it's a special case for - // Clique blocks where they can share state among each other, so importing an - // older block might complete the state of the subsequent one. In this case, - // just skip the block (we already validated it once fully (and crashed), since - // its header and body was already in the database). But if the corresponding - // snapshot layer is missing, forcibly rerun the execution to build it. - if bc.skipBlock(err, it) { - logger := log.Debug - if bc.chainConfig.Clique == nil { - logger = log.Warn - } - logger("Inserted known block", "number", block.Number(), "hash", block.Hash(), - "uncles", len(block.Uncles()), "txs", len(block.Transactions()), "gas", block.GasUsed(), - "root", block.Root()) + PostMergeTime += time.Since(mstart) - // Special case. Commit the empty receipt slice if we meet the known - // block in the middle. It can only happen in the clique chain. Whenever - // we insert blocks via `insertSideChain`, we only commit `td`, `header` - // and `body` if it's non-existent. Since we don't have receipts without - // reexecution, so nothing to commit. But if the sidechain will be adopted - // as the canonical chain eventually, it needs to be reexecuted for missing - // state, but if it's this special case here(skip reexecution) we will lose - // the empty receipt entry. - if len(block.Transactions()) == 0 { - rawdb.WriteReceipts(bc.db, block.Hash(), block.NumberU64(), nil) - } else { - log.Error("Please file an issue, skip known block execution without receipt", - "hash", block.Hash(), "number", block.NumberU64()) + // Prewarm the trie for future committing + pstart := time.Now() + statedb.PrefetchTrie() + PrefetchTrieTimer += time.Since(pstart) + }() + + go func() { + defer wg.Done() + + hstart := time.Now() + // Write all headers then parallel executing to validate post-tx BALs + bc.writeNBlockHeaders(chain) + headerTime = time.Since(hstart) + + for ; block != nil && err == nil || errors.Is(err, ErrKnownBlock); block, err = it.next() { + // If the chain is terminating, stop processing blocks + if bc.insertStopped() { + log.Debug("Abort during block processing") + break } - if err := bc.writeKnownBlock(block); err != nil { - return nil, it.index, err + // If the block is known (in the middle of the chain), it's a special case for + // Clique blocks where they can share state among each other, so importing an + // older block might complete the state of the subsequent one. In this case, + // just skip the block (we already validated it once fully (and crashed), since + // its header and body was already in the database). But if the corresponding + // snapshot layer is missing, forcibly rerun the execution to build it. + if bc.skipBlock(err, it) { + logger := log.Debug + if bc.chainConfig.Clique == nil { + logger = log.Warn + } + logger("Inserted known block", "number", block.Number(), "hash", block.Hash(), + "uncles", len(block.Uncles()), "txs", len(block.Transactions()), "gas", block.GasUsed(), + "root", block.Root()) + + // Special case. Commit the empty receipt slice if we meet the known + // block in the middle. It can only happen in the clique chain. Whenever + // we insert blocks via `insertSideChain`, we only commit `td`, `header` + // and `body` if it's non-existent. Since we don't have receipts without + // reexecution, so nothing to commit. But if the sidechain will be adopted + // as the canonical chain eventually, it needs to be reexecuted for missing + // state, but if it's this special case here(skip reexecution) we will lose + // the empty receipt entry. + if len(block.Transactions()) == 0 { + rawdb.WriteReceipts(bc.db, block.Hash(), block.NumberU64(), nil) + } else { + log.Error("Please file an issue, skip known block execution without receipt", + "hash", block.Hash(), "number", block.NumberU64()) + } + if err := bc.writeKnownBlock(block); err != nil { + witness = nil + return + } + stats.processed++ + if bc.logger != nil && bc.logger.OnSkippedBlock != nil { + bc.logger.OnSkippedBlock(tracing.BlockEvent{ + Block: block, + Finalized: bc.CurrentFinalBlock(), + Safe: bc.CurrentSafeBlock(), + }) + } + // We can assume that logs are empty here, since the only way for consecutive + // Clique blocks to have the same state is if there are no transactions. + lastCanon = block + continue } + // Retrieve the parent block and it's state to execute on top + parent := it.previous() + if parent == nil { + parent = bc.GetHeader(block.ParentHash(), block.NumberU64()-1) + } + // The traced section of block import. + pstart := time.Now() + statedbForBlock := <-prestateCh + PrefetchChTime += time.Since(pstart) + + res, err := bc.processBlockWithState(parent.Root, block, setHead, makeWitness && len(chain) == 1, statedbForBlock) + if err != nil { + witness = nil + log.Crit("Failed to processBlock", "error", err) + return + } + + // collect logs and receipts for N-blocks + allReceipts = append(allReceipts, res.receipts...) + allLogs = append(allLogs, res.logs...) + totalGasUsed += res.usedGas + // Report the import stats before returning the various results stats.processed++ - if bc.logger != nil && bc.logger.OnSkippedBlock != nil { - bc.logger.OnSkippedBlock(tracing.BlockEvent{ - Block: block, - Finalized: bc.CurrentFinalBlock(), - Safe: bc.CurrentSafeBlock(), - }) + stats.usedGas += res.usedGas + witness = res.witness + + var snapDiffItems, snapBufItems common.StorageSize + if bc.snaps != nil { + snapDiffItems, snapBufItems = bc.snaps.Size() + } + trieDiffNodes, trieBufNodes, _ := bc.triedb.Size() + stats.report(chain, it.index, snapDiffItems, snapBufItems, trieDiffNodes, trieBufNodes, setHead) + + // Print confirmation that a future fork is scheduled, but not yet active. + bc.logForkReadiness(block) + + if !setHead { + // After merge we expect few side chains. Simply count + // all blocks the CL gives us for GC processing time + bc.gcproc += res.procTime + witness = nil + return // Direct block insertion of a single block } - // We can assume that logs are empty here, since the only way for consecutive - // Clique blocks to have the same state is if there are no transactions. - lastCanon = block - continue } - // Retrieve the parent block and it's state to execute on top - parent := it.previous() - if parent == nil { - parent = bc.GetHeader(block.ParentHash(), block.NumberU64()-1) - } - // The traced section of block import. - start := time.Now() - res, err := bc.processBlock(parent.Root, block, setHead, makeWitness && len(chain) == 1) - if err != nil { - return nil, it.index, err - } - // Report the import stats before returning the various results - stats.processed++ - stats.usedGas += res.usedGas - witness = res.witness - var snapDiffItems, snapBufItems common.StorageSize - if bc.snaps != nil { - snapDiffItems, snapBufItems = bc.snaps.Size() - } - trieDiffNodes, trieBufNodes, _ := bc.triedb.Size() - stats.report(chain, it.index, snapDiffItems, snapBufItems, trieDiffNodes, trieBufNodes, setHead) + stats.ignored += it.remaining() + }() + wg.Wait() - // Print confirmation that a future fork is scheduled, but not yet active. - bc.logForkReadiness(block) + // Validate stateRoot for N-blocks at once. + xvtime := time.Now() + header := endBlock.Header() + if root := statedb.IntermediateRoot(true); header.Root != root { + log.Crit("invalid merkle root (remote: %x local: %x) dberr: %w", header.Root, root, statedb.Error()) + } + blockCrossValidationTimer.Update(time.Since(xvtime)) - if !setHead { - // After merge we expect few side chains. Simply count - // all blocks the CL gives us for GC processing time - bc.gcproc += res.procTime - return witness, it.index, nil // Direct block insertion of a single block - } - switch res.status { - case CanonStatTy: - log.Debug("Inserted new block", "number", block.Number(), "hash", block.Hash(), - "uncles", len(block.Uncles()), "txs", len(block.Transactions()), "gas", block.GasUsed(), - "elapsed", common.PrettyDuration(time.Since(start)), - "root", block.Root()) - - lastCanon = block - - // Only count canonical blocks for GC processing time - bc.gcproc += res.procTime - - case SideStatTy: - log.Debug("Inserted forked block", "number", block.Number(), "hash", block.Hash(), - "diff", block.Difficulty(), "elapsed", common.PrettyDuration(time.Since(start)), - "txs", len(block.Transactions()), "gas", block.GasUsed(), "uncles", len(block.Uncles()), - "root", block.Root()) - - default: - // This in theory is impossible, but lets be nice to our future selves and leave - // a log, instead of trying to track down blocks imports that don't emit logs. - log.Warn("Inserted block with unknown status", "number", block.Number(), "hash", block.Hash(), - "diff", block.Difficulty(), "elapsed", common.PrettyDuration(time.Since(start)), - "txs", len(block.Transactions()), "gas", block.GasUsed(), "uncles", len(block.Uncles()), - "root", block.Root()) - } + // Commit N-blocks together to the chain and get the status. + var ( + wstart = time.Now() + status WriteStatus + ) + if !setHead { + // Don't set the head, only insert the block + err = bc.writeNBlocksWithState(startBlock, endBlock, allReceipts, statedb) + } else { + status, err = bc.writeNBlocksAndSetHead(startBlock, endBlock, allReceipts, allLogs, statedb, false) + } + if err != nil { + return nil, it.index, err } - stats.ignored += it.remaining() + // Update the metrics touched during N-blocks commit + accountCommitTimer.Update(statedb.AccountCommits) // Account commits are complete, we can mark them + storageCommitTimer.Update(statedb.StorageCommits) // Storage commits are complete, we can mark them + snapshotCommitTimer.Update(statedb.SnapshotCommits) // Snapshot commits are complete, we can mark them + triedbCommitTimer.Update(statedb.TrieDBCommits) // Trie database commits are complete, we can mark them + + blockWriteTimer.Update(time.Since(wstart) - max(statedb.AccountCommits, statedb.StorageCommits) /* concurrent */ - statedb.SnapshotCommits - statedb.TrieDBCommits + headerTime) + + switch status { + case CanonStatTy: + log.Debug("Inserted new blocks", "numberStart", startBlock.Number(), "numberEnd", endBlock.Number(), + "elapsed", common.PrettyDuration(time.Since(prefetchStart)), + "root", endBlock.Root()) + + lastCanon = endBlock + + default: + // This in theory is impossible, but lets be nice to our future selves and leave + // a log, instead of trying to track down blocks imports that don't emit logs. + log.Warn("Inserted block with unknown status", "number", endBlock.Number(), "hash", endBlock.Hash(), + "diff", endBlock.Difficulty(), "elapsed", common.PrettyDuration(time.Since(prefetchStart)), + "txs", len(endBlock.Transactions()), "gas", endBlock.GasUsed(), "uncles", len(endBlock.Uncles()), + "root", endBlock.Root()) + } + + elapsed := time.Since(prefetchStart) + 1 // prevent zero division + blockInsertTimer.Update(elapsed) + + // TODO(rjl493456442) generalize the ResettingTimer + mgasps := float64(totalGasUsed) * 1000 / float64(elapsed) + chainMgaspsMeter.Update(time.Duration(mgasps)) + return witness, it.index, err } -func (bc *BlockChain) writeNBlocksWithState(block *types.Block, receipts []*types.Receipt, statedb *state.StateDB) error { - if !bc.HasHeader(block.ParentHash(), block.NumberU64()-1) { +func (bc *BlockChain) writeNBlocksWithState(startBlock, endBlock *types.Block, receipts []*types.Receipt, statedb *state.StateDB) error { + if !bc.HasHeader(startBlock.ParentHash(), startBlock.NumberU64()-1) { return consensus.ErrUnknownAncestor } // Irrelevant of the canonical status, write the block itself to the database. @@ -259,14 +370,14 @@ func (bc *BlockChain) writeNBlocksWithState(block *types.Block, receipts []*type // Note all the components of block(hash->number map, header, body, receipts) // should be written atomically. BlockBatch is used for containing all components. blockBatch := bc.db.NewBatch() - rawdb.WriteBlock(blockBatch, block) - rawdb.WriteReceipts(blockBatch, block.Hash(), block.NumberU64(), receipts) + rawdb.WriteBlock(blockBatch, endBlock) + rawdb.WriteReceipts(blockBatch, endBlock.Hash(), endBlock.NumberU64(), receipts) rawdb.WritePreimages(blockBatch, statedb.Preimages()) if err := blockBatch.Write(); err != nil { log.Crit("Failed to write block into disk", "err", err) } // Commit all cached state changes into underlying memory database. - root, err := statedb.Commit(block.NumberU64(), bc.chainConfig.IsEIP158(block.Number()), bc.chainConfig.IsCancun(block.Number(), block.Time())) + root, err := statedb.Commit(endBlock.NumberU64(), bc.chainConfig.IsEIP158(endBlock.Number()), bc.chainConfig.IsCancun(endBlock.Number(), endBlock.Time())) if err != nil { return err } @@ -281,10 +392,10 @@ func (bc *BlockChain) writeNBlocksWithState(block *types.Block, receipts []*type } // Full but not archive node, do proper garbage collection bc.triedb.Reference(root, common.Hash{}) // metadata reference to keep trie alive - bc.triegc.Push(root, -int64(block.NumberU64())) + bc.triegc.Push(root, -int64(endBlock.NumberU64())) // Flush limits are not considered for the first TriesInMemory blocks. - current := block.NumberU64() + current := endBlock.NumberU64() if current <= state.TriesInMemory { return nil } @@ -330,23 +441,34 @@ func (bc *BlockChain) writeNBlocksWithState(block *types.Block, receipts []*type return nil } -func (bc *BlockChain) writeNBlocksAndSetHead(block *types.Block, receipts []*types.Receipt, logs []*types.Log, state *state.StateDB, emitHeadEvent bool) (status WriteStatus, err error) { - if err := bc.writeBlockWithState(block, receipts, state); err != nil { +// Due to blockHash opCode, blockHeaders must be written to DB before commit. +func (bc *BlockChain) writeNBlockHeaders(chain types.Blocks) { + blockBatch := bc.db.NewBatch() + for _, block := range chain { + rawdb.WriteHeader(blockBatch, block.Header()) + } + if err := blockBatch.Write(); err != nil { + log.Crit("Failed to write block into disk", "err", err) + } +} + +func (bc *BlockChain) writeNBlocksAndSetHead(startBlock, endBlock *types.Block, receipts []*types.Receipt, logs []*types.Log, state *state.StateDB, emitHeadEvent bool) (status WriteStatus, err error) { + if err := bc.writeBlockWithState(endBlock, receipts, state); err != nil { return NonStatTy, err } currentBlock := bc.CurrentBlock() // Reorganise the chain if the parent is not the head block - if block.ParentHash() != currentBlock.Hash() { - if err := bc.reorg(currentBlock, block.Header()); err != nil { + if startBlock.ParentHash() != currentBlock.Hash() { + if err := bc.reorg(currentBlock, endBlock.Header()); err != nil { return NonStatTy, err } } // Set new head. - bc.writeHeadBlock(block) + bc.writeHeadBlock(endBlock) - bc.chainFeed.Send(ChainEvent{Header: block.Header()}) + bc.chainFeed.Send(ChainEvent{Header: endBlock.Header()}) if len(logs) > 0 { bc.logsFeed.Send(logs) } @@ -356,33 +478,24 @@ func (bc *BlockChain) writeNBlocksAndSetHead(block *types.Block, receipts []*typ // we will fire an accumulated ChainHeadEvent and disable fire // event here. if emitHeadEvent { - bc.chainHeadFeed.Send(ChainHeadEvent{Header: block.Header()}) + bc.chainHeadFeed.Send(ChainHeadEvent{Header: endBlock.Header()}) } return CanonStatTy, nil } -func (bc *BlockChain) processBlockWithState(parentRoot common.Hash, block *types.Block, setHead bool, makeWitness bool) (_ *blockProcessingResult, blockEndErr error) { +func (bc *BlockChain) processBlockWithState(parentRoot common.Hash, block *types.Block, setHead bool, makeWitness bool, statedb *state.StateDB) (_ *blockProcessingResult, blockEndErr error) { var ( err error startTime = time.Now() - statedb *state.StateDB interrupt atomic.Bool ) defer interrupt.Store(true) // terminate the prefetch at the end + if !bc.cfg.NoPrefetch { + panic("Must enable --cache.noprefetch to perf N-blocks") + } if bc.cfg.NoPrefetch { - statedb, err = state.New(parentRoot, bc.statedb) - if err != nil { - return nil, err - } - // reader, err := bc.statedb.ReaderWithCache(parentRoot) - // if err != nil { - // return nil, err - // } - // statedb, err = state.NewWithReader(parentRoot, bc.statedb, reader) - // if err != nil { - // return nil, err - // } + // Use stateDB snapshotted from statedb.MergePostBalStates } else { fmt.Println("prefetch enabled===========================") // If prefetching is enabled, run that against the current state to pre-cache @@ -443,8 +556,9 @@ func (bc *BlockChain) processBlockWithState(parentRoot common.Hash, block *types return nil, err } } - statedb.StartPrefetcher("chain", witness) - defer statedb.StopPrefetcher() + // Don't start the prefetcher, as it can significantly degrade performance. We've already prefetched the trie in insertChainN. + // statedb.StartPrefetcher("chain", witness) + // defer statedb.StopPrefetcher() } if bc.logger != nil && bc.logger.OnBlockStart != nil { @@ -525,38 +639,10 @@ func (bc *BlockChain) processBlockWithState(parentRoot common.Hash, block *types blockValidationTimer.Update(vtime - (triehash + trieUpdate)) // The time spent on block validation blockCrossValidationTimer.Update(xvtime) // The time spent on stateless cross validation - // Write the block to the chain and get the status. - var ( - wstart = time.Now() - status WriteStatus - ) - if !setHead { - // Don't set the head, only insert the block - err = bc.writeBlockWithState(block, res.Receipts, statedb) - } else { - status, err = bc.writeBlockAndSetHead(block, res.Receipts, res.Logs, statedb, false) - } - if err != nil { - return nil, err - } - // Update the metrics touched during block commit - accountCommitTimer.Update(statedb.AccountCommits) // Account commits are complete, we can mark them - storageCommitTimer.Update(statedb.StorageCommits) // Storage commits are complete, we can mark them - snapshotCommitTimer.Update(statedb.SnapshotCommits) // Snapshot commits are complete, we can mark them - triedbCommitTimer.Update(statedb.TrieDBCommits) // Trie database commits are complete, we can mark them - - blockWriteTimer.Update(time.Since(wstart) - max(statedb.AccountCommits, statedb.StorageCommits) /* concurrent */ - statedb.SnapshotCommits - statedb.TrieDBCommits) - elapsed := time.Since(startTime) + 1 // prevent zero division - blockInsertTimer.Update(elapsed) - - // TODO(rjl493456442) generalize the ResettingTimer - mgasps := float64(res.GasUsed) * 1000 / float64(elapsed) - chainMgaspsMeter.Update(time.Duration(mgasps)) - return &blockProcessingResult{ usedGas: res.GasUsed, procTime: proctime, - status: status, + status: CanonStatTy, // Assue write status is always correct witness: witness, }, nil } diff --git a/core/blockchain_insert.go b/core/blockchain_insert.go index 7edd674f17..56e6ffc1fb 100644 --- a/core/blockchain_insert.go +++ b/core/blockchain_insert.go @@ -46,8 +46,6 @@ func (st *insertStats) report(chain []*types.Block, index int, snapDiffItems, sn elapsed = now.Sub(st.startTime) + 1 // prevent zero division mgasps = float64(st.usedGas) * 1000 / float64(elapsed) ) - // Update the Mgas per second gauge - mgaspsHist.Update(int64(mgasps)) // If we're at the last block of the batch or report period reached, log if index == len(chain)-1 || elapsed >= statsReportLimit { diff --git a/core/parallel_state_processor.go b/core/parallel_state_processor.go index 2b76ce468e..d8ea4c75da 100644 --- a/core/parallel_state_processor.go +++ b/core/parallel_state_processor.go @@ -4,7 +4,6 @@ import ( "fmt" "math/big" "runtime" - "sync" "time" "github.com/ethereum/go-ethereum/common" @@ -17,11 +16,9 @@ import ( ) var ( - PrefetchBALTime = time.Duration(0) - PrefetchMergeBALTime = time.Duration(0) ParallelExeTime = time.Duration(0) - PostMergeTime = time.Duration(0) - PrefetchTrieTimer = time.Duration(0) + PrefetchMergeBALTime = time.Duration(0) + EmptyStatedb = state.NewEmptyDB() ) type ParallelStateProcessor struct { @@ -43,10 +40,8 @@ func (p *ParallelStateProcessor) Process(block *types.Block, statedb *state.Stat context vm.BlockContext gp = new(GasPool).AddGas(block.GasLimit()) signer = types.MakeSigner(p.config, header.Number, header.Time) - initialdb = statedb.Copy() - postState *state.StateDB + initialdb = EmptyStatedb - wg sync.WaitGroup result *ProcessResult err error ) @@ -56,14 +51,9 @@ func (p *ParallelStateProcessor) Process(block *types.Block, statedb *state.Stat misc.ApplyDAOHardFork(statedb) } - if preStateType == BALPreState { - // copy initialdb before PrefetchStateBAL to avoid redundant copy - 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) - } + preCalPostStart := time.Now() + statedb.PreComputePostState(block.NumberU64(), runtime.NumCPU()/2) + PrefetchMergeBALTime += time.Since(preCalPostStart) // Apply pre-execution system calls. var tracingStateDB = vm.StateDB(statedb) @@ -80,43 +70,9 @@ func (p *ParallelStateProcessor) Process(block *types.Block, statedb *state.Stat ProcessParentBlockHash(block.ParentHash(), evm) } - wg.Add(2) - go func() { - defer wg.Done() - - start := time.Now() - postState = statedb.CopyState() - // Stop prefetcher cause we'll directly fetch tries in parallel - postState.StopPrefetcher() - - postState.MergePostBalStates() - PostMergeTime += time.Since(start) - - // Prewarm the updating trie - start = time.Now() - postState.PrefetchTrie() - PrefetchTrieTimer += time.Since(start) - }() - - go func() { - defer wg.Done() - - preCalPostStart := time.Now() - statedb.PreComputePostState(block.NumberU64(), runtime.NumCPU()/2) - PrefetchMergeBALTime += time.Since(preCalPostStart) - - 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 - } + exeStart := time.Now() + result, err = p.executeParallel(block, statedb, cfg, gp, signer, context, initialdb) + ParallelExeTime += time.Since(exeStart) return result, err } @@ -135,15 +91,16 @@ func (p *ParallelStateProcessor) executeParallel(block *types.Block, statedb *st ) // leave some cpus for prefetching - workers.SetLimit(runtime.NumCPU() / 2) switch preStateType { case BALPreState: { + workers.SetLimit(runtime.NumCPU() / 2) } case SeqPreState: // must set workers limit = 1 { + workers.SetLimit(1) preStatedb := statedb.Copy() gpcp := *gp preStateProvider = &SequentialPrestateProvider{ @@ -297,7 +254,7 @@ func ApplyTransactionWithParallelEVM(msg *Message, gp *GasPool, statedb *state.S // Merge the tx-local access event into the "block-local" one, in order to collect // all values, so that the witness can be built. - if statedb.GetTrie().IsVerkle() { + if statedb.GetTrie() != nil && statedb.GetTrie().IsVerkle() { statedb.AccessEvents().Merge(evm.AccessEvents) } diff --git a/core/state/statedb.go b/core/state/statedb.go index b37a980a2c..8badaa3940 100644 --- a/core/state/statedb.go +++ b/core/state/statedb.go @@ -246,6 +246,23 @@ func New(root common.Hash, db Database) (*StateDB, error) { return NewWithReader(root, db, reader) } +// Empty statedb without tr +func NewEmptyDB() *StateDB { + sdb := &StateDB{ + stateObjects: make(map[common.Address]*stateObject), + stateObjectsDestruct: make(map[common.Address]*stateObject), + 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(), + postStates: make(map[int]*StateDB), + } + return sdb +} + // NewWithReader creates a new state for the specified state root. Unlike New, // this function accepts an additional Reader which is bound to the given root. func NewWithReader(root common.Hash, db Database, reader Reader) (*StateDB, error) { @@ -357,7 +374,7 @@ func (s *StateDB) PrefetchStateBAL(blockNumber uint64) { } func (s *StateDB) prefetchBalPreblockKeys() { - log.Info("PrefetchBalPreblockKeys...") + log.Info("PrefetchBalPreblockKeys for:", "block:", s.blockNumber) type StorageKV struct { addr *common.Address key *common.Hash @@ -370,12 +387,17 @@ func (s *StateDB) prefetchBalPreblockKeys() { workers errgroup.Group ) - workers.SetLimit(runtime.NumCPU() - 2) + workers.SetLimit(runtime.NumCPU() / 2) // Fetch pre-block acccount state accounts := make(chan *stateObject, len(preBal)) + lenAcctCh := 0 for _, acl := range preBal { addr := acl.Address lenMaxSlots += len(acl.StorageKeys) + if s.stateObjects[addr] != nil { // Skip already fetched pre-block bals or newObject will override existing storage! + continue + } + lenAcctCh++ workers.Go(func() error { acctBal, err := s.reader.Account(addr) @@ -393,9 +415,8 @@ func (s *StateDB) prefetchBalPreblockKeys() { return nil }) } - workers.Wait() - for range len(preBal) { + for range lenAcctCh { obj := <-accounts // must set it first to avoid accounts read later in storage fetching if obj != nil { @@ -404,7 +425,6 @@ func (s *StateDB) prefetchBalPreblockKeys() { } close(accounts) - // Fetch pre-block storage state storages := make(chan *StorageKV, lenMaxSlots) for _, acl := range preBal { @@ -417,9 +437,13 @@ func (s *StateDB) prefetchBalPreblockKeys() { continue } - lenSlots += len(acl.StorageKeys) - for _, key := range keys { + // Skip already fetched slots to simulate merged N-blocks BALs + if _, cached := obj.originStorage[key]; cached { + continue + } + lenSlots++ + workers.Go(func() error { val, err := s.reader.Storage(addr, key) kv := &StorageKV{&addr, &key, &val} @@ -433,7 +457,6 @@ func (s *StateDB) prefetchBalPreblockKeys() { }) } } - workers.Wait() for range lenSlots { kv := <-storages @@ -443,7 +466,6 @@ func (s *StateDB) prefetchBalPreblockKeys() { s.setStateObject(account) } } - close(storages) } @@ -683,16 +705,17 @@ func (s *StateDB) MergeState(entries []JournalEntry) { } } -func (s *StateDB) MergePostBalStates() { +func (s *StateDB) MergePostBalStates(blockNumber uint64) { if balType != BalPreblockKeysPostValues { panic("MergePostBal is only supported with BalPreblockKeysPostValues") } var ( - postBal = AllBlockTxPostValues[s.blockNumber] + postBal = AllBlockTxPostValues[blockNumber] ) + s.blockNumber = blockNumber - // -1 and len(postBal)-1 (not used) are syscalls pre and post all txs. - for txIndex := -1; txIndex < len(postBal)-1; txIndex++ { + // -1 and len(postBal)-1 (used for N-blocks) are syscalls pre and post all txs. + for txIndex := -1; txIndex < len(postBal); txIndex++ { postVals := postBal[txIndex] for addr, acct := range postVals { @@ -1241,23 +1264,26 @@ func (s *StateDB) Copy() *StateDB { } func (s *StateDB) CopyState() *StateDB { - // Copy all the basic fields, initialize the memory ones + // Only copy all the basic fields for N-blocks state, initialize the memory ones state := &StateDB{ - db: s.db, - prefetcher: s.prefetcher, + db: s.db, + // prefetcher: s.prefetcher, // don't need prefetcher for BAL 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, len(s.mutations)), - dbErr: s.dbErr, - refund: s.refund, - thash: s.thash, - txIndex: s.txIndex, - logs: make(map[common.Hash][]*types.Log, len(s.logs)), - logSize: s.logSize, - preimages: maps.Clone(s.preimages), + // mutations: make(map[common.Address]*mutation, len(s.mutations)), + mutations: make(map[common.Address]*mutation), + dbErr: s.dbErr, + refund: s.refund, + thash: s.thash, + txIndex: s.txIndex, + // logs: make(map[common.Hash][]*types.Log, len(s.logs)), + logs: make(map[common.Hash][]*types.Log), + + logSize: s.logSize, + // preimages: maps.Clone(s.preimages), // Do we need to copy the access list and transient storage? // In practice: No. At the start of a transaction, these two lists are empty. @@ -1265,9 +1291,12 @@ func (s *StateDB) CopyState() *StateDB { // in the middle of a transaction. However, it doesn't cost us much to copy // empty lists, so we do it anyway to not blow up if we ever decide copy them // in the middle of a transaction. - accessList: s.accessList.Copy(), - transientStorage: s.transientStorage.Copy(), - journal: s.journal.copy(), + // accessList: s.accessList.Copy(), + // transientStorage: s.transientStorage.Copy(), + // journal: s.journal.copy(), + accessList: newAccessList(), + transientStorage: newTransientStorage(), + journal: newJournal(), // The update journal is not copies to avoid duplicated updates in later transactions. blockNumber: s.blockNumber, updateJournal: newJournal(), @@ -1277,9 +1306,9 @@ func (s *StateDB) CopyState() *StateDB { if s.witness != nil { state.witness = s.witness.Copy() } - if s.accessEvents != nil { - state.accessEvents = s.accessEvents.Copy() - } + // if s.accessEvents != nil { + // state.accessEvents = s.accessEvents.Copy() + // } // Deep copy cached state objects. for addr, obj := range s.stateObjects { state.stateObjects[addr] = obj.simpleCopy(state) @@ -1289,18 +1318,19 @@ func (s *StateDB) CopyState() *StateDB { state.stateObjectsDestruct[addr] = obj.simpleCopy(state) } // Deep copy the object state markers. - for addr, op := range s.mutations { - state.mutations[addr] = op.copy() - } + // 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 - } + // 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 } diff --git a/core/state/statedb_readerbal.go b/core/state/statedb_readerbal.go index 37845e918f..263e37024a 100644 --- a/core/state/statedb_readerbal.go +++ b/core/state/statedb_readerbal.go @@ -28,6 +28,13 @@ 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 { + // Must use dirtyStorage and pendingStorage first, because statedb.MergePostBalStates will modify originStorage during N-blocks scenerio. + if value, cached := stateObject.dirtyStorage[slot]; cached { + return value, nil + } + if value, cached := stateObject.pendingStorage[slot]; cached { + return value, nil + } if value, cached := stateObject.originStorage[slot]; cached { return value, nil } diff --git a/core/state_processor.go b/core/state_processor.go index ee98326467..a3ed60fe47 100644 --- a/core/state_processor.go +++ b/core/state_processor.go @@ -161,7 +161,7 @@ func ApplyTransactionWithEVM(msg *Message, gp *GasPool, statedb *state.StateDB, // Merge the tx-local access event into the "block-local" one, in order to collect // all values, so that the witness can be built. - if statedb.Database().TrieDB().IsVerkle() { + if statedb.Database() != nil && statedb.Database().TrieDB().IsVerkle() { statedb.AccessEvents().Merge(evm.AccessEvents) } return MakeReceipt(evm, result, statedb, blockNumber, blockHash, blockTime, tx, *usedGas, root), nil