mirror of
https://github.com/ethereum/go-ethereum.git
synced 2026-08-20 10:52:25 +00:00
pathdb: add more docs for asyncnodebuffer
This commit is contained in:
joeylichang
parent
37ec06a5ce
commit
8677240ff8
4 changed files with 80 additions and 67 deletions
|
|
@ -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
|
||||
|
|
|
|||
|
|
@ -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()
|
||||
|
|
|
|||
|
|
@ -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
|
||||
|
|
|
|||
|
|
@ -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
|
||||
}
|
||||
|
|
@ -204,7 +209,9 @@ type nodebuffer struct {
|
|||
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
|
||||
// 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{
|
||||
nb := &nodebuffer{
|
||||
layers: layers,
|
||||
nodes: nodes,
|
||||
size: size,
|
||||
limit: uint64(limit),
|
||||
immutable: 0,
|
||||
}
|
||||
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
|
||||
}
|
||||
|
|
|
|||
Loading…
Reference in a new issue