trie: support mid-trie pruning interruption

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))

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
+	taskCh   chan []*prunerTarget // Task queue receiving the pruning targets to delete
-	pauseCh     chan chan struct{} // Notification channel to pause the pruner
+	abortCh  chan chan struct{}   // Notification channel to terminate the pruner
 	resumeCh chan chan struct{}   // Notification channel to resume the pruner
-	terminateCh chan chan struct{} // Notification channel to terminate the pruner
+
+	interrupt uint32 // Signals to a running deep pruning to suspend itself
 }
 
 // prunerTarget represents a single marked target for potential pruning.
@@ -56,31 +58,35 @@ type prunerTarget struct {
 func newPruner(db *Database) *pruner {
 	p := &pruner{
 		db:       db,
-		taskCh:      make(chan *prunerTarget, 128),
+		taskCh:   make(chan []*prunerTarget),
-		pauseCh:     make(chan chan struct{}),
+		abortCh:  make(chan chan struct{}),
 		resumeCh: make(chan chan struct{}),
-		terminateCh: make(chan chan struct{}),
 	}
 	go p.loop()
 	return p
 }
 
-// enqueue adds a potential prune target to the removal queue to be inspected and
+// enqueue adds a batch of potential prune targets to the removal queue to be
-// removed from the database if deemed unreferenced by recent and snapshot tries.
+// inspected and removed from the database if deemed unreferenced by recent
-func (p *pruner) enqueue(owner common.Hash, hash common.Hash, path []byte) {
+// and snapshot tries.
-	p.taskCh <- &prunerTarget{
+//
-		owner: owner,
+// It's important to queue in batches as a single block might enque hundreds or
-		hash:  hash,
+// thousands of targets. Queueing individually entails a huge performance hit.
-		path:  common.CopyBytes(path),
+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
+// Note, calling resume on an already running pruner will deadlock!
-// initially paused.
 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
+// Note, calling pause on a non-running pruner will panic!
-// paused.
 func (p *pruner) pause() {
-	// We don't really need to wait for the pause to complete here as we're unable
+	// Notify the pruner to abort right now.
-	// to obtain a write-lock sooner anyway, but it's perhaps nicer code to make it
+	atomic.StoreUint32(&p.interrupt, 1)
-	// 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.
+//
+// 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,42 +120,32 @@ 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
+		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.
@@ -169,103 +164,64 @@ func (p *pruner) loop() {
 					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
+
+			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
+			// If all tasks have been procesed, get rid of any allocated task slice and
-
+			// terminate the runner pathway.
-		case ch := <-p.pauseCh:
+			if !interrupted {
-			// Pruner was requestd to pause operation. We can just release the read lock
+				tasks = nil
-			// and stop processing the queued tasks.
+			}
-
+			// Update all the stats with the results until now
-			// 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
+			p.db.prunetime += time.Since(start)
-			if batch.ValueSize() > 0 {
+			p.db.lock.RUnlock()
+
+			// 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()
-				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.
+// it, cascading until all dangling nodes are removed. If the pruner's interrupt
-func (p *pruner) prune(owner common.Hash, hash common.Hash, path []byte, tries []*traverser, batch ethdb.Batch) {
+// 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