From 81157fc4a8fa60dfc36e849852f068253158b298 Mon Sep 17 00:00:00 2001 From: user Date: Thu, 30 Jun 2022 17:05:44 +0800 Subject: [PATCH] temprorarily stash the implementation --- core/blockchain.go | 97 ++++++------------------------ core/rawdb/chain_thrower.go | 115 ++++++++++++++++++++++++++++++++---- 2 files changed, 120 insertions(+), 92 deletions(-) diff --git a/core/blockchain.go b/core/blockchain.go index 366782800d..8050cd8de7 100644 --- a/core/blockchain.go +++ b/core/blockchain.go @@ -406,6 +406,9 @@ func NewBlockChain(db ethdb.Database, cacheConfig *CacheConfig, chainConfig *par // Start tx indexer/unindexer. if txLookupLimit != nil { bc.txLookupLimit = *txLookupLimit + if bc.cacheConfig.AncientRecentLimit != 0 { + bc.txLookupLimit = params.FullImmutabilityThreshold + } bc.wg.Add(1) go bc.maintainTxIndex(txIndexBlock) @@ -937,6 +940,11 @@ func (bc *BlockChain) InsertReceiptChain(blockChain types.Blocks, receiptChain [ bc.wg.Add(1) defer bc.wg.Done() + if bc.cacheConfig.AncientRecentLimit != 0 { + log.Info("throwing ancient turned on, set ancientLimit to 0 during snap sync") + ancientLimit = 0 + } + var ( ancientBlocks, liveBlocks types.Blocks ancientReceipts, liveReceipts []types.Receipts @@ -2270,38 +2278,19 @@ 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) - // 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 - } + 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{}{} }() @@ -2344,56 +2333,9 @@ func (bc *BlockChain) maintainTxIndex(ancients uint64) { } } - // 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(txIndexTail *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 - <-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", "last frozen", last, "first frozen", first, "ancientRecentLimit", ancientLimit) - return - } - - pruneTo := last - ancientLimit - storedSections := rawdb.ReadStoredBloomSections(bc.db) - if storedSections*params.BloomBitsBlocks-1 < pruneTo { - log.Warn("Attempt to prune the ancient blocks that bloom filter haven't finished yet, postpone to next round", "storedSections", storedSections, "pruneTo", pruneTo) - return - } - - // Double ensure we don't prune the blocks having dangling transaction indices - if txIndexTail != nil && pruneTo > *txIndexTail { - log.Warn("Attempt to prune the ancient blocks that still have tx indices, postpone to next round", "txIndexTail", *txIndexTail, "pruneTo", pruneTo) - return - } - - // truncate - bc.db.TruncateTail(pruneTo) - - // Log something friendly for the user - context := []interface{}{ - "blocks", pruneTo - first, "from block", first, "to", pruneTo, "current last block in ancient", last, "elapsed", common.PrettyDuration(time.Since(start)), - } - log.Info("Cleaned ancient chain segment", context...) - } - // Any reindexing done, start listening to chain events and moving the index window var ( - 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. + done chan struct{} // For tx indexing 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) @@ -2405,19 +2347,16 @@ func (bc *BlockChain) maintainTxIndex(ancients uint64) { for { select { case head := <-headCh: - 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) + if done == nil { + done = make(chan struct{}) + go indexBlocks(rawdb.ReadTxIndexTail(bc.db), head.Block.NumberU64(), done) } - case <-donePr: - donePr = nil - doneTx = nil + case <-done: + done = nil case <-bc.quit: - if donePr != nil { + if done != nil { log.Info("Waiting background transaction indexer and ancient pruner to exit") - <-donePr + <-done } return } diff --git a/core/rawdb/chain_thrower.go b/core/rawdb/chain_thrower.go index 1433cd5746..1ceadf2771 100644 --- a/core/rawdb/chain_thrower.go +++ b/core/rawdb/chain_thrower.go @@ -28,21 +28,21 @@ import ( ) const ( - // freezerRecheckInterval is the frequency to check the key-value database for + // throwerRecheckInterval is the frequency to check the key-value database for // chain progression that might permit new blocks to be frozen into immutable // storage. - nofreezerRecheckInterval = time.Minute + throwerRecheckInterval = time.Minute - // freezerBatchLimit is the maximum number of blocks to freeze in one batch + // throwerBatchLimit is the maximum number of blocks to freeze in one batch // before doing an fsync and deleting it from the key-value store. - nofreezerBatchLimit = 30000 + throwerBatchLimit = 30000 ) // chainThrower is a wrapper of freezer with additional chain freezing feature. // The background thread will keep moving ancient chain segments from key-value // database to flat files for saving space on live database. type chainThrower struct { - nofreezedb + throwdb // WARNING: The `threshold` field is accessed atomically. On 32 bit platforms, only // 64-bit aligned fields can be atomic. The struct is guaranteed to be so aligned, @@ -54,6 +54,88 @@ type chainThrower struct { trigger chan chan struct{} // Manual blocking freeze trigger, test determinism } +// throwdb is a database wrapper that disables freezer data retrievals. +type throwdb struct { + ethdb.KeyValueStore +} + +// HasAncient always returns false as `throwdb` has thrown everything written to it +func (db *throwdb) HasAncient(kind string, number uint64) (bool, error) { + return false, nil +} + +// Ancient returns an empty result as `throwdb` has thrown everything written to it +func (db *throwdb) Ancient(kind string, number uint64) ([]byte, error) { + return []byte{}, nil +} + +// AncientRange returns an empty result as `throwdb` has thrown everything written to it +func (db *throwdb) AncientRange(kind string, start, max, maxByteSize uint64) ([][]byte, error) { + return [][]byte{}, nil +} + +// Ancients returns 0 as we don't have a backing chain freezer. +func (db *throwdb) Ancients() (uint64, error) { + return 0, nil +} + +// Tail returns 0 as we don't have a backing chain freezer. +func (db *throwdb) Tail() (uint64, error) { + return 0, errNotSupported +} + +// AncientSize returns an error as we don't have a backing chain freezer. +func (db *throwdb) AncientSize(kind string) (uint64, error) { + return 0, nil +} + +// ModifyAncients is not supported. +func (db *throwdb) ModifyAncients(func(ethdb.AncientWriteOp) error) (int64, error) { + return 0, nil +} + +// TruncateHead returns an error as we don't have a backing chain freezer. +func (db *throwdb) TruncateHead(items uint64) error { + return nil +} + +// TruncateTail returns an error as we don't have a backing chain freezer. +func (db *throwdb) TruncateTail(items uint64) error { + return nil +} + +// Sync returns an error as we don't have a backing chain freezer. +func (db *throwdb) Sync() error { + return nil +} + +func (db *throwdb) ReadAncients(fn func(reader ethdb.AncientReaderOp) error) (err error) { + // Unlike other ancient-related methods, this method does not return + // errNotSupported when invoked. + // The reason for this is that the caller might want to do several things: + // 1. Check if something is in freezer, + // 2. If not, check leveldb. + // + // This will work, since the ancient-checks inside 'fn' will return errors, + // and the leveldb work will continue. + // + // If we instead were to return errNotSupported here, then the caller would + // have to explicitly check for that, having an extra clause to do the + // non-ancient operations. + return fn(db) +} + +// MigrateTable processes the entries in a given table in sequence +// converting them to a new format if they're of an old format. +func (db *throwdb) MigrateTable(kind string, convert convertLegacyFn) error { + return errNotSupported +} + +// AncientDatadir returns an error as we don't have a backing chain freezer. +func (db *throwdb) AncientDatadir() (string, error) { + return "", errNotSupported +} + // newChainFreezer initializes the freezer for ancient chain data. func newChainThrower(datadir string, namespace string, readonly bool) (*chainThrower, error) { return &chainThrower{ @@ -80,7 +162,7 @@ func (f *chainThrower) Close() error { // This functionality is deliberately broken off from block importing to avoid // incurring additional data shuffling delays on block propagation. func (f *chainThrower) throw(db ethdb.KeyValueStore) { - nfdb := &nofreezedb{KeyValueStore: db} + nfdb := &throwdb{KeyValueStore: db} var ( backoff bool @@ -100,7 +182,7 @@ func (f *chainThrower) throw(db ethdb.KeyValueStore) { triggered = nil } select { - case <-time.NewTimer(freezerRecheckInterval).C: + case <-time.NewTimer(throwerRecheckInterval).C: backoff = false case triggered = <-f.trigger: backoff = false @@ -138,6 +220,12 @@ func (f *chainThrower) throw(db ethdb.KeyValueStore) { continue } + storedSections := ReadStoredBloomSections(nfdb) + if storedSections*params.BloomBitsBlocks-1 < *last { + log.Warn("Attempt to prune the ancient blocks that bloom filter haven't finished yet, postpone to next round", "storedSections", storedSections, "pruneTo", *last) + return + } + head := ReadHeader(nfdb, hash, *number) if head == nil { log.Error("Current full block unavailable", "number", *number, "hash", hash) @@ -154,6 +242,7 @@ func (f *chainThrower) throw(db ethdb.KeyValueStore) { if limit-first > freezerBatchLimit { limit = first + freezerBatchLimit } + log.Info("schedule throwing blocks", "from", first, "to", limit) ancients, err := f.throwRange(nfdb, first, limit) if err != nil { log.Error("Error in block freeze operation", "err", err) @@ -245,34 +334,34 @@ func (f *chainThrower) throw(db ethdb.KeyValueStore) { } } -func (f *chainThrower) throwRange(nfdb *nofreezedb, number, limit uint64) (hashes []common.Hash, err error) { +func (f *chainThrower) throwRange(nfdb *throwdb, number, limit uint64) (hashes []common.Hash, err error) { hashes = make([]common.Hash, 0, limit-number) for ; number <= limit; number++ { // Retrieve all the components of the canonical block. hash := ReadCanonicalHash(nfdb, number) if hash == (common.Hash{}) { - log.Error("canonical hash missing, can't freeze block %d", number) + log.Error("canonical hash missing, can't freeze", "block %d", number) continue } header := ReadHeaderRLP(nfdb, hash, number) if len(header) == 0 { - log.Error("block header missing, can't freeze block %d", number) + log.Error("block header missing, can't freeze", "block %d", number) continue } body := ReadBodyRLP(nfdb, hash, number) if len(body) == 0 { - log.Error("block body missing, can't freeze block %d", number) + log.Error("block body missing, can't freeze", "block %d", number) continue } receipts := ReadReceiptsRLP(nfdb, hash, number) if len(receipts) == 0 { - log.Error("block receipts missing, can't freeze block %d", number) + log.Error("block receipts missing, can't freeze", "block %d", number) continue } td := ReadTdRLP(nfdb, hash, number) if len(td) == 0 { - log.Error("total difficulty missing, can't freeze block %d", number) + log.Error("total difficulty missing, can't freeze", "block %d", number) continue }