From a15efaa26c33aa7dbcb8094d5155cd049c757203 Mon Sep 17 00:00:00 2001 From: Rayhaan Jaufeerally Date: Sat, 10 Oct 2026 15:43:48 +0200 Subject: [PATCH] export: give the leave to gather back once the table is gathered The last change held a session's leave to gather a table, of which there are four, until its table had gone out. A peer that reads at the pace of its receive window takes hours over a table, and two such peers held two of the four for that long, keeping other sessions from gathering theirs: they waited for a leave before their first UPDATE. The leave is for the memory a gather takes; what is still held once the table is gathered is let go of as it goes out, as it was before the gather was reworked. The mesh benchmark's new peer can also come once the first router has its own peers' routes and the other router's are still arriving (BENCH_RECEIVER_EARLY=2), as a router's sessions come up after a restart while its peers' tables arrive; the walk of a full table while the table fills costs the arriving routes nothing measurable in the VM. Co-Authored-By: Claude Fable 5.1 --- crates/bench/benches/mesh.rs | 40 +++++++++++++++++++++++++--------- crates/server/src/advertise.rs | 6 +++-- 2 files changed, 34 insertions(+), 12 deletions(-) diff --git a/crates/bench/benches/mesh.rs b/crates/bench/benches/mesh.rs index b9a8280..33ad328 100644 --- a/crates/bench/benches/mesh.rs +++ b/crates/bench/benches/mesh.rs @@ -17,7 +17,10 @@ //! second; `BENCH_SCALE_DOWN` shrinks the table. The routers peer over //! IPv6 loopback and the feeders come over IPv4 loopback; the new peer //! comes from 127.0.0.2, which Linux has and macOS does not, so that -//! part is skipped on macOS. +//! part is skipped on macOS. With `BENCH_RECEIVER_EARLY=1` the new peer +//! comes as the feeders start instead, so that its exporter goes +//! through the table while the table is being filled, as a router's +//! do after a restart. use std::net::{Ipv4Addr, Ipv6Addr, SocketAddr}; use std::sync::Arc; @@ -32,6 +35,7 @@ use bgp_packet::open::{AddPathEntry, Capability, SendReceive}; use bgp_server::config::PeerConfig; use bgp_server::family::SUPPORTED; use bgp_server::policy::{ExportPolicy, ImportPolicy}; +use bgp_server::session::SessionRegistry; use tokio::io::{AsyncReadExt, AsyncWriteExt, BufReader}; /// An AS number that is in none of the synthetic paths. @@ -158,7 +162,21 @@ fn main() { let own_a = per_peer * peers_a as usize; let own_b = per_peer * peers_b as usize; + let receiver_routes = shape.ipv4_prefixes * peers_a as usize; let started = Instant::now(); + // Early: as the feeders start, or (2) once A has its own peers' + // routes and B's are still coming, as a router's sessions come + // up after a restart while its peers' tables arrive. + let when = env("BENCH_RECEIVER_EARLY", 0); + let early_receiver = (when > 0).then(|| { + let (addr, registry, rib) = (a.addr, a.registry.clone(), Arc::clone(&a.rib)); + tokio::spawn(async move { + while when == 2 && rib.len() < own_a { + tokio::time::sleep(Duration::from_millis(5)).await; + } + resync(addr, registry, receiver_routes).await + }) + }); let mut sending = Vec::new(); for (mut session, stream) in feed_a.into_iter().chain(feed_b) { sending.push(tokio::spawn(async move { @@ -235,20 +253,24 @@ fn main() { for session in sending { feeders.push(session.await.expect("a feeder is done")); } - resync(&a, shape.ipv4_prefixes * peers_a as usize).await; + match early_receiver { + Some(receiver) => receiver.await.expect("the receiver is done"), + None => resync(a.addr, a.registry.clone(), receiver_routes).await, + } drop(feeders); }); } -/// A new peer takes router `a`'s table, `routes` routes with every path: -/// how long until the first UPDATE, and until all of it is sent. -async fn resync(a: &TestServer, routes: usize) { +/// A new peer takes the table of the router at `addr`, whose sessions +/// are in `registry`: `routes` routes with every path. How long until +/// the first UPDATE, and until all of it is sent. +async fn resync(addr: SocketAddr, registry: SessionRegistry, routes: usize) { let socket = tokio::net::TcpSocket::new_v4().expect("a socket"); if let Err(e) = socket.bind(SocketAddr::new(RECEIVER.into(), 0)) { println!("A new peer taking A's routes: skipped, {RECEIVER} is not ours ({e})"); return; } - let stream = socket.connect(a.addr).await.expect("A listens"); + let stream = socket.connect(addr).await.expect("A listens"); let mut session = TestPeer::from_stream(stream); let mut capabilities = vec![Capability::FourByteAsn { asn: ASN }]; capabilities.extend(SUPPORTED.into_iter().map(Capability::multiprotocol)); @@ -328,8 +350,7 @@ async fn resync(a: &TestServer, routes: usize) { let deadline = Duration::from_secs(env("BENCH_DEADLINE", 600) as u64); let advertised = loop { tokio::time::sleep(Duration::from_millis(5)).await; - let advertised = a - .registry + let advertised = registry .list() .iter() .find(|s| s.peer.addr.ip() == RECEIVER) @@ -338,8 +359,7 @@ async fn resync(a: &TestServer, routes: usize) { break advertised; } if started.elapsed() > deadline { - let sessions: Vec = a - .registry + let sessions: Vec = registry .list() .iter() .map(|s| { diff --git a/crates/server/src/advertise.rs b/crates/server/src/advertise.rs index 9091d1f..36a2c14 100644 --- a/crates/server/src/advertise.rs +++ b/crates/server/src/advertise.rs @@ -955,8 +955,10 @@ impl Exporter { let Some(gathered) = learned.gathering.take() else { return Ok(()); }; - // Held until the gathered routes are let go of. - let _permit = learned.permit.take(); + // The leave to gather is given back now that the table is all + // there: what is still held is let go of as it goes out, and a + // peer that reads slowly must not keep the others from gathering. + learned.permit = None; let Gathered { groups, .. } = gathered; let egress = Self::egress(session, self.control, self.cluster_id, towards); // What the peer would hold once this goes out: none of it goes if