diff --git a/Cargo.lock b/Cargo.lock index 6e6368d..3df0705 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -156,6 +156,22 @@ dependencies = [ "tracing", ] +[[package]] +name = "bgp-bench" +version = "0.0.2" +dependencies = [ + "bgp-integration-tests", + "bgp-packet", + "bgp-primitives", + "bgp-rib", + "bgp-server", + "bytes", + "stats_alloc", + "tokio", + "tracing", + "tracing-subscriber", +] + [[package]] name = "bgp-integration-tests" version = "0.0.2" @@ -1356,6 +1372,12 @@ version = "1.2.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "6ce2be8dc25455e1f91df71bfa12ad37d7af1092ae736f3a6cd0e37bc7810596" +[[package]] +name = "stats_alloc" +version = "0.1.10" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "5c0e04424e733e69714ca1bbb9204c1a57f09f5493439520f9f68c132ad25eec" + [[package]] name = "strsim" version = "0.11.1" diff --git a/Cargo.toml b/Cargo.toml index ebb418f..57de73c 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -33,6 +33,7 @@ bgp-rib = { path = "crates/rib" } bgp-server = { path = "crates/server" } bgp-web = { path = "crates/web" } bgp-integration-tests = { path = "crates/integration-tests" } +bgp-bench = { path = "crates/bench" } # --- General --- bitfield = "0.19.0" diff --git a/crates/bench/Cargo.toml b/crates/bench/Cargo.toml new file mode 100644 index 0000000..f07659b --- /dev/null +++ b/crates/bench/Cargo.toml @@ -0,0 +1,31 @@ +[package] +name = "bgp-bench" +description = "Benchmarks at the scale of the Internet routing table" +authors.workspace = true +edition.workspace = true +homepage.workspace = true +license.workspace = true +repository.workspace = true +version.workspace = true +publish = false + +[dependencies] +bgp-packet.workspace = true +bgp-primitives.workspace = true +bytes.workspace = true + +[dev-dependencies] +bgp-integration-tests.workspace = true +bgp-rib.workspace = true +bgp-server.workspace = true +stats_alloc = "0.1.10" +tokio.workspace = true +tracing.workspace = true +tracing-subscriber.workspace = true + +[[bench]] +name = "full_table" +harness = false + +[lints] +workspace = true diff --git a/crates/bench/benches/full_table.rs b/crates/bench/benches/full_table.rs new file mode 100644 index 0000000..87a4ce6 --- /dev/null +++ b/crates/bench/benches/full_table.rs @@ -0,0 +1,342 @@ +//! What it costs to take in the Internet routing table. +//! +//! Run with `cargo bench -p bgp-bench`. Set `BENCH_PEERS` (default 4) and +//! `BENCH_SCALE_DOWN` (default 1; 10 makes the table ten times smaller) to +//! change the size. +//! +//! This is a stopwatch and an allocation counter around whole operations at +//! full size, not a statistical micro-benchmark: the questions are "how +//! long until a full table is in" and "how much memory does it take", and +//! their answers are seconds and gigabytes. + +use std::alloc::System; +use std::hint::black_box; +use std::net::{Ipv4Addr, SocketAddr}; +use std::time::{Duration, Instant}; + +use bgp_bench::{Prefixes, Shape, announcements, encoded, wire_context}; +use bgp_integration_tests::test_peer::TestPeer; +use bgp_integration_tests::test_server::TestServer; +use bgp_packet::message::{Message, PathAttribute, UpdateMessage}; +use bgp_primitives::ip_prefix::IpPrefix; +use bgp_rib::{AdjRibIn, PeerId, ReceivedUpdate, RibEntry, SnapshotEntry}; +use bytes::BytesMut; +use stats_alloc::{INSTRUMENTED_SYSTEM, Region, StatsAlloc}; + +#[global_allocator] +static GLOBAL: &StatsAlloc = &INSTRUMENTED_SYSTEM; + +fn env(name: &str, default: usize) -> usize { + std::env::var(name) + .ok() + .and_then(|v| v.parse().ok()) + .unwrap_or(default) +} + +/// Bytes allocated and not freed since `region` began. +fn retained(region: &Region<'_, System>) -> i64 { + let stats = region.change(); + stats.bytes_allocated as i64 - stats.bytes_deallocated as i64 +} + +fn mib(bytes: i64) -> f64 { + bytes as f64 / (1024.0 * 1024.0) +} + +fn per_second(count: usize, elapsed: Duration) -> String { + let rate = count as f64 / elapsed.as_secs_f64(); + if rate >= 1e6 { + format!("{:.2} M/s", rate / 1e6) + } else { + format!("{:.0} k/s", rate / 1e3) + } +} + +fn row(what: &str, elapsed: Duration, detail: String) { + println!("{what:<44} {:>9.3} s {detail}", elapsed.as_secs_f64()); +} + +fn peer_id(peer: u32) -> PeerId { + PeerId { + addr: SocketAddr::new(Ipv4Addr::new(192, 0, 2, peer as u8).into(), 179), + asn: 64512 + peer, + } +} + +fn main() { + let peers = env("BENCH_PEERS", 4) as u32; + let shape = Shape::INTERNET.scaled_down(env("BENCH_SCALE_DOWN", 1)); + + println!("== Sizes of the types a route is made of"); + for (name, size) in [ + ("IpPrefix", size_of::()), + ("RibEntry", size_of::()), + ( + "(IpPrefix, RibEntry), a table slot", + size_of::<(IpPrefix, RibEntry)>(), + ), + ("PathAttribute", size_of::()), + ("SnapshotEntry", size_of::()), + ] { + println!("{name:<44} {size:>5} bytes"); + } + + println!( + "\n== Table: {} IPv4 + {} IPv6 prefixes, from each of {peers} peers", + shape.ipv4_prefixes, shape.ipv6_prefixes + ); + let prefixes = Prefixes::generate(shape); + let per_peer: Vec> = (1..=peers) + .map(|peer| announcements(&prefixes, shape, peer)) + .collect(); + let wire: Vec> = per_peer.iter().map(|updates| encoded(updates)).collect(); + let updates: usize = per_peer.iter().map(Vec::len).sum(); + let routes = shape.prefixes() * peers as usize; + let wire_bytes: usize = wire.iter().flatten().map(BytesMut::len).sum(); + println!( + "{updates} UPDATE messages, {:.1} prefixes each, {:.1} MiB on the wire, {routes} routes", + routes as f64 / updates as f64, + mib(wire_bytes as i64), + ); + + println!("\n== Reading the messages (one thread)"); + let ctx = wire_context(); + let started = Instant::now(); + let decoded: Vec> = wire + .iter() + .map(|messages| { + messages + .iter() + .map(|m| Message::decode(&ctx, m).expect("what we wrote is well formed")) + .collect() + }) + .collect(); + let elapsed = started.elapsed(); + row( + "decode UPDATEs", + elapsed, + format!( + "{} routes, {} messages, {:.0} MiB/s", + per_second(routes, elapsed), + per_second(updates, elapsed), + mib(wire_bytes as i64) / elapsed.as_secs_f64() + ), + ); + drop(decoded); + + floor(&prefixes, &per_peer, routes); + + println!("\n== The same, as the daemon does it today (one thread)"); + // From here to the filled table: what the routes cost to hold. + let table_region = Region::new(GLOBAL); + let region = Region::new(GLOBAL); + let started = Instant::now(); + let received: Vec> = per_peer + .iter() + .map(|updates| updates.iter().map(ReceivedUpdate::from_message).collect()) + .collect(); + let elapsed = started.elapsed(); + row( + "UPDATE -> routes (ReceivedUpdate)", + elapsed, + format!( + "{} routes, holding {:.0} MiB", + per_second(routes, elapsed), + mib(retained(®ion)) + ), + ); + + let runtime = tokio::runtime::Builder::new_multi_thread() + .enable_all() + .build() + .expect("a runtime"); + + println!("\n== Adj-RIB-In (the actor, fed from one task)"); + runtime.block_on(async { + let rib = AdjRibIn::spawn(); + let started = Instant::now(); + for (peer, updates) in received.into_iter().enumerate() { + let peer = peer_id(peer as u32 + 1); + for update in updates { + rib.apply(peer, update).await.expect("the actor runs"); + } + } + let stored = rib.len().await.expect("the actor runs"); + let elapsed = started.elapsed(); + assert_eq!(stored, routes); + let held = retained(&table_region); + row( + "insert every route", + elapsed, + format!( + "{} routes; table holds {:.0} MiB = {:.0} bytes a route", + per_second(routes, elapsed), + mib(held), + held as f64 / routes as f64 + ), + ); + + let started = Instant::now(); + black_box(rib.len().await.expect("the actor runs")); + row("count routes", started.elapsed(), String::new()); + + let before = Region::new(GLOBAL); + let started = Instant::now(); + let snapshot = rib.snapshot().await.expect("the actor runs"); + let elapsed = started.elapsed(); + row( + "snapshot (what /api/routes/received does)", + elapsed, + format!( + "copies {:.0} MiB; nothing else is applied meanwhile", + mib(retained(&before)) + ), + ); + drop(snapshot); + + // A second announcement of everything by one peer: every route is + // replaced, as after a route refresh. + let again: Vec = per_peer[0] + .iter() + .map(ReceivedUpdate::from_message) + .collect(); + let started = Instant::now(); + for update in again { + rib.apply(peer_id(1), update).await.expect("the actor runs"); + } + rib.len().await.expect("the actor runs"); + let elapsed = started.elapsed(); + row( + "replace one peer's routes", + elapsed, + per_second(shape.prefixes(), elapsed) + " routes", + ); + + let started = Instant::now(); + let removed = rib.remove_peer(peer_id(1)).await.expect("the actor runs"); + let elapsed = started.elapsed(); + assert_eq!(removed, shape.prefixes()); + row( + "remove one peer (session down)", + elapsed, + per_second(removed, elapsed) + " routes", + ); + }); + + println!("\n== End to end: {peers} peers send the table over TCP at once"); + for logging in [false, true] { + // The daemon logs a line for every UPDATE at INFO. With no subscriber + // installed that costs almost nothing; this shows what it costs + // when the lines are formatted, as they are in production (here to + // nowhere, so without the cost of the terminal or the disk). + if logging { + tracing::subscriber::set_global_default( + tracing_subscriber::fmt() + .with_writer(std::io::sink) + .with_max_level(tracing::Level::INFO) + .finish(), + ) + .expect("no subscriber was installed before"); + } + runtime.block_on(async { + let server = TestServer::start(65000).await.expect("a server"); + let mut senders = Vec::new(); + for (index, messages) in wire.iter().enumerate() { + let peer = index as u32 + 1; + let mut session = TestPeer::connect(server.addr).await.expect("a connection"); + session + .handshake(64512 + peer, Ipv4Addr::new(10, 0, 0, peer as u8), 90) + .await + .expect("a session"); + let stream: Vec = messages.iter().flat_map(|m| m.iter().copied()).collect(); + senders.push((session, stream)); + } + let started = Instant::now(); + let sending: Vec<_> = senders + .into_iter() + .map(|(mut session, stream)| { + tokio::spawn(async move { + session.send_raw(&stream).await.expect("the server reads"); + // Held open until the table is in. + session + }) + }) + .collect(); + while server.adj_rib_in.len().await.expect("the actor runs") < routes { + tokio::time::sleep(Duration::from_millis(5)).await; + } + let elapsed = started.elapsed(); + row( + if logging { + "with a log line per UPDATE" + } else { + "with logging off" + }, + elapsed, + per_second(routes, elapsed) + " routes", + ); + for session in sending { + drop(session.await); + } + }); + } +} + +/// A straw man, not a design: the same routes in the plainest compact form, +/// to show how far the current table is from what a hash table can do. +/// +/// A prefix is one integer and a route is the number of its attribute set +/// (here simply the number of the UPDATE that announced it; a real table +/// would intern the sets). One map per peer and address family, filled on +/// one thread with no channel in between. +fn floor(prefixes: &Prefixes, per_peer: &[Vec], routes: usize) { + use std::collections::HashMap; + + fn key(prefix: &IpPrefix) -> u128 { + let mut address = [0u8; 16]; + address[..prefix.prefix.len()].copy_from_slice(&prefix.prefix); + // The address in the high bits, the length in the low eight. An IPv6 + // prefix longer than /120 would not fit; the table has none. + (u128::from_be_bytes(address) & !0xff) | u128::from(prefix.length) + } + + println!("\n== Straw man: packed prefix -> attribute set number, no actor"); + let region = Region::new(GLOBAL); + let started = Instant::now(); + let mut tables: Vec<(HashMap, HashMap)> = Vec::new(); + for updates in per_peer { + let mut v4: HashMap = HashMap::with_capacity(prefixes.ipv4.len()); + let mut v6: HashMap = HashMap::with_capacity(prefixes.ipv6.len()); + for (number, update) in updates.iter().enumerate() { + for prefix in &update.announced_routes { + v4.insert( + (key(prefix) >> 64) as u64 | u64::from(prefix.length), + number as u32, + ); + } + for attribute in &update.path_attributes { + if let PathAttribute::MpReachNlri(mp) = attribute { + for prefix in &mp.prefixes { + v6.insert(key(prefix), number as u32); + } + } + } + } + tables.push((v4, v6)); + } + let elapsed = started.elapsed(); + let stored: usize = tables.iter().map(|(v4, v6)| v4.len() + v6.len()).sum(); + assert_eq!(stored, routes); + let held = retained(®ion); + row( + "insert every route", + elapsed, + format!( + "{} routes; {:.0} MiB = {:.0} bytes a route (attribute sets not counted)", + per_second(routes, elapsed), + mib(held), + held as f64 / routes as f64 + ), + ); + black_box(tables); +} diff --git a/crates/bench/src/lib.rs b/crates/bench/src/lib.rs new file mode 100644 index 0000000..7057ef9 --- /dev/null +++ b/crates/bench/src/lib.rs @@ -0,0 +1,290 @@ +//! A synthetic copy of the Internet routing table, as the UPDATE messages a +//! peer would send to announce it. +//! +//! It is made up, not captured, so that the benchmarks need no data files +//! and give the same input every run. It is shaped like the real thing where +//! that matters for cost: +//! +//! - about a million IPv4 prefixes, most of them /24s, and about a quarter +//! of a million IPv6 prefixes, most of them /48s; +//! - prefixes that share their attributes are announced together, a handful +//! to an UPDATE, as routers do; +//! - AS paths a few hops long, and communities on most routes; +//! - several peers announce the same prefixes, each with attributes of its +//! own. + +use std::collections::HashSet; +use std::net::{Ipv4Addr, Ipv6Addr}; + +use bgp_packet::message::{ + AsPathAttribute, AsPathSegment, CommunitiesPathAttribute, LargeCommunitiesPathAttribute, + LargeCommunity, Message, MpReachNlriPathAttribute, MultiExitDiscPathAttribute, + NextHopPathAttribute, NlriNextHop, OriginPathAttribute, PathAttribute, UpdateMessage, +}; +use bgp_primitives::constants::{AddressFamilyId, Afi, Safi}; +use bgp_primitives::ip_prefix::IpPrefix; +use bgp_primitives::parser::ParserContext; +use bytes::BytesMut; + +/// How big a table to make. +#[derive(Debug, Clone, Copy)] +pub struct Shape { + pub ipv4_prefixes: usize, + pub ipv6_prefixes: usize, + /// How many prefixes share one set of attributes, on average. + pub prefixes_per_attribute_set: usize, +} + +impl Shape { + /// Roughly the default-free zone in 2026. + pub const INTERNET: Self = Self { + ipv4_prefixes: 1_000_000, + ipv6_prefixes: 230_000, + prefixes_per_attribute_set: 6, + }; + + pub fn prefixes(&self) -> usize { + self.ipv4_prefixes + self.ipv6_prefixes + } + + /// The same shape, `factor` times smaller. + pub fn scaled_down(&self, factor: usize) -> Self { + Self { + ipv4_prefixes: self.ipv4_prefixes / factor, + ipv6_prefixes: self.ipv6_prefixes / factor, + ..*self + } + } +} + +/// A small deterministic generator (SplitMix64); no need for a dependency. +pub struct Rng(u64); + +impl Rng { + pub fn new(seed: u64) -> Self { + Self(seed) + } + + pub fn next_u64(&mut self) -> u64 { + self.0 = self.0.wrapping_add(0x9e37_79b9_7f4a_7c15); + let mut z = self.0; + z = (z ^ (z >> 30)).wrapping_mul(0xbf58_476d_1ce4_e5b9); + z = (z ^ (z >> 27)).wrapping_mul(0x94d0_49bb_1331_11eb); + z ^ (z >> 31) + } + + /// A number below `bound`. + pub fn below(&mut self, bound: u64) -> u64 { + self.next_u64() % bound + } + + /// An entry of `weighted`, picked with probability proportional to its + /// weight. + fn pick(&mut self, weighted: &[(T, u64)]) -> T { + let total: u64 = weighted.iter().map(|(_, w)| w).sum(); + let mut at = self.below(total); + for (value, weight) in weighted { + if at < *weight { + return *value; + } + at -= weight; + } + weighted[weighted.len() - 1].0 + } +} + +/// The prefixes of the table. Every peer announces these same prefixes. +pub struct Prefixes { + pub ipv4: Vec, + pub ipv6: Vec, +} + +impl Prefixes { + pub fn generate(shape: Shape) -> Self { + let mut rng = Rng::new(0x1234_5678); + // Shares of prefix lengths, close to what the table looks like. + let v4_lengths: [(u8, u64); 9] = [ + (24, 60), + (23, 10), + (22, 11), + (21, 5), + (20, 4), + (19, 3), + (18, 2), + (16, 4), + (15, 1), + ]; + let v6_lengths: [(u8, u64); 7] = [ + (48, 48), + (32, 14), + (40, 10), + (44, 10), + (36, 6), + (29, 6), + (47, 6), + ]; + + let mut seen = HashSet::new(); + let mut ipv4 = Vec::with_capacity(shape.ipv4_prefixes); + while ipv4.len() < shape.ipv4_prefixes { + let length = rng.pick(&v4_lengths); + // Unicast space, 1.0.0.0 to 223.255.255.255. + let address = 0x0100_0000 + (rng.next_u64() as u32) % 0xdf00_0000; + let prefix = IpPrefix::new( + AddressFamilyId::Ipv4, + address.to_be_bytes().to_vec(), + length, + ) + .expect("four bytes cover any IPv4 prefix"); + if seen.insert(prefix.clone()) { + ipv4.push(prefix); + } + } + let mut ipv6 = Vec::with_capacity(shape.ipv6_prefixes); + while ipv6.len() < shape.ipv6_prefixes { + let length = rng.pick(&v6_lengths); + // Global unicast, 2000::/3. + let address = (0x2000_u128 << 112) | (u128::from(rng.next_u64()) << 61 >> 3); + let prefix = IpPrefix::new( + AddressFamilyId::Ipv6, + address.to_be_bytes().to_vec(), + length, + ) + .expect("sixteen bytes cover any IPv6 prefix"); + if seen.insert(prefix.clone()) { + ipv6.push(prefix); + } + } + Self { ipv4, ipv6 } + } +} + +/// The UPDATE messages with which the peer numbered `peer` announces the +/// whole table. The peer's AS is `64512 + peer`. +pub fn announcements(prefixes: &Prefixes, shape: Shape, peer: u32) -> Vec { + let mut rng = Rng::new(0x9999 + u64::from(peer)); + let peer_asn = 64512 + peer; + let attributes = |rng: &mut Rng| { + let hops = 2 + rng.below(4) as usize; + let mut path = vec![peer_asn]; + path.extend((0..hops).map(|_| 1 + rng.below(400_000) as u32)); + let mut attributes = vec![ + PathAttribute::Origin(OriginPathAttribute::IGP), + PathAttribute::AsPath(AsPathAttribute { + segments: vec![AsPathSegment { + ordered: true, + path, + }], + }), + ]; + let mut later = Vec::new(); + if rng.below(4) == 0 { + later.push(PathAttribute::MultiExitDisc(MultiExitDiscPathAttribute( + rng.below(1000) as u32, + ))); + } + let communities = rng.below(7) as usize; + if communities > 0 { + later.push(PathAttribute::Communities(CommunitiesPathAttribute( + (0..communities) + .map(|_| (rng.below(65000) as u16, rng.below(1000) as u16)) + .collect(), + ))); + } + let large = if rng.below(5) == 0 { + Some(PathAttribute::LargeCommunities( + LargeCommunitiesPathAttribute { + communities: vec![LargeCommunity { + global_admin: peer_asn, + data_1: rng.below(100) as u32, + data_2: rng.below(100) as u32, + }], + }, + )) + } else { + None + }; + (std::mem::take(&mut attributes), later, large) + }; + + let mut updates = Vec::new(); + // Prefixes that share attributes travel together. A set's prefixes are a + // run of the (randomly ordered) list, a few long. + let mut grouped = |list: &[IpPrefix], rng: &mut Rng, ipv6: bool| { + let mut rest = list; + while !rest.is_empty() { + let size = (1 + rng.below(2 * shape.prefixes_per_attribute_set as u64 - 1) as usize) + .min(rest.len()); + let (group, after) = rest.split_at(size); + rest = after; + + let (mut path_attributes, later, large) = attributes(rng); + let update = if ipv6 { + // In order of type code: MED and COMMUNITIES come before + // MP_REACH_NLRI. + path_attributes.extend(later); + path_attributes.push(PathAttribute::MpReachNlri(MpReachNlriPathAttribute { + afi: Afi::IPV6, + safi: Safi::UNICAST, + next_hop: NlriNextHop::Ipv6(Ipv6Addr::new( + 0x2001, + 0xdb8, + 0, + 0, + 0, + 0, + 0, + peer as u16, + )), + prefixes: group.to_vec(), + })); + path_attributes.extend(large); + UpdateMessage { + withdrawn_routes: vec![], + path_attributes, + announced_routes: vec![], + } + } else { + path_attributes.push(PathAttribute::NextHop(NextHopPathAttribute(Ipv4Addr::new( + 192, 0, 2, peer as u8, + )))); + path_attributes.extend(later); + path_attributes.extend(large); + UpdateMessage { + withdrawn_routes: vec![], + path_attributes, + announced_routes: group.to_vec(), + } + }; + updates.push(update); + } + }; + grouped(&prefixes.ipv4, &mut rng, false); + grouped(&prefixes.ipv6, &mut rng, true); + updates +} + +/// `updates` as they are on the wire, on a session with four-octet AS +/// numbers. +pub fn encoded(updates: &[UpdateMessage]) -> Vec { + let ctx = wire_context(); + updates + .iter() + .map(|update| { + let mut wire = BytesMut::new(); + Message::Update(update.clone()) + .to_wire(&ctx, &mut wire) + .expect("a few prefixes fit one message"); + wire + }) + .collect() +} + +/// The context of the sessions the benchmarks pretend to have: an external +/// peer with four-octet AS numbers. +pub fn wire_context() -> ParserContext { + ParserContext { + four_octet_asn: Some(true), + ibgp: false, + } +}