sync_fd/

channel.rs

use std::io::{self};
use std::mem::MaybeUninit;
use std::os::fd::BorrowedFd;
use std::os::unix::io::{AsRawFd, RawFd};
use std::sync::mpsc::{self, Receiver, Sender};
use std::time::{Duration, Instant};

use libc::FD_ZERO;

#[derive(Debug, PartialEq, Eq, Clone, Copy)]
pub enum SendError<T> {
    Disconnected(T),
}

#[derive(Debug, PartialEq, Eq, Clone, Copy)]
pub enum TrySendError<T> {
    Full(T),
    Disconnected(T),
}

impl<T> From<mpsc::SendError<T>> for SendError<T> {
    fn from(err: mpsc::SendError<T>) -> Self {
        SendError::Disconnected(err.0)
    }
}

#[derive(Debug, PartialEq, Eq, Clone, Copy)]
pub enum RecvError {
    Disconnected,
}

impl From<mpsc::RecvError> for RecvError {
    fn from(_: mpsc::RecvError) -> Self {
        RecvError::Disconnected
    }
}

#[derive(Debug, PartialEq, Eq, Clone, Copy)]
pub enum TryRecvError {
    Empty,
    Disconnected,
}

impl From<mpsc::TryRecvError> for TryRecvError {
    fn from(err: mpsc::TryRecvError) -> Self {
        match err {
            mpsc::TryRecvError::Empty => TryRecvError::Empty,
            mpsc::TryRecvError::Disconnected => TryRecvError::Disconnected,
        }
    }
}

pub enum RecvTimeoutError {
    Timeout,
    Disconnected,
}

impl From<mpsc::RecvTimeoutError> for RecvTimeoutError {
    fn from(err: mpsc::RecvTimeoutError) -> Self {
        match err {
            mpsc::RecvTimeoutError::Timeout => RecvTimeoutError::Timeout,
            mpsc::RecvTimeoutError::Disconnected => RecvTimeoutError::Disconnected,
        }
    }
}

/// Sender half of an MpscFd pair
pub struct MpscFdSender<T> {
    sender: Sender<T>,
    pipe_write: std::io::PipeWriter,
}

/// Receiver half of an MpscFd pair
pub struct MpscFdReceiver<T> {
    receiver: Receiver<T>,
    blocking: bool,
    pipe_read: std::io::PipeReader,
}

/// Create a new MpscFd pair (sender and receiver). This acts like a
/// `std::sync::mpsc` channel, but with a file descriptor associated with each
/// end. The file descriptor may be polled with `libc::poll` or `libc::select`
/// to wait for data to be available alongside other file descriptors.
///
/// One byte is written to the pipe for each message sent. One byte is read from
/// the pipe for each message received.
///
/// The underlying implementation currently uses mpsc channels and pipes, but
/// this may change in the future.
///
/// The underlying file descriptor is useful for the receiver
pub fn mpsc_fd_pair<T>() -> io::Result<(MpscFdSender<T>, MpscFdReceiver<T>)> {
    let (sender, receiver) = mpsc::channel();
    let (pipe_read, pipe_write) = std::io::pipe()?;

    set_nonblocking(pipe_write.as_raw_fd())?;
    set_nonblocking(pipe_read.as_raw_fd())?;

    Ok((
        MpscFdSender { sender, pipe_write },
        MpscFdReceiver {
            receiver,
            blocking: true,
            pipe_read,
        },
    ))
}

fn set_nonblocking(fd: RawFd) -> io::Result<()> {
    let flags = libc_fcntl(fd, libc::F_GETFL)?;
    let _ = libc_fcntl_int(fd, libc::F_SETFL, flags | libc::O_NONBLOCK)?;
    Ok(())
}

fn set_blocking(fd: RawFd) -> io::Result<()> {
    let flags = libc_fcntl(fd, libc::F_GETFL)?;
    let _ = libc_fcntl_int(fd, libc::F_SETFL, flags & !libc::O_NONBLOCK)?;
    Ok(())
}

fn libc_fcntl(fd: RawFd, request: std::ffi::c_int) -> io::Result<std::ffi::c_int> {
    loop {
        let result = unsafe { libc::fcntl(fd, request) };
        if result == -1 {
            let err = io::Error::last_os_error();
            if err.kind() == io::ErrorKind::Interrupted {
                continue;
            }
            break Err(err);
        }
        break Ok(result);
    }
}

fn libc_fcntl_int(fd: RawFd, request: std::ffi::c_int, arg: std::ffi::c_int) -> io::Result<()> {
    loop {
        let result = unsafe { libc::fcntl(fd, request, arg) };
        if result == -1 {
            let err = io::Error::last_os_error();
            if err.kind() == io::ErrorKind::Interrupted {
                continue;
            }
            break Err(err);
        }
        break Ok(());
    }
}

fn libc_read(fd: RawFd, buf: &mut [u8]) -> io::Result<usize> {
    loop {
        let result = unsafe { libc::read(fd, buf.as_mut_ptr() as *mut _, buf.len()) };
        if result <= -1 {
            let err = io::Error::last_os_error();
            if err.kind() == io::ErrorKind::Interrupted {
                continue;
            }
            break Err(err);
        }
        break Ok(result as usize);
    }
}

fn libc_write(fd: RawFd, buf: &[u8]) -> io::Result<usize> {
    loop {
        let result = unsafe { libc::write(fd, buf.as_ptr() as *const _, buf.len()) };
        if result <= -1 {
            let err = io::Error::last_os_error();
            if err.kind() == io::ErrorKind::Interrupted {
                continue;
            }
            break Err(err);
        }
        break Ok(result as usize);
    }
}

struct FdSet {
    fdset: MaybeUninit<libc::fd_set>,
    max: RawFd,
}

impl FdSet {
    fn new() -> Self {
        let mut fdset = MaybeUninit::uninit();
        unsafe {
            FD_ZERO(fdset.as_mut_ptr());
        }

        Self { fdset, max: 0 }
    }

    unsafe fn copy(&self, fds: *mut libc::fd_set) {
        unsafe {
            libc::memcpy(
                fds as *mut _,
                self.fdset.as_ptr() as *const _,
                std::mem::size_of::<libc::fd_set>(),
            )
        };
    }

    fn set(&mut self, fd: RawFd) {
        self.max = self.max.max(fd + 1);
        unsafe {
            libc::FD_SET(fd, self.fdset.as_mut_ptr());
        }
    }

    fn is_set(&self, fd: RawFd) -> bool {
        unsafe { libc::FD_ISSET(fd, self.fdset.as_ptr()) }
    }
}

/// `select()` returns the number of ready descriptors that are contained in the
/// descriptor sets,.  If the time limit expires, `select()` returns 0.
fn libc_select(
    read: Option<&FdSet>,
    write: Option<&FdSet>,
    error: Option<&FdSet>,
    timeout: std::time::Duration,
) -> io::Result<usize> {
    let mut readfds = read.map(|_| MaybeUninit::uninit());
    let read_ptr = readfds
        .as_mut()
        .map(|r| r.as_mut_ptr())
        .unwrap_or(std::ptr::null_mut());
    let mut writefds = write.map(|_| MaybeUninit::uninit());
    let write_ptr = writefds
        .as_mut()
        .map(|w| w.as_mut_ptr())
        .unwrap_or(std::ptr::null_mut());
    let mut errorfds = error.map(|_| MaybeUninit::uninit());
    let error_ptr = errorfds
        .as_mut()
        .map(|e| e.as_mut_ptr())
        .unwrap_or(std::ptr::null_mut());
    let max = read
        .map(|r| r.max)
        .max(write.map(|w| w.max))
        .max(error.map(|e| e.max))
        .unwrap_or(0);

    loop {
        unsafe {
            if let Some(read) = read {
                read.copy(read_ptr);
            }
            if let Some(write) = write {
                write.copy(write_ptr);
            }
            if let Some(error) = error {
                error.copy(error_ptr);
            }

            let mut timeout = libc::timeval {
                tv_sec: timeout.as_secs() as _,
                tv_usec: timeout.subsec_micros() as _,
            };

            let result = libc::select(max, read_ptr, write_ptr, error_ptr, &mut timeout);
            if result <= -1 {
                let err = io::Error::last_os_error();
                if err.kind() == io::ErrorKind::Interrupted {
                    continue;
                }
                return Err(io::Error::last_os_error());
            }
            break Ok(result as usize);
        }
    }
}

impl<T> MpscFdSender<T> {
    /// Send a message through the mpsc channel and write a byte to the pipe.
    /// Fails immediately if the pipe is full.
    pub fn try_send(&mut self, msg: T) -> Result<(), TrySendError<T>> {
        // Write a byte to the pipe using libc
        let fd = self.pipe_write.as_raw_fd();
        let buf = [0u8; 1];
        loop {
            let result = unsafe { libc::write(fd, buf.as_ptr() as *const _, 1) };
            if result == -1 {
                let err = io::Error::last_os_error();
                if err.kind() == io::ErrorKind::WouldBlock {
                    return Err(TrySendError::Full(msg));
                }
                if err.kind() == io::ErrorKind::Interrupted {
                    continue;
                }
            }
            if result != 1 {
                return Err(TrySendError::Full(msg));
            }

            // If this fails, the receiver has hung up. We don't care about the
            // pipe, it'll be closed.
            self.sender
                .send(msg)
                .map_err(|msg| TrySendError::Disconnected(msg.0))?;
            break;
        }

        Ok(())
    }

    /// Send a message through the mpsc channel and block until the message is
    /// sent.
    ///
    /// This may not be fair in the case of multiple senders.
    pub fn blocking_send(&mut self, msg: T) -> Result<(), SendError<T>> {
        let fd = self.pipe_write.as_raw_fd();
        let mut writefds = FdSet::new();
        let mut errorfds = FdSet::new();

        writefds.set(fd);
        errorfds.set(fd);

        loop {
            // Wait for the pipe to be writable (or error)
            match libc_select(
                None,
                Some(&writefds),
                Some(&errorfds),
                std::time::Duration::from_secs(60),
            ) {
                Ok(0) => continue,
                Ok(_) => {}
                Err(_) => return Err(SendError::Disconnected(msg)),
            };

            // We only have one FD so it's either writable or errored.

            // Write a byte to the pipe using libc
            let buf = [0u8; 1];
            match libc_write(fd, &buf) {
                Ok(1) => {}
                Ok(_) => unreachable!("Should only write 1 byte"),
                Err(e) if e.kind() == io::ErrorKind::WouldBlock => continue,
                Err(_) => return Err(SendError::Disconnected(msg)),
            };

            self.sender.send(msg)?;
            break Ok(());
        }
    }

    pub fn try_clone(&self) -> io::Result<Self> {
        Ok(MpscFdSender {
            sender: self.sender.clone(),
            pipe_write: self.pipe_write.try_clone()?,
        })
    }
}

impl<T> MpscFdReceiver<T> {
    /// Receive a message from the mpsc channel and read a byte from the pipe.
    pub fn recv(&mut self) -> Result<T, RecvError> {
        let msg = self.receiver.recv()?;

        // If we got a byte, or the pipe is closed, return this message
        if !matches!(
            self.read_byte_nonblocking()
                .map_err(|_| RecvError::Disconnected)?,
            ReadByteResult::WouldBlock
        ) {
            return Ok(msg);
        }

        loop {
            // Keep waiting for the pipe to be readable or failed
            self.select_timeout(Duration::from_secs(60 * 60))
                .map_err(|_| RecvError::Disconnected)?;
            if matches!(
                self.read_byte_nonblocking()
                    .map_err(|_| RecvError::Disconnected)?,
                ReadByteResult::WouldBlock
            ) {
                continue;
            }

            break Ok(msg);
        }
    }

    /// Try to receive a message without blocking.
    ///
    /// Readability of the fd is a hint that this will likely return a value, however
    /// this may return Empty and leave the fd readable in certain cases.
    pub fn try_recv(&mut self) -> Result<T, TryRecvError> {
        // If we got a byte, we block on the channel
        if matches!(self.read_byte_nonblocking(), Ok(ReadByteResult::Success)) {
            // This may block slightly if the pipe was written but the task was
            // unscheduled.
            return self.receiver.recv().map_err(|_| TryRecvError::Disconnected);
        }

        // Otherwise it's Eof or WouldBlock, so delegate to try_recv on the channel
        Ok(self.receiver.try_recv()?)
    }

    /// Receive with timeout. Note that this may block slightly more than `timeout`
    /// if the pipe was written but the channel was not.
    ///
    /// Readability of the fd is a hint that this will likely return a value, however
    /// this may return Empty and leave the fd readable in certain cases.
    pub fn recv_timeout(&mut self, mut timeout: Duration) -> Result<T, RecvTimeoutError> {
        loop {
            let start = Instant::now();
            if !self
                .select_timeout(timeout)
                .map_err(|_| RecvTimeoutError::Disconnected)?
            {
                return Err(RecvTimeoutError::Timeout);
            }

            match self.read_byte_nonblocking() {
                Ok(ReadByteResult::Success) => {
                    // This may block slightly more than `timeout` if the pipe was
                    // written but the channel was not due to task scheduling.
                    break self
                        .receiver
                        .recv()
                        .map_err(|_| RecvTimeoutError::Disconnected);
                }
                Ok(ReadByteResult::Eof) => {
                    // The pipe was closed, so delegate to the channel's recv_timeout
                    break Ok(self.receiver.recv_timeout(
                        timeout
                            .checked_sub(start.elapsed())
                            .unwrap_or(Duration::from_secs(0)),
                    )?);
                }
                Ok(ReadByteResult::WouldBlock) => {
                    timeout = timeout
                        .checked_sub(start.elapsed())
                        .unwrap_or(Duration::from_secs(0));
                    if timeout == Duration::from_secs(0) {
                        return Err(RecvTimeoutError::Timeout);
                    }
                }
                Err(_) => return Err(RecvTimeoutError::Disconnected),
            }
        }
    }

    fn select_timeout(&mut self, timeout: Duration) -> Result<bool, TryRecvError> {
        let mut readfds = FdSet::new();
        let mut errorfds = FdSet::new();

        let fd = self.pipe_read.as_raw_fd();
        readfds.set(fd);
        errorfds.set(fd);

        match libc_select(Some(&readfds), None, Some(&errorfds), timeout) {
            Ok(0) => Ok(false),
            Ok(_) => Ok(true),
            Err(_) => Err(TryRecvError::Disconnected),
        }
    }

    fn read_byte_nonblocking(&mut self) -> io::Result<ReadByteResult> {
        // Try to read a byte from the pipe
        let fd = self.pipe_read.as_raw_fd();
        let mut buf = [0u8; 1];
        Ok(match libc_read(fd, &mut buf) {
            // If the pipe is closed, we'll continue reading out of the channel
            Ok(1) => ReadByteResult::Success,
            Ok(0) => ReadByteResult::Eof,
            Err(e) if e.kind() == io::ErrorKind::BrokenPipe => ReadByteResult::Eof,

            Ok(_) => unreachable!("Should only read at most 1 byte"),
            Err(e) if e.kind() == io::ErrorKind::WouldBlock => ReadByteResult::WouldBlock,
            Err(e) => return Err(e),
        })
    }
}

enum ReadByteResult {
    Success,
    Eof,
    WouldBlock,
}

#[cfg(unix)]
impl<T> AsRawFd for MpscFdReceiver<T> {
    fn as_raw_fd(&self) -> RawFd {
        self.pipe_read.as_raw_fd()
    }
}

#[cfg(unix)]
impl<T> AsRawFd for MpscFdSender<T> {
    fn as_raw_fd(&self) -> RawFd {
        self.pipe_write.as_raw_fd()
    }
}

#[cfg(unix)]
impl<'a, T> Into<BorrowedFd<'a>> for &'a MpscFdReceiver<T> {
    fn into(self) -> BorrowedFd<'a> {
        unsafe { BorrowedFd::borrow_raw(self.pipe_read.as_raw_fd() as _) }
    }
}

#[cfg(unix)]
impl<'a, T> Into<BorrowedFd<'a>> for &'a MpscFdSender<T> {
    fn into(self) -> BorrowedFd<'a> {
        unsafe { BorrowedFd::borrow_raw(self.pipe_write.as_raw_fd() as _) }
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use std::sync::{Arc, Mutex};
    use std::thread;
    use std::time::{Duration, Instant};

    #[test]
    fn test_mpsc_fd_pair() {
        let (mut sender, mut receiver) = mpsc_fd_pair::<u32>().unwrap();

        sender.blocking_send(1).unwrap();
        assert_eq!(receiver.recv().unwrap(), 1);

        sender.blocking_send(2).unwrap();
        assert_eq!(receiver.recv().unwrap(), 2);

        sender.blocking_send(3).unwrap();
        assert_eq!(receiver.recv().unwrap(), 3);

        drop(sender);
        assert_eq!(receiver.recv().unwrap_err(), RecvError::Disconnected);
    }

    /// Torture test 1: Rapid switching between blocking and non-blocking receive
    /// with multiple senders and receivers, testing edge cases around pipe buffer limits
    #[test]
    fn torture_test_rapid_mode_switching() {
        let (mut sender1, mut receiver) = mpsc_fd_pair::<u32>().unwrap();
        let mut sender2 = sender1.try_clone().unwrap();
        let mut sender3 = sender1.try_clone().unwrap();

        let start_time = Instant::now();

        // Spawn receiver thread that rapidly switches between modes
        let receiver_handle = thread::spawn(move || {
            let mut local_received = Vec::new();
            let mut mode_switch_count = 0;

            while start_time.elapsed() < Duration::from_secs(5) {
                // Switch modes every few iterations
                if mode_switch_count % 10 == 0 {
                    // Non-blocking mode
                    loop {
                        match receiver.try_recv() {
                            Ok(msg) => local_received.push(msg),
                            Err(TryRecvError::Empty) => break,
                            Err(TryRecvError::Disconnected) => return local_received,
                        }
                    }
                } else if mode_switch_count % 10 == 5 {
                    // Timeout mode with very short timeout
                    match receiver.recv_timeout(Duration::from_millis(1)) {
                        Ok(msg) => local_received.push(msg),
                        Err(RecvTimeoutError::Timeout) => {}
                        Err(RecvTimeoutError::Disconnected) => return local_received,
                    }
                } else {
                    // Blocking mode
                    match receiver.recv() {
                        Ok(msg) => local_received.push(msg),
                        Err(RecvError::Disconnected) => return local_received,
                    }
                }
                mode_switch_count += 1;
            }
            local_received
        });

        // Spawn multiple sender threads with different patterns
        let sender1_handle = thread::spawn(move || {
            for i in 0..1000 {
                if let Err(TrySendError::Full(_)) = sender1.try_send(i) {
                    // If pipe is full, use blocking send
                    sender1.blocking_send(i).unwrap();
                }
                thread::sleep(Duration::from_micros(100));
            }
        });

        let sender2_handle = thread::spawn(move || {
            for i in 1000..2000 {
                sender2.blocking_send(i).unwrap();
                thread::sleep(Duration::from_micros(50));
            }
        });

        let sender3_handle = thread::spawn(move || {
            for i in 2000..3000 {
                // Rapid fire with try_send
                for j in 0..10 {
                    let msg = i * 10 + j;
                    if msg >= 3000 {
                        break; // Don't send messages outside the expected range
                    }
                    if let Err(TrySendError::Full(_)) = sender3.try_send(msg) {
                        break;
                    }
                }
                thread::sleep(Duration::from_millis(1));
            }
        });

        // Wait for all threads to complete
        sender1_handle.join().unwrap();
        sender2_handle.join().unwrap();
        sender3_handle.join().unwrap();

        let received_messages = receiver_handle.join().unwrap();

        // Verify we received a substantial number of messages
        assert!(
            received_messages.len() > 100,
            "Should receive many messages, got {}",
            received_messages.len()
        );

        // Verify no duplicates and all messages are in expected ranges
        let mut sorted = received_messages.clone();
        sorted.sort();
        sorted.dedup();
        assert_eq!(
            sorted.len(),
            received_messages.len(),
            "No duplicates should exist"
        );

        // Check that messages are from expected ranges
        for &msg in &received_messages {
            assert!(msg < 3000, "Message {} is out of expected range", msg);
        }
    }

    /// Torture test 2: Stress test with pipe buffer exhaustion and recovery
    /// Tests the channel under extreme pressure with rapid send/receive cycles
    #[test]
    fn torture_test_pipe_buffer_stress() {
        let (mut sender, mut receiver) = mpsc_fd_pair::<String>().unwrap();

        let iterations = 1000; // Reduced for more reliable testing

        // Track sent messages to ensure all are sent
        let sent_count = Arc::new(Mutex::new(0));
        let sent_count_clone = sent_count.clone();
        let burst_count = Arc::new(Mutex::new(0));
        let burst_count_clone = burst_count.clone();

        // Receiver thread that switches between all three receive modes
        let receiver_handle = thread::spawn(move || {
            let mut local_received = Vec::new();
            let mut consecutive_empty = 0;

            // Keep receiving until sender disconnects
            loop {
                // Cycle through different receive modes
                match local_received.len() % 3 {
                    0 => {
                        // Non-blocking mode
                        match receiver.try_recv() {
                            Ok(msg) => {
                                local_received.push(msg);
                                consecutive_empty = 0;
                            }
                            Err(TryRecvError::Empty) => {
                                consecutive_empty += 1;
                                if consecutive_empty > 100 {
                                    // Switch to blocking mode if too many empty reads
                                    match receiver.recv() {
                                        Ok(msg) => {
                                            local_received.push(msg);
                                            consecutive_empty = 0;
                                        }
                                        Err(RecvError::Disconnected) => break,
                                    }
                                }
                            }
                            Err(TryRecvError::Disconnected) => break,
                        }
                    }
                    1 => {
                        // Timeout mode with varying timeouts
                        let timeout = if local_received.len() % 100 == 0 {
                            Duration::from_millis(10)
                        } else {
                            Duration::from_micros(100)
                        };
                        match receiver.recv_timeout(timeout) {
                            Ok(msg) => {
                                local_received.push(msg);
                                consecutive_empty = 0;
                            }
                            Err(RecvTimeoutError::Timeout) => {
                                consecutive_empty += 1;
                            }
                            Err(RecvTimeoutError::Disconnected) => break,
                        }
                    }
                    2 => {
                        // Blocking mode
                        match receiver.recv() {
                            Ok(msg) => {
                                local_received.push(msg);
                                consecutive_empty = 0;
                            }
                            Err(RecvError::Disconnected) => break,
                        }
                    }
                    _ => unreachable!(),
                }
            }
            local_received
        });

        // Sender thread that ensures all messages are sent
        let sender_handle = thread::spawn(move || {
            for i in 0..iterations {
                let msg = format!("message-{}", i);

                // Always use blocking send to ensure message is sent
                match sender.blocking_send(msg) {
                    Ok(_) => {
                        *sent_count_clone.lock().unwrap() += 1;
                    }
                    Err(SendError::Disconnected(_)) => {
                        break; // Receiver disconnected, stop sending
                    }
                }

                // Occasionally create small bursts to stress the buffer
                if i % 100 == 0 {
                    for j in 0..5 {
                        // Reduced burst size
                        let burst_msg = format!("burst-{}-{}", i, j);
                        if let Err(TrySendError::Full(_)) = sender.try_send(burst_msg) {
                            // If pipe is full, skip this burst
                            break;
                        } else {
                            *burst_count_clone.lock().unwrap() += 1;
                        }
                    }
                }
            }
        });

        sender_handle.join().unwrap();
        let received_messages = receiver_handle.join().unwrap();
        let final_sent_count = *sent_count.lock().unwrap();
        let final_burst_count = *burst_count.lock().unwrap();
        let total_sent = final_sent_count + final_burst_count;

        // Verify all messages were sent
        assert_eq!(
            final_sent_count, iterations,
            "All regular messages should be sent, sent: {}",
            final_sent_count
        );

        // Verify we received all sent messages
        assert_eq!(
            received_messages.len(),
            total_sent,
            "Should receive all sent messages, sent: {}, received: {}",
            total_sent,
            received_messages.len()
        );

        // Verify message integrity - check that we have the expected message types
        let mut message_count = 0;
        for msg in received_messages.iter() {
            if msg.starts_with("message-") {
                message_count += 1;
            }
        }

        // We should have received all regular messages
        assert_eq!(
            message_count, iterations,
            "Should receive all regular messages, got {}",
            message_count
        );
    }

    /// Torture test 3: Concurrent access with multiple senders and single receiver
    /// Tests race conditions and edge cases in multi-threaded scenarios
    #[test]
    fn torture_test_concurrent_access() {
        let (sender, mut receiver) = mpsc_fd_pair::<u64>().unwrap();
        let sender = Arc::new(Mutex::new(sender));

        let num_senders = 5;
        let messages_per_sender = 100; // Reduced for more reliable testing
        let total_messages = num_senders * messages_per_sender;

        // Track sent messages to ensure all are sent
        let sent_count = Arc::new(Mutex::new(0));
        let sent_count_clone = sent_count.clone();

        // Spawn multiple sender threads
        let sender_handles: Vec<_> = (0..num_senders)
            .map(|sender_id| {
                let sender = sender.clone();
                let local_sent_count = sent_count_clone.clone();

                thread::spawn(move || {
                    let mut local_sender = sender.lock().unwrap().try_clone().unwrap();

                    for i in 0..messages_per_sender {
                        let msg = (sender_id as u64) * 10000 + i;

                        // Always use blocking send to ensure message is sent
                        match local_sender.blocking_send(msg) {
                            Ok(_) => {
                                *local_sent_count.lock().unwrap() += 1;
                            }
                            Err(SendError::Disconnected(_)) => {
                                break; // Receiver disconnected, stop sending
                            }
                        }

                        // Small delay to create interleaving
                        thread::sleep(Duration::from_micros(10));
                    }
                })
            })
            .collect();

        // Single receiver thread that switches between different receive modes
        let receiver_handle = thread::spawn(move || {
            let mut all_received = Vec::new();
            let mut mode_counter = 0;
            let start_time = Instant::now();

            // Keep receiving until all senders disconnect or timeout
            while start_time.elapsed() < Duration::from_secs(10) {
                // Cycle through different receive modes
                match mode_counter % 4 {
                    0 => {
                        // Non-blocking mode
                        match receiver.try_recv() {
                            Ok(msg) => all_received.push(msg),
                            Err(TryRecvError::Empty) => {
                                thread::sleep(Duration::from_micros(100));
                            }
                            Err(TryRecvError::Disconnected) => break,
                        }
                    }
                    1 => {
                        // Short timeout mode
                        match receiver.recv_timeout(Duration::from_millis(1)) {
                            Ok(msg) => all_received.push(msg),
                            Err(RecvTimeoutError::Timeout) => {}
                            Err(RecvTimeoutError::Disconnected) => break,
                        }
                    }
                    2 => {
                        // Longer timeout mode
                        match receiver.recv_timeout(Duration::from_millis(10)) {
                            Ok(msg) => all_received.push(msg),
                            Err(RecvTimeoutError::Timeout) => {}
                            Err(RecvTimeoutError::Disconnected) => break,
                        }
                    }
                    3 => {
                        // Blocking mode with short timeout
                        match receiver.recv_timeout(Duration::from_millis(100)) {
                            Ok(msg) => all_received.push(msg),
                            Err(RecvTimeoutError::Timeout) => {}
                            Err(RecvTimeoutError::Disconnected) => break,
                        }
                    }
                    _ => unreachable!(),
                }
                mode_counter += 1;
            }

            all_received
        });

        // Wait for all threads to complete
        for handle in sender_handles {
            handle.join().unwrap();
        }

        let all_received = receiver_handle.join().unwrap();
        let final_sent_count = *sent_count.lock().unwrap();

        // Verify all messages were sent
        assert_eq!(
            final_sent_count, total_messages as usize,
            "All messages should be sent, sent: {}",
            final_sent_count
        );

        // Verify we received all sent messages
        assert_eq!(
            all_received.len(),
            final_sent_count,
            "Should receive all sent messages, sent: {}, received: {}",
            final_sent_count,
            all_received.len()
        );

        // Verify message ranges
        for &msg in all_received.iter() {
            let sender_id = msg / 10000;
            let msg_id = msg % 10000;
            assert!(
                sender_id < num_senders as u64,
                "Invalid sender ID: {}",
                sender_id
            );
            assert!(
                msg_id < messages_per_sender as u64,
                "Invalid message ID: {}",
                msg_id
            );
        }

        // Check for duplicates
        let mut sorted = all_received.clone();
        sorted.sort();
        sorted.dedup();
        assert_eq!(
            sorted.len(),
            all_received.len(),
            "No duplicates should exist"
        );
    }
}