From bcfd6e013f8a34d47f2c9c114cdc4588cb48724b Mon Sep 17 00:00:00 2001 From: Rayhaan Jaufeerally Date: Sat, 10 Oct 2026 22:44:32 +0200 Subject: [PATCH] rib: the lock-free number in a slot of its own, and the reader in named steps A code-quality pass over the table. The latest sequence number of each journal, which readers compare their cursors with without the shard's lock, was an atomic inside the journal under the lock with the table holding the other end of an Arc to it, stored on every route; it is now in a slot beside each shard's lock, `ShardSlot`, stored once a shard a change while the write guard is still held and before the readers are woken. "One of each kind of journal" is one type, `ByKind`, where three hand-written dispatchers and an array index were. `Reader::poll` is the loop skeleton over `resync_step` and `follow_pass`; the journal's `tidy` is three named stages; `live`'s nested Option is an enum that says what an entry says; the place of a cursor in a journal is found by one function for both the time chosen for a pass and the pass itself. The attribute set's four byte-readers share one walk over the raw attributes, with the next hop decoded once for both of its readers. `best` returns a `Route` like the other lookups. Docs that described an earlier design (the reader's `seen`, "every method takes &self", a `Routes::is_empty` that could not be true) are put right. Testing: new tests pin the lock-free skip, which nothing checked directly before: the number equals the journal's after every kind of change, with and without readers of each kind; a shard passed over is read again once it has news, from a reader of the whole table and from one that starts now. The OTC flag and the received-attributes interning gain tests. The crate's tests, both proptests included, the server tests, the integration tests, the kernel tests and every lab phase pass. Co-Authored-By: Claude Fable 5.1 --- crates/rib/src/table/attrs.rs | 146 +++++--- crates/rib/src/table/journal.rs | 488 ++++++++++++++------------ crates/rib/src/table/journal_tests.rs | 111 +++++- crates/rib/src/table/mod.rs | 71 ++-- crates/rib/src/table/tests.rs | 2 +- crates/server/src/reach.rs | 2 +- 6 files changed, 509 insertions(+), 311 deletions(-) diff --git a/crates/rib/src/table/attrs.rs b/crates/rib/src/table/attrs.rs index c5b85f3..f0d2be6 100644 --- a/crates/rib/src/table/attrs.rs +++ b/crates/rib/src/table/attrs.rs @@ -13,7 +13,7 @@ use std::hash::BuildHasher; use std::sync::{Arc, Mutex, MutexGuard, PoisonError}; -use bgp_packet::message::{NlriNextHop, PathAttribute, attr_type, split_attribute}; +use bgp_packet::message::{NlriNextHop, PathAttribute, RawAttribute, attr_type, split_attribute}; use bgp_primitives::parser::{EncodeError, ParserContext}; use bytes::{BufMut, BytesMut}; use foldhash::fast::RandomState; @@ -21,6 +21,10 @@ use hashbrown::HashTable; const OPTIONAL: u8 = 0x80; +/// The type of an AS_PATH segment whose AS numbers are in order; the other +/// type, AS_SET, is an unordered handful (RFC 4271 §4.3). +const AS_SEQUENCE: u8 = 2; + /// What route selection reads from a set of attributes (RFC 4271 §9.1.2.2). #[derive(Debug, Clone, Copy, PartialEq, Eq)] pub struct SelectionKey { @@ -116,7 +120,6 @@ impl SelectionKey { /// Reads an AS_PATH value with four-octet AS numbers. fn scan_as_path(mut segments: &[u8]) -> Option { - const AS_SEQUENCE: u8 = 2; let mut hops = 0usize; let (mut first, mut last) = (0, 0); let mut at_start = true; @@ -150,7 +153,6 @@ fn scan_as_path(mut segments: &[u8]) -> Option { /// AS numbers to `out`, and returns whether any segment is not a sequence. /// Stops at what is not a segment. fn read_as_path(mut segments: &[u8], out: &mut Vec) -> bool { - const AS_SEQUENCE: u8 = 2; let mut has_set = false; while let [kind, count, rest @ ..] = segments { let Some((asns, rest)) = rest.split_at_checked(usize::from(*count) * 4) else { @@ -187,24 +189,37 @@ impl AttrSet { &self.wire } + /// The attributes one by one, as they lie in the stored bytes. The + /// bytes were checked when the set was made, so this is all of them; + /// it stops at the first that does not split as an attribute all the + /// same, rather than go on from the middle of one. + fn raw_attributes(&self) -> impl Iterator> { + let mut wire = &*self.wire; + std::iter::from_fn(move || { + let (attribute, rest) = split_attribute(wire).ok()?; + wire = rest; + Some(attribute) + }) + } + + /// The value of the attribute of type `type_code`, if the set has one. + fn attribute(&self, type_code: u8) -> Option<&[u8]> { + self.raw_attributes() + .find(|attribute| attribute.type_code == type_code) + .map(|attribute| attribute.value) + } + /// The route's next hop: the one that came in MP_REACH_NLRI, or else /// the NEXT_HOP attribute. Read off the stored bytes without building /// anything else, for those who ask it of every route. pub fn next_hop(&self) -> Option { let mut plain = None; - let mut wire = &*self.wire; - while let Ok((attribute, rest)) = split_attribute(wire) { + for attribute in self.raw_attributes() { match attribute.type_code { - attr_type::MP_REACH_NLRI => { - // Kept as the next hop's length and the next hop. - let (length, address) = attribute.value.split_first()?; - return NlriNextHop::from_bytes(address) - .filter(|_| usize::from(*length) == address.len()); - } + attr_type::MP_REACH_NLRI => return decode_next_hop(attribute.value), attr_type::NEXT_HOP => plain = NlriNextHop::from_bytes(attribute.value), _ => {} } - wire = rest; } plain } @@ -215,14 +230,7 @@ impl AttrSet { /// for those who ask it of every route: a session to a peer or a /// provider asks it of most of the table, to pass over them. pub fn is_only_to_customer(&self) -> bool { - let mut wire = &*self.wire; - while let Ok((attribute, rest)) = split_attribute(wire) { - if attribute.type_code == attr_type::ONLY_TO_CUSTOMER { - return true; - } - wire = rest; - } - false + self.attribute(attr_type::ONLY_TO_CUSTOMER).is_some() } /// Whether the route carries the BLACKHOLE community (RFC 7999), which @@ -231,15 +239,8 @@ impl AttrSet { /// of every route. pub fn is_blackhole(&self) -> bool { const BLACKHOLE: [u8; 4] = [0xFF, 0xFF, 0x02, 0x9A]; - let mut wire = &*self.wire; - while let Ok((attribute, rest)) = split_attribute(wire) { - if attribute.type_code == attr_type::COMMUNITIES { - let (communities, _) = attribute.value.as_chunks::<4>(); - return communities.contains(&BLACKHOLE); - } - wire = rest; - } - false + self.attribute(attr_type::COMMUNITIES) + .is_some_and(|communities| communities.as_chunks::<4>().0.contains(&BLACKHOLE)) } /// Puts the AS numbers of the route's path in `out`, in place of what @@ -253,14 +254,10 @@ impl AttrSet { /// else, for those who ask it of every route. pub fn as_path(&self, out: &mut Vec) -> bool { out.clear(); - let mut wire = &*self.wire; - while let Ok((attribute, rest)) = split_attribute(wire) { - if attribute.type_code == attr_type::AS_PATH { - return read_as_path(attribute.value, out); - } - wire = rest; + match self.attribute(attr_type::AS_PATH) { + Some(segments) => read_as_path(segments, out), + None => false, } - false } /// The attributes, parsed, and the next hop if it came in @@ -307,7 +304,8 @@ fn encode( Ok(()) } -/// Writes a next hop as an MP_REACH_NLRI that holds nothing else. +/// Writes a next hop as an MP_REACH_NLRI that holds nothing else: the +/// next hop's length, and the next hop. fn encode_next_hop(next_hop: &NlriNextHop, out: &mut BytesMut) { out.put_u8(OPTIONAL); out.put_u8(attr_type::MP_REACH_NLRI); @@ -316,6 +314,13 @@ fn encode_next_hop(next_hop: &NlriNextHop, out: &mut BytesMut) { next_hop.to_wire(out); } +/// Reads the next hop out of the value of an MP_REACH_NLRI as +/// [`encode_next_hop`] writes it. `None` if the value is not one. +fn decode_next_hop(value: &[u8]) -> Option { + let (length, address) = value.split_first()?; + NlriNextHop::from_bytes(address).filter(|_| usize::from(*length) == address.len()) +} + /// 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. @@ -326,11 +331,7 @@ fn parse(mut wire: &[u8]) -> Option<(Vec, Option)> { while !wire.is_empty() { let (raw, rest) = split_attribute(wire).ok()?; if raw.type_code == attr_type::MP_REACH_NLRI { - let (length, address) = raw.value.split_first()?; - if usize::from(*length) != address.len() { - return None; - } - next_hop = Some(NlriNextHop::from_bytes(address)?); + next_hop = Some(decode_next_hop(raw.value)?); } else { let (_, attribute) = PathAttribute::from_wire(&ctx, raw.wire).ok()?; attributes.push(attribute); @@ -650,6 +651,67 @@ mod tests { ); } + /// The OTC attribute is read off the stored bytes, wherever in them it + /// lies. + #[test] + fn otc_is_told_from_the_bytes() { + let store = AttrStore::new(); + let without = store.intern(&attributes(vec![3333]), None).unwrap(); + assert!(!without.is_only_to_customer()); + + let mut with = attributes(vec![3333]); + with.insert(1, PathAttribute::OnlyToCustomer(3333)); + assert!(store.intern(&with, None).unwrap().is_only_to_customer()); + + // After COMMUNITIES, which is the last of the usual attributes. + let mut after = attributes(vec![3333]); + after.push(PathAttribute::OnlyToCustomer(3333)); + assert!(store.intern(&after, None).unwrap().is_only_to_customer()); + } + + /// What a session received, with attributes of our own added after it, + /// is one set with the parsed form of the whole; the bytes are not + /// parsed to find that out, so the bytes must say the same thing. + #[test] + fn received_attributes_with_ours_added_are_the_same_set_as_the_parsed_form() { + let store = AttrStore::new(); + let ctx = four_octet(); + let received = attributes(vec![65001, 65002]); + let mut wire = BytesMut::new(); + encode(&received, None, &mut wire).unwrap(); + let ours = [ + PathAttribute::OnlyToCustomer(65001), + PathAttribute::LocalPref(LocalPrefPathAttribute(150)), + ]; + let mut added = BytesMut::new(); + for attribute in &ours { + attribute.to_wire(&ctx, &mut added).unwrap(); + } + let next_hop = NlriNextHop::Ipv6("2001:db8::1".parse().unwrap()); + + let mut whole = received.clone(); + whole.extend(ours.iter().cloned()); + let parsed = store.intern(&whole, Some(&next_hop)).unwrap(); + let from_bytes = store + .intern_received_with(&wire, &added, Some(&next_hop)) + .unwrap(); + assert!(Arc::ptr_eq(&parsed, &from_bytes)); + assert!(from_bytes.is_only_to_customer()); + assert_eq!(from_bytes.next_hop(), Some(next_hop)); + + // Without a next hop too. + let parsed = store.intern(&whole, None).unwrap(); + let from_bytes = store.intern_received_with(&wire, &added, None).unwrap(); + assert!(Arc::ptr_eq(&parsed, &from_bytes)); + + // An attribute of ours of a type the peer sent is not a set. + let mut repeated = BytesMut::new(); + PathAttribute::MultiExitDisc(MultiExitDiscPathAttribute(1)) + .to_wire(&ctx, &mut repeated) + .unwrap(); + assert!(store.intern_received_with(&wire, &repeated, None).is_none()); + } + /// Two of the same attribute are not a set: its readers could each /// take a different one. #[test] diff --git a/crates/rib/src/table/journal.rs b/crates/rib/src/table/journal.rs index 0d92b18..0aff74e 100644 --- a/crates/rib/src/table/journal.rs +++ b/crates/rib/src/table/journal.rs @@ -95,7 +95,7 @@ struct Entry { } /// One journal of one shard. -struct Journal { +pub(super) struct Journal { kind: Kind, /// The sequence number of the latest change. Counts every change this /// journal is for, whether or not anyone was reading. @@ -113,10 +113,6 @@ struct Journal { /// Where each reader has read to, by reader slot. Written by the /// readers themselves while they hold the shard for reading. cursors: Vec, - /// `seq`, for reading without the shard's lock: the table holds the - /// other end, so that a reader can tell a shard with nothing new for - /// it from one with something without stopping the shard's writers. - latest: Arc, /// How many reader slots are in use. readers: usize, limits: Limits, @@ -129,7 +125,6 @@ impl Journal { Self { kind, limits, - latest: Arc::new(AtomicU64::new(0)), seq: 0, floor: 0, entries: VecDeque::new(), @@ -144,7 +139,6 @@ impl Journal { /// number for the row to remember. fn changed(&mut self, prefix: IpPrefix) -> u64 { self.seq += 1; - self.latest.store(self.seq, Ordering::Release); if self.readers > 0 { self.entries.push_back(Entry { seq: self.seq, @@ -167,7 +161,6 @@ impl Journal { /// Notes that `prefix` lost its last route and has no row any more. fn disappeared(&mut self, prefix: IpPrefix) { self.seq += 1; - self.latest.store(self.seq, Ordering::Release); if self.readers > 0 { self.entries.push_back(Entry { seq: self.seq, @@ -178,25 +171,35 @@ impl Journal { } } - /// Forgets that `entry`'s prefix had gone, if `entry` is what said so. - fn drop_entry(gone: &mut HashMap, entry: &Entry) { + /// Voids the note that `entry`'s prefix had gone, if `entry` is what + /// said so: an entry that is dropped must not leave the note behind, + /// or the prefix would be given out as gone on the strength of a later + /// entry that says something else. + fn void_gone_note(gone: &mut HashMap, entry: &Entry) { if gone.get(&entry.prefix) == Some(&entry.seq) { gone.remove(&entry.prefix); } } - /// What `entry` says, if it still says something: it is the latest - /// change of its prefix. The row, or `None` for a prefix that has gone. - fn live<'a>( - &self, - entry: &Entry, - rows: &'a HashTable, - hasher: &RandomState, - ) -> Option> { + /// What `entry` says: something only if it is the latest change of its + /// prefix. + fn says<'a>(&self, entry: &Entry, rows: &'a HashTable, hasher: &RandomState) -> Says<'a> { let hash = hasher.hash_one(entry.prefix); match rows.find(hash, |row| row.prefix == entry.prefix) { - Some(row) => (row.seqs.of(self.kind) == entry.seq).then_some(Some(row)), - None => (self.gone.get(&entry.prefix) == Some(&entry.seq)).then_some(None), + Some(row) if *row.seqs.of(self.kind) == entry.seq => Says::Row(row), + None if self.gone.get(&entry.prefix) == Some(&entry.seq) => Says::Gone, + _ => Says::Nothing, + } + } + + /// Where a reader whose cursor is `cursor` is to read from. + fn place_of(&self, cursor: u64) -> Place { + if cursor < self.floor { + Place::Lagged + } else if cursor == self.seq { + Place::Nothing + } else { + Place::From(self.entries.partition_point(|entry| entry.seq <= cursor)) } } @@ -206,7 +209,26 @@ impl Journal { if self.entries.len() <= self.limits.tidy_at { return; } - // What every reader has passed can go. + self.drop_passed(); + // Entries for prefixes that have a row are at most as many as the + // rows once the dead ones are out. Entries for prefixes that have + // gone are not counted in: they are what a reader that has stopped + // reading makes pile up. Twice that, so that the pass over every + // entry that finding the dead ones takes is made only once the + // journal has doubled, and the cost per change stays small. + let limit = 2 * (rows.len() + self.limits.slack); + if self.entries.len() <= limit { + return; + } + self.drop_dead(rows, hasher); + if self.entries.len() <= limit { + return; + } + self.drop_oldest_half(); + } + + /// Drops what every reader has passed. + fn drop_passed(&mut self) { let passed = self .cursors .iter() @@ -217,40 +239,57 @@ impl Journal { if front.seq > passed { break; } - Self::drop_entry(&mut self.gone, front); + Self::void_gone_note(&mut self.gone, front); self.entries.pop_front(); } + } - // Entries for prefixes that have a row are at most as many as the - // rows once the dead ones are out. Entries for prefixes that have - // gone are not counted in: they are what a reader that has stopped - // reading makes pile up. - let limit = 2 * (rows.len() + self.limits.slack); - if self.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. + /// Drops the dead entries: those whose prefix has changed again since, + /// so that a later entry says what they said. This looks every entry + /// up. + fn drop_dead(&mut self, rows: &HashTable, hasher: &RandomState) { let mut entries = std::mem::take(&mut self.entries); - entries.retain(|entry| self.live(entry, rows, hasher).is_some()); + entries.retain(|entry| !matches!(self.says(entry, rows, hasher), Says::Nothing)); self.entries = entries; + } - if self.entries.len() <= limit { - return; - } - // Still too long: a reader is not reading while prefixes come and - // go, or more prefixes went at once than are left. The older half - // goes, and a reader that had not read it loses its place. Half, - // so that readers that are only a little behind keep theirs, and - // so that getting a long journal down to size takes few passes. + /// Drops the older half of the entries without regard to the readers, + /// for a journal that is too long even without its dead entries: a + /// reader is not reading while prefixes come and go, or more prefixes + /// went at once than are left. A reader that had not read what goes + /// loses its place. Half, so that readers that are only a little + /// behind keep theirs, and so that getting a long journal down to size + /// takes few passes. + fn drop_oldest_half(&mut self) { for entry in self.entries.drain(..self.entries.len() / 2) { - Self::drop_entry(&mut self.gone, &entry); + Self::void_gone_note(&mut self.gone, &entry); self.floor = entry.seq; } } } +/// What an entry says; see [`Journal::says`]. +enum Says<'a> { + /// The prefix's row, whose latest change the entry is. + Row(&'a Row), + /// The prefix has no route any more, and the entry is what says so. + Gone, + /// Nothing: the prefix has changed again since, and a later entry has + /// it. + Nothing, +} + +/// Where a reader is to read a journal from; see [`Journal::place_of`]. +enum Place { + /// Entries the reader had not read were dropped; it has lost its + /// place. + Lagged, + /// The reader has read everything up to the latest change. + Nothing, + /// The index of the first entry the reader has not read. + From(usize), +} + /// Which of a shard's two journals. #[derive(Debug, Clone, Copy, PartialEq, Eq)] pub(super) enum Kind { @@ -260,39 +299,36 @@ pub(super) enum Kind { All, } -impl Kind { - /// Its place in what the table keeps for each shard's two journals. - pub(super) fn index(self) -> usize { - match self { - Self::Best => 0, - Self::All => 1, +/// One of something for each [`Kind`] of journal. +#[derive(Debug, Clone, Copy, Default)] +pub(super) struct ByKind { + best: T, + all: T, +} + +impl ByKind { + pub(super) fn of(&self, kind: Kind) -> &T { + match kind { + Kind::Best => &self.best, + Kind::All => &self.all, + } + } + + pub(super) fn of_mut(&mut self, kind: Kind) -> &mut T { + match kind { + Kind::Best => &mut self.best, + Kind::All => &mut self.all, } } } /// The sequence numbers a row remembers, one for each journal: of the /// latest change of its best route, and of any of its routes. -#[derive(Debug, Clone, Copy)] -pub(super) struct Seqs { - best: u64, - all: u64, -} - -impl Seqs { - fn of(self, kind: Kind) -> u64 { - match kind { - Kind::Best => self.best, - Kind::All => self.all, - } - } -} +pub(super) type Seqs = ByKind; /// The two journals of a shard, written together: every change to a row is /// a change of one of its routes, and some are a change of its best route. -pub(super) struct Journals { - best: Journal, - all: Journal, -} +pub(super) type Journals = ByKind; impl Journals { pub(super) fn new(limits: Limits) -> Self { @@ -302,24 +338,13 @@ impl Journals { } } - /// The latest sequence number of each journal, by [`Kind::index`], - /// as something to read without the shard's lock. - pub(super) fn latest(&self) -> [Arc; 2] { - [Arc::clone(&self.best.latest), Arc::clone(&self.all.latest)] - } - - fn of(&self, kind: Kind) -> &Journal { - match kind { - Kind::Best => &self.best, - Kind::All => &self.all, - } - } - - fn of_mut(&mut self, kind: Kind) -> &mut Journal { - match kind { - Kind::Best => &mut self.best, - Kind::All => &mut self.all, - } + /// Copies the journals' sequence numbers to `latest`, which the table + /// keeps outside the shard's lock for readers to look at + /// ([`RibTable::latest_seq`]). Called with the lock held, after every + /// change to the shard. + pub(super) fn publish(&self, latest: &ByKind) { + latest.best.store(self.best.seq, Ordering::Release); + latest.all.store(self.all.seq, Ordering::Release); } /// The time by the table's clock that the changes made from now on @@ -385,18 +410,11 @@ pub(super) struct Mark(u64, u64); pub struct Routes<'a>(&'a Row); impl<'a> Routes<'a> { + /// The routes: at least one, since a prefix with none has no row, and + /// is given out as [`Event::Withdrawn`] instead. pub fn iter(&self) -> impl Iterator> + use<'a> { self.0.routes() } - - /// How many there are: at least one. - pub fn len(&self) -> usize { - self.0.paths.len() - } - - pub fn is_empty(&self) -> bool { - self.len() == 0 - } } impl std::fmt::Debug for Routes<'_> { @@ -447,19 +465,15 @@ pub struct Progress { } enum Phase { - /// Giving out the whole table; `next` is the shard to do next, and - /// `announced` whether [`Event::Reset`] has been given. `upto` is the - /// time on the table's clock the table is given out as of: the shards - /// are gone through one at a time while the table changes, and the - /// place taken in each shard's journal is as of that one time, so that - /// what follows is consistent from shard to shard. A change after it - /// that a later shard's rows already show is read again from the + /// Giving out the whole table; `next` is the shard to do next. `upto` + /// is the time on the table's clock the table is given out as of, and + /// is taken when [`Event::Reset`] is given, so `None` until then: the + /// shards are gone through one at a time while the table changes, and + /// the place taken in each shard's journal is as of that one time, so + /// that what follows is consistent from shard to shard. A change after + /// it that a later shard's rows already show is read again from the /// journal; that is harmless. - Resync { - next: usize, - announced: bool, - upto: u64, - }, + Resync { next: usize, upto: Option }, /// Following the journals; `next` is the shard to read next, and /// `upto` the time by the table's clock up to which this pass reads /// (see [`Entry::stamp`]). @@ -470,9 +484,11 @@ impl Phase { /// The beginning of a resynchronisation. const RESYNC: Self = Self::Resync { next: 0, - announced: false, - upto: 0, + upto: None, }; + + /// The beginning of a pass over the journals. + const FOLLOW: Self = Self::Follow { next: 0, upto: 0 }; } /// Follows the routes of a table: the best of each prefix, or all of them. @@ -489,7 +505,10 @@ pub struct Reader { /// is for the shard's benefit. cursors: Vec, phase: Phase, - /// The table's version when the reader was last found caught up. + /// The table's version as it was when the current pass over the + /// journals began. [`Reader::changed`] waits while the version is still + /// this one: whatever was written since the pass began has moved it + /// on, and is there to be read. seen: u64, caught_up: bool, } @@ -505,7 +524,7 @@ impl RibTable { /// those who want the news and not the table. If it falls too far /// behind it is reset like any other, and is then given the table. pub fn reader_from_now(self: &Arc) -> Reader { - let mut reader = self.register(Kind::Best, Phase::Follow { next: 0, upto: 0 }); + let mut reader = self.register(Kind::Best, Phase::FOLLOW); for (index, shard) in self.shards.iter().enumerate() { let shard = read(shard); reader.set_cursor(index, &shard, shard.journals.best.seq); @@ -625,114 +644,124 @@ impl Reader { // shard of every pass, by every reader, was the most of what a // router with many peers did. let table = Arc::clone(&self.table); - let shards = table.shards.len(); let mut events = 0; loop { - match self.phase { - Phase::Resync { - next, - announced, - upto, - } => { - let upto = if announced { - upto - } else { - self.forget_place(); - visit(Event::Reset); - events += 1; - self.table.settled() - }; - if next == shards { - visit(Event::Synced); - events += 1; - self.phase = Phase::Follow { next: 0, upto: 0 }; - continue; - } - events += self.scan(&table, next, upto, &mut visit); - self.phase = Phase::Resync { - next: next + 1, - announced: true, - upto, - }; - if events >= budget { - return self.progress(events, false); - } + let done = match self.phase { + Phase::Resync { next, upto } => { + self.resync_step(&table, next, upto, budget, &mut events, &mut visit) } Phase::Follow { next, upto } => { - // Read before the pass: a change made during it shows as - // a newer version, and `changed` then returns at once. - // And the pass reads up to one time in every shard - // alike, chosen so that about the budget is read. - let (version, upto) = if next == 0 { - ( - self.table.version.load(Ordering::Acquire), - self.time_for(budget - events), - ) - } else { - (self.seen, upto) - }; - self.seen = version; - let mut shard = next; - let mut lagged = false; - let mut consumed_any = false; - let mut more_somewhere = false; - while shard < shards { - if events >= budget { - self.phase = Phase::Follow { next: shard, upto }; - return self.progress(events, false); - } - // Nothing new in this shard, by the number that is - // read without its lock: it is left alone, and so - // are the sessions writing to it. - if self.cursors[shard] == self.table.latest_seq(shard, self.kind) { - shard += 1; - continue; - } - match self.drain(&table, shard, upto, &mut visit) { - Drained::Lagged => { - lagged = true; - break; - } - Drained::Read { - events: read, - consumed, - more, - } => { - events += read; - consumed_any |= consumed; - more_somewhere |= more; - } - } - shard += 1; - } - if lagged { - self.phase = Phase::RESYNC; - continue; - } - self.phase = Phase::Follow { next: 0, upto: 0 }; - if next == 0 && !consumed_any && more_somewhere { - // A whole pass got nowhere, though there is more: - // it is all beyond the earliest change still under - // way, which no pass can read past until it ends. - // Asking again at once would spin until then. The - // change wakes the readers when it ends. - return self.progress(events, true); - } - if next != 0 || more_somewhere { - // This pass took up where an earlier call stopped, - // so the shards before that point have not been - // looked at in this call; or some shard had more - // than its share, or changes later than this - // pass's time. Go round again from the first. - if events >= budget { - return self.progress(events, false); - } - continue; - } - return self.progress(events, true); + self.follow_pass(&table, next, upto, budget, &mut events, &mut visit) + } + }; + if let Some(progress) = done { + return progress; + } + } + } + + /// One step of a resynchronisation: the shard `next`, or the end of it + /// once every shard has been given out. Returns what to tell the + /// caller if the budget is spent, and `None` to go on. + fn resync_step( + &mut self, + table: &RibTable, + next: usize, + upto: Option, + budget: usize, + events: &mut usize, + visit: &mut impl FnMut(Event<'_>), + ) -> Option { + let upto = upto.unwrap_or_else(|| { + self.forget_place(); + visit(Event::Reset); + *events += 1; + table.settled() + }); + if next == table.shards.len() { + visit(Event::Synced); + *events += 1; + self.phase = Phase::FOLLOW; + return None; + } + *events += self.scan(table, next, upto, visit); + self.phase = Phase::Resync { + next: next + 1, + upto: Some(upto), + }; + (*events >= budget).then(|| self.progress(*events, false)) + } + + /// A pass over the journals from shard `next` on, or the rest of one + /// that an earlier call cut short. Returns what to tell the caller if + /// the pass is to be the call's last, and `None` to go round again. + fn follow_pass( + &mut self, + table: &RibTable, + next: usize, + upto: u64, + budget: usize, + events: &mut usize, + visit: &mut impl FnMut(Event<'_>), + ) -> Option { + // Read before the pass: a change made during it shows as a newer + // version, and `changed` then returns at once. And the pass reads + // up to one time in every shard alike, chosen so that about the + // budget is read. + let (version, upto) = if next == 0 { + ( + table.version.load(Ordering::Acquire), + self.time_for(budget - *events), + ) + } else { + (self.seen, upto) + }; + self.seen = version; + let mut consumed_any = false; + let mut more_somewhere = false; + for shard in next..table.shards.len() { + if *events >= budget { + self.phase = Phase::Follow { next: shard, upto }; + return Some(self.progress(*events, false)); + } + // Nothing new in this shard, by the number that is read + // without its lock: it is left alone, and so are the sessions + // writing to it. + if self.cursors[shard] == table.latest_seq(shard, self.kind) { + continue; + } + match self.drain(table, shard, upto, visit) { + Drained::Lagged => { + self.phase = Phase::RESYNC; + return None; + } + Drained::Read { + events: read, + consumed, + more, + } => { + *events += read; + consumed_any |= consumed; + more_somewhere |= more; } } } + self.phase = Phase::FOLLOW; + if next == 0 && !consumed_any && more_somewhere { + // A whole pass got nowhere, though there is more: it is all + // beyond the earliest change still under way, which no pass + // can read past until it ends. Asking again at once would spin + // until then. The change wakes the readers when it ends. + return Some(self.progress(*events, true)); + } + if next != 0 || more_somewhere { + // This pass took up where an earlier call stopped, so the + // shards before that point have not been looked at in this + // call; or some shard had more than its share, or changes + // later than this pass's time. Go round again from the first. + return (*events >= budget).then(|| self.progress(*events, false)); + } + Some(self.progress(*events, true)) } fn progress(&mut self, events: usize, caught_up: bool) -> Progress { @@ -847,10 +876,9 @@ impl Reader { } let shard = read(shard); let journal = shard.journals.of(self.kind); - if cursor < journal.floor || cursor == journal.seq { + let Place::From(start) = journal.place_of(cursor) else { continue; - } - let start = journal.entries.partition_point(|entry| entry.seq <= cursor); + }; if let Some(next) = journal.entries.get(start) { earliest = Some(earliest.map_or(next.stamp, |e: u64| e.min(next.stamp))); } @@ -874,18 +902,17 @@ impl Reader { let shard = read(&table.shards[index]); let journal = shard.journals.of(self.kind); let mut cursor = self.cursors[index]; - if cursor < journal.floor { - return Drained::Lagged; - } - if cursor == journal.seq { - return Drained::Read { - events: 0, - consumed: false, - more: false, - }; - } - - let start = journal.entries.partition_point(|entry| entry.seq <= cursor); + let start = match journal.place_of(cursor) { + Place::Lagged => return Drained::Lagged, + Place::Nothing => { + return Drained::Read { + events: 0, + consumed: false, + more: false, + }; + } + Place::From(start) => start, + }; let mut events = 0; let mut more = false; for entry in journal.entries.range(start..) { @@ -894,13 +921,12 @@ impl Reader { break; } cursor = entry.seq; - let event = match journal.live(entry, &shard.rows, &shard.hasher) { - Some(Some(row)) => self.event_for(row), - Some(None) => Event::Withdrawn { + let event = match journal.says(entry, &shard.rows, &shard.hasher) { + Says::Row(row) => self.event_for(row), + Says::Gone => Event::Withdrawn { prefix: entry.prefix, }, - // The prefix has changed again; a later entry has it. - None => continue, + Says::Nothing => continue, }; visit(event); events += 1; diff --git a/crates/rib/src/table/journal_tests.rs b/crates/rib/src/table/journal_tests.rs index f02c6ef..d11a32b 100644 --- a/crates/rib/src/table/journal_tests.rs +++ b/crates/rib/src/table/journal_tests.rs @@ -8,7 +8,7 @@ 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}; +use super::{Event, Kind, Limits, Reader, read}; /// What a follower would keep: its own copy of the best routes, brought /// up to date from the events it is given. @@ -53,7 +53,10 @@ impl Mirror { } Event::Routes { prefix, routes } => { self.routes_events += 1; - assert!(!routes.is_empty(), "a prefix with no routes is withdrawn"); + assert!( + routes.iter().next().is_some(), + "a prefix with no routes is withdrawn" + ); self.routes.insert( prefix, sorted( @@ -115,9 +118,6 @@ impl Mirror { ); } } -} - -impl Mirror { /// For the mirror of a follower of all routes: it holds every route in /// the table, and knows which is best. #[track_caller] @@ -213,6 +213,107 @@ fn a_reader_from_now_is_given_the_changes_and_not_the_table() { assert_eq!((best, withdrawn), (vec![new], vec![old])); } +/// The number a reader checks without a shard's lock must be the journal's +/// own after every kind of change, whether or not anyone is reading: a +/// reader skips a shard on the strength of it, and a shard with news that +/// said it had none would never be read until the next change. +#[test] +fn the_lock_free_number_is_the_journals_after_every_kind_of_change() { + let mut fixture = Fixture::new(&[external(1), external(2)]); + let prefixes: Vec = (0..16).map(nth_prefix).collect(); + let agree = |table: &RibTable| { + for (index, shard) in table.shards.iter().enumerate() { + let shard = read(shard); + for kind in [Kind::Best, Kind::All] { + assert_eq!( + table.latest_seq(index, kind), + shard.journals.of(kind).seq, + "shard {index}, {kind:?}" + ); + } + } + }; + + // Without a reader, with one of each kind, and with both. + for readers in 0..4 { + let _best = (readers & 1 != 0).then(|| fixture.table.reader()); + let _all = (readers & 2 != 0).then(|| fixture.table.reader_of_all_routes()); + agree(&fixture.table); + + for p in &prefixes { + fixture.announce(0, p, &Route::via(&[1])); + } + agree(&fixture.table); + // The same prefixes again, with other attributes. + for p in &prefixes { + fixture.announce(0, p, &Route::via(&[1, 2])); + } + agree(&fixture.table); + // A second route for each, then the last route of one taken away. + for p in &prefixes { + fixture.announce(1, p, &Route::via(&[2, 3])); + } + agree(&fixture.table); + fixture.withdraw(0, &prefixes[0]); + agree(&fixture.table); + fixture.withdraw(1, &prefixes[0]); + agree(&fixture.table); + + fixture.table.remove_peer(fixture.peers[1]); + agree(&fixture.table); + fixture.peers[1] = fixture.table.add_peer(external(2)).unwrap(); + + fixture + .table + .mark_stale(fixture.peers[0], &[IpVersion::Ipv4]); + agree(&fixture.table); + fixture + .table + .sweep_stale(fixture.peers[0], &[IpVersion::Ipv4]); + agree(&fixture.table); + assert!( + fixture.table.is_empty(), + "round {readers} left routes behind" + ); + } +} + +/// A shard that had nothing new for a reader is passed over without its +/// lock. The next change to it must bring the reader back. +#[test] +fn a_shard_passed_over_is_read_again_once_it_has_news() { + let fixture = Fixture::on(RibTable::new(1), &[external(1)]); + let mut reader = fixture.table.reader(); + let mut mirror = Mirror::default(); + mirror.catch_up(&mut reader, 100); + assert_eq!(reader.poll(100, |_| panic!("no event expected")).events, 0); + + fixture.announce(0, &prefix("10.0.0.0/8"), &Route::via(&[1])); + let mut events = Vec::new(); + let progress = reader.poll(100, |event| events.push(format!("{event:?}"))); + assert!(progress.caught_up); + assert_eq!(events.len(), 1, "{events:?}"); + assert!(events[0].starts_with("Best"), "{events:?}"); +} + +/// As [`a_shard_passed_over_is_read_again_once_it_has_news`], for a reader +/// that begins where the journals are rather than with the table. +#[test] +fn a_shard_passed_over_by_a_reader_from_now_is_read_again_once_it_has_news() { + let fixture = Fixture::on(RibTable::new(1), &[external(1)]); + fixture.announce(0, &prefix("192.0.2.0/24"), &Route::via(&[1])); + let mut reader = fixture.table.reader_from_now(); + assert!(reader.poll(100, |_| panic!("no event expected")).caught_up); + assert_eq!(reader.poll(100, |_| panic!("no event expected")).events, 0); + + fixture.announce(0, &prefix("10.0.0.0/8"), &Route::via(&[1])); + let mut events = Vec::new(); + let progress = reader.poll(100, |event| events.push(format!("{event:?}"))); + assert!(progress.caught_up); + assert_eq!(events.len(), 1, "{events:?}"); + assert!(events[0].starts_with("Best"), "{events:?}"); +} + /// A change to a route that is not the best is nothing to a follower of /// the best routes, and is what a follower of all routes is there for. #[test] diff --git a/crates/rib/src/table/mod.rs b/crates/rib/src/table/mod.rs index e28ed8d..783f5f8 100644 --- a/crates/rib/src/table/mod.rs +++ b/crates/rib/src/table/mod.rs @@ -313,13 +313,21 @@ struct Shard { journals: journal::Journals, } -/// The routing table. Shared between threads by reference; every method -/// takes `&self`. +/// A shard, and beside its lock what is read without taking it. +struct ShardSlot { + lock: RwLock, + /// The sequence number of each of the shard's journals as of the last + /// change to the shard, stored by the change while it still holds the + /// lock ([`RibTable::change`]). A reader looks at it first, and passes + /// over a shard with nothing new for it without stopping the shard's + /// writers. + latest: journal::ByKind, +} + +/// The routing table. Shared between threads by reference: nothing takes +/// `&mut self`, and the readers ([`Reader`]) keep an `Arc` of their own. pub struct RibTable { - shards: Box<[RwLock]>, - /// The latest sequence number of each shard's two journals, kept by - /// the journals themselves, for reading without the shard's lock. - latest: Box<[[Arc; 2]]>, + shards: Box<[ShardSlot]>, hasher: RandomState, attrs: AttrStore, /// Which peer numbers are in use. Changing the peer list also takes @@ -358,23 +366,20 @@ impl RibTable { /// sizes. Tests use small ones to get there quickly. fn with_journal_limits(shards: usize, limits: journal::Limits) -> Self { let hasher = RandomState::default(); - let mut latest = Vec::with_capacity(shards.max(1)); Self { shards: (0..shards.max(1)) - .map(|_| { - let journals = journal::Journals::new(limits); - latest.push(journals.latest()); - RwLock::new(Shard { + .map(|_| ShardSlot { + lock: RwLock::new(Shard { rows: HashTable::new(), routes: 0, peers: Vec::new(), hasher: hasher.clone(), scratch: select::Scratch::default(), - journals, - }) + journals: journal::Journals::new(limits), + }), + latest: journal::ByKind::default(), }) .collect(), - latest: latest.into_boxed_slice(), hasher, attrs: AttrStore::new(), peers: Mutex::default(), @@ -523,7 +528,7 @@ impl RibTable { /// How many routes of `peer` are stale; see [`RibTable::mark_stale`]. /// Looks at every prefix in the table. #[cfg(test)] - pub(super) fn count_stale(&self, peer: PeerIndex) -> usize { + fn count_stale(&self, peer: PeerIndex) -> usize { self.shards .iter() .map(|shard| { @@ -623,18 +628,22 @@ impl RibTable { /// journal one of them follows was written to. fn change<'a, T>( &'a self, - parts: impl IntoIterator, T)>, + parts: impl IntoIterator, mut apply: impl FnMut(&mut Shard, T), ) { let change = self.begin_change(); let mut news = false; - for (shard, part) in parts { - let mut shard = write(shard); + for (slot, part) in parts { + let mut shard = write(slot); shard.journals.set_stamp(change.stamp); let before = shard.journals.mark(); apply(&mut shard, part); shard.tidy_journals(); news |= shard.journals.has_news_since(before); + // Once per change, not once per route; and before the lock is + // let go, so that a reader that sees the new number and takes + // the lock finds the entries behind it. + shard.journals.publish(&slot.latest); } drop(change); if news { @@ -658,15 +667,18 @@ impl RibTable { } } - /// The latest sequence number of shard `shard`'s journal of `kind`, - /// without the shard's lock. + /// The sequence number of shard `shard`'s journal of `kind` as of the + /// last change to the shard, without the shard's lock. fn latest_seq(&self, shard: usize, kind: journal::Kind) -> u64 { - self.latest[shard][kind.index()].load(std::sync::atomic::Ordering::Acquire) + self.shards[shard] + .latest + .of(kind) + .load(std::sync::atomic::Ordering::Acquire) } /// The time on the clock up to which every change is complete in every /// shard: everything before the earliest one still under way. - pub(super) fn settled(&self) -> u64 { + fn settled(&self) -> u64 { let in_progress = self .in_progress .lock() @@ -709,11 +721,8 @@ impl RibTable { } /// 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[row.best()]; - (best.peer(), Arc::clone(&best.attrs)) - }) + pub fn best(&self, prefix: &IpPrefix) -> Option { + self.with_row(prefix, |row| row.route(row.best()).to_owned()) } /// Every route for `prefix`. @@ -811,14 +820,14 @@ impl RibTable { } } -fn read(shard: &RwLock) -> RwLockReadGuard<'_, Shard> { +fn read(shard: &ShardSlot) -> 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) + shard.lock.read().unwrap_or_else(PoisonError::into_inner) } -fn write(shard: &RwLock) -> RwLockWriteGuard<'_, Shard> { - shard.write().unwrap_or_else(PoisonError::into_inner) +fn write(shard: &ShardSlot) -> RwLockWriteGuard<'_, Shard> { + shard.lock.write().unwrap_or_else(PoisonError::into_inner) } /// A change to the table that is under way; see [`RibTable::begin_change`]. diff --git a/crates/rib/src/table/tests.rs b/crates/rib/src/table/tests.rs index fa8dbe2..04db89e 100644 --- a/crates/rib/src/table/tests.rs +++ b/crates/rib/src/table/tests.rs @@ -162,7 +162,7 @@ impl Fixture { /// Which peer (by order added) the best route for `prefix` is from. fn best(&self, prefix: &IpPrefix) -> Option { - let (peer, _) = self.table.best(prefix)?; + let peer = self.table.best(prefix)?.peer; self.peers.iter().position(|p| *p == peer) } } diff --git a/crates/server/src/reach.rs b/crates/server/src/reach.rs index c11f8d4..82dfe24 100644 --- a/crates/server/src/reach.rs +++ b/crates/server/src/reach.rs @@ -370,7 +370,7 @@ mod tests { } fn best(&self, route: &str) -> Option { - self.table.best(&prefix(route)).map(|(peer, _)| peer) + self.table.best(&prefix(route)).map(|best| best.peer) } }