ferranet 0.2.0

//! Hardware integration tests against real network interfaces.
//!
//! These are `#[ignore]`d by default and additionally **skip** unless both
//! `FERRANET_HW_IFACE_A` and `FERRANET_HW_IFACE_B` are set. They are self-contained: each test
//! transmits on interface A and receives on interface B, so point them at two cabled ports
//! (each ideally in its own network namespace so the host stack doesn't short-circuit the link),
//! or at a single port in NIC loopback mode (set both vars to the same name).
//!
//! ```sh
//! cargo test --test hw --no-run
//! sudo FERRANET_HW_IFACE_A=enp1s0f0 FERRANET_HW_IFACE_B=enp1s0f1 \
//!     ./target/debug/deps/hw-<hash> --ignored --test-threads=1
//! # (or run the whole binary under `setcap cap_net_raw,cap_net_admin+ep`)
//! ```
//!
//! Tip: disable RX coalescing offloads first so you see real wire frames, not merged ones:
//! `ethtool -K <dev> gro off lro off`.

use std::sync::Arc;
use std::sync::atomic::{AtomicBool, Ordering};
use std::thread;
use std::time::{Duration, Instant};

use ferranet::{Channel, FanoutMode, Receiver, RingConfig, Sender};

mod common;
use common::{contains, hw_ifaces, make_frame, make_vlan_frame, wait_readable};

const DEADLINE: Duration = Duration::from_secs(3);

/// Skips the test (printing why) if the hardware interfaces aren't configured.
macro_rules! ifaces_or_skip {
    () => {
        match hw_ifaces() {
            Some(pair) => pair,
            None => {
                eprintln!(
                    "skipping: set FERRANET_HW_IFACE_A and FERRANET_HW_IFACE_B to two cabled ports"
                );
                return;
            }
        }
    };
}

fn sender_on(iface: &str) -> Sender {
    Channel::builder(iface).build_sync().expect("open sender").0
}

fn ring_receiver_on(iface: &str) -> Receiver {
    Channel::builder(iface)
        .promiscuous(true)
        .rx_ring(RingConfig { retire_blk_tov_ms: 10, ..RingConfig::rx_default() })
        .build_sync()
        .expect("open receiver")
        .1
}

/// Repeatedly transmits `frame` on A and scans frames received on B, returning the first frame
/// (its data copied out) for which `accept` holds, or `None` past the deadline.
fn send_until_received(
    a: &str,
    b: &str,
    frame: &[u8],
    accept: impl Fn(&ferranet::Frame<'_>) -> bool,
) -> Option<Vec<u8>> {
    let mut tx = sender_on(a);
    let mut rx = ring_receiver_on(b);

    let deadline = Instant::now() + DEADLINE;
    while Instant::now() < deadline {
        for _ in 0..32 {
            let _ = tx.send(frame);
        }
        if wait_readable(rx.as_fd(), 200) {
            if let Ok(block) = rx.recv_block() {
                for f in block.frames() {
                    if accept(&f) {
                        return Some(f.data().to_vec());
                    }
                }
            }
        }
    }
    None
}

#[test]
#[ignore = "requires real interfaces; set FERRANET_HW_IFACE_A/B"]
fn hw_ring_roundtrip() {
    let (a, b) = ifaces_or_skip!();
    let payload = b"ferranet-hw-roundtrip";
    let frame = make_frame(payload);

    let got = send_until_received(&a, &b, &frame, |f| contains(f.data(), payload));
    assert!(got.is_some(), "no matching frame received on {b} from {a}");
}

#[test]
#[ignore = "requires real interfaces; set FERRANET_HW_IFACE_A/B"]
fn hw_vlan_tag() {
    let (a, b) = ifaces_or_skip!();
    let vid = 100u16;
    let payload = b"ferranet-hw-vlan";
    let frame = make_vlan_frame(vid, payload);

    let mut tx = sender_on(&a);
    let mut rx = ring_receiver_on(&b);

    let deadline = Instant::now() + DEADLINE;
    let mut checked = false;
    'outer: while Instant::now() < deadline {
        for _ in 0..32 {
            let _ = tx.send(&frame);
        }
        if wait_readable(rx.as_fd(), 200) {
            if let Ok(block) = rx.recv_block() {
                for f in block.frames() {
                    if contains(f.data(), payload) {
                        match f.vlan() {
                            // Driver/NIC stripped the tag into metadata: validate our parsing.
                            Some(tag) => {
                                assert_eq!(tag.vid(), vid, "decoded VLAN id mismatch");
                                eprintln!("VLAN delivered as metadata: vid={}", tag.vid());
                            }
                            // Driver delivered the tag in-band: the 802.1Q header is in the bytes.
                            None => {
                                assert!(
                                    contains(f.data(), &[0x81, 0x00]),
                                    "VLAN tag neither in metadata nor in-band"
                                );
                                eprintln!("VLAN delivered in-band (no RX VLAN offload)");
                            }
                        }
                        checked = true;
                        break 'outer;
                    }
                }
            }
        }
    }
    assert!(checked, "no VLAN frame received on {b}");
}

#[test]
#[ignore = "requires real interfaces; set FERRANET_HW_IFACE_A/B"]
fn hw_rx_timestamps() {
    let (a, b) = ifaces_or_skip!();
    let frame = make_frame(b"ferranet-hw-timestamp");

    // The kernel software-timestamps ring frames by default; at least one received frame should
    // carry a timestamp.
    let stamped = send_until_received(&a, &b, &frame, |f| f.timestamp().is_some());
    assert!(stamped.is_some(), "received no frame with a timestamp on {b}");
}

#[test]
#[ignore = "requires real interfaces; set FERRANET_HW_IFACE_A/B"]
fn hw_fanout_distributes() {
    let (a, b) = ifaces_or_skip!();

    let receivers = Channel::builder(&b)
        .promiscuous(true)
        .fanout(FanoutMode::LoadBalance)
        .rx_ring(RingConfig { retire_blk_tov_ms: 10, ..RingConfig::rx_default() })
        .build_fanout_rx(2)
        .expect("open fanout group");

    let stop = Arc::new(AtomicBool::new(false));
    let workers: Vec<_> = receivers
        .into_iter()
        .map(|mut rx| {
            let stop = stop.clone();
            thread::spawn(move || {
                let mut count: u64 = 0;
                while !stop.load(Ordering::Relaxed) {
                    if wait_readable(rx.as_fd(), 100) {
                        if let Ok(block) = rx.recv_block() {
                            count += block.frames().count() as u64;
                        }
                    }
                }
                count
            })
        })
        .collect();

    let mut tx = sender_on(&a);
    let frame = make_frame(b"ferranet-hw-fanout");
    let deadline = Instant::now() + Duration::from_secs(2);
    while Instant::now() < deadline {
        for _ in 0..64 {
            let _ = tx.send(&frame);
        }
    }

    stop.store(true, Ordering::Relaxed);
    let counts: Vec<u64> = workers.into_iter().map(|w| w.join().unwrap()).collect();

    let total: u64 = counts.iter().sum();
    assert!(total > 0, "fanout group received no frames on {b}");
    assert!(
        counts.iter().all(|&c| c > 0),
        "fanout did not distribute across both receivers: {counts:?}"
    );
}