triedb/pathdb: polish

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
 }
 

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

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)

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 {
+// 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
 }
-	return dl.frozen.waitFlush()
-}

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

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,