Merge #695

695: Simplify Channel. r=Dirbaio a=Dirbaio

- Allow initializing in a static, without Forever.
- Remove ability to close, since in embedded enviromnents channels usually live forever and don't get closed.
- Remove MPSC restriction, it's MPMC now. Rename "mpsc" to "channel".
- `Sender` and `Receiver` are still available if you want to enforce a piece of code only has send/receive access, but are optional: you can send/receive directly into the Channel if you want.

Co-authored-by: Dario Nieuwenhuis <[email protected]>

6 files changed, 139 insertions, 126 deletions

diff --git a/examples/nrf/src/bin/channel.rs b/examples/nrf/src/bin/channel.rs
new file mode 100644
index 000000000..476ec09a1
--- /dev/null
+++ b/examples/nrf/src/bin/channel.rs
@@ -0,0 +1,45 @@
+#![no_std]
+#![no_main]
+#![feature(type_alias_impl_trait)]
+
+use defmt::unwrap;
+use embassy::blocking_mutex::raw::ThreadModeRawMutex;
+use embassy::channel::channel::Channel;
+use embassy::executor::Spawner;
+use embassy::time::{Duration, Timer};
+use embassy_nrf::gpio::{Level, Output, OutputDrive};
+use embassy_nrf::Peripherals;
+
+use defmt_rtt as _; // global logger
+use panic_probe as _;
+
+enum LedState {
+    On,
+    Off,
+}
+
+static CHANNEL: Channel<ThreadModeRawMutex, LedState, 1> = Channel::new();
+
+#[embassy::task]
+async fn my_task() {
+    loop {
+        CHANNEL.send(LedState::On).await;
+        Timer::after(Duration::from_secs(1)).await;
+        CHANNEL.send(LedState::Off).await;
+        Timer::after(Duration::from_secs(1)).await;
+    }
+}
+
+#[embassy::main]
+async fn main(spawner: Spawner, p: Peripherals) {
+    let mut led = Output::new(p.P0_13, Level::Low, OutputDrive::Standard);
+
+    unwrap!(spawner.spawn(my_task()));
+
+    loop {
+        match CHANNEL.recv().await {
+            LedState::On => led.set_high(),
+            LedState::Off => led.set_low(),
+        }
+    }
+}
diff --git a/examples/nrf/src/bin/channel_sender_receiver.rs b/examples/nrf/src/bin/channel_sender_receiver.rs
new file mode 100644
index 000000000..c79f2fd6b
--- /dev/null
+++ b/examples/nrf/src/bin/channel_sender_receiver.rs
@@ -0,0 +1,52 @@
+#![no_std]
+#![no_main]
+#![feature(type_alias_impl_trait)]
+
+use defmt::unwrap;
+use embassy::blocking_mutex::raw::NoopRawMutex;
+use embassy::channel::channel::{Channel, Receiver, Sender};
+use embassy::executor::Spawner;
+use embassy::time::{Duration, Timer};
+use embassy::util::Forever;
+use embassy_nrf::gpio::{AnyPin, Level, Output, OutputDrive, Pin};
+use embassy_nrf::Peripherals;
+
+use defmt_rtt as _; // global logger
+use panic_probe as _;
+
+enum LedState {
+    On,
+    Off,
+}
+
+static CHANNEL: Forever<Channel<NoopRawMutex, LedState, 1>> = Forever::new();
+
+#[embassy::task]
+async fn send_task(sender: Sender<'static, NoopRawMutex, LedState, 1>) {
+    loop {
+        sender.send(LedState::On).await;
+        Timer::after(Duration::from_secs(1)).await;
+        sender.send(LedState::Off).await;
+        Timer::after(Duration::from_secs(1)).await;
+    }
+}
+
+#[embassy::task]
+async fn recv_task(led: AnyPin, receiver: Receiver<'static, NoopRawMutex, LedState, 1>) {
+    let mut led = Output::new(led, Level::Low, OutputDrive::Standard);
+
+    loop {
+        match receiver.recv().await {
+            LedState::On => led.set_high(),
+            LedState::Off => led.set_low(),
+        }
+    }
+}
+
+#[embassy::main]
+async fn main(spawner: Spawner, p: Peripherals) {
+    let channel = CHANNEL.put(Channel::new());
+
+    unwrap!(spawner.spawn(send_task(channel.sender())));
+    unwrap!(spawner.spawn(recv_task(p.P0_13.degrade(), channel.receiver())));
+}
diff --git a/examples/nrf/src/bin/mpsc.rs b/examples/nrf/src/bin/mpsc.rs
deleted file mode 100644
index 0cb182755..000000000
--- a/examples/nrf/src/bin/mpsc.rs
+++ /dev/null
@@ -1,60 +0,0 @@
-#![no_std]
-#![no_main]
-#![feature(type_alias_impl_trait)]
-
-use defmt::unwrap;
-use embassy::blocking_mutex::raw::NoopRawMutex;
-use embassy::channel::mpsc::{self, Channel, Sender, TryRecvError};
-use embassy::executor::Spawner;
-use embassy::time::{Duration, Timer};
-use embassy::util::Forever;
-use embassy_nrf::gpio::{Level, Output, OutputDrive};
-use embassy_nrf::Peripherals;
-
-use defmt_rtt as _; // global logger
-use panic_probe as _;
-
-enum LedState {
-    On,
-    Off,
-}
-
-static CHANNEL: Forever<Channel<NoopRawMutex, LedState, 1>> = Forever::new();
-
-#[embassy::task(pool_size = 1)]
-async fn my_task(sender: Sender<'static, NoopRawMutex, LedState, 1>) {
-    loop {
-        let _ = sender.send(LedState::On).await;
-        Timer::after(Duration::from_secs(1)).await;
-        let _ = sender.send(LedState::Off).await;
-        Timer::after(Duration::from_secs(1)).await;
-    }
-}
-
-#[embassy::main]
-async fn main(spawner: Spawner, p: Peripherals) {
-    let mut led = Output::new(p.P0_13, Level::Low, OutputDrive::Standard);
-
-    let channel = CHANNEL.put(Channel::new());
-    let (sender, mut receiver) = mpsc::split(channel);
-
-    unwrap!(spawner.spawn(my_task(sender)));
-
-    // We could just loop on `receiver.recv()` for simplicity. The code below
-    // is optimized to drain the queue as fast as possible in the spirit of
-    // handling events as fast as possible. This optimization is benign when in
-    // thread mode, but can be useful when interrupts are sending messages
-    // with the channel having been created via with_critical_sections.
-    loop {
-        let maybe_message = match receiver.try_recv() {
-            m @ Ok(..) => m.ok(),
-            Err(TryRecvError::Empty) => receiver.recv().await,
-            Err(TryRecvError::Closed) => break,
-        };
-        match maybe_message {
-            Some(LedState::On) => led.set_high(),
-            Some(LedState::Off) => led.set_low(),
-            _ => (),
-        }
-    }
-}
diff --git a/examples/nrf/src/bin/uart_split.rs b/examples/nrf/src/bin/uart_split.rs
index 909429b1a..3fde2f0d8 100644
--- a/examples/nrf/src/bin/uart_split.rs
+++ b/examples/nrf/src/bin/uart_split.rs
@@ -3,10 +3,9 @@
 #![feature(type_alias_impl_trait)]
 
 use defmt::*;
-use embassy::blocking_mutex::raw::NoopRawMutex;
-use embassy::channel::mpsc::{self, Channel, Sender};
+use embassy::blocking_mutex::raw::ThreadModeRawMutex;
+use embassy::channel::channel::Channel;
 use embassy::executor::Spawner;
-use embassy::util::Forever;
 use embassy_nrf::peripherals::UARTE0;
 use embassy_nrf::uarte::UarteRx;
 use embassy_nrf::{interrupt, uarte, Peripherals};
@@ -14,7 +13,7 @@ use embassy_nrf::{interrupt, uarte, Peripherals};
 use defmt_rtt as _; // global logger
 use panic_probe as _;
 
-static CHANNEL: Forever<Channel<NoopRawMutex, [u8; 8], 1>> = Forever::new();
+static CHANNEL: Channel<ThreadModeRawMutex, [u8; 8], 1> = Channel::new();
 
 #[embassy::main]
 async fn main(spawner: Spawner, p: Peripherals) {
@@ -26,14 +25,11 @@ async fn main(spawner: Spawner, p: Peripherals) {
     let uart = uarte::Uarte::new(p.UARTE0, irq, p.P0_08, p.P0_06, config);
     let (mut tx, rx) = uart.split();
 
-    let c = CHANNEL.put(Channel::new());
-    let (s, mut r) = mpsc::split(c);
-
     info!("uarte initialized!");
 
     // Spawn a task responsible purely for reading
 
-    unwrap!(spawner.spawn(reader(rx, s)));
+    unwrap!(spawner.spawn(reader(rx)));
 
     // Message must be in SRAM
     {
@@ -48,19 +44,18 @@ async fn main(spawner: Spawner, p: Peripherals) {
     // back out the buffer we receive from the read
     // task.
     loop {
-        if let Some(buf) = r.recv().await {
-            info!("writing...");
-            unwrap!(tx.write(&buf).await);
-        }
+        let buf = CHANNEL.recv().await;
+        info!("writing...");
+        unwrap!(tx.write(&buf).await);
     }
 }
 
 #[embassy::task]
-async fn reader(mut rx: UarteRx<'static, UARTE0>, s: Sender<'static, NoopRawMutex, [u8; 8], 1>) {
+async fn reader(mut rx: UarteRx<'static, UARTE0>) {
     let mut buf = [0; 8];
     loop {
         info!("reading...");
         unwrap!(rx.read(&mut buf).await);
-        unwrap!(s.send(buf).await);
+        CHANNEL.send(buf).await;
     }
 }
diff --git a/examples/stm32f3/src/bin/button_events.rs b/examples/stm32f3/src/bin/button_events.rs
index 99aab3027..06e8eec1f 100644
--- a/examples/stm32f3/src/bin/button_events.rs
+++ b/examples/stm32f3/src/bin/button_events.rs
@@ -11,11 +11,10 @@
 #![feature(type_alias_impl_trait)]
 
 use defmt::*;
-use embassy::blocking_mutex::raw::NoopRawMutex;
-use embassy::channel::mpsc::{self, Channel, Receiver, Sender};
+use embassy::blocking_mutex::raw::ThreadModeRawMutex;
+use embassy::channel::channel::Channel;
 use embassy::executor::Spawner;
 use embassy::time::{with_timeout, Duration, Timer};
-use embassy::util::Forever;
 use embassy_stm32::exti::ExtiInput;
 use embassy_stm32::gpio::{AnyPin, Input, Level, Output, Pin, Pull, Speed};
 use embassy_stm32::peripherals::PA0;
@@ -51,14 +50,15 @@ impl<'a> Leds<'a> {
         }
     }
 
-    async fn show(&mut self, queue: &mut Receiver<'static, NoopRawMutex, ButtonEvent, 4>) {
+    async fn show(&mut self) {
         self.leds[self.current_led].set_high();
-        if let Ok(new_message) = with_timeout(Duration::from_millis(500), queue.recv()).await {
+        if let Ok(new_message) = with_timeout(Duration::from_millis(500), CHANNEL.recv()).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), queue.recv()).await {
+            if let Ok(new_message) = with_timeout(Duration::from_millis(200), CHANNEL.recv()).await
+            {
                 self.process_event(new_message).await;
             }
         }
@@ -77,15 +77,18 @@ impl<'a> Leds<'a> {
         }
     }
 
-    async fn process_event(&mut self, event: Option<ButtonEvent>) {
+    async fn process_event(&mut self, event: ButtonEvent) {
         match event {
-            Some(ButtonEvent::SingleClick) => self.move_next(),
-            Some(ButtonEvent::DoubleClick) => {
+            ButtonEvent::SingleClick => {
+                self.move_next();
+            }
+            ButtonEvent::DoubleClick => {
                 self.change_direction();
-                self.move_next()
+                self.move_next();
+            }
+            ButtonEvent::Hold => {
+                self.flash().await;
             }
-            Some(ButtonEvent::Hold) => self.flash().await,
-            _ => {}
         }
     }
 }
@@ -97,7 +100,7 @@ enum ButtonEvent {
     Hold,
 }
 
-static BUTTON_EVENTS_QUEUE: Forever<Channel<NoopRawMutex, ButtonEvent, 4>> = Forever::new();
+static CHANNEL: Channel<ThreadModeRawMutex, ButtonEvent, 4> = Channel::new();
 
 #[embassy::main]
 async fn main(spawner: Spawner, p: Peripherals) {
@@ -116,27 +119,19 @@ async fn main(spawner: Spawner, p: Peripherals) {
     ];
     let leds = Leds::new(leds);
 
-    let buttons_queue = BUTTON_EVENTS_QUEUE.put(Channel::new());
-    let (sender, receiver) = mpsc::split(buttons_queue);
-    spawner.spawn(button_waiter(button, sender)).unwrap();
-    spawner.spawn(led_blinker(leds, receiver)).unwrap();
+    spawner.spawn(button_waiter(button)).unwrap();
+    spawner.spawn(led_blinker(leds)).unwrap();
 }
 
 #[embassy::task]
-async fn led_blinker(
-    mut leds: Leds<'static>,
-    mut queue: Receiver<'static, NoopRawMutex, ButtonEvent, 4>,
-) {
+async fn led_blinker(mut leds: Leds<'static>) {
     loop {
-        leds.show(&mut queue).await;
+        leds.show().await;
     }
 }
 
 #[embassy::task]
-async fn button_waiter(
-    mut button: ExtiInput<'static, PA0>,
-    queue: Sender<'static, NoopRawMutex, ButtonEvent, 4>,
-) {
+async fn button_waiter(mut button: ExtiInput<'static, PA0>) {
     const DOUBLE_CLICK_DELAY: u64 = 250;
     const HOLD_DELAY: u64 = 1000;
 
@@ -150,9 +145,7 @@ async fn button_waiter(
         .is_err()
         {
             info!("Hold");
-            if queue.send(ButtonEvent::Hold).await.is_err() {
-                break;
-            }
+            CHANNEL.send(ButtonEvent::Hold).await;
             button.wait_for_falling_edge().await;
         } else if with_timeout(
             Duration::from_millis(DOUBLE_CLICK_DELAY),
@@ -161,15 +154,11 @@ async fn button_waiter(
         .await
         .is_err()
         {
-            if queue.send(ButtonEvent::SingleClick).await.is_err() {
-                break;
-            }
             info!("Single click");
+            CHANNEL.send(ButtonEvent::SingleClick).await;
         } else {
             info!("Double click");
-            if queue.send(ButtonEvent::DoubleClick).await.is_err() {
-                break;
-            }
+            CHANNEL.send(ButtonEvent::DoubleClick).await;
             button.wait_for_falling_edge().await;
         }
         button.wait_for_rising_edge().await;
diff --git a/examples/stm32h7/src/bin/usart_split.rs b/examples/stm32h7/src/bin/usart_split.rs
index ee1763aa4..40a7c3e44 100644
--- a/examples/stm32h7/src/bin/usart_split.rs
+++ b/examples/stm32h7/src/bin/usart_split.rs
@@ -4,10 +4,9 @@
 
 use defmt::*;
 use defmt_rtt as _; // global logger
-use embassy::blocking_mutex::raw::NoopRawMutex;
-use embassy::channel::mpsc::{self, Channel, Sender};
+use embassy::blocking_mutex::raw::ThreadModeRawMutex;
+use embassy::channel::channel::Channel;
 use embassy::executor::Spawner;
-use embassy::util::Forever;
 use embassy_stm32::dma::NoDma;
 use embassy_stm32::{
     peripherals::{DMA1_CH1, UART7},
@@ -28,7 +27,7 @@ async fn writer(mut usart: Uart<'static, UART7, NoDma, NoDma>) {
     }
 }
 
-static CHANNEL: Forever<Channel<NoopRawMutex, [u8; 8], 1>> = Forever::new();
+static CHANNEL: Channel<ThreadModeRawMutex, [u8; 8], 1> = Channel::new();
 
 #[embassy::main]
 async fn main(spawner: Spawner, p: Peripherals) -> ! {
@@ -40,28 +39,21 @@ async fn main(spawner: Spawner, p: Peripherals) -> ! {
 
     let (mut tx, rx) = usart.split();
 
-    let c = CHANNEL.put(Channel::new());
-    let (s, mut r) = mpsc::split(c);
-
-    unwrap!(spawner.spawn(reader(rx, s)));
+    unwrap!(spawner.spawn(reader(rx)));
 
     loop {
-        if let Some(buf) = r.recv().await {
-            info!("writing...");
-            unwrap!(tx.write(&buf).await);
-        }
+        let buf = CHANNEL.recv().await;
+        info!("writing...");
+        unwrap!(tx.write(&buf).await);
     }
 }
 
 #[embassy::task]
-async fn reader(
-    mut rx: UartRx<'static, UART7, DMA1_CH1>,
-    s: Sender<'static, NoopRawMutex, [u8; 8], 1>,
-) {
+async fn reader(mut rx: UartRx<'static, UART7, DMA1_CH1>) {
     let mut buf = [0; 8];
     loop {
         info!("reading...");
         unwrap!(rx.read(&mut buf).await);
-        unwrap!(s.send(buf).await);
+        CHANNEL.send(buf).await;
     }
 }