aboutsummaryrefslogtreecommitdiff
path: root/embassy-net-driver-channel
diff options
context:
space:
mode:
authorDario Nieuwenhuis <[email protected]>2022-12-26 04:31:49 +0100
committerDario Nieuwenhuis <[email protected]>2022-12-26 04:49:08 +0100
commit007246b16036c3aa874e78d0665567a27ab35fa9 (patch)
tree451d75b390e751e2ef6401b0b60c137bcaf92b29 /embassy-net-driver-channel
parent1f033d509afb4e590a81896de66af683fda4e706 (diff)
net: split channel-based driver impl from usb cdc-ncm into a separate crate.
Diffstat (limited to 'embassy-net-driver-channel')
-rw-r--r--embassy-net-driver-channel/Cargo.toml12
-rw-r--r--embassy-net-driver-channel/src/fmt.rs225
-rw-r--r--embassy-net-driver-channel/src/lib.rs525
3 files changed, 762 insertions, 0 deletions
diff --git a/embassy-net-driver-channel/Cargo.toml b/embassy-net-driver-channel/Cargo.toml
new file mode 100644
index 000000000..700a4e8a0
--- /dev/null
+++ b/embassy-net-driver-channel/Cargo.toml
@@ -0,0 +1,12 @@
1[package]
2name = "embassy-net-driver-channel"
3version = "0.1.0"
4edition = "2021"
5
6[dependencies]
7defmt = { version = "0.3", optional = true }
8log = { version = "0.4.14", optional = true }
9
10embassy-sync = { version = "0.1.0", path = "../embassy-sync" }
11embassy-futures = { version = "0.1.0", path = "../embassy-futures" }
12embassy-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 000000000..066970813
--- /dev/null
+++ b/embassy-net-driver-channel/src/fmt.rs
@@ -0,0 +1,225 @@
1#![macro_use]
2#![allow(unused_macros)]
3
4#[cfg(all(feature = "defmt", feature = "log"))]
5compile_error!("You may not enable both `defmt` and `log` features.");
6
7macro_rules! assert {
8 ($($x:tt)*) => {
9 {
10 #[cfg(not(feature = "defmt"))]
11 ::core::assert!($($x)*);
12 #[cfg(feature = "defmt")]
13 ::defmt::assert!($($x)*);
14 }
15 };
16}
17
18macro_rules! assert_eq {
19 ($($x:tt)*) => {
20 {
21 #[cfg(not(feature = "defmt"))]
22 ::core::assert_eq!($($x)*);
23 #[cfg(feature = "defmt")]
24 ::defmt::assert_eq!($($x)*);
25 }
26 };
27}
28
29macro_rules! assert_ne {
30 ($($x:tt)*) => {
31 {
32 #[cfg(not(feature = "defmt"))]
33 ::core::assert_ne!($($x)*);
34 #[cfg(feature = "defmt")]
35 ::defmt::assert_ne!($($x)*);
36 }
37 };
38}
39
40macro_rules! debug_assert {
41 ($($x:tt)*) => {
42 {
43 #[cfg(not(feature = "defmt"))]
44 ::core::debug_assert!($($x)*);
45 #[cfg(feature = "defmt")]
46 ::defmt::debug_assert!($($x)*);
47 }
48 };
49}
50
51macro_rules! debug_assert_eq {
52 ($($x:tt)*) => {
53 {
54 #[cfg(not(feature = "defmt"))]
55 ::core::debug_assert_eq!($($x)*);
56 #[cfg(feature = "defmt")]
57 ::defmt::debug_assert_eq!($($x)*);
58 }
59 };
60}
61
62macro_rules! debug_assert_ne {
63 ($($x:tt)*) => {
64 {
65 #[cfg(not(feature = "defmt"))]
66 ::core::debug_assert_ne!($($x)*);
67 #[cfg(feature = "defmt")]
68 ::defmt::debug_assert_ne!($($x)*);
69 }
70 };
71}
72
73macro_rules! todo {
74 ($($x:tt)*) => {
75 {
76 #[cfg(not(feature = "defmt"))]
77 ::core::todo!($($x)*);
78 #[cfg(feature = "defmt")]
79 ::defmt::todo!($($x)*);
80 }
81 };
82}
83
84macro_rules! unreachable {
85 ($($x:tt)*) => {
86 {
87 #[cfg(not(feature = "defmt"))]
88 ::core::unreachable!($($x)*);
89 #[cfg(feature = "defmt")]
90 ::defmt::unreachable!($($x)*);
91 }
92 };
93}
94
95macro_rules! panic {
96 ($($x:tt)*) => {
97 {
98 #[cfg(not(feature = "defmt"))]
99 ::core::panic!($($x)*);
100 #[cfg(feature = "defmt")]
101 ::defmt::panic!($($x)*);
102 }
103 };
104}
105
106macro_rules! trace {
107 ($s:literal $(, $x:expr)* $(,)?) => {
108 {
109 #[cfg(feature = "log")]
110 ::log::trace!($s $(, $x)*);
111 #[cfg(feature = "defmt")]
112 ::defmt::trace!($s $(, $x)*);
113 #[cfg(not(any(feature = "log", feature="defmt")))]
114 let _ = ($( & $x ),*);
115 }
116 };
117}
118
119macro_rules! debug {
120 ($s:literal $(, $x:expr)* $(,)?) => {
121 {
122 #[cfg(feature = "log")]
123 ::log::debug!($s $(, $x)*);
124 #[cfg(feature = "defmt")]
125 ::defmt::debug!($s $(, $x)*);
126 #[cfg(not(any(feature = "log", feature="defmt")))]
127 let _ = ($( & $x ),*);
128 }
129 };
130}
131
132macro_rules! info {
133 ($s:literal $(, $x:expr)* $(,)?) => {
134 {
135 #[cfg(feature = "log")]
136 ::log::info!($s $(, $x)*);
137 #[cfg(feature = "defmt")]
138 ::defmt::info!($s $(, $x)*);
139 #[cfg(not(any(feature = "log", feature="defmt")))]
140 let _ = ($( & $x ),*);
141 }
142 };
143}
144
145macro_rules! warn {
146 ($s:literal $(, $x:expr)* $(,)?) => {
147 {
148 #[cfg(feature = "log")]
149 ::log::warn!($s $(, $x)*);
150 #[cfg(feature = "defmt")]
151 ::defmt::warn!($s $(, $x)*);
152 #[cfg(not(any(feature = "log", feature="defmt")))]
153 let _ = ($( & $x ),*);
154 }
155 };
156}
157
158macro_rules! error {
159 ($s:literal $(, $x:expr)* $(,)?) => {
160 {
161 #[cfg(feature = "log")]
162 ::log::error!($s $(, $x)*);
163 #[cfg(feature = "defmt")]
164 ::defmt::error!($s $(, $x)*);
165 #[cfg(not(any(feature = "log", feature="defmt")))]
166 let _ = ($( & $x ),*);
167 }
168 };
169}
170
171#[cfg(feature = "defmt")]
172macro_rules! unwrap {
173 ($($x:tt)*) => {
174 ::defmt::unwrap!($($x)*)
175 };
176}
177
178#[cfg(not(feature = "defmt"))]
179macro_rules! unwrap {
180 ($arg:expr) => {
181 match $crate::fmt::Try::into_result($arg) {
182 ::core::result::Result::Ok(t) => t,
183 ::core::result::Result::Err(e) => {
184 ::core::panic!("unwrap of `{}` failed: {:?}", ::core::stringify!($arg), e);
185 }
186 }
187 };
188 ($arg:expr, $($msg:expr),+ $(,)? ) => {
189 match $crate::fmt::Try::into_result($arg) {
190 ::core::result::Result::Ok(t) => t,
191 ::core::result::Result::Err(e) => {
192 ::core::panic!("unwrap of `{}` failed: {}: {:?}", ::core::stringify!($arg), ::core::format_args!($($msg,)*), e);
193 }
194 }
195 }
196}
197
198#[derive(Debug, Copy, Clone, Eq, PartialEq)]
199pub struct NoneError;
200
201pub trait Try {
202 type Ok;
203 type Error;
204 fn into_result(self) -> Result<Self::Ok, Self::Error>;
205}
206
207impl<T> Try for Option<T> {
208 type Ok = T;
209 type Error = NoneError;
210
211 #[inline]
212 fn into_result(self) -> Result<T, NoneError> {
213 self.ok_or(NoneError)
214 }
215}
216
217impl<T, E> Try for Result<T, E> {
218 type Ok = T;
219 type Error = E;
220
221 #[inline]
222 fn into_result(self) -> Self {
223 self
224 }
225}
diff --git a/embassy-net-driver-channel/src/lib.rs b/embassy-net-driver-channel/src/lib.rs
new file mode 100644
index 000000000..369dc5a9d
--- /dev/null
+++ b/embassy-net-driver-channel/src/lib.rs
@@ -0,0 +1,525 @@
1#![no_std]
2
3// must go first!
4mod fmt;
5
6use core::cell::RefCell;
7use core::mem::MaybeUninit;
8use core::task::{Context, Poll};
9
10pub use embassy_net_driver as driver;
11use embassy_net_driver::{Capabilities, LinkState, Medium};
12use embassy_sync::blocking_mutex::raw::NoopRawMutex;
13use embassy_sync::blocking_mutex::Mutex;
14use embassy_sync::waitqueue::WakerRegistration;
15
16pub struct State<const MTU: usize, const N_RX: usize, const N_TX: usize> {
17 rx: [PacketBuf<MTU>; N_RX],
18 tx: [PacketBuf<MTU>; N_TX],
19 inner: MaybeUninit<StateInner<'static, MTU>>,
20}
21
22impl<const MTU: usize, const N_RX: usize, const N_TX: usize> State<MTU, N_RX, N_TX> {
23 const NEW_PACKET: PacketBuf<MTU> = PacketBuf::new();
24
25 pub const fn new() -> Self {
26 Self {
27 rx: [Self::NEW_PACKET; N_RX],
28 tx: [Self::NEW_PACKET; N_TX],
29 inner: MaybeUninit::uninit(),
30 }
31 }
32}
33
34struct StateInner<'d, const MTU: usize> {
35 rx: zerocopy_channel::Channel<'d, NoopRawMutex, PacketBuf<MTU>>,
36 tx: zerocopy_channel::Channel<'d, NoopRawMutex, PacketBuf<MTU>>,
37 link_state: Mutex<NoopRawMutex, RefCell<LinkStateState>>,
38}
39
40/// State of the LinkState
41struct LinkStateState {
42 state: LinkState,
43 waker: WakerRegistration,
44}
45
46pub struct Runner<'d, const MTU: usize> {
47 tx_chan: zerocopy_channel::Receiver<'d, NoopRawMutex, PacketBuf<MTU>>,
48 rx_chan: zerocopy_channel::Sender<'d, NoopRawMutex, PacketBuf<MTU>>,
49 link_state: &'d Mutex<NoopRawMutex, RefCell<LinkStateState>>,
50}
51
52pub struct RxRunner<'d, const MTU: usize> {
53 rx_chan: zerocopy_channel::Sender<'d, NoopRawMutex, PacketBuf<MTU>>,
54 link_state: &'d Mutex<NoopRawMutex, RefCell<LinkStateState>>,
55}
56
57pub struct TxRunner<'d, const MTU: usize> {
58 tx_chan: zerocopy_channel::Receiver<'d, NoopRawMutex, PacketBuf<MTU>>,
59}
60
61impl<'d, const MTU: usize> Runner<'d, MTU> {
62 pub fn split(self) -> (RxRunner<'d, MTU>, TxRunner<'d, MTU>) {
63 (
64 RxRunner {
65 link_state: self.link_state,
66 rx_chan: self.rx_chan,
67 },
68 TxRunner { tx_chan: self.tx_chan },
69 )
70 }
71
72 pub fn set_link_state(&mut self, state: LinkState) {
73 self.link_state.lock(|s| {
74 let s = &mut *s.borrow_mut();
75 s.state = state;
76 s.waker.wake();
77 });
78 }
79
80 pub async fn rx_buf(&mut self) -> &mut [u8] {
81 let p = self.rx_chan.send().await;
82 &mut p.buf
83 }
84
85 pub fn try_rx_buf(&mut self) -> Option<&mut [u8]> {
86 let p = self.rx_chan.try_send()?;
87 Some(&mut p.buf)
88 }
89
90 pub fn poll_rx_buf(&mut self, cx: &mut Context) -> Poll<&mut [u8]> {
91 match self.rx_chan.poll_send(cx) {
92 Poll::Ready(p) => Poll::Ready(&mut p.buf),
93 Poll::Pending => Poll::Pending,
94 }
95 }
96
97 pub fn rx_done(&mut self, len: usize) {
98 let p = self.rx_chan.try_send().unwrap();
99 p.len = len;
100 self.rx_chan.send_done();
101 }
102
103 pub async fn tx_buf(&mut self) -> &mut [u8] {
104 let p = self.tx_chan.recv().await;
105 &mut p.buf[..p.len]
106 }
107
108 pub fn try_tx_buf(&mut self) -> Option<&mut [u8]> {
109 let p = self.tx_chan.try_recv()?;
110 Some(&mut p.buf[..p.len])
111 }
112
113 pub fn poll_tx_buf(&mut self, cx: &mut Context) -> Poll<&mut [u8]> {
114 match self.tx_chan.poll_recv(cx) {
115 Poll::Ready(p) => Poll::Ready(&mut p.buf[..p.len]),
116 Poll::Pending => Poll::Pending,
117 }
118 }
119
120 pub fn tx_done(&mut self) {
121 self.tx_chan.recv_done();
122 }
123}
124
125impl<'d, const MTU: usize> RxRunner<'d, MTU> {
126 pub fn set_link_state(&mut self, state: LinkState) {
127 self.link_state.lock(|s| {
128 let s = &mut *s.borrow_mut();
129 s.state = state;
130 s.waker.wake();
131 });
132 }
133
134 pub async fn rx_buf(&mut self) -> &mut [u8] {
135 let p = self.rx_chan.send().await;
136 &mut p.buf
137 }
138
139 pub fn try_rx_buf(&mut self) -> Option<&mut [u8]> {
140 let p = self.rx_chan.try_send()?;
141 Some(&mut p.buf)
142 }
143
144 pub fn poll_rx_buf(&mut self, cx: &mut Context) -> Poll<&mut [u8]> {
145 match self.rx_chan.poll_send(cx) {
146 Poll::Ready(p) => Poll::Ready(&mut p.buf),
147 Poll::Pending => Poll::Pending,
148 }
149 }
150
151 pub fn rx_done(&mut self, len: usize) {
152 let p = self.rx_chan.try_send().unwrap();
153 p.len = len;
154 self.rx_chan.send_done();
155 }
156}
157
158impl<'d, const MTU: usize> TxRunner<'d, MTU> {
159 pub async fn tx_buf(&mut self) -> &mut [u8] {
160 let p = self.tx_chan.recv().await;
161 &mut p.buf[..p.len]
162 }
163
164 pub fn try_tx_buf(&mut self) -> Option<&mut [u8]> {
165 let p = self.tx_chan.try_recv()?;
166 Some(&mut p.buf[..p.len])
167 }
168
169 pub fn poll_tx_buf(&mut self, cx: &mut Context) -> Poll<&mut [u8]> {
170 match self.tx_chan.poll_recv(cx) {
171 Poll::Ready(p) => Poll::Ready(&mut p.buf[..p.len]),
172 Poll::Pending => Poll::Pending,
173 }
174 }
175
176 pub fn tx_done(&mut self) {
177 self.tx_chan.recv_done();
178 }
179}
180
181pub fn new<'d, const MTU: usize, const N_RX: usize, const N_TX: usize>(
182 state: &'d mut State<MTU, N_RX, N_TX>,
183 ethernet_address: [u8; 6],
184) -> (Runner<'d, MTU>, Device<'d, MTU>) {
185 let mut caps = Capabilities::default();
186 caps.max_transmission_unit = MTU;
187 caps.medium = Medium::Ethernet;
188
189 // safety: this is a self-referential struct, however:
190 // - it can't move while the `'d` borrow is active.
191 // - when the borrow ends, the dangling references inside the MaybeUninit will never be used again.
192 let state_uninit: *mut MaybeUninit<StateInner<'d, MTU>> =
193 (&mut state.inner as *mut MaybeUninit<StateInner<'static, MTU>>).cast();
194 let state = unsafe { &mut *state_uninit }.write(StateInner {
195 rx: zerocopy_channel::Channel::new(&mut state.rx[..]),
196 tx: zerocopy_channel::Channel::new(&mut state.tx[..]),
197 link_state: Mutex::new(RefCell::new(LinkStateState {
198 state: LinkState::Down,
199 waker: WakerRegistration::new(),
200 })),
201 });
202
203 let (rx_sender, rx_receiver) = state.rx.split();
204 let (tx_sender, tx_receiver) = state.tx.split();
205
206 (
207 Runner {
208 tx_chan: tx_receiver,
209 rx_chan: rx_sender,
210 link_state: &state.link_state,
211 },
212 Device {
213 caps,
214 ethernet_address,
215 link_state: &state.link_state,
216 rx: rx_receiver,
217 tx: tx_sender,
218 },
219 )
220}
221
222pub struct PacketBuf<const MTU: usize> {
223 len: usize,
224 buf: [u8; MTU],
225}
226
227impl<const MTU: usize> PacketBuf<MTU> {
228 pub const fn new() -> Self {
229 Self { len: 0, buf: [0; MTU] }
230 }
231}
232
233pub struct Device<'d, const MTU: usize> {
234 rx: zerocopy_channel::Receiver<'d, NoopRawMutex, PacketBuf<MTU>>,
235 tx: zerocopy_channel::Sender<'d, NoopRawMutex, PacketBuf<MTU>>,
236 link_state: &'d Mutex<NoopRawMutex, RefCell<LinkStateState>>,
237 caps: Capabilities,
238 ethernet_address: [u8; 6],
239}
240
241impl<'d, const MTU: usize> embassy_net_driver::Driver for Device<'d, MTU> {
242 type RxToken<'a> = RxToken<'a, MTU> where Self: 'a ;
243 type TxToken<'a> = TxToken<'a, MTU> where Self: 'a ;
244
245 fn receive(&mut self, cx: &mut Context) -> Option<(Self::RxToken<'_>, Self::TxToken<'_>)> {
246 if self.rx.poll_recv(cx).is_ready() && self.tx.poll_send(cx).is_ready() {
247 Some((RxToken { rx: self.rx.borrow() }, TxToken { tx: self.tx.borrow() }))
248 } else {
249 None
250 }
251 }
252
253 /// Construct a transmit token.
254 fn transmit(&mut self, cx: &mut Context) -> Option<Self::TxToken<'_>> {
255 if self.tx.poll_send(cx).is_ready() {
256 Some(TxToken { tx: self.tx.borrow() })
257 } else {
258 None
259 }
260 }
261
262 /// Get a description of device capabilities.
263 fn capabilities(&self) -> Capabilities {
264 self.caps.clone()
265 }
266
267 fn ethernet_address(&self) -> [u8; 6] {
268 self.ethernet_address
269 }
270
271 fn link_state(&mut self, cx: &mut Context) -> LinkState {
272 self.link_state.lock(|s| {
273 let s = &mut *s.borrow_mut();
274 s.waker.register(cx.waker());
275 s.state
276 })
277 }
278}
279
280pub struct RxToken<'a, const MTU: usize> {
281 rx: zerocopy_channel::Receiver<'a, NoopRawMutex, PacketBuf<MTU>>,
282}
283
284impl<'a, const MTU: usize> embassy_net_driver::RxToken for RxToken<'a, MTU> {
285 fn consume<R, F>(mut self, f: F) -> R
286 where
287 F: FnOnce(&mut [u8]) -> R,
288 {
289 // NOTE(unwrap): we checked the queue wasn't full when creating the token.
290 let pkt = unwrap!(self.rx.try_recv());
291 let r = f(&mut pkt.buf[..pkt.len]);
292 self.rx.recv_done();
293 r
294 }
295}
296
297pub struct TxToken<'a, const MTU: usize> {
298 tx: zerocopy_channel::Sender<'a, NoopRawMutex, PacketBuf<MTU>>,
299}
300
301impl<'a, const MTU: usize> embassy_net_driver::TxToken for TxToken<'a, MTU> {
302 fn consume<R, F>(mut self, len: usize, f: F) -> R
303 where
304 F: FnOnce(&mut [u8]) -> R,
305 {
306 // NOTE(unwrap): we checked the queue wasn't full when creating the token.
307 let pkt = unwrap!(self.tx.try_send());
308 let r = f(&mut pkt.buf[..len]);
309 pkt.len = len;
310 self.tx.send_done();
311 r
312 }
313}
314
315mod zerocopy_channel {
316 use core::cell::RefCell;
317 use core::future::poll_fn;
318 use core::marker::PhantomData;
319 use core::task::{Context, Poll};
320
321 use embassy_sync::blocking_mutex::raw::RawMutex;
322 use embassy_sync::blocking_mutex::Mutex;
323 use embassy_sync::waitqueue::WakerRegistration;
324
325 pub struct Channel<'a, M: RawMutex, T> {
326 buf: *mut T,
327 phantom: PhantomData<&'a mut T>,
328 state: Mutex<M, RefCell<State>>,
329 }
330
331 impl<'a, M: RawMutex, T> Channel<'a, M, T> {
332 pub fn new(buf: &'a mut [T]) -> Self {
333 let len = buf.len();
334 assert!(len != 0);
335
336 Self {
337 buf: buf.as_mut_ptr(),
338 phantom: PhantomData,
339 state: Mutex::new(RefCell::new(State {
340 len,
341 front: 0,
342 back: 0,
343 full: false,
344 send_waker: WakerRegistration::new(),
345 recv_waker: WakerRegistration::new(),
346 })),
347 }
348 }
349
350 pub fn split(&mut self) -> (Sender<'_, M, T>, Receiver<'_, M, T>) {
351 (Sender { channel: self }, Receiver { channel: self })
352 }
353 }
354
355 pub struct Sender<'a, M: RawMutex, T> {
356 channel: &'a Channel<'a, M, T>,
357 }
358
359 impl<'a, M: RawMutex, T> Sender<'a, M, T> {
360 pub fn borrow(&mut self) -> Sender<'_, M, T> {
361 Sender { channel: self.channel }
362 }
363
364 pub fn try_send(&mut self) -> Option<&mut T> {
365 self.channel.state.lock(|s| {
366 let s = &mut *s.borrow_mut();
367 match s.push_index() {
368 Some(i) => Some(unsafe { &mut *self.channel.buf.add(i) }),
369 None => None,
370 }
371 })
372 }
373
374 pub fn poll_send(&mut self, cx: &mut Context) -> Poll<&mut T> {
375 self.channel.state.lock(|s| {
376 let s = &mut *s.borrow_mut();
377 match s.push_index() {
378 Some(i) => Poll::Ready(unsafe { &mut *self.channel.buf.add(i) }),
379 None => {
380 s.recv_waker.register(cx.waker());
381 Poll::Pending
382 }
383 }
384 })
385 }
386
387 pub async fn send(&mut self) -> &mut T {
388 let i = poll_fn(|cx| {
389 self.channel.state.lock(|s| {
390 let s = &mut *s.borrow_mut();
391 match s.push_index() {
392 Some(i) => Poll::Ready(i),
393 None => {
394 s.recv_waker.register(cx.waker());
395 Poll::Pending
396 }
397 }
398 })
399 })
400 .await;
401 unsafe { &mut *self.channel.buf.add(i) }
402 }
403
404 pub fn send_done(&mut self) {
405 self.channel.state.lock(|s| s.borrow_mut().push_done())
406 }
407 }
408 pub struct Receiver<'a, M: RawMutex, T> {
409 channel: &'a Channel<'a, M, T>,
410 }
411
412 impl<'a, M: RawMutex, T> Receiver<'a, M, T> {
413 pub fn borrow(&mut self) -> Receiver<'_, M, T> {
414 Receiver { channel: self.channel }
415 }
416
417 pub fn try_recv(&mut self) -> Option<&mut T> {
418 self.channel.state.lock(|s| {
419 let s = &mut *s.borrow_mut();
420 match s.pop_index() {
421 Some(i) => Some(unsafe { &mut *self.channel.buf.add(i) }),
422 None => None,
423 }
424 })
425 }
426
427 pub fn poll_recv(&mut self, cx: &mut Context) -> Poll<&mut T> {
428 self.channel.state.lock(|s| {
429 let s = &mut *s.borrow_mut();
430 match s.pop_index() {
431 Some(i) => Poll::Ready(unsafe { &mut *self.channel.buf.add(i) }),
432 None => {
433 s.send_waker.register(cx.waker());
434 Poll::Pending
435 }
436 }
437 })
438 }
439
440 pub async fn recv(&mut self) -> &mut T {
441 let i = poll_fn(|cx| {
442 self.channel.state.lock(|s| {
443 let s = &mut *s.borrow_mut();
444 match s.pop_index() {
445 Some(i) => Poll::Ready(i),
446 None => {
447 s.send_waker.register(cx.waker());
448 Poll::Pending
449 }
450 }
451 })
452 })
453 .await;
454 unsafe { &mut *self.channel.buf.add(i) }
455 }
456
457 pub fn recv_done(&mut self) {
458 self.channel.state.lock(|s| s.borrow_mut().pop_done())
459 }
460 }
461
462 struct State {
463 len: usize,
464
465 /// Front index. Always 0..=(N-1)
466 front: usize,
467 /// Back index. Always 0..=(N-1).
468 back: usize,
469
470 /// Used to distinguish "empty" and "full" cases when `front == back`.
471 /// May only be `true` if `front == back`, always `false` otherwise.
472 full: bool,
473
474 send_waker: WakerRegistration,
475 recv_waker: WakerRegistration,
476 }
477
478 impl State {
479 fn increment(&self, i: usize) -> usize {
480 if i + 1 == self.len {
481 0
482 } else {
483 i + 1
484 }
485 }
486
487 fn is_full(&self) -> bool {
488 self.full
489 }
490
491 fn is_empty(&self) -> bool {
492 self.front == self.back && !self.full
493 }
494
495 fn push_index(&mut self) -> Option<usize> {
496 match self.is_full() {
497 true => None,
498 false => Some(self.back),
499 }
500 }
501
502 fn push_done(&mut self) {
503 assert!(!self.is_full());
504 self.back = self.increment(self.back);
505 if self.back == self.front {
506 self.full = true;
507 }
508 self.send_waker.wake();
509 }
510
511 fn pop_index(&mut self) -> Option<usize> {
512 match self.is_empty() {
513 true => None,
514 false => Some(self.front),
515 }
516 }
517
518 fn pop_done(&mut self) {
519 assert!(!self.is_empty());
520 self.front = self.increment(self.front);
521 self.full = false;
522 self.recv_waker.wake();
523 }
524 }
525}