ferrix/crates/bench/benches/mesh.rs

//! Two routers of one AS, each taking in full tables from its own peers
//! and sending the other every path (ADD-PATH both ways, `all_paths`), as
//! a mesh of routers that each hold the network's whole view do. How
//! long until each router has the other's routes is what this measures:
//! the exporter reading the table's journal while sessions fill the
//! table, which nothing else exercises at this size.
//!
//! Run with `cargo bench -p bgp-bench --bench mesh`. `BENCH_PEERS`
//! (default 4) feed the first router and `BENCH_PEERS_B` (default 2) the
//! second; `BENCH_SCALE_DOWN` shrinks the table. The routers peer over
//! IPv6 loopback and the feeders come over IPv4 loopback.

use std::net::{Ipv4Addr, Ipv6Addr, SocketAddr};
use std::sync::Arc;
use std::time::{Duration, Instant};

use bgp_bench::{Prefixes, Shape, announcements, encoded, env};
use bgp_integration_tests::peer_config;
use bgp_integration_tests::test_peer::TestPeer;
use bgp_integration_tests::test_server::TestServer;
use bgp_packet::open::SendReceive;
use bgp_server::config::PeerConfig;
use bgp_server::policy::{ExportPolicy, ImportPolicy};

/// An AS number that is in none of the synthetic paths.
const ASN: u32 = 4_200_000_000;

fn free_port() -> u16 {
    std::net::TcpListener::bind((Ipv6Addr::LOCALHOST, 0))
        .expect("a port")
        .local_addr()
        .expect("an address")
        .port()
}

/// A router of the mesh, listening for feeders on IPv4 loopback and for
/// the other router on IPv6 loopback at `port`, which it dials at
/// `other`.
async fn router(name: &'static str, id: u8, port: u16, other: u16) -> TestServer {
    TestServer::start_with(ASN, |c| {
        // Two routers of one AS must not share a BGP identifier.
        c.router_id = Ipv4Addr::new(192, 0, 2, id);
        c.cluster_id = c.router_id;
        c.listen_endpoints = vec![
            SocketAddr::new(Ipv4Addr::LOCALHOST.into(), 0),
            SocketAddr::new(Ipv6Addr::LOCALHOST.into(), port),
        ];
        c.peers = vec![PeerConfig {
            add_path: Some(SendReceive::Both),
            import: ImportPolicy::permissive(),
            export: Arc::new(ExportPolicy {
                learned_routes: true,
                all_paths: true,
                ..ExportPolicy::default()
            }),
            ..peer_config(
                name,
                SocketAddr::new(Ipv6Addr::LOCALHOST.into(), other),
                ASN,
            )
        }];
        c.connect_retry = Duration::from_millis(200);
    })
    .await
    .expect("a router")
}

/// Connects `count` feeders to `router`, numbered from `first`, and
/// returns their sessions with the bytes each is to send.
async fn feeders(
    router: &TestServer,
    prefixes: &Prefixes,
    shape: Shape,
    first: u32,
    count: u32,
) -> Vec<(TestPeer, Vec<u8>)> {
    let mut sessions = Vec::new();
    for peer in first..first + count {
        let stream: Vec<u8> = encoded(&announcements(prefixes, shape, peer))
            .iter()
            .flat_map(|m| m.iter().copied())
            .collect();
        let mut session = TestPeer::connect(router.addr).await.expect("a connection");
        session
            .handshake(64512 + peer, Ipv4Addr::new(10, 0, 0, peer as u8), 90)
            .await
            .expect("a session");
        sessions.push((session, stream));
    }
    sessions
}

fn main() {
    let peers_a = env("BENCH_PEERS", 4) as u32;
    let peers_b = env("BENCH_PEERS_B", 2) as u32;
    let shape = Shape::INTERNET.scaled_down(env("BENCH_SCALE_DOWN", 1));
    let per_peer = shape.prefixes();
    println!(
        "== Two routers of AS{ASN}, every path both ways: A fed by {peers_a} peers, B by {peers_b}, {} prefixes each",
        per_peer
    );
    let prefixes = Prefixes::generate(shape);

    let runtime = tokio::runtime::Builder::new_multi_thread()
        .enable_all()
        .build()
        .expect("a runtime");
    runtime.block_on(async {
        let (port_a, port_b) = (free_port(), free_port());
        let a = router("b", 1, port_a, port_b).await;
        let b = router("a", 2, port_b, port_a).await;
        a.wait_for_sessions(1).await.expect("the mesh session at A");
        b.wait_for_sessions(1).await.expect("the mesh session at B");

        let feed_a = feeders(&a, &prefixes, shape, 1, peers_a).await;
        let feed_b = feeders(&b, &prefixes, shape, 1 + peers_a, peers_b).await;
        let own_a = per_peer * peers_a as usize;
        let own_b = per_peer * peers_b as usize;

        let started = Instant::now();
        let mut sending = Vec::new();
        for (mut session, stream) in feed_a.into_iter().chain(feed_b) {
            sending.push(tokio::spawn(async move {
                session.send_raw(&stream).await.expect("the router reads");
                session
            }));
        }
        let mut a_own = None;
        let mut b_own = None;
        let mut a_all = None;
        let mut b_all = None;
        while a_all.is_none() || b_all.is_none() {
            tokio::time::sleep(Duration::from_millis(5)).await;
            let (at_a, at_b) = (a.rib.len(), b.rib.len());
            let now = started.elapsed();
            if a_own.is_none() && at_a >= own_a {
                a_own = Some(now);
            }
            if b_own.is_none() && at_b >= own_b {
                b_own = Some(now);
            }
            if a_all.is_none() && at_a >= own_a + own_b {
                a_all = Some(now);
            }
            if b_all.is_none() && at_b >= own_a + own_b {
                b_all = Some(now);
            }
            if now > Duration::from_secs(env("BENCH_DEADLINE", 600) as u64) {
                let sessions = |router: &TestServer| -> Vec<String> {
                    router
                        .registry
                        .list()
                        .iter()
                        .map(|s| {
                            format!(
                                "{} routes {} rejected {}",
                                s.peer,
                                s.counters.routes(),
                                s.counters.rejected()
                            )
                        })
                        .collect()
                };
                panic!(
                    "not all routes arrived: A has {at_a} {:?}, B has {at_b} {:?}",
                    sessions(&a),
                    sessions(&b)
                );
            }
        }
        let row = |what: &str, when: Option<Duration>, routes: usize| {
            let when = when.unwrap_or_default();
            println!(
                "{what:<52} {:>8.3} s   {:.2} M routes/s",
                when.as_secs_f64(),
                routes as f64 / when.as_secs_f64() / 1e6
            );
        };
        row("A has its own peers' routes", a_own, own_a);
        row("B has its own peers' routes", b_own, own_b);
        row(
            "A has B's routes too (B's exporter, A's session)",
            a_all,
            own_b,
        );
        row(
            "B has A's routes too (A's exporter, B's session)",
            b_all,
            own_a,
        );
        for session in sending {
            drop(session.await);
        }
    });
}