diff --git a/embassy-net-driver-channel/Cargo.toml b/embassy-net-driver-channel/Cargo.toml new file mode 100644 index 00000000..700a4e8a --- /dev/null +++ b/embassy-net-driver-channel/Cargo.toml @@ -0,0 +1,12 @@ +[package] +name = "embassy-net-driver-channel" +version = "0.1.0" +edition = "2021" + +[dependencies] +defmt = { version = "0.3", optional = true } +log = { version = "0.4.14", optional = true } + +embassy-sync = { version = "0.1.0", path = "../embassy-sync" } +embassy-futures = { version = "0.1.0", path = "../embassy-futures" } +embassy-net-driver = { version = "0.1.0", path = "../embassy-net-driver" } diff --git a/embassy-net-driver-channel/src/fmt.rs b/embassy-net-driver-channel/src/fmt.rs new file mode 100644 index 00000000..06697081 --- /dev/null +++ b/embassy-net-driver-channel/src/fmt.rs @@ -0,0 +1,225 @@ +#![macro_use] +#![allow(unused_macros)] + +#[cfg(all(feature = "defmt", feature = "log"))] +compile_error!("You may not enable both `defmt` and `log` features."); + +macro_rules! assert { + ($($x:tt)*) => { + { + #[cfg(not(feature = "defmt"))] + ::core::assert!($($x)*); + #[cfg(feature = "defmt")] + ::defmt::assert!($($x)*); + } + }; +} + +macro_rules! assert_eq { + ($($x:tt)*) => { + { + #[cfg(not(feature = "defmt"))] + ::core::assert_eq!($($x)*); + #[cfg(feature = "defmt")] + ::defmt::assert_eq!($($x)*); + } + }; +} + +macro_rules! assert_ne { + ($($x:tt)*) => { + { + #[cfg(not(feature = "defmt"))] + ::core::assert_ne!($($x)*); + #[cfg(feature = "defmt")] + ::defmt::assert_ne!($($x)*); + } + }; +} + +macro_rules! debug_assert { + ($($x:tt)*) => { + { + #[cfg(not(feature = "defmt"))] + ::core::debug_assert!($($x)*); + #[cfg(feature = "defmt")] + ::defmt::debug_assert!($($x)*); + } + }; +} + +macro_rules! debug_assert_eq { + ($($x:tt)*) => { + { + #[cfg(not(feature = "defmt"))] + ::core::debug_assert_eq!($($x)*); + #[cfg(feature = "defmt")] + ::defmt::debug_assert_eq!($($x)*); + } + }; +} + +macro_rules! debug_assert_ne { + ($($x:tt)*) => { + { + #[cfg(not(feature = "defmt"))] + ::core::debug_assert_ne!($($x)*); + #[cfg(feature = "defmt")] + ::defmt::debug_assert_ne!($($x)*); + } + }; +} + +macro_rules! todo { + ($($x:tt)*) => { + { + #[cfg(not(feature = "defmt"))] + ::core::todo!($($x)*); + #[cfg(feature = "defmt")] + ::defmt::todo!($($x)*); + } + }; +} + +macro_rules! unreachable { + ($($x:tt)*) => { + { + #[cfg(not(feature = "defmt"))] + ::core::unreachable!($($x)*); + #[cfg(feature = "defmt")] + ::defmt::unreachable!($($x)*); + } + }; +} + +macro_rules! panic { + ($($x:tt)*) => { + { + #[cfg(not(feature = "defmt"))] + ::core::panic!($($x)*); + #[cfg(feature = "defmt")] + ::defmt::panic!($($x)*); + } + }; +} + +macro_rules! trace { + ($s:literal $(, $x:expr)* $(,)?) => { + { + #[cfg(feature = "log")] + ::log::trace!($s $(, $x)*); + #[cfg(feature = "defmt")] + ::defmt::trace!($s $(, $x)*); + #[cfg(not(any(feature = "log", feature="defmt")))] + let _ = ($( & $x ),*); + } + }; +} + +macro_rules! debug { + ($s:literal $(, $x:expr)* $(,)?) => { + { + #[cfg(feature = "log")] + ::log::debug!($s $(, $x)*); + #[cfg(feature = "defmt")] + ::defmt::debug!($s $(, $x)*); + #[cfg(not(any(feature = "log", feature="defmt")))] + let _ = ($( & $x ),*); + } + }; +} + +macro_rules! info { + ($s:literal $(, $x:expr)* $(,)?) => { + { + #[cfg(feature = "log")] + ::log::info!($s $(, $x)*); + #[cfg(feature = "defmt")] + ::defmt::info!($s $(, $x)*); + #[cfg(not(any(feature = "log", feature="defmt")))] + let _ = ($( & $x ),*); + } + }; +} + +macro_rules! warn { + ($s:literal $(, $x:expr)* $(,)?) => { + { + #[cfg(feature = "log")] + ::log::warn!($s $(, $x)*); + #[cfg(feature = "defmt")] + ::defmt::warn!($s $(, $x)*); + #[cfg(not(any(feature = "log", feature="defmt")))] + let _ = ($( & $x ),*); + } + }; +} + +macro_rules! error { + ($s:literal $(, $x:expr)* $(,)?) => { + { + #[cfg(feature = "log")] + ::log::error!($s $(, $x)*); + #[cfg(feature = "defmt")] + ::defmt::error!($s $(, $x)*); + #[cfg(not(any(feature = "log", feature="defmt")))] + let _ = ($( & $x ),*); + } + }; +} + +#[cfg(feature = "defmt")] +macro_rules! unwrap { + ($($x:tt)*) => { + ::defmt::unwrap!($($x)*) + }; +} + +#[cfg(not(feature = "defmt"))] +macro_rules! unwrap { + ($arg:expr) => { + match $crate::fmt::Try::into_result($arg) { + ::core::result::Result::Ok(t) => t, + ::core::result::Result::Err(e) => { + ::core::panic!("unwrap of `{}` failed: {:?}", ::core::stringify!($arg), e); + } + } + }; + ($arg:expr, $($msg:expr),+ $(,)? ) => { + match $crate::fmt::Try::into_result($arg) { + ::core::result::Result::Ok(t) => t, + ::core::result::Result::Err(e) => { + ::core::panic!("unwrap of `{}` failed: {}: {:?}", ::core::stringify!($arg), ::core::format_args!($($msg,)*), e); + } + } + } +} + +#[derive(Debug, Copy, Clone, Eq, PartialEq)] +pub struct NoneError; + +pub trait Try { + type Ok; + type Error; + fn into_result(self) -> Result; +} + +impl Try for Option { + type Ok = T; + type Error = NoneError; + + #[inline] + fn into_result(self) -> Result { + self.ok_or(NoneError) + } +} + +impl Try for Result { + type Ok = T; + type Error = E; + + #[inline] + fn into_result(self) -> Self { + self + } +} diff --git a/embassy-net-driver-channel/src/lib.rs b/embassy-net-driver-channel/src/lib.rs new file mode 100644 index 00000000..369dc5a9 --- /dev/null +++ b/embassy-net-driver-channel/src/lib.rs @@ -0,0 +1,525 @@ +#![no_std] + +// must go first! +mod fmt; + +use core::cell::RefCell; +use core::mem::MaybeUninit; +use core::task::{Context, Poll}; + +pub use embassy_net_driver as driver; +use embassy_net_driver::{Capabilities, LinkState, Medium}; +use embassy_sync::blocking_mutex::raw::NoopRawMutex; +use embassy_sync::blocking_mutex::Mutex; +use embassy_sync::waitqueue::WakerRegistration; + +pub struct State { + rx: [PacketBuf; N_RX], + tx: [PacketBuf; N_TX], + inner: MaybeUninit>, +} + +impl State { + const NEW_PACKET: PacketBuf = PacketBuf::new(); + + pub const fn new() -> Self { + Self { + rx: [Self::NEW_PACKET; N_RX], + tx: [Self::NEW_PACKET; N_TX], + inner: MaybeUninit::uninit(), + } + } +} + +struct StateInner<'d, const MTU: usize> { + rx: zerocopy_channel::Channel<'d, NoopRawMutex, PacketBuf>, + tx: zerocopy_channel::Channel<'d, NoopRawMutex, PacketBuf>, + link_state: Mutex>, +} + +/// State of the LinkState +struct LinkStateState { + state: LinkState, + waker: WakerRegistration, +} + +pub struct Runner<'d, const MTU: usize> { + tx_chan: zerocopy_channel::Receiver<'d, NoopRawMutex, PacketBuf>, + rx_chan: zerocopy_channel::Sender<'d, NoopRawMutex, PacketBuf>, + link_state: &'d Mutex>, +} + +pub struct RxRunner<'d, const MTU: usize> { + rx_chan: zerocopy_channel::Sender<'d, NoopRawMutex, PacketBuf>, + link_state: &'d Mutex>, +} + +pub struct TxRunner<'d, const MTU: usize> { + tx_chan: zerocopy_channel::Receiver<'d, NoopRawMutex, PacketBuf>, +} + +impl<'d, const MTU: usize> Runner<'d, MTU> { + pub fn split(self) -> (RxRunner<'d, MTU>, TxRunner<'d, MTU>) { + ( + RxRunner { + link_state: self.link_state, + rx_chan: self.rx_chan, + }, + TxRunner { tx_chan: self.tx_chan }, + ) + } + + pub fn set_link_state(&mut self, state: LinkState) { + self.link_state.lock(|s| { + let s = &mut *s.borrow_mut(); + s.state = state; + s.waker.wake(); + }); + } + + pub async fn rx_buf(&mut self) -> &mut [u8] { + let p = self.rx_chan.send().await; + &mut p.buf + } + + pub fn try_rx_buf(&mut self) -> Option<&mut [u8]> { + let p = self.rx_chan.try_send()?; + Some(&mut p.buf) + } + + pub fn poll_rx_buf(&mut self, cx: &mut Context) -> Poll<&mut [u8]> { + match self.rx_chan.poll_send(cx) { + Poll::Ready(p) => Poll::Ready(&mut p.buf), + Poll::Pending => Poll::Pending, + } + } + + pub fn rx_done(&mut self, len: usize) { + let p = self.rx_chan.try_send().unwrap(); + p.len = len; + self.rx_chan.send_done(); + } + + pub async fn tx_buf(&mut self) -> &mut [u8] { + let p = self.tx_chan.recv().await; + &mut p.buf[..p.len] + } + + pub fn try_tx_buf(&mut self) -> Option<&mut [u8]> { + let p = self.tx_chan.try_recv()?; + Some(&mut p.buf[..p.len]) + } + + pub fn poll_tx_buf(&mut self, cx: &mut Context) -> Poll<&mut [u8]> { + match self.tx_chan.poll_recv(cx) { + Poll::Ready(p) => Poll::Ready(&mut p.buf[..p.len]), + Poll::Pending => Poll::Pending, + } + } + + pub fn tx_done(&mut self) { + self.tx_chan.recv_done(); + } +} + +impl<'d, const MTU: usize> RxRunner<'d, MTU> { + pub fn set_link_state(&mut self, state: LinkState) { + self.link_state.lock(|s| { + let s = &mut *s.borrow_mut(); + s.state = state; + s.waker.wake(); + }); + } + + pub async fn rx_buf(&mut self) -> &mut [u8] { + let p = self.rx_chan.send().await; + &mut p.buf + } + + pub fn try_rx_buf(&mut self) -> Option<&mut [u8]> { + let p = self.rx_chan.try_send()?; + Some(&mut p.buf) + } + + pub fn poll_rx_buf(&mut self, cx: &mut Context) -> Poll<&mut [u8]> { + match self.rx_chan.poll_send(cx) { + Poll::Ready(p) => Poll::Ready(&mut p.buf), + Poll::Pending => Poll::Pending, + } + } + + pub fn rx_done(&mut self, len: usize) { + let p = self.rx_chan.try_send().unwrap(); + p.len = len; + self.rx_chan.send_done(); + } +} + +impl<'d, const MTU: usize> TxRunner<'d, MTU> { + pub async fn tx_buf(&mut self) -> &mut [u8] { + let p = self.tx_chan.recv().await; + &mut p.buf[..p.len] + } + + pub fn try_tx_buf(&mut self) -> Option<&mut [u8]> { + let p = self.tx_chan.try_recv()?; + Some(&mut p.buf[..p.len]) + } + + pub fn poll_tx_buf(&mut self, cx: &mut Context) -> Poll<&mut [u8]> { + match self.tx_chan.poll_recv(cx) { + Poll::Ready(p) => Poll::Ready(&mut p.buf[..p.len]), + Poll::Pending => Poll::Pending, + } + } + + pub fn tx_done(&mut self) { + self.tx_chan.recv_done(); + } +} + +pub fn new<'d, const MTU: usize, const N_RX: usize, const N_TX: usize>( + state: &'d mut State, + ethernet_address: [u8; 6], +) -> (Runner<'d, MTU>, Device<'d, MTU>) { + let mut caps = Capabilities::default(); + caps.max_transmission_unit = MTU; + caps.medium = Medium::Ethernet; + + // safety: this is a self-referential struct, however: + // - it can't move while the `'d` borrow is active. + // - when the borrow ends, the dangling references inside the MaybeUninit will never be used again. + let state_uninit: *mut MaybeUninit> = + (&mut state.inner as *mut MaybeUninit>).cast(); + let state = unsafe { &mut *state_uninit }.write(StateInner { + rx: zerocopy_channel::Channel::new(&mut state.rx[..]), + tx: zerocopy_channel::Channel::new(&mut state.tx[..]), + link_state: Mutex::new(RefCell::new(LinkStateState { + state: LinkState::Down, + waker: WakerRegistration::new(), + })), + }); + + let (rx_sender, rx_receiver) = state.rx.split(); + let (tx_sender, tx_receiver) = state.tx.split(); + + ( + Runner { + tx_chan: tx_receiver, + rx_chan: rx_sender, + link_state: &state.link_state, + }, + Device { + caps, + ethernet_address, + link_state: &state.link_state, + rx: rx_receiver, + tx: tx_sender, + }, + ) +} + +pub struct PacketBuf { + len: usize, + buf: [u8; MTU], +} + +impl PacketBuf { + pub const fn new() -> Self { + Self { len: 0, buf: [0; MTU] } + } +} + +pub struct Device<'d, const MTU: usize> { + rx: zerocopy_channel::Receiver<'d, NoopRawMutex, PacketBuf>, + tx: zerocopy_channel::Sender<'d, NoopRawMutex, PacketBuf>, + link_state: &'d Mutex>, + caps: Capabilities, + ethernet_address: [u8; 6], +} + +impl<'d, const MTU: usize> embassy_net_driver::Driver for Device<'d, MTU> { + type RxToken<'a> = RxToken<'a, MTU> where Self: 'a ; + type TxToken<'a> = TxToken<'a, MTU> where Self: 'a ; + + fn receive(&mut self, cx: &mut Context) -> Option<(Self::RxToken<'_>, Self::TxToken<'_>)> { + if self.rx.poll_recv(cx).is_ready() && self.tx.poll_send(cx).is_ready() { + Some((RxToken { rx: self.rx.borrow() }, TxToken { tx: self.tx.borrow() })) + } else { + None + } + } + + /// Construct a transmit token. + fn transmit(&mut self, cx: &mut Context) -> Option> { + if self.tx.poll_send(cx).is_ready() { + Some(TxToken { tx: self.tx.borrow() }) + } else { + None + } + } + + /// Get a description of device capabilities. + fn capabilities(&self) -> Capabilities { + self.caps.clone() + } + + fn ethernet_address(&self) -> [u8; 6] { + self.ethernet_address + } + + fn link_state(&mut self, cx: &mut Context) -> LinkState { + self.link_state.lock(|s| { + let s = &mut *s.borrow_mut(); + s.waker.register(cx.waker()); + s.state + }) + } +} + +pub struct RxToken<'a, const MTU: usize> { + rx: zerocopy_channel::Receiver<'a, NoopRawMutex, PacketBuf>, +} + +impl<'a, const MTU: usize> embassy_net_driver::RxToken for RxToken<'a, MTU> { + fn consume(mut self, f: F) -> R + where + F: FnOnce(&mut [u8]) -> R, + { + // NOTE(unwrap): we checked the queue wasn't full when creating the token. + let pkt = unwrap!(self.rx.try_recv()); + let r = f(&mut pkt.buf[..pkt.len]); + self.rx.recv_done(); + r + } +} + +pub struct TxToken<'a, const MTU: usize> { + tx: zerocopy_channel::Sender<'a, NoopRawMutex, PacketBuf>, +} + +impl<'a, const MTU: usize> embassy_net_driver::TxToken for TxToken<'a, MTU> { + fn consume(mut self, len: usize, f: F) -> R + where + F: FnOnce(&mut [u8]) -> R, + { + // NOTE(unwrap): we checked the queue wasn't full when creating the token. + let pkt = unwrap!(self.tx.try_send()); + let r = f(&mut pkt.buf[..len]); + pkt.len = len; + self.tx.send_done(); + r + } +} + +mod zerocopy_channel { + use core::cell::RefCell; + use core::future::poll_fn; + use core::marker::PhantomData; + use core::task::{Context, Poll}; + + use embassy_sync::blocking_mutex::raw::RawMutex; + use embassy_sync::blocking_mutex::Mutex; + use embassy_sync::waitqueue::WakerRegistration; + + pub struct Channel<'a, M: RawMutex, T> { + buf: *mut T, + phantom: PhantomData<&'a mut T>, + state: Mutex>, + } + + impl<'a, M: RawMutex, T> Channel<'a, M, T> { + pub fn new(buf: &'a mut [T]) -> Self { + let len = buf.len(); + assert!(len != 0); + + Self { + buf: buf.as_mut_ptr(), + phantom: PhantomData, + state: Mutex::new(RefCell::new(State { + len, + front: 0, + back: 0, + full: false, + send_waker: WakerRegistration::new(), + recv_waker: WakerRegistration::new(), + })), + } + } + + pub fn split(&mut self) -> (Sender<'_, M, T>, Receiver<'_, M, T>) { + (Sender { channel: self }, Receiver { channel: self }) + } + } + + pub struct Sender<'a, M: RawMutex, T> { + channel: &'a Channel<'a, M, T>, + } + + impl<'a, M: RawMutex, T> Sender<'a, M, T> { + pub fn borrow(&mut self) -> Sender<'_, M, T> { + Sender { channel: self.channel } + } + + pub fn try_send(&mut self) -> Option<&mut T> { + self.channel.state.lock(|s| { + let s = &mut *s.borrow_mut(); + match s.push_index() { + Some(i) => Some(unsafe { &mut *self.channel.buf.add(i) }), + None => None, + } + }) + } + + pub fn poll_send(&mut self, cx: &mut Context) -> Poll<&mut T> { + self.channel.state.lock(|s| { + let s = &mut *s.borrow_mut(); + match s.push_index() { + Some(i) => Poll::Ready(unsafe { &mut *self.channel.buf.add(i) }), + None => { + s.recv_waker.register(cx.waker()); + Poll::Pending + } + } + }) + } + + pub async fn send(&mut self) -> &mut T { + let i = poll_fn(|cx| { + self.channel.state.lock(|s| { + let s = &mut *s.borrow_mut(); + match s.push_index() { + Some(i) => Poll::Ready(i), + None => { + s.recv_waker.register(cx.waker()); + Poll::Pending + } + } + }) + }) + .await; + unsafe { &mut *self.channel.buf.add(i) } + } + + pub fn send_done(&mut self) { + self.channel.state.lock(|s| s.borrow_mut().push_done()) + } + } + pub struct Receiver<'a, M: RawMutex, T> { + channel: &'a Channel<'a, M, T>, + } + + impl<'a, M: RawMutex, T> Receiver<'a, M, T> { + pub fn borrow(&mut self) -> Receiver<'_, M, T> { + Receiver { channel: self.channel } + } + + pub fn try_recv(&mut self) -> Option<&mut T> { + self.channel.state.lock(|s| { + let s = &mut *s.borrow_mut(); + match s.pop_index() { + Some(i) => Some(unsafe { &mut *self.channel.buf.add(i) }), + None => None, + } + }) + } + + pub fn poll_recv(&mut self, cx: &mut Context) -> Poll<&mut T> { + self.channel.state.lock(|s| { + let s = &mut *s.borrow_mut(); + match s.pop_index() { + Some(i) => Poll::Ready(unsafe { &mut *self.channel.buf.add(i) }), + None => { + s.send_waker.register(cx.waker()); + Poll::Pending + } + } + }) + } + + pub async fn recv(&mut self) -> &mut T { + let i = poll_fn(|cx| { + self.channel.state.lock(|s| { + let s = &mut *s.borrow_mut(); + match s.pop_index() { + Some(i) => Poll::Ready(i), + None => { + s.send_waker.register(cx.waker()); + Poll::Pending + } + } + }) + }) + .await; + unsafe { &mut *self.channel.buf.add(i) } + } + + pub fn recv_done(&mut self) { + self.channel.state.lock(|s| s.borrow_mut().pop_done()) + } + } + + struct State { + len: usize, + + /// Front index. Always 0..=(N-1) + front: usize, + /// Back index. Always 0..=(N-1). + back: usize, + + /// Used to distinguish "empty" and "full" cases when `front == back`. + /// May only be `true` if `front == back`, always `false` otherwise. + full: bool, + + send_waker: WakerRegistration, + recv_waker: WakerRegistration, + } + + impl State { + fn increment(&self, i: usize) -> usize { + if i + 1 == self.len { + 0 + } else { + i + 1 + } + } + + fn is_full(&self) -> bool { + self.full + } + + fn is_empty(&self) -> bool { + self.front == self.back && !self.full + } + + fn push_index(&mut self) -> Option { + match self.is_full() { + true => None, + false => Some(self.back), + } + } + + fn push_done(&mut self) { + assert!(!self.is_full()); + self.back = self.increment(self.back); + if self.back == self.front { + self.full = true; + } + self.send_waker.wake(); + } + + fn pop_index(&mut self) -> Option { + match self.is_empty() { + true => None, + false => Some(self.front), + } + } + + fn pop_done(&mut self) { + assert!(!self.is_empty()); + self.front = self.increment(self.front); + self.full = false; + self.recv_waker.wake(); + } + } +} diff --git a/embassy-usb/Cargo.toml b/embassy-usb/Cargo.toml index d0742ceb..1e567bb9 100644 --- a/embassy-usb/Cargo.toml +++ b/embassy-usb/Cargo.toml @@ -19,7 +19,7 @@ default = ["usbd-hid"] embassy-futures = { version = "0.1.0", path = "../embassy-futures" } embassy-usb-driver = { version = "0.1.0", path = "../embassy-usb-driver" } embassy-sync = { version = "0.1.0", path = "../embassy-sync" } -embassy-net-driver = { version = "0.1.0", path = "../embassy-net-driver" } +embassy-net-driver-channel = { version = "0.1.0", path = "../embassy-net-driver-channel" } defmt = { version = "0.3", optional = true } log = { version = "0.4.14", optional = true } diff --git a/embassy-usb/src/class/cdc_ncm/embassy_net.rs b/embassy-usb/src/class/cdc_ncm/embassy_net.rs index 093afeff..7ecf693d 100644 --- a/embassy-usb/src/class/cdc_ncm/embassy_net.rs +++ b/embassy-usb/src/class/cdc_ncm/embassy_net.rs @@ -1,81 +1,45 @@ -use core::cell::RefCell; -use core::mem::MaybeUninit; -use core::task::Context; - use embassy_futures::select::{select, Either}; -use embassy_net_driver::{Capabilities, LinkState, Medium}; -use embassy_sync::blocking_mutex::raw::NoopRawMutex; -use embassy_sync::blocking_mutex::Mutex; -use embassy_sync::waitqueue::WakerRegistration; +use embassy_net_driver_channel as ch; +use embassy_net_driver_channel::driver::LinkState; use embassy_usb_driver::Driver; use super::{CdcNcmClass, Receiver, Sender}; -pub struct State<'d, const MTU: usize, const N_RX: usize, const N_TX: usize> { - rx: [PacketBuf; N_RX], - tx: [PacketBuf; N_TX], - inner: MaybeUninit>, +pub struct State { + ch_state: ch::State, } -impl<'d, const MTU: usize, const N_RX: usize, const N_TX: usize> State<'d, MTU, N_RX, N_TX> { - const NEW_PACKET: PacketBuf = PacketBuf::new(); - +impl State { pub const fn new() -> Self { Self { - rx: [Self::NEW_PACKET; N_RX], - tx: [Self::NEW_PACKET; N_TX], - inner: MaybeUninit::uninit(), + ch_state: ch::State::new(), } } } -struct StateInner<'d, const MTU: usize> { - rx: zerocopy_channel::Channel<'d, NoopRawMutex, PacketBuf>, - tx: zerocopy_channel::Channel<'d, NoopRawMutex, PacketBuf>, - link_state: Mutex>, -} - -/// State of the LinkState -struct LinkStateState { - state: LinkState, - waker: WakerRegistration, -} - pub struct Runner<'d, D: Driver<'d>, const MTU: usize> { tx_usb: Sender<'d, D>, - tx_chan: zerocopy_channel::Receiver<'d, NoopRawMutex, PacketBuf>, rx_usb: Receiver<'d, D>, - rx_chan: zerocopy_channel::Sender<'d, NoopRawMutex, PacketBuf>, - link_state: &'d Mutex>, + ch: ch::Runner<'d, MTU>, } impl<'d, D: Driver<'d>, const MTU: usize> Runner<'d, D, MTU> { pub async fn run(mut self) -> ! { + let (mut rx_chan, mut tx_chan) = self.ch.split(); let rx_fut = async move { loop { trace!("WAITING for connection"); - self.link_state.lock(|s| { - let s = &mut *s.borrow_mut(); - s.state = LinkState::Down; - s.waker.wake(); - }); + rx_chan.set_link_state(LinkState::Down); self.rx_usb.wait_connection().await.unwrap(); trace!("Connected"); - self.link_state.lock(|s| { - let s = &mut *s.borrow_mut(); - s.state = LinkState::Up; - s.waker.wake(); - }); + rx_chan.set_link_state(LinkState::Up); loop { - let p = self.rx_chan.send().await; - match self.rx_usb.read_packet(&mut p.buf).await { - Ok(n) => { - p.len = n; - self.rx_chan.send_done(); - } + let p = rx_chan.rx_buf().await; + match self.rx_usb.read_packet(p).await { + Ok(n) => rx_chan.rx_done(n), Err(e) => { warn!("error reading packet: {:?}", e); break; @@ -86,11 +50,11 @@ impl<'d, D: Driver<'d>, const MTU: usize> Runner<'d, D, MTU> { }; let tx_fut = async move { loop { - let p = self.tx_chan.recv().await; - if let Err(e) = self.tx_usb.write_packet(&p.buf[..p.len]).await { + let p = tx_chan.tx_buf().await; + if let Err(e) = self.tx_usb.write_packet(p).await { warn!("Failed to TX packet: {:?}", e); } - self.tx_chan.recv_done(); + tx_chan.tx_done(); } }; match select(rx_fut, tx_fut).await { @@ -100,350 +64,26 @@ impl<'d, D: Driver<'d>, const MTU: usize> Runner<'d, D, MTU> { } } +// would be cool to use a TAIT here, but it gives a "may not live long enough". rustc bug? +//pub type Device<'d, const MTU: usize> = impl embassy_net_driver_channel::driver::Driver + 'd; +pub type Device<'d, const MTU: usize> = embassy_net_driver_channel::Device<'d, MTU>; + impl<'d, D: Driver<'d>> CdcNcmClass<'d, D> { pub fn into_embassy_net_device( self, - state: &'d mut State<'d, MTU, N_RX, N_TX>, + state: &'d mut State, ethernet_address: [u8; 6], ) -> (Runner<'d, D, MTU>, Device<'d, MTU>) { let (tx_usb, rx_usb) = self.split(); - - let mut caps = Capabilities::default(); - caps.max_transmission_unit = 1514; // 1500 IP + 14 ethernet header - caps.medium = Medium::Ethernet; - - let state = state.inner.write(StateInner { - rx: zerocopy_channel::Channel::new(&mut state.rx[..]), - tx: zerocopy_channel::Channel::new(&mut state.tx[..]), - link_state: Mutex::new(RefCell::new(LinkStateState { - state: LinkState::Down, - waker: WakerRegistration::new(), - })), - }); - - let (rx_sender, rx_receiver) = state.rx.split(); - let (tx_sender, tx_receiver) = state.tx.split(); + let (runner, device) = ch::new(&mut state.ch_state, ethernet_address); ( Runner { tx_usb, - tx_chan: tx_receiver, rx_usb, - rx_chan: rx_sender, - link_state: &state.link_state, - }, - Device { - caps, - ethernet_address, - link_state: &state.link_state, - rx: rx_receiver, - tx: tx_sender, + ch: runner, }, + device, ) } } - -pub struct PacketBuf { - len: usize, - buf: [u8; MTU], -} - -impl PacketBuf { - pub const fn new() -> Self { - Self { len: 0, buf: [0; MTU] } - } -} - -pub struct Device<'d, const MTU: usize> { - rx: zerocopy_channel::Receiver<'d, NoopRawMutex, PacketBuf>, - tx: zerocopy_channel::Sender<'d, NoopRawMutex, PacketBuf>, - link_state: &'d Mutex>, - caps: Capabilities, - ethernet_address: [u8; 6], -} - -impl<'d, const MTU: usize> embassy_net_driver::Driver for Device<'d, MTU> { - type RxToken<'a> = RxToken<'a, MTU> where Self: 'a ; - type TxToken<'a> = TxToken<'a, MTU> where Self: 'a ; - - fn receive(&mut self, cx: &mut Context) -> Option<(Self::RxToken<'_>, Self::TxToken<'_>)> { - if self.rx.poll_recv(cx).is_ready() && self.tx.poll_send(cx).is_ready() { - Some((RxToken { rx: self.rx.borrow() }, TxToken { tx: self.tx.borrow() })) - } else { - None - } - } - - /// Construct a transmit token. - fn transmit(&mut self, cx: &mut Context) -> Option> { - if self.tx.poll_send(cx).is_ready() { - Some(TxToken { tx: self.tx.borrow() }) - } else { - None - } - } - - /// Get a description of device capabilities. - fn capabilities(&self) -> Capabilities { - self.caps.clone() - } - - fn ethernet_address(&self) -> [u8; 6] { - self.ethernet_address - } - - fn link_state(&mut self, cx: &mut Context) -> LinkState { - self.link_state.lock(|s| { - let s = &mut *s.borrow_mut(); - s.waker.register(cx.waker()); - s.state - }) - } -} - -pub struct RxToken<'a, const MTU: usize> { - rx: zerocopy_channel::Receiver<'a, NoopRawMutex, PacketBuf>, -} - -impl<'a, const MTU: usize> embassy_net_driver::RxToken for RxToken<'a, MTU> { - fn consume(mut self, f: F) -> R - where - F: FnOnce(&mut [u8]) -> R, - { - // NOTE(unwrap): we checked the queue wasn't full when creating the token. - let pkt = unwrap!(self.rx.try_recv()); - let r = f(&mut pkt.buf[..pkt.len]); - self.rx.recv_done(); - r - } -} - -pub struct TxToken<'a, const MTU: usize> { - tx: zerocopy_channel::Sender<'a, NoopRawMutex, PacketBuf>, -} - -impl<'a, const MTU: usize> embassy_net_driver::TxToken for TxToken<'a, MTU> { - fn consume(mut self, len: usize, f: F) -> R - where - F: FnOnce(&mut [u8]) -> R, - { - // NOTE(unwrap): we checked the queue wasn't full when creating the token. - let pkt = unwrap!(self.tx.try_send()); - let r = f(&mut pkt.buf[..len]); - pkt.len = len; - self.tx.send_done(); - r - } -} - -mod zerocopy_channel { - use core::cell::RefCell; - use core::future::poll_fn; - use core::marker::PhantomData; - use core::task::{Context, Poll}; - - use embassy_sync::blocking_mutex::raw::RawMutex; - use embassy_sync::blocking_mutex::Mutex; - use embassy_sync::waitqueue::WakerRegistration; - - pub struct Channel<'a, M: RawMutex, T> { - buf: *mut T, - phantom: PhantomData<&'a mut T>, - state: Mutex>, - } - - impl<'a, M: RawMutex, T> Channel<'a, M, T> { - pub fn new(buf: &'a mut [T]) -> Self { - let len = buf.len(); - assert!(len != 0); - - Self { - buf: buf.as_mut_ptr(), - phantom: PhantomData, - state: Mutex::new(RefCell::new(State { - len, - front: 0, - back: 0, - full: false, - send_waker: WakerRegistration::new(), - recv_waker: WakerRegistration::new(), - })), - } - } - - pub fn split(&mut self) -> (Sender<'_, M, T>, Receiver<'_, M, T>) { - (Sender { channel: self }, Receiver { channel: self }) - } - } - - pub struct Sender<'a, M: RawMutex, T> { - channel: &'a Channel<'a, M, T>, - } - - impl<'a, M: RawMutex, T> Sender<'a, M, T> { - pub fn borrow(&mut self) -> Sender<'_, M, T> { - Sender { channel: self.channel } - } - - pub fn try_send(&mut self) -> Option<&mut T> { - self.channel.state.lock(|s| { - let s = &mut *s.borrow_mut(); - match s.push_index() { - Some(i) => Some(unsafe { &mut *self.channel.buf.add(i) }), - None => None, - } - }) - } - - pub fn poll_send(&mut self, cx: &mut Context) -> Poll<&mut T> { - self.channel.state.lock(|s| { - let s = &mut *s.borrow_mut(); - match s.push_index() { - Some(i) => Poll::Ready(unsafe { &mut *self.channel.buf.add(i) }), - None => { - s.recv_waker.register(cx.waker()); - Poll::Pending - } - } - }) - } - - pub async fn send(&mut self) -> &mut T { - let i = poll_fn(|cx| { - self.channel.state.lock(|s| { - let s = &mut *s.borrow_mut(); - match s.push_index() { - Some(i) => Poll::Ready(i), - None => { - s.recv_waker.register(cx.waker()); - Poll::Pending - } - } - }) - }) - .await; - unsafe { &mut *self.channel.buf.add(i) } - } - - pub fn send_done(&mut self) { - self.channel.state.lock(|s| s.borrow_mut().push_done()) - } - } - pub struct Receiver<'a, M: RawMutex, T> { - channel: &'a Channel<'a, M, T>, - } - - impl<'a, M: RawMutex, T> Receiver<'a, M, T> { - pub fn borrow(&mut self) -> Receiver<'_, M, T> { - Receiver { channel: self.channel } - } - - pub fn try_recv(&mut self) -> Option<&mut T> { - self.channel.state.lock(|s| { - let s = &mut *s.borrow_mut(); - match s.pop_index() { - Some(i) => Some(unsafe { &mut *self.channel.buf.add(i) }), - None => None, - } - }) - } - - pub fn poll_recv(&mut self, cx: &mut Context) -> Poll<&mut T> { - self.channel.state.lock(|s| { - let s = &mut *s.borrow_mut(); - match s.pop_index() { - Some(i) => Poll::Ready(unsafe { &mut *self.channel.buf.add(i) }), - None => { - s.send_waker.register(cx.waker()); - Poll::Pending - } - } - }) - } - - pub async fn recv(&mut self) -> &mut T { - let i = poll_fn(|cx| { - self.channel.state.lock(|s| { - let s = &mut *s.borrow_mut(); - match s.pop_index() { - Some(i) => Poll::Ready(i), - None => { - s.send_waker.register(cx.waker()); - Poll::Pending - } - } - }) - }) - .await; - unsafe { &mut *self.channel.buf.add(i) } - } - - pub fn recv_done(&mut self) { - self.channel.state.lock(|s| s.borrow_mut().pop_done()) - } - } - - struct State { - len: usize, - - /// Front index. Always 0..=(N-1) - front: usize, - /// Back index. Always 0..=(N-1). - back: usize, - - /// Used to distinguish "empty" and "full" cases when `front == back`. - /// May only be `true` if `front == back`, always `false` otherwise. - full: bool, - - send_waker: WakerRegistration, - recv_waker: WakerRegistration, - } - - impl State { - fn increment(&self, i: usize) -> usize { - if i + 1 == self.len { - 0 - } else { - i + 1 - } - } - - fn is_full(&self) -> bool { - self.full - } - - fn is_empty(&self) -> bool { - self.front == self.back && !self.full - } - - fn push_index(&mut self) -> Option { - match self.is_full() { - true => None, - false => Some(self.back), - } - } - - fn push_done(&mut self) { - assert!(!self.is_full()); - self.back = self.increment(self.back); - if self.back == self.front { - self.full = true; - } - self.send_waker.wake(); - } - - fn pop_index(&mut self) -> Option { - match self.is_empty() { - true => None, - false => Some(self.front), - } - } - - fn pop_done(&mut self) { - assert!(!self.is_empty()); - self.front = self.increment(self.front); - self.full = false; - self.recv_waker.wake(); - } - } -}