netmap_rs/
fallback.rs

1//! Fallback implementation for platforms without Netnap support
2
3use std::collections::VecDeque;
4use std::sync::{Arc, Mutex};
5
6use crate::error::Error;
7use crate::frame::Frame;
8
9#[derive(Clone)]
10struct SharedRing {
11    queue: Arc<Mutex<VecDeque<Vec<u8>>>>,
12    max_size: usize,
13}
14
15/// fallback implememntation for a Netmap TX ring
16pub struct FallbackTxRing(SharedRing);
17
18/// fallback implememntation for a Netmap RX ring
19pub struct FallbackRxRing(SharedRing);
20
21impl FallbackTxRing {
22    /// create new fallback TX ring
23    pub fn new(max_size: usize) -> Self {
24        Self(SharedRing {
25            queue: Arc::new(Mutex::new(VecDeque::new())),
26            max_size,
27        })
28    }
29
30    /// send a packet
31    pub fn send(&self, buf: &[u8]) -> Result<(), Error> {
32        let mut queue = self.0.queue.lock().unwrap();
33        if queue.len() >= self.0.max_size {
34            return Err(Error::WouldBlock);
35        }
36        queue.push_back(buf.to_vec());
37        Ok(())
38    }
39}
40
41impl FallbackRxRing {
42    /// create a new fallback RX ring
43    pub fn new(max_size: usize) -> Self {
44        Self(SharedRing {
45            queue: Arc::new(Mutex::new(VecDeque::new())),
46            max_size,
47        })
48    }
49
50    /// recieve a packet
51    pub fn recv(&self) -> Option<Frame<'static>> {
52        let mut queue = self.0.queue.lock().unwrap();
53        queue.pop_front().map(Frame::new_owned)
54    }
55}
56
57/// Creates a connected pair of fallback TX and RX rings.
58pub fn create_fallback_channel(max_size: usize) -> (FallbackTxRing, FallbackRxRing) {
59    let shared_ring = SharedRing {
60        queue: Arc::new(Mutex::new(VecDeque::new())),
61        max_size,
62    };
63    (
64        FallbackTxRing(shared_ring.clone()),
65        FallbackRxRing(shared_ring),
66    )
67}
68
69#[cfg(test)]
70mod tests {
71    use super::*;
72
73    #[test]
74    fn empty_ring_returns_none() {
75        let rx = FallbackRxRing::new(4);
76        assert!(rx.recv().is_none());
77    }
78
79    #[test]
80    fn send_recv_single() {
81        let tx = FallbackTxRing::new(4);
82        let rx = FallbackRxRing::new(4);
83        // Not connected; create channel instead.
84        drop(tx);
85        drop(rx);
86        let (tx, rx) = create_fallback_channel(4);
87        tx.send(b"ping").unwrap();
88        let frame = rx.recv().unwrap();
89        assert_eq!(frame.payload(), b"ping");
90    }
91
92    #[test]
93    fn send_recv_in_order() {
94        let (tx, rx) = create_fallback_channel(16);
95        for i in 0..8u8 {
96            tx.send(&[i]).unwrap();
97        }
98        for i in 0..8u8 {
99            let frame = rx.recv().unwrap();
100            assert_eq!(frame.payload(), &[i]);
101        }
102        assert!(rx.recv().is_none());
103    }
104
105    #[test]
106    fn full_ring_returns_would_block() {
107        let (tx, rx) = create_fallback_channel(2);
108        tx.send(b"a").unwrap();
109        tx.send(b"b").unwrap();
110        assert!(matches!(tx.send(b"c"), Err(Error::WouldBlock)));
111        // Drain one and retry.
112        rx.recv().unwrap();
113        assert!(tx.send(b"c").is_ok());
114    }
115
116    #[test]
117    fn oversize_buffers_are_copied() {
118        let (tx, rx) = create_fallback_channel(4);
119        let payload = vec![0xabu8; 4096];
120        tx.send(&payload).unwrap();
121        let frame = rx.recv().unwrap();
122        assert_eq!(frame.payload(), payload.as_slice());
123    }
124}
netmap_rs/fallback.rs

netmap_rs/
fallback.rs
