session: write messages in batches, with TCP_NODELAY

A session wrote every message to its socket as it came: one write and,
with Nagle's algorithm, mostly one segment for each UPDATE, a hundred or
two octets. On the IPv6 cross-connect between the two routers, whose
round trip is 50 microseconds, a million routes took 42 seconds to go
out: the status page's new TCP row showed the connection busy for that
long with neither window nor buffer limiting it, a congestion window
stuck at ten segments, and a smoothed round-trip time of 18 ms. That is
the receiver's delayed acknowledgements, 40 ms at a time, on a stream
of small segments, holding the sender to a few segments a round trip.

Now a session gathers what is waiting for it, up to the outbound queue's
length, in one wake, encodes the messages into one buffer, and writes
the buffer when it reaches 64 KiB or when the actions of the wake are
done; the socket has TCP_NODELAY, so a batch's last, partial segment is
not held back for the acknowledgement of the one before. The
NOTIFICATION that ends a session, including one sent after a panic, is
flushed on its own.

Measured in the Linux VM with four peers' tables going to a collector
over a veth pair, 151 MiB: 121,850 segments instead of 313,773, the
connection busy 0.8 s instead of 2.0 s, and the congestion window grown
to 47 segments instead of 13. Over loopback the transit benchmark is
unchanged at 2.5 s for the second neighbor to have the table: there the
receiver's parsing sets the pace, not the segments.

Testing: the session tests, including the one that sends a NOTIFICATION
after a panic; the integration tests; the kernel tests and every lab
phase.

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

crates/server/src/connection.rs

@@ -60,6 +60,13 @@ use crate::throttle::Throttle;
 /// senders are back-pressured.
 const OUTBOUND_QUEUE_CAPACITY: usize = 1024;
 
+/// 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
+/// UPDATEs; written one at a time, each was a segment of its own, and a
+/// receiver that acknowledges late held the sender at a few segments a
+/// round trip. Gathered, they go as full segments in one write.
+const WRITE_BATCH_BYTES: usize = 64 * 1024;
+
 /// The least time a write to the peer may make no progress before the
 /// session is given up: the Send Hold Timer of RFC 9687, which is this or
 /// twice the hold time, whichever is longer. A peer that has stopped
@@ -595,6 +602,9 @@ struct Validation {
 enum Wake {
     /// Something for the state machine.
     Event(Event),
+    /// Messages from the rest of the server to send, as many as were
+    /// waiting, to be written together.
+    Outbound(Vec<Message>),
     /// The validation data has changed; these routes of the peer's are
     /// now invalid.
     Revalidate(Vec<Nlri>),
@@ -671,6 +681,10 @@ impl Session {
         peer: PeerSettings,
     ) -> io::Result<(Self, Vec<Action>)> {
         let local_addr = stream.local_addr()?;
+        // Messages are written in batches of full segments (see
+        // `WRITE_BATCH_BYTES`); a batch's last, partial segment is not to
+        // wait for the acknowledgement of the one before it.
+        let _ = stream.set_nodelay(true);
         let (reader, writer) = stream.into_split();
         let (fsm, first_actions) = Fsm::new(FsmConfig {
             router_id: server.router_id,
@@ -728,8 +742,15 @@ impl Session {
         error!(peer_addr = %socket_text(self.peer_addr), err = %error, "Closing session after a panic");
         if let Some(notification) = error.notification() {
             let notification = Message::Notification(notification);
-            // Contained as well: whatever panicked may do so again.
-            if !matches!(contained(self.send(&notification)).await, Ok(Ok(()))) {
+            // Whatever was gathered to be written when the panic came is
+            // not to go; the NOTIFICATION is, now. Contained as well:
+            // whatever panicked may do so again.
+            self.write_buf.clear();
+            let sent = async {
+                self.send(&notification).await?;
+                self.flush().await
+            };
+            if !matches!(contained(sent).await, Ok(Ok(()))) {
                 warn!(peer_addr = %socket_text(self.peer_addr), "Could not send NOTIFICATION after a panic");
             }
         }
@@ -758,6 +779,10 @@ impl Session {
             }
             actions = match self.next_event().await? {
                 Wake::Event(event) => self.fsm.handle(event),
+                Wake::Outbound(messages) => messages
+                    .into_iter()
+                    .flat_map(|msg| self.fsm.handle(Event::Outbound(msg)))
+                    .collect(),
                 Wake::Revalidate(suspects) => self.revalidate(&suspects),
                 Wake::Swept(outcome) => self.swept(&outcome),
                 Wake::StaleTimeout => {
@@ -826,8 +851,19 @@ impl Session {
     }
 
     /// Carries out `actions` in order. Returns the session's result once it
-    /// is over.
+    /// is over. What the actions send is written once they are all done,
+    /// or as each batch of `WRITE_BATCH_BYTES` fills.
     async fn execute(&mut self, actions: Vec<Action>) -> Option<Result<(), SessionError>> {
+        let result = self.execute_all(actions).await;
+        // Written whatever came of the actions: a NOTIFICATION that ends
+        // the session is the last thing the peer is to get.
+        match self.flush().await {
+            Ok(()) => result,
+            Err(e) => Some(result.unwrap_or(Err(e))),
+        }
+    }
+
+    async fn execute_all(&mut self, actions: Vec<Action>) -> Option<Result<(), SessionError>> {
         let mut queue: VecDeque<Action> = actions.into();
         while let Some(action) = queue.pop_front() {
             match action {
@@ -1285,6 +1321,8 @@ impl Session {
         vec![]
     }
 
+    /// Puts `msg` in the write buffer, and writes the buffer out if it
+    /// has a batch's worth; the rest goes with [`Session::flush`].
     async fn send(&mut self, msg: &Message) -> Result<(), SessionError> {
         if let (Message::Update(update), Some(established)) = (msg, &self.established) {
             let (withdrawn, announced) = prefix_counts(update);
@@ -1294,17 +1332,32 @@ impl Session {
             info!(peer_addr = %socket_text(self.peer_addr), error = %notification.error, "Sending NOTIFICATION, closing session");
             self.notified = true;
         }
-        self.write_buf.clear();
-        msg.to_wire(&self.send_ctx, &mut self.write_buf)?;
-        match timeout(self.send_hold_time, self.writer.write_all(&self.write_buf)).await {
-            Ok(written) => written?,
-            Err(_) => return Err(SessionError::SendHoldTimerExpired(self.send_hold_time)),
+        let before = self.write_buf.len();
+        if let Err(e) = msg.to_wire(&self.send_ctx, &mut self.write_buf) {
+            self.write_buf.truncate(before);
+            return Err(e.into());
         }
         MESSAGES_SENT
             .with_label_values(&[message_type_name(msg)])
             .inc();
+        if self.write_buf.len() >= WRITE_BATCH_BYTES {
+            self.flush().await?;
+        }
         Ok(())
     }
+
+    /// Writes what is in the write buffer to the peer.
+    async fn flush(&mut self) -> Result<(), SessionError> {
+        if self.write_buf.is_empty() {
+            return Ok(());
+        }
+        let written = timeout(self.send_hold_time, self.writer.write_all(&self.write_buf)).await;
+        self.write_buf.clear();
+        match written {
+            Ok(written) => Ok(written?),
+            Err(_) => Err(SessionError::SendHoldTimerExpired(self.send_hold_time)),
+        }
+    }
 }
 
 // ---------------------------------------------------------------------------
@@ -1379,7 +1432,16 @@ async fn from_server(established: &mut Option<Established>) -> Wake {
     };
     tokio::select! {
         Some(error) = ending.recv() => Wake::Event(Event::LocalFailure(error)),
-        Some(msg) = outbound.recv() => Wake::Event(Event::Outbound(msg)),
+        Some(msg) = outbound.recv() => {
+            // And whatever else is waiting, to go out together.
+            let mut messages = vec![msg];
+            while messages.len() < OUTBOUND_QUEUE_CAPACITY
+                && let Ok(more) = outbound.try_recv()
+            {
+                messages.push(more);
+            }
+            Wake::Outbound(messages)
+        }
         Some(suspects) = invalidated => Wake::Revalidate(suspects),
         Some(outcome) = swept.reported.recv() => Wake::Swept(outcome),
         () = stale_timeout => Wake::StaleTimeout,

scripts/profile/full-table.sh

@@ -195,6 +195,9 @@ if [ "$collector" = 1 ]; then
         last=$now_held
     done
     echo "collector: $(advertised) routes advertised after $(since "$start") s"
+    collector_addr=$(status peers '.[] | select(.name == "collector") | .peer_addr')
+    run curl -sf "http://127.0.0.1:8080/api/tcp?peer=$collector_addr" \
+        | jq -r '.connection | "collector tcp: busy \(.busy_ms / 1000 | . * 10 | round / 10) s, rwnd-limited \(.limited_by_receive_window_share * 100 | round)%, rtt \(.rtt_ms) ms, cwnd \(.cwnd_segments), sent \(.bytes_sent / 1048576 | round) MiB in \(.segments_out) segments"'
 fi
 memory_report() {
     rss=$(awk '/VmRSS/ { printf "%.0f", $2 / 1024 }' "/proc/$daemon_pid/status")