Merge pull request #1763 from rubdos/sender-receiver-with-ctx

Refactor Channel/Sender/Receiver poll methods

embassy-stm32-wpan/src/mac/driver.rs

@@ -28,7 +28,9 @@ impl<'d> embassy_net_driver::Driver for Driver<'d> {
     type TxToken<'a> = TxToken<'d> where Self: 'a;
 
     fn receive(&mut self, cx: &mut Context) -> Option<(Self::RxToken<'_>, Self::TxToken<'_>)> {
-        if self.runner.rx_channel.poll_ready_to_receive(cx) && self.runner.tx_buf_channel.poll_ready_to_receive(cx) {
+        if self.runner.rx_channel.poll_ready_to_receive(cx).is_ready()
+            && self.runner.tx_buf_channel.poll_ready_to_receive(cx).is_ready()
+        {
             Some((
                 RxToken {
                     rx: &self.runner.rx_channel,
@@ -44,7 +46,7 @@ impl<'d> embassy_net_driver::Driver for Driver<'d> {
     }
 
     fn transmit(&mut self, cx: &mut Context) -> Option<Self::TxToken<'_>> {
-        if self.runner.tx_buf_channel.poll_ready_to_receive(cx) {
+        if self.runner.tx_buf_channel.poll_ready_to_receive(cx).is_ready() {
             Some(TxToken {
                 tx: &self.runner.tx_channel,
                 tx_buf: &self.runner.tx_buf_channel,
@@ -91,7 +93,7 @@ impl<'d> embassy_net_driver::RxToken for RxToken<'d> {
     {
         // Only valid data events should be put into the queue
 
-        let data_event = match self.rx.try_recv().unwrap() {
+        let data_event = match self.rx.try_receive().unwrap() {
             MacEvent::McpsDataInd(data_event) => data_event,
             _ => unreachable!(),
         };
@@ -111,7 +113,7 @@ impl<'d> embassy_net_driver::TxToken for TxToken<'d> {
         F: FnOnce(&mut [u8]) -> R,
     {
         // Only valid tx buffers should be put into the queue
-        let buf = self.tx_buf.try_recv().unwrap();
+        let buf = self.tx_buf.try_receive().unwrap();
         let r = f(&mut buf[..len]);
 
         // The tx channel should always be of equal capacity to the tx_buf channel

embassy-stm32-wpan/src/mac/runner.rs

@@ -73,7 +73,7 @@ impl<'a> Runner<'a> {
                 let mut msdu_handle = 0x02;
 
                 loop {
-                    let (buf, len) = self.tx_channel.recv().await;
+                    let (buf, len) = self.tx_channel.receive().await;
                     let _wm = self.write_mutex.lock().await;
 
                     // The mutex should be dropped on the next loop iteration

embassy-stm32/src/can/bxcan.rs

@@ -478,7 +478,7 @@ impl<'c, 'd, T: Instance> CanRx<'c, 'd, T> {
     pub async fn read(&mut self) -> Result<Envelope, BusError> {
         poll_fn(|cx| {
             T::state().err_waker.register(cx.waker());
-            if let Poll::Ready(envelope) = T::state().rx_queue.recv().poll_unpin(cx) {
+            if let Poll::Ready(envelope) = T::state().rx_queue.receive().poll_unpin(cx) {
                 return Poll::Ready(Ok(envelope));
             } else if let Some(err) = self.curr_error() {
                 return Poll::Ready(Err(err));
@@ -493,7 +493,7 @@ impl<'c, 'd, T: Instance> CanRx<'c, 'd, T> {
     ///
     /// Returns [Err(TryReadError::Empty)] if there are no frames in the rx queue.
     pub fn try_read(&mut self) -> Result<Envelope, TryReadError> {
-        if let Ok(envelope) = T::state().rx_queue.try_recv() {
+        if let Ok(envelope) = T::state().rx_queue.try_receive() {
             return Ok(envelope);
         }
 
@@ -506,14 +506,7 @@ impl<'c, 'd, T: Instance> CanRx<'c, 'd, T> {
 
     /// Waits while receive queue is empty.
     pub async fn wait_not_empty(&mut self) {
-        poll_fn(|cx| {
+        poll_fn(|cx| T::state().rx_queue.poll_ready_to_receive(cx)).await
-            if T::state().rx_queue.poll_ready_to_receive(cx) {
-                Poll::Ready(())
-            } else {
-                Poll::Pending
-            }
-        })
-        .await
     }
 
     fn curr_error(&self) -> Option<BusError> {

embassy-sync/src/channel.rs

@@ -65,6 +65,13 @@ where
     pub fn try_send(&self, message: T) -> Result<(), TrySendError<T>> {
         self.channel.try_send(message)
     }
+
+    /// Allows a poll_fn to poll until the channel is ready to send
+    ///
+    /// See [`Channel::poll_ready_to_send()`]
+    pub fn poll_ready_to_send(&self, cx: &mut Context<'_>) -> Poll<()> {
+        self.channel.poll_ready_to_send(cx)
+    }
 }
 
 /// Send-only access to a [`Channel`] without knowing channel size.
@@ -106,6 +113,13 @@ impl<'ch, T> DynamicSender<'ch, T> {
     pub fn try_send(&self, message: T) -> Result<(), TrySendError<T>> {
         self.channel.try_send_with_context(message, None)
     }
+
+    /// Allows a poll_fn to poll until the channel is ready to send
+    ///
+    /// See [`Channel::poll_ready_to_send()`]
+    pub fn poll_ready_to_send(&self, cx: &mut Context<'_>) -> Poll<()> {
+        self.channel.poll_ready_to_send(cx)
+    }
 }
 
 /// Receive-only access to a [`Channel`].
@@ -133,16 +147,30 @@ where
 {
     /// Receive the next value.
     ///
-    /// See [`Channel::recv()`].
+    /// See [`Channel::receive()`].
-    pub fn recv(&self) -> RecvFuture<'_, M, T, N> {
+    pub fn receive(&self) -> ReceiveFuture<'_, M, T, N> {
-        self.channel.recv()
+        self.channel.receive()
     }
 
     /// Attempt to immediately receive the next value.
     ///
-    /// See [`Channel::try_recv()`]
+    /// See [`Channel::try_receive()`]
-    pub fn try_recv(&self) -> Result<T, TryRecvError> {
+    pub fn try_receive(&self) -> Result<T, TryReceiveError> {
-        self.channel.try_recv()
+        self.channel.try_receive()
+    }
+
+    /// Allows a poll_fn to poll until the channel is ready to receive
+    ///
+    /// See [`Channel::poll_ready_to_receive()`]
+    pub fn poll_ready_to_receive(&self, cx: &mut Context<'_>) -> Poll<()> {
+        self.channel.poll_ready_to_receive(cx)
+    }
+
+    /// Poll the channel for the next item
+    ///
+    /// See [`Channel::poll_receive()`]
+    pub fn poll_receive(&self, cx: &mut Context<'_>) -> Poll<T> {
+        self.channel.poll_receive(cx)
     }
 }
 
@@ -162,16 +190,30 @@ impl<'ch, T> Copy for DynamicReceiver<'ch, T> {}
 impl<'ch, T> DynamicReceiver<'ch, T> {
     /// Receive the next value.
     ///
-    /// See [`Channel::recv()`].
+    /// See [`Channel::receive()`].
-    pub fn recv(&self) -> DynamicRecvFuture<'_, T> {
+    pub fn receive(&self) -> DynamicReceiveFuture<'_, T> {
-        DynamicRecvFuture { channel: self.channel }
+        DynamicReceiveFuture { channel: self.channel }
     }
 
     /// Attempt to immediately receive the next value.
     ///
-    /// See [`Channel::try_recv()`]
+    /// See [`Channel::try_receive()`]
-    pub fn try_recv(&self) -> Result<T, TryRecvError> {
+    pub fn try_receive(&self) -> Result<T, TryReceiveError> {
-        self.channel.try_recv_with_context(None)
+        self.channel.try_receive_with_context(None)
+    }
+
+    /// Allows a poll_fn to poll until the channel is ready to receive
+    ///
+    /// See [`Channel::poll_ready_to_receive()`]
+    pub fn poll_ready_to_receive(&self, cx: &mut Context<'_>) -> Poll<()> {
+        self.channel.poll_ready_to_receive(cx)
+    }
+
+    /// Poll the channel for the next item
+    ///
+    /// See [`Channel::poll_receive()`]
+    pub fn poll_receive(&self, cx: &mut Context<'_>) -> Poll<T> {
+        self.channel.poll_receive(cx)
     }
 }
 
@@ -184,42 +226,39 @@ where
     }
 }
 
-/// Future returned by [`Channel::recv`] and  [`Receiver::recv`].
+/// Future returned by [`Channel::receive`] and  [`Receiver::receive`].
 #[must_use = "futures do nothing unless you `.await` or poll them"]
-pub struct RecvFuture<'ch, M, T, const N: usize>
+pub struct ReceiveFuture<'ch, M, T, const N: usize>
 where
     M: RawMutex,
 {
     channel: &'ch Channel<M, T, N>,
 }
 
-impl<'ch, M, T, const N: usize> Future for RecvFuture<'ch, M, T, N>
+impl<'ch, M, T, const N: usize> Future for ReceiveFuture<'ch, M, T, N>
 where
     M: RawMutex,
 {
     type Output = T;
 
     fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<T> {
-        match self.channel.try_recv_with_context(Some(cx)) {
+        self.channel.poll_receive(cx)
-            Ok(v) => Poll::Ready(v),
-            Err(TryRecvError::Empty) => Poll::Pending,
-        }
     }
 }
 
-/// Future returned by [`DynamicReceiver::recv`].
+/// Future returned by [`DynamicReceiver::receive`].
 #[must_use = "futures do nothing unless you `.await` or poll them"]
-pub struct DynamicRecvFuture<'ch, T> {
+pub struct DynamicReceiveFuture<'ch, T> {
     channel: &'ch dyn DynamicChannel<T>,
 }
 
-impl<'ch, T> Future for DynamicRecvFuture<'ch, T> {
+impl<'ch, T> Future for DynamicReceiveFuture<'ch, T> {
     type Output = T;
 
     fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<T> {
-        match self.channel.try_recv_with_context(Some(cx)) {
+        match self.channel.try_receive_with_context(Some(cx)) {
             Ok(v) => Poll::Ready(v),
-            Err(TryRecvError::Empty) => Poll::Pending,
+            Err(TryReceiveError::Empty) => Poll::Pending,
         }
     }
 }
@@ -285,13 +324,18 @@ impl<'ch, T> Unpin for DynamicSendFuture<'ch, T> {}
 trait DynamicChannel<T> {
     fn try_send_with_context(&self, message: T, cx: Option<&mut Context<'_>>) -> Result<(), TrySendError<T>>;
 
-    fn try_recv_with_context(&self, cx: Option<&mut Context<'_>>) -> Result<T, TryRecvError>;
+    fn try_receive_with_context(&self, cx: Option<&mut Context<'_>>) -> Result<T, TryReceiveError>;
+
+    fn poll_ready_to_send(&self, cx: &mut Context<'_>) -> Poll<()>;
+    fn poll_ready_to_receive(&self, cx: &mut Context<'_>) -> Poll<()>;
+
+    fn poll_receive(&self, cx: &mut Context<'_>) -> Poll<T>;
 }
 
-/// Error returned by [`try_recv`](Channel::try_recv).
+/// Error returned by [`try_receive`](Channel::try_receive).
 #[derive(PartialEq, Eq, Clone, Copy, Debug)]
 #[cfg_attr(feature = "defmt", derive(defmt::Format))]
-pub enum TryRecvError {
+pub enum TryReceiveError {
     /// A message could not be received because the channel is empty.
     Empty,
 }
@@ -320,11 +364,11 @@ impl<T, const N: usize> ChannelState<T, N> {
         }
     }
 
-    fn try_recv(&mut self) -> Result<T, TryRecvError> {
+    fn try_receive(&mut self) -> Result<T, TryReceiveError> {
-        self.try_recv_with_context(None)
+        self.try_receive_with_context(None)
     }
 
-    fn try_recv_with_context(&mut self, cx: Option<&mut Context<'_>>) -> Result<T, TryRecvError> {
+    fn try_receive_with_context(&mut self, cx: Option<&mut Context<'_>>) -> Result<T, TryReceiveError> {
         if self.queue.is_full() {
             self.senders_waker.wake();
         }
@@ -335,14 +379,31 @@ impl<T, const N: usize> ChannelState<T, N> {
             if let Some(cx) = cx {
                 self.receiver_waker.register(cx.waker());
             }
-            Err(TryRecvError::Empty)
+            Err(TryReceiveError::Empty)
         }
     }
 
-    fn poll_ready_to_receive(&mut self, cx: &mut Context<'_>) -> bool {
+    fn poll_receive(&mut self, cx: &mut Context<'_>) -> Poll<T> {
+        if self.queue.is_full() {
+            self.senders_waker.wake();
+        }
+
+        if let Some(message) = self.queue.pop_front() {
+            Poll::Ready(message)
+        } else {
+            self.receiver_waker.register(cx.waker());
+            Poll::Pending
+        }
+    }
+
+    fn poll_ready_to_receive(&mut self, cx: &mut Context<'_>) -> Poll<()> {
         self.receiver_waker.register(cx.waker());
 
-        !self.queue.is_empty()
+        if !self.queue.is_empty() {
+            Poll::Ready(())
+        } else {
+            Poll::Pending
+        }
     }
 
     fn try_send(&mut self, message: T) -> Result<(), TrySendError<T>> {
@@ -364,10 +425,14 @@ impl<T, const N: usize> ChannelState<T, N> {
         }
     }
 
-    fn poll_ready_to_send(&mut self, cx: &mut Context<'_>) -> bool {
+    fn poll_ready_to_send(&mut self, cx: &mut Context<'_>) -> Poll<()> {
         self.senders_waker.register(cx.waker());
 
-        !self.queue.is_full()
+        if !self.queue.is_full() {
+            Poll::Ready(())
+        } else {
+            Poll::Pending
+        }
     }
 }
 
@@ -409,8 +474,13 @@ where
         self.inner.lock(|rc| f(&mut *rc.borrow_mut()))
     }
 
-    fn try_recv_with_context(&self, cx: Option<&mut Context<'_>>) -> Result<T, TryRecvError> {
+    fn try_receive_with_context(&self, cx: Option<&mut Context<'_>>) -> Result<T, TryReceiveError> {
-        self.lock(|c| c.try_recv_with_context(cx))
+        self.lock(|c| c.try_receive_with_context(cx))
+    }
+
+    /// Poll the channel for the next message
+    pub fn poll_receive(&self, cx: &mut Context<'_>) -> Poll<T> {
+        self.lock(|c| c.poll_receive(cx))
     }
 
     fn try_send_with_context(&self, m: T, cx: Option<&mut Context<'_>>) -> Result<(), TrySendError<T>> {
@@ -418,12 +488,12 @@ where
     }
 
     /// Allows a poll_fn to poll until the channel is ready to receive
-    pub fn poll_ready_to_receive(&self, cx: &mut Context<'_>) -> bool {
+    pub fn poll_ready_to_receive(&self, cx: &mut Context<'_>) -> Poll<()> {
         self.lock(|c| c.poll_ready_to_receive(cx))
     }
 
     /// Allows a poll_fn to poll until the channel is ready to send
-    pub fn poll_ready_to_send(&self, cx: &mut Context<'_>) -> bool {
+    pub fn poll_ready_to_send(&self, cx: &mut Context<'_>) -> Poll<()> {
         self.lock(|c| c.poll_ready_to_send(cx))
     }
 
@@ -466,16 +536,16 @@ where
     ///
     /// 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> {
+    pub fn receive(&self) -> ReceiveFuture<'_, M, T, N> {
-        RecvFuture { channel: self }
+        ReceiveFuture { 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<T, TryRecvError> {
+    pub fn try_receive(&self) -> Result<T, TryReceiveError> {
-        self.lock(|c| c.try_recv())
+        self.lock(|c| c.try_receive())
     }
 }
 
@@ -489,8 +559,20 @@ where
         Channel::try_send_with_context(self, m, cx)
     }
 
-    fn try_recv_with_context(&self, cx: Option<&mut Context<'_>>) -> Result<T, TryRecvError> {
+    fn try_receive_with_context(&self, cx: Option<&mut Context<'_>>) -> Result<T, TryReceiveError> {
-        Channel::try_recv_with_context(self, cx)
+        Channel::try_receive_with_context(self, cx)
+    }
+
+    fn poll_ready_to_send(&self, cx: &mut Context<'_>) -> Poll<()> {
+        Channel::poll_ready_to_send(self, cx)
+    }
+
+    fn poll_ready_to_receive(&self, cx: &mut Context<'_>) -> Poll<()> {
+        Channel::poll_ready_to_receive(self, cx)
+    }
+
+    fn poll_receive(&self, cx: &mut Context<'_>) -> Poll<T> {
+        Channel::poll_receive(self, cx)
     }
 }
 
@@ -534,15 +616,15 @@ mod tests {
     fn receiving_once_with_one_send() {
         let mut c = ChannelState::<u32, 3>::new();
         assert!(c.try_send(1).is_ok());
-        assert_eq!(c.try_recv().unwrap(), 1);
+        assert_eq!(c.try_receive().unwrap(), 1);
         assert_eq!(capacity(&c), 3);
     }
 
     #[test]
     fn receiving_when_empty() {
         let mut c = ChannelState::<u32, 3>::new();
-        match c.try_recv() {
+        match c.try_receive() {
-            Err(TryRecvError::Empty) => assert!(true),
+            Err(TryReceiveError::Empty) => assert!(true),
             _ => assert!(false),
         }
         assert_eq!(capacity(&c), 3);
@@ -552,7 +634,7 @@ mod tests {
     fn simple_send_and_receive() {
         let c = Channel::<NoopRawMutex, u32, 3>::new();
         assert!(c.try_send(1).is_ok());
-        assert_eq!(c.try_recv().unwrap(), 1);
+        assert_eq!(c.try_receive().unwrap(), 1);
     }
 
     #[test]
@@ -572,7 +654,7 @@ mod tests {
         let r: DynamicReceiver<'_, u32> = c.receiver().into();
 
         assert!(s.try_send(1).is_ok());
-        assert_eq!(r.try_recv().unwrap(), 1);
+        assert_eq!(r.try_receive().unwrap(), 1);
     }
 
     #[futures_test::test]
@@ -587,14 +669,14 @@ mod tests {
                 assert!(c2.try_send(1).is_ok());
             })
             .is_ok());
-        assert_eq!(c.recv().await, 1);
+        assert_eq!(c.receive().await, 1);
     }
 
     #[futures_test::test]
     async fn sender_send_completes_if_capacity() {
         let c = Channel::<CriticalSectionRawMutex, u32, 1>::new();
         c.send(1).await;
-        assert_eq!(c.recv().await, 1);
+        assert_eq!(c.receive().await, 1);
     }
 
     #[futures_test::test]
@@ -612,11 +694,11 @@ mod tests {
         // 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_eq!(c.receive().await, 1);
         assert!(executor
             .spawn(async move {
                 loop {
-                    c.recv().await;
+                    c.receive().await;
                 }
             })
             .is_ok());

examples/nrf52840/src/bin/channel.rs

@@ -35,7 +35,7 @@ async fn main(spawner: Spawner) {
     unwrap!(spawner.spawn(my_task()));
 
     loop {
-        match CHANNEL.recv().await {
+        match CHANNEL.receive().await {
             LedState::On => led.set_high(),
             LedState::Off => led.set_low(),
         }

examples/nrf52840/src/bin/channel_sender_receiver.rs

@@ -33,7 +33,7 @@ async fn recv_task(led: AnyPin, receiver: Receiver<'static, NoopRawMutex, LedSta
     let mut led = Output::new(led, Level::Low, OutputDrive::Standard);
 
     loop {
-        match receiver.recv().await {
+        match receiver.receive().await {
             LedState::On => led.set_high(),
             LedState::Off => led.set_low(),
         }

examples/nrf52840/src/bin/uart_split.rs

@@ -46,7 +46,7 @@ async fn main(spawner: Spawner) {
     // back out the buffer we receive from the read
     // task.
     loop {
-        let buf = CHANNEL.recv().await;
+        let buf = CHANNEL.receive().await;
         info!("writing...");
         unwrap!(tx.write(&buf).await);
     }

examples/rp/src/bin/lora_p2p_send_multicore.rs

@@ -113,7 +113,7 @@ async fn core1_task(
     };
 
     loop {
-        let buffer: [u8; 3] = CHANNEL.recv().await;
+        let buffer: [u8; 3] = CHANNEL.receive().await;
         match lora.prepare_for_tx(&mdltn_params, 20, false).await {
             Ok(()) => {}
             Err(err) => {

examples/rp/src/bin/multicore.rs

@@ -56,7 +56,7 @@ async fn core0_task() {
 async fn core1_task(mut led: Output<'static, PIN_25>) {
     info!("Hello from core 1");
     loop {
-        match CHANNEL.recv().await {
+        match CHANNEL.receive().await {
             LedState::On => led.set_high(),
             LedState::Off => led.set_low(),
         }

examples/stm32f3/src/bin/button_events.rs

@@ -49,12 +49,12 @@ impl<'a> Leds<'a> {
 
     async fn show(&mut self) {
         self.leds[self.current_led].set_high();
-        if let Ok(new_message) = with_timeout(Duration::from_millis(500), CHANNEL.recv()).await {
+        if let Ok(new_message) = with_timeout(Duration::from_millis(500), CHANNEL.receive()).await {
             self.leds[self.current_led].set_low();
             self.process_event(new_message).await;
         } else {
             self.leds[self.current_led].set_low();
-            if let Ok(new_message) = with_timeout(Duration::from_millis(200), CHANNEL.recv()).await {
+            if let Ok(new_message) = with_timeout(Duration::from_millis(200), CHANNEL.receive()).await {
                 self.process_event(new_message).await;
             }
         }

examples/stm32h5/src/bin/usart_split.rs

@@ -44,7 +44,7 @@ async fn main(spawner: Spawner) -> ! {
     unwrap!(spawner.spawn(reader(rx)));
 
     loop {
-        let buf = CHANNEL.recv().await;
+        let buf = CHANNEL.receive().await;
         info!("writing...");
         unwrap!(tx.write(&buf).await);
     }

examples/stm32h7/src/bin/usart_split.rs

@@ -44,7 +44,7 @@ async fn main(spawner: Spawner) -> ! {
     unwrap!(spawner.spawn(reader(rx)));
 
     loop {
-        let buf = CHANNEL.recv().await;
+        let buf = CHANNEL.receive().await;
         info!("writing...");
         unwrap!(tx.write(&buf).await);
     }

tests/rp/src/bin/gpio_multicore.rs

@@ -37,11 +37,11 @@ async fn core0_task(p: PIN_0) {
     let mut pin = Output::new(p, Level::Low);
 
     CHANNEL0.send(()).await;
-    CHANNEL1.recv().await;
+    CHANNEL1.receive().await;
 
     pin.set_high();
 
-    CHANNEL1.recv().await;
+    CHANNEL1.receive().await;
 
     info!("Test OK");
     cortex_m::asm::bkpt();
@@ -51,7 +51,7 @@ async fn core0_task(p: PIN_0) {
 async fn core1_task(p: PIN_1) {
     info!("CORE1 is running");
 
-    CHANNEL0.recv().await;
+    CHANNEL0.receive().await;
 
     let mut pin = Input::new(p, Pull::Down);
     let wait = pin.wait_for_rising_edge();

tests/rp/src/bin/multicore.rs

@@ -33,7 +33,7 @@ async fn core0_task() {
     info!("CORE0 is running");
     let ping = true;
     CHANNEL0.send(ping).await;
-    let pong = CHANNEL1.recv().await;
+    let pong = CHANNEL1.receive().await;
     assert_eq!(ping, pong);
 
     info!("Test OK");
@@ -43,6 +43,6 @@ async fn core0_task() {
 #[embassy_executor::task]
 async fn core1_task() {
     info!("CORE1 is running");
-    let ping = CHANNEL0.recv().await;
+    let ping = CHANNEL0.receive().await;
     CHANNEL1.send(ping).await;
 }