From aa21c1f2f46b86290d75f3916938bef5fa8b31c9 Mon Sep 17 00:00:00 2001 From: Gary Rong Date: Mon, 22 Jan 2024 12:08:54 +0800 Subject: [PATCH] core, eth: minor fixes --- core/blockchain.go | 28 +++++++++---------- core/blockchain_reader.go | 6 ----- eth/downloader/api.go | 57 +++++++++++++++++++-------------------- 3 files changed, 40 insertions(+), 51 deletions(-) diff --git a/core/blockchain.go b/core/blockchain.go index 764189dddc..f67f071e36 100644 --- a/core/blockchain.go +++ b/core/blockchain.go @@ -244,7 +244,6 @@ type BlockChain struct { chainHeadFeed event.Feed logsFeed event.Feed blockProcFeed event.Feed - txIndexFeed event.Feed scope event.SubscriptionScope genesisBlock *types.Block @@ -267,10 +266,12 @@ type BlockChain struct { futureBlocks *lru.Cache[common.Hash, *types.Block] wg sync.WaitGroup - quit chan struct{} // shutdown signal, closed in Stop. - stopping atomic.Bool // false if chain is running, true when stopped - procInterrupt atomic.Bool // interrupt signaler for block processing - txIndexProgCh chan chan TxIndexProgress // chan for querying the progress of transaction indexing + quit chan struct{} // shutdown signal, closed in Stop. + stopping atomic.Bool // false if chain is running, true when stopped + procInterrupt atomic.Bool // interrupt signaler for block processing + + txIndexRunning bool // flag if the background tx indexer is activated + txIndexProgCh chan chan TxIndexProgress // chan for querying the progress of transaction indexing engine consensus.Engine validator Validator // Block and state validator interface @@ -487,6 +488,7 @@ func NewBlockChain(db ethdb.Database, cacheConfig *CacheConfig, genesis *Genesis // Start tx indexer/unindexer if required. if txLookupLimit != nil { bc.txLookupLimit = *txLookupLimit + bc.txIndexRunning = true bc.wg.Add(1) go bc.maintainTxIndex() @@ -2462,7 +2464,7 @@ func (bc *BlockChain) reportTxIndexProgress(head uint64) TxIndexProgress { indexed = head - *tail + 1 } // The value of indexed might be larger than total if some blocks need - // to unindexed, avoiding a negative remaining. + // to be unindexed, avoiding a negative remaining. if indexed < total { remaining = total - indexed } @@ -2472,8 +2474,12 @@ func (bc *BlockChain) reportTxIndexProgress(head uint64) TxIndexProgress { } } -// TxIndexProgress retrieves the tx indexing progress. +// TxIndexProgress retrieves the tx indexing progress, or an error if the +// background tx indexer is not activated or already stopped. func (bc *BlockChain) TxIndexProgress() (TxIndexProgress, error) { + if !bc.txIndexRunning { + return TxIndexProgress{}, errors.New("tx indexer is not activated") + } ch := make(chan TxIndexProgress, 1) select { case bc.txIndexProgCh <- ch: @@ -2527,14 +2533,6 @@ func (bc *BlockChain) maintainTxIndex() { lastHead = head.Block.NumberU64() case <-done: done = nil - - // WARNING, the event will be fired for each signal once the - // transaction indexing is finished. Subscribers need to manage - // the event stream by themselves. It's recommended to unsubscribe - // once the event is received. - if bc.reportTxIndexProgress(lastHead).Done() { - bc.txIndexFeed.Send(true) - } case ch := <-bc.txIndexProgCh: ch <- bc.reportTxIndexProgress(lastHead) case <-bc.quit: diff --git a/core/blockchain_reader.go b/core/blockchain_reader.go index fbec721e49..0592329460 100644 --- a/core/blockchain_reader.go +++ b/core/blockchain_reader.go @@ -444,9 +444,3 @@ func (bc *BlockChain) SubscribeLogsEvent(ch chan<- []*types.Log) event.Subscript func (bc *BlockChain) SubscribeBlockProcessingEvent(ch chan<- bool) event.Subscription { return bc.scope.Track(bc.blockProcFeed.Subscribe(ch)) } - -// SubscribeTxIndexEvent registers a subscription of bool where true means -// transaction indexing has finished. -func (bc *BlockChain) SubscribeTxIndexEvent(ch chan<- bool) event.Subscription { - return bc.scope.Track(bc.txIndexFeed.Subscribe(ch)) -} diff --git a/eth/downloader/api.go b/eth/downloader/api.go index ef61a9b9f2..1e06aa9e1d 100644 --- a/eth/downloader/api.go +++ b/eth/downloader/api.go @@ -19,6 +19,7 @@ package downloader import ( "context" "sync" + "time" "github.com/ethereum/go-ethereum" "github.com/ethereum/go-ethereum/core" @@ -57,25 +58,33 @@ func NewDownloaderAPI(d *Downloader, chain *core.BlockChain, m *event.TypeMux) * // new sync subscriptions and broadcasts sync status updates to the installed sync // subscriptions. // -// The sync status pushed to subscriptions can a stream like: +// The sync status pushed to subscriptions can be a stream like: // >>> {Syncing: true, Progress: {...}} // >>> {false} // -// If the node is already synced up, then only a single event will be pushed {false}. +// If the node is already synced up, then only a single event subscribers will +// receive is {false}. func (api *DownloaderAPI) eventLoop() { var ( sub = api.mux.Subscribe(StartEvent{}) syncSubscriptions = make(map[chan interface{}]struct{}) - txIndexCh = make(chan bool, 1) - done bool - ) - txIndexSub := api.chain.SubscribeTxIndexEvent(txIndexCh) - defer func() { - if txIndexSub == nil { - return + checkInterval = time.Second * 30 + checkTimer = time.NewTimer(checkInterval) + + // status flags + started bool + done bool + + getProgress = func() ethereum.SyncProgress { + prog := api.d.Progress() + if txProg, err := api.chain.TxIndexProgress(); err == nil { + prog.TxIndexFinishedBlocks = txProg.Indexed + prog.TxIndexRemainingBlocks = txProg.Remaining + } + return prog } - txIndexSub.Unsubscribe() - }() + ) + defer checkTimer.Stop() for { select { @@ -93,11 +102,7 @@ func (api *DownloaderAPI) eventLoop() { } switch event.Data.(type) { case StartEvent: - prog := api.d.Progress() - if txProg, err := api.chain.TxIndexProgress(); err == nil { - prog.TxIndexFinishedBlocks = txProg.Indexed - prog.TxIndexRemainingBlocks = txProg.Remaining - } + prog := getProgress() notification := &SyncingResult{ Syncing: true, Status: prog, @@ -106,30 +111,22 @@ func (api *DownloaderAPI) eventLoop() { for c := range syncSubscriptions { c <- notification } + started = true } - case synced := <-txIndexCh: - if !synced { + case <-checkTimer.C: + if !started { + checkTimer.Reset(checkInterval) continue } - prog := api.d.Progress() + prog := getProgress() if !prog.Done() { - continue - } - txProg, err := api.chain.TxIndexProgress() - if err != nil || !txProg.Done() { + checkTimer.Reset(checkInterval) continue } for c := range syncSubscriptions { c <- false } done = true - - // Unsubscribe the tx indexing events as the whole - // state sync is already finished. - if txIndexSub != nil { - txIndexSub.Unsubscribe() - txIndexSub = nil - } } } }