feat(s2n-quic-xdp): add async tasks

.github/workflows/ci.yml

@@ -666,6 +666,9 @@ jobs:
 
   kani:
     runs-on: ubuntu-latest
+    strategy:
+      matrix:
+        crate: [quic/s2n-quic-core, tools/xdp/s2n-quic-xdp]
     steps:
       - uses: actions/checkout@v3
         with:
@@ -674,7 +677,7 @@ jobs:
       - name: Kani run
         uses: model-checking/[email protected]
         with:
-          working-directory: quic/s2n-quic-core
+          working-directory: ${{ matrix.crate }}
           args: --tests
 
   dhat:

tools/xdp/s2n-quic-xdp/Cargo.toml

@@ -3,11 +3,18 @@ name = "s2n-quic-xdp"
 version = "0.1.0"
 edition = "2021"
 
+[features]
+default = ["tokio"]
+
 [dependencies]
 bitflags = "2"
 errno = "0.3"
 libc = "0.2"
 s2n-quic-core = { path = "../../../quic/s2n-quic-core", version = "0.19" }
+tokio = { version = "1", optional = true }
 
 [dev-dependencies]
 bolero = "0.9"
+rand = "0.8"
+s2n-quic-core = { path = "../../../quic/s2n-quic-core", version = "0.19", features = ["testing"] }
+tokio = { version = "1", features = ["full"] }

tools/xdp/s2n-quic-xdp/src/if_xdp.rs

@@ -292,3 +292,24 @@ pub struct UmemDescriptor {
     /// Offset into the umem where the packet starts
     pub address: u64,
 }
+
+impl UmemDescriptor {
+    /// Sets the length for the descriptor and converts it into a [`RxTxDescriptor`]
+    #[inline]
+    pub fn with_len(self, len: u32) -> RxTxDescriptor {
+        RxTxDescriptor {
+            address: self.address,
+            len,
+            options: 0,
+        }
+    }
+}
+
+impl From<RxTxDescriptor> for UmemDescriptor {
+    #[inline]
+    fn from(desc: RxTxDescriptor) -> Self {
+        Self {
+            address: desc.address,
+        }
+    }
+}

tools/xdp/s2n-quic-xdp/src/lib.rs

@@ -15,5 +15,7 @@ mod ring;
 mod socket;
 /// Helpers for making API calls to AF-XDP sockets
 mod syscall;
+/// A set of async tasks responsible for managing ring buffer and queue state
+mod task;
 /// A shared region of memory for holding frame (packet) data
 mod umem;

tools/xdp/s2n-quic-xdp/src/ring.rs

@@ -16,6 +16,10 @@ use cursor::Cursor;
 #[derive(Debug)]
 struct Ring<T: Copy + fmt::Debug> {
     cursor: Cursor<T>,
+    // make the area clonable in test mode
+    #[cfg(test)]
+    area: std::sync::Arc<Mmap>,
+    #[cfg(not(test))]
     area: Mmap,
     socket: socket::Fd,
 }
@@ -53,6 +57,9 @@ macro_rules! impl_producer {
                 cursor.init_producer();
             }
 
+            #[cfg(test)]
+            let area = std::sync::Arc::new(area);
+
             Ok(Self(Ring {
                 cursor,
                 area,
@@ -95,6 +102,12 @@ macro_rules! impl_producer {
                 self.0.cursor.producer_data()
             }
         }
+
+        /// Returns the overall size of the ring
+        #[inline]
+        pub fn capacity(&self) -> usize {
+            self.0.cursor.capacity() as _
+        }
     };
 }
 
@@ -120,6 +133,9 @@ macro_rules! impl_consumer {
                 Cursor::new(&area, offsets, size)
             };
 
+            #[cfg(test)]
+            let area = std::sync::Arc::new(area);
+
             Ok(Self(Ring {
                 cursor,
                 area,
@@ -156,31 +172,46 @@ macro_rules! impl_consumer {
                 self.0.cursor.consumer_data()
             }
         }
+
+        /// Returns the overall size of the ring
+        #[inline]
+        pub fn capacity(&self) -> usize {
+            self.0.cursor.capacity() as _
+        }
+
+        #[cfg(test)]
+        pub fn set_flags(&mut self, flags: crate::if_xdp::RingFlags) {
+            *self.0.cursor.flags_mut() = flags;
+        }
     };
 }
 
 /// A transmission ring for entries to be transmitted
+#[derive(Debug)]
 pub struct Tx(Ring<RxTxDescriptor>);
 
 impl Tx {
     impl_producer!(RxTxDescriptor, set_tx_ring_size, tx, TX_RING);
 }
 
 /// A receive ring for entries to be processed
+#[derive(Debug)]
 pub struct Rx(Ring<RxTxDescriptor>);
 
 impl Rx {
     impl_consumer!(RxTxDescriptor, set_rx_ring_size, rx, RX_RING);
 }
 
 /// The fill ring for entries to be populated
+#[derive(Debug)]
 pub struct Fill(Ring<UmemDescriptor>);
 
 impl Fill {
     impl_producer!(UmemDescriptor, set_fill_ring_size, fill, FILL_RING);
 }
 
 /// The completion ring for entries to be reused for transmission
+#[derive(Debug)]
 pub struct Completion(Ring<UmemDescriptor>);
 
 impl Completion {
@@ -191,3 +222,80 @@ impl Completion {
         COMPLETION_RING
     );
 }
+
+#[cfg(test)]
+pub mod testing {
+    use super::*;
+    use crate::{if_xdp, socket::Fd};
+
+    fn offsets() -> if_xdp::RingOffsetV2 {
+        if_xdp::RingOffsetV2 {
+            producer: 0,
+            consumer: core::mem::size_of::<usize>() as _,
+            flags: (core::mem::size_of::<usize>() * 2) as _,
+            desc: (core::mem::size_of::<usize>() * 3) as _,
+        }
+    }
+
+    macro_rules! impl_pair {
+        ($name:ident, $consumer:ident, $producer:ident, $T:ident) => {
+            /// Creates a pair of rings used for testing
+            pub fn $name(size: u32) -> ($consumer, $producer) {
+                assert!(size.is_power_of_two());
+
+                let offsets = offsets();
+
+                // start with the descriptor offset as the total length
+                let mut len = offsets.desc as usize;
+                // extend the length by the `size` multiplied the entry size
+                len += size as usize * size_of::<$T>();
+
+                let area = Mmap::new(len, 0, None).unwrap();
+
+                let consumer_cursor = unsafe {
+                    // Safety: `area` lives as long as `cursor`
+                    Cursor::new(&area, &offsets, size)
+                };
+
+                let mut producer_cursor = unsafe {
+                    // Safety: `area` lives as long as `cursor`
+                    Cursor::new(&area, &offsets, size)
+                };
+
+                unsafe {
+                    // Safety: this is only called by a producer
+                    producer_cursor.init_producer();
+                }
+
+                let area = std::sync::Arc::new(area);
+
+                let cons = $consumer(Ring {
+                    cursor: consumer_cursor,
+                    area: area.clone(),
+                    socket: Fd::invalid(),
+                });
+
+                let prod = $producer(Ring {
+                    cursor: producer_cursor,
+                    area,
+                    socket: Fd::invalid(),
+                });
+
+                (cons, prod)
+            }
+        };
+    }
+
+    impl_pair!(rx_tx, Rx, Tx, RxTxDescriptor);
+    impl_pair!(completion_fill, Completion, Fill, UmemDescriptor);
+
+    #[test]
+    fn rx_tx_test() {
+        let _ = rx_tx(16);
+    }
+
+    #[test]
+    fn comp_fill_test() {
+        let _ = completion_fill(16);
+    }
+}