Merge #783

783: Reimplement BufRead for BufferedUart r=Dirbaio a=chemicstry

The `AsyncBufRead` implementation for `BufferedUart` was removed in https://github.com/embassy-rs/embassy/pull/752, this PR reimplements it from `embedded-io`. This allows reading `BufferedUart` without copying slices.

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

5 files changed, 121 insertions, 15 deletions

diff --git a/embassy-nrf/src/buffered_uarte.rs b/embassy-nrf/src/buffered_uarte.rs
index c42fa1138..e1d32a311 100644
--- a/embassy-nrf/src/buffered_uarte.rs
+++ b/embassy-nrf/src/buffered_uarte.rs
@@ -236,6 +236,48 @@ impl<'d, U: UarteInstance, T: TimerInstance> embedded_io::asynch::Read for Buffe
     }
 }
 
+impl<'d, U: UarteInstance, T: TimerInstance> embedded_io::asynch::BufRead
+    for BufferedUarte<'d, U, 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);
+                trace!("fill_buf");
+
+                // We have data ready in buffer? Return it.
+                let buf = state.rx.pop_buf();
+                if !buf.is_empty() {
+                    trace!("  got {:?} {:?}", buf.as_ptr() as u32, buf.len());
+                    let buf: &[u8] = buf;
+                    // Safety: buffer lives as long as uart
+                    let buf: &[u8] = unsafe { core::mem::transmute(buf) };
+                    return Poll::Ready(Ok(buf));
+                }
+
+                trace!("  empty");
+                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, U: UarteInstance, T: TimerInstance> embedded_io::asynch::Write
     for BufferedUarte<'d, U, T>
 {
diff --git a/embassy-stm32/src/usart/buffered.rs b/embassy-stm32/src/usart/buffered.rs
index b2fcb504e..7b63638a0 100644
--- a/embassy-stm32/src/usart/buffered.rs
+++ b/embassy-stm32/src/usart/buffered.rs
@@ -191,6 +191,43 @@ impl<'d, T: Instance> embedded_io::asynch::Read for BufferedUart<'d, T> {
     }
 }
 
+impl<'d, T: Instance> 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> embedded_io::asynch::Write for BufferedUart<'d, T> {
     type WriteFuture<'a> = impl Future<Output = Result<usize, Self::Error>>
     where
diff --git a/examples/nrf/src/bin/buffered_uart.rs b/examples/nrf/src/bin/buffered_uart.rs
index a64c5821b..782c39499 100644
--- a/examples/nrf/src/bin/buffered_uart.rs
+++ b/examples/nrf/src/bin/buffered_uart.rs
@@ -6,7 +6,7 @@ use defmt::*;
 use embassy::executor::Spawner;
 use embassy_nrf::buffered_uarte::State;
 use embassy_nrf::{buffered_uarte::BufferedUarte, interrupt, uarte, Peripherals};
-use embedded_io::asynch::{Read, Write};
+use embedded_io::asynch::{BufRead, Write};
 use futures::pin_mut;
 
 use defmt_rtt as _; // global logger
@@ -46,23 +46,13 @@ async fn main(_spawner: Spawner, p: Peripherals) {
     unwrap!(u.write_all(b"Hello!\r\n").await);
     info!("wrote hello in uart!");
 
-    // Simple demo, reading 8-char chunks and echoing them back reversed.
     loop {
         info!("reading...");
-        let mut buf = [0u8; 8];
-        unwrap!(u.read_exact(&mut buf).await);
+        let buf = unwrap!(u.fill_buf().await);
         info!("read done, got {}", buf);
 
-        // Reverse buf
-        for i in 0..4 {
-            buf.swap(i, 7 - i);
-        }
-
-        info!("writing...");
-        unwrap!(u.write_all(&buf).await);
-        info!("write done");
-
-        // Wait until the bytes are actually finished being transmitted
-        unwrap!(u.flush().await);
+        // Read bytes have to be explicitly consumed, otherwise fill_buf() will return them again
+        let n = buf.len();
+        u.consume(n);
     }
 }
diff --git a/examples/stm32f4/Cargo.toml b/examples/stm32f4/Cargo.toml
index e2065bed9..651d01388 100644
--- a/examples/stm32f4/Cargo.toml
+++ b/examples/stm32f4/Cargo.toml
@@ -16,6 +16,7 @@ defmt-rtt = "0.3"
 cortex-m = "0.7.3"
 cortex-m-rt = "0.7.0"
 embedded-hal = "0.2.6"
+embedded-io = "0.3.0"
 panic-probe = { version = "0.3", features = ["print-defmt"] }
 futures = { version = "0.3.17", default-features = false, features = ["async-await"] }
 heapless = { version = "0.7.5", default-features = false }
diff --git a/examples/stm32f4/src/bin/usart_buffered.rs b/examples/stm32f4/src/bin/usart_buffered.rs
new file mode 100644
index 000000000..80b65f0d4
--- /dev/null
+++ b/examples/stm32f4/src/bin/usart_buffered.rs
@@ -0,0 +1,36 @@
+#![no_std]
+#![no_main]
+#![feature(type_alias_impl_trait)]
+
+use defmt::*;
+use defmt_rtt as _; // global logger
+use embassy::executor::Spawner;
+use embassy_stm32::dma::NoDma;
+use embassy_stm32::usart::{BufferedUart, Config, State, Uart};
+use embassy_stm32::{interrupt, Peripherals};
+use embedded_io::asynch::BufRead;
+use panic_probe as _;
+
+#[embassy::main]
+async fn main(_spawner: Spawner, p: Peripherals) {
+    info!("Hello World!");
+
+    let config = Config::default();
+    let usart = Uart::new(p.USART3, p.PD9, p.PD8, NoDma, NoDma, config);
+
+    let mut state = State::new();
+    let irq = interrupt::take!(USART3);
+    let mut tx_buf = [0u8; 32];
+    let mut rx_buf = [0u8; 32];
+    let mut buf_usart =
+        unsafe { BufferedUart::new(&mut state, usart, irq, &mut tx_buf, &mut rx_buf) };
+
+    loop {
+        let buf = buf_usart.fill_buf().await.unwrap();
+        info!("Received: {}", buf);
+
+        // Read bytes have to be explicitly consumed, otherwise fill_buf() will return them again
+        let n = buf.len();
+        buf_usart.consume(n);
+    }
+}