diff --git a/core/blockchain.go b/core/blockchain.go index 654b4fbdca..564410a1bc 100644 --- a/core/blockchain.go +++ b/core/blockchain.go @@ -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 diff --git a/core/state/state_object.go b/core/state/state_object.go index d75ba01376..e3200815f7 100644 --- a/core/state/state_object.go +++ b/core/state/state_object.go @@ -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 { diff --git a/core/state/statedb.go b/core/state/statedb.go index f4022f7f18..cd91c21196 100644 --- a/core/state/statedb.go +++ b/core/state/statedb.go @@ -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()) diff --git a/core/state/trie_prefetcher.go b/core/state/trie_prefetcher.go index f9d52b9844..c20915cea1 100644 --- a/core/state/trie_prefetcher.go +++ b/core/state/trie_prefetcher.go @@ -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). } } } diff --git a/core/state/trie_prefetcher_test.go b/core/state/trie_prefetcher_test.go index 8a54e4beab..4e5414f0e3 100644 --- a/core/state/trie_prefetcher_test.go +++ b/core/state/trie_prefetcher_test.go @@ -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) } }