diff --git a/crates/server/src/advertise.rs b/crates/server/src/advertise.rs index 94d0f99..9091d1f 100644 --- a/crates/server/src/advertise.rs +++ b/crates/server/src/advertise.rs @@ -337,15 +337,13 @@ 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. +/// out alike go in one UPDATE wherever in the table they are. Nothing is +/// dressed until it goes. The table is given out shard by shard, each +/// prefix once, so there is no second word on a prefix to keep track of: +/// a change meanwhile comes after, as news of its own. #[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>, } @@ -358,41 +356,15 @@ struct Group { 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. +/// A route gathered: of which prefix, which of its routes, and the path +/// identifier it is to go out with where routes are told apart. 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); @@ -886,7 +858,8 @@ impl Exporter { /// 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. + /// yet. A prefix the peer is not to have is not gathered: what it + /// held of it is withdrawn once the table is through. fn gather(&mut self, item: News) { let session = &self.session; let Some(Learned { @@ -908,34 +881,27 @@ impl Exporter { 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::Synced | News::Withdrawn(_) => {} 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); + } => { + // The peer has gone, and its routes with it. + if let Some(source) = source_of(&self.rib, &mut gathered.sources, peer) + && may_go(&prefix) + { 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 @@ -949,7 +915,6 @@ impl Exporter { .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. @@ -970,23 +935,17 @@ impl Exporter { .expect("fewer routes than identifiers"), }; taken.push(id); - let entry = Entry { - prefix, - key, - id, - word, - }; - gathered.add(&attrs, source, version, entry); + gathered.add(&attrs, source, version, Entry { prefix, key, id }); } } } } /// 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. + /// all there: each set of routes that go out alike is dressed as its + /// turn comes 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; @@ -998,49 +957,35 @@ impl Exporter { }; // Held until the gathered routes are let go of. let _permit = learned.permit.take(); - let Gathered { groups, heard, .. } = gathered; + let Gathered { groups, .. } = 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; + // that is more than it may. Gathered routes are at most that + // many; only if those are over the limit is each set dressed + // first, to know which cannot be and take their routes with them. 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 at_most: usize = groups.values().map(|group| group.routes.len()).sum(); + if limits.is_some_and(|limits| { + limits + .exceeded_by(self.own_announced.len() + at_most) + .is_some() + }) { + let mut will_hold = 0; + for (n, ((_, _, version), group)) in groups.iter().enumerate() { + if n % EXPORT_BATCH == EXPORT_BATCH - 1 { + tokio::task::yield_now().await; + } + if learned + .dressed(&group.attrs, group.source, *version, &egress) + .is_some() + { + will_hold += group.routes.len(); + } + } + let routes = self.own_announced.len() + will_hold; + 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, @@ -1052,13 +997,14 @@ impl Exporter { }, }); let mut entries = 0; - for Ready { - version, - attrs, - dressed, - routes, - } in ready - { + for ((_, _, version), group) in groups { + // Dressed as its turn comes: the peer is reading the sets + // before it meanwhile. + let Some(dressed) = learned.dressed(&group.attrs, group.source, version, &egress) + else { + continue; + }; + let Group { attrs, routes, .. } = group; let mut nlris = Vec::with_capacity(routes.len()); for entry in routes { entries += 1; diff --git a/crates/server/src/connection.rs b/crates/server/src/connection.rs index 34ef113..ee98e80 100644 --- a/crates/server/src/connection.rs +++ b/crates/server/src/connection.rs @@ -60,9 +60,10 @@ use crate::session::{ }; use crate::throttle::Throttle; -/// Number of outbound messages that may be queued for a session before -/// senders are back-pressured. -const OUTBOUND_QUEUE_CAPACITY: usize = 1024; +/// How many items may be queued for a session before senders wait: each +/// a message, or a piece of up to [`export::ENCODED_PIECE`] octets of +/// UPDATEs from the exporter, so a few megabytes of a table at most. +const OUTBOUND_QUEUE_CAPACITY: usize = 64; /// 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 @@ -1415,10 +1416,20 @@ impl Session { Ok(()) } - /// Puts an UPDATE that is already encoded in the write buffer. - fn send_encoded(&mut self, update: BytesMut) { - self.write_buf.extend_from_slice(&update); - MESSAGES_SENT.with_label_values(&["update"]).inc(); + /// Puts a piece of UPDATEs that are already encoded in the write + /// buffer. + fn send_encoded(&mut self, piece: BytesMut) { + // Counted by their headers: a piece is of whole messages. + let mut messages = 0; + let mut at = 0; + while at + 18 <= piece.len() { + at += usize::from(u16::from_be_bytes([piece[at + 16], piece[at + 17]])).max(19); + messages += 1; + } + self.write_buf.extend_from_slice(&piece); + MESSAGES_SENT + .with_label_values(&["update"]) + .inc_by(messages); if self.write_buf.len() >= WRITE_BATCH_BYTES { self.flush(); } diff --git a/crates/server/src/export.rs b/crates/server/src/export.rs index 1374867..1bccdf4 100644 --- a/crates/server/src/export.rs +++ b/crates/server/src/export.rs @@ -453,11 +453,17 @@ impl EncodedAttributes { /// and a little to spare. const NEXT_HOP_ROOM: usize = 64; +/// How many octets of UPDATEs go to a session in one piece: what the +/// session hands its writer at a time. A piece holds whole messages, +/// so it runs over by at most one. +pub(crate) const ENCODED_PIECE: usize = 64 * 1024; + /// The UPDATE messages, complete with their headers, that announce /// `routes`, all of IP version `version`, with the attributes `encoded` -/// and `next_hop`: as few as they fit in. `ctx` is the session's, and -/// says whether path identifiers go with the family's prefixes -/// (RFC 7911). +/// and `next_hop`: as few as they fit in, in pieces of +/// [`ENCODED_PIECE`] octets or so, each of whole messages. `ctx` is the +/// session's, and says whether path identifiers go with the family's +/// prefixes (RFC 7911). /// /// What [`announcements`] builds as messages, built as bytes: the /// attributes are not parsed, cloned or encoded again for each message. @@ -472,10 +478,10 @@ pub(crate) fn encoded_announcements( let at_once = (room / nlri_octets(version)).max(1); let in_body = matches!(next_hop, NlriNextHop::Ipv4(_)) && version == IpVersion::Ipv4; let with_path_id = ctx.add_path.contains(&AfiSafi::unicast(version)); - routes - .chunks(at_once) - .map(|routes| { - let mut out = BytesMut::with_capacity(4096); + let mut pieces = Vec::new(); + let mut out = BytesMut::with_capacity(ENCODED_PIECE.min(routes.len() * 64 + 4096)); + for routes in routes.chunks(at_once) { + { out.put_slice(&BGP_MARKER); let length_at = out.len(); out.put_u16(0); @@ -511,11 +517,17 @@ pub(crate) fn encoded_announcements( if in_body { write_nlri(routes, with_path_id, &mut out); } - let length = out.len(); + let length = out.len() - length_at + BGP_MARKER.len(); out[length_at..length_at + 2].copy_from_slice(&(length as u16).to_be_bytes()); - out - }) - .collect() + } + if out.len() >= ENCODED_PIECE { + pieces.push(out.split()); + } + } + if !out.is_empty() { + pieces.push(out); + } + pieces } /// The UPDATEs that announce `routes`, all of IP version `version`, as @@ -710,7 +722,7 @@ mod tests { }], }), ]; - let routes: Vec = (0..1000u32) + let routes: Vec = (0..30_000u32) .map(|i| Nlri { prefix: match version { IpVersion::Ipv4 => IpPrefix::new( @@ -762,11 +774,33 @@ mod tests { "{version:?}, path ids {with_path_id}, four-octet {four_octet_asn}" ); assert!(expected.len() > 1, "{what}: more than one message"); - assert_eq!(bytes.len(), expected.len(), "{what}: as many messages"); - for (n, (got, want)) in bytes.iter().zip(&expected).enumerate() { - assert!(got.len() <= 4096, "{what}: message {n} fits"); - assert_eq!(&got[..], &want[..], "{what}: message {n}"); + for update in &expected { + assert!(update.len() <= 4096, "{what}: a message fits"); } + // In pieces of whole messages, each of about the size + // meant, all but the last at least that. + assert!(bytes.len() > 1, "{what}: more than one piece"); + for (n, piece) in bytes.iter().enumerate() { + assert!( + n == bytes.len() - 1 + || (ENCODED_PIECE..ENCODED_PIECE + 4096).contains(&piece.len()), + "{what}: piece {n} is {} octets", + piece.len() + ); + let mut at = 0; + while at < piece.len() { + let length = + usize::from(u16::from_be_bytes([piece[at + 16], piece[at + 17]])); + at += length; + } + assert_eq!(at, piece.len(), "{what}: piece {n} is of whole messages"); + } + let got: Vec = bytes + .iter() + .flat_map(|piece| piece.iter().copied()) + .collect(); + let want: Vec = expected.iter().flat_map(|m| m.iter().copied()).collect(); + assert_eq!(got, want, "{what}: the same octets"); } } } diff --git a/crates/server/src/session.rs b/crates/server/src/session.rs index 9603d25..114ffe0 100644 --- a/crates/server/src/session.rs +++ b/crates/server/src/session.rs @@ -318,19 +318,19 @@ impl SessionHandle { .map_err(|_| SessionClosed(self.peer)) } - /// Sends the peer an UPDATE that is already encoded, header and all, - /// as the exporter builds them. - pub async fn send_encoded(&self, update: BytesMut) -> Result<(), SessionClosed> { + /// Sends the peer a piece of one or more UPDATEs that are already + /// encoded, headers and all, as the exporter builds them. + pub async fn send_encoded(&self, piece: BytesMut) -> Result<(), SessionClosed> { self.outbound - .send(Outbound::Encoded(update)) + .send(Outbound::Encoded(piece)) .await .map_err(|_| SessionClosed(self.peer)) } } -/// What the rest of the server gives a session to send: a message, or -/// an UPDATE already encoded, header and all, which goes to the peer as -/// it is once the session is established. +/// What the rest of the server gives a session to send: a message, or a +/// piece of one or more UPDATEs already encoded, headers and all, which +/// goes to the peer as it is once the session is established. #[derive(Debug)] pub enum Outbound { Message(Message),