export: a lighter gather, and UPDATEs to the session by the piece

Two things kept an exporter from its peer. A table being gathered to
go out anew kept a map of every prefix heard, to let a later word on a
prefix supersede an earlier one, and dressed every set of attributes
before the first UPDATE went: a map insert and a parse-and-encode a
route that the peer waited on. And the session's queue took UPDATEs
one by one, a thousand at most, so the exporter stood at the queue
once a hundred kilobytes were in it, behind a writer that takes a
megabyte: the two took turns instead of running together.

The journal gives a table out shard by shard, each prefix once, and a
change meanwhile comes after the table as news of its own, so there is
no second word to keep track of while gathering: the map is gone, and
a prefix the peer is not to have is simply not gathered, as what it
held of it is withdrawn once the table is through. Each set is dressed
as its turn to go comes, so the first UPDATE goes once the table is
gathered; only when the gathered routes alone would be over the
peer's limit are the sets dressed first, to know which cannot be and
take their routes with them, so that none goes if too many would. The
UPDATEs for a set now go to the session in pieces of 64 KiB or so,
each of whole messages, and the queue takes 64 of those: a few
megabytes ahead of the writer, not a hundred kilobytes. The session
counts the messages in a piece by their headers.

Testing: the byte-for-byte test of the encoded UPDATEs now encodes
thirty thousand routes and checks the pieces are of whole messages
and of the size meant. The server tests, the integration tests, the
kernel tests and every lab phase pass. In the Linux VM, three rounds
alternating with the previous commit, a new peer's first UPDATE comes
1.8 s after its session is up instead of 3.1 s, and it has its four
million routes in 3.8 s instead of 4.1 s; the smaller of the two
routers' tables crosses in 3.0 s instead of 4.2 s and the larger in
5.0 s instead of 5.2 s, at the same CPU time.

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

crates/server/src/advertise.rs

@@ -337,15 +337,13 @@ type Verdicts = HashMap<IpPrefix, Verdict>;
 
 /// What is gathered while the table is given out anew, to go out once it
 /// is all there: the routes by what dresses them, so that those that go
-/// out alike go in one UPDATE wherever in the table they are; and the
+/// out alike go in one UPDATE wherever in the table they are. Nothing is
-/// last word on each prefix, which is the one that counts. Nothing is
+/// dressed until it goes. The table is given out shard by shard, each
-/// dressed until it goes.
+/// prefix once, so there is no second word on a prefix to keep track of:
+/// a change meanwhile comes after, as news of its own.
 #[derive(Default)]
 struct Gathered {
     groups: HashMap<DressedKey, Group>,
-    /// Each prefix heard of: which word on it is the latest, and how
-    /// many routes that word gives it.
-    heard: HashMap<IpPrefix, Heard>,
     /// Who the routes are from, asked of the table once for each peer.
     sources: HashMap<PeerIndex, Option<Source>>,
 }
@@ -358,41 +356,15 @@ struct Group {
     routes: Vec<Entry>,
 }
 
-/// A route gathered: of which prefix, which of its routes, the path
+/// A route gathered: of which prefix, which of its routes, and the path
-/// identifier it is to go out with where routes are told apart, and
+/// identifier it is to go out with where routes are told apart.
-/// under which word on the prefix.
 struct Entry {
     prefix: IpPrefix,
     key: PathKey,
     id: u32,
-    word: u32,
-}
-
-#[derive(Clone, Copy, Default)]
-struct Heard {
-    word: u32,
-    routes: u32,
-}
-
-/// A group of gathered routes dressed and ready to go: those of its
-/// routes that count.
-struct Ready {
-    version: IpVersion,
-    attrs: Arc<AttrSet>,
-    dressed: Arc<Dressed>,
-    routes: Vec<Entry>,
 }
 
 impl Gathered {
-    /// A new word on `prefix`, giving it `routes` routes: whatever was
-    /// gathered for it before no longer counts. Returns the word.
-    fn word(&mut self, prefix: IpPrefix, routes: u32) -> u32 {
-        let heard = self.heard.entry(prefix).or_default();
-        heard.word += 1;
-        heard.routes = routes;
-        heard.word
-    }
-
     /// Gathers a route with `attrs` from `source`.
     fn add(&mut self, attrs: &Arc<AttrSet>, source: Source, version: IpVersion, entry: Entry) {
         let key = (Arc::as_ptr(attrs) as usize, source, version);
@@ -886,7 +858,8 @@ impl Exporter {
     /// Gathers one piece of news on a table that is being given out anew:
     /// the routes by what they go out as, so that those that go out alike
     /// go together wherever in the table they are, and nothing is dressed
-    /// yet. A prefix heard of again supersedes what was heard before.
+    /// yet. A prefix the peer is not to have is not gathered: what it
+    /// held of it is withdrawn once the table is through.
     fn gather(&mut self, item: News) {
         let session = &self.session;
         let Some(Learned {
@@ -908,34 +881,27 @@ impl Exporter {
         match item {
             // Again, before it was all there: from the start.
             News::Reset => *gathered = Gathered::default(),
-            News::Synced => {}
+            News::Synced | News::Withdrawn(_) => {}
-            News::Withdrawn(prefix) => {
-                gathered.word(prefix, 0);
-            }
             News::Best {
                 prefix,
                 peer,
                 attrs,
-            } => match source_of(&self.rib, &mut gathered.sources, peer) {
+            } => {
-                Some(source) if may_go(&prefix) => {
+                // The peer has gone, and its routes with it.
-                    let word = gathered.word(prefix, 1);
+                if let Some(source) = source_of(&self.rib, &mut gathered.sources, peer)
+                    && may_go(&prefix)
+                {
                     let entry = Entry {
                         prefix,
                         key: (peer, 0),
                         id: 0,
-                        word,
                     };
                     gathered.add(&attrs, source, prefix.address_family(), entry);
                 }
-                // The peer has gone, and its routes with it.
+            }
-                _ => {
-                    gathered.word(prefix, 0);
-                }
-            },
             News::Routes { prefix, routes } => {
                 let version = prefix.address_family();
                 if !may_go(&prefix) {
-                    gathered.word(prefix, 0);
                     return;
                 }
                 // Where the peer did not agree to tell the routes of a
@@ -949,7 +915,6 @@ impl Exporter {
                             .map(|source| ((peer, path_id), attrs, source))
                     })
                     .collect();
-                let word = gathered.word(prefix, wanted.len() as u32);
                 // A route the peer had keeps its identifier; a new one
                 // gets the lowest not in use for the prefix, in the order
                 // the table has the routes.
@@ -970,23 +935,17 @@ impl Exporter {
                             .expect("fewer routes than identifiers"),
                     };
                     taken.push(id);
-                    let entry = Entry {
+                    gathered.add(&attrs, source, version, Entry { prefix, key, id });
-                        prefix,
-                        key,
-                        id,
-                        word,
-                    };
-                    gathered.add(&attrs, source, version, entry);
                 }
             }
         }
     }
 
     /// Sends the peer the table gathered since the reset, now that it is
-    /// all there: each set of routes that go out alike is dressed once
+    /// all there: each set of routes that go out alike is dressed as its
-    /// and goes, and what was announced before and not heard of again is
+    /// turn comes and goes, and what was announced before and not heard
-    /// withdrawn after. Nothing goes if the peer would then hold more
+    /// of again is withdrawn after. Nothing goes if the peer would then
-    /// than it may. The routes are counted as they go.
+    /// hold more than it may. The routes are counted as they go.
     async fn stream(&mut self) -> Result<(), Finished> {
         let towards = self.towards();
         let session = &self.session;
@@ -998,49 +957,35 @@ impl Exporter {
         };
         // Held until the gathered routes are let go of.
         let _permit = learned.permit.take();
-        let Gathered { groups, heard, .. } = gathered;
+        let Gathered { groups, .. } = gathered;
         let egress = Self::egress(session, self.control, self.cluster_id, towards);
-        // Dressed, each set once, and down to the routes that count:
-        // the latest word on their prefix.
-        let mut ready: Vec<Ready> = Vec::with_capacity(groups.len());
-        let mut will_hold = 0;
-        for (n, ((_, _, version), group)) in groups.into_iter().enumerate() {
-            // Millions, for the Internet: the thread is lent to the other
-            // tasks on the way through.
-            if n % EXPORT_BATCH == EXPORT_BATCH - 1 {
-                tokio::task::yield_now().await;
-            }
-            let Some(dressed) = learned.dressed(&group.attrs, group.source, version, &egress)
-            else {
-                continue;
-            };
-            let routes: Vec<Entry> = group
-                .routes
-                .into_iter()
-                .filter(|entry| {
-                    heard
-                        .get(&entry.prefix)
-                        .is_some_and(|heard| heard.word == entry.word)
-                })
-                .collect();
-            if routes.is_empty() {
-                continue;
-            }
-            will_hold += routes.len();
-            ready.push(Ready {
-                version,
-                attrs: group.attrs,
-                dressed,
-                routes,
-            });
-        }
-        drop(heard);
         // What the peer would hold once this goes out: none of it goes if
-        // that is more than it may.
+        // that is more than it may. Gathered routes are at most that
-        let routes = self.own_announced.len() + will_hold;
+        // many; only if those are over the limit is each set dressed
+        // first, to know which cannot be and take their routes with them.
         let limits = session.export_limits.as_deref();
-        if let Some(limit) = limits.and_then(|limits| limits.exceeded_by(routes)) {
+        let at_most: usize = groups.values().map(|group| group.routes.len()).sum();
-            return Err(over_limit(session, limit, routes));
+        if limits.is_some_and(|limits| {
+            limits
+                .exceeded_by(self.own_announced.len() + at_most)
+                .is_some()
+        }) {
+            let mut will_hold = 0;
+            for (n, ((_, _, version), group)) in groups.iter().enumerate() {
+                if n % EXPORT_BATCH == EXPORT_BATCH - 1 {
+                    tokio::task::yield_now().await;
+                }
+                if learned
+                    .dressed(&group.attrs, group.source, *version, &egress)
+                    .is_some()
+                {
+                    will_hold += group.routes.len();
+                }
+            }
+            let routes = self.own_announced.len() + will_hold;
+            if let Some(limit) = limits.and_then(|limits| limits.exceeded_by(routes)) {
+                return Err(over_limit(session, limit, routes));
+            }
         }
         let contexts = [IpVersion::Ipv4, IpVersion::Ipv6].map(|version| ParserContext {
             four_octet_asn: towards.four_octet_asn,
@@ -1052,13 +997,14 @@ impl Exporter {
             },
         });
         let mut entries = 0;
-        for Ready {
+        for ((_, _, version), group) in groups {
-            version,
+            // Dressed as its turn comes: the peer is reading the sets
-            attrs,
+            // before it meanwhile.
-            dressed,
+            let Some(dressed) = learned.dressed(&group.attrs, group.source, version, &egress)
-            routes,
+            else {
-        } in ready
+                continue;
-        {
+            };
+            let Group { attrs, routes, .. } = group;
             let mut nlris = Vec::with_capacity(routes.len());
             for entry in routes {
                 entries += 1;

crates/server/src/connection.rs

@@ -60,9 +60,10 @@ use crate::session::{
 };
 use crate::throttle::Throttle;
 
-/// Number of outbound messages that may be queued for a session before
+/// How many items may be queued for a session before senders wait: each
-/// senders are back-pressured.
+/// a message, or a piece of up to [`export::ENCODED_PIECE`] octets of
-const OUTBOUND_QUEUE_CAPACITY: usize = 1024;
+/// UPDATEs from the exporter, so a few megabytes of a table at most.
+const OUTBOUND_QUEUE_CAPACITY: usize = 64;
 
 /// How many octets of messages are gathered before they are written to
 /// the socket, when more are waiting. A full table is a million small
@@ -1415,10 +1416,20 @@ impl Session {
         Ok(())
     }
 
-    /// Puts an UPDATE that is already encoded in the write buffer.
+    /// Puts a piece of UPDATEs that are already encoded in the write
-    fn send_encoded(&mut self, update: BytesMut) {
+    /// buffer.
-        self.write_buf.extend_from_slice(&update);
+    fn send_encoded(&mut self, piece: BytesMut) {
-        MESSAGES_SENT.with_label_values(&["update"]).inc();
+        // Counted by their headers: a piece is of whole messages.
+        let mut messages = 0;
+        let mut at = 0;
+        while at + 18 <= piece.len() {
+            at += usize::from(u16::from_be_bytes([piece[at + 16], piece[at + 17]])).max(19);
+            messages += 1;
+        }
+        self.write_buf.extend_from_slice(&piece);
+        MESSAGES_SENT
+            .with_label_values(&["update"])
+            .inc_by(messages);
         if self.write_buf.len() >= WRITE_BATCH_BYTES {
             self.flush();
         }

crates/server/src/export.rs

@@ -453,11 +453,17 @@ impl EncodedAttributes {
 /// and a little to spare.
 const NEXT_HOP_ROOM: usize = 64;
 
+/// How many octets of UPDATEs go to a session in one piece: what the
+/// session hands its writer at a time. A piece holds whole messages,
+/// so it runs over by at most one.
+pub(crate) const ENCODED_PIECE: usize = 64 * 1024;
+
 /// The UPDATE messages, complete with their headers, that announce
 /// `routes`, all of IP version `version`, with the attributes `encoded`
-/// and `next_hop`: as few as they fit in. `ctx` is the session's, and
+/// and `next_hop`: as few as they fit in, in pieces of
-/// says whether path identifiers go with the family's prefixes
+/// [`ENCODED_PIECE`] octets or so, each of whole messages. `ctx` is the
-/// (RFC 7911).
+/// session's, and says whether path identifiers go with the family's
+/// prefixes (RFC 7911).
 ///
 /// What [`announcements`] builds as messages, built as bytes: the
 /// attributes are not parsed, cloned or encoded again for each message.
@@ -472,10 +478,10 @@ pub(crate) fn encoded_announcements(
     let at_once = (room / nlri_octets(version)).max(1);
     let in_body = matches!(next_hop, NlriNextHop::Ipv4(_)) && version == IpVersion::Ipv4;
     let with_path_id = ctx.add_path.contains(&AfiSafi::unicast(version));
-    routes
+    let mut pieces = Vec::new();
-        .chunks(at_once)
+    let mut out = BytesMut::with_capacity(ENCODED_PIECE.min(routes.len() * 64 + 4096));
-        .map(|routes| {
+    for routes in routes.chunks(at_once) {
-            let mut out = BytesMut::with_capacity(4096);
+        {
             out.put_slice(&BGP_MARKER);
             let length_at = out.len();
             out.put_u16(0);
@@ -511,11 +517,17 @@ pub(crate) fn encoded_announcements(
             if in_body {
                 write_nlri(routes, with_path_id, &mut out);
             }
-            let length = out.len();
+            let length = out.len() - length_at + BGP_MARKER.len();
             out[length_at..length_at + 2].copy_from_slice(&(length as u16).to_be_bytes());
-            out
+        }
-        })
+        if out.len() >= ENCODED_PIECE {
-        .collect()
+            pieces.push(out.split());
+        }
+    }
+    if !out.is_empty() {
+        pieces.push(out);
+    }
+    pieces
 }
 
 /// The UPDATEs that announce `routes`, all of IP version `version`, as
@@ -710,7 +722,7 @@ mod tests {
                             }],
                         }),
                     ];
-                    let routes: Vec<Nlri> = (0..1000u32)
+                    let routes: Vec<Nlri> = (0..30_000u32)
                         .map(|i| Nlri {
                             prefix: match version {
                                 IpVersion::Ipv4 => IpPrefix::new(
@@ -762,11 +774,33 @@ mod tests {
                         "{version:?}, path ids {with_path_id}, four-octet {four_octet_asn}"
                     );
                     assert!(expected.len() > 1, "{what}: more than one message");
-                    assert_eq!(bytes.len(), expected.len(), "{what}: as many messages");
+                    for update in &expected {
-                    for (n, (got, want)) in bytes.iter().zip(&expected).enumerate() {
+                        assert!(update.len() <= 4096, "{what}: a message fits");
-                        assert!(got.len() <= 4096, "{what}: message {n} fits");
-                        assert_eq!(&got[..], &want[..], "{what}: message {n}");
                     }
+                    // In pieces of whole messages, each of about the size
+                    // meant, all but the last at least that.
+                    assert!(bytes.len() > 1, "{what}: more than one piece");
+                    for (n, piece) in bytes.iter().enumerate() {
+                        assert!(
+                            n == bytes.len() - 1
+                                || (ENCODED_PIECE..ENCODED_PIECE + 4096).contains(&piece.len()),
+                            "{what}: piece {n} is {} octets",
+                            piece.len()
+                        );
+                        let mut at = 0;
+                        while at < piece.len() {
+                            let length =
+                                usize::from(u16::from_be_bytes([piece[at + 16], piece[at + 17]]));
+                            at += length;
+                        }
+                        assert_eq!(at, piece.len(), "{what}: piece {n} is of whole messages");
+                    }
+                    let got: Vec<u8> = bytes
+                        .iter()
+                        .flat_map(|piece| piece.iter().copied())
+                        .collect();
+                    let want: Vec<u8> = expected.iter().flat_map(|m| m.iter().copied()).collect();
+                    assert_eq!(got, want, "{what}: the same octets");
                 }
             }
         }

crates/server/src/session.rs

@@ -318,19 +318,19 @@ impl SessionHandle {
             .map_err(|_| SessionClosed(self.peer))
     }
 
-    /// Sends the peer an UPDATE that is already encoded, header and all,
+    /// Sends the peer a piece of one or more UPDATEs that are already
-    /// as the exporter builds them.
+    /// encoded, headers and all, as the exporter builds them.
-    pub async fn send_encoded(&self, update: BytesMut) -> Result<(), SessionClosed> {
+    pub async fn send_encoded(&self, piece: BytesMut) -> Result<(), SessionClosed> {
         self.outbound
-            .send(Outbound::Encoded(update))
+            .send(Outbound::Encoded(piece))
             .await
             .map_err(|_| SessionClosed(self.peer))
     }
 }
 
-/// What the rest of the server gives a session to send: a message, or
+/// What the rest of the server gives a session to send: a message, or a
-/// an UPDATE already encoded, header and all, which goes to the peer as
+/// piece of one or more UPDATEs already encoded, headers and all, which
-/// it is once the session is established.
+/// goes to the peer as it is once the session is established.
 #[derive(Debug)]
 pub enum Outbound {
     Message(Message),