Add BufferedUart implementation, and feature-guard time-driver initialization, to free up TIMER peripheral if not used with embassy executor

3 files changed, 369 insertions, 2 deletions

diff --git a/embassy-rp/Cargo.toml b/embassy-rp/Cargo.toml
index c43fd7e7..211b6a40 100644
--- a/embassy-rp/Cargo.toml
+++ b/embassy-rp/Cargo.toml
@@ -52,6 +52,7 @@ cortex-m = "0.7.6"
 critical-section = "1.1"
 futures = { version = "0.3.17", default-features = false, features = ["async-await"] }
 chrono = { version = "0.4", default-features = false, optional = true }
+embedded-io = { version = "0.3.0", features = ["async"], optional = true }
 
 rp2040-pac2 = { git = "https://github.com/embassy-rs/rp2040-pac2", rev="017e3c9007b2d3b6965f0d85b5bf8ce3fa6d7364", features = ["rt"] }
 #rp2040-pac2 = { path = "../../rp2040-pac2", features = ["rt"] }
diff --git a/embassy-rp/src/uart/buffered.rs b/embassy-rp/src/uart/buffered.rs
new file mode 100644
index 00000000..c31af801
--- /dev/null
+++ b/embassy-rp/src/uart/buffered.rs
@@ -0,0 +1,286 @@
+use core::future::Future;
+use core::task::Poll;
+
+use atomic_polyfill::{compiler_fence, Ordering};
+use embassy_cortex_m::peripheral::{PeripheralMutex, PeripheralState, StateStorage};
+use embassy_hal_common::ring_buffer::RingBuffer;
+use embassy_sync::waitqueue::WakerRegistration;
+use futures::future::poll_fn;
+
+use super::*;
+
+pub struct State<'d, T: Instance>(StateStorage<StateInner<'d, T>>);
+impl<'d, T: Instance> State<'d, T> {
+    pub fn new() -> Self {
+        Self(StateStorage::new())
+    }
+}
+
+struct StateInner<'d, T: Instance> {
+    phantom: PhantomData<&'d mut T>,
+
+    rx_waker: WakerRegistration,
+    rx: RingBuffer<'d>,
+
+    tx_waker: WakerRegistration,
+    tx: RingBuffer<'d>,
+}
+
+unsafe impl<'d, T: Instance> Send for StateInner<'d, T> {}
+unsafe impl<'d, T: Instance> Sync for StateInner<'d, T> {}
+
+pub struct BufferedUart<'d, T: Instance> {
+    inner: PeripheralMutex<'d, StateInner<'d, T>>,
+}
+
+impl<'d, T: Instance> Unpin for BufferedUart<'d, T> {}
+
+impl<'d, T: Instance> BufferedUart<'d, T> {
+    pub fn new<M: Mode>(
+        state: &'d mut State<'d, T>,
+        _uart: Uart<'d, T, M>,
+        irq: impl Peripheral<P = T::Interrupt> + 'd,
+        tx_buffer: &'d mut [u8],
+        rx_buffer: &'d mut [u8],
+    ) -> BufferedUart<'d, T> {
+        into_ref!(irq);
+
+        let r = T::regs();
+        unsafe {
+            r.uartimsc().modify(|w| {
+                // TODO: Should and more or fewer interrupts be enabled?
+                w.set_rxim(true);
+                w.set_rtim(true);
+            });
+        }
+
+        Self {
+            inner: PeripheralMutex::new(irq, &mut state.0, move || StateInner {
+                phantom: PhantomData,
+                tx: RingBuffer::new(tx_buffer),
+                tx_waker: WakerRegistration::new(),
+
+                rx: RingBuffer::new(rx_buffer),
+                rx_waker: WakerRegistration::new(),
+            }),
+        }
+    }
+}
+
+impl<'d, T: Instance> StateInner<'d, T>
+where
+    Self: 'd,
+{
+    fn on_rx(&mut self) {
+        let r = T::regs();
+        unsafe {
+            let ris = r.uartris().read();
+            // Clear interrupt flags
+            r.uarticr().write(|w| {
+                w.set_rxic(true);
+                w.set_rtic(true);
+            });
+
+            if ris.rxris() {
+                if ris.peris() {
+                    warn!("Parity error");
+                }
+                if ris.feris() {
+                    warn!("Framing error");
+                }
+                if ris.beris() {
+                    warn!("Break error");
+                }
+                if ris.oeris() {
+                    warn!("Overrun error");
+                }
+
+                let buf = self.rx.push_buf();
+                if !buf.is_empty() {
+                    buf[0] = r.uartdr().read().data();
+                    self.rx.push(1);
+                } else {
+                    warn!("RX buffer full, discard received byte");
+                }
+
+                if self.rx.is_full() {
+                    self.rx_waker.wake();
+                }
+            }
+
+            if ris.rtris() {
+                self.rx_waker.wake();
+            };
+        }
+    }
+
+    fn on_tx(&mut self) {
+        let r = T::regs();
+        unsafe {
+            let ris = r.uartris().read();
+            // Clear interrupt flags
+            r.uarticr().write(|w| {
+                w.set_rtic(true);
+            });
+
+            if ris.txris() {
+                let buf = self.tx.pop_buf();
+                if !buf.is_empty() {
+                    r.uartimsc().modify(|w| {
+                        w.set_txim(true);
+                    });
+                    r.uartdr().write(|w| w.set_data(buf[0].into()));
+                    self.tx.pop(1);
+                    self.tx_waker.wake();
+                } else {
+                    // Disable interrupt until we have something to transmit again
+                    r.uartimsc().modify(|w| {
+                        w.set_txim(false);
+                    });
+                }
+            }
+        }
+    }
+}
+
+impl<'d, T: Instance> PeripheralState for StateInner<'d, T>
+where
+    Self: 'd,
+{
+    type Interrupt = T::Interrupt;
+    fn on_interrupt(&mut self) {
+        self.on_rx();
+        self.on_tx();
+    }
+}
+
+impl embedded_io::Error for Error {
+    fn kind(&self) -> embedded_io::ErrorKind {
+        embedded_io::ErrorKind::Other
+    }
+}
+
+impl<'d, T: Instance> embedded_io::Io for BufferedUart<'d, T> {
+    type Error = Error;
+}
+
+impl<'d, T: Instance + 'd> embedded_io::asynch::Read for BufferedUart<'d, T> {
+    type ReadFuture<'a> = impl Future<Output = Result<usize, Self::Error>> 
+    where
+        Self: 'a;
+
+    fn read<'a>(&'a mut self, buf: &'a mut [u8]) -> Self::ReadFuture<'a> {
+        poll_fn(move |cx| {
+            let mut do_pend = false;
+            let res = self.inner.with(|state| {
+                compiler_fence(Ordering::SeqCst);
+
+                // We have data ready in buffer? Return it.
+                let data = state.rx.pop_buf();
+                if !data.is_empty() {
+                    let len = data.len().min(buf.len());
+                    buf[..len].copy_from_slice(&data[..len]);
+
+                    if state.rx.is_full() {
+                        do_pend = true;
+                    }
+                    state.rx.pop(len);
+
+                    return Poll::Ready(Ok(len));
+                }
+
+                state.rx_waker.register(cx.waker());
+                Poll::Pending
+            });
+
+            if do_pend {
+                self.inner.pend();
+            }
+
+            res
+        })
+    }
+}
+
+impl<'d, T: Instance + 'd> embedded_io::asynch::BufRead for BufferedUart<'d, T> {
+    type FillBufFuture<'a> = impl Future<Output = Result<&'a [u8], Self::Error>>
+    where
+        Self: 'a;
+
+    fn fill_buf<'a>(&'a mut self) -> Self::FillBufFuture<'a> {
+        poll_fn(move |cx| {
+            self.inner.with(|state| {
+                compiler_fence(Ordering::SeqCst);
+
+                // We have data ready in buffer? Return it.
+                let buf = state.rx.pop_buf();
+                if !buf.is_empty() {
+                    let buf: &[u8] = buf;
+                    // Safety: buffer lives as long as uart
+                    let buf: &[u8] = unsafe { core::mem::transmute(buf) };
+                    return Poll::Ready(Ok(buf));
+                }
+
+                state.rx_waker.register(cx.waker());
+                Poll::<Result<&[u8], Self::Error>>::Pending
+            })
+        })
+    }
+
+    fn consume(&mut self, amt: usize) {
+        let signal = self.inner.with(|state| {
+            let full = state.rx.is_full();
+            state.rx.pop(amt);
+            full
+        });
+        if signal {
+            self.inner.pend();
+        }
+    }
+}
+
+impl<'d, T: Instance + 'd> embedded_io::asynch::Write for BufferedUart<'d, T> {
+    type WriteFuture<'a> = impl Future<Output = Result<usize, Self::Error>>
+    where
+        Self: 'a;
+
+    fn write<'a>(&'a mut self, buf: &'a [u8]) -> Self::WriteFuture<'a> {
+        poll_fn(move |cx| {
+            let (poll, empty) = self.inner.with(|state| {
+                let empty = state.tx.is_empty();
+                let tx_buf = state.tx.push_buf();
+                if tx_buf.is_empty() {
+                    state.tx_waker.register(cx.waker());
+                    return (Poll::Pending, empty);
+                }
+
+                let n = core::cmp::min(tx_buf.len(), buf.len());
+                tx_buf[..n].copy_from_slice(&buf[..n]);
+                state.tx.push(n);
+
+                (Poll::Ready(Ok(n)), empty)
+            });
+            if empty {
+                self.inner.pend();
+            }
+            poll
+        })
+    }
+
+    type FlushFuture<'a> = impl Future<Output = Result<(), Self::Error>>
+    where
+        Self: 'a;
+
+    fn flush<'a>(&'a mut self) -> Self::FlushFuture<'a> {
+        poll_fn(move |cx| {
+            self.inner.with(|state| {
+                if !state.tx.is_empty() {
+                    state.tx_waker.register(cx.waker());
+                    return Poll::Pending;
+                }
+
+                Poll::Ready(Ok(()))
+            })
+        })
+    }
+}
diff --git a/embassy-rp/src/uart.rs b/embassy-rp/src/uart/mod.rs
index 987b716b..3b71d87b 100644
--- a/embassy-rp/src/uart.rs
+++ b/embassy-rp/src/uart/mod.rs
@@ -475,6 +475,76 @@ mod eh1 {
     impl<'d, T: Instance, M: Mode> embedded_hal_1::serial::ErrorType for UartRx<'d, T, M> {
         type Error = Error;
     }
+
+    impl<'d, T: Instance, M: Mode> embedded_hal_1::serial::nb::Read for UartRx<'d, T, M> {
+        fn read(&mut self) -> nb::Result<u8, Self::Error> {
+            let r = T::regs();
+            unsafe {
+                let dr = r.uartdr().read();
+
+                if dr.oe() {
+                    Err(nb::Error::Other(Error::Overrun))
+                } else if dr.be() {
+                    Err(nb::Error::Other(Error::Break))
+                } else if dr.pe() {
+                    Err(nb::Error::Other(Error::Parity))
+                } else if dr.fe() {
+                    Err(nb::Error::Other(Error::Framing))
+                } else if dr.fe() {
+                    Ok(dr.data())
+                } else {
+                    Err(nb::Error::WouldBlock)
+                }
+            }
+        }
+    }
+
+    impl<'d, T: Instance, M: Mode> embedded_hal_1::serial::blocking::Write for UartTx<'d, T, M> {
+        fn write(&mut self, buffer: &[u8]) -> Result<(), Self::Error> {
+            self.blocking_write(buffer)
+        }
+
+        fn flush(&mut self) -> Result<(), Self::Error> {
+            self.blocking_flush()
+        }
+    }
+
+    impl<'d, T: Instance, M: Mode> embedded_hal_1::serial::nb::Write for UartTx<'d, T, M> {
+        fn write(&mut self, char: u8) -> nb::Result<(), Self::Error> {
+            self.blocking_write(&[char]).map_err(nb::Error::Other)
+        }
+
+        fn flush(&mut self) -> nb::Result<(), Self::Error> {
+            self.blocking_flush().map_err(nb::Error::Other)
+        }
+    }
+
+    impl<'d, T: Instance, M: Mode> embedded_hal_1::serial::nb::Read for Uart<'d, T, M> {
+        fn read(&mut self) -> Result<u8, nb::Error<Self::Error>> {
+            embedded_hal_02::serial::Read::read(&mut self.rx)
+        }
+    }
+
+    impl<'d, T: Instance, M: Mode> embedded_hal_1::serial::blocking::Write for Uart<'d, T, M> {
+        fn write(&mut self, buffer: &[u8]) -> Result<(), Self::Error> {
+            self.blocking_write(buffer)
+        }
+
+        fn flush(&mut self) -> Result<(), Self::Error> {
+            self.blocking_flush()
+        }
+    }
+
+    impl<'d, T: Instance, M: Mode> embedded_hal_1::serial::nb::Write for Uart<'d, T, M> {
+        fn write(&mut self, char: u8) -> nb::Result<(), Self::Error> {
+            self.blocking_write(&[char]).map_err(nb::Error::Other)
+        }
+
+        fn flush(&mut self) -> nb::Result<(), Self::Error> {
+            self.blocking_flush().map_err(nb::Error::Other)
+        }
+    }
+
 }
 
 #[cfg(all(
@@ -532,6 +602,12 @@ mod eha {
     }
 }
 
+#[cfg(feature = "nightly")]
+mod buffered;
+#[cfg(feature = "nightly")]
+pub use buffered::*;
+
+
 mod sealed {
     use super::*;
 
@@ -541,6 +617,8 @@ mod sealed {
         const TX_DREQ: u8;
         const RX_DREQ: u8;
 
+        type Interrupt: crate::interrupt::Interrupt;
+
         fn regs() -> pac::uart::Uart;
     }
     pub trait TxPin<T: Instance> {}
@@ -571,6 +649,8 @@ macro_rules! impl_instance {
         impl sealed::Instance for peripherals::$inst {
             const TX_DREQ: u8 = $tx_dreq;
             const RX_DREQ: u8 = $rx_dreq;
+            
+            type Interrupt = crate::interrupt::$irq;
 
             fn regs() -> pac::uart::Uart {
                 pac::$inst
@@ -580,8 +660,8 @@ macro_rules! impl_instance {
     };
 }
 
-impl_instance!(UART0, UART0, 20, 21);
-impl_instance!(UART1, UART1, 22, 23);
+impl_instance!(UART0, UART0_IRQ, 20, 21);
+impl_instance!(UART1, UART1_IRQ, 22, 23);
 
 pub trait TxPin<T: Instance>: sealed::TxPin<T> + crate::gpio::Pin {}
 pub trait RxPin<T: Instance>: sealed::RxPin<T> + crate::gpio::Pin {}