From e270da3d77b03099f65355d3eb2f91e78a404638 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?P=C3=A9ter=20Szil=C3=A1gyi?= Date: Tue, 30 Apr 2024 09:06:24 +0300 Subject: [PATCH] core/state: use errgroups, commit accounts concurrently --- core/state/statedb.go | 103 ++++++++++++++++++----------------- internal/syncx/workerpool.go | 94 -------------------------------- 2 files changed, 53 insertions(+), 144 deletions(-) delete mode 100644 internal/syncx/workerpool.go diff --git a/core/state/statedb.go b/core/state/statedb.go index f173da0b92..957cb531cc 100644 --- a/core/state/statedb.go +++ b/core/state/statedb.go @@ -21,7 +21,6 @@ import ( "fmt" "maps" "math/big" - "runtime" "slices" "sort" "sync" @@ -33,13 +32,13 @@ import ( "github.com/ethereum/go-ethereum/core/tracing" "github.com/ethereum/go-ethereum/core/types" "github.com/ethereum/go-ethereum/crypto" - "github.com/ethereum/go-ethereum/internal/syncx" "github.com/ethereum/go-ethereum/log" "github.com/ethereum/go-ethereum/params" "github.com/ethereum/go-ethereum/trie" "github.com/ethereum/go-ethereum/trie/trienode" "github.com/ethereum/go-ethereum/trie/triestate" "github.com/holiman/uint256" + "golang.org/x/sync/errgroup" ) type revision struct { @@ -1159,11 +1158,51 @@ func (s *StateDB) Commit(block uint64, deleteEmptyObjects bool) (common.Hash, er // writes to run in parallel with the computations. start := time.Now() var ( - code = s.db.DiskDB().NewBatch() - lock sync.Mutex + code = s.db.DiskDB().NewBatch() + lock sync.Mutex + root common.Hash + workers errgroup.Group ) - workers := syncx.NewWorkerPool[*stateObject, error](len(s.mutations), min(len(s.mutations), runtime.NumCPU()), - func(obj *stateObject) error { + // Schedule the account trie first since that will be the biggest, so give + // it the most time to crunch. + workers.Go(func() error { + // Write the account trie changes, measuring the amount of wasted time + start = time.Now() + + newroot, set, err := s.trie.Commit(true) + if err != nil { + return err + } + root = newroot + + // Merge the dirty nodes of account trie into global set + if set != nil { + lock.Lock() + defer lock.Unlock() + + if err = nodes.Merge(set); err != nil { + return err + } + accountTrieNodesUpdated, accountTrieNodesDeleted = set.Size() + } + // Report the commit metrics + s.AccountCommits += time.Since(start) + return nil + }) + // Schedule each of the storage tries that need to be updated, so they can + // run concurrently to one another. + for addr, op := range s.mutations { + if op.isDelete() { + continue + } + // Write any contract code associated with the state object + obj := s.stateObjects[addr] + if obj.code != nil && obj.dirtyCode { + rawdb.WriteCode(code, common.BytesToHash(obj.CodeHash()), obj.code) + obj.dirtyCode = false + } + // Run the storage updates concurrently to one another + workers.Go(func() error { // Write any storage changes in the state object to its storage trie set, err := obj.commit() if err != nil { @@ -1185,60 +1224,24 @@ func (s *StateDB) Commit(block uint64, deleteEmptyObjects bool) (common.Hash, er } return nil }) - - for addr, op := range s.mutations { - if op.isDelete() { - continue - } - // Write any contract code associated with the state object - obj := s.stateObjects[addr] - if obj.code != nil && obj.dirtyCode { - rawdb.WriteCode(code, common.BytesToHash(obj.CodeHash()), obj.code) - obj.dirtyCode = false - } - // Run the storage updates concurrently to one another - workers.Schedule(obj) } - workers.Close() - // Updates running concurrently, wait for them to complete; running the code // writes in the meantime. done := make(chan struct{}) go func() { - // This goroutine is only needed to accurately measure the storage commit - // and not have the concurrent code write dirty the stats. defer close(done) - workers.Wait() - s.StorageCommits += time.Since(start) + if code.ValueSize() > 0 { + if err := code.Write(); err != nil { + log.Crit("Failed to commit dirty codes", "error", err) + } + } }() - if code.ValueSize() > 0 { - if err := code.Write(); err != nil { - log.Crit("Failed to commit dirty codes", "error", err) - } - } - <-done - for err := range workers.Results() { - if err != nil { - return common.Hash{}, err - } - } - // Write the account trie changes, measuring the amount of wasted time - start = time.Now() - - root, set, err := s.trie.Commit(true) - if err != nil { + if err := workers.Wait(); err != nil { return common.Hash{}, err } - // Merge the dirty nodes of account trie into global set - if set != nil { - if err := nodes.Merge(set); err != nil { - return common.Hash{}, err - } - accountTrieNodesUpdated, accountTrieNodesDeleted = set.Size() - } - // Report the commit metrics - s.AccountCommits += time.Since(start) + s.StorageCommits += time.Since(start) + <-done accountUpdatedMeter.Mark(int64(s.AccountUpdated)) storageUpdatedMeter.Mark(int64(s.StorageUpdated)) diff --git a/internal/syncx/workerpool.go b/internal/syncx/workerpool.go deleted file mode 100644 index f207ac669e..0000000000 --- a/internal/syncx/workerpool.go +++ /dev/null @@ -1,94 +0,0 @@ -// Copyright 2024 The go-ethereum Authors -// This file is part of the go-ethereum library. -// -// The go-ethereum library is free software: you can redistribute it and/or modify -// it under the terms of the GNU Lesser General Public License as published by -// the Free Software Foundation, either version 3 of the License, or -// (at your option) any later version. -// -// The go-ethereum library is distributed in the hope that it will be useful, -// but WITHOUT ANY WARRANTY; without even the implied warranty of -// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the -// GNU Lesser General Public License for more details. -// -// You should have received a copy of the GNU Lesser General Public License -// along with the go-ethereum library. If not, see . - -package syncx - -import ( - "runtime" - "sync" -) - -// WorkerPool is a concurrent task processor, scheduling and running tasks from -// a source channel, feeding any errors into a sink. -type WorkerPool[T any, R any] struct { - tasks chan T // Input channel waiting to consume tasks - results chan R // Result channel for consuers to wait on - working sync.WaitGroup // Waitgroup blocking on worker liveness -} - -// NewWorkerPool creates a worker pool with the given number of max task capacity -// and an optional goroutine count to execute on. If 0 threads are requested, the -// pool will default to the number of (logical) CPUs. -func NewWorkerPool[T any, R any](tasks int, threads int, f func(T) R) *WorkerPool[T, R] { - // Create the worker pool - pool := &WorkerPool[T, R]{ - tasks: make(chan T, tasks), - results: make(chan R, tasks), - } - // Start all the data processor routines - if threads == 0 { - threads = runtime.NumCPU() - } - pool.working.Add(threads) - for i := 0; i < threads; i++ { - go pool.work(f) - } - return pool -} - -// Close signals the end of the task stream. It does not block execution, rather -// returns immediately and users have to explicitly call Wait to block until the -// pool actually spins down. Alternatively, consumers can read the results chan, -// which will be closed after the last result is delivered. -// -// Calling Close multiple times will panic. Not particularly hard to avoid, but -// it's really a programming error. -func (pool *WorkerPool[T, R]) Close() { - close(pool.tasks) - go func() { - pool.working.Wait() - close(pool.results) - }() -} - -// Wait blocks until all the scheduled tasks have been processed. -func (pool *WorkerPool[T, R]) Wait() { - pool.working.Wait() -} - -// Schedule adds a task to the work queue. -func (pool *WorkerPool[T, R]) Schedule(task T) { - pool.tasks <- task -} - -// Results retrieves the result channel to consume the output of the individual -// work tasks. The channel will be closed after all tasks are done. -// -// Note, as long as the number of actually scheduled tasks are smaller or equal -// to the requested number form the constructor, it's fine to not consume this -// channel. -func (pool *WorkerPool[T, R]) Results() chan R { - return pool.results -} - -// work is the (one of many) goroutine consuming input tasks and executing them -// to compute the results. -func (pool *WorkerPool[T, R]) work(f func(T) R) { - defer pool.working.Done() - for task := range pool.tasks { - pool.results <- f(task) - } -}