flowrs-engine/crates/flowrs-core/tests/perf_tiers.rs

//! The partitioner's measured tiers against the harness that measured them.
//!
//! `codegen/perf_tiers.rs` is generated by tools/perfbench from its newest results CSV, and
//! `codegen::partition` takes two latency classes' tails from it (a call under a static
//! schedule, and the generated ROS 2 node's publish) instead of constants. These tests read
//! that CSV themselves, with their own parser and their own rounding, so a hand-edited table,
//! a table left behind by a newer run, or a partitioner that stopped using the table fails.
//!
//! The stated tolerance: a tier's recorded measurement equals the CSV's largest accepted
//! p99.9 exactly, and its tail is that value rounded up to two significant figures, which is
//! never below the measurement and at most 10% above it.

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

use flowrs_core::codegen::partition::{
    CALL_TAIL_NS, DDS_PUBLISH_NS, LOOP_MARGIN, LatencyClass, Options, analyze,
};
use flowrs_core::codegen::perf_tiers::{CALL, DDS_PUBLISH, SOURCE_CSV, Tier};
use flowrs_core::graph::WireMode;
use flowrs_core::graph::ir::Graph;
use flowrs_core::graph::placement::KEY;

type Row = BTreeMap<String, String>;

fn results_dir() -> PathBuf {
    Path::new(env!("CARGO_MANIFEST_DIR")).join("../../tools/perfbench/results")
}

/// RFC 4180 records (quoted fields may hold commas, quotes and newlines).
fn records(text: &str) -> Vec<Vec<String>> {
    let mut out = Vec::new();
    let (mut rec, mut field, mut quoted) = (Vec::new(), String::new(), false);
    let mut it = text.chars().peekable();
    while let Some(c) = it.next() {
        match (quoted, c) {
            (true, '"') if it.peek() == Some(&'"') => {
                field.push('"');
                it.next();
            }
            (true, '"') => quoted = false,
            (true, c) => field.push(c),
            (false, '"') => quoted = true,
            (false, ',') => rec.push(std::mem::take(&mut field)),
            (false, '\n') => {
                rec.push(std::mem::take(&mut field));
                out.push(std::mem::take(&mut rec));
            }
            (false, '\r') => {}
            (false, c) => field.push(c),
        }
    }
    if !field.is_empty() || !rec.is_empty() {
        rec.push(field);
        out.push(rec);
    }
    out
}

fn rows(path: &Path) -> Vec<Row> {
    let text = std::fs::read_to_string(path).unwrap_or_else(|e| panic!("{}: {e}", path.display()));
    let mut recs = records(&text).into_iter();
    let head = recs.next().expect("a header");
    recs.filter(|r| r.len() == head.len())
        .map(|r| head.iter().cloned().zip(r).collect())
        .collect()
}

/// The newest results CSV: the one whose rows carry the latest `date_utc`; the file name
/// breaks ties.
fn newest() -> (String, Vec<Row>) {
    let mut best: Option<(String, String, Vec<Row>)> = None;
    for e in std::fs::read_dir(results_dir()).expect("tools/perfbench/results exists") {
        let p = e.expect("a directory entry").path();
        if p.extension().and_then(|x| x.to_str()) != Some("csv") {
            continue;
        }
        let name = p.file_name().unwrap().to_string_lossy().into_owned();
        let t = rows(&p);
        let date = t
            .iter()
            .filter_map(|r| r.get("date_utc").cloned())
            .max()
            .unwrap_or_default();
        let rank = format!("{date}|{name}");
        if best.as_ref().is_none_or(|(b, _, _)| rank > *b) {
            best = Some((rank, name, t));
        }
    }
    let (_, name, t) = best.expect("at least one results CSV");
    (name, t)
}

fn key(r: &Row) -> String {
    ["family", "bench", "dataset", "contender", "op"]
        .map(|k| r.get(k).map_or("", String::as_str))
        .join("|")
}

/// The largest p99.9 over the accepted rows of `row` (over all of them when none was
/// accepted), ns, rounded up to a whole ns.
fn measured(t: &[Row], row: &str) -> u64 {
    let all: Vec<&Row> = t.iter().filter(|r| key(r) == row).collect();
    assert!(!all.is_empty(), "the CSV has no row {row}");
    let accepted: Vec<&Row> = all
        .iter()
        .copied()
        .filter(|r| r.get("accepted").is_some_and(|a| a == "1"))
        .collect();
    let pool = if accepted.is_empty() { all } else { accepted };
    pool.iter()
        .map(|r| {
            r["p999"]
                .parse::<f64>()
                .unwrap_or_else(|_| panic!("{row}: p999 {:?}", r["p999"]))
                .ceil() as u64
        })
        .max()
        .unwrap()
}

/// Rounded up to two significant figures.
fn two_figures_up(x: u64) -> u64 {
    let digits = x.checked_ilog10().map_or(1, |d| d + 1);
    if digits <= 2 {
        return x;
    }
    let unit = 10u64.pow(digits - 2);
    x.div_ceil(unit) * unit
}

#[test]
fn the_tier_table_is_generated_from_the_newest_csv() {
    let (name, t) = newest();
    assert_eq!(
        SOURCE_CSV, name,
        "codegen/perf_tiers.rs was generated from {SOURCE_CSV}, but the newest CSV in \
         tools/perfbench/results is {name}: re-run tools/perfbench/run.sh, which regenerates it"
    );
    for (what, tier) in [("CALL", CALL), ("DDS_PUBLISH", DDS_PUBLISH)] {
        let m = measured(&t, tier.row);
        assert_eq!(
            tier.measured_p999_ns, m,
            "{what}: the table records {} ns, the CSV's largest accepted p99.9 is {m} ns",
            tier.measured_p999_ns
        );
        // The tolerance: never below the measurement, at most 10% above it.
        assert!(tier.tail_ns >= m, "{what}: tail {} below {m}", tier.tail_ns);
        assert!(
            tier.tail_ns * 10 <= m * 11 + 9,
            "{what}: tail {} more than 10% above {m}",
            tier.tail_ns
        );
        assert_eq!(tier.tail_ns, two_figures_up(m), "{what}");
    }
}

#[test]
fn rounding_up_to_two_figures() {
    for (x, want) in [
        (0, 0),
        (7, 7),
        (99, 99),
        (100, 100),
        (101, 110),
        (4_321, 4_400),
    ] {
        assert_eq!(two_figures_up(x), want, "{x}");
    }
}

/// Host-side nodes publishing into ROS, a call edge beside it, and a loop on the host.
fn graph(period_ms: u64) -> Graph {
    let mut g = Graph::new();
    let set = |g: &mut Graph, n, k: &str, v: &str| {
        g.get_node_mut(n)
            .unwrap()
            .properties
            .insert(k.into(), v.into());
    };
    let value = g.add_node("Control", 0.0, 0.0);
    set(&mut g, value, "value", "12");
    let print = g.add_node("PrintLn", 0.0, 0.0);
    g.add_wire(value, "value", print, "input", WireMode::Copy)
        .unwrap();
    let publish = g.add_node("ros.publish", 0.0, 0.0);
    for (k, v) in [
        ("topic", "ratio"),
        ("type", "std_msgs/msg/Float64"),
        ("field", "data"),
        (KEY, "ros(/robot)"),
    ] {
        set(&mut g, publish, k, v);
    }
    g.add_wire(value, "value", publish, "value", WireMode::Copy)
        .unwrap();
    let timer = g.add_node("embedded.timer.interval", 0.0, 0.0);
    set(&mut g, timer, "period_ms", &period_ms.to_string());
    g
}

/// End to end: the CSV's measurement is the tail `analyze` gives a call edge and a DDS
/// publication, and whether a loop on the host is refused follows from it.
#[test]
fn the_partitioner_costs_its_edges_with_the_measured_tiers() {
    let (_, t) = newest();
    let call = two_figures_up(measured(&t, CALL.row));
    let dds = two_figures_up(measured(&t, DDS_PUBLISH.row));
    assert_eq!(CALL_TAIL_NS, call);
    assert_eq!(DDS_PUBLISH_NS, dds);
    let a = analyze(&graph(1), &Options::default()).unwrap();
    assert!(a.problems.is_empty(), "{:?}", a.problems);
    let tail = |class: LatencyClass| {
        a.edges
            .iter()
            .find(|e| e.class == class)
            .unwrap_or_else(|| panic!("no {class} edge in {:?}", a.edges))
            .tail_ns()
    };
    assert_eq!(tail(LatencyClass::Call), call);
    assert_eq!(tail(LatencyClass::DdsPublish), dds);
    assert_eq!(LatencyClass::Call.wire_ns(1 << 20), call);

    // The host's loop crosses both: it is refused exactly when 1 ms is under 10x the larger
    // measured tail, and a loop slow enough for it passes.
    let worst = call.max(dds);
    assert_eq!(
        !a.violations.is_empty(),
        1_000_000 < LOOP_MARGIN * worst,
        "{:?}",
        a.violations
    );
    let enough = (LOOP_MARGIN * worst).div_ceil(1_000_000).max(1);
    let a = analyze(&graph(enough), &Options::default()).unwrap();
    assert!(a.violations.is_empty(), "{:?}", a.violations);
}

/// The table's own shape: the rows it names are the harness's.
#[test]
fn the_tiers_name_the_harness_rows() {
    let rows: [Tier; 2] = [CALL, DDS_PUBLISH];
    assert_eq!(rows[0].row, "chain|chain16|msg64|flowrs-static-call|pass");
    assert_eq!(
        rows[1].row,
        "ros|one-way|sensor_msgs/Imu|flowrs-ros2edge-rustdds|send-call"
    );
}