mirror of
https://github.com/ethereum/go-ethereum.git
synced 2026-08-20 10:52:25 +00:00
core: init tx lookup in background
This commit is contained in:
parent
a978adfd7c
commit
4f0cc2d7e2
4 changed files with 113 additions and 25 deletions
|
|
@ -230,7 +230,13 @@ func NewBlockChain(db ethdb.Database, cacheConfig *CacheConfig, chainConfig *par
|
||||||
}
|
}
|
||||||
// Initialize the chain with ancient data if it isn't empty.
|
// Initialize the chain with ancient data if it isn't empty.
|
||||||
if bc.empty() {
|
if bc.empty() {
|
||||||
rawdb.InitDatabaseFromFreezer(bc.db)
|
rawdb.InitBlockIndexFromFreezer(bc.db)
|
||||||
|
rawdb.WriteAncientTxLookupProgress(bc.db, 0) // Explicitly mark the missing of txlookup.
|
||||||
|
}
|
||||||
|
// Re-initialise all ancient txlookup indexes in the background.
|
||||||
|
if number := rawdb.ReadAncientTxLookupProgress(bc.db); number != nil {
|
||||||
|
// Genesis block doesn't have transaction, just ignore it.
|
||||||
|
go rawdb.InitTxsLookupFromFreezer(bc.db, *number+1)
|
||||||
}
|
}
|
||||||
if err := bc.loadLastState(); err != nil {
|
if err := bc.loadLastState(); err != nil {
|
||||||
return nil, err
|
return nil, err
|
||||||
|
|
|
||||||
|
|
@ -171,6 +171,32 @@ func WriteFastTrieProgress(db ethdb.KeyValueWriter, count uint64) {
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// ReadAncientTxLookupProgress retrieves the number of ancient blocks which
|
||||||
|
// txlookup has been inserted to allow reporting correct numbers across restarts.
|
||||||
|
func ReadAncientTxLookupProgress(db ethdb.KeyValueReader) *uint64 {
|
||||||
|
data, _ := db.Get(ancientTxLookupProgressKey)
|
||||||
|
if len(data) != 8 {
|
||||||
|
return nil
|
||||||
|
}
|
||||||
|
number := binary.BigEndian.Uint64(data)
|
||||||
|
return &number
|
||||||
|
}
|
||||||
|
|
||||||
|
// WriteAncientTxLookupProgress stores the ancient txlookup process counter to support
|
||||||
|
// retrieving it across restarts.
|
||||||
|
func WriteAncientTxLookupProgress(db ethdb.KeyValueWriter, number uint64) {
|
||||||
|
if err := db.Put(ancientTxLookupProgressKey, encodeBlockNumber(number)); err != nil {
|
||||||
|
log.Crit("Failed to store head number of txlookup", "err", err)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
// DeleteAncientTxLookupProgress deletes the ancient txlookup progress.
|
||||||
|
func DeleteAncientTxLookupProgress(db ethdb.KeyValueWriter) {
|
||||||
|
if err := db.Delete(ancientTxLookupProgressKey); err != nil {
|
||||||
|
log.Crit("Failed to delete ancient txlookup progress entry", "err", err)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
// ReadHeaderRLP retrieves a block header in its raw RLP database encoding.
|
// ReadHeaderRLP retrieves a block header in its raw RLP database encoding.
|
||||||
func ReadHeaderRLP(db ethdb.Reader, hash common.Hash, number uint64) rlp.RawValue {
|
func ReadHeaderRLP(db ethdb.Reader, hash common.Hash, number uint64) rlp.RawValue {
|
||||||
data, _ := db.Ancient(freezerHeaderTable, number)
|
data, _ := db.Ancient(freezerHeaderTable, number)
|
||||||
|
|
|
||||||
|
|
@ -29,21 +29,34 @@ import (
|
||||||
"github.com/ethereum/go-ethereum/log"
|
"github.com/ethereum/go-ethereum/log"
|
||||||
)
|
)
|
||||||
|
|
||||||
// InitDatabaseFromFreezer reinitializes an empty database from a previous batch
|
type (
|
||||||
// of frozen ancient blocks. The method iterates over all the frozen blocks and
|
initPrepare func(*types.Block) // The callback for customized prepare operation.
|
||||||
// injects into the database the block hash->number mappings and the transaction
|
initAction func(ethdb.Batch, *types.Block) // The callback for customized initialisation action.
|
||||||
// lookup entries.
|
)
|
||||||
func InitDatabaseFromFreezer(db ethdb.Database) error {
|
|
||||||
|
// iterateAncient iterates the specified range blocks from ancient database
|
||||||
|
// and then apply initialisation action.
|
||||||
|
func iterateAncient(db ethdb.Database, from uint64, typ string, prepare initPrepare, action initAction) error {
|
||||||
|
// Short circuit if the init action is nil.
|
||||||
|
if action == nil {
|
||||||
|
return nil
|
||||||
|
}
|
||||||
// If we can't access the freezer or it's empty, abort
|
// If we can't access the freezer or it's empty, abort
|
||||||
frozen, err := db.Ancients()
|
frozen, err := db.Ancients()
|
||||||
if err != nil || frozen == 0 {
|
if err != nil || frozen == 0 {
|
||||||
return err
|
return err
|
||||||
}
|
}
|
||||||
// Blocks previously frozen, iterate over- and hash them concurrently
|
// Spawn multi-routines, iterate over the specified blocks and invoke prepare
|
||||||
|
// callback concurrently.
|
||||||
var (
|
var (
|
||||||
number = ^uint64(0) // -1
|
number uint64
|
||||||
results = make(chan *types.Block, 4*runtime.NumCPU())
|
results = make(chan *types.Block, 4*runtime.NumCPU())
|
||||||
)
|
)
|
||||||
|
if from == 0 {
|
||||||
|
number = ^uint64(0) // -1
|
||||||
|
} else {
|
||||||
|
number = from - 1
|
||||||
|
}
|
||||||
abort := make(chan struct{})
|
abort := make(chan struct{})
|
||||||
defer close(abort)
|
defer close(abort)
|
||||||
|
|
||||||
|
|
@ -56,14 +69,10 @@ func InitDatabaseFromFreezer(db ethdb.Database) error {
|
||||||
return
|
return
|
||||||
}
|
}
|
||||||
// Retrieve the block from the freezer (no need for the hash, we pull by
|
// 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
|
// number from the freezer).
|
||||||
// the individual transaction hashes for storing into the database.
|
|
||||||
block := ReadBlock(db, common.Hash{}, n)
|
block := ReadBlock(db, common.Hash{}, n)
|
||||||
if block != nil {
|
if prepare != nil && block != nil {
|
||||||
block.Hash()
|
prepare(block)
|
||||||
for _, tx := range block.Transactions() {
|
|
||||||
tx.Hash()
|
|
||||||
}
|
|
||||||
}
|
}
|
||||||
// Feed the block to the aggregator, or abort on interrupt
|
// Feed the block to the aggregator, or abort on interrupt
|
||||||
select {
|
select {
|
||||||
|
|
@ -74,20 +83,20 @@ func InitDatabaseFromFreezer(db ethdb.Database) error {
|
||||||
}
|
}
|
||||||
}()
|
}()
|
||||||
}
|
}
|
||||||
// Reassemble the blocks into a contiguous stream and push them out to disk
|
// Reassemble the blocks into a contiguous stream and apply the action callback.
|
||||||
var (
|
var (
|
||||||
queue = prque.New(nil)
|
queue = prque.New(nil)
|
||||||
next = int64(0)
|
next = int64(from)
|
||||||
|
|
||||||
batch = db.NewBatch()
|
batch = db.NewBatch()
|
||||||
start = time.Now()
|
start = time.Now()
|
||||||
logged time.Time
|
logged time.Time
|
||||||
)
|
)
|
||||||
for i := uint64(0); i < frozen; i++ {
|
for i := from; i < frozen; i++ {
|
||||||
// Retrieve the next result and bail if it's nil
|
// Retrieve the next result and bail if it's nil
|
||||||
block := <-results
|
block := <-results
|
||||||
if block == nil {
|
if block == nil {
|
||||||
return errors.New("broken ancient database")
|
return errors.New("broken database")
|
||||||
}
|
}
|
||||||
// Push the block into the import queue and process contiguous ranges
|
// Push the block into the import queue and process contiguous ranges
|
||||||
queue.Push(block, -int64(block.NumberU64()))
|
queue.Push(block, -int64(block.NumberU64()))
|
||||||
|
|
@ -100,9 +109,8 @@ func InitDatabaseFromFreezer(db ethdb.Database) error {
|
||||||
block = queue.PopItem().(*types.Block)
|
block = queue.PopItem().(*types.Block)
|
||||||
next++
|
next++
|
||||||
|
|
||||||
// Inject hash<->number mapping and txlookup indexes
|
// Invoke action to inject specified data into key-value database.
|
||||||
WriteHeaderNumber(batch, block.Hash(), block.NumberU64())
|
action(batch, block)
|
||||||
WriteTxLookupEntries(batch, block)
|
|
||||||
|
|
||||||
// If enough data was accumulated in memory or we're at the last block, dump to disk
|
// 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 batch.ValueSize() > ethdb.IdealBatchSize || uint64(next) == frozen {
|
||||||
|
|
@ -113,15 +121,60 @@ func InitDatabaseFromFreezer(db ethdb.Database) error {
|
||||||
}
|
}
|
||||||
// If we've spent too much time already, notify the user of what we're doing
|
// If we've spent too much time already, notify the user of what we're doing
|
||||||
if time.Since(logged) > 8*time.Second {
|
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)))
|
log.Info("Initializing chain from ancient data", "type", typ, "number", block.Number(), "hash", block.Hash(), "total", uint64(next)-from, "elapsed", common.PrettyDuration(time.Since(start)))
|
||||||
logged = time.Now()
|
logged = time.Now()
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
log.Info("Initialized chain from ancient data", "type", typ, "number", frozen-from, "elapsed", common.PrettyDuration(time.Since(start)))
|
||||||
|
return nil
|
||||||
|
}
|
||||||
|
|
||||||
|
// InitBlockIndexFromFreezer 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 InitBlockIndexFromFreezer(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
|
||||||
|
}
|
||||||
|
// hashBlock calculates block hash in advance using the multi-routine's concurrent
|
||||||
|
// computing power.
|
||||||
|
hashBlock := func(block *types.Block) { block.Hash() }
|
||||||
|
|
||||||
|
// writeIndex injects hash <-> number mapping into the database.
|
||||||
|
writeIndex := func(batch ethdb.Batch, block *types.Block) { WriteHeaderNumber(batch, block.Hash(), block.NumberU64()) }
|
||||||
|
|
||||||
|
if err := iterateAncient(db, 0, "blocks", hashBlock, writeIndex); err != nil {
|
||||||
|
return err
|
||||||
|
}
|
||||||
hash := ReadCanonicalHash(db, frozen-1)
|
hash := ReadCanonicalHash(db, frozen-1)
|
||||||
WriteHeadHeaderHash(db, hash)
|
WriteHeadHeaderHash(db, hash)
|
||||||
WriteHeadFastBlockHash(db, hash)
|
WriteHeadFastBlockHash(db, hash)
|
||||||
|
return nil
|
||||||
log.Info("Initialized chain from ancient data", "number", frozen-1, "hash", hash, "elapsed", common.PrettyDuration(time.Since(start)))
|
}
|
||||||
|
|
||||||
|
// InitTxsLookupFromFreezer initializes txlookup indexes in the database.
|
||||||
|
func InitTxsLookupFromFreezer(db ethdb.Database, from uint64) error {
|
||||||
|
// hashTxs calculates transaction hash in advance using the multi-routine's
|
||||||
|
// concurrent computing power.
|
||||||
|
hashTxs := func(block *types.Block) {
|
||||||
|
for _, tx := range block.Transactions() {
|
||||||
|
tx.Hash()
|
||||||
|
}
|
||||||
|
}
|
||||||
|
// writeIndex injects txlookup indexes into the database.
|
||||||
|
writeIndex := func(batch ethdb.Batch, block *types.Block) {
|
||||||
|
WriteTxLookupEntries(batch, block)
|
||||||
|
if block.NumberU64()%10000 == 0 {
|
||||||
|
WriteAncientTxLookupProgress(batch, block.NumberU64())
|
||||||
|
}
|
||||||
|
}
|
||||||
|
if err := iterateAncient(db, from, "txlookup", hashTxs, writeIndex); err != nil {
|
||||||
|
return err
|
||||||
|
}
|
||||||
|
DeleteAncientTxLookupProgress(db) // Mark all txlookup indexes of ancient blocks have been inserted.
|
||||||
return nil
|
return nil
|
||||||
}
|
}
|
||||||
|
|
@ -41,6 +41,9 @@ var (
|
||||||
// fastTrieProgressKey tracks the number of trie entries imported during fast sync.
|
// fastTrieProgressKey tracks the number of trie entries imported during fast sync.
|
||||||
fastTrieProgressKey = []byte("TrieSync")
|
fastTrieProgressKey = []byte("TrieSync")
|
||||||
|
|
||||||
|
// ancientTxLookupProgressKey tracks the progress of ancient txs lookup insertion.
|
||||||
|
ancientTxLookupProgressKey = []byte("AncientTxsLookup")
|
||||||
|
|
||||||
// Data item prefixes (use single byte to avoid mixing data types, avoid `i`, used for indexes).
|
// Data item prefixes (use single byte to avoid mixing data types, avoid `i`, used for indexes).
|
||||||
headerPrefix = []byte("h") // headerPrefix + num (uint64 big endian) + hash -> header
|
headerPrefix = []byte("h") // headerPrefix + num (uint64 big endian) + hash -> header
|
||||||
headerTDSuffix = []byte("t") // headerPrefix + num (uint64 big endian) + hash + headerTDSuffix -> td
|
headerTDSuffix = []byte("t") // headerPrefix + num (uint64 big endian) + hash + headerTDSuffix -> td
|
||||||
|
|
|
||||||
Loading…
Reference in a new issue