diff --git a/crates/rib/src/table/journal.rs b/crates/rib/src/table/journal.rs index 24116c5..d65c90e 100644 --- a/crates/rib/src/table/journal.rs +++ b/crates/rib/src/table/journal.rs @@ -387,9 +387,8 @@ impl std::fmt::Debug for Routes<'_> { pub enum Event<'a> { /// Forget what you know. The whole table follows, as [`Event::Best`] /// or [`Event::Routes`] 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. + /// 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`. Given to a follower of the /// best routes. @@ -577,8 +576,14 @@ impl Drop for Reader { 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 }, + /// `events` were given out; `consumed` if the reader's place in the + /// journal moved at all; `more` if entries were left past the time + /// read up to. + Read { + events: usize, + consumed: bool, + more: bool, + }, } impl Reader { @@ -642,6 +647,7 @@ impl Reader { 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 { @@ -653,8 +659,13 @@ impl Reader { lagged = true; break; } - Drained::Read { events: read, more } => { + Drained::Read { + events: read, + consumed, + more, + } => { events += read; + consumed_any |= consumed; more_somewhere |= more; } } @@ -665,6 +676,14 @@ impl Reader { 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 @@ -809,6 +828,7 @@ impl Reader { if cursor == journal.seq { return Drained::Read { events: 0, + consumed: false, more: false, }; } @@ -837,8 +857,13 @@ impl Reader { // Past the last entry there is nothing up to the latest change. cursor = journal.seq; } + let consumed = cursor != self.cursors[index]; self.set_cursor(index, &shard, cursor); - Drained::Read { events, more } + Drained::Read { + events, + consumed, + more, + } } } diff --git a/crates/rib/src/table/journal_tests.rs b/crates/rib/src/table/journal_tests.rs index 4691850..f02c6ef 100644 --- a/crates/rib/src/table/journal_tests.rs +++ b/crates/rib/src/table/journal_tests.rs @@ -517,6 +517,45 @@ async fn a_waiting_reader_is_woken_by_a_change_of_best_route() { .expect("the change was already there"); } +/// A reader does not read past a change that is still under way, which +/// a change to every shard (a peer's whole table going) is for a while. +/// Changes made by others meanwhile are beyond it. A reader that finds +/// only those must say it is caught up and wait, not ask again at once +/// for as long as the change takes; and when the change ends, it is +/// woken, though the change itself wrote nothing for it. +#[tokio::test] +async fn a_change_under_way_does_not_keep_a_reader_spinning() { + let fixture = Fixture::new(&[external(1), external(2)]); + let mut reader = fixture.table.reader(); + let mut mirror = Mirror::default(); + mirror.catch_up(&mut reader, 100); + + // A long change begins, and another writer announces meanwhile. + let change = fixture.table.begin_change(); + fixture.announce(1, &prefix("10.0.0.0/8"), &Route::via(&[2])); + + let progress = reader.poll(100, |_| {}); + assert_eq!(progress.events, 0, "nothing can be read past the change"); + assert!( + progress.caught_up, + "and the reader is to wait, not ask again" + ); + let waiting = tokio::time::timeout(std::time::Duration::from_millis(50), reader.changed()); + assert!( + waiting.await.is_err(), + "nothing to be had until the change ends" + ); + + // The change ends without writing anything itself. + drop(change); + tokio::time::timeout(std::time::Duration::from_secs(5), reader.changed()) + .await + .expect("woken when the change ended"); + mirror.catch_up(&mut reader, 100); + mirror.assert_matches(&fixture.table); + assert_eq!(mirror.best.len(), 1); +} + /// 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. diff --git a/crates/rib/src/table/mod.rs b/crates/rib/src/table/mod.rs index 83d289d..26def64 100644 --- a/crates/rib/src/table/mod.rs +++ b/crates/rib/src/table/mod.rs @@ -382,13 +382,6 @@ impl RibTable { &self.attrs } - /// Makes room for a peer's routes and returns the number to announce - /// them under. - /// - /// The number is the peer's until [`RibTable::remove_peer`] is called - /// with it, and may then be given to another. Whoever was given it must - /// not use it after that: the table cannot tell a late announcement - /// from one of the number's next owner. /// What is known of the peer with this number, if it is in use. Not to /// be called from a visitor that is given routes with a shard held. pub fn peer(&self, index: PeerIndex) -> Option { @@ -396,6 +389,13 @@ impl RibTable { peers.get(index.as_usize()).copied().flatten() } + /// Makes room for a peer's routes and returns the number to announce + /// them under. + /// + /// The number is the peer's until [`RibTable::remove_peer`] is called + /// with it, and may then be given to another. Whoever was given it must + /// not use it after that: the table cannot tell a late announcement + /// from one of the number's next owner. pub fn add_peer(&self, info: PeerInfo) -> Result { let mut peers = self.peers.lock().unwrap_or_else(PoisonError::into_inner); let index = match peers.iter().position(Option::is_none) { @@ -428,29 +428,18 @@ impl RibTable { /// held while it does, so other peers can come and go meanwhile. pub fn remove_peer(&self, peer: PeerIndex) -> Outcome { let mut outcome = Outcome::default(); - let mut news = false; - let change = self.begin_change(); - for shard in &self.shards { - let mut shard = write(shard); - shard.journals.set_stamp(change.stamp); - let before = shard.journals.mark(); + self.change(self.shards.iter().map(|shard| (shard, ())), |shard, ()| { outcome += shard.remove_peer(peer); // From here on the shard takes no more routes from the peer. if let Some(slot) = shard.peers.get_mut(peer.as_usize()) { *slot = None; } - shard.tidy_journals(); - news |= shard.journals.has_news_since(before); - } + }); // Only now can the number be given out again: no shard knows it. let mut peers = self.peers.lock().unwrap_or_else(PoisonError::into_inner); if let Some(slot) = peers.get_mut(peer.as_usize()) { *slot = None; } - drop(peers); - if news { - self.wake_readers(); - } outcome } @@ -511,30 +500,20 @@ impl RibTable { /// again, before or during this, is fresh and stays. pub fn sweep_stale(&self, peer: PeerIndex, versions: &[IpVersion]) -> Outcome { let mut outcome = Outcome::default(); - let mut news = false; - let change = self.begin_change(); - for shard in &self.shards { - let mut shard = write(shard); - shard.journals.set_stamp(change.stamp); - let before = shard.journals.mark(); + self.change(self.shards.iter().map(|shard| (shard, ())), |shard, ()| { outcome += shard.remove_where(|row, path| { path.peer() == peer && path.is_stale() && versions.contains(&row.prefix.address_family()) }); - shard.tidy_journals(); - news |= shard.journals.has_news_since(before); - } - drop(change); - if news { - self.wake_readers(); - } + }); outcome } /// How many routes of `peer` are stale; see [`RibTable::mark_stale`]. /// Looks at every prefix in the table. - pub fn count_stale(&self, peer: PeerIndex) -> usize { + #[cfg(test)] + pub(super) fn count_stale(&self, peer: PeerIndex) -> usize { self.shards .iter() .map(|shard| { @@ -594,7 +573,7 @@ impl RibTable { /// Calls `apply` once for each shard that some of `routes` fall in, /// with that shard locked and those routes. Wakes the readers if a /// journal one of them follows was written to. - fn by_shard(&self, routes: &[N], mut apply: impl FnMut(&mut Shard, &[(u64, Nlri)])) + fn by_shard(&self, routes: &[N], apply: impl FnMut(&mut Shard, &[(u64, Nlri)])) where N: Copy + Into, { @@ -620,17 +599,34 @@ impl RibTable { }; placed.sort_unstable_by_key(|(hash, _)| self.shard_of(*hash)); - let mut news = false; - // One time for all of this call's changes, in every shard. + let runs = placed + .chunk_by(|a, b| self.shard_of(a.0) == self.shard_of(b.0)) + .map(|run| (&self.shards[self.shard_of(run[0].0)], run)); + self.change(runs, apply); + } + + /// Makes one change to the table, part by part: `apply` is called with + /// each shard that `parts` names locked for writing, and the part's + /// own item. All the parts are one change, at one time on the clock. + /// + /// Readers are woken once the change is complete in every shard, if a + /// journal one of them follows was written to. + fn change<'a, T>( + &'a self, + parts: impl IntoIterator, T)>, + mut apply: impl FnMut(&mut Shard, T), + ) { let change = self.begin_change(); - for run in placed.chunk_by(|a, b| self.shard_of(a.0) == self.shard_of(b.0)) { - let mut shard = write(&self.shards[self.shard_of(run[0].0)]); + let mut news = false; + for (shard, part) in parts { + let mut shard = write(shard); shard.journals.set_stamp(change.stamp); let before = shard.journals.mark(); - apply(&mut shard, run); + apply(&mut shard, part); shard.tidy_journals(); news |= shard.journals.has_news_since(before); } + drop(change); if news { self.wake_readers(); } @@ -639,12 +635,17 @@ impl RibTable { /// Takes a time on the clock for a change that is about to be made, /// and notes that it is under way until the guard drops. fn begin_change(&self) -> Change<'_> { + let version = self.version.load(std::sync::atomic::Ordering::Acquire); let stamp = self.clock.fetch_add(1, std::sync::atomic::Ordering::AcqRel) + 1; self.in_progress .lock() .unwrap_or_else(PoisonError::into_inner) .insert(stamp); - Change { table: self, stamp } + Change { + table: self, + stamp, + version, + } } /// The time on the clock up to which every change is complete in every @@ -713,13 +714,6 @@ impl RibTable { } } - /// Up to `limit` routes: every route of as many prefixes as fit, in no - /// particular order. For showing some of a table that is too large to - /// show all of. - pub fn some_routes(&self, limit: usize) -> Vec { - self.routes_where(limit, |_| true) - } - /// Up to `limit` of the routes that `wanted` picks, in no particular /// order. This looks at routes until it has enough, which for a choosy /// `wanted` is every route in the table: it is for looking into the @@ -781,6 +775,9 @@ fn write(shard: &RwLock) -> RwLockWriteGuard<'_, Shard> { struct Change<'a> { table: &'a RibTable, stamp: u64, + /// The readers' version when it began, to tell whether they were + /// woken while it was under way. + version: u64, } impl Drop for Change<'_> { @@ -790,6 +787,17 @@ impl Drop for Change<'_> { .lock() .unwrap_or_else(PoisonError::into_inner) .remove(&self.stamp); + // A reader does not read past a change still under way. One that + // another change woke meanwhile found nothing it could read yet, + // and is woken again now that it can. + if self + .table + .version + .load(std::sync::atomic::Ordering::Acquire) + != self.version + { + self.table.wake_readers(); + } } }