forks/go-ethereum
1
0
Fork 1
mirror of https://github.com/ethereum/go-ethereum.git synced 2026-07-24 05:36:46 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 1

feat: N-blocks gas limit

Browse source
This commit is contained in:
Po 2025-08-08 03:58:24 +02:00
parent a94d9ee531
commit f2c21d5814
10 changed files with 367 additions and 284 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

8
cmd/geth/chaincmd.go
View file

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

2
cmd/utils/cmd.go
View file

@ -52,7 +52,7 @@ import (
)
const (
importBatchSize = 1
importBatchSize = 10
)
// ErrImportInterrupted is returned when the user interrupts the import process.

37
core/block_validator.go
View file

@ -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
}

8
core/blockchain.go
View file

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

406
core/blockchainN.go
View file

@ -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
}

2
core/blockchain_insert.go
View file

@ -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 {

67
core/parallel_state_processor.go
View file

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

112
core/state/statedb.go
View file

@ -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
}

7
core/state/statedb_readerbal.go
View file

@ -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
}

2
core/state_processor.go
View file

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