diff --git a/cmd/utils/flags.go b/cmd/utils/flags.go index 1709a5aed8..35273132bb 100644 --- a/cmd/utils/flags.go +++ b/cmd/utils/flags.go @@ -217,7 +217,7 @@ var ( } TxLookupLimitFlag = cli.Int64Flag{ Name: "txlookuplimit", - Usage: "The maximum number of blocks from head whose tx indices are reserved(0 means reserve all indices)", + Usage: "Number of recent blocks to index transactions-by-hash in (default = index all blocks)", Value: 0, } LightKDFFlag = cli.BoolFlag{ @@ -1427,7 +1427,7 @@ func SetEthConfig(ctx *cli.Context, stack *node.Node, cfg *eth.Config) { // todo(rjl493456442) make it available for les server // Ancient tx indices pruning is not available for les server now // since light client relies on the server for transaction status query. - CheckExclusive(ctx, SyncModeFlag, "light", TxLookupLimitFlag) + CheckExclusive(ctx, LightLegacyServFlag, LightServeFlag, TxLookupLimitFlag) var ks *keystore.KeyStore if keystores := stack.AccountManager().Backends(keystore.KeyStoreType); len(keystores) > 0 { ks = keystores[0].(*keystore.KeyStore) diff --git a/core/blockchain.go b/core/blockchain.go index 246aee9cd1..8bfdaf98b6 100644 --- a/core/blockchain.go +++ b/core/blockchain.go @@ -139,10 +139,10 @@ type BlockChain struct { gcproc time.Duration // Accumulates canonical block processing for trie dumping // txLookupLimit is the maximum number of blocks from head whose tx indices - // are reserved. - // * nil do nothing - // * 0 means no limit (and regenerate any missing) - // * N means N blocks limit [HEAD-N, HEAD] + // are reserved: + // * 0: means no limit and regenerate any missing indexes + // * N: means N block limit [HEAD-N, HEAD] and delete extra indexes + // * nil: disable tx reindexer/deleter, but still index new blocks txLookupLimit uint64 hc *HeaderChain @@ -236,14 +236,15 @@ func NewBlockChain(db ethdb.Database, cacheConfig *CacheConfig, chainConfig *par return nil, ErrNoGenesis } // Initialize the chain with ancient data if it isn't empty. - var ancients uint64 + var txIndexBlock uint64 + if bc.empty() { - rawdb.InitBlockIndexFromFreezer(bc.db) + rawdb.InitDatabaseFromFreezer(bc.db) // If ancient database is not empty, reconstruct all missing // indices in the background. frozen, _ := bc.db.Ancients() if frozen > 0 { - ancients = frozen + txIndexBlock = frozen } } if err := bc.loadLastState(); err != nil { @@ -304,7 +305,7 @@ func NewBlockChain(db ethdb.Database, cacheConfig *CacheConfig, chainConfig *par go bc.update() if txLookupLimit != nil { bc.txLookupLimit = *txLookupLimit - go bc.maintainTxIndex(ancients) + go bc.maintainTxIndex(txIndexBlock) } return bc, nil } @@ -1187,8 +1188,7 @@ 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 always write tx indices for live block since we assume the indices are needed. - rawdb.WriteTxLookupEntries(batch, block) + rawdb.WriteTxLookupEntries(batch, block) // Always write tx indices for live blocks, we assume they are needed stats.processed++ if batch.ValueSize() >= ethdb.IdealBatchSize { @@ -1208,7 +1208,7 @@ func (bc *BlockChain) InsertReceiptChain(blockChain types.Blocks, receiptChain [ updateHead(blockChain[len(blockChain)-1]) return 0, nil } - // Write downloaded chain data and corresponding receipt chain data. + // Write downloaded chain data and corresponding receipt chain data if len(ancientBlocks) > 0 { if n, err := writeAncient(ancientBlocks, ancientReceipts); err != nil { if err == errInsertionInterrupted { @@ -1217,10 +1217,10 @@ func (bc *BlockChain) InsertReceiptChain(blockChain types.Blocks, receiptChain [ return n, err } } - // Write tx indices tail before write any live block. - if len(ancientBlocks) > 0 && len(liveBlocks) > 0 { + // Write the tx index tail (block number from where we index) before write any live blocks + if len(liveBlocks) > 0 && liveBlocks[0].NumberU64() == ancientLimit+1 { // The tx index tail can only be one of the following two options: - // * 0: all ancient blocks have been indexed. + // * 0: all ancient blocks have been indexed // * ancient-limit: the indices of blocks before ancient-limit are ignored if tail := rawdb.ReadTxIndexTail(bc.db); tail == nil { if bc.txLookupLimit == 0 || ancientLimit <= bc.txLookupLimit { @@ -1253,15 +1253,14 @@ func (bc *BlockChain) InsertReceiptChain(blockChain types.Blocks, receiptChain [ return 0, nil } -// SetTxLookupLimit is responsible for updating the txlookup limit -// to the original one stored in db if the new old mismatch with the old -// one. +// SetTxLookupLimit is responsible for updating the txlookup limit to the +// original one stored in db if the new mismatches with the old one. func (bc *BlockChain) SetTxLookupLimit(limit uint64) { bc.txLookupLimit = limit } // TxLookupLimit retrieves the txlookup limit used by blockchain to prune -// stale tx indices. +// stale transaction indices. func (bc *BlockChain) TxLookupLimit() uint64 { return bc.txLookupLimit } @@ -2098,64 +2097,60 @@ func (bc *BlockChain) update() { // sync, Geth will automatically construct the missing indices and delete // the extra indices. func (bc *BlockChain) maintainTxIndex(ancients uint64) { - // indexBlocks reindex or unindex transaction indices depends - // on user's requirement. - indexBlocks := func(tail *uint64, head uint64, done chan struct{}) { - defer func() { done <- struct{}{} }() - - if tail == nil { - // This is a special case that user upgrades Geth to a new version - // which supports tx indices pruning feature but the tx index tail - // is missing. So that we can assume all blocks in db are indexed. - if bc.txLookupLimit == 0 || head < bc.txLookupLimit { - // Nothing to delete, write the tail and return. - rawdb.WriteTxIndexTail(bc.db, 0) - } else { - // Prune all stale tx indices and record the tx index tail. - rawdb.RemoveTxsLookup(bc.db, 0, head-bc.txLookupLimit+1) - } - return - } - // All indices should be reserved. - if bc.txLookupLimit == 0 || head < bc.txLookupLimit { - if *tail == 0 { - // Short circuit if nothing to delete. - } else { - // Reindex all indices if necessary - rawdb.IndexTxLookup(bc.db, 0, *tail) - } - return - } - if head-bc.txLookupLimit+1 < *tail { - // Reindex a part of missing indices and rewind oldest indexed - // point to HEAD-limit - rawdb.IndexTxLookup(bc.db, head-bc.txLookupLimit+1, *tail) - } else { - // Unindex a part of stale indices and forward oldest indexed - // point to HEAD-limit - rawdb.RemoveTxsLookup(bc.db, *tail, head-bc.txLookupLimit+1) - } - } - // Special case here: user might init Geth with an external ancient database. - // If so, we should reindex all necessary indices before start processing any - // indices pruning requests. + // Before starting the actual maintenance, we need to handle a special case, + // where user might init Geth with an external ancient database. If so, we + // need to reindex all necessary transactions before starting to process any + // pruning requests. if ancients > 0 { var from = uint64(0) if bc.txLookupLimit != 0 && ancients > bc.txLookupLimit { from = ancients - bc.txLookupLimit } - rawdb.IndexTxLookup(bc.db, from, ancients) + rawdb.IndexTransactions(bc.db, from, ancients) } + // indexBlocks reindexes or unindexes transactions depending on user configuration + indexBlocks := func(tail *uint64, head uint64, done chan struct{}) { + defer func() { done <- struct{}{} }() + + // If the user just upgraded Geth to a new version which supports transaction + // index pruning, write the new tail and remove anything older. + if tail == nil { + if bc.txLookupLimit == 0 || head < bc.txLookupLimit { + // Nothing to delete, write the tail and return + rawdb.WriteTxIndexTail(bc.db, 0) + } else { + // Prune all stale tx indices and record the tx index tail + rawdb.UnindexTransactions(bc.db, 0, head-bc.txLookupLimit+1) + } + return + } + // If a previous indexing existed, make sure that we fill in any missing entries + if bc.txLookupLimit == 0 || head < bc.txLookupLimit { + if *tail > 0 { + rawdb.IndexTransactions(bc.db, 0, *tail) + } + return + } + // Update the transaction index to the new chain state + if head-bc.txLookupLimit+1 < *tail { + // Reindex a part of missing indices and rewind index tail to HEAD-limit + rawdb.IndexTransactions(bc.db, head-bc.txLookupLimit+1, *tail) + } else { + // Unindex a part of stale indices and forward index tail to HEAD-limit + rawdb.UnindexTransactions(bc.db, *tail, head-bc.txLookupLimit+1) + } + } + // Any reindexing done, start listening to chain events and moving the index window var ( - done chan struct{} // Non-nil if background unindexing or reindexing routine is active. - headCh = make(chan ChainHeadEvent) + done chan struct{} // Non-nil if background unindexing or reindexing routine is active. + headCh = make(chan ChainHeadEvent, 1) // Buffered to avoid locking up the event feed ) sub := bc.SubscribeChainHeadEvent(headCh) - defer func() { - if sub != nil { - sub.Unsubscribe() - } - }() + if sub == nil { + return + } + defer sub.Unsubscribe() + for { select { case head := <-headCh: diff --git a/core/rawdb/accessors_chain.go b/core/rawdb/accessors_chain.go index 1963728d28..3564605ee5 100644 --- a/core/rawdb/accessors_chain.go +++ b/core/rawdb/accessors_chain.go @@ -187,7 +187,7 @@ func ReadTxIndexTail(db ethdb.KeyValueReader) *uint64 { // into database. func WriteTxIndexTail(db ethdb.KeyValueWriter, number uint64) { if err := db.Put(txIndexTailKey, encodeBlockNumber(number)); err != nil { - log.Crit("Failed to store the number of oldest indexed block", "err", err) + log.Crit("Failed to store the transaction index tail", "err", err) } } @@ -204,7 +204,7 @@ func ReadFastTxLookupLimit(db ethdb.KeyValueReader) *uint64 { // WriteFastTxLookupLimit stores the txlookup limit used in fast sync into database. func WriteFastTxLookupLimit(db ethdb.KeyValueWriter, number uint64) { if err := db.Put(fastTxLookupLimitKey, encodeBlockNumber(number)); err != nil { - log.Crit("Failed to store txlookup limit for fast sync", "err", err) + log.Crit("Failed to store transaction lookup limit for fast sync", "err", err) } } diff --git a/core/rawdb/accessors_indexes.go b/core/rawdb/accessors_indexes.go index 4f99436bf3..6ae97a951b 100644 --- a/core/rawdb/accessors_indexes.go +++ b/core/rawdb/accessors_indexes.go @@ -63,8 +63,14 @@ func WriteTxLookupEntries(db ethdb.KeyValueWriter, block *types.Block) { } } -// DeleteTxLookupEntries removes all transaction lookup indices contained in -// given block. +// DeleteTxLookupEntry removes all transaction data associated with a hash. +func DeleteTxLookupEntry(db ethdb.KeyValueWriter, hash common.Hash) { + if err := db.Delete(txLookupKey(hash)); err != nil { + log.Crit("Failed to delete transaction lookup entry", "err", err) + } +} + +// DeleteTxLookupEntries removes all transaction lookups for a given block. func DeleteTxLookupEntries(db ethdb.KeyValueWriter, block *types.Block) { for _, tx := range block.Transactions() { if err := db.Delete(txLookupKey(tx.Hash())); err != nil { @@ -73,11 +79,6 @@ func DeleteTxLookupEntries(db ethdb.KeyValueWriter, block *types.Block) { } } -// DeleteTxLookupEntry removes all transaction data associated with a hash. -func DeleteTxLookupEntry(db ethdb.KeyValueWriter, hash common.Hash) { - db.Delete(txLookupKey(hash)) -} - // ReadTransaction retrieves a specific transaction from the database, along with // its added positional metadata. func ReadTransaction(db ethdb.Reader, hash common.Hash) (*types.Transaction, common.Hash, uint64, uint64) { diff --git a/core/rawdb/chain_iterator.go b/core/rawdb/chain_iterator.go index cab1ea6f9c..bb238c8d78 100644 --- a/core/rawdb/chain_iterator.go +++ b/core/rawdb/chain_iterator.go @@ -31,19 +31,18 @@ import ( ) type ( - prepareCallback func(*types.Block) // The callback for customized prepare operation. - actionCallback func(ethdb.Batch, *types.Block) // The callback for customized action. + prepareCallback func(*types.Block) // Callback for custom concurrent pre-computations on block data + actionCallback func(ethdb.Batch, *types.Block) // Callback for custom sequential operations on block data. ) -// iterateCanonicalChain iterates the specified range canonical chain and then -// apply the given action callback. -// Note both for forward and backward iteration, the range is [from, to). -func iterateCanonicalChain(db ethdb.Database, from uint64, to uint64, typ string, prepare prepareCallback, action actionCallback, reverse bool, report bool) error { - // Short circuit if the action is nil. +// iterateCanonicalChain iterates the specified range of canonical blocks and applies +// the given action callback. Note, both for forward and backward iteration, the range +// is [from, to). +func iterateCanonicalChain(db ethdb.Database, from uint64, to uint64, prepare prepareCallback, action actionCallback, reverse bool, progMsg, doneMsg string) error { + // Short circuit if the action is nil or if the range is invalid if action == nil { return nil } - // Short circuit if the iteration range is invalid. if from >= to { return nil } @@ -61,7 +60,11 @@ func iterateCanonicalChain(db ethdb.Database, from uint64, to uint64, typ string abort := make(chan struct{}) defer close(abort) - for i := 0; i < runtime.NumCPU(); i++ { + threads := to - from + if cpus := runtime.NumCPU(); threads > uint64(cpus) { + threads = uint64(cpus) + } + for i := 0; i < int(threads); i++ { go func() { for { // Fetch the next task number, terminating if everything's done @@ -90,7 +93,7 @@ func iterateCanonicalChain(db ethdb.Database, from uint64, to uint64, typ string } }() } - // Reassemble the blocks into a contiguous stream and apply the action callback. + // Reassemble the blocks into a contiguous stream and apply the action callback var ( next, first, last int64 queue = prque.New(nil) @@ -104,10 +107,6 @@ func iterateCanonicalChain(db ethdb.Database, from uint64, to uint64, typ string } else { next, first, last = int64(to-1), int64(to-1), int64(from-1) } - logFn := log.Debug - if report { - logFn = log.Info - } for i := from; i < to; i++ { // Retrieve the next result and bail if it's nil block := <-results @@ -145,19 +144,19 @@ func iterateCanonicalChain(db ethdb.Database, from uint64, to uint64, typ string } // If we've spent too much time already, notify the user of what we're doing if time.Since(logged) > 8*time.Second { - logFn("Iterating canonical chain", "type", typ, "reserve", reverse, "number", block.Number(), "hash", block.Hash(), "total", int64(math.Abs(float64(next-first))), "elapsed", common.PrettyDuration(time.Since(start))) + log.Info(progMsg, "blocks", int64(math.Abs(float64(next-first))), "total", to-from, "number", block.Number(), "hash", block.Hash(), "elapsed", common.PrettyDuration(time.Since(start))) logged = time.Now() } } } - logFn("Iterated canonical chain", "type", typ, "reverse", reverse, "total", to-from, "elapsed", common.PrettyDuration(time.Since(start))) + log.Info(doneMsg, "blocks", to-from, "elapsed", common.PrettyDuration(time.Since(start))) return nil } -// InitBlockIndexFromFreezer reinitializes an empty database from a previous batch +// 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. -func InitBlockIndexFromFreezer(db ethdb.Database) { +func InitDatabaseFromFreezer(db ethdb.Database) { // If we can't access the freezer or it's empty, abort frozen, err := db.Ancients() if err != nil || frozen == 0 { @@ -170,22 +169,20 @@ func InitBlockIndexFromFreezer(db ethdb.Database) { // writeIndex injects hash <-> number mapping into the database. writeIndex := func(batch ethdb.Batch, block *types.Block) { WriteHeaderNumber(batch, block.Hash(), block.NumberU64()) } - if err := iterateCanonicalChain(db, 0, frozen, "blocks", hashBlock, writeIndex, false, true); err != nil { - log.Crit("Failed to iterate canonical chain", "err", err) + if err := iterateCanonicalChain(db, 0, frozen, hashBlock, writeIndex, false, "Initializing database from freezer", "Initialized database from freezer"); err != nil { + log.Crit("Failed to init database from freezer", "err", err) } hash := ReadCanonicalHash(db, frozen-1) WriteHeadHeaderHash(db, hash) WriteHeadFastBlockHash(db, hash) - log.Info("Initialized chain from ancient data", "number", frozen-1, "hash", hash) } -// IndexTxLookup initializes txlookup indices of the specified range blocks into -// the database. +// IndexTransactions creates txlookup indices of the specified block range. // // This function iterates canonical chain in reverse order, it has one main advantage: // We can write tx index tail flag periodically even without the whole indexing // procedure is finished. So that we can resume indexing procedure next time quickly. -func IndexTxLookup(db ethdb.Database, from uint64, to uint64) { +func IndexTransactions(db ethdb.Database, from uint64, to uint64) { // hashTxs calculates transaction hash in advance using the multi-routine's // concurrent computing power. hashTxs := func(block *types.Block) { @@ -200,29 +197,26 @@ func IndexTxLookup(db ethdb.Database, from uint64, to uint64) { 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) + if err := iterateCanonicalChain(db, from, to, hashTxs, writeIndices, true, "Indexing transactions", "Indexed transactions"); err != nil { + log.Crit("Failed to index transactions", "err", err) } WriteTxIndexTail(db, 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. -func RemoveTxsLookup(db ethdb.Database, from uint64, to uint64) { +// UnindexTransactions removes txlookup indices of the specified block range. +func UnindexTransactions(db ethdb.Database, from uint64, to uint64) { // Write flag first and then unindex the transaction indices. Some indices // will be left in the database if crash happens but it's fine. WriteTxIndexTail(db, to) + // If only one block is unindexed, do it direclty if from+1 == to { - hash := ReadCanonicalHash(db, from) - DeleteTxLookupEntries(db, ReadBlock(db, hash, from)) - log.Debug("Removed transaction indices", "number", from, "hash", hash) - } else { - deleteIndices := func(batch ethdb.Batch, block *types.Block) { DeleteTxLookupEntries(batch, block) } - if err := iterateCanonicalChain(db, from, to, "txlookup", nil, deleteIndices, false, false); err != nil { - log.Crit("Failed to iterate canonical chain", "err", err) - } - log.Debug("Removed transaction indices", "from", from, "to", to, "count", to-from) + DeleteTxLookupEntries(db, ReadBlock(db, ReadCanonicalHash(db, from), from)) + return + } + // Otherwise spin up the concurrent iterator and unindexer + deleteIndices := func(batch ethdb.Batch, block *types.Block) { DeleteTxLookupEntries(batch, block) } + if err := iterateCanonicalChain(db, from, to, nil, deleteIndices, false, "Unindexing transactions", "Unindexed transactions"); err != nil { + log.Crit("Failed to unindex transactions", "err", err) } } diff --git a/core/rawdb/chain_iterator_test.go b/core/rawdb/chain_iterator_test.go index fc00fb1de7..d9505c62da 100644 --- a/core/rawdb/chain_iterator_test.go +++ b/core/rawdb/chain_iterator_test.go @@ -56,7 +56,7 @@ func TestChainIterator(t *testing.T) { } for i, c := range cases { var visit []uint64 - err := iterateCanonicalChain(chainDb, c.from, c.to, "", nil, func(db ethdb.Batch, b *types.Block) { visit = append(visit, b.NumberU64()) }, c.reverse, false) + err := iterateCanonicalChain(chainDb, c.from, c.to, nil, func(db ethdb.Batch, b *types.Block) { visit = append(visit, b.NumberU64()) }, c.reverse, "", "") if err != nil { t.Fatalf("Case %d failed, err %v", i, err) } diff --git a/core/rawdb/schema.go b/core/rawdb/schema.go index c38260aa4c..745a59922b 100644 --- a/core/rawdb/schema.go +++ b/core/rawdb/schema.go @@ -41,11 +41,11 @@ var ( // fastTrieProgressKey tracks the number of trie entries imported during fast sync. fastTrieProgressKey = []byte("TrieSync") - // txIndexTailKey tracks the oldest block whose transaction indices(txlookup) has been indexed. - txIndexTailKey = []byte("TxIndexTail") + // txIndexTailKey tracks the oldest block whose transactions have been indexed. + txIndexTailKey = []byte("TransactionIndexTail") - // fastTxLookupLimitKey tracks the tx indices limit during fast sync - fastTxLookupLimitKey = []byte("FSTxLookupLimit") + // fastTxLookupLimitKey tracks the transaction lookup limit during fast sync. + fastTxLookupLimitKey = []byte("FastTransactionLookupLimit") // 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