[core/state] Prefetch Trie Nodes Concurrently

core/blockchain.go

@@ -148,11 +148,12 @@ type CacheConfig struct {
 	TrieCleanRejournal              time.Duration // Time interval to dump clean cache to disk periodically
 	TrieDirtyLimit                  int           // Memory limit (MB) at which to block on insert and force a flush of dirty trie nodes to disk
 	TrieDirtyCommitTarget           int           // Memory limit (MB) to target for the dirties cache before invoking commit
+	TriePrefetcherParallelism       int           // Max concurrent disk reads trie prefetcher should perform at once
 	CommitInterval                  uint64        // Commit the trie every [CommitInterval] blocks.
 	Pruning                         bool          // Whether to disable trie write caching and GC altogether (archive node)
 	AcceptorQueueLimit              int           // Blocks to queue before blocking during acceptance
 	PopulateMissingTries            *uint64       // If non-nil, sets the starting height for re-generating historical tries.
-	PopulateMissingTriesParallelism int           // Is the number of readers to use when trying to populate missing tries.
+	PopulateMissingTriesParallelism int           // Number of readers to use when trying to populate missing tries.
 	AllowMissingTries               bool          // Whether to allow an archive node to run with pruning enabled
 	SnapshotDelayInit               bool          // Whether to initialize snapshots on startup or wait for external call
 	SnapshotLimit                   int           // Memory allowance (MB) to use for caching snapshot entries in memory
@@ -166,14 +167,15 @@ type CacheConfig struct {
 }
 
 var DefaultCacheConfig = &CacheConfig{
-	TrieCleanLimit:        256,
-	TrieDirtyLimit:        256,
-	TrieDirtyCommitTarget: 20, // 20% overhead in memory counting (this targets 16 MB)
-	Pruning:               true,
-	CommitInterval:        4096,
-	AcceptorQueueLimit:    64, // Provides 2 minutes of buffer (2s block target) for a commit delay
-	SnapshotLimit:         256,
-	AcceptedCacheSize:     32,
+	TrieCleanLimit:            256,
+	TrieDirtyLimit:            256,
+	TrieDirtyCommitTarget:     20, // 20% overhead in memory counting (this targets 16 MB)
+	TriePrefetcherParallelism: 16,
+	Pruning:                   true,
+	CommitInterval:            4096,
+	AcceptorQueueLimit:        64, // Provides 2 minutes of buffer (2s block target) for a commit delay
+	SnapshotLimit:             256,
+	AcceptedCacheSize:         32,
 }
 
 // BlockChain represents the canonical chain given a database with a genesis
@@ -1312,7 +1314,7 @@ func (bc *BlockChain) insertBlock(block *types.Block, writes bool) error {
 	blockStateInitTimer.Inc(time.Since(substart).Milliseconds())
 
 	// Enable prefetching to pull in trie node paths while processing transactions
-	statedb.StartPrefetcher("chain")
+	statedb.StartPrefetcher("chain", bc.cacheConfig.TriePrefetcherParallelism)
 	activeState = statedb
 
 	// If we have a followup block, run that against the current state to pre-cache
@@ -1687,7 +1689,7 @@ func (bc *BlockChain) reprocessBlock(parent *types.Block, current *types.Block)
 	}
 
 	// Enable prefetching to pull in trie node paths while processing transactions
-	statedb.StartPrefetcher("chain")
+	statedb.StartPrefetcher("chain", bc.cacheConfig.TriePrefetcherParallelism)
 	defer func() {
 		statedb.StopPrefetcher()
 	}()
@@ -2094,3 +2096,11 @@ func (bc *BlockChain) ResetToStateSyncedBlock(block *types.Block) error {
 	bc.initSnapshot(head)
 	return nil
 }
+
+// CacheConfig returns a reference to [bc.cacheConfig]
+//
+// This is used by [miner] to set prefetch parallelism
+// during block building.
+func (bc *BlockChain) CacheConfig() *CacheConfig {
+	return bc.cacheConfig
+}

core/blockchain_repair_test.go

@@ -524,9 +524,10 @@ func testRepair(t *testing.T, tt *rewindTest, snapshots bool) {
 		}
 		engine = dummy.NewFullFaker()
 		config = &CacheConfig{
-			TrieCleanLimit: 256,
-			TrieDirtyLimit: 256,
-			SnapshotLimit:  0, // Disable snapshot by default
+			TrieCleanLimit:            256,
+			TrieDirtyLimit:            256,
+			TriePrefetcherParallelism: 4,
+			SnapshotLimit:             0, // Disable snapshot by default
 		}
 	)
 	defer engine.Close()

core/blockchain_test.go

@@ -28,22 +28,24 @@ import (
 
 var (
 	archiveConfig = &CacheConfig{
-		TrieCleanLimit:        256,
-		TrieDirtyLimit:        256,
-		TrieDirtyCommitTarget: 20,
-		Pruning:               false, // Archive mode
-		SnapshotLimit:         256,
-		AcceptorQueueLimit:    64,
+		TrieCleanLimit:            256,
+		TrieDirtyLimit:            256,
+		TrieDirtyCommitTarget:     20,
+		TriePrefetcherParallelism: 4,
+		Pruning:                   false, // Archive mode
+		SnapshotLimit:             256,
+		AcceptorQueueLimit:        64,
 	}
 
 	pruningConfig = &CacheConfig{
-		TrieCleanLimit:        256,
-		TrieDirtyLimit:        256,
-		TrieDirtyCommitTarget: 20,
-		Pruning:               true, // Enable pruning
-		CommitInterval:        4096,
-		SnapshotLimit:         256,
-		AcceptorQueueLimit:    64,
+		TrieCleanLimit:            256,
+		TrieDirtyLimit:            256,
+		TrieDirtyCommitTarget:     20,
+		TriePrefetcherParallelism: 4,
+		Pruning:                   true, // Enable pruning
+		CommitInterval:            4096,
+		SnapshotLimit:             256,
+		AcceptorQueueLimit:        64,
 	}
 )
 
@@ -180,12 +182,13 @@ func TestArchiveBlockChainSnapsDisabled(t *testing.T) {
 		return createBlockChain(
 			db,
 			&CacheConfig{
-				TrieCleanLimit:        256,
-				TrieDirtyLimit:        256,
-				TrieDirtyCommitTarget: 20,
-				Pruning:               false, // Archive mode
-				SnapshotLimit:         0,     // Disable snapshots
-				AcceptorQueueLimit:    64,
+				TrieCleanLimit:            256,
+				TrieDirtyLimit:            256,
+				TrieDirtyCommitTarget:     20,
+				TriePrefetcherParallelism: 4,
+				Pruning:                   false, // Archive mode
+				SnapshotLimit:             0,     // Disable snapshots
+				AcceptorQueueLimit:        64,
 			},
 			gspec,
 			lastAcceptedHash,
@@ -214,13 +217,14 @@ func TestPruningBlockChainSnapsDisabled(t *testing.T) {
 		return createBlockChain(
 			db,
 			&CacheConfig{
-				TrieCleanLimit:        256,
-				TrieDirtyLimit:        256,
-				TrieDirtyCommitTarget: 20,
-				Pruning:               true, // Enable pruning
-				CommitInterval:        4096,
-				SnapshotLimit:         0, // Disable snapshots
-				AcceptorQueueLimit:    64,
+				TrieCleanLimit:            256,
+				TrieDirtyLimit:            256,
+				TrieDirtyCommitTarget:     20,
+				TriePrefetcherParallelism: 4,
+				Pruning:                   true, // Enable pruning
+				CommitInterval:            4096,
+				SnapshotLimit:             0, // Disable snapshots
+				AcceptorQueueLimit:        64,
 			},
 			gspec,
 			lastAcceptedHash,
@@ -263,13 +267,14 @@ func TestPruningBlockChainUngracefulShutdownSnapsDisabled(t *testing.T) {
 		blockchain, err := createBlockChain(
 			db,
 			&CacheConfig{
-				TrieCleanLimit:        256,
-				TrieDirtyLimit:        256,
-				TrieDirtyCommitTarget: 20,
-				Pruning:               true, // Enable pruning
-				CommitInterval:        4096,
-				SnapshotLimit:         0, // Disable snapshots
-				AcceptorQueueLimit:    64,
+				TrieCleanLimit:            256,
+				TrieDirtyLimit:            256,
+				TrieDirtyCommitTarget:     20,
+				TriePrefetcherParallelism: 4,
+				Pruning:                   true, // Enable pruning
+				CommitInterval:            4096,
+				SnapshotLimit:             0, // Disable snapshots
+				AcceptorQueueLimit:        64,
 			},
 			gspec,
 			lastAcceptedHash,
@@ -298,13 +303,14 @@ func TestEnableSnapshots(t *testing.T) {
 		blockchain, err := createBlockChain(
 			db,
 			&CacheConfig{
-				TrieCleanLimit:        256,
-				TrieDirtyLimit:        256,
-				TrieDirtyCommitTarget: 20,
-				Pruning:               true, // Enable pruning
-				CommitInterval:        4096,
-				SnapshotLimit:         snapLimit,
-				AcceptorQueueLimit:    64,
+				TrieCleanLimit:            256,
+				TrieDirtyLimit:            256,
+				TrieDirtyCommitTarget:     20,
+				TriePrefetcherParallelism: 4,
+				Pruning:                   true, // Enable pruning
+				CommitInterval:            4096,
+				SnapshotLimit:             snapLimit,
+				AcceptorQueueLimit:        64,
 			},
 			gspec,
 			lastAcceptedHash,
@@ -455,6 +461,7 @@ func testRepopulateMissingTriesParallel(t *testing.T, parallelism int) {
 			TrieCleanLimit:                  256,
 			TrieDirtyLimit:                  256,
 			TrieDirtyCommitTarget:           20,
+			TriePrefetcherParallelism:       4,
 			Pruning:                         false, // Archive mode
 			SnapshotLimit:                   256,
 			PopulateMissingTries:            &startHeight, // Starting point for re-populating.
@@ -487,14 +494,15 @@ func TestUngracefulAsyncShutdown(t *testing.T) {
 	var (
 		create = func(db ethdb.Database, gspec *Genesis, lastAcceptedHash common.Hash) (*BlockChain, error) {
 			blockchain, err := createBlockChain(db, &CacheConfig{
-				TrieCleanLimit:        256,
-				TrieDirtyLimit:        256,
-				TrieDirtyCommitTarget: 20,
-				Pruning:               true,
-				CommitInterval:        4096,
-				SnapshotLimit:         256,
-				SnapshotNoBuild:       true, // Ensure the test errors if snapshot initialization fails
-				AcceptorQueueLimit:    1000, // ensure channel doesn't block
+				TrieCleanLimit:            256,
+				TrieDirtyLimit:            256,
+				TrieDirtyCommitTarget:     20,
+				TriePrefetcherParallelism: 4,
+				Pruning:                   true,
+				CommitInterval:            4096,
+				SnapshotLimit:             256,
+				SnapshotNoBuild:           true, // Ensure the test errors if snapshot initialization fails
+				AcceptorQueueLimit:        1000, // ensure channel doesn't block
 			}, gspec, lastAcceptedHash)
 			if err != nil {
 				return nil, err
@@ -681,14 +689,15 @@ func TestTransactionIndices(t *testing.T) {
 	}
 
 	conf := &CacheConfig{
-		TrieCleanLimit:        256,
-		TrieDirtyLimit:        256,
-		TrieDirtyCommitTarget: 20,
-		Pruning:               true,
-		CommitInterval:        4096,
-		SnapshotLimit:         256,
-		SnapshotNoBuild:       true, // Ensure the test errors if snapshot initialization fails
-		AcceptorQueueLimit:    64,
+		TrieCleanLimit:            256,
+		TrieDirtyLimit:            256,
+		TrieDirtyCommitTarget:     20,
+		TriePrefetcherParallelism: 4,
+		Pruning:                   true,
+		CommitInterval:            4096,
+		SnapshotLimit:             256,
+		SnapshotNoBuild:           true, // Ensure the test errors if snapshot initialization fails
+		AcceptorQueueLimit:        64,
 	}
 
 	// Init block chain and check all needed indices has been indexed.
@@ -851,15 +860,16 @@ func TestCanonicalHashMarker(t *testing.T) {
 
 func TestTxLookupBlockChain(t *testing.T) {
 	cacheConf := &CacheConfig{
-		TrieCleanLimit:        256,
-		TrieDirtyLimit:        256,
-		TrieDirtyCommitTarget: 20,
-		Pruning:               true,
-		CommitInterval:        4096,
-		SnapshotLimit:         256,
-		SnapshotNoBuild:       true, // Ensure the test errors if snapshot initialization fails
-		AcceptorQueueLimit:    64,   // ensure channel doesn't block
-		TxLookupLimit:         5,
+		TrieCleanLimit:            256,
+		TrieDirtyLimit:            256,
+		TrieDirtyCommitTarget:     20,
+		TriePrefetcherParallelism: 4,
+		Pruning:                   true,
+		CommitInterval:            4096,
+		SnapshotLimit:             256,
+		SnapshotNoBuild:           true, // Ensure the test errors if snapshot initialization fails
+		AcceptorQueueLimit:        64,   // ensure channel doesn't block
+		TxLookupLimit:             5,
 	}
 	createTxLookupBlockChain := func(db ethdb.Database, gspec *Genesis, lastAcceptedHash common.Hash) (*BlockChain, error) {
 		return createBlockChain(db, cacheConf, gspec, lastAcceptedHash)

core/state/statedb.go

@@ -186,13 +186,13 @@ func NewWithSnapshot(root common.Hash, db Database, snap snapshot.Snapshot) (*St
 // StartPrefetcher initializes a new trie prefetcher to pull in nodes from the
 // state trie concurrently while the state is mutated so that when we reach the
 // commit phase, most of the needed data is already hot.
-func (s *StateDB) StartPrefetcher(namespace string) {
+func (s *StateDB) StartPrefetcher(namespace string, maxConcurrency int) {
 	if s.prefetcher != nil {
 		s.prefetcher.close()
 		s.prefetcher = nil
 	}
 	if s.snap != nil {
-		s.prefetcher = newTriePrefetcher(s.db, s.originalRoot, namespace)
+		s.prefetcher = newTriePrefetcher(s.db, s.originalRoot, namespace, maxConcurrency)
 	}
 }