diff --git a/Cargo.lock b/Cargo.lock index 3df0705..8e64adb 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -2,6 +2,12 @@ # It is not intended for manual editing. version = 4 +[[package]] +name = "adler2" +version = "2.0.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "320119579fcad9c21884f5c4861d16174d0e06250625266f50fe6898340abefa" + [[package]] name = "aho-corasick" version = "1.1.4" @@ -166,6 +172,7 @@ dependencies = [ "bgp-rib", "bgp-server", "bytes", + "flate2", "stats_alloc", "tokio", "tracing", @@ -372,6 +379,15 @@ version = "1.0.4" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "b05b61dc5112cbb17e4b6cd61790d9845d13888356391624cbe7e41efeac1e75" +[[package]] +name = "crc32fast" +version = "1.5.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "01a7799fd6b852db0e61728dde9a204c423b44d689dbd432522543614b490e78" +dependencies = [ + "cfg-if", +] + [[package]] name = "displaydoc" version = "0.2.7" @@ -421,6 +437,17 @@ version = "2.3.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "37909eebbb50d72f9059c3b6d82c0463f2ff062c9e95845c43a6c9c0355411be" +[[package]] +name = "flate2" +version = "1.1.10" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "6e634e2e0ebac1ee034020da1ca582e17ffe4e0f5e985823721e168928136dcb" +dependencies = [ + "crc32fast", + "miniz_oxide", + "zlib-rs", +] + [[package]] name = "fnv" version = "1.0.7" @@ -868,6 +895,16 @@ version = "0.3.17" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "6877bb514081ee2a7ff5ef9de3281f14a4dd4bceac4c09388074a6b5df8a139a" +[[package]] +name = "miniz_oxide" +version = "0.9.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b63fbc4a50860e98e7b2aa7804ded1db5cbc3aff9193adaff57a6931bf7c4b4c" +dependencies = [ + "adler2", + "simd-adler32", +] + [[package]] name = "mio" version = "1.2.3" @@ -1344,6 +1381,12 @@ dependencies = [ "libc", ] +[[package]] +name = "simd-adler32" +version = "0.3.10" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "3a219298ac11a56ea9a6d2120044824d6f01aeb034955e7af7bc16858527deea" + [[package]] name = "slab" version = "0.4.12" @@ -2036,6 +2079,12 @@ dependencies = [ "syn 3.0.5", ] +[[package]] +name = "zlib-rs" +version = "0.6.8" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b268e58e7c693d7c271f93ffc4ba3b380412554231c85bf61ca7af91042a4112" + [[package]] name = "zmij" version = "1.0.23" diff --git a/crates/bench/Cargo.toml b/crates/bench/Cargo.toml index f07659b..78542ad 100644 --- a/crates/bench/Cargo.toml +++ b/crates/bench/Cargo.toml @@ -13,6 +13,7 @@ publish = false bgp-packet.workspace = true bgp-primitives.workspace = true bytes.workspace = true +flate2 = "1" [dev-dependencies] bgp-integration-tests.workspace = true @@ -27,5 +28,9 @@ tracing-subscriber.workspace = true name = "full_table" harness = false +[[bench]] +name = "real_table" +harness = false + [lints] workspace = true diff --git a/crates/bench/benches/real_table.rs b/crates/bench/benches/real_table.rs new file mode 100644 index 0000000..d108969 --- /dev/null +++ b/crates/bench/benches/real_table.rs @@ -0,0 +1,274 @@ +//! The same questions as `full_table`, asked of a real table dump: every +//! route that the peers of a RIPE RIS route collector had announced to it. +//! +//! Needs a dump in `crates/bench/data/` (see the README there) or at the +//! path in `BENCH_MRT`. Without one it says so and does nothing, so that +//! `cargo bench` works on a fresh checkout. `BENCH_PEERS` limits the run to +//! the peers with the most routes (default: all of them). + +use std::alloc::System; +use std::collections::HashSet; +use std::net::SocketAddr; +use std::path::PathBuf; +use std::sync::Arc; +use std::time::{Duration, Instant, SystemTime}; + +use bgp_bench::mrt::{Dump, parse_attributes}; +use bgp_packet::message::{NlriNextHop, PathAttribute}; +use bgp_primitives::ip_prefix::IpPrefix; +use bgp_rib::{AdjRibIn, PeerId, ReceivedUpdate, RibEntry}; +use stats_alloc::{INSTRUMENTED_SYSTEM, Region, StatsAlloc}; + +#[global_allocator] +static GLOBAL: &StatsAlloc = &INSTRUMENTED_SYSTEM; + +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 { + format!("{:.2} M/s", count as f64 / elapsed.as_secs_f64() / 1e6) +} + +fn row(what: &str, elapsed: Duration, detail: String) { + println!("{what:<44} {:>9.3} s {detail}", elapsed.as_secs_f64()); +} + +fn dump_path() -> Option { + if let Ok(path) = std::env::var("BENCH_MRT") { + return Some(path.into()); + } + let data = PathBuf::from(env!("CARGO_MANIFEST_DIR")).join("data"); + let mut dumps: Vec = std::fs::read_dir(data) + .ok()? + .filter_map(|entry| Some(entry.ok()?.path())) + .filter(|path| path.extension().is_some_and(|e| e == "gz")) + .collect(); + dumps.sort(); + dumps.pop() +} + +type Attributes = (Arc<[PathAttribute]>, Option); + +fn main() { + let Some(path) = dump_path() else { + println!("No table dump found; see crates/bench/data/README.md. Nothing to do."); + return; + }; + println!("== Reading {}", path.display()); + let started = Instant::now(); + let dump = Dump::read(&path).expect("a readable dump"); + println!( + "{} peers, {} routes, {} distinct attribute sets, read in {:.0} s", + dump.peers.len(), + dump.route_count(), + dump.attribute_sets.len(), + started.elapsed().as_secs_f64() + ); + + let limit = std::env::var("BENCH_PEERS") + .ok() + .and_then(|v| v.parse().ok()) + .unwrap_or(dump.peers.len()); + let peers = dump.largest_peers(limit); + let peers: Vec = peers + .into_iter() + .filter(|&peer| !dump.routes[peer].is_empty()) + .collect(); + describe(&dump, &peers); + + // Parsed once, outside the timings: reading messages is measured by + // `full_table`, and here the attributes arrive already parsed. + let region = Region::new(GLOBAL); + let parsed: Vec> = dump + .attribute_sets + .iter() + .map(|wire| parse_attributes(wire).map(|(a, next_hop)| (Arc::from(a), next_hop))) + .collect(); + let parsed_bytes = retained(®ion); + println!( + "\nattribute sets, parsed (Arc<[PathAttribute]>) {:.0} MiB = {:.0} bytes a set", + mib(parsed_bytes), + parsed_bytes as f64 / parsed.len() as f64 + ); + let unparsed = parsed.iter().filter(|a| a.is_none()).count(); + if unparsed > 0 { + println!("{unparsed} attribute sets could not be parsed; their routes are left out"); + } + + let runtime = tokio::runtime::Builder::new_multi_thread() + .enable_all() + .build() + .expect("a runtime"); + current_table(&runtime, &dump, &peers, &parsed); +} + +/// What the table looks like, as far as it decides what a RIB costs. +fn describe(dump: &Dump, peers: &[usize]) { + let routes: usize = peers.iter().map(|&p| dump.routes[p].len()).sum(); + let full = peers + .iter() + .filter(|&&p| dump.routes[p].len() > 800_000) + .count(); + let prefixes: HashSet = peers + .iter() + .flat_map(|&p| dump.routes[p].iter().map(|(prefix, _)| *prefix)) + .collect(); + let mut sets_used = HashSet::new(); + let mut sets_per_peer = 0usize; + for &peer in peers { + let own: HashSet = dump.routes[peer].iter().map(|(_, set)| *set).collect(); + sets_per_peer += own.len(); + sets_used.extend(own); + } + let attribute_bytes: usize = sets_used + .iter() + .map(|&set| dump.attribute_sets[set as usize].len()) + .sum(); + println!("\n== The table ({} peers used)", peers.len()); + println!("routes {routes:>12}"); + println!("peers with more than 800,000 routes {full:>12}"); + println!( + "distinct prefixes {:>12}", + prefixes.len() + ); + println!( + "routes per prefix {:>12.1}", + routes as f64 / prefixes.len() as f64 + ); + println!( + "distinct attribute sets {:>12}", + sets_used.len() + ); + println!( + " if each peer's were kept apart {sets_per_peer:>12} ({:.2}x: sets shared between peers)", + sets_per_peer as f64 / sets_used.len() as f64 + ); + println!( + "routes per attribute set {:>12.1}", + routes as f64 / sets_used.len() as f64 + ); + println!( + "attribute set on the wire, mean {:>12.0} bytes", + attribute_bytes as f64 / sets_used.len() as f64 + ); +} + +fn peer_id(dump: &Dump, peer: usize) -> PeerId { + PeerId { + // The port tells apart peers that share an address in the dump. + addr: SocketAddr::new(dump.peers[peer].address, peer as u16), + asn: dump.peers[peer].asn, + } +} + +/// A peer's routes as the UPDATEs that would have announced them: routes +/// that share their attributes arrive together. +fn as_updates( + dump: &Dump, + peer: usize, + parsed: &[Option], + received_at: SystemTime, +) -> Vec { + let mut routes = dump.routes[peer].clone(); + routes.sort_by_key(|(_, set)| *set); + routes + .chunk_by(|a, b| a.1 == b.1) + .filter_map(|group| { + let (attributes, next_hop) = parsed[group[0].1 as usize].as_ref()?; + Some(ReceivedUpdate { + withdrawn: vec![], + announced: group + .iter() + .map(|(prefix, _)| { + ( + *prefix, + RibEntry { + path_attributes: Arc::clone(attributes), + next_hop: next_hop.clone(), + received_at, + }, + ) + }) + .collect(), + }) + }) + .collect() +} + +fn current_table( + runtime: &tokio::runtime::Runtime, + dump: &Dump, + peers: &[usize], + parsed: &[Option], +) { + println!("\n== Adj-RIB-In as it is today (the actor, fed from one task)"); + // From before the routes exist to the filled table: what it holds, + // apart from the attribute sets, which were parsed above. + let region = Region::new(GLOBAL); + let now = SystemTime::now(); + let updates: Vec<(PeerId, Vec)> = peers + .iter() + .map(|&peer| (peer_id(dump, peer), as_updates(dump, peer, parsed, now))) + .collect(); + let routes: usize = updates + .iter() + .flat_map(|(_, updates)| updates) + .map(|update| update.announced.len()) + .sum(); + let messages: usize = updates.iter().map(|(_, updates)| updates.len()).sum(); + println!( + "{routes} routes in {messages} UPDATEs, {:.1} routes each", + routes as f64 / messages as f64 + ); + let first = updates[0].0; + let first_routes: usize = updates[0].1.iter().map(|u| u.announced.len()).sum(); + + runtime.block_on(async { + let rib = AdjRibIn::spawn(); + let started = Instant::now(); + for (peer, updates) in updates { + 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(); + let held = retained(®ion); + row( + "insert every route", + elapsed, + format!( + "{} routes; {stored} stored; {:.0} MiB = {:.0} bytes a route, attribute sets not counted", + per_second(routes, elapsed), + mib(held), + held as f64 / stored as f64 + ), + ); + + let before = Region::new(GLOBAL); + let started = Instant::now(); + let snapshot = rib.snapshot().await.expect("the actor runs"); + row( + "snapshot", + started.elapsed(), + format!("copies {:.0} MiB", mib(retained(&before))), + ); + drop(snapshot); + + let started = Instant::now(); + let removed = rib.remove_peer(first).await.expect("the actor runs"); + let elapsed = started.elapsed(); + assert_eq!(removed, first_routes); + row( + "remove one peer (session down)", + elapsed, + per_second(removed, elapsed) + " routes", + ); + }); +} diff --git a/crates/bench/data/README.md b/crates/bench/data/README.md new file mode 100644 index 0000000..3d8ad5a --- /dev/null +++ b/crates/bench/data/README.md @@ -0,0 +1,18 @@ +# Table dumps for the benchmarks + +`cargo bench -p bgp-bench --bench real_table` runs on a real routing table: +a dump of every route that the peers of a route collector had announced to +it. The dumps are large and are not checked in. Fetch one into this +directory: + +```bash +curl -O https://data.ris.ripe.net/rrc03/2026.09/bview.20260915.0000.gz +``` + +That is RIPE RIS collector `rrc03`, which peers at AMS-IX: about 340 MB, +3.4 GB uncompressed. Any `bview` file from or +a RouteViews RIB works; the most recent `.gz` here is used, or the one named +by `BENCH_MRT`. + +The benchmark needs memory in proportion to the dump. `BENCH_PEERS=8` +limits it to the eight peers with the most routes. diff --git a/crates/bench/src/lib.rs b/crates/bench/src/lib.rs index 8a43d29..2d7e204 100644 --- a/crates/bench/src/lib.rs +++ b/crates/bench/src/lib.rs @@ -13,6 +13,8 @@ //! - several peers announce the same prefixes, each with attributes of its //! own. +pub mod mrt; + use std::collections::HashSet; use std::net::{Ipv4Addr, Ipv6Addr}; diff --git a/crates/bench/src/mrt.rs b/crates/bench/src/mrt.rs new file mode 100644 index 0000000..f6796a8 --- /dev/null +++ b/crates/bench/src/mrt.rs @@ -0,0 +1,237 @@ +//! Reads a table dump in MRT format (RFC 6396, TABLE_DUMP_V2), as published +//! by RIPE RIS and RouteViews: the routes that every peer of a route +//! collector had announced to it at one moment. +//! +//! Only what the benchmarks need is read: the IPv4 and IPv6 unicast tables. + +use std::collections::HashMap; +use std::fs::File; +use std::io::{self, BufReader, Read}; +use std::net::{IpAddr, Ipv4Addr, Ipv6Addr}; +use std::path::Path; + +use bgp_packet::message::{NlriNextHop, PathAttribute}; +use bgp_primitives::constants::AddressFamilyId; +use bgp_primitives::ip_prefix::IpPrefix; +use flate2::read::MultiGzDecoder; + +use crate::wire_context; + +const TABLE_DUMP_V2: u16 = 13; +const PEER_INDEX_TABLE: u16 = 1; +const RIB_IPV4_UNICAST: u16 = 2; +const RIB_IPV6_UNICAST: u16 = 4; +const MP_REACH_NLRI: u8 = 14; + +/// A peer of the collector. +#[derive(Debug, Clone)] +pub struct DumpPeer { + pub address: IpAddr, + pub asn: u32, +} + +/// The contents of a table dump. +pub struct Dump { + /// When the dump was taken, in seconds since the Unix epoch. + pub timestamp: u32, + pub peers: Vec, + /// Every distinct list of path attributes in the dump, as it is on the + /// wire. Routes refer to these by index. + pub attribute_sets: Vec>, + /// For each peer (by index into `peers`), its routes: a prefix and the + /// index of its attributes. + pub routes: Vec>, +} + +impl Dump { + /// Reads the dump at `path`, gzip-compressed as it is published. + pub fn read(path: &Path) -> io::Result { + let mut input = BufReader::with_capacity( + 1 << 20, + MultiGzDecoder::new(BufReader::new(File::open(path)?)), + ); + let mut dump = Dump { + timestamp: 0, + peers: Vec::new(), + attribute_sets: Vec::new(), + routes: Vec::new(), + }; + let mut interned: HashMap, u32> = HashMap::new(); + let mut body = Vec::new(); + loop { + let mut header = [0u8; 12]; + match input.read_exact(&mut header) { + Ok(()) => {} + Err(e) if e.kind() == io::ErrorKind::UnexpectedEof => break, + Err(e) => return Err(e), + } + let timestamp = u32::from_be_bytes([header[0], header[1], header[2], header[3]]); + let kind = u16::from_be_bytes([header[4], header[5]]); + let subtype = u16::from_be_bytes([header[6], header[7]]); + let length = u32::from_be_bytes([header[8], header[9], header[10], header[11]]); + body.resize(length as usize, 0); + input.read_exact(&mut body)?; + if kind != TABLE_DUMP_V2 { + continue; + } + let mut record = Cursor(&body); + let malformed = || io::Error::new(io::ErrorKind::InvalidData, "truncated MRT record"); + match subtype { + PEER_INDEX_TABLE => { + dump.timestamp = timestamp; + dump.peers = peer_index_table(&mut record).ok_or_else(malformed)?; + dump.routes = vec![Vec::new(); dump.peers.len()]; + } + RIB_IPV4_UNICAST | RIB_IPV6_UNICAST => { + let version = if subtype == RIB_IPV4_UNICAST { + AddressFamilyId::Ipv4 + } else { + AddressFamilyId::Ipv6 + }; + rib_entries(&mut record, version, &mut dump, &mut interned) + .ok_or_else(malformed)?; + } + _ => {} + } + } + Ok(dump) + } + + pub fn route_count(&self) -> usize { + self.routes.iter().map(Vec::len).sum() + } + + /// The peers with the most routes, most first: the ones that send a + /// full table. + pub fn largest_peers(&self, count: usize) -> Vec { + let mut peers: Vec = (0..self.peers.len()).collect(); + peers.sort_by_key(|&peer| std::cmp::Reverse(self.routes[peer].len())); + peers.truncate(count); + peers + } +} + +struct Cursor<'a>(&'a [u8]); + +impl<'a> Cursor<'a> { + fn take(&mut self, n: usize) -> Option<&'a [u8]> { + if self.0.len() < n { + return None; + } + let (taken, rest) = self.0.split_at(n); + self.0 = rest; + Some(taken) + } + fn u8(&mut self) -> Option { + self.take(1).map(|b| b[0]) + } + fn u16(&mut self) -> Option { + self.take(2).map(|b| u16::from_be_bytes([b[0], b[1]])) + } + fn u32(&mut self) -> Option { + self.take(4) + .map(|b| u32::from_be_bytes([b[0], b[1], b[2], b[3]])) + } +} + +fn peer_index_table(record: &mut Cursor<'_>) -> Option> { + record.u32()?; // collector's BGP identifier + let name = record.u16()?; + record.take(name.into())?; + let count = record.u16()?; + let mut peers = Vec::with_capacity(count.into()); + for _ in 0..count { + let kind = record.u8()?; + record.u32()?; // peer's BGP identifier + let address = if kind & 1 == 1 { + let bytes: [u8; 16] = record.take(16)?.try_into().ok()?; + IpAddr::from(Ipv6Addr::from(bytes)) + } else { + let bytes: [u8; 4] = record.take(4)?.try_into().ok()?; + IpAddr::from(Ipv4Addr::from(bytes)) + }; + let asn = if kind & 2 == 2 { + record.u32()? + } else { + record.u16()?.into() + }; + peers.push(DumpPeer { address, asn }); + } + Some(peers) +} + +fn rib_entries( + record: &mut Cursor<'_>, + version: AddressFamilyId, + dump: &mut Dump, + interned: &mut HashMap, u32>, +) -> Option<()> { + record.u32()?; // sequence number + let length = record.u8()?; + let bytes = record.take(usize::from(length.div_ceil(8)))?; + // A prefix the collector should not have had is not ours to fix. + let prefix = IpPrefix::new(version, bytes, length).ok(); + let count = record.u16()?; + for _ in 0..count { + let peer = usize::from(record.u16()?); + record.u32()?; // when the route was received + let attributes_len = record.u16()?; + let attributes = record.take(attributes_len.into())?; + let (Some(prefix), Some(routes)) = (prefix, dump.routes.get_mut(peer)) else { + continue; + }; + let id = match interned.get(attributes) { + Some(id) => *id, + None => { + let id = u32::try_from(dump.attribute_sets.len()).ok()?; + dump.attribute_sets.push(attributes.into()); + interned.insert(attributes.into(), id); + id + } + }; + routes.push((prefix, id)); + } + Some(()) +} + +/// The path attributes of a route in a dump, parsed. +/// +/// In a dump the AS numbers are always four octets wide, and +/// MP_REACH_NLRI holds only the next hop (RFC 6396 ยง4.3.4): the prefix is in +/// the record. The next hop is returned beside the other attributes, as the +/// RIB keeps it. `None` if the attributes cannot be parsed. +pub fn parse_attributes(mut wire: &[u8]) -> Option<(Vec, Option)> { + let ctx = wire_context(); + let mut attributes = Vec::new(); + let mut next_hop = None; + while !wire.is_empty() { + if wire.get(1) == Some(&MP_REACH_NLRI) { + let extended = wire[0] & 0x10 != 0; + let (length, header) = if extended { + ( + usize::from(u16::from_be_bytes([*wire.get(2)?, *wire.get(3)?])), + 4, + ) + } else { + (usize::from(*wire.get(2)?), 3) + }; + let value = wire.get(header..header + length)?; + let address = |bytes: &[u8]| Some(Ipv6Addr::from(<[u8; 16]>::try_from(bytes).ok()?)); + next_hop = match value { + [4, a, b, c, d] => Some(NlriNextHop::Ipv4(Ipv4Addr::new(*a, *b, *c, *d))), + [16, rest @ ..] => Some(NlriNextHop::Ipv6(address(rest)?)), + [32, rest @ ..] if rest.len() == 32 => Some(NlriNextHop::Ipv6WithLl { + global: address(&rest[..16])?, + link_local: address(&rest[16..])?, + }), + _ => return None, + }; + wire = &wire[header + length..]; + continue; + } + let (rest, attribute) = PathAttribute::from_wire(&ctx, wire).ok()?; + attributes.push(attribute); + wire = rest; + } + Some((attributes, next_hop)) +}