From 59bc98f10853a663700db757aafe50a9b896a236 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?P=C3=A9ter=20Szil=C3=A1gyi?= Date: Mon, 12 Aug 2019 23:48:06 +0300 Subject: [PATCH] core/state: rebase concurrent updater on state accumulator --- core/state/state_object.go | 9 ------- core/state/statedb.go | 55 +++++++++++++++++++++++++++++++++----- core/state/workers.go | 47 ++++++++++++++++++++++++++++++++ 3 files changed, 96 insertions(+), 15 deletions(-) create mode 100644 core/state/workers.go diff --git a/core/state/state_object.go b/core/state/state_object.go index 6c0ce625e7..90a815e338 100644 --- a/core/state/state_object.go +++ b/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() } diff --git a/core/state/statedb.go b/core/state/statedb.go index 56708d85a3..5e645c92da 100644 --- a/core/state/statedb.go +++ b/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()) diff --git a/core/state/workers.go b/core/state/workers.go new file mode 100644 index 0000000000..0e7cab5c9f --- /dev/null +++ b/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 . + +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 +}