nuclei 0.5.4

//! The bounded channel: [`channel`], its [`Sender`] and its [`WeakSender`].
//!
//! The channel's admission control is a [`FairSemaphore`] of `buffer`
//! permits. A send takes one permit (a reservation is exactly that), the
//! receiver gives one back per value it takes, and a permit dropped unused
//! goes back too. So the permits free, plus the ones set aside for a waiting
//! reservation, plus the values queued, plus the reservations held, are
//! always `buffer`, and the queue can never hold more than `buffer` values.
//!
//! The semaphore serves reservations strictly in the order they began to
//! wait, as the established API does: a `try_` attempt or a new reservation
//! never takes a slot while an older one waits, and a reservation for many
//! slots keeps the ones set aside for it while it waits for the rest. The
//! receiver closing closes the semaphore: waiting senders end at once, what
//! was set aside for them comes back, and every later reservation fails.

#![allow(unsafe_code)]

use core::fmt;
use core::future::Future;
use core::marker::PhantomPinned;
use core::pin::Pin;
use core::task::{Context, Poll};

use super::chan::{Chan, Limit};
use super::error::{SendError, SendTimeoutError, TrySendError};
use super::permit::{OwnedPermit, Permit, PermitIterator};
use super::queue;
use super::receiver::Receiver;
use crate::sync::fair_semaphore::{FairSemaphore, FairWait};
use crate::sync::internal::{self, Arc};
use crate::sync::raw_semaphore::observed_closed;

/// The bounded flavor's limit: `bound` permits.
pub(super) struct Bounded {
    pub(super) sem: FairSemaphore,
    pub(super) bound: usize,
}

impl Limit for Bounded {
    type Queue<T> = queue::Bounded<T>;

    fn queue<T>(&self) -> queue::Bounded<T> {
        queue::Bounded::with_capacity(self.bound)
    }

    #[inline]
    fn is_idle(&self) -> bool {
        self.sem.available() == self.bound
    }

    #[inline]
    fn release(&self, n: usize) {
        self.sem.release(n);
    }

    fn close(&self) {
        self.sem.close();
    }

    #[inline]
    fn is_closed(&self) -> bool {
        self.sem.is_closed()
    }
}

/// Creates a bounded channel for communicating between tasks with
/// backpressure.
///
/// The channel buffers up to `buffer` messages. Once the buffer is full, a
/// send waits until a message is received. The buffer must hold at least one
/// message.
///
/// Everything sent on a [`Sender`] comes out of the [`Receiver`], in the order
/// it was sent (see [Ordering](super#ordering)). The [`Sender`] can be cloned
/// to send to the same channel from many places; only one [`Receiver`] exists.
///
/// If the [`Receiver`] is closed or dropped, a send returns a
/// [`SendError`] with the value. When every [`Sender`] is dropped, `recv`
/// returns `None` once the buffered messages are received.
///
/// # Panics
///
/// Panics if `buffer` is 0, or larger than
/// [`Semaphore::MAX_PERMITS`](crate::sync::Semaphore::MAX_PERMITS).
///
/// ```
/// use nuclei::sync::mpsc;
///
/// let (tx, mut rx) = mpsc::channel(4);
/// let producer = std::thread::spawn(move || {
///     for i in 0..10 {
///         tx.blocking_send(i).expect("the receiver is still there");
///     }
/// });
/// for i in 0..10 {
///     assert_eq!(rx.blocking_recv(), Some(i));
/// }
/// assert_eq!(rx.blocking_recv(), None);
/// producer.join().unwrap();
/// ```
#[track_caller]
pub fn channel<T>(buffer: usize) -> (Sender<T>, Receiver<T>) {
    assert!(buffer > 0, "mpsc bounded channel requires buffer > 0");
    let chan = Arc::new(Chan::new(Bounded {
        sem: FairSemaphore::new(buffer),
        bound: buffer,
    }));
    let tx = Sender {
        chan: Arc::clone(&chan),
    };
    (tx, Receiver::new(chan))
}

/// Sends values to the associated [`Receiver`].
///
/// Created by [`channel`]. Cloning it is a reference-count increment.
///
/// # Progress
///
/// Wait-free (a bounded number of the caller's own atomic steps whatever other
/// threads do): [`is_closed`](Self::is_closed), [`capacity`](Self::capacity),
/// [`max_capacity`](Self::max_capacity), [`same_channel`](Self::same_channel),
/// [`strong_count`](Self::strong_count), [`weak_count`](Self::weak_count),
/// [`downgrade`](Self::downgrade), `clone` and `drop`. The drop of the last
/// sender ends with a wake of the receiver's waker cell, which is two atomic
/// steps; what the waker then does belongs to the executor.
///
/// Lock-free, not wait-free: [`try_send`](Self::try_send), the `try_reserve`
/// family, and every poll of [`send`](Self::send) and the `reserve` family.
/// A permit is taken with a compare-and-swap that retries when another thread
/// changed the semaphore first; a wait that has to queue pushes a node on the
/// semaphore's lock-free arrival stack and may serve the queue (one pass, plus
/// one for each operation of another thread that lands meanwhile); and the
/// push of a value onto the message queue retries when it loses a slot to
/// another sender and allocates a segment when the last one is full. Some
/// thread's call completes whenever any retries. [`closed`](Self::closed) is
/// lock-free for the same reason.
///
/// A waiting reservation is granted by whichever thread serves the queue once
/// its slots are free (the receiver taking a value, a permit dropped), so a
/// thread the scheduler stops while it serves the queue delays that grant
/// until it runs again; for that moment a `try_` attempt fails as well, even
/// if the last waiting reservation has just left. No call waits for it.
///
/// `blocking_send` parks the calling thread until there is room.
pub struct Sender<T> {
    pub(super) chan: Arc<Chan<T, Bounded>>,
}

/// A sender that does not keep the channel open.
///
/// When every [`Sender`] is dropped and only `WeakSender`s remain, the channel
/// is closed for the receiver. A `WeakSender` sends nothing itself:
/// [`upgrade`](Self::upgrade) turns it into a [`Sender`] while one exists.
///
/// ```
/// use nuclei::sync::mpsc;
///
/// let (tx, _rx) = mpsc::channel::<i32>(15);
/// let weak = tx.downgrade();
/// assert!(weak.upgrade().is_some());
/// drop(tx);
/// assert!(weak.clone().upgrade().is_none());
/// ```
///
/// # Progress
///
/// `clone`, `drop`, [`strong_count`](Self::strong_count) and
/// [`weak_count`](Self::weak_count) are wait-free. [`upgrade`](Self::upgrade)
/// is lock-free, not wait-free: it is a compare-and-swap on the sender count
/// that retries when a sender was cloned or dropped meanwhile.
pub struct WeakSender<T> {
    chan: Arc<Chan<T, Bounded>>,
}

impl<T> Sender<T> {
    /// Sends a value, waiting until there is capacity.
    ///
    /// A successful send means the other end had not hung up when the value
    /// was published. `Ok` does not mean the value will be received: the
    /// receiver can hang up right after.
    ///
    /// # Errors
    ///
    /// If the receiving half is closed, by [`Receiver::close`] or by being
    /// dropped, the function returns an error that holds the value passed to
    /// `send`.
    ///
    /// # Cancel safety
    ///
    /// If the returned future is dropped before it completes, the message was
    /// not sent, and the value is dropped. To keep the value, reserve capacity
    /// with [`reserve`](Self::reserve) and send through the [`Permit`].
    ///
    /// Waiting sends and reservations are served in the order they began to
    /// wait. Dropping the future gives up its place, and a slot it was handed
    /// meanwhile goes on to the next one.
    ///
    /// ```
    /// # use std::{future::Future, pin::pin, task::{Context, Poll, Waker}};
    /// # fn now<F: Future>(f: F) -> F::Output {
    /// #     match pin!(f).poll(&mut Context::from_waker(Waker::noop())) {
    /// #         Poll::Ready(out) => out,
    /// #         Poll::Pending => panic!("not ready"),
    /// #     }
    /// # }
    /// use nuclei::sync::mpsc;
    ///
    /// let (tx, mut rx) = mpsc::channel(1);
    /// now(tx.send(1)).unwrap();
    /// assert_eq!(rx.try_recv(), Ok(1));
    ///
    /// drop(rx);
    /// let err = now(tx.send(2)).unwrap_err();
    /// assert_eq!(err.0, 2);
    /// ```
    pub fn send(&self, value: T) -> impl Future<Output = Result<(), SendError<T>>> + '_ {
        self.sending(value)
    }

    fn sending(&self, value: T) -> Sending<'_, T> {
        Sending {
            chan: &self.chan,
            wait: FairWait::new(),
            value: Some(value),
            _pinned: PhantomPinned,
        }
    }

    /// Sends a value, waiting for room at most `timeout`.
    ///
    /// As [`send`](Self::send), with a deadline: the value comes back in
    /// [`SendTimeoutError::Timeout`] if the channel stayed full until then,
    /// and in [`SendTimeoutError::Closed`] if the receiver is gone. The wait
    /// is on nuclei's timers: on a thread that runs no nuclei shard, the
    /// default runtime fires it.
    ///
    /// # Cancel safety
    ///
    /// As [`send`](Self::send): a dropped future sent nothing, and drops its
    /// value.
    ///
    /// ```
    /// use std::time::Duration;
    /// use nuclei::sync::mpsc::{self, error::SendTimeoutError};
    ///
    /// nuclei::runtime::block_on(async {
    ///     let (tx, _rx) = mpsc::channel(1);
    ///     tx.send_timeout(1, Duration::from_millis(10)).await.unwrap();
    ///     let full = tx.send_timeout(2, Duration::from_millis(10)).await;
    ///     assert!(matches!(full, Err(SendTimeoutError::Timeout(2))));
    /// });
    /// ```
    pub async fn send_timeout(
        &self,
        value: T,
        timeout: std::time::Duration,
    ) -> Result<(), SendTimeoutError<T>> {
        match crate::time::timeout(timeout, self.reserve()).await {
            Ok(Ok(permit)) => {
                permit.send(value);
                Ok(())
            }
            Ok(Err(_)) => Err(SendTimeoutError::Closed(value)),
            Err(_) => Err(SendTimeoutError::Timeout(value)),
        }
    }

    /// Completes when the receiver has been closed or dropped.
    ///
    /// This lets a producer learn that nobody wants its values any more and
    /// stop work early.
    ///
    /// # Cancel safety
    ///
    /// Cancel safe. Once the channel is closed it stays closed, and every later
    /// call completes at once.
    ///
    /// ```
    /// # use std::{future::Future, pin::pin, task::{Context, Poll, Waker}};
    /// use nuclei::sync::mpsc;
    ///
    /// let (tx, rx) = mpsc::channel::<()>(1);
    /// let mut cx = Context::from_waker(Waker::noop());
    /// let mut closed = pin!(tx.closed());
    /// assert_eq!(closed.as_mut().poll(&mut cx), Poll::Pending);
    /// drop(rx);
    /// assert_eq!(closed.as_mut().poll(&mut cx), Poll::Ready(()));
    /// ```
    pub async fn closed(&self) {
        self.chan.closed().await;
    }

    /// Sends a value at once, or says why it cannot be sent.
    ///
    /// Unlike [`send`](Self::send) it never waits: a full buffer is an error.
    /// It never takes a slot ahead of a send or reservation that is waiting.
    ///
    /// # Errors
    ///
    /// [`TrySendError::Full`] if no slot is free or a send or reservation is
    /// waiting, [`TrySendError::Closed`] if the receiver closed or was
    /// dropped. Both hold the value back.
    ///
    /// ```
    /// use nuclei::sync::mpsc::{self, error::TrySendError};
    ///
    /// let (tx, mut rx) = mpsc::channel(1);
    /// tx.try_send("first").unwrap();
    /// assert!(matches!(tx.try_send("second"), Err(TrySendError::Full("second"))));
    /// assert_eq!(rx.try_recv(), Ok("first"));
    /// ```
    pub fn try_send(&self, message: T) -> Result<(), TrySendError<T>> {
        match try_reserve_inner(&self.chan, 1) {
            Ok(()) => {
                self.chan.send(message);
                Ok(())
            }
            Err(refused) => Err(refused.with_value(message)),
        }
    }

    /// Sends a value from a thread that may block.
    ///
    /// It parks the calling thread until there is room. On a nuclei shard (inside one of its tasks) it does not park the thread: it runs a nested `block_on`, so the shard keeps running its other tasks meanwhile. Inside another
    /// runtime's task it blocks that task's thread. It works whether or not
    /// the receiver uses [`Receiver::blocking_recv`].
    ///
    /// A wait whose frame runs a task that opens another blocking frame gives
    /// up its place in line and asks again, at the back, once its frame is on
    /// top again, so nested blocking waits never wait on each other with the
    /// resource free.
    ///
    /// # Panics
    ///
    /// On `wasm32`, where no other thread can make room, when the channel is
    /// full: the call could never end.
    ///
    /// ```
    /// use nuclei::sync::mpsc;
    ///
    /// let (tx, mut rx) = mpsc::channel::<u8>(1);
    /// let sender = std::thread::spawn(move || tx.blocking_send(10).unwrap());
    /// assert_eq!(rx.blocking_recv(), Some(10));
    /// sender.join().unwrap();
    /// ```
    pub fn blocking_send(&self, value: T) -> Result<(), SendError<T>> {
        internal::blocking_in_line(self.sending(value))
    }

    /// Whether the channel is closed: the receiver was closed or dropped.
    ///
    /// ```
    /// use nuclei::sync::mpsc;
    ///
    /// let (tx, rx) = mpsc::channel::<()>(42);
    /// let tx2 = tx.clone();
    /// assert!(!tx.is_closed());
    /// drop(rx);
    /// assert!(tx.is_closed() && tx2.is_closed());
    /// ```
    pub fn is_closed(&self) -> bool {
        self.chan.is_closed()
    }

    /// Waits for capacity for one message and reserves it.
    ///
    /// A [`Permit`] tracks the reserved slot; [`Permit::send`] uses it, and
    /// dropping the permit unused gives the slot back.
    ///
    /// # Errors
    ///
    /// A [`SendError`] if the receiver is closed or dropped, now or while
    /// waiting.
    ///
    /// # Cancel safety
    ///
    /// Dropping the future gives up its place in the queue of waiters, and a
    /// slot it was handed meanwhile goes on to the next one. No slot is held
    /// by a dropped future.
    ///
    /// ```
    /// # use std::{future::Future, pin::pin, task::{Context, Poll, Waker}};
    /// # fn now<F: Future>(f: F) -> F::Output {
    /// #     match pin!(f).poll(&mut Context::from_waker(Waker::noop())) {
    /// #         Poll::Ready(out) => out,
    /// #         Poll::Pending => panic!("not ready"),
    /// #     }
    /// # }
    /// use nuclei::sync::mpsc;
    ///
    /// let (tx, mut rx) = mpsc::channel(1);
    /// let permit = now(tx.reserve()).unwrap();
    /// assert!(tx.try_send(123).is_err(), "the only slot is reserved");
    /// permit.send(456);
    /// assert_eq!(rx.try_recv(), Ok(456));
    /// ```
    pub async fn reserve(&self) -> Result<Permit<'_, T>, SendError<()>> {
        reserve_inner(&self.chan, 1).await?;
        Ok(Permit::new(&self.chan))
    }

    /// Waits for capacity for `n` messages and reserves them.
    ///
    /// The returned [`PermitIterator`] yields `n` permits; dropping it before
    /// it is exhausted gives the rest back. Waiting requests are served in the
    /// order they began to wait, whatever each asks for: the slots that come
    /// free are set aside for the oldest one until it has all `n`, and nothing
    /// behind it is served first. Slots set aside for a request do not count
    /// in [`capacity`](Self::capacity).
    ///
    /// # Errors
    ///
    /// A [`SendError`] if the receiver is closed or dropped, now or while
    /// waiting, or if `n` is more than the channel can ever hold
    /// ([`max_capacity`](Self::max_capacity)). `n == 0` succeeds with an empty
    /// iterator unless the channel is closed.
    ///
    /// # Cancel safety
    ///
    /// As for [`reserve`](Self::reserve): the slots set aside for a dropped
    /// request go on to the next one.
    ///
    /// ```
    /// # use std::{future::Future, pin::pin, task::{Context, Poll, Waker}};
    /// # fn now<F: Future>(f: F) -> F::Output {
    /// #     match pin!(f).poll(&mut Context::from_waker(Waker::noop())) {
    /// #         Poll::Ready(out) => out,
    /// #         Poll::Pending => panic!("not ready"),
    /// #     }
    /// # }
    /// use nuclei::sync::mpsc;
    ///
    /// let (tx, mut rx) = mpsc::channel(2);
    /// let mut permits = now(tx.reserve_many(2)).unwrap();
    /// assert!(tx.try_send(123).is_err());
    /// permits.next().unwrap().send(456);
    /// permits.next().unwrap().send(457);
    /// assert!(permits.next().is_none());
    /// assert_eq!(rx.try_recv(), Ok(456));
    /// assert_eq!(rx.try_recv(), Ok(457));
    /// ```
    pub async fn reserve_many(&self, n: usize) -> Result<PermitIterator<'_, T>, SendError<()>> {
        reserve_inner(&self.chan, n).await?;
        Ok(PermitIterator::new(&self.chan, n))
    }

    /// Waits for capacity for one message, moving the sender into the
    /// returned [`OwnedPermit`].
    ///
    /// Unlike [`reserve`](Self::reserve), the permit borrows nothing, so it can
    /// be stored or moved into another task. Cloning a sender first keeps a
    /// handle to send with meanwhile. [`OwnedPermit::send`] and [`OwnedPermit::release`] hand
    /// the sender back.
    ///
    /// # Errors
    ///
    /// A [`SendError`] if the receiver is closed or dropped, now or while
    /// waiting.
    ///
    /// # Cancel safety
    ///
    /// Dropping the future gives up its place in the queue of waiters, passes
    /// on a slot it was handed meanwhile, and drops the sender it holds.
    ///
    /// ```
    /// # use std::{future::Future, pin::pin, task::{Context, Poll, Waker}};
    /// # fn now<F: Future>(f: F) -> F::Output {
    /// #     match pin!(f).poll(&mut Context::from_waker(Waker::noop())) {
    /// #         Poll::Ready(out) => out,
    /// #         Poll::Pending => panic!("not ready"),
    /// #     }
    /// # }
    /// use nuclei::sync::mpsc;
    ///
    /// let (tx, mut rx) = mpsc::channel(1);
    /// let permit = now(tx.reserve_owned()).unwrap();
    /// let tx = permit.send(123);
    /// assert_eq!(rx.try_recv(), Ok(123));
    /// tx.try_send(456).unwrap();
    /// ```
    pub async fn reserve_owned(self) -> Result<OwnedPermit<T>, SendError<()>> {
        reserve_inner(&self.chan, 1).await?;
        Ok(OwnedPermit::new(self))
    }

    /// Reserves capacity for one message if there is any, without waiting.
    /// It never takes a slot ahead of a send or reservation that is waiting.
    ///
    /// # Errors
    ///
    /// [`TrySendError::Full`] when no slot is free or a send or reservation
    /// is waiting, [`TrySendError::Closed`] when the receiver closed or was
    /// dropped.
    ///
    /// ```
    /// use nuclei::sync::mpsc;
    ///
    /// let (tx, mut rx) = mpsc::channel(1);
    /// let permit = tx.try_reserve().unwrap();
    /// assert!(tx.try_reserve().is_err(), "no slot is left");
    /// permit.send("foo");
    /// assert_eq!(rx.try_recv(), Ok("foo"));
    /// ```
    pub fn try_reserve(&self) -> Result<Permit<'_, T>, TrySendError<()>> {
        try_reserve_inner(&self.chan, 1)?;
        Ok(Permit::new(&self.chan))
    }

    /// Reserves capacity for `n` messages if there is enough, without
    /// waiting.
    ///
    /// # Errors
    ///
    /// [`TrySendError::Full`] when fewer than `n` slots are free, a send or
    /// reservation is waiting, or `n` is more than the channel can ever hold;
    /// [`TrySendError::Closed`] when the receiver closed or was dropped.
    /// `n == 0` succeeds with an empty iterator unless the channel is closed.
    ///
    /// ```
    /// use nuclei::sync::mpsc;
    ///
    /// let (tx, mut rx) = mpsc::channel(2);
    /// let mut permits = tx.try_reserve_many(2).unwrap();
    /// assert!(tx.try_send(123).is_err());
    /// permits.next().unwrap().send(456);
    /// permits.next().unwrap().send(457);
    /// assert_eq!(rx.try_recv(), Ok(456));
    /// assert_eq!(rx.try_recv(), Ok(457));
    ///
    /// assert!(tx.try_reserve_many(3).is_err(), "more than the channel holds");
    /// ```
    pub fn try_reserve_many(&self, n: usize) -> Result<PermitIterator<'_, T>, TrySendError<()>> {
        if n > self.chan.limit.bound {
            return Err(TrySendError::Full(()));
        }
        try_reserve_inner(&self.chan, n)?;
        Ok(PermitIterator::new(&self.chan, n))
    }

    /// Reserves capacity for one message if there is any, moving the sender
    /// into the returned [`OwnedPermit`].
    ///
    /// # Errors
    ///
    /// On failure the sender is handed back inside the error:
    /// [`TrySendError::Full`] when no slot is free or a send or reservation
    /// is waiting, [`TrySendError::Closed`] when the receiver closed or was
    /// dropped.
    ///
    /// ```
    /// use nuclei::sync::mpsc;
    ///
    /// let (tx, mut rx) = mpsc::channel(1);
    /// let permit = tx.clone().try_reserve_owned().unwrap();
    /// assert!(tx.try_send(123).is_err());
    /// permit.send(456);
    /// assert_eq!(rx.try_recv(), Ok(456));
    /// ```
    pub fn try_reserve_owned(self) -> Result<OwnedPermit<T>, TrySendError<Self>> {
        match try_reserve_inner(&self.chan, 1) {
            Ok(()) => Ok(OwnedPermit::new(self)),
            Err(refused) => Err(refused.with_value(self)),
        }
    }

    /// Whether two senders belong to the same channel.
    ///
    /// ```
    /// use nuclei::sync::mpsc;
    ///
    /// let (tx, _rx) = mpsc::channel::<()>(1);
    /// let tx2 = tx.clone();
    /// assert!(tx.same_channel(&tx2));
    ///
    /// let (tx3, _rx3) = mpsc::channel::<()>(1);
    /// assert!(!tx3.same_channel(&tx2));
    /// ```
    pub fn same_channel(&self, other: &Self) -> bool {
        Arc::ptr_eq(&self.chan, &other.chan)
    }

    /// The channel's current capacity: the slots that are free.
    ///
    /// It goes down when a value is sent or a slot is reserved, and up when the
    /// receiver takes a value or a permit is dropped unused. A slot set aside
    /// for a waiting [`reserve_many`](Self::reserve_many) is not free. Compare
    /// [`max_capacity`](Self::max_capacity), which never changes.
    ///
    /// ```
    /// use nuclei::sync::mpsc;
    ///
    /// let (tx, mut rx) = mpsc::channel::<()>(5);
    /// assert_eq!(tx.capacity(), 5);
    /// let permit = tx.try_reserve().unwrap();
    /// assert_eq!(tx.capacity(), 4);
    /// permit.send(());
    /// rx.try_recv().unwrap();
    /// assert_eq!(tx.capacity(), 5);
    /// ```
    pub fn capacity(&self) -> usize {
        self.chan.limit.sem.available()
    }

    /// Converts the sender into a [`WeakSender`], which does not keep the
    /// channel open. The sender itself stays.
    #[must_use = "Downgrade creates a WeakSender without destroying the original non-weak sender."]
    pub fn downgrade(&self) -> WeakSender<T> {
        self.chan.add_weak();
        WeakSender {
            chan: Arc::clone(&self.chan),
        }
    }

    /// The channel's maximum capacity: the `buffer` it was created with.
    ///
    /// ```
    /// use nuclei::sync::mpsc;
    ///
    /// let (tx, _rx) = mpsc::channel::<()>(5);
    /// let _permit = tx.try_reserve().unwrap();
    /// assert_eq!(tx.max_capacity(), 5);
    /// assert_eq!(tx.capacity(), 4);
    /// ```
    pub fn max_capacity(&self) -> usize {
        self.chan.limit.bound
    }

    /// How many [`Sender`] handles exist (an [`OwnedPermit`] holds one).
    pub fn strong_count(&self) -> usize {
        self.chan.strong_count()
    }

    /// How many [`WeakSender`] handles exist.
    pub fn weak_count(&self) -> usize {
        self.chan.weak_count()
    }
}

impl<T> Clone for Sender<T> {
    fn clone(&self) -> Self {
        self.chan.add_sender();
        Sender {
            chan: Arc::clone(&self.chan),
        }
    }
}

impl<T> Drop for Sender<T> {
    fn drop(&mut self) {
        self.chan.drop_sender();
    }
}

impl<T> fmt::Debug for Sender<T> {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        f.debug_struct("Sender")
            .field("closed", &self.is_closed())
            .field("capacity", &self.capacity())
            .field("max_capacity", &self.max_capacity())
            .finish()
    }
}

impl<T> WeakSender<T> {
    /// Tries to turn the weak sender into a [`Sender`].
    ///
    /// Returns `Some` while at least one [`Sender`] (or [`OwnedPermit`]) is
    /// alive, and `None` once the last is gone.
    pub fn upgrade(&self) -> Option<Sender<T>> {
        self.chan.try_add_sender().then(|| Sender {
            chan: Arc::clone(&self.chan),
        })
    }

    /// How many [`Sender`] handles exist.
    pub fn strong_count(&self) -> usize {
        self.chan.strong_count()
    }

    /// How many [`WeakSender`] handles exist.
    pub fn weak_count(&self) -> usize {
        self.chan.weak_count()
    }
}

impl<T> Clone for WeakSender<T> {
    fn clone(&self) -> Self {
        self.chan.add_weak();
        WeakSender {
            chan: Arc::clone(&self.chan),
        }
    }
}

impl<T> Drop for WeakSender<T> {
    fn drop(&mut self) {
        self.chan.drop_weak();
    }
}

impl<T> fmt::Debug for WeakSender<T> {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        f.debug_struct("WeakSender").finish()
    }
}

/// Takes `n` slots without waiting: the permits are the caller's to hand to
/// [`Permit`]s, which return them.
fn try_reserve_inner<T>(chan: &Chan<T, Bounded>, n: usize) -> Result<(), TrySendError<()>> {
    match chan.limit.sem.try_acquire(n) {
        Ok(()) => Ok(()),
        Err(seen) if observed_closed(seen) => Err(TrySendError::Closed(())),
        Err(_) => Err(TrySendError::Full(())),
    }
}

/// Waits for `n` slots, which the caller then owns as [`Permit`]s.
async fn reserve_inner<T>(chan: &Chan<T, Bounded>, n: usize) -> Result<(), SendError<()>> {
    if n > chan.limit.bound {
        return Err(SendError(()));
    }
    Reserve {
        chan,
        n,
        wait: FairWait::new(),
        taken: false,
    }
    .await
}

/// The future of [`Sender::send`]: the wait of [`reserve_inner`] for one
/// slot and the push of the value into it, as one future, so a send that
/// finds a slot free takes it and pushes with no state of nested waits.
///
/// It ends as a [`Reserve`] for one slot followed by a [`Permit`] send does:
/// a wait that fails, or that is dropped before it completes, lets go of its
/// place (and of a slot it was handed meanwhile) and wakes a receiver that
/// waits for a closed channel to go idle; the value is dropped after that.
struct Sending<'a, T> {
    chan: &'a Chan<T, Bounded>,
    wait: FairWait,
    /// The value to send: taken when the wait ends, so `None` once the
    /// future completed.
    value: Option<T>,
    /// Not `Unpin`, as the established API's send future is not.
    _pinned: PhantomPinned,
}

impl<T> Sending<'_, T> {
    /// What a [`Reserve`] that did not get its slot does as it is dropped.
    fn let_go(&mut self) {
        self.wait.cancel(&self.chan.limit.sem, 1);
        self.chan.wake_if_finished();
    }
}

impl<T> Future for Sending<'_, T> {
    type Output = Result<(), SendError<T>>;

    fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
        // SAFETY: nothing of the future is pinned in place: the wait's node
        // lives in its queue's pool, and the value moves into the channel or
        // back to the caller, never borrowed across polls.
        let this = unsafe { self.get_unchecked_mut() };
        if this.value.is_none() {
            polled_after_completion();
        }
        let chan = this.chan;
        let granted = core::task::ready!(this.wait.poll(&chan.limit.sem, 1, cx.waker(), |n| {
            chan.give_back(n);
        }));
        let Some(value) = this.value.take() else {
            polled_after_completion();
        };
        match granted {
            Ok(()) => {
                // The slot goes to the value, as a permit's send does.
                this.chan.send(value);
                Poll::Ready(Ok(()))
            }
            Err(_) => {
                this.let_go();
                Poll::Ready(Err(SendError(value)))
            }
        }
    }
}

impl<T> Drop for Sending<'_, T> {
    fn drop(&mut self) {
        if self.value.is_some() {
            self.let_go();
        }
    }
}

impl<T> internal::Withdraw for Sending<'_, T> {
    fn withdraw(self: Pin<&mut Self>) {
        // SAFETY: as in `poll`: nothing of the future is pinned in place.
        let this = unsafe { self.get_unchecked_mut() };
        // The value stays: only the wait for the slot is given up.
        if this.value.is_some() {
            this.let_go();
        }
    }
}

/// A `send` future polled again after it completed, which the `Future`
/// contract rules out.
#[cold]
#[expect(
    clippy::panic,
    reason = "polled after it completed, which the `Future` contract rules out"
)]
fn polled_after_completion() -> ! {
    panic!("`Sender::send` polled after it completed");
}

/// The wait of [`reserve_inner`].
///
/// A wait that ends without its slots becoming the caller's (it failed, or
/// it was dropped, perhaps just after a grant) can leave a closed channel
/// idle once what it held is back, with no receiver call to notice: so its
/// drop wakes a receiver that waits for a closed channel to go idle, after
/// the wait has left the queue.
struct Reserve<'a, T> {
    chan: &'a Chan<T, Bounded>,
    n: usize,
    wait: FairWait,
    /// The slots are the caller's: they leave with its permits.
    taken: bool,
}

impl<T> Future for Reserve<'_, T> {
    type Output = Result<(), SendError<()>>;

    fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
        let this = self.get_mut();
        let chan = this.chan;
        this.wait
            .poll(&chan.limit.sem, this.n, cx.waker(), |n| chan.give_back(n))
            .map(|taken| {
                this.taken = taken.is_ok();
                taken.map_err(|_| SendError(()))
            })
    }
}

impl<T> Drop for Reserve<'_, T> {
    fn drop(&mut self) {
        if !self.taken {
            self.wait.cancel(&self.chan.limit.sem, self.n);
            self.chan.wake_if_finished();
        }
    }
}

#[cfg(all(test, not(loom)))]
mod tests {
    use super::*;
    use crate::sync::mpsc::error::TryRecvError;
    use crate::sync::test_support::{Polled, hooked, with_budget_spent};

    #[test]
    fn the_free_slots_plus_the_queued_values_are_the_buffer() {
        let (tx, mut rx) = channel::<u32>(4);
        let permit = tx.try_reserve().unwrap();
        tx.try_send(1).unwrap();
        assert_eq!(tx.capacity() + rx.len() + 1, 4);
        permit.send(2);
        assert_eq!(tx.capacity() + rx.len(), 4);
        rx.try_recv().unwrap();
        assert_eq!(tx.capacity() + rx.len(), 4);
    }

    #[test]
    fn a_send_granted_as_it_queues_spends_the_budget() {
        let (tx, rx) = channel::<u32>(1);
        tx.try_send(0).unwrap();
        let rx = std::rc::Rc::new(core::cell::RefCell::new(rx));
        let frees = std::rc::Rc::clone(&rx);
        // The send finds no slot; between that look and its queueing, the
        // receiver takes the value, so the queueing's own pass grants it.
        let waker = hooked(
            Some(Box::new(move || {
                assert_eq!(frees.borrow_mut().try_recv(), Ok(0));
            })),
            None,
        );
        let mut send = Box::pin(tx.send(1));
        let mut cx = Context::from_waker(&waker);
        let polled = with_budget_spent(|| send.as_mut().poll(&mut cx));
        assert!(
            polled.is_pending(),
            "a send completed past the spent budget"
        );
        assert!(matches!(send.as_mut().poll(&mut cx), Poll::Ready(Ok(()))));
        assert_eq!(rx.borrow_mut().try_recv(), Ok(1));
        assert_eq!(tx.capacity(), 1);
    }

    #[test]
    fn a_send_refused_for_the_budget_gives_its_slot_back_before_it_wakes() {
        let (tx, rx) = channel::<u32>(1);
        let rx = std::rc::Rc::new(core::cell::RefCell::new(rx));
        let receiver = Polled::new(core::future::pending::<()>());
        let rx_waker = receiver.waker();
        let closes = std::rc::Rc::clone(&rx);
        // The send takes the free slot and is refused for the budget; at the
        // refusal's wake the slot is back already, so a receiver that closes
        // there finds the channel idle and ends.
        let waker = hooked(
            None,
            Some(Box::new(move || {
                let mut rx = closes.borrow_mut();
                rx.close();
                // The receiver's own poll, with a budget of its own.
                let looked =
                    crate::coop::outside(|| rx.poll_recv(&mut Context::from_waker(&rx_waker)));
                assert!(matches!(looked, Poll::Ready(None)), "a slot is still out");
            })),
        );
        let mut send = Box::pin(tx.send(1));
        let mut cx = Context::from_waker(&waker);
        let polled = with_budget_spent(|| send.as_mut().poll(&mut cx));
        assert!(polled.is_pending(), "the send was refused for the budget");
        assert_eq!(tx.capacity(), 1, "the refused send gave its slot back");
        assert!(matches!(send.as_mut().poll(&mut cx), Poll::Ready(Err(_))));
    }

    #[test]
    fn a_zero_buffer_is_refused() {
        let refused = std::panic::catch_unwind(|| channel::<u8>(0));
        assert!(refused.is_err());
    }

    #[test]
    fn a_reserve_granted_then_dropped_after_the_close_wakes_the_receiver() {
        let (tx, mut rx) = channel::<()>(1);
        let held = tx.try_reserve().unwrap();
        let mut reserve = Polled::new(tx.reserve());
        reserve.pending();
        drop(held);
        assert_eq!(reserve.wakes(), 1, "the slot was granted to the waiter");
        assert_eq!(tx.capacity(), 0);
        rx.close();
        let mut recv = Polled::new(rx.recv());
        recv.pending();
        drop(reserve);
        assert_eq!(tx.capacity(), 1, "the granted slot came back");
        assert_eq!(recv.wakes(), 1, "and the closed receiver was told");
        assert_eq!(recv.ready(), None);
    }

    #[test]
    fn a_failed_reserve_wakes_a_receiver_waiting_for_idle() {
        let (tx, mut rx) = channel::<()>(1);
        let permit = tx.try_reserve().unwrap();
        let mut reserve = Polled::new(tx.reserve());
        reserve.pending();
        rx.close();
        // The reservation was granted the error by the close.
        assert!(reserve.wakes() >= 1);
        assert!(matches!(reserve.ready(), Err(SendError(()))));
        let mut recv = Polled::new(rx.recv());
        recv.pending();
        drop(permit);
        assert_eq!(recv.wakes(), 1, "the last permit back ends the wait");
        assert_eq!(recv.ready(), None);
    }

    #[test]
    fn dropping_the_receiver_fails_a_waiting_reserve_with_the_value_kept() {
        let (tx, rx) = channel::<String>(1);
        tx.try_send("held".to_string()).unwrap();
        let mut send = Polled::new(tx.send("kept".to_string()));
        send.pending();
        drop(rx);
        let SendError(value) = send.ready().unwrap_err();
        assert_eq!(value, "kept");
    }

    /// A sender queued behind a full buffer whose waker panics when the
    /// receive frees its slot.
    fn panicking_sender() -> core::task::Waker {
        hooked(None, Some(Box::new(|| panic!("a sender's waker panicked"))))
    }

    fn quiet() -> core::task::Waker {
        hooked(None, None)
    }

    #[test]
    fn a_send_refused_for_the_budget_whose_wake_panics_gives_its_slot_back() {
        let (tx, _rx) = channel::<u32>(1);
        let waker = panicking_sender();
        let mut send = Box::pin(tx.send(1));
        let mut cx = Context::from_waker(&waker);
        let caught = with_budget_spent(|| {
            std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| send.as_mut().poll(&mut cx)))
        });
        assert!(caught.is_err(), "the panic reaches the caller");
        assert_eq!(tx.capacity(), 1, "the slot was lost to the panic");
    }

    #[test]
    fn a_receive_whose_release_panics_keeps_its_value_for_the_next_one() {
        let (tx, mut rx) = channel::<u32>(1);
        tx.try_send(1).unwrap();
        let waker = panicking_sender();
        let mut send = Box::pin(tx.send(2));
        assert!(
            send.as_mut()
                .poll(&mut Context::from_waker(&waker))
                .is_pending()
        );
        let caught = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| rx.try_recv()));
        assert!(caught.is_err(), "the panic reaches the receiver");
        assert_eq!(rx.len(), 1, "the stashed value counts");
        assert_eq!(rx.try_recv(), Ok(1), "the value was lost to the panic");
        assert_eq!(rx.try_recv(), Err(TryRecvError::Empty));
        let quiet = quiet();
        assert!(matches!(
            send.as_mut().poll(&mut Context::from_waker(&quiet)),
            Poll::Ready(Ok(()))
        ));
        assert_eq!(rx.try_recv(), Ok(2));
        assert_eq!(tx.capacity(), 1, "the stash made up an admission");
        tx.try_send(3).unwrap();
        assert!(matches!(tx.try_send(4), Err(TrySendError::Full(4))));
    }

    #[test]
    fn a_polled_receive_whose_release_panics_keeps_its_value() {
        let (tx, mut rx) = channel::<u32>(1);
        tx.try_send(1).unwrap();
        let waker = panicking_sender();
        let mut send = Box::pin(tx.send(2));
        assert!(
            send.as_mut()
                .poll(&mut Context::from_waker(&waker))
                .is_pending()
        );
        let mut recv = Polled::new(rx.recv());
        let caught = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| recv.poll()));
        assert!(caught.is_err(), "the panic reaches the poll");
        drop(recv);
        assert_eq!(Polled::new(rx.recv()).ready(), Some(1));
        let quiet = quiet();
        assert!(matches!(
            send.as_mut().poll(&mut Context::from_waker(&quiet)),
            Poll::Ready(Ok(()))
        ));
        assert_eq!(Polled::new(rx.recv()).ready(), Some(2));
    }

    #[test]
    fn a_batch_receive_takes_a_stashed_value_without_a_release() {
        let (tx, mut rx) = channel::<u32>(2);
        tx.try_send(1).unwrap();
        tx.try_send(2).unwrap();
        let waker = panicking_sender();
        let mut send = Box::pin(tx.send(3));
        assert!(
            send.as_mut()
                .poll(&mut Context::from_waker(&waker))
                .is_pending()
        );
        let caught = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| rx.try_recv()));
        assert!(caught.is_err());
        let mut buf = Vec::new();
        let mut recv = Polled::new(rx.recv_many(&mut buf, 8));
        assert_eq!(recv.ready(), 2);
        drop(recv);
        assert_eq!(buf, [1, 2]);
        drop(send);
        assert_eq!(tx.capacity(), 2, "every admission is back, none made up");
    }

    #[test]
    fn a_channel_dropped_with_a_stashed_value_drops_it_once() {
        use std::sync::Arc;
        use std::sync::atomic::{AtomicUsize, Ordering};
        struct Counted(Arc<AtomicUsize>);
        impl Drop for Counted {
            fn drop(&mut self) {
                self.0.fetch_add(1, Ordering::SeqCst);
            }
        }
        let drops = Arc::new(AtomicUsize::new(0));
        let (tx, mut rx) = channel::<Counted>(1);
        assert!(tx.try_send(Counted(Arc::clone(&drops))).is_ok());
        let waker = panicking_sender();
        let mut send = Box::pin(tx.send(Counted(Arc::clone(&drops))));
        assert!(
            send.as_mut()
                .poll(&mut Context::from_waker(&waker))
                .is_pending()
        );
        let caught = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| rx.try_recv()));
        assert!(caught.is_err());
        assert_eq!(drops.load(Ordering::SeqCst), 0, "the value is stashed");
        drop(send);
        drop(rx);
        drop(tx);
        assert_eq!(drops.load(Ordering::SeqCst), 2, "each dropped exactly once");
    }
}