mirror of
https://github.com/rustfs/rustfs.git
synced 2026-10-26 22:07:53 +00:00
fix(scanner): streamline budgeted metadata walks (#8837)
* fix(scanner): budget directory enumeration I/O Co-Authored-By: heihutu <heihutu@gmail.com> Co-Authored-By: zhi22915 <qiuzgang@gmail.com> * refactor(scanner): skip redundant metadata walk work Co-Authored-By: heihutu <heihutu@gmail.com> Co-Authored-By: zhi22915 <qiuzgang@gmail.com> * refactor(scanner): share budgeted directory traversal Co-Authored-By: heihutu <heihutu@gmail.com> Co-Authored-By: zhi22915 <qiuzgang@gmail.com> --------- Co-authored-by: zhi22915 <qiuzgang@gmail.com>
This commit is contained in:
@@ -912,6 +912,28 @@ fn prune_size_reconciliation(info: &mut DataUsageCacheInfo, now: u64) {
|
||||
}
|
||||
}
|
||||
|
||||
async fn wait_for_scanner_io<T>(ctx: &CancellationToken, wait: impl std::future::Future<Output = T>) -> Result<T, ScannerError> {
|
||||
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<Option<tokio::fs::DirEntry>, 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<CachedFolder> = Vec::new();
|
||||
let mut new_folders: Vec<CachedFolder> = 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;
|
||||
}
|
||||
|
||||
@@ -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<bool, ScannerError> {
|
||||
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<bool, ScannerError> {
|
||||
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<bool, Scann
|
||||
continue;
|
||||
}
|
||||
|
||||
match entry.file_type().await {
|
||||
match wait_for_scanner_io(ctx, entry.file_type()).await? {
|
||||
Ok(file_type) if file_type.is_file() => return Ok(true),
|
||||
Ok(_) => {}
|
||||
Err(err) if matches!(err.kind(), ErrorKind::NotFound | ErrorKind::TooManyLinks) => {}
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -134,7 +134,7 @@ pub static SCANNER_SLEEPER: LazyLock<DynamicSleeper> = 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<SleeperParams>,
|
||||
@@ -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.
|
||||
|
||||
@@ -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() {
|
||||
|
||||
@@ -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.
|
||||
|
||||
Reference in New Issue
Block a user