From eab891567b3a17b222bb83a95d47dfe8694700ef Mon Sep 17 00:00:00 2001 From: Rayhaan Jaufeerally Date: Fri, 2 Oct 2026 00:01:50 +0200 Subject: [PATCH] Add the sharded routing table: Adj-RIBs-In and Loc-RIB in one bgp_rib::table::RibTable holds every route from every peer and the best one for each prefix. It is not used by the server yet; this adds it with its tests and benchmarks. - Organised by prefix. A row holds the prefix's routes from all peers in one array; choosing again after a change reads that row and nothing else. - A route is sixteen bytes: a pointer to its attribute set and one word holding the peer and the first three steps of selection (LOCAL_PREF, AS path length, ORIGIN) as a single number. That number alone settles most changes; a tie goes on to MED, compared only within a neighbouring AS and in a way that does not depend on arrival order, and the tie-breaks. - Attribute sets are stored once, in wire form, with what selection needs from them worked out when they are stored. - Sharded by hash of the prefix, a lock per shard. A session applies its own UPDATEs; there is no queue or owning thread in front of the table. - Readers take one shard at a time for reading and look at rows in place. Route selection is checked against a plain implementation with no shortcuts by a property test over random announcements, withdrawals and session flaps, for both the chosen route and the count of changes. On a RIPE RIS dump from AMS-IX (148 peers, 39.3 million routes), against the Adj-RIB-In actor, which does no selection: actor table memory 11.0 GiB 1.8 GiB (279 -> 48 bytes a route) load, 1 thread 10.2 s 12.3 s load, 12 threads (no gain) 2.1 s (19 million routes a second) read every route 1.6 s, copying 8 GiB, table stopped 0.2 s, in place, table running The benchmark crate counts memory only with --features count-memory: the counters are shared between threads and distort the timings. Co-Authored-By: Claude Fable 5.1 --- Cargo.lock | 29 +- Cargo.toml | 2 + crates/bench/Cargo.toml | 8 +- crates/bench/benches/full_table.rs | 27 +- crates/bench/benches/real_table.rs | 177 +++++- crates/bench/src/lib.rs | 1 + crates/bench/src/memory.rs | 60 ++ crates/rib/Cargo.toml | 6 + .../rib/proptest-regressions/table/tests.txt | 7 + crates/rib/src/lib.rs | 1 + crates/rib/src/table/attrs.rs | 493 +++++++++++++++ crates/rib/src/table/mod.rs | 571 +++++++++++++++++ crates/rib/src/table/select.rs | 158 +++++ crates/rib/src/table/tests.rs | 578 ++++++++++++++++++ 14 files changed, 2091 insertions(+), 27 deletions(-) create mode 100644 crates/bench/src/memory.rs create mode 100644 crates/rib/proptest-regressions/table/tests.txt create mode 100644 crates/rib/src/table/attrs.rs create mode 100644 crates/rib/src/table/mod.rs create mode 100644 crates/rib/src/table/select.rs create mode 100644 crates/rib/src/table/tests.rs diff --git a/Cargo.lock b/Cargo.lock index 8e64adb..3dbfa6b 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -17,6 +17,12 @@ dependencies = [ "memchr", ] +[[package]] +name = "allocator-api2" +version = "0.2.21" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "683d7910e743518b0e34f1186f92494becacb047c7b6bf616c96772180fef923" + [[package]] name = "anstream" version = "0.6.21" @@ -232,6 +238,10 @@ version = "0.0.2" dependencies = [ "bgp-packet", "bgp-primitives", + "bytes", + "foldhash", + "hashbrown 0.17.1", + "proptest", "serde", "thiserror 2.0.12", "tokio", @@ -454,6 +464,12 @@ version = "1.0.7" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "3f9eec918d3f24069decb9af1554cad7c880e2da24a9afd88aca000531ab82c1" +[[package]] +name = "foldhash" +version = "0.2.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "77ce24cb58228fbb8aa041425bb1050850ac19177686ea6e0f41a70416f56fdb" + [[package]] name = "form_urlencoded" version = "1.2.2" @@ -569,6 +585,17 @@ version = "0.16.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "841d1cc9bed7f9236f321df977030373f4a4163ae1a7dbfe1a51a2c1a51d9100" +[[package]] +name = "hashbrown" +version = "0.17.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ed5909b6e89a2db4456e54cd5f673791d7eca6732202bbf2a9cc504fe2f9b84a" +dependencies = [ + "allocator-api2", + "equivalent", + "foldhash", +] + [[package]] name = "heck" version = "0.5.0" @@ -790,7 +817,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "0ad4bb2b565bca0645f4d68c5c9af97fba094e9791da685bf83cb5f3ce74acf2" dependencies = [ "equivalent", - "hashbrown", + "hashbrown 0.16.1", ] [[package]] diff --git a/Cargo.toml b/Cargo.toml index 57de73c..752b0d3 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -53,7 +53,9 @@ proptest = "1.7" protobuf = { package = "protobuf", version = "3.7.2" } protobuf-codegen = { package = "protobuf-codegen", version = "3.7.2" } clap = { version = "4.5.53", features = ["derive"] } +foldhash = "0.2" futures = "0.3" +hashbrown = "0.17" axum = "0.8.9" tokio-stream = { version = "0.1.19", features = ["sync"] } reqwest = { version = "0.13.5", default-features = false, features = ["json"] } diff --git a/crates/bench/Cargo.toml b/crates/bench/Cargo.toml index 78542ad..fd8bf36 100644 --- a/crates/bench/Cargo.toml +++ b/crates/bench/Cargo.toml @@ -14,16 +14,22 @@ bgp-packet.workspace = true bgp-primitives.workspace = true bytes.workspace = true flate2 = "1" +stats_alloc = { version = "0.1.10", optional = 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 +[features] +# Counts every allocation, to report how much memory a table holds. The +# counters are shared by all threads and slow them down, so timings are +# taken without this and memory is measured in a run of its own. +count-memory = ["dep:stats_alloc"] + [[bench]] name = "full_table" harness = false diff --git a/crates/bench/benches/full_table.rs b/crates/bench/benches/full_table.rs index ce9e0e6..64b9dce 100644 --- a/crates/bench/benches/full_table.rs +++ b/crates/bench/benches/full_table.rs @@ -9,11 +9,11 @@ //! 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::memory::Region; use bgp_bench::{Prefixes, Shape, announcements, encoded, wire_context}; use bgp_integration_tests::test_peer::TestPeer; use bgp_integration_tests::test_server::TestServer; @@ -21,10 +21,6 @@ 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) @@ -33,10 +29,10 @@ fn env(name: &str, default: usize) -> usize { .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 +/// Bytes allocated and not freed since `region` began; zero when memory +/// is not being counted (see `bgp_bench::memory`). +fn retained(region: &Region) -> i64 { + region.held().unwrap_or(0) } fn mib(bytes: i64) -> f64 { @@ -64,6 +60,11 @@ fn peer_id(peer: u32) -> PeerId { } fn main() { + if Region::begin().held().is_none() { + println!( + "Memory is not counted in this run: sizes print as 0. For sizes (and slower\ntimings) add --features count-memory.\n" + ); + } let peers = env("BENCH_PEERS", 4) as u32; let shape = Shape::INTERNET.scaled_down(env("BENCH_SCALE_DOWN", 1)); @@ -128,8 +129,8 @@ fn main() { 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 table_region = Region::begin(); + let region = Region::begin(); let started = Instant::now(); let received: Vec> = per_peer .iter() @@ -180,7 +181,7 @@ fn main() { black_box(rib.len().await.expect("the actor runs")); row("count routes", started.elapsed(), String::new()); - let before = Region::new(GLOBAL); + let before = Region::begin(); let started = Instant::now(); let snapshot = rib.snapshot().await.expect("the actor runs"); let elapsed = started.elapsed(); @@ -301,7 +302,7 @@ fn floor(prefixes: &Prefixes, per_peer: &[Vec], routes: usize) { } println!("\n== Straw man: packed prefix -> attribute set number, no actor"); - let region = Region::new(GLOBAL); + let region = Region::begin(); let started = Instant::now(); let mut tables: Vec<(HashMap, HashMap)> = Vec::new(); for updates in per_peer { diff --git a/crates/bench/benches/real_table.rs b/crates/bench/benches/real_table.rs index d108969..cb4d0dd 100644 --- a/crates/bench/benches/real_table.rs +++ b/crates/bench/benches/real_table.rs @@ -6,25 +6,24 @@ //! `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::net::{Ipv4Addr, SocketAddr}; use std::path::PathBuf; use std::sync::Arc; +use std::sync::atomic::{AtomicUsize, Ordering}; use std::time::{Duration, Instant, SystemTime}; +use bgp_bench::memory::Region; use bgp_bench::mrt::{Dump, parse_attributes}; use bgp_packet::message::{NlriNextHop, PathAttribute}; use bgp_primitives::ip_prefix::IpPrefix; +use bgp_rib::table::{PeerInfo, RibTable}; 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 +/// Bytes allocated and not freed since `region` began; zero when memory +/// is not being counted (see `bgp_bench::memory`). +fn retained(region: &Region) -> i64 { + region.held().unwrap_or(0) } fn mib(bytes: i64) -> f64 { @@ -56,6 +55,11 @@ fn dump_path() -> Option { type Attributes = (Arc<[PathAttribute]>, Option); fn main() { + if Region::begin().held().is_none() { + println!( + "Memory is not counted in this run: sizes print as 0. For sizes (and slower\ntimings) add --features count-memory.\n" + ); + } let Some(path) = dump_path() else { println!("No table dump found; see crates/bench/data/README.md. Nothing to do."); return; @@ -84,7 +88,7 @@ fn main() { // 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 region = Region::begin(); let parsed: Vec> = dump .attribute_sets .iter() @@ -106,6 +110,11 @@ fn main() { .build() .expect("a runtime"); current_table(&runtime, &dump, &peers, &parsed); + drop(runtime); + + let threads = std::thread::available_parallelism().map_or(1, |n| n.get()); + sharded_table(&dump, &peers, &parsed, 1); + sharded_table(&dump, &peers, &parsed, threads); } /// What the table looks like, as far as it decides what a RIB costs. @@ -210,7 +219,7 @@ fn current_table( 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 region = Region::begin(); let now = SystemTime::now(); let updates: Vec<(PeerId, Vec)> = peers .iter() @@ -251,7 +260,7 @@ fn current_table( ), ); - let before = Region::new(GLOBAL); + let before = Region::begin(); let started = Instant::now(); let snapshot = rib.snapshot().await.expect("the actor runs"); row( @@ -272,3 +281,147 @@ fn current_table( ); }); } + +/// The sharded table, filled by `threads` threads. Each takes whole peers, +/// as sessions would, and for every UPDATE interns the attributes and then +/// announces the prefixes: everything between a decoded message and a +/// chosen best route. +fn sharded_table(dump: &Dump, peers: &[usize], parsed: &[Option], threads: usize) { + println!("\n== The sharded table, {threads} thread(s)"); + // A peer's routes grouped as UPDATEs: the attribute set and the + // prefixes announced with it. + let updates: Vec)>> = peers + .iter() + .map(|&peer| { + let mut routes = dump.routes[peer].clone(); + routes.sort_by_key(|(_, set)| *set); + routes + .chunk_by(|a, b| a.1 == b.1) + .filter(|group| parsed[group[0].1 as usize].is_some()) + .map(|group| { + ( + group[0].1, + group.iter().map(|(prefix, _)| *prefix).collect(), + ) + }) + .collect() + }) + .collect(); + let routes: usize = updates + .iter() + .flatten() + .map(|(_, prefixes)| prefixes.len()) + .sum(); + + let region = Region::begin(); + let table = RibTable::new(threads * 16); + let indices: Vec<_> = peers + .iter() + .map(|&peer| { + table + .add_peer(PeerInfo { + asn: dump.peers[peer].asn, + router_id: Ipv4Addr::from(peer as u32 + 1), + address: dump.peers[peer].address, + ibgp: false, + }) + .expect("room for the peer") + }) + .collect(); + + let next = AtomicUsize::new(0); + let best_changes = AtomicUsize::new(0); + let started = Instant::now(); + std::thread::scope(|scope| { + for _ in 0..threads { + scope.spawn(|| { + let mut changes = 0; + loop { + let at = next.fetch_add(1, Ordering::Relaxed); + let Some(updates) = updates.get(at) else { + break; + }; + for (set, prefixes) in updates { + // From the bytes, as a dump has them and as a + // session could hand them over without parsing. + let Some(attrs) = table + .attrs() + .intern_wire(&dump.attribute_sets[*set as usize]) + else { + continue; + }; + changes += table.announce(indices[at], &attrs, prefixes).best_changed; + } + } + best_changes.fetch_add(changes, Ordering::Relaxed); + }); + } + }); + let elapsed = started.elapsed(); + assert_eq!(table.len(), routes); + let held = retained(®ion); + row( + "intern attributes, insert, choose best", + elapsed, + format!( + "{} routes; {:.0} MiB = {:.0} bytes a route, attribute sets included", + per_second(routes, elapsed), + mib(held), + held as f64 / routes as f64 + ), + ); + println!( + " {} prefixes, {} attribute sets, best route changed {} times", + table.prefix_count(), + table.attrs().len(), + best_changes.load(Ordering::Relaxed) + ); + + let started = Instant::now(); + let mut best = 0usize; + table.for_each_best(|prefix, _, _| { + best += usize::from(std::hint::black_box(prefix).length() > 0) + }); + row( + "read every best route", + started.elapsed(), + format!("{best} of them, in place, table not stopped"), + ); + + // Reading has to look at each route for the timing to mean anything. + let started = Instant::now(); + let mut all = 0usize; + let mut hops = 0u64; + table.for_each_route(|_, _, attrs, _| { + all += 1; + hops += u64::from(attrs.key().as_path_len); + }); + row( + "read every route and its attributes", + started.elapsed(), + format!( + "{all} of them, mean AS path {:.1}", + hops as f64 / all as f64 + ), + ); + + let started = Instant::now(); + let outcome = table.remove_peer(indices[0]); + let elapsed = started.elapsed(); + row( + "remove one peer (session down)", + elapsed, + format!( + "{} routes; best route changed for {} prefixes", + per_second(outcome.removed, elapsed), + outcome.best_changed + ), + ); + let started = Instant::now(); + let swept = table.attrs().sweep(); + row( + "drop attribute sets no longer used", + started.elapsed(), + format!("{swept} dropped"), + ); +} diff --git a/crates/bench/src/lib.rs b/crates/bench/src/lib.rs index 2d7e204..f074a99 100644 --- a/crates/bench/src/lib.rs +++ b/crates/bench/src/lib.rs @@ -13,6 +13,7 @@ //! - several peers announce the same prefixes, each with attributes of its //! own. +pub mod memory; pub mod mrt; use std::collections::HashSet; diff --git a/crates/bench/src/memory.rs b/crates/bench/src/memory.rs new file mode 100644 index 0000000..11a961e --- /dev/null +++ b/crates/bench/src/memory.rs @@ -0,0 +1,60 @@ +//! Counting the memory a benchmark holds. +//! +//! With the `count-memory` feature every allocation is counted, and +//! [`Region::held`] says how many bytes were allocated and not freed since +//! the region began. Without it nothing is counted and the answer is +//! `None`: the counters are shared by all threads, and counting slows down +//! exactly the code whose speed is being measured. + +#[cfg(feature = "count-memory")] +mod counting { + use std::alloc::System; + + use stats_alloc::{INSTRUMENTED_SYSTEM, StatsAlloc}; + + #[global_allocator] + static GLOBAL: &StatsAlloc = &INSTRUMENTED_SYSTEM; + + pub struct Region(stats_alloc::Region<'static, System>); + + impl Region { + pub fn begin() -> Self { + Self(stats_alloc::Region::new(GLOBAL)) + } + + pub fn held(&self) -> Option { + let stats = self.0.change(); + Some(stats.bytes_allocated as i64 - stats.bytes_deallocated as i64) + } + } +} + +#[cfg(not(feature = "count-memory"))] +mod counting { + pub struct Region; + + impl Region { + pub fn begin() -> Self { + Self + } + + pub fn held(&self) -> Option { + None + } + } +} + +pub use counting::Region; + +/// "`bytes` MiB = n bytes a `unit`", or a note that memory is not being +/// counted. +pub fn describe(held: Option, count: usize, unit: &str) -> String { + match held { + Some(bytes) => format!( + "{:.0} MiB = {:.0} bytes a {unit}", + bytes as f64 / (1024.0 * 1024.0), + bytes as f64 / count as f64 + ), + None => "memory not counted (run with --features count-memory)".into(), + } +} diff --git a/crates/rib/Cargo.toml b/crates/rib/Cargo.toml index 8eb31f6..ce39351 100644 --- a/crates/rib/Cargo.toml +++ b/crates/rib/Cargo.toml @@ -10,10 +10,16 @@ version.workspace = true [dependencies] bgp-packet.workspace = true bgp-primitives.workspace = true +bytes.workspace = true +foldhash.workspace = true +hashbrown.workspace = true serde.workspace = true thiserror.workspace = true tokio = { workspace = true, features = ["full"] } tracing.workspace = true +[dev-dependencies] +proptest.workspace = true + [lints] workspace = true diff --git a/crates/rib/proptest-regressions/table/tests.txt b/crates/rib/proptest-regressions/table/tests.txt new file mode 100644 index 0000000..160a710 --- /dev/null +++ b/crates/rib/proptest-regressions/table/tests.txt @@ -0,0 +1,7 @@ +# Seeds for failure cases proptest has generated in the past. It is +# automatically read and these particular cases re-run before any +# novel cases are generated. +# +# It is recommended to check this file in to source control so that +# everyone who runs the test benefits from these saved cases. +cc 6bcfc317c976928c2fbb29e943f385127b91770a76f18e7061b80b31f4276955 # shrinks to ops = [Announce { peer: 1, prefix: 1, route: Route { local_pref: None, path: [2], origin: 0, med: None } }, Announce { peer: 2, prefix: 1, route: Route { local_pref: None, path: [1], origin: 0, med: Some(2) } }, Announce { peer: 3, prefix: 1, route: Route { local_pref: None, path: [1], origin: 0, med: None } }, Announce { peer: 3, prefix: 1, route: Route { local_pref: None, path: [1], origin: 1, med: None } }] diff --git a/crates/rib/src/lib.rs b/crates/rib/src/lib.rs index 2e18adc..a599934 100644 --- a/crates/rib/src/lib.rs +++ b/crates/rib/src/lib.rs @@ -3,6 +3,7 @@ mod entry; mod error; mod event; mod local_rib; +pub mod table; pub use adj_rib_in::AdjRibIn; pub use entry::{PeerId, ReceivedUpdate, RibEntry, SnapshotEntry}; diff --git a/crates/rib/src/table/attrs.rs b/crates/rib/src/table/attrs.rs new file mode 100644 index 0000000..88c11bd --- /dev/null +++ b/crates/rib/src/table/attrs.rs @@ -0,0 +1,493 @@ +//! Attribute sets, stored once. +//! +//! A full table has millions of routes and far fewer distinct lists of path +//! attributes: a peer announces prefixes that share their attributes +//! together, about six to an UPDATE on the Internet today. A route in the +//! table is therefore a pointer to an [`AttrSet`], and each distinct set +//! exists once, in the [`AttrStore`]. +//! +//! A set is kept as its bytes on the wire, which is a fifth the size of the +//! parsed form. What choosing between routes needs from it is worked out +//! once, when the set is stored, and kept beside the bytes. + +use std::hash::BuildHasher; +use std::net::{Ipv4Addr, Ipv6Addr}; +use std::sync::{Arc, Mutex, PoisonError}; + +use bgp_packet::message::{NlriNextHop, OriginPathAttribute, PathAttribute}; +use bgp_primitives::parser::{EncodeError, ParserContext}; +use bytes::{BufMut, BytesMut}; +use foldhash::fast::RandomState; +use hashbrown::HashTable; + +const MP_REACH_NLRI: u8 = 14; +const OPTIONAL: u8 = 0x80; + +/// What route selection reads from a set of attributes (RFC 4271 §9.1.2.2). +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub struct SelectionKey { + /// LOCAL_PREF, if the set has one. + pub local_pref: Option, + /// MULTI_EXIT_DISC, if the set has one. + pub med: Option, + /// The AS that the route was learned from: the first AS of the path, or + /// zero if the path is empty or begins with a set. MEDs are comparable + /// only between routes with the same one. + pub neighbor_as: u32, + /// The length of the AS path as selection counts it: an AS_SET is one + /// hop however many AS numbers it holds. + pub as_path_len: u16, + /// ORIGIN: 0 for IGP, 1 for EGP, 2 for INCOMPLETE. + pub origin: u8, +} + +impl SelectionKey { + fn of(attributes: &[PathAttribute]) -> Self { + let mut key = Self { + local_pref: None, + med: None, + neighbor_as: 0, + as_path_len: 0, + // A route without ORIGIN does not get this far; if one did, it + // would be the least preferred. + origin: 2, + }; + for attribute in attributes { + match attribute { + PathAttribute::Origin(origin) => { + key.origin = match origin { + OriginPathAttribute::IGP => 0, + OriginPathAttribute::EGP => 1, + OriginPathAttribute::INCOMPLETE => 2, + } + } + PathAttribute::AsPath(path) => { + let hops: usize = path + .segments + .iter() + .map(|s| if s.ordered { s.path.len() } else { 1 }) + .sum(); + key.as_path_len = u16::try_from(hops).unwrap_or(u16::MAX); + key.neighbor_as = path + .segments + .first() + .filter(|s| s.ordered) + .and_then(|s| s.path.first().copied()) + .unwrap_or(0); + } + PathAttribute::MultiExitDisc(med) => key.med = Some(med.0), + PathAttribute::LocalPref(local_pref) => key.local_pref = Some(local_pref.0), + _ => {} + } + } + key + } +} + +/// Reads the selection key straight off attributes in the form an +/// [`AttrSet`] keeps, without building anything. `None` if the bytes are +/// not a well-formed list of attributes. +fn scan(mut wire: &[u8]) -> Option { + const EXTENDED_LENGTH: u8 = 0x10; + let mut key = SelectionKey { + local_pref: None, + med: None, + neighbor_as: 0, + as_path_len: 0, + origin: 2, + }; + let number = |value: &[u8]| Some(u32::from_be_bytes(<[u8; 4]>::try_from(value).ok()?)); + while !wire.is_empty() { + let (&flags, rest) = wire.split_first()?; + let (&type_code, rest) = rest.split_first()?; + let (length, rest) = if flags & EXTENDED_LENGTH != 0 { + let (length, rest) = rest.split_first_chunk::<2>()?; + (usize::from(u16::from_be_bytes(*length)), rest) + } else { + let (&length, rest) = rest.split_first()?; + (usize::from(length), rest) + }; + let (value, rest) = rest.split_at_checked(length)?; + wire = rest; + match type_code { + 1 => key.origin = *value.first()?, + 2 => { + let mut segments = value; + let mut hops = 0usize; + let mut first = true; + while let [kind, count, rest @ ..] = segments { + let (asns, rest) = rest.split_at_checked(usize::from(*count) * 4)?; + let ordered = *kind == 2; + hops += if ordered { usize::from(*count) } else { 1 }; + if first && ordered { + key.neighbor_as = asns + .first_chunk::<4>() + .map_or(0, |a| u32::from_be_bytes(*a)); + } + first = false; + segments = rest; + } + if !segments.is_empty() { + return None; + } + key.as_path_len = u16::try_from(hops).unwrap_or(u16::MAX); + } + 4 => key.med = Some(number(value)?), + 5 => key.local_pref = Some(number(value)?), + _ => {} + } + } + Some(key) +} + +/// One distinct list of path attributes, with the next hop that goes with +/// it. Made by [`AttrStore::intern`]; two routes have the same attributes +/// exactly when they point to the same `AttrSet`. +#[derive(Debug)] +pub struct AttrSet { + hash: u64, + key: SelectionKey, + /// The attributes as on the wire, with four-octet AS numbers. A next + /// hop that arrived in MP_REACH_NLRI is kept as an MP_REACH_NLRI that + /// holds nothing else, the form RFC 6396 §4.3.4 uses in table dumps. + wire: Box<[u8]>, +} + +impl AttrSet { + pub fn key(&self) -> &SelectionKey { + &self.key + } + + /// The attributes as they are on the wire. + pub fn wire(&self) -> &[u8] { + &self.wire + } + + /// The attributes, parsed, and the next hop if it came in + /// MP_REACH_NLRI. + pub fn attributes(&self) -> (Vec, Option) { + // The bytes were written by `encode`, so they parse. + parse(&self.wire).unwrap_or_default() + } +} + +fn four_octet() -> ParserContext { + ParserContext { + four_octet_asn: Some(true), + ibgp: true, + } +} + +/// Writes `attributes` and `next_hop` in the form an [`AttrSet`] keeps. +fn encode( + attributes: &[PathAttribute], + next_hop: Option<&NlriNextHop>, + out: &mut BytesMut, +) -> Result<(), EncodeError> { + let ctx = four_octet(); + for attribute in attributes { + // These name prefixes, which are not a property of the route. + if matches!( + attribute, + PathAttribute::MpReachNlri(_) | PathAttribute::MpUnreachNlri(_) + ) { + continue; + } + attribute.to_wire(&ctx, out)?; + } + if let Some(next_hop) = next_hop { + out.put_u8(OPTIONAL); + out.put_u8(MP_REACH_NLRI); + out.put_u8(1 + next_hop.wire_len()); + out.put_u8(next_hop.wire_len()); + next_hop.to_wire(&ctx, out)?; + } + Ok(()) +} + +/// Parses path attributes in the form [`AttrSet`] keeps and table dumps +/// use (RFC 6396 §4.3.4): four-octet AS numbers, and an MP_REACH_NLRI that +/// holds only a next hop. `None` if they are malformed. +pub fn parse(mut wire: &[u8]) -> Option<(Vec, Option)> { + let ctx = four_octet(); + 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)) +} + +/// How many locks the store is split over. Sessions intern concurrently, +/// once per UPDATE; this many locks keep them out of each other's way. +const STRIPES: usize = 64; + +/// The one place attribute sets are kept. Safe to share between threads. +pub struct AttrStore { + stripes: Box<[Mutex>>]>, + hasher: RandomState, +} + +impl Default for AttrStore { + fn default() -> Self { + Self::new() + } +} + +impl AttrStore { + pub fn new() -> Self { + Self { + stripes: (0..STRIPES).map(|_| Mutex::default()).collect(), + hasher: RandomState::default(), + } + } + + /// The stored set equal to `attributes` and `next_hop`, added if it was + /// not there. `next_hop` is the one from MP_REACH_NLRI; routes whose + /// next hop is the NEXT_HOP attribute pass `None`. + /// + /// Fails only if the attributes cannot be encoded, which attributes that + /// were decoded from a message always can. + pub fn intern( + &self, + attributes: &[PathAttribute], + next_hop: Option<&NlriNextHop>, + ) -> Result, EncodeError> { + let mut wire = BytesMut::with_capacity(256); + encode(attributes, next_hop, &mut wire)?; + let hash = self.hasher.hash_one(&wire[..]); + + let mut stripe = self.stripe(hash); + if let Some(existing) = stripe.find(hash, |set| set.wire[..] == wire[..]) { + return Ok(Arc::clone(existing)); + } + let key = SelectionKey::of(attributes); + debug_assert_eq!(scan(&wire), Some(key), "both ways of reading the key agree"); + let set = Arc::new(AttrSet { + hash, + key, + wire: wire[..].into(), + }); + stripe.insert_unique(hash, Arc::clone(&set), |set| set.hash); + Ok(set) + } + + /// As [`AttrStore::intern`], for attributes that are already in the + /// form a set keeps: on the wire, with four-octet AS numbers, no + /// MP_UNREACH_NLRI, and an MP_REACH_NLRI (if any) that holds only the + /// next hop. That is how a table dump has them (RFC 6396 §4.3.4). + /// + /// Nothing is parsed: finding a set that is already stored costs a hash + /// of the bytes and a comparison. `None` if the bytes are not a + /// well-formed list of attributes. + pub fn intern_wire(&self, wire: &[u8]) -> Option> { + let hash = self.hasher.hash_one(wire); + let mut stripe = self.stripe(hash); + if let Some(existing) = stripe.find(hash, |set| *set.wire == *wire) { + return Some(Arc::clone(existing)); + } + let set = Arc::new(AttrSet { + hash, + key: scan(wire)?, + wire: wire.into(), + }); + stripe.insert_unique(hash, Arc::clone(&set), |set| set.hash); + Some(set) + } + + /// How many distinct sets are stored. + pub fn len(&self) -> usize { + (0..STRIPES).map(|i| self.stripe_at(i).len()).sum() + } + + pub fn is_empty(&self) -> bool { + self.len() == 0 + } + + /// Drops the sets that no route refers to any more, and returns how + /// many that was. Sets are not dropped as their last route goes, since + /// that would mean a lock here for every route removed; this is called + /// now and then instead. + pub fn sweep(&self) -> usize { + let mut dropped = 0; + for i in 0..STRIPES { + let mut stripe = self.stripe_at(i); + let before = stripe.len(); + // The store's own reference is the only one left. Nobody can + // be taking another: that is done under this lock. + stripe.retain(|set| Arc::strong_count(set) > 1); + dropped += before - stripe.len(); + } + dropped + } + + fn stripe(&self, hash: u64) -> std::sync::MutexGuard<'_, HashTable>> { + // The high bits: the table inside uses the low ones. + self.stripe_at((hash >> 56) as usize % STRIPES) + } + + fn stripe_at(&self, index: usize) -> std::sync::MutexGuard<'_, HashTable>> { + // A panic while holding the lock cannot leave the table half + // changed in a way that matters: see `IgnorePoison` in the server. + self.stripes[index] + .lock() + .unwrap_or_else(PoisonError::into_inner) + } +} + +#[cfg(test)] +mod tests { + use bgp_packet::message::{ + AsPathAttribute, AsPathSegment, CommunitiesPathAttribute, LocalPrefPathAttribute, + MultiExitDiscPathAttribute, NextHopPathAttribute, + }; + + use super::*; + + fn attributes(path: Vec) -> Vec { + vec![ + PathAttribute::Origin(OriginPathAttribute::EGP), + AsPathAttribute::from_asns(path), + PathAttribute::NextHop(NextHopPathAttribute(Ipv4Addr::new(192, 0, 2, 1))), + PathAttribute::MultiExitDisc(MultiExitDiscPathAttribute(50)), + PathAttribute::Communities(CommunitiesPathAttribute(vec![(65000, 1)])), + ] + } + + #[test] + fn equal_attributes_are_one_set() { + let store = AttrStore::new(); + let a = store.intern(&attributes(vec![65001, 65002]), None).unwrap(); + let b = store.intern(&attributes(vec![65001, 65002]), None).unwrap(); + let c = store.intern(&attributes(vec![65001, 65003]), None).unwrap(); + assert!(Arc::ptr_eq(&a, &b)); + assert!(!Arc::ptr_eq(&a, &c)); + assert_eq!(store.len(), 2); + + // The next hop is part of what a set is. + let next_hop = NlriNextHop::Ipv6("2001:db8::1".parse().unwrap()); + let d = store + .intern(&attributes(vec![65001, 65002]), Some(&next_hop)) + .unwrap(); + assert!(!Arc::ptr_eq(&a, &d)); + assert_eq!(store.len(), 3); + } + + #[test] + fn a_set_gives_back_what_went_in() { + let store = AttrStore::new(); + let next_hop = NlriNextHop::Ipv6WithLl { + global: "2001:db8::1".parse().unwrap(), + link_local: "fe80::1".parse().unwrap(), + }; + let original = attributes(vec![65001, 4_200_000_001]); + let set = store.intern(&original, Some(&next_hop)).unwrap(); + assert_eq!(set.attributes(), (original, Some(next_hop))); + } + + #[test] + fn the_selection_key_is_read_off_the_attributes() { + let store = AttrStore::new(); + let mut attrs = vec![ + PathAttribute::Origin(OriginPathAttribute::INCOMPLETE), + PathAttribute::AsPath(AsPathAttribute { + segments: vec![ + AsPathSegment { + ordered: true, + path: vec![65001, 65002], + }, + // A set counts as one hop. + AsPathSegment { + ordered: false, + path: vec![65003, 65004, 65005], + }, + ], + }), + PathAttribute::LocalPref(LocalPrefPathAttribute(200)), + ]; + let key = *store.intern(&attrs, None).unwrap().key(); + assert_eq!( + key, + SelectionKey { + local_pref: Some(200), + med: None, + neighbor_as: 65001, + as_path_len: 3, + origin: 2, + } + ); + + attrs.push(PathAttribute::MultiExitDisc(MultiExitDiscPathAttribute(7))); + assert_eq!(store.intern(&attrs, None).unwrap().key().med, Some(7)); + } + + #[test] + fn attributes_on_the_wire_are_interned_without_parsing_them() { + let store = AttrStore::new(); + let next_hop = NlriNextHop::Ipv6("2001:db8::1".parse().unwrap()); + let mut attrs = attributes(vec![65001, 65002, 65003]); + attrs.push(PathAttribute::LocalPref(LocalPrefPathAttribute(300))); + let parsed = store.intern(&attrs, Some(&next_hop)).unwrap(); + + // The same set by its bytes is the same set, key and all. + let by_wire = store.intern_wire(parsed.wire()).unwrap(); + assert!(Arc::ptr_eq(&parsed, &by_wire)); + + // And one that arrives as bytes first gets the same key as it would + // have had from the parsed form. + let other = AttrStore::new(); + let fresh = other.intern_wire(parsed.wire()).unwrap(); + assert_eq!(fresh.key(), parsed.key()); + assert_eq!(fresh.key().as_path_len, 3); + assert_eq!(fresh.key().neighbor_as, 65001); + assert_eq!(fresh.attributes(), (attrs, Some(next_hop))); + + // Bytes that are not attributes are refused. + assert!(other.intern_wire(&[0x40, 0x01, 0x05, 0x00]).is_none()); + assert!( + other + .intern_wire(&[0x40, 0x02, 0x03, 0x02, 0x01, 0x00]) + .is_none() + ); + } + + #[test] + fn sweeping_drops_only_sets_nobody_holds() { + let store = AttrStore::new(); + let kept = store.intern(&attributes(vec![65001]), None).unwrap(); + drop(store.intern(&attributes(vec![65002]), None).unwrap()); + assert_eq!(store.len(), 2); + + assert_eq!(store.sweep(), 1); + assert_eq!(store.len(), 1); + // The one that is left is still the one that is found. + let again = store.intern(&attributes(vec![65001]), None).unwrap(); + assert!(Arc::ptr_eq(&kept, &again)); + } +} diff --git a/crates/rib/src/table/mod.rs b/crates/rib/src/table/mod.rs new file mode 100644 index 0000000..b1ece43 --- /dev/null +++ b/crates/rib/src/table/mod.rs @@ -0,0 +1,571 @@ +//! The routing table: every route every peer has announced, and for each +//! prefix the one chosen as best. +//! +//! This is the Adj-RIBs-In and the Loc-RIB of RFC 4271 §3.2 in one +//! structure, organised by prefix: a prefix's row holds the routes for it +//! from all peers, side by side, and which of them is best. Choosing again +//! after a change looks at one row and nothing else. +//! +//! It is built for a table of hundreds of full feeds, hundreds of millions +//! of routes, taken in on many cores at once: +//! +//! - **A route is small**: sixteen bytes. A pointer to its attribute set, +//! which is shared with every other route that has the same attributes +//! ([`attrs`]), and one word that holds the peer it is from and the first +//! steps of route selection as a single number to compare. +//! - **The table is in shards**, by hash of the prefix, each behind its own +//! lock. A session applies an UPDATE itself, on its own thread, taking +//! the locks of the few shards its prefixes fall in. There is no queue in +//! front of the table and no thread that owns it; sessions that touch +//! different shards do not wait for each other, and one that has to wait +//! is slowed down, which is the back-pressure that is wanted. +//! - **Readers do not stop the table.** Reading takes a shard's lock for +//! reading, one shard at a time, and looks at the rows in place. + +pub mod attrs; +mod select; +#[cfg(test)] +mod tests; + +use std::hash::BuildHasher; +use std::net::{IpAddr, Ipv4Addr}; +use std::sync::{Arc, Mutex, PoisonError, RwLock, RwLockReadGuard, RwLockWriteGuard}; + +use bgp_primitives::ip_prefix::IpPrefix; +use foldhash::fast::RandomState; +use hashbrown::HashTable; +use hashbrown::hash_table::Entry; + +pub use attrs::{AttrSet, AttrStore, SelectionKey}; + +/// What route selection needs to know about a peer. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub struct PeerInfo { + pub asn: u32, + /// The peer's BGP Identifier. + pub router_id: Ipv4Addr, + pub address: IpAddr, + /// Whether the peer is in our own AS. + pub ibgp: bool, +} + +/// A peer's number in a table, given out by [`RibTable::add_peer`]. Small, +/// because every route carries one. +#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, PartialOrd, Ord)] +pub struct PeerIndex(u16); + +impl PeerIndex { + pub fn as_usize(self) -> usize { + self.0.into() + } +} + +/// The LOCAL_PREF of a route that does not carry one: every route from an +/// external peer, until import policy says otherwise. +pub const DEFAULT_LOCAL_PREF: u32 = 100; + +/// What a change to the table did. +#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)] +pub struct Outcome { + /// Routes that were not in the table before. + pub added: usize, + /// Routes that took the place of one from the same peer. + pub replaced: usize, + /// Routes that were taken out. + pub removed: usize, + /// Prefixes whose best route is now another one, or none. + pub best_changed: usize, +} + +impl std::ops::AddAssign for Outcome { + fn add_assign(&mut self, other: Self) { + self.added += other.added; + self.replaced += other.replaced; + self.removed += other.removed; + self.best_changed += other.best_changed; + } +} + +/// One route of a prefix: sixteen bytes. +struct Path { + /// Two things in one word. The high 48 bits are the first steps of + /// route selection as one number, the larger the better; see + /// [`select::preference`]. The low 16 are the peer the route is from. + word: u64, + attrs: Arc, +} + +impl Path { + fn new(preference: u64, peer: PeerIndex, attrs: &Arc) -> Self { + Self { + word: (preference << 16) | u64::from(peer.0), + attrs: Arc::clone(attrs), + } + } + + fn preference(&self) -> u64 { + self.word >> 16 + } + + fn peer(&self) -> PeerIndex { + PeerIndex(self.word as u16) + } +} + +/// Everything about one prefix. +struct Row { + prefix: IpPrefix, + /// Index of the best route in `paths`. A row always has at least one + /// route, and so always a best one. + best: u16, + /// The routes, one from each peer that has announced the prefix, in one + /// run of memory: finding a peer's route and choosing the best are + /// scans of it. + paths: Vec, +} + +impl Row { + fn position(&self, peer: PeerIndex) -> Option { + self.paths.iter().position(|path| path.peer() == peer) + } +} + +struct Shard { + rows: HashTable, + routes: usize, + /// A copy of the peer list, so that route selection reads it under the + /// lock it already holds. + peers: Vec>, + hasher: RandomState, + /// Reused between selections. + scratch: select::Scratch, +} + +/// The routing table. Shared between threads by reference; every method +/// takes `&self`. +pub struct RibTable { + shards: Box<[RwLock]>, + hasher: RandomState, + attrs: AttrStore, + /// Which peer numbers are in use. Changing the peer list also takes + /// every shard's lock in turn; this serialises that. + peers: Mutex>>, +} + +/// No more peers can be added: every [`PeerIndex`] is in use. +#[derive(Debug, Clone, Copy, PartialEq, Eq, thiserror::Error)] +#[error("the table has no room for another peer")] +pub struct TooManyPeers; + +impl RibTable { + /// A table in `shards` shards. More shards let more sessions write at + /// once; a few times the number of cores is plenty. + pub fn new(shards: usize) -> Self { + let hasher = RandomState::default(); + Self { + shards: (0..shards.max(1)) + .map(|_| { + RwLock::new(Shard { + rows: HashTable::new(), + routes: 0, + peers: Vec::new(), + hasher: hasher.clone(), + scratch: select::Scratch::default(), + }) + }) + .collect(), + hasher, + attrs: AttrStore::new(), + peers: Mutex::default(), + } + } + + /// Where attribute sets for this table's routes are made. + pub fn attrs(&self) -> &AttrStore { + &self.attrs + } + + /// Makes room for a peer's routes and returns the number to announce + /// them under. + pub fn add_peer(&self, info: PeerInfo) -> Result { + let mut peers = self.peers.lock().unwrap_or_else(PoisonError::into_inner); + let index = match peers.iter().position(Option::is_none) { + Some(free) => free, + None => { + // u16::MAX is kept back: see `select::NONE`. + if peers.len() >= usize::from(u16::MAX) { + return Err(TooManyPeers); + } + peers.push(None); + peers.len() - 1 + } + }; + peers[index] = Some(info); + for shard in &self.shards { + let mut shard = write(shard); + if shard.peers.len() <= index { + shard.peers.resize(index + 1, None); + } + shard.peers[index] = Some(info); + } + Ok(PeerIndex(index as u16)) + } + + /// Takes out every route from `peer` and gives its number back for + /// reuse (RFC 4271 §8.2.2: the routes of a session that ended are + /// deleted). + pub fn remove_peer(&self, peer: PeerIndex) -> Outcome { + let mut peers = self.peers.lock().unwrap_or_else(PoisonError::into_inner); + let mut outcome = Outcome::default(); + for shard in &self.shards { + let mut shard = write(shard); + outcome += shard.remove_peer(peer); + if let Some(slot) = shard.peers.get_mut(peer.as_usize()) { + *slot = None; + } + } + if let Some(slot) = peers.get_mut(peer.as_usize()) { + *slot = None; + } + outcome + } + + /// Announces `prefixes` from `peer`, all with the attributes `attrs`: + /// what one UPDATE does. A route the peer had announced before for the + /// same prefix is replaced. + pub fn announce( + &self, + peer: PeerIndex, + attrs: &Arc, + prefixes: &[IpPrefix], + ) -> Outcome { + let mut outcome = Outcome::default(); + self.by_shard(prefixes, |shard, placed| { + let Some(info) = shard.peers.get(peer.as_usize()).copied().flatten() else { + // The peer was removed while this was on its way. + return; + }; + let preference = select::preference(attrs.key(), &info); + for (hash, prefix) in placed { + if let Some(prefix) = prefix { + outcome += shard.announce(*hash, *prefix, peer, preference, attrs); + } + } + }); + outcome + } + + /// Withdraws `peer`'s routes for `prefixes`. Withdrawing a route that + /// is not there does nothing. + pub fn withdraw(&self, peer: PeerIndex, prefixes: &[IpPrefix]) -> Outcome { + let mut outcome = Outcome::default(); + self.by_shard(prefixes, |shard, placed| { + for (hash, prefix) in placed { + if let Some(prefix) = prefix { + outcome += shard.withdraw(*hash, prefix, peer); + } + } + }); + outcome + } + + /// Calls `apply` once for each shard that some of `prefixes` fall in, + /// with that shard locked and those prefixes. + fn by_shard( + &self, + prefixes: &[IpPrefix], + mut apply: impl FnMut(&mut Shard, &[(u64, Option)]), + ) { + let shards = self.shards.len() as u64; + // The high bits pick the shard; the table inside uses the low ones. + let shard_of = |hash: u64| ((hash >> 32) * shards) >> 32; + // An UPDATE has a handful of prefixes; they are sorted on the stack. + const INLINE: usize = 32; + let mut inline = [(0u64, prefixes.first().copied()); INLINE]; + let mut spilled = Vec::new(); + let placed: &mut [(u64, Option)] = if prefixes.len() <= INLINE { + &mut inline[..prefixes.len()] + } else { + spilled.resize(prefixes.len(), (0, None)); + &mut spilled + }; + for (slot, prefix) in placed.iter_mut().zip(prefixes) { + *slot = (self.hasher.hash_one(prefix), Some(*prefix)); + } + placed.sort_unstable_by_key(|(hash, _)| shard_of(*hash)); + for run in placed.chunk_by(|a, b| shard_of(a.0) == shard_of(b.0)) { + let mut shard = write(&self.shards[shard_of(run[0].0) as usize]); + apply(&mut shard, run); + } + } + + /// How many routes the table holds, from all peers. + pub fn len(&self) -> usize { + self.shards.iter().map(|shard| read(shard).routes).sum() + } + + pub fn is_empty(&self) -> bool { + self.len() == 0 + } + + /// How many prefixes have at least one route. + pub fn prefix_count(&self) -> usize { + self.shards.iter().map(|shard| read(shard).rows.len()).sum() + } + + /// The best route for `prefix`: who it is from, and its attributes. + pub fn best(&self, prefix: &IpPrefix) -> Option<(PeerIndex, Arc)> { + self.with_row(prefix, |row| { + let best = &row.paths[usize::from(row.best)]; + (best.peer(), Arc::clone(&best.attrs)) + }) + } + + /// Every route for `prefix`, with whether it is the best one. + pub fn routes(&self, prefix: &IpPrefix) -> Vec<(PeerIndex, Arc, bool)> { + self.with_row(prefix, |row| { + row.paths + .iter() + .enumerate() + .map(|(i, path)| { + ( + path.peer(), + Arc::clone(&path.attrs), + i == usize::from(row.best), + ) + }) + .collect() + }) + .unwrap_or_default() + } + + fn with_row(&self, prefix: &IpPrefix, read_row: impl FnOnce(&Row) -> T) -> Option { + let hash = self.hasher.hash_one(prefix); + let shards = self.shards.len() as u64; + let shard = read(&self.shards[(((hash >> 32) * shards) >> 32) as usize]); + shard + .rows + .find(hash, |row| row.prefix == *prefix) + .map(read_row) + } + + /// Calls `visit` with the best route of every prefix, in no particular + /// order. The table is not stopped: one shard at a time is held for + /// reading, so the routes seen are not all of one moment. + pub fn for_each_best(&self, mut visit: impl FnMut(&IpPrefix, PeerIndex, &Arc)) { + for shard in &self.shards { + for row in &read(shard).rows { + let best = &row.paths[usize::from(row.best)]; + visit(&row.prefix, best.peer(), &best.attrs); + } + } + } + + /// Calls `visit` with every route in the table and whether it is its + /// prefix's best. As [`RibTable::for_each_best`], without stopping the + /// table. + pub fn for_each_route(&self, mut visit: impl FnMut(&IpPrefix, PeerIndex, &Arc, bool)) { + for shard in &self.shards { + for row in &read(shard).rows { + for (i, path) in row.paths.iter().enumerate() { + visit( + &row.prefix, + path.peer(), + &path.attrs, + i == usize::from(row.best), + ); + } + } + } + } +} + +fn read(shard: &RwLock) -> RwLockReadGuard<'_, Shard> { + // A panic under the lock cannot leave a row half changed in a way that + // is unsafe to read; the poison is ignored, as for the server's tables. + shard.read().unwrap_or_else(PoisonError::into_inner) +} + +fn write(shard: &RwLock) -> RwLockWriteGuard<'_, Shard> { + shard.write().unwrap_or_else(PoisonError::into_inner) +} + +/// Adds to a vector without the doubling that `push` does: a row of a few +/// hundred routes should not hold room for as many again. Grows by a +/// quarter, which keeps adding cheap on average. +fn push_tight(vec: &mut Vec, value: T) { + if vec.len() == vec.capacity() { + vec.reserve_exact((vec.len() / 4).max(1)); + } + vec.push(value); +} + +impl Shard { + fn announce( + &mut self, + hash: u64, + prefix: IpPrefix, + peer: PeerIndex, + preference: u64, + attrs: &Arc, + ) -> Outcome { + let hasher = &self.hasher; + let entry = self.rows.entry( + hash, + |row| row.prefix == prefix, + |row| hasher.hash_one(row.prefix), + ); + let path = Path::new(preference, peer, attrs); + let row = match entry { + Entry::Vacant(vacant) => { + vacant.insert(Row { + prefix, + best: 0, + paths: vec![path], + }); + self.routes += 1; + return Outcome { + added: 1, + best_changed: 1, + ..Outcome::default() + }; + } + Entry::Occupied(occupied) => occupied.into_mut(), + }; + + let best = usize::from(row.best); + let best_before = row.paths[best].preference(); + match row.position(peer) { + None => { + push_tight(&mut row.paths, path); + self.routes += 1; + let added = row.paths.len() - 1; + // Better than the best in the first steps of selection is + // better than everything; worse than it changes nothing. A + // tie has to be settled by the later steps. + let best_changed = if preference > best_before { + row.best = added as u16; + true + } else if preference < best_before { + false + } else if let Some(wins) = select::settles_tie(row, added, &self.peers) { + if wins { + row.best = added as u16; + } + wins + } else { + select::choose(row, &self.peers, &mut self.scratch) + }; + Outcome { + added: 1, + best_changed: best_changed.into(), + ..Outcome::default() + } + } + Some(at) => { + let before = std::mem::replace(&mut row.paths[at], path); + let same = Arc::ptr_eq(&before.attrs, attrs) && before.preference() == preference; + let best_changed = if same { + false + } else if at == best { + // The best route itself changed. If it got better in + // the first steps it is still the best, but it is a + // different route now. + if preference <= best_before { + select::choose(row, &self.peers, &mut self.scratch); + } + true + } else if preference > best_before { + row.best = at as u16; + true + } else if preference < best_before && before.preference() < best_before { + // It was not in the running and still is not. + false + } else { + select::choose(row, &self.peers, &mut self.scratch) + }; + Outcome { + replaced: 1, + best_changed: best_changed.into(), + ..Outcome::default() + } + } + } + } + + fn withdraw(&mut self, hash: u64, prefix: &IpPrefix, peer: PeerIndex) -> Outcome { + let Ok(mut entry) = self.rows.find_entry(hash, |row| row.prefix == *prefix) else { + return Outcome::default(); + }; + let row = entry.get_mut(); + let Some(at) = row.position(peer) else { + return Outcome::default(); + }; + self.routes -= 1; + if row.paths.len() == 1 { + entry.remove(); + return Outcome { + removed: 1, + best_changed: 1, + ..Outcome::default() + }; + } + let best_changed = Self::remove_path(row, at, &self.peers, &mut self.scratch); + Outcome { + removed: 1, + best_changed: best_changed.into(), + ..Outcome::default() + } + } + + /// Takes the route at `at` out of a row that has others, and returns + /// whether the best route changed. + fn remove_path( + row: &mut Row, + at: usize, + peers: &[Option], + scratch: &mut select::Scratch, + ) -> bool { + let best = usize::from(row.best); + let best_before = row.paths[best].preference(); + let last = row.paths.len() - 1; + let removed = row.paths.swap_remove(at); + if at == best { + select::choose(row, peers, scratch); + return true; + } + // The last route moved into the gap. + if best == last { + row.best = at as u16; + } + // A route that tied with the best in the first steps may have been + // what kept another from being best, by its MED. + if removed.preference() == best_before { + select::choose(row, peers, scratch) + } else { + false + } + } + + fn remove_peer(&mut self, peer: PeerIndex) -> Outcome { + let mut outcome = Outcome::default(); + let peers = &self.peers; + let scratch = &mut self.scratch; + self.rows.retain(|row| { + let Some(at) = row.position(peer) else { + return true; + }; + outcome.removed += 1; + if row.paths.len() == 1 { + outcome.best_changed += 1; + return false; + } + if Self::remove_path(row, at, peers, scratch) { + outcome.best_changed += 1; + } + true + }); + self.routes -= outcome.removed; + outcome + } +} diff --git a/crates/rib/src/table/select.rs b/crates/rib/src/table/select.rs new file mode 100644 index 0000000..29d93b4 --- /dev/null +++ b/crates/rib/src/table/select.rs @@ -0,0 +1,158 @@ +//! Choosing the best route for a prefix (RFC 4271 §9.1.2.2). +//! +//! The steps, in order. A route wins at the first step where it differs: +//! +//! 1. highest LOCAL_PREF; +//! 2. shortest AS path; +//! 3. lowest ORIGIN; +//! 4. lowest MULTI_EXIT_DISC, among routes from the same neighbouring AS; +//! 5. a route from an external peer over one from an internal peer; +//! 6. the peer with the lowest BGP Identifier; +//! 7. the peer with the lowest address. +//! +//! (The RFC's step between 5 and 6, the interior cost to the next hop, has +//! nothing to compare until there is an IGP to ask.) +//! +//! Steps 1 to 3 are put into one number per route when it is stored, +//! [`preference`]. Most of the time that number alone settles it: a new +//! route with a larger one than the best is the new best, with a smaller +//! one changes nothing, and neither needs a look at any other route. Only a +//! tie goes on to [`choose`], which looks at the whole row, and that begins +//! with a pass over the numbers to find the routes still in the running. + +use super::attrs::SelectionKey; +use super::{DEFAULT_LOCAL_PREF, PeerInfo, Row}; + +/// Steps 1 to 3 as one number of 48 bits: the larger, the better. +/// +/// LOCAL_PREF fills the high 32 bits. Below it come the AS path length in +/// 12 bits and the ORIGIN in 2, each turned upside down so that shorter and +/// lower compare as larger. Twelve bits are enough for any path: a BGP +/// message of 4096 octets has room for about a thousand AS numbers. +pub(super) fn preference(key: &SelectionKey, peer: &PeerInfo) -> u64 { + // LOCAL_PREF is a matter between routers of one AS. A route from an + // external peer has the default until import policy gives it another. + let local_pref = if peer.ibgp { + key.local_pref.unwrap_or(DEFAULT_LOCAL_PREF) + } else { + DEFAULT_LOCAL_PREF + }; + const LONGEST: u16 = (1 << 12) - 1; + (u64::from(local_pref) << 16) + | (u64::from(LONGEST - key.as_path_len.min(LONGEST)) << 4) + | (u64::from(3 - key.origin.min(3)) << 2) +} + +/// Buffers that [`choose`] reuses. +#[derive(Default)] +pub(super) struct Scratch { + /// Routes still in the running: (neighbouring AS, MED, index in row). + candidates: Vec<(u32, u32, u16)>, +} + +/// Chooses the best route of `row` from all its routes, records it in the +/// row, and returns whether it is another route than before. +/// +/// The row must have at least one route. +pub(super) fn choose(row: &mut Row, peers: &[Option], scratch: &mut Scratch) -> bool { + let before = row.best; + let top = row + .paths + .iter() + .map(|path| path.preference()) + .max() + .unwrap_or(0); + + let candidates = &mut scratch.candidates; + candidates.clear(); + for (index, path) in row.paths.iter().enumerate() { + if path.preference() == top { + let key = path.attrs.key(); + // A route without a MED is taken to have the lowest (RFC 4271 + // §9.1.2.2 c). + candidates.push((key.neighbor_as, key.med.unwrap_or(0), index as u16)); + } + } + + // Step 4. MEDs are compared only between routes from the same + // neighbouring AS, so "lower MED" is not an order over all the routes, + // and comparing them pair by pair can go round in circles. Instead: + // within each neighbouring AS only the routes with its lowest MED stay + // in, and the steps after are an order over what is left. + candidates.sort_unstable(); + let mut best: Option<(TieBreak, u16)> = None; + let mut group = None; + for &(neighbor_as, med, index) in candidates.iter() { + // Sorted by AS and then MED: the first of each AS has its lowest. + let lowest = match group { + Some((group_as, lowest)) if group_as == neighbor_as => lowest, + _ => { + group = Some((neighbor_as, med)); + med + } + }; + if med != lowest { + continue; + } + let tie_break = TieBreak::of(row, index, peers); + if best.as_ref().is_none_or(|(least, _)| tie_break < *least) { + best = Some((tie_break, index)); + } + } + + row.best = best.map_or(0, |(_, index)| index); + row.best != before +} + +/// For a route just added at `added` that ties with the best in steps 1 to +/// 3: whether it is the new best, if that can be said from the two routes +/// alone. `None` if the whole row has to be looked at. +/// +/// It can be said when MED cannot knock either of them out. The new route +/// has the lowest MED there is, so nothing knocks it out. It knocks out +/// only routes of its own neighbouring AS with a higher MED, and if the +/// best is not one of those, the best is still the best of all the others. +/// So the better of the two is the best of all. This is the usual case: +/// most routes have no MED. +pub(super) fn settles_tie(row: &Row, added: usize, peers: &[Option]) -> Option { + let new = row.paths[added].attrs.key(); + let best = row.paths[usize::from(row.best)].attrs.key(); + if new.med.unwrap_or(0) != 0 { + return None; + } + if best.neighbor_as == new.neighbor_as && best.med.unwrap_or(0) != 0 { + return None; + } + Some(TieBreak::of(row, added as u16, peers) < TieBreak::of(row, row.best, peers)) +} + +/// Steps 5 to 7, and then the peer's number so that the choice never +/// depends on the order routes arrived in. The least wins. +#[derive(PartialEq, Eq, PartialOrd, Ord)] +struct TieBreak { + internal: bool, + router_id: u32, + address: std::net::IpAddr, + peer: u16, +} + +impl TieBreak { + fn of(row: &Row, index: u16, peers: &[Option]) -> Self { + let peer = row.paths[usize::from(index)].peer(); + match peers.get(peer.as_usize()).copied().flatten() { + Some(info) => Self { + internal: info.ibgp, + router_id: info.router_id.into(), + address: info.address, + peer: peer.0, + }, + // A peer on its way out; anything known is preferred to it. + None => Self { + internal: true, + router_id: u32::MAX, + address: std::net::Ipv6Addr::from(u128::MAX).into(), + peer: peer.0, + }, + } + } +} diff --git a/crates/rib/src/table/tests.rs b/crates/rib/src/table/tests.rs new file mode 100644 index 0000000..d26592b --- /dev/null +++ b/crates/rib/src/table/tests.rs @@ -0,0 +1,578 @@ +use std::collections::{BTreeMap, HashMap}; +use std::net::{IpAddr, Ipv4Addr}; +use std::sync::Arc; + +use bgp_packet::message::{ + AsPathAttribute, LocalPrefPathAttribute, MultiExitDiscPathAttribute, NextHopPathAttribute, + OriginPathAttribute, PathAttribute, +}; +use bgp_primitives::ip_prefix::IpPrefix; +use proptest::prelude::*; + +use super::*; + +fn prefix(s: &str) -> IpPrefix { + IpPrefix::try_from(s).unwrap() +} + +/// What a test says about a route; everything selection looks at. +#[derive(Debug, Clone, PartialEq, Eq)] +struct Route { + local_pref: Option, + path: Vec, + origin: u8, + med: Option, +} + +impl Route { + fn via(path: &[u32]) -> Self { + Self { + local_pref: None, + path: path.to_vec(), + origin: 0, + med: None, + } + } + + fn attributes(&self) -> Vec { + let mut attributes = vec![ + PathAttribute::Origin(match self.origin { + 0 => OriginPathAttribute::IGP, + 1 => OriginPathAttribute::EGP, + _ => OriginPathAttribute::INCOMPLETE, + }), + AsPathAttribute::from_asns(self.path.clone()), + PathAttribute::NextHop(NextHopPathAttribute(Ipv4Addr::new(192, 0, 2, 1))), + ]; + attributes.extend( + self.med + .map(|med| PathAttribute::MultiExitDisc(MultiExitDiscPathAttribute(med))), + ); + attributes.extend( + self.local_pref + .map(|lp| PathAttribute::LocalPref(LocalPrefPathAttribute(lp))), + ); + attributes + } +} + +fn external(number: u8) -> PeerInfo { + PeerInfo { + asn: 65000 + u32::from(number), + router_id: Ipv4Addr::new(10, 0, 0, number), + address: IpAddr::from(Ipv4Addr::new(192, 0, 2, number)), + ibgp: false, + } +} + +fn internal(number: u8) -> PeerInfo { + PeerInfo { + asn: 64999, + ibgp: true, + ..external(number) + } +} + +/// A table and the peers added to it, for tests that speak of peers by the +/// order they were added in. +struct Fixture { + table: RibTable, + peers: Vec, +} + +impl Fixture { + fn new(peers: &[PeerInfo]) -> Self { + // More than one shard, so that a test's prefixes are spread. + let table = RibTable::new(4); + let peers = peers.iter().map(|p| table.add_peer(*p).unwrap()).collect(); + Self { table, peers } + } + + fn announce(&self, peer: usize, prefix: &IpPrefix, route: &Route) -> Outcome { + let attrs = self + .table + .attrs() + .intern(&route.attributes(), None) + .unwrap(); + self.table.announce(self.peers[peer], &attrs, &[*prefix]) + } + + fn withdraw(&self, peer: usize, prefix: &IpPrefix) -> Outcome { + self.table.withdraw(self.peers[peer], &[*prefix]) + } + + /// Which peer (by order added) the best route for `prefix` is from. + fn best(&self, prefix: &IpPrefix) -> Option { + let (peer, _) = self.table.best(prefix)?; + self.peers.iter().position(|p| *p == peer) + } +} + +#[test] +fn each_step_of_selection_decides_when_the_ones_before_tie() { + let p = prefix("10.0.0.0/8"); + // Peer 0 has the lower identifier, so it wins whenever nothing else + // decides; each case makes peer 1's route better in one step. + let cases: [(&str, [PeerInfo; 2], Route, Route); 6] = [ + ( + "higher LOCAL_PREF, despite a longer path", + [internal(1), internal(2)], + Route { + local_pref: Some(100), + ..Route::via(&[1]) + }, + Route { + local_pref: Some(200), + ..Route::via(&[1, 2, 3]) + }, + ), + ( + "shorter AS path", + [external(1), external(2)], + Route::via(&[1, 2, 3]), + Route::via(&[4, 5]), + ), + ( + "lower ORIGIN", + [external(1), external(2)], + Route { + origin: 2, + ..Route::via(&[1, 2]) + }, + Route { + origin: 0, + ..Route::via(&[3, 4]) + }, + ), + ( + "lower MED from the same neighbouring AS", + [external(1), external(2)], + Route { + med: Some(20), + ..Route::via(&[7, 2]) + }, + Route { + med: Some(10), + ..Route::via(&[7, 3]) + }, + ), + ( + "no MED counts as the lowest", + [external(1), external(2)], + Route { + med: Some(1), + ..Route::via(&[7, 2]) + }, + Route::via(&[7, 3]), + ), + ( + "external over internal", + [internal(1), external(2)], + Route::via(&[7, 2]), + Route::via(&[8, 3]), + ), + ]; + for (case, peers, first, second) in cases { + // In both orders of arrival: the choice must not depend on it. + for order in [[0, 1], [1, 0]] { + let fixture = Fixture::new(&peers); + for peer in order { + fixture.announce(peer, &p, [&first, &second][peer]); + } + assert_eq!(fixture.best(&p), Some(1), "{case}, arriving {order:?}"); + } + } + + // And when nothing decides: the lowest identifier. + let fixture = Fixture::new(&[external(2), external(1)]); + fixture.announce(0, &p, &Route::via(&[1, 2])); + fixture.announce(1, &p, &Route::via(&[3, 4])); + assert_eq!(fixture.best(&p), Some(1)); +} + +/// MEDs are compared only within a neighbouring AS. Taken pair by pair +/// that is not an order: A beats B, B beats C, C beats A. Whatever the +/// order of arrival, the answer must be the same. +#[test] +fn med_is_compared_only_within_a_neighbouring_as_and_never_goes_in_circles() { + let p = prefix("10.0.0.0/8"); + let routes = [ + // From AS 1, MED 10, and the highest identifier. + Route { + med: Some(10), + ..Route::via(&[1, 9]) + }, + // From AS 2: its MED is not comparable with the others'. + Route { + med: Some(50), + ..Route::via(&[2, 9]) + }, + // From AS 1, MED 20: loses to the first on MED. + Route { + med: Some(20), + ..Route::via(&[1, 8]) + }, + ]; + // Identifiers: the third route's peer lowest, then the second's, then + // the first's. Without MED the third would win; it is knocked out by + // the first, and of the two left the second has the lower identifier. + let peers = [external(3), external(2), external(1)]; + for order in [ + [0, 1, 2], + [0, 2, 1], + [1, 0, 2], + [1, 2, 0], + [2, 0, 1], + [2, 1, 0], + ] { + let fixture = Fixture::new(&peers); + for peer in order { + fixture.announce(peer, &p, &routes[peer]); + } + assert_eq!(fixture.best(&p), Some(1), "arriving {order:?}"); + } + + // Take away the route that knocked the third out, and the third wins. + let fixture = Fixture::new(&peers); + for (peer, route) in routes.iter().enumerate() { + fixture.announce(peer, &p, route); + } + let outcome = fixture.withdraw(0, &p); + assert_eq!(outcome.best_changed, 1); + assert_eq!(fixture.best(&p), Some(2)); +} + +/// A route that is not the best can still decide which is: by its MED it +/// keeps another route of its neighbouring AS out. When it gets worse or +/// goes, that other route is back in the running. The shortcuts must not +/// skip this. +#[test] +fn a_route_that_was_keeping_another_out_matters_when_it_changes() { + let p = prefix("10.0.0.0/8"); + // Peer 0: from AS 1 with the lowest identifier, but MED 1. + // Peer 1: from AS 2. Best while peer 0's route is kept out. + // Peer 2: from AS 1 with MED 0, which keeps peer 0's route out. + let peers = [external(1), external(2), external(3)]; + let held_out = Route { + med: Some(1), + ..Route::via(&[1, 9]) + }; + let best = Route::via(&[2, 9]); + let holding_out = Route { + med: Some(0), + ..Route::via(&[1, 8]) + }; + let set_up = || { + let fixture = Fixture::new(&peers); + fixture.announce(0, &p, &held_out); + fixture.announce(1, &p, &best); + fixture.announce(2, &p, &holding_out); + assert_eq!(fixture.best(&p), Some(1)); + fixture + }; + + // It is replaced by a route with a longer path. + let fixture = set_up(); + let outcome = fixture.announce(2, &p, &Route::via(&[1, 8, 7])); + assert_eq!(fixture.best(&p), Some(0)); + assert_eq!(outcome.best_changed, 1); + + // It is withdrawn. + let fixture = set_up(); + let outcome = fixture.withdraw(2, &p); + assert_eq!(fixture.best(&p), Some(0)); + assert_eq!(outcome.best_changed, 1); + + // Its peer goes away. + let fixture = set_up(); + let outcome = fixture.table.remove_peer(fixture.peers[2]); + assert_eq!(fixture.best(&p), Some(0)); + assert_eq!(outcome.best_changed, 1); + + // Its MED gets worse than the one it was keeping out. + let fixture = set_up(); + fixture.announce( + 2, + &p, + &Route { + med: Some(5), + ..Route::via(&[1, 8]) + }, + ); + assert_eq!(fixture.best(&p), Some(0)); +} + +#[test] +fn what_a_change_did_is_reported() { + let fixture = Fixture::new(&[external(1), external(2)]); + let p = prefix("10.0.0.0/8"); + let long = Route::via(&[1, 2, 3]); + let short = Route::via(&[4]); + + let first = fixture.announce(0, &p, &long); + assert_eq!((first.added, first.best_changed), (1, 1)); + let worse = fixture.announce(1, &p, &Route::via(&[1, 2, 3, 4])); + assert_eq!((worse.added, worse.best_changed), (1, 0)); + // The same route again changes nothing. + let again = fixture.announce(0, &p, &long); + assert_eq!((again.replaced, again.best_changed), (1, 0)); + // The best route replaced by a better one is still a change of best. + let better = fixture.announce(0, &p, &short); + assert_eq!((better.replaced, better.best_changed), (1, 1)); + assert_eq!((fixture.table.len(), fixture.table.prefix_count()), (2, 1)); + + assert_eq!(fixture.withdraw(1, &p).best_changed, 0); + let last = fixture.withdraw(0, &p); + assert_eq!((last.removed, last.best_changed), (1, 1)); + assert_eq!(fixture.withdraw(0, &p), Outcome::default(), "nothing there"); + assert!(fixture.table.is_empty()); + assert_eq!(fixture.best(&p), None); +} + +#[test] +fn removing_a_peer_takes_its_routes_and_frees_its_number() { + let fixture = Fixture::new(&[external(1), external(2)]); + let prefixes: Vec = (0..200u32) + .map(|i| { + IpPrefix::new( + bgp_primitives::constants::AddressFamilyId::Ipv4, + (i << 8).to_be_bytes(), + 24, + ) + .unwrap() + }) + .collect(); + let attrs = |route: &Route| { + fixture + .table + .attrs() + .intern(&route.attributes(), None) + .unwrap() + }; + // Peer 0 is best everywhere; peer 1 has only the first hundred. + fixture + .table + .announce(fixture.peers[0], &attrs(&Route::via(&[1])), &prefixes); + fixture.table.announce( + fixture.peers[1], + &attrs(&Route::via(&[2, 3])), + &prefixes[..100], + ); + assert_eq!(fixture.table.len(), 300); + + let outcome = fixture.table.remove_peer(fixture.peers[0]); + assert_eq!(outcome.removed, 200); + assert_eq!(outcome.best_changed, 200); + assert_eq!( + (fixture.table.len(), fixture.table.prefix_count()), + (100, 100) + ); + assert_eq!(fixture.best(&prefixes[0]), Some(1)); + assert_eq!(fixture.best(&prefixes[150]), None); + + // Its routes are gone, so its attribute set can be dropped. + assert_eq!(fixture.table.attrs().sweep(), 1); + // And its number is given to the next peer. + assert_eq!(fixture.table.add_peer(external(3)), Ok(fixture.peers[0])); +} + +#[test] +fn readers_see_every_route_and_which_is_best() { + let fixture = Fixture::new(&[external(1), external(2)]); + let (a, b) = (prefix("10.0.0.0/8"), prefix("2001:db8::/32")); + fixture.announce(0, &a, &Route::via(&[1, 2])); + fixture.announce(1, &a, &Route::via(&[3])); + fixture.announce(0, &b, &Route::via(&[1, 2])); + + let mut best = Vec::new(); + fixture + .table + .for_each_best(|prefix, peer, _| best.push((*prefix, peer))); + best.sort_by_key(|(prefix, _)| prefix.to_string()); + assert_eq!(best, [(a, fixture.peers[1]), (b, fixture.peers[0])]); + + let mut all = 0; + let mut marked_best = 0; + fixture.table.for_each_route(|_, _, _, is_best| { + all += 1; + marked_best += usize::from(is_best); + }); + assert_eq!((all, marked_best), (3, 2)); + + let routes = fixture.table.routes(&a); + assert_eq!(routes.len(), 2); + assert!( + routes + .iter() + .any(|(peer, _, best)| *peer == fixture.peers[1] && *best) + ); + // Routes with the same attributes share one set. + let (_, shared, _) = &fixture.table.routes(&b)[0]; + assert!( + routes + .iter() + .any(|(_, attrs, _)| Arc::ptr_eq(attrs, shared)) + ); +} + +// --------------------------------------------------------------------------- +// The table against a model +// --------------------------------------------------------------------------- + +/// Route selection written the plain way: all routes of a prefix, step by +/// step, with no shortcuts. The table takes shortcuts and must agree. +fn chosen_plainly(routes: &BTreeMap, peers: &[PeerInfo]) -> Option { + let local_pref = |peer: usize, route: &Route| { + if peers[peer].ibgp { + route.local_pref.unwrap_or(DEFAULT_LOCAL_PREF) + } else { + DEFAULT_LOCAL_PREF + } + }; + let mut left: Vec<(usize, &Route)> = + routes.iter().map(|(peer, route)| (*peer, route)).collect(); + let most = left.iter().map(|(p, r)| local_pref(*p, r)).max()?; + left.retain(|(p, r)| local_pref(*p, r) == most); + let shortest = left.iter().map(|(_, r)| r.path.len()).min()?; + left.retain(|(_, r)| r.path.len() == shortest); + let lowest = left.iter().map(|(_, r)| r.origin).min()?; + left.retain(|(_, r)| r.origin == lowest); + + let neighbor = |route: &Route| route.path.first().copied().unwrap_or(0); + let med = |route: &Route| route.med.unwrap_or(0); + let mut lowest_med: HashMap = HashMap::new(); + for (_, route) in &left { + let entry = lowest_med.entry(neighbor(route)).or_insert(u32::MAX); + *entry = (*entry).min(med(route)); + } + left.retain(|(_, r)| med(r) == lowest_med[&neighbor(r)]); + + left.into_iter().map(|(peer, _)| peer).min_by_key(|&peer| { + ( + peers[peer].ibgp, + peers[peer].router_id, + peers[peer].address, + peer, + ) + }) +} + +#[derive(Debug, Clone)] +enum Op { + Announce { + peer: usize, + prefix: usize, + route: Route, + }, + Withdraw { + peer: usize, + prefix: usize, + }, + /// The session goes down and comes straight back. + Flap { + peer: usize, + }, +} + +const PEERS: usize = 6; +// Few prefixes, so that each gathers routes from most peers. +const PREFIXES: usize = 2; + +fn route() -> impl Strategy { + // Few values of everything, so that ties at every step are common: + // ties are where the shortcuts could be wrong. + ( + proptest::option::of(100u32..102), + proptest::collection::vec(1u32..3, 1..3), + 0u8..2, + proptest::option::of(0u32..3), + ) + .prop_map(|(local_pref, path, origin, med)| Route { + local_pref, + path, + origin, + med, + }) +} + +fn op() -> impl Strategy { + prop_oneof![ + 6 => (0..PEERS, 0..PREFIXES, route()) + .prop_map(|(peer, prefix, route)| Op::Announce { peer, prefix, route }), + 3 => (0..PEERS, 0..PREFIXES).prop_map(|(peer, prefix)| Op::Withdraw { peer, prefix }), + 1 => (0..PEERS).prop_map(|peer| Op::Flap { peer }), + ] +} + +proptest! { + #![proptest_config(ProptestConfig::with_cases(2000))] + + /// After any sequence of changes the table holds exactly the routes it + /// should, has chosen for every prefix what plain selection chooses, + /// and has reported a change of best route exactly when there was one. + #[test] + fn the_table_agrees_with_plain_selection(ops in proptest::collection::vec(op(), 1..80)) { + // Internal and external peers, and identifiers that are not in the + // order the peers are added in. + let infos: Vec = (0..PEERS) + .map(|i| { + let number = (PEERS - i) as u8; + if i % 3 == 0 { internal(number) } else { external(number) } + }) + .collect(); + let mut fixture = Fixture::new(&infos); + let prefixes: Vec = (0..PREFIXES) + .map(|i| prefix(&format!("10.{i}.0.0/16"))) + .collect(); + let mut model: Vec> = vec![BTreeMap::new(); PREFIXES]; + + for op in ops { + let before: Vec> = model + .iter() + .map(|routes| { + chosen_plainly(routes, &infos).map(|peer| (peer, routes[&peer].clone())) + }) + .collect(); + let outcome = match &op { + Op::Announce { peer, prefix, route } => { + model[*prefix].insert(*peer, route.clone()); + fixture.announce(*peer, &prefixes[*prefix], route) + } + Op::Withdraw { peer, prefix } => { + model[*prefix].remove(peer); + fixture.withdraw(*peer, &prefixes[*prefix]) + } + Op::Flap { peer } => { + for routes in &mut model { + routes.remove(peer); + } + let outcome = fixture.table.remove_peer(fixture.peers[*peer]); + fixture.peers[*peer] = fixture.table.add_peer(infos[*peer]).unwrap(); + outcome + } + }; + + let mut best_changes = 0; + for (i, routes) in model.iter().enumerate() { + let expected = chosen_plainly(routes, &infos); + prop_assert_eq!(fixture.best(&prefixes[i]), expected, "after {:?}", &op); + let now = expected.map(|peer| (peer, routes[&peer].clone())); + best_changes += usize::from(now != before[i]); + + let mut held: Vec = fixture + .table + .routes(&prefixes[i]) + .iter() + .map(|(peer, _, _)| fixture.peers.iter().position(|p| p == peer).unwrap()) + .collect(); + held.sort_unstable(); + prop_assert_eq!(held, routes.keys().copied().collect::>()); + } + prop_assert_eq!(outcome.best_changed, best_changes, "after {:?}", &op); + prop_assert_eq!( + fixture.table.len(), + model.iter().map(BTreeMap::len).sum::() + ); + } + } +}