From c35de00e8a0551b349668c403369c5a2406c854b Mon Sep 17 00:00:00 2001 From: Hauser Date: Sat, 19 Sep 2026 19:32:02 +0800 Subject: [PATCH] fix(scanner): protect single-disk foreground latency (#8022) Co-authored-by: zhi22915 --- crates/scanner/src/remote_scanner/stream.rs | 13 +++-- crates/scanner/src/scanner.rs | 4 +- crates/scanner/src/scanner/heal_info.rs | 15 ----- crates/scanner/src/scanner_folder.rs | 8 +++ crates/scanner/src/scanner_io.rs | 4 +- crates/scanner/src/scanner_io/cache.rs | 65 ++++++++++++++++++--- crates/scanner/src/scanner_io/io_cache.rs | 9 ++- 7 files changed, 82 insertions(+), 36 deletions(-) diff --git a/crates/scanner/src/remote_scanner/stream.rs b/crates/scanner/src/remote_scanner/stream.rs index f35af52dd..df8c1deba 100644 --- a/crates/scanner/src/remote_scanner/stream.rs +++ b/crates/scanner/src/remote_scanner/stream.rs @@ -16,8 +16,8 @@ use crate::RUSTFS_META_BUCKET; use crate::scanner_budget::{ScannerCycleBudget, ScannerCycleBudgetConfig}; use crate::scanner_io::{ - DataUsageCacheReuseOptions, DataUsageCacheScanState, ScannerCheckpointPersistResult, ScannerDiskScanOptions, - ScannerDiskScanOutcome, ScannerIODisk, acquire_scanner_cache_locks, cache_root_entry_info, + DataUsageCacheReuseOptions, DataUsageCacheScanState, ScannerCheckpointPersistContext, ScannerCheckpointPersistResult, + ScannerDiskScanOptions, ScannerDiskScanOutcome, ScannerIODisk, acquire_scanner_cache_locks, cache_root_entry_info, current_cache_root_or_prepare_with_generation, persist_scanner_checkpoint, scanner_set_disk_inventory, }; use crate::storage_api::owner::NS_SCANNER_PROTOCOL_VERSION; @@ -840,12 +840,15 @@ async fn scan_and_persist_local_bucket( } match persist_scanner_checkpoint( set.clone(), + ScannerCheckpointPersistContext { + ctx: &scan_ctx, + expected_publication_epoch, + cycle: next_cycle, + leader_epoch, + }, &cache_name, &checkpoint, &mut revisions, - expected_publication_epoch, - next_cycle, - leader_epoch, ) .await { diff --git a/crates/scanner/src/scanner.rs b/crates/scanner/src/scanner.rs index ea358cadb..fcd313ea9 100644 --- a/crates/scanner/src/scanner.rs +++ b/crates/scanner/src/scanner.rs @@ -1778,9 +1778,7 @@ where mark_scan_cycle_idle(cycle_info, &mut cycle_metrics_guard).await; return ScannerCycleOutcome::Failed; } - BackgroundHealInfoReadStatus::ErasureSd - | BackgroundHealInfoReadStatus::Loaded - | BackgroundHealInfoReadStatus::Missing => {} + BackgroundHealInfoReadStatus::Loaded | BackgroundHealInfoReadStatus::Missing => {} } let mut background_heal_info = background_heal_read.info; let background_heal_epoch = background_heal_read.expected_epoch; diff --git a/crates/scanner/src/scanner/heal_info.rs b/crates/scanner/src/scanner/heal_info.rs index f7e3f61f9..5f898d44a 100644 --- a/crates/scanner/src/scanner/heal_info.rs +++ b/crates/scanner/src/scanner/heal_info.rs @@ -28,7 +28,6 @@ pub struct BackgroundHealInfo { #[derive(Clone, Copy, Debug, Eq, PartialEq)] pub(super) enum BackgroundHealInfoReadStatus { - ErasureSd, Loaded, Missing, Blocked, @@ -66,15 +65,6 @@ pub(super) async fn read_background_heal_info_with_epoch(storeapi: Arc) -> where S: ScannerStorage, { - // Skip for ErasureSD setup - if storeapi.setup_is_erasure_sd().await { - return BackgroundHealInfoRead { - info: BackgroundHealInfo::default(), - expected_epoch: None, - status: BackgroundHealInfoReadStatus::ErasureSd, - }; - } - let expected_epoch = scanner_publication_epoch(storeapi.clone()).await; if expected_epoch.is_none() { return BackgroundHealInfoRead { @@ -146,11 +136,6 @@ pub(super) async fn save_background_heal_info_for_epoch( ) where S: ScannerStorage, { - // Skip for ErasureSD setup - if storeapi.setup_is_erasure_sd().await { - return; - } - // Serialize to JSON let data = match serde_json::to_vec(&info) { Ok(data) => data, diff --git a/crates/scanner/src/scanner_folder.rs b/crates/scanner/src/scanner_folder.rs index eedf06e4a..ac7c1abdb 100644 --- a/crates/scanner/src/scanner_folder.rs +++ b/crates/scanner/src/scanner_folder.rs @@ -1806,6 +1806,14 @@ impl FolderScanner { continue; } + // Do not start another metadata read while foreground work is + // active. The post-object timer protects the next request only + // after the read has already been dispatched; this admission + // point keeps the scanner from extending a single-disk I/O + // burst across foreground requests. + if crate::workload_admission::foreground_workload_activity() > 0 { + self.sleeper.sleep_folder().await; + } let timer = self.sleeper.timer(); let heal_enabled = this_hash.mod_alt( diff --git a/crates/scanner/src/scanner_io.rs b/crates/scanner/src/scanner_io.rs index e0374b4af..0ff8b2dfc 100644 --- a/crates/scanner/src/scanner_io.rs +++ b/crates/scanner/src/scanner_io.rs @@ -1609,8 +1609,8 @@ use dirty_usage::*; use guards::*; pub(crate) use cache::{ - DataUsageCacheReuseOptions, DataUsageCacheScanState, ScannerCheckpointPersistResult, acquire_scanner_cache_locks, - current_cache_root_or_prepare_with_generation, persist_scanner_checkpoint, + DataUsageCacheReuseOptions, DataUsageCacheScanState, ScannerCheckpointPersistContext, ScannerCheckpointPersistResult, + acquire_scanner_cache_locks, current_cache_root_or_prepare_with_generation, persist_scanner_checkpoint, }; pub use dirty_usage::{ ScannerDirtyUsageAckError, ScannerDirtyUsageBucket, ScannerDirtyUsageClearObserver, ScannerDirtyUsageMutationObserver, diff --git a/crates/scanner/src/scanner_io/cache.rs b/crates/scanner/src/scanner_io/cache.rs index 9fc8a9004..fe4b6f36b 100644 --- a/crates/scanner/src/scanner_io/cache.rs +++ b/crates/scanner/src/scanner_io/cache.rs @@ -31,6 +31,41 @@ pub(crate) enum ScannerCheckpointPersistResult { Failed(StorageError), } +pub(crate) struct ScannerCheckpointPersistContext<'a> { + pub(crate) ctx: &'a CancellationToken, + pub(crate) expected_publication_epoch: u64, + pub(crate) cycle: u64, + pub(crate) leader_epoch: u64, +} + +const CHECKPOINT_FOREGROUND_QUIET_WAIT: Duration = Duration::from_secs(1); + +async fn wait_for_checkpoint_foreground_quiet(ctx: &CancellationToken) -> bool { + let deadline = tokio::time::Instant::now() + CHECKPOINT_FOREGROUND_QUIET_WAIT; + loop { + if crate::workload_admission::foreground_workload_activity() == 0 { + return true; + } + + let remaining = deadline.saturating_duration_since(tokio::time::Instant::now()); + if remaining.is_zero() { + return false; + } + + let backoff = Duration::from_millis( + crate::workload_admission::foreground_workload_activity() + .saturating_mul(10) + .min(250), + ) + .max(Duration::from_millis(10)) + .min(remaining); + tokio::select! { + _ = ctx.cancelled() => return false, + _ = tokio::time::sleep(backoff) => {} + } + } +} + /// Persist one bounded checkpoint and refresh its CAS revisions. /// /// Local and remote workers share the same publication/leader fencing and @@ -38,23 +73,37 @@ pub(crate) enum ScannerCheckpointPersistResult { /// at the caller because those guards have different concrete types. pub(crate) async fn persist_scanner_checkpoint( store: Arc, + context: ScannerCheckpointPersistContext<'_>, cache_name: &str, checkpoint: &DataUsageCache, revisions: &mut DataUsageCacheRevisions, - expected_publication_epoch: u64, - cycle: u64, - leader_epoch: u64, ) -> ScannerCheckpointPersistResult where S: ScannerObjectIO + ScannerConfigObjectDelete, { - if crate::remote_scanner::validate_remote_scanner_request_fence_with_store(cycle, leader_epoch, store.clone()) + let foreground_quiet = wait_for_checkpoint_foreground_quiet(context.ctx).await; + if !foreground_quiet && context.ctx.is_cancelled() { + return ScannerCheckpointPersistResult::FenceChanged; + } + if !foreground_quiet { + debug!( + target: "rustfs::scanner::io", + event = EVENT_SCANNER_CACHE_PERSIST_STATE, + component = LOG_COMPONENT_SCANNER, + subsystem = LOG_SUBSYSTEM_IO, + cache_name, + state = "checkpoint_foreground_wait_expired", + "Scanner checkpoint foreground quiet wait expired; preserving bounded progress" + ); + } + + if crate::remote_scanner::validate_remote_scanner_request_fence_with_store(context.cycle, context.leader_epoch, store.clone()) .await .is_err() { return ScannerCheckpointPersistResult::FenceChanged; } - if scanner_publication_admission_for_epoch(store.clone(), expected_publication_epoch) + if scanner_publication_admission_for_epoch(store.clone(), context.expected_publication_epoch) .await .is_none() { @@ -62,16 +111,16 @@ where } if let Err(error) = checkpoint - .save_with_revisions_for_epoch(store.clone(), cache_name, revisions, expected_publication_epoch) + .save_with_revisions_for_epoch(store.clone(), cache_name, revisions, context.expected_publication_epoch) .await { return ScannerCheckpointPersistResult::Failed(error); } - if crate::remote_scanner::validate_remote_scanner_request_fence_with_store(cycle, leader_epoch, store.clone()) + if crate::remote_scanner::validate_remote_scanner_request_fence_with_store(context.cycle, context.leader_epoch, store.clone()) .await .is_err() - || scanner_publication_admission_for_epoch(store.clone(), expected_publication_epoch) + || scanner_publication_admission_for_epoch(store.clone(), context.expected_publication_epoch) .await .is_none() { diff --git a/crates/scanner/src/scanner_io/io_cache.rs b/crates/scanner/src/scanner_io/io_cache.rs index c3b190f7d..e64b1965e 100644 --- a/crates/scanner/src/scanner_io/io_cache.rs +++ b/crates/scanner/src/scanner_io/io_cache.rs @@ -1191,12 +1191,15 @@ impl ScannerIOCache for SetDisks { } match persist_scanner_checkpoint( store_clone_clone.clone(), + ScannerCheckpointPersistContext { + ctx: &ctx_clone, + expected_publication_epoch: expected_publication_epoch_clone, + cycle: want_cycle, + leader_epoch, + }, cache_name.as_str(), &checkpoint, &mut revisions, - expected_publication_epoch_clone, - want_cycle, - leader_epoch, ) .await {