patchbay loss test

output.txt

running 1 test
unimpaired
dev1 -> dev2: 1000 of 1000 (100%)
dev2 -> dev1: 1000 of 1000 (100%)
impaired: dev1 on Both with Manual(LinkLimits { rate_kbit: 0, loss_pct: 50.0, latency_ms: 0, jitter_ms: 0, reorder_pct: 0.0, duplicate_pct: 0.0, corrupt_pct: 0.0 })
dev1 -> dev2: 521 of 1000 (52.100002%)
dev2 -> dev1: 500 of 1000 (50%)

impaired: dev1 on Egress with Manual(LinkLimits { rate_kbit: 0, loss_pct: 50.0, latency_ms: 0, jitter_ms: 0, reorder_pct: 0.0, duplicate_pct: 0.0, corrupt_pct: 0.0 })
dev1 -> dev2: 526 of 1000 (52.600002%)
dev2 -> dev1: 1000 of 1000 (100%)

impaired: dev1 on Ingress with Manual(LinkLimits { rate_kbit: 0, loss_pct: 50.0, latency_ms: 0, jitter_ms: 0, reorder_pct: 0.0, duplicate_pct: 0.0, corrupt_pct: 0.0 })
dev1 -> dev2: 1000 of 1000 (100%)
dev2 -> dev1: 488 of 1000 (48.8%)

test tests::loss::loss_smoke ... ok

test result: ok. 1 passed; 0 failed; 0 ignored; 0 measured; 161 filtered out; finished in 32.17s

test.rs

use std::{
    net::{Ipv4Addr, SocketAddr},
    time::Duration,
};

use anyhow::Result;
use n0_tracing_test::traced_test;
use tokio::{net::UdpSocket, sync::oneshot};

use crate::{Device, Lab, LinkCondition, LinkDirection, LinkLimits};

#[tokio::test]
#[traced_test]
async fn loss_smoke() -> Result<()> {
    let var_name = Lab::new().await?;
    let lab = var_name;
    let gw1 = lab.add_router("gw1").build().await?;
    let gw2 = lab.add_router("gw2").build().await?;
    let dev1 = lab.add_device("dev1").uplink(gw1.id()).build().await?;
    let dev2 = lab.add_device("dev2").uplink(gw2.id()).build().await?;

    let count = 1000usize;
    let pace = Duration::from_millis(1);

    println!("unimpaired");
    run(&dev1, &dev2, count, pace).await?;
    run(&dev2, &dev1, count, pace).await?;

    for direction in [
        LinkDirection::Both,
        LinkDirection::Egress,
        LinkDirection::Ingress,
    ] {
        let iface = dev1.iface("eth0").unwrap();
        iface.clear_condition(LinkDirection::Both).await?;
        let condition = LinkCondition::Manual(LinkLimits {
            loss_pct: 50.,
            ..Default::default()
        });
        println!("impaired: dev1 on {direction:?} with {condition:?}");
        iface.set_condition(condition, direction).await?;
        run(&dev1, &dev2, count, pace).await?;
        run(&dev2, &dev1, count, pace).await?;
        println!();
    }

    Ok(())
}

async fn run(sender: &Device, receiver: &Device, count: usize, pace: Duration) -> Result<usize> {
    let timeout = (count as u32) * pace * 4;
    let (bound_tx, bound_rx) = oneshot::channel();
    let recv_addr = receiver.iface("eth0").unwrap().ip().unwrap();
    let recv_task = receiver.spawn(async move |_dev| recv_for(timeout, 1234, bound_tx).await)?;
    bound_rx.await.unwrap();
    let send_task = sender.spawn(async move |_dev| send_n((recv_addr, 1234), count, pace).await)?;
    send_task.await.unwrap()?;
    let recvd = recv_task.await.unwrap()?;
    println!(
        "{} -> {}: {recvd} of {count} ({}%)",
        sender.name(),
        receiver.name(),
        (recvd as f32 / count as f32) * 100.
    );
    Ok(recvd)
}

async fn recv_for(timeout: Duration, port: u16, bound_tx: oneshot::Sender<()>) -> Result<usize> {
    let socket = UdpSocket::bind(SocketAddr::from((Ipv4Addr::UNSPECIFIED, port))).await?;
    bound_tx.send(()).ok();
    let mut buf = [0u8; 2048];
    let mut count = 0;
    let _ = tokio::time::timeout(timeout, async {
        loop {
            let (_n, _from) = socket.recv_from(&mut buf).await.expect("socket died");
            count += 1;
        }
    })
    .await;
    Ok(count)
}

async fn send_n(to: impl Into<SocketAddr>, count: usize, pace: Duration) -> Result<()> {
    let to = to.into();
    let socket = UdpSocket::bind(SocketAddr::from((Ipv4Addr::UNSPECIFIED, 0))).await?;
    for _i in 0..count {
        socket.send_to(&[0u8; 256], to).await?;
        tokio::time::sleep(pace).await;
    }
    Ok(())
}