mirror of
https://github.com/rustfs/rustfs.git
synced 2026-07-26 16:28:15 +00:00
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 <heihutu@gmail.com>
This commit is contained in:
houseme
@@ -0,0 +1 @@
|
||||
/target
|
||||
+103
@@ -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"
|
||||
@@ -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"] }
|
||||
@@ -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<File> + 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<File>`)。只保 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 主体重启时可直接沿用此所有权模型。
|
||||
+36
@@ -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"
|
||||
@@ -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<File>,
|
||||
offset: u64,
|
||||
len: usize,
|
||||
done: oneshot::Sender<io::Result<Vec<u8>>>,
|
||||
},
|
||||
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<u8>,
|
||||
_file: Arc<File>,
|
||||
done: Option<oneshot::Sender<io::Result<Vec<u8>>>>,
|
||||
}
|
||||
|
||||
/// 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<io::Result<Vec<u8>>>,
|
||||
tx: mpsc::Sender<Msg>,
|
||||
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<Vec<u8>>;
|
||||
|
||||
fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
|
||||
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<Msg>,
|
||||
handle: Option<JoinHandle<()>>,
|
||||
stats: Arc<DriverStats>,
|
||||
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<Self, ProbeFailure> {
|
||||
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<File>, 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<File>, len: usize) -> ReadHandle {
|
||||
self.submit(file, CURRENT_POSITION, len)
|
||||
}
|
||||
|
||||
fn submit(&self, file: Arc<File>, 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<u8> = (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<Msg>, stats: Arc<DriverStats>) {
|
||||
let mut pending: HashMap<u64, Pending> = HashMap::new();
|
||||
let mut backlog: VecDeque<io_uring::squeue::Entry> = 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));
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -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};
|
||||
@@ -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<UringDriver> {
|
||||
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<u8> {
|
||||
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<File>) {
|
||||
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>, 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);
|
||||
}
|
||||
Reference in New Issue
Block a user