rd_net/
lib.rs

1#![no_std]
2
3extern crate alloc;
4
5use alloc::{boxed::Box, collections::BTreeMap, sync::Arc};
6use core::{alloc::Layout, cell::UnsafeCell};
7
8use dma_api::{ContiguousBuffer, ContiguousBufferPool, DeviceDma, DmaDirection, DmaOp};
9use futures::task::AtomicWaker;
10pub use rdif_eth::*;
11
12fn other_error(msg: &'static str) -> NetError {
13    NetError::Other(Box::new(KError::Unknown(msg)))
14}
15
16struct QueueWakerMap(UnsafeCell<BTreeMap<usize, Arc<AtomicWaker>>>);
17
18impl QueueWakerMap {
19    fn new() -> Self {
20        Self(UnsafeCell::new(BTreeMap::new()))
21    }
22
23    fn register(&self, queue_id: usize) -> Arc<AtomicWaker> {
24        let waker = Arc::new(AtomicWaker::new());
25        unsafe { &mut *self.0.get() }.insert(queue_id, waker.clone());
26        waker
27    }
28
29    fn wake(&self, queue_id: usize) {
30        if let Some(waker) = unsafe { &*self.0.get() }.get(&queue_id) {
31            waker.wake();
32        }
33    }
34}
35
36struct NetInner {
37    interface: UnsafeCell<Box<dyn Interface>>,
38    dma_op: &'static dyn DmaOp,
39    tx_wakers: QueueWakerMap,
40    rx_wakers: QueueWakerMap,
41}
42
43unsafe impl Send for NetInner {}
44unsafe impl Sync for NetInner {}
45
46struct IrqGuard<'a> {
47    enabled: bool,
48    inner: &'a Net,
49}
50
51impl Drop for IrqGuard<'_> {
52    fn drop(&mut self) {
53        if self.enabled {
54            self.inner.interface().enable_irq();
55        }
56    }
57}
58
59pub struct Net {
60    inner: Arc<NetInner>,
61}
62
63impl DriverGeneric for Net {
64    fn name(&self) -> &str {
65        self.interface().name()
66    }
67
68    fn raw_any(&self) -> Option<&dyn core::any::Any> {
69        Some(self)
70    }
71
72    fn raw_any_mut(&mut self) -> Option<&mut dyn core::any::Any> {
73        Some(self)
74    }
75}
76
77impl Net {
78    pub fn new(interface: impl Interface, dma_op: &'static dyn DmaOp) -> Self {
79        Self {
80            inner: Arc::new(NetInner {
81                interface: UnsafeCell::new(Box::new(interface)),
82                dma_op,
83                tx_wakers: QueueWakerMap::new(),
84                rx_wakers: QueueWakerMap::new(),
85            }),
86        }
87    }
88
89    #[allow(clippy::mut_from_ref)]
90    fn interface(&self) -> &mut dyn Interface {
91        unsafe { &mut **self.inner.interface.get() }
92    }
93
94    fn irq_guard(&self) -> IrqGuard<'_> {
95        let enabled = self.interface().is_irq_enabled();
96        if enabled {
97            self.interface().disable_irq();
98        }
99        IrqGuard {
100            enabled,
101            inner: self,
102        }
103    }
104
105    pub fn mac_address(&self) -> [u8; 6] {
106        self.interface().mac_address()
107    }
108
109    pub fn create_tx_queue(&mut self) -> Result<TxQueue, NetError> {
110        let irq_guard = self.irq_guard();
111        let queue = self
112            .interface()
113            .create_tx_queue()
114            .ok_or_else(|| other_error("failed to create tx queue"))?;
115        let config = queue.config();
116        let pool = make_pool(self.inner.dma_op, config, DmaDirection::ToDevice)?;
117        let waker = self.inner.tx_wakers.register(queue.id());
118        drop(irq_guard);
119
120        Ok(TxQueue {
121            interface: queue,
122            pool,
123            inflight: BTreeMap::new(),
124            config,
125            _waker: waker,
126        })
127    }
128
129    pub fn create_rx_queue(&mut self) -> Result<RxQueue, NetError> {
130        let irq_guard = self.irq_guard();
131        let queue = self
132            .interface()
133            .create_rx_queue()
134            .ok_or_else(|| other_error("failed to create rx queue"))?;
135        let config = queue.config();
136        let pool = make_pool(self.inner.dma_op, config, DmaDirection::FromDevice)?;
137        let waker = self.inner.rx_wakers.register(queue.id());
138        drop(irq_guard);
139
140        let mut rx = RxQueue {
141            interface: queue,
142            pool,
143            inflight: BTreeMap::new(),
144            config,
145            _waker: waker,
146        };
147        rx.prefill()?;
148        Ok(rx)
149    }
150
151    pub fn irq_handler(&self) -> IrqHandler {
152        IrqHandler {
153            inner: self.inner.clone(),
154        }
155    }
156}
157
158fn make_pool(
159    dma_op: &'static dyn DmaOp,
160    config: QueueConfig,
161    direction: DmaDirection,
162) -> Result<ContiguousBufferPool, NetError> {
163    let layout = Layout::from_size_align(config.buf_size, config.align.max(1))
164        .map_err(|_| other_error("invalid queue layout"))?;
165    let dma = DeviceDma::new(config.dma_mask, dma_op);
166    Ok(dma.contiguous_buffer_pool(layout, direction, config.ring_size))
167}
168
169pub struct IrqHandler {
170    inner: Arc<NetInner>,
171}
172
173unsafe impl Sync for IrqHandler {}
174
175impl IrqHandler {
176    pub fn handle(&self) {
177        let iface = unsafe { &mut **self.inner.interface.get() };
178        let event = iface.handle_irq();
179        for id in event.tx_queue.iter() {
180            self.inner.tx_wakers.wake(id);
181        }
182        for id in event.rx_queue.iter() {
183            self.inner.rx_wakers.wake(id);
184        }
185    }
186}
187
188pub struct TxQueue {
189    interface: Box<dyn ITxQueue>,
190    pool: ContiguousBufferPool,
191    inflight: BTreeMap<u64, ContiguousBuffer>,
192    config: QueueConfig,
193    _waker: Arc<AtomicWaker>,
194}
195
196impl TxQueue {
197    fn capacity(&self) -> usize {
198        self.config.ring_size.saturating_sub(1)
199    }
200
201    fn reclaim_bounded(&mut self, limit: usize) -> Result<usize, NetError> {
202        let mut reclaimed = 0;
203        while reclaimed < limit {
204            let Some(bus_addr) = self.interface.reclaim() else {
205                break;
206            };
207            let Some(buff) = self.inflight.remove(&bus_addr) else {
208                return Err(other_error("reclaimed unknown tx buffer"));
209            };
210            drop(buff);
211            reclaimed += 1;
212        }
213        Ok(reclaimed)
214    }
215
216    pub fn id(&self) -> usize {
217        self.interface.id()
218    }
219
220    pub fn buf_size(&self) -> usize {
221        self.config.buf_size
222    }
223
224    pub fn prepare_send<R>(
225        &mut self,
226        len: usize,
227        f: impl FnOnce(&mut [u8]) -> R,
228    ) -> Result<(R, TxPending<'_>), NetError> {
229        if len > self.config.buf_size {
230            return Err(other_error("tx packet too large"));
231        }
232
233        self.reclaim_bounded(self.capacity().max(1))?;
234
235        let mut buff = self.pool.alloc()?;
236        let bus_addr = buff.dma_addr().as_u64();
237        let ret = buff.write_with_cpu(len, f);
238        Ok((
239            ret,
240            TxPending {
241                queue: self,
242                len,
243                bus_addr,
244                buff: Some(buff),
245            },
246        ))
247    }
248}
249
250pub struct TxPending<'a> {
251    queue: &'a mut TxQueue,
252    len: usize,
253    bus_addr: u64,
254    buff: Option<ContiguousBuffer>,
255}
256
257impl TxPending<'_> {
258    pub fn bus_addr(&self) -> u64 {
259        self.bus_addr
260    }
261
262    pub fn len(&self) -> usize {
263        self.len
264    }
265
266    pub fn is_empty(&self) -> bool {
267        self.len == 0
268    }
269
270    pub fn try_submit(&mut self) -> Result<(), NetError> {
271        self.queue.reclaim_bounded(self.queue.capacity().max(1))?;
272        let buff = self
273            .buff
274            .as_ref()
275            .expect("tx pending buffer should exist until submit succeeds");
276        buff.prepare_for_device(0, self.len);
277        self.queue.interface.submit(DmaBuffer {
278            virt: buff.as_ptr(),
279            bus_addr: self.bus_addr,
280            len: self.len,
281        })?;
282        let buff = self
283            .buff
284            .take()
285            .expect("tx pending buffer should exist until submit succeeds");
286        self.queue.inflight.insert(self.bus_addr, buff);
287        Ok(())
288    }
289}
290
291pub struct RxQueue {
292    interface: Box<dyn IRxQueue>,
293    pool: ContiguousBufferPool,
294    inflight: BTreeMap<u64, ContiguousBuffer>,
295    config: QueueConfig,
296    _waker: Arc<AtomicWaker>,
297}
298
299impl RxQueue {
300    fn capacity(&self) -> usize {
301        self.config.ring_size.saturating_sub(1)
302    }
303
304    fn prefill(&mut self) -> Result<(), NetError> {
305        while self.inflight.len() < self.capacity() {
306            let buff = self.pool.alloc()?;
307            if let Err(err) = self.submit_buffer(buff) {
308                if matches!(err, NetError::Retry) {
309                    break;
310                }
311                return Err(err);
312            }
313        }
314        Ok(())
315    }
316
317    fn submit_buffer(&mut self, buff: ContiguousBuffer) -> Result<(), NetError> {
318        let bus_addr = buff.dma_addr().as_u64();
319        let len = self.config.buf_size.min(buff.len());
320        buff.prepare_for_device(0, len);
321        self.interface.submit(DmaBuffer {
322            virt: buff.as_ptr(),
323            bus_addr,
324            len,
325        })?;
326        self.inflight.insert(bus_addr, buff);
327        Ok(())
328    }
329
330    fn reclaim_packet(&mut self) -> Result<Option<(ContiguousBuffer, usize)>, NetError> {
331        let Some((bus_addr, len)) = self.interface.reclaim() else {
332            return Ok(None);
333        };
334        let Some(buff) = self.inflight.remove(&bus_addr) else {
335            return Err(other_error("reclaimed unknown rx buffer"));
336        };
337        let packet_len = len.min(self.config.buf_size).min(buff.len());
338        buff.complete_for_cpu(0, packet_len);
339        Ok(Some((buff, packet_len)))
340    }
341
342    pub fn id(&self) -> usize {
343        self.interface.id()
344    }
345
346    pub fn buf_size(&self) -> usize {
347        self.config.buf_size
348    }
349
350    pub fn try_receive(&mut self) -> Option<RxPacket<'_>> {
351        match self.reclaim_packet() {
352            Ok(Some((buff, len))) => Some(RxPacket {
353                queue: self,
354                len,
355                buff: Some(buff),
356            }),
357            Ok(None) | Err(_) => None,
358        }
359    }
360
361    pub fn receive<R>(&mut self, f: impl FnOnce(&[u8]) -> R) -> Option<R> {
362        let packet = self.try_receive()?;
363        Some(packet.consume(f))
364    }
365}
366
367pub struct RxPacket<'a> {
368    queue: &'a mut RxQueue,
369    len: usize,
370    buff: Option<ContiguousBuffer>,
371}
372
373impl RxPacket<'_> {
374    pub fn len(&self) -> usize {
375        self.len
376    }
377
378    pub fn is_empty(&self) -> bool {
379        self.len == 0
380    }
381
382    pub fn consume<R>(mut self, f: impl FnOnce(&[u8]) -> R) -> R {
383        let buff = self.buff.as_ref().expect("rx packet buffer should exist");
384        let ret = buff.read_with_cpu(self.len, f);
385        if let Some(buff) = self.buff.take() {
386            let _ = self.queue.submit_buffer(buff);
387        }
388        ret
389    }
390}
391
392impl Drop for RxPacket<'_> {
393    fn drop(&mut self) {
394        if let Some(buff) = self.buff.take() {
395            let _ = self.queue.submit_buffer(buff);
396        }
397    }
398}
rd_net/lib.rs

rd_net/
lib.rs
