forks/go-ethereum
1
0
Fork 1
mirror of https://github.com/ethereum/go-ethereum.git synced 2026-08-20 10:52:25 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 1

core, trie: move state pruner onto its own background thread

Browse source
This commit is contained in:
Péter Szilágyi 2019-01-14 12:53:36 +02:00
parent a32e4c7f69
commit 8a301cc929
No known key found for this signature in database
GPG key ID: E9AE538CEDF8293D
8 changed files with 302 additions and 106 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

21
core/blockchain.go
View file

@ -208,6 +208,8 @@ func NewBlockChain(db ethdb.Database, cacheConfig *CacheConfig, chainConfig *par
bc.gcsave = head.Root()
bc.stateCache.TrieDB().ForbidPrune(bc.gcsave)
}
bc.stateCache.TrieDB().ResumePruning()
// Take ownership of this particular state
go bc.update()
return bc, nil
@ -708,6 +710,9 @@ func (bc *BlockChain) Stop() {
bc.wg.Wait()
// Terminate the pruner, we don't want it to remove recent stuff while quitting
bc.stateCache.TrieDB().TerminatePruning()
// Ensure the state of a recent block is also stored to disk before exiting.
// We're writing three different states to catch different restart scenarios:
// - HEAD: So we don't need to reprocess any blocks in the general case
@ -727,7 +732,7 @@ func (bc *BlockChain) Stop() {
}
}
for !bc.gcqueue.Empty() {
triedb.Dereference(bc.gcqueue.PopItem().(common.Hash), false)
triedb.Dereference(bc.gcqueue.PopItem().(common.Hash))
}
if size, _ := triedb.Size(); size != 0 {
log.Error("Dangling trie nodes after full cleanup", "size", size)
@ -961,13 +966,17 @@ func (bc *BlockChain) writeBlockWithState(block *types.Block, receipts []*types.
}
rawdb.WriteBlock(bc.db, block)
// Pause pruning while the tries are being mutated
triedb := bc.stateCache.TrieDB()
triedb.PausePruning()
defer triedb.ResumePruning()
// Commit the state into the dirty memory-cache and either flush, or garbage collect
root, err := state.Commit(bc.chainConfig.IsEIP158(block.Number()))
if err != nil {
return NonStatTy, err
}
triedb := bc.stateCache.TrieDB()
// If we're running an archive node, always flush
if bc.cacheConfig.Disabled {
if err := triedb.Commit(root, false); err != nil {
return NonStatTy, err
@ -1011,7 +1020,7 @@ func (bc *BlockChain) writeBlockWithState(block *types.Block, receipts []*types.
triedb.ForbidPrune(header.Root)
if bc.gcsave != (common.Hash{}) {
triedb.PermitPrune(bc.gcsave)
triedb.Dereference(bc.gcsave, true)
triedb.Dereference(bc.gcsave)
}
bc.gcsave = header.Root
}
@ -1023,7 +1032,7 @@ func (bc *BlockChain) writeBlockWithState(block *types.Block, receipts []*types.
bc.gcqueue.Push(root, number)
break
}
triedb.Dereference(root.(common.Hash), true)
triedb.Dereference(root.(common.Hash))
}
}
}

2
core/state/database.go
View file

@ -84,7 +84,7 @@ func NewDatabase(db ethdb.Database) Database {
func NewDatabaseWithCache(db ethdb.Database, cache int) Database {
csc, _ := lru.New(codeSizeCacheSize)
return &cachingDB{
db: trie.NewDatabaseWithCache(db, cache),
db: trie.NewDatabaseWithCache(db, cache, true),
codeSizeCache: csc,
}
}

6
eth/api_tracer.go
View file

@ -311,7 +311,7 @@ func (api *PrivateDebugAPI) traceChain(ctx context.Context, start, end *types.Bl
}
// Dereference all past tries we ourselves are done working with
if proot != (common.Hash{}) {
database.TrieDB().Dereference(proot, false)
database.TrieDB().Dereference(proot)
}
proot = root
@ -335,7 +335,7 @@ func (api *PrivateDebugAPI) traceChain(ctx context.Context, start, end *types.Bl
done[uint64(result.Block)] = result
// Dereference any paret tries held in memory by this task
database.TrieDB().Dereference(res.rootref, false)
database.TrieDB().Dereference(res.rootref)
// Stream completed traces to the user, aborting on the first error
for result, ok := done[next]; ok; result, ok = done[next] {
@ -690,7 +690,7 @@ func (api *PrivateDebugAPI) computeStateDB(block *types.Block, reexec uint64) (*
}
database.TrieDB().Reference(common.Hash{}, root, common.Hash{})
if proot != (common.Hash{}) {
database.TrieDB().Dereference(proot, false)
database.TrieDB().Dereference(proot)
}
proot = root
}

4
les/handler.go
View file

@ -907,7 +907,7 @@ func (pm *ProtocolManager) handleMsg(p *peer) error {
if reject(uint64(reqCnt), MaxHelperTrieProofsFetch) {
return errResp(ErrRequestRejected, "")
}
trieDb := trie.NewDatabase(ethdb.NewTable(pm.chainDb, light.ChtTablePrefix))
trieDb := trie.NewDatabase(ethdb.NewTable(pm.chainDb, light.ChtTablePrefix), false)
for _, req := range req.Reqs {
if header := pm.blockchain.GetHeaderByNumber(req.BlockNum); header != nil {
sectionHead := rawdb.ReadCanonicalHash(pm.chainDb, req.ChtNum*pm.iConfig.ChtSize-1)
@ -966,7 +966,7 @@ func (pm *ProtocolManager) handleMsg(p *peer) error {
var prefix string
if root, prefix = pm.getHelperTrie(req.Type, req.TrieIdx); root != (common.Hash{}) {
auxTrie, _ = trie.New(root, trie.NewDatabase(ethdb.NewTable(pm.chainDb, prefix)))
auxTrie, _ = trie.New(root, trie.NewDatabase(ethdb.NewTable(pm.chainDb, prefix), false))
}
}
if req.AuxReq == auxRoot {

4
light/postprocess.go
View file

@ -159,7 +159,7 @@ func NewChtIndexer(db ethdb.Database, odr OdrBackend, size, confirms uint64) *co
diskdb: db,
odr: odr,
trieTable: trieTable,
triedb: trie.NewDatabaseWithCache(trieTable, 1), // Use a tiny cache only to keep memory down
triedb: trie.NewDatabaseWithCache(trieTable, 1, false), // Use a tiny cache only to keep memory down
sectionSize: size,
}
return core.NewChainIndexer(db, ethdb.NewTable(db, "chtIndex-"), backend, size, confirms, time.Millisecond*100, "cht")
@ -281,7 +281,7 @@ func NewBloomTrieIndexer(db ethdb.Database, odr OdrBackend, parentSize, size uin
diskdb: db,
odr: odr,
trieTable: trieTable,
triedb: trie.NewDatabaseWithCache(trieTable, 1), // Use a tiny cache only to keep memory down
triedb: trie.NewDatabaseWithCache(trieTable, 1, false), // Use a tiny cache only to keep memory down
parentSize: parentSize,
size: size,
}

4
light/trie.go
View file

@ -151,7 +151,7 @@ func (t *odrTrie) do(key []byte, fn func() error) error {
for {
var err error
if t.trie == nil {
t.trie, err = trie.New(t.id.Root, trie.NewDatabase(t.db.backend.Database()))
t.trie, err = trie.New(t.id.Root, trie.NewDatabase(t.db.backend.Database(), false))
}
if err == nil {
err = fn()
@ -177,7 +177,7 @@ func newNodeIterator(t *odrTrie, startkey []byte) trie.NodeIterator {
// Open the actual non-ODR trie if that hasn't happened yet.
if t.trie == nil {
it.do(func() error {
t, err := trie.New(t.id.Root, trie.NewDatabase(t.db.backend.Database()))
t, err := trie.New(t.id.Root, trie.NewDatabase(t.db.backend.Database(), false))
if err == nil {
it.t.trie = t
}

108
trie/database.go
View file

@ -20,6 +20,7 @@ import (
"fmt"
"io"
"sync"
"sync/atomic"
"time"
"github.com/allegro/bigcache"
@ -104,6 +105,9 @@ type Database struct {
diskdb ethdb.Database // Persistent storage for matured trie nodes
noprune map[common.Hash]struct{} // Root hashes of the tries that aren't prunable
pruner *pruner // Background pruner to remove unreferenced trie nodes
pruning uint32 // Flag whether the pruner is running (sanity checks)
cleans *bigcache.BigCache // GC friendly memory cache of clean node RLPs
dirties map[string]*cachedNode // Data and references relationships of dirty nodes
oldest string // Oldest tracked node, flush-list head
@ -115,7 +119,7 @@ type Database struct {
gcnodes uint64 // Nodes garbage collected since last commit
gcsize common.StorageSize // Data storage garbage collected since last commit
prunetime time.Duration // Time spend on disk pruning since last commit
prunetime time.Duration // Time spent on disk pruning since last commit
prunenodes uint64 // Nodes pruned from disk since last commit
prunesize common.StorageSize // Data storage pruned from disk since last commit
@ -344,14 +348,14 @@ func expandNode(hash hashNode, n node, cachegen uint16) node {
// NewDatabase creates a new trie database to store ephemeral trie content before
// its written out to disk or garbage collected. No read cache is created, so all
// data retrievals will hit the underlying disk database.
func NewDatabase(diskdb ethdb.Database) *Database {
return NewDatabaseWithCache(diskdb, 0)
func NewDatabase(diskdb ethdb.Database, prune bool) *Database {
return NewDatabaseWithCache(diskdb, 0, prune)
}
// NewDatabaseWithCache creates a new trie database to store ephemeral trie content
// before its written out to disk or garbage collected. It also acts as a read cache
// for nodes loaded from disk.
func NewDatabaseWithCache(diskdb ethdb.Database, cache int) *Database {
func NewDatabaseWithCache(diskdb ethdb.Database, cache int, prune bool) *Database {
var cleans *bigcache.BigCache
if cache > 0 {
cleans, _ = bigcache.NewBigCache(bigcache.Config{
@ -362,13 +366,48 @@ func NewDatabaseWithCache(diskdb ethdb.Database, cache int) *Database {
HardMaxCacheSize: cache,
})
}
return &Database{
db := &Database{
diskdb: diskdb,
noprune: make(map[common.Hash]struct{}),
cleans: cleans,
dirties: map[string]*cachedNode{metaRoot: {}},
preimages: make(map[common.Hash][]byte),
}
if prune {
db.pruner = newPruner(db)
}
return db
}
// ResumePruning permits the pruner to continue deleting unreferenced trie nodes.
// It is essential to only ever resume pruning after all data is commited, capped
// and properly referenced, otherwise the pruner might delete data that's *going-
// to-be* referenced.
func (db *Database) ResumePruning() {
if db.pruner != nil {
atomic.StoreUint32(&db.pruning, 1)
db.pruner.resume()
}
}
// PausePruning waits until the pruner is done with processing its current task
// and then pauses it so a new trie might be properly integrated into the dirty
// caches and reference counts.
func (db *Database) PausePruning() {
if db.pruner != nil {
atomic.StoreUint32(&db.pruning, 0)
db.pruner.pause()
}
}
// TerminatePruning waits until the pruner is done with processing all its queued
// tasls and then permanently terminates it.
func (db *Database) TerminatePruning() {
if db.pruner != nil {
atomic.StoreUint32(&db.pruning, 0)
db.pruner.terminate()
db.pruner = nil // TODO(karalabe): raceyyyy....
}
}
// ForbidPrune adds a root hash to the list of tries that are disallowed from
@ -562,6 +601,10 @@ func (db *Database) Nodes() []string {
// to break genericity here and assume that parent nodes are not owned (account
// trie) whereas child nodes may be owned (storage trie or bytecode).
func (db *Database) Reference(owner common.Hash, child common.Hash, parent common.Hash) {
// If pruning is enabled and running, something's very wrong
if db.pruner != nil && atomic.LoadUint32(&db.pruning) == 1 {
panic("pruner running during referencing")
}
db.lock.Lock()
defer db.lock.Unlock()
@ -583,7 +626,11 @@ func (db *Database) Reference(owner common.Hash, child common.Hash, parent commo
}
// Dereference removes an existing reference from a root node.
func (db *Database) Dereference(root common.Hash, prune bool) error {
func (db *Database) Dereference(root common.Hash) error {
// If pruning is enabled and running, something's very wrong
if db.pruner != nil && atomic.LoadUint32(&db.pruning) == 1 {
panic("pruner running during dereferencing")
}
// Sanity check to ensure that the meta-root is not removed
if root == (common.Hash{}) {
log.Error("Attempted to dereference the trie cache meta root")
@ -593,15 +640,10 @@ func (db *Database) Dereference(root common.Hash, prune bool) error {
db.lock.Lock()
defer db.lock.Unlock()
nodes, storage, start := len(db.dirties), db.dirtiesSize, time.Now()
prunetime, prunenodes, prunesize := db.prunetime, db.prunenodes, db.prunesize
// Dereference the trie and accumulate prune targets if needed
var pruner *pruner
if prune {
pruner = db.newPruner()
}
if err := db.dereference(common.Hash{}, root, common.Hash{}, common.Hash{}, nil, pruner); err != nil {
nodes, storage, start := len(db.dirties), db.dirtiesSize, time.Now()
if err := db.dereference(common.Hash{}, root, common.Hash{}, common.Hash{}, nil); err != nil {
return err
}
db.gcnodes += uint64(nodes - len(db.dirties))
@ -612,31 +654,11 @@ func (db *Database) Dereference(root common.Hash, prune bool) error {
memcacheGCSizeMeter.Mark(int64(storage - db.dirtiesSize))
memcacheGCNodesMeter.Mark(int64(nodes - len(db.dirties)))
// If pruning was requested, execute on a background thread
go func() {
db.lock.RLock()
defer db.lock.RUnlock()
if pruner != nil {
start := time.Now()
pruner.execute()
go func() {
if err := pruner.flush(); err != nil {
log.Crit("Failed to prune database", "err", err)
}
}()
db.prunetime += time.Since(start) // TODO(karalabe): unsafe, stats are off too
}
// Pruned or not, update the stats and log
memcachePruneTimeTimer.Update(db.prunetime - prunetime)
memcachePruneNodesMeter.Mark(int64(db.prunenodes - prunenodes))
memcachePruneSizeMeter.Mark(int64(db.prunesize - prunesize))
}()
return nil
}
// dereference is the private locked version of Dereference.
func (db *Database) dereference(childOwner common.Hash, childHash common.Hash, parentOwner common.Hash, parentHash common.Hash, path []byte, pruner *pruner) error {
func (db *Database) dereference(childOwner common.Hash, childHash common.Hash, parentOwner common.Hash, parentHash common.Hash, path []byte) error {
// Dereference the parent-child
parentKey := makeNodeKey(parentOwner, parentHash)
parent := db.dirties[parentKey]
@ -651,8 +673,8 @@ func (db *Database) dereference(childOwner common.Hash, childHash common.Hash, p
// If the child does not exist, it's a previously committed node.
child, ok := db.dirties[childKey]
if !ok {
if pruner != nil {
pruner.mark(childOwner, childHash, path)
if db.pruner != nil {
db.pruner.enqueue(childOwner, childHash, path)
}
return nil
}
@ -679,12 +701,12 @@ func (db *Database) dereference(childOwner common.Hash, childHash common.Hash, p
}
// Dereference all children and delete the node
child.iterateRefs(path, func(path []byte, hash common.Hash) error {
db.dereference(childOwner, hash, childOwner, childHash, path, pruner)
db.dereference(childOwner, hash, childOwner, childHash, path)
return nil
})
for key := range child.children {
owner, hash := splitNodeKey(key)
db.dereference(owner, hash, childOwner, childHash, nil, pruner)
db.dereference(owner, hash, childOwner, childHash, nil)
}
delete(db.dirties, childKey)
db.dirtiesSize -= common.StorageSize(common.HashLength + int(child.size))
@ -695,6 +717,10 @@ func (db *Database) dereference(childOwner common.Hash, childHash common.Hash, p
// Cap iteratively flushes old but still referenced trie nodes until the total
// memory usage goes below the given threshold.
func (db *Database) Cap(limit common.StorageSize) error {
// If pruning is enabled and running, something's very wrong
if db.pruner != nil && atomic.LoadUint32(&db.pruning) == 1 {
panic("pruner running during capping")
}
// Create a database batch to flush persistent data out. It is important that
// outside code doesn't see an inconsistent state (referenced data removed from
// memory cache during commit but not yet in persistent storage). This is ensured
@ -798,6 +824,10 @@ func (db *Database) Cap(limit common.StorageSize) error {
//
// As a side effect, all pre-images accumulated up to this point are also written.
func (db *Database) Commit(node common.Hash, report bool) error {
// If pruning is enabled and running, something's very wrong
if db.pruner != nil && atomic.LoadUint32(&db.pruning) == 1 {
panic("pruner running during committing")
}
// Create a database batch to flush persistent data out. It is important that
// outside code doesn't see an inconsistent state (referenced data removed from
// memory cache during commit but not yet in persistent storage). This is ensured

235
trie/pruning.go
View file

@ -20,6 +20,7 @@ import (
"bytes"
"fmt"
"math/big"
"time"
"github.com/ethereum/go-ethereum/common"
"github.com/ethereum/go-ethereum/ethdb"
@ -30,12 +31,17 @@ import (
// pruner is responsible for pruning the state trie based on liveness checks
// whenever the in-memory garbage collector attempt to dereference a node from
// disk.
//
// Note, the pruner is not a standalone construct, rather an extension to the
// trie database. No attempt was made to separate the API surface and make one
// a disjoint client of the other.
type pruner struct {
db *Database // Trie database for accessing dirty and clean data
tries []*traverser // Individual stateful trie traversers for fast liveness checks
marks []*prunerTarget // Nodes marked for potential pruning
batch ethdb.Batch // Write batch to minimize database trashing
taskCh chan *prunerTarget // Task queue receiving the pruning targets to delete
pauseCh chan chan struct{} // Notification channel to pause the pruner
resumeCh chan chan struct{} // Notification channel to resume the pruner
terminateCh chan chan struct{} // Notification channel to terminate the pruner
}
// prunerTarget represents a single marked target for potential pruning.
@ -45,63 +51,214 @@ type prunerTarget struct {
hash common.Hash // Hash of the node to delete
}
// newPruner creates a new trie pruner tied to the liveness of all the currently
// referenced in-memory nodes.
func (db *Database) newPruner() *pruner {
return &pruner{
// newPruner creates a new background trie pruner to delete unreferenced nodes
// whenever the tries are not being actively written.
func newPruner(db *Database) *pruner {
p := &pruner{
db: db,
batch: db.diskdb.NewBatch(),
taskCh: make(chan *prunerTarget, 128),
pauseCh: make(chan chan struct{}),
resumeCh: make(chan chan struct{}),
terminateCh: make(chan chan struct{}),
}
go p.loop()
return p
}
// mark adds a new prune target to be deleted on the pruning run.
func (p *pruner) mark(owner common.Hash, hash common.Hash, path []byte) {
p.marks = append(p.marks, &prunerTarget{
// enqueue adds a potential prune target to the removal queue to be inspected and
// removed from the database if deemed unreferenced by recent and snapshot tries.
func (p *pruner) enqueue(owner common.Hash, hash common.Hash, path []byte) {
p.taskCh <- &prunerTarget{
owner: owner,
hash: hash,
path: common.CopyBytes(path),
})
}
}
// execute runs the pruning procedure, deleting everything that has no live
// reference any more.
func (p *pruner) execute() {
// Create the set of traversers based on the live tries
// resume (re)starts the pruning, locking the dirty caches for reads to prevent
// trie nodes going missing due to concurrent pruning/referencing.
//
// Note, calling resume on an already running pruner will deadlock! The pruner is
// initially paused.
func (p *pruner) resume() {
// We *must* wait for the pruner to obtain the lock, otherwise the caller might
// race forward and lock the database for writing, messing up the state machine.
ch := make(chan struct{})
p.resumeCh <- ch
<-ch
}
// pause signals the pruner to interrupt its operation and release its held lock.
// This is needed for the block processor to obtain a write lock on the dirty
// caches, which are otherwise held hostage by the pruner.
//
// Note, calling pause on a non-running pruner will panic! The pruner is initially
// paused.
func (p *pruner) pause() {
// We don't really need to wait for the pause to complete here as we're unable
// to obtain a write-lock sooner anyway, but it's perhaps nicer code to make it
// symmetrical to `resume`.
ch := make(chan struct{})
p.pauseCh <- ch
<-ch
}
// terminate signals the pruner to finish all remaining tasks and permanently
// release all locks and clean itself up.
func (p *pruner) terminate() {
ch := make(chan struct{})
p.terminateCh <- ch
<-ch
}
// loop is the pruner background gorutineo that waits for pruning targets the be
// added, causing liveness checks and potentially database deletions in response.
func (p *pruner) loop() {
var (
runner chan struct{} // Runner channel acting as a boolean 'running' flag
tasks []*prunerTarget // Batch of trie nodes queued for potential pruning
tries []*traverser // Individual trie traversers for liveness checks
done int // Number of pruning tasks done, for smarter CG
batch = p.db.diskdb.NewBatch() // Create a write batch to minimize thrashing
start time.Time // Time instance when the pruner was resumed
nodes uint64 // Number of nodes pruned when the pruner was resumed
size common.StorageSize // Number of bytes pruned when the pruner was resumed
quit chan struct{} // Quit signal channel when termination is requested
quitting <-chan time.Time // Ticker to periodically log termination progress
)
// Wait for different events and process them accordingly
for {
select {
case task := <-p.taskCh:
// New task received, queue it up. We will not start immediately processing
// this as the enqueueing is done whilst doing in-memory garbage collection,
// so the dirty caches are locked for writing.
tasks = append(tasks, task)
case ch := <-p.resumeCh:
// Pruner was requested to resume operation. Obtain the necessary locks to
// prevent the block processor for modifying the dirty caches, but allow any
// goroutines to still read the data.
p.db.lock.RLock()
ch <- struct{}{} // signal back that the lock was obtained
// Only proceed with task processing if there's something available
if len(tasks) > 0 {
// Create a runner channel that will allow running whenever checked
runner = make(chan struct{})
close(runner)
// Ensure the traversers are pointing to the currently live tries. Usually
// after each pause/resume cycle, one (new block) or two (new snapshot) tries
// get swapped out.
tries = nil // cheat a bit for now and just reconstruct them
for key := range p.db.dirties[metaRoot].children {
_, root := splitNodeKey(key)
p.tries = append(p.tries, &traverser{
tries = append(tries, &traverser{
db: p.db,
state: &traverserState{
node: hashNode(root[:]),
hash: root,
},
state: &traverserState{hash: root, node: hashNode(root[:])},
})
}
// Beside all the tries kept in memory, keep anything forbidden from pruning
for hash := range p.db.noprune {
p.tries = append(p.tries, &traverser{
tries = append(tries, &traverser{
db: p.db,
state: &traverserState{
node: hashNode(common.CopyBytes(hash[:])), // Need closure, take care!!
hash: hash,
},
state: &traverserState{hash: hash, node: hashNode(common.CopyBytes(hash[:]))}, // need closure!
})
}
// Iterate over all the nodes marked for pruning and delete them
for _, mark := range p.marks {
p.prune(mark.owner, mark.hash, mark.path)
}
}
// Mark the resumption to track the pruning time
start, nodes, size = time.Now(), p.db.prunenodes, p.db.prunesize
// flush commits any pending database writes. It does not reset the batch since
// we only ever supposed to commit once per prune run.
func (p *pruner) flush() error {
return p.batch.Write()
case ch := <-p.pauseCh:
// Pruner was requestd to pause operation. We can just release the read lock
// and stop processing the queued tasks.
// Destroy the runner, disabling the deletion part of the event loop.
if runner != nil {
memcachePruneNodesMeter.Mark(int64(p.db.prunenodes - nodes))
memcachePruneSizeMeter.Mark(int64(p.db.prunesize - size))
memcachePruneTimeTimer.Update(time.Since(start))
p.db.prunetime += time.Since(start)
runner = nil
}
// Signal back that the lock was released and nothing touches the database
// filds any more.
p.db.lock.RUnlock()
ch <- struct{}{}
// If we have anything queued up for writing, might as well push it out now
if batch.ValueSize() > 0 {
if err := batch.Write(); err != nil {
log.Crit("Failed to flush pruned nodes", "err", err)
}
}
batch.Reset()
case quit = <-p.terminateCh:
// Pruner was requetsed to terminate. If everything was already processed, we
// can exit cleanly. Otherwise we must schedule a cleanup.
if len(tasks) == 0 {
p.db.lock.RUnlock()
quit <- struct{}{}
return
}
// Still some tasks left, create a progress ticker to not hang the user
log.Info("Pruner finishing pending jobs", "count", len(tasks))
quitter := time.NewTicker(8 * time.Second)
defer quitter.Stop()
quitting = quitter.C
case <-quitting:
// A bit of time passed since the last info log, print our progress
log.Info("Pruner finishing pending jobs", "count", len(tasks))
case <-runner:
// No interesting events available, but pruner is permitted to delete queued
// up tasks. Process the next one.
p.prune(tasks[0].owner, tasks[0].hash, tasks[0].path, tries, batch)
// Delete the task from the queue. Here let's be a bit smarter to prevent the
// task slice growing indefinitely.
if done++; done%1024 == 0 {
tasks = append([]*prunerTarget{}, tasks[1:]...)
} else {
tasks = tasks[1:]
}
// If we're out of pruning tasks, stop looping the runner (but don't release
// the lock, that's up to higher layer code to request).
if len(tasks) == 0 {
// Update all the stats and disable the runner
memcachePruneNodesMeter.Mark(int64(p.db.prunenodes - nodes))
memcachePruneSizeMeter.Mark(int64(p.db.prunesize - size))
memcachePruneTimeTimer.Update(time.Since(start))
p.db.prunetime += time.Since(start)
runner = nil
// If we're actually shutting down, clean up everything
if quit != nil {
if err := batch.Write(); err != nil {
log.Crit("Failed to flush pruned nodes", "err", err)
}
batch.Reset()
p.db.lock.RUnlock()
quit <- struct{}{}
return
}
}
}
}
}
// prune deletes a trie node from disk if there are no more live references to
// it, cascading until all dangling nodes are removed.
func (p *pruner) prune(owner common.Hash, hash common.Hash, path []byte) {
func (p *pruner) prune(owner common.Hash, hash common.Hash, path []byte, tries []*traverser, batch ethdb.Batch) {
// If the node is still live in the memory cache, it's still referenced so we
// can abort. This case is important when and old trie being pruned references
// a new node (maybe that node was recreted since), since currently live nodes
@ -116,7 +273,7 @@ func (p *pruner) prune(owner common.Hash, hash common.Hash, path []byte) {
crosspath = append(append(keybytesToHex(owner[:]), 0xff), crosspath...)
}
unrefs := make(map[common.Hash]bool)
for _, trie := range p.tries {
for _, trie := range tries {
// If the node is still live, abort
if trie.live(owner, hash, crosspath, unrefs) {
return
@ -135,12 +292,12 @@ func (p *pruner) prune(owner common.Hash, hash common.Hash, path []byte) {
// Prune the node and its children if it's not a bytecode blob
p.db.cleans.Delete(string(hash[:]))
p.batch.Delete(dead)
batch.Delete(dead)
p.db.prunenodes++
p.db.prunesize += common.StorageSize(len(blob))
iterateRefs(node, path, func(path []byte, hash common.Hash) error {
p.prune(owner, hash, path)
p.prune(owner, hash, path, tries, batch)
return nil
})
}