feat: support to control memory usage for header map

ckb-bin/src/subcommand/run.rs

@@ -67,6 +67,12 @@ pub fn run(args: RunArgs, version: Version) -> Result<(), ExitCode> {
 
     let sync_shared = Arc::new(SyncShared::with_tmpdir(
         shared.clone(),
+        args.config
+            .network
+            .sync
+            .as_ref()
+            .cloned()
+            .unwrap_or_default(),
         args.config.tmp_dir.as_ref(),
     ));
     let network_state = Arc::new(

network/src/protocols/test.rs

@@ -116,6 +116,7 @@ fn net_service_start(name: String) -> Node {
         upnp: false,
         bootnode_mode: true,
         max_send_buffer: None,
+        sync: None,
     };
 
     let network_state =

rpc/src/test.rs

@@ -180,7 +180,7 @@ fn setup_node(height: u64) -> (Shared, ChainController, RpcServer) {
         .start(Some("rpc-test-network"))
         .expect("Start network service failed")
     };
-    let sync_shared = Arc::new(SyncShared::new(shared.clone()));
+    let sync_shared = Arc::new(SyncShared::new(shared.clone(), Default::default()));
     let synchronizer = Synchronizer::new(chain_controller.clone(), Arc::clone(&sync_shared));
     let indexer_store = {
         let mut indexer_config = IndexerConfig::default();

sync/Cargo.toml

@@ -10,6 +10,7 @@ ckb-chain = { path = "../chain" }
 ckb-shared = { path = "../shared" }
 ckb-store = { path = "../store" }
 ckb-db = { path = "../db" }
+ckb-app-config = {path = "../util/app-config"}
 ckb-types = {path = "../util/types"}
 ckb-network = { path = "../network" }
 ckb-logger = {path = "../util/logger"}

sync/src/relayer/tests/helper.rs

@@ -151,7 +151,7 @@ pub(crate) fn build_chain(tip: BlockNumber) -> (Relayer, OutPoint) {
             .expect("processing block should be ok");
     }
 
-    let sync_shared = Arc::new(SyncShared::new(shared));
+    let sync_shared = Arc::new(SyncShared::new(shared, Default::default()));
     (
         Relayer::new(
             chain_controller,

sync/src/synchronizer/mod.rs

@@ -739,7 +739,7 @@ mod tests {
     }
 
     fn gen_synchronizer(chain_controller: ChainController, shared: Shared) -> Synchronizer {
-        let shared = Arc::new(SyncShared::new(shared));
+        let shared = Arc::new(SyncShared::new(shared, Default::default()));
         Synchronizer::new(chain_controller, shared)
     }
 

sync/src/tests/sync_shared.rs

@@ -64,7 +64,10 @@ fn test_insert_parent_unknown_block() {
             let chain_service = ChainService::new(shared.clone(), table);
             chain_service.start::<&str>(None)
         };
-        (SyncShared::new(shared), chain_controller)
+        (
+            SyncShared::new(shared, Default::default()),
+            chain_controller,
+        )
     };
 
     let block = shared1

sync/src/tests/synchronizer.rs

@@ -166,7 +166,7 @@ fn setup_node(height: u64) -> (TestNode, Shared) {
             .expect("process block should be OK");
     }
 
-    let sync_shared = Arc::new(SyncShared::new(shared.clone()));
+    let sync_shared = Arc::new(SyncShared::new(shared.clone(), Default::default()));
     let synchronizer = Synchronizer::new(chain_controller, sync_shared);
     let mut node = TestNode::default();
     let protocol = Arc::new(RwLock::new(synchronizer)) as Arc<_>;

sync/src/tests/util.rs

@@ -29,7 +29,7 @@ pub fn build_chain(tip: BlockNumber) -> (SyncShared, ChainController) {
         chain_service.start::<&str>(None)
     };
     generate_blocks(&shared, &chain_controller, tip);
-    let sync_shared = SyncShared::new(shared);
+    let sync_shared = SyncShared::new(shared, Default::default());
     (sync_shared, chain_controller)
 }
 

sync/src/types/header_map/backend_rocksdb.rs

@@ -2,7 +2,7 @@ use std::path;
 
 use ckb_db::internal::{
     ops::{Delete as _, GetPinned as _, Open as _, Put as _},
-    DB,
+    BlockBasedOptions, Options, DB,
 };
 use ckb_logger::{debug, warn};
 use ckb_types::{packed::Byte32, prelude::*};
@@ -47,7 +47,20 @@ impl KeyValueBackend for RocksDBBackend {
                 builder.tempdir()
             };
             if let Ok(cache_dir) = cache_dir_res {
-                if let Ok(db) = DB::open_default(cache_dir.path()) {
+                // We minimize memory usage at all costs here.
+                // If we want to use more memory, we should increase the limit of KeyValueMemory.
+                let opts = {
+                    let mut block_opts = BlockBasedOptions::default();
+                    block_opts.disable_cache();
+                    let mut opts = Options::default();
+                    opts.create_if_missing(true);
+                    opts.set_block_based_table_factory(&block_opts);
+                    opts.set_write_buffer_size(4 * 1024 * 1024);
+                    opts.set_max_write_buffer_number(2);
+                    opts.set_min_write_buffer_number_to_merge(1);
+                    opts
+                };
+                if let Ok(db) = DB::open(&opts, cache_dir.path()) {
                     debug!(
                         "open a key-value database({}) to save header map into disk",
                         cache_dir.path().to_str().unwrap_or("")

sync/src/types/mod.rs

@@ -7,6 +7,7 @@ use crate::{
     MAX_TIP_AGE, NORMAL_INDEX, POW_INTERVAL, RETRY_ASK_TX_TIMEOUT_INCREASE, SUSPEND_SYNC_TIME,
     TIME_TRACE_SIZE,
 };
+use ckb_app_config::SyncConfig;
 use ckb_chain::chain::ChainController;
 use ckb_chain_spec::consensus::Consensus;
 use ckb_logger::{debug, debug_target, error, trace};
@@ -1174,11 +1175,11 @@ pub struct SyncShared {
 }
 
 impl SyncShared {
-    pub fn new(shared: Shared) -> SyncShared {
-        Self::with_tmpdir::<PathBuf>(shared, None)
+    pub fn new(shared: Shared, sync_config: SyncConfig) -> SyncShared {
+        Self::with_tmpdir::<PathBuf>(shared, sync_config, None)
     }
 
-    pub fn with_tmpdir<P>(shared: Shared, tmpdir: Option<P>) -> SyncShared
+    pub fn with_tmpdir<P>(shared: Shared, sync_config: SyncConfig, tmpdir: Option<P>) -> SyncShared
     where
         P: AsRef<Path>,
     {
@@ -1190,7 +1191,11 @@ impl SyncShared {
             )
         };
         let shared_best_header = RwLock::new(HeaderView::new(header, total_difficulty));
-        let header_map = HeaderMap::new(tmpdir, 300_000, 20_000);
+        let header_map = HeaderMap::new(
+            tmpdir,
+            sync_config.header_map.primary_limit,
+            sync_config.header_map.backend_close_threshold,
+        );
 
         let state = SyncState {
             n_sync_started: AtomicUsize::new(0),

test/src/net.rs

@@ -114,6 +114,7 @@ impl Net {
             upnp: false,
             bootnode_mode: false,
             max_send_buffer: None,
+            sync: None,
         };
 
         let network_state =

util/app-config/src/configs/mod.rs

@@ -16,7 +16,7 @@ pub use miner::{
     ClientConfig as MinerClientConfig, Config as MinerConfig, DummyConfig, EaglesongSimpleConfig,
     ExtraHashFunction, WorkerConfig as MinerWorkerConfig,
 };
-pub use network::Config as NetworkConfig;
+pub use network::{Config as NetworkConfig, HeaderMapConfig, SyncConfig};
 pub use network_alert::Config as NetworkAlertConfig;
 pub use notify::Config as NotifyConfig;
 pub use rpc::{Config as RpcConfig, Module as RpcModule};

util/app-config/src/configs/network.rs

@@ -37,6 +37,30 @@ pub struct Config {
     pub bootnode_mode: bool,
     // Max send buffer size
     pub max_send_buffer: Option<usize>,
+    pub sync: Option<SyncConfig>,
+}
+
+#[derive(Clone, Debug, Serialize, Deserialize, Default)]
+pub struct SyncConfig {
+    #[serde(default)]
+    pub header_map: HeaderMapConfig,
+}
+
+#[derive(Clone, Debug, Serialize, Deserialize)]
+pub struct HeaderMapConfig {
+    // The maximum size of data in memory
+    pub primary_limit: usize,
+    // Disable cache if the size of data in memory less than this threshold
+    pub backend_close_threshold: usize,
+}
+
+impl Default for HeaderMapConfig {
+    fn default() -> Self {
+        Self {
+            primary_limit: 300_000,
+            backend_close_threshold: 20_000,
+        }
+    }
 }
 
 pub(crate) fn generate_random_key() -> [u8; 32] {