cmd, core: minor polishes to the transaction indexer

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)

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:

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)
 	}
 }
 

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) {

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)
 	}
 }

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)
 		}

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