zenith_linux/

ring.rs

//! AF_XDP Ring 操作封装
//!
//! 实现与内核 AF_XDP Ring 兼容的四环操作（Fill/RX/TX/Completion）：
//! - 支持模拟模式（Vec<XdpDesc> backing store，用于测试和开发）
//! - 支持内核 mmap 模式（通过 XDP_MMAP_OFFSETS 获取内核 ring 地址）
//! - SPSC 无锁批量操作
//! - 基于 XdpDesc（内核 xdp_desc）的描述符传递
//!
//! # 内核 Ring 结构
//!
//! AF_XDP Ring 在内核中的内存布局（每 ring）：
//! ```text
//! [flags]      (4 bytes) - XDP_RING_NEED_WAKEUP 等标志
//! [producer]   (4 bytes) - 生产者索引（用户态写，内核态读）
//! [consumer]   (4 bytes) - 消费者索引（内核态写，用户态读）
//! [desc0]      (16 bytes) - 第一个 xdp_desc
//! [desc1]      (16 bytes) - 第二个 xdp_desc
//! ...
//! ```
//!
//! # 偏移量
//!
//! 通过 getsockopt(XDP_MMAP_OFFSETS) 获取各 ring 偏移：
//! - XDP_RX_RING_OFFSET: RX Ring 在 mmap 中的偏移
//! - XDP_TX_RING_OFFSET: TX Ring 在 mmap 中的偏移
//! - XDP_FILL_RING_OFFSET: Fill Ring 在 mmap 中的偏移
//! - XDP_COMPLETION_RING_OFFSET: Completion Ring 在 mmap 中的偏移
//!
//! # unsafe 使用
//! 本文件实现与内核共享内存的 Ring 原语，包含 mmap 地址读写与裸指针访问。
//! 是 crate 内部四个被精确放开 `#[allow(unsafe_code)]` 的文件之一。

#![allow(unsafe_code)]

use crate::descriptor::XdpDesc;
use crate::error::{Result, RingError};
use std::sync::atomic::{AtomicU32, Ordering};

/// XDP_RING_NEED_WAKEUP 标志（内核 UAPI `include/uapi/linux/if_xdp.h`：`(1 << 0)` = 0x1）
///
/// 当此标志置位时，用户态必须通过 sendto()/recvfrom() 唤醒内核轮询。
/// - TX Ring：调用 sendto(fd, NULL, 0, MSG_DONTWAIT, NULL, 0)
/// - Fill Ring：调用 recvfrom(fd, NULL, 0, MSG_DONTWAIT, NULL, NULL)
///
/// 注：0x1 是内核唯一定义；此前误写 0x2（bit 1）时内核从不写该位，
/// `need_wakeup()` 恒为 false，notify_tx/notify_fill 路径静默永不唤醒。
pub const XDP_RING_NEED_WAKEUP: u32 = 0x1;

/// 模拟模式（[`RingStorage::Simulated`]）环容量硬上限（2^20 = 1M 项）
///
/// Simulated 模式的 descriptors 为真实堆分配（16B × capacity = ≤ 16 MiB）；
/// 不设上限时调用方可请求 2^31 项（32 GiB）触发 OOM abort（`vec!` 分配失败
/// 走 alloc_error_handler 直接 abort，Fail-Closed 无法返回错误）。
pub const MAX_SIMULATED_CAPACITY: u32 = 1 << 20;

/// Ring 类型
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum RingType {
    /// Fill Ring（用户态 → 内核态：填充可接收的描述符）
    Fill,
    /// RX Ring（内核态 → 用户态：接收完成的描述符）
    Rx,
    /// TX Ring（用户态 → 内核态：发送的描述符）
    Tx,
    /// Completion Ring（内核态 → 用户态：发送完成的描述符）
    Completion,
}

/// Ring 内存布局偏移（相对于 ring 起始地址）
///
/// 与内核 `xdp_ring_offset` 结构一致：
/// - producer: 生产者索引的偏移
/// - consumer: 消费者索引的偏移
/// - desc: 描述符数组起始偏移
/// - flags: 标志位偏移
/// - len: ring 长度（字节）
#[derive(Debug, Clone, Copy)]
pub struct RingOffsets {
    /// 生产者索引偏移
    pub producer: u64,
    /// 消费者索引偏移
    pub consumer: u64,
    /// 描述符数组起始偏移
    pub desc: u64,
    /// 标志位偏移
    pub flags: u64,
    /// ring 总长度（字节）
    pub len: u64,
}

/// AF_XDP Ring
///
/// # 两种模式（枚举互斥，禁止混用）
///
/// 1. **模拟模式**（[`RingStorage::Simulated`]）：使用 `Vec<XdpDesc>` 作为 backing store，
///    索引通过 AtomicU32 模拟，适合无 root 权限的开发和测试。
///
/// 2. **内核 mmap 模式**（[`RingStorage::Kernel`]）：通过 mmap 映射内核 ring 内存，
///    producer/consumer 索引与描述符数组均直接读写内核共享区。
pub struct XskRing {
    /// Ring 类型
    ring_type: RingType,
    /// 容量（必须是 2 的幂）
    capacity: u32,
    /// 位掩码（capacity - 1）
    mask: u32,
    /// 存储后端（模拟 Vec / 内核 mmap 二选一，类型层面禁止混用）
    storage: RingStorage,
}

/// Ring 存储后端
///
/// 修复「内核模式描述符来自空 Vec 导致越界」问题：
/// 内核模式的描述符来自 mmap 共享区，模拟模式使用用户态 Vec，
/// 两种后端由枚举互斥表达，任何路径都不可能拿错存储。
enum RingStorage {
    /// 模拟模式：用户态 backing store（测试/开发）
    Simulated {
        /// 生产者索引（用户态写）
        producer_idx: AtomicU32,
        /// 消费者索引（用户态读）
        consumer_idx: AtomicU32,
        /// 描述符数组 backing store
        descriptors: Vec<XdpDesc>,
    },
    /// 内核 mmap 模式：所有指针均指向内核共享映射区
    Kernel(KernelRing),
}

/// 内核 mmap 模式的 ring 视图
///
/// 持有 mmap 区域所有权：`Drop` 时执行 `munmap`。
struct KernelRing {
    /// mmap 区域基址
    mmap_base: *mut u8,
    /// mmap 区域长度（字节）
    mmap_len: usize,
    /// 生产者索引（内核共享区，用户态写）
    producer: *mut AtomicU32,
    /// 消费者索引（内核共享区，内核态写）
    consumer: *mut AtomicU32,
    /// 描述符数组起始地址（内核共享区）
    descs: *mut XdpDesc,
    /// flags 字段指针（XDP_RING_NEED_WAKEUP）；旧内核（< 5.4）无此字段时为 None
    flags: Option<*const AtomicU32>,
}

impl Drop for KernelRing {
    fn drop(&mut self) {
        // SAFETY: mmap_base/mmap_len 来自 with_kernel_ring 接管的有效 mmap 区域，
        // 本结构独占其所有权，Drop 时精确 munmap 一次（此后无任何引用）。
        unsafe {
            libc::munmap(self.mmap_base as *mut libc::c_void, self.mmap_len);
        }
    }
}

impl std::fmt::Debug for XskRing {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("XskRing")
            .field("ring_type", &self.ring_type)
            .field("capacity", &self.capacity)
            .field("producer_idx", &self.producer_index())
            .field("consumer_idx", &self.consumer_index())
            .field("kernel_mode", &self.is_kernel_mode())
            .finish()
    }
}

// SAFETY: XskRing 设计为单 Owner（Worker）独占，无 Arc/Rc 共享路径。
// Send/Sync 安全的原因：
// 1. 模拟模式：所有数据存储在 Vec<XdpDesc> 中，AtomicU32 索引操作本身线程安全，
//    且 Worker 独占调用 enqueue/dequeue，不存在数据竞争。
// 2. 内核模式（RingStorage::Kernel）：mmap 区域由 KernelRing 独占持有（Drop 时 munmap），
//    producer/consumer/descs 指针指向该区域，Worker 独占访问内核 ring（SPSC 模型：
//    用户态与内核各持一端，通过 Acquire/Release 原子索引同步）。
// 3. 裸指针不自动实现 Send/Sync，此处手动 impl 是对上述安全契约的显式声明。
unsafe impl Send for XskRing {}
unsafe impl Sync for XskRing {}

impl XskRing {
    /// 创建模拟模式的 Ring
    ///
    /// # 参数
    /// * `ring_type` - Ring 类型
    /// * `capacity` - 容量（向上取整到 2 的幂，最小 16）
    ///
    /// # 返回
    /// * `Result<Self>` - Ring 实例
    ///
    /// # Fail-Closed
    /// - `capacity == 0`：拒绝（原 `max(16)` 会静默掩盖调用方传 0 的错误）
    /// - 取整后超出 [`MAX_SIMULATED_CAPACITY`]：拒绝。模拟模式的
    ///   descriptors 是真实堆分配（`XdpDesc` 16B × capacity），不设上限时
    ///   恶意/失控调用方可请求高达 2^31 项（32 GiB）触发 OOM abort——
    ///   `vec!` 分配失败走 `alloc_error_handler` 直接 abort，连错误都返回不了。
    /// - 取整后超出 u32 可表示的最大 2 的幂（2^31）：拒绝。原
    ///   `next_power_of_two()` 在 `capacity > 2^31` 时 debug panic、
    ///   release 回绕为 0（mask = 0 - 1 = u32::MAX，后续索引全越界），
    ///   此处改用 `checked_next_power_of_two()` 显式报错。
    pub fn new(ring_type: RingType, capacity: u32) -> Result<Self> {
        if capacity == 0 {
            return Err(RingError::InvalidOffsets(
                "ring capacity 必须 > 0".to_string(),
            )
            .into());
        }
        let actual_capacity = capacity
            .max(16)
            .checked_next_power_of_two()
            .ok_or_else(|| {
                RingError::InvalidOffsets(format!(
                    "ring capacity {capacity} 超出最大可表示容量 2^31"
                ))
            })?;
        if actual_capacity > MAX_SIMULATED_CAPACITY {
            return Err(RingError::InvalidOffsets(format!(
                "ring capacity {actual_capacity} 超出模拟模式上限 {MAX_SIMULATED_CAPACITY}"
            ))
            .into());
        }
        let mask = actual_capacity - 1;

        Ok(Self {
            ring_type,
            capacity: actual_capacity,
            mask,
            storage: RingStorage::Simulated {
                producer_idx: AtomicU32::new(0),
                consumer_idx: AtomicU32::new(0),
                descriptors: vec![XdpDesc::zero(); actual_capacity as usize],
            },
        })
    }

    /// 创建内核 mmap 模式的 Ring（接管 mmap 区域所有权，Drop 时 munmap）
    ///
    /// # 参数
    /// * `ring_type` - Ring 类型
    /// * `mmap_base` - mmap 区域基址（必须指向有效的、已映射的内核 ring 内存）
    /// * `offsets` - Ring 偏移量（由 getsockopt(XDP_MMAP_OFFSETS) 获取；
    ///   `len` 为该 ring 的 mmap 总字节数 = `desc + ring_size * sizeof(XdpDesc)`）
    ///
    /// # 返回
    /// * `Result<Self>` - Ring 实例；偏移/长度非法时返回 `RingError::InvalidOffsets`（Fail-Closed）
    ///
    /// # Safety
    /// 调用者必须保证以下条件：
    /// 1. `mmap_base` 指向长度 ≥ `offsets.len` 的有效 mmap 映射区域（来自 xsk fd 的
    ///    ring mmap）。本函数成功后该区域所有权转移给 XskRing，由 `Drop` 执行 `munmap`，
    ///    调用者不得再自行 `munmap`。
    /// 2. `offsets` 中的偏移量必须与内核 ring 布局一致（通过 getsockopt 获取，
    ///    由内核保证正确性）；本函数仍会对偏移做边界校验，非法即返回错误。
    /// 3. 调用者必须保证 XskRing 仅在单 Worker 上下文中使用（单 Owner，无并发共享），
    ///    因为内核 ring 的 producer/consumer 索引需要严格的 happens-before 关系。
    pub unsafe fn with_kernel_ring(
        ring_type: RingType,
        mmap_base: *mut u8,
        offsets: RingOffsets,
    ) -> Result<Self> {
        // ── Fail-Closed 校验：任何偏移/长度非法直接返回错误 ──────────────
        if mmap_base.is_null() {
            return Err(RingError::InvalidOffsets("mmap_base 为空指针".to_string()).into());
        }
        // checked 算术：desc 区长度 = len - desc（防止 desc > len 时下溢）
        let desc_area = offsets.len.checked_sub(offsets.desc).ok_or_else(|| {
            RingError::InvalidOffsets(format!(
                "desc 偏移 {} 超出 ring 长度 {}",
                offsets.desc, offsets.len
            ))
        })?;
        let desc_size = std::mem::size_of::<XdpDesc>() as u64;
        let capacity_u64 = desc_area / desc_size;
        if capacity_u64 == 0 || capacity_u64 > u64::from(u32::MAX) {
            return Err(RingError::InvalidOffsets(format!(
                "描述符容量 {capacity_u64} 非法"
            ))
            .into());
        }
        let capacity = capacity_u64 as u32;
        // 内核 ring 大小必须是 2 的幂（xsk_setsockopt 强制）；不合法即拒绝，
        // 绝不向上取整（否则 mask 与内核实际 ring 大小不一致导致越界）。
        if !capacity.is_power_of_two() {
            return Err(RingError::InvalidOffsets(format!(
                "容量 {capacity} 不是 2 的幂"
            ))
            .into());
        }
        // producer/consumer/flags 偏移必须落在映射区内（off < len 后 len - off 不会下溢）
        let word = std::mem::size_of::<u32>() as u64;
        for (name, off) in [
            ("producer", offsets.producer),
            ("consumer", offsets.consumer),
        ] {
            if off >= offsets.len || offsets.len - off < word {
                return Err(RingError::InvalidOffsets(format!(
                    "{name} 偏移 {off} 越界（len={}）",
                    offsets.len
                ))
                .into());
            }
        }
        if offsets.flags > 0 && (offsets.flags >= offsets.len || offsets.len - offsets.flags < word)
        {
            return Err(RingError::InvalidOffsets(format!(
                "flags 偏移 {} 越界（len={}）",
                offsets.flags, offsets.len
            ))
            .into());
        }

        // SAFETY: 调用者契约保证 mmap_base 指向长度 ≥ offsets.len 的有效映射区；
        // 上述校验已保证 producer/consumer/desc/flags 各偏移均落在映射区内，
        // 且内核保证这些偏移满足 u32/xdp_desc 对齐要求。
        let kernel = KernelRing {
            mmap_base,
            mmap_len: offsets.len as usize,
            producer: unsafe { mmap_base.add(offsets.producer as usize) as *mut AtomicU32 },
            consumer: unsafe { mmap_base.add(offsets.consumer as usize) as *mut AtomicU32 },
            descs: unsafe { mmap_base.add(offsets.desc as usize) as *mut XdpDesc },
            flags: if offsets.flags > 0 {
                Some(unsafe { mmap_base.add(offsets.flags as usize) as *const AtomicU32 })
            } else {
                None
            },
        };

        Ok(Self {
            ring_type,
            capacity,
            mask: capacity - 1,
            storage: RingStorage::Kernel(kernel),
        })
    }

    /// 计算 ring 槽位：`base + offset` 后按 mask 回绕
    ///
    /// producer/consumer 为 u32 单调递增索引，到达 `u32::MAX` 后按内核语义
    /// 自然回绕（wrapping_add）；mask = capacity - 1（2 的幂减 1）保证结果
    /// 恒落在 `[0, capacity)`，不会越界。
    #[inline]
    fn ring_slot(base: u32, offset: u32, mask: u32) -> u32 {
        base.wrapping_add(offset) & mask
    }

    /// 获取 producer 原子索引引用（按存储后端分发）
    #[inline]
    fn producer_atomic(&self) -> &AtomicU32 {
        match &self.storage {
            RingStorage::Simulated { producer_idx, .. } => producer_idx,
            // SAFETY: producer 指向 KernelRing 持有的内核共享映射区，
            // 偏移经 with_kernel_ring 校验，映射在 KernelRing Drop 前始终有效。
            RingStorage::Kernel(k) => unsafe { &*k.producer },
        }
    }

    /// 获取 consumer 原子索引引用（按存储后端分发）
    #[inline]
    fn consumer_atomic(&self) -> &AtomicU32 {
        match &self.storage {
            RingStorage::Simulated { consumer_idx, .. } => consumer_idx,
            // SAFETY: 同 producer_atomic，consumer 指针经校验且映射有效。
            RingStorage::Kernel(k) => unsafe { &*k.consumer },
        }
    }

    /// 读取指定槽位的描述符（按存储后端分发）
    #[inline]
    fn read_desc(&self, slot: u32) -> XdpDesc {
        match &self.storage {
            RingStorage::Simulated { descriptors, .. } => descriptors[slot as usize],
            // SAFETY: slot = index & mask < capacity，descs 指向内核共享描述符数组，
            // 容量由 with_kernel_ring 按 mmap 长度校验，slot 必然落在映射区内。
            // XdpDesc 为 Copy 类型，直接按值读取。
            RingStorage::Kernel(k) => unsafe { k.descs.add(slot as usize).read() },
        }
    }

    /// 写入指定槽位的描述符（按存储后端分发）
    #[inline]
    fn write_desc(&mut self, slot: u32, desc: XdpDesc) {
        match &mut self.storage {
            RingStorage::Simulated { descriptors, .. } => descriptors[slot as usize] = desc,
            // SAFETY: 同 read_desc；SPSC 单生产者语义，描述符写入先于
            // producer 的 Release 存储完成，对消费者（内核）建立 happens-before。
            RingStorage::Kernel(k) => unsafe { k.descs.add(slot as usize).write(desc) },
        }
    }

    /// 获取可用空间（生产者可写入的位置数）
    #[inline]
    pub fn available_space(&self) -> u32 {
        let producer = self.producer_atomic().load(Ordering::Relaxed);
        let consumer = self.consumer_atomic().load(Ordering::Acquire);
        // wrapping_sub：索引回绕后差值仍为正确的占用数（不变式：0 ≤ p - c ≤ capacity）
        self.capacity - (producer.wrapping_sub(consumer))
    }

    /// 获取可读数据（消费者可读取的位置数）
    #[inline]
    pub fn available_data(&self) -> u32 {
        let producer = self.producer_atomic().load(Ordering::Acquire);
        let consumer = self.consumer_atomic().load(Ordering::Relaxed);
        producer.wrapping_sub(consumer)
    }

    /// 批量入队（生产者操作）
    ///
    /// # 参数
    /// * `descriptors` - 要入队的 XdpDesc 数组
    ///
    /// # 返回
    /// * `Result<u32>` - 实际入队数量
    pub fn enqueue_batch(&mut self, descriptors: &[XdpDesc]) -> Result<u32> {
        if descriptors.is_empty() {
            return Ok(0);
        }

        let available = self.available_space();
        let to_enqueue = descriptors.len() as u32;

        if to_enqueue > available {
            return Err(RingError::BatchSizeExceeded {
                requested: to_enqueue,
                maximum: available,
            }
            .into());
        }

        let start_idx = self.producer_atomic().load(Ordering::Relaxed);

        for (i, desc) in descriptors.iter().enumerate() {
            let slot = Self::ring_slot(start_idx, i as u32, self.mask);
            self.write_desc(slot, *desc);
        }

        // Release 存储 producer：保证上述描述符写入先于索引对消费者（内核）可见。
        // wrapping_add：u32 索引按内核语义自然回绕。
        self.producer_atomic()
            .store(start_idx.wrapping_add(to_enqueue), Ordering::Release);

        Ok(to_enqueue)
    }

    /// 批量出队（消费者操作）
    ///
    /// # 参数
    /// * `max_batch` - 最大批量大小
    ///
    /// # 返回
    /// * `Result<Vec<XdpDesc>>` - 描述符数组
    pub fn dequeue_batch(&mut self, max_batch: u32) -> Result<Vec<XdpDesc>> {
        let available = self.available_data();
        let to_dequeue = available.min(max_batch);

        if to_dequeue == 0 {
            return Ok(Vec::new());
        }

        let consumer = self.consumer_atomic().load(Ordering::Relaxed);
        let mut result = Vec::with_capacity(to_dequeue as usize);

        for i in 0..to_dequeue {
            let slot = Self::ring_slot(consumer, i, self.mask);
            result.push(self.read_desc(slot));
        }

        // Release 存储 consumer：保证描述符读取先于索引对生产者（内核）可见
        self.consumer_atomic()
            .store(consumer.wrapping_add(to_dequeue), Ordering::Release);

        Ok(result)
    }

    /// 批量出队到预分配缓冲区（零堆分配路径）
    ///
    /// # 参数
    /// * `buffer` - 预分配的 XdpDesc 缓冲区
    ///
    /// # 返回
    /// * `Result<u32>` - 实际出队数量
    pub fn dequeue_batch_to(&mut self, buffer: &mut [XdpDesc]) -> Result<u32> {
        let available = self.available_data();
        let max = buffer.len() as u32;
        let to_dequeue = available.min(max);

        if to_dequeue == 0 {
            return Ok(0);
        }

        let consumer = self.consumer_atomic().load(Ordering::Relaxed);

        for (i, slot) in buffer.iter_mut().enumerate().take(to_dequeue as usize) {
            let idx = Self::ring_slot(consumer, i as u32, self.mask);
            *slot = self.read_desc(idx);
        }

        self.consumer_atomic()
            .store(consumer.wrapping_add(to_dequeue), Ordering::Release);

        Ok(to_dequeue)
    }

    /// 批量入队到预分配描述符（零 Vec 分配）
    ///
    /// # 参数
    /// * `descriptors` - 预分配的 XdpDesc 数组切片
    ///
    /// # 返回
    /// * `Result<u32>` - 实际入队数量
    pub fn enqueue_batch_from(&mut self, descriptors: &[XdpDesc]) -> Result<u32> {
        self.enqueue_batch(descriptors)
    }

    /// 回退消费者索引
    ///
    /// count 不得超过当前消费者索引，否则索引回绕到 u32::MAX 导致后续读取未初始化数据
    pub fn revert_consumer(&mut self, count: u32) {
        debug_assert!(
            count <= self.consumer_index(),
            "回退计数 {count} 超过当前消费者索引 {}",
            self.consumer_index()
        );
        self.consumer_atomic().fetch_sub(count, Ordering::Release);
    }

    /// 回退生产者索引
    ///
    /// count 不得超过当前生产者索引，否则索引回绕到 u32::MAX 导致后续写入越界
    pub fn revert_producer(&mut self, count: u32) {
        debug_assert!(
            count <= self.producer_index(),
            "回退计数 {count} 超过当前生产者索引 {}",
            self.producer_index()
        );
        self.producer_atomic().fetch_sub(count, Ordering::Release);
    }

    /// 获取容量
    #[inline]
    pub fn capacity(&self) -> u32 {
        self.capacity
    }

    /// 获取 Ring 类型
    #[inline]
    pub fn ring_type(&self) -> RingType {
        self.ring_type
    }

    /// 获取当前生产者索引
    #[inline]
    pub fn producer_index(&self) -> u32 {
        self.producer_atomic().load(Ordering::Relaxed)
    }

    /// 获取当前消费者索引
    #[inline]
    pub fn consumer_index(&self) -> u32 {
        self.consumer_atomic().load(Ordering::Relaxed)
    }

    /// 是否为内核 mmap 模式
    #[inline]
    pub fn is_kernel_mode(&self) -> bool {
        matches!(self.storage, RingStorage::Kernel(_))
    }

    /// 检查内核是否需要唤醒（XDP_RING_NEED_WAKEUP 标志）
    ///
    /// 在内核模式下，读取内核 ring 的 flags 字段。
    /// 当 `XDP_RING_NEED_WAKEUP` 标志置位时，用户态须通过 syscall 唤醒内核：
    /// `sendto(fd, NULL, 0, MSG_DONTWAIT, NULL, 0)`。
    ///
    /// 模拟模式下始终返回 false（无需唤醒）。
    #[inline]
    pub fn need_wakeup(&self) -> bool {
        match &self.storage {
            RingStorage::Kernel(k) => match k.flags {
                // SAFETY: flags 指针在 with_kernel_ring 中由内核 mmap 区域计算得出并经边界校验，
                // 映射由 KernelRing 持有（Drop 才 munmap），且仅单 Worker 访问（无数据竞争）。
                Some(ptr) => unsafe {
                    let flags = (*ptr).load(Ordering::Acquire);
                    (flags & XDP_RING_NEED_WAKEUP) != 0
                },
                None => false,
            },
            RingStorage::Simulated { .. } => false,
        }
    }
}

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

    fn make_desc(addr: u64, len: u32) -> XdpDesc {
        XdpDesc {
            addr,
            len,
            options: 0,
        }
    }

    #[test]
    fn test_ring_creation() {
        let ring = XskRing::new(RingType::Fill, 64).unwrap();
        assert_eq!(ring.capacity(), 64);
        assert_eq!(ring.ring_type(), RingType::Fill);
        assert_eq!(ring.available_data(), 0);
        assert_eq!(ring.available_space(), 64);
        assert!(!ring.is_kernel_mode());
    }

    #[test]
    fn test_ring_capacity_power_of_two() {
        let ring = XskRing::new(RingType::Rx, 10).unwrap();
        assert_eq!(ring.capacity(), 16);
    }

    #[test]
    fn test_ring_enqueue_dequeue() {
        let mut ring = XskRing::new(RingType::Tx, 16).unwrap();

        let descs = [
            make_desc(0, 64),
            make_desc(4096, 128),
            make_desc(8192, 256),
            make_desc(12288, 512),
            make_desc(16384, 1024),
        ];
        let enqueued = ring.enqueue_batch(&descs).unwrap();
        assert_eq!(enqueued, 5);
        assert_eq!(ring.available_data(), 5);
        assert_eq!(ring.available_space(), 11);

        let dequeued = ring.dequeue_batch(3).unwrap();
        assert_eq!(dequeued.len(), 3);
        assert_eq!(dequeued[0].addr, 0);
        assert_eq!(dequeued[1].addr, 4096);
        assert_eq!(dequeued[2].addr, 8192);
        assert_eq!(ring.available_data(), 2);

        let dequeued = ring.dequeue_batch(10).unwrap();
        assert_eq!(dequeued.len(), 2);
        assert_eq!(ring.available_data(), 0);
    }

    #[test]
    fn test_ring_full() {
        let mut ring = XskRing::new(RingType::Completion, 16).unwrap();
        let descs: Vec<XdpDesc> = (0..17).map(|i| make_desc(i * 4096, 0)).collect();
        let result = ring.enqueue_batch(&descs);
        assert!(result.is_err());
    }

    #[test]
    fn test_ring_wrapping() {
        let mut ring = XskRing::new(RingType::Fill, 16).unwrap();

        let descs: Vec<XdpDesc> = (0..16).map(|i| make_desc(i * 4096, 0)).collect();
        ring.enqueue_batch(&descs).unwrap();
        ring.dequeue_batch(16).unwrap();

        assert_eq!(ring.available_data(), 0);
        assert_eq!(ring.available_space(), 16);

        ring.enqueue_batch(&[make_desc(65536, 0), make_desc(69632, 0)]).unwrap();
        assert_eq!(ring.available_data(), 2);
    }

    #[test]
    fn test_ring_dequeue_to_buffer() {
        let mut ring = XskRing::new(RingType::Rx, 8).unwrap();
        let descs = [make_desc(1024, 100), make_desc(2048, 200), make_desc(3072, 300), make_desc(4096, 400)];
        ring.enqueue_batch(&descs).unwrap();

        let mut buffer = [XdpDesc::zero(); 3];
        let count = ring.dequeue_batch_to(&mut buffer).unwrap();
        assert_eq!(count, 3);
        assert_eq!(buffer[0].addr, 1024);
        assert_eq!(buffer[1].addr, 2048);
        assert_eq!(buffer[2].addr, 3072);
    }

    #[test]
    fn test_ring_revert() {
        let mut ring = XskRing::new(RingType::Tx, 16).unwrap();

        ring.enqueue_batch(&[make_desc(0, 0), make_desc(4096, 0), make_desc(8192, 0)])
            .unwrap();
        assert_eq!(ring.available_data(), 3);

        ring.revert_producer(2);
        assert_eq!(ring.available_data(), 1);

        ring.dequeue_batch(1).unwrap();
        assert_eq!(ring.available_data(), 0);

        ring.revert_consumer(1);
        assert_eq!(ring.available_data(), 1);
    }

    #[test]
    fn test_ring_empty_dequeue() {
        let mut ring = XskRing::new(RingType::Rx, 16).unwrap();
        let result = ring.dequeue_batch(10).unwrap();
        assert!(result.is_empty());
    }

    #[test]
    fn test_ring_zero_length() {
        let mut ring = XskRing::new(RingType::Fill, 16).unwrap();
        let result = ring.enqueue_batch(&[]).unwrap();
        assert_eq!(result, 0);
    }

    #[test]
    fn test_ring_capacity_minimum_16() {
        let ring = XskRing::new(RingType::Rx, 1).unwrap();
        assert_eq!(ring.capacity(), 16);
        assert!(ring.capacity().is_power_of_two());
    }

    #[test]
    fn test_ring_capacity_exact_power_of_two() {
        for cap in [16, 32, 64, 128, 256, 512, 1024, 2048, 4096] {
            let ring = XskRing::new(RingType::Tx, cap).unwrap();
            assert_eq!(ring.capacity(), cap);
            assert!(ring.capacity().is_power_of_two());
        }
    }

    #[test]
    fn test_ring_capacity_rounds_up() {
        let test_cases = vec![
            (17, 32),
            (33, 64),
            (100, 128),
            (255, 256),
            (1000, 1024),
        ];
        for (input, expected) in test_cases {
            let ring = XskRing::new(RingType::Completion, input).unwrap();
            assert_eq!(
                ring.capacity(),
                expected,
                "Input {} should round up to {}",
                input,
                expected
            );
            assert!(ring.capacity().is_power_of_two());
        }
    }

    #[test]
    fn test_ring_new_fail_closed_on_invalid_capacity() {
        // Fail-Closed：capacity == 0 必须返回错误（不再被 max(16) 静默掩盖）
        assert!(XskRing::new(RingType::Rx, 0).is_err());
        // Fail-Closed：capacity > 2^31 时 next_power_of_two 会 debug panic /
        // release 回绕为 0，必须显式返回错误而非 panic/回绕
        assert!(XskRing::new(RingType::Rx, u32::MAX).is_err());
        assert!(XskRing::new(RingType::Rx, (1u32 << 31) + 1).is_err());
        // Fail-Closed：超出模拟模式堆分配上限必须返回错误而非 OOM abort——
        // 2^31 项 × 16B = 32 GiB 的真实堆分配在容量受限环境直接 abort（alloc_error_handler），
        // 连错误都返回不了；MAX_SIMULATED_CAPACITY 是 Simulated 模式可安全分配的上界
        assert!(XskRing::new(RingType::Rx, 1u32 << 31).is_err());
        assert!(XskRing::new(RingType::Rx, MAX_SIMULATED_CAPACITY + 1).is_err());
        // 边界：上限本身（2^20 = 1M 项 ≤ 16 MiB）合法
        assert!(XskRing::new(RingType::Rx, MAX_SIMULATED_CAPACITY).is_ok());
    }

    #[test]
    fn test_ring_empty_boundary() {
        let mut ring = XskRing::new(RingType::Rx, 32).unwrap();
        assert_eq!(ring.available_data(), 0);
        assert_eq!(ring.available_space(), 32);

        let result = ring.dequeue_batch(10).unwrap();
        assert!(result.is_empty());
        assert_eq!(ring.consumer_index(), 0);
    }

    #[test]
    fn test_ring_full_boundary() {
        let mut ring = XskRing::new(RingType::Tx, 16).unwrap();
        let descs: Vec<XdpDesc> = (0..16).map(|i| make_desc(i * 4096, 64)).collect();

        let enqueued = ring.enqueue_batch(&descs).unwrap();
        assert_eq!(enqueued, 16);
        assert_eq!(ring.available_data(), 16);
        assert_eq!(ring.available_space(), 0);

        let overflow = ring.enqueue_batch(&[make_desc(65536, 0)]);
        assert!(overflow.is_err());
    }

    #[test]
    fn test_ring_full_then_partial_consume() {
        let mut ring = XskRing::new(RingType::Rx, 16).unwrap();
        let descs: Vec<XdpDesc> = (0..16).map(|i| make_desc(i * 4096, 64)).collect();
        ring.enqueue_batch(&descs).unwrap();
        assert_eq!(ring.available_space(), 0);

        let consumed = ring.dequeue_batch(5).unwrap();
        assert_eq!(consumed.len(), 5);
        assert_eq!(ring.available_space(), 5);
        assert_eq!(ring.available_data(), 11);
    }

    #[test]
    fn test_ring_index_wrapping_producer() {
        let mut ring = XskRing::new(RingType::Fill, 16).unwrap();

        for i in 0..16 {
            ring.enqueue_batch(&[make_desc(i * 4096, 0)]).unwrap();
        }
        assert_eq!(ring.producer_index(), 16);

        ring.dequeue_batch(8).unwrap();
        assert_eq!(ring.consumer_index(), 8);

        for i in 16..24 {
            ring.enqueue_batch(&[make_desc(i * 4096, 0)]).unwrap();
        }
        assert_eq!(ring.producer_index(), 24);
        assert_eq!(ring.available_data(), 16);
    }

    #[test]
    fn test_ring_index_wrapping_consumer() {
        let mut ring = XskRing::new(RingType::Rx, 16).unwrap();

        for _ in 0..3 {
            let descs: Vec<XdpDesc> = (0..16).map(|i| make_desc(i * 4096, 0)).collect();
            ring.enqueue_batch(&descs).unwrap();
            let consumed = ring.dequeue_batch(16).unwrap();
            assert_eq!(consumed.len(), 16);
        }

        assert_eq!(ring.producer_index(), 48);
        assert_eq!(ring.consumer_index(), 48);
        assert_eq!(ring.available_data(), 0);
    }

    #[test]
    fn test_ring_batch_exceeds_available_space() {
        let mut ring = XskRing::new(RingType::Tx, 32).unwrap();
        let descs: Vec<XdpDesc> = (0..10).map(|i| make_desc(i * 4096, 0)).collect();
        ring.enqueue_batch(&descs).unwrap();

        assert_eq!(ring.available_space(), 22);

        let too_many: Vec<XdpDesc> = (0..30).map(|i| make_desc(i * 4096, 0)).collect();
        let result = ring.enqueue_batch(&too_many);
        assert!(result.is_err());
    }

    #[test]
    fn test_ring_dequeue_batch_to_empty_buffer() {
        let mut ring = XskRing::new(RingType::Rx, 16).unwrap();
        let descs = [make_desc(1024, 100)];
        ring.enqueue_batch(&descs).unwrap();

        let mut buffer = [];
        let count = ring.dequeue_batch_to(&mut buffer).unwrap();
        assert_eq!(count, 0);
    }

    #[test]
    fn test_ring_enqueue_batch_from_alias() {
        let mut ring = XskRing::new(RingType::Fill, 16).unwrap();
        let descs = [make_desc(4096, 128), make_desc(8192, 256)];
        let count = ring.enqueue_batch_from(&descs).unwrap();
        assert_eq!(count, 2);
        assert_eq!(ring.available_data(), 2);
    }

    #[test]
    fn test_ring_revert_producer_zero() {
        let mut ring = XskRing::new(RingType::Tx, 16).unwrap();
        ring.revert_producer(0);
        assert_eq!(ring.producer_index(), 0);
        assert_eq!(ring.available_data(), 0);
    }

    #[test]
    fn test_ring_revert_consumer_zero() {
        let mut ring = XskRing::new(RingType::Rx, 16).unwrap();
        ring.revert_consumer(0);
        assert_eq!(ring.consumer_index(), 0);
        assert_eq!(ring.available_data(), 0);
    }

    #[test]
    fn test_ring_types_all_work() {
        let types = vec![
            RingType::Fill,
            RingType::Rx,
            RingType::Tx,
            RingType::Completion,
        ];
        for ring_type in types {
            let mut ring = XskRing::new(ring_type, 16).unwrap();
            assert_eq!(ring.ring_type(), ring_type);
            let desc = make_desc(0, 0);
            ring.enqueue_batch(&[desc]).unwrap();
            let result = ring.dequeue_batch(1).unwrap();
            assert_eq!(result.len(), 1);
        }
    }

    #[test]
    fn test_ring_fifo_order() {
        let mut ring = XskRing::new(RingType::Rx, 32).unwrap();
        let descs: Vec<XdpDesc> = (0..10u32).map(|i| make_desc(i as u64 * 4096, i * 100)).collect();

        ring.enqueue_batch(&descs).unwrap();
        let dequeued = ring.dequeue_batch(10).unwrap();

        for (i, desc) in dequeued.iter().enumerate() {
            assert_eq!(desc.addr, i as u64 * 4096);
            assert_eq!(desc.len, i as u32 * 100);
        }
    }

    #[test]
    fn test_need_wakeup_simulated_mode() {
        // 模拟模式下 need_wakeup 始终返回 false（无内核需要唤醒）
        let ring = XskRing::new(RingType::Tx, 64).unwrap();
        assert!(!ring.need_wakeup());

        let fill_ring = XskRing::new(RingType::Fill, 64).unwrap();
        assert!(!fill_ring.need_wakeup());
    }

    #[test]
    fn test_xdp_ring_need_wakeup_constant() {
        // XDP_RING_NEED_WAKEUP = 0x1（内核 UAPI if_xdp.h: (1 << 0)，固化防回归）
        assert_eq!(XDP_RING_NEED_WAKEUP, 0x1);
    }

    #[test]
    fn test_ring_slot_wrapping_pure_logic() {
        // 纯逻辑：u32 索引在 u32::MAX 边界自然回绕，mask 保证落入 [0, capacity)
        assert_eq!(XskRing::ring_slot(0, 0, 15), 0);
        assert_eq!(XskRing::ring_slot(u32::MAX, 1, 15), 0); // MAX+1 回绕到 0
        assert_eq!(XskRing::ring_slot(u32::MAX - 1, 3, 15), 1); // (MAX-1)+3 回绕到 1
        assert_eq!(XskRing::ring_slot(u32::MAX - 2, 4, 15), 1); // (MAX-2)+4 回绕到 1
        assert_eq!(XskRing::ring_slot(5, 3, 15), 8);
        // mask 为 2 的幂减 1 时结果恒 < capacity
        for base in [0u32, 1, 15, 16, u32::MAX - 7, u32::MAX] {
            for off in [0u32, 1, 7, 16, 100] {
                assert!(XskRing::ring_slot(base, off, 15) < 16);
            }
        }
    }

    #[test]
    fn test_simulated_ring_index_wrap_at_u32_max() {
        // 模拟模式：producer/consumer 推到 u32::MAX 附近，验证跨回绕边界 FIFO 正确
        let mut ring = XskRing::new(RingType::Tx, 16).unwrap();
        if let RingStorage::Simulated {
            producer_idx,
            consumer_idx,
            ..
        } = &ring.storage
        {
            producer_idx.store(u32::MAX - 2, Ordering::SeqCst);
            consumer_idx.store(u32::MAX - 2, Ordering::SeqCst);
        } else {
            panic!("new() 必须是模拟模式");
        }

        let descs: Vec<XdpDesc> = (0..4u32).map(|i| make_desc(u64::from(i) * 4096, i)).collect();
        assert_eq!(ring.enqueue_batch(&descs).unwrap(), 4);
        assert_eq!(ring.available_data(), 4);
        // producer 已回绕：(MAX-2) + 4 = MAX+2 → 1
        assert_eq!(ring.producer_index(), 1);

        let out = ring.dequeue_batch(4).unwrap();
        assert_eq!(out.len(), 4);
        for (i, d) in out.iter().enumerate() {
            assert_eq!(d.addr, (i as u64) * 4096);
        }
        assert_eq!(ring.consumer_index(), 1);
        assert_eq!(ring.available_data(), 0);
        assert_eq!(ring.available_space(), 16);
    }

    /// 构造匿名 mmap 上的「假内核 ring」布局（无需真实 AF_XDP socket）
    ///
    /// 返回 (mmap 基址, RingOffsets)。映射所有权随 with_kernel_ring 转移，
    /// 由 XskRing Drop 时 munmap。
    fn fake_kernel_ring_layout(ring_size: u32, flags_value: u32) -> (*mut u8, RingOffsets) {
        let desc_off = 64u64;
        let len = desc_off + u64::from(ring_size) * std::mem::size_of::<XdpDesc>() as u64;
        let map_len = len as usize;
        // SAFETY: 测试内匿名 mmap，参数均为有效常量；下方检查 MAP_FAILED。
        let base = unsafe {
            libc::mmap(
                std::ptr::null_mut(),
                map_len,
                libc::PROT_READ | libc::PROT_WRITE,
                libc::MAP_SHARED | libc::MAP_ANONYMOUS,
                -1,
                0,
            )
        };
        assert_ne!(base, libc::MAP_FAILED, "匿名 mmap 失败");
        let base = base as *mut u8;
        // SAFETY: base 指向刚映射的 map_len 字节区域，偏移 16..20 在区域内。
        unsafe {
            (base.add(16) as *mut u32).write(flags_value);
        }
        (
            base,
            RingOffsets {
                producer: 0,
                consumer: 8,
                desc: desc_off,
                flags: 16,
                len,
            },
        )
    }

    #[test]
    fn test_kernel_mode_ring_enqueue_dequeue_via_mmap() {
        let (base, offsets) = fake_kernel_ring_layout(4, 0);
        // SAFETY: base 指向有效的匿名映射区，offsets 布局经手工构造且在映射区内；
        // 测试单线程访问，满足单 Owner 契约。
        let mut ring = unsafe { XskRing::with_kernel_ring(RingType::Tx, base, offsets) }
            .expect("合法偏移必须成功");

        assert!(ring.is_kernel_mode());
        assert_eq!(ring.capacity(), 4);
        assert_eq!(ring.available_space(), 4);

        let descs = [make_desc(4096, 64), make_desc(8192, 128)];
        assert_eq!(ring.enqueue_batch(&descs).unwrap(), 2);

        // 验证数据真实写入内核共享区：producer == 2，desc[0].addr == 4096
        // SAFETY: 读取本测试刚写入的映射区，偏移均在映射长度内。
        unsafe {
            assert_eq!((base as *const u32).read(), 2, "producer 索引应写入共享区");
            let d0 = (base.add(64) as *const XdpDesc).read();
            assert_eq!(d0.addr, 4096);
            assert_eq!(d0.len, 64);
            let d1 = (base.add(64 + 16) as *const XdpDesc).read();
            assert_eq!(d1.addr, 8192);
        }

        let out = ring.dequeue_batch(4).unwrap();
        assert_eq!(out.len(), 2);
        assert_eq!(out[0].addr, 4096);
        assert_eq!(out[1].addr, 8192);
        // SAFETY: 读取共享区 consumer 索引。
        unsafe {
            assert_eq!((base.add(8) as *const u32).read(), 2, "consumer 索引应写入共享区");
        }
        // ring Drop 时自动 munmap（映射所有权已转移）
    }

    #[test]
    fn test_kernel_mode_ring_need_wakeup_flag() {
        let (base, offsets) = fake_kernel_ring_layout(4, XDP_RING_NEED_WAKEUP);
        // SAFETY: 同上一个测试，映射有效、单线程。
        let ring = unsafe { XskRing::with_kernel_ring(RingType::Fill, base, offsets) }
            .expect("合法偏移必须成功");
        assert!(ring.need_wakeup(), "flags 置位时 need_wakeup 必须为 true");

        // 清除 flags 后应为 false
        // SAFETY: base 仍有效（ring 未 drop），偏移 16 在映射区内。
        unsafe {
            (base.add(16) as *mut u32).write(0u32);
        }
        assert!(!ring.need_wakeup());
    }

    #[test]
    fn test_kernel_mode_ring_no_flags_field() {
        // 旧内核（< 5.4）无 flags 字段：offsets.flags == 0 → need_wakeup 恒 false
        let (base, mut offsets) = fake_kernel_ring_layout(4, 0);
        offsets.flags = 0;
        // SAFETY: 映射有效、单线程。
        let ring = unsafe { XskRing::with_kernel_ring(RingType::Rx, base, offsets) }
            .expect("合法偏移必须成功");
        assert!(!ring.need_wakeup());
    }

    #[test]
    fn test_kernel_mode_ring_rejects_invalid_offsets() {
        let (base, offsets) = fake_kernel_ring_layout(4, 0);

        // 空指针
        let r = unsafe { XskRing::with_kernel_ring(RingType::Rx, std::ptr::null_mut(), offsets) };
        assert!(r.is_err(), "空 mmap_base 必须拒绝");

        // desc > len（下溢路径）
        let bad = RingOffsets { desc: 4096, ..offsets };
        let r = unsafe { XskRing::with_kernel_ring(RingType::Rx, base, bad) };
        assert!(r.is_err(), "desc > len 必须拒绝");

        // 容量非 2 的幂（len - desc = 3 * 16）
        let bad = RingOffsets { len: offsets.desc + 3 * 16, ..offsets };
        let r = unsafe { XskRing::with_kernel_ring(RingType::Rx, base, bad) };
        assert!(r.is_err(), "非 2 的幂容量必须拒绝");

        // producer 偏移越界
        let bad = RingOffsets { producer: offsets.len, ..offsets };
        let r = unsafe { XskRing::with_kernel_ring(RingType::Rx, base, bad) };
        assert!(r.is_err(), "producer 越界必须拒绝");

        // flags 偏移越界
        let bad = RingOffsets { flags: offsets.len + 4, ..offsets };
        let r = unsafe { XskRing::with_kernel_ring(RingType::Rx, base, bad) };
        assert!(r.is_err(), "flags 越界必须拒绝");

        // 清理：上述 Err 路径未接管映射，需手动 munmap
        // SAFETY: base 指向本测试映射的 offsets.len 字节区域，且未转移所有权。
        unsafe {
            libc::munmap(base as *mut libc::c_void, offsets.len as usize);
        }
    }

    #[test]
    fn test_kernel_mode_ring_wrap_across_u32_max() {
        let (base, offsets) = fake_kernel_ring_layout(16, 0);
        // SAFETY: 映射有效、单线程。
        let mut ring = unsafe { XskRing::with_kernel_ring(RingType::Completion, base, offsets) }
            .expect("合法偏移必须成功");

        // 直接把共享区 producer/consumer 写到 u32::MAX 附近（模拟内核长期运行后的状态）
        // SAFETY: base 有效，偏移 0/8 在映射区内。
        unsafe {
            (base as *mut u32).write(u32::MAX - 1);
            (base.add(8) as *mut u32).write(u32::MAX - 1);
        }
        assert_eq!(ring.available_data(), 0);
        assert_eq!(ring.available_space(), 16);

        let descs: Vec<XdpDesc> = (0..4u32).map(|i| make_desc(u64::from(i) * 4096, i)).collect();
        assert_eq!(ring.enqueue_batch(&descs).unwrap(), 4);
        // producer 回绕：(MAX-1) + 4 = 2
        assert_eq!(ring.producer_index(), 2);
        assert_eq!(ring.available_data(), 4);

        let out = ring.dequeue_batch(4).unwrap();
        assert_eq!(out.len(), 4);
        for (i, d) in out.iter().enumerate() {
            assert_eq!(d.addr, (i as u64) * 4096, "跨回绕边界 FIFO 顺序必须保持");
        }
        assert_eq!(ring.consumer_index(), 2);
    }
}