//! RFC 004 spinning-worker tests.
//!
//! The futex scheduler-park path is NOT loom-modelled (sync_shim.rs: the full
//! runtime — context switches, futexes, TLS — is never run under cfg(loom)),
//! so the Dekker submit/park ordering is guarded here instead, by driving the
//! one true deadlock state and catching it as a timeout rather than a hung
//! binary: every scheduler parked on the futex while runnable work exists.
//!
//! All tests use a NO-IO runtime (pure message passing) so the idle path is
//! the bare `_` arm RFC 004 replaces — the futex park — not the timer/io arms,
//! which keep their own wake sources. A lost wakeup therefore manifests as a
//! permanent stall: a `recv` that never returns, an `rt.run` that never
//! reaches AllDone. The watchdog turns that into a failed assertion.

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

/// Run `body` on a side thread; fail (rather than hang the test binary) if it
/// does not finish within `secs`. A hang here means a scheduler parked on the
/// futex with work outstanding and nothing woke it — exactly the RFC 004
/// failure mode. The leaked thread is acceptable: the process exits non-zero.
fn run_with_timeout(secs: u64, body: impl FnOnce() + Send + 'static) {
    let (tx, rx) = std::sync::mpsc::channel();
    std::thread::spawn(move || {
        body();
        let _ = tx.send(());
    });
    if rx.recv_timeout(Duration::from_secs(secs)).is_err() {
        panic!(
            "RFC 004 deadlock: workload did not terminate within {secs}s \
             — a scheduler wakeup was lost (work stranded with all schedulers \
             parked on the futex)"
        );
    }
}

/// A single token ping-ponging between two actors for `rounds` rounds, on a
/// `threads`-wide no-io runtime with the given spin budget. With more threads
/// than active actors, the surplus schedulers go idle and park; every send to
/// the (parked) peer is a cross-thread wakeup. A lost wake stalls the bounce.
/// The post-run count confirms no message was silently dropped.
fn bounce(threads: usize, rounds: u64, budget: u64) {
    run_with_timeout(30, move || {
        let rt = init(Config::exact(threads).spin_budget_cycles(budget));
        let echoed = Arc::new(AtomicU64::new(0));
        let echoed_actor = echoed.clone();
        rt.run(move || {
            let (tx_ab, rx_ab) = channel::<u64>();
            let (tx_ba, rx_ba) = channel::<u64>();
            let echo = spawn(move || {
                for _ in 0..rounds {
                    let v = rx_ab.recv().expect("echo recv");
                    echoed_actor.fetch_add(1, Ordering::Relaxed);
                    tx_ba.send(v + 1).expect("echo send");
                }
            });
            for i in 0..rounds {
                tx_ab.send(i).expect("ping send");
                let r = rx_ba.recv().expect("ping recv");
                assert_eq!(r, i + 1, "token corrupted in flight");
            }
            echo.join().expect("echo join");
        });
        assert_eq!(
            echoed.load(Ordering::Relaxed),
            rounds,
            "messages were lost, not just delayed"
        );
    });
}

/// Tiny budget: schedulers park almost immediately, so essentially every
/// cross-thread handoff exercises the futex wake. Sharpest stress on the
/// submit-rule / lost-wakeup guards.
#[test]
fn bounce_forces_park_and_wake() {
    bounce(4, 20_000, 2_000);
}

/// Large budget: idle schedulers stay hot spinners, so the submit rule mostly
/// takes the `n_spinning > 0` skip and the last-spinner handoff (rule 4) does
/// the waking. Confirms that path also makes progress and stays correct.
#[test]
fn bounce_with_hot_spinners() {
    bounce(4, 20_000, 2_000_000);
}

/// Oversubscribed: more active actors than schedulers, so the run queue is
/// rarely empty and the park/spin/submit machinery is hammered from every
/// thread at once.
#[test]
fn bounce_oversubscribed() {
    bounce(2, 30_000, 5_000);
}

/// Bursty fork: a parent makes many children runnable at once (the motivating
/// fan-out), each sends one reply, the parent collects all of them. Exercises
/// the submit rule under a burst — the first job eats one wake, the woken
/// worker becomes a spinner and absorbs the rest — and confirms none are lost.
fn fanout(threads: usize, children: u64, budget: u64) {
    run_with_timeout(30, move || {
        let rt = init(Config::exact(threads).spin_budget_cycles(budget));
        rt.run(move || {
            let (tx, rx) = channel::<u64>();
            for c in 0..children {
                let tx = tx.clone();
                spawn(move || {
                    tx.send(c).expect("child send");
                });
            }
            drop(tx);
            let mut sum = 0u64;
            for _ in 0..children {
                sum += rx.recv().expect("collector recv");
            }
            assert_eq!(sum, (0..children).sum(), "a child reply was lost");
        });
    });
}

#[test]
fn fanout_burst_parks() {
    fanout(4, 256, 2_000);
}

#[test]
fn fanout_burst_hot() {
    fanout(4, 256, 2_000_000);
}

/// Shutdown guard: a mostly-idle spinning runtime must still terminate. Most
/// schedulers park on the futex with no work ever coming; only the AllDone
/// broadcast (`unpark_all_schedulers`) can release them. If that broadcast is
/// missing, `rt.run` never returns and the watchdog fires.
#[test]
fn idle_runtime_shuts_down() {
    run_with_timeout(15, || {
        let rt = init(Config::exact(8).spin_budget_cycles(2_000));
        rt.run(|| {
            // One trivial cross-thread handoff, then everyone goes idle and the
            // runtime must drain to AllDone through parked schedulers.
            let (tx, rx) = channel::<u64>();
            let h = spawn(move || {
                tx.send(42).expect("send");
            });
            assert_eq!(rx.recv().expect("recv"), 42);
            h.join().expect("join");
        });
    });
}

/// Repeated lifecycles: init→run→AllDone several times in one process, to
/// catch any park/broadcast state that fails to reset between runtimes.
#[test]
fn repeated_spinning_lifecycles() {
    run_with_timeout(30, || {
        for _ in 0..20 {
            let rt = init(Config::exact(4).spin_budget_cycles(2_000));
            let echoed = Arc::new(AtomicU64::new(0));
            let e = echoed.clone();
            rt.run(move || {
                let (tx, rx) = channel::<u64>();
                let worker = spawn(move || {
                    for _ in 0..500 {
                        let v = rx.recv().expect("recv");
                        e.fetch_add(v, Ordering::Relaxed);
                    }
                });
                for i in 0..500u64 {
                    tx.send(i).expect("send");
                }
                worker.join().expect("worker join");
            });
            assert_eq!(echoed.load(Ordering::Relaxed), (0..500u64).sum());
        }
    });
}

/// Shutdown with siblings *pinned* in the futex park — the test that actually
/// exercises the AllDone broadcast. A single long-running actor holds `live`
/// at 1 (CPU-bound, never sending or spawning) long enough for every other
/// scheduler to idle, spin out its budget, and commit to `futex_wait`. Because
/// nothing is ever enqueued during that window, the submit-rule wakes never
/// fire and those siblings stay parked; the final finalize enqueues no one, so
/// the live-check self-termination cannot save them either. Only
/// `unpark_all_schedulers` at AllDone can. Gut that broadcast and this hangs.
#[test]
fn shutdown_releases_pinned_parked_siblings() {
    run_with_timeout(15, || {
        let rt = init(Config::exact(8).spin_budget_cycles(2_000));
        rt.run(|| {
            // ~50ms of pure CPU: far longer than the µs it takes the other
            // seven schedulers to park, so they are provably in futex_wait
            // before this returns and live drops to 0.
            let start = std::time::Instant::now();
            let mut x = 0u64;
            while start.elapsed() < Duration::from_millis(50) {
                x = x.wrapping_mul(6364136223846793005).wrapping_add(1);
                std::hint::black_box(x);
            }
            std::hint::black_box(x);
        });
    });
}

/// cross-thread handoff — but the futex park path must still be inert and
/// correct (the lone thread is never parked while it has work). Pins the
/// degenerate cap=0 case.
#[test]
fn single_thread_spinning_is_inert() {
    run_with_timeout(15, || {
        let rt = init(Config::exact(1).spin_budget_cycles(2_000));
        let echoed = Arc::new(AtomicU64::new(0));
        let e = echoed.clone();
        rt.run(move || {
            let (tx, rx) = channel::<u64>();
            let worker = spawn(move || {
                for _ in 0..1_000 {
                    let v = rx.recv().expect("recv");
                    e.fetch_add(v, Ordering::Relaxed);
                }
            });
            for i in 0..1_000u64 {
                tx.send(i).expect("send");
            }
            worker.join().expect("join");
        });
        assert_eq!(echoed.load(Ordering::Relaxed), (0..1_000u64).sum());
    });
}

/// `spin_budget_cycles == 0` is the opt-out: the feature is fully off and the
/// runtime uses the historical `thread::sleep` idle path. Must still run a
/// cross-thread workload to completion (it always did) — this just guards that
/// the gating left the off-path intact.
#[test]
fn budget_zero_keeps_old_behaviour() {
    run_with_timeout(20, || {
        let rt = init(Config::exact(4).spin_budget_cycles(0));
        let sum = Arc::new(AtomicU64::new(0));
        let s = sum.clone();
        rt.run(move || {
            let (tx, rx) = channel::<u64>();
            let worker = spawn(move || {
                for _ in 0..1_000 {
                    s.fetch_add(rx.recv().expect("recv"), Ordering::Relaxed);
                }
            });
            for i in 0..1_000u64 {
                tx.send(i).expect("send");
            }
            worker.join().expect("join");
        });
        assert_eq!(sum.load(Ordering::Relaxed), (0..1_000u64).sum());
    });
}