diff --git a/Cargo.lock b/Cargo.lock index 9941abe..846b932 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -289,6 +289,7 @@ dependencies = [ "bgp-rib", "bgp-rpki", "bytes", + "foldhash", "futures", "nix", "prometheus", diff --git a/crates/packet/src/message.rs b/crates/packet/src/message.rs index 285d253..b34b67e 100644 --- a/crates/packet/src/message.rs +++ b/crates/packet/src/message.rs @@ -30,7 +30,7 @@ pub const BGP_HEADER_LEN: u16 = 19; /// Maximum BGP message length per RFC 4271 §4.1. pub const BGP_MAX_MSG_LEN: u16 = 4096; -const BGP_MARKER: [u8; 16] = [0xFF; 16]; +pub const BGP_MARKER: [u8; 16] = [0xFF; 16]; /// BGP message type codes (RFC 4271 §4.1, RFC 2918). mod msg_type { @@ -358,7 +358,7 @@ impl UpdateMessage { /// Writes prefixes as an UPDATE names them: for each, its path identifier /// if the session uses them for the family (RFC 7911 §3), its length in /// bits, and the bytes that length covers. -pub(crate) fn write_nlri(nlri: &[Nlri], with_path_id: bool, out: &mut BytesMut) { +pub fn write_nlri(nlri: &[Nlri], with_path_id: bool, out: &mut BytesMut) { for n in nlri { if with_path_id { out.put_u32(n.path_id); diff --git a/crates/server/Cargo.toml b/crates/server/Cargo.toml index 4e81d34..7cd989b 100644 --- a/crates/server/Cargo.toml +++ b/crates/server/Cargo.toml @@ -8,6 +8,7 @@ repository.workspace = true version.workspace = true [dependencies] +foldhash.workspace = true bgp-packet.workspace = true bgp-primitives.workspace = true bgp-rib.workspace = true diff --git a/crates/server/src/advertise.rs b/crates/server/src/advertise.rs index 78128dd..2fdd3f2 100644 --- a/crates/server/src/advertise.rs +++ b/crates/server/src/advertise.rs @@ -28,7 +28,10 @@ //! peer has sent its own routes since the start. A peer that kept our //! routes over a restart of ours drops those it has not had again then. -use std::collections::{HashMap, HashSet}; +/// The maps of this module, hashed fast: a full table is a million +/// prefixes in each of them. +type HashMap = std::collections::HashMap; +type HashSet = std::collections::HashSet; use std::net::{IpAddr, Ipv4Addr, Ipv6Addr}; use std::sync::Arc; use std::time::Duration; @@ -43,7 +46,6 @@ use bgp_primitives::ip_prefix::{IpPrefix, Nlri}; use bgp_primitives::parser::ParserContext; use bgp_rib::table::{AttrSet, DEFAULT_LOCAL_PREF, Event, PeerIndex, Reader, RibTable}; use bgp_rib::{LocalRib, LocalRibEvent, LocalRoute}; -use bytes::BytesMut; use prometheus::IntGauge; use tokio::sync::broadcast::Receiver; use tokio::sync::broadcast::error::RecvError; @@ -52,7 +54,7 @@ use tokio::task::JoinHandle; use tracing::{debug, error, info, warn}; use crate::error::SessionError; -use crate::export::{self, Egress, Outgoing, Reflection, Source}; +use crate::export::{self, Egress, EncodedAttributes, Outgoing, Reflection, Source}; use crate::family::FamilySet; use crate::marks::Control; use crate::metrics::{PEER_ADVERTISED, ROUTES_EXPORTED}; @@ -80,7 +82,7 @@ pub fn spawn_advertiser( control, cluster_id, resyncs: Arc::new(Semaphore::new(RESYNCS_AT_ONCE)), - exporters: HashMap::new(), + exporters: HashMap::default(), }; tokio::spawn(advertiser.run()) } @@ -168,7 +170,7 @@ impl Advertiser { session: session.clone(), local_asn: self.local_asn, resend: Arc::clone(&resend), - own_announced: HashSet::new(), + own_announced: HashSet::default(), end_of_rib: session .graceful_restart .as_ref() @@ -254,8 +256,9 @@ const RESYNCS_AT_ONCE: usize = 4; /// How many routes ready to go out are remembered, by the attributes they /// came with. A table has a few hundred thousand distinct sets, each a -/// kilobyte or so in this form; what a flush is sending holds its own. -const DRESSED_KEPT: usize = 65_536; +/// few hundred octets in this form: some tens of megabytes a session at +/// most. What a flush is sending holds its own. +const DRESSED_KEPT: usize = 131_072; fn version_index(version: IpVersion) -> usize { match version { @@ -290,7 +293,9 @@ type DressedKey = (usize, Source, IpVersion); /// A learned route as this session announces it. struct Dressed { - outgoing: Outgoing, + /// Its attributes as they go on this session's wire, encoded once. + encoded: EncodedAttributes, + next_hop: NlriNextHop, /// How many octets its attributes take on this session's wire. octets: usize, } @@ -343,11 +348,11 @@ impl Announced { fn new(all_paths: bool) -> Self { if all_paths { Self::All { - routes: HashMap::new(), + routes: HashMap::default(), count: 0, } } else { - Self::Best(HashSet::new()) + Self::Best(HashSet::default()) } } @@ -504,7 +509,7 @@ impl Learned { announced: Announced::new(all_paths), synced: false, stale: None, - dressed: HashMap::new(), + dressed: HashMap::default(), } } @@ -526,22 +531,22 @@ impl Learned { .and_then(|(attributes, next_hop)| { egress.outgoing(attributes, next_hop, source, version) }) - .map(|outgoing| { - // As long as they will be on the wire, to know how many - // prefixes fit beside them. - let ctx = ParserContext { - four_octet_asn: egress.towards.four_octet_asn, - ibgp: egress.towards.ibgp, - add_path: Vec::new(), - }; - let mut wire = BytesMut::new(); - let mut attributes = outgoing.attributes.clone(); + .and_then(|outgoing| { + // Encoded here, once: every UPDATE with these attributes + // carries the same octets. + let Outgoing { + mut attributes, + next_hop, + } = outgoing; export::finish(&mut attributes, &egress.towards); - let octets = attributes - .iter() - .try_for_each(|attribute| attribute.to_wire(&ctx, &mut wire)) - .map_or(usize::MAX, |()| wire.len()); - Arc::new(Dressed { outgoing, octets }) + let encoded = + EncodedAttributes::encode(&attributes, egress.towards.four_octet_asn)?; + let octets = encoded.octets(); + Some(Arc::new(Dressed { + encoded, + next_hop, + octets, + })) }) // One that cannot be written is one that does not go. .filter(|dressed| dressed.octets < 3072); @@ -755,7 +760,7 @@ impl Exporter { 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::new(); + 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. @@ -857,7 +862,8 @@ impl Exporter { return Ok(()); }; let mut withdrawn: [Vec; 2] = Default::default(); - let mut batches: HashMap<(usize, IpVersion), (Arc, Vec)> = HashMap::new(); + let mut batches: HashMap<(usize, IpVersion), (Arc, Vec)> = + HashMap::default(); for (prefix, verdict) in verdicts { let version = prefix.address_family(); let withdrawn = &mut withdrawn[version_index(version)]; @@ -979,20 +985,32 @@ impl Exporter { .values() .map(|(_, routes)| routes.len()) .sum::(); + // The session's context for each IP version, with whether path + // identifiers go with the family's prefixes: made once a flush, + // not once an UPDATE. + 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 announced = 0; for ((_, version), (dressed, routes)) in batches { ROUTES_EXPORTED .with_label_values(&["announced"]) .inc_by(routes.len() as u64); - let updates = export::announcements( - &dressed.outgoing, + let updates = export::encoded_announcements( + &dressed.encoded, + dressed.next_hop, version, &routes, - &towards, - dressed.octets, + &contexts[version_index(version)], ); for update in updates { - session.send(Message::Update(update)).await?; + session.send_encoded(update).await?; } announced += routes.len(); note(before + announced); @@ -1663,7 +1681,25 @@ mod tests { use crate::collision::Direction; use crate::family::FamilySet; use crate::fsm::Negotiated; - use crate::session::SessionGuard; + use crate::session::{Outbound, SessionGuard}; + + /// What a session was given to send, as a message: an encoded UPDATE + /// decoded as the peer decodes it, without ADD-PATH, in our own AS + /// (`ibgp`) or not: what attributes are expected and kept depends on + /// which. + fn message_of(outbound: Outbound, ibgp: bool) -> Message { + match outbound { + Outbound::Message(message) => message, + Outbound::Encoded(bytes) => { + let ctx = ParserContext { + four_octet_asn: true, + ibgp, + add_path: Vec::new(), + }; + Message::decode(&ctx, &bytes).expect("an encoded UPDATE decodes") + } + } + } /// A registry and local RIB with an advertiser running, and a table to /// take peer numbers from. @@ -1676,7 +1712,7 @@ mod tests { /// The far end of a session's queue: what the session's task would /// write to the peer. struct Queue { - outbound: mpsc::Receiver, + outbound: mpsc::Receiver, /// What the peer holds, going by what it has been sent. routes: HashSet, _registration: SessionGuard, @@ -1746,7 +1782,7 @@ mod tests { ); Queue { outbound, - routes: HashSet::new(), + routes: HashSet::default(), _registration: self.registry.register(handle), } } @@ -1773,7 +1809,9 @@ mod tests { async fn read_until_it_has(&mut self, expected: &HashSet) { let reading = async { while self.routes != *expected { - let Some(Message::Update(update)) = self.outbound.recv().await else { + let Some(Message::Update(update)) = + self.outbound.recv().await.map(|m| message_of(m, false)) + else { panic!("expected an UPDATE"); }; for nlri in &update.withdrawn_routes { @@ -1873,7 +1911,7 @@ mod tests { let update = tokio::time::timeout(Duration::from_secs(10), async { loop { - match outbound.recv().await { + match outbound.recv().await.map(|m| message_of(m, true)) { Some(Message::Update(update)) if !update.announced_routes.is_empty() => { return update; } @@ -2053,19 +2091,19 @@ mod tests { /// The far end of the limited peer's session: what it is sent, and /// what ends it. struct LimitedPeer { - outbound: mpsc::Receiver, + outbound: mpsc::Receiver, ending: mpsc::UnboundedReceiver, _registration: SessionGuard, } /// The routes announced on a session's queue until it has been quiet /// for a while. - async fn announced(outbound: &mut mpsc::Receiver) -> HashSet { - let mut routes = HashSet::new(); + async fn announced(outbound: &mut mpsc::Receiver) -> HashSet { + let mut routes = HashSet::default(); while let Ok(Some(message)) = tokio::time::timeout(Duration::from_millis(500), outbound.recv()).await { - if let Message::Update(update) = message { + if let Message::Update(update) = message_of(message, false) { for nlri in &update.withdrawn_routes { routes.remove(&nlri.prefix); } @@ -2107,7 +2145,7 @@ mod tests { .await; assert_eq!( announced(&mut peer.outbound).await, - HashSet::from([nth_prefix(2), nth_prefix(11)]) + HashSet::from_iter([nth_prefix(2), nth_prefix(11)]) ); assert!(peer.ending.try_recv().is_err(), "the session goes on"); } @@ -2137,7 +2175,7 @@ mod tests { ), "{why:?}" ); - assert_eq!(announced(&mut peer.outbound).await, HashSet::new()); + assert_eq!(announced(&mut peer.outbound).await, HashSet::default()); } /// Our own routes count too, and none goes past the limit: those that diff --git a/crates/server/src/connection.rs b/crates/server/src/connection.rs index bf0135e..34ef113 100644 --- a/crates/server/src/connection.rs +++ b/crates/server/src/connection.rs @@ -55,7 +55,9 @@ use crate::rejections::{PeerRejections, RejectionLog}; use crate::restart::{self, Agreed, GracefulRestartConfig, Leave, Restarts}; use crate::routes::{self, Held}; use crate::rpki::{Checks, Subject, Validator}; -use crate::session::{PeerId, SessionCounters, SessionGuard, SessionHandle, SessionRegistry}; +use crate::session::{ + Outbound, PeerId, SessionCounters, SessionGuard, SessionHandle, SessionRegistry, +}; use crate::throttle::Throttle; /// Number of outbound messages that may be queued for a session before @@ -371,7 +373,7 @@ struct Established { /// The peer's number in the routing table. rib_peer: PeerIndex, /// Messages the rest of the server queued for the peer. - outbound: mpsc::Receiver, + outbound: mpsc::Receiver, /// Where the rest of the server asks for the session to end, and why. ending: mpsc::UnboundedReceiver, /// The routes learned from the peer, which go when the session does. @@ -609,10 +611,10 @@ enum Wake { /// Something for the state machine. Event(Event), /// The writer has room for a batch that was waiting. - WriterRoom(mpsc::OwnedPermit>), + WriterRoom(mpsc::OwnedPermit), /// Messages from the rest of the server to send, as many as were /// waiting, to be written together. - Outbound(Vec), + Outbound(Vec), /// The validation data has changed; these routes of the peer's are /// now invalid. Revalidate(Vec), @@ -648,12 +650,12 @@ struct Session { frames: FrameReader, /// Batches of messages on their way to the writer task, which writes /// them to the peer; see [`write_to_peer`]. - writer: mpsc::Sender>, + writer: mpsc::Sender, writer_task: JoinHandle<()>, /// Batches the writer had no room for yet, oldest first. While there /// are any, no more messages are taken from the rest of the server; /// the peer is still read. - pending: VecDeque>, + pending: VecDeque, /// What the writer task says if it fails. write_failed: mpsc::Receiver, /// Messages encoded and not yet made a batch of. @@ -818,10 +820,25 @@ impl Session { } actions = match self.next_event().await? { Wake::Event(event) => self.fsm.handle(event), - Wake::Outbound(messages) => messages - .into_iter() - .flat_map(|msg| self.fsm.handle(Event::Outbound(msg))) - .collect(), + Wake::Outbound(messages) => { + let mut actions = Vec::new(); + for item in messages { + match item { + Outbound::Message(msg) => { + actions.extend(self.fsm.handle(Event::Outbound(msg))); + } + // Encoded by the exporter for this session as + // it is now; one for a session that is no + // longer established is void. + Outbound::Encoded(update) => { + if self.established.is_some() { + self.send_encoded(update); + } + } + } + } + actions + } Wake::WriterRoom(permit) => { if let Some(batch) = self.pending.pop_front() { permit.send(batch); @@ -1125,7 +1142,7 @@ impl Session { negotiated: &Negotiated, peer_id: PeerId, rib_peer: PeerIndex, - outbound: mpsc::Sender, + outbound: mpsc::Sender, ) -> SessionHandle { let peer = &self.peer; let mut handle = SessionHandle::new( @@ -1398,6 +1415,15 @@ 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(); + if self.write_buf.len() >= WRITE_BATCH_BYTES { + self.flush(); + } + } + /// Makes a batch of the write buffer and gives it to the writer, or /// keeps it until the writer has room. Never waits: a session that /// waited to write would not be reading, and a peer doing the same @@ -1406,7 +1432,8 @@ impl Session { if self.write_buf.is_empty() { return; } - self.pending.push_back(self.write_buf.split().to_vec()); + // Split off, not copied: the buffer's bytes go as they are. + self.pending.push_back(self.write_buf.split()); self.push_pending(); } @@ -1528,7 +1555,7 @@ async fn from_server(established: &mut Option, may_take_outbound: b /// session has nothing more to send, our side of the connection is shut. async fn write_to_peer( mut writer: OwnedWriteHalf, - mut queue: mpsc::Receiver>, + mut queue: mpsc::Receiver, send_hold: watch::Receiver, failed: mpsc::Sender, ) { diff --git a/crates/server/src/export.rs b/crates/server/src/export.rs index 1487802..1374867 100644 --- a/crates/server/src/export.rs +++ b/crates/server/src/export.rs @@ -33,12 +33,15 @@ use std::net::{IpAddr, Ipv4Addr}; use bgp_packet::message::{ - AsPathAttribute, AsPathSegment, LocalPrefPathAttribute, MpReachNlriPathAttribute, + AsPathAttribute, AsPathSegment, BGP_MARKER, LocalPrefPathAttribute, MpReachNlriPathAttribute, MpUnreachNlriPathAttribute, NextHopPathAttribute, NlriNextHop, PathAttribute, UpdateMessage, + attr_type, write_nlri, }; use bgp_primitives::constants::{AfiSafi, IpVersion}; use bgp_primitives::ip_prefix::Nlri; +use bgp_primitives::parser::ParserContext; use bgp_rib::table::{DEFAULT_LOCAL_PREF, PeerIndex}; +use bytes::{BufMut, Bytes, BytesMut}; use crate::advertise::{Towards, ipv6_next_hop}; use crate::marks::{self, Control, Named, communities_of, well_known_allow}; @@ -399,9 +402,129 @@ fn nlri_octets(routes: IpVersion) -> usize { 4 + 1 + usize::from(routes.bits()) / 8 } +/// A route's attributes as they go out on one session, encoded once and +/// kept: what every UPDATE announcing a route with them carries, less the +/// next hop, which goes with the prefixes. `at_next_hop` and +/// `at_mp_reach` are where in `bytes` a NEXT_HOP (type 3) and an +/// MP_REACH_NLRI (type 14) go, so that the attributes stay in order of +/// type code. +#[derive(Debug)] +pub struct EncodedAttributes { + pub bytes: Bytes, + pub at_next_hop: usize, + pub at_mp_reach: usize, +} + +impl EncodedAttributes { + /// Encodes `attributes`, which are `finish`ed, for a session with + /// `four_octet_asn`. `None` if they cannot be encoded. + pub fn encode(attributes: &[PathAttribute], four_octet_asn: bool) -> Option { + let ctx = ParserContext { + four_octet_asn, + ibgp: false, + add_path: Vec::new(), + }; + let mut out = BytesMut::with_capacity(256); + let mut at_next_hop = None; + let mut at_mp_reach = None; + for attribute in attributes { + let code = attribute.type_code(); + if code > attr_type::NEXT_HOP && at_next_hop.is_none() { + at_next_hop = Some(out.len()); + } + if code > attr_type::MP_REACH_NLRI && at_mp_reach.is_none() { + at_mp_reach = Some(out.len()); + } + attribute.to_wire(&ctx, &mut out).ok()?; + } + Some(Self { + at_next_hop: at_next_hop.unwrap_or(out.len()), + at_mp_reach: at_mp_reach.unwrap_or(out.len()), + bytes: out.freeze(), + }) + } + + pub fn octets(&self) -> usize { + self.bytes.len() + } +} + +/// What the next hop takes, in its own attribute or with the prefixes, +/// and a little to spare. +const NEXT_HOP_ROOM: usize = 64; + +/// 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). +/// +/// What [`announcements`] builds as messages, built as bytes: the +/// attributes are not parsed, cloned or encoded again for each message. +pub(crate) fn encoded_announcements( + encoded: &EncodedAttributes, + next_hop: NlriNextHop, + version: IpVersion, + routes: &[Nlri], + ctx: &ParserContext, +) -> Vec { + let room = UPDATE_ROOM.saturating_sub(encoded.octets() + NEXT_HOP_ROOM); + 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); + out.put_slice(&BGP_MARKER); + let length_at = out.len(); + out.put_u16(0); + // The type of an UPDATE (RFC 4271 §4.1). + out.put_u8(2); + // No withdrawn routes. + out.put_u16(0); + let attributes_at = out.len(); + out.put_u16(0); + if in_body { + let NlriNextHop::Ipv4(next_hop) = next_hop else { + unreachable!("in_body means an IPv4 next hop"); + }; + out.put_slice(&encoded.bytes[..encoded.at_next_hop]); + PathAttribute::NextHop(NextHopPathAttribute(next_hop)) + .to_wire(ctx, &mut out) + .expect("a next hop encodes"); + out.put_slice(&encoded.bytes[encoded.at_next_hop..]); + } else { + out.put_slice(&encoded.bytes[..encoded.at_mp_reach]); + PathAttribute::MpReachNlri(MpReachNlriPathAttribute { + family: AfiSafi::unicast(version), + next_hop, + prefixes: routes.to_vec(), + }) + .to_wire(ctx, &mut out) + .expect("a chunk that fits encodes"); + out.put_slice(&encoded.bytes[encoded.at_mp_reach..]); + } + let attributes_len = out.len() - attributes_at - 2; + out[attributes_at..attributes_at + 2] + .copy_from_slice(&(attributes_len as u16).to_be_bytes()); + if in_body { + write_nlri(routes, with_path_id, &mut out); + } + let length = out.len(); + out[length_at..length_at + 2].copy_from_slice(&(length as u16).to_be_bytes()); + out + }) + .collect() +} + /// The UPDATEs that announce `routes`, all of IP version `version`, as /// `outgoing`: as few as they fit in. `attributes_octets` is how long /// `outgoing`'s attributes are on this session's wire. +/// +/// What [`encoded_announcements`] does, built as messages: kept as what +/// its bytes are checked against. +#[cfg(test)] pub(crate) fn announcements( outgoing: &Outgoing, version: IpVersion, @@ -484,7 +607,7 @@ mod tests { use std::net::Ipv4Addr; use bgp_packet::message::{ - CommunitiesPathAttribute, LargeCommunitiesPathAttribute, LargeCommunity, + CommunitiesPathAttribute, LargeCommunitiesPathAttribute, LargeCommunity, Message, MultiExitDiscPathAttribute, OriginPathAttribute, PathAttributeFlags, }; use bgp_primitives::ip_prefix::IpPrefix; @@ -549,6 +672,106 @@ mod tests { PathAttribute::Communities(CommunitiesPathAttribute(communities.to_vec())) } + /// What `encoded_announcements` writes is, byte for byte, what the + /// messages `announcements` builds encode to: with the next hop in + /// its own attribute (IPv4) and in MP_REACH_NLRI (IPv6), with and + /// without path identifiers, to two-octet and four-octet AS speakers + /// (so with the attributes around NEXT_HOP, MP_REACH_NLRI and + /// AS4_PATH in order), and over as many messages when the routes do + /// not fit in one. + #[test] + fn encoded_announcements_are_the_messages_on_the_wire() { + use std::net::Ipv6Addr; + + for four_octet_asn in [true, false] { + for with_path_id in [false, true] { + for version in [IpVersion::Ipv4, IpVersion::Ipv6] { + let towards = Towards { + four_octet_asn, + ..towards(None, false) + }; + let next_hop = match version { + IpVersion::Ipv4 => NlriNextHop::Ipv4(Ipv4Addr::new(192, 0, 2, 1)), + IpVersion::Ipv6 => NlriNextHop::Ipv6WithLl { + global: Ipv6Addr::new(0x2001, 0xdb8, 0, 0, 0, 0, 0, 1), + link_local: Ipv6Addr::new(0xfe80, 0, 0, 0, 0, 0, 0, 1), + }, + }; + let attributes = vec![ + PathAttribute::Origin(OriginPathAttribute::IGP), + AsPathAttribute::from_asns(vec![4_200_000_000, 3356, 64700]), + PathAttribute::MultiExitDisc(MultiExitDiscPathAttribute(10)), + communities(&[(3356, 100), (64700, 1)]), + PathAttribute::LargeCommunities(LargeCommunitiesPathAttribute { + communities: vec![LargeCommunity { + global_admin: 64700, + data_1: 1, + data_2: 2, + }], + }), + ]; + let routes: Vec = (0..1000u32) + .map(|i| Nlri { + prefix: match version { + IpVersion::Ipv4 => IpPrefix::new( + version, + Ipv4Addr::from(0x0a00_0000 + i * 256).octets(), + 24, + ), + IpVersion::Ipv6 => IpPrefix::new( + version, + Ipv6Addr::new(0x2001, 0xdb8, i as u16, 0, 0, 0, 0, 0).octets(), + 48, + ), + } + .unwrap(), + path_id: if with_path_id { i + 1 } else { 0 }, + }) + .collect(); + + let ctx = ParserContext { + four_octet_asn, + ibgp: false, + add_path: if with_path_id { + vec![AfiSafi::unicast(version)] + } else { + Vec::new() + }, + }; + let mut finished = attributes.clone(); + finish(&mut finished, &towards); + let encoded = EncodedAttributes::encode(&finished, four_octet_asn).unwrap(); + let bytes = encoded_announcements(&encoded, next_hop, version, &routes, &ctx); + + let outgoing = Outgoing { + attributes, + next_hop, + }; + let messages = + announcements(&outgoing, version, &routes, &towards, encoded.octets()); + let expected: Vec = messages + .into_iter() + .map(|update| { + let mut out = BytesMut::new(); + Message::Update(update).to_wire(&ctx, &mut out).unwrap(); + out + }) + .collect(); + + let what = format!( + "{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}"); + } + } + } + } + } + struct Case<'a> { policy: &'a ExportPolicy, role: Option, diff --git a/crates/server/src/session.rs b/crates/server/src/session.rs index 3b7fd66..9603d25 100644 --- a/crates/server/src/session.rs +++ b/crates/server/src/session.rs @@ -17,6 +17,7 @@ use std::time::{Duration, SystemTime}; use bgp_packet::message::Message; use bgp_primitives::ip_prefix::Nlri; use bgp_rib::table::PeerIndex; +use bytes::BytesMut; use tokio::sync::{broadcast, mpsc, watch}; use crate::address::{Endpoint, PeerAddress, socket_text}; @@ -199,7 +200,7 @@ pub struct SessionHandle { pub reflector_client: bool, /// The peer's BGP Identifier. pub peer_router_id: Ipv4Addr, - outbound: mpsc::Sender, + outbound: mpsc::Sender, /// Where the session is asked to end, if it listens for that. ending: Option>, /// Which of the RPKI's checks the peer's routes are subject to, if @@ -219,7 +220,7 @@ impl SessionHandle { local_addr: SocketAddr, negotiated: &Negotiated, rib_peer: PeerIndex, - outbound: mpsc::Sender, + outbound: mpsc::Sender, ) -> Self { Self { peer, @@ -312,10 +313,28 @@ impl SessionHandle { /// space if the session's writer is behind. pub async fn send(&self, msg: Message) -> Result<(), SessionClosed> { self.outbound - .send(msg) + .send(Outbound::Message(msg)) .await .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> { + self.outbound + .send(Outbound::Encoded(update)) + .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. +#[derive(Debug)] +pub enum Outbound { + Message(Message), + Encoded(BytesMut), } #[derive(Debug, Clone)]