feat: implement parallel account trie hashing with prefetching of dirty accounts

This commit is contained in:
Sahil-4555 2025-10-17 15:15:42 +05:30
parent 342285b139
commit 5f53eda30f
2 changed files with 46 additions and 5 deletions

View file

@ -1155,6 +1155,7 @@ func (s *StateDB) GetTrie() Trie {
// commit gathers the state mutations accumulated along with the associated // commit gathers the state mutations accumulated along with the associated
// trie changes, resetting all internal flags with the new state as the base. // trie changes, resetting all internal flags with the new state as the base.
// Modified to parallelize account trie hashing with storage trie commits.
func (s *StateDB) commit(deleteEmptyObjects bool, noStorageWiping bool, blockNumber uint64) (*stateUpdate, error) { func (s *StateDB) commit(deleteEmptyObjects bool, noStorageWiping bool, blockNumber uint64) (*stateUpdate, error) {
// Short circuit in case any database failure occurred earlier. // Short circuit in case any database failure occurred earlier.
if s.dbErr != nil { if s.dbErr != nil {
@ -1218,6 +1219,15 @@ func (s *StateDB) commit(deleteEmptyObjects bool, noStorageWiping bool, blockNum
return nil, err return nil, err
} }
} }
if s.prefetcher != nil {
// Gather all account addresses with mutations
dirtyAccounts := s.getDirtyAccountAddresses()
if err := s.prefetcher.PrefetchAccounts(dirtyAccounts); err != nil {
log.Warn("Failed to prefetch account trie nodes before parallel hashing", "err", err)
}
}
// Handle all state updates afterwards, concurrently to one another to shave // Handle all state updates afterwards, concurrently to one another to shave
// off some milliseconds from the commit operation. Also accumulate the code // off some milliseconds from the commit operation. Also accumulate the code
// writes to run in parallel with the computations. // writes to run in parallel with the computations.
@ -1225,6 +1235,7 @@ func (s *StateDB) commit(deleteEmptyObjects bool, noStorageWiping bool, blockNum
start = time.Now() start = time.Now()
root common.Hash root common.Hash
workers errgroup.Group workers errgroup.Group
accountTrieHash common.Hash
) )
// Schedule the account trie first since that will be the biggest, so give // Schedule the account trie first since that will be the biggest, so give
// it the most time to crunch. // it the most time to crunch.
@ -1246,6 +1257,14 @@ func (s *StateDB) commit(deleteEmptyObjects bool, noStorageWiping bool, blockNum
s.AccountCommits = time.Since(start) s.AccountCommits = time.Since(start)
return nil return nil
}) })
// Step 2: Launch account trie hashing concurrently (CPU only, no I/O).
trieCopy := s.trie
workers.Go(func() error {
accountTrieHash = trieCopy.Hash()
return nil
})
// Schedule each of the storage tries that need to be updated, so they can // Schedule each of the storage tries that need to be updated, so they can
// run concurrently to one another. // run concurrently to one another.
// //
@ -1305,9 +1324,13 @@ func (s *StateDB) commit(deleteEmptyObjects bool, noStorageWiping bool, blockNum
s.stateObjectsDestruct = make(map[common.Address]*stateObject) s.stateObjectsDestruct = make(map[common.Address]*stateObject)
origin := s.originalRoot origin := s.originalRoot
if accountTrieHash != (common.Hash{}) {
s.originalRoot = accountTrieHash
} else {
s.originalRoot = root s.originalRoot = root
}
return newStateUpdate(noStorageWiping, origin, root, blockNumber, deletes, updates, nodes), nil return newStateUpdate(noStorageWiping, origin, s.originalRoot, blockNumber, deletes, updates, nodes), nil
} }
// commitAndFlush is a wrapper of commit which also commits the state mutations // commitAndFlush is a wrapper of commit which also commits the state mutations
@ -1388,6 +1411,15 @@ func (s *StateDB) CommitWithUpdate(block uint64, deleteEmptyObjects bool, noStor
return ret.root, ret, nil return ret.root, ret, nil
} }
// getDirtyAccountAddresses returns the list of addresses of accounts that have pending mutations.
func (s *StateDB) getDirtyAccountAddresses() []common.Address {
addrs := make([]common.Address, 0, len(s.mutations))
for addr := range s.mutations {
addrs = append(addrs, addr)
}
return addrs
}
// Prepare handles the preparatory steps for executing a state transition with. // Prepare handles the preparatory steps for executing a state transition with.
// This method must be invoked before state transition. // This method must be invoked before state transition.
// //

View file

@ -218,6 +218,15 @@ func (p *triePrefetcher) trieID(owner common.Hash, root common.Hash) string {
return string(trieID) return string(trieID)
} }
// PrefetchAccounts triggers prefetching account trie nodes given a list of accounts.
func (p *triePrefetcher) PrefetchAccounts(accounts []common.Address) error {
if p == nil {
return nil
}
// The account trie owner hash is zero (empty) as per convention.
return p.prefetch(common.Hash{}, p.root, common.Address{}, accounts, nil, false)
}
// subfetcher is a trie fetcher goroutine responsible for pulling entries for a // subfetcher is a trie fetcher goroutine responsible for pulling entries for a
// single trie. It is spawned when a new root is encountered and lives until the // single trie. It is spawned when a new root is encountered and lives until the
// main prefetcher is paused and either all requested items are processed or if // main prefetcher is paused and either all requested items are processed or if