[NIT-3021] paralellise pruning inside accounts

core/state/pruner/pruner.go

@@ -22,6 +22,7 @@ import (
 	"errors"
 	"fmt"
 	"math"
+	"math/big"
 	"os"
 	"path/filepath"
 	"sync"
@@ -370,6 +371,11 @@ func dumpRawTrieDescendants(db ethdb.Database, root common.Hash, output *stateBl
 	for i := 0; i < threads; i++ {
 		results <- nil
 	}
+	// We also create a semaphore for the storage trie traversal, to limit the number of goroutines
+	resultsPerAccount := make(chan error, threads)
+	for i := 0; i < threads; i++ {
+		resultsPerAccount <- nil
+	}
 	var threadsRunning atomic.Int32
 
 	for accountIt.Next(true) {
@@ -422,31 +428,60 @@ func dumpRawTrieDescendants(db ethdb.Database, root common.Hash, output *stateBl
 					threadStartedAt := time.Now()
 					threadLastLog := time.Now()
 
-					storageIt, err := storageTr.NodeIterator(nil)
-					if err != nil {
-						return
-					}
 					var processedNodes uint64
-					for storageIt.Next(true) {
-						storageTrieHash := storageIt.Hash()
-						if storageTrieHash != (common.Hash{}) {
-							// The inner bloomfilter library has a mutex so concurrency is fine here
-							err = output.Put(storageTrieHash.Bytes(), nil)
+					for i := int64(1); i <= 32; i++ {
+						err = <-resultsPerAccount
+						if err != nil {
+							return
+						}
+						var startIterator trie.NodeIterator
+						startIterator, err = storageTr.NodeIterator(big.NewInt((i - 1) << 3).Bytes())
+						if err != nil {
+							return
+						}
+						go func(startIt trie.NodeIterator, iteration int64) {
+							threadsRunning.Add(1)
+							defer threadsRunning.Add(-1)
+							var err error
+							defer func() {
+								resultsPerAccount <- err
+							}()
+
+							// Traverse the storage trie, and stop if we reach the end of the trie or the end of the current part
+							var startItPath, endItPath []byte
+
+							key := keybytesToHex(big.NewInt((iteration) << 3).Bytes())
+							key = key[:len(key)-1]
+							for startIt.Next(true) {
+								if iteration != 32 && bytes.Compare(startIt.Path(), key) >= 0 {
+									break
+								}
+								if startItPath == nil {
+									startItPath = startIt.Path()
+								}
+								endItPath = startIt.Path()
+								storageTrieHash := startIt.Hash()
+								if storageTrieHash != (common.Hash{}) {
+									// The inner bloomfilter library has a mutex so concurrency is fine here
+									err = output.Put(storageTrieHash.Bytes(), nil)
+									if err != nil {
+										return
+									}
+								}
+								processedNodes++
+								if time.Since(threadLastLog) > 5*time.Minute {
+									elapsedTotal := time.Since(startedAt)
+									elapsedThread := time.Since(threadStartedAt)
+									log.Info("traversing trie database - traversing storage trie taking long", "key", key, "elapsedTotal", elapsedTotal, "elapsedThread", elapsedThread, "processedNodes", processedNodes, "threadsRunning", threadsRunning.Load())
+									threadLastLog = time.Now()
+								}
+							}
+							err = startIt.Error()
 							if err != nil {
 								return
 							}
-						}
-						processedNodes++
-						if time.Since(threadLastLog) > 5*time.Minute {
-							elapsedTotal := time.Since(startedAt)
-							elapsedThread := time.Since(threadStartedAt)
-							log.Info("traversing trie database - traversing storage trie taking long", "key", key, "elapsedTotal", elapsedTotal, "elapsedThread", elapsedThread, "processedNodes", processedNodes, "threadsRunning", threadsRunning.Load())
-							threadLastLog = time.Now()
-						}
-					}
-					err = storageIt.Error()
-					if err != nil {
-						return
+							log.Trace("Finished traversing storage trie", "key", key, "startPath", startItPath, "endPath", endItPath)
+						}(startIterator, i)
 					}
 				}()
 			}
@@ -461,9 +496,26 @@ func dumpRawTrieDescendants(db ethdb.Database, root common.Hash, output *stateBl
 			return err
 		}
 	}
+	for i := 0; i < threads; i++ {
+		err = <-resultsPerAccount
+		if err != nil {
+			return err
+		}
+	}
 	return nil
 }
 
+func keybytesToHex(str []byte) []byte {
+	l := len(str)*2 + 1
+	var nibbles = make([]byte, l)
+	for i, b := range str {
+		nibbles[i*2] = b / 16
+		nibbles[i*2+1] = b % 16
+	}
+	nibbles[l-1] = 16
+	return nibbles
+}
+
 // Prune deletes all historical state nodes except the nodes belong to the
 // specified state version. If user doesn't specify the state version, use
 // the bottom-most snapshot diff layer as the target.