yip_io/
af_xdp.rs

1//! AF_XDP (XSK) kernel-bypass driver and UMEM shared memory pool.
2//!
3//! Provides zero-copy memory buffers (`UmemPool`) and ring buffer descriptors
4//! (`FillRing`, `CompletionRing`) for kernel-bypass packet I/O.
5//!
6//! Low-level libc XSK/XDP structures and unsafe memory mappings are quarantined
7//! within this module, with explicit `// SAFETY:` justifications on all unsafe blocks.
8
9use std::ffi::CString;
10use std::io;
11use std::os::fd::RawFd;
12use std::sync::Arc;
13
14/// Default UMEM chunk size in bytes (2 KiB aligns with MTU 1500 + wire overhead).
15pub const UMEM_CHUNK_SIZE: usize = 2048;
16
17/// Default ring size in descriptors (must be a power of two).
18pub const UMEM_RING_SIZE: u32 = 2048;
19
20/// Configuration parameters for allocating an AF_XDP [`UmemPool`].
21#[derive(Debug, Clone, Copy, PartialEq, Eq)]
22pub struct UmemPoolConfig {
23    /// Number of chunk buffers in the pool.
24    pub num_chunks: usize,
25    /// Size of each chunk in bytes (typically 2048 or 4096).
26    pub chunk_size: usize,
27    /// Pre-fault all memory pages during allocation via `MAP_POPULATE`.
28    ///
29    /// When `true` (the default), eliminates page fault jitter during packet I/O
30    /// at the expense of higher startup latency and immediate RSS allocation.
31    /// When `false`, pages are faulted lazily on-demand, reducing startup time
32    /// and memory footprint on resource-constrained hosts.
33    pub pre_fault: bool,
34    /// Pin pages in physical RAM via `libc::mlock` to prevent swapping.
35    ///
36    /// Requires `CAP_IPC_LOCK` or sufficient `RLIMIT_MEMLOCK`. If `true` and `mlock`
37    /// fails, an error is returned.
38    pub lock_memory: bool,
39}
40
41impl Default for UmemPoolConfig {
42    fn default() -> Self {
43        Self {
44            num_chunks: 4096,
45            chunk_size: 2048,
46            pre_fault: true,
47            lock_memory: false,
48        }
49    }
50}
51
52/// Pre-allocated page-aligned shared memory pool for AF_XDP kernel-bypass I/O.
53pub struct UmemPool {
54    area: *mut libc::c_void,
55    size: usize,
56    chunk_size: usize,
57    free_chunks: Vec<u64>,
58}
59
60impl UmemPool {
61    /// Allocates an anonymous, page-aligned shared memory region for `num_chunks` chunks
62    /// of size `chunk_size` with default pre-faulting (`MAP_POPULATE`) enabled.
63    pub fn new(num_chunks: usize, chunk_size: usize) -> io::Result<Self> {
64        Self::with_config(UmemPoolConfig {
65            num_chunks,
66            chunk_size,
67            pre_fault: true,
68            lock_memory: false,
69        })
70    }
71
72    /// Allocates a new `UmemPool` with lazy on-demand page allocation (`MAP_POPULATE` disabled),
73    /// minimizing startup latency and initial RSS footprint.
74    pub fn new_lazy(num_chunks: usize, chunk_size: usize) -> io::Result<Self> {
75        Self::with_config(UmemPoolConfig {
76            num_chunks,
77            chunk_size,
78            pre_fault: false,
79            lock_memory: false,
80        })
81    }
82
83    /// Allocates a new `UmemPool` according to the provided [`UmemPoolConfig`].
84    pub fn with_config(config: UmemPoolConfig) -> io::Result<Self> {
85        if config.num_chunks == 0 || config.chunk_size == 0 {
86            return Err(io::Error::new(
87                io::ErrorKind::InvalidInput,
88                "num_chunks and chunk_size must be positive",
89            ));
90        }
91
92        let size = config
93            .num_chunks
94            .checked_mul(config.chunk_size)
95            .ok_or_else(|| io::Error::new(io::ErrorKind::InvalidInput, "UMEM size overflow"))?;
96
97        let mut flags = libc::MAP_ANONYMOUS | libc::MAP_SHARED;
98        if config.pre_fault {
99            flags |= libc::MAP_POPULATE;
100        }
101
102        // SAFETY: We request an anonymous, shared, page-aligned memory mapping of `size` bytes.
103        // The address is NULL so the kernel selects the starting address.
104        // We verify that the returned pointer is not `MAP_FAILED` before proceeding.
105        let area = unsafe {
106            libc::mmap(
107                std::ptr::null_mut(),
108                size,
109                libc::PROT_READ | libc::PROT_WRITE,
110                flags,
111                -1,
112                0,
113            )
114        };
115
116        if area == libc::MAP_FAILED {
117            return Err(io::Error::last_os_error());
118        }
119
120        if config.lock_memory {
121            // SAFETY: `area` points to `size` bytes of freshly mapped valid memory.
122            let lock_ret = unsafe { libc::mlock(area, size) };
123            if lock_ret != 0 {
124                let err = io::Error::last_os_error();
125                // SAFETY: On mlock failure, unmap the allocated area before returning error.
126                unsafe {
127                    libc::munmap(area, size);
128                }
129                return Err(err);
130            }
131        }
132
133        // Populates free_chunks so popping yields 0, chunk_size, 2*chunk_size, ...
134        let free_chunks: Vec<u64> = (0..config.num_chunks)
135            .rev()
136            .map(|i| (i * config.chunk_size) as u64)
137            .collect();
138
139        Ok(Self {
140            area,
141            size,
142            chunk_size: config.chunk_size,
143            free_chunks,
144        })
145    }
146
147    /// Allocates a chunk address from the free pool, returning `None` if exhausted.
148    pub fn alloc_chunk(&mut self) -> Option<u64> {
149        self.free_chunks.pop()
150    }
151
152    /// Returns a chunk address to the free pool.
153    pub fn free_chunk(&mut self, addr: u64) {
154        debug_assert!(
155            (addr as usize) < self.size && addr.is_multiple_of(self.chunk_size as u64),
156            "corrupt chunk address returned to free pool: addr={addr}, size={}, chunk_size={}",
157            self.size,
158            self.chunk_size
159        );
160        self.free_chunks.push(addr);
161    }
162
163    /// Returns a bounds-checked immutable slice into the chunk starting at `addr`.
164    pub fn chunk_slice(&self, addr: u64, len: usize) -> &[u8] {
165        assert!(
166            (addr as usize)
167                .checked_add(len)
168                .is_some_and(|end| end <= self.size),
169            "UMEM slice out of bounds: addr={addr}, len={len}, total size={}",
170            self.size
171        );
172        let offset = (addr as usize) % self.chunk_size;
173        assert!(
174            offset
175                .checked_add(len)
176                .is_some_and(|end| end <= self.chunk_size),
177            "UMEM slice len={len} with intra-chunk offset={offset} exceeds chunk_size={}",
178            self.chunk_size
179        );
180
181        // SAFETY: `self.area` is valid mapped memory of `self.size` bytes. We verified that
182        // `addr + len <= self.size`, so `self.area.add(addr as usize)` points to at least `len`
183        // contiguous initialized bytes within the mapped region. The lifetime is tied to `&self`,
184        // ensuring the memory remains mapped during the slice's lifetime.
185        unsafe {
186            let ptr = (self.area as *const u8).add(addr as usize);
187            std::slice::from_raw_parts(ptr, len)
188        }
189    }
190
191    /// Returns a bounds-checked mutable slice into the chunk starting at `addr`.
192    pub fn chunk_slice_mut(&mut self, addr: u64, len: usize) -> &mut [u8] {
193        assert!(
194            (addr as usize)
195                .checked_add(len)
196                .is_some_and(|end| end <= self.size),
197            "UMEM slice out of bounds: addr={addr}, len={len}, total size={}",
198            self.size
199        );
200        let offset = (addr as usize) % self.chunk_size;
201        assert!(
202            offset
203                .checked_add(len)
204                .is_some_and(|end| end <= self.chunk_size),
205            "UMEM slice len={len} with intra-chunk offset={offset} exceeds chunk_size={}",
206            self.chunk_size
207        );
208
209        // SAFETY: `self.area` is valid mapped memory of `self.size` bytes. We verified that
210        // `addr + len <= self.size`, so `self.area.add(addr as usize)` points to at least `len`
211        // contiguous initialized bytes within the mapped region. The exclusive lifetime `&mut self`
212        // guarantees that no other references to this memory exist during the returned slice's lifetime.
213        unsafe {
214            let ptr = (self.area as *mut u8).add(addr as usize);
215            std::slice::from_raw_parts_mut(ptr, len)
216        }
217    }
218
219    /// Raw pointer to the mapped UMEM memory area.
220    pub fn area_ptr(&self) -> *mut libc::c_void {
221        self.area
222    }
223
224    /// Total size in bytes of the mapped UMEM memory area.
225    pub fn size(&self) -> usize {
226        self.size
227    }
228
229    /// Chunk size in bytes.
230    pub fn chunk_size(&self) -> usize {
231        self.chunk_size
232    }
233
234    /// Number of free chunks currently available in the pool.
235    pub fn free_chunk_count(&self) -> usize {
236        self.free_chunks.len()
237    }
238
239    /// Returns `true` if `addr` is an aligned chunk address within this pool's bounds.
240    #[inline]
241    pub fn is_valid_chunk_addr(&self, addr: u64) -> bool {
242        (addr as usize) < self.size && addr.is_multiple_of(self.chunk_size as u64)
243    }
244
245    /// Unsafe direct access to a mutable raw pointer for a chunk.
246    ///
247    /// # Safety
248    /// Caller must ensure that no other thread or reference is concurrently accessing
249    /// the memory range `[addr, addr + len)`.
250    pub(crate) unsafe fn chunk_ptr_mut(&self, addr: u64, len: usize) -> *mut u8 {
251        assert!(
252            (addr as usize)
253                .checked_add(len)
254                .is_some_and(|end| end <= self.size),
255            "UMEM slice out of bounds: addr={addr}, len={len}, total size={}",
256            self.size
257        );
258        let offset = (addr as usize) % self.chunk_size;
259        assert!(
260            offset
261                .checked_add(len)
262                .is_some_and(|end| end <= self.chunk_size),
263            "UMEM slice len={len} with intra-chunk offset={offset} exceeds chunk_size={}",
264            self.chunk_size
265        );
266
267        // SAFETY: The caller guarantees exclusive access to this slice. `self.area` is valid
268        // mapped memory of `self.size` bytes, and we verified that `addr + len <= self.size`.
269        unsafe { (self.area as *mut u8).add(addr as usize) }
270    }
271
272    /// Partitions an `Arc<UmemPool>` into `num_slabs` disjoint, per-core chunk slabs.
273    ///
274    /// Each slab receives an equal partition of chunks (`total_chunks / num_slabs`).
275    /// Slabs operate completely independently without shared locks or cross-core
276    /// cache line contention.
277    pub fn create_slabs(self: &Arc<Self>, num_slabs: usize) -> io::Result<Vec<UmemSlab>> {
278        if num_slabs == 0 {
279            return Err(io::Error::new(
280                io::ErrorKind::InvalidInput,
281                "num_slabs must be positive",
282            ));
283        }
284
285        let total_chunks = self.size / self.chunk_size;
286        let chunks_per_slab = total_chunks / num_slabs;
287        if chunks_per_slab == 0 {
288            return Err(io::Error::new(
289                io::ErrorKind::InvalidInput,
290                format!("not enough chunks ({total_chunks}) for requested slabs ({num_slabs})"),
291            ));
292        }
293
294        let mut slabs = Vec::with_capacity(num_slabs);
295        for slab_id in 0..num_slabs {
296            let base_addr = (slab_id * chunks_per_slab * self.chunk_size) as u64;
297            // Pre-allocate free chunks so popping yields base_addr, base_addr + chunk_size, ...
298            let free_chunks: Vec<u64> = (0..chunks_per_slab)
299                .rev()
300                .map(|i| base_addr + (i * self.chunk_size) as u64)
301                .collect();
302
303            slabs.push(UmemSlab {
304                pool: Arc::clone(self),
305                slab_id,
306                base_addr,
307                num_chunks: chunks_per_slab,
308                chunk_size: self.chunk_size,
309                free_chunks,
310            });
311        }
312
313        Ok(slabs)
314    }
315
316    /// Allocates a new `UmemPool` with default configuration and partitions it into `num_slabs` disjoint slabs.
317    pub fn new_slabs(
318        num_chunks: usize,
319        chunk_size: usize,
320        num_slabs: usize,
321    ) -> io::Result<(Arc<Self>, Vec<UmemSlab>)> {
322        Self::new_slabs_with_config(
323            UmemPoolConfig {
324                num_chunks,
325                chunk_size,
326                pre_fault: true,
327                lock_memory: false,
328            },
329            num_slabs,
330        )
331    }
332
333    /// Allocates a new `UmemPool` using the specified [`UmemPoolConfig`] and partitions it into `num_slabs` disjoint slabs.
334    pub fn new_slabs_with_config(
335        config: UmemPoolConfig,
336        num_slabs: usize,
337    ) -> io::Result<(Arc<Self>, Vec<UmemSlab>)> {
338        let pool = Arc::new(Self::with_config(config)?);
339        let slabs = pool.create_slabs(num_slabs)?;
340        Ok((pool, slabs))
341    }
342}
343
344/// Dedicated per-core / per-queue UMEM chunk slab for lock-free AF_XDP multi-worker scaling.
345///
346/// Each `UmemSlab` manages an isolated, non-overlapping range of UMEM chunk addresses
347/// allocated from a parent `UmemPool`. Worker threads allocate and recycle chunks
348/// locally from their dedicated slab without cross-core coordination, atomic locks,
349/// or MESI cache-line invalidation.
350pub struct UmemSlab {
351    pool: Arc<UmemPool>,
352    slab_id: usize,
353    base_addr: u64,
354    num_chunks: usize,
355    chunk_size: usize,
356    free_chunks: Vec<u64>,
357}
358
359impl UmemSlab {
360    /// Slab identifier (e.g. core ID or queue ID).
361    #[inline]
362    pub fn slab_id(&self) -> usize {
363        self.slab_id
364    }
365
366    /// Base memory address offset of this slab within the parent UMEM area.
367    #[inline]
368    pub fn base_addr(&self) -> u64 {
369        self.base_addr
370    }
371
372    /// Upper non-inclusive address boundary of this slab within the parent UMEM area.
373    #[inline]
374    pub fn end_addr(&self) -> u64 {
375        self.base_addr + (self.num_chunks * self.chunk_size) as u64
376    }
377
378    /// Total number of chunks assigned to this slab.
379    #[inline]
380    pub fn num_chunks(&self) -> usize {
381        self.num_chunks
382    }
383
384    /// Total number of chunks assigned to this slab (alias for `num_chunks`).
385    #[inline]
386    pub fn total_chunks(&self) -> usize {
387        self.num_chunks
388    }
389
390    /// Number of free chunks currently available in this slab.
391    #[inline]
392    pub fn free_chunk_count(&self) -> usize {
393        self.free_chunks.len()
394    }
395
396    /// Chunk size in bytes.
397    #[inline]
398    pub fn chunk_size(&self) -> usize {
399        self.chunk_size
400    }
401
402    /// Reference to the parent `UmemPool`.
403    #[inline]
404    pub fn pool(&self) -> &Arc<UmemPool> {
405        &self.pool
406    }
407
408    /// Returns `true` if `addr` is an aligned chunk address belonging to this slab.
409    #[inline]
410    pub fn contains_addr(&self, addr: u64) -> bool {
411        addr >= self.base_addr
412            && addr < self.end_addr()
413            && (addr - self.base_addr).is_multiple_of(self.chunk_size as u64)
414    }
415
416    /// Returns `true` if `addr..addr+len` lies completely within this slab's memory range.
417    #[inline]
418    pub fn contains_addr_range(&self, addr: u64, len: usize) -> bool {
419        addr >= self.base_addr
420            && (addr as usize)
421                .checked_add(len)
422                .is_some_and(|end| end as u64 <= self.end_addr())
423    }
424
425    /// Allocates a chunk address from this slab, returning `None` if exhausted.
426    #[inline]
427    pub fn alloc_chunk(&mut self) -> Option<u64> {
428        self.free_chunks.pop()
429    }
430
431    /// Returns a chunk address to this slab's free pool.
432    #[inline]
433    pub fn free_chunk(&mut self, addr: u64) {
434        debug_assert!(
435            self.contains_addr(addr),
436            "corrupt chunk address returned to slab {}: addr={addr}, base={}, end={}, chunk_size={}",
437            self.slab_id,
438            self.base_addr,
439            self.end_addr(),
440            self.chunk_size
441        );
442        self.free_chunks.push(addr);
443    }
444
445    /// Allocates up to `out.len()` chunk addresses into `out`. Returns count allocated.
446    pub fn alloc_batch(&mut self, out: &mut [u64]) -> usize {
447        let count = out.len().min(self.free_chunks.len());
448        if count == 0 {
449            return 0;
450        }
451        let start = self.free_chunks.len() - count;
452        for (i, slot) in out.iter_mut().take(count).enumerate() {
453            *slot = self.free_chunks[start + count - 1 - i];
454        }
455        self.free_chunks.truncate(start);
456        count
457    }
458
459    /// Frees a slice of chunk addresses back to this slab.
460    pub fn free_batch(&mut self, addrs: &[u64]) {
461        for &addr in addrs {
462            debug_assert!(
463                self.contains_addr(addr),
464                "corrupt chunk address in batch for slab {}: addr={addr}",
465                self.slab_id
466            );
467        }
468        self.free_chunks.extend_from_slice(addrs);
469    }
470
471    /// Pre-populates a `FillRing` with chunks from this slab up to `max_count`.
472    ///
473    /// Allocates in batches of 64 chunks to maximize memory pipeline throughput.
474    /// Returns the total number of chunks enqueued into the fill ring.
475    pub fn populate_fill_ring(&mut self, fill_ring: &mut FillRing, max_count: usize) -> usize {
476        let mut total = 0;
477        let mut buf = [0u64; 64];
478        while total < max_count && !self.free_chunks.is_empty() && !fill_ring.is_full() {
479            let to_alloc = (max_count - total).min(buf.len());
480            let allocated = self.alloc_batch(&mut buf[..to_alloc]);
481            if allocated == 0 {
482                break;
483            }
484            let produced = fill_ring.produce_batch(&buf[..allocated]);
485            total += produced;
486            if produced < allocated {
487                self.free_batch(&buf[produced..allocated]);
488                break;
489            }
490        }
491        total
492    }
493
494    /// Recycles completed chunk descriptors from `CompletionRing` directly back into `FillRing`.
495    ///
496    /// Consumes up to `max_batch` descriptors from `comp_ring` and attempts to produce them
497    /// directly into `fill_ring`. Any surplus chunks that do not fit into `fill_ring` are
498    /// returned to this slab's free pool without heap allocations.
499    ///
500    /// Returns the total number of completed descriptors consumed from `comp_ring`.
501    pub fn recycle_from_rings(
502        &mut self,
503        comp_ring: &mut CompletionRing,
504        fill_ring: &mut FillRing,
505        max_batch: usize,
506    ) -> usize {
507        let mut buf = [0u64; 64];
508        let batch_size = max_batch.min(buf.len());
509        let consumed = comp_ring.consume_batch(&mut buf[..batch_size]);
510        if consumed == 0 {
511            return 0;
512        }
513
514        let produced = fill_ring.produce_batch(&buf[..consumed]);
515        if produced < consumed {
516            self.free_batch(&buf[produced..consumed]);
517        }
518
519        consumed
520    }
521
522    /// Returns a bounds-checked immutable slice into the chunk starting at `addr`.
523    pub fn chunk_slice(&self, addr: u64, len: usize) -> &[u8] {
524        assert!(
525            self.contains_addr_range(addr, len),
526            "UMEM slab {}: addr={addr}, len={len} out of slab bounds [{}..{})",
527            self.slab_id,
528            self.base_addr,
529            self.end_addr()
530        );
531        self.pool.chunk_slice(addr, len)
532    }
533
534    /// Returns a bounds-checked mutable slice into the chunk starting at `addr`.
535    pub fn chunk_slice_mut(&mut self, addr: u64, len: usize) -> &mut [u8] {
536        assert!(
537            self.contains_addr_range(addr, len),
538            "UMEM slab {}: addr={addr}, len={len} out of slab bounds [{}..{})",
539            self.slab_id,
540            self.base_addr,
541            self.end_addr()
542        );
543
544        // SAFETY: Each `UmemSlab` owns an isolated, non-overlapping range of the UMEM memory area.
545        // We verified that `addr..addr+len` lies strictly within this slab's exclusive address range.
546        // The `&mut self` borrow guarantees that no other code is currently accessing this slab's memory.
547        unsafe {
548            let ptr = self.pool.chunk_ptr_mut(addr, len);
549            std::slice::from_raw_parts_mut(ptr, len)
550        }
551    }
552}
553
554impl Drop for UmemPool {
555    fn drop(&mut self) {
556        if !self.area.is_null() && self.area != libc::MAP_FAILED {
557            // SAFETY: `self.area` was mapped by `mmap` with size `self.size` in `new()`,
558            // and has not been unmapped yet.
559            unsafe {
560                libc::munmap(self.area, self.size);
561            }
562        }
563    }
564}
565
566// SAFETY: `UmemPool` exclusively owns the mapped memory region `self.area` of `self.size`
567// bytes. Transferring ownership of `UmemPool` across threads transfers ownership of the
568// entire memory pool safely without aliasing.
569unsafe impl Send for UmemPool {}
570
571// SAFETY: All methods taking `&self` only provide immutable, bounds-checked access to the
572// mapped memory region and immutable fields. Any mutable access requires `&mut self`
573// (such as `chunk_slice_mut`, `alloc_chunk`, `free_chunk`), which Rust's borrow checker
574// guarantees to be exclusive.
575unsafe impl Sync for UmemPool {}
576
577/// Ring buffer for UMEM fill descriptors (chunk addresses pushed by userspace for the kernel/NIC to fill).
578#[derive(Debug, Clone)]
579pub struct FillRing {
580    entries: Vec<u64>,
581    capacity: u32,
582    head: u32,
583    tail: u32,
584}
585
586impl FillRing {
587    /// Creates a new fill ring with `capacity` descriptors. `capacity` must be a power of two.
588    pub fn new(capacity: u32) -> Self {
589        assert!(
590            capacity > 0 && capacity.is_power_of_two(),
591            "FillRing capacity must be a power of two, got {capacity}"
592        );
593        Self {
594            entries: vec![0u64; capacity as usize],
595            capacity,
596            head: 0,
597            tail: 0,
598        }
599    }
600
601    /// Maximum number of descriptors the ring can hold.
602    pub fn capacity(&self) -> u32 {
603        self.capacity
604    }
605
606    /// Number of descriptors currently queued in the ring.
607    pub fn len(&self) -> usize {
608        self.tail.wrapping_sub(self.head) as usize
609    }
610
611    /// Returns `true` if the ring has no queued descriptors.
612    pub fn is_empty(&self) -> bool {
613        self.head == self.tail
614    }
615
616    /// Returns `true` if the ring is at maximum capacity.
617    pub fn is_full(&self) -> bool {
618        self.len() >= self.capacity as usize
619    }
620
621    /// Enqueues a chunk address for the kernel/NIC to fill. Returns `false` if full.
622    pub fn produce(&mut self, addr: u64) -> bool {
623        if self.is_full() {
624            return false;
625        }
626        let idx = (self.tail as usize) & ((self.capacity - 1) as usize);
627        self.entries[idx] = addr;
628        self.tail = self.tail.wrapping_add(1);
629        true
630    }
631
632    /// Enqueues a batch of chunk addresses into the ring. Returns count enqueued.
633    pub fn produce_batch(&mut self, addrs: &[u64]) -> usize {
634        let available = (self.capacity as usize).saturating_sub(self.len());
635        let to_produce = addrs.len().min(available);
636        if to_produce == 0 {
637            return 0;
638        }
639
640        let mask = (self.capacity - 1) as usize;
641        let start = (self.tail as usize) & mask;
642        let first_len = to_produce.min(self.entries.len() - start);
643        self.entries[start..start + first_len].copy_from_slice(&addrs[..first_len]);
644
645        if first_len < to_produce {
646            let second_len = to_produce - first_len;
647            self.entries[..second_len].copy_from_slice(&addrs[first_len..to_produce]);
648        }
649
650        self.tail = self.tail.wrapping_add(to_produce as u32);
651        to_produce
652    }
653
654    /// Dequeues a chunk address from the ring.
655    pub fn consume(&mut self) -> Option<u64> {
656        if self.is_empty() {
657            return None;
658        }
659        let idx = (self.head as usize) & ((self.capacity - 1) as usize);
660        let addr = self.entries[idx];
661        self.head = self.head.wrapping_add(1);
662        Some(addr)
663    }
664
665    /// Dequeues up to `out.len()` chunk addresses into `out`. Returns count dequeued.
666    pub fn consume_batch(&mut self, out: &mut [u64]) -> usize {
667        let available = self.len();
668        let to_consume = out.len().min(available);
669        if to_consume == 0 {
670            return 0;
671        }
672
673        let mask = (self.capacity - 1) as usize;
674        let start = (self.head as usize) & mask;
675        let first_len = to_consume.min(self.entries.len() - start);
676        out[..first_len].copy_from_slice(&self.entries[start..start + first_len]);
677
678        if first_len < to_consume {
679            let second_len = to_consume - first_len;
680            out[first_len..to_consume].copy_from_slice(&self.entries[..second_len]);
681        }
682
683        self.head = self.head.wrapping_add(to_consume as u32);
684        to_consume
685    }
686}
687
688/// Ring buffer for UMEM completion descriptors (chunk addresses pushed by kernel/NIC after TX completion).
689#[derive(Debug, Clone)]
690pub struct CompletionRing {
691    entries: Vec<u64>,
692    capacity: u32,
693    head: u32,
694    tail: u32,
695}
696
697impl CompletionRing {
698    /// Creates a new completion ring with `capacity` descriptors. `capacity` must be a power of two.
699    pub fn new(capacity: u32) -> Self {
700        assert!(
701            capacity > 0 && capacity.is_power_of_two(),
702            "CompletionRing capacity must be a power of two, got {capacity}"
703        );
704        Self {
705            entries: vec![0u64; capacity as usize],
706            capacity,
707            head: 0,
708            tail: 0,
709        }
710    }
711
712    /// Maximum number of descriptors the ring can hold.
713    pub fn capacity(&self) -> u32 {
714        self.capacity
715    }
716
717    /// Number of descriptors currently queued in the ring.
718    pub fn len(&self) -> usize {
719        self.tail.wrapping_sub(self.head) as usize
720    }
721
722    /// Returns `true` if the ring has no queued descriptors.
723    pub fn is_empty(&self) -> bool {
724        self.head == self.tail
725    }
726
727    /// Returns `true` if the ring is at maximum capacity.
728    pub fn is_full(&self) -> bool {
729        self.len() >= self.capacity as usize
730    }
731
732    /// Enqueues a chunk address into the completion ring. Returns `false` if full.
733    pub fn produce(&mut self, addr: u64) -> bool {
734        if self.is_full() {
735            return false;
736        }
737        let idx = (self.tail as usize) & ((self.capacity - 1) as usize);
738        self.entries[idx] = addr;
739        self.tail = self.tail.wrapping_add(1);
740        true
741    }
742
743    /// Enqueues a batch of chunk addresses into the ring. Returns count enqueued.
744    pub fn produce_batch(&mut self, addrs: &[u64]) -> usize {
745        let available = (self.capacity as usize).saturating_sub(self.len());
746        let to_produce = addrs.len().min(available);
747        if to_produce == 0 {
748            return 0;
749        }
750
751        let mask = (self.capacity - 1) as usize;
752        let start = (self.tail as usize) & mask;
753        let first_len = to_produce.min(self.entries.len() - start);
754        self.entries[start..start + first_len].copy_from_slice(&addrs[..first_len]);
755
756        if first_len < to_produce {
757            let second_len = to_produce - first_len;
758            self.entries[..second_len].copy_from_slice(&addrs[first_len..to_produce]);
759        }
760
761        self.tail = self.tail.wrapping_add(to_produce as u32);
762        to_produce
763    }
764
765    /// Dequeues a completed chunk address from the ring.
766    pub fn consume(&mut self) -> Option<u64> {
767        if self.is_empty() {
768            return None;
769        }
770        let idx = (self.head as usize) & ((self.capacity - 1) as usize);
771        let addr = self.entries[idx];
772        self.head = self.head.wrapping_add(1);
773        Some(addr)
774    }
775
776    /// Dequeues up to `out.len()` chunk addresses into `out`. Returns count dequeued.
777    pub fn consume_batch(&mut self, out: &mut [u64]) -> usize {
778        let available = self.len();
779        let to_consume = out.len().min(available);
780        if to_consume == 0 {
781            return 0;
782        }
783
784        let mask = (self.capacity - 1) as usize;
785        let start = (self.head as usize) & mask;
786        let first_len = to_consume.min(self.entries.len() - start);
787        out[..first_len].copy_from_slice(&self.entries[start..start + first_len]);
788
789        if first_len < to_consume {
790            let second_len = to_consume - first_len;
791            out[first_len..to_consume].copy_from_slice(&self.entries[..second_len]);
792        }
793
794        self.head = self.head.wrapping_add(to_consume as u32);
795        to_consume
796    }
797}
798
799/// Flag for copy-mode bind in AF_XDP (`XDP_COPY`).
800pub const XDP_COPY: u16 = 1 << 1;
801
802/// Flag for zero-copy mode bind in AF_XDP (`XDP_ZEROCOPY`).
803pub const XDP_ZERO_COPY: u16 = 1 << 2;
804
805/// Alias for `XDP_ZERO_COPY` matching kernel naming `XDP_ZEROCOPY`.
806pub const XDP_ZEROCOPY: u16 = XDP_ZERO_COPY;
807
808/// Linux internal errno for unsupported operation (`ENOTSUPP`).
809pub const ENOTSUPP: i32 = 524;
810
811/// The operational mode for AF_XDP binding, with automatic fallback.
812#[derive(Debug, Clone, Copy, PartialEq, Eq)]
813pub enum XskBindMode {
814    /// Hardware/driver zero-copy mode directly to NIC ring.
815    ZeroCopy,
816    /// Kernel copy mode (skb copy into UMEM, used in cloud VMs).
817    Copy,
818    /// Graceful portable fallback using standard recvmmsg/sendmmsg socket I/O.
819    FallbackRecvmmsg,
820}
821
822/// Descriptor for AF_XDP packets in RX and TX rings.
823#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
824pub struct XskDesc {
825    pub addr: u64,
826    pub len: u32,
827    pub flags: u32,
828}
829
830impl XskDesc {
831    /// Creates a new `XskDesc` with the specified address, length, and flags.
832    pub const fn new(addr: u64, len: u32, flags: u32) -> Self {
833        Self { addr, len, flags }
834    }
835}
836
837/// Circular descriptor ring for AF_XDP RX packets.
838#[derive(Debug, Clone)]
839pub struct RxRing {
840    entries: Vec<XskDesc>,
841    capacity: u32,
842    head: u32,
843    tail: u32,
844}
845
846impl RxRing {
847    /// Creates a new RX ring with `capacity` descriptors. `capacity` must be a power of two.
848    pub fn new(capacity: u32) -> Self {
849        assert!(
850            capacity > 0 && capacity.is_power_of_two(),
851            "RxRing capacity must be a power of two, got {capacity}"
852        );
853        Self {
854            entries: vec![XskDesc::default(); capacity as usize],
855            capacity,
856            head: 0,
857            tail: 0,
858        }
859    }
860
861    /// Maximum number of descriptors the ring can hold.
862    pub fn capacity(&self) -> u32 {
863        self.capacity
864    }
865
866    /// Number of descriptors currently queued in the ring.
867    pub fn len(&self) -> usize {
868        self.tail.wrapping_sub(self.head) as usize
869    }
870
871    /// Returns `true` if the ring has no queued descriptors.
872    pub fn is_empty(&self) -> bool {
873        self.head == self.tail
874    }
875
876    /// Returns `true` if the ring is at maximum capacity.
877    pub fn is_full(&self) -> bool {
878        self.len() >= self.capacity as usize
879    }
880
881    /// Enqueues a descriptor into the RX ring. Returns `false` if full.
882    pub fn produce(&mut self, desc: XskDesc) -> bool {
883        if self.is_full() {
884            return false;
885        }
886        let idx = (self.tail as usize) & ((self.capacity - 1) as usize);
887        self.entries[idx] = desc;
888        self.tail = self.tail.wrapping_add(1);
889        true
890    }
891
892    /// Enqueues a batch of descriptors into the RX ring. Returns count enqueued.
893    pub fn produce_batch(&mut self, descs: &[XskDesc]) -> usize {
894        let available = (self.capacity as usize).saturating_sub(self.len());
895        let to_produce = descs.len().min(available);
896        if to_produce == 0 {
897            return 0;
898        }
899
900        let mask = (self.capacity - 1) as usize;
901        let start = (self.tail as usize) & mask;
902        let first_len = to_produce.min(self.entries.len() - start);
903        self.entries[start..start + first_len].copy_from_slice(&descs[..first_len]);
904
905        if first_len < to_produce {
906            let second_len = to_produce - first_len;
907            self.entries[..second_len].copy_from_slice(&descs[first_len..to_produce]);
908        }
909
910        self.tail = self.tail.wrapping_add(to_produce as u32);
911        to_produce
912    }
913
914    /// Dequeues a descriptor from the RX ring.
915    pub fn consume(&mut self) -> Option<XskDesc> {
916        if self.is_empty() {
917            return None;
918        }
919        let idx = (self.head as usize) & ((self.capacity - 1) as usize);
920        let desc = self.entries[idx];
921        self.head = self.head.wrapping_add(1);
922        Some(desc)
923    }
924
925    /// Dequeues up to `out.len()` descriptors into `out`. Returns count dequeued.
926    pub fn consume_batch(&mut self, out: &mut [XskDesc]) -> usize {
927        let available = self.len();
928        let to_consume = out.len().min(available);
929        if to_consume == 0 {
930            return 0;
931        }
932
933        let mask = (self.capacity - 1) as usize;
934        let start = (self.head as usize) & mask;
935        let first_len = to_consume.min(self.entries.len() - start);
936        out[..first_len].copy_from_slice(&self.entries[start..start + first_len]);
937
938        if first_len < to_consume {
939            let second_len = to_consume - first_len;
940            out[first_len..to_consume].copy_from_slice(&self.entries[..second_len]);
941        }
942
943        self.head = self.head.wrapping_add(to_consume as u32);
944        to_consume
945    }
946}
947
948/// Circular descriptor ring for AF_XDP TX packets.
949#[derive(Debug, Clone)]
950pub struct TxRing {
951    entries: Vec<XskDesc>,
952    capacity: u32,
953    head: u32,
954    tail: u32,
955}
956
957impl TxRing {
958    /// Creates a new TX ring with `capacity` descriptors. `capacity` must be a power of two.
959    pub fn new(capacity: u32) -> Self {
960        assert!(
961            capacity > 0 && capacity.is_power_of_two(),
962            "TxRing capacity must be a power of two, got {capacity}"
963        );
964        Self {
965            entries: vec![XskDesc::default(); capacity as usize],
966            capacity,
967            head: 0,
968            tail: 0,
969        }
970    }
971
972    /// Maximum number of descriptors the ring can hold.
973    pub fn capacity(&self) -> u32 {
974        self.capacity
975    }
976
977    /// Number of descriptors currently queued in the ring.
978    pub fn len(&self) -> usize {
979        self.tail.wrapping_sub(self.head) as usize
980    }
981
982    /// Returns `true` if the ring has no queued descriptors.
983    pub fn is_empty(&self) -> bool {
984        self.head == self.tail
985    }
986
987    /// Returns `true` if the ring is at maximum capacity.
988    pub fn is_full(&self) -> bool {
989        self.len() >= self.capacity as usize
990    }
991
992    /// Enqueues a descriptor into the TX ring. Returns `false` if full.
993    pub fn produce(&mut self, desc: XskDesc) -> bool {
994        if self.is_full() {
995            return false;
996        }
997        let idx = (self.tail as usize) & ((self.capacity - 1) as usize);
998        self.entries[idx] = desc;
999        self.tail = self.tail.wrapping_add(1);
1000        true
1001    }
1002
1003    /// Enqueues a batch of descriptors into the TX ring. Returns count enqueued.
1004    pub fn produce_batch(&mut self, descs: &[XskDesc]) -> usize {
1005        let available = (self.capacity as usize).saturating_sub(self.len());
1006        let to_produce = descs.len().min(available);
1007        if to_produce == 0 {
1008            return 0;
1009        }
1010
1011        let mask = (self.capacity - 1) as usize;
1012        let start = (self.tail as usize) & mask;
1013        let first_len = to_produce.min(self.entries.len() - start);
1014        self.entries[start..start + first_len].copy_from_slice(&descs[..first_len]);
1015
1016        if first_len < to_produce {
1017            let second_len = to_produce - first_len;
1018            self.entries[..second_len].copy_from_slice(&descs[first_len..to_produce]);
1019        }
1020
1021        self.tail = self.tail.wrapping_add(to_produce as u32);
1022        to_produce
1023    }
1024
1025    /// Dequeues a descriptor from the TX ring.
1026    pub fn consume(&mut self) -> Option<XskDesc> {
1027        if self.is_empty() {
1028            return None;
1029        }
1030        let idx = (self.head as usize) & ((self.capacity - 1) as usize);
1031        let desc = self.entries[idx];
1032        self.head = self.head.wrapping_add(1);
1033        Some(desc)
1034    }
1035
1036    /// Dequeues up to `out.len()` descriptors into `out`. Returns count dequeued.
1037    pub fn consume_batch(&mut self, out: &mut [XskDesc]) -> usize {
1038        let available = self.len();
1039        let to_consume = out.len().min(available);
1040        if to_consume == 0 {
1041            return 0;
1042        }
1043
1044        let mask = (self.capacity - 1) as usize;
1045        let start = (self.head as usize) & mask;
1046        let first_len = to_consume.min(self.entries.len() - start);
1047        out[..first_len].copy_from_slice(&self.entries[start..start + first_len]);
1048
1049        if first_len < to_consume {
1050            let second_len = to_consume - first_len;
1051            out[first_len..to_consume].copy_from_slice(&self.entries[..second_len]);
1052        }
1053
1054        self.head = self.head.wrapping_add(to_consume as u32);
1055        to_consume
1056    }
1057}
1058
1059/// RAII helper ensuring raw file descriptors are closed if initialization fails early.
1060struct AutoCloseFd(RawFd);
1061
1062impl AutoCloseFd {
1063    fn into_raw(mut self) -> RawFd {
1064        let fd = self.0;
1065        self.0 = -1;
1066        fd
1067    }
1068}
1069
1070impl Drop for AutoCloseFd {
1071    fn drop(&mut self) {
1072        if self.0 >= 0 {
1073            // SAFETY: `self.0` is an open file descriptor owned by `AutoCloseFd`.
1074            // Closing it releases OS resources upon early return or failure.
1075            unsafe {
1076                libc::close(self.0);
1077            }
1078        }
1079    }
1080}
1081
1082/// AF_XDP socket with three-tier opportunistic fallback engine.
1083pub struct XskSocket {
1084    fd: RawFd,
1085    mode: XskBindMode,
1086    rx_ring: RxRing,
1087    tx_ring: TxRing,
1088    filter: Option<crate::bpf::XdpRedirectFilter>,
1089}
1090
1091impl XskSocket {
1092    /// Attempts opportunistic binding of an AF_XDP socket.
1093    ///
1094    /// Fallback order:
1095    /// 1. Zero-Copy mode (`XskBindMode::ZeroCopy`)
1096    /// 2. Copy mode (`XskBindMode::Copy`)
1097    /// 3. Portable recvmmsg fallback (`XskBindMode::FallbackRecvmmsg`)
1098    ///
1099    /// If running unprivileged or kernel lacks AF_XDP, cleanly returns `FallbackRecvmmsg`.
1100    pub fn bind_opportunistic(ifname: &str, queue_id: u32, umem: &UmemPool) -> io::Result<Self> {
1101        // SAFETY: Calling `socket(AF_XDP, SOCK_RAW, 0)` requests a raw AF_XDP socket from the kernel.
1102        // It requires no initialized pointers or preconditions.
1103        let fd = unsafe { libc::socket(libc::AF_XDP, libc::SOCK_RAW, 0) };
1104        if fd < 0 {
1105            let err = io::Error::last_os_error();
1106            match err.raw_os_error() {
1107                Some(
1108                    libc::EPERM
1109                    | libc::EACCES
1110                    | libc::EAFNOSUPPORT
1111                    | libc::ENOPROTOOPT
1112                    | libc::EPROTONOSUPPORT,
1113                ) => return Ok(Self::fallback()),
1114                _ => return Ok(Self::fallback()),
1115            }
1116        }
1117
1118        let sock_guard = AutoCloseFd(fd);
1119
1120        // 1. Configure UMEM registration
1121        let mr = libc::xdp_umem_reg {
1122            addr: umem.area_ptr() as u64,
1123            len: umem.size() as u64,
1124            chunk_size: umem.chunk_size() as u32,
1125            headroom: 0,
1126            flags: 0,
1127            tx_metadata_len: 0,
1128        };
1129
1130        // SAFETY: `fd` is a valid open AF_XDP socket. `mr` is a stack-local `xdp_umem_reg`
1131        // initialized with valid UMEM memory bounds and chunk parameters. The memory region
1132        // is guaranteed valid by `umem`. Size passed matches `sizeof(xdp_umem_reg)`.
1133        let ret_umem = unsafe {
1134            libc::setsockopt(
1135                fd,
1136                libc::SOL_XDP,
1137                libc::XDP_UMEM_REG,
1138                std::ptr::addr_of!(mr).cast::<libc::c_void>(),
1139                std::mem::size_of::<libc::xdp_umem_reg>() as libc::socklen_t,
1140            )
1141        };
1142        if ret_umem != 0 {
1143            return Ok(Self::fallback());
1144        }
1145
1146        let ring_size: u32 = UMEM_RING_SIZE;
1147
1148        // 2. Configure Fill Ring
1149        // SAFETY: `fd` is an open AF_XDP socket. `ring_size` is a valid stack-local u32.
1150        // Size passed matches `sizeof(u32)`.
1151        let ret_fill = unsafe {
1152            libc::setsockopt(
1153                fd,
1154                libc::SOL_XDP,
1155                libc::XDP_UMEM_FILL_RING,
1156                std::ptr::addr_of!(ring_size).cast::<libc::c_void>(),
1157                std::mem::size_of::<u32>() as libc::socklen_t,
1158            )
1159        };
1160        if ret_fill != 0 {
1161            return Ok(Self::fallback());
1162        }
1163
1164        // 3. Configure Completion Ring
1165        // SAFETY: `fd` is an open AF_XDP socket. `ring_size` is a valid stack-local u32.
1166        // Size passed matches `sizeof(u32)`.
1167        let ret_comp = unsafe {
1168            libc::setsockopt(
1169                fd,
1170                libc::SOL_XDP,
1171                libc::XDP_UMEM_COMPLETION_RING,
1172                std::ptr::addr_of!(ring_size).cast::<libc::c_void>(),
1173                std::mem::size_of::<u32>() as libc::socklen_t,
1174            )
1175        };
1176        if ret_comp != 0 {
1177            return Ok(Self::fallback());
1178        }
1179
1180        // 4. Configure RX Ring
1181        // SAFETY: `fd` is an open AF_XDP socket. `ring_size` is a valid stack-local u32.
1182        // Size passed matches `sizeof(u32)`.
1183        let ret_rx = unsafe {
1184            libc::setsockopt(
1185                fd,
1186                libc::SOL_XDP,
1187                libc::XDP_RX_RING,
1188                std::ptr::addr_of!(ring_size).cast::<libc::c_void>(),
1189                std::mem::size_of::<u32>() as libc::socklen_t,
1190            )
1191        };
1192        if ret_rx != 0 {
1193            return Ok(Self::fallback());
1194        }
1195
1196        // 5. Configure TX Ring
1197        // SAFETY: `fd` is an open AF_XDP socket. `ring_size` is a valid stack-local u32.
1198        // Size passed matches `sizeof(u32)`.
1199        let ret_tx = unsafe {
1200            libc::setsockopt(
1201                fd,
1202                libc::SOL_XDP,
1203                libc::XDP_TX_RING,
1204                std::ptr::addr_of!(ring_size).cast::<libc::c_void>(),
1205                std::mem::size_of::<u32>() as libc::socklen_t,
1206            )
1207        };
1208        if ret_tx != 0 {
1209            return Ok(Self::fallback());
1210        }
1211
1212        // 6. Resolve network interface index
1213        let c_ifname = match CString::new(ifname) {
1214            Ok(c) => c,
1215            Err(_) => return Ok(Self::fallback()),
1216        };
1217
1218        // SAFETY: `c_ifname` is a valid null-terminated C string.
1219        let ifindex = unsafe { libc::if_nametoindex(c_ifname.as_ptr()) };
1220        if ifindex == 0 {
1221            return Ok(Self::fallback());
1222        }
1223
1224        // 7. Attempt bind with XDP_ZERO_COPY
1225        let mut sxdp: libc::sockaddr_xdp = unsafe {
1226            // SAFETY: zeroing a C struct with POD fields is safe.
1227            std::mem::zeroed()
1228        };
1229        sxdp.sxdp_family = libc::AF_XDP as u16;
1230        sxdp.sxdp_ifindex = ifindex;
1231        sxdp.sxdp_queue_id = queue_id;
1232        sxdp.sxdp_shared_umem_fd = 0;
1233        sxdp.sxdp_flags = XDP_ZERO_COPY;
1234
1235        // SAFETY: `fd` is an open AF_XDP socket. `sxdp` is a valid stack-local sockaddr_xdp.
1236        // Size passed matches `sizeof(sockaddr_xdp)`.
1237        let ret_zc = unsafe {
1238            libc::bind(
1239                fd,
1240                std::ptr::addr_of!(sxdp).cast::<libc::sockaddr>(),
1241                std::mem::size_of::<libc::sockaddr_xdp>() as libc::socklen_t,
1242            )
1243        };
1244
1245        if ret_zc == 0 {
1246            let fd = sock_guard.into_raw();
1247            return Ok(Self {
1248                fd,
1249                mode: XskBindMode::ZeroCopy,
1250                rx_ring: RxRing::new(UMEM_RING_SIZE),
1251                tx_ring: TxRing::new(UMEM_RING_SIZE),
1252                filter: None,
1253            });
1254        }
1255
1256        // 8. Attempt bind with XDP_COPY if ZeroCopy returned unsupported error
1257        let zc_err = io::Error::last_os_error();
1258        let should_try_copy = matches!(
1259            zc_err.raw_os_error(),
1260            Some(libc::EOPNOTSUPP | ENOTSUPP | libc::EINVAL)
1261        );
1262
1263        if should_try_copy {
1264            sxdp.sxdp_flags = XDP_COPY;
1265            // SAFETY: `fd` is an open AF_XDP socket. `sxdp` is a valid stack-local sockaddr_xdp.
1266            // Size passed matches `sizeof(sockaddr_xdp)`.
1267            let ret_copy = unsafe {
1268                libc::bind(
1269                    fd,
1270                    std::ptr::addr_of!(sxdp).cast::<libc::sockaddr>(),
1271                    std::mem::size_of::<libc::sockaddr_xdp>() as libc::socklen_t,
1272                )
1273            };
1274
1275            if ret_copy == 0 {
1276                let fd = sock_guard.into_raw();
1277                return Ok(Self {
1278                    fd,
1279                    mode: XskBindMode::Copy,
1280                    rx_ring: RxRing::new(UMEM_RING_SIZE),
1281                    tx_ring: TxRing::new(UMEM_RING_SIZE),
1282                    filter: None,
1283                });
1284            }
1285        }
1286
1287        // All AF_XDP attempts failed; fallback to portable recvmmsg mode
1288        Ok(Self::fallback())
1289    }
1290
1291    /// Creates a fallback socket instance in `FallbackRecvmmsg` mode without an active AF_XDP socket.
1292    pub fn fallback() -> Self {
1293        Self {
1294            fd: -1,
1295            mode: XskBindMode::FallbackRecvmmsg,
1296            rx_ring: RxRing::new(UMEM_RING_SIZE),
1297            tx_ring: TxRing::new(UMEM_RING_SIZE),
1298            filter: None,
1299        }
1300    }
1301
1302    /// The active bind mode of this socket (ZeroCopy, Copy, or FallbackRecvmmsg).
1303    pub fn mode(&self) -> XskBindMode {
1304        self.mode
1305    }
1306
1307    /// The raw file descriptor of the underlying AF_XDP socket, or -1 in fallback mode.
1308    pub fn fd(&self) -> RawFd {
1309        self.fd
1310    }
1311
1312    /// Mutable reference to the socket's RX ring buffer.
1313    pub fn rx_ring_mut(&mut self) -> &mut RxRing {
1314        &mut self.rx_ring
1315    }
1316
1317    /// Mutable reference to the socket's TX ring buffer.
1318    pub fn tx_ring_mut(&mut self) -> &mut TxRing {
1319        &mut self.tx_ring
1320    }
1321
1322    /// Immutable reference to the socket's RX ring buffer.
1323    pub fn rx_ring(&self) -> &RxRing {
1324        &self.rx_ring
1325    }
1326
1327    /// Immutable reference to the socket's TX ring buffer.
1328    pub fn tx_ring(&self) -> &TxRing {
1329        &self.tx_ring
1330    }
1331
1332    /// Reference to the attached eBPF redirect filter, if active.
1333    pub fn filter(&self) -> Option<&crate::bpf::XdpRedirectFilter> {
1334        self.filter.as_ref()
1335    }
1336
1337    /// Mutable reference to the attached eBPF redirect filter, if active.
1338    pub fn filter_mut(&mut self) -> Option<&mut crate::bpf::XdpRedirectFilter> {
1339        self.filter.as_mut()
1340    }
1341
1342    /// Opportunistically attaches an eBPF XDP redirect filter steering incoming UDP traffic
1343    /// for `listen_port` on `ifname` into this socket's RX ring.
1344    pub fn attach_bpf_filter(
1345        &mut self,
1346        ifname: &str,
1347        listen_port: u16,
1348        queue_id: u32,
1349    ) -> crate::bpf::BpfFilterStatus {
1350        if self.fd < 0 {
1351            return crate::bpf::BpfFilterStatus::Unsupported;
1352        }
1353        match crate::bpf::XdpRedirectFilter::load_and_attach(ifname, listen_port, queue_id, self.fd)
1354        {
1355            Ok(f) => {
1356                let prog_fd = f.prog_fd();
1357                self.filter = Some(f);
1358                crate::bpf::BpfFilterStatus::Attached(prog_fd)
1359            }
1360            Err(status) => status,
1361        }
1362    }
1363
1364    /// Opportunistically attaches a multi-queue eBPF XDP redirect filter steering incoming UDP traffic
1365    /// for `listen_port` on `ifname` across multiple RX queues into core-pinned AF_XDP rings.
1366    pub fn attach_bpf_filter_multi_queue(
1367        &mut self,
1368        ifname: &str,
1369        listen_port: u16,
1370        queue_fds: &[(u32, RawFd)],
1371    ) -> crate::bpf::BpfFilterStatus {
1372        if self.fd < 0 {
1373            return crate::bpf::BpfFilterStatus::Unsupported;
1374        }
1375        match crate::bpf::XdpRedirectFilter::load_multi_queue(ifname, listen_port, queue_fds) {
1376            Ok(f) => {
1377                let prog_fd = f.prog_fd();
1378                self.filter = Some(f);
1379                crate::bpf::BpfFilterStatus::Attached(prog_fd)
1380            }
1381            Err(status) => status,
1382        }
1383    }
1384
1385    /// Sets or updates the socket file descriptor for a specific queue ID in the attached BPF filter.
1386    pub fn set_bpf_filter_queue_socket(&mut self, queue_id: u32, xsk_fd: RawFd) -> io::Result<()> {
1387        match self.filter.as_mut() {
1388            Some(filter) => filter.set_socket_for_queue(queue_id, xsk_fd),
1389            None => Err(io::Error::new(
1390                io::ErrorKind::NotFound,
1391                "no BPF filter attached to socket",
1392            )),
1393        }
1394    }
1395
1396    /// Binds multiple AF_XDP sockets across `queue_ids` for the given interface and UMEM pool.
1397    ///
1398    /// Returns a list of `(queue_id, XskSocket)` pairs. If binding fails for a queue,
1399    /// a socket in fallback mode is returned for that queue.
1400    pub fn bind_multi_queue(
1401        ifname: &str,
1402        queue_ids: &[u32],
1403        umem: &UmemPool,
1404    ) -> io::Result<Vec<(u32, Self)>> {
1405        let mut sockets = Vec::with_capacity(queue_ids.len());
1406        for &qid in queue_ids {
1407            let sock = Self::bind_opportunistic(ifname, qid, umem)?;
1408            sockets.push((qid, sock));
1409        }
1410        Ok(sockets)
1411    }
1412}
1413
1414impl Drop for XskSocket {
1415    fn drop(&mut self) {
1416        if self.fd >= 0 {
1417            // SAFETY: `self.fd` is an open AF_XDP socket descriptor exclusively owned by
1418            // this `XskSocket`. Closing it releases the kernel socket resources.
1419            unsafe {
1420                libc::close(self.fd);
1421            }
1422            self.fd = -1;
1423        }
1424    }
1425}
yip_io/af_xdp.rs

yip_io/
af_xdp.rs
