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 enable_irq(&mut self) {
110        self.interface().enable_irq();
111    }
112
113    pub fn disable_irq(&mut self) {
114        self.interface().disable_irq();
115    }
116
117    pub fn is_irq_enabled(&self) -> bool {
118        self.interface().is_irq_enabled()
119    }
120
121    pub fn create_tx_queue(&mut self) -> Result<TxQueue, NetError> {
122        let irq_guard = self.irq_guard();
123        let queue = self
124            .interface()
125            .create_tx_queue()
126            .ok_or_else(|| other_error("failed to create tx queue"))?;
127        let config = queue.config();
128        let pool = make_pool(self.inner.dma_op, config, DmaDirection::ToDevice)?;
129        let waker = self.inner.tx_wakers.register(queue.id());
130        drop(irq_guard);
131
132        Ok(TxQueue {
133            interface: queue,
134            pool,
135            inflight: BTreeMap::new(),
136            config,
137            _waker: waker,
138        })
139    }
140
141    pub fn create_rx_queue(&mut self) -> Result<RxQueue, NetError> {
142        let irq_guard = self.irq_guard();
143        let queue = self
144            .interface()
145            .create_rx_queue()
146            .ok_or_else(|| other_error("failed to create rx queue"))?;
147        let config = queue.config();
148        let pool = make_pool(self.inner.dma_op, config, DmaDirection::FromDevice)?;
149        let waker = self.inner.rx_wakers.register(queue.id());
150        drop(irq_guard);
151
152        let mut rx = RxQueue {
153            interface: queue,
154            pool,
155            inflight: BTreeMap::new(),
156            config,
157            _waker: waker,
158        };
159        rx.prefill()?;
160        Ok(rx)
161    }
162
163    pub fn irq_handler(&self) -> IrqHandler {
164        IrqHandler {
165            inner: self.inner.clone(),
166        }
167    }
168}
169
170fn make_pool(
171    dma_op: &'static dyn DmaOp,
172    config: QueueConfig,
173    direction: DmaDirection,
174) -> Result<ContiguousBufferPool, NetError> {
175    let layout = Layout::from_size_align(config.buf_size, config.align.max(1))
176        .map_err(|_| other_error("invalid queue layout"))?;
177    let dma = DeviceDma::new(config.dma_mask, dma_op);
178    Ok(dma.contiguous_buffer_pool(layout, direction, config.ring_size))
179}
180
181pub struct IrqHandler {
182    inner: Arc<NetInner>,
183}
184
185unsafe impl Sync for IrqHandler {}
186
187impl IrqHandler {
188    pub fn enable(&self) {
189        let iface = unsafe { &mut **self.inner.interface.get() };
190        iface.enable_irq();
191    }
192
193    pub fn disable(&self) {
194        let iface = unsafe { &mut **self.inner.interface.get() };
195        iface.disable_irq();
196    }
197
198    pub fn handle(&self) {
199        let iface = unsafe { &mut **self.inner.interface.get() };
200        let event = iface.handle_irq();
201        for id in event.tx_queue.iter() {
202            self.inner.tx_wakers.wake(id);
203        }
204        for id in event.rx_queue.iter() {
205            self.inner.rx_wakers.wake(id);
206        }
207    }
208}
209
210pub struct TxQueue {
211    interface: Box<dyn ITxQueue>,
212    pool: ContiguousBufferPool,
213    inflight: BTreeMap<u64, ContiguousBuffer>,
214    config: QueueConfig,
215    _waker: Arc<AtomicWaker>,
216}
217
218impl TxQueue {
219    fn capacity(&self) -> usize {
220        self.config.ring_size.saturating_sub(1)
221    }
222
223    fn reclaim_bounded(&mut self, limit: usize) -> Result<usize, NetError> {
224        let mut reclaimed = 0;
225        while reclaimed < limit {
226            let Some(bus_addr) = self.interface.reclaim() else {
227                break;
228            };
229            let Some(buff) = self.inflight.remove(&bus_addr) else {
230                return Err(other_error("reclaimed unknown tx buffer"));
231            };
232            drop(buff);
233            reclaimed += 1;
234        }
235        Ok(reclaimed)
236    }
237
238    pub fn id(&self) -> usize {
239        self.interface.id()
240    }
241
242    pub fn buf_size(&self) -> usize {
243        self.config.buf_size
244    }
245
246    pub fn prepare_send<R>(
247        &mut self,
248        len: usize,
249        f: impl FnOnce(&mut [u8]) -> R,
250    ) -> Result<(R, TxPending<'_>), NetError> {
251        if len > self.config.buf_size {
252            return Err(other_error("tx packet too large"));
253        }
254
255        self.reclaim_bounded(self.capacity().max(1))?;
256
257        let mut buff = self.pool.alloc()?;
258        let bus_addr = buff.dma_addr().as_u64();
259        let ret = buff.write_with_cpu(len, f);
260        Ok((
261            ret,
262            TxPending {
263                queue: self,
264                len,
265                bus_addr,
266                buff: Some(buff),
267            },
268        ))
269    }
270}
271
272pub struct TxPending<'a> {
273    queue: &'a mut TxQueue,
274    len: usize,
275    bus_addr: u64,
276    buff: Option<ContiguousBuffer>,
277}
278
279impl TxPending<'_> {
280    pub fn bus_addr(&self) -> u64 {
281        self.bus_addr
282    }
283
284    pub fn len(&self) -> usize {
285        self.len
286    }
287
288    pub fn is_empty(&self) -> bool {
289        self.len == 0
290    }
291
292    pub fn try_submit(&mut self) -> Result<(), NetError> {
293        self.queue.reclaim_bounded(self.queue.capacity().max(1))?;
294        let buff = self
295            .buff
296            .as_ref()
297            .expect("tx pending buffer should exist until submit succeeds");
298        buff.prepare_for_device(0, self.len);
299        self.queue.interface.submit(DmaBuffer {
300            virt: buff.as_ptr(),
301            bus_addr: self.bus_addr,
302            len: self.len,
303        })?;
304        let buff = self
305            .buff
306            .take()
307            .expect("tx pending buffer should exist until submit succeeds");
308        self.queue.inflight.insert(self.bus_addr, buff);
309        Ok(())
310    }
311}
312
313pub struct RxQueue {
314    interface: Box<dyn IRxQueue>,
315    pool: ContiguousBufferPool,
316    inflight: BTreeMap<u64, ContiguousBuffer>,
317    config: QueueConfig,
318    _waker: Arc<AtomicWaker>,
319}
320
321impl RxQueue {
322    fn capacity(&self) -> usize {
323        self.config.ring_size.saturating_sub(1)
324    }
325
326    fn prefill(&mut self) -> Result<(), NetError> {
327        while self.inflight.len() < self.capacity() {
328            let buff = self.pool.alloc()?;
329            if let Err(err) = self.submit_buffer(buff) {
330                if matches!(err, NetError::Retry) {
331                    break;
332                }
333                return Err(err);
334            }
335        }
336        Ok(())
337    }
338
339    fn submit_buffer(&mut self, buff: ContiguousBuffer) -> Result<(), NetError> {
340        let bus_addr = buff.dma_addr().as_u64();
341        let len = self.config.buf_size.min(buff.len());
342        buff.prepare_for_device(0, len);
343        self.interface.submit(DmaBuffer {
344            virt: buff.as_ptr(),
345            bus_addr,
346            len,
347        })?;
348        self.inflight.insert(bus_addr, buff);
349        Ok(())
350    }
351
352    fn reclaim_packet(&mut self) -> Result<Option<(ContiguousBuffer, usize)>, NetError> {
353        let Some((bus_addr, len)) = self.interface.reclaim() else {
354            return Ok(None);
355        };
356        let Some(buff) = self.inflight.remove(&bus_addr) else {
357            return Err(other_error("reclaimed unknown rx buffer"));
358        };
359        let packet_len = len.min(self.config.buf_size).min(buff.len());
360        buff.complete_for_cpu(0, packet_len);
361        Ok(Some((buff, packet_len)))
362    }
363
364    pub fn id(&self) -> usize {
365        self.interface.id()
366    }
367
368    pub fn buf_size(&self) -> usize {
369        self.config.buf_size
370    }
371
372    pub fn try_receive(&mut self) -> Option<RxPacket<'_>> {
373        match self.reclaim_packet() {
374            Ok(Some((buff, len))) => Some(RxPacket {
375                queue: self,
376                len,
377                buff: Some(buff),
378            }),
379            Ok(None) | Err(_) => None,
380        }
381    }
382
383    pub fn receive<R>(&mut self, f: impl FnOnce(&[u8]) -> R) -> Option<R> {
384        let packet = self.try_receive()?;
385        Some(packet.consume(f))
386    }
387}
388
389pub struct RxPacket<'a> {
390    queue: &'a mut RxQueue,
391    len: usize,
392    buff: Option<ContiguousBuffer>,
393}
394
395impl RxPacket<'_> {
396    pub fn len(&self) -> usize {
397        self.len
398    }
399
400    pub fn is_empty(&self) -> bool {
401        self.len == 0
402    }
403
404    pub fn consume<R>(mut self, f: impl FnOnce(&[u8]) -> R) -> R {
405        let buff = self.buff.as_ref().expect("rx packet buffer should exist");
406        let ret = buff.read_with_cpu(self.len, f);
407        if let Some(buff) = self.buff.take() {
408            let _ = self.queue.submit_buffer(buff);
409        }
410        ret
411    }
412}
413
414impl Drop for RxPacket<'_> {
415    fn drop(&mut self) {
416        if let Some(buff) = self.buff.take() {
417            let _ = self.queue.submit_buffer(buff);
418        }
419    }
420}
rd_net/lib.rs

rd_net/
lib.rs
