core/state: rebase concurrent updater on state accumulator

core/state/state_object.go

@@ -276,10 +276,6 @@ func (s *stateObject) updateTrie(db Database) Trie {
 	// Make sure all dirty slots are finalized into the pending storage area
 	s.finalise()
 
-	// Track the amount of time wasted on updating the storge trie
-	if metrics.EnabledExpensive {
-		defer func(start time.Time) { s.db.StorageUpdates += time.Since(start) }(time.Now())
-	}
 	// Insert all the pending updates into the trie
 	tr := s.getTrie(db)
 	for key, value := range s.pendingStorage {
@@ -306,11 +302,6 @@ func (s *stateObject) updateTrie(db Database) Trie {
 // UpdateRoot sets the trie root to the current root hash of
 func (s *stateObject) updateRoot(db Database) {
 	s.updateTrie(db)
-
-	// Track the amount of time wasted on hashing the storge trie
-	if metrics.EnabledExpensive {
-		defer func(start time.Time) { s.db.StorageHashes += time.Since(start) }(time.Now())
-	}
 	s.data.Root = s.trie.Hash()
 }
 

core/state/statedb.go

@@ -22,6 +22,7 @@ import (
 	"fmt"
 	"math/big"
 	"sort"
+	"sync/atomic"
 	"time"
 
 	"github.com/ethereum/go-ethereum/common"
@@ -681,21 +682,63 @@ func (s *StateDB) Finalise(deleteEmptyObjects bool) {
 // It is called in between transactions to get the root hash that
 // goes into transaction receipts.
 func (s *StateDB) IntermediateRoot(deleteEmptyObjects bool) common.Hash {
-	// Finalise all the dirty storage states and write them into the tries
+	// Finalise all the dirty storage states and prepare them for writing
 	s.Finalise(deleteEmptyObjects)
 
+	// Storage tries are independent, so update them in the background
+	var (
+		deleted = make([]*stateObject, 0, len(s.stateObjectsPending))
+		changed = make([]*stateObject, 0, len(s.stateObjectsPending)) // Only account updated
+		updated = make([]*stateObject, 0, len(s.stateObjectsPending)) // Storage updates
+	)
 	for addr := range s.stateObjectsPending {
 		obj := s.stateObjects[addr]
-		if obj.deleted {
-			s.deleteStateObject(obj)
-		} else {
-			obj.updateRoot(s.db)
-			s.updateStateObject(obj)
+		switch {
+		case obj.deleted:
+			deleted = append(deleted, obj)
+		case len(obj.pendingStorage) == 0:
+			changed = append(changed, obj)
+		default:
+			updated = append(updated, obj)
 		}
 	}
 	if len(s.stateObjectsPending) > 0 {
 		s.stateObjectsPending = make(map[common.Address]struct{})
 	}
+	done := make(chan *stateObject, len(updated))
+
+	// Beside running the updates in the background, track the longest running one
+	// so we can report it as a storage update metric.
+	start := time.Now()
+	var end atomic.Value
+	end.Store(start)
+
+	for _, obj := range updated {
+		obj := obj // Closure, take care
+		workers.Schedule(func() {
+			obj.updateRoot(s.db)
+			if metrics.EnabledExpensive {
+				end.Store(time.Now())
+			}
+			done <- obj
+		})
+	}
+	// Storage tries are updating in the background, concurrently remove deleted
+	// and naively updated accounts from the main trie ,then progress to handling
+	// the storage updates as they are finishing.
+	for _, obj := range deleted {
+		s.deleteStateObject(obj)
+	}
+	for _, obj := range changed {
+		s.updateStateObject(obj)
+	}
+	for range updated {
+		s.updateStateObject(<-done)
+	}
+	// Track the amount of time wasted on updating the storge tries
+	if metrics.EnabledExpensive {
+		s.StorageUpdates += end.Load().(time.Time).Sub(start) // TODO(karalabe): StorageHashes is not counted now
+	}
 	// Track the amount of time wasted on hashing the account trie
 	if metrics.EnabledExpensive {
 		defer func(start time.Time) { s.AccountHashes += time.Since(start) }(time.Now())

core/state/workers.go

@@ -0,0 +1,47 @@
+// Copyright 2019 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 <http://www.gnu.org/licenses/>.
+
+package state
+
+import "runtime"
+
+// workers is a singleton pool of goroutines to run arbitrary tasks concurrently.
+var workers = newWorkerPool(16 * runtime.GOMAXPROCS(0))
+
+// workerPool is a set of goroutines that execute arbitrary tasks concurrently.
+type workerPool struct {
+	tasks chan func()
+}
+
+// newWorkerPool creates the worker task channel and launches the goroutines.
+func newWorkerPool(queue int) *workerPool {
+	pool := &workerPool{
+		tasks: make(chan func(), queue),
+	}
+	for i := 0; i < runtime.GOMAXPROCS(0); i++ {
+		go func() {
+			for task := range pool.tasks {
+				task()
+			}
+		}()
+	}
+	return pool
+}
+
+// Schedule inserts a new task into the worker pool.
+func (pool *workerPool) Schedule(fn func()) {
+	pool.tasks <- fn
+}