//! RFC 008 — fd arms in select. Beyond the functional cases, the
//! *_stays_usable tests are the soundness probes for the one asymmetry the
//! RFC must close: a losing CHANNEL arm's stale registration is inert, but
//! a losing FD arm's registration (waiters entry + kernel ONESHOT) poisons
//! the fd with AlreadyExists until the eager cleanup pass removes it. Every
//! "loser" scenario therefore re-waits on the same fd afterwards and must
//! succeed — pre-cleanup, each of those re-waits errors or hangs.
//!
//! House pattern: actor panics are trampoline-caught and `run` returns
//! normally, so every test funnels its result into an outcome flag asserted
//! OUTSIDE `run` — an in-actor assertion alone passes vacuously.

use smarm::{
    channel, run, select, select_timeout, spawn, try_select, wait_readable,
    wait_readable_timeout, wait_writable_timeout, yield_now, FdArm,
};
use std::os::fd::RawFd;
use std::sync::atomic::{AtomicBool, AtomicU32, Ordering};
use std::sync::Arc;
use std::time::{Duration, Instant};

// ---------------------------------------------------------------------------
// Pipe helper (as in io_epoll.rs)
// ---------------------------------------------------------------------------

struct Pipe {
    read: RawFd,
    write: RawFd,
}

impl Pipe {
    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");
        Pipe { read: fds[0], write: fds[1] }
    }
}

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

fn raw_write(fd: RawFd, buf: &[u8]) -> isize {
    unsafe { libc::write(fd, buf.as_ptr() as *const _, buf.len()) }
}

fn raw_read(fd: RawFd, buf: &mut [u8]) -> isize {
    unsafe { libc::read(fd, buf.as_mut_ptr() as *mut _, buf.len()) }
}

fn flag() -> (Arc<AtomicBool>, Arc<AtomicBool>) {
    let f = Arc::new(AtomicBool::new(false));
    (f.clone(), f)
}

// ---------------------------------------------------------------------------
// Ready-now: data already pending retires the wait without parking.
// ---------------------------------------------------------------------------

#[test]
fn fd_arm_ready_now_returns_without_parking() {
    let (ok, ok2) = flag();
    run(move || {
        let p = Pipe::new();
        assert_eq!(raw_write(p.write, b"x"), 1);
        let (_tx, rx) = channel::<i64>();
        let fd_arm = FdArm::readable(p.read);
        // fd arm at index 1: the ready-now path must also work for a
        // non-first arm (and clean nothing — channel arms are inert).
        let i = select(&[&rx, &fd_arm]);
        assert_eq!(i, 1);
        let mut buf = [0u8; 1];
        assert_eq!(raw_read(p.read, &mut buf), 1);
        ok2.store(true, Ordering::SeqCst);
    });
    assert!(ok.load(Ordering::SeqCst));
}

// ---------------------------------------------------------------------------
// Park-then-wake: fd arm wins against an idle channel arm.
// ---------------------------------------------------------------------------

#[test]
fn fd_arm_parks_until_data_then_wins() {
    let got = Arc::new(AtomicU32::new(0));
    let got2 = got.clone();
    run(move || {
        let p = Pipe::new();
        let (rfd, wfd) = (p.read, p.write);
        let (_tx_keepalive, rx) = channel::<i64>();

        let h = spawn(move || {
            let fd_arm = FdArm::readable(rfd);
            let i = select(&[&fd_arm, &rx]);
            assert_eq!(i, 0);
            let mut buf = [0u8; 1];
            assert_eq!(raw_read(rfd, &mut buf), 1);
            got2.store(buf[0] as u32, Ordering::SeqCst);
        });
        yield_now(); // let it park
        assert_eq!(raw_write(wfd, b"y"), 1);
        let _ = h.join();
    });
    assert_eq!(got.load(Ordering::SeqCst), b'y' as u32);
}

// ---------------------------------------------------------------------------
// THE asymmetry probe: channel arm wins, losing fd arm must be cleaned —
// the same actor (and the io thread) must be able to wait that fd again.
// ---------------------------------------------------------------------------

#[test]
fn losing_fd_arm_is_cleaned_up_and_fd_stays_usable() {
    let got = Arc::new(AtomicU32::new(0));
    let got2 = got.clone();
    run(move || {
        let p = Pipe::new();
        let (rfd, wfd) = (p.read, p.write);
        let (tx, rx) = channel::<i64>();

        let h = spawn(move || {
            let fd_arm = FdArm::readable(rfd);
            // Channel wins; the fd arm registered and lost.
            let i = select(&[&fd_arm, &rx]);
            assert_eq!(i, 1);
            assert_eq!(rx.try_recv().unwrap(), Some(7));

            // Pre-cleanup this wait_readable fails AlreadyExists (the
            // waiters entry is stale-ours) — the cleanup pass must have
            // removed it.
            wait_readable(rfd).unwrap();
            let mut buf = [0u8; 1];
            assert_eq!(raw_read(rfd, &mut buf), 1);
            got2.store(buf[0] as u32, Ordering::SeqCst);
        });
        yield_now(); // let it park in the select
        tx.send(7).unwrap();
        yield_now(); // let it reach the second wait
        assert_eq!(raw_write(wfd, b"z"), 1);
        let _ = h.join();
    });
    assert_eq!(got.load(Ordering::SeqCst), b'z' as u32);
}

// ---------------------------------------------------------------------------
// Ready-now on a LATER arm must unregister an earlier fd arm (the
// register_arms prefix-cleanup path: no park ever happens).
// ---------------------------------------------------------------------------

#[test]
fn ready_now_later_arm_cleans_earlier_fd_arm() {
    let (ok, ok2) = flag();
    run(move || {
        let p = Pipe::new();
        let (rfd, wfd) = (p.read, p.write);
        let (tx, rx) = channel::<i64>();
        tx.send(1).unwrap(); // arm 1 ready before the select

        let fd_arm = FdArm::readable(rfd);
        let i = select(&[&fd_arm, &rx]);
        assert_eq!(i, 1);
        assert_eq!(rx.try_recv().unwrap(), Some(1));

        // The fd arm registered (idle pipe), then arm 1 retired the wait.
        // Its registration must have been removed in the same pass.
        assert_eq!(raw_write(wfd, b"a"), 1);
        wait_readable(rfd).unwrap();
        let mut buf = [0u8; 1];
        assert_eq!(raw_read(rfd, &mut buf), 1);
        ok2.store(true, Ordering::SeqCst);
    });
    assert!(ok.load(Ordering::SeqCst));
}

// ---------------------------------------------------------------------------
// Two fd arms in one select (phase-1: distinct fds, nothing special).
// ---------------------------------------------------------------------------

#[test]
fn two_fd_arms_second_fires_first_stays_usable() {
    let (ok, ok2) = flag();
    run(move || {
        let pa = Pipe::new();
        let pb = Pipe::new();
        let (rfd_a, wfd_a) = (pa.read, pa.write);
        let (rfd_b, wfd_b) = (pb.read, pb.write);

        let h = spawn(move || {
            let a = FdArm::readable(rfd_a);
            let b = FdArm::readable(rfd_b);
            let i = select(&[&a, &b]);
            assert_eq!(i, 1);
            let mut buf = [0u8; 1];
            assert_eq!(raw_read(rfd_b, &mut buf), 1);

            // Arm a lost; its fd must be immediately re-waitable.
            assert_eq!(raw_write(wfd_a, b"q"), 1);
            wait_readable(rfd_a).unwrap();
            assert_eq!(raw_read(rfd_a, &mut buf), 1);
            assert_eq!(buf[0], b'q');
            ok2.store(true, Ordering::SeqCst);
        });
        yield_now();
        assert_eq!(raw_write(wfd_b, b"b"), 1);
        let _ = h.join();
    });
    assert!(ok.load(Ordering::SeqCst));
}

// ---------------------------------------------------------------------------
// select_timeout: timer wins over an idle fd arm; the fd is left clean.
// ---------------------------------------------------------------------------

#[test]
fn select_timeout_timer_beats_idle_fd_arm_and_fd_stays_usable() {
    let (ok, ok2) = flag();
    run(move || {
        let p = Pipe::new();
        let (rfd, wfd) = (p.read, p.write);

        let fd_arm = FdArm::readable(rfd);
        let start = Instant::now();
        let r = select_timeout(&[&fd_arm], Duration::from_millis(30));
        assert!(r.is_none(), "idle fd must time out");
        assert!(start.elapsed() >= Duration::from_millis(30));

        // Timer win is exactly the case where the fd arm's registration is
        // left behind without an eager pass.
        assert_eq!(raw_write(wfd, b"c"), 1);
        wait_readable(rfd).unwrap();
        let mut buf = [0u8; 1];
        assert_eq!(raw_read(rfd, &mut buf), 1);
        ok2.store(true, Ordering::SeqCst);
    });
    assert!(ok.load(Ordering::SeqCst));
}

// ---------------------------------------------------------------------------
// Derived wrappers.
// ---------------------------------------------------------------------------

#[test]
fn wait_readable_timeout_times_out_then_succeeds_with_data() {
    let (ok, ok2) = flag();
    run(move || {
        let p = Pipe::new();
        let (rfd, wfd) = (p.read, p.write);

        let start = Instant::now();
        assert_eq!(wait_readable_timeout(rfd, Duration::from_millis(30)).unwrap(), false);
        assert!(start.elapsed() >= Duration::from_millis(30));

        // Timed-out wait must leave the fd clean; ready path returns true.
        assert_eq!(raw_write(wfd, b"d"), 1);
        assert_eq!(wait_readable_timeout(rfd, Duration::from_secs(5)).unwrap(), true);
        let mut buf = [0u8; 1];
        assert_eq!(raw_read(rfd, &mut buf), 1);
        ok2.store(true, Ordering::SeqCst);
    });
    assert!(ok.load(Ordering::SeqCst));
}

#[test]
fn wait_readable_timeout_wakes_on_late_data() {
    let got = Arc::new(AtomicU32::new(0));
    let got2 = got.clone();
    run(move || {
        let p = Pipe::new();
        let (rfd, wfd) = (p.read, p.write);
        let h = spawn(move || {
            assert_eq!(wait_readable_timeout(rfd, Duration::from_secs(5)).unwrap(), true);
            let mut buf = [0u8; 1];
            assert_eq!(raw_read(rfd, &mut buf), 1);
            got2.store(buf[0] as u32, Ordering::SeqCst);
        });
        yield_now();
        assert_eq!(raw_write(wfd, b"e"), 1);
        let _ = h.join();
    });
    assert_eq!(got.load(Ordering::SeqCst), b'e' as u32);
}

#[test]
fn wait_writable_timeout_ready_now_on_empty_pipe() {
    let (ok, ok2) = flag();
    run(move || {
        let p = Pipe::new();
        // An empty pipe's write end is writable: ready-now path, no park.
        assert_eq!(wait_writable_timeout(p.write, Duration::from_secs(5)).unwrap(), true);
        ok2.store(true, Ordering::SeqCst);
    });
    assert!(ok.load(Ordering::SeqCst));
}

// ---------------------------------------------------------------------------
// Error surface: registration failure is an Err from try_select, with the
// wait retired (the actor can immediately wait on something else).
// ---------------------------------------------------------------------------

#[test]
fn try_select_surfaces_registration_error_and_retires_the_wait() {
    let (ok, ok2) = flag();
    run(move || {
        let bad: RawFd = {
            let p = Pipe::new();
            p.read
        }; // both ends closed by Drop: EBADF on registration

        let fd_arm = FdArm::readable(bad);
        let err = try_select(&[&fd_arm]).unwrap_err();
        // EBADF, whether the pre-poll or epoll_ctl ADD reports it.
        assert_eq!(err.raw_os_error(), Some(libc::EBADF));

        // The wait was retired: a normal select right after works.
        let (tx, rx) = channel::<i64>();
        tx.send(9).unwrap();
        let i = select(&[&rx]);
        assert_eq!(i, 0);
        assert_eq!(rx.try_recv().unwrap(), Some(9));
        ok2.store(true, Ordering::SeqCst);
    });
    assert!(ok.load(Ordering::SeqCst));
}

// ---------------------------------------------------------------------------
// Stop-unwind: an actor stopped while parked in an fd-arm select must not
// poison the fd (the UnregisterGuard generalization of wait_fd's Dereg).
// Mirrors io_epoll.rs::stopped_waiter_does_not_poison_the_fd.
// ---------------------------------------------------------------------------

#[test]
fn stopped_selector_does_not_poison_the_fd() {
    let seen = Arc::new(AtomicU32::new(0));
    let seen_outer = seen.clone();
    run(move || {
        let p = Pipe::new();
        let (rfd, wfd) = (p.read, p.write);
        let (_tx_keepalive, rx) = channel::<i64>();

        let h = spawn(move || {
            let fd_arm = FdArm::readable(rfd);
            select(&[&fd_arm, &rx]);
            unreachable!("neither arm ever fires while this actor lives");
        });
        yield_now(); // let it reach the park
        smarm::request_stop(h.pid());
        let _ = h.join(); // Ok(()): stopped, not panicked

        // Second waiter on the SAME fd must register and be woken.
        let seen2 = seen.clone();
        let h2 = spawn(move || {
            wait_readable(rfd).unwrap();
            let mut buf = [0u8; 1];
            assert_eq!(raw_read(rfd, &mut buf), 1);
            seen2.store(buf[0] as u32, Ordering::SeqCst);
        });
        yield_now();
        assert_eq!(raw_write(wfd, b"x"), 1);
        let _ = h2.join();
    });
    assert_eq!(seen_outer.load(Ordering::SeqCst), b'x' as u32);
}

// ---------------------------------------------------------------------------
// Phase-1 misuse: a second waiter on an fd that already has one is an Err
// (AlreadyExists), not a hang — and does not disturb the first waiter.
// ---------------------------------------------------------------------------

#[test]
fn second_waiter_on_same_fd_errs_without_disturbing_the_first() {
    let got = Arc::new(AtomicU32::new(0));
    let got2 = got.clone();
    let (ok, ok2) = flag();
    run(move || {
        let p = Pipe::new();
        let (rfd, wfd) = (p.read, p.write);

        let h = spawn(move || {
            wait_readable(rfd).unwrap();
            let mut buf = [0u8; 1];
            assert_eq!(raw_read(rfd, &mut buf), 1);
            got2.store(buf[0] as u32, Ordering::SeqCst);
        });
        yield_now(); // first waiter parked

        let fd_arm = FdArm::readable(rfd);
        let err = try_select(&[&fd_arm]).unwrap_err();
        assert_eq!(err.kind(), std::io::ErrorKind::AlreadyExists);

        // First waiter still wakes normally.
        assert_eq!(raw_write(wfd, b"w"), 1);
        let _ = h.join();
        ok2.store(true, Ordering::SeqCst);
    });
    assert_eq!(got.load(Ordering::SeqCst), b'w' as u32);
    assert!(ok.load(Ordering::SeqCst));
}