mirror of
https://github.com/rustfs/rustfs.git
synced 2026-07-26 16:28:15 +00:00
improve run_data_scanner
Signed-off-by: junxiang Mu <1948535941@qq.com>
This commit is contained in:
junxiang Mu
@@ -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)]
|
||||
|
||||
Reference in New Issue
Block a user