Repository: tikv/tikv (language: rust)

Issue:
br: The ResizableRuntime make some unit test taking too long to finish
## Bug Report

<!-- Thanks for your bug report! Don't worry if you can't fill out all the sections. -->

### What version of TiKV are you using?
<!-- You can run `tikv-server --version` -->

master

### What operating system and CPU are you using?
<!-- If you're using Linux, you can run `cat /proc/cpuinfo` -->

### Steps to reproduce
<!-- If possible, provide a recipe for reproducing the error. A complete runnable program is good. -->

Run unit test `test_txn_api_version::test_raftstore_v2_must_new_and_configure_cluster`.
It should finish in 20s, but it actually takes more than 120s.

### What did you expect?

The unit tests could finish faster.

### What did happened?

It takes too long time to finish.



Discussion / hints:
@BornChanger: The label(s) `serity/major` cannot be applied. These labels are supported: `challenge-program, compatibility-breaker, high-performance, hptc, wontfix, do-not-merge/cherry-pick-not-approved, needs-cherry-pick-release-5.4, needs-cherry-pick-release-6.1, needs-cherry-pick-release-6.5, needs-cherry-pick-release-7.1, needs-cherry-pick-release-7.5, needs-cherry-pick-release-8.1, needs-cherry-pick-release-8.5, affects-5.4, affects-6.1, affects-6.5, affects-7.1, affects-7.5, affects-8.1, affects-8.4, affects-8.5, may-affects-5.4, may-affects-6.1, may-affects-6.5, may-affects-7.1, may-affects-7.5, may-affects-8.1, may-affects-8.5`.

<details>

In response to [this](https://github.com/tikv/tikv/issues/17807#issuecomment-2475895066):

>/label serity/major


Instructions for interacting with me using PR comments are available [here](https://prow.tidb.net/command-help).  If you have questions or suggestions related to my behavior, please file an issue against the [ti-community-infra/tichi](https://github.com/ti-community-infra/tichi/issues/new?title=Prow%20issue:) repository.
</details>
@BornChanger: The label(s) `serity/major` cannot be applied. These labels are supported: `challenge-program, compatibility-breaker, high-performance, hptc, wontfix, do-not-merge/cherry-pick-not-approved, needs-cherry-pick-release-5.4, needs-cherry-pick-release-6.1, needs-cherry-pick-release-6.5, needs-cherry-pick-release-7.1, needs-cherry-pick-release-7.5, needs-cherry-pick-release-8.1, needs-cherry-pick-release-8.5, affects-5.4, affects-6.1, affects-6.5, affects-7.1, affects-7.5, affects-8.1, affects-8.4, affects-8.5, may-affects-5.4, may-affects-6.1, may-affects-6.5, may-affects-7.1, may-affects-7.5, may-affects-8.1, may-affects-8.5`.

<details>

In response to [this](https://github.com/tikv/tikv/issues/17807#issuecomment-2475895066):

>/label serity/major


Instructions for interacting with me using PR comments are available [here](https://prow.tidb.net/command-help).  If you have questions or suggestions related to my behavior, please file an issue against the [ti-community-infra/tichi](https://github.com/ti-community-infra/tichi/issues/new?title=Prow%20issue:) repository.
</details>

Fix patch:
diff --git a/Cargo.lock b/Cargo.lock
index 31b1787fb1b..f5400b4358f 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -3216,7 +3216,7 @@ dependencies = [
  "httpdate",
  "itoa 1.0.1",
  "pin-project-lite",
- "socket2",
+ "socket2 0.4.7",
  "tokio",
  "tower-service",
  "tracing",
@@ -4034,13 +4034,14 @@ dependencies = [
 
 [[package]]
 name = "mio"
-version = "0.8.11"
+version = "1.0.2"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "a4a650543ca06a924e8b371db273b2756685faae30f8487da1b56505a8f78b0c"
+checksum = "80e04d1dcff3aae0704555fe5fee3bcfaf3d1fdf8a7e521d5b9d2b42acb52cec"
 dependencies = [
+ "hermit-abi 0.3.9",
  "libc 0.2.151",
  "wasi 0.11.0+wasi-snapshot-preview1",
- "windows-sys 0.48.0",
+ "windows-sys 0.52.0",
 ]
 
 [[package]]
@@ -4759,9 +4760,9 @@ dependencies = [
 
 [[package]]
 name = "pin-project-lite"
-version = "0.2.9"
+version = "0.2.15"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "e0a7ae3ac2f1173085d398531c705756c94a4c56843785df85a60c1a0afac116"
+checksum = "915a1e146535de9163f3987b8944ed8cf49a18bb0056bcebcdcece385cece4ff"
 
 [[package]]
 name = "pin-utils"
@@ -6386,6 +6387,16 @@ dependencies = [
  "winapi 0.3.9",
 ]
 
+[[package]]
+name = "socket2"
+version = "0.5.7"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "ce305eb0b4296696835b71df73eb912e0f1ffd2556a501fcede6e0c50349191c"
+dependencies = [
+ "libc 0.2.151",
+ "windows-sys 0.52.0",
+]
+
 [[package]]
 name = "sst_importer"
 version = "0.1.0"
@@ -7574,6 +7585,7 @@ dependencies = [
  "tokio",
  "tokio-executor",
  "tokio-timer",
+ "tokio-util",
  "toml",
  "tracker",
  "url",
@@ -7676,21 +7688,20 @@ dependencies = [
 
 [[package]]
 name = "tokio"
-version = "1.25.3"
+version = "1.41.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "8666f87015685834a42aa61a391303d3bee0b1442dd9cf93e3adf4cbaf8de75a"
+checksum = "22cfb5bee7a6a52939ca9224d6ac897bb669134078daa8735560897f69de4d33"
 dependencies = [
- "autocfg",
+ "backtrace",
  "bytes",
  "libc 0.2.151",
- "mio 0.8.11",
- "num_cpus",
+ "mio 1.0.2",
  "parking_lot 0.12.1",
  "pin-project-lite",
  "signal-hook-registry",
- "socket2",
+ "socket2 0.5.7",
  "tokio-macros",
- "windows-sys 0.42.0",
+ "windows-sys 0.52.0",
 ]
 
 [[package]]
@@ -7704,13 +7715,13 @@ dependencies = [
 
 [[package]]
 name = "tokio-macros"
-version = "1.7.0"
+version = "2.4.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "b557f72f448c511a979e2564e55d74e6c4432fc96ff4f6241bc6bded342643b7"
+checksum = "693d596312e88961bc67d7f1f97af8a70227d9f90c31bba5806eec004978d752"
 dependencies = [
  "proc-macro2",
  "quote",
- "syn 1.0.103",
+ "syn 2.0.79",
 ]
 
 [[package]]
@@ -7759,17 +7770,18 @@ dependencies = [
 
 [[package]]
 name = "tokio-util"
-version = "0.7.2"
+version = "0.7.12"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "f988a1a1adc2fb21f9c12aa96441da33a1728193ae0b95d2be22dbd17fcb4e5c"
+checksum = "61e7c3654c13bcd040d4a03abee2c75b1d14a37b423cf5a813ceae1cc903ec6a"
 dependencies = [
  "bytes",
  "futures-core",
  "futures-io",
  "futures-sink",
+ "futures-util",
+ "hashbrown 0.14.0",
  "pin-project-lite",
  "tokio",
- "tracing",
 ]
 
 [[package]]
diff --git a/components/backup-stream/src/subscription_manager.rs b/components/backup-stream/src/subscription_manager.rs
index e1803048d71..15903d11e72 100644
--- a/components/backup-stream/src/subscription_manager.rs
+++ b/components/backup-stream/src/subscription_manager.rs
@@ -920,17 +920,15 @@ mod test {
         fail::cfg("execute_scan_command_sleep_100", "return").unwrap();
         for _ in 0..100 {
             let wg = wg.clone();
-            assert!(
-                block_on(pool.request(ScanCmd {
-                    region: Default::default(),
-                    handle: Default::default(),
-                    last_checkpoint: Default::default(),
-                    feedback_channel: tx.clone(),
-                    // Note: Maybe make here a Box<dyn FnOnce()> or some other trait?
-                    _work: wg.work(),
-                }))
-                .is_ok()
-            )
+            block_on(pool.request(ScanCmd {
+                region: Default::default(),
+                handle: Default::default(),
+                last_checkpoint: Default::default(),
+                feedback_channel: tx.clone(),
+                // Note: Maybe make here a Box<dyn FnOnce()> or some other trait?
+                _work: wg.work(),
+            }))
+            .unwrap();
         }
 
         should_finish_in(move || drop(pool), Duration::from_secs(5));
diff --git a/components/backup/src/endpoint.rs b/components/backup/src/endpoint.rs
index 67cd07c0a13..4bd02cafd70 100644
--- a/components/backup/src/endpoint.rs
+++ b/components/backup/src/endpoint.rs
@@ -38,6 +38,7 @@ use tikv_util::{
     box_err, debug, error, error_unknown,
     future::RescheduleChecker,
     impl_display_as_debug, info,
+    resizable_threadpool::ResizableRuntime,
     store::find_peer,
     time::{Instant, Limiter},
     warn,
@@ -49,7 +50,7 @@ use txn_types::{Key, Lock, TimeStamp};
 use crate::{
     metrics::*,
     softlimit::{CpuStatistics, SoftLimit, SoftLimitByCpu},
-    utils::{ControlThreadPool, KeyValueCodec},
+    utils::KeyValueCodec,
     writer::{BackupWriterBuilder, CfNameWrap},
     Error, *,
 };
@@ -704,7 +705,7 @@ impl SoftLimitKeeper {
 /// It coordinates backup tasks and dispatches them to different workers.
 pub struct Endpoint<E: Engine, R: RegionInfoProvider + Clone + 'static> {
     store_id: u64,
-    pool: RefCell<ControlThreadPool>,
+    pool: RefCell<ResizableRuntime>,
     io_pool: Runtime,
     tablets: LocalTablets<E::Local>,
     config_manager: ConfigManager,
@@ -713,7 +714,6 @@ pub struct Endpoint<E: Engine, R: RegionInfoProvider + Clone + 'static> {
     api_version: ApiVersion,
     causal_ts_provider: Option<Arc<CausalTsProviderImpl>>, // used in rawkv apiv2 only
     resource_ctl: Option<Arc<ResourceGroupManager>>,
-
     pub(crate) engine: E,
     pub(crate) region_info: R,
 }
@@ -879,7 +879,12 @@ impl<E: Engine, R: RegionInfoProvider + Clone + 'static> Endpoint<E, R> {
         causal_ts_provider: Option<Arc<CausalTsProviderImpl>>,
         resource_ctl: Option<Arc<ResourceGroupManager>>,
     ) -> Endpoint<E, R> {
-        let pool = ControlThreadPool::new();
+        let pool = ResizableRuntime::new(
+            config.num_threads,
+            "bkwkr",
+            Box::new(utils::create_tokio_runtime),
+            Box::new(|new_size| BACKUP_THREAD_POOL_SIZE_GAUGE.set(new_size as i64)),
+        );
         let rt = utils::create_tokio_runtime(config.io_thread_size, "backup-io").unwrap();
         let config_manager = ConfigManager(Arc::new(RwLock::new(config)));
         let softlimit = SoftLimitKeeper::new(config_manager.clone());
@@ -1502,8 +1507,12 @@ pub mod tests {
         use std::thread::sleep;
 
         let counter = Arc::new(AtomicU32::new(0));
-        let mut pool = ControlThreadPool::new();
-        pool.adjust_with(3);
+        let mut pool = ResizableRuntime::new(
+            3,
+            "bkwkr",
+            Box::new(utils::create_tokio_runtime),
+            Box::new(|new_size: usize| BACKUP_THREAD_POOL_SIZE_GAUGE.set(new_size as i64)),
+        );
 
         for i in 0..8 {
             let ctr = counter.clone();
@@ -2540,20 +2549,20 @@ pub mod tests {
         endpoint.get_config_manager().set_num_threads(15);
         let (task, _) = Task::new(req.clone(), tx.clone()).unwrap();
         endpoint.handle_backup_task(task);
-        assert!(endpoint.pool.borrow().size == 15);
+        assert!(endpoint.pool.borrow().size() == 15);
 
         // shrink thread pool only if there are too many idle threads
         endpoint.get_config_manager().set_num_threads(10);
         req.set_start_key(vec![b'2']);
         let (task, _) = Task::new(req.clone(), tx.clone()).unwrap();
         endpoint.handle_backup_task(task);
-        assert!(endpoint.pool.borrow().size == 15);
+        assert!(endpoint.pool.borrow().size() == 10);
 
         endpoint.get_config_manager().set_num_threads(3);
         req.set_start_key(vec![b'3']);
         let (task, _) = Task::new(req, tx).unwrap();
         endpoint.handle_backup_task(task);
-        assert!(endpoint.pool.borrow().size == 3);
+        assert!(endpoint.pool.borrow().size() == 3);
 
         // make sure all tasks can finish properly.
         let responses = block_on(rx.collect::<Vec<_>>());
@@ -2562,8 +2571,12 @@ pub mod tests {
         // for testing whether dropping the pool before all tasks finished causes panic.
         // but the panic must be checked manually. (It may panic at tokio runtime
         // threads)
-        let mut pool = ControlThreadPool::new();
-        pool.adjust_with(1);
+        let mut pool = ResizableRuntime::new(
+            1,
+            "bkwkr",
+            Box::new(utils::create_tokio_runtime),
+            Box::new(|new_size: usize| BACKUP_THREAD_POOL_SIZE_GAUGE.set(new_size as i64)),
+        );
         pool.spawn(async { tokio::time::sleep(Duration::from_millis(100)).await });
         pool.adjust_with(2);
         drop(pool);
diff --git a/components/backup/src/utils.rs b/components/backup/src/utils.rs
index 9d85eb664eb..5ca859f80f4 100644
--- a/components/backup/src/utils.rs
+++ b/components/backup/src/utils.rs
@@ -1,106 +1,18 @@
 // Copyright 2021 TiKV Project Authors. Licensed under Apache-2.0.
 
-use std::sync::Arc;
-
 use api_version::{dispatch_api_version, ApiV2, KeyMode, KvFormat};
 use file_system::IoType;
-use futures::Future;
 use kvproto::kvrpcpb::ApiVersion;
 use tikv_util::{error, sys::thread::ThreadBuildWrapper};
 use tokio::{io::Result as TokioResult, runtime::Runtime};
 use txn_types::{Key, TimeStamp};
 
-use crate::{metrics::*, Result};
+use crate::Result;
 
 // BACKUP_V1_TO_V2_TS is used as causal timestamp to backup RawKV api version
 // V1/V1Ttl data and save to V2 format. Use 1 other than 0 because 0 is not a
 // acceptable value for causal timestamp. See api_version::ApiV2::is_valid_ts.
 pub const BACKUP_V1_TO_V2_TS: u64 = 1;
-/// DaemonRuntime is a "background" runtime, which contains "daemon" tasks:
-/// any task spawn into it would run until finish even the runtime isn't
-/// referenced.
-pub struct DaemonRuntime(Option<Runtime>);
-
-impl DaemonRuntime {
-    /// spawn a daemon task to the runtime.
-    pub fn spawn(self: &Arc<Self>, f: impl Future<Output = ()> + Send + 'static) {
-        let wkr = self.clone();
-        self.0.as_ref().unwrap().spawn(async move {
-            f.await;
-            drop(wkr)
-        });
-    }
-
-    /// create a daemon runtime from some runtime.
-    pub fn from_runtime(rt: Runtime) -> Arc<Self> {
-        Arc::new(Self(Some(rt)))
-    }
-}
-
-impl Drop for DaemonRuntime {
-    fn drop(&mut self) {
-        // it is safe because all tasks should be finished.
-        self.0.take().unwrap().shutdown_background()
-    }
-}
-pub struct ControlThreadPool {
-    pub(crate) size: usize,
-    workers: Option<Arc<DaemonRuntime>>,
-}
-
-impl ControlThreadPool {
-    pub fn new() -> Self {
-        ControlThreadPool {
-            size: 0,
-            workers: None,
-        }
-    }
-
-    pub fn spawn<F>(&self, func: F)
-    where
-        F: Future<Output = ()> + Send + 'static,
-    {
-        self.workers
-            .as_ref()
-            .expect("ControlThreadPool: please call adjust_with() before spawn()")
-            .spawn(func);
-    }
-
-    /// Lazily adjust the thread pool's size
-    ///
-    /// Resizing if the thread pool need to expend or there
-    /// are too many idle threads. Otherwise do nothing.
-    pub fn adjust_with(&mut self, new_size: usize) {
-        if self.size >= new_size && self.size - new_size <= 10 {
-            return;
-        }
-        // TODO: after tokio supports adjusting thread pool size(https://github.com/tokio-rs/tokio/issues/3329),
-        //   adapt it.
-        let workers = create_tokio_runtime(new_size, "bkwkr")
-            .expect("failed to create tokio runtime for backup worker.");
-        self.workers = Some(DaemonRuntime::from_runtime(workers));
-        self.size = new_size;
-        BACKUP_THREAD_POOL_SIZE_GAUGE.set(new_size as i64);
-    }
-}
-
-/// Create a standard tokio runtime.
-/// (which allows io and time reactor, involve thread memory accessor),
-/// and set the I/O type to export.
-pub fn create_tokio_runtime(thread_count: usize, thread_name: &str) -> TokioResult<Runtime> {
-    tokio::runtime::Builder::new_multi_thread()
-        .thread_name(thread_name)
-        .enable_io()
-        .enable_time()
-        .with_sys_and_custom_hooks(
-            || {
-                file_system::set_io_type(IoType::Export);
-            },
-            || {},
-        )
-        .worker_threads(thread_count)
-        .build()
-}
 
 #[derive(Debug, Copy, Clone)]
 pub struct KeyValueCodec {
@@ -264,6 +176,24 @@ impl KeyValueCodec {
     }
 }
 
+/// Create a standard tokio runtime.
+/// (which allows io and time reactor, involve thread memory accessor),
+/// and set the I/O type to export.
+pub(crate) fn create_tokio_runtime(thread_count: usize, thread_name: &str) -> TokioResult<Runtime> {
+    tokio::runtime::Builder::new_multi_thread()
+        .thread_name(thread_name)
+        .enable_io()
+        .enable_time()
+        .with_sys_and_custom_hooks(
+            || {
+                file_system::set_io_type(IoType::Export);
+            },
+            || {},
+        )
+        .worker_threads(thread_count)
+        .build()
+}
+
 #[cfg(test)]
 pub mod tests {
     use api_version::{KvFormat, RawValue};
diff --git a/components/sst_importer/src/config.rs b/components/sst_importer/src/config.rs
index 7e83a07f2b2..cdb6e79afc7 100644
--- a/components/sst_importer/src/config.rs
+++ b/components/sst_importer/src/config.rs
@@ -1,19 +1,17 @@
 // Copyright 2018 TiKV Project Authors. Licensed under Apache-2.0.
-
 use std::{
     error::Error,
     result::Result,
-    sync::{Arc, RwLock},
+    sync::{Arc, Mutex, RwLock, Weak},
 };
 
 use online_config::{self, OnlineConfig};
-use tikv_util::{config::ReadableDuration, HandyRwLock};
+use tikv_util::{config::ReadableDuration, resizable_threadpool::ResizableRuntime, HandyRwLock};
 
 #[derive(Clone, Serialize, Deserialize, PartialEq, Debug, OnlineConfig)]
 #[serde(default)]
 #[serde(rename_all = "kebab-case")]
 pub struct Config {
-    #[online_config(skip)]
     pub num_threads: usize,
     #[online_config(skip)]
     pub stream_channel_window: usize,
@@ -62,11 +60,17 @@ impl Config {
 }
 
 #[derive(Clone)]
-pub struct ConfigManager(pub Arc<RwLock<Config>>);
+pub struct ConfigManager {
+    pub config: Arc<RwLock<Config>>,
+    pool: Weak<Mutex<ResizableRuntime>>,
+}
 
 impl ConfigManager {
-    pub fn new(cfg: Config) -> Self {
-        ConfigManager(Arc::new(RwLock::new(cfg)))
+    pub fn new(cfg: Config, pool: Weak<Mutex<ResizableRuntime>>) -> Self {
+        ConfigManager {
+            config: Arc::new(RwLock::new(cfg)),
+            pool,
+        }
     }
 }
 
@@ -88,6 +92,11 @@ impl online_config::ConfigManager for ConfigManager {
             return Err(e);
         }
 
+        if let Some(pool) = self.pool.upgrade() {
+            let mut pool = pool.lock().unwrap();
+            pool.adjust_with(cfg.num_threads);
+        }
+
         *self.wl() = cfg;
         Ok(())
     }
@@ -95,8 +104,7 @@ impl online_config::ConfigManager for ConfigManager {
 
 impl std::ops::Deref for ConfigManager {
     type Target = RwLock<Config>;
-
     fn deref(&self) -> &Self::Target {
-        self.0.as_ref()
+        self.config.as_ref()
     }
 }
diff --git a/components/sst_importer/src/import_mode.rs b/components/sst_importer/src/import_mode.rs
index 5f5b5d1060e..7456b3706ee 100644
--- a/components/sst_importer/src/import_mode.rs
+++ b/components/sst_importer/src/import_mode.rs
@@ -11,8 +11,7 @@ use std::{
 use engine_traits::{CfOptions, DbOptions, KvEngine};
 use futures_util::compat::Future01CompatExt;
 use kvproto::import_sstpb::*;
-use tikv_util::timer::GLOBAL_TIMER_HANDLE;
-use tokio::runtime::Handle;
+use tikv_util::{resizable_threadpool::DeamonRuntimeHandle, timer::GLOBAL_TIMER_HANDLE};
 
 use super::{Config, Result};
 
@@ -88,7 +87,8 @@ impl ImportModeSwitcher {
         ImportModeSwitcher { inner, is_import }
     }
 
-    pub fn start<E: KvEngine>(&self, executor: &Handle, db: E) {
+    // start_resizable_threads only serves for resizable runtime
+    pub fn start_resizable_threads<E: KvEngine>(&self, executor: &DeamonRuntimeHandle, db: E) {
         // spawn a background future to put TiKV back into normal mode after timeout
         let inner = self.inner.clone();
         let switcher = Arc::downgrade(&inner);
@@ -245,10 +245,17 @@ mod tests {
     use engine_traits::{KvEngine, CF_DEFAULT};
     use tempfile::Builder;
     use test_sst_importer::{new_test_engine, new_test_engine_with_options};
-    use tikv_util::config::ReadableDuration;
+    use tikv_util::{config::ReadableDuration, resizable_threadpool::ResizableRuntime};
+    use tokio::{io::Result as TokioResult, runtime::Runtime};
 
     use super::*;
 
+    fn create_tokio_runtime(_: usize, _: &str) -> TokioResult<Runtime> {
+        tokio::runtime::Builder::new_multi_thread()
+            .enable_all()
+            .build()
+    }
+
     fn check_import_options<E>(
         db: &E,
         expected_db_opts: &ImportModeDbOptions,
@@ -305,13 +312,15 @@ mod tests {
         fn mf(_cf: &str, _name: &str, _v: f64) {}
 
         let cfg = Config::default();
-        let threads = tokio::runtime::Builder::new_current_thread()
-            .enable_all()
-            .build()
-            .unwrap();
 
+        let threads = ResizableRuntime::new(
+            cfg.num_threads,
+            "test",
+            Box::new(create_tokio_runtime),
+            Box::new(|_| {}),
+        );
         let switcher = ImportModeSwitcher::new(&cfg);
-        switcher.start(threads.handle(), db.clone());
+        switcher.start_resizable_threads(&threads.handle(), db.clone());
         check_import_options(&db, &normal_db_options, &normal_cf_options);
         assert!(switcher.enter_import_mode(&db, mf).unwrap());
         check_import_options(&db, &import_db_options, &import_cf_options);
@@ -343,13 +352,12 @@ mod tests {
             ..Config::default()
         };
 
-        let threads = tokio::runtime::Builder::new_current_thread()
-            .enable_all()
-            .build()
-            .unwrap();
-
+        let threads =
+            ResizableRuntime::new(4, "test", Box::new(create_tokio_runtime), Box::new(|_| {}));
         let switcher = ImportModeSwitcher::new(&cfg);
-        switcher.start(threads.handle(), db.clone());
+        let handle = threads.handle();
+
+        switcher.start_resizable_threads(&handle.clone(), db.clone());
         check_import_options(&db, &normal_db_options, &normal_cf_options);
         switcher.enter_import_mode(&db, mf).unwrap();
         check_import_options(&db, &import_db_options, &import_cf_options);
diff --git a/components/sst_importer/src/import_mode2.rs b/components/sst_importer/src/import_mode2.rs
index 4db29c47a6f..73669e5921e 100644
--- a/components/sst_importer/src/import_mode2.rs
+++ b/components/sst_importer/src/import_mode2.rs
@@ -8,8 +8,7 @@ use std::{
 use collections::{HashMap, HashSet};
 use futures_util::compat::Future01CompatExt;
 use kvproto::{import_sstpb::Range, metapb::Region};
-use tikv_util::timer::GLOBAL_TIMER_HANDLE;
-use tokio::runtime::Handle;
+use tikv_util::{resizable_threadpool::DeamonRuntimeHandle, timer::GLOBAL_TIMER_HANDLE};
 
 use super::Config;
 
@@ -56,9 +55,7 @@ impl ImportModeSwitcherV2 {
         ImportModeSwitcherV2 { inner }
     }
 
-    // Periodically perform timeout check to change import mode of some regions back
-    // to normal mode.
-    pub fn start(&self, executor: &Handle) {
+    pub fn start_resizable_threads(&self, executor: &DeamonRuntimeHandle) {
         // spawn a background future to put regions back into normal mode after timeout
         let inner = self.inner.clone();
         let switcher = Arc::downgrade(&inner);
@@ -157,10 +154,17 @@ pub fn range_overlaps(range1: &HashRange, range2: &Range) -> bool {
 mod test {
     use std::thread;
 
-    use tikv_util::config::ReadableDuration;
+    use tikv_util::{config::ReadableDuration, resizable_threadpool::ResizableRuntime};
+    use tokio::{io::Result as TokioResult, runtime::Runtime};
 
     use super::*;
 
+    fn create_tokio_runtime(_: usize, _: &str) -> TokioResult<Runtime> {
+        tokio::runtime::Builder::new_multi_thread()
+            .enable_all()
+            .build()
+    }
+
     #[test]
     fn test_region_range_overlaps() {
         let verify_overlap = |ranges1: &[(&str, &str)], ranges2: &[(&str, &str)], overlap: bool| {
@@ -270,10 +274,9 @@ mod test {
             ..Config::default()
         };
 
-        let threads = tokio::runtime::Builder::new_current_thread()
-            .enable_all()
-            .build()
-            .unwrap();
+        let threads =
+            ResizableRuntime::new(4, "test", Box::new(create_tokio_runtime), Box::new(|_| {}));
+        let handle = threads.handle();
 
         let switcher = ImportModeSwitcherV2::new(&cfg);
         let mut region = Region::default();
@@ -300,7 +303,7 @@ mod test {
         assert!(switcher.region_in_import_mode(&region2));
         assert!(switcher.region_in_import_mode(&region3));
 
-        switcher.start(threads.handle());
+        switcher.start_resizable_threads(&handle.clone());
 
         thread::sleep(Duration::from_millis(400));
         // renew the timeout of key_range2
diff --git a/components/sst_importer/src/sst_importer.rs b/components/sst_importer/src/sst_importer.rs
index 207a859c700..be3a1e123ed 100644
--- a/components/sst_importer/src/sst_importer.rs
+++ b/components/sst_importer/src/sst_importer.rs
@@ -39,14 +39,12 @@ use tikv_util::{
     },
     future::RescheduleChecker,
     memory::{MemoryQuota, OwnedAllocated},
+    resizable_threadpool::DeamonRuntimeHandle,
     sys::{thread::ThreadBuildWrapper, SysQuota},
     time::{Instant, Limiter},
     Either, HandyRwLock,
 };
-use tokio::{
-    runtime::{Handle, Runtime},
-    sync::OnceCell,
-};
+use tokio::{runtime::Runtime, sync::OnceCell};
 use txn_types::{Key, TimeStamp, WriteRef};
 
 use crate::{
@@ -267,10 +265,10 @@ impl<E: KvEngine> SstImporter<E> {
         }
     }
 
-    pub fn start_switch_mode_check(&self, executor: &Handle, db: Option<E>) {
+    pub fn start_switch_mode_check(&self, executor: &DeamonRuntimeHandle, db: Option<E>) {
         match &self.switcher {
-            Either::Left(switcher) => switcher.start(executor, db.unwrap()),
-            Either::Right(switcher) => switcher.start(executor),
+            Either::Left(switcher) => switcher.start_resizable_threads(executor, db.unwrap()),
+            Either::Right(switcher) => switcher.start_resizable_threads(executor),
         }
     }
 
@@ -1624,6 +1622,10 @@ mod tests {
     use std::{
         io::{self, Cursor},
         ops::Sub,
+        sync::{
+            atomic::{AtomicUsize, Ordering},
+            Mutex,
+        },
         usize,
     };
 
@@ -1647,8 +1649,14 @@ mod tests {
     use tempfile::{Builder, TempDir};
     use test_sst_importer::*;
     use test_util::new_test_key_manager;
-    use tikv_util::{codec::stream_event::EventEncoder, stream::block_on_external_io};
-    use tokio::io::{AsyncWrite, AsyncWriteExt};
+    use tikv_util::{
+        codec::stream_event::EventEncoder, resizable_threadpool::ResizableRuntime,
+        stream::block_on_external_io,
+    };
+    use tokio::{
+        io::{AsyncWrite, AsyncWriteExt, Result as TokioResult},
+        runtime::Runtime,
+    };
     use tokio_util::compat::{FuturesAsyncWriteCompatExt, TokioAsyncWriteCompatExt};
     use txn_types::{Value, WriteType};
     use uuid::Uuid;
@@ -1656,6 +1664,14 @@ mod tests {
     use super::*;
     use crate::{import_file::ImportPath, *};
 
+    static COUNTER: AtomicUsize = AtomicUsize::new(0);
+
+    fn create_tokio_runtime(_: usize, _: &str) -> TokioResult<Runtime> {
+        tokio::runtime::Builder::new_current_thread()
+            .enable_all()
+            .build()
+    }
+
     fn do_test_import_dir(key_manager: Option<Arc<DataKeyManager>>) {
         let temp_dir = Builder::new().prefix("test_import_dir").tempdir().unwrap();
         let dir = ImportDir::new(temp_dir.path()).unwrap();
@@ -2294,8 +2310,17 @@ mod tests {
         };
         let change = cfg.diff(&cfg_new);
 
+        let threads = ResizableRuntime::new(
+            cfg.num_threads,
+            "test",
+            Box::new(create_tokio_runtime),
+            Box::new(|_| {}),
+        );
+
+        let threads_clone = Arc::new(Mutex::new(threads));
+
         // create config manager and update config.
-        let mut cfg_mgr = ImportConfigManager::new(cfg);
+        let mut cfg_mgr = ImportConfigManager::new(cfg, Arc::downgrade(&threads_clone));
         cfg_mgr.dispatch(change).unwrap();
         importer.update_config_memory_use_ratio(&cfg_mgr);
 
@@ -2318,11 +2343,50 @@ mod tests {
             ..Default::default()
         };
         let change = cfg.diff(&cfg_new);
-        let mut cfg_mgr = ImportConfigManager::new(cfg);
+
+        let threads = ResizableRuntime::new(
+            cfg.num_threads,
+            "test",
+            Box::new(create_tokio_runtime),
+            Box::new(|_| {}),
+        );
+
+        let threads_clone = Arc::new(Mutex::new(threads));
+
+        let mut cfg_mgr = ImportConfigManager::new(cfg, Arc::downgrade(&threads_clone));
         let r = cfg_mgr.dispatch(change);
         assert!(r.is_err());
     }
 
+    #[test]
+    fn test_update_import_num_threads() {
+        let cfg = Config::default();
+        let threads = ResizableRuntime::new(
+            Config::default().num_threads,
+            "test",
+            Box::new(create_tokio_runtime),
+            Box::new(|new_size: usize| {
+                COUNTER.store(new_size, Ordering::SeqCst);
+            }),
+        );
+
+        let threads_clone = Arc::new(Mutex::new(threads));
+        let mut cfg_mgr = ImportConfigManager::new(cfg, Arc::downgrade(&threads_clone));
+
+        assert_eq!(cfg_mgr.rl().num_threads, Config::default().num_threads);
+
+        let cfg_new = Config {
+            num_threads: 10,
+            ..Default::default()
+        };
+        let change = Config::default().diff(&cfg_new);
+        let r = cfg_mgr.dispatch(change);
+
+        r.unwrap();
+        assert_eq!(cfg_mgr.rl().num_threads, cfg_new.num_threads);
+        assert_eq!(COUNTER.load(Ordering::SeqCst), cfg_mgr.rl().num_threads);
+    }
+
     #[test]
     fn test_download_kv_file_to_mem_cache() {
         // create a sample kv file.
diff --git a/components/tikv_util/Cargo.toml b/components/tikv_util/Cargo.toml
index e1ab04cc732..1ef223db4e9 100644
--- a/components/tikv_util/Cargo.toml
+++ b/components/tikv_util/Cargo.toml
@@ -60,9 +60,10 @@ sysinfo = "0.26"
 thiserror = "1.0"
 tikv_alloc = { workspace = true }
 time = { workspace = true }
-tokio = { version = "1.5", features = ["rt-multi-thread"] }
+tokio = { version = "1.5", features = ["rt-multi-thread","time", "rt", "macros", "sync", "full"] }
 tokio-executor = { workspace = true }
 tokio-timer = { workspace = true }
+tokio-util = { version = "0.7", features = ["rt"] }
 tracker = { workspace = true }
 url = "2"
 yatp = { workspace = true }
diff --git a/components/tikv_util/src/lib.rs b/components/tikv_util/src/lib.rs
index 1171e5f4286..7d30fe11874 100644
--- a/components/tikv_util/src/lib.rs
+++ b/components/tikv_util/src/lib.rs
@@ -54,6 +54,7 @@ pub mod memory;
 pub mod metrics;
 pub mod mpsc;
 pub mod quota_limiter;
+pub mod resizable_threadpool;
 pub mod resource_control;
 pub mod smoother;
 pub mod store;
diff --git a/components/tikv_util/src/resizable_threadpool.rs b/components/tikv_util/src/resizable_threadpool.rs
new file mode 100644
index 00000000000..a393dd9f8d2
--- /dev/null
+++ b/components/tikv_util/src/resizable_threadpool.rs
@@ -0,0 +1,346 @@
+// Copyright 2021 TiKV Project Authors. Licensed under Apache-2.0.
+
+use std::sync::{Arc, Mutex, Weak};
+
+use futures::Future;
+use tokio::{
+    io::Result as TokioResult,
+    runtime::{Builder, Runtime},
+};
+use tokio_util::task::task_tracker::TaskTracker;
+
+struct DeamonRuntime {
+    inner: Option<Runtime>,
+    tracker: TaskTracker,
+}
+
+impl DeamonRuntime {
+    pub fn spawn<Fut>(&self, fut: Fut)
+    where
+        Fut: Future<Output = ()> + Send + 'static,
+    {
+        self.inner
+            .as_ref()
+            .unwrap()
+            .spawn(self.tracker.track_future(fut));
+    }
+}
+
+impl Drop for DeamonRuntime {
+    fn drop(&mut self) {
+        if let Some(runtime) = self.inner.take() {
+            runtime.shutdown_background();
+        }
+    }
+}
+
+#[derive(Clone)]
+pub struct DeamonRuntimeHandle {
+    inner: Weak<Mutex<DeamonRuntime>>,
+}
+
+impl DeamonRuntimeHandle {
+    pub fn spawn<Fut>(&self, fut: Fut)
+    where
+        Fut: Future<Output = ()> + Send + 'static,
+    {
+        let runtime = match self.inner.upgrade() {
+            Some(runtime) => runtime,
+            None => {
+                error!("Daemon runtime has been dropped. Task will be ignored.");
+                return;
+            }
+        };
+
+        let (handle, tracker) = {
+            let lock_guard = runtime.lock().unwrap();
+            let inner = lock_guard
+                .inner
+                .as_ref()
+                .expect("Runtime inner should exist");
+            (inner.handle().clone(), lock_guard.tracker.clone())
+        };
+
+        handle.spawn(tracker.track_future(fut));
+    }
+
+    pub fn block_on<Fut>(&self, fut: Fut)
+    where
+        Fut: Future<Output = ()> + Send + 'static,
+    {
+        let runtime = match self.inner.upgrade() {
+            Some(runtime) => runtime,
+            None => {
+                error!("Daemon runtime has been dropped. Task will be ignored.");
+                return;
+            }
+        };
+
+        let (handle, tracker) = {
+            let lock_guard = runtime.lock().unwrap();
+            let inner = lock_guard
+                .inner
+                .as_ref()
+                .expect("Runtime inner should exist");
+            (inner.handle().clone(), lock_guard.tracker.clone())
+        };
+
+        handle.block_on(tracker.track_future(fut));
+    }
+}
+
+pub struct ResizableRuntime {
+    size: usize,
+    version: usize,
+    thread_prefix: String,
+    gc_runtime: DeamonRuntime,
+    current_runtime: Arc<Mutex<DeamonRuntime>>,
+    replace_pool_rule: Box<dyn Fn(usize, &str) -> TokioResult<Runtime> + Send + Sync>,
+    after_adjust: Box<dyn Fn(usize) + Send + Sync>,
+}
+
+impl ResizableRuntime {
+    pub fn new(
+        thread_size: usize,
+        thread_prefix: &str,
+        replace_pool_rule: Box<dyn Fn(usize, &str) -> TokioResult<Runtime> + Send + Sync>,
+        after_adjust: Box<dyn Fn(usize) + Send + Sync>,
+    ) -> Self {
+        let init_name = format!("{}-v0", thread_prefix);
+        let keeper = Builder::new_multi_thread()
+            .worker_threads(1)
+            .thread_name("rtkp")
+            .enable_all()
+            .build()
+            .expect("Failed to create runtime-keeper");
+        let new_runtime = (replace_pool_rule)(thread_size, &init_name)
+            .unwrap_or_else(|_| panic!("failed to create tokio runtime {}", thread_prefix));
+
+        ResizableRuntime {
+            size: thread_size,
+            version: 0,
+            thread_prefix: thread_prefix.to_owned(),
+            gc_runtime: DeamonRuntime {
+                inner: Some(keeper),
+                tracker: TaskTracker::new(),
+            },
+            current_runtime: Arc::new(Mutex::new(DeamonRuntime {
+                inner: Some(new_runtime),
+                tracker: TaskTracker::new(),
+            })),
+            replace_pool_rule,
+            after_adjust,
+        }
+    }
+
+    pub fn size(&self) -> usize {
+        self.size
+    }
+
+    pub fn spawn<Fut>(&self, fut: Fut)
+    where
+        Fut: Future<Output = ()> + Send + 'static,
+    {
+        let handle = self.handle();
+        handle.spawn(fut);
+    }
+
+    pub fn block_on<Fut>(&self, fut: Fut)
+    where
+        Fut: Future<Output = ()> + Send + 'static,
+    {
+        let handle = self.handle();
+        handle.block_on(fut);
+    }
+
+    pub fn handle(&self) -> DeamonRuntimeHandle {
+        DeamonRuntimeHandle {
+            inner: Arc::downgrade(&self.current_runtime),
+        }
+    }
+
+    // TODO: after tokio supports adjusting thread pool size(https://github.com/tokio-rs/tokio/issues/3329),
+    //   adapt it.
+    pub fn adjust_with(&mut self, new_size: usize) {
+        if self.size == new_size {
+            return;
+        }
+
+        self.version += 1;
+        let thread_name = format!("{}-v{}", self.thread_prefix, self.version);
+        let new_runtime = (self.replace_pool_rule)(new_size, &thread_name)
+            .unwrap_or_else(|_| panic!("failed to create tokio runtime {}", thread_name));
+
+        let old_runtime: DeamonRuntime;
+        {
+            let mut runtime_guard = self.current_runtime.lock().unwrap();
+
+            old_runtime = std::mem::replace(
+                &mut *runtime_guard,
+                DeamonRuntime {
+                    inner: Some(new_runtime),
+                    tracker: TaskTracker::new(),
+                },
+            );
+        }
+
+        self.size = new_size;
+
+        info!(
+            "Resizing thread pool";
+            "thread_name" => &thread_name,
+            "new_size" => new_size
+        );
+        self.gc_runtime.spawn(async move {
+            old_runtime.tracker.close();
+            old_runtime.tracker.wait().await;
+            drop(old_runtime);
+        });
+        (self.after_adjust)(new_size);
+    }
+}
+
+#[cfg(test)]
+mod test {
+    use std::{
+        future,
+        sync::atomic::{AtomicUsize, Ordering},
+        thread::{self, sleep},
+        time::Duration,
+    };
+
+    use super::*;
+    use crate::time::Instant;
+
+    fn replace_pool_rule(thread_count: usize, thread_name: &str) -> TokioResult<Runtime> {
+        let rt = tokio::runtime::Builder::new_multi_thread()
+            .worker_threads(thread_count)
+            .thread_name(thread_name)
+            .enable_all()
+            .build()
+            .unwrap();
+        Ok(rt)
+    }
+
+    #[test]
+    fn test_adjust_thread_num() {
+        static COUNTER: AtomicUsize = AtomicUsize::new(4);
+        let after_adjust = |new_size: usize| {
+            COUNTER.store(new_size, Ordering::SeqCst);
+        };
+        let mut threads = ResizableRuntime::new(
+            COUNTER.load(Ordering::SeqCst),
+            "test",
+            Box::new(replace_pool_rule),
+            Box::new(after_adjust),
+        );
+        assert_eq!(COUNTER.load(Ordering::SeqCst), threads.size());
+
+        let handle = threads.handle();
+        handle.block_on(async {
+            COUNTER.fetch_add(1, Ordering::SeqCst);
+        });
+        assert_eq!(COUNTER.load(Ordering::SeqCst), threads.size() + 1);
+
+        threads.adjust_with(8);
+        assert!(!threads.gc_runtime.tracker.is_empty());
+        assert_eq!(COUNTER.load(Ordering::SeqCst), threads.size());
+
+        sleep(Duration::from_secs(1));
+        assert!(threads.gc_runtime.tracker.is_empty());
+
+        // New task should be scheduled to the new runtime
+        handle.block_on(async {
+            COUNTER.fetch_add(1, Ordering::SeqCst);
+        });
+        assert_eq!(COUNTER.load(Ordering::SeqCst), threads.size() + 1);
+    }
+
+    #[test]
+    fn test_infinite_loop() {
+        static COUNTER: AtomicUsize = AtomicUsize::new(4);
+        let after_adjust = |new_size: usize| {
+            COUNTER.store(new_size, Ordering::SeqCst);
+        };
+        let mut threads = ResizableRuntime::new(
+            COUNTER.load(Ordering::SeqCst),
+            "test",
+            Box::new(replace_pool_rule),
+            Box::new(after_adjust),
+        );
+        assert_eq!(COUNTER.load(Ordering::SeqCst), threads.size());
+
+        let handle = threads.handle();
+        // infinite loop should not be cleaned
+        handle.spawn(async {
+            future::pending::<()>().await;
+        });
+
+        threads.adjust_with(8);
+        sleep(Duration::from_secs(1));
+        assert!(!threads.gc_runtime.tracker.is_empty());
+        assert_eq!(COUNTER.load(Ordering::SeqCst), threads.size());
+    }
+
+    #[test]
+    fn test_drop() {
+        let start = Instant::now();
+        let threads =
+            ResizableRuntime::new(4, "test", Box::new(replace_pool_rule), Box::new(|_| {}));
+        let handle = threads.handle();
+        let handle_clone = handle.clone();
+        handle.spawn(async {
+            future::pending::<()>().await;
+        });
+        let thread = thread::spawn(move || {
+            handle_clone.block_on(async {
+                future::pending::<()>().await;
+            });
+        });
+        drop(threads);
+        handle.spawn(async {
+            future::pending::<()>().await;
+        });
+        handle.block_on(async {
+            future::pending::<()>().await;
+        });
+        thread.join().unwrap();
+
+        assert!(Instant::now() - start < Duration::from_secs(5));
+    }
+
+    #[test]
+    fn test_multi_tasks() {
+        let threads = Arc::new(ResizableRuntime::new(
+            8,
+            "test",
+            Box::new(replace_pool_rule),
+            Box::new(|_| {}),
+        ));
+        let handle = threads.handle();
+
+        let handles: Vec<_> = (0..2000)
+            .map(|i| {
+                let runtime_handle = handle.clone();
+                thread::spawn(move || {
+                    if i % 2 == 0 {
+                        runtime_handle.block_on(async move {
+                            sleep(Duration::from_millis(500));
+                            println!("Thread {} finished", i);
+                        });
+                    } else {
+                        runtime_handle.spawn(async move {
+                            sleep(Duration::from_millis(500));
+                            println!("Thread {} finished", i);
+                        })
+                    }
+                })
+            })
+            .collect();
+
+        // Wait for all threads to complete
+        for handle in handles {
+            handle.join().expect("Thread panicked");
+        }
+    }
+}
diff --git a/components/tikv_util/src/stream.rs b/components/tikv_util/src/stream.rs
index 4f16a75ef57..84b8eede06d 100644
--- a/components/tikv_util/src/stream.rs
+++ b/components/tikv_util/src/stream.rs
@@ -76,8 +76,7 @@ pub fn error_stream(e: io::Error) -> impl Stream<Item = io::Result<Bytes>> + Unp
 pub fn block_on_external_io<F: Future>(f: F) -> F::Output {
     // we need a Tokio runtime, Tokio futures require Tokio executor.
     Builder::new_current_thread()
-        .enable_io()
-        .enable_time()
+        .enable_all()
         .build()
         .expect("failed to create Tokio runtime")
         .block_on(f)
diff --git a/src/import/sst_service.rs b/src/import/sst_service.rs
index 2973dda3caf..8efff75660d 100644
--- a/src/import/sst_service.rs
+++ b/src/import/sst_service.rs
@@ -38,6 +38,7 @@ use tikv_kv::{Engine, LocalTablets, Modify, WriteData};
 use tikv_util::{
     config::ReadableSize,
     future::{create_stream_with_buffer, paired_future_callback},
+    resizable_threadpool::{DeamonRuntimeHandle, ResizableRuntime},
     sys::{
         disk::{get_disk_status, DiskUsage},
         thread::ThreadBuildWrapper,
@@ -45,7 +46,7 @@ use tikv_util::{
     time::{Instant, Limiter},
     HandyRwLock,
 };
-use tokio::{runtime::Runtime, time::sleep};
+use tokio::time::sleep;
 use txn_types::{Key, WriteRef, WriteType};
 
 use super::{
@@ -120,7 +121,10 @@ pub struct ImportSstService<E: Engine> {
     cfg: ConfigManager,
     tablets: LocalTablets<E::Local>,
     engine: E,
-    threads: Arc<Runtime>,
+    threads: DeamonRuntimeHandle,
+    // threads_ref is for safely cleanning
+    #[allow(dead_code)]
+    threads_ref: Arc<Mutex<ResizableRuntime>>,
     importer: Arc<SstImporter<E::Local>>,
     limiter: Limiter,
     ingest_latch: Arc<IngestLatch>,
@@ -322,47 +326,60 @@ impl<E: Engine> ImportSstService<E> {
         resource_manager: Option<Arc<ResourceGroupManager>>,
         region_info_accessor: Arc<RegionInfoAccessor>,
     ) -> Self {
-        let props = tikv_util::thread_group::current_properties();
-        let eng = Mutex::new(engine.clone());
-        let threads = tokio::runtime::Builder::new_multi_thread()
-            .worker_threads(cfg.num_threads)
-            .enable_all()
-            .thread_name("sst-importer")
-            .with_sys_and_custom_hooks(
-                move || {
-                    tikv_util::thread_group::set_properties(props.clone());
-
-                    set_io_type(IoType::Import);
-                    tikv_kv::set_tls_engine(eng.lock().unwrap().clone());
-                },
-                move || {
-                    // SAFETY: we have set the engine at some lines above with type `E`.
-                    unsafe { tikv_kv::destroy_tls_engine::<E>() };
-                },
-            )
-            .build()
-            .unwrap();
+        let eng = Arc::new(Mutex::new(engine.clone()));
+        let create_tokio_runtime = move |thread_count: usize, thread_name: &str| {
+            let props = tikv_util::thread_group::current_properties();
+            let eng = eng.clone();
+            tokio::runtime::Builder::new_multi_thread()
+                .worker_threads(thread_count)
+                .enable_all()
+                .thread_name(thread_name)
+                .with_sys_and_custom_hooks(
+                    move || {
+                        tikv_util::thread_group::set_properties(props.clone());
+                        set_io_type(IoType::Import);
+                        tikv_kv::set_tls_engine(eng.lock().unwrap().clone());
+                    },
+                    move || {
+                        // SAFETY: we have set the engine at some lines above with type `E`.
+                        unsafe { tikv_kv::destroy_tls_engine::<E>() };
+                    },
+                )
+                .build()
+        };
+
+        let threads = ResizableRuntime::new(
+            4,
+            "impwkr",
+            Box::new(create_tokio_runtime),
+            Box::new(|_| ()),
+        );
+
+        let handle = threads.handle();
+        let threads_clone = Arc::new(Mutex::new(threads));
         if let LocalTablets::Singleton(tablet) = &tablets {
-            importer.start_switch_mode_check(threads.handle(), Some(tablet.clone()));
+            importer.start_switch_mode_check(&handle.clone(), Some(tablet.clone()));
         } else {
-            importer.start_switch_mode_check(threads.handle(), None);
+            importer.start_switch_mode_check(&handle.clone(), None);
         }
-
         let writer = raft_writer::ThrottledTlsEngineWriter::default();
         let gc_handle = writer.clone();
-        threads.spawn(async move {
+        handle.spawn(async move {
             while gc_handle.try_gc() {
                 tokio::time::sleep(WRITER_GC_INTERVAL).await;
             }
         });
-
-        let cfg_mgr = ConfigManager::new(cfg);
-        threads.spawn(Self::tick(importer.clone(), cfg_mgr.clone()));
+        let num_threads = cfg.num_threads;
+        let cfg_mgr = ConfigManager::new(cfg, Arc::downgrade(&threads_clone));
+        handle.spawn(Self::tick(importer.clone(), cfg_mgr.clone()));
+        // Drop the initial pool to accept new tasks
+        threads_clone.lock().unwrap().adjust_with(num_threads);
 
         ImportSstService {
             cfg: cfg_mgr,
             tablets,
-            threads: Arc::new(threads),
+            threads: handle.clone(),
+            threads_ref: threads_clone,
             engine,
             importer,
             limiter: Limiter::new(f64::INFINITY),


Test patch: