diff --git a/trie/database.go b/trie/database.go index 6270ba5364..6fafec953a 100644 --- a/trie/database.go +++ b/trie/database.go @@ -643,9 +643,13 @@ func (db *Database) Dereference(root common.Hash) error { // Dereference the trie and accumulate prune targets if needed nodes, storage, start := len(db.dirties), db.dirtiesSize, time.Now() - if err := db.dereference(common.Hash{}, root, common.Hash{}, common.Hash{}, nil); err != nil { + derefs := new([]*prunerTarget) + if err := db.dereference(common.Hash{}, root, common.Hash{}, common.Hash{}, nil, derefs); err != nil { return err } + if db.pruner != nil { + db.pruner.enqueue(*derefs) + } db.gcnodes += uint64(nodes - len(db.dirties)) db.gcsize += storage - db.dirtiesSize db.gctime += time.Since(start) @@ -658,7 +662,7 @@ func (db *Database) Dereference(root common.Hash) error { } // 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) error { +func (db *Database) dereference(childOwner common.Hash, childHash common.Hash, parentOwner common.Hash, parentHash common.Hash, path []byte, derefs *[]*prunerTarget) error { // Dereference the parent-child parentKey := makeNodeKey(parentOwner, parentHash) parent := db.dirties[parentKey] @@ -674,7 +678,11 @@ func (db *Database) dereference(childOwner common.Hash, childHash common.Hash, p child, ok := db.dirties[childKey] if !ok { if db.pruner != nil { - db.pruner.enqueue(childOwner, childHash, path) + *derefs = append(*derefs, &prunerTarget{ + owner: childOwner, + hash: childHash, + path: common.CopyBytes(path), + }) } return nil } @@ -701,12 +709,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) + db.dereference(childOwner, hash, childOwner, childHash, path, derefs) return nil }) for key := range child.children { owner, hash := splitNodeKey(key) - db.dereference(owner, hash, childOwner, childHash, nil) + db.dereference(owner, hash, childOwner, childHash, nil, derefs) } delete(db.dirties, childKey) db.dirtiesSize -= common.StorageSize(common.HashLength + int(child.size)) diff --git a/trie/pruning.go b/trie/pruning.go index 26738d3a8d..05a7f3ac3e 100644 --- a/trie/pruning.go +++ b/trie/pruning.go @@ -20,6 +20,7 @@ import ( "bytes" "fmt" "math/big" + "sync/atomic" "time" "github.com/ethereum/go-ethereum/common" @@ -38,10 +39,11 @@ import ( type pruner struct { db *Database // Trie database for accessing dirty and clean data - 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 + taskCh chan []*prunerTarget // Task queue receiving the pruning targets to delete + abortCh chan chan struct{} // Notification channel to terminate the pruner + resumeCh chan chan struct{} // Notification channel to resume the pruner + + interrupt uint32 // Signals to a running deep pruning to suspend itself } // prunerTarget represents a single marked target for potential pruning. @@ -55,32 +57,36 @@ type prunerTarget struct { // 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{}), + db: db, + taskCh: make(chan []*prunerTarget), + abortCh: make(chan chan struct{}), + resumeCh: make(chan chan struct{}), } go p.loop() return p } -// 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), - } +// enqueue adds a batch of potential prune targets to the removal queue to be +// inspected and removed from the database if deemed unreferenced by recent +// and snapshot tries. +// +// It's important to queue in batches as a single block might enque hundreds or +// thousands of targets. Queueing individually entails a huge performance hit. +func (p *pruner) enqueue(targets []*prunerTarget) { + p.taskCh <- targets } // 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. +// Note, calling resume on an already running pruner will deadlock! func (p *pruner) resume() { + // The prumer might have been interrupted previously, so we need to ensure the + // interrut is cleared before requesting a resumption. This could be done by the + // pruner ron loop too, but figured it might be cleaner to set the interrupt at + // the same scope as with `pause`, + atomic.StoreUint32(&p.interrupt, 0) + // 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{}) @@ -92,22 +98,21 @@ func (p *pruner) resume() { // 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. +// Note, calling pause on a non-running pruner will panic! 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 + // Notify the pruner to abort right now. + atomic.StoreUint32(&p.interrupt, 1) } // terminate signals the pruner to finish all remaining tasks and permanently // release all locks and clean itself up. +// +// Note, calling terminate on a non-running pruner will panic! func (p *pruner) terminate() { + // Signal to the pruner that it should terminate itself gracefully and wait for + // it to confirm before pulling the rug from underneath. ch := make(chan struct{}) - p.terminateCh <- ch + p.abortCh <- ch <-ch } @@ -115,157 +120,108 @@ func (p *pruner) terminate() { // 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 + tasks []*prunerTarget // Batch of trie nodes queued for potential pruning + tries []*traverser // Individual trie traversers for liveness checks + quit chan struct{} // Quit signal channel when termination is requested 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: + case targets := <-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) + tasks = append(tasks, targets...) 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. + if len(tasks) == 0 { + ch <- struct{}{} // signal back, but nothing to do really + continue + } 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 - // 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! + }) + } + // Process the tasks until an interrupt arrives + start, nodes, size := time.Now(), p.db.prunenodes, p.db.prunesize - 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! - }) + interrupted := false + for i, task := range tasks { + remain := p.prune(task.owner, task.hash, task.path, tries, batch) + if len(remain) > 0 { + tasks = append(remain, tasks[i+1:]...) + interrupted = true + break } } - // 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 + // If all tasks have been procesed, get rid of any allocated task slice and + // terminate the runner pathway. + if !interrupted { + tasks = nil } - // Signal back that the lock was released and nothing touches the database - // filds any more. + // Update all the stats with the results until now + 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) 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) - } + // Push any change to disk + 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() + // If we're actually shutting down, clean up everything + if quit != nil { 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 - } - } + case quit = <-p.abortCh: + // Pruner was requetsed to terminate. Since termination doesn't interrupt, we + // can at this point safely assume everything was pruned. + 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, tries []*traverser, batch ethdb.Batch) { +// it, cascading until all dangling nodes are removed. If the pruner's interrupt +// has been triggered (block processing pending), the remaining nodes are bubbled +// up to the caller to reschedule later. +func (p *pruner) prune(owner common.Hash, hash common.Hash, path []byte, tries []*traverser, batch ethdb.Batch) []*prunerTarget { // 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 // are stored expanded, not as hashes. key := makeNodeKey(owner, hash) if p.db.dirties[key] != nil { - return + return nil } // Iterate over all the live tries and check node liveliness crosspath := path @@ -276,7 +232,7 @@ func (p *pruner) prune(owner common.Hash, hash common.Hash, path []byte, tries [ for _, trie := range tries { // If the node is still live, abort if trie.live(owner, hash, crosspath, unrefs) { - return + return nil } // Node dead in this trie, cache the result for subsequent traversals trie.unref(2, unrefs) @@ -286,7 +242,7 @@ func (p *pruner) prune(owner common.Hash, hash common.Hash, path []byte, tries [ blob, err := p.db.diskdb.Get(dead) if blob == nil || err != nil { log.Error("Missing prune target", "owner", owner, "hash", hash, "path", fmt.Sprintf("%x", path)) - return + return nil } node := mustDecodeNode(hash[:], blob, 0) @@ -296,10 +252,20 @@ func (p *pruner) prune(owner common.Hash, hash common.Hash, path []byte, tries [ p.db.prunenodes++ p.db.prunesize += common.StorageSize(len(blob)) + var remain []*prunerTarget iterateRefs(node, path, func(path []byte, hash common.Hash) error { - p.prune(owner, hash, path, tries, batch) + // If the pruner was interrupted, accumulate the remaining targets + if atomic.LoadUint32(&p.interrupt) == 1 { + remain = append(remain, &prunerTarget{owner: owner, hash: hash, path: common.CopyBytes(path)}) + return nil + } + // Pruning not interrupted until now, attempt to process children too. It's + // fine to assign the result directly to the `remain` slice because it's nil + // anyway until the interrupt triggers. + remain = p.prune(owner, hash, path, tries, batch) return nil }) + return remain } // traverser is a stateful trie traversal data structure used by the pruner to