From 7dc03603da8cf5192bd59489d51465ede80e8d92 Mon Sep 17 00:00:00 2001 From: houseme Date: Fri, 10 Jul 2026 08:54:59 +0800 Subject: [PATCH] feat(ecstore): add io_uring degradation latch and per-disk probe cache (#4635) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit feat(ecstore): io_uring runtime degradation latch + per-disk probe cache (rustfs/backlog#1101) Follow-up to #1104. Add the runtime half of the io_uring degradation contract: - Runtime latch: UringBackend gains an `active` flag (mirrors DirectIoReadState::supported). A read that returns a restriction-class errno (EPERM/EACCES/ENOSYS/EOPNOTSUPP) — io_uring became unusable on this disk — trips the latch, and every further read goes straight to StdBackend with no more per-read io_uring attempts. Data errors (EIO), missing files, and parameter errors do NOT latch, since StdBackend would hit them too. - Per-disk probe cache: a process-wide negative cache of disks whose probe failed, so try_new skips re-probing (creating a ring + driver thread) on reconnect. Tests: the errno classifier (restriction-only) and a latched-off read that still returns correct bytes via StdBackend; the #1104 differential test still passes under real io_uring. Co-authored-by: heihutu --- crates/ecstore/src/disk/local.rs | 118 +++++++++++++++++++++++++++++-- 1 file changed, 113 insertions(+), 5 deletions(-) diff --git a/crates/ecstore/src/disk/local.rs b/crates/ecstore/src/disk/local.rs index 00d48eb80..ee8a2eeac 100644 --- a/crates/ecstore/src/disk/local.rs +++ b/crates/ecstore/src/disk/local.rs @@ -2288,9 +2288,37 @@ pub(crate) struct UringBackend { root: PathBuf, inner: StdBackend, driver: Arc, + /// Runtime degradation latch (backlog#1101). Starts `true`; once a read + /// returns a restriction-class errno (io_uring became unusable on this + /// disk), it is set `false` and all further reads go straight to + /// `StdBackend` — no more per-read io_uring attempts. Mirrors + /// [`DirectIoReadState::supported`]. + active: std::sync::atomic::AtomicBool, fallback_logged: std::sync::atomic::AtomicBool, } +/// Disks whose io_uring probe failed, so `UringBackend::try_new` can skip +/// re-probing them on reconnect (per-disk probe cache, backlog#1101). A probe +/// creates a ring and spawns a driver thread; caching the negative result +/// avoids repeating that on every `LocalDisk` reconstruction of a disk that +/// does not support io_uring. +#[cfg(target_os = "linux")] +static URING_UNSUPPORTED_DISKS: std::sync::LazyLock>> = + std::sync::LazyLock::new(|| std::sync::Mutex::new(std::collections::HashSet::new())); + +/// True when a runtime io_uring read error means io_uring itself is unusable on +/// this disk (→ latch off), as opposed to a per-read/file error `StdBackend` +/// would hit too (backlog#1101). Restriction-class errnos only; data errors +/// (EIO), missing files, and parameter errors do NOT latch. Mirrors the shape +/// of [`is_direct_io_unsupported`]. +#[cfg(target_os = "linux")] +fn is_io_uring_unsupported(err: &std::io::Error) -> bool { + matches!( + err.raw_os_error(), + Some(libc::EPERM) | Some(libc::EACCES) | Some(libc::ENOSYS) | Some(libc::EOPNOTSUPP) + ) +} + #[cfg(target_os = "linux")] impl std::fmt::Debug for UringBackend { fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { @@ -2306,6 +2334,15 @@ impl UringBackend { /// to `StdBackend`. A restricted-environment errno degrades quietly; an /// unexpected errno is surfaced as a warning (both still fall back). pub(crate) fn try_new(root: PathBuf) -> Option { + // Per-disk probe cache: skip a disk already known not to support + // io_uring (backlog#1101). + if URING_UNSUPPORTED_DISKS + .lock() + .expect("uring probe cache mutex poisoned") + .contains(&root) + { + return None; + } match rustfs_uring::UringDriver::probe_and_start(URING_QUEUE_DEPTH) { Ok(driver) => { info!( @@ -2318,6 +2355,7 @@ impl UringBackend { inner: StdBackend::new(root.clone()), root, driver: Arc::new(driver), + active: std::sync::atomic::AtomicBool::new(true), fallback_logged: std::sync::atomic::AtomicBool::new(false), }) } @@ -2339,6 +2377,10 @@ impl UringBackend { "io_uring probe failed unexpectedly; using StdBackend" ); } + URING_UNSUPPORTED_DISKS + .lock() + .expect("uring probe cache mutex poisoned") + .insert(root); None } } @@ -2378,11 +2420,18 @@ impl UringBackend { return Ok(Bytes::new()); } - let bytes = self - .driver - .read_at(Arc::new(file), offset_u64, length) - .await - .map_err(DiskError::from)?; + let bytes = match self.driver.read_at(Arc::new(file), offset_u64, length).await { + Ok(bytes) => bytes, + Err(io_err) => { + // Latch io_uring off for this disk if the errno says the + // subsystem is unusable (backlog#1101); the caller falls back + // to StdBackend for this and every future read. + if is_io_uring_unsupported(&io_err) { + self.active.store(false, Ordering::Relaxed); + } + return Err(DiskError::from(io_err)); + } + }; if bytes.len() != length { return Err(DiskError::other("io_uring returned a short read")); } @@ -2401,14 +2450,21 @@ impl LocalIoBackend for UringBackend { length: usize, metrics: Option, ) -> Result { + // Latched off (backlog#1101): io_uring proved unusable on this disk, so + // skip it entirely and read via StdBackend. + if !self.active.load(Ordering::Relaxed) { + return self.inner.pread_bytes(volume, path, offset, length, metrics).await; + } match self.pread_uring(volume, path, offset, length).await { Ok(bytes) => Ok(bytes), Err(err) => { if !self.fallback_logged.swap(true, Ordering::Relaxed) { + let latched = !self.active.load(Ordering::Relaxed); debug!( component = LOG_COMPONENT_ECSTORE, subsystem = LOG_SUBSYSTEM_DISK_LOCAL, error = ?err, + latched_off = latched, "io_uring read fell back to StdBackend (logged once per disk)" ); } @@ -11282,6 +11338,58 @@ mod test { } } + /// The runtime degradation latch classifier (backlog#1101): only + /// restriction-class errnos latch io_uring off; data/parameter/missing-file + /// errors do not. + #[cfg(target_os = "linux")] + #[test] + fn io_uring_unsupported_classifies_restriction_errnos_only() { + use std::io::Error; + for errno in [libc::EPERM, libc::EACCES, libc::ENOSYS, libc::EOPNOTSUPP] { + assert!(is_io_uring_unsupported(&Error::from_raw_os_error(errno)), "errno {errno} should latch"); + } + for errno in [libc::EIO, libc::EINVAL, libc::ENOENT, libc::EAGAIN] { + assert!(!is_io_uring_unsupported(&Error::from_raw_os_error(errno)), "errno {errno} must not latch"); + } + assert!(!is_io_uring_unsupported(&Error::other("driver gone"))); + } + + /// Once io_uring is latched off (backlog#1101), reads still return correct + /// bytes via StdBackend. Lay out the disk with LocalDisk, then read the same + /// root through a UringBackend with the latch tripped. + #[cfg(target_os = "linux")] + #[tokio::test(flavor = "multi_thread")] + async fn uring_backend_latched_off_reads_via_std() { + use std::sync::atomic::Ordering; + use tempfile::tempdir; + + let root_dir = tempdir().expect("operation should succeed"); + let root = root_dir.path().to_path_buf(); + let content: Vec = (0..4096u32).map(|i| (i.wrapping_mul(31).wrapping_add(7)) as u8).collect(); + let content = Bytes::from(content); + + { + let endpoint = Endpoint::try_from(root.to_string_lossy().as_ref()).expect("operation should succeed"); + let disk = LocalDisk::new(&endpoint, false).await.expect("operation should succeed"); + disk.make_volume("test-volume").await.expect("operation should succeed"); + disk.write_all("test-volume", "blob.bin", content.clone()) + .await + .expect("operation should succeed"); + } + + // Skip if io_uring is unavailable on this host (restricted env). + let Some(backend) = UringBackend::try_new(root) else { + return; + }; + backend.active.store(false, Ordering::Relaxed); + + let got = backend + .pread_bytes("test-volume", "blob.bin", 100, 512, None) + .await + .expect("latched-off read must succeed via StdBackend"); + assert_eq!(got, content.slice(100..612), "latched-off read returned wrong bytes"); + } + /// The O_DIRECT read path must return the same bytes as the buffered /// path for unaligned shard sizes and ranges, and must silently fall /// back (never error) on filesystems that reject O_DIRECT (e.g. tmpfs).