Merge #1414

1414: rp: report errors from buffered and dma uart receives r=Dirbaio a=pennae

neither of these reported errors so far, which is not ideal. add error reporting to both of them that matches the blocking error reporting as closely as is feasible, even allowing partial receives from buffered uarts before errors are reported where they would have been by the blocking code. dma transfers don't do this, if an errors applies to any byte in a transfer the entire transfer is nuked (though we probably could report how many bytes have been transferred).

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

6 files changed, 1005 insertions, 151 deletions

diff --git a/embassy-rp/src/uart/buffered.rs b/embassy-rp/src/uart/buffered.rs
index cb0461930..59fec8f1b 100644
--- a/embassy-rp/src/uart/buffered.rs
+++ b/embassy-rp/src/uart/buffered.rs
@@ -2,11 +2,14 @@ use core::future::{poll_fn, Future};
 use core::slice;
 use core::task::Poll;
 
+use atomic_polyfill::{AtomicU8, Ordering};
 use embassy_cortex_m::interrupt::{Interrupt, InterruptExt};
 use embassy_hal_common::atomic_ring_buffer::RingBuffer;
 use embassy_sync::waitqueue::AtomicWaker;
+use embassy_time::{Duration, Timer};
 
 use super::*;
+use crate::clocks::clk_peri_freq;
 use crate::RegExt;
 
 pub struct State {
@@ -14,8 +17,15 @@ pub struct State {
     tx_buf: RingBuffer,
     rx_waker: AtomicWaker,
     rx_buf: RingBuffer,
+    rx_error: AtomicU8,
 }
 
+// these must match bits 8..11 in UARTDR
+const RXE_OVERRUN: u8 = 8;
+const RXE_BREAK: u8 = 4;
+const RXE_PARITY: u8 = 2;
+const RXE_FRAMING: u8 = 1;
+
 impl State {
     pub const fn new() -> Self {
         Self {
@@ -23,6 +33,7 @@ impl State {
             tx_buf: RingBuffer::new(),
             rx_waker: AtomicWaker::new(),
             tx_waker: AtomicWaker::new(),
+            rx_error: AtomicU8::new(0),
         }
     }
 }
@@ -45,7 +56,7 @@ pub(crate) fn init_buffers<'d, T: Instance + 'd>(
     tx_buffer: &'d mut [u8],
     rx_buffer: &'d mut [u8],
 ) {
-    let state = T::state();
+    let state = T::buffered_state();
     let len = tx_buffer.len();
     unsafe { state.tx_buf.init(tx_buffer.as_mut_ptr(), len) };
     let len = rx_buffer.len();
@@ -63,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);
@@ -136,6 +147,14 @@ impl<'d, T: Instance> BufferedUart<'d, T> {
         self.rx.blocking_read(buffer)
     }
 
+    pub fn busy(&self) -> bool {
+        self.tx.busy()
+    }
+
+    pub async fn send_break(&mut self, bits: u32) {
+        self.tx.send_break(bits).await
+    }
+
     pub fn split(self) -> (BufferedUartRx<'d, T>, BufferedUartTx<'d, T>) {
         (self.rx, self.tx)
     }
@@ -173,90 +192,113 @@ impl<'d, T: Instance> BufferedUartRx<'d, T> {
         Self { phantom: PhantomData }
     }
 
-    fn read<'a>(buf: &'a mut [u8]) -> impl Future<Output = Result<usize, Error>> + 'a {
+    fn read<'a>(buf: &'a mut [u8]) -> impl Future<Output = Result<usize, Error>> + 'a
+    where
+        T: 'd,
+    {
         poll_fn(move |cx| {
-            if buf.is_empty() {
-                return Poll::Ready(Ok(0));
-            }
-
-            let state = T::state();
-            let mut rx_reader = unsafe { state.rx_buf.reader() };
-            let n = rx_reader.pop(|data| {
-                let n = data.len().min(buf.len());
-                buf[..n].copy_from_slice(&data[..n]);
-                n
-            });
-            if n == 0 {
-                state.rx_waker.register(cx.waker());
-                return Poll::Pending;
-            }
-
-            // (Re-)Enable the interrupt to receive more data in case it was
-            // disabled because the buffer was full.
-            let regs = T::regs();
-            unsafe {
-                regs.uartimsc().write_set(|w| {
-                    w.set_rxim(true);
-                    w.set_rtim(true);
-                });
+            if let Poll::Ready(r) = Self::try_read(buf) {
+                return Poll::Ready(r);
             }
-
-            Poll::Ready(Ok(n))
+            T::buffered_state().rx_waker.register(cx.waker());
+            Poll::Pending
         })
     }
 
-    pub fn blocking_read(&mut self, buf: &mut [u8]) -> Result<usize, Error> {
+    fn get_rx_error() -> Option<Error> {
+        let errs = T::buffered_state().rx_error.swap(0, Ordering::Relaxed);
+        if errs & RXE_OVERRUN != 0 {
+            Some(Error::Overrun)
+        } else if errs & RXE_BREAK != 0 {
+            Some(Error::Break)
+        } else if errs & RXE_PARITY != 0 {
+            Some(Error::Parity)
+        } else if errs & RXE_FRAMING != 0 {
+            Some(Error::Framing)
+        } else {
+            None
+        }
+    }
+
+    fn try_read(buf: &mut [u8]) -> Poll<Result<usize, Error>>
+    where
+        T: 'd,
+    {
         if buf.is_empty() {
-            return Ok(0);
+            return Poll::Ready(Ok(0));
         }
 
-        loop {
-            let state = T::state();
-            let mut rx_reader = unsafe { state.rx_buf.reader() };
-            let n = rx_reader.pop(|data| {
-                let n = data.len().min(buf.len());
-                buf[..n].copy_from_slice(&data[..n]);
-                n
+        let state = T::buffered_state();
+        let mut rx_reader = unsafe { state.rx_buf.reader() };
+        let n = rx_reader.pop(|data| {
+            let n = data.len().min(buf.len());
+            buf[..n].copy_from_slice(&data[..n]);
+            n
+        });
+
+        let result = if n == 0 {
+            match Self::get_rx_error() {
+                None => return Poll::Pending,
+                Some(e) => Err(e),
+            }
+        } else {
+            Ok(n)
+        };
+
+        // (Re-)Enable the interrupt to receive more data in case it was
+        // disabled because the buffer was full or errors were detected.
+        let regs = T::regs();
+        unsafe {
+            regs.uartimsc().write_set(|w| {
+                w.set_rxim(true);
+                w.set_rtim(true);
             });
+        }
 
-            if n > 0 {
-                // (Re-)Enable the interrupt to receive more data in case it was
-                // disabled because the buffer was full.
-                let regs = T::regs();
-                unsafe {
-                    regs.uartimsc().write_set(|w| {
-                        w.set_rxim(true);
-                        w.set_rtim(true);
-                    });
-                }
+        Poll::Ready(result)
+    }
 
-                return Ok(n);
+    pub fn blocking_read(&mut self, buf: &mut [u8]) -> Result<usize, Error> {
+        loop {
+            match Self::try_read(buf) {
+                Poll::Ready(res) => return res,
+                Poll::Pending => continue,
             }
         }
     }
 
-    fn fill_buf<'a>() -> impl Future<Output = Result<&'a [u8], Error>> {
+    fn fill_buf<'a>() -> impl Future<Output = Result<&'a [u8], Error>>
+    where
+        T: 'd,
+    {
         poll_fn(move |cx| {
-            let state = T::state();
+            let state = T::buffered_state();
             let mut rx_reader = unsafe { state.rx_buf.reader() };
             let (p, n) = rx_reader.pop_buf();
-            if n == 0 {
-                state.rx_waker.register(cx.waker());
-                return Poll::Pending;
-            }
+            let result = if n == 0 {
+                match Self::get_rx_error() {
+                    None => {
+                        state.rx_waker.register(cx.waker());
+                        return Poll::Pending;
+                    }
+                    Some(e) => Err(e),
+                }
+            } else {
+                let buf = unsafe { slice::from_raw_parts(p, n) };
+                Ok(buf)
+            };
 
-            let buf = unsafe { slice::from_raw_parts(p, n) };
-            Poll::Ready(Ok(buf))
+            Poll::Ready(result)
         })
     }
 
     fn consume(amt: usize) {
-        let state = T::state();
+        let state = T::buffered_state();
         let mut rx_reader = unsafe { state.rx_buf.reader() };
         rx_reader.pop_done(amt);
 
         // (Re-)Enable the interrupt to receive more data in case it was
-        // disabled because the buffer was full.
+        // disabled because the buffer was full or errors were detected.
         let regs = T::regs();
         unsafe {
             regs.uartimsc().write_set(|w| {
@@ -305,7 +347,7 @@ impl<'d, T: Instance> BufferedUartTx<'d, T> {
                 return Poll::Ready(Ok(0));
             }
 
-            let state = T::state();
+            let state = T::buffered_state();
             let mut tx_writer = unsafe { state.tx_buf.writer() };
             let n = tx_writer.push(|data| {
                 let n = data.len().min(buf.len());
@@ -328,7 +370,7 @@ impl<'d, T: Instance> BufferedUartTx<'d, T> {
 
     fn flush() -> impl Future<Output = Result<(), Error>> {
         poll_fn(move |cx| {
-            let state = T::state();
+            let state = T::buffered_state();
             if !state.tx_buf.is_empty() {
                 state.tx_waker.register(cx.waker());
                 return Poll::Pending;
@@ -344,7 +386,7 @@ impl<'d, T: Instance> BufferedUartTx<'d, T> {
         }
 
         loop {
-            let state = T::state();
+            let state = T::buffered_state();
             let mut tx_writer = unsafe { state.tx_buf.writer() };
             let n = tx_writer.push(|data| {
                 let n = data.len().min(buf.len());
@@ -365,17 +407,54 @@ impl<'d, T: Instance> BufferedUartTx<'d, T> {
 
     pub fn blocking_flush(&mut self) -> Result<(), Error> {
         loop {
-            let state = T::state();
+            let state = T::buffered_state();
             if state.tx_buf.is_empty() {
                 return Ok(());
             }
         }
     }
+
+    pub fn busy(&self) -> bool {
+        unsafe { T::regs().uartfr().read().busy() }
+    }
+
+    /// Assert a break condition after waiting for the transmit buffers to empty,
+    /// for the specified number of bit times. This condition must be asserted
+    /// for at least two frame times to be effective, `bits` will adjusted
+    /// according to frame size, parity, and stop bit settings to ensure this.
+    ///
+    /// This method may block for a long amount of time since it has to wait
+    /// for the transmit fifo to empty, which may take a while on slow links.
+    pub async fn send_break(&mut self, bits: u32) {
+        let regs = T::regs();
+        let bits = bits.max(unsafe {
+            let lcr = regs.uartlcr_h().read();
+            let width = lcr.wlen() as u32 + 5;
+            let parity = lcr.pen() as u32;
+            let stops = 1 + lcr.stp2() as u32;
+            2 * (1 + width + parity + stops)
+        });
+        let divx64 = unsafe {
+            ((regs.uartibrd().read().baud_divint() as u32) << 6) + regs.uartfbrd().read().baud_divfrac() as u32
+        } as u64;
+        let div_clk = clk_peri_freq() as u64 * 64;
+        let wait_usecs = (1_000_000 * bits as u64 * divx64 * 16 + div_clk - 1) / div_clk;
+
+        Self::flush().await.unwrap();
+        while self.busy() {}
+        unsafe {
+            regs.uartlcr_h().write_set(|w| w.set_brk(true));
+        }
+        Timer::after(Duration::from_micros(wait_usecs)).await;
+        unsafe {
+            regs.uartlcr_h().write_clear(|w| w.set_brk(true));
+        }
+    }
 }
 
 impl<'d, T: Instance> Drop for BufferedUartRx<'d, T> {
     fn drop(&mut self) {
-        let state = T::state();
+        let state = T::buffered_state();
         unsafe {
             state.rx_buf.deinit();
 
@@ -390,7 +469,7 @@ impl<'d, T: Instance> Drop for BufferedUartRx<'d, T> {
 
 impl<'d, T: Instance> Drop for BufferedUartTx<'d, T> {
     fn drop(&mut self) {
-        let state = T::state();
+        let state = T::buffered_state();
         unsafe {
             state.tx_buf.deinit();
 
@@ -405,7 +484,7 @@ impl<'d, T: Instance> Drop for BufferedUartTx<'d, T> {
 
 pub(crate) unsafe fn on_interrupt<T: Instance>(_: *mut ()) {
     let r = T::regs();
-    let s = T::state();
+    let s = T::buffered_state();
 
     unsafe {
         // Clear TX and error interrupt flags
@@ -439,19 +518,37 @@ pub(crate) unsafe fn on_interrupt<T: Instance>(_: *mut ()) {
         let mut rx_writer = s.rx_buf.writer();
         let rx_buf = rx_writer.push_slice();
         let mut n_read = 0;
+        let mut error = false;
         for rx_byte in rx_buf {
             if r.uartfr().read().rxfe() {
                 break;
             }
-            *rx_byte = r.uartdr().read().data();
+            let dr = r.uartdr().read();
+            if (dr.0 >> 8) != 0 {
+                s.rx_error.fetch_or((dr.0 >> 8) as u8, Ordering::Relaxed);
+                error = true;
+                // only fill the buffer with valid characters. the current character is fine
+                // if the error is an overrun, but if we add it to the buffer we'll report
+                // the overrun one character too late. drop it instead and pretend we were
+                // a bit slower at draining the rx fifo than we actually were.
+                // this is consistent with blocking uart error reporting.
+                break;
+            }
+            *rx_byte = dr.data();
             n_read += 1;
         }
         if n_read > 0 {
             rx_writer.push_done(n_read);
             s.rx_waker.wake();
+        } else if error {
+            s.rx_waker.wake();
         }
-        // Disable any further RX interrupts when the buffer becomes full.
-        if s.rx_buf.is_full() {
+        // Disable any further RX interrupts when the buffer becomes full or
+        // errors have occured. this lets us buffer additional errors in the
+        // fifo without needing more error storage locations, and most applications
+        // will want to do a full reset of their uart state anyway once an error
+        // has happened.
+        if s.rx_buf.is_full() || error {
             r.uartimsc().write_clear(|w| {
                 w.set_rxim(true);
                 w.set_rtim(true);
diff --git a/embassy-rp/src/uart/mod.rs b/embassy-rp/src/uart/mod.rs
index dedc390f0..ebe32cc66 100644
--- a/embassy-rp/src/uart/mod.rs
+++ b/embassy-rp/src/uart/mod.rs
@@ -1,12 +1,21 @@
+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};
 use crate::gpio::sealed::Pin;
 use crate::gpio::AnyPin;
 use crate::pac::io::vals::{Inover, Outover};
-use crate::{pac, peripherals, Peripheral};
+use crate::{pac, peripherals, Peripheral, RegExt};
 
 #[cfg(feature = "nightly")]
 mod buffered;
@@ -95,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>,
@@ -146,6 +160,43 @@ impl<'d, T: Instance, M: Mode> UartTx<'d, T, M> {
         unsafe { while !r.uartfr().read().txfe() {} }
         Ok(())
     }
+
+    pub fn busy(&self) -> bool {
+        unsafe { T::regs().uartfr().read().busy() }
+    }
+
+    /// Assert a break condition after waiting for the transmit buffers to empty,
+    /// for the specified number of bit times. This condition must be asserted
+    /// for at least two frame times to be effective, `bits` will adjusted
+    /// according to frame size, parity, and stop bit settings to ensure this.
+    ///
+    /// This method may block for a long amount of time since it has to wait
+    /// for the transmit fifo to empty, which may take a while on slow links.
+    pub async fn send_break(&mut self, bits: u32) {
+        let regs = T::regs();
+        let bits = bits.max(unsafe {
+            let lcr = regs.uartlcr_h().read();
+            let width = lcr.wlen() as u32 + 5;
+            let parity = lcr.pen() as u32;
+            let stops = 1 + lcr.stp2() as u32;
+            2 * (1 + width + parity + stops)
+        });
+        let divx64 = unsafe {
+            ((regs.uartibrd().read().baud_divint() as u32) << 6) + regs.uartfbrd().read().baud_divfrac() as u32
+        } as u64;
+        let div_clk = clk_peri_freq() as u64 * 64;
+        let wait_usecs = (1_000_000 * bits as u64 * divx64 * 16 + div_clk - 1) / div_clk;
+
+        self.blocking_flush().unwrap();
+        while self.busy() {}
+        unsafe {
+            regs.uartlcr_h().write_set(|w| w.set_brk(true));
+        }
+        Timer::after(Duration::from_micros(wait_usecs)).await;
+        unsafe {
+            regs.uartlcr_h().write_clear(|w| w.set_brk(true));
+        }
+    }
 }
 
 impl<'d, T: Instance> UartTx<'d, T, Blocking> {
@@ -167,7 +218,7 @@ impl<'d, T: Instance> UartTx<'d, T, Async> {
     pub async fn write(&mut self, buffer: &[u8]) -> Result<(), Error> {
         let ch = self.tx_dma.as_mut().unwrap();
         let transfer = unsafe {
-            T::regs().uartdmacr().modify(|reg| {
+            T::regs().uartdmacr().write_set(|reg| {
                 reg.set_txdmae(true);
             });
             // If we don't assign future to a variable, the data register pointer
@@ -184,51 +235,86 @@ 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,
         }
     }
 
-    pub fn blocking_read(&mut self, buffer: &mut [u8]) -> Result<(), Error> {
+    pub fn blocking_read(&mut self, mut buffer: &mut [u8]) -> Result<(), Error> {
+        while buffer.len() > 0 {
+            let received = self.drain_fifo(buffer)?;
+            buffer = &mut buffer[received..];
+        }
+        Ok(())
+    }
+
+    fn drain_fifo(&mut self, buffer: &mut [u8]) -> Result<usize, Error> {
         let r = T::regs();
-        unsafe {
-            for b in buffer {
-                *b = loop {
-                    if r.uartfr().read().rxfe() {
-                        continue;
-                    }
-
-                    let dr = r.uartdr().read();
-
-                    if dr.oe() {
-                        return Err(Error::Overrun);
-                    } else if dr.be() {
-                        return Err(Error::Break);
-                    } else if dr.pe() {
-                        return Err(Error::Parity);
-                    } else if dr.fe() {
-                        return Err(Error::Framing);
-                    } else {
-                        break dr.data();
-                    }
-                };
+        for (i, b) in buffer.iter_mut().enumerate() {
+            if unsafe { r.uartfr().read().rxfe() } {
+                return Ok(i);
+            }
+
+            let dr = unsafe { r.uartdr().read() };
+
+            if dr.oe() {
+                return Err(Error::Overrun);
+            } else if dr.be() {
+                return Err(Error::Break);
+            } else if dr.pe() {
+                return Err(Error::Parity);
+            } else if dr.fe() {
+                return Err(Error::Framing);
+            } else {
+                *b = dr.data();
+            }
+        }
+        Ok(buffer.len())
+    }
+}
+
+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();
             }
         }
-        Ok(())
     }
 }
 
 impl<'d, T: Instance> UartRx<'d, T, Blocking> {
+    pub fn new_blocking(
+        _uart: impl Peripheral<P = T> + 'd,
+        rx: impl Peripheral<P = impl RxPin<T>> + 'd,
+        config: Config,
+    ) -> Self {
+        into_ref!(rx);
+        Uart::<T, Blocking>::init(None, Some(rx.map_into()), None, None, config);
+        Self::new_inner(None, None)
+    }
+
     #[cfg(feature = "nightly")]
     pub fn into_buffered(
         self,
@@ -243,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().uartdmacr().modify(|reg| {
+            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");
     }
 }
 
@@ -268,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)
@@ -289,6 +449,7 @@ impl<'d, T: Instance> Uart<'d, T, Blocking> {
             Some(cts.map_into()),
             None,
             None,
+            None,
             config,
         )
     }
@@ -317,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,
@@ -341,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,
@@ -366,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,
@@ -380,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),
         }
     }
 
@@ -516,6 +682,14 @@ impl<'d, T: Instance, M: Mode> Uart<'d, T, M> {
         self.rx.blocking_read(buffer)
     }
 
+    pub fn busy(&self) -> bool {
+        self.tx.busy()
+    }
+
+    pub async fn send_break(&mut self, bits: u32) {
+        self.tx.send_break(bits).await
+    }
+
     /// Split the Uart into a transmitter and receiver, which is particuarly
     /// useful when having two tasks correlating to transmitting and receiving.
     pub fn split(self) -> (UartTx<'d, T, M>, UartRx<'d, T, M>) {
@@ -700,13 +874,16 @@ mod sealed {
     pub trait Instance {
         const TX_DREQ: u8;
         const RX_DREQ: u8;
+        const ID: usize;
 
         type Interrupt: crate::interrupt::Interrupt;
 
         fn regs() -> pac::uart::Uart;
 
         #[cfg(feature = "nightly")]
-        fn state() -> &'static buffered::State;
+        fn buffered_state() -> &'static buffered::State;
+
+        fn dma_state() -> &'static DmaState;
     }
     pub trait TxPin<T: Instance> {}
     pub trait RxPin<T: Instance> {}
@@ -732,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;
 
@@ -744,17 +922,25 @@ macro_rules! impl_instance {
             }
 
             #[cfg(feature = "nightly")]
-            fn state() -> &'static buffered::State {
+            fn buffered_state() -> &'static buffered::State {
                 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.rs b/tests/rp/src/bin/uart.rs
index 92f61464e..80c18c02e 100644
--- a/tests/rp/src/bin/uart.rs
+++ b/tests/rp/src/bin/uart.rs
@@ -4,28 +4,165 @@
 
 use defmt::{assert_eq, *};
 use embassy_executor::Spawner;
-use embassy_rp::uart::{Config, Uart};
+use embassy_rp::gpio::{Level, Output};
+use embassy_rp::uart::{Blocking, Config, Error, Instance, Parity, Uart, UartRx};
+use embassy_time::{Duration, Timer};
 use {defmt_rtt as _, panic_probe as _};
 
+fn read<const N: usize>(uart: &mut Uart<'_, impl Instance, Blocking>) -> Result<[u8; N], Error> {
+    let mut buf = [255; N];
+    uart.blocking_read(&mut buf)?;
+    Ok(buf)
+}
+
+fn read1<const N: usize>(uart: &mut UartRx<'_, impl Instance, Blocking>) -> Result<[u8; N], Error> {
+    let mut buf = [255; N];
+    uart.blocking_read(&mut buf)?;
+    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());
     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 config = Config::default();
+        let mut uart = Uart::new_blocking(&mut uart, &mut tx, &mut rx, config);
+
+        // 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 data = [0xC0, 0xDE];
+        uart.blocking_write(&data).unwrap();
+        assert_eq!(read(&mut uart).unwrap(), data);
+    }
+
+    info!("test overflow detection");
+    {
+        let config = Config::default();
+        let mut uart = Uart::new_blocking(&mut uart, &mut tx, &mut rx, config);
+
+        let data = [
+            1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13, 14, 15, 16, 17, 18, 19, 20, 21, 22, 23, 24, 25, 26, 27, 28, 29,
+            30, 31, 32,
+        ];
+        let overflow = [
+            101, 102, 103, 104, 105, 106, 107, 108, 109, 110, 111, 112, 113, 114, 115, 116,
+        ];
+        uart.blocking_write(&data).unwrap();
+        uart.blocking_write(&overflow).unwrap();
+        while uart.busy() {}
+
+        // prefix in fifo is valid
+        assert_eq!(read(&mut uart).unwrap(), data);
+        // next received character causes overrun error and is discarded
+        uart.blocking_write(&[1, 2, 3]).unwrap();
+        assert_eq!(read::<1>(&mut uart).unwrap_err(), Error::Overrun);
+        assert_eq!(read(&mut uart).unwrap(), [2, 3]);
+    }
+
+    info!("test break detection");
+    {
+        let config = Config::default();
+        let mut uart = Uart::new_blocking(&mut uart, &mut tx, &mut rx, config);
+
+        // break on empty fifo
+        uart.send_break(20).await;
+        uart.blocking_write(&[64]).unwrap();
+        assert_eq!(read::<1>(&mut uart).unwrap_err(), Error::Break);
+        assert_eq!(read(&mut uart).unwrap(), [64]);
+
+        // break on partially filled fifo
+        uart.blocking_write(&[65; 2]).unwrap();
+        uart.send_break(20).await;
+        uart.blocking_write(&[66]).unwrap();
+        assert_eq!(read(&mut uart).unwrap(), [65; 2]);
+        assert_eq!(read::<1>(&mut uart).unwrap_err(), Error::Break);
+        assert_eq!(read(&mut uart).unwrap(), [66]);
+    }
+
+    // parity detection. here we bitbang to not require two uarts.
+    info!("test parity error detection");
+    {
+        let mut pin = Output::new(&mut tx, Level::High);
+        let mut config = Config::default();
+        config.baudrate = 1000;
+        config.parity = Parity::ParityEven;
+        let mut uart = UartRx::new_blocking(&mut uart, &mut rx, config);
+
+        async fn chr(pin: &mut Output<'_, impl embassy_rp::gpio::Pin>, v: u8, parity: u8) {
+            send(pin, v, Some(parity != 0)).await;
+        }
+
+        // first check that we can send correctly
+        chr(&mut pin, 64, 1).await;
+        assert_eq!(read1(&mut uart).unwrap(), [64]);
+
+        // all good, check real errors
+        chr(&mut pin, 2, 1).await;
+        chr(&mut pin, 3, 0).await;
+        chr(&mut pin, 4, 0).await;
+        chr(&mut pin, 5, 0).await;
+        assert_eq!(read1(&mut uart).unwrap(), [2, 3]);
+        assert_eq!(read1::<1>(&mut uart).unwrap_err(), Error::Parity);
+        assert_eq!(read1(&mut uart).unwrap(), [5]);
+    }
 
-    let config = Config::default();
-    let mut uart = Uart::new_blocking(uart, tx, rx, config);
+    // 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);
+        let mut config = Config::default();
+        config.baudrate = 1000;
+        let mut uart = UartRx::new_blocking(&mut uart, &mut rx, config);
 
-    // 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.
+        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;
+            }
+        }
 
-    let data = [0xC0, 0xDE];
-    uart.blocking_write(&data).unwrap();
+        // first check that we can send correctly
+        chr(&mut pin, 64, true).await;
+        assert_eq!(read1(&mut uart).unwrap(), [64]);
 
-    let mut buf = [0; 2];
-    uart.blocking_read(&mut buf).unwrap();
-    assert_eq!(buf, data);
+        // all good, check real errors
+        chr(&mut pin, 2, true).await;
+        chr(&mut pin, 3, true).await;
+        chr(&mut pin, 4, false).await;
+        chr(&mut pin, 5, true).await;
+        assert_eq!(read1(&mut uart).unwrap(), [2, 3]);
+        assert_eq!(read1::<1>(&mut uart).unwrap_err(), Error::Framing);
+        assert_eq!(read1(&mut uart).unwrap(), [5]);
+    }
 
     info!("Test OK");
     cortex_m::asm::bkpt();
diff --git a/tests/rp/src/bin/uart_buffered.rs b/tests/rp/src/bin/uart_buffered.rs
index bea9283e7..1deb22ce6 100644
--- a/tests/rp/src/bin/uart_buffered.rs
+++ b/tests/rp/src/bin/uart_buffered.rs
@@ -2,39 +2,248 @@
 #![no_main]
 #![feature(type_alias_impl_trait)]
 
-use defmt::{assert_eq, *};
+use defmt::{assert_eq, panic, *};
 use embassy_executor::Spawner;
+use embassy_rp::gpio::{Level, Output};
 use embassy_rp::interrupt;
-use embassy_rp::uart::{BufferedUart, Config};
-use embedded_io::asynch::{Read, Write};
+use embassy_rp::uart::{BufferedUart, BufferedUartRx, Config, Error, Instance, Parity};
+use embassy_time::{Duration, Timer};
+use embedded_io::asynch::{Read, ReadExactError, Write};
 use {defmt_rtt as _, panic_probe as _};
 
+async fn read<const N: usize>(uart: &mut BufferedUart<'_, impl Instance>) -> Result<[u8; N], Error> {
+    let mut buf = [255; N];
+    match uart.read_exact(&mut buf).await {
+        Ok(()) => Ok(buf),
+        // we should not ever produce an Eof condition
+        Err(ReadExactError::UnexpectedEof) => panic!(),
+        Err(ReadExactError::Other(e)) => Err(e),
+    }
+}
+
+async fn read1<const N: usize>(uart: &mut BufferedUartRx<'_, impl Instance>) -> Result<[u8; N], Error> {
+    let mut buf = [255; N];
+    match uart.read_exact(&mut buf).await {
+        Ok(()) => Ok(buf),
+        // we should not ever produce an Eof condition
+        Err(ReadExactError::UnexpectedEof) => panic!(),
+        Err(ReadExactError::Other(e)) => Err(e),
+    }
+}
+
+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());
     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 tx_buf = &mut [0u8; 16];
+        let rx_buf = &mut [0u8; 16];
+        let mut uart = BufferedUart::new(&mut uart, &mut irq, &mut tx, &mut rx, tx_buf, rx_buf, config);
+
+        // Make sure we send more bytes than fits in the FIFO, to test the actual
+        // bufferedUart.
+
+        let data = [
+            1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13, 14, 15, 16, 17, 18, 19, 20, 21, 22, 23, 24, 25, 26, 27, 28, 29,
+            30, 31, 32, 33, 34, 35, 36, 37, 38, 39, 40, 41, 42, 43, 44, 45, 46, 47, 48,
+        ];
+        uart.write_all(&data).await.unwrap();
+        info!("Done writing");
+
+        assert_eq!(read(&mut uart).await.unwrap(), data);
+    }
+
+    info!("test overflow detection");
+    {
+        let config = Config::default();
+        let tx_buf = &mut [0u8; 16];
+        let rx_buf = &mut [0u8; 16];
+        let mut uart = BufferedUart::new(&mut uart, &mut irq, &mut tx, &mut rx, tx_buf, rx_buf, config);
+
+        // Make sure we send more bytes than fits in the FIFO, to test the actual
+        // bufferedUart.
+
+        let data = [
+            1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13, 14, 15, 16, 17, 18, 19, 20, 21, 22, 23, 24, 25, 26, 27, 28, 29,
+            30, 31, 32, 33, 34, 35, 36, 37, 38, 39, 40, 41, 42, 43, 44, 45, 46, 47, 48,
+        ];
+        let overflow = [
+            101, 102, 103, 104, 105, 106, 107, 108, 109, 110, 111, 112, 113, 114, 115, 116,
+        ];
+        // give each block time to settle into the fifo. we want the overrun to occur at a well-defined point.
+        uart.write_all(&data).await.unwrap();
+        uart.blocking_flush().unwrap();
+        while uart.busy() {}
+        uart.write_all(&overflow).await.unwrap();
+        uart.blocking_flush().unwrap();
+        while uart.busy() {}
+
+        // already buffered/fifod prefix is valid
+        assert_eq!(read(&mut uart).await.unwrap(), data);
+        // next received character causes overrun error and is discarded
+        uart.write_all(&[1, 2, 3]).await.unwrap();
+        uart.blocking_flush().unwrap();
+        assert_eq!(read::<1>(&mut uart).await.unwrap_err(), Error::Overrun);
+        assert_eq!(read(&mut uart).await.unwrap(), [2, 3]);
+    }
+
+    info!("test break detection");
+    {
+        let mut config = Config::default();
+        config.baudrate = 1000;
+        let tx_buf = &mut [0u8; 16];
+        let rx_buf = &mut [0u8; 16];
+        let mut uart = BufferedUart::new(&mut uart, &mut irq, &mut tx, &mut rx, tx_buf, rx_buf, config);
+
+        // break on empty buffer
+        uart.send_break(20).await;
+        assert_eq!(read::<1>(&mut uart).await.unwrap_err(), Error::Break);
+        uart.write_all(&[64]).await.unwrap();
+        assert_eq!(read(&mut uart).await.unwrap(), [64]);
+
+        // break on partially filled buffer
+        uart.write_all(&[65; 2]).await.unwrap();
+        uart.send_break(20).await;
+        uart.write_all(&[66]).await.unwrap();
+        assert_eq!(read(&mut uart).await.unwrap(), [65; 2]);
+        assert_eq!(read::<1>(&mut uart).await.unwrap_err(), Error::Break);
+        assert_eq!(read(&mut uart).await.unwrap(), [66]);
+
+        // break on full buffer
+        uart.write_all(&[64; 16]).await.unwrap();
+        uart.send_break(20).await;
+        uart.write_all(&[65]).await.unwrap();
+        assert_eq!(read(&mut uart).await.unwrap(), [64; 16]);
+        assert_eq!(read::<1>(&mut uart).await.unwrap_err(), Error::Break);
+        assert_eq!(read(&mut uart).await.unwrap(), [65]);
+    }
+
+    // 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 rx_buf = &mut [0u8; 16];
+        let mut uart = BufferedUartRx::new(&mut uart, &mut irq, &mut rx, rx_buf, 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, 64, 1).await;
+        assert_eq!(read1(&mut uart).await.unwrap(), [64]);
+
+        // parity on empty buffer
+        chr(&mut pin, 64, 0).await;
+        chr(&mut pin, 4, 1).await;
+        assert_eq!(read1::<1>(&mut uart).await.unwrap_err(), Error::Parity);
+        assert_eq!(read1(&mut uart).await.unwrap(), [4]);
+
+        // parity on partially filled buffer
+        chr(&mut pin, 64, 1).await;
+        chr(&mut pin, 32, 1).await;
+        chr(&mut pin, 64, 0).await;
+        chr(&mut pin, 65, 0).await;
+        assert_eq!(read1(&mut uart).await.unwrap(), [64, 32]);
+        assert_eq!(read1::<1>(&mut uart).await.unwrap_err(), Error::Parity);
+        assert_eq!(read1(&mut uart).await.unwrap(), [65]);
+
+        // parity on full buffer
+        for i in 0..16 {
+            chr(&mut pin, i, i.count_ones() % 2).await;
+        }
+        chr(&mut pin, 64, 0).await;
+        chr(&mut pin, 65, 0).await;
+        assert_eq!(
+            read1(&mut uart).await.unwrap(),
+            [0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13, 14, 15]
+        );
+        assert_eq!(read1::<1>(&mut uart).await.unwrap_err(), Error::Parity);
+        assert_eq!(read1(&mut uart).await.unwrap(), [65]);
+    }
+
+    // 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 rx_buf = &mut [0u8; 16];
+        let mut uart = BufferedUartRx::new(&mut uart, &mut irq, &mut rx, rx_buf, 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;
+            }
+        }
 
-    let config = Config::default();
-    let irq = interrupt::take!(UART0_IRQ);
-    let tx_buf = &mut [0u8; 16];
-    let rx_buf = &mut [0u8; 16];
-    let mut uart = BufferedUart::new(uart, irq, tx, rx, tx_buf, rx_buf, config);
+        // first check that we can send correctly
+        chr(&mut pin, 64, true).await;
+        assert_eq!(read1(&mut uart).await.unwrap(), [64]);
 
-    // Make sure we send more bytes than fits in the FIFO, to test the actual
-    // bufferedUart.
+        // framing on empty buffer
+        chr(&mut pin, 64, false).await;
+        assert_eq!(read1::<1>(&mut uart).await.unwrap_err(), Error::Framing);
+        chr(&mut pin, 65, true).await;
+        assert_eq!(read1(&mut uart).await.unwrap(), [65]);
 
-    let data = [
-        1u8, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13, 14, 15, 16, 17, 18, 19, 20, 21, 22, 23, 24, 25, 26, 27, 28, 29,
-        30, 31,
-    ];
-    uart.write_all(&data).await.unwrap();
-    info!("Done writing");
+        // framing on partially filled buffer
+        chr(&mut pin, 64, true).await;
+        chr(&mut pin, 32, true).await;
+        chr(&mut pin, 64, false).await;
+        chr(&mut pin, 65, true).await;
+        assert_eq!(read1(&mut uart).await.unwrap(), [64, 32]);
+        assert_eq!(read1::<1>(&mut uart).await.unwrap_err(), Error::Framing);
+        assert_eq!(read1(&mut uart).await.unwrap(), [65]);
 
-    let mut buf = [0; 31];
-    uart.read_exact(&mut buf).await.unwrap();
-    assert_eq!(buf, data);
+        // framing on full buffer
+        for i in 0..16 {
+            chr(&mut pin, i, true).await;
+        }
+        chr(&mut pin, 64, false).await;
+        chr(&mut pin, 65, true).await;
+        assert_eq!(
+            read1(&mut uart).await.unwrap(),
+            [0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13, 14, 15]
+        );
+        assert_eq!(read1::<1>(&mut uart).await.unwrap_err(), Error::Framing);
+        assert_eq!(read1(&mut uart).await.unwrap(), [65]);
+    }
 
     info!("Test OK");
     cortex_m::asm::bkpt();
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();