smarm/src/scheduler.rs

//! Scheduler public API — thin façade over the multi-scheduler runtime.
//!
//! All heavy lifting lives in `runtime.rs`. This module exposes the same
//! surface that the rest of the codebase (channel, mutex, io, timer, actor)
//! calls into, plus the public API re-exported from `lib.rs`.
//!
//! The single-threaded `run()` entry point is kept as a convenience wrapper
//! around `runtime::init(Config::exact(1)).run(f)`.

use crate::actor::current_pid;
use crate::channel::Sender;
use crate::pid::{Name, Pid};
use crate::runtime::{
    self, RuntimeInner, YieldIntent, RUNTIME,
};
use crate::supervisor::Signal;
use std::sync::Arc;

// ---------------------------------------------------------------------------
// with_runtime / try_with_runtime
// ---------------------------------------------------------------------------

/// Borrow the current runtime. Panics if called outside `Runtime::run()`.
///
/// The whole span runs with preemption disabled. Two reasons, both load-
/// bearing:
///
/// - The `RUNTIME` thread-local borrow is live across `f`. A preemption-
///   driven context switch inside `f` can resume the actor on a DIFFERENT OS
///   thread; the borrow guard would then increment this thread's RefCell but
///   decrement the other's — a count underflow that leaves that thread's
///   RefCell permanently "mutably borrowed". Holding a thread-local guard
///   across a potential switch point is the one unforgivable sin of green
///   threads; disabling preemption makes "no switch inside `f`" structural
///   instead of an accident of which bodies happen to allocate.
/// - `f` is runtime bookkeeping. Suspending an actor halfway through it (or
///   unwinding via the stop sentinel, which shares the gate) is never wanted.
pub(crate) fn with_runtime<R>(f: impl FnOnce(&Arc<RuntimeInner>) -> R) -> R {
    let prev = crate::preempt::PREEMPTION_ENABLED.with(|c| c.replace(false));
    let result = RUNTIME.with(|r| {
        let b = r.borrow();
        let inner = match b.as_ref() {
            Some(inner) => inner,
            None => panic!("smarm: not inside Runtime::run()"),
        };
        f(inner)
    });
    crate::preempt::PREEMPTION_ENABLED.with(|c| c.set(prev));
    result
}

/// Borrow the runtime if present; returns `None` otherwise.
/// Used on cleanup paths (channel Drop during teardown).
/// Same preemption gate as [`with_runtime`]; same reasons.
pub(crate) fn try_with_runtime<R>(f: impl FnOnce(&Arc<RuntimeInner>) -> R) -> Option<R> {
    let prev = crate::preempt::PREEMPTION_ENABLED.with(|c| c.replace(false));
    let result = RUNTIME.with(|r| r.borrow().as_ref().map(f));
    crate::preempt::PREEMPTION_ENABLED.with(|c| c.set(prev));
    result
}

// ---------------------------------------------------------------------------
// JoinHandle / JoinError
// ---------------------------------------------------------------------------

#[derive(Debug)]
pub struct JoinError {
    pub payload: Box<dyn std::any::Any + Send>,
}

pub struct JoinHandle {
    pid: Pid,
    consumed: bool,
}

impl JoinHandle {
    pub fn pid(&self) -> Pid { self.pid }

    pub fn join(mut self) -> Result<(), JoinError> {
        use crate::actor::Outcome;

        let me = match current_pid() {
            Some(pid) => pid,
            None => panic!("join() called outside an actor"),
        };

        loop {
            // Check-Done-or-register-waiter is atomic under the target's cold
            // lock; finalize publishes Done and takes the waiter list under
            // the same lock, so we either see the outcome or are woken.
            let outcome = with_runtime(|inner| {
                let slot = match inner.slot_at(self.pid) {
                    Some(slot) => slot,
                    None => panic!("join: pid index out of range: {:?}", self.pid),
                };
                let mut cold = slot.cold.lock();
                match slot.status_for(self.pid) {
                    // Our outstanding handle pins the slot: it cannot be
                    // reclaimed (generation cannot change) while we hold it.
                    crate::slot_state::Status::Stale => {
                        panic!("join: target slot has been reused")
                    }
                    crate::slot_state::Status::Done => {
                        Some(match cold.outcome.take() {
                            Some(outcome) => outcome,
                            None => panic!("Done slot must have outcome"),
                        })
                    }
                    crate::slot_state::Status::Live => {
                        // begin_wait is lock-free, legal under the cold lock;
                        // registering under it makes the epoch atomic with
                        // the check-Done-or-register linearization point.
                        cold.waiters.push((me, begin_wait()));
                        None
                    }
                }
            });

            match outcome {
                Some(o) => {
                    self.consumed = true;
                    self.decrement_handle_count();
                    return match o {
                        Outcome::Exit => Ok(()),
                        Outcome::Panic(p) => Err(JoinError { payload: p }),
                        // A cooperative stop carries no panic payload to
                        // propagate; the *reason* is observable via monitors
                        // (DownReason::Stopped). join() therefore reports Ok.
                        Outcome::Stopped => Ok(()),
                    };
                }
                None => {
                    let _np = NoPreempt::enter();
                    park_current();
                }
            }
        }
    }

    fn decrement_handle_count(&mut self) {
        with_runtime(|inner| {
            let should_reclaim = match inner.slot_at(self.pid) {
                Some(slot) => {
                    let mut cold = slot.cold.lock();
                    match slot.status_for(self.pid) {
                        crate::slot_state::Status::Stale => false,
                        status => {
                            cold.outstanding_handles =
                                cold.outstanding_handles.saturating_sub(1);
                            cold.outstanding_handles == 0
                                && status == crate::slot_state::Status::Done
                        }
                    }
                }
                None => false,
            };
            if should_reclaim {
                // Re-verified inside; benign if finalize's reclaim won a race.
                crate::runtime::reclaim_slot(inner, self.pid);
            }
        });
    }
}

impl Drop for JoinHandle {
    fn drop(&mut self) {
        if !self.consumed {
            // May be called outside run() if handle is dropped after teardown.
            if try_with_runtime(|_| ()).is_some() {
                self.decrement_handle_count();
            }
        }
    }
}

// ---------------------------------------------------------------------------
// spawn / spawn_under / self_pid
// ---------------------------------------------------------------------------

pub fn spawn(f: impl FnOnce() + Send + 'static) -> JoinHandle {
    let parent = current_pid().unwrap_or_else(|| {
        // Outside an actor but inside run(): the initial spawn. with_runtime
        // panics with "not inside Runtime::run()" if there's no runtime at all.
        with_runtime(|_| crate::runtime::ROOT_PID)
    });
    spawn_under(parent, f)
}

pub fn spawn_under<A>(supervisor: Pid<A>, f: impl FnOnce() + Send + 'static) -> JoinHandle {
    let supervisor = supervisor.erase();
    // Stack + closure boxing happen before ANY runtime lock is taken: no
    // syscall and no allocation ever stalls another scheduler thread.
    let stack = with_runtime(|inner| inner.stack_pool.lock().pop())
        .unwrap_or_else(|| {
            match crate::stack::Stack::new(crate::runtime::ACTOR_STACK_SIZE) {
                Ok(stack) => stack,
                Err(e) => panic!("stack allocation failed: {e}"),
            }
        });
    let sp = init_actor_stack(stack.top(), crate::actor::trampoline);
    let closure: crate::runtime::Closure = Box::new(f);

    let pid = with_runtime(|inner| {
        let idx = inner.allocate_slot(); // panics loudly on slab exhaustion
        crate::runtime::install_actor(inner, idx, sp, stack, supervisor, closure)
    });

    JoinHandle { pid, consumed: false }
}

/// Spawn a typed, single-message actor and hand back its identity-bound
/// [`Pid<A>`] (RFC 014's typed-path producer). The runtime makes the actor's
/// inbox, hands the body its [`Receiver<A::Msg>`], installs the sender, and
/// returns the parent a `Pid<A>`.
///
/// The inbox is published from the parent side **before** the pid is returned
/// (see [`registry::install_for`](crate::registry::install_for)), so the address
/// is live the instant the caller holds it: an immediate
/// [`send_to`](crate::send_to) always resolves, never racing the body's first
/// instruction. The actor is detached — its lifetime is governed by its own
/// logic (an explicit stop message, or returning), like
/// [`GenServerBuilder::start`](crate::GenServerBuilder::start) — so the backing join
/// handle is dropped. Spawns under the current actor (via [`spawn`]).
///
/// Panics if called outside `Runtime::run()`.
pub fn spawn_addr<A: crate::pid::Addressable>(
    body: impl FnOnce(crate::channel::Receiver<A::Msg>) + Send + 'static,
) -> Pid<A> {
    let (tx, rx) = crate::channel::channel::<A::Msg>();
    let handle = spawn(move || body(rx));
    let pid = handle.pid();
    // Publish the sender for `pid` before returning the typed address. `handle`
    // drops at end of scope (detached).
    crate::registry::install_for::<A::Msg>(pid, tx);
    crate::pid::assert_type::<A>(pid)
}

use crate::context::init_actor_stack;

pub fn self_pid() -> Pid {
    match current_pid() {
        Some(pid) => pid,
        None => panic!("self_pid() called outside an actor"),
    }
}

// ---------------------------------------------------------------------------
// yield_now / park_current / unpark
// ---------------------------------------------------------------------------

pub fn yield_now() {
    runtime::set_yield_intent(YieldIntent::Yield);
    unsafe { crate::context::switch_to_scheduler() };
    // Observation point: we may have been resumed only to be cancelled.
    crate::preempt::check_cancelled();
}

pub fn park_current() {
    // Entry-side observation point: a stop flagged while we were QUEUED is
    // otherwise lost — the stop's wildcard unpark no-ops on a Queued actor
    // (the pending run "is" the wake), so if our first action on resume is
    // this park, no wake is ever coming and the wake-side check below is
    // unreachable. Checking here closes that hole; a flag that lands after
    // this check is covered by the existing protocol (the stop's unpark
    // finds Running / the prep-to-park window, sets Notified, and the
    // park-return re-queues us into the wake-side check). Unwinding from
    // here is the same unwind path as the wake side: leftover wait
    // registrations are stale-epoch / dead-generation and self-clean at
    // their wakers' failed CAS.
    crate::preempt::check_cancelled();
    runtime::set_yield_intent(YieldIntent::Park);
    unsafe { crate::context::switch_to_scheduler() };
    // Observation point on the wakeup side of every blocking primitive
    // (recv/sleep/mutex/io/join). Past the prep-to-park window, so this never
    // races a wakeup: a stop unparks us, we resume here, and unwind out of
    // whatever blocking call parked us — running Drop along the way.
    crate::preempt::check_cancelled();
}

pub fn unpark(pid: Pid) {
    // The whole protocol lives on the slot's packed word (gen + epoch +
    // state in one CAS) — see slot_state.rs. No runtime lock unless we
    // enqueue. WILDCARD form: reserved for terminal wakes (request_stop);
    // every registration-based waker must use `unpark_at`.
    let _ = try_with_runtime(|inner| inner.unpark(pid));
}

/// Epoch-matched unpark: wake `pid` only if its current wait is still the
/// one this waker registered for. The form every registration-based waker
/// (channel senders, mutex grants, wait-timers, io completions, joiner
/// wakes) must use — see slot_state.rs for the consuming-wake rules.
pub(crate) fn unpark_at(pid: Pid, epoch: u32) {
    let _ = try_with_runtime(|inner| inner.unpark_at(pid, epoch));
}

/// Open a new wait for the CURRENT actor: bump its park-epoch and return
/// it. Call once per wait, before registering `(pid, epoch)` with any
/// waker. Lock-free (one CAS on the own slot word), so it is legal under
/// any lock, including a Channel-class lock.
pub(crate) fn begin_wait() -> u32 {
    let me = match current_pid() {
        Some(pid) => pid,
        None => panic!("begin_wait() called outside an actor"),
    };
    with_runtime(|inner| inner.begin_wait(me))
}

/// Close the current actor's wait WITHOUT parking on it. For the no-park
/// exit of multi-registration waits (`select` finding an arm ready at
/// registration time): bumps the epoch so in-flight wakes die at their CAS,
/// eats a notification that already landed, then observes a pending stop —
/// in that order. After this, leftover registrations are stale-epoch and
/// self-clean at their wakers' failed CAS; nothing can fault the actor's
/// next one-shot park.
pub(crate) fn retire_wait() {
    let me = match current_pid() {
        Some(pid) => pid,
        None => panic!("retire_wait() called outside an actor"),
    };
    with_runtime(|inner| inner.retire_wait(me));
    // A request_stop that fired before the clear had its notification
    // eaten, but it set the stop flag first — observe it here and unwind.
    // One that fires after re-notifies a Running word as usual.
    crate::preempt::check_cancelled();
}

/// Request cooperative cancellation of `pid`.
///
/// Sets the actor's stop flag and wakes it so it observes the flag promptly:
/// a parked target is re-queued; a running target is marked notified, so its
/// next park returns immediately to an observation point. The actor realises
/// the stop as a controlled unwind at its next observation point
/// (`check!()`/allocation, or the wakeup side of a blocking park),
/// terminating with `Outcome::Stopped`.
///
/// This is best-effort and cooperative: an actor that never reaches an
/// observation point — a tight loop with no `check!()`, no allocation, and no
/// blocking op — cannot be stopped, exactly as it cannot be preempted. A no-op
/// if `pid` is already gone.
pub fn request_stop<A>(pid: Pid<A>) {
    let pid = pid.erase();
    let _ = try_with_runtime(|inner| request_stop_inner(inner, pid));
}

/// The core of [`request_stop`], taking the runtime directly so it can be
/// driven from inside the runtime (e.g. the root-exit sweep in
/// `finalize_actor`) without re-borrowing the thread-local. Sets the stop flag
/// under the target's cold lock (generation re-verified there; a mismatch or a
/// slot with no live actor is a no-op) and wakes it.
pub(crate) fn request_stop_inner(inner: &RuntimeInner, pid: Pid) {
    if let Some(slot) = inner.slot_at(pid) {
        {
            let cold = slot.cold.lock();
            // Verify under the cold lock: generation can't change while
            // we hold it (reclaim takes the same lock).
            if slot.generation() == pid.generation() {
                if let Some(actor) = cold.actor.as_ref() {
                    actor.stop.store(true, std::sync::atomic::Ordering::Relaxed);
                }
            }
        }
        inner.unpark(pid);
    }
}

// ---------------------------------------------------------------------------
// NoPreempt
// ---------------------------------------------------------------------------

pub struct NoPreempt(bool);

impl NoPreempt {
    pub fn enter() -> Self {
        let prev = crate::preempt::PREEMPTION_ENABLED.with(|c| c.replace(false));
        NoPreempt(prev)
    }
}

impl Drop for NoPreempt {
    fn drop(&mut self) {
        crate::preempt::PREEMPTION_ENABLED.with(|c| c.set(self.0));
    }
}

// ---------------------------------------------------------------------------
// sleep / insert_wait_timer
// ---------------------------------------------------------------------------

pub fn sleep(duration: std::time::Duration) {
    let me = match current_pid() {
        Some(pid) => pid,
        None => panic!("sleep() called outside an actor"),
    };
    let _np = NoPreempt::enter();
    let epoch = begin_wait();
    let deadline = crate::timer::deadline_from_now(duration);
    with_runtime(|inner| {
        match inner.timers.lock() {
            Ok(mut timers) => timers.insert_sleep(deadline, me, epoch),
            Err(e) => panic!("smarm: timers lock poisoned (core corrupt): {e}"),
        }
    });
    park_current();
}

pub fn insert_wait_timer(
    deadline: std::time::Instant,
    pid: Pid,
    target: std::sync::Arc<dyn crate::timer::TimerTarget>,
    epoch: u32,
) {
    with_runtime(|inner| {
        match inner.timers.lock() {
            Ok(mut timers) => timers.insert(
                deadline,
                pid,
                crate::timer::Reason::WaitTimeout { target, epoch },
            ),
            Err(e) => panic!("smarm: timers lock poisoned (core corrupt): {e}"),
        }
    });
}

// ---------------------------------------------------------------------------
// send_after / cancel_timer — message-delivery timers (Erlang send_after).
//
// Arm a timer that delivers `msg` to an address after `after`, returning a
// `TimerId`. The destination is resolved *on fire*, not at arm time: a
// `Pid<A>` that has since died yields `SendError::Dead`, a `Name<M>` resolves
// to whoever currently holds it (so a restarted server is reached). Either way
// a failed resolve / closed inbox is dropped, matching `erlang:send_after`.
// `cancel_timer` prevents an as-yet-unfired delivery; it returns whether the
// timer was still armed.
// ---------------------------------------------------------------------------

/// Deliver `msg` to the exact actor named by `dest` (identity-bound, no
/// redirect — see [`send_to`](crate::registry::send_to)) after `after`.
/// Returns a [`TimerId`](crate::timer::TimerId) for [`cancel_timer`].
pub fn send_after<A: crate::pid::Addressable>(
    after: std::time::Duration,
    dest: Pid<A>,
    msg: A::Msg,
) -> crate::timer::TimerId {
    let deadline = crate::timer::deadline_from_now(after);
    let fire = Box::new(move || {
        let _ = crate::registry::send_to(dest, msg);
    });
    with_runtime(|inner| {
        match inner.timers.lock() {
            Ok(mut timers) => timers.insert_send(deadline, dest.erase(), fire),
            Err(e) => panic!("smarm: timers lock poisoned (core corrupt): {e}"),
        }
    })
}

/// Deliver `msg` to whichever actor holds the name `dest` at fire time
/// (re-resolving [`send`](crate::registry::send) semantics) after `after`.
/// Returns a [`TimerId`](crate::timer::TimerId) for [`cancel_timer`].
pub fn send_after_named<M: Send + 'static>(
    after: std::time::Duration,
    dest: Name<M>,
    msg: M,
) -> crate::timer::TimerId {
    let deadline = crate::timer::deadline_from_now(after);
    // Informational only (who armed it); not used for delivery.
    let armed_by = current_pid().unwrap_or(Pid::new(0, 0));
    let fire = Box::new(move || {
        let _ = crate::registry::send(dest, msg);
    });
    with_runtime(|inner| {
        match inner.timers.lock() {
            Ok(mut timers) => timers.insert_send(deadline, armed_by, fire),
            Err(e) => panic!("smarm: timers lock poisoned (core corrupt): {e}"),
        }
    })
}

/// Deliver `msg` onto a channel the caller owns after `after`, rather than to a
/// registry address. The sibling of [`send_after`] used by the gen_server timer
/// layer (RFC 015 §5): arming a server timer must land the fire on the loop's
/// own `Sys` channel — giving it the loop's arm position — not in the inbox.
///
/// Same substrate as [`send_after`]: a `Reason::Send` entry, the same `armed`
/// set, the same [`TimerId`](crate::timer::TimerId) for [`cancel_timer`]. Only
/// the fire thunk differs — `tx.send(msg)` instead of a registry resolve — so a
/// send onto a channel whose receiver is gone is dropped, exactly as a failed
/// address resolve is (Erlang `send_after` semantics). `pid` is informational
/// (who armed it); delivery lives entirely in the thunk.
pub(crate) fn send_after_to<T: Send + 'static>(
    after: std::time::Duration,
    tx: Sender<T>,
    msg: T,
) -> crate::timer::TimerId {
    let deadline = crate::timer::deadline_from_now(after);
    let armed_by = current_pid().unwrap_or(Pid::new(0, 0));
    let fire = Box::new(move || {
        let _ = tx.send(msg);
    });
    with_runtime(|inner| {
        match inner.timers.lock() {
            Ok(mut timers) => timers.insert_send(deadline, armed_by, fire),
            Err(e) => panic!("smarm: timers lock poisoned (core corrupt): {e}"),
        }
    })
}

/// Cancel a timer armed by [`send_after`] / [`send_after_named`]. Returns
/// `true` if it was still pending (delivery now prevented), `false` if it had
/// already fired or been cancelled.
pub fn cancel_timer(id: crate::timer::TimerId) -> bool {
    with_runtime(|inner| {
        match inner.timers.lock() {
            Ok(mut timers) => timers.cancel(id),
            Err(e) => panic!("smarm: timers lock poisoned (core corrupt): {e}"),
        }
    })
}

// ---------------------------------------------------------------------------
// block_on_io / wait_readable / wait_writable / read / write
// ---------------------------------------------------------------------------

pub fn block_on_io<F, T>(f: F) -> T
where
    F: FnOnce() -> T + Send + 'static,
    T: Send + 'static,
{
    let me = match current_pid() {
        Some(pid) => pid,
        None => panic!("block_on_io() called outside an actor"),
    };
    let work: Box<dyn FnOnce() -> crate::io::IoResult + Send> = Box::new(move || {
        let v: T = f();
        Ok(Box::new(v) as Box<dyn std::any::Any + Send>)
    });
    {
        let _np = NoPreempt::enter();
        let epoch = begin_wait();
        with_runtime(|inner| {
            let mut io = match inner.io.lock() {
                Ok(io) => io,
                Err(e) => panic!("smarm: io lock poisoned (core corrupt): {e}"),
            };
            match io.as_mut() {
                Some(io) => io.submit(me, epoch, work),
                None => panic!("io thread not started"),
            }
        });
        park_current();
    }
    let result = with_runtime(|inner| {
        let slot = match inner.slot_at(me) {
            Some(slot) => slot,
            None => panic!("block_on_io: own slot vanished"),
        };
        let mut cold = slot.cold.lock();
        debug_assert_eq!(
            slot.generation(), me.generation(),
            "block_on_io: own slot reused mid-park"
        );
        match cold.pending_io_result.take() {
            Some(result) => result,
            None => panic!("block_on_io: resumed without a result"),
        }
    });
    match result {
        Ok(any) => match any.downcast::<T>() {
            Ok(typed) => *typed,
            Err(_) => panic!("block_on_io: type mismatch"),
        },
        Err(payload) => std::panic::resume_unwind(payload),
    }
}

pub fn wait_readable(fd: std::os::fd::RawFd) -> std::io::Result<()> {
    wait_fd(fd, true, false)
}

pub fn wait_writable(fd: std::os::fd::RawFd) -> std::io::Result<()> {
    wait_fd(fd, false, true)
}

fn wait_fd(fd: std::os::fd::RawFd, readable: bool, writable: bool) -> std::io::Result<()> {
    let me = match current_pid() {
        Some(pid) => pid,
        None => panic!("wait_*() called outside an actor"),
    };
    let _np = NoPreempt::enter();
    let epoch = begin_wait();
    with_runtime(|inner| {
        let mut io = match inner.io.lock() {
            Ok(io) => io,
            Err(e) => panic!("smarm: io lock poisoned (core corrupt): {e}"),
        };
        match io.as_mut() {
            Some(io) => io.epoll_register(fd, me, epoch, readable, writable),
            None => panic!("io thread not started"),
        }
    })?;

    // If a terminal stop unwinds us out of the park below, the registration
    // must not outlive us: a stale `waiters` entry fails every future
    // `wait_*` on this fd with AlreadyExists, and the kernel-side ADD leaks
    // until the fd happens to be reused. Clean up iff the entry is still
    // ours — a `FdReady` racing the stop may have consumed it already (it
    // removes + DELs under the io lock), after which the fd may even carry
    // ANOTHER actor's fresh registration; in that case touch nothing.
    struct Dereg {
        fd: std::os::fd::RawFd,
        me: Pid,
        epoch: u32,
    }
    impl Drop for Dereg {
        fn drop(&mut self) {
            with_runtime(|inner| {
                let mut io = match inner.io.lock() {
                    Ok(io) => io,
                    Err(e) => panic!("smarm: io lock poisoned (core corrupt): {e}"),
                };
                if let Some(io) = io.as_mut() {
                    if io.waiters.get(&self.fd) == Some(&(self.me, self.epoch)) {
                        io.waiters.remove(&self.fd);
                        io.epoll_deregister(self.fd);
                    }
                }
            });
        }
    }
    let guard = Dereg { fd, me, epoch };
    park_current();
    // Normal wake: the FdReady path removed the entry and DEL'd the fd
    // before the unpark, so the guard's check would be a guaranteed no-op —
    // skip the io lock on the hot path. (Dereg owns nothing; forget leaks
    // no resource.)
    std::mem::forget(guard);
    Ok(())
}

// ---------------------------------------------------------------------------
// FdArm — fd readiness as a select arm (RFC 008)
// ---------------------------------------------------------------------------

/// An fd-readiness arm for [`crate::select`] / [`crate::select_timeout`]:
/// ready when the fd is readable (resp. writable), composable with channel
/// receivers on one wait epoch. Phase-1 rules apply: one waiter per fd at a
/// time, one direction per arm (duplex on a single fd needs `dup`; epoll
/// registrations key on the open file description, so dup'd fds register
/// independently).
pub struct FdArm {
    fd: std::os::fd::RawFd,
    readable: bool,
    writable: bool,
}

impl FdArm {
    pub fn readable(fd: std::os::fd::RawFd) -> Self {
        FdArm { fd, readable: true, writable: false }
    }

    pub fn writable(fd: std::os::fd::RawFd) -> Self {
        FdArm { fd, readable: false, writable: true }
    }
}

impl crate::channel::sealed::Sealed for FdArm {}

impl crate::channel::Selectable for FdArm {
    /// Ready-now check is a zero-timeout `poll(2)`; if the requested events
    /// are pending the wait is retired without registering (`Ok(false)`,
    /// the channel-arm contract). Otherwise register with the io thread —
    /// every failure surfaces as `Err` (EBADF including a closed-fd
    /// POLLNVAL, EMFILE on the epoll set, AlreadyExists for a second
    /// waiter on the fd): the fallible-out, nothing-left-behind rule, in
    /// deviation from RFC 008's permanently-ready lean, which would spin a
    /// consumer whose fd is healthy but unregistrable (EMFILE).
    fn sel_register(&self, pid: Pid, epoch: u32) -> std::io::Result<bool> {
        if poll_events(self.fd, self.readable, self.writable)? {
            return Ok(false);
        }
        with_runtime(|inner| {
            let mut io = match inner.io.lock() {
                Ok(io) => io,
                Err(e) => panic!("smarm: io lock poisoned (core corrupt): {e}"),
            };
            match io.as_mut() {
                Some(io) => {
                    io.epoll_register(self.fd, pid, epoch, self.readable, self.writable)
                }
                None => panic!("io thread not started"),
            }
        })?;
        Ok(true)
    }

    /// Classification is the same zero-timeout poll: a pure function of fd
    /// state, independent of the registration the cleanup pass removed. An
    /// error here (EBADF: fd closed mid-wait) reports READY — the
    /// consumer's read/write surfaces the errno; a dead arm is an event,
    /// not a hang.
    fn sel_ready(&self) -> bool {
        poll_events(self.fd, self.readable, self.writable).unwrap_or(true)
    }

    /// `wait_fd`'s `Dereg` compare, verbatim: remove the waiters entry and
    /// kernel-side registration iff the entry is still `(pid, epoch)`-ours.
    /// A `FdReady` racing the wake may have consumed it (it removes + DELs
    /// under the io lock), after which the fd may even carry ANOTHER
    /// actor's fresh registration; in that case touch nothing.
    fn sel_unregister(&self, pid: Pid, epoch: u32) {
        with_runtime(|inner| {
            let mut io = match inner.io.lock() {
                Ok(io) => io,
                Err(e) => panic!("smarm: io lock poisoned (core corrupt): {e}"),
            };
            if let Some(io) = io.as_mut() {
                if io.waiters.get(&self.fd) == Some(&(pid, epoch)) {
                    io.waiters.remove(&self.fd);
                    io.epoll_deregister(self.fd);
                }
            }
        });
    }

    fn sel_eager_cleanup(&self) -> bool {
        true
    }
}

/// Zero-timeout `poll(2)`: are any of the requested events (or ERR/HUP,
/// which make the consumer's read/write fail loudly rather than park
/// forever) pending on `fd`? POLLNVAL maps to `Err(EBADF)`.
fn poll_events(fd: std::os::fd::RawFd, readable: bool, writable: bool) -> std::io::Result<bool> {
    let mut events: libc::c_short = 0;
    if readable {
        events |= libc::POLLIN;
    }
    if writable {
        events |= libc::POLLOUT;
    }
    let mut pfd = libc::pollfd { fd, events, revents: 0 };
    loop {
        let r = unsafe { libc::poll(&mut pfd, 1, 0) };
        if r < 0 {
            let e = std::io::Error::last_os_error();
            if e.kind() == std::io::ErrorKind::Interrupted {
                continue;
            }
            return Err(e);
        }
        if r == 0 {
            return Ok(false);
        }
        if pfd.revents & libc::POLLNVAL != 0 {
            return Err(std::io::Error::from_raw_os_error(libc::EBADF));
        }
        return Ok(pfd.revents & (events | libc::POLLERR | libc::POLLHUP) != 0);
    }
}

/// Wait until `fd` is readable or `timeout` elapses: `Ok(true)` = ready,
/// `Ok(false)` = timed out. A one-arm [`crate::try_select_timeout`].
pub fn wait_readable_timeout(
    fd: std::os::fd::RawFd,
    timeout: std::time::Duration,
) -> std::io::Result<bool> {
    let arm = FdArm::readable(fd);
    Ok(crate::channel::try_select_timeout(&[&arm], timeout)?.is_some())
}

/// Wait until `fd` is writable or `timeout` elapses: `Ok(true)` = ready,
/// `Ok(false)` = timed out.
pub fn wait_writable_timeout(
    fd: std::os::fd::RawFd,
    timeout: std::time::Duration,
) -> std::io::Result<bool> {
    let arm = FdArm::writable(fd);
    Ok(crate::channel::try_select_timeout(&[&arm], timeout)?.is_some())
}

pub fn read(fd: std::os::fd::RawFd, buf: &mut [u8]) -> std::io::Result<usize> {
    wait_readable(fd)?;
    let n = unsafe { libc::read(fd, buf.as_mut_ptr() as *mut _, buf.len()) };
    if n < 0 { Err(std::io::Error::last_os_error()) } else { Ok(n as usize) }
}

pub fn write(fd: std::os::fd::RawFd, buf: &[u8]) -> std::io::Result<usize> {
    wait_writable(fd)?;
    let n = unsafe { libc::write(fd, buf.as_ptr() as *const _, buf.len()) };
    if n < 0 { Err(std::io::Error::last_os_error()) } else { Ok(n as usize) }
}

// ---------------------------------------------------------------------------
// register_supervisor_channel
// ---------------------------------------------------------------------------

pub fn register_supervisor_channel(pid: Pid, sender: Sender<Signal>) {
    with_runtime(|inner| {
        let slot = inner.slot_at(pid)
            .unwrap_or_else(|| panic!("register_supervisor_channel: pid {:?} not found", pid));
        let mut cold = slot.cold.lock();
        assert_eq!(
            slot.generation(), pid.generation(),
            "register_supervisor_channel: pid {:?} not found", pid
        );
        cold.supervisor_channel = Some(sender);
    });
}

// ---------------------------------------------------------------------------
// Legacy run() — convenience wrapper
// ---------------------------------------------------------------------------

/// Single-threaded runtime entry point (backwards-compatible wrapper).
/// Equivalent to `runtime::init(Config::exact(1)).run(f)`.
pub fn run<F: FnOnce() + Send + 'static>(f: F) {
    crate::runtime::init(crate::runtime::Config::exact(1)).run(f);
}

#[cfg(all(test, not(loom)))]
mod send_after_to_tests {
    use super::*;
    use crate::channel::channel;
    use std::sync::atomic::{AtomicBool, Ordering};
    use std::sync::Arc;
    use std::time::Duration;

    // A fired send_after_to lands its message on the caller's own channel.
    #[test]
    fn delivers_onto_the_channel() {
        run(|| {
            let (tx, rx) = channel::<u32>();
            send_after_to(Duration::from_millis(10), tx, 42);
            // recv parks until the scheduler fires the timer thunk.
            assert_eq!(rx.recv().unwrap(), 42);
        });
    }

    // cancel_timer before the deadline prevents delivery and reports the race
    // win; the channel then closes with no message once the lone sender (moved
    // into the now-discarded thunk) is gone.
    #[test]
    fn cancel_prevents_delivery() {
        run(|| {
            let (tx, rx) = channel::<u32>();
            let id = send_after_to(Duration::from_millis(50), tx, 7);
            assert!(cancel_timer(id), "cancel before fire should win the race");
            // No delivery: the discarded thunk drops the only sender, so recv
            // sees a closed channel rather than the value.
            assert!(rx.recv().is_err());
        });
    }

    // The thunk runs on the scheduler thread; a closed receiver makes the send
    // a harmless no-op (Erlang send_after semantics) rather than a panic.
    #[test]
    fn send_to_closed_channel_is_harmless() {
        let reached = Arc::new(AtomicBool::new(false));
        let r2 = reached.clone();
        run(move || {
            let (tx, rx) = channel::<u32>();
            send_after_to(Duration::from_millis(10), tx, 1);
            drop(rx); // receiver gone before the timer fires
            crate::sleep(Duration::from_millis(30));
            r2.store(true, Ordering::SeqCst);
        });
        assert!(reached.load(Ordering::SeqCst), "runtime survived the dead-channel fire");
    }
}