diff --git a/Cargo.lock b/Cargo.lock index 846b932..fe7136c 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -184,6 +184,7 @@ dependencies = [ "stats_alloc", "tikv-jemallocator", "tokio", + "tracing-subscriber", ] [[package]] diff --git a/crates/bench/Cargo.toml b/crates/bench/Cargo.toml index dbbeaf1..0d57d9d 100644 --- a/crates/bench/Cargo.toml +++ b/crates/bench/Cargo.toml @@ -24,6 +24,7 @@ bgp-integration-tests.workspace = true bgp-rib.workspace = true bgp-server.workspace = true tokio.workspace = true +tracing-subscriber.workspace = true [features] # Counts every allocation, to report how much memory a table holds. The diff --git a/crates/bench/benches/mesh.rs b/crates/bench/benches/mesh.rs index f483868..b9a8280 100644 --- a/crates/bench/benches/mesh.rs +++ b/crates/bench/benches/mesh.rs @@ -5,26 +5,50 @@ //! the exporter reading the table's journal while sessions fill the //! table, which nothing else exercises at this size. //! +//! Then a new peer takes the first router's table, as a collector or a +//! router coming back does: how soon its first UPDATE arrives and how +//! long the table takes are measured too. It is a peer of the routers' +//! AS over IPv4, so what it gets is the first router's IPv4 routes from +//! its own peers: the other router's are not reflected to it, and IPv6 +//! routes need an IPv6 session. +//! //! Run with `cargo bench -p bgp-bench --bench mesh`. `BENCH_PEERS` //! (default 4) feed the first router and `BENCH_PEERS_B` (default 2) the //! second; `BENCH_SCALE_DOWN` shrinks the table. The routers peer over -//! IPv6 loopback and the feeders come over IPv4 loopback. +//! 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. use std::net::{Ipv4Addr, Ipv6Addr, SocketAddr}; use std::sync::Arc; +use std::sync::atomic::{AtomicU64, Ordering}; use std::time::{Duration, Instant}; use bgp_bench::{Prefixes, Shape, announcements, encoded, env}; use bgp_integration_tests::peer_config; use bgp_integration_tests::test_peer::TestPeer; use bgp_integration_tests::test_server::TestServer; -use bgp_packet::open::SendReceive; +use bgp_packet::open::{AddPathEntry, Capability, SendReceive}; use bgp_server::config::PeerConfig; +use bgp_server::family::SUPPORTED; use bgp_server::policy::{ExportPolicy, ImportPolicy}; +use tokio::io::{AsyncReadExt, AsyncWriteExt, BufReader}; /// An AS number that is in none of the synthetic paths. const ASN: u32 = 4_200_000_000; +/// Where the peer that takes the whole table connects from. +const RECEIVER: Ipv4Addr = Ipv4Addr::new(127, 0, 0, 2); + +/// What the routers send each other, and the receiver: every path. +fn every_path() -> Arc { + Arc::new(ExportPolicy { + learned_routes: true, + all_paths: true, + ..ExportPolicy::default() + }) +} + fn free_port() -> u16 { std::net::TcpListener::bind((Ipv6Addr::LOCALHOST, 0)) .expect("a port") @@ -36,7 +60,13 @@ fn free_port() -> u16 { /// A router of the mesh, listening for feeders on IPv4 loopback and for /// the other router on IPv6 loopback at `port`, which it dials at /// `other`. -async fn router(name: &'static str, id: u8, port: u16, other: u16) -> TestServer { +async fn router( + name: &'static str, + id: u8, + port: u16, + other: u16, + with_receiver: bool, +) -> TestServer { TestServer::start_with(ASN, |c| { // Two routers of one AS must not share a BGP identifier. c.router_id = Ipv4Addr::new(192, 0, 2, id); @@ -48,17 +78,20 @@ async fn router(name: &'static str, id: u8, port: u16, other: u16) -> TestServer c.peers = vec![PeerConfig { add_path: Some(SendReceive::Both), import: ImportPolicy::permissive(), - export: Arc::new(ExportPolicy { - learned_routes: true, - all_paths: true, - ..ExportPolicy::default() - }), + export: every_path(), ..peer_config( name, SocketAddr::new(Ipv6Addr::LOCALHOST.into(), other), ASN, ) }]; + if with_receiver { + c.peers.push(PeerConfig { + add_path: Some(SendReceive::Send), + export: every_path(), + ..peer_config("receiver", SocketAddr::new(RECEIVER.into(), 179), ASN) + }); + } c.connect_retry = Duration::from_millis(200); }) .await @@ -81,8 +114,9 @@ async fn feeders( .flat_map(|m| m.iter().copied()) .collect(); let mut session = TestPeer::connect(router.addr).await.expect("a connection"); + // No hold timer: a feeder has nothing to say after its table. session - .handshake(64512 + peer, Ipv4Addr::new(10, 0, 0, peer as u8), 90) + .handshake(64512 + peer, Ipv4Addr::new(10, 0, 0, peer as u8), 0) .await .expect("a session"); sessions.push((session, stream)); @@ -91,6 +125,13 @@ async fn feeders( } fn main() { + // The routers' logs, with `BENCH_LOG` as a `RUST_LOG` filter. + if let Ok(filter) = std::env::var("BENCH_LOG") { + tracing_subscriber::fmt() + .with_env_filter(filter) + .with_writer(std::io::stderr) + .init(); + } let peers_a = env("BENCH_PEERS", 4) as u32; let peers_b = env("BENCH_PEERS_B", 2) as u32; let shape = Shape::INTERNET.scaled_down(env("BENCH_SCALE_DOWN", 1)); @@ -107,8 +148,8 @@ fn main() { .expect("a runtime"); runtime.block_on(async { let (port_a, port_b) = (free_port(), free_port()); - let a = router("b", 1, port_a, port_b).await; - let b = router("a", 2, port_b, port_a).await; + let a = router("b", 1, port_a, port_b, true).await; + let b = router("a", 2, port_b, port_a, false).await; a.wait_for_sessions(1).await.expect("the mesh session at A"); b.wait_for_sessions(1).await.expect("the mesh session at B"); @@ -188,8 +229,146 @@ fn main() { b_all, own_a, ); + // The feeders' sessions stay up until the end: a feeder going + // would have its routes withdrawn from both tables meanwhile. + let mut feeders = Vec::new(); for session in sending { - drop(session.await); + feeders.push(session.await.expect("a feeder is done")); } + resync(&a, shape.ipv4_prefixes * peers_a as usize).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) { + 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 mut session = TestPeer::from_stream(stream); + let mut capabilities = vec![Capability::FourByteAsn { asn: ASN }]; + capabilities.extend(SUPPORTED.into_iter().map(Capability::multiprotocol)); + capabilities.push(Capability::AddPath { + entries: SUPPORTED + .iter() + .map(|family| AddPathEntry { + afi: family.afi, + safi: family.safi, + send_recv: SendReceive::Receive, + }) + .collect(), + }); + session + .handshake_with(ASN, Ipv4Addr::new(192, 0, 2, 9), 90, capabilities) + .await + .expect("A takes the receiver's session"); + let started = Instant::now(); + let (reader, mut writer) = session.into_stream().into_split(); + // KEEPALIVEs, so that the session outlives the hold time. + let keepalives = tokio::spawn(async move { + let mut keepalive = [0xffu8; 19]; + keepalive[16..19].copy_from_slice(&[0, 19, 4]); + loop { + tokio::time::sleep(Duration::from_secs(20)).await; + if writer.write_all(&keepalive).await.is_err() { + break; + } + } + }); + // When the first UPDATE arrived, in microseconds; 0 until it has. + let first = Arc::new(AtomicU64::new(0)); + let updates = Arc::new(AtomicU64::new(0)); + let counting = tokio::spawn({ + let (first, updates) = (Arc::clone(&first), Arc::clone(&updates)); + async move { + let mut reader = BufReader::with_capacity(1 << 20, reader); + let mut header = [0u8; 19]; + let mut body = vec![0u8; 4096]; + while reader.read_exact(&mut header).await.is_ok() { + let length = usize::from(u16::from_be_bytes([header[16], header[17]])); + if reader.read_exact(&mut body[..length - 19]).await.is_err() { + break; + } + match header[18] { + // An UPDATE (RFC 4271 ยง4.1). + 2 => { + let _ = first.compare_exchange( + 0, + started.elapsed().as_micros() as u64, + Ordering::Relaxed, + Ordering::Relaxed, + ); + let n = updates.fetch_add(1, Ordering::Relaxed) + 1; + if n % 1_000_000 == 0 { + println!( + " the receiver has read {n} UPDATEs after {:.2} s", + started.elapsed().as_secs_f64() + ); + } + } + 3 => println!( + "the receiver was sent a NOTIFICATION, error {} subcode {}, after {} UPDATEs", + body[0], + body[1], + updates.load(Ordering::Relaxed) + ), + _ => {} + } + } + println!( + "the receiver's connection closed after {} UPDATEs", + updates.load(Ordering::Relaxed) + ); + } + }); + 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 + .list() + .iter() + .find(|s| s.peer.addr.ip() == RECEIVER) + .map_or(0, |s| s.counters.advertised()); + if advertised as usize >= routes { + break advertised; + } + if started.elapsed() > deadline { + let sessions: Vec = a + .registry + .list() + .iter() + .map(|s| { + format!( + "{} {:?} export {:?} add_path_send {:?} advertised {}", + s.peer, + s.name, + s.export, + s.add_path_send, + s.counters.advertised() + ) + }) + .collect(); + panic!( + "the receiver has {advertised} of {routes} routes and read {} UPDATEs; sessions at A: {sessions:#?}", + updates.load(Ordering::Relaxed) + ); + } + }; + let took = started.elapsed(); + counting.abort(); + keepalives.abort(); + let first = Duration::from_micros(first.load(Ordering::Relaxed)); + println!( + "{:<52} {:>8.3} s {:.2} M routes/s first UPDATE after {:.0} ms, {} UPDATEs", + "A new peer has A's IPv4 routes (A's exporter)", + took.as_secs_f64(), + advertised as f64 / took.as_secs_f64() / 1e6, + first.as_secs_f64() * 1e3, + updates.load(Ordering::Relaxed), + ); +} diff --git a/crates/bench/examples/packing.rs b/crates/bench/examples/packing.rs new file mode 100644 index 0000000..00964cb --- /dev/null +++ b/crates/bench/examples/packing.rs @@ -0,0 +1,71 @@ +//! How many UPDATEs, and how many octets, a table takes when the routes +//! that share attributes are grouped only within slices of the table +//! taken in shard order, for a few slice sizes: what streaming a table +//! out in slices costs against grouping it whole. + +use std::collections::HashMap; +use std::hash::{BuildHasher, RandomState}; +use std::path::PathBuf; + +use bgp_bench::mrt::Dump; + +fn main() { + let path = std::env::var("BENCH_MRT") + .map(PathBuf::from) + .unwrap_or_else(|_| "crates/bench/data/rrc03-bview.20260915.0000.gz".into()); + let dump = Dump::read(&path).expect("a readable dump"); + let peers = dump.largest_peers(usize::MAX); + let full: Vec = peers + .iter() + .copied() + .filter(|&p| dump.routes[p].len() > 800_000) + .collect(); + println!("{} peers with a full table", full.len()); + let hasher = RandomState::new(); + for (what, chosen) in [ + ("one full table", vec![full[0]]), + ("every full table, every path", full.clone()), + ] { + // (hash of prefix, attribute set, prefix octets) in shard order. + let mut routes: Vec<(u64, u32, usize)> = chosen + .iter() + .flat_map(|&p| dump.routes[p].iter()) + .map(|(prefix, set)| { + ( + hasher.hash_one(prefix), + *set, + 1 + usize::from(prefix.length()).div_ceil(8) + 4, + ) + }) + .collect(); + routes.sort_unstable_by_key(|r| r.0); + println!("== {what}: {} routes", routes.len()); + for slice in [4_096usize, 16_384, 65_536, 262_144, usize::MAX] { + let mut updates = 0usize; + let mut octets = 0usize; + for chunk in routes.chunks(slice.min(routes.len())) { + let mut groups: HashMap = HashMap::new(); + for &(_, set, prefix_octets) in chunk { + *groups.entry(set).or_insert(0) += prefix_octets; + } + for (set, prefix_octets) in groups { + // One UPDATE per 4096 octets of prefixes, roughly. + let attrs = dump.attribute_sets[set as usize].len(); + let messages = prefix_octets.div_ceil(4000 - attrs.min(3000)).max(1); + updates += messages; + octets += messages * (23 + attrs) + prefix_octets; + } + } + println!( + " slice {:>9}: {:>9} UPDATEs, {:>5.1} MB on the wire", + if slice == usize::MAX { + "whole".to_string() + } else { + slice.to_string() + }, + updates, + octets as f64 / 1e6 + ); + } + } +} diff --git a/crates/server/src/advertise.rs b/crates/server/src/advertise.rs index 11c6544..94d0f99 100644 --- a/crates/server/src/advertise.rs +++ b/crates/server/src/advertise.rs @@ -335,6 +335,79 @@ enum Verdict { /// What the peer is to end up with for each prefix heard of. type Verdicts = HashMap; +/// 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 +/// last word on each prefix, which is the one that counts. Nothing is +/// dressed until it goes. +#[derive(Default)] +struct Gathered { + groups: HashMap, + /// Each prefix heard of: which word on it is the latest, and how + /// many routes that word gives it. + heard: HashMap, + /// Who the routes are from, asked of the table once for each peer. + sources: HashMap>, +} + +/// The routes that share their attributes, where they are from and +/// their IP version: what one dressing serves. +struct Group { + attrs: Arc, + source: Source, + routes: Vec, +} + +/// A route gathered: of which prefix, which of its routes, the path +/// identifier it is to go out with where routes are told apart, and +/// 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, + dressed: Arc, + routes: Vec, +} + +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, source: Source, version: IpVersion, entry: Entry) { + let key = (Arc::as_ptr(attrs) as usize, source, version); + self.groups + .entry(key) + .or_insert_with(|| Group { + attrs: Arc::clone(attrs), + source, + routes: Vec::new(), + }) + .routes + .push(entry); + } +} + /// The learned routes announced to the peer: our Adj-RIB-Out for it, as /// far as it needs keeping. enum Announced { @@ -420,9 +493,9 @@ impl Announced { /// What an exporter knows of the learned routes it has passed on. struct Learned { reader: Reader, - /// The verdicts so far on a table that is being given out anew, kept - /// until the table is all there; and the leave to keep them. - gathered: Option, + /// What has been heard of a table that is being given out anew, + /// kept until the table is all there; and the leave to keep it. + gathering: Option, permit: Option, /// Whether every route goes, and not only the best of each prefix: /// the reader is then one of all routes. Only for a session on which @@ -506,7 +579,7 @@ impl Learned { fn new(reader: Reader, all_paths: bool) -> Self { Self { reader, - gathered: None, + gathering: None, permit: None, all_paths, announced: Announced::new(all_paths), @@ -750,33 +823,349 @@ impl Exporter { return Ok(()); }; let news = learned.drain(); - match self.judge(news).await { - Some((verdicts, synced)) => self.flush(verdicts, synced).await, - // Not all there yet. - None => Ok(()), + // What is heard while the table is not being given out anew is + // judged and sent together; what is heard before a reset is + // superseded by the table as it is gone through. + let mut steady = Vec::new(); + for item in news { + let gathering = self + .learned + .as_ref() + .is_some_and(|learned| learned.gathering.is_some()); + match (gathering, item) { + (true, News::Synced) => self.stream().await?, + (true, item) => self.gather(item), + (false, News::Reset) => { + steady.clear(); + self.reset().await; + } + (false, News::Synced) => {} + (false, item) => steady.push(item), + } + } + if !steady.is_empty() { + let verdicts = self.judge(steady); + self.flush(verdicts).await?; + } + Ok(()) + } + + /// The table is being given out anew, shard by shard: what was + /// announced is in doubt until it is heard of again, and what is + /// heard is gathered until the table is all there. + async fn reset(&mut self) { + let Some(learned) = &mut self.learned else { + return; + }; + if learned.synced { + // The table was changing faster than this exporter read it, + // and the journal let the oldest changes go: the whole table + // is gone through again, and what is not heard of again is + // withdrawn then. + info!( + peer = %self.session.peer, + "The table changed too fast to follow for the peer; going through it again" + ); + } + // Whatever was announced and is not heard of again by the time + // the table is through is withdrawn then. + let known = learned.announced.take(); + match &mut learned.stale { + Some(stale) => stale.absorb(known), + // Moved, not copied: a table's worth. + stale @ None => *stale = Some(known), + } + learned.gathering = Some(Gathered::default()); + // Gathering a table takes memory: only so many sessions do it at + // once, the others wait their turn. + if learned.permit.is_none() { + learned.permit = self.resyncs.clone().acquire_owned().await.ok(); } } - /// Decides what the peer is to end up with for each prefix the news - /// is about: the last word on a prefix counts. While the whole table - /// is being given out anew, which is shard by shard, the verdicts are - /// kept until it is all there, and this returns `None`: the routes - /// that share attributes are in every shard, and are only sent - /// together if gathered first. Otherwise the verdicts, and whether - /// the table has been given out whole. - async fn judge(&mut self, news: Vec) -> Option<(Verdicts, bool)> { + /// 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. + fn gather(&mut self, item: News) { + let session = &self.session; + let Some(Learned { + gathering: Some(gathered), + stale, + .. + }) = &mut self.learned + else { + return; + }; + let may_go = |prefix: &IpPrefix| { + session.families.carries(prefix) + && !self.own_announced.contains(prefix) + && session + .export_limits + .as_deref() + .is_none_or(|limits| limits.allows(prefix)) + }; + match item { + // Again, before it was all there: from the start. + News::Reset => *gathered = Gathered::default(), + News::Synced => {} + 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) => { + let word = gathered.word(prefix, 1); + 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 + // prefix apart, only the best goes, as plain BGP has it. + let told_apart = session.add_path_send.contains(AfiSafi::unicast(version)); + let wanted: Vec<(PathKey, Arc, Source)> = routes + .into_iter() + .filter(|(_, _, _, best)| told_apart || *best) + .filter_map(|(peer, path_id, attrs, _)| { + source_of(&self.rib, &mut gathered.sources, peer) + .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. + let had: &[Sent] = match stale { + Some(Announced::All { routes, .. }) => { + routes.get(&prefix).map_or(&[], Vec::as_slice) + } + _ => &[], + }; + let mut taken: Vec = Vec::with_capacity(wanted.len()); + for (key, attrs, source) in wanted { + let id = match had.iter().find(|sent| sent.key == key) { + Some(sent) => sent.id, + None => (1..) + .find(|id| { + !had.iter().any(|sent| sent.id == *id) && !taken.contains(id) + }) + .expect("fewer routes than identifiers"), + }; + taken.push(id); + let entry = Entry { + 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 + /// and goes, and what was announced before and not heard of again is + /// withdrawn after. Nothing goes if the peer would then 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; - let learned = self.learned.as_mut()?; + let Some(learned) = &mut self.learned else { + return Ok(()); + }; + let Some(gathered) = learned.gathering.take() else { + return Ok(()); + }; + // Held until the gathered routes are let go of. + let _permit = learned.permit.take(); + let Gathered { groups, heard, .. } = 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 = 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 = 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. + let routes = self.own_announced.len() + will_hold; + let limits = session.export_limits.as_deref(); + 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, + ibgp: towards.ibgp, + add_path: if session.add_path_send.contains(AfiSafi::unicast(version)) { + vec![AfiSafi::unicast(version)] + } else { + Vec::new() + }, + }); + let mut entries = 0; + for Ready { + version, + attrs, + dressed, + routes, + } in ready + { + let mut nlris = Vec::with_capacity(routes.len()); + for entry in routes { + entries += 1; + if entries % EXPORT_BATCH == 0 { + tokio::task::yield_now().await; + } + match (&mut learned.announced, &mut learned.stale) { + (Announced::Best(announced), stale) => { + if let Some(Announced::Best(stale)) = stale { + stale.remove(&entry.prefix); + } + announced.insert(entry.prefix); + nlris.push(Nlri::from(entry.prefix)); + } + (Announced::All { routes, count }, stale) => { + // What the peer has of this route, in doubt since + // the reset: it goes out again only if it changed. + let mut known = None; + if let Some(Announced::All { + routes: stale_routes, + count: stale_count, + }) = stale + && let Some(sent) = stale_routes.get_mut(&entry.prefix) + { + if let Some(at) = sent.iter().position(|sent| sent.key == entry.key) { + known = Some(sent.swap_remove(at)); + *stale_count -= 1; + } + if sent.is_empty() { + stale_routes.remove(&entry.prefix); + } + } + let has = routes.entry(entry.prefix).or_default(); + *count += 1; + if let Some(sent) = known + && Arc::ptr_eq(&sent.attrs, &attrs) + { + has.push(sent); + continue; + } + has.push(Sent { + key: entry.key, + id: entry.id, + attrs: Arc::clone(&attrs), + }); + nlris.push(Nlri { + prefix: entry.prefix, + path_id: entry.id, + }); + } + } + } + if nlris.is_empty() { + continue; + } + ROUTES_EXPORTED + .with_label_values(&["announced"]) + .inc_by(nlris.len() as u64); + let updates = export::encoded_announcements( + &dressed.encoded, + dressed.next_hop, + version, + &nlris, + &contexts[version_index(version)], + ); + for update in updates { + session.send_encoded(update).await?; + } + let advertised = self.own_announced.len() + learned.held_by_peer(); + session.counters.set_advertised(advertised); + self.advertised_metric.set(advertised as i64); + } + // Not heard of again: gone. + learned.synced = true; + if let Some(stale) = learned.stale.take() { + for (version, routes) in [IpVersion::Ipv4, IpVersion::Ipv6] + .into_iter() + .zip(stale.withdrawals()) + { + if routes.is_empty() { + continue; + } + ROUTES_EXPORTED + .with_label_values(&["withdrawn"]) + .inc_by(routes.len() as u64); + for update in export::withdrawals(version, &routes, &towards) { + session.send(Message::Update(update)).await?; + } + } + } + self.note_advertised(); + Ok(()) + } + + /// Decides what the peer is to end up with for each prefix the news + /// is about: the last word on a prefix counts. + fn judge(&mut self, news: Vec) -> Verdicts { + let towards = self.towards(); + let session = &self.session; + let mut verdicts = Verdicts::default(); + let Some(learned) = self.learned.as_mut() else { + return verdicts; + }; let limits = session.export_limits.as_deref(); let egress = Self::egress(session, self.control, self.cluster_id, towards); - let mut verdicts = learned.gathered.take().unwrap_or_default(); - let mut synced = false; // Who the routes are from, asked of the table once for each. let mut sources: HashMap> = HashMap::default(); - // Whether a learned route for `prefix` may go to the peer at all: - // what we originate ourselves is announced as that, and not as - // someone else's route for it. let may_go = |prefix: &IpPrefix| { session.families.carries(prefix) && !self.own_announced.contains(prefix) @@ -784,36 +1173,8 @@ impl Exporter { }; for item in news { match item { - News::Reset => { - if learned.synced { - // The table was changing faster than this exporter - // read it, and the journal let the oldest changes - // go: the whole table is gone through again, and - // what is not heard of again is withdrawn then. - info!( - peer = %session.peer, - "The table changed too fast to follow for the peer; going through it again" - ); - } - // Whatever was announced and is not heard of again by - // the time the table is through is withdrawn then. - let known = learned.announced.take(); - learned - .stale - .get_or_insert_with(|| Announced::new(learned.all_paths)) - .absorb(known); - verdicts.clear(); - synced = false; - // Gathering a table takes memory: only so many sessions - // do it at once, the others wait their turn. - if learned.permit.is_none() { - learned.permit = self.resyncs.clone().acquire_owned().await.ok(); - } - } - News::Synced => { - synced = true; - learned.permit = None; - } + // Dealt with by `follow`. + News::Reset | News::Synced => {} News::Withdrawn(prefix) => { verdicts.insert(prefix, learned.nothing()); } @@ -823,8 +1184,6 @@ impl Exporter { attrs, } => { let dressed = match source_of(&self.rib, &mut sources, peer) { - // What we originate ourselves is announced as - // that, and not as someone else's route for it. Some(source) if may_go(&prefix) => { learned.dressed(&attrs, source, prefix.address_family(), &egress) } @@ -857,18 +1216,12 @@ impl Exporter { } } } - if learned.permit.is_some() && !synced { - learned.gathered = Some(verdicts); - return None; - } - Some((verdicts, synced)) + verdicts } /// Sends the peer what the verdicts say: withdrawals first, then - /// announcements, routes that go out alike together. `synced` says - /// the table was given out whole, so what was not heard of again is - /// withdrawn too. - async fn flush(&mut self, verdicts: Verdicts, synced: bool) -> Result<(), Finished> { + /// announcements, routes that go out alike together. + async fn flush(&mut self, verdicts: Verdicts) -> Result<(), Finished> { let towards = self.towards(); let session = &self.session; let Some(learned) = &mut self.learned else { @@ -887,37 +1240,22 @@ impl Exporter { .1 .push(nlri); }; - match (verdict, &mut learned.announced, &mut learned.stale) { - (Verdict::Best(verdict), Announced::Best(announced), stale) => { - // Heard of again: no longer in doubt, whichever way it - // goes. - let was_stale = match stale { - Some(Announced::Best(stale)) => stale.remove(&prefix), - _ => false, - }; - match verdict { - Some(dressed) => { - announced.insert(prefix); - announce(dressed, prefix.into()); - } - None => { - if announced.remove(&prefix) || was_stale { - withdrawn.push(prefix.into()); - } + match (verdict, &mut learned.announced) { + (Verdict::Best(verdict), Announced::Best(announced)) => match verdict { + Some(dressed) => { + announced.insert(prefix); + announce(dressed, prefix.into()); + } + None => { + if announced.remove(&prefix) { + withdrawn.push(prefix.into()); } } - } - (Verdict::All(wanted), Announced::All { routes, count }, stale) => { - // What the peer has for the prefix: announced, or in - // doubt since the table was reset. + }, + (Verdict::All(wanted), Announced::All { routes, count }) => { + // What the peer has for the prefix. let mut had = routes.remove(&prefix).unwrap_or_default(); *count -= had.len(); - if let Some(Announced::All { routes, count }) = stale - && let Some(stale) = routes.remove(&prefix) - { - *count -= stale.len(); - had.extend(stale); - } // A route it had keeps its identifier, and goes out // again only if it changed; a new one gets the lowest // identifier not in use for the prefix. @@ -958,14 +1296,6 @@ impl Exporter { _ => {} } } - if synced { - learned.synced = true; - if let Some(stale) = learned.stale.take() { - for (version, routes) in withdrawn.iter_mut().zip(stale.withdrawals()) { - version.extend(routes); - } - } - } // What the peer would hold once this goes out: none of it goes if // that is more than it may. let routes = self.own_announced.len() + learned.held_by_peer(); @@ -1949,6 +2279,102 @@ mod tests { ); } + /// A table given out anew to a new session goes out with the routes + /// that share their attributes in one UPDATE, wherever in the table + /// they are: the table is gathered by what dresses its routes before + /// any of it goes. + #[tokio::test] + async fn a_table_given_out_anew_goes_out_packed_by_attributes() { + let registry = SessionRegistry::new(); + let rib = Arc::new(RibTable::new(4)); + spawn_advertiser( + LocalRib::spawn(), + Arc::clone(&rib), + registry.clone(), + 65000, + None, + Ipv4Addr::new(10, 0, 0, 100), + watch::channel(true).1, + ); + let peer = |n: u8| PeerInfo { + asn: 64496 + u32::from(n), + router_id: Ipv4Addr::new(10, 0, 0, n), + address: IpAddr::V4(Ipv4Addr::new(192, 0, 2, n)), + interface: 0, + ibgp: false, + reflector_client: false, + }; + // A peer with 300 routes that share three sets of attributes, + // spread over the table's shards. + let from = rib.add_peer(peer(1)).unwrap(); + for set in 0..3u32 { + let came = [ + PathAttribute::Origin(OriginPathAttribute::IGP), + AsPathAttribute::from_asns(vec![64497, 64510 + set]), + PathAttribute::NextHop(NextHopPathAttribute(Ipv4Addr::new(192, 0, 2, 1))), + ]; + let attrs = rib.attrs().intern(&came, None).unwrap(); + let routes: Vec = (0..100) + .map(|n| Nlri::from(nth_prefix(n * 3 + set))) + .collect(); + rib.announce(from, &attrs, &routes); + } + + // A new session that is told of learned routes. + let negotiated = Negotiated { + peer_asn: 64498, + ibgp: false, + peer_identifier: Ipv4Addr::new(10, 0, 0, 2), + hold_time_secs: 90, + four_octet_asn: true, + route_refresh: false, + families: FamilySet::supported(), + add_path_receive: FamilySet::default(), + add_path_send: FamilySet::default(), + role_confirmed: false, + extended_next_hop: false, + graceful_restart: None, + }; + let rib_peer = rib.add_peer(peer(2)).unwrap(); + let (tx, mut outbound) = mpsc::channel(1024); + let mut handle = SessionHandle::new( + PeerId { + addr: (Ipv4Addr::new(192, 0, 2, 2), 179).into(), + asn: 64498, + }, + None, + Direction::Inbound, + (Ipv4Addr::new(192, 0, 2, 100), 179).into(), + &negotiated, + rib_peer, + tx, + ); + handle.export = Arc::new(crate::policy::ExportPolicy { + learned_routes: true, + all_paths: false, + ..crate::policy::ExportPolicy::default() + }); + let _registration = registry.register(handle); + + let mut updates = 0; + let mut routes = HashSet::default(); + tokio::time::timeout(Duration::from_secs(10), async { + while routes.len() < 300 { + match outbound.recv().await.map(|m| message_of(m, false)) { + Some(Message::Update(update)) if !update.announced_routes.is_empty() => { + updates += 1; + routes.extend(update.announced_routes.iter().map(|nlri| nlri.prefix)); + } + Some(_) => {} + None => panic!("the session's queue closed"), + } + } + }) + .await + .expect("the session is told of every route"); + assert_eq!(updates, 3, "one UPDATE for each set of attributes"); + } + /// One peer that takes nothing must not keep the others from being /// sent their routes. There used to be one task announcing to every /// session in turn, and it waited for whichever queue was full.