zenith_linux/

syscalls.rs

//! Linux 高性能零拷贝/批量系统调用封装
//!
//! 本模块集中收敛与「全链路最高性能最低消耗」相关的 Linux 内核系统调用封装：
//!
//! - `splice(2)`：在两个 fd 之间内核态零拷贝移动数据（pipe ↔ socket）
//! - `sendfile(2)`：从文件 fd 直接发送到 socket fd，绕过用户态缓冲区
//! - `recvmmsg(2)`：单次系统调用接收多个 UDP/UDP-Lite 数据报，降低系统调用开销
//! - `sendmmsg(2)`：单次系统调用发送多个 UDP/UDP-Lite 数据报
//! - `pipe2(2)`：创建用于 splice 中继的管道对
//!
//! # 设计原则（规范 §1.1 / §6.1）
//! - **Fail-Closed**：任何系统调用失败立即返回错误，绝不静默降级
//! - **零堆分配热路径**：批量数组使用栈上数组 `[libc::mmsghdr; N]`
//! - **checked 算术**：所有偏移、长度计算使用 `checked_*` 防止溢出
//! - **unsafe 隔离**：所有 unsafe 集中于本文件，附 `// SAFETY:` 注释
//! - **错误语义完整**：errno → `LinuxError::Syscall`，保留 syscall 名与 errno
//!
//! # 极致性能
//! - 单次 `splice` 可传输最多 2^31-1 字节（Linux 限制）
//! - `recvmmsg/sendmmsg` 单次最多 64 个数据报（`UIO_MAXIOV`）
//! - 与 AF_XDP 路径互补：内核套接字场景的最优解

#![allow(unsafe_code)]

use crate::error::{LinuxError, Result};

/// 批量收发最大数据报数（与 `UIO_MAXIOV` 对齐，glibc 上限）
pub const MAX_BATCH_DATAGRAMS: usize = 64;

/// splice(2) 标志位
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
#[repr(u32)]
pub enum SpliceFlags {
    /// 默认（无标志）
    None = 0,
    /// 移动数据（仅引用计数迁移，零拷贝）
    Move = libc::SPLICE_F_MOVE,
    /// 非阻塞
    NonBlock = libc::SPLICE_F_NONBLOCK,
    /// 不阻塞，返回 EAGAIN
    More = libc::SPLICE_F_MORE,
}

impl SpliceFlags {
    /// 合并为原始 u32 标志
    #[inline]
    pub fn bits(self) -> u32 {
        self as u32
    }
}

/// splice(2) 在两个 fd 之间内核态零拷贝移动数据。
///
/// 至少一个 fd 必须是管道（pipe）。常用于 socket ↔ pipe ↔ socket 中继。
///
/// # 参数
/// * `fd_in` - 输入 fd（socket / file / pipe）
/// * `off_in` - 输入偏移指针，None 表示使用当前文件偏移
/// * `fd_out` - 输出 fd
/// * `off_out` - 输出偏移指针，None 表示使用当前文件偏移
/// * `len` - 最多移动的字节数（上限 2^31-1）
/// * `flags` - splice 标志
///
/// # 返回
/// 实际移动的字节数（0 表示 EOF）
///
/// # 错误
/// - `LinuxError::Syscall { "splice", errno }` - 系统调用失败
pub fn splice(
    fd_in: i32,
    off_in: Option<&mut i64>,
    fd_out: i32,
    off_out: Option<&mut i64>,
    len: usize,
    flags: u32,
) -> Result<usize> {
    // Linux splice 长度上限为 2^31-1（splice 内部用 ssize_t）
    let safe_len = len.min(0x7FFF_FFFFusize);

    let off_in_ptr = off_in.map_or(std::ptr::null_mut(), |p| p as *mut i64);
    let off_out_ptr = off_out.map_or(std::ptr::null_mut(), |p| p as *mut i64);

    // SAFETY: splice(2) 系统调用。fd_in/fd_out 必须是有效的打开文件描述符，
    // 至少一个为管道。off_in/off_out 可空，非空时指向 64 位有符号偏移。
    // safe_len 已被钳制到 Linux 上限内。flags 直接传递内核。
    let ret = unsafe {
        libc::splice(
            fd_in,
            off_in_ptr,
            fd_out,
            off_out_ptr,
            safe_len,
            flags,
        )
    };

    if ret < 0 {
        Err(LinuxError::Syscall {
            syscall: "splice",
            errno: std::io::Error::last_os_error().raw_os_error().unwrap_or(0),
        })
    } else {
        Ok(ret as usize)
    }
}

/// sendfile(2) 从文件 fd 直接发送到 socket fd，绕过用户态缓冲区。
///
/// 适用于静态文件服务（HTTP 静态资源、缓存持久化等）。
///
/// # 参数
/// * `out_fd` - 输出 fd（必须是 socket 或类似可写 fd）
/// * `in_fd` - 输入 fd（必须是支持 sendfile 的文件 fd，不能是 socket）
/// * `offset` - 输入偏移指针，None 表示使用当前文件偏移
/// * `count` - 最多发送的字节数
///
/// # 返回
/// 实际发送的字节数（0 表示 EOF）
pub fn sendfile(
    out_fd: i32,
    in_fd: i32,
    offset: Option<&mut i64>,
    count: usize,
) -> Result<usize> {
    let off_ptr = offset.map_or(std::ptr::null_mut(), |p| p as *mut i64);

    // SAFETY: sendfile(2) 系统调用。out_fd 通常是 socket，in_fd 必须是支持
    // mmap 的文件（如普通文件、块设备）。off_ptr 可空。count 不超过 SSIZE_MAX。
    let ret = unsafe { libc::sendfile(out_fd, in_fd, off_ptr, count.min(0x7FFF_FFFFusize)) };

    if ret < 0 {
        Err(LinuxError::Syscall {
            syscall: "sendfile",
            errno: std::io::Error::last_os_error().raw_os_error().unwrap_or(0),
        })
    } else {
        Ok(ret as usize)
    }
}

/// pipe2(2) 创建管道对，用于 splice 中继。
///
/// # 参数
/// * `flags` - 管道标志（如 `O_NONBLOCK`）
///
/// # 返回
/// (read_fd, write_fd)
pub fn pipe2(flags: i32) -> Result<(i32, i32)> {
    let mut fds = [0i32; 2];

    // SAFETY: pipe2(2) 系统调用。fds 数组长度为 2，flags 直接传递内核。
    let ret = unsafe { libc::pipe2(fds.as_mut_ptr(), flags) };

    if ret < 0 {
        Err(LinuxError::Syscall {
            syscall: "pipe2",
            errno: std::io::Error::last_os_error().raw_os_error().unwrap_or(0),
        })
    } else {
        Ok((fds[0], fds[1]))
    }
}

/// recvmmsg(2) 批量接收多个数据报（UDP/UDP-Lite）。
///
/// 单次系统调用最多接收 `bufs.len()` 个数据报（上限 64），显著降低 PPS 场景系统调用开销。
///
/// # 参数
/// * `fd` - UDP socket fd
/// * `bufs` - 接收缓冲区切片（每个元素对应一个数据报）
/// * `flags` - recvmsg 标志（如 `MSG_DONTWAIT`）
///
/// # 返回
/// 实际接收的数据报数量（0 表示无数据）
pub fn recvmmsg(bufs: &mut [&mut [u8]], fd: i32, flags: i32) -> Result<usize> {
    let count = bufs.len().min(MAX_BATCH_DATAGRAMS);
    if count == 0 {
        return Ok(0);
    }

    // 栈上预分配 mmsghdr 数组，零堆分配
    let mut msgs = [libc::mmsghdr {
        msg_hdr: libc::msghdr {
            msg_name: std::ptr::null_mut(),
            msg_namelen: 0,
            msg_iov: std::ptr::null_mut(),
            msg_iovlen: 0,
            msg_control: std::ptr::null_mut(),
            msg_controllen: 0,
            msg_flags: 0,
        },
        msg_len: 0,
    }; MAX_BATCH_DATAGRAMS];

    // 栈上预分配 iovec 数组
    let mut iovs = [libc::iovec {
        iov_base: std::ptr::null_mut(),
        iov_len: 0,
    }; MAX_BATCH_DATAGRAMS];

    // 填充 iovs 指向用户缓冲区
    for i in 0..count {
        iovs[i] = libc::iovec {
            iov_base: bufs[i].as_mut_ptr() as *mut std::ffi::c_void,
            iov_len: bufs[i].len(),
        };
        msgs[i].msg_hdr.msg_iov = &mut iovs[i];
        msgs[i].msg_hdr.msg_iovlen = 1;
    }

    // SAFETY: recvmmsg(2) 系统调用。msgs 数组长度 ≥ count，每个 msg_hdr.msg_iov
    // 指向一个有效的 iovec，iovec 指向用户提供的可写缓冲区。
    // timeout 传空表示阻塞（或受 flags 中 MSG_DONTWAIT 控制）。
    let ret = unsafe {
        libc::recvmmsg(
            fd,
            msgs.as_mut_ptr(),
            count as u32,
            flags,
            std::ptr::null_mut(),
        )
    };

    if ret < 0 {
        Err(LinuxError::Syscall {
            syscall: "recvmmsg",
            errno: std::io::Error::last_os_error().raw_os_error().unwrap_or(0),
        })
    } else {
        Ok(ret as usize)
    }
}

/// sendmmsg(2) 批量发送多个数据报（UDP/UDP-Lite）。
///
/// 单次系统调用最多发送 `bufs.len()` 个数据报（上限 64）。
///
/// # 参数
/// * `fd` - UDP socket fd
/// * `bufs` - 发送缓冲区切片（每个元素对应一个数据报）
/// * `dest` - 目标地址（所有数据报发往同一地址）
/// * `flags` - sendmsg 标志（如 `MSG_DONTWAIT`）
///
/// # 返回
/// 实际发送的数据报数量
pub fn sendmmsg(bufs: &[&[u8]], fd: i32, dest: &libc::sockaddr_storage, flags: i32) -> Result<usize> {
    let count = bufs.len().min(MAX_BATCH_DATAGRAMS);
    if count == 0 {
        return Ok(0);
    }

    let mut msgs = [libc::mmsghdr {
        msg_hdr: libc::msghdr {
            msg_name: std::ptr::null_mut(),
            msg_namelen: 0,
            msg_iov: std::ptr::null_mut(),
            msg_iovlen: 0,
            msg_control: std::ptr::null_mut(),
            msg_controllen: 0,
            msg_flags: 0,
        },
        msg_len: 0,
    }; MAX_BATCH_DATAGRAMS];

    let mut iovs = [libc::iovec {
        iov_base: std::ptr::null_mut(),
        iov_len: 0,
    }; MAX_BATCH_DATAGRAMS];

    // 目标地址转换为原始字节指针（所有数据报共用同一目标）
    let dest_ptr = dest as *const libc::sockaddr_storage as *mut std::ffi::c_void;
    let dest_len = std::mem::size_of::<libc::sockaddr_storage>() as u32;

    for i in 0..count {
        iovs[i] = libc::iovec {
            iov_base: bufs[i].as_ptr() as *mut std::ffi::c_void,
            iov_len: bufs[i].len(),
        };
        msgs[i].msg_hdr.msg_name = dest_ptr;
        msgs[i].msg_hdr.msg_namelen = dest_len;
        msgs[i].msg_hdr.msg_iov = &mut iovs[i];
        msgs[i].msg_hdr.msg_iovlen = 1;
    }

    // SAFETY: sendmmsg(2) 系统调用。msgs 数组长度 ≥ count，每个 msg_hdr.msg_iov
    // 指向一个有效的 iovec，iovec 指向用户提供的只读缓冲区。
    // dest_ptr 指向有效的 sockaddr_storage，生命周期由调用者保证。
    let ret = unsafe {
        libc::sendmmsg(
            fd,
            msgs.as_mut_ptr(),
            count as u32,
            flags,
        )
    };

    if ret < 0 {
        Err(LinuxError::Syscall {
            syscall: "sendmmsg",
            errno: std::io::Error::last_os_error().raw_os_error().unwrap_or(0),
        })
    } else {
        Ok(ret as usize)
    }
}

/// TCP 双向中继：使用 splice(2) 实现内核态零拷贝双向转发。
///
/// 创建两条管道，分别中继 `client → upstream` 和 `upstream → client`。
/// 全程数据不进入用户态缓冲区，达到内核零拷贝极限性能。
///
/// 采用双线程模型（每方向一线程），避免单线程中一个方向阻塞导致另一方向饥饿。
/// 任一方向 EOF 即关闭对端 socket（shutdown(SHUT_WR)），触发另一方向也快速退出。
///
/// # 参数
/// * `client_fd` - 客户端 socket fd
/// * `upstream_fd` - 上游 socket fd
/// * `pipe_buf_size` - 管道缓冲区大小（建议 65536）
///
/// # 返回
/// `(c2u_bytes, u2c_bytes)`：分方向字节数
/// （c2u = client→upstream，u2c = upstream→client），便于调用方分账 rx/tx
///
/// # 错误
/// - `LinuxError::Syscall { "pipe2", .. }` - 管道创建失败
/// - `LinuxError::Syscall { "splice", .. }` - splice 失败（含对端关闭 EPIPE/ECONNRESET）
pub fn splice_bidirectional(
    client_fd: i32,
    upstream_fd: i32,
    pipe_buf_size: usize,
) -> Result<(usize, usize)> {
    // pipe_buf_size 为 0 时 splice 每次传输 0 字节，导致无限空转
    if pipe_buf_size == 0 {
        return Err(LinuxError::InsufficientResources(
            "pipe_buf_size 不能为 0（会导致无限 0 字节 splice 调用）".to_string(),
        ));
    }

    // 阻塞管道：让 splice 在管道满/空时自然阻塞，避免 EAGAIN 自旋
    let (c2u_read, c2u_write) = pipe2(0)?;
    let (u2c_read, u2c_write) = pipe2(0)?;
    // 全部 pipe fd 纳入 RAII 守卫：spawn 失败 / relay 异常 / join 失败等任何
    // 提前返回路径均闭环回收；spawn 成功的方向写端所有权移交 relay 线程
    // （relay 各退出路径统一 close），守卫 into_raw 摘除防止 double-close。
    let c2u_read_guard = FdGuard::new(c2u_read);
    let mut c2u_write_guard = Some(FdGuard::new(c2u_write));
    let u2c_read_guard = FdGuard::new(u2c_read);
    let mut u2c_write_guard = Some(FdGuard::new(u2c_write));

    // 单向 splice 中继：src → pipe(w) → pipe(r) → dst，直到任一端 EOF
    // pipe_w 所有权归本闭包：所有退出路径（EOF/EPIPE/错误）都负责 close 一次
    let relay = |src_fd: i32, pipe_r: i32, pipe_w: i32, dst_fd: i32, p_size: usize| -> std::result::Result<usize, LinuxError> {
        let mut total = 0usize;
        loop {
            // src → pipe 写端
            let n = match splice(src_fd, None, pipe_w, None, p_size, libc::SPLICE_F_MOVE) {
                Ok(n) => n,
                Err(LinuxError::Syscall { syscall: _, errno }) if errno == libc::EPIPE => {
                    // dst 已关闭写端（对端中继已 EOF），停止本方向
                    // 关闭 pipe 写端防止 fd 泄漏（读端由主函数统一清理）
                    // SAFETY: close(2) 关闭写端（本路径唯一一次）
                    unsafe { libc::close(pipe_w); }
                    return Ok(total);
                }
                Err(e) => {
                    // 错误路径同样回收写端，杜绝 fd 泄漏
                    // SAFETY: close(2) 关闭写端（本路径唯一一次）
                    unsafe { libc::close(pipe_w); }
                    return Err(e);
                }
            };
            if n == 0 {
                // src EOF：关闭 pipe 写端，使 pipe 读端 splice 收到 0
                // SAFETY: close(2) 关闭写端，触发读端 EOF（本路径唯一一次）
                unsafe { libc::close(pipe_w); }
                // 排空管道剩余数据到 dst
                loop {
                    let m = splice(pipe_r, None, dst_fd, None, p_size, libc::SPLICE_F_MOVE)?;
                    if m == 0 {
                        break;
                    }
                    total = total.checked_add(m).ok_or_else(|| LinuxError::InsufficientResources(
                        "splice byte count overflow".to_string()
                    ))?;
                }
                // 通知 dst 没有更多数据
                // SAFETY: shutdown(2) 语义安全，EBADF/ENOTCONN 忽略
                unsafe { let _ = libc::shutdown(dst_fd, libc::SHUT_WR); }
                return Ok(total);
            }
            total = total.checked_add(n).ok_or_else(|| LinuxError::InsufficientResources(
                "splice byte count overflow".to_string()
            ))?;
            // pipe 读端 → dst
            let m = splice(pipe_r, None, dst_fd, None, n, libc::SPLICE_F_MOVE)?;
            // 管道内 splice 必须保持字节守恒
            debug_assert_eq!(m, n, "splice through pipe must preserve byte count");
        }
    };

    // 双线程并行：互不阻塞。spawn 成功后写端所有权移交对应 relay 线程
    let t1 = match std::thread::Builder::new()
        .name("zenith-splice-c2u".to_string())
        .spawn(move || relay(client_fd, c2u_read, c2u_write, upstream_fd, pipe_buf_size))
    {
        Ok(t) => {
            if let Some(g) = c2u_write_guard.take() {
                let _ = g.into_raw();
            }
            t
        }
        Err(e) => {
            // spawn 失败：无任何线程持有 fd，守卫随作用域闭环回收全部 4 个 fd
            return Err(LinuxError::InsufficientResources(format!(
                "spawn c2u thread failed: {e}"
            )));
        }
    };

    let t2 = match std::thread::Builder::new()
        .name("zenith-splice-u2c".to_string())
        .spawn(move || relay(upstream_fd, u2c_read, u2c_write, client_fd, pipe_buf_size))
    {
        Ok(t) => {
            if let Some(g) = u2c_write_guard.take() {
                let _ = g.into_raw();
            }
            t
        }
        Err(e) => {
            // t1 正在对已移交的 client_fd/upstream_fd 做 splice：先 shutdown
            // 两端 socket 让 t1 的 splice 返回 EOF/EPIPE 退出，join 回收线程
            // 后再返回 Err（c2u 写端由 t1 闭环，其余 fd 随守卫闭环）。
            // SAFETY: shutdown(2) 语义安全；fd 已关闭/未连接时的 EBADF/ENOTCONN 忽略。
            unsafe {
                let _ = libc::shutdown(client_fd, libc::SHUT_RDWR);
                let _ = libc::shutdown(upstream_fd, libc::SHUT_RDWR);
            }
            let _ = t1.join();
            return Err(LinuxError::InsufficientResources(format!(
                "spawn u2c thread failed: {e}"
            )));
        }
    };

    let r1 = t1.join().map_err(|_| LinuxError::InsufficientResources(
        "c2u relay thread panicked".to_string()
    ))??;
    let r2 = t2.join().map_err(|_| LinuxError::InsufficientResources(
        "u2c relay thread panicked".to_string()
    ))??;

    // pipe 写端由 relay 线程在各自退出路径闭环关闭；读端与（spawn 失败路径的）
    // 未移交写端随 FdGuard 在函数返回时回收——全程无泄漏分支
    let _ = (c2u_read_guard, u2c_read_guard);

    // r1 = c2u 字节（client→upstream），r2 = u2c 字节（upstream→client）分方向返回
    Ok((r1, r2))
}

/// 将系统 fd 包装为 RAII 守卫，确保 close(2) 被调用。
#[derive(Debug)]
pub struct FdGuard {
    fd: i32,
    closed: bool,
}

impl FdGuard {
    /// 创建 fd 守卫。`fd` 必须是有效的打开文件描述符。
    #[inline]
    pub fn new(fd: i32) -> Self {
        Self { fd, closed: false }
    }

    /// 获取原始 fd。
    #[inline]
    pub fn as_raw(&self) -> i32 {
        self.fd
    }

    /// 取出 fd 的所有权（不再由 Drop 关闭）。
    #[inline]
    pub fn into_raw(mut self) -> i32 {
        self.closed = true;
        self.fd
    }
}

impl Drop for FdGuard {
    fn drop(&mut self) {
        if !self.closed && self.fd >= 0 {
            // SAFETY: close(2) 释放 fd。忽略 EBADF（已被其他途径关闭）
            unsafe { let _ = libc::close(self.fd); }
        }
    }
}

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

    #[test]
    fn test_splice_flags_bits() {
        assert_eq!(SpliceFlags::None.bits(), 0);
        assert_eq!(SpliceFlags::Move.bits(), libc::SPLICE_F_MOVE);
        assert_eq!(SpliceFlags::NonBlock.bits(), libc::SPLICE_F_NONBLOCK);
        assert_eq!(SpliceFlags::More.bits(), libc::SPLICE_F_MORE);
    }

    #[test]
    fn test_pipe2_create() {
        // pipe2(O_NONBLOCK) 应创建两个非阻塞 fd
        let (r, w) = pipe2(libc::O_NONBLOCK).expect("pipe2");
        assert!(r >= 0);
        assert!(w >= 0);
        assert_ne!(r, w);
        // SAFETY: close(2) 清理 fd
        unsafe {
            libc::close(r);
            libc::close(w);
        }
    }

    #[test]
    fn test_pipe2_invalid_flags() {
        // 不应 panic，应返回错误
        let result = pipe2(0x7FFF_FFFF);
        // pipe2 对未知标志可能返回 EINVAL 或忽略，不强制断言
        let _ = result;
    }

    #[test]
    fn test_splice_eof_pipe() {
        // 写端关闭后，splice 应返回 0（EOF）
        let (r, w) = pipe2(0).expect("pipe2");
        // SAFETY: close 写端，触发读端 EOF
        unsafe { libc::close(w); }
        let n = splice(r, None, -1, None, 1024, 0);
        // fd_out = -1 应失败
        assert!(n.is_err());
        unsafe { libc::close(r); }
    }

    #[test]
    fn test_fd_guard_closes() {
        let (r, w) = pipe2(0).expect("pipe2");
        unsafe { libc::close(w); }
        {
            let _guard = FdGuard::new(r);
            // guard drop 时关闭 r
        }
        // 再次 close 应返回 EBADF（已关闭），但我们不暴露 close 结果
    }

    #[test]
    fn test_fd_guard_into_raw() {
        let (r, w) = pipe2(0).expect("pipe2");
        unsafe { libc::close(w); }
        let guard = FdGuard::new(r);
        let raw = guard.into_raw();
        assert_eq!(raw, r);
        // 需要手动 close
        unsafe { libc::close(r); }
    }

    #[test]
    fn test_recvmmsg_empty_bufs() {
        // 空缓冲区切片应返回 Ok(0)
        let mut bufs: [&mut [u8]; 0] = [];
        let result = recvmmsg(&mut bufs, -1, 0);
        assert_eq!(result.unwrap(), 0);
    }

    #[test]
    fn test_sendmmsg_empty_bufs() {
        let bufs: [&[u8]; 0] = [];
        // SAFETY: sockaddr_storage 全零是合法的零初始化（ss_family=AF_UNSPEC）
        let dest: libc::sockaddr_storage = unsafe { std::mem::zeroed() };
        let result = sendmmsg(&bufs, -1, &dest, 0);
        assert_eq!(result.unwrap(), 0);
    }

    #[test]
    fn test_sendfile_invalid_fd() {
        let result = sendfile(-1, -1, None, 1024);
        assert!(result.is_err());
        let err = result.unwrap_err();
        match err {
            LinuxError::Syscall { syscall, .. } => assert_eq!(syscall, "sendfile"),
            _ => panic!("expected Syscall error"),
        }
    }

    #[test]
    fn test_splice_invalid_fd() {
        let result = splice(-1, None, -1, None, 1024, 0);
        assert!(result.is_err());
        let err = result.unwrap_err();
        match err {
            LinuxError::Syscall { syscall, .. } => assert_eq!(syscall, "splice"),
            _ => panic!("expected Syscall error"),
        }
    }

    #[test]
    fn test_max_batch_datagrams_constant() {
        // 与 UIO_MAXIOV 对齐，glibc 上限
        assert_eq!(MAX_BATCH_DATAGRAMS, 64);
        const { assert!(MAX_BATCH_DATAGRAMS > 0) };
    }

    /// 枚举当前进程打开的 pipe fd 的 inode 标识（/proc/self/fd 符号链接目标）
    fn open_pipe_inodes() -> std::collections::HashSet<String> {
        let mut set = std::collections::HashSet::new();
        if let Ok(dir) = std::fs::read_dir("/proc/self/fd") {
            for entry in dir.flatten() {
                if let Ok(target) = std::fs::read_link(entry.path()) {
                    let s = target.to_string_lossy().into_owned();
                    if s.starts_with("pipe:[") {
                        set.insert(s);
                    }
                }
            }
        }
        set
    }

    /// 真实回环 TCP 的 splice 双向中继集成测试：
    /// echo-server ↔ splice_bidirectional(client, upstream) ↔ 半关闭客户端
    ///
    /// 验证（EOF/半关闭语义）：
    /// 1. 字节数守恒：c2u/u2c 分方向计数 == 负载长度，echo 内容逐字节一致
    /// 2. 半关闭传播：客户端 shutdown(Wr) → relay c2u EOF → upstream 半关闭 →
    ///    echo 收口 → relay u2c EOF → 客户端读到 EOF(0)
    /// 3. 无泄漏：relay 结束后全部 pipe fd 关闭（/proc/self/fd 对比）
    #[test]
    fn test_splice_bidirectional_tcp_loopback() {
        use std::io::{Read, Write};
        use std::net::{Shutdown, TcpListener, TcpStream};
        use std::os::unix::io::AsRawFd;

        let pipe_baseline = open_pipe_inodes();

        // echo 服务端：读到的数据原样写回；客户端半关闭（EOF）后自身半关闭写侧
        let echo_listener = match TcpListener::bind("127.0.0.1:0") {
            Ok(l) => l,
            Err(_) => return, // 极端环境无回环网络栈，跳过
        };
        let echo_addr = echo_listener.local_addr().unwrap();
        let echo_thread = std::thread::spawn(move || -> usize {
            let (mut s, _) = echo_listener.accept().unwrap();
            let mut buf = [0u8; 4096];
            let mut echoed = 0usize;
            loop {
                match s.read(&mut buf) {
                    Ok(0) | Err(_) => break, // EOF：上游半关闭已传播
                    Ok(n) => {
                        s.write_all(&buf[..n]).unwrap();
                        echoed += n;
                    }
                }
            }
            let _ = s.shutdown(Shutdown::Write); // echo 侧半关闭，触发 relay U2C EOF
            echoed
        });

        // relay 两端：upstream 侧主动连 echo；client 侧接受测试驱动连接
        let proxy_listener = TcpListener::bind("127.0.0.1:0").unwrap();
        let proxy_addr = proxy_listener.local_addr().unwrap();
        let upstream = TcpStream::connect(echo_addr).unwrap();
        let mut client = TcpStream::connect(proxy_addr).unwrap();
        let (proxy_side, _) = proxy_listener.accept().unwrap();

        let proxy_fd = proxy_side.as_raw_fd();
        let upstream_fd = upstream.as_raw_fd();
        let relay_thread = std::thread::spawn(move || {
            splice_bidirectional(proxy_fd, upstream_fd, 16384)
        });

        // 测试负载：64KB（pipe_buf 16384 → 多轮 splice），随后半关闭写侧
        let payload: Vec<u8> = (0..65536u32).map(|i| (i % 251) as u8).collect();
        client.write_all(&payload).unwrap();
        client.shutdown(Shutdown::Write).unwrap();

        // 读回全部 echo，直至 EOF（半关闭传播链的最终证据）
        let mut got = Vec::with_capacity(payload.len());
        let mut tmp = [0u8; 8192];
        loop {
            match client.read(&mut tmp) {
                Ok(0) => break, // EOF：relay U2C 完成并传播了 client 侧半关闭
                Ok(n) => got.extend_from_slice(&tmp[..n]),
                Err(e) => panic!("client read failed: {e}"),
            }
        }
        assert_eq!(got.len(), payload.len(), "echo 字节数必须守恒");
        assert!(got == payload, "echo 内容必须逐字节一致");

        // echo 侧读到的字节数 == 负载长度
        let echoed = echo_thread.join().unwrap();
        assert_eq!(echoed, payload.len(), "echo 服务端必须读满全部负载");

        // relay 双向 EOF 后自然结束：分方向字节数核对
        let (c2u, u2c) = relay_thread.join().unwrap().unwrap();
        assert_eq!(c2u, payload.len(), "c2u 方向字节数必须守恒");
        assert_eq!(u2c, payload.len(), "u2c 方向字节数必须守恒");

        // 显式断开测试侧 socket，释放其 fd 后再做泄漏判定
        drop(client);
        drop(proxy_side);
        drop(upstream);
        drop(proxy_listener);

        // 无泄漏判定：relay 的 4 个 pipe fd 必须全部关闭。
        // 并发测试进程中其他用例的瞬时 pipe 会造成抖动：短暂重试等待其消失；
        // 泄漏的 pipe fd 不会自行消失，重试窗口后仍存在即判失败。
        let mut extra = Vec::new();
        for _ in 0..100 {
            extra = open_pipe_inodes()
                .difference(&pipe_baseline)
                .cloned()
                .collect::<Vec<_>>();
            if extra.is_empty() {
                break;
            }
            std::thread::sleep(std::time::Duration::from_millis(10));
        }
        assert!(extra.is_empty(), "splice_bidirectional 泄漏 pipe fd: {extra:?}");
    }
}