//! Unbounded multi-producer, single-consumer channels: how actors talk to
//! each other.
//!
//! A channel is a queue with a typed [`Sender`] on one end and a typed
//! [`Receiver`] on the other. Any number of actors can hold a clone of the
//! `Sender` and push messages onto the same queue; exactly one [`Receiver`]
//! reads them back out, in the order they arrived. This is the basic wiring
//! smarm's other actor primitives (`gen_server`, `pg`, the registry) are all
//! built out of, and it is directly usable on its own for a worker that just
//! needs an inbox.
//!
//! ## A first channel
//!
//! ```
//! use smarm::{channel, run, spawn};
//!
//! run(|| {
//!     let (tx, rx) = channel::<u64>();
//!
//!     let worker = spawn(move || {
//!         // Blocks until a message arrives.
//!         let n = rx.recv().unwrap();
//!         assert_eq!(n, 42);
//!
//!         // Once every Sender is dropped, recv() reports the channel closed
//!         // instead of blocking forever.
//!         assert!(rx.recv().is_err());
//!     });
//!
//!     tx.send(42).unwrap();
//!     drop(tx); // last sender gone: the channel is now closed
//!     worker.join().unwrap();
//! });
//! ```
//!
//! ## Sending
//!
//! [`Sender`] is cheaply clonable: hand a clone to every actor that needs to
//! push messages into this queue. The channel stays open as long as at least
//! one clone exists; [`Sender::send`] never blocks and always succeeds while
//! the channel is open, since the queue is unbounded. Once the [`Receiver`]
//! has been dropped, `send` returns the message back to you in
//! [`SendError`] instead of delivering it.
//!
//! ## Receiving
//!
//! There is exactly one [`Receiver`] per channel (it is not clonable).
//! [`Receiver::recv`] returns the next message in arrival order, parking the
//! calling actor if the queue is currently empty. Once every `Sender` has
//! been dropped and the queue has been drained, `recv` stops parking and
//! returns [`RecvError`] instead, so a receiver never blocks forever waiting
//! on senders that are never coming back.
//!
//! Beyond plain `recv`, three variants cover the common needs:
//!
//! - [`Receiver::try_recv`]: never parks: reports an empty-but-open channel
//!   as `Ok(None)` instead of waiting.
//! - [`Receiver::recv_timeout`]: parks, but gives up and returns
//!   [`RecvTimeoutError::Timeout`] if no message arrives before a deadline.
//! - [`Receiver::recv_match`] / [`Receiver::try_recv_match`]: selective
//!   receive. Instead of taking whatever is at the front of the queue, pick
//!   out the first message matching a predicate, leaving the rest queued in
//!   order. Handy for an actor that wants to prioritise one kind of message
//!   over others already waiting.
//!
//! ## Waiting on several channels: `select`
//!
//! [`select`] parks an actor across several receivers at once and reports
//! the index of the first one that is ready (has a message queued, or has
//! been closed). [`select_timeout`] adds a deadline, the way `recv_timeout`
//! does for a single channel. See their docs for the full contract,
//! including the priority-order and no-fairness guarantee.
//!
//! ## Implementation notes
//!
//! The queue and its bookkeeping live behind `Arc<RawMutex<Inner<T>>>`
//! rather than a `std::sync::Mutex`, so that a channel can be freely shared
//! and sent across the OS threads backing the multi-scheduler runtime.
//! `RawMutex` matters here for a subtler reason too: an ordinary pthread
//! mutex can be released from a different OS thread than the one that took
//! it (smarm's preemption can migrate a timesliced actor between scheduler
//! threads mid-critical-section), and doing that to a `std::sync::Mutex` is
//! undefined behavior. `RawMutex` disables preemption for the guard's short
//! lifetime instead, so the release always happens on the thread that
//! acquired it, and it has no poisoning to worry about besides. Channel
//! locks are cheap and are never held across another lock acquisition or a
//! blocking call; the predicate passed to `recv_match` runs under this lock,
//! which is why it needs to stay cheap, pure, and must not call back into
//! the same channel.

use crate::pid::Pid;
use crate::raw_mutex::RawMutex;
use std::collections::VecDeque;
use std::sync::Arc;

/// Create a new channel and return its `(Sender, Receiver)` halves.
///
/// The channel is unbounded (no capacity limit) and open until every
/// `Sender` has been dropped.
pub fn channel<T>() -> (Sender<T>, Receiver<T>) {
    let inner = Arc::new(RawMutex::new_channel(Inner {
        queue: VecDeque::new(),
        parked_receiver: None,
        senders: 1,
        receiver_alive: true,
    }));
    (Sender { inner: inner.clone() }, Receiver { inner })
}

struct Inner<T> {
    queue: VecDeque<T>,
    /// The parked receiver's `(pid, park-epoch)`, if one is currently
    /// waiting. The epoch identifies exactly which wait this is, so a waker
    /// left over from a wait that already ended (a losing `select` arm, a
    /// `recv_timeout` whose timer fired after it was already satisfied) is
    /// inert and does nothing when it fires.
    parked_receiver: Option<(Pid, u32)>,
    senders: usize,
    receiver_alive: bool,
}

/// The sending half of a channel, created by [`channel`]. Clonable: every
/// clone pushes onto the same queue, and the channel stays open as long as
/// any clone is alive. Dropping the last `Sender` closes the channel, which
/// wakes a parked [`Receiver`] so it can observe the closure.
pub struct Sender<T> {
    inner: Arc<RawMutex<Inner<T>>>,
}

/// The receiving half of a channel, created by [`channel`]. Not clonable:
/// a channel has exactly one receiver. Reads messages in the order they
/// were sent, via [`recv`](Receiver::recv) and its variants.
pub struct Receiver<T> {
    inner: Arc<RawMutex<Inner<T>>>,
}

/// Returned by [`Sender::send`] when the channel's [`Receiver`] has already
/// been dropped. Carries the message back so it is never silently lost;
/// recover it with `.0` or by matching.
#[derive(Debug, PartialEq, Eq)]
pub struct SendError<T>(pub T);

/// Returned by [`Receiver::recv`] (and the other receive methods, in their
/// own error types) when the channel is closed: every `Sender` has been
/// dropped and no message is left queued.
#[derive(Debug, PartialEq, Eq, Clone, Copy)]
pub struct RecvError;

impl std::fmt::Display for RecvError {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        write!(f, "channel closed")
    }
}

impl std::error::Error for RecvError {}

/// Returned by [`Receiver::recv_timeout`].
#[derive(Debug, PartialEq, Eq, Clone, Copy)]
pub enum RecvTimeoutError {
    /// The deadline passed with no message available.
    Timeout,
    /// Every sender was dropped with no message available. The
    /// timeout-aware counterpart of plain [`RecvError`].
    Disconnected,
}

impl std::fmt::Display for RecvTimeoutError {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        match self {
            RecvTimeoutError::Timeout => write!(f, "recv timed out"),
            RecvTimeoutError::Disconnected => write!(f, "channel closed"),
        }
    }
}

impl std::error::Error for RecvTimeoutError {}

impl<T> Clone for Sender<T> {
    fn clone(&self) -> Self {
        self.inner.lock().senders += 1;
        Sender { inner: self.inner.clone() }
    }
}

impl<T> Drop for Sender<T> {
    fn drop(&mut self) {
        let unpark = {
            let mut g = self.inner.lock();
            g.senders -= 1;
            // Wake the parked receiver on the last sender drop regardless of
            // whether the queue is empty. A plain `recv` only ever parks on an
            // empty queue (so this is unchanged for it), but a selective
            // `recv_match` may be parked on a non-empty queue holding only
            // non-matching messages. It must wake to observe closure and
            // return Err rather than sleep forever.
            if g.senders == 0 {
                g.parked_receiver.take()
            } else {
                None
            }
        };
        if let Some((pid, epoch)) = unpark {
            crate::scheduler::unpark_at(pid, epoch);
        }
    }
}

impl<T> Drop for Receiver<T> {
    fn drop(&mut self) {
        // The only consumer is gone: queued messages can never be delivered.
        // Drop them now instead of leaving them queued until the last Sender
        // happens to go away, which can be long after this receiver's owner
        // has exited if some other part of the runtime is still holding a
        // clone of the Sender. Draining runs each queued message's own drop
        // glue, which matters for a gen_server call: dropping a queued call
        // envelope drops its reply channel too, which wakes the caller with
        // an error instead of leaving it parked forever. Drain under the
        // lock, then run the drops after releasing it, since a message's
        // drop glue may itself touch a different channel or the scheduler.
        let drained = {
            let mut g = self.inner.lock();
            g.receiver_alive = false;
            std::mem::take(&mut g.queue)
        };
        drop(drained);
    }
}

impl<T> Sender<T> {
    /// Number of messages currently queued and not yet received. For
    /// introspection and monitoring; takes the channel's internal lock, so
    /// avoid calling it from a hot path.
    pub(crate) fn queued_len(&self) -> usize {
        self.inner.lock().queue.len()
    }

    /// Push `value` onto the channel. Succeeds unconditionally as long as
    /// the [`Receiver`] is still alive: the queue has no capacity limit, so
    /// this never blocks and never fails except when the channel is closed,
    /// in which case `value` comes back in [`SendError`].
    pub fn send(&self, value: T) -> Result<(), SendError<T>> {
        let unpark = {
            let mut g = self.inner.lock();
            if !g.receiver_alive {
                return Err(SendError(value));
            }
            g.queue.push_back(value);
            g.parked_receiver.take()
        };
        if let Some((pid, epoch)) = unpark {
            crate::te!(crate::trace::Event::Send { sender: crate::actor::current_pid().unwrap_or(crate::pid::Pid::new(u32::MAX, u32::MAX)), receiver: Some(pid) });
            crate::scheduler::unpark_at(pid, epoch);
        } else {
            crate::te!(crate::trace::Event::Send { sender: crate::actor::current_pid().unwrap_or(crate::pid::Pid::new(u32::MAX, u32::MAX)), receiver: None });
        }
        Ok(())
    }
}

impl<T> Receiver<T> {
    /// Block until a message is available and return it. Messages come back
    /// in the order they were sent. If the queue is empty and every
    /// [`Sender`] has already been dropped, returns [`RecvError`] instead of
    /// blocking forever.
    pub fn recv(&self) -> Result<T, RecvError> {
        loop {
            {
                let mut g = self.inner.lock();
                if let Some(v) = g.queue.pop_front() {
                    crate::preempt::note_message_received();
                    return Ok(v);
                }
                if g.senders == 0 {
                    return Err(RecvError);
                }
                let me = match crate::actor::current_pid() {
                    Some(me) => me,
                    None => panic!("smarm: recv() called outside an actor"),
                };
                debug_assert!(
                    g.parked_receiver.is_none_or(|(p, _)| p == me),
                    "channel has more than one receiver"
                );
                // begin_wait is lock-free, so it's legal under the Channel lock;
                // registering in the same critical section makes the epoch
                // atomic with the senders' view of the registration.
                g.parked_receiver = Some((me, crate::scheduler::begin_wait()));
                crate::te!(crate::trace::Event::RecvPark(me));
            }
            // Release the lock before parking: the unparker will need it.
            crate::scheduler::park_current();
            // Woken up. Record it before looping to check the queue.
            crate::te!(crate::trace::Event::RecvWake(match crate::actor::current_pid() {
                Some(p) => p,
                None => panic!("smarm: RecvWake outside an actor (core corrupt)"),
            }));
        }
    }

    /// Like [`recv`](Self::recv), but gives up and returns
    /// [`RecvTimeoutError::Timeout`] if no message has arrived by the time
    /// `timeout` elapses.
    ///
    /// If a message arrives at essentially the same moment the deadline
    /// passes, the message wins: you get `Ok` rather than `Timeout`. If
    /// every sender is dropped before a message arrives or the deadline
    /// passes, you get [`RecvTimeoutError::Disconnected`].
    ///
    /// `Duration::ZERO` is a valid timeout: it still gives any
    /// already-queued message a chance to be returned, and only then
    /// reports `Timeout`.
    pub fn recv_timeout(&self, timeout: std::time::Duration) -> Result<T, RecvTimeoutError>
    where
        T: Send + 'static,
    {
        let me = match crate::actor::current_pid() {
            Some(me) => me,
            None => panic!("smarm: recv_timeout() called outside an actor"),
        };

        // Fast path + wait registration, one critical section.
        let epoch;
        {
            let mut g = self.inner.lock();
            if let Some(v) = g.queue.pop_front() {
                crate::preempt::note_message_received();
                return Ok(v);
            }
            if g.senders == 0 {
                return Err(RecvTimeoutError::Disconnected);
            }
            debug_assert!(
                g.parked_receiver.is_none_or(|(p, _)| p == me),
                "channel has more than one receiver"
            );
            epoch = crate::scheduler::begin_wait();
            g.parked_receiver = Some((me, epoch));
            crate::te!(crate::trace::Event::RecvPark(me));
        }

        // Arm the timer after releasing the channel lock (insert takes the
        // timers lock; never nest under a Channel lock). A send or even the
        // timer itself may unpark us before we park; the runtime's wake
        // protocol makes the park below return immediately in that case.
        let deadline = crate::timer::deadline_from_now(timeout);
        let target: std::sync::Arc<dyn crate::timer::TimerTarget> = self.inner.clone();
        crate::scheduler::insert_wait_timer(deadline, me, target, epoch);

        crate::scheduler::park_current();
        crate::te!(crate::trace::Event::RecvWake(match crate::actor::current_pid() {
            Some(p) => p,
            None => panic!("smarm: RecvWake outside an actor (core corrupt)"),
        }));
        let mut g = self.inner.lock();
        if let Some(v) = g.queue.pop_front() {
            crate::preempt::note_message_received();
            return Ok(v);
        }
        if g.senders == 0 {
            return Err(RecvTimeoutError::Disconnected);
        }
        Err(RecvTimeoutError::Timeout)
    }

    /// Selective receive: find and return the first queued message for
    /// which `pred` returns `true`, leaving every other message in the
    /// queue untouched and in order. Useful when an actor's inbox mixes
    /// message kinds and it wants to handle one kind out of turn, without
    /// discarding the rest.
    ///
    /// If nothing queued matches, this blocks and re-checks every time a new
    /// message arrives, the same way [`recv`](Self::recv) blocks on an empty
    /// queue: a selective receiver can be waiting even while the queue holds
    /// messages, just none that match yet. Returns [`RecvError`] only once
    /// the channel is closed and still nothing matches.
    ///
    /// `pred` runs while the channel is locked, so keep it cheap, side
    /// effect free, and make sure it never calls back into this same
    /// channel. It takes `&T` and is called fresh on every scan (not `FnMut`
    /// with running state), so it should judge each message purely on its
    /// own content.
    pub fn recv_match<F>(&self, pred: F) -> Result<T, RecvError>
    where
        F: Fn(&T) -> bool,
    {
        loop {
            {
                let mut g = self.inner.lock();
                if let Some(i) = g.queue.iter().position(&pred) {
                    // position() found it, so remove() returns Some.
                    crate::preempt::note_message_received();
                    let v = match g.queue.remove(i) {
                        Some(v) => v,
                        None => panic!("smarm: channel queue.remove after position (logic bug)"),
                    };
                    return Ok(v);
                }
                if g.senders == 0 {
                    // Closed and nothing queued can ever match.
                    return Err(RecvError);
                }
                let me = match crate::actor::current_pid() {
                    Some(me) => me,
                    None => panic!("smarm: recv_match() called outside an actor"),
                };
                debug_assert!(
                    g.parked_receiver.is_none_or(|(p, _)| p == me),
                    "channel has more than one receiver"
                );
                g.parked_receiver = Some((me, crate::scheduler::begin_wait()));
                crate::te!(crate::trace::Event::RecvPark(me));
            }
            // Release the lock before parking: the unparker will need it.
            crate::scheduler::park_current();
            crate::te!(crate::trace::Event::RecvWake(match crate::actor::current_pid() {
                Some(p) => p,
                None => panic!("smarm: RecvWake outside an actor (core corrupt)"),
            }));
        }
    }

    /// The non-blocking counterpart of [`recv_match`](Self::recv_match):
    /// returns immediately either way. `Ok(Some(v))` if a queued message
    /// matched `pred` (removed; the rest stay queued in order), `Ok(None)`
    /// if the channel is open but nothing currently matches, `Err(RecvError)`
    /// if the channel is closed and nothing matches. Same predicate contract
    /// as `recv_match`.
    pub fn try_recv_match<F>(&self, pred: F) -> Result<Option<T>, RecvError>
    where
        F: Fn(&T) -> bool,
    {
        let mut g = self.inner.lock();
        if let Some(i) = g.queue.iter().position(&pred) {
            crate::preempt::note_message_received();
            let v = match g.queue.remove(i) {
                Some(v) => v,
                None => panic!("smarm: channel queue.remove after position (logic bug)"),
            };
            return Ok(Some(v));
        }
        if g.senders == 0 {
            return Err(RecvError);
        }
        Ok(None)
    }

    /// The non-blocking counterpart of [`recv`](Self::recv): returns
    /// immediately either way. `Ok(Some(v))` if a message was queued,
    /// `Ok(None)` if the channel is open but currently empty, `Err(RecvError)`
    /// if the channel is closed and the queue is drained.
    pub fn try_recv(&self) -> Result<Option<T>, RecvError> {
        let mut g = self.inner.lock();
        if let Some(v) = g.queue.pop_front() {
            crate::preempt::note_message_received();
            return Ok(Some(v));
        }
        if g.senders == 0 {
            return Err(RecvError);
        }
        Ok(None)
    }
}

// ---------------------------------------------------------------------------
// TimerTarget: the expiry half of recv_timeout
// ---------------------------------------------------------------------------

impl<T: Send + 'static> crate::timer::TimerTarget for RawMutex<Inner<T>> {
    fn on_timeout(&self, pid: Pid, epoch: u32) {
        // Cancel the wait only if THIS wait (epoch match) is still
        // registered. If a sender already took `parked_receiver`, the
        // receiver is waking with a message: message wins, the timer
        // no-ops. If a later wait by the same receiver is registered, the
        // epoch mismatches: stale entry, no-op. (unpark_at would fail its
        // internal check in either case anyway; checking under the lock
        // keeps the registration bookkeeping exact.)
        let unpark = {
            let mut g = self.lock();
            if g.parked_receiver == Some((pid, epoch)) {
                g.parked_receiver = None;
                true
            } else {
                false
            }
        };
        // Unpark outside the channel lock: it may take the run-queue lock;
        // legal under a Channel lock, but pointless to nest.
        if unpark {
            crate::scheduler::unpark_at(pid, epoch);
        }
    }
}

// ---------------------------------------------------------------------------
// select: ready-index wait over multiple receivers
// ---------------------------------------------------------------------------

pub(crate) mod sealed {
    pub trait Sealed {}
}
impl<T> sealed::Sealed for Receiver<T> {}

/// An arm of a [`select`]: something you can wait on alongside other arms
/// and be told when it becomes ready. Implemented by [`Receiver`]; sealed
/// (cannot be implemented outside this crate), since the registration
/// contract below is part of the runtime's internal wake protocol.
///
/// Contract (all under the arm's own lock): `sel_register` checks-or-
/// registers atomically. If the arm is ready it does not register and
/// returns `Ok(false)`; otherwise it publishes `(pid, epoch)` where its
/// wakers will find it and returns `Ok(true)`. "Ready" means a receive
/// would not block: a message is queued, or the arm is closed. `Err` means
/// the arm could not register at all (only fd arms can fail; channel
/// registration always succeeds), and the wait must be retired and earlier
/// eager-cleanup arms unregistered.
pub trait Selectable: sealed::Sealed {
    #[doc(hidden)]
    fn sel_register(&self, pid: Pid, epoch: u32) -> std::io::Result<bool>;
    #[doc(hidden)]
    fn sel_ready(&self) -> bool;
    /// Remove this arm's `(pid, epoch)` registration if, and only if, it is
    /// still in place. Default no-op: a losing channel arm's stale
    /// registration is harmless and self-cleans. Fd arms override this:
    /// their staleness would otherwise leave the fd unusable for future
    /// selects, so they need an eager cleanup pass.
    #[doc(hidden)]
    fn sel_unregister(&self, _pid: Pid, _epoch: u32) {}
    /// Whether this arm requires the eager cleanup pass at all. Gates the
    /// post-wake `sel_unregister` sweep so channel-only selects keep their
    /// cheap, cleanup-free path.
    #[doc(hidden)]
    fn sel_eager_cleanup(&self) -> bool {
        false
    }
}

impl<T> Selectable for Receiver<T> {
    fn sel_register(&self, pid: Pid, epoch: u32) -> std::io::Result<bool> {
        let mut g = self.inner.lock();
        if !g.queue.is_empty() || g.senders == 0 {
            return Ok(false);
        }
        debug_assert!(
            g.parked_receiver.is_none_or(|(p, _)| p == pid),
            "channel has more than one receiver"
        );
        g.parked_receiver = Some((pid, epoch));
        Ok(true)
    }

    fn sel_ready(&self) -> bool {
        let g = self.inner.lock();
        !g.queue.is_empty() || g.senders == 0
    }
}

/// Wait on several channels at once and return the index of the first one
/// that is ready, instead of blocking on just one with [`Receiver::recv`].
///
/// "Ready" means a receive on that arm would not block: a message is
/// queued, or the arm is closed (so the caller's own `try_recv` observes
/// the disconnect: a dead arm is something to react to, not something to
/// hang on). `select` only tells you which arm is ready; read the actual
/// message yourself, typically with [`Receiver::try_recv`] on that arm.
///
/// A closed arm stays ready forever. Once you have observed its disconnect,
/// drop it from the arm set you pass in next time: otherwise, under the
/// priority order below, it would win every subsequent call and starve
/// every arm listed after it.
///
/// Arms are checked **in order**: index 0 is the highest priority, both
/// when checking immediately and after being woken. This is a deliberate,
/// documented guarantee, not an accident of implementation: put a control
/// or shutdown channel first so it is always noticed promptly. The
/// flip side is that there is **no fairness guarantee**: a busy arm 0 can
/// starve arm 1 indefinitely by design.
///
/// One actor can `select` on a channel and later plain `recv` on it (or
/// `select` again on an overlapping set of arms) with no restriction. What
/// stays illegal is what was always illegal for a channel: two *different*
/// actors receiving on the same one.
///
/// Panics if `arms` is empty, if called outside an actor, or if an fd arm
/// fails to register (see [`try_select`] for the fallible form; a
/// channel-only `select` can never fail).
pub fn select(arms: &[&dyn Selectable]) -> usize {
    match try_select(arms) {
        Ok(i) => i,
        Err(e) => panic!("smarm: select() fd arm failed to register (use try_select): {e}"),
    }
}

/// The fallible form of [`select`]: `Err` when an arm fails to register.
/// Only fd arms can fail this way (for example, the file descriptor is
/// invalid, or something else is already waiting on it); a channel-only
/// select can never fail. On `Err` the wait is fully retired and no
/// registration is left behind: every arm registered before the failing
/// one has been unregistered.
pub fn try_select(arms: &[&dyn Selectable]) -> std::io::Result<usize> {
    assert!(!arms.is_empty(), "select() on an empty arm list");
    let me = match crate::actor::current_pid() {
        Some(me) => me,
        None => panic!("smarm: select() called outside an actor"),
    };
    loop {
        let epoch = crate::scheduler::begin_wait();
        if let Some(i) = register_arms(me, epoch, arms)? {
            return Ok(i);
        }

        // Stale fd registrations are not harmless (a losing fd arm's
        // leftover registration can make the fd unusable for the next
        // select until a kernel event happens to clear it), so selects
        // containing fd arms run an eager cleanup pass after the park,
        // including when a terminal stop unwinds out of it, via the guard.
        // Channel-only selects skip all of it: `eager` is false, the guard
        // is disarmed, and the loser-arm self-cleaning story is unchanged.
        let eager = arms.iter().any(|a| a.sel_eager_cleanup());
        let mut guard = UnregisterGuard { arms, me, epoch, armed: eager };

        crate::scheduler::park_current();

        if eager {
            unregister_arms(arms, me, epoch);
        }
        guard.armed = false;
        drop(guard);

        // Woken precisely: an arm's send (message) or last-sender drop
        // (closure) is what woke us, and both leave their arm ready.
        // Return the first ready one, in priority order (which may be a
        // different, higher-priority arm than the one that woke us; its
        // message stays queued and re-reports ready on the next call).
        // Fd arms classify by a fresh zero-timeout poll, so they too are
        // a pure function of current state, independent of the
        // registration the cleanup pass just removed.
        for (i, arm) in arms.iter().enumerate() {
            if arm.sel_ready() {
                return Ok(i);
            }
        }
        // Unreachable in practice (a stop wake unwinds out of
        // park_current before we get here). Defensive: re-open the wait
        // and re-register; stale own-registrations are overwritten
        // (channels) or were removed by the cleanup pass above (fds).
    }
}

/// Eager-cleanup sweep: remove every fd arm's registration that is still
/// ours. No-op per channel arm (one virtual call); one io-lock visit per
/// fd arm.
fn unregister_arms(arms: &[&dyn Selectable], me: Pid, epoch: u32) {
    for arm in arms {
        if arm.sel_eager_cleanup() {
            arm.sel_unregister(me, epoch);
        }
    }
}

// Stop-unwind twin of the explicit cleanup pass: a terminal stop unwinds
// out of `park_current`, and a registered fd arm must not outlive its
// actor. Disarmed on the normal path after the explicit pass runs; never
// armed when no fd arm is registered, keeping the channel-only path
// guard-free in effect.
struct UnregisterGuard<'a> {
    arms: &'a [&'a dyn Selectable],
    me: Pid,
    epoch: u32,
    armed: bool,
}

impl Drop for UnregisterGuard<'_> {
    fn drop(&mut self) {
        if self.armed {
            unregister_arms(self.arms, self.me, self.epoch);
        }
    }
}

// The registration pass shared by `select` and `select_timeout`: check-or-
// register each arm, in priority order, each atomically under its own lock.
// Cross-arm atomicity is unnecessary: an arm becoming ready right after its
// registration still wakes the caller through the normal wake path.
//
// `Ok(Some(i))` = arm `i` was already ready, the pass stopped, and the wait
// has been fully retired (no park may follow): earlier fd arms are
// unregistered eagerly so none are left dangling. `Err` = an arm failed to
// register; same unwind (earlier fd arms unregistered, wait retired).
// `Ok(None)` = every arm registered successfully; the caller parks.
fn register_arms(
    me: Pid,
    epoch: u32,
    arms: &[&dyn Selectable],
) -> std::io::Result<Option<usize>> {
    for (i, arm) in arms.iter().enumerate() {
        let registered = match arm.sel_register(me, epoch) {
            Ok(r) => r,
            Err(e) => {
                unregister_arms(&arms[..i], me, epoch);
                crate::scheduler::retire_wait();
                return Err(e);
            }
        };
        if !registered {
            unregister_arms(&arms[..i], me, epoch);
            crate::scheduler::retire_wait();
            return Ok(Some(i));
        }
    }
    Ok(None)
}

// The `select_timeout` timer target: stateless, because a wake's cause can
// always be read back off plain channel state (an arm ready, or not). If
// an arm already won before the deadline, this timer's fire is simply
// ignored, the way any other stale wakeup is.
struct SelectTimeout;
impl crate::timer::TimerTarget for SelectTimeout {
    fn on_timeout(&self, pid: Pid, epoch: u32) {
        crate::scheduler::unpark_at(pid, epoch);
    }
}

/// Like [`select`], but gives up and returns `None` if no arm becomes
/// ready before `timeout` elapses.
///
/// All of `select`'s semantics carry over: arms are still checked in
/// priority order, a closed arm is still permanently ready, and there is
/// still no fairness guarantee across arms. A message that arrives at
/// essentially the same moment the deadline passes still wins, the same
/// way [`Receiver::recv_timeout`] resolves that race.
///
/// `Duration::ZERO` is a valid timeout: it still gives an already-ready arm
/// a chance to be reported before falling through to `None`.
///
/// Panics if `arms` is empty, if called outside an actor, or if an fd arm
/// fails to register (see [`try_select_timeout`] for the fallible form; a
/// channel-only select can never fail).
pub fn select_timeout(
    arms: &[&dyn Selectable],
    timeout: std::time::Duration,
) -> Option<usize> {
    match try_select_timeout(arms, timeout) {
        Ok(r) => r,
        Err(e) => panic!(
            "smarm: select_timeout() fd arm failed to register (use try_select_timeout): {e}"
        ),
    }
}

/// The fallible form of [`select_timeout`]: `Err` when an arm fails to
/// register (only fd arms can). On `Err` the wait is fully retired and no
/// registration is left behind on any arm.
pub fn try_select_timeout(
    arms: &[&dyn Selectable],
    timeout: std::time::Duration,
) -> std::io::Result<Option<usize>> {
    assert!(!arms.is_empty(), "select_timeout() on an empty arm list");
    let me = match crate::actor::current_pid() {
        Some(me) => me,
        None => panic!("smarm: select_timeout() called outside an actor"),
    };
    let epoch = crate::scheduler::begin_wait();
    if let Some(i) = register_arms(me, epoch, arms)? {
        return Ok(Some(i)); // ready now: the timer was never armed
    }

    // Arm the timer after the registration pass, outside every channel
    // lock (inserting a timer takes the timers lock).
    let deadline = crate::timer::deadline_from_now(timeout);
    let target: std::sync::Arc<dyn crate::timer::TimerTarget> = std::sync::Arc::new(SelectTimeout);
    crate::scheduler::insert_wait_timer(deadline, me, target, epoch);

    // Same eager-cleanup story as `try_select`: a timer win in particular
    // leaves every fd arm's registration behind, which without this pass
    // would leave those fds unusable until a kernel event happened to
    // clear them.
    let eager = arms.iter().any(|a| a.sel_eager_cleanup());
    let mut guard = UnregisterGuard { arms, me, epoch, armed: eager };

    crate::scheduler::park_current();

    if eager {
        unregister_arms(arms, me, epoch);
    }
    guard.armed = false;
    drop(guard);

    // Woken precisely: an arm (ready below) or the timer (nothing ready).
    Ok(arms.iter().position(|arm| arm.sel_ready()))
}