From 99d19c7dcf077cdd06113bf013d6a458289b1ecb Mon Sep 17 00:00:00 2001 From: huntc Date: Sun, 12 Jun 2022 15:16:56 +1000 Subject: Rename channel to mpmc I've renamed the channel module for the MPMC as mpmc. There was a previous debate about this, but I feel that the strategy here avoids importing `channel::channel`. The change leaves `signal::Signal`, but I think that's ok. It is all a bit subjective of course. The bottom line for me is that I really like the term mpmc - it means something to me and aligns with broader naming e.g. in Tokio. --- embassy/src/channel/channel.rs | 637 ------------------------ embassy/src/channel/mod.rs | 4 +- embassy/src/channel/mpmc.rs | 637 ++++++++++++++++++++++++ embassy/src/channel/signal.rs | 2 +- examples/nrf/src/bin/channel.rs | 2 +- examples/nrf/src/bin/channel_sender_receiver.rs | 2 +- examples/nrf/src/bin/uart_split.rs | 2 +- examples/nrf/src/bin/usb_ethernet.rs | 2 +- examples/nrf/src/bin/usb_hid_keyboard.rs | 2 +- examples/stm32f3/src/bin/button_events.rs | 2 +- examples/stm32h7/src/bin/signal.rs | 2 +- examples/stm32h7/src/bin/usart_split.rs | 2 +- examples/stm32l5/src/bin/usb_ethernet.rs | 2 +- 13 files changed, 648 insertions(+), 650 deletions(-) delete mode 100644 embassy/src/channel/channel.rs create mode 100644 embassy/src/channel/mpmc.rs diff --git a/embassy/src/channel/channel.rs b/embassy/src/channel/channel.rs deleted file mode 100644 index d74f6365d..000000000 --- a/embassy/src/channel/channel.rs +++ /dev/null @@ -1,637 +0,0 @@ -//! A queue for sending values between asynchronous tasks. -//! -//! It can be used concurrently by multiple producers (senders) and multiple -//! consumers (receivers), i.e. it is an "MPMC channel". -//! -//! Receivers are competing for messages. So a message that is received by -//! one receiver is not received by any other. -//! -//! This queue takes a Mutex type so that various -//! targets can be attained. For example, a ThreadModeMutex can be used -//! for single-core Cortex-M targets where messages are only passed -//! between tasks running in thread mode. Similarly, a CriticalSectionMutex -//! can also be used for single-core targets where messages are to be -//! passed from exception mode e.g. out of an interrupt handler. -//! -//! This module provides a bounded channel that has a limit on the number of -//! messages that it can store, and if this limit is reached, trying to send -//! another message will result in an error being returned. -//! - -use core::cell::RefCell; -use core::pin::Pin; -use core::task::Context; -use core::task::Poll; - -use futures::Future; -use heapless::Deque; - -use crate::blocking_mutex::raw::RawMutex; -use crate::blocking_mutex::Mutex; -use crate::waitqueue::WakerRegistration; - -/// Send-only access to a [`Channel`]. -#[derive(Copy)] -pub struct Sender<'ch, M, T, const N: usize> -where - M: RawMutex, -{ - channel: &'ch Channel, -} - -impl<'ch, M, T, const N: usize> Clone for Sender<'ch, M, T, N> -where - M: RawMutex, -{ - fn clone(&self) -> Self { - Sender { - channel: self.channel, - } - } -} - -impl<'ch, M, T, const N: usize> Sender<'ch, M, T, N> -where - M: RawMutex, -{ - /// Sends a value. - /// - /// See [`Channel::send()`] - pub fn send(&self, message: T) -> SendFuture<'ch, M, T, N> { - self.channel.send(message) - } - - /// Attempt to immediately send a message. - /// - /// See [`Channel::send()`] - pub fn try_send(&self, message: T) -> Result<(), TrySendError> { - self.channel.try_send(message) - } -} - -/// Send-only access to a [`Channel`] without knowing channel size. -#[derive(Copy)] -pub struct DynamicSender<'ch, T> { - channel: &'ch dyn DynamicChannel, -} - -impl<'ch, T> Clone for DynamicSender<'ch, T> { - fn clone(&self) -> Self { - DynamicSender { - channel: self.channel, - } - } -} - -impl<'ch, M, T, const N: usize> From> for DynamicSender<'ch, T> -where - M: RawMutex, -{ - fn from(s: Sender<'ch, M, T, N>) -> Self { - Self { channel: s.channel } - } -} - -impl<'ch, T> DynamicSender<'ch, T> { - /// Sends a value. - /// - /// See [`Channel::send()`] - pub fn send(&self, message: T) -> DynamicSendFuture<'ch, T> { - DynamicSendFuture { - channel: self.channel, - message: Some(message), - } - } - - /// Attempt to immediately send a message. - /// - /// See [`Channel::send()`] - pub fn try_send(&self, message: T) -> Result<(), TrySendError> { - self.channel.try_send_with_context(message, None) - } -} - -/// Receive-only access to a [`Channel`]. -#[derive(Copy)] -pub struct Receiver<'ch, M, T, const N: usize> -where - M: RawMutex, -{ - channel: &'ch Channel, -} - -impl<'ch, M, T, const N: usize> Clone for Receiver<'ch, M, T, N> -where - M: RawMutex, -{ - fn clone(&self) -> Self { - Receiver { - channel: self.channel, - } - } -} - -impl<'ch, M, T, const N: usize> Receiver<'ch, M, T, N> -where - M: RawMutex, -{ - /// Receive the next value. - /// - /// See [`Channel::recv()`]. - pub fn recv(&self) -> RecvFuture<'_, M, T, N> { - self.channel.recv() - } - - /// Attempt to immediately receive the next value. - /// - /// See [`Channel::try_recv()`] - pub fn try_recv(&self) -> Result { - self.channel.try_recv() - } -} - -/// Receive-only access to a [`Channel`] without knowing channel size. -#[derive(Copy)] -pub struct DynamicReceiver<'ch, T> { - channel: &'ch dyn DynamicChannel, -} - -impl<'ch, T> Clone for DynamicReceiver<'ch, T> { - fn clone(&self) -> Self { - DynamicReceiver { - channel: self.channel, - } - } -} - -impl<'ch, T> DynamicReceiver<'ch, T> { - /// Receive the next value. - /// - /// See [`Channel::recv()`]. - pub fn recv(&self) -> DynamicRecvFuture<'_, T> { - DynamicRecvFuture { - channel: self.channel, - } - } - - /// Attempt to immediately receive the next value. - /// - /// See [`Channel::try_recv()`] - pub fn try_recv(&self) -> Result { - self.channel.try_recv_with_context(None) - } -} - -impl<'ch, M, T, const N: usize> From> for DynamicReceiver<'ch, T> -where - M: RawMutex, -{ - fn from(s: Receiver<'ch, M, T, N>) -> Self { - Self { channel: s.channel } - } -} - -pub struct RecvFuture<'ch, M, T, const N: usize> -where - M: RawMutex, -{ - channel: &'ch Channel, -} - -impl<'ch, M, T, const N: usize> Future for RecvFuture<'ch, M, T, N> -where - M: RawMutex, -{ - type Output = T; - - fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll { - match self.channel.try_recv_with_context(Some(cx)) { - Ok(v) => Poll::Ready(v), - Err(TryRecvError::Empty) => Poll::Pending, - } - } -} - -pub struct DynamicRecvFuture<'ch, T> { - channel: &'ch dyn DynamicChannel, -} - -impl<'ch, T> Future for DynamicRecvFuture<'ch, T> { - type Output = T; - - fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll { - match self.channel.try_recv_with_context(Some(cx)) { - Ok(v) => Poll::Ready(v), - Err(TryRecvError::Empty) => Poll::Pending, - } - } -} - -pub struct SendFuture<'ch, M, T, const N: usize> -where - M: RawMutex, -{ - channel: &'ch Channel, - message: Option, -} - -impl<'ch, M, T, const N: usize> Future for SendFuture<'ch, M, T, N> -where - M: RawMutex, -{ - type Output = (); - - fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll { - match self.message.take() { - Some(m) => match self.channel.try_send_with_context(m, Some(cx)) { - Ok(..) => Poll::Ready(()), - Err(TrySendError::Full(m)) => { - self.message = Some(m); - Poll::Pending - } - }, - None => panic!("Message cannot be None"), - } - } -} - -impl<'ch, M, T, const N: usize> Unpin for SendFuture<'ch, M, T, N> where M: RawMutex {} - -pub struct DynamicSendFuture<'ch, T> { - channel: &'ch dyn DynamicChannel, - message: Option, -} - -impl<'ch, T> Future for DynamicSendFuture<'ch, T> { - type Output = (); - - fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll { - match self.message.take() { - Some(m) => match self.channel.try_send_with_context(m, Some(cx)) { - Ok(..) => Poll::Ready(()), - Err(TrySendError::Full(m)) => { - self.message = Some(m); - Poll::Pending - } - }, - None => panic!("Message cannot be None"), - } - } -} - -impl<'ch, T> Unpin for DynamicSendFuture<'ch, T> {} - -trait DynamicChannel { - fn try_send_with_context( - &self, - message: T, - cx: Option<&mut Context<'_>>, - ) -> Result<(), TrySendError>; - - fn try_recv_with_context(&self, cx: Option<&mut Context<'_>>) -> Result; -} - -/// Error returned by [`try_recv`](Channel::try_recv). -#[derive(PartialEq, Eq, Clone, Copy, Debug)] -#[cfg_attr(feature = "defmt", derive(defmt::Format))] -pub enum TryRecvError { - /// A message could not be received because the channel is empty. - Empty, -} - -/// Error returned by [`try_send`](Channel::try_send). -#[derive(PartialEq, Eq, Clone, Copy, Debug)] -#[cfg_attr(feature = "defmt", derive(defmt::Format))] -pub enum TrySendError { - /// The data could not be sent on the channel because the channel is - /// currently full and sending would require blocking. - Full(T), -} - -struct ChannelState { - queue: Deque, - receiver_waker: WakerRegistration, - senders_waker: WakerRegistration, -} - -impl ChannelState { - const fn new() -> Self { - ChannelState { - queue: Deque::new(), - receiver_waker: WakerRegistration::new(), - senders_waker: WakerRegistration::new(), - } - } - - fn try_recv(&mut self) -> Result { - self.try_recv_with_context(None) - } - - fn try_recv_with_context(&mut self, cx: Option<&mut Context<'_>>) -> Result { - if self.queue.is_full() { - self.senders_waker.wake(); - } - - if let Some(message) = self.queue.pop_front() { - Ok(message) - } else { - if let Some(cx) = cx { - self.receiver_waker.register(cx.waker()); - } - Err(TryRecvError::Empty) - } - } - - fn try_send(&mut self, message: T) -> Result<(), TrySendError> { - self.try_send_with_context(message, None) - } - - fn try_send_with_context( - &mut self, - message: T, - cx: Option<&mut Context<'_>>, - ) -> Result<(), TrySendError> { - match self.queue.push_back(message) { - Ok(()) => { - self.receiver_waker.wake(); - Ok(()) - } - Err(message) => { - if let Some(cx) = cx { - self.senders_waker.register(cx.waker()); - } - Err(TrySendError::Full(message)) - } - } - } -} - -/// A bounded channel for communicating between asynchronous tasks -/// with backpressure. -/// -/// The channel will buffer up to the provided number of messages. Once the -/// buffer is full, attempts to `send` new messages will wait until a message is -/// received from the channel. -/// -/// All data sent will become available in the same order as it was sent. -pub struct Channel -where - M: RawMutex, -{ - inner: Mutex>>, -} - -impl Channel -where - M: RawMutex, -{ - /// Establish a new bounded channel. For example, to create one with a NoopMutex: - /// - /// ``` - /// use embassy::channel::channel::Channel; - /// use embassy::blocking_mutex::raw::NoopRawMutex; - /// - /// // Declare a bounded channel of 3 u32s. - /// let mut channel = Channel::::new(); - /// ``` - #[cfg(feature = "nightly")] - pub const fn new() -> Self { - Self { - inner: Mutex::new(RefCell::new(ChannelState::new())), - } - } - - /// Establish a new bounded channel. For example, to create one with a NoopMutex: - /// - /// ``` - /// use embassy::channel::channel::Channel; - /// use embassy::blocking_mutex::raw::NoopRawMutex; - /// - /// // Declare a bounded channel of 3 u32s. - /// let mut channel = Channel::::new(); - /// ``` - #[cfg(not(feature = "nightly"))] - pub fn new() -> Self { - Self { - inner: Mutex::new(RefCell::new(ChannelState::new())), - } - } - - fn lock(&self, f: impl FnOnce(&mut ChannelState) -> R) -> R { - self.inner.lock(|rc| f(&mut *rc.borrow_mut())) - } - - fn try_recv_with_context(&self, cx: Option<&mut Context<'_>>) -> Result { - self.lock(|c| c.try_recv_with_context(cx)) - } - - fn try_send_with_context( - &self, - m: T, - cx: Option<&mut Context<'_>>, - ) -> Result<(), TrySendError> { - self.lock(|c| c.try_send_with_context(m, cx)) - } - - /// Get a sender for this channel. - pub fn sender(&self) -> Sender<'_, M, T, N> { - Sender { channel: self } - } - - /// Get a receiver for this channel. - pub fn receiver(&self) -> Receiver<'_, M, T, N> { - Receiver { channel: self } - } - - /// Send a value, waiting until there is capacity. - /// - /// Sending completes when the value has been pushed to the channel's queue. - /// This doesn't mean the value has been received yet. - pub fn send(&self, message: T) -> SendFuture<'_, M, T, N> { - SendFuture { - channel: self, - message: Some(message), - } - } - - /// Attempt to immediately send a message. - /// - /// This method differs from [`send`] by returning immediately if the channel's - /// buffer is full, instead of waiting. - /// - /// # Errors - /// - /// If the channel capacity has been reached, i.e., the channel has `n` - /// buffered values where `n` is the argument passed to [`Channel`], then an - /// error is returned. - pub fn try_send(&self, message: T) -> Result<(), TrySendError> { - self.lock(|c| c.try_send(message)) - } - - /// Receive the next value. - /// - /// If there are no messages in the channel's buffer, this method will - /// wait until a message is sent. - pub fn recv(&self) -> RecvFuture<'_, M, T, N> { - RecvFuture { channel: self } - } - - /// Attempt to immediately receive a message. - /// - /// This method will either receive a message from the channel immediately or return an error - /// if the channel is empty. - pub fn try_recv(&self) -> Result { - self.lock(|c| c.try_recv()) - } -} - -/// Implements the DynamicChannel to allow creating types that are unaware of the queue size with the -/// tradeoff cost of dynamic dispatch. -impl DynamicChannel for Channel -where - M: RawMutex, -{ - fn try_send_with_context( - &self, - m: T, - cx: Option<&mut Context<'_>>, - ) -> Result<(), TrySendError> { - Channel::try_send_with_context(self, m, cx) - } - - fn try_recv_with_context(&self, cx: Option<&mut Context<'_>>) -> Result { - Channel::try_recv_with_context(self, cx) - } -} - -#[cfg(test)] -mod tests { - use core::time::Duration; - - use futures::task::SpawnExt; - use futures_executor::ThreadPool; - use futures_timer::Delay; - - use crate::blocking_mutex::raw::{CriticalSectionRawMutex, NoopRawMutex}; - use crate::util::Forever; - - use super::*; - - fn capacity(c: &ChannelState) -> usize { - c.queue.capacity() - c.queue.len() - } - - #[test] - fn sending_once() { - let mut c = ChannelState::::new(); - assert!(c.try_send(1).is_ok()); - assert_eq!(capacity(&c), 2); - } - - #[test] - fn sending_when_full() { - let mut c = ChannelState::::new(); - let _ = c.try_send(1); - let _ = c.try_send(1); - let _ = c.try_send(1); - match c.try_send(2) { - Err(TrySendError::Full(2)) => assert!(true), - _ => assert!(false), - } - assert_eq!(capacity(&c), 0); - } - - #[test] - fn receiving_once_with_one_send() { - let mut c = ChannelState::::new(); - assert!(c.try_send(1).is_ok()); - assert_eq!(c.try_recv().unwrap(), 1); - assert_eq!(capacity(&c), 3); - } - - #[test] - fn receiving_when_empty() { - let mut c = ChannelState::::new(); - match c.try_recv() { - Err(TryRecvError::Empty) => assert!(true), - _ => assert!(false), - } - assert_eq!(capacity(&c), 3); - } - - #[test] - fn simple_send_and_receive() { - let c = Channel::::new(); - assert!(c.try_send(1).is_ok()); - assert_eq!(c.try_recv().unwrap(), 1); - } - - #[test] - fn cloning() { - let c = Channel::::new(); - let r1 = c.receiver(); - let s1 = c.sender(); - - let _ = r1.clone(); - let _ = s1.clone(); - } - - #[test] - fn dynamic_dispatch() { - let c = Channel::::new(); - let s: DynamicSender<'_, u32> = c.sender().into(); - let r: DynamicReceiver<'_, u32> = c.receiver().into(); - - assert!(s.try_send(1).is_ok()); - assert_eq!(r.try_recv().unwrap(), 1); - } - - #[futures_test::test] - async fn receiver_receives_given_try_send_async() { - let executor = ThreadPool::new().unwrap(); - - static CHANNEL: Forever> = Forever::new(); - let c = &*CHANNEL.put(Channel::new()); - let c2 = c; - assert!(executor - .spawn(async move { - assert!(c2.try_send(1).is_ok()); - }) - .is_ok()); - assert_eq!(c.recv().await, 1); - } - - #[futures_test::test] - async fn sender_send_completes_if_capacity() { - let c = Channel::::new(); - c.send(1).await; - assert_eq!(c.recv().await, 1); - } - - #[futures_test::test] - async fn senders_sends_wait_until_capacity() { - let executor = ThreadPool::new().unwrap(); - - static CHANNEL: Forever> = Forever::new(); - let c = &*CHANNEL.put(Channel::new()); - assert!(c.try_send(1).is_ok()); - - let c2 = c; - let send_task_1 = executor.spawn_with_handle(async move { c2.send(2).await }); - let c2 = c; - let send_task_2 = executor.spawn_with_handle(async move { c2.send(3).await }); - // Wish I could think of a means of determining that the async send is waiting instead. - // However, I've used the debugger to observe that the send does indeed wait. - Delay::new(Duration::from_millis(500)).await; - assert_eq!(c.recv().await, 1); - assert!(executor - .spawn(async move { - loop { - c.recv().await; - } - }) - .is_ok()); - send_task_1.unwrap().await; - send_task_2.unwrap().await; - } -} diff --git a/embassy/src/channel/mod.rs b/embassy/src/channel/mod.rs index 4d37fe859..05edc55d1 100644 --- a/embassy/src/channel/mod.rs +++ b/embassy/src/channel/mod.rs @@ -1,7 +1,5 @@ //! Async channels -pub mod channel; -pub use channel::*; +pub mod mpmc; pub mod signal; -pub use signal::*; diff --git a/embassy/src/channel/mpmc.rs b/embassy/src/channel/mpmc.rs new file mode 100644 index 000000000..2377a9665 --- /dev/null +++ b/embassy/src/channel/mpmc.rs @@ -0,0 +1,637 @@ +//! A queue for sending values between asynchronous tasks. +//! +//! It can be used concurrently by multiple producers (senders) and multiple +//! consumers (receivers), i.e. it is an "MPMC channel". +//! +//! Receivers are competing for messages. So a message that is received by +//! one receiver is not received by any other. +//! +//! This queue takes a Mutex type so that various +//! targets can be attained. For example, a ThreadModeMutex can be used +//! for single-core Cortex-M targets where messages are only passed +//! between tasks running in thread mode. Similarly, a CriticalSectionMutex +//! can also be used for single-core targets where messages are to be +//! passed from exception mode e.g. out of an interrupt handler. +//! +//! This module provides a bounded channel that has a limit on the number of +//! messages that it can store, and if this limit is reached, trying to send +//! another message will result in an error being returned. +//! + +use core::cell::RefCell; +use core::pin::Pin; +use core::task::Context; +use core::task::Poll; + +use futures::Future; +use heapless::Deque; + +use crate::blocking_mutex::raw::RawMutex; +use crate::blocking_mutex::Mutex; +use crate::waitqueue::WakerRegistration; + +/// Send-only access to a [`Channel`]. +#[derive(Copy)] +pub struct Sender<'ch, M, T, const N: usize> +where + M: RawMutex, +{ + channel: &'ch Channel, +} + +impl<'ch, M, T, const N: usize> Clone for Sender<'ch, M, T, N> +where + M: RawMutex, +{ + fn clone(&self) -> Self { + Sender { + channel: self.channel, + } + } +} + +impl<'ch, M, T, const N: usize> Sender<'ch, M, T, N> +where + M: RawMutex, +{ + /// Sends a value. + /// + /// See [`Channel::send()`] + pub fn send(&self, message: T) -> SendFuture<'ch, M, T, N> { + self.channel.send(message) + } + + /// Attempt to immediately send a message. + /// + /// See [`Channel::send()`] + pub fn try_send(&self, message: T) -> Result<(), TrySendError> { + self.channel.try_send(message) + } +} + +/// Send-only access to a [`Channel`] without knowing channel size. +#[derive(Copy)] +pub struct DynamicSender<'ch, T> { + channel: &'ch dyn DynamicChannel, +} + +impl<'ch, T> Clone for DynamicSender<'ch, T> { + fn clone(&self) -> Self { + DynamicSender { + channel: self.channel, + } + } +} + +impl<'ch, M, T, const N: usize> From> for DynamicSender<'ch, T> +where + M: RawMutex, +{ + fn from(s: Sender<'ch, M, T, N>) -> Self { + Self { channel: s.channel } + } +} + +impl<'ch, T> DynamicSender<'ch, T> { + /// Sends a value. + /// + /// See [`Channel::send()`] + pub fn send(&self, message: T) -> DynamicSendFuture<'ch, T> { + DynamicSendFuture { + channel: self.channel, + message: Some(message), + } + } + + /// Attempt to immediately send a message. + /// + /// See [`Channel::send()`] + pub fn try_send(&self, message: T) -> Result<(), TrySendError> { + self.channel.try_send_with_context(message, None) + } +} + +/// Receive-only access to a [`Channel`]. +#[derive(Copy)] +pub struct Receiver<'ch, M, T, const N: usize> +where + M: RawMutex, +{ + channel: &'ch Channel, +} + +impl<'ch, M, T, const N: usize> Clone for Receiver<'ch, M, T, N> +where + M: RawMutex, +{ + fn clone(&self) -> Self { + Receiver { + channel: self.channel, + } + } +} + +impl<'ch, M, T, const N: usize> Receiver<'ch, M, T, N> +where + M: RawMutex, +{ + /// Receive the next value. + /// + /// See [`Channel::recv()`]. + pub fn recv(&self) -> RecvFuture<'_, M, T, N> { + self.channel.recv() + } + + /// Attempt to immediately receive the next value. + /// + /// See [`Channel::try_recv()`] + pub fn try_recv(&self) -> Result { + self.channel.try_recv() + } +} + +/// Receive-only access to a [`Channel`] without knowing channel size. +#[derive(Copy)] +pub struct DynamicReceiver<'ch, T> { + channel: &'ch dyn DynamicChannel, +} + +impl<'ch, T> Clone for DynamicReceiver<'ch, T> { + fn clone(&self) -> Self { + DynamicReceiver { + channel: self.channel, + } + } +} + +impl<'ch, T> DynamicReceiver<'ch, T> { + /// Receive the next value. + /// + /// See [`Channel::recv()`]. + pub fn recv(&self) -> DynamicRecvFuture<'_, T> { + DynamicRecvFuture { + channel: self.channel, + } + } + + /// Attempt to immediately receive the next value. + /// + /// See [`Channel::try_recv()`] + pub fn try_recv(&self) -> Result { + self.channel.try_recv_with_context(None) + } +} + +impl<'ch, M, T, const N: usize> From> for DynamicReceiver<'ch, T> +where + M: RawMutex, +{ + fn from(s: Receiver<'ch, M, T, N>) -> Self { + Self { channel: s.channel } + } +} + +pub struct RecvFuture<'ch, M, T, const N: usize> +where + M: RawMutex, +{ + channel: &'ch Channel, +} + +impl<'ch, M, T, const N: usize> Future for RecvFuture<'ch, M, T, N> +where + M: RawMutex, +{ + type Output = T; + + fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll { + match self.channel.try_recv_with_context(Some(cx)) { + Ok(v) => Poll::Ready(v), + Err(TryRecvError::Empty) => Poll::Pending, + } + } +} + +pub struct DynamicRecvFuture<'ch, T> { + channel: &'ch dyn DynamicChannel, +} + +impl<'ch, T> Future for DynamicRecvFuture<'ch, T> { + type Output = T; + + fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll { + match self.channel.try_recv_with_context(Some(cx)) { + Ok(v) => Poll::Ready(v), + Err(TryRecvError::Empty) => Poll::Pending, + } + } +} + +pub struct SendFuture<'ch, M, T, const N: usize> +where + M: RawMutex, +{ + channel: &'ch Channel, + message: Option, +} + +impl<'ch, M, T, const N: usize> Future for SendFuture<'ch, M, T, N> +where + M: RawMutex, +{ + type Output = (); + + fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll { + match self.message.take() { + Some(m) => match self.channel.try_send_with_context(m, Some(cx)) { + Ok(..) => Poll::Ready(()), + Err(TrySendError::Full(m)) => { + self.message = Some(m); + Poll::Pending + } + }, + None => panic!("Message cannot be None"), + } + } +} + +impl<'ch, M, T, const N: usize> Unpin for SendFuture<'ch, M, T, N> where M: RawMutex {} + +pub struct DynamicSendFuture<'ch, T> { + channel: &'ch dyn DynamicChannel, + message: Option, +} + +impl<'ch, T> Future for DynamicSendFuture<'ch, T> { + type Output = (); + + fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll { + match self.message.take() { + Some(m) => match self.channel.try_send_with_context(m, Some(cx)) { + Ok(..) => Poll::Ready(()), + Err(TrySendError::Full(m)) => { + self.message = Some(m); + Poll::Pending + } + }, + None => panic!("Message cannot be None"), + } + } +} + +impl<'ch, T> Unpin for DynamicSendFuture<'ch, T> {} + +trait DynamicChannel { + fn try_send_with_context( + &self, + message: T, + cx: Option<&mut Context<'_>>, + ) -> Result<(), TrySendError>; + + fn try_recv_with_context(&self, cx: Option<&mut Context<'_>>) -> Result; +} + +/// Error returned by [`try_recv`](Channel::try_recv). +#[derive(PartialEq, Eq, Clone, Copy, Debug)] +#[cfg_attr(feature = "defmt", derive(defmt::Format))] +pub enum TryRecvError { + /// A message could not be received because the channel is empty. + Empty, +} + +/// Error returned by [`try_send`](Channel::try_send). +#[derive(PartialEq, Eq, Clone, Copy, Debug)] +#[cfg_attr(feature = "defmt", derive(defmt::Format))] +pub enum TrySendError { + /// The data could not be sent on the channel because the channel is + /// currently full and sending would require blocking. + Full(T), +} + +struct ChannelState { + queue: Deque, + receiver_waker: WakerRegistration, + senders_waker: WakerRegistration, +} + +impl ChannelState { + const fn new() -> Self { + ChannelState { + queue: Deque::new(), + receiver_waker: WakerRegistration::new(), + senders_waker: WakerRegistration::new(), + } + } + + fn try_recv(&mut self) -> Result { + self.try_recv_with_context(None) + } + + fn try_recv_with_context(&mut self, cx: Option<&mut Context<'_>>) -> Result { + if self.queue.is_full() { + self.senders_waker.wake(); + } + + if let Some(message) = self.queue.pop_front() { + Ok(message) + } else { + if let Some(cx) = cx { + self.receiver_waker.register(cx.waker()); + } + Err(TryRecvError::Empty) + } + } + + fn try_send(&mut self, message: T) -> Result<(), TrySendError> { + self.try_send_with_context(message, None) + } + + fn try_send_with_context( + &mut self, + message: T, + cx: Option<&mut Context<'_>>, + ) -> Result<(), TrySendError> { + match self.queue.push_back(message) { + Ok(()) => { + self.receiver_waker.wake(); + Ok(()) + } + Err(message) => { + if let Some(cx) = cx { + self.senders_waker.register(cx.waker()); + } + Err(TrySendError::Full(message)) + } + } + } +} + +/// A bounded channel for communicating between asynchronous tasks +/// with backpressure. +/// +/// The channel will buffer up to the provided number of messages. Once the +/// buffer is full, attempts to `send` new messages will wait until a message is +/// received from the channel. +/// +/// All data sent will become available in the same order as it was sent. +pub struct Channel +where + M: RawMutex, +{ + inner: Mutex>>, +} + +impl Channel +where + M: RawMutex, +{ + /// Establish a new bounded channel. For example, to create one with a NoopMutex: + /// + /// ``` + /// use embassy::channel::mpmc::Channel; + /// use embassy::blocking_mutex::raw::NoopRawMutex; + /// + /// // Declare a bounded channel of 3 u32s. + /// let mut channel = Channel::::new(); + /// ``` + #[cfg(feature = "nightly")] + pub const fn new() -> Self { + Self { + inner: Mutex::new(RefCell::new(ChannelState::new())), + } + } + + /// Establish a new bounded channel. For example, to create one with a NoopMutex: + /// + /// ``` + /// use embassy::channel::mpmc::Channel; + /// use embassy::blocking_mutex::raw::NoopRawMutex; + /// + /// // Declare a bounded channel of 3 u32s. + /// let mut channel = Channel::::new(); + /// ``` + #[cfg(not(feature = "nightly"))] + pub fn new() -> Self { + Self { + inner: Mutex::new(RefCell::new(ChannelState::new())), + } + } + + fn lock(&self, f: impl FnOnce(&mut ChannelState) -> R) -> R { + self.inner.lock(|rc| f(&mut *rc.borrow_mut())) + } + + fn try_recv_with_context(&self, cx: Option<&mut Context<'_>>) -> Result { + self.lock(|c| c.try_recv_with_context(cx)) + } + + fn try_send_with_context( + &self, + m: T, + cx: Option<&mut Context<'_>>, + ) -> Result<(), TrySendError> { + self.lock(|c| c.try_send_with_context(m, cx)) + } + + /// Get a sender for this channel. + pub fn sender(&self) -> Sender<'_, M, T, N> { + Sender { channel: self } + } + + /// Get a receiver for this channel. + pub fn receiver(&self) -> Receiver<'_, M, T, N> { + Receiver { channel: self } + } + + /// Send a value, waiting until there is capacity. + /// + /// Sending completes when the value has been pushed to the channel's queue. + /// This doesn't mean the value has been received yet. + pub fn send(&self, message: T) -> SendFuture<'_, M, T, N> { + SendFuture { + channel: self, + message: Some(message), + } + } + + /// Attempt to immediately send a message. + /// + /// This method differs from [`send`] by returning immediately if the channel's + /// buffer is full, instead of waiting. + /// + /// # Errors + /// + /// If the channel capacity has been reached, i.e., the channel has `n` + /// buffered values where `n` is the argument passed to [`Channel`], then an + /// error is returned. + pub fn try_send(&self, message: T) -> Result<(), TrySendError> { + self.lock(|c| c.try_send(message)) + } + + /// Receive the next value. + /// + /// If there are no messages in the channel's buffer, this method will + /// wait until a message is sent. + pub fn recv(&self) -> RecvFuture<'_, M, T, N> { + RecvFuture { channel: self } + } + + /// Attempt to immediately receive a message. + /// + /// This method will either receive a message from the channel immediately or return an error + /// if the channel is empty. + pub fn try_recv(&self) -> Result { + self.lock(|c| c.try_recv()) + } +} + +/// Implements the DynamicChannel to allow creating types that are unaware of the queue size with the +/// tradeoff cost of dynamic dispatch. +impl DynamicChannel for Channel +where + M: RawMutex, +{ + fn try_send_with_context( + &self, + m: T, + cx: Option<&mut Context<'_>>, + ) -> Result<(), TrySendError> { + Channel::try_send_with_context(self, m, cx) + } + + fn try_recv_with_context(&self, cx: Option<&mut Context<'_>>) -> Result { + Channel::try_recv_with_context(self, cx) + } +} + +#[cfg(test)] +mod tests { + use core::time::Duration; + + use futures::task::SpawnExt; + use futures_executor::ThreadPool; + use futures_timer::Delay; + + use crate::blocking_mutex::raw::{CriticalSectionRawMutex, NoopRawMutex}; + use crate::util::Forever; + + use super::*; + + fn capacity(c: &ChannelState) -> usize { + c.queue.capacity() - c.queue.len() + } + + #[test] + fn sending_once() { + let mut c = ChannelState::::new(); + assert!(c.try_send(1).is_ok()); + assert_eq!(capacity(&c), 2); + } + + #[test] + fn sending_when_full() { + let mut c = ChannelState::::new(); + let _ = c.try_send(1); + let _ = c.try_send(1); + let _ = c.try_send(1); + match c.try_send(2) { + Err(TrySendError::Full(2)) => assert!(true), + _ => assert!(false), + } + assert_eq!(capacity(&c), 0); + } + + #[test] + fn receiving_once_with_one_send() { + let mut c = ChannelState::::new(); + assert!(c.try_send(1).is_ok()); + assert_eq!(c.try_recv().unwrap(), 1); + assert_eq!(capacity(&c), 3); + } + + #[test] + fn receiving_when_empty() { + let mut c = ChannelState::::new(); + match c.try_recv() { + Err(TryRecvError::Empty) => assert!(true), + _ => assert!(false), + } + assert_eq!(capacity(&c), 3); + } + + #[test] + fn simple_send_and_receive() { + let c = Channel::::new(); + assert!(c.try_send(1).is_ok()); + assert_eq!(c.try_recv().unwrap(), 1); + } + + #[test] + fn cloning() { + let c = Channel::::new(); + let r1 = c.receiver(); + let s1 = c.sender(); + + let _ = r1.clone(); + let _ = s1.clone(); + } + + #[test] + fn dynamic_dispatch() { + let c = Channel::::new(); + let s: DynamicSender<'_, u32> = c.sender().into(); + let r: DynamicReceiver<'_, u32> = c.receiver().into(); + + assert!(s.try_send(1).is_ok()); + assert_eq!(r.try_recv().unwrap(), 1); + } + + #[futures_test::test] + async fn receiver_receives_given_try_send_async() { + let executor = ThreadPool::new().unwrap(); + + static CHANNEL: Forever> = Forever::new(); + let c = &*CHANNEL.put(Channel::new()); + let c2 = c; + assert!(executor + .spawn(async move { + assert!(c2.try_send(1).is_ok()); + }) + .is_ok()); + assert_eq!(c.recv().await, 1); + } + + #[futures_test::test] + async fn sender_send_completes_if_capacity() { + let c = Channel::::new(); + c.send(1).await; + assert_eq!(c.recv().await, 1); + } + + #[futures_test::test] + async fn senders_sends_wait_until_capacity() { + let executor = ThreadPool::new().unwrap(); + + static CHANNEL: Forever> = Forever::new(); + let c = &*CHANNEL.put(Channel::new()); + assert!(c.try_send(1).is_ok()); + + let c2 = c; + let send_task_1 = executor.spawn_with_handle(async move { c2.send(2).await }); + let c2 = c; + let send_task_2 = executor.spawn_with_handle(async move { c2.send(3).await }); + // Wish I could think of a means of determining that the async send is waiting instead. + // However, I've used the debugger to observe that the send does indeed wait. + Delay::new(Duration::from_millis(500)).await; + assert_eq!(c.recv().await, 1); + assert!(executor + .spawn(async move { + loop { + c.recv().await; + } + }) + .is_ok()); + send_task_1.unwrap().await; + send_task_2.unwrap().await; + } +} diff --git a/embassy/src/channel/signal.rs b/embassy/src/channel/signal.rs index e1f6c4b1d..5a2c9d47a 100644 --- a/embassy/src/channel/signal.rs +++ b/embassy/src/channel/signal.rs @@ -5,7 +5,7 @@ use core::task::{Context, Poll, Waker}; /// Synchronization primitive. Allows creating awaitable signals that may be passed between tasks. /// For a simple use-case where the receiver is only ever interested in the latest value of -/// something, Signals work well. For more advanced use cases, you might want to use [`Channel`](crate::channel::channel::Channel) instead.. +/// something, Signals work well. For more advanced use cases, you might want to use [`Channel`](crate::channel::mpmc::Channel) instead.. /// /// Signals are generally declared as being a static const and then borrowed as required. /// diff --git a/examples/nrf/src/bin/channel.rs b/examples/nrf/src/bin/channel.rs index 476ec09a1..12b463581 100644 --- a/examples/nrf/src/bin/channel.rs +++ b/examples/nrf/src/bin/channel.rs @@ -4,7 +4,7 @@ use defmt::unwrap; use embassy::blocking_mutex::raw::ThreadModeRawMutex; -use embassy::channel::channel::Channel; +use embassy::channel::mpmc::Channel; use embassy::executor::Spawner; use embassy::time::{Duration, Timer}; use embassy_nrf::gpio::{Level, Output, OutputDrive}; diff --git a/examples/nrf/src/bin/channel_sender_receiver.rs b/examples/nrf/src/bin/channel_sender_receiver.rs index c79f2fd6b..1742dcc9e 100644 --- a/examples/nrf/src/bin/channel_sender_receiver.rs +++ b/examples/nrf/src/bin/channel_sender_receiver.rs @@ -4,7 +4,7 @@ use defmt::unwrap; use embassy::blocking_mutex::raw::NoopRawMutex; -use embassy::channel::channel::{Channel, Receiver, Sender}; +use embassy::channel::mpmc::{Channel, Receiver, Sender}; use embassy::executor::Spawner; use embassy::time::{Duration, Timer}; use embassy::util::Forever; diff --git a/examples/nrf/src/bin/uart_split.rs b/examples/nrf/src/bin/uart_split.rs index 3fde2f0d8..958cfeba0 100644 --- a/examples/nrf/src/bin/uart_split.rs +++ b/examples/nrf/src/bin/uart_split.rs @@ -4,7 +4,7 @@ use defmt::*; use embassy::blocking_mutex::raw::ThreadModeRawMutex; -use embassy::channel::channel::Channel; +use embassy::channel::mpmc::Channel; use embassy::executor::Spawner; use embassy_nrf::peripherals::UARTE0; use embassy_nrf::uarte::UarteRx; diff --git a/examples/nrf/src/bin/usb_ethernet.rs b/examples/nrf/src/bin/usb_ethernet.rs index 49f2fb89a..3d85d22bb 100644 --- a/examples/nrf/src/bin/usb_ethernet.rs +++ b/examples/nrf/src/bin/usb_ethernet.rs @@ -8,7 +8,7 @@ use core::sync::atomic::{AtomicBool, Ordering}; use core::task::Waker; use defmt::*; use embassy::blocking_mutex::raw::ThreadModeRawMutex; -use embassy::channel::Channel; +use embassy::channel::mpmc::Channel; use embassy::executor::Spawner; use embassy::util::Forever; use embassy_net::tcp::TcpSocket; diff --git a/examples/nrf/src/bin/usb_hid_keyboard.rs b/examples/nrf/src/bin/usb_hid_keyboard.rs index d855a3a57..d7affae11 100644 --- a/examples/nrf/src/bin/usb_hid_keyboard.rs +++ b/examples/nrf/src/bin/usb_hid_keyboard.rs @@ -6,7 +6,7 @@ use core::mem; use core::sync::atomic::{AtomicBool, Ordering}; use defmt::*; -use embassy::channel::Signal; +use embassy::channel::signal::Signal; use embassy::executor::Spawner; use embassy::interrupt::InterruptExt; use embassy::time::Duration; diff --git a/examples/stm32f3/src/bin/button_events.rs b/examples/stm32f3/src/bin/button_events.rs index 06e8eec1f..593701799 100644 --- a/examples/stm32f3/src/bin/button_events.rs +++ b/examples/stm32f3/src/bin/button_events.rs @@ -12,7 +12,7 @@ use defmt::*; use embassy::blocking_mutex::raw::ThreadModeRawMutex; -use embassy::channel::channel::Channel; +use embassy::channel::mpmc::Channel; use embassy::executor::Spawner; use embassy::time::{with_timeout, Duration, Timer}; use embassy_stm32::exti::ExtiInput; diff --git a/examples/stm32h7/src/bin/signal.rs b/examples/stm32h7/src/bin/signal.rs index d2531b2ba..c63b2836c 100644 --- a/examples/stm32h7/src/bin/signal.rs +++ b/examples/stm32h7/src/bin/signal.rs @@ -8,7 +8,7 @@ use defmt_rtt as _; use panic_probe as _; -use embassy::channel::Signal; +use embassy::channel::signal::Signal; use embassy::executor::Spawner; use embassy::time::{Duration, Timer}; diff --git a/examples/stm32h7/src/bin/usart_split.rs b/examples/stm32h7/src/bin/usart_split.rs index 40a7c3e44..eb3f9578c 100644 --- a/examples/stm32h7/src/bin/usart_split.rs +++ b/examples/stm32h7/src/bin/usart_split.rs @@ -5,7 +5,7 @@ use defmt::*; use defmt_rtt as _; // global logger use embassy::blocking_mutex::raw::ThreadModeRawMutex; -use embassy::channel::channel::Channel; +use embassy::channel::mpmc::Channel; use embassy::executor::Spawner; use embassy_stm32::dma::NoDma; use embassy_stm32::{ diff --git a/examples/stm32l5/src/bin/usb_ethernet.rs b/examples/stm32l5/src/bin/usb_ethernet.rs index fa445eece..526b27530 100644 --- a/examples/stm32l5/src/bin/usb_ethernet.rs +++ b/examples/stm32l5/src/bin/usb_ethernet.rs @@ -8,7 +8,7 @@ use core::task::Waker; use defmt::*; use defmt_rtt as _; // global logger use embassy::blocking_mutex::raw::ThreadModeRawMutex; -use embassy::channel::Channel; +use embassy::channel::mpmc::Channel; use embassy::executor::Spawner; use embassy::util::Forever; use embassy_net::tcp::TcpSocket; -- cgit