@ -36,12 +36,13 @@ type diskLayer struct {
db * Database // Path-based trie database
cleans * fastcache . Cache // GC friendly memory cache of clean node RLPs
buffer * nodebuffer // Node buffer to aggregate writes
frozen * nodebuffer // Frozen node buffer waiting for flushing
stale bool // Signals that the layer became stale (state progressed)
lock sync . RWMutex // Lock used to protect stale flag
}
// newDiskLayer creates a new disk layer based on the passing arguments.
func newDiskLayer ( root common . Hash , id uint64 , db * Database , cleans * fastcache . Cache , buffer * nodebuffer ) * diskLayer {
func newDiskLayer ( root common . Hash , id uint64 , db * Database , cleans * fastcache . Cache , buffer * nodebuffer , frozen * nodebuffer ) * diskLayer {
// Initialize a clean cache if the memory allowance is not zero
// or reuse the provided cache if it is not nil (inherited from
// the original disk layer).
@ -54,6 +55,7 @@ func newDiskLayer(root common.Hash, id uint64, db *Database, cleans *fastcache.C
db : db ,
cleans : cleans ,
buffer : buffer ,
frozen : frozen ,
}
}
@ -102,16 +104,19 @@ func (dl *diskLayer) node(owner common.Hash, path []byte, depth int) ([]byte, co
if dl . stale {
return nil , common . Hash { } , nil , errSnapshotStale
}
// Try to retrieve the trie node from the not-yet-written
// node buffer first. Note the buffer is lock free since
// it's impossible to mutate the buffer before tagging the
// layer as stale.
n , found := dl . buffer . node ( owner , path )
if found {
dirtyHitMeter . Mark ( 1 )
dirtyReadMeter . Mark ( int64 ( len ( n . Blob ) ) )
dirtyNodeHitDepthHist . Update ( int64 ( depth ) )
return n . Blob , n . Hash , & nodeLoc { loc : locDirtyCache , depth : depth } , nil
// Try to retrieve the trie node from the not-yet-written node buffer first
// (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 [ ] * nodebuffer { dl . buffer , dl . frozen } {
if buffer != nil {
n , found := buffer . node ( owner , path )
if found {
dirtyHitMeter . Mark ( 1 )
dirtyReadMeter . Mark ( int64 ( len ( n . Blob ) ) )
dirtyNodeHitDepthHist . Update ( int64 ( depth ) )
return n . Blob , n . Hash , & nodeLoc { loc : locDirtyCache , depth : depth } , nil
}
}
}
dirtyMissMeter . Mark ( 1 )
@ -182,7 +187,7 @@ func (dl *diskLayer) commit(bottom *diffLayer, force bool) (*diskLayer, error) {
// Mark the diskLayer as stale before applying any mutations on top.
dl . stale = true
// Store the root->id lookup afterwards . All stored lookups are identified
// Store the root->id lookup afterward. All stored lookups are identified
// by the **unique** state root. It's impossible that in the same chain
// blocks are not adjacent but have the same root.
if dl . id == 0 {
@ -190,21 +195,43 @@ func (dl *diskLayer) commit(bottom *diffLayer, force bool) (*diskLayer, error) {
}
rawdb . WriteStateID ( dl . db . diskdb , bottom . rootHash ( ) , bottom . stateID ( ) )
// Construct a new disk layer by merging the nodes from the provided diff
// layer, and flush the content in disk layer if there are too many nodes
// cached. The clean cache is inherited from the original disk layer.
ndl := newDiskLayer ( bottom . root , bottom . stateID ( ) , dl . db , dl . cleans , dl . buffer . commit ( bottom . nodes ) )
// In a unique scenario where the ID of the oldest history object (after tail
// truncation) surpasses the persisted state ID, we take the necessary action
// of forcibly committing the cached dirty nod es to ensure that the persisted
// of forcibly committing the cached dirty states to ensure that the persisted
// state ID remains higher.
if ! force && rawdb . ReadPersistentStateID ( dl . db . diskdb ) < oldest {
persistedID := rawdb . ReadPersistentStateID ( dl . db . diskdb )
if ! force && persistedID < oldest {
force = true
}
if err := ndl . buffer . flush ( ndl . db . diskdb , ndl . cleans , ndl . id , force ) ; err != nil {
return nil , err
// Merge the nodes of the bottom-most diff layer into the buffer as the combined one
combined := dl . buffer . commit ( bottom . nodes )
if combined . full ( ) || force {
// Wait until the previous frozen buffer is fully flushed
if dl . frozen != nil {
if err := dl . frozen . flushed ( ) ; err != nil {
return nil , err
}
}
dl . frozen = nil
// Freeze the live buffer and schedule background flushing
dl . frozen = combined
dl . frozen . flush ( dl . db . diskdb , dl . cleans , bottom . stateID ( ) )
// Block until the frozen buffer is fully flushed out if the oldest history
// surpasses the persisted state ID.
if persistedID < oldest {
if err := dl . frozen . flushed ( ) ; err != nil {
return nil , err
}
}
combined = newNodeBuffer ( dl . db . bufferSize , nil , 0 )
}
// Construct a new disk layer by merging the nodes from the provided diff
// layer, and flush the content in disk layer if there are too many nodes
// cached. The clean cache is inherited from the original disk layer.
ndl := newDiskLayer ( bottom . root , bottom . stateID ( ) , dl . db , dl . cleans , combined , dl . frozen )
// To remove outdated history objects from the end, we set the 'tail' parameter
// to 'oldest-1' due to the offset between the freezer index and the history ID.
if overflow {
@ -249,6 +276,13 @@ func (dl *diskLayer) revert(h *history) (*diskLayer, error) {
return nil , err
}
} else {
// Block until the frozen buffer is fully flushed
if dl . frozen != nil {
if err := dl . frozen . flushed ( ) ; err != nil {
return nil , err
}
dl . frozen = nil // unset the frozen buffer
}
batch := dl . db . diskdb . NewBatch ( )
writeNodes ( batch , nodes , dl . cleans )
rawdb . WritePersistentStateID ( batch , dl . id - 1 )
@ -256,18 +290,7 @@ func (dl *diskLayer) revert(h *history) (*diskLayer, error) {
log . Crit ( "Failed to write states" , "err" , err )
}
}
return newDiskLayer ( h . meta . parent , dl . id - 1 , dl . db , dl . cleans , dl . buffer ) , nil
}
// setBufferSize sets the node buffer size to the provided value.
func ( dl * diskLayer ) setBufferSize ( size int ) error {
dl . lock . RLock ( )
defer dl . lock . RUnlock ( )
if dl . stale {
return errSnapshotStale
}
return dl . buffer . setSize ( size , dl . db . diskdb , dl . cleans , dl . id )
return newDiskLayer ( h . meta . parent , dl . id - 1 , dl . db , dl . cleans , dl . buffer , dl . frozen ) , nil
}
// size returns the approximate size of cached nodes in the disk layer.
Gary Rong
2 days ago