From c8cb1d86d338481b5413e8d58fe06bccbdd00219 Mon Sep 17 00:00:00 2001 From: Hauser Date: Fri, 23 Oct 2026 20:07:09 +0800 Subject: [PATCH] fix(scanner): streamline budgeted metadata walks (#8837) * fix(scanner): budget directory enumeration I/O Co-Authored-By: heihutu Co-Authored-By: zhi22915 * refactor(scanner): skip redundant metadata walk work Co-Authored-By: heihutu Co-Authored-By: zhi22915 * refactor(scanner): share budgeted directory traversal Co-Authored-By: heihutu Co-Authored-By: zhi22915 --------- Co-authored-by: zhi22915 --- crates/scanner/src/scanner_folder.rs | 69 +++-- .../src/scanner_folder/item_actions.rs | 19 +- crates/scanner/src/scanner_folder/tests.rs | 253 ++++++++++++++++++ crates/scanner/src/sleeper.rs | 19 +- crates/scanner/src/sleeper/io_budget.rs | 19 +- docs/operations/scanner-runtime-controls.md | 10 +- 6 files changed, 349 insertions(+), 40 deletions(-) diff --git a/crates/scanner/src/scanner_folder.rs b/crates/scanner/src/scanner_folder.rs index 7d1c054ab..e9fc06af9 100644 --- a/crates/scanner/src/scanner_folder.rs +++ b/crates/scanner/src/scanner_folder.rs @@ -912,6 +912,28 @@ fn prune_size_reconciliation(info: &mut DataUsageCacheInfo, now: u64) { } } +async fn wait_for_scanner_io(ctx: &CancellationToken, wait: impl std::future::Future) -> Result { + select! { + biased; + _ = ctx.cancelled() => Err(ScannerError::Other("Operation cancelled".to_string())), + result = wait => Ok(result), + } +} + +async fn next_scanner_dir_entry( + ctx: &CancellationToken, + sleeper: &DynamicSleeper, + entries: &mut tokio::fs::ReadDir, +) -> Result, ScannerError> { + let entry = wait_for_scanner_io(ctx, entries.next_entry()).await??; + if entry.is_some() { + // Tokio prefetches bounded chunks. Pacing consumption also paces the + // next refill, without charging an extra entry for EOF. + wait_for_scanner_io(ctx, sleeper.admit_disk_io(1, 0)).await?; + } + Ok(entry) +} + impl FolderScanner { fn now_secs() -> u64 { SystemTime::now() @@ -1683,7 +1705,7 @@ impl FolderScanner { }; let active_object_lock = self.old_cache.info.object_lock.clone(); - self.sleeper.sleep_folder().await; + wait_for_scanner_io(&ctx, self.sleeper.sleep_folder()).await?; let mut existing_folders: Vec = Vec::new(); let mut new_folders: Vec = Vec::new(); @@ -1709,7 +1731,7 @@ impl FolderScanner { // directory that vanished during iteration instead of failing the // whole parent subtree. The bucket root (no parent) still propagates // NotFound so namespace removal is detected by the caller. - let mut dir_reader = match tokio::fs::read_dir(&dir_path).await { + let mut dir_reader = match wait_for_scanner_io(&ctx, tokio::fs::read_dir(&dir_path)).await? { Ok(dir_reader) => Some(dir_reader), Err(e) if e.kind() == ErrorKind::NotFound && folder.parent.is_some() => { debug!( @@ -1736,13 +1758,13 @@ impl FolderScanner { raw_enumeration_complete = true; break; }; - let entry = match reader.next_entry().await { + let entry = match next_scanner_dir_entry(&ctx, &self.sleeper, reader).await { Ok(Some(entry)) => entry, Ok(None) => { raw_enumeration_complete = true; break; } - Err(e) if e.kind() == ErrorKind::NotFound => { + Err(ScannerError::Io(e)) if e.kind() == ErrorKind::NotFound => { debug!( target: "rustfs::scanner::folder", event = EVENT_SCANNER_FOLDER_STATE, @@ -1756,7 +1778,7 @@ impl FolderScanner { raw_enumeration_complete = true; break; } - Err(e) if e.kind() == ErrorKind::NotADirectory => { + Err(ScannerError::Io(e)) if e.kind() == ErrorKind::NotADirectory => { warn!( target: "rustfs::scanner::folder", event = EVENT_SCANNER_FOLDER_STATE, @@ -1770,7 +1792,7 @@ impl FolderScanner { raw_enumeration_complete = true; break; } - Err(e) => return Err(ScannerError::Io(e)), + Err(e) => return Err(e), }; let file_name = entry.file_name().to_string_lossy().to_string(); if file_name.is_empty() || file_name == "." || file_name == ".." { @@ -1805,7 +1827,7 @@ impl FolderScanner { // Ignore entries that disappeared during traversal or hit symlink // loops, but propagate other walk errors. - let mut entry_type = match entry.file_type().await { + let mut entry_type = match wait_for_scanner_io(&ctx, entry.file_type()).await? { Ok(entry_type) => entry_type, Err(e) if e.kind() == ErrorKind::NotFound => { debug!( @@ -1844,8 +1866,10 @@ impl FolderScanner { found_object_metadata = true; } + let mut entry_metadata_len = None; if entry_type.is_symlink() { - let metadata = match tokio::fs::metadata(&file_path).await { + wait_for_scanner_io(&ctx, self.sleeper.admit_disk_io(1, 0)).await?; + let metadata = match wait_for_scanner_io(&ctx, tokio::fs::metadata(&file_path)).await? { Ok(metadata) => metadata, Err(e) if e.kind() == ErrorKind::NotFound => { debug!( @@ -1890,6 +1914,7 @@ impl FolderScanner { } entry_type = metadata.file_type(); + entry_metadata_len = Some(metadata.len()); if is_storage_format_entry { found_object_metadata = true; } @@ -1945,6 +1970,12 @@ impl FolderScanner { continue; } + // Only xl.meta contributes object accounting. Payload files + // must not incur a size stat or a metadata-read reservation. + if !is_storage_format_entry { + continue; + } + let timer = self.sleeper.timer(); let heal_enabled = this_hash.mod_alt( @@ -1978,18 +2009,22 @@ impl FolderScanner { continue; } - // Charge the disk's background I/O budget before the metadata - // read. The `xl.meta` size is only measured when bytes are - // budgeted, so an uncapped byte axis costs no extra stat. + // Reuse a resolved symlink's size; otherwise admit the size + // stat only when the byte budget needs it. let budgeted_bytes = if self.sleeper.budgets_disk_io_bytes() { - tokio::fs::metadata(&item.path) - .await - .map(|metadata| metadata.len()) - .unwrap_or(0) + if let Some(len) = entry_metadata_len { + len + } else { + wait_for_scanner_io(&ctx, self.sleeper.admit_disk_io(1, 0)).await?; + wait_for_scanner_io(&ctx, tokio::fs::metadata(&item.path)) + .await? + .map(|metadata| metadata.len()) + .unwrap_or(0) + } } else { 0 }; - self.sleeper.admit_disk_io(1, budgeted_bytes).await; + wait_for_scanner_io(&ctx, self.sleeper.admit_disk_io(1, budgeted_bytes)).await?; let sz = match self .local_disk @@ -2132,7 +2167,7 @@ impl FolderScanner { let mut found_erasure_data_directory = false; if self.is_erasure_mode && !found_object_metadata { for (_, _, path) in &erasure_data_directory_candidates { - if contains_erasure_part_file(path).await? { + if contains_erasure_part_file(path, &ctx, &self.sleeper).await? { found_erasure_data_directory = true; break; } diff --git a/crates/scanner/src/scanner_folder/item_actions.rs b/crates/scanner/src/scanner_folder/item_actions.rs index d205bf1ee..bb866bd8d 100644 --- a/crates/scanner/src/scanner_folder/item_actions.rs +++ b/crates/scanner/src/scanner_folder/item_actions.rs @@ -1296,19 +1296,26 @@ pub(super) fn classify_get_size_failure(item: &ScannerItem, err: &StorageError) GetSizeFailureAction::RecordFailed } -pub(super) async fn contains_erasure_part_file(path: &str) -> Result { - let mut entries = match tokio::fs::read_dir(path).await { +pub(super) async fn contains_erasure_part_file( + path: &str, + ctx: &CancellationToken, + sleeper: &DynamicSleeper, +) -> Result { + wait_for_scanner_io(ctx, sleeper.admit_disk_io(1, 0)).await?; + let mut entries = match wait_for_scanner_io(ctx, tokio::fs::read_dir(path)).await? { Ok(entries) => entries, Err(err) if matches!(err.kind(), ErrorKind::NotFound | ErrorKind::NotADirectory) => return Ok(false), Err(err) => return Err(ScannerError::Io(err)), }; for _ in 0..ERASURE_DATA_DIR_PROBE_ENTRY_LIMIT { - let entry = match entries.next_entry().await { + let entry = match next_scanner_dir_entry(ctx, sleeper, &mut entries).await { Ok(Some(entry)) => entry, Ok(None) => return Ok(false), - Err(err) if matches!(err.kind(), ErrorKind::NotFound | ErrorKind::NotADirectory) => return Ok(false), - Err(err) => return Err(ScannerError::Io(err)), + Err(ScannerError::Io(err)) if matches!(err.kind(), ErrorKind::NotFound | ErrorKind::NotADirectory) => { + return Ok(false); + } + Err(err) => return Err(err), }; let file_name = entry.file_name(); let Some(part_number) = file_name @@ -1322,7 +1329,7 @@ pub(super) async fn contains_erasure_part_file(path: &str) -> Result return Ok(true), Ok(_) => {} Err(err) if matches!(err.kind(), ErrorKind::NotFound | ErrorKind::TooManyLinks) => {} diff --git a/crates/scanner/src/scanner_folder/tests.rs b/crates/scanner/src/scanner_folder/tests.rs index e10d28287..f45888efd 100644 --- a/crates/scanner/src/scanner_folder/tests.rs +++ b/crates/scanner/src/scanner_folder/tests.rs @@ -2314,6 +2314,247 @@ async fn test_scan_folder_treats_child_removed_before_open_as_empty() { assert_eq!(into.size, 0); } +#[cfg(unix)] +#[tokio::test] +#[serial] +async fn scan_folder_charges_each_enumerated_entry_to_the_disk_budget() { + let (mut scanner, temp_dir) = build_test_scanner().await; + let _guard = TestGuard::new(60, 100, &mut scanner, temp_dir.clone()); + let bucket_dir = temp_dir.join("bucket"); + tokio::fs::create_dir_all(&bucket_dir) + .await + .expect("failed to create bucket directory"); + for name in ["one", "two", "three"] { + symlink("missing-target", bucket_dir.join(name)).expect("failed to create dangling symlink"); + } + + scanner.sleeper = SCANNER_SLEEPER.for_disk(&temp_dir); + scanner.old_cache.info.name = "bucket".to_string(); + scanner.new_cache.info.name = "bucket".to_string(); + scanner.update_cache.info.name = "bucket".to_string(); + + let folder = CachedFolder { + name: "bucket".to_string(), + parent: None, + object_heal_prob_div: 1, + }; + let mut into = DataUsageEntry::default(); + scanner + .scan_folder(CancellationToken::new(), folder, &mut into) + .await + .expect("directory enumeration should complete"); + + let disk = temp_dir.to_string_lossy(); + let status = crate::sleeper::scanner_disk_io_budget_status() + .into_iter() + .find(|status| status.disk == disk) + .expect("scanner disk budget should expose its charge count"); + assert_eq!(status.charged_ops, 7, "one directory open, three entries, and three target stats"); +} + +#[tokio::test] +#[serial] +async fn scan_folder_budgets_payload_entries_without_metadata_io() { + let (mut scanner, temp_dir) = build_test_scanner().await; + let _guard = TestGuard::new(60, 100, &mut scanner, temp_dir.clone()); + let bucket_dir = temp_dir.join("bucket"); + tokio::fs::create_dir_all(&bucket_dir).await.expect("create bucket"); + for index in 0..33 { + tokio::fs::write(bucket_dir.join(format!("part.{index}")), b"payload") + .await + .expect("create payload entry"); + } + scanner.sleeper = DynamicSleeper::new(rustfs_config::ScannerSpeed::Fastest).for_disk_with_budget_for_tests( + &temp_dir, + crate::sleeper::ScannerDiskIoBudgetLimits { + iops: Some(1_000_000), + bytes_per_sec: Some(1_000_000), + }, + ); + let mut into = DataUsageEntry::default(); + scanner + .scan_folder( + CancellationToken::new(), + CachedFolder { + name: "bucket".to_string(), + parent: None, + object_heal_prob_div: 1, + }, + &mut into, + ) + .await + .expect("scan payload entries across a directory prefetch boundary"); + + let status = crate::sleeper::scanner_disk_io_budget_status() + .into_iter() + .find(|status| status.disk == temp_dir.to_string_lossy()) + .expect("bound disk budget"); + assert_eq!(status.charged_ops, 34, "one directory open and 33 entries, without size stats or reads"); + assert_eq!(status.charged_bytes, 0); + assert_eq!(into.objects, 0); + assert_eq!(into.failed_objects, 0); +} + +#[cfg(unix)] +#[tokio::test] +#[serial] +async fn scan_folder_reuses_symlink_size_without_bypassing_disk_validation() { + let (mut scanner, temp_dir) = build_test_scanner().await; + let _guard = TestGuard::new(60, 100, &mut scanner, temp_dir.clone()); + let object_dir = temp_dir.join("bucket/object"); + tokio::fs::create_dir_all(&object_dir).await.expect("create object directory"); + let metadata = metadata_for_object("bucket", "object"); + let target = temp_dir.join("metadata-target"); + tokio::fs::write(&target, &metadata).await.expect("write metadata target"); + symlink(&target, object_dir.join(STORAGE_FORMAT_FILE)).expect("create metadata symlink"); + scanner.sleeper = DynamicSleeper::new(rustfs_config::ScannerSpeed::Fastest).for_disk_with_budget_for_tests( + &temp_dir, + crate::sleeper::ScannerDiskIoBudgetLimits { + iops: Some(1_000_000), + bytes_per_sec: Some(1_000_000), + }, + ); + let mut into = DataUsageEntry::default(); + scanner + .scan_folder( + CancellationToken::new(), + CachedFolder { + name: "bucket/object".to_string(), + parent: None, + object_heal_prob_div: 1, + }, + &mut into, + ) + .await + .expect("scan symlinked object metadata"); + + let status = crate::sleeper::scanner_disk_io_budget_status() + .into_iter() + .find(|status| status.disk == temp_dir.to_string_lossy()) + .expect("bound disk budget"); + assert_eq!( + status.charged_ops, 4, + "directory, entry, target stat, and read reservation without a second stat" + ); + assert_eq!(status.charged_bytes, u64::try_from(metadata.len()).expect("metadata length fits u64")); + assert_eq!(into.objects, 0, "the disk must continue rejecting symlinked metadata"); + assert_eq!(into.failed_objects, 1); + assert!( + scanner + .new_cache + .info + .failed_objects + .contains_key(object_dir.join(STORAGE_FORMAT_FILE).to_string_lossy().as_ref()) + ); +} + +#[tokio::test(start_paused = true)] +async fn scanner_directory_probe_budget_wait_cancels_without_rate_delay() { + let path = std::env::temp_dir().join(format!("scanner-cancel-budget-{}", Uuid::new_v4())); + let sleeper = DynamicSleeper::new(rustfs_config::ScannerSpeed::Fastest).for_disk_with_budget_for_tests( + &path, + crate::sleeper::ScannerDiskIoBudgetLimits { + iops: Some(0), + bytes_per_sec: None, + }, + ); + sleeper.admit_disk_io(1, 0).await; + let ctx = CancellationToken::new(); + let cancel = ctx.clone(); + let task = tokio::spawn(async move { + tokio::task::yield_now().await; + cancel.cancel(); + }); + let started = tokio::time::Instant::now(); + assert!( + contains_erasure_part_file(path.to_string_lossy().as_ref(), &ctx, &sleeper) + .await + .is_err() + ); + task.await.expect("cancellation task"); + assert_eq!( + started.elapsed(), + Duration::ZERO, + "cancellation must interrupt the one-second budget wait" + ); + + assert!(wait_for_scanner_io(&ctx, sleeper.admit_disk_io(1, 0)).await.is_err()); + let status = crate::sleeper::scanner_disk_io_budget_status() + .into_iter() + .find(|status| status.disk == path.to_string_lossy()) + .expect("bound disk budget"); + assert_eq!(status.charged_ops, 2, "an already cancelled wait must not charge another operation"); +} + +#[tokio::test] +async fn erasure_directory_probe_respects_entry_limit_and_disk_budget() { + let temp_dir = tempfile::tempdir().expect("create probe directory"); + for index in 0..=ERASURE_DATA_DIR_PROBE_ENTRY_LIMIT { + tokio::fs::write(temp_dir.path().join(format!("namespace-{index}")), b"") + .await + .expect("create non-part entry"); + } + let sleeper = DynamicSleeper::new(rustfs_config::ScannerSpeed::Fastest).for_disk_with_budget_for_tests( + temp_dir.path(), + crate::sleeper::ScannerDiskIoBudgetLimits { + iops: Some(1_000_000), + bytes_per_sec: None, + }, + ); + assert!( + !contains_erasure_part_file(temp_dir.path().to_string_lossy().as_ref(), &CancellationToken::new(), &sleeper) + .await + .expect("probe directory without erasure parts") + ); + let status = crate::sleeper::scanner_disk_io_budget_status() + .into_iter() + .find(|status| status.disk == temp_dir.path().to_string_lossy()) + .expect("bound disk budget"); + assert_eq!( + status.charged_ops, + 1 + u64::try_from(ERASURE_DATA_DIR_PROBE_ENTRY_LIMIT).expect("entry limit fits u64"), + "one directory open plus at most the bounded probe entries" + ); + assert_eq!(status.charged_bytes, 0); +} + +#[tokio::test(start_paused = true)] +async fn scanner_directory_probe_cancels_during_entry_budget_wait() { + let temp_dir = tempfile::tempdir().expect("create probe directory"); + tokio::fs::write(temp_dir.path().join("part.1"), b"shard") + .await + .expect("create erasure part"); + let sleeper = DynamicSleeper::new(rustfs_config::ScannerSpeed::Fastest).for_disk_with_budget_for_tests( + temp_dir.path(), + crate::sleeper::ScannerDiskIoBudgetLimits { + iops: Some(0), + bytes_per_sec: None, + }, + ); + let ctx = CancellationToken::new(); + let cancel = ctx.clone(); + let disk = temp_dir.path().to_string_lossy().into_owned(); + let task = tokio::spawn(async move { + loop { + if crate::sleeper::scanner_disk_io_budget_status() + .iter() + .any(|status| status.disk == disk && status.charged_ops == 2) + { + cancel.cancel(); + return; + } + tokio::task::yield_now().await; + } + }); + let started = tokio::time::Instant::now(); + let result = contains_erasure_part_file(temp_dir.path().to_string_lossy().as_ref(), &ctx, &sleeper).await; + task.abort(); + assert!(ctx.is_cancelled(), "probe must reach the entry admission after opening its directory"); + task.await.expect("entry-wait cancellation task"); + assert!(result.is_err(), "cancelled probe must not report an erasure part"); + assert_eq!(started.elapsed(), Duration::ZERO, "entry admission must not wait out its one-second debt"); +} + #[tokio::test] #[serial] async fn test_scan_folder_missing_bucket_root_still_reports_not_found() { @@ -2681,6 +2922,13 @@ async fn test_scan_folder_missing_xl_meta_stops_erasure_data_dir_descent() { scanner.new_cache.info.name = "bucket".to_string(); scanner.update_cache.info.name = "bucket".to_string(); scanner.is_erasure_mode = true; + scanner.sleeper = DynamicSleeper::new(rustfs_config::ScannerSpeed::Fastest).for_disk_with_budget_for_tests( + &temp_dir, + crate::sleeper::ScannerDiskIoBudgetLimits { + iops: Some(1_000_000), + bytes_per_sec: None, + }, + ); let parent = CancellationToken::new(); let budget = ScannerCycleBudget::new_with_progress_tracking( @@ -2703,6 +2951,11 @@ async fn test_scan_folder_missing_xl_meta_stops_erasure_data_dir_descent() { .await .expect("missing metadata must not make scanner descend into erasure data directories"); + let disk_budget = crate::sleeper::scanner_disk_io_budget_status() + .into_iter() + .find(|status| status.disk == temp_dir.to_string_lossy()) + .expect("bound disk budget"); + assert_eq!(disk_budget.charged_ops, 4, "parent open/entry and part-directory probe open/entry"); assert_eq!(into.failed_objects, 1); assert!( scanner diff --git a/crates/scanner/src/sleeper.rs b/crates/scanner/src/sleeper.rs index 115b348eb..da35fa3e8 100644 --- a/crates/scanner/src/sleeper.rs +++ b/crates/scanner/src/sleeper.rs @@ -134,7 +134,7 @@ pub static SCANNER_SLEEPER: LazyLock = LazyLock::new(|| { /// For folder-level gaps, [`sleep_folder`] uses `MIN_SLEEP * factor`, /// clamped only by `max_sleep`. All sleeps are gated on [`SCANNER_IDLE_MODE`]; /// the per-disk background I/O budget bound with [`DynamicSleeper::for_disk`] -/// is a hard ceiling that applies regardless of idle mode or preset. +/// applies regardless of idle mode or preset. #[derive(Clone)] pub struct DynamicSleeper { inner: Arc, @@ -166,6 +166,15 @@ impl DynamicSleeper { } } + #[cfg(test)] + pub(crate) fn for_disk_with_budget_for_tests(&self, disk_path: &Path, limits: ScannerDiskIoBudgetLimits) -> Self { + let sleeper = self.for_disk(disk_path); + if let Some(budget) = &sleeper.disk_io_budget { + budget.reconfigure(limits); + } + sleeper + } + /// Whether byte charges can change pacing on the bound disk. pub fn budgets_disk_io_bytes(&self) -> bool { self.disk_io_budget.as_ref().is_some_and(|budget| budget.budgets_bytes()) @@ -173,9 +182,9 @@ impl DynamicSleeper { /// Charge the bound disk's background I/O budget and wait it out. /// - /// This is a hard ceiling layered on the local-disk foreground-priority - /// admission gate: it runs regardless of idle mode or speed preset, and - /// it only ever delays, so it can never admit more than the gate allows. + /// This pacing applies regardless of idle mode or speed preset and is + /// layered on the local-disk foreground-priority admission gate. It is a + /// scanner work budget, not a strict OS-level IOPS or bandwidth limit. pub async fn admit_disk_io(&self, ops: u64, bytes: u64) { if let Some(budget) = self.disk_io_budget.as_ref() { budget.admit(ops, bytes).await; @@ -477,7 +486,7 @@ mod tests { #[tokio::test(start_paused = true)] #[serial] - async fn sleeper_bound_to_a_disk_charges_folder_and_object_io_as_a_hard_ceiling() { + async fn sleeper_bound_to_a_disk_paces_folder_and_object_io() { let prev_mode = SCANNER_IDLE_MODE.load(Ordering::Relaxed); // Sleeps are disabled and the preset is `fastest`: only the explicit // budget can pace this walk. diff --git a/crates/scanner/src/sleeper/io_budget.rs b/crates/scanner/src/sleeper/io_budget.rs index ed4701bc2..ba8f7e2bc 100644 --- a/crates/scanner/src/sleeper/io_budget.rs +++ b/crates/scanner/src/sleeper/io_budget.rs @@ -15,13 +15,14 @@ //! Per-disk background I/O budget for the data scanner. //! //! Each local disk owns two token buckets, one for operations per second and -//! one for bytes per second. Every scanner directory listing and object -//! metadata read charges them before it runs; an exhausted bucket delays the -//! walk until the rate allows the charge again. The budget is a hard ceiling -//! layered on top of the local-disk foreground-priority admission gate: -//! admission decides *when* background I/O may start while foreground -//! requests are active, the budget bounds *how much* background I/O one disk -//! sees per second even while it is idle. A charge can never bypass the gate. +//! one for bytes per second. Directory opens, enumerated entries, scanner +//! metadata/stat operations, and metadata bytes charge the relevant buckets; +//! an exhausted bucket delays the walk. This paces scanner work on top of the +//! local-disk foreground-priority admission gate: admission decides *when* +//! background I/O may start while foreground requests are active, while the +//! budget slows scanner work even when the disk is idle. Bounded directory +//! prefetch and the maximum wait mean this is not a strict OS-level IOPS or +//! bandwidth limit. A charge can never bypass the admission gate. //! //! A zero or tiny budget still admits one charge per //! [`SCANNER_DISK_IO_BUDGET_MAX_WAIT`], so a walk keeps making progress @@ -307,7 +308,7 @@ impl ScannerDiskIoBudget { self.limits().bytes_per_sec.is_some() } - fn reconfigure(&self, limits: ScannerDiskIoBudgetLimits) { + pub(super) fn reconfigure(&self, limits: ScannerDiskIoBudgetLimits) { let now = Instant::now(); self.state .lock() @@ -337,7 +338,7 @@ impl ScannerDiskIoBudget { } /// Charge and wait out the budget. Runs regardless of idle mode or - /// speed preset: the budget is a ceiling, not a preset sleep. + /// speed preset, independently of the preset sleep. pub async fn admit(&self, ops: u64, bytes: u64) { let wait = self.charge(ops, bytes); if wait.is_zero() { diff --git a/docs/operations/scanner-runtime-controls.md b/docs/operations/scanner-runtime-controls.md index b53e5a76d..479607014 100644 --- a/docs/operations/scanner-runtime-controls.md +++ b/docs/operations/scanner-runtime-controls.md @@ -45,8 +45,8 @@ Bitrot cycle resolution differs because the canonical persistent key belongs to | `scanner.max_concurrent_set_scans` | `RUSTFS_SCANNER_MAX_CONCURRENT_SET_SCANS` | count | `4` (`DEFAULT_SCANNER_MAX_CONCURRENT_SET_SCANS`) | Caps concurrent set-level scanner tasks. `0` keeps topology-derived concurrency. | | `scanner.max_concurrent_disk_scans` | `RUSTFS_SCANNER_MAX_CONCURRENT_DISK_SCANS` | count | `4` (`DEFAULT_SCANNER_MAX_CONCURRENT_DISK_SCANS`) | Caps concurrent disk bucket walks per set. `0` keeps walker-count-derived concurrency. | | `scanner.walkers_per_disk` | `RUSTFS_SCANNER_WALKERS_PER_DISK` | count | `1` (`DEFAULT_SCANNER_WALKERS_PER_DISK`) | Bucket walkers spawned per local disk of a set. The walkers of one disk share its scanner admission, so a peer still sees one scan session per physical disk; remote disks keep one walker. The resulting walker count is what `scanner.max_concurrent_disk_scans` caps, and the disk count stays the floor. Valid range is `1` through `16` (`MAX_SCANNER_WALKERS_PER_DISK`); `1` keeps today's I/O profile and is the right value for rotational media. The `disk_concurrency_budget` log line reports `walkers_per_disk`, `walkers`, the effective `concurrency_limit`, and its `reason` (`available_walkers`, `configured_cap`, or `count_budget`). | -| `scanner.disk_io_budget_iops` | `RUSTFS_SCANNER_DISK_IO_BUDGET_IOPS` | operations/s | `auto` (`DEFAULT_SCANNER_DISK_IO_BUDGET`) | Per-local-disk ceiling on scanner directory listings and object metadata reads. `auto` picks the media default (`100` on rotational or unknown media, `4000` on flash); `off` removes the cap; a number is an explicit rate. See [Per-disk background I/O budget](#per-disk-background-io-budget). | -| `scanner.disk_io_budget_bytes_per_sec` | `RUSTFS_SCANNER_DISK_IO_BUDGET_BYTES_PER_SEC` | bytes/s | `auto` (`DEFAULT_SCANNER_DISK_IO_BUDGET`) | Per-local-disk ceiling on the metadata bytes the scanner reads. `auto` picks the media default (`16777216` on rotational or unknown media, `268435456` on flash); `off` removes the cap; a number is an explicit rate. | +| `scanner.disk_io_budget_iops` | `RUSTFS_SCANNER_DISK_IO_BUDGET_IOPS` | operations/s | `auto` (`DEFAULT_SCANNER_DISK_IO_BUDGET`) | Per-local-disk pacing for scanner directory opens, enumerated entries, symlink target stats, and object metadata reads. `auto` picks the media default (`100` on rotational or unknown media, `4000` on flash); `off` removes the cap; a number is an explicit rate. See [Per-disk background I/O budget](#per-disk-background-io-budget). | +| `scanner.disk_io_budget_bytes_per_sec` | `RUSTFS_SCANNER_DISK_IO_BUDGET_BYTES_PER_SEC` | bytes/s | `auto` (`DEFAULT_SCANNER_DISK_IO_BUDGET`) | Per-local-disk pacing for the metadata bytes the scanner reads. `auto` picks the media default (`16777216` on rotational or unknown media, `268435456` on flash); `off` removes the cap; a number is an explicit rate. | | `scanner.yield_every_n_objects` | `RUSTFS_SCANNER_YIELD_EVERY_N_OBJECTS` | objects | `128` (`DEFAULT_SCANNER_YIELD_EVERY_N_OBJECTS`) | Controls how often object loops yield to the async runtime. `0` disables this extra yield. | | `scanner.alert_excess_versions` | `RUSTFS_SCANNER_ALERT_EXCESS_VERSIONS` | versions | `100` (`DEFAULT_SCANNER_ALERT_EXCESS_VERSIONS`) | Version count threshold for scanner alerts. | | `scanner.alert_excess_version_size` | `RUSTFS_SCANNER_ALERT_EXCESS_VERSION_SIZE` | bytes | `1099511627776` (`DEFAULT_SCANNER_ALERT_EXCESS_VERSION_SIZE`) | Retained version byte threshold for scanner alerts. | @@ -137,7 +137,11 @@ This is logical operation admission, not an OS disk scheduler: foreground and ba ## Per-disk background I/O budget -The admission gate orders scanner I/O behind foreground requests but does not bound how much background I/O an idle disk sees. `scanner.disk_io_budget_iops` and `scanner.disk_io_budget_bytes_per_sec` add that bound. Each local disk owns one token bucket per axis (`crates/scanner/src/sleeper/io_budget.rs`); every directory listing charges one operation, and every object metadata read charges one operation plus the `xl.meta` size. A charge that exceeds the remaining budget delays that walker until the rate allows it. The budget is layered on top of admission: a budgeted operation still queues behind foreground requests, so the budget can only slow the scanner down and never lets it exceed what admission allows. It applies whether or not `idle_mode` is enabled. +The admission gate orders scanner I/O behind foreground requests but does not bound how much background I/O an idle disk sees. `scanner.disk_io_budget_iops` and `scanner.disk_io_budget_bytes_per_sec` pace scanner work in addition to that gate. Each local disk owns one token bucket per axis (`crates/scanner/src/sleeper/io_budget.rs`); charges cover directory opens, each enumerated entry, symlink target stats, optional `xl.meta` size stats, and metadata reads. A charge that exceeds the remaining budget delays that walker. Tokio may prefetch a bounded directory chunk, and waits are capped to preserve progress, so these settings pace scanner work rather than enforcing a strict OS-level IOPS or bandwidth ceiling. The budget applies whether or not `idle_mode` is enabled and does not control unrelated host processes. + +Only `xl.meta` files enter object metadata accounting. Other payload files consume entry charges without extra size stats or metadata-read charges. An `xl.meta` symlink reuses its resolved target size for the byte charge. Cancellation interrupts folder pacing, budget waits, and directory/stat waits. Filesystem work already dispatched to a blocking worker can still finish. + +Erasure data-directory probes for missing `xl.meta` use the same directory-open and entry budget. They retain their 64-entry limit and remain cancellation-aware without adding a preset folder delay per probe. - Media is detected once per disk path from the backing block device's `rotational` flag on Linux, through partitions, `dm-*`, and software RAID; any spindle in the stack counts as rotational. Undetectable media, including non-Linux hosts and overlay filesystems without a bind-mounted volume, uses the rotational defaults because a spindle is the case that needs protection. - `auto` resolves to `100` operations/s and `16 MiB/s` on rotational or unknown media and to `4000` operations/s and `256 MiB/s` on flash. The rotational values leave roughly half of a 7200 rpm spindle's random reads to foreground traffic; the flash values are a few percent of a SATA SSD. The `fastest` speed preset resolves `auto` to no cap; explicit numbers always apply.