deal comments in PR

Signed-off-by: joccau <zak.zhao@pingcap.com>
tikv · Mar 17, 2023 · e0054f1 · e0054f1
1 parent a516a43
commit e0054f1
Show file tree

Hide file tree

Showing 3 changed files with 19 additions and 15 deletions.
diff --git a/components/sst_importer/src/config.rs b/components/sst_importer/src/config.rs
@@ -7,7 +7,7 @@ use std::{
 };
 
 use online_config::{self, OnlineConfig};
-use tikv_util::config::ReadableDuration;
+use tikv_util::{config::ReadableDuration, HandyRwLock};
 
 #[derive(Clone, Serialize, Deserialize, PartialEq, Debug, OnlineConfig)]
 #[serde(default)]
@@ -85,7 +85,7 @@ impl online_config::ConfigManager for ConfigManager {
             return Err(e);
         }
 
-        *self.write().unwrap() = cfg;
+        *self.wl() = cfg;
         Ok(())
     }
 }

diff --git a/components/sst_importer/src/sst_importer.rs b/components/sst_importer/src/sst_importer.rs
@@ -38,6 +38,7 @@ use tikv_util::{
     },
     sys::{thread::ThreadBuildWrapper, SysQuota},
     time::{Instant, Limiter},
+    HandyRwLock,
 };
 use tokio::runtime::{Handle, Runtime};
 use txn_types::{Key, TimeStamp, WriteRef};
@@ -590,8 +591,8 @@ impl SstImporter {
         Ok(())
     }
 
-    pub fn update_config_memory_use_ratio(&self, cfg_mgr: ImportConfigManager) {
-        let mem_ratio = cfg_mgr.read().unwrap().memory_use_ratio;
+    pub fn update_config_memory_use_ratio(&self, cfg_mgr: &ImportConfigManager) {
+        let mem_ratio = cfg_mgr.rl().memory_use_ratio;
         let memory_limit = Self::calcualte_usage_mem(mem_ratio);
 
         if self.mem_limit.load(Ordering::SeqCst) != memory_limit {
@@ -663,7 +664,7 @@ impl SstImporter {
     // If mem_limit is 0, which represent download kv-file when import.
     // Or read kv-file into buffer directly.
     pub fn import_support_download(&self) -> bool {
-        self.mem_limit.load(Ordering::SeqCst) == 0
+        self.mem_limit.load(Ordering::SeqCst) <= 0
     }
 
     fn request_memory(&self, meta: &KvMeta) -> Option<MemUsePermit> {
@@ -2000,13 +2001,17 @@ mod tests {
         // create config manager and update config.
         let mut cfg_mgr = ImportConfigManager::new(cfg);
         cfg_mgr.dispatch(change).unwrap();
-        importer.update_config_memory_use_ratio(cfg_mgr);
+        importer.update_config_memory_use_ratio(&cfg_mgr);
 
         let mem_limit_new = importer.mem_limit.load(Ordering::SeqCst);
         assert!(mem_limit_old > mem_limit_new);
-
-        let delt = (mem_limit_old - mem_limit_new * 3) as f64 / mem_limit_old as f64;
-        assert!(delt < 0.000001)
+        assert_eq!(
+            mem_limit_old / 3,
+            mem_limit_new,
+            "mem_limit_old / 3 = {} mem_limit_new = {}",
+            mem_limit_old / 3,
+            mem_limit_new
+        );
     }
 
     #[test]

diff --git a/src/import/sst_service.rs b/src/import/sst_service.rs
@@ -35,6 +35,7 @@ use tikv_util::{
     future::create_stream_with_buffer,
     sys::thread::ThreadBuildWrapper,
     time::{Instant, Limiter},
+    HandyRwLock,
 };
 use tokio::{runtime::Runtime, time::sleep};
 use txn_types::{Key, WriteRef, WriteType};
@@ -321,7 +322,7 @@ impl<E: Engine> ImportSstService<E> {
         loop {
             sleep(Duration::from_secs(10)).await;
 
-            importer.update_config_memory_use_ratio(cfg.clone());
+            importer.update_config_memory_use_ratio(&cfg);
             importer.shrink_by_tick();
         }
     }
@@ -551,10 +552,8 @@ macro_rules! impl_write {
         ) {
             let import = self.importer.clone();
             let tablets = self.tablets.clone();
-            let (rx, buf_driver) = create_stream_with_buffer(
-                stream,
-                self.cfg.0.as_ref().read().unwrap().stream_channel_window,
-            );
+            let (rx, buf_driver) =
+                create_stream_with_buffer(stream, self.cfg.rl().stream_channel_window);
             let mut rx = rx.map_err(Error::from);
 
             let timer = Instant::now_coarse();
@@ -663,7 +662,7 @@ impl<E: Engine> ImportSst for ImportSstService<E> {
         let timer = Instant::now_coarse();
         let import = self.importer.clone();
         let (rx, buf_driver) =
-            create_stream_with_buffer(stream, self.cfg.read().unwrap().stream_channel_window);
+            create_stream_with_buffer(stream, self.cfg.rl().stream_channel_window);
         let mut map_rx = rx.map_err(Error::from);
 
         let handle_task = async move {