ferranet 0.2.0

//! `pibench` — a capture/flood harness for comparing the **CPU cost** of ferranet against libpnet
//! on low-power hardware (e.g. a Raspberry Pi Zero 2 W with a USB Ethernet adapter).
//!
//! Two modes:
//!
//! * `capture` — the device under test: receive frames for a fixed window on one backend
//!   (`ring`, `basic`, or `pnet`) and report received pps, throughput, kernel drops, and the CPU
//!   load it cost — **process CPU as a percentage of one core**, CPU microseconds per frame, and
//!   the system-wide CPU used over the window (`getrusage` + `/proc/stat`).
//! * `flood` — saturate an interface via ferranet's batched TX, optionally rate-capped with
//!   `--mbps` so you can compare CPU load at a *fixed* offered rate.
//!
//! Pair them over a `veth` pair on the Pi (controlled, repeatable) or point `capture` at the USB
//! Ethernet interface and flood it from another host. See `docs/pi-bench.md`.
//!
//! ```sh
//! pibench capture --backend ring  --iface eth1 --secs 10 [--snaplen 128]
//! pibench flood   --iface pi0 --secs 13 --size 64 [--mbps 100]
//! ```

use std::env;
use std::mem;
use std::time::{Duration, Instant};

use etherparse::{NetSlice, SlicedPacket, TransportSlice};
use ferranet::{Channel, MacAddr, RingConfig};
use pnet_datalink::{Channel as PnetChannel, Config as PnetConfig};

/// One measurement window.
struct Sample {
    frames: u64,
    bytes: u64,
    drops: Option<u64>,
    wall: f64,
    /// Process user+system CPU seconds (this capture loop's own cost).
    cpu_s: f64,
    /// System-wide CPU-seconds used over the window (all cores; includes softirq/driver).
    sys_cpu_s: f64,
    /// Per-frame parse/checksum/classification tally, when `--verify` is set.
    verify: Option<Verify>,
}

/// Correctness check + protocol classification for captured frames, via etherparse.
///
/// Only populated under `--verify`. Parsing every frame adds CPU, so a verify run's `proc_cpu`
/// is NOT the clean performance number — use it to confirm integrity, not to compare backends.
#[derive(Default)]
struct Verify {
    parsed: u64,        // frames etherparse sliced without error
    parse_err: u64,     // frames etherparse rejected (corrupt / mis-bounded framing)
    ipv4_bad_csum: u64, // IPv4 frames whose header checksum did not recompute
    udp: u64,
    tcp: u64,
    icmp: u64,
    arp: u64,
    other: u64,
}

impl Verify {
    /// Slice one Ethernet frame, validate its IPv4 header checksum, and bump the class histogram.
    fn observe(&mut self, frame: &[u8]) {
        let pkt = match SlicedPacket::from_ethernet(frame) {
            Ok(p) => p,
            Err(_) => {
                self.parse_err += 1;
                return;
            }
        };
        self.parsed += 1;
        if let Some(NetSlice::Ipv4(ip)) = &pkt.net {
            let h = ip.header();
            if h.to_header().calc_header_checksum() != h.header_checksum() {
                self.ipv4_bad_csum += 1;
            }
        }
        match pkt.transport {
            Some(TransportSlice::Udp(_)) => self.udp += 1,
            Some(TransportSlice::Tcp(_)) => self.tcp += 1,
            Some(TransportSlice::Icmpv4(_) | TransportSlice::Icmpv6(_)) => self.icmp += 1,
            // No L4 we track: ARP has no IP/transport, so disambiguate it by ethertype (0x0806).
            _ if frame.len() >= 14 && u16::from_be_bytes([frame[12], frame[13]]) == 0x0806 => {
                self.arp += 1
            }
            _ => self.other += 1,
        }
    }
}

fn main() {
    let args: Vec<String> = env::args().collect();
    match args.get(1).map(String::as_str) {
        Some("capture") => capture(&args),
        Some("flood") => flood(&args),
        _ => {
            eprintln!(
                "usage:\n  pibench capture --backend <ring|basic|pnet|pcap> --iface <if> [--secs N] [--warmup S] [--snaplen N] [--verify]\n    (pcap backend requires building with --features pcap-bench)\n  pibench flood --iface <if> [--secs N] [--size N] [--mbps N]"
            );
            std::process::exit(2);
        }
    }
}

fn capture(args: &[String]) {
    let backend = flag(args, "--backend").unwrap_or_else(|| "ring".into());
    let iface = flag(args, "--iface").unwrap_or_else(|| die("--iface is required"));
    let secs: f64 = flag(args, "--secs").and_then(|s| s.parse().ok()).unwrap_or(10.0);
    let snaplen: Option<u32> = flag(args, "--snaplen").and_then(|s| s.parse().ok());
    // Drain (and discard) this many seconds before the measurement window opens, so the
    // ring's first-touch page faults, cold caches, and any backlog that built up between
    // bind and the first read don't show up as startup drops / inflated CPU. 0 disables it.
    let warmup: f64 = flag(args, "--warmup").and_then(|s| s.parse().ok()).unwrap_or(0.5);
    // Parse & classify every captured frame with etherparse, validating IPv4 header checksums.
    // Adds per-frame CPU, so it's a correctness pass, not a perf measurement. Best without --snaplen.
    let verify = args.iter().any(|a| a == "--verify");

    let sample = match backend.as_str() {
        "ring" | "basic" => capture_ferranet(&iface, &backend, secs, warmup, verify, snaplen),
        "pnet" => capture_pnet(&iface, secs, warmup, verify),
        #[cfg(feature = "pcap-bench")]
        "pcap" => capture_pcap(&iface, secs, warmup, verify),
        #[cfg(not(feature = "pcap-bench"))]
        "pcap" => die("pcap backend not compiled in — rebuild with --features pcap-bench"),
        other => die(&format!("unknown backend {other:?} (ring|basic|pnet|pcap)")),
    };

    let note = match (backend.as_str(), snaplen) {
        ("ring", Some(s)) => format!("snaplen={s}"),
        _ => String::new(),
    };
    report(&backend, &iface, &note, &sample);
}

fn capture_ferranet(iface: &str, backend: &str, secs: f64, warmup: f64, verify: bool, snaplen: Option<u32>) -> Sample {
    let mut builder = Channel::builder(iface).promiscuous(true);
    if backend == "basic" {
        builder = builder.basic();
    } else if let Some(s) = snaplen {
        builder = builder.rx_ring(RingConfig::small().snaplen(s));
    }
    let (_tx, mut rx) = builder.build_sync().unwrap_or_else(|e| die(&format!("open: {e}")));

    // Warm up by draining for `warmup` seconds: lets first-touch page faults and cold caches settle.
    let warm_until = Instant::now() + Duration::from_secs_f64(warmup);
    while Instant::now() < warm_until {
        if rx.recv_block().is_err() {
            break;
        }
    }

    let mut ver = verify.then(Verify::default);
    let start = Instant::now();
    let (cpu0, sys0) = (cpu_seconds(), system_cpu_seconds());
    // Reset the kernel drop counter as the LAST thing before draining: stats() reads
    // PACKET_STATISTICS, which the kernel resets on read. Doing it after the cpu_seconds() /
    // system_cpu_seconds() syscalls (the /proc/stat read is slow on an A53) means no un-drained
    // gap sits between the reset and the first recv_block — that gap was letting the ring fill and
    // show up as phantom "startup" drops.
    let _ = rx.stats();
    let deadline = start + Duration::from_secs_f64(secs);
    let (mut frames, mut bytes) = (0u64, 0u64);
    while Instant::now() < deadline {
        match rx.recv_block() {
            Ok(block) => {
                for f in block.frames() {
                    frames += 1;
                    bytes += f.wire_len() as u64;
                    if let Some(v) = ver.as_mut() {
                        v.observe(f.data());
                    }
                }
            }
            Err(_) => break,
        }
    }
    let wall = start.elapsed().as_secs_f64();
    Sample {
        frames,
        bytes,
        drops: rx.stats().ok().map(|s| s.dropped),
        wall,
        cpu_s: cpu_seconds() - cpu0,
        sys_cpu_s: system_cpu_seconds() - sys0,
        verify: ver,
    }
}

fn capture_pnet(iface: &str, secs: f64, warmup: f64, verify: bool) -> Sample {
    let interface = pnet_datalink::interfaces()
        .into_iter()
        .find(|i| i.name == iface)
        .unwrap_or_else(|| die(&format!("libpnet: interface {iface} not found")));
    let config = PnetConfig {
        read_timeout: Some(Duration::from_millis(200)),
        promiscuous: true,
        ..PnetConfig::default()
    };
    let mut rx = match pnet_datalink::channel(&interface, config) {
        Ok(PnetChannel::Ethernet(_tx, rx)) => rx,
        Ok(_) => die("libpnet: unexpected channel type"),
        Err(e) => die(&format!("libpnet open: {e}")),
    };

    // Warm up (matches the ferranet path) so the steady-state CPU comparison is apples-to-apples.
    let warm_until = Instant::now() + Duration::from_secs_f64(warmup);
    while Instant::now() < warm_until {
        let _ = rx.next();
    }

    let mut ver = verify.then(Verify::default);
    let start = Instant::now();
    let (cpu0, sys0) = (cpu_seconds(), system_cpu_seconds());
    let deadline = start + Duration::from_secs_f64(secs);
    let (mut frames, mut bytes) = (0u64, 0u64);
    while Instant::now() < deadline {
        if let Ok(pkt) = rx.next() {
            frames += 1;
            bytes += pkt.len() as u64;
            if let Some(v) = ver.as_mut() {
                v.observe(pkt);
            }
        }
    }
    let wall = start.elapsed().as_secs_f64();
    Sample {
        frames,
        bytes,
        drops: None, // libpnet exposes no per-socket drop counter
        wall,
        cpu_s: cpu_seconds() - cpu0,
        sys_cpu_s: system_cpu_seconds() - sys0,
        verify: ver,
    }
}

/// libpcap backend, for comparison. Like ferranet's ring it captures off a `PACKET_MMAP` ring
/// (not a recvfrom-per-packet like pnet), so the interesting question is the per-frame CPU gap
/// between libpcap's default ring and ferranet's TPACKET_V3 block model. Given an 8 MiB buffer to
/// match ferranet's ring headroom; full snaplen so it captures whole frames like the ring default.
#[cfg(feature = "pcap-bench")]
fn capture_pcap(iface: &str, secs: f64, warmup: f64, verify: bool) -> Sample {
    let mut cap = pcap::Capture::from_device(iface)
        .unwrap_or_else(|e| die(&format!("pcap device: {e}")))
        .promisc(true)
        .snaplen(65535)
        .buffer_size(8 << 20)
        .timeout(100)
        .open()
        .unwrap_or_else(|e| die(&format!("pcap open: {e}")));

    let warm_until = Instant::now() + Duration::from_secs_f64(warmup);
    while Instant::now() < warm_until {
        match cap.next_packet() {
            Ok(_) | Err(pcap::Error::TimeoutExpired) => {}
            Err(_) => break,
        }
    }

    let mut ver = verify.then(Verify::default);
    let start = Instant::now();
    let (cpu0, sys0) = (cpu_seconds(), system_cpu_seconds());
    let deadline = start + Duration::from_secs_f64(secs);
    let (mut frames, mut bytes) = (0u64, 0u64);
    while Instant::now() < deadline {
        match cap.next_packet() {
            Ok(pkt) => {
                frames += 1;
                bytes += pkt.header.len as u64; // original wire length, like ferranet's wire_len()
                if let Some(v) = ver.as_mut() {
                    v.observe(pkt.data);
                }
            }
            Err(pcap::Error::TimeoutExpired) => {}
            Err(_) => break,
        }
    }
    let wall = start.elapsed().as_secs_f64();
    Sample {
        frames,
        bytes,
        drops: cap.stats().ok().map(|s| u64::from(s.dropped)), // ps_drop: buffer-full drops
        wall,
        cpu_s: cpu_seconds() - cpu0,
        sys_cpu_s: system_cpu_seconds() - sys0,
        verify: ver,
    }
}

fn flood(args: &[String]) {
    let iface = flag(args, "--iface").unwrap_or_else(|| die("--iface is required"));
    let secs: f64 = flag(args, "--secs").and_then(|s| s.parse().ok()).unwrap_or(13.0);
    let size: usize = flag(args, "--size").and_then(|s| s.parse().ok()).unwrap_or(64);
    let rate_bps: Option<f64> = flag(args, "--mbps").and_then(|s| s.parse::<f64>().ok()).map(|m| m * 1e6 / 8.0);

    let (mut tx, _rx) =
        Channel::builder(&iface).build_sync().unwrap_or_else(|e| die(&format!("open: {e}")));
    let frame = make_frame(size);
    let batch: Vec<&[u8]> = (0..64).map(|_| frame.as_slice()).collect();

    let start = Instant::now();
    let deadline = start + Duration::from_secs_f64(secs);
    let (mut sent, mut bytes) = (0u64, 0u64);
    while Instant::now() < deadline {
        let n = tx.send_batch(&batch).unwrap_or(0);
        sent += n as u64;
        bytes += (n * size) as u64;
        if let Some(bps) = rate_bps {
            // Pace to the target rate: sleep until the wall clock catches up to the bytes sent.
            let expected = bytes as f64 / bps;
            let actual = start.elapsed().as_secs_f64();
            if expected > actual {
                std::thread::sleep(Duration::from_secs_f64(expected - actual));
            }
        }
    }
    let wall = start.elapsed().as_secs_f64();
    eprintln!(
        "flood     iface={iface} secs={wall:.2} sent={sent} pps={:.0} Mb/s={:.1} ({size}B)",
        sent as f64 / wall,
        bytes as f64 * 8.0 / wall / 1e6,
    );
}

// ---- helpers ------------------------------------------------------------------------------------

fn report(backend: &str, iface: &str, note: &str, s: &Sample) {
    let pps = s.frames as f64 / s.wall;
    let mbps = (s.bytes as f64 * 8.0) / s.wall / 1e6;
    let proc_pct = s.cpu_s / s.wall * 100.0; // % of one core
    let cpu_us_per_frame = if s.frames > 0 { s.cpu_s * 1e6 / s.frames as f64 } else { 0.0 };
    let sys_cores = s.sys_cpu_s / s.wall; // cores busy system-wide
    let drops = s.drops.map_or_else(|| "n/a".to_string(), |d| d.to_string());
    println!(
        "{backend:<6} iface={iface} secs={s_wall:.2} frames={frames} pps={pps:.0} Mb/s={mbps:.1} \
         drops={drops} proc_cpu={proc_pct:.1}%core cpu_us/frame={cpu_us_per_frame:.3} \
         sys_cpu={sys_cores:.2}cores {note}",
        s_wall = s.wall,
        frames = s.frames,
    );
    if let Some(v) = &s.verify {
        println!(
            "       verify parsed={} parse_err={} ipv4_bad_csum={} | class udp={} tcp={} icmp={} arp={} other={}",
            v.parsed, v.parse_err, v.ipv4_bad_csum, v.udp, v.tcp, v.icmp, v.arp, v.other,
        );
    }
}

/// Total user+system CPU seconds consumed by this process so far.
fn cpu_seconds() -> f64 {
    // SAFETY: rusage is plain-old-data; an all-zero value is a valid initial state.
    let mut ru: libc::rusage = unsafe { mem::zeroed() };
    // SAFETY: `ru` is a valid, writable rusage; RUSAGE_SELF has no other preconditions.
    unsafe { libc::getrusage(libc::RUSAGE_SELF, &mut ru) };
    let secs = |t: libc::timeval| t.tv_sec as f64 + t.tv_usec as f64 / 1e6;
    secs(ru.ru_utime) + secs(ru.ru_stime)
}

/// System-wide busy CPU-seconds since boot (all cores), from `/proc/stat`.
fn system_cpu_seconds() -> f64 {
    let stat = std::fs::read_to_string("/proc/stat").unwrap_or_default();
    let line = stat.lines().next().unwrap_or("");
    let vals: Vec<u64> = line.split_whitespace().skip(1).filter_map(|v| v.parse().ok()).collect();
    if vals.len() < 4 {
        return 0.0;
    }
    let total: u64 = vals.iter().sum();
    let idle = vals[3] + vals.get(4).copied().unwrap_or(0); // idle + iowait
    let busy = total.saturating_sub(idle);
    // SAFETY: sysconf with a valid name has no preconditions.
    let clk = unsafe { libc::sysconf(libc::_SC_CLK_TCK) };
    let clk = if clk > 0 { clk as f64 } else { 100.0 };
    busy as f64 / clk
}

fn make_frame(size: usize) -> Vec<u8> {
    let mut frame = Vec::with_capacity(size.max(14));
    frame.extend_from_slice(&MacAddr::BROADCAST.octets());
    frame.extend_from_slice(&[0x02, 0x00, 0x00, 0x00, 0x00, 0x01]);
    frame.extend_from_slice(&[0x88, 0xb5]);
    frame.resize(size.max(14), 0x5a);
    frame
}

fn flag(args: &[String], name: &str) -> Option<String> {
    args.iter().position(|a| a == name).and_then(|i| args.get(i + 1)).cloned()
}

fn die(msg: &str) -> ! {
    eprintln!("pibench: {msg}");
    std::process::exit(1);
}