1use std::fs::{File, OpenOptions};
31use std::io::{Read, Seek, SeekFrom, Write};
32use std::os::unix::fs::OpenOptionsExt;
33use std::path::{Path, PathBuf};
34use std::time::{Duration, Instant};
35
36use rustix::fs::{fadvise, Advice, OFlags};
37
38pub const ENV: &str = "GUNNAR_DISKTEST";
41pub const DIR_NAME: &str = ".korp-disktest";
43pub const FILE_BYTES: u64 = 256 << 20;
45pub const MIN_FREE_BYTES: u64 = 1 << 30;
48pub const SYNC_WINDOW: Duration = Duration::from_millis(500);
50pub const CACHED_BPS: f64 = 5e9;
52const CHUNK: usize = 4 << 20;
55const BLOCK: usize = 4096;
56const DIRECT_ALIGN: usize = 4096;
59
60#[derive(Debug, Clone, Copy, PartialEq, Eq)]
62pub enum ReadHow {
63 Direct,
65 Dropped,
68}
69
70impl ReadHow {
71 pub fn as_str(self) -> &'static str {
72 match self {
73 ReadHow::Direct => "direct",
74 ReadHow::Dropped => "dropped",
75 }
76 }
77}
78
79#[derive(Debug, Clone, PartialEq)]
81pub struct Report {
82 pub seq_write_bps: f64,
83 pub seq_read_bps: f64,
84 pub read_how: ReadHow,
86 pub sync_ops_per_s: f64,
87 pub elapsed: Duration,
88 pub file_bytes: u64,
93 pub write_elapsed: Duration,
94 pub read_elapsed: Duration,
95 pub sync_ops: u64,
96 pub sync_elapsed: Duration,
97}
98
99impl Report {
100 pub fn row(&self) -> String {
103 let read = if self.seq_read_bps > CACHED_BPS {
104 "cached".to_owned()
105 } else {
106 fmt_rate(self.seq_read_bps)
107 };
108 format!(
109 "seq w {} \u{b7} r {} \u{b7} 4k sync {:.0} ops/s ({:.1} s)",
110 fmt_rate(self.seq_write_bps),
111 read,
112 self.sync_ops_per_s,
113 self.elapsed.as_secs_f64()
114 )
115 }
116}
117
118pub fn fmt_rate(bps: f64) -> String {
120 if bps >= 1e9 {
121 format!("{:.1} GB/s", bps / 1e9)
122 } else if bps >= 1e6 {
123 format!("{:.0} MB/s", bps / 1e6)
124 } else {
125 format!("{:.0} kB/s", bps / 1e3)
126 }
127}
128
129#[derive(Debug, Clone, PartialEq)]
131pub enum Outcome {
132 Done(Report),
133 Skipped {
135 free: u64,
136 need: u64,
137 },
138 Failed(String),
139}
140
141impl Outcome {
142 pub fn row(&self) -> (String, bool) {
144 match self {
145 Outcome::Done(r) => (r.row(), false),
146 Outcome::Skipped { free, need } => (
147 format!(
148 "skipped: {} free (< {})",
149 crate::dashboard::fmt_bytes(*free),
150 crate::dashboard::fmt_bytes(*need)
151 ),
152 false,
153 ),
154 Outcome::Failed(why) => (format!("failed: {why}"), true),
155 }
156 }
157}
158
159pub fn should_skip(free: u64, need: u64) -> bool {
162 free < need
163}
164
165pub fn run(data_root: &Path) -> Outcome {
167 run_sized(data_root, FILE_BYTES, MIN_FREE_BYTES, SYNC_WINDOW)
168}
169
170pub fn run_sized(
172 data_root: &Path,
173 file_bytes: u64,
174 min_free: u64,
175 sync_window: Duration,
176) -> Outcome {
177 let free = match crate::dashboard::probe::disk(data_root) {
178 Some((_, avail)) => avail,
179 None => {
180 return Outcome::Failed(format!(
181 "statfs {}: not a mounted filesystem",
182 data_root.display()
183 ))
184 }
185 };
186 if should_skip(free, min_free) {
187 return Outcome::Skipped {
188 free,
189 need: min_free,
190 };
191 }
192 let dir = data_root.join(DIR_NAME);
193 let path = dir.join("probe");
194 let result = measure(&dir, &path, file_bytes, sync_window);
195 let _ = std::fs::remove_file(&path);
197 let _ = std::fs::remove_dir(&dir);
198 match result {
199 Ok(report) => Outcome::Done(report),
200 Err(why) => Outcome::Failed(why),
201 }
202}
203
204fn measure(
205 dir: &Path,
206 path: &Path,
207 file_bytes: u64,
208 sync_window: Duration,
209) -> Result<Report, String> {
210 let started = Instant::now();
211 std::fs::create_dir_all(dir).map_err(|e| format!("mkdir {}: {e}", dir.display()))?;
212 let mut buf = AlignedBuf::new(CHUNK, DIRECT_ALIGN);
217 buf.fill(&pattern(CHUNK));
218
219 let t = Instant::now();
221 let mut f = File::create(path).map_err(|e| format!("create: {e}"))?;
222 let mut left = file_bytes;
223 while left > 0 {
224 let n = (left as usize).min(CHUNK);
225 f.write_all(&buf.as_slice()[..n])
226 .map_err(|e| format!("write: {e}"))?;
227 left -= n as u64;
228 }
229 f.sync_all().map_err(|e| format!("fsync: {e}"))?;
230 drop(f);
231 let write_elapsed = t.elapsed();
232 let seq_write_bps = file_bytes as f64 / write_elapsed.as_secs_f64().max(1e-9);
233
234 let (got, secs, read_how) = read_back(path, &mut buf, true)?;
236 if got != file_bytes {
237 return Err(format!("read back {got} of {file_bytes} bytes"));
238 }
239 let seq_read_bps = got as f64 / secs.max(1e-9);
240
241 let block_src = pattern(BLOCK);
245 buf.as_mut_slice()[..BLOCK].copy_from_slice(&block_src);
246 let mut f = OpenOptions::new()
247 .write(true)
248 .open(path)
249 .map_err(|e| format!("reopen: {e}"))?;
250 let blocks = (file_bytes / BLOCK as u64).max(1);
251 let block = &buf.as_slice()[..BLOCK];
252 let mut rng = started.elapsed().as_nanos() as u64 | 1;
253 let mut ops = 0u64;
254 let t = Instant::now();
255 while t.elapsed() < sync_window {
256 rng ^= rng << 13;
258 rng ^= rng >> 7;
259 rng ^= rng << 17;
260 let at = (rng % blocks) * BLOCK as u64;
261 f.seek(SeekFrom::Start(at))
262 .map_err(|e| format!("seek: {e}"))?;
263 f.write_all(block).map_err(|e| format!("write 4k: {e}"))?;
264 f.sync_data().map_err(|e| format!("fdatasync: {e}"))?;
265 ops += 1;
266 }
267 let sync_elapsed = t.elapsed();
268 let sync_ops_per_s = ops as f64 / sync_elapsed.as_secs_f64().max(1e-9);
269 drop(f);
270
271 Ok(Report {
272 seq_write_bps,
273 seq_read_bps,
274 read_how,
275 sync_ops_per_s,
276 elapsed: started.elapsed(),
277 file_bytes,
278 write_elapsed,
279 read_elapsed: Duration::from_secs_f64(secs.max(0.0)),
280 sync_ops: ops,
281 sync_elapsed,
282 })
283}
284
285pub fn read_back(
291 path: &Path,
292 buf: &mut AlignedBuf,
293 try_direct: bool,
294) -> Result<(u64, f64, ReadHow), String> {
295 let direct = if try_direct {
296 OpenOptions::new()
297 .read(true)
298 .custom_flags(OFlags::DIRECT.bits() as i32)
299 .open(path)
300 .ok()
301 } else {
302 None
303 };
304 let (mut f, how) = match direct {
305 Some(f) => (f, ReadHow::Direct),
306 None => {
307 let f = File::open(path).map_err(|e| format!("open: {e}"))?;
308 let _ = fadvise(&f, 0, None, Advice::DontNeed);
311 (f, ReadHow::Dropped)
312 }
313 };
314 let t = Instant::now();
315 let mut got = 0u64;
316 loop {
317 let n = f
318 .read(buf.as_mut_slice())
319 .map_err(|e| format!("read ({}): {e}", how.as_str()))?;
320 if n == 0 {
321 break;
322 }
323 got += n as u64;
324 }
325 Ok((got, t.elapsed().as_secs_f64(), how))
326}
327
328pub struct AlignedBuf {
331 raw: Vec<u8>,
332 at: usize,
333 len: usize,
334}
335
336impl AlignedBuf {
337 pub fn new(len: usize, align: usize) -> Self {
338 debug_assert!(align.is_power_of_two());
339 let raw = vec![0u8; len + align];
340 let at = raw.as_ptr().align_offset(align);
341 Self { raw, at, len }
342 }
343 pub fn fill(&mut self, src: &[u8]) {
344 let n = src.len().min(self.len);
345 self.as_mut_slice()[..n].copy_from_slice(&src[..n]);
346 }
347 pub fn as_slice(&self) -> &[u8] {
348 &self.raw[self.at..self.at + self.len]
349 }
350 pub fn as_mut_slice(&mut self) -> &mut [u8] {
351 &mut self.raw[self.at..self.at + self.len]
352 }
353}
354
355fn pattern(n: usize) -> Vec<u8> {
356 let mut v = Vec::with_capacity(n);
357 let mut x = 0x9E37_79B9_7F4A_7C15u64;
358 while v.len() < n {
359 x ^= x << 13;
360 x ^= x >> 7;
361 x ^= x << 17;
362 v.extend_from_slice(&x.to_le_bytes());
363 }
364 v.truncate(n);
365 v
366}
367
368static LAST: std::sync::OnceLock<std::sync::Mutex<Option<Outcome>>> = std::sync::OnceLock::new();
378
379fn latch() -> &'static std::sync::Mutex<Option<Outcome>> {
380 LAST.get_or_init(|| std::sync::Mutex::new(None))
381}
382
383pub fn record(outcome: &Outcome) {
385 if let Ok(mut slot) = latch().lock() {
386 *slot = Some(outcome.clone());
387 }
388}
389
390pub fn last() -> Option<Outcome> {
394 latch().lock().ok().and_then(|slot| slot.clone())
395}
396
397pub fn announce(data_root: &Path, outcome: &Outcome) {
400 let (text, warn) = outcome.row();
401 let read = match outcome {
402 Outcome::Done(r) => r.read_how.as_str(),
403 _ => "-",
404 };
405 if warn {
406 tracing::warn!(event = "disktest", path = %data_root.display(), read, "{text}");
407 } else {
408 tracing::info!(event = "disktest", path = %data_root.display(), read, "{text}");
409 }
410}
411
412pub fn spawn(data_root: PathBuf) -> bool {
416 let explicit = std::env::var(ENV)
417 .ok()
418 .filter(|v| !v.is_empty() && v != "0");
419 let gated = appliance_pid1::net::is_init() && appliance_pid1::net::booted_by_korp_start();
420 let started = std::time::Instant::now();
421 if explicit.is_none() && !gated {
422 return false;
426 }
427 crate::readiness::begin("the start-up disk test");
429 let spawned = std::thread::Builder::new()
430 .name("gunnar-disktest".into())
431 .spawn(move || {
432 let outcome = std::panic::catch_unwind(|| run(&data_root))
433 .unwrap_or_else(|_| Outcome::Failed("the disk test panicked".to_owned()));
434 record(&outcome);
435 announce(&data_root, &outcome);
436 crate::readiness::mark_ready("the start-up disk test", started);
442 })
443 .is_ok();
444 if !spawned {
445 crate::readiness::mark_ready("a disk test that could not be spawned", started);
448 }
449 spawned
450}
451
452#[cfg(test)]
453mod tests {
454 use super::*;
455
456 #[test]
457 fn the_row_reads_as_one_line_and_a_page_cache_read_is_called_cached() {
458 let r = Report {
459 seq_write_bps: 812e6,
460 seq_read_bps: 1.9e9,
461 read_how: ReadHow::Direct,
462 sync_ops_per_s: 3100.4,
463 elapsed: Duration::from_millis(1600),
464 file_bytes: 256 << 20,
465 write_elapsed: Duration::from_millis(330),
466 read_elapsed: Duration::from_millis(141),
467 sync_ops: 1550,
468 sync_elapsed: Duration::from_millis(500),
469 };
470 assert_eq!(
471 r.row(),
472 "seq w 812 MB/s \u{b7} r 1.9 GB/s \u{b7} 4k sync 3100 ops/s (1.6 s)"
473 );
474 let cached = Report {
475 seq_read_bps: 7.2e9,
476 ..r.clone()
477 };
478 assert_eq!(
479 cached.row(),
480 "seq w 812 MB/s \u{b7} r cached \u{b7} 4k sync 3100 ops/s (1.6 s)"
481 );
482 assert_eq!(fmt_rate(512e3), "512 kB/s");
483 assert!(!Outcome::Done(r).row().1);
484 assert_eq!(
485 Outcome::Skipped {
486 free: 800 << 20,
487 need: 1 << 30
488 }
489 .row(),
490 ("skipped: 800M free (< 1.0G)".to_owned(), false)
491 );
492 assert_eq!(
493 Outcome::Failed("write: No space left on device".into()).row(),
494 ("failed: write: No space left on device".to_owned(), true)
495 );
496 }
497
498 #[test]
499 fn the_skip_rule_is_free_space_below_the_need() {
500 assert!(should_skip((1 << 30) - 1, 1 << 30));
501 assert!(!should_skip(1 << 30, 1 << 30));
502 assert!(!should_skip(u64::MAX, MIN_FREE_BYTES));
503 let dir = tempfile::tempdir().unwrap();
506 let out = run_sized(dir.path(), 4 << 20, u64::MAX, Duration::from_millis(1));
507 assert!(
508 matches!(out, Outcome::Skipped { need: u64::MAX, .. }),
509 "{out:?}"
510 );
511 assert!(!dir.path().join(DIR_NAME).exists());
512 }
513
514 #[test]
515 fn a_small_run_measures_three_numbers_and_leaves_nothing_behind() {
516 let dir = tempfile::tempdir().unwrap();
517 let out = run_sized(dir.path(), 4 << 20, 0, Duration::from_millis(50));
518 let Outcome::Done(r) = out else {
519 panic!("{out:?}")
520 };
521 assert!(
522 r.seq_write_bps > 0.0 && r.seq_read_bps > 0.0 && r.sync_ops_per_s > 0.0,
523 "{r:?}"
524 );
525 assert!(r.elapsed >= Duration::from_millis(50));
526 assert!(
527 !dir.path().join(DIR_NAME).exists(),
528 "the probe directory is removed"
529 );
530 let text = r.row();
531 assert!(
532 text.starts_with("seq w ") && text.contains(" 4k sync "),
533 "{text}"
534 );
535 }
536
537 fn fixture(dir: &Path) -> (PathBuf, u64) {
540 let path = dir.join("fixture");
541 let body = pattern(4 * (1 << 20) + 1234);
542 std::fs::write(&path, &body).unwrap();
543 File::open(&path).unwrap().sync_all().unwrap();
544 (path, body.len() as u64)
545 }
546
547 #[test]
548 fn the_read_leg_reads_the_fixture_back_off_the_device_with_o_direct() {
549 let base = std::env::var_os("GUNNAR_DISKTEST_FIXTURE_DIR").map(PathBuf::from);
552 let dir = match base {
553 Some(b) => tempfile::tempdir_in(b).unwrap(),
554 None => tempfile::tempdir().unwrap(),
555 };
556 let (path, len) = fixture(dir.path());
557 let mut buf = AlignedBuf::new(CHUNK, DIRECT_ALIGN);
558 assert_eq!(
559 buf.as_slice().as_ptr() as usize % DIRECT_ALIGN,
560 0,
561 "the buffer is 4 KiB-aligned"
562 );
563 assert_eq!(buf.as_slice().len(), CHUNK);
564 let (got, secs, how) = read_back(&path, &mut buf, true).unwrap();
565 assert_eq!(got, len, "every byte, including the partial last block");
566 assert!(secs > 0.0);
567 let tail = &pattern(len as usize)[(len as usize) - 1234..];
570 assert_eq!(&buf.as_slice()[..1234], tail);
571 let fs_direct = OpenOptions::new()
572 .read(true)
573 .custom_flags(OFlags::DIRECT.bits() as i32)
574 .open(&path)
575 .is_ok();
576 assert_eq!(
577 how,
578 if fs_direct {
579 ReadHow::Direct
580 } else {
581 ReadHow::Dropped
582 },
583 "O_DIRECT when the filesystem takes it, the fallback when not"
584 );
585 }
586
587 #[test]
588 fn the_read_leg_falls_back_to_dropping_the_cache_and_reading_buffered() {
589 let dir = tempfile::tempdir().unwrap();
590 let (path, len) = fixture(dir.path());
591 let mut buf = AlignedBuf::new(CHUNK, DIRECT_ALIGN);
592 let (got, _, how) = read_back(&path, &mut buf, false).unwrap();
593 assert_eq!(got, len);
594 assert_eq!(how, ReadHow::Dropped);
595 assert!(
596 read_back(Path::new("/nonexistent/korp/fixture"), &mut buf, false).is_err(),
597 "a missing file is an error, not a panic"
598 );
599 }
600
601 #[test]
602 fn a_full_run_names_how_it_read() {
603 let dir = tempfile::tempdir().unwrap();
604 let out = run_sized(dir.path(), 4 << 20, 0, Duration::from_millis(20));
605 let Outcome::Done(r) = out else {
606 panic!("{out:?}")
607 };
608 assert!(matches!(r.read_how, ReadHow::Direct | ReadHow::Dropped));
609 assert!(r.seq_read_bps > 0.0);
610 }
611
612 #[test]
613 fn a_missing_root_is_a_failure_row_not_a_panic() {
614 let out = run_sized(
615 Path::new("/nonexistent/korp"),
616 4 << 20,
617 0,
618 Duration::from_millis(1),
619 );
620 assert!(matches!(out, Outcome::Failed(_)), "{out:?}");
621 }
622}
623
624#[cfg(test)]
637mod noise {
638 use super::*;
639
640 #[test]
641 #[ignore = "a measurement: minutes, and gigabytes written"]
642 fn the_spread_of_ten_runs_on_one_idle_disk() {
643 let runs: usize = std::env::var("GUNNAR_DISKTEST_RUNS")
644 .ok()
645 .and_then(|v| v.parse().ok())
646 .unwrap_or(10);
647 let dir = match std::env::var_os("GUNNAR_DISKTEST_FIXTURE_DIR") {
648 Some(b) => tempfile::tempdir_in(PathBuf::from(b)).unwrap(),
649 None => tempfile::tempdir().unwrap(),
650 };
651 let mut w = Vec::with_capacity(runs);
652 let mut r = Vec::with_capacity(runs);
653 let mut s = Vec::with_capacity(runs);
654 for i in 0..runs {
655 match run_sized(dir.path(), FILE_BYTES, 0, SYNC_WINDOW) {
656 Outcome::Done(rep) => {
657 println!("run {i:2} {}", rep.row());
658 w.push(rep.seq_write_bps);
659 r.push(rep.seq_read_bps);
660 s.push(rep.sync_ops_per_s);
661 }
662 other => panic!("run {i}: {other:?}"),
663 }
664 }
665 for (name, mut v) in [("seq write", w), ("seq read", r), ("4k sync", s)] {
666 v.sort_by(|a, b| a.partial_cmp(b).unwrap());
667 let median = v[v.len() / 2];
668 let (lo, hi) = (v[0], v[v.len() - 1]);
669 println!(
670 "{name:<10} median {median:>14.0} min {lo:>14.0} max {hi:>14.0} \
671 spread {:>6.1}% of median",
672 (hi - lo) / median * 100.0
673 );
674 }
675 }
676}