RMT: add poll() method to blocking transactions

this allows properly interleaving several such transactions (as the HIL
loopback tests do, in fact) or other work

Author: wisp3rwind

esp-hal/CHANGELOG.md

@@ -23,6 +23,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 - RMT `TxChannelCreator` and `RxChannelCreator` now have a `DriverMode` generic parameter; `TxChannelCreatorAsync` and `RxChannelCreatorAsync` have been removed. (#3505)
 - RMT `ChannelCreator` methods have been renamed from `configure` to `configure_tx` and `configure_rx` to avoid trait disambiguation issues. (#3505)
 - The RMT `Error` type has been marked `non_exhaustive` (#3701)
+- Blocking RMT transactions can now be `poll`ed without blocking, returning whether they have completed. (#3716)
 
 ### Fixed
 

esp-hal/src/rmt.rs

@@ -998,6 +998,10 @@ where
 }
 
 /// An in-progress transaction for a single shot TX transaction.
+///
+/// If the data size exceeds the size of the internal buffer, `.poll()` or
+/// `.wait()` needs to be called before the entire buffer has been sent to avoid
+/// underruns.
 pub struct SingleShotTxTransaction<'a, Raw>
 where
     Raw: TxChannelInternal,
@@ -1017,15 +1021,61 @@ impl<Raw> SingleShotTxTransaction<'_, Raw>
 where
     Raw: TxChannelInternal,
 {
-    /// Wait for the transaction to complete
-    pub fn wait(mut self) -> Result<Channel<Blocking, Raw>, (Error, Channel<Blocking, Raw>)> {
+    fn poll_internal(&mut self) -> Option<Event> {
         let raw = self.channel.raw;
-        let memsize = raw.memsize().codes();
 
+        let status = raw.get_tx_status();
+        if status == Some(Event::Threshold) {
+            raw.reset_tx_threshold_set();
+
+            if !self.remaining_data.is_empty() {
+                // re-fill TX RAM
+                let memsize = raw.memsize().codes();
+                let ptr = unsafe { raw.channel_ram_start().add(self.ram_index) };
+                let count = self.remaining_data.len().min(memsize / 2);
+                let (chunk, remaining) = self.remaining_data.split_at(count);
+                for (idx, entry) in chunk.iter().enumerate() {
+                    unsafe {
+                        ptr.add(idx).write_volatile(*entry);
+                    }
+                }
+
+                // If count == memsize / 2 codes were written, update ram_index as
+                // - 0 -> memsize / 2
+                // - memsize / 2 -> 0
+                // Otherwise, for count < memsize / 2, the new position is invalid but the new
+                // slice is empty and we won't use ram_index again.
+                self.ram_index = memsize / 2 - self.ram_index;
+                self.remaining_data = remaining;
+                debug_assert!(
+                    self.ram_index == 0
+                        || self.ram_index == memsize / 2
+                        || self.remaining_data.is_empty()
+                );
+            }
+        }
+
+        status
+    }
+
+    /// Check transmission status and write new data to the hardware if
+    /// necessary.
+    ///
+    /// Returns whether transmission has ended (whether successfully or with an
+    /// error). In that case, a subsequent call to `wait()` returns immediately.
+    pub fn poll(&mut self) -> bool {
+        match self.poll_internal() {
+            Some(Event::Error | Event::End) => true,
+            Some(Event::Threshold) | None => false,
+        }
+    }
+
+    /// Wait for the transaction to complete
+    pub fn wait(mut self) -> Result<Channel<Blocking, Raw>, (Error, Channel<Blocking, Raw>)> {
+        // Not sure that all the error cases below can happen. However, it's best to
+        // handle them to be sure that we don't lock up here in case they can happen.
         loop {
-            // Not sure that all the error cases below can happen. However, it's best to
-            // handle them to be sure that we don't lock up here in case they can happen.
-            match raw.get_tx_status() {
+            match self.poll_internal() {
                 Some(Event::Error) => break Err((Error::TransmissionError, self.channel)),
                 Some(Event::End) => {
                     if !self.remaining_data.is_empty() {
@@ -1036,34 +1086,6 @@ where
                         break Ok(self.channel);
                     }
                 }
-                Some(Event::Threshold) => {
-                    raw.reset_tx_threshold_set();
-
-                    if !self.remaining_data.is_empty() {
-                        // re-fill TX RAM
-                        let ptr = unsafe { raw.channel_ram_start().add(self.ram_index) };
-                        let count = self.remaining_data.len().min(memsize / 2);
-                        let (chunk, remaining) = self.remaining_data.split_at(count);
-                        for (idx, entry) in chunk.iter().enumerate() {
-                            unsafe {
-                                ptr.add(idx).write_volatile(*entry);
-                            }
-                        }
-
-                        // If count == memsize / 2 codes were written, update ram_index as
-                        // - 0 -> memsize / 2
-                        // - memsize / 2 -> 0
-                        // Otherwise, for count < memsize / 2, the new position is invalid but the new
-                        // slice is empty and we won't use ram_index again.
-                        self.ram_index = memsize / 2 - self.ram_index;
-                        self.remaining_data = remaining;
-                        debug_assert!(
-                            self.ram_index == 0
-                                || self.ram_index == memsize / 2
-                                || self.remaining_data.is_empty()
-                        );
-                    }
-                }
                 _ => continue,
             }
         }
@@ -1298,32 +1320,53 @@ pub struct RxTransaction<'a, Raw: RxChannelInternal> {
 }
 
 impl<Raw: RxChannelInternal> RxTransaction<'_, Raw> {
-    /// Wait for the transaction to complete
-    pub fn wait(self) -> Result<Channel<Blocking, Raw>, (Error, Channel<Blocking, Raw>)> {
+    fn poll_internal(&mut self) -> Option<Event> {
         let raw = self.channel.raw;
 
-        loop {
-            if raw.is_error() {
-                return Err((Error::ReceiverError, self.channel));
-            }
+        let status = raw.get_rx_status();
+        if status == Some(Event::End) {
+            // Do not clear the interrupt flags here: Subsequent calls of wait() must
+            // be able to observe them if this is currently called via poll()
+            raw.stop_rx();
+            raw.update();
 
-            if raw.is_rx_done() {
-                break;
+            let ptr = raw.channel_ram_start();
+            // SAFETY: RxChannel.receive() verifies that the length of self.data does not
+            // exceed the channel RAM size.
+            for (idx, entry) in self.data.iter_mut().enumerate() {
+                *entry = unsafe { ptr.add(idx).read_volatile() };
             }
         }
 
-        raw.stop_rx();
-        raw.clear_rx_interrupts();
-        raw.update();
+        status
+    }
 
-        let ptr = raw.channel_ram_start();
-        // SAFETY: RxChannel.receive() verifies that the length of self.data does not
-        // exceed the channel RAM size.
-        for (idx, entry) in self.data.iter_mut().enumerate() {
-            *entry = unsafe { ptr.add(idx).read_volatile() };
+    /// Check receive status
+    ///
+    /// Returns whether reception has ended (whether successfully or with an
+    /// error). In that case, a subsequent call to `wait()` returns immediately.
+    pub fn poll(&mut self) -> bool {
+        match self.poll_internal() {
+            Some(Event::Error | Event::End) => true,
+            Some(Event::Threshold) | None => false,
         }
+    }
 
-        Ok(self.channel)
+    /// Wait for the transaction to complete
+    pub fn wait(mut self) -> Result<Channel<Blocking, Raw>, (Error, Channel<Blocking, Raw>)> {
+        let raw = self.channel.raw;
+
+        let result = loop {
+            match self.poll_internal() {
+                Some(Event::Error) => break Err((Error::ReceiverError, self.channel)),
+                Some(Event::End) => break Ok(self.channel),
+                _ => continue,
+            }
+        };
+
+        raw.clear_rx_interrupts();
+
+        result
     }
 }
 

hil-test/tests/rmt.rs

@@ -79,7 +79,7 @@ fn generate_tx_data<const TX_LEN: usize>(write_end_marker: bool) -> [u32; TX_LEN
 
 // Run a test where some data is sent from one channel and looped back to
 // another one for receive, and verify that the data matches.
-fn do_rmt_loopback<const TX_LEN: usize>(tx_memsize: u8, rx_memsize: u8, wait_tx_first: bool) {
+fn do_rmt_loopback<const TX_LEN: usize>(tx_memsize: u8, rx_memsize: u8) {
     use esp_hal::rmt::{RxChannel, TxChannel};
 
     let peripherals = esp_hal::init(esp_hal::Config::default());
@@ -96,17 +96,20 @@ fn do_rmt_loopback<const TX_LEN: usize>(tx_memsize: u8, rx_memsize: u8, wait_tx_
     let tx_data: [_; TX_LEN] = generate_tx_data(true);
     let mut rcv_data: [u32; TX_LEN] = [PulseCode::empty(); TX_LEN];
 
-    let rx_transaction = rx_channel.receive(&mut rcv_data).unwrap();
-    let tx_transaction = tx_channel.transmit(&tx_data).unwrap();
+    let mut rx_transaction = rx_channel.receive(&mut rcv_data).unwrap();
+    let mut tx_transaction = tx_channel.transmit(&tx_data).unwrap();
 
-    if wait_tx_first {
-        tx_transaction.wait().unwrap();
-        rx_transaction.wait().unwrap();
-    } else {
-        rx_transaction.wait().unwrap();
-        tx_transaction.wait().unwrap();
+    loop {
+        let tx_done = tx_transaction.poll();
+        let rx_done = rx_transaction.poll();
+        if tx_done && rx_done {
+            break;
+        }
     }
 
+    tx_transaction.wait().unwrap();
+    rx_transaction.wait().unwrap();
+
     // the last two pulse-codes are the ones which wait for the timeout so
     // they can't be equal
     assert_eq!(&tx_data[..TX_LEN - 2], &rcv_data[..TX_LEN - 2]);
@@ -180,7 +183,7 @@ mod tests {
     #[test]
     fn rmt_loopback_simple() {
         // 20 codes fit a single RAM block
-        do_rmt_loopback::<20>(1, 1, false);
+        do_rmt_loopback::<20>(1, 1);
     }
 
     #[test]
@@ -191,7 +194,7 @@ mod tests {
     #[test]
     fn rmt_loopback_extended_ram() {
         // 80 codes require two RAM blocks
-        do_rmt_loopback::<80>(2, 2, false);
+        do_rmt_loopback::<80>(2, 2);
     }
 
     // FIXME: This test currently fails on esp32 with an rmt::Error::ReceiverError,
@@ -204,9 +207,8 @@ mod tests {
     #[test]
     fn rmt_loopback_tx_wrap() {
         // 80 codes require two RAM blocks; thus a tx channel with only 1 block requires
-        // wrapping. We need to .wait() on the tx transaction first to handle
-        // this.
-        do_rmt_loopback::<80>(1, 2, true);
+        // wrapping.
+        do_rmt_loopback::<80>(1, 2);
     }
 
     // FIXME: This test can't work right now, because wrapping rx is not
@@ -216,7 +218,7 @@ mod tests {
     // fn rmt_loopback_rx_wrap() {
     //     // 80 codes require two RAM blocks; thus an rx channel with only 1 block
     //     // requires wrapping
-    //     do_rmt_loopback<80>(2, 1, false);
+    //     do_rmt_loopback<80>(2, 1);
     // }
 
     #[test]