From 89ad90d335edae7b9d70a59bc485a6200e8dda26 Mon Sep 17 00:00:00 2001 From: rjl493456442 Date: Fri, 26 Jul 2019 13:36:26 +0800 Subject: [PATCH] core, eth: skip transaction indexing during fast sync if required --- core/blockchain.go | 85 ++++++++++++++++++------------- core/rawdb/chain_iterator.go | 5 +- eth/downloader/downloader.go | 8 ++- eth/downloader/downloader_test.go | 2 +- eth/handler.go | 2 +- 5 files changed, 62 insertions(+), 40 deletions(-) diff --git a/core/blockchain.go b/core/blockchain.go index 3a9e4ab98f..e86f1242de 100644 --- a/core/blockchain.go +++ b/core/blockchain.go @@ -457,7 +457,7 @@ func (bc *BlockChain) SetHead(head uint64) error { // FastSyncCommitHead sets the current head block to the one defined by the hash // irrelevant what the chain contents were prior. -func (bc *BlockChain) FastSyncCommitHead(hash common.Hash) error { +func (bc *BlockChain) FastSyncCommitHead(hash common.Hash, from uint64, ancient uint64) error { // Make sure that both the block as well at its state trie exists block := bc.GetBlockByHash(hash) if block == nil { @@ -472,6 +472,22 @@ func (bc *BlockChain) FastSyncCommitHead(hash common.Hash) error { headBlockGauge.Update(int64(block.NumberU64())) bc.chainmu.Unlock() + // Write tx indices tail if it doesn't exist in database. + // The tx index tail can only be one of the following two options: + // * the start point of fast sync: + // in this case all blocks imported during fast sync have been indexed + // * ancient - limit: + // in this case the block before ancient-limit won't be indexed + if tail := rawdb.ReadTxIndexTail(bc.db); tail == nil { + if bc.txLookupLimit != 0 && ancient >= bc.txLookupLimit && ancient-bc.txLookupLimit > from { + rawdb.WriteTxIndexTail(bc.db, ancient-bc.txLookupLimit) + } else { + if from == 1 { + from = 0 + } + rawdb.WriteTxIndexTail(bc.db, from) + } + } log.Info("Committed new head block", "number", block.Number(), "hash", hash) return nil } @@ -1091,11 +1107,14 @@ func (bc *BlockChain) InsertReceiptChain(blockChain types.Blocks, receiptChain [ // Flush data into ancient database. size += rawdb.WriteAncientBlock(bc.db, block, receiptChain[i], bc.GetTd(block.Hash(), block.NumberU64())) - // We don't write tx lookup indices here because Geth can offer a CLI flag `txlookuplimt` - // with which user can choose to drop historical indices data and only keep latest indices. - // After the fast sync, we will reconstruct all missing indices even user requires to keep - // all historical indices. - + // Write tx indices if any condition is satisfied: + // * If user requires to reserve all tx indices(txlookuplimit=0) + // * If all ancient tx indices are required to be reserved(txlookuplimit is even higher than ancientlimit) + // * If block number is large enough to be regarded as a recent block + // It means blocks below the ancientLimit-txlookupLimit won't be indexed. + if bc.txLookupLimit == 0 || ancientLimit <= bc.txLookupLimit || block.NumberU64() >= ancientLimit-bc.txLookupLimit { + rawdb.WriteTxLookupEntries(batch, block) + } stats.processed++ } // Flush all tx-lookup index data. @@ -1169,11 +1188,8 @@ func (bc *BlockChain) InsertReceiptChain(blockChain types.Blocks, receiptChain [ // Write all the data out into the database rawdb.WriteBody(batch, block.Hash(), block.NumberU64(), block.Body()) rawdb.WriteReceipts(batch, block.Hash(), block.NumberU64(), receiptChain[i]) - - // We don't write tx lookup indices here because Geth can offer a CLI flag `txlookuplimt` - // with which user can choose to drop historical indices data and only keep latest indices. - // After the fast sync, we will reconstruct all missing indices even user requires to keep - // all historical indices. + // We always write tx indices for live block since we assume the indices are needed. + rawdb.WriteTxLookupEntries(batch, block) stats.processed++ if batch.ValueSize() >= ethdb.IdealBatchSize { @@ -2056,10 +2072,12 @@ func (bc *BlockChain) update() { // The user can adjust the txlookuplimit value for each launch, Geth will // automatically construct the missing indices and delete the extra indices. func (bc *BlockChain) maintainTxIndex() { - // initialiseIndices inits txlookup indices into the database. - // If there already exists some indices, this function will find - // the oldest block which has been indexed and start indexing from - // this point. + // initialiseIndices inits tx indices into the database if `TxIndexTail` + // is missing in database. + // Note for archive sync or fast sync, this code path will only be triggered + // after importing the first batch of blocks(e.g. 1024). But these block + // actually are already indexed. So a binary search will be performed to + // skip reindexing. initialiseIndices := func(head uint64, done chan struct{}) { defer func() { done <- struct{}{} }() @@ -2067,27 +2085,26 @@ func (bc *BlockChain) maintainTxIndex() { if bc.txLookupLimit != 0 && head > bc.txLookupLimit { from = head - bc.txLookupLimit } - // Find oldest indexed block via binary search when we don't - // have this flag in database. - start := time.Now() - oldest := rawdb.FindTxIndexTail(bc.db, from, to) - log.Debug("Find oldest indexed block", "oldest", oldest, "elapsed", common.PrettyDuration(time.Since(start))) + if tail := rawdb.FindTxIndexTail(bc.db, from, to); tail != nil { + // Special case here is genesis block doesn't contain any transaction + // that will be regarded as unindexed. + if *tail == from || (from == 0 && *tail == 1) { + rawdb.WriteTxIndexTail(bc.db, from) - // Re-construct missing tx indices. - if oldest == nil { - rawdb.IndexTxLookup(bc.db, from, to) // No block has been indexed. - } else { - rawdb.IndexTxLookup(bc.db, from, *oldest) - - // Drop all useless tx indices below the HEAD-limit. - if from > 0 { - oldest := rawdb.FindTxIndexTail(bc.db, 0, from) - if oldest != nil { - rawdb.RemoveTxsLookup(bc.db, *oldest, from) + // Drop all useless tx indices below the HEAD-limit. + if from > 0 { + rawdb.RemoveTxsLookup(bc.db, 0, from) } + return } } - log.Debug("Initialised transaction indices", "elapsed", common.PrettyDuration(time.Since(start))) + // Re-construct missing tx indices. + rawdb.IndexTxLookup(bc.db, from, to) + + // Drop all useless tx indices below the HEAD-limit. + if from > 0 { + rawdb.RemoveTxsLookup(bc.db, 0, from) + } } // indexBlocks reindex or unindex transaction indices depends // on user's requirement. @@ -2131,10 +2148,10 @@ func (bc *BlockChain) maintainTxIndex() { case head := <-headCh: if done == nil { done = make(chan struct{}) - if number := rawdb.ReadTxIndexTail(bc.db); number == nil { + if tail := rawdb.ReadTxIndexTail(bc.db); tail == nil { go initialiseIndices(head.Block.NumberU64(), done) } else { - go indexBlocks(*number, head.Block.NumberU64(), done) + go indexBlocks(*tail, head.Block.NumberU64(), done) } } case <-done: diff --git a/core/rawdb/chain_iterator.go b/core/rawdb/chain_iterator.go index 6755a37899..ba955ac92e 100644 --- a/core/rawdb/chain_iterator.go +++ b/core/rawdb/chain_iterator.go @@ -199,15 +199,16 @@ func IndexTxLookup(db ethdb.Database, from uint64, to uint64) { // writeIndices injects txlookup indices into the database. writeIndices := func(batch ethdb.Batch, block *types.Block) { WriteTxLookupEntries(batch, block) - if block.NumberU64()%1000000 == 0 { + if block.NumberU64()%100000 == 0 { WriteTxIndexTail(batch, block.NumberU64()) } } + start := time.Now() if err := iterateCanonicalChain(db, from, to, "txlookup", hashTxs, writeIndices, true, true); err != nil { log.Crit("Failed to iterate canonical chain", "err", err) } WriteTxIndexTail(db, from) - log.Info("Constructed transaction indices", "from", from, "to", to, "count", to-from) + log.Info("Constructed transaction indices", "from", from, "to", to, "count", to-from, "elapsed", common.PrettyDuration(time.Since(start))) } // RemoveTxsLookup removes txlookup indices of the specified range blocks. diff --git a/eth/downloader/downloader.go b/eth/downloader/downloader.go index edd0eb4d95..255f553db4 100644 --- a/eth/downloader/downloader.go +++ b/eth/downloader/downloader.go @@ -200,7 +200,7 @@ type BlockChain interface { CurrentFastBlock() *types.Block // FastSyncCommitHead directly commits the head block to a certain entity. - FastSyncCommitHead(common.Hash) error + FastSyncCommitHead(common.Hash, uint64, uint64) error // InsertChain inserts a batch of blocks into the local chain. InsertChain(types.Blocks) (int, error) @@ -1721,7 +1721,11 @@ func (d *Downloader) commitPivotBlock(result *fetchResult) error { if _, err := d.blockchain.InsertReceiptChain([]*types.Block{block}, []types.Receipts{result.Receipts}, d.ancientLimit); err != nil { return err } - if err := d.blockchain.FastSyncCommitHead(block.Hash()); err != nil { + // Use origin block number + 1 as the number of the first inserted block. + d.syncStatsLock.RLock() + from := d.syncStatsChainOrigin + 1 + d.syncStatsLock.RUnlock() + if err := d.blockchain.FastSyncCommitHead(block.Hash(), from, d.ancientLimit); err != nil { return err } atomic.StoreInt32(&d.committed, 1) diff --git a/eth/downloader/downloader_test.go b/eth/downloader/downloader_test.go index b23043b1c0..7fec1925c4 100644 --- a/eth/downloader/downloader_test.go +++ b/eth/downloader/downloader_test.go @@ -218,7 +218,7 @@ func (dl *downloadTester) CurrentFastBlock() *types.Block { } // FastSyncCommitHead manually sets the head block to a given hash. -func (dl *downloadTester) FastSyncCommitHead(hash common.Hash) error { +func (dl *downloadTester) FastSyncCommitHead(hash common.Hash, from uint64, ancient uint64) error { // For now only check that the state trie is correct if block := dl.GetBlockByHash(hash); block != nil { _, err := trie.NewSecure(block.Root(), trie.NewDatabase(dl.stateDb)) diff --git a/eth/handler.go b/eth/handler.go index 4ce2d1c82f..2e4c887bda 100644 --- a/eth/handler.go +++ b/eth/handler.go @@ -50,7 +50,7 @@ const ( // The number is referenced from the size of tx pool. txChanSize = 4096 - // minimim number of peers to broadcast new blocks to + // minBroadcastPeers is the minimal number of peers to broadcast new blocks to. minBroadcastPeers = 4 )