From d608e320f751690ec6aad9a74073452f0976624d Mon Sep 17 00:00:00 2001 From: houseme Date: Tue, 7 Jul 2026 23:39:14 +0800 Subject: [PATCH] chore(experiments): add io_uring cancel-safety spike (backlog#894) Standalone-workspace prototype proving the buffer-ownership model P2 requires before any production io_uring work: buffer and fd are owned by the driver's pending table from SQE submission to CQE, dropping the future never reclaims the buffer, IORING_OP_ASYNC_CANCEL only accelerates the CQE, and shutdown drains to in_flight == 0 before the ring is unmapped. Deliberately excluded from the rustfs workspace (own [workspace] table): P2 stays deferred per the P1.5 NO-GO gate, so io-uring must not enter the main Cargo.lock. Verified via run-docker.sh on a real Linux kernel: default-seccomp leg degrades gracefully (EPERM probe, the #4313 environment), unconfined leg passes all five cancel-safety tests. Co-Authored-By: heihutu --- experiments/io-uring-cancel-spike/.gitignore | 1 + experiments/io-uring-cancel-spike/Cargo.lock | 103 +++++ experiments/io-uring-cancel-spike/Cargo.toml | 20 + experiments/io-uring-cancel-spike/SPIKE.md | 85 ++++ .../io-uring-cancel-spike/run-docker.sh | 36 ++ .../io-uring-cancel-spike/src/driver.rs | 421 ++++++++++++++++++ experiments/io-uring-cancel-spike/src/lib.rs | 23 + .../io-uring-cancel-spike/tests/cancel.rs | 245 ++++++++++ 8 files changed, 934 insertions(+) create mode 100644 experiments/io-uring-cancel-spike/.gitignore create mode 100644 experiments/io-uring-cancel-spike/Cargo.lock create mode 100644 experiments/io-uring-cancel-spike/Cargo.toml create mode 100644 experiments/io-uring-cancel-spike/SPIKE.md create mode 100755 experiments/io-uring-cancel-spike/run-docker.sh create mode 100644 experiments/io-uring-cancel-spike/src/driver.rs create mode 100644 experiments/io-uring-cancel-spike/src/lib.rs create mode 100644 experiments/io-uring-cancel-spike/tests/cancel.rs diff --git a/experiments/io-uring-cancel-spike/.gitignore b/experiments/io-uring-cancel-spike/.gitignore new file mode 100644 index 000000000..ea8c4bf7f --- /dev/null +++ b/experiments/io-uring-cancel-spike/.gitignore @@ -0,0 +1 @@ +/target diff --git a/experiments/io-uring-cancel-spike/Cargo.lock b/experiments/io-uring-cancel-spike/Cargo.lock new file mode 100644 index 000000000..33e1b8191 --- /dev/null +++ b/experiments/io-uring-cancel-spike/Cargo.lock @@ -0,0 +1,103 @@ +# This file is automatically @generated by Cargo. +# It is not intended for manual editing. +version = 4 + +[[package]] +name = "bitflags" +version = "2.13.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b4388bee8683e3d04af747c73422af53102d2bd24d9eadb6cbc100baef4b43f8" + +[[package]] +name = "cfg-if" +version = "1.0.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "9330f8b2ff13f34540b44e946ef35111825727b38d33286ef986142615121801" + +[[package]] +name = "io-uring" +version = "0.7.13" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "9080b15e63775b9a2ac7dca720f7050a8b955e092ea0f6020a4a80f69998cdc0" +dependencies = [ + "bitflags", + "cfg-if", + "libc", +] + +[[package]] +name = "io-uring-cancel-spike" +version = "0.1.0" +dependencies = [ + "io-uring", + "libc", + "tokio", +] + +[[package]] +name = "libc" +version = "0.2.186" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "68ab91017fe16c622486840e4c83c9a37afeff978bd239b5293d61ece587de66" + +[[package]] +name = "pin-project-lite" +version = "0.2.17" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "a89322df9ebe1c1578d689c92318e070967d1042b512afbe49518723f4e6d5cd" + +[[package]] +name = "proc-macro2" +version = "1.0.106" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "8fd00f0bb2e90d81d1044c2b32617f68fcb9fa3bb7640c23e9c748e53fb30934" +dependencies = [ + "unicode-ident", +] + +[[package]] +name = "quote" +version = "1.0.46" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "dfbc457d0c7a0759a614551b11a6409e5951f6c7537be1f1b7682b9ae9230368" +dependencies = [ + "proc-macro2", +] + +[[package]] +name = "syn" +version = "2.0.118" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "1b9ae57f904213ebb649ce6895b8a66c66f0203b9319718f69a5612a065b1422" +dependencies = [ + "proc-macro2", + "quote", + "unicode-ident", +] + +[[package]] +name = "tokio" +version = "1.52.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "8fc7f01b389ac15039e4dc9531aa973a135d7a4135281b12d7c1bc79fd57fffe" +dependencies = [ + "pin-project-lite", + "tokio-macros", +] + +[[package]] +name = "tokio-macros" +version = "2.7.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "385a6cb71ab9ab790c5fe8d67f1645e6c450a7ce006a33de03daa956cf70a496" +dependencies = [ + "proc-macro2", + "quote", + "syn", +] + +[[package]] +name = "unicode-ident" +version = "1.0.24" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e6e4313cd5fcd3dad5cafa179702e2b244f760991f45397d14d4ebf38247da75" diff --git a/experiments/io-uring-cancel-spike/Cargo.toml b/experiments/io-uring-cancel-spike/Cargo.toml new file mode 100644 index 000000000..6e65db1a5 --- /dev/null +++ b/experiments/io-uring-cancel-spike/Cargo.toml @@ -0,0 +1,20 @@ +[package] +name = "io-uring-cancel-spike" +version = "0.1.0" +edition = "2024" +publish = false +description = "Spike 0 for backlog#894 (P2 io_uring read backend): cancel-safety / buffer-ownership prototype. NOT production code." + +# Deliberately a standalone workspace: P2 is deferred (P1.5 NO-GO), so the +# io-uring crate must not enter the main rustfs Cargo.lock or build graph. +[workspace] + +[dependencies] +tokio = { version = "1", default-features = false, features = ["sync"] } + +[target.'cfg(target_os = "linux")'.dependencies] +io-uring = "0.7" +libc = "0.2" + +[dev-dependencies] +tokio = { version = "1", features = ["rt-multi-thread", "macros", "sync", "time"] } diff --git a/experiments/io-uring-cancel-spike/SPIKE.md b/experiments/io-uring-cancel-spike/SPIKE.md new file mode 100644 index 000000000..c45ddea26 --- /dev/null +++ b/experiments/io-uring-cancel-spike/SPIKE.md @@ -0,0 +1,85 @@ +# Spike 0: io_uring 取消安全原型(backlog#894 P2 前置) + +## 这是什么 + +rustfs/backlog#897 路线图中 P2(io_uring 读后端)被 P1.5 基准判 NO-GO 而 defer。本 spike 是 #894 明确要求先行的**取消安全原型**——P2 中风险最高、最容易随时间流失的知识,按"只实现原型、不进主干、不启用"的方案 B 存档。重启条件满足前,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 quorum 达成 / 断连 / 超时会 drop 在途的 `BitrotReaderTask` future(main 上位于 `crates/ecstore/src/set_disk/core/io_primitives.rs:1455` 的 `FuturesUnordered`)。若该 future 已向内核提交了 read SQE,内核在 CQE 之前始终可能向目标 buffer 写入。**future 的 drop 不能回收 buffer,否则是 use-after-free。** + +## 验证的所有权模型 + +``` +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`)。只保 buffer 不保 fd 的话,future drop 会关闭 fd,fd 号被复用后内核的写落到别人的文件上——这是本 spike 落笔前未在 #894 文字中显式化的坑,已验证必须纳入 P2 设计。 +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 直接断言失败。 + +## 测试矩阵 + +| 测试 | 验证点 | +|---|---| +| `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 | + +## 如何运行 + +需要 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 不覆盖) + +- eventfd + tokio `AsyncFd` 收割替换轮询驱动循环;SQ 深度 `Semaphore` 背压。 +- 进程级单例 ring 的生命周期管理(本 spike 每测试一个 ring)。 +- O_DIRECT 对齐 buffer(P1 的 statx 探测复用)、三条读形态接入 `LocalIoBackend`。 +- per-disk 探测缓存与运行期 errno 降级闩锁(参照 main 上 `DirectIoReadState`,local.rs:278)。 +- registered buffers(P3)/写路径(P4)完全不涉及。 + +## 实测记录 + +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 主体重启时可直接沿用此所有权模型。 diff --git a/experiments/io-uring-cancel-spike/run-docker.sh b/experiments/io-uring-cancel-spike/run-docker.sh new file mode 100755 index 000000000..b068abdf5 --- /dev/null +++ b/experiments/io-uring-cancel-spike/run-docker.sh @@ -0,0 +1,36 @@ +#!/usr/bin/env bash +# Runs the spike test suite in the two environments that matter for P2 +# (backlog#894): +# +# leg 1 default Docker seccomp — io_uring is (usually) blocked, which is +# exactly the #4313 incident environment. Tests must degrade to a +# graceful skip via the probe, never fail. +# leg 2 seccomp=unconfined — real io_uring against the host kernel. +# The full cancel-safety suite runs. +set -euo pipefail +cd "$(dirname "$0")" + +IMG="${SPIKE_IMAGE:-rust:1-bookworm}" +CACHE_REG=uring-spike-cargo-registry +CACHE_TARGET=uring-spike-target + +run_leg() { + local title="$1" + shift + echo "==================================================================" + echo "== $title" + echo "==================================================================" + docker run --rm "$@" \ + -v "$PWD":/spike:ro \ + -v "$CACHE_REG":/usr/local/cargo/registry \ + -v "$CACHE_TARGET":/spike-target \ + -e CARGO_TARGET_DIR=/spike-target \ + -e CARGO_TERM_COLOR=always \ + -w /spike \ + "$IMG" \ + cargo test --release -- --nocapture --test-threads=1 +} + +run_leg "leg 1: default seccomp (restricted env expected)" +run_leg "leg 2: seccomp=unconfined (real io_uring)" --security-opt seccomp=unconfined +echo "both legs passed" diff --git a/experiments/io-uring-cancel-spike/src/driver.rs b/experiments/io-uring-cancel-spike/src/driver.rs new file mode 100644 index 000000000..9d71ae7bd --- /dev/null +++ b/experiments/io-uring-cancel-spike/src/driver.rs @@ -0,0 +1,421 @@ +use std::collections::{HashMap, VecDeque}; +use std::fs::File; +use std::io; +use std::os::fd::AsRawFd; +use std::pin::Pin; +use std::sync::Arc; +use std::sync::atomic::{AtomicU64, Ordering}; +use std::sync::mpsc::{self, RecvTimeoutError, TryRecvError}; +use std::task::{Context, Poll}; +use std::thread::JoinHandle; +use std::time::Duration; + +use io_uring::{IoUring, opcode, types}; +use tokio::sync::oneshot; + +/// user_data bit marking the CQE of an `AsyncCancel` SQE itself (as opposed +/// to the CQE of the read op it targets). +const CANCEL_BIT: u64 = 1 << 63; + +/// `offset` value meaning "use the file's current position" (read(2) +/// semantics); required for pipes/sockets where pread returns ESPIPE. +const CURRENT_POSITION: u64 = u64::MAX; + +/// Why the probe refused to start the io_uring driver. +/// +/// Mirrors the P2 degradation contract (backlog#894): a restricted +/// environment must be recognized and answered with a silent fallback to the +/// std backend, never surfaced to callers. +#[derive(Debug)] +pub enum ProbeFailure { + /// `io_uring_setup` itself failed (seccomp/gVisor/old kernel). + Setup(io::Error), + /// The ring was created but a real `IORING_OP_READ` did not complete + /// correctly (gVisor accepts setup but fails ops; also covers silent + /// data corruption, which we treat as "unusable"). + ReadOp(io::Error), +} + +impl ProbeFailure { + /// True when the errno belongs to the "expected restriction" class that + /// P2 maps to permanent per-disk fallback: EACCES/EPERM/ENOSYS/EINVAL/ + /// EOPNOTSUPP. Anything else would be a genuine bug worth surfacing. + pub fn is_expected_restriction(&self) -> bool { + let err = match self { + ProbeFailure::Setup(e) | ProbeFailure::ReadOp(e) => e, + }; + matches!( + err.raw_os_error(), + Some(libc::EACCES) | Some(libc::EPERM) | Some(libc::ENOSYS) | Some(libc::EINVAL) | Some(libc::EOPNOTSUPP) + ) + } +} + +#[derive(Default)] +struct DriverStats { + submitted: AtomicU64, + delivered: AtomicU64, + orphan_reclaimed: AtomicU64, + in_flight: AtomicU64, +} + +/// Point-in-time copy of the driver counters. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub struct StatsSnapshot { + /// Read ops handed to the kernel. + pub submitted: u64, + /// CQEs whose result was received by a live caller. + pub delivered: u64, + /// CQEs whose caller had dropped the future: the buffer stayed in the + /// pending table the whole time and was reclaimed here, at the CQE. + pub orphan_reclaimed: u64, + /// Ops submitted but not yet completed. The kernel may still write into + /// their buffers. + pub in_flight: u64, +} + +enum Msg { + Read { + id: u64, + file: Arc, + offset: u64, + len: usize, + done: oneshot::Sender>>, + }, + Cancel { + id: u64, + }, + Shutdown, +} + +/// One in-flight read. This struct — not the caller — owns everything the +/// kernel touches: +/// +/// - `buf`: the destination buffer. Its heap allocation must stay put until +/// the CQE; the `Vec` itself may move (HashMap rehash) since that never +/// relocates the heap block. It is never read, resized, or dropped before +/// the CQE handler removes this entry. +/// - `_file`: keeps the fd open even if every caller-side clone is dropped. +/// Without this, dropping the future could close the fd mid-read and a +/// recycled fd number would receive the kernel's write (spike finding: +/// the orphan table must own the file handle, not just the buffer). +struct Pending { + buf: Vec, + _file: Arc, + done: Option>>>, +} + +/// Handle to a submitted read. Await it for the result. +/// +/// Dropping it before completion abandons the result only; by default it +/// also submits `IORING_OP_ASYNC_CANCEL` (best effort) so the CQE — and with +/// it the buffer reclamation — arrives sooner. `without_cancel_on_drop` +/// disables that to model the bare "quorum drops the future" case. +pub struct ReadHandle { + id: u64, + rx: oneshot::Receiver>>, + tx: mpsc::Sender, + finished: bool, + cancel_on_drop: bool, +} + +impl ReadHandle { + pub fn without_cancel_on_drop(mut self) -> Self { + self.cancel_on_drop = false; + self + } +} + +impl Future for ReadHandle { + type Output = io::Result>; + + fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll { + match Pin::new(&mut self.rx).poll(cx) { + Poll::Ready(res) => { + self.finished = true; + Poll::Ready(match res { + Ok(inner) => inner, + Err(_) => Err(io::Error::other("uring driver shut down before completion")), + }) + } + Poll::Pending => Poll::Pending, + } + } +} + +impl Drop for ReadHandle { + fn drop(&mut self) { + // The buffer is deliberately NOT touched here: the driver owns it + // until the CQE. All we may do is ask the kernel to hurry up. + if !self.finished && self.cancel_on_drop { + let _ = self.tx.send(Msg::Cancel { id: self.id }); + } + } +} + +/// Process-level io_uring driver: one ring, one driver thread. +pub struct UringDriver { + tx: mpsc::Sender, + handle: Option>, + stats: Arc, + next_id: AtomicU64, +} + +impl UringDriver { + /// Create the ring AND verify a real `IORING_OP_READ` round-trip on a + /// temp file before accepting work. `io_uring_setup` succeeding is not + /// enough: gVisor/seccomp environments can create a ring whose ops then + /// fail with ENOSYS/EINVAL (backlog#894 probe design). + pub fn probe_and_start(entries: u32) -> Result { + let mut ring = IoUring::new(entries).map_err(ProbeFailure::Setup)?; + probe_real_read(&mut ring).map_err(ProbeFailure::ReadOp)?; + + let (tx, rx) = mpsc::channel(); + let stats = Arc::new(DriverStats::default()); + let thread_stats = Arc::clone(&stats); + let handle = std::thread::Builder::new() + .name("uring-spike-driver".into()) + .spawn(move || drive(ring, rx, thread_stats)) + .expect("spawn driver thread"); + + Ok(Self { + tx, + handle: Some(handle), + stats, + next_id: AtomicU64::new(1), + }) + } + + /// Positioned read (pread semantics) — regular files. + pub fn read_at(&self, file: Arc, offset: u64, len: usize) -> ReadHandle { + assert_ne!(offset, CURRENT_POSITION, "offset u64::MAX is reserved"); + self.submit(file, offset, len) + } + + /// Read at the file's current position (read(2) semantics) — pipes. + pub fn read_current(&self, file: Arc, len: usize) -> ReadHandle { + self.submit(file, CURRENT_POSITION, len) + } + + fn submit(&self, file: Arc, offset: u64, len: usize) -> ReadHandle { + let id = self.next_id.fetch_add(1, Ordering::Relaxed); + assert_eq!(id & CANCEL_BIT, 0, "op id overflowed into the cancel bit"); + let (done, rx) = oneshot::channel(); + // If the driver already shut down, the send fails and `done` is + // dropped, which the caller observes as a driver-gone error. + let _ = self.tx.send(Msg::Read { + id, + file, + offset, + len, + done, + }); + ReadHandle { + id, + rx, + tx: self.tx.clone(), + finished: false, + cancel_on_drop: true, + } + } + + pub fn stats(&self) -> StatsSnapshot { + StatsSnapshot { + submitted: self.stats.submitted.load(Ordering::SeqCst), + delivered: self.stats.delivered.load(Ordering::SeqCst), + orphan_reclaimed: self.stats.orphan_reclaimed.load(Ordering::SeqCst), + in_flight: self.stats.in_flight.load(Ordering::SeqCst), + } + } + + /// Stop accepting work, cancel all in-flight ops, drain the ring to + /// `in_flight == 0`, then join the driver thread. Only after that is the + /// ring dropped/unmapped — the shutdown ordering P2 requires. + pub fn shutdown(mut self) -> StatsSnapshot { + let _ = self.tx.send(Msg::Shutdown); + if let Some(h) = self.handle.take() { + let _ = h.join(); + } + let snap = self.stats(); + assert_eq!(snap.in_flight, 0, "driver exited with ops still in flight"); + snap + } +} + +impl Drop for UringDriver { + fn drop(&mut self) { + if let Some(h) = self.handle.take() { + let _ = self.tx.send(Msg::Shutdown); + let _ = h.join(); + } + } +} + +fn probe_real_read(ring: &mut IoUring) -> io::Result<()> { + static PROBE_SEQ: AtomicU64 = AtomicU64::new(0); + let pattern: Vec = (0..512u32).map(|i| (i * 7 + 13) as u8).collect(); + let path = std::env::temp_dir().join(format!( + "uring-spike-probe-{}-{}", + std::process::id(), + PROBE_SEQ.fetch_add(1, Ordering::Relaxed) + )); + let result = (|| { + std::fs::write(&path, &pattern)?; + let file = File::open(&path)?; + let mut buf = vec![0u8; pattern.len()]; + let sqe = opcode::Read::new(types::Fd(file.as_raw_fd()), buf.as_mut_ptr(), buf.len() as u32) + .offset(0) + .build() + .user_data(0xB0BE); + // SAFETY: `buf` and `file` outlive the synchronous wait below. + unsafe { + ring.submission() + .push(&sqe) + .map_err(|_| io::Error::other("probe: submission queue full"))?; + } + ring.submit_and_wait(1)?; + let cqe = ring.completion().next().ok_or_else(|| io::Error::other("probe: no CQE"))?; + if cqe.result() < 0 { + return Err(io::Error::from_raw_os_error(-cqe.result())); + } + if cqe.result() as usize != pattern.len() || buf != pattern { + return Err(io::Error::other("probe: read completed but data mismatched")); + } + Ok(()) + })(); + let _ = std::fs::remove_file(&path); + result +} + +fn drive(mut ring: IoUring, rx: mpsc::Receiver, stats: Arc) { + let mut pending: HashMap = HashMap::new(); + let mut backlog: VecDeque = VecDeque::new(); + let mut shutting_down = false; + + loop { + // 1. Intake. Block (with timeout) only when fully idle; once ops are + // in flight we must keep reaping, so only try_recv. + loop { + let idle = pending.is_empty() && backlog.is_empty() && !shutting_down; + let msg = if idle { + match rx.recv_timeout(Duration::from_millis(50)) { + Ok(m) => m, + Err(RecvTimeoutError::Timeout) => break, + Err(RecvTimeoutError::Disconnected) => { + shutting_down = true; + break; + } + } + } else { + match rx.try_recv() { + Ok(m) => m, + Err(TryRecvError::Empty) => break, + Err(TryRecvError::Disconnected) => { + shutting_down = true; + break; + } + } + }; + match msg { + Msg::Read { + id, + file, + offset, + len, + done, + } => { + if shutting_down { + let _ = done.send(Err(io::Error::other("uring driver shutting down"))); + continue; + } + let mut buf = vec![0u8; len]; + // 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. + let sqe = opcode::Read::new(types::Fd(file.as_raw_fd()), buf.as_mut_ptr(), len as u32) + .offset(offset) + .build() + .user_data(id); + pending.insert( + id, + Pending { + buf, + _file: file, + done: Some(done), + }, + ); + stats.submitted.fetch_add(1, Ordering::SeqCst); + stats.in_flight.fetch_add(1, Ordering::SeqCst); + backlog.push_back(sqe); + } + Msg::Cancel { id } => { + if pending.contains_key(&id) { + backlog.push_back(opcode::AsyncCancel::new(id).build().user_data(id | CANCEL_BIT)); + } + } + Msg::Shutdown => { + shutting_down = true; + for id in pending.keys() { + backlog.push_back(opcode::AsyncCancel::new(*id).build().user_data(*id | CANCEL_BIT)); + } + } + } + } + + // 2. Push backlog into the SQ (stop when full; retry next turn). + { + 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; + } + } + } + if ring.submit().is_err() { + // EINTR and friends: retry on the next loop turn. + } + + // 3. Reap. This is the ONLY place a Pending entry (and thus its + // buffer) is dropped. + while let Some(cqe) = ring.completion().next() { + let ud = cqe.user_data(); + if ud & CANCEL_BIT != 0 { + // Result of the AsyncCancel op itself; the read's own CQE + // (ECANCELED or success) still arrives separately. + continue; + } + let Some(mut p) = pending.remove(&ud) else { + continue; + }; + let res = cqe.result(); + let outcome = if res < 0 { + Err(io::Error::from_raw_os_error(-res)) + } else { + p.buf.truncate(res as usize); + Ok(std::mem::take(&mut p.buf)) + }; + match p.done.take().expect("done sender set at submit").send(outcome) { + Ok(()) => stats.delivered.fetch_add(1, Ordering::SeqCst), + // Caller dropped the future: the buffer survived in the + // table until this very CQE and is reclaimed here. + Err(_) => stats.orphan_reclaimed.fetch_add(1, Ordering::SeqCst), + }; + stats.in_flight.fetch_sub(1, Ordering::SeqCst); + } + + // 4. Exit only when drained: kernel no longer references any buffer, + // so dropping the ring (unmap) is safe. + if shutting_down && pending.is_empty() && backlog.is_empty() { + return; + } + + // Spike-grade pacing: production replaces this poll with eventfd + + // tokio AsyncFd reaping (backlog#894). + if !pending.is_empty() { + std::thread::sleep(Duration::from_micros(200)); + } + } +} diff --git a/experiments/io-uring-cancel-spike/src/lib.rs b/experiments/io-uring-cancel-spike/src/lib.rs new file mode 100644 index 000000000..2139697df --- /dev/null +++ b/experiments/io-uring-cancel-spike/src/lib.rs @@ -0,0 +1,23 @@ +//! Spike 0 for rustfs/backlog#894 (P2 io_uring read backend). +//! +//! Proves the cancel-safety ownership model before any production work: +//! +//! - The read buffer and the file handle are owned by the driver's pending +//! (orphan) table from SQE submission until the CQE arrives. The kernel may +//! write into the buffer at any point in that window, so nothing else is +//! allowed to free or move its heap allocation. +//! - Dropping the caller-side future only abandons the *result*. It never +//! touches the buffer. Optionally it submits `IORING_OP_ASYNC_CANCEL` to +//! accelerate the CQE; reclamation still happens only at the CQE. +//! - Driver shutdown cancels all in-flight ops and drains the ring to +//! `in_flight == 0` before the ring is dropped (unmapped). +//! +//! NOT production code: the driver thread uses a coarse poll loop instead of +//! eventfd + `AsyncFd` reaping, there is no SQ-depth semaphore backpressure, +//! no O_DIRECT alignment, and only one read shape. See SPIKE.md. + +#[cfg(target_os = "linux")] +mod driver; + +#[cfg(target_os = "linux")] +pub use driver::{ProbeFailure, StatsSnapshot, UringDriver}; diff --git a/experiments/io-uring-cancel-spike/tests/cancel.rs b/experiments/io-uring-cancel-spike/tests/cancel.rs new file mode 100644 index 000000000..a78881fc1 --- /dev/null +++ b/experiments/io-uring-cancel-spike/tests/cancel.rs @@ -0,0 +1,245 @@ +//! Cancel-safety acceptance tests for backlog#894 Spike 0. +//! +//! In a restricted environment (Docker default seccomp, gVisor) the probe +//! fails with an expected-restriction errno and every test degrades to a +//! skip — the same contract P2's production probe must honor. Run under +//! `--security-opt seccomp=unconfined` (see run-docker.sh) to exercise the +//! real io_uring paths. +#![cfg(target_os = "linux")] + +use std::fs::File; +use std::io::Write; +use std::os::fd::FromRawFd; +use std::sync::Arc; +use std::time::{Duration, Instant}; + +use io_uring_cancel_spike::UringDriver; + +fn driver_or_skip(name: &str) -> Option { + match UringDriver::probe_and_start(64) { + Ok(d) => Some(d), + Err(e) => { + assert!( + e.is_expected_restriction(), + "probe failed OUTSIDE the expected restriction errno class \ + (EACCES/EPERM/ENOSYS/EINVAL/EOPNOTSUPP): {e:?}" + ); + eprintln!("SKIP {name}: restricted environment, graceful degradation path taken ({e:?})"); + None + } + } +} + +/// Deterministic pseudo-random content so reads are verifiable. +fn make_content(len: usize) -> Vec { + let mut state: u64 = 0x2545F4914F6CDD1D; + (0..len) + .map(|_| { + state ^= state << 13; + state ^= state >> 7; + state ^= state << 17; + state as u8 + }) + .collect() +} + +fn temp_file_with(content: &[u8], tag: &str) -> (std::path::PathBuf, Arc) { + let path = std::env::temp_dir().join(format!("uring-spike-{tag}-{}", std::process::id())); + std::fs::write(&path, content).expect("write temp file"); + let file = Arc::new(File::open(&path).expect("open temp file")); + (path, file) +} + +/// An OS pipe whose read side never completes until we write — the only +/// deterministic way to hold an op in flight across a future drop. +fn os_pipe() -> (Arc, File) { + let mut fds = [0i32; 2]; + // SAFETY: fds is a valid out-array; on success both fds are owned here + // and immediately wrapped in File which takes over closing them. + let rc = unsafe { libc::pipe(fds.as_mut_ptr()) }; + assert_eq!(rc, 0, "pipe(2) failed"); + let read = unsafe { File::from_raw_fd(fds[0]) }; + let write = unsafe { File::from_raw_fd(fds[1]) }; + (Arc::new(read), write) +} + +async fn wait_until(deadline: Duration, mut cond: impl FnMut() -> bool) -> bool { + let start = Instant::now(); + while start.elapsed() < deadline { + if cond() { + return true; + } + tokio::time::sleep(Duration::from_millis(5)).await; + } + cond() +} + +/// Baseline: completed reads return exactly what pread would. +#[tokio::test(flavor = "multi_thread")] +async fn read_matches_std() { + let Some(driver) = driver_or_skip("read_matches_std") else { + return; + }; + const LEN: usize = 8 << 20; + let content = make_content(LEN); + let (path, file) = temp_file_with(&content, "correctness"); + + for i in 0..64usize { + let offset = (i * 131_071) % (LEN - 70_000); + let len = 1 + (i * 8_191) % 65_536; + let got = driver + .read_at(Arc::clone(&file), offset as u64, len) + .await + .expect("read failed"); + assert_eq!(got, &content[offset..offset + len], "mismatch at offset {offset} len {len}"); + } + + let snap = driver.shutdown(); + assert_eq!(snap.delivered, 64); + assert_eq!(snap.orphan_reclaimed, 0); + let _ = std::fs::remove_file(path); +} + +/// THE core spike assertion: drop the future while the op is provably still +/// in flight (blocked pipe read, no cancel submitted) and verify the buffer +/// stays owned by the driver until the CQE finally arrives. +#[tokio::test(flavor = "multi_thread")] +async fn dropped_future_buffer_lives_until_cqe() { + let Some(driver) = driver_or_skip("dropped_future_buffer_lives_until_cqe") else { + return; + }; + let (pipe_read, mut pipe_write) = os_pipe(); + + 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).await, + "read never reached in-flight state" + ); + + drop(handle); + + // No cancel was submitted and the pipe is empty: the op MUST stay in + // flight and the buffer MUST NOT be reclaimed, no matter how long the + // future has been gone. + tokio::time::sleep(Duration::from_millis(300)).await; + let snap = driver.stats(); + assert_eq!(snap.in_flight, 1, "op vanished without a CQE"); + assert_eq!(snap.orphan_reclaimed, 0, "buffer reclaimed before CQE — UAF window!"); + + // Now let the kernel complete the read; the CQE both writes into the + // driver-owned buffer (safely) and triggers reclamation. + pipe_write.write_all(&[0xAB; 512]).expect("pipe write"); + assert!( + wait_until(Duration::from_secs(2), || { + let s = driver.stats(); + s.in_flight == 0 && s.orphan_reclaimed == 1 + }) + .await, + "orphaned op was not reclaimed at CQE: {:?}", + driver.stats() + ); + + driver.shutdown(); +} + +/// Default drop path: ASYNC_CANCEL accelerates the CQE so the orphaned +/// buffer is reclaimed promptly without any data ever arriving. +#[tokio::test(flavor = "multi_thread")] +async fn async_cancel_accelerates_reclaim() { + let Some(driver) = driver_or_skip("async_cancel_accelerates_reclaim") else { + return; + }; + let (pipe_read, pipe_write) = os_pipe(); + + let handle = driver.read_current(Arc::clone(&pipe_read), 4096); + assert!( + wait_until(Duration::from_secs(2), || driver.stats().in_flight == 1).await, + "read never reached in-flight state" + ); + + drop(handle); // submits IORING_OP_ASYNC_CANCEL + + assert!( + wait_until(Duration::from_secs(2), || { + let s = driver.stats(); + s.in_flight == 0 && s.orphan_reclaimed == 1 + }) + .await, + "cancel did not reclaim the orphan: {:?}", + driver.stats() + ); + + drop(pipe_write); + driver.shutdown(); +} + +/// Volume test modeling the EC quorum pattern: many concurrent reads, half +/// the futures dropped immediately. Every op must be accounted for as either +/// delivered or orphan-reclaimed, and survivors must return correct bytes. +#[tokio::test(flavor = "multi_thread")] +async fn cancel_stress_accounts_for_every_buffer() { + let Some(driver) = driver_or_skip("cancel_stress_accounts_for_every_buffer") else { + return; + }; + const LEN: usize = 8 << 20; + const OPS: usize = 256; + const READ_LEN: usize = 64 << 10; + let content = make_content(LEN); + let (path, file) = temp_file_with(&content, "stress"); + + let mut kept = Vec::new(); + for i in 0..OPS { + let offset = (i * 97_611) % (LEN - READ_LEN); + let handle = driver.read_at(Arc::clone(&file), offset as u64, READ_LEN); + if i % 2 == 0 { + drop(handle); // dropped mid-flight or post-completion — both must be safe + } else { + kept.push((offset, handle)); + } + } + + for (offset, handle) in kept { + let got = handle.await.expect("kept read failed"); + assert_eq!(got, &content[offset..offset + READ_LEN], "mismatch at offset {offset}"); + } + + let snap = driver.shutdown(); + assert_eq!(snap.submitted, OPS as u64); + assert_eq!(snap.delivered, (OPS / 2) as u64); + assert_eq!( + snap.delivered + snap.orphan_reclaimed, + OPS as u64, + "some buffers are unaccounted for: {snap:?}" + ); + let _ = std::fs::remove_file(path); +} + +/// Shutdown with ops still blocked in flight must cancel + drain them before +/// the driver thread exits (and the ring is unmapped). +#[tokio::test(flavor = "multi_thread")] +async fn shutdown_drains_in_flight_ops() { + let Some(driver) = driver_or_skip("shutdown_drains_in_flight_ops") else { + return; + }; + let (pipe_read, pipe_write) = os_pipe(); + + let h1 = driver.read_current(Arc::clone(&pipe_read), 1024); + let h2 = driver.read_current(Arc::clone(&pipe_read), 1024); + assert!( + wait_until(Duration::from_secs(2), || driver.stats().in_flight == 2).await, + "reads never reached in-flight state" + ); + + // shutdown() cancels both, drains to in_flight == 0 (asserted inside), + // and joins the thread. The held futures then resolve with ECANCELED. + let snap = driver.shutdown(); + assert_eq!(snap.in_flight, 0); + assert_eq!(snap.delivered + snap.orphan_reclaimed, snap.submitted); + + for h in [h1, h2] { + let err = h.await.expect_err("blocked pipe read cannot have succeeded"); + assert_eq!(err.raw_os_error(), Some(libc::ECANCELED), "unexpected error: {err:?}"); + } + + drop(pipe_write); +}