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.


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...


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 {

fn new_stream(capture_inactive: Capture<Inactive>) -> Result<PacketStream<Active, BoxCodec>, Error> {
    let cap = capture_inactive

// 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

    // Source MAC address

    // 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];

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



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

htop and program output


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.


  • \$\begingroup\$ Welcome to Code Review! Please edit your question so that the title describes the purpose of the code, rather than its mechanism. We really need to understand the motivational context to give good reviews. Please see How to get the best value out of Code Review: Asking Questions for guidance on writing good question titles. \$\endgroup\$ Aug 25 at 18:41
  • \$\begingroup\$ I really thought it was pretty clear. I'm trying to maximize the throughput of i/o so that I could process a target packet rate of 350k packets/s. I've hit a limit, and I'm looking for ways to improve it. \$\endgroup\$
    – PilotGuy
    Aug 25 at 19:05
  • \$\begingroup\$ Dropping 26 packets out of 4 million is really not so bad. Is your code slow? No, the throughput is fine, it's a good match for the offered load. Do you have adequate buffering? No, not if you desire fewer drops. Sometimes a packet burst arrives and/or your system is busy with other things, maybe bringing in a hard page fault from backing store. Think about the 95-th or 99-th percentile latencies. You need more buffering to ride them out. Using single core for single file descriptor is a function of the kernel interface, right? \$\endgroup\$
    – J_H
    Aug 25 at 19:24
  • \$\begingroup\$ Are you sure you want immediate mode? "In immediate mode, packets are always delivered as soon as they arrive, with no buffering." Also, it's unclear how nonblocking is helping you. Might as well patiently block for a millisecond until next packet arrives, no? \$\endgroup\$
    – J_H
    Aug 25 at 19:36
  • \$\begingroup\$ I might be misunderstanding the output of pcap stats, then. I was under the impression that if the if_drops was 0 and the stats.dropped was anything, that the failure of stats.dropped was pointing to an inefficiency in my code. I have increased the buffering to 5_000_000 from the default 1_000_000, and I increased the snap_length to 250_000. And that did help, but I'm also not completely sure why I would have needed to increase the snap_length when all of the packets on my network have a max length of 800 bytes -- with the majority at around 200 bytes max. \$\endgroup\$
    – PilotGuy
    Aug 25 at 19:38


Your Answer

By clicking “Post Your Answer”, you agree to our terms of service and acknowledge that you have read and understand our privacy policy and code of conduct.

Browse other questions tagged or ask your own question.