diff --git a/trie/database.go b/trie/database.go index 6929b9c0da..0db779f38c 100644 --- a/trie/database.go +++ b/trie/database.go @@ -148,9 +148,11 @@ func (db *Database) Commit(root common.Hash, report bool) error { return db.backend.Commit(root, report) } -// Size returns the storage size of diff layer nodes above the persistent disk -// layer, the dirty nodes buffered within the disk layer, and the size of cached -// preimages. +// Size returns the sizes of: +// - the diff layer nodes above the persistent disk layer, +// - the mutable dirty nodes buffered within the disk layer, +// - the immutable nodes in the disk layer, +// - the cached preimages. func (db *Database) Size() (common.StorageSize, common.StorageSize, common.StorageSize, common.StorageSize) { var ( diffs, nodes, nodesImmutable common.StorageSize diff --git a/trie/triedb/pathdb/disklayer.go b/trie/triedb/pathdb/disklayer.go index c6d9008882..bfa506fd3e 100644 --- a/trie/triedb/pathdb/disklayer.go +++ b/trie/triedb/pathdb/disklayer.go @@ -64,7 +64,9 @@ type trienodebuffer interface { // empty returns an indicator if trienodebuffer contains any state transition inside. empty() bool - // getSize return the trienodebuffer used size. + // getSize return the trienodebuffer used size, includes: + // - the mutable dirty nodes buffered within the disk layer, + // - the immutable nodes in the disk layer. getSize() (uint64, uint64) // getAllNodes return all the trie nodes are cached in trienodebuffer. @@ -336,7 +338,9 @@ func (dl *diskLayer) setBufferSize(size int) error { return dl.buffer.setSize(size, dl.db.diskdb, dl.cleans, dl.id) } -// size returns the approximate size of cached nodes in the disk layer. +// size returns the approximate size of: +// - the mutable dirty nodes buffered within the disk layer, +// - the immutable nodes in the disk layer. func (dl *diskLayer) size() (common.StorageSize, common.StorageSize) { dl.lock.RLock() defer dl.lock.RUnlock() diff --git a/trie/triedb/pathdb/errors.go b/trie/triedb/pathdb/errors.go index d8a5805317..f78c82af4e 100644 --- a/trie/triedb/pathdb/errors.go +++ b/trie/triedb/pathdb/errors.go @@ -51,11 +51,9 @@ var ( errUnexpectedNode = errors.New("unexpected node") // errWriteImmutable is returned if write to background immutable nodebuffer - // under asyncnodebuffer errWriteImmutable = errors.New("write immutable node buffer") - // errFlushMutable is returned if flush the background mutable nodebuffer - // to disk, under asyncnodebuffer + // errFlushMutable is returned if flush the background mutable nodebuffer to disk errFlushMutable = errors.New("flush mutable node buffer") // errRevertImmutable is returned if revert the background immutable nodebuffer diff --git a/trie/triedb/pathdb/nodebuffer.go b/trie/triedb/pathdb/nodebuffer.go index 65e64badc6..3293fcebbd 100644 --- a/trie/triedb/pathdb/nodebuffer.go +++ b/trie/triedb/pathdb/nodebuffer.go @@ -33,12 +33,15 @@ import ( var _ trienodebuffer = &asyncnodebuffer{} -// asyncnodebuffer implement trienodebuffer interface, and aysnc the nodecache -// to disk. +// asyncnodebuffer implement trienodebuffer interface, and async flush nodebuffer +// to disk. It includes two nodebuffer, the mutable and immutable nodebuffer. The +// mutable nodebuffer that up to the size limit switches the immutable, and the new +// mutable nodebuffer can continue to be committed nodes. Retrieves node will access +// mutable nodebuffer firstly, then immutable nodebuffer. type asyncnodebuffer struct { - mux sync.RWMutex - current *nodebuffer - background *nodebuffer + mu sync.RWMutex // Lock used to protect current and background switch + current *nodebuffer // mutable nodebuffer is used to write and read nodes + background *nodebuffer // immutable nodebuffer is readonly and async flush to disk } // newAsyncNodeBuffer initializes the async node buffer with the provided nodes. @@ -49,10 +52,10 @@ func newAsyncNodeBuffer(limit int, nodes map[common.Hash]map[string]*trienode.No } } -// node retrieves the trie node with given node info. +// node retrieves the trie node with given node info, retrieves the current, then background. func (a *asyncnodebuffer) node(owner common.Hash, path []byte, hash common.Hash) (*trienode.Node, error) { - a.mux.RLock() - defer a.mux.RUnlock() + a.mu.RLock() + defer a.mu.RUnlock() node, err := a.current.node(owner, path, hash) if err != nil { @@ -64,13 +67,13 @@ func (a *asyncnodebuffer) node(owner common.Hash, path []byte, hash common.Hash) return node, nil } -// commit merges the dirty nodes into the nodebuffer. This operation won't take -// the ownership of the nodes map which belongs to the bottom-most diff layer. -// It will just hold the node references from the given map which are safe to -// copy. +// commit merges the dirty nodes into the current nodebuffer. This operation +// won't take the ownership of the nodes map which belongs to the bottom-most +// diff layer. It will just hold the node references from the given map which +// are safe to copy. func (a *asyncnodebuffer) commit(nodes map[common.Hash]map[string]*trienode.Node) trienodebuffer { - a.mux.Lock() - defer a.mux.Unlock() + a.mu.Lock() + defer a.mu.Unlock() if err := a.current.commit(nodes); err != nil { log.Warn("Failed to commit trie nodes", "error", err) @@ -82,8 +85,8 @@ func (a *asyncnodebuffer) commit(nodes map[common.Hash]map[string]*trienode.Node // into the nodebuffer, the difference is that the provided node set should // revert the changes made by the last state transition. func (a *asyncnodebuffer) revert(db ethdb.KeyValueReader, nodes map[common.Hash]map[string]*trienode.Node) error { - a.mux.Lock() - defer a.mux.Unlock() + a.mu.Lock() + defer a.mu.Unlock() newBuf, err := a.current.merge(a.background) if err != nil { @@ -95,10 +98,10 @@ func (a *asyncnodebuffer) revert(db ethdb.KeyValueReader, nodes map[common.Hash] return a.current.revert(db, nodes) } -// setSize is unsupported in asyncnodebuffer, due to the double buffer, blocking will occur. +// setSize sets the nodebuffer size limit. func (a *asyncnodebuffer) setSize(size int, db ethdb.KeyValueStore, clean *fastcache.Cache, id uint64) error { - a.mux.Lock() - defer a.mux.Unlock() + a.mu.Lock() + defer a.mu.Unlock() newBuf, err := a.current.merge(a.background) if err != nil { log.Warn("[BUG] failed to merge node cache under revert async node buffer", "error", err) @@ -113,8 +116,8 @@ func (a *asyncnodebuffer) setSize(size int, db ethdb.KeyValueStore, clean *fastc // reset cleans up the disk cache. func (a *asyncnodebuffer) reset() { - a.mux.Lock() - defer a.mux.Unlock() + a.mu.Lock() + defer a.mu.Unlock() a.current.reset() a.background.reset() @@ -122,26 +125,28 @@ func (a *asyncnodebuffer) reset() { // empty returns an indicator if nodebuffer contains any state transition inside. func (a *asyncnodebuffer) empty() bool { - a.mux.RLock() - defer a.mux.RUnlock() + a.mu.RLock() + defer a.mu.RUnlock() return a.current.empty() && a.background.empty() } -// flush persists the in-memory dirty trie node into the disk if the configured -// memory threshold is reached. Note, all data must be written atomically. +// flush persists the immutable dirty trie node into the disk. If the configured +// memory threshold is reached, switch the mutable nodebuffer to immutable, if the +// previous immutable nodebuffer flushing to disk immediately return. Note, all +// data belongs the same nodebuffer must be written atomically. func (a *asyncnodebuffer) flush(db ethdb.KeyValueStore, clean *fastcache.Cache, id uint64, force bool) error { - a.mux.Lock() - defer a.mux.Unlock() + a.mu.Lock() + defer a.mu.Unlock() if force { for { - if atomic.LoadUint64(&a.background.immutable) == 1 { + if a.background.immutable.Load() { time.Sleep(time.Duration(DefaultBackgroundFlushInterval) * time.Second) log.Info("waiting background memory table flush to disk for force flush node buffer") continue } - atomic.StoreUint64(&a.current.immutable, 1) + a.current.immutable.Store(true) return a.current.flush(db, clean, id, true) } } @@ -151,11 +156,11 @@ func (a *asyncnodebuffer) flush(db ethdb.KeyValueStore, clean *fastcache.Cache, } // background flush doing - if atomic.LoadUint64(&a.background.immutable) == 1 { + if a.background.immutable.Load() { return nil } - - atomic.StoreUint64(&a.current.immutable, 1) + // immutable the current nodebuffer, ready for switching + a.current.immutable.Store(true) a.current, a.background = a.background, a.current go func(persistId uint64) { @@ -172,8 +177,8 @@ func (a *asyncnodebuffer) flush(db ethdb.KeyValueStore, clean *fastcache.Cache, } func (a *asyncnodebuffer) getAllNodes() map[common.Hash]map[string]*trienode.Node { - a.mux.Lock() - defer a.mux.Unlock() + a.mu.Lock() + defer a.mu.Unlock() cached, err := a.current.merge(a.background) if err != nil { @@ -183,15 +188,15 @@ func (a *asyncnodebuffer) getAllNodes() map[common.Hash]map[string]*trienode.Nod } func (a *asyncnodebuffer) getLayers() uint64 { - a.mux.RLock() - defer a.mux.RUnlock() + a.mu.RLock() + defer a.mu.RUnlock() return a.current.layers + a.background.layers } func (a *asyncnodebuffer) getSize() (uint64, uint64) { - a.mux.RLock() - defer a.mux.RUnlock() + a.mu.RLock() + defer a.mu.RUnlock() return a.current.size, a.background.size } @@ -200,11 +205,13 @@ func (a *asyncnodebuffer) getSize() (uint64, uint64) { // write. The content of the nodebuffer must be checked before diving into // disk (since it basically is not-yet-written data). type nodebuffer struct { - layers uint64 // The number of diff layers aggregated inside - size uint64 // The size of aggregated writes - limit uint64 // The maximum memory allowance in bytes - nodes map[common.Hash]map[string]*trienode.Node // The dirty node set, mapped by owner and path - immutable uint64 // The flag equal 1, flush nodes to disk background + layers uint64 // The number of diff layers aggregated inside + size uint64 // The size of aggregated writes + limit uint64 // The maximum memory allowance in bytes + nodes map[common.Hash]map[string]*trienode.Node // The dirty node set, mapped by owner and path + // If this is set to true, then this nodebuffer is immutable and any write-operations to it will exit with error. + // If this is set to true, then some other thread is performing a flush in the background, and thus nonblock the write/read-operations. + immutable atomic.Bool // The flag equal true, readonly wait to flush nodes to disk background } // newNodeBuffer initializes the node buffer with the provided nodes. @@ -218,13 +225,14 @@ func newNodeBuffer(limit int, nodes map[common.Hash]map[string]*trienode.Node, l size += uint64(len(n.Blob) + len(path)) } } - return &nodebuffer{ - layers: layers, - nodes: nodes, - size: size, - limit: uint64(limit), - immutable: 0, + nb := &nodebuffer{ + layers: layers, + nodes: nodes, + size: size, + limit: uint64(limit), } + nb.immutable.Store(false) + return nb } // node retrieves the trie node with given node info. @@ -250,7 +258,7 @@ func (b *nodebuffer) node(owner common.Hash, path []byte, hash common.Hash) (*tr // It will just hold the node references from the given map which are safe to // copy. func (b *nodebuffer) commit(nodes map[common.Hash]map[string]*trienode.Node) error { - if atomic.LoadUint64(&b.immutable) == 1 { + if b.immutable.Load() { return errWriteImmutable } @@ -298,7 +306,7 @@ func (b *nodebuffer) commit(nodes map[common.Hash]map[string]*trienode.Node) err // into the nodebuffer, the difference is that the provided node set should // revert the changes made by the last state transition. func (b *nodebuffer) revert(db ethdb.KeyValueReader, nodes map[common.Hash]map[string]*trienode.Node) error { - if atomic.LoadUint64(&b.immutable) == 1 { + if b.immutable.Load() { return errRevertImmutable } @@ -364,7 +372,7 @@ func (b *nodebuffer) updateSize(delta int64) { // reset cleans up the disk cache. func (b *nodebuffer) reset() { - atomic.StoreUint64(&b.immutable, 0) + b.immutable.Store(false) b.layers = 0 b.size = 0 b.nodes = make(map[common.Hash]map[string]*trienode.Node) @@ -378,8 +386,8 @@ func (b *nodebuffer) empty() bool { // setSize sets the buffer size to the provided number, and invokes a flush // operation if the current memory usage exceeds the new limit. func (b *nodebuffer) setSize(size int, db ethdb.KeyValueStore, clean *fastcache.Cache, id uint64) error { - if atomic.LoadUint64(&b.immutable) == 1 { - return errRevertImmutable + if b.immutable.Load() { + return errWriteImmutable } b.limit = uint64(size) @@ -393,15 +401,15 @@ func (b *nodebuffer) merge(nb *nodebuffer) (*nodebuffer, error) { } if b == nil || b.empty() { res := copyNodeBuffer(nb) - atomic.StoreUint64(&res.immutable, 0) + res.immutable.Store(false) return nb, nil } if nb == nil || nb.empty() { res := copyNodeBuffer(b) - atomic.StoreUint64(&res.immutable, 0) + res.immutable.Store(false) return b, nil } - if atomic.LoadUint64(&b.immutable) == atomic.LoadUint64(&nb.immutable) { + if b.immutable.Load() == nb.immutable.Load() { return nil, errIncompatibleMerge } @@ -409,7 +417,7 @@ func (b *nodebuffer) merge(nb *nodebuffer) (*nodebuffer, error) { immutable *nodebuffer mutable *nodebuffer ) - if atomic.LoadUint64(&b.immutable) == 1 { + if b.immutable.Load() { immutable = b mutable = nb } else { @@ -441,7 +449,7 @@ func (b *nodebuffer) merge(nb *nodebuffer) (*nodebuffer, error) { // flush persists the in-memory dirty trie node into the disk if the configured // memory threshold is reached. Note, all data must be written atomically. func (b *nodebuffer) flush(db ethdb.KeyValueStore, clean *fastcache.Cache, id uint64, force bool) error { - if atomic.LoadUint64(&b.immutable) == 0 { + if !b.immutable.Load() { return errFlushMutable } @@ -527,5 +535,6 @@ func copyNodeBuffer(n *nodebuffer) *nodebuffer { } } nb := newNodeBuffer(int(n.limit), nodes, n.layers) + nb.immutable.Store(n.immutable.Load()) return nb }