nordisk-pairing 0.5.28

//! **The start-up disk test the election reads** (lane B3, A9 row 5): what `/data` writes,
//! measured once at the start on an idle box, so the weaker box becomes the twin by its
//! disk too, not by cores, RAM and room alone. Product-neutral: it measures the data root,
//! nothing of the product's.
//!
//! Gunnar's `disktest` is the pattern (its `hello` answers "ask again" until the test is done,
//! at most 300 s): a sequential write of [`FILE_BYTES`] in 4 MiB chunks, `fsync`ed, timed;
//! the file under `<data>/`[`DIR_NAME`] is removed afterwards. Never a panic and never a
//! reason the box does not start: a failure is a result that says why, and the election
//! then reads 0 MiB/s.

use std::io::Write as _;
use std::path::{Path, PathBuf};
use std::sync::{Arc, OnceLock};
use std::time::{Duration, Instant};

/// The test's directory under the data root; removed when the test is done.
pub const DIR_NAME: &str = ".nordisk-disktest";
/// What the test writes.
pub const FILE_BYTES: u64 = 64 << 20;
/// How long a box's `hello` is held for the test (the election compares its number).
pub const HOLD_HELLO: Duration = Duration::from_secs(300);
const CHUNK: usize = 4 << 20;

/// **Measure `/data` once**: [`FILE_BYTES`] written in 4 MiB chunks and `fsync`ed, the MiB/s
/// that took (at least 1 for any disk that finished). The file and its directory are removed
/// whatever happened.
pub fn measure(data_root: &Path) -> Result<u32, String> {
    let dir = data_root.join(DIR_NAME);
    let file = dir.join("seq-write");
    let run = || -> std::io::Result<Duration> {
        std::fs::create_dir_all(&dir)?;
        let mut f = std::fs::File::create(&file)?;
        let chunk = vec![0xA5u8; CHUNK];
        let t0 = Instant::now();
        let mut left = FILE_BYTES;
        while left > 0 {
            let n = left.min(CHUNK as u64) as usize;
            f.write_all(&chunk[..n])?;
            left -= n as u64;
        }
        f.sync_all()?;
        Ok(t0.elapsed())
    };
    let took = run();
    let _ = std::fs::remove_file(&file);
    let _ = std::fs::remove_dir(&dir);
    let took = took.map_err(|e| format!("the start-up disk test on {} failed: {e}", data_root.display()))?;
    let mib = (FILE_BYTES >> 20) as f64;
    Ok(((mib / took.as_secs_f64().max(1e-6)) as u64).clamp(1, u32::MAX as u64) as u32)
}

/// **The start-up disk test of this box**: started once, read by the box's `hello`.
#[derive(Debug)]
pub struct DiskTest {
    started: Instant,
    result: OnceLock<Result<u32, String>>,
}

impl DiskTest {
    /// A test not finished yet (the box side of a test, or before [`DiskTest::start`] lands).
    pub fn pending(started: Instant) -> DiskTest {
        DiskTest { started, result: OnceLock::new() }
    }

    /// **Start the test on `data_root`** on a thread of its own; returns at once. `say` gets
    /// its one line when it is done (the MiB/s, or why it failed).
    pub fn start(data_root: PathBuf, say: impl Fn(&str) + Send + 'static) -> Arc<DiskTest> {
        let t = Arc::new(DiskTest::pending(Instant::now()));
        let done = Arc::clone(&t);
        // A28 M18, stays a thread: one bounded measurement; its result, its failure and a start that fails are all answered.
        let spawned = std::thread::Builder::new().name("disktest".into()).spawn(move || {
            let r = measure(&data_root);
            say(&match &r {
                Ok(v) => format!("disktest {}: seq write {v} MiB/s ({} MiB, fsync'd) — the election reads it", data_root.display(), FILE_BYTES >> 20),
                Err(e) => format!("disktest: {e} — the election reads 0 MiB/s"),
            });
            done.finish(r);
        });
        if let Err(e) = spawned {
            t.finish(Err(format!("the start-up disk test could not start its thread: {e}")));
        }
        t
    }

    /// Record the result (the first one wins).
    pub fn finish(&self, r: Result<u32, String>) {
        let _ = self.result.set(r);
    }

    /// The result, once the test is done.
    pub fn result(&self) -> Option<&Result<u32, String>> {
        self.result.get()
    }

    /// **What the election reads**: the MiB/s, 0 while not done or failed.
    pub fn disk_mib_s(&self) -> u32 {
        match self.result.get() {
            Some(Ok(v)) => *v,
            _ => 0,
        }
    }

    /// **Whether `hello` may answer at `now`**: `Err` ("ask again") while the test runs and
    /// [`HOLD_HELLO`] has not passed since its start; `Ok` once it is done or the hold ran out.
    pub fn hold_hello(&self, now: Instant) -> Result<(), String> {
        if self.result.get().is_none() && now.saturating_duration_since(self.started) < HOLD_HELLO {
            return Err("the start-up disk test has not finished; the election compares its numbers — ask again".into());
        }
        Ok(())
    }
}

#[cfg(test)]
mod tests {
    use super::*;

    /// ★ **The election gets the disk's number, and hello waits for it** (lane B3, A9 row 5):
    /// a real measure on a tempdir is at least 1 MiB/s and leaves nothing behind; hello is held
    /// while the test runs, answered once it is done or the hold ran out; a failure reads 0.
    #[test]
    fn the_disk_test_feeds_the_election_and_holds_hello_until_done() {
        let root = std::env::temp_dir().join(format!("disktest-{}", std::process::id()));
        std::fs::create_dir_all(&root).unwrap();
        let v = measure(&root).unwrap();
        assert!(v >= 1, "{v}");
        assert!(!root.join(DIR_NAME).exists(), "nothing left behind");
        let t0 = Instant::now();
        let t = DiskTest::pending(t0);
        assert!(t.hold_hello(t0 + Duration::from_secs(1)).is_err(), "running: ask again");
        assert!(t.hold_hello(t0 + HOLD_HELLO).is_ok(), "the hold ran out");
        assert_eq!(t.disk_mib_s(), 0);
        t.finish(Ok(812));
        assert!(t.hold_hello(t0).is_ok());
        assert_eq!(t.disk_mib_s(), 812);
        assert!(measure(&root.join("absent/deeper\0bad")).is_err(), "a failure says why");
        let started = DiskTest::start(root.clone(), |_| {});
        for _ in 0..200 {
            if started.result().is_some() {
                break;
            }
            std::thread::sleep(Duration::from_millis(20));
        }
        assert!(started.disk_mib_s() >= 1);
        let _ = std::fs::remove_dir_all(&root);
    }
}