forks/go-ethereum
1
0
Fork 1
mirror of https://github.com/ethereum/go-ethereum.git synced 2026-08-19 18:32:23 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 1

core/state: restore async prefetcher stask scheduling

Browse source
This commit is contained in:
Péter Szilágyi 2024-04-12 15:04:06 +03:00
parent 43babe4f66
commit f166ce17e7
5 changed files with 181 additions and 94 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

8
core/blockchain.go
View file

@ -1806,8 +1806,12 @@ func (bc *BlockChain) insertChain(chain types.Blocks, setHead bool) (int, error)
}
statedb.SetLogger(bc.logger)
// Enable prefetching to pull in trie node paths while processing transactions
statedb.StartPrefetcher("chain")
// If we are past Byzantium, enable prefetching to pull in trie node paths
// while processing transactions. Before Byzantium the prefetcher is mostly
// useless due to the intermediate root hashing after each transaction.
if bc.chainConfig.IsByzantium(block.Number()) {
statedb.StartPrefetcher("chain")
}
activeState = statedb
// If we have a followup block, run that against the current state to pre-cache

17
core/state/state_object.go
View file

@ -126,7 +126,12 @@ func (s *stateObject) getTrie() (Trie, error) {
// Try fetching from prefetcher first
if s.data.Root != types.EmptyRootHash && s.db.prefetcher != nil {
// When the miner is creating the pending state, there is no prefetcher
s.trie = s.db.prefetcher.trie(s.addrHash, s.data.Root)
trie, err := s.db.prefetcher.trie(s.addrHash, s.data.Root)
if err != nil {
log.Error("Failed to retrieve storage pre-fetcher trie", "addr", s.address, "err", err)
} else {
s.trie = trie
}
}
if s.trie == nil {
tr, err := s.db.db.OpenStorageTrie(s.db.originalRoot, s.address, s.data.Root, s.db.trie)
@ -253,7 +258,7 @@ func (s *stateObject) setState(key common.Hash, value *common.Hash) {
// finalise moves all dirty storage slots into the pending area to be hashed or
// committed later. It is invoked at the end of every transaction.
func (s *stateObject) finalise(prefetch bool) {
func (s *stateObject) finalise() {
slotsToPrefetch := make([][]byte, 0, len(s.dirtyStorage))
for key, value := range s.dirtyStorage {
// If the slot is different from its original value, move it into the
@ -268,8 +273,10 @@ func (s *stateObject) finalise(prefetch bool) {
delete(s.pendingStorage, key)
}
}
if s.db.prefetcher != nil && prefetch && len(slotsToPrefetch) > 0 && s.data.Root != types.EmptyRootHash {
s.db.prefetcher.prefetch(s.addrHash, s.data.Root, s.address, slotsToPrefetch)
if s.db.prefetcher != nil && len(slotsToPrefetch) > 0 && s.data.Root != types.EmptyRootHash {
if err := s.db.prefetcher.prefetch(s.addrHash, s.data.Root, s.address, slotsToPrefetch); err != nil {
log.Error("Failed to prefetch slots", "addr", s.address, "slots", len(slotsToPrefetch), "err", err)
}
}
if len(s.dirtyStorage) > 0 {
s.dirtyStorage = make(Storage)
@ -288,7 +295,7 @@ func (s *stateObject) finalise(prefetch bool) {
// storage change at all.
func (s *stateObject) updateTrie() (Trie, error) {
// Make sure all dirty slots are finalized into the pending storage area
s.finalise(false)
s.finalise()
// Short circuit if nothing changed, don't bother with hashing anything
if len(s.pendingStorage) == 0 {

50
core/state/statedb.go
View file

@ -210,7 +210,8 @@ func (s *StateDB) SetLogger(l *tracing.Hooks) {
// commit phase, most of the needed data is already hot.
func (s *StateDB) StartPrefetcher(namespace string) {
if s.prefetcher != nil {
s.prefetcher.close()
s.prefetcher.terminate()
s.prefetcher.report()
s.prefetcher = nil
}
if s.snap != nil {
@ -222,7 +223,8 @@ func (s *StateDB) StartPrefetcher(namespace string) {
// from the gathered metrics.
func (s *StateDB) StopPrefetcher() {
if s.prefetcher != nil {
s.prefetcher.close()
s.prefetcher.terminate()
s.prefetcher.report()
s.prefetcher = nil
}
}
@ -809,7 +811,7 @@ func (s *StateDB) Finalise(deleteEmptyObjects bool) {
delete(s.accountsOrigin, obj.address) // Clear out any previously updated account data (may be recreated via a resurrect)
delete(s.storagesOrigin, obj.address) // Clear out any previously updated storage data (may be recreated via a resurrect)
} else {
obj.finalise(true) // Prefetch slots in the background
obj.finalise()
s.markUpdate(addr)
}
// At this point, also ship the address off to the precacher. The precacher
@ -818,7 +820,9 @@ func (s *StateDB) Finalise(deleteEmptyObjects bool) {
addressesToPrefetch = append(addressesToPrefetch, common.CopyBytes(addr[:])) // Copy needed for closure
}
if s.prefetcher != nil && len(addressesToPrefetch) > 0 {
s.prefetcher.prefetch(common.Hash{}, s.originalRoot, common.Address{}, addressesToPrefetch)
if err := s.prefetcher.prefetch(common.Hash{}, s.originalRoot, common.Address{}, addressesToPrefetch); err != nil {
log.Error("Failed to prefetch addresses", "addresses", len(addressesToPrefetch), "err", err)
}
}
// Invalidate journal because reverting across transactions is not allowed.
s.clearJournalAndRefund()
@ -831,18 +835,14 @@ func (s *StateDB) IntermediateRoot(deleteEmptyObjects bool) common.Hash {
// Finalise all the dirty storage states and write them into the tries
s.Finalise(deleteEmptyObjects)
// If there was a trie prefetcher operating, it gets aborted and irrevocably
// modified after we start retrieving tries. Remove it from the statedb after
// this round of use.
//
// This is weird pre-byzantium since the first tx runs with a prefetcher and
// the remainder without, but pre-byzantium even the initial prefetcher is
// useless, so no sleep lost.
prefetcher := s.prefetcher
// If there was a trie prefetcher operating, terminate it (blocking until
// all tasks finish) and then proceed with the trie hashing.
var subfetchers chan *subfetcher
if s.prefetcher != nil {
subfetchers = s.prefetcher.terminateAsync()
defer func() {
s.prefetcher.close()
s.prefetcher = nil
s.prefetcher.report()
s.prefetcher = nil // Pre-byzantium, unset any used up prefetcher
}()
}
// Although naively it makes sense to retrieve the account trie and then do
@ -851,6 +851,18 @@ func (s *StateDB) IntermediateRoot(deleteEmptyObjects bool) common.Hash {
// first, giving the account prefetches just a few more milliseconds of time
// to pull useful data from disk.
start := time.Now()
updated := make(map[common.Address]struct{})
if subfetchers != nil {
for f := range subfetchers {
if op, ok := s.mutations[f.addr]; ok {
if !op.applied && !op.isDelete() {
s.stateObjects[f.addr].updateRoot()
}
updated[f.addr] = struct{}{}
}
}
}
for addr, op := range s.mutations {
if op.applied {
continue
@ -865,8 +877,10 @@ func (s *StateDB) IntermediateRoot(deleteEmptyObjects bool) common.Hash {
// Now we're about to start to write changes to the trie. The trie is so far
// _untouched_. We can check with the prefetcher, if it can give us a trie
// which has the same root, but also has some content loaded into it.
if prefetcher != nil {
if trie := prefetcher.trie(common.Hash{}, s.originalRoot); trie != nil {
if s.prefetcher != nil {
if trie, err := s.prefetcher.trie(common.Hash{}, s.originalRoot); err != nil {
log.Error("Failed to retrieve account pre-fetcher trie", "err", err)
} else if trie != nil {
s.trie = trie
}
}
@ -902,8 +916,8 @@ func (s *StateDB) IntermediateRoot(deleteEmptyObjects bool) common.Hash {
s.deleteStateObject(deletedAddr)
s.AccountDeleted += 1
}
if prefetcher != nil {
prefetcher.used(common.Hash{}, s.originalRoot, usedAddrs)
if s.prefetcher != nil {
s.prefetcher.used(common.Hash{}, s.originalRoot, usedAddrs)
}
// Track the amount of time wasted on hashing the account trie
defer func(start time.Time) { s.AccountHashes += time.Since(start) }(time.Now())

177
core/state/trie_prefetcher.go
View file

@ -29,8 +29,13 @@ var (
// triePrefetchMetricsPrefix is the prefix under which to publish the metrics.
triePrefetchMetricsPrefix = "trie/prefetch/"
// errTerminated is returned if any invocation is applied on a terminated fetcher.
// errTerminated is returned if a fetcher is attempted to be operated after it
// has already terminated.
errTerminated = errors.New("fetcher is already terminated")
// errNotTerminated is returned if a fetchers data is attempted to be retrieved
// before it terminates.
errNotTerminated = errors.New("fetcher is not yet terminated")
)
// triePrefetcher is an active prefetcher, which receives accounts or storage
@ -42,7 +47,7 @@ type triePrefetcher struct {
db Database // Database to fetch trie nodes through
root common.Hash // Root hash of the account trie for metrics
fetchers map[string]*subfetcher // Subfetchers for each trie
closed bool
term chan struct{} // Channel to signal interruption
deliveryMissMeter metrics.Meter
accountLoadMeter metrics.Meter
@ -59,6 +64,7 @@ func newTriePrefetcher(db Database, root common.Hash, namespace string) *triePre
db: db,
root: root,
fetchers: make(map[string]*subfetcher), // Active prefetchers use the fetchers map
term: make(chan struct{}),
deliveryMissMeter: metrics.GetOrRegisterMeter(prefix+"/deliverymiss", nil),
accountLoadMeter: metrics.GetOrRegisterMeter(prefix+"/account/load", nil),
@ -70,36 +76,74 @@ func newTriePrefetcher(db Database, root common.Hash, namespace string) *triePre
}
}
// close iterates over all the subfetchers, waits on any that were left spinning
// and reports the stats to the metrics subsystem.
func (p *triePrefetcher) close() {
// Short circuit if the fetcher is already closed.
if p.closed {
// terminate iterates over all the subfetchers, waiting on any that still spin.
func (p *triePrefetcher) terminate() {
// Short circuit if the fetcher is already closed
select {
case <-p.term:
return
default:
}
// Termiante all sub-fetchers synchronously and close the main fetcher
for _, fetcher := range p.fetchers {
fetcher.terminate()
}
close(p.term)
}
// terminateAsync iterates over all the subfetchers and terminates them async,
// feeding each into a result channel as they finish.
func (p *triePrefetcher) terminateAsync() chan *subfetcher {
// Short circuit if the fetcher is already closed
select {
case <-p.term:
return nil
default:
}
// Terminate all the sub-fetchers asynchronously and feed them into a result
// channel as they finish
var (
res = make(chan *subfetcher, len(p.fetchers))
pend sync.WaitGroup
)
for _, fetcher := range p.fetchers {
pend.Add(1)
go func(f *subfetcher) {
f.terminate()
res <- f
pend.Done()
}(fetcher)
}
go func() {
pend.Wait()
close(res)
}()
close(p.term)
return res
}
// report aggregates the pre-fetching and usage metrics and reports them.
func (p *triePrefetcher) report() {
if !metrics.Enabled {
return
}
for _, fetcher := range p.fetchers {
fetcher.close()
if metrics.Enabled {
if fetcher.root == p.root {
p.accountLoadMeter.Mark(int64(len(fetcher.seen)))
p.accountDupMeter.Mark(int64(fetcher.dups))
for _, key := range fetcher.used {
delete(fetcher.seen, string(key))
}
p.accountWasteMeter.Mark(int64(len(fetcher.seen)))
} else {
p.storageLoadMeter.Mark(int64(len(fetcher.seen)))
p.storageDupMeter.Mark(int64(fetcher.dups))
for _, key := range fetcher.used {
delete(fetcher.seen, string(key))
}
p.storageWasteMeter.Mark(int64(len(fetcher.seen)))
if fetcher.root == p.root {
p.accountLoadMeter.Mark(int64(len(fetcher.seen)))
p.accountDupMeter.Mark(int64(fetcher.dups))
for _, key := range fetcher.used {
delete(fetcher.seen, string(key))
}
p.accountWasteMeter.Mark(int64(len(fetcher.seen)))
} else {
p.storageLoadMeter.Mark(int64(len(fetcher.seen)))
p.storageDupMeter.Mark(int64(fetcher.dups))
for _, key := range fetcher.used {
delete(fetcher.seen, string(key))
}
p.storageWasteMeter.Mark(int64(len(fetcher.seen)))
}
}
p.closed = true
p.fetchers = nil
}
// prefetch schedules a batch of trie items to prefetch. After the prefetcher is
@ -114,8 +158,11 @@ func (p *triePrefetcher) close() {
// repeated.
// 2. Finalize of the main account trie. This happens only once per block.
func (p *triePrefetcher) prefetch(owner common.Hash, root common.Hash, addr common.Address, keys [][]byte) error {
if p.closed {
// Ensure the subfetcher is still alive
select {
case <-p.term:
return errTerminated
default:
}
id := p.trieID(owner, root)
fetcher := p.fetchers[id]
@ -128,15 +175,13 @@ func (p *triePrefetcher) prefetch(owner common.Hash, root common.Hash, addr comm
// trie returns the trie matching the root hash, or nil if either the fetcher
// is terminated or the trie is not available.
func (p *triePrefetcher) trie(owner common.Hash, root common.Hash) Trie {
if p.closed {
return nil
}
func (p *triePrefetcher) trie(owner common.Hash, root common.Hash) (Trie, error) {
// Bail if no trie was prefetched for this root
fetcher := p.fetchers[p.trieID(owner, root)]
if fetcher == nil {
log.Warn("Prefetcher missed to load trie", "owner", owner, "root", root)
p.deliveryMissMeter.Mark(1)
return nil
return nil, nil
}
return fetcher.peek()
}
@ -144,9 +189,6 @@ func (p *triePrefetcher) trie(owner common.Hash, root common.Hash) Trie {
// used marks a batch of state items used to allow creating statistics as to
// how useful or wasteful the fetcher is.
func (p *triePrefetcher) used(owner common.Hash, root common.Hash, used [][]byte) {
if p.closed {
return
}
if fetcher := p.fetchers[p.trieID(owner, root)]; fetcher != nil {
fetcher.used = used
}
@ -175,10 +217,9 @@ type subfetcher struct {
tasks [][]byte // Items queued up for retrieval
lock sync.Mutex // Lock protecting the task queue
wake chan struct{} // Wake channel if a new task is scheduled
stop chan struct{} // Channel to interrupt processing
term chan struct{} // Channel to signal interruption
copy chan chan Trie // channel for retrieving copies of the subfetcher's trie
wake chan struct{} // Wake channel if a new task is scheduled
stop chan struct{} // Channel to interrupt processing
term chan struct{} // Channel to signal interruption
seen map[string]struct{} // Tracks the entries already loaded
dups int // Number of duplicate preload tasks
@ -194,10 +235,9 @@ func newSubfetcher(db Database, state common.Hash, owner common.Hash, root commo
owner: owner,
root: root,
addr: addr,
wake: make(chan struct{}),
wake: make(chan struct{}, 1),
stop: make(chan struct{}),
term: make(chan struct{}),
copy: make(chan chan Trie),
seen: make(map[string]struct{}),
}
go sf.loop()
@ -206,6 +246,12 @@ func newSubfetcher(db Database, state common.Hash, owner common.Hash, root commo
// schedule adds a batch of trie keys to the queue to prefetch.
func (sf *subfetcher) schedule(keys [][]byte) error {
// Ensure the subfetcher is still alive
select {
case <-sf.term:
return errTerminated
default:
}
// Append the tasks to the current queue
sf.lock.Lock()
sf.tasks = append(sf.tasks, keys...)
@ -214,27 +260,31 @@ func (sf *subfetcher) schedule(keys [][]byte) error {
// Notify the background thread to execute scheduled tasks
select {
case sf.wake <- struct{}{}:
return nil
case <-sf.term:
return errTerminated
// Wake signal sent
default:
// Wake signal not sent as a previous is already queued
}
return nil
}
// peek tries to retrieve a deep copy of the fetcher's trie. Nil is returned
// if the fetcher is already terminated, or the associated trie is failing
// for opening.
func (sf *subfetcher) peek() Trie {
ch := make(chan Trie)
// peek retrieves the fetcher's trie, populated with any pre-fetched data. The
// returned trie will be a shallow copy, so modifying it will break subsequent
// peeks for the original data.
//
// This method can only be called after closing the subfetcher.
func (sf *subfetcher) peek() (Trie, error) {
// Ensure the subfetcher finished operating on its trie
select {
case sf.copy <- ch:
return <-ch
case <-sf.term:
return nil
default:
return nil, errNotTerminated
}
return sf.trie, nil
}
// close waits for the subfetcher to finish its tasks. It cannot be called multiple times
func (sf *subfetcher) close() {
// terminate waits for the subfetcher to finish its tasks, after which it tears
// down all the internal background loaders.
func (sf *subfetcher) terminate() {
select {
case <-sf.stop:
default:
@ -249,7 +299,7 @@ func (sf *subfetcher) loop() {
// No matter how the loop stops, signal anyone waiting that it's terminated
defer close(sf.term)
// Start by opening the trie and stop processing if it fails.
// Start by opening the trie and stop processing if it fails
if sf.owner == (common.Hash{}) {
trie, err := sf.db.OpenTrie(sf.root)
if err != nil {
@ -287,13 +337,20 @@ func (sf *subfetcher) loop() {
}
sf.seen[string(task)] = struct{}{}
}
case ch := <-sf.copy:
// Somebody wants a copy of the current trie, grant them.
ch <- sf.db.CopyTrie(sf.trie)
case <-sf.stop:
// Termination is requested, abort
return
// Termination is requested, abort if no more tasks are pending. If
// there are some, exhaust them first.
sf.lock.Lock()
done := sf.tasks == nil
sf.lock.Unlock()
if done {
return
}
// Some tasks are pending, loop and pick them up (that wake branch
// will be selected eventually, whilst stop remains closed to this
// branch will also run afterwards).
}
}
}

23
core/state/trie_prefetcher_test.go
View file

@ -45,18 +45,23 @@ func filledStateDB() *StateDB {
return state
}
func TestUseAfterClose(t *testing.T) {
func TestUseAfterTerminate(t *testing.T) {
db := filledStateDB()
prefetcher := newTriePrefetcher(db.db, db.originalRoot, "")
skey := common.HexToHash("aaa")
prefetcher.prefetch(common.Hash{}, db.originalRoot, common.Address{}, [][]byte{skey.Bytes()})
a := prefetcher.trie(common.Hash{}, db.originalRoot)
prefetcher.close()
b := prefetcher.trie(common.Hash{}, db.originalRoot)
if a == nil {
t.Fatal("Prefetching before close should not return nil")
if err := prefetcher.prefetch(common.Hash{}, db.originalRoot, common.Address{}, [][]byte{skey.Bytes()}); err != nil {
t.Errorf("Prefetch failed before terminate: %v", err)
}
if b != nil {
t.Fatal("Trie after close should return nil")
if _, err := prefetcher.trie(common.Hash{}, db.originalRoot); err == nil {
t.Errorf("Trie retrieval succeeded before terminate")
}
prefetcher.terminate()
if err := prefetcher.prefetch(common.Hash{}, db.originalRoot, common.Address{}, [][]byte{skey.Bytes()}); err == nil {
t.Errorf("Prefetch succeeded after terminate: %v", err)
}
if _, err := prefetcher.trie(common.Hash{}, db.originalRoot); err != nil {
t.Errorf("Trie retrieval failed after terminate: %v", err)
}
}