yip-io 0.1.0

//! AF_XDP (XSK) kernel-bypass driver and UMEM shared memory pool.
//!
//! Provides zero-copy memory buffers (`UmemPool`) and ring buffer descriptors
//! (`FillRing`, `CompletionRing`) for kernel-bypass packet I/O.
//!
//! Low-level libc XSK/XDP structures and unsafe memory mappings are quarantined
//! within this module, with explicit `// SAFETY:` justifications on all unsafe blocks.

use std::ffi::CString;
use std::io;
use std::os::fd::RawFd;
use std::sync::Arc;

/// Default UMEM chunk size in bytes (2 KiB aligns with MTU 1500 + wire overhead).
pub const UMEM_CHUNK_SIZE: usize = 2048;

/// Default ring size in descriptors (must be a power of two).
pub const UMEM_RING_SIZE: u32 = 2048;

/// Configuration parameters for allocating an AF_XDP [`UmemPool`].
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct UmemPoolConfig {
    /// Number of chunk buffers in the pool.
    pub num_chunks: usize,
    /// Size of each chunk in bytes (typically 2048 or 4096).
    pub chunk_size: usize,
    /// Pre-fault all memory pages during allocation via `MAP_POPULATE`.
    ///
    /// When `true` (the default), eliminates page fault jitter during packet I/O
    /// at the expense of higher startup latency and immediate RSS allocation.
    /// When `false`, pages are faulted lazily on-demand, reducing startup time
    /// and memory footprint on resource-constrained hosts.
    pub pre_fault: bool,
    /// Pin pages in physical RAM via `libc::mlock` to prevent swapping.
    ///
    /// Requires `CAP_IPC_LOCK` or sufficient `RLIMIT_MEMLOCK`. If `true` and `mlock`
    /// fails, an error is returned.
    pub lock_memory: bool,
}

impl Default for UmemPoolConfig {
    fn default() -> Self {
        Self {
            num_chunks: 4096,
            chunk_size: 2048,
            pre_fault: true,
            lock_memory: false,
        }
    }
}

/// Pre-allocated page-aligned shared memory pool for AF_XDP kernel-bypass I/O.
pub struct UmemPool {
    area: *mut libc::c_void,
    size: usize,
    chunk_size: usize,
    free_chunks: Vec<u64>,
}

impl UmemPool {
    /// Allocates an anonymous, page-aligned shared memory region for `num_chunks` chunks
    /// of size `chunk_size` with default pre-faulting (`MAP_POPULATE`) enabled.
    pub fn new(num_chunks: usize, chunk_size: usize) -> io::Result<Self> {
        Self::with_config(UmemPoolConfig {
            num_chunks,
            chunk_size,
            pre_fault: true,
            lock_memory: false,
        })
    }

    /// Allocates a new `UmemPool` with lazy on-demand page allocation (`MAP_POPULATE` disabled),
    /// minimizing startup latency and initial RSS footprint.
    pub fn new_lazy(num_chunks: usize, chunk_size: usize) -> io::Result<Self> {
        Self::with_config(UmemPoolConfig {
            num_chunks,
            chunk_size,
            pre_fault: false,
            lock_memory: false,
        })
    }

    /// Allocates a new `UmemPool` according to the provided [`UmemPoolConfig`].
    pub fn with_config(config: UmemPoolConfig) -> io::Result<Self> {
        if config.num_chunks == 0 || config.chunk_size == 0 {
            return Err(io::Error::new(
                io::ErrorKind::InvalidInput,
                "num_chunks and chunk_size must be positive",
            ));
        }

        let size = config
            .num_chunks
            .checked_mul(config.chunk_size)
            .ok_or_else(|| io::Error::new(io::ErrorKind::InvalidInput, "UMEM size overflow"))?;

        let mut flags = libc::MAP_ANONYMOUS | libc::MAP_SHARED;
        if config.pre_fault {
            flags |= libc::MAP_POPULATE;
        }

        // SAFETY: We request an anonymous, shared, page-aligned memory mapping of `size` bytes.
        // The address is NULL so the kernel selects the starting address.
        // We verify that the returned pointer is not `MAP_FAILED` before proceeding.
        let area = unsafe {
            libc::mmap(
                std::ptr::null_mut(),
                size,
                libc::PROT_READ | libc::PROT_WRITE,
                flags,
                -1,
                0,
            )
        };

        if area == libc::MAP_FAILED {
            return Err(io::Error::last_os_error());
        }

        if config.lock_memory {
            // SAFETY: `area` points to `size` bytes of freshly mapped valid memory.
            let lock_ret = unsafe { libc::mlock(area, size) };
            if lock_ret != 0 {
                let err = io::Error::last_os_error();
                // SAFETY: On mlock failure, unmap the allocated area before returning error.
                unsafe {
                    libc::munmap(area, size);
                }
                return Err(err);
            }
        }

        // Populates free_chunks so popping yields 0, chunk_size, 2*chunk_size, ...
        let free_chunks: Vec<u64> = (0..config.num_chunks)
            .rev()
            .map(|i| (i * config.chunk_size) as u64)
            .collect();

        Ok(Self {
            area,
            size,
            chunk_size: config.chunk_size,
            free_chunks,
        })
    }

    /// Allocates a chunk address from the free pool, returning `None` if exhausted.
    pub fn alloc_chunk(&mut self) -> Option<u64> {
        self.free_chunks.pop()
    }

    /// Returns a chunk address to the free pool.
    pub fn free_chunk(&mut self, addr: u64) {
        debug_assert!(
            (addr as usize) < self.size && addr.is_multiple_of(self.chunk_size as u64),
            "corrupt chunk address returned to free pool: addr={addr}, size={}, chunk_size={}",
            self.size,
            self.chunk_size
        );
        self.free_chunks.push(addr);
    }

    /// Returns a bounds-checked immutable slice into the chunk starting at `addr`.
    pub fn chunk_slice(&self, addr: u64, len: usize) -> &[u8] {
        assert!(
            (addr as usize)
                .checked_add(len)
                .is_some_and(|end| end <= self.size),
            "UMEM slice out of bounds: addr={addr}, len={len}, total size={}",
            self.size
        );
        let offset = (addr as usize) % self.chunk_size;
        assert!(
            offset
                .checked_add(len)
                .is_some_and(|end| end <= self.chunk_size),
            "UMEM slice len={len} with intra-chunk offset={offset} exceeds chunk_size={}",
            self.chunk_size
        );

        // SAFETY: `self.area` is valid mapped memory of `self.size` bytes. We verified that
        // `addr + len <= self.size`, so `self.area.add(addr as usize)` points to at least `len`
        // contiguous initialized bytes within the mapped region. The lifetime is tied to `&self`,
        // ensuring the memory remains mapped during the slice's lifetime.
        unsafe {
            let ptr = (self.area as *const u8).add(addr as usize);
            std::slice::from_raw_parts(ptr, len)
        }
    }

    /// Returns a bounds-checked mutable slice into the chunk starting at `addr`.
    pub fn chunk_slice_mut(&mut self, addr: u64, len: usize) -> &mut [u8] {
        assert!(
            (addr as usize)
                .checked_add(len)
                .is_some_and(|end| end <= self.size),
            "UMEM slice out of bounds: addr={addr}, len={len}, total size={}",
            self.size
        );
        let offset = (addr as usize) % self.chunk_size;
        assert!(
            offset
                .checked_add(len)
                .is_some_and(|end| end <= self.chunk_size),
            "UMEM slice len={len} with intra-chunk offset={offset} exceeds chunk_size={}",
            self.chunk_size
        );

        // SAFETY: `self.area` is valid mapped memory of `self.size` bytes. We verified that
        // `addr + len <= self.size`, so `self.area.add(addr as usize)` points to at least `len`
        // contiguous initialized bytes within the mapped region. The exclusive lifetime `&mut self`
        // guarantees that no other references to this memory exist during the returned slice's lifetime.
        unsafe {
            let ptr = (self.area as *mut u8).add(addr as usize);
            std::slice::from_raw_parts_mut(ptr, len)
        }
    }

    /// Raw pointer to the mapped UMEM memory area.
    pub fn area_ptr(&self) -> *mut libc::c_void {
        self.area
    }

    /// Total size in bytes of the mapped UMEM memory area.
    pub fn size(&self) -> usize {
        self.size
    }

    /// Chunk size in bytes.
    pub fn chunk_size(&self) -> usize {
        self.chunk_size
    }

    /// Number of free chunks currently available in the pool.
    pub fn free_chunk_count(&self) -> usize {
        self.free_chunks.len()
    }

    /// Returns `true` if `addr` is an aligned chunk address within this pool's bounds.
    #[inline]
    pub fn is_valid_chunk_addr(&self, addr: u64) -> bool {
        (addr as usize) < self.size && addr.is_multiple_of(self.chunk_size as u64)
    }

    /// Unsafe direct access to a mutable raw pointer for a chunk.
    ///
    /// # Safety
    /// Caller must ensure that no other thread or reference is concurrently accessing
    /// the memory range `[addr, addr + len)`.
    pub(crate) unsafe fn chunk_ptr_mut(&self, addr: u64, len: usize) -> *mut u8 {
        assert!(
            (addr as usize)
                .checked_add(len)
                .is_some_and(|end| end <= self.size),
            "UMEM slice out of bounds: addr={addr}, len={len}, total size={}",
            self.size
        );
        let offset = (addr as usize) % self.chunk_size;
        assert!(
            offset
                .checked_add(len)
                .is_some_and(|end| end <= self.chunk_size),
            "UMEM slice len={len} with intra-chunk offset={offset} exceeds chunk_size={}",
            self.chunk_size
        );

        // SAFETY: The caller guarantees exclusive access to this slice. `self.area` is valid
        // mapped memory of `self.size` bytes, and we verified that `addr + len <= self.size`.
        unsafe { (self.area as *mut u8).add(addr as usize) }
    }

    /// Partitions an `Arc<UmemPool>` into `num_slabs` disjoint, per-core chunk slabs.
    ///
    /// Each slab receives an equal partition of chunks (`total_chunks / num_slabs`).
    /// Slabs operate completely independently without shared locks or cross-core
    /// cache line contention.
    pub fn create_slabs(self: &Arc<Self>, num_slabs: usize) -> io::Result<Vec<UmemSlab>> {
        if num_slabs == 0 {
            return Err(io::Error::new(
                io::ErrorKind::InvalidInput,
                "num_slabs must be positive",
            ));
        }

        let total_chunks = self.size / self.chunk_size;
        let chunks_per_slab = total_chunks / num_slabs;
        if chunks_per_slab == 0 {
            return Err(io::Error::new(
                io::ErrorKind::InvalidInput,
                format!("not enough chunks ({total_chunks}) for requested slabs ({num_slabs})"),
            ));
        }

        let mut slabs = Vec::with_capacity(num_slabs);
        for slab_id in 0..num_slabs {
            let base_addr = (slab_id * chunks_per_slab * self.chunk_size) as u64;
            // Pre-allocate free chunks so popping yields base_addr, base_addr + chunk_size, ...
            let free_chunks: Vec<u64> = (0..chunks_per_slab)
                .rev()
                .map(|i| base_addr + (i * self.chunk_size) as u64)
                .collect();

            slabs.push(UmemSlab {
                pool: Arc::clone(self),
                slab_id,
                base_addr,
                num_chunks: chunks_per_slab,
                chunk_size: self.chunk_size,
                free_chunks,
            });
        }

        Ok(slabs)
    }

    /// Allocates a new `UmemPool` with default configuration and partitions it into `num_slabs` disjoint slabs.
    pub fn new_slabs(
        num_chunks: usize,
        chunk_size: usize,
        num_slabs: usize,
    ) -> io::Result<(Arc<Self>, Vec<UmemSlab>)> {
        Self::new_slabs_with_config(
            UmemPoolConfig {
                num_chunks,
                chunk_size,
                pre_fault: true,
                lock_memory: false,
            },
            num_slabs,
        )
    }

    /// Allocates a new `UmemPool` using the specified [`UmemPoolConfig`] and partitions it into `num_slabs` disjoint slabs.
    pub fn new_slabs_with_config(
        config: UmemPoolConfig,
        num_slabs: usize,
    ) -> io::Result<(Arc<Self>, Vec<UmemSlab>)> {
        let pool = Arc::new(Self::with_config(config)?);
        let slabs = pool.create_slabs(num_slabs)?;
        Ok((pool, slabs))
    }
}

/// Dedicated per-core / per-queue UMEM chunk slab for lock-free AF_XDP multi-worker scaling.
///
/// Each `UmemSlab` manages an isolated, non-overlapping range of UMEM chunk addresses
/// allocated from a parent `UmemPool`. Worker threads allocate and recycle chunks
/// locally from their dedicated slab without cross-core coordination, atomic locks,
/// or MESI cache-line invalidation.
pub struct UmemSlab {
    pool: Arc<UmemPool>,
    slab_id: usize,
    base_addr: u64,
    num_chunks: usize,
    chunk_size: usize,
    free_chunks: Vec<u64>,
}

impl UmemSlab {
    /// Slab identifier (e.g. core ID or queue ID).
    #[inline]
    pub fn slab_id(&self) -> usize {
        self.slab_id
    }

    /// Base memory address offset of this slab within the parent UMEM area.
    #[inline]
    pub fn base_addr(&self) -> u64 {
        self.base_addr
    }

    /// Upper non-inclusive address boundary of this slab within the parent UMEM area.
    #[inline]
    pub fn end_addr(&self) -> u64 {
        self.base_addr + (self.num_chunks * self.chunk_size) as u64
    }

    /// Total number of chunks assigned to this slab.
    #[inline]
    pub fn num_chunks(&self) -> usize {
        self.num_chunks
    }

    /// Total number of chunks assigned to this slab (alias for `num_chunks`).
    #[inline]
    pub fn total_chunks(&self) -> usize {
        self.num_chunks
    }

    /// Number of free chunks currently available in this slab.
    #[inline]
    pub fn free_chunk_count(&self) -> usize {
        self.free_chunks.len()
    }

    /// Chunk size in bytes.
    #[inline]
    pub fn chunk_size(&self) -> usize {
        self.chunk_size
    }

    /// Reference to the parent `UmemPool`.
    #[inline]
    pub fn pool(&self) -> &Arc<UmemPool> {
        &self.pool
    }

    /// Returns `true` if `addr` is an aligned chunk address belonging to this slab.
    #[inline]
    pub fn contains_addr(&self, addr: u64) -> bool {
        addr >= self.base_addr
            && addr < self.end_addr()
            && (addr - self.base_addr).is_multiple_of(self.chunk_size as u64)
    }

    /// Returns `true` if `addr..addr+len` lies completely within this slab's memory range.
    #[inline]
    pub fn contains_addr_range(&self, addr: u64, len: usize) -> bool {
        addr >= self.base_addr
            && (addr as usize)
                .checked_add(len)
                .is_some_and(|end| end as u64 <= self.end_addr())
    }

    /// Allocates a chunk address from this slab, returning `None` if exhausted.
    #[inline]
    pub fn alloc_chunk(&mut self) -> Option<u64> {
        self.free_chunks.pop()
    }

    /// Returns a chunk address to this slab's free pool.
    #[inline]
    pub fn free_chunk(&mut self, addr: u64) {
        debug_assert!(
            self.contains_addr(addr),
            "corrupt chunk address returned to slab {}: addr={addr}, base={}, end={}, chunk_size={}",
            self.slab_id,
            self.base_addr,
            self.end_addr(),
            self.chunk_size
        );
        self.free_chunks.push(addr);
    }

    /// Allocates up to `out.len()` chunk addresses into `out`. Returns count allocated.
    pub fn alloc_batch(&mut self, out: &mut [u64]) -> usize {
        let count = out.len().min(self.free_chunks.len());
        if count == 0 {
            return 0;
        }
        let start = self.free_chunks.len() - count;
        for (i, slot) in out.iter_mut().take(count).enumerate() {
            *slot = self.free_chunks[start + count - 1 - i];
        }
        self.free_chunks.truncate(start);
        count
    }

    /// Frees a slice of chunk addresses back to this slab.
    pub fn free_batch(&mut self, addrs: &[u64]) {
        for &addr in addrs {
            debug_assert!(
                self.contains_addr(addr),
                "corrupt chunk address in batch for slab {}: addr={addr}",
                self.slab_id
            );
        }
        self.free_chunks.extend_from_slice(addrs);
    }

    /// Pre-populates a `FillRing` with chunks from this slab up to `max_count`.
    ///
    /// Allocates in batches of 64 chunks to maximize memory pipeline throughput.
    /// Returns the total number of chunks enqueued into the fill ring.
    pub fn populate_fill_ring(&mut self, fill_ring: &mut FillRing, max_count: usize) -> usize {
        let mut total = 0;
        let mut buf = [0u64; 64];
        while total < max_count && !self.free_chunks.is_empty() && !fill_ring.is_full() {
            let to_alloc = (max_count - total).min(buf.len());
            let allocated = self.alloc_batch(&mut buf[..to_alloc]);
            if allocated == 0 {
                break;
            }
            let produced = fill_ring.produce_batch(&buf[..allocated]);
            total += produced;
            if produced < allocated {
                self.free_batch(&buf[produced..allocated]);
                break;
            }
        }
        total
    }

    /// Recycles completed chunk descriptors from `CompletionRing` directly back into `FillRing`.
    ///
    /// Consumes up to `max_batch` descriptors from `comp_ring` and attempts to produce them
    /// directly into `fill_ring`. Any surplus chunks that do not fit into `fill_ring` are
    /// returned to this slab's free pool without heap allocations.
    ///
    /// Returns the total number of completed descriptors consumed from `comp_ring`.
    pub fn recycle_from_rings(
        &mut self,
        comp_ring: &mut CompletionRing,
        fill_ring: &mut FillRing,
        max_batch: usize,
    ) -> usize {
        let mut buf = [0u64; 64];
        let batch_size = max_batch.min(buf.len());
        let consumed = comp_ring.consume_batch(&mut buf[..batch_size]);
        if consumed == 0 {
            return 0;
        }

        let produced = fill_ring.produce_batch(&buf[..consumed]);
        if produced < consumed {
            self.free_batch(&buf[produced..consumed]);
        }

        consumed
    }

    /// Returns a bounds-checked immutable slice into the chunk starting at `addr`.
    pub fn chunk_slice(&self, addr: u64, len: usize) -> &[u8] {
        assert!(
            self.contains_addr_range(addr, len),
            "UMEM slab {}: addr={addr}, len={len} out of slab bounds [{}..{})",
            self.slab_id,
            self.base_addr,
            self.end_addr()
        );
        self.pool.chunk_slice(addr, len)
    }

    /// Returns a bounds-checked mutable slice into the chunk starting at `addr`.
    pub fn chunk_slice_mut(&mut self, addr: u64, len: usize) -> &mut [u8] {
        assert!(
            self.contains_addr_range(addr, len),
            "UMEM slab {}: addr={addr}, len={len} out of slab bounds [{}..{})",
            self.slab_id,
            self.base_addr,
            self.end_addr()
        );

        // SAFETY: Each `UmemSlab` owns an isolated, non-overlapping range of the UMEM memory area.
        // We verified that `addr..addr+len` lies strictly within this slab's exclusive address range.
        // The `&mut self` borrow guarantees that no other code is currently accessing this slab's memory.
        unsafe {
            let ptr = self.pool.chunk_ptr_mut(addr, len);
            std::slice::from_raw_parts_mut(ptr, len)
        }
    }
}

impl Drop for UmemPool {
    fn drop(&mut self) {
        if !self.area.is_null() && self.area != libc::MAP_FAILED {
            // SAFETY: `self.area` was mapped by `mmap` with size `self.size` in `new()`,
            // and has not been unmapped yet.
            unsafe {
                libc::munmap(self.area, self.size);
            }
        }
    }
}

// SAFETY: `UmemPool` exclusively owns the mapped memory region `self.area` of `self.size`
// bytes. Transferring ownership of `UmemPool` across threads transfers ownership of the
// entire memory pool safely without aliasing.
unsafe impl Send for UmemPool {}

// SAFETY: All methods taking `&self` only provide immutable, bounds-checked access to the
// mapped memory region and immutable fields. Any mutable access requires `&mut self`
// (such as `chunk_slice_mut`, `alloc_chunk`, `free_chunk`), which Rust's borrow checker
// guarantees to be exclusive.
unsafe impl Sync for UmemPool {}

/// Ring buffer for UMEM fill descriptors (chunk addresses pushed by userspace for the kernel/NIC to fill).
#[derive(Debug, Clone)]
pub struct FillRing {
    entries: Vec<u64>,
    capacity: u32,
    head: u32,
    tail: u32,
}

impl FillRing {
    /// Creates a new fill ring with `capacity` descriptors. `capacity` must be a power of two.
    pub fn new(capacity: u32) -> Self {
        assert!(
            capacity > 0 && capacity.is_power_of_two(),
            "FillRing capacity must be a power of two, got {capacity}"
        );
        Self {
            entries: vec![0u64; capacity as usize],
            capacity,
            head: 0,
            tail: 0,
        }
    }

    /// Maximum number of descriptors the ring can hold.
    pub fn capacity(&self) -> u32 {
        self.capacity
    }

    /// Number of descriptors currently queued in the ring.
    pub fn len(&self) -> usize {
        self.tail.wrapping_sub(self.head) as usize
    }

    /// Returns `true` if the ring has no queued descriptors.
    pub fn is_empty(&self) -> bool {
        self.head == self.tail
    }

    /// Returns `true` if the ring is at maximum capacity.
    pub fn is_full(&self) -> bool {
        self.len() >= self.capacity as usize
    }

    /// Enqueues a chunk address for the kernel/NIC to fill. Returns `false` if full.
    pub fn produce(&mut self, addr: u64) -> bool {
        if self.is_full() {
            return false;
        }
        let idx = (self.tail as usize) & ((self.capacity - 1) as usize);
        self.entries[idx] = addr;
        self.tail = self.tail.wrapping_add(1);
        true
    }

    /// Enqueues a batch of chunk addresses into the ring. Returns count enqueued.
    pub fn produce_batch(&mut self, addrs: &[u64]) -> usize {
        let available = (self.capacity as usize).saturating_sub(self.len());
        let to_produce = addrs.len().min(available);
        if to_produce == 0 {
            return 0;
        }

        let mask = (self.capacity - 1) as usize;
        let start = (self.tail as usize) & mask;
        let first_len = to_produce.min(self.entries.len() - start);
        self.entries[start..start + first_len].copy_from_slice(&addrs[..first_len]);

        if first_len < to_produce {
            let second_len = to_produce - first_len;
            self.entries[..second_len].copy_from_slice(&addrs[first_len..to_produce]);
        }

        self.tail = self.tail.wrapping_add(to_produce as u32);
        to_produce
    }

    /// Dequeues a chunk address from the ring.
    pub fn consume(&mut self) -> Option<u64> {
        if self.is_empty() {
            return None;
        }
        let idx = (self.head as usize) & ((self.capacity - 1) as usize);
        let addr = self.entries[idx];
        self.head = self.head.wrapping_add(1);
        Some(addr)
    }

    /// Dequeues up to `out.len()` chunk addresses into `out`. Returns count dequeued.
    pub fn consume_batch(&mut self, out: &mut [u64]) -> usize {
        let available = self.len();
        let to_consume = out.len().min(available);
        if to_consume == 0 {
            return 0;
        }

        let mask = (self.capacity - 1) as usize;
        let start = (self.head as usize) & mask;
        let first_len = to_consume.min(self.entries.len() - start);
        out[..first_len].copy_from_slice(&self.entries[start..start + first_len]);

        if first_len < to_consume {
            let second_len = to_consume - first_len;
            out[first_len..to_consume].copy_from_slice(&self.entries[..second_len]);
        }

        self.head = self.head.wrapping_add(to_consume as u32);
        to_consume
    }
}

/// Ring buffer for UMEM completion descriptors (chunk addresses pushed by kernel/NIC after TX completion).
#[derive(Debug, Clone)]
pub struct CompletionRing {
    entries: Vec<u64>,
    capacity: u32,
    head: u32,
    tail: u32,
}

impl CompletionRing {
    /// Creates a new completion ring with `capacity` descriptors. `capacity` must be a power of two.
    pub fn new(capacity: u32) -> Self {
        assert!(
            capacity > 0 && capacity.is_power_of_two(),
            "CompletionRing capacity must be a power of two, got {capacity}"
        );
        Self {
            entries: vec![0u64; capacity as usize],
            capacity,
            head: 0,
            tail: 0,
        }
    }

    /// Maximum number of descriptors the ring can hold.
    pub fn capacity(&self) -> u32 {
        self.capacity
    }

    /// Number of descriptors currently queued in the ring.
    pub fn len(&self) -> usize {
        self.tail.wrapping_sub(self.head) as usize
    }

    /// Returns `true` if the ring has no queued descriptors.
    pub fn is_empty(&self) -> bool {
        self.head == self.tail
    }

    /// Returns `true` if the ring is at maximum capacity.
    pub fn is_full(&self) -> bool {
        self.len() >= self.capacity as usize
    }

    /// Enqueues a chunk address into the completion ring. Returns `false` if full.
    pub fn produce(&mut self, addr: u64) -> bool {
        if self.is_full() {
            return false;
        }
        let idx = (self.tail as usize) & ((self.capacity - 1) as usize);
        self.entries[idx] = addr;
        self.tail = self.tail.wrapping_add(1);
        true
    }

    /// Enqueues a batch of chunk addresses into the ring. Returns count enqueued.
    pub fn produce_batch(&mut self, addrs: &[u64]) -> usize {
        let available = (self.capacity as usize).saturating_sub(self.len());
        let to_produce = addrs.len().min(available);
        if to_produce == 0 {
            return 0;
        }

        let mask = (self.capacity - 1) as usize;
        let start = (self.tail as usize) & mask;
        let first_len = to_produce.min(self.entries.len() - start);
        self.entries[start..start + first_len].copy_from_slice(&addrs[..first_len]);

        if first_len < to_produce {
            let second_len = to_produce - first_len;
            self.entries[..second_len].copy_from_slice(&addrs[first_len..to_produce]);
        }

        self.tail = self.tail.wrapping_add(to_produce as u32);
        to_produce
    }

    /// Dequeues a completed chunk address from the ring.
    pub fn consume(&mut self) -> Option<u64> {
        if self.is_empty() {
            return None;
        }
        let idx = (self.head as usize) & ((self.capacity - 1) as usize);
        let addr = self.entries[idx];
        self.head = self.head.wrapping_add(1);
        Some(addr)
    }

    /// Dequeues up to `out.len()` chunk addresses into `out`. Returns count dequeued.
    pub fn consume_batch(&mut self, out: &mut [u64]) -> usize {
        let available = self.len();
        let to_consume = out.len().min(available);
        if to_consume == 0 {
            return 0;
        }

        let mask = (self.capacity - 1) as usize;
        let start = (self.head as usize) & mask;
        let first_len = to_consume.min(self.entries.len() - start);
        out[..first_len].copy_from_slice(&self.entries[start..start + first_len]);

        if first_len < to_consume {
            let second_len = to_consume - first_len;
            out[first_len..to_consume].copy_from_slice(&self.entries[..second_len]);
        }

        self.head = self.head.wrapping_add(to_consume as u32);
        to_consume
    }
}

/// Flag for copy-mode bind in AF_XDP (`XDP_COPY`).
pub const XDP_COPY: u16 = 1 << 1;

/// Flag for zero-copy mode bind in AF_XDP (`XDP_ZEROCOPY`).
pub const XDP_ZERO_COPY: u16 = 1 << 2;

/// Alias for `XDP_ZERO_COPY` matching kernel naming `XDP_ZEROCOPY`.
pub const XDP_ZEROCOPY: u16 = XDP_ZERO_COPY;

/// Linux internal errno for unsupported operation (`ENOTSUPP`).
pub const ENOTSUPP: i32 = 524;

/// The operational mode for AF_XDP binding, with automatic fallback.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum XskBindMode {
    /// Hardware/driver zero-copy mode directly to NIC ring.
    ZeroCopy,
    /// Kernel copy mode (skb copy into UMEM, used in cloud VMs).
    Copy,
    /// Graceful portable fallback using standard recvmmsg/sendmmsg socket I/O.
    FallbackRecvmmsg,
}

/// Descriptor for AF_XDP packets in RX and TX rings.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
pub struct XskDesc {
    pub addr: u64,
    pub len: u32,
    pub flags: u32,
}

impl XskDesc {
    /// Creates a new `XskDesc` with the specified address, length, and flags.
    pub const fn new(addr: u64, len: u32, flags: u32) -> Self {
        Self { addr, len, flags }
    }
}

/// Circular descriptor ring for AF_XDP RX packets.
#[derive(Debug, Clone)]
pub struct RxRing {
    entries: Vec<XskDesc>,
    capacity: u32,
    head: u32,
    tail: u32,
}

impl RxRing {
    /// Creates a new RX ring with `capacity` descriptors. `capacity` must be a power of two.
    pub fn new(capacity: u32) -> Self {
        assert!(
            capacity > 0 && capacity.is_power_of_two(),
            "RxRing capacity must be a power of two, got {capacity}"
        );
        Self {
            entries: vec![XskDesc::default(); capacity as usize],
            capacity,
            head: 0,
            tail: 0,
        }
    }

    /// Maximum number of descriptors the ring can hold.
    pub fn capacity(&self) -> u32 {
        self.capacity
    }

    /// Number of descriptors currently queued in the ring.
    pub fn len(&self) -> usize {
        self.tail.wrapping_sub(self.head) as usize
    }

    /// Returns `true` if the ring has no queued descriptors.
    pub fn is_empty(&self) -> bool {
        self.head == self.tail
    }

    /// Returns `true` if the ring is at maximum capacity.
    pub fn is_full(&self) -> bool {
        self.len() >= self.capacity as usize
    }

    /// Enqueues a descriptor into the RX ring. Returns `false` if full.
    pub fn produce(&mut self, desc: XskDesc) -> bool {
        if self.is_full() {
            return false;
        }
        let idx = (self.tail as usize) & ((self.capacity - 1) as usize);
        self.entries[idx] = desc;
        self.tail = self.tail.wrapping_add(1);
        true
    }

    /// Enqueues a batch of descriptors into the RX ring. Returns count enqueued.
    pub fn produce_batch(&mut self, descs: &[XskDesc]) -> usize {
        let available = (self.capacity as usize).saturating_sub(self.len());
        let to_produce = descs.len().min(available);
        if to_produce == 0 {
            return 0;
        }

        let mask = (self.capacity - 1) as usize;
        let start = (self.tail as usize) & mask;
        let first_len = to_produce.min(self.entries.len() - start);
        self.entries[start..start + first_len].copy_from_slice(&descs[..first_len]);

        if first_len < to_produce {
            let second_len = to_produce - first_len;
            self.entries[..second_len].copy_from_slice(&descs[first_len..to_produce]);
        }

        self.tail = self.tail.wrapping_add(to_produce as u32);
        to_produce
    }

    /// Dequeues a descriptor from the RX ring.
    pub fn consume(&mut self) -> Option<XskDesc> {
        if self.is_empty() {
            return None;
        }
        let idx = (self.head as usize) & ((self.capacity - 1) as usize);
        let desc = self.entries[idx];
        self.head = self.head.wrapping_add(1);
        Some(desc)
    }

    /// Dequeues up to `out.len()` descriptors into `out`. Returns count dequeued.
    pub fn consume_batch(&mut self, out: &mut [XskDesc]) -> usize {
        let available = self.len();
        let to_consume = out.len().min(available);
        if to_consume == 0 {
            return 0;
        }

        let mask = (self.capacity - 1) as usize;
        let start = (self.head as usize) & mask;
        let first_len = to_consume.min(self.entries.len() - start);
        out[..first_len].copy_from_slice(&self.entries[start..start + first_len]);

        if first_len < to_consume {
            let second_len = to_consume - first_len;
            out[first_len..to_consume].copy_from_slice(&self.entries[..second_len]);
        }

        self.head = self.head.wrapping_add(to_consume as u32);
        to_consume
    }
}

/// Circular descriptor ring for AF_XDP TX packets.
#[derive(Debug, Clone)]
pub struct TxRing {
    entries: Vec<XskDesc>,
    capacity: u32,
    head: u32,
    tail: u32,
}

impl TxRing {
    /// Creates a new TX ring with `capacity` descriptors. `capacity` must be a power of two.
    pub fn new(capacity: u32) -> Self {
        assert!(
            capacity > 0 && capacity.is_power_of_two(),
            "TxRing capacity must be a power of two, got {capacity}"
        );
        Self {
            entries: vec![XskDesc::default(); capacity as usize],
            capacity,
            head: 0,
            tail: 0,
        }
    }

    /// Maximum number of descriptors the ring can hold.
    pub fn capacity(&self) -> u32 {
        self.capacity
    }

    /// Number of descriptors currently queued in the ring.
    pub fn len(&self) -> usize {
        self.tail.wrapping_sub(self.head) as usize
    }

    /// Returns `true` if the ring has no queued descriptors.
    pub fn is_empty(&self) -> bool {
        self.head == self.tail
    }

    /// Returns `true` if the ring is at maximum capacity.
    pub fn is_full(&self) -> bool {
        self.len() >= self.capacity as usize
    }

    /// Enqueues a descriptor into the TX ring. Returns `false` if full.
    pub fn produce(&mut self, desc: XskDesc) -> bool {
        if self.is_full() {
            return false;
        }
        let idx = (self.tail as usize) & ((self.capacity - 1) as usize);
        self.entries[idx] = desc;
        self.tail = self.tail.wrapping_add(1);
        true
    }

    /// Enqueues a batch of descriptors into the TX ring. Returns count enqueued.
    pub fn produce_batch(&mut self, descs: &[XskDesc]) -> usize {
        let available = (self.capacity as usize).saturating_sub(self.len());
        let to_produce = descs.len().min(available);
        if to_produce == 0 {
            return 0;
        }

        let mask = (self.capacity - 1) as usize;
        let start = (self.tail as usize) & mask;
        let first_len = to_produce.min(self.entries.len() - start);
        self.entries[start..start + first_len].copy_from_slice(&descs[..first_len]);

        if first_len < to_produce {
            let second_len = to_produce - first_len;
            self.entries[..second_len].copy_from_slice(&descs[first_len..to_produce]);
        }

        self.tail = self.tail.wrapping_add(to_produce as u32);
        to_produce
    }

    /// Dequeues a descriptor from the TX ring.
    pub fn consume(&mut self) -> Option<XskDesc> {
        if self.is_empty() {
            return None;
        }
        let idx = (self.head as usize) & ((self.capacity - 1) as usize);
        let desc = self.entries[idx];
        self.head = self.head.wrapping_add(1);
        Some(desc)
    }

    /// Dequeues up to `out.len()` descriptors into `out`. Returns count dequeued.
    pub fn consume_batch(&mut self, out: &mut [XskDesc]) -> usize {
        let available = self.len();
        let to_consume = out.len().min(available);
        if to_consume == 0 {
            return 0;
        }

        let mask = (self.capacity - 1) as usize;
        let start = (self.head as usize) & mask;
        let first_len = to_consume.min(self.entries.len() - start);
        out[..first_len].copy_from_slice(&self.entries[start..start + first_len]);

        if first_len < to_consume {
            let second_len = to_consume - first_len;
            out[first_len..to_consume].copy_from_slice(&self.entries[..second_len]);
        }

        self.head = self.head.wrapping_add(to_consume as u32);
        to_consume
    }
}

/// RAII helper ensuring raw file descriptors are closed if initialization fails early.
struct AutoCloseFd(RawFd);

impl AutoCloseFd {
    fn into_raw(mut self) -> RawFd {
        let fd = self.0;
        self.0 = -1;
        fd
    }
}

impl Drop for AutoCloseFd {
    fn drop(&mut self) {
        if self.0 >= 0 {
            // SAFETY: `self.0` is an open file descriptor owned by `AutoCloseFd`.
            // Closing it releases OS resources upon early return or failure.
            unsafe {
                libc::close(self.0);
            }
        }
    }
}

/// AF_XDP socket with three-tier opportunistic fallback engine.
pub struct XskSocket {
    fd: RawFd,
    mode: XskBindMode,
    rx_ring: RxRing,
    tx_ring: TxRing,
    filter: Option<crate::bpf::XdpRedirectFilter>,
}

impl XskSocket {
    /// Attempts opportunistic binding of an AF_XDP socket.
    ///
    /// Fallback order:
    /// 1. Zero-Copy mode (`XskBindMode::ZeroCopy`)
    /// 2. Copy mode (`XskBindMode::Copy`)
    /// 3. Portable recvmmsg fallback (`XskBindMode::FallbackRecvmmsg`)
    ///
    /// If running unprivileged or kernel lacks AF_XDP, cleanly returns `FallbackRecvmmsg`.
    pub fn bind_opportunistic(ifname: &str, queue_id: u32, umem: &UmemPool) -> io::Result<Self> {
        // SAFETY: Calling `socket(AF_XDP, SOCK_RAW, 0)` requests a raw AF_XDP socket from the kernel.
        // It requires no initialized pointers or preconditions.
        let fd = unsafe { libc::socket(libc::AF_XDP, libc::SOCK_RAW, 0) };
        if fd < 0 {
            let err = io::Error::last_os_error();
            match err.raw_os_error() {
                Some(
                    libc::EPERM
                    | libc::EACCES
                    | libc::EAFNOSUPPORT
                    | libc::ENOPROTOOPT
                    | libc::EPROTONOSUPPORT,
                ) => return Ok(Self::fallback()),
                _ => return Ok(Self::fallback()),
            }
        }

        let sock_guard = AutoCloseFd(fd);

        // 1. Configure UMEM registration
        let mr = libc::xdp_umem_reg {
            addr: umem.area_ptr() as u64,
            len: umem.size() as u64,
            chunk_size: umem.chunk_size() as u32,
            headroom: 0,
            flags: 0,
            tx_metadata_len: 0,
        };

        // SAFETY: `fd` is a valid open AF_XDP socket. `mr` is a stack-local `xdp_umem_reg`
        // initialized with valid UMEM memory bounds and chunk parameters. The memory region
        // is guaranteed valid by `umem`. Size passed matches `sizeof(xdp_umem_reg)`.
        let ret_umem = unsafe {
            libc::setsockopt(
                fd,
                libc::SOL_XDP,
                libc::XDP_UMEM_REG,
                std::ptr::addr_of!(mr).cast::<libc::c_void>(),
                std::mem::size_of::<libc::xdp_umem_reg>() as libc::socklen_t,
            )
        };
        if ret_umem != 0 {
            return Ok(Self::fallback());
        }

        let ring_size: u32 = UMEM_RING_SIZE;

        // 2. Configure Fill Ring
        // SAFETY: `fd` is an open AF_XDP socket. `ring_size` is a valid stack-local u32.
        // Size passed matches `sizeof(u32)`.
        let ret_fill = unsafe {
            libc::setsockopt(
                fd,
                libc::SOL_XDP,
                libc::XDP_UMEM_FILL_RING,
                std::ptr::addr_of!(ring_size).cast::<libc::c_void>(),
                std::mem::size_of::<u32>() as libc::socklen_t,
            )
        };
        if ret_fill != 0 {
            return Ok(Self::fallback());
        }

        // 3. Configure Completion Ring
        // SAFETY: `fd` is an open AF_XDP socket. `ring_size` is a valid stack-local u32.
        // Size passed matches `sizeof(u32)`.
        let ret_comp = unsafe {
            libc::setsockopt(
                fd,
                libc::SOL_XDP,
                libc::XDP_UMEM_COMPLETION_RING,
                std::ptr::addr_of!(ring_size).cast::<libc::c_void>(),
                std::mem::size_of::<u32>() as libc::socklen_t,
            )
        };
        if ret_comp != 0 {
            return Ok(Self::fallback());
        }

        // 4. Configure RX Ring
        // SAFETY: `fd` is an open AF_XDP socket. `ring_size` is a valid stack-local u32.
        // Size passed matches `sizeof(u32)`.
        let ret_rx = unsafe {
            libc::setsockopt(
                fd,
                libc::SOL_XDP,
                libc::XDP_RX_RING,
                std::ptr::addr_of!(ring_size).cast::<libc::c_void>(),
                std::mem::size_of::<u32>() as libc::socklen_t,
            )
        };
        if ret_rx != 0 {
            return Ok(Self::fallback());
        }

        // 5. Configure TX Ring
        // SAFETY: `fd` is an open AF_XDP socket. `ring_size` is a valid stack-local u32.
        // Size passed matches `sizeof(u32)`.
        let ret_tx = unsafe {
            libc::setsockopt(
                fd,
                libc::SOL_XDP,
                libc::XDP_TX_RING,
                std::ptr::addr_of!(ring_size).cast::<libc::c_void>(),
                std::mem::size_of::<u32>() as libc::socklen_t,
            )
        };
        if ret_tx != 0 {
            return Ok(Self::fallback());
        }

        // 6. Resolve network interface index
        let c_ifname = match CString::new(ifname) {
            Ok(c) => c,
            Err(_) => return Ok(Self::fallback()),
        };

        // SAFETY: `c_ifname` is a valid null-terminated C string.
        let ifindex = unsafe { libc::if_nametoindex(c_ifname.as_ptr()) };
        if ifindex == 0 {
            return Ok(Self::fallback());
        }

        // 7. Attempt bind with XDP_ZERO_COPY
        let mut sxdp: libc::sockaddr_xdp = unsafe {
            // SAFETY: zeroing a C struct with POD fields is safe.
            std::mem::zeroed()
        };
        sxdp.sxdp_family = libc::AF_XDP as u16;
        sxdp.sxdp_ifindex = ifindex;
        sxdp.sxdp_queue_id = queue_id;
        sxdp.sxdp_shared_umem_fd = 0;
        sxdp.sxdp_flags = XDP_ZERO_COPY;

        // SAFETY: `fd` is an open AF_XDP socket. `sxdp` is a valid stack-local sockaddr_xdp.
        // Size passed matches `sizeof(sockaddr_xdp)`.
        let ret_zc = unsafe {
            libc::bind(
                fd,
                std::ptr::addr_of!(sxdp).cast::<libc::sockaddr>(),
                std::mem::size_of::<libc::sockaddr_xdp>() as libc::socklen_t,
            )
        };

        if ret_zc == 0 {
            let fd = sock_guard.into_raw();
            return Ok(Self {
                fd,
                mode: XskBindMode::ZeroCopy,
                rx_ring: RxRing::new(UMEM_RING_SIZE),
                tx_ring: TxRing::new(UMEM_RING_SIZE),
                filter: None,
            });
        }

        // 8. Attempt bind with XDP_COPY if ZeroCopy returned unsupported error
        let zc_err = io::Error::last_os_error();
        let should_try_copy = matches!(
            zc_err.raw_os_error(),
            Some(libc::EOPNOTSUPP | ENOTSUPP | libc::EINVAL)
        );

        if should_try_copy {
            sxdp.sxdp_flags = XDP_COPY;
            // SAFETY: `fd` is an open AF_XDP socket. `sxdp` is a valid stack-local sockaddr_xdp.
            // Size passed matches `sizeof(sockaddr_xdp)`.
            let ret_copy = unsafe {
                libc::bind(
                    fd,
                    std::ptr::addr_of!(sxdp).cast::<libc::sockaddr>(),
                    std::mem::size_of::<libc::sockaddr_xdp>() as libc::socklen_t,
                )
            };

            if ret_copy == 0 {
                let fd = sock_guard.into_raw();
                return Ok(Self {
                    fd,
                    mode: XskBindMode::Copy,
                    rx_ring: RxRing::new(UMEM_RING_SIZE),
                    tx_ring: TxRing::new(UMEM_RING_SIZE),
                    filter: None,
                });
            }
        }

        // All AF_XDP attempts failed; fallback to portable recvmmsg mode
        Ok(Self::fallback())
    }

    /// Creates a fallback socket instance in `FallbackRecvmmsg` mode without an active AF_XDP socket.
    pub fn fallback() -> Self {
        Self {
            fd: -1,
            mode: XskBindMode::FallbackRecvmmsg,
            rx_ring: RxRing::new(UMEM_RING_SIZE),
            tx_ring: TxRing::new(UMEM_RING_SIZE),
            filter: None,
        }
    }

    /// The active bind mode of this socket (ZeroCopy, Copy, or FallbackRecvmmsg).
    pub fn mode(&self) -> XskBindMode {
        self.mode
    }

    /// The raw file descriptor of the underlying AF_XDP socket, or -1 in fallback mode.
    pub fn fd(&self) -> RawFd {
        self.fd
    }

    /// Mutable reference to the socket's RX ring buffer.
    pub fn rx_ring_mut(&mut self) -> &mut RxRing {
        &mut self.rx_ring
    }

    /// Mutable reference to the socket's TX ring buffer.
    pub fn tx_ring_mut(&mut self) -> &mut TxRing {
        &mut self.tx_ring
    }

    /// Immutable reference to the socket's RX ring buffer.
    pub fn rx_ring(&self) -> &RxRing {
        &self.rx_ring
    }

    /// Immutable reference to the socket's TX ring buffer.
    pub fn tx_ring(&self) -> &TxRing {
        &self.tx_ring
    }

    /// Reference to the attached eBPF redirect filter, if active.
    pub fn filter(&self) -> Option<&crate::bpf::XdpRedirectFilter> {
        self.filter.as_ref()
    }

    /// Mutable reference to the attached eBPF redirect filter, if active.
    pub fn filter_mut(&mut self) -> Option<&mut crate::bpf::XdpRedirectFilter> {
        self.filter.as_mut()
    }

    /// Opportunistically attaches an eBPF XDP redirect filter steering incoming UDP traffic
    /// for `listen_port` on `ifname` into this socket's RX ring.
    pub fn attach_bpf_filter(
        &mut self,
        ifname: &str,
        listen_port: u16,
        queue_id: u32,
    ) -> crate::bpf::BpfFilterStatus {
        if self.fd < 0 {
            return crate::bpf::BpfFilterStatus::Unsupported;
        }
        match crate::bpf::XdpRedirectFilter::load_and_attach(ifname, listen_port, queue_id, self.fd)
        {
            Ok(f) => {
                let prog_fd = f.prog_fd();
                self.filter = Some(f);
                crate::bpf::BpfFilterStatus::Attached(prog_fd)
            }
            Err(status) => status,
        }
    }

    /// Opportunistically attaches a multi-queue eBPF XDP redirect filter steering incoming UDP traffic
    /// for `listen_port` on `ifname` across multiple RX queues into core-pinned AF_XDP rings.
    pub fn attach_bpf_filter_multi_queue(
        &mut self,
        ifname: &str,
        listen_port: u16,
        queue_fds: &[(u32, RawFd)],
    ) -> crate::bpf::BpfFilterStatus {
        if self.fd < 0 {
            return crate::bpf::BpfFilterStatus::Unsupported;
        }
        match crate::bpf::XdpRedirectFilter::load_multi_queue(ifname, listen_port, queue_fds) {
            Ok(f) => {
                let prog_fd = f.prog_fd();
                self.filter = Some(f);
                crate::bpf::BpfFilterStatus::Attached(prog_fd)
            }
            Err(status) => status,
        }
    }

    /// Sets or updates the socket file descriptor for a specific queue ID in the attached BPF filter.
    pub fn set_bpf_filter_queue_socket(&mut self, queue_id: u32, xsk_fd: RawFd) -> io::Result<()> {
        match self.filter.as_mut() {
            Some(filter) => filter.set_socket_for_queue(queue_id, xsk_fd),
            None => Err(io::Error::new(
                io::ErrorKind::NotFound,
                "no BPF filter attached to socket",
            )),
        }
    }

    /// Binds multiple AF_XDP sockets across `queue_ids` for the given interface and UMEM pool.
    ///
    /// Returns a list of `(queue_id, XskSocket)` pairs. If binding fails for a queue,
    /// a socket in fallback mode is returned for that queue.
    pub fn bind_multi_queue(
        ifname: &str,
        queue_ids: &[u32],
        umem: &UmemPool,
    ) -> io::Result<Vec<(u32, Self)>> {
        let mut sockets = Vec::with_capacity(queue_ids.len());
        for &qid in queue_ids {
            let sock = Self::bind_opportunistic(ifname, qid, umem)?;
            sockets.push((qid, sock));
        }
        Ok(sockets)
    }
}

impl Drop for XskSocket {
    fn drop(&mut self) {
        if self.fd >= 0 {
            // SAFETY: `self.fd` is an open AF_XDP socket descriptor exclusively owned by
            // this `XskSocket`. Closing it releases the kernel socket resources.
            unsafe {
                libc::close(self.fd);
            }
            self.fd = -1;
        }
    }
}