UART: make async operations cancellation-safe, update others for consistency (#3142)

* Make async operations cancellable

* Handle thresholds

* Fix written amount calculation

* Fix waiting for events

* Don't return 0 bytes, don't modify TX threshold

* Add read_exact implementation

* Fix tests

* Add tests

* Inline constant into the default values

* Make FIFO config stable

* Fix doc links

* Changelog

* Remove duplicate changelog entries

* Check for RX error

* Fix write_async not recalculating available space

* Fix ESP32 timeout loop

* Check the unmasked interrupt bits to actually detect errors

* Improve naming

* Change async handler to not clear interrupt status

* Test and fix case where write_async returned Ok(0)

* Tweak docs

* Address review feedback

* Use embedded_io to fill buffer in test

* Rename

esp-hal/CHANGELOG.md

@@ -11,6 +11,8 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 - SPI: Added support for 3-wire SPI (#2919)
 - UART: Add separate config for Rx and Tx (#2965)
+- UART: `read_exact_async` (unstable) (#3142)
+- UART: `TxConfig::fifo_empty_threshold` (#3142)
 - Added accessor methods to config structs (#3011)
 - `esp_hal::time::{Rate, Duration, Instant}` (#3083)
 - Async support for ADC oneshot reads for ESP32C2, ESP32C3, ESP32C6 and ESP32H2 (#2925, #3082)
@@ -24,11 +26,6 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 - RMT: `TxChannelConfig` and `RxChannelConfig` now support the builder-lite pattern (#2978)
 - RMT: Some fields of `TxChannelConfig` and `RxChannelConfig` are now `gpio::Level`-valued instead of `bool` (#2989)
 - RMT: The `PulseCode` trait now uses `gpio::Level` to specify output levels instead of `bool` (#2989)
-- Uart `write_bytes` and `read_bytes` are now blocking and return the number of bytes written/read (#2882)
-- Uart `read_bytes` is now blocking  returns the number of bytes read (#2882)
-- Uart `flush` is now blocking (#2882)
-- Uart errors have been split into `RxError` and `TxError`. A combined `IoError` has been created for embedded-io. (#3138)
-- `{Uart, UartTx}::flush()` is now fallible. (#3138)
 - Removed `embedded-hal-nb` traits (#2882)
 - `timer::wait` is now blocking (#2882)
 - By default, set `tx_idle_num` to 0 so that bytes written to TX FIFO are always immediately transmitted. (#2859)
@@ -46,12 +43,13 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 - The `system::RadioClockController` trait has been replaced by the `clock::RadioClockController` struct. (#3100)
 - The `Cpu` struct and contents of the `reset` and `cpu_control` modules have been moved into `cpu`. (#3099)
 - The `software_reset_cpu` now takes which CPU to reset as parameter. (#3099)
-- The `Cpu` struct and contents of the `reset` and `cpu_control` modules have been moved into `cpu`. (#)
-- The `software_reset_cpu` now takes which CPU to reset as parameter. (#)
 - `read_bytes` and `write_bytes` methods on drivers have been renamed to `read` and `write` (#3137)
 - `Uart::write` and `Uart::read` are now blocking and return the number of bytes written/read (#2882)
 - `Uart::flush` is now blocking (#2882)
 - `Uart::split` and the respective split halves have been marked as unstable (#3137)
+- Uart errors have been split into `RxError` and `TxError`. A combined `IoError` has been created for embedded-io. (#3138)
+- `{Uart, UartTx}::flush()` is now fallible. (#3138)
+- `Uart::{read_async, write_async}` are now cancellation-safe (#3142)
 
 - I2C: Async functions are postfixed with `_async`, non-async functions are available in async-mode (#3056)
 

examples/src/bin/ieee802154_sniffer.rs

@@ -77,7 +77,7 @@ fn main() -> ! {
         }
 
         let mut buf = [0u8; 1];
-        if let Ok(_) = uart0.read_buffered_bytes(&mut buf) {
+        if let Ok(_) = uart0.read_buffered(&mut buf) {
             if buf[0] == b'r' {
                 software_reset();
             }

hil-test/tests/uart.rs

@@ -82,13 +82,7 @@ mod tests {
 
         // Calls to read may not fill the buffer, wait until read returns 0
         let mut buffer = [0; BUF_SIZE];
-        let mut n = 0;
-        loop {
-            match ctx.uart.read(&mut buffer[n..]).unwrap() {
-                0 => break,
-                cnt => n += cnt,
-            };
-        }
+        embedded_io::Read::read_exact(&mut ctx.uart, &mut buffer).unwrap();
 
         assert_eq!(data, buffer);
     }

hil-test/tests/uart_regression.rs

@@ -36,7 +36,7 @@ mod tests {
             .unwrap()
             .with_tx(tx);
 
-        tx.flush();
+        tx.flush().unwrap();
         tx.write(&[0x42]).unwrap();
         let mut byte = [0u8; 1];
         rx.read(&mut byte).unwrap();

hil-test/tests/uart_tx_rx.rs

@@ -42,7 +42,7 @@ mod tests {
     fn test_send_receive(mut ctx: Context) {
         let byte = [0x42];
 
-        ctx.tx.flush();
+        ctx.tx.flush().unwrap();
         ctx.tx.write(&byte).unwrap();
         let mut buf = [0u8; 1];
         ctx.rx.read(&mut buf).unwrap();
@@ -55,17 +55,10 @@ mod tests {
         let bytes = [0x42, 0x43, 0x44];
         let mut buf = [0u8; 3];
 
-        ctx.tx.flush();
+        ctx.tx.flush().unwrap();
         ctx.tx.write(&bytes).unwrap();
 
-        // Calls to read may not fill the buffer, wait until read returns 0
-        let mut n = 0;
-        loop {
-            match ctx.rx.read(&mut buf[n..]).unwrap() {
-                0 => break,
-                cnt => n += cnt,
-            };
-        }
+        embedded_io::Read::read_exact(&mut ctx.rx, &mut buf).unwrap();
 
         assert_eq!(buf, bytes);
     }

hil-test/tests/uart_tx_rx_async.rs

@@ -1,7 +1,7 @@
 //! UART TX/RX Async Test
 
 //% CHIPS: esp32 esp32c2 esp32c3 esp32c6 esp32h2 esp32s2 esp32s3
-//% FEATURES: unstable
+//% FEATURES: unstable embassy
 
 #![no_std]
 #![no_main]
@@ -20,6 +20,16 @@ struct Context {
 #[cfg(test)]
 #[embedded_test::tests(default_timeout = 3, executor = hil_test::Executor::new())]
 mod tests {
+    use embassy_futures::{
+        join::join,
+        select::{select, Either},
+    };
+    use embassy_time::{Duration, Timer};
+    use esp_hal::{
+        timer::timg::TimerGroup,
+        uart::{RxConfig, RxError},
+    };
+
     use super::*;
 
     #[init]
@@ -28,6 +38,9 @@ mod tests {
 
         let (rx, tx) = hil_test::common_test_pins!(peripherals);
 
+        let timg0 = TimerGroup::new(peripherals.TIMG0);
+        esp_hal_embassy::init(timg0.timer0);
+
         let tx = UartTx::new(peripherals.UART0, uart::Config::default())
             .unwrap()
             .with_tx(tx)
@@ -41,7 +54,7 @@ mod tests {
     }
 
     #[test]
-    async fn test_send_receive(mut ctx: Context) {
+    async fn test_write_read(mut ctx: Context) {
         let byte = [0x42];
         let mut read = [0u8; 1];
 
@@ -51,4 +64,133 @@ mod tests {
 
         assert_eq!(read, byte);
     }
+
+    #[test]
+    async fn write_async_does_not_return_0(mut ctx: Context) {
+        let bytes = [0; 200];
+
+        for i in 0..50 {
+            let written = ctx.tx.write_async(&bytes).await.unwrap();
+            assert_ne!(written, 0, "Iteration {}", i);
+        }
+    }
+
+    #[test]
+    async fn rx_overflow_is_detected(mut ctx: Context) {
+        let mut to_send: &[u8] = &[0; 250];
+        let mut read = [0u8; 1];
+
+        while !to_send.is_empty() {
+            let written = ctx.tx.write_async(to_send).await.unwrap();
+            to_send = &to_send[written..];
+        }
+        ctx.tx.flush_async().await.unwrap();
+
+        // The read is supposed to return an error because the FIFO is just 128 bytes
+        // long.
+        let res = ctx.rx.read_async(&mut read).await;
+        assert!(matches!(res, Err(RxError::FifoOverflowed)), "{:?}", res);
+    }
+
+    #[test]
+    async fn read_returns_with_partial_data_if_read_times_out(mut ctx: Context) {
+        let mut data = [0u8; 16];
+
+        // We write 4 bytes, but read 16. This should return 4 bytes because the default
+        // RX timeout is 10 symbols' worth of time.
+        let (read, _) = join(ctx.rx.read_async(&mut data), async {
+            ctx.tx.write_async(&[1, 2, 3, 4]).await.unwrap();
+            ctx.tx.flush_async().await.unwrap();
+        })
+        .await;
+
+        assert_eq!(read, Ok(4));
+        assert_eq!(&data[..4], &[1, 2, 3, 4]);
+    }
+
+    #[test]
+    async fn read_exact_does_not_return_partial_data(mut ctx: Context) {
+        let mut data = [0u8; 8];
+
+        // Write 4 bytes
+        ctx.tx.write_async(&[1, 2, 3, 4]).await.unwrap();
+        ctx.tx.flush_async().await.unwrap();
+
+        // read_exact should either read exactly 8 bytes, or time out.
+        let should_timeout = select(
+            // Wait for 8 bytes or FIFO threshold which is more.
+            ctx.rx.read_exact_async(&mut data),
+            // We expect to time out
+            Timer::after(Duration::from_secs(1)),
+        )
+        .await;
+
+        assert!(matches!(should_timeout, Either::Second(_)));
+    }
+
+    #[test]
+    async fn read_does_not_resolve_when_there_is_not_enough_data_and_no_timeout(mut ctx: Context) {
+        let mut data = [0u8; 8];
+
+        ctx.rx
+            .apply_config(&uart::Config::default().with_rx(RxConfig::default().with_timeout_none()))
+            .unwrap();
+        // Default RX FIFO threshold is higher than 8 bytes, so we should be able to
+        // read all 8 bytes in one go.
+        let (should_timeout, _) = join(
+            select(
+                // Wait for 8 bytes or FIFO threshold which is more.
+                ctx.rx.read_async(&mut data),
+                // We expect to time out
+                Timer::after(Duration::from_secs(1)),
+            ),
+            async {
+                // Write 4 bytes
+                ctx.tx.write_async(&[1, 2, 3, 4]).await.unwrap();
+                ctx.tx.flush_async().await.unwrap();
+            },
+        )
+        .await;
+
+        assert!(matches!(should_timeout, Either::Second(_)));
+    }
+
+    #[test]
+    async fn read_resolves_when_buffer_is_full(mut ctx: Context) {
+        let mut data = [0u8; 8];
+
+        ctx.rx
+            .apply_config(&uart::Config::default().with_rx(RxConfig::default().with_timeout_none()))
+            .unwrap();
+        // Default RX FIFO threshold is higher than 8 bytes, so we should be able to
+        // read all 8 bytes in one go.
+        let (read, _) = join(ctx.rx.read_async(&mut data), async {
+            ctx.tx.write_async(&[1, 2, 3, 4, 5, 6, 7, 8]).await.unwrap();
+            ctx.tx.flush_async().await.unwrap();
+        })
+        .await;
+
+        assert_eq!(read, Ok(8));
+        assert_eq!(&data, &[1, 2, 3, 4, 5, 6, 7, 8]);
+    }
+
+    #[test]
+    async fn cancelling_read_does_not_drop_data(mut ctx: Context) {
+        let mut data = [0u8; 16];
+
+        let should_be_tx = select(ctx.rx.read_async(&mut data), async {
+            ctx.tx.write_async(&[1, 2, 3]).await.unwrap();
+            ctx.tx.flush_async().await.unwrap();
+        })
+        .await;
+
+        // Default RX FIFO threshold should be more than 3 bytes, so we don't expect
+        // read to become ready.
+        assert!(matches!(should_be_tx, Either::Second(_)));
+
+        let read = ctx.rx.read_async(&mut data).await;
+
+        assert_eq!(read, Ok(3));
+        assert_eq!(&data[..3], &[1, 2, 3]);
+    }
 }