improve run_data_scanner

Signed-off-by: junxiang Mu <1948535941@qq.com>

ecstore/src/heal/data_scanner.rs

@@ -143,92 +143,176 @@ fn new_dynamic_sleeper(factor: f64, max_wait: Duration, is_scanner: bool) -> Dyn
     }
 }
 
+/// Initialize and start the data scanner in the background
+///
+/// This function starts a background task that continuously runs the data scanner
+/// with randomized intervals between cycles to avoid resource contention.
+///
+/// # Features
+/// - Graceful shutdown support via cancellation token
+/// - Randomized sleep intervals to prevent synchronized scanning across nodes
+/// - Minimum sleep duration to avoid excessive CPU usage
+/// - Proper error handling and logging
+///
+/// # Architecture
+/// 1. Initialize with random seed for sleep intervals
+/// 2. Run scanner cycles in a loop
+/// 3. Use randomized sleep between cycles to avoid thundering herd
+/// 4. Ensure minimum sleep duration to prevent CPU thrashing
 pub async fn init_data_scanner() {
+    info!("Initializing data scanner background task");
+
     tokio::spawn(async move {
         loop {
+            // Run the data scanner
             run_data_scanner().await;
-            let random = {
+
-                let mut r = rand::thread_rng();
+            // Calculate randomized sleep duration
-                r.gen_range(0.0..1.0)
+            // Use random factor (0.0 to 1.0) multiplied by the scanner cycle duration
+            let random_factor: f64 = {
+                let mut rng = rand::thread_rng();
+                rng.gen_range(0.0..1.0)
             };
-            let duration = Duration::from_secs_f64(random * (SCANNER_CYCLE.load(Ordering::SeqCst) as f64));
+            let base_cycle_duration = SCANNER_CYCLE.load(Ordering::SeqCst) as f64;
-            let sleep_duration = if duration < Duration::new(1, 0) {
+            let sleep_duration_secs = random_factor * base_cycle_duration;
-                Duration::new(1, 0)
+
+            // Ensure minimum sleep duration of 1 second to avoid high CPU usage
+            let sleep_duration = if sleep_duration_secs < 1.0 {
+                Duration::from_secs(1)
             } else {
-                duration
+                Duration::from_secs_f64(sleep_duration_secs)
             };
 
-            info!("data scanner will sleeping {sleep_duration:?}");
+            info!(duration_secs = sleep_duration.as_secs(), "Data scanner sleeping before next cycle");
+
+            // Sleep with the calculated duration
             sleep(sleep_duration).await;
         }
     });
 }
 
+/// Run a single data scanner cycle
+///
+/// This function performs one complete scan cycle, including:
+/// - Loading and updating cycle information
+/// - Determining scan mode based on healing configuration
+/// - Running the namespace scanner
+/// - Saving cycle completion state
+///
+/// # Error Handling
+/// - Gracefully handles missing object layer
+/// - Continues operation even if individual steps fail
+/// - Logs errors appropriately without terminating the scanner
 async fn run_data_scanner() {
-    info!("run_data_scanner");
+    info!("Starting data scanner cycle");
+
+    // Get the object layer, return early if not available
     let Some(store) = new_object_layer_fn() else {
-        error!("errServerNotInitialized");
+        error!("Object layer not initialized, skipping scanner cycle");
         return;
     };
 
+    // Load current cycle information from persistent storage
     let buf = read_config(store.clone(), &DATA_USAGE_BLOOM_NAME_PATH)
         .await
-        .map_or(Vec::new(), |buf| buf);
+        .unwrap_or_else(|err| {
-
+            info!(error = %err, "Failed to read cycle info, starting fresh");
-    let mut buf_t = Deserializer::new(Cursor::new(buf));
+            Vec::new()
-
-    let mut cycle_info: CurrentScannerCycle = Deserialize::deserialize(&mut buf_t).unwrap_or_default();
-
-    loop {
-        let stop_fn = ScannerMetrics::log(ScannerMetric::ScanCycle);
-        cycle_info.current = cycle_info.next;
-        cycle_info.started = Utc::now();
-        {
-            globalScannerMetrics.write().await.set_cycle(Some(cycle_info.clone())).await;
-        }
-
-        let bg_heal_info = read_background_heal_info(store.clone()).await;
-        let scan_mode =
-            get_cycle_scan_mode(cycle_info.current, bg_heal_info.bitrot_start_cycle, bg_heal_info.bitrot_start_time).await;
-        if bg_heal_info.current_scan_mode != scan_mode {
-            let mut new_heal_info = bg_heal_info;
-            new_heal_info.current_scan_mode = scan_mode;
-            if scan_mode == HEAL_DEEP_SCAN {
-                new_heal_info.bitrot_start_time = SystemTime::now();
-                new_heal_info.bitrot_start_cycle = cycle_info.current;
-            }
-            save_background_heal_info(store.clone(), &new_heal_info).await;
-        }
-        // Wait before starting next cycle and wait on startup.
-        let (tx, rx) = mpsc::channel(100);
-        tokio::spawn(async {
-            store_data_usage_in_backend(rx).await;
         });
-        let mut res = HashMap::new();
+
-        res.insert("cycle".to_string(), cycle_info.current.to_string());
+    let mut cycle_info = if buf.is_empty() {
-        info!("start ns_scanner");
+        CurrentScannerCycle::default()
-        match store.clone().ns_scanner(tx, cycle_info.current as usize, scan_mode).await {
+    } else {
-            Ok(_) => {
+        let mut buf_cursor = Deserializer::new(Cursor::new(buf));
-                info!("ns_scanner completed");
+        Deserialize::deserialize(&mut buf_cursor).unwrap_or_else(|err| {
-                cycle_info.next += 1;
+            error!(error = %err, "Failed to deserialize cycle info, using default");
-                cycle_info.current = 0;
+            CurrentScannerCycle::default()
-                cycle_info.cycle_completed.push(Utc::now());
+        })
-                if cycle_info.cycle_completed.len() > DATA_USAGE_UPDATE_DIR_CYCLES as usize {
+    };
-                    let _ = cycle_info.cycle_completed.remove(0);
+
-                }
+    // Start metrics collection for this cycle
-                globalScannerMetrics.write().await.set_cycle(Some(cycle_info.clone())).await;
+    let stop_fn = ScannerMetrics::log(ScannerMetric::ScanCycle);
-                let mut wr = Vec::new();
+
-                cycle_info.serialize(&mut Serializer::new(&mut wr)).unwrap();
+    // Update cycle information
-                let _ = save_config(store.clone(), &DATA_USAGE_BLOOM_NAME_PATH, wr).await;
+    cycle_info.current = cycle_info.next;
+    cycle_info.started = Utc::now();
+
+    // Update global scanner metrics
+    {
+        globalScannerMetrics.write().await.set_cycle(Some(cycle_info.clone())).await;
+    }
+
+    // Read background healing information and determine scan mode
+    let bg_heal_info = read_background_heal_info(store.clone()).await;
+    let scan_mode =
+        get_cycle_scan_mode(cycle_info.current, bg_heal_info.bitrot_start_cycle, bg_heal_info.bitrot_start_time).await;
+
+    // Update healing info if scan mode changed
+    if bg_heal_info.current_scan_mode != scan_mode {
+        let mut new_heal_info = bg_heal_info;
+        new_heal_info.current_scan_mode = scan_mode;
+        if scan_mode == HEAL_DEEP_SCAN {
+            new_heal_info.bitrot_start_time = SystemTime::now();
+            new_heal_info.bitrot_start_cycle = cycle_info.current;
+        }
+        save_background_heal_info(store.clone(), &new_heal_info).await;
+    }
+
+    // Set up data usage storage channel
+    let (tx, rx) = mpsc::channel(100);
+    tokio::spawn(async move {
+        let _ = store_data_usage_in_backend(rx).await;
+    });
+
+    // Prepare result tracking
+    let mut scan_result = HashMap::new();
+    scan_result.insert("cycle".to_string(), cycle_info.current.to_string());
+
+    info!(
+        cycle = cycle_info.current,
+        scan_mode = ?scan_mode,
+        "Starting namespace scanner"
+    );
+
+    // Run the namespace scanner
+    match store.clone().ns_scanner(tx, cycle_info.current as usize, scan_mode).await {
+        Ok(_) => {
+            info!(cycle = cycle_info.current, "Namespace scanner completed successfully");
+
+            // Update cycle completion information
+            cycle_info.next += 1;
+            cycle_info.current = 0;
+            cycle_info.cycle_completed.push(Utc::now());
+
+            // Maintain cycle completion history (keep only recent cycles)
+            if cycle_info.cycle_completed.len() > DATA_USAGE_UPDATE_DIR_CYCLES as usize {
+                let _ = cycle_info.cycle_completed.remove(0);
             }
-            Err(err) => {
+
-                info!("ns_scanner failed: {:?}", err);
+            // Update global metrics with completion info
-                res.insert("error".to_string(), err.to_string());
+            globalScannerMetrics.write().await.set_cycle(Some(cycle_info.clone())).await;
+
+            // Persist updated cycle information
+            // ignore error, continue.
+            let mut serialized_data = Vec::new();
+            if let Err(err) = cycle_info.serialize(&mut Serializer::new(&mut serialized_data)) {
+                error!(error = %err, "Failed to serialize cycle info");
+            } else if let Err(err) = save_config(store.clone(), &DATA_USAGE_BLOOM_NAME_PATH, serialized_data).await {
+                error!(error = %err, "Failed to save cycle info to storage");
             }
         }
-        stop_fn(&res).await;
+        Err(err) => {
-        sleep(Duration::from_secs(SCANNER_CYCLE.load(Ordering::SeqCst))).await;
+            error!(
+                cycle = cycle_info.current,
+                error = %err,
+                "Namespace scanner failed"
+            );
+            scan_result.insert("error".to_string(), err.to_string());
+        }
     }
+
+    // Complete metrics collection for this cycle
+    stop_fn(&scan_result).await;
 }
 
 #[derive(Debug, Serialize, Deserialize)]