From 581eb6641c29de3fbdff9c05707eba688053cf9b Mon Sep 17 00:00:00 2001 From: junxiang Mu <1948535941@qq.com> Date: Wed, 28 May 2025 07:36:03 +0000 Subject: [PATCH] improve run_data_scanner Signed-off-by: junxiang Mu <1948535941@qq.com> --- ecstore/src/heal/data_scanner.rs | 206 ++++++++++++++++++++++--------- 1 file changed, 145 insertions(+), 61 deletions(-) diff --git a/ecstore/src/heal/data_scanner.rs b/ecstore/src/heal/data_scanner.rs index 627f044e6..0e3f21074 100644 --- a/ecstore/src/heal/data_scanner.rs +++ b/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(); - r.gen_range(0.0..1.0) + + // Calculate randomized sleep duration + // 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 sleep_duration = if duration < Duration::new(1, 0) { - Duration::new(1, 0) + let base_cycle_duration = SCANNER_CYCLE.load(Ordering::SeqCst) as f64; + let sleep_duration_secs = random_factor * base_cycle_duration; + + // 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); - - let mut buf_t = Deserializer::new(Cursor::new(buf)); - - 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; + .unwrap_or_else(|err| { + info!(error = %err, "Failed to read cycle info, starting fresh"); + Vec::new() }); - let mut res = HashMap::new(); - res.insert("cycle".to_string(), cycle_info.current.to_string()); - info!("start ns_scanner"); - match store.clone().ns_scanner(tx, cycle_info.current as usize, scan_mode).await { - Ok(_) => { - info!("ns_scanner completed"); - cycle_info.next += 1; - cycle_info.current = 0; - 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); - } - globalScannerMetrics.write().await.set_cycle(Some(cycle_info.clone())).await; - let mut wr = Vec::new(); - cycle_info.serialize(&mut Serializer::new(&mut wr)).unwrap(); - let _ = save_config(store.clone(), &DATA_USAGE_BLOOM_NAME_PATH, wr).await; + + let mut cycle_info = if buf.is_empty() { + CurrentScannerCycle::default() + } else { + let mut buf_cursor = Deserializer::new(Cursor::new(buf)); + Deserialize::deserialize(&mut buf_cursor).unwrap_or_else(|err| { + error!(error = %err, "Failed to deserialize cycle info, using default"); + CurrentScannerCycle::default() + }) + }; + + // Start metrics collection for this cycle + let stop_fn = ScannerMetrics::log(ScannerMetric::ScanCycle); + + // Update cycle information + 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); - res.insert("error".to_string(), err.to_string()); + + // Update global metrics with completion info + 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; - sleep(Duration::from_secs(SCANNER_CYCLE.load(Ordering::SeqCst))).await; + Err(err) => { + 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)]