//! Runtime-level run-queue benches (ROADMAP_v0.5 phase 4; slot dimension
//! added for the v0.9 slot shootout, RFC 005).
//!
//! These exercise the WHOLE scheduler with the compile-time-selected queue,
//! so comparing variants means rebuilding per rq-* feature — that's what
//! scripts/bench_rq.sh does. The RFC 005 wake slot is a *runtime* Config
//! knob, so one binary benches both arms; the slot on/off sweep happens
//! inside this binary. Workloads:
//!
//!   yield-storm     — N actors yield K times each. Pure queue churn:
//!                     every yield is a push + pop with nothing in between.
//!                     Slot role: REGRESSION GUARD — yields never touch the
//!                     slot, any slot-on delta is pop-path overhead.
//!   ping-pong-pairs — P channel pairs, M roundtrips each. Park/unpark
//!                     latency through the queue.
//!                     Slot role: TARGET METRIC — every send-wake is an
//!                     actor-context unpark, the slot's home pattern.
//!   spawn-storm     — S spawn+join of trivial actors. Slab + queue + free
//!                     list under churn.
//!                     Slot role: NEUTRALITY CHECK — spawns bypass the slot
//!                     by policy; join wakes fire from finalize (scheduler
//!                     context), also shared.
//!
//! Knobs (env):
//!   SMARM_BENCH_THREADS  scheduler-count sweep, default "1 2 4"
//!   SMARM_BENCH_SLOT     wake-slot sweep, default "0 1" (off then on)
//!   SMARM_BENCH_RUNS     repetitions per config (median), default 5
//!   SMARM_BENCH_YIELD_ACTORS / _YIELDS      default 200 / 500
//!   SMARM_BENCH_PAIRS / _ROUNDTRIPS         default 32 / 1000
//!   SMARM_BENCH_SPAWNS                      default 5000
//!
//! Output: house table + one line per config:
//!   RQCSV,runtime,<variant>,<slot>,<bench>,<threads>,<work>,<median_us>,<ops_per_s>
//! plus, for slot-on configs, the RFC 005 observability counters:
//!   RQSLOT,<variant>,<bench>,<threads>,<slot_hits>,<slot_displacements>
//! (hits/displacements are taken from the same run as the median time).
//!
//! NOTE: a 1-core sandbox validates the harness, not the scaling story;
//! real curves come from the many-core box.

use smarm::runtime::{init, Config};
use std::sync::atomic::{AtomicU64, Ordering};
use std::sync::Arc;
use std::time::Instant;

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_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, 4])
}

fn env_slots() -> Vec<bool> {
    std::env::var("SMARM_BENCH_SLOT")
        .map(|v| {
            v.split_whitespace()
                .filter_map(|t| match t {
                    "0" | "off" | "false" => Some(false),
                    "1" | "on" | "true" => Some(true),
                    _ => None,
                })
                .collect()
        })
        .unwrap_or_else(|_| vec![false, true])
}

/// One measured run: (total_ops, elapsed_µs, slot_hits, slot_displacements).
struct Sample {
    ops: u64,
    us: u128,
    hits: u64,
    displacements: u64,
}

fn yield_storm(threads: usize, slot: bool, actors: usize, yields: usize) -> Sample {
    let rt = init(Config::exact(threads).wake_slot(slot));
    let start = Instant::now();
    rt.run(move || {
        let handles: Vec<_> = (0..actors)
            .map(|_| {
                smarm::spawn(move || {
                    for _ in 0..yields {
                        smarm::yield_now();
                    }
                })
            })
            .collect();
        for h in handles {
            let _ = h.join();
        }
    });
    let us = start.elapsed().as_micros();
    let stats = rt.stats();
    Sample {
        ops: (actors * yields) as u64,
        us,
        hits: stats.slot_hits(),
        displacements: stats.slot_displacements(),
    }
}

fn ping_pong_pairs(threads: usize, slot: bool, pairs: usize, roundtrips: usize) -> Sample {
    let rt = init(Config::exact(threads).wake_slot(slot));
    let total = Arc::new(AtomicU64::new(0));
    let t2 = total.clone();
    let start = Instant::now();
    rt.run(move || {
        let handles: Vec<_> = (0..pairs)
            .map(|_| {
                let total = t2.clone();
                smarm::spawn(move || {
                    let (tx_ab, rx_ab) = smarm::channel::channel::<u64>();
                    let (tx_ba, rx_ba) = smarm::channel::channel::<u64>();
                    let n = roundtrips as u64;
                    let echo = smarm::spawn(move || {
                        for _ in 0..n {
                            let v = rx_ab.recv().expect("echo recv");
                            tx_ba.send(v + 1).expect("echo send");
                        }
                    });
                    for i in 0..n {
                        tx_ab.send(i).expect("ping send");
                        let v = rx_ba.recv().expect("ping recv");
                        assert_eq!(v, i + 1);
                    }
                    let _ = echo.join();
                    total.fetch_add(n, Ordering::Relaxed);
                })
            })
            .collect();
        for h in handles {
            let _ = h.join();
        }
    });
    let us = start.elapsed().as_micros();
    let stats = rt.stats();
    Sample {
        ops: total.load(Ordering::Relaxed),
        us,
        hits: stats.slot_hits(),
        displacements: stats.slot_displacements(),
    }
}

fn spawn_storm(threads: usize, slot: bool, spawns: usize) -> Sample {
    let rt = init(Config::exact(threads).wake_slot(slot));
    let start = Instant::now();
    rt.run(move || {
        // Batches bound simultaneous liveness well below the slab cap.
        const BATCH: usize = 1024;
        let mut left = spawns;
        while left > 0 {
            let n = left.min(BATCH);
            let handles: Vec<_> = (0..n).map(|_| smarm::spawn(|| {})).collect();
            for h in handles {
                let _ = h.join();
            }
            left -= n;
        }
    });
    let us = start.elapsed().as_micros();
    let stats = rt.stats();
    Sample {
        ops: spawns as u64,
        us,
        hits: stats.slot_hits(),
        displacements: stats.slot_displacements(),
    }
}

fn main() {
    let threads_sweep = env_threads();
    let slot_sweep = env_slots();
    let runs = env_usize("SMARM_BENCH_RUNS", 5);
    let ya = env_usize("SMARM_BENCH_YIELD_ACTORS", 200);
    let yy = env_usize("SMARM_BENCH_YIELDS", 500);
    let pp = env_usize("SMARM_BENCH_PAIRS", 32);
    let pr = env_usize("SMARM_BENCH_ROUNDTRIPS", 1000);
    let ss = env_usize("SMARM_BENCH_SPAWNS", 5000);

    println!("\n{}", "=".repeat(106));
    println!(
        "  runtime benches — variant={}, runs={runs} (median)",
        variant()
    );
    println!("{}", "=".repeat(106));
    println!(
        "{:>16} | {:>7} | {:>4} | {:>16} | {:>10} | {:>14} | {:>10} | {:>9}",
        "bench", "threads", "slot", "work", "median µs", "ops/s", "slot hits", "displaced"
    );
    println!("{}", "-".repeat(106));

    type Bench = (&'static str, String, Box<dyn Fn(usize, bool) -> Sample>);
    let benches: Vec<Bench> = vec![
        (
            "yield-storm",
            format!("{ya}x{yy}"),
            Box::new(move |t, s| yield_storm(t, s, ya, yy)),
        ),
        (
            "ping-pong-pairs",
            format!("{pp}x{pr}"),
            Box::new(move |t, s| ping_pong_pairs(t, s, pp, pr)),
        ),
        (
            "spawn-storm",
            format!("{ss}"),
            Box::new(move |t, s| spawn_storm(t, s, ss)),
        ),
    ];

    for (name, work, f) in &benches {
        for &t in &threads_sweep {
            for &slot in &slot_sweep {
                let mut samples: Vec<Sample> = (0..runs).map(|_| f(t, slot)).collect();
                // Median by elapsed time; report the counters from that
                // same run so hits/time stay paired.
                samples.sort_unstable_by_key(|s| s.us);
                let mid = &samples[samples.len() / 2];
                let per_s = (mid.ops as f64 / (mid.us as f64 / 1e6)) as u64;
                let slot_str = if slot { "on" } else { "off" };
                println!(
                    "{:>16} | {:>7} | {:>4} | {:>16} | {:>10} | {:>14} | {:>10} | {:>9}",
                    name, t, slot_str, work, mid.us, per_s, mid.hits, mid.displacements
                );
                println!(
                    "RQCSV,runtime,{},{},{},{},{},{},{}",
                    variant(), slot_str, name, t, work, mid.us, per_s
                );
                if slot {
                    println!(
                        "RQSLOT,{},{},{},{},{}",
                        variant(), name, t, mid.hits, mid.displacements
                    );
                }
            }
        }
    }
}