//! Terminal wake: a scheduler thread blocked in the idle wait must be woken
//! when a sibling reaches the all-done verdict, or `rt.run` stalls in its
//! worker join.
//!
//! Mechanism (runtime.rs, `Pop::Idle`): an idle scheduler snapshots
//! `peek_deadline()` / `io_outstanding` and blocks in `poll_wake` (or
//! `thread::sleep`) on that snapshot. `enqueue` does not write the wake
//! pipe — only IO completions do — so the snapshot can go terminally stale:
//!
//! - An actor parked in `wait_readable` that is `request_stop`ped produces
//!   NO completion (cancellation deregisters the waiter); a sibling that
//!   blocked on `io_outstanding > 0` with no timers is in `poll(-1)` forever.
//! - An actor cancelled out of a long `sleep` leaves its timer entry
//!   orphaned; a sibling that blocked on that deadline sleeps it out in full.
//!
//! In both cases the remaining work completes on the *other* scheduler
//! thread, which hits AllDone and returns — and nothing wakes the blocked
//! one. `Runtime::run` joins it: a permanent hang in the first case, a
//! full-deadline stall in the second.
//!
//! Both tests force the window deterministically: the root busy-spins
//! (creating no timer entries and occupying one scheduler thread) so the
//! other thread settles into the stale idle wait before the stop is issued.

use std::sync::mpsc;
use std::time::{Duration, Instant};

struct PipePair {
    read: libc::c_int,
    write: libc::c_int,
}

impl PipePair {
    fn new() -> Self {
        let mut fds: [libc::c_int; 2] = [0; 2];
        let r = unsafe { libc::pipe2(fds.as_mut_ptr(), libc::O_CLOEXEC | libc::O_NONBLOCK) };
        assert_eq!(r, 0, "pipe2 failed");
        PipePair { read: fds[0], write: fds[1] }
    }
}

impl Drop for PipePair {
    fn drop(&mut self) {
        unsafe {
            libc::close(self.read);
            libc::close(self.write);
        }
    }
}

/// Occupy the current scheduler thread without creating timer entries or
/// parking. A plain spin has no observation points, so the root stays
/// on-CPU and the sibling scheduler is left alone with the idle branch.
fn spin_for(d: Duration) {
    let t0 = Instant::now();
    while t0.elapsed() < d {
        std::hint::spin_loop();
    }
}

/// Run `body` under a 2-scheduler runtime on a watchdog thread; fail if
/// `Runtime::run` has not returned within `limit`.
fn run_with_watchdog(limit: Duration, body: impl FnOnce() + Send + 'static) {
    let (done_tx, done_rx) = mpsc::channel::<()>();
    std::thread::spawn(move || {
        let rt = smarm::init(smarm::Config::exact(2));
        rt.run(body);
        let _ = done_tx.send(());
    });
    done_rx
        .recv_timeout(limit)
        .expect("Runtime::run did not return: idle scheduler thread was never woken at termination");
}

/// Permanent-hang variant: sibling blocked in `poll_wake(wake_fd, None)`
/// because `io_outstanding > 0` (one actor parked in `wait_readable`) and no
/// timers are pending. The waiter is then stop-cancelled — no IO completion
/// ever writes the wake pipe — and everything else finishes on the root's
/// thread. Without a terminal wake, `rt.run` never returns.
#[test]
fn run_returns_after_io_waiter_is_stop_cancelled() {
    run_with_watchdog(Duration::from_secs(10), || {
        let pipe = PipePair::new();
        let rfd = pipe.read;
        let h = smarm::spawn(move || {
            // Never-readable fd (write end open, nothing written).
            let _ = smarm::wait_readable(rfd);
        });
        // Let the waiter park and the sibling scheduler settle into the
        // io_outstanding>0 / no-timers idle wait: poll(wake_fd, -1).
        spin_for(Duration::from_millis(200));
        smarm::request_stop(h.pid());
        let _ = h.join();
        drop(pipe);
    });
}

/// Finite-stall variant: sibling blocked on an orphaned long timer
/// deadline. An actor cancelled out of `sleep(60s)` leaves its timer entry
/// behind (documented as harmless at AllDone — but a sibling that already
/// blocked on that deadline sleeps it out in full, stalling `rt.run` for
/// the better part of a minute).
#[test]
fn run_returns_after_long_sleeper_is_stop_cancelled() {
    run_with_watchdog(Duration::from_secs(10), || {
        let h = smarm::spawn(|| {
            smarm::sleep(Duration::from_secs(60));
        });
        // Let the sleeper park and the sibling scheduler block on the 60s
        // deadline: poll(wake_fd, ~60_000ms).
        spin_for(Duration::from_millis(200));
        smarm::request_stop(h.pid());
        let _ = h.join();
    });
}