#![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;
const PREFILL_CHUNK: usize = 64;
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,
}
}
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,
}
}
#[derive(Debug)]
struct RingMmapGuard {
entries: Vec<RingMmapEntry>,
}
#[derive(Debug)]
struct RingMmapEntry {
base: *mut u8,
len: usize,
disarmed: bool,
}
impl RingMmapGuard {
fn new() -> Self {
Self {
entries: Vec::new(),
}
}
fn register(&mut self, base: *mut u8, len: usize) -> usize {
self.entries.push(RingMmapEntry {
base,
len,
disarmed: false,
});
self.entries.len() - 1
}
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;
}
guard_unmap(entry.base, entry.len);
}
}
}
#[cfg(not(test))]
fn guard_unmap(base: *mut u8, len: usize) {
unsafe {
libc::munmap(base as *mut libc::c_void, len);
}
}
#[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;
}
unsafe {
libc::munmap(base as *mut libc::c_void, len);
}
}
#[cfg(test)]
thread_local! {
static GUARD_TEST_STUB: std::cell::Cell<bool> = const { std::cell::Cell::new(false) };
static GUARD_UNMAP_LOG: std::cell::RefCell<Vec<(usize, usize)>> = const { std::cell::RefCell::new(Vec::new()) };
}
#[derive(Debug, Clone)]
pub struct XskConfig {
pub ifindex: u32,
pub queue_id: u32,
pub zero_copy: bool,
pub fill_ring_size: u32,
pub rx_ring_size: u32,
pub tx_ring_size: u32,
pub completion_ring_size: u32,
pub shared_umem: bool,
pub frame_size: u32,
pub headroom: u32,
pub so_rcvbuf: i32,
pub so_sndbuf: i32,
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,
}
}
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum XskState {
Created,
Configured,
Bound,
Closed,
}
pub struct XskSocket {
fd: Option<i32>,
config: XskConfig,
state: XskState,
umem: Option<Arc<UmemManager>>,
fill_ring: XskRing,
rx_ring: XskRing,
tx_ring: XskRing,
completion_ring: XskRing,
descriptor_engine: DescriptorEngine,
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 {
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;
let total_descriptors = u64::from(fill_ring_size)
+ u64::from(rx_ring_size)
+ u64::from(tx_ring_size)
+ u64::from(completion_ring_size);
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),
})
}
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)
}
pub fn create_socket(&mut self) -> Result<()> {
if self.state != XskState::Created {
return Err(XskError::InvalidState(format!(
"create_socket expects state=Created, actual state={:?}",
self.state
))
.into());
}
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(())
}
pub fn configure(&mut self) -> Result<()> {
let fd = self.get_fd()?;
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());
}
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(())
}
fn set_ring_size(&self, optname: libc::c_int, size: u32) -> Result<()> {
let fd = self.get_fd()?;
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(())
}
fn get_mmap_offsets(fd: i32) -> Result<libc::xdp_mmap_offsets> {
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;
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)
}
fn mmap_kernel_ring(fd: i32, pgoff: libc::off_t, length: usize, name: &str) -> Result<*mut u8> {
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)
}
pub fn bind(&mut self, umem: Arc<UmemManager>) -> Result<()> {
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());
}
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());
}
let frame_size = self.config.frame_size;
let umem_frames = (umem.size() / frame_size as usize).min(u32::MAX as usize) as u32;
let new_engine =
DescriptorEngine::with_frame_shift(u64::from(umem_frames), frame_size.trailing_zeros())?;
let fd = self.get_fd()?;
let reg = build_umem_reg(
umem.as_ptr(),
umem.size(),
self.config.frame_size,
self.config.headroom,
);
let ret = unsafe {
libc::setsockopt(
fd,
libc::SOL_XDP,
libc::XDP_UMEM_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());
}
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)?;
let offsets = Self::get_mmap_offsets(fd)?;
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);
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);
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;
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;
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());
}
self.fill_ring = fill_ring;
self.rx_ring = rx_ring;
self.tx_ring = tx_ring;
self.completion_ring = completion_ring;
self.descriptor_engine = new_engine;
self.prefilled_frames = 0;
self.umem = Some(umem);
self.state = XskState::Bound;
self.active.store(true, Ordering::SeqCst);
self.prefill_fill_ring()?;
Ok(())
}
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), };
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;
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);
let batch = ((target - frame_base) as usize).min(chunk_size);
for (j, slot) in chunk.iter_mut().enumerate().take(batch) {
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; }
}
self.prefilled_frames = start.saturating_add(filled);
if filled > 0 && self.fill_ring.need_wakeup() {
self.notify_fill()?;
}
Ok(filled)
}
#[inline]
pub fn is_kernel_mode(&self) -> bool {
self.rx_ring.is_kernel_mode()
}
pub fn close(&mut self) -> Result<()> {
self.active.store(false, Ordering::SeqCst);
if let Some(fd) = self.fd.take() {
let ret = unsafe { libc::close(fd) };
if ret < 0 {
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(())
}
#[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)
}
#[inline]
pub fn fill_ring(&self) -> &XskRing {
&self.fill_ring
}
#[inline]
pub fn fill_ring_mut(&mut self) -> &mut XskRing {
&mut self.fill_ring
}
#[inline]
pub fn rx_ring(&self) -> &XskRing {
&self.rx_ring
}
#[inline]
pub fn rx_ring_mut(&mut self) -> &mut XskRing {
&mut self.rx_ring
}
#[inline]
pub fn tx_ring(&self) -> &XskRing {
&self.tx_ring
}
#[inline]
pub fn tx_ring_mut(&mut self) -> &mut XskRing {
&mut self.tx_ring
}
#[inline]
pub fn completion_ring(&self) -> &XskRing {
&self.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
}
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 {
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) => {
for &idx in &allocated_indices {
self.descriptor_engine.rollback_allocate(idx);
}
return Err(e);
}
};
if self.fill_ring.need_wakeup() {
self.notify_fill()?;
}
Ok(filled)
}
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 {
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)
}
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)
}
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)?;
if self.tx_ring.need_wakeup() {
self.notify_tx()?;
}
Ok(sent)
}
pub fn send_xdp_batch(&mut self, descriptors: &[XdpDesc]) -> Result<u32> {
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)
}
#[inline]
pub fn notify_tx(&self) -> Result<()> {
let fd = match self.fd {
Some(f) if f >= 0 => f,
_ => return Ok(()), };
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();
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(())
}
#[inline]
pub fn notify_fill(&self) -> Result<()> {
let fd = match self.fd {
Some(f) if f >= 0 => f,
_ => return Ok(()),
};
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();
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(())
}
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 {
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)
}
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)
}
pub fn umem(&self) -> Option<&Arc<UmemManager>> {
self.umem.as_ref()
}
pub fn get_fd(&self) -> Result<i32> {
self.fd.ok_or_else(|| XskError::NotBound.into())
}
#[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();
let filled = socket.fill_descriptors(4).unwrap();
assert_eq!(filled, 4);
assert_eq!(socket.fill_ring().available_data(), 4);
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();
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();
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();
let _ = socket.descriptor_engine().allocate(0, 42).unwrap();
let _ = socket.descriptor_engine().allocate(1, 42).unwrap();
let xdp_descs = [
XdpDesc {
addr: 0, len: 100,
options: 0,
},
XdpDesc {
addr: 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() {
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");
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() {
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 偏移 + 全部描述符"
);
assert_eq!(std::mem::size_of::<XdpDesc>(), 16);
}
#[test]
fn test_xsk_new_is_simulation_mode() {
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() {
let mut socket = match XskSocket::new(XskConfig::default()) {
Ok(s) => s,
Err(_) => return,
};
if socket.create_socket().is_err() {
return; }
let umem = Arc::new(
UmemManager::new(crate::umem::UmemConfig {
size: 4096 * 16,
hugepage: false,
locked: false,
shared: false,
})
.unwrap(),
);
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() {
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) => {
assert!(s.is_kernel_mode());
assert_eq!(s.state(), XskState::Bound);
}
Err(e) => {
let msg = format!("{e}");
assert!(!msg.is_empty());
}
}
}
#[test]
fn test_prefill_fill_ring_kernel_mode_logic() {
let config = XskConfig {
fill_ring_size: 16,
frame_size: 4096,
..Default::default()
};
let mut socket = XskSocket::new(config).unwrap();
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));
let desc_off = 64u64;
let ring_size = 16u32;
let len = desc_off + u64::from(ring_size) * 16;
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,
};
socket.fill_ring = unsafe {
XskRing::with_kernel_ring(RingType::Fill, base, offsets)
}
.unwrap();
let filled = socket.prefill_fill_ring().unwrap();
assert_eq!(filled, 8);
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);
}
}
assert!(socket.descriptor_engine().verify_conservation());
}
#[test]
fn test_prefill_fill_ring_simulation_mode_no_umem() {
let mut socket = XskSocket::new(XskConfig::default()).unwrap();
assert_eq!(socket.prefill_fill_ring().unwrap(), 0);
}
fn anon_mmap_page() -> *mut u8 {
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
}
fn page_is_mapped(addr: *const u8) -> bool {
let mut vec = [0u8; 1];
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() {
let base = anon_mmap_page();
{
let mut guard = RingMmapGuard::new();
let slot = guard.register(base, 4096);
guard.disarm(slot);
} unsafe {
base.write(0xAB);
assert_eq!(base.read(), 0xAB, "disarm 后映射必须仍可访问");
}
assert!(page_is_mapped(base), "disarm 的映射不得被 munmap");
unsafe {
libc::munmap(base as *mut libc::c_void, 4096);
}
}
#[test]
fn test_ring_mmap_guard_failure_path_munmaps_all() {
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);
}
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());
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() {
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());
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() {
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());
}
}