forks/go-ethereum
1
0
Fork 1
mirror of https://github.com/ethereum/go-ethereum.git synced 2026-06-19 21:31:37 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 1

core: prefetch next block state concurrently #19328

Browse source
This commit is contained in:
Daniel Liu 2025-04-16 16:46:15 +08:00 committed by Daniel Liu
parent b628212d8e
commit 1d3e5cb455
8 changed files with 209 additions and 58 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

1
cmd/XDC/main.go
View file

@ -98,6 +98,7 @@ var (
utils.CacheDatabaseFlag,
utils.CacheTrieFlag,
utils.CacheGCFlag,
utils.CacheNoPrefetchFlag,
//utils.TrieCacheGenFlag,
utils.CacheLogSizeFlag,
utils.FDLimitFlag,

15
cmd/utils/flags.go
View file

@ -302,6 +302,11 @@ var (
Value: 25,
Category: flags.PerfCategory,
}
CacheNoPrefetchFlag = &cli.BoolFlag{
Name: "cache.noprefetch",
Usage: "Disable heuristic state prefetch during block import (less CPU and disk IO, more time waiting for data)",
Category: flags.PerfCategory,
}
CacheLogSizeFlag = &cli.IntFlag{
Name: "cache-blocklogs",
Aliases: []string{"cache.blocklogs"},
@ -1474,6 +1479,7 @@ func SetEthConfig(ctx *cli.Context, stack *node.Node, cfg *ethconfig.Config) {
Fatalf("--%s must be either 'full' or 'archive'", GCModeFlag.Name)
}
cfg.NoPruning = ctx.String(GCModeFlag.Name) == "archive"
cfg.NoPrefetch = ctx.Bool(CacheNoPrefetchFlag.Name)
if ctx.IsSet(CacheFlag.Name) || ctx.IsSet(CacheTrieFlag.Name) {
cfg.TrieCleanCache = ctx.Int(CacheFlag.Name) * ctx.Int(CacheTrieFlag.Name) / 100
@ -1736,10 +1742,11 @@ func MakeChain(ctx *cli.Context, stack *node.Node, readonly bool) (chain *core.B
Fatalf("--%s must be either 'full' or 'archive'", GCModeFlag.Name)
}
cache := &core.CacheConfig{
Disabled: ctx.String(GCModeFlag.Name) == "archive",
TrieCleanLimit: ethconfig.Defaults.TrieCleanCache,
TrieDirtyLimit: ethconfig.Defaults.TrieDirtyCache,
TrieTimeLimit: ethconfig.Defaults.TrieTimeout,
TrieCleanLimit: ethconfig.Defaults.TrieCleanCache,
TrieCleanNoPrefetch: ctx.Bool(CacheNoPrefetchFlag.Name),
TrieDirtyLimit: ethconfig.Defaults.TrieDirtyCache,
TrieDirtyDisabled: ctx.String(GCModeFlag.Name) == "archive",
TrieTimeLimit: ethconfig.Defaults.TrieTimeout,
}
if ctx.IsSet(CacheFlag.Name) || ctx.IsSet(CacheTrieFlag.Name) {
cache.TrieCleanLimit = ctx.Int(CacheFlag.Name) * ctx.Int(CacheTrieFlag.Name) / 100

82
core/blockchain.go
View file

@ -81,6 +81,9 @@ var (
blockReorgAddMeter = metrics.NewRegisteredMeter("chain/reorg/add", nil)
blockReorgDropMeter = metrics.NewRegisteredMeter("chain/reorg/drop", nil)
blockPrefetchExecuteTimer = metrics.NewRegisteredTimer("chain/prefetch/executes", nil)
blockPrefetchInterruptMeter = metrics.NewRegisteredMeter("chain/prefetch/interrupts", nil)
errInsertionInterrupted = errors.New("insertion is interrupted")
errChainStopped = errors.New("blockchain is stopped")
errInvalidOldChain = errors.New("invalid old chain")
@ -127,10 +130,11 @@ const (
// CacheConfig contains the configuration values for the trie caching/pruning
// that's resident in a blockchain.
type CacheConfig struct {
Disabled bool // Whether to disable trie write caching (archive node)
TrieCleanLimit int // Memory allowance (MB) to use for caching trie nodes in memory
TrieDirtyLimit int // Memory limit (MB) at which to start flushing dirty trie nodes to disk
TrieTimeLimit time.Duration // Time limit after which to flush the current in-memory trie to disk
TrieCleanLimit int // Memory allowance (MB) to use for caching trie nodes in memory
TrieCleanNoPrefetch bool // Whether to disable heuristic state prefetching for followup blocks
TrieDirtyLimit int // Memory limit (MB) at which to start flushing dirty trie nodes to disk
TrieDirtyDisabled bool // Whether to disable trie write caching and GC altogether (archive node)
TrieTimeLimit time.Duration // Time limit after which to flush the current in-memory trie to disk
}
type ResultProcessBlock struct {
@ -179,8 +183,6 @@ type BlockChain struct {
// Readers don't need to take it, they can just read the database.
chainmu *syncx.ClosableMutex
procmu sync.RWMutex // block processor lock
currentBlock atomic.Value // Current head of the block chain
currentFastBlock atomic.Value // Current head of the fast-sync chain (may be above the block chain!)
@ -203,10 +205,11 @@ type BlockChain struct {
running int32 // 0 if chain is running, 1 when stopped
procInterrupt int32 // interrupt signaler for block processing
engine consensus.Engine
processor Processor // block processor interface
validator Validator // block and state validator interface
vmConfig vm.Config
engine consensus.Engine
validator Validator // Block and state validator interface
prefetcher Prefetcher // Block state prefetcher interface
processor Processor // Block transaction processor interface
vmConfig vm.Config
IPCEndpoint string
Client bind.ContractBackend // Global ipc client instance.
@ -260,8 +263,9 @@ func NewBlockChain(db ethdb.Database, cacheConfig *CacheConfig, chainConfig *par
rejectedLendingItem: lru.NewCache[common.Hash, interface{}](tradingstate.OrderCacheLimit),
finalizedTrade: lru.NewCache[common.Hash, interface{}](tradingstate.OrderCacheLimit),
}
bc.SetValidator(NewBlockValidator(chainConfig, bc, engine))
bc.SetProcessor(NewStateProcessor(chainConfig, bc, engine))
bc.validator = NewBlockValidator(chainConfig, bc, engine)
bc.prefetcher = newStatePrefetcher(chainConfig, bc, engine)
bc.processor = NewStateProcessor(chainConfig, bc, engine)
var err error
bc.hc, err = NewHeaderChain(db, chainConfig, engine, bc.insertStopped)
@ -554,31 +558,13 @@ func (bc *BlockChain) CurrentFastBlock() *types.Block {
return bc.currentFastBlock.Load().(*types.Block)
}
// SetProcessor sets the processor required for making state modifications.
func (bc *BlockChain) SetProcessor(processor Processor) {
bc.procmu.Lock()
defer bc.procmu.Unlock()
bc.processor = processor
}
// SetValidator sets the validator which is used to validate incoming blocks.
func (bc *BlockChain) SetValidator(validator Validator) {
bc.procmu.Lock()
defer bc.procmu.Unlock()
bc.validator = validator
}
// Validator returns the current validator.
func (bc *BlockChain) Validator() Validator {
bc.procmu.RLock()
defer bc.procmu.RUnlock()
return bc.validator
}
// Processor returns the current processor.
func (bc *BlockChain) Processor() Processor {
bc.procmu.RLock()
defer bc.procmu.RUnlock()
return bc.processor
}
@ -1045,7 +1031,7 @@ func (bc *BlockChain) saveData() {
// - HEAD: So we don't need to reprocess any blocks in the general case
// - HEAD-1: So we don't do large reorgs if our HEAD becomes an uncle
// - HEAD-127: So we have a hard limit on the number of blocks reexecuted
if !bc.cacheConfig.Disabled {
if !bc.cacheConfig.TrieDirtyDisabled {
var tradingTriedb *trie.Database
var lendingTriedb *trie.Database
engine, _ := bc.Engine().(*XDPoS.XDPoS)
@ -1426,7 +1412,7 @@ func (bc *BlockChain) writeBlockWithState(block *types.Block, receipts []*types.
}
// If we're running an archive node, always flush
if bc.cacheConfig.Disabled {
if bc.cacheConfig.TrieDirtyDisabled {
if err := triedb.Commit(root, false); err != nil {
return NonStatTy, err
}
@ -1666,7 +1652,7 @@ func (bc *BlockChain) insertChain(chain types.Blocks, verifySeals bool) (int, []
defer close(abort)
// Peek the error for the first block to decide the directing import logic
it := newInsertIterator(chain, results, bc.Validator())
it := newInsertIterator(chain, results, bc.validator)
block, err := it.next()
switch {
@ -1733,12 +1719,31 @@ func (bc *BlockChain) insertChain(chain types.Blocks, verifySeals bool) (int, []
if err != nil {
return it.index, events, coalescedLogs, err
}
// If we have a followup block, run that against the current state to pre-cache
// transactions and probabilistically some of the account/storage trie nodes.
var followupInterrupt uint32
if !bc.cacheConfig.TrieCleanNoPrefetch {
if followup, err := it.peek(); followup != nil && err == nil {
go func(start time.Time) {
throwaway, _ := state.New(parent.Root, bc.stateCache)
bc.prefetcher.Prefetch(followup, throwaway, bc.vmConfig, &followupInterrupt)
blockPrefetchExecuteTimer.Update(time.Since(start))
if atomic.LoadUint32(&followupInterrupt) == 1 {
blockPrefetchInterruptMeter.Mark(1)
}
}(time.Now())
}
}
// Process block using the parent state as reference point.
t0 := time.Now()
isTIPXDCXReceiver := bc.Config().IsTIPXDCXReceiver(block.Number())
tradingState, lendingState, err := bc.processTradingAndLendingStates(isTIPXDCXReceiver, block, parent, statedb)
if err != nil {
bc.reportBlock(block, nil, err)
atomic.StoreUint32(&followupInterrupt, 1)
return it.index, events, coalescedLogs, err
}
feeCapacity := state.GetTRC21FeeCapacityFromStateWithCache(parent.Root, statedb)
@ -1746,10 +1751,11 @@ func (bc *BlockChain) insertChain(chain types.Blocks, verifySeals bool) (int, []
t1 := time.Now()
if err != nil {
bc.reportBlock(block, receipts, err)
atomic.StoreUint32(&followupInterrupt, 1)
return it.index, events, coalescedLogs, err
}
// Validate the state using the default validator
err = bc.Validator().ValidateState(block, statedb, receipts, usedGas)
err = bc.validator.ValidateState(block, statedb, receipts, usedGas)
if err != nil {
bc.reportBlock(block, receipts, err)
return it.index, events, coalescedLogs, err
@ -1761,8 +1767,10 @@ func (bc *BlockChain) insertChain(chain types.Blocks, verifySeals bool) (int, []
status, err := bc.writeBlockWithState(block, receipts, statedb, tradingState, lendingState)
t3 := time.Now()
if err != nil {
atomic.StoreUint32(&followupInterrupt, 1)
return it.index, events, coalescedLogs, err
}
atomic.StoreUint32(&followupInterrupt, 1)
// Update the metrics subsystem with all the measurements
accountReadTimer.Update(statedb.AccountReads)
@ -2021,7 +2029,7 @@ func (bc *BlockChain) getResultBlock(block *types.Block, verifiedM2 bool) (*Resu
}
// Wait for the block's verification to complete
bstart := time.Now()
err := bc.Validator().ValidateBody(block)
err := bc.validator.ValidateBody(block)
switch {
case err == ErrKnownBlock:
// Block and state both already known. However if the current block is below
@ -2953,7 +2961,7 @@ func (bc *BlockChain) processTradingAndLendingStates(isValidBlockNumber bool, bl
}
for _, txMatchBatch := range txMatchBatchData {
log.Debug("Verify matching transaction", "txHash", txMatchBatch.TxHash.Hex())
err := bc.Validator().ValidateTradingOrder(statedb, tradingState, txMatchBatch, author, block.Header())
err := bc.validator.ValidateTradingOrder(statedb, tradingState, txMatchBatch, author, block.Header())
if err != nil {
return tradingState, lendingState, err
}
@ -2964,7 +2972,7 @@ func (bc *BlockChain) processTradingAndLendingStates(isValidBlockNumber bool, bl
}
for _, batch := range batches {
log.Debug("Verify matching transaction", "txHash", batch.TxHash.Hex())
err := bc.Validator().ValidateLendingOrder(statedb, lendingState, tradingState, batch, author, block.Header())
err := bc.validator.ValidateLendingOrder(statedb, lendingState, tradingState, batch, author, block.Header())
if err != nil {
return tradingState, lendingState, err
}

45
core/blockchain_insert.go
View file

@ -63,7 +63,7 @@ func (st *insertStats) report(chain []*types.Block, index int, cache common.Stor
if timestamp := time.Unix(end.Time().Int64(), 0); time.Since(timestamp) > time.Minute {
context = append(context, []interface{}{"age", common.PrettyAge(timestamp)}...)
}
context = append(context, []interface{}{"cache", cache}...)
context = append(context, []interface{}{"dirty", cache}...)
if st.queued > 0 {
context = append(context, []interface{}{"queued", st.queued}...)
@ -80,10 +80,13 @@ func (st *insertStats) report(chain []*types.Block, index int, cache common.Stor
// insertIterator is a helper to assist during chain import.
type insertIterator struct {
chain types.Blocks
results <-chan error
index int
validator Validator
chain types.Blocks // Chain of blocks being iterated over
results <-chan error // Verification result sink from the consensus engine
errors []error // Header verification errors for the blocks
index int // Current offset of the iterator
validator Validator // Validator to run if verification succeeds
}
// newInsertIterator creates a new iterator based on the given blocks, which are
@ -92,6 +95,7 @@ func newInsertIterator(chain types.Blocks, results <-chan error, validator Valid
return &insertIterator{
chain: chain,
results: results,
errors: make([]error, 0, len(chain)),
index: -1,
validator: validator,
}
@ -100,17 +104,44 @@ func newInsertIterator(chain types.Blocks, results <-chan error, validator Valid
// next returns the next block in the iterator, along with any potential validation
// error for that block. When the end is reached, it will return (nil, nil).
func (it *insertIterator) next() (*types.Block, error) {
// If we reached the end of the chain, abort
if it.index+1 >= len(it.chain) {
it.index = len(it.chain)
return nil, nil
}
// Advance the iterator and wait for verification result if not yet done
it.index++
if err := <-it.results; err != nil {
return it.chain[it.index], err
if len(it.errors) <= it.index {
it.errors = append(it.errors, <-it.results)
}
if it.errors[it.index] != nil {
return it.chain[it.index], it.errors[it.index]
}
// Block header valid, run body validation and return
return it.chain[it.index], it.validator.ValidateBody(it.chain[it.index])
}
// peek returns the next block in the iterator, along with any potential validation
// error for that block, but does **not** advance the iterator.
//
// Both header and body validation errors (nil too) is cached into the iterator
// to avoid duplicating work on the following next() call.
func (it *insertIterator) peek() (*types.Block, error) {
// If we reached the end of the chain, abort
if it.index+1 >= len(it.chain) {
return nil, nil
}
// Wait for verification result if not yet done
if len(it.errors) <= it.index+1 {
it.errors = append(it.errors, <-it.results)
}
if it.errors[it.index+1] != nil {
return it.chain[it.index+1], it.errors[it.index+1]
}
// Block header valid, ignore body validation since we don't have a parent anyway
return it.chain[it.index+1], nil
}
// previous returns the previous header that was being processed, or nil.
func (it *insertIterator) previous() *types.Header {
if it.index < 1 {

95
core/state_prefetcher.go Normal file
View file

@ -0,0 +1,95 @@
// Copyright 2019 The go-ethereum Authors
// This file is part of the go-ethereum library.
//
// The go-ethereum library is free software: you can redistribute it and/or modify
// it under the terms of the GNU Lesser General Public License as published by
// the Free Software Foundation, either version 3 of the License, or
// (at your option) any later version.
//
// The go-ethereum library is distributed in the hope that it will be useful,
// but WITHOUT ANY WARRANTY; without even the implied warranty of
// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
// GNU Lesser General Public License for more details.
//
// You should have received a copy of the GNU Lesser General Public License
// along with the go-ethereum library. If not, see <http://www.gnu.org/licenses/>.
package core
import (
"sync/atomic"
"github.com/XinFinOrg/XDPoSChain/common"
"github.com/XinFinOrg/XDPoSChain/consensus"
"github.com/XinFinOrg/XDPoSChain/core/state"
"github.com/XinFinOrg/XDPoSChain/core/types"
"github.com/XinFinOrg/XDPoSChain/core/vm"
"github.com/XinFinOrg/XDPoSChain/params"
)
// statePrefetcher is a basic Prefetcher, which blindly executes a block on top
// of an arbitrary state with the goal of prefetching potentially useful state
// data from disk before the main block processor start executing.
type statePrefetcher struct {
config *params.ChainConfig // Chain configuration options
bc *BlockChain // Canonical block chain
engine consensus.Engine // Consensus engine used for block rewards
}
// newStatePrefetcher initialises a new statePrefetcher.
func newStatePrefetcher(config *params.ChainConfig, bc *BlockChain, engine consensus.Engine) *statePrefetcher {
return &statePrefetcher{
config: config,
bc: bc,
engine: engine,
}
}
// Prefetch processes the state changes according to the Ethereum rules by running
// the transaction messages using the statedb, but any changes are discarded. The
// only goal is to pre-cache transaction signatures and state trie nodes.
func (p *statePrefetcher) Prefetch(block *types.Block, statedb *state.StateDB, cfg vm.Config, interrupt *uint32) {
var (
header = block.Header()
gaspool = new(GasPool).AddGas(block.GasLimit())
blockContext = NewEVMBlockContext(header, p.bc, nil)
evm = vm.NewEVM(blockContext, vm.TxContext{}, statedb, nil, p.config, cfg)
signer = types.MakeSigner(p.config, header.Number)
)
// Iterate over and process the individual transactions
byzantium := p.config.IsByzantium(block.Number())
for i, tx := range block.Transactions() {
// If block precaching was interrupted, abort
if interrupt != nil && atomic.LoadUint32(interrupt) == 1 {
return
}
// Convert the transaction into an executable message and pre-cache its sender
msg, err := tx.AsMessage(signer, nil, nil, header.BaseFee)
if err != nil {
return // Also invalid block, bail out
}
statedb.SetTxContext(tx.Hash(), i)
if err := precacheTransaction(msg, p.config, gaspool, statedb, header, evm); err != nil {
return // Ugh, something went horribly wrong, bail out
}
// If we're pre-byzantium, pre-load trie nodes for the intermediate root
if !byzantium {
statedb.IntermediateRoot(true)
}
}
// If were post-byzantium, pre-load trie nodes for the final root hash
if byzantium {
statedb.IntermediateRoot(true)
}
}
// precacheTransaction attempts to apply a transaction to the given state database
// and uses the input parameters for its environment. The goal is not to execute
// the transaction successfully, rather to warm up touched data slots.
func precacheTransaction(msg types.Message, config *params.ChainConfig, gaspool *GasPool, statedb *state.StateDB, header *types.Header, evm *vm.EVM) error {
// Update the evm with the new transaction context.
evm.Reset(NewEVMTxContext(msg), statedb)
// Add addresses to access list if applicable
_, err := ApplyMessage(evm, msg, gaspool, common.Address{})
return err
}

16
core/types.go
View file

@ -43,13 +43,19 @@ type Validator interface {
ValidateLendingOrder(statedb *state.StateDB, lendingStateDb *lendingstate.LendingStateDB, XDCxStatedb *tradingstate.TradingStateDB, batch lendingstate.TxLendingBatch, coinbase common.Address, header *types.Header) error
}
// Prefetcher is an interface for pre-caching transaction signatures and state.
type Prefetcher interface {
// Prefetch processes the state changes according to the Ethereum rules by running
// the transaction messages using the statedb, but any changes are discarded. The
// only goal is to pre-cache transaction signatures and state trie nodes.
Prefetch(block *types.Block, statedb *state.StateDB, cfg vm.Config, interrupt *uint32)
}
// Processor is an interface for processing blocks using a given initial state.
//
// Process takes the block to be processed and the statedb upon which the
// initial state is based. It should return the receipts generated, amount
// of gas used in the process and return an error if any of the internal rules
// failed.
type Processor interface {
// Process processes the state changes according to the Ethereum rules by running
// the transaction messages using the statedb and applying any rewards to both
// the processor (coinbase) and any included uncles.
Process(block *types.Block, statedb *state.StateDB, tradingState *tradingstate.TradingStateDB, cfg vm.Config, balanceFee map[common.Address]*big.Int) (types.Receipts, []*types.Log, uint64, error)
ProcessBlockNoValidator(block *CalculatedBlock, statedb *state.StateDB, tradingState *tradingstate.TradingStateDB, cfg vm.Config, balanceFee map[common.Address]*big.Int) (types.Receipts, []*types.Log, uint64, error)

9
eth/backend.go
View file

@ -176,10 +176,11 @@ func New(stack *node.Node, config *ethconfig.Config, XDCXServ *XDCx.XDCX, lendin
var (
vmConfig = vm.Config{EnablePreimageRecording: config.EnablePreimageRecording}
cacheConfig = &core.CacheConfig{
Disabled: config.NoPruning,
TrieCleanLimit: config.TrieCleanCache,
TrieDirtyLimit: config.TrieDirtyCache,
TrieTimeLimit: config.TrieTimeout,
TrieCleanLimit: config.TrieCleanCache,
TrieCleanNoPrefetch: config.NoPrefetch,
TrieDirtyLimit: config.TrieDirtyCache,
TrieDirtyDisabled: config.NoPruning,
TrieTimeLimit: config.TrieTimeout,
}
)
if eth.chainConfig.XDPoS != nil {

4
eth/ethconfig/config.go
View file

@ -103,7 +103,9 @@ type Config struct {
// zero, the chain ID is used as network ID.
NetworkId uint64
SyncMode downloader.SyncMode
NoPruning bool
NoPruning bool // Whether to disable pruning and flush everything to disk
NoPrefetch bool // Whether to disable prefetching and only load state on demand
// Light client options
LightServ int `toml:",omitempty"` // Maximum percentage of time allowed for serving LES requests