diff --git a/embassy-stm32-wpan/src/mac/driver.rs b/embassy-stm32-wpan/src/mac/driver.rs index f8e3a2b0..bfc4f1ee 100644 --- a/embassy-stm32-wpan/src/mac/driver.rs +++ b/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> { - 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 diff --git a/embassy-stm32-wpan/src/mac/runner.rs b/embassy-stm32-wpan/src/mac/runner.rs index 1be6df8a..d3099b6b 100644 --- a/embassy-stm32-wpan/src/mac/runner.rs +++ b/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 diff --git a/embassy-stm32/src/can/bxcan.rs b/embassy-stm32/src/can/bxcan.rs index fb223e4a..7ad13cec 100644 --- a/embassy-stm32/src/can/bxcan.rs +++ b/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 { 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 { - 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| { - if T::state().rx_queue.poll_ready_to_receive(cx) { - Poll::Ready(()) - } else { - Poll::Pending - } - }) - .await + poll_fn(|cx| T::state().rx_queue.poll_ready_to_receive(cx)).await } fn curr_error(&self) -> Option { diff --git a/embassy-sync/src/channel.rs b/embassy-sync/src/channel.rs index d6f36f53..62ea1307 100644 --- a/embassy-sync/src/channel.rs +++ b/embassy-sync/src/channel.rs @@ -65,6 +65,13 @@ where pub fn try_send(&self, message: T) -> Result<(), TrySendError> { 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> { 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()`]. - pub fn recv(&self) -> RecvFuture<'_, M, T, N> { - self.channel.recv() + /// See [`Channel::receive()`]. + pub fn receive(&self) -> ReceiveFuture<'_, M, T, N> { + self.channel.receive() } /// Attempt to immediately receive the next value. /// - /// See [`Channel::try_recv()`] - pub fn try_recv(&self) -> Result { - self.channel.try_recv() + /// See [`Channel::try_receive()`] + pub fn try_receive(&self) -> Result { + 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 { + 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()`]. - pub fn recv(&self) -> DynamicRecvFuture<'_, T> { - DynamicRecvFuture { channel: self.channel } + /// See [`Channel::receive()`]. + pub fn receive(&self) -> DynamicReceiveFuture<'_, T> { + DynamicReceiveFuture { 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) + /// See [`Channel::try_receive()`] + pub fn try_receive(&self) -> Result { + 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 { + 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, } -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 { - match self.channel.try_recv_with_context(Some(cx)) { - Ok(v) => Poll::Ready(v), - Err(TryRecvError::Empty) => Poll::Pending, - } + self.channel.poll_receive(cx) } } -/// 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, } -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 { - 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 { 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; + fn try_receive_with_context(&self, cx: Option<&mut Context<'_>>) -> Result; + + 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; } -/// 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 ChannelState { } } - fn try_recv(&mut self) -> Result { - self.try_recv_with_context(None) + fn try_receive(&mut self) -> Result { + self.try_receive_with_context(None) } - fn try_recv_with_context(&mut self, cx: Option<&mut Context<'_>>) -> Result { + fn try_receive_with_context(&mut self, cx: Option<&mut Context<'_>>) -> Result { if self.queue.is_full() { self.senders_waker.wake(); } @@ -335,14 +379,31 @@ impl ChannelState { 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 { + 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> { @@ -364,10 +425,14 @@ impl ChannelState { } } - 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 { - self.lock(|c| c.try_recv_with_context(cx)) + fn try_receive_with_context(&self, cx: Option<&mut Context<'_>>) -> Result { + 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 { + self.lock(|c| c.poll_receive(cx)) } fn try_send_with_context(&self, m: T, cx: Option<&mut Context<'_>>) -> Result<(), TrySendError> { @@ -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> { - RecvFuture { channel: self } + pub fn receive(&self) -> ReceiveFuture<'_, M, T, N> { + 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 { - self.lock(|c| c.try_recv()) + pub fn try_receive(&self) -> Result { + 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 { - Channel::try_recv_with_context(self, cx) + fn try_receive_with_context(&self, cx: Option<&mut Context<'_>>) -> Result { + 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 { + Channel::poll_receive(self, cx) } } @@ -534,15 +616,15 @@ mod tests { 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!(c.try_receive().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), + match c.try_receive() { + 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::::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::::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()); diff --git a/examples/nrf52840/src/bin/channel.rs b/examples/nrf52840/src/bin/channel.rs index d782a79e..bd9c909d 100644 --- a/examples/nrf52840/src/bin/channel.rs +++ b/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(), } diff --git a/examples/nrf52840/src/bin/channel_sender_receiver.rs b/examples/nrf52840/src/bin/channel_sender_receiver.rs index fcccdaed..ec4f1d80 100644 --- a/examples/nrf52840/src/bin/channel_sender_receiver.rs +++ b/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(), } diff --git a/examples/nrf52840/src/bin/uart_split.rs b/examples/nrf52840/src/bin/uart_split.rs index 9979a1d5..b748bfcd 100644 --- a/examples/nrf52840/src/bin/uart_split.rs +++ b/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); } diff --git a/examples/rp/src/bin/lora_p2p_send_multicore.rs b/examples/rp/src/bin/lora_p2p_send_multicore.rs index 89a62818..b54cc92f 100644 --- a/examples/rp/src/bin/lora_p2p_send_multicore.rs +++ b/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) => { diff --git a/examples/rp/src/bin/multicore.rs b/examples/rp/src/bin/multicore.rs index 893b724b..bf017f6a 100644 --- a/examples/rp/src/bin/multicore.rs +++ b/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(), } diff --git a/examples/stm32f3/src/bin/button_events.rs b/examples/stm32f3/src/bin/button_events.rs index 02c475f6..8e97e85e 100644 --- a/examples/stm32f3/src/bin/button_events.rs +++ b/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; } } diff --git a/examples/stm32h5/src/bin/usart_split.rs b/examples/stm32h5/src/bin/usart_split.rs index debd6f45..a6b2e690 100644 --- a/examples/stm32h5/src/bin/usart_split.rs +++ b/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); } diff --git a/examples/stm32h7/src/bin/usart_split.rs b/examples/stm32h7/src/bin/usart_split.rs index 330d1ce0..aa075345 100644 --- a/examples/stm32h7/src/bin/usart_split.rs +++ b/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); } diff --git a/tests/rp/src/bin/gpio_multicore.rs b/tests/rp/src/bin/gpio_multicore.rs index 22be7824..6ab7f671 100644 --- a/tests/rp/src/bin/gpio_multicore.rs +++ b/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(); diff --git a/tests/rp/src/bin/multicore.rs b/tests/rp/src/bin/multicore.rs index ec794c48..f4188135 100644 --- a/tests/rp/src/bin/multicore.rs +++ b/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; }