indexbus_platform_ops/linux/

preflight.rs

//! Host preflight checks for latency-sensitive Linux deployments.

use std::path::{Path, PathBuf};

use crate::errors::{Error, Result};

#[derive(Debug, Default)]
/// A report returned by [`run`].
pub struct PreflightReport {
    /// Observations captured during the run (key/value pairs).
    pub observations: Vec<(String, String)>,
    /// Non-fatal findings.
    pub warnings: Vec<String>,
    /// Fatal findings.
    pub errors: Vec<String>,
}

impl PreflightReport {
    /// Returns `true` when the report contains no errors.
    pub fn is_ok(&self) -> bool {
        self.errors.is_empty()
    }
}

#[derive(Debug, Clone)]
/// Inputs to the preflight checker.
pub struct PreflightInput {
    /// SHM/mmap directory to validate (existence + writability).
    pub shm_dir: PathBuf,
    /// If `true`, treat findings as errors where possible.
    pub strict: bool,
}

/// Run host preflight checks and return a report.
pub fn run(input: &PreflightInput) -> Result<PreflightReport> {
    let mut report = PreflightReport::default();

    report.observations.push((
        "shm_dir".to_string(),
        format!(
            "{} (exists={})",
            input.shm_dir.display(),
            input.shm_dir.exists()
        ),
    ));

    // 1) SHM dir existence + basic writeability.
    if !input.shm_dir.exists() {
        report.errors.push(format!(
            "shm_dir does not exist: {}",
            input.shm_dir.display()
        ));
    } else if let Err(e) = write_test_dir(&input.shm_dir) {
        report.errors.push(format!(
            "shm_dir is not writable: {} ({e})",
            input.shm_dir.display()
        ));
    }

    // 2) Linux-specific latency hygiene checks.
    if let Err(e) = check_thp(
        input.strict,
        &mut report.observations,
        &mut report.warnings,
        &mut report.errors,
    ) {
        report.warnings.push(format!("THP check failed: {e}"));
    }

    if let Err(e) = check_cpu_governor(
        input.strict,
        &mut report.observations,
        &mut report.warnings,
        &mut report.errors,
    ) {
        report
            .warnings
            .push(format!("CPU governor check failed: {e}"));
    }

    if let Err(e) = check_memlock(
        input.strict,
        &mut report.observations,
        &mut report.warnings,
        &mut report.errors,
    ) {
        report.warnings.push(format!("memlock check failed: {e}"));
    }

    if let Err(e) = check_hugetlbfs(
        &input.shm_dir,
        &mut report.observations,
        &mut report.warnings,
    ) {
        report.warnings.push(format!("hugetlbfs check failed: {e}"));
    }

    if let Err(e) = check_irq_policy(input.strict, &mut report.observations, &mut report.warnings) {
        report
            .warnings
            .push(format!("IRQ/isolation check failed: {e}"));
    }

    Ok(report)
}

fn write_test_dir(dir: &Path) -> Result<()> {
    let path = dir.join(".indexbus_preflight_write_test");

    let mut f = std::fs::OpenOptions::new()
        .create(true)
        .write(true)
        .truncate(true)
        .open(&path)?;

    // hugetlbfs can return EINVAL for small writes. Use a set_len probe instead.
    // hugetlbfs magic: 0x958458f6
    const HUGETLBFS_MAGIC: u64 = 0x9584_58f6;

    let mut s: libc::statfs = unsafe { std::mem::zeroed() };
    let c_path = std::ffi::CString::new(dir.to_string_lossy().as_bytes())
        .map_err(|e| Error::msg(format!("invalid shm_dir path for CString: {e}")))?;
    let rc = unsafe { libc::statfs(c_path.as_ptr(), &mut s as *mut libc::statfs) };

    if rc == 0 && (s.f_type as u64) == HUGETLBFS_MAGIC {
        const TWO_MIB: u64 = 2 * 1024 * 1024;
        f.set_len(TWO_MIB)?;
    } else {
        use std::io::Write;
        f.write_all(b"ok\n")?;
    }

    drop(f);
    let _ = std::fs::remove_file(&path);
    Ok(())
}

fn read_first_line(path: impl AsRef<Path>) -> Result<String> {
    let s = std::fs::read_to_string(path.as_ref())?;
    Ok(s.lines().next().unwrap_or("").trim().to_string())
}

fn check_thp(
    strict: bool,
    observations: &mut Vec<(String, String)>,
    warnings: &mut Vec<String>,
    errors: &mut Vec<String>,
) -> Result<()> {
    // Common sysfs locations.
    let enabled =
        read_first_line("/sys/kernel/mm/transparent_hugepage/enabled").unwrap_or_default();
    let defrag = read_first_line("/sys/kernel/mm/transparent_hugepage/defrag").unwrap_or_default();

    observations.push((
        "thp".to_string(),
        format!("enabled='{enabled}' defrag='{defrag}'"),
    ));

    // Values look like: "always [madvise] never" with current in brackets.
    let enabled_is_bad = enabled.contains("[always]");
    let defrag_is_bad = defrag.contains("[always]");

    if enabled_is_bad || defrag_is_bad {
        let msg = format!(
            "THP looks enabled (enabled='{enabled}', defrag='{defrag}'); for low-latency consider: echo never | sudo tee /sys/kernel/mm/transparent_hugepage/enabled and /defrag"
        );
        if strict {
            errors.push(msg);
        } else {
            warnings.push(msg);
        }
    }

    Ok(())
}

fn check_cpu_governor(
    strict: bool,
    observations: &mut Vec<(String, String)>,
    warnings: &mut Vec<String>,
    errors: &mut Vec<String>,
) -> Result<()> {
    // Best-effort: read governor for cpu0.
    let gov_path = "/sys/devices/system/cpu/cpu0/cpufreq/scaling_governor";
    let gov = std::fs::read_to_string(gov_path).unwrap_or_default();
    let gov = gov.trim();

    if !gov.is_empty() {
        observations.push(("cpu_governor".to_string(), gov.to_string()));
    }

    if gov.is_empty() {
        warnings.push(
            "CPU governor not readable (cpufreq may be unavailable in this environment)"
                .to_string(),
        );
        return Ok(());
    }

    if gov != "performance" {
        let msg = format!(
            "CPU governor is '{gov}' (expected 'performance' for lowest jitter); consider: sudo cpupower frequency-set -g performance"
        );
        if strict {
            errors.push(msg);
        } else {
            warnings.push(msg);
        }
    }

    Ok(())
}

fn check_memlock(
    strict: bool,
    observations: &mut Vec<(String, String)>,
    warnings: &mut Vec<String>,
    errors: &mut Vec<String>,
) -> Result<()> {
    // If mlockall is enabled by an app, a low limit causes startup failure.
    // We only warn/error based on the current ulimit.
    let (cur, max) = rlimit_memlock()?;

    observations.push(("memlock".to_string(), format!("cur={cur} max={max}")));

    if cur < 1024 * 1024 * 64 {
        let msg = format!(
            "memlock ulimit is low (cur={cur}, max={max}); if using mlockall, set LimitMEMLOCK=infinity or `ulimit -l unlimited`"
        );
        if strict {
            errors.push(msg);
        } else {
            warnings.push(msg);
        }
    }

    Ok(())
}

fn rlimit_memlock() -> Result<(u64, u64)> {
    let mut lim = libc::rlimit {
        rlim_cur: 0,
        rlim_max: 0,
    };

    let rc = unsafe { libc::getrlimit(libc::RLIMIT_MEMLOCK, &mut lim as *mut libc::rlimit) };
    if rc != 0 {
        return Err(Error::msg(format!(
            "getrlimit(RLIMIT_MEMLOCK) failed: {}",
            std::io::Error::last_os_error()
        )));
    }

    Ok((lim.rlim_cur, lim.rlim_max))
}

fn check_hugetlbfs(
    shm_dir: &Path,
    observations: &mut Vec<(String, String)>,
    warnings: &mut Vec<String>,
) -> Result<()> {
    // If shm_dir is on hugetlbfs, you'll see a magic fs type. Best-effort via statfs.
    let mut s: libc::statfs = unsafe { std::mem::zeroed() };
    let cstr = std::ffi::CString::new(shm_dir.to_string_lossy().as_bytes())
        .map_err(|e| Error::msg(format!("invalid shm_dir path for CString: {e}")))?;
    let rc = unsafe { libc::statfs(cstr.as_ptr(), &mut s as *mut libc::statfs) };
    if rc != 0 {
        warnings.push(format!(
            "statfs({}) failed: {}",
            shm_dir.display(),
            std::io::Error::last_os_error()
        ));
        return Ok(());
    }

    // hugetlbfs magic: 0x958458f6
    let is_hugetlbfs = (s.f_type as u64) == 0x9584_58f6;

    observations.push(("hugetlbfs".to_string(), format!("{}", is_hugetlbfs)));
    if !is_hugetlbfs {
        let msg = format!(
            "shm_dir ({}) does not appear to be hugetlbfs (ok, but for tighter latency you may mount hugetlbfs and point shm_dir there)",
            shm_dir.display()
        );
        // Advisory.
        warnings.push(msg);
    }

    Ok(())
}

fn check_irq_policy(
    _strict: bool,
    observations: &mut Vec<(String, String)>,
    warnings: &mut Vec<String>,
) -> Result<()> {
    // 1) Kernel cmdline hints (best-effort). Optional if using cpusets/systemd CPUAffinity.
    let cmdline = std::fs::read_to_string("/proc/cmdline").unwrap_or_default();
    let has_isol = cmdline.contains("isolcpus=");
    let has_nohz = cmdline.contains("nohz_full=");
    let has_rcu = cmdline.contains("rcu_nocbs=");

    observations.push((
        "kernel_isolation_flags".to_string(),
        format!("isolcpus={has_isol} nohz_full={has_nohz} rcu_nocbs={has_rcu}"),
    ));

    if !has_isol && !has_nohz && !has_rcu {
        let msg = "no kernel CPU isolation flags detected (ok if using cpusets/CPUAffinity); ensure IRQ affinity keeps noisy interrupts off your isolated CPU set".to_string();
        // Advisory.
        warnings.push(msg);
    }

    // 2) irqbalance status (best-effort). irqbalance can move IRQs onto isolated CPUs.
    let mut irqbalance_active = None;

    if let Ok(out) = std::process::Command::new("systemctl")
        .args(["is-active", "irqbalance"])
        .output()
    {
        if out.status.success() {
            let s = String::from_utf8_lossy(&out.stdout);
            irqbalance_active = Some(s.trim() == "active");
        }
    }

    if irqbalance_active.is_none() {
        if let Ok(out) = std::process::Command::new("pgrep")
            .args(["-x", "irqbalance"])
            .output()
        {
            irqbalance_active = Some(out.status.success());
        }
    }

    if let Some(active) = irqbalance_active {
        observations.push(("irqbalance".to_string(), format!("active={active}")));
        if active {
            let msg = "irqbalance appears active; for isolated CPU sets ensure IRQ affinity and irqbalance config don't place interrupts on your isolated CPUs".to_string();
            warnings.push(msg);
        }
    } else {
        observations.push(("irqbalance".to_string(), "unknown".to_string()));
        warnings.push(
            "unable to determine irqbalance status (systemctl/pgrep unavailable?)".to_string(),
        );
    }

    Ok(())
}

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

    #[test]
    fn preflight_errors_when_shm_dir_missing() {
        let missing = std::path::PathBuf::from(format!(
            "/tmp/indexbus_missing_shm_{}_{}",
            std::process::id(),
            crate::time::monotonic_now_ns()
        ));

        let input = PreflightInput {
            shm_dir: missing,
            strict: true,
        };

        let report = run(&input).unwrap();
        assert!(!report.is_ok());
        assert!(report
            .errors
            .iter()
            .any(|e| e.contains("shm_dir does not exist")));
    }
}