Routing table: a reader waits out a change under way instead of spinning

A reader does not read past the earliest change still under way, which
may not have reached every shard. A change to every shard, as when a
peer's whole table goes, is under way for a while; changes others made
meanwhile are beyond it. A reader that found only those saw "more to
read" with nothing it could read, and asked again at once, round and
round, taking every shard's read lock each time, until the change ended:
every exporter and the kernel installer, spinning for as long as a full
table took to remove.

Now a pass that moves no cursor though there is more says it is caught
up, and waits. And a change wakes the readers when it ends if any other
change woke them while it was under way, since what they were woken for
is only readable now. The three per-shard change loops are one
(RibTable::change), which drops the change before waking, as only one of
them did.

Also: the add_peer doc was attached to peer(); some_routes had no
callers; count_stale is for tests.

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>

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,
+        }
     }
 }
 

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.

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<PeerInfo> {
@@ -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<PeerIndex, TooManyPeers> {
         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<N>(&self, routes: &[N], mut apply: impl FnMut(&mut Shard, &[(u64, Nlri)]))
+    fn by_shard<N>(&self, routes: &[N], apply: impl FnMut(&mut Shard, &[(u64, Nlri)]))
     where
         N: Copy + Into<Nlri>,
     {
@@ -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<Item = (&'a RwLock<Shard>, 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<Route> {
-        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<Shard>) -> 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();
+        }
     }
 }