4 files changed, 375 insertions, 25 deletions

diff --git a/embassy-rp/src/uart/buffered.rs b/embassy-rp/src/uart/buffered.rs
index 58fede765..59fec8f1b 100644
--- a/embassy-rp/src/uart/buffered.rs
+++ b/embassy-rp/src/uart/buffered.rs
@@ -74,7 +74,7 @@ pub(crate) fn init_buffers<'d, T: Instance + 'd>(
     // to pend the ISR when we want data transmission to start.
     let regs = T::regs();
     unsafe {
-        regs.uartimsc().write_set(|w| {
+        regs.uartimsc().write(|w| {
             w.set_rxim(true);
             w.set_rtim(true);
             w.set_txim(true);
diff --git a/embassy-rp/src/uart/mod.rs b/embassy-rp/src/uart/mod.rs
index 8612e9cda..ebe32cc66 100644
--- a/embassy-rp/src/uart/mod.rs
+++ b/embassy-rp/src/uart/mod.rs
@@ -1,7 +1,14 @@
+use core::future::poll_fn;
 use core::marker::PhantomData;
+use core::task::Poll;
 
+use atomic_polyfill::{AtomicU16, Ordering};
+use embassy_cortex_m::interrupt::{Interrupt, InterruptExt};
+use embassy_futures::select::{select, Either};
 use embassy_hal_common::{into_ref, PeripheralRef};
+use embassy_sync::waitqueue::AtomicWaker;
 use embassy_time::{Duration, Timer};
+use pac::uart::regs::Uartris;
 
 use crate::clocks::clk_peri_freq;
 use crate::dma::{AnyChannel, Channel};
@@ -97,6 +104,11 @@ pub enum Error {
     Framing,
 }
 
+pub struct DmaState {
+    rx_err_waker: AtomicWaker,
+    rx_errs: AtomicU16,
+}
+
 pub struct Uart<'d, T: Instance, M: Mode> {
     tx: UartTx<'d, T, M>,
     rx: UartRx<'d, T, M>,
@@ -223,15 +235,26 @@ impl<'d, T: Instance, M: Mode> UartRx<'d, T, M> {
     pub fn new(
         _uart: impl Peripheral<P = T> + 'd,
         rx: impl Peripheral<P = impl RxPin<T>> + 'd,
+        irq: impl Peripheral<P = T::Interrupt> + 'd,
         rx_dma: impl Peripheral<P = impl Channel> + 'd,
         config: Config,
     ) -> Self {
-        into_ref!(rx, rx_dma);
+        into_ref!(rx, irq, rx_dma);
         Uart::<T, M>::init(None, Some(rx.map_into()), None, None, config);
-        Self::new_inner(Some(rx_dma.map_into()))
+        Self::new_inner(Some(irq), Some(rx_dma.map_into()))
     }
 
-    fn new_inner(rx_dma: Option<PeripheralRef<'d, AnyChannel>>) -> Self {
+    fn new_inner(irq: Option<PeripheralRef<'d, T::Interrupt>>, rx_dma: Option<PeripheralRef<'d, AnyChannel>>) -> Self {
+        debug_assert_eq!(irq.is_some(), rx_dma.is_some());
+        if let Some(irq) = irq {
+            unsafe {
+                // disable all error interrupts initially
+                T::regs().uartimsc().write(|w| w.0 = 0);
+            }
+            irq.set_handler(on_interrupt::<T>);
+            irq.unpend();
+            irq.enable();
+        }
         Self {
             rx_dma,
             phantom: PhantomData,
@@ -271,6 +294,16 @@ impl<'d, T: Instance, M: Mode> UartRx<'d, T, M> {
     }
 }
 
+impl<'d, T: Instance, M: Mode> Drop for UartRx<'d, T, M> {
+    fn drop(&mut self) {
+        if let Some(_) = self.rx_dma {
+            unsafe {
+                T::Interrupt::steal().disable();
+            }
+        }
+    }
+}
+
 impl<'d, T: Instance> UartRx<'d, T, Blocking> {
     pub fn new_blocking(
         _uart: impl Peripheral<P = T> + 'd,
@@ -279,7 +312,7 @@ impl<'d, T: Instance> UartRx<'d, T, Blocking> {
     ) -> Self {
         into_ref!(rx);
         Uart::<T, Blocking>::init(None, Some(rx.map_into()), None, None, config);
-        Self::new_inner(None)
+        Self::new_inner(None, None)
     }
 
     #[cfg(feature = "nightly")]
@@ -296,19 +329,93 @@ impl<'d, T: Instance> UartRx<'d, T, Blocking> {
     }
 }
 
+unsafe fn on_interrupt<T: Instance>(_: *mut ()) {
+    let uart = T::regs();
+    let state = T::dma_state();
+    let errs = uart.uartris().read();
+    state.rx_errs.store(errs.0 as u16, Ordering::Relaxed);
+    state.rx_err_waker.wake();
+    // disable the error interrupts instead of clearing the flags. clearing the
+    // flags would allow the dma transfer to continue, potentially signaling
+    // completion before we can check for errors that happened *during* the transfer.
+    uart.uartimsc().write_clear(|w| w.0 = errs.0);
+}
+
 impl<'d, T: Instance> UartRx<'d, T, Async> {
     pub async fn read(&mut self, buffer: &mut [u8]) -> Result<(), Error> {
+        // clear error flags before we drain the fifo. errors that have accumulated
+        // in the flags will also be present in the fifo.
+        T::dma_state().rx_errs.store(0, Ordering::Relaxed);
+        unsafe {
+            T::regs().uarticr().write(|w| {
+                w.set_oeic(true);
+                w.set_beic(true);
+                w.set_peic(true);
+                w.set_feic(true);
+            });
+        }
+
+        // then drain the fifo. we need to read at most 32 bytes. errors that apply
+        // to fifo bytes will be reported directly.
+        let buffer = match {
+            let limit = buffer.len().min(32);
+            self.drain_fifo(&mut buffer[0..limit])
+        } {
+            Ok(len) if len < buffer.len() => &mut buffer[len..],
+            Ok(_) => return Ok(()),
+            Err(e) => return Err(e),
+        };
+
+        // start a dma transfer. if errors have happened in the interim some error
+        // interrupt flags will have been raised, and those will be picked up immediately
+        // by the interrupt handler.
         let ch = self.rx_dma.as_mut().unwrap();
         let transfer = unsafe {
+            T::regs().uartimsc().write_set(|w| {
+                w.set_oeim(true);
+                w.set_beim(true);
+                w.set_peim(true);
+                w.set_feim(true);
+            });
             T::regs().uartdmacr().write_set(|reg| {
                 reg.set_rxdmae(true);
+                reg.set_dmaonerr(true);
             });
             // If we don't assign future to a variable, the data register pointer
             // is held across an await and makes the future non-Send.
             crate::dma::read(ch, T::regs().uartdr().ptr() as *const _, buffer, T::RX_DREQ)
         };
-        transfer.await;
-        Ok(())
+
+        // wait for either the transfer to complete or an error to happen.
+        let transfer_result = select(
+            transfer,
+            poll_fn(|cx| {
+                T::dma_state().rx_err_waker.register(cx.waker());
+                match T::dma_state().rx_errs.swap(0, Ordering::Relaxed) {
+                    0 => Poll::Pending,
+                    e => Poll::Ready(Uartris(e as u32)),
+                }
+            }),
+        )
+        .await;
+
+        let errors = match transfer_result {
+            Either::First(()) => return Ok(()),
+            Either::Second(e) => e,
+        };
+
+        if errors.0 == 0 {
+            return Ok(());
+        } else if errors.oeris() {
+            return Err(Error::Overrun);
+        } else if errors.beris() {
+            return Err(Error::Break);
+        } else if errors.peris() {
+            return Err(Error::Parity);
+        } else if errors.feris() {
+            return Err(Error::Framing);
+        }
+        unreachable!("unrecognized rx error");
     }
 }
 
@@ -321,7 +428,7 @@ impl<'d, T: Instance> Uart<'d, T, Blocking> {
         config: Config,
     ) -> Self {
         into_ref!(tx, rx);
-        Self::new_inner(uart, tx.map_into(), rx.map_into(), None, None, None, None, config)
+        Self::new_inner(uart, tx.map_into(), rx.map_into(), None, None, None, None, None, config)
     }
 
     /// Create a new UART with hardware flow control (RTS/CTS)
@@ -342,6 +449,7 @@ impl<'d, T: Instance> Uart<'d, T, Blocking> {
             Some(cts.map_into()),
             None,
             None,
+            None,
             config,
         )
     }
@@ -370,17 +478,19 @@ impl<'d, T: Instance> Uart<'d, T, Async> {
         uart: impl Peripheral<P = T> + 'd,
         tx: impl Peripheral<P = impl TxPin<T>> + 'd,
         rx: impl Peripheral<P = impl RxPin<T>> + 'd,
+        irq: impl Peripheral<P = T::Interrupt> + 'd,
         tx_dma: impl Peripheral<P = impl Channel> + 'd,
         rx_dma: impl Peripheral<P = impl Channel> + 'd,
         config: Config,
     ) -> Self {
-        into_ref!(tx, rx, tx_dma, rx_dma);
+        into_ref!(tx, rx, irq, tx_dma, rx_dma);
         Self::new_inner(
             uart,
             tx.map_into(),
             rx.map_into(),
             None,
             None,
+            Some(irq),
             Some(tx_dma.map_into()),
             Some(rx_dma.map_into()),
             config,
@@ -394,17 +504,19 @@ impl<'d, T: Instance> Uart<'d, T, Async> {
         rx: impl Peripheral<P = impl RxPin<T>> + 'd,
         rts: impl Peripheral<P = impl RtsPin<T>> + 'd,
         cts: impl Peripheral<P = impl CtsPin<T>> + 'd,
+        irq: impl Peripheral<P = T::Interrupt> + 'd,
         tx_dma: impl Peripheral<P = impl Channel> + 'd,
         rx_dma: impl Peripheral<P = impl Channel> + 'd,
         config: Config,
     ) -> Self {
-        into_ref!(tx, rx, cts, rts, tx_dma, rx_dma);
+        into_ref!(tx, rx, cts, rts, irq, tx_dma, rx_dma);
         Self::new_inner(
             uart,
             tx.map_into(),
             rx.map_into(),
             Some(rts.map_into()),
             Some(cts.map_into()),
+            Some(irq),
             Some(tx_dma.map_into()),
             Some(rx_dma.map_into()),
             config,
@@ -419,6 +531,7 @@ impl<'d, T: Instance + 'd, M: Mode> Uart<'d, T, M> {
         mut rx: PeripheralRef<'d, AnyPin>,
         mut rts: Option<PeripheralRef<'d, AnyPin>>,
         mut cts: Option<PeripheralRef<'d, AnyPin>>,
+        irq: Option<PeripheralRef<'d, T::Interrupt>>,
         tx_dma: Option<PeripheralRef<'d, AnyChannel>>,
         rx_dma: Option<PeripheralRef<'d, AnyChannel>>,
         config: Config,
@@ -433,7 +546,7 @@ impl<'d, T: Instance + 'd, M: Mode> Uart<'d, T, M> {
 
         Self {
             tx: UartTx::new_inner(tx_dma),
-            rx: UartRx::new_inner(rx_dma),
+            rx: UartRx::new_inner(irq, rx_dma),
         }
     }
 
@@ -761,6 +874,7 @@ mod sealed {
     pub trait Instance {
         const TX_DREQ: u8;
         const RX_DREQ: u8;
+        const ID: usize;
 
         type Interrupt: crate::interrupt::Interrupt;
 
@@ -768,6 +882,8 @@ mod sealed {
 
         #[cfg(feature = "nightly")]
         fn buffered_state() -> &'static buffered::State;
+
+        fn dma_state() -> &'static DmaState;
     }
     pub trait TxPin<T: Instance> {}
     pub trait RxPin<T: Instance> {}
@@ -793,10 +909,11 @@ impl_mode!(Async);
 pub trait Instance: sealed::Instance {}
 
 macro_rules! impl_instance {
-    ($inst:ident, $irq:ident, $tx_dreq:expr, $rx_dreq:expr) => {
+    ($inst:ident, $irq:ident, $id:expr, $tx_dreq:expr, $rx_dreq:expr) => {
         impl sealed::Instance for peripherals::$inst {
             const TX_DREQ: u8 = $tx_dreq;
             const RX_DREQ: u8 = $rx_dreq;
+            const ID: usize = $id;
 
             type Interrupt = crate::interrupt::$irq;
 
@@ -809,13 +926,21 @@ macro_rules! impl_instance {
                 static STATE: buffered::State = buffered::State::new();
                 &STATE
             }
+
+            fn dma_state() -> &'static DmaState {
+                static STATE: DmaState = DmaState {
+                    rx_err_waker: AtomicWaker::new(),
+                    rx_errs: AtomicU16::new(0),
+                };
+                &STATE
+            }
         }
         impl Instance for peripherals::$inst {}
     };
 }
 
-impl_instance!(UART0, UART0_IRQ, 20, 21);
-impl_instance!(UART1, UART1_IRQ, 22, 23);
+impl_instance!(UART0, UART0_IRQ, 0, 20, 21);
+impl_instance!(UART1, UART1_IRQ, 1, 22, 23);
 
 pub trait TxPin<T: Instance>: sealed::TxPin<T> + crate::gpio::Pin {}
 pub trait RxPin<T: Instance>: sealed::RxPin<T> + crate::gpio::Pin {}
diff --git a/examples/rp/src/bin/uart_unidir.rs b/examples/rp/src/bin/uart_unidir.rs
index f56e7009f..4119a309f 100644
--- a/examples/rp/src/bin/uart_unidir.rs
+++ b/examples/rp/src/bin/uart_unidir.rs
@@ -7,6 +7,7 @@
 
 use defmt::*;
 use embassy_executor::Spawner;
+use embassy_rp::interrupt;
 use embassy_rp::peripherals::UART1;
 use embassy_rp::uart::{Async, Config, UartRx, UartTx};
 use embassy_time::{Duration, Timer};
@@ -17,7 +18,13 @@ async fn main(spawner: Spawner) {
     let p = embassy_rp::init(Default::default());
 
     let mut uart_tx = UartTx::new(p.UART0, p.PIN_0, p.DMA_CH0, Config::default());
-    let uart_rx = UartRx::new(p.UART1, p.PIN_5, p.DMA_CH1, Config::default());
+    let uart_rx = UartRx::new(
+        p.UART1,
+        p.PIN_5,
+        interrupt::take!(UART1_IRQ),
+        p.DMA_CH1,
+        Config::default(),
+    );
 
     unwrap!(spawner.spawn(reader(uart_rx)));
 
diff --git a/tests/rp/src/bin/uart_dma.rs b/tests/rp/src/bin/uart_dma.rs
index 963c22707..92aa205c9 100644
--- a/tests/rp/src/bin/uart_dma.rs
+++ b/tests/rp/src/bin/uart_dma.rs
@@ -4,28 +4,246 @@
 
 use defmt::{assert_eq, *};
 use embassy_executor::Spawner;
-use embassy_rp::uart::{Config, Uart};
+use embassy_rp::gpio::{Level, Output};
+use embassy_rp::interrupt;
+use embassy_rp::uart::{Async, Config, Error, Instance, Parity, Uart, UartRx};
+use embassy_time::{Duration, Timer};
 use {defmt_rtt as _, panic_probe as _};
 
+async fn read<const N: usize>(uart: &mut Uart<'_, impl Instance, Async>) -> Result<[u8; N], Error> {
+    let mut buf = [255; N];
+    uart.read(&mut buf).await?;
+    Ok(buf)
+}
+
+async fn read1<const N: usize>(uart: &mut UartRx<'_, impl Instance, Async>) -> Result<[u8; N], Error> {
+    let mut buf = [255; N];
+    uart.read(&mut buf).await?;
+    Ok(buf)
+}
+
+async fn send(pin: &mut Output<'_, impl embassy_rp::gpio::Pin>, v: u8, parity: Option<bool>) {
+    pin.set_low();
+    Timer::after(Duration::from_millis(1)).await;
+    for i in 0..8 {
+        if v & (1 << i) == 0 {
+            pin.set_low();
+        } else {
+            pin.set_high();
+        }
+        Timer::after(Duration::from_millis(1)).await;
+    }
+    if let Some(b) = parity {
+        if b {
+            pin.set_high();
+        } else {
+            pin.set_low();
+        }
+        Timer::after(Duration::from_millis(1)).await;
+    }
+    pin.set_high();
+    Timer::after(Duration::from_millis(1)).await;
+}
+
 #[embassy_executor::main]
 async fn main(_spawner: Spawner) {
-    let p = embassy_rp::init(Default::default());
+    let mut p = embassy_rp::init(Default::default());
     info!("Hello World!");
 
-    let (tx, rx, uart) = (p.PIN_0, p.PIN_1, p.UART0);
+    let (mut tx, mut rx, mut uart) = (p.PIN_0, p.PIN_1, p.UART0);
+    let mut irq = interrupt::take!(UART0_IRQ);
 
-    let config = Config::default();
-    let mut uart = Uart::new(uart, tx, rx, p.DMA_CH0, p.DMA_CH1, config);
+    // TODO
+    // nuclear error reporting. just abort the entire transfer and invalidate the
+    // dma buffer, buffered buffer, fifo etc.
 
     // We can't send too many bytes, they have to fit in the FIFO.
     // This is because we aren't sending+receiving at the same time.
+    {
+        let config = Config::default();
+        let mut uart = Uart::new(
+            &mut uart,
+            &mut tx,
+            &mut rx,
+            &mut irq,
+            &mut p.DMA_CH0,
+            &mut p.DMA_CH1,
+            config,
+        );
+
+        let data = [0xC0, 0xDE];
+        uart.write(&data).await.unwrap();
+
+        let mut buf = [0; 2];
+        uart.read(&mut buf).await.unwrap();
+        assert_eq!(buf, data);
+    }
+
+    info!("test overflow detection");
+    {
+        let config = Config::default();
+        let mut uart = Uart::new(
+            &mut uart,
+            &mut tx,
+            &mut rx,
+            &mut irq,
+            &mut p.DMA_CH0,
+            &mut p.DMA_CH1,
+            config,
+        );
+
+        uart.blocking_write(&[42; 32]).unwrap();
+        uart.blocking_write(&[1, 2, 3]).unwrap();
+        uart.blocking_flush().unwrap();
+
+        // can receive regular fifo contents
+        assert_eq!(read(&mut uart).await, Ok([42; 16]));
+        assert_eq!(read(&mut uart).await, Ok([42; 16]));
+        // receiving the rest fails with overrun
+        assert_eq!(read::<16>(&mut uart).await, Err(Error::Overrun));
+        // new data is accepted, latest overrunning byte first
+        assert_eq!(read(&mut uart).await, Ok([3]));
+        uart.blocking_write(&[8, 9]).unwrap();
+        Timer::after(Duration::from_millis(1)).await;
+        assert_eq!(read(&mut uart).await, Ok([8, 9]));
+    }
+
+    info!("test break detection");
+    {
+        let config = Config::default();
+        let (mut tx, mut rx) = Uart::new(
+            &mut uart,
+            &mut tx,
+            &mut rx,
+            &mut irq,
+            &mut p.DMA_CH0,
+            &mut p.DMA_CH1,
+            config,
+        )
+        .split();
+
+        // break before read
+        tx.send_break(20).await;
+        tx.write(&[64]).await.unwrap();
+        assert_eq!(read1::<1>(&mut rx).await.unwrap_err(), Error::Break);
+        assert_eq!(read1(&mut rx).await.unwrap(), [64]);
+
+        // break during read
+        {
+            let r = read1::<2>(&mut rx);
+            tx.write(&[2]).await.unwrap();
+            tx.send_break(20).await;
+            tx.write(&[3]).await.unwrap();
+            assert_eq!(r.await.unwrap_err(), Error::Break);
+            assert_eq!(read1(&mut rx).await.unwrap(), [3]);
+        }
+
+        // break after read
+        {
+            let r = read1(&mut rx);
+            tx.write(&[2]).await.unwrap();
+            tx.send_break(20).await;
+            tx.write(&[3]).await.unwrap();
+            assert_eq!(r.await.unwrap(), [2]);
+            assert_eq!(read1::<1>(&mut rx).await.unwrap_err(), Error::Break);
+            assert_eq!(read1(&mut rx).await.unwrap(), [3]);
+        }
+    }
+
+    // parity detection. here we bitbang to not require two uarts.
+    info!("test parity error detection");
+    {
+        let mut pin = Output::new(&mut tx, Level::High);
+        // choose a very slow baud rate to make tests reliable even with O0
+        let mut config = Config::default();
+        config.baudrate = 1000;
+        config.parity = Parity::ParityEven;
+        let mut uart = UartRx::new(&mut uart, &mut rx, &mut irq, &mut p.DMA_CH0, config);
+
+        async fn chr(pin: &mut Output<'_, impl embassy_rp::gpio::Pin>, v: u8, parity: u32) {
+            send(pin, v, Some(parity != 0)).await;
+        }
+
+        // first check that we can send correctly
+        chr(&mut pin, 32, 1).await;
+        assert_eq!(read1(&mut uart).await.unwrap(), [32]);
+
+        // parity error before read
+        chr(&mut pin, 32, 0).await;
+        chr(&mut pin, 31, 1).await;
+        assert_eq!(read1::<1>(&mut uart).await.unwrap_err(), Error::Parity);
+        assert_eq!(read1(&mut uart).await.unwrap(), [31]);
+
+        // parity error during read
+        {
+            let r = read1::<2>(&mut uart);
+            chr(&mut pin, 2, 1).await;
+            chr(&mut pin, 32, 0).await;
+            chr(&mut pin, 3, 0).await;
+            assert_eq!(r.await.unwrap_err(), Error::Parity);
+            assert_eq!(read1(&mut uart).await.unwrap(), [3]);
+        }
+
+        // parity error after read
+        {
+            let r = read1(&mut uart);
+            chr(&mut pin, 2, 1).await;
+            chr(&mut pin, 32, 0).await;
+            chr(&mut pin, 3, 0).await;
+            assert_eq!(r.await.unwrap(), [2]);
+            assert_eq!(read1::<1>(&mut uart).await.unwrap_err(), Error::Parity);
+            assert_eq!(read1(&mut uart).await.unwrap(), [3]);
+        }
+    }
+
+    // framing error detection. here we bitbang because there's no other way.
+    info!("test framing error detection");
+    {
+        let mut pin = Output::new(&mut tx, Level::High);
+        // choose a very slow baud rate to make tests reliable even with O0
+        let mut config = Config::default();
+        config.baudrate = 1000;
+        let mut uart = UartRx::new(&mut uart, &mut rx, &mut irq, &mut p.DMA_CH0, config);
+
+        async fn chr(pin: &mut Output<'_, impl embassy_rp::gpio::Pin>, v: u8, good: bool) {
+            if good {
+                send(pin, v, None).await;
+            } else {
+                send(pin, v, Some(false)).await;
+            }
+        }
+
+        // first check that we can send correctly
+        chr(&mut pin, 32, true).await;
+        assert_eq!(read1(&mut uart).await.unwrap(), [32]);
+
+        // parity error before read
+        chr(&mut pin, 32, false).await;
+        chr(&mut pin, 31, true).await;
+        assert_eq!(read1::<1>(&mut uart).await.unwrap_err(), Error::Framing);
+        assert_eq!(read1(&mut uart).await.unwrap(), [31]);
 
-    let data = [0xC0, 0xDE];
-    uart.write(&data).await.unwrap();
+        // parity error during read
+        {
+            let r = read1::<2>(&mut uart);
+            chr(&mut pin, 2, true).await;
+            chr(&mut pin, 32, false).await;
+            chr(&mut pin, 3, true).await;
+            assert_eq!(r.await.unwrap_err(), Error::Framing);
+            assert_eq!(read1(&mut uart).await.unwrap(), [3]);
+        }
 
-    let mut buf = [0; 2];
-    uart.read(&mut buf).await.unwrap();
-    assert_eq!(buf, data);
+        // parity error after read
+        {
+            let r = read1(&mut uart);
+            chr(&mut pin, 2, true).await;
+            chr(&mut pin, 32, false).await;
+            chr(&mut pin, 3, true).await;
+            assert_eq!(r.await.unwrap(), [2]);
+            assert_eq!(read1::<1>(&mut uart).await.unwrap_err(), Error::Framing);
+            assert_eq!(read1(&mut uart).await.unwrap(), [3]);
+        }
+    }
 
     info!("Test OK");
     cortex_m::asm::bkpt();