gsadmin/rustfs
1
0
Fork 0
mirror of https://github.com/rustfs/rustfs.git synced 2026-10-27 14:27:10 +00:00
Code Issues Packages Projects Releases Wiki Activity

perf(ecstore,filemeta): batch listings and buffer metacache streams (#8791)

Browse Source 
* 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 <heihutu@gmail.com>
Co-Authored-By: zhi22915 <qiuzgang@gmail.com>

* 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 <heihutu@gmail.com>
Co-Authored-By: zhi22915 <qiuzgang@gmail.com>

* 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 <heihutu@gmail.com>
Co-Authored-By: zhi22915 <qiuzgang@gmail.com>

* 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 <heihutu@gmail.com>
Co-Authored-By: zhi22915 <qiuzgang@gmail.com>

---------

Co-authored-by: Hauser <housemecn@gmail.com>
Co-authored-by: heihutu <heihutu@gmail.com>
Co-authored-by: zhi22915 <qiuzgang@gmail.com>
This commit is contained in:
Ramakrishna Chilaka
2026-10-22 12:24:34 +05:30
committed by GitHub
parent 5749752217
commit 0bade0ca3e
8 changed files with 834 additions and 128 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/ecstore/Cargo.toml
+4
View File
@@ -286,5 +286,9 @@ harness = false
name = "multipart_read_parts_benchmark"
harness = false
[[bench]]
name = "local_walk_dir_benchmark"
harness = false
[lib]
doctest = false
crates/ecstore/benches/local_walk_dir_benchmark.rs
+85
View File
@@ -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<MetaCacheEntry> {
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);
crates/ecstore/benches/storage_api/mod.rs
+4
View File
@@ -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};
}
crates/ecstore/src/disk/local.rs
+388 -94
View File
@@ -235,9 +235,19 @@ enum ReadAllError {
struct ListingMetadataRead {
bytes: Vec<u8>,
#[cfg(test)]
file_meta: Option<FileMeta>,
data_dirs: HashSet<String>,
data_dirs: Vec<Uuid>,
has_namespace_child_candidate: bool,
counts_toward_limit: bool,
}
type ListingMetadataItem = core::result::Result<ListingMetadataRead, ReadAllError>;
struct ListingMetadataBatch {
volume_dir: PathBuf,
rx: tokio::sync::mpsc::UnboundedReceiver<ListingMetadataItem>,
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<std::fs::File>,
) -> core::result::Result<(Vec<u8>, Option<OffsetDateTime>), 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<ListingMetadataItem>,
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<FileMeta>, Vec<Uuid>, 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<ListingMetadataRead, ReadAllError> {
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<ListingMetadataRead> {
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<String>,
stall: Option<Duration>,
) -> Result<ListingMetadataBatch> {
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<String> = 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<T>(
@@ -9213,8 +9315,10 @@ impl LocalDisk {
}
};
prefix = "".to_owned();
let mut prefetched_batch: Option<ListingMetadataBatch> = 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<String> = 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<u8>, 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<String> = (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]
crates/ecstore/src/disk/os.rs
+95
View File
@@ -2061,11 +2061,13 @@ pub async fn is_empty_dir(path: impl AsRef<Path>) -> bool {
const READ_DIR_PROBE_RAW_LIMIT: usize = 256;
#[cfg(test)]
pub(crate) struct ReadDirProbe {
pub entries: Vec<String>,
pub complete: bool,
}
#[cfg(test)]
pub(crate) fn read_dir_probe(path: impl AsRef<Path>, entry_limit: usize) -> io::Result<ReadDirProbe> {
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<Path>, entry_limit: usize) -> io::
})
}
pub(crate) fn has_non_data_dir_entry(path: impl AsRef<Path>, data_dirs: &[uuid::Uuid]) -> io::Result<bool> {
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<bool> {
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]
crates/filemeta/benches/xl_meta_bench.rs
+93 -1
View File
@@ -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);
crates/filemeta/src/metacache.rs
+162 -31
View File
@@ -911,18 +911,30 @@ impl<W: AsyncWrite + Unpin> MetacacheWriter<W> {
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<u8>, 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<R> {
in_entry: bool,
err: Option<Error>,
buf: Vec<u8>,
pos: usize,
has_current: bool,
}
@@ -953,63 +966,111 @@ impl<R: AsyncRead + Unpin> MetacacheReader<R> {
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<R: AsyncRead + Unpin> MetacacheReader<R> {
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<std::io::Result<()>> {
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];
crates/scanner/src/scanner/tests/wakeup_metrics.rs
+3 -2
View File
@@ -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;