Merge #544

544: Introduces split on the nRF Uarte r=Dirbaio a=huntc

A new `split` method is introduced such that the Uarte tx and rx can be used from separate tasks. An MPSC is used in an example to illustrate how data may be passed between these tasks.

The approach taken within the `Uarte` struct is to split into tx and rx fields on calling `Uarte::new`. These fields are returned given a call to `Uarte::split`, but otherwise, if that call isn't made, then the API remains as it was before.

Here's a snippet from a new example introduced:

```rust
#[embassy::main]
async fn main(spawner: Spawner, p: Peripherals) {
    // ...

    let uart = uarte::Uarte::new(p.UARTE0, irq, p.P0_08, p.P0_06, NoPin, NoPin, config);
    let (mut tx, rx) = uart.split();

    // ...

    // Spawn a task responsible purely for reading

    unwrap!(spawner.spawn(reader(rx, s)));

    // ...

    // Continue reading in this main task and write
    // 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);
        }
    }
}

#[embassy::task]
async fn reader(mut rx: UarteRx<'static, UARTE0>, s: Sender<'static, Noop, [u8; 8], 1>) {
    let mut buf = [0; 8];
    loop {
        info!("reading...");
        unwrap!(rx.read(&mut buf).await);
        unwrap!(s.send(buf).await);
    }
}
```


Co-authored-by: huntc <[email protected]>

2 files changed, 220 insertions, 63 deletions

diff --git a/embassy-nrf/src/uarte.rs b/embassy-nrf/src/uarte.rs
index ff781cabd..17417c0e2 100644
--- a/embassy-nrf/src/uarte.rs
+++ b/embassy-nrf/src/uarte.rs
@@ -54,6 +54,20 @@ impl Default for Config {
 /// Interface to the UARTE peripheral
 pub struct Uarte<'d, T: Instance> {
     phantom: PhantomData<&'d mut T>,
+    tx: UarteTx<'d, T>,
+    rx: UarteRx<'d, T>,
+}
+
+/// Transmitter interface to the UARTE peripheral obtained
+/// via [Uarte]::split.
+pub struct UarteTx<'d, T: Instance> {
+    phantom: PhantomData<&'d mut T>,
+}
+
+/// Receiver interface to the UARTE peripheral obtained
+/// via [Uarte]::split.
+pub struct UarteRx<'d, T: Instance> {
+    phantom: PhantomData<&'d mut T>,
 }
 
 impl<'d, T: Instance> Uarte<'d, T> {
@@ -119,11 +133,24 @@ impl<'d, T: Instance> Uarte<'d, T> {
         apply_workaround_for_enable_anomaly(&r);
         r.enable.write(|w| w.enable().enabled());
 
+        let s = T::state();
+
+        s.tx_rx_refcount.store(2, Ordering::Relaxed);
+
         Self {
             phantom: PhantomData,
+            tx: UarteTx::new(),
+            rx: UarteRx::new(),
         }
     }
 
+    /// Split the Uarte into a transmitter and receiver, which is
+    /// particuarly useful when having two tasks correlating to
+    /// transmitting and receiving.
+    pub fn split(self) -> (UarteTx<'d, T>, UarteRx<'d, T>) {
+        (self.tx, self.rx)
+    }
+
     fn on_interrupt(_: *mut ()) {
         let r = T::regs();
         let s = T::state();
@@ -139,72 +166,72 @@ impl<'d, T: Instance> Uarte<'d, T> {
     }
 }
 
-impl<'a, T: Instance> Drop for Uarte<'a, T> {
-    fn drop(&mut self) {
-        info!("uarte drop");
-
-        let r = T::regs();
-
-        let did_stoprx = r.events_rxstarted.read().bits() != 0;
-        let did_stoptx = r.events_txstarted.read().bits() != 0;
-        info!("did_stoprx {} did_stoptx {}", did_stoprx, did_stoptx);
+impl<'d, T: Instance> Read for Uarte<'d, T> {
+    #[rustfmt::skip]
+    type ReadFuture<'a> where Self: 'a = impl Future<Output = Result<(), TraitError>> + 'a;
 
-        // Wait for rxto or txstopped, if needed.
-        while (did_stoprx && r.events_rxto.read().bits() == 0)
-            || (did_stoptx && r.events_txstopped.read().bits() == 0)
-        {}
+    fn read<'a>(&'a mut self, rx_buffer: &'a mut [u8]) -> Self::ReadFuture<'a> {
+        self.rx.read(rx_buffer)
+    }
+}
 
-        // Finally we can disable!
-        r.enable.write(|w| w.enable().disabled());
+impl<'d, T: Instance> Write for Uarte<'d, T> {
+    #[rustfmt::skip]
+    type WriteFuture<'a> where Self: 'a = impl Future<Output = Result<(), TraitError>> + 'a;
 
-        gpio::deconfigure_pin(r.psel.rxd.read().bits());
-        gpio::deconfigure_pin(r.psel.txd.read().bits());
-        gpio::deconfigure_pin(r.psel.rts.read().bits());
-        gpio::deconfigure_pin(r.psel.cts.read().bits());
+    fn write<'a>(&'a mut self, tx_buffer: &'a [u8]) -> Self::WriteFuture<'a> {
+        self.tx.write(tx_buffer)
+    }
+}
 
-        info!("uarte drop: done");
+impl<'d, T: Instance> UarteTx<'d, T> {
+    pub fn new() -> Self {
+        Self {
+            phantom: PhantomData,
+        }
     }
 }
 
-impl<'d, T: Instance> Read for Uarte<'d, T> {
+impl<'d, T: Instance> Write for UarteTx<'d, T> {
     #[rustfmt::skip]
-    type ReadFuture<'a> where Self: 'a = impl Future<Output = Result<(), TraitError>> + 'a;
+    type WriteFuture<'a> where Self: 'a = impl Future<Output = Result<(), TraitError>> + 'a;
 
-    fn read<'a>(&'a mut self, rx_buffer: &'a mut [u8]) -> Self::ReadFuture<'a> {
+    fn write<'a>(&'a mut self, tx_buffer: &'a [u8]) -> Self::WriteFuture<'a> {
         async move {
-            let ptr = rx_buffer.as_ptr();
-            let len = rx_buffer.len();
+            let ptr = tx_buffer.as_ptr();
+            let len = tx_buffer.len();
             assert!(len <= EASY_DMA_SIZE);
+            // TODO: panic if buffer is not in SRAM
 
             let r = T::regs();
             let s = T::state();
 
             let drop = OnDrop::new(move || {
-                info!("read drop: stopping");
-
-                r.intenclr.write(|w| w.endrx().clear());
-                r.events_rxto.reset();
-                r.tasks_stoprx.write(|w| unsafe { w.bits(1) });
+                info!("write drop: stopping");
 
-                while r.events_endrx.read().bits() == 0 {}
+                r.intenclr.write(|w| w.endtx().clear());
+                r.events_txstopped.reset();
+                r.tasks_stoptx.write(|w| unsafe { w.bits(1) });
 
-                info!("read drop: stopped");
+                // TX is stopped almost instantly, spinning is fine.
+                while r.events_endtx.read().bits() == 0 {}
+                info!("write drop: stopped");
             });
 
-            r.rxd.ptr.write(|w| unsafe { w.ptr().bits(ptr as u32) });
-            r.rxd.maxcnt.write(|w| unsafe { w.maxcnt().bits(len as _) });
+            r.txd.ptr.write(|w| unsafe { w.ptr().bits(ptr as u32) });
+            r.txd.maxcnt.write(|w| unsafe { w.maxcnt().bits(len as _) });
 
-            r.events_endrx.reset();
-            r.intenset.write(|w| w.endrx().set());
+            r.events_endtx.reset();
+            r.intenset.write(|w| w.endtx().set());
 
             compiler_fence(Ordering::SeqCst);
 
-            trace!("startrx");
-            r.tasks_startrx.write(|w| unsafe { w.bits(1) });
+            trace!("starttx");
+            r.tasks_starttx.write(|w| unsafe { w.bits(1) });
 
             poll_fn(|cx| {
-                s.endrx_waker.register(cx.waker());
-                if r.events_endrx.read().bits() != 0 {
+                s.endtx_waker.register(cx.waker());
+                if r.events_endtx.read().bits() != 0 {
                     return Poll::Ready(());
                 }
                 Poll::Pending
@@ -212,7 +239,7 @@ impl<'d, T: Instance> Read for Uarte<'d, T> {
             .await;
 
             compiler_fence(Ordering::SeqCst);
-            r.events_rxstarted.reset();
+            r.events_txstarted.reset();
             drop.defuse();
 
             Ok(())
@@ -220,46 +247,71 @@ impl<'d, T: Instance> Read for Uarte<'d, T> {
     }
 }
 
-impl<'d, T: Instance> Write for Uarte<'d, T> {
+impl<'a, T: Instance> Drop for UarteTx<'a, T> {
+    fn drop(&mut self) {
+        info!("uarte tx drop");
+
+        let r = T::regs();
+
+        let did_stoptx = r.events_txstarted.read().bits() != 0;
+        info!("did_stoptx {}", did_stoptx);
+
+        // Wait for txstopped, if needed.
+        while did_stoptx && r.events_txstopped.read().bits() == 0 {}
+
+        let s = T::state();
+
+        drop_tx_rx(&r, &s);
+    }
+}
+
+impl<'d, T: Instance> UarteRx<'d, T> {
+    pub fn new() -> Self {
+        Self {
+            phantom: PhantomData,
+        }
+    }
+}
+
+impl<'d, T: Instance> Read for UarteRx<'d, T> {
     #[rustfmt::skip]
-    type WriteFuture<'a> where Self: 'a = impl Future<Output = Result<(), TraitError>> + 'a;
+    type ReadFuture<'a> where Self: 'a = impl Future<Output = Result<(), TraitError>> + 'a;
 
-    fn write<'a>(&'a mut self, tx_buffer: &'a [u8]) -> Self::WriteFuture<'a> {
+    fn read<'a>(&'a mut self, rx_buffer: &'a mut [u8]) -> Self::ReadFuture<'a> {
         async move {
-            let ptr = tx_buffer.as_ptr();
-            let len = tx_buffer.len();
+            let ptr = rx_buffer.as_ptr();
+            let len = rx_buffer.len();
             assert!(len <= EASY_DMA_SIZE);
-            // TODO: panic if buffer is not in SRAM
 
             let r = T::regs();
             let s = T::state();
 
             let drop = OnDrop::new(move || {
-                info!("write drop: stopping");
+                info!("read drop: stopping");
 
-                r.intenclr.write(|w| w.endtx().clear());
-                r.events_txstopped.reset();
-                r.tasks_stoptx.write(|w| unsafe { w.bits(1) });
+                r.intenclr.write(|w| w.endrx().clear());
+                r.events_rxto.reset();
+                r.tasks_stoprx.write(|w| unsafe { w.bits(1) });
 
-                // TX is stopped almost instantly, spinning is fine.
-                while r.events_endtx.read().bits() == 0 {}
-                info!("write drop: stopped");
+                while r.events_endrx.read().bits() == 0 {}
+
+                info!("read drop: stopped");
             });
 
-            r.txd.ptr.write(|w| unsafe { w.ptr().bits(ptr as u32) });
-            r.txd.maxcnt.write(|w| unsafe { w.maxcnt().bits(len as _) });
+            r.rxd.ptr.write(|w| unsafe { w.ptr().bits(ptr as u32) });
+            r.rxd.maxcnt.write(|w| unsafe { w.maxcnt().bits(len as _) });
 
-            r.events_endtx.reset();
-            r.intenset.write(|w| w.endtx().set());
+            r.events_endrx.reset();
+            r.intenset.write(|w| w.endrx().set());
 
             compiler_fence(Ordering::SeqCst);
 
-            trace!("starttx");
-            r.tasks_starttx.write(|w| unsafe { w.bits(1) });
+            trace!("startrx");
+            r.tasks_startrx.write(|w| unsafe { w.bits(1) });
 
             poll_fn(|cx| {
-                s.endtx_waker.register(cx.waker());
-                if r.events_endtx.read().bits() != 0 {
+                s.endrx_waker.register(cx.waker());
+                if r.events_endrx.read().bits() != 0 {
                     return Poll::Ready(());
                 }
                 Poll::Pending
@@ -267,7 +319,7 @@ impl<'d, T: Instance> Write for Uarte<'d, T> {
             .await;
 
             compiler_fence(Ordering::SeqCst);
-            r.events_txstarted.reset();
+            r.events_rxstarted.reset();
             drop.defuse();
 
             Ok(())
@@ -275,6 +327,24 @@ impl<'d, T: Instance> Write for Uarte<'d, T> {
     }
 }
 
+impl<'a, T: Instance> Drop for UarteRx<'a, T> {
+    fn drop(&mut self) {
+        info!("uarte rx drop");
+
+        let r = T::regs();
+
+        let did_stoprx = r.events_rxstarted.read().bits() != 0;
+        info!("did_stoprx {}", did_stoprx);
+
+        // Wait for rxto, if needed.
+        while did_stoprx && r.events_rxto.read().bits() == 0 {}
+
+        let s = T::state();
+
+        drop_tx_rx(&r, &s);
+    }
+}
+
 #[cfg(not(any(feature = "_nrf9160", feature = "nrf5340")))]
 pub(in crate) fn apply_workaround_for_enable_anomaly(_r: &crate::pac::uarte0::RegisterBlock) {
     // Do nothing
@@ -328,6 +398,21 @@ pub(in crate) fn apply_workaround_for_enable_anomaly(r: &crate::pac::uarte0::Reg
     }
 }
 
+pub(in crate) fn drop_tx_rx(r: &pac::uarte0::RegisterBlock, s: &sealed::State) {
+    if s.tx_rx_refcount.fetch_sub(1, Ordering::Relaxed) == 1 {
+        // Finally we can disable, and we do so for the peripheral
+        // i.e. not just rx concerns.
+        r.enable.write(|w| w.enable().disabled());
+
+        gpio::deconfigure_pin(r.psel.rxd.read().bits());
+        gpio::deconfigure_pin(r.psel.txd.read().bits());
+        gpio::deconfigure_pin(r.psel.rts.read().bits());
+        gpio::deconfigure_pin(r.psel.cts.read().bits());
+
+        info!("uarte tx and rx drop: done");
+    }
+}
+
 /// Interface to an UARTE peripheral that uses an additional timer and two PPI channels,
 /// allowing it to implement the ReadUntilIdle trait.
 pub struct UarteWithIdle<'d, U: Instance, T: TimerInstance> {
@@ -487,6 +572,8 @@ impl<'d, U: Instance, T: TimerInstance> Write for UarteWithIdle<'d, U, T> {
 }
 
 pub(crate) mod sealed {
+    use core::sync::atomic::AtomicU8;
+
     use embassy::waitqueue::AtomicWaker;
 
     use super::*;
@@ -494,12 +581,14 @@ pub(crate) mod sealed {
     pub struct State {
         pub endrx_waker: AtomicWaker,
         pub endtx_waker: AtomicWaker,
+        pub tx_rx_refcount: AtomicU8,
     }
     impl State {
         pub const fn new() -> Self {
             Self {
                 endrx_waker: AtomicWaker::new(),
                 endtx_waker: AtomicWaker::new(),
+                tx_rx_refcount: AtomicU8::new(0),
             }
         }
     }
diff --git a/examples/nrf/src/bin/uart_split.rs b/examples/nrf/src/bin/uart_split.rs
new file mode 100644
index 000000000..4b5dbb21f
--- /dev/null
+++ b/examples/nrf/src/bin/uart_split.rs
@@ -0,0 +1,68 @@
+#![no_std]
+#![no_main]
+#![feature(type_alias_impl_trait)]
+
+#[path = "../example_common.rs"]
+mod example_common;
+use embassy::blocking_mutex::kind::Noop;
+use embassy::channel::mpsc::{self, Channel, Sender};
+use embassy::util::Forever;
+use embassy_nrf::peripherals::UARTE0;
+use embassy_nrf::uarte::UarteRx;
+use example_common::*;
+
+use embassy::executor::Spawner;
+use embassy::traits::uart::{Read, Write};
+use embassy_nrf::gpio::NoPin;
+use embassy_nrf::{interrupt, uarte, Peripherals};
+
+static CHANNEL: Forever<Channel<Noop, [u8; 8], 1>> = Forever::new();
+
+#[embassy::main]
+async fn main(spawner: Spawner, p: Peripherals) {
+    let mut config = uarte::Config::default();
+    config.parity = uarte::Parity::EXCLUDED;
+    config.baudrate = uarte::Baudrate::BAUD115200;
+
+    let irq = interrupt::take!(UARTE0_UART0);
+    let uart = uarte::Uarte::new(p.UARTE0, irq, p.P0_08, p.P0_06, NoPin, NoPin, 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)));
+
+    // Message must be in SRAM
+    {
+        let mut buf = [0; 23];
+        buf.copy_from_slice(b"Type 8 chars to echo!\r\n");
+
+        unwrap!(tx.write(&buf).await);
+        info!("wrote hello in uart!");
+    }
+
+    // Continue reading in this main task and write
+    // 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);
+        }
+    }
+}
+
+#[embassy::task]
+async fn reader(mut rx: UarteRx<'static, UARTE0>, s: Sender<'static, Noop, [u8; 8], 1>) {
+    let mut buf = [0; 8];
+    loop {
+        info!("reading...");
+        unwrap!(rx.read(&mut buf).await);
+        unwrap!(s.send(buf).await);
+    }
+}