diff --git a/crates/bench/benches/mesh.rs b/crates/bench/benches/mesh.rs index 33ad328..9c09c32 100644 --- a/crates/bench/benches/mesh.rs +++ b/crates/bench/benches/mesh.rs @@ -20,7 +20,12 @@ //! 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. +//! do after a restart. `BENCH_COLLECTORS` (default 0) more peers of the +//! first router, from 127.0.1.0/24, are peers in the sense of RFC 9234 +//! told of learned routes, and the feeders' routes are marked as having +//! come across from a peer, so none of them goes: as many exporters +//! following the table while it fills and passing nearly all of it +//! over, as a router with many peers at exchanges has. use std::net::{Ipv4Addr, Ipv6Addr, SocketAddr}; use std::sync::Arc; @@ -31,10 +36,11 @@ 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::attributes::PathAttribute; 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::policy::{ExportPolicy, ImportPolicy, PeerRole}; use bgp_server::session::SessionRegistry; use tokio::io::{AsyncReadExt, AsyncWriteExt, BufReader}; @@ -44,6 +50,14 @@ 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); +/// Where the collectors connect from: this and the addresses after it. +fn collector_address(n: u32) -> Ipv4Addr { + Ipv4Addr::new(127, 0, 1, n as u8) +} + +/// The collectors' AS numbers start here. +const COLLECTOR_ASN: u32 = 65000; + /// What the routers send each other, and the receiver: every path. fn every_path() -> Arc { Arc::new(ExportPolicy { @@ -70,6 +84,7 @@ async fn router( port: u16, other: u16, with_receiver: bool, + collectors: u32, ) -> TestServer { TestServer::start_with(ASN, |c| { // Two routers of one AS must not share a BGP identifier. @@ -96,6 +111,21 @@ async fn router( ..peer_config("receiver", SocketAddr::new(RECEIVER.into(), 179), ASN) }); } + for n in 1..=collectors { + c.peers.push(PeerConfig { + role: Some(PeerRole::Peer), + export: Arc::new(ExportPolicy { + learned_routes: true, + all_paths: false, + ..ExportPolicy::default() + }), + ..peer_config( + "collector", + SocketAddr::new(collector_address(n).into(), 179), + COLLECTOR_ASN + n, + ) + }); + } c.connect_retry = Duration::from_millis(200); }) .await @@ -110,10 +140,21 @@ async fn feeders( shape: Shape, first: u32, count: u32, + marked: bool, ) -> Vec<(TestPeer, Vec)> { let mut sessions = Vec::new(); for peer in first..first + count { - let stream: Vec = encoded(&announcements(prefixes, shape, peer)) + let mut updates = announcements(prefixes, shape, peer); + if marked { + // As having come across from a peer (RFC 9234): not for the + // collectors, which are peers too. + for update in &mut updates { + update + .path_attributes + .push(PathAttribute::OnlyToCustomer(64512 + peer)); + } + } + let stream: Vec = encoded(&updates) .iter() .flat_map(|m| m.iter().copied()) .collect(); @@ -152,13 +193,46 @@ 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, true).await; - let b = router("a", 2, port_b, port_a, false).await; + let collectors = env("BENCH_COLLECTORS", 0) as u32; + let a = router("b", 1, port_a, port_b, true, collectors).await; + let b = router("a", 2, port_b, port_a, false, 0).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"); + // The collectors are there before the feeders, as a router's + // peers are before its tables arrive after a restart. + let mut draining = Vec::new(); + for n in 1..=collectors { + let socket = tokio::net::TcpSocket::new_v4().expect("a socket"); + if let Err(e) = socket.bind(SocketAddr::new(collector_address(n).into(), 0)) { + println!( + "collectors: skipped, {} is not ours ({e})", + collector_address(n) + ); + break; + } + let stream = socket.connect(a.addr).await.expect("A listens"); + let mut session = TestPeer::from_stream(stream); + session + .handshake(COLLECTOR_ASN + n, Ipv4Addr::new(192, 0, 3, n as u8), 0) + .await + .expect("A takes a collector's session"); + let (mut reader, writer) = session.into_stream().into_split(); + draining.push(( + writer, + tokio::spawn(async move { + let mut sink = vec![0u8; 1 << 16]; + while matches!(reader.read(&mut sink).await, Ok(n) if n > 0) {} + }), + )); + } + if collectors > 0 { + a.wait_for_sessions(1 + draining.len()) + .await + .expect("the collectors' sessions at A"); + } - let feed_a = feeders(&a, &prefixes, shape, 1, peers_a).await; - let feed_b = feeders(&b, &prefixes, shape, 1 + peers_a, peers_b).await; + let feed_a = feeders(&a, &prefixes, shape, 1, peers_a, collectors > 0).await; + let feed_b = feeders(&b, &prefixes, shape, 1 + peers_a, peers_b, false).await; let own_a = per_peer * peers_a as usize; let own_b = per_peer * peers_b as usize; @@ -258,6 +332,7 @@ fn main() { None => resync(a.addr, a.registry.clone(), receiver_routes).await, } drop(feeders); + drop(draining); }); } diff --git a/crates/rib/src/table/attrs.rs b/crates/rib/src/table/attrs.rs index d9ffdc3..c5b85f3 100644 --- a/crates/rib/src/table/attrs.rs +++ b/crates/rib/src/table/attrs.rs @@ -209,6 +209,22 @@ impl AttrSet { plain } + /// Whether the route carries the OTC attribute (RFC 9234): it came + /// down from a provider or across from a peer, and is to go on down + /// only. Read off the stored bytes without building anything else, + /// for those who ask it of every route: a session to a peer or a + /// provider asks it of most of the table, to pass over them. + pub fn is_only_to_customer(&self) -> bool { + let mut wire = &*self.wire; + while let Ok((attribute, rest)) = split_attribute(wire) { + if attribute.type_code == attr_type::ONLY_TO_CUSTOMER { + return true; + } + wire = rest; + } + false + } + /// Whether the route carries the BLACKHOLE community (RFC 7999), which /// asks for the packets to its prefix to be dropped. Read off the /// stored bytes without building anything else, for those who ask it diff --git a/crates/rib/src/table/journal.rs b/crates/rib/src/table/journal.rs index 3b17d97..0d92b18 100644 --- a/crates/rib/src/table/journal.rs +++ b/crates/rib/src/table/journal.rs @@ -113,6 +113,10 @@ struct Journal { /// Where each reader has read to, by reader slot. Written by the /// readers themselves while they hold the shard for reading. cursors: Vec, + /// `seq`, for reading without the shard's lock: the table holds the + /// other end, so that a reader can tell a shard with nothing new for + /// it from one with something without stopping the shard's writers. + latest: Arc, /// How many reader slots are in use. readers: usize, limits: Limits, @@ -125,6 +129,7 @@ impl Journal { Self { kind, limits, + latest: Arc::new(AtomicU64::new(0)), seq: 0, floor: 0, entries: VecDeque::new(), @@ -139,6 +144,7 @@ impl Journal { /// number for the row to remember. fn changed(&mut self, prefix: IpPrefix) -> u64 { self.seq += 1; + self.latest.store(self.seq, Ordering::Release); if self.readers > 0 { self.entries.push_back(Entry { seq: self.seq, @@ -161,6 +167,7 @@ impl Journal { /// Notes that `prefix` lost its last route and has no row any more. fn disappeared(&mut self, prefix: IpPrefix) { self.seq += 1; + self.latest.store(self.seq, Ordering::Release); if self.readers > 0 { self.entries.push_back(Entry { seq: self.seq, @@ -253,6 +260,16 @@ pub(super) enum Kind { All, } +impl Kind { + /// Its place in what the table keeps for each shard's two journals. + pub(super) fn index(self) -> usize { + match self { + Self::Best => 0, + Self::All => 1, + } + } +} + /// The sequence numbers a row remembers, one for each journal: of the /// latest change of its best route, and of any of its routes. #[derive(Debug, Clone, Copy)] @@ -285,6 +302,12 @@ impl Journals { } } + /// The latest sequence number of each journal, by [`Kind::index`], + /// as something to read without the shard's lock. + pub(super) fn latest(&self) -> [Arc; 2] { + [Arc::clone(&self.best.latest), Arc::clone(&self.all.latest)] + } + fn of(&self, kind: Kind) -> &Journal { match kind { Kind::Best => &self.best, @@ -598,7 +621,11 @@ impl Reader { pub fn poll(&mut self, budget: usize, mut visit: impl FnMut(Event<'_>)) -> Progress { // A call that may give nothing out would never get anywhere. let budget = budget.max(1); - let shards = self.table.shards.len(); + // Held once for the call: a count taken on the table for every + // shard of every pass, by every reader, was the most of what a + // router with many peers did. + let table = Arc::clone(&self.table); + let shards = table.shards.len(); let mut events = 0; loop { match self.phase { @@ -621,7 +648,7 @@ impl Reader { self.phase = Phase::Follow { next: 0, upto: 0 }; continue; } - events += self.scan(next, upto, &mut visit); + events += self.scan(&table, next, upto, &mut visit); self.phase = Phase::Resync { next: next + 1, announced: true, @@ -654,7 +681,14 @@ impl Reader { self.phase = Phase::Follow { next: shard, upto }; return self.progress(events, false); } - match self.drain(shard, upto, &mut visit) { + // Nothing new in this shard, by the number that is + // read without its lock: it is left alone, and so + // are the sessions writing to it. + if self.cursors[shard] == self.table.latest_seq(shard, self.kind) { + shard += 1; + continue; + } + match self.drain(&table, shard, upto, &mut visit) { Drained::Lagged => { lagged = true; break; @@ -765,8 +799,13 @@ impl Reader { /// as of `upto` on the table's clock, in one hold of the shard: what /// changed after that time is in the journal after that place, whether /// or not the rows given out show it already. - fn scan(&mut self, index: usize, upto: u64, visit: &mut impl FnMut(Event<'_>)) -> usize { - let table = Arc::clone(&self.table); + fn scan( + &mut self, + table: &RibTable, + index: usize, + upto: u64, + visit: &mut impl FnMut(Event<'_>), + ) -> usize { let shard = read(&table.shards[index]); for row in &shard.rows { visit(self.event_for(row)); @@ -802,9 +841,12 @@ impl Reader { // many changes were made with it. let mut earliest = None; for (index, shard) in self.table.shards.iter().enumerate() { + let cursor = self.cursors[index]; + if cursor == self.table.latest_seq(index, self.kind) { + continue; + } let shard = read(shard); let journal = shard.journals.of(self.kind); - let cursor = self.cursors[index]; if cursor < journal.floor || cursor == journal.seq { continue; } @@ -822,8 +864,13 @@ impl Reader { /// Reads a shard's journal from this reader's place, no further than /// the changes made by `upto` on the table's clock. - fn drain(&mut self, index: usize, upto: u64, visit: &mut impl FnMut(Event<'_>)) -> Drained { - let table = Arc::clone(&self.table); + fn drain( + &mut self, + table: &RibTable, + index: usize, + upto: u64, + visit: &mut impl FnMut(Event<'_>), + ) -> Drained { let shard = read(&table.shards[index]); let journal = shard.journals.of(self.kind); let mut cursor = self.cursors[index]; diff --git a/crates/rib/src/table/mod.rs b/crates/rib/src/table/mod.rs index a4cd95e..e28ed8d 100644 --- a/crates/rib/src/table/mod.rs +++ b/crates/rib/src/table/mod.rs @@ -317,6 +317,9 @@ struct Shard { /// takes `&self`. pub struct RibTable { shards: Box<[RwLock]>, + /// The latest sequence number of each shard's two journals, kept by + /// the journals themselves, for reading without the shard's lock. + latest: Box<[[Arc; 2]]>, hasher: RandomState, attrs: AttrStore, /// Which peer numbers are in use. Changing the peer list also takes @@ -355,19 +358,23 @@ impl RibTable { /// sizes. Tests use small ones to get there quickly. fn with_journal_limits(shards: usize, limits: journal::Limits) -> Self { let hasher = RandomState::default(); + let mut latest = Vec::with_capacity(shards.max(1)); Self { shards: (0..shards.max(1)) .map(|_| { + let journals = journal::Journals::new(limits); + latest.push(journals.latest()); RwLock::new(Shard { rows: HashTable::new(), routes: 0, peers: Vec::new(), hasher: hasher.clone(), scratch: select::Scratch::default(), - journals: journal::Journals::new(limits), + journals, }) }) .collect(), + latest: latest.into_boxed_slice(), hasher, attrs: AttrStore::new(), peers: Mutex::default(), @@ -651,6 +658,12 @@ impl RibTable { } } + /// The latest sequence number of shard `shard`'s journal of `kind`, + /// without the shard's lock. + fn latest_seq(&self, shard: usize, kind: journal::Kind) -> u64 { + self.latest[shard][kind.index()].load(std::sync::atomic::Ordering::Acquire) + } + /// The time on the clock up to which every change is complete in every /// shard: everything before the earliest one still under way. pub(super) fn settled(&self) -> u64 { diff --git a/crates/server/src/advertise.rs b/crates/server/src/advertise.rs index 36a2c14..b5e6fff 100644 --- a/crates/server/src/advertise.rs +++ b/crates/server/src/advertise.rs @@ -248,6 +248,9 @@ struct Exporter { /// Small enough that what one batch sends is being read by the peer /// while the next is being dressed: a batch is dressed whole before any /// of it goes, and the session's queue holds the exporter once full. +/// And small enough that a shard is held for reading only briefly on +/// the way: the sessions writing to it wait for that, and a batch +/// sixteen times this size made them wait three times as long in all. const EXPORT_BATCH: usize = 4_096; /// How long changes are gathered before they go out, once there are any. @@ -495,6 +498,16 @@ impl Learned { self.announced.routes() + self.stale.as_ref().map_or(0, Announced::routes) } + /// Whether the peer holds anything for `prefix`, announced or in + /// doubt since a reset. + fn holds(&self, prefix: &IpPrefix) -> bool { + let held = |announced: &Announced| match announced { + Announced::Best(prefixes) => prefixes.contains(prefix), + Announced::All { routes, .. } => routes.contains_key(prefix), + }; + held(&self.announced) || self.stale.as_ref().is_some_and(held) + } + /// The verdict that the peer is to have nothing for a prefix. fn nothing(&self) -> Verdict { if self.all_paths { @@ -574,8 +587,19 @@ impl Learned { if let Some((_, dressed)) = self.dressed.get(&key) { return dressed.clone(); } - let dressed = attrs - .attributes() + // A route that came down or across goes on down only (RFC 9234): + // told from the bytes, as a session to a peer or a provider is + // told it of most of the table, and parsing them for it would be + // most of the work of following the table for such a session. + let towards = &egress.towards; + let only_down = !towards.ibgp + && towards + .role + .is_some_and(|role| !role.may_be_sent_only_to_customer_routes()) + && attrs.is_only_to_customer(); + let dressed = (!only_down) + .then(|| attrs.attributes()) + .flatten() .and_then(|(attributes, next_hop)| { egress.outgoing(attributes, next_hop, source, version) }) @@ -1123,8 +1147,14 @@ impl Exporter { match item { // Dealt with by `follow`. News::Reset | News::Synced => {} + // Nothing for a prefix the peer does not hold is no news + // for it: most of a table is that for a session that is + // sent only some of it, and the verdict would cost a map + // insert here and a lookup on its way out. News::Withdrawn(prefix) => { - verdicts.insert(prefix, learned.nothing()); + if learned.holds(&prefix) { + verdicts.insert(prefix, learned.nothing()); + } } News::Best { prefix, @@ -1138,7 +1168,9 @@ impl Exporter { // The peer has gone, and its routes with it. _ => None, }; - verdicts.insert(prefix, Verdict::Best(dressed)); + if dressed.is_some() || learned.holds(&prefix) { + verdicts.insert(prefix, Verdict::Best(dressed)); + } } News::Routes { prefix, routes } => { let version = prefix.address_family(); @@ -1160,7 +1192,9 @@ impl Exporter { } } } - verdicts.insert(prefix, Verdict::All(wanted)); + if !wanted.is_empty() || learned.holds(&prefix) { + verdicts.insert(prefix, Verdict::All(wanted)); + } } } }