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) } }