raftstore: make threads of snap-generator pool configurable (tikv#11274)

* add snap-generator-pool-size for configurable snapshot generator thread pool

Signed-off-by: iosmanthus <[email protected]>

* rename config macro

Signed-off-by: iosmanthus <[email protected]>

Co-authored-by: Andy Lok <[email protected]>

components/raftstore/src/store/config.rs

@@ -164,6 +164,9 @@ pub struct Config {
     #[online_config(hidden)]
     pub use_delete_range: bool,
 
+    #[online_config(skip)]
+    pub snap_generator_pool_size: usize,
+
     pub cleanup_import_sst_interval: ReadableDuration,
 
     /// Maximum size of every local read task batch.
@@ -305,6 +308,7 @@ impl Default for Config {
             merge_max_log_gap: 10,
             merge_check_tick_interval: ReadableDuration::secs(2),
             use_delete_range: false,
+            snap_generator_pool_size: 2,
             cleanup_import_sst_interval: ReadableDuration::minutes(10),
             local_read_batch_size: 1024,
             apply_batch_system: BatchSystemConfig::default(),
@@ -536,6 +540,12 @@ impl Config {
             ));
         }
 
+        if self.snap_generator_pool_size == 0 {
+            return Err(box_err!(
+                "snap-generator-pool-size should be greater than 0."
+            ));
+        }
+
         Ok(())
     }
 
@@ -711,6 +721,9 @@ impl Config {
         CONFIG_RAFTSTORE_GAUGE
             .with_label_values(&["future_poll_size"])
             .set(self.future_poll_size as f64);
+        CONFIG_RAFTSTORE_GAUGE
+            .with_label_values(&["snap_generator_pool_size"])
+            .set(self.snap_generator_pool_size as f64);
         CONFIG_RAFTSTORE_GAUGE
             .with_label_values(&["hibernate_regions"])
             .set((self.hibernate_regions as i32).into());
@@ -930,6 +943,10 @@ mod tests {
         cfg.future_poll_size = 0;
         assert!(cfg.validate().is_err());
 
+        cfg = Config::new();
+        cfg.snap_generator_pool_size = 0;
+        assert!(cfg.validate().is_err());
+
         cfg = Config::new();
         cfg.raft_base_tick_interval = ReadableDuration::secs(1);
         cfg.raft_election_timeout_ticks = 11;

components/raftstore/src/store/fsm/store.rs

@@ -1321,6 +1321,7 @@ impl<EK: KvEngine, ER: RaftEngine> RaftBatchSystem<EK, ER> {
             mgr.clone(),
             cfg.value().snap_apply_batch_size.0 as usize,
             cfg.value().use_delete_range,
+            cfg.value().snap_generator_pool_size,
             workers.coprocessor_host.clone(),
             self.router(),
         );

components/raftstore/src/store/peer_storage.rs

@@ -2105,6 +2105,7 @@ mod tests {
             mgr,
             0,
             true,
+            2,
             CoprocessorHost::<KvTestEngine>::default(),
             router,
         );
@@ -2482,6 +2483,7 @@ mod tests {
             mgr,
             0,
             true,
+            2,
             CoprocessorHost::<KvTestEngine>::default(),
             router,
         );

components/raftstore/src/store/worker/region.rs

@@ -35,8 +35,6 @@ use tikv_util::worker::{Runnable, RunnableWithTimer};
 
 use super::metrics::*;
 
-const GENERATE_POOL_SIZE: usize = 2;
-
 // used to periodically check whether we should delete a stale peer's range in region runner
 
 #[cfg(test)]
@@ -625,12 +623,13 @@ where
         mgr: SnapManager,
         batch_size: usize,
         use_delete_range: bool,
+        snap_generator_pool_size: usize,
         coprocessor_host: CoprocessorHost<EK>,
         router: R,
     ) -> Runner<EK, R> {
         Runner {
             pool: Builder::new(thd_name!("snap-generator"))
-                .max_thread_count(GENERATE_POOL_SIZE)
+                .max_thread_count(snap_generator_pool_size)
                 .build_future_pool(),
             ctx: SnapContext {
                 engine,
@@ -873,6 +872,7 @@ mod tests {
             mgr,
             0,
             false,
+            2,
             CoprocessorHost::<KvTestEngine>::default(),
             router,
         );
@@ -977,6 +977,7 @@ mod tests {
             mgr,
             0,
             true,
+            2,
             CoprocessorHost::<KvTestEngine>::default(),
             router,
         );

etc/config-template.toml

@@ -425,6 +425,9 @@
 ## When the size of raft db writebatch exceeds this value, write will be triggered.
 # raft-write-size-limit = "1MB"
 
+## threads to generate raft snapshots
+# snap-generator-pool-size = 2
+
 [coprocessor]
 ## When it is set to `true`, TiKV will try to split a Region with table prefix if that Region
 ## crosses tables.

tests/integrations/config/mod.rs

@@ -202,6 +202,7 @@ fn test_serde_custom_tikv_config() {
         merge_max_log_gap: 3,
         merge_check_tick_interval: ReadableDuration::secs(11),
         use_delete_range: true,
+        snap_generator_pool_size: 2,
         cleanup_import_sst_interval: ReadableDuration::minutes(12),
         region_max_size: ReadableSize(0),
         region_split_size: ReadableSize(0),