From 8a301cc929ab0b92d63eb4f8b6388acc396b5fb0 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?P=C3=A9ter=20Szil=C3=A1gyi?= Date: Mon, 14 Jan 2019 12:53:36 +0200 Subject: [PATCH] core, trie: move state pruner onto its own background thread --- core/blockchain.go | 21 +++- core/state/database.go | 2 +- eth/api_tracer.go | 6 +- les/handler.go | 4 +- light/postprocess.go | 4 +- light/trie.go | 4 +- trie/database.go | 108 ++++++++++------- trie/pruning.go | 259 +++++++++++++++++++++++++++++++++-------- 8 files changed, 302 insertions(+), 106 deletions(-) diff --git a/core/blockchain.go b/core/blockchain.go index 8da18f8853..521413ebdc 100644 --- a/core/blockchain.go +++ b/core/blockchain.go @@ -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)) } } } diff --git a/core/state/database.go b/core/state/database.go index 01872bf3c4..600d0a7cf3 100644 --- a/core/state/database.go +++ b/core/state/database.go @@ -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, } } diff --git a/eth/api_tracer.go b/eth/api_tracer.go index b682376ec5..f1d9c88e54 100644 --- a/eth/api_tracer.go +++ b/eth/api_tracer.go @@ -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 } diff --git a/les/handler.go b/les/handler.go index eb89d032f4..7bfaf39fcc 100644 --- a/les/handler.go +++ b/les/handler.go @@ -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 { diff --git a/light/postprocess.go b/light/postprocess.go index dd1b74a7be..5019921fd1 100644 --- a/light/postprocess.go +++ b/light/postprocess.go @@ -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, } diff --git a/light/trie.go b/light/trie.go index 4d4abe1375..8587b42b7d 100644 --- a/light/trie.go +++ b/light/trie.go @@ -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 } diff --git a/trie/database.go b/trie/database.go index f701464c90..6270ba5364 100644 --- a/trie/database.go +++ b/trie/database.go @@ -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 diff --git a/trie/pruning.go b/trie/pruning.go index 3305b7defe..26738d3a8d 100644 --- a/trie/pruning.go +++ b/trie/pruning.go @@ -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 + db *Database // Trie database for accessing dirty and clean data - 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{ - db: db, - batch: db.diskdb.NewBatch(), +// 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, + 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 - for key := range p.db.dirties[metaRoot].children { - _, root := splitNodeKey(key) - p.tries = append(p.tries, &traverser{ - db: p.db, - state: &traverserState{ - node: hashNode(root[:]), - hash: 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{ - db: p.db, - state: &traverserState{ - node: hashNode(common.CopyBytes(hash[:])), // Need closure, take care!! - hash: hash, - }, - }) - } - // Iterate over all the nodes marked for pruning and delete them - for _, mark := range p.marks { - p.prune(mark.owner, mark.hash, mark.path) } } -// 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() +// 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) + tries = append(tries, &traverser{ + db: p.db, + state: &traverserState{hash: root, node: hashNode(root[:])}, + }) + } + for hash := range p.db.noprune { + tries = append(tries, &traverser{ + db: p.db, + state: &traverserState{hash: hash, node: hashNode(common.CopyBytes(hash[:]))}, // need closure! + }) + } + } + // Mark the resumption to track the pruning time + start, nodes, size = time.Now(), p.db.prunenodes, p.db.prunesize + + 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 }) }