triedb/pathdb: optimize db commit

triedb/pathdb/buffer.go

@@ -19,6 +19,9 @@ package pathdb
 import (
 	"errors"
 	"fmt"
+	"slices"
+	"sync"
+	"sync/atomic"
 	"time"
 
 	"github.com/VictoriaMetrics/fastcache"
@@ -29,15 +32,38 @@ import (
 	"github.com/ethereum/go-ethereum/trie/trienode"
 )
 
+// mergeYieldThreshold is the number of entries the background merger integrates
+// while holding the buffer lock before it releases and re-acquires the lock,
+// giving any waiting frontend reads a chance to run. It is deliberately small:
+// the merge is allowed to be slow, but it must not stall reads. With a
+// writer-priority RWMutex the merger effectively owns the lock for the whole
+// merge, so this threshold bounds the worst-case read stall (roughly
+// mergeYieldThreshold map operations) rather than the total merge duration.
+const mergeYieldThreshold = 32
+
+// pendingMerge represents a merge operation which is handed to the buffer but
+// not yet completed.
+type pendingMerge struct {
+	nodes  *nodeSet
+	states *stateSet
+	size   uint64
+}
+
 // buffer is a collection of modified states along with the modified trie nodes.
 // They are cached here to aggregate the disk write. The content of the buffer
 // must be checked before diving into disk (since it basically is not yet written
 // data).
 type buffer struct {
-	layers uint64    // The number of diff layers aggregated inside
-	limit  uint64    // The maximum memory allowance in bytes
-	nodes  *nodeSet  // Aggregated trie node set
-	states *stateSet // Aggregated state set
+	layers uint64 // The number of diff layers aggregated inside
+	limit  uint64 // The maximum memory allowance in bytes
+
+	nodes       *nodeSet                        // Aggregated trie node set
+	states      *stateSet                       // Aggregated state set
+	pending     atomic.Pointer[[]*pendingMerge] // The list of pending merges
+	pendingSize uint64                          // The size of pending merges
+	merging     bool                            // Flag whether the background merger is running
+	lock        sync.RWMutex                    // Guard the five fields above
+	wg          sync.WaitGroup
 
 	// done is the notifier whether the content in buffer has been flushed or not.
 	// This channel is nil if the buffer is not frozen.
@@ -64,33 +90,195 @@ func newBuffer(limit int, nodes *nodeSet, states *stateSet, layers uint64) *buff
 	}
 }
 
+// loadPending returns the current immutable pending-overlay snapshot.
+func (b *buffer) loadPending() []*pendingMerge {
+	if p := b.pending.Load(); p != nil {
+		return *p
+	}
+	return nil
+}
+
+// storePending sets the pending snapshot to the given list.
+func (b *buffer) storePending(list []*pendingMerge) {
+	b.pending.Store(&list)
+}
+
 // account retrieves the account blob with account address hash.
 func (b *buffer) account(hash common.Hash) ([]byte, bool) {
+	pending := b.loadPending()
+	for i := len(pending) - 1; i >= 0; i-- {
+		if blob, found := pending[i].states.account(hash); found {
+			return blob, true
+		}
+	}
+	b.lock.RLock()
+	defer b.lock.RUnlock()
 	return b.states.account(hash)
 }
 
 // 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) {
+	pending := b.loadPending()
+	for i := len(pending) - 1; i >= 0; i-- {
+		if blob, found := pending[i].states.storage(addrHash, storageHash); found {
+			return blob, true
+		}
+	}
+	b.lock.RLock()
+	defer b.lock.RUnlock()
 	return b.states.storage(addrHash, storageHash)
 }
 
 // node retrieves the trie node with node path and its trie identifier.
 func (b *buffer) node(owner common.Hash, path []byte) (*trienode.Node, bool) {
+	pending := b.loadPending()
+	for i := len(pending) - 1; i >= 0; i-- {
+		if n, found := pending[i].nodes.node(owner, path); found {
+			return n, true
+		}
+	}
+	b.lock.RLock()
+	defer b.lock.RUnlock()
 	return b.nodes.node(owner, path)
 }
 
-// commit merges the provided states and trie nodes into the buffer.
+// accountList returns the sorted list of all accounts held by the buffer.
+func (b *buffer) accountList() []common.Hash {
+	b.lock.RLock()
+	defer b.lock.RUnlock()
+
+	pending := b.loadPending()
+	if len(pending) == 0 {
+		return b.states.accountList()
+	}
+	var (
+		list []common.Hash
+		seen = make(map[common.Hash]struct{})
+	)
+	add := func(hashes []common.Hash) {
+		for _, h := range hashes {
+			if _, ok := seen[h]; !ok {
+				seen[h] = struct{}{}
+				list = append(list, h)
+			}
+		}
+	}
+	add(b.states.accountList())
+	for _, pf := range pending {
+		add(pf.states.accountList())
+	}
+	slices.SortFunc(list, common.Hash.Cmp)
+	return list
+}
+
+// storageList returns the sorted list of all storage slot hashes held by the
+// buffer for the given account.
+func (b *buffer) storageList(account common.Hash) []common.Hash {
+	b.lock.RLock()
+	defer b.lock.RUnlock()
+
+	pending := b.loadPending()
+	if len(pending) == 0 {
+		return b.states.storageList(account)
+	}
+	var (
+		list []common.Hash
+		seen = make(map[common.Hash]struct{})
+	)
+	add := func(hashes []common.Hash) {
+		for _, h := range hashes {
+			if _, ok := seen[h]; !ok {
+				seen[h] = struct{}{}
+				list = append(list, h)
+			}
+		}
+	}
+	add(b.states.storageList(account))
+	for _, m := range pending {
+		add(m.states.storageList(account))
+	}
+	slices.SortFunc(list, common.Hash.Cmp)
+	return list
+}
+
+// commit hands the provided states and trie nodes to the buffer as an immutable
+// pending overlay and wakes the background folder. It returns immediately,
+// keeping the (expensive) merge off the caller's critical path.
 func (b *buffer) commit(nodes *nodeSet, states *stateSet) *buffer {
+	m := &pendingMerge{
+		nodes:  nodes,
+		states: states,
+		size:   nodes.size + states.size,
+	}
 	b.layers++
-	b.nodes.merge(nodes)
-	b.states.merge(states)
+
+	b.lock.Lock()
+	b.pendingSize += m.size
+	b.storePending(append(slices.Clone(b.loadPending()), m))
+
+	// Spawn an ephemeral merger if one isn't already running. It exits on its
+	// own once it has drained the pending queue, so there is no goroutine to
+	// stop/clean up later.
+	if !b.merging {
+		b.merging = true
+		b.wg.Add(1)
+		go b.merge()
+	}
+	b.lock.Unlock()
+
 	return b
 }
 
+// merge integrates pending overlays into the aggregated maps until the queue is
+// empty, then exits.
+func (b *buffer) merge() {
+	defer b.wg.Done()
+
+	for {
+		b.lock.Lock()
+		cur := b.loadPending()
+		if len(cur) == 0 {
+			b.merging = false
+			b.lock.Unlock()
+			return
+		}
+		m := cur[0]
+
+		// Periodically release the lock to unblock the frontend reads
+		n := 0
+		yield := func() {
+			if n++; n >= mergeYieldThreshold {
+				n = 0
+				b.lock.Unlock()
+				b.lock.Lock()
+			}
+		}
+		b.nodes.merge(m.nodes, yield)
+		b.states.merge(m.states, yield)
+
+		// Drop the just-merged overlay from the front of the (possibly grown)
+		// snapshot. Re-load rather than reuse `cur`: commit may have appended new
+		// overlays during a yield window, but the merged one is always the front.
+		b.storePending(slices.Clone(b.loadPending()[1:]))
+		b.pendingSize -= m.size
+		b.lock.Unlock()
+	}
+}
+
+// waitMerge blocks until the background merger (if any) has drained all pending
+// overlays and exited.
+func (b *buffer) waitMerge() {
+	b.wg.Wait()
+}
+
 // revertTo is the reverse operation of commit. It also merges the provided states
 // and trie nodes into the buffer. The key difference is that the provided state
 // set should reverse the changes made by the most recent state transition.
 func (b *buffer) revertTo(db ethdb.KeyValueReader, nodes map[common.Hash]map[string]*trienode.Node, accounts map[common.Hash][]byte, storages map[common.Hash]map[common.Hash][]byte) error {
+	// Reverting operates on the fully merged maps; wait for the merger to drain
+	// all outstanding overlays first (no commit can be in flight here).
+	b.waitMerge()
+
 	// Short circuit if no embedded state transition to revert
 	if b.layers == 0 {
 		return errStateUnrecoverable
@@ -99,7 +287,8 @@ func (b *buffer) revertTo(db ethdb.KeyValueReader, nodes map[common.Hash]map[str
 
 	// Reset the entire buffer if only a single transition left
 	if b.layers == 0 {
-		b.reset()
+		b.nodes.reset()
+		b.states.reset()
 		return nil
 	}
 	b.nodes.revertTo(db, nodes)
@@ -107,13 +296,6 @@ func (b *buffer) revertTo(db ethdb.KeyValueReader, nodes map[common.Hash]map[str
 	return nil
 }
 
-// reset cleans up the disk cache.
-func (b *buffer) reset() {
-	b.layers = 0
-	b.nodes.reset()
-	b.states.reset()
-}
-
 // empty returns an indicator if buffer is empty.
 func (b *buffer) empty() bool {
 	return b.layers == 0
@@ -125,9 +307,13 @@ func (b *buffer) full() bool {
 	return b.size() > b.limit
 }
 
-// size returns the approximate memory size of the held content.
+// size returns the approximate memory size of the held content, including the
+// not-yet-merged overlays.
 func (b *buffer) size() uint64 {
-	return b.states.size + b.nodes.size
+	b.lock.RLock()
+	defer b.lock.RUnlock()
+
+	return b.states.size + b.nodes.size + b.pendingSize // pendingSize is mutable
 }
 
 // flush persists the in-memory dirty trie node into the disk if the configured
@@ -136,6 +322,11 @@ func (b *buffer) flush(root common.Hash, db ethdb.KeyValueStore, freezers []ethd
 	if b.done != nil {
 		panic("duplicated flush operation")
 	}
+	// Wait for the merger to integrate any remaining overlays before reading the
+	// maps. The buffer is frozen at this point, so no new overlays can arrive and
+	// the maps are stable once waitMerge returns.
+	b.waitMerge()
+
 	b.done = make(chan struct{}) // allocate the channel for notification
 
 	// Schedule the background thread to construct the batch, which usually
@@ -188,12 +379,6 @@ func (b *buffer) flush(root common.Hash, db ethdb.KeyValueStore, freezers []ethd
 		commitStoragesMeter.Mark(int64(slots))
 		commitTimeTimer.UpdateSince(start)
 
-		// 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??
-		// 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/iterator_binary.go

@@ -49,11 +49,7 @@ func (dl *diskLayer) initBinaryAccountIterator(seek common.Hash) *binaryIterator
 	if err := dl.waitFlush(); err != nil {
 		panic(err)
 	}
-	// The state set in the disk layer is mutable, hold the lock before obtaining
-	// the account list to prevent concurrent map iteration and write.
-	dl.lock.RLock()
-	accountList := dl.buffer.states.accountList()
-	dl.lock.RUnlock()
+	accountList := dl.buffer.accountList()
 
 	// Create two iterators for state buffer and the persistent state in disk
 	// respectively and combine them as a binary iterator.
@@ -121,11 +117,7 @@ func (dl *diskLayer) initBinaryStorageIterator(account common.Hash, seek common.
 	if err := dl.waitFlush(); err != nil {
 		panic(err)
 	}
-	// The state set in the disk layer is mutable, hold the lock before obtaining
-	// the storage list to prevent concurrent map iteration and write.
-	dl.lock.RLock()
-	storageList := dl.buffer.states.storageList(account)
-	dl.lock.RUnlock()
+	storageList := dl.buffer.storageList(account)
 
 	// Create two iterators for state buffer and the persistent state in disk
 	// respectively and combine them as a binary iterator.

triedb/pathdb/iterator_fast.go

@@ -81,11 +81,7 @@ func newFastIterator(db *Database, root common.Hash, account common.Hash, seek c
 				if err := dl.waitFlush(); err != nil {
 					return nil, err
 				}
-				// The state set in the disk layer is mutable, hold the lock before obtaining
-				// the account list to prevent concurrent map iteration and write.
-				dl.lock.RLock()
-				accountList := dl.buffer.states.accountList()
-				dl.lock.RUnlock()
+				accountList := dl.buffer.accountList()
 
 				fi.iterators = append(fi.iterators, &weightedIterator{
 					// The state set in the disk layer is mutable, and the entire state becomes stale
@@ -98,7 +94,11 @@ func newFastIterator(db *Database, root common.Hash, account common.Hash, seek c
 						if dl.stale {
 							return nil, errSnapshotStale
 						}
-						return dl.buffer.states.mustAccount(hash)
+						blob, found := dl.buffer.account(hash)
+						if !found {
+							return nil, fmt.Errorf("account is not found, %x", hash)
+						}
+						return blob, nil
 					}),
 					priority: depth,
 				})
@@ -123,11 +123,7 @@ func newFastIterator(db *Database, root common.Hash, account common.Hash, seek c
 				if err := dl.waitFlush(); err != nil {
 					return nil, err
 				}
-				// The state set in the disk layer is mutable, hold the lock before obtaining
-				// the storage list to prevent concurrent map iteration and write.
-				dl.lock.RLock()
-				storageList := dl.buffer.states.storageList(account)
-				dl.lock.RUnlock()
+				storageList := dl.buffer.storageList(account)
 
 				fi.iterators = append(fi.iterators, &weightedIterator{
 					// The state set in the disk layer is mutable, and the entire state becomes stale
@@ -140,7 +136,11 @@ func newFastIterator(db *Database, root common.Hash, account common.Hash, seek c
 						if dl.stale {
 							return nil, errSnapshotStale
 						}
-						return dl.buffer.states.mustStorage(addrHash, slotHash)
+						blob, found := dl.buffer.storage(addrHash, slotHash)
+						if !found {
+							return nil, fmt.Errorf("storage slot is not found, %x %x", addrHash, slotHash)
+						}
+						return blob, nil
 					}),
 					priority: depth,
 				})

triedb/pathdb/journal.go

@@ -263,6 +263,12 @@ func (dl *diskLayer) journal(w io.Writer) error {
 	if dl.stale {
 		return errSnapshotStale
 	}
+	// Merge any outstanding overlays into the buffer maps before serializing them,
+	// so the journal captures the complete content and the background merger is
+	// no longer mutating the maps concurrently with the encode below. Journaling
+	// runs under the database write lock, so no commit can be in flight.
+	dl.buffer.waitMerge()
+
 	// Step one, write the disk root into the journal.
 	if err := rlp.Encode(w, dl.root); err != nil {
 		return err

triedb/pathdb/nodes.go

@@ -23,7 +23,6 @@ import (
 	"fmt"
 	"hash/fnv"
 	"io"
-	"maps"
 
 	"github.com/VictoriaMetrics/fastcache"
 	"github.com/ethereum/go-ethereum/common"
@@ -109,7 +108,7 @@ func (s *nodeSet) node(owner common.Hash, path []byte) (*trienode.Node, bool) {
 
 // merge integrates the provided dirty nodes into the set. The provided nodeset
 // will remain unchanged, as it may still be referenced by other layers.
-func (s *nodeSet) merge(set *nodeSet) {
+func (s *nodeSet) merge(set *nodeSet, yield func()) {
 	var (
 		delta     int64   // size difference resulting from node merging
 		overwrite counter // counter of nodes being overwritten
@@ -124,22 +123,20 @@ func (s *nodeSet) merge(set *nodeSet) {
 			overwrite.add(len(orig.Blob) + len(path))
 		}
 		s.accountNodes[path] = n
+		yield()
 	}
 
 	// Merge storage nodes
 	for owner, subset := range set.storageNodes {
 		current, exist := s.storageNodes[owner]
 		if !exist {
+			current = make(map[string]*trienode.Node, len(subset))
 			for path, n := range subset {
+				current[path] = n
 				delta += int64(common.HashLength + len(n.Blob) + len(path))
+				yield()
 			}
-			// Perform a shallow copy of the map for the subset instead of claiming it
-			// directly from the provided nodeset to avoid potential concurrent map
-			// read/write issues. The nodes belonging to the original diff layer remain
-			// accessible even after merging. Therefore, ownership of the nodes map
-			// should still belong to the original layer, and any modifications to it
-			// should be prevented.
-			s.storageNodes[owner] = maps.Clone(subset)
+			s.storageNodes[owner] = current
 			continue
 		}
 		for path, n := range subset {
@@ -150,8 +147,8 @@ func (s *nodeSet) merge(set *nodeSet) {
 				overwrite.add(common.HashLength + len(orig.Blob) + len(path))
 			}
 			current[path] = n
+			yield()
 		}
-		s.storageNodes[owner] = current
 	}
 	overwrite.report(gcTrieNodeMeter, gcTrieNodeBytesMeter)
 	s.updateSize(delta)

triedb/pathdb/states.go

@@ -227,7 +227,7 @@ func (s *stateSet) clearLists() {
 //
 // The stateSet supplied as parameter set will not be mutated by this operation,
 // as it may still be referenced by other layers.
-func (s *stateSet) merge(other *stateSet) {
+func (s *stateSet) merge(other *stateSet, yield func()) {
 	var (
 		delta             int
 		accountOverwrites counter
@@ -242,6 +242,7 @@ func (s *stateSet) merge(other *stateSet) {
 			delta += common.HashLength + len(data)
 		}
 		s.accountData[accountHash] = data
+		yield()
 	}
 	// Apply all the updated storage slots (individually)
 	for accountHash, storage := range other.storageData {
@@ -256,6 +257,7 @@ func (s *stateSet) merge(other *stateSet) {
 			for storageHash, data := range storage {
 				slots[storageHash] = data
 				delta += 2*common.HashLength + len(data)
+				yield()
 			}
 			s.storageData[accountHash] = slots
 			continue
@@ -270,6 +272,7 @@ func (s *stateSet) merge(other *stateSet) {
 				delta += 2*common.HashLength + len(data)
 			}
 			slots[storageHash] = data
+			yield()
 		}
 	}
 	accountOverwrites.report(gcAccountMeter, gcAccountBytesMeter)

triedb/pathdb/states_test.go

@@ -67,7 +67,7 @@ func TestStatesMerge(t *testing.T) {
 		},
 		false,
 	)
-	a.merge(b)
+	a.merge(b, func() {})
 
 	blob, exist := a.account(common.Hash{0xa})
 	if !exist || !bytes.Equal(blob, []byte{0xa1}) {
@@ -157,7 +157,7 @@ func TestStatesRevert(t *testing.T) {
 		},
 		false,
 	)
-	a.merge(b)
+	a.merge(b, func() {})
 	a.revertTo(
 		map[common.Hash][]byte{
 			{0xa}: {0xa0},
@@ -236,7 +236,7 @@ func TestStateRevertAccountNullMarker(t *testing.T) {
 		nil,
 		false,
 	)
-	a.merge(b) // create account 0xa
+	a.merge(b, func() {}) // create account 0xa
 	a.revertTo(
 		map[common.Hash][]byte{
 			{0xa}: nil,
@@ -270,7 +270,7 @@ func TestStateRevertStorageNullMarker(t *testing.T) {
 		},
 		false,
 	)
-	a.merge(b) // create slot 0x1
+	a.merge(b, func() {}) // create slot 0x1
 	a.revertTo(
 		nil,
 		map[common.Hash]map[common.Hash][]byte{
@@ -437,7 +437,7 @@ func TestStateSizeTracking(t *testing.T) {
 		t.Fatalf("Unexpected size, want: %d, got: %d", expSizeB, b.size)
 	}
 
-	a.merge(b)
+	a.merge(b, func() {})
 	mergeSize := expSizeA + 1 /* account a data change */ + 2 /* account b data change */ - 1 /* account c data change */
 	mergeSize += 2*common.HashLength + 2 + 2                                                  /* storage a change */
 	mergeSize += 2*common.HashLength + 2 - 1                                                  /* storage b change */