core/blockchain, core/state: implement new trie prefetcher

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()
+}

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
 		}

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.

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)

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()

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 <http://www.gnu.org/licenses/>.
+
+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)))
+	}
+
+}