Tokio与Pcap性能优化：如何提升高吞吐量数据包处理能力？

我开发了一个需要处理大量数据包的程序，目标是达到约350k pkts/s的处理能力。经过多轮代码迭代后，得到了下方的实现。程序运行在配备4个专属核心、NIC带宽远超需求的Ubuntu虚拟机上。

运行当前代码时，基本仅占用一个核心，且大部分为IO相关的内核调用。Pcap仅报告程序缓冲区的丢包，而非网卡本身的丢包。在仅87k pkts/s的吞吐量下就出现了丢包，需大幅优化以实现4-5倍的负载提升。

Cargo.toml #

[dependencies]
pcap = { version = "1.1.0", features = ["all-features", "capture-stream"] }
tokio = { version = "1.32.0", features = ["full"] }
futures = { version = "0.3.28"}

主程序代码 #

use pcap::{Active, Capture, Inactive, Error, Packet, PacketCodec, PacketStream};
use tokio::sync::mpsc;
use futures::StreamExt;

// Simple codec that returns owned copies, since the result may not
// reference the input packet.
pub struct BoxCodec;

impl PacketCodec for BoxCodec {
    type Item = Box<[u8]>;

    fn decode(&mut self, packet: Packet) -> Self::Item {
        packet.data.into()
    }
}

fn new_stream(capture_inactive: Capture<Inactive>) -> Result<PacketStream<Active, BoxCodec>, Error> {
    let cap = capture_inactive
        .promisc(true)
        .immediate_mode(true)
        .open()?
        .setnonblock()?;
    cap.stream(BoxCodec)
}

// generate a dummy layer 2 packet that we can easily find in wireshark
async fn generate_packet() -> Vec<u8> {
    // Define source and destination MAC addresses
    let src_mac = [0x00, 0x11, 0x22, 0x33, 0x44, 0x55];
    let dest_mac = [0xAA, 0xBB, 0xCC, 0xDD, 0xEE, 0xFF];

    // Create the Ethernet frame
    let mut pkt: Vec<u8> = Vec::new();

    // Destination MAC address
    pkt.extend_from_slice(&dest_mac);

    // Source MAC address
    pkt.extend_from_slice(&src_mac);

    // EtherType (0x0800 for IPv4)
    pkt.extend_from_slice(&[0x08, 0x00]);

    // Custom payload
    let payload: [u8; 10] = [0x01, 0x02, 0x03, 0x04, 0x05, 0x06, 0x07, 0x08, 0x09, 0x0A];
    pkt.extend_from_slice(&payload);
    
    pkt
}

#[tokio::main]
async fn main() {
    let capture_inactive = Capture::from_device("ens192").unwrap();

    let (tx, mut rx): (mpsc::Sender<Vec<u8>>, mpsc::Receiver<Vec<u8>>) = mpsc::channel(1024);
    let (tx_msg, mut rx_msg): (mpsc::Sender<Vec<u8>>, mpsc::Receiver<Vec<u8>>) = mpsc::channel(1024);

    // spawn the process for reading packets from the interface...
    tokio::spawn(async move {
        let mut stream = new_stream(capture_inactive).expect("Failed to create stream");
        let mut count = 0;
        loop {
            tokio::select! {
                packet = stream.next() => { // packet is Option<Result<Box>>
                    count += 1;
                    if count % 1_000_000 == 0 {
                        let cap = stream.capture_mut();
                        let stats = cap.stats().unwrap();
                        println!(
                            "Received: {}, dropped: {}, if_dropped: {}",
                            stats.received, stats.dropped, stats.if_dropped
                        );
                    }

                    if let Some(Ok(data)) = packet {
                        let _send_result = tx.send(data.to_vec()).await;
                    }
                },
                data = rx_msg.recv() => {
                    let _ = stream.capture_mut().sendpacket(data.unwrap());
                }
            }
        }
    });

    let worker_handle = tokio::spawn(async move {
        let mut count = 0;
        loop {
            match rx.recv().await {
                Some(_packet) => {
                    count += 1;
                    if count % 100_000 == 0 {
                        println!("Processed {} packets", count);
                    }
                    if count % 100_000 == 0 {
                        let data = generate_packet().await;
                        let _msg = tx_msg.send(data).await;
                    }
                }
                None => {
                    println!("Worker task ended");
                    break;
                }
            }
        }
    });

    worker_handle.await.unwrap();

}

我该从哪些方面入手优化性能？我曾了解到Tokio默认会使用所有可用核心，也尝试过启用多线程模式（#[tokio::main(flavor = "multi_thread", worker_threads = 8)]），但调整工作线程数并未带来性能提升。我希望让主线程专注于数据包接收，但当前并未达到预期效果，求优化思路。

一、降低数据包接收路径开销 #

• 减少内存拷贝：当前BoxCodec将packet.data转为Box<[u8]>后，又调用data.to_vec()发送到通道，存在两次冗余拷贝。可改用Bytes类型替代Vec<u8>，利用其共享特性实现零拷贝传递；或者直接传递Packet的引用（需注意生命周期管理）。
• 调大Pcap缓冲区：默认缓冲区过小易导致用户态溢出丢包，通过capture_inactive.buffer_size(64 * 1024 * 1024)设置64MB级别的缓冲区，降低内核到用户态的拷贝频率。
• 剥离非核心逻辑：接收循环中每百万包调用一次cap.stats()会触发系统调用，增加延迟。可将统计逻辑拆分到单独定时任务中，避免干扰数据包接收流程。

二、优化任务调度与核心绑定 #

• 核心隔离绑定：用core_affinity crate将数据包接收任务绑定到专属核心，避免调度切换；将处理任务绑定到剩余3个核心，实现接收与处理的物理核心隔离。
• 拆分收发逻辑：当前接收任务通过tokio::select!混合处理数据包接收与发送，易导致接收阻塞。将发送逻辑拆分到独立Tokio任务，让接收循环仅专注于读包。

三、优化通道与任务模型 #

• 增大通道容量：当前1024的通道容量在高吞吐量下易导致发送阻塞，可将容量提升至16384或使用无界通道（需注意内存占用），避免因通道满导致接收停滞。
• 批量处理数据包：不要单包发送到通道，积累128个或更多数据包后批量传递，减少通道收发开销与任务调度次数。
• 多Worker并行处理：仅一个Worker任务无法利用多核心，启动3个Worker任务对应剩余核心，通过mpsc::broadcast或信号量分配任务，并行处理提升吞吐量。

四、替换高性能捕获库 #

Pcap依赖内核libpcap，高吞吐量下开销较大。可替换为：

• xdp-rs：基于AF_XDP实现零拷贝捕获，大幅降低IO开销
• rte_rs：DPDK的Rust绑定，适合超高性能数据包处理场景
• tokio-uring结合原始套接字：减少内核态到用户态的拷贝

五、系统层面优化 #

• NIC参数调整：关闭NIC不必要的校验和卸载，将NIC中断绑定到接收任务所在核心，减少中断处理的上下文切换。
• 内核参数调优：修改/proc/sys/net/core/rmem_max、/proc/sys/net/core/wmem_max增大套接字缓冲区；调整/proc/sys/net/core/netdev_max_backlog增大网卡接收队列，避免内核层面丢包。

内容的提问来源于stack exchange，提问作者PilotGuy