zenith_linux/

xsk.rs

//! AF_XDP Socket 管理
//!
//! 封装 AF_XDP Socket 的创建、配置、绑定和生命周期管理：
//! - Socket 创建（AF_XDP, SOCK_RAW）
//! - Socket 选项配置（SO_RCVBUF, SO_SNDBUF, XDP_MMAP_OFFSETS 等）
//! - 绑定到网卡队列（bind）
//! - 与 UMEM 和 Ring 集成
//!
//! # 设计原则
//! - 单 Owner：每个 XskSocket 由一个 Worker 独占拥有
//! - 零堆分配热路径：批量操作使用预分配缓冲区
//! - 类型安全：使用 XdpDesc（内核 xdp_desc）而非裸 u64
//!
//! # unsafe 使用
//! 本文件封装 AF_XDP 系统调用（socket/bind/close/setsockopt 等）
//! 是 crate 内部四个被精确放开 `#[allow(unsafe_code)]` 的文件之一。

#![allow(unsafe_code)]

use crate::descriptor::{Descriptor, DescriptorEngine, XdpDesc};
use crate::error::{LinuxError, Result, UmemError, XskError};
use crate::ring::{RingOffsets, RingType, XskRing};
use crate::umem::UmemManager;
use std::sync::atomic::{AtomicBool, Ordering};
use std::sync::Arc;

/// Fill Ring 预填的分块大小（避免大块堆分配，栈上数组批量入队）
const PREFILL_CHUNK: usize = 64;

/// 构造 `xdp_umem_reg` 注册参数（纯函数，便于单测）
///
/// 对应内核 UAPI `struct xdp_umem_reg`（include/uapi/linux/if_xdp.h）：
/// - `addr`：UMEM 映射区基址（用户态虚拟地址）
/// - `len`：UMEM 总字节数
/// - `chunk_size`：帧大小（必须 2 的幂且 ≥ 2048）
/// - `headroom`：帧头预留
/// - `flags`/`tx_metadata_len`：保留字段，置 0（不启用 TX metadata）
fn build_umem_reg(addr: *const u8, len: usize, chunk_size: u32, headroom: u32) -> libc::xdp_umem_reg {
    libc::xdp_umem_reg {
        addr: addr as u64,
        len: len as u64,
        chunk_size,
        headroom,
        flags: 0,
        tx_metadata_len: 0,
    }
}

/// 将内核 `xdp_ring_offset` 转换为 [`RingOffsets`]（纯函数，便于单测）
///
/// `len` 为该 ring 的 mmap 总字节数：`desc + ring_size * sizeof(XdpDesc)`，
/// 与内核 `xsk_mmap` 期望的映射长度一致。
fn to_ring_offsets(k: &libc::xdp_ring_offset, ring_size: u32) -> RingOffsets {
    RingOffsets {
        producer: k.producer,
        consumer: k.consumer,
        desc: k.desc,
        flags: k.flags,
        len: k.desc + u64::from(ring_size) * std::mem::size_of::<XdpDesc>() as u64,
    }
}

/// bind() 四环 mmap 局部 RAII 守卫
///
/// 修复「四环 mmap 任一失败时已成功的裸 base 泄漏」：每张成功的 ring mmap
/// 逐张登记进守卫；Drop 时对未 disarm 的条目执行 munmap。映射成功移交
/// `XskRing`（所有权归 KernelRing、由其 Drop munmap）后才 disarm 摘除，
/// 保证任何 `?` 提前返回路径都精确回收前序裸映射、且绝不 double-munmap
/// 已移交的映射。
#[derive(Debug)]
struct RingMmapGuard {
    /// 已登记映射（登记顺序 = RX → TX → FILL → COMPLETION）
    entries: Vec<RingMmapEntry>,
}

/// 单张待回收 ring mmap 的登记条目
#[derive(Debug)]
struct RingMmapEntry {
    /// mmap 基址
    base: *mut u8,
    /// mmap 字节数
    len: usize,
    /// 已移交 XskRing（不再由本守卫回收）
    disarmed: bool,
}

impl RingMmapGuard {
    /// 创建空守卫
    fn new() -> Self {
        Self {
            entries: Vec::new(),
        }
    }

    /// 登记一张成功映射，返回其登记序号（后续 `disarm` 凭序号摘除）
    fn register(&mut self, base: *mut u8, len: usize) -> usize {
        self.entries.push(RingMmapEntry {
            base,
            len,
            disarmed: false,
        });
        self.entries.len() - 1
    }

    /// 摘除登记（映射所有权已移交 XskRing/KernelRing，此后由其 Drop munmap）
    ///
    /// 序号为 `register` 返回值；非法序号静默忽略（防御，不 panic）。
    fn disarm(&mut self, idx: usize) {
        if let Some(entry) = self.entries.get_mut(idx) {
            entry.disarmed = true;
        }
    }
}

impl Drop for RingMmapGuard {
    fn drop(&mut self) {
        for entry in &self.entries {
            if entry.disarmed || entry.base.is_null() {
                continue;
            }
            // SAFETY 不变量由 guard_unmap 收口执行（见该函数 SAFETY 注释）。
            guard_unmap(entry.base, entry.len);
        }
    }
}

/// 守卫回收单条映射的收口点。
///
/// 生产构建：真实 `munmap`。
/// 单元测试构建：若线程开启 [`GUARD_TEST_STUB`]，仅把 (base, len) 记入
/// [`GUARD_UNMAP_LOG`] 而不真正 `munmap`——guard 单测据此做确定性断言，
/// 消除"并发测试里的 mmap 立即复用刚释放地址"带来的页可达性竞态；
/// 测试创建的匿名页由各测试显式 `munmap` 清理。
#[cfg(not(test))]
fn guard_unmap(base: *mut u8, len: usize) {
    // SAFETY: 未 disarm 的映射所有权仍归守卫（每张 mmap 成功仅 register
    // 一次、此处仅 munmap 一次；已移交 XskRing 的映射均已 disarm 跳过）；
    // base/len 来自注册时成功的 mmap，munmap 精确回收。
    unsafe {
        libc::munmap(base as *mut libc::c_void, len);
    }
}

/// 见上方生产版本的 SAFETY 注释与 stub 语义说明。
#[cfg(test)]
fn guard_unmap(base: *mut u8, len: usize) {
    if GUARD_TEST_STUB.with(|s| s.get()) {
        GUARD_UNMAP_LOG.with(|l| l.borrow_mut().push((base as usize, len)));
        return;
    }
    // SAFETY: 与生产路径同一套所有权不变量，调用方为真实匿名映射。
    unsafe {
        libc::munmap(base as *mut libc::c_void, len);
    }
}

#[cfg(test)]
thread_local! {
    /// 见 [`guard_unmap`]：stub 开关（线程局部，测试相互隔离）
    static GUARD_TEST_STUB: std::cell::Cell<bool> = const { std::cell::Cell::new(false) };
    /// 见 [`guard_unmap`]：stub 模式下 Drop 的 munmap 记账 (base, len)
    static GUARD_UNMAP_LOG: std::cell::RefCell<Vec<(usize, usize)>> = const { std::cell::RefCell::new(Vec::new()) };
}

/// AF_XDP Socket 配置
#[derive(Debug, Clone)]
pub struct XskConfig {
    /// 网卡接口索引
    pub ifindex: u32,
    /// 队列 ID
    pub queue_id: u32,
    /// 是否使用 Zero-Copy 模式
    pub zero_copy: bool,
    /// Fill Ring 大小（必须是 2 的幂）
    pub fill_ring_size: u32,
    /// RX Ring 大小
    pub rx_ring_size: u32,
    /// TX Ring 大小
    pub tx_ring_size: u32,
    /// Completion Ring 大小
    pub completion_ring_size: u32,
    /// UMEM 共享标志
    pub shared_umem: bool,
    /// UMEM 帧（chunk）大小，必须是 2 的幂且 ≥ 2048（XDP_UMEM_MIN_CHUNK_SIZE）
    pub frame_size: u32,
    /// UMEM 帧头预留（headroom），必须 < frame_size
    pub headroom: u32,
    /// SO_RCVBUF 内核接收缓冲（字节）。默认 1MB；调用方可按需调整。
    pub so_rcvbuf: i32,
    /// SO_SNDBUF 内核发送缓冲（字节）。默认 1MB；调用方可按需调整。
    pub so_sndbuf: i32,
    /// Fill Ring 预填/再填充的分块大小（栈上批量数组）。默认 64。
    pub prefill_chunk: usize,
}

impl Default for XskConfig {
    fn default() -> Self {
        Self {
            ifindex: 0,
            queue_id: 0,
            zero_copy: true,
            fill_ring_size: 2048,
            rx_ring_size: 2048,
            tx_ring_size: 2048,
            completion_ring_size: 2048,
            shared_umem: false,
            frame_size: 4096,
            headroom: 0,
            so_rcvbuf: 1024 * 1024,
            so_sndbuf: 1024 * 1024,
            prefill_chunk: 64,
        }
    }
}

/// AF_XDP Socket 状态
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum XskState {
    /// 已创建（未配置）
    Created,
    /// 已配置（已设置选项）
    Configured,
    /// 已绑定（已绑定到网卡队列）
    Bound,
    /// 已关闭
    Closed,
}

/// AF_XDP Socket
///
/// 封装单个 AF_XDP Socket 的完整生命周期管理。
/// 单 Owner 设计：每个 Socket 由一个 Worker 独占。
pub struct XskSocket {
    /// Socket 文件描述符
    fd: Option<i32>,
    /// 配置
    config: XskConfig,
    /// 当前状态
    state: XskState,
    /// UMEM 管理器（共享所有权，Arc 仅用于多 Worker 共享 UMEM 的场景）
    umem: Option<Arc<UmemManager>>,
    /// Fill Ring
    fill_ring: XskRing,
    /// RX Ring
    rx_ring: XskRing,
    /// TX Ring
    tx_ring: XskRing,
    /// Completion Ring
    completion_ring: XskRing,
    /// 描述符引擎
    descriptor_engine: DescriptorEngine,
    /// Fill Ring 预填游标（已登记的 UMEM 帧数，防重复登记同一批帧）
    prefilled_frames: u32,
    /// 是否活跃
    active: AtomicBool,
}

impl std::fmt::Debug for XskSocket {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("XskSocket")
            .field("fd", &self.fd)
            .field("config", &self.config)
            .field("state", &self.state)
            .field("active", &self.active)
            .finish()
    }
}

impl XskSocket {
    /// 创建 AF_XDP Socket（**模拟模式**，显式选择）
    ///
    /// 四环使用用户态 `Vec<XdpDesc>` backing store，不触碰任何内核 AF_XDP 资源，
    /// 适合无 root / 无 XDP 内核支持环境下的开发与测试。
    ///
    /// 真实内核模式请显式使用 [`XskSocket::new_real`]（或手动串联
    /// `create_socket → configure → bind`）。本类型**绝不静默降级**：
    /// 模式在构造时选定，`bind` 失败一律返回错误（Fail-Closed）。
    pub fn new(config: XskConfig) -> Result<Self> {
        let fill_ring_size = config.fill_ring_size;
        let rx_ring_size = config.rx_ring_size;
        let tx_ring_size = config.tx_ring_size;
        let completion_ring_size = config.completion_ring_size;

        // checked 算术：四个 u32 先提升为 u64 再求和，杜绝 u32 加法溢出
        let total_descriptors = u64::from(fill_ring_size)
            + u64::from(rx_ring_size)
            + u64::from(tx_ring_size)
            + u64::from(completion_ring_size);
        // fail-closed：四环容量之和超出 20-bit 帧索引域（> 2^20）时拒绝构造
        let descriptor_engine = DescriptorEngine::new(total_descriptors)?;

        Ok(Self {
            fd: None,
            config,
            state: XskState::Created,
            umem: None,
            fill_ring: XskRing::new(RingType::Fill, fill_ring_size)?,
            rx_ring: XskRing::new(RingType::Rx, rx_ring_size)?,
            tx_ring: XskRing::new(RingType::Tx, tx_ring_size)?,
            completion_ring: XskRing::new(RingType::Completion, completion_ring_size)?,
            descriptor_engine,
            prefilled_frames: 0,
            active: AtomicBool::new(false),
        })
    }

    /// 创建**真实内核模式** AF_XDP Socket（显式选择，Fail-Closed）
    ///
    /// 完整执行真实系统调用链：
    /// 1. `socket(AF_XDP, SOCK_RAW, 0)`
    /// 2. `setsockopt(SO_RCVBUF/SO_SNDBUF)`
    /// 3. `bind()`：`XDP_UMEM_REG` 注册 → 四环大小 → `getsockopt(XDP_MMAP_OFFSETS)`
    ///    → 四环 mmap → `bind(sockaddr_xdp)` → Fill Ring 预填
    ///
    /// 任何一步失败立即返回错误（WSL2 等不支持环境下返回明确错误），
    /// **绝不回退到模拟模式**；模拟模式请显式使用 [`XskSocket::new`]。
    ///
    /// # 参数
    /// * `config` - Socket 配置（ifindex/queue_id/四环大小/帧大小等）
    /// * `umem` - 已完成 `create()`（mmap）的 UMEM 管理器
    pub fn new_real(config: XskConfig, umem: Arc<UmemManager>) -> Result<Self> {
        let mut socket = Self::new(config)?;
        socket.create_socket()?;
        socket.configure()?;
        socket.bind(umem)?;
        debug_assert!(socket.is_kernel_mode());
        Ok(socket)
    }

    /// 创建 Socket（系统调用）
    pub fn create_socket(&mut self) -> Result<()> {
        // 错误语义精确化：非 Created 状态（Configured/Bound/Closed 均非法；
        // AlreadyBound 仅描述"已绑定"子集）统一报 InvalidState 并注明实际状态
        if self.state != XskState::Created {
            return Err(XskError::InvalidState(format!(
                "create_socket expects state=Created, actual state={:?}",
                self.state
            ))
            .into());
        }

        // SAFETY: AF_XDP(44)、SOCK_RAW、protocol 0 为有效常量。
        // socket() 是纯系统调用，参数为编译期常量，不存在运行时风险。
        // 注意：AF_XDP 正确值为 44（linux/socket.h: #define AF_XDP 44），
        // 此前误写为 48 导致 EAFNOSUPPORT。统一使用 libc::AF_XDP 命名常量。
        let fd = unsafe { libc::socket(libc::AF_XDP, libc::SOCK_RAW, 0) };
        if fd < 0 {
            let err = std::io::Error::last_os_error();
            return Err(XskError::SocketCreate(format!("socket() failed: {}", err)).into());
        }

        self.fd = Some(fd);
        self.state = XskState::Configured;

        Ok(())
    }

    /// 配置 Socket 选项
    pub fn configure(&mut self) -> Result<()> {
        let fd = self.get_fd()?;

        // SAFETY: fd 已通过 get_fd()? 验证为非负有效值。
        // rcvbuf 为栈变量，指针在调用期间有效；size 参数为 sizeof(i32)，与实际数据一致。
        let rcvbuf: i32 = self.config.so_rcvbuf;
        let ret = unsafe {
            libc::setsockopt(
                fd,
                libc::SOL_SOCKET,
                libc::SO_RCVBUF,
                &rcvbuf as *const i32 as *const libc::c_void,
                std::mem::size_of::<i32>() as libc::socklen_t,
            )
        };
        if ret < 0 {
            let err = std::io::Error::last_os_error();
            return Err(XskError::SocketOption(format!("SO_RCVBUF failed: {}", err)).into());
        }

        // SAFETY: fd 已通过 get_fd()? 验证为非负有效值。
        // sndbuf 为栈变量，指针在调用期间有效；size 参数为 sizeof(i32)，与实际数据一致。
        let sndbuf: i32 = self.config.so_sndbuf;
        let ret = unsafe {
            libc::setsockopt(
                fd,
                libc::SOL_SOCKET,
                libc::SO_SNDBUF,
                &sndbuf as *const i32 as *const libc::c_void,
                std::mem::size_of::<i32>() as libc::socklen_t,
            )
        };
        if ret < 0 {
            let err = std::io::Error::last_os_error();
            return Err(XskError::SocketOption(format!("SO_SNDBUF failed: {}", err)).into());
        }

        Ok(())
    }

    /// 配置单个 Ring 的大小（setsockopt SOL_XDP）
    ///
    /// optname 为 XDP_UMEM_FILL_RING / XDP_UMEM_COMPLETION_RING / XDP_RX_RING / XDP_TX_RING。
    /// size 必须是 2 的幂。
    fn set_ring_size(&self, optname: libc::c_int, size: u32) -> Result<()> {
        let fd = self.get_fd()?;
        // SAFETY: fd 已验证有效；size 为栈变量，指针在调用期间有效；
        // optlen 为 sizeof(u32)，与实际传入的数据一致。
        let ret = unsafe {
            libc::setsockopt(
                fd,
                libc::SOL_XDP,
                optname,
                &size as *const u32 as *const libc::c_void,
                std::mem::size_of::<u32>() as libc::socklen_t,
            )
        };
        if ret < 0 {
            let err = std::io::Error::last_os_error();
            return Err(XskError::SocketOption(format!(
                "ring size setsockopt(opt={optname}, size={size}) failed: {err}"
            ))
            .into());
        }
        Ok(())
    }

    /// 通过 `getsockopt(XDP_MMAP_OFFSETS)` 获取内核四环偏移
    fn get_mmap_offsets(fd: i32) -> Result<libc::xdp_mmap_offsets> {
        // SAFETY: offsets 为栈上零初始化结构体，getsockopt 按 optlen 写入；
        // optlen 传入结构体完整大小，旧内核（v1 布局，无 flags 字段）会按其实际
        // 大小裁剪写入，剩余字段保持零值（flags=0 → need_wakeup 恒 false）。
        let mut offsets: libc::xdp_mmap_offsets = unsafe { std::mem::zeroed() };
        let mut optlen = std::mem::size_of::<libc::xdp_mmap_offsets>() as libc::socklen_t;
        // SAFETY: fd 已验证有效；offsets/optlen 均为栈上有效变量，指针在调用期间有效。
        let ret = unsafe {
            libc::getsockopt(
                fd,
                libc::SOL_XDP,
                libc::XDP_MMAP_OFFSETS,
                &mut offsets as *mut libc::xdp_mmap_offsets as *mut libc::c_void,
                &mut optlen,
            )
        };
        if ret < 0 {
            let err = std::io::Error::last_os_error();
            return Err(XskError::SocketOption(format!(
                "getsockopt(XDP_MMAP_OFFSETS) failed: {err}"
            ))
            .into());
        }
        Ok(offsets)
    }

    /// mmap 单个内核 ring 区域（MAP_SHARED + 内核 UAPI 页偏移）
    ///
    /// `pgoff` 为内核 UAPI 定义的 ring 页偏移（`libc::XDP_PGOFF_RX_RING` 等，
    /// 权威值见 include/uapi/linux/if_xdp.h：RX=0x0、TX=0x80000000、
    /// FILL=0x100000000、COMPLETION=0x180000000）。
    fn mmap_kernel_ring(fd: i32, pgoff: libc::off_t, length: usize, name: &str) -> Result<*mut u8> {
        // SAFETY: fd 已验证有效；pgoff 为内核 UAPI 常量；length 由
        // getsockopt(XDP_MMAP_OFFSETS) 偏移 + ring 容量计算，非零。
        // MAP_SHARED 是 AF_XDP ring 映射的必需标志（与内核共享同一段内存）。
        // 返回值已检查 MAP_FAILED。
        let addr = unsafe {
            libc::mmap(
                std::ptr::null_mut(),
                length,
                libc::PROT_READ | libc::PROT_WRITE,
                libc::MAP_SHARED,
                fd,
                pgoff,
            )
        };
        if addr == libc::MAP_FAILED {
            let err = std::io::Error::last_os_error();
            return Err(XskError::SocketOption(format!(
                "{name} ring mmap(pgoff={pgoff:#x}, len={length}) failed: {err}"
            ))
            .into());
        }
        Ok(addr as *mut u8)
    }

    /// 绑定到网卡队列（真实内核注册，Fail-Closed）
    ///
    /// 完整 AF_XDP 绑定序列（与 libbpf `xsk_socket__create` 顺序一致）：
    /// 1. `setsockopt(XDP_UMEM_REG)` 注册 UMEM 内存区
    /// 2. `setsockopt(XDP_UMEM_FILL_RING/COMPLETION_RING/RX_RING/TX_RING)` 配置四环大小
    /// 3. `getsockopt(XDP_MMAP_OFFSETS)` 获取四环内核布局偏移
    /// 4. `mmap` 四环（XDP_PGOFF_RX_RING / XDP_PGOFF_TX_RING /
    ///    XDP_UMEM_PGOFF_FILL_RING / XDP_UMEM_PGOFF_COMPLETION_RING），
    ///    四环切换为内核 mmap 模式（任一失败，已映射区域由 RingMmapGuard 回收）
    /// 5. `bind(sockaddr_xdp)` 绑定到 ifindex+queue_id
    /// 6. Fill Ring 预填帧地址，`XDP_RING_NEED_WAKEUP` 置位时
    ///    `sendto(MSG_DONTWAIT)` 唤醒内核
    ///
    /// WSL2 等不支持环境下任一步骤失败即返回明确错误，**绝不静默降级为模拟模式**。
    pub fn bind(&mut self, umem: Arc<UmemManager>) -> Result<()> {
        // ── 0. 全部纯参数校验前移到任何 syscall/setsockopt 之前（Fail-Closed）──
        // 此前 is_power_of_two 校验在 bind() 成功之后才执行：校验失败时已
        // 完成内核 bind，socket 处于"已绑定但配置非法"的矛盾状态。
        if !umem.is_initialized() || umem.as_ptr().is_null() {
            return Err(LinuxError::Umem(UmemError::NotCreated));
        }
        if self.config.frame_size == 0 {
            return Err(XskError::SocketOption(
                "frame_size 必须 > 0（XDP_UMEM_MIN_CHUNK_SIZE=2048）".to_string(),
            )
            .into());
        }
        // frame_size 为 XDP_UMEM chunk_size 契约，必须 2 的幂（纯参数校验）
        if !self.config.frame_size.is_power_of_two() {
            return Err(XskError::SocketOption(format!(
                "frame_size={} 必须是 2 的幂（XDP_UMEM chunk_size 契约）",
                self.config.frame_size
            ))
            .into());
        }
        // UMEM 帧数与描述符引擎容量也是纯参数：在任何 syscall 前完成计算与
        // 构造校验（fail-closed：超 20-bit 帧索引域 / 非法帧移位直接拒绝）。
        let frame_size = self.config.frame_size;
        let umem_frames = (umem.size() / frame_size as usize).min(u32::MAX as usize) as u32;
        // with_frame_shift fail-closed：frame_size 已校验为 2 的幂且 > 0，
        // trailing_zeros ∈ 11..=15 仍由引擎构造器二次校验（双重防护）
        let new_engine =
            DescriptorEngine::with_frame_shift(u64::from(umem_frames), frame_size.trailing_zeros())?;

        let fd = self.get_fd()?;

        // ── 1. 注册 UMEM（XDP_UMEM_REG）────────────────────────────
        // UMEM 必须已 mmap（umem.create() 先于本调用），as_ptr() 返回映射基址。
        let reg = build_umem_reg(
            umem.as_ptr(),
            umem.size(),
            self.config.frame_size,
            self.config.headroom,
        );
        // SAFETY: fd 已验证有效；reg 为栈变量，指针在调用期间有效；
        // optlen 为 sizeof(xdp_umem_reg)，与结构体一致。
        let ret = unsafe {
            libc::setsockopt(
                fd,
                libc::SOL_XDP,
                libc::XDP_UMEM_REG,
                &reg as *const libc::xdp_umem_reg as *const libc::c_void,
                std::mem::size_of::<libc::xdp_umem_reg>() as libc::socklen_t,
            )
        };
        if ret < 0 {
            let err = std::io::Error::last_os_error();
            return Err(XskError::SocketOption(format!("XDP_UMEM_REG failed: {err}")).into());
        }

        // ── 2. 配置四环大小 ────────────────────────────────────────
        self.set_ring_size(libc::XDP_UMEM_FILL_RING, self.config.fill_ring_size)?;
        self.set_ring_size(libc::XDP_UMEM_COMPLETION_RING, self.config.completion_ring_size)?;
        self.set_ring_size(libc::XDP_RX_RING, self.config.rx_ring_size)?;
        self.set_ring_size(libc::XDP_TX_RING, self.config.tx_ring_size)?;

        // ── 3. 获取内核四环偏移（XDP_MMAP_OFFSETS）──────────────────
        let offsets = Self::get_mmap_offsets(fd)?;

        // ── 4. mmap 四环并构造内核模式 XskRing ──────────────────────
        // 任一步失败时 self 上的四环保持原状（绝不出现半内核半模拟的混合状态）；
        // 已成功的 ring mmap 全部登记进 RingMmapGuard：任何 `?` 提前返回由
        // guard Drop 回收前序映射，彻底消除裸 base 泄漏。
        let rx_off = to_ring_offsets(&offsets.rx, self.config.rx_ring_size);
        let tx_off = to_ring_offsets(&offsets.tx, self.config.tx_ring_size);
        let fr_off = to_ring_offsets(&offsets.fr, self.config.fill_ring_size);
        let cr_off = to_ring_offsets(&offsets.cr, self.config.completion_ring_size);

        let mut mmap_guard = RingMmapGuard::new();
        let rx_base = Self::mmap_kernel_ring(
            fd,
            libc::XDP_PGOFF_RX_RING,
            rx_off.len as usize,
            "RX",
        )?;
        let rx_slot = mmap_guard.register(rx_base, rx_off.len as usize);
        let tx_base = Self::mmap_kernel_ring(
            fd,
            libc::XDP_PGOFF_TX_RING,
            tx_off.len as usize,
            "TX",
        )?;
        let tx_slot = mmap_guard.register(tx_base, tx_off.len as usize);
        let fr_base = Self::mmap_kernel_ring(
            fd,
            libc::XDP_UMEM_PGOFF_FILL_RING as libc::off_t,
            fr_off.len as usize,
            "FILL",
        )?;
        let fr_slot = mmap_guard.register(fr_base, fr_off.len as usize);
        let cr_base = Self::mmap_kernel_ring(
            fd,
            libc::XDP_UMEM_PGOFF_COMPLETION_RING as libc::off_t,
            cr_off.len as usize,
            "COMPLETION",
        )?;
        let cr_slot = mmap_guard.register(cr_base, cr_off.len as usize);

        // SAFETY: 四个 base 均来自上方成功的 ring mmap（长度 = offsets.len），
        // offsets 来自内核 getsockopt；本 Socket 单 Owner。构造成功即移交映射
        // 所有权给 XskRing（KernelRing Drop 时 munmap）并从 guard 摘除；
        // 构造失败/中途 `?` 时剩余映射由 guard Drop 精确回收。
        let rx_ring = unsafe { XskRing::with_kernel_ring(RingType::Rx, rx_base, rx_off) }?;
        mmap_guard.disarm(rx_slot);
        let tx_ring = unsafe { XskRing::with_kernel_ring(RingType::Tx, tx_base, tx_off) }?;
        mmap_guard.disarm(tx_slot);
        let fill_ring = unsafe { XskRing::with_kernel_ring(RingType::Fill, fr_base, fr_off) }?;
        mmap_guard.disarm(fr_slot);
        let completion_ring =
            unsafe { XskRing::with_kernel_ring(RingType::Completion, cr_base, cr_off) }?;
        mmap_guard.disarm(cr_slot);

        // ── 5. 准备 bind 地址（sockaddr_xdp）───────────────────────
        // SAFETY: sockaddr_xdp 为栈分配，通过 zeroed() 初始化为全零；
        // 随后所有字段（sxdp_family, sxdp_ifindex, sxdp_queue_id, sxdp_flags）均被显式赋值，
        // 保证传给 bind() 的结构体完整且有效。
        let mut addr: libc::sockaddr_xdp = unsafe { std::mem::zeroed() };
        addr.sxdp_family = libc::AF_XDP as u16;
        addr.sxdp_ifindex = self.config.ifindex;
        addr.sxdp_queue_id = self.config.queue_id;
        // sxdp_flags 命名常量（内核 UAPI）：
        // XDP_ZEROCOPY(1<<2) 强制零拷贝 / XDP_COPY(1<<1) 强制复制模式；
        // 共享 UMEM 时叠加 XDP_SHARED_UMEM(1<<0)。
        let mut flags: u16 = if self.config.zero_copy {
            libc::XDP_ZEROCOPY
        } else {
            libc::XDP_COPY
        };
        if self.config.shared_umem {
            flags |= libc::XDP_SHARED_UMEM;
        }
        addr.sxdp_flags = flags;

        let addr_ptr = &addr as *const libc::sockaddr_xdp as *const libc::sockaddr;
        let addr_len = std::mem::size_of::<libc::sockaddr_xdp>() as libc::socklen_t;

        // SAFETY: fd 已通过 get_fd()? 验证为非负有效值。
        // addr_ptr 指向栈上有效的 sockaddr_xdp 结构体，addr_len 为结构体正确大小。
        let ret = unsafe { libc::bind(fd, addr_ptr, addr_len) };
        if ret < 0 {
            let err = std::io::Error::last_os_error();
            return Err(XskError::BindFailed(format!(
                "bind() failed: {err}（WSL2/无驱动支持环境属预期失败，绝不降级模拟）"
            ))
            .into());
        }

        // ── 6. bind 成功：四环统一切换为内核 mmap 模式 ──────────────
        self.fill_ring = fill_ring;
        self.rx_ring = rx_ring;
        self.tx_ring = tx_ring;
        self.completion_ring = completion_ring;

        // 以 bind() 前置校验阶段预构的真实 UMEM 帧引擎替换 new() 中的
        // 四环容量占位引擎（帧数 = UMEM 字节 / 帧大小，构造已通过
        // 20-bit 帧索引域与帧移位双重校验）
        self.descriptor_engine = new_engine;
        self.prefilled_frames = 0;

        self.umem = Some(umem);
        self.state = XskState::Bound;
        self.active.store(true, Ordering::SeqCst);

        // ── 7. Fill Ring 预填帧地址，needs_wakeup 时唤醒内核 ────────
        self.prefill_fill_ring()?;

        Ok(())
    }

    /// Fill Ring 预填：将 UMEM 前 N 个帧地址写入 Fill Ring
    ///
    /// N = min(fill_ring 容量, UMEM 帧数)。每个帧同时在描述符引擎中登记
    /// （位图置位），保证守恒等式成立。预填后若内核 `XDP_RING_NEED_WAKEUP`
    /// 置位，通过 `sendto(fd, NULL, 0, MSG_DONTWAIT, NULL, 0)` 唤醒内核轮询。
    ///
    /// # 返回
    /// * `Result<u32>` - 实际预填的帧数
    pub fn prefill_fill_ring(&mut self) -> Result<u32> {
        let frame_size = self.config.frame_size;
        if frame_size == 0 {
            return Err(XskError::SocketOption("frame_size 必须 > 0".to_string()).into());
        }
        let umem = match &self.umem {
            Some(u) => Arc::clone(u),
            None => return Ok(0), // 模拟模式无 UMEM，无帧可填
        };

        // UMEM 帧数 = 总字节 / 帧大小（checked：usize 除法不溢出；
        // 帧数钳制到 u32 范围，防巨型 UMEM 截断）
        let umem_frames = (umem.size() / frame_size as usize).min(u32::MAX as usize) as u32;
        let target = self.fill_ring.capacity().min(umem_frames);
        let owner = self.config.queue_id as u16;

        // 游标续填：从 prefilled_frames 继续，禁止从 0 重复登记同一批帧
        // （重复 allocate 会重复计数，破坏守恒等式）
        let start = self.prefilled_frames.min(target);
        let mut filled: u32 = 0;
        // 预填分块大小可配置（钳制到栈上数组容量内，保持零堆分配热路径）
        let chunk_size = self.config.prefill_chunk.clamp(1, PREFILL_CHUNK);
        let mut chunk = [XdpDesc::zero(); PREFILL_CHUNK];
        while start.saturating_add(filled) < target {
            let frame_base = start.saturating_add(filled);
            // target - frame_base ≤ target ≤ u32::MAX，转 usize 安全
            let batch = ((target - frame_base) as usize).min(chunk_size);
            for (j, slot) in chunk.iter_mut().enumerate().take(batch) {
                // frame_base + j < target ≤ u32::MAX，不会溢出
                let frame_index = frame_base + j as u32;
                // 在描述符引擎登记帧（位图置位），保持守恒等式
                let _ = self.descriptor_engine.allocate(frame_index, owner)?;
                *slot = XdpDesc {
                    addr: u64::from(frame_index) * u64::from(frame_size),
                    len: 0,
                    options: 0,
                };
            }
            let n = self.fill_ring.enqueue_batch(&chunk[..batch])?;
            filled = filled.saturating_add(n);
            if n as usize != batch {
                break; // ring 已满，停止预填
            }
        }
        self.prefilled_frames = start.saturating_add(filled);

        // XDP_RING_NEED_WAKEUP 置位时唤醒内核消费 Fill Ring
        // （模拟模式 need_wakeup 恒 false，自动跳过）
        if filled > 0 && self.fill_ring.need_wakeup() {
            self.notify_fill()?;
        }
        Ok(filled)
    }

    /// 是否为内核 mmap 模式（bind 成功后四环已切换为内核共享映射）
    #[inline]
    pub fn is_kernel_mode(&self) -> bool {
        self.rx_ring.is_kernel_mode()
    }

    /// 关闭 Socket
    pub fn close(&mut self) -> Result<()> {
        self.active.store(false, Ordering::SeqCst);

        // take() 确保 fd 只被尝试关闭一次：即使 close() 返回错误，
        // fd 也已从 self.fd 移除，Drop 不会再次关闭（避免 double-close）
        if let Some(fd) = self.fd.take() {
            // SAFETY: fd 在 self.fd 中仅当 create_socket() 成功时才存在（非负）。
            // close() 系统调用对有效 fd 是安全的，且 take() 确保 fd 不会被并发关闭。
            let ret = unsafe { libc::close(fd) };
            if ret < 0 {
                // fd 已被 Linux 关闭（close(2) 即使返回 EINTR 也已关闭 fd），
                // 仅报告错误，不重试
                self.state = XskState::Closed;
                let err = std::io::Error::last_os_error();
                return Err(XskError::CloseFailed(format!("close() failed: {}", err)).into());
            }
        }

        self.state = XskState::Closed;

        Ok(())
    }

    /// 获取 Socket FD
    #[inline]
    pub fn fd(&self) -> Option<i32> {
        self.fd
    }

    /// 获取状态
    #[inline]
    pub fn state(&self) -> XskState {
        self.state
    }

    /// 是否活跃
    #[inline]
    pub fn is_active(&self) -> bool {
        self.active.load(Ordering::SeqCst)
    }

    /// 获取 Fill Ring 引用
    #[inline]
    pub fn fill_ring(&self) -> &XskRing {
        &self.fill_ring
    }

    /// 获取 Fill Ring 可变引用
    #[inline]
    pub fn fill_ring_mut(&mut self) -> &mut XskRing {
        &mut self.fill_ring
    }

    /// 获取 RX Ring 引用
    #[inline]
    pub fn rx_ring(&self) -> &XskRing {
        &self.rx_ring
    }

    /// 获取 RX Ring 可变引用
    #[inline]
    pub fn rx_ring_mut(&mut self) -> &mut XskRing {
        &mut self.rx_ring
    }

    /// 获取 TX Ring 引用
    #[inline]
    pub fn tx_ring(&self) -> &XskRing {
        &self.tx_ring
    }

    /// 获取 TX Ring 可变引用
    #[inline]
    pub fn tx_ring_mut(&mut self) -> &mut XskRing {
        &mut self.tx_ring
    }

    /// 获取 Completion Ring 引用
    #[inline]
    pub fn completion_ring(&self) -> &XskRing {
        &self.completion_ring
    }

    /// 获取 Completion Ring 可变引用
    #[inline]
    pub fn completion_ring_mut(&mut self) -> &mut XskRing {
        &mut self.completion_ring
    }

    /// 获取描述符引擎引用
    #[inline]
    pub fn descriptor_engine(&self) -> &DescriptorEngine {
        &self.descriptor_engine
    }

    /// 批量填充 Fill Ring
    ///
    /// 向 Fill Ring 注入可用的 XdpDesc，供内核接收数据包。
    ///
    /// # 参数
    /// * `count` - 填充数量
    ///
    /// # 返回
    /// * `Result<u32>` - 实际填充数量
    pub fn fill_descriptors(&mut self, count: u32) -> Result<u32> {
        let mut descs = Vec::with_capacity(count as usize);
        let mut allocated_indices = Vec::with_capacity(count as usize);
        for i in 0..count {
            // 修复「从 0 重复登记帧」：引擎位图已置位的帧不得再次 allocate
            // （allocate 对已分配帧 fail-closed 返回 AlreadyAllocated）；
            // 只登记尚未被引擎分配的帧，与 prefill_fill_ring 的游标语义一致，
            // 保证 allocated == freed + active 守恒等式不被双登记破坏。
            if self.descriptor_engine.is_allocated(i) {
                continue;
            }
            let desc = self
                .descriptor_engine
                .allocate(i, self.config.queue_id as u16)?;
            descs.push(*desc.xdp_desc());
            allocated_indices.push(i);
        }
        let filled = match self.fill_ring.enqueue_batch(&descs) {
            Ok(n) => n,
            Err(e) => {
                // fail-closed 回滚：enqueue 失败时，已分配的帧位图置位但未入环，
                // 形成孤立帧（无法回收）。必须以位图为权威回滚这些帧，
                // 保持 allocated == freed + active 守恒等式。
                for &idx in &allocated_indices {
                    self.descriptor_engine.rollback_allocate(idx);
                }
                return Err(e);
            }
        };
        // 真实内核模式：enqueue 后检查是否需要唤醒内核轮询 Fill Ring
        if self.fill_ring.need_wakeup() {
            self.notify_fill()?;
        }
        Ok(filled)
    }

    /// 接收数据包（从 RX Ring）
    ///
    /// # 参数
    /// * `max_batch` - 最大批量
    ///
    /// # 返回
    /// * `Result<Vec<Descriptor>>` - 接收到的描述符列表
    pub fn receive_batch(&mut self, max_batch: u32) -> Result<Vec<Descriptor>> {
        let raw_descs = self.rx_ring.dequeue_batch(max_batch)?;
        let mut descriptors = Vec::with_capacity(raw_descs.len());

        for raw_xdp in raw_descs {
            // fail-closed：frame index 高位域非法（> MAX_FRAME_INDEX）时
            // from_xdp 直接拒绝，非法描述符绝不进入验证管线。
            // 使用引擎实际 frame_shift 而非硬编码 12，保证非 4096 帧大小下索引正确
            let desc = Descriptor::from_xdp(
                raw_xdp,
                self.config.queue_id as u16,
                0,
                self.descriptor_engine.frame_shift(),
            )?;
            self.descriptor_engine
                .verify(&desc, self.config.queue_id as u16)?;
            descriptors.push(desc);
        }

        Ok(descriptors)
    }

    /// 接收数据包到预分配缓冲区（零堆分配路径）
    ///
    /// # 参数
    /// * `buffer` - 预分配的 Descriptor 缓冲区
    ///
    /// # 返回
    /// * `Result<u32>` - 实际接收数量
    pub fn receive_batch_to(&mut self, buffer: &mut [Descriptor]) -> Result<u32> {
        let max_batch = buffer.len() as u32;
        let raw_descs = self.rx_ring.dequeue_batch(max_batch)?;
        let count = raw_descs.len() as u32;

        for (i, raw_xdp) in raw_descs.iter().enumerate() {
            let desc = Descriptor::from_xdp(
                *raw_xdp,
                self.config.queue_id as u16,
                0,
                self.descriptor_engine.frame_shift(),
            )?;
            self.descriptor_engine
                .verify(&desc, self.config.queue_id as u16)?;
            buffer[i] = desc;
        }

        Ok(count)
    }

    /// 发送数据包（到 TX Ring）
    ///
    /// # 参数
    /// * `descriptors` - 要发送的描述符列表
    ///
    /// # 返回
    /// * `Result<u32>` - 实际发送数量
    pub fn send_batch(&mut self, descriptors: &[Descriptor]) -> Result<u32> {
        let mut raw_descs: Vec<XdpDesc> = Vec::with_capacity(descriptors.len());

        for desc in descriptors {
            self.descriptor_engine
                .verify(desc, self.config.queue_id as u16)?;
            raw_descs.push(*desc.xdp_desc());
        }

        let sent = self.tx_ring.enqueue_batch(&raw_descs)?;
        // 真实内核模式：enqueue 后检查是否需要唤醒内核轮询 TX Ring
        if self.tx_ring.need_wakeup() {
            self.notify_tx()?;
        }
        Ok(sent)
    }

    /// 批量发送（从预分配缓冲区）
    ///
    /// # 参数
    /// * `descriptors` - 预分配的 XdpDesc 数组
    ///
    /// # 返回
    /// * `Result<u32>` - 实际发送数量
    pub fn send_xdp_batch(&mut self, descriptors: &[XdpDesc]) -> Result<u32> {
        // 与 send_batch 对称的描述符验证管线：每个 raw XdpDesc 必须通过
        // from_xdp 帧索引域校验 + descriptor_engine.verify 所有权/代际验证，
        // 杜绝未验证的 raw addr 直接进入内核 TX Ring（conservation 破坏 / 帧混淆）
        for raw_xdp in descriptors {
            let desc = Descriptor::from_xdp(
                *raw_xdp,
                self.config.queue_id as u16,
                0,
                self.descriptor_engine.frame_shift(),
            )?;
            self.descriptor_engine
                .verify(&desc, self.config.queue_id as u16)?;
        }
        let sent = self.tx_ring.enqueue_batch(descriptors)?;
        if self.tx_ring.need_wakeup() {
            self.notify_tx()?;
        }
        Ok(sent)
    }

    /// 唤醒内核处理 TX Ring（sendto 系统调用）
    ///
    /// 当 `XDP_RING_NEED_WAKEUP` 标志置位时，通过 `sendto(fd, NULL, 0, MSG_DONTWAIT, NULL, 0)`
    /// 通知内核从 TX Ring 消费描述符并发送数据包。
    ///
    /// 模拟模式下为空操作（无需唤醒）。
    #[inline]
    pub fn notify_tx(&self) -> Result<()> {
        let fd = match self.fd {
            Some(f) if f >= 0 => f,
            _ => return Ok(()), // 无效 fd 或模拟模式，静默跳过
        };
        // SAFETY: fd 已验证为有效值。sendto with NULL/0 是 AF_XDP 的标准唤醒路径，
        // 不涉及数据拷贝，仅通知内核轮询 TX Ring。
        let ret = unsafe {
            libc::sendto(
                fd,
                std::ptr::null(),
                0,
                libc::MSG_DONTWAIT,
                std::ptr::null(),
                0,
            )
        };
        if ret < 0 {
            let err = std::io::Error::last_os_error();
            // EAGAIN/EWOULDBLOCK/EINTR 为非致命错误，不阻断数据面
            if matches!(
                err.raw_os_error(),
                Some(libc::EAGAIN) | Some(libc::EINTR)
            ) {
                return Ok(());
            }
            return Err(
                XskError::NotifyFailed(format!("notify_tx sendto failed: {err}")).into(),
            );
        }
        Ok(())
    }

    /// 唤醒内核处理 Fill Ring（sendto 系统调用）
    ///
    /// 当 `XDP_RING_NEED_WAKEUP` 标志置位时，通过
    /// `sendto(fd, NULL, 0, MSG_DONTWAIT, NULL, 0)`
    /// 通知内核从 Fill Ring 获取空描述符并接收数据包。
    /// （本实现四环齐备，xsk_sendmsg 路径可正常进入内核唤醒逻辑。）
    ///
    /// 模拟模式下为空操作（无 fd，无需唤醒）。
    #[inline]
    pub fn notify_fill(&self) -> Result<()> {
        let fd = match self.fd {
            Some(f) if f >= 0 => f,
            _ => return Ok(()),
        };
        // SAFETY: fd 已验证为有效值。sendto with NULL/0 + MSG_DONTWAIT 是 AF_XDP 的
        // 标准唤醒路径，不涉及数据拷贝，仅通知内核轮询已绑定队列的 Fill Ring。
        let ret = unsafe {
            libc::sendto(
                fd,
                std::ptr::null(),
                0,
                libc::MSG_DONTWAIT,
                std::ptr::null(),
                0,
            )
        };
        if ret < 0 {
            let err = std::io::Error::last_os_error();
            // EAGAIN/EWOULDBLOCK/EINTR 为非致命错误，不阻断数据面
            if matches!(
                err.raw_os_error(),
                Some(libc::EAGAIN) | Some(libc::EINTR)
            ) {
                return Ok(());
            }
            return Err(
                XskError::NotifyFailed(format!("notify_fill sendto failed: {err}")).into(),
            );
        }
        Ok(())
    }

    /// 回收发送完成的描述符（从 Completion Ring）
    ///
    /// # 参数
    /// * `max_batch` - 最大批量
    ///
    /// # 返回
    /// * `Result<u32>` - 回收数量
    pub fn recycle_completed(&mut self, max_batch: u32) -> Result<u32> {
        let raw_descs = self.completion_ring.dequeue_batch(max_batch)?;
        let count = raw_descs.len() as u32;

        for raw_xdp in &raw_descs {
            // frame index 域校验 fail-closed（非法 raw addr 直接拒绝该批次）
            let desc = Descriptor::from_xdp(
                *raw_xdp,
                self.config.queue_id as u16,
                0,
                self.descriptor_engine.frame_shift(),
            )?;
            self.descriptor_engine.release(&desc)?;
        }

        Ok(count)
    }

    /// 批量接收并回收（零拷贝路径）
    ///
    /// # 参数
    /// * `max_batch` - 最大批量
    ///
    /// # 返回
    /// * `Result<Vec<Descriptor>>` - 接收到的描述符
    pub fn receive_and_refill(&mut self, max_batch: u32) -> Result<Vec<Descriptor>> {
        let received = self.receive_batch(max_batch)?;

        if !received.is_empty() {
            self.fill_descriptors(received.len() as u32)?;
        }

        Ok(received)
    }

    /// 获取 UMEM 引用
    pub fn umem(&self) -> Option<&Arc<UmemManager>> {
        self.umem.as_ref()
    }

    /// 返回底层 AF_XDP socket 文件描述符。
    ///
    /// - socket 已创建且未关闭：`Ok(fd)`，fd ≥ 0
    /// - 尚未调用 `create_socket()` 或已关闭：`Err(XskError::NotBound)`
    ///
    /// 该 fd 可用于 `poll(2)`/`select(2)`/`epoll(2)` 等待 RX 队列唤醒，
    /// 或用于 SO_RCVBUF / SO_SNDBUF 调优。调用方不应 close 该 fd。
    pub fn get_fd(&self) -> Result<i32> {
        self.fd.ok_or_else(|| XskError::NotBound.into())
    }

    /// 队列 ID（构造配置原值；XSKMAP 注册 `queue_id → xsk_fd` 的 map key）
    #[inline]
    pub fn queue_id(&self) -> u32 {
        self.config.queue_id
    }
}

impl Drop for XskSocket {
    fn drop(&mut self) {
        let _ = self.close();
    }
}

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

    #[test]
    fn test_xsk_config_default() {
        let config = XskConfig::default();
        assert_eq!(config.ifindex, 0);
        assert_eq!(config.queue_id, 0);
        assert!(config.zero_copy);
    }

    #[test]
    fn test_xsk_socket_creation() {
        let config = XskConfig::default();
        let socket = XskSocket::new(config);
        assert!(socket.is_ok());
        let socket = socket.unwrap();
        assert_eq!(socket.state(), XskState::Created);
        assert!(!socket.is_active());
    }

    #[test]
    fn test_xsk_socket_create_and_configure() {
        let config = XskConfig::default();
        let mut socket = XskSocket::new(config).unwrap();

        let result = socket.create_socket();
        if result.is_ok() {
            let result = socket.configure();
            assert!(result.is_ok());
        }
    }

    #[test]
    fn test_xsk_ring_operations() {
        let config = XskConfig::default();
        let socket = XskSocket::new(config).unwrap();

        assert_eq!(socket.fill_ring().capacity(), 2048);
        assert_eq!(socket.rx_ring().capacity(), 2048);
        assert_eq!(socket.tx_ring().capacity(), 2048);
        assert_eq!(socket.completion_ring().capacity(), 2048);
    }

    #[test]
    fn test_xsk_descriptor_engine() {
        let config = XskConfig::default();
        let socket = XskSocket::new(config).unwrap();

        let engine = socket.descriptor_engine();
        let desc = engine.allocate(0, 1).unwrap();
        assert_eq!(desc.index(), 0);
        assert_eq!(desc.addr(), 0);

        engine.verify(&desc, 1).unwrap();
        engine.release(&desc).unwrap();

        assert!(engine.verify_conservation());
    }

    #[test]
    fn test_xsk_fill_and_receive() {
        let config = XskConfig::default();
        let mut socket = XskSocket::new(config).unwrap();

        // 填充 Fill Ring
        let filled = socket.fill_descriptors(4).unwrap();
        assert_eq!(filled, 4);
        assert_eq!(socket.fill_ring().available_data(), 4);

        // 手动将描述符转移到 RX Ring（模拟内核行为）
        let descs = socket.fill_ring_mut().dequeue_batch(4).unwrap();
        assert_eq!(descs.len(), 4);
        socket.rx_ring_mut().enqueue_batch(&descs).unwrap();

        // 接收
        let received = socket.receive_batch(4).unwrap();
        assert_eq!(received.len(), 4);

        for desc in &received {
            assert_eq!(desc.addr(), (desc.index() as u64) << 12);
        }
    }

    #[test]
    fn test_xsk_send_and_complete() {
        let config = XskConfig::default();
        let mut socket = XskSocket::new(config).unwrap();

        // 分配描述符模拟 TX 场景
        let d0 = socket.descriptor_engine().allocate(100, 0).unwrap();
        let d1 = socket.descriptor_engine().allocate(101, 0).unwrap();

        let descs = [d0, d1];
        let sent = socket.send_batch(&descs).unwrap();
        assert_eq!(sent, 2);

        // 模拟内核完成
        let tx_descs = socket.tx_ring_mut().dequeue_batch(2).unwrap();
        socket.completion_ring_mut().enqueue_batch(&tx_descs).unwrap();

        // 回收
        let recycled = socket.recycle_completed(2).unwrap();
        assert_eq!(recycled, 2);

        assert!(socket.descriptor_engine().verify_conservation());
    }

    #[test]
    fn test_xsk_receive_batch_to_buffer() {
        let config = XskConfig::default();
        let mut socket = XskSocket::new(config).unwrap();

        // 填充并转移到 RX
        socket.fill_descriptors(3).unwrap();
        let descs = socket.fill_ring_mut().dequeue_batch(3).unwrap();
        socket.rx_ring_mut().enqueue_batch(&descs).unwrap();

        // 使用预分配缓冲区
        let mut buffer = [Descriptor::default(); 4];
        let count = socket.receive_batch_to(&mut buffer).unwrap();
        assert_eq!(count, 3);
    }

    #[test]
    fn test_xsk_config_custom_values() {
        let config = XskConfig {
            ifindex: 5,
            queue_id: 3,
            zero_copy: false,
            fill_ring_size: 512,
            rx_ring_size: 1024,
            tx_ring_size: 256,
            completion_ring_size: 128,
            shared_umem: true,
            frame_size: 4096,
            headroom: 0,
            so_rcvbuf: 1024 * 1024,
            so_sndbuf: 1024 * 1024,
            prefill_chunk: 64,
        };
        assert_eq!(config.ifindex, 5);
        assert_eq!(config.queue_id, 3);
        assert!(!config.zero_copy);
        assert_eq!(config.fill_ring_size, 512);
        assert_eq!(config.rx_ring_size, 1024);
        assert_eq!(config.tx_ring_size, 256);
        assert_eq!(config.completion_ring_size, 128);
        assert!(config.shared_umem);
    }

    #[test]
    fn test_xsk_config_clone() {
        let config = XskConfig {
            ifindex: 42,
            queue_id: 7,
            ..Default::default()
        };
        let cloned = config.clone();
        assert_eq!(cloned.ifindex, 42);
        assert_eq!(cloned.queue_id, 7);
    }

    #[test]
    fn test_xsk_state_transitions() {
        let config = XskConfig::default();
        let mut socket = XskSocket::new(config).unwrap();
        assert_eq!(socket.state(), XskState::Created);
        assert!(!socket.is_active());

        let result = socket.create_socket();
        if result.is_ok() {
            assert_eq!(socket.state(), XskState::Configured);
        }
    }

    #[test]
    fn test_xsk_socket_debug_format() {
        let config = XskConfig::default();
        let socket = XskSocket::new(config).unwrap();
        let debug = format!("{:?}", socket);
        assert!(debug.contains("XskSocket"));
        assert!(debug.contains("state"));
        assert!(debug.contains("active"));
    }

    #[test]
    fn test_xsk_ring_sizes_from_config() {
        let config = XskConfig {
            fill_ring_size: 256,
            rx_ring_size: 512,
            tx_ring_size: 1024,
            completion_ring_size: 2048,
            ..Default::default()
        };
        let socket = XskSocket::new(config).unwrap();

        assert_eq!(socket.fill_ring().capacity(), 256);
        assert_eq!(socket.rx_ring().capacity(), 512);
        assert_eq!(socket.tx_ring().capacity(), 1024);
        assert_eq!(socket.completion_ring().capacity(), 2048);
    }

    #[test]
    fn test_xsk_queue_id_config() {
        let config = XskConfig {
            queue_id: 42,
            ..Default::default()
        };
        let socket = XskSocket::new(config).unwrap();
        let desc = socket.descriptor_engine().allocate(0, 42).unwrap();
        assert_eq!(desc.owner(), 42);
    }

    #[test]
    fn test_xsk_zero_copy_config() {
        let config = XskConfig {
            zero_copy: true,
            ..Default::default()
        };
        assert!(config.zero_copy);

        let config = XskConfig {
            zero_copy: false,
            ..Default::default()
        };
        assert!(!config.zero_copy);
    }

    #[test]
    fn test_xsk_send_xdp_batch() {
        let config = XskConfig::default();
        let mut socket = XskSocket::new(config).unwrap();

        // send_xdp_batch 现在与 send_batch 对称验证所有权：
        // 每个 raw XdpDesc 必须通过 from_xdp 帧索引域校验 + engine.verify。
        // 先在引擎中分配帧 0 和帧 1（addr = frame_index << frame_shift = i << 12）
        let _ = socket.descriptor_engine().allocate(0, 42).unwrap();
        let _ = socket.descriptor_engine().allocate(1, 42).unwrap();

        let xdp_descs = [
            XdpDesc {
                addr: 0, // frame 0: 0 << 12 = 0
                len: 100,
                options: 0,
            },
            XdpDesc {
                addr: 4096, // frame 1: 1 << 12 = 4096
                len: 200,
                options: 0,
            },
        ];

        let sent = socket.send_xdp_batch(&xdp_descs).unwrap();
        assert_eq!(sent, 2);
        assert_eq!(socket.tx_ring().available_data(), 2);
    }

    #[test]
    fn test_xsk_receive_and_refill() {
        let config = XskConfig::default();
        let mut socket = XskSocket::new(config).unwrap();

        socket.fill_descriptors(4).unwrap();
        let descs = socket.fill_ring_mut().dequeue_batch(4).unwrap();
        socket.rx_ring_mut().enqueue_batch(&descs).unwrap();

        let received = socket.receive_and_refill(4).unwrap();
        assert_eq!(received.len(), 4);
    }

    #[test]
    fn test_xsk_fill_descriptors_zero_count() {
        let config = XskConfig::default();
        let mut socket = XskSocket::new(config).unwrap();

        let filled = socket.fill_descriptors(0).unwrap();
        assert_eq!(filled, 0);
    }

    #[test]
    fn test_xsk_receive_empty_rx_ring() {
        let config = XskConfig::default();
        let mut socket = XskSocket::new(config).unwrap();

        let received = socket.receive_batch(10).unwrap();
        assert!(received.is_empty());
    }

    #[test]
    fn test_xsk_recycle_empty_completion_ring() {
        let config = XskConfig::default();
        let mut socket = XskSocket::new(config).unwrap();

        let recycled = socket.recycle_completed(10).unwrap();
        assert_eq!(recycled, 0);
    }

    #[test]
    fn test_xsk_state_variants() {
        let states = vec![
            XskState::Created,
            XskState::Configured,
            XskState::Bound,
            XskState::Closed,
        ];
        for state in states {
            let debug = format!("{:?}", state);
            assert!(!debug.is_empty());
        }
    }

    #[test]
    fn test_xsk_socket_fd_none_in_simulation() {
        let config = XskConfig::default();
        let socket = XskSocket::new(config).unwrap();
        assert!(socket.fd().is_none());
    }

    #[test]
    fn test_xsk_umem_none_initially() {
        let config = XskConfig::default();
        let socket = XskSocket::new(config).unwrap();
        assert!(socket.umem().is_none());
    }

    #[test]
    fn test_xsk_conservation_after_send_and_recycle() {
        let config = XskConfig::default();
        let mut socket = XskSocket::new(config).unwrap();

        let d0 = socket.descriptor_engine().allocate(10, 0).unwrap();
        let d1 = socket.descriptor_engine().allocate(11, 0).unwrap();

        assert!(socket.descriptor_engine().verify_conservation());

        let descs = [d0, d1];
        socket.send_batch(&descs).unwrap();

        let tx_descs = socket.tx_ring_mut().dequeue_batch(2).unwrap();
        socket.completion_ring_mut().enqueue_batch(&tx_descs).unwrap();

        socket.recycle_completed(2).unwrap();

        assert!(socket.descriptor_engine().verify_conservation());
    }

    #[test]
    fn test_build_umem_reg_pure_logic() {
        // UMEM 注册参数纯逻辑：字段必须与内核 xdp_umem_reg 布局一致
        let addr = 0x7f00_0000usize as *const u8;
        let reg = build_umem_reg(addr, 2 * 1024 * 1024, 4096, 256);
        assert_eq!(reg.addr, 0x7f00_0000u64);
        assert_eq!(reg.len, 2 * 1024 * 1024u64);
        assert_eq!(reg.chunk_size, 4096);
        assert_eq!(reg.headroom, 256);
        assert_eq!(reg.flags, 0, "保留标志必须置 0");
        assert_eq!(reg.tx_metadata_len, 0, "不启用 TX metadata");

        // 空地址/零 headroom 边界
        let reg0 = build_umem_reg(std::ptr::null(), 4096, 2048, 0);
        assert_eq!(reg0.addr, 0);
        assert_eq!(reg0.headroom, 0);
    }

    #[test]
    fn test_to_ring_offsets_pure_logic() {
        // ring 偏移纯逻辑：len = desc + ring_size * sizeof(XdpDesc)
        let k = libc::xdp_ring_offset {
            producer: 0,
            consumer: 64,
            desc: 128,
            flags: 192,
        };
        let off = to_ring_offsets(&k, 256);
        assert_eq!(off.producer, 0);
        assert_eq!(off.consumer, 64);
        assert_eq!(off.desc, 128);
        assert_eq!(off.flags, 192);
        assert_eq!(
            off.len,
            128 + 256 * std::mem::size_of::<XdpDesc>() as u64,
            "mmap 长度必须覆盖 desc 偏移 + 全部描述符"
        );
        // XdpDesc 必须与内核 xdp_desc 同为 16 字节
        assert_eq!(std::mem::size_of::<XdpDesc>(), 16);
    }

    #[test]
    fn test_xsk_new_is_simulation_mode() {
        // 构造时显式选择：new() = 模拟模式，绝不隐式进入内核模式
        let socket = XskSocket::new(XskConfig::default()).unwrap();
        assert!(!socket.is_kernel_mode());
        assert!(!socket.rx_ring().is_kernel_mode());
        assert!(!socket.fill_ring().is_kernel_mode());
        assert!(socket.fd().is_none());
    }

    #[test]
    fn test_xsk_bind_rejects_uncreated_umem() {
        // Fail-Closed：UMEM 未 create() 时 bind 必须返回明确错误（NotCreated）
        let mut socket = match XskSocket::new(XskConfig::default()) {
            Ok(s) => s,
            Err(_) => return,
        };
        if socket.create_socket().is_err() {
            return; // 环境不支持 AF_XDP socket（如无权限），跳过
        }
        let umem = Arc::new(
            UmemManager::new(crate::umem::UmemConfig {
                size: 4096 * 16,
                hugepage: false,
                locked: false,
                shared: false,
            })
            .unwrap(),
        );
        // 注意：未调用 umem.create()
        let result = socket.bind(umem);
        assert!(result.is_err(), "UMEM 未 mmap 时 bind 必须失败");
        assert!(!socket.is_kernel_mode(), "失败后不得进入内核模式（无混合状态）");
    }

    #[test]
    fn test_xsk_new_real_fail_closed_on_invalid_ifindex() {
        // Fail-Closed：ifindex=0 永远非法，new_real 必须返回错误且绝不降级模拟
        let mut umem = UmemManager::new(crate::umem::UmemConfig {
            size: 4096 * 64,
            hugepage: false,
            locked: false,
            shared: false,
        })
        .unwrap();
        umem.create().unwrap();
        let config = XskConfig {
            ifindex: 0, // 非法接口索引
            queue_id: 0,
            zero_copy: false,
            fill_ring_size: 64,
            rx_ring_size: 64,
            tx_ring_size: 64,
            completion_ring_size: 64,
            shared_umem: false,
            frame_size: 4096,
            headroom: 0,
            so_rcvbuf: 1024 * 1024,
            so_sndbuf: 1024 * 1024,
            prefill_chunk: 64,
        };
        let result = XskSocket::new_real(config, Arc::new(umem));
        match result {
            Ok(s) => {
                // 极端环境（ifindex 0 竟可绑定）：必须是完整内核模式
                assert!(s.is_kernel_mode());
                assert_eq!(s.state(), XskState::Bound);
            }
            Err(e) => {
                // 预期路径：明确错误，无 panic、无静默降级
                let msg = format!("{e}");
                assert!(!msg.is_empty());
            }
        }
    }

    #[test]
    fn test_prefill_fill_ring_kernel_mode_logic() {
        // 预填纯逻辑：注入匿名 mmap 假内核 Fill Ring + 真实 UMEM，
        // 验证帧地址按 frame_size 步进写入共享区、producer 索引正确、守恒成立
        let config = XskConfig {
            fill_ring_size: 16,
            frame_size: 4096,
            ..Default::default()
        };
        let mut socket = XskSocket::new(config).unwrap();

        // 真实 UMEM：8 帧 × 4096
        let mut umem = UmemManager::new(crate::umem::UmemConfig {
            size: 4096 * 8,
            hugepage: false,
            locked: false,
            shared: false,
        })
        .unwrap();
        umem.create().unwrap();
        socket.umem = Some(Arc::new(umem));

        // 假内核 Fill Ring（匿名 mmap，容量 16）
        let desc_off = 64u64;
        let ring_size = 16u32;
        let len = desc_off + u64::from(ring_size) * 16;
        // SAFETY: 测试内匿名 mmap，参数有效；检查 MAP_FAILED。
        let base = unsafe {
            libc::mmap(
                std::ptr::null_mut(),
                len as usize,
                libc::PROT_READ | libc::PROT_WRITE,
                libc::MAP_SHARED | libc::MAP_ANONYMOUS,
                -1,
                0,
            )
        };
        assert_ne!(base, libc::MAP_FAILED);
        let base = base as *mut u8;
        let offsets = crate::ring::RingOffsets {
            producer: 0,
            consumer: 8,
            desc: desc_off,
            flags: 16,
            len,
        };
        // SAFETY: base 有效、offsets 在映射区内、测试单线程。
        socket.fill_ring = unsafe {
            XskRing::with_kernel_ring(RingType::Fill, base, offsets)
        }
        .unwrap();

        // UMEM 仅 8 帧 → 预填数 = min(16, 8) = 8
        let filled = socket.prefill_fill_ring().unwrap();
        assert_eq!(filled, 8);

        // 验证共享区：producer == 8，desc[i].addr == i * 4096
        // SAFETY: 读取本测试拥有的映射区，偏移在 len 内。
        unsafe {
            assert_eq!((base as *const u32).read(), 8);
            for i in 0..8u64 {
                let d = (base.add(desc_off as usize) as *const XdpDesc)
                    .add(i as usize)
                    .read();
                assert_eq!(d.addr, i * 4096, "帧地址必须按 frame_size 步进");
                assert_eq!(d.len, 0);
            }
        }
        // 守恒：8 个帧已在引擎登记（allocated == freed + active）
        assert!(socket.descriptor_engine().verify_conservation());
    }

    #[test]
    fn test_prefill_fill_ring_simulation_mode_no_umem() {
        // 模拟模式无 UMEM：预填为无操作，返回 0
        let mut socket = XskSocket::new(XskConfig::default()).unwrap();
        assert_eq!(socket.prefill_fill_ring().unwrap(), 0);
    }

    // ── RingMmapGuard 单元测试（真实匿名 mmap 验证回收/移交语义）───────

    /// 测试辅助：匿名 mmap 4096 字节共享页，返回基址（失败即测试失败）
    fn anon_mmap_page() -> *mut u8 {
        // SAFETY: 测试内匿名 mmap，参数为有效常量；下方检查 MAP_FAILED。
        let base = unsafe {
            libc::mmap(
                std::ptr::null_mut(),
                4096,
                libc::PROT_READ | libc::PROT_WRITE,
                libc::MAP_SHARED | libc::MAP_ANONYMOUS,
                -1,
                0,
            )
        };
        assert_ne!(base, libc::MAP_FAILED, "匿名 mmap 失败");
        base as *mut u8
    }

    /// 测试辅助：mincore 探测页是否仍在映射中（false = 已被 munmap）
    fn page_is_mapped(addr: *const u8) -> bool {
        let mut vec = [0u8; 1];
        // SAFETY: mincore 仅查询页驻留状态；vec 为栈上 1 字节有效缓冲区，
        // addr 为页对齐地址（mmap 返回值）；页未映射时返回 ENOMEM 而非 UB。
        let ret = unsafe { libc::mincore(addr as *mut libc::c_void, 4096, vec.as_mut_ptr()) };
        ret == 0
    }

    #[test]
    fn test_ring_mmap_guard_disarm_path_keeps_mapping() {
        // 正常路径：disarm 后 guard Drop 不得 munmap（映射所有权已移交）
        let base = anon_mmap_page();
        {
            let mut guard = RingMmapGuard::new();
            let slot = guard.register(base, 4096);
            guard.disarm(slot);
        } // guard Drop：唯一条目已 disarm，跳过 munmap
        // SAFETY: base 未被 munmap（disarm 语义），4096 字节区域内。
        unsafe {
            base.write(0xAB);
            assert_eq!(base.read(), 0xAB, "disarm 后映射必须仍可访问");
        }
        assert!(page_is_mapped(base), "disarm 的映射不得被 munmap");
        // 测试清理（模拟 KernelRing Drop 的 munmap）
        // SAFETY: base 为本测试映射的 4096 字节区域，此后不再使用。
        unsafe {
            libc::munmap(base as *mut libc::c_void, 4096);
        }
    }

    #[test]
    fn test_ring_mmap_guard_failure_path_munmaps_all() {
        // 失败路径：未 disarm 的映射在 guard Drop 时全部回收（防裸 base 泄漏）。
        // 用记账桩做确定性断言（避免并发测试复用刚释放地址的竞态抖动）。
        GUARD_TEST_STUB.with(|s| s.set(true));
        GUARD_UNMAP_LOG.with(|l| l.borrow_mut().clear());
        let b1 = anon_mmap_page();
        let b2 = anon_mmap_page();
        {
            let mut guard = RingMmapGuard::new();
            guard.register(b1, 4096);
            guard.register(b2, 4096);
            // 模拟中途 `?` 提前返回：不 disarm 任何条目，guard 直接 Drop
        }
        let log = GUARD_UNMAP_LOG.with(|l| l.borrow().clone());
        assert_eq!(
            log,
            vec![(b1 as usize, 4096), (b2 as usize, 4096)],
            "失败路径遗留映射必须全部按登记顺序回收"
        );
        GUARD_TEST_STUB.with(|s| s.set(false));
        GUARD_UNMAP_LOG.with(|l| l.borrow_mut().clear());
        // SAFETY: b1/b2 为本测试映射的页，测试结束清理。
        unsafe {
            libc::munmap(b1 as *mut libc::c_void, 4096);
            libc::munmap(b2 as *mut libc::c_void, 4096);
        }
    }

    #[test]
    fn test_ring_mmap_guard_partial_disarm_only_munmaps_armed() {
        // 部分移交路径：已移交（disarm）的保留，未移交（armed）的全部回收
        GUARD_TEST_STUB.with(|s| s.set(true));
        GUARD_UNMAP_LOG.with(|l| l.borrow_mut().clear());
        let b1 = anon_mmap_page();
        let b2 = anon_mmap_page();
        let b3 = anon_mmap_page();
        {
            let mut guard = RingMmapGuard::new();
            let s1 = guard.register(b1, 4096);
            guard.register(b2, 4096);
            guard.register(b3, 4096);
            guard.disarm(s1);
        }
        let log = GUARD_UNMAP_LOG.with(|l| l.borrow().clone());
        assert_eq!(
            log,
            vec![(b2 as usize, 4096), (b3 as usize, 4096)],
            "仅未移交（armed）映射 b2/b3 须被回收，已移交 b1 不得回收"
        );
        GUARD_TEST_STUB.with(|s| s.set(false));
        GUARD_UNMAP_LOG.with(|l| l.borrow_mut().clear());
        // SAFETY: 三页均为本测试映射的匿名页，测试结束清理。
        unsafe {
            libc::munmap(b1 as *mut libc::c_void, 4096);
            libc::munmap(b2 as *mut libc::c_void, 4096);
            libc::munmap(b3 as *mut libc::c_void, 4096);
        }
    }

    #[test]
    fn test_xsk_fill_descriptors_no_double_register() {
        // 修复「fill_descriptors 从 0 重复登记帧破坏守恒」：
        // 第二次填充同一批帧时必须跳过已分配帧（返回 0），allocate 不再双计数
        let config = XskConfig::default();
        let mut socket = XskSocket::new(config).unwrap();

        let filled = socket.fill_descriptors(4).unwrap();
        assert_eq!(filled, 4);
        let (allocated, _) = socket.descriptor_engine().conservation_stats();
        assert_eq!(allocated, 4);

        let again = socket.fill_descriptors(4).unwrap();
        assert_eq!(again, 0, "已分配帧不得重复登记");
        let (allocated_after, _) = socket.descriptor_engine().conservation_stats();
        assert_eq!(allocated_after, 4, "重复填充不得双计数");
        assert!(socket.descriptor_engine().verify_conservation());
    }
}