core/rawdb: reinit from freezer in sequential order

common/prque/prque.go

@@ -1,5 +1,20 @@
+// 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).
+
 // This is a duplicated and slightly modified version of "gopkg.in/karalabe/cookiejar.v2/collections/prque".
 
+// Package prque implements a priority queue data structure supporting arbitrary
+// value types and int64 priorities.
+//
+// If you would like to use a min-priority queue, simply negate the priorities.
+//
+// Internally the queue is based on the standard heap package working on a
+// sortable version of the block based stack.
 package prque
 
 import (
@@ -11,8 +26,8 @@ type Prque struct {
 	cont *sstack
 }
 
-// Creates a new priority queue.
-func New(setIndex setIndexCallback) *Prque {
+// New creates a new priority queue.
+func New(setIndex SetIndexCallback) *Prque {
 	return &Prque{newSstack(setIndex)}
 }
 
@@ -21,6 +36,12 @@ func (p *Prque) Push(data interface{}, priority int64) {
 	heap.Push(p.cont, &item{data, priority})
 }
 
+// Peek returns the value with the greates priority but does not pop it off.
+func (p *Prque) Peek() (interface{}, int64) {
+	item := p.cont.blocks[0][0]
+	return item.value, item.priority
+}
+
 // Pops the value with the greates priority off the stack and returns it.
 // Currently no shrinking is done.
 func (p *Prque) Pop() (interface{}, int64) {

common/prque/sstack.go

@@ -1,3 +1,11 @@
+// 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).
+
 // This is a duplicated and slightly modified version of "gopkg.in/karalabe/cookiejar.v2/collections/prque".
 
 package prque
@@ -14,16 +22,16 @@ type item struct {
 	priority int64
 }
 
-// setIndexCallback is called when the element is moved to a new index.
-// Providing setIndexCallback is optional, it is needed only if the application needs
+// SetIndexCallback is called when the element is moved to a new index.
+// Providing SetIndexCallback is optional, it is needed only if the application needs
 // to delete elements other than the top one.
-type setIndexCallback func(a interface{}, i int)
+type SetIndexCallback func(data interface{}, index int)
 
 // Internal sortable stack data structure. Implements the Push and Pop ops for
 // the stack (heap) functionality and the Len, Less and Swap methods for the
 // sortability requirements of the heaps.
 type sstack struct {
-	setIndex setIndexCallback
+	setIndex SetIndexCallback
 	size     int
 	capacity int
 	offset   int
@@ -33,7 +41,7 @@ type sstack struct {
 }
 
 // Creates a new, empty stack.
-func newSstack(setIndex setIndexCallback) *sstack {
+func newSstack(setIndex SetIndexCallback) *sstack {
 	result := new(sstack)
 	result.setIndex = setIndex
 	result.active = make([]*item, blockSize)

core/rawdb/freezer_reinit.go

@@ -23,6 +23,7 @@ import (
 	"time"
 
 	"github.com/ethereum/go-ethereum/common"
+	"github.com/ethereum/go-ethereum/common/prque"
 	"github.com/ethereum/go-ethereum/core/types"
 	"github.com/ethereum/go-ethereum/ethdb"
 	"github.com/ethereum/go-ethereum/log"
@@ -75,6 +76,9 @@ func InitDatabaseFromFreezer(db ethdb.Database) error {
 	}
 	// Reassemble the blocks into a contiguous stream and push them out to disk
 	var (
+		queue = prque.New(nil)
+		next  = int64(0)
+
 		batch  = db.NewBatch()
 		start  = time.Now()
 		logged time.Time
@@ -85,20 +89,33 @@ func InitDatabaseFromFreezer(db ethdb.Database) error {
 		if block == nil {
 			return errors.New("broken ancient database")
 		}
-		// Inject hash<->number mapping and txlookup indexes
-		WriteHeaderNumber(batch, block.Hash(), block.NumberU64())
-		WriteTxLookupEntries(batch, block)
-
-		if batch.ValueSize() > ethdb.IdealBatchSize || i == frozen-1 {
-			if err := batch.Write(); err != nil {
-				return err
+		// Push the block into the import queue and process contiguous ranges
+		queue.Push(block, -int64(block.NumberU64()))
+		for !queue.Empty() {
+			// If the next available item is gapped, return
+			if _, priority := queue.Peek(); -priority != next {
+				break
+			}
+			// Next block available, pop it off and index it
+			block = queue.PopItem().(*types.Block)
+			next++
+
+			// Inject hash<->number mapping and txlookup indexes
+			WriteHeaderNumber(batch, block.Hash(), block.NumberU64())
+			WriteTxLookupEntries(batch, block)
+
+			// If enough data was accumulated in memory or we're at the last block, dump to disk
+			if batch.ValueSize() > ethdb.IdealBatchSize || uint64(next) == frozen {
+				if err := batch.Write(); err != nil {
+					return err
+				}
+				batch.Reset()
+			}
+			// If we've spent too much time already, notify the user of what we're doing
+			if time.Since(logged) > 8*time.Second {
+				log.Info("Initializing chain from ancient data", "number", block.Number(), "hash", block.Hash(), "total", frozen-1, "elapsed", common.PrettyDuration(time.Since(start)))
+				logged = time.Now()
 			}
-			batch.Reset()
-		}
-		// If we've spent too much time already, notify the user of what we're doing
-		if time.Since(logged) > 8*time.Second {
-			log.Info("Initializing chain from ancient data", "number", block.Number(), "hash", block.Hash(), "total", frozen-1, "elapsed", common.PrettyDuration(time.Since(start)))
-			logged = time.Now()
 		}
 	}
 	hash := ReadCanonicalHash(db, frozen-1)