calloop/sources/

channel.rs

//! An MPSC channel whose receiving end is an event source
//!
//! Create a channel using [`channel()`](channel), which returns a
//! [`Sender`] that can be cloned and sent accross threads if `T: Send`,
//! and a [`Channel`] that can be inserted into an [`EventLoop`](crate::EventLoop).
//! It will generate one event per message.
//!
//! A synchronous version of the channel is provided by [`sync_channel`], in which
//! the [`SyncSender`] will block when the channel is full.

use std::cmp;
use std::fmt;
use std::ops;
use std::sync::{mpsc, Arc};

use crate::{EventSource, Poll, PostAction, Readiness, Token, TokenFactory};

use super::ping::{make_ping, Ping, PingError, PingSource};

const MAX_EVENTS_CHECK: usize = 1024;

/// The events generated by the channel event source
#[derive(Debug)]
pub enum Event<T> {
    /// A message was received and is bundled here
    Msg(T),
    /// The channel was closed
    ///
    /// This means all the `Sender`s associated with this channel
    /// have been dropped, no more messages will ever be received.
    Closed,
}

#[derive(Debug)]
struct PingOnDrop(Ping);

impl ops::Deref for PingOnDrop {
    type Target = Ping;

    fn deref(&self) -> &Ping {
        &self.0
    }
}

impl Drop for PingOnDrop {
    fn drop(&mut self) {
        self.0.ping();
    }
}

/// The sender end of a channel
///
/// It can be cloned and sent accross threads (if `T` is).
#[derive(Debug)]
pub struct Sender<T> {
    sender: mpsc::Sender<T>,
    // Dropped after `sender` so receiver is guaranteed to get `Disconnected`
    // after ping.
    ping: PingOnDrop,
}

impl<T> Clone for Sender<T> {
    #[cfg_attr(feature = "nightly_coverage", coverage(off))]
    fn clone(&self) -> Sender<T> {
        Sender {
            sender: self.sender.clone(),
            ping: PingOnDrop(self.ping.clone()),
        }
    }
}

impl<T> Sender<T> {
    /// Send a message to the channel
    ///
    /// This will wake the event loop and deliver an `Event::Msg` to
    /// it containing the provided value.
    pub fn send(&self, t: T) -> Result<(), mpsc::SendError<T>> {
        self.sender.send(t).map(|()| self.ping.ping())
    }
}

/// The sender end of a synchronous channel
///
/// It can be cloned and sent accross threads (if `T` is).
#[derive(Debug)]
pub struct SyncSender<T> {
    sender: mpsc::SyncSender<T>,
    // Dropped after `sender` so receiver is guaranteed to get `Disconnected`
    // after ping.
    ping: Arc<PingOnDrop>,
}

impl<T> Clone for SyncSender<T> {
    #[cfg_attr(feature = "nightly_coverage", coverage(off))]
    fn clone(&self) -> SyncSender<T> {
        SyncSender {
            sender: self.sender.clone(),
            ping: self.ping.clone(),
        }
    }
}

impl<T> SyncSender<T> {
    /// Send a message to the synchronous channel
    ///
    /// This will wake the event loop and deliver an `Event::Msg` to
    /// it containing the provided value. If the channel is full, this
    /// function will block until the event loop empties it and it can
    /// deliver the message.
    ///
    /// Due to the blocking behavior, this method should not be used on the
    /// same thread as the one running the event loop, as it could cause deadlocks.
    pub fn send(&self, t: T) -> Result<(), mpsc::SendError<T>> {
        let ret = self.try_send(t);
        match ret {
            Ok(()) => Ok(()),
            Err(mpsc::TrySendError::Full(t)) => self.sender.send(t).map(|()| self.ping.ping()),
            Err(mpsc::TrySendError::Disconnected(t)) => Err(mpsc::SendError(t)),
        }
    }

    /// Send a message to the synchronous channel
    ///
    /// This will wake the event loop and deliver an `Event::Msg` to
    /// it containing the provided value. If the channel is full, this
    /// function will return an error, but the event loop will still be
    /// signaled for readiness.
    pub fn try_send(&self, t: T) -> Result<(), mpsc::TrySendError<T>> {
        let ret = self.sender.try_send(t);
        if let Ok(()) | Err(mpsc::TrySendError::Full(_)) = ret {
            self.ping.ping();
        }
        ret
    }
}

/// The receiving end of the channel
///
/// This is the event source to be inserted into your `EventLoop`.
#[derive(Debug)]
pub struct Channel<T> {
    receiver: mpsc::Receiver<T>,
    source: PingSource,
    ping: Ping,
    capacity: usize,
}

// This impl is safe because the Channel is only able to move around threads
// when it is not inserted into an event loop. (Otherwise it is stuck into
// a Source<_> and the internals of calloop, which are not Send).
// At this point, the Arc<Receiver> has a count of 1, and it is obviously
// safe to Send between threads.
unsafe impl<T: Send> Send for Channel<T> {}

impl<T> Channel<T> {
    /// Proxy for [`mpsc::Receiver::recv`] to manually poll events.
    ///
    /// *Note*: Normally you would want to use the `Channel` by inserting
    /// it into an event loop instead. Use this for example to immediately
    /// dispatch pending events after creation.
    pub fn recv(&self) -> Result<T, mpsc::RecvError> {
        self.receiver.recv()
    }

    /// Proxy for [`mpsc::Receiver::try_recv`] to manually poll events.
    ///
    /// *Note*: Normally you would want to use the `Channel` by inserting
    /// it into an event loop instead. Use this for example to immediately
    /// dispatch pending events after creation.
    pub fn try_recv(&self) -> Result<T, mpsc::TryRecvError> {
        self.receiver.try_recv()
    }
}

/// Create a new asynchronous channel
pub fn channel<T>() -> (Sender<T>, Channel<T>) {
    let (sender, receiver) = mpsc::channel();
    let (ping, source) = make_ping().expect("Failed to create a Ping.");
    (
        Sender {
            sender,
            ping: PingOnDrop(ping.clone()),
        },
        Channel {
            receiver,
            ping,
            source,
            capacity: usize::MAX,
        },
    )
}

/// Create a new synchronous, bounded channel
pub fn sync_channel<T>(bound: usize) -> (SyncSender<T>, Channel<T>) {
    let (sender, receiver) = mpsc::sync_channel(bound);
    let (ping, source) = make_ping().expect("Failed to create a Ping.");
    (
        SyncSender {
            sender,
            ping: Arc::new(PingOnDrop(ping.clone())),
        },
        Channel {
            receiver,
            source,
            ping,
            capacity: bound,
        },
    )
}

impl<T> EventSource for Channel<T> {
    type Event = Event<T>;
    type Metadata = ();
    type Ret = ();
    type Error = ChannelError;

    fn process_events<C>(
        &mut self,
        readiness: Readiness,
        token: Token,
        mut callback: C,
    ) -> Result<PostAction, Self::Error>
    where
        C: FnMut(Self::Event, &mut Self::Metadata) -> Self::Ret,
    {
        let receiver = &self.receiver;
        let capacity = self.capacity;
        let mut clear_readiness = false;
        let mut disconnected = false;

        let action = self
            .source
            .process_events(readiness, token, |(), &mut ()| {
                // Limit the number of elements we process at a time to the channel's capacity, or 1024.
                let max = cmp::min(capacity.saturating_add(1), MAX_EVENTS_CHECK);
                for _ in 0..max {
                    match receiver.try_recv() {
                        Ok(val) => callback(Event::Msg(val), &mut ()),
                        Err(mpsc::TryRecvError::Empty) => {
                            clear_readiness = true;
                            break;
                        }
                        Err(mpsc::TryRecvError::Disconnected) => {
                            callback(Event::Closed, &mut ());
                            disconnected = true;
                            break;
                        }
                    }
                }
            })
            .map_err(ChannelError)?;

        if disconnected {
            Ok(PostAction::Remove)
        } else if clear_readiness {
            Ok(action)
        } else {
            // Re-notify the ping source so we can try again.
            self.ping.ping();
            Ok(PostAction::Continue)
        }
    }

    fn register(&mut self, poll: &mut Poll, token_factory: &mut TokenFactory) -> crate::Result<()> {
        self.source.register(poll, token_factory)
    }

    fn reregister(
        &mut self,
        poll: &mut Poll,
        token_factory: &mut TokenFactory,
    ) -> crate::Result<()> {
        self.source.reregister(poll, token_factory)
    }

    fn unregister(&mut self, poll: &mut Poll) -> crate::Result<()> {
        self.source.unregister(poll)
    }
}

/// An error arising from processing events for a channel.
#[derive(Debug)]
pub struct ChannelError(PingError);

impl fmt::Display for ChannelError {
    #[cfg_attr(feature = "nightly_coverage", coverage(off))]
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        fmt::Display::fmt(&self.0, f)
    }
}

impl std::error::Error for ChannelError {
    #[cfg_attr(feature = "nightly_coverage", coverage(off))]
    fn source(&self) -> Option<&(dyn std::error::Error + 'static)> {
        Some(&self.0)
    }
}

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn basic_channel() {
        let mut event_loop = crate::EventLoop::try_new().unwrap();

        let handle = event_loop.handle();

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

        // (got_msg, got_closed)
        let mut got = (false, false);

        let _channel_token = handle
            .insert_source(rx, move |evt, &mut (), got: &mut (bool, bool)| match evt {
                Event::Msg(()) => {
                    got.0 = true;
                }
                Event::Closed => {
                    got.1 = true;
                }
            })
            .unwrap();

        // nothing is sent, nothing is received
        event_loop
            .dispatch(Some(::std::time::Duration::ZERO), &mut got)
            .unwrap();

        assert_eq!(got, (false, false));

        // a message is send
        tx.send(()).unwrap();
        event_loop
            .dispatch(Some(::std::time::Duration::ZERO), &mut got)
            .unwrap();

        assert_eq!(got, (true, false));

        // the sender is dropped
        ::std::mem::drop(tx);
        event_loop
            .dispatch(Some(::std::time::Duration::ZERO), &mut got)
            .unwrap();

        assert_eq!(got, (true, true));
    }

    #[test]
    fn basic_sync_channel() {
        let mut event_loop = crate::EventLoop::try_new().unwrap();

        let handle = event_loop.handle();

        let (tx, rx) = sync_channel::<()>(2);

        let mut received = (0, false);

        let _channel_token = handle
            .insert_source(
                rx,
                move |evt, &mut (), received: &mut (u32, bool)| match evt {
                    Event::Msg(()) => {
                        received.0 += 1;
                    }
                    Event::Closed => {
                        received.1 = true;
                    }
                },
            )
            .unwrap();

        // nothing is sent, nothing is received
        event_loop
            .dispatch(Some(::std::time::Duration::ZERO), &mut received)
            .unwrap();

        assert_eq!(received.0, 0);
        assert!(!received.1);

        // fill the channel
        tx.send(()).unwrap();
        tx.send(()).unwrap();
        assert!(tx.try_send(()).is_err());

        // empty it
        event_loop
            .dispatch(Some(::std::time::Duration::ZERO), &mut received)
            .unwrap();

        assert_eq!(received.0, 2);
        assert!(!received.1);

        // send a final message and drop the sender
        tx.send(()).unwrap();
        std::mem::drop(tx);

        // final read of the channel
        event_loop
            .dispatch(Some(::std::time::Duration::ZERO), &mut received)
            .unwrap();

        assert_eq!(received.0, 3);
        assert!(received.1);
    }

    #[test]
    fn test_more_than_1024() {
        let mut event_loop = crate::EventLoop::try_new().unwrap();
        let handle = event_loop.handle();

        let (tx, rx) = channel::<()>();
        let mut received = (0u32, false);

        handle
            .insert_source(
                rx,
                move |evt, &mut (), received: &mut (u32, bool)| match evt {
                    Event::Msg(()) => received.0 += 1,
                    Event::Closed => received.1 = true,
                },
            )
            .unwrap();

        event_loop
            .dispatch(Some(std::time::Duration::ZERO), &mut received)
            .unwrap();

        assert_eq!(received.0, 0);
        assert!(!received.1);

        // Send 1025 elements into the channel.
        for _ in 0..MAX_EVENTS_CHECK + 1 {
            tx.send(()).unwrap();
        }

        event_loop
            .dispatch(Some(std::time::Duration::ZERO), &mut received)
            .unwrap();

        assert_eq!(received.0, MAX_EVENTS_CHECK as u32);
        assert!(!received.1);

        event_loop
            .dispatch(Some(std::time::Duration::ZERO), &mut received)
            .unwrap();

        assert_eq!(received.0, (MAX_EVENTS_CHECK + 1) as u32);
        assert!(!received.1);
    }
}