perf: Make kura drop old blocks from memory (#5103)

Signed-off-by: Dmitry Murzin <[email protected]>

crates/iroha_config/src/parameters/actual.rs

@@ -101,6 +101,7 @@ pub struct Queue {
 pub struct Kura {
     pub init_mode: InitMode,
     pub store_dir: WithOrigin<PathBuf>,
+    pub max_blocks_in_memory: NonZeroUsize,
     pub debug_output_new_blocks: bool,
 }
 

crates/iroha_config/src/parameters/defaults.rs

@@ -20,7 +20,12 @@ pub mod queue {
 }
 
 pub mod kura {
+    use std::num::NonZeroUsize;
+
+    use nonzero_ext::nonzero;
+
     pub const STORE_DIR: &str = "./storage";
+    pub const MAX_BLOCKS_IN_MEMORY: NonZeroUsize = nonzero!(128_usize);
 }
 
 pub mod network {

crates/iroha_config/src/parameters/user.rs

@@ -181,6 +181,11 @@ pub struct Kura {
         default = "PathBuf::from(defaults::kura::STORE_DIR)"
     )]
     pub store_dir: WithOrigin<PathBuf>,
+    #[config(
+        env = "KURA_MAX_BLOCKS_IN_MEMORY",
+        default = "defaults::kura::MAX_BLOCKS_IN_MEMORY"
+    )]
+    pub max_blocks_in_memory: NonZeroUsize,
     #[config(nested)]
     pub debug: KuraDebug,
 }
@@ -190,6 +195,7 @@ impl Kura {
         let Self {
             init_mode,
             store_dir,
+            max_blocks_in_memory,
             debug:
                 KuraDebug {
                     output_new_blocks: debug_output_new_blocks,
@@ -199,6 +205,7 @@ impl Kura {
         actual::Kura {
             init_mode,
             store_dir,
+            max_blocks_in_memory,
             debug_output_new_blocks,
         }
     }

crates/iroha_config/tests/fixtures.rs

@@ -127,6 +127,7 @@ fn minimal_config_snapshot() {
                         id: ParameterId(kura.store_dir),
                     },
                 },
+                max_blocks_in_memory: 128,
                 debug_output_new_blocks: false,
             },
             sumeragi: Sumeragi {

crates/iroha_config/tests/fixtures/full.env

@@ -7,6 +7,7 @@ GENESIS=./genesis.signed.scale
 API_ADDRESS=127.0.0.1:8080
 KURA_INIT_MODE=strict
 KURA_STORE_DIR=/store/path/from/env
+KURA_MAX_BLOCKS_IN_MEMORY=128
 KURA_DEBUG_OUTPUT_NEW_BLOCKS=false
 LOG_LEVEL=DEBUG
 LOG_FORMAT=pretty

crates/iroha_core/benches/kura.rs

@@ -2,7 +2,10 @@
 
 use byte_unit::{Byte, UnitType};
 use criterion::{criterion_group, criterion_main, Criterion};
-use iroha_config::{base::WithOrigin, parameters::actual::Kura as Config};
+use iroha_config::{
+    base::WithOrigin,
+    parameters::{actual::Kura as Config, defaults::kura::MAX_BLOCKS_IN_MEMORY},
+};
 use iroha_core::{
     block::*,
     kura::BlockStore,
@@ -22,6 +25,7 @@ async fn measure_block_size_for_n_executors(n_executors: u32) {
     let cfg = Config {
         init_mode: iroha_config::kura::InitMode::Strict,
         debug_output_new_blocks: false,
+        max_blocks_in_memory: MAX_BLOCKS_IN_MEMORY,
         store_dir: WithOrigin::inline(dir.path().to_path_buf()),
     };
     let chain_id = ChainId::from("00000000-0000-0000-0000-000000000000");

crates/iroha_core/src/kura.rs

@@ -11,7 +11,10 @@ use std::{
     time::Duration,
 };
 
-use iroha_config::{kura::InitMode, parameters::actual::Kura as Config};
+use iroha_config::{
+    kura::InitMode,
+    parameters::{actual::Kura as Config, defaults::kura::MAX_BLOCKS_IN_MEMORY},
+};
 use iroha_crypto::{Hash, HashOf};
 use iroha_data_model::block::{BlockHeader, SignedBlock};
 use iroha_futures::supervisor::{spawn_os_thread_as_future, Child, OnShutdown, ShutdownSignal};
@@ -37,6 +40,9 @@ pub struct Kura {
     block_data: Mutex<BlockData>,
     /// Path to file for plain text blocks.
     block_plain_text_path: Option<PathBuf>,
+    /// At most N last blocks will be stored in memory.
+    /// Older blocks will be dropped from memory and loaded from the disk if they are needed.
+    max_blocks_in_memory: NonZeroUsize,
     /// Amount of blocks loaded during initialization
     init_block_count: usize,
 }
@@ -68,6 +74,7 @@ impl Kura {
             block_store: Mutex::new(block_store),
             block_data: Mutex::new(block_data),
             block_plain_text_path,
+            max_blocks_in_memory: config.max_blocks_in_memory,
             init_block_count: block_count,
         });
 
@@ -81,6 +88,7 @@ impl Kura {
             block_store: Mutex::new(BlockStore::new(PathBuf::new())),
             block_data: Mutex::new(Vec::new()),
             block_plain_text_path: None,
+            max_blocks_in_memory: MAX_BLOCKS_IN_MEMORY,
             init_block_count: 0,
         })
     }
@@ -201,7 +209,7 @@ impl Kura {
                 should_exit = true;
             }
 
-            let block_data = kura.block_data.lock();
+            let mut block_data = kura.block_data.lock();
 
             let new_latest_written_block_hash = written_block_count
                 .checked_sub(1)
@@ -231,6 +239,11 @@ impl Kura {
                     "INTERNAL BUG: The block to be written is None. Check store_block function.",
                 );
                 blocks_to_be_written.push(Arc::clone(block_ref));
+                Self::drop_old_block(
+                    &mut block_data,
+                    written_block_count,
+                    kura.max_blocks_in_memory.get(),
+                );
                 written_block_count += 1;
             }
 
@@ -318,7 +331,10 @@ impl Kura {
             .expect("INTERNAL BUG: Failed to decode block");
 
         let block_arc = Arc::new(block);
-        data_array_guard[block_index].1 = Some(Arc::clone(&block_arc));
+        // Only last N blocks should be kept in memory
+        if block_index + self.max_blocks_in_memory.get() >= data_array_guard.len() {
+            data_array_guard[block_index].1 = Some(Arc::clone(&block_arc));
+        }
         Some(block_arc)
     }
 
@@ -335,6 +351,20 @@ impl Kura {
         data.pop();
         data.push((block.hash(), Some(block)));
     }
+
+    // Drop old block to prevent unbounded memory usage.
+    // It will be loaded from the disk if needed later.
+    fn drop_old_block(
+        block_data: &mut BlockData,
+        written_block_count: usize,
+        max_blocks_in_memory: usize,
+    ) {
+        // Keep last N blocks and genesis block.
+        // (genesis block is used in metrics to get genesis timestamp)
+        if written_block_count > max_blocks_in_memory {
+            block_data[written_block_count - max_blocks_in_memory].1 = None;
+        }
+    }
 }
 
 /// Loaded block count
@@ -780,6 +810,7 @@ impl<T> AddErrContextExt<T> for Result<T, std::io::Error> {
 mod tests {
     use std::{str::FromStr, thread, time::Duration};
 
+    use iroha_config::parameters::defaults::kura::MAX_BLOCKS_IN_MEMORY;
     use iroha_crypto::KeyPair;
     use iroha_data_model::{
         account::Account,
@@ -978,6 +1009,7 @@ mod tests {
             store_dir: iroha_config::base::WithOrigin::inline(
                 temp_dir.path().to_str().unwrap().into(),
             ),
+            max_blocks_in_memory: MAX_BLOCKS_IN_MEMORY,
             debug_output_new_blocks: false,
         })
         .unwrap();
@@ -1007,6 +1039,7 @@ mod tests {
                 store_dir: iroha_config::base::WithOrigin::inline(
                     temp_dir.path().to_str().unwrap().into(),
                 ),
+                max_blocks_in_memory: MAX_BLOCKS_IN_MEMORY,
                 debug_output_new_blocks: false,
             })
             .unwrap();
@@ -1057,6 +1090,7 @@ mod tests {
             store_dir: iroha_config::base::WithOrigin::inline(
                 temp_dir.path().to_str().unwrap().into(),
             ),
+            max_blocks_in_memory: MAX_BLOCKS_IN_MEMORY,
             debug_output_new_blocks: false,
         })
         .unwrap();