diff --git a/Cargo.lock b/Cargo.lock index 6f83b14..9941abe 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -182,6 +182,7 @@ dependencies = [ "bytes", "flate2", "stats_alloc", + "tikv-jemallocator", "tokio", ] diff --git a/crates/bench/Cargo.toml b/crates/bench/Cargo.toml index 3236b1d..dbbeaf1 100644 --- a/crates/bench/Cargo.toml +++ b/crates/bench/Cargo.toml @@ -16,6 +16,9 @@ bytes.workspace = true flate2 = "1" stats_alloc = { version = "0.1.10", optional = true } +[target.'cfg(not(target_env = "msvc"))'.dependencies] +tikv-jemallocator.workspace = true + [dev-dependencies] bgp-integration-tests.workspace = true bgp-rib.workspace = true @@ -42,3 +45,7 @@ harness = false [lints] workspace = true + +[[bench]] +name = "mesh" +harness = false diff --git a/crates/bench/benches/mesh.rs b/crates/bench/benches/mesh.rs new file mode 100644 index 0000000..f483868 --- /dev/null +++ b/crates/bench/benches/mesh.rs @@ -0,0 +1,195 @@ +//! 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)> { + let mut sessions = Vec::new(); + for peer in first..first + count { + let stream: Vec = 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 { + 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, 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); + } + }); +} diff --git a/crates/bench/src/lib.rs b/crates/bench/src/lib.rs index 0940dcd..e74a061 100644 --- a/crates/bench/src/lib.rs +++ b/crates/bench/src/lib.rs @@ -17,6 +17,13 @@ pub mod memory; pub mod mrt; +// The benchmarks allocate as the daemon does (bin/bgp/src/main.rs), with +// jemalloc; the C library's allocator would put its own locks in the +// profile. Not when memory is counted: that needs the counting one. +#[cfg(all(not(feature = "count-memory"), not(target_env = "msvc")))] +#[global_allocator] +static ALLOCATOR: tikv_jemallocator::Jemalloc = tikv_jemallocator::Jemalloc; + use std::collections::HashSet; use std::net::{Ipv4Addr, Ipv6Addr}; diff --git a/crates/server/src/config.rs b/crates/server/src/config.rs index 133d32a..097ff2c 100644 --- a/crates/server/src/config.rs +++ b/crates/server/src/config.rs @@ -646,6 +646,7 @@ fn parse_import(import: &proto::ImportPolicy) -> Result