diff --git a/core/rawdb/accessors_chain.go b/core/rawdb/accessors_chain.go index 8ea2e2ca72..4fbba22b55 100644 --- a/core/rawdb/accessors_chain.go +++ b/core/rawdb/accessors_chain.go @@ -168,6 +168,24 @@ func DeleteHeaderNumber(db ethdb.KeyValueWriter, hash common.Hash) { } } +// ReadAncientThrowLastBlock retrieves the hash of the finalized block. +func ReadAncientThrowLastBlockNumber(db ethdb.KeyValueReader) *uint64 { + data, _ := db.Get(ancientThrowLastBlock) + if len(data) != 8 { + return nil + } + number := binary.BigEndian.Uint64(data) + return &number +} + +// WriteAncientThrowLastBlock stores the hash of the finalized block. +func WriteAncientThrowLastBlockNumber(db ethdb.KeyValueWriter, number uint64) { + enc := encodeBlockNumber(number) + if err := db.Put(ancientThrowLastBlock, enc); err != nil { + log.Crit("Failed to store last throwed ancient block's hash", "err", err) + } +} + // ReadHeadHeaderHash retrieves the hash of the current canonical head header. func ReadHeadHeaderHash(db ethdb.KeyValueReader) common.Hash { data, _ := db.Get(headHeaderKey) diff --git a/core/rawdb/chain_freezer.go b/core/rawdb/chain_freezer.go index 6f602f7b51..4c49db2748 100644 --- a/core/rawdb/chain_freezer.go +++ b/core/rawdb/chain_freezer.go @@ -37,10 +37,6 @@ const ( // freezerBatchLimit is the maximum number of blocks to freeze in one batch // before doing an fsync and deleting it from the key-value store. freezerBatchLimit = 30000 - - // cleanerRecheckInterval is the frequency to check the freezer database to - // find the bloks that might be pruned - cleanerRecheckInterval = 1 * time.Minute ) // chainFreezer is a wrapper of freezer with additional chain freezing feature. @@ -52,10 +48,6 @@ type chainFreezer struct { // 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) - // ancientRecentLimit Number of recent blocks to keep in ancient db - // if not 0, blocks older than `HEAD - ancientRecentLimit` will be purged from ancient db - ancientRecentLimit uint64 - *Freezer quit chan struct{} wg sync.WaitGroup @@ -63,17 +55,16 @@ type chainFreezer struct { } // newChainFreezer initializes the freezer for ancient chain data. -func newChainFreezer(datadir string, namespace string, readonly bool, maxTableSize uint32, tables map[string]bool, ancientRecentLimit uint64) (*chainFreezer, error) { +func newChainFreezer(datadir string, namespace string, readonly bool, maxTableSize uint32, tables map[string]bool) (*chainFreezer, error) { freezer, err := NewFreezer(datadir, namespace, readonly, maxTableSize, tables) if err != nil { return nil, err } return &chainFreezer{ - Freezer: freezer, - ancientRecentLimit: ancientRecentLimit, - threshold: params.FullImmutabilityThreshold, - quit: make(chan struct{}), - trigger: make(chan chan struct{}), + Freezer: freezer, + threshold: params.FullImmutabilityThreshold, + quit: make(chan struct{}), + trigger: make(chan chan struct{}), }, nil } diff --git a/core/rawdb/chain_thrower.go b/core/rawdb/chain_thrower.go new file mode 100644 index 0000000000..1433cd5746 --- /dev/null +++ b/core/rawdb/chain_thrower.go @@ -0,0 +1,283 @@ +// 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 ( + // freezerRecheckInterval 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 + + // freezerBatchLimit 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 +) + +// 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 + + // 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 +} + +// 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 := &nofreezedb{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(freezerRecheckInterval).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 + } + + 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 + } + 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 *nofreezedb, 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 6cb5f69f99..4960e439cd 100644 --- a/core/rawdb/database.go +++ b/core/rawdb/database.go @@ -165,9 +165,9 @@ 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, ancientRecentLimit uint64) (ethdb.Database, error) { +func NewDatabaseWithFreezer(db ethdb.KeyValueStore, freezer string, namespace string, readonly bool) (ethdb.Database, error) { // Create the idle freezer instance - frdb, err := newChainFreezer(freezer, namespace, readonly, freezerTableSize, FreezerNoSnappy, ancientRecentLimit) + frdb, err := newChainFreezer(freezer, namespace, readonly, freezerTableSize, FreezerNoSnappy) if err != nil { return nil, err } @@ -194,7 +194,7 @@ func NewDatabaseWithFreezer(db ethdb.KeyValueStore, freezer string, namespace st // 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 kvgenesis, _ := db.Get(headerHashKey(0)); len(kvgenesis) > 0 { if frozen, _ := frdb.Ancients(); frozen > 0 { // If the freezer already contains something, ensure that the genesis blocks // match, otherwise we might mix up freezers across chains and destroy both @@ -250,6 +250,29 @@ 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 { @@ -275,12 +298,29 @@ 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, ancientRecentLimit uint64) (ethdb.Database, error) { +func NewLevelDBDatabaseWithFreezer(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 := NewDatabaseWithFreezer(kvdb, freezer, namespace, readonly, ancientRecentLimit) + + 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) if err != nil { kvdb.Close() return nil, err diff --git a/core/rawdb/schema.go b/core/rawdb/schema.go index 041c9f0449..dce29a2d14 100644 --- a/core/rawdb/schema.go +++ b/core/rawdb/schema.go @@ -84,6 +84,9 @@ var ( // transitionStatusKey tracks the eth2 transition status. transitionStatusKey = []byte("eth2-transition") + // ancientThrowLastBlock tracks the last recent ancient block that has been thrown + ancientThrowLastBlock = []byte("AncientThrowLastBlock") + // 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 headerTDSuffix = []byte("t") // headerPrefix + num (uint64 big endian) + hash + headerTDSuffix -> td diff --git a/eth/backend.go b/eth/backend.go index 48d7d1c249..9a887ef2f0 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.OpenDatabaseWithFreezer("chaindata", config.DatabaseCache, config.DatabaseHandles, config.DatabaseFreezer, "eth/db/chaindata/", false, config.AncientRecentLimit) + chainDb, err := stack.OpenDatabaseWrapper("chaindata", config.DatabaseCache, config.DatabaseHandles, config.DatabaseFreezer, "eth/db/chaindata/", false, config.AncientRecentLimit != 0) if err != nil { return nil, err } diff --git a/node/node.go b/node/node.go index c67bbd01d8..b390abdb4b 100644 --- a/node/node.go +++ b/node/node.go @@ -712,6 +712,38 @@ 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 @@ -736,7 +768,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, ancientRecentLimit) + db, err = rawdb.NewLevelDBDatabaseWithFreezer(root, cache, handles, freezer, namespace, readonly) } if err == nil {