Merge pull request #30 from DeBankDeFi/ancient-db-prunning

Initial implement of the ancient db prunning

.gitignore

@@ -47,3 +47,6 @@ profile.cov
 /dashboard/assets/package-lock.json
 
 **/yarn-error.log
+
+# vim swap files
+*.swp

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)
 		}

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)

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)
+		// 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 {
+			if doneTx == nil && donePr == nil {
-				done = make(chan struct{})
+				doneTx = make(chan struct{})
-				go indexBlocks(rawdb.ReadTxIndexTail(bc.db), head.Block.NumberU64(), done)
+				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:
+		case <-donePr:
-			done = nil
+			donePr = nil
+			doneTx = nil
 		case <-bc.quit:
-			if done != nil {
+			if donePr != nil {
-				log.Info("Waiting background transaction indexer to exit")
+				log.Info("Waiting background transaction indexer and ancient pruner to exit")
-				<-done
+				<-donePr
 			}
 			return
 		}

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) {

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,13 +63,14 @@ 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,
+		ancientRecentLimit: ancientRecentLimit,
 		threshold:          params.FullImmutabilityThreshold,
 		quit:               make(chan struct{}),
 		trigger:            make(chan chan struct{}),

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

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)

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

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 {