diff --git a/core/blockchain.go b/core/blockchain.go index 8050cd8de7..92331c6bea 100644 --- a/core/blockchain.go +++ b/core/blockchain.go @@ -941,7 +941,6 @@ func (bc *BlockChain) InsertReceiptChain(blockChain types.Blocks, receiptChain [ defer bc.wg.Done() if bc.cacheConfig.AncientRecentLimit != 0 { - log.Info("throwing ancient turned on, set ancientLimit to 0 during snap sync") ancientLimit = 0 } @@ -2303,7 +2302,7 @@ func (bc *BlockChain) maintainTxIndex(ancients uint64) { rawdb.WriteTxIndexTail(bc.db, 0) } else { // Prune all stale tx indices and record the tx index tail - log.Info("Scheduled transactions unindexing", "from block", 0, "to", head-bc.txLookupLimit+1) + log.Info("Scheduled blocks & transactions unindexing", "from block", 0, "to", head-bc.txLookupLimit+1) rawdb.UnindexTransactions(bc.db, 0, head-bc.txLookupLimit+1, bc.quit) } return @@ -2328,7 +2327,7 @@ 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("Scheduled transactions unindexing", "from block", *tail, "to", head-bc.txLookupLimit+1) + log.Info("Scheduled blocks & transactions unindexing", "from block", *tail, "to", head-bc.txLookupLimit+1) rawdb.UnindexTransactions(bc.db, *tail, head-bc.txLookupLimit+1, bc.quit) } } diff --git a/core/rawdb/chain_iterator.go b/core/rawdb/chain_iterator.go index 21e42f42d4..4d4e42b9dc 100644 --- a/core/rawdb/chain_iterator.go +++ b/core/rawdb/chain_iterator.go @@ -84,8 +84,9 @@ func InitDatabaseFromFreezer(db ethdb.Database) { } type blockTxHashes struct { - number uint64 - hashes []common.Hash + number uint64 + blockHash common.Hash + hashes []common.Hash } // iterateTransactions iterates over all transactions in the (canon) block @@ -96,6 +97,7 @@ func iterateTransactions(db ethdb.Database, from uint64, to uint64, reverse bool // One thread sequentially reads data from db type numberRlp struct { number uint64 + hash common.Hash rlp rlp.RawValue } if to == from { @@ -118,9 +120,10 @@ func iterateTransactions(db ethdb.Database, from uint64, to uint64, reverse bool defer close(rlpCh) for n != end { data := ReadCanonicalBodyRLP(db, n) + hash := ReadCanonicalHash(db, n) // Feed the block to the aggregator, or abort on interrupt select { - case rlpCh <- &numberRlp{n, data}: + case rlpCh <- &numberRlp{n, hash, data}: case <-interrupt: return } @@ -151,8 +154,9 @@ func iterateTransactions(db ethdb.Database, from uint64, to uint64, reverse bool hashes = append(hashes, tx.Hash()) } result := &blockTxHashes{ - hashes: hashes, - number: data.number, + hashes: hashes, + blockHash: data.hash, + number: data.number, } // Feed the block to the aggregator, or abort on interrupt select { @@ -303,6 +307,10 @@ func unindexTransactions(db ethdb.Database, from uint64, to uint64, interrupt ch nextNum = delivery.number + 1 DeleteTxLookupEntries(batch, delivery.hashes) txs += len(delivery.hashes) + // Delete all about the block + if delivery.number != 0 { // never delete the genesis block + DeleteBlock(batch, delivery.blockHash, delivery.number) + } blocks++ // If enough data was accumulated in memory or we're at the last block, dump to disk diff --git a/core/rawdb/chain_thrower.go b/core/rawdb/chain_thrower.go deleted file mode 100644 index 1ceadf2771..0000000000 --- a/core/rawdb/chain_thrower.go +++ /dev/null @@ -1,372 +0,0 @@ -// Copyright 2022 The go-ethereum Authors -// This file is part of the go-ethereum library. -// -// The go-ethereum library is free software: you can redistribute it and/or modify -// it under the terms of the GNU Lesser General Public License as published by -// the Free Software Foundation, either version 3 of the License, or -// (at your option) any later version. -// -// The go-ethereum library is distributed in the hope that it will be useful, -// but WITHOUT ANY WARRANTY; without even the implied warranty of -// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the -// GNU Lesser General Public License for more details. -// -// You should have received a copy of the GNU Lesser General Public License -// along with the go-ethereum library. If not, see . - -package rawdb - -import ( - "sync" - "sync/atomic" - "time" - - "github.com/ethereum/go-ethereum/common" - "github.com/ethereum/go-ethereum/ethdb" - "github.com/ethereum/go-ethereum/log" - "github.com/ethereum/go-ethereum/params" -) - -const ( - // throwerRecheckInterval is the frequency to check the key-value database for - // chain progression that might permit new blocks to be frozen into immutable - // storage. - throwerRecheckInterval = time.Minute - - // throwerBatchLimit is the maximum number of blocks to freeze in one batch - // before doing an fsync and deleting it from the key-value store. - 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 { - 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, - // so take advantage of that (https://golang.org/pkg/sync/atomic/#pkg-note-BUG). - threshold uint64 // Number of recent blocks not to freeze (params.FullImmutabilityThreshold apart from tests) - - quit chan struct{} - wg sync.WaitGroup - 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{ - threshold: params.FullImmutabilityThreshold, - quit: make(chan struct{}), - trigger: make(chan chan struct{}), - }, nil -} - -// Close closes the chain freezer instance and terminates the background thread. -func (f *chainThrower) Close() error { - select { - case <-f.quit: - default: - close(f.quit) - } - f.wg.Wait() - return nil -} - -// freeze is a background thread that periodically checks the blockchain for any -// import progress and moves ancient data from the fast database into the freezer. -// -// 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 := &throwdb{KeyValueStore: db} - - var ( - backoff bool - triggered chan struct{} // Used in tests - ) - for { - select { - case <-f.quit: - log.Info("Thrower shutting down") - return - default: - } - if backoff { - // If we were doing a manual trigger, notify it - if triggered != nil { - triggered <- struct{}{} - triggered = nil - } - select { - case <-time.NewTimer(throwerRecheckInterval).C: - backoff = false - case triggered = <-f.trigger: - backoff = false - case <-f.quit: - return - } - } - // Retrieve the freezing threshold. - hash := ReadHeadBlockHash(nfdb) - if hash == (common.Hash{}) { - log.Debug("Current full block hash unavailable") // new chain, empty database - backoff = true - continue - } - number := ReadHeaderNumber(nfdb, hash) - threshold := atomic.LoadUint64(&f.threshold) - last := ReadAncientThrowLastBlockNumber(db) - if last == nil { - last = new(uint64) - } - switch { - case number == nil: - log.Error("Current full block number unavailable", "hash", hash) - backoff = true - continue - - case *number < threshold: - log.Debug("Current full block not old enough", "number", *number, "hash", hash, "delay", threshold) - backoff = true - continue - - case *number-threshold <= *last: - log.Debug("Ancient blocks frozen already", "number", *number, "hash", hash, "last", last) - backoff = true - 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) - backoff = true - continue - } - - // Seems we have data ready to be frozen, process in usable batches - var ( - start = time.Now() - first = *last - limit = *number - threshold - ) - 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) - backoff = true - continue - } - - // Wipe out all data from the active database - batch := db.NewBatch() - for i := 0; i < len(ancients); i++ { - // Always keep the genesis block in active database - if first+uint64(i) != 0 { - DeleteBlockWithoutNumber(batch, ancients[i], first+uint64(i)) - DeleteCanonicalHash(batch, first+uint64(i)) - } - } - if err := batch.Write(); err != nil { - log.Crit("Failed to throw ancient blocks", "err", err) - } - batch.Reset() - - // record - WriteAncientThrowLastBlockNumber(db, limit) - - // Wipe out side chains also and track dangling side chains - var dangling []common.Hash - for number := first; number < limit; number++ { - // Always keep the genesis block in active database - if number != 0 { - dangling = ReadAllHashes(db, number) - for _, hash := range dangling { - log.Trace("Deleting side chain", "number", number, "hash", hash) - DeleteBlock(batch, hash, number) - } - } - } - if err := batch.Write(); err != nil { - log.Crit("Failed to delete frozen side blocks", "err", err) - } - batch.Reset() - - // Step into the future and delete and dangling side chains - if limit > 0 { - tip := limit - for len(dangling) > 0 { - drop := make(map[common.Hash]struct{}) - for _, hash := range dangling { - log.Debug("Dangling parent from Freezer", "number", tip-1, "hash", hash) - drop[hash] = struct{}{} - } - children := ReadAllHashes(db, tip) - for i := 0; i < len(children); i++ { - // Dig up the child and ensure it's dangling - child := ReadHeader(nfdb, children[i], tip) - if child == nil { - log.Error("Missing dangling header", "number", tip, "hash", children[i]) - continue - } - if _, ok := drop[child.ParentHash]; !ok { - children = append(children[:i], children[i+1:]...) - i-- - continue - } - // Delete all block data associated with the child - log.Debug("Deleting dangling block", "number", tip, "hash", children[i], "parent", child.ParentHash) - DeleteBlock(batch, children[i], tip) - } - dangling = children - tip++ - } - if err := batch.Write(); err != nil { - log.Crit("Failed to delete dangling side blocks", "err", err) - } - } - - // Log something friendly for the user - context := []interface{}{ - "blocks", limit - first, "elapsed", common.PrettyDuration(time.Since(start)), "number", limit - 1, - } - if n := len(ancients); n > 0 { - context = append(context, []interface{}{"hash", ancients[n-1]}...) - } - log.Info("Deep throw chain segment", context...) - - // Avoid database thrashing with tiny writes - if limit-first < freezerBatchLimit { - backoff = true - } - } -} - -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) - continue - } - header := ReadHeaderRLP(nfdb, hash, number) - if len(header) == 0 { - 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) - continue - } - receipts := ReadReceiptsRLP(nfdb, hash, number) - if len(receipts) == 0 { - 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) - continue - } - - hashes = append(hashes, hash) - } - - return hashes, err -} diff --git a/core/rawdb/database.go b/core/rawdb/database.go index 4960e439cd..adf9d0f2f0 100644 --- a/core/rawdb/database.go +++ b/core/rawdb/database.go @@ -165,7 +165,7 @@ func NewDatabase(db ethdb.KeyValueStore) ethdb.Database { // NewDatabaseWithFreezer creates a high level database on top of a given key- // value data store with a freezer moving immutable chain segments into cold // storage. -func NewDatabaseWithFreezer(db ethdb.KeyValueStore, freezer string, namespace string, readonly bool) (ethdb.Database, error) { +func NewDatabaseWithFreezer(db ethdb.KeyValueStore, freezer string, namespace string, readonly, discardAncient bool) (ethdb.Database, error) { // Create the idle freezer instance frdb, err := newChainFreezer(freezer, namespace, readonly, freezerTableSize, FreezerNoSnappy) if err != nil { @@ -237,7 +237,7 @@ func NewDatabaseWithFreezer(db ethdb.KeyValueStore, freezer string, namespace st } } // Freezer is consistent with the key-value database, permit combining the two - if !frdb.readonly { + if !frdb.readonly && !discardAncient { frdb.wg.Add(1) go func() { frdb.freeze(db) @@ -250,29 +250,6 @@ func NewDatabaseWithFreezer(db ethdb.KeyValueStore, freezer string, namespace st }, nil } -// NewDatabaseWithThrower creates a high level database on top of a given key- -// value data store with a freezer moving immutable chain segments into cold -// storage. -func NewDatabaseWithThrower(db ethdb.KeyValueStore, freezer string, namespace string, readonly bool) (ethdb.Database, error) { - // Create the idle freezer instance - throwdb, err := newChainThrower(freezer, namespace, readonly) - if err != nil { - return nil, err - } - - // Thrower is consistent with the key-value database, permit combining the two - throwdb.wg.Add(1) - go func() { - throwdb.throw(db) - throwdb.wg.Done() - }() - - return &freezerdb{ - KeyValueStore: db, - AncientStore: throwdb, - }, nil -} - // NewMemoryDatabase creates an ephemeral in-memory key-value database without a // freezer moving immutable chain segments into cold storage. func NewMemoryDatabase() ethdb.Database { @@ -298,29 +275,13 @@ func NewLevelDBDatabase(file string, cache int, handles int, namespace string, r // NewLevelDBDatabaseWithFreezer creates a persistent key-value database with a // freezer moving immutable chain segments into cold storage. -func NewLevelDBDatabaseWithFreezer(file string, cache int, handles int, freezer string, namespace string, readonly bool) (ethdb.Database, error) { +func NewLevelDBDatabaseWithFreezer(file string, cache int, handles int, freezer string, namespace string, readonly bool, ancientRecentLimit uint64) (ethdb.Database, error) { kvdb, err := leveldb.New(file, cache, handles, namespace, readonly) if err != nil { return nil, err } - frdb, err := NewDatabaseWithFreezer(kvdb, freezer, namespace, readonly) - if err != nil { - kvdb.Close() - return nil, err - } - return frdb, nil -} - -// NewLevelDBDatabaseWithFreezer creates a persistent key-value database with a -// freezer moving immutable chain segments into cold storage. -func NewLevelDBDatabaseWithThrower(file string, cache int, handles int, freezer string, namespace string, readonly bool) (ethdb.Database, error) { - kvdb, err := leveldb.New(file, cache, handles, namespace, readonly) - if err != nil { - return nil, err - } - - frdb, err := NewDatabaseWithThrower(kvdb, freezer, namespace, readonly) + frdb, err := NewDatabaseWithFreezer(kvdb, freezer, namespace, readonly, ancientRecentLimit != 0) if err != nil { kvdb.Close() return nil, err diff --git a/eth/backend.go b/eth/backend.go index 9a887ef2f0..4016ce5848 100644 --- a/eth/backend.go +++ b/eth/backend.go @@ -133,7 +133,7 @@ func New(stack *node.Node, config *ethconfig.Config) (*Ethereum, error) { ethashConfig.NotifyFull = config.Miner.NotifyFull // Assemble the Ethereum object - chainDb, err := stack.OpenDatabaseWrapper("chaindata", config.DatabaseCache, config.DatabaseHandles, config.DatabaseFreezer, "eth/db/chaindata/", false, config.AncientRecentLimit != 0) + chainDb, err := stack.OpenDatabaseWithFreezer("chaindata", config.DatabaseCache, config.DatabaseHandles, config.DatabaseFreezer, "eth/db/chaindata/", false, config.AncientRecentLimit) if err != nil { return nil, err } @@ -215,7 +215,9 @@ func New(stack *node.Node, config *ethconfig.Config) (*Ethereum, error) { eth.blockchain.SetHead(compat.RewindTo) rawdb.WriteChainConfig(chainDb, genesisHash, chainConfig) } - eth.bloomIndexer.Start(eth.blockchain) + if config.AncientRecentLimit == 0 { + eth.bloomIndexer.Start(eth.blockchain) + } if config.TxPool.Journal != "" { config.TxPool.Journal = stack.ResolvePath(config.TxPool.Journal) diff --git a/node/node.go b/node/node.go index b390abdb4b..c67bbd01d8 100644 --- a/node/node.go +++ b/node/node.go @@ -712,38 +712,6 @@ func (n *Node) OpenDatabase(name string, cache, handles int, namespace string, r return db, err } -// OpenDatabaseWrapper ... -func (n *Node) OpenDatabaseWrapper(name string, cache, handles int, freezer, namespace string, readonly, throwAncient bool) (ethdb.Database, error) { - n.lock.Lock() - defer n.lock.Unlock() - if n.state == closedState { - return nil, ErrNodeStopped - } - - var db ethdb.Database - var err error - if n.config.DataDir == "" { - db = rawdb.NewMemoryDatabase() - } else if !throwAncient { - root := n.ResolvePath(name) - switch { - case freezer == "": - freezer = filepath.Join(root, "ancient") - case !filepath.IsAbs(freezer): - freezer = n.ResolvePath(freezer) - } - db, err = rawdb.NewLevelDBDatabaseWithFreezer(root, cache, handles, freezer, namespace, readonly) - } else { - root := n.ResolvePath(name) - db, err = rawdb.NewLevelDBDatabaseWithThrower(root, cache, handles, freezer, namespace, readonly) - } - - if err == nil { - db = n.wrapDatabase(db) - } - return db, err -} - // OpenDatabaseWithFreezer opens an existing database with the given name (or // creates one if no previous can be found) from within the node's data directory, // also attaching a chain freezer to it that moves ancient chain data from the @@ -768,7 +736,7 @@ func (n *Node) OpenDatabaseWithFreezer(name string, cache, handles int, freezer, case !filepath.IsAbs(freezer): freezer = n.ResolvePath(freezer) } - db, err = rawdb.NewLevelDBDatabaseWithFreezer(root, cache, handles, freezer, namespace, readonly) + db, err = rawdb.NewLevelDBDatabaseWithFreezer(root, cache, handles, freezer, namespace, readonly, ancientRecentLimit) } if err == nil {