From dbf6d0460b3d74cf8aac0b0376333a265b722235 Mon Sep 17 00:00:00 2001 From: Gary Rong Date: Wed, 18 Jun 2025 15:58:56 +0800 Subject: [PATCH] triedb/pathdb: polish --- core/state/snapshot/generate_test.go | 13 +------ triedb/pathdb/buffer.go | 16 ++++++-- triedb/pathdb/database.go | 18 ++++----- triedb/pathdb/disklayer.go | 56 +++++++++++++++++----------- triedb/pathdb/journal.go | 6 +-- triedb/pathdb/layertree.go | 4 -- 6 files changed, 56 insertions(+), 57 deletions(-) diff --git a/core/state/snapshot/generate_test.go b/core/state/snapshot/generate_test.go index d5cf6dce31..4488630095 100644 --- a/core/state/snapshot/generate_test.go +++ b/core/state/snapshot/generate_test.go @@ -168,6 +168,7 @@ func newHelper(scheme string) *testHelper { if scheme == rawdb.PathScheme { config.PathDB = &pathdb.Config{ SnapshotNoBuild: true, + NoAsyncFlush: true, } // disable caching } else { config.HashDB = &hashdb.Config{} // disable caching @@ -242,18 +243,6 @@ func (t *testHelper) Commit() common.Hash { } t.triedb.Update(root, types.EmptyRootHash, 0, t.nodes, t.states) t.triedb.Commit(root, false) - - // re-open the trie database to ensure the frozen buffer - // is not referenced - //config := &triedb.Config{} - //if t.triedb.Scheme() == rawdb.PathScheme { - // config.PathDB = &pathdb.Config{ - // SnapshotNoBuild: true, - // } // disable caching - //} else { - // config.HashDB = &hashdb.Config{} // disable caching - //} - //t.triedb = triedb.NewDatabase(t.triedb.Disk(), config) return root } diff --git a/triedb/pathdb/buffer.go b/triedb/pathdb/buffer.go index 8d2a6b6943..138962110f 100644 --- a/triedb/pathdb/buffer.go +++ b/triedb/pathdb/buffer.go @@ -39,8 +39,12 @@ type buffer struct { nodes *nodeSet // Aggregated trie node set states *stateSet // Aggregated state set - done chan struct{} // notifier whether the content in buffer has been flushed or not - flushErr error // error if any exception occurs during flushing + // done is the notifier whether the content in buffer has been flushed or not. + // This channel is nil if the buffer is not frozen. + done chan struct{} + + // flushErr memorizes the error if any exception occurs during flushing + flushErr error } // newBuffer initializes the buffer with the provided states and trie nodes. @@ -65,7 +69,7 @@ func (b *buffer) account(hash common.Hash) ([]byte, bool) { return b.states.account(hash) } -// storage retrieves the storage slot with account address hash and slot key. +// storage retrieves the storage slot with account address hash and slot key hash. func (b *buffer) storage(addrHash common.Hash, storageHash common.Hash) ([]byte, bool) { return b.states.storage(addrHash, storageHash) } @@ -134,6 +138,8 @@ func (b *buffer) flush(root common.Hash, db ethdb.KeyValueStore, freezer ethdb.A } b.done = make(chan struct{}) // allocate the channel for notification + // Schedule the background thread to construct the batch, which usually + // take a few seconds. go func() { defer func() { if postFlush != nil { @@ -185,7 +191,9 @@ func (b *buffer) flush(root common.Hash, db ethdb.KeyValueStore, freezer ethdb.A // The content in the frozen buffer is kept for consequent state access, // TODO (rjl493456442) measure the gc overhead for holding this struct. // TODO (rjl493456442) can we somehow get rid of it after flushing?? - b.reset() + // TODO (rjl493456442) buffer itself is not thread-safe, add the lock + // protection if try to reset the buffer here. + // b.reset() log.Debug("Persisted buffer content", "nodes", nodes, "accounts", accounts, "slots", slots, "bytes", common.StorageSize(size), "elapsed", common.PrettyDuration(time.Since(start))) }() } diff --git a/triedb/pathdb/database.go b/triedb/pathdb/database.go index 83fb6ac431..159c91a7c1 100644 --- a/triedb/pathdb/database.go +++ b/triedb/pathdb/database.go @@ -437,12 +437,9 @@ func (db *Database) Disable() error { // Terminate the state generator if it's active and mark the disk layer // as stale to prevent access to persistent state. disk := db.tree.bottom() - if err := disk.waitFlush(); err != nil { + if err := disk.terminate(); err != nil { return err } - if disk.generator != nil { - disk.generator.stop() - } disk.markStale() // Write the initial sync flag to persist it across restarts. @@ -602,13 +599,9 @@ func (db *Database) Close() error { // be done before terminating the potential background snapshot // generator. dl := db.tree.bottom() - if err := dl.waitFlush(); err != nil { + if err := dl.terminate(); err != nil { return err } - // Terminate the background generation if it's active - if dl.generator != nil { - dl.generator.stop() - } dl.resetCache() // release the memory held by clean cache // Close the attached state history freezer. @@ -677,6 +670,9 @@ func (db *Database) HistoryRange() (uint64, uint64, error) { // waitGeneration waits until the background generation is finished. It assumes // that the generation is permitted; otherwise, it will block indefinitely. func (db *Database) waitGeneration() { + db.lock.RLock() + defer db.lock.RUnlock() + gen := db.tree.bottom().generator if gen == nil || gen.completed() { return @@ -693,7 +689,7 @@ func (db *Database) AccountIterator(root common.Hash, seek common.Hash) (Account if wait { return nil, errDatabaseWaitSync } - if gen := db.tree.bottom().generator; gen != nil && !gen.completed() { + if !db.tree.bottom().genComplete() { return nil, errNotConstructed } return newFastAccountIterator(db, root, seek) @@ -708,7 +704,7 @@ func (db *Database) StorageIterator(root common.Hash, account common.Hash, seek if wait { return nil, errDatabaseWaitSync } - if gen := db.tree.bottom().generator; gen != nil && !gen.completed() { + if !db.tree.bottom().genComplete() { return nil, errNotConstructed } return newFastStorageIterator(db, root, account, seek) diff --git a/triedb/pathdb/disklayer.go b/triedb/pathdb/disklayer.go index 1973231771..b360a8f158 100644 --- a/triedb/pathdb/disklayer.go +++ b/triedb/pathdb/disklayer.go @@ -49,6 +49,7 @@ type diskLayer struct { // The generator is set if the state snapshot was not fully completed, // regardless of whether the background generation is running or not. + // It should only be unset if the generation completes. generator *generator } @@ -121,7 +122,7 @@ func (dl *diskLayer) node(owner common.Hash, path []byte, depth int) ([]byte, co // (both the live one and the frozen one). Note the buffer is lock free since // it's impossible to mutate the buffer before tagging the layer as stale. for _, buffer := range []*buffer{dl.buffer, dl.frozen} { - if buffer != nil && !buffer.empty() { + if buffer != nil { n, found := buffer.node(owner, path) if found { dirtyNodeHitMeter.Mark(1) @@ -177,7 +178,7 @@ func (dl *diskLayer) account(hash common.Hash, depth int) ([]byte, error) { // (both the live one and the frozen one). Note the buffer is lock free since // it's impossible to mutate the buffer before tagging the layer as stale. for _, buffer := range []*buffer{dl.buffer, dl.frozen} { - if buffer != nil && !buffer.empty() { + if buffer != nil { blob, found := buffer.account(hash) if found { dirtyStateHitMeter.Mark(1) @@ -255,7 +256,7 @@ func (dl *diskLayer) storage(accountHash, storageHash common.Hash, depth int) ([ // (both the live one and the frozen one). Note the buffer is lock free since // it's impossible to mutate the buffer before tagging the layer as stale. for _, buffer := range []*buffer{dl.buffer, dl.frozen} { - if buffer != nil && !buffer.empty() { + if buffer != nil { if blob, found := buffer.storage(accountHash, storageHash); found { dirtyStateHitMeter.Mark(1) dirtyStateReadMeter.Mark(int64(len(blob))) @@ -403,7 +404,6 @@ func (dl *diskLayer) commit(bottom *diffLayer, force bool) (*diskLayer, error) { // If the snapshot has been fully generated, unset the generator if progress == nil { - gen = nil dl.setGenerator(nil) } else { log.Info("Paused snapshot generation") @@ -425,18 +425,12 @@ func (dl *diskLayer) commit(bottom *diffLayer, force bool) (*diskLayer, error) { } }) // Block until the frozen buffer is fully flushed out if the async flushing - // is not allowed. - if dl.db.config.NoAsyncFlush { - if err := dl.frozen.waitFlush(); err != nil { - return nil, err - } - } - // Block until the frozen buffer is fully flushed out if the oldest history - // surpasses the persisted state ID. - if persistedID < oldest { + // is not allowed, or if the oldest history surpasses the persisted state ID. + if dl.db.config.NoAsyncFlush || persistedID < oldest { if err := dl.frozen.waitFlush(); err != nil { return nil, err } + dl.frozen = nil } combined = newBuffer(dl.db.config.WriteBufferSize, nil, nil, 0) } @@ -512,7 +506,9 @@ func (dl *diskLayer) revert(h *history) (*diskLayer, error) { dl.frozen = nil } - // Terminate the generation before writing any data into database + // Terminate the generator before writing any data to the database. + // This must be done after flushing the frozen buffer, as the generator + // may be restarted at the end of the flush process. var progress []byte if dl.generator != nil { dl.generator.stop() @@ -576,11 +572,29 @@ func (dl *diskLayer) genMarker() []byte { return dl.generator.progressMarker() } -// waitFlush blocks until the buffer has been fully flushed and returns any -// stored errors that occurred during the process. -func (dl *diskLayer) waitFlush() error { - if dl.frozen == nil { - return nil - } - return dl.frozen.waitFlush() +// genComplete returns a flag indicating whether the state snapshot has been +// fully generated. +func (dl *diskLayer) genComplete() bool { + dl.lock.RLock() + defer dl.lock.RUnlock() + + return dl.genMarker() == nil +} + +// terminate releases the frozen buffer if it's not nil and terminates the +// background state generator. +func (dl *diskLayer) terminate() error { + dl.lock.Lock() + defer dl.lock.Unlock() + + if dl.frozen != nil { + if err := dl.frozen.waitFlush(); err != nil { + return err + } + dl.frozen = nil + } + if dl.generator != nil { + dl.generator.stop() + } + return nil } diff --git a/triedb/pathdb/journal.go b/triedb/pathdb/journal.go index cf81e2199d..d4a6b3e262 100644 --- a/triedb/pathdb/journal.go +++ b/triedb/pathdb/journal.go @@ -304,13 +304,9 @@ func (db *Database) Journal(root common.Hash) error { // Block until the background flushing is finished. It must // be done before terminating the potential background snapshot // generator. - if err := disk.waitFlush(); err != nil { + if err := disk.terminate(); err != nil { return err } - // Terminate the background state generation if it's active - if disk.generator != nil { - disk.generator.stop() - } start := time.Now() // Run the journaling diff --git a/triedb/pathdb/layertree.go b/triedb/pathdb/layertree.go index a0615bc8d4..b2f3f7f37d 100644 --- a/triedb/pathdb/layertree.go +++ b/triedb/pathdb/layertree.go @@ -195,10 +195,6 @@ func (tree *layerTree) cap(root common.Hash, layers int) error { } tree.base = base - // Block until the frozen buffer is fully flushed - if err := base.waitFlush(); err != nil { - return err - } // Reset the layer tree with the single new disk layer tree.layers = map[common.Hash]layer{ base.rootHash(): base,