Address goroutine leak

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
+			parallelStatedb.StartPrefetcher("chain", nil)
-			var witness *stateless.Witness
-			if bc.vmConfig.EnableWitnessCollection {
-				witness, err = stateless.NewWitness(bc, block)
-				if err != nil {
-					return
-				}
-			}
-			parallelStatedb.StartPrefetcher("chain", witness)
 			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
+			statedb.StartPrefetcher("chain", nil)
-			var witness *stateless.Witness
-			if bc.vmConfig.EnableWitnessCollection {
-				witness, err = stateless.NewWitness(bc, block)
-				if err != nil {
-					return
-				}
-			}
-			statedb.StartPrefetcher("chain", witness)
-
 			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

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
 

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
 	}

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

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.

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.