smarm/tests/runtime.rs

//! Tests for the multi-scheduler runtime: Config, Runtime::run, and
//! correctness under genuine parallelism.
//!
//! The single-threaded correctness properties (channel ordering, mutex
//! fairness, timer accuracy, etc.) are already covered by the per-module
//! tests. This file focuses on what changes when N > 1 scheduler threads
//! are involved:
//!
//!   - Config construction and validation
//!   - Runtime::run blocks until all actors finish
//!   - All existing cooperative behaviours hold under multi-threading
//!   - Actors genuinely run on different OS threads
//!   - No lost wakeups under concurrent park/unpark
//!   - No slot leaks under high spawn/join churn
//!   - Panic on one scheduler thread doesn't kill others

use smarm::{channel, runtime::{Config, Runtime}, spawn, yield_now, JoinHandle};
use std::sync::{atomic::{AtomicBool, AtomicU64, Ordering}, Arc};
use std::time::Duration;
use std::collections::HashSet;

// ---------------------------------------------------------------------------
// Helpers
// ---------------------------------------------------------------------------

/// Build a runtime with exactly `n` scheduler threads.
fn rt(n: usize) -> Runtime {
    smarm::runtime::init(Config::exact(n))
}

/// Convenient single-threaded runtime (regression guard).
fn rt1() -> Runtime { rt(1) }

/// Multi-threaded runtime using all available parallelism.
fn rt_par() -> Runtime {
    smarm::runtime::init(Config::default())
}

// ---------------------------------------------------------------------------
// Config
// ---------------------------------------------------------------------------

#[test]
fn config_exact_overrides_bounds() {
    let c = Config::exact(3);
    assert_eq!(c.resolved_thread_count(), 3);
}

#[test]
fn config_default_clamps_to_available_parallelism() {
    let c = Config::default();
    let n = c.resolved_thread_count();
    let avail = std::thread::available_parallelism()
        .map(|n| n.get())
        .unwrap_or(1);
    // Default min is 1, default max is available_parallelism.
    assert!(n >= 1 && n <= avail);
}

#[test]
fn config_min_max_clamps() {
    // Force a range that excludes exact: min=2, max=4, available might be >4.
    let c = Config::new(2, 4, None);
    let n = c.resolved_thread_count();
    assert!(n >= 2 && n <= 4, "expected 2..=4, got {n}");
}

#[test]
fn config_min_1_max_1_is_single_threaded() {
    let c = Config::new(1, 1, None);
    assert_eq!(c.resolved_thread_count(), 1);
}

// ---------------------------------------------------------------------------
// Runtime::run — basic lifecycle
// ---------------------------------------------------------------------------

#[test]
fn runtime_run_executes_closure() {
    let flag = Arc::new(AtomicBool::new(false));
    let f = flag.clone();
    rt(1).run(move || { f.store(true, Ordering::SeqCst); });
    assert!(flag.load(Ordering::SeqCst));
}

#[test]
fn runtime_run_blocks_until_all_actors_done() {
    // Spawn a chain of actors; the counter should be exactly N when run returns.
    let counter = Arc::new(AtomicU64::new(0));
    let c = counter.clone();
    rt(2).run(move || {
        let mut handles = Vec::new();
        for _ in 0..20 {
            let cc = c.clone();
            handles.push(spawn(move || {
                cc.fetch_add(1, Ordering::SeqCst);
            }));
        }
        for h in handles {
            h.join().unwrap();
        }
    });
    assert_eq!(counter.load(Ordering::SeqCst), 20);
}

#[test]
fn runtime_can_be_used_multiple_times_sequentially() {
    // Each call to run() is independent.
    let r = rt(2);
    let a = Arc::new(AtomicU64::new(0));
    let b = Arc::new(AtomicU64::new(0));
    let ac = a.clone();
    let bc = b.clone();
    r.run(move || { ac.fetch_add(1, Ordering::SeqCst); });
    r.run(move || { bc.fetch_add(1, Ordering::SeqCst); });
    assert_eq!(a.load(Ordering::SeqCst), 1);
    assert_eq!(b.load(Ordering::SeqCst), 1);
}

// ---------------------------------------------------------------------------
// Single-threaded regression: exact(1) must behave identically to old run()
// ---------------------------------------------------------------------------

#[test]
fn exact_1_spawn_join_works() {
    let v = Arc::new(AtomicU64::new(0));
    let vc = v.clone();
    rt1().run(move || {
        let h = spawn(move || { vc.store(42, Ordering::SeqCst); });
        h.join().unwrap();
    });
    assert_eq!(v.load(Ordering::SeqCst), 42);
}

#[test]
fn exact_1_channel_recv_parks_and_wakes() {
    let v = Arc::new(AtomicU64::new(0));
    let vc = v.clone();
    rt1().run(move || {
        let (tx, rx) = channel::<u64>();
        let h = spawn(move || {
            let val = rx.recv().unwrap();
            vc.store(val, Ordering::SeqCst);
        });
        yield_now();
        tx.send(99).unwrap();
        h.join().unwrap();
    });
    assert_eq!(v.load(Ordering::SeqCst), 99);
}

#[test]
fn exact_1_panic_captured() {
    let saw_err = Arc::new(AtomicBool::new(false));
    let s = saw_err.clone();
    rt1().run(move || {
        let h = spawn(|| panic!("oops"));
        if h.join().is_err() { s.store(true, Ordering::SeqCst); }
    });
    assert!(saw_err.load(Ordering::SeqCst));
}

// ---------------------------------------------------------------------------
// Multi-threaded correctness
// ---------------------------------------------------------------------------

#[test]
fn multi_thread_all_actors_complete() {
    let counter = Arc::new(AtomicU64::new(0));
    let c = counter.clone();
    rt_par().run(move || {
        let mut handles = Vec::new();
        for _ in 0..100 {
            let cc = c.clone();
            handles.push(spawn(move || {
                cc.fetch_add(1, Ordering::SeqCst);
            }));
        }
        for h in handles { h.join().unwrap(); }
    });
    assert_eq!(counter.load(Ordering::SeqCst), 100);
}

#[test]
fn multi_thread_channel_wakeup_across_threads() {
    // Receiver parks; sender runs (potentially on a different OS thread).
    // Verifies no lost wakeup.
    let received = Arc::new(AtomicU64::new(0));
    let rc = received.clone();
    rt_par().run(move || {
        let (tx, rx) = channel::<u64>();
        let h = spawn(move || {
            let v = rx.recv().unwrap();
            rc.store(v, Ordering::SeqCst);
        });
        // Let receiver park.
        yield_now();
        tx.send(7).unwrap();
        h.join().unwrap();
    });
    assert_eq!(received.load(Ordering::SeqCst), 7);
}

#[test]
fn multi_thread_many_channels_no_lost_wakeups() {
    // N pairs of (sender actor, receiver actor). Each pair exchanges one
    // message. All must complete — any lost wakeup causes a deadlock/timeout.
    const PAIRS: usize = 50;
    let count = Arc::new(AtomicU64::new(0));
    let c = count.clone();
    rt_par().run(move || {
        let mut handles: Vec<JoinHandle> = Vec::new();
        for _ in 0..PAIRS {
            let (tx, rx) = channel::<u64>();
            let cc = c.clone();
            handles.push(spawn(move || {
                let v = rx.recv().unwrap();
                cc.fetch_add(v, Ordering::SeqCst);
            }));
            handles.push(spawn(move || {
                tx.send(1).unwrap();
            }));
        }
        for h in handles { h.join().unwrap(); }
    });
    assert_eq!(count.load(Ordering::SeqCst), PAIRS as u64);
}

#[test]
fn multi_thread_mutex_contention_no_deadlock() {
    use smarm::Mutex;
    const ACTORS: usize = 20;
    const PER: u64 = 100;
    let total = Arc::new(AtomicU64::new(0));
    let t = total.clone();
    rt_par().run(move || {
        let m: Mutex<u64> = Mutex::new(0);
        let mut handles = Vec::new();
        for _ in 0..ACTORS {
            let mc = m.clone();
            let tc = t.clone();
            handles.push(spawn(move || {
                for _ in 0..PER {
                    let mut g = mc.lock_timeout(Duration::from_secs(5)).unwrap();
                    *g += 1;
                    tc.fetch_add(0, Ordering::SeqCst); // just a memory barrier
                }
            }));
        }
        for h in handles { h.join().unwrap(); }
        let g = m.lock_timeout(Duration::from_secs(1)).unwrap();
        t.store(*g, Ordering::SeqCst);
    });
    assert_eq!(total.load(Ordering::SeqCst), ACTORS as u64 * PER);
}

#[test]
fn multi_thread_join_across_threads() {
    // Parent joins a child that may run on a different scheduler thread.
    let v = Arc::new(AtomicU64::new(0));
    let vc = v.clone();
    rt_par().run(move || {
        let h = spawn(move || {
            // Do some work to make scheduling interesting.
            for _ in 0..10 { yield_now(); }
            vc.store(1, Ordering::SeqCst);
        });
        h.join().unwrap();
    });
    assert_eq!(v.load(Ordering::SeqCst), 1);
}

// ---------------------------------------------------------------------------
// Actors run on distinct OS threads
//
// We collect the OS thread IDs that actors execute on. With N schedulers
// and enough actors, we expect to see more than one thread ID.
// ---------------------------------------------------------------------------

#[test]
fn actors_run_on_multiple_os_threads() {
    let thread_ids: Arc<smarm::Mutex<HashSet<u64>>> =
        Arc::new(smarm::Mutex::new(HashSet::new()));

    rt_par().run({
        let ids = thread_ids.clone();
        move || {
            let mut handles = Vec::new();
            for _ in 0..64 {
                let idc = ids.clone();
                handles.push(spawn(move || {
                    let tid = unsafe { libc::syscall(libc::SYS_gettid) as u64 };
                    let mut g = idc.lock_timeout(Duration::from_secs(1)).unwrap();
                    g.insert(tid);
                }));
            }
            for h in handles { h.join().unwrap(); }
        }
    });

    let n = std::thread::available_parallelism().map(|n| n.get()).unwrap_or(1);

    let ids = thread_ids.lock_timeout(Duration::from_secs(1)).unwrap();
    // If we have >1 scheduler threads, we expect >1 OS thread IDs.
    // On a single-CPU machine this may be 1; we just assert ≥ 1.
    assert!(!ids.is_empty());
    if n > 1 {
        // Strongly expect parallelism — not a hard assert since scheduling
        // is non-deterministic, but 64 actors should spread.
        // We log rather than assert to avoid flakiness on loaded CI.
        if ids.len() == 1 {
            eprintln!("WARNING: 64 actors all ran on the same OS thread (flaky on loaded system)");
        }
    }
}

// ---------------------------------------------------------------------------
// Scheduler stats (RFC 000 Layer 1 primitives)
// ---------------------------------------------------------------------------

#[test]
fn scheduler_stats_run_queue_len_is_observable() {
    // After spawning actors but before they run, the queue should be non-empty.
    // We can't observe this from inside run() without a snapshot API, but we
    // can verify the stats struct is accessible and returns sane values after
    // run() completes (queue len == 0 at quiescence).
    let r = rt_par();
    r.run(|| {
        for _ in 0..10 { spawn(|| {}); }
        // Don't join — let them drain naturally.
    });
    let stats = r.stats();
    assert_eq!(stats.total_run_queue_len(), 0, "queue should be empty after run()");
}

#[test]
fn scheduler_stats_thread_count_matches_config() {
    let r = rt(3);
    r.run(|| {});
    assert_eq!(r.stats().scheduler_count(), 3);
}

// ---------------------------------------------------------------------------
// Panic isolation: a panicking actor doesn't kill the scheduler thread
// ---------------------------------------------------------------------------

#[test]
fn panic_in_actor_does_not_kill_runtime() {
    let completed = Arc::new(AtomicU64::new(0));
    let c = completed.clone();
    rt_par().run(move || {
        // Spawn a panicker alongside well-behaved actors.
        let bad = spawn(|| panic!("deliberate"));
        let mut good_handles = Vec::new();
        for _ in 0..10 {
            let cc = c.clone();
            good_handles.push(spawn(move || {
                cc.fetch_add(1, Ordering::SeqCst);
            }));
        }
        let _ = bad.join(); // expect Err
        for h in good_handles { h.join().unwrap(); }
    });
    assert_eq!(completed.load(Ordering::SeqCst), 10);
}

// ---------------------------------------------------------------------------
// No slot leaks: rapid spawn/join churn
// ---------------------------------------------------------------------------

#[test]
fn no_slot_leak_under_churn() {
    // Spawn and join many short actors in a loop. If slots leak, the slot
    // table grows unboundedly. We can't directly measure it without an
    // introspection API, but the test at least checks correctness under
    // churn and will OOM if there's a severe leak.
    let counter = Arc::new(AtomicU64::new(0));
    let c = counter.clone();
    rt_par().run(move || {
        for _ in 0..500 {
            let cc = c.clone();
            spawn(move || { cc.fetch_add(1, Ordering::SeqCst); })
                .join()
                .unwrap();
        }
    });
    assert_eq!(counter.load(Ordering::SeqCst), 500);
}

// ---------------------------------------------------------------------------
// Ping-pong: channel round-trips between two actors
// ---------------------------------------------------------------------------

#[test]
fn ping_pong_completes() {
    const ROUNDS: u64 = 1_000;
    let final_val = Arc::new(AtomicU64::new(0));
    let fv = final_val.clone();
    rt_par().run(move || {
        let (tx_a, rx_a) = channel::<u64>();
        let (tx_b, rx_b) = channel::<u64>();
        let h_a = spawn(move || {
            tx_a.send(0).unwrap();
            for _ in 0..ROUNDS {
                let v = rx_b.recv().unwrap();
                tx_a.send(v + 1).unwrap();
            }
        });
        let h_b = spawn(move || {
            for _ in 0..=ROUNDS {
                let v = rx_a.recv().unwrap();
                if v < ROUNDS {
                    tx_b.send(v).unwrap();
                } else {
                    fv.store(v, Ordering::SeqCst);
                }
            }
        });
        h_a.join().unwrap();
        h_b.join().unwrap();
    });
    assert_eq!(final_val.load(Ordering::SeqCst), ROUNDS);
}

/// Regression test for the multi-scheduler timer-only hang.
///
/// Bug: when the run queue is empty and only timers are pending (no IO
/// outstanding), all N scheduler threads called `poll_wake(wake_fd,
/// Some(timeout))` on the *same* pipe fd. When the timer fired, the one
/// thread that won `drain_lock` consumed the single wake byte and re-queued
/// the actors; the other N-1 threads stayed blocked in `poll()` for the full
/// timeout duration. After actors completed and `all_clear` became true, the
/// stuck threads had to wait out the remainder of their poll timeout before
/// noticing — adding up to one full sleep duration of extra latency.
///
/// Fix: use `thread::sleep(timeout)` when `io_outstanding == 0`, so every
/// scheduler thread independently wakes at the deadline without contending
/// on the wake pipe.
///
/// Signal: with SLEEP_MS=300 and ≥2 scheduler threads, the broken impl
/// takes ≥2×SLEEP_MS (actors sleep + stuck threads drain their poll timeout
/// before seeing all_clear). The fixed impl takes ≈SLEEP_MS + epsilon.
#[test]
fn multi_thread_timer_only_no_pipe_contention() {
    const SLEEP_MS: u64 = 300;
    const ACTORS: usize = 100;

    // Need at least 2 scheduler threads: one wins drain_lock and does work,
    // the rest pile into poll_wake and get stuck.
    let n = std::thread::available_parallelism()
        .map(|n| n.get())
        .unwrap_or(1)
        .max(2);
    let r = smarm::runtime::init(Config::exact(n.min(4)));

    let count = Arc::new(AtomicU64::new(0));
    let c = count.clone();
    let start = std::time::Instant::now();

    r.run(move || {
        let mut handles = Vec::new();
        for _ in 0..ACTORS {
            let cc = c.clone();
            handles.push(spawn(move || {
                // Pure timer sleep: no channels, no IO.
                // All scheduler threads will be idle simultaneously while
                // these timers are pending — the condition that triggers the bug.
                smarm::sleep(Duration::from_millis(SLEEP_MS));
                cc.fetch_add(1, Ordering::SeqCst);
            }));
        }
        for h in handles {
            h.join().unwrap();
        }
    });

    assert_eq!(count.load(Ordering::SeqCst), ACTORS as u64, "not all actors completed");

    let elapsed = start.elapsed();
    assert!(
        elapsed < Duration::from_millis(SLEEP_MS * 2),
        "run() took {elapsed:?} — expected <{}ms; likely stuck threads draining \
         their poll_wake timeout after all_clear (bug: scheduler threads poll \
         the shared wake pipe instead of sleeping independently for timer-only workloads)",
        SLEEP_MS * 2,
    );
}

// ---------------------------------------------------------------------------
// Root panic propagation

/// A panic in the root actor escapes `run()` to the caller. Anything else
/// makes every assert inside `run` silently vacuous — found live when a
/// failing-first test passed: the tripped assert was caught by the
/// trampoline, recorded as `Outcome::Panic` on the root slot, and dropped
/// unread with the initial handle.
#[test]
#[should_panic(expected = "root actor panic escapes")]
fn root_panic_escapes_run() {
    rt1().run(|| {
        panic!("root actor panic escapes");
    });
}

/// Teardown completes before the root panic propagates: a caller that
/// catches it can immediately `run()` again on the same `Runtime` (the
/// documented sequential-reuse contract).
#[test]
fn runtime_reusable_after_root_panic() {
    let r = rt1();
    let caught = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| {
        r.run(|| panic!("boom"));
    }));
    assert!(caught.is_err(), "root panic must escape run()");
    let ran = Arc::new(AtomicBool::new(false));
    let ran_t = ran.clone();
    r.run(move || ran_t.store(true, Ordering::Relaxed));
    assert!(ran.load(Ordering::Relaxed), "runtime unusable after root panic");
}