From 607e67f51a33d83c59db5e091348e4bda0b5ddbb Mon Sep 17 00:00:00 2001 From: Dario Nieuwenhuis Date: Tue, 5 Jan 2021 21:14:04 +0100 Subject: [PATCH] Cleanup BufferedUarte --- embassy-nrf/src/buffered_uarte.rs | 157 +++++++++++++----------------- 1 file changed, 67 insertions(+), 90 deletions(-) diff --git a/embassy-nrf/src/buffered_uarte.rs b/embassy-nrf/src/buffered_uarte.rs index dbdb6f51..2e29da25 100644 --- a/embassy-nrf/src/buffered_uarte.rs +++ b/embassy-nrf/src/buffered_uarte.rs @@ -33,12 +33,25 @@ enum RxState { ReceivingReady, Stopping, } + #[derive(Copy, Clone, Debug, PartialEq)] enum TxState { Idle, Transmitting(usize), } +struct State<'a, T: Instance> { + inner: T, + + rx: RingBuffer<'a>, + rx_state: RxState, + rx_waker: WakerRegistration, + + tx: RingBuffer<'a>, + tx_state: TxState, + tx_waker: WakerRegistration, +} + /// Interface to a UARTE instance /// /// This is a very basic interface that comes with the following limitations: @@ -145,6 +158,10 @@ impl<'a, T: Instance> BufferedUarte<'a, T> { ), } } + + fn inner(self: Pin<&mut Self>) -> Pin<&mut PeripheralMutex>> { + unsafe { Pin::new_unchecked(&mut self.get_unchecked_mut().inner) } + } } impl<'a, T: Instance> Drop for BufferedUarte<'a, T> { @@ -156,111 +173,71 @@ impl<'a, T: Instance> Drop for BufferedUarte<'a, T> { impl<'a, T: Instance> AsyncBufRead for BufferedUarte<'a, T> { fn poll_fill_buf(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll> { - let this = unsafe { self.get_unchecked_mut() }; - let reg = unsafe { Pin::new_unchecked(&mut this.inner) }; - reg.with(|_irq, state| { - let z: Poll> = state.poll_fill_buf(cx); - let z: Poll> = unsafe { mem::transmute(z) }; - z + self.inner().with(|_irq, state| { + // Conservative compiler fence to prevent optimizations that do not + // take in to account actions by DMA. The fence has been placed here, + // before any DMA action has started + compiler_fence(Ordering::SeqCst); + trace!("poll_read"); + + // We have data ready in buffer? Return it. + let buf = state.rx.pop_buf(); + if buf.len() != 0 { + trace!(" got {:?} {:?}", buf.as_ptr() as u32, buf.len()); + let buf: &[u8] = buf; + let buf: &[u8] = unsafe { mem::transmute(buf) }; + return Poll::Ready(Ok(buf)); + } + + trace!(" empty"); + + if state.rx_state == RxState::ReceivingReady { + trace!(" stopping"); + state.rx_state = RxState::Stopping; + state.inner.tasks_stoprx.write(|w| unsafe { w.bits(1) }); + } + + state.rx_waker.register(cx.waker()); + Poll::>::Pending }) } fn consume(self: Pin<&mut Self>, amt: usize) { - let this = unsafe { self.get_unchecked_mut() }; - let reg = unsafe { Pin::new_unchecked(&mut this.inner) }; - reg.with(|irq, state| state.consume(irq, amt)) + self.inner().with(|irq, state| { + trace!("consume {:?}", amt); + state.rx.pop(amt); + irq.pend(); + }) } } impl<'a, T: Instance> AsyncWrite for BufferedUarte<'a, T> { fn poll_write(self: Pin<&mut Self>, cx: &mut Context<'_>, buf: &[u8]) -> Poll> { - let this = unsafe { self.get_unchecked_mut() }; - let reg = unsafe { Pin::new_unchecked(&mut this.inner) }; - reg.with(|irq, state| state.poll_write(irq, cx, buf)) - } -} + self.inner().with(|irq, state| { + trace!("poll_write: {:?}", buf.len()); -// ==================================== -// ==================================== -// ==================================== + let tx_buf = state.tx.push_buf(); + if tx_buf.len() == 0 { + trace!("poll_write: pending"); + state.tx_waker.register(cx.waker()); + return Poll::Pending; + } -// public because it needs to be used in Instance trait, but -// should not be used outside the module -#[doc(hidden)] -pub struct State<'a, T: Instance> { - inner: T, + let n = min(tx_buf.len(), buf.len()); + tx_buf[..n].copy_from_slice(&buf[..n]); + state.tx.push(n); - rx: RingBuffer<'a>, - rx_state: RxState, - rx_waker: WakerRegistration, + trace!("poll_write: queued {:?}", n); - tx: RingBuffer<'a>, - tx_state: TxState, - tx_waker: WakerRegistration, -} + // Conservative compiler fence to prevent optimizations that do not + // take in to account actions by DMA. The fence has been placed here, + // before any DMA action has started + compiler_fence(Ordering::SeqCst); -impl<'a, T: Instance> State<'a, T> { - fn poll_fill_buf(&mut self, cx: &mut Context<'_>) -> Poll> { - // Conservative compiler fence to prevent optimizations that do not - // take in to account actions by DMA. The fence has been placed here, - // before any DMA action has started - compiler_fence(Ordering::SeqCst); - trace!("poll_read"); + irq.pend(); - // We have data ready in buffer? Return it. - let buf = self.rx.pop_buf(); - if buf.len() != 0 { - trace!(" got {:?} {:?}", buf.as_ptr() as u32, buf.len()); - return Poll::Ready(Ok(buf)); - } - - trace!(" empty"); - - if self.rx_state == RxState::ReceivingReady { - trace!(" stopping"); - self.rx_state = RxState::Stopping; - self.inner.tasks_stoprx.write(|w| unsafe { w.bits(1) }); - } - - self.rx_waker.register(cx.waker()); - Poll::Pending - } - - fn consume(&mut self, irq: &mut T::Interrupt, amt: usize) { - trace!("consume {:?}", amt); - self.rx.pop(amt); - irq.pend(); - } - - fn poll_write( - &mut self, - irq: &mut T::Interrupt, - cx: &mut Context<'_>, - buf: &[u8], - ) -> Poll> { - trace!("poll_write: {:?}", buf.len()); - - let tx_buf = self.tx.push_buf(); - if tx_buf.len() == 0 { - trace!("poll_write: pending"); - self.tx_waker.register(cx.waker()); - return Poll::Pending; - } - - let n = min(tx_buf.len(), buf.len()); - tx_buf[..n].copy_from_slice(&buf[..n]); - self.tx.push(n); - - trace!("poll_write: queued {:?}", n); - - // Conservative compiler fence to prevent optimizations that do not - // take in to account actions by DMA. The fence has been placed here, - // before any DMA action has started - compiler_fence(Ordering::SeqCst); - - irq.pend(); - - Poll::Ready(Ok(n)) + Poll::Ready(Ok(n)) + }) } }