gsadmin/rustfs
1
0
Fork 0
mirror of https://github.com/rustfs/rustfs.git synced 2026-07-28 09:08:58 +00:00
Code Issues Packages Projects Releases Wiki Activity

feat(ecstore): cache part-file descriptors for io_uring reads (#4658)

Browse Source 
* feat(ecstore): run one io_uring ring per shard on each disk (backlog#1145)

A buffered read that hits the page cache completes inline inside
`io_uring_enter`, so the thread driving a ring performs that read's
memcpy. One ring per disk therefore capped cache-hit reads at a single
core's memory bandwidth: measured on a 16-core host, one driver thread sat
pinned at 100% CPU while throughput stayed flat at ~5 GB/s regardless of
read size, against 50 GB/s for the blocking-pool baseline.

rustfs/uring#6 taught the driver to hold N independent rings, each with its
own thread, pending table, backpressure semaphore, and eventfd. Wire it up:
`UringBackend::try_new` now calls `probe_and_start_sharded`, and
`RUSTFS_IO_URING_SHARDS` selects the count per disk.

The default is a quarter of the available parallelism clamped to `1..=4`,
because the cost is `disks × shards` driver threads (each normally blocked
in `poll(2)`). Any override is clamped to `1..=16`, so a mistyped value can
neither disable the driver (0) nor spawn threads without bound; an
unparseable value falls back to the default.

Effect (warm page cache, 16-core, rustfs/uring's concurrent_pread_bench):

  1 MiB, conc 8:    1 shard  4911 MB/s -> 8 shards 47361 MB/s (9.6x);
                    the blocking-pool baseline is 50662 MB/s
  64 KiB, conc 32:  StdBackend 153678 IOPS, p999 3030 us
                    8 shards   345402 IOPS, p999  897 us
  64 KiB, conc 128: StdBackend 135155 IOPS, p999 10716 us
                    8 shards    389047 IOPS, p999  4092 us

Sharding removes the throughput deficit *and* keeps io_uring's tail-latency
advantage, rather than trading one for the other.

Unchanged: io_uring read stays gray-off by default
(`RUSTFS_IO_URING_READ_ENABLE`), reads are byte-for-byte identical to
StdBackend, the per-disk degradation latches and probe cache (backlog#1101)
and the O_DIRECT tiered fallback (backlog#1102) all still apply. Rings stay
per-disk, so a stalled disk cannot starve another disk's rings
(backlog#1055). Bumps the rustfs-uring pin to the merged #6 commit.

Verified on a real Linux host (16-core, real io_uring): cargo clippy
--tests -D warnings clean; disk::local tests 132 passed, 0 failed —
including the existing io_uring and O_DIRECT cases now running on the
sharded driver, plus a new test covering the shard-count default, override,
and clamping.

Co-Authored-By: heihutu <heihutu@gmail.com>

* feat(ecstore): cache part-file descriptors for io_uring reads (backlog#1145)

`pread_uring` opened the file on the blocking pool for every read, so each
read paid a `spawn_blocking` round trip — the very thread hop io_uring
exists to avoid. Sharding the driver (backlog#1145) removed the previous
ceiling and left this as the binding cost. Measured on a 16-core host with
a 4-shard driver, warm page cache:

  64 KiB, conc 8:   143942 -> 263054 IOPS (+83%),  p999 240 -> 65 us
  64 KiB, conc 32:  150128 -> 204876 IOPS (+36%),  p999 2508 -> 871 us
  64 KiB, conc 128: 129172 -> 361287 IOPS (+180%), p999 15329 -> 3046 us
  1 MiB,  conc 32:   33875 ->  42301 IOPS (+25%)

At 64 KiB / conc 128 the open is what masked io_uring entirely: with it,
io_uring beat StdBackend by 3.5%; without it, by 189%.

Add a bounded per-disk descriptor cache used by the io_uring read path.
A hit takes no `open` and no `spawn_blocking`, so the read never leaves the
runtime worker.

Why caching a part-file descriptor is safe:
  * only `<object>/<data_dir>/part.N` reaches this backend's `pread_bytes`;
    `xl.meta` — the one path replaced in place — is read through `read_all`
    / `read_metadata` and never gets here;
  * part files are never rewritten in place. A replacement is always
    write-new-tmp then `rename`, which swaps the inode, so a cached
    descriptor can never observe a torn shard.

Why invalidation is nevertheless REQUIRED: heal reuses the existing
version's `data_dir` and renames a rebuilt shard onto the SAME part path.
A cached descriptor would keep serving the pre-heal (corrupt) inode,
defeating the heal and eroding read quorum. `delete` likewise unlinks a
part that a cached descriptor would keep readable. So `rename_data`,
`rename_file`, and `delete` all call the new
`LocalIoBackend::invalidate_cached_fds` after they mutate, and a 5s TTL
bounds the blast radius should a future mutation path forget to.

Two preamble checks the miss path runs are not silently lost on a hit:
  * bounds — the driver only short-reads at EOF (it resubmits otherwise),
    so `bytes.len() != length` is exactly the old `meta.len() < end_offset`
    check, and now yields the same `FileCorrupt`;
  * volume access — skipped while an entry is live. An unreachable disk
    keeps serving already-open descriptors for at most the TTL, after which
    the re-open re-runs the check. Disk health is tracked independently of
    this per-read probe.

Scope: buffered io_uring reads only. The O_DIRECT path keeps opening per
read (its reads are >= 4 MiB, so the open is a small fraction), and
StdBackend is untouched — it must take the blocking hop for the pread
regardless, so caching would buy it only 2-6% while carrying the same
invalidation risk. `RUSTFS_IO_URING_FD_CACHE=false` restores open-per-read.

Verified on a real Linux host (16-core, real io_uring): clippy --tests
-D warnings clean; disk::local tests 135 passed, 0 failed. The new
heal-staleness test first asserts a read still returns the PRE-heal bytes —
proving the cache is live and the hazard real — then that invalidation makes
the healed shard visible. A second test drives `rename_file` and `delete`
through `LocalDisk` to prove those paths actually invalidate, and a unit
test pins prefix invalidation to component boundaries (`a/b` must not drop
`a/bc`).

Co-Authored-By: heihutu <heihutu@gmail.com>

---------

Co-authored-by: heihutu <heihutu@gmail.com>
This commit is contained in:
houseme
2026-07-10 16:33:50 +08:00
committed by GitHub
parent aaffd3ede8
commit 15b8b13698
1 changed files with 442 additions and 22 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/ecstore/src/disk/local.rs
+442 -22
View File
@@ -1935,6 +1935,20 @@ pub(crate) trait LocalIoBackend: Send + Sync + Debug + 'static {
/// Open a writer (mirrors `create_file`/`append_file` per [`WriteMode`]).
async fn open_write(&self, volume: &str, path: &str, mode: WriteMode) -> Result<FileWriter>;
/// Drop any descriptor this backend caches for exactly `volume`'s `path`
/// (backlog#1145). Preferred wherever the caller knows the affected paths.
///
/// MUST be called by every path that replaces a file a reader may have opened
/// — `rename_data` (heal reuses a version's `data_dir` and lands a rebuilt
/// shard on the same part path) and `rename_file`.
async fn invalidate_cached_fd(&self, _volume: &str, _path: &str) {}
/// Drop every descriptor for `volume`'s `path` and anything beneath it.
/// For callers that cannot enumerate the affected paths (`delete`).
///
/// Backends that hold no descriptors ignore both methods.
fn invalidate_cached_fds_under(&self, _volume: &str, _path: &str) {}
}
/// Default [`LocalIoBackend`]: tokio blocking-pool file I/O plus the
@@ -2549,6 +2563,137 @@ impl LocalIoBackend for StdBackend {
/// Runtime-probed io_uring read backend (backlog#1104).
///
/// Wraps a [`StdBackend`] for everything except positioned reads, which go
/// Enable the per-disk descriptor cache for io_uring reads (backlog#1145).
#[cfg(target_os = "linux")]
const ENV_RUSTFS_IO_URING_FD_CACHE: &str = "RUSTFS_IO_URING_FD_CACHE";
#[cfg(target_os = "linux")]
const DEFAULT_RUSTFS_IO_URING_FD_CACHE: bool = true;
/// Open descriptors kept per disk. Each entry holds an fd, so this bounds the
/// cache's share of `RLIMIT_NOFILE`. moka evicts asynchronously, so the count may
/// briefly exceed this.
#[cfg(target_os = "linux")]
const FD_CACHE_CAPACITY: u64 = 512;
/// Backstop on how long a cached descriptor may serve reads. Explicit
/// invalidation (below) is the correctness mechanism; this only bounds the
/// blast radius if a future mutation path forgets to call it.
#[cfg(target_os = "linux")]
const FD_CACHE_TTL: std::time::Duration = std::time::Duration::from_secs(5);
#[cfg(target_os = "linux")]
fn is_io_uring_fd_cache_enabled() -> bool {
rustfs_utils::get_env_bool(ENV_RUSTFS_IO_URING_FD_CACHE, DEFAULT_RUSTFS_IO_URING_FD_CACHE)
}
/// A cached descriptor is keyed by the open flags too: the O_DIRECT and buffered
/// read paths must never hand each other a descriptor opened the other way.
/// Only the buffered path caches today, so `direct` is always `false`; keeping it
/// in the key stops a future O_DIRECT cache from colliding with this one.
#[cfg(target_os = "linux")]
#[derive(PartialEq, Eq, Hash, Clone)]
struct FdKey {
volume: String,
path: String,
direct: bool,
}
/// Per-disk cache of open descriptors for io_uring reads (backlog#1145).
///
/// Why this exists: `pread_uring` opened the file on the blocking pool for every
/// read, so each read paid a `spawn_blocking` round trip — the very thread hop
/// io_uring exists to avoid. Measured on a 16-core host with a 4-shard driver,
/// removing it is worth +36% to +180% IOPS and 3-5x better p999.
///
/// Why it is safe to cache a *part file* descriptor:
/// - only `<object>/<data_dir>/part.N` reaches this backend's `pread_bytes`;
/// `xl.meta` (the one path replaced in place) is read through `read_all` /
/// `read_metadata` and never gets here;
/// - part files are never rewritten in place — a replacement is always
/// write-new-tmp then `rename`, which swaps the inode, so a cached descriptor
/// can never observe a torn shard.
///
/// Why invalidation is nevertheless REQUIRED: heal reuses the existing version's
/// `data_dir` and renames a reconstructed shard onto the *same* part path. A
/// cached descriptor would keep serving the pre-heal (corrupt) inode, defeating
/// the heal and eroding read quorum. `delete` likewise unlinks the part while a
/// cached descriptor would keep the inode readable.
///
/// Backed by `moka`, already a dependency: its `get` is sharded rather than
/// behind one mutex (this cache is touched on every read, at >300k IOPS), and it
/// evicts by TinyLFU instead of arbitrarily. `time_to_live` supplies the backstop
/// TTL should a future mutation path forget to invalidate; `max_capacity` bounds
/// this cache's share of `RLIMIT_NOFILE`. Eviction drops the `Arc<File>`, closing
/// the descriptor once no in-flight read still holds it.
#[cfg(target_os = "linux")]
struct FdCache {
cache: moka::future::Cache<FdKey, Arc<std::fs::File>>,
}
#[cfg(target_os = "linux")]
impl FdCache {
fn new() -> Self {
Self {
cache: moka::future::Cache::builder()
.max_capacity(FD_CACHE_CAPACITY)
.time_to_live(FD_CACHE_TTL)
// Required for `invalidate_entries_if`; without it that call
// fails at runtime instead of dropping stale descriptors.
.support_invalidation_closures()
.build(),
}
}
async fn get(&self, key: &FdKey) -> Option<Arc<std::fs::File>> {
self.cache.get(key).await
}
async fn insert(&self, key: FdKey, file: Arc<std::fs::File>) {
self.cache.insert(key, file).await;
}
/// Drop the descriptor for exactly this path. Preferred wherever the caller
/// knows the keys: unlike a predicate it costs nothing on later reads.
async fn invalidate_exact(&self, volume: &str, path: &str) {
self.cache
.invalidate(&FdKey {
volume: volume.to_owned(),
path: path.to_owned(),
direct: false,
})
.await;
}
/// Drop every descriptor for `volume` whose path is `prefix` or lies under it.
/// A component-boundary check keeps `a/b` from invalidating `a/bc`.
///
/// moka applies the predicate to entries inserted at or before this call and
/// guarantees a later `get` never returns one of them, so the invalidation is
/// effective immediately even though the removal itself is deferred. Reserved
/// for paths whose exact keys are unknown (`delete`); every predicate is
/// re-evaluated by subsequent `get`s, so registering one per write would tax
/// the read path.
fn invalidate_under(&self, volume: &str, prefix: &str) {
let volume = volume.to_owned();
let prefix = prefix.trim_end_matches('/').to_owned();
let matches = move |k: &FdKey, _: &Arc<std::fs::File>| {
k.volume == volume && (k.path == prefix || k.path.strip_prefix(&prefix).is_some_and(|r| r.starts_with('/')))
};
if self.cache.invalidate_entries_if(matches).is_err() {
// Closure support is enabled at construction, so this is unreachable.
// If it ever changes, over-invalidate rather than serve a stale
// descriptor: correctness first, the cache refills on the next read.
self.cache.invalidate_all();
}
}
#[cfg(test)]
async fn entry_count(&self) -> u64 {
self.cache.run_pending_tasks().await;
self.cache.entry_count()
}
}
/// through rustfs-uring's cancel-safe `UringDriver`. Constructed only when
/// `RUSTFS_IO_URING_READ_ENABLE` is set AND the per-disk probe succeeds; on any
/// per-read driver error a read falls back to the inner `StdBackend`, so
@@ -2575,6 +2720,9 @@ pub(crate) struct UringBackend {
/// `active` gates io_uring as a whole, this gates only the native O_DIRECT
/// read shape.
direct_uring: DirectIoReadState,
/// Per-disk descriptor cache (backlog#1145). `None` when
/// `RUSTFS_IO_URING_FD_CACHE` is off, which restores the open-per-read path.
fd_cache: Option<FdCache>,
}
/// Disks whose io_uring probe failed, so `UringBackend::try_new` can skip
@@ -2640,6 +2788,7 @@ impl UringBackend {
active: std::sync::atomic::AtomicBool::new(true),
fallback_logged: std::sync::atomic::AtomicBool::new(false),
direct_uring: DirectIoReadState::new(),
fd_cache: is_io_uring_fd_cache_enabled().then(FdCache::new),
})
}
Err(err) => {
@@ -2676,34 +2825,67 @@ impl UringBackend {
let Some(end_offset) = offset.checked_add(length) else {
return Err(DiskError::FileCorrupt);
};
let root = self.root.clone();
let volume_owned = volume.to_owned();
let path_owned = path.to_owned();
let offset_u64 = u64::try_from(offset).map_err(|_| DiskError::FileCorrupt)?;
let (file, offset_u64) = tokio::task::spawn_blocking(move || -> Result<(std::fs::File, u64)> {
let volume_dir = local_disk_bucket_path(&root, &volume_owned)?;
if !skip_access_checks(&volume_owned) {
access_std(&volume_dir).map_err(|e| DiskError::from(to_access_error(e, DiskError::VolumeAccessDenied)))?;
// Descriptor-cache hit: no `open`, no `spawn_blocking`, so the read path
// never leaves the runtime worker — which is the whole point of io_uring
// (backlog#1145). The two preamble checks the miss path runs are not
// silently lost:
// * bounds — a file shorter than `end_offset` short-reads, and the
// driver only short-reads at EOF (it resubmits otherwise), so the
// `bytes.len() != length` check below is exactly the old
// `meta.len() < end_offset` check and yields the same `FileCorrupt`;
// * volume access — skipped while an entry is live. A disk that became
// unreachable keeps serving its already-open descriptors for at most
// `FD_CACHE_TTL`, after which the re-open re-runs the check. Disk
// health is tracked independently of this per-read probe.
let key = self.fd_cache.as_ref().map(|_| FdKey {
volume: volume.to_owned(),
path: path.to_owned(),
direct: false,
});
let cached = match (self.fd_cache.as_ref(), key.as_ref()) {
(Some(cache), Some(key)) => cache.get(key).await,
_ => None,
};
let file = match cached {
Some(file) => file,
None => {
let root = self.root.clone();
let volume_owned = volume.to_owned();
let path_owned = path.to_owned();
let file = tokio::task::spawn_blocking(move || -> Result<std::fs::File> {
let volume_dir = local_disk_bucket_path(&root, &volume_owned)?;
if !skip_access_checks(&volume_owned) {
access_std(&volume_dir)
.map_err(|e| DiskError::from(to_access_error(e, DiskError::VolumeAccessDenied)))?;
}
let file_path = local_disk_object_path(&root, &volume_owned, &path_owned)?;
check_path_length(file_path.to_string_lossy().as_ref())?;
let file = std::fs::File::open(&file_path).map_err(DiskError::from)?;
let meta = file.metadata().map_err(DiskError::from)?;
let end_offset_u64 = u64::try_from(end_offset).map_err(|_| DiskError::FileCorrupt)?;
if meta.len() < end_offset_u64 {
return Err(DiskError::FileCorrupt);
}
Ok(file)
})
.await
.map_err(|e| DiskError::other(format!("uring pread join error: {e}")))??;
let file = Arc::new(file);
if let (Some(cache), Some(key)) = (self.fd_cache.as_ref(), key) {
cache.insert(key, Arc::clone(&file)).await;
}
file
}
let file_path = local_disk_object_path(&root, &volume_owned, &path_owned)?;
check_path_length(file_path.to_string_lossy().as_ref())?;
let file = std::fs::File::open(&file_path).map_err(DiskError::from)?;
let meta = file.metadata().map_err(DiskError::from)?;
let end_offset_u64 = u64::try_from(end_offset).map_err(|_| DiskError::FileCorrupt)?;
if meta.len() < end_offset_u64 {
return Err(DiskError::FileCorrupt);
}
let offset_u64 = u64::try_from(offset).map_err(|_| DiskError::FileCorrupt)?;
Ok((file, offset_u64))
})
.await
.map_err(|e| DiskError::other(format!("uring pread join error: {e}")))??;
};
if length == 0 {
return Ok(Bytes::new());
}
let bytes = match self.driver.read_at(Arc::new(file), offset_u64, length).await {
let bytes = match self.driver.read_at(file, offset_u64, length).await {
Ok(bytes) => bytes,
Err(io_err) => {
// Latch io_uring off for this disk if the errno says the
@@ -2716,7 +2898,11 @@ impl UringBackend {
}
};
if bytes.len() != length {
return Err(DiskError::other("io_uring returned a short read"));
// The driver resubmits short reads, so a short result means EOF: the
// file is shorter than `offset + length`. That is precisely what the
// miss path's `meta.len() < end_offset` check rejects, so report the
// same error whether or not the descriptor came from the cache.
return Err(DiskError::FileCorrupt);
}
Ok(Bytes::from(bytes))
}
@@ -2909,6 +3095,18 @@ impl LocalIoBackend for UringBackend {
async fn open_write(&self, volume: &str, path: &str, mode: WriteMode) -> Result<FileWriter> {
self.inner.open_write(volume, path, mode).await
}
async fn invalidate_cached_fd(&self, volume: &str, path: &str) {
if let Some(cache) = self.fd_cache.as_ref() {
cache.invalidate_exact(volume, path).await;
}
}
fn invalidate_cached_fds_under(&self, volume: &str, path: &str) {
if let Some(cache) = self.fd_cache.as_ref() {
cache.invalidate_under(volume, path);
}
}
}
/// Select the local read backend: the runtime-probed io_uring backend when
@@ -5268,6 +5466,12 @@ impl DiskAPI for LocalDisk {
self.delete_file(&volume_dir, &file_path, opt.recursive, opt.immediate)
.await?;
// The inode is unlinked, but a cached descriptor would keep it readable —
// a deleted shard must not keep answering reads (backlog#1145). The part
// numbers under `path` are not known here, so this is the one caller that
// needs the predicate form.
self.io_backend.invalidate_cached_fds_under(volume, path);
Ok(())
}
@@ -5685,6 +5889,11 @@ impl DiskAPI for LocalDisk {
rename_all(&src_file_path, &dst_file_path, &dst_volume_dir).await?;
// Both ends changed identity: the source path no longer exists and the
// destination now resolves to a different inode (backlog#1145).
self.io_backend.invalidate_cached_fd(src_volume, src_path).await;
self.io_backend.invalidate_cached_fd(dst_volume, dst_path).await;
if let Some(parent) = src_file_path.parent() {
let _ = self.delete_file(&src_volume_dir, &parent.to_path_buf(), false, false).await;
}
@@ -5889,6 +6098,17 @@ impl DiskAPI for LocalDisk {
dst_path: &str,
) -> Result<RenameDataResp> {
crate::hp_guard!("LocalDisk::rename_data");
// Snapshot the destination part paths before `fi` is consumed below. These
// are the descriptors a reader may hold for the version this call is about
// to replace (backlog#1145); readers build the identical string in
// `io_primitives`. An inline-data version has no parts and yields none.
let invalidate_part_paths: Vec<String> = {
let data_dir = fi.data_dir.unwrap_or_default();
fi.parts
.iter()
.map(|part| format!("{dst_path}/{data_dir}/part.{}", part.number))
.collect()
};
let src_volume_dir = self.get_bucket_path(src_volume)?;
if !skip_access_checks(src_volume)
&& let Err(e) = super::fs::access_std(&src_volume_dir)
@@ -6228,6 +6448,21 @@ impl DiskAPI for LocalDisk {
}
}
// Heal reuses a version's `data_dir` and lands the rebuilt shard on
// the SAME `<object>/<data_dir>/part.N` path. Without this, a cached
// descriptor would keep serving the pre-heal inode, defeating the heal
// and eroding read quorum (backlog#1145).
//
// The exact keys are derivable here, and this runs on every write, so
// use them rather than registering a predicate the read path would then
// have to evaluate. Readers build the same string
// (`{object}/{data_dir}/part.{n}`), and `fi.parts` enumerates every
// part of the version now at `dst_path` — any part path absent from it
// no longer exists for readers to ask for.
for part_path in &invalidate_part_paths {
self.io_backend.invalidate_cached_fd(dst_volume, part_path).await;
}
Ok(RenameDataResp {
old_data_dir: rollback_data_dir,
sign: version_signature,
@@ -6407,6 +6642,21 @@ impl DiskAPI for LocalDisk {
let _ = remove_std(parent);
}
// Heal reuses a version's `data_dir` and lands the rebuilt shard on
// the SAME `<object>/<data_dir>/part.N` path. Without this, a cached
// descriptor would keep serving the pre-heal inode, defeating the heal
// and eroding read quorum (backlog#1145).
//
// The exact keys are derivable here, and this runs on every write, so
// use them rather than registering a predicate the read path would then
// have to evaluate. Readers build the same string
// (`{object}/{data_dir}/part.{n}`), and `fi.parts` enumerates every
// part of the version now at `dst_path` — any part path absent from it
// no longer exists for readers to ask for.
for part_path in &invalidate_part_paths {
self.io_backend.invalidate_cached_fd(dst_volume, part_path).await;
}
Ok(RenameDataResp {
old_data_dir,
sign: version_signature,
@@ -12789,6 +13039,176 @@ mod test {
});
}
/// Prefix invalidation must respect component boundaries: dropping the
/// descriptors for object `a/b` must not also drop those for `a/bc`.
/// moka applies the predicate lazily, so this also pins the guarantee we rely
/// on — a `get` after `invalidate_under` never returns a matched entry.
#[cfg(target_os = "linux")]
#[tokio::test(flavor = "multi_thread")]
async fn fd_cache_prefix_invalidation_respects_component_boundary() {
use std::fs::File;
let dir = tempfile::tempdir().expect("operation should succeed");
let make = || Arc::new(File::create(dir.path().join("f")).expect("operation should succeed"));
let cache = FdCache::new();
let key = |path: &str| FdKey {
volume: "v".into(),
path: path.into(),
direct: false,
};
for path in ["a/b", "a/b/dir/part.1", "a/bc", "a/bc/dir/part.1"] {
cache.insert(key(path), make()).await;
}
assert_eq!(cache.entry_count().await, 4);
cache.invalidate_under("v", "a/b");
// `a/b` itself and everything under it go; the sibling `a/bc` stays.
assert!(cache.get(&key("a/b")).await.is_none(), "the prefix itself must be dropped");
assert!(cache.get(&key("a/b/dir/part.1")).await.is_none(), "children must be dropped");
assert!(cache.get(&key("a/bc")).await.is_some(), "a/bc is not under a/b");
assert!(cache.get(&key("a/bc/dir/part.1")).await.is_some(), "a/bc/... is not under a/b");
// A different volume with the same path must be untouched.
cache.insert(key("a/b"), make()).await;
cache.invalidate_under("other", "a/b");
assert!(cache.get(&key("a/b")).await.is_some(), "invalidation must be scoped to its volume");
// Exact invalidation drops only its own key.
cache.invalidate_exact("v", "a/bc").await;
assert!(cache.get(&key("a/bc")).await.is_none(), "exact key must be dropped");
assert!(
cache.get(&key("a/bc/dir/part.1")).await.is_some(),
"exact invalidation must not touch children"
);
}
/// The descriptor cache (backlog#1145) must not let a healed shard be masked
/// by a stale descriptor.
///
/// Heal reuses the version's `data_dir` and renames a rebuilt shard onto the
/// SAME `<object>/<data_dir>/part.N` path. This test reproduces exactly that:
/// read the part (which caches its descriptor), swap the file for new content
/// by rename, and assert that a read still returns the OLD bytes — proving the
/// hazard is real and the cache is actually in play — then invalidate and
/// assert the healed bytes become visible.
#[cfg(target_os = "linux")]
#[tokio::test(flavor = "multi_thread")]
async fn uring_fd_cache_hides_a_healed_shard_until_invalidated() {
use tempfile::tempdir;
let root_dir = tempdir().expect("operation should succeed");
let root = root_dir.path().to_path_buf();
let Some(backend) = temp_env::with_vars(
[
(ENV_RUSTFS_IO_URING_READ_ENABLE, Some("true")),
(ENV_RUSTFS_IO_URING_FD_CACHE, Some("true")),
],
|| UringBackend::try_new(root.clone()),
) else {
// Restricted environment (CI seccomp): io_uring is unavailable, so
// there is no descriptor cache to exercise. Do not vacuously pass.
eprintln!("SKIP uring_fd_cache_hides_a_healed_shard_until_invalidated: io_uring unavailable");
return;
};
assert!(backend.fd_cache.is_some(), "the cache must be on for this test to mean anything");
let volume = "bucket";
let object = "obj/0d1e2f/part.1";
let dir = root.join(volume).join("obj/0d1e2f");
std::fs::create_dir_all(&dir).expect("operation should succeed");
let part = root.join(volume).join(object);
std::fs::write(&part, b"corrupt-shard").expect("operation should succeed");
let before = backend
.pread_bytes(volume, object, 0, b"corrupt-shard".len(), None)
.await
.expect("operation should succeed");
assert_eq!(before, Bytes::from_static(b"corrupt-shard"));
// Heal: write the rebuilt shard beside the old one and rename it into
// place, exactly as rename_data does. The path is unchanged; the inode is not.
let rebuilt = dir.join("part.1.rebuilt");
std::fs::write(&rebuilt, b"healed--shard").expect("operation should succeed");
std::fs::rename(&rebuilt, &part).expect("operation should succeed");
let stale = backend
.pread_bytes(volume, object, 0, b"healed--shard".len(), None)
.await
.expect("operation should succeed");
assert_eq!(
stale,
Bytes::from_static(b"corrupt-shard"),
"a cached descriptor is expected to still see the pre-heal inode — this is the hazard \
invalidate_cached_fds exists to close, and the assertion proves the cache is live"
);
backend.invalidate_cached_fds_under(volume, "obj/0d1e2f");
let healed = backend
.pread_bytes(volume, object, 0, b"healed--shard".len(), None)
.await
.expect("operation should succeed");
assert_eq!(
healed,
Bytes::from_static(b"healed--shard"),
"after invalidation the healed shard must be visible"
);
}
/// The mutation paths on `LocalDisk` must actually call
/// `invalidate_cached_fds`, not merely have it available (backlog#1145).
/// `rename_file` replaces the inode at a path a reader has already cached;
/// `delete` unlinks it while a cached descriptor would keep it readable.
#[cfg(target_os = "linux")]
#[tokio::test(flavor = "multi_thread")]
async fn uring_fd_cache_is_invalidated_by_rename_file_and_delete() {
use tempfile::tempdir;
let root_dir = tempdir().expect("operation should succeed");
temp_env::async_with_vars(
[
(ENV_RUSTFS_IO_URING_READ_ENABLE, Some("true")),
(ENV_RUSTFS_IO_URING_FD_CACHE, Some("true")),
],
async {
let endpoint = Endpoint::try_from(root_dir.path().to_string_lossy().as_ref()).expect("operation should succeed");
let disk = LocalDisk::new(&endpoint, false).await.expect("operation should succeed");
disk.make_volume("bucket").await.expect("operation should succeed");
disk.write_all("bucket", "obj/part.1", Bytes::from_static(b"v1"))
.await
.expect("operation should succeed");
let got = disk
.read_file_mmap_copy("bucket", "obj/part.1", 0, 2)
.await
.expect("operation should succeed");
assert_eq!(got, Bytes::from_static(b"v1"), "first read seeds the descriptor cache");
// rename_file over the same destination path: the inode changes.
disk.write_all("bucket", "staging/part.1", Bytes::from_static(b"v2"))
.await
.expect("operation should succeed");
disk.rename_file("bucket", "staging/part.1", "bucket", "obj/part.1")
.await
.expect("operation should succeed");
let got = disk
.read_file_mmap_copy("bucket", "obj/part.1", 0, 2)
.await
.expect("operation should succeed");
assert_eq!(got, Bytes::from_static(b"v2"), "rename_file must invalidate the cached descriptor");
// delete must stop the path from answering reads at all.
disk.delete("bucket", "obj/part.1", DeleteOptions::default())
.await
.expect("operation should succeed");
let err = disk.read_file_mmap_copy("bucket", "obj/part.1", 0, 2).await;
assert!(
err.is_err(),
"a deleted part must not keep answering reads from a cached descriptor, got {err:?}"
);
},
)
.await;
}
/// O_DIRECT interop (backlog#1102): with BOTH io_uring and O_DIRECT enabled,
/// an O_DIRECT-eligible read keeps O_DIRECT semantics via the native
/// `read_at_direct` path (or, if that disk can't do io_uring+O_DIRECT, the