ferranet 0.2.0

//! `PACKET_MMAP` ring backends: a `TPACKET_V3` receive ring and a `TPACKET_V2` transmit ring.
//!
//! Because `PACKET_VERSION` is a per-socket setting, the RX (v3, block-based) and TX (v2) rings
//! live on *separate* sockets bound to the same interface. This maps naturally onto `ferranet`'s
//! split into a [`Sender`](crate::channel::Sender) and [`Receiver`](crate::channel::Receiver):
//! each half owns one socket, one mmap, and one file descriptor for readiness.
//!
//! This module contains essentially all of the crate's `unsafe`. Each block is justified with a
//! `SAFETY` comment; the invariants rest on the kernel's `TP_STATUS_*` ownership protocol, which
//! we observe with acquire/release atomics on the per-block / per-frame status word.

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

use crate::block::{Block, Frame, RawRingMeta};
use crate::error::{Error, Result};
use crate::sys::Stats;
use crate::sys::linux::socket::PacketSocket;
use crate::sys::linux::tpacket::{RingLayout, TPACKET_V2, TPACKET_V3, tpacket_align};

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

// SAFETY: the mapping is a stable address range owned exclusively by this value; sending it to
// another thread merely moves the pointer. Concurrent access is mediated by the kernel's status
// protocol, not by Rust aliasing of this handle.
unsafe impl Send for Mmap {}

impl Mmap {
    /// Maps `len` bytes of the packet ring backing `fd`.
    fn new(fd: BorrowedFd<'_>, len: usize) -> Result<Self> {
        // SAFETY: a standard mmap of a kernel-backed ring; `len` is non-zero and validated by the
        // ring layout. We pass offset 0 as the kernel requires for packet rings.
        let raw = unsafe {
            libc::mmap(
                ptr::null_mut(),
                len,
                libc::PROT_READ | libc::PROT_WRITE,
                libc::MAP_SHARED,
                fd.as_raw_fd(),
                0,
            )
        };
        if raw == libc::MAP_FAILED {
            return Err(Error::Mmap(io::Error::last_os_error()));
        }
        // SAFETY: mmap returned a non-MAP_FAILED pointer, which is therefore non-null.
        let ptr = unsafe { NonNull::new_unchecked(raw.cast::<u8>()) };
        Ok(Mmap { ptr, len })
    }

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

    /// Borrows the `len` bytes at `off` — deliberately *not* the whole mapping: a reference over
    /// the full region would assert immutability (and `dereferenceable`) over ring blocks the
    /// kernel is concurrently writing, so slices are only ever formed over a single USER-owned
    /// block.
    fn bytes(&self, off: usize, len: usize) -> &[u8] {
        assert!(off.checked_add(len).is_some_and(|end| end <= self.len));
        // SAFETY: `off..off + len` is within the mapping (checked above) and covers only a block
        // the caller observed as USER-owned via an acquire load, so the contents are stable for
        // the borrow of `&self`.
        unsafe { std::slice::from_raw_parts(self.ptr.as_ptr().add(off), len) }
    }
}

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

/// A `TPACKET_V3`, block-based, zero-copy receive ring.
#[derive(Debug)]
pub struct RxRing {
    sock: PacketSocket,
    mmap: Mmap,
    layout: RingLayout,
    /// Index of the next block we will inspect.
    current_block: usize,
}

impl RxRing {
    /// Builds a v3 RX ring on `sock` with the given geometry.
    pub fn open(sock: PacketSocket, layout: RingLayout) -> Result<Self> {
        sock.set_packet_opt_int(libc::PACKET_VERSION, "PACKET_VERSION", TPACKET_V3)?;
        let req = layout.to_req3();
        // SAFETY: `req` is a fully-initialised tpacket_req3, the struct PACKET_RX_RING expects.
        unsafe { sock.set_packet_opt(libc::PACKET_RX_RING, "PACKET_RX_RING", &req)? };
        let mmap = Mmap::new(sock.as_fd(), layout.map_len())?;
        Ok(RxRing { sock, mmap, layout, current_block: 0 })
    }

    /// Returns the next ready block, or `None` if the kernel has not finished one yet.
    ///
    /// The returned [`Block`] borrows directly from the mapped ring; dropping it releases the
    /// block back to the kernel and advances to the next one.
    pub fn recv_block(&mut self) -> Option<Block<'_>> {
        let Self { mmap, layout, current_block, .. } = self;
        let block_size = layout.block_size as usize;
        let block_off = *current_block * block_size;

        // The block descriptor begins each block; `block_status` is the first field of its header
        // union, located at `offset_of!(hdr)`.
        let status_off = mem::offset_of!(libc::tpacket_block_desc, hdr);
        // SAFETY: `block_off + status_off` is within the mapping (block 0..block_count), and the
        // pointer is 4-byte aligned because blocks are page-aligned. `from_ptr` borrows for the
        // mapping's lifetime, which outlives the returned block.
        let status: &AtomicU32 = unsafe {
            AtomicU32::from_ptr(mmap.base().add(block_off + status_off).cast::<u32>())
        };

        // Acquire so that, once we observe USER ownership, all frame writes are visible.
        let block_status = status.load(Ordering::Acquire);
        if block_status & libc::TP_STATUS_USER == 0 {
            return None;
        }
        // The kernel ORs TP_STATUS_LOSING into the v3 block status when packets were dropped since
        // the last PACKET_STATISTICS read — a zero-syscall "we are losing data" signal.
        let losing = block_status & libc::TP_STATUS_LOSING != 0;

        // SAFETY: the block is owned by us (USER); reading its v1 header is in-bounds and the
        // bytes are stable until we release it.
        let hdr = unsafe {
            ptr::read_unaligned(
                mmap.base().add(block_off + status_off).cast::<libc::tpacket_hdr_v1>(),
            )
        };

        let block_bytes = mmap.bytes(block_off, block_size);
        Some(Block::from_ring(RxBlock {
            status,
            block: block_bytes,
            num_pkts: hdr.num_pkts,
            first_off: hdr.offset_to_first_pkt as usize,
            cursor: current_block,
            block_count: layout.block_count as usize,
            losing,
        }))
    }

    /// Returns `true` if the current block has been handed to user space by the kernel.
    ///
    /// Lets a caller wait for readiness (via `poll`/`AsyncFd`) without holding the block borrow.
    pub fn block_ready(&self) -> bool {
        let block_off = self.current_block * self.layout.block_size as usize;
        let status_off = mem::offset_of!(libc::tpacket_block_desc, hdr);
        // SAFETY: identical bounds/alignment reasoning as `recv_block`; we only read the status.
        let status: &AtomicU32 = unsafe {
            AtomicU32::from_ptr(self.mmap.base().add(block_off + status_off).cast::<u32>())
        };
        status.load(Ordering::Acquire) & libc::TP_STATUS_USER != 0
    }

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

    /// Enables or disables promiscuous mode.
    pub fn set_promiscuous(&self, on: bool) -> Result<()> {
        self.sock.set_promiscuous(on)
    }

    /// Joins this ring's socket to a `PACKET_FANOUT` group.
    pub fn set_fanout(&self, group_id: u16, mode: u16) -> Result<()> {
        self.sock.set_fanout(group_id, mode)
    }

    /// Claims a kernel-allocated unique fanout group, returning its id (see
    /// [`PacketSocket::set_fanout_unique`]).
    pub fn set_fanout_unique(&self, mode: u16, fallback_id: u16) -> Result<u16> {
        self.sock.set_fanout_unique(mode, fallback_id)
    }

    /// Reads kernel receive/drop statistics.
    pub fn stats(&self) -> io::Result<Stats> {
        self.sock.statistics()
    }
}

/// A ready `TPACKET_V3` block, borrowed from the RX ring.
///
/// Dropping the block stores `TP_STATUS_KERNEL` to its status word (releasing it to the kernel)
/// and advances the ring cursor.
#[derive(Debug)]
pub struct RxBlock<'a> {
    status: &'a AtomicU32,
    block: &'a [u8],
    num_pkts: u32,
    first_off: usize,
    cursor: &'a mut usize,
    block_count: usize,
    losing: bool,
}

impl<'a> RxBlock<'a> {
    /// The number of frames in this block.
    pub fn frame_count(&self) -> usize {
        self.num_pkts as usize
    }

    /// Whether the kernel flagged packet drops (`TP_STATUS_LOSING`) for this block.
    pub fn is_losing(&self) -> bool {
        self.losing
    }

    /// Iterates the frames in this block.
    ///
    /// The iterator borrows from `self`, not the ring: it must not outlive the block, whose drop
    /// releases these bytes back to the kernel.
    pub fn frames(&self) -> RxFrames<'_> {
        RxFrames { block: self.block, off: self.first_off, remaining: self.num_pkts }
    }
}

impl Drop for RxBlock<'_> {
    fn drop(&mut self) {
        // Release so the kernel sees our reads as complete before it reuses the block.
        self.status.store(libc::TP_STATUS_KERNEL, Ordering::Release);
        *self.cursor = (*self.cursor + 1) % self.block_count;
    }
}

/// Iterator over the frames of an [`RxBlock`].
#[derive(Debug)]
pub struct RxFrames<'a> {
    block: &'a [u8],
    off: usize,
    remaining: u32,
}

/// Builds a frame iterator over a raw block buffer.
///
/// Used internally by the RX ring and exposed (crate-internally) so benchmarks can drive the
/// zero-copy parse loop over a synthetic block without privileges.
pub(crate) fn frames_over(block: &[u8], first_off: usize, num_pkts: u32) -> RxFrames<'_> {
    RxFrames { block, off: first_off, remaining: num_pkts }
}

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

    fn next(&mut self) -> Option<Frame<'a>> {
        if self.remaining == 0 {
            return None;
        }

        // All offset arithmetic is overflow-checked: the offsets come from kernel-written header
        // fields, but treating them as untrusted keeps the pointer read sound even against a
        // malformed block or 32-bit wraparound (this is the fuzzed surface).
        let hdr_end = self.off.checked_add(mem::size_of::<libc::tpacket3_hdr>());
        if hdr_end.is_none_or(|e| e > self.block.len()) {
            self.remaining = 0;
            return None;
        }

        // SAFETY: the bounds check above guarantees the header is fully inside `block`.
        // `read_unaligned` makes no alignment assumption.
        let hdr = unsafe {
            ptr::read_unaligned(self.block.as_ptr().add(self.off).cast::<libc::tpacket3_hdr>())
        };

        let mac = hdr.tp_mac as usize;
        let snap = hdr.tp_snaplen as usize;
        let Some(start) = self.off.checked_add(mac) else {
            self.remaining = 0;
            return None;
        };
        let end = start.checked_add(snap);
        if end.is_none_or(|e| e > self.block.len()) {
            self.remaining = 0;
            return None;
        }
        let data = &self.block[start..end.unwrap()];

        // Keep only the cheap header scalars; timestamp/VLAN/packet-type are decoded lazily, so a
        // caller that only reads `data()` pays nothing for them. This matters on slow cores.
        let raw = RawRingMeta {
            wire_len: hdr.tp_len,
            ts_sec: hdr.tp_sec,
            ts_nsec: hdr.tp_nsec,
            status: hdr.tp_status,
            vlan_tci: hdr.hv1.tp_vlan_tci as u16,
            vlan_tpid: hdr.hv1.tp_vlan_tpid,
        };

        // Advance to the next frame within the block (overflow-checked; a bad offset just ends
        // iteration on the next call via the header bounds check).
        match hdr.tp_next_offset {
            0 => self.remaining = 1, // last frame: terminate after this one
            // Saturate on overflow; the next iteration's header bounds check then ends iteration.
            next => self.off = self.off.saturating_add(next as usize),
        }
        self.remaining -= 1;

        Some(Frame::ring(data, raw))
    }

    fn size_hint(&self) -> (usize, Option<usize>) {
        // `remaining` is the kernel's claimed count, but a malformed block ends iteration early,
        // so it is only honest as an upper bound.
        (0, Some(self.remaining as usize))
    }
}

/// A `TPACKET_V2` transmit ring with batched submission.
#[derive(Debug)]
pub struct TxRing {
    sock: PacketSocket,
    mmap: Mmap,
    layout: RingLayout,
    /// Index of the next frame slot to fill.
    cursor: usize,
    /// Index of the most recently filled slot, if any — used to detect frames the kernel has not
    /// yet picked up (the TX ring is only walked inside `send(2)`; nothing drains it in the
    /// background).
    last_filled: Option<usize>,
    /// Offset from a frame slot's start to its data area.
    data_off: usize,
}

impl TxRing {
    /// Builds a v2 TX ring on `sock` with the given geometry.
    pub fn open(sock: PacketSocket, layout: RingLayout) -> Result<Self> {
        sock.set_packet_opt_int(libc::PACKET_VERSION, "PACKET_VERSION", TPACKET_V2)?;
        let req = layout.to_req();
        // SAFETY: `req` is a fully-initialised tpacket_req, the struct PACKET_TX_RING expects.
        unsafe { sock.set_packet_opt(libc::PACKET_TX_RING, "PACKET_TX_RING", &req)? };
        let mmap = Mmap::new(sock.as_fd(), layout.map_len())?;
        let data_off = tpacket_align(mem::size_of::<libc::tpacket2_hdr>());
        Ok(TxRing { sock, mmap, layout, cursor: 0, last_filled: None, data_off })
    }

    /// The maximum frame length that fits in one TX slot.
    pub fn max_frame_len(&self) -> usize {
        self.layout.frame_size as usize - self.data_off
    }

    fn slot_status(&self, idx: usize) -> &AtomicU32 {
        let off = idx * self.layout.frame_size as usize;
        // SAFETY: `off` is within the mapping (idx < frame_count), and `tp_status` is the first,
        // 4-byte-aligned field of the frame slot's header.
        unsafe { AtomicU32::from_ptr(self.mmap.base().add(off).cast::<u32>()) }
    }

    /// Fills the next available slot with `frame`. Returns `WouldBlock` if the ring is full.
    fn fill_slot(&mut self, frame: &[u8]) -> io::Result<()> {
        if frame.len() > self.max_frame_len() {
            return Err(io::Error::new(io::ErrorKind::InvalidInput, "frame exceeds TX frame size"));
        }
        let idx = self.cursor;
        let status = self.slot_status(idx);
        let s = status.load(Ordering::Acquire);
        // WRONG_FORMAT is the kernel handing back a slot whose frame it rejected (e.g. over the
        // device MTU): the slot is user-owned again and must be reclaimed here, or the cursor
        // would wedge on it and every future send would report a full ring.
        if s != libc::TP_STATUS_AVAILABLE && s != libc::TP_STATUS_WRONG_FORMAT {
            return Err(io::Error::from(io::ErrorKind::WouldBlock));
        }

        let slot_off = idx * self.layout.frame_size as usize;
        // SAFETY: the slot is AVAILABLE (kernel-released), so we may write its header and data.
        // `tp_len`/`tp_snaplen` are the u32 fields at offsets 4 and 8 of tpacket2_hdr; the data
        // area begins at `data_off`, and `frame.len() <= max_frame_len()` keeps it in bounds.
        unsafe {
            let base = self.mmap.base().add(slot_off);
            ptr::write_unaligned(base.add(4).cast::<u32>(), frame.len() as u32);
            ptr::write_unaligned(base.add(8).cast::<u32>(), frame.len() as u32);
            ptr::copy_nonoverlapping(frame.as_ptr(), base.add(self.data_off), frame.len());
        }
        // Release so the kernel observes the header/data writes before it sees SEND_REQUEST.
        status.store(libc::TP_STATUS_SEND_REQUEST, Ordering::Release);

        self.last_filled = Some(idx);
        self.cursor = (idx + 1) % self.layout.frame_count as usize;
        Ok(())
    }

    /// Kicks the kernel to transmit all queued (`SEND_REQUEST`) slots.
    ///
    /// A `WouldBlock` from the kick (send buffer full) is not fatal — the slots stay
    /// `SEND_REQUEST` — but the kernel only walks the TX ring inside `send(2)`, so such frames
    /// sit unsent until another kick. Callers use [`TxRing::kick_pending`] to guarantee pickup.
    fn flush(&self) -> io::Result<()> {
        // SAFETY: a zero-length send on a TX-ring socket flushes queued frames; passing a null
        // buffer with length 0 is the documented way to trigger transmission.
        let rc = unsafe {
            libc::send(self.sock.as_raw_fd(), ptr::null(), 0, libc::MSG_DONTWAIT)
        };
        if rc < 0 {
            let err = io::Error::last_os_error();
            if err.kind() == io::ErrorKind::WouldBlock {
                return Ok(());
            }
            return Err(err);
        }
        Ok(())
    }

    /// Whether the most recently filled slot is still waiting for the kernel to pick it up.
    ///
    /// `tpacket_snd` consumes `SEND_REQUEST` slots in order, so if the last-filled slot has left
    /// `SEND_REQUEST` (to `SENDING`/`AVAILABLE`), every earlier frame has been picked up too.
    pub fn has_unsent(&self) -> bool {
        match self.last_filled {
            Some(idx) => {
                self.slot_status(idx).load(Ordering::Acquire) == libc::TP_STATUS_SEND_REQUEST
            }
            None => false,
        }
    }

    /// Blocking follow-up to a [`TxRing::flush`] whose kick was refused (`EAGAIN`) or cut short:
    /// hands queued frames to the kernel, waiting out a full send buffer.
    ///
    /// This is a single blocking kick, not a drain-until-empty loop: if the kernel's TX head is
    /// parked on a slot it rejected (`TP_STATUS_WRONG_FORMAT`), frames queued behind it cannot be
    /// sent until the ring wraps and reclaims that slot, and the kick returns immediately —
    /// looping until `has_unsent` clears would spin forever. `POLLOUT` cannot replace this: it
    /// reflects the kernel head slot's availability, which stays false exactly while frames are
    /// parked, so a poll-based wait deadlocks.
    pub fn finish_kick(&self) -> io::Result<()> {
        if !self.has_unsent() {
            return Ok(());
        }
        blocking_kick(self.sock.as_fd())
    }

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

/// Kicks a TX-ring socket with a *blocking* zero-length send: returns once every currently
/// sendable `SEND_REQUEST` slot has been handed to the device (waiting out a full send buffer),
/// or immediately if the kernel head is parked on a rejected slot. Retries `EINTR`.
pub(crate) fn blocking_kick(fd: BorrowedFd<'_>) -> io::Result<()> {
    loop {
        // SAFETY: a zero-length send on a TX-ring socket flushes queued frames; passing a null
        // buffer with length 0 is the documented way to trigger transmission.
        let rc = unsafe { libc::send(fd.as_raw_fd(), ptr::null(), 0, 0) };
        if rc < 0 {
            let err = io::Error::last_os_error();
            if err.kind() == io::ErrorKind::Interrupted {
                continue;
            }
            return Err(err);
        }
        return Ok(());
    }
}

impl crate::sys::RawChannel for TxRing {
    fn send(&mut self, frame: &[u8]) -> io::Result<usize> {
        self.fill_slot(frame)?;
        self.flush()?;
        Ok(frame.len())
    }

    fn send_batch(&mut self, frames: &[&[u8]]) -> io::Result<usize> {
        let mut filled = 0;
        for frame in frames {
            match self.fill_slot(frame) {
                Ok(()) => filled += 1,
                Err(e) if filled > 0 && e.kind() == io::ErrorKind::WouldBlock => break,
                Err(e) => return Err(e),
            }
        }
        if filled > 0 {
            self.flush()?;
        }
        Ok(filled)
    }
}

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

    #[test]
    fn size_hint_lower_bound_is_zero_for_malformed_blocks() {
        // A block that claims 5 frames but is too short to hold even one header: iteration ends
        // early, so the lower bound must not promise frames that may never come.
        let block = [0u8; 8];
        let it = frames_over(&block, 0, 5);
        assert_eq!(it.size_hint(), (0, Some(5)));
        assert_eq!(it.count(), 0);
    }

    #[test]
    fn tx_data_offset_matches_kernel_header() {
        // The TX data area starts after the aligned v2 header (TPACKET2_HDRLEN minus the trailing
        // sockaddr_ll the macro adds).
        let expected = tpacket_align(mem::size_of::<libc::tpacket2_hdr>());
        assert_eq!(expected % libc::TPACKET_ALIGNMENT, 0);
        assert!(expected >= mem::size_of::<libc::tpacket2_hdr>());
    }
}