From 41b7e4a434c0d1fe1fc5d93afe76f8058aa411db Mon Sep 17 00:00:00 2001
From: Corey Schuhen <cschuhen@gmail.com>
Date: Sun, 24 Mar 2024 08:08:12 +1000
Subject: [PATCH] BXCAN: Create TxMode in order to support buffered TX.
---
embassy-stm32/src/can/bxcan.rs | 130 ++++++++++++++++++++++++++++++---
embassy-stm32/src/can/fdcan.rs | 5 +-
2 files changed, 120 insertions(+), 15 deletions(-)
diff --git a/embassy-stm32/src/can/bxcan.rs b/embassy-stm32/src/can/bxcan.rs
index 66f6e7067..45a3836c0 100644
--- a/embassy-stm32/src/can/bxcan.rs
+++ b/embassy-stm32/src/can/bxcan.rs
@@ -17,7 +17,6 @@ use crate::rcc::RccPeripheral;
use crate::{interrupt, peripherals, Peripheral};
pub mod enums;
-use enums::*;
pub mod frame;
pub mod util;
@@ -49,8 +48,7 @@ impl<T: Instance> interrupt::typelevel::Handler<T::TXInterrupt> for TxInterruptH
v.set_rqcp(1, true);
v.set_rqcp(2, true);
});
-
- T::state().tx_waker.wake();
+ T::state().tx_mode.on_interrupt::<T>();
}
}
@@ -258,7 +256,7 @@ impl<'d, T: Instance> CanTx<'d, T> {
/// If the TX queue is full, this will wait until there is space, therefore exerting backpressure.
pub async fn write(&mut self, frame: &Frame) -> crate::can::bx::TransmitStatus {
poll_fn(|cx| {
- T::state().tx_waker.register(cx.waker());
+ T::state().tx_mode.register(cx.waker());
if let Ok(status) = self.tx.transmit(frame) {
return Poll::Ready(status);
}
@@ -277,7 +275,7 @@ impl<'d, T: Instance> CanTx<'d, T> {
async fn flush_inner(mb: crate::can::bx::Mailbox) {
poll_fn(|cx| {
- T::state().tx_waker.register(cx.waker());
+ T::state().tx_mode.register(cx.waker());
if T::regs().tsr().read().tme(mb.index()) {
return Poll::Ready(());
}
@@ -294,7 +292,7 @@ impl<'d, T: Instance> CanTx<'d, T> {
async fn flush_any_inner() {
poll_fn(|cx| {
- T::state().tx_waker.register(cx.waker());
+ T::state().tx_mode.register(cx.waker());
let tsr = T::regs().tsr().read();
if tsr.tme(crate::can::bx::Mailbox::Mailbox0.index())
@@ -316,7 +314,7 @@ impl<'d, T: Instance> CanTx<'d, T> {
async fn flush_all_inner() {
poll_fn(|cx| {
- T::state().tx_waker.register(cx.waker());
+ T::state().tx_mode.register(cx.waker());
let tsr = T::regs().tsr().read();
if tsr.tme(crate::can::bx::Mailbox::Mailbox0.index())
@@ -335,6 +333,62 @@ impl<'d, T: Instance> CanTx<'d, T> {
pub async fn flush_all(&self) {
Self::flush_all_inner().await
}
+
+ /// Return a buffered instance of driver. User must supply Buffers
+ pub fn buffered<const TX_BUF_SIZE: usize>(
+ self,
+ txb: &'static mut TxBuf<TX_BUF_SIZE>,
+ ) -> BufferedCanTx<'d, T, TX_BUF_SIZE> {
+ BufferedCanTx::new(self.tx, txb)
+ }
+}
+
+/// User supplied buffer for TX buffering
+pub type TxBuf<const BUF_SIZE: usize> = Channel<CriticalSectionRawMutex, Frame, BUF_SIZE>;
+
+/// CAN driver, transmit half.
+pub struct BufferedCanTx<'d, T: Instance, const TX_BUF_SIZE: usize> {
+ _tx: crate::can::bx::Tx<BxcanInstance<'d, T>>,
+ tx_buf: &'static TxBuf<TX_BUF_SIZE>,
+}
+
+impl<'d, T: Instance, const TX_BUF_SIZE: usize> BufferedCanTx<'d, T, TX_BUF_SIZE> {
+ fn new(_tx: crate::can::bx::Tx<BxcanInstance<'d, T>>, tx_buf: &'static TxBuf<TX_BUF_SIZE>) -> Self {
+ Self { _tx, tx_buf }.setup()
+ }
+
+ fn setup(self) -> Self {
+ // We don't want interrupts being processed while we change modes.
+ critical_section::with(|_| unsafe {
+ let tx_inner = self::common::ClassicBufferedTxInner {
+ tx_receiver: self.tx_buf.receiver().into(),
+ };
+ T::mut_state().tx_mode = TxMode::Buffered(tx_inner);
+ });
+ self
+ }
+
+ /// Async write frame to TX buffer.
+ pub async fn write(&mut self, frame: &Frame) {
+ self.tx_buf.send(*frame).await;
+ T::TXInterrupt::pend(); // Wake for Tx
+ }
+
+ /// Returns a sender that can be used for sending CAN frames.
+ pub fn writer(&self) -> BufferedCanSender {
+ BufferedCanSender {
+ tx_buf: self.tx_buf.sender().into(),
+ waker: T::TXInterrupt::pend,
+ }
+ }
+}
+
+impl<'d, T: Instance, const TX_BUF_SIZE: usize> Drop for BufferedCanTx<'d, T, TX_BUF_SIZE> {
+ fn drop(&mut self) {
+ critical_section::with(|_| unsafe {
+ T::mut_state().tx_mode = TxMode::NonBuffered(embassy_sync::waitqueue::AtomicWaker::new());
+ });
+ }
}
/// CAN driver, receive half.
@@ -365,7 +419,7 @@ impl<'d, T: Instance> CanRx<'d, T> {
T::state().rx_mode.wait_not_empty::<T>().await
}
- /// Return a buffered instance of driver without CAN FD support. User must supply Buffers
+ /// Return a buffered instance of driver. User must supply Buffers
pub fn buffered<const RX_BUF_SIZE: usize>(
self,
rxb: &'static mut RxBuf<RX_BUF_SIZE>,
@@ -442,6 +496,14 @@ impl<'d, T: Instance, const RX_BUF_SIZE: usize> BufferedCanRx<'d, T, RX_BUF_SIZE
}
}
+impl<'d, T: Instance, const RX_BUF_SIZE: usize> Drop for BufferedCanRx<'d, T, RX_BUF_SIZE> {
+ fn drop(&mut self) {
+ critical_section::with(|_| unsafe {
+ T::mut_state().rx_mode = RxMode::NonBuffered(embassy_sync::waitqueue::AtomicWaker::new());
+ });
+ }
+}
+
use crate::can::bx::RxFifo;
impl<'d, T: Instance> Drop for Can<'d, T> {
@@ -568,18 +630,62 @@ impl RxMode {
}
}
}
+
+enum TxMode {
+ NonBuffered(AtomicWaker),
+ Buffered(self::common::ClassicBufferedTxInner),
+}
+
+impl TxMode {
+ pub fn buffer_free<T: Instance>(&self) -> bool {
+ let tsr = T::regs().tsr().read();
+ tsr.tme(crate::can::bx::Mailbox::Mailbox0.index())
+ || tsr.tme(crate::can::bx::Mailbox::Mailbox1.index())
+ || tsr.tme(crate::can::bx::Mailbox::Mailbox2.index())
+ }
+ pub fn on_interrupt<T: Instance>(&self) {
+ match &T::state().tx_mode {
+ TxMode::NonBuffered(waker) => waker.wake(),
+ TxMode::Buffered(buf) => {
+ while self.buffer_free::<T>() {
+ match buf.tx_receiver.try_receive() {
+ Ok(frame) => {
+ let mut registers = crate::can::bx::Registers { canregs: T::regs() };
+ _ = registers.transmit(&frame);
+ }
+ Err(_) => {
+ break;
+ }
+ }
+ }
+ }
+ }
+ }
+
+ fn register(&self, arg: &core::task::Waker) {
+ match self {
+ TxMode::NonBuffered(waker) => {
+ waker.register(arg);
+ }
+ _ => {
+ panic!("Bad mode");
+ }
+ }
+ }
+}
+
struct State {
- pub tx_waker: AtomicWaker,
- pub err_waker: AtomicWaker,
pub(crate) rx_mode: RxMode,
+ pub(crate) tx_mode: TxMode,
+ pub err_waker: AtomicWaker,
}
impl State {
pub const fn new() -> Self {
Self {
- tx_waker: AtomicWaker::new(),
- err_waker: AtomicWaker::new(),
rx_mode: RxMode::NonBuffered(AtomicWaker::new()),
+ tx_mode: TxMode::NonBuffered(AtomicWaker::new()),
+ err_waker: AtomicWaker::new(),
}
}
}
diff --git a/embassy-stm32/src/can/fdcan.rs b/embassy-stm32/src/can/fdcan.rs
index 42c9bd9f6..e58d8c0ec 100644
--- a/embassy-stm32/src/can/fdcan.rs
+++ b/embassy-stm32/src/can/fdcan.rs
@@ -507,7 +507,7 @@ pub struct BufferedCanFd<'d, T: Instance, const TX_BUF_SIZE: usize, const RX_BUF
/// Sender that can be used for sending CAN frames.
#[derive(Copy, Clone)]
pub struct BufferedFdCanSender {
- tx_buf: embassy_sync::channel::DynamicSender<'static, FdFrame>,
+ tx_buf: DynamicSender<'static, FdFrame>,
waker: fn(),
}
@@ -532,8 +532,7 @@ impl BufferedFdCanSender {
}
/// Receiver that can be used for receiving CAN frames. Note, each CAN frame will only be received by one receiver.
-pub type BufferedFdCanReceiver =
- embassy_sync::channel::DynamicReceiver<'static, Result<(FdFrame, Timestamp), BusError>>;
+pub type BufferedFdCanReceiver = DynamicReceiver<'static, Result<(FdFrame, Timestamp), BusError>>;
impl<'c, 'd, T: Instance, const TX_BUF_SIZE: usize, const RX_BUF_SIZE: usize>
BufferedCanFd<'d, T, TX_BUF_SIZE, RX_BUF_SIZE>