diff --git a/triedb/pathdb/lookup.go b/triedb/pathdb/lookup.go index 6ed9634e3f..50fce47a16 100644 --- a/triedb/pathdb/lookup.go +++ b/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 { diff --git a/triedb/pathdb/lookup_test.go b/triedb/pathdb/lookup_test.go index 3d9a2be3a7..827f479763 100644 --- a/triedb/pathdb/lookup_test.go +++ b/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) + } +}