//! RFC 004 spinning-worker measurement tool — the latency-vs-idle-CPU curve
//! over `spin_budget_cycles`.
//!
//! Spinning workers trade idle CPU for wake latency: an idle worker burns
//! cycles polling `queue_len` (up to `spin_budget_cycles`, capped at
//! `max_spinners` concurrent spinners) so that incoming work is claimed with
//! no futex syscall. `spin_budget_cycles = 0` is the feature fully off — the
//! historical `thread::sleep` park, exactly preserved. This binary measures
//! both sides of that trade so the sweep can pick the budget default (RFC 004
//! Q3) and the on/off posture (Q4).
//!
//! Workloads — each one is a fresh runtime per sample (no cross-contamination):
//!
//!   remote-wake   — a consumer blocks on recv (its worker idles); a producer
//!                   on another scheduler timestamps just before send. Latency
//!                   = send -> receipt, the quantity spinning is meant to cut.
//!                   A fixed gap (SMARM_SPIN_GAP_US) between rounds lets the
//!                   consumer re-idle each round. THE LATENCY TARGET.
//!   fork-join     — root forks FANOUT trivial actors and joins them (one
//!                   burst), gap, repeat. Latency = fork -> all-joined; stresses
//!                   the fan-out wake. Secondary latency target.
//!   half-load     — work dispatched every PERIOD_US for ITEMS items, so
//!                   workers idle a fraction of each period. THE COST TARGET:
//!                   cores-busy climbs with budget as spin fills the idle gaps.
//!   pure-idle     — one actor sleeps WINDOW_MS, nothing else runs. The cleanest
//!                   single picture of spin burn: cores-busy is ~0 at budget 0
//!                   and climbs toward max_spinners as the budget grows.
//!
//! Two axes, reported together per config:
//!   latency  — p50 / p90 / p99 / min / max over the POOLED rounds of every run
//!              (pooling gives a stabler tail than a median-of-percentiles).
//!   cost     — cores-busy = process CPU-seconds (getrusage RUSAGE_SELF, all
//!              runtime threads) / wall-seconds over the measured window. 1.0 =
//!              one core pegged; approaches max_spinners when every spinner is
//!              hot. Reported as the median over runs.
//!
//! The crossover the sweep reveals: spinning only wins when work arrives while
//! a worker is still inside its budget. With a fixed inter-arrival gap, that
//! means budgets whose spin-time (budget_cycles / TSC_freq) exceeds the gap
//! start catching work before the park; smaller budgets park and pay the full
//! wake. So GAP_US sets the knee — tune it against your box's TSC frequency.
//!
//! Knobs (env):
//!   SMARM_BENCH_THREADS   scheduler-count sweep            default "1 2"
//!   SMARM_BENCH_RUNS      runs per config (pooled/median)  default 3
//!   SMARM_SPIN_BUDGET     spin_budget_cycles sweep         default "0 1000 10000 100000 1000000"
//!   SMARM_MAX_SPINNERS    max_spinners sweep; "-" = runtime default (N/2)  default "-"
//!   SMARM_SPIN_GAP_US     idle gap between rounds/bursts    default 150
//!   SMARM_SPIN_ROUNDS     remote-wake rounds                default 100
//!   SMARM_SPIN_FANOUT     fork-join actors per burst        default 64
//!   SMARM_SPIN_BURSTS     fork-join bursts                  default 30
//!   SMARM_SPIN_ITEMS      half-load dispatch count          default 150
//!   SMARM_SPIN_PERIOD_US  half-load inter-dispatch period   default 300
//!   SMARM_SPIN_WINDOW_MS  pure-idle window                  default 25
//!
//! Output: house table + one greppable line per config:
//!   SPINCSV,<variant>,<workload>,<threads>,<budget>,<spinners>,<gap_us>,
//!           <n>,<p50_us>,<p90_us>,<p99_us>,<min_us>,<max_us>,<cores_busy>,<ops_s>
//! ("spinners" is the literal "default" when SMARM_MAX_SPINNERS is "-".)
//!
//! NOTE: on a 1-core box (CI, sandboxes) spinning is *pathological* — a hot
//! spinner starves the very thread that would enqueue work, so the optimal
//! budget there is 0 and you only ever see the cost, never the benefit. This
//! binary on one core validates the harness and catches gross pathologies; the
//! real latency-vs-idle-CPU curve comes from the many-core box.

use smarm::runtime::{init, Config};
use smarm::{channel, sleep, spawn};
use std::sync::{Arc, Mutex};
use std::time::{Duration, Instant};

// --------------------------------------------------------------------------
// env helpers (house style, matching rq_runtime.rs)
// --------------------------------------------------------------------------

fn variant() -> &'static str {
    if cfg!(feature = "rq-mpmc") {
        "rq-mpmc"
    } else if cfg!(feature = "rq-striped") {
        "rq-striped"
    } else {
        "rq-mutex"
    }
}

fn env_usize(key: &str, default: usize) -> usize {
    std::env::var(key).ok().and_then(|v| v.parse().ok()).unwrap_or(default)
}

fn env_u64(key: &str, default: u64) -> u64 {
    std::env::var(key).ok().and_then(|v| v.parse().ok()).unwrap_or(default)
}

fn env_threads() -> Vec<usize> {
    std::env::var("SMARM_BENCH_THREADS")
        .map(|v| v.split_whitespace().filter_map(|t| t.parse().ok()).collect())
        .unwrap_or_else(|_| vec![1, 2])
}

fn env_budgets() -> Vec<u64> {
    std::env::var("SMARM_SPIN_BUDGET")
        .map(|v| v.split_whitespace().filter_map(|t| t.parse().ok()).collect())
        .unwrap_or_else(|_| vec![0, 1_000, 10_000, 100_000, 1_000_000])
}

/// `None` means "leave the runtime default (N/2)"; encoded in env as "-".
fn env_spinners() -> Vec<Option<u32>> {
    std::env::var("SMARM_MAX_SPINNERS")
        .map(|v| {
            v.split_whitespace()
                .map(|t| if t == "-" { None } else { t.parse().ok() })
                .collect()
        })
        .unwrap_or_else(|_| vec![None])
}

fn make_config(threads: usize, budget: u64, spinners: Option<u32>) -> Config {
    let cfg = Config::exact(threads).spin_budget_cycles(budget);
    match spinners {
        Some(n) => cfg.max_spinners(n),
        None => cfg,
    }
}

// --------------------------------------------------------------------------
// CPU accounting — process-wide (all runtime threads) via getrusage
// --------------------------------------------------------------------------

/// Total CPU seconds (user + system) consumed by the whole process so far.
/// RUSAGE_SELF aggregates every thread, which is exactly what "how much did
/// the runtime burn" needs — the spinners are separate threads from the one
/// reading this.
fn cpu_seconds() -> f64 {
    // SAFETY: getrusage with RUSAGE_SELF and a valid out-pointer; we check the
    // return code and only read the (now-initialised) timeval fields.
    let mut ru: libc::rusage = unsafe { std::mem::zeroed() };
    let rc = unsafe { libc::getrusage(libc::RUSAGE_SELF, &mut ru) };
    assert_eq!(rc, 0, "getrusage(RUSAGE_SELF) failed");
    let tv = |t: libc::timeval| t.tv_sec as f64 + t.tv_usec as f64 * 1e-6;
    tv(ru.ru_utime) + tv(ru.ru_stime)
}

// --------------------------------------------------------------------------
// One sample = one fresh runtime run
// --------------------------------------------------------------------------

struct Sample {
    /// Per-event latencies in nanoseconds (empty for pure-idle).
    lat_ns: Vec<u64>,
    /// CPU-seconds / wall-seconds over the measured window.
    cores_busy: f64,
    /// Completed work units (for ops/s); 0 where not meaningful.
    ops: u64,
    /// Wall-seconds of the measured window (for ops/s).
    wall_s: f64,
}

/// Shared scratch the root closure writes and the caller reads after `run()`
/// returns. (`run` takes a `FnOnce` and yields no value, so we hand results
/// out through an Arc rather than a return.)
#[derive(Default)]
struct Scratch {
    lat_ns: Vec<u64>,
    cpu_delta: f64,
    wall_s: f64,
    ops: u64,
}

fn finish(scratch: Arc<Mutex<Scratch>>) -> Sample {
    let mut s = scratch.lock().unwrap();
    let cores_busy = if s.wall_s > 0.0 { s.cpu_delta / s.wall_s } else { 0.0 };
    Sample {
        lat_ns: std::mem::take(&mut s.lat_ns),
        cores_busy,
        ops: s.ops,
        wall_s: s.wall_s,
    }
}

fn remote_wake(threads: usize, budget: u64, spinners: Option<u32>, rounds: usize, gap: Duration) -> Sample {
    let rt = init(make_config(threads, budget, spinners));
    let scratch = Arc::new(Mutex::new(Scratch::default()));
    let out = scratch.clone();
    rt.run(move || {
        let (tx, rx) = channel::channel::<Instant>();
        let lat_sink = out.clone();
        // Consumer parks on recv; its worker idles (spins or parks per budget)
        // until each send wakes it. It records the wake latency itself.
        let consumer = spawn(move || {
            let mut lats = Vec::with_capacity(rounds);
            for _ in 0..rounds {
                let sent = rx.recv().expect("remote-wake recv");
                lats.push(sent.elapsed().as_nanos() as u64);
            }
            lat_sink.lock().unwrap().lat_ns = lats;
        });
        // Producer: sleep the gap (so the consumer's worker re-idles), then
        // timestamp and send. The send is the wake event under measurement.
        let cpu0 = cpu_seconds();
        let w0 = Instant::now();
        for _ in 0..rounds {
            sleep(gap);
            tx.send(Instant::now()).expect("remote-wake send");
        }
        let wall = w0.elapsed().as_secs_f64();
        let cpu = cpu_seconds() - cpu0;
        let _ = consumer.join();
        let mut s = out.lock().unwrap();
        s.cpu_delta = cpu;
        s.wall_s = wall;
        s.ops = rounds as u64;
    });
    finish(scratch)
}

fn fork_join(threads: usize, budget: u64, spinners: Option<u32>, fanout: usize, bursts: usize, gap: Duration) -> Sample {
    let rt = init(make_config(threads, budget, spinners));
    let scratch = Arc::new(Mutex::new(Scratch::default()));
    let out = scratch.clone();
    rt.run(move || {
        let mut lats = Vec::with_capacity(bursts);
        let cpu0 = cpu_seconds();
        let w0 = Instant::now();
        for _ in 0..bursts {
            // Gap lets the workers park between bursts, so the fan-out has real
            // wakes to do rather than landing on already-hot workers.
            sleep(gap);
            let t = Instant::now();
            let handles: Vec<_> = (0..fanout).map(|_| spawn(|| {})).collect();
            for h in handles {
                let _ = h.join();
            }
            lats.push(t.elapsed().as_nanos() as u64);
        }
        let wall = w0.elapsed().as_secs_f64();
        let cpu = cpu_seconds() - cpu0;
        let mut s = out.lock().unwrap();
        s.lat_ns = lats;
        s.cpu_delta = cpu;
        s.wall_s = wall;
        s.ops = (bursts * fanout) as u64;
    });
    finish(scratch)
}

fn half_load(threads: usize, budget: u64, spinners: Option<u32>, items: usize, period: Duration) -> Sample {
    let rt = init(make_config(threads, budget, spinners));
    let scratch = Arc::new(Mutex::new(Scratch::default()));
    let out = scratch.clone();
    rt.run(move || {
        let cpu0 = cpu_seconds();
        let w0 = Instant::now();
        // Dispatch one trivial actor every `period`. Between dispatches the
        // workers have nothing to do, so they idle for a fraction of each
        // period — that idle fraction is what spinning fills (or doesn't).
        let mut handles = Vec::with_capacity(items);
        for i in 0..items {
            let target = period * i as u32;
            let elapsed = w0.elapsed();
            if target > elapsed {
                sleep(target - elapsed);
            }
            handles.push(spawn(|| {}));
        }
        for h in handles {
            let _ = h.join();
        }
        let wall = w0.elapsed().as_secs_f64();
        let cpu = cpu_seconds() - cpu0;
        let mut s = out.lock().unwrap();
        s.cpu_delta = cpu;
        s.wall_s = wall;
        s.ops = items as u64;
    });
    finish(scratch)
}

fn pure_idle(threads: usize, budget: u64, spinners: Option<u32>, window: Duration) -> Sample {
    let rt = init(make_config(threads, budget, spinners));
    let scratch = Arc::new(Mutex::new(Scratch::default()));
    let out = scratch.clone();
    rt.run(move || {
        // One actor sleeps; every other worker thread has nothing to do and
        // spins-or-parks per budget for the whole window.
        let cpu0 = cpu_seconds();
        let w0 = Instant::now();
        sleep(window);
        let wall = w0.elapsed().as_secs_f64();
        let cpu = cpu_seconds() - cpu0;
        let mut s = out.lock().unwrap();
        s.cpu_delta = cpu;
        s.wall_s = wall;
        s.ops = 0;
    });
    finish(scratch)
}

// --------------------------------------------------------------------------
// stats
// --------------------------------------------------------------------------

/// Nearest-rank percentile over an already-sorted slice. `p` in [0, 100].
fn pct(sorted: &[u64], p: f64) -> u64 {
    if sorted.is_empty() {
        return 0;
    }
    let idx = ((p / 100.0) * (sorted.len() - 1) as f64).round() as usize;
    sorted[idx.min(sorted.len() - 1)]
}

fn median_f64(xs: &mut [f64]) -> f64 {
    if xs.is_empty() {
        return 0.0;
    }
    xs.sort_by(|a, b| a.partial_cmp(b).unwrap());
    xs[xs.len() / 2]
}

struct Aggregate {
    n: usize,
    p50_us: f64,
    p90_us: f64,
    p99_us: f64,
    min_us: f64,
    max_us: f64,
    cores_busy: f64,
    ops_s: f64,
}

/// Pool latencies across all runs (stabler tail); take the median cores-busy
/// and median ops/s across runs.
fn aggregate(samples: Vec<Sample>) -> Aggregate {
    let mut pooled: Vec<u64> = Vec::new();
    let mut cores: Vec<f64> = Vec::new();
    let mut ops_s: Vec<f64> = Vec::new();
    for s in &samples {
        pooled.extend_from_slice(&s.lat_ns);
        cores.push(s.cores_busy);
        if s.wall_s > 0.0 && s.ops > 0 {
            ops_s.push(s.ops as f64 / s.wall_s);
        }
    }
    pooled.sort_unstable();
    let ns_to_us = |ns: u64| ns as f64 / 1000.0;
    Aggregate {
        n: pooled.len(),
        p50_us: ns_to_us(pct(&pooled, 50.0)),
        p90_us: ns_to_us(pct(&pooled, 90.0)),
        p99_us: ns_to_us(pct(&pooled, 99.0)),
        min_us: pooled.first().map(|&v| ns_to_us(v)).unwrap_or(0.0),
        max_us: pooled.last().map(|&v| ns_to_us(v)).unwrap_or(0.0),
        cores_busy: median_f64(&mut cores),
        ops_s: median_f64(&mut ops_s),
    }
}

// --------------------------------------------------------------------------
// driver
// --------------------------------------------------------------------------

fn main() {
    let threads_sweep = env_threads();
    let budgets = env_budgets();
    let spinners_sweep = env_spinners();
    let runs = env_usize("SMARM_BENCH_RUNS", 3);
    let gap = Duration::from_micros(env_u64("SMARM_SPIN_GAP_US", 150));
    let rounds = env_usize("SMARM_SPIN_ROUNDS", 100);
    let fanout = env_usize("SMARM_SPIN_FANOUT", 64);
    let bursts = env_usize("SMARM_SPIN_BURSTS", 30);
    let items = env_usize("SMARM_SPIN_ITEMS", 150);
    let period = Duration::from_micros(env_u64("SMARM_SPIN_PERIOD_US", 300));
    let window = Duration::from_millis(env_u64("SMARM_SPIN_WINDOW_MS", 25));

    // (name, closure producing one Sample for a (threads,budget,spinners) point)
    type Workload = (&'static str, Box<dyn Fn(usize, u64, Option<u32>) -> Sample>);
    let workloads: Vec<Workload> = vec![
        ("remote-wake", Box::new(move |t, b, s| remote_wake(t, b, s, rounds, gap))),
        ("fork-join", Box::new(move |t, b, s| fork_join(t, b, s, fanout, bursts, gap))),
        ("half-load", Box::new(move |t, b, s| half_load(t, b, s, items, period))),
        ("pure-idle", Box::new(move |t, b, s| pure_idle(t, b, s, window))),
    ];

    let width = 122;
    println!("\n{}", "=".repeat(width));
    println!(
        "  spin sweep — variant={}, runs={runs} (latency pooled, cores median), gap={}µs",
        variant(),
        gap.as_micros()
    );
    println!("{}", "=".repeat(width));
    println!(
        "{:>11} | {:>3} | {:>9} | {:>8} | {:>6} | {:>9} | {:>9} | {:>9} | {:>9} | {:>9} | {:>7} | {:>10}",
        "workload", "thr", "budget", "spinners", "n", "p50 µs", "p90 µs", "p99 µs", "min µs", "max µs", "cores", "ops/s"
    );
    println!("{}", "-".repeat(width));

    for (name, run_one) in &workloads {
        for &threads in &threads_sweep {
            for &budget in &budgets {
                for &spinners in &spinners_sweep {
                    let samples: Vec<Sample> =
                        (0..runs).map(|_| run_one(threads, budget, spinners)).collect();
                    let a = aggregate(samples);
                    let sp_str = match spinners {
                        Some(n) => n.to_string(),
                        None => "default".to_string(),
                    };
                    // Latency columns are blank for the no-latency probe.
                    let has_lat = a.n > 0;
                    let lat = |v: f64| if has_lat { format!("{v:.2}") } else { "-".to_string() };
                    println!(
                        "{:>11} | {:>3} | {:>9} | {:>8} | {:>6} | {:>9} | {:>9} | {:>9} | {:>9} | {:>9} | {:>7.3} | {:>10}",
                        name,
                        threads,
                        budget,
                        sp_str,
                        a.n,
                        lat(a.p50_us),
                        lat(a.p90_us),
                        lat(a.p99_us),
                        lat(a.min_us),
                        lat(a.max_us),
                        a.cores_busy,
                        if a.ops_s > 0.0 { format!("{:.0}", a.ops_s) } else { "-".to_string() },
                    );
                    println!(
                        "SPINCSV,{},{},{},{},{},{},{},{:.3},{:.3},{:.3},{:.3},{:.3},{:.4},{:.0}",
                        variant(), name, threads, budget, sp_str, gap.as_micros(),
                        a.n, a.p50_us, a.p90_us, a.p99_us, a.min_us, a.max_us, a.cores_busy, a.ops_s
                    );
                }
            }
        }
    }
}