From 3ed1810a99a2a7100aa81b4e19cb3d64ec155b2d Mon Sep 17 00:00:00 2001 From: Martin Holst Swende Date: Fri, 7 Feb 2020 16:41:53 +0100 Subject: [PATCH] core/blockchain, core/state: implement new trie prefetcher --- common/types.go | 8 ++ core/blockchain.go | 53 ++++++----- core/state/database.go | 12 ++- core/state/state_object.go | 5 + core/state/statedb.go | 11 +++ core/trie_prefetcher.go | 182 +++++++++++++++++++++++++++++++++++++ 6 files changed, 247 insertions(+), 24 deletions(-) create mode 100644 core/trie_prefetcher.go diff --git a/common/types.go b/common/types.go index cdcc6c20ad..86239eedd5 100644 --- a/common/types.go +++ b/common/types.go @@ -368,3 +368,11 @@ func (ma *MixedcaseAddress) ValidChecksum() bool { func (ma *MixedcaseAddress) Original() string { return ma.original } + +type TriePrefetcher interface { + Pause() + Reset(number uint64, root Hash) + PrefetchAddress(addr Address) + PrefetchStorage(root Hash, slots []Hash) + Close() +} diff --git a/core/blockchain.go b/core/blockchain.go index 0d1c27f95e..3b9e2ed776 100644 --- a/core/blockchain.go +++ b/core/blockchain.go @@ -178,9 +178,9 @@ type BlockChain struct { wg sync.WaitGroup // chain processing wait group for shutting down engine consensus.Engine - validator Validator // Block and state validator interface - prefetcher Prefetcher // Block state prefetcher interface - processor Processor // Block transaction processor interface + validator Validator // Block and state validator interface + prefetcher common.TriePrefetcher // Trie prefetcher interface + processor Processor // Block transaction processor interface vmConfig vm.Config badBlocks *lru.Cache // Bad block cache @@ -228,7 +228,11 @@ func NewBlockChain(db ethdb.Database, cacheConfig *CacheConfig, chainConfig *par badBlocks: badBlocks, } bc.validator = NewBlockValidator(chainConfig, bc, engine) - bc.prefetcher = newStatePrefetcher(chainConfig, bc, engine) + //bc.prefetcher = newStatePrefetcher(chainConfig, bc, engine) + tp := newTriePrefetcher(bc.stateCache) + go tp.loop() + bc.prefetcher = tp + bc.processor = NewStateProcessor(chainConfig, bc, engine) var err error @@ -866,6 +870,9 @@ func (bc *BlockChain) Stop() { atomic.StoreInt32(&bc.procInterrupt, 1) bc.wg.Wait() + if bc.prefetcher != nil { + bc.prefetcher.Close() + } // Ensure that the entirety of the state snapshot is journalled to disk. var snapBase common.Hash @@ -1690,31 +1697,33 @@ func (bc *BlockChain) insertChain(chain types.Blocks, verifySeals bool) (int, er parent = bc.GetHeader(block.ParentHash(), block.NumberU64()-1) } statedb, err := state.New(parent.Root, bc.stateCache, bc.snaps) + bc.prefetcher.Reset(block.NumberU64(), parent.Root) + statedb.UsePrefetcher(bc.prefetcher) if err != nil { return it.index, err } // If we have a followup block, run that against the current state to pre-cache // transactions and probabilistically some of the account/storage trie nodes. - var followupInterrupt uint32 - if !bc.cacheConfig.TrieCleanNoPrefetch { - if followup, err := it.peek(); followup != nil && err == nil { - throwaway, _ := state.New(parent.Root, bc.stateCache, bc.snaps) - go func(start time.Time, followup *types.Block, throwaway *state.StateDB, interrupt *uint32) { - bc.prefetcher.Prefetch(followup, throwaway, bc.vmConfig, &followupInterrupt) - - blockPrefetchExecuteTimer.Update(time.Since(start)) - if atomic.LoadUint32(interrupt) == 1 { - blockPrefetchInterruptMeter.Mark(1) - } - }(time.Now(), followup, throwaway, &followupInterrupt) - } - } + //var followupInterrupt uint32 + //if !bc.cacheConfig.TrieCleanNoPrefetch { + // if followup, err := it.peek(); followup != nil && err == nil { + // throwaway, _ := state.New(parent.Root, bc.stateCache, bc.snaps) + // go func(start time.Time, followup *types.Block, throwaway *state.StateDB, interrupt *uint32) { + // bc.prefetcher.Prefetch(followup, throwaway, bc.vmConfig, &followupInterrupt) + // + // blockPrefetchExecuteTimer.Update(time.Since(start)) + // if atomic.LoadUint32(interrupt) == 1 { + // blockPrefetchInterruptMeter.Mark(1) + // } + // }(time.Now(), followup, throwaway, &followupInterrupt) + // } + //} // Process block using the parent state as reference point substart := time.Now() receipts, logs, usedGas, err := bc.processor.Process(block, statedb, bc.vmConfig) if err != nil { bc.reportBlock(block, receipts, err) - atomic.StoreUint32(&followupInterrupt, 1) + //atomic.StoreUint32(&followupInterrupt, 1) return it.index, err } // Update the metrics touched during block processing @@ -1724,7 +1733,7 @@ func (bc *BlockChain) insertChain(chain types.Blocks, verifySeals bool) (int, er storageUpdateTimer.Update(statedb.StorageUpdates) // Storage updates are complete, we can mark them snapshotAccountReadTimer.Update(statedb.SnapshotAccountReads) // Account reads are complete, we can mark them snapshotStorageReadTimer.Update(statedb.SnapshotStorageReads) // Storage reads are complete, we can mark them - + bc.prefetcher.Pause() triehash := statedb.AccountHashes + statedb.StorageHashes // Save to not double count in validation trieproc := statedb.SnapshotAccountReads + statedb.AccountReads + statedb.AccountUpdates trieproc += statedb.SnapshotStorageReads + statedb.StorageReads + statedb.StorageUpdates @@ -1735,7 +1744,7 @@ func (bc *BlockChain) insertChain(chain types.Blocks, verifySeals bool) (int, er substart = time.Now() if err := bc.validator.ValidateState(block, statedb, receipts, usedGas); err != nil { bc.reportBlock(block, receipts, err) - atomic.StoreUint32(&followupInterrupt, 1) + //atomic.StoreUint32(&followupInterrupt, 1) return it.index, err } proctime := time.Since(start) @@ -1749,7 +1758,7 @@ func (bc *BlockChain) insertChain(chain types.Blocks, verifySeals bool) (int, er // Write the block to the chain and get the status. substart = time.Now() status, err := bc.writeBlockWithState(block, receipts, logs, statedb, false) - atomic.StoreUint32(&followupInterrupt, 1) + //atomic.StoreUint32(&followupInterrupt, 1) if err != nil { return it.index, err } diff --git a/core/state/database.go b/core/state/database.go index ecc2c134da..3dd3bab84f 100644 --- a/core/state/database.go +++ b/core/state/database.go @@ -121,12 +121,20 @@ type cachingDB struct { // OpenTrie opens the main account trie at a specific root hash. func (db *cachingDB) OpenTrie(root common.Hash) (Trie, error) { - return trie.NewSecure(root, db.db) + tr, err := trie.NewSecure(root, db.db) + if err != nil { + return nil, err + } + return tr, nil } // OpenStorageTrie opens the storage trie of an account. func (db *cachingDB) OpenStorageTrie(addrHash, root common.Hash) (Trie, error) { - return trie.NewSecure(root, db.db) + tr, err := trie.NewSecure(root, db.db) + if err != nil { + return nil, err + } + return tr, nil } // CopyTrie returns an independent copy of the given trie. diff --git a/core/state/state_object.go b/core/state/state_object.go index 0e88d53c44..5911594149 100644 --- a/core/state/state_object.go +++ b/core/state/state_object.go @@ -301,8 +301,13 @@ func (s *stateObject) setState(key, 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() { + trieChanges := make([]common.Hash, 0, len(s.dirtyStorage)) for key, value := range s.dirtyStorage { s.pendingStorage[key] = value + trieChanges = append(trieChanges, key) + } + if len(trieChanges) > 0 && s.db.prefetcher != nil { + s.db.prefetcher.PrefetchStorage(s.data.Root, trieChanges) } if len(s.dirtyStorage) > 0 { s.dirtyStorage = make(Storage) diff --git a/core/state/statedb.go b/core/state/statedb.go index 9bbb49c50b..cca6abfb3f 100644 --- a/core/state/statedb.go +++ b/core/state/statedb.go @@ -65,6 +65,7 @@ func (n *proofList) Delete(key []byte) error { // * Accounts type StateDB struct { db Database + prefetcher common.TriePrefetcher trie Trie hasher crypto.KeccakHasher @@ -144,6 +145,10 @@ func New(root common.Hash, db Database, snaps *snapshot.Tree) (*StateDB, error) return sdb, nil } +func (s *StateDB) UsePrefetcher(prefetcher common.TriePrefetcher){ + s.prefetcher = prefetcher +} + // setError remembers the first non-nil error it is called with. func (s *StateDB) setError(err error) { if s.dbErr == nil { @@ -758,6 +763,12 @@ func (s *StateDB) Finalise(deleteEmptyObjects bool) { } s.stateObjectsPending[addr] = struct{}{} s.stateObjectsDirty[addr] = struct{}{} + // At this point, also ship the address off to the precacher. The precacher + // will start loading tries, and when the change is eventually committed, + // the commit-phase will be a lot faster + if s.prefetcher != nil{ + s.prefetcher.PrefetchAddress(addr) + } } // Invalidate journal because reverting across transactions is not allowed. s.clearJournalAndRefund() diff --git a/core/trie_prefetcher.go b/core/trie_prefetcher.go new file mode 100644 index 0000000000..27cfa7fe76 --- /dev/null +++ b/core/trie_prefetcher.go @@ -0,0 +1,182 @@ +// Copyright 2020 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 core + +import ( + "sync/atomic" + + "github.com/ethereum/go-ethereum/common" + "github.com/ethereum/go-ethereum/core/state" + "github.com/ethereum/go-ethereum/log" + "github.com/ethereum/go-ethereum/metrics" +) + +var ( + triePrefetchFetchMeter = metrics.NewRegisteredMeter("trie/prefetch/fetch", nil) + triePrefetchSkipMeter = metrics.NewRegisteredMeter("trie/prefetch/skip", nil) + triePrefetchDropMeter = metrics.NewRegisteredMeter("trie/prefetch/drop", nil) +) + +// triePrefetcher is an active prefetcher, which receives accounts or storage +// items on two channels, and does trie-loading of the items. +// The goal is to get as much useful content into the caches as possible +type triePrefetcher struct { + cmdCh chan (command) + abortCh chan (struct{}) + db state.Database + stale uint64 +} + +func newTriePrefetcher(db state.Database) *triePrefetcher { + return &triePrefetcher{ + cmdCh: make(chan command, 200), + abortCh: make(chan struct{}), + db: db, + } +} + +type command struct { + root *common.Hash + address *common.Address + slots []common.Hash +} + +func (p *triePrefetcher) loop() { + var ( + tr state.Trie + err error + currentRoot common.Hash + // Some tracking of performance + skipped int64 + fetched int64 + ) + for { + select { + case cmd := <-p.cmdCh: + // New roots are sent synchoronously + if cmd.root != nil && cmd.slots == nil { + // Update metrics at new block events + triePrefetchFetchMeter.Mark(fetched) + fetched = 0 + triePrefetchSkipMeter.Mark(skipped) + skipped = 0 + // New root and number + currentRoot = *cmd.root + tr, err = p.db.OpenTrie(currentRoot) + if err != nil { + log.Warn("trie prefetcher failed opening trie", "root", currentRoot, "err", err) + } + // Open for business again + atomic.StoreUint64(&p.stale, 0) + continue + } + // Don't get stuck precaching on old blocks + if atomic.LoadUint64(&p.stale) == 1 { + if nSlots := len(cmd.slots); nSlots > 0 { + skipped += int64(nSlots) + } else { + skipped++ + } + // Keep reading until we're in step with the chain + continue + } + // It's either storage slots or an account + if cmd.slots != nil { + storageTrie, err := p.db.OpenTrie(*cmd.root) + if err != nil { + log.Warn("trie prefetcher failed opening storage trie", "root", *cmd.root, "err", err) + skipped += int64(len(cmd.slots)) + continue + } + for i, key := range cmd.slots { + storageTrie.TryGet(key[:]) + fetched++ + // Abort if we fall behind + if atomic.LoadUint64(&p.stale) == 1 { + skipped += int64(len(cmd.slots[i:])) + break + } + } + } else { // an account + if tr == nil { + skipped++ + continue + } + // We're in sync with the chain, do preloading + if cmd.address != nil { + fetched++ + addr := *cmd.address + tr.TryGet(addr[:]) + } + } + case <-p.abortCh: + return + } + } +} + +// Close stops the prefetcher +func (p *triePrefetcher) Close() { + p.abortCh <- struct{}{} +} + +// Reset prevent the prefetcher from entering a state where it is +// behind the actual block processing. +// It causes any existing (stale) work to be ignored, and the prefetcher will skip ahead +// to current tasks +func (p *triePrefetcher) Reset(number uint64, root common.Hash) { + // Set staleness + atomic.StoreUint64(&p.stale, 1) + // Do a synced send, so we're sure it punches through any old (now stale) commands + cmd := command{ + root: &root, + } + p.cmdCh <- cmd +} + +func (p *triePrefetcher) Pause() { + // Set staleness + atomic.StoreUint64(&p.stale, 1) +} + +// PrefetchAddress adds an address for prefetching +func (p *triePrefetcher) PrefetchAddress(addr common.Address) { + cmd := command{ + address: &addr, + } + // We do an async send here, to not cause the caller to block + select { + case p.cmdCh <- cmd: + default: + triePrefetchDropMeter.Mark(1) + } +} + +// PrefetchStorage adds a storage root and a set of keys for prefetching +func (p *triePrefetcher) PrefetchStorage(root common.Hash, slots []common.Hash) { + cmd := command{ + root: &root, + slots: slots, + } + // We do an async send here, to not cause the caller to block + select { + case p.cmdCh <- cmd: + default: + triePrefetchDropMeter.Mark(int64(len(slots))) + } + +}