From 4943bd8bbb3dbee33ce7b54623305609ed2d2b00 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?P=C3=A9ter=20Szil=C3=A1gyi?= Date: Mon, 12 Feb 2024 17:49:54 +0200 Subject: [PATCH 1/3] core/state: minor prefetcher polishes --- core/state/state_object.go | 13 +++----- core/state/statedb.go | 61 +++++++++++------------------------ core/state/trie_prefetcher.go | 54 ++++--------------------------- 3 files changed, 29 insertions(+), 99 deletions(-) diff --git a/core/state/state_object.go b/core/state/state_object.go index e7c21e6881..778356f0a3 100644 --- a/core/state/state_object.go +++ b/core/state/state_object.go @@ -194,13 +194,6 @@ func (s *stateObject) GetCommittedState(key common.Hash) common.Hash { err error value common.Hash ) - - if s.db.witness != nil && s.db.snap != nil && s.origin != nil { - // when building a witness with snapshot enabled, prefetch all read slots to be collected - // and included in the witness when the block root hash is committed (intermediateroot/commit?) - s.db.readPrefetcher.prefetch(s.addrHash, s.origin.Root, s.address, [][]byte{key[:]}) - } - if s.db.snap != nil { start := time.Now() enc, err = s.db.snap.Storage(s.addrHash, crypto.Keccak256Hash(key.Bytes())) @@ -214,6 +207,11 @@ func (s *stateObject) GetCommittedState(key common.Hash) common.Hash { } value.SetBytes(content) } + // If witness building is enabled, prefetch any trie paths loaded directly + // via the snapshots + if s.db.prefetcher != nil && err == nil && s.db.witness != nil && s.data.Root != types.EmptyRootHash { + s.db.prefetcher.prefetch(s.addrHash, s.origin.Root, s.address, [][]byte{key[:]}) + } } // If the snapshot is unavailable or reading from it fails, load from the database. if s.db.snap == nil || err != nil { @@ -224,7 +222,6 @@ func (s *stateObject) GetCommittedState(key common.Hash) common.Hash { return common.Hash{} } val, err := tr.GetStorage(s.address, key.Bytes()) - //fmt.Printf("trie access list is %v\n", tr.AccessList()) if metrics.EnabledExpensive { s.db.StorageReads += time.Since(start) } diff --git a/core/state/statedb.go b/core/state/statedb.go index 53d194aba5..646216c194 100644 --- a/core/state/statedb.go +++ b/core/state/statedb.go @@ -62,6 +62,7 @@ type StateDB struct { db Database prefetcher *triePrefetcher trie Trie + witness *Witness hasher crypto.KeccakState snaps *snapshot.Tree // Nil if snapshot is not available snap snapshot.Snapshot // Nil if snapshot is not available @@ -138,9 +139,6 @@ type StateDB struct { // Testing hooks onCommit func(states *triestate.Set) // Hook invoked when commit is performed - - witness *Witness - readPrefetcher *triePrefetcher } // NewWithWitnessRecording creates a new state from a given trie. The state is configured to construct a stateless @@ -191,20 +189,11 @@ func New(root common.Hash, db Database, snaps *snapshot.Tree) (*StateDB, error) // commit phase, most of the needed data is already hot. func (s *StateDB) StartPrefetcher(namespace string) { if s.prefetcher != nil { - s.prefetcher.wait() s.prefetcher.close() s.prefetcher = nil } - if s.readPrefetcher != nil { - s.readPrefetcher.wait() - s.readPrefetcher.close() - s.readPrefetcher = nil - } if s.snap != nil { s.prefetcher = newTriePrefetcher(s.db, s.originalRoot, namespace) - if s.witness != nil { - s.readPrefetcher = newTriePrefetcher(s.db, s.originalRoot, namespace) - } } } @@ -212,15 +201,9 @@ func (s *StateDB) StartPrefetcher(namespace string) { // from the gathered metrics. func (s *StateDB) StopPrefetcher() { if s.prefetcher != nil { - s.prefetcher.wait() s.prefetcher.close() s.prefetcher = nil } - if s.readPrefetcher != nil { - s.readPrefetcher.wait() - s.readPrefetcher.close() - s.readPrefetcher = nil - } } // setError remembers the first non-nil error it is called with. @@ -601,6 +584,11 @@ func (s *StateDB) getDeletedStateObject(addr common.Address) *stateObject { s.SnapshotAccountReads += time.Since(start) } if err == nil { + // If witness building is enabled, prefetch any trie paths loaded directly + // via the snapshots + if s.prefetcher != nil && s.witness != nil { + s.prefetcher.prefetch(common.Hash{}, s.originalRoot, common.Address{}, [][]byte{addr[:]}) + } if acc == nil { return nil } @@ -634,15 +622,9 @@ func (s *StateDB) getDeletedStateObject(addr common.Address) *stateObject { return nil } } - // Insert into the live set obj := newObject(s, addr, data) s.setStateObject(obj) - if s.witness != nil && s.snap != nil { - // when building witness with snap enabled, prefetch all read accounts to later be collected and - // included in the witness - s.readPrefetcher.prefetch(common.Hash{}, s.originalRoot, common.Address{}, [][]byte{addr[:]}) - } return obj } @@ -719,7 +701,10 @@ func (s *StateDB) CreateAccount(addr common.Address) { } // Copy creates a deep, independent copy of the state. -// Snapshots of the copied state cannot be applied to the copy. +// +// Note: +// - Snapshots of the copied state cannot be applied to the copy. +// - Pre-fetchers will not be copied nor active in the copy. func (s *StateDB) Copy() *StateDB { // Copy all the basic fields, initialize the memory ones state := &StateDB{ @@ -816,12 +801,6 @@ func (s *StateDB) Copy() *StateDB { state.accessList = s.accessList.Copy() state.transientStorage = s.transientStorage.Copy() - // If there's a prefetcher running, make an inactive copy of it that can - // only access data but does not actively preload (since the user will not - // know that they need to explicitly terminate an active copy). - if s.prefetcher != nil { - state.prefetcher = s.prefetcher.copy() - } return state } @@ -909,7 +888,7 @@ func (s *StateDB) collectReadStorageAccessLists() { for _, obj := range s.stateObjects { // load read storage slots from the finished trie in the prefetcher as these continue to be prefetched // until commit. - tr := s.readPrefetcher.trie(obj.addrHash, obj.data.Root) + tr := s.prefetcher.trie(obj.addrHash, obj.data.Root) if tr == nil { continue } @@ -921,7 +900,7 @@ func (s *StateDB) collectReadStorageAccessLists() { } func (s *StateDB) collectReadAccountsAccessLists() { - tr := s.readPrefetcher.trie(common.Hash{}, s.originalRoot) + tr := s.prefetcher.trie(common.Hash{}, s.originalRoot) if tr == nil { // TODO: ensure this case is b/c of empty block return @@ -943,20 +922,16 @@ func (s *StateDB) IntermediateRoot(deleteEmptyObjects bool) common.Hash { if s.prefetcher != nil { defer func() { // TODO: need to wait for read accounts to be resolved in prefetcher main trie? - s.prefetcher.wait() s.prefetcher.close() + + if s.witness != nil { + // TODO: move read prefetcher logic into Commit? + s.collectReadStorageAccessLists() + s.collectReadAccountsAccessLists() + } s.prefetcher = nil }() } - if s.readPrefetcher != nil { - // TODO: move read prefetcher logic into Commit? - s.readPrefetcher.wait() - s.collectReadStorageAccessLists() - s.collectReadAccountsAccessLists() - s.readPrefetcher.close() - s.readPrefetcher = nil - } - // Although naively it makes sense to retrieve the account trie and then do // the contract storage and account updates sequentially, that short circuits // the account prefetcher. Instead, let's process all the storage updates diff --git a/core/state/trie_prefetcher.go b/core/state/trie_prefetcher.go index e624e65162..6e3a594bd6 100644 --- a/core/state/trie_prefetcher.go +++ b/core/state/trie_prefetcher.go @@ -38,7 +38,6 @@ var ( type triePrefetcher struct { db Database // Database to fetch trie nodes through root common.Hash // Root hash of the account trie for metrics - fetches map[string]Trie // Partially or fully fetched tries. Only populated for inactive copies. fetchers map[string]*subfetcher // Subfetchers for each trie deliveryMissMeter metrics.Meter @@ -71,16 +70,13 @@ func newTriePrefetcher(db Database, root common.Hash, namespace string) *triePre } return p } -func (p *triePrefetcher) wait() { - for _, fetcher := range p.fetchers { - fetcher.wait() - } -} -// close iterates over all the subfetchers, aborts any that were left spinning +// 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() { for _, fetcher := range p.fetchers { + fetcher.wait() // safe to do multiple times + if metrics.Enabled { if fetcher.root == p.root { p.accountLoadMeter.Mark(int64(len(fetcher.seen))) @@ -107,30 +103,8 @@ func (p *triePrefetcher) close() { p.fetchers = nil } -// copy creates a deep-but-inactive copy of the trie prefetcher. Any trie data -// already loaded will be copied over, but no goroutines will be started. This -// is mostly used in the miner which creates a copy of it's actively mutated -// state to be sealed while it may further mutate the state. -func (p *triePrefetcher) copy() *triePrefetcher { - copy := &triePrefetcher{ - db: p.db, - root: p.root, - fetches: make(map[string]Trie), // Active prefetchers use the fetches map - - deliveryMissMeter: p.deliveryMissMeter, - accountLoadMeter: p.accountLoadMeter, - accountDupMeter: p.accountDupMeter, - accountSkipMeter: p.accountSkipMeter, - accountWasteMeter: p.accountWasteMeter, - storageLoadMeter: p.storageLoadMeter, - storageDupMeter: p.storageDupMeter, - storageSkipMeter: p.storageSkipMeter, - storageWasteMeter: p.storageWasteMeter, - } - return copy -} - // prefetch schedules a batch of trie items to prefetch. +// // prefetch is called from two locations: // 1. Finalize of the state-objects storage roots. This happens at the end // of every transaction, meaning that if several transactions touches @@ -138,12 +112,6 @@ func (p *triePrefetcher) copy() *triePrefetcher { // 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) { - // If the prefetcher is an inactive one, bail out - if p.fetches != nil { - return - } - - // Active fetcher, schedule the retrievals id := p.trieID(owner, root) fetcher := p.fetchers[id] if fetcher == nil { @@ -156,18 +124,8 @@ func (p *triePrefetcher) prefetch(owner common.Hash, root common.Hash, addr comm // trie returns the trie matching the root hash, or nil if the prefetcher doesn't // have it. func (p *triePrefetcher) trie(owner common.Hash, root common.Hash) Trie { - // If the prefetcher is inactive, return from existing deep copies - id := p.trieID(owner, root) - if p.fetches != nil { - trie := p.fetches[id] - if trie == nil { - p.deliveryMissMeter.Mark(1) - return nil - } - return p.db.CopyTrie(trie) - } - // Otherwise the prefetcher is active, bail if no trie was prefetched for this root - fetcher := p.fetchers[id] + // Bail if no trie was prefetched for this root + fetcher := p.fetchers[p.trieID(owner, root)] if fetcher == nil { p.deliveryMissMeter.Mark(1) return nil From 494ba791a17d1c923801fa7df23936c0be2e96be Mon Sep 17 00:00:00 2001 From: Jared Wasinger Date: Tue, 13 Feb 2024 07:44:55 -0800 Subject: [PATCH 2/3] ensure non-mutated storage slots are included in witness when snapshot is enabled --- core/state/statedb.go | 5 ++--- core/state/trie_prefetcher.go | 8 ++++++++ 2 files changed, 10 insertions(+), 3 deletions(-) diff --git a/core/state/statedb.go b/core/state/statedb.go index 646216c194..8aa610c1b6 100644 --- a/core/state/statedb.go +++ b/core/state/statedb.go @@ -921,14 +921,13 @@ func (s *StateDB) IntermediateRoot(deleteEmptyObjects bool) common.Hash { prefetcher := s.prefetcher if s.prefetcher != nil { defer func() { - // TODO: need to wait for read accounts to be resolved in prefetcher main trie? - s.prefetcher.close() - + s.prefetcher.wait() if s.witness != nil { // TODO: move read prefetcher logic into Commit? s.collectReadStorageAccessLists() s.collectReadAccountsAccessLists() } + s.prefetcher.close() s.prefetcher = nil }() } diff --git a/core/state/trie_prefetcher.go b/core/state/trie_prefetcher.go index 6e3a594bd6..f8258b51ae 100644 --- a/core/state/trie_prefetcher.go +++ b/core/state/trie_prefetcher.go @@ -210,6 +210,14 @@ func (sf *subfetcher) schedule(keys [][]byte) { } } +// wait instructs all subfetchers to finish their tasks +// and stop receiving new requests. +func (p *triePrefetcher) wait() { + for _, fetcher := range p.fetchers { + fetcher.wait() + } +} + // wait waits for the subfetcher to finish it's task. It is safe to call wait multiple // times but it is not thread safe. func (sf *subfetcher) wait() { From 259f3e2b4ebb7087ae5d04c3b2b5df99b0caad31 Mon Sep 17 00:00:00 2001 From: Jared Wasinger Date: Wed, 14 Feb 2024 19:19:17 -0800 Subject: [PATCH 3/3] remove wait function from trie prefetcher. make close not remove fetchers. Fix stateless blockchain test-runner. --- core/state/statedb.go | 3 +-- core/state/trie_prefetcher.go | 13 ++----------- tests/block_test.go | 8 ++++---- 3 files changed, 7 insertions(+), 17 deletions(-) diff --git a/core/state/statedb.go b/core/state/statedb.go index 8aa610c1b6..1ae52ab632 100644 --- a/core/state/statedb.go +++ b/core/state/statedb.go @@ -921,13 +921,12 @@ func (s *StateDB) IntermediateRoot(deleteEmptyObjects bool) common.Hash { prefetcher := s.prefetcher if s.prefetcher != nil { defer func() { - s.prefetcher.wait() + s.prefetcher.close() if s.witness != nil { // TODO: move read prefetcher logic into Commit? s.collectReadStorageAccessLists() s.collectReadAccountsAccessLists() } - s.prefetcher.close() s.prefetcher = nil }() } diff --git a/core/state/trie_prefetcher.go b/core/state/trie_prefetcher.go index f8258b51ae..c6332576b3 100644 --- a/core/state/trie_prefetcher.go +++ b/core/state/trie_prefetcher.go @@ -72,7 +72,8 @@ 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. +// and reports the stats to the metrics subsystem. close should not be called +// more than once on a triePrefetcher instance. func (p *triePrefetcher) close() { for _, fetcher := range p.fetchers { fetcher.wait() // safe to do multiple times @@ -99,8 +100,6 @@ func (p *triePrefetcher) close() { } } } - // Clear out all fetchers (will crash on a second call, deliberate) - p.fetchers = nil } // prefetch schedules a batch of trie items to prefetch. @@ -210,14 +209,6 @@ func (sf *subfetcher) schedule(keys [][]byte) { } } -// wait instructs all subfetchers to finish their tasks -// and stop receiving new requests. -func (p *triePrefetcher) wait() { - for _, fetcher := range p.fetchers { - fetcher.wait() - } -} - // wait waits for the subfetcher to finish it's task. It is safe to call wait multiple // times but it is not thread safe. func (sf *subfetcher) wait() { diff --git a/tests/block_test.go b/tests/block_test.go index 744b588f0b..fca3458388 100644 --- a/tests/block_test.go +++ b/tests/block_test.go @@ -194,21 +194,21 @@ func execBlockTestStateless(t *testing.T, bt *testMatcher, test *BlockTest) { tracer := logger.NewJSONLogger(logconfig, os.Stdout) _ = tracer - if err := bt.checkFailure(t, test.RunStateless(false, rawdb.HashScheme, nil)); err != nil { + if err := bt.checkFailure(t, test.RunStateless(false, rawdb.HashScheme, nil, nil)); err != nil { t.Errorf("test in hash mode without snapshotter failed: %v", err) return } - if err := bt.checkFailure(t, test.RunStateless(true, rawdb.HashScheme, nil)); err != nil { + if err := bt.checkFailure(t, test.RunStateless(true, rawdb.HashScheme, nil, nil)); err != nil { t.Errorf("test in hash mode with snapshotter failed: %v", err) return } - if err := bt.checkFailure(t, test.RunStateless(false, rawdb.PathScheme, nil)); err != nil { + if err := bt.checkFailure(t, test.RunStateless(false, rawdb.PathScheme, nil, nil)); err != nil { t.Errorf("test in path mode without snapshotter failed: %v", err) return } - if err := bt.checkFailure(t, test.RunStateless(true, rawdb.PathScheme, nil)); err != nil { + if err := bt.checkFailure(t, test.RunStateless(true, rawdb.PathScheme, nil, nil)); err != nil { t.Errorf("test in path mode with snapshotter failed: %v", err) return }