use 16 concurrent workers

Signed-off-by: jsvisa <delweng@gmail.com>

triedb/pathdb/lookup.go

@@ -74,7 +74,7 @@ func getStorageShardIndex(path string) int {
 	hash := 0
 	for i, r := range path {
 		hash = hash*31 + int(r)
-		if i >= 4 { // Use first 4 characters for good distribution
+		if i >= 8 { // Use first 8 characters for better distribution
 			break
 		}
 	}
@@ -292,55 +292,79 @@ func (l *lookup) addLayer(diff *diffLayer) {
 		st2 = time.Since(st)
 	}()
 
-	wg.Add(1)
-	go func() {
-		defer wg.Done()
-		st := time.Now()
+	// Use concurrent workers for storage nodes updates, one per shard
+	var storageWg sync.WaitGroup
+	storageWg.Add(16)
 
-		count := 0
-		var stHash, stKey, stMapGet, stMapNew, stMapApp, stMapSet time.Duration
+	// Create channels to distribute work to workers
+	type storageWork struct {
+		accountHash common.Hash
+		accountHex  string
+		path        string
+	}
 
-		for accountHash, slots := range diff.nodes.storageNodes {
-			st00 := time.Now()
-			accountHex := accountHash.Hex()
-			st01 := time.Now()
-			stHash += st01.Sub(st00)
+	workChannels := make([]chan storageWork, 16)
+	for i := 0; i < 16; i++ {
+		workChannels[i] = make(chan storageWork, 1000) // Buffer to avoid blocking
+	}
 
-			for path := range slots {
-				count += 1
+	// Start 16 workers, each handling its own shard
+	for shardIndex := 0; shardIndex < 16; shardIndex++ {
+		go func(shardIdx int) {
+			defer storageWg.Done()
+			st := time.Now()
+			count := 0
 
-				st00 := time.Now()
-				// Construct the combined key and find the correct shard
-				key := accountHex + path
-				shardIndex := getStorageShardIndex(path)
-				st01 := time.Now()
+			for work := range workChannels[shardIdx] {
+				// Construct the combined key
+				key := work.accountHex + work.path
 
-				// Access the specific shard map
-				shardMap := l.storageNodes[shardIndex]
+				// Access the specific shard map (no lock needed as each worker owns its shard)
+				shardMap := l.storageNodes[shardIdx]
 				list, exists := shardMap[key]
-				st02 := time.Now()
 				if !exists {
 					list = make([]common.Hash, 0, 16) // TODO(rjl493456442) use sync pool
 				}
-				st03 := time.Now()
 				list = append(list, state)
-				st04 := time.Now()
 				shardMap[key] = list
-				st05 := time.Now()
-
-				stKey += st01.Sub(st00)
-				stMapGet += st02.Sub(st01)
-				stMapNew += st03.Sub(st02)
-				stMapApp += st04.Sub(st03)
-				stMapSet += st05.Sub(st04)
+				count++
 			}
-		}
 
-		st3 = time.Since(st)
-		if st3 > time.Millisecond {
-			log.Info("PathDB lookup add storage nodes", "count", count, "accounts", len(diff.nodes.storageNodes), "stHash", stHash, "stKey", stKey, "stMapGet", stMapGet, "stMapNew", stMapNew, "stMapApp", stMapApp, "stMapSet", stMapSet)
+			st3 := time.Since(st)
+			if st3 > time.Millisecond {
+				log.Info("PathDB lookup add storage nodes worker", "shard", shardIdx, "count", count, "elapsed", st3)
+			}
+		}(shardIndex)
+	}
+
+	// Distribute work to workers based on shard index
+	distributeStart := time.Now()
+	totalCount := 0
+	for accountHash, slots := range diff.nodes.storageNodes {
+		accountHex := accountHash.Hex()
+		for path := range slots {
+			shardIndex := getStorageShardIndex(path)
+			workChannels[shardIndex] <- storageWork{
+				accountHash: accountHash,
+				accountHex:  accountHex,
+				path:        path,
+			}
+			totalCount++
 		}
-	}()
+	}
+
+	// Close all channels to signal workers to finish
+	for i := 0; i < 16; i++ {
+		close(workChannels[i])
+	}
+
+	// Wait for all storage workers to complete
+	storageWg.Wait()
+	st3 = time.Since(distributeStart)
+
+	if st3 > time.Millisecond {
+		log.Info("PathDB lookup add storage nodes", "total_count", totalCount, "accounts", len(diff.nodes.storageNodes), "elapsed", st3)
+	}
 
 	wg.Wait()
 	if elapsed := time.Since(st00); elapsed > time.Millisecond {

triedb/pathdb/lookup_test.go

@@ -2,6 +2,7 @@ package pathdb
 
 import (
 	"crypto/rand"
+	"fmt"
 	"testing"
 
 	"github.com/ethereum/go-ethereum/common"
@@ -133,3 +134,123 @@ func BenchmarkAddNodes(b *testing.B) {
 		})
 	}
 }
+
+func TestConcurrentStorageNodesUpdate(b *testing.T) {
+	// Create a lookup instance
+	lookup := &lookup{
+		accountNodes: make(map[string][]common.Hash),
+	}
+	// Initialize all 16 storage node shards
+	for i := 0; i < 16; i++ {
+		lookup.storageNodes[i] = make(map[string][]common.Hash)
+	}
+
+	// Create test data with known shard distribution
+	testData := map[common.Hash]map[string]*trienode.Node{}
+
+	// Create accounts that will distribute across different shards
+	for i := 0; i < 100; i++ {
+		var accountHash common.Hash
+		accountHash[0] = byte(i)
+
+		testData[accountHash] = make(map[string]*trienode.Node)
+
+		// Create paths that will hash to different shards
+		for j := 0; j < 10; j++ {
+			path := fmt.Sprintf("path_%d_%d", i, j)
+			var nodeHash common.Hash
+			nodeHash[0] = byte(j)
+
+			testData[accountHash][path] = &trienode.Node{Hash: nodeHash}
+		}
+	}
+
+	// Add nodes using the concurrent method
+	var stateHash common.Hash
+	stateHash[0] = 0x42
+	lookup.addNodes(stateHash, nil, testData)
+
+	// Verify that all nodes were added correctly
+	totalNodes := 0
+	for accountHash, slots := range testData {
+		accountHex := accountHash.Hex()
+		for path := range slots {
+			key := accountHex + path
+			shardIndex := getStorageShardIndex(path)
+
+			list, exists := lookup.storageNodes[shardIndex][key]
+			if !exists {
+				b.Errorf("Node not found: account=%x, path=%s, shard=%d", accountHash, path, shardIndex)
+				continue
+			}
+
+			if len(list) != 1 {
+				b.Errorf("Expected 1 state hash, got %d: account=%x, path=%s", len(list), accountHash, path)
+				continue
+			}
+
+			if list[0] != stateHash {
+				b.Errorf("Expected state hash %x, got %x: account=%x, path=%s", stateHash, list[0], accountHash, path)
+				continue
+			}
+
+			totalNodes++
+		}
+	}
+
+	expectedTotal := 100 * 10 // 100 accounts * 10 nodes each
+	if totalNodes != expectedTotal {
+		b.Errorf("Expected %d total nodes, got %d", expectedTotal, totalNodes)
+	}
+
+	// Verify shard distribution
+	for i := 0; i < 16; i++ {
+		shardSize := len(lookup.storageNodes[i])
+		if shardSize == 0 {
+			b.Logf("Shard %d is empty", i)
+		} else {
+			b.Logf("Shard %d has %d entries", i, shardSize)
+		}
+	}
+}
+
+func TestShardDistribution(b *testing.T) {
+	// Test shard distribution with different path patterns
+	paths := []string{
+		"path_0_0", "path_0_1", "path_0_2", "path_0_3",
+		"path_1_0", "path_1_1", "path_1_2", "path_1_3",
+		"path_2_0", "path_2_1", "path_2_2", "path_2_3",
+		"path_3_0", "path_3_1", "path_3_2", "path_3_3",
+		"path_4_0", "path_4_1", "path_4_2", "path_4_3",
+		"path_5_0", "path_5_1", "path_5_2", "path_5_3",
+		"path_6_0", "path_6_1", "path_6_2", "path_6_3",
+		"path_7_0", "path_7_1", "path_7_2", "path_7_3",
+		"path_8_0", "path_8_1", "path_8_2", "path_8_3",
+		"path_9_0", "path_9_1", "path_9_2", "path_9_3",
+	}
+
+	shardCounts := make(map[int]int)
+	for _, path := range paths {
+		shardIndex := getStorageShardIndex(path)
+		shardCounts[shardIndex]++
+		b.Logf("Path: %s -> Shard: %d", path, shardIndex)
+	}
+
+	b.Logf("Shard distribution:")
+	for i := 0; i < 16; i++ {
+		count := shardCounts[i]
+		b.Logf("  Shard %d: %d paths", i, count)
+	}
+
+	// Check if we have a reasonable distribution
+	usedShards := 0
+	for _, count := range shardCounts {
+		if count > 0 {
+			usedShards++
+		}
+	}
+
+	if usedShards < 4 {
+		b.Logf("Warning: Only %d shards are being used out of 16", usedShards)
+	}
+}