rayhaan/ferrix
0
0 Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

export: encode a route's attributes once a session

Browse Source 
The exporter built each UPDATE as a message: the attributes of the
route's set, parsed from the table, were cloned, had the next hop
added, were sorted, and were handed to the session, which encoded them
to the wire with the rest of the message. For a table whose attribute
sets each cover a few routes that is an encoding of the same octets for
nearly every route, and a copy of the message's bytes on the way to the
writer besides. A profile of a production router taking the other
router's table with ADD-PATH had the exporters at 55% of the CPU time,
their dressing of routes at 23%, and memcpy at 12%.

A route's attributes are now encoded once for a session, when it is
first dressed for it, and kept with where in the octets a NEXT_HOP or
an MP_REACH_NLRI belongs so that the attributes stay in order of type
code. An UPDATE is then its header, the encoded octets with the next
hop or the MP_REACH_NLRI spliced in, and the prefixes: written straight
into a buffer that goes to the session as it is, and from the session
to the writer without a copy. The session's queue takes either a
message or an encoded UPDATE; an encoded one for a session that is no
longer established is dropped, as the exporter made it for the session
as it was. What is kept for a route, now a few hundred octets instead
of a kilobyte, lets twice as many be kept. The exporter's maps hash
with foldhash instead of SipHash, and the parser context for an UPDATE
is made once a flush instead of once an UPDATE.

Testing: a new test encodes a thousand routes with each of the forms
an UPDATE takes (IPv4 with the next hop in its own attribute, IPv6 in
MP_REACH_NLRI, with and without path identifiers, to two-octet and
four-octet AS speakers, over several messages) both ways and checks the
octets are the same byte for byte; the message-building form is kept
for the test. The exporter tests decode what the session is handed.
The server tests, the integration tests, the kernel tests and every
lab phase pass. On the mesh benchmark in the Linux VM, three
rounds each alternating with the previous commit, the two routers'
exchange of 7.4 million routes takes 20.6 s of CPU time instead of
25.4 s (-19%) over 9.0 s of wall-clock time instead of 11.1 s, and the
larger of the two tables is across in 7.1 s instead of 9.3 s. The
smaller table's time varies too much from run to run (5-8 s) for a
difference to be read from it.

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
This commit is contained in:
Rayhaan JaufeerallyandClaude Fable 5.1 committed 2026-10-10 13:47:47 +02:00
1 parent f1aca37ebe
commit 381eabd886
7 files changed
+373 -64
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

crates/packet/src/message.rs
+2 -2
View File
@@ -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);
crates/server/Cargo.toml
+1
View File
@@ -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
crates/server/src/advertise.rs
+82 -44
View File
@@ -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<K, V> = std::collections::HashMap<K, V, foldhash::fast::RandomState>;
type HashSet<K> = std::collections::HashSet<K, foldhash::fast::RandomState>;
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<PeerIndex, Option<Source>> = HashMap::new();
let mut sources: HashMap<PeerIndex, Option<Source>> = 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<Nlri>; 2] = Default::default();
let mut batches: HashMap<(usize, IpVersion), (Arc<Dressed>, Vec<Nlri>)> = HashMap::new();
let mut batches: HashMap<(usize, IpVersion), (Arc<Dressed>, Vec<Nlri>)> =
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::<usize>();
// 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<Message>,
outbound: mpsc::Receiver<Outbound>,
/// What the peer holds, going by what it has been sent.
routes: HashSet<IpPrefix>,
_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<IpPrefix>) {
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<Message>,
outbound: mpsc::Receiver<Outbound>,
ending: mpsc::UnboundedReceiver<SessionError>,
_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<Message>) -> HashSet<IpPrefix> {
let mut routes = HashSet::new();
async fn announced(outbound: &mut mpsc::Receiver<Outbound>) -> HashSet<IpPrefix> {
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
crates/server/src/connection.rs
+40 -13
View File
@@ -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<Message>,
outbound: mpsc::Receiver<Outbound>,
/// Where the rest of the server asks for the session to end, and why.
ending: mpsc::UnboundedReceiver<SessionError>,
/// 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<Vec<u8>>),
WriterRoom(mpsc::OwnedPermit<BytesMut>),
/// Messages from the rest of the server to send, as many as were
/// waiting, to be written together.
Outbound(Vec<Message>),
Outbound(Vec<Outbound>),
/// The validation data has changed; these routes of the peer's are
/// now invalid.
Revalidate(Vec<Nlri>),
@@ -648,12 +650,12 @@ struct Session {
frames: FrameReader<OwnedReadHalf>,
/// Batches of messages on their way to the writer task, which writes
/// them to the peer; see [`write_to_peer`].
writer: mpsc::Sender<Vec<u8>>,
writer: mpsc::Sender<BytesMut>,
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<Vec<u8>>,
pending: VecDeque<BytesMut>,
/// What the writer task says if it fails.
write_failed: mpsc::Receiver<SessionError>,
/// 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<Message>,
outbound: mpsc::Sender<Outbound>,
) -> 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<Established>, 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<Vec<u8>>,
mut queue: mpsc::Receiver<BytesMut>,
send_hold: watch::Receiver<Duration>,
failed: mpsc::Sender<SessionError>,
) {
crates/server/src/export.rs
+225 -2
View File
@@ -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<Self> {
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<BytesMut> {
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<Nlri> = (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<BytesMut> = 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<PeerRole>,
crates/server/src/session.rs
+22 -3
View File
@@ -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<Message>,
outbound: mpsc::Sender<Outbound>,
/// Where the session is asked to end, if it listens for that.
ending: Option<mpsc::UnboundedSender<SessionError>>,
/// 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<Message>,
outbound: mpsc::Sender<Outbound>,
) -> 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)]