ferranet 0.2.0

//! Criterion end-to-end throughput benchmarks over a `veth` pair: zero-copy ring vs. the basic
//! syscall-per-packet backend (which approximates libpnet's mainline Linux model), plus
//! `PACKET_FANOUT` RX scaling.
//!
//! Needs `CAP_NET_RAW`/`CAP_NET_ADMIN`. Build outside the namespace (it has no network), then run
//! inside an unprivileged user+net namespace:
//!
//! ```sh
//! cargo bench --bench throughput --no-run
//! unshare --user --map-root-user --net cargo bench --bench throughput
//! # HTML report + plots: target/criterion/report/index.html
//! ```
//!
//! Outside a namespace it prints a skip notice and exits cleanly.
//!
//! These are throughput benchmarks driven through Criterion's `iter_custom`: each "iteration"
//! moves `UNIT` frames, and the closure returns the wall-clock time taken, so Criterion reports a
//! statistically sampled frames/second (with plots) rather than a per-call latency.

use std::process::{Command, Stdio};
use std::sync::Arc;
use std::sync::OnceLock;
use std::sync::atomic::{AtomicBool, AtomicU64, Ordering};
use std::thread::{self, JoinHandle};
use std::time::{Duration, Instant};

use criterion::{BenchmarkId, Criterion, Throughput};
use ferranet::{Channel, FanoutMode, MacAddr, Receiver, RingConfig, Sender};
use pnet_datalink::{Channel as PnetChannel, Config as PnetConfig, DataLinkReceiver, DataLinkSender};

const BATCH: usize = 64;
const FLOODERS: usize = 4;
const UNIT: u64 = 50_000; // frames moved per Criterion "iteration"
const SIZES: &[usize] = &[64, 1500];

/// The transmit-side interface: `$FERRANET_BENCH_IFACE_A`, or a temporary `veth` end by default.
fn iface_a() -> &'static str {
    static A: OnceLock<String> = OnceLock::new();
    A.get_or_init(|| std::env::var("FERRANET_BENCH_IFACE_A").unwrap_or_else(|_| "fnbencha".into()))
}

/// The receive-side interface: `$FERRANET_BENCH_IFACE_B`, or a temporary `veth` end by default.
fn iface_b() -> &'static str {
    static B: OnceLock<String> = OnceLock::new();
    B.get_or_init(|| std::env::var("FERRANET_BENCH_IFACE_B").unwrap_or_else(|_| "fnbenchb".into()))
}

/// True when both interfaces are supplied via the environment, i.e. real NICs the bench must not
/// create or destroy.
fn real_nics() -> bool {
    std::env::var_os("FERRANET_BENCH_IFACE_A").is_some()
        && std::env::var_os("FERRANET_BENCH_IFACE_B").is_some()
}

fn main() {
    let real = real_nics();
    if real {
        eprintln!("throughput bench: using real interfaces {} -> {}", iface_a(), iface_b());
    } else if !setup_veth() {
        eprintln!(
            "throughput bench: could not create a veth pair.\n\
             Run inside a namespace: `unshare --user --map-root-user --net cargo bench --bench throughput`,\n\
             or point it at real NICs with FERRANET_BENCH_IFACE_A / FERRANET_BENCH_IFACE_B.\n\
             Skipping."
        );
        return;
    }
    // Only tear down interfaces we created ourselves. `then` (not `then_some`) is essential here:
    // `VethGuard` must be constructed lazily, or its Drop would delete a real, user-supplied NIC.
    let _guard = (!real).then(|| VethGuard);

    // Throughput benches are coarse; fewer/shorter samples keep total runtime sane.
    let mut c = Criterion::default()
        .configure_from_args()
        .sample_size(10)
        .warm_up_time(Duration::from_secs(1))
        .measurement_time(Duration::from_secs(3));

    bench_tx(&mut c);
    bench_rx(&mut c);
    bench_fanout(&mut c);

    c.final_summary();
}

/// Transmit submission rate: time to hand `UNIT` frames to the kernel — ferranet ring vs basic vs
/// libpnet's `send_to`.
fn bench_tx(c: &mut Criterion) {
    let mut group = c.benchmark_group("tx_throughput");
    group.throughput(Throughput::Elements(UNIT));

    for &size in SIZES {
        let frames: Vec<Vec<u8>> = (0..BATCH).map(|_| make_frame(size)).collect();
        let batch: Vec<&[u8]> = frames.iter().map(Vec::as_slice).collect();

        for &basic in &[false, true] {
            let (mut tx, _rx) = build(iface_a(), basic);
            group.bench_with_input(BenchmarkId::new(mode(basic), size), &size, |b, _| {
                b.iter_custom(|iters| {
                    let target = iters * UNIT;
                    let mut sent = 0u64;
                    let start = Instant::now();
                    while sent < target {
                        sent += tx.send_batch(&batch).unwrap_or(0) as u64;
                    }
                    start.elapsed()
                });
            });
        }

        // libpnet: one send_to per frame (its mainline Linux model).
        let (mut tx, _rx) = pnet_channel(iface_a());
        let frame = make_frame(size);
        group.bench_with_input(BenchmarkId::new("libpnet", size), &size, |b, _| {
            b.iter_custom(|iters| {
                let target = iters * UNIT;
                let mut sent = 0u64;
                let start = Instant::now();
                while sent < target {
                    if let Some(Ok(())) = tx.send_to(&frame, None) {
                        sent += 1;
                    }
                }
                start.elapsed()
            });
        });
    }
    group.finish();
}

/// Receive absorption rate while the peer is over-driven, ring vs basic.
fn bench_rx(c: &mut Criterion) {
    let mut group = c.benchmark_group("rx_throughput");
    group.throughput(Throughput::Elements(UNIT));

    for &size in SIZES {
        let stop = Arc::new(AtomicBool::new(false));
        let flooders = spawn_flooders(size, &stop);
        thread::sleep(Duration::from_millis(150)); // let the flood reach steady state

        for &basic in &[false, true] {
            let (_tx, mut rx) = build_rx(iface_b(), basic);
            group.bench_with_input(BenchmarkId::new(mode(basic), size), &size, |b, _| {
                b.iter_custom(|iters| {
                    let target = iters * UNIT;
                    let mut got = 0u64;
                    let start = Instant::now();
                    while got < target {
                        match rx.recv_block() {
                            Ok(block) => got += block.frames().count() as u64,
                            Err(_) => break,
                        }
                    }
                    start.elapsed()
                });
            });
        }

        // libpnet: one next() per frame.
        let (_tx, mut rx) = pnet_channel(iface_b());
        group.bench_with_input(BenchmarkId::new("libpnet", size), &size, |b, _| {
            b.iter_custom(|iters| {
                let target = iters * UNIT;
                let mut got = 0u64;
                let start = Instant::now();
                while got < target {
                    match rx.next() {
                        Ok(_) => got += 1,
                        Err(_) => break,
                    }
                }
                start.elapsed()
            });
        });

        stop.store(true, Ordering::Relaxed);
        join_all(flooders);
    }
    group.finish();
}

/// `PACKET_FANOUT` aggregate RX rate as the receiver count grows (64-byte frames).
fn bench_fanout(c: &mut Criterion) {
    let mut group = c.benchmark_group("fanout_rx_64B");
    group.throughput(Throughput::Elements(UNIT));

    for &n in &[1usize, 2, 4] {
        let counter = Arc::new(AtomicU64::new(0));
        let recv_stop = Arc::new(AtomicBool::new(false));

        let receivers = Channel::builder(iface_b())
            .fanout(FanoutMode::LoadBalance)
            .rx_ring(rx_cfg())
            .build_fanout_rx(n)
            .expect("open fanout group");

        let workers: Vec<_> = receivers
            .into_iter()
            .map(|mut rx| {
                let counter = counter.clone();
                let stop = recv_stop.clone();
                thread::spawn(move || {
                    while !stop.load(Ordering::Relaxed) {
                        match rx.recv_block() {
                            Ok(block) => {
                                counter.fetch_add(block.frames().count() as u64, Ordering::Relaxed);
                            }
                            Err(_) => break,
                        }
                    }
                })
            })
            .collect();

        let flood_stop = Arc::new(AtomicBool::new(false));
        let flooders = spawn_flooders(64, &flood_stop);
        thread::sleep(Duration::from_millis(150));

        group.bench_with_input(BenchmarkId::from_parameter(n), &n, |b, _| {
            b.iter_custom(|iters| {
                let target = iters * UNIT;
                let start_count = counter.load(Ordering::Relaxed);
                let start = Instant::now();
                while counter.load(Ordering::Relaxed).wrapping_sub(start_count) < target {
                    std::hint::spin_loop();
                }
                start.elapsed()
            });
        });

        flood_stop.store(true, Ordering::Relaxed);
        join_all(flooders);
        recv_stop.store(true, Ordering::Relaxed);
        join_all(workers);
    }
    group.finish();
}

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

fn spawn_flooders(size: usize, stop: &Arc<AtomicBool>) -> Vec<JoinHandle<()>> {
    (0..FLOODERS)
        .map(|_| {
            let stop = stop.clone();
            thread::spawn(move || {
                let (mut tx, _rx) = build(iface_a(), false);
                let frames: Vec<Vec<u8>> = (0..BATCH).map(|_| make_frame(size)).collect();
                let batch: Vec<&[u8]> = frames.iter().map(Vec::as_slice).collect();
                while !stop.load(Ordering::Relaxed) {
                    let _ = tx.send_batch(&batch);
                }
            })
        })
        .collect()
}

fn join_all(handles: Vec<JoinHandle<()>>) {
    for h in handles {
        let _ = h.join();
    }
}

fn mode(basic: bool) -> &'static str {
    if basic { "basic" } else { "ring" }
}

fn rx_cfg() -> RingConfig {
    RingConfig { retire_blk_tov_ms: 10, ..RingConfig::rx_default() }
}

fn build(iface: &str, basic: bool) -> (Sender, Receiver) {
    let mut b = Channel::builder(iface);
    if basic {
        b = b.basic();
    }
    b.build_sync().expect("open channel")
}

fn build_rx(iface: &str, basic: bool) -> (Sender, Receiver) {
    let mut b = Channel::builder(iface).rx_ring(rx_cfg());
    if basic {
        b = b.basic();
    }
    b.build_sync().expect("open channel")
}

/// Opens a libpnet Ethernet channel on `iface` for head-to-head comparison.
fn pnet_channel(iface: &str) -> (Box<dyn DataLinkSender>, Box<dyn DataLinkReceiver>) {
    let interface = pnet_datalink::interfaces()
        .into_iter()
        .find(|i| i.name == iface)
        .unwrap_or_else(|| panic!("libpnet could not find interface {iface}"));
    let config = PnetConfig { read_timeout: None, ..PnetConfig::default() };
    match pnet_datalink::channel(&interface, config) {
        Ok(PnetChannel::Ethernet(tx, rx)) => (tx, rx),
        Ok(_) => panic!("unexpected libpnet channel type"),
        Err(e) => panic!("libpnet channel error on {iface}: {e}"),
    }
}

fn make_frame(size: usize) -> Vec<u8> {
    let mut frame = Vec::with_capacity(size);
    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 setup_veth() -> bool {
    let _ = run("ip", &["link", "del", iface_a()]);
    run("ip", &["link", "add", iface_a(), "type", "veth", "peer", "name", iface_b()])
        && run("ip", &["link", "set", iface_a(), "up"])
        && run("ip", &["link", "set", iface_b(), "up"])
}

fn run(cmd: &str, args: &[&str]) -> bool {
    Command::new(cmd)
        .args(args)
        .stdout(Stdio::null())
        .stderr(Stdio::null())
        .status()
        .map(|s| s.success())
        .unwrap_or(false)
}

struct VethGuard;
impl Drop for VethGuard {
    fn drop(&mut self) {
        let _ = run("ip", &["link", "del", iface_a()]);
    }
}