use std::fs::{File, OpenOptions};
use std::io::{Read, Seek, SeekFrom, Write};
use std::os::unix::fs::OpenOptionsExt;
use std::path::{Path, PathBuf};
use std::time::{Duration, Instant};
use rustix::fs::{fadvise, Advice, OFlags};
pub const ENV: &str = "GUNNAR_DISKTEST";
pub const DIR_NAME: &str = ".korp-disktest";
pub const FILE_BYTES: u64 = 256 << 20;
pub const MIN_FREE_BYTES: u64 = 1 << 30;
pub const SYNC_WINDOW: Duration = Duration::from_millis(500);
pub const CACHED_BPS: f64 = 5e9;
const CHUNK: usize = 4 << 20;
const BLOCK: usize = 4096;
const DIRECT_ALIGN: usize = 4096;
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum ReadHow {
Direct,
Dropped,
}
impl ReadHow {
pub fn as_str(self) -> &'static str {
match self {
ReadHow::Direct => "direct",
ReadHow::Dropped => "dropped",
}
}
}
#[derive(Debug, Clone, PartialEq)]
pub struct Report {
pub seq_write_bps: f64,
pub seq_read_bps: f64,
pub read_how: ReadHow,
pub sync_ops_per_s: f64,
pub elapsed: Duration,
pub file_bytes: u64,
pub write_elapsed: Duration,
pub read_elapsed: Duration,
pub sync_ops: u64,
pub sync_elapsed: Duration,
}
impl Report {
pub fn row(&self) -> String {
let read = if self.seq_read_bps > CACHED_BPS {
"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()
)
}
}
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)
}
}
#[derive(Debug, Clone, PartialEq)]
pub enum Outcome {
Done(Report),
Skipped {
free: u64,
need: u64,
},
Failed(String),
}
impl Outcome {
pub fn row(&self) -> (String, bool) {
match self {
Outcome::Done(r) => (r.row(), false),
Outcome::Skipped { free, need } => (
format!(
"skipped: {} free (< {})",
crate::dashboard::fmt_bytes(*free),
crate::dashboard::fmt_bytes(*need)
),
false,
),
Outcome::Failed(why) => (format!("failed: {why}"), true),
}
}
}
pub fn should_skip(free: u64, need: u64) -> bool {
free < need
}
pub fn run(data_root: &Path) -> Outcome {
run_sized(data_root, FILE_BYTES, MIN_FREE_BYTES, SYNC_WINDOW)
}
pub fn run_sized(
data_root: &Path,
file_bytes: u64,
min_free: u64,
sync_window: Duration,
) -> Outcome {
let free = match crate::dashboard::probe::disk(data_root) {
Some((_, avail)) => avail,
None => {
return Outcome::Failed(format!(
"statfs {}: not a mounted filesystem",
data_root.display()
))
}
};
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);
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()))?;
let mut buf = AlignedBuf::new(CHUNK, DIRECT_ALIGN);
buf.fill(&pattern(CHUNK));
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 write_elapsed = t.elapsed();
let seq_write_bps = file_bytes as f64 / write_elapsed.as_secs_f64().max(1e-9);
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);
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 {
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_elapsed = t.elapsed();
let sync_ops_per_s = ops as f64 / sync_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(),
file_bytes,
write_elapsed,
read_elapsed: Duration::from_secs_f64(secs.max(0.0)),
sync_ops: ops,
sync_elapsed,
})
}
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}"))?;
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))
}
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
}
static LAST: std::sync::OnceLock<std::sync::Mutex<Option<Outcome>>> = std::sync::OnceLock::new();
fn latch() -> &'static std::sync::Mutex<Option<Outcome>> {
LAST.get_or_init(|| std::sync::Mutex::new(None))
}
pub fn record(outcome: &Outcome) {
if let Ok(mut slot) = latch().lock() {
*slot = Some(outcome.clone());
}
}
pub fn last() -> Option<Outcome> {
latch().lock().ok().and_then(|slot| slot.clone())
}
pub fn announce(data_root: &Path, outcome: &Outcome) {
let (text, warn) = outcome.row();
let read = match outcome {
Outcome::Done(r) => r.read_how.as_str(),
_ => "-",
};
if warn {
tracing::warn!(event = "disktest", path = %data_root.display(), read, "{text}");
} else {
tracing::info!(event = "disktest", path = %data_root.display(), read, "{text}");
}
}
pub fn spawn(data_root: PathBuf) -> bool {
let explicit = std::env::var(ENV)
.ok()
.filter(|v| !v.is_empty() && v != "0");
let gated = appliance_pid1::net::is_init() && appliance_pid1::net::booted_by_korp_start();
let started = std::time::Instant::now();
if explicit.is_none() && !gated {
return false;
}
crate::readiness::begin("the start-up disk test");
let spawned = std::thread::Builder::new()
.name("gunnar-disktest".into())
.spawn(move || {
let outcome = std::panic::catch_unwind(|| run(&data_root))
.unwrap_or_else(|_| Outcome::Failed("the disk test panicked".to_owned()));
record(&outcome);
announce(&data_root, &outcome);
crate::readiness::mark_ready("the start-up disk test", started);
})
.is_ok();
if !spawned {
crate::readiness::mark_ready("a disk test that could not be spawned", started);
}
spawned
}
#[cfg(test)]
mod tests {
use super::*;
#[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),
file_bytes: 256 << 20,
write_elapsed: Duration::from_millis(330),
read_elapsed: Duration::from_millis(141),
sync_ops: 1550,
sync_elapsed: Duration::from_millis(500),
};
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_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_off_the_device_with_o_direct() {
let base = std::env::var_os("GUNNAR_DISKTEST_FIXTURE_DIR").map(PathBuf::from);
let dir = match base {
Some(b) => tempfile::tempdir_in(b).unwrap(),
None => 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"
);
assert_eq!(buf.as_slice().len(), CHUNK);
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
},
"O_DIRECT when the filesystem takes it, the fallback when not"
);
}
#[test]
fn the_read_leg_falls_back_to_dropping_the_cache_and_reading_buffered() {
let dir = tempfile::tempdir().unwrap();
let (path, len) = fixture(dir.path());
let mut buf = AlignedBuf::new(CHUNK, DIRECT_ALIGN);
let (got, _, how) = read_back(&path, &mut buf, false).unwrap();
assert_eq!(got, len);
assert_eq!(how, ReadHow::Dropped);
assert!(
read_back(Path::new("/nonexistent/korp/fixture"), &mut buf, false).is_err(),
"a missing file is an error, not a panic"
);
}
#[test]
fn a_full_run_names_how_it_read() {
let dir = tempfile::tempdir().unwrap();
let out = run_sized(dir.path(), 4 << 20, 0, Duration::from_millis(20));
let Outcome::Done(r) = out else {
panic!("{out:?}")
};
assert!(matches!(r.read_how, ReadHow::Direct | ReadHow::Dropped));
assert!(r.seq_read_bps > 0.0);
}
#[test]
fn a_missing_root_is_a_failure_row_not_a_panic() {
let out = run_sized(
Path::new("/nonexistent/korp"),
4 << 20,
0,
Duration::from_millis(1),
);
assert!(matches!(out, Outcome::Failed(_)), "{out:?}");
}
}
#[cfg(test)]
mod noise {
use super::*;
#[test]
#[ignore = "a measurement: minutes, and gigabytes written"]
fn the_spread_of_ten_runs_on_one_idle_disk() {
let runs: usize = std::env::var("GUNNAR_DISKTEST_RUNS")
.ok()
.and_then(|v| v.parse().ok())
.unwrap_or(10);
let dir = match std::env::var_os("GUNNAR_DISKTEST_FIXTURE_DIR") {
Some(b) => tempfile::tempdir_in(PathBuf::from(b)).unwrap(),
None => tempfile::tempdir().unwrap(),
};
let mut w = Vec::with_capacity(runs);
let mut r = Vec::with_capacity(runs);
let mut s = Vec::with_capacity(runs);
for i in 0..runs {
match run_sized(dir.path(), FILE_BYTES, 0, SYNC_WINDOW) {
Outcome::Done(rep) => {
println!("run {i:2} {}", rep.row());
w.push(rep.seq_write_bps);
r.push(rep.seq_read_bps);
s.push(rep.sync_ops_per_s);
}
other => panic!("run {i}: {other:?}"),
}
}
for (name, mut v) in [("seq write", w), ("seq read", r), ("4k sync", s)] {
v.sort_by(|a, b| a.partial_cmp(b).unwrap());
let median = v[v.len() / 2];
let (lo, hi) = (v[0], v[v.len() - 1]);
println!(
"{name:<10} median {median:>14.0} min {lo:>14.0} max {hi:>14.0} \
spread {:>6.1}% of median",
(hi - lo) / median * 100.0
);
}
}
}