From d9daacad1723670284a63664099fbacaa53e6b92 Mon Sep 17 00:00:00 2001 From: Jerry Date: Thu, 12 Sep 2024 15:55:25 -0700 Subject: [PATCH] Address goroutine leak --- core/blockchain.go | 43 ++------------------------------ core/parallel_state_processor.go | 2 -- core/state/statedb.go | 4 +++ core/state/trie_prefetcher.go | 17 +++++++++++++ eth/handler.go | 2 -- eth/sync.go | 2 +- 6 files changed, 24 insertions(+), 46 deletions(-) diff --git a/core/blockchain.go b/core/blockchain.go index b113f3f521..7bc283397b 100644 --- a/core/blockchain.go +++ b/core/blockchain.go @@ -43,7 +43,6 @@ import ( "github.com/ethereum/go-ethereum/core/rawdb" "github.com/ethereum/go-ethereum/core/state" "github.com/ethereum/go-ethereum/core/state/snapshot" - "github.com/ethereum/go-ethereum/core/stateless" "github.com/ethereum/go-ethereum/core/tracing" "github.com/ethereum/go-ethereum/core/types" "github.com/ethereum/go-ethereum/core/vm" @@ -617,15 +616,7 @@ func (bc *BlockChain) ProcessBlock(block *types.Block, parent *types.Header) (_ processorCount++ go func() { - // todo: @anshalshukla || @cffls - check witness collection and requirement in start prefetchers - var witness *stateless.Witness - if bc.vmConfig.EnableWitnessCollection { - witness, err = stateless.NewWitness(bc, block) - if err != nil { - return - } - } - parallelStatedb.StartPrefetcher("chain", witness) + parallelStatedb.StartPrefetcher("chain", nil) receipts, logs, usedGas, err := bc.parallelProcessor.Process(block, parallelStatedb, bc.vmConfig, ctx) resultChan <- Result{receipts, logs, usedGas, err, parallelStatedb, blockExecutionParallelCounter} }() @@ -641,16 +632,7 @@ func (bc *BlockChain) ProcessBlock(block *types.Block, parent *types.Header) (_ processorCount++ go func() { - // todo: @anshalshukla || @cffls - check witness collection and requirement in start prefetchers - var witness *stateless.Witness - if bc.vmConfig.EnableWitnessCollection { - witness, err = stateless.NewWitness(bc, block) - if err != nil { - return - } - } - statedb.StartPrefetcher("chain", witness) - + statedb.StartPrefetcher("chain", nil) receipts, logs, usedGas, err := bc.processor.Process(block, statedb, bc.vmConfig, ctx) resultChan <- Result{receipts, logs, usedGas, err, statedb, blockExecutionSerialCounter} }() @@ -2319,27 +2301,6 @@ func (bc *BlockChain) insertChain(chain types.Blocks, setHead bool) (int, error) parent = bc.GetHeader(block.ParentHash(), block.NumberU64()-1) } - statedb, err := state.New(parent.Root, bc.stateCache, bc.snaps) - if err != nil { - return it.index, err - } - statedb.SetLogger(bc.logger) - - // If we are past Byzantium, enable prefetching to pull in trie node paths - // while processing transactions. Before Byzantium the prefetcher is mostly - // useless due to the intermediate root hashing after each transaction. - if bc.chainConfig.IsByzantium(block.Number()) { - var witness *stateless.Witness - if bc.vmConfig.EnableWitnessCollection { - witness, err = stateless.NewWitness(bc, block) - if err != nil { - return it.index, err - } - } - statedb.StartPrefetcher("chain", witness) - } - activeState = statedb - // 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 atomic.Bool diff --git a/core/parallel_state_processor.go b/core/parallel_state_processor.go index e44b943273..f521764d28 100644 --- a/core/parallel_state_processor.go +++ b/core/parallel_state_processor.go @@ -363,8 +363,6 @@ func (p *ParallelStateProcessor) Process(block *types.Block, statedb *state.Stat if task.shouldRerunWithoutFeeDelay { shouldDelayFeeCal = false - statedb.StopPrefetcher() - // nolint *statedb = *backupStateDB diff --git a/core/state/statedb.go b/core/state/statedb.go index 35c90bd8b3..19eedbbc03 100644 --- a/core/state/statedb.go +++ b/core/state/statedb.go @@ -1179,6 +1179,10 @@ func (s *StateDB) Copy() *StateDB { state.accessList = s.accessList.Copy() state.transientStorage = s.transientStorage.Copy() + if s.prefetcher != nil { + state.prefetcher = s.prefetcher + } + if s.mvHashmap != nil { state.mvHashmap = s.mvHashmap } diff --git a/core/state/trie_prefetcher.go b/core/state/trie_prefetcher.go index 81f3e1effa..0165ba78e3 100644 --- a/core/state/trie_prefetcher.go +++ b/core/state/trie_prefetcher.go @@ -61,6 +61,8 @@ type triePrefetcher struct { storageDupWriteMeter metrics.Meter storageDupCrossMeter metrics.Meter storageWasteMeter metrics.Meter + + lock sync.RWMutex // Use RWMutex for better read/write locking } func newTriePrefetcher(db Database, root common.Hash, namespace string, noreads bool) *triePrefetcher { @@ -94,6 +96,9 @@ func newTriePrefetcher(db Database, root common.Hash, namespace string, noreads // to all of them. Depending on the async parameter, the method will either block // until all subfetchers spin down, or return immediately. func (p *triePrefetcher) terminate(async bool) { + p.lock.Lock() // Lock for writing + defer p.lock.Unlock() // Ensure the lock is released after the function + // Short circuit if the fetcher is already closed select { case <-p.term: @@ -109,6 +114,9 @@ func (p *triePrefetcher) terminate(async bool) { // report aggregates the pre-fetching and usage metrics and reports them. func (p *triePrefetcher) report() { + p.lock.RLock() // Lock for reading + defer p.lock.RUnlock() // Ensure the lock is released after the function + if !metrics.Enabled { return } @@ -157,6 +165,9 @@ func (p *triePrefetcher) report() { // repeated. // 2. Finalize of the main account trie. This happens only once per block. func (p *triePrefetcher) prefetch(owner common.Hash, root common.Hash, addr common.Address, keys [][]byte, read bool) error { + p.lock.Lock() // Lock for writing + defer p.lock.Unlock() // Ensure the lock is released after the function + // If the state item is only being read, but reads are disabled, return if read && p.noreads { return nil @@ -181,6 +192,9 @@ func (p *triePrefetcher) prefetch(owner common.Hash, root common.Hash, addr comm // the given trie terminates. If no fetcher exists for the request, nil will be // returned. func (p *triePrefetcher) trie(owner common.Hash, root common.Hash) Trie { + p.lock.RLock() // Lock for reading + defer p.lock.RUnlock() // Ensure the lock is released after the function + // Bail if no trie was prefetched for this root fetcher := p.fetchers[p.trieID(owner, root)] if fetcher == nil { @@ -195,6 +209,9 @@ func (p *triePrefetcher) trie(owner common.Hash, root common.Hash) Trie { // used marks a batch of state items used to allow creating statistics as to // how useful or wasteful the fetcher is. func (p *triePrefetcher) used(owner common.Hash, root common.Hash, used [][]byte) { + p.lock.Lock() // Lock for writing + defer p.lock.Unlock() // Ensure the lock is released after the function + if fetcher := p.fetchers[p.trieID(owner, root)]; fetcher != nil { fetcher.wait() // ensure the fetcher's idle before poking in its internals fetcher.used = used diff --git a/eth/handler.go b/eth/handler.go index 12d826bdb6..6ca929f880 100644 --- a/eth/handler.go +++ b/eth/handler.go @@ -525,8 +525,6 @@ func (h *handler) Start(maxPeers int) { func (h *handler) Stop() { h.txsSub.Unsubscribe() // quits txBroadcastLoop - h.txFetcher.Stop() - h.downloader.Terminate() h.minedBlockSub.Unsubscribe() // Quit chainSync and txsync64. diff --git a/eth/sync.go b/eth/sync.go index eeb2781091..6475b80f9f 100644 --- a/eth/sync.go +++ b/eth/sync.go @@ -94,7 +94,7 @@ func (cs *chainSyncer) loop() { cs.handler.txFetcher.Start() defer cs.handler.blockFetcher.Stop() - // defer cs.handler.txFetcher.Stop() + defer cs.handler.txFetcher.Stop() defer cs.handler.downloader.Terminate() // The force timer lowers the peer count threshold down to one when it fires.