core/rawdb: concurrent database reinit from freezer dump

core/blockchain.go

@@ -220,53 +220,16 @@ func NewBlockChain(db ethdb.Database, cacheConfig *CacheConfig, chainConfig *par
 	}
 	// Initialize the chain with ancient data if it isn't empty.
 	if bc.empty() {
-		if frozen, err := bc.db.Ancients(); err == nil && frozen > 0 {
-			var (
-				batch  = bc.db.NewBatch()
-				start  = time.Now()
-				logged time.Time
-			)
-			for i := uint64(0); i < frozen; i++ {
-				// Retrieve the canonical hash and block for tx index
-				hash := rawdb.ReadCanonicalHash(bc.db, i)
-				if hash == (common.Hash{}) {
-					return nil, errors.New("broken ancient database")
-				}
-				block := rawdb.ReadBlock(bc.db, hash, i)
-				if block == nil {
-					return nil, errors.New("broken ancient database")
-				}
-				// Inject hash<->number mapping and txlookup indexes
-				rawdb.WriteHeaderNumber(batch, hash, i)
-				rawdb.WriteTxLookupEntries(batch, block)
-
-				if batch.ValueSize() > ethdb.IdealBatchSize || i == frozen-1 {
-					if err := batch.Write(); err != nil {
-						return nil, 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", i, "hash", hash, "total", frozen-1, "elapsed", common.PrettyDuration(time.Since(start)))
-					logged = time.Now()
-				}
-			}
-			hash := rawdb.ReadCanonicalHash(bc.db, frozen-1)
-			rawdb.WriteHeadHeaderHash(bc.db, hash)
-			rawdb.WriteHeadFastBlockHash(bc.db, hash)
-
-			// The first thing the node will do is reconstruct the verification data for
-			// the head block (ethash cache or clique voting snapshot). Might as well do
-			// it in advance.
-			bc.engine.VerifyHeader(bc, rawdb.ReadHeader(bc.db, hash, frozen-1), true)
-
-			log.Info("Initialized chain from ancient data", "number", frozen-1, "hash", hash, "elapsed", common.PrettyDuration(time.Since(start)))
-		}
+		rawdb.InitDatabaseFromFreezer(bc.db)
 	}
 	if err := bc.loadLastState(); err != nil {
 		return nil, err
 	}
+	// The first thing the node will do is reconstruct the verification data for
+	// the head block (ethash cache or clique voting snapshot). Might as well do
+	// it in advance.
+	bc.engine.VerifyHeader(bc, bc.CurrentHeader(), true)
+
 	if frozen, err := bc.db.Ancients(); err == nil && frozen > 0 {
 		var (
 			needRewind bool

core/rawdb/accessors_indexes.go

@@ -55,8 +55,9 @@ func ReadTxLookupEntry(db ethdb.Reader, hash common.Hash) *uint64 {
 // WriteTxLookupEntries stores a positional metadata for every transaction from
 // a block, enabling hash based transaction and receipt lookups.
 func WriteTxLookupEntries(db ethdb.KeyValueWriter, block *types.Block) {
+	number := block.Number().Bytes()
 	for _, tx := range block.Transactions() {
-		if err := db.Put(txLookupKey(tx.Hash()), block.Number().Bytes()); err != nil {
+		if err := db.Put(txLookupKey(tx.Hash()), number); err != nil {
 			log.Crit("Failed to store transaction lookup entry", "err", err)
 		}
 	}

core/rawdb/freezer_reinit.go

@@ -0,0 +1,110 @@
+// Copyright 2019 The go-ethereum Authors
+// This file is part of the go-ethereum library.
+//
+// The go-ethereum library is free software: you can redistribute it and/or modify
+// it under the terms of the GNU Lesser General Public License as published by
+// the Free Software Foundation, either version 3 of the License, or
+// (at your option) any later version.
+//
+// The go-ethereum library is distributed in the hope that it will be useful,
+// but WITHOUT ANY WARRANTY; without even the implied warranty of
+// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+// GNU Lesser General Public License for more details.
+//
+// You should have received a copy of the GNU Lesser General Public License
+// along with the go-ethereum library. If not, see <http://www.gnu.org/licenses/>.
+
+package rawdb
+
+import (
+	"errors"
+	"runtime"
+	"sync/atomic"
+	"time"
+
+	"github.com/ethereum/go-ethereum/common"
+	"github.com/ethereum/go-ethereum/core/types"
+	"github.com/ethereum/go-ethereum/ethdb"
+	"github.com/ethereum/go-ethereum/log"
+)
+
+// InitDatabaseFromFreezer reinitializes an empty database from a previous batch
+// of frozen ancient blocks. The method iterates over all the frozen blocks and
+// injects into the database the block hash->number mappings and the transaction
+// lookup entries.
+func InitDatabaseFromFreezer(db ethdb.Database) error {
+	// If we can't access the freezer or it's empty, abort
+	frozen, err := db.Ancients()
+	if err != nil || frozen == 0 {
+		return err
+	}
+	// Blocks previously frozen, iterate over- and hash them concurrently
+	var (
+		number  = ^uint64(0) // -1
+		results = make(chan *types.Block, 4*runtime.NumCPU())
+	)
+	abort := make(chan struct{})
+	defer close(abort)
+
+	for i := 0; i < runtime.NumCPU(); i++ {
+		go func() {
+			for {
+				// Fetch the next task number, terminating if everything's done
+				n := atomic.AddUint64(&number, 1)
+				if n >= frozen {
+					return
+				}
+				// Retrieve the block from the freezer (no need for the hash, we pull by
+				// number from the freezer). If successful, pre-cache the block hash and
+				// the individual transaction hashes for storing into the database.
+				block := ReadBlock(db, common.Hash{}, n)
+				if block != nil {
+					block.Hash()
+					for _, tx := range block.Transactions() {
+						tx.Hash()
+					}
+				}
+				// Feed the block to the aggregator, or abort on interrupt
+				select {
+				case results <- block:
+				case <-abort:
+					return
+				}
+			}
+		}()
+	}
+	// Reassemble the blocks into a contiguous stream and push them out to disk
+	var (
+		batch  = db.NewBatch()
+		start  = time.Now()
+		logged time.Time
+	)
+	for i := uint64(0); i < frozen; i++ {
+		// Retrieve the next result and bail if it's nil
+		block := <-results
+		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
+			}
+			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)
+	WriteHeadHeaderHash(db, hash)
+	WriteHeadFastBlockHash(db, hash)
+
+	log.Info("Initialized chain from ancient data", "number", frozen-1, "hash", hash, "elapsed", common.PrettyDuration(time.Since(start)))
+	return nil
+}