nexus-channel 1.0.0

//! Multi-producer single-consumer bounded channel.
//!
//! A blocking channel wrapping [`nexus_queue::mpsc`] with an optimized parking
//! strategy that minimizes syscall overhead.
//!
//! # Example
//!
//! ```
//! use nexus_channel::mpsc;
//! use std::thread;
//!
//! let (tx, mut rx) = mpsc::channel::<u64>(1024);
//!
//! let tx2 = tx.clone();
//! thread::spawn(move || {
//!     tx2.send(1).unwrap();
//! });
//!
//! thread::spawn(move || {
//!     tx.send(2).unwrap();
//! });
//!
//! let mut received = vec![rx.recv().unwrap(), rx.recv().unwrap()];
//! received.sort();
//! assert_eq!(received, vec![1, 2]);
//! ```

use core::fmt;
use std::mem::ManuallyDrop;
use std::sync::Arc;
use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering};
use std::time::{Duration, Instant};

use crossbeam_utils::sync::{Parker, Unparker};
use crossbeam_utils::{Backoff, CachePadded};
use nexus_queue::Full;
use nexus_queue::mpsc::{Consumer, Producer, bounded};

use crate::{RecvError, RecvTimeoutError, SendError, TryRecvError, TrySendError};

/// Default number of backoff snooze iterations before parking.
const DEFAULT_SNOOZE_ITERS: usize = 8;

/// Shared state between senders and receiver.
#[allow(clippy::struct_field_names)]
struct Shared {
    /// Flag indicating the receiver is parked.
    receiver_parked: CachePadded<AtomicBool>,
    /// Flag indicating the receiver has been dropped.
    receiver_disconnected: AtomicBool,
    /// Number of active senders. When this reaches 0, receiver should wake.
    sender_count: AtomicUsize,
    /// Unparker to wake the receiver.
    receiver_unparker: Unparker,
}

/// Creates a bounded MPSC channel with the given capacity.
///
/// Returns a `(Sender, Receiver)` pair. The sender can be cloned to create
/// additional producers. The actual capacity will be rounded up to the next
/// power of two.
///
/// # Panics
///
/// Panics if `capacity` is 0.
///
/// # Example
///
/// ```
/// use nexus_channel::mpsc;
///
/// let (tx, mut rx) = mpsc::channel::<String>(100);
///
/// tx.send("hello".to_string()).unwrap();
/// assert_eq!(rx.recv().unwrap(), "hello");
/// ```
pub fn channel<T>(capacity: usize) -> (Sender<T>, Receiver<T>) {
    channel_with_config(capacity, DEFAULT_SNOOZE_ITERS)
}

/// Creates a bounded MPSC channel with custom backoff configuration.
///
/// # Arguments
///
/// * `capacity` - Maximum number of messages the channel can hold (rounded to power of 2)
/// * `snooze_iters` - Number of backoff iterations before parking. Higher values
///   burn more CPU but reduce latency for bursty workloads.
///
/// # Panics
///
/// Panics if `capacity` is 0.
pub fn channel_with_config<T>(capacity: usize, snooze_iters: usize) -> (Sender<T>, Receiver<T>) {
    let (producer, consumer) = bounded(capacity);

    let receiver_parker = Parker::new();
    let receiver_unparker = receiver_parker.unparker().clone();

    let shared = Arc::new(Shared {
        receiver_parked: CachePadded::new(AtomicBool::new(false)),
        receiver_disconnected: AtomicBool::new(false),
        sender_count: AtomicUsize::new(1),
        receiver_unparker,
    });

    (
        Sender {
            producer: ManuallyDrop::new(producer),
            shared: Arc::clone(&shared),
            snooze_iters,
        },
        Receiver {
            consumer: ManuallyDrop::new(consumer),
            shared,
            parker: receiver_parker,
            snooze_iters,
        },
    )
}

/// The sending half of an MPSC channel.
///
/// This can be cloned to create multiple producers. Each clone is independent
/// and can be sent to different threads.
///
/// Messages can be sent with [`send`](Sender::send) (blocking) or
/// [`try_send`](Sender::try_send) (non-blocking).
pub struct Sender<T> {
    producer: ManuallyDrop<Producer<T>>,
    shared: Arc<Shared>,
    snooze_iters: usize,
}

impl<T> Clone for Sender<T> {
    fn clone(&self) -> Self {
        // Increment sender count before creating the clone
        self.shared.sender_count.fetch_add(1, Ordering::AcqRel);
        Sender {
            producer: ManuallyDrop::new((*self.producer).clone()),
            shared: Arc::clone(&self.shared),
            snooze_iters: self.snooze_iters,
        }
    }
}

impl<T> Sender<T> {
    /// Sends a message into the channel, blocking if necessary.
    ///
    /// If the channel is full, this method will block until space is available
    /// or the receiver disconnects.
    ///
    /// Returns `Err(SendError(value))` if the receiver has been dropped.
    pub fn send(&self, value: T) -> Result<(), SendError<T>> {
        if self.is_disconnected() {
            return Err(SendError(value));
        }

        // Clone producer for mutable access (each send needs its own mutable ref)
        let mut producer = (*self.producer).clone();

        let mut val = value;

        // Fast path
        match producer.push(val) {
            Ok(()) => {
                self.notify_receiver();
                return Ok(());
            }
            Err(Full(v)) => val = v,
        }

        // Backoff phase
        let backoff = Backoff::new();
        for _ in 0..self.snooze_iters {
            backoff.snooze();

            if self.is_disconnected() {
                return Err(SendError(val));
            }

            match producer.push(val) {
                Ok(()) => {
                    self.notify_receiver();
                    return Ok(());
                }
                Err(Full(v)) => val = v,
            }
        }

        // Spin phase (no parking for MPSC senders - they compete via CAS)
        // We use a lighter spin loop here since multiple producers can't all park
        loop {
            if self.is_disconnected() {
                return Err(SendError(val));
            }

            match producer.push(val) {
                Ok(()) => {
                    self.notify_receiver();
                    return Ok(());
                }
                Err(Full(v)) => val = v,
            }

            // Yield to avoid burning CPU when truly full
            std::thread::yield_now();
        }
    }

    /// Attempts to send a message without blocking.
    ///
    /// Returns immediately with:
    /// - `Ok(())` if the message was sent
    /// - `Err(TrySendError::Full(value))` if the channel is full
    /// - `Err(TrySendError::Disconnected(value))` if the receiver was dropped
    pub fn try_send(&self, value: T) -> Result<(), TrySendError<T>> {
        if self.is_disconnected() {
            return Err(TrySendError::Disconnected(value));
        }

        let mut producer = (*self.producer).clone();

        match producer.push(value) {
            Ok(()) => {
                self.notify_receiver();
                Ok(())
            }
            Err(Full(v)) => Err(TrySendError::Full(v)),
        }
    }

    #[inline]
    fn notify_receiver(&self) {
        if self.shared.receiver_parked.load(Ordering::SeqCst) {
            self.shared.receiver_unparker.unpark();
        }
    }

    /// Returns `true` if the receiver has been dropped.
    #[inline]
    pub fn is_disconnected(&self) -> bool {
        self.shared.receiver_disconnected.load(Ordering::Acquire)
    }

    /// Returns the capacity of the channel.
    #[inline]
    pub fn capacity(&self) -> usize {
        (*self.producer).capacity()
    }
}

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

impl<T> Drop for Sender<T> {
    fn drop(&mut self) {
        // Drop the producer first - this allows consumer.is_disconnected() to return true
        // SAFETY: We're in drop, producer won't be used again
        unsafe { ManuallyDrop::drop(&mut self.producer) };

        // Decrement sender count. If we were the last sender, wake the receiver.
        // fetch_sub returns the previous value, so if it was 1, we just dropped the last sender.
        if self.shared.sender_count.fetch_sub(1, Ordering::AcqRel) == 1 {
            self.shared.receiver_unparker.unpark();
        }
    }
}

/// The receiving half of an MPSC channel.
///
/// This cannot be cloned - only one receiver is allowed. Messages can be
/// received with [`recv`](Receiver::recv) (blocking),
/// [`recv_timeout`](Receiver::recv_timeout) (blocking with timeout), or
/// [`try_recv`](Receiver::try_recv) (non-blocking).
pub struct Receiver<T> {
    consumer: ManuallyDrop<Consumer<T>>,
    shared: Arc<Shared>,
    parker: Parker,
    snooze_iters: usize,
}

impl<T> Receiver<T> {
    /// Receives a message from the channel, blocking if necessary.
    ///
    /// If the channel is empty, this method will block until a message arrives
    /// or all senders disconnect.
    ///
    /// Returns `Err(RecvError)` if all senders have been dropped and no messages
    /// remain in the channel.
    pub fn recv(&mut self) -> Result<T, RecvError> {
        // Fast path
        if let Some(v) = self.consumer.pop() {
            return Ok(v);
        }

        // Backoff phase
        let backoff = Backoff::new();
        for _ in 0..self.snooze_iters {
            backoff.snooze();

            if let Some(v) = self.consumer.pop() {
                return Ok(v);
            }

            if self.consumer.is_disconnected() {
                return self.consumer.pop().ok_or(RecvError);
            }
        }

        // Park phase
        loop {
            self.shared.receiver_parked.store(true, Ordering::SeqCst);

            if let Some(v) = self.consumer.pop() {
                self.shared.receiver_parked.store(false, Ordering::Relaxed);
                return Ok(v);
            }

            if self.consumer.is_disconnected() {
                self.shared.receiver_parked.store(false, Ordering::Relaxed);
                return Err(RecvError);
            }

            self.parker.park();
            self.shared.receiver_parked.store(false, Ordering::Relaxed);

            if let Some(v) = self.consumer.pop() {
                return Ok(v);
            }

            if self.consumer.is_disconnected() {
                return Err(RecvError);
            }
        }
    }

    /// Receives a message from the channel, blocking for at most `timeout`.
    ///
    /// Returns:
    /// - `Ok(value)` if a message was received
    /// - `Err(RecvTimeoutError::Timeout)` if the timeout elapsed
    /// - `Err(RecvTimeoutError::Disconnected)` if all senders were dropped
    pub fn recv_timeout(&mut self, timeout: Duration) -> Result<T, RecvTimeoutError> {
        let deadline = Instant::now() + timeout;

        // Fast path
        if let Some(v) = self.consumer.pop() {
            return Ok(v);
        }

        // Backoff phase
        let backoff = Backoff::new();
        for _ in 0..self.snooze_iters {
            if Instant::now() >= deadline {
                return Err(RecvTimeoutError::Timeout);
            }

            backoff.snooze();

            if let Some(v) = self.consumer.pop() {
                return Ok(v);
            }

            if self.consumer.is_disconnected() {
                return self.consumer.pop().ok_or(RecvTimeoutError::Disconnected);
            }
        }

        // Park phase with timeout
        loop {
            let now = Instant::now();
            if now >= deadline {
                return Err(RecvTimeoutError::Timeout);
            }

            self.shared.receiver_parked.store(true, Ordering::SeqCst);

            if let Some(v) = self.consumer.pop() {
                self.shared.receiver_parked.store(false, Ordering::Relaxed);
                return Ok(v);
            }

            if self.consumer.is_disconnected() {
                self.shared.receiver_parked.store(false, Ordering::Relaxed);
                return Err(RecvTimeoutError::Disconnected);
            }

            let remaining = deadline - now;
            self.parker.park_timeout(remaining);
            self.shared.receiver_parked.store(false, Ordering::Relaxed);

            if let Some(v) = self.consumer.pop() {
                return Ok(v);
            }

            if self.consumer.is_disconnected() {
                return Err(RecvTimeoutError::Disconnected);
            }
        }
    }

    /// Attempts to receive a message without blocking.
    ///
    /// Returns immediately with:
    /// - `Ok(value)` if a message was available
    /// - `Err(TryRecvError::Empty)` if the channel is empty
    /// - `Err(TryRecvError::Disconnected)` if all senders were dropped and channel is empty
    pub fn try_recv(&mut self) -> Result<T, TryRecvError> {
        match self.consumer.pop() {
            Some(v) => Ok(v),
            None => {
                if self.consumer.is_disconnected() {
                    Err(TryRecvError::Disconnected)
                } else {
                    Err(TryRecvError::Empty)
                }
            }
        }
    }

    /// Returns `true` if all senders have been dropped.
    #[inline]
    pub fn is_disconnected(&self) -> bool {
        self.consumer.is_disconnected()
    }

    /// Returns the capacity of the channel.
    #[inline]
    pub fn capacity(&self) -> usize {
        self.consumer.capacity()
    }
}

impl<T> fmt::Debug for Receiver<T> {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        f.debug_struct("Receiver")
            .field("capacity", &self.capacity())
            .field("disconnected", &self.is_disconnected())
            .finish_non_exhaustive()
    }
}

impl<T> Drop for Receiver<T> {
    fn drop(&mut self) {
        // Signal disconnection before dropping consumer
        self.shared
            .receiver_disconnected
            .store(true, Ordering::Release);
        unsafe { ManuallyDrop::drop(&mut self.consumer) };
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use std::sync::atomic::{AtomicBool, AtomicUsize};
    use std::thread;

    // ============================================================================
    // Basic Operations
    // ============================================================================

    #[test]
    fn basic_send_recv() {
        let (tx, mut rx) = channel::<u64>(4);

        tx.send(1).unwrap();
        tx.send(2).unwrap();
        tx.send(3).unwrap();

        assert_eq!(rx.recv().unwrap(), 1);
        assert_eq!(rx.recv().unwrap(), 2);
        assert_eq!(rx.recv().unwrap(), 3);
    }

    #[test]
    fn try_send_try_recv() {
        let (tx, mut rx) = channel::<u64>(2);

        assert!(tx.try_send(1).is_ok());
        assert!(tx.try_send(2).is_ok());
        assert!(matches!(tx.try_send(3), Err(TrySendError::Full(3))));

        assert_eq!(rx.try_recv().unwrap(), 1);
        assert_eq!(rx.try_recv().unwrap(), 2);
        assert!(matches!(rx.try_recv(), Err(TryRecvError::Empty)));
    }

    #[test]
    fn send_fills_then_recv_drains() {
        let (tx, mut rx) = channel::<u64>(4);

        for i in 0..4 {
            tx.try_send(i).unwrap();
        }
        assert!(matches!(tx.try_send(99), Err(TrySendError::Full(99))));

        for i in 0..4 {
            assert_eq!(rx.recv().unwrap(), i);
        }
        assert!(matches!(rx.try_recv(), Err(TryRecvError::Empty)));
    }

    // ============================================================================
    // Timeout Operations
    // ============================================================================

    #[test]
    fn recv_timeout_success() {
        let (tx, mut rx) = channel::<u64>(4);

        tx.send(42).unwrap();

        let result = rx.recv_timeout(Duration::from_millis(100));
        assert_eq!(result.unwrap(), 42);
    }

    #[test]
    fn recv_timeout_expires() {
        let (_tx, mut rx) = channel::<u64>(4);

        let start = Instant::now();
        let result = rx.recv_timeout(Duration::from_millis(50));

        assert!(matches!(result, Err(RecvTimeoutError::Timeout)));
        assert!(start.elapsed() >= Duration::from_millis(50));
    }

    #[test]
    fn recv_timeout_disconnected() {
        let (tx, mut rx) = channel::<u64>(4);

        drop(tx);

        let result = rx.recv_timeout(Duration::from_millis(100));
        assert!(matches!(result, Err(RecvTimeoutError::Disconnected)));
    }

    #[test]
    fn recv_timeout_data_arrives() {
        let (tx, mut rx) = channel::<u64>(4);

        let handle = thread::spawn(move || {
            thread::sleep(Duration::from_millis(25));
            tx.send(42).unwrap();
        });

        let result = rx.recv_timeout(Duration::from_millis(100));
        assert_eq!(result.unwrap(), 42);

        handle.join().unwrap();
    }

    #[test]
    fn recv_timeout_disconnect_while_waiting() {
        let (tx, mut rx) = channel::<u64>(4);

        let handle = thread::spawn(move || {
            thread::sleep(Duration::from_millis(25));
            drop(tx);
        });

        let result = rx.recv_timeout(Duration::from_millis(100));
        assert!(matches!(result, Err(RecvTimeoutError::Disconnected)));

        handle.join().unwrap();
    }

    #[test]
    fn recv_timeout_all_senders_disconnect() {
        let (tx1, mut rx) = channel::<u64>(4);
        let tx2 = tx1.clone();
        let tx3 = tx1.clone();

        let h1 = thread::spawn(move || {
            thread::sleep(Duration::from_millis(10));
            drop(tx1);
        });

        let h2 = thread::spawn(move || {
            thread::sleep(Duration::from_millis(20));
            drop(tx2);
        });

        let h3 = thread::spawn(move || {
            thread::sleep(Duration::from_millis(30));
            drop(tx3);
        });

        let result = rx.recv_timeout(Duration::from_millis(100));
        assert!(matches!(result, Err(RecvTimeoutError::Disconnected)));

        h1.join().unwrap();
        h2.join().unwrap();
        h3.join().unwrap();
    }

    // ============================================================================
    // Disconnection - Single Sender
    // ============================================================================

    #[test]
    fn recv_returns_error_when_sender_dropped() {
        let (tx, mut rx) = channel::<u64>(4);

        drop(tx);

        assert!(rx.recv().is_err());
        assert!(matches!(rx.try_recv(), Err(TryRecvError::Disconnected)));
    }

    #[test]
    fn recv_drains_before_error_when_sender_dropped() {
        let (tx, mut rx) = channel::<u64>(4);

        tx.send(1).unwrap();
        tx.send(2).unwrap();
        drop(tx);

        assert_eq!(rx.recv().unwrap(), 1);
        assert_eq!(rx.recv().unwrap(), 2);
        assert!(rx.recv().is_err());
    }

    #[test]
    fn send_returns_error_when_receiver_dropped() {
        let (tx, rx) = channel::<u64>(4);

        drop(rx);

        assert!(tx.send(1).is_err());
        assert!(matches!(tx.try_send(1), Err(TrySendError::Disconnected(1))));
    }

    #[test]
    fn is_disconnected_sender() {
        let (tx, rx) = channel::<u64>(4);

        assert!(!tx.is_disconnected());
        drop(rx);
        assert!(tx.is_disconnected());
    }

    #[test]
    fn is_disconnected_receiver() {
        let (tx, rx) = channel::<u64>(4);

        assert!(!rx.is_disconnected());
        drop(tx);
        assert!(rx.is_disconnected());
    }

    // ============================================================================
    // Disconnection - Multiple Senders
    // ============================================================================

    #[test]
    fn disconnection_all_senders_dropped() {
        let (tx, mut rx) = channel::<u64>(4);
        let tx2 = tx.clone();

        assert!(!rx.is_disconnected());
        drop(tx);
        assert!(!rx.is_disconnected()); // tx2 still alive
        drop(tx2);
        assert!(rx.is_disconnected());

        assert!(rx.recv().is_err());
    }

    #[test]
    fn disconnection_receiver_dropped() {
        let (tx, rx) = channel::<u64>(4);
        drop(rx);

        assert!(tx.send(1).is_err());
        assert!(matches!(tx.try_send(1), Err(TrySendError::Disconnected(1))));
    }

    #[test]
    fn receiver_drop_detected_by_cloned_sender() {
        let (tx1, rx) = channel::<u64>(4);
        let tx2 = tx1.clone();
        let tx3 = tx1.clone();

        drop(rx);

        assert!(tx1.is_disconnected());
        assert!(tx2.is_disconnected());
        assert!(tx3.is_disconnected());

        assert!(tx1.send(1).is_err());
        assert!(tx2.send(2).is_err());
        assert!(tx3.send(3).is_err());
    }

    #[test]
    fn partial_sender_drop_keeps_channel_alive() {
        let (tx1, mut rx) = channel::<u64>(4);
        let tx2 = tx1.clone();
        let tx3 = tx1.clone();

        tx1.send(1).unwrap();
        drop(tx1);

        tx2.send(2).unwrap();
        drop(tx2);

        tx3.send(3).unwrap();
        drop(tx3);

        assert_eq!(rx.recv().unwrap(), 1);
        assert_eq!(rx.recv().unwrap(), 2);
        assert_eq!(rx.recv().unwrap(), 3);
        assert!(rx.recv().is_err());
    }

    #[test]
    fn many_senders_drop_sequentially() {
        let (tx, mut rx) = channel::<u64>(64);

        let senders: Vec<_> = (0..10).map(|_| tx.clone()).collect();
        drop(tx);

        for (i, sender) in senders.into_iter().enumerate() {
            assert!(!rx.is_disconnected());
            sender.send(i as u64).unwrap();
            drop(sender);
        }

        assert!(rx.is_disconnected());

        for i in 0..10 {
            assert_eq!(rx.recv().unwrap(), i);
        }
        assert!(rx.recv().is_err());
    }

    // ============================================================================
    // Multiple Producers
    // ============================================================================

    #[test]
    fn multiple_producers() {
        let (tx, mut rx) = channel::<u64>(64);

        let handles: Vec<_> = (0..4)
            .map(|i| {
                let tx = tx.clone();
                thread::spawn(move || {
                    for j in 0..100 {
                        tx.send(i * 100 + j).unwrap();
                    }
                })
            })
            .collect();

        drop(tx); // Drop original sender

        let mut received = Vec::new();
        while let Ok(v) = rx.recv() {
            received.push(v);
        }

        for h in handles {
            h.join().unwrap();
        }

        assert_eq!(received.len(), 400);
        received.sort();
        let expected: Vec<u64> = (0..4)
            .flat_map(|i| (0..100).map(move |j| i * 100 + j))
            .collect();
        let mut expected_sorted = expected;
        expected_sorted.sort();
        assert_eq!(received, expected_sorted);
    }

    #[test]
    fn sender_clone_is_independent() {
        let (tx1, mut rx) = channel::<u64>(4);
        let tx2 = tx1.clone();

        tx1.send(1).unwrap();
        tx2.send(2).unwrap();

        assert_eq!(rx.recv().unwrap(), 1);
        assert_eq!(rx.recv().unwrap(), 2);
    }

    #[test]
    fn concurrent_sends_from_clones() {
        let (tx, mut rx) = channel::<u64>(1024);

        let handles: Vec<_> = (0..8)
            .map(|id| {
                let tx = tx.clone();
                thread::spawn(move || {
                    for i in 0..1000 {
                        tx.send(id * 1000 + i).unwrap();
                    }
                })
            })
            .collect();

        drop(tx);

        let mut count = 0;
        while let Ok(_) = rx.recv() {
            count += 1;
        }

        for h in handles {
            h.join().unwrap();
        }

        assert_eq!(count, 8000);
    }

    // ============================================================================
    // Cross-Thread Basic
    // ============================================================================

    #[test]
    fn cross_thread_single_message() {
        let (tx, mut rx) = channel::<u64>(4);

        let handle = thread::spawn(move || rx.recv().unwrap());

        tx.send(42).unwrap();

        assert_eq!(handle.join().unwrap(), 42);
    }

    #[test]
    fn cross_thread_multiple_messages() {
        let (tx, mut rx) = channel::<u64>(4);

        let handle = thread::spawn(move || {
            let mut sum = 0;
            for _ in 0..100 {
                sum += rx.recv().unwrap();
            }
            sum
        });

        for i in 0..100 {
            tx.send(i).unwrap();
        }

        let sum = handle.join().unwrap();
        assert_eq!(sum, 99 * 100 / 2);
    }

    // ============================================================================
    // Blocking Behavior
    // ============================================================================

    #[test]
    fn recv_blocks_until_send() {
        let (tx, mut rx) = channel::<u64>(4);

        let start = Instant::now();

        let handle = thread::spawn(move || rx.recv().unwrap());

        thread::sleep(Duration::from_millis(50));
        tx.send(42).unwrap();

        let val = handle.join().unwrap();
        assert_eq!(val, 42);
        assert!(start.elapsed() >= Duration::from_millis(50));
    }

    #[test]
    fn send_spins_until_space_available() {
        let (tx, mut rx) = channel::<u64>(2);

        // Fill the buffer
        tx.try_send(1).unwrap();
        tx.try_send(2).unwrap();

        let start = Instant::now();

        let handle = thread::spawn(move || {
            tx.send(3).unwrap(); // Should spin waiting
        });

        thread::sleep(Duration::from_millis(50));
        rx.recv().unwrap(); // Free up space

        handle.join().unwrap();
        assert!(start.elapsed() >= Duration::from_millis(50));
    }

    // ============================================================================
    // Wake on Disconnect
    // ============================================================================

    #[test]
    fn recv_wakes_on_all_senders_drop() {
        let (tx1, mut rx) = channel::<u64>(4);
        let tx2 = tx1.clone();

        let handle = thread::spawn(move || {
            let result = rx.recv();
            assert!(result.is_err());
        });

        thread::sleep(Duration::from_millis(50));
        drop(tx1);
        drop(tx2);

        // Should complete, not hang
        handle.join().unwrap();
    }

    #[test]
    fn send_detects_receiver_drop_while_spinning() {
        let (tx, rx) = channel::<u64>(1);

        tx.try_send(1).unwrap(); // Fill it

        let handle = thread::spawn(move || {
            let result = tx.send(2); // Will spin, then detect disconnect
            assert!(result.is_err());
        });

        thread::sleep(Duration::from_millis(50));
        drop(rx);

        // Should complete, not hang
        handle.join().unwrap();
    }

    // ============================================================================
    // Capacity Edge Cases
    // ============================================================================

    #[test]
    fn capacity_one() {
        let (tx, mut rx) = channel::<u64>(1);

        for i in 0..100 {
            tx.send(i).unwrap();
            assert_eq!(rx.recv().unwrap(), i);
        }
    }

    #[test]
    fn capacity_one_cross_thread() {
        let (tx, mut rx) = channel::<u64>(1);

        let handle = thread::spawn(move || {
            for _ in 0..1000 {
                rx.recv().unwrap();
            }
        });

        for i in 0..1000 {
            tx.send(i).unwrap();
        }

        handle.join().unwrap();
    }

    #[test]
    fn capacity_one_multiple_producers() {
        let (tx, mut rx) = channel::<u64>(1);

        let handles: Vec<_> = (0..4)
            .map(|_| {
                let tx = tx.clone();
                thread::spawn(move || {
                    for i in 0..100 {
                        tx.send(i).unwrap();
                    }
                })
            })
            .collect();

        drop(tx);

        let mut count = 0;
        while let Ok(_) = rx.recv() {
            count += 1;
        }

        for h in handles {
            h.join().unwrap();
        }

        assert_eq!(count, 400);
    }

    // ============================================================================
    // Drop Behavior
    // ============================================================================

    #[test]
    fn values_dropped_on_channel_drop() {
        static DROP_COUNT: AtomicUsize = AtomicUsize::new(0);

        #[derive(Debug)]
        struct DropCounter;
        impl Drop for DropCounter {
            fn drop(&mut self) {
                DROP_COUNT.fetch_add(1, Ordering::SeqCst);
            }
        }

        DROP_COUNT.store(0, Ordering::SeqCst);

        let (tx, rx) = channel::<DropCounter>(4);

        tx.try_send(DropCounter).unwrap();
        tx.try_send(DropCounter).unwrap();
        tx.try_send(DropCounter).unwrap();

        assert_eq!(DROP_COUNT.load(Ordering::SeqCst), 0);

        drop(tx);
        drop(rx);

        assert_eq!(DROP_COUNT.load(Ordering::SeqCst), 3);
    }

    #[test]
    fn failed_send_returns_value() {
        let (tx, rx) = channel::<String>(1);

        tx.try_send("hello".to_string()).unwrap();

        let err = tx.try_send("world".to_string());
        match err {
            Err(TrySendError::Full(s)) => assert_eq!(s, "world"),
            _ => panic!("expected Full error"),
        }

        drop(rx);

        let err = tx.try_send("test".to_string());
        match err {
            Err(TrySendError::Disconnected(s)) => assert_eq!(s, "test"),
            _ => panic!("expected Disconnected error"),
        }
    }

    #[test]
    fn drop_sender_while_value_in_flight() {
        static DROP_COUNT: AtomicUsize = AtomicUsize::new(0);

        #[derive(Debug)]
        struct DropCounter(u64);
        impl Drop for DropCounter {
            fn drop(&mut self) {
                DROP_COUNT.fetch_add(1, Ordering::SeqCst);
            }
        }

        DROP_COUNT.store(0, Ordering::SeqCst);

        let (tx, mut rx) = channel::<DropCounter>(4);

        tx.send(DropCounter(1)).unwrap();
        tx.send(DropCounter(2)).unwrap();
        drop(tx);

        assert_eq!(DROP_COUNT.load(Ordering::SeqCst), 0);

        let v = rx.recv().unwrap();
        assert_eq!(v.0, 1);
        drop(v);
        assert_eq!(DROP_COUNT.load(Ordering::SeqCst), 1);

        let v = rx.recv().unwrap();
        assert_eq!(v.0, 2);
        drop(v);
        assert_eq!(DROP_COUNT.load(Ordering::SeqCst), 2);
    }

    #[test]
    fn drop_receiver_with_pending_values() {
        static DROP_COUNT: AtomicUsize = AtomicUsize::new(0);

        #[derive(Debug)]
        struct DropCounter;
        impl Drop for DropCounter {
            fn drop(&mut self) {
                DROP_COUNT.fetch_add(1, Ordering::SeqCst);
            }
        }

        DROP_COUNT.store(0, Ordering::SeqCst);

        let (tx, rx) = channel::<DropCounter>(4);

        tx.send(DropCounter).unwrap();
        tx.send(DropCounter).unwrap();

        // Values not dropped yet - still in shared buffer held by tx and rx
        assert_eq!(DROP_COUNT.load(Ordering::SeqCst), 0);

        // Drop receiver without consuming
        drop(rx);

        // Values still not dropped - shared buffer held by tx
        assert_eq!(DROP_COUNT.load(Ordering::SeqCst), 0);

        // Sender should detect disconnect
        assert!(tx.is_disconnected());

        // Drop sender - now shared buffer can be freed
        drop(tx);

        // Values should now be dropped
        assert_eq!(DROP_COUNT.load(Ordering::SeqCst), 2);
    }

    // ============================================================================
    // Special Types
    // ============================================================================

    #[test]
    fn zero_sized_type() {
        let (tx, mut rx) = channel::<()>(4);

        tx.send(()).unwrap();
        tx.send(()).unwrap();

        assert_eq!(rx.recv().unwrap(), ());
        assert_eq!(rx.recv().unwrap(), ());
    }

    #[test]
    fn large_message_type() {
        #[derive(Clone, PartialEq, Debug)]
        struct LargeMessage {
            data: [u8; 4096],
        }

        let (tx, mut rx) = channel::<LargeMessage>(4);

        let msg = LargeMessage { data: [42u8; 4096] };
        tx.send(msg.clone()).unwrap();

        let received = rx.recv().unwrap();
        assert_eq!(received.data[0], 42);
        assert_eq!(received.data[4095], 42);
    }

    // ============================================================================
    // Multiple Laps
    // ============================================================================

    #[test]
    fn many_laps_single_thread() {
        let (tx, mut rx) = channel::<u64>(4);

        // 1000 messages through 4-slot buffer = 250 laps
        for i in 0..1000 {
            tx.send(i).unwrap();
            assert_eq!(rx.recv().unwrap(), i);
        }
    }

    #[test]
    fn many_laps_cross_thread() {
        const COUNT: u64 = 100_000;

        let (tx, mut rx) = channel::<u64>(4); // Small buffer, many laps

        let producer = thread::spawn(move || {
            for i in 0..COUNT {
                tx.send(i).unwrap();
            }
        });

        let consumer = thread::spawn(move || {
            let mut expected = 0u64;
            while expected < COUNT {
                let val = rx.recv().unwrap();
                assert_eq!(val, expected);
                expected += 1;
            }
        });

        producer.join().unwrap();
        consumer.join().unwrap();
    }

    // ============================================================================
    // Stress Tests - Single Producer
    // ============================================================================

    #[test]
    fn stress_high_volume() {
        const COUNT: u64 = 100_000;

        let (tx, mut rx) = channel::<u64>(1024);

        let producer = thread::spawn(move || {
            for i in 0..COUNT {
                tx.send(i).unwrap();
            }
        });

        let consumer = thread::spawn(move || {
            let mut sum = 0u64;
            for _ in 0..COUNT {
                sum = sum.wrapping_add(rx.recv().unwrap());
            }
            sum
        });

        producer.join().unwrap();
        let sum = consumer.join().unwrap();
        assert_eq!(sum, COUNT * (COUNT - 1) / 2);
    }

    #[test]
    fn stress_small_buffer() {
        const COUNT: u64 = 10_000;

        let (tx, mut rx) = channel::<u64>(4);

        let producer = thread::spawn(move || {
            for i in 0..COUNT {
                tx.send(i).unwrap();
            }
        });

        let consumer = thread::spawn(move || {
            let mut received = 0u64;
            while received < COUNT {
                rx.recv().unwrap();
                received += 1;
            }
            received
        });

        producer.join().unwrap();
        let received = consumer.join().unwrap();
        assert_eq!(received, COUNT);
    }

    #[test]
    fn stress_capacity_one_high_volume() {
        const COUNT: u64 = 10_000;

        let (tx, mut rx) = channel::<u64>(1);

        let producer = thread::spawn(move || {
            for i in 0..COUNT {
                tx.send(i).unwrap();
            }
        });

        let consumer = thread::spawn(move || {
            let mut expected = 0u64;
            while expected < COUNT {
                let val = rx.recv().unwrap();
                assert_eq!(val, expected);
                expected += 1;
            }
        });

        producer.join().unwrap();
        consumer.join().unwrap();
    }

    // ============================================================================
    // Stress Tests - Multiple Producers
    // ============================================================================

    #[test]
    fn stress_multiple_producers() {
        const PRODUCERS: usize = 4;
        const PER_PRODUCER: u64 = 25_000;

        let (tx, mut rx) = channel::<u64>(1024);

        let handles: Vec<_> = (0..PRODUCERS)
            .map(|_| {
                let tx = tx.clone();
                thread::spawn(move || {
                    for i in 0..PER_PRODUCER {
                        tx.send(i).unwrap();
                    }
                })
            })
            .collect();

        drop(tx);

        let mut received = 0u64;
        while let Ok(_) = rx.recv() {
            received += 1;
        }

        for h in handles {
            h.join().unwrap();
        }

        assert_eq!(received, PRODUCERS as u64 * PER_PRODUCER);
    }

    #[test]
    fn stress_many_producers_small_buffer() {
        const PRODUCERS: usize = 8;
        const PER_PRODUCER: u64 = 1_000;

        let (tx, mut rx) = channel::<u64>(4);

        let handles: Vec<_> = (0..PRODUCERS)
            .map(|id| {
                let tx = tx.clone();
                thread::spawn(move || {
                    for i in 0..PER_PRODUCER {
                        tx.send(id as u64 * PER_PRODUCER + i).unwrap();
                    }
                })
            })
            .collect();

        drop(tx);

        let mut received = 0u64;
        while let Ok(_) = rx.recv() {
            received += 1;
        }

        for h in handles {
            h.join().unwrap();
        }

        assert_eq!(received, PRODUCERS as u64 * PER_PRODUCER);
    }

    #[test]
    fn stress_producers_dropping_mid_stream() {
        const PRODUCERS: usize = 8;

        let (tx, mut rx) = channel::<u64>(64);

        let handles: Vec<_> = (0..PRODUCERS)
            .map(|id| {
                let tx = tx.clone();
                thread::spawn(move || {
                    for i in 0..1000 {
                        // Randomly fail some sends to simulate producer drops
                        if i == 500 && id % 2 == 0 {
                            return; // Early exit for some producers
                        }
                        tx.send(id as u64 * 1000 + i).unwrap();
                    }
                })
            })
            .collect();

        drop(tx);

        let mut count = 0;
        while let Ok(_) = rx.recv() {
            count += 1;
        }

        for h in handles {
            h.join().unwrap();
        }

        // Half producers send 500, half send 1000
        // 4 * 500 + 4 * 1000 = 6000
        assert_eq!(count, 6000);
    }

    // ============================================================================
    // Producer Contention
    // ============================================================================

    #[test]
    fn high_contention_producers() {
        const PRODUCERS: usize = 16;
        const PER_PRODUCER: u64 = 5_000;

        let (tx, mut rx) = channel::<u64>(256);

        let handles: Vec<_> = (0..PRODUCERS)
            .map(|_| {
                let tx = tx.clone();
                thread::spawn(move || {
                    for i in 0..PER_PRODUCER {
                        tx.send(i).unwrap();
                    }
                })
            })
            .collect();

        drop(tx);

        let mut count = 0u64;
        while let Ok(_) = rx.recv() {
            count += 1;
        }

        for h in handles {
            h.join().unwrap();
        }

        assert_eq!(count, PRODUCERS as u64 * PER_PRODUCER);
    }

    #[test]
    fn bursting_producers() {
        let (tx, mut rx) = channel::<u64>(64);

        // Multiple producers burst-sending
        let handles: Vec<_> = (0..4)
            .map(|id| {
                let tx = tx.clone();
                thread::spawn(move || {
                    for burst in 0..100 {
                        // Burst of 10 messages
                        for i in 0..10 {
                            tx.send(id * 1000 + burst * 10 + i).unwrap();
                        }
                        // Small delay between bursts
                        if burst % 10 == 0 {
                            thread::yield_now();
                        }
                    }
                })
            })
            .collect();

        drop(tx);

        let mut count = 0;
        while let Ok(_) = rx.recv() {
            count += 1;
        }

        for h in handles {
            h.join().unwrap();
        }

        assert_eq!(count, 4 * 100 * 10);
    }

    // ============================================================================
    // Deadlock Prevention Tests
    // ============================================================================

    #[test]
    fn no_deadlock_disconnect_while_blocked_recv() {
        let (tx, mut rx) = channel::<u64>(1);

        let handle = thread::spawn(move || {
            // Will block waiting for data
            let result = rx.recv();
            assert!(result.is_err()); // Should error, not deadlock
        });

        thread::sleep(Duration::from_millis(50));
        drop(tx); // Disconnect while receiver is blocked

        handle.join().unwrap();
    }

    #[test]
    fn no_deadlock_disconnect_while_spinning_send() {
        let (tx, rx) = channel::<u64>(1);
        tx.try_send(1).unwrap(); // Fill it

        let handle = thread::spawn(move || {
            // Will spin waiting for space
            let result = tx.send(2);
            assert!(result.is_err()); // Should error, not deadlock
        });

        thread::sleep(Duration::from_millis(50));
        drop(rx); // Disconnect while sender is spinning

        handle.join().unwrap();
    }

    #[test]
    fn no_deadlock_multiple_senders_disconnect() {
        let (tx1, rx) = channel::<u64>(1);
        let tx2 = tx1.clone();

        tx1.try_send(1).unwrap(); // Fill it

        let h1 = thread::spawn(move || {
            let _ = tx1.send(2); // Will spin
        });

        let h2 = thread::spawn(move || {
            let _ = tx2.send(3); // Will spin
        });

        thread::sleep(Duration::from_millis(50));
        drop(rx); // Disconnect

        h1.join().unwrap();
        h2.join().unwrap();
    }

    // ============================================================================
    // Race Condition Tests
    // ============================================================================

    #[test]
    fn race_send_before_recv_parks() {
        // Send happens just before recv decides to park
        for _ in 0..100 {
            let (tx, mut rx) = channel::<u64>(1);

            let handle = thread::spawn(move || rx.recv().unwrap());

            // Tiny delay to let receiver potentially start parking
            thread::yield_now();
            tx.send(42).unwrap();

            assert_eq!(handle.join().unwrap(), 42);
        }
    }

    #[test]
    fn race_disconnect_during_park_transition() {
        // Disconnect happens during the brief window of parking
        for _ in 0..100 {
            let (tx, mut rx) = channel::<u64>(1);

            let handle = thread::spawn(move || {
                let _ = rx.recv(); // May succeed or fail, shouldn't deadlock
            });

            // Immediately disconnect
            drop(tx);

            handle.join().unwrap();
        }
    }

    #[test]
    fn race_multiple_producers_one_slot() {
        // Multiple producers competing for one slot
        for _ in 0..50 {
            let (tx, mut rx) = channel::<u64>(1);

            let handles: Vec<_> = (0..4)
                .map(|id| {
                    let tx = tx.clone();
                    thread::spawn(move || {
                        tx.send(id).unwrap();
                    })
                })
                .collect();

            drop(tx);

            let mut received = Vec::new();
            while let Ok(v) = rx.recv() {
                received.push(v);
            }

            for h in handles {
                h.join().unwrap();
            }

            assert_eq!(received.len(), 4);
        }
    }

    // ============================================================================
    // Stress: Rapid Park/Unpark Cycles
    // ============================================================================

    #[test]
    fn stress_rapid_unpark_receiver() {
        let (tx, mut rx) = channel::<u64>(1);

        let handle = thread::spawn(move || {
            for i in 0..10_000 {
                tx.send(i).unwrap();
            }
        });

        for _ in 0..10_000 {
            rx.recv().unwrap();
        }

        handle.join().unwrap();
    }

    #[test]
    fn stress_receiver_parks_frequently() {
        // Single producer, small buffer - receiver will park often
        let (tx, mut rx) = channel::<u64>(1);

        let handle = thread::spawn(move || {
            let mut count = 0;
            for _ in 0..50_000 {
                rx.recv().unwrap();
                count += 1;
            }
            count
        });

        for i in 0..50_000 {
            tx.send(i).unwrap();
        }

        assert_eq!(handle.join().unwrap(), 50_000);
    }

    // ============================================================================
    // Timed Tests (ensure no indefinite blocking)
    // ============================================================================

    #[test]
    fn completes_in_reasonable_time() {
        use std::sync::mpsc;

        let (done_tx, done_rx) = mpsc::channel();

        let handle = thread::spawn(move || {
            let (tx, mut rx) = channel::<u64>(1);

            let h = thread::spawn(move || {
                for i in 0..1000 {
                    tx.send(i).unwrap();
                }
            });

            for _ in 0..1000 {
                rx.recv().unwrap();
            }

            h.join().unwrap();
            done_tx.send(()).unwrap();
        });

        // Should complete in well under a second
        let result = done_rx.recv_timeout(Duration::from_secs(5));
        assert!(result.is_ok(), "Test timed out - possible deadlock!");

        handle.join().unwrap();
    }

    #[test]
    fn does_not_hang_on_disconnect_during_recv() {
        let done = Arc::new(AtomicBool::new(false));
        let done_clone = done.clone();

        let (tx, mut rx) = channel::<u64>(4);

        let handle = thread::spawn(move || {
            let _ = rx.recv(); // Will block, then return Err on disconnect
            done_clone.store(true, Ordering::SeqCst);
        });

        thread::sleep(Duration::from_millis(50));
        assert!(!done.load(Ordering::SeqCst)); // Still blocked

        drop(tx);

        handle.join().unwrap();
        assert!(done.load(Ordering::SeqCst)); // Completed
    }

    #[test]
    fn does_not_hang_on_disconnect_during_send() {
        let done = Arc::new(AtomicBool::new(false));
        let done_clone = done.clone();

        let (tx, rx) = channel::<u64>(1);
        tx.try_send(1).unwrap(); // Fill it

        let handle = thread::spawn(move || {
            let _ = tx.send(2); // Will spin, then return Err on disconnect
            done_clone.store(true, Ordering::SeqCst);
        });

        thread::sleep(Duration::from_millis(50));
        assert!(!done.load(Ordering::SeqCst)); // Still spinning

        drop(rx);

        handle.join().unwrap();
        assert!(done.load(Ordering::SeqCst)); // Completed
    }

    // ============================================================================
    // Rapid Connect/Disconnect
    // ============================================================================

    #[test]
    fn rapid_channel_creation() {
        for _ in 0..1000 {
            let (tx, mut rx) = channel::<u64>(4);
            tx.try_send(1).unwrap();
            assert_eq!(rx.recv().unwrap(), 1);
        }
    }

    #[test]
    fn rapid_disconnect() {
        for _ in 0..1000 {
            let (tx, rx) = channel::<u64>(4);
            drop(tx);
            drop(rx);
        }
    }

    #[test]
    fn rapid_clone_and_drop() {
        for _ in 0..100 {
            // Buffer must be large enough to hold all messages since we send before receiving
            let (tx, mut rx) = channel::<u64>(16);

            let clones: Vec<_> = (0..10).map(|_| tx.clone()).collect();

            for (i, c) in clones.into_iter().enumerate() {
                c.send(i as u64).unwrap();
                drop(c);
            }

            drop(tx);

            let mut count = 0;
            while let Ok(_) = rx.recv() {
                count += 1;
            }

            assert_eq!(count, 10);
        }
    }

    // ============================================================================
    // Debug Trait
    // ============================================================================

    #[test]
    fn debug_impl_sender() {
        let (tx, _rx) = channel::<u64>(4);
        let debug_str = format!("{:?}", tx);
        assert!(debug_str.contains("Sender"));
        assert!(debug_str.contains("capacity"));
    }

    #[test]
    fn debug_impl_receiver() {
        let (_tx, rx) = channel::<u64>(4);
        let debug_str = format!("{:?}", rx);
        assert!(debug_str.contains("Receiver"));
        assert!(debug_str.contains("capacity"));
    }

    // ============================================================================
    // Capacity Method
    // ============================================================================

    #[test]
    fn capacity_method() {
        let (tx, rx) = channel::<u64>(100);
        // Capacity rounds up to power of 2
        assert!(tx.capacity() >= 100);
        assert!(rx.capacity() >= 100);
        assert_eq!(tx.capacity(), rx.capacity());
    }

    // ============================================================================
    // Edge Cases
    // ============================================================================

    #[test]
    fn send_recv_after_partial_fill_drain() {
        let (tx, mut rx) = channel::<u64>(4);

        // Partial fill
        tx.send(1).unwrap();
        tx.send(2).unwrap();

        // Partial drain
        assert_eq!(rx.recv().unwrap(), 1);

        // More sends
        tx.send(3).unwrap();
        tx.send(4).unwrap();

        // Drain rest
        assert_eq!(rx.recv().unwrap(), 2);
        assert_eq!(rx.recv().unwrap(), 3);
        assert_eq!(rx.recv().unwrap(), 4);
    }

    #[test]
    fn interleaved_send_recv() {
        let (tx, mut rx) = channel::<u64>(2);

        for i in 0..100 {
            tx.send(i).unwrap();
            assert_eq!(rx.recv().unwrap(), i);
        }
    }
}