ferranet 0.2.0

//! AF_XDP backend (feature `xdp`): the kernel-bypass tier above `AF_PACKET`.
//!
//! An [`XdpSocket`] owns a UMEM (a contiguous frame buffer shared with the kernel) and the four
//! single-producer/single-consumer rings AF_XDP uses:
//!
//! * **fill** — user hands the kernel free UMEM chunks to receive into,
//! * **rx** — kernel hands the user received-frame descriptors (zero-copy),
//! * **tx** — user hands the kernel descriptors to transmit,
//! * **completion** — kernel hands back transmitted chunks for reuse.
//!
//! Receive reuses the same zero-copy [`Block`]/[`Frame`] surface as the `AF_PACKET` ring; transmit
//! implements [`RawChannel`](crate::sys::RawChannel), so AF_XDP slots into the same abstraction
//! boundary.
//!
//! # Prerequisite: an XDP redirect program
//!
//! Unlike `AF_PACKET`, an AF_XDP socket only receives frames that an **XDP program** redirects into
//! an `XSKMAP` for its `(ifindex, queue_id)` — the kernel ships no default. `ferranet` builds and
//! drives the socket and rings; attaching the redirect program is left to the operator or a higher
//! layer (e.g. `libxdp`/`aya`). Without one, the socket binds but `recv` stays empty. This mirrors
//! how `libxdp` and `xsk-rs` separate the *socket* from the *program*.
//!
//! This module is the crate's second concentration of `unsafe`; each block carries a `SAFETY`
//! note. The ring index arithmetic and UMEM chunking are pure and unit-tested; end-to-end
//! validation requires a real NIC/queue (or `veth` in generic XDP mode) with a redirect program
//! attached.

use std::ffi::{c_int, c_void};
use std::io;
use std::mem;
use std::os::fd::{AsFd, AsRawFd, BorrowedFd, FromRawFd, OwnedFd, RawFd};
use std::ptr::{self, NonNull};
use std::sync::atomic::{AtomicU32, Ordering};

use crate::block::{Block, Frame, FrameMeta, PacketType};
use crate::error::{Error, Result};
use crate::interface::IfIndex;
use crate::sys::Stats;

/// Geometry and mode for an [`XdpSocket`].
///
/// All ring sizes and the frame count must be powers of two. The defaults (a 4096 × 2 KiB UMEM
/// with 2048-entry rings, copy mode, need-wakeup) are broadly portable; request
/// [`zero_copy`](XdpConfig::zero_copy) on drivers that support it for the full kernel-bypass path.
#[derive(Debug, Clone, Copy)]
pub struct XdpConfig {
    /// The NIC receive queue to bind to.
    pub queue_id: u32,
    /// Number of UMEM chunks (frames). Power of two.
    pub frame_count: u32,
    /// Size of each UMEM chunk in bytes (typically 2048 or 4096). Power of two.
    pub frame_size: u32,
    /// Fill-ring entries. Power of two.
    pub fill_size: u32,
    /// Completion-ring entries. Power of two.
    pub comp_size: u32,
    /// RX-ring entries. Power of two.
    pub rx_size: u32,
    /// TX-ring entries. Power of two.
    pub tx_size: u32,
    /// Request zero-copy mode (`XDP_ZEROCOPY`); falls back is *not* automatic — binding fails if
    /// the driver lacks support. When `false`, `XDP_COPY` is requested.
    pub zero_copy: bool,
    /// Enable `XDP_USE_NEED_WAKEUP`, letting the kernel signal when a `sendto`/`poll` is needed.
    pub need_wakeup: bool,
}

impl Default for XdpConfig {
    fn default() -> Self {
        XdpConfig {
            queue_id: 0,
            frame_count: 4096,
            frame_size: 2048,
            fill_size: 2048,
            comp_size: 2048,
            rx_size: 2048,
            tx_size: 2048,
            zero_copy: false,
            need_wakeup: true,
        }
    }
}

impl XdpConfig {
    fn validate(&self) -> Result<()> {
        let pow2 = |v: u32| v != 0 && v.is_power_of_two();
        if !pow2(self.frame_count) || !pow2(self.frame_size) {
            return Err(Error::invalid_config("frame_count and frame_size must be powers of two"));
        }
        if self.frame_size < 2048 {
            return Err(Error::invalid_config("frame_size must be at least 2048"));
        }
        if !pow2(self.fill_size) || !pow2(self.comp_size) || !pow2(self.rx_size) || !pow2(self.tx_size)
        {
            return Err(Error::invalid_config("all ring sizes must be powers of two"));
        }
        if self.fill_size > self.frame_count {
            return Err(Error::invalid_config("fill_size cannot exceed frame_count"));
        }
        Ok(())
    }

    fn umem_len(&self) -> usize {
        self.frame_count as usize * self.frame_size as usize
    }

    fn bind_flags(&self) -> u16 {
        let mut f = if self.zero_copy { libc::XDP_ZEROCOPY } else { libc::XDP_COPY };
        if self.need_wakeup {
            f |= libc::XDP_USE_NEED_WAKEUP;
        }
        f
    }
}

/// An owned `mmap` region, unmapped on drop.
#[derive(Debug)]
struct Mmap {
    ptr: NonNull<u8>,
    len: usize,
}

impl Mmap {
    /// Maps `len` bytes of `fd` at `offset` (for the AF_XDP ring regions).
    fn ring(fd: BorrowedFd<'_>, len: usize, offset: libc::off_t) -> Result<Self> {
        // SAFETY: standard mmap of a kernel ring region; `len` is computed from negotiated
        // offsets and ring sizes, `offset` is a documented XDP ring page-offset constant.
        let raw = unsafe {
            libc::mmap(
                ptr::null_mut(),
                len,
                libc::PROT_READ | libc::PROT_WRITE,
                libc::MAP_SHARED | libc::MAP_POPULATE,
                fd.as_raw_fd(),
                offset,
            )
        };
        Self::wrap(raw, len)
    }

    /// Allocates an anonymous, page-aligned region for the UMEM.
    fn anon(len: usize) -> Result<Self> {
        // SAFETY: anonymous private mapping; no fd, offset 0. `len` is non-zero (validated).
        let raw = unsafe {
            libc::mmap(
                ptr::null_mut(),
                len,
                libc::PROT_READ | libc::PROT_WRITE,
                libc::MAP_PRIVATE | libc::MAP_ANONYMOUS,
                -1,
                0,
            )
        };
        Self::wrap(raw, len)
    }

    fn wrap(raw: *mut c_void, len: usize) -> Result<Self> {
        if raw == libc::MAP_FAILED {
            return Err(Error::Mmap(io::Error::last_os_error()));
        }
        // SAFETY: a non-MAP_FAILED return from mmap is non-null.
        Ok(Mmap { ptr: unsafe { NonNull::new_unchecked(raw.cast::<u8>()) }, len })
    }

    fn base(&self) -> *mut u8 {
        self.ptr.as_ptr()
    }
}

impl Drop for Mmap {
    fn drop(&mut self) {
        // SAFETY: `ptr`/`len` are exactly the mapping created above, unmapped once.
        unsafe {
            libc::munmap(self.ptr.as_ptr().cast::<c_void>(), self.len);
        }
    }
}

/// A single-producer/single-consumer AF_XDP ring over an mmap'd region.
///
/// `producer`/`consumer` are the kernel-shared head/tail counters; `desc` is the base of the
/// descriptor array (entries of `u64` for fill/completion, `xdp_desc` for rx/tx). All wrap at
/// `mask + 1` entries.
#[derive(Debug)]
struct Ring {
    producer: *const AtomicU32,
    consumer: *const AtomicU32,
    desc: *mut u8,
    mask: u32,
    cached_producer: u32,
    cached_consumer: u32,
    _map: Mmap,
}

impl Ring {
    fn new(map: Mmap, off: &libc::xdp_ring_offset, size: u32) -> Ring {
        let base = map.base();
        // SAFETY: the kernel guarantees these byte offsets lie within the mapped ring region.
        let (producer, consumer, desc) = unsafe {
            (
                base.add(off.producer as usize).cast::<AtomicU32>().cast_const(),
                base.add(off.consumer as usize).cast::<AtomicU32>().cast_const(),
                base.add(off.desc as usize),
            )
        };
        Ring { producer, consumer, desc, mask: size - 1, cached_producer: 0, cached_consumer: 0, _map: map }
    }

    fn prod(&self) -> &AtomicU32 {
        // SAFETY: `producer` points to a live AtomicU32 in the mapping, valid for the ring's life.
        unsafe { &*self.producer }
    }

    fn cons(&self) -> &AtomicU32 {
        // SAFETY: `consumer` points to a live AtomicU32 in the mapping, valid for the ring's life.
        unsafe { &*self.consumer }
    }

    fn entry<T>(&self, idx: u32) -> *mut T {
        let slot = (idx & self.mask) as usize;
        // SAFETY: `slot < size`; the desc array holds `size` entries of `T`.
        unsafe { self.desc.add(slot * mem::size_of::<T>()).cast::<T>() }
    }

    /// Reserves space to produce `n` entries, returning how many are actually free.
    fn reserve(&mut self, n: u32) -> u32 {
        let size = self.mask + 1;
        let mut free = size - (self.cached_producer.wrapping_sub(self.cached_consumer));
        if free < n {
            self.cached_consumer = self.cons().load(Ordering::Acquire);
            free = size - (self.cached_producer.wrapping_sub(self.cached_consumer));
        }
        free.min(n)
    }

    /// Publishes `n` produced entries to the kernel.
    fn submit(&mut self, n: u32) {
        self.cached_producer = self.cached_producer.wrapping_add(n);
        self.prod().store(self.cached_producer, Ordering::Release);
    }

    /// Returns how many entries are available to consume (up to `n`).
    fn available(&mut self, n: u32) -> u32 {
        let mut avail = self.cached_producer.wrapping_sub(self.cached_consumer);
        if avail < n {
            self.cached_producer = self.prod().load(Ordering::Acquire);
            avail = self.cached_producer.wrapping_sub(self.cached_consumer);
        }
        avail.min(n)
    }

    /// Marks `n` consumed entries as released back to the kernel.
    fn release(&mut self, n: u32) {
        self.cached_consumer = self.cached_consumer.wrapping_add(n);
        self.cons().store(self.cached_consumer, Ordering::Release);
    }
}

// SAFETY: an XdpSocket exclusively owns its mappings and fd; the raw pointers address that
// owned, stable memory. Concurrent access is mediated by the kernel ring protocol, not by Rust
// aliasing of this handle, so moving it between threads is sound.
unsafe impl Send for XdpSocket {}

/// An AF_XDP socket: UMEM + fill/completion/rx/tx rings bound to one interface queue.
pub struct XdpSocket {
    fd: OwnedFd,
    umem: Mmap,
    fill: Ring,
    comp: Ring,
    rx: Ring,
    tx: Ring,
    /// Free UMEM chunk addresses available for transmit.
    free: Vec<u64>,
    config: XdpConfig,
    ifindex: IfIndex,
}

impl XdpSocket {
    /// Opens an AF_XDP socket, registers a UMEM, sets up the four rings, and binds to
    /// `(ifindex, config.queue_id)`.
    pub fn open(ifindex: IfIndex, config: XdpConfig) -> Result<Self> {
        config.validate()?;

        // SAFETY: a plain socket(2) call with valid constants; the fd is immediately owned.
        let raw = unsafe { libc::socket(libc::AF_XDP, libc::SOCK_RAW, 0) };
        if raw < 0 {
            return Err(Error::Socket(io::Error::last_os_error()));
        }
        // SAFETY: `raw` is a fresh valid owned fd from socket(2).
        let fd = unsafe { OwnedFd::from_raw_fd(raw) };

        let umem = Mmap::anon(config.umem_len())?;

        let reg = libc::xdp_umem_reg {
            addr: umem.base() as u64,
            len: config.umem_len() as u64,
            chunk_size: config.frame_size,
            headroom: 0,
            flags: 0,
            tx_metadata_len: 0,
        };
        // SAFETY: `reg` is a fully-initialised xdp_umem_reg, the struct XDP_UMEM_REG expects.
        unsafe { setsockopt(fd.as_fd(), libc::XDP_UMEM_REG, &reg, "XDP_UMEM_REG")? };

        set_ring_size(fd.as_fd(), libc::XDP_UMEM_FILL_RING, config.fill_size, "XDP_UMEM_FILL_RING")?;
        set_ring_size(
            fd.as_fd(),
            libc::XDP_UMEM_COMPLETION_RING,
            config.comp_size,
            "XDP_UMEM_COMPLETION_RING",
        )?;
        set_ring_size(fd.as_fd(), libc::XDP_RX_RING, config.rx_size, "XDP_RX_RING")?;
        set_ring_size(fd.as_fd(), libc::XDP_TX_RING, config.tx_size, "XDP_TX_RING")?;

        let offsets = get_mmap_offsets(fd.as_fd())?;

        let fill = Ring::new(
            Mmap::ring(fd.as_fd(), ring_bytes(&offsets.fr, config.fill_size, 8), fill_ring_offset())?,
            &offsets.fr,
            config.fill_size,
        );
        let comp = Ring::new(
            Mmap::ring(fd.as_fd(), ring_bytes(&offsets.cr, config.comp_size, 8), comp_ring_offset())?,
            &offsets.cr,
            config.comp_size,
        );
        let desc_sz = mem::size_of::<libc::xdp_desc>();
        let rx = Ring::new(
            Mmap::ring(
                fd.as_fd(),
                ring_bytes(&offsets.rx, config.rx_size, desc_sz),
                libc::XDP_PGOFF_RX_RING,
            )?,
            &offsets.rx,
            config.rx_size,
        );
        let tx = Ring::new(
            Mmap::ring(
                fd.as_fd(),
                ring_bytes(&offsets.tx, config.tx_size, desc_sz),
                libc::XDP_PGOFF_TX_RING,
            )?,
            &offsets.tx,
            config.tx_size,
        );

        // Chunk addresses: hand the first `fill_size` to the kernel for RX, keep the rest free
        // for transmit.
        let free: Vec<u64> = (config.fill_size..config.frame_count)
            .map(|i| u64::from(i) * u64::from(config.frame_size))
            .collect();

        let mut sock =
            XdpSocket { fd, umem, fill, comp, rx, tx, free, config, ifindex };
        sock.fill_initial()?;
        sock.bind()?;
        Ok(sock)
    }

    /// Seeds the fill ring with the first `fill_size` chunks so the kernel has buffers to RX into.
    fn fill_initial(&mut self) -> Result<()> {
        let n = self.config.fill_size;
        let got = self.fill.reserve(n);
        if got != n {
            return Err(Error::invalid_config("could not seed the fill ring"));
        }
        for i in 0..n {
            let addr = u64::from(i) * u64::from(self.config.frame_size);
            // SAFETY: `entry` is in-bounds for the reserved slot; fill descriptors are u64 addrs.
            unsafe { ptr::write(self.fill.entry::<u64>(self.fill.cached_producer + i), addr) };
        }
        self.fill.submit(n);
        Ok(())
    }

    fn bind(&self) -> Result<()> {
        // SAFETY: sockaddr_xdp is plain-old-data; an all-zero value is a valid initial state.
        let mut addr: libc::sockaddr_xdp = unsafe { mem::zeroed() };
        addr.sxdp_family = libc::AF_XDP as u16;
        addr.sxdp_ifindex = self.ifindex.0;
        addr.sxdp_queue_id = self.config.queue_id;
        addr.sxdp_flags = self.config.bind_flags();
        // SAFETY: `addr` is a fully-initialised sockaddr_xdp passed with its true size.
        let rc = unsafe {
            libc::bind(
                self.fd.as_raw_fd(),
                (&raw const addr).cast::<libc::sockaddr>(),
                mem::size_of::<libc::sockaddr_xdp>() as libc::socklen_t,
            )
        };
        if rc < 0 {
            return Err(Error::Bind(io::Error::last_os_error()));
        }
        Ok(())
    }

    /// Receives the currently-available batch of frames (non-blocking).
    ///
    /// The returned [`Block`] borrows frame bytes directly from the UMEM; dropping it returns those
    /// chunks to the fill ring for the kernel to reuse. Use [`XdpSocket::as_fd`] with `poll`/
    /// `AsyncFd` to wait for readiness. Returns an empty block when nothing is ready.
    pub fn recv(&mut self) -> Result<Block<'_>> {
        let batch = self.config.rx_size;
        let n = self.rx.available(batch);

        let mut frames = Vec::with_capacity(n as usize);
        let umem_base: *const u8 = self.umem.base();
        for i in 0..n {
            // SAFETY: `i < n <= available`; rx entries are xdp_desc.
            let desc = unsafe { ptr::read(self.rx.entry::<libc::xdp_desc>(self.rx.cached_consumer + i)) };
            // Drop descriptors that don't fit the UMEM instead of slicing out of bounds; the
            // chunk is deliberately not re-queued to the fill ring, since its address is exactly
            // the value we couldn't trust.
            if desc_within_umem(desc.addr, desc.len, self.config.umem_len()) {
                frames.push((desc.addr, desc.len));
            }
        }
        if n > 0 {
            self.rx.release(n);
        }

        Ok(Block::from_xdp(XdpBlock { frames, umem_base, fill: &mut self.fill }))
    }

    /// Borrows the socket file descriptor for readiness integration.
    pub fn as_fd(&self) -> BorrowedFd<'_> {
        self.fd.as_fd()
    }

    /// Reclaims transmitted chunks from the completion ring back onto the free list.
    fn reclaim_completions(&mut self) {
        let n = self.comp.available(self.config.comp_size);
        for i in 0..n {
            // SAFETY: `i < n <= available`; completion entries are u64 addrs.
            let addr = unsafe { ptr::read(self.comp.entry::<u64>(self.comp.cached_consumer + i)) };
            self.free.push(addr);
        }
        if n > 0 {
            self.comp.release(n);
        }
    }

    /// Kicks the kernel to process the TX ring when need-wakeup indicates it is necessary.
    fn kick_tx(&self) -> io::Result<()> {
        // SAFETY: a zero-length sendto on the xsk fd kicks transmit; null buffer with len 0.
        let rc = unsafe {
            libc::sendto(self.fd.as_raw_fd(), ptr::null(), 0, libc::MSG_DONTWAIT, ptr::null(), 0)
        };
        if rc < 0 {
            let err = io::Error::last_os_error();
            // EAGAIN/EBUSY just mean the kernel is already busy draining; not fatal.
            if matches!(err.raw_os_error(), Some(libc::EAGAIN | libc::EBUSY | libc::ENOBUFS)) {
                return Ok(());
            }
            return Err(err);
        }
        Ok(())
    }

    /// Reads AF_XDP drop/error statistics.
    pub fn stats(&self) -> io::Result<Stats> {
        // SAFETY: xdp_statistics is plain-old-data; an all-zero value is a valid initial state.
        let mut s: libc::xdp_statistics = unsafe { mem::zeroed() };
        let mut len = mem::size_of::<libc::xdp_statistics>() as libc::socklen_t;
        // SAFETY: `s`/`len` are valid storage and its length for getsockopt to fill.
        let rc = unsafe {
            libc::getsockopt(
                self.fd.as_raw_fd(),
                libc::SOL_XDP,
                libc::XDP_STATISTICS,
                (&raw mut s).cast(),
                &raw mut len,
            )
        };
        if rc < 0 {
            return Err(io::Error::last_os_error());
        }
        Ok(Stats { received: 0, dropped: s.rx_dropped, freezes: 0 })
    }

    fn frame_capacity(&self) -> usize {
        self.config.frame_size as usize
    }

    /// Sends a single frame by copying it into a free UMEM chunk and kicking the TX ring.
    ///
    /// Non-blocking: returns `WouldBlock` when no free chunk or TX descriptor is available (wait
    /// for writability on [`XdpSocket::as_fd`] and retry). Completed transmissions are reclaimed
    /// opportunistically on each call.
    pub fn send(&mut self, frame: &[u8]) -> io::Result<usize> {
        if frame.len() > self.frame_capacity() {
            return Err(io::Error::new(io::ErrorKind::InvalidInput, "frame exceeds UMEM chunk size"));
        }
        self.reclaim_completions();
        let addr = match self.free.pop() {
            Some(a) => a,
            None => return Err(io::Error::from(io::ErrorKind::WouldBlock)),
        };
        if self.tx.reserve(1) != 1 {
            self.free.push(addr);
            return Err(io::Error::from(io::ErrorKind::WouldBlock));
        }
        // SAFETY: `addr` is a valid chunk offset within the UMEM and `frame.len() <= chunk_size`,
        // so the copy stays inside the chunk.
        unsafe {
            ptr::copy_nonoverlapping(frame.as_ptr(), self.umem.base().add(addr as usize), frame.len());
            ptr::write(
                self.tx.entry::<libc::xdp_desc>(self.tx.cached_producer),
                libc::xdp_desc { addr, len: frame.len() as u32, options: 0 },
            );
        }
        self.tx.submit(1);
        self.kick_tx()?;
        Ok(frame.len())
    }

    /// Sends a batch of frames, returning how many were accepted before the UMEM or TX ring
    /// filled up (at least one, or `WouldBlock`).
    pub fn send_batch(&mut self, frames: &[&[u8]]) -> io::Result<usize> {
        let mut sent = 0;
        for frame in frames {
            match self.send(frame) {
                Ok(_) => sent += 1,
                Err(e) if sent > 0 && e.kind() == io::ErrorKind::WouldBlock => break,
                Err(e) => return Err(e),
            }
        }
        Ok(sent)
    }
}

impl crate::sys::RawChannel for XdpSocket {
    fn send(&mut self, frame: &[u8]) -> io::Result<usize> {
        XdpSocket::send(self, frame)
    }

    fn send_batch(&mut self, frames: &[&[u8]]) -> io::Result<usize> {
        XdpSocket::send_batch(self, frames)
    }
}

impl AsFd for XdpSocket {
    fn as_fd(&self) -> BorrowedFd<'_> {
        self.fd.as_fd()
    }
}

impl AsRawFd for XdpSocket {
    fn as_raw_fd(&self) -> RawFd {
        self.fd.as_raw_fd()
    }
}

/// A batch of received frames borrowed from the UMEM. Dropping it returns the chunks to the fill
/// ring so the kernel can receive into them again.
#[derive(Debug)]
pub struct XdpBlock<'a> {
    frames: Vec<(u64, u32)>,
    umem_base: *const u8,
    fill: &'a mut Ring,
}

impl<'a> XdpBlock<'a> {
    /// Number of frames in this block.
    pub fn frame_count(&self) -> usize {
        self.frames.len()
    }

    /// Iterates the frames, each borrowing zero-copy from the UMEM.
    pub fn frames(&self) -> XdpFrames<'_> {
        XdpFrames { iter: self.frames.iter(), umem_base: self.umem_base }
    }
}

impl Drop for XdpBlock<'_> {
    fn drop(&mut self) {
        // Return the consumed chunks to the fill ring for reuse.
        let n = self.frames.len() as u32;
        if n == 0 {
            return;
        }
        let got = self.fill.reserve(n);
        // With one socket per UMEM and the fill ring seeded to capacity, every received frame
        // implies a consumed fill entry, so space for all `n` chunks is always available. If a
        // future change breaks that invariant, the unreturned chunks leak and RX slowly starves —
        // catch it loudly rather than degrade silently.
        debug_assert_eq!(got, n, "fill ring out of space; leaking {} UMEM chunks", n - got);
        for (i, (addr, _)) in self.frames.iter().take(got as usize).enumerate() {
            // RX descriptor addresses may include the frame's data offset within its chunk; the
            // kernel masks fill-ring addresses down to the chunk base itself in aligned mode.
            // SAFETY: slot is within the reserved range; fill entries are u64 addrs.
            unsafe { ptr::write(self.fill.entry::<u64>(self.fill.cached_producer + i as u32), *addr) };
        }
        self.fill.submit(got);
    }
}

/// Iterator over the frames of an [`XdpBlock`].
#[derive(Debug)]
pub struct XdpFrames<'a> {
    iter: std::slice::Iter<'a, (u64, u32)>,
    umem_base: *const u8,
}

impl<'a> Iterator for XdpFrames<'a> {
    type Item = Frame<'a>;

    fn next(&mut self) -> Option<Frame<'a>> {
        let &(addr, len) = self.iter.next()?;
        // SAFETY: `recv` checked `(addr, len)` against the UMEM length (`desc_within_umem`)
        // before storing it, so the range is within the mapping; the bytes stay valid until this
        // block is dropped (chunks not yet returned to the fill ring).
        let data = unsafe { std::slice::from_raw_parts(self.umem_base.add(addr as usize), len as usize) };
        let packet_type = data
            .get(0..6)
            .map(|b| PacketType::from_dest_mac(crate::interface::MacAddr([b[0], b[1], b[2], b[3], b[4], b[5]])))
            .unwrap_or(PacketType::Other(0));
        let meta = FrameMeta { wire_len: len as usize, timestamp: None, vlan: None, packet_type };
        Some(Frame::new(data, meta))
    }

    fn size_hint(&self) -> (usize, Option<usize>) {
        self.iter.size_hint()
    }
}

// ---- setsockopt / getsockopt helpers ------------------------------------------------------------

/// # Safety
/// `T` must be the POD request struct the kernel expects for `opt` under `SOL_XDP`.
unsafe fn setsockopt<T>(fd: BorrowedFd<'_>, opt: c_int, val: &T, name: &'static str) -> Result<()> {
    // SAFETY: by this function's contract `val` is a valid `T` of the size the kernel expects for
    // `opt`; we pass its real size. The fd is valid.
    let rc = unsafe {
        libc::setsockopt(
            fd.as_raw_fd(),
            libc::SOL_XDP,
            opt,
            (val as *const T).cast(),
            mem::size_of::<T>() as libc::socklen_t,
        )
    };
    if rc < 0 {
        return Err(Error::SetSockOpt { option: name, source: io::Error::last_os_error() });
    }
    Ok(())
}

fn set_ring_size(fd: BorrowedFd<'_>, opt: c_int, size: u32, name: &'static str) -> Result<()> {
    // SAFETY: the ring-size options take a single u32 descriptor count.
    unsafe { setsockopt(fd, opt, &size, name) }
}

fn get_mmap_offsets(fd: BorrowedFd<'_>) -> Result<libc::xdp_mmap_offsets> {
    // SAFETY: xdp_mmap_offsets is plain-old-data; an all-zero value is a valid initial state.
    let mut offsets: libc::xdp_mmap_offsets = unsafe { mem::zeroed() };
    let mut len = mem::size_of::<libc::xdp_mmap_offsets>() as libc::socklen_t;
    // SAFETY: `offsets`/`len` are valid storage and its length for getsockopt to fill.
    let rc = unsafe {
        libc::getsockopt(
            fd.as_raw_fd(),
            libc::SOL_XDP,
            libc::XDP_MMAP_OFFSETS,
            (&raw mut offsets).cast(),
            &raw mut len,
        )
    };
    if rc < 0 {
        return Err(Error::SetSockOpt {
            option: "XDP_MMAP_OFFSETS",
            source: io::Error::last_os_error(),
        });
    }
    Ok(offsets)
}

/// Total bytes to map for a ring: the descriptor-array offset plus `size` entries of `entry_size`.
fn ring_bytes(off: &libc::xdp_ring_offset, size: u32, entry_size: usize) -> usize {
    off.desc as usize + size as usize * entry_size
}

/// Whether an RX descriptor's `(addr, len)` range lies entirely within the UMEM.
///
/// RX descriptors are kernel-written and, in zero-copy mode, driver-written; like the `AF_PACKET`
/// block parser, we treat them as untrusted rather than let a buggy driver hand us a range that
/// would put a frame slice outside the mapping.
fn desc_within_umem(addr: u64, len: u32, umem_len: usize) -> bool {
    match addr.checked_add(u64::from(len)) {
        Some(end) => end <= umem_len as u64,
        None => false,
    }
}

const fn fill_ring_offset() -> libc::off_t {
    libc::XDP_UMEM_PGOFF_FILL_RING as libc::off_t
}

const fn comp_ring_offset() -> libc::off_t {
    libc::XDP_UMEM_PGOFF_COMPLETION_RING as libc::off_t
}

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn config_validation() {
        assert!(XdpConfig::default().validate().is_ok());
        assert!(XdpConfig { frame_count: 1000, ..Default::default() }.validate().is_err()); // not pow2
        assert!(XdpConfig { frame_size: 1024, ..Default::default() }.validate().is_err()); // < 2048
        assert!(XdpConfig { rx_size: 3, ..Default::default() }.validate().is_err()); // not pow2
        assert!(XdpConfig { fill_size: 8192, frame_count: 4096, ..Default::default() }
            .validate()
            .is_err()); // fill > frames
    }

    #[test]
    fn umem_len_and_chunks() {
        let c = XdpConfig { frame_count: 8, frame_size: 2048, ..Default::default() };
        assert_eq!(c.umem_len(), 8 * 2048);
        // chunk addresses are frame_size-aligned offsets.
        let addrs: Vec<u64> = (0..c.frame_count).map(|i| u64::from(i) * u64::from(c.frame_size)).collect();
        assert_eq!(addrs, vec![0, 2048, 4096, 6144, 8192, 10240, 12288, 14336]);
    }

    #[test]
    fn bind_flags_encode_mode() {
        let copy = XdpConfig { zero_copy: false, need_wakeup: false, ..Default::default() };
        assert_eq!(copy.bind_flags(), libc::XDP_COPY);
        let zc = XdpConfig { zero_copy: true, need_wakeup: true, ..Default::default() };
        assert_eq!(zc.bind_flags(), libc::XDP_ZEROCOPY | libc::XDP_USE_NEED_WAKEUP);
    }

    #[test]
    fn rx_descriptors_outside_umem_are_rejected() {
        let umem_len = 8usize * 2048;
        // In-bounds descriptors, including one ending exactly at the UMEM boundary.
        assert!(desc_within_umem(0, 100, umem_len));
        assert!(desc_within_umem((umem_len - 64) as u64, 64, umem_len));
        assert!(desc_within_umem(0, 0, umem_len));
        // Past the end, straddling the boundary, and u64 overflow.
        assert!(!desc_within_umem(umem_len as u64, 1, umem_len));
        assert!(!desc_within_umem((umem_len - 64) as u64, 65, umem_len));
        assert!(!desc_within_umem(u64::MAX, 1, umem_len));
        assert!(!desc_within_umem(u64::MAX, u32::MAX, umem_len));
    }

    #[test]
    #[cfg_attr(miri, ignore = "uses mmap; not interpretable by Miri")]
    fn ring_reserve_available_wraps() {
        // Build a standalone ring over a heap buffer to exercise the index math (no kernel).
        let size = 4u32;
        let entry = 8usize; // u64 entries
        // Layout: [producer u32][consumer u32][pad][desc array]
        let desc_off = 64usize;
        let map = Mmap::anon(desc_off + size as usize * entry).unwrap();
        let off = libc::xdp_ring_offset { producer: 0, consumer: 4, desc: desc_off as u64, flags: 8 };
        let mut ring = Ring::new(map, &off, size);

        // Initially empty: nothing to consume, full capacity to produce.
        assert_eq!(ring.available(size), 0);
        assert_eq!(ring.reserve(size), size);

        // Produce 3, then they become available to a consumer view.
        for i in 0..3 {
            // SAFETY: slot `i` is within the reserved range of this `size`-entry ring.
            unsafe { ptr::write(ring.entry::<u64>(ring.cached_producer + i), u64::from(i)) };
        }
        ring.submit(3);
        assert_eq!(ring.available(size), 3);
        // SAFETY: at least one entry is available to consume (we just produced 3).
        let v0 = unsafe { ptr::read(ring.entry::<u64>(ring.cached_consumer)) };
        assert_eq!(v0, 0);
        ring.release(3);
        assert_eq!(ring.available(size), 0);
    }
}