diff --git a/triedb/pathdb/buffer.go b/triedb/pathdb/buffer.go index 5d3099285f..db5fd5033f 100644 --- a/triedb/pathdb/buffer.go +++ b/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))) }() } diff --git a/triedb/pathdb/iterator_binary.go b/triedb/pathdb/iterator_binary.go index 97a2918989..cc982b93ba 100644 --- a/triedb/pathdb/iterator_binary.go +++ b/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. diff --git a/triedb/pathdb/iterator_fast.go b/triedb/pathdb/iterator_fast.go index 04bf89e874..91ef408021 100644 --- a/triedb/pathdb/iterator_fast.go +++ b/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, }) diff --git a/triedb/pathdb/journal.go b/triedb/pathdb/journal.go index ad0111e6c7..871c202bab 100644 --- a/triedb/pathdb/journal.go +++ b/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 diff --git a/triedb/pathdb/nodes.go b/triedb/pathdb/nodes.go index 62c72c1953..83a33d8c9b 100644 --- a/triedb/pathdb/nodes.go +++ b/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) diff --git a/triedb/pathdb/states.go b/triedb/pathdb/states.go index 27a6c1d422..af6e2d19d6 100644 --- a/triedb/pathdb/states.go +++ b/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) diff --git a/triedb/pathdb/states_test.go b/triedb/pathdb/states_test.go index 4d181cc914..e408c334bf 100644 --- a/triedb/pathdb/states_test.go +++ b/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 */