crates/disk-snapshot/src/speed.rs
//! **The disk speed test**: what a data root can actually do, measured once.
//!
//! ```text
//! seq w 812 MB/s · r 1.9 GB/s · 4k sync 3100 ops/s (1.6 s)
//! ```
//!
//! Moved AS IS out of gunnar-server's `disktest.rs` (gunnar-live 50c3e69) when
//! holger's admin view needed the same number (owner 2026-09-28: "measured disk
//! speed"; RULE 2: product-neutral code goes to edda). What stayed behind is
//! the product's: WHEN to run it (gunnar gates it on being PID 1 and holds its
//! boot-readiness window around it), and HOW to say it (gunnar's dashboard
//! row). Here is only the measurement and the latch that keeps its answer.
//!
//! Three numbers, because they are the three a store lives on: a sequential
//! write (a large upload arriving), a sequential read (a download leaving), and
//! 4 KiB overwrites each followed by `fsync` — the small-metadata-update
//! pattern, and the number that separates a slow filesystem from a fast one.
//!
//! The write and the 4 KiB legs are plain `std::fs`. The READ leg measures the
//! device, not the page cache: the file it just wrote is still cached, so the
//! read reopens it with `O_DIRECT` (one 4 MiB buffer, 4 KiB-aligned, allocated
//! once for the whole run) and streams it back off the disk. Where `O_DIRECT`
//! is refused (tmpfs, some FUSE) the fallback is `posix_fadvise(DONTNEED)` to
//! drop the cached pages, then a buffered read — and a result that is still
//! implausible (over 5 GB/s) is said to be `cached` rather than reported as disk
//! speed. 256 MiB under `<data>/.korp-disktest`, deleted afterwards.
//!
//! Never a panic: too little free space is [`Outcome::Skipped`], any I/O error
//! is [`Outcome::Failed`], and both leave the file gone. **Not run yet is
//! [`last`] = `None`** — a different fact from a disk that measured badly, and
//! never a zero.
use std::fs::{File, OpenOptions};
use std::io::{Read, Seek, SeekFrom, Write};
use std::os::unix::fs::OpenOptionsExt;
use std::path::Path;
use std::time::{Duration, Instant};
use rustix::fs::{fadvise, Advice, OFlags};
/// The file's directory under the data root; removed when the test is done.
pub const DIR_NAME: &str = ".korp-disktest";
/// What the test writes.
pub const FILE_BYTES: u64 = 256 << 20;
/// Skip below this much free space: the test must never be the thing that
/// fills the disk.
pub const MIN_FREE_BYTES: u64 = 1 << 30;
/// How long the 4 KiB fsync loop runs.
pub const SYNC_WINDOW: Duration = Duration::from_millis(500);
/// A read faster than this did not touch the disk.
pub const CACHED_BPS: f64 = 5e9;
/// The write chunk and the read buffer: one buffer of this size, allocated
/// once per run, aligned for `O_DIRECT`.
const CHUNK: usize = 4 << 20;
const BLOCK: usize = 4096;
/// `O_DIRECT` wants the buffer, the offset and the length aligned to the
/// logical block; 4 KiB satisfies every disk an appliance runs on.
const DIRECT_ALIGN: usize = 4096;
/// How the read leg got its bytes.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum ReadHow {
/// `O_DIRECT`: the page cache was bypassed, the number is the device's.
Direct,
/// `O_DIRECT` refused; the cached pages were dropped (`fadvise
/// DONTNEED`) and the file read through the cache.
Dropped,
}
impl ReadHow {
pub fn as_str(self) -> &'static str {
match self {
ReadHow::Direct => "direct",
ReadHow::Dropped => "dropped",
}
}
}
/// What the test measured.
#[derive(Debug, Clone, PartialEq)]
pub struct Report {
pub seq_write_bps: f64,
pub seq_read_bps: f64,
/// Whether the read bypassed the page cache or only dropped it first.
pub read_how: ReadHow,
pub sync_ops_per_s: f64,
pub elapsed: Duration,
}
impl Report {
/// True when the read leg was faster than any disk: the page cache answered.
pub fn read_was_cached(&self) -> bool {
self.seq_read_bps > CACHED_BPS
}
/// `seq w 812 MB/s · r 1.9 GB/s · 4k sync 3100 ops/s (1.6 s)`; a read
/// over [`CACHED_BPS`] reads `r cached`.
pub fn row(&self) -> String {
let read = if self.read_was_cached() { "cached".to_owned() } else { fmt_rate(self.seq_read_bps) };
format!(
"seq w {} \u{b7} r {} \u{b7} 4k sync {:.0} ops/s ({:.1} s)",
fmt_rate(self.seq_write_bps),
read,
self.sync_ops_per_s,
self.elapsed.as_secs_f64()
)
}
}
/// `812 MB/s` / `1.9 GB/s` (decimal units, as disks are sold).
pub fn fmt_rate(bps: f64) -> String {
if bps >= 1e9 {
format!("{:.1} GB/s", bps / 1e9)
} else if bps >= 1e6 {
format!("{:.0} MB/s", bps / 1e6)
} else {
format!("{:.0} kB/s", bps / 1e3)
}
}
/// `800M` / `1.0G` (binary units, as free space is read).
fn fmt_size(bytes: u64) -> String {
const G: u64 = 1 << 30;
if bytes >= G {
format!("{:.1}G", bytes as f64 / G as f64)
} else {
format!("{}M", bytes >> 20)
}
}
/// How one run ended.
#[derive(Debug, Clone, PartialEq)]
pub enum Outcome {
Done(Report),
/// Too little free space: `(free, needed)` bytes.
Skipped { free: u64, need: u64 },
Failed(String),
}
impl Outcome {
/// The row's text, and whether it is a warning (`Failed`).
pub fn row(&self) -> (String, bool) {
match self {
Outcome::Done(r) => (r.row(), false),
Outcome::Skipped { free, need } => (format!("skipped: {} free (< {})", fmt_size(*free), fmt_size(*need)), false),
Outcome::Failed(why) => (format!("failed: {why}"), true),
}
}
}
/// The skip rule, pure: `free` bytes available against the need.
pub fn should_skip(free: u64, need: u64) -> bool {
free < need
}
/// Run the test under `data_root` with the production sizes.
pub fn run(data_root: &Path) -> Outcome {
run_sized(data_root, FILE_BYTES, MIN_FREE_BYTES, SYNC_WINDOW)
}
/// The test at any size — what the unit test calls with a few MiB.
pub fn run_sized(data_root: &Path, file_bytes: u64, min_free: u64, sync_window: Duration) -> Outcome {
let free = match crate::statvfs(data_root) {
Ok(f) => f.available_bytes,
Err(why) => return Outcome::Failed(why),
};
if should_skip(free, min_free) {
return Outcome::Skipped { free, need: min_free };
}
let dir = data_root.join(DIR_NAME);
let path = dir.join("probe");
let result = measure(&dir, &path, file_bytes, sync_window);
// Leave nothing behind whichever way it went.
let _ = std::fs::remove_file(&path);
let _ = std::fs::remove_dir(&dir);
match result {
Ok(report) => Outcome::Done(report),
Err(why) => Outcome::Failed(why),
}
}
fn measure(dir: &Path, path: &Path, file_bytes: u64, sync_window: Duration) -> Result<Report, String> {
let started = Instant::now();
std::fs::create_dir_all(dir).map_err(|e| format!("mkdir {}: {e}", dir.display()))?;
// Not zeros: a filesystem or a virtual disk that compresses or
// deduplicates would report a write it never made. ONE buffer for the
// whole run — the write source, the O_DIRECT read target and the 4 KiB
// block all live in it (allocate once, reuse).
let mut buf = AlignedBuf::new(CHUNK, DIRECT_ALIGN);
buf.fill(&pattern(CHUNK));
// ── sequential write + fsync ───────────────────────────────────────────
let t = Instant::now();
let mut f = File::create(path).map_err(|e| format!("create: {e}"))?;
let mut left = file_bytes;
while left > 0 {
let n = (left as usize).min(CHUNK);
f.write_all(&buf.as_slice()[..n]).map_err(|e| format!("write: {e}"))?;
left -= n as u64;
}
f.sync_all().map_err(|e| format!("fsync: {e}"))?;
drop(f);
let seq_write_bps = file_bytes as f64 / t.elapsed().as_secs_f64().max(1e-9);
// ── sequential read, off the device ───────────────────────────────────
let (got, secs, read_how) = read_back(path, &mut buf, true)?;
if got != file_bytes {
return Err(format!("read back {got} of {file_bytes} bytes"));
}
let seq_read_bps = got as f64 / secs.max(1e-9);
// ── 4 KiB random overwrites, fsync each ────────────────────────────────
let block_src = pattern(BLOCK);
buf.as_mut_slice()[..BLOCK].copy_from_slice(&block_src);
let mut f = OpenOptions::new().write(true).open(path).map_err(|e| format!("reopen: {e}"))?;
let blocks = (file_bytes / BLOCK as u64).max(1);
let block = &buf.as_slice()[..BLOCK];
let mut rng = started.elapsed().as_nanos() as u64 | 1;
let mut ops = 0u64;
let t = Instant::now();
while t.elapsed() < sync_window {
// xorshift64: enough randomness to defeat any write-combining.
rng ^= rng << 13;
rng ^= rng >> 7;
rng ^= rng << 17;
let at = (rng % blocks) * BLOCK as u64;
f.seek(SeekFrom::Start(at)).map_err(|e| format!("seek: {e}"))?;
f.write_all(block).map_err(|e| format!("write 4k: {e}"))?;
f.sync_data().map_err(|e| format!("fdatasync: {e}"))?;
ops += 1;
}
let sync_ops_per_s = ops as f64 / t.elapsed().as_secs_f64().max(1e-9);
drop(f);
Ok(Report { seq_write_bps, seq_read_bps, read_how, sync_ops_per_s, elapsed: started.elapsed() })
}
/// Read `path` back sequentially into `buf`, past the page cache: `O_DIRECT`
/// when `try_direct` and the filesystem takes it, else drop the cached pages
/// (`fadvise DONTNEED`, after the caller's fsync they are clean) and read
/// buffered. Returns `(bytes, seconds, how)`; the clock starts after the
/// open so the number is the transfer, not the open.
pub fn read_back(path: &Path, buf: &mut AlignedBuf, try_direct: bool) -> Result<(u64, f64, ReadHow), String> {
let direct = if try_direct { OpenOptions::new().read(true).custom_flags(OFlags::DIRECT.bits() as i32).open(path).ok() } else { None };
let (mut f, how) = match direct {
Some(f) => (f, ReadHow::Direct),
None => {
let f = File::open(path).map_err(|e| format!("open: {e}"))?;
// Best effort: a filesystem that ignores the advice is one whose
// "read" is memory anyway, and the > 5 GB/s rule catches it.
let _ = fadvise(&f, 0, None, Advice::DontNeed);
(f, ReadHow::Dropped)
}
};
let t = Instant::now();
let mut got = 0u64;
loop {
let n = f.read(buf.as_mut_slice()).map_err(|e| format!("read ({}): {e}", how.as_str()))?;
if n == 0 {
break;
}
got += n as u64;
}
Ok((got, t.elapsed().as_secs_f64(), how))
}
/// A heap buffer whose start is aligned to `align` (a power of two): what
/// `O_DIRECT` needs and `Vec<u8>` cannot promise. Allocated once per run.
pub struct AlignedBuf {
raw: Vec<u8>,
at: usize,
len: usize,
}
impl AlignedBuf {
pub fn new(len: usize, align: usize) -> Self {
debug_assert!(align.is_power_of_two());
let raw = vec![0u8; len + align];
let at = raw.as_ptr().align_offset(align);
Self { raw, at, len }
}
pub fn fill(&mut self, src: &[u8]) {
let n = src.len().min(self.len);
self.as_mut_slice()[..n].copy_from_slice(&src[..n]);
}
pub fn as_slice(&self) -> &[u8] {
&self.raw[self.at..self.at + self.len]
}
pub fn as_mut_slice(&mut self) -> &mut [u8] {
&mut self.raw[self.at..self.at + self.len]
}
}
fn pattern(n: usize) -> Vec<u8> {
let mut v = Vec::with_capacity(n);
let mut x = 0x9E37_79B9_7F4A_7C15u64;
while v.len() < n {
x ^= x << 13;
x ^= x >> 7;
x ^= x << 17;
v.extend_from_slice(&x.to_le_bytes());
}
v.truncate(n);
v
}
// ── the latch ───────────────────────────────────────────────────────────────
/// One recorded run: what it measured, where, and when (Unix ms, the caller's
/// clock).
#[derive(Debug, Clone, PartialEq)]
pub struct Latched {
pub outcome: Outcome,
pub path: std::path::PathBuf,
pub at_unix_ms: i64,
}
/// **The latch.** A reader (an admin view, a twin election) asks what this
/// process's disk did at start-up; the answer is kept here. `None` means the
/// test has not finished — never a zero and never an empty [`Report`].
static LAST: std::sync::OnceLock<std::sync::Mutex<Option<Latched>>> = std::sync::OnceLock::new();
fn latch() -> &'static std::sync::Mutex<Option<Latched>> {
LAST.get_or_init(|| std::sync::Mutex::new(None))
}
/// Record the outcome of a run.
pub fn record(path: &Path, outcome: &Outcome, at_unix_ms: i64) {
if let Ok(mut slot) = latch().lock() {
*slot = Some(Latched { outcome: outcome.clone(), path: path.to_path_buf(), at_unix_ms });
}
}
/// The last outcome this process measured, or `None` if it has not finished
/// one. A lookup, never a run: asking must never start a disk test.
pub fn last() -> Option<Latched> {
latch().lock().ok().and_then(|slot| slot.clone())
}
/// **Run once on its own thread** and latch the answer; `now_unix_ms` stamps it
/// (the product's trusted clock). Returns whether the thread started. The
/// product decides WHETHER to call this (gunnar: only as PID 1).
pub fn spawn(data_root: std::path::PathBuf, thread_name: &str, now_unix_ms: fn() -> i64) -> bool {
// A32-14, stays a thread: the one disk speed test, run once and latched (the product calls it only as PID 1).
std::thread::Builder::new()
.name(thread_name.into())
.spawn(move || {
let outcome = std::panic::catch_unwind(|| run(&data_root)).unwrap_or_else(|_| Outcome::Failed("the disk test panicked".to_owned()));
record(&data_root, &outcome, now_unix_ms());
})
.is_ok()
}
#[cfg(test)]
mod tests {
use super::*;
use std::path::PathBuf;
#[test]
fn the_row_reads_as_one_line_and_a_page_cache_read_is_called_cached() {
let r = Report { seq_write_bps: 812e6, seq_read_bps: 1.9e9, read_how: ReadHow::Direct, sync_ops_per_s: 3100.4, elapsed: Duration::from_millis(1600) };
assert_eq!(r.row(), "seq w 812 MB/s \u{b7} r 1.9 GB/s \u{b7} 4k sync 3100 ops/s (1.6 s)");
let cached = Report { seq_read_bps: 7.2e9, ..r.clone() };
assert!(cached.read_was_cached());
assert_eq!(cached.row(), "seq w 812 MB/s \u{b7} r cached \u{b7} 4k sync 3100 ops/s (1.6 s)");
assert_eq!(fmt_rate(512e3), "512 kB/s");
assert!(!Outcome::Done(r).row().1);
assert_eq!(Outcome::Skipped { free: 800 << 20, need: 1 << 30 }.row(), ("skipped: 800M free (< 1.0G)".to_owned(), false));
assert_eq!(Outcome::Failed("write: No space left on device".into()).row(), ("failed: write: No space left on device".to_owned(), true));
}
#[test]
fn the_skip_rule_is_free_space_below_the_need() {
assert!(should_skip((1 << 30) - 1, 1 << 30));
assert!(!should_skip(1 << 30, 1 << 30));
assert!(!should_skip(u64::MAX, MIN_FREE_BYTES));
let dir = tempfile::tempdir().unwrap();
let out = run_sized(dir.path(), 4 << 20, u64::MAX, Duration::from_millis(1));
assert!(matches!(out, Outcome::Skipped { need: u64::MAX, .. }), "{out:?}");
assert!(!dir.path().join(DIR_NAME).exists());
}
#[test]
fn a_small_run_measures_three_numbers_and_leaves_nothing_behind() {
let dir = tempfile::tempdir().unwrap();
let out = run_sized(dir.path(), 4 << 20, 0, Duration::from_millis(50));
let Outcome::Done(r) = out else { panic!("{out:?}") };
assert!(r.seq_write_bps > 0.0 && r.seq_read_bps > 0.0 && r.sync_ops_per_s > 0.0, "{r:?}");
assert!(r.elapsed >= Duration::from_millis(50));
assert!(!dir.path().join(DIR_NAME).exists(), "the probe directory is removed");
let text = r.row();
assert!(text.starts_with("seq w ") && text.contains(" 4k sync "), "{text}");
}
fn fixture(dir: &Path) -> (PathBuf, u64) {
let path = dir.join("fixture");
let body = pattern(4 * (1 << 20) + 1234);
std::fs::write(&path, &body).unwrap();
File::open(&path).unwrap().sync_all().unwrap();
(path, body.len() as u64)
}
#[test]
fn the_read_leg_reads_the_fixture_back() {
let dir = tempfile::tempdir().unwrap();
let (path, len) = fixture(dir.path());
let mut buf = AlignedBuf::new(CHUNK, DIRECT_ALIGN);
assert_eq!(buf.as_slice().as_ptr() as usize % DIRECT_ALIGN, 0, "the buffer is 4 KiB-aligned");
let (got, secs, how) = read_back(&path, &mut buf, true).unwrap();
assert_eq!(got, len, "every byte, including the partial last block");
assert!(secs > 0.0);
let tail = &pattern(len as usize)[(len as usize) - 1234..];
assert_eq!(&buf.as_slice()[..1234], tail);
let fs_direct = OpenOptions::new().read(true).custom_flags(OFlags::DIRECT.bits() as i32).open(&path).is_ok();
assert_eq!(how, if fs_direct { ReadHow::Direct } else { ReadHow::Dropped });
let (got, _, how) = read_back(&path, &mut buf, false).unwrap();
assert_eq!((got, how), (len, ReadHow::Dropped));
assert!(read_back(Path::new("/nonexistent/korp/fixture"), &mut buf, false).is_err());
}
#[test]
fn a_missing_root_is_a_failure_not_a_panic_and_the_latch_starts_empty_then_keeps_the_run() {
let out = run_sized(Path::new("/nonexistent/korp"), 4 << 20, 0, Duration::from_millis(1));
assert!(matches!(out, Outcome::Failed(_)), "{out:?}");
assert!(last().is_none(), "not run yet is None, never a zero");
record(Path::new("/data"), &out, 42);
let l = last().unwrap();
assert_eq!((l.path.as_path(), l.at_unix_ms), (Path::new("/data"), 42));
assert_eq!(l.outcome, out);
}
}