From 0e5226cc6286f92d635d2b289a70c4a3b3628c5e Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?P=C3=A9ter=20Szil=C3=A1gyi?= Date: Mon, 27 May 2019 14:46:44 +0300 Subject: [PATCH] core/rawdb: reinit from freezer in sequential order --- common/prque/prque.go | 25 +++++++++++++++++++-- common/prque/sstack.go | 18 ++++++++++----- core/rawdb/freezer_reinit.go | 43 +++++++++++++++++++++++++----------- 3 files changed, 66 insertions(+), 20 deletions(-) diff --git a/common/prque/prque.go b/common/prque/prque.go index 9fd31a2e5d..3cc5a1adaf 100755 --- a/common/prque/prque.go +++ b/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) { diff --git a/common/prque/sstack.go b/common/prque/sstack.go index 4875dae99d..8518af54ff 100755 --- a/common/prque/sstack.go +++ b/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) diff --git a/core/rawdb/freezer_reinit.go b/core/rawdb/freezer_reinit.go index be6a8c839c..ea4dd33d1d 100644 --- a/core/rawdb/freezer_reinit.go +++ b/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)