diff --git a/crates/bench/benches/real_table.rs b/crates/bench/benches/real_table.rs index cb4d0dd..fba8768 100644 --- a/crates/bench/benches/real_table.rs +++ b/crates/bench/benches/real_table.rs @@ -10,14 +10,14 @@ use std::collections::HashSet; use std::net::{Ipv4Addr, SocketAddr}; use std::path::PathBuf; use std::sync::Arc; -use std::sync::atomic::{AtomicUsize, Ordering}; +use std::sync::atomic::{AtomicBool, 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::table::{Event, PeerIndex, PeerInfo, RibTable}; use bgp_rib::{AdjRibIn, PeerId, ReceivedUpdate, RibEntry}; /// Bytes allocated and not freed since `region` began; zero when memory @@ -115,6 +115,7 @@ fn main() { let threads = std::thread::available_parallelism().map_or(1, |n| n.get()); sharded_table(&dump, &peers, &parsed, 1); sharded_table(&dump, &peers, &parsed, threads); + followed_table(&dump, &peers, &parsed, threads); } /// What the table looks like, as far as it decides what a RIB costs. @@ -288,25 +289,7 @@ fn current_table( /// 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 updates = grouped(dump, peers, parsed); let routes: usize = updates .iter() .flatten() @@ -314,49 +297,10 @@ fn sharded_table(dump: &Dump, peers: &[usize], parsed: &[Option], th .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 table = Arc::new(RibTable::new(threads * 16)); + let indices = add_peers(&table, dump, peers); 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 best_changes = load(&table, dump, &updates, &indices, threads); let elapsed = started.elapsed(); assert_eq!(table.len(), routes); let held = retained(®ion); @@ -374,7 +318,7 @@ fn sharded_table(dump: &Dump, peers: &[usize], parsed: &[Option], th " {} prefixes, {} attribute sets, best route changed {} times", table.prefix_count(), table.attrs().len(), - best_changes.load(Ordering::Relaxed) + best_changes ); let started = Instant::now(); @@ -425,3 +369,183 @@ fn sharded_table(dump: &Dump, peers: &[usize], parsed: &[Option], th format!("{swept} dropped"), ); } + +/// A peer's routes grouped as UPDATEs: the attribute set and the prefixes +/// announced with it. +type Updates = Vec)>>; + +fn grouped(dump: &Dump, peers: &[usize], parsed: &[Option]) -> Updates { + 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() +} + +fn add_peers(table: &RibTable, dump: &Dump, peers: &[usize]) -> 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() +} + +/// Announces everything in `updates` on `threads` threads, each taking +/// whole peers. Returns how many times a best route changed. +fn load( + table: &RibTable, + dump: &Dump, + updates: &Updates, + indices: &[PeerIndex], + threads: usize, +) -> usize { + let next = AtomicUsize::new(0); + let best_changes = AtomicUsize::new(0); + 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); + }); + } + }); + best_changes.load(Ordering::Relaxed) +} + +/// The same load with someone following the best routes through the +/// journal: once reading as fast as it can while the table fills, once not +/// reading at all until the table is full. +fn followed_table(dump: &Dump, peers: &[usize], parsed: &[Option], threads: usize) { + println!("\n== The sharded table with a follower, {threads} thread(s)"); + let updates = grouped(dump, peers, parsed); + let routes: usize = updates + .iter() + .flatten() + .map(|(_, prefixes)| prefixes.len()) + .sum(); + let count = |counts: &mut (usize, usize, usize), event: Event<'_>| match event { + Event::Best { attrs, .. } => { + // Look at the route, as a follower would. + counts.0 += usize::from(attrs.key().as_path_len > 0); + } + Event::Withdrawn { .. } => counts.1 += 1, + Event::Reset | Event::Synced => counts.2 += 1, + }; + + // Reading while the table fills. + let table = Arc::new(RibTable::new(threads * 16)); + let indices = add_peers(&table, dump, peers); + let done = AtomicBool::new(false); + let started = Instant::now(); + let (best_changes, counts, longest, after_load) = std::thread::scope(|scope| { + let follower = scope.spawn(|| { + let mut reader = table.reader(); + let mut counts = (0, 0, 0); + let mut longest = 0; + while !done.load(Ordering::Acquire) { + reader.poll(4096, |event| count(&mut counts, event)); + longest = longest.max(table.journal_len()); + } + let finishing = Instant::now(); + while !reader + .poll(4096, |event| count(&mut counts, event)) + .caught_up + {} + (counts, longest, finishing.elapsed()) + }); + let best_changes = load(&table, dump, &updates, &indices, threads); + done.store(true, Ordering::Release); + let (counts, longest, after_load) = follower.join().expect("the follower runs"); + (best_changes, counts, longest, after_load) + }); + let elapsed = started.elapsed(); + assert_eq!(table.len(), routes); + row( + "load, follower reading throughout", + elapsed, + format!( + "{} routes; caught up {:.3} s after the last route", + per_second(routes, elapsed), + after_load.as_secs_f64() + ), + ); + println!( + " best route changed {best_changes} times; the follower was given {} events for it; \n journal at most {longest} entries for {} prefixes", + counts.0 + counts.1, + table.prefix_count() + ); + drop(table); + + // Not reading: the journal has to stay small on its own account. + let table = Arc::new(RibTable::new(threads * 16)); + let indices = add_peers(&table, dump, peers); + let mut reader = table.reader(); + let mut counts = (0, 0, 0); + reader.poll(usize::MAX, |event| count(&mut counts, event)); + let started = Instant::now(); + load(&table, dump, &updates, &indices, threads); + let elapsed = started.elapsed(); + row( + "load, follower not reading", + elapsed, + format!( + "{} routes; journal holds {} entries for {} prefixes", + per_second(routes, elapsed), + table.journal_len(), + table.prefix_count() + ), + ); + let started = Instant::now(); + let mut counts = (0, 0, 0); + while !reader + .poll(4096, |event| count(&mut counts, event)) + .caught_up + {} + let elapsed = started.elapsed(); + row( + "the follower then catches up", + elapsed, + format!( + "{} events, {} of them a reset", + counts.0 + counts.1, + counts.2 + ), + ); +} diff --git a/crates/rib/proptest-regressions/table/journal_tests.txt b/crates/rib/proptest-regressions/table/journal_tests.txt new file mode 100644 index 0000000..7ee2a27 --- /dev/null +++ b/crates/rib/proptest-regressions/table/journal_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 7107cf5580fe7bb95e138f1fead386d89273532f83aa1d11808aa9586aa9da63 # shrinks to ops = [Withdraw { peer: 0, prefix: 1 }, Withdraw { peer: 0, prefix: 1 }, Poll { reader: 2, budget: 2 }, Withdraw { peer: 0, prefix: 10 }, Poll { reader: 2, budget: 4 }, Announce { peer: 2, prefix: 7, route: Route { local_pref: None, path: [1, 2], origin: 0, med: Some(0) } }, Announce { peer: 3, prefix: 6, route: Route { local_pref: None, path: [1], origin: 0, med: None } }, Announce { peer: 3, prefix: 6, route: Route { local_pref: None, path: [1, 1], origin: 0, med: None } }, Withdraw { peer: 1, prefix: 5 }, Withdraw { peer: 0, prefix: 11 }, Withdraw { peer: 3, prefix: 6 }, Announce { peer: 0, prefix: 3, route: Route { local_pref: None, path: [1, 2], origin: 1, med: None } }, Poll { reader: 2, budget: 3 }, Withdraw { peer: 0, prefix: 8 }, Flap { peer: 3 }, Withdraw { peer: 1, prefix: 1 }, Announce { peer: 2, prefix: 5, route: Route { local_pref: None, path: [1], origin: 1, med: Some(0) } }, Announce { peer: 0, prefix: 1, route: Route { local_pref: None, path: [1, 1], origin: 1, med: None } }, Withdraw { peer: 0, prefix: 1 }, Withdraw { peer: 2, prefix: 8 }, Poll { reader: 1, budget: 2 }, Flap { peer: 1 }, Poll { reader: 2, budget: 2 }, Announce { peer: 0, prefix: 9, route: Route { local_pref: None, path: [1], origin: 0, med: Some(1) } }, Announce { peer: 2, prefix: 2, route: Route { local_pref: Some(100), path: [2], origin: 0, med: Some(1) } }, Poll { reader: 1, budget: 3 }, Withdraw { peer: 3, prefix: 1 }, Announce { peer: 1, prefix: 4, route: Route { local_pref: None, path: [2], origin: 1, med: None } }, Announce { peer: 3, prefix: 7, route: Route { local_pref: None, path: [2], origin: 0, med: None } }, Announce { peer: 1, prefix: 11, route: Route { local_pref: None, path: [1], origin: 1, med: Some(1) } }, Withdraw { peer: 0, prefix: 6 }, Withdraw { peer: 0, prefix: 4 }, Announce { peer: 3, prefix: 10, route: Route { local_pref: Some(100), path: [2, 2], origin: 1, med: None } }, Announce { peer: 1, prefix: 2, route: Route { local_pref: None, path: [2], origin: 0, med: Some(2) } }, Announce { peer: 1, prefix: 3, route: Route { local_pref: None, path: [1], origin: 1, med: None } }, Withdraw { peer: 0, prefix: 2 }, Announce { peer: 1, prefix: 4, route: Route { local_pref: None, path: [1, 1], origin: 0, med: None } }, Poll { reader: 1, budget: 2 }, Poll { reader: 1, budget: 4 }, Poll { reader: 2, budget: 1 }, Poll { reader: 1, budget: 3 }, Poll { reader: 1, budget: 5 }, Poll { reader: 1, budget: 1 }, Withdraw { peer: 3, prefix: 5 }, Withdraw { peer: 3, prefix: 9 }, Withdraw { peer: 2, prefix: 10 }], shards = 3 diff --git a/crates/rib/src/table/journal.rs b/crates/rib/src/table/journal.rs new file mode 100644 index 0000000..0dea3d6 --- /dev/null +++ b/crates/rib/src/table/journal.rs @@ -0,0 +1,554 @@ +//! The journal: how the rest of the router follows the best routes. +//! +//! Whatever acts on the table's choices (the peers that are sent the best +//! routes, the forwarding table, the status page) has to learn when a best +//! route changes, at its own pace, without the table waiting for it. Each +//! shard keeps a journal for that, and each follower reads it through a +//! [`Reader`]. +//! +//! # What the journal holds +//! +//! It records *that* a prefix's best route changed, not what it changed to. +//! A reader that comes to an entry looks the prefix up and is given the best +//! route as it is at that moment. A follower does not want the history; it +//! wants to end up with what the table has now. +//! +//! This is what keeps the journal small. If a prefix changes again before a +//! reader has got to its entry, the older entry no longer says anything: the +//! newer one will deliver the current route. Each row remembers the +//! sequence number of its latest change, an entry with any other number is +//! dead, and dead entries are thrown out when the journal is tidied. So a +//! prefix that flaps a thousand times costs a reader one event, and the +//! journal never has more live entries than the table has prefixes. +//! +//! # Readers that fall behind +//! +//! Entries are dropped once every reader has passed them. A reader that +//! stops reading would hold them all back, so there is a limit: when the +//! journal is too long even after tidying, its oldest half is dropped +//! regardless. A reader that was behind that point is told so the next time +//! it reads ([`Event::Reset`]) and is given the whole table again. Reading +//! too slowly costs the slow reader a resynchronisation; it costs the +//! table, and the other readers, nothing. +//! +//! A new reader starts the same way: `Reset`, every best route, `Synced`, +//! and changes from then on. +//! +//! # Order +//! +//! For any one prefix, a reader is given its states in the order they came +//! about, and always, in the end, the latest. Between prefixes there is no +//! order worth relying on. + +use std::collections::VecDeque; +use std::hash::BuildHasher; +use std::sync::Arc; +use std::sync::atomic::{AtomicU64, Ordering}; + +use bgp_primitives::ip_prefix::IpPrefix; +use foldhash::fast::RandomState; +use hashbrown::HashMap; + +use super::{AttrSet, PeerIndex, RibTable, Shard, read, write}; + +/// A cursor that holds nothing back: a reader slot that is free, or a +/// reader that has not read this shard yet. +const NO_CURSOR: u64 = u64::MAX; + +/// When a journal is tidied, and when it is too long. +#[derive(Debug, Clone, Copy)] +pub(super) struct Limits { + /// Below this many entries a journal is not worth tidying. + pub(super) tidy_at: usize, + /// Room allowed beyond the number of prefixes before the journal is + /// thought too long. + pub(super) slack: usize, +} + +impl Default for Limits { + fn default() -> Self { + Self { + tidy_at: 1024, + slack: 64, + } + } +} + +struct Entry { + seq: u64, + prefix: IpPrefix, +} + +/// One shard's journal. +pub(super) struct Journal { + /// The sequence number of the latest change. Counts every change of a + /// best route in the shard, whether or not anyone was reading. + seq: u64, + /// Entries up to and including this number were dropped without regard + /// to the readers. A reader whose cursor is below it missed something. + floor: u64, + /// In order of sequence number. Not every number is here: dead entries + /// are removed, and nothing is recorded while there are no readers. + entries: VecDeque, + /// Prefixes that have no route any more and whose disappearance is + /// still to be read, with the sequence number of the entry that says + /// so. A row keeps this number itself; a prefix without a row cannot. + gone: HashMap, + /// Where each reader has read to, by reader slot. Written by the + /// readers themselves while they hold the shard for reading. + cursors: Vec, + /// How many reader slots are in use. + readers: usize, + limits: Limits, +} + +impl Journal { + pub(super) fn new(limits: Limits) -> Self { + Self { + limits, + seq: 0, + floor: 0, + entries: VecDeque::new(), + gone: HashMap::default(), + cursors: Vec::new(), + readers: 0, + } + } + + /// Notes that the best route of `prefix`, which has a row, changed. + /// Returns the sequence number for the row to remember. + pub(super) fn best_changed(&mut self, prefix: IpPrefix) -> u64 { + self.seq += 1; + if self.readers > 0 { + self.entries.push_back(Entry { + seq: self.seq, + prefix, + }); + } + self.seq + } + + /// As [`Journal::best_changed`], for a prefix whose row was just made: + /// if it had gone and is back, the note that it had gone is void. + pub(super) fn appeared(&mut self, prefix: IpPrefix) -> u64 { + if !self.gone.is_empty() { + self.gone.remove(&prefix); + } + self.best_changed(prefix) + } + + /// Notes that `prefix` lost its last route and has no row any more. + pub(super) fn disappeared(&mut self, prefix: IpPrefix) { + self.seq += 1; + if self.readers > 0 { + self.entries.push_back(Entry { + seq: self.seq, + prefix, + }); + self.gone.insert(prefix, self.seq); + } + } + + fn drop_entry(gone: &mut HashMap, entry: &Entry) { + if gone.get(&entry.prefix) == Some(&entry.seq) { + gone.remove(&entry.prefix); + } + } + + pub(super) fn len(&self) -> usize { + self.entries.len() + } +} + +impl Shard { + /// Whether `entry` still says something: it is the latest change of + /// its prefix. + fn is_live(&self, entry: &Entry) -> bool { + let hash = self.hasher.hash_one(entry.prefix); + match self.rows.find(hash, |row| row.prefix == entry.prefix) { + Some(row) => row.changed == entry.seq, + None => self.journal.gone.get(&entry.prefix) == Some(&entry.seq), + } + } + + /// Keeps the journal from growing. Called after every change to the + /// shard; does nothing until there is something worth doing. + pub(super) fn tidy_journal(&mut self) { + if self.journal.entries.len() <= self.journal.limits.tidy_at { + return; + } + // What every reader has passed can go. + let journal = &mut self.journal; + let passed = journal + .cursors + .iter() + .map(|cursor| cursor.load(Ordering::Acquire)) + .min() + .unwrap_or(NO_CURSOR); + while let Some(front) = journal.entries.front() { + if front.seq > passed { + break; + } + Journal::drop_entry(&mut journal.gone, front); + journal.entries.pop_front(); + } + + let limit = 2 * (self.rows.len() + self.journal.limits.slack); + if self.journal.entries.len() <= limit { + return; + } + // Dead entries: their prefix has changed again since. This looks + // every entry up, but only once the journal has doubled, so the + // cost per change stays small. + let mut entries = std::mem::take(&mut self.journal.entries); + entries.retain(|entry| self.is_live(entry)); + self.journal.entries = entries; + + if self.journal.entries.len() <= limit { + return; + } + // Still too long: some reader is not reading, and prefixes keep + // coming and going. It loses its place. + let journal = &mut self.journal; + for entry in journal.entries.drain(..limit / 2) { + Journal::drop_entry(&mut journal.gone, &entry); + journal.floor = entry.seq; + } + } +} + +/// What a [`Reader`] is given. +#[derive(Debug, Clone, Copy)] +pub enum Event<'a> { + /// Forget what you know. The whole table follows, as [`Event::Best`] + /// for every prefix, and then [`Event::Synced`]. The first thing a new + /// reader is given, and what a reader that fell too far behind is + /// given in place of what it missed. + Reset, + /// This is now the best route for `prefix`. + Best { + prefix: IpPrefix, + peer: PeerIndex, + attrs: &'a Arc, + }, + /// `prefix` has no route any more. + Withdrawn { prefix: IpPrefix }, + /// The table that began with [`Event::Reset`] is complete: a prefix + /// not heard of since the reset has no route. + Synced, +} + +/// What a call to [`Reader::poll`] got done. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub struct Progress { + /// How many events were given out. + pub events: usize, + /// Whether the reader has now been given everything there was when the + /// call began. If not, call again. + pub caught_up: bool, +} + +enum Phase { + /// Giving out the whole table; `next` is the shard to do next. + Resync { next: usize, announced: bool }, + /// Following the journals; `next` is the shard to read next. + Follow { next: usize }, +} + +/// Follows the best routes of a table. Made by [`RibTable::reader`]. +/// +/// A reader holds its place in every shard's journal. Dropping it gives +/// the place up, so that the journals need not keep entries for it. +pub struct Reader { + table: Arc, + slot: usize, + /// Where this reader has read to in each shard; the copy in the shard + /// is for the shard's benefit. + cursors: Vec, + phase: Phase, + /// The table's version when the reader was last found caught up. + seen: u64, + caught_up: bool, +} + +impl RibTable { + /// A new follower of this table's best routes. It is given the whole + /// table first and every change after. + pub fn reader(self: &Arc) -> Reader { + let mut slots = self + .reader_slots + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner); + let slot = match slots.iter().position(|used| !used) { + Some(free) => free, + None => { + slots.push(false); + slots.len() - 1 + } + }; + slots[slot] = true; + for shard in &self.shards { + let journal = &mut write(shard).journal; + while journal.cursors.len() <= slot { + journal.cursors.push(AtomicU64::new(NO_CURSOR)); + } + journal.cursors[slot].store(NO_CURSOR, Ordering::Release); + journal.readers += 1; + } + self.readers.fetch_add(1, Ordering::AcqRel); + Reader { + table: Arc::clone(self), + slot, + cursors: vec![NO_CURSOR; self.shards.len()], + phase: Phase::Resync { + next: 0, + announced: false, + }, + seen: 0, + caught_up: false, + } + } + + /// How many entries the journals hold, over all shards. For tests and + /// for watching: it stays bounded whatever the readers do. + pub fn journal_len(&self) -> usize { + self.shards + .iter() + .map(|shard| read(shard).journal.len()) + .sum() + } + + /// Tells readers that are waiting that there is something to read. + pub(super) fn wake_readers(&self) { + if self.readers.load(Ordering::Acquire) > 0 { + self.version.fetch_add(1, Ordering::AcqRel); + self.wake.notify_waiters(); + } + } +} + +impl Drop for Reader { + fn drop(&mut self) { + for shard in &self.table.shards { + let journal = &mut write(shard).journal; + journal.cursors[self.slot].store(NO_CURSOR, Ordering::Release); + journal.readers -= 1; + if journal.readers == 0 { + // Nobody left to read it. + journal.entries.clear(); + journal.gone.clear(); + } + } + self.table.readers.fetch_sub(1, Ordering::AcqRel); + let mut slots = self + .table + .reader_slots + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner); + slots[self.slot] = false; + } +} + +enum Drained { + /// Entries were dropped that this reader had not read. + Lagged, + /// `events` were given out; `more` if the budget ran out first. + Read { events: usize, more: bool }, +} + +impl Reader { + /// Gives `visit` what has changed since the last call, in up to about + /// `budget` events (a shard's part of a resynchronisation is not cut + /// short, so it can be more). + /// + /// `visit` is called with a shard of the table held for reading. It + /// should be quick, and it must not change the table. + pub fn poll(&mut self, budget: usize, mut visit: impl FnMut(Event<'_>)) -> Progress { + let shards = self.table.shards.len(); + let mut events = 0; + loop { + match self.phase { + Phase::Resync { next, announced } => { + if !announced { + self.forget_place(); + visit(Event::Reset); + events += 1; + } + if next == shards { + visit(Event::Synced); + events += 1; + self.phase = Phase::Follow { next: 0 }; + continue; + } + events += self.scan(next, &mut visit); + self.phase = Phase::Resync { + next: next + 1, + announced: true, + }; + if events >= budget { + return self.progress(events, false); + } + } + Phase::Follow { next } => { + // Read before the pass: a change made during it shows as + // a newer version, and `changed` then returns at once. + let version = if next == 0 { + self.table.version.load(Ordering::Acquire) + } else { + self.seen + }; + self.seen = version; + let mut shard = next; + let mut lagged = false; + while shard < shards { + if events >= budget { + self.phase = Phase::Follow { next: shard }; + return self.progress(events, false); + } + match self.drain(shard, budget - events, &mut visit) { + Drained::Lagged => { + lagged = true; + break; + } + Drained::Read { events: read, more } => { + events += read; + if more { + self.phase = Phase::Follow { next: shard }; + return self.progress(events, false); + } + } + } + shard += 1; + } + if lagged { + self.phase = Phase::Resync { + next: 0, + announced: false, + }; + continue; + } + self.phase = Phase::Follow { next: 0 }; + if next != 0 { + // This pass took up where an earlier call stopped. + // The shards before that point have not been looked + // at in this call, so go round again from the first. + continue; + } + return self.progress(events, true); + } + } + } + } + + fn progress(&mut self, events: usize, caught_up: bool) -> Progress { + self.caught_up = caught_up; + Progress { events, caught_up } + } + + /// Waits until there may be something to read. Returns at once if the + /// last [`Reader::poll`] did not catch up. + pub async fn changed(&mut self) { + loop { + // Registered before the check, so that a change between the + // check and the wait is not missed. + let woken = self.table.wake.notified(); + tokio::pin!(woken); + woken.as_mut().enable(); + if !self.caught_up || self.table.version.load(Ordering::Acquire) != self.seen { + return; + } + woken.await; + } + } + + /// Gives up this reader's place in every journal, before a + /// resynchronisation takes a new one. + fn forget_place(&mut self) { + for (index, shard) in self.table.shards.iter().enumerate() { + read(shard).journal.cursors[self.slot].store(NO_CURSOR, Ordering::Release); + self.cursors[index] = NO_CURSOR; + } + } + + fn set_cursor(&mut self, index: usize, shard: &Shard, cursor: u64) { + self.cursors[index] = cursor; + shard.journal.cursors[self.slot].store(cursor, Ordering::Release); + } + + /// Gives out every best route in a shard and takes a place at the end + /// of its journal, in one hold of the shard: what changes afterwards is + /// in the journal after that place. + fn scan(&mut self, index: usize, visit: &mut impl FnMut(Event<'_>)) -> usize { + let table = Arc::clone(&self.table); + let shard = read(&table.shards[index]); + for row in &shard.rows { + let best = &row.paths[usize::from(row.best)]; + visit(Event::Best { + prefix: row.prefix, + peer: best.peer(), + attrs: &best.attrs, + }); + } + self.set_cursor(index, &shard, shard.journal.seq); + shard.rows.len() + } + + /// Reads a shard's journal from this reader's place, for at most `max` + /// events. + fn drain(&mut self, index: usize, max: usize, visit: &mut impl FnMut(Event<'_>)) -> Drained { + let table = Arc::clone(&self.table); + let shard = read(&table.shards[index]); + let journal = &shard.journal; + let mut cursor = self.cursors[index]; + if cursor < journal.floor { + return Drained::Lagged; + } + if cursor == journal.seq { + return Drained::Read { + events: 0, + more: false, + }; + } + + let start = journal.entries.partition_point(|entry| entry.seq <= cursor); + let mut events = 0; + let mut more = false; + for entry in journal.entries.range(start..) { + if events == max { + more = true; + break; + } + cursor = entry.seq; + let hash = shard.hasher.hash_one(entry.prefix); + match shard.rows.find(hash, |row| row.prefix == entry.prefix) { + Some(row) if row.changed == entry.seq => { + let best = &row.paths[usize::from(row.best)]; + visit(Event::Best { + prefix: row.prefix, + peer: best.peer(), + attrs: &best.attrs, + }); + events += 1; + } + None if journal.gone.get(&entry.prefix) == Some(&entry.seq) => { + visit(Event::Withdrawn { + prefix: entry.prefix, + }); + events += 1; + } + // The prefix has changed again; a later entry has it. + _ => {} + } + } + if !more { + // Past the last entry there is nothing up to the latest change. + cursor = journal.seq; + } + self.set_cursor(index, &shard, cursor); + Drained::Read { events, more } + } +} + +#[cfg(test)] +#[path = "journal_tests.rs"] +mod tests; diff --git a/crates/rib/src/table/journal_tests.rs b/crates/rib/src/table/journal_tests.rs new file mode 100644 index 0000000..0c1013a --- /dev/null +++ b/crates/rib/src/table/journal_tests.rs @@ -0,0 +1,466 @@ +use std::collections::HashMap; +use std::sync::Arc; +use std::sync::atomic::{AtomicBool, Ordering}; + +use bgp_primitives::constants::AddressFamilyId; +use bgp_primitives::ip_prefix::IpPrefix; +use proptest::prelude::*; + +use super::super::tests::{Fixture, Route, external, internal, prefix, route}; +use super::super::{AttrSet, PeerIndex, PeerInfo, RibTable}; +use super::{Event, Limits, Reader}; + +/// What a follower would keep: its own copy of the best routes, brought +/// up to date from the events it is given. +#[derive(Default)] +struct Mirror { + best: HashMap)>, + /// What was held before a reset and has not been heard of since. + stale: HashMap)>, + resets: usize, + synced: usize, + best_events: usize, + withdrawn_events: usize, +} + +impl Mirror { + fn apply(&mut self, event: Event<'_>) { + match event { + Event::Reset => { + self.resets += 1; + self.stale = std::mem::take(&mut self.best); + } + Event::Best { + prefix, + peer, + attrs, + } => { + self.best_events += 1; + self.stale.remove(&prefix); + self.best.insert(prefix, (peer, Arc::clone(attrs))); + } + Event::Withdrawn { prefix } => { + self.withdrawn_events += 1; + self.best.remove(&prefix); + } + Event::Synced => { + self.synced += 1; + self.stale.clear(); + } + } + } + + /// Reads until there is nothing more, in steps of `budget`. + fn catch_up(&mut self, reader: &mut Reader, budget: usize) { + while !reader.poll(budget, |event| self.apply(event)).caught_up {} + } + + #[track_caller] + fn assert_matches(&self, table: &RibTable) { + let mut expected = HashMap::new(); + table.for_each_best(|prefix, peer, attrs| { + expected.insert(*prefix, (peer, Arc::clone(attrs))); + }); + assert_eq!( + self.best.len(), + expected.len(), + "the mirror has {} prefixes, the table {}", + self.best.len(), + expected.len() + ); + for (prefix, (peer, attrs)) in &expected { + let Some((mirrored_peer, mirrored_attrs)) = self.best.get(prefix) else { + panic!("{prefix} is in the table and not in the mirror"); + }; + assert_eq!(mirrored_peer, peer, "{prefix}: from the wrong peer"); + assert!( + Arc::ptr_eq(mirrored_attrs, attrs), + "{prefix}: wrong attributes" + ); + } + } +} + +fn nth_prefix(n: u32) -> IpPrefix { + IpPrefix::new(AddressFamilyId::Ipv4, (n << 8).to_be_bytes(), 24).unwrap() +} + +#[test] +fn a_new_reader_is_given_the_table_and_then_the_changes() { + 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(0, &b, &Route::via(&[1, 2])); + + let mut reader = fixture.table.reader(); + let mut mirror = Mirror::default(); + let mut events = Vec::new(); + let progress = reader.poll(100, |event| { + events.push( + format!("{event:?}") + .split([' ', '{']) + .next() + .unwrap() + .to_string(), + ); + mirror.apply(event); + }); + assert!(progress.caught_up); + assert_eq!(events, ["Reset", "Best", "Best", "Synced"]); + mirror.assert_matches(&fixture.table); + + // Nothing has happened: nothing to read. + assert_eq!(reader.poll(100, |_| panic!("no event expected")).events, 0); + + // A change that does not change the best route is not an event. + fixture.announce(1, &a, &Route::via(&[3, 4, 5])); + assert_eq!(reader.poll(100, |_| panic!("no event expected")).events, 0); + + fixture.announce(1, &a, &Route::via(&[3])); + fixture.withdraw(0, &b); + mirror.catch_up(&mut reader, 100); + assert_eq!((mirror.best_events, mirror.withdrawn_events), (3, 1)); + assert_eq!((mirror.resets, mirror.synced), (1, 1)); + mirror.assert_matches(&fixture.table); +} + +/// A follower wants what the table has now, not how it got there. +#[test] +fn many_changes_to_a_prefix_between_reads_are_one_event() { + let fixture = Fixture::new(&[external(1), external(2)]); + let p = prefix("10.0.0.0/8"); + let mut reader = fixture.table.reader(); + let mut mirror = Mirror::default(); + mirror.catch_up(&mut reader, 100); + + for round in 0..50 { + fixture.announce(0, &p, &Route::via(&[1, 2, 3])); + fixture.announce(1, &p, &Route::via(&[4, 5])); + fixture.withdraw(1, &p); + if round % 2 == 0 { + fixture.withdraw(0, &p); + } + } + // It ended with peer 0's route in place. + mirror.catch_up(&mut reader, 100); + assert_eq!((mirror.best_events, mirror.withdrawn_events), (1, 0)); + mirror.assert_matches(&fixture.table); + + // And a prefix that came and went again is, at most, a withdrawal. + let q = prefix("10.1.0.0/16"); + for _ in 0..50 { + fixture.announce(0, &q, &Route::via(&[1])); + fixture.withdraw(0, &q); + } + mirror.catch_up(&mut reader, 100); + assert_eq!((mirror.best_events, mirror.withdrawn_events), (1, 1)); + mirror.assert_matches(&fixture.table); +} + +#[test] +fn reading_in_small_steps_gives_the_same_as_reading_at_once() { + let fixture = Fixture::new(&[external(1)]); + for n in 0..500 { + fixture.announce(0, &nth_prefix(n), &Route::via(&[1, n % 7 + 1])); + } + let mut reader = fixture.table.reader(); + let mut mirror = Mirror::default(); + mirror.catch_up(&mut reader, 7); + mirror.assert_matches(&fixture.table); + + for n in 0..500 { + if n % 3 == 0 { + fixture.withdraw(0, &nth_prefix(n)); + } else { + fixture.announce(0, &nth_prefix(n), &Route::via(&[2, n % 5 + 1])); + } + } + let mut steps = 0; + while !reader.poll(7, |event| mirror.apply(event)).caught_up { + steps += 1; + } + assert!(steps > 50, "the budget was kept to: {steps} steps"); + mirror.assert_matches(&fixture.table); + assert_eq!(mirror.resets, 1); +} + +#[test] +fn with_no_reader_nothing_is_kept() { + let fixture = Fixture::new(&[external(1)]); + for n in 0..5000 { + fixture.announce(0, &nth_prefix(n), &Route::via(&[1])); + fixture.withdraw(0, &nth_prefix(n)); + } + assert_eq!(fixture.table.journal_len(), 0); + + // And what a reader was holding back goes when the reader does. + let reader = fixture.table.reader(); + for n in 0..100 { + fixture.announce(0, &nth_prefix(n), &Route::via(&[1])); + } + assert_eq!(fixture.table.journal_len(), 100); + drop(reader); + assert_eq!(fixture.table.journal_len(), 0); +} + +/// A reader that stops reading must not make the table keep a growing +/// journal for it. It is the one that pays: it is given the table again. +#[test] +fn a_reader_that_stops_reading_is_reset_and_holds_nothing_up() { + // One shard, so that the journal's length is that shard's. + let fixture = Fixture::on(RibTable::new(1), &[external(1)]); + fixture.announce(0, &prefix("192.0.2.0/24"), &Route::via(&[1])); + + let mut stuck = fixture.table.reader(); + let mut stuck_mirror = Mirror::default(); + stuck_mirror.catch_up(&mut stuck, 100); + let mut diligent = fixture.table.reader(); + let mut diligent_mirror = Mirror::default(); + + // Prefixes come and go, each a different one, so no entry is made + // dead by a later one for the same prefix. + let limits = Limits::default(); + let mut longest = 0; + for n in 0..20_000 { + fixture.announce(0, &nth_prefix(n), &Route::via(&[1])); + fixture.withdraw(0, &nth_prefix(n)); + longest = longest.max(fixture.table.journal_len()); + if n % 100 == 0 { + diligent_mirror.catch_up(&mut diligent, 1000); + } + } + assert!( + longest <= limits.tidy_at.max(2 * (1 + limits.slack)) + 2, + "the journal reached {longest} entries" + ); + + // The one that kept up never noticed. + diligent_mirror.catch_up(&mut diligent, 1000); + assert_eq!( + diligent_mirror.resets, 1, + "only the reset every reader starts with" + ); + diligent_mirror.assert_matches(&fixture.table); + + // The one that did not is told, and ends up right all the same. + stuck_mirror.catch_up(&mut stuck, 1000); + assert_eq!(stuck_mirror.resets, 2); + assert_eq!(stuck_mirror.synced, 2); + stuck_mirror.assert_matches(&fixture.table); +} + +#[tokio::test] +async fn a_waiting_reader_is_woken_by_a_change_of_best_route() { + let fixture = Fixture::new(&[external(1), external(2)]); + let p = prefix("10.0.0.0/8"); + let mut reader = fixture.table.reader(); + let mut mirror = Mirror::default(); + + // Not caught up yet: no waiting. + reader.changed().await; + mirror.catch_up(&mut reader, 100); + + let waiting = tokio::time::timeout(std::time::Duration::from_millis(50), reader.changed()); + assert!(waiting.await.is_err(), "nothing changed, so it waits"); + + let table = Arc::clone(&fixture.table); + let attrs = table + .attrs() + .intern(&Route::via(&[1]).attributes(), None) + .unwrap(); + let peer = fixture.peers[0]; + let writer = tokio::spawn(async move { + tokio::time::sleep(std::time::Duration::from_millis(20)).await; + table.announce(peer, &attrs, &[p]); + }); + tokio::time::timeout(std::time::Duration::from_secs(5), reader.changed()) + .await + .expect("woken by the announcement"); + writer.await.unwrap(); + mirror.catch_up(&mut reader, 100); + mirror.assert_matches(&fixture.table); + + // A change made before the wait began is not missed either. + fixture.withdraw(0, &p); + tokio::time::timeout(std::time::Duration::from_secs(5), reader.changed()) + .await + .expect("the change was already there"); +} + +/// Writers on several threads, a reader on another, reading while they +/// write. When the writers are done and the reader has caught up, it has +/// what the table has. +#[test] +fn a_reader_keeps_up_with_writers_on_other_threads() { + const WRITERS: usize = 4; + const PREFIXES: u32 = 300; + let infos: Vec = (0..WRITERS).map(|i| external(i as u8 + 1)).collect(); + // Small journals, so that tidying and falling behind happen too. + let fixture = Fixture::on( + RibTable::with_journal_limits( + 8, + Limits { + tidy_at: 16, + slack: 4, + }, + ), + &infos, + ); + let table = &fixture.table; + let done = AtomicBool::new(false); + let mut mirror = Mirror::default(); + + std::thread::scope(|scope| { + let writers: Vec<_> = (0..WRITERS) + .map(|writer| { + let peer = fixture.peers[writer]; + scope.spawn(move || { + // A simple generator; each writer its own sequence. + let mut state = 0x9e37_79b9_7f4a_7c15u64.wrapping_mul(writer as u64 + 1); + let mut next = || { + state ^= state << 13; + state ^= state >> 7; + state ^= state << 17; + state + }; + for _ in 0..30_000 { + let p = nth_prefix((next() % u64::from(PREFIXES)) as u32); + if next() % 3 == 0 { + table.withdraw(peer, &[p]); + } else { + let route = Route { + med: Some((next() % 3) as u32), + ..Route::via( + &[(next() % 3) as u32 + 1, 9][..(next() % 2 + 1) as usize], + ) + }; + let attrs = table.attrs().intern(&route.attributes(), None).unwrap(); + table.announce(peer, &attrs, &[p]); + } + } + }) + }) + .collect(); + + let mirror = &mut mirror; + let done = &done; + let reading = scope.spawn(move || { + let mut reader = table.reader(); + while !done.load(Ordering::Acquire) { + reader.poll(50, |event| mirror.apply(event)); + } + mirror.catch_up(&mut reader, 50); + }); + for writer in writers { + writer.join().unwrap(); + } + done.store(true, Ordering::Release); + reading.join().unwrap(); + }); + + mirror.assert_matches(table); + assert!(mirror.best_events > 0); +} + +// --------------------------------------------------------------------------- +// Readers against the table, under any interleaving +// --------------------------------------------------------------------------- + +#[derive(Debug, Clone)] +enum Op { + Announce { + peer: usize, + prefix: usize, + route: Route, + }, + Withdraw { + peer: usize, + prefix: usize, + }, + Flap { + peer: usize, + }, + Poll { + reader: usize, + budget: usize, + }, + /// The reader is dropped and a new one takes its place. + Replace { + reader: usize, + }, +} + +const PEERS: usize = 4; +const PREFIXES: usize = 12; +const READERS: usize = 3; + +fn op() -> impl Strategy { + prop_oneof![ + 8 => (0..PEERS, 0..PREFIXES, route()) + .prop_map(|(peer, prefix, route)| Op::Announce { peer, prefix, route }), + 5 => (0..PEERS, 0..PREFIXES).prop_map(|(peer, prefix)| Op::Withdraw { peer, prefix }), + 1 => (0..PEERS).prop_map(|peer| Op::Flap { peer }), + // Reader 0 is polled rarely: it is the one that falls behind. + 4 => (1..READERS, 1usize..6).prop_map(|(reader, budget)| Op::Poll { reader, budget }), + 1 => (0..READERS, 1usize..6).prop_map(|(reader, budget)| Op::Poll { reader, budget }), + 1 => (1..READERS).prop_map(|reader| Op::Replace { reader }), + ] +} + +proptest! { + #![proptest_config(ProptestConfig::with_cases(1500))] + + /// However changes and reads are interleaved, and however small the + /// journals: a reader that has caught up holds exactly the table's best + /// routes, and every reader gets there in the end. + #[test] + fn a_reader_that_has_caught_up_has_what_the_table_has( + ops in proptest::collection::vec(op(), 1..150), + shards in 1usize..4, + ) { + let infos: Vec = (0..PEERS) + .map(|i| if i == 0 { internal(9) } else { external(i as u8) }) + .collect(); + // Journals that are tidied, and found too long, almost at once. + let limits = Limits { tidy_at: 3, slack: 1 }; + let mut fixture = Fixture::on(RibTable::with_journal_limits(shards, limits), &infos); + let prefixes: Vec = (0..PREFIXES as u32).map(nth_prefix).collect(); + let mut readers: Vec<(Reader, Mirror)> = (0..READERS) + .map(|_| (fixture.table.reader(), Mirror::default())) + .collect(); + + for op in ops { + match op { + Op::Announce { peer, prefix, route } => { + fixture.announce(peer, &prefixes[prefix], &route); + } + Op::Withdraw { peer, prefix } => { + fixture.withdraw(peer, &prefixes[prefix]); + } + Op::Flap { peer } => { + fixture.table.remove_peer(fixture.peers[peer]); + fixture.peers[peer] = fixture.table.add_peer(infos[peer]).unwrap(); + } + Op::Poll { reader, budget } => { + let (reader, mirror) = &mut readers[reader]; + let progress = reader.poll(budget, |event| mirror.apply(event)); + if progress.caught_up { + mirror.assert_matches(&fixture.table); + } + } + Op::Replace { reader } => { + readers[reader] = (fixture.table.reader(), Mirror::default()); + } + } + // Never much longer than the table, whatever the readers do. + let bound = shards * (2 * (PREFIXES + limits.slack) + limits.tidy_at + PREFIXES); + prop_assert!(fixture.table.journal_len() <= bound); + } + + for (reader, mirror) in &mut readers { + mirror.catch_up(reader, 5); + mirror.assert_matches(&fixture.table); + prop_assert_eq!(mirror.resets, mirror.synced); + } + } +} diff --git a/crates/rib/src/table/mod.rs b/crates/rib/src/table/mod.rs index b1ece43..30c0f71 100644 --- a/crates/rib/src/table/mod.rs +++ b/crates/rib/src/table/mod.rs @@ -21,22 +21,28 @@ //! 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. +//! - **Followers go at their own pace.** What acts on the best routes reads +//! the changes from a [`journal`], and a slow reader holds nothing up. pub mod attrs; +mod journal; mod select; #[cfg(test)] mod tests; use std::hash::BuildHasher; use std::net::{IpAddr, Ipv4Addr}; +use std::sync::atomic::{AtomicU64, AtomicUsize}; 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; +use tokio::sync::Notify; pub use attrs::{AttrSet, AttrStore, SelectionKey}; +pub use journal::{Event, Progress, Reader}; /// What route selection needs to know about a peer. #[derive(Debug, Clone, Copy, PartialEq, Eq)] @@ -118,6 +124,9 @@ struct Row { /// Index of the best route in `paths`. A row always has at least one /// route, and so always a best one. best: u16, + /// The journal's sequence number for the latest change of this row's + /// best route; see [`journal`]. + changed: u64, /// 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. @@ -139,6 +148,8 @@ struct Shard { hasher: RandomState, /// Reused between selections. scratch: select::Scratch, + /// Which prefixes' best routes have changed, for those who follow. + journal: journal::Journal, } /// The routing table. Shared between threads by reference; every method @@ -150,6 +161,13 @@ pub struct RibTable { /// Which peer numbers are in use. Changing the peer list also takes /// every shard's lock in turn; this serialises that. peers: Mutex>>, + /// Which reader slots are in use; see [`journal`]. + reader_slots: Mutex>, + /// How many readers there are, to skip waking them when there are none. + readers: AtomicUsize, + /// Goes up whenever a best route changes while there are readers. + version: AtomicU64, + wake: Notify, } /// No more peers can be added: every [`PeerIndex`] is in use. @@ -161,6 +179,12 @@ 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 { + Self::with_journal_limits(shards, journal::Limits::default()) + } + + /// A table whose journals are tidied, and found too long, at the given + /// sizes. Tests use small ones to get there quickly. + fn with_journal_limits(shards: usize, limits: journal::Limits) -> Self { let hasher = RandomState::default(); Self { shards: (0..shards.max(1)) @@ -171,12 +195,17 @@ impl RibTable { peers: Vec::new(), hasher: hasher.clone(), scratch: select::Scratch::default(), + journal: journal::Journal::new(limits), }) }) .collect(), hasher, attrs: AttrStore::new(), peers: Mutex::default(), + reader_slots: Mutex::default(), + readers: AtomicUsize::new(0), + version: AtomicU64::new(0), + wake: Notify::new(), } } @@ -192,7 +221,7 @@ impl RibTable { let index = match peers.iter().position(Option::is_none) { Some(free) => free, None => { - // u16::MAX is kept back: see `select::NONE`. + // A peer number shares a word with other things; see `Path`. if peers.len() >= usize::from(u16::MAX) { return Err(TooManyPeers); } @@ -220,6 +249,7 @@ impl RibTable { for shard in &self.shards { let mut shard = write(shard); outcome += shard.remove_peer(peer); + shard.tidy_journal(); if let Some(slot) = shard.peers.get_mut(peer.as_usize()) { *slot = None; } @@ -227,6 +257,9 @@ impl RibTable { if let Some(slot) = peers.get_mut(peer.as_usize()) { *slot = None; } + if outcome.best_changed > 0 { + self.wake_readers(); + } outcome } @@ -252,6 +285,9 @@ impl RibTable { } } }); + if outcome.best_changed > 0 { + self.wake_readers(); + } outcome } @@ -266,6 +302,9 @@ impl RibTable { } } }); + if outcome.best_changed > 0 { + self.wake_readers(); + } outcome } @@ -296,6 +335,7 @@ impl RibTable { 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); + shard.tidy_journal(); } } @@ -421,6 +461,7 @@ impl Shard { vacant.insert(Row { prefix, best: 0, + changed: self.journal.appeared(prefix), paths: vec![path], }); self.routes += 1; @@ -456,6 +497,9 @@ impl Shard { } else { select::choose(row, &self.peers, &mut self.scratch) }; + if best_changed { + row.changed = self.journal.best_changed(prefix); + } Outcome { added: 1, best_changed: best_changed.into(), @@ -484,6 +528,9 @@ impl Shard { } else { select::choose(row, &self.peers, &mut self.scratch) }; + if best_changed { + row.changed = self.journal.best_changed(prefix); + } Outcome { replaced: 1, best_changed: best_changed.into(), @@ -504,6 +551,7 @@ impl Shard { self.routes -= 1; if row.paths.len() == 1 { entry.remove(); + self.journal.disappeared(*prefix); return Outcome { removed: 1, best_changed: 1, @@ -511,6 +559,9 @@ impl Shard { }; } let best_changed = Self::remove_path(row, at, &self.peers, &mut self.scratch); + if best_changed { + row.changed = self.journal.best_changed(*prefix); + } Outcome { removed: 1, best_changed: best_changed.into(), @@ -551,6 +602,7 @@ impl Shard { let mut outcome = Outcome::default(); let peers = &self.peers; let scratch = &mut self.scratch; + let journal = &mut self.journal; self.rows.retain(|row| { let Some(at) = row.position(peer) else { return true; @@ -558,10 +610,12 @@ impl Shard { outcome.removed += 1; if row.paths.len() == 1 { outcome.best_changed += 1; + journal.disappeared(row.prefix); return false; } if Self::remove_path(row, at, peers, scratch) { outcome.best_changed += 1; + row.changed = journal.best_changed(row.prefix); } true }); diff --git a/crates/rib/src/table/tests.rs b/crates/rib/src/table/tests.rs index d26592b..7379c44 100644 --- a/crates/rib/src/table/tests.rs +++ b/crates/rib/src/table/tests.rs @@ -11,21 +11,21 @@ use proptest::prelude::*; use super::*; -fn prefix(s: &str) -> IpPrefix { +pub(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, +pub(super) struct Route { + pub(super) local_pref: Option, + pub(super) path: Vec, + pub(super) origin: u8, + pub(super) med: Option, } impl Route { - fn via(path: &[u32]) -> Self { + pub(super) fn via(path: &[u32]) -> Self { Self { local_pref: None, path: path.to_vec(), @@ -34,7 +34,7 @@ impl Route { } } - fn attributes(&self) -> Vec { + pub(super) fn attributes(&self) -> Vec { let mut attributes = vec![ PathAttribute::Origin(match self.origin { 0 => OriginPathAttribute::IGP, @@ -56,7 +56,7 @@ impl Route { } } -fn external(number: u8) -> PeerInfo { +pub(super) fn external(number: u8) -> PeerInfo { PeerInfo { asn: 65000 + u32::from(number), router_id: Ipv4Addr::new(10, 0, 0, number), @@ -65,7 +65,7 @@ fn external(number: u8) -> PeerInfo { } } -fn internal(number: u8) -> PeerInfo { +pub(super) fn internal(number: u8) -> PeerInfo { PeerInfo { asn: 64999, ibgp: true, @@ -75,20 +75,24 @@ fn internal(number: u8) -> PeerInfo { /// 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, +pub(super) struct Fixture { + pub(super) table: Arc, + pub(super) peers: Vec, } impl Fixture { - fn new(peers: &[PeerInfo]) -> Self { + pub(super) fn new(peers: &[PeerInfo]) -> Self { // More than one shard, so that a test's prefixes are spread. - let table = RibTable::new(4); + Self::on(RibTable::new(4), peers) + } + + pub(super) fn on(table: RibTable, peers: &[PeerInfo]) -> Self { + let table = Arc::new(table); 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 { + pub(super) fn announce(&self, peer: usize, prefix: &IpPrefix, route: &Route) -> Outcome { let attrs = self .table .attrs() @@ -97,7 +101,7 @@ impl Fixture { self.table.announce(self.peers[peer], &attrs, &[*prefix]) } - fn withdraw(&self, peer: usize, prefix: &IpPrefix) -> Outcome { + pub(super) fn withdraw(&self, peer: usize, prefix: &IpPrefix) -> Outcome { self.table.withdraw(self.peers[peer], &[*prefix]) } @@ -478,7 +482,7 @@ const PEERS: usize = 6; // Few prefixes, so that each gathers routes from most peers. const PREFIXES: usize = 2; -fn route() -> impl Strategy { +pub(super) fn route() -> impl Strategy { // Few values of everything, so that ties at every step are common: // ties are where the shortcuts could be wrong. (