diff --git a/.gitignore b/.gitignore index 1ee8b83022..f43105dfdb 100644 --- a/.gitignore +++ b/.gitignore @@ -47,3 +47,6 @@ profile.cov /dashboard/assets/package-lock.json **/yarn-error.log + +# vim swap files +*.swp diff --git a/cmd/geth/chaincmd.go b/cmd/geth/chaincmd.go index 6e19edeb46..0b09344bf0 100644 --- a/cmd/geth/chaincmd.go +++ b/cmd/geth/chaincmd.go @@ -192,7 +192,7 @@ func initGenesis(ctx *cli.Context) error { stack, _ := makeConfigNode(ctx) defer stack.Close() for _, name := range []string{"chaindata", "lightchaindata"} { - chaindb, err := stack.OpenDatabaseWithFreezer(name, 0, 0, ctx.GlobalString(utils.AncientFlag.Name), "", false) + chaindb, err := stack.OpenDatabaseWithFreezer(name, 0, 0, ctx.GlobalString(utils.AncientFlag.Name), "", false, ctx.GlobalUint64(utils.AncientRecentLimitFlag.Name)) if err != nil { utils.Fatalf("Failed to open database: %v", err) } diff --git a/cmd/utils/flags.go b/cmd/utils/flags.go index db78a78a69..c80de61104 100644 --- a/cmd/utils/flags.go +++ b/cmd/utils/flags.go @@ -122,6 +122,11 @@ var ( Name: "datadir.ancient", Usage: "Data directory for ancient chain segments (default = inside chaindata)", } + AncientRecentLimitFlag = cli.Uint64Flag{ + Name: "ancient.recentlimit", + Usage: "Keep only the specified amount of recent ancient blocks and won't do ancient related checks on startup (default = 0, means keep all)", + Value: 0, + } MinFreeDiskSpaceFlag = DirectoryFlag{ Name: "datadir.minfreedisk", Usage: "Minimum free disk space in MB, once reached triggers auto shut down (default = --cache.gc converted to MB, 0 = disabled)", @@ -849,6 +854,7 @@ var ( DatabasePathFlags = []cli.Flag{ DataDirFlag, AncientFlag, + AncientRecentLimitFlag, RemoteDBFlag, } ) @@ -1589,6 +1595,10 @@ func SetEthConfig(ctx *cli.Context, stack *node.Node, cfg *ethconfig.Config) { ctx.GlobalSet(TxLookupLimitFlag.Name, "0") log.Warn("Disable transaction unindexing for archive node") } + if ctx.GlobalString(GCModeFlag.Name) == "archive" && ctx.GlobalUint64(AncientRecentLimitFlag.Name) != 0 { + ctx.GlobalSet(AncientRecentLimitFlag.Name, "0") + log.Warn("Disable ancient prunning for archive node") + } if ctx.GlobalIsSet(LightServeFlag.Name) && ctx.GlobalUint64(TxLookupLimitFlag.Name) != 0 { log.Warn("LES server cannot serve old transaction status and cannot connect below les/4 protocol version if transaction lookup index is limited") } @@ -1637,7 +1647,6 @@ func SetEthConfig(ctx *cli.Context, stack *node.Node, cfg *ethconfig.Config) { if ctx.GlobalIsSet(AncientFlag.Name) { cfg.DatabaseFreezer = ctx.GlobalString(AncientFlag.Name) } - if gcmode := ctx.GlobalString(GCModeFlag.Name); gcmode != "full" && gcmode != "archive" { Fatalf("--%s must be either 'full' or 'archive'", GCModeFlag.Name) } @@ -1656,6 +1665,9 @@ func SetEthConfig(ctx *cli.Context, stack *node.Node, cfg *ethconfig.Config) { if ctx.GlobalIsSet(TxLookupLimitFlag.Name) { cfg.TxLookupLimit = ctx.GlobalUint64(TxLookupLimitFlag.Name) } + if ctx.GlobalIsSet(AncientRecentLimitFlag.Name) { + cfg.AncientRecentLimit = ctx.GlobalUint64(AncientRecentLimitFlag.Name) + } if ctx.GlobalIsSet(CacheFlag.Name) || ctx.GlobalIsSet(CacheTrieFlag.Name) { cfg.TrieCleanCache = ctx.GlobalInt(CacheFlag.Name) * ctx.GlobalInt(CacheTrieFlag.Name) / 100 } @@ -1979,7 +1991,7 @@ func MakeChainDatabase(ctx *cli.Context, stack *node.Node, readonly bool) ethdb. case ctx.GlobalString(SyncModeFlag.Name) == "light": chainDb, err = stack.OpenDatabase("lightchaindata", cache, handles, "", readonly) default: - chainDb, err = stack.OpenDatabaseWithFreezer("chaindata", cache, handles, ctx.GlobalString(AncientFlag.Name), "", readonly) + chainDb, err = stack.OpenDatabaseWithFreezer("chaindata", cache, handles, ctx.GlobalString(AncientFlag.Name), "", readonly, ctx.GlobalUint64(AncientRecentLimitFlag.Name)) } if err != nil { Fatalf("Could not open database: %v", err) diff --git a/core/blockchain.go b/core/blockchain.go index 2a58ed11fe..366782800d 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{}{} }() @@ -2289,6 +2314,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) rawdb.UnindexTransactions(bc.db, 0, head-bc.txLookupLimit+1, bc.quit) } return @@ -2313,13 +2339,61 @@ 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) 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(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 ( - 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 +2405,19 @@ 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) + 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/accessors_indexes.go b/core/rawdb/accessors_indexes.go index 25a44354bf..369996834b 100644 --- a/core/rawdb/accessors_indexes.go +++ b/core/rawdb/accessors_indexes.go @@ -18,6 +18,7 @@ package rawdb import ( "bytes" + "encoding/binary" "math/big" "github.com/ethereum/go-ethereum/common" @@ -141,6 +142,19 @@ func ReadReceipt(db ethdb.Reader, hash common.Hash, config *params.ChainConfig) return nil, common.Hash{}, 0, 0 } +// ReadStoredBloomSections reads the number of valid sections from the index database +func ReadStoredBloomSections(db ethdb.Database) uint64 { + var storedSections uint64 + + table := NewTable(db, string(BloomBitsIndexPrefix)) + data, _ := table.Get([]byte("count")) + if len(data) == 8 { + storedSections = binary.BigEndian.Uint64(data) + } + + return storedSections +} + // ReadBloomBits retrieves the compressed bloom bit vector belonging to the given // section and bit index from the. func ReadBloomBits(db ethdb.KeyValueReader, bit uint, section uint64, head common.Hash) ([]byte, error) { diff --git a/core/rawdb/chain_freezer.go b/core/rawdb/chain_freezer.go index 4c49db2748..6f602f7b51 100644 --- a/core/rawdb/chain_freezer.go +++ b/core/rawdb/chain_freezer.go @@ -37,6 +37,10 @@ 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. @@ -48,6 +52,10 @@ 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 @@ -55,16 +63,17 @@ 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) (*chainFreezer, error) { +func newChainFreezer(datadir string, namespace string, readonly bool, maxTableSize uint32, tables map[string]bool, ancientRecentLimit uint64) (*chainFreezer, error) { freezer, err := NewFreezer(datadir, namespace, readonly, maxTableSize, tables) if err != nil { return nil, err } return &chainFreezer{ - Freezer: freezer, - threshold: params.FullImmutabilityThreshold, - quit: make(chan struct{}), - trigger: make(chan chan struct{}), + Freezer: freezer, + ancientRecentLimit: ancientRecentLimit, + threshold: params.FullImmutabilityThreshold, + quit: make(chan struct{}), + trigger: make(chan chan struct{}), }, nil } diff --git a/core/rawdb/database.go b/core/rawdb/database.go index 2b870d16d4..6cb5f69f99 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) (ethdb.Database, error) { +func NewDatabaseWithFreezer(db ethdb.KeyValueStore, freezer string, namespace string, readonly bool, ancientRecentLimit uint64) (ethdb.Database, error) { // Create the idle freezer instance - frdb, err := newChainFreezer(freezer, namespace, readonly, freezerTableSize, FreezerNoSnappy) + frdb, err := newChainFreezer(freezer, namespace, readonly, freezerTableSize, FreezerNoSnappy, ancientRecentLimit) if err != nil { return nil, err } @@ -193,7 +193,8 @@ 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. - if kvgenesis, _ := db.Get(headerHashKey(0)); len(kvgenesis) > 0 { + // 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 // match, otherwise we might mix up freezers across chains and destroy both @@ -274,12 +275,12 @@ 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) + frdb, err := NewDatabaseWithFreezer(kvdb, freezer, namespace, readonly, ancientRecentLimit) if err != nil { kvdb.Close() return nil, err diff --git a/eth/backend.go b/eth/backend.go index 6bd87fbbc0..48d7d1c249 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) + chainDb, err := stack.OpenDatabaseWithFreezer("chaindata", config.DatabaseCache, config.DatabaseHandles, config.DatabaseFreezer, "eth/db/chaindata/", false, config.AncientRecentLimit) if err != nil { return nil, err } @@ -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) diff --git a/eth/ethconfig/config.go b/eth/ethconfig/config.go index b86f7a5012..c37faa1688 100644 --- a/eth/ethconfig/config.go +++ b/eth/ethconfig/config.go @@ -163,6 +163,7 @@ type Config struct { DatabaseHandles int `toml:"-"` DatabaseCache int DatabaseFreezer string + AncientRecentLimit uint64 TrieCleanCache int TrieCleanCacheJournal string `toml:",omitempty"` // Disk journal directory for trie cache to survive node restarts diff --git a/node/node.go b/node/node.go index d7b35af2fa..c67bbd01d8 100644 --- a/node/node.go +++ b/node/node.go @@ -717,7 +717,7 @@ func (n *Node) OpenDatabase(name string, cache, handles int, namespace string, r // also attaching a chain freezer to it that moves ancient chain data from the // database to immutable append-only files. If the node is an ephemeral one, a // memory database is returned. -func (n *Node) OpenDatabaseWithFreezer(name string, cache, handles int, freezer, namespace string, readonly bool) (ethdb.Database, error) { +func (n *Node) OpenDatabaseWithFreezer(name string, cache, handles int, freezer, namespace string, readonly bool, ancientRecentLimit uint64) (ethdb.Database, error) { n.lock.Lock() defer n.lock.Unlock() if n.state == closedState { @@ -736,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 {