Merge pull request #1743 from xoviat/dma-2

stm32/dma: consolidate ringbuf

3 files changed, 120 insertions, 131 deletions

diff --git a/embassy-stm32/src/dma/bdma.rs b/embassy-stm32/src/dma/bdma.rs
index 60f4fbd09..20ff29bef 100644
--- a/embassy-stm32/src/dma/bdma.rs
+++ b/embassy-stm32/src/dma/bdma.rs
@@ -393,6 +393,10 @@ impl<'a, C: Channel> DmaCtrl for DmaCtrlImpl<'a, C> {
     fn reset_complete_count(&mut self) -> usize {
         STATE.complete_count[self.0.index()].swap(0, Ordering::AcqRel)
     }
+
+    fn set_waker(&mut self, waker: &Waker) {
+        STATE.ch_wakers[self.0.index()].register(waker);
+    }
 }
 
 pub struct ReadableRingBuffer<'a, C: Channel, W: Word> {
@@ -463,7 +467,7 @@ impl<'a, C: Channel, W: Word> ReadableRingBuffer<'a, C, W> {
     }
 
     pub fn clear(&mut self) {
-        self.ringbuf.clear(DmaCtrlImpl(self.channel.reborrow()));
+        self.ringbuf.clear(&mut DmaCtrlImpl(self.channel.reborrow()));
     }
 
     /// Read elements from the ring buffer
@@ -472,7 +476,7 @@ impl<'a, C: Channel, W: Word> ReadableRingBuffer<'a, C, W> {
     /// The length remaining is the capacity, ring_buf.len(), less the elements remaining after the read
     /// OverrunError is returned if the portion to be read was overwritten by the DMA controller.
     pub fn read(&mut self, buf: &mut [W]) -> Result<(usize, usize), OverrunError> {
-        self.ringbuf.read(DmaCtrlImpl(self.channel.reborrow()), buf)
+        self.ringbuf.read(&mut DmaCtrlImpl(self.channel.reborrow()), buf)
     }
 
     /// Read an exact number of elements from the ringbuffer.
@@ -487,39 +491,18 @@ impl<'a, C: Channel, W: Word> ReadableRingBuffer<'a, C, W> {
     /// - If M equals N/2 or N/2 divides evenly into M, this function will return every N/2 elements read on the DMA source.
     /// - Otherwise, this function may need up to N/2 extra elements to arrive before returning.
     pub async fn read_exact(&mut self, buffer: &mut [W]) -> Result<usize, OverrunError> {
-        use core::future::poll_fn;
-        use core::sync::atomic::compiler_fence;
-
-        let mut read_data = 0;
-        let buffer_len = buffer.len();
-
-        poll_fn(|cx| {
-            self.set_waker(cx.waker());
-
-            compiler_fence(Ordering::SeqCst);
-
-            match self.read(&mut buffer[read_data..buffer_len]) {
-                Ok((len, remaining)) => {
-                    read_data += len;
-                    if read_data == buffer_len {
-                        Poll::Ready(Ok(remaining))
-                    } else {
-                        Poll::Pending
-                    }
-                }
-                Err(e) => Poll::Ready(Err(e)),
-            }
-        })
-        .await
+        self.ringbuf
+            .read_exact(&mut DmaCtrlImpl(self.channel.reborrow()), buffer)
+            .await
     }
 
     /// The capacity of the ringbuffer.
-    pub fn cap(&self) -> usize {
+    pub const fn cap(&self) -> usize {
         self.ringbuf.cap()
     }
 
     pub fn set_waker(&mut self, waker: &Waker) {
-        STATE.ch_wakers[self.channel.index()].register(waker);
+        DmaCtrlImpl(self.channel.reborrow()).set_waker(waker);
     }
 
     fn clear_irqs(&mut self) {
@@ -628,50 +611,29 @@ impl<'a, C: Channel, W: Word> WritableRingBuffer<'a, C, W> {
     }
 
     pub fn clear(&mut self) {
-        self.ringbuf.clear(DmaCtrlImpl(self.channel.reborrow()));
+        self.ringbuf.clear(&mut DmaCtrlImpl(self.channel.reborrow()));
     }
 
     /// Write elements to the ring buffer
     /// Return a tuple of the length written and the length remaining in the buffer
     pub fn write(&mut self, buf: &[W]) -> Result<(usize, usize), OverrunError> {
-        self.ringbuf.write(DmaCtrlImpl(self.channel.reborrow()), buf)
+        self.ringbuf.write(&mut DmaCtrlImpl(self.channel.reborrow()), buf)
     }
 
     /// Write an exact number of elements to the ringbuffer.
     pub async fn write_exact(&mut self, buffer: &[W]) -> Result<usize, OverrunError> {
-        use core::future::poll_fn;
-        use core::sync::atomic::compiler_fence;
-
-        let mut written_data = 0;
-        let buffer_len = buffer.len();
-
-        poll_fn(|cx| {
-            self.set_waker(cx.waker());
-
-            compiler_fence(Ordering::SeqCst);
-
-            match self.write(&buffer[written_data..buffer_len]) {
-                Ok((len, remaining)) => {
-                    written_data += len;
-                    if written_data == buffer_len {
-                        Poll::Ready(Ok(remaining))
-                    } else {
-                        Poll::Pending
-                    }
-                }
-                Err(e) => Poll::Ready(Err(e)),
-            }
-        })
-        .await
+        self.ringbuf
+            .write_exact(&mut DmaCtrlImpl(self.channel.reborrow()), buffer)
+            .await
     }
 
     /// The capacity of the ringbuffer.
-    pub fn cap(&self) -> usize {
+    pub const fn cap(&self) -> usize {
         self.ringbuf.cap()
     }
 
     pub fn set_waker(&mut self, waker: &Waker) {
-        STATE.ch_wakers[self.channel.index()].register(waker);
+        DmaCtrlImpl(self.channel.reborrow()).set_waker(waker);
     }
 
     fn clear_irqs(&mut self) {
diff --git a/embassy-stm32/src/dma/dma.rs b/embassy-stm32/src/dma/dma.rs
index 9cd7aa8d5..5033ae477 100644
--- a/embassy-stm32/src/dma/dma.rs
+++ b/embassy-stm32/src/dma/dma.rs
@@ -623,6 +623,10 @@ impl<'a, C: Channel> DmaCtrl for DmaCtrlImpl<'a, C> {
     fn reset_complete_count(&mut self) -> usize {
         STATE.complete_count[self.0.index()].swap(0, Ordering::AcqRel)
     }
+
+    fn set_waker(&mut self, waker: &Waker) {
+        STATE.ch_wakers[self.0.index()].register(waker);
+    }
 }
 
 pub struct ReadableRingBuffer<'a, C: Channel, W: Word> {
@@ -708,7 +712,7 @@ impl<'a, C: Channel, W: Word> ReadableRingBuffer<'a, C, W> {
     }
 
     pub fn clear(&mut self) {
-        self.ringbuf.clear(DmaCtrlImpl(self.channel.reborrow()));
+        self.ringbuf.clear(&mut DmaCtrlImpl(self.channel.reborrow()));
     }
 
     /// Read elements from the ring buffer
@@ -717,7 +721,7 @@ impl<'a, C: Channel, W: Word> ReadableRingBuffer<'a, C, W> {
     /// The length remaining is the capacity, ring_buf.len(), less the elements remaining after the read
     /// OverrunError is returned if the portion to be read was overwritten by the DMA controller.
     pub fn read(&mut self, buf: &mut [W]) -> Result<(usize, usize), OverrunError> {
-        self.ringbuf.read(DmaCtrlImpl(self.channel.reborrow()), buf)
+        self.ringbuf.read(&mut DmaCtrlImpl(self.channel.reborrow()), buf)
     }
 
     /// Read an exact number of elements from the ringbuffer.
@@ -732,39 +736,18 @@ impl<'a, C: Channel, W: Word> ReadableRingBuffer<'a, C, W> {
     /// - If M equals N/2 or N/2 divides evenly into M, this function will return every N/2 elements read on the DMA source.
     /// - Otherwise, this function may need up to N/2 extra elements to arrive before returning.
     pub async fn read_exact(&mut self, buffer: &mut [W]) -> Result<usize, OverrunError> {
-        use core::future::poll_fn;
-        use core::sync::atomic::compiler_fence;
-
-        let mut read_data = 0;
-        let buffer_len = buffer.len();
-
-        poll_fn(|cx| {
-            self.set_waker(cx.waker());
-
-            compiler_fence(Ordering::SeqCst);
-
-            match self.read(&mut buffer[read_data..buffer_len]) {
-                Ok((len, remaining)) => {
-                    read_data += len;
-                    if read_data == buffer_len {
-                        Poll::Ready(Ok(remaining))
-                    } else {
-                        Poll::Pending
-                    }
-                }
-                Err(e) => Poll::Ready(Err(e)),
-            }
-        })
-        .await
+        self.ringbuf
+            .read_exact(&mut DmaCtrlImpl(self.channel.reborrow()), buffer)
+            .await
     }
 
     // The capacity of the ringbuffer
-    pub fn cap(&self) -> usize {
+    pub const fn cap(&self) -> usize {
         self.ringbuf.cap()
     }
 
     pub fn set_waker(&mut self, waker: &Waker) {
-        STATE.ch_wakers[self.channel.index()].register(waker);
+        DmaCtrlImpl(self.channel.reborrow()).set_waker(waker);
     }
 
     fn clear_irqs(&mut self) {
@@ -890,50 +873,29 @@ impl<'a, C: Channel, W: Word> WritableRingBuffer<'a, C, W> {
     }
 
     pub fn clear(&mut self) {
-        self.ringbuf.clear(DmaCtrlImpl(self.channel.reborrow()));
+        self.ringbuf.clear(&mut DmaCtrlImpl(self.channel.reborrow()));
     }
 
     /// Write elements from the ring buffer
     /// Return a tuple of the length written and the length remaining in the buffer
     pub fn write(&mut self, buf: &[W]) -> Result<(usize, usize), OverrunError> {
-        self.ringbuf.write(DmaCtrlImpl(self.channel.reborrow()), buf)
+        self.ringbuf.write(&mut DmaCtrlImpl(self.channel.reborrow()), buf)
     }
 
     /// Write an exact number of elements to the ringbuffer.
     pub async fn write_exact(&mut self, buffer: &[W]) -> Result<usize, OverrunError> {
-        use core::future::poll_fn;
-        use core::sync::atomic::compiler_fence;
-
-        let mut written_data = 0;
-        let buffer_len = buffer.len();
-
-        poll_fn(|cx| {
-            self.set_waker(cx.waker());
-
-            compiler_fence(Ordering::SeqCst);
-
-            match self.write(&buffer[written_data..buffer_len]) {
-                Ok((len, remaining)) => {
-                    written_data += len;
-                    if written_data == buffer_len {
-                        Poll::Ready(Ok(remaining))
-                    } else {
-                        Poll::Pending
-                    }
-                }
-                Err(e) => Poll::Ready(Err(e)),
-            }
-        })
-        .await
+        self.ringbuf
+            .write_exact(&mut DmaCtrlImpl(self.channel.reborrow()), buffer)
+            .await
     }
 
     // The capacity of the ringbuffer
-    pub fn cap(&self) -> usize {
+    pub const fn cap(&self) -> usize {
         self.ringbuf.cap()
     }
 
     pub fn set_waker(&mut self, waker: &Waker) {
-        STATE.ch_wakers[self.channel.index()].register(waker);
+        DmaCtrlImpl(self.channel.reborrow()).set_waker(waker);
     }
 
     fn clear_irqs(&mut self) {
diff --git a/embassy-stm32/src/dma/ringbuffer.rs b/embassy-stm32/src/dma/ringbuffer.rs
index 6ce59e455..c9f7a3026 100644
--- a/embassy-stm32/src/dma/ringbuffer.rs
+++ b/embassy-stm32/src/dma/ringbuffer.rs
@@ -1,7 +1,9 @@
 #![cfg_attr(gpdma, allow(unused))]
 
+use core::future::poll_fn;
 use core::ops::Range;
 use core::sync::atomic::{compiler_fence, Ordering};
+use core::task::{Poll, Waker};
 
 use super::word::Word;
 
@@ -49,6 +51,9 @@ pub trait DmaCtrl {
 
     /// Reset the transfer completed counter to 0 and return the value just prior to the reset.
     fn reset_complete_count(&mut self) -> usize;
+
+    /// Set the waker for a running poll_fn
+    fn set_waker(&mut self, waker: &Waker);
 }
 
 impl<'a, W: Word> ReadableDmaRingBuffer<'a, W> {
@@ -57,7 +62,7 @@ impl<'a, W: Word> ReadableDmaRingBuffer<'a, W> {
     }
 
     /// Reset the ring buffer to its initial state
-    pub fn clear(&mut self, mut dma: impl DmaCtrl) {
+    pub fn clear(&mut self, dma: &mut impl DmaCtrl) {
         self.start = 0;
         dma.reset_complete_count();
     }
@@ -68,8 +73,43 @@ impl<'a, W: Word> ReadableDmaRingBuffer<'a, W> {
     }
 
     /// The current position of the ringbuffer
-    fn pos(&self, remaining_transfers: usize) -> usize {
-        self.cap() - remaining_transfers
+    fn pos(&self, dma: &mut impl DmaCtrl) -> usize {
+        self.cap() - dma.get_remaining_transfers()
+    }
+
+    /// Read an exact number of elements from the ringbuffer.
+    ///
+    /// Returns the remaining number of elements available for immediate reading.
+    /// OverrunError is returned if the portion to be read was overwritten by the DMA controller.
+    ///
+    /// Async/Wake Behavior:
+    /// The underlying DMA peripheral only can wake us when its buffer pointer has reached the halfway point,
+    /// and when it wraps around. This means that when called with a buffer of length 'M', when this
+    /// ring buffer was created with a buffer of size 'N':
+    /// - If M equals N/2 or N/2 divides evenly into M, this function will return every N/2 elements read on the DMA source.
+    /// - Otherwise, this function may need up to N/2 extra elements to arrive before returning.
+    pub async fn read_exact(&mut self, dma: &mut impl DmaCtrl, buffer: &mut [W]) -> Result<usize, OverrunError> {
+        let mut read_data = 0;
+        let buffer_len = buffer.len();
+
+        poll_fn(|cx| {
+            dma.set_waker(cx.waker());
+
+            compiler_fence(Ordering::SeqCst);
+
+            match self.read(dma, &mut buffer[read_data..buffer_len]) {
+                Ok((len, remaining)) => {
+                    read_data += len;
+                    if read_data == buffer_len {
+                        Poll::Ready(Ok(remaining))
+                    } else {
+                        Poll::Pending
+                    }
+                }
+                Err(e) => Poll::Ready(Err(e)),
+            }
+        })
+        .await
     }
 
     /// Read elements from the ring buffer
@@ -77,7 +117,7 @@ impl<'a, W: Word> ReadableDmaRingBuffer<'a, W> {
     /// If not all of the elements were read, then there will be some elements in the buffer remaining
     /// The length remaining is the capacity, ring_buf.len(), less the elements remaining after the read
     /// OverrunError is returned if the portion to be read was overwritten by the DMA controller.
-    pub fn read(&mut self, mut dma: impl DmaCtrl, buf: &mut [W]) -> Result<(usize, usize), OverrunError> {
+    pub fn read(&mut self, dma: &mut impl DmaCtrl, buf: &mut [W]) -> Result<(usize, usize), OverrunError> {
         /*
             This algorithm is optimistic: we assume we haven't overrun more than a full buffer and then check
             after we've done our work to see we have. This is because on stm32, an interrupt is not guaranteed
@@ -93,7 +133,7 @@ impl<'a, W: Word> ReadableDmaRingBuffer<'a, W> {
             rather than the data we actually copied because it costs nothing and confirms an error condition
             earlier.
         */
-        let end = self.pos(dma.get_remaining_transfers());
+        let end = self.pos(dma);
         if self.start == end && dma.get_complete_count() == 0 {
             // No elements are available in the buffer
             Ok((0, self.cap()))
@@ -114,8 +154,7 @@ impl<'a, W: Word> ReadableDmaRingBuffer<'a, W> {
                 then, get the current position of of the dma write and check
                 if it's inside data we could have copied
             */
-            let (pos, complete_count) =
-                critical_section::with(|_| (self.pos(dma.get_remaining_transfers()), dma.get_complete_count()));
+            let (pos, complete_count) = critical_section::with(|_| (self.pos(dma), dma.get_complete_count()));
             if (pos >= self.start && pos < end) || (complete_count > 0 && pos >= end) || complete_count > 1 {
                 Err(OverrunError)
             } else {
@@ -141,7 +180,7 @@ impl<'a, W: Word> ReadableDmaRingBuffer<'a, W> {
                 then, get the current position of of the dma write and check
                 if it's inside data we could have copied
             */
-            let pos = self.pos(dma.get_remaining_transfers());
+            let pos = self.pos(dma);
             if pos > self.start || pos < end || dma.get_complete_count() > 1 {
                 Err(OverrunError)
             } else {
@@ -169,7 +208,7 @@ impl<'a, W: Word> ReadableDmaRingBuffer<'a, W> {
                 then, get the current position of of the dma write and check
                 if it's inside data we could have copied
             */
-            let pos = self.pos(dma.get_remaining_transfers());
+            let pos = self.pos(dma);
             if pos > self.start || pos < end || dma.reset_complete_count() > 1 {
                 Err(OverrunError)
             } else {
@@ -209,7 +248,7 @@ impl<'a, W: Word> WritableDmaRingBuffer<'a, W> {
     }
 
     /// Reset the ring buffer to its initial state
-    pub fn clear(&mut self, mut dma: impl DmaCtrl) {
+    pub fn clear(&mut self, dma: &mut impl DmaCtrl) {
         self.end = 0;
         dma.reset_complete_count();
     }
@@ -220,14 +259,39 @@ impl<'a, W: Word> WritableDmaRingBuffer<'a, W> {
     }
 
     /// The current position of the ringbuffer
-    fn pos(&self, remaining_transfers: usize) -> usize {
-        self.cap() - remaining_transfers
+    fn pos(&self, dma: &mut impl DmaCtrl) -> usize {
+        self.cap() - dma.get_remaining_transfers()
+    }
+
+    /// Write an exact number of elements to the ringbuffer.
+    pub async fn write_exact(&mut self, dma: &mut impl DmaCtrl, buffer: &[W]) -> Result<usize, OverrunError> {
+        let mut written_data = 0;
+        let buffer_len = buffer.len();
+
+        poll_fn(|cx| {
+            dma.set_waker(cx.waker());
+
+            compiler_fence(Ordering::SeqCst);
+
+            match self.write(dma, &buffer[written_data..buffer_len]) {
+                Ok((len, remaining)) => {
+                    written_data += len;
+                    if written_data == buffer_len {
+                        Poll::Ready(Ok(remaining))
+                    } else {
+                        Poll::Pending
+                    }
+                }
+                Err(e) => Poll::Ready(Err(e)),
+            }
+        })
+        .await
     }
 
     /// Write elements from the ring buffer
     /// Return a tuple of the length written and the capacity remaining to be written in the buffer
-    pub fn write(&mut self, mut dma: impl DmaCtrl, buf: &[W]) -> Result<(usize, usize), OverrunError> {
-        let start = self.pos(dma.get_remaining_transfers());
+    pub fn write(&mut self, dma: &mut impl DmaCtrl, buf: &[W]) -> Result<(usize, usize), OverrunError> {
+        let start = self.pos(dma);
         if start > self.end {
             // The occupied portion in the ring buffer DOES wrap
             let len = self.copy_from(buf, self.end..start);
@@ -235,8 +299,7 @@ impl<'a, W: Word> WritableDmaRingBuffer<'a, W> {
             compiler_fence(Ordering::SeqCst);
 
             // Confirm that the DMA is not inside data we could have written
-            let (pos, complete_count) =
-                critical_section::with(|_| (self.pos(dma.get_remaining_transfers()), dma.get_complete_count()));
+            let (pos, complete_count) = critical_section::with(|_| (self.pos(dma), dma.get_complete_count()));
             if (pos >= self.end && pos < start) || (complete_count > 0 && pos >= start) || complete_count > 1 {
                 Err(OverrunError)
             } else {
@@ -256,7 +319,7 @@ impl<'a, W: Word> WritableDmaRingBuffer<'a, W> {
             compiler_fence(Ordering::SeqCst);
 
             // Confirm that the DMA is not inside data we could have written
-            let pos = self.pos(dma.get_remaining_transfers());
+            let pos = self.pos(dma);
             if pos > self.end || pos < start || dma.get_complete_count() > 1 {
                 Err(OverrunError)
             } else {
@@ -274,7 +337,7 @@ impl<'a, W: Word> WritableDmaRingBuffer<'a, W> {
             compiler_fence(Ordering::SeqCst);
 
             // Confirm that the DMA is not inside data we could have written
-            let pos = self.pos(dma.get_remaining_transfers());
+            let pos = self.pos(dma);
             if pos > self.end || pos < start || dma.reset_complete_count() > 1 {
                 Err(OverrunError)
             } else {
@@ -323,7 +386,7 @@ mod tests {
         requests: cell::RefCell<vec::Vec<TestCircularTransferRequest>>,
     }
 
-    impl DmaCtrl for &mut TestCircularTransfer {
+    impl DmaCtrl for TestCircularTransfer {
         fn get_remaining_transfers(&self) -> usize {
             match self.requests.borrow_mut().pop().unwrap() {
                 TestCircularTransferRequest::PositionRequest(pos) => {
@@ -350,6 +413,8 @@ mod tests {
                 _ => unreachable!(),
             }
         }
+
+        fn set_waker(&mut self, waker: &Waker) {}
     }
 
     impl TestCircularTransfer {