From 0bade0ca3ef19cfa0940f134f6f2d91e390965db Mon Sep 17 00:00:00 2001 From: Ramakrishna Chilaka <49393831+RamakrishnaChilaka@users.noreply.github.com> Date: Thu, 22 Oct 2026 12:24:34 +0530 Subject: [PATCH] perf(ecstore,filemeta): batch listings and buffer metacache streams (#8791) * perf(ecstore,filemeta): batch local walk_dir metadata reads and buffer metacache streams * fix(ecstore): stream completed listing metadata reads Publish each result before starting the next blocking read, preserving per-object progress and cancellation without a timed chunk queue. Keep dispatch batching, byte limits, and namespace/error stops intact. Cover multiple batches, page boundaries, and oversized-envelope resumption. Co-Authored-By: heihutu Co-Authored-By: zhi22915 * test(ecstore,filemeta): verify production listing streams Replace the simulated directory-walk benchmark with the public local disk path. Validate fixture output before timing and retain metacache-only stream benchmarks. Pin fragmented reads, read coalescing, and canonical UUID namespace probing against the reference behavior. Co-Authored-By: heihutu Co-Authored-By: zhi22915 * fix(ecstore): route listing benchmark through storage API Centralize local listing benchmark imports in the shared benchmark storage API boundary required by the architecture migration rules. Co-Authored-By: heihutu Co-Authored-By: zhi22915 * test(scanner): keep catch-up fixture timestamps consistent Admit the seeded catch-up using its observation timestamp so a durable writer claim crossing a wall-clock second cannot spuriously rate-limit it. Model an older claim timestamp to exercise that boundary deterministically. Co-Authored-By: heihutu Co-Authored-By: zhi22915 --------- Co-authored-by: Hauser Co-authored-by: heihutu Co-authored-by: zhi22915 --- crates/ecstore/Cargo.toml | 4 + .../benches/local_walk_dir_benchmark.rs | 85 +++ crates/ecstore/benches/storage_api/mod.rs | 4 + crates/ecstore/src/disk/local.rs | 482 ++++++++++++++---- crates/ecstore/src/disk/os.rs | 95 ++++ crates/filemeta/benches/xl_meta_bench.rs | 94 +++- crates/filemeta/src/metacache.rs | 193 +++++-- .../src/scanner/tests/wakeup_metrics.rs | 5 +- 8 files changed, 834 insertions(+), 128 deletions(-) create mode 100644 crates/ecstore/benches/local_walk_dir_benchmark.rs diff --git a/crates/ecstore/Cargo.toml b/crates/ecstore/Cargo.toml index 07b29ac4b..a4271c25d 100644 --- a/crates/ecstore/Cargo.toml +++ b/crates/ecstore/Cargo.toml @@ -286,5 +286,9 @@ harness = false name = "multipart_read_parts_benchmark" harness = false +[[bench]] +name = "local_walk_dir_benchmark" +harness = false + [lib] doctest = false diff --git a/crates/ecstore/benches/local_walk_dir_benchmark.rs b/crates/ecstore/benches/local_walk_dir_benchmark.rs new file mode 100644 index 000000000..b65f7aa80 --- /dev/null +++ b/crates/ecstore/benches/local_walk_dir_benchmark.rs @@ -0,0 +1,85 @@ +// Copyright 2024 RustFS Team +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +use criterion::{Criterion, Throughput, criterion_group, criterion_main}; +use rustfs_filemeta::{MetaCacheEntry, MetacacheReader, test_data::create_real_xlmeta}; +use std::hint::black_box; + +mod storage_api; +use storage_api::local_walk_dir::{Disk, DiskAPI, DiskOption, Endpoint, WalkDirOptions, new_disk}; + +const OBJECTS: usize = 1000; + +async fn walk_page(disk: &Disk, recursive: bool) -> Vec { + let opts = WalkDirOptions { + bucket: "bench-bucket".to_string(), + base_dir: "prefix".to_string(), + recursive, + limit: i32::try_from(OBJECTS).expect("page limit fits"), + ..Default::default() + }; + let (reader, mut writer) = tokio::io::duplex(4096); + let produce = async move { disk.walk_dir(opts, &mut writer).await }; + let consume = async { MetacacheReader::new(reader).read_all().await.expect("decode listing stream") }; + let (result, entries) = tokio::join!(produce, consume); + result.expect("production walk_dir must succeed"); + entries +} + +fn bench_local_walk_dir(c: &mut Criterion) { + let root = tempfile::tempdir().expect("create benchmark disk"); + let metadata = create_real_xlmeta().expect("create object metadata"); + let decoded = rustfs_filemeta::FileMeta::load(&metadata).expect("decode fixture"); + let data_dirs = decoded.get_data_dirs().expect("fixture data directories"); + for i in 0..OBJECTS { + let object_dir = root.path().join("bench-bucket/prefix").join(format!("object-{i:04}")); + std::fs::create_dir_all(&object_dir).expect("create object directory"); + for data_dir in data_dirs.iter().flatten() { + std::fs::create_dir_all(object_dir.join(data_dir.to_string())).expect("create data directory"); + } + std::fs::write(object_dir.join("xl.meta"), &metadata).expect("write object metadata"); + } + let runtime = tokio::runtime::Builder::new_multi_thread() + .worker_threads(4) + .enable_all() + .build() + .expect("build benchmark runtime"); + let endpoint = Endpoint::try_from(root.path().to_str().expect("UTF-8 disk path")).expect("disk endpoint"); + let disk = runtime + .block_on(new_disk( + &endpoint, + &DiskOption { + cleanup: false, + health_check: false, + }, + )) + .expect("open local disk"); + let mut group = c.benchmark_group("local_walk_dir"); + group.throughput(Throughput::Elements(u64::try_from(OBJECTS).expect("object count fits"))); + for (name, recursive) in [("recursive_1000_objects", true), ("non_recursive_1000_objects", false)] { + let entries = runtime.block_on(walk_page(&disk, recursive)); + let objects: Vec<_> = entries.iter().filter(|entry| !entry.metadata.is_empty()).collect(); + assert_eq!(objects.len(), OBJECTS); + for (i, entry) in objects.iter().enumerate() { + assert_eq!(entry.name, format!("prefix/object-{i:04}")); + } + group.bench_function(name, |b| { + b.iter(|| black_box(runtime.block_on(walk_page(&disk, recursive)))); + }); + } + group.finish(); +} + +criterion_group!(benches, bench_local_walk_dir); +criterion_main!(benches); diff --git a/crates/ecstore/benches/storage_api/mod.rs b/crates/ecstore/benches/storage_api/mod.rs index ae80a1d42..ea435d0d6 100644 --- a/crates/ecstore/benches/storage_api/mod.rs +++ b/crates/ecstore/benches/storage_api/mod.rs @@ -33,3 +33,7 @@ pub(crate) mod single_block_non_inline { pub(crate) mod multipart { pub(crate) use rustfs_ecstore::api::disk::{DiskAPI, DiskOption, Endpoint, new_disk}; } + +pub(crate) mod local_walk_dir { + pub(crate) use rustfs_ecstore::api::disk::{Disk, DiskAPI, DiskOption, Endpoint, WalkDirOptions, new_disk}; +} diff --git a/crates/ecstore/src/disk/local.rs b/crates/ecstore/src/disk/local.rs index b3f3a6201..574403565 100644 --- a/crates/ecstore/src/disk/local.rs +++ b/crates/ecstore/src/disk/local.rs @@ -235,9 +235,19 @@ enum ReadAllError { struct ListingMetadataRead { bytes: Vec, + #[cfg(test)] file_meta: Option, - data_dirs: HashSet, + data_dirs: Vec, has_namespace_child_candidate: bool, + counts_toward_limit: bool, +} + +type ListingMetadataItem = core::result::Result; + +struct ListingMetadataBatch { + volume_dir: PathBuf, + rx: tokio::sync::mpsc::UnboundedReceiver, + remaining: usize, } const LISTING_META_INITIAL_READ_BYTES: usize = 4 << 10; @@ -272,31 +282,56 @@ fn read_listing_metadata_bytes_std( opened: std::io::Result, ) -> core::result::Result<(Vec, Option), ReadAllError> { let mut file = opened.map_err(ReadAllError::Open)?; - let metadata = file.metadata().map_err(|err| ReadAllError::Disk(to_file_error(err).into()))?; + let mut initial_buf = [0u8; LISTING_META_INITIAL_READ_BYTES]; + let mut n = 0; + let mut reached_eof = false; - if metadata.is_dir() { - return Err(ReadAllError::Disk(DiskError::FileNotFound)); + while n < LISTING_META_INITIAL_READ_BYTES { + match std::io::Read::read(&mut file, &mut initial_buf[n..]) { + Ok(0) => { + reached_eof = true; + break; + } + Ok(read_bytes) => { + n += read_bytes; + if let Ok((tmp_buf, 1, minor @ 1..=3)) = FileMeta::check_xl2_v1(&initial_buf[..n]) + && let Ok((sz, rest)) = FileMeta::read_bytes_header(tmp_buf) + { + let header_len = n - rest.len(); + let crc_len = if minor >= 2 { 5 } else { 0 }; + if let Some(want) = (sz as usize).checked_add(header_len).and_then(|v| v.checked_add(crc_len)) + && want <= n + { + return Ok((initial_buf[..want].to_vec(), None)); + } + } + } + Err(err) if err.kind() == ErrorKind::Interrupted => continue, + #[cfg(unix)] + Err(err) if err.raw_os_error() == Some(libc::EISDIR) => { + return Err(ReadAllError::Disk(DiskError::FileNotFound)); + } + Err(err) => { + #[cfg(not(unix))] + if file.metadata().is_ok_and(|m| m.is_dir()) { + return Err(ReadAllError::Disk(DiskError::FileNotFound)); + } + return Err(ReadAllError::Disk(to_file_error(err).into())); + } + } } - let size = usize::try_from(metadata.len()).map_err(|err| ReadAllError::Disk(DiskError::other(err)))?; - let modtime = metadata.modified().ok().map(OffsetDateTime::from); - if size == 0 { - return Ok((Vec::new(), modtime)); + if n == 0 { + return Ok((Vec::new(), None)); } - let initial = size.min(LISTING_META_INITIAL_READ_BYTES); - let mut bytes = Vec::new(); - bytes - .try_reserve_exact(initial) - .map_err(|err| ReadAllError::Disk(Error::other(err)))?; - std::io::Read::read_to_end(&mut std::io::Read::take(&mut file, initial as u64), &mut bytes) - .map_err(|err| ReadAllError::Disk(to_file_error(err).into()))?; - - if let Ok((tmp_buf, major, minor)) = FileMeta::check_xl2_v1(&bytes) { + if !reached_eof && let Ok((tmp_buf, major, minor)) = FileMeta::check_xl2_v1(&initial_buf) { + let metadata = file.metadata().map_err(|err| ReadAllError::Disk(to_file_error(err).into()))?; + let size = usize::try_from(metadata.len()).map_err(|err| ReadAllError::Disk(DiskError::other(err)))?; let want = match (major, minor) { (1, 0) => Some(size), (1, 1..=3) => FileMeta::read_bytes_header(tmp_buf).ok().and_then(|(sz, rest)| { - let header_len = bytes.len() - rest.len(); + let header_len = initial_buf.len() - rest.len(); let crc_len = if minor >= 2 { 5 } else { 0 }; (sz as usize).checked_add(header_len)?.checked_add(crc_len) }), @@ -304,21 +339,83 @@ fn read_listing_metadata_bytes_std( }; if let Some(want) = want && want <= size + && want > initial_buf.len() { - if want > bytes.len() && bytes.len() == initial { - let extra = want - bytes.len(); - bytes - .try_reserve_exact(extra) - .map_err(|err| ReadAllError::Disk(Error::other(err)))?; - std::io::Read::read_to_end(&mut std::io::Read::take(&mut file, extra as u64), &mut bytes) - .map_err(|err| ReadAllError::Disk(to_file_error(err).into()))?; - } else if want < bytes.len() { - bytes.truncate(want); - } + let mut bytes = Vec::new(); + bytes + .try_reserve_exact(want.saturating_add(1)) + .map_err(|err| ReadAllError::Disk(Error::other(err)))?; + bytes.extend_from_slice(&initial_buf); + let extra = want - initial_buf.len(); + std::io::Read::read_to_end(&mut std::io::Read::take(&mut file, extra as u64), &mut bytes) + .map_err(|err| ReadAllError::Disk(to_file_error(err).into()))?; + return Ok((bytes, None)); } } - Ok((bytes, modtime)) + Ok((initial_buf[..n].to_vec(), None)) +} + +const LISTING_META_BATCH_MAX_BYTES: usize = 1 << 20; + +// Deliver progress before starting another blocking read. A timer checked after a +// read cannot publish completed entries while the next filesystem call is stalled. +fn stream_listing_metadata_batch( + object_names: &[String], + tx: tokio::sync::mpsc::UnboundedSender, + mut read: impl FnMut(&str) -> ListingMetadataItem, +) { + let mut buffered_bytes = 0usize; + for object_name in object_names { + if tx.is_closed() { + break; + } + let res = read(object_name); + if let Ok(read) = &res { + buffered_bytes = buffered_bytes.saturating_add(read.bytes.len()); + } + let stop = res.as_ref().map_or(true, |read| read.has_namespace_child_candidate) + || buffered_bytes >= LISTING_META_BATCH_MAX_BYTES; + if tx.send(res).is_err() || stop { + break; + } + } +} + +fn parse_listing_metadata_bytes(bytes: &[u8]) -> (Option, Vec, bool) { + let file_meta = FileMeta::load(bytes).ok(); + let mut data_dirs = Vec::new(); + if let Some(meta) = file_meta.as_ref() + && let Ok(dirs) = meta.get_data_dirs() + { + data_dirs.extend(dirs.into_iter().flatten()); + data_dirs.sort_unstable(); + data_dirs.dedup(); + } + let counts_toward_limit = file_meta.as_ref().is_none_or(file_meta_counts_toward_limit); + (file_meta, data_dirs, counts_toward_limit) +} + +fn read_listing_metadata_entry_std( + root: &os::PublicationRoot, + object_dir: &Path, +) -> core::result::Result { + let metadata_path = object_dir.join(STORAGE_FORMAT_FILE); + let (bytes, _) = read_listing_metadata_bytes_std(root.open_readonly(&metadata_path))?; + let (file_meta, data_dirs, counts_toward_limit) = parse_listing_metadata_bytes(&bytes); + #[cfg(not(test))] + let _ = file_meta; + let has_namespace_child_candidate = + os::has_non_data_dir_entry(object_dir, &data_dirs).map_err(|err| ReadAllError::Disk(to_file_error(err).into()))?; + + Ok(ListingMetadataRead { + bytes, + #[cfg(test)] + file_meta, + data_dirs, + has_namespace_child_candidate, + counts_toward_limit, + }) } pub(crate) fn inline_metadata_rollback_dir(version_id: Uuid, meta: &FileMeta) -> Uuid { @@ -7962,49 +8059,54 @@ impl LocalDisk { })) } + #[cfg(test)] async fn read_listing_metadata(&self, volume: &str, object_name: &str) -> Result { - let io_permit = acquire_context_disk_io_permit().await.map_err(DiskError::Io)?; - let object_dir = self.io_get_object_path(volume, object_name)?; - let metadata_path = object_dir.join(STORAGE_FORMAT_FILE); + let mut batch = self + .spawn_listing_metadata_batch(volume, vec![object_name.to_owned()], None) + .await?; + let raw = batch + .rx + .recv() + .await + .ok_or_else(|| DiskError::other("listing metadata batch closed unexpectedly"))?; + self.resolve_read_all_result(volume, &batch.volume_dir, raw).await + } + + async fn spawn_listing_metadata_batch( + &self, + volume: &str, + object_names: Vec, + stall: Option, + ) -> Result { + let io_permit = with_walk_stall_deadline(stall, acquire_context_disk_io_permit()) + .await? + .map_err(DiskError::Io)?; let volume_dir = self.io_get_bucket_path(volume)?; + let io_root = self.io_root().to_path_buf(); + #[cfg(target_os = "linux")] + let mount_lease = self.mount_lease.clone(); + let volume_str = volume.to_owned(); let root = self.publication_root.clone(); - let result = spawn_blocking_with_disk_io_permit(io_permit, move || { - let (bytes, _) = read_listing_metadata_bytes_std(root.open_readonly(&metadata_path))?; - let file_meta = FileMeta::load(&bytes).ok(); - let data_dirs: HashSet = file_meta - .as_ref() - .and_then(|meta| meta.get_data_dirs().ok()) - .into_iter() - .flatten() - .flatten() - .map(|data_dir| data_dir.to_string()) - .collect(); - let probe = match os::read_dir_probe(&object_dir, data_dirs.len().saturating_add(2)) { - Ok(probe) => probe, - Err(err) if err.kind() == ErrorKind::NotFound => os::ReadDirProbe { - entries: Vec::new(), - complete: true, - }, - Err(err) => return Err(ReadAllError::Disk(to_file_error(err).into())), - }; - let found_child = probe.entries.into_iter().any(|entry| { - entry - .strip_suffix(SLASH_SEPARATOR) - .is_some_and(|child| !child.is_empty() && !data_dirs.contains(child)) + let remaining = object_names.len(); + let (tx, rx) = tokio::sync::mpsc::unbounded_channel(); + drop(spawn_blocking_with_disk_io_permit(io_permit, move || { + stream_listing_metadata_batch(&object_names, tx, |object_name| { + let (bucket_path, object_dir) = build_local_disk_object_path(&io_root, &volume_str, object_name); + #[cfg(target_os = "linux")] + let path_check = check_local_disk_valid_object_path_at(&io_root, &mount_lease, &bucket_path, &object_dir); + #[cfg(not(target_os = "linux"))] + let path_check = check_local_disk_valid_object_path(&io_root, &bucket_path, &object_dir); + match path_check { + Ok(()) => read_listing_metadata_entry_std(&root, &object_dir), + Err(err) => Err(ReadAllError::Disk(err)), + } }); - let has_namespace_child_candidate = found_child || !probe.complete; - - Ok(ListingMetadataRead { - bytes, - file_meta, - data_dirs, - has_namespace_child_candidate, - }) + })); + Ok(ListingMetadataBatch { + volume_dir, + rx, + remaining, }) - .await - .map_err(DiskError::from)?; - - self.resolve_read_all_result(volume, &volume_dir, result).await } async fn resolve_read_all_result( @@ -9213,8 +9315,10 @@ impl LocalDisk { } }; prefix = "".to_owned(); + let mut prefetched_batch: Option = None; + let mut adaptive_batch_cap = WALK_DIR_BATCH_SIZE; - for entry in entries.iter() { + for (entry_idx, entry) in entries.iter().enumerate() { if entry.is_empty() { continue; } @@ -9299,50 +9403,93 @@ impl LocalDisk { meta.name.push_str(GLOBAL_DIR_SUFFIX_WITH_SLASH); } - let fname = format!("{}/{}", meta.name, STORAGE_FORMAT_FILE); let metadata_read = if !is_dir_obj { - with_walk_stall_timeout(stall, self.read_listing_metadata(&opts.bucket, &meta.name)) - .await - .map(|read| (read.bytes, read.file_meta, Some(read.data_dirs), read.has_namespace_child_candidate)) + async { + let mut spawned = false; + let (volume_dir, raw_res) = loop { + if let Some(batch) = prefetched_batch.as_mut() { + if let Some(raw_res) = with_walk_stall_deadline(stall, batch.rx.recv()).await? { + batch.remaining -= 1; + let stop = batch.remaining == 0 + || raw_res.as_ref().map_or(true, |r| r.has_namespace_child_candidate); + let volume_dir = batch.volume_dir.clone(); + if stop { + prefetched_batch = None; + } + break (volume_dir, raw_res); + } + prefetched_batch = None; + if spawned { + return Err(DiskError::other("listing metadata batch closed unexpectedly")); + } + } + spawned = true; + let batch_limit = if opts.limit > 0 { + usize::try_from(opts.limit - *objs_returned) + .unwrap_or(1) + .min(adaptive_batch_cap) + } else { + adaptive_batch_cap + }; + let batch_names = if batch_limit > 1 + && dir_objes.is_empty() + && dir_stack.iter().all(|(_, _, _, scan_required)| !*scan_required) + { + entries[entry_idx..] + .iter() + .take_while(|c| !c.is_empty()) + .take(batch_limit) + .map(|c| path_join_buf(&[current.as_str(), c.as_str()])) + .collect() + } else { + vec![meta.name.clone()] + }; + prefetched_batch = Some(self.spawn_listing_metadata_batch(&opts.bucket, batch_names, stall).await?); + }; + with_walk_stall_timeout(stall, self.resolve_read_all_result(&opts.bucket, &volume_dir, raw_res)).await + } + .await } else { + let fname = format!("{}/{}", meta.name, STORAGE_FORMAT_FILE); with_walk_stall_timeout(stall, self.read_metadata(&opts.bucket, fname.as_str())) .await - .map(|metadata| (Vec::from(metadata), None, None, true)) + .map(|metadata| { + let bytes = Vec::from(metadata); + let (file_meta, data_dirs, counts_toward_limit) = parse_listing_metadata_bytes(&bytes); + #[cfg(not(test))] + let _ = file_meta; + ListingMetadataRead { + bytes, + #[cfg(test)] + file_meta, + data_dirs, + has_namespace_child_candidate: true, + counts_toward_limit, + } + }) }; match metadata_read { - Ok((res, prefetched_file_meta, prefetched_data_dirs, has_namespace_child_candidate)) => { + Ok(read) => { + adaptive_batch_cap = WALK_DIR_BATCH_SIZE; if is_dir_obj { meta.name = meta.name.trim_end_matches(GLOBAL_DIR_SUFFIX_WITH_SLASH).to_owned(); meta.name.push_str(SLASH_SEPARATOR); } - let file_meta = match prefetched_file_meta { - Some(file_meta) => Some(file_meta), - None if opts.limit > 0 || opts.recursive || !is_dir_obj => FileMeta::load(&res).ok(), - None => None, - }; - - meta.metadata = res; + meta.metadata = read.bytes; write_metacache_obj(out, &meta).await?; - if opts.limit <= 0 || file_meta.as_ref().is_none_or(file_meta_counts_toward_limit) { + if opts.limit <= 0 || read.counts_toward_limit { *objs_returned += 1; } - let dir_to_skip = if let Some(data_dirs) = prefetched_data_dirs { - data_dirs + let has_namespace_child_candidate = read.has_namespace_child_candidate; + let dir_to_skip: HashSet = if has_namespace_child_candidate || !dir_objes.is_empty() { + read.data_dirs.into_iter().map(|dir| dir.to_string()).collect() } else { - let mut data_dirs_to_skip = HashSet::new(); - if let Some(file_meta) = file_meta.as_ref() - && let Ok(data_dirs) = file_meta.get_data_dirs() - { - for data_dir in data_dirs.iter().flatten() { - data_dirs_to_skip.insert(data_dir.to_string()); - } - } - data_dirs_to_skip + HashSet::new() }; if opts.recursive { @@ -9376,6 +9523,7 @@ impl LocalDisk { } Err(err) => { if err == Error::FileNotFound || err == Error::IsNotRegular { + adaptive_batch_cap = 1; // NOT an object, append to stack (with slash) // If dirObject, but no metadata (which is unexpected) we skip it. if !is_dir_obj && { @@ -9404,6 +9552,7 @@ impl LocalDisk { continue; } + let fname = format!("{}/{}", meta.name, STORAGE_FORMAT_FILE); error!( event = EVENT_DISK_LOCAL_SCAN_FAILED, component = LOG_COMPONENT_ECSTORE, @@ -22534,7 +22683,7 @@ mod test { .await .expect("read leaf metadata"); assert!(!leaf.has_namespace_child_candidate); - assert!(leaf.data_dirs.contains(&data_dir.to_string())); + assert!(leaf.data_dirs.contains(&data_dir)); fs::create_dir(object_dir.join("child")) .await @@ -22546,6 +22695,94 @@ mod test { assert!(parent.has_namespace_child_candidate); } + fn listing_batch_fixture(bytes: Vec, has_namespace_child_candidate: bool) -> ListingMetadataRead { + ListingMetadataRead { + bytes, + file_meta: None, + data_dirs: Vec::new(), + has_namespace_child_candidate, + counts_toward_limit: true, + } + } + + #[tokio::test] + async fn listing_metadata_batch_publishes_before_the_next_read_completes() { + let (tx, mut rx) = tokio::sync::mpsc::unbounded_channel(); + let (started_tx, started_rx) = tokio::sync::oneshot::channel(); + let (resume_tx, resume_rx) = std::sync::mpsc::channel(); + let worker = tokio::task::spawn_blocking(move || { + let mut started_tx = Some(started_tx); + stream_listing_metadata_batch(&["first".to_string(), "second".to_string()], tx, |name| { + if name == "second" { + started_tx + .take() + .expect("second read starts once") + .send(()) + .expect("notify reader"); + resume_rx.recv().expect("resume second read"); + } + Ok(listing_batch_fixture(name.as_bytes().to_vec(), false)) + }); + }); + started_rx.await.expect("worker must reach the blocked second read"); + let first = rx.try_recv(); + // Release the blocking task before asserting, including on regression. + resume_tx.send(()).expect("release blocked worker"); + worker.await.expect("worker must finish"); + assert_eq!( + first + .expect("first result must already be published") + .ok() + .expect("first read succeeds") + .bytes, + b"first" + ); + assert_eq!( + rx.recv() + .await + .expect("second result") + .ok() + .expect("second read succeeds") + .bytes, + b"second" + ); + assert!(rx.recv().await.is_none()); + } + + #[test] + fn listing_metadata_batch_stops_on_boundaries_and_cancellation() { + for boundary in ["bytes", "child", "error", "cancel"] { + let (tx, mut rx) = tokio::sync::mpsc::unbounded_channel(); + if boundary == "cancel" { + rx.close(); + } + let mut reads = 0; + stream_listing_metadata_batch(&["first".to_string(), "second".to_string()], tx, |_| { + reads += 1; + if boundary == "error" { + Err(ReadAllError::Disk(DiskError::FileNotFound)) + } else { + Ok(listing_batch_fixture( + vec![ + 0; + if boundary == "bytes" { + LISTING_META_BATCH_MAX_BYTES + } else { + 1 + } + ], + boundary == "child", + )) + } + }); + assert_eq!(reads, usize::from(boundary != "cancel"), "boundary: {boundary}"); + if boundary != "cancel" { + assert!(rx.try_recv().is_ok(), "the boundary result must be delivered: {boundary}"); + } + assert!(rx.try_recv().is_err(), "no result may follow the boundary: {boundary}"); + } + } + #[tokio::test] async fn listing_metadata_skips_inline_payload_bytes_and_preserves_corrupt_classification() { use tempfile::tempdir; @@ -22753,6 +22990,12 @@ mod test { fi.data_dir = Some(data_dir); fi.size = 16; fi.mod_time = Some(OffsetDateTime::from_unix_timestamp(1_700_000_000).expect("ts")); + if name == "obj-20" { + // A single envelope can exhaust the byte budget before the name + // batch ends; subsequent keys must resume in a fresh worker. + fi.metadata + .insert("x-amz-meta-padding".to_string(), "p".repeat(LISTING_META_BATCH_MAX_BYTES)); + } fi.data = Some(Bytes::from(vec![b'a'; 16 * 1024])); fi.set_inline_data(); fm.add_version(fi).expect("add version"); @@ -22812,6 +23055,57 @@ mod test { assert_eq!(entries[2].name, "parent/"); assert!(entries[2].is_dir()); + + // Cross multiple batches, including an early stop for a nested child, and + // pin both sides of the 32-object batch boundary and an unlimited walk. + let batch_bucket = "walk-batch-bucket"; + let batch_bucket_dir = dir.path().join(batch_bucket); + for idx in 0..80 { + let name = format!("obj-{idx:02}"); + let obj_dir = batch_bucket_dir.join(&name); + fs::create_dir_all(obj_dir.join(data_dir.to_string())) + .await + .expect("create batch obj data_dir"); + fs::write(obj_dir.join(STORAGE_FORMAT_FILE), make_obj_meta(&name)) + .await + .expect("write batch obj xl.meta"); + if idx == 10 { + let nested_dir = obj_dir.join("nested"); + fs::create_dir_all(&nested_dir).await.expect("create nested child dir"); + fs::write(nested_dir.join(STORAGE_FORMAT_FILE), make_obj_meta("obj-10/nested")) + .await + .expect("write nested child xl.meta"); + } + } + + let mut expected: Vec = (0..80).map(|i| format!("obj-{i:02}")).collect(); + expected.insert(11, "obj-10/nested".to_string()); + for limit in [15, 32, 33, 0] { + let (reader, mut writer) = tokio::io::duplex(64 * 1024); + let batch_opts = WalkDirOptions { + bucket: batch_bucket.to_string(), + base_dir: String::new(), + recursive: true, + limit, + ..Default::default() + }; + let disk_ref = &disk; + let produce = async move { disk_ref.walk_dir(batch_opts, &mut writer).await }; + let consume = async { MetacacheReader::new(reader).read_all().await.expect("read batched stream") }; + let (result, batch_entries) = tokio::join!(produce, consume); + result.expect("batched recursive walk_dir must succeed"); + let visible_names: Vec<_> = batch_entries + .iter() + .filter(|e| !e.metadata.is_empty()) + .map(|e| e.name.as_str()) + .collect(); + let count = if limit == 0 { + expected.len() + } else { + usize::try_from(limit).expect("positive limit") + }; + assert_eq!(visible_names, expected[..count], "page limit {limit}"); + } } #[tokio::test] diff --git a/crates/ecstore/src/disk/os.rs b/crates/ecstore/src/disk/os.rs index 0f3e3f8a7..849897035 100644 --- a/crates/ecstore/src/disk/os.rs +++ b/crates/ecstore/src/disk/os.rs @@ -2061,11 +2061,13 @@ pub async fn is_empty_dir(path: impl AsRef) -> bool { const READ_DIR_PROBE_RAW_LIMIT: usize = 256; +#[cfg(test)] pub(crate) struct ReadDirProbe { pub entries: Vec, pub complete: bool, } +#[cfg(test)] pub(crate) fn read_dir_probe(path: impl AsRef, entry_limit: usize) -> io::Result { let mut dir = std::fs::read_dir(path)?; let mut entries = Vec::with_capacity(entry_limit.min(READ_DIR_PROBE_RAW_LIMIT)); @@ -2102,6 +2104,59 @@ pub(crate) fn read_dir_probe(path: impl AsRef, entry_limit: usize) -> io:: }) } +pub(crate) fn has_non_data_dir_entry(path: impl AsRef, data_dirs: &[uuid::Uuid]) -> io::Result { + match has_non_data_dir_entry_inner(path.as_ref(), data_dirs) { + Ok(res) => Ok(res), + Err(err) if err.kind() == io::ErrorKind::NotFound => Ok(false), + Err(err) => Err(err), + } +} + +fn has_non_data_dir_entry_inner(path: &Path, data_dirs: &[uuid::Uuid]) -> io::Result { + let mut dir = std::fs::read_dir(path)?; + let entry_limit = data_dirs.len().saturating_add(2); + let mut valid_entries = 0usize; + + for _ in 0..READ_DIR_PROBE_RAW_LIMIT { + let Some(entry) = dir.next() else { + return Ok(false); + }; + let entry = entry?; + let file_name = entry.file_name(); + if file_name.is_empty() || file_name == "." || file_name == ".." { + continue; + } + + let file_type = entry.file_type()?; + if file_type.is_dir() { + let is_data_dir = file_name.len() == uuid::fmt::Hyphenated::LENGTH + && file_name + .to_str() + .and_then(|s| { + let parsed = uuid::Uuid::try_parse(s).ok()?; + let mut buf = [0u8; uuid::fmt::Hyphenated::LENGTH]; + (parsed.hyphenated().encode_lower(&mut buf) == s && data_dirs.binary_search(&parsed).is_ok()) + .then_some(()) + }) + .is_some(); + if !is_data_dir { + return Ok(true); + } + valid_entries += 1; + } else if file_type.is_file() { + valid_entries += 1; + } else { + continue; + } + + if valid_entries == entry_limit { + return Ok(true); + } + } + + Ok(true) +} + // read_dir count read limit. when count == 0 unlimit. /// Return file names in the directory. #[tracing::instrument(level = "debug", skip_all)] @@ -5456,6 +5511,45 @@ mod tests { } } + #[test] + fn read_dir_probe_matches_uuid_namespace_probe() { + let data_dir = uuid::Uuid::from_u128(0xabcdef); + let canonical = data_dir.to_string(); + let uppercase = canonical.to_uppercase(); + let simple = data_dir.simple().to_string(); + for (directories, file_count) in [ + (vec![], 0), + (vec![], 1), + (vec![canonical.as_str()], 1), + (vec![canonical.as_str(), "child"], 1), + (vec![uppercase.as_str()], 1), + (vec![simple.as_str()], 1), + (vec![canonical.as_str()], 2), + (vec![canonical.as_str()], 3), + ] { + let root = tempdir().expect("create probe directory"); + for name in &directories { + std::fs::create_dir(root.path().join(name)).expect("create test directory"); + } + for i in 0..file_count { + std::fs::write(root.path().join(format!("file-{i}")), []).expect("create test file"); + } + let probe = read_dir_probe(root.path(), 3).expect("reference probe"); + let expected = !probe.complete + || probe + .entries + .iter() + .any(|entry| entry.strip_suffix('/').is_some_and(|name| name != canonical)); + assert_eq!( + has_non_data_dir_entry(root.path(), &[data_dir]).expect("UUID probe"), + expected, + "directories={directories:?}, files={file_count}" + ); + } + let missing = tempdir().expect("missing probe parent"); + assert!(!has_non_data_dir_entry(missing.path().join("absent"), &[]).expect("missing directory is not a child")); + } + #[cfg(unix)] #[test] fn read_dir_probe_bounds_unsupported_entries() { @@ -5470,6 +5564,7 @@ mod tests { assert!(probe.entries.is_empty()); assert!(!probe.complete, "a bounded probe must not claim that an oversized directory is complete"); + assert!(has_non_data_dir_entry(temp_dir.path(), &[]).expect("probe directory")); } #[test] diff --git a/crates/filemeta/benches/xl_meta_bench.rs b/crates/filemeta/benches/xl_meta_bench.rs index dc093e21d..b96560322 100644 --- a/crates/filemeta/benches/xl_meta_bench.rs +++ b/crates/filemeta/benches/xl_meta_bench.rs @@ -204,6 +204,97 @@ fn bench_delete_inline_versions(c: &mut Criterion) { group.finish(); } +fn bench_metacache_stream_page(c: &mut Criterion) { + let xl_bytes = create_real_xlmeta().expect("create_real_xlmeta failed"); + let mut entries = Vec::with_capacity(2000); + for i in 0..1000 { + entries.push(MetaCacheEntry { + name: format!("prefix/subprefix/object-{i:04}"), + metadata: xl_bytes.clone(), + cached: None, + reusable: false, + }); + entries.push(MetaCacheEntry { + name: format!("prefix/subprefix/object-{i:04}/"), + metadata: Vec::new(), + cached: None, + reusable: false, + }); + } + + let rt = tokio::runtime::Builder::new_current_thread() + .enable_all() + .build() + .expect("build current_thread runtime"); + + let encoded_stream = rt.block_on(async { + let mut cursor = std::io::Cursor::new(Vec::with_capacity(512 * 1024)); + let mut writer = rustfs_filemeta::MetacacheWriter::new(&mut cursor); + writer.write(&entries).await.expect("encode entries"); + writer.close().await.expect("close stream"); + cursor.into_inner() + }); + + let mut group = c.benchmark_group("metacache_stream_page"); + + group.bench_function("reader_read_all/1000_objects", |b| { + b.iter(|| { + rt.block_on(async { + let mut reader = + rustfs_filemeta::MetacacheReader::new(std::io::Cursor::new(black_box(encoded_stream.as_slice()))); + let out = reader.read_all().await.expect("read_all"); + black_box(out.len()); + }); + }); + }); + + group.bench_function("duplex_4k_write_and_read/1000_objects", |b| { + b.iter(|| { + rt.block_on(async { + let (rd, mut wr) = tokio::io::duplex(4096); + let entries_ref = &entries; + let writer_fut = async move { + let mut out = rustfs_filemeta::MetacacheWriter::new(&mut wr); + for chunk in entries_ref.chunks(64) { + out.write(chunk).await.expect("write"); + } + out.close().await.expect("close"); + }; + let reader_fut = async move { + let mut reader = rustfs_filemeta::MetacacheReader::new(rd); + let out = reader.read_all().await.expect("read_all"); + black_box(out.len()); + }; + tokio::join!(writer_fut, reader_fut); + }); + }); + }); + + group.bench_function("duplex_4k_write_obj_and_read/1000_objects", |b| { + b.iter(|| { + rt.block_on(async { + let (rd, mut wr) = tokio::io::duplex(4096); + let entries_ref = &entries; + let writer_fut = async move { + let mut out = rustfs_filemeta::MetacacheWriter::new(&mut wr); + for entry in entries_ref { + out.write_obj(entry).await.expect("write_obj"); + } + out.close().await.expect("close"); + }; + let reader_fut = async move { + let mut reader = rustfs_filemeta::MetacacheReader::new(rd); + let out = reader.read_all().await.expect("read_all"); + black_box(out.len()); + }; + tokio::join!(writer_fut, reader_fut); + }); + }); + }); + + group.finish(); +} + criterion_group!( benches, bench_create_real_xlmeta, @@ -218,7 +309,8 @@ criterion_group!( bench_validate_integrity, bench_into_fileinfo_realistic, bench_list_page_to_fileinfo, - bench_delete_inline_versions + bench_delete_inline_versions, + bench_metacache_stream_page ); criterion_main!(benches); diff --git a/crates/filemeta/src/metacache.rs b/crates/filemeta/src/metacache.rs index 2fbec7df2..5887be925 100644 --- a/crates/filemeta/src/metacache.rs +++ b/crates/filemeta/src/metacache.rs @@ -911,18 +911,30 @@ impl MetacacheWriter { return Err(Error::other("metacacheWriter: no name")); } - self.write_obj(obj).await?; + Self::encode_obj(&mut self.buf, obj)?; + if self.buf.len() >= METACACHE_READ_CHUNK_SIZE { + self.flush().await?; + } } + if !self.buf.is_empty() { + self.flush().await?; + } + + Ok(()) + } + + fn encode_obj(buf: &mut Vec, obj: &MetaCacheEntry) -> Result<()> { + rmp::encode::write_bool(buf, true).map_err(|e| Error::other(format!("{e:?}")))?; + rmp::encode::write_str(buf, &obj.name).map_err(|e| Error::other(format!("{e:?}")))?; + rmp::encode::write_bin(buf, &obj.metadata).map_err(|e| Error::other(format!("{e:?}")))?; Ok(()) } pub async fn write_obj(&mut self, obj: &MetaCacheEntry) -> Result<()> { self.init().await?; - rmp::encode::write_bool(&mut self.buf, true).map_err(|e| Error::other(format!("{e:?}")))?; - rmp::encode::write_str(&mut self.buf, &obj.name).map_err(|e| Error::other(format!("{e:?}")))?; - rmp::encode::write_bin(&mut self.buf, &obj.metadata).map_err(|e| Error::other(format!("{e:?}")))?; + Self::encode_obj(&mut self.buf, obj)?; self.flush().await?; Ok(()) @@ -942,6 +954,7 @@ pub struct MetacacheReader { in_entry: bool, err: Option, buf: Vec, + pos: usize, has_current: bool, } @@ -953,63 +966,111 @@ impl MetacacheReader { in_entry: false, err: None, buf: Vec::new(), + pos: 0, has_current: false, } } pub async fn read_more(&mut self, read_size: usize) -> Result<&[u8]> { + self.read_more_inner(read_size, true).await + } + + async fn read_more_inner(&mut self, read_size: usize, lookahead: bool) -> Result<&[u8]> { if let Some(err) = &self.err { return Err(err.clone()); } - let pref = self.buf.len(); - let Some(ext_size) = pref.checked_add(read_size) else { - let err = Error::other("metacache stream: element length overflow"); - self.err = Some(err.clone()); - return Err(err); - }; + if self.buf.len() - self.pos >= read_size { + let start = self.pos; + self.pos += read_size; + return Ok(&self.buf[start..self.pos]); + } + + if self.pos > 0 { + self.buf.drain(..self.pos); + self.pos = 0; + } // A whole xl.meta can legitimately exceed the limit on its individual // msgpack values. Grow only as payload arrives, so a corrupt length // prefix cannot cause a speculative allocation of the advertised size. - while self.buf.len() < ext_size { + while self.buf.len() < read_size { let start = self.buf.len(); - let extra = (ext_size - start).min(METACACHE_READ_CHUNK_SIZE); + let extra = if lookahead { + let min_chunk = METACACHE_READ_CHUNK_SIZE.saturating_sub(start); + (read_size - start).clamp(min_chunk, METACACHE_READ_CHUNK_SIZE) + } else { + (read_size - start).min(METACACHE_READ_CHUNK_SIZE) + }; if let Err(e) = self.buf.try_reserve(extra) { let err = Error::other(format!("metacache stream: buffer allocation of {extra} bytes failed: {e}")); self.err = Some(err.clone()); return Err(err); } - self.buf.resize(start + extra, 0); - if let Err(e) = self.rd.read_exact(&mut self.buf[start..]).await { - // Listings treat Unexpected/UnexpectedEof as clean exhaustion. - // An incomplete entry must not hide remaining keys. Preserve - // the existing empty/between-entry EOF handling. - let err = match e.kind() { - std::io::ErrorKind::UnexpectedEof if self.in_entry => Error::FileCorrupt, - _ => Error::from(e), - }; - self.err = Some(err.clone()); - return Err(err); + let mut dst = bytes::BufMut::limit(&mut self.buf, extra); + match self.rd.read_buf(&mut dst).await { + Ok(0) => { + // Listings treat Unexpected/UnexpectedEof as clean exhaustion. + // An incomplete entry must not hide remaining keys. Preserve + // the existing empty/between-entry EOF handling. + let err = if self.in_entry { + Error::FileCorrupt + } else { + Error::Unexpected + }; + self.err = Some(err.clone()); + return Err(err); + } + Ok(_) => {} + Err(e) => { + let err = match e.kind() { + std::io::ErrorKind::UnexpectedEof if self.in_entry => Error::FileCorrupt, + _ => Error::from(e), + }; + self.err = Some(err.clone()); + return Err(err); + } } } - Ok(&self.buf[pref..ext_size]) + self.pos = read_size; + Ok(&self.buf[..read_size]) } async fn discard(&mut self, mut size: usize) -> Result<()> { - self.reset(); + let avail = self.buf.len() - self.pos; + if size <= avail { + self.pos += size; + self.reset(); + return Ok(()); + } + size -= avail; + self.buf.clear(); + self.pos = 0; while size > 0 { let chunk = size.min(METACACHE_READ_CHUNK_SIZE); - self.read_more(chunk).await?; + self.read_more_inner(chunk, false).await?; size -= chunk; - self.reset(); + self.buf.clear(); + self.pos = 0; } Ok(()) } fn reset(&mut self) { - self.buf.clear(); + if self.pos >= self.buf.len() { + self.buf.clear(); + self.pos = 0; + if self.buf.capacity() > METACACHE_READ_CHUNK_SIZE { + self.buf.shrink_to(METACACHE_READ_CHUNK_SIZE); + } + } else if self.pos >= METACACHE_READ_CHUNK_SIZE { + self.buf.drain(..self.pos); + self.pos = 0; + if self.buf.capacity() > METACACHE_READ_CHUNK_SIZE { + self.buf.shrink_to(METACACHE_READ_CHUNK_SIZE); + } + } } async fn check_init(&mut self) -> Result<()> { @@ -1164,9 +1225,11 @@ impl MetacacheReader { let l = self.read_bin_len().await?; - let buf = self.read_more(l as usize).await?; - - let metadata = buf.to_vec(); + let metadata = if l == 0 { + Vec::new() + } else { + self.read_more(l as usize).await?.to_vec() + }; self.in_entry = false; self.reset(); @@ -1369,6 +1432,74 @@ mod tests { assert_eq!(reader.read_all().await.expect("remaining entries must decode"), entries[3..]); } + struct FragmentedReader<'a> { + data: &'a [u8], + chunk: usize, + reads: usize, + } + + impl AsyncRead for FragmentedReader<'_> { + fn poll_read( + self: Pin<&mut Self>, + _cx: &mut std::task::Context<'_>, + buf: &mut tokio::io::ReadBuf<'_>, + ) -> std::task::Poll> { + let this = self.get_mut(); + let count = this.data.len().min(this.chunk).min(buf.remaining()); + buf.put_slice(&this.data[..count]); + this.data = &this.data[count..]; + this.reads += 1; + std::task::Poll::Ready(Ok(())) + } + } + + #[tokio::test] + async fn test_reader_buffered_wire_frames_survive_fragmented_reads_and_skips() { + for version in [1, 2] { + let mut bytes = vec![0xcc, version]; + let payloads = [ + vec![1; METACACHE_READ_CHUNK_SIZE - 1], + vec![], + vec![3; METACACHE_READ_CHUNK_SIZE + 1], + ]; + for (i, payload) in payloads.iter().enumerate() { + rmp::encode::write_bool(&mut bytes, true).expect("entry marker"); + rmp::encode::write_str(&mut bytes, &format!("key-{i}")).expect("entry name"); + rmp::encode::write_bin(&mut bytes, payload).expect("entry metadata"); + } + rmp::encode::write_bool(&mut bytes, false).expect("stream end"); + for chunk in [1, 2, 7, 4096, METACACHE_READ_CHUNK_SIZE] { + let mut reader = MetacacheReader::new(FragmentedReader { + data: &bytes, + chunk, + reads: 0, + }); + assert_eq!(reader.peek().await.expect("first entry").expect("present").metadata, payloads[0]); + reader.skip(2).await.expect("consume first and empty second entries"); + let last = reader.peek().await.expect("last entry").expect("present"); + assert_eq!(last.name, "key-2"); + assert_eq!(last.metadata, payloads[2]); + assert!(reader.peek().await.expect("valid terminator").is_none()); + } + } + } + + #[tokio::test] + async fn test_reader_coalesces_small_entries_without_per_field_reads() { + let mut bytes = vec![0xcc, 2]; + for _ in 0..1000 { + bytes.extend_from_slice(&[0xc3, 0xa1, b'x', 0xc4, 1, 0x42]); + } + bytes.push(0xc2); + let mut reader = MetacacheReader::new(FragmentedReader { + data: &bytes, + chunk: METACACHE_READ_CHUNK_SIZE, + reads: 0, + }); + assert_eq!(reader.read_all().await.expect("decode page").len(), 1000); + assert_eq!(reader.rd.reads, 1, "a ready page smaller than 64 KiB needs one stream read"); + } + #[tokio::test] async fn test_reader_invalid_utf8_name_remains_a_sticky_error() { let stream = vec![0xcc, 2, 0xc3, 0xa1, 0xff, 0xc4, 0, 0xc2]; diff --git a/crates/scanner/src/scanner/tests/wakeup_metrics.rs b/crates/scanner/src/scanner/tests/wakeup_metrics.rs index afd742b2c..98dc81454 100644 --- a/crates/scanner/src/scanner/tests/wakeup_metrics.rs +++ b/crates/scanner/src/scanner/tests/wakeup_metrics.rs @@ -188,7 +188,8 @@ async fn scanner_trigger_metrics_record_recovery_and_config_without_rate_limited crate::scanner_io::clear_dirty_usage_buckets_for_tests(); global_metrics().set_cycle(None).await; let (_temp_dir, store) = setup_scanner_cycle_store().await; - let now = scanner_pause_backlog_now(); + // Model a second passing during the durable writer claim without sleeping. + let now = scanner_pause_backlog_now().saturating_sub(1); let mut backlog = ScannerPauseBacklogController::claim(store.clone(), now) .await .expect("seed writer must claim the pause backlog"); @@ -198,7 +199,7 @@ async fn scanner_trigger_metrics_record_recovery_and_config_without_rate_limited backlog.observe(observation).await; observation.paused = false; backlog.observe(observation).await; - let ScannerPauseBacklogAttemptDecision::Tracked(serial) = backlog.begin_attempt(now).await else { + let ScannerPauseBacklogAttemptDecision::Tracked(serial) = backlog.begin_attempt(observation.now_unix_secs).await else { panic!("seeded catch-up must admit its first attempt"); }; backlog.finish_attempt(serial, ScannerCycleOutcome::Failed, observation).await;