Optimizing tokio and pcap Operations in Rust for Max Throughput

Question

Background

I'm mostly a c# and python programmer and am slowly learning Rust. I'm not sure if that is contributing to incorrect concepts as to how threading and io work in Rust, but that's what I'm trying to figure out here.

I've got a program that needs to handle a lot of packets. Ultimately, I'm looking to handle around 350k pkts/s. To tackle this, I've iterated through several versions of code and ended up with the code below.

The code layout is to ultimately allow me to do calculations on the incoming network traffic and periodically send out messages. Some of those calculations are long-running, thus my need to offload the packets as quickly as possible to worker threads. Unfortunately, even this code (without my calculations) is dropping packets.

For reference, the code is running on an ubuntu VM with 4 dedicated cores and a NIC way bigger than is needed for this application.

Observations

Running the below code only loads one core, for the most part. It's mostly kernel calls, which I'm guessing is for IO.

Pcap (if accurate) is only reporting dropped packets for the program buffer -- not the interface itself.

The dropped packets are being observed with only 87k pkts/s. I've got to significantly improve this if I want to get 4-5X data loads through it.

See the question below the code...

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"}

Main.rs

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();

}

Here is a screenshot showing the program output, and the htop stats.

Question

What can I look at to improve performance? Although I can't find it at the moment, I thought I read that tokio defaults to using all of the available cores. For grins, I did run a version with the multi-thread options on tokio. Playing with this and changing the number of available workers did not seem to impact performance in any positive way.

#[tokio::main(flavor = "multi_thread", worker_threads = 8)]

I'm trying to offload the main thread to allow it to do nothing but ingest packets, but that doesn't appear to be what's happening here.

Ideas?