support flag to control size of block index cache (#1453)

* support flag to control size of block index cache

* fix calls to NewBlockchain to include block index size

* update run.go to use DefaultBlockIndexSize

Author: lazynina

cmd/config.go

@@ -49,6 +49,9 @@ type Config struct {
 	DisableEncoderMigrations  bool
 	HypersyncMaxQueueSize     uint32
 
+	// Block Index Size
+	BlockIndexSize int
+
 	// PoS Validator
 	PosValidatorSeed string
 
@@ -140,6 +143,18 @@ func LoadConfig() *Config {
 	config.DisableEncoderMigrations = viper.GetBool("disable-encoder-migrations")
 	config.HypersyncMaxQueueSize = viper.GetUint32("hypersync-max-queue-size")
 
+	// Block Index Size
+	config.BlockIndexSize = viper.GetInt("block-index-size")
+	if config.BlockIndexSize <= 0 {
+		glog.Fatal("Block index size must be greater than 0")
+	}
+	if config.BlockIndexSize < lib.MinBlockIndexSize {
+		glog.Fatalf("Block index size must be greater than %d", lib.MinBlockIndexSize)
+	}
+	if config.BlockIndexSize > lib.MaxBlockIndexSize {
+		glog.Fatalf("Block index size must be less than %d", lib.MaxBlockIndexSize)
+	}
+
 	// PoS Validator
 	config.PosValidatorSeed = viper.GetString("pos-validator-seed")
 
@@ -270,6 +285,8 @@ func (config *Config) Print() {
 		glog.Infof("MaxSyncBlockHeight: %v", config.MaxSyncBlockHeight)
 	}
 
+	glog.Infof("BlockIndexSize: %v", config.BlockIndexSize)
+
 	if len(config.ConnectIPs) > 0 {
 		glog.Infof("Connect IPs: %s", config.ConnectIPs)
 	}

cmd/node.go

@@ -245,6 +245,7 @@ func (node *Node) Start(exitChannels ...*chan struct{}) {
 		node.Config.TransactionValidationRefreshIntervalMillis,
 		node.Config.StateSyncerMempoolTxnSyncLimit,
 		node.Config.CheckpointSyncingProviders,
+		node.Config.BlockIndexSize,
 	)
 	if err != nil {
 		// shouldRestart can be true if, on the previous run, we did not finish flushing all ancestral

cmd/run.go

@@ -86,6 +86,10 @@ func SetupRunFlags(cmd *cobra.Command) {
 		  is true.
 		- hypersync: Will sync by downloading historical state, and will NOT
 		  download historical blocks. Can only be set if HyperSync is true.`)
+	// Block Index Size
+	cmd.PersistentFlags().Uint64("block-index-size", lib.DefaultBlockIndexSize, "The number of blocks to keep in memory in the block "+
+		"index. Setting this to a lower number will reduce memory usage, but will require the node to go to disk "+
+		"more when fetching block nodes more often.")
 
 	// PoS Validator
 	cmd.PersistentFlags().String("pos-validator-seed", "", "A BIP39 seed phrase or seed hex used to generate the "+

lib/blockchain.go

@@ -553,24 +553,32 @@ type BlockIndex struct {
 	blockIndexByHash *collections.LruCache[BlockHash, *BlockNode]
 	tip              *BlockNode
 	headerTip        *BlockNode
+	size             int
 }
 
 // NewBlockIndex creates a new BlockIndex with the provided snapshot and tip node.
-func NewBlockIndex(db *badger.DB, snapshot *Snapshot, tipNode *BlockNode) *BlockIndex {
-	blockIndexByHash, _ := collections.NewLruCache[BlockHash, *BlockNode](MaxBlockIndexNodes) // TODO: parameterize this?
+func NewBlockIndex(db *badger.DB, snapshot *Snapshot, tipNode *BlockNode, blockIndexSize int) (*BlockIndex, error) {
+	blockIndexByHash, err := collections.NewLruCache[BlockHash, *BlockNode](blockIndexSize)
+	if err != nil {
+		return nil, errors.Wrapf(err, "NewBlockIndex: Problem creating LRU cache")
+	}
 	return &BlockIndex{
 		db:               db,
 		snapshot:         snapshot,
 		blockIndexByHash: blockIndexByHash,
 		tip:              tipNode,
 		headerTip:        tipNode,
-	}
+		size:             blockIndexSize,
+	}, nil
 }
 
 // setBlockIndexFromMap is a helper function only used in tests. It constructs the
 // block index from the provided map of block hashes to block nodes.
-func (bi *BlockIndex) setBlockIndexFromMap(input map[BlockHash]*BlockNode) {
-	newHashToBlockNodeMap, _ := collections.NewLruCache[BlockHash, *BlockNode](MaxBlockIndexNodes) // TODO: parameterize this?
+func (bi *BlockIndex) setBlockIndexFromMap(input map[BlockHash]*BlockNode) error {
+	newHashToBlockNodeMap, err := collections.NewLruCache[BlockHash, *BlockNode](bi.size)
+	if err != nil {
+		return errors.Wrapf(err, "setBlockIndexFromMap: Problem creating LRU cache")
+	}
 	bi.blockIndexByHash = newHashToBlockNodeMap
 	for _, val := range input {
 		bi.addNewBlockNodeToBlockIndex(val)
@@ -582,6 +590,7 @@ func (bi *BlockIndex) setBlockIndexFromMap(input map[BlockHash]*BlockNode) {
 			bi.tip = val
 		}
 	}
+	return nil
 }
 
 // setHeaderTip sets the header tip of the block index to the provided block node.
@@ -879,7 +888,7 @@ func (bc *Blockchain) CopyBlockIndexes() (
 	_blockIndexByHash *collections.LruCache[BlockHash, *BlockNode],
 ) {
 	// Create a new lru cache.
-	newBlockIndexByHash, _ := collections.NewLruCache[BlockHash, *BlockNode](MaxBlockIndexNodes)
+	newBlockIndexByHash, _ := collections.NewLruCache[BlockHash, *BlockNode](bc.blockIndex.size)
 	// Iterate over the keys of the block index and copy them to the new lru cache.
 	for _, key := range bc.blockIndex.blockIndexByHash.Keys() {
 		val, _ := bc.blockIndex.blockIndexByHash.Get(key)
@@ -1039,10 +1048,15 @@ func (bc *Blockchain) _initChain() error {
 			if !tipNodeExists {
 				return fmt.Errorf("_initChain: Best hash (%#v) not found in block index", bestBlockHash)
 			}
-			// Walk back the last 6 hours of blocks.
+			// Walk back the last 6 hours of blocks. Or enough to fill the block index cache if the
+			// size of the block index cache is less than 6 * 3600
 			currBlockCounter := 1
 			parentNode := tipNode.GetParent(bc.blockIndex)
-			for currBlockCounter < 3600*6 && parentNode != nil {
+			numBlockNodesToFetch := 3600 * 6
+			if bc.blockIndex.size < numBlockNodesToFetch {
+				numBlockNodesToFetch = bc.blockIndex.size
+			}
+			for currBlockCounter < numBlockNodesToFetch && parentNode != nil {
 				parentNode = parentNode.GetParent(bc.blockIndex)
 				currBlockCounter++
 			}
@@ -1119,6 +1133,7 @@ func NewBlockchain(
 	snapshot *Snapshot,
 	archivalMode bool,
 	checkpointSyncingProviders []string,
+	blockIndexSize int,
 ) (*Blockchain, error) {
 	if err := RunBlockIndexMigrationOnce(db, params); err != nil {
 		return nil, errors.Wrapf(err, "NewBlockchain: Problem running block index migration")
@@ -1137,6 +1152,10 @@ func NewBlockchain(
 	timer := &Timer{}
 	timer.Initialize()
 	blockViewCache, _ := collections.NewLruCache[BlockHash, *BlockViewAndUtxoOps](100) // TODO: parameterize
+	blockIndex, err := NewBlockIndex(db, snapshot, nil, blockIndexSize)                // This tip will be set in _initChain.
+	if err != nil {
+		return nil, errors.Wrapf(err, "NewBlockchain: Problem creating block index")
+	}
 	bc := &Blockchain{
 		db:                              db,
 		postgres:                        postgres,
@@ -1149,7 +1168,7 @@ func NewBlockchain(
 		eventManager:                    eventManager,
 		archivalMode:                    archivalMode,
 
-		blockIndex:     NewBlockIndex(db, snapshot, nil), // This tip will be set in _initChain.
+		blockIndex:     blockIndex,
 		blockViewCache: blockViewCache,
 		snapshotCache:  NewSnapshotCache(),
 

lib/blockchain_test.go

@@ -153,7 +153,7 @@ func NewTestBlockchain(t *testing.T) (*Blockchain, *DeSoParams, *badger.DB) {
 	paramsCopy := DeSoTestnetParams
 
 	chain, err := NewBlockchain([]string{blockSignerPk}, 0, 0, &paramsCopy,
-		timesource, db, nil, nil, nil, false, nil)
+		timesource, db, nil, nil, nil, false, nil, MinBlockIndexSize)
 	if err != nil {
 		log.Fatal(err)
 	}
@@ -255,7 +255,7 @@ func NewLowDifficultyBlockchainWithParamsAndDb(t *testing.T, params *DeSoParams,
 		}
 	}
 	chain, err := NewBlockchain([]string{blockSignerPk}, 0, 0,
-		&testParams, timesource, db, postgresDb, NewEventManager(), snap, false, nil)
+		&testParams, timesource, db, postgresDb, NewEventManager(), snap, false, nil, MinBlockIndexSize)
 	if err != nil {
 		log.Fatal(err)
 	}

lib/constants.go

@@ -1962,3 +1962,7 @@ const BlockIndexMigrationFileName = "block_index_migration.txt"
 
 // BtcecPubKeyBytesLenUncompressed is a constant that was removed from newer version of Btcec
 const BtcecPubKeyBytesLenUncompressed = 65
+
+const DefaultBlockIndexSize = MaxBlockIndexNodes
+const MaxBlockIndexSize = 10000000
+const MinBlockIndexSize = 10000

lib/db_utils_test.go

@@ -195,7 +195,8 @@ func TestBlockNodePutGet(t *testing.T) {
 
 	// Check that getting the best chain works.
 	{
-		bi := NewBlockIndex(db, nil, b4)
+		bi, err := NewBlockIndex(db, nil, b4, 1000)
+		require.NoError(err)
 		bestChain, err := GetBestChain(b3Ret, bi)
 		require.NoError(err)
 		require.Len(bestChain, 3)
@@ -229,7 +230,8 @@ func TestInitDbWithGenesisBlock(t *testing.T) {
 	require.Equal(&genesisHash, genesis.Hash)
 
 	// Check the bestChain.
-	bi := NewBlockIndex(db, nil, genesis)
+	bi, err := NewBlockIndex(db, nil, genesis, 1000)
+	require.NoError(err)
 	bestChain, err := GetBestChain(genesis, bi)
 	require.NoError(err)
 	require.Len(bestChain, 1)

lib/pos_blockchain_test.go

@@ -252,7 +252,7 @@ func TestHasValidBlockHeight(t *testing.T) {
 		ValidatorsVoteQC:             nil,
 		ValidatorsTimeoutAggregateQC: nil,
 	}, StatusBlockStored|StatusBlockValidated)
-	bc.blockIndex.setBlockIndexFromMap(map[BlockHash]*BlockNode{*genesisBlock.Hash: genesisBlock})
+	require.NoError(t, bc.blockIndex.setBlockIndexFromMap(map[BlockHash]*BlockNode{*genesisBlock.Hash: genesisBlock}))
 	bc.blockIndex.blockIndexByHash.Put(*genesisBlock.Hash, genesisBlock)
 	// Create a block with a valid header.
 	randomPayload := RandomBytes(256)
@@ -303,7 +303,8 @@ func TestHasValidBlockHeight(t *testing.T) {
 
 	block.Header.Height = 2
 	// TODO: make sure setting to genesis block works.
-	bc.blockIndex = NewBlockIndex(bc.db, bc.snapshot, genesisBlock)
+	bc.blockIndex, err = NewBlockIndex(bc.db, bc.snapshot, genesisBlock, 1000)
+	require.NoError(t, err)
 	err = bc.hasValidBlockHeightPoS(block.Header)
 	require.Equal(t, err, RuleErrorMissingParentBlock)
 }
@@ -331,10 +332,10 @@ func TestUpsertBlockAndBlockNodeToDB(t *testing.T) {
 		ValidatorsVoteQC:             nil,
 		ValidatorsTimeoutAggregateQC: nil,
 	}, StatusBlockStored|StatusBlockValidated)
-	bc.blockIndex.setBlockIndexFromMap(map[BlockHash]*BlockNode{
+	require.NoError(t, bc.blockIndex.setBlockIndexFromMap(map[BlockHash]*BlockNode{
 		*hash1: genesisNode,
 		*hash2: block2,
-	})
+	}))
 	randomPayload := RandomBytes(256)
 	randomBLSPrivateKey := _generateRandomBLSPrivateKey(t)
 	signature, err := randomBLSPrivateKey.Sign(randomPayload)
@@ -470,10 +471,10 @@ func TestHasValidBlockViewPoS(t *testing.T) {
 		ValidatorsVoteQC:             nil,
 		ValidatorsTimeoutAggregateQC: nil,
 	}, StatusBlockStored|StatusBlockValidated)
-	bc.blockIndex.setBlockIndexFromMap(map[BlockHash]*BlockNode{
+	require.NoError(t, bc.blockIndex.setBlockIndexFromMap(map[BlockHash]*BlockNode{
 		*hash1: genesisNode,
 		*hash2: block2,
-	})
+	}))
 	randomPayload := RandomBytes(256)
 	randomBLSPrivateKey := _generateRandomBLSPrivateKey(t)
 	signature, err := randomBLSPrivateKey.Sign(randomPayload)
@@ -802,9 +803,9 @@ func TestGetLineageFromCommittedTip(t *testing.T) {
 		Height:         1,
 		ProposedInView: 1,
 	}, StatusBlockStored|StatusBlockValidated|StatusBlockCommitted)
-	bc.blockIndex.setBlockIndexFromMap(map[BlockHash]*BlockNode{
+	require.NoError(t, bc.blockIndex.setBlockIndexFromMap(map[BlockHash]*BlockNode{
 		*hash1: genesisNode,
-	})
+	}))
 	block := &MsgDeSoBlock{
 		Header: &MsgDeSoHeader{
 			Version:        HeaderVersion2,
@@ -1241,10 +1242,10 @@ func TestShouldReorg(t *testing.T) {
 			Height: 1,
 		},
 	}
-	bc.blockIndex.setBlockIndexFromMap(map[BlockHash]*BlockNode{
+	require.NoError(t, bc.blockIndex.setBlockIndexFromMap(map[BlockHash]*BlockNode{
 		*hash1: chain[0],
 		*hash3: chain[1],
-	})
+	}))
 
 	newBlock := &BlockNode{
 		Header: &MsgDeSoHeader{

lib/pos_consensus_test.go

@@ -102,10 +102,11 @@ func TestFastHotStuffConsensusHandleLocalTimeoutEvent(t *testing.T) {
 	currentView := blockHeader.ValidatorsVoteQC.GetView() + 1
 	nextView := currentView + 1
 
-	blockIndex := NewBlockIndex(nil, nil, nil)
-	blockIndex.setBlockIndexFromMap(map[BlockHash]*BlockNode{
+	blockIndex, err := NewBlockIndex(nil, nil, nil, 1000)
+	require.NoError(t, err)
+	require.NoError(t, blockIndex.setBlockIndexFromMap(map[BlockHash]*BlockNode{
 		*blockHash: {Header: blockHeader, Height: uint32(blockHeader.Height), Hash: blockHash},
-	})
+	}))
 
 	// Create a mock consensus
 	fastHotStuffConsensus := FastHotStuffConsensus{

lib/server.go

@@ -452,6 +452,7 @@ func NewServer(
 	_transactionValidationRefreshIntervalMillis uint64,
 	_stateSyncerMempoolTxnSyncLimit uint64,
 	_checkpointSyncingProviders []string,
+	_blockIndexSize int,
 ) (
 	_srv *Server,
 	_err error,
@@ -548,7 +549,8 @@ func NewServer(
 
 	_chain, err := NewBlockchain(
 		_trustedBlockProducerPublicKeys, _trustedBlockProducerStartHeight, _maxSyncBlockHeight,
-		_params, timesource, _db, postgres, eventManager, _snapshot, archivalMode, _checkpointSyncingProviders)
+		_params, timesource, _db, postgres, eventManager, _snapshot, archivalMode, _checkpointSyncingProviders,
+		_blockIndexSize)
 	if err != nil {
 		return nil, errors.Wrapf(err, "NewServer: Problem initializing blockchain"), true
 	}

lib/txindex.go

@@ -130,7 +130,7 @@ func NewTXIndex(coreChain *Blockchain, params *DeSoParams, dataDirectory string)
 	// Note that we *DONT* pass server here because it is already tied to the main blockchain.
 	txIndexChain, err := NewBlockchain(
 		[]string{}, 0, coreChain.MaxSyncBlockHeight, params, chainlib.NewMedianTime(),
-		txIndexDb, nil, nil, nil, false, nil)
+		txIndexDb, nil, nil, nil, false, nil, coreChain.blockIndex.size)
 	if err != nil {
 		return nil, fmt.Errorf("NewTXIndex: Error initializing TxIndex: %v", err)
 	}