refine the implementation

cmd/utils/flags.go

@@ -1663,12 +1663,7 @@ func SetEthConfig(ctx *cli.Context, stack *node.Node, cfg *ethconfig.Config) {
 	}
 	if ctx.GlobalIsSet(AncientRecentLimitFlag.Name) {
 		cfg.AncientRecentLimit = ctx.GlobalUint64(AncientRecentLimitFlag.Name)
-		if cfg.TxLookupLimit == 0 || cfg.TxLookupLimit > cfg.AncientRecentLimit+params.FullImmutabilityThreshold {
-			cfg.TxLookupLimit = cfg.AncientRecentLimit + params.FullImmutabilityThreshold
-			log.Warn("Reducing TxLookupLimit to meet ancient db purging, as it's insane to lookup txs in purged blocks")
-		}
 	}
-
 	if ctx.GlobalIsSet(CacheFlag.Name) || ctx.GlobalIsSet(CacheTrieFlag.Name) {
 		cfg.TrieCleanCache = ctx.GlobalInt(CacheFlag.Name) * ctx.GlobalInt(CacheTrieFlag.Name) / 100
 	}

core/blockchain.go

@@ -127,6 +127,7 @@ const (
 
 // CacheConfig contains the configuration values for the trie caching/pruning
 // that's resident in a blockchain.
+// FIXME Change this config's name so it suits for both cache and prune config
 type CacheConfig struct {
 	TriesInMemory       int           // Keeps the latest n blocks when pruning.
 	TrieCleanLimit      int           // Memory allowance (MB) to use for caching trie nodes in memory
@@ -139,6 +140,8 @@ type CacheConfig struct {
 	SnapshotLimit       int           // Memory allowance (MB) to use for caching snapshot entries in memory
 	Preimages           bool          // Whether to store preimage of trie key to the disk
 
+	AncientRecentLimit uint64
+
 	SnapshotWait bool // Wait for snapshot construction on startup. TODO(karalabe): This is a dirty hack for testing, nuke it
 }
 
@@ -926,6 +929,8 @@ const (
 
 // InsertReceiptChain attempts to complete an already existing header chain with
 // transaction and receipt data.
+// (The function name is confusing here, though it calls "insert receipt", it inserts block body for sure,
+//  but it doesn't process the transaction in the block, which is different from the InsertChain() function)
 func (bc *BlockChain) InsertReceiptChain(blockChain types.Blocks, receiptChain []types.Receipts, ancientLimit uint64) (int, error) {
 	// We don't require the chainMu here since we want to maximize the
 	// concurrency of header insertion and receipt insertion.
@@ -2265,18 +2270,38 @@ func (bc *BlockChain) skipBlock(err error, it *insertIterator) bool {
 func (bc *BlockChain) maintainTxIndex(ancients uint64) {
 	defer bc.wg.Done()
 
+	ancientLimit := bc.cacheConfig.AncientRecentLimit
+	frozen, _ := bc.db.Ancients()
+	log.Info("maintainTxIndex", "ancientLimit", ancientLimit, "ancients", ancients, "frozen", frozen)
 	// 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
+		// let `from` be a "more" recent block
+		if ancientLimit > 0 {
+			if ancientLimit < ancients {
+				from = ancients - ancientLimit
+			}
+			if bc.txLookupLimit != 0 && bc.txLookupLimit < ancientLimit {
+				from = ancients - bc.txLookupLimit
+			}
+		} else {
+			if bc.txLookupLimit != 0 && ancients > bc.txLookupLimit {
+				from = ancients - bc.txLookupLimit
+			}
 		}
+
+		log.Info("maintainTxIndex", "ancientLimit", ancientLimit, "ancients", ancients, "from", from)
 		rawdb.IndexTransactions(bc.db, from, ancients, bc.quit)
 	}
 
+	if ancientLimit > 0 && (bc.txLookupLimit == 0 || bc.txLookupLimit > ancientLimit) {
+		log.Warn("reduece txLookupLimit to meet ancient purging purpose, as it's insane to lookup txs in purged blocks")
+		bc.txLookupLimit = ancientLimit
+	}
+
 	// indexBlocks reindexes or unindexes transactions depending on user configuration
 	indexBlocks := func(tail *uint64, head uint64, done chan struct{}) {
 		defer func() { done <- struct{}{} }()
@@ -2313,13 +2338,58 @@ func (bc *BlockChain) maintainTxIndex(ancients uint64) {
 			rawdb.IndexTransactions(bc.db, head-bc.txLookupLimit+1, *tail, bc.quit)
 		} else {
 			// Unindex a part of stale indices and forward index tail to HEAD-limit
+			log.Info("unindex transactions", "tail", *tail, "to", head-bc.txLookupLimit+1)
 			rawdb.UnindexTransactions(bc.db, *tail, head-bc.txLookupLimit+1, bc.quit)
 		}
 	}
 
+	// pruneAncient is a background thread that periodically checks the blockchain for any
+	// import progress and cleans ancient data from the freezer.
+	//
+	// it calculates the highest block(named `cleanTo`) that can be cleaned, then starts
+	// cleaning from the genesis+1 block in batches, until reaches the `cleanTo` block.
+	pruneAncient := func(tail *uint64, head uint64, wait, done chan struct{}) {
+		defer func() { done <- struct{}{} }()
+		// Wait till the tx index routine finishes, in case that it cannot find the ancient
+		// blocks for iterating transactions for unindex because we have pruned the blocks
+		select {
+		case <-wait:
+		}
+
+		start := time.Now()
+
+		first, _ := bc.db.Tail()
+		last, _ := bc.db.Ancients()
+		ancientLimit := bc.cacheConfig.AncientRecentLimit
+
+		if last < ancientLimit || last < first {
+			// It should not reach here
+			log.Error("prune ancient error", "frozen", frozen, "tail", tail, "ancientRecentLimit", ancientLimit)
+			return
+		}
+
+		pruneTo := last - ancientLimit
+
+		// Double ensure we don't prune the blocks having dangling transaction indices
+		if tail != nil && pruneTo > *tail {
+			log.Warn("Attempt to prune the ancient blocks that still have tx indices, postpone to next round")
+			return
+		}
+
+		// truncate
+		bc.db.TruncateTail(pruneTo)
+
+		// Log something friendly for the user
+		context := []interface{}{
+			"blocks", pruneTo - first, "frozen", frozen, "ancientRecentLimit", ancientLimit, "elapsed", common.PrettyDuration(time.Since(start)), "pruned to", pruneTo,
+		}
+		log.Info("cleaned chain segment", context...)
+	}
+
 	// 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.
+		doneTx chan struct{}                  // For tx indexing routine, non-nil if the routine is active.
+		donePr chan struct{}                  // For prune ancient routine, non-nil if the routine is active.
 		headCh = make(chan ChainHeadEvent, 1) // Buffered to avoid locking up the event feed
 	)
 	sub := bc.SubscribeChainHeadEvent(headCh)
@@ -2331,16 +2401,20 @@ func (bc *BlockChain) maintainTxIndex(ancients uint64) {
 	for {
 		select {
 		case head := <-headCh:
-			if done == nil {
-				done = make(chan struct{})
-				go indexBlocks(rawdb.ReadTxIndexTail(bc.db), head.Block.NumberU64(), done)
+			log.Info("chainHeadEvent", "received", head)
+			if doneTx == nil && donePr == nil {
+				doneTx = make(chan struct{})
+				donePr = make(chan struct{})
+				go indexBlocks(rawdb.ReadTxIndexTail(bc.db), head.Block.NumberU64(), doneTx)
+				go pruneAncient(rawdb.ReadTxIndexTail(bc.db), head.Block.NumberU64(), doneTx, donePr)
 			}
-		case <-done:
-			done = nil
+		case <-donePr:
+			donePr = nil
+			doneTx = nil
 		case <-bc.quit:
-			if done != nil {
-				log.Info("Waiting background transaction indexer to exit")
-				<-done
+			if donePr != nil {
+				log.Info("Waiting background transaction indexer and ancient pruner to exit")
+				<-donePr
 			}
 			return
 		}

core/rawdb/chain_freezer.go

@@ -259,57 +259,6 @@ func (f *chainFreezer) freeze(db ethdb.KeyValueStore) {
 	}
 }
 
-// clean is a background thread that periodically checks the blockchain for any
-// import progress and cleans ancient data from the freezer.
-//
-// it calculates the highest block(named `cleanTo`) that can be cleaned, then starts
-// cleaning from the genesis+1 block in batches, until reaches the `cleanTo` block.
-func (f *chainFreezer) clean(db ethdb.KeyValueStore) {
-	nfdb := &nofreezedb{KeyValueStore: db}
-
-	for {
-		select {
-		case <-time.NewTimer(cleanerRecheckInterval).C:
-		case <-f.quit:
-			log.Info("Freezer cleaner shutting down")
-			return
-		}
-
-		// we won't start the prunning until chain has initialised itself
-		hash := ReadHeadBlockHash(nfdb)
-		log.Info("Read head block hash", hash.Hex())
-		if hash == (common.Hash{}) {
-			log.Info("current full block hash unavailable, schedule next round") // new chain, empty database
-			continue
-		}
-
-		// Calculates the cleanTo block
-		number := ReadHeaderNumber(nfdb, hash)
-		if number == nil {
-			log.Info("Current full block number unavailable, schedule next round", "hash", hash)
-			continue
-		}
-
-		start := time.Now()
-		tail, _ := f.Tail()
-		frozen, _ := f.Ancients()
-		if frozen < f.ancientRecentLimit || frozen < tail {
-			log.Error("ancient clean error target block", "frozen", frozen, "tail", tail, "ancientRecentLimit", f.ancientRecentLimit)
-			continue
-		}
-
-		cleanTo := frozen - f.ancientRecentLimit
-		// truncate
-		f.TruncateTail(cleanTo)
-
-		// Log something friendly for the user
-		context := []interface{}{
-			"blocks", cleanTo - tail, "frozen", frozen, "ancientRecentLimit", f.ancientRecentLimit, "elapsed", common.PrettyDuration(time.Since(start)), "cleaned to", cleanTo,
-		}
-		log.Info("cleaned chain segment", context...)
-	}
-}
-
 func (f *chainFreezer) freezeRange(nfdb *nofreezedb, number, limit uint64) (hashes []common.Hash, err error) {
 	hashes = make([]common.Hash, 0, limit-number)
 

core/rawdb/database.go

@@ -193,6 +193,7 @@ func NewDatabaseWithFreezer(db ethdb.KeyValueStore, freezer string, namespace st
 	// If the genesis hash is empty, we have a new key-value store, so nothing to
 	// validate in this method. If, however, the genesis hash is not nil, compare
 	// it to the freezer content.
+	// TODO consider deprecate this check
 	if kvgenesis, _ := db.Get(headerHashKey(0)); len(kvgenesis) > 0 && frdb.ancientRecentLimit == 0 {
 		if frozen, _ := frdb.Ancients(); frozen > 0 {
 			// If the freezer already contains something, ensure that the genesis blocks
@@ -242,14 +243,6 @@ func NewDatabaseWithFreezer(db ethdb.KeyValueStore, freezer string, namespace st
 			frdb.freeze(db)
 			frdb.wg.Done()
 		}()
-
-		if frdb.ancientRecentLimit != 0 {
-			frdb.wg.Add(1)
-			go func() {
-				frdb.clean(db)
-				frdb.wg.Done()
-			}()
-		}
 	}
 	return &freezerdb{
 		KeyValueStore: db,

eth/backend.go

@@ -202,6 +202,7 @@ func New(stack *node.Node, config *ethconfig.Config) (*Ethereum, error) {
 			TrieTimeLimit:       config.TrieTimeout,
 			SnapshotLimit:       config.SnapshotCache,
 			Preimages:           config.Preimages,
+			AncientRecentLimit:  config.AncientRecentLimit,
 		}
 	)
 	eth.blockchain, err = core.NewBlockChain(chainDb, cacheConfig, chainConfig, eth.engine, vmConfig, eth.shouldPreserve, &config.TxLookupLimit)