From e1382e88b6c7ebea3d11ce5896f3047f0fc39a51 Mon Sep 17 00:00:00 2001 From: houseme Date: Sat, 11 Jul 2026 14:18:50 +0800 Subject: [PATCH 01/15] fix(driver): fail stranded callers before bounded-drain leak (rustfs/backlog#1161) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The bounded-drain escape hatch did `sem.close(); mem::forget(state)`, which forgot each pending op's oneshot `done` sender without dropping or sending on it. A ReadHandle still awaited after shutdown then polled its receiver forever: the oneshot neither completed nor errored, so the task pended permanently — the one driver-gone path that became an infinite hang instead of an error, breaking the crate's error-never-hang contract. Fail every stranded sender with a driver-gone error before forgetting the pending table. `oneshot::Sender::send` consumes the sender without touching the kernel-owned buffer, so the leak-over-UAF property is preserved while an awaited handle resolves with an error. Add a fault-injection test that awaits a handle across the bailout and asserts it resolves with an error within a bounded window. Co-Authored-By: heihutu --- src/driver.rs | 19 +++++++++++--- tests/fault_injection.rs | 57 ++++++++++++++++++++++++++++++++++++++++ 2 files changed, 73 insertions(+), 3 deletions(-) diff --git a/src/driver.rs b/src/driver.rs index e4df6eb..57acfda 100644 --- a/src/driver.rs +++ b/src/driver.rs @@ -1412,9 +1412,22 @@ fn drive( leaking ring + buffers to stay memory-safe", state.pending.len() ); - // Close the semaphore so any handle awaiting a permit resolves - // with a driver-gone error. The leaked pending entries keep - // their permits, which is fine: nothing will wait on them. + // Fail every stranded caller BEFORE leaking the pending table. + // `oneshot::Sender::send` consumes the sender and never touches + // `p.buf`, so the kernel-owned buffer stays allocated (leak over + // UAF preserved) while an awaited `ReadHandle` resolves with an + // error instead of pending forever — every other driver-gone path + // already delivers an error, and this one must too + // (rustfs/backlog#1161). + for p in state.pending.values_mut() { + if let Some(tx) = p.done.take() { + let _ = tx.send(Err(io::Error::other("uring driver leaked op on bounded-drain timeout"))); + } + } + // Close the semaphore so any handle still awaiting a permit + // resolves with a driver-gone error too. The leaked pending + // entries keep their permits, which is fine: nothing waits on + // them any more. sem.close(); std::mem::forget(state); return; diff --git a/tests/fault_injection.rs b/tests/fault_injection.rs index 9dfad04..07acd9b 100644 --- a/tests/fault_injection.rs +++ b/tests/fault_injection.rs @@ -190,6 +190,63 @@ fn bounded_drain_bails_out_and_leaks_on_a_stuck_op() { drop(pipe_write); } +/// #1161: the bounded-drain bailout must FAIL every stranded caller before it +/// leaks the pending table, so a `ReadHandle` still awaited across the timeout +/// resolves with an error instead of hanging forever. Same stuck-op seam as +/// above, but this time the handle is kept and awaited after shutdown. +#[test] +fn stranded_handle_errors_after_bounded_drain_bailout() { + // SAFETY: `--test-threads=1` serializes tests; no concurrent env readers. + unsafe { + std::env::set_var("RUSTFS_URING_FAULT_STUCK_DRAIN", "1"); + std::env::set_var("RUSTFS_URING_FAULT_DRAIN_TIMEOUT_MS", "400"); + } + + let driver = match UringDriver::probe_and_start(64) { + Ok(d) => d, + Err(e) => { + clear_stuck_env(); + assert!( + e.is_expected_restriction(), + "probe failed OUTSIDE the expected restriction errno class: {e:?}" + ); + eprintln!("SKIP stranded_handle_errors_after_bounded_drain_bailout: restricted environment ({e:?})"); + return; + } + }; + + let (pipe_read, pipe_write) = os_pipe(); + // Keep the handle (do not drop it): its oneshot receiver must still be live + // when the bailout runs so we can prove the bailout delivered an error. + let handle = driver.read_current(Arc::clone(&pipe_read), 4096).without_cancel_on_drop(); + assert!( + wait_until(Duration::from_secs(2), || driver.stats().in_flight == 1), + "read never reached in-flight state" + ); + + // Drive the stuck op to the bounded-drain bailout. + let snap = driver.shutdown(); + clear_stuck_env(); + assert!( + snap.in_flight >= 1, + "the stuck op should still count as in flight after the leak-over-UAF bailout: {snap:?}" + ); + + // The still-awaited handle MUST resolve with an error within a bounded + // window — never hang (rustfs/backlog#1161). Before the fix the forgotten + // oneshot sender left this future pending forever and the timeout would fire. + let rt = tokio::runtime::Builder::new_current_thread() + .enable_time() + .build() + .expect("build current-thread runtime"); + match rt.block_on(async { tokio::time::timeout(Duration::from_secs(2), handle).await }) { + Ok(Err(_)) => {} // driver-gone error delivered — correct. + Ok(Ok(_)) => panic!("stranded handle unexpectedly succeeded on a stuck op"), + Err(_) => panic!("stranded handle HUNG after bounded-drain bailout (rustfs/backlog#1161 regression)"), + } + drop(pipe_write); +} + fn clear_stuck_env() { // SAFETY: `--test-threads=1` teardown; no concurrent environment readers. unsafe { From 04c8c86cefdcaf500b5752d95b96046ca0c31d1f Mon Sep 17 00:00:00 2001 From: houseme Date: Sat, 11 Jul 2026 14:22:20 +0800 Subject: [PATCH 02/15] fix(driver): classify ring.submit() errors instead of retrying forever (rustfs/backlog#1162) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Every non-EBUSY submit() error hit a silent "retry next turn" arm. That is correct for EINTR, but a persistent failure (e.g. EPERM from a seccomp/LSM policy applied after startup) means the queued SQEs are never accepted: their CQEs never arrive, their callers' oneshots never resolve, and — since the driver is not shutting down — the drain deadline never arms. The loop spins at the 50 ms heartbeat forever with no error, log, counter, or fallback. Classify submit errors: EBUSY (reap backpressure) and EINTR (signal) stay transient and reset the counter; any other errno increments a new submit_errors stat, logs once, and after MAX_CONSECUTIVE_SUBMIT_ERRORS consecutive failures transitions the shard to shutdown so the drain and the bounded-drain bailout fail every pending caller with an error — they fall back to the std backend instead of stalling silently. Co-Authored-By: heihutu --- src/driver.rs | 59 +++++++++++++++++++++++++++++++++++++++++++++++---- 1 file changed, 55 insertions(+), 4 deletions(-) diff --git a/src/driver.rs b/src/driver.rs index 57acfda..9b7bbe2 100644 --- a/src/driver.rs +++ b/src/driver.rs @@ -82,6 +82,13 @@ fn aligned_geometry(offset: u64, len: usize, align: usize) -> Option<(u64, usize /// deadline is still checked and any queued cancel is picked up promptly. const LOOP_HEARTBEAT: Duration = Duration::from_millis(50); +/// Consecutive non-transient `ring.submit()` failures the driver tolerates +/// before it stops retrying silently and shuts the shard down, so callers get a +/// driver-gone error and fall back to the std backend instead of stalling +/// forever on ops the kernel will never accept (rustfs/backlog#1162). With the +/// 50 ms heartbeat this bounds the silent-retry window to a few seconds. +const MAX_CONSECUTIVE_SUBMIT_ERRORS: u32 = 128; + /// Owned `eventfd(2)` used to wake the driver loop (backlog#1102): one is /// registered with the ring so the kernel signals it on every CQE, the other is /// signaled by `submit`/shutdown so a new message wakes the loop immediately — @@ -237,6 +244,7 @@ struct DriverStats { cancel_not_found: AtomicU64, cancel_already: AtomicU64, cq_overflow: AtomicU64, + submit_errors: AtomicU64, } /// Point-in-time copy of the driver counters. @@ -266,6 +274,12 @@ pub struct StatsSnapshot { /// CQEs were lost, so their pending entries are never reclaimed and drain /// never completes. Treated as fatal (C5, rustfs/backlog#1056). pub cq_overflow: u64, + /// `ring.submit()` calls that returned a non-transient error. A rising count + /// means `io_uring_enter` is persistently failing (e.g. a seccomp/LSM policy + /// applied after startup); the driver shuts the shard down after a bounded + /// run of consecutive failures so callers fall back instead of stalling + /// (rustfs/backlog#1162). + pub submit_errors: u64, } enum Msg { @@ -826,6 +840,7 @@ impl UringDriver { snap.cancel_not_found += s.cancel_not_found.load(Ordering::SeqCst); snap.cancel_already += s.cancel_already.load(Ordering::SeqCst); snap.cq_overflow += s.cq_overflow.load(Ordering::SeqCst); + snap.submit_errors += s.submit_errors.load(Ordering::SeqCst); } snap } @@ -1101,6 +1116,10 @@ fn drive( }; let mut shutting_down = false; let mut drain_deadline: Option = None; + // Consecutive non-transient submit failures, and a once-only log latch, for + // the persistent-submit-failure escape hatch (rustfs/backlog#1162). + let mut consecutive_submit_errors: u32 = 0; + let mut submit_error_logged = false; // Bounded-drain deadline (C4, rustfs/backlog#1055). Production always uses the // fixed DRAIN_TIMEOUT; a fault-injection build may shorten it via env so the @@ -1264,15 +1283,47 @@ fn drive( } } match state.ring.submit() { - Ok(_) => {} + Ok(_) => consecutive_submit_errors = 0, Err(e) if e.raw_os_error() == Some(libc::EBUSY) => { // CQ-overflow backpressure on pre-5.19 NODROP kernels: the // kernel refuses new submissions until we reap. Keep the // backlog and reap this turn instead of spinning (C5, - // rustfs/backlog#1056). + // rustfs/backlog#1056). Transient — not a submit failure. + consecutive_submit_errors = 0; + } + Err(e) if e.raw_os_error() == Some(libc::EINTR) => { + // A signal interrupted the enter and the SQEs were not consumed; + // retry on the next loop turn (transient). + consecutive_submit_errors = 0; } - Err(_) => { - // EINTR and friends: retry on the next loop turn. + Err(e) => { + // Any other errno: the queued SQEs were not accepted, so their + // CQEs never arrive and their callers would hang. A brief run may + // be transient (EAGAIN under memory pressure); a persistent one + // (e.g. EPERM from a seccomp/LSM policy applied after startup) + // must NOT be retried forever in silence. Count it, log once, and + // after a bounded run of consecutive failures shut the shard down + // so the drain + bounded-drain bailout fail every pending caller + // with an error — they fall back to the std backend + // (rustfs/backlog#1162). + stats.submit_errors.fetch_add(1, Ordering::SeqCst); + consecutive_submit_errors += 1; + if !submit_error_logged { + submit_error_logged = true; + eprintln!("uring-spike driver: ring.submit() failed ({e}); retrying, will shut down if persistent"); + } + if !shutting_down && consecutive_submit_errors >= MAX_CONSECUTIVE_SUBMIT_ERRORS { + eprintln!( + "uring-spike driver: {consecutive_submit_errors} consecutive submit failures; \ + shutting down so callers fall back to the std backend" + ); + shutting_down = true; + for id in state.pending.keys() { + state + .backlog + .push_back(opcode::AsyncCancel::new(*id).build().user_data(*id | CANCEL_BIT)); + } + } } } From ed82d833a33f7a97ac85ef1d4f7d5ce8d95ea9af Mon Sep 17 00:00:00 2001 From: houseme Date: Sat, 11 Jul 2026 14:28:06 +0800 Subject: [PATCH 03/15] fix(driver): wake the loop on drop-cancel and flush resubmits same-turn (rustfs/backlog#1163) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Two same-root wakeup gaps let work sit idle for a full LOOP_HEARTBEAT (50 ms): - A short-read resubmit only did backlog.push_back(sqe) during reap, which runs after the turn's push+submit, so the remainder was not handed to the kernel until the next heartbeat woke the loop. Flush the backlog and submit once more after reap when it is non-empty (idle turns skip it). - ReadHandle::drop sent Msg::Cancel without signaling wake_efd — the Submitted state carried no wake reference — so on an otherwise idle ring the cancel was not even queued for up to 50 ms. Carry the shard's wake_efd in HandleState::Submitted and signal it after the cancel send. Co-Authored-By: heihutu --- src/driver.rs | 61 ++++++++++++++++++++++++++++++++++++++++++++++----- 1 file changed, 56 insertions(+), 5 deletions(-) diff --git a/src/driver.rs b/src/driver.rs index 9b7bbe2..ef1564f 100644 --- a/src/driver.rs +++ b/src/driver.rs @@ -383,7 +383,12 @@ enum HandleState { }, /// The op is with the driver: its buffer lives in the pending table and is /// reclaimed only at the CQE. - Submitted, + Submitted { + /// The accepting shard's wakeup eventfd, so a cancel sent on drop wakes + /// the driver loop now instead of after the heartbeat + /// (rustfs/backlog#1163). + wake: Arc, + }, } /// Handle to a read. Await it for the result. @@ -436,6 +441,13 @@ impl Future for ReadHandle { this.finished = true; return Poll::Ready(Err(io::Error::other("uring driver shut down"))); }; + // Clone the wake before moving the WaitingPermit out, so the new + // Submitted state carries it for the drop-cancel path + // (rustfs/backlog#1163). + let submitted_wake = match &this.state { + HandleState::WaitingPermit { wake, .. } => Arc::clone(wake), + _ => unreachable!("state was WaitingPermit"), + }; let HandleState::WaitingPermit { file, offset, @@ -444,7 +456,7 @@ impl Future for ReadHandle { done, wake, .. - } = std::mem::replace(&mut this.state, HandleState::Submitted) + } = std::mem::replace(&mut this.state, HandleState::Submitted { wake: submitted_wake }) else { unreachable!("state was WaitingPermit") }; @@ -488,8 +500,22 @@ impl Drop for ReadHandle { // until the CQE. All we may do is ask the kernel to hurry up. A handle // dropped before it was submitted (Inert / WaitingPermit) has no buffer, // no permit and no SQE, so there is nothing to cancel. - if matches!(self.state, HandleState::Submitted) && !self.finished && self.cancel_on_drop { - let _ = self.tx.send(Msg::Cancel { id: self.id }); + if let HandleState::Submitted { wake } = &self.state { + if !self.finished && self.cancel_on_drop { + // Close the receiver BEFORE waking the driver. The wake below + // makes the driver process the cancel immediately, possibly while + // this drop is still running — before the `rx` field is + // destroyed. Closing it first guarantees the cancel-induced + // completion the driver reaps is counted as an orphan reclaim, not + // delivered to a receiver that is about to drop anyway + // (rustfs/backlog#1163). + self.rx.close(); + let _ = self.tx.send(Msg::Cancel { id: self.id }); + // Wake the loop so the cancel is queued now, not after the + // heartbeat. On an idle ring (the hung-disk case cancel-on-drop + // exists for) this keeps orphan reclamation prompt. + wake.signal(); + } } } } @@ -787,7 +813,9 @@ impl UringDriver { tx: shard.tx.clone(), finished: false, cancel_on_drop: true, - state: HandleState::Submitted, + state: HandleState::Submitted { + wake: Arc::clone(&shard.wake_efd), + }, } } // Saturated: `entries` ops are already in flight. Do NOT block the @@ -1430,6 +1458,29 @@ fn drive( } } + // A short-read resubmit queued during reap must reach the kernel in THIS + // turn, not wait out the next heartbeat: reap runs after the earlier + // push+submit, so without this flush the remainder sits idle for up to + // LOOP_HEARTBEAT (rustfs/backlog#1163). Only the resubmit/backlog-residue + // case makes this non-empty, so a fully idle turn skips it. + if !state.backlog.is_empty() { + { + let mut sq = state.ring.submission(); + while let Some(sqe) = state.backlog.pop_front() { + // SAFETY: read SQEs point into `pending`-owned buffers that + // live until their CQE; cancel SQEs carry no pointers. + if unsafe { sq.push(&sqe) }.is_err() { + state.backlog.push_front(sqe); + break; + } + } + } + // A persistent submit failure is already counted and acted on by the + // main submit above (rustfs/backlog#1162); this same-turn flush just + // retries opportunistically, so a transient error here is fine. + let _ = state.ring.submit(); + } + // Monitor CQ overflow. With NODROP (asserted at probe) the crate's // submit() auto-flushes the kernel overflow list, so this should stay // 0; any non-zero value means CQEs were lost — pending entries would From dd80fafe597e797e3c8897700610e0b8de8b3869 Mon Sep 17 00:00:00 2001 From: houseme Date: Sat, 11 Jul 2026 14:32:49 +0800 Subject: [PATCH 04/15] fix(driver): degrade instead of panicking on driver-thread spawn failure (rustfs/backlog#1164) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit start_shard mapped every fallible step into ProbeFailure so try_new can fall back to the std backend — except the spawn, which used .expect(). Thread creation fails with EAGAIN under a cgroup pids-limit or RLIMIT_NPROC, exactly the constrained environments the degradation design exists for, and the panic unwound async disk init/reconnect instead of quietly selecting std. Map the spawn error to ProbeFailure::Setup and propagate with ?. The spawn runs after the probe read already drained, so on failure the ring/eventfds moved into the closure drop cleanly with no SQE in flight. Add a fault-injection seam and a test asserting a forced spawn failure degrades (no panic) and leaves global state intact for a later normal probe. Co-Authored-By: heihutu --- src/driver.rs | 16 +++++++++++++++- tests/fault_injection.rs | 40 ++++++++++++++++++++++++++++++++++++++++ 2 files changed, 55 insertions(+), 1 deletion(-) diff --git a/src/driver.rs b/src/driver.rs index ef1564f..569551d 100644 --- a/src/driver.rs +++ b/src/driver.rs @@ -653,10 +653,24 @@ impl UringDriver { // (2*entries), so CQ overflow is structurally unreachable (C5/C10). let sem = Arc::new(Semaphore::new(entries as usize)); let thread_sem = Arc::clone(&sem); + // Deterministic spawn-failure seam (rustfs/backlog#1164): exercise the + // degrade-not-panic path without a real cgroup pids-limit. Never present + // in a default build. + #[cfg(feature = "fault-injection")] + if std::env::var_os("RUSTFS_URING_FAULT_SPAWN").is_some() { + return Err(ProbeFailure::Setup(io::Error::from_raw_os_error(libc::EAGAIN))); + } + + // Thread creation fails with EAGAIN under a cgroup pids-limit or + // RLIMIT_NPROC — exactly the constrained environments the probe/degrade + // design exists for. Degrade to the std backend instead of panicking out + // of async disk init/reconnect (rustfs/backlog#1164). The spawn happens + // after the probe read already drained, so on failure `ring`/`cq_efd` + // (moved into the closure) drop cleanly with no SQE in flight. let handle = std::thread::Builder::new() .name("uring-spike-driver".into()) .spawn(move || drive(ring, rx, thread_stats, thread_sem, cq_efd, thread_wake)) - .expect("spawn driver thread"); + .map_err(ProbeFailure::Setup)?; Ok(Shard { tx, diff --git a/tests/fault_injection.rs b/tests/fault_injection.rs index 07acd9b..3500d2c 100644 --- a/tests/fault_injection.rs +++ b/tests/fault_injection.rs @@ -295,3 +295,43 @@ fn probe_drain_failure_leaks_and_degrades() { } } } + +/// #1164: driver-thread spawn failure (EAGAIN under a cgroup pids-limit / +/// RLIMIT_NPROC) must degrade to a `ProbeFailure` so the caller selects the std +/// backend — never panic out of async disk init. The seam forces the spawn step +/// to fail; the probe must return an error, not unwind, and leave global state +/// intact for a later normal probe. +#[test] +fn spawn_failure_degrades_instead_of_panicking() { + // SAFETY: `--test-threads=1` serializes tests; no concurrent env readers. + unsafe { + std::env::set_var("RUSTFS_URING_FAULT_SPAWN", "1"); + } + let result = UringDriver::probe_and_start_sharded(64, 3); + // SAFETY: `--test-threads=1` teardown. + unsafe { + std::env::remove_var("RUSTFS_URING_FAULT_SPAWN"); + } + + match result { + Ok(_) => panic!("driver started despite the forced spawn failure"), + Err(e) if e.is_expected_restriction() => { + // leg 1: io_uring was blocked before the probe reached the spawn seam. + eprintln!("SKIP spawn_failure_degrades_instead_of_panicking: restricted environment ({e:?})"); + } + Err(e) => { + // leg 2: the forced EAGAIN propagated as a ProbeFailure (degrade, not + // panic). EAGAIN is not a restriction errno, so the caller logs an + // unexpected-probe warning and still falls back to the std backend. + assert!( + !e.is_expected_restriction(), + "forced spawn EAGAIN must not look like an environment restriction: {e:?}" + ); + // A subsequent NORMAL probe must still succeed — the aborted start + // cleaned up any partially-started shard without corrupting state. + let driver = + UringDriver::probe_and_start_sharded(64, 2).expect("a normal probe after a forced spawn failure must still work"); + let _ = driver.shutdown(); + } + } +} From 996a627e4988a9d9a1d6f85c3e0084efc53955d9 Mon Sep 17 00:00:00 2001 From: houseme Date: Sat, 11 Jul 2026 14:35:01 +0800 Subject: [PATCH 05/15] fix(driver): time-bound the probe wait and probe only the first shard (rustfs/backlog#1165) Two probe defects that both bite startup: - drain_probe_cqe bounded only the EINTR retry count, not wall-clock time. submit_and_wait(1) parks in io_cqring_wait until a CQE or a signal, so a probe read that never completes (temp_dir on a hung/D-state or NFS device) blocked async disk init forever. Bound the wait by time: on EXT_ARG kernels (>= 5.11) pass a Timespec to the enter and treat ETIME as "re-check the deadline"; on older kernels fall back to the blocking wait, which the deadline still re-checks between returns. On expiry return an error so the caller's leak-over-UAF fallback degrades the disk to the std backend. - probe_and_start_sharded ran probe_real_read on EVERY shard, contradicting its own doc ("probing happens on the first shard") and doing shards-1 redundant O_TMPFILE round-trips per disk on every start and renew. Probe only the first shard; the rest still create a ring and verify NODROP. Co-Authored-By: heihutu --- src/driver.rs | 54 +++++++++++++++++++++++++++++++++++++++++---------- 1 file changed, 44 insertions(+), 10 deletions(-) diff --git a/src/driver.rs b/src/driver.rs index 569551d..b39dda1 100644 --- a/src/driver.rs +++ b/src/driver.rs @@ -604,9 +604,14 @@ impl UringDriver { /// joined before the error is returned. pub fn probe_and_start_sharded(entries: u32, shards: usize) -> Result { let mut started = Vec::with_capacity(shards.max(1)); - for _ in 0..shards.max(1) { + for i in 0..shards.max(1) { + // Probe only the first shard (rustfs/backlog#1165): the probe read + // exercises io_uring against the environment-global temp_dir, so one + // confirmation is representative. Shards 2..n only create a ring and + // verify NODROP — this avoids `shards - 1` extra O_TMPFILE + // create+write+read round-trips per disk on every start and renew. // `?` drops `started`, whose `Shard::drop` joins each running thread. - started.push(Self::start_shard(entries)?); + started.push(Self::start_shard(entries, i == 0)?); } Ok(Self { shards: started, @@ -623,7 +628,7 @@ impl UringDriver { &self.shards[self.rr.fetch_add(1, Ordering::Relaxed) % n] } - fn start_shard(entries: u32) -> Result { + fn start_shard(entries: u32, probe: bool) -> Result { let mut ring = IoUring::new(entries).map_err(ProbeFailure::Setup)?; // Require the NODROP feature (kernel >= 5.5). Without it, CQ overflow // silently drops CQEs, stranding pending entries forever and hanging @@ -632,7 +637,12 @@ impl UringDriver { if !ring.params().is_feature_nodrop() { return Err(ProbeFailure::Setup(io::Error::from_raw_os_error(libc::ENOSYS))); } - probe_real_read(&mut ring).map_err(ProbeFailure::ReadOp)?; + // Only the first shard runs the real-read probe (rustfs/backlog#1165); the + // rest still create a ring and check NODROP above, which is what makes + // io_uring usable, but skip the redundant temp_dir round-trip. + if probe { + probe_real_read(&mut ring).map_err(ProbeFailure::ReadOp)?; + } // Wake the driver loop on CQEs (kernel-signaled via a registered // eventfd) and on new messages (submit-signaled), replacing the 200 µs @@ -1053,13 +1063,38 @@ fn open_probe_file_exclusive(dir: &std::path::Path, pattern: &[u8]) -> io::Resul /// device from blocking forever; exhausting it returns an error that drives /// the caller's leak-over-UAF fallback. fn drain_probe_cqe(ring: &mut IoUring) -> io::Result { - const MAX_WAIT_ATTEMPTS: u32 = 4096; - for _ in 0..MAX_WAIT_ATTEMPTS { - match ring.submit_and_wait(1) { + // Bound the wait by WALL-CLOCK, not by an attempt count. `submit_and_wait(1)` + // parks in the kernel's io_cqring_wait until a CQE or a signal, so a single + // call can block forever when the probe read never completes — e.g. a + // temp_dir backed by a hung/D-state or NFS device. Since this runs on the + // caller's (async disk-init) thread, an unbounded block hangs startup. On + // kernels with EXT_ARG (>= 5.11) pass a timeout to the enter; on older + // kernels fall back to the blocking wait, whose only real risk is a hung + // temp_dir (rare) and which the deadline still re-checks between returns + // (rustfs/backlog#1165). On expiry, error out so the caller's leak-over-UAF + // fallback degrades the disk to the std backend instead of hanging. + const PROBE_TIMEOUT: Duration = Duration::from_secs(2); + let deadline = Instant::now() + PROBE_TIMEOUT; + let ext_arg = ring.params().is_feature_ext_arg(); + loop { + let remaining = deadline.saturating_duration_since(Instant::now()); + if remaining.is_zero() { + return Err(io::Error::other("probe: no CQE within the bounded wait")); + } + let waited = if ext_arg { + let ts = types::Timespec::new().sec(remaining.as_secs()).nsec(remaining.subsec_nanos()); + let args = types::SubmitArgs::new().timespec(&ts); + ring.submitter().submit_with_args(1, &args) + } else { + ring.submit_and_wait(1) + }; + match waited { Ok(_) => {} - // Signal interrupted the wait; the SQE is already in flight, so - // just wait again (do NOT re-push). + // Signal interrupted the wait; the SQE is already in flight, so wait + // again (do NOT re-push). The deadline still bounds the total time. Err(e) if e.raw_os_error() == Some(libc::EINTR) => {} + // EXT_ARG timeout elapsed with no CQE: loop to re-check the deadline. + Err(e) if e.raw_os_error() == Some(libc::ETIME) => {} Err(e) => return Err(e), } if let Some(cqe) = ring.completion().next() { @@ -1074,7 +1109,6 @@ fn drain_probe_cqe(ring: &mut IoUring) -> io::Result { return Ok(cqe.result()); } } - Err(io::Error::other("probe: no CQE after bounded wait")) } /// Owns everything the kernel can still be writing into: the ring, the From e4f04e84f01ace67b32930ddcdea4bc38784e0dc Mon Sep 17 00:00:00 2001 From: houseme Date: Sat, 11 Jul 2026 14:37:31 +0800 Subject: [PATCH 06/15] fix(driver): complete the C7 errno contract in reap and the offset guard (rustfs/backlog#1166) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Two gaps versus the driver's own documented three-class errno contract: - The reap loop turned every res < 0 into a final error, including EINTR/EAGAIN, which the contract classifies as "transient -> retry, never surface" — and surfacing them discarded a resubmit's already-read prefix. Retry EINTR/EAGAIN by resubmitting the remaining range, bounded per logical read by MAX_TRANSIENT_RETRIES (reset on progress) so a storm cannot spin the driver thread. Streams still finish; ECANCELED and every other errno terminate. - The submit offset guard only rejected offset > i64::MAX, so a request whose block-aligned end (kernel_offset + region_len) crossed i64::MAX still reached the kernel as a runtime EINVAL/EOVERFLOW — the errno class C7 says to pre-empt at submit. Reject it in submit, which also makes every resubmit offset provably <= i64::MAX. Co-Authored-By: heihutu --- src/driver.rs | 68 +++++++++++++++++++++++++++++++++++++++++++++------ 1 file changed, 61 insertions(+), 7 deletions(-) diff --git a/src/driver.rs b/src/driver.rs index b39dda1..4a170c4 100644 --- a/src/driver.rs +++ b/src/driver.rs @@ -89,6 +89,11 @@ const LOOP_HEARTBEAT: Duration = Duration::from_millis(50); /// 50 ms heartbeat this bounds the silent-retry window to a few seconds. const MAX_CONSECUTIVE_SUBMIT_ERRORS: u32 = 128; +/// How many times a single logical read retries a transient CQE errno +/// (EINTR/EAGAIN) without making progress before it surfaces the error, so a +/// pathological storm cannot spin the driver thread (rustfs/backlog#1166). +const MAX_TRANSIENT_RETRIES: u32 = 16; + /// Owned `eventfd(2)` used to wake the driver loop (backlog#1102): one is /// registered with the ring so the kernel signals it on every CQE, the other is /// signaled by `submit`/shutdown so a new message wakes the loop immediately — @@ -361,6 +366,10 @@ struct Pending { region_len: usize, /// `1` for buffered, the block size for `O_DIRECT`. align: usize, + /// Consecutive transient-errno (EINTR/EAGAIN) retries since the last byte of + /// progress, bounded by `MAX_TRANSIENT_RETRIES` so a storm cannot spin the + /// driver thread (rustfs/backlog#1166). Reset whenever a read makes progress. + transient_retries: u32, } /// Where a [`ReadHandle`] is in its lifecycle (rustfs/backlog#1102). @@ -773,15 +782,29 @@ impl UringDriver { }; } - // Reject a bad O_DIRECT alignment, and a request whose block-aligned - // superset range would exceed the kernel's single-read cap - // (rustfs/backlog#1102). `align == 1` (buffered) always passes. + // Reject a bad O_DIRECT alignment, a request whose block-aligned superset + // range would exceed the kernel's single-read cap, and one whose aligned + // END crosses i64::MAX — the kernel reads pos as a signed loff_t, so + // `kernel_offset + region_len > i64::MAX` fails at runtime with + // EINVAL/EOVERFLOW, exactly the errno class the C7 guard must pre-empt at + // submit (rustfs/backlog#1102, #1166). Pre-empting it here also makes + // every resubmit's `next_off < kernel_offset + region_len` provably + // <= i64::MAX. `align == 1` (buffered) always passes the alignment part. match aligned_geometry(offset, len, align) { - Some((_, _, region_len)) if region_len <= MAX_READ_LEN => {} + // CURRENT_POSITION (stream) reads use no positional offset — the + // kernel reads from the current file position — so the i64::MAX end + // check does not apply to them (their sentinel offset would overflow + // it). Exempt them exactly as the offset guard above does. + Some((kernel_offset, _, region_len)) + if region_len <= MAX_READ_LEN + && (offset == CURRENT_POSITION + || kernel_offset + .checked_add(region_len as u64) + .is_some_and(|end| end <= i64::MAX as u64)) => {} _ => { let _ = done.send(Err(io::Error::new( io::ErrorKind::InvalidInput, - "alignment must be a power of two and the block-aligned range must fit MAX_RW_COUNT", + "alignment must be a power of two, and the block-aligned range must fit MAX_RW_COUNT and end within i64::MAX", ))); return ReadHandle { id, @@ -1312,6 +1335,7 @@ fn drive( want: len, region_len, align, + transient_retries: 0, }, ); stats.submitted.fetch_add(1, Ordering::SeqCst); @@ -1442,13 +1466,43 @@ fn drive( let step = { let p = state.pending.get_mut(&ud).expect("checked above"); if res < 0 { - // Error (incl. ECANCELED) terminates the logical read. - ReapStep::Finish(Err(io::Error::from_raw_os_error(-res))) + let err = -res; + // C7 three-class contract (rustfs/backlog#1166): a transient + // errno (EINTR/EAGAIN) must be retried, not surfaced as the + // read's final result — surfacing it would also discard the + // already-read prefix of a resubmit. Bounded per logical read + // so a storm cannot spin the driver thread. Streams + // (CURRENT_POSITION) cannot resubmit positionally; ECANCELED + // and every other errno terminate the logical read. + let transient = err == libc::EINTR || err == libc::EAGAIN; + if transient + && p.offset != CURRENT_POSITION + && p.nread < p.region_len + && p.transient_retries < MAX_TRANSIENT_RETRIES + { + p.transient_retries += 1; + let remaining = p.region_len - p.nread; + // SAFETY: `pad + nread < pad + region_len <= buf.len()`, + // and the buffer lives in the pending table until the CQE. + let ptr = unsafe { p.buf.as_mut_ptr().add(p.pad + p.nread) }; + let next_off = p.offset + p.nread as u64; + let sqe = opcode::Read::new(types::Fd(p.file.as_raw_fd()), ptr, remaining as u32) + .offset(next_off) + .build() + .user_data(ud); + ReapStep::Resubmit(sqe) + } else { + // Error (incl. ECANCELED, or a transient errno past its + // retry budget) terminates the logical read. + ReapStep::Finish(Err(io::Error::from_raw_os_error(err))) + } } else if res == 0 { // Real EOF: deliver whatever of the logical range was read. ReapStep::Finish(Ok(deliver(p))) } else { p.nread += res as usize; + // Progress resets the transient-retry budget (rustfs/backlog#1166). + p.transient_retries = 0; // Only POSITIONED reads (read_at / read_at_direct, whole-range // pread contract) resubmit a short read. CURRENT_POSITION // reads (read_current on pipes/streams) follow read(2) From 58fea858b70c8449a2022c85f0a8cad354a19093 Mon Sep 17 00:00:00 2001 From: houseme Date: Sat, 11 Jul 2026 14:40:20 +0800 Subject: [PATCH 07/15] fix(driver): eventfd leak on bailout, NODROP overflow wording, cancel dedup (rustfs/backlog#1167) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Three diagnostic-consistency defects: - The bounded-drain bailout leaked the ring (still holding a registered cq_efd and in-flight ops) via mem::forget(state), then returned and dropped the cq_efd parameter, closing the fd out from under the still-mapped ring — violating start_shard's documented "cq_efd outlives the ring" invariant. Leak the eventfd alongside the ring on that path. - Overflow was reported as fatal "CQEs lost", but with NODROP (asserted at probe) overflow CQEs are buffered in the kernel overflow list and flushed on the next enter, not lost. Reword the log and StatsSnapshot doc to a backpressure warning. - AsyncCancel SQEs were not deduplicated: a drop-cancel plus a shutdown could queue two cancels per op, pushing completions past 2*entries and making transient overflow reachable. Track queued cancel ids in a HashSet (bounded by the pending table, entries removed on reap) so each op gets at most one cancel. Co-Authored-By: heihutu --- src/driver.rs | 60 ++++++++++++++++++++++++++++++++++++--------------- 1 file changed, 43 insertions(+), 17 deletions(-) diff --git a/src/driver.rs b/src/driver.rs index 4a170c4..80c894f 100644 --- a/src/driver.rs +++ b/src/driver.rs @@ -12,7 +12,7 @@ // See the License for the specific language governing permissions and // limitations under the License. -use std::collections::{HashMap, VecDeque}; +use std::collections::{HashMap, HashSet, VecDeque}; use std::fs::File; use std::io; use std::io::Write as _; @@ -275,9 +275,12 @@ pub struct StatsSnapshot { /// signal that makes drain-to-zero non-terminating (C4, /// rustfs/backlog#1055). pub cancel_already: u64, - /// Kernel CQ-ring overflow counter. MUST stay 0: a non-zero value means - /// CQEs were lost, so their pending entries are never reclaimed and drain - /// never completes. Treated as fatal (C5, rustfs/backlog#1056). + /// Kernel CQ-ring overflow counter. With NODROP (asserted at probe) overflow + /// CQEs are buffered in the kernel overflow list and flushed on the next + /// enter, NOT lost — so a non-zero value is a backpressure warning, not fatal + /// loss (C5, rustfs/backlog#1056, #1167). In-flight is capped at `entries` + /// and cancels are deduped, keeping completions <= 2*entries, so it should + /// stay 0 in practice. pub cq_overflow: u64, /// `ring.submit()` calls that returned a non-transient error. A rising count /// means `io_uring_enter` is persistently failing (e.g. a seccomp/LSM policy @@ -1219,6 +1222,12 @@ fn drive( // the persistent-submit-failure escape hatch (rustfs/backlog#1162). let mut consecutive_submit_errors: u32 = 0; let mut submit_error_logged = false; + // Ids with an AsyncCancel already queued, so a drop-cancel followed by a + // shutdown (or vice versa) does not enqueue a second cancel for the same op — + // keeping total completions <= 2*entries and CQ overflow unreachable + // (rustfs/backlog#1167). Ids are monotonic, so an entry is removed only when + // its pending op is reaped; the set stays bounded by the pending table. + let mut queued_cancels: HashSet = HashSet::new(); // Bounded-drain deadline (C4, rustfs/backlog#1055). Production always uses the // fixed DRAIN_TIMEOUT; a fault-injection build may shorten it via env so the @@ -1343,7 +1352,8 @@ fn drive( state.backlog.push_back(sqe); } Msg::Cancel { id } => { - if state.pending.contains_key(&id) { + // Dedup: at most one AsyncCancel per op (rustfs/backlog#1167). + if state.pending.contains_key(&id) && queued_cancels.insert(id) { state .backlog .push_back(opcode::AsyncCancel::new(id).build().user_data(id | CANCEL_BIT)); @@ -1352,9 +1362,11 @@ fn drive( Msg::Shutdown => { shutting_down = true; for id in state.pending.keys() { - state - .backlog - .push_back(opcode::AsyncCancel::new(*id).build().user_data(*id | CANCEL_BIT)); + if queued_cancels.insert(*id) { + state + .backlog + .push_back(opcode::AsyncCancel::new(*id).build().user_data(*id | CANCEL_BIT)); + } } } #[cfg(feature = "fault-injection")] @@ -1419,9 +1431,11 @@ fn drive( ); shutting_down = true; for id in state.pending.keys() { - state - .backlog - .push_back(opcode::AsyncCancel::new(*id).build().user_data(*id | CANCEL_BIT)); + if queued_cancels.insert(*id) { + state + .backlog + .push_back(opcode::AsyncCancel::new(*id).build().user_data(*id | CANCEL_BIT)); + } } } } @@ -1552,6 +1566,10 @@ fn drive( Err(_) => stats.orphan_reclaimed.fetch_add(1, Ordering::SeqCst), }; stats.in_flight.fetch_sub(1, Ordering::SeqCst); + // Drop any queued-cancel bookkeeping for this now-gone op so + // the dedup set stays bounded by the pending table + // (rustfs/backlog#1167). + queued_cancels.remove(&ud); // `p` (and with it `_permit`) is dropped here, at the CQE // and pending-table removal — never at future drop (C10, // rustfs/backlog#1060). No manual release to forget. @@ -1583,15 +1601,16 @@ fn drive( let _ = state.ring.submit(); } - // Monitor CQ overflow. With NODROP (asserted at probe) the crate's - // submit() auto-flushes the kernel overflow list, so this should stay - // 0; any non-zero value means CQEs were lost — pending entries would - // never be reclaimed. Record it as a fatal signal (C5, - // rustfs/backlog#1056). + // Monitor CQ overflow. With NODROP (asserted at probe) overflowed CQEs + // are BUFFERED in the kernel overflow list and flushed on the next enter, + // never lost — so a non-zero value is a backpressure warning, not fatal + // loss (rustfs/backlog#1056, #1167). In-flight reads are capped at + // `entries` and cancels are deduped (at most one per op), keeping total + // completions <= 2*entries, so this should stay 0 in practice. let overflow = state.ring.completion().overflow(); if overflow != 0 { stats.cq_overflow.store(overflow as u64, Ordering::SeqCst); - eprintln!("uring-spike driver: CQ overflow = {overflow}; CQEs lost — treat as fatal in P2"); + eprintln!("uring-spike driver: CQ overflow = {overflow}; CQEs buffered (NODROP), not lost — backpressure warning"); } // 4. Exit when drained: the kernel no longer references any buffer, so @@ -1633,6 +1652,13 @@ fn drive( // entries keep their permits, which is fine: nothing waits on // them any more. sem.close(); + // The leaked ring still has `cq_efd` registered via + // IORING_REGISTER_EVENTFD and in-flight ops that may post CQEs, so + // the eventfd must outlive it. Leak it alongside the ring instead + // of letting the returning `drive` drop (close) it out from under + // the still-mapped ring, honoring start_shard's documented "cq_efd + // outlives the ring" invariant on this exit too (rustfs/backlog#1167). + std::mem::forget(cq_efd); std::mem::forget(state); return; } From 7cb72a0491a82f439aa0773b9f47155556fe1a17 Mon Sep 17 00:00:00 2001 From: houseme Date: Sat, 11 Jul 2026 14:42:03 +0800 Subject: [PATCH 08/15] fix(driver): disambiguate O_DIRECT tail short reads via fstat, not a heuristic (rustfs/backlog#1168) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The short-read resubmit loop treated any non-block-multiple O_DIRECT read as EOF ("the kernel returns block multiples except at the file tail"). That holds for local block filesystems but not universally: NFS O_DIRECT partial-RPC completions, FUSE chunking, and signal-split direct I/O can legally return a non-block-multiple count mid-file. Assuming EOF there silently truncated the delivered range, violating the whole-range pread contract. Confirm EOF with the actual file length (fstat, cheap, driver-thread) instead of inferring it: deliver only when offset+nread is genuinely at/past EOF; for a mid-file non-multiple short read — which an O_DIRECT read cannot resubmit from a non-block-aligned offset — surface an error so the integration falls back to the std backend, preserving correctness rather than truncating. A failed fstat keeps the conservative EOF assumption. Buffered reads (align == 1) are unaffected. Co-Authored-By: heihutu --- src/driver.rs | 47 +++++++++++++++++++++++++++++++++++++---------- 1 file changed, 37 insertions(+), 10 deletions(-) diff --git a/src/driver.rs b/src/driver.rs index 80c894f..304a200 100644 --- a/src/driver.rs +++ b/src/driver.rs @@ -1173,6 +1173,15 @@ impl Drop for DriverState { } } +/// Best-effort file length via `fstat` on the driver thread, used to tell a +/// genuine O_DIRECT tail short read from a non-block-multiple short read that +/// happened mid-file on a stacked filesystem (rustfs/backlog#1168). `None` when +/// the stat fails, in which case the caller keeps the conservative EOF +/// assumption rather than risk a wrong error or an unbounded resubmit loop. +fn file_len(file: &File) -> Option { + file.metadata().ok().map(|m| m.len()) +} + /// Hand the caller exactly the logical range `[head, head + want)` of the read /// region, truncated to what was actually read (rustfs/backlog#1102). /// @@ -1525,18 +1534,36 @@ fn drive( // for stream data that may never come. let is_stream = p.offset == CURRENT_POSITION; let covered = p.nread >= p.head + p.want; - // An O_DIRECT resubmit must stay block-aligned. The kernel - // returns block multiples except at the file tail, so a - // non-multiple means we reached EOF: stop and deliver. - let unaligned_tail = p.align > 1 && !p.nread.is_multiple_of(p.align); - if is_stream || covered || unaligned_tail || p.nread >= p.region_len { + if is_stream || covered || p.nread >= p.region_len { ReapStep::Finish(Ok(deliver(p))) + } else if p.align > 1 && !p.nread.is_multiple_of(p.align) { + // O_DIRECT non-block-multiple short read below the covered + // range. The kernel returns block multiples EXCEPT at the + // file tail — but a stacked filesystem (NFS/FUSE, or a + // signal-split direct I/O) can legally return a non-multiple + // mid-file, and assuming EOF there would silently truncate + // the delivered range. Disambiguate with the actual file + // length instead of inferring it (rustfs/backlog#1168). + match file_len(&p.file) { + // Genuine tail: at or past EOF — deliver what we read. + Some(len) if p.offset + p.nread as u64 >= len => ReapStep::Finish(Ok(deliver(p))), + // Mid-file non-multiple: an O_DIRECT read cannot resubmit + // from a non-block-aligned offset, so surface an error + // rather than truncate. The integration falls back to + // the std backend for this read, preserving correctness. + Some(_) => ReapStep::Finish(Err(io::Error::other( + "io_uring O_DIRECT: non-block-aligned short read before EOF", + ))), + // fstat failed: keep the conservative EOF assumption + // rather than risk a wrong error or an infinite loop. + None => ReapStep::Finish(Ok(deliver(p))), + } } else { - // Positioned short read, not EOF: resubmit the remainder - // into the read region. The buffer stays owned by the - // driver and in_flight is unchanged — one logical op. - // For a direct read, `pad + nread` and `offset + nread` - // are both block-aligned, as is `remaining`. + // Positioned short read, not EOF, block-aligned: resubmit + // the remainder into the read region. The buffer stays + // owned by the driver and in_flight is unchanged — one + // logical op. For a direct read, `pad + nread` and + // `offset + nread` are both block-aligned, as is `remaining`. let remaining = p.region_len - p.nread; // SAFETY: `pad + nread < pad + region_len <= buf.len()`, // and the buffer lives in the pending table until the CQE. From 32fabcc6f63a560ab61e6133dd2469bae24fb303 Mon Sep 17 00:00:00 2001 From: houseme Date: Sat, 11 Jul 2026 14:46:05 +0800 Subject: [PATCH 09/15] perf(driver): cut idle churn and cap io-wq workers (rustfs/backlog#1169) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Each shard thread woke every 50 ms and, even on a fully idle turn, drained two eventfds and called ring.submit(), which on a non-SQPOLL ring always issues an io_uring_enter syscall with zero SQEs. Across disks x shards driver threads this is thousands of pointless timer wakeups and syscalls per second, forever. - Skip submit() when the SQ is empty: nothing to submit means nothing to enter. A non-empty SQ (freshly pushed, or residual after EBUSY) still submits. - Adaptive heartbeat: 50 ms only while shutting down or with in-flight ops to reap; otherwise wait up to IDLE_HEARTBEAT (1 s). New work still wakes the loop instantly via wake_efd and completions via cq_efd, so latency is unaffected. Also cap each ring's io-wq bounded worker pool via register_iowq_max_workers so a cold-read burst cannot materialize thousands of PF_IO_WORKER threads against TasksMax/RLIMIT_NPROC — the kernel default is min(sq_entries, 4*nCPU) per ring, and there is one ring per shard per disk. Best-effort: older kernels keep the default. Co-Authored-By: heihutu --- src/driver.rs | 120 +++++++++++++++++++++++++++++++++----------------- 1 file changed, 80 insertions(+), 40 deletions(-) diff --git a/src/driver.rs b/src/driver.rs index 304a200..428e612 100644 --- a/src/driver.rs +++ b/src/driver.rs @@ -82,6 +82,21 @@ fn aligned_geometry(offset: u64, len: usize, align: usize) -> Option<(u64, usize /// deadline is still checked and any queued cancel is picked up promptly. const LOOP_HEARTBEAT: Duration = Duration::from_millis(50); +/// Heartbeat used when the shard is fully idle — no in-flight ops and not +/// shutting down (rustfs/backlog#1169). New work still wakes the loop instantly +/// via `wake_efd` and completions via the registered `cq_efd`; this only bounds +/// the fallback wait, so a much longer value cuts idle timer/​syscall churn +/// across many per-disk shards without affecting latency. +const IDLE_HEARTBEAT: Duration = Duration::from_secs(1); + +/// Per-ring cap on io-wq BOUNDED workers (rustfs/backlog#1169). Cold buffered +/// and O_DIRECT reads punted to io-wq each spawn a bounded worker, and the +/// kernel default is min(sq_entries, 4*nCPU) PER ring — one ring per shard per +/// disk can otherwise materialize thousands of PF_IO_WORKER threads under a +/// cold-read burst. Best-effort (needs kernel >= 5.15); older kernels keep the +/// default. +const IOWQ_MAX_BOUNDED_WORKERS: u32 = 16; + /// Consecutive non-transient `ring.submit()` failures the driver tolerates /// before it stops retrying silently and shuts the shard down, so callers get a /// driver-gone error and fall back to the std backend instead of stalling @@ -665,6 +680,15 @@ impl UringDriver { ring.submitter() .register_eventfd(cq_efd.as_raw()) .map_err(ProbeFailure::Setup)?; + + // Cap the ring's io-wq bounded worker pool so a cold-read burst cannot + // materialize thousands of PF_IO_WORKER threads against the process's + // TasksMax/RLIMIT_NPROC (rustfs/backlog#1169). Best-effort: 0 leaves the + // unbounded pool unchanged, and a kernel without this op (< 5.15) keeps + // the default — neither is fatal to a working ring. + let mut iowq_max = [IOWQ_MAX_BOUNDED_WORKERS, 0u32]; + let _ = ring.submitter().register_iowq_max_workers(&mut iowq_max); + let wake_efd = Arc::new(EventFd::new().map_err(ProbeFailure::Setup)?); let thread_wake = Arc::clone(&wake_efd); @@ -1263,7 +1287,17 @@ fn drive( // both eventfds after waking keeps them from staying spuriously // readable; a missed edge is harmless because the CQ/mpsc are re-checked // unconditionally below. - wait_for_events(&cq_efd, &wake_efd, LOOP_HEARTBEAT); + // Adaptive heartbeat (rustfs/backlog#1169): poll at 50 ms only while + // there is in-flight work to reap or a drain deadline to honor; when the + // shard is fully idle, wait up to IDLE_HEARTBEAT. New work still wakes us + // immediately via wake_efd and completions via cq_efd, so the longer idle + // wait only cuts timer/syscall churn. + let heartbeat = if shutting_down || !state.pending.is_empty() { + LOOP_HEARTBEAT + } else { + IDLE_HEARTBEAT + }; + wait_for_events(&cq_efd, &wake_efd, heartbeat); cq_efd.drain(); wake_efd.drain(); @@ -1403,47 +1437,53 @@ fn drive( } } } - match state.ring.submit() { - Ok(_) => consecutive_submit_errors = 0, - Err(e) if e.raw_os_error() == Some(libc::EBUSY) => { - // CQ-overflow backpressure on pre-5.19 NODROP kernels: the - // kernel refuses new submissions until we reap. Keep the - // backlog and reap this turn instead of spinning (C5, - // rustfs/backlog#1056). Transient — not a submit failure. - consecutive_submit_errors = 0; - } - Err(e) if e.raw_os_error() == Some(libc::EINTR) => { - // A signal interrupted the enter and the SQEs were not consumed; - // retry on the next loop turn (transient). - consecutive_submit_errors = 0; - } - Err(e) => { - // Any other errno: the queued SQEs were not accepted, so their - // CQEs never arrive and their callers would hang. A brief run may - // be transient (EAGAIN under memory pressure); a persistent one - // (e.g. EPERM from a seccomp/LSM policy applied after startup) - // must NOT be retried forever in silence. Count it, log once, and - // after a bounded run of consecutive failures shut the shard down - // so the drain + bounded-drain bailout fail every pending caller - // with an error — they fall back to the std backend - // (rustfs/backlog#1162). - stats.submit_errors.fetch_add(1, Ordering::SeqCst); - consecutive_submit_errors += 1; - if !submit_error_logged { - submit_error_logged = true; - eprintln!("uring-spike driver: ring.submit() failed ({e}); retrying, will shut down if persistent"); + // Skip the io_uring_enter syscall on a fully idle turn (rustfs/backlog#1169): + // an empty SQ has nothing to submit, yet a non-SQPOLL submit() still enters + // the kernel unconditionally. A non-empty SQ — freshly pushed, or residual + // after EBUSY — still submits. + if !state.ring.submission().is_empty() { + match state.ring.submit() { + Ok(_) => consecutive_submit_errors = 0, + Err(e) if e.raw_os_error() == Some(libc::EBUSY) => { + // CQ-overflow backpressure on pre-5.19 NODROP kernels: the + // kernel refuses new submissions until we reap. Keep the + // backlog and reap this turn instead of spinning (C5, + // rustfs/backlog#1056). Transient — not a submit failure. + consecutive_submit_errors = 0; + } + Err(e) if e.raw_os_error() == Some(libc::EINTR) => { + // A signal interrupted the enter and the SQEs were not consumed; + // retry on the next loop turn (transient). + consecutive_submit_errors = 0; } - if !shutting_down && consecutive_submit_errors >= MAX_CONSECUTIVE_SUBMIT_ERRORS { - eprintln!( - "uring-spike driver: {consecutive_submit_errors} consecutive submit failures; \ + Err(e) => { + // Any other errno: the queued SQEs were not accepted, so their + // CQEs never arrive and their callers would hang. A brief run may + // be transient (EAGAIN under memory pressure); a persistent one + // (e.g. EPERM from a seccomp/LSM policy applied after startup) + // must NOT be retried forever in silence. Count it, log once, and + // after a bounded run of consecutive failures shut the shard down + // so the drain + bounded-drain bailout fail every pending caller + // with an error — they fall back to the std backend + // (rustfs/backlog#1162). + stats.submit_errors.fetch_add(1, Ordering::SeqCst); + consecutive_submit_errors += 1; + if !submit_error_logged { + submit_error_logged = true; + eprintln!("uring-spike driver: ring.submit() failed ({e}); retrying, will shut down if persistent"); + } + if !shutting_down && consecutive_submit_errors >= MAX_CONSECUTIVE_SUBMIT_ERRORS { + eprintln!( + "uring-spike driver: {consecutive_submit_errors} consecutive submit failures; \ shutting down so callers fall back to the std backend" - ); - shutting_down = true; - for id in state.pending.keys() { - if queued_cancels.insert(*id) { - state - .backlog - .push_back(opcode::AsyncCancel::new(*id).build().user_data(*id | CANCEL_BIT)); + ); + shutting_down = true; + for id in state.pending.keys() { + if queued_cancels.insert(*id) { + state + .backlog + .push_back(opcode::AsyncCancel::new(*id).build().user_data(*id | CANCEL_BIT)); + } } } } From 15b5518e712c24646a87a4f5204e3fa8089c99ad Mon Sep 17 00:00:00 2001 From: houseme Date: Sat, 11 Jul 2026 15:12:59 +0800 Subject: [PATCH 10/15] test(driver): deterministic sharded cancel-routing test (rustfs/backlog#1180) The sharded stress test cannot prove cancel routing: its dropped regular-file reads complete on their own, so a cancel routed to the wrong ring (answered -ENOENT) still conserves. Add a test where every op is a blocked-pipe read that can only finish by being canceled, with more ops than shards so every shard owns one. A mis-routed cancel would leave its read stuck and in_flight would never reach 0; the test asserts every orphan is reclaimed and that a cancel, not a timeout, did it. Co-Authored-By: heihutu --- tests/cancel.rs | 49 +++++++++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 49 insertions(+) diff --git a/tests/cancel.rs b/tests/cancel.rs index 5ff104b..3ccebd4 100644 --- a/tests/cancel.rs +++ b/tests/cancel.rs @@ -712,3 +712,52 @@ async fn sharded_driver_with_one_shard_matches_single_ring() { assert_eq!(snap.delivered + snap.orphan_reclaimed, snap.submitted, "{snap:?}"); let _ = std::fs::remove_file(path); } + +/// Deterministic sharded cancel routing (backlog#1180). A `ReadHandle` carries +/// the tx/wake of the shard that accepted its op, so a drop-cancel must reach +/// THAT shard's ring. The existing sharded stress test cannot prove this: its +/// dropped regular-file reads complete on their own, so a mis-routed cancel +/// (answered -ENOENT) still conserves. Here every op is a blocked-pipe read that +/// can ONLY finish by being canceled, and there is more than one op per shard, so +/// a cancel that landed on the wrong ring would leave its read stuck and +/// in_flight would never reach 0. +#[tokio::test(flavor = "multi_thread")] +async fn sharded_cancel_routes_to_the_owning_ring() { + const SHARDS: usize = 4; + // Comfortably more than SHARDS so round-robin gives every shard at least one. + const OPS: usize = 16; + let Some(driver) = sharded_driver_or_skip("sharded_cancel_routes_to_the_owning_ring", SHARDS) else { + return; + }; + let (pipe_read, pipe_write) = os_pipe(); + + let mut handles = Vec::new(); + for _ in 0..OPS { + handles.push(driver.read_current(Arc::clone(&pipe_read), 4096)); + } + assert!( + wait_until(Duration::from_secs(2), || driver.stats().in_flight == OPS as u64).await, + "not all reads reached in-flight: {:?}", + driver.stats() + ); + + // Drop every handle: each sends AsyncCancel to the shard that accepted it. + drop(handles); + + assert!( + wait_until(Duration::from_secs(3), || { + let s = driver.stats(); + s.in_flight == 0 && s.orphan_reclaimed == OPS as u64 + }) + .await, + "sharded cancels did not reclaim every orphan — a cancel routed to the wrong ring? {:?}", + driver.stats() + ); + // Blocked-pipe reads cannot complete on their own, so reclaiming every one + // proves the cancels — not a timeout — did it. + let snap = driver.stats(); + assert!(snap.cancel_succeeded > 0, "orphans not reclaimed by cancel: {snap:?}"); + + drop(pipe_write); + driver.shutdown(); +} From 5bd6a6b7bb9a50f3063a57dabcc716df8426525b Mon Sep 17 00:00:00 2001 From: houseme Date: Sat, 11 Jul 2026 15:13:09 +0800 Subject: [PATCH 11/15] docs(changelog): record the audit hardening pass and fix v0.1.0 links (rustfs/backlog#1181) The changelog claimed "No unreleased changes" although PR #11 (fault injection) landed after the 0.1.0 tag, and every reference link pointed at a nonexistent `v0.1.0` ref (the only tag is `0.1.0`, no `v`). Record the #11 seams and the backlog#1160 audit fixes under [Unreleased], add the [#11] link, and rewrite the compare/tag/blob links to the existing `0.1.0` ref. Co-Authored-By: heihutu --- CHANGELOG.md | 49 +++++++++++++++++++++++++++++++++++++++++++++---- 1 file changed, 45 insertions(+), 4 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index e7c65ed..df4bbac 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -23,7 +23,47 @@ aims to follow [Semantic Versioning](https://semver.org/spec/v2.0.0.html). ## [Unreleased] -No unreleased changes. +Hardening pass from the rustfs/backlog#1160 audit. All changes are on the read +path and preserve the cancel-safety ownership model. + +### Added + +- **Test-only fault-injection seams** for the cancel-safety escape hatches + (driver-thread panic, stuck bounded drain, forced probe-drain failure), gated + behind the `fault-injection` feature and never present in a default build. + ([#11], backlog#1103) +- `StatsSnapshot::submit_errors` — count of non-transient `ring.submit()` + failures, so a persistently failing `io_uring_enter` is observable. + (backlog#1162) + +### Fixed + +- **Bounded-drain bailout no longer hangs awaited handles.** It now fails every + stranded caller with a driver-gone error before leaking the pending table, and + leaks the ring-registered eventfd alongside the ring so the "cq_efd outlives + the ring" invariant holds on that exit too. (backlog#1161, #1167) +- **Persistent `ring.submit()` errors are classified** instead of retried + forever in silence: EINTR/EBUSY stay transient; any other errno is counted and, + after a bounded run, shuts the shard down so callers fall back. (backlog#1162) +- **Wakeup gaps closed:** short-read resubmits are flushed the same turn, and a + drop-cancel now signals the wake eventfd (after closing the receiver so the + reclaim is still counted as an orphan), removing up-to-50 ms stalls. + (backlog#1163) +- **Driver-thread spawn failure degrades** to a `ProbeFailure` instead of + panicking out of disk init. (backlog#1164) +- **The startup probe is time-bounded** (EXT_ARG timeout with a pre-5.11 + fallback) and runs on the first shard only. (backlog#1165) +- **EINTR/EAGAIN completions are retried** (bounded) rather than surfaced as the + read's final error, and the submit offset guard also rejects an aligned end + crossing `i64::MAX`. (backlog#1166) +- **CQ overflow is reported as a NODROP backpressure warning**, not fatal loss, + and AsyncCancel SQEs are deduplicated per op. (backlog#1167) +- **O_DIRECT tail short reads are disambiguated with `fstat`** instead of assuming + any non-block-multiple read is EOF, so a stacked filesystem cannot cause a + silent truncation. (backlog#1168) +- **Idle churn cut**: the loop skips `io_uring_enter` on an empty SQ and uses a + longer heartbeat when idle; each ring caps its io-wq bounded workers. + (backlog#1169) ## [0.1.0] - 2026-07-11 @@ -116,7 +156,7 @@ polished (`39018c0`). The cancel-safety invariants that survived that audit — the driver's pending table owning the buffer and fd from SQE submission until the CQE, drop-abandons-result-only, bounded shutdown drain, abort-before-free on a driver panic — are the ones every change above is measured -against. See the [design notes](https://github.com/rustfs/uring/blob/v0.1.0/docs/DESIGN.md). +against. See the [design notes](https://github.com/rustfs/uring/blob/0.1.0/docs/DESIGN.md). ## Decisions recorded, not implemented @@ -145,8 +185,9 @@ built. They are listed so nobody re-opens them without new evidence. [#4]: https://github.com/rustfs/uring/pull/4 [#5]: https://github.com/rustfs/uring/pull/5 [#6]: https://github.com/rustfs/uring/pull/6 +[#11]: https://github.com/rustfs/uring/pull/11 [rustfs/backlog#1051]: https://github.com/rustfs/backlog/issues/1051 [rustfs/backlog#1144]: https://github.com/rustfs/backlog/issues/1144 [rustfs/backlog#1159]: https://github.com/rustfs/backlog/issues/1159 -[Unreleased]: https://github.com/rustfs/uring/compare/v0.1.0...HEAD -[0.1.0]: https://github.com/rustfs/uring/releases/tag/v0.1.0 +[Unreleased]: https://github.com/rustfs/uring/compare/0.1.0...HEAD +[0.1.0]: https://github.com/rustfs/uring/releases/tag/0.1.0 From f5a29b657c25d5c08bd1399c1a67985a63abf448 Mon Sep 17 00:00:00 2001 From: houseme Date: Sat, 11 Jul 2026 15:48:08 +0800 Subject: [PATCH 12/15] refactor(driver): consolidate the submit/cancel/resubmit paths (rustfs/backlog#1160) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The audit-fix series accreted duplicated logic across the driver loop. Fold it into single, named helpers without changing behavior (both run-docker.sh legs still pass, real io_uring included): - Two submit blocks — the main push+submit with full errno classification and a bare `let _ = submit()` for the reap-flush — become one `submit_ring` helper. The reap-flush now shares the same classification instead of silently dropping submit errors. - Three AsyncCancel-enqueue sites (drop-cancel, shutdown, submit-error shutdown) become `queue_cancel`, keeping the per-op dedup in one place. - Two identical resubmit-SQE builds (short-read and transient-errno retry) become `Pending::resubmit_sqe`. - Two push-to-SQ loops become `flush_backlog`. The loop's submit → reap → conditional re-flush order is kept deliberately: a cache-hit read completes inline inside submit()'s io_uring_enter and is reaped the same turn, so reordering would cost the hot path an extra iteration; the second submit only fires when reap queued a resubmit. Co-Authored-By: heihutu --- src/driver.rs | 260 +++++++++++++++++++++++++------------------------- 1 file changed, 132 insertions(+), 128 deletions(-) diff --git a/src/driver.rs b/src/driver.rs index 428e612..5b0ffaa 100644 --- a/src/driver.rs +++ b/src/driver.rs @@ -390,6 +390,29 @@ struct Pending { transient_retries: u32, } +impl Pending { + /// Build the read SQE for the not-yet-read remainder + /// `[pad + nread, pad + region_len)` at file offset `offset + nread`. This is + /// the single place a read SQE is constructed: the initial submit calls it + /// with `nread == 0` (the whole region), and a short-read or transient-errno + /// resubmit calls it after `nread` has advanced (rustfs/backlog#1058/#1166). + /// For an `O_DIRECT` read `pad + nread`, `offset + nread`, and the remaining + /// length are all block-aligned. + fn read_sqe(&self, ud: u64) -> io_uring::squeue::Entry { + let remaining = self.region_len - self.nread; + // SAFETY: `pad + nread < pad + region_len <= buf.len()`, and the buffer + // lives in the pending table until the CQE, so the kernel may write here. + // The read region is exclusively owned by this entry (no live aliases), + // so deriving a `*mut` from the shared `as_ptr` is sound. + let ptr = unsafe { self.buf.as_ptr().add(self.pad + self.nread).cast_mut() }; + let next_off = self.offset + self.nread as u64; + opcode::Read::new(types::Fd(self.file.as_raw_fd()), ptr, remaining as u32) + .offset(next_off) + .build() + .user_data(ud) + } +} + /// Where a [`ReadHandle`] is in its lifecycle (rustfs/backlog#1102). enum HandleState { /// Nothing was ever handed to the driver (a rejected parameter, or the @@ -1236,6 +1259,79 @@ enum ReapStep { Resubmit(io_uring::squeue::Entry), } +/// Queue at most one `AsyncCancel` per op (rustfs/backlog#1167): a drop-cancel +/// followed by a shutdown, or the submit-error shutdown, must not enqueue a +/// second cancel for the same id. The set is bounded by the pending table +/// because ids are monotonic and an entry is removed when its op is reaped. +fn queue_cancel(backlog: &mut VecDeque, queued_cancels: &mut HashSet, id: u64) { + if queued_cancels.insert(id) { + backlog.push_back(opcode::AsyncCancel::new(id).build().user_data(id | CANCEL_BIT)); + } +} + +/// Push as much of the backlog into the SQ as fits, stopping when the ring is +/// full (the remainder retries next turn). +fn flush_backlog(ring: &mut IoUring, backlog: &mut VecDeque) { + let mut sq = ring.submission(); + while let Some(sqe) = backlog.pop_front() { + // SAFETY: read SQEs point into `pending`-owned buffers that live until + // their CQE; cancel SQEs carry no pointers. + if unsafe { sq.push(&sqe) }.is_err() { + backlog.push_front(sqe); + break; + } + } +} + +/// Flush the backlog into the SQ and submit it, with submit-error classification +/// (rustfs/backlog#1162). The single submit path for the whole loop: called once +/// after intake and once more after reap when resubmits were queued. Skips the +/// `io_uring_enter` syscall on an empty SQ (rustfs/backlog#1169). EINTR/EBUSY are +/// transient; any other errno is counted and, after a bounded run, transitions +/// the shard to shutdown so callers fall back to the std backend. +fn submit_ring( + state: &mut DriverState, + stats: &DriverStats, + consecutive_submit_errors: &mut u32, + submit_error_logged: &mut bool, + shutting_down: &mut bool, + queued_cancels: &mut HashSet, +) { + flush_backlog(&mut state.ring, &mut state.backlog); + if state.ring.submission().is_empty() { + return; + } + match state.ring.submit() { + Ok(_) => *consecutive_submit_errors = 0, + // CQ-overflow backpressure (EBUSY) and signal interruption (EINTR) are + // transient — retry next turn without counting them (C5, backlog#1056). + Err(e) if matches!(e.raw_os_error(), Some(libc::EBUSY) | Some(libc::EINTR)) => *consecutive_submit_errors = 0, + Err(e) => { + // The queued SQEs were not accepted, so their CQEs never arrive. A + // brief run may be transient (EAGAIN); a persistent one (e.g. EPERM + // from a seccomp/LSM policy applied after startup) must not be retried + // forever in silence. + stats.submit_errors.fetch_add(1, Ordering::SeqCst); + *consecutive_submit_errors += 1; + if !*submit_error_logged { + *submit_error_logged = true; + eprintln!("uring-spike driver: ring.submit() failed ({e}); retrying, will shut down if persistent"); + } + if !*shutting_down && *consecutive_submit_errors >= MAX_CONSECUTIVE_SUBMIT_ERRORS { + eprintln!( + "uring-spike driver: {} consecutive submit failures; shutting down so callers fall back to the std backend", + *consecutive_submit_errors + ); + *shutting_down = true; + let ids: Vec = state.pending.keys().copied().collect(); + for id in ids { + queue_cancel(&mut state.backlog, queued_cancels, id); + } + } + } + } +} + fn drive( ring: IoUring, rx: mpsc::Receiver, @@ -1359,18 +1455,12 @@ fn drive( continue; } - // The raw pointer is captured before `buf` moves into the - // table; moving the Vec never relocates its heap block, and - // the entry is only removed at the CQE. `region_len as u32` - // is lossless: `submit` rejected anything > MAX_READ_LEN. - // - // SAFETY: `pad <= align - 1` and `pad + region_len <= - // buf.len()`, so the pointer stays inside the allocation. - let region_ptr = unsafe { buf.as_mut_ptr().add(pad) }; - let sqe = opcode::Read::new(types::Fd(file.as_raw_fd()), region_ptr, region_len as u32) - .offset(kernel_offset) - .build() - .user_data(id); + // Move the buffer into the pending table (which owns it until + // the CQE), THEN build the SQE from the entry: the initial read + // is `read_sqe` with `nread == 0`, so the read-region pointer + // math and the `Read` builder live in exactly one place. + // Moving the Vec never relocates its heap block, so the pointer + // the SQE captures stays valid. state.pending.insert( id, Pending { @@ -1390,26 +1480,21 @@ fn drive( transient_retries: 0, }, ); + let sqe = state.pending.get(&id).expect("just inserted").read_sqe(id); stats.submitted.fetch_add(1, Ordering::SeqCst); stats.in_flight.fetch_add(1, Ordering::SeqCst); state.backlog.push_back(sqe); } Msg::Cancel { id } => { - // Dedup: at most one AsyncCancel per op (rustfs/backlog#1167). - if state.pending.contains_key(&id) && queued_cancels.insert(id) { - state - .backlog - .push_back(opcode::AsyncCancel::new(id).build().user_data(id | CANCEL_BIT)); + if state.pending.contains_key(&id) { + queue_cancel(&mut state.backlog, &mut queued_cancels, id); } } Msg::Shutdown => { shutting_down = true; - for id in state.pending.keys() { - if queued_cancels.insert(*id) { - state - .backlog - .push_back(opcode::AsyncCancel::new(*id).build().user_data(*id | CANCEL_BIT)); - } + let ids: Vec = state.pending.keys().copied().collect(); + for id in ids { + queue_cancel(&mut state.backlog, &mut queued_cancels, id); } } #[cfg(feature = "fault-injection")] @@ -1425,70 +1510,16 @@ fn drive( } } - // 2. Push backlog into the SQ (stop when full; retry next turn). - { - let mut sq = state.ring.submission(); - while let Some(sqe) = state.backlog.pop_front() { - // SAFETY: read SQEs point into `pending`-owned buffers that - // live until their CQE; cancel SQEs carry no pointers. - if unsafe { sq.push(&sqe) }.is_err() { - state.backlog.push_front(sqe); - break; - } - } - } - // Skip the io_uring_enter syscall on a fully idle turn (rustfs/backlog#1169): - // an empty SQ has nothing to submit, yet a non-SQPOLL submit() still enters - // the kernel unconditionally. A non-empty SQ — freshly pushed, or residual - // after EBUSY — still submits. - if !state.ring.submission().is_empty() { - match state.ring.submit() { - Ok(_) => consecutive_submit_errors = 0, - Err(e) if e.raw_os_error() == Some(libc::EBUSY) => { - // CQ-overflow backpressure on pre-5.19 NODROP kernels: the - // kernel refuses new submissions until we reap. Keep the - // backlog and reap this turn instead of spinning (C5, - // rustfs/backlog#1056). Transient — not a submit failure. - consecutive_submit_errors = 0; - } - Err(e) if e.raw_os_error() == Some(libc::EINTR) => { - // A signal interrupted the enter and the SQEs were not consumed; - // retry on the next loop turn (transient). - consecutive_submit_errors = 0; - } - Err(e) => { - // Any other errno: the queued SQEs were not accepted, so their - // CQEs never arrive and their callers would hang. A brief run may - // be transient (EAGAIN under memory pressure); a persistent one - // (e.g. EPERM from a seccomp/LSM policy applied after startup) - // must NOT be retried forever in silence. Count it, log once, and - // after a bounded run of consecutive failures shut the shard down - // so the drain + bounded-drain bailout fail every pending caller - // with an error — they fall back to the std backend - // (rustfs/backlog#1162). - stats.submit_errors.fetch_add(1, Ordering::SeqCst); - consecutive_submit_errors += 1; - if !submit_error_logged { - submit_error_logged = true; - eprintln!("uring-spike driver: ring.submit() failed ({e}); retrying, will shut down if persistent"); - } - if !shutting_down && consecutive_submit_errors >= MAX_CONSECUTIVE_SUBMIT_ERRORS { - eprintln!( - "uring-spike driver: {consecutive_submit_errors} consecutive submit failures; \ - shutting down so callers fall back to the std backend" - ); - shutting_down = true; - for id in state.pending.keys() { - if queued_cancels.insert(*id) { - state - .backlog - .push_back(opcode::AsyncCancel::new(*id).build().user_data(*id | CANCEL_BIT)); - } - } - } - } - } - } + // 2. Flush the backlog into the SQ and submit it (the single submit path; + // see `submit_ring`). + submit_ring( + &mut state, + &stats, + &mut consecutive_submit_errors, + &mut submit_error_logged, + &mut shutting_down, + &mut queued_cancels, + ); // 3. Reap. A Pending entry (and thus its buffer) is dropped ONLY when // the logical read finishes; a short read is resubmitted for the @@ -1544,16 +1575,7 @@ fn drive( && p.transient_retries < MAX_TRANSIENT_RETRIES { p.transient_retries += 1; - let remaining = p.region_len - p.nread; - // SAFETY: `pad + nread < pad + region_len <= buf.len()`, - // and the buffer lives in the pending table until the CQE. - let ptr = unsafe { p.buf.as_mut_ptr().add(p.pad + p.nread) }; - let next_off = p.offset + p.nread as u64; - let sqe = opcode::Read::new(types::Fd(p.file.as_raw_fd()), ptr, remaining as u32) - .offset(next_off) - .build() - .user_data(ud); - ReapStep::Resubmit(sqe) + ReapStep::Resubmit(p.read_sqe(ud)) } else { // Error (incl. ECANCELED, or a transient errno past its // retry budget) terminates the logical read. @@ -1602,18 +1624,8 @@ fn drive( // Positioned short read, not EOF, block-aligned: resubmit // the remainder into the read region. The buffer stays // owned by the driver and in_flight is unchanged — one - // logical op. For a direct read, `pad + nread` and - // `offset + nread` are both block-aligned, as is `remaining`. - let remaining = p.region_len - p.nread; - // SAFETY: `pad + nread < pad + region_len <= buf.len()`, - // and the buffer lives in the pending table until the CQE. - let ptr = unsafe { p.buf.as_mut_ptr().add(p.pad + p.nread) }; - let next_off = p.offset + p.nread as u64; - let sqe = opcode::Read::new(types::Fd(p.file.as_raw_fd()), ptr, remaining as u32) - .offset(next_off) - .build() - .user_data(ud); - ReapStep::Resubmit(sqe) + // logical op. + ReapStep::Resubmit(p.read_sqe(ud)) } } }; @@ -1646,26 +1658,18 @@ fn drive( } // A short-read resubmit queued during reap must reach the kernel in THIS - // turn, not wait out the next heartbeat: reap runs after the earlier - // push+submit, so without this flush the remainder sits idle for up to - // LOOP_HEARTBEAT (rustfs/backlog#1163). Only the resubmit/backlog-residue - // case makes this non-empty, so a fully idle turn skips it. + // turn, not wait out the next heartbeat (rustfs/backlog#1163). Reap runs + // after the submit above, so re-run the single submit path when reap left + // work in the backlog; an idle turn leaves it empty and skips the call. if !state.backlog.is_empty() { - { - let mut sq = state.ring.submission(); - while let Some(sqe) = state.backlog.pop_front() { - // SAFETY: read SQEs point into `pending`-owned buffers that - // live until their CQE; cancel SQEs carry no pointers. - if unsafe { sq.push(&sqe) }.is_err() { - state.backlog.push_front(sqe); - break; - } - } - } - // A persistent submit failure is already counted and acted on by the - // main submit above (rustfs/backlog#1162); this same-turn flush just - // retries opportunistically, so a transient error here is fine. - let _ = state.ring.submit(); + submit_ring( + &mut state, + &stats, + &mut consecutive_submit_errors, + &mut submit_error_logged, + &mut shutting_down, + &mut queued_cancels, + ); } // Monitor CQ overflow. With NODROP (asserted at probe) overflowed CQEs From 467657fe125867117c438ee42839a41b1b5f2a47 Mon Sep 17 00:00:00 2001 From: houseme Date: Sat, 11 Jul 2026 15:50:09 +0800 Subject: [PATCH 13/15] chore: release 0.2.0 Bump to 0.2.0 for the backlog#1160 cancel-safety / degradation hardening pass and the driver-loop refactor. Promote the CHANGELOG [Unreleased] section to [0.2.0], and update the version and design-note references in Cargo.toml, the README, and the crate docs. Co-Authored-By: heihutu --- CHANGELOG.md | 24 ++++++++++++++++++++++-- Cargo.lock | 2 +- Cargo.toml | 2 +- README.md | 6 +++--- src/lib.rs | 4 ++-- 5 files changed, 29 insertions(+), 9 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index df4bbac..5c3b2d0 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -23,8 +23,13 @@ aims to follow [Semantic Versioning](https://semver.org/spec/v2.0.0.html). ## [Unreleased] +No unreleased changes. + +## [0.2.0] - 2026-07-11 + Hardening pass from the rustfs/backlog#1160 audit. All changes are on the read -path and preserve the cancel-safety ownership model. +path and preserve the cancel-safety ownership model; both `run-docker.sh` legs +(seccomp-blocked degradation and real io_uring) pass. ### Added @@ -65,6 +70,20 @@ path and preserve the cancel-safety ownership model. longer heartbeat when idle; each ring caps its io-wq bounded workers. (backlog#1169) +### Changed + +- **Internal driver-loop refactor** folding the duplicated submit, + cancel-enqueue, and resubmit-SQE paths into single helpers (`submit_ring`, + `queue_cancel`, `Pending::resubmit_sqe`, `flush_backlog`). No behavior change; + the perf-sensitive submit → reap → conditional re-flush order is kept so a + cache-hit read is still reaped the same turn it completes inline. + (backlog#1160) + +### Tests + +- Deterministic sharded cancel-routing test proving a drop-cancel reaches the + ring that accepted the op. (backlog#1180) + ## [0.1.0] - 2026-07-11 ### Added @@ -189,5 +208,6 @@ built. They are listed so nobody re-opens them without new evidence. [rustfs/backlog#1051]: https://github.com/rustfs/backlog/issues/1051 [rustfs/backlog#1144]: https://github.com/rustfs/backlog/issues/1144 [rustfs/backlog#1159]: https://github.com/rustfs/backlog/issues/1159 -[Unreleased]: https://github.com/rustfs/uring/compare/0.1.0...HEAD +[Unreleased]: https://github.com/rustfs/uring/compare/0.2.0...HEAD +[0.2.0]: https://github.com/rustfs/uring/compare/0.1.0...0.2.0 [0.1.0]: https://github.com/rustfs/uring/releases/tag/0.1.0 diff --git a/Cargo.lock b/Cargo.lock index 3dc98f5..90402c2 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -57,7 +57,7 @@ dependencies = [ [[package]] name = "rustfs-uring" -version = "0.1.0" +version = "0.2.0" dependencies = [ "io-uring", "libc", diff --git a/Cargo.toml b/Cargo.toml index 0fd4dc3..07bcda7 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -14,7 +14,7 @@ [package] name = "rustfs-uring" -version = "0.1.0" +version = "0.2.0" edition = "2024" rust-version = "1.96" license = "Apache-2.0" diff --git a/README.md b/README.md index 00d8a04..7ee53ed 100644 --- a/README.md +++ b/README.md @@ -15,11 +15,11 @@ wired into the storage layer. > **Status:** read path only, Linux only. The read path is wired into `rustfs/rustfs` behind a runtime probe and is > **off by default** (`RUSTFS_IO_URING_READ_ENABLE`). See [`CHANGELOG.md`](CHANGELOG.md) for what has landed and the -> [design notes](https://github.com/rustfs/uring/blob/v0.1.0/docs/DESIGN.md) for the invariants. +> [design notes](https://github.com/rustfs/uring/blob/0.2.0/docs/DESIGN.md) for the invariants. ```toml [target.'cfg(target_os = "linux")'.dependencies] -rustfs-uring = "0.1.0" +rustfs-uring = "0.2.0" ``` ## The ownership model it enforces @@ -40,7 +40,7 @@ proves and enforces the invariants any production io_uring integration must foll Each invariant holds **per shard** (see below), because a shard is an independent instance of the same driver. The full invariant list, the corrected fd-reuse mechanism, and the design constraints are in -the [design notes](https://github.com/rustfs/uring/blob/v0.1.0/docs/DESIGN.md). +the [design notes](https://github.com/rustfs/uring/blob/0.2.0/docs/DESIGN.md). ## Usage diff --git a/src/lib.rs b/src/lib.rs index aada8cf..e66c2d2 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -33,8 +33,8 @@ //! reads, `O_DIRECT` reads with internal alignment, sharded rings, async //! backpressure, eventfd-driven reaping, graceful restricted-environment //! detection, and bounded shutdown drain. The write path is intentionally out -//! of scope for `0.1.0`; see the -//! [design notes](https://github.com/rustfs/uring/blob/v0.1.0/docs/DESIGN.md) +//! of scope; see the +//! [design notes](https://github.com/rustfs/uring/blob/0.2.0/docs/DESIGN.md) //! for the invariant details. #[cfg(target_os = "linux")] From 8db27ef3bf61d78834f01f307bdc6c9e45fada6f Mon Sep 17 00:00:00 2001 From: houseme Date: Sat, 11 Jul 2026 17:39:01 +0800 Subject: [PATCH 14/15] docs: rewrite README from the public API and remove docs/DESIGN.md MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Refactor the README to match what the code actually exposes: all three read entry points (read_at / read_at_direct / read_current), ReadHandle and StatsSnapshot, and the degradation contract added in the backlog#1160 hardening (classified probe/submit failures, transient-errno retry, fstat-disambiguated EOF). Correct the acceptance-test counts (16 cancel-safety + 5 fault-injection). Remove docs/DESIGN.md and its now-dangling links from the README and crate docs so there is a single source of truth — the per-invariant rationale lives in the module/function docs and the README. The historical CHANGELOG link into the 0.1.0 tag stays valid. Drop the obsolete `exclude = ["docs/**"]` from Cargo.toml. Co-Authored-By: heihutu --- CHANGELOG.md | 8 +++- Cargo.toml | 1 - README.md | 76 +++++++++++++++++++++------------ docs/DESIGN.md | 114 ------------------------------------------------- src/lib.rs | 11 +++-- 5 files changed, 61 insertions(+), 149 deletions(-) delete mode 100644 docs/DESIGN.md diff --git a/CHANGELOG.md b/CHANGELOG.md index 5c3b2d0..ea6bd9a 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -23,7 +23,13 @@ aims to follow [Semantic Versioning](https://semver.org/spec/v2.0.0.html). ## [Unreleased] -No unreleased changes. +### Docs + +- Rewrote the README from the current public API — all three read entry points + (`read_at`, `read_at_direct`, `read_current`) and the degradation contract — + and removed the `docs/DESIGN.md` design notes. The per-invariant rationale now + lives in the module and function docs and in the README, so there is a single + source of truth. ## [0.2.0] - 2026-07-11 diff --git a/Cargo.toml b/Cargo.toml index 07bcda7..e93422a 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -25,7 +25,6 @@ readme = "README.md" homepage = "https://rustfs.com" keywords = ["io_uring", "async", "storage", "rustfs", "linux"] categories = ["asynchronous", "filesystem"] -exclude = ["docs/**"] [features] # Test-only fault-injection seams (rustfs/backlog#1103). OFF by default and in diff --git a/README.md b/README.md index 7ee53ed..1bdf85b 100644 --- a/README.md +++ b/README.md @@ -7,15 +7,14 @@ Cancel-safe async `io_uring` read backend for [RustFS](https://github.com/rustfs/rustfs) storage. -This crate is the io_uring integration that RustFS's read path is built on. It started as the Spike 0 cancel-safety -prototype ([rustfs/backlog#894](https://github.com/rustfs/backlog/issues/894)) and was hardened per -the [#1048/#1051 audit](https://github.com/rustfs/backlog/issues/1051). It lives in its own repository so it can be +This crate is the io_uring integration that RustFS's read path is built on. It lives in its own repository so it can be verified in isolation — with a real io_uring CI leg that the main `rustfs/rustfs` workspace cannot run — before being -wired into the storage layer. +wired into the storage layer behind a runtime probe. -> **Status:** read path only, Linux only. The read path is wired into `rustfs/rustfs` behind a runtime probe and is -> **off by default** (`RUSTFS_IO_URING_READ_ENABLE`). See [`CHANGELOG.md`](CHANGELOG.md) for what has landed and the -> [design notes](https://github.com/rustfs/uring/blob/0.2.0/docs/DESIGN.md) for the invariants. +> **Status:** read path only, Linux only. On any other target the crate compiles to an empty stub. The read path is +> wired into `rustfs/rustfs` behind a runtime probe and is **off by default** (`RUSTFS_IO_URING_READ_ENABLE`). What has +> landed is in [`CHANGELOG.md`](CHANGELOG.md); the per-invariant rationale lives in the module and function docs on +> [docs.rs](https://docs.rs/rustfs-uring/) and inline in `src/`. ```toml [target.'cfg(target_os = "linux")'.dependencies] @@ -26,21 +25,31 @@ rustfs-uring = "0.2.0" When a caller drops the future of an in-flight read (EC quorum reached, timeout, disconnect), the kernel may still write into the read buffer at any point until the CQE. Freeing the buffer at future-drop is a use-after-free. This crate -proves and enforces the invariants any production io_uring integration must follow: +enforces the invariants any production io_uring integration must follow: - **The buffer and the file handle are owned by the driver's pending (orphan) table** from SQE submission until the CQE — never by the caller's future. -- **Dropping the future abandons only the result**; reclamation always happens at the CQE (optionally accelerated by - `IORING_OP_ASYNC_CANCEL`). -- **Shutdown drains in-flight ops to zero** (with a bounded escape hatch for hung disks) before the ring is unmapped. -- A driver-thread panic **aborts before freeing in-flight buffers** (leak over UAF); backpressure permits are released - at the CQE, not at future drop; short reads on positioned reads are resubmitted to satisfy the whole-range contract; +- **Dropping the future abandons only the result**; reclamation always happens at the CQE, optionally accelerated by an + `IORING_OP_ASYNC_CANCEL` sent on drop. +- **Shutdown drains in-flight ops to zero** before the ring is unmapped, with a bounded escape hatch for a hung disk: + on timeout it fails the stranded callers with an error and *leaks* the ring and its buffers rather than free memory + the kernel may still touch (leak over UAF). +- A driver-thread panic **aborts the process before freeing in-flight buffers**; backpressure permits are released at + the CQE, not at future drop; a short read on a positioned read is resubmitted to satisfy the whole-range contract; the probe file is opened via `O_TMPFILE`. -Each invariant holds **per shard** (see below), because a shard is an independent instance of the same driver. +It also has a **degradation contract** — the reason it can ship default-on-probe safely: -The full invariant list, the corrected fd-reuse mechanism, and the design constraints are in -the [design notes](https://github.com/rustfs/uring/blob/0.2.0/docs/DESIGN.md). +- The startup probe runs a real `IORING_OP_READ` under a wall-clock timeout. A restricted environment + (seccomp / gVisor / old kernel) fails with a `ProbeFailure` whose `is_expected_restriction()` tells the caller to + degrade to the std backend quietly; an unexpected failure is surfaced, not hidden. +- A persistently failing `io_uring_enter` is classified (not retried forever in silence): the shard shuts down so + callers fall back, and the failure count is visible in `StatsSnapshot`. +- Transient CQE errnos (`EINTR`/`EAGAIN`) are retried; the whole-range and EOF contracts are honoured even on stacked + filesystems by disambiguating a short read with `fstat` rather than assuming EOF. + +Each invariant holds **per shard** (see below), because a shard is an independent instance of the same driver. Every +one is pinned by an acceptance test (see [Testing](#testing)). ## Usage @@ -50,9 +59,8 @@ use std::sync::Arc; use rustfs_uring::UringDriver; # async fn demo() -> std::io::Result<()> { - // Probe a real IORING_OP_READ before accepting work. A restricted environment - // (seccomp/gVisor/old kernel) returns a ProbeFailure whose - // `is_expected_restriction()` tells you to degrade to the std backend quietly. + // Probe a real IORING_OP_READ before accepting work. On a restricted host the + // ProbeFailure's `is_expected_restriction()` tells you to degrade quietly. let driver = UringDriver::probe_and_start(64).expect("io_uring available"); let file = Arc::new(File::open("/data/object")?); @@ -70,6 +78,17 @@ use rustfs_uring::UringDriver; } ``` +Three read entry points, all returning an awaitable `ReadHandle`: + +- `read_at(file, offset, len)` — positioned (pread) read, whole-range: short reads are resubmitted until the range is + satisfied or a real EOF. +- `read_at_direct(file, offset, len, align)` — the same, for an fd opened with `O_DIRECT` (see below). +- `read_current(file, len)` — reads from the fd's current position with `read(2)` semantics: a short read is a valid + final result and is *not* resubmitted. Use it for pipes and other non-seekable fds. + +`ReadHandle::without_cancel_on_drop()` opts a handle out of the drop-time `ASYNC_CANCEL` when you know the op will +complete on its own; and `driver.stats()` returns a `StatsSnapshot` at any time. + ### Sharded rings A buffered read that hits the page cache completes *inline* inside `io_uring_enter`, so the thread driving a ring @@ -135,15 +154,19 @@ cargo test -- --nocapture --test-threads=1 # Two legs in Docker (also runs on macOS via Docker Desktop / OrbStack): # leg 1 — io_uring blocked by an explicit seccomp profile → the suite MUST -# degrade to a graceful skip (reproduces the #4313 restricted env); +# degrade to a graceful skip (reproduces a restricted environment); # leg 2 — seccomp=unconfined → real io_uring, and NO test may skip. ./run-docker.sh ``` The harness fails on either a non-degrading leg 1 or a vacuous-pass leg 2, so a skipped suite can never masquerade as -real coverage. The 15 acceptance tests are the cancel-safety contract: buffer conservation under a mixed -drop/keep stress across shards, an orphaned op reclaimed only at its CQE, bounded shutdown drain, `O_DIRECT` returning -exact unaligned ranges, and backpressure deferring rather than blocking a runtime worker. +real coverage. The cancel-safety contract is pinned by 16 acceptance tests in `tests/cancel.rs` — buffer conservation +under a mixed drop/keep stress across shards, an orphaned op reclaimed only at its CQE, sharded cancel routed to the +ring that owns the op, bounded shutdown drain, `O_DIRECT` returning exact unaligned ranges, and backpressure deferring +rather than blocking a runtime worker. Five deterministic fault-injection tests in `tests/fault_injection.rs` (behind +the test-only `fault-injection` feature, never compiled into a release build) drive the escape hatches: the +panic-abort barrier, the bounded-drain leak-and-error path, a forced probe-drain failure, and a driver-thread spawn +failure that must degrade rather than panic. ## Benchmarks @@ -168,10 +191,9 @@ alone cannot tell a correct strategy from one reading the wrong offsets. - **`SQPOLL`.** Eliminates `io_uring_enter` under sustained load, at the cost of a kernel polling thread per ring — which multiplies by shards and by disks. Only for high-end deployments. -Closed by measurement, not built — see [`CHANGELOG.md`](CHANGELOG.md#decisions-recorded-not-implemented): streaming -reads through io_uring (NO-GO), `AsyncFd` reaping without a driver thread (would break the public API), a process-wide -singleton ring (conflicts with per-disk isolation), and registered buffers (conflicts with the `Vec` ownership -model, and the bottleneck is elsewhere). +Closed by measurement, not built (see the CHANGELOG): streaming reads through io_uring (NO-GO), `AsyncFd` reaping +without a driver thread (would break the public API), a process-wide singleton ring (conflicts with per-disk +isolation), and registered buffers (conflicts with the `Vec` ownership model, and the bottleneck is elsewhere). ## License diff --git a/docs/DESIGN.md b/docs/DESIGN.md deleted file mode 100644 index dd26588..0000000 --- a/docs/DESIGN.md +++ /dev/null @@ -1,114 +0,0 @@ -# rustfs-uring 设计:io_uring 取消安全读后端(backlog#894 / #1048 / #1051) - -> 本文档源自 Spike 0 取消安全原型(原 `SPIKE.md`),经 rustfs/backlog#1051 审计整改后作为 `rustfs-uring` 库的设计与不变量说明保留。逐 issue 修复的历史见本仓库 git log。 - -## 这是什么 - -rustfs/backlog#897 路线图中 P2(io_uring 读后端)被 P1.5 基准判 NO-GO 而 defer。本 spike 是 #894 明确要求先行的**取消安全原型**——P2 中风险最高、最容易随时间流失的知识,按"只实现原型、不进主干、不启用"的方案 B 存档。 - -> **现状更新(2026-07):** P2 主体已在此库实现并接入 `rustfs/rustfs`,但**默认灰度关**(`RUSTFS_IO_URING_READ_ENABLE`)。端到端 A/B(rustfs/backlog#1159)显示 io_uring 对 S3 GET 大致中性(−7%~+4%),瓶颈在用户态拷贝而非磁盘读,因此 #1048 转为 **[Watch] 看护 issue**——真实直连 NVMe 证据满足前不默认启用。下方"对 P2 主体实现的遗留项"一节记录了每项的落地/决策现状。本文其余部分(所有权模型、不变量、测试矩阵)是这些实现共同遵守的契约,持续有效。 - -**本 crate 是独立 workspace**(Cargo.toml 内含空 `[workspace]` 表),io-uring 依赖不进入 rustfs 主 Cargo.lock、不参与主工程构建与 CI。这与守卫脚本 `scripts/check_no_tokio_io_uring.sh` 的约束一致:禁的是 tokio 的 io-uring runtime feature,应用层显式 io-uring 集成必须走运行时探测的独立后端(即本原型验证的模型)。 - -## 要证明的问题 - -EC 读会 drop 在途的分片读 future;若该 future 已向内核提交了 read SQE,内核在 CQE 之前始终可能向目标 buffer 写入。**future 的 drop 不能回收 buffer,否则是 use-after-free。** - -> **该场景在生产 GET 上被真实行使(2026-07 核对 main):** 主触发点是**读者建立阶段**——`crates/ecstore/src/set_disk/core/io_primitives.rs` 的 `create_bitrot_readers_until_quorum_all_shards`(`FuturesUnordered` 于 :1362 建立、setup quorum break 于 :1403、多余任务于 :1428 `drop(reader_tasks)`)。数据分片在此阶段经 `read_file_mmap_copy → UringBackend::pread_bytes → driver.read_at(...).await` **急切**读入,因此一个仍停在该 `.await` 的任务被 drop 时,其 `ReadHandle` 在 `Submitted` 态被 drop、触发 `ASYNC_CANCEL`(`ReadHandle::drop`,`src/driver.rs`)。这在**每个响应盘多于 setup quorum 的 GET(常态)**都发生,与后续是否重建无关。decode 阶段(`ParallelReader`,`crates/ecstore/src/erasure/coding/decode.rs`)的 `FuturesUnordered` 也在 quorum 处 drop 落败者,但对 io_uring 只在**非 lockstep 路径的延迟 parity reader 边打开边被 drop**时命中(数据分片此时已是内存 `Bytes`,lockstep 路径则全量 drain、从不中途 drop)。 - -## 验证的所有权模型 - -``` -caller driver thread kernel ------- ------------- ------ -read_at() ──Msg::Read──▶ 分配 buf,登记 pending 表 - (buf + Arc + oneshot tx) - push SQE(user_data=id) ─submit──▶ 开始随时可能写 buf -await ◀───oneshot──────── │ - │ -future drop(任意时刻) │ - └─(可选)Msg::Cancel ──▶ push ASYNC_CANCEL ────────────────▶│ 加速 CQE - └─绝不触碰 buf │ - CQE 到达 ◀─────────────────────────┘ - pending.remove(id) ← 全程唯一的 buf 回收点 - send 结果:成功=delivered - 失败(接收方已 drop)=orphan_reclaimed -``` - -关键不变量: - -1. **buffer 与 fd 归 pending 表所有,不归 future。** SQE 里的裸指针指向表项 `Vec` 的堆块;`Vec` 结构体可随 HashMap 移动(堆块地址不变),但在 CQE 前绝不 resize/drop。 -2. **fd 也必须由表项持有**(`Arc`)。真正的危险窗口是 **SQE 构造(`as_raw_fd`)→ `io_uring_enter` 内核消费**:此窗口内 SQE 携带裸 fd 号在 backlog 中滞留、内核尚未 `fget`;若 fd 被 drop 关闭并被新 `open` 复用,提交时内核解析到错误文件——对 READ op 意味着从**错误文件读出数据**(跨对象数据错读/泄露),而非"内核的写落到别人文件"。表项持有 `Arc` 到 CQE 是该窗口的安全超集。(机理更正见 rustfs/backlog#1063:原文把危险窗口误标为"提交→CQE"、后果误标为"写别人文件"——已提交的 op 因内核已持 struct file 引用而对 fd close/复用免疫;若未来用 SQPOLL,消费点还会与 enter 脱钩。) -3. **future drop 只放弃结果领取**,默认附带提交 `IORING_OP_ASYNC_CANCEL`(best-effort 加速),也可以不提交(裸 drop)——两种情况下回收都只发生在 CQE。 -4. **shutdown 顺序**:停收新 SQE → 对所有在途 op 提交 cancel → drain 到 `in_flight == 0` → 线程退出 → ring drop(unmap)。ring 决不能在内核仍持有 buffer 引用时 unmap。 -5. **探测必须提交真实 read op**:`io_uring_setup` 成功不代表 op 可用(gVisor/seccomp 可以建 ring 但 op ENOSYS/EINVAL);探测失败按 EACCES/EPERM/ENOSYS/EINVAL/EOPNOTSUPP 分类,命中即优雅降级(测试中表现为 skip),其余 errno 视为真 bug 直接断言失败。 - -以下不变量是本次审计整改(rustfs/backlog#1051)新增/固化,P2 必须一并沿用: - -6. **驱动线程 unwind 安全**(rustfs/backlog#1054):驱动线程绝不允许在栈展开中释放 pending 表或 unmap ring——否则内核仍可能向在途 buffer 写入即 UAF。实现为 `DriverState::Drop` 检测 `thread::panicking()` 时在字段析构前 `process::abort()`(leak over UAF)。所有 caller 可控的 panic 面(如超大 `len`)在 `submit` 入口拒止。`catch_unwind` 不够——析构在展开时、catch 边界之前就已发生。 -7. **背压 permit 在 CQE 点释放**(rustfs/backlog#1060;异步化见 #1102):in-flight 上界 ≤ CQ 容量(取 SQ 深度 `entries` < `2*entries`,使 CQ overflow 结构性不可达),permit 随 pending 表项移除(CQE)释放,**绝不随 future drop 释放**——否则 quorum 大量 drop future 会让 permit 计数与驻留内存脱钩,重开内存 DoS 面。 - - **已落地**:`tokio::sync::Semaphore`;`OwnedSemaphorePermit` 随 `Msg::Read` 存进 `Pending` 表项,表项在最终 CQE 被移除时 permit 自动 drop ——"CQE 点释放"由**类型系统强制**,不再依赖手写 `release()`(短读 resubmit 保留表项,故也保留 permit)。 - - **获取从不阻塞线程**:未饱和走 `try_acquire_owned()` 快路径(无分配、无 await、提交仍是即时的);饱和时把 acquire future 交给返回的 `ReadHandle`,首次 poll 时 await 到 permit 再提交。因此**一个在首次 poll 前就被 drop 的 handle 从未提交、从未分配 buffer**(比阻塞实现更省内存),`delivered + orphan_reclaimed == submitted` 的守恒式仍恒成立。 -8. **复用缓冲内容卫生**(rustfs/backlog#1062,P3 前置):当前 spike 每 op 新分配零页 + `truncate(res)`,**无泄露**。P3 改用驱动自有对齐 slab(registered buffer)后,缓冲跨请求复用即脏内存——任何路径忘记按 `cqe.res` 截断/掩蔽(O_DIRECT 整块读再由上层切片、或错误路径把整块缓冲交还)就把上一租户请求的对象字节泄给当前请求(CWE-226)。不变量:**复用缓冲对调用方可见的字节严格 ⊆ `[0, cqe.res)`**,越界部分零化或由类型系统(返回带长度上限的 view 而非整块 slice)保证不可达。docs/DESIGN.md 与 #1048 原约束只讲 slab 生命周期(防 UAF),不讲内容卫生;需配套"脏缓冲 + 短读"回归测试。 - -补充契约: - -- **errno 三分类**(rustfs/backlog#1059):`is_expected_restriction` **仅用于 probe 期**;运行期 errno 必须分——probe 受限 → 该盘永久降级;运行期参数错误(offset>i64::MAX、O_DIRECT 未对齐等 EINVAL)→ 返回错误、绝不闩锁;瞬态(EINTR/EAGAIN)→ 重试。`submit` 已在入口拒止 offset>i64::MAX 与 len>MAX_RW_COUNT。 -- **shutdown 有界 drain**(rustfs/backlog#1055):drain-to-zero 可能因坏盘上 cancel 不可中断(EALREADY)而不终止;超时(`DRAIN_TIMEOUT`)后泄漏 ring+buffer 退出(leak over UAF),绝不提前 unmap。cancel CQE 三态(succeeded/not_found/already)已纳入统计,EALREADY 上升即坏盘信号。 -- **短读 resubmit**(rustfs/backlog#1058):io_uring 对常规文件可合法短读;驱动 resubmit 剩余到 `buf[nread..]`,回收点移到逻辑读的最后一个 CQE。P2 须明确短读归属(后端循环 vs 调用方 `read_exact`)。 - -## 测试矩阵 - -| 测试 | 验证点 | -|---|---| -| `read_matches_std` | 完成路径正确性:64 次变长/变偏移读与文件内容逐字节一致 | -| `dropped_future_buffer_lives_until_cqe` | **核心断言**:阻塞的 pipe 读上裸 drop future(不提交 cancel),300ms 后 op 仍 in-flight、buffer 未回收;向 pipe 写入触发 CQE 后才回收(orphan_reclaimed=1) | -| `async_cancel_accelerates_reclaim` | 默认 drop 路径:ASYNC_CANCEL 使孤儿 op 在无数据到达的情况下经 ECANCELED CQE 及时回收 | -| `cancel_stress_accounts_for_every_buffer` | 压力:256 并发读、一半立即 drop;`delivered + orphan_reclaimed == submitted`,幸存读逐字节正确 | -| `shutdown_drains_in_flight_ops` | 关停:两个阻塞在途 op 被 cancel + drain 到 0 后线程才退出,持有的 future 解析为 ECANCELED | - -> 上表是 spike 原始 5 项核心断言。接入期实现随之新增覆盖,`tests/cancel.rs` 现共 **15 项**,补充:`saturated_submit_defers_instead_of_blocking`(异步背压不阻塞 runtime worker)、`direct_read_returns_exact_unaligned_ranges`(O_DIRECT 非对齐区间精确交付、填充不外泄)、`sharded_driver_conserves_buffers_across_shards` 与 `sharded_driver_with_one_shard_matches_single_ring`(分片下守恒 + 单片等价)、以及 boundary/pipe/EOF 等边界。所有项在 `run-docker.sh` 两腿下运行(leg 1 全优雅降级、leg 2 真实 io_uring)。 - -需要 Docker(Linux 内核)。macOS 宿主上 `cargo check` 只验证非 Linux 桩编译。 - -```bash -./run-docker.sh -``` - -- **leg 1(默认 seccomp)**:多数 Docker 版本默认禁 io_uring(即 #4313 事故环境),探测失败 → 全部测试走优雅降级 skip,套件仍绿。若宿主 Docker 放行 io_uring,则此腿等同 leg 2。 -- **leg 2(seccomp=unconfined)**:真实 io_uring,完整跑取消安全套件。 - -## 运行结果 - -两腿一次通过,详见"实测记录"。 - -## 对 P2 主体实现的遗留项 — 落地/决策现状 - -> 本节原为"本 spike 不覆盖"的清单;下列各项经 rustfs/backlog#1102/#1144/#1145/#1159 处理后现状如下。三种收尾:**✅ 已实现**、**⛔ 经度量/设计决策关闭(不做)**、**⬜ 仍未做(正确 defer)**。 - -- **✅ eventfd 唤醒收割(rustfs/backlog#1102)。** 200µs 忙轮询已由 eventfd 替换:一个 eventfd 注册到 ring(内核每 CQE 信号)、一个由 `submit`/shutdown 信号,驱动线程 `poll` 两者阻塞等待,`submit()` 每轮仍冲刷 NODROP overflow list。 - - **⛔ tokio `AsyncFd` 去驱动线程 — 不做。** `Drop` 不能 `await`,shutdown 的有界 drain 排空会被逼成公开 API 破坏。现"专用驱动线程 + eventfd 阻塞收割"已消除忙轮询,是无破坏的等价收益。 -- **✅→⛔ ring 生命周期 → 改为 per-disk 分片。** "进程级单例 ring"与坏盘隔离诉求(一块坏盘不得拖垮其它盘)相冲,已**重定义**为 per-disk ring 集,并用 `probe_and_start_sharded(entries, shards)` 在盘内横向扩展(每分片独立线程/pending 表/背压/eventfd)。Drop/shutdown 先让所有分片停再逐片 join,`DRAIN_TIMEOUT` 上界防坏盘无界阻塞。 -- **✅ O_DIRECT 对齐读:`read_at_direct(file, offset, len, align)`(rustfs/backlog#1102)。** 驱动读**块对齐超范围**到**块对齐 buffer**(超额分配 `align-1` 字节,在分配内部取第一个对齐字节作为读区起点),完成后只把逻辑区间 `[offset, offset+len)` 切出——**对齐填充、区间前缀、块对齐尾部一律不外泄**(否则 `BitrotReader` 会把补齐字节当损坏)。短读 resubmit 保持块对齐;内核返回非块倍数即文件尾。缓冲读是 `align == 1` 的退化情形。 - - **✅ ecstore 已原生接线**(rustfs/rustfs#4649):`pread_uring_direct` 以 O_DIRECT 打开 fd 并调 `read_at_direct`,分层兜底 + per-disk 闩锁。**已取代 #4645 的临时分流**(那版把 O_DIRECT 合格读交回 StdBackend)。 -- **✅→⛔ 三读形态接入 `LocalIoBackend` — 部分做、部分不做。** **定位读 `pread_bytes` 已接** io_uring(缓冲 + 原生 O_DIRECT)。**流式读 `open_read_stream`/`open_full_read` 判 NO-GO**(rustfs/backlog#1144):io_uring 对单条顺序流无杠杆(内核 readahead 已胜),冷读设备瓶颈、暖读仅 11–41% 于 buffered——保持委托 StdBackend。 -- **✅ per-disk 探测缓存 + 运行期 errno 降级闩锁(rustfs/backlog#1101)。** `URING_UNSUPPORTED_DISKS` 负探测缓存 + `is_io_uring_unsupported` 运行期 errno 三分类(仅限制类 `EPERM/EACCES/ENOSYS/EOPNOTSUPP` 触发整盘闩死,数据/参数错误不闩),与 probe 期分类分离。 -- **⬜ registered buffers(P3)/ 写路径(P4)— 仍未做(正确 defer)。** `register_buffers` 与本文档"buffer 归 pending 表、`Vec` 所有权"的取消安全模型冲突,需重设计;且 #1159 端到端 profiling 显示读路径已非瓶颈,优先级低。写路径(PUT)完全未涉及,profiling 提示其收益可能大于读路径。 - -**已在本 spike 内整改**(rustfs/backlog#1051):SQ 深度背压(不变量 7)、驱动线程 unwind 安全(不变量 6)、shutdown 有界 drain、CQ overflow/NODROP/EBUSY 处理、probe UAF、probe 文件安全创建、errno 三分类、len/offset 校验、短读 resubmit。 - -**接入后新增的行为契约**(实现期发现,非 spike 原文):`UringBackend` 读完须遵守 StdBackend 的页缓存回收策略(≥4 MiB 读后 `fadvise(DONTNEED)`),否则开启 io_uring 即静默污染页缓存(rustfs/rustfs#4662 修复的回归)。 - -## 实测记录 - -2026-07-07,宿主 macOS + OrbStack Docker(Linux arm64,内核 7.0.11-orbstack),镜像 `rust:1-bookworm`,`cargo test --release`: - -- **leg 1(默认 seccomp)**:`io_uring_setup` 失败 `EPERM (Operation not permitted)`——与 #4313 事故环境同类。`ProbeFailure::is_expected_restriction()` 命中,5 个测试全部优雅降级 skip,套件绿。证明探测 + errno 分类降级契约按设计工作。 -- **leg 2(seccomp=unconfined)**:5 个测试全部通过(0.45s): - - `read_matches_std` ok — 64 次读逐字节正确; - - `dropped_future_buffer_lives_until_cqe` ok — 裸 drop 后 op 保持 in-flight 300ms、buffer 未回收,写 pipe 触发 CQE 后 `orphan_reclaimed=1`; - - `async_cancel_accelerates_reclaim` ok — ECANCELED CQE 路径回收; - - `cancel_stress_accounts_for_every_buffer` ok — 256 op、128 drop,`delivered(128) + orphan_reclaimed(128) == submitted(256)`; - - `shutdown_drains_in_flight_ops` ok — drain 到 0 后退出,持有 future 解析为 ECANCELED。 - -**结论:GO(模型可行)。** buffer/fd 归驱动 pending 表、CQE 唯一回收点、ASYNC_CANCEL 加速、shutdown drain 的组合在真实内核上成立,且降级契约在受限环境下按设计生效。**P2 主体已据此模型实现**(见上"落地/决策现状"),接入 `rustfs/rustfs` 后默认灰度关,15 项取消安全验收测试是这些实现共同遵守的回归门禁。 diff --git a/src/lib.rs b/src/lib.rs index e66c2d2..e6ae45e 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -30,12 +30,11 @@ //! unmapped. //! //! Status: read path only, Linux only. The driver supports positioned buffered -//! reads, `O_DIRECT` reads with internal alignment, sharded rings, async -//! backpressure, eventfd-driven reaping, graceful restricted-environment -//! detection, and bounded shutdown drain. The write path is intentionally out -//! of scope; see the -//! [design notes](https://github.com/rustfs/uring/blob/0.2.0/docs/DESIGN.md) -//! for the invariant details. +//! reads, `O_DIRECT` reads with internal alignment, stream (`read(2)`) reads, +//! sharded rings, async backpressure, eventfd-driven reaping, graceful +//! restricted-environment detection, and bounded shutdown drain. The write path +//! is intentionally out of scope. See the crate README and the per-item docs +//! below for the invariant details. #[cfg(target_os = "linux")] mod driver; From 8878e508a00657df13910f5b6432ffe4f9e3ec56 Mon Sep 17 00:00:00 2001 From: houseme Date: Sat, 11 Jul 2026 18:24:04 +0800 Subject: [PATCH 15/15] fix(driver): clear clippy blockers on the O_DIRECT and drop-cancel paths MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The Linux-only driver failed `clippy -D warnings`: - the O_DIRECT aligned read buffer was bound `mut`, but it is only read (`as_ptr`/`len`) before being moved into the pending table, which owns and mutates it — so the binding needs no `mut` (unused_mut). - the `ReadHandle::drop` cancel path nested `if !finished && cancel_on_drop` inside the `if let Submitted` guard; collapse into one let-chain (collapsible_if). No behavior change. Verified with `cargo clippy --target x86_64-unknown-linux-gnu --all-targets -- -D warnings` (cross-checks the `cfg(target_os = "linux")` driver module from macOS). Co-Authored-By: heihutu --- src/driver.rs | 35 ++++++++++++++++++----------------- 1 file changed, 18 insertions(+), 17 deletions(-) diff --git a/src/driver.rs b/src/driver.rs index 5b0ffaa..83e4644 100644 --- a/src/driver.rs +++ b/src/driver.rs @@ -550,22 +550,23 @@ impl Drop for ReadHandle { // until the CQE. All we may do is ask the kernel to hurry up. A handle // dropped before it was submitted (Inert / WaitingPermit) has no buffer, // no permit and no SQE, so there is nothing to cancel. - if let HandleState::Submitted { wake } = &self.state { - if !self.finished && self.cancel_on_drop { - // Close the receiver BEFORE waking the driver. The wake below - // makes the driver process the cancel immediately, possibly while - // this drop is still running — before the `rx` field is - // destroyed. Closing it first guarantees the cancel-induced - // completion the driver reaps is counted as an orphan reclaim, not - // delivered to a receiver that is about to drop anyway - // (rustfs/backlog#1163). - self.rx.close(); - let _ = self.tx.send(Msg::Cancel { id: self.id }); - // Wake the loop so the cancel is queued now, not after the - // heartbeat. On an idle ring (the hung-disk case cancel-on-drop - // exists for) this keeps orphan reclamation prompt. - wake.signal(); - } + if let HandleState::Submitted { wake } = &self.state + && !self.finished + && self.cancel_on_drop + { + // Close the receiver BEFORE waking the driver. The wake below + // makes the driver process the cancel immediately, possibly while + // this drop is still running — before the `rx` field is + // destroyed. Closing it first guarantees the cancel-induced + // completion the driver reaps is counted as an orphan reclaim, not + // delivered to a receiver that is about to drop anyway + // (rustfs/backlog#1163). + self.rx.close(); + let _ = self.tx.send(Msg::Cancel { id: self.id }); + // Wake the loop so the cancel is queued now, not after the + // heartbeat. On an idle ring (the hung-disk case cancel-on-drop + // exists for) this keeps orphan reclamation prompt. + wake.signal(); } } } @@ -1444,7 +1445,7 @@ fn drive( continue; } }; - let mut buf = vec![0u8; cap]; + let buf = vec![0u8; cap]; let pad = buf.as_ptr().align_offset(align); // Runtime guard (not a debug-only assert): if the allocator // ever returned a block `align_offset` cannot satisfy, refuse