From d23afb431b8281c6536e23fb7522423fe8860fc4 Mon Sep 17 00:00:00 2001 From: cifer76 Date: Fri, 24 Jun 2022 03:30:43 +0800 Subject: [PATCH] refine the implementation --- cmd/utils/flags.go | 5 -- core/blockchain.go | 96 ++++++++++++++++++++++++++++++++----- core/rawdb/chain_freezer.go | 51 -------------------- core/rawdb/database.go | 9 +--- eth/backend.go | 1 + 5 files changed, 87 insertions(+), 75 deletions(-) diff --git a/cmd/utils/flags.go b/cmd/utils/flags.go index e9df3dcdfa..59804e4da0 100644 --- a/cmd/utils/flags.go +++ b/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 } diff --git a/core/blockchain.go b/core/blockchain.go index 2a58ed11fe..6a405e36be 100644 --- a/core/blockchain.go +++ b/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 } diff --git a/core/rawdb/chain_freezer.go b/core/rawdb/chain_freezer.go index 6b637a3b5e..6f602f7b51 100644 --- a/core/rawdb/chain_freezer.go +++ b/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) diff --git a/core/rawdb/database.go b/core/rawdb/database.go index e4a2617c65..6cb5f69f99 100644 --- a/core/rawdb/database.go +++ b/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, diff --git a/eth/backend.go b/eth/backend.go index 6dcfe1e7d9..48d7d1c249 100644 --- a/eth/backend.go +++ b/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)