trie: replace naive pruner queue with a proper deque

trie/pruning.go

@@ -28,6 +28,7 @@ import (
 	"github.com/ethereum/go-ethereum/log"
 	"github.com/ethereum/go-ethereum/metrics"
 	"github.com/ethereum/go-ethereum/rlp"
+	"github.com/karalabe/cookiejar/collections/deque"
 )
 
 var (
@@ -36,9 +37,10 @@ var (
 	memcachePruneSizeMeter  = metrics.NewRegisteredMeter("trie/memcache/prune/size", nil)
 
 	memcachePruneAssignHistogram    = metrics.NewRegisteredHistogram("trie/memcache/prune/assign", nil, metrics.NewUniformSample(1024))
+	memcachePruneAssignDupHistogram = metrics.NewRegisteredHistogram("trie/memcache/prune/assigndup", nil, metrics.NewUniformSample(1024))
 	memcachePruneRemainHistogram    = metrics.NewRegisteredHistogram("trie/memcache/prune/remain", nil, metrics.NewUniformSample(1024))
+	memcachePruneRemainDupHistogram = metrics.NewRegisteredHistogram("trie/memcache/prune/remaindup", nil, metrics.NewUniformSample(1024))
 	memcachePruneQueueHistogram     = metrics.NewRegisteredHistogram("trie/memcache/prune/queue", nil, metrics.NewUniformSample(1024))
-	memcachePruneDedupHistogram  = metrics.NewRegisteredHistogram("trie/memcache/prune/dedup", nil, metrics.NewUniformSample(1024))
 )
 
 // pruner is responsible for pruning the state trie based on liveness checks
@@ -132,7 +134,7 @@ func (p *pruner) terminate() {
 // added, causing liveness checks and potentially database deletions in response.
 func (p *pruner) loop() {
 	var (
-		tasks   []*prunerTarget             // Batch of trie nodes queued for potential pruning
+		tasks   = deque.New()               // Queue of trie nodes queued for potential pruning
 		taskset = make(map[string]struct{}) // Set of trie nodes queued to prevent duplication
 
 		tries []*traverser  // Individual trie traversers for liveness checks
@@ -154,18 +156,18 @@ func (p *pruner) loop() {
 					duplicates++
 					continue
 				}
-				tasks = append(tasks, task)
+				tasks.PushRight(task)
 				taskset[key] = struct{}{}
 			}
+			memcachePruneQueueHistogram.Update(int64(tasks.Size()))
 			memcachePruneAssignHistogram.Update(int64(len(targets)))
-			memcachePruneQueueHistogram.Update(int64(len(tasks)))
+			memcachePruneAssignDupHistogram.Update(int64(duplicates))
-			memcachePruneDedupHistogram.Update(int64(duplicates))
 
 		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 {
+			if tasks.Size() == 0 {
 				ch <- struct{}{} // signal back, but nothing to do really
 				continue
 			}
@@ -193,42 +195,41 @@ func (p *pruner) loop() {
 			// Process the tasks until an interrupt arrives
 			start, nodes, size := time.Now(), p.db.prunenodes, p.db.prunesize
 
-			interrupted := false
+			for !tasks.Empty() {
-			for i, task := range tasks {
 				// Delete this particular task from the deduplication set
+				task := tasks.PopLeft().(*prunerTarget)
 				delete(taskset, makeNodeKey(task.owner, task.hash))
 
+				// Prune the target and reschedule any interrupted sub-tasks
 				remain := p.prune(task.owner, task.hash, task.path, taskset, tries, batch)
-				if len(remain) > 0 {
+				if atomic.LoadUint32(&p.interrupt) == 1 {
-					// Schedule any newly discovered but interrupted tasks for later
+					duplicates := 0
-					for j := 0; j < len(remain); j++ {
+					for j := len(remain) - 1; j >= 0; j-- { // reverse to keep the depth priority
 						// Dedup already scheduled tasks, no need to prune twice
 						key := makeNodeKey(remain[j].owner, remain[j].hash)
 						if _, exist := taskset[key]; exist {
-							if j == 0 {
+							duplicates++
-								remain = remain[j+1:]
-							} else {
-								remain = append(remain[:j-1], remain[j+1:]...)
-							}
-							j--
 							continue
 						}
+						// Reschedule (high priority) anything that's not a duplicate
+						tasks.PushLeft(remain[j])
 						taskset[key] = struct{}{}
 					}
-					tasks = append(remain, tasks[i+1:]...)
+					memcachePruneQueueHistogram.Update(int64(tasks.Size()))
-					interrupted = true
-
 					memcachePruneRemainHistogram.Update(int64(len(remain)))
-					memcachePruneQueueHistogram.Update(int64(len(tasks)))
+					memcachePruneRemainDupHistogram.Update(int64(duplicates))
 
 					break
 				}
 			}
 			// If all tasks have been procesed, get rid of any allocated task slice and
 			// terminate the runner pathway.
-			if !interrupted {
+			if tasks.Empty() {
+				tasks.Reset()
+
 				memcachePruneQueueHistogram.Update(0)
-				tasks = nil
+				memcachePruneRemainHistogram.Update(0)
+				memcachePruneRemainDupHistogram.Update(0)
 			}
 			// Update all the stats with the results until now
 			memcachePruneNodesMeter.Mark(int64(p.db.prunenodes - nodes))
@@ -266,7 +267,7 @@ func (p *pruner) loop() {
 func (p *pruner) prune(owner common.Hash, hash common.Hash, path []byte, taskset map[string]struct{}, tries []*traverser, batch ethdb.Batch) []*prunerTarget {
 	// If the node is already queued for pruning, don't duplicate any effort on it
 	key := makeNodeKey(owner, hash)
-	if _, ok := taskset[makeNodeKey(owner, hash)]; ok {
+	if _, ok := taskset[key]; ok {
 		return nil
 	}
 	// If the node is still live in the memory cache, it's still referenced so we
@@ -291,7 +292,7 @@ func (p *pruner) prune(owner common.Hash, hash common.Hash, path []byte, taskset
 		trie.unref(2, unrefs)
 	}
 	// Dead node found, delete it from the database
-	dead := []byte(makeNodeKey(owner, hash))
+	dead := []byte(key)
 	blob, err := p.db.diskdb.Get(dead)
 	if blob == nil || err != nil {
 		// Node already deleted by something else, happens with delayed pruning

vendor/github.com/karalabe/cookiejar/LICENSE

@@ -0,0 +1,25 @@
+Copyright (c) 2014 Péter Szilágyi. All rights reserved.
+
+Redistribution and use in source and binary forms, with or without modification,
+are permitted provided that the following conditions are met:
+
+    * Redistributions of source code must retain the above copyright notice,
+      this list of conditions and the following disclaimer.
+    * Redistributions in binary form must reproduce the above copyright notice,
+      this list of conditions and the following disclaimer in the documentation
+      and/or other materials provided with the distribution.
+
+THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS" AND
+ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED
+WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE
+DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT HOLDER OR CONTRIBUTORS BE LIABLE FOR
+ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES
+(INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES;
+LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON
+ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
+(INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF THIS
+SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
+
+Alternatively, the CookieJar toolbox may be used in accordance with the terms
+and conditions contained in a signed written agreement between you and the
+author(s).

vendor/github.com/karalabe/cookiejar/collections/deque/deque.go

@@ -0,0 +1,141 @@
+// CookieJar - A contestant's algorithm toolbox
+// Copyright (c) 2013 Peter Szilagyi. All rights reserved.
+//
+// CookieJar is dual licensed: use of this source code is governed by a BSD
+// license that can be found in the LICENSE file. Alternatively, the CookieJar
+// toolbox may be used in accordance with the terms and conditions contained
+// in a signed written agreement between you and the author(s).
+
+// Package deque implements a double ended queue supporting arbitrary types
+// (even a mixture).
+//
+// Internally it uses a dynamically growing circular slice of blocks, resulting
+// in faster resizes than a simple dynamic array/slice would allow, yet less gc
+// overhead.
+package deque
+
+// The size of a block of data
+const blockSize = 4096
+
+// Double ended queue data structure.
+type Deque struct {
+	leftIdx  int
+	leftOff  int
+	rightIdx int
+	rightOff int
+
+	blocks [][]interface{}
+	left   []interface{}
+	right  []interface{}
+}
+
+// Creates a new, empty deque.
+func New() *Deque {
+	result := new(Deque)
+	result.blocks = [][]interface{}{make([]interface{}, blockSize)}
+	result.right = result.blocks[0]
+	result.left = result.blocks[0]
+	return result
+}
+
+// Pushes a new element into the queue from the right, expanding it if necessary.
+func (d *Deque) PushRight(data interface{}) {
+	d.right[d.rightOff] = data
+	d.rightOff++
+	if d.rightOff == blockSize {
+		d.rightOff = 0
+		d.rightIdx = (d.rightIdx + 1) % len(d.blocks)
+
+		// If we wrapped over to the left, insert a new block and update indices
+		if d.rightIdx == d.leftIdx {
+			buffer := make([][]interface{}, len(d.blocks)+1)
+			copy(buffer[:d.rightIdx], d.blocks[:d.rightIdx])
+			buffer[d.rightIdx] = make([]interface{}, blockSize)
+			copy(buffer[d.rightIdx+1:], d.blocks[d.rightIdx:])
+			d.blocks = buffer
+			d.leftIdx++
+			d.left = d.blocks[d.leftIdx]
+		}
+		d.right = d.blocks[d.rightIdx]
+	}
+}
+
+// Pops out an element from the queue from the right. Note, no bounds checking are done.
+func (d *Deque) PopRight() (res interface{}) {
+	d.rightOff--
+	if d.rightOff < 0 {
+		d.rightOff = blockSize - 1
+		d.rightIdx = (d.rightIdx - 1 + len(d.blocks)) % len(d.blocks)
+		d.right = d.blocks[d.rightIdx]
+	}
+	res, d.right[d.rightOff] = d.right[d.rightOff], nil
+	return
+}
+
+// Returns the rightmost element from the deque. No bounds are checked.
+func (d *Deque) Right() interface{} {
+	if d.rightOff > 0 {
+		return d.right[d.rightOff-1]
+	} else {
+		return d.blocks[(d.rightIdx-1+len(d.blocks))%len(d.blocks)][blockSize-1]
+	}
+}
+
+// Pushes a new element into the queue from the left, expanding it if necessary.
+func (d *Deque) PushLeft(data interface{}) {
+	d.leftOff--
+	if d.leftOff < 0 {
+		d.leftOff = blockSize - 1
+		d.leftIdx = (d.leftIdx - 1 + len(d.blocks)) % len(d.blocks)
+
+		// If we wrapped over to the right, insert a new block and update indices
+		if d.leftIdx == d.rightIdx {
+			d.leftIdx++
+			buffer := make([][]interface{}, len(d.blocks)+1)
+			copy(buffer[:d.leftIdx], d.blocks[:d.leftIdx])
+			buffer[d.leftIdx] = make([]interface{}, blockSize)
+			copy(buffer[d.leftIdx+1:], d.blocks[d.leftIdx:])
+			d.blocks = buffer
+		}
+		d.left = d.blocks[d.leftIdx]
+	}
+	d.left[d.leftOff] = data
+}
+
+// Pops out an element from the queue from the left. Note, no bounds checking are done.
+func (d *Deque) PopLeft() (res interface{}) {
+	res, d.left[d.leftOff] = d.left[d.leftOff], nil
+	d.leftOff++
+	if d.leftOff == blockSize {
+		d.leftOff = 0
+		d.leftIdx = (d.leftIdx + 1) % len(d.blocks)
+		d.left = d.blocks[d.leftIdx]
+	}
+	return
+}
+
+// Returns the leftmost element from the deque. No bounds are checked.
+func (d *Deque) Left() interface{} {
+	return d.left[d.leftOff]
+}
+
+// Checks whether the queue is empty.
+func (d *Deque) Empty() bool {
+	return d.leftIdx == d.rightIdx && d.leftOff == d.rightOff
+}
+
+// Returns the number of elements in the queue.
+func (d *Deque) Size() int {
+	if d.rightIdx > d.leftIdx {
+		return (d.rightIdx-d.leftIdx)*blockSize - d.leftOff + d.rightOff
+	} else if d.rightIdx < d.leftIdx {
+		return (len(d.blocks)-d.leftIdx+d.rightIdx)*blockSize - d.leftOff + d.rightOff
+	} else {
+		return d.rightOff - d.leftOff
+	}
+}
+
+// Clears out the contents of the queue.
+func (d *Deque) Reset() {
+	*d = *New()
+}

vendor/vendor.json

@@ -266,6 +266,12 @@
 			"revision": "975b5c4c7c21c0e3d2764200bf2aa8e34657ae6e",
 			"revisionTime": "2017-04-30T22:20:11Z"
 		},
+		{
+			"checksumSHA1": "SDHLlmY5ED3dBFr1HfHvdcNaHao=",
+			"path": "github.com/karalabe/cookiejar/collections/deque",
+			"revision": "8dcd6a7f4951f6ff3ee9cbb919a06d8925822e57",
+			"revisionTime": "2015-07-24T13:16:13Z"
+		},
 		{
 			"checksumSHA1": "6XsjAARQFvlW6dS15al0ibTFPOQ=",
 			"path": "github.com/karalabe/hid",