amberjs 1.17.0

// 零拷贝接收器 - 高性能网络数据接收
//
// Stage 39.0: 网络零拷贝优化
//
// 该模块提供高性能的零拷贝数据接收功能，通过 splice 系统调用实现
// 网络套接字到文件的零拷贝接收，最小化数据拷贝开销。

use std::collections::{BTreeMap, HashMap};
use std::io::{Read, Seek, SeekFrom, self};
use std::sync::{Arc, Mutex};
use std::fs::File;
use std::time::{Duration, Instant};

/// 零拷贝接收配置
#[derive(Debug, Clone)]
pub struct ZeroCopyReceiverConfig {
    /// 接收缓冲区大小
    pub buffer_size: usize,
    /// 最大接收大小
    pub max_receive_size: usize,
    /// 接收超时时间
    pub receive_timeout: Duration,
    /// 启用预读
    pub enable_read_ahead: bool,
    /// 预读大小
    pub read_ahead_size: usize,
}
impl Default for ZeroCopyReceiverConfig {
    fn default() -> Self {
        Self {
            buffer_size: 64 * 1024,
            max_receive_size: 1024 * 1024 * 1024, // 1GB
            receive_timeout: Duration::from_secs(30),
            enable_read_ahead: true,
            read_ahead_size: 128 * 1024,
        }
    }
}
/// 零拷贝接收统计信息
#[derive(Debug, Clone, Default)]
pub struct ZeroCopyReceiverStats {
    /// 总接收字节数
    pub total_bytes_received: u64,
    /// 接收成功次数
    pub success_count: u64,
    /// 接收失败次数
    pub error_count: u64,
    /// 平均接收速度 (bytes/sec)
    pub avg_receive_speed: f64,
    /// 峰值接收速度 (bytes/sec)
    pub peak_receive_speed: f64,
    /// splice 调用次数
    pub splice_count: u64,
    /// 总接收时间
    pub total_receive_time: Duration,
    /// 最后接收时间
    pub last_receive: Option<Instant>,
}
/// 零拷贝接收器
///
/// 该结构体提供高性能的零拷贝数据接收功能：
/// - 使用 splice 系统调用实现零拷贝接收
/// - 支持大文件接收
/// - 实时性能统计
/// - 错误处理和重试机制
#[derive(Debug)]
pub struct ZeroCopyReceiver {
    /// 接收配置
    config: ZeroCopyReceiverConfig,
    /// 接收统计
    stats: Arc<Mutex<ZeroCopyReceiverStats>>,
    /// splice 传输器
    splice: Splice,
    /// 当前接收位置
    current_pos: u64,
}
impl ZeroCopyReceiver {
    /// 创建新的零拷贝接收器
    ///
    /// # 参数
    /// - `config`: 接收配置，如果为 None 则使用默认配置
    ///
    /// # 返回值
    /// 返回创建结果
    pub fn new(config: Option<ZeroCopyReceiverConfig>) -> io::Result<Self> {
        let config: _ = config.unwrap_or_default();
        let splice: _ = Splice::new();
        Ok(Self {
            config,
            stats: Arc::new(Mutex::new(ZeroCopyReceiverStats::default())),
            splice,
            current_pos: 0,
        })
    }
    /// 从网络套接字零拷贝接收数据到文件
    ///
    /// 使用 splice 系统调用实现网络套接字到文件的零拷贝接收
    ///
    /// # 参数
    /// - `socket`: 网络套接字（实现 AsRawFd）
    /// - `file`: 目标文件
    /// - `max_bytes`: 最大接收字节数
    ///
    /// # 返回值
    /// 返回接收的字节数
    pub fn receive_from_socket<S: AsRawFd>(
        &mut self,
        socket: &S,
        file: &mut File,
        max_bytes: usize,
    ) -> io::Result<u64> {
        let start_time: _ = Instant::now();
        let socket_fd: _ = socket.as_raw_fd();
        // 设置文件位置
        file.seek(SeekFrom::Start(self.current_pos))?;
        // 使用 splice 进行零拷贝接收
        match self.splice.fd_to_pipe(socket_fd, &std::os::unix::net::UnixStream::pair()?.1, max_bytes) {
            Ok(bytes_received) => {
                self.current_pos += bytes_received;
                // 更新统计信息
                self.update_stats_on_success(bytes_received, &start_time);
                println!("✅ 零拷贝接收成功: {} bytes", bytes_received);
                Ok(bytes_received)
            }
            Err(e) => {
                self.update_stats_on_error(&start_time);
                println!("❌ 零拷贝接收失败: {}", e);
                Err(e)
            }
        }
    }
    /// 从网络套接字零拷贝接收数据到内存缓冲区
    ///
    /// # 参数
    /// - `socket`: 网络套接字（实现 AsRawFd）
    /// - `buffer`: 目标缓冲区
    /// - `max_bytes`: 最大接收字节数
    ///
    /// # 返回值
    /// 返回接收的字节数
    pub fn receive_to_buffer<S: AsRawFd>(
        &mut self,
        socket: &S,
        buffer: &mut Vec<u8>,
        max_bytes: usize,
    ) -> io::Result<u64> {
        let start_time: _ = Instant::now();
        let socket_fd: _ = socket.as_raw_fd();
        // 确保缓冲区有足够的空间
        if buffer.len() < max_bytes {
            buffer.resize(max_bytes, 0);
        }
        // 使用 splice 将数据接收到临时文件，然后读取到缓冲区
        let mut temp_file = tempfile::tempfile()?;
        let pipe: _ = std::os::unix::net::UnixStream::pair()?.1;
        match self.splice.fd_to_pipe(socket_fd, &pipe, max_bytes) {
            Ok(bytes_received) => {
                // 从临时文件读取到缓冲区
                temp_file.seek(SeekFrom::Start(0))?;
                let bytes_to_read: _ = bytes_received as usize;
                temp_file.read_exact(&mut buffer[0..bytes_to_read])?;
                self.current_pos += bytes_received;
                // 更新统计信息
                self.update_stats_on_success(bytes_received, &start_time);
                println!("✅ 零拷贝接收数据到缓冲区成功: {} bytes", bytes_received);
                Ok(bytes_received)
            }
            Err(e) => {
                self.update_stats_on_error(&start_time);
                println!("❌ 零拷贝接收数据到缓冲区失败: {}", e);
                Err(e)
            }
        }
    }
    /// 设置接收位置
    ///
    /// # 参数
    /// - `pos`: 新的接收位置
    pub fn set_position(&mut self, pos: u64) {
        self.current_pos = pos;
    }
    /// 获取当前接收位置
    ///
    /// # 返回值
    /// 返回当前接收位置
    pub fn position(&self) -> u64 {
        self.current_pos
    }
    /// 重置接收状态
    pub fn reset(&mut self) {
        self.current_pos = 0;
        // 重置统计信息
        let mut stats = self.stats.lock().unwrap();
        *stats = ZeroCopyReceiverStats::default();
    }
    /// 获取接收统计信息
    ///
    /// # 返回值
    /// 返回统计信息副本
    pub fn get_stats(&self) -> ZeroCopyReceiverStats {
        self.stats.lock().unwrap().clone()
    }
    /// 计算接收进度百分比
    ///
    /// # 参数
    /// - `total_size`: 总大小
    ///
    /// # 返回值
    /// 返回接收进度 (0.0-100.0)
    pub fn progress(&self, total_size: u64) -> f64 {
        if total_size > 0 {
            (self.current_pos as f64 / total_size as f64) * 100.0
        } else {
            0.0
        }
    }
    /// 获取接收速度 (bytes/sec)
    ///
    /// # 返回值
    /// 返回当前接收速度
    pub fn speed(&self) -> f64 {
        let stats: _ = self.stats.lock().unwrap();
        stats.avg_receive_speed
    }
    /// 检查是否可以继续接收
    ///
    /// # 参数
    /// - `max_size`: 最大大小限制
    ///
    /// # 返回值
    /// 返回是否可以继续接收
    pub fn can_continue(&self, max_size: u64) -> bool {
        self.current_pos < max_size && self.current_pos < self.config.max_receive_size as u64
    }
    /// 更新成功接收的统计信息
    fn update_stats_on_success(&self, bytes: u64, start_time: &Instant) {
        let mut stats = self.stats.lock().unwrap();
        stats.total_bytes_received += bytes;
        stats.success_count += 1;
        stats.last_receive = Some(Instant::now());
        let elapsed: _ = start_time.elapsed();
        stats.total_receive_time += elapsed;
        if elapsed.as_secs_f64() > 0.0 {
            let current_speed: _ = bytes as f64 / elapsed.as_secs_f64();
            stats.avg_receive_speed = (stats.avg_receive_speed * (stats.success_count - 1) as f64
                + current_speed)
                / stats.success_count as f64;
            if current_speed > stats.peak_receive_speed {
                stats.peak_receive_speed = current_speed;
            }
        }
        stats.splice_count += 1;
    }
    /// 更新失败接收的统计信息
    fn update_stats_on_error(&self, start_time: &Instant) {
        let mut stats = self.stats.lock().unwrap();
        stats.error_count += 1;
        stats.total_receive_time += start_time.elapsed();
    }
}
impl Default for ZeroCopyReceiver {
    fn default() -> Self {
        Self::new(None).expect("Failed to create default ZeroCopyReceiver")
    }
}
#[cfg(test)]
mod tests {
    /// 测试创建零拷贝接收器
    #[test]
    fn test_zero_copy_receiver_creation() {
        let config: _ = ZeroCopyReceiverConfig::default();
        let receiver: _ = ZeroCopyReceiver::new(Some(config)).expect("创建接收器失败");
        let stats: _ = receiver.get_stats();
        assert_eq!(stats.total_bytes_received, 0);
        assert_eq!(stats.success_count, 0);
        assert_eq!(stats.error_count, 0);
        println!("✅ 测试通过: 零拷贝接收器创建");
    }
    /// 测试接收位置设置
    #[test]
    fn test_position_setting() {
        let mut receiver = ZeroCopyReceiver::new(None).expect("创建接收器失败");
        receiver.set_position(2048);
        assert_eq!(receiver.position(), 2048);
        println!("✅ 测试通过: 接收位置设置");
    }
    /// 测试进度计算
    #[test]
    fn test_progress_calculation() {
        let mut receiver = ZeroCopyReceiver::new(None).expect("创建接收器失败");
        // 设置位置为总大小的 50%
        receiver.set_position(512);
        let progress: _ = receiver.progress(1024);
        assert_eq!(progress, 50.0);
        println!("进度: {:.1}%", progress);
        println!("✅ 测试通过: 进度计算");
    }
    /// 测试接收统计
    #[test]
    fn test_receive_stats() {
        let receiver: _ = ZeroCopyReceiver::new(None).expect("创建接收器失败");
        // 模拟成功接收
        let start_time: _ = Instant::now();
        receiver.update_stats_on_success(2048, &start_time);
        let stats: _ = receiver.get_stats();
        assert_eq!(stats.total_bytes_received, 2048);
        assert_eq!(stats.success_count, 1);
        assert_eq!(stats.error_count, 0);
        println!("✅ 测试通过: 接收统计");
    }
}