From d90e354b11c51b73166d924455c129406b3d08ef Mon Sep 17 00:00:00 2001 From: houseme Date: Fri, 10 Jul 2026 00:06:15 +0800 Subject: [PATCH] chore(experiments): remove io_uring spike, migrated to rustfs/uring (#4629) The io_uring cancel-safety spike (imported in #4625) has been migrated to a dedicated repository, https://github.com/rustfs/uring (crate `rustfs-uring`), with the full per-issue commit history preserved. The standalone repo has independent CI that this workspace cannot run: it excludes the crate from the build graph, so its checks passed without ever building or testing it. rustfs/uring CI is green on fmt + clippy, a native leg that runs the suite with real io_uring and fails on any skip, and a docker leg exercising both the graceful-degradation and real-io_uring paths. Removing the in-tree copy keeps rustfs/rustfs clean; the crate will be wired in later as a git dependency behind runtime probing (rustfs/backlog#1051). 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 | 99 -- .../io-uring-cancel-spike/run-docker.sh | 78 -- .../seccomp-block-uring.json | 11 - .../io-uring-cancel-spike/src/driver.rs | 905 ------------------ experiments/io-uring-cancel-spike/src/lib.rs | 37 - .../io-uring-cancel-spike/tests/cancel.rs | 472 --------- 9 files changed, 1726 deletions(-) delete mode 100644 experiments/io-uring-cancel-spike/.gitignore delete mode 100644 experiments/io-uring-cancel-spike/Cargo.lock delete mode 100644 experiments/io-uring-cancel-spike/Cargo.toml delete mode 100644 experiments/io-uring-cancel-spike/SPIKE.md delete mode 100755 experiments/io-uring-cancel-spike/run-docker.sh delete mode 100644 experiments/io-uring-cancel-spike/seccomp-block-uring.json delete mode 100644 experiments/io-uring-cancel-spike/src/driver.rs delete mode 100644 experiments/io-uring-cancel-spike/src/lib.rs delete 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 deleted file mode 100644 index ea8c4bf7f..000000000 --- a/experiments/io-uring-cancel-spike/.gitignore +++ /dev/null @@ -1 +0,0 @@ -/target diff --git a/experiments/io-uring-cancel-spike/Cargo.lock b/experiments/io-uring-cancel-spike/Cargo.lock deleted file mode 100644 index 33e1b8191..000000000 --- a/experiments/io-uring-cancel-spike/Cargo.lock +++ /dev/null @@ -1,103 +0,0 @@ -# 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 deleted file mode 100644 index 6e65db1a5..000000000 --- a/experiments/io-uring-cancel-spike/Cargo.toml +++ /dev/null @@ -1,20 +0,0 @@ -[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 deleted file mode 100644 index 9ded1323d..000000000 --- a/experiments/io-uring-cancel-spike/SPIKE.md +++ /dev/null @@ -1,99 +0,0 @@ -# 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`)。真正的危险窗口是 **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):in-flight 上界 ≤ CQ 容量(取 SQ 深度 `entries` < `2*entries`,使 CQ overflow 结构性不可达),permit 随 pending 表项移除(CQE)释放,**绝不随 future drop 释放**——否则 quorum 大量 drop future 会让 permit 计数与驻留内存脱钩,重开内存 DoS 面。 -8. **复用缓冲内容卫生**(rustfs/backlog#1062,P3 前置):当前 spike 每 op 新分配零页 + `truncate(res)`,**无泄露**。P3 改用驱动自有对齐 slab(registered buffer)后,缓冲跨请求复用即脏内存——任何路径忘记按 `cqe.res` 截断/掩蔽(O_DIRECT 整块读再由上层切片、或错误路径把整块缓冲交还)就把上一租户请求的对象字节泄给当前请求(CWE-226)。不变量:**复用缓冲对调用方可见的字节严格 ⊆ `[0, cqe.res)`**,越界部分零化或由类型系统(返回带长度上限的 view 而非整块 slice)保证不可达。SPIKE.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 | - -## 如何运行 - -需要 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` 收割替换轮询驱动循环(注意:换收割方式后仍须周期性进 `io_uring_enter(GETEVENTS)` 冲刷 NODROP overflow list,否则 rustfs/backlog#1056 的挂起会复现)。 -- 进程级单例 ring 的生命周期管理(本 spike 每测试一个 ring);Drop 路径不得无界阻塞 tokio worker。 -- O_DIRECT 对齐 buffer(P1 的 statx 探测复用)、三条读形态接入 `LocalIoBackend`。 -- per-disk 探测缓存与运行期 errno 降级闩锁(参照 main 上 `DirectIoReadState`,`crates/ecstore/src/disk/local.rs`;运行期 errno 分类须按不变量补充里的三分类,勿复用 probe 期分类)。 -- registered buffers(P3,内容卫生见不变量 8 / rustfs/backlog#1062)/写路径(P4)完全不涉及。 - -**已在本 spike 内整改**(rustfs/backlog#1051):SQ 深度背压(不变量 7)、驱动线程 unwind 安全(不变量 6)、shutdown 有界 drain、CQ overflow/NODROP/EBUSY 处理、probe UAF、probe 文件安全创建、errno 三分类、len/offset 校验、短读 resubmit。 - -## 实测记录 - -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 deleted file mode 100755 index 41030ba6f..000000000 --- a/experiments/io-uring-cancel-spike/run-docker.sh +++ /dev/null @@ -1,78 +0,0 @@ -#!/usr/bin/env bash -# Copyright 2024 RustFS Team -# -# Licensed under the Apache License, Version 2.0 (the "License"); -# you may not use this file except in compliance with the License. -# You may obtain a copy of the License at -# -# http://www.apache.org/licenses/LICENSE-2.0 -# -# Unless required by applicable law or agreed to in writing, software -# distributed under the License is distributed on an "AS IS" BASIS, -# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. -# See the License for the specific language governing permissions and -# limitations under the License. - -# Runs the spike test suite in the two environments that matter for P2 -# (backlog#894), and — crucially — ASSERTS which path each leg actually took -# (rustfs/backlog#1067), so neither leg can silently degenerate into a vacuous -# pass: -# -# leg 1 io_uring blocked by an EXPLICIT seccomp profile (not the host -# Docker default), reproducing the #4313 restricted environment. -# Every test MUST degrade to a graceful skip; if none skip, io_uring -# was not actually blocked and the leg FAILS. -# leg 2 seccomp=unconfined — real io_uring against the host kernel. NO test -# may skip; a single skip means io_uring did not really run and the -# leg FAILS. -set -euo pipefail -cd "$(dirname "$0")" - -IMG="${SPIKE_IMAGE:-rust:1-bookworm}" -CACHE_REG=uring-spike-cargo-registry -CACHE_TARGET=uring-spike-target -SECCOMP_PROFILE="$PWD/seccomp-block-uring.json" - -# Run the suite once, capturing stdout+stderr (so the tests' "SKIP" lines, -# printed to stderr, are inspectable). -run_and_capture() { - 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 2>&1 -} - -echo "==================================================================" -echo "== leg 1: io_uring blocked by explicit seccomp profile (degradation expected)" -echo "==================================================================" -leg1_out=$(run_and_capture --security-opt seccomp="$SECCOMP_PROFILE") || { - echo "$leg1_out" - echo "leg 1 FAILED: the suite errored (a blocked probe must degrade, not fail)" - exit 1 -} -echo "$leg1_out" -if ! grep -q "SKIP " <<<"$leg1_out"; then - echo "leg 1 FAILED: expected graceful-degradation SKIPs but saw none — io_uring was NOT blocked" - exit 1 -fi - -echo "==================================================================" -echo "== leg 2: seccomp=unconfined (real io_uring)" -echo "==================================================================" -leg2_out=$(run_and_capture --security-opt seccomp=unconfined) || { - echo "$leg2_out" - echo "leg 2 FAILED: the suite errored" - exit 1 -} -echo "$leg2_out" -if grep -q "SKIP " <<<"$leg2_out"; then - echo "leg 2 FAILED: a test skipped — io_uring did not actually run here (vacuous pass)" - exit 1 -fi - -echo "both legs passed (leg 1 degraded gracefully, leg 2 ran real io_uring)" diff --git a/experiments/io-uring-cancel-spike/seccomp-block-uring.json b/experiments/io-uring-cancel-spike/seccomp-block-uring.json deleted file mode 100644 index b3d4ab784..000000000 --- a/experiments/io-uring-cancel-spike/seccomp-block-uring.json +++ /dev/null @@ -1,11 +0,0 @@ -{ - "comment": "Deterministically block io_uring for run-docker.sh leg 1 (rustfs/backlog#1067): io_uring_setup/enter/register return EPERM, so the probe fails with an expected-restriction errno and the suite must degrade to a graceful skip — reproducing the #4313 restricted environment without depending on the host Docker's default seccomp.", - "defaultAction": "SCMP_ACT_ALLOW", - "syscalls": [ - { - "names": ["io_uring_setup", "io_uring_enter", "io_uring_register"], - "action": "SCMP_ACT_ERRNO", - "errnoRet": 1 - } - ] -} diff --git a/experiments/io-uring-cancel-spike/src/driver.rs b/experiments/io-uring-cancel-spike/src/driver.rs deleted file mode 100644 index 9eca64822..000000000 --- a/experiments/io-uring-cancel-spike/src/driver.rs +++ /dev/null @@ -1,905 +0,0 @@ -// Copyright 2024 RustFS Team -// -// Licensed under the Apache License, Version 2.0 (the "License"); -// you may not use this file except in compliance with the License. -// You may obtain a copy of the License at -// -// http://www.apache.org/licenses/LICENSE-2.0 -// -// Unless required by applicable law or agreed to in writing, software -// distributed under the License is distributed on an "AS IS" BASIS, -// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. -// See the License for the specific language governing permissions and -// limitations under the License. - -use std::collections::{HashMap, VecDeque}; -use std::fs::File; -use std::io; -use std::io::Write as _; -use std::os::fd::{AsRawFd, FromRawFd}; -use std::os::unix::ffi::OsStrExt; -use std::pin::Pin; -use std::sync::atomic::{AtomicU64, Ordering}; -use std::sync::mpsc::{self, RecvTimeoutError, TryRecvError}; -use std::sync::{Arc, Condvar, Mutex}; -use std::task::{Context, Poll}; -use std::thread::JoinHandle; -use std::time::{Duration, Instant}; - -use io_uring::{IoUring, opcode, types}; - -/// Upper bound on how long shutdown waits for in-flight ops to drain before -/// leaking the ring+buffers and exiting (C4, rustfs/backlog#1055). ASYNC_CANCEL -/// cannot interrupt an in-execution regular-file read on a D-state/NFS-hung -/// disk, so drain-to-zero can be non-terminating; this bounds it. -const DRAIN_TIMEOUT: Duration = Duration::from_secs(5); -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; - -/// Kernel single-read cap: `MAX_RW_COUNT = INT_MAX & PAGE_MASK` (2 GiB − 4 KiB -/// on 4 KiB pages). io_uring's READ length field is a u32, and any request -/// above this short-reads. We reject beyond it in `submit` so a `len as u32` -/// truncation can never silently turn a huge read into a 0-byte "EOF" (C6, -/// rustfs/backlog#1057); P2 must chunk reads larger than this. -const MAX_READ_LEN: usize = 0x7fff_f000; - -/// 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 **probe-time** errno belongs to the "expected - /// restriction" class that P2 maps to permanent per-disk fallback: - /// EACCES/EPERM/ENOSYS/EINVAL/EOPNOTSUPP. Anything else is a genuine bug - /// worth surfacing. - /// - /// IMPORTANT (C7, rustfs/backlog#1059): this classification is valid ONLY - /// for a one-shot startup probe, where these errnos unambiguously mean - /// "io_uring is unusable here" (gVisor/seccomp/old kernel). Runtime - /// per-op errnos have different semantics and MUST NOT reuse this class. - /// In particular EINVAL is triple-meaning at runtime — offset > i64::MAX - /// (signed loff_t), O_DIRECT buffer/offset/len misalignment (P2 will use - /// O_DIRECT), and setup `entries` over the cap — none of which imply the - /// disk should be permanently degraded off io_uring. P2's degradation - /// contract must split errnos into three classes: - /// - /// * probe-time restriction -> degrade this disk to the std backend; - /// * runtime parameter error -> return the error to the caller (and, - /// for a suspected bug, re-verify once via std pread) — never latch; - /// * transient (EINTR/EAGAIN) -> retry, never surface. - /// - /// See `submit` for the offset guard that keeps a caller arithmetic bug - /// from ever reaching the kernel as a runtime EINVAL. - 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) - ) - } -} - -/// Submission-side backpressure (C10, rustfs/backlog#1060). -/// -/// Bounds in-flight ops so the count can never exceed CQ capacity, and — the -/// load-bearing part — releases a permit at the CQE (pending-table removal), -/// NOT at future drop. Tying a permit to the future (the natural RAII shape) -/// would let a quorum dropping many futures return permits while their orphan -/// buffers still sit in the pending table awaiting slow-disk CQEs, decoupling -/// the permit count from resident memory and reopening the memory-DoS surface. -/// P2 uses a tokio `Semaphore` with `acquire_owned`; the RELEASE POINT is what -/// matters and must stay at the CQE. -struct Backpressure { - state: Mutex, - cv: Condvar, -} - -struct BpState { - permits: usize, - shutdown: bool, -} - -impl Backpressure { - fn new(permits: usize) -> Self { - Self { - state: Mutex::new(BpState { - permits, - shutdown: false, - }), - cv: Condvar::new(), - } - } - - /// Block until a permit is free. Returns false if the driver has shut down - /// (so submit stops blocking forever once the driver is gone). - fn acquire(&self) -> bool { - let mut g = self.state.lock().expect("backpressure mutex poisoned"); - loop { - if g.shutdown { - return false; - } - if g.permits > 0 { - g.permits -= 1; - return true; - } - g = self.cv.wait(g).expect("backpressure mutex poisoned"); - } - } - - fn release(&self) { - let mut g = self.state.lock().expect("backpressure mutex poisoned"); - g.permits += 1; - self.cv.notify_one(); - } - - fn shutdown(&self) { - let mut g = self.state.lock().expect("backpressure mutex poisoned"); - g.shutdown = true; - self.cv.notify_all(); - } -} - -#[derive(Default)] -struct DriverStats { - submitted: AtomicU64, - delivered: AtomicU64, - orphan_reclaimed: AtomicU64, - in_flight: AtomicU64, - cancel_succeeded: AtomicU64, - cancel_not_found: AtomicU64, - cancel_already: AtomicU64, - cq_overflow: 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, - /// ASYNC_CANCEL CQEs that reported the target op was canceled (res == 0). - pub cancel_succeeded: u64, - /// ASYNC_CANCEL CQEs that reported the target was not found (-ENOENT): - /// the op had already completed. - pub cancel_not_found: u64, - /// ASYNC_CANCEL CQEs that reported the target was already executing and - /// could not be interrupted (-EALREADY). A rising count is the hung-disk - /// 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). - pub cq_overflow: u64, -} - -enum Msg { - Read { - id: u64, - file: Arc, - offset: u64, - len: usize, - done: oneshot::Sender>>, - }, - Cancel { - id: u64, - }, - Shutdown, -} - -/// One in-flight LOGICAL read. This struct — not the caller — owns everything -/// the kernel touches: -/// -/// - `buf`: the destination buffer. Its heap allocation must stay put until -/// the final CQE; the `Vec` itself may move (HashMap rehash) since that -/// never relocates the heap block. It is never resized or dropped before -/// the CQE handler removes this entry. -/// - `file`: keeps the fd open even if every caller-side clone is dropped, and -/// supplies the fd for short-read resubmission. Without it, dropping the -/// future could close the fd while an SQE built from that fd still sits in -/// the backlog (SQE construction → io_uring_enter window), and a recycled -/// fd number would make the kernel read the WRONG file (spike finding, with -/// the corrected mechanism per rustfs/backlog#1063). -/// - `offset`/`nread`: track a short-read resubmit loop (C9, -/// rustfs/backlog#1058). io_uring may legally short-read a regular file; -/// the driver resubmits the remainder into `buf[nread..]` until the request -/// is fully satisfied or a real EOF (res == 0) is seen, so reclamation -/// happens only at the FINAL CQE of the logical read. -struct Pending { - buf: Vec, - file: Arc, - done: Option>>>, - offset: u64, - nread: usize, -} - -/// 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, - bp: Arc, -} - -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)?; - // Require the NODROP feature (kernel >= 5.5). Without it, CQ overflow - // silently drops CQEs, stranding pending entries forever and hanging - // shutdown (C5, rustfs/backlog#1056). ENOSYS is in the expected- - // restriction class, so this degrades to the std backend cleanly. - 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)?; - - let (tx, rx) = mpsc::channel(); - let stats = Arc::new(DriverStats::default()); - let thread_stats = Arc::clone(&stats); - // Cap in-flight at the SQ depth (entries), which is < CQ capacity - // (2*entries), so CQ overflow is structurally unreachable (C5/C10). - let bp = Arc::new(Backpressure::new(entries as usize)); - let thread_bp = Arc::clone(&bp); - let handle = std::thread::Builder::new() - .name("uring-spike-driver".into()) - .spawn(move || drive(ring, rx, thread_stats, thread_bp)) - .expect("spawn driver thread"); - - Ok(Self { - tx, - handle: Some(handle), - stats, - next_id: AtomicU64::new(1), - bp, - }) - } - - /// 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(); - - // Reject an offset the kernel would answer with a runtime EINVAL that - // must NOT be mistaken for an environment restriction (C7, - // rustfs/backlog#1059). The kernel reads `off` as a signed loff_t, so - // offset > i64::MAX becomes a negative ki_pos → EINVAL. A caller - // offset-arithmetic bug has to surface as an error here, never as a - // permanent per-disk fallback. CURRENT_POSITION is the reserved - // read(2) sentinel and bypasses this check. - if offset != CURRENT_POSITION && offset > i64::MAX as u64 { - let _ = done.send(Err(io::Error::new( - io::ErrorKind::InvalidInput, - "offset exceeds i64::MAX (kernel loff_t is signed)", - ))); - return ReadHandle { - id, - rx, - tx: self.tx.clone(), - finished: false, - cancel_on_drop: false, - }; - } - - // Reject a length the kernel would short-read past MAX_RW_COUNT and - // that the SQE's u32 `len` field would silently truncate: len == 2^32 - // becomes a 0-byte read the caller decodes as a false EOF (C6, - // rustfs/backlog#1057). Failing fast here also removes the caller- - // controlled `vec![0u8; len]` capacity-overflow panic that made the - // unwind-UAF (rustfs/backlog#1054) reachable. P2 must chunk instead. - if len > MAX_READ_LEN { - let _ = done.send(Err(io::Error::new( - io::ErrorKind::InvalidInput, - "read length exceeds MAX_RW_COUNT (2 GiB - 4 KiB); caller must chunk", - ))); - return ReadHandle { - id, - rx, - tx: self.tx.clone(), - finished: false, - cancel_on_drop: false, - }; - } - - // Acquire a backpressure permit BEFORE handing the op to the driver; - // the driver releases it at the CQE (C10, rustfs/backlog#1060). This - // blocks the caller once `entries` ops are in flight, bounding both - // in-flight count (< CQ capacity) and resident buffer memory. - if !self.bp.acquire() { - let _ = done.send(Err(io::Error::other("uring driver shut down"))); - return ReadHandle { - id, - rx, - tx: self.tx.clone(), - finished: false, - cancel_on_drop: false, - }; - } - - // If the driver shut down between acquire and send, give the permit - // back — no CQE will ever release it. `done` is dropped with the - // returned Msg, which the caller observes as a driver-gone error. - if self - .tx - .send(Msg::Read { - id, - file, - offset, - len, - done, - }) - .is_err() - { - self.bp.release(); - return ReadHandle { - id, - rx, - tx: self.tx.clone(), - finished: false, - cancel_on_drop: false, - }; - } - 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), - cancel_succeeded: self.stats.cancel_succeeded.load(Ordering::SeqCst), - cancel_not_found: self.stats.cancel_not_found.load(Ordering::SeqCst), - cancel_already: self.stats.cancel_already.load(Ordering::SeqCst), - cq_overflow: self.stats.cq_overflow.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(); - // A clean drain leaves in_flight == 0. A non-zero count here means the - // bounded drain bailed out on a hung device and leaked the ring+buffers - // to stay memory-safe (C4, rustfs/backlog#1055) — a degraded but safe - // outcome, not a panic. Callers/tests that require a clean drain assert - // on the returned snapshot themselves. - if snap.in_flight != 0 { - eprintln!( - "uring-spike shutdown: {} ops still in flight (bounded-drain bailout on a hung device)", - snap.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<()> { - let pattern: Vec = (0..512u32).map(|i| (i * 7 + 13) as u8).collect(); - - // Open an anonymous probe file seeded with the pattern. File setup runs - // BEFORE any SQE, so its errors early-return safely — nothing is in flight. - let file = open_probe_file(&pattern)?; - - 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: a push failure means the kernel never accepted the SQE, so - // `buf`/`file` may be dropped safely on this early return. - if unsafe { ring.submission().push(&sqe) }.is_err() { - return Err(io::Error::other("probe: submission queue full")); - } - - // C1 (rustfs/backlog#1053): once the SQE is handed to the kernel, the read - // may be punted to io-wq and write into `buf` at ANY later point. Until its - // CQE arrives, `buf`/`file` must NOT be dropped and the ring must NOT be - // unmapped — otherwise the kernel writes into freed memory (UAF). The probe - // path has no pending-table backstop, so we must drain to the CQE here, and - // any early exit first leaks the buffer ("leak over UAF"). - let res = match drain_probe_cqe(ring) { - Ok(res) => res, - Err(e) => { - // Could not confirm the op terminated: leak `buf` (the real UAF - // hazard — the kernel may still write 512 bytes into it) and, - // defensively, `file`. Leaking one 512-byte startup-probe buffer is - // trivially cheaper than a silent heap corruption. - std::mem::forget(buf); - std::mem::forget(file); - return Err(e); - } - }; - - // The CQE has arrived: the kernel is done with `buf`, so dropping it and - // `file` below is now safe. - if res < 0 { - Err(io::Error::from_raw_os_error(-res)) - } else if res as usize != pattern.len() || buf != pattern { - Err(io::Error::other("probe: read completed but data mismatched")) - } else { - Ok(()) - } -} - -/// Open a probe file seeded with `pattern`, avoiding the symlink/TOCTOU/ -/// leftover hazards of a predictable temp path (C3, rustfs/backlog#1061). -/// -/// Primary: `O_TMPFILE` — an anonymous inode with no name at all, so there is -/// nothing for an attacker to pre-plant a symlink at, no TOCTOU window, and no -/// leftover file. Fallback (filesystems without O_TMPFILE): create in the temp -/// dir with `O_CREAT|O_EXCL|O_NOFOLLOW` + 0600 + a per-process nonce, then -/// unlink immediately so no attacker-planted symlink is followed and no named -/// file survives. -fn open_probe_file(pattern: &[u8]) -> io::Result { - let dir = std::env::temp_dir(); - let c_dir = std::ffi::CString::new(dir.as_os_str().as_bytes()).map_err(|_| io::Error::other("probe dir path has NUL"))?; - // SAFETY: `c_dir` is a valid NUL-terminated path; O_TMPFILE requires a - // directory and O_RDWR/O_WRONLY. On success we own the returned fd. - let fd = unsafe { libc::open(c_dir.as_ptr(), libc::O_TMPFILE | libc::O_RDWR | libc::O_CLOEXEC, 0o600) }; - if fd >= 0 { - let mut file = unsafe { File::from_raw_fd(fd) }; - file.write_all(pattern)?; - return Ok(file); - } - open_probe_file_exclusive(&dir, pattern) -} - -fn open_probe_file_exclusive(dir: &std::path::Path, pattern: &[u8]) -> io::Result { - static SEQ: AtomicU64 = AtomicU64::new(0); - let nonce = SEQ.fetch_add(1, Ordering::Relaxed); - let path = dir.join(format!("uring-spike-probe-{}-{}", std::process::id(), nonce)); - let c_path = std::ffi::CString::new(path.as_os_str().as_bytes()).map_err(|_| io::Error::other("probe path has NUL"))?; - // O_EXCL refuses a pre-existing file; O_NOFOLLOW refuses a symlink; 0600 is - // owner-only. SAFETY: `c_path` is a valid NUL-terminated path; on success - // we own the fd. - let fd = unsafe { - libc::open( - c_path.as_ptr(), - libc::O_CREAT | libc::O_EXCL | libc::O_NOFOLLOW | libc::O_RDWR | libc::O_CLOEXEC, - 0o600, - ) - }; - if fd < 0 { - return Err(io::Error::last_os_error()); - } - let mut file = unsafe { File::from_raw_fd(fd) }; - file.write_all(pattern)?; - // Unlink now: the fd stays valid, no named leftover remains. - // SAFETY: `c_path` is still a valid NUL-terminated path. - unsafe { - libc::unlink(c_path.as_ptr()); - } - Ok(file) -} - -/// Wait for the probe SQE's CQE and return its raw result. -/// -/// The SQE has already been pushed; this only drains it. `submit_and_wait` -/// interrupted by a signal returns EINTR — since the kernel consumed the SQE -/// atomically before the wait phase, we retry the WAIT only and never re-push -/// (C8, backlog#1059). A bounded attempt count keeps a probe that hit a hung -/// 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) { - Ok(_) => {} - // Signal interrupted the wait; the SQE is already in flight, so - // just wait again (do NOT re-push). - Err(e) if e.raw_os_error() == Some(libc::EINTR) => {} - Err(e) => return Err(e), - } - if let Some(cqe) = ring.completion().next() { - 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 -/// pending (orphan) table of in-flight buffers, and the SQE backlog. -/// -/// C2 (rustfs/backlog#1054): the "CQE is the only reclamation point" -/// invariant holds only while the driver thread does NOT unwind. On a panic, -/// Rust would drop the pending table (freeing every in-flight buffer) while -/// the kernel may still write into them → mass UAF; reversing drop order does -/// not help because io_uring teardown on ring drop is asynchronous and does -/// not wait for in-flight ops. So this type's `Drop` refuses to run field -/// destructors during an unwind: it aborts the process first, leaving the -/// ring mapped and the buffers allocated (leak over UAF). A storage read path -/// silently corrupting memory is worse than a crash. -struct DriverState { - ring: IoUring, - pending: HashMap, - backlog: VecDeque, -} - -impl Drop for DriverState { - fn drop(&mut self) { - if std::thread::panicking() { - // Abort BEFORE any field destructor runs: the ring stays mapped - // and the in-flight buffers stay allocated, so the kernel can - // never write into freed memory. - eprintln!( - "uring-spike driver thread panicked with {} ops in flight; \ - aborting to avoid UAF of in-flight buffers", - self.pending.len() - ); - std::process::abort(); - } - // Normal drop: the shutdown invariant guarantees pending/backlog are - // empty and in_flight == 0, so unmapping the ring here is safe. - } -} - -/// What to do with a pending entry after its CQE (C9, rustfs/backlog#1058). -enum ReapStep { - /// The logical read is done: remove the entry and deliver this result. - Finish(io::Result>), - /// Short read, not EOF: re-queue this SQE for the remainder; keep the entry. - Resubmit(io_uring::squeue::Entry), -} - -fn drive(ring: IoUring, rx: mpsc::Receiver, stats: Arc, bp: Arc) { - let mut state = DriverState { - ring, - pending: HashMap::new(), - backlog: VecDeque::new(), - }; - let mut shutting_down = false; - let mut drain_deadline: Option = None; - - 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 = state.pending.is_empty() && state.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"))); - // The op never became in-flight; return its permit. - bp.release(); - 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. `len as u32` - // is lossless: `submit` rejected anything > MAX_READ_LEN. - let sqe = opcode::Read::new(types::Fd(file.as_raw_fd()), buf.as_mut_ptr(), len as u32) - .offset(offset) - .build() - .user_data(id); - state.pending.insert( - id, - Pending { - buf, - file, - done: Some(done), - offset, - nread: 0, - }, - ); - stats.submitted.fetch_add(1, Ordering::SeqCst); - stats.in_flight.fetch_add(1, Ordering::SeqCst); - state.backlog.push_back(sqe); - } - Msg::Cancel { id } => { - if state.pending.contains_key(&id) { - state - .backlog - .push_back(opcode::AsyncCancel::new(id).build().user_data(id | CANCEL_BIT)); - } - } - 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)); - } - } - } - } - - // 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; - } - } - } - match state.ring.submit() { - Ok(_) => {} - 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). - } - Err(_) => { - // EINTR and friends: retry on the next loop turn. - } - } - - // 3. Reap. A Pending entry (and thus its buffer) is dropped ONLY when - // the logical read finishes; a short read is resubmitted for the - // remainder and the entry stays put (C9, rustfs/backlog#1058). - while let Some(cqe) = state.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. Record the - // three-state outcome for diagnosability (C4, - // rustfs/backlog#1055): EALREADY means the read is executing - // and cannot be interrupted, i.e. its CQE may never come on a - // hung device — the signal the bounded drain below relies on. - match cqe.result() { - 0 => stats.cancel_succeeded.fetch_add(1, Ordering::SeqCst), - r if r == -libc::ENOENT => stats.cancel_not_found.fetch_add(1, Ordering::SeqCst), - r if r == -libc::EALREADY => stats.cancel_already.fetch_add(1, Ordering::SeqCst), - _ => 0, - }; - continue; - } - let res = cqe.result(); - if !state.pending.contains_key(&ud) { - continue; - } - - // Decide the next step while borrowing the entry, then act after - // the borrow ends (finish removes it; resubmit re-queues an SQE). - 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))) - } else if res == 0 { - // Real EOF: deliver what was read so far. - p.buf.truncate(p.nread); - ReapStep::Finish(Ok(std::mem::take(&mut p.buf))) - } else { - p.nread += res as usize; - // Only POSITIONED reads (read_at, whole-range pread - // contract) resubmit a short read. CURRENT_POSITION reads - // (read_current on pipes/streams) follow read(2) semantics: - // a short read is a valid final result and must be - // delivered as-is — resubmitting would block forever - // waiting for stream data that may never come. - let is_stream = p.offset == CURRENT_POSITION; - if is_stream || p.nread >= p.buf.len() { - p.buf.truncate(p.nread); - ReapStep::Finish(Ok(std::mem::take(&mut p.buf))) - } else { - // Positioned short read, not EOF: resubmit the - // remainder into buf[nread..]. The buffer stays owned by - // the driver and in_flight is unchanged — one logical op. - let remaining = p.buf.len() - p.nread; - // SAFETY: buf[nread..] is in bounds (nread < len) and - // stays alive in the pending table until the CQE. - let ptr = unsafe { p.buf.as_mut_ptr().add(p.nread) }; - let next_off = if p.offset == CURRENT_POSITION { - CURRENT_POSITION - } else { - 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) - } - } - }; - - match step { - ReapStep::Finish(outcome) => { - // Content hygiene (C12, rustfs/backlog#1062): the delivered - // bytes are ⊆ [0, res) — buf was freshly zeroed per op and - // truncated to res. When P3 reuses a driver-owned slab - // across requests, this ⊆ [0, res) property MUST be - // preserved or a previous tenant's object bytes leak. - let mut p = state.pending.remove(&ud).expect("checked above"); - 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 final CQE and is reclaimed here. - Err(_) => stats.orphan_reclaimed.fetch_add(1, Ordering::SeqCst), - }; - stats.in_flight.fetch_sub(1, Ordering::SeqCst); - // Release the permit HERE, at the CQE (pending removed), - // not at future drop (C10, rustfs/backlog#1060). - bp.release(); - } - ReapStep::Resubmit(sqe) => state.backlog.push_back(sqe), - } - } - - // 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). - 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"); - } - - // 4. Exit when drained: the kernel no longer references any buffer, so - // dropping the ring (unmap) is safe. If a hung device keeps a CQE - // from ever arriving, bail out under a bounded deadline instead of - // blocking forever (C4, rustfs/backlog#1055). - if shutting_down { - if state.pending.is_empty() && state.backlog.is_empty() { - bp.shutdown(); // unblock any submit() waiter before we exit - return; // clean drain: DriverState drops normally, ring unmaps. - } - let deadline = *drain_deadline.get_or_insert_with(|| Instant::now() + DRAIN_TIMEOUT); - if Instant::now() >= deadline { - // A CQE may never arrive (ASYNC_CANCEL cannot interrupt an - // in-execution regular-file read on a hung disk). We must NOT - // unmap the ring or free the still-in-flight buffers — leak the - // whole state (leak over UAF) and exit so shutdown() returns. - eprintln!( - "uring-spike driver: bounded drain timed out with {} ops still in flight; \ - leaking ring + buffers to stay memory-safe", - state.pending.len() - ); - bp.shutdown(); // unblock any submit() waiter - std::mem::forget(state); - return; - } - } - - // Spike-grade pacing: production replaces this poll with eventfd + - // tokio AsyncFd reaping (backlog#894). - if !state.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 deleted file mode 100644 index 59909c5bf..000000000 --- a/experiments/io-uring-cancel-spike/src/lib.rs +++ /dev/null @@ -1,37 +0,0 @@ -// Copyright 2024 RustFS Team -// -// Licensed under the Apache License, Version 2.0 (the "License"); -// you may not use this file except in compliance with the License. -// You may obtain a copy of the License at -// -// http://www.apache.org/licenses/LICENSE-2.0 -// -// Unless required by applicable law or agreed to in writing, software -// distributed under the License is distributed on an "AS IS" BASIS, -// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. -// See the License for the specific language governing permissions and -// limitations under the License. - -//! 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 deleted file mode 100644 index 68612583b..000000000 --- a/experiments/io-uring-cancel-spike/tests/cancel.rs +++ /dev/null @@ -1,472 +0,0 @@ -// Copyright 2024 RustFS Team -// -// Licensed under the Apache License, Version 2.0 (the "License"); -// you may not use this file except in compliance with the License. -// You may obtain a copy of the License at -// -// http://www.apache.org/licenses/LICENSE-2.0 -// -// Unless required by applicable law or agreed to in writing, software -// distributed under the License is distributed on an "AS IS" BASIS, -// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. -// See the License for the specific language governing permissions and -// limitations under the License. - -//! 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::{AsRawFd, 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); - // The exact split delivered == OPS/2 was flaky (C18, rustfs/backlog#1066): - // an even-i read can finish between read_at returning and drop(handle), - // delivering to the still-live receiver and flipping the split — all while - // the ownership model behaves correctly. Only the conservation identity is - // deterministic; the kept half guarantees delivered >= OPS/2. - assert!(snap.delivered >= (OPS / 2) as u64, "kept half not all delivered: {snap:?}"); - 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); -} - -/// Invariant 2 regression (C20, rustfs/backlog#1064): the pending table — not -/// the caller — owns the fd. Deleting `Pending.file` used to leave every test -/// green because each test kept its own Arc alive; this pins it. Drop the -/// caller's Arc while the op is in flight (bare drop, no cancel) and assert the -/// fd is STILL open: only the pending table's clone keeps it alive. If that -/// field were removed, the File would close the fd and F_GETFD returns EBADF. -#[tokio::test(flavor = "multi_thread")] -async fn pending_table_owns_fd_after_caller_drop() { - let Some(driver) = driver_or_skip("pending_table_owns_fd_after_caller_drop") else { - return; - }; - let (pipe_read, mut pipe_write) = os_pipe(); - let raw_fd = pipe_read.as_raw_fd(); - - let handle = driver.read_current(Arc::clone(&pipe_read), 64).without_cancel_on_drop(); - assert!( - wait_until(Duration::from_secs(2), || driver.stats().in_flight == 1).await, - "read never reached in-flight state" - ); - - // The pending table is now the ONLY owner of the fd. - drop(handle); - drop(pipe_read); - tokio::time::sleep(Duration::from_millis(50)).await; - - // SAFETY: F_GETFD only queries the descriptor; it neither closes nor - // mutates it. - let rc = unsafe { libc::fcntl(raw_fd, libc::F_GETFD) }; - assert_ne!( - rc, - -1, - "fd was closed while an op is in flight — the pending table does not own it \ - (invariant 2 unprotected): {}", - std::io::Error::last_os_error() - ); - - // Complete the op so the driver reclaims cleanly. - pipe_write.write_all(&[0x5A; 64]).expect("pipe write"); - assert!( - wait_until(Duration::from_secs(2), || driver.stats().in_flight == 0).await, - "op was not reclaimed at CQE" - ); - driver.shutdown(); -} - -/// Memory-safety integrity (C14, rustfs/backlog#1064): while an orphaned -/// blocked-pipe read holds a driver-owned buffer in flight the whole time, many -/// delivered reads must still come back byte-exact — the kernel writing into -/// the orphan's still-owned buffer must not corrupt anything, and the orphan -/// buffer must NOT be reclaimed before its own CQE. This is a direct -/// data-integrity observation, not just a counter identity. (A driver-level -/// poison/canary leg is a P2 acceptance-matrix item; ASAN cannot see a kernel -/// write into a freed buffer, so it is not the mechanism.) -#[tokio::test(flavor = "multi_thread")] -async fn orphan_in_flight_does_not_corrupt_delivered_reads() { - let Some(driver) = driver_or_skip("orphan_in_flight_does_not_corrupt_delivered_reads") else { - return; - }; - let (pipe_read, mut pipe_write) = os_pipe(); - let orphan = 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, - "orphan never reached in-flight state" - ); - drop(orphan); // bare drop: buffer stays owned by the driver until its CQE - - const LEN: usize = 1 << 20; - let content = make_content(LEN); - let (path, file) = temp_file_with(&content, "canary"); - for i in 0..64usize { - let offset = (i * 9_973) % (LEN - 4096); - let got = driver - .read_at(Arc::clone(&file), offset as u64, 4096) - .await - .expect("read failed"); - assert_eq!(got, &content[offset..offset + 4096], "delivered read corrupted at offset {offset}"); - } - assert_eq!(driver.stats().orphan_reclaimed, 0, "orphan buffer reclaimed before its CQE"); - - pipe_write.write_all(&[0u8; 512]).expect("pipe write"); - driver.shutdown(); - let _ = std::fs::remove_file(path); -} - -/// CQ-overflow safety (C15, rustfs/backlog#1065). With backpressure capping -/// in-flight at the SQ depth (64) below CQ capacity (128), overflow is -/// structurally unreachable. Drive far more ops than CQ capacity through and -/// assert the kernel overflow counter stays 0 and every op is delivered. -#[tokio::test(flavor = "multi_thread")] -async fn no_cq_overflow_under_load() { - let Some(driver) = driver_or_skip("no_cq_overflow_under_load") else { - return; - }; - const LEN: usize = 8 << 20; - const OPS: usize = 300; // > CQ capacity (128) many times over - const READ_LEN: usize = 4096; - let content = make_content(LEN); - let (path, file) = temp_file_with(&content, "overflow"); - - let mut kept = Vec::new(); - for i in 0..OPS { - let offset = (i * 4_093) % (LEN - READ_LEN); - kept.push((offset, driver.read_at(Arc::clone(&file), offset as u64, READ_LEN))); - } - for (offset, handle) in kept { - let got = handle.await.expect("read failed"); - assert_eq!(got, &content[offset..offset + READ_LEN], "mismatch at offset {offset}"); - } - - let snap = driver.shutdown(); - assert_eq!(snap.cq_overflow, 0, "CQ overflowed under load: {snap:?}"); - assert_eq!(snap.delivered, OPS as u64); - let _ = std::fs::remove_file(path); -} - -/// Boundary reads on a regular file (C16, rustfs/backlog#1065): len==0, read at -/// EOF, a cross-EOF short read delivered to a live receiver (exercises the C9 -/// resubmit loop), and the rejected huge-len / huge-offset guards (C6/C7). -#[tokio::test(flavor = "multi_thread")] -async fn boundary_reads() { - let Some(driver) = driver_or_skip("boundary_reads") else { - return; - }; - const LEN: usize = 4096; - let content = make_content(LEN); - let (path, file) = temp_file_with(&content, "boundary"); - - // len == 0 → empty Ok. - let got = driver.read_at(Arc::clone(&file), 0, 0).await.expect("len=0 read"); - assert!(got.is_empty(), "len=0 should return empty, got {}", got.len()); - - // offset == file size → EOF → empty Ok. - let got = driver.read_at(Arc::clone(&file), LEN as u64, 128).await.expect("EOF read"); - assert!(got.is_empty(), "read at EOF should be empty, got {}", got.len()); - - // Read spanning past EOF → the available tail bytes, delivered to a live - // receiver via the resubmit-then-EOF path. - let got = driver - .read_at(Arc::clone(&file), (LEN - 10) as u64, 100) - .await - .expect("cross-EOF read"); - assert_eq!(got, &content[LEN - 10..LEN], "cross-EOF read should return the tail only"); - - // len > MAX_RW_COUNT → rejected (C6); offset > i64::MAX → rejected (C7). - let err = driver - .read_at(Arc::clone(&file), 0, (1usize << 32) + 1) - .await - .expect_err("huge len must be rejected"); - assert_eq!(err.kind(), std::io::ErrorKind::InvalidInput, "huge len error: {err:?}"); - let err = driver - .read_at(Arc::clone(&file), (i64::MAX as u64) + 1, 16) - .await - .expect_err("huge offset must be rejected"); - assert_eq!(err.kind(), std::io::ErrorKind::InvalidInput, "huge offset error: {err:?}"); - - driver.shutdown(); - let _ = std::fs::remove_file(path); -} - -/// Pipe half-close boundary (C16, rustfs/backlog#1065): an in-flight read whose -/// write end is closed observes EOF (res=0) and returns empty. -#[tokio::test(flavor = "multi_thread")] -async fn pipe_half_close_reads_eof() { - let Some(driver) = driver_or_skip("pipe_half_close_reads_eof") else { - return; - }; - let (pipe_read, pipe_write) = os_pipe(); - let handle = driver.read_current(Arc::clone(&pipe_read), 128); - assert!( - wait_until(Duration::from_secs(2), || driver.stats().in_flight == 1).await, - "read never reached in-flight state" - ); - drop(pipe_write); // close write end → blocked read observes EOF - let got = handle.await.expect("pipe EOF read"); - assert!(got.is_empty(), "closed-pipe read should be empty EOF, got {}", got.len()); - driver.shutdown(); -} - -/// Drop-without-shutdown path (C17, rustfs/backlog#1066): every other test ends -/// via explicit shutdown(), so the UringDriver `Drop` impl's live-thread branch -/// was never exercised. Drop the driver directly with ops still in flight; the -/// Drop must send Shutdown BEFORE joining (send-after-join would deadlock at -/// recv), cancel + drain the in-flight ops, and join — the held futures then -/// resolve to ECANCELED. A regression (join-first, or an unbounded hang) makes -/// this test hang. -#[tokio::test(flavor = "multi_thread")] -async fn drop_without_shutdown_drains_and_cancels() { - let Some(driver) = driver_or_skip("drop_without_shutdown_drains_and_cancels") 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" - ); - - // No shutdown() — exercise Drop directly. - drop(driver); - - 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); -}