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