diff --git a/crates/server/src/connection.rs b/crates/server/src/connection.rs index 15a650b..914a419 100644 --- a/crates/server/src/connection.rs +++ b/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), /// The validation data has changed; these routes of the peer's are /// now invalid. Revalidate(Vec), @@ -671,6 +681,10 @@ impl Session { peer: PeerSettings, ) -> io::Result<(Self, Vec)> { 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(¬ification)).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(¬ification).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) -> Option> { + 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) -> Option> { let mut queue: VecDeque = 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) -> 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, diff --git a/scripts/profile/full-table.sh b/scripts/profile/full-table.sh index 46e46cb..db0ad8a 100755 --- a/scripts/profile/full-table.sh +++ b/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")