From ec1616d6e00283e3713e5cec91d5083df161f8b5 Mon Sep 17 00:00:00 2001 From: Gary Rong Date: Sun, 4 May 2025 15:36:56 +0800 Subject: [PATCH] core, eth, internal: split txIndexProgress from the main loop --- core/blockchain_reader.go | 9 ++- core/txindexer.go | 71 +++++++++++------------- core/txindexer_test.go | 20 +++---- eth/api_backend.go | 6 +- eth/downloader/api.go | 2 +- eth/tracers/api.go | 4 +- eth/tracers/api_test.go | 2 +- ethclient/ethclient_test.go | 2 +- internal/ethapi/api.go | 8 +-- internal/ethapi/api_test.go | 2 +- internal/ethapi/backend.go | 2 +- internal/ethapi/transaction_args_test.go | 2 +- 12 files changed, 57 insertions(+), 73 deletions(-) diff --git a/core/blockchain_reader.go b/core/blockchain_reader.go index 7bfdd692fe..fefeb37542 100644 --- a/core/blockchain_reader.go +++ b/core/blockchain_reader.go @@ -17,7 +17,6 @@ package core import ( - "context" "errors" "github.com/ethereum/go-ethereum/common" @@ -299,8 +298,8 @@ func (bc *BlockChain) GetTransactionLookup(hash common.Hash) (*rawdb.LegacyTxLoo } // TxIndexDone returns true if the transaction indexer has finished indexing. -func (bc *BlockChain) TxIndexDone(ctx context.Context) bool { - progress, err := bc.TxIndexProgress(ctx) +func (bc *BlockChain) TxIndexDone() bool { + progress, err := bc.TxIndexProgress() if err != nil { // No error is returned if the transaction indexing progress is unreachable // due to unexpected internal errors. In such cases, it is impossible to @@ -403,11 +402,11 @@ func (bc *BlockChain) GetVMConfig() *vm.Config { } // TxIndexProgress returns the transaction indexing progress. -func (bc *BlockChain) TxIndexProgress(ctx context.Context) (TxIndexProgress, error) { +func (bc *BlockChain) TxIndexProgress() (TxIndexProgress, error) { if bc.txIndexer == nil { return TxIndexProgress{}, errors.New("tx indexer is not enabled") } - return bc.txIndexer.txIndexProgress(ctx) + return bc.txIndexer.txIndexProgress(), nil } // HistoryPruningCutoff returns the configured history pruning point. diff --git a/core/txindexer.go b/core/txindexer.go index 135b56e777..3c448b33fc 100644 --- a/core/txindexer.go +++ b/core/txindexer.go @@ -17,9 +17,8 @@ package core import ( - "context" - "errors" "fmt" + "sync/atomic" "github.com/ethereum/go-ethereum/common" "github.com/ethereum/go-ethereum/core/rawdb" @@ -48,26 +47,35 @@ type txIndexer struct { // and all others shouldn't. limit uint64 + // The current tail of the indexed transactions, null indicates + // that no transactions have been indexed yet. + // + // This field is accessed by both the indexer and the indexing + // progress queries. + tail atomic.Pointer[uint64] + // cutoff denotes the block number before which the chain segment should // be pruned and not available locally. - cutoff uint64 - db ethdb.Database - progress chan chan TxIndexProgress - term chan chan struct{} - closed chan struct{} + cutoff uint64 + chain *BlockChain + db ethdb.Database + term chan chan struct{} + closed chan struct{} } // newTxIndexer initializes the transaction indexer. func newTxIndexer(limit uint64, chain *BlockChain) *txIndexer { cutoff, _ := chain.HistoryPruningCutoff() indexer := &txIndexer{ - limit: limit, - cutoff: cutoff, - db: chain.db, - progress: make(chan chan TxIndexProgress), - term: make(chan chan struct{}), - closed: make(chan struct{}), + limit: limit, + cutoff: cutoff, + chain: chain, + db: chain.db, + term: make(chan chan struct{}), + closed: make(chan struct{}), } + indexer.tail.Store(rawdb.ReadTxIndexTail(chain.db)) + go indexer.loop(chain) var msg string @@ -217,10 +225,9 @@ func (indexer *txIndexer) loop(chain *BlockChain) { // Listening to chain events and manipulate the transaction indexes. var ( - stop chan struct{} // Non-nil if background routine is active - done chan struct{} // Non-nil if background routine is active - head = indexer.resolveHead() // The latest announced chain head - lastTail = rawdb.ReadTxIndexTail(indexer.db) // The oldest indexed block, nil means nothing indexed + stop chan struct{} // Non-nil if background routine is active + done chan struct{} // Non-nil if background routine is active + head = indexer.resolveHead() // The latest announced chain head headCh = make(chan ChainHeadEvent) sub = chain.SubscribeChainHeadEvent(headCh) @@ -245,13 +252,10 @@ func (indexer *txIndexer) loop(chain *BlockChain) { done = make(chan struct{}) go indexer.run(h.Header.Number.Uint64(), stop, done) } - head = h.Header.Number.Uint64() case <-done: stop = nil done = nil - lastTail = rawdb.ReadTxIndexTail(indexer.db) - case ch := <-indexer.progress: - ch <- indexer.report(head, lastTail) + indexer.tail.Store(rawdb.ReadTxIndexTail(indexer.db)) case ch := <-indexer.term: if stop != nil { close(stop) @@ -302,25 +306,12 @@ func (indexer *txIndexer) report(head uint64, tail *uint64) TxIndexProgress { } } -// txIndexProgress retrieves the tx indexing progress, or an error if the -// background tx indexer is already stopped. -func (indexer *txIndexer) txIndexProgress(ctx context.Context) (TxIndexProgress, error) { - ch := make(chan TxIndexProgress, 1) - select { - case indexer.progress <- ch: - select { - case prog := <-ch: - return prog, nil - case <-ctx.Done(): - // Since the channel is buffered the loop will not block - // eventually when it prepares the response. - return TxIndexProgress{}, ctx.Err() - } - case <-indexer.closed: - return TxIndexProgress{}, errors.New("indexer is closed") - case <-ctx.Done(): - return TxIndexProgress{}, ctx.Err() - } +// txIndexProgress retrieves the transaction indexing progress. The reported +// progress may slightly lag behind the actual indexing state, as the tail is +// only updated at the end of each indexing operation. However, this delay is +// considered acceptable. +func (indexer *txIndexer) txIndexProgress() TxIndexProgress { + return indexer.report(indexer.resolveHead(), indexer.tail.Load()) } // close shutdown the indexer. Safe to be called for multiple times. diff --git a/core/txindexer_test.go b/core/txindexer_test.go index 2de704d365..6543ff429d 100644 --- a/core/txindexer_test.go +++ b/core/txindexer_test.go @@ -121,9 +121,8 @@ func TestTxIndexer(t *testing.T) { // Index the initial blocks from ancient store indexer := &txIndexer{ - limit: 0, - db: db, - progress: make(chan chan TxIndexProgress), + limit: 0, + db: db, } for i, limit := range c.limits { indexer.limit = limit @@ -241,9 +240,8 @@ func TestTxIndexerRepair(t *testing.T) { // Index the initial blocks from ancient store indexer := &txIndexer{ - limit: c.limit, - db: db, - progress: make(chan chan TxIndexProgress), + limit: c.limit, + db: db, } indexer.run(chainHead, make(chan struct{}), make(chan struct{})) @@ -432,13 +430,9 @@ func TestTxIndexerReport(t *testing.T) { // Index the initial blocks from ancient store indexer := &txIndexer{ - limit: c.limit, - cutoff: c.cutoff, - db: db, - progress: make(chan chan TxIndexProgress), - } - if c.tail != nil { - rawdb.WriteTxIndexTail(db, *c.tail) + limit: c.limit, + cutoff: c.cutoff, + db: db, } p := indexer.report(c.head, c.tail) if p.Indexed != c.expIndexed { diff --git a/eth/api_backend.go b/eth/api_backend.go index 681016e424..57f5a50837 100644 --- a/eth/api_backend.go +++ b/eth/api_backend.go @@ -361,8 +361,8 @@ func (b *EthAPIBackend) GetTransaction(txHash common.Hash) (bool, *types.Transac } // TxIndexDone returns true if the transaction indexer has finished indexing. -func (b *EthAPIBackend) TxIndexDone(ctx context.Context) bool { - return b.eth.blockchain.TxIndexDone(ctx) +func (b *EthAPIBackend) TxIndexDone() bool { + return b.eth.blockchain.TxIndexDone() } func (b *EthAPIBackend) GetPoolNonce(ctx context.Context, addr common.Address) (uint64, error) { @@ -391,7 +391,7 @@ func (b *EthAPIBackend) SubscribeNewTxsEvent(ch chan<- core.NewTxsEvent) event.S func (b *EthAPIBackend) SyncProgress(ctx context.Context) ethereum.SyncProgress { prog := b.eth.Downloader().Progress() - if txProg, err := b.eth.blockchain.TxIndexProgress(ctx); err == nil { + if txProg, err := b.eth.blockchain.TxIndexProgress(); err == nil { prog.TxIndexFinishedBlocks = txProg.Indexed prog.TxIndexRemainingBlocks = txProg.Remaining } diff --git a/eth/downloader/api.go b/eth/downloader/api.go index 850f1a10a9..ac175672a0 100644 --- a/eth/downloader/api.go +++ b/eth/downloader/api.go @@ -77,7 +77,7 @@ func (api *DownloaderAPI) eventLoop() { getProgress = func() ethereum.SyncProgress { prog := api.d.Progress() - if txProg, err := api.chain.TxIndexProgress(context.Background()); err == nil { + if txProg, err := api.chain.TxIndexProgress(); err == nil { prog.TxIndexFinishedBlocks = txProg.Indexed prog.TxIndexRemainingBlocks = txProg.Remaining } diff --git a/eth/tracers/api.go b/eth/tracers/api.go index 3c8e2c9924..17a0ad687a 100644 --- a/eth/tracers/api.go +++ b/eth/tracers/api.go @@ -83,7 +83,7 @@ type Backend interface { BlockByHash(ctx context.Context, hash common.Hash) (*types.Block, error) BlockByNumber(ctx context.Context, number rpc.BlockNumber) (*types.Block, error) GetTransaction(txHash common.Hash) (bool, *types.Transaction, common.Hash, uint64, uint64) - TxIndexDone(ctx context.Context) bool + TxIndexDone() bool RPCGasCap() uint64 ChainConfig() *params.ChainConfig Engine() consensus.Engine @@ -862,7 +862,7 @@ func (api *API) TraceTransaction(ctx context.Context, hash common.Hash, config * found, _, blockHash, blockNumber, index := api.backend.GetTransaction(hash) if !found { // Warn in case tx indexer is not done. - if !api.backend.TxIndexDone(ctx) { + if !api.backend.TxIndexDone() { return nil, ethapi.NewTxIndexingError() } // Only mined txes are supported diff --git a/eth/tracers/api_test.go b/eth/tracers/api_test.go index 6bb2416dbf..fa39187694 100644 --- a/eth/tracers/api_test.go +++ b/eth/tracers/api_test.go @@ -121,7 +121,7 @@ func (b *testBackend) GetTransaction(txHash common.Hash) (bool, *types.Transacti return tx != nil, tx, hash, blockNumber, index } -func (b *testBackend) TxIndexDone(ctx context.Context) bool { +func (b *testBackend) TxIndexDone() bool { return true } diff --git a/ethclient/ethclient_test.go b/ethclient/ethclient_test.go index a855ff59cd..29e311c1b4 100644 --- a/ethclient/ethclient_test.go +++ b/ethclient/ethclient_test.go @@ -119,7 +119,7 @@ func newTestBackend(config *node.Config) (*node.Node, []*types.Block, error) { } // Ensure the tx indexing is fully generated for ; ; time.Sleep(time.Millisecond * 100) { - progress, err := ethservice.BlockChain().TxIndexProgress(context.Background()) + progress, err := ethservice.BlockChain().TxIndexProgress() if err == nil && progress.Done() { break } diff --git a/internal/ethapi/api.go b/internal/ethapi/api.go index 945aa881e8..8f736226c7 100644 --- a/internal/ethapi/api.go +++ b/internal/ethapi/api.go @@ -1340,7 +1340,7 @@ func (api *TransactionAPI) GetTransactionByHash(ctx context.Context, hash common return NewRPCPendingTransaction(tx, api.b.CurrentHeader(), api.b.ChainConfig()), nil } // If also not in the pool there is a chance the tx indexer is still in progress. - if !api.b.TxIndexDone(ctx) { + if !api.b.TxIndexDone() { return nil, NewTxIndexingError() } // If the transaction is not found in the pool and the indexer is done, return nil @@ -1362,7 +1362,7 @@ func (api *TransactionAPI) GetRawTransactionByHash(ctx context.Context, hash com return tx.MarshalBinary() } // If also not in the pool there is a chance the tx indexer is still in progress. - if !api.b.TxIndexDone(ctx) { + if !api.b.TxIndexDone() { return nil, NewTxIndexingError() } // If the transaction is not found in the pool and the indexer is done, return nil @@ -1376,7 +1376,7 @@ func (api *TransactionAPI) GetTransactionReceipt(ctx context.Context, hash commo found, tx, blockHash, blockNumber, index := api.b.GetTransaction(hash) if !found { // Make sure indexer is done. - if !api.b.TxIndexDone(ctx) { + if !api.b.TxIndexDone() { return nil, NewTxIndexingError() } // No such tx. @@ -1786,7 +1786,7 @@ func (api *DebugAPI) GetRawTransaction(ctx context.Context, hash common.Hash) (h return tx.MarshalBinary() } // If also not in the pool there is a chance the tx indexer is still in progress. - if !api.b.TxIndexDone(ctx) { + if !api.b.TxIndexDone() { return nil, NewTxIndexingError() } // Transaction is not found in the pool and the indexer is done. diff --git a/internal/ethapi/api_test.go b/internal/ethapi/api_test.go index dc51e906de..ef799d9994 100644 --- a/internal/ethapi/api_test.go +++ b/internal/ethapi/api_test.go @@ -595,7 +595,7 @@ func (b testBackend) GetTransaction(txHash common.Hash) (bool, *types.Transactio tx, blockHash, blockNumber, index := rawdb.ReadTransaction(b.db, txHash) return true, tx, blockHash, blockNumber, index } -func (b testBackend) TxIndexDone(ctx context.Context) bool { +func (b testBackend) TxIndexDone() bool { return true } func (b testBackend) GetPoolTransactions() (types.Transactions, error) { panic("implement me") } diff --git a/internal/ethapi/backend.go b/internal/ethapi/backend.go index 9ac313b2a1..49c3a37560 100644 --- a/internal/ethapi/backend.go +++ b/internal/ethapi/backend.go @@ -75,7 +75,7 @@ type Backend interface { // Transaction pool API SendTx(ctx context.Context, signedTx *types.Transaction) error GetTransaction(txHash common.Hash) (bool, *types.Transaction, common.Hash, uint64, uint64) - TxIndexDone(ctx context.Context) bool + TxIndexDone() bool GetPoolTransactions() (types.Transactions, error) GetPoolTransaction(txHash common.Hash) *types.Transaction GetPoolNonce(ctx context.Context, addr common.Address) (uint64, error) diff --git a/internal/ethapi/transaction_args_test.go b/internal/ethapi/transaction_args_test.go index 364bd28732..9b86e452a5 100644 --- a/internal/ethapi/transaction_args_test.go +++ b/internal/ethapi/transaction_args_test.go @@ -383,7 +383,7 @@ func (b *backendMock) SendTx(ctx context.Context, signedTx *types.Transaction) e func (b *backendMock) GetTransaction(txHash common.Hash) (bool, *types.Transaction, common.Hash, uint64, uint64) { return false, nil, [32]byte{}, 0, 0 } -func (b *backendMock) TxIndexDone(ctx context.Context) bool { return true } +func (b *backendMock) TxIndexDone() bool { return true } func (b *backendMock) GetPoolTransactions() (types.Transactions, error) { return nil, nil } func (b *backendMock) GetPoolTransaction(txHash common.Hash) *types.Transaction { return nil } func (b *backendMock) GetPoolNonce(ctx context.Context, addr common.Address) (uint64, error) {