From b0e0549b48f4bc00f4e5cbb37f64df82a8961e96 Mon Sep 17 00:00:00 2001 From: houseme Date: Fri, 10 Jul 2026 23:45:14 +0800 Subject: [PATCH 1/2] docs: refresh README for shipped features and add a CHANGELOG MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The README described eventfd reaping, O_DIRECT, and LocalIoBackend integration as unlanded and listed streaming reads on the roadmap; all of that has since merged or been closed by measurement. Bring it in line with the actual public API and record the honest performance picture. README changes: - status now reflects that the read path is wired into rustfs/rustfs behind a runtime probe and off by default (RUSTFS_IO_URING_READ_ENABLE), and how to pin the git dependency; - document sharded rings (probe_and_start_sharded) and native O_DIRECT (read_at_direct) with runnable examples matching the current signatures; - a "when this crate helps — and when it does not" section reporting the measured results, including where io_uring loses (single sequential stream, low concurrency) and the roughly-neutral end-to-end S3 GET, plus the two benchmarking traps this repo hit (a 76x regression mistaken for a win, and page-cache-hit microbenchmarks not transferring end-to-end); - roadmap trimmed to what is actually open (write path, register_files, SQPOLL) with the closed-by-measurement decisions moved to the CHANGELOG. Add CHANGELOG.md (Keep a Changelog format): every landed change #1..#6 with the commit that carried it, the pre-history of the audited spike, and a "decisions recorded, not implemented" section so the NO-GO items are not silently re-opened. Verified README claims against source: the six documented signatures match src/driver.rs, ProbeFailure/StatsSnapshot/UringDriver are the lib exports, and the O_DIRECT example's alignment passes the driver's validation. Co-Authored-By: heihutu --- CHANGELOG.md | 153 +++++++++++++++++++++++++++++++++++++++++++++++++++ README.md | 125 ++++++++++++++++++++++++++++++++++------- 2 files changed, 258 insertions(+), 20 deletions(-) create mode 100644 CHANGELOG.md diff --git a/CHANGELOG.md b/CHANGELOG.md new file mode 100644 index 0000000..f9516d8 --- /dev/null +++ b/CHANGELOG.md @@ -0,0 +1,153 @@ + + +# Changelog + +All notable changes to `rustfs-uring` are documented here. + +The format follows [Keep a Changelog](https://keepachangelog.com/en/1.1.0/), and the project +aims to follow [Semantic Versioning](https://semver.org/spec/v2.0.0.html). + +The crate is **not published to crates.io** (`publish = false`); consumers pin a git revision. +Because of that, entries are grouped under `[Unreleased]` until the first published release, +and each one names the commit that landed it so a pin can be matched to a feature. + +## [Unreleased] + +Everything below is in `main` and reachable by pinning the listed revision. + +### Added + +- **Sharded rings** — `UringDriver::probe_and_start_sharded(entries, shards)` runs `shards` + independent rings, each with its own driver thread, pending table, backpressure semaphore, + and eventfd. `probe_and_start(entries)` is unchanged and equals `..._sharded(entries, 1)`. + ([#6], `719b245`) + + A buffered read that hits the page cache completes *inline* inside `io_uring_enter`, so the + thread driving a ring performs that read's `memcpy`. A single-ring driver is therefore capped + at one core's memory bandwidth. Measured on a 16-core host: one thread pinned at 100% CPU, + throughput flat at ~5 GB/s regardless of read size, while a blocking-pool baseline reached + 50 GB/s. Independent rings scaled near-linearly (1/2/4 rings → 4890/8969/15806 MB/s), which is + what identified the driver thread as the ceiling. With 8 shards, 1 MiB reads went from + 4911 MB/s to 47361 MB/s, and 64 KiB reads at concurrency 32 from 124k to 345k IOPS — while + *keeping* io_uring's tail-latency advantage rather than trading it away. + + In-flight ops are capped at `entries` **per shard**, so the invariant that makes CQ overflow + structurally unreachable holds per ring; the driver admits up to `shards * entries` + concurrent reads. Every cancel-safety invariant holds per shard: a `ReadHandle` carries the + `tx`/`wake` of the shard that accepted it, so a cancel or a deferred submission necessarily + routes back to the ring whose pending table holds the op. + +- **Native aligned `O_DIRECT` positioned read** — `UringDriver::read_at_direct(file, offset, len, align)`. + The caller opens the fd with `O_DIRECT` and passes the device's logical block size; `offset` + and `len` need **not** be aligned. The driver reads a block-aligned superset range into a + block-aligned buffer and hands back exactly `[offset, offset + len)`. ([#3], `f577ba0`) + + Alignment padding, the bytes preceding the range, and the block-aligned tail never escape — a + `BitrotReader` expecting an exact shard length would flag padded output as corruption. The + buffer remains a `Vec`, over-allocated by `align - 1` with the read region starting at the + first aligned byte inside the allocation, so the cancel-safety ownership model is untouched. + Buffered reads are the degenerate `align == 1` case of the same geometry. + +- **Benchmark harnesses** (`examples/` + runner scripts), both hardened so they cannot pass + vacuously: files are created only when absent (`create_new` + `O_NOFOLLOW`, never truncating a + caller-supplied path even though the sweeps run as root), content is offset-addressable, and + `BENCH_VERIFY=1` checks every delivered byte at its claimed offset. + + - `streaming_bench` + `bench-streaming.sh` — sequential whole-file read across + `std_buffered` / `std_odirect` / pipelined `uring_read_at` / `uring_read_at_direct`, in warm + and cold page-cache legs. ([#4], `4299b92`) + - `concurrent_pread_bench` + `bench-concurrent-pread.sh` — many concurrent positioned reads on + one disk, the shape erasure-coded shard reads actually serve. Four strategies isolate the + cost of the per-read `open` and of the `spawn_blocking` hop. ([#5], `d03977a`) + +### Changed + +- **Async backpressure** — the driver's synchronous backpressure gate became a tokio + `Semaphore`. `submit` takes a permit with `try_acquire_owned()` on the fast path and, when + saturated, hands the acquire future to the `ReadHandle` to await on its first poll instead of + parking the calling runtime worker. ([#2], `2b7cae4`) + + The permit is stored in the pending entry, so it is released exactly when that entry is + dropped at the final CQE — never at future drop. Public API is unchanged: `read_at`'s + signature and its "in flight the moment it returns" behaviour hold on the fast path. + +- **eventfd-driven reaping** — the 200 µs busy-poll loop was replaced by a `poll(2)` on two + eventfds: one registered with the ring so the kernel signals every CQE, one signalled by + `submit`/shutdown so a new message wakes the loop immediately. A bounded heartbeat still runs + a loop turn so the drain deadline is checked and the NODROP overflow list is flushed. + ([#1], `ea6b920`) + +### Fixed + +- `aligned_geometry` rejects an `align` above the kernel read cap, foreclosing the + `align_offset == usize::MAX` path that would have made the driver's pointer arithmetic + undefined, and the `region_len + align - 1` allocation overflow. The driver's aligned-buffer + setup replaced a release-mode-only `debug_assert!` with a checked allocation size and a + runtime guard: an unsatisfiable alignment now fails the read instead of doing UB pointer + arithmetic. ([#3], `f577ba0`) +- `deliver()` skips the full-buffer `copy_within` on the buffered path (`align == 1`, where the + logical range already starts at byte 0), so a buffered read no longer pays a per-read + `memmove`. ([#3], `f577ba0`) +- The fast-path `submit` now surfaces an explicit driver-gone error through the result channel + when the driver has exited, instead of letting the caller infer one from a dropped oneshot. + ([#2], `2b7cae4`) + +## Pre-history + +Before this repository existed, the code was the Spike 0 cancel-safety prototype in +`rustfs/rustfs` under `experiments/`. It was audited line by line +([rustfs/backlog#1051]) and every confirmed finding was remediated there. The library was then +promoted into this repository with its history preserved (`60993c9`), given a real-io_uring CI +leg that the main workspace cannot run (`d5b28fc`), and had its crate metadata and README +polished (`39018c0`). + +The cancel-safety invariants that survived that audit — the driver's pending table owning the +buffer and fd from SQE submission until the CQE, drop-abandons-result-only, bounded shutdown +drain, abort-before-free on a driver panic — are the ones every change above is measured +against. See [`docs/DESIGN.md`](docs/DESIGN.md). + +## Decisions recorded, not implemented + +These were on the roadmap and were **closed by measurement or by design conflict** rather than +built. They are listed so nobody re-opens them without new evidence. + +- **Three read shapes / streaming reads through io_uring** — **NO-GO** ([rustfs/backlog#1144]). + io_uring's lever on a *sequential* stream is pipelining; kernel readahead already has it. Cold + reads are device-bound, so io_uring at best ties; on a warm page cache it *loses badly* + (11–41% of a buffered read) because a single sequential stream exploits neither batched + submission nor the absence of a blocking thread. Streaming reads stay on the std backend. +- **`AsyncFd` reaping without a driver thread** — **not done**. `Drop` cannot `await`, so the + bounded shutdown drain would have to break the public API. The dedicated driver thread plus + eventfd reaping already removed the busy-poll, which was the actual cost. +- **Process-wide singleton ring** — **redefined**. A singleton conflicts with per-disk + isolation: a stalled disk must not starve another disk's ring. Rings stay per-disk, and + sharding ([#6]) scales them within a disk instead. +- **Registered buffers (`register_buffers`)** — **deprioritized**. It conflicts with the + `Vec` ownership model the cancel-safety proof rests on, and end-to-end profiling showed + the S3 GET bottleneck is userspace copies, not the disk read + ([rustfs/backlog#1159]). See "When this crate helps" in the README. + +[#1]: https://github.com/rustfs/uring/pull/1 +[#2]: https://github.com/rustfs/uring/pull/2 +[#3]: https://github.com/rustfs/uring/pull/3 +[#4]: https://github.com/rustfs/uring/pull/4 +[#5]: https://github.com/rustfs/uring/pull/5 +[#6]: https://github.com/rustfs/uring/pull/6 +[rustfs/backlog#1051]: https://github.com/rustfs/backlog/issues/1051 +[rustfs/backlog#1144]: https://github.com/rustfs/backlog/issues/1144 +[rustfs/backlog#1159]: https://github.com/rustfs/backlog/issues/1159 +[Unreleased]: https://github.com/rustfs/uring/commits/main diff --git a/README.md b/README.md index 03f4af3..da0adf8 100644 --- a/README.md +++ b/README.md @@ -1,20 +1,25 @@ # rustfs-uring [![CI](https://github.com/rustfs/uring/actions/workflows/ci.yml/badge.svg)](https://github.com/rustfs/uring/actions/workflows/ci.yml) -[![crates](https://img.shields.io/crates/v/rustfs-uring.svg)](https://crates.io/crates/rustfs-uring) -[![license](https://img.shields.io/badge/License-Apache%202.0-blue.svg)](https://github.com/rustfs/uring/blob/master/LICENSE) -[![docs.rs](https://docs.rs/rustfs-uring/badge.svg)](https://docs.rs/rustfs-uring/) +[![license](https://img.shields.io/badge/License-Apache%202.0-blue.svg)](https://github.com/rustfs/uring/blob/main/LICENSE) Cancel-safe async `io_uring` read backend for [RustFS](https://github.com/rustfs/rustfs) storage. -This crate is the io_uring integration that RustFS's P2 read path is built on. It started as the Spike 0 cancel-safety +This crate is the io_uring integration that RustFS's read path is built on. It started as the Spike 0 cancel-safety prototype ([rustfs/backlog#894](https://github.com/rustfs/backlog/issues/894)) and was hardened per the [#1048/#1051 audit](https://github.com/rustfs/backlog/issues/1051). It lives in its own repository so it can be verified in isolation — with a real io_uring CI leg that the main `rustfs/rustfs` workspace cannot run — before being wired into the storage layer. -> **Status:** read path only, Linux only. Not yet published to crates.io — depend on it via git until the P2 work (write -> path, eventfd/`AsyncFd` reaping, O_DIRECT, `LocalIoBackend` integration) lands. See [`docs/DESIGN.md`](docs/DESIGN.md). +> **Status:** read path only, Linux only. **Not published to crates.io** (`publish = false`) — depend on it by pinning a +> git revision. The read path is wired into `rustfs/rustfs` behind a runtime probe and is **off by default** +> (`RUSTFS_IO_URING_READ_ENABLE`). See [`CHANGELOG.md`](CHANGELOG.md) for what has landed and +> [`docs/DESIGN.md`](docs/DESIGN.md) for the invariants. + +```toml +[target.'cfg(target_os = "linux")'.dependencies] +rustfs-uring = { git = "https://github.com/rustfs/uring", rev = "" } +``` ## The ownership model it enforces @@ -31,8 +36,10 @@ proves and enforces the invariants any production io_uring integration must foll at the CQE, not at future drop; short reads on positioned reads are resubmitted to satisfy the whole-range contract; the probe file is opened via `O_TMPFILE`. -The full invariant list, the corrected fd-reuse mechanism, and the P2 design constraints are in [ -`docs/DESIGN.md`](docs/DESIGN.md). +Each invariant holds **per shard** (see below), because a shard is an independent instance of the same driver. + +The full invariant list, the corrected fd-reuse mechanism, and the design constraints are in +[`docs/DESIGN.md`](docs/DESIGN.md). ## Usage @@ -42,25 +49,81 @@ use std::sync::Arc; use rustfs_uring::UringDriver; # async fn demo() -> std::io::Result<()> { - // Probe a real IORING_OP_READ before accepting work; a restricted environment - // (seccomp/gVisor/old kernel) returns a ProbeFailure you degrade to the std - // backend on. + // Probe a real IORING_OP_READ before accepting work. A restricted environment + // (seccomp/gVisor/old kernel) returns a ProbeFailure whose + // `is_expected_restriction()` tells you to degrade to the std backend quietly. let driver = UringDriver::probe_and_start(64).expect("io_uring available"); let file = Arc::new(File::open("/data/object")?); - // Positioned read (pread semantics, whole-range). + // Positioned read (pread semantics, whole-range: short reads are resubmitted). let bytes = driver.read_at(Arc::clone(&file), 0, 65536).await?; // Dropping the returned future before it completes is safe — the driver owns // the buffer until the CQE. - driver.shutdown(); + let snapshot = driver.shutdown(); + assert_eq!(snapshot.delivered + snapshot.orphan_reclaimed, snapshot.submitted); # Ok(()) # } ``` +### Sharded rings + +A buffered read that hits the page cache completes *inline* inside `io_uring_enter`, so the thread driving a ring +performs that read's `memcpy`. One ring is therefore capped at a single core's memory bandwidth (~5 GB/s measured). +Give a disk several rings when its reads hit the cache: + +```rust +# use rustfs_uring::UringDriver; +// Four independent rings, each with `entries` SQ slots and its own driver thread. +// In-flight is capped per shard, so the driver admits up to `shards * entries` reads. +let driver = UringDriver::probe_and_start_sharded(64, 4)?; +# Ok::<_, rustfs_uring::ProbeFailure>(()) +``` + +`probe_and_start(entries)` is exactly `probe_and_start_sharded(entries, 1)`, so nobody grows threads by upgrading. +Rings stay per-disk: a stalled disk cannot starve another disk's rings. + +### `O_DIRECT` + +Open the fd with `O_DIRECT`, pass the device's logical block size, and let the driver do the alignment. `offset` and +`len` need **not** be aligned — it reads a block-aligned superset into a block-aligned buffer and hands back exactly +the range you asked for. Padding, the bytes before the range, and the block-aligned tail never escape. + +```rust +# use rustfs_uring::UringDriver; +# use std::{fs::File, sync::Arc}; +# async fn demo(driver: &UringDriver, file: Arc) -> std::io::Result<()> { +// `file` was opened with O_DIRECT; 4096 is the probed logical block size. +let bytes = driver.read_at_direct(file, 8_191, 100, 4096).await?; +assert_eq!(bytes.len(), 100); +# Ok(()) +# } +``` + +## When this crate helps — and when it does not + +These numbers come from the harnesses in this repository and from end-to-end profiling of RustFS +([rustfs/backlog#1159](https://github.com/rustfs/backlog/issues/1159)). They are reported as measured, including the +cases where io_uring loses. + +| workload | result | +| --- | --- | +| **Many concurrent positioned reads on one disk** (erasure-coded shard reads) | **Where it wins.** With sharded rings and a cached fd: 64 KiB at concurrency 128 → 361k IOPS vs 125k for a blocking-pool baseline, and p999 3.0 ms vs 13.5 ms. | +| **A single sequential stream** | **It loses.** Kernel readahead already does what pipelining would buy. Cold reads are device-bound; on a warm page cache io_uring reaches only 11–41% of a buffered read. Streaming reads should stay on the std backend. | +| **One read at a time (low concurrency)** | **It loses.** Per-op submission overhead exceeds a page-cache `memcpy`. | +| **End-to-end S3 GET** | **Roughly neutral today (−7% … +4%).** The disk read is not the bottleneck: a cached 1 MiB GET spends ~25% of CPU in `memcpy` and ~10% in `memset`, and 0% on device reads. Optimising the read path further only pays once those copies are gone. | + +Two traps this crate's own benchmarking fell into, documented so others do not repeat them: + +- A `76×` apparent speedup turned out to be a **behaviour regression**, not a win: the io_uring path had silently + stopped honouring RustFS's `fadvise(DONTNEED)` page-cache reclaim policy, so one leg served everything from cache + while the other read the device. Always check `disk_read` and page-cache deltas, not just throughput. +- Microbenchmarks of the read path measured a page-cache-hit regime that production *deliberately avoids* for large + reads. Isolated-path gains do not transfer end-to-end for free. + ## Testing This is a Linux-only crate; on a non-Linux host `cargo check` only builds the empty stub. @@ -77,15 +140,37 @@ cargo test -- --nocapture --test-threads=1 ``` The harness fails on either a non-degrading leg 1 or a vacuous-pass leg 2, so a skipped suite can never masquerade as -real coverage. +real coverage. The 15 acceptance tests are the cancel-safety contract: buffer conservation under a mixed +drop/keep stress across shards, an orphaned op reclaimed only at its CQE, bounded shutdown drain, `O_DIRECT` returning +exact unaligned ranges, and backpressure deferring rather than blocking a runtime worker. + +## Benchmarks + +Both harnesses refuse to overwrite or follow a symlink at a caller-supplied path (the sweeps run as root), fill their +files with an offset-addressable pattern, and check every delivered byte against it under `BENCH_VERIFY=1` — throughput +alone cannot tell a correct strategy from one reading the wrong offsets. + +```bash +# Sequential whole-file read: buffered vs O_DIRECT vs pipelined io_uring, warm and cold cache. +./bench-streaming.sh + +# Many concurrent positioned reads on one disk — the shape shard reads actually serve. +# Isolates the cost of the per-read open and of the spawn_blocking hop. +./bench-concurrent-pread.sh +``` + +## Roadmap -## Roadmap (P2) +- **Write path.** Untouched today; PUT still goes through the blocking pool, and profiling suggests the win there may + exceed the read path's. +- **`register_files`.** Would remove the per-op fd lookup. Lower value now that the consumer caches descriptors. +- **`SQPOLL`.** Eliminates `io_uring_enter` under sustained load, at the cost of a kernel polling thread per ring — + which multiplies by shards and by disks. Only for high-end deployments. -- eventfd + tokio `AsyncFd` reaping to replace the poll loop (must still flush the NODROP overflow list). -- O_DIRECT aligned buffers and the three read shapes. -- `LocalIoBackend` integration in `rustfs/rustfs` behind runtime probing. -- Per-disk probe cache and runtime errno degradation latch. -- Registered buffers (P3) with the content-hygiene invariant, and the write path (P4). +Closed by measurement, not built — see [`CHANGELOG.md`](CHANGELOG.md#decisions-recorded-not-implemented): streaming +reads through io_uring (NO-GO), `AsyncFd` reaping without a driver thread (would break the public API), a process-wide +singleton ring (conflicts with per-disk isolation), and registered buffers (conflicts with the `Vec` ownership +model, and the bottleneck is elsewhere). ## License From 4ae3718822e5e97d28c76900fac540f769829756 Mon Sep 17 00:00:00 2001 From: houseme Date: Sat, 11 Jul 2026 01:14:46 +0800 Subject: [PATCH 2/2] fix: gate Linux-only benchmark examples --- examples/concurrent_pread_bench.rs | 431 +--------------------- examples/concurrent_pread_bench/linux.rs | 439 ++++++++++++++++++++++ examples/streaming_bench.rs | 434 +--------------------- examples/streaming_bench/linux.rs | 442 +++++++++++++++++++++++ 4 files changed, 901 insertions(+), 845 deletions(-) create mode 100644 examples/concurrent_pread_bench/linux.rs create mode 100644 examples/streaming_bench/linux.rs diff --git a/examples/concurrent_pread_bench.rs b/examples/concurrent_pread_bench.rs index 4502af2..6ca009e 100644 --- a/examples/concurrent_pread_bench.rs +++ b/examples/concurrent_pread_bench.rs @@ -12,428 +12,17 @@ // See the License for the specific language governing permissions and // limitations under the License. -//! Concurrent positioned-read benchmark — the shape ecstore actually serves. -//! -//! `streaming_bench` measured a single sequential stream and (correctly) found -//! io_uring loses to kernel readahead. That is not io_uring's lever. Its levers -//! are (a) batching many independent IOs into one `io_uring_enter`, and (b) -//! keeping the IO off a blocking thread. Both only appear when *many concurrent -//! positioned reads* hit one disk — exactly what erasure-coded shard reads -//! (`pread_bytes`) do under load. -//! -//! Four strategies isolate where the time actually goes: -//! -//! | strategy | models | -//! |---------------------|-----------------------------------------------------------| -//! | `std_open_pread` | today's StdBackend: spawn_blocking{open; pread} | -//! | `std_cached_pread` | spawn_blocking{pread} on a pre-opened fd | -//! | `uring_open_read` | today's UringBackend: spawn_blocking{open;stat} + read_at | -//! | `uring_cached_read` | pre-opened fd + read_at, NO spawn_blocking | -//! -//! `std_open_pread` vs `uring_open_read` answers "is today's io_uring wiring -//! worth anything". `*_cached_*` vs `*_open_*` prices the per-read open. -//! `uring_cached_read` is the ceiling an fd cache + registered files would buy. -//! -//! Usage: -//! concurrent_pread_bench -//! -//! The file is created (filled + fsync'd) only if it does not exist. An existing -//! path is never truncated, overwritten, or followed through a symlink, so a -//! mistyped path cannot destroy data even though the sweep runs as root. -//! -//! Correctness: IOPS alone cannot tell a correct strategy from one that reads -//! the wrong offsets, so the file is filled with an offset-addressable pattern -//! (`byte[i] = splitmix64(i / 8)[i % 8]`). Setting `BENCH_VERIFY=1` checks every -//! byte each strategy delivers against that pattern at its claimed offset. -//! Verification is skipped entirely when unset, so it never perturbs a timed -//! run; a `BENCH_VERIFY=1` run is a correctness check, not a measurement. +#[cfg(target_os = "linux")] +#[path = "concurrent_pread_bench/linux.rs"] +mod linux; -use std::fs::{File, OpenOptions}; -use std::io::{self, Write}; -use std::os::unix::fs::{FileExt, OpenOptionsExt}; -use std::process::ExitCode; -use std::sync::Arc; -use std::time::{Duration, Instant}; - -use rustfs_uring::UringDriver; - -/// Keeps `(concurrency * 2).next_power_of_two()` ring entries under the -/// kernel's 32768-entry limit and the arithmetic far from overflow. -const MAX_CONCURRENCY: usize = 4096; -const MAX_READ_SIZE: usize = 1 << 30; - -#[derive(Clone, Copy, PartialEq, Eq)] -enum Strategy { - StdOpenPread, - StdCachedPread, - UringOpenRead, - UringCachedRead, -} - -impl Strategy { - fn parse(s: &str) -> Result { - match s { - "std_open_pread" => Ok(Self::StdOpenPread), - "std_cached_pread" => Ok(Self::StdCachedPread), - "uring_open_read" => Ok(Self::UringOpenRead), - "uring_cached_read" => Ok(Self::UringCachedRead), - other => Err(format!( - "unknown strategy {other:?}; expected one of std_open_pread, std_cached_pread, uring_open_read, uring_cached_read" - )), - } - } - - fn name(self) -> &'static str { - match self { - Self::StdOpenPread => "std_open_pread", - Self::StdCachedPread => "std_cached_pread", - Self::UringOpenRead => "uring_open_read", - Self::UringCachedRead => "uring_cached_read", - } - } - - /// Strategies that reuse one pre-opened fd across every read. - fn uses_cached_fd(self) -> bool { - matches!(self, Self::StdCachedPread | Self::UringCachedRead) - } - - fn uses_uring(self) -> bool { - matches!(self, Self::UringOpenRead | Self::UringCachedRead) - } -} - -#[derive(Clone)] -struct Config { - strategy: Strategy, - file: String, - file_size: u64, - read_size: usize, - concurrency: usize, - total_ops: usize, - shards: usize, - verify: bool, -} - -fn parse_args() -> Result { - let a: Vec = std::env::args().collect(); - if a.len() != 7 && a.len() != 8 { - return Err( - "usage: concurrent_pread_bench [shards]" - .to_string(), - ); - } - let parse = |name: &str, raw: &str| -> Result { - raw.parse::() - .map_err(|_| format!("{name}: {raw:?} is not a non-negative integer")) - }; - let cfg = Config { - strategy: Strategy::parse(&a[1])?, - file: a[2].clone(), - file_size: parse("file_size", &a[3])?, - read_size: parse("read_size", &a[4])? as usize, - concurrency: parse("concurrency", &a[5])? as usize, - total_ops: parse("total_ops", &a[6])? as usize, - shards: if a.len() == 8 { parse("shards", &a[7])? as usize } else { 1 }, - verify: matches!(std::env::var("BENCH_VERIFY").as_deref(), Ok("1") | Ok("true")), - }; - validate(&cfg)?; - Ok(cfg) -} - -/// Reject geometry the strategies cannot honour, before any I/O runs. -fn validate(cfg: &Config) -> Result<(), String> { - if cfg.read_size == 0 || cfg.read_size > MAX_READ_SIZE { - return Err(format!("read_size must be in 1..={MAX_READ_SIZE}, got {}", cfg.read_size)); - } - if cfg.file_size > i64::MAX as u64 { - return Err("file_size exceeds i64::MAX (the kernel's signed loff_t)".to_string()); - } - // Offsets are drawn from `file_size - read_size` rounded down to 4 KiB, so - // the file must hold at least one whole read plus a block to pick from. - if cfg.file_size < cfg.read_size as u64 + 4096 { - return Err(format!( - "file_size ({}) must exceed read_size ({}) by at least 4096 bytes", - cfg.file_size, cfg.read_size - )); - } - if cfg.concurrency == 0 || cfg.concurrency > MAX_CONCURRENCY { - return Err(format!("concurrency must be in 1..={MAX_CONCURRENCY}, got {}", cfg.concurrency)); - } - if cfg.total_ops == 0 { - return Err("total_ops must be > 0".to_string()); - } - Ok(()) -} - -// --------------------------------------------------------------------------- -// Offset-addressable file content (shared shape with streaming_bench) -// --------------------------------------------------------------------------- - -fn splitmix64(seed: u64) -> u64 { - let mut z = seed.wrapping_add(0x9e37_79b9_7f4a_7c15); - z = (z ^ (z >> 30)).wrapping_mul(0xbf58_476d_1ce4_e5b9); - z = (z ^ (z >> 27)).wrapping_mul(0x94d0_49bb_1331_11eb); - z ^ (z >> 31) -} - -/// The expected 8-byte little-endian word covering byte offsets -/// `idx * 8 .. idx * 8 + 8`. Content is a pure function of absolute offset, so a -/// read at any offset can be checked without reading anything before it. -fn expected_word(idx: u64) -> [u8; 8] { - splitmix64(idx).to_le_bytes() -} - -/// Check that `data` is exactly what the file holds at `offset`. Catches a -/// strategy that reads the right *number* of bytes from the wrong offsets — a -/// length-only assertion cannot. -fn verify_range(offset: u64, data: &[u8]) -> Result<(), String> { - let (mut cached_idx, mut word) = (u64::MAX, [0u8; 8]); - for (i, &got) in data.iter().enumerate() { - let pos = offset + i as u64; - let idx = pos / 8; - if idx != cached_idx { - cached_idx = idx; - word = expected_word(idx); - } - let want = word[(pos % 8) as usize]; - if got != want { - return Err(format!("content mismatch at byte {pos}: got {got:#04x}, want {want:#04x}")); - } - } - Ok(()) -} - -/// Create the bench file if missing. An existing path is left untouched: never -/// truncated, never followed through a symlink (`create_new` implies `O_EXCL`; -/// `symlink_metadata` does not resolve one). A wrong-size or non-regular path is -/// an error rather than an overwrite, because the sweep runs as root and the -/// path is caller-supplied. -fn ensure_file(path: &str, size: u64) -> Result<(), String> { - match std::fs::symlink_metadata(path) { - Ok(meta) if !meta.file_type().is_file() => Err(format!( - "{path} exists and is not a regular file ({:?}); refusing to touch it", - meta.file_type() - )), - Ok(meta) if meta.len() != size => Err(format!( - "{path} exists with {} bytes but {size} were requested; refusing to truncate it — delete it or pick another path", - meta.len() - )), - Ok(_) => Ok(()), - Err(e) if e.kind() == io::ErrorKind::NotFound => create_file(path, size).map_err(|e| { - // A half-written file would silently become the "wrong size" error - // on the next run; leave nothing behind. - let _ = std::fs::remove_file(path); - format!("create bench file {path}: {e}") - }), - Err(e) => Err(format!("stat {path}: {e}")), - } -} - -fn create_file(path: &str, size: u64) -> io::Result<()> { - let mut f = OpenOptions::new() - .write(true) - .create_new(true) - .custom_flags(libc::O_NOFOLLOW | libc::O_CLOEXEC) - .open(path)?; - let mut buf = vec![0u8; 1 << 20]; - let mut written = 0u64; - while written < size { - // `written` is always a multiple of 8 (the buffer is), so word indices - // line up with absolute offsets. - let base = written / 8; - for (w, word) in buf.chunks_exact_mut(8).enumerate() { - word.copy_from_slice(&expected_word(base + w as u64)); - } - let n = ((size - written) as usize).min(buf.len()); - f.write_all(&buf[..n])?; - written += n as u64; - } - f.sync_all() -} - -/// O_NOFOLLOW closes the window between `ensure_file`'s stat and this open. -fn open_read(path: &str) -> io::Result { - OpenOptions::new() - .read(true) - .custom_flags(libc::O_NOFOLLOW | libc::O_CLOEXEC) - .open(path) -} - -/// Deterministic block-aligned offsets, so every strategy reads the exact same -/// blocks and the comparison is not confounded by a different access pattern. -fn offsets(cfg: &Config) -> Vec { - let blocks = (cfg.file_size - cfg.read_size as u64) / 4096; - let mut state = 0x2545_f491_4f6c_dd1du64; - (0..cfg.total_ops) - .map(|_| { - state = state.wrapping_mul(6364136223846793005).wrapping_add(1442695040888963407); - ((state >> 33) % blocks) * 4096 - }) - .collect() -} - -fn percentile(sorted: &[Duration], p: f64) -> u128 { - if sorted.is_empty() { - return 0; - } - let idx = ((sorted.len() as f64 - 1.0) * p).round() as usize; - sorted[idx].as_micros() -} - -async fn run(cfg: &Config) -> Result, String> { - let offs = Arc::new(offsets(cfg)); - let path = Arc::new(cfg.file.clone()); - - // A shared fd is safe for the cached strategies: pread is positional and - // never touches the file description's shared offset. - let cached: Option> = if cfg.strategy.uses_cached_fd() { - Some(Arc::new(open_read(&cfg.file).map_err(|e| format!("open cached fd: {e}"))?)) - } else { - None - }; - let driver = if cfg.strategy.uses_uring() { - let depth = (cfg.concurrency.max(64) * 2).next_power_of_two() as u32; - Some(Arc::new( - UringDriver::probe_and_start_sharded(depth, cfg.shards).map_err(|e| format!("probe io_uring: {e:?}"))?, - )) - } else { - None - }; - - let (strategy, read_size, verify) = (cfg.strategy, cfg.read_size, cfg.verify); - let per_task = cfg.total_ops.div_ceil(cfg.concurrency); - let mut set = tokio::task::JoinSet::new(); - for t in 0..cfg.concurrency { - let (offs, path, cached, driver) = (offs.clone(), path.clone(), cached.clone(), driver.clone()); - let start_idx = (t * per_task).min(offs.len()); - let end_idx = (start_idx + per_task).min(offs.len()); - set.spawn(async move { - let mut lats = Vec::with_capacity(end_idx - start_idx); - for &off in &offs[start_idx..end_idx] { - let t0 = Instant::now(); - let bytes: Vec = match strategy { - // Today's StdBackend: a blocking-pool hop that opens then preads. - Strategy::StdOpenPread => { - let p = path.clone(); - tokio::task::spawn_blocking(move || { - let f = open_read(&p)?; - let mut buf = vec![0u8; read_size]; - f.read_exact_at(&mut buf, off)?; - io::Result::Ok(buf) - }) - .await - .map_err(|e| format!("join: {e}"))? - .map_err(|e| format!("open+pread: {e}"))? - } - // Blocking pread on a pre-opened fd: prices the open away. - Strategy::StdCachedPread => { - let f = cached.clone().expect("cached fd"); - tokio::task::spawn_blocking(move || { - let mut buf = vec![0u8; read_size]; - f.read_exact_at(&mut buf, off)?; - io::Result::Ok(buf) - }) - .await - .map_err(|e| format!("join: {e}"))? - .map_err(|e| format!("pread: {e}"))? - } - // Today's UringBackend: still a blocking-pool hop for open+stat. - Strategy::UringOpenRead => { - let p = path.clone(); - let file = tokio::task::spawn_blocking(move || { - let f = open_read(&p)?; - f.metadata()?; - io::Result::Ok(f) - }) - .await - .map_err(|e| format!("join: {e}"))? - .map_err(|e| format!("open+stat: {e}"))?; - let d = driver.clone().expect("driver"); - d.read_at(Arc::new(file), off, read_size) - .await - .map_err(|e| format!("uring read: {e}"))? - } - // The ceiling: no blocking pool at all on the read path. - Strategy::UringCachedRead => { - let d = driver.clone().expect("driver"); - let f = cached.clone().expect("cached fd"); - d.read_at(f, off, read_size).await.map_err(|e| format!("uring read: {e}"))? - } - }; - let elapsed = t0.elapsed(); - if bytes.len() != read_size { - return Err(format!("short read at {off}: {} of {read_size}", bytes.len())); - } - if verify { - verify_range(off, &bytes)?; - } - lats.push(elapsed); - } - Ok::<_, String>(lats) - }); - } - - let mut all = Vec::with_capacity(cfg.total_ops); - while let Some(r) = set.join_next().await { - all.extend(r.map_err(|e| format!("task panicked: {e}"))??); - } - Ok(all) +#[cfg(target_os = "linux")] +fn main() -> std::process::ExitCode { + linux::main() } -fn main() -> ExitCode { - let cfg = match parse_args() { - Ok(cfg) => cfg, - Err(e) => { - eprintln!("error: {e}"); - return ExitCode::from(2); - } - }; - if let Err(e) = ensure_file(&cfg.file, cfg.file_size) { - eprintln!("error: {e}"); - return ExitCode::FAILURE; - } - - let rt = match tokio::runtime::Builder::new_multi_thread().enable_all().build() { - Ok(rt) => rt, - Err(e) => { - eprintln!("error: build runtime: {e}"); - return ExitCode::FAILURE; - } - }; - - let start = Instant::now(); - let mut lats = match rt.block_on(run(&cfg)) { - Ok(l) => l, - Err(e) => { - eprintln!("error: {e}"); - return ExitCode::FAILURE; - } - }; - let secs = start.elapsed().as_secs_f64(); - - if cfg.verify { - eprintln!("{}: verified byte-exact across {} reads", cfg.strategy.name(), lats.len()); - } - - lats.sort_unstable(); - let ops = lats.len(); - let iops = ops as f64 / secs; - let mbps = (ops as f64 * cfg.read_size as f64 / (1024.0 * 1024.0)) / secs; - // CSV: strategy,shards,file_size,read_size,concurrency,ops,secs,IOPS,MBps,p50_us,p99_us,p999_us - println!( - "{},{},{},{},{},{},{:.6},{:.0},{:.1},{},{},{}", - cfg.strategy.name(), - cfg.shards, - cfg.file_size, - cfg.read_size, - cfg.concurrency, - ops, - secs, - iops, - mbps, - percentile(&lats, 0.50), - percentile(&lats, 0.99), - percentile(&lats, 0.999), - ); - ExitCode::SUCCESS +#[cfg(not(target_os = "linux"))] +fn main() -> std::process::ExitCode { + eprintln!("concurrent_pread_bench is only supported on Linux because it benchmarks io_uring."); + std::process::ExitCode::from(2) } diff --git a/examples/concurrent_pread_bench/linux.rs b/examples/concurrent_pread_bench/linux.rs new file mode 100644 index 0000000..60fb6fc --- /dev/null +++ b/examples/concurrent_pread_bench/linux.rs @@ -0,0 +1,439 @@ +// 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. + +//! Concurrent positioned-read benchmark — the shape ecstore actually serves. +//! +//! `streaming_bench` measured a single sequential stream and (correctly) found +//! io_uring loses to kernel readahead. That is not io_uring's lever. Its levers +//! are (a) batching many independent IOs into one `io_uring_enter`, and (b) +//! keeping the IO off a blocking thread. Both only appear when *many concurrent +//! positioned reads* hit one disk — exactly what erasure-coded shard reads +//! (`pread_bytes`) do under load. +//! +//! Four strategies isolate where the time actually goes: +//! +//! | strategy | models | +//! |---------------------|-----------------------------------------------------------| +//! | `std_open_pread` | today's StdBackend: spawn_blocking{open; pread} | +//! | `std_cached_pread` | spawn_blocking{pread} on a pre-opened fd | +//! | `uring_open_read` | today's UringBackend: spawn_blocking{open;stat} + read_at | +//! | `uring_cached_read` | pre-opened fd + read_at, NO spawn_blocking | +//! +//! `std_open_pread` vs `uring_open_read` answers "is today's io_uring wiring +//! worth anything". `*_cached_*` vs `*_open_*` prices the per-read open. +//! `uring_cached_read` is the ceiling a fd cache + registered files would buy. +//! +//! Usage: +//! concurrent_pread_bench +//! +//! The file is created (filled + fsync'd) only if it does not exist. An existing +//! path is never truncated, overwritten, or followed through a symlink, so a +//! mistyped path cannot destroy data even though the sweep runs as root. +//! +//! Correctness: IOPS alone cannot tell a correct strategy from one that reads +//! the wrong offsets, so the file is filled with an offset-addressable pattern +//! (`byte[i] = splitmix64(i / 8)[i % 8]`). Setting `BENCH_VERIFY=1` checks every +//! byte each strategy delivers against that pattern at its claimed offset. +//! Verification is skipped entirely when unset, so it never perturbs a timed +//! run; a `BENCH_VERIFY=1` run is a correctness check, not a measurement. + +use std::fs::{File, OpenOptions}; +use std::io::{self, Write}; +use std::os::unix::fs::{FileExt, OpenOptionsExt}; +use std::process::ExitCode; +use std::sync::Arc; +use std::time::{Duration, Instant}; + +use rustfs_uring::UringDriver; + +/// Keeps `(concurrency * 2).next_power_of_two()` ring entries under the +/// kernel's 32768-entry limit and the arithmetic far from overflow. +const MAX_CONCURRENCY: usize = 4096; +const MAX_READ_SIZE: usize = 1 << 30; + +#[derive(Clone, Copy, PartialEq, Eq)] +enum Strategy { + StdOpenPread, + StdCachedPread, + UringOpenRead, + UringCachedRead, +} + +impl Strategy { + fn parse(s: &str) -> Result { + match s { + "std_open_pread" => Ok(Self::StdOpenPread), + "std_cached_pread" => Ok(Self::StdCachedPread), + "uring_open_read" => Ok(Self::UringOpenRead), + "uring_cached_read" => Ok(Self::UringCachedRead), + other => Err(format!( + "unknown strategy {other:?}; expected one of std_open_pread, std_cached_pread, uring_open_read, uring_cached_read" + )), + } + } + + fn name(self) -> &'static str { + match self { + Self::StdOpenPread => "std_open_pread", + Self::StdCachedPread => "std_cached_pread", + Self::UringOpenRead => "uring_open_read", + Self::UringCachedRead => "uring_cached_read", + } + } + + /// Strategies that reuse one pre-opened fd across every read. + fn uses_cached_fd(self) -> bool { + matches!(self, Self::StdCachedPread | Self::UringCachedRead) + } + + fn uses_uring(self) -> bool { + matches!(self, Self::UringOpenRead | Self::UringCachedRead) + } +} + +#[derive(Clone)] +struct Config { + strategy: Strategy, + file: String, + file_size: u64, + read_size: usize, + concurrency: usize, + total_ops: usize, + shards: usize, + verify: bool, +} + +fn parse_args() -> Result { + let a: Vec = std::env::args().collect(); + if a.len() != 7 && a.len() != 8 { + return Err( + "usage: concurrent_pread_bench [shards]" + .to_string(), + ); + } + let parse = |name: &str, raw: &str| -> Result { + raw.parse::() + .map_err(|_| format!("{name}: {raw:?} is not a non-negative integer")) + }; + let cfg = Config { + strategy: Strategy::parse(&a[1])?, + file: a[2].clone(), + file_size: parse("file_size", &a[3])?, + read_size: parse("read_size", &a[4])? as usize, + concurrency: parse("concurrency", &a[5])? as usize, + total_ops: parse("total_ops", &a[6])? as usize, + shards: if a.len() == 8 { parse("shards", &a[7])? as usize } else { 1 }, + verify: matches!(std::env::var("BENCH_VERIFY").as_deref(), Ok("1") | Ok("true")), + }; + validate(&cfg)?; + Ok(cfg) +} + +/// Reject geometry the strategies cannot honour, before any I/O runs. +fn validate(cfg: &Config) -> Result<(), String> { + if cfg.read_size == 0 || cfg.read_size > MAX_READ_SIZE { + return Err(format!("read_size must be in 1..={MAX_READ_SIZE}, got {}", cfg.read_size)); + } + if cfg.file_size > i64::MAX as u64 { + return Err("file_size exceeds i64::MAX (the kernel's signed loff_t)".to_string()); + } + // Offsets are drawn from `file_size - read_size` rounded down to 4 KiB, so + // the file must hold at least one whole read plus a block to pick from. + if cfg.file_size < cfg.read_size as u64 + 4096 { + return Err(format!( + "file_size ({}) must exceed read_size ({}) by at least 4096 bytes", + cfg.file_size, cfg.read_size + )); + } + if cfg.concurrency == 0 || cfg.concurrency > MAX_CONCURRENCY { + return Err(format!("concurrency must be in 1..={MAX_CONCURRENCY}, got {}", cfg.concurrency)); + } + if cfg.total_ops == 0 { + return Err("total_ops must be > 0".to_string()); + } + Ok(()) +} + +// --------------------------------------------------------------------------- +// Offset-addressable file content (shared shape with streaming_bench) +// --------------------------------------------------------------------------- + +fn splitmix64(seed: u64) -> u64 { + let mut z = seed.wrapping_add(0x9e37_79b9_7f4a_7c15); + z = (z ^ (z >> 30)).wrapping_mul(0xbf58_476d_1ce4_e5b9); + z = (z ^ (z >> 27)).wrapping_mul(0x94d0_49bb_1331_11eb); + z ^ (z >> 31) +} + +/// The expected 8-byte little-endian word covering byte offsets +/// `idx * 8 .. idx * 8 + 8`. Content is a pure function of absolute offset, so a +/// read at any offset can be checked without reading anything before it. +fn expected_word(idx: u64) -> [u8; 8] { + splitmix64(idx).to_le_bytes() +} + +/// Check that `data` is exactly what the file holds at `offset`. Catches a +/// strategy that reads the right *number* of bytes from the wrong offsets — a +/// length-only assertion cannot. +fn verify_range(offset: u64, data: &[u8]) -> Result<(), String> { + let (mut cached_idx, mut word) = (u64::MAX, [0u8; 8]); + for (i, &got) in data.iter().enumerate() { + let pos = offset + i as u64; + let idx = pos / 8; + if idx != cached_idx { + cached_idx = idx; + word = expected_word(idx); + } + let want = word[(pos % 8) as usize]; + if got != want { + return Err(format!("content mismatch at byte {pos}: got {got:#04x}, want {want:#04x}")); + } + } + Ok(()) +} + +/// Create the bench file if missing. An existing path is left untouched: never +/// truncated, never followed through a symlink (`create_new` implies `O_EXCL`; +/// `symlink_metadata` does not resolve one). A wrong-size or non-regular path is +/// an error rather than an overwrite, because the sweep runs as root and the +/// path is caller-supplied. +fn ensure_file(path: &str, size: u64) -> Result<(), String> { + match std::fs::symlink_metadata(path) { + Ok(meta) if !meta.file_type().is_file() => Err(format!( + "{path} exists and is not a regular file ({:?}); refusing to touch it", + meta.file_type() + )), + Ok(meta) if meta.len() != size => Err(format!( + "{path} exists with {} bytes but {size} were requested; refusing to truncate it — delete it or pick another path", + meta.len() + )), + Ok(_) => Ok(()), + Err(e) if e.kind() == io::ErrorKind::NotFound => create_file(path, size).map_err(|e| { + // A half-written file would silently become the "wrong size" error + // on the next run; leave nothing behind. + let _ = std::fs::remove_file(path); + format!("create bench file {path}: {e}") + }), + Err(e) => Err(format!("stat {path}: {e}")), + } +} + +fn create_file(path: &str, size: u64) -> io::Result<()> { + let mut f = OpenOptions::new() + .write(true) + .create_new(true) + .custom_flags(libc::O_NOFOLLOW | libc::O_CLOEXEC) + .open(path)?; + let mut buf = vec![0u8; 1 << 20]; + let mut written = 0u64; + while written < size { + // `written` is always a multiple of 8 (the buffer is), so word indices + // line up with absolute offsets. + let base = written / 8; + for (w, word) in buf.chunks_exact_mut(8).enumerate() { + word.copy_from_slice(&expected_word(base + w as u64)); + } + let n = ((size - written) as usize).min(buf.len()); + f.write_all(&buf[..n])?; + written += n as u64; + } + f.sync_all() +} + +/// O_NOFOLLOW closes the window between `ensure_file`'s stat and this open. +fn open_read(path: &str) -> io::Result { + OpenOptions::new() + .read(true) + .custom_flags(libc::O_NOFOLLOW | libc::O_CLOEXEC) + .open(path) +} + +/// Deterministic block-aligned offsets, so every strategy reads the exact same +/// blocks and the comparison is not confounded by a different access pattern. +fn offsets(cfg: &Config) -> Vec { + let blocks = (cfg.file_size - cfg.read_size as u64) / 4096; + let mut state = 0x2545_f491_4f6c_dd1du64; + (0..cfg.total_ops) + .map(|_| { + state = state.wrapping_mul(6364136223846793005).wrapping_add(1442695040888963407); + ((state >> 33) % blocks) * 4096 + }) + .collect() +} + +fn percentile(sorted: &[Duration], p: f64) -> u128 { + if sorted.is_empty() { + return 0; + } + let idx = ((sorted.len() as f64 - 1.0) * p).round() as usize; + sorted[idx].as_micros() +} + +async fn run(cfg: &Config) -> Result, String> { + let offs = Arc::new(offsets(cfg)); + let path = Arc::new(cfg.file.clone()); + + // A shared fd is safe for the cached strategies: pread is positional and + // never touches the file description's shared offset. + let cached: Option> = if cfg.strategy.uses_cached_fd() { + Some(Arc::new(open_read(&cfg.file).map_err(|e| format!("open cached fd: {e}"))?)) + } else { + None + }; + let driver = if cfg.strategy.uses_uring() { + let depth = (cfg.concurrency.max(64) * 2).next_power_of_two() as u32; + Some(Arc::new( + UringDriver::probe_and_start_sharded(depth, cfg.shards).map_err(|e| format!("probe io_uring: {e:?}"))?, + )) + } else { + None + }; + + let (strategy, read_size, verify) = (cfg.strategy, cfg.read_size, cfg.verify); + let per_task = cfg.total_ops.div_ceil(cfg.concurrency); + let mut set = tokio::task::JoinSet::new(); + for t in 0..cfg.concurrency { + let (offs, path, cached, driver) = (offs.clone(), path.clone(), cached.clone(), driver.clone()); + let start_idx = (t * per_task).min(offs.len()); + let end_idx = (start_idx + per_task).min(offs.len()); + set.spawn(async move { + let mut lats = Vec::with_capacity(end_idx - start_idx); + for &off in &offs[start_idx..end_idx] { + let t0 = Instant::now(); + let bytes: Vec = match strategy { + // Today's StdBackend: a blocking-pool hop that opens then preads. + Strategy::StdOpenPread => { + let p = path.clone(); + tokio::task::spawn_blocking(move || { + let f = open_read(&p)?; + let mut buf = vec![0u8; read_size]; + f.read_exact_at(&mut buf, off)?; + Ok::<_, io::Error>(buf) + }) + .await + .map_err(|e| format!("join: {e}"))? + .map_err(|e| format!("open+pread: {e}"))? + } + // Blocking pread on a pre-opened fd: prices the open away. + Strategy::StdCachedPread => { + let f = cached.clone().expect("cached fd"); + tokio::task::spawn_blocking(move || { + let mut buf = vec![0u8; read_size]; + f.read_exact_at(&mut buf, off)?; + Ok::<_, io::Error>(buf) + }) + .await + .map_err(|e| format!("join: {e}"))? + .map_err(|e| format!("pread: {e}"))? + } + // Today's UringBackend: still a blocking-pool hop for open+stat. + Strategy::UringOpenRead => { + let p = path.clone(); + let file = tokio::task::spawn_blocking(move || { + let f = open_read(&p)?; + f.metadata()?; + Ok::<_, io::Error>(f) + }) + .await + .map_err(|e| format!("join: {e}"))? + .map_err(|e| format!("open+stat: {e}"))?; + let d = driver.clone().expect("driver"); + d.read_at(Arc::new(file), off, read_size) + .await + .map_err(|e| format!("uring read: {e}"))? + } + // The ceiling: no blocking pool at all on the read path. + Strategy::UringCachedRead => { + let d = driver.clone().expect("driver"); + let f = cached.clone().expect("cached fd"); + d.read_at(f, off, read_size).await.map_err(|e| format!("uring read: {e}"))? + } + }; + let elapsed = t0.elapsed(); + if bytes.len() != read_size { + return Err(format!("short read at {off}: {} of {read_size}", bytes.len())); + } + if verify { + verify_range(off, &bytes)?; + } + lats.push(elapsed); + } + Ok::<_, String>(lats) + }); + } + + let mut all = Vec::with_capacity(cfg.total_ops); + while let Some(r) = set.join_next().await { + all.extend(r.map_err(|e| format!("task panicked: {e}"))??); + } + Ok(all) +} + +pub(super) fn main() -> ExitCode { + let cfg = match parse_args() { + Ok(cfg) => cfg, + Err(e) => { + eprintln!("error: {e}"); + return ExitCode::from(2); + } + }; + if let Err(e) = ensure_file(&cfg.file, cfg.file_size) { + eprintln!("error: {e}"); + return ExitCode::FAILURE; + } + + let rt = match tokio::runtime::Builder::new_multi_thread().enable_all().build() { + Ok(rt) => rt, + Err(e) => { + eprintln!("error: build runtime: {e}"); + return ExitCode::FAILURE; + } + }; + + let start = Instant::now(); + let mut lats = match rt.block_on(run(&cfg)) { + Ok(l) => l, + Err(e) => { + eprintln!("error: {e}"); + return ExitCode::FAILURE; + } + }; + let secs = start.elapsed().as_secs_f64(); + + if cfg.verify { + eprintln!("{}: verified byte-exact across {} reads", cfg.strategy.name(), lats.len()); + } + + lats.sort_unstable(); + let ops = lats.len(); + let iops = ops as f64 / secs; + let mbps = (ops as f64 * cfg.read_size as f64 / (1024.0 * 1024.0)) / secs; + // CSV: strategy,shards,file_size,read_size,concurrency,ops,secs,IOPS,MBps,p50_us,p99_us,p999_us + println!( + "{},{},{},{},{},{},{:.6},{:.0},{:.1},{},{},{}", + cfg.strategy.name(), + cfg.shards, + cfg.file_size, + cfg.read_size, + cfg.concurrency, + ops, + secs, + iops, + mbps, + percentile(&lats, 0.50), + percentile(&lats, 0.99), + percentile(&lats, 0.999), + ); + ExitCode::SUCCESS +} diff --git a/examples/streaming_bench.rs b/examples/streaming_bench.rs index 997710e..903239a 100644 --- a/examples/streaming_bench.rs +++ b/examples/streaming_bench.rs @@ -12,431 +12,17 @@ // See the License for the specific language governing permissions and // limitations under the License. -//! Streaming-read A/B benchmark for rustfs/backlog#1144 (the 3c go/no-go). -//! -//! Reads one file end-to-end with a chosen strategy and prints a single CSV -//! row (throughput is the decision metric). One measurement per invocation so -//! the runner script can control page-cache state (cold vs warm) between runs. -//! -//! The question it answers: does routing ecstore's *sequential* streaming reads -//! through io_uring beat StdBackend's buffered read (which rides kernel -//! readahead)? io_uring's only lever on a sequential stream is *pipelining* -//! (queue depth > 1), so the io_uring strategies take a `qd` and keep that many -//! reads in flight. A single-depth io_uring read is expected to lose; the -//! interesting comparison is deep-pipelined io_uring vs buffered+readahead. -//! -//! Usage: -//! streaming_bench -//! -//! strategy ∈ { std_buffered, std_odirect, uring_read_at, uring_read_at_direct } -//! -//! The file is created (filled + fsync'd) only if it does not exist. An existing -//! path is never truncated, overwritten, or followed through a symlink: a wrong -//! size or a non-regular file is a hard error, so a mistyped path cannot destroy -//! data even though the sweep runs as root. Once created, the file is reused -//! across runs so the runner can drop caches without rewriting it. -//! -//! Correctness: throughput alone cannot tell a correct strategy from one that -//! reads the wrong offsets, so the file is filled with an offset-addressable -//! pattern (`byte[i] = splitmix64(i / 8)[i % 8]`). Setting `BENCH_VERIFY=1` -//! checks every byte a strategy delivers against that pattern, at its claimed -//! offset. Verification is skipped entirely when the variable is unset, so it -//! never perturbs a timed run; a `BENCH_VERIFY=1` run is a correctness check, -//! not a measurement, and its CSV row must be discarded. +#[cfg(target_os = "linux")] +#[path = "streaming_bench/linux.rs"] +mod linux; -use std::fs::{File, OpenOptions}; -use std::io::{self, Read, Write}; -use std::os::unix::fs::{FileExt, OpenOptionsExt}; -use std::process::ExitCode; -use std::sync::Arc; -use std::time::Instant; - -use rustfs_uring::UringDriver; - -/// Queue depth ceiling. `probe_and_start` gets `(qd * 2).next_power_of_two()` -/// entries, so this keeps the ring under the kernel's 32768-entry limit and the -/// arithmetic far from overflow. -const MAX_QD: usize = 4096; -/// Smallest logical block size any device reports. -const MIN_ALIGN: usize = 512; -const MAX_ALIGN: usize = 1 << 20; -/// One chunk is one in-flight buffer; `qd` of them are live at once. -const MAX_CHUNK: usize = 1 << 30; - -#[derive(Clone, Copy, PartialEq, Eq)] -enum Strategy { - StdBuffered, - StdODirect, - UringReadAt, - UringReadAtDirect, -} - -impl Strategy { - fn parse(s: &str) -> Result { - match s { - "std_buffered" => Ok(Self::StdBuffered), - "std_odirect" => Ok(Self::StdODirect), - "uring_read_at" => Ok(Self::UringReadAt), - "uring_read_at_direct" => Ok(Self::UringReadAtDirect), - other => Err(format!( - "unknown strategy {other:?}; expected one of std_buffered, std_odirect, uring_read_at, uring_read_at_direct" - )), - } - } - - fn name(self) -> &'static str { - match self { - Self::StdBuffered => "std_buffered", - Self::StdODirect => "std_odirect", - Self::UringReadAt => "uring_read_at", - Self::UringReadAtDirect => "uring_read_at_direct", - } - } - - fn is_direct(self) -> bool { - matches!(self, Self::StdODirect | Self::UringReadAtDirect) - } -} - -#[derive(Clone)] -struct Config { - strategy: Strategy, - file: String, - size: usize, - chunk: usize, - qd: usize, - align: usize, - verify: bool, -} - -fn parse_usize(name: &str, raw: &str) -> Result { - raw.parse() - .map_err(|_| format!("{name}: {raw:?} is not a non-negative integer")) -} - -fn parse_args() -> Result { - let a: Vec = std::env::args().collect(); - if a.len() != 7 { - return Err("usage: streaming_bench ".to_string()); - } - let cfg = Config { - strategy: Strategy::parse(&a[1])?, - file: a[2].clone(), - size: parse_usize("size_bytes", &a[3])?, - chunk: parse_usize("chunk_bytes", &a[4])?, - qd: parse_usize("qd", &a[5])?, - align: parse_usize("align", &a[6])?, - verify: matches!(std::env::var("BENCH_VERIFY").as_deref(), Ok("1") | Ok("true")), - }; - validate(&cfg)?; - Ok(cfg) -} - -/// Reject geometry the strategies cannot honour, before any I/O runs. Without -/// this, `chunk == 0` panics in `step_by`, `align == 0` panics in -/// `next_multiple_of`/`align_offset`, and a large `qd` overflows the ring-entry -/// arithmetic. -fn validate(cfg: &Config) -> Result<(), String> { - if cfg.size == 0 { - return Err("size_bytes must be > 0".to_string()); - } - if cfg.size as u64 > i64::MAX as u64 { - return Err("size_bytes exceeds i64::MAX (the kernel's signed loff_t)".to_string()); - } - if cfg.chunk == 0 || cfg.chunk > MAX_CHUNK { - return Err(format!("chunk_bytes must be in 1..={MAX_CHUNK}, got {}", cfg.chunk)); - } - if cfg.qd == 0 || cfg.qd > MAX_QD { - return Err(format!("qd must be in 1..={MAX_QD}, got {}", cfg.qd)); - } - // `align_offset` and the kernel's O_DIRECT contract both require a power of - // two; validate it for every strategy so the sweep cannot pass one geometry - // to the baselines and another to io_uring. - if !cfg.align.is_power_of_two() || cfg.align < MIN_ALIGN || cfg.align > MAX_ALIGN { - return Err(format!("align must be a power of two in {MIN_ALIGN}..={MAX_ALIGN}, got {}", cfg.align)); - } - if cfg.strategy.is_direct() && !cfg.chunk.is_multiple_of(cfg.align) { - return Err(format!( - "O_DIRECT strategies need chunk_bytes ({}) to be a multiple of align ({}), so every read offset is block-aligned", - cfg.chunk, cfg.align - )); - } - if cfg.chunk.checked_add(cfg.align).is_none() { - return Err("chunk_bytes + align overflows usize".to_string()); - } - Ok(()) -} - -// --------------------------------------------------------------------------- -// Offset-addressable file content -// --------------------------------------------------------------------------- - -fn splitmix64(seed: u64) -> u64 { - let mut z = seed.wrapping_add(0x9e37_79b9_7f4a_7c15); - z = (z ^ (z >> 30)).wrapping_mul(0xbf58_476d_1ce4_e5b9); - z = (z ^ (z >> 27)).wrapping_mul(0x94d0_49bb_1331_11eb); - z ^ (z >> 31) -} - -/// The expected 8-byte little-endian word at word index `idx`, i.e. covering -/// byte offsets `idx * 8 .. idx * 8 + 8`. Content is a pure function of absolute -/// offset, so any chunk can be checked without reading the ones before it. -fn expected_word(idx: u64) -> [u8; 8] { - splitmix64(idx).to_le_bytes() -} - -/// Check that `data` is exactly what the file holds at `offset`. Catches a -/// strategy that reads the right *number* of bytes from the wrong offsets, or -/// repeats a chunk — both of which pass a length-only assertion. -fn verify_range(offset: u64, data: &[u8]) -> Result<(), String> { - let (mut cached_idx, mut word) = (u64::MAX, [0u8; 8]); - for (i, &got) in data.iter().enumerate() { - let pos = offset + i as u64; - let idx = pos / 8; - if idx != cached_idx { - cached_idx = idx; - word = expected_word(idx); - } - let want = word[(pos % 8) as usize]; - if got != want { - return Err(format!("content mismatch at byte {pos}: got {got:#04x}, want {want:#04x}")); - } - } - Ok(()) +#[cfg(target_os = "linux")] +fn main() -> std::process::ExitCode { + linux::main() } -/// Create the bench file if it is missing. An existing path is left untouched: -/// this never truncates, and never follows a symlink (`create_new` implies -/// `O_EXCL`, which fails on one; `symlink_metadata` does not resolve one). A -/// wrong-size or non-regular path is an error rather than an overwrite, because -/// the sweep runs as root and `BENCH_DIR` is caller-supplied. -fn ensure_file(path: &str, size: usize) -> Result<(), String> { - match std::fs::symlink_metadata(path) { - Ok(meta) if !meta.file_type().is_file() => Err(format!( - "{path} exists and is not a regular file ({:?}); refusing to touch it", - meta.file_type() - )), - Ok(meta) if meta.len() != size as u64 => Err(format!( - "{path} exists with {} bytes but {size} were requested; refusing to truncate it — delete it or pick another path", - meta.len() - )), - Ok(_) => Ok(()), - Err(e) if e.kind() == io::ErrorKind::NotFound => create_file(path, size).map_err(|e| { - // A half-written file would silently become the "wrong size" error - // on the next run; leave nothing behind. - let _ = std::fs::remove_file(path); - format!("create bench file {path}: {e}") - }), - Err(e) => Err(format!("stat {path}: {e}")), - } -} - -fn create_file(path: &str, size: usize) -> io::Result<()> { - let mut f = OpenOptions::new() - .write(true) - .create_new(true) - .custom_flags(libc::O_NOFOLLOW | libc::O_CLOEXEC) - .open(path)?; - let mut buf = vec![0u8; 1 << 20]; - let mut written = 0usize; - while written < size { - // `written` is always a multiple of 8 here (the buffer is), so word - // indices line up with absolute offsets. - let base = (written / 8) as u64; - for (w, word) in buf.chunks_exact_mut(8).enumerate() { - word.copy_from_slice(&expected_word(base + w as u64)); - } - let n = (size - written).min(buf.len()); - f.write_all(&buf[..n])?; - written += n; - } - f.sync_all() -} - -fn open_read(cfg: &Config, direct: bool) -> io::Result { - let mut opts = OpenOptions::new(); - opts.read(true); - // O_NOFOLLOW closes the window between `ensure_file`'s stat and this open. - let mut flags = libc::O_NOFOLLOW | libc::O_CLOEXEC; - if direct { - flags |= libc::O_DIRECT; - } - opts.custom_flags(flags).open(&cfg.file) -} - -// --------------------------------------------------------------------------- -// Strategies -// --------------------------------------------------------------------------- - -/// Sequential buffered read: the StdBackend baseline that rides kernel readahead. -fn run_std_buffered(cfg: &Config) -> Result<(usize, usize), String> { - let mut f = open_read(cfg, false).map_err(|e| format!("open: {e}"))?; - let mut buf = vec![0u8; cfg.chunk]; - let (mut total, mut ops) = (0usize, 0usize); - loop { - let n = f.read(&mut buf).map_err(|e| format!("read: {e}"))?; - if n == 0 { - break; - } - if cfg.verify { - verify_range(total as u64, &buf[..n])?; - } - total += n; - ops += 1; - } - Ok((total, ops)) -} - -/// Heap buffer whose start is aligned to `align` (for O_DIRECT). -fn aligned_buf(len: usize, align: usize) -> (Vec, usize) { - let v = vec![0u8; len + align]; - let pad = v.as_ptr().align_offset(align); - assert!(pad + len <= v.len()); - (v, pad) -} - -/// Sequential O_DIRECT read: page-cache-bypassing baseline. -fn run_std_odirect(cfg: &Config) -> Result<(usize, usize), String> { - let f = open_read(cfg, true).map_err(|e| format!("open O_DIRECT: {e}"))?; - // `validate` already requires chunk % align == 0; this is the identity. - let chunk = cfg.chunk; - let (mut buf, pad) = aligned_buf(chunk, cfg.align); - let (mut total, mut ops, mut off) = (0usize, 0usize, 0u64); - while (off as usize) < cfg.size { - let n = f - .read_at(&mut buf[pad..pad + chunk], off) - .map_err(|e| format!("O_DIRECT read_at: {e}"))?; - if n == 0 { - break; - } - // The kernel returns whole blocks; count only the logical remainder. - let logical = n.min(cfg.size - off as usize); - if cfg.verify { - verify_range(off, &buf[pad..pad + logical])?; - } - total += logical; - ops += 1; - off += n as u64; - } - Ok((total, ops)) -} - -/// Pipelined io_uring read at depth `qd`. `direct` selects read_at_direct. -async fn run_uring(cfg: &Config, direct: bool) -> Result<(usize, usize), String> { - // `validate` caps qd at MAX_QD, so this cannot overflow u32. - let depth = (cfg.qd * 2).next_power_of_two() as u32; - let driver = Arc::new(UringDriver::probe_and_start(depth).map_err(|e| format!("probe io_uring: {e:?}"))?); - let file = Arc::new(open_read(cfg, direct).map_err(|e| format!("open: {e}"))?); - - let offsets: Vec<(u64, usize)> = (0..cfg.size) - .step_by(cfg.chunk) - .map(|o| (o as u64, cfg.chunk.min(cfg.size - o))) - .collect(); - - let (mut total, mut ops) = (0usize, 0usize); - let mut set = tokio::task::JoinSet::new(); - let mut idx = 0usize; - let align = cfg.align; - let verify = cfg.verify; - let spawn = |set: &mut tokio::task::JoinSet>, idx: usize| { - let (off, len) = offsets[idx]; - let d = driver.clone(); - let f = file.clone(); - set.spawn(async move { - let bytes = if direct { - d.read_at_direct(f, off, len, align).await? - } else { - d.read_at(f, off, len).await? - }; - if verify { - if bytes.len() != len { - return Err(io::Error::other(format!( - "short read at offset {off}: got {} bytes, want {len}", - bytes.len() - ))); - } - verify_range(off, &bytes).map_err(io::Error::other)?; - } - Ok(bytes.len()) - }); - }; - while set.len() < cfg.qd && idx < offsets.len() { - spawn(&mut set, idx); - idx += 1; - } - while let Some(res) = set.join_next().await { - let n = res.map_err(|e| format!("join: {e}"))?.map_err(|e| format!("read: {e}"))?; - total += n; - ops += 1; - if idx < offsets.len() { - spawn(&mut set, idx); - idx += 1; - } - } - Ok((total, ops)) -} - -fn run(cfg: &Config) -> Result<(usize, usize, f64), String> { - let start = Instant::now(); - let (total, ops) = match cfg.strategy { - Strategy::StdBuffered => run_std_buffered(cfg)?, - Strategy::StdODirect => run_std_odirect(cfg)?, - Strategy::UringReadAt | Strategy::UringReadAtDirect => { - let direct = cfg.strategy == Strategy::UringReadAtDirect; - let rt = tokio::runtime::Builder::new_multi_thread() - .enable_all() - .build() - .map_err(|e| format!("runtime: {e}"))?; - rt.block_on(run_uring(cfg, direct))? - } - }; - let secs = start.elapsed().as_secs_f64(); - if total != cfg.size { - return Err(format!("strategy {} read {total} of {} bytes", cfg.strategy.name(), cfg.size)); - } - Ok((total, ops, secs)) -} - -fn main() -> ExitCode { - let cfg = match parse_args().and_then(|cfg| ensure_file(&cfg.file, cfg.size).map(|()| cfg)) { - Ok(cfg) => cfg, - Err(e) => { - eprintln!("streaming_bench: {e}"); - return ExitCode::from(2); - } - }; - - let (total, ops, secs) = match run(&cfg) { - Ok(v) => v, - Err(e) => { - eprintln!("streaming_bench: {e}"); - return ExitCode::FAILURE; - } - }; - - if cfg.verify { - eprintln!( - "streaming_bench: verify=ok strategy={} size={} chunk={} qd={} align={} (timing below is NOT a measurement)", - cfg.strategy.name(), - cfg.size, - cfg.chunk, - cfg.qd, - cfg.align - ); - } - let mbps = (total as f64 / (1024.0 * 1024.0)) / secs; - // CSV: strategy,size,chunk,qd,align,bytes,secs,MBps,ops - println!( - "{},{},{},{},{},{},{:.6},{:.1},{}", - cfg.strategy.name(), - cfg.size, - cfg.chunk, - cfg.qd, - cfg.align, - total, - secs, - mbps, - ops - ); - ExitCode::SUCCESS +#[cfg(not(target_os = "linux"))] +fn main() -> std::process::ExitCode { + eprintln!("streaming_bench is only supported on Linux because it benchmarks io_uring."); + std::process::ExitCode::from(2) } diff --git a/examples/streaming_bench/linux.rs b/examples/streaming_bench/linux.rs new file mode 100644 index 0000000..f91a9bc --- /dev/null +++ b/examples/streaming_bench/linux.rs @@ -0,0 +1,442 @@ +// 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. + +//! Streaming-read A/B benchmark for rustfs/backlog#1144 (the 3c go/no-go). +//! +//! Reads one file end-to-end with a chosen strategy and prints a single CSV +//! row (throughput is the decision metric). One measurement per invocation so +//! the runner script can control page-cache state (cold vs warm) between runs. +//! +//! The question it answers: does routing ecstore's *sequential* streaming reads +//! through io_uring beat StdBackend's buffered read (which rides kernel +//! readahead)? io_uring's only lever on a sequential stream is *pipelining* +//! (queue depth > 1), so the io_uring strategies take a `qd` and keep that many +//! reads in flight. A single-depth io_uring read is expected to lose; the +//! interesting comparison is deep-pipelined io_uring vs buffered+readahead. +//! +//! Usage: +//! streaming_bench +//! +//! strategy ∈ { std_buffered, std_odirect, uring_read_at, uring_read_at_direct } +//! +//! The file is created (filled + fsync'd) only if it does not exist. An existing +//! path is never truncated, overwritten, or followed through a symlink: a wrong +//! size or a non-regular file is a hard error, so a mistyped path cannot destroy +//! data even though the sweep runs as root. Once created, the file is reused +//! across runs so the runner can drop caches without rewriting it. +//! +//! Correctness: throughput alone cannot tell a correct strategy from one that +//! reads the wrong offsets, so the file is filled with an offset-addressable +//! pattern (`byte[i] = splitmix64(i / 8)[i % 8]`). Setting `BENCH_VERIFY=1` +//! checks every byte a strategy delivers against that pattern, at its claimed +//! offset. Verification is skipped entirely when the variable is unset, so it +//! never perturbs a timed run; a `BENCH_VERIFY=1` run is a correctness check, +//! not a measurement, and its CSV row must be discarded. + +use std::fs::{File, OpenOptions}; +use std::io::{self, Read, Write}; +use std::os::unix::fs::{FileExt, OpenOptionsExt}; +use std::process::ExitCode; +use std::sync::Arc; +use std::time::Instant; + +use rustfs_uring::UringDriver; + +/// Queue depth ceiling. `probe_and_start` gets `(qd * 2).next_power_of_two()` +/// entries, so this keeps the ring under the kernel's 32768-entry limit and the +/// arithmetic far from overflow. +const MAX_QD: usize = 4096; +/// Smallest logical block size any device reports. +const MIN_ALIGN: usize = 512; +const MAX_ALIGN: usize = 1 << 20; +/// One chunk is one in-flight buffer; `qd` of them are live at once. +const MAX_CHUNK: usize = 1 << 30; + +#[derive(Clone, Copy, PartialEq, Eq)] +enum Strategy { + StdBuffered, + StdODirect, + UringReadAt, + UringReadAtDirect, +} + +impl Strategy { + fn parse(s: &str) -> Result { + match s { + "std_buffered" => Ok(Self::StdBuffered), + "std_odirect" => Ok(Self::StdODirect), + "uring_read_at" => Ok(Self::UringReadAt), + "uring_read_at_direct" => Ok(Self::UringReadAtDirect), + other => Err(format!( + "unknown strategy {other:?}; expected one of std_buffered, std_odirect, uring_read_at, uring_read_at_direct" + )), + } + } + + fn name(self) -> &'static str { + match self { + Self::StdBuffered => "std_buffered", + Self::StdODirect => "std_odirect", + Self::UringReadAt => "uring_read_at", + Self::UringReadAtDirect => "uring_read_at_direct", + } + } + + fn is_direct(self) -> bool { + matches!(self, Self::StdODirect | Self::UringReadAtDirect) + } +} + +#[derive(Clone)] +struct Config { + strategy: Strategy, + file: String, + size: usize, + chunk: usize, + qd: usize, + align: usize, + verify: bool, +} + +fn parse_usize(name: &str, raw: &str) -> Result { + raw.parse() + .map_err(|_| format!("{name}: {raw:?} is not a non-negative integer")) +} + +fn parse_args() -> Result { + let a: Vec = std::env::args().collect(); + if a.len() != 7 { + return Err("usage: streaming_bench ".to_string()); + } + let cfg = Config { + strategy: Strategy::parse(&a[1])?, + file: a[2].clone(), + size: parse_usize("size_bytes", &a[3])?, + chunk: parse_usize("chunk_bytes", &a[4])?, + qd: parse_usize("qd", &a[5])?, + align: parse_usize("align", &a[6])?, + verify: matches!(std::env::var("BENCH_VERIFY").as_deref(), Ok("1") | Ok("true")), + }; + validate(&cfg)?; + Ok(cfg) +} + +/// Reject geometry the strategies cannot honour, before any I/O runs. Without +/// this, `chunk == 0` panics in `step_by`, `align == 0` panics in +/// `next_multiple_of`/`align_offset`, and a large `qd` overflows the ring-entry +/// arithmetic. +fn validate(cfg: &Config) -> Result<(), String> { + if cfg.size == 0 { + return Err("size_bytes must be > 0".to_string()); + } + if cfg.size as u64 > i64::MAX as u64 { + return Err("size_bytes exceeds i64::MAX (the kernel's signed loff_t)".to_string()); + } + if cfg.chunk == 0 || cfg.chunk > MAX_CHUNK { + return Err(format!("chunk_bytes must be in 1..={MAX_CHUNK}, got {}", cfg.chunk)); + } + if cfg.qd == 0 || cfg.qd > MAX_QD { + return Err(format!("qd must be in 1..={MAX_QD}, got {}", cfg.qd)); + } + // `align_offset` and the kernel's O_DIRECT contract both require a power of + // two; validate it for every strategy so the sweep cannot pass one geometry + // to the baselines and another to io_uring. + if !cfg.align.is_power_of_two() || cfg.align < MIN_ALIGN || cfg.align > MAX_ALIGN { + return Err(format!("align must be a power of two in {MIN_ALIGN}..={MAX_ALIGN}, got {}", cfg.align)); + } + if cfg.strategy.is_direct() && !cfg.chunk.is_multiple_of(cfg.align) { + return Err(format!( + "O_DIRECT strategies need chunk_bytes ({}) to be a multiple of align ({}), so every read offset is block-aligned", + cfg.chunk, cfg.align + )); + } + if cfg.chunk.checked_add(cfg.align).is_none() { + return Err("chunk_bytes + align overflows usize".to_string()); + } + Ok(()) +} + +// --------------------------------------------------------------------------- +// Offset-addressable file content +// --------------------------------------------------------------------------- + +fn splitmix64(seed: u64) -> u64 { + let mut z = seed.wrapping_add(0x9e37_79b9_7f4a_7c15); + z = (z ^ (z >> 30)).wrapping_mul(0xbf58_476d_1ce4_e5b9); + z = (z ^ (z >> 27)).wrapping_mul(0x94d0_49bb_1331_11eb); + z ^ (z >> 31) +} + +/// The expected 8-byte little-endian word at word index `idx`, i.e. covering +/// byte offsets `idx * 8 .. idx * 8 + 8`. Content is a pure function of absolute +/// offset, so any chunk can be checked without reading the ones before it. +fn expected_word(idx: u64) -> [u8; 8] { + splitmix64(idx).to_le_bytes() +} + +/// Check that `data` is exactly what the file holds at `offset`. Catches a +/// strategy that reads the right *number* of bytes from the wrong offsets, or +/// repeats a chunk — both of which pass a length-only assertion. +fn verify_range(offset: u64, data: &[u8]) -> Result<(), String> { + let (mut cached_idx, mut word) = (u64::MAX, [0u8; 8]); + for (i, &got) in data.iter().enumerate() { + let pos = offset + i as u64; + let idx = pos / 8; + if idx != cached_idx { + cached_idx = idx; + word = expected_word(idx); + } + let want = word[(pos % 8) as usize]; + if got != want { + return Err(format!("content mismatch at byte {pos}: got {got:#04x}, want {want:#04x}")); + } + } + Ok(()) +} + +/// Create the bench file if it is missing. An existing path is left untouched: +/// this never truncates, and never follows a symlink (`create_new` implies +/// `O_EXCL`, which fails on one; `symlink_metadata` does not resolve one). A +/// wrong-size or non-regular path is an error rather than an overwrite, because +/// the sweep runs as root and `BENCH_DIR` is caller-supplied. +fn ensure_file(path: &str, size: usize) -> Result<(), String> { + match std::fs::symlink_metadata(path) { + Ok(meta) if !meta.file_type().is_file() => Err(format!( + "{path} exists and is not a regular file ({:?}); refusing to touch it", + meta.file_type() + )), + Ok(meta) if meta.len() != size as u64 => Err(format!( + "{path} exists with {} bytes but {size} were requested; refusing to truncate it — delete it or pick another path", + meta.len() + )), + Ok(_) => Ok(()), + Err(e) if e.kind() == io::ErrorKind::NotFound => create_file(path, size).map_err(|e| { + // A half-written file would silently become the "wrong size" error + // on the next run; leave nothing behind. + let _ = std::fs::remove_file(path); + format!("create bench file {path}: {e}") + }), + Err(e) => Err(format!("stat {path}: {e}")), + } +} + +fn create_file(path: &str, size: usize) -> io::Result<()> { + let mut f = OpenOptions::new() + .write(true) + .create_new(true) + .custom_flags(libc::O_NOFOLLOW | libc::O_CLOEXEC) + .open(path)?; + let mut buf = vec![0u8; 1 << 20]; + let mut written = 0usize; + while written < size { + // `written` is always a multiple of 8 here (the buffer is), so word + // indices line up with absolute offsets. + let base = (written / 8) as u64; + for (w, word) in buf.chunks_exact_mut(8).enumerate() { + word.copy_from_slice(&expected_word(base + w as u64)); + } + let n = (size - written).min(buf.len()); + f.write_all(&buf[..n])?; + written += n; + } + f.sync_all() +} + +fn open_read(cfg: &Config, direct: bool) -> io::Result { + let mut opts = OpenOptions::new(); + opts.read(true); + // O_NOFOLLOW closes the window between `ensure_file`'s stat and this open. + let mut flags = libc::O_NOFOLLOW | libc::O_CLOEXEC; + if direct { + flags |= libc::O_DIRECT; + } + opts.custom_flags(flags).open(&cfg.file) +} + +// --------------------------------------------------------------------------- +// Strategies +// --------------------------------------------------------------------------- + +/// Sequential buffered read: the StdBackend baseline that rides kernel readahead. +fn run_std_buffered(cfg: &Config) -> Result<(usize, usize), String> { + let mut f = open_read(cfg, false).map_err(|e| format!("open: {e}"))?; + let mut buf = vec![0u8; cfg.chunk]; + let (mut total, mut ops) = (0usize, 0usize); + loop { + let n = f.read(&mut buf).map_err(|e| format!("read: {e}"))?; + if n == 0 { + break; + } + if cfg.verify { + verify_range(total as u64, &buf[..n])?; + } + total += n; + ops += 1; + } + Ok((total, ops)) +} + +/// Heap buffer whose start is aligned to `align` (for O_DIRECT). +fn aligned_buf(len: usize, align: usize) -> (Vec, usize) { + let v = vec![0u8; len + align]; + let pad = v.as_ptr().align_offset(align); + assert!(pad + len <= v.len()); + (v, pad) +} + +/// Sequential O_DIRECT read: page-cache-bypassing baseline. +fn run_std_odirect(cfg: &Config) -> Result<(usize, usize), String> { + let f = open_read(cfg, true).map_err(|e| format!("open O_DIRECT: {e}"))?; + // `validate` already requires chunk % align == 0; this is the identity. + let chunk = cfg.chunk; + let (mut buf, pad) = aligned_buf(chunk, cfg.align); + let (mut total, mut ops, mut off) = (0usize, 0usize, 0u64); + while (off as usize) < cfg.size { + let n = f + .read_at(&mut buf[pad..pad + chunk], off) + .map_err(|e| format!("O_DIRECT read_at: {e}"))?; + if n == 0 { + break; + } + // The kernel returns whole blocks; count only the logical remainder. + let logical = n.min(cfg.size - off as usize); + if cfg.verify { + verify_range(off, &buf[pad..pad + logical])?; + } + total += logical; + ops += 1; + off += n as u64; + } + Ok((total, ops)) +} + +/// Pipelined io_uring read at depth `qd`. `direct` selects read_at_direct. +async fn run_uring(cfg: &Config, direct: bool) -> Result<(usize, usize), String> { + // `validate` caps qd at MAX_QD, so this cannot overflow u32. + let depth = (cfg.qd * 2).next_power_of_two() as u32; + let driver = Arc::new(UringDriver::probe_and_start(depth).map_err(|e| format!("probe io_uring: {e:?}"))?); + let file = Arc::new(open_read(cfg, direct).map_err(|e| format!("open: {e}"))?); + + let offsets: Vec<(u64, usize)> = (0..cfg.size) + .step_by(cfg.chunk) + .map(|o| (o as u64, cfg.chunk.min(cfg.size - o))) + .collect(); + + let (mut total, mut ops) = (0usize, 0usize); + let mut set = tokio::task::JoinSet::new(); + let mut idx = 0usize; + let align = cfg.align; + let verify = cfg.verify; + let spawn = |set: &mut tokio::task::JoinSet>, idx: usize| { + let (off, len) = offsets[idx]; + let d = driver.clone(); + let f = file.clone(); + set.spawn(async move { + let bytes = if direct { + d.read_at_direct(f, off, len, align).await? + } else { + d.read_at(f, off, len).await? + }; + if verify { + if bytes.len() != len { + return Err(io::Error::other(format!( + "short read at offset {off}: got {} bytes, want {len}", + bytes.len() + ))); + } + verify_range(off, &bytes).map_err(io::Error::other)?; + } + Ok(bytes.len()) + }); + }; + while set.len() < cfg.qd && idx < offsets.len() { + spawn(&mut set, idx); + idx += 1; + } + while let Some(res) = set.join_next().await { + let n = res.map_err(|e| format!("join: {e}"))?.map_err(|e| format!("read: {e}"))?; + total += n; + ops += 1; + if idx < offsets.len() { + spawn(&mut set, idx); + idx += 1; + } + } + Ok((total, ops)) +} + +fn run(cfg: &Config) -> Result<(usize, usize, f64), String> { + let start = Instant::now(); + let (total, ops) = match cfg.strategy { + Strategy::StdBuffered => run_std_buffered(cfg)?, + Strategy::StdODirect => run_std_odirect(cfg)?, + Strategy::UringReadAt | Strategy::UringReadAtDirect => { + let direct = cfg.strategy == Strategy::UringReadAtDirect; + let rt = tokio::runtime::Builder::new_multi_thread() + .enable_all() + .build() + .map_err(|e| format!("runtime: {e}"))?; + rt.block_on(run_uring(cfg, direct))? + } + }; + let secs = start.elapsed().as_secs_f64(); + if total != cfg.size { + return Err(format!("strategy {} read {total} of {} bytes", cfg.strategy.name(), cfg.size)); + } + Ok((total, ops, secs)) +} + +pub(super) fn main() -> ExitCode { + let cfg = match parse_args().and_then(|cfg| ensure_file(&cfg.file, cfg.size).map(|()| cfg)) { + Ok(cfg) => cfg, + Err(e) => { + eprintln!("streaming_bench: {e}"); + return ExitCode::from(2); + } + }; + + let (total, ops, secs) = match run(&cfg) { + Ok(v) => v, + Err(e) => { + eprintln!("streaming_bench: {e}"); + return ExitCode::FAILURE; + } + }; + + if cfg.verify { + eprintln!( + "streaming_bench: verify=ok strategy={} size={} chunk={} qd={} align={} (timing below is NOT a measurement)", + cfg.strategy.name(), + cfg.size, + cfg.chunk, + cfg.qd, + cfg.align + ); + } + let mbps = (total as f64 / (1024.0 * 1024.0)) / secs; + // CSV: strategy,size,chunk,qd,align,bytes,secs,MBps,ops + println!( + "{},{},{},{},{},{},{:.6},{:.1},{}", + cfg.strategy.name(), + cfg.size, + cfg.chunk, + cfg.qd, + cfg.align, + total, + secs, + mbps, + ops + ); + ExitCode::SUCCESS +}