theta_sync/

lib.rs

#![no_std]

extern crate alloc;

#[cfg(test)]
#[path = "attacks/mod.rs"]
pub mod attacks;

use alloc::{boxed::Box, sync::Arc};
use core::{
    cell::UnsafeCell,
    fmt,
    future::Future,
    hash::{Hash, Hasher},
    mem::{ManuallyDrop, MaybeUninit},
    pin::Pin,
    ptr,
    sync::atomic::{AtomicBool, AtomicPtr, AtomicUsize, Ordering},
    task::{Context, Poll, Waker},
};

/// Block capacity - platform dependent for optimal memory usage
/// 32 messages on 64-bit targets, 16 on 32-bit targets (same as Tokio)
#[cfg(target_pointer_width = "64")]
const BLOCK_CAP: usize = 32;
#[cfg(target_pointer_width = "32")]
const BLOCK_CAP: usize = 16;

/// Error returned when sending on a closed channel
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct SendError<T>(pub T);

impl<T> fmt::Display for SendError<T> {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        write!(f, "sending on a closed channel")
    }
}

/// Error returned when receiving fails
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum TryRecvError {
    /// The channel is currently empty
    Empty,
    /// The channel is closed and empty
    Disconnected,
}

impl fmt::Display for TryRecvError {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        match self {
            TryRecvError::Empty => write!(f, "channel is empty"),
            TryRecvError::Disconnected => write!(f, "channel is disconnected"),
        }
    }
}

/// Creates an unbounded MPSC channel
pub fn unbounded_channel<T>() -> (UnboundedSender<T>, UnboundedReceiver<T>) {
    let block = Block::new(0);
    let block_ptr = Box::into_raw(Box::new(block));

    let shared = Arc::new(Shared {
        head: AtomicPtr::new(block_ptr),
        tail: AtomicPtr::new(block_ptr),
        rx_waker: AtomicPtr::new(ptr::null_mut()),
        waker_lock: AtomicBool::new(false),
        num_senders: AtomicUsize::new(1),
        num_weak_senders: AtomicUsize::new(0),
        closed: AtomicBool::new(false),
    });

    let sender = UnboundedSender {
        shared: Arc::clone(&shared),
    };
    let receiver = UnboundedReceiver {
        shared,
        recv_index: 0,
    };

    (sender, receiver)
}

/// Block structure for the linked list
struct Block<T> {
    /// Next block in the linked list
    next: AtomicPtr<Block<T>>,
    /// Starting index for this block
    start_index: usize,
    /// Values stored in this block
    values: UnsafeCell<[MaybeUninit<ManuallyDrop<T>>; BLOCK_CAP]>,
    /// Bit set indicating which slots are ready
    ready_slots: AtomicUsize,
    /// Number of values written to this block
    len: AtomicUsize,
}

impl<T> Block<T> {
    fn new(start_index: usize) -> Self {
        Self {
            next: AtomicPtr::new(ptr::null_mut()),
            start_index,
            values: UnsafeCell::new([const { MaybeUninit::uninit() }; BLOCK_CAP]),
            ready_slots: AtomicUsize::new(0),
            len: AtomicUsize::new(0),
        }
    }

    /// Returns the relative index within this block for the given global index
    fn relative_index(&self, index: usize) -> Option<usize> {
        if index >= self.start_index && index < self.start_index + BLOCK_CAP {
            Some(index - self.start_index)
        } else {
            None
        }
    }

    /// Writes a value to the specified slot
    /// Returns Ok(()) if successful, Err(value) if slot is already occupied
    fn write(&self, relative_index: usize, value: T) -> Result<(), T> {
        if relative_index >= BLOCK_CAP {
            return Err(value);
        }

        let mask = 1 << relative_index;
        
        // Try to set the ready bit atomically
        let prev_ready = self.ready_slots.fetch_or(mask, Ordering::AcqRel);

        if prev_ready & mask != 0 {
            // Slot already occupied, return the value back
            return Err(value);
        }

        // Write the value to the slot
        unsafe {
            let values = &mut *self.values.get();
            values[relative_index].write(ManuallyDrop::new(value));
        }

        // Don't increment len here since it's already done atomically in send()
        Ok(())
    }

    /// Reads a value from the specified slot
    /// Returns None if slot is empty or not ready
    fn read(&self, relative_index: usize) -> Option<T> {
        if relative_index >= BLOCK_CAP {
            return None;
        }

        let mask = 1 << relative_index;

        // Try to clear the ready bit atomically to claim the slot
        let prev_ready = self.ready_slots.fetch_and(!mask, Ordering::AcqRel);

        if prev_ready & mask == 0 {
            // Slot was not ready, restore the bit and return None
            return None;
        }

        unsafe {
            let values = &*self.values.get();
            Some(ManuallyDrop::into_inner(values[relative_index].assume_init_read()))
        }
    }

    /// Checks if a slot is ready
    fn is_ready(&self, relative_index: usize) -> bool {
        if relative_index >= BLOCK_CAP {
            return false;
        }

        let mask = 1 << relative_index;
        self.ready_slots.load(Ordering::Acquire) & mask != 0
    }

    /// Returns the number of ready slots
    fn ready_count(&self) -> usize {
        self.ready_slots.load(Ordering::Acquire).count_ones() as usize
    }
}

impl<T> Drop for Block<T> {
    fn drop(&mut self) {
        let ready = self.ready_slots.load(Ordering::Relaxed);
        unsafe {
            let values = &mut *self.values.get();
            for i in 0..BLOCK_CAP {
                if ready & (1 << i) != 0 {
                    ManuallyDrop::drop(values[i].assume_init_mut());
                }
            }
        }
    }
}

/// Shared state between senders and receiver
struct Shared<T> {
    /// Pointer to the head block (where new values are written)
    head: AtomicPtr<Block<T>>,
    /// Pointer to the tail block (where values are read)
    tail: AtomicPtr<Block<T>>,
    /// Waker for the receiver task - using atomic approach for no_std compatibility
    rx_waker: AtomicPtr<Waker>,
    /// Atomic flag to prevent concurrent waker access
    waker_lock: AtomicBool,
    /// Number of active senders (strong references)
    num_senders: AtomicUsize,
    /// Number of weak senders
    num_weak_senders: AtomicUsize,
    /// Channel closed flag
    closed: AtomicBool,
}

impl<T> Shared<T> {
    /// Atomically take and wake the stored waker
    fn wake_receiver(&self) {
        // Quick check if there's even a waker to avoid unnecessary locking
        if self.rx_waker.load(Ordering::Acquire).is_null() {
            return; // No waker, skip the expensive spin lock
        }
        
        // Spin lock to ensure exclusive access to waker
        while self.waker_lock.compare_exchange_weak(
            false, 
            true, 
            Ordering::Acquire, 
            Ordering::Relaxed
        ).is_err() {
            core::hint::spin_loop();
        }
        
        // Take the waker if it exists
        let waker_ptr = self.rx_waker.swap(ptr::null_mut(), Ordering::Acquire);
        if !waker_ptr.is_null() {
            let waker = unsafe { Box::from_raw(waker_ptr) };
            // Release lock before waking to avoid holding it during wake
            self.waker_lock.store(false, Ordering::Release);
            waker.wake();
        } else {
            // Release lock
            self.waker_lock.store(false, Ordering::Release);
        }
    }
    
    /// Atomically store a new waker
    fn store_waker(&self, waker: Waker) {
        // Spin lock to ensure exclusive access to waker  
        while self.waker_lock.compare_exchange_weak(
            false,
            true,
            Ordering::Acquire,
            Ordering::Relaxed
        ).is_err() {
            core::hint::spin_loop();
        }
        
        // Replace the old waker
        let new_waker_ptr = Box::into_raw(Box::new(waker));
        let old_waker_ptr = self.rx_waker.swap(new_waker_ptr, Ordering::Release);
        
        // Clean up old waker if it existed
        if !old_waker_ptr.is_null() {
            unsafe { drop(Box::from_raw(old_waker_ptr)) };
        }
        
        // Release lock
        self.waker_lock.store(false, Ordering::Release);
    }
}

unsafe impl<T: Send> Send for Shared<T> {}
unsafe impl<T: Send> Sync for Shared<T> {}

impl<T> Drop for Shared<T> {
    fn drop(&mut self) {
        // Clean up waker
        let waker_ptr = self.rx_waker.load(Ordering::Relaxed);
        if !waker_ptr.is_null() {
            unsafe { drop(Box::from_raw(waker_ptr)) };
        }
        
        // Clean up all blocks in the linked list
        let mut current = self.tail.load(Ordering::Relaxed);
        while !current.is_null() {
            let block = unsafe { Box::from_raw(current) };
            current = block.next.load(Ordering::Relaxed);
        }
    }
}

/// The sending half of an unbounded MPSC channel
pub struct UnboundedSender<T> {
    shared: Arc<Shared<T>>,
}

/// An unbounded sender that does not prevent the channel from being closed.
///
/// If all [`UnboundedSender`] instances of a channel were dropped and only
/// `WeakUnboundedSender` instances remain, the channel is closed.
///
/// In order to send messages, the `WeakUnboundedSender` needs to be upgraded using
/// [`WeakUnboundedSender::upgrade`], which returns `Option<UnboundedSender>`. It returns `None`
/// if all `UnboundedSender`s have been dropped, and otherwise it returns an `UnboundedSender`.
///
/// [`UnboundedSender`]: UnboundedSender
/// [`WeakUnboundedSender::upgrade`]: WeakUnboundedSender::upgrade
///
/// # Examples
///
/// ```
/// use theta_sync::unbounded_channel;
///
/// let (tx, _rx) = unbounded_channel::<i32>();
/// let tx_weak = tx.downgrade();
///
/// // Upgrading will succeed because `tx` still exists.
/// assert!(tx_weak.upgrade().is_some());
///
/// // If we drop `tx`, then it will fail.
/// drop(tx);
/// assert!(tx_weak.upgrade().is_none());
/// ```
pub struct WeakUnboundedSender<T> {
    shared: Arc<Shared<T>>,
}

impl<T> Clone for WeakUnboundedSender<T> {
    fn clone(&self) -> Self {
        self.shared.num_weak_senders.fetch_add(1, Ordering::Relaxed);
        Self {
            shared: Arc::clone(&self.shared),
        }
    }
}

impl<T> Drop for WeakUnboundedSender<T> {
    fn drop(&mut self) {
        self.shared.num_weak_senders.fetch_sub(1, Ordering::AcqRel);
    }
}

impl<T> Clone for UnboundedSender<T> {
    fn clone(&self) -> Self {
        self.shared.num_senders.fetch_add(1, Ordering::Relaxed);
        Self {
            shared: Arc::clone(&self.shared),
        }
    }
}

impl<T> Drop for UnboundedSender<T> {
    fn drop(&mut self) {
        let prev_count = self.shared.num_senders.fetch_sub(1, Ordering::AcqRel);
        if prev_count == 1 {
            // Last sender dropped, close the channel
            self.shared.closed.store(true, Ordering::Release);

            // Wake up the receiver
            self.shared.wake_receiver();
        }
    }
}

impl<T> UnboundedSender<T> {
    /// Sends a value on this channel
    ///
    /// This method never blocks. It will return an error if the receiver has been dropped.
    pub fn send(&self, mut value: T) -> Result<(), SendError<T>> {
        if self.shared.closed.load(Ordering::Acquire) {
            return Err(SendError(value));
        }

        let mut attempts = 0;
        loop {
            let head_ptr = self.shared.head.load(Ordering::Acquire);
            let head = unsafe { &*head_ptr };

            // Atomically allocate the next slot to maintain FIFO ordering
            let slot_idx = head.len.fetch_add(1, Ordering::AcqRel);
            
            if slot_idx < BLOCK_CAP {
                // We have a valid slot index, try to write to it
                // This should always succeed since we atomically allocated the slot
                match head.write(slot_idx, value) {
                    Ok(()) => {
                        // Successfully written, wake receiver
                        self.shared.wake_receiver();
                        return Ok(());
                    }
                    Err(returned_value) => {
                        // This should rarely happen, but if it does, the slot was somehow occupied
                        // We need to retry with a new allocation
                        value = returned_value;
                        head.len.fetch_sub(1, Ordering::AcqRel); // Rollback the allocation
                        // Continue the loop to try again
                        continue;
                    }
                }
            } else {
                // Block is full, reset the len counter to prevent overflow
                head.len.store(BLOCK_CAP, Ordering::Release);
            }

            // Block is full, need to allocate a new one
            let next_ptr = head.next.load(Ordering::Acquire);
            if next_ptr.is_null() {
                // Try to allocate and link a new block
                let new_block = Box::into_raw(Box::new(Block::new(head.start_index + BLOCK_CAP)));

                match head.next.compare_exchange_weak(
                    ptr::null_mut(),
                    new_block,
                    Ordering::AcqRel,
                    Ordering::Acquire,
                ) {
                    Ok(_) => {
                        // Successfully linked new block, update head
                        self.shared.head.store(new_block, Ordering::Release);
                    }
                    Err(_) => {
                        // Someone else allocated a block, free ours
                        unsafe { drop(Box::from_raw(new_block)) };
                    }
                }
            } else {
                // Move head to the next block
                self.shared
                    .head
                    .compare_exchange_weak(head_ptr, next_ptr, Ordering::AcqRel, Ordering::Acquire)
                    .ok();
            }

            attempts += 1;
            if attempts > 1000 {
                // Prevent infinite loops in pathological cases
                core::hint::spin_loop();
                attempts = 0;
            }
        }
    }

    /// Returns a unique identifier for this channel based on the shared pointer address
    pub fn id(&self) -> ChannelId {
        ChannelId(Arc::as_ptr(&self.shared) as usize)
    }

    /// Checks if the channel is closed
    pub fn is_closed(&self) -> bool {
        self.shared.closed.load(Ordering::Acquire)
    }

    /// Returns true if this sender and another sender send to the same channel
    pub fn same_channel(&self, other: &Self) -> bool {
        Arc::ptr_eq(&self.shared, &other.shared)
    }

    /// Converts the `UnboundedSender` to a [`WeakUnboundedSender`] that does not count
    /// towards RAII semantics, i.e. if all `UnboundedSender` instances of the
    /// channel were dropped and only `WeakUnboundedSender` instances remain,
    /// the channel is closed.
    #[must_use = "Downgrade creates a WeakSender without destroying the original non-weak sender."]
    pub fn downgrade(&self) -> WeakUnboundedSender<T> {
        self.shared.num_weak_senders.fetch_add(1, Ordering::Relaxed);
        WeakUnboundedSender {
            shared: Arc::clone(&self.shared),
        }
    }

    /// Returns the number of [`UnboundedSender`] handles.
    pub fn strong_count(&self) -> usize {
        self.shared.num_senders.load(Ordering::Acquire)
    }

    /// Returns the number of [`WeakUnboundedSender`] handles.
    pub fn weak_count(&self) -> usize {
        self.shared.num_weak_senders.load(Ordering::Acquire)
    }

    /// Completes when the receiver has dropped.
    ///
    /// This allows the producers to get notified when interest in the produced values is canceled and immediately stop doing work.
    pub async fn closed(&self) {
        ClosedFuture { sender: self }.await
    }
}

impl<T> WeakUnboundedSender<T> {
    /// Tries to convert a `WeakUnboundedSender` into an [`UnboundedSender`].
    /// This will return `Some` if there are other `UnboundedSender` instances alive and
    /// the channel wasn't previously dropped, otherwise `None` is returned.
    pub fn upgrade(&self) -> Option<UnboundedSender<T>> {
        let mut count = self.shared.num_senders.load(Ordering::Acquire);

        loop {
            if count == 0 {
                // No strong senders remaining, cannot upgrade
                return None;
            }

            match self.shared.num_senders.compare_exchange_weak(
                count,
                count + 1,
                Ordering::AcqRel,
                Ordering::Acquire,
            ) {
                Ok(_) => {
                    return Some(UnboundedSender {
                        shared: Arc::clone(&self.shared),
                    });
                }
                Err(actual) => count = actual,
            }
        }
    }

    /// Returns the number of [`UnboundedSender`] handles.
    pub fn strong_count(&self) -> usize {
        self.shared.num_senders.load(Ordering::Acquire)
    }

    /// Returns the number of [`WeakUnboundedSender`] handles.
    pub fn weak_count(&self) -> usize {
        self.shared.num_weak_senders.load(Ordering::Acquire)
    }
}

impl<T> PartialEq for UnboundedSender<T> {
    fn eq(&self, other: &Self) -> bool {
        self.id() == other.id()
    }
}

impl<T> Eq for UnboundedSender<T> {}

impl<T> PartialOrd for UnboundedSender<T> {
    fn partial_cmp(&self, other: &Self) -> Option<core::cmp::Ordering> {
        Some(self.cmp(other))
    }
}

impl<T> Ord for UnboundedSender<T> {
    fn cmp(&self, other: &Self) -> core::cmp::Ordering {
        self.id().cmp(&other.id())
    }
}

impl<T> Hash for UnboundedSender<T> {
    fn hash<H: Hasher>(&self, state: &mut H) {
        self.id().hash(state);
    }
}

impl<T> PartialEq for WeakUnboundedSender<T> {
    fn eq(&self, other: &Self) -> bool {
        Arc::ptr_eq(&self.shared, &other.shared)
    }
}

impl<T> Eq for WeakUnboundedSender<T> {}

impl<T> PartialOrd for WeakUnboundedSender<T> {
    fn partial_cmp(&self, other: &Self) -> Option<core::cmp::Ordering> {
        Some(self.cmp(other))
    }
}

impl<T> Ord for WeakUnboundedSender<T> {
    fn cmp(&self, other: &Self) -> core::cmp::Ordering {
        let self_ptr = Arc::as_ptr(&self.shared) as usize;
        let other_ptr = Arc::as_ptr(&other.shared) as usize;
        self_ptr.cmp(&other_ptr)
    }
}

impl<T> Hash for WeakUnboundedSender<T> {
    fn hash<H: Hasher>(&self, state: &mut H) {
        let ptr = Arc::as_ptr(&self.shared) as usize;
        ptr.hash(state);
    }
}

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

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

/// The receiving half of an unbounded MPSC channel
pub struct UnboundedReceiver<T> {
    shared: Arc<Shared<T>>,
    /// Current position for reading
    recv_index: usize,
}

impl<T> UnboundedReceiver<T> {
    /// Receives a value from the channel asynchronously
    pub async fn recv(&mut self) -> Option<T> {
        RecvFuture { receiver: self }.await
    }

    /// Attempts to receive a value without blocking
    pub fn try_recv(&mut self) -> Result<T, TryRecvError> {
        loop {
            let tail_ptr = self.shared.tail.load(Ordering::Acquire);
            let tail = unsafe { &*tail_ptr };

            if let Some(relative_idx) = tail.relative_index(self.recv_index) {
                // We're in the current tail block
                if tail.is_ready(relative_idx) {
                    if let Some(value) = tail.read(relative_idx) {
                        self.recv_index += 1;
                        return Ok(value);
                    }
                }

                // Check if we need to move to the next block
                if relative_idx == BLOCK_CAP - 1
                    || tail.ready_count() == tail.len.load(Ordering::Acquire)
                {
                    let next_ptr = tail.next.load(Ordering::Acquire);
                    if !next_ptr.is_null() {
                        // Move to next block and try to free the current one
                        if self
                            .shared
                            .tail
                            .compare_exchange(
                                tail_ptr,
                                next_ptr,
                                Ordering::AcqRel,
                                Ordering::Acquire,
                            )
                            .is_ok()
                        {
                            // Successfully moved tail, free the old block
                            unsafe { drop(Box::from_raw(tail_ptr)) };
                        }
                        continue;
                    }
                }
            } else {
                // We're behind the current tail block, advance
                let next_ptr = tail.next.load(Ordering::Acquire);
                if !next_ptr.is_null() {
                    if self
                        .shared
                        .tail
                        .compare_exchange(tail_ptr, next_ptr, Ordering::AcqRel, Ordering::Acquire)
                        .is_ok()
                    {
                        unsafe { drop(Box::from_raw(tail_ptr)) };
                    }
                    continue;
                }
            }

            // No data available
            if self.shared.closed.load(Ordering::Acquire)
                && self.shared.num_senders.load(Ordering::Acquire) == 0
            {
                return Err(TryRecvError::Disconnected);
            }

            return Err(TryRecvError::Empty);
        }
    }

    /// Returns a unique identifier for this channel based on the shared pointer address
    pub fn id(&self) -> ChannelId {
        ChannelId(Arc::as_ptr(&self.shared) as usize)
    }

    /// Checks if the channel is closed and empty
    pub fn is_closed(&self) -> bool {
        self.shared.closed.load(Ordering::Acquire)
            && self.shared.num_senders.load(Ordering::Acquire) == 0
    }

    /// Checks if the channel is currently empty
    pub fn is_empty(&self) -> bool {
        // We need to create a temporary receiver to check if it's empty
        // Since this method takes &self, we can't call try_recv which requires &mut self
        // Instead, we'll check the state without modifying the receiver

        let tail_ptr = self.shared.tail.load(Ordering::Acquire);
        let tail = unsafe { &*tail_ptr };

        if let Some(relative_idx) = tail.relative_index(self.recv_index) {
            // Check if there's data ready at our current position
            if tail.is_ready(relative_idx) {
                return false; // Not empty
            }
        }

        // Check if channel is disconnected
        if self.shared.closed.load(Ordering::Acquire)
            && self.shared.num_senders.load(Ordering::Acquire) == 0
        {
            return true; // Empty and disconnected
        }

        // Assume empty if we can't find ready data at current position
        true
    }

    /// Closes the receiving half of the channel without dropping it
    pub fn close(&mut self) {
        self.shared.closed.store(true, Ordering::Release);
    }

    /// Returns the number of [`UnboundedSender`] handles.
    pub fn sender_strong_count(&self) -> usize {
        self.shared.num_senders.load(Ordering::Acquire)
    }

    /// Returns the number of [`WeakUnboundedSender`] handles.
    pub fn sender_weak_count(&self) -> usize {
        self.shared.num_weak_senders.load(Ordering::Acquire)
    }

    /// Returns the number of messages in the channel.
    pub fn len(&self) -> usize {
        // Count ready messages across all blocks
        let mut count = 0;
        let mut current = self.shared.tail.load(Ordering::Acquire);
        
        while !current.is_null() {
            let block = unsafe { &*current };
            count += block.ready_count();
            current = block.next.load(Ordering::Acquire);
        }
        
        count
    }

    fn poll_recv(&mut self, cx: &mut Context<'_>) -> Poll<Option<T>> {
        match self.try_recv() {
            Ok(value) => Poll::Ready(Some(value)),
            Err(TryRecvError::Disconnected) => Poll::Ready(None),
            Err(TryRecvError::Empty) => {
                // Store waker and return Pending
                self.shared.store_waker(cx.waker().clone());
                Poll::Pending
            }
        }
    }
}

impl<T> Drop for UnboundedReceiver<T> {
    fn drop(&mut self) {
        self.shared.closed.store(true, Ordering::Release);
    }
}

/// Future returned by `UnboundedReceiver::recv()`
struct RecvFuture<'a, T> {
    receiver: &'a mut UnboundedReceiver<T>,
}

impl<'a, T> Future for RecvFuture<'a, T> {
    type Output = Option<T>;

    fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
        self.receiver.poll_recv(cx)
    }
}

/// Future returned by `UnboundedSender::closed()`
struct ClosedFuture<'a, T> {
    sender: &'a UnboundedSender<T>,
}

impl<'a, T> Future for ClosedFuture<'a, T> {
    type Output = ();

    fn poll(self: Pin<&mut Self>, _cx: &mut Context<'_>) -> Poll<Self::Output> {
        if self.sender.is_closed() {
            Poll::Ready(())
        } else {
            // In a real implementation, we'd store the waker and wake it when the channel closes
            // For now, just return Pending - this is a simplified implementation
            Poll::Pending
        }
    }
}

/// Unique identifier for a channel based on its memory address
#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord)]
pub struct ChannelId(usize);

impl Hash for ChannelId {
    fn hash<H: Hasher>(&self, state: &mut H) {
        self.0.hash(state);
    }
}

impl fmt::Display for ChannelId {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        write!(f, "ChannelId(0x{:x})", self.0)
    }
}

#[cfg(test)]  
mod tests {
    use super::*;
    use alloc::{vec, vec::Vec};

    #[test]
    fn test_basic_send_recv() {
        let (tx, mut rx) = unbounded_channel::<i32>();

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

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

    #[test]
    fn test_channel_id() {
        let (tx1, rx1) = unbounded_channel::<i32>();
        let (tx2, rx2) = unbounded_channel::<i32>();

        assert_eq!(tx1.id(), rx1.id());
        assert_ne!(tx1.id(), tx2.id());
        assert_ne!(rx1.id(), rx2.id());
    }

    #[test]
    fn test_clone_sender() {
        let (tx, mut rx) = unbounded_channel::<i32>();
        let tx2 = tx.clone();

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

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

    #[test]
    fn test_large_number_of_messages() {
        let (tx, mut rx) = unbounded_channel::<usize>();

        // Send more messages than a single block can hold
        for i in 0..100 {
            tx.send(i).unwrap();
        }

        // Receive all messages
        for i in 0..100 {
            assert_eq!(rx.try_recv().unwrap(), i);
        }
    }

    #[test]
    fn test_drop_sender_closes_channel() {
        let (tx, mut rx) = unbounded_channel::<i32>();

        tx.send(42).unwrap();
        drop(tx);

        assert_eq!(rx.try_recv().unwrap(), 42);
        assert!(matches!(rx.try_recv(), Err(TryRecvError::Disconnected)));
    }

    #[test]
    fn test_same_channel() {
        let (tx1, _rx) = unbounded_channel::<i32>();
        let tx2 = tx1.clone();
        let (tx3, _rx2) = unbounded_channel::<i32>();

        assert!(tx1.same_channel(&tx2));
        assert!(!tx1.same_channel(&tx3));
    }

    // === EDGE CASES AND INTENSIVE TESTS ===

    #[test]
    fn test_stress_many_messages() {
        let (tx, mut rx) = unbounded_channel::<usize>();
        const NUM_MESSAGES: usize = 10_000;

        // Send many messages
        for i in 0..NUM_MESSAGES {
            tx.send(i).unwrap();
        }

        // Receive all messages in order
        for i in 0..NUM_MESSAGES {
            assert_eq!(rx.try_recv().unwrap(), i);
        }

        assert!(matches!(rx.try_recv(), Err(TryRecvError::Empty)));
    }

    #[test]
    fn test_send_recv_interleaved() {
        let (tx, mut rx) = unbounded_channel::<i32>();

        // Interleave sends and receives
        let mut expected_recv = 0;
        for i in 0..100 {
            tx.send(i).unwrap();
            if i % 2 == 0 {
                assert_eq!(rx.try_recv().unwrap(), expected_recv);
                expected_recv += 1;
            }
        }

        // Receive remaining messages
        while let Ok(value) = rx.try_recv() {
            assert_eq!(value, expected_recv);
            expected_recv += 1;
        }
        
        assert_eq!(expected_recv, 100); // Should have received all 100 messages
    }

    #[test]
    fn test_drop_receiver_while_sending() {
        let (tx, rx) = unbounded_channel::<i32>();

        // Send some messages
        tx.send(1).unwrap();
        tx.send(2).unwrap();

        // Drop receiver
        drop(rx);

        // Further sends should fail
        assert!(matches!(tx.send(3), Err(SendError(3))));
        assert!(tx.is_closed());
    }

    #[test]
    fn test_multiple_sender_drops() {
        let (tx, mut rx) = unbounded_channel::<i32>();
        let tx2 = tx.clone();
        let tx3 = tx.clone();

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

        // Drop senders one by one
        drop(tx);
        assert!(!rx.is_closed());

        drop(tx2);
        assert!(!rx.is_closed());

        // Last sender drop should close channel
        drop(tx3);

        // Receive existing messages
        assert_eq!(rx.try_recv().unwrap(), 1);
        assert_eq!(rx.try_recv().unwrap(), 2);
        assert_eq!(rx.try_recv().unwrap(), 3);

        // Channel should be disconnected
        assert!(matches!(rx.try_recv(), Err(TryRecvError::Disconnected)));
        assert!(rx.is_closed());
    }

    #[test]
    fn test_zero_sized_types() {
        #[derive(Debug, PartialEq)]
        struct ZeroSized;

        let (tx, mut rx) = unbounded_channel::<ZeroSized>();

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

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

    #[test]
    fn test_large_types() {
        #[derive(Debug, PartialEq)]
        struct LargeType([u8; 1024]);

        let (tx, mut rx) = unbounded_channel::<LargeType>();
        let large_value = LargeType([42; 1024]);

        tx.send(large_value).unwrap();
        let received = rx.try_recv().unwrap();
        assert_eq!(received.0[0], 42);
        assert_eq!(received.0[1023], 42);
    }

    #[test]
    fn test_unwind_safety_basic() {
        #[derive(Debug)]
        struct PanicOnDrop(bool);
        impl Drop for PanicOnDrop {
            fn drop(&mut self) {
                if self.0 {
                    // In no_std, we can't actually panic in tests easily,
                    // so we'll just test the structure
                }
            }
        }

        let (tx, mut rx) = unbounded_channel::<PanicOnDrop>();

        // Send a value that won't panic on drop
        tx.send(PanicOnDrop(false)).unwrap();

        // Channel should work normally
        assert_eq!(rx.try_recv().unwrap().0, false);
        tx.send(PanicOnDrop(false)).unwrap();
        assert_eq!(rx.try_recv().unwrap().0, false);
    }

    #[test]
    fn test_block_boundary_conditions() {
        let (tx, mut rx) = unbounded_channel::<usize>();

        // Send exactly BLOCK_CAP messages to fill first block
        for i in 0..BLOCK_CAP {
            tx.send(i).unwrap();
        }

        // Send one more to trigger new block allocation
        tx.send(BLOCK_CAP).unwrap();

        // Receive all messages
        for i in 0..=BLOCK_CAP {
            assert_eq!(rx.try_recv().unwrap(), i);
        }

        // Send messages across multiple blocks
        for i in 0..(BLOCK_CAP * 3) {
            tx.send(i).unwrap();
        }

        for i in 0..(BLOCK_CAP * 3) {
            assert_eq!(rx.try_recv().unwrap(), i);
        }
    }

    #[test]
    fn test_receiver_state_consistency() {
        let (tx, mut rx) = unbounded_channel::<i32>();

        // Test empty state
        assert!(rx.is_empty());
        assert!(!rx.is_closed());

        // Send and test non-empty state
        tx.send(42).unwrap();
        assert!(!rx.is_empty());
        assert!(!rx.is_closed());

        // Receive and test empty again
        assert_eq!(rx.try_recv().unwrap(), 42);
        assert!(rx.is_empty());
        assert!(!rx.is_closed());

        // Close channel and test closed state
        drop(tx);
        assert!(rx.is_empty());
        assert!(rx.is_closed());
        assert!(matches!(rx.try_recv(), Err(TryRecvError::Disconnected)));
    }

    #[test]
    fn test_manual_close() {
        let (tx, mut rx) = unbounded_channel::<i32>();

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

        // Manually close receiver
        rx.close();

        // Should still be able to receive existing messages
        assert_eq!(rx.try_recv().unwrap(), 1);
        assert_eq!(rx.try_recv().unwrap(), 2);

        // But sender should see channel as closed
        assert!(tx.is_closed());
        assert!(matches!(tx.send(3), Err(SendError(3))));
    }

    #[test]
    fn test_channel_id_consistency() {
        // Test that sender and receiver from same channel have same ID
        let (tx, rx) = unbounded_channel::<i32>();
        assert_eq!(tx.id(), rx.id());

        // Test that different channels have different IDs (at creation time)
        let (tx2, rx2) = unbounded_channel::<i32>();
        assert_eq!(tx2.id(), rx2.id());

        // Note: Different channels may reuse memory addresses after deallocation,
        // so we only test that senders/receivers from the same channel match
        let tx_clone = tx.clone();
        assert_eq!(tx.id(), tx_clone.id());
        assert!(tx.same_channel(&tx_clone));
        assert!(!tx.same_channel(&tx2));
    }

    #[test]
    fn test_drop_semantics() {
        use alloc::rc::Rc;

        let drop_count = Rc::new(core::cell::RefCell::new(0));

        #[derive(Debug)]
        struct DropCounter(Rc<core::cell::RefCell<i32>>);
        impl Drop for DropCounter {
            fn drop(&mut self) {
                *self.0.borrow_mut() += 1;
            }
        }

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

        // Send some values
        tx.send(DropCounter(drop_count.clone())).unwrap();
        tx.send(DropCounter(drop_count.clone())).unwrap();
        tx.send(DropCounter(drop_count.clone())).unwrap();

        assert_eq!(*drop_count.borrow(), 0); // Nothing dropped yet

        // Receive one value
        let _value1 = rx.try_recv().unwrap();
        assert_eq!(*drop_count.borrow(), 0); // Still holding reference

        drop(_value1);
        assert_eq!(*drop_count.borrow(), 1); // One dropped

        // Drop the channel with remaining values
        drop(tx);
        drop(rx);

        assert_eq!(*drop_count.borrow(), 3); // All should be dropped
    }

    #[test]
    fn test_memory_safety_after_close() {
        let (tx, mut rx) = unbounded_channel::<Vec<u8>>();

        // Send some data
        tx.send(vec![1, 2, 3]).unwrap();
        tx.send(vec![4, 5, 6]).unwrap();

        // Close the receiver
        rx.close();

        // Sender should fail
        assert!(matches!(tx.send(vec![7, 8, 9]), Err(_)));

        // But we should still be able to receive existing data
        assert_eq!(rx.try_recv().unwrap(), vec![1, 2, 3]);
        assert_eq!(rx.try_recv().unwrap(), vec![4, 5, 6]);
    }

    #[test]
    fn test_ordering_guarantees() {
        let (tx, mut rx) = unbounded_channel::<usize>();

        // Send messages in order
        for i in 0..1000 {
            tx.send(i).unwrap();
        }

        // Should receive in the same order
        for i in 0..1000 {
            assert_eq!(rx.try_recv().unwrap(), i);
        }
    }

    #[test]
    fn test_empty_channel_operations() {
        let (tx, mut rx) = unbounded_channel::<i32>();

        // Operations on empty channel
        assert!(rx.is_empty());
        assert!(!rx.is_closed());
        assert!(matches!(rx.try_recv(), Err(TryRecvError::Empty)));

        // Close and test again
        drop(tx);
        assert!(rx.is_empty());
        assert!(rx.is_closed());
        assert!(matches!(rx.try_recv(), Err(TryRecvError::Disconnected)));
    }

    #[test]
    fn test_channel_reuse_after_empty() {
        let (tx, mut rx) = unbounded_channel::<i32>();

        // Send, receive, repeat multiple times
        for round in 0..10 {
            for i in 0..10 {
                tx.send(round * 10 + i).unwrap();
            }

            for i in 0..10 {
                assert_eq!(rx.try_recv().unwrap(), round * 10 + i);
            }

            assert!(matches!(rx.try_recv(), Err(TryRecvError::Empty)));
        }
    }

    #[test]
    fn test_mixed_operation_patterns() {
        let (tx, mut rx) = unbounded_channel::<usize>();

        // Pattern: send some, receive some, repeat
        let mut next_send = 0;
        let mut next_recv = 0;

        for _ in 0..100 {
            // Send 1-5 messages
            let send_count = (next_send % 5) + 1;
            for _ in 0..send_count {
                tx.send(next_send).unwrap();
                next_send += 1;
            }

            // Receive 1-3 messages (if available)
            let recv_count = (next_recv % 3) + 1;
            for _ in 0..recv_count {
                if let Ok(value) = rx.try_recv() {
                    assert_eq!(value, next_recv);
                    next_recv += 1;
                } else {
                    break;
                }
            }
        }

        // Receive remaining messages
        while let Ok(value) = rx.try_recv() {
            assert_eq!(value, next_recv);
            next_recv += 1;
        }

        assert_eq!(next_send, next_recv);
    }

    // === WEAK SENDER TESTS ===

    #[test]
    fn test_weak_sender_basic() {
        let (tx, mut rx) = unbounded_channel::<i32>();

        // Create weak sender
        let weak_tx = tx.downgrade();

        // Upgrade should succeed while strong sender exists
        let upgraded_tx = weak_tx.upgrade().unwrap();

        // Send through upgraded sender
        upgraded_tx.send(42).unwrap();
        assert_eq!(rx.try_recv().unwrap(), 42);

        // Drop original strong sender
        drop(tx);

        // Should still work with upgraded sender
        upgraded_tx.send(43).unwrap();
        assert_eq!(rx.try_recv().unwrap(), 43);

        // Drop upgraded sender
        drop(upgraded_tx);

        // Now upgrade should fail
        assert!(weak_tx.upgrade().is_none());
    }

    #[test]
    fn test_weak_sender_upgrade_failure() {
        let (tx, _rx) = unbounded_channel::<i32>();
        let weak_tx = tx.downgrade();

        // Drop strong sender
        drop(tx);

        // Upgrade should fail
        assert!(weak_tx.upgrade().is_none());
    }

    #[test]
    fn test_weak_sender_counts() {
        let (tx, rx) = unbounded_channel::<i32>();

        // Initial counts
        assert_eq!(tx.strong_count(), 1);
        assert_eq!(tx.weak_count(), 0);
        assert_eq!(rx.sender_strong_count(), 1);
        assert_eq!(rx.sender_weak_count(), 0);

        // Create weak sender
        let weak_tx = tx.downgrade();
        assert_eq!(tx.strong_count(), 1);
        assert_eq!(tx.weak_count(), 1);
        assert_eq!(weak_tx.strong_count(), 1);
        assert_eq!(weak_tx.weak_count(), 1);
        assert_eq!(rx.sender_strong_count(), 1);
        assert_eq!(rx.sender_weak_count(), 1);

        // Clone strong sender
        let tx2 = tx.clone();
        assert_eq!(tx.strong_count(), 2);
        assert_eq!(tx.weak_count(), 1);
        assert_eq!(tx2.strong_count(), 2);
        assert_eq!(weak_tx.strong_count(), 2);
        assert_eq!(weak_tx.weak_count(), 1);
        assert_eq!(rx.sender_strong_count(), 2);
        assert_eq!(rx.sender_weak_count(), 1);

        // Clone weak sender
        let weak_tx2 = weak_tx.clone();
        assert_eq!(tx.strong_count(), 2);
        assert_eq!(tx.weak_count(), 2);
        assert_eq!(weak_tx.weak_count(), 2);
        assert_eq!(weak_tx2.weak_count(), 2);
        assert_eq!(rx.sender_strong_count(), 2);
        assert_eq!(rx.sender_weak_count(), 2);

        // Drop weak sender
        drop(weak_tx);
        assert_eq!(tx.weak_count(), 1);
        assert_eq!(weak_tx2.weak_count(), 1);
        assert_eq!(rx.sender_weak_count(), 1);

        // Drop strong sender
        drop(tx);
        assert_eq!(tx2.strong_count(), 1);
        assert_eq!(weak_tx2.strong_count(), 1);
        assert_eq!(rx.sender_strong_count(), 1);

        // Drop last strong sender
        drop(tx2);
        assert_eq!(weak_tx2.strong_count(), 0);
        assert_eq!(weak_tx2.weak_count(), 1);
        assert_eq!(rx.sender_strong_count(), 0);
        assert_eq!(rx.sender_weak_count(), 1);

        // Upgrade should now fail
        assert!(weak_tx2.upgrade().is_none());
    }

    #[test]
    fn test_weak_sender_channel_close() {
        let (tx, rx) = unbounded_channel::<i32>();
        let weak_tx = tx.downgrade();

        // Drop strong sender
        drop(tx);

        // Channel should be closed
        assert!(rx.is_closed());

        // Weak sender should not be able to upgrade
        assert!(weak_tx.upgrade().is_none());
    }

    #[test]
    fn test_sender_ordering_and_equality() {
        let (tx1, _rx1) = unbounded_channel::<i32>();
        let (tx2, _rx2) = unbounded_channel::<i32>();

        let tx1_clone = tx1.clone();
        let weak_tx1 = tx1.downgrade();
        let weak_tx2 = tx2.downgrade();

        // Same channel senders should be equal
        assert_eq!(tx1, tx1_clone);

        // Different channels should not be equal
        assert_ne!(tx1, tx2);

        // Weak senders from same channel should be equal
        assert_eq!(weak_tx1, weak_tx1.clone());

        // Weak senders from different channels should not be equal
        assert_ne!(weak_tx1, weak_tx2);

        // Test ordering (consistent but not necessarily meaningful)
        let ordering1 = tx1.cmp(&tx2);
        let ordering2 = tx1.cmp(&tx2);
        assert_eq!(ordering1, ordering2); // Should be consistent

        // Test hashing (same senders should have same hash)
        use alloc::collections::BTreeSet;
        let mut set = BTreeSet::new();
        set.insert(tx1.clone());
        set.insert(tx1_clone.clone());
        assert_eq!(set.len(), 1); // Same sender, so only one in set

        set.insert(tx2.clone());
        assert_eq!(set.len(), 2); // Different sender, so now two in set
    }

    #[test]
    fn test_weak_sender_multiple_upgrades() {
        let (tx, mut rx) = unbounded_channel::<i32>();
        let weak_tx = tx.downgrade();

        // Multiple upgrades should work
        let upgraded1 = weak_tx.upgrade().unwrap();
        let upgraded2 = weak_tx.upgrade().unwrap();

        upgraded1.send(1).unwrap();
        upgraded2.send(2).unwrap();

        assert_eq!(rx.try_recv().unwrap(), 1);
        assert_eq!(rx.try_recv().unwrap(), 2);

        // Drop original and one upgrade
        drop(tx);
        drop(upgraded1);

        // Should still be able to upgrade and send
        let upgraded3 = weak_tx.upgrade().unwrap();
        upgraded3.send(3).unwrap();
        assert_eq!(rx.try_recv().unwrap(), 3);

        // Drop remaining senders
        drop(upgraded2);
        drop(upgraded3);

        // Now upgrade should fail
        assert!(weak_tx.upgrade().is_none());
    }

    #[test]
    fn test_sender_hash_collections() {
        use alloc::collections::BTreeSet;

        let (tx1, _rx1) = unbounded_channel::<i32>();
        let (tx2, _rx2) = unbounded_channel::<i32>();
        let tx1_clone = tx1.clone();

        // Test with HashSet equivalent (BTreeSet in no_std)
        let mut set = BTreeSet::new();

        // Insert original sender
        set.insert(tx1.clone());
        assert_eq!(set.len(), 1);

        // Insert clone of same sender - should not increase size
        set.insert(tx1_clone);
        assert_eq!(set.len(), 1);

        // Insert different sender - should increase size
        set.insert(tx2);
        assert_eq!(set.len(), 2);

        // Test weak senders
        let weak_tx1 = tx1.downgrade();
        let weak_tx1_clone = weak_tx1.clone();

        let mut weak_set = BTreeSet::new();
        weak_set.insert(weak_tx1);
        weak_set.insert(weak_tx1_clone); // Should not increase size
        assert_eq!(weak_set.len(), 1);
    }

    // === NEW TOKIO-COMPATIBLE TESTS ===

    #[test]
    fn test_len_method() {
        let (tx, rx) = unbounded_channel::<i32>();
        
        // Empty channel
        assert_eq!(rx.len(), 0);
        assert!(rx.is_empty());
        
        // Send some messages
        tx.send(1).unwrap();
        assert_eq!(rx.len(), 1);
        assert!(!rx.is_empty());
        
        tx.send(2).unwrap();
        tx.send(3).unwrap();
        assert_eq!(rx.len(), 3);
        assert!(!rx.is_empty());
    }

    /// Compare drop behavior with Tokio's implementation
    /// Tests whether unread messages are properly dropped when receiver is dropped
    #[test]
    fn test_tokio_drop_behavior_compatibility() {
        use alloc::sync::Arc;
        use core::sync::atomic::{AtomicUsize, Ordering};

        // Drop counter to track when messages are dropped
        #[derive(Debug)]
        struct DropCounter {
            id: usize,
            counter: Arc<AtomicUsize>,
        }

        impl Drop for DropCounter {
            fn drop(&mut self) {
                self.counter.fetch_add(1, Ordering::SeqCst);
            }
        }

        const NUM_MESSAGES: usize = 100; // Smaller for unit test
        
        // Test our implementation
        let our_drop_counter = Arc::new(AtomicUsize::new(0));
        {
            let (tx, mut rx) = unbounded_channel::<DropCounter>();
            
            // Send messages
            for i in 0..NUM_MESSAGES {
                let msg = DropCounter {
                    id: i,
                    counter: Arc::clone(&our_drop_counter),
                };
                tx.send(msg).unwrap();
            }
            
            // Receive only first few messages
            for _ in 0..10 {
                let _msg = rx.try_recv().unwrap();
                // Let them drop immediately
            }
            
            // Drop receiver - this should drop all remaining messages
            drop(rx);
            drop(tx);
        }
        
        let our_dropped_count = our_drop_counter.load(Ordering::SeqCst);
        
        // All messages should have been dropped
        assert_eq!(our_dropped_count, NUM_MESSAGES, 
                  "Our implementation should drop all {} messages", NUM_MESSAGES);
    }

    /// Test the exact same condition as described in Tokio docs:
    /// "If the Receiver handle is dropped, then messages can no longer be read out of the channel.
    /// In this case, all further attempts to send will result in an error. Additionally,
    /// all unread messages will be drained from the channel and dropped."
    #[test]
    fn test_tokio_exact_drop_condition() {
        use alloc::sync::Arc;
        use core::sync::atomic::{AtomicUsize, Ordering};
        use alloc::vec::Vec;

        #[derive(Debug)]
        struct DropTracker {
            id: usize,
            counter: Arc<AtomicUsize>,
        }

        impl Drop for DropTracker {
            fn drop(&mut self) {
                self.counter.fetch_add(1, Ordering::SeqCst);
            }
        }

        const NUM_MESSAGES: usize = 50; // Smaller for unit test
        let drop_counter = Arc::new(AtomicUsize::new(0));

        let (tx, mut rx) = unbounded_channel::<DropTracker>();
        let tx_clone = tx.clone();

        // Fill the channel with messages
        for i in 0..NUM_MESSAGES {
            let msg = DropTracker {
                id: i,
                counter: Arc::clone(&drop_counter),
            };
            tx.send(msg).unwrap();
        }

        // Receive some messages (but not all)
        let received_before_drop = 10;
        for _ in 0..received_before_drop {
            let _msg = rx.try_recv().unwrap();
        }

        // Test condition 1: Drop receiver while messages remain
        drop(rx);

        // Test condition 2: Further attempts to send should result in error
        let mut send_errors = 0;
        let mut failed_messages = Vec::new();
        
        // Try to send more messages after receiver is dropped  
        for i in NUM_MESSAGES..NUM_MESSAGES + 10 {
            let msg = DropTracker {
                id: i,
                counter: Arc::clone(&drop_counter),
            };
            
            match tx_clone.send(msg) {
                Ok(_) => {
                    // This shouldn't happen if receiver is properly dropped
                    panic!("Send succeeded after receiver drop - this violates Tokio behavior");
                }
                Err(send_error) => {
                    send_errors += 1;
                    failed_messages.push(send_error.0); // Extract the message from SendError
                }
            }
        }

        // Drop the senders to clean up
        drop(tx);
        drop(tx_clone);
        
        // Drop the failed messages explicitly to ensure they're counted
        drop(failed_messages);

        let final_drop_count = drop_counter.load(Ordering::SeqCst);

        // Verify the Tokio documented behavior:
        // 1. All unread messages are drained and dropped
        // 2. All received messages are dropped
        // 3. All failed send messages are dropped
        let expected_total_drops = NUM_MESSAGES + send_errors;
        assert_eq!(final_drop_count, expected_total_drops, 
                  "Expected {} drops (original {} + failed sends {}), got {}", 
                  expected_total_drops, NUM_MESSAGES, send_errors, final_drop_count);
        
        // 4. Further send attempts result in errors
        assert!(send_errors > 0, "Send attempts after receiver drop should fail");
    }

    #[test]
    fn test_tokio_api_compatibility() {
        let (tx, mut rx) = unbounded_channel::<i32>();
        
        // Test basic Tokio-like APIs
        assert!(!tx.is_closed());
        assert!(!rx.is_closed());
        assert!(rx.is_empty());
        assert_eq!(rx.len(), 0);
        
        // Test sender counts
        assert_eq!(tx.strong_count(), 1);
        assert_eq!(tx.weak_count(), 0);
        assert_eq!(rx.sender_strong_count(), 1);
        assert_eq!(rx.sender_weak_count(), 0);
        
        // Test channel identification
        assert!(tx.same_channel(&tx));
        assert_eq!(tx.id(), rx.id());
        
        // Test weak sender creation
        let _weak_tx = tx.downgrade();
        assert_eq!(tx.weak_count(), 1);
        assert_eq!(rx.sender_weak_count(), 1);
        
        // Test message sending and length tracking
        tx.send(42).unwrap();
        assert_eq!(rx.len(), 1);
        assert!(!rx.is_empty());
        
        // Test message receiving
        assert_eq!(rx.try_recv().unwrap(), 42);
        assert_eq!(rx.len(), 0);
        assert!(rx.is_empty());
    }

    /// Test drop behavior when all senders are dropped (with and without remaining messages)
    /// This complements the receiver drop tests and ensures full Tokio compatibility
    #[test]
    fn test_sender_drop_behavior_comprehensive() {
        use alloc::sync::Arc;
        use core::sync::atomic::{AtomicUsize, Ordering};

        #[derive(Debug)]
        struct DropTracker {
            id: usize,
            counter: Arc<AtomicUsize>,
        }

        impl Drop for DropTracker {
            fn drop(&mut self) {
                self.counter.fetch_add(1, Ordering::SeqCst);
            }
        }

        // Test Case 1: All senders dropped with remaining messages in channel
        {
            let drop_counter = Arc::new(AtomicUsize::new(0));
            const NUM_MESSAGES: usize = 50;
            
            let (tx, mut rx) = unbounded_channel::<DropTracker>();
            let tx2 = tx.clone();
            let tx3 = tx.clone();
            
            // Send messages from multiple senders
            for i in 0..NUM_MESSAGES {
                let msg = DropTracker {
                    id: i,
                    counter: Arc::clone(&drop_counter),
                };
                
                // Distribute sends across different senders
                match i % 3 {
                    0 => tx.send(msg).unwrap(),
                    1 => tx2.send(msg).unwrap(),
                    2 => tx3.send(msg).unwrap(),
                    _ => unreachable!(),
                }
            }
            
            // Receive only some messages
            let received_count = 15;
            for _ in 0..received_count {
                let _msg = rx.try_recv().unwrap();
            }
            
            assert_eq!(rx.len(), NUM_MESSAGES - received_count);
            assert!(!rx.is_closed());
            
            // Drop all senders - this should make channel disconnected
            drop(tx);
            drop(tx2); 
            drop(tx3);
            
            // Channel should now be closed
            assert!(rx.is_closed());
            
            // Should still be able to receive remaining messages
            let mut remaining_received = 0;
            while let Ok(_msg) = rx.try_recv() {
                remaining_received += 1;
            }
            
            assert_eq!(remaining_received, NUM_MESSAGES - received_count);
            
            // Now channel should show disconnected
            assert!(matches!(rx.try_recv(), Err(TryRecvError::Disconnected)));
            
            // Drop receiver to clean up remaining messages (if any)
            drop(rx);
            
            // All messages should be dropped
            assert_eq!(drop_counter.load(Ordering::SeqCst), NUM_MESSAGES,
                      "All messages should be dropped when senders and receiver are dropped");
        }
        
        // Test Case 2: All senders dropped with empty channel
        {
            let drop_counter = Arc::new(AtomicUsize::new(0));
            const NUM_MESSAGES: usize = 30;
            
            let (tx, mut rx) = unbounded_channel::<DropTracker>();
            let tx2 = tx.clone();
            
            // Send and immediately receive all messages
            for i in 0..NUM_MESSAGES {
                let msg = DropTracker {
                    id: i + 100, // Different ID range
                    counter: Arc::clone(&drop_counter),
                };
                
                if i % 2 == 0 {
                    tx.send(msg).unwrap();
                } else {
                    tx2.send(msg).unwrap();
                }
                
                // Immediately receive
                let _received = rx.try_recv().unwrap();
            }
            
            // Channel should be empty but not closed
            assert!(rx.is_empty());
            assert!(!rx.is_closed());
            assert!(matches!(rx.try_recv(), Err(TryRecvError::Empty)));
            
            // Drop all senders
            drop(tx);
            drop(tx2);
            
            // Channel should now be closed and empty
            assert!(rx.is_empty());
            assert!(rx.is_closed());
            assert!(matches!(rx.try_recv(), Err(TryRecvError::Disconnected)));
            
            drop(rx);
            
            // All messages should be dropped (they were all received and dropped)
            assert_eq!(drop_counter.load(Ordering::SeqCst), NUM_MESSAGES,
                      "All messages should be dropped when received");
        }
        
        // Test Case 3: Gradual sender drop with weak senders
        {
            let drop_counter = Arc::new(AtomicUsize::new(0));
            const NUM_MESSAGES: usize = 40;
            
            let (tx, mut rx) = unbounded_channel::<DropTracker>();
            let tx2 = tx.clone();
            let weak_tx = tx.downgrade();
            
            // Send some messages
            for i in 0..NUM_MESSAGES / 2 {
                let msg = DropTracker {
                    id: i + 200, // Different ID range
                    counter: Arc::clone(&drop_counter),
                };
                tx.send(msg).unwrap();
            }
            
            // Drop one strong sender
            drop(tx);
            assert!(!rx.is_closed()); // Still have tx2
            
            // Send more messages with remaining sender
            for i in NUM_MESSAGES / 2..NUM_MESSAGES {
                let msg = DropTracker {
                    id: i + 200,
                    counter: Arc::clone(&drop_counter),
                };
                tx2.send(msg).unwrap();
            }
            
            // Weak sender should still be able to upgrade
            let upgraded = weak_tx.upgrade();
            assert!(upgraded.is_some());
            drop(upgraded); // Important: drop the upgraded sender
            
            // Drop last strong sender
            drop(tx2);
            
            // Receive all messages first
            let mut received_all = 0;
            while let Ok(_msg) = rx.try_recv() {
                received_all += 1;
            }
            
            assert_eq!(received_all, NUM_MESSAGES);
            
            // Now channel should show as disconnected and closed
            assert!(matches!(rx.try_recv(), Err(TryRecvError::Disconnected)));
            assert!(rx.is_closed());
            
            // Weak sender should no longer be able to upgrade
            assert!(weak_tx.upgrade().is_none());
            
            drop(rx);
            drop(weak_tx);
            
            // All messages should be dropped
            assert_eq!(drop_counter.load(Ordering::SeqCst), NUM_MESSAGES,
                      "All messages should be dropped with gradual sender drop");
        }
    }
}