Review pass on the codec and the table: bugs fixed, encoders simplified

A review of the wire codec and the routing table, with the fixes.

Bugs in the codec:

- A NOTIFICATION with a code or subcode we have no name for (subcode 0,
  "unspecific", is common) failed to decode and was answered as a bad
  message. Every pair now reads, as BgpError::Other.
- Optional parameters in the extended form of RFC 9072 were read and
  written with one-octet parameter lengths; they are two.
- An ORF capability listing more than one address family made the whole
  OPEN malformed.
- Bytes after an OPEN's optional parameters were accepted.
- Extended communities were split into fields by rules that were wrong
  for the common types. They are kept as their eight octets.
- Several length fields were computed in u16 and could wrap on encode.

Simplifications in the codec:

- Lengths are filled in after the value is written, for messages,
  attributes, capabilities and the UPDATE's two fields. The parallel
  wire_len functions, which had to be kept in step by hand, are gone.
  Message::to_wire no longer encodes into a second buffer.
- Type codes for messages, attributes and capabilities are each defined
  once. The strum discriminant enums and their hand-written conversions
  are removed.
- One function splits an attribute off a buffer; the UPDATE decoder, the
  plain attribute parser and the table's key reader all use it.
- The UPDATE decoder's 185-line loop is an AttributeReader with one
  method that decides what becomes of an attribute. Duplicate detection
  is a bitset, not a HashSet allocated per UPDATE.
- ParserContext::four_octet_asn is a bool. The "context not set" error
  path, which treated a programming error as a malformed route, is gone.
- Codec functions take a context only if they read it, and return a
  Result only if they can fail.
- MP_REACH_NLRI and MP_UNREACH_NLRI hold an AfiSafi.
- AddressFamilyId is IpVersion, which is what it was.
- Deserialize is no longer derived on codec types: nothing used it, and
  it bypassed IpPrefix's invariants. IpPrefix::to_wire, which was
  broken and unused, is removed.

Bugs in the table:

- The index of a row's best route was sixteen bits. A peer using
  ADD-PATH could send more routes for a prefix than that counts, and
  the index wrapped. It is 32 bits, at no cost in row size.
- Shards and attribute-store stripes were picked from the top bits of
  the hash, which are the bits hashbrown keeps as the tag it compares
  first. With 64 shards each shard's table had 2 of 128 tag values.
  They are now picked from bits the table does not use.
- remove_peer held the peer list's lock for the whole scan of the
  table, so no session could establish while a peer was removed.
- When more prefixes went at once than were left, the forced trim
  dropped a fixed few entries, reset every reader, and re-ran the tidy
  on every later write. It drops the older half.
- Readers were woken by every change, including a repeated announcement
  and changes their journal does not record.
- Reader::poll with a budget of zero never made progress.

Simplifications in the table:

- The two journals of a shard are one Journals value written through
  three calls; the five places that kept them in step by hand are gone.
  Whether an entry is still live is decided in one place.
- The shortcuts of route selection moved to select.rs, next to the
  invariant they depend on, which is now written down.
- announce and withdraw take routes with or without path identifiers;
  the _paths variants are removed.
- The selection key is read off the wire form only; the second
  implementation over parsed attributes is removed.

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
This commit is contained in:
Rayhaan JaufeerallyandClaude Fable 5.1 committed 2026-10-02 10:32:27 +02:00
1 parent d49bab58a5
commit a8aba52562
30 files changed
+2386 -2878

No files matched your search

+2 -2
View File
@@ -185,7 +185,7 @@ fn into_table(wire: &[Vec<BytesMut>], routes: usize, threads: usize) {
let wire = decoded.attributes_wire.as_deref().expect("kept");
if !update.announced_routes.is_empty() {
let attrs = table.attrs().intern_received(wire, None).expect("valid");
table.announce_paths(peers[at], &attrs, &update.announced_routes);
table.announce(peers[at], &attrs, &update.announced_routes);
}
for attribute in &update.path_attributes {
if let PathAttribute::MpReachNlri(mp) = attribute {
@@ -193,7 +193,7 @@ fn into_table(wire: &[Vec<BytesMut>], routes: usize, threads: usize) {
.attrs()
.intern_received(wire, Some(&mp.next_hop))
.expect("valid");
table.announce_paths(peers[at], &attrs, &mp.prefixes);
table.announce(peers[at], &attrs, &mp.prefixes);
}
}
}
+5 -6
View File
@@ -24,7 +24,7 @@ use bgp_packet::message::{
LargeCommunity, Message, MpReachNlriPathAttribute, MultiExitDiscPathAttribute,
NextHopPathAttribute, NlriNextHop, OriginPathAttribute, PathAttribute, UpdateMessage,
};
use bgp_primitives::constants::{AddressFamilyId, Afi, Safi};
use bgp_primitives::constants::{AfiSafi, IpVersion};
use bgp_primitives::ip_prefix::{IpPrefix, Nlri};
use bgp_primitives::parser::ParserContext;
use bytes::BytesMut;
@@ -133,7 +133,7 @@ impl Prefixes {
let length = rng.pick(&v4_lengths);
// Unicast space, 1.0.0.0 to 223.255.255.255.
let address = 0x0100_0000 + (rng.next_u64() as u32) % 0xdf00_0000;
let prefix = IpPrefix::new(AddressFamilyId::Ipv4, address.to_be_bytes(), length)
let prefix = IpPrefix::new(IpVersion::Ipv4, address.to_be_bytes(), length)
.expect("four bytes cover any IPv4 prefix");
if seen.insert(prefix) {
ipv4.push(prefix);
@@ -144,7 +144,7 @@ impl Prefixes {
let length = rng.pick(&v6_lengths);
// Global unicast, 2000::/3.
let address = (0x2000_u128 << 112) | (u128::from(rng.next_u64()) << 61 >> 3);
let prefix = IpPrefix::new(AddressFamilyId::Ipv6, address.to_be_bytes(), length)
let prefix = IpPrefix::new(IpVersion::Ipv6, address.to_be_bytes(), length)
.expect("sixteen bytes cover any IPv6 prefix");
if seen.insert(prefix) {
ipv6.push(prefix);
@@ -219,8 +219,7 @@ pub fn announcements(prefixes: &Prefixes, shape: Shape, peer: u32) -> Vec<Update
// MP_REACH_NLRI.
path_attributes.extend(later);
path_attributes.push(PathAttribute::MpReachNlri(MpReachNlriPathAttribute {
afi: Afi::IPV6,
safi: Safi::UNICAST,
family: AfiSafi::IPV6_UNICAST,
next_hop: NlriNextHop::Ipv6(Ipv6Addr::new(
0x2001,
0xdb8,
@@ -279,7 +278,7 @@ pub fn encoded(updates: &[UpdateMessage]) -> Vec<BytesMut> {
/// peer with four-octet AS numbers.
pub fn wire_context() -> ParserContext {
ParserContext {
four_octet_asn: Some(true),
four_octet_asn: true,
ibgp: false,
add_path: Vec::new(),
}
+4 -4
View File
@@ -11,7 +11,7 @@ use std::net::{IpAddr, Ipv4Addr, Ipv6Addr};
use std::path::Path;
use bgp_packet::message::{NlriNextHop, PathAttribute};
use bgp_primitives::constants::AddressFamilyId;
use bgp_primitives::constants::IpVersion;
use bgp_primitives::ip_prefix::IpPrefix;
use flate2::read::MultiGzDecoder;
@@ -84,9 +84,9 @@ impl Dump {
}
RIB_IPV4_UNICAST | RIB_IPV6_UNICAST => {
let version = if subtype == RIB_IPV4_UNICAST {
AddressFamilyId::Ipv4
IpVersion::Ipv4
} else {
AddressFamilyId::Ipv6
IpVersion::Ipv6
};
rib_entries(&mut record, version, &mut dump, &mut interned)
.ok_or_else(malformed)?;
@@ -162,7 +162,7 @@ fn peer_index_table(record: &mut Cursor<'_>) -> Option<Vec<DumpPeer>> {
fn rib_entries(
record: &mut Cursor<'_>,
version: AddressFamilyId,
version: IpVersion,
dump: &mut Dump,
interned: &mut HashMap<Box<[u8]>, u32>,
) -> Option<()> {
+5 -7
View File
@@ -69,9 +69,7 @@ impl TestPeer {
.read_exact(&mut buf[BGP_HEADER_LEN as usize..])
.await?;
let (_, msg) = Message::from_wire(&self.ctx, &buf)
.map_err(|e| eyre!("Failed to parse BGP message: {e:?}"))?;
Ok(msg)
Message::decode(&self.ctx, &buf).map_err(|e| eyre!("Failed to parse BGP message: {e}"))
}
/// Read messages until an UPDATE arrives, skipping KEEPALIVEs. Any other
@@ -95,7 +93,7 @@ impl TestPeer {
matches!(opt, OpenOption::Capabilities(caps)
if caps.iter().any(|c| matches!(c, Capability::FourByteAsn { .. })))
});
self.ctx.four_octet_asn = Some(four_byte && self.offered_four_octet_asn);
self.ctx.four_octet_asn = four_byte && self.offered_four_octet_asn;
Ok(o)
}
other => bail!("Expected OPEN, got {:?}", other),
@@ -138,9 +136,9 @@ impl TestPeer {
self.offered_four_octet_asn = capabilities
.iter()
.any(|c| matches!(c, Capability::FourByteAsn { .. }));
if let Some(negotiated) = &mut self.ctx.four_octet_asn {
*negotiated &= self.offered_four_octet_asn;
}
// If the server's OPEN has been read already, what was settled
// then still depends on what is offered now.
self.ctx.four_octet_asn &= self.offered_four_octet_asn;
let asn_field = if local_asn > u16::MAX as u32 {
23456
} else {
+1 -1
View File
@@ -142,7 +142,7 @@ impl TestServer {
pub fn received_routes(&self) -> Vec<ReceivedRoute> {
let mut routes = Vec::new();
self.rib.for_each_route(|route| {
let (attributes, next_hop) = route.attrs.attributes();
let (attributes, next_hop) = route.attrs.attributes().expect("stored attributes parse");
routes.push(ReceivedRoute {
prefix: route.prefix,
peer: route.peer,
@@ -10,7 +10,7 @@ use bgp_packet::message::{
NextHopPathAttribute, NlriNextHop, NotificationMessage, OriginPathAttribute, PathAttribute,
UpdateMessage,
};
use bgp_primitives::constants::{AddressFamilyId, Afi, Safi};
use bgp_primitives::constants::{AfiSafi, IpVersion};
use bgp_primitives::ip_prefix::IpPrefix;
use eyre::Result;
@@ -213,7 +213,7 @@ async fn test_message_split_across_a_keepalive_tick_is_still_one_message() -> Re
let mut peer = TestPeer::connect(server.addr).await?;
peer.handshake(65001, Ipv4Addr::new(10, 0, 0, 1), 3).await?;
let prefix = IpPrefix::new(AddressFamilyId::Ipv4, vec![10], 8)?;
let prefix = IpPrefix::new(IpVersion::Ipv4, vec![10], 8)?;
let mut update = bytes::BytesMut::new();
Message::Update(UpdateMessage {
withdrawn_routes: vec![],
@@ -340,7 +340,7 @@ async fn test_server_accepts_update() -> Result<()> {
withdrawn_routes: vec![],
path_attributes: mandatory_attributes(),
// 10.0.0.0/8 — one byte of prefix for a /8.
announced_routes: vec![IpPrefix::new(AddressFamilyId::Ipv4, vec![10], 8)?.into()],
announced_routes: vec![IpPrefix::new(IpVersion::Ipv4, vec![10], 8)?.into()],
});
peer.send(&update).await?;
@@ -358,13 +358,12 @@ async fn test_server_accepts_ipv6_update() -> Result<()> {
.await?;
// 2001:db8::/32
let prefix = IpPrefix::new(AddressFamilyId::Ipv6, vec![0x20, 0x01, 0x0d, 0xb8], 32)?;
let prefix = IpPrefix::new(IpVersion::Ipv6, vec![0x20, 0x01, 0x0d, 0xb8], 32)?;
let update = Message::Update(UpdateMessage {
withdrawn_routes: vec![],
path_attributes: with_origin_and_as_path(PathAttribute::MpReachNlri(
MpReachNlriPathAttribute {
afi: Afi::IPV6,
safi: Safi::UNICAST,
family: AfiSafi::IPV6_UNICAST,
next_hop: NlriNextHop::Ipv6("2001:db8::1".parse().unwrap()),
prefixes: vec![prefix.into()],
},
@@ -392,7 +391,7 @@ async fn test_rib_records_ipv4_announce() -> Result<()> {
peer.handshake(65001, Ipv4Addr::new(10, 0, 0, 1), 90)
.await?;
let prefix = IpPrefix::new(AddressFamilyId::Ipv4, vec![10], 8)?;
let prefix = IpPrefix::new(IpVersion::Ipv4, vec![10], 8)?;
peer.send(&Message::Update(UpdateMessage {
withdrawn_routes: vec![],
path_attributes: mandatory_attributes(),
@@ -424,7 +423,7 @@ async fn test_rib_records_ipv4_withdraw() -> Result<()> {
peer.handshake(65001, Ipv4Addr::new(10, 0, 0, 1), 90)
.await?;
let prefix = IpPrefix::new(AddressFamilyId::Ipv4, vec![10], 8)?;
let prefix = IpPrefix::new(IpVersion::Ipv4, vec![10], 8)?;
// Announce first.
peer.send(&Message::Update(UpdateMessage {
@@ -467,13 +466,12 @@ async fn test_rib_records_ipv6_announce() -> Result<()> {
peer.handshake(65001, Ipv4Addr::new(10, 0, 0, 1), 90)
.await?;
let prefix = IpPrefix::new(AddressFamilyId::Ipv6, vec![0x20, 0x01, 0x0d, 0xb8], 32)?;
let prefix = IpPrefix::new(IpVersion::Ipv6, vec![0x20, 0x01, 0x0d, 0xb8], 32)?;
peer.send(&Message::Update(UpdateMessage {
withdrawn_routes: vec![],
path_attributes: with_origin_and_as_path(PathAttribute::MpReachNlri(
MpReachNlriPathAttribute {
afi: Afi::IPV6,
safi: Safi::UNICAST,
family: AfiSafi::IPV6_UNICAST,
next_hop: NlriNextHop::Ipv6("2001:db8::1".parse().unwrap()),
prefixes: vec![prefix.into()],
},
@@ -506,15 +504,14 @@ async fn test_rib_records_ipv6_withdraw() -> Result<()> {
peer.handshake(65001, Ipv4Addr::new(10, 0, 0, 1), 90)
.await?;
let prefix = IpPrefix::new(AddressFamilyId::Ipv6, vec![0x20, 0x01, 0x0d, 0xb8], 32)?;
let prefix = IpPrefix::new(IpVersion::Ipv6, vec![0x20, 0x01, 0x0d, 0xb8], 32)?;
// Announce first.
peer.send(&Message::Update(UpdateMessage {
withdrawn_routes: vec![],
path_attributes: with_origin_and_as_path(PathAttribute::MpReachNlri(
MpReachNlriPathAttribute {
afi: Afi::IPV6,
safi: Safi::UNICAST,
family: AfiSafi::IPV6_UNICAST,
next_hop: NlriNextHop::Ipv6("2001:db8::1".parse().unwrap()),
prefixes: vec![prefix.into()],
},
@@ -528,8 +525,7 @@ async fn test_rib_records_ipv6_withdraw() -> Result<()> {
peer.send(&Message::Update(UpdateMessage {
withdrawn_routes: vec![],
path_attributes: vec![PathAttribute::MpUnreachNlri(MpUnreachNlriPathAttribute {
afi: Afi::IPV6,
safi: Safi::UNICAST,
family: AfiSafi::IPV6_UNICAST,
prefixes: vec![prefix.into()],
})],
announced_routes: vec![],
+2 -2
View File
@@ -7,7 +7,7 @@ use bgp_integration_tests::test_peer::TestPeer;
use bgp_integration_tests::test_server::TestServer;
use bgp_packet::message::{Message, PathAttribute};
use bgp_packet::open::{Capability, OpenOption};
use bgp_primitives::constants::{Afi, Safi};
use bgp_primitives::constants::{Afi, AfiSafi, Safi};
use bgp_primitives::ip_prefix::IpPrefix;
use bgp_rib::{LocalRoute, LocalRouteAttrs};
use eyre::Result;
@@ -133,7 +133,7 @@ async fn test_ipv6_prefix_update_shape() -> Result<()> {
.path_attributes
.iter()
.any(|a| matches!(a, PathAttribute::MpReachNlri(m)
if m.afi == Afi::IPV6 && m.prefixes[0].to_string() == "2001:db8:1::/48"))
if m.family == AfiSafi::IPV6_UNICAST && m.prefixes[0].to_string() == "2001:db8:1::/48"))
);
let _ = Ipv4Addr::LOCALHOST;
Ok(())
@@ -58,8 +58,7 @@ fn dual_stack_update() -> UpdateMessage {
AsPathAttribute::from_asns(vec![PEER_ASN]),
PathAttribute::NextHop(NextHopPathAttribute(Ipv4Addr::new(192, 0, 2, 1))),
PathAttribute::MpReachNlri(MpReachNlriPathAttribute {
afi: Afi::IPV6,
safi: Safi::UNICAST,
family: AfiSafi::IPV6_UNICAST,
next_hop: NlriNextHop::Ipv6(Ipv6Addr::new(0x2001, 0xdb8, 0, 0, 0, 0, 0, 1)),
prefixes: vec![prefix("2001:db8:1::/48").into()],
}),
+46 -97
View File
@@ -1,4 +1,4 @@
use serde::{Deserialize, Serialize};
use serde::Serialize;
use strum::FromRepr;
use thiserror::Error;
@@ -23,67 +23,57 @@ pub struct DecodeError {
pub detail: String,
}
#[derive(Debug, Error)]
#[error("Unknown BGP error code {code} subcode {subcode}")]
pub struct UnknownErrorCode {
pub code: u8,
pub subcode: u8,
}
#[derive(Debug, Clone, PartialEq, Eq, Error, Serialize, Deserialize)]
#[repr(u8)]
/// The error a NOTIFICATION reports: its code and subcode (RFC 4271 §4.5).
#[derive(Debug, Clone, PartialEq, Eq, Error, Serialize)]
pub enum BgpError {
#[error("Message header error")]
MsgHeader(MsgHeaderSubcode) = 1,
#[error("Message header error: {0}")]
MsgHeader(MsgHeaderSubcode),
#[error("Open message error")]
OpenMsg(OpenMsgSubcode) = 2,
#[error("Open message error: {0}")]
OpenMsg(OpenMsgSubcode),
#[error("Update message error")]
UpdateMsg(UpdateMsgSubcode) = 3,
#[error("Update message error: {0}")]
UpdateMsg(UpdateMsgSubcode),
#[error("Hold timer expired")]
HoldTimer = 4,
HoldTimer,
#[error("Finite state machine error")]
FsmError(FsmSubcode) = 5,
#[error("Finite state machine error: {0}")]
FsmError(FsmSubcode),
#[error("Cease")]
Cease(CeaseSubcode) = 6,
#[error("Cease: {0}")]
Cease(CeaseSubcode),
#[error("Route refresh message error")]
RouteRefresh(RouteRefreshSubcode) = 7,
#[error("Route refresh message error: {0}")]
RouteRefresh(RouteRefreshSubcode),
#[error("Send hold timer expired")]
SendHoldTimer = 8,
SendHoldTimer,
/// A code or subcode this implementation has no name for. A peer may
/// send any: new ones are assigned, and zero ("unspecific") is always
/// allowed. The session ends the same way, and the numbers are kept
/// for the log.
#[error("error code {code} subcode {subcode}")]
Other { code: u8, subcode: u8 },
}
impl BgpError {
pub fn from_code_subcode(code: u8, subcode: u8) -> Result<Self, UnknownErrorCode> {
let err = || UnknownErrorCode { code, subcode };
match code {
1 => Ok(Self::MsgHeader(
MsgHeaderSubcode::try_from(subcode).map_err(|_| err())?,
)),
2 => Ok(Self::OpenMsg(
OpenMsgSubcode::try_from(subcode).map_err(|_| err())?,
)),
3 => Ok(Self::UpdateMsg(
UpdateMsgSubcode::try_from(subcode).map_err(|_| err())?,
)),
4 => Ok(Self::HoldTimer),
5 => Ok(Self::FsmError(
FsmSubcode::try_from(subcode).map_err(|_| err())?,
)),
6 => Ok(Self::Cease(
CeaseSubcode::try_from(subcode).map_err(|_| err())?,
)),
7 => Ok(Self::RouteRefresh(
RouteRefreshSubcode::try_from(subcode).map_err(|_| err())?,
)),
8 => Ok(Self::SendHoldTimer),
_ => Err(err()),
}
/// The error for a code and subcode off the wire. Every pair has one,
/// so that a peer's NOTIFICATION can always be read.
pub fn from_code_subcode(code: u8, subcode: u8) -> Self {
let known = match code {
1 => MsgHeaderSubcode::from_repr(subcode).map(Self::MsgHeader),
2 => OpenMsgSubcode::from_repr(subcode).map(Self::OpenMsg),
3 => UpdateMsgSubcode::from_repr(subcode).map(Self::UpdateMsg),
4 => (subcode == 0).then_some(Self::HoldTimer),
5 => FsmSubcode::from_repr(subcode).map(Self::FsmError),
6 => CeaseSubcode::from_repr(subcode).map(Self::Cease),
7 => RouteRefreshSubcode::from_repr(subcode).map(Self::RouteRefresh),
8 => (subcode == 0).then_some(Self::SendHoldTimer),
_ => None,
};
known.unwrap_or(Self::Other { code, subcode })
}
pub fn code_subcode(&self) -> (u8, u8) {
@@ -96,11 +86,12 @@ impl BgpError {
Self::Cease(s) => (6, *s as u8),
Self::RouteRefresh(s) => (7, *s as u8),
Self::SendHoldTimer => (8, 0),
Self::Other { code, subcode } => (*code, *subcode),
}
}
}
#[derive(Debug, Clone, Copy, PartialEq, Eq, Error, Serialize, Deserialize, FromRepr)]
#[derive(Debug, Clone, Copy, PartialEq, Eq, Error, Serialize, FromRepr)]
#[repr(u8)]
pub enum MsgHeaderSubcode {
#[error("Connection not synchronized")]
@@ -113,14 +104,7 @@ pub enum MsgHeaderSubcode {
BadMessageType = 3,
}
impl TryFrom<u8> for MsgHeaderSubcode {
type Error = ();
fn try_from(v: u8) -> Result<Self, ()> {
Self::from_repr(v).ok_or(())
}
}
#[derive(Debug, Clone, Copy, PartialEq, Eq, Error, Serialize, Deserialize, FromRepr)]
#[derive(Debug, Clone, Copy, PartialEq, Eq, Error, Serialize, FromRepr)]
#[repr(u8)]
pub enum OpenMsgSubcode {
#[error("Unspecific")]
@@ -148,14 +132,7 @@ pub enum OpenMsgSubcode {
RoleMismatch = 8,
}
impl TryFrom<u8> for OpenMsgSubcode {
type Error = ();
fn try_from(v: u8) -> Result<Self, ()> {
Self::from_repr(v).ok_or(())
}
}
#[derive(Debug, Clone, Copy, PartialEq, Eq, Error, Serialize, Deserialize, FromRepr)]
#[derive(Debug, Clone, Copy, PartialEq, Eq, Error, Serialize, FromRepr)]
#[repr(u8)]
pub enum UpdateMsgSubcode {
#[error("Malformed attribute list")]
@@ -189,14 +166,7 @@ pub enum UpdateMsgSubcode {
MalformedAsPath = 11,
}
impl TryFrom<u8> for UpdateMsgSubcode {
type Error = ();
fn try_from(v: u8) -> Result<Self, ()> {
Self::from_repr(v).ok_or(())
}
}
#[derive(Debug, Clone, Copy, PartialEq, Eq, Error, Serialize, Deserialize, FromRepr)]
#[derive(Debug, Clone, Copy, PartialEq, Eq, Error, Serialize, FromRepr)]
#[repr(u8)]
pub enum FsmSubcode {
#[error("Received an unexpected message in OpenSent state")]
@@ -209,14 +179,7 @@ pub enum FsmSubcode {
UnexpectedEstablished = 3,
}
impl TryFrom<u8> for FsmSubcode {
type Error = ();
fn try_from(v: u8) -> Result<Self, ()> {
Self::from_repr(v).ok_or(())
}
}
#[derive(Debug, Clone, Copy, PartialEq, Eq, Error, Serialize, Deserialize, FromRepr)]
#[derive(Debug, Clone, Copy, PartialEq, Eq, Error, Serialize, FromRepr)]
#[repr(u8)]
pub enum CeaseSubcode {
#[error("Maximum number of prefixes reached")]
@@ -250,23 +213,9 @@ pub enum CeaseSubcode {
BfdDown = 10,
}
impl TryFrom<u8> for CeaseSubcode {
type Error = ();
fn try_from(v: u8) -> Result<Self, ()> {
Self::from_repr(v).ok_or(())
}
}
#[derive(Debug, Clone, Copy, PartialEq, Eq, Error, Serialize, Deserialize, FromRepr)]
#[derive(Debug, Clone, Copy, PartialEq, Eq, Error, Serialize, FromRepr)]
#[repr(u8)]
pub enum RouteRefreshSubcode {
#[error("Invalid message length")]
InvalidLength = 1,
}
impl TryFrom<u8> for RouteRefreshSubcode {
type Error = ();
fn try_from(v: u8) -> Result<Self, ()> {
Self::from_repr(v).ok_or(())
}
}
+716 -1216
View File
File diff suppressed because it is too large. Load diff
+531 -593
View File
File diff suppressed because it is too large. Load diff
+228 -236
View File
@@ -19,16 +19,15 @@
//! be acted on, with treat-as-withdraw and attribute discard already carried
//! out, together with a report of what was wrong, for the log.
use std::collections::HashSet;
use bgp_primitives::constants::{AddressFamilyId, AfiSafi};
use bgp_primitives::constants::{AfiSafi, IpVersion};
use bgp_primitives::ip_prefix::Nlri;
use bgp_primitives::parser::{BgpParserError, ParserContext};
use bgp_primitives::parser::ParserContext;
use crate::errors::{BgpError, DecodeError, UpdateMsgSubcode};
use crate::message::attr_type::*;
use crate::message::{
MpUnreachNlriPathAttribute, PathAttribute, PathAttributeDiscriminants, PathAttributeFlags,
UpdateMessage, defined_flags, describe, parse_nlri,
MpUnreachNlriPathAttribute, NotAnAttribute, PathAttribute, RawAttribute, UpdateMessage,
defined_flags, describe, parse_nlri_list, split_attribute,
};
/// How an error in an UPDATE was handled, short of resetting the session.
@@ -70,25 +69,12 @@ pub struct DecodedUpdate {
/// again. MP_REACH_NLRI and MP_UNREACH_NLRI are left out, since they
/// name prefixes, and so is every attribute that was discarded.
///
/// `None` when there are no routes for them to belong to, or when the
/// bytes are not in the one form such a table keeps: on a session with
/// two-octet AS numbers.
/// `None` when the UPDATE has no such attributes, when its routes were
/// withdrawn because of an error, or when the bytes are not in the one
/// form such a table keeps: on a session with two-octet AS numbers.
pub attributes_wire: Option<Vec<u8>>,
}
const ORIGIN: u8 = 1;
const AS_PATH: u8 = 2;
const NEXT_HOP: u8 = 3;
const LOCAL_PREF: u8 = 5;
const ATOMIC_AGGREGATE: u8 = 6;
const AGGREGATOR: u8 = 7;
const ORIGINATOR_ID: u8 = 9;
const CLUSTER_LIST: u8 = 10;
const MP_REACH_NLRI: u8 = 14;
const MP_UNREACH_NLRI: u8 = 15;
const AS4_PATH: u8 = 17;
const AS4_AGGREGATOR: u8 = 18;
/// How a malformed value of the attribute with this type code is handled
/// (RFC 7606 §7, and RFC 6793 §6 for the AS4 attributes). `None` means the
/// session is reset: an error inside MP_REACH_NLRI or MP_UNREACH_NLRI leaves
@@ -115,22 +101,37 @@ fn reset(subcode: UpdateMsgSubcode, data: Vec<u8>, detail: String) -> DecodeErro
/// Parses a field of the message body that is nothing but prefixes. An
/// error here cannot be recovered from: the prefixes are what
/// treat-as-withdraw would act on.
fn prefixes(ctx: &ParserContext, field: &str, mut buf: &[u8]) -> Result<Vec<Nlri>, DecodeError> {
fn prefixes(ctx: &ParserContext, field: &str, buf: &[u8]) -> Result<Vec<Nlri>, DecodeError> {
// The prefixes in the body are IPv4 unicast (RFC 4271 §4.3).
let with_path_id = ctx.has_add_path(AfiSafi::IPV4_UNICAST);
let mut prefixes = Vec::new();
while !buf.is_empty() {
let (rest, prefix) = parse_nlri(AddressFamilyId::Ipv4, with_path_id, buf).map_err(|e| {
reset(
UpdateMsgSubcode::InvalidNetworkField,
vec![],
format!("malformed {field}: {}", describe(&e)),
)
})?;
prefixes.push(prefix);
buf = rest;
match parse_nlri_list(IpVersion::Ipv4, with_path_id, buf) {
Ok((_, prefixes)) => Ok(prefixes),
Err(e) => Err(reset(
UpdateMsgSubcode::InvalidNetworkField,
vec![],
format!("malformed {field}: {}", describe(&e)),
)),
}
Ok(prefixes)
}
/// Splits off the field at the front of `buf` that begins with its own
/// two-octet length: the field, and what follows it. The two such fields
/// frame an UPDATE; if they do not fit, there is no telling where anything
/// is (RFC 7606 §3.e).
fn length_prefixed<'a>(buf: &'a [u8], what: &str) -> Result<(&'a [u8], &'a [u8]), DecodeError> {
let framing = |detail: String| {
reset(
UpdateMsgSubcode::MalformedAttrs,
vec![],
format!("malformed UPDATE: {detail}"),
)
};
let Some((length, rest)) = buf.split_first_chunk::<2>() else {
return Err(framing(format!("no room for the {what} length")));
};
let length = usize::from(u16::from_be_bytes(*length));
rest.split_at_checked(length)
.ok_or_else(|| framing(format!("{what} length {length} exceeds the message")))
}
impl UpdateMessage {
@@ -140,66 +141,42 @@ impl UpdateMessage {
/// `Err` is an error that resets the session, with the NOTIFICATION to
/// send. Every other error is handled here and listed in the result.
pub fn decode(ctx: &ParserContext, body: &[u8]) -> Result<DecodedUpdate, DecodeError> {
// The two length fields frame the message. If they do not fit, there
// is no telling where anything is (RFC 7606 §3.e).
let framing = |detail: &str| {
reset(
UpdateMsgSubcode::MalformedAttrs,
vec![],
format!("malformed UPDATE: {detail}"),
)
};
let split_field = |buf: &'_ [u8], what: &str| -> Result<(usize, usize), DecodeError> {
let length = buf
.get(..2)
.map(|l| usize::from(u16::from_be_bytes([l[0], l[1]])))
.ok_or_else(|| framing(&format!("no room for the {what} length")))?;
if buf.len() - 2 < length {
return Err(framing(&format!(
"{what} length {length} exceeds the message"
)));
}
Ok((2, 2 + length))
};
let (start, end) = split_field(body, "withdrawn routes")?;
let withdrawn_buf = &body[start..end];
let body = &body[end..];
let (start, end) = split_field(body, "path attribute")?;
let attributes_buf = &body[start..end];
let nlri_buf = &body[end..];
let (withdrawn_buf, body) = length_prefixed(body, "withdrawn routes")?;
let (attributes_buf, nlri_buf) = length_prefixed(body, "path attribute")?;
let withdrawn_routes = prefixes(ctx, "withdrawn routes", withdrawn_buf)?;
let announced_routes = prefixes(ctx, "NLRI", nlri_buf)?;
let mut errors = Vec::new();
let mut kept_wire = Vec::with_capacity(attributes_buf.len());
let path_attributes = attributes(ctx, attributes_buf, &mut errors, &mut kept_wire)?;
let mut reader = AttributeReader::new(ctx, attributes_buf.len());
reader.read_all(attributes_buf)?;
let AttributeReader {
attributes: path_attributes,
mut errors,
kept_wire,
..
} = reader;
let mut update = UpdateMessage {
withdrawn_routes,
path_attributes,
announced_routes,
};
if !errors
let mut withdrawn = errors
.iter()
.any(|e| e.handling == Handling::TreatAsWithdraw)
&& let Some(missing) = missing_mandatory(ctx, &update)
{
.any(|e| e.handling == Handling::TreatAsWithdraw);
if !withdrawn && let Some(missing) = missing_mandatory(ctx, &update) {
errors.push(UpdateError {
handling: Handling::TreatAsWithdraw,
attribute: Some(missing),
detail: format!("mandatory attribute {missing} is missing"),
});
withdrawn = true;
}
let withdrawn = errors
.iter()
.any(|e| e.handling == Handling::TreatAsWithdraw);
if withdrawn {
update = as_withdrawal(update);
}
let attributes_wire =
(!withdrawn && ctx.four_octet_asn == Some(true) && !kept_wire.is_empty())
.then_some(kept_wire);
(!withdrawn && ctx.four_octet_asn && !kept_wire.is_empty()).then_some(kept_wire);
Ok(DecodedUpdate {
update,
errors,
@@ -208,86 +185,112 @@ impl UpdateMessage {
}
}
/// Reads the path attributes, leaving out the ones that are discarded. The
/// bytes of the ones that are kept and are a property of a route (all but
/// MP_REACH_NLRI and MP_UNREACH_NLRI) are added to `kept_wire`.
fn attributes(
ctx: &ParserContext,
mut buf: &[u8],
errors: &mut Vec<UpdateError>,
kept_wire: &mut Vec<u8>,
) -> Result<Vec<PathAttribute>, DecodeError> {
let mut attributes = Vec::new();
let mut seen = HashSet::new();
let mut error = |handling, attribute: Option<u8>, detail: String| {
errors.push(UpdateError {
/// What to do with an attribute, once it is clear.
enum Verdict {
/// Use it. `route_property` is false for one that was read only to find
/// the routes it names, and is no part of the routes' attributes.
Keep {
attribute: PathAttribute,
route_property: bool,
},
/// Leave it out, having noted why.
Skip,
}
/// Reads the path attributes of one UPDATE, leaving out the ones that are
/// discarded and noting every error that does not reset the session.
struct AttributeReader<'a> {
ctx: &'a ParserContext,
attributes: Vec<PathAttribute>,
errors: Vec<UpdateError>,
/// The bytes of the attributes that are kept and are a property of a
/// route: all but MP_REACH_NLRI and MP_UNREACH_NLRI.
kept_wire: Vec<u8>,
/// The type codes met so far, one bit each.
seen: [u64; 4],
}
impl<'a> AttributeReader<'a> {
fn new(ctx: &'a ParserContext, wire_len: usize) -> Self {
Self {
ctx,
attributes: Vec::new(),
errors: Vec::new(),
kept_wire: Vec::with_capacity(wire_len),
seen: [0; 4],
}
}
fn error(&mut self, handling: Handling, attribute: Option<u8>, detail: String) {
self.errors.push(UpdateError {
handling,
attribute,
detail,
})
};
});
}
while !buf.is_empty() {
// An attribute that does not fit in what is left of the attributes
// (RFC 7606 §4). Where the next one would start is unknown, so the
// rest is not read. Routes found so far are withdrawn.
let header = match buf {
[flags, type_code, rest @ ..] => {
let flags = PathAttributeFlags(*flags);
match (flags.extended_length(), rest) {
(true, [hi, lo, rest @ ..]) => Some((
flags,
*type_code,
usize::from(u16::from_be_bytes([*hi, *lo])),
rest,
)),
(false, [length, rest @ ..]) => {
Some((flags, *type_code, usize::from(*length), rest))
}
_ => None,
/// Notes that `type_code` has been met, and returns whether it had
/// been already.
fn seen_before(&mut self, type_code: u8) -> bool {
let (word, bit) = (usize::from(type_code / 64), 1u64 << (type_code % 64));
let before = self.seen[word] & bit != 0;
self.seen[word] |= bit;
before
}
fn read_all(&mut self, mut buf: &[u8]) -> Result<(), DecodeError> {
while !buf.is_empty() {
let (raw, rest) = match split_attribute(buf) {
Ok(split) => split,
Err(e) => {
// An attribute that does not fit in what is left of the
// attributes (RFC 7606 §4). Where the next one would
// start is unknown, so the rest is not read. Routes
// found so far are withdrawn.
let attribute = match e {
NotAnAttribute::NoHeader => None,
NotAnAttribute::Truncated { type_code, .. } => Some(type_code),
};
self.error(Handling::TreatAsWithdraw, attribute, e.to_string());
return Ok(());
}
};
buf = rest;
if let Verdict::Keep {
attribute,
route_property,
} = self.judge(&raw)?
{
if route_property {
self.kept_wire.extend_from_slice(raw.wire);
}
self.attributes.push(attribute);
}
_ => None,
};
let Some((flags, type_code, length, rest)) = header else {
error(
Handling::TreatAsWithdraw,
None,
format!("{} octets are too few for an attribute", buf.len()),
);
break;
};
if rest.len() < length {
error(
Handling::TreatAsWithdraw,
Some(type_code),
format!(
"attribute {type_code} has length {length} but only {} octets remain",
rest.len()
),
);
break;
}
let (value, rest) = rest.split_at(length);
let attribute_wire = &buf[..buf.len() - rest.len()];
buf = rest;
Ok(())
}
/// Decides what becomes of one attribute. `Err` resets the session.
fn judge(&mut self, raw: &RawAttribute<'_>) -> Result<Verdict, DecodeError> {
let (flags, type_code, wire) = (&raw.flags, raw.type_code, raw.wire);
let names_routes = matches!(type_code, MP_REACH_NLRI | MP_UNREACH_NLRI);
// RFC 7606 §3.g: a second MP_REACH_NLRI or MP_UNREACH_NLRI makes it
// unclear which routes are meant; a second anything else is ignored.
if !seen.insert(type_code) {
if matches!(type_code, MP_REACH_NLRI | MP_UNREACH_NLRI) {
if self.seen_before(type_code) {
if names_routes {
return Err(reset(
UpdateMsgSubcode::MalformedAttrs,
vec![],
format!("attribute {type_code} appears more than once"),
));
}
error(
self.error(
Handling::AttributeDiscard,
Some(type_code),
format!("attribute {type_code} appears more than once; kept the first"),
);
continue;
return Ok(Verdict::Skip);
}
let Some((optional, transitive)) = defined_flags(type_code) else {
@@ -297,23 +300,20 @@ fn attributes(
if !flags.optional() {
return Err(reset(
UpdateMsgSubcode::UnrecognizedWellKnownAttr,
attribute_wire.to_vec(),
wire.to_vec(),
format!("unrecognised well-known attribute {type_code}"),
));
}
attributes.push(PathAttribute::UnknownPathAttribute {
flags,
type_code,
payload: value.to_vec(),
return Ok(Verdict::Keep {
attribute: opaque(raw),
route_property: true,
});
kept_wire.extend_from_slice(attribute_wire);
continue;
};
let flags_are_wrong = flags.optional() != optional || flags.transitive() != transitive;
if flags_are_wrong {
// RFC 7606 §3.c. The value is still read below if it is one that
// names routes, since those have to be withdrawn.
error(
self.error(
Handling::TreatAsWithdraw,
Some(type_code),
format!(
@@ -321,99 +321,89 @@ fn attributes(
flags.0
),
);
if !matches!(type_code, MP_REACH_NLRI | MP_UNREACH_NLRI) {
continue;
if !names_routes {
return Ok(Verdict::Skip);
}
}
// Attributes that only mean something between internal peers are
// dropped when an external one sends them (RFC 7606 §7.5, §7.9,
// §7.10), well formed or not.
if !ctx.ibgp && matches!(type_code, LOCAL_PREF | ORIGINATOR_ID | CLUSTER_LIST) {
error(
if !self.ctx.ibgp && matches!(type_code, LOCAL_PREF | ORIGINATOR_ID | CLUSTER_LIST) {
self.error(
Handling::AttributeDiscard,
Some(type_code),
format!("attribute {type_code} from an external peer"),
);
continue;
return Ok(Verdict::Skip);
}
// A speaker with four-octet AS numbers has no use for AS4_PATH and
// AS4_AGGREGATOR and must not send them to another (RFC 6793 §6).
if ctx.four_octet_asn == Some(true) && matches!(type_code, AS4_PATH | AS4_AGGREGATOR) {
error(
if self.ctx.four_octet_asn && matches!(type_code, AS4_PATH | AS4_AGGREGATOR) {
self.error(
Handling::AttributeDiscard,
Some(type_code),
format!("attribute {type_code} on a session with four-octet AS numbers"),
);
continue;
return Ok(Verdict::Skip);
}
let parsed = match type_code {
// Known by their definition but not modelled: kept as bytes,
// once their length has been checked.
ORIGINATOR_ID | CLUSTER_LIST | AS4_AGGREGATOR => {
let well_formed = match type_code {
ORIGINATOR_ID => value.len() == 4,
CLUSTER_LIST => !value.is_empty() && value.len().is_multiple_of(4),
_ => value.len() == 8,
};
if well_formed {
Ok(PathAttribute::UnknownPathAttribute {
flags,
type_code,
payload: value.to_vec(),
})
} else {
Err(format!("a length of {} is not valid", value.len()))
}
}
_ => {
PathAttribute::parse_value(ctx, PathAttributeDiscriminants::from(type_code), value)
.map_err(|e: nom::Err<BgpParserError<&[u8]>>| describe(&e))
}
};
match parsed {
Ok(attribute) => {
// With wrong flags it was read only to find its routes.
if !flags_are_wrong && !matches!(type_code, MP_REACH_NLRI | MP_UNREACH_NLRI) {
kept_wire.extend_from_slice(attribute_wire);
}
attributes.push(attribute);
}
Err(detail) => match handling_of_malformed(type_code) {
Some(handling) => error(
handling,
Some(type_code),
format!("malformed attribute {type_code}: {detail}"),
),
None => {
return Err(reset(
match self.parse(raw) {
Ok(attribute) => Ok(Verdict::Keep {
attribute,
route_property: !flags_are_wrong && !names_routes,
}),
Err(detail) => {
let detail = format!("malformed attribute {type_code}: {detail}");
match handling_of_malformed(type_code) {
Some(handling) => {
self.error(handling, Some(type_code), detail);
Ok(Verdict::Skip)
}
None => Err(reset(
UpdateMsgSubcode::OptionalAttribute,
attribute_wire.to_vec(),
format!("malformed attribute {type_code}: {detail}"),
));
wire.to_vec(),
detail,
)),
}
},
}
}
}
Ok(attributes)
/// Reads the value of an attribute whose definition is known. The error
/// is what was wrong with it, as text.
fn parse(&self, raw: &RawAttribute<'_>) -> Result<PathAttribute, String> {
let value = raw.value;
// Known by their definition but not modelled: kept as bytes, once
// their length has been checked.
let length_is_valid = match raw.type_code {
ORIGINATOR_ID => Some(value.len() == 4),
CLUSTER_LIST => Some(!value.is_empty() && value.len().is_multiple_of(4)),
AS4_AGGREGATOR => Some(value.len() == 8),
_ => None,
};
match length_is_valid {
Some(true) => Ok(opaque(raw)),
Some(false) => Err(format!("a length of {} is not valid", value.len())),
None => PathAttribute::parse_value(self.ctx, raw.flags.clone(), raw.type_code, value)
.map_err(|e| describe(&e)),
}
}
}
/// The attribute kept as bytes.
fn opaque(raw: &RawAttribute<'_>) -> PathAttribute {
PathAttribute::UnknownPathAttribute {
flags: raw.flags.clone(),
type_code: raw.type_code,
payload: raw.value.to_vec(),
}
}
/// The type code of a mandatory attribute that an UPDATE announcing routes
/// lacks (RFC 4271 §5, RFC 4760 §3, RFC 7606 §3.d). An UPDATE that only
/// withdraws needs no attributes.
fn missing_mandatory(ctx: &ParserContext, update: &UpdateMessage) -> Option<u8> {
let has = |type_code: u8| {
update.path_attributes.iter().any(|a| {
matches!(
(type_code, a),
(ORIGIN, PathAttribute::Origin(_))
| (AS_PATH, PathAttribute::AsPath(_))
| (NEXT_HOP, PathAttribute::NextHop(_))
| (LOCAL_PREF, PathAttribute::LocalPref(_))
)
})
};
let in_body = !update.announced_routes.is_empty();
let in_mp_reach = update
.path_attributes
@@ -422,15 +412,24 @@ fn missing_mandatory(ctx: &ParserContext, update: &UpdateMessage) -> Option<u8>
if !in_body && !in_mp_reach {
return None;
}
let mut required = vec![ORIGIN, AS_PATH];
if in_body {
// Routes in MP_REACH_NLRI carry their next hop there.
required.push(NEXT_HOP);
}
if ctx.ibgp {
required.push(LOCAL_PREF);
}
required.into_iter().find(|type_code| !has(*type_code))
// Routes in MP_REACH_NLRI carry their next hop there; LOCAL_PREF is
// for internal peers.
let required = [
(ORIGIN, true),
(AS_PATH, true),
(NEXT_HOP, in_body),
(LOCAL_PREF, ctx.ibgp),
];
required
.into_iter()
.filter(|(_, needed)| *needed)
.map(|(type_code, _)| type_code)
.find(|type_code| {
!update
.path_attributes
.iter()
.any(|attribute| attribute.type_code() == *type_code)
})
}
/// `update` with every route it announces withdrawn instead, and no other
@@ -440,21 +439,15 @@ fn as_withdrawal(update: UpdateMessage) -> UpdateMessage {
withdrawn_routes.extend(update.announced_routes);
let mut unreach: Vec<MpUnreachNlriPathAttribute> = Vec::new();
let mut withdraw = |afi, safi, prefixes: Vec<Nlri>| match unreach
.iter_mut()
.find(|u| u.afi == afi && u.safi == safi)
{
Some(existing) => existing.prefixes.extend(prefixes),
None => unreach.push(MpUnreachNlriPathAttribute {
afi,
safi,
prefixes,
}),
};
let mut withdraw =
|family, prefixes: Vec<Nlri>| match unreach.iter_mut().find(|u| u.family == family) {
Some(existing) => existing.prefixes.extend(prefixes),
None => unreach.push(MpUnreachNlriPathAttribute { family, prefixes }),
};
for attribute in update.path_attributes {
match attribute {
PathAttribute::MpUnreachNlri(mp) => withdraw(mp.afi, mp.safi, mp.prefixes),
PathAttribute::MpReachNlri(mp) => withdraw(mp.afi, mp.safi, mp.prefixes),
PathAttribute::MpUnreachNlri(mp) => withdraw(mp.family, mp.prefixes),
PathAttribute::MpReachNlri(mp) => withdraw(mp.family, mp.prefixes),
_ => {}
}
}
@@ -472,7 +465,7 @@ fn as_withdrawal(update: UpdateMessage) -> UpdateMessage {
mod tests {
use std::net::Ipv4Addr;
use bgp_primitives::constants::{Afi, Safi};
use crate::message::PathAttributeFlags;
use bgp_primitives::ip_prefix::IpPrefix;
use super::*;
@@ -492,7 +485,7 @@ mod tests {
fn ebgp() -> ParserContext {
ParserContext {
four_octet_asn: Some(true),
four_octet_asn: true,
ibgp: false,
add_path: Vec::new(),
}
@@ -500,7 +493,7 @@ mod tests {
fn ibgp() -> ParserContext {
ParserContext {
four_octet_asn: Some(true),
four_octet_asn: true,
ibgp: true,
add_path: Vec::new(),
}
@@ -619,7 +612,7 @@ mod tests {
// Nor on a session with two-octet AS numbers, where the bytes are
// not in the form a table keeps.
let two_octet = ParserContext {
four_octet_asn: Some(false),
four_octet_asn: false,
ibgp: false,
add_path: Vec::new(),
};
@@ -851,7 +844,7 @@ mod tests {
#[test]
fn as4_path_is_read_from_a_two_octet_peer_and_dropped_if_malformed() {
let ctx = ParserContext {
four_octet_asn: Some(false),
four_octet_asn: false,
ibgp: false,
add_path: Vec::new(),
};
@@ -911,8 +904,7 @@ mod tests {
assert_eq!(
decoded.update.path_attributes,
[PathAttribute::MpUnreachNlri(MpUnreachNlriPathAttribute {
afi: Afi::IPV6,
safi: Safi::UNICAST,
family: AfiSafi::IPV6_UNICAST,
prefixes: vec![v6.into()],
})],
"{case}"
@@ -925,7 +917,7 @@ mod tests {
assert!(decoded.errors.is_empty());
assert!(matches!(
&decoded.update.path_attributes[0],
PathAttribute::MpReachNlri(mp) if mp.family() == AfiSafi::IPV6_UNICAST
PathAttribute::MpReachNlri(mp) if mp.family == AfiSafi::IPV6_UNICAST
));
withdrawn(
+47 -68
View File
@@ -3,7 +3,7 @@
//! Two things are checked for every kind of message:
//!
//! - **Round trip.** A message that can be written can be read back, comes
//! back equal, and is exactly as long as `wire_len` said.
//! back equal.
//! - **Hostile input.** Whatever bytes arrive, decoding returns a message or
//! an error: it never panics. What it does accept can be shown, written
//! again, and read back as the same message.
@@ -27,9 +27,9 @@ use bgp_packet::message::{
use bgp_packet::open::{
AddPathEntry, AddPathSendRecv, BgpRole, BgpSecHeader, Capability, ExtendedNextHop,
GracefulRestartEntry, LongLivedGracefulRestart, MultiLabelEntry, OpenMessage, OpenOption,
OrfSendRecv, OrfType, OutboundRouteFilterEntry,
OrfFamily, OrfSendRecv, OrfType, OutboundRouteFilterEntry,
};
use bgp_primitives::constants::{AddressFamilyId, Afi, AfiSafi, Safi};
use bgp_primitives::constants::{Afi, AfiSafi, IpVersion, Safi};
use bgp_primitives::ip_prefix::{IpPrefix, Nlri};
use bgp_primitives::parser::{EncodeError, ParserContext};
use bytes::BytesMut;
@@ -49,7 +49,7 @@ struct Session {
impl Session {
fn ctx(self) -> ParserContext {
ParserContext {
four_octet_asn: Some(self.four_octet_asn),
four_octet_asn: self.four_octet_asn,
ibgp: self.ibgp,
add_path: [AfiSafi::IPV4_UNICAST, AfiSafi::IPV6_UNICAST]
.into_iter()
@@ -59,11 +59,11 @@ impl Session {
}
}
fn has_add_path(self, version: AddressFamilyId, safi: Safi) -> bool {
fn has_add_path(self, version: IpVersion, safi: Safi) -> bool {
safi == Safi::UNICAST
&& match version {
AddressFamilyId::Ipv4 => self.add_path[0],
AddressFamilyId::Ipv6 => self.add_path[1],
IpVersion::Ipv4 => self.add_path[0],
IpVersion::Ipv6 => self.add_path[1],
}
}
}
@@ -80,10 +80,10 @@ fn session() -> impl Strategy<Value = Session> {
// Generators
// ---------------------------------------------------------------------------
fn prefix(version: AddressFamilyId) -> impl Strategy<Value = IpPrefix> {
fn prefix(version: IpVersion) -> impl Strategy<Value = IpPrefix> {
let max_len = match version {
AddressFamilyId::Ipv4 => 32u8,
AddressFamilyId::Ipv6 => 128,
IpVersion::Ipv4 => 32u8,
IpVersion::Ipv6 => 128,
};
(0..=max_len, any::<[u8; 16]>()).prop_map(move |(length, bytes)| {
IpPrefix::new(version, bytes, length).expect("16 bytes are enough for any prefix")
@@ -92,11 +92,7 @@ fn prefix(version: AddressFamilyId) -> impl Strategy<Value = IpPrefix> {
/// Prefixes as an UPDATE names them. Without ADD-PATH for the family there
/// is no identifier on the wire, and it reads back as zero.
fn nlri(
session: Session,
version: AddressFamilyId,
safi: Safi,
) -> impl Strategy<Value = Vec<Nlri>> {
fn nlri(session: Session, version: IpVersion, safi: Safi) -> impl Strategy<Value = Vec<Nlri>> {
let with_path_id = session.has_add_path(version, safi);
vec(
(prefix(version), any::<u32>()).prop_map(move |(prefix, path_id)| Nlri {
@@ -135,14 +131,8 @@ fn as_path(asn: BoxedStrategy<u32>) -> impl Strategy<Value = AsPathAttribute> {
})
}
/// Extended communities are classified by their type octets when read, so
/// they are generated from the eight octets they are on the wire.
fn extended_community() -> impl Strategy<Value = ExtendedCommunity> {
any::<[u8; 8]>().prop_map(|bytes| {
ExtendedCommunity::from_wire(&ParserContext::default(), &bytes)
.expect("any eight octets are an extended community")
.1
})
any::<[u8; 8]>().prop_map(ExtendedCommunity)
}
fn mp_reach(session: Session) -> impl Strategy<Value = MpReachNlriPathAttribute> {
@@ -152,12 +142,11 @@ fn mp_reach(session: Session) -> impl Strategy<Value = MpReachNlriPathAttribute>
(
Just(safi),
any::<[u8; 4]>(),
nlri(session, AddressFamilyId::Ipv4, safi),
nlri(session, IpVersion::Ipv4, safi),
)
})
.prop_map(|(safi, next_hop, prefixes)| MpReachNlriPathAttribute {
afi: Afi::IPV4,
safi,
family: AfiSafi::new(Afi::IPV4, safi),
next_hop: NlriNextHop::Ipv4(Ipv4Addr::from(next_hop)),
prefixes,
});
@@ -177,12 +166,11 @@ fn mp_reach(session: Session) -> impl Strategy<Value = MpReachNlriPathAttribute>
(
Just(safi),
v6_next_hop(),
nlri(session, AddressFamilyId::Ipv6, safi),
nlri(session, IpVersion::Ipv6, safi),
)
})
.prop_map(|(safi, next_hop, prefixes)| MpReachNlriPathAttribute {
afi: Afi::IPV6,
safi,
family: AfiSafi::new(Afi::IPV6, safi),
next_hop,
prefixes,
});
@@ -190,14 +178,13 @@ fn mp_reach(session: Session) -> impl Strategy<Value = MpReachNlriPathAttribute>
}
fn mp_unreach(session: Session) -> impl Strategy<Value = MpUnreachNlriPathAttribute> {
let of = move |version: AddressFamilyId| {
let of = move |version: IpVersion| {
nlri(session, version, Safi::UNICAST).prop_map(move |prefixes| MpUnreachNlriPathAttribute {
afi: version.into(),
safi: Safi::UNICAST,
family: AfiSafi::new(version.into(), Safi::UNICAST),
prefixes,
})
};
prop_oneof![of(AddressFamilyId::Ipv4), of(AddressFamilyId::Ipv6)]
prop_oneof![of(IpVersion::Ipv4), of(IpVersion::Ipv6)]
}
/// Type codes whose definition the decoder knows; an unknown attribute has
@@ -331,9 +318,9 @@ fn path_attributes(session: Session) -> impl Strategy<Value = Vec<PathAttribute>
fn update(session: Session) -> impl Strategy<Value = UpdateMessage> {
(
nlri(session, AddressFamilyId::Ipv4, Safi::UNICAST),
nlri(session, IpVersion::Ipv4, Safi::UNICAST),
path_attributes(session),
nlri(session, AddressFamilyId::Ipv4, Safi::UNICAST),
nlri(session, IpVersion::Ipv4, Safi::UNICAST),
)
.prop_map(
|(withdrawn_routes, path_attributes, announced_routes)| UpdateMessage {
@@ -349,7 +336,7 @@ const KNOWN_CAPABILITIES: [u8; 13] = [1, 2, 3, 5, 6, 7, 8, 9, 64, 65, 69, 70, 71
fn capability() -> impl Strategy<Value = Capability> {
let orf = (
any::<u8>().prop_map(OrfType::from),
any::<u8>().prop_map(OrfType),
prop_oneof![
Just(OrfSendRecv::Receive),
Just(OrfSendRecv::Send),
@@ -365,8 +352,14 @@ fn capability() -> impl Strategy<Value = Capability> {
prop_oneof![
(afi(), safi()).prop_map(|(afi, safi)| Capability::MultiProtocol { afi, safi }),
Just(Capability::RouteRefresh),
(afi(), safi(), vec(orf, 0..4))
.prop_map(|(afi, safi, orfs)| Capability::OutboundRouteFilter { afi, safi, orfs }),
vec((afi(), safi(), vec(orf, 0..4)), 0..3).prop_map(|families| {
Capability::OutboundRouteFilter(
families
.into_iter()
.map(|(afi, safi, orfs)| OrfFamily { afi, safi, orfs })
.collect(),
)
}),
vec((afi(), safi(), afi()), 1..4).prop_map(|entries| Capability::ExtendedNextHop {
entries: entries
.into_iter()
@@ -439,11 +432,7 @@ fn capability() -> impl Strategy<Value = Capability> {
}),
vec(any::<u8>(), 0..20)
)
.prop_map(|(type_code, value)| Capability::UnknownCapability {
type_code,
length: value.len() as u8,
value,
}),
.prop_map(|(type_code, value)| Capability::UnknownCapability { type_code, value }),
]
}
@@ -466,14 +455,13 @@ fn open() -> impl Strategy<Value = OpenMessage> {
}
fn notification() -> impl Strategy<Value = NotificationMessage> {
(1u8..=8, 0u8..12, vec(any::<u8>(), 0..40)).prop_filter_map(
"an error code and subcode that are defined",
|(code, subcode, data)| {
BgpError::from_code_subcode(code, subcode)
.ok()
.map(|error| NotificationMessage { error, data })
},
)
// Any code and subcode: a peer may send ones we have no name for.
(any::<u8>(), any::<u8>(), vec(any::<u8>(), 0..40)).prop_map(|(code, subcode, data)| {
NotificationMessage {
error: BgpError::from_code_subcode(code, subcode),
data,
}
})
}
fn message(session: Session) -> impl Strategy<Value = Message> {
@@ -540,7 +528,6 @@ fn check_decoding(ctx: &ParserContext, wire: &[u8]) -> Result<(), TestCaseError>
)));
}
}
prop_assert_eq!(usize::from(msg.wire_len(ctx).unwrap()), rewritten.len());
if repairs.is_empty() {
let again = Message::decode_with_report(ctx, &rewritten)
.map_err(|e| TestCaseError::fail(format!("written but cannot be read: {e}")))?;
@@ -551,15 +538,13 @@ fn check_decoding(ctx: &ParserContext, wire: &[u8]) -> Result<(), TestCaseError>
}
/// Writes a generated message. `None` when it is too big for one BGP
/// message, which the generators can produce and the encoder must refuse;
/// any other refusal is a bug, since everything generated is meant to fit
/// the session.
fn written(ctx: &ParserContext, msg: &Message) -> Result<Option<BytesMut>, TestCaseError> {
/// message, or some part of it for its length field, which the generators
/// can produce and the encoder must refuse.
fn written(ctx: &ParserContext, msg: &Message) -> Option<BytesMut> {
let mut wire = BytesMut::new();
match msg.to_wire(ctx, &mut wire) {
Ok(()) => Ok(Some(wire)),
Err(EncodeError::TooLong { .. }) => Ok(None),
Err(e) => Err(TestCaseError::fail(format!("cannot be written: {e}"))),
Ok(()) => Some(wire),
Err(EncodeError::TooLong { .. }) => None,
}
}
@@ -638,18 +623,14 @@ proptest! {
#[test]
fn a_message_that_can_be_written_reads_back_equal((session, msg) in session_and_message()) {
let ctx = session.ctx();
let Some(wire) = written(&ctx, &msg)? else {
let Some(wire) = written(&ctx, &msg) else {
return Ok(());
};
prop_assert_eq!(usize::from(msg.wire_len(&ctx).unwrap()), wire.len());
let decoded = Message::decode_with_report(&ctx, &wire)
.map_err(|e| TestCaseError::fail(format!("{e}")))?;
// What we write is well formed: nothing for the decoder to repair.
prop_assert!(decoded.update_errors.is_empty(), "{:?}", decoded.update_errors);
prop_assert_eq!(&decoded.message, &msg);
let (rest, _) = Message::from_wire(&ctx, &wire).unwrap();
prop_assert!(rest.is_empty());
}
#[test]
@@ -672,7 +653,7 @@ proptest! {
damage in vec(damage(), 1..4),
) {
let ctx = session.ctx();
let Some(wire) = written(&ctx, &msg)? else {
let Some(wire) = written(&ctx, &msg) else {
return Ok(());
};
check_decoding(&ctx, &damaged(&wire, &damage))?;
@@ -685,11 +666,9 @@ proptest! {
bytes in vec(any::<u8>(), 0..120),
) {
let ctx = session.ctx();
let _ = OpenMessage::from_wire(&ctx, &bytes);
let _ = OpenOption::from_wire(&ctx, &bytes);
let _ = Capability::from_wire(&ctx, &bytes);
let _ = OpenMessage::from_wire(&bytes);
let _ = Capability::from_wire(&bytes);
let _ = UpdateMessage::decode(&ctx, &bytes);
let _ = PathAttribute::from_wire(&ctx, &bytes);
let _ = NotificationMessage::from_wire(&ctx, &bytes);
}
}
+20 -26
View File
@@ -1,7 +1,7 @@
use std::fmt::Display;
use serde::{Deserialize, Serialize};
use serde_repr::{Deserialize_repr, Serialize_repr};
use serde::Serialize;
use serde_repr::Serialize_repr;
use thiserror::Error;
/// An address family number this implementation has no name for.
@@ -11,14 +11,14 @@ pub struct UnknownAfi(pub u16);
/// The IP version of a prefix. For the address family field of a capability
/// or an MP_REACH_NLRI, which can name families that are not IP, see [`Afi`].
#[derive(Debug, Copy, Clone, Hash, PartialEq, Eq, Serialize_repr, Deserialize_repr)]
#[derive(Debug, Copy, Clone, Hash, PartialEq, Eq, Serialize_repr)]
#[repr(u16)]
pub enum AddressFamilyId {
pub enum IpVersion {
Ipv4 = 1,
Ipv6 = 2,
}
impl AddressFamilyId {
impl IpVersion {
/// How many bits an address of this version has.
pub fn bits(self) -> u8 {
match self {
@@ -28,7 +28,7 @@ impl AddressFamilyId {
}
}
impl TryFrom<u16> for AddressFamilyId {
impl TryFrom<u16> for IpVersion {
type Error = UnknownAfi;
fn try_from(value: u16) -> Result<Self, Self::Error> {
@@ -40,11 +40,11 @@ impl TryFrom<u16> for AddressFamilyId {
}
}
impl Display for AddressFamilyId {
impl Display for IpVersion {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
match self {
AddressFamilyId::Ipv4 => write!(f, "Ipv4"),
AddressFamilyId::Ipv6 => write!(f, "Ipv6"),
IpVersion::Ipv4 => write!(f, "Ipv4"),
IpVersion::Ipv6 => write!(f, "Ipv6"),
}
}
}
@@ -55,7 +55,7 @@ impl Display for AddressFamilyId {
/// has never heard of, and that must be readable so it can be ignored. Which
/// families are *supported* is a separate question, answered where routes
/// are stored.
#[derive(Debug, Copy, Clone, Hash, PartialEq, Eq, PartialOrd, Ord, Serialize, Deserialize)]
#[derive(Debug, Copy, Clone, Hash, PartialEq, Eq, PartialOrd, Ord, Serialize)]
#[serde(transparent)]
pub struct Afi(pub u16);
@@ -66,13 +66,13 @@ impl Afi {
pub const BGP_LS: Self = Self(16388);
}
impl From<AddressFamilyId> for Afi {
fn from(version: AddressFamilyId) -> Self {
impl From<IpVersion> for Afi {
fn from(version: IpVersion) -> Self {
Self(version as u16)
}
}
impl TryFrom<Afi> for AddressFamilyId {
impl TryFrom<Afi> for IpVersion {
type Error = UnknownAfi;
fn try_from(afi: Afi) -> Result<Self, Self::Error> {
@@ -82,7 +82,7 @@ impl TryFrom<Afi> for AddressFamilyId {
/// A Subsequent Address Family Identifier as it appears on the wire. Like
/// [`Afi`], any number is a valid value.
#[derive(Debug, Copy, Clone, Hash, PartialEq, Eq, PartialOrd, Ord, Serialize, Deserialize)]
#[derive(Debug, Copy, Clone, Hash, PartialEq, Eq, PartialOrd, Ord, Serialize)]
#[serde(transparent)]
pub struct Safi(pub u8);
@@ -99,7 +99,7 @@ impl Safi {
/// An address family: the pair that says what kind of routes a capability
/// or an MP_REACH_NLRI is about.
#[derive(Debug, Copy, Clone, Hash, PartialEq, Eq, PartialOrd, Ord, Serialize, Deserialize)]
#[derive(Debug, Copy, Clone, Hash, PartialEq, Eq, PartialOrd, Ord, Serialize)]
pub struct AfiSafi {
pub afi: Afi,
pub safi: Safi,
@@ -114,7 +114,7 @@ impl AfiSafi {
}
/// The unicast family of an IP version.
pub fn unicast(version: AddressFamilyId) -> Self {
pub fn unicast(version: IpVersion) -> Self {
Self::new(version.into(), Safi::UNICAST)
}
@@ -162,19 +162,13 @@ mod tests {
AfiSafi::new(Afi(999), Safi(7)).to_string(),
"AFI 999 SAFI 7"
);
assert_eq!(
AfiSafi::unicast(AddressFamilyId::Ipv4),
AfiSafi::IPV4_UNICAST
);
assert_eq!(AfiSafi::unicast(IpVersion::Ipv4), AfiSafi::IPV4_UNICAST);
}
#[test]
fn only_ipv4_and_ipv6_are_ip_versions() {
assert_eq!(
AddressFamilyId::try_from(Afi::IPV6),
Ok(AddressFamilyId::Ipv6)
);
assert_eq!(AddressFamilyId::try_from(Afi::L2VPN), Err(UnknownAfi(25)));
assert_eq!(AddressFamilyId::try_from(3u16), Err(UnknownAfi(3)));
assert_eq!(IpVersion::try_from(Afi::IPV6), Ok(IpVersion::Ipv6));
assert_eq!(IpVersion::try_from(Afi::L2VPN), Err(UnknownAfi(25)));
assert_eq!(IpVersion::try_from(3u16), Err(UnknownAfi(3)));
}
}
+25 -47
View File
@@ -2,12 +2,10 @@ use std::fmt::Display;
use std::net::{Ipv4Addr, Ipv6Addr};
use std::str::FromStr;
use bytes::{Buf, BufMut, BytesMut};
use serde::{Deserialize, Serialize};
use serde::Serialize;
use thiserror::Error;
use crate::constants::AddressFamilyId;
use crate::parser::{EncodeError, ParserContext};
use crate::constants::IpVersion;
/// Why something is not a valid prefix.
#[derive(Debug, Clone, PartialEq, Eq, Error)]
@@ -33,9 +31,9 @@ pub enum PrefixError {
/// value. A prefix has one representation: the bits past its length are
/// always zero, so two prefixes are equal exactly when they are the same
/// prefix.
#[derive(Debug, PartialEq, Eq, Clone, Copy, Hash, Serialize, Deserialize)]
#[derive(Debug, PartialEq, Eq, Clone, Copy, Hash, Serialize)]
pub struct IpPrefix {
address_family: AddressFamilyId,
address_family: IpVersion,
length: u8,
/// The address, left-aligned, zero past `length` bits.
address: [u8; 16],
@@ -46,7 +44,7 @@ impl IpPrefix {
/// hold at least the bytes that `length` bits cover; anything past
/// those bits is ignored (RFC 4271 §4.3: such bits are irrelevant).
pub fn new(
address_family: AddressFamilyId,
address_family: IpVersion,
prefix: impl AsRef<[u8]>,
length: u8,
) -> Result<Self, PrefixError> {
@@ -75,7 +73,7 @@ impl IpPrefix {
})
}
pub fn address_family(&self) -> AddressFamilyId {
pub fn address_family(&self) -> IpVersion {
self.address_family
}
@@ -108,20 +106,6 @@ impl IpPrefix {
let mask = u8::MAX << (8 - rem_bits);
(self.address[full_bytes] & mask) == (other.address[full_bytes] & mask)
}
pub fn to_wire(&self, _: &ParserContext, out: &mut BytesMut) -> Result<(), EncodeError> {
let bytes = self.bytes();
// Verify that there is enough space to write the IpPrefix.
if out.remaining() < (bytes.len() + 1) {
return Err(EncodeError::OutBufferOverflow);
}
// Write length and prefix.
out.put_u8(self.length);
out.put(bytes);
Ok(())
}
}
/// A prefix as an UPDATE names it: the prefix and, on a session where
@@ -130,7 +114,7 @@ impl IpPrefix {
///
/// Whether the identifier is on the wire is a matter of what the session
/// negotiated, not of its value. Where it is not, it is zero here.
#[derive(Debug, PartialEq, Eq, Clone, Copy, Hash, Serialize, Deserialize)]
#[derive(Debug, PartialEq, Eq, Clone, Copy, Hash, Serialize)]
pub struct Nlri {
pub prefix: IpPrefix,
pub path_id: u32,
@@ -181,10 +165,10 @@ impl TryFrom<&str> for IpPrefix {
let (afi, octets) = if addr.contains(':') {
let addr = Ipv6Addr::from_str(addr).map_err(|_| address())?;
(AddressFamilyId::Ipv6, addr.octets().to_vec())
(IpVersion::Ipv6, addr.octets().to_vec())
} else {
let addr = Ipv4Addr::from_str(addr).map_err(|_| address())?;
(AddressFamilyId::Ipv4, addr.octets().to_vec())
(IpVersion::Ipv4, addr.octets().to_vec())
};
IpPrefix::new(afi, octets, length)
}
@@ -193,11 +177,11 @@ impl TryFrom<&str> for IpPrefix {
impl Display for IpPrefix {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
match self.address_family {
AddressFamilyId::Ipv4 => {
IpVersion::Ipv4 => {
let [a, b, c, d, ..] = self.address;
write!(f, "{}/{}", Ipv4Addr::new(a, b, c, d), self.length)
}
AddressFamilyId::Ipv6 => {
IpVersion::Ipv6 => {
write!(f, "{}/{}", Ipv6Addr::from(self.address), self.length)
}
}
@@ -206,7 +190,7 @@ impl Display for IpPrefix {
#[cfg(test)]
mod tests {
use crate::constants::AddressFamilyId;
use crate::constants::IpVersion;
use proptest::prelude::*;
@@ -231,7 +215,7 @@ mod tests {
verify_roundtrip!(
verify_roundtrip_ipv4_24,
"10.1.2.0/24",
AddressFamilyId::Ipv4,
IpVersion::Ipv4,
vec![10, 1, 2],
24
);
@@ -240,7 +224,7 @@ mod tests {
verify_roundtrip!(
verify_roundtrip_ipv4_19,
"10.245.123.0/19",
AddressFamilyId::Ipv4,
IpVersion::Ipv4,
vec![10, 245, 96],
19
);
@@ -249,7 +233,7 @@ mod tests {
verify_roundtrip!(
verify_roundtrip_ipv4_3,
"192.168.1.0/3",
AddressFamilyId::Ipv4,
IpVersion::Ipv4,
vec![192],
3
);
@@ -258,7 +242,7 @@ mod tests {
verify_roundtrip!(
verify_roundtrip_ipv4_0,
"0.0.0.0/0",
AddressFamilyId::Ipv4,
IpVersion::Ipv4,
vec![],
0
);
@@ -266,7 +250,7 @@ mod tests {
verify_roundtrip!(
verify_roundtrip_ipv6_48,
"2001:db8:cafe::/48",
AddressFamilyId::Ipv6,
IpVersion::Ipv6,
vec![32, 1, 13, 184, 202, 254],
48
);
@@ -275,7 +259,7 @@ mod tests {
verify_roundtrip!(
verify_roundtrip_ipv6_32,
"2001:db8:cafe::/32",
AddressFamilyId::Ipv6,
IpVersion::Ipv6,
vec![32, 1, 13, 184],
32
);
@@ -307,7 +291,7 @@ mod tests {
assert!(IpPrefix::try_from("0.0.0.0/0").is_ok());
assert!(IpPrefix::try_from("2001:db8::1/128").is_ok());
assert_eq!(
IpPrefix::new(AddressFamilyId::Ipv4, vec![10], 16),
IpPrefix::new(IpVersion::Ipv4, vec![10], 16),
Err(PrefixError::TooFewBytes {
length: 16,
needed: 2,
@@ -319,11 +303,11 @@ mod tests {
#[test]
fn a_prefix_has_one_representation() {
// Extra bytes and stray bits past the length are dropped.
let noisy = IpPrefix::new(AddressFamilyId::Ipv4, vec![10, 0xff, 3, 4], 12).unwrap();
let noisy = IpPrefix::new(IpVersion::Ipv4, vec![10, 0xff, 3, 4], 12).unwrap();
assert_eq!(noisy, IpPrefix::try_from("10.240.0.0/12").unwrap());
assert_eq!(noisy.bytes(), [10, 0xf0]);
assert_eq!(
IpPrefix::new(AddressFamilyId::Ipv4, vec![0xff; 13], 99),
IpPrefix::new(IpVersion::Ipv4, vec![0xff; 13], 99),
Err(PrefixError::LengthTooLong {
length: 99,
max: 32
@@ -344,7 +328,7 @@ mod tests {
proptest! {
#[test]
fn text_round_trips(version: bool, bytes: [u8; 16], length in 0u8..=128) {
let version = if version { AddressFamilyId::Ipv6 } else { AddressFamilyId::Ipv4 };
let version = if version { IpVersion::Ipv6 } else { IpVersion::Ipv4 };
let length = length.min(version.bits());
let prefix = IpPrefix::new(version, bytes, length).unwrap();
let text = prefix.to_string();
@@ -365,8 +349,8 @@ mod tests {
fn contains_is_a_comparison_of_leading_bits(
a: [u8; 4], a_len in 0u8..=32, b: [u8; 4], b_len in 0u8..=32,
) {
let outer = IpPrefix::new(AddressFamilyId::Ipv4, a, a_len).unwrap();
let inner = IpPrefix::new(AddressFamilyId::Ipv4, b, b_len).unwrap();
let outer = IpPrefix::new(IpVersion::Ipv4, a, a_len).unwrap();
let inner = IpPrefix::new(IpVersion::Ipv4, b, b_len).unwrap();
let mask = if a_len == 0 { 0 } else { u32::MAX << (32 - a_len) };
let expected = b_len >= a_len
&& (u32::from_be_bytes(a) & mask) == (u32::from_be_bytes(b) & mask);
@@ -390,11 +374,5 @@ mod tests {
assert!(p("2001:db8::/32").contains(&p("2001:db8:cafe::/48")));
}
verify_roundtrip!(
verify_roundtrip_ipv6_0,
"::/0",
AddressFamilyId::Ipv6,
vec![],
0
);
verify_roundtrip!(verify_roundtrip_ipv6_0, "::/0", IpVersion::Ipv6, vec![], 0);
}
+4 -10
View File
@@ -3,12 +3,13 @@ use nom::error::{ErrorKind, ParseError};
use crate::constants::AfiSafi;
use thiserror::Error;
/// What a parser has to know about the session a message belongs to.
/// What reading or writing a message has to know about the session it
/// belongs to. The default is a session before its OPENs are exchanged;
/// only an UPDATE depends on anything here, and there are none until then.
#[derive(Debug, Clone, Default)]
pub struct ParserContext {
/// Whether AS numbers are four octets wide on this session (RFC 6793).
/// `None` until the OPENs have been exchanged.
pub four_octet_asn: Option<bool>,
pub four_octet_asn: bool,
/// Whether the peer is in our own AS. Some attributes mean something
/// only between internal peers and are dropped when an external one
/// sends them (RFC 7606 §7).
@@ -29,9 +30,6 @@ impl ParserContext {
/// Why a value could not be written to the wire.
#[derive(Debug, Clone, PartialEq, Eq, Error)]
pub enum EncodeError {
/// There was not enough space in the output buffer.
#[error("output buffer too small")]
OutBufferOverflow,
/// A length does not fit the field that has to carry it.
#[error("{what} is too long to encode: {len} exceeds {max}")]
TooLong {
@@ -39,10 +37,6 @@ pub enum EncodeError {
len: usize,
max: usize,
},
/// The value has no encoding in this context, e.g. a four-octet AS
/// number on a session that negotiated two-octet ones.
#[error("{0}")]
Unencodable(&'static str),
}
// Custom error type for the parser.
+3 -3
View File
@@ -10,7 +10,7 @@ use std::time::SystemTime;
use bgp_packet::message::{LargeCommunity, OriginPathAttribute};
use bgp_primitives::ip_prefix::IpPrefix;
use serde::{Deserialize, Serialize};
use serde::Serialize;
use tokio::sync::{broadcast, mpsc, oneshot};
use tokio::task::AbortHandle;
@@ -20,7 +20,7 @@ const EVENT_CHANNEL_CAPACITY: usize = 1024;
const CMD_CHANNEL_CAPACITY: usize = 256;
/// Attributes of a locally originated route.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[derive(Debug, Clone, PartialEq, Eq, Serialize)]
pub struct LocalRouteAttrs {
pub origin: OriginPathAttribute,
/// Number of *additional* copies of the local ASN to prepend to the
@@ -46,7 +46,7 @@ impl Default for LocalRouteAttrs {
}
/// A route originated by this router.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[derive(Debug, Clone, PartialEq, Eq, Serialize)]
pub struct LocalRoute {
pub prefix: IpPrefix,
pub attrs: LocalRouteAttrs,
+79 -142
View File
@@ -11,16 +11,14 @@
//! once, when the set is stored, and kept beside the bytes.
use std::hash::BuildHasher;
use std::net::{Ipv4Addr, Ipv6Addr};
use std::sync::{Arc, Mutex, PoisonError};
use std::sync::{Arc, Mutex, MutexGuard, PoisonError};
use bgp_packet::message::{NlriNextHop, OriginPathAttribute, PathAttribute};
use bgp_packet::message::{NlriNextHop, PathAttribute, attr_type, split_attribute};
use bgp_primitives::parser::{EncodeError, ParserContext};
use bytes::{BufMut, BytesMut};
use foldhash::fast::RandomState;
use hashbrown::HashTable;
const MP_REACH_NLRI: u8 = 14;
const OPTIONAL: u8 = 0x80;
/// What route selection reads from a set of attributes (RFC 4271 §9.1.2.2).
@@ -41,107 +39,62 @@ pub struct SelectionKey {
pub origin: u8,
}
impl SelectionKey {
fn of(attributes: &[PathAttribute]) -> Self {
let mut key = Self {
local_pref: None,
med: None,
neighbor_as: 0,
as_path_len: 0,
// A route without ORIGIN does not get this far; if one did, it
// would be the least preferred.
origin: 2,
};
for attribute in attributes {
match attribute {
PathAttribute::Origin(origin) => {
key.origin = match origin {
OriginPathAttribute::IGP => 0,
OriginPathAttribute::EGP => 1,
OriginPathAttribute::INCOMPLETE => 2,
}
}
PathAttribute::AsPath(path) => {
let hops: usize = path
.segments
.iter()
.map(|s| if s.ordered { s.path.len() } else { 1 })
.sum();
key.as_path_len = u16::try_from(hops).unwrap_or(u16::MAX);
key.neighbor_as = path
.segments
.first()
.filter(|s| s.ordered)
.and_then(|s| s.path.first().copied())
.unwrap_or(0);
}
PathAttribute::MultiExitDisc(med) => key.med = Some(med.0),
PathAttribute::LocalPref(local_pref) => key.local_pref = Some(local_pref.0),
_ => {}
}
}
key
}
}
/// Reads the selection key straight off attributes in the form an
/// [`AttrSet`] keeps, without building anything. `None` if the bytes are
/// not a well-formed list of attributes.
fn scan(mut wire: &[u8]) -> Option<SelectionKey> {
const EXTENDED_LENGTH: u8 = 0x10;
let mut key = SelectionKey {
local_pref: None,
med: None,
neighbor_as: 0,
as_path_len: 0,
// A route without ORIGIN does not get this far; if one did, it
// would be the least preferred.
origin: 2,
};
let number = |value: &[u8]| Some(u32::from_be_bytes(<[u8; 4]>::try_from(value).ok()?));
while !wire.is_empty() {
let (&flags, rest) = wire.split_first()?;
let (&type_code, rest) = rest.split_first()?;
let (length, rest) = if flags & EXTENDED_LENGTH != 0 {
let (length, rest) = rest.split_first_chunk::<2>()?;
(usize::from(u16::from_be_bytes(*length)), rest)
} else {
let (&length, rest) = rest.split_first()?;
(usize::from(length), rest)
};
let (value, rest) = rest.split_at_checked(length)?;
let (attribute, rest) = split_attribute(wire).ok()?;
wire = rest;
match type_code {
1 => key.origin = *value.first()?,
2 => {
let mut segments = value;
let mut hops = 0usize;
let mut first = true;
while let [kind, count, rest @ ..] = segments {
let (asns, rest) = rest.split_at_checked(usize::from(*count) * 4)?;
let ordered = *kind == 2;
hops += if ordered { usize::from(*count) } else { 1 };
if first && ordered {
key.neighbor_as = asns
.first_chunk::<4>()
.map_or(0, |a| u32::from_be_bytes(*a));
}
first = false;
segments = rest;
}
if !segments.is_empty() {
return None;
}
key.as_path_len = u16::try_from(hops).unwrap_or(u16::MAX);
}
4 => key.med = Some(number(value)?),
5 => key.local_pref = Some(number(value)?),
let value = attribute.value;
match attribute.type_code {
attr_type::ORIGIN => key.origin = *value.first()?,
attr_type::AS_PATH => (key.as_path_len, key.neighbor_as) = scan_as_path(value)?,
attr_type::MULTI_EXIT_DISC => key.med = Some(number(value)?),
attr_type::LOCAL_PREF => key.local_pref = Some(number(value)?),
_ => {}
}
}
Some(key)
}
/// The length of an AS_PATH as selection counts it, and the neighbouring
/// AS, from the attribute's value with four-octet AS numbers.
fn scan_as_path(mut segments: &[u8]) -> Option<(u16, u32)> {
const AS_SEQUENCE: u8 = 2;
let mut hops = 0usize;
let mut neighbor_as = 0;
let mut first = true;
while let [kind, count, rest @ ..] = segments {
let (asns, rest) = rest.split_at_checked(usize::from(*count) * 4)?;
let ordered = *kind == AS_SEQUENCE;
hops += if ordered { usize::from(*count) } else { 1 };
if first && ordered {
neighbor_as = asns
.first_chunk::<4>()
.map_or(0, |asn| u32::from_be_bytes(*asn));
}
first = false;
segments = rest;
}
if !segments.is_empty() {
return None;
}
Some((u16::try_from(hops).unwrap_or(u16::MAX), neighbor_as))
}
/// One distinct list of path attributes, with the next hop that goes with
/// it. Made by [`AttrStore::intern`]; two routes have the same attributes
/// it. Made by an [`AttrStore`]; two routes have the same attributes
/// exactly when they point to the same `AttrSet`.
#[derive(Debug)]
pub struct AttrSet {
@@ -165,15 +118,20 @@ impl AttrSet {
/// The attributes, parsed, and the next hop if it came in
/// MP_REACH_NLRI.
pub fn attributes(&self) -> (Vec<PathAttribute>, Option<NlriNextHop>) {
// The bytes were written by `encode`, so they parse.
parse(&self.wire).unwrap_or_default()
///
/// `None` if they do not parse. Sets made from parsed attributes or
/// from an UPDATE the decoder accepted always do; one made from bytes
/// of another source ([`AttrStore::intern_wire`]) was checked only as
/// far as selection reads it.
pub fn attributes(&self) -> Option<(Vec<PathAttribute>, Option<NlriNextHop>)> {
parse(&self.wire)
}
}
/// How the attributes of a set are written: with four-octet AS numbers.
fn four_octet() -> ParserContext {
ParserContext {
four_octet_asn: Some(true),
four_octet_asn: true,
ibgp: true,
add_path: Vec::new(),
}
@@ -197,49 +155,39 @@ fn encode(
attribute.to_wire(&ctx, out)?;
}
if let Some(next_hop) = next_hop {
out.put_u8(OPTIONAL);
out.put_u8(MP_REACH_NLRI);
out.put_u8(1 + next_hop.wire_len());
out.put_u8(next_hop.wire_len());
next_hop.to_wire(&ctx, out)?;
encode_next_hop(next_hop, out);
}
Ok(())
}
/// Writes a next hop as an MP_REACH_NLRI that holds nothing else.
fn encode_next_hop(next_hop: &NlriNextHop, out: &mut BytesMut) {
out.put_u8(OPTIONAL);
out.put_u8(attr_type::MP_REACH_NLRI);
out.put_u8(1 + next_hop.wire_len());
out.put_u8(next_hop.wire_len());
next_hop.to_wire(out);
}
/// Parses path attributes in the form [`AttrSet`] keeps and table dumps
/// use (RFC 6396 §4.3.4): four-octet AS numbers, and an MP_REACH_NLRI that
/// holds only a next hop. `None` if they are malformed.
pub fn parse(mut wire: &[u8]) -> Option<(Vec<PathAttribute>, Option<NlriNextHop>)> {
fn parse(mut wire: &[u8]) -> Option<(Vec<PathAttribute>, Option<NlriNextHop>)> {
let ctx = four_octet();
let mut attributes = Vec::new();
let mut next_hop = None;
while !wire.is_empty() {
if wire.get(1) == Some(&MP_REACH_NLRI) {
let extended = wire[0] & 0x10 != 0;
let (length, header) = if extended {
(
usize::from(u16::from_be_bytes([*wire.get(2)?, *wire.get(3)?])),
4,
)
} else {
(usize::from(*wire.get(2)?), 3)
};
let value = wire.get(header..header + length)?;
let address = |bytes: &[u8]| Some(Ipv6Addr::from(<[u8; 16]>::try_from(bytes).ok()?));
next_hop = match value {
[4, a, b, c, d] => Some(NlriNextHop::Ipv4(Ipv4Addr::new(*a, *b, *c, *d))),
[16, rest @ ..] => Some(NlriNextHop::Ipv6(address(rest)?)),
[32, rest @ ..] if rest.len() == 32 => Some(NlriNextHop::Ipv6WithLl {
global: address(&rest[..16])?,
link_local: address(&rest[16..])?,
}),
_ => return None,
};
wire = &wire[header + length..];
continue;
let (raw, rest) = split_attribute(wire).ok()?;
if raw.type_code == attr_type::MP_REACH_NLRI {
let (length, address) = raw.value.split_first()?;
if usize::from(*length) != address.len() {
return None;
}
next_hop = Some(NlriNextHop::from_bytes(address)?);
} else {
let (_, attribute) = PathAttribute::from_wire(&ctx, raw.wire).ok()?;
attributes.push(attribute);
}
let (rest, attribute) = PathAttribute::from_wire(&ctx, wire).ok()?;
attributes.push(attribute);
wire = rest;
}
Some((attributes, next_hop))
@@ -282,21 +230,9 @@ impl AttrStore {
) -> Result<Arc<AttrSet>, EncodeError> {
let mut wire = BytesMut::with_capacity(256);
encode(attributes, next_hop, &mut wire)?;
let hash = self.hasher.hash_one(&wire[..]);
let mut stripe = self.stripe(hash);
if let Some(existing) = stripe.find(hash, |set| set.wire[..] == wire[..]) {
return Ok(Arc::clone(existing));
}
let key = SelectionKey::of(attributes);
debug_assert_eq!(scan(&wire), Some(key), "both ways of reading the key agree");
let set = Arc::new(AttrSet {
hash,
key,
wire: wire[..].into(),
});
stripe.insert_unique(hash, Arc::clone(&set), |set| set.hash);
Ok(set)
Ok(self
.intern_wire(&wire)
.expect("attributes written here are well formed"))
}
/// As [`AttrStore::intern`], for attributes that are already in the
@@ -336,7 +272,7 @@ impl AttrStore {
};
let mut with_next_hop = BytesMut::with_capacity(wire.len() + 36);
with_next_hop.put_slice(wire);
encode(&[], Some(next_hop), &mut with_next_hop).ok()?;
encode_next_hop(next_hop, &mut with_next_hop);
self.intern_wire(&with_next_hop)
}
@@ -366,12 +302,11 @@ impl AttrStore {
dropped
}
fn stripe(&self, hash: u64) -> std::sync::MutexGuard<'_, HashTable<Arc<AttrSet>>> {
// The high bits: the table inside uses the low ones.
self.stripe_at((hash >> 56) as usize % STRIPES)
fn stripe(&self, hash: u64) -> MutexGuard<'_, HashTable<Arc<AttrSet>>> {
self.stripe_at(super::spread(hash, STRIPES))
}
fn stripe_at(&self, index: usize) -> std::sync::MutexGuard<'_, HashTable<Arc<AttrSet>>> {
fn stripe_at(&self, index: usize) -> MutexGuard<'_, HashTable<Arc<AttrSet>>> {
// A panic while holding the lock cannot leave the table half
// changed in a way that matters: see `IgnorePoison` in the server.
self.stripes[index]
@@ -382,9 +317,11 @@ impl AttrStore {
#[cfg(test)]
mod tests {
use std::net::Ipv4Addr;
use bgp_packet::message::{
AsPathAttribute, AsPathSegment, CommunitiesPathAttribute, LocalPrefPathAttribute,
MultiExitDiscPathAttribute, NextHopPathAttribute,
MultiExitDiscPathAttribute, NextHopPathAttribute, OriginPathAttribute,
};
use super::*;
@@ -427,7 +364,7 @@ mod tests {
};
let original = attributes(vec![65001, 4_200_000_001]);
let set = store.intern(&original, Some(&next_hop)).unwrap();
assert_eq!(set.attributes(), (original, Some(next_hop)));
assert_eq!(set.attributes(), Some((original, Some(next_hop))));
}
#[test]
@@ -485,7 +422,7 @@ mod tests {
assert_eq!(fresh.key(), parsed.key());
assert_eq!(fresh.key().as_path_len, 3);
assert_eq!(fresh.key().neighbor_as, 65001);
assert_eq!(fresh.attributes(), (attrs, Some(next_hop)));
assert_eq!(fresh.attributes(), Some((attrs, Some(next_hop))));
// Bytes that are not attributes are refused.
assert!(other.intern_wire(&[0x40, 0x01, 0x05, 0x00]).is_none());
+195 -129
View File
@@ -57,7 +57,7 @@ use std::sync::atomic::{AtomicU64, Ordering};
use bgp_primitives::ip_prefix::IpPrefix;
use foldhash::fast::RandomState;
use hashbrown::HashMap;
use hashbrown::{HashMap, HashTable};
use super::{AttrSet, PeerIndex, RibTable, RouteRef, Row, Shard, read, write};
@@ -89,10 +89,11 @@ struct Entry {
prefix: IpPrefix,
}
/// One shard's journal.
pub(super) struct Journal {
/// The sequence number of the latest change. Counts every change of a
/// best route in the shard, whether or not anyone was reading.
/// One journal of one shard.
struct Journal {
kind: Kind,
/// The sequence number of the latest change. Counts every change this
/// journal is for, whether or not anyone was reading.
seq: u64,
/// Entries up to and including this number were dropped without regard
/// to the readers. A reader whose cursor is below it missed something.
@@ -113,8 +114,9 @@ pub(super) struct Journal {
}
impl Journal {
pub(super) fn new(limits: Limits) -> Self {
fn new(kind: Kind, limits: Limits) -> Self {
Self {
kind,
limits,
seq: 0,
floor: 0,
@@ -125,10 +127,9 @@ impl Journal {
}
}
/// Notes a change to `prefix`, which has a row: of its best route, or
/// of any of its routes, whichever this journal is for. Returns the
/// sequence number for the row to remember.
pub(super) fn changed(&mut self, prefix: IpPrefix) -> u64 {
/// Notes a change to `prefix`, which has a row. Returns the sequence
/// number for the row to remember.
fn changed(&mut self, prefix: IpPrefix) -> u64 {
self.seq += 1;
if self.readers > 0 {
self.entries.push_back(Entry {
@@ -141,7 +142,7 @@ impl Journal {
/// As [`Journal::changed`], for a prefix whose row was just made:
/// if it had gone and is back, the note that it had gone is void.
pub(super) fn appeared(&mut self, prefix: IpPrefix) -> u64 {
fn appeared(&mut self, prefix: IpPrefix) -> u64 {
if !self.gone.is_empty() {
self.gone.remove(&prefix);
}
@@ -149,7 +150,7 @@ impl Journal {
}
/// Notes that `prefix` lost its last route and has no row any more.
pub(super) fn disappeared(&mut self, prefix: IpPrefix) {
fn disappeared(&mut self, prefix: IpPrefix) {
self.seq += 1;
if self.readers > 0 {
self.entries.push_back(Entry {
@@ -160,14 +161,76 @@ impl Journal {
}
}
/// Forgets that `entry`'s prefix had gone, if `entry` is what said so.
fn drop_entry(gone: &mut HashMap<IpPrefix, u64, RandomState>, entry: &Entry) {
if gone.get(&entry.prefix) == Some(&entry.seq) {
gone.remove(&entry.prefix);
}
}
pub(super) fn len(&self) -> usize {
self.entries.len()
/// What `entry` says, if it still says something: it is the latest
/// change of its prefix. The row, or `None` for a prefix that has gone.
fn live<'a>(
&self,
entry: &Entry,
rows: &'a HashTable<Row>,
hasher: &RandomState,
) -> Option<Option<&'a Row>> {
let hash = hasher.hash_one(entry.prefix);
match rows.find(hash, |row| row.prefix == entry.prefix) {
Some(row) => (row.seqs.of(self.kind) == entry.seq).then_some(Some(row)),
None => (self.gone.get(&entry.prefix) == Some(&entry.seq)).then_some(None),
}
}
/// Keeps the journal from growing. Called after every change to the
/// shard; does nothing until there is something worth doing.
fn tidy(&mut self, rows: &HashTable<Row>, hasher: &RandomState) {
if self.entries.len() <= self.limits.tidy_at {
return;
}
// What every reader has passed can go.
let passed = self
.cursors
.iter()
.map(|cursor| cursor.load(Ordering::Acquire))
.min()
.unwrap_or(NO_CURSOR);
while let Some(front) = self.entries.front() {
if front.seq > passed {
break;
}
Self::drop_entry(&mut self.gone, front);
self.entries.pop_front();
}
// Entries for prefixes that have a row are at most as many as the
// rows once the dead ones are out. Entries for prefixes that have
// gone are not counted in: they are what a reader that has stopped
// reading makes pile up.
let limit = 2 * (rows.len() + self.limits.slack);
if self.entries.len() <= limit {
return;
}
// Dead entries: their prefix has changed again since. This looks
// every entry up, but only once the journal has doubled, so the
// cost per change stays small.
let mut entries = std::mem::take(&mut self.entries);
entries.retain(|entry| self.live(entry, rows, hasher).is_some());
self.entries = entries;
if self.entries.len() <= limit {
return;
}
// Still too long: a reader is not reading while prefixes come and
// go, or more prefixes went at once than are left. The older half
// goes, and a reader that had not read it loses its place. Half,
// so that readers that are only a little behind keep theirs, and
// so that getting a long journal down to size takes few passes.
for entry in self.entries.drain(..self.entries.len() / 2) {
Self::drop_entry(&mut self.gone, &entry);
self.floor = entry.seq;
}
}
}
@@ -180,92 +243,103 @@ pub(super) enum Kind {
All,
}
impl Kind {
/// The sequence number `row` remembers for this journal.
fn seq_of(self, row: &Row) -> u64 {
match self {
Self::Best => row.changed,
Self::All => row.touched,
/// 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)]
pub(super) struct Seqs {
best: u64,
all: u64,
}
impl Seqs {
fn of(self, kind: Kind) -> u64 {
match kind {
Kind::Best => self.best,
Kind::All => self.all,
}
}
}
impl Shard {
fn journal_of(&self, kind: Kind) -> &Journal {
/// The two journals of a shard, written together: every change to a row is
/// a change of one of its routes, and some are a change of its best route.
pub(super) struct Journals {
best: Journal,
all: Journal,
}
impl Journals {
pub(super) fn new(limits: Limits) -> Self {
Self {
best: Journal::new(Kind::Best, limits),
all: Journal::new(Kind::All, limits),
}
}
fn of(&self, kind: Kind) -> &Journal {
match kind {
Kind::Best => &self.journal,
Kind::All => &self.all_journal,
Kind::Best => &self.best,
Kind::All => &self.all,
}
}
fn journal_of_mut(&mut self, kind: Kind) -> &mut Journal {
fn of_mut(&mut self, kind: Kind) -> &mut Journal {
match kind {
Kind::Best => &mut self.journal,
Kind::All => &mut self.all_journal,
Kind::Best => &mut self.best,
Kind::All => &mut self.all,
}
}
/// Whether `entry` still says something: it is the latest change of
/// its prefix.
fn is_live(&self, kind: Kind, entry: &Entry) -> bool {
let hash = self.hasher.hash_one(entry.prefix);
match self.rows.find(hash, |row| row.prefix == entry.prefix) {
Some(row) => kind.seq_of(row) == entry.seq,
None => self.journal_of(kind).gone.get(&entry.prefix) == Some(&entry.seq),
/// Notes that `prefix` has its first route, and so a row. Returns the
/// sequence numbers for the new row.
pub(super) fn appeared(&mut self, prefix: IpPrefix) -> Seqs {
Seqs {
best: self.best.appeared(prefix),
all: self.all.appeared(prefix),
}
}
/// Keeps the journals from growing. Called after every change to the
/// shard; does nothing until there is something worth doing.
pub(super) fn tidy_journals(&mut self) {
self.tidy_journal(Kind::Best);
self.tidy_journal(Kind::All);
/// Notes that a route of `row` changed, and whether its best route did.
pub(super) fn changed(&mut self, row: &mut Row, best_changed: bool) {
if best_changed {
row.seqs.best = self.best.changed(row.prefix);
}
row.seqs.all = self.all.changed(row.prefix);
}
fn tidy_journal(&mut self, kind: Kind) {
if self.journal_of(kind).entries.len() <= self.journal_of(kind).limits.tidy_at {
return;
}
// What every reader has passed can go.
let journal = self.journal_of_mut(kind);
let passed = journal
.cursors
.iter()
.map(|cursor| cursor.load(Ordering::Acquire))
.min()
.unwrap_or(NO_CURSOR);
while let Some(front) = journal.entries.front() {
if front.seq > passed {
break;
}
Journal::drop_entry(&mut journal.gone, front);
journal.entries.pop_front();
}
/// Notes that `prefix` lost its last route and has no row any more.
pub(super) fn disappeared(&mut self, prefix: IpPrefix) {
self.best.disappeared(prefix);
self.all.disappeared(prefix);
}
let limit = 2 * (self.rows.len() + self.journal_of(kind).limits.slack);
if self.journal_of(kind).entries.len() <= limit {
return;
}
// Dead entries: their prefix has changed again since. This looks
// every entry up, but only once the journal has doubled, so the
// cost per change stays small.
let mut entries = std::mem::take(&mut self.journal_of_mut(kind).entries);
entries.retain(|entry| self.is_live(kind, entry));
self.journal_of_mut(kind).entries = entries;
/// Keeps the journals from growing; see [`Journal::tidy`].
pub(super) fn tidy(&mut self, rows: &HashTable<Row>, hasher: &RandomState) {
self.best.tidy(rows, hasher);
self.all.tidy(rows, hasher);
}
if self.journal_of(kind).entries.len() <= limit {
return;
}
// Still too long: some reader is not reading, and prefixes keep
// coming and going. It loses its place.
let journal = self.journal_of_mut(kind);
for entry in journal.entries.drain(..limit / 2) {
Journal::drop_entry(&mut journal.gone, &entry);
journal.floor = entry.seq;
}
/// Where the journals are now, to ask [`Journals::has_news_since`]
/// about later.
pub(super) fn mark(&self) -> Mark {
Mark(self.best.seq, self.all.seq)
}
/// Whether anything was written since `mark` that some reader is there
/// to read.
pub(super) fn has_news_since(&self, mark: Mark) -> bool {
(self.best.readers > 0 && self.best.seq != mark.0)
|| (self.all.readers > 0 && self.all.seq != mark.1)
}
fn len(&self) -> usize {
self.best.entries.len() + self.all.entries.len()
}
}
/// A point in a shard's journals; see [`Journals::mark`].
#[derive(Clone, Copy)]
pub(super) struct Mark(u64, u64);
/// The routes of one prefix, as a follower of all routes is given them.
#[derive(Clone, Copy)]
pub struct Routes<'a>(&'a Row);
@@ -295,7 +369,8 @@ impl std::fmt::Debug for Routes<'_> {
#[derive(Debug, Clone, Copy)]
pub enum Event<'a> {
/// Forget what you know. The whole table follows, as [`Event::Best`]
/// for every prefix, and then [`Event::Synced`]. The first thing a new
/// or [`Event::Routes`] for every prefix, and then [`Event::Synced`].
/// The first thing a new
/// reader is given, and what a reader that fell too far behind is
/// given in place of what it missed.
Reset,
@@ -333,13 +408,23 @@ pub struct Progress {
}
enum Phase {
/// Giving out the whole table; `next` is the shard to do next.
/// Giving out the whole table; `next` is the shard to do next, and
/// `announced` whether [`Event::Reset`] has been given.
Resync { next: usize, announced: bool },
/// Following the journals; `next` is the shard to read next.
Follow { next: usize },
}
/// Follows the best routes of a table. Made by [`RibTable::reader`].
impl Phase {
/// The beginning of a resynchronisation.
const RESYNC: Self = Self::Resync {
next: 0,
announced: false,
};
}
/// Follows the routes of a table: the best of each prefix, or all of them.
/// Made by [`RibTable::reader`] and its siblings.
///
/// A reader holds its place in every shard's journal. Dropping it gives
/// the place up, so that the journals need not keep entries for it.
@@ -361,13 +446,7 @@ impl RibTable {
/// A new follower of this table's best routes. It is given the whole
/// table first and every change after.
pub fn reader(self: &Arc<Self>) -> Reader {
self.register(
Kind::Best,
Phase::Resync {
next: 0,
announced: false,
},
)
self.register(Kind::Best, Phase::RESYNC)
}
/// A new follower that is given only the changes from now on, for
@@ -377,7 +456,7 @@ impl RibTable {
let mut reader = self.register(Kind::Best, Phase::Follow { next: 0 });
for (index, shard) in self.shards.iter().enumerate() {
let shard = read(shard);
reader.set_cursor(index, &shard, shard.journal_of(Kind::Best).seq);
reader.set_cursor(index, &shard, shard.journals.best.seq);
}
reader
}
@@ -387,13 +466,7 @@ impl RibTable {
/// prefix's routes ([`Event::Routes`]). The whole table first, like
/// [`RibTable::reader`].
pub fn reader_of_all_routes(self: &Arc<Self>) -> Reader {
self.register(
Kind::All,
Phase::Resync {
next: 0,
announced: false,
},
)
self.register(Kind::All, Phase::RESYNC)
}
fn register(self: &Arc<Self>, kind: Kind, phase: Phase) -> Reader {
@@ -411,7 +484,7 @@ impl RibTable {
slots[slot] = true;
for shard in &self.shards {
let mut shard = write(shard);
let journal = shard.journal_of_mut(kind);
let journal = shard.journals.of_mut(kind);
while journal.cursors.len() <= slot {
journal.cursors.push(AtomicU64::new(NO_CURSOR));
}
@@ -435,10 +508,7 @@ impl RibTable {
pub fn journal_len(&self) -> usize {
self.shards
.iter()
.map(|shard| {
let shard = read(shard);
shard.journal.len() + shard.all_journal.len()
})
.map(|shard| read(shard).journals.len())
.sum()
}
@@ -455,7 +525,7 @@ impl Drop for Reader {
fn drop(&mut self) {
for shard in &self.table.shards {
let mut shard = write(shard);
let journal = shard.journal_of_mut(self.kind);
let journal = shard.journals.of_mut(self.kind);
journal.cursors[self.slot].store(NO_CURSOR, Ordering::Release);
journal.readers -= 1;
if journal.readers == 0 {
@@ -487,8 +557,12 @@ impl Reader {
/// short, so it can be more).
///
/// `visit` is called with a shard of the table held for reading. It
/// should be quick, and it must not change the table.
/// should be quick, and it must not call into the table, even to read:
/// taking the shard's lock again can wait for ever on a writer that is
/// waiting for this hold to end.
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();
let mut events = 0;
loop {
@@ -546,10 +620,7 @@ impl Reader {
shard += 1;
}
if lagged {
self.phase = Phase::Resync {
next: 0,
announced: false,
};
self.phase = Phase::RESYNC;
continue;
}
self.phase = Phase::Follow { next: 0 };
@@ -590,7 +661,7 @@ impl Reader {
fn event_for<'a>(&self, row: &'a Row) -> Event<'a> {
match self.kind {
Kind::Best => {
let best = row.route(usize::from(row.best));
let best = row.route(row.best());
Event::Best {
prefix: row.prefix,
peer: best.peer,
@@ -609,7 +680,7 @@ impl Reader {
/// resynchronisation takes a new one.
fn forget_place(&mut self) {
for (index, shard) in self.table.shards.iter().enumerate() {
read(shard).journal_of(self.kind).cursors[self.slot]
read(shard).journals.of(self.kind).cursors[self.slot]
.store(NO_CURSOR, Ordering::Release);
self.cursors[index] = NO_CURSOR;
}
@@ -617,19 +688,19 @@ impl Reader {
fn set_cursor(&mut self, index: usize, shard: &Shard, cursor: u64) {
self.cursors[index] = cursor;
shard.journal_of(self.kind).cursors[self.slot].store(cursor, Ordering::Release);
shard.journals.of(self.kind).cursors[self.slot].store(cursor, Ordering::Release);
}
/// Gives out every best route in a shard and takes a place at the end
/// of its journal, in one hold of the shard: what changes afterwards is
/// in the journal after that place.
/// Gives out every prefix in a shard and takes a place at the end of
/// its journal, in one hold of the shard: what changes afterwards is in
/// the journal after that place.
fn scan(&mut self, index: usize, visit: &mut impl FnMut(Event<'_>)) -> usize {
let table = Arc::clone(&self.table);
let shard = read(&table.shards[index]);
for row in &shard.rows {
visit(self.event_for(row));
}
self.set_cursor(index, &shard, shard.journal_of(self.kind).seq);
self.set_cursor(index, &shard, shard.journals.of(self.kind).seq);
shard.rows.len()
}
@@ -638,7 +709,7 @@ impl Reader {
fn drain(&mut self, index: usize, max: usize, visit: &mut impl FnMut(Event<'_>)) -> Drained {
let table = Arc::clone(&self.table);
let shard = read(&table.shards[index]);
let journal = shard.journal_of(self.kind);
let journal = shard.journals.of(self.kind);
let mut cursor = self.cursors[index];
if cursor < journal.floor {
return Drained::Lagged;
@@ -659,21 +730,16 @@ impl Reader {
break;
}
cursor = entry.seq;
let hash = shard.hasher.hash_one(entry.prefix);
match shard.rows.find(hash, |row| row.prefix == entry.prefix) {
Some(row) if self.kind.seq_of(row) == entry.seq => {
visit(self.event_for(row));
events += 1;
}
None if journal.gone.get(&entry.prefix) == Some(&entry.seq) => {
visit(Event::Withdrawn {
prefix: entry.prefix,
});
events += 1;
}
let event = match journal.live(entry, &shard.rows, &shard.hasher) {
Some(Some(row)) => self.event_for(row),
Some(None) => Event::Withdrawn {
prefix: entry.prefix,
},
// The prefix has changed again; a later entry has it.
_ => {}
}
None => continue,
};
visit(event);
events += 1;
}
if !more {
// Past the last entry there is nothing up to the latest change.
+86 -2
View File
@@ -2,7 +2,7 @@ use std::collections::HashMap;
use std::sync::Arc;
use std::sync::atomic::{AtomicBool, Ordering};
use bgp_primitives::constants::AddressFamilyId;
use bgp_primitives::constants::IpVersion;
use bgp_primitives::ip_prefix::IpPrefix;
use proptest::prelude::*;
@@ -146,7 +146,7 @@ impl Mirror {
}
fn nth_prefix(n: u32) -> IpPrefix {
IpPrefix::new(AddressFamilyId::Ipv4, (n << 8).to_be_bytes(), 24).unwrap()
IpPrefix::new(IpVersion::Ipv4, (n << 8).to_be_bytes(), 24).unwrap()
}
#[test]
@@ -395,6 +395,90 @@ fn a_reader_that_stops_reading_is_reset_and_holds_nothing_up() {
stuck_mirror.assert_matches(&fixture.table);
}
/// When a peer that was the only source of most prefixes goes, more
/// prefixes disappear at once than are left. The journal still comes down
/// to its bound, and a reader ends up with what the table has.
#[test]
fn many_prefixes_going_at_once_leaves_the_journal_bounded() {
let fixture = Fixture::on(RibTable::new(1), &[external(1), external(2)]);
for n in 0..5000 {
fixture.announce(0, &nth_prefix(n), &Route::via(&[1]));
}
fixture.announce(1, &nth_prefix(0), &Route::via(&[2, 3]));
let mut reader = fixture.table.reader();
let mut mirror = Mirror::default();
mirror.catch_up(&mut reader, 1000);
fixture.table.remove_peer(fixture.peers[0]);
// Some later change, as there always is, lets the journal be tidied.
for _ in 0..8 {
fixture.announce(1, &nth_prefix(1), &Route::via(&[2]));
fixture.withdraw(1, &nth_prefix(1));
}
let limits = Limits::default();
assert!(
fixture.table.journal_len() <= limits.tidy_at.max(2 * (1 + limits.slack)) + 2,
"the journal holds {} entries",
fixture.table.journal_len()
);
mirror.catch_up(&mut reader, 1000);
mirror.assert_matches(&fixture.table);
assert_eq!(mirror.best.len(), 1);
}
/// Readers are woken to read. An announcement that changes nothing, or
/// nothing a reader follows, writes nothing for them and must not wake
/// them: each would look through every shard to find that out.
#[test]
fn readers_are_not_woken_when_nothing_was_written_for_them() {
let fixture = Fixture::new(&[external(1), external(2)]);
let p = prefix("10.0.0.0/8");
let version = || fixture.table.version.load(Ordering::Acquire);
// Nobody reading: nobody to wake.
fixture.announce(0, &p, &Route::via(&[1]));
assert_eq!(version(), 0);
let _best = fixture.table.reader();
let before = version();
// The same route again.
fixture.announce(0, &p, &Route::via(&[1]));
assert_eq!(version(), before);
// A worse route from another peer: the best route stands.
fixture.announce(1, &p, &Route::via(&[2, 3, 4]));
assert_eq!(version(), before);
// A better one.
fixture.announce(1, &p, &Route::via(&[]));
assert_eq!(version(), before + 1);
// A follower of all routes is told of any change to a route.
let _all = fixture.table.reader_of_all_routes();
let before = version();
fixture.announce(0, &p, &Route::via(&[1, 1, 1]));
assert_eq!(version(), before + 1);
fixture.announce(0, &p, &Route::via(&[1, 1, 1]));
assert_eq!(version(), before + 1);
}
/// A budget of nothing would let a reader ask for ever without getting
/// anywhere; it is taken as the smallest that does.
#[test]
fn a_budget_of_nothing_still_makes_progress() {
let fixture = Fixture::new(&[external(1)]);
for n in 0..20 {
fixture.announce(0, &nth_prefix(n), &Route::via(&[1]));
}
let mut reader = fixture.table.reader();
let mut mirror = Mirror::default();
mirror.catch_up(&mut reader, 0);
for n in 0..20 {
fixture.withdraw(0, &nth_prefix(n));
}
mirror.catch_up(&mut reader, 0);
mirror.assert_matches(&fixture.table);
assert!(mirror.best.is_empty());
}
#[tokio::test]
async fn a_waiting_reader_is_woken_by_a_change_of_best_route() {
let fixture = Fixture::new(&[external(1), external(2)]);
+156 -204
View File
@@ -122,14 +122,15 @@ impl Path {
struct Row {
prefix: IpPrefix,
/// Index of the best route in `paths`. A row always has at least one
/// route, and so always a best one.
best: u16,
/// The journal's sequence number for the latest change of this row's
/// best route; see [`journal`].
changed: u64,
/// The journal's sequence number for the latest change of any of this
/// row's routes, kept only while someone follows all routes.
touched: u64,
/// route, and so always a best one; and the best one always has the
/// highest [`Path::preference`] in the row, which is what lets most
/// changes be settled without looking at the other routes
/// ([`select`]).
best: u32,
/// The sequence numbers of the latest change of this row's best route
/// and of any of its routes, by which the journals tell their entries
/// for it that still say something from those that do not.
seqs: journal::Seqs,
/// The routes, in one run of memory: finding a peer's route and
/// choosing the best are scans of it. One from each peer that has
/// announced the prefix, or several from a peer that sends more than one
@@ -143,12 +144,11 @@ struct Row {
}
impl Row {
fn new(prefix: IpPrefix, path: Path, path_id: u32, changed: u64, touched: u64) -> Self {
fn new(prefix: IpPrefix, path: Path, path_id: u32, seqs: journal::Seqs) -> Self {
Self {
prefix,
best: 0,
changed,
touched,
seqs,
paths: vec![path],
path_ids: if path_id == 0 { vec![] } else { vec![path_id] },
}
@@ -189,6 +189,16 @@ impl Row {
self.paths.swap_remove(at)
}
fn best(&self) -> usize {
self.best as usize
}
fn set_best(&mut self, index: usize) {
// An index into `paths`, which cannot have that many: each is
// sixteen bytes.
self.best = index as u32;
}
fn route(&self, index: usize) -> RouteRef<'_> {
let path = &self.paths[index];
RouteRef {
@@ -196,7 +206,7 @@ impl Row {
peer: path.peer(),
path_id: self.path_id(index),
attrs: &path.attrs,
best: index == usize::from(self.best),
best: index == self.best(),
}
}
@@ -251,11 +261,8 @@ struct Shard {
hasher: RandomState,
/// Reused between selections.
scratch: select::Scratch,
/// Which prefixes' best routes have changed, for those who follow.
journal: journal::Journal,
/// Which prefixes have had any route change, for those who follow all
/// the routes and not only the best.
all_journal: journal::Journal,
/// Which prefixes have changed, for those who follow.
journals: journal::Journals,
}
/// The routing table. Shared between threads by reference; every method
@@ -271,7 +278,8 @@ pub struct RibTable {
reader_slots: Mutex<Vec<bool>>,
/// How many readers there are, to skip waking them when there are none.
readers: AtomicUsize,
/// Goes up whenever a best route changes while there are readers.
/// Goes up whenever something is written to a journal that has
/// readers.
version: AtomicU64,
wake: Notify,
}
@@ -301,8 +309,7 @@ impl RibTable {
peers: Vec::new(),
hasher: hasher.clone(),
scratch: select::Scratch::default(),
journal: journal::Journal::new(limits),
all_journal: journal::Journal::new(limits),
journals: journal::Journals::new(limits),
})
})
.collect(),
@@ -323,13 +330,18 @@ impl RibTable {
/// Makes room for a peer's routes and returns the number to announce
/// them under.
///
/// The number is the peer's until [`RibTable::remove_peer`] is called
/// with it, and may then be given to another. Whoever was given it must
/// not use it after that: the table cannot tell a late announcement
/// from one of the number's next owner.
pub fn add_peer(&self, info: PeerInfo) -> Result<PeerIndex, TooManyPeers> {
let mut peers = self.peers.lock().unwrap_or_else(PoisonError::into_inner);
let index = match peers.iter().position(Option::is_none) {
Some(free) => free,
None => {
// A peer number shares a word with other things; see `Path`.
if peers.len() >= usize::from(u16::MAX) {
if peers.len() > usize::from(u16::MAX) {
return Err(TooManyPeers);
}
peers.push(None);
@@ -350,58 +362,47 @@ impl RibTable {
/// Takes out every route from `peer` and gives its number back for
/// reuse (RFC 4271 §8.2.2: the routes of a session that ended are
/// deleted).
///
/// This looks at every prefix in the table. The list of peers is not
/// held while it does, so other peers can come and go meanwhile.
pub fn remove_peer(&self, peer: PeerIndex) -> Outcome {
let mut peers = self.peers.lock().unwrap_or_else(PoisonError::into_inner);
let mut outcome = Outcome::default();
let mut news = false;
for shard in &self.shards {
let mut shard = write(shard);
let before = shard.journals.mark();
outcome += shard.remove_peer(peer);
shard.tidy_journals();
// From here on the shard takes no more routes from the peer.
if let Some(slot) = shard.peers.get_mut(peer.as_usize()) {
*slot = None;
}
shard.tidy_journals();
news |= shard.journals.has_news_since(before);
}
// Only now can the number be given out again: no shard knows it.
let mut peers = self.peers.lock().unwrap_or_else(PoisonError::into_inner);
if let Some(slot) = peers.get_mut(peer.as_usize()) {
*slot = None;
}
if outcome != Outcome::default() {
drop(peers);
if news {
self.wake_readers();
}
outcome
}
/// Announces `prefixes` from `peer`, all with the attributes `attrs`:
/// what one UPDATE does. A route the peer had announced before for the
/// same prefix is replaced.
pub fn announce(
&self,
peer: PeerIndex,
attrs: &Arc<AttrSet>,
prefixes: &[IpPrefix],
) -> Outcome {
self.announce_each(
peer,
attrs,
prefixes.iter().map(|prefix| Nlri::from(*prefix)),
)
}
/// As [`RibTable::announce`], for a peer that may send several routes
/// for a prefix (ADD-PATH, RFC 7911): each route is known by its prefix
/// and path identifier together, and replaces only the route the peer
/// had announced with both the same.
pub fn announce_paths(&self, peer: PeerIndex, attrs: &Arc<AttrSet>, nlri: &[Nlri]) -> Outcome {
self.announce_each(peer, attrs, nlri.iter().copied())
}
fn announce_each(
&self,
peer: PeerIndex,
attrs: &Arc<AttrSet>,
nlri: impl ExactSizeIterator<Item = Nlri>,
) -> Outcome {
/// Announces `routes` from `peer`, all with the attributes `attrs`:
/// what one UPDATE does. A route is known by its prefix and path
/// identifier together ([`Nlri`]; a plain [`IpPrefix`] is the route
/// with identifier zero, which is every route of a peer that sends one
/// for each prefix). Announcing a route the peer had announced before
/// replaces it.
pub fn announce<N>(&self, peer: PeerIndex, attrs: &Arc<AttrSet>, routes: &[N]) -> Outcome
where
N: Copy + Into<Nlri>,
{
let mut outcome = Outcome::default();
self.by_shard(nlri, |shard, placed| {
self.by_shard(routes, |shard, placed| {
let Some(info) = shard.peers.get(peer.as_usize()).copied().flatten() else {
// The peer was removed while this was on its way.
return;
@@ -411,71 +412,67 @@ impl RibTable {
outcome += shard.announce(*hash, *nlri, peer, preference, attrs);
}
});
if outcome != Outcome::default() {
self.wake_readers();
}
outcome
}
/// Withdraws `peer`'s routes for `prefixes`. Withdrawing a route that
/// is not there does nothing.
pub fn withdraw(&self, peer: PeerIndex, prefixes: &[IpPrefix]) -> Outcome {
self.withdraw_each(peer, prefixes.iter().map(|prefix| Nlri::from(*prefix)))
}
/// As [`RibTable::withdraw`], for routes known by prefix and path
/// identifier (ADD-PATH, RFC 7911).
pub fn withdraw_paths(&self, peer: PeerIndex, nlri: &[Nlri]) -> Outcome {
self.withdraw_each(peer, nlri.iter().copied())
}
fn withdraw_each(&self, peer: PeerIndex, nlri: impl ExactSizeIterator<Item = Nlri>) -> Outcome {
/// Withdraws `routes` of `peer`. Withdrawing a route that is not there
/// does nothing.
pub fn withdraw<N>(&self, peer: PeerIndex, routes: &[N]) -> Outcome
where
N: Copy + Into<Nlri>,
{
let mut outcome = Outcome::default();
self.by_shard(nlri, |shard, placed| {
self.by_shard(routes, |shard, placed| {
for (hash, nlri) in placed {
outcome += shard.withdraw(*hash, nlri, peer);
}
});
if outcome != Outcome::default() {
self.wake_readers();
}
outcome
}
/// Calls `apply` once for each shard that some of `nlri` fall in, with
/// that shard locked and those prefixes.
fn by_shard(
&self,
nlri: impl ExactSizeIterator<Item = Nlri>,
mut apply: impl FnMut(&mut Shard, &[(u64, Nlri)]),
) {
let shards = self.shards.len() as u64;
// The high bits pick the shard; the table inside uses the low ones.
let shard_of = |hash: u64| ((hash >> 32) * shards) >> 32;
let hashed = |nlri: Nlri| (self.hasher.hash_one(nlri.prefix), nlri);
// An UPDATE has a handful of prefixes; they are sorted on the stack.
const INLINE: usize = 32;
let count = nlri.len();
let mut nlri = nlri.peekable();
let Some(first) = nlri.peek().copied() else {
fn shard_of(&self, hash: u64) -> usize {
spread(hash, self.shards.len())
}
/// Calls `apply` once for each shard that some of `routes` fall in,
/// with that shard locked and those routes. Wakes the readers if a
/// journal one of them follows was written to.
fn by_shard<N>(&self, routes: &[N], mut apply: impl FnMut(&mut Shard, &[(u64, Nlri)]))
where
N: Copy + Into<Nlri>,
{
let Some(first) = routes.first() else {
return;
};
let mut inline = [(0u64, first); INLINE];
let hashed = |route: &N| {
let nlri: Nlri = (*route).into();
(self.hasher.hash_one(nlri.prefix), nlri)
};
// An UPDATE has a handful of prefixes; they are sorted on the stack.
const INLINE: usize = 32;
let mut inline = [hashed(first); INLINE];
let mut spilled = Vec::new();
let placed: &mut [(u64, Nlri)] = if count <= INLINE {
for (slot, nlri) in inline.iter_mut().zip(nlri) {
*slot = hashed(nlri);
let placed: &mut [(u64, Nlri)] = if routes.len() <= INLINE {
for (slot, route) in inline.iter_mut().zip(routes) {
*slot = hashed(route);
}
&mut inline[..count]
&mut inline[..routes.len()]
} else {
spilled.extend(nlri.map(hashed));
spilled.extend(routes.iter().map(hashed));
&mut spilled
};
placed.sort_unstable_by_key(|(hash, _)| shard_of(*hash));
for run in placed.chunk_by(|a, b| shard_of(a.0) == shard_of(b.0)) {
let mut shard = write(&self.shards[shard_of(run[0].0) as usize]);
placed.sort_unstable_by_key(|(hash, _)| self.shard_of(*hash));
let mut news = false;
for run in placed.chunk_by(|a, b| self.shard_of(a.0) == self.shard_of(b.0)) {
let mut shard = write(&self.shards[self.shard_of(run[0].0)]);
let before = shard.journals.mark();
apply(&mut shard, run);
shard.tidy_journals();
news |= shard.journals.has_news_since(before);
}
if news {
self.wake_readers();
}
}
@@ -496,7 +493,7 @@ impl RibTable {
/// The best route for `prefix`: who it is from, and its attributes.
pub fn best(&self, prefix: &IpPrefix) -> Option<(PeerIndex, Arc<AttrSet>)> {
self.with_row(prefix, |row| {
let best = &row.paths[usize::from(row.best)];
let best = &row.paths[row.best()];
(best.peer(), Arc::clone(&best.attrs))
})
}
@@ -509,8 +506,7 @@ impl RibTable {
fn with_row<T>(&self, prefix: &IpPrefix, read_row: impl FnOnce(&Row) -> T) -> Option<T> {
let hash = self.hasher.hash_one(prefix);
let shards = self.shards.len() as u64;
let shard = read(&self.shards[(((hash >> 32) * shards) >> 32) as usize]);
let shard = read(&self.shards[self.shard_of(hash)]);
shard
.rows
.find(hash, |row| row.prefix == *prefix)
@@ -520,10 +516,14 @@ impl RibTable {
/// Calls `visit` with the best route of every prefix, in no particular
/// order. The table is not stopped: one shard at a time is held for
/// reading, so the routes seen are not all of one moment.
///
/// `visit` is called with a shard held, and must not call into the
/// table itself: taking the shard's lock again can wait for ever on a
/// writer that is waiting for this hold to end.
pub fn for_each_best(&self, mut visit: impl FnMut(&IpPrefix, PeerIndex, &Arc<AttrSet>)) {
for shard in &self.shards {
for row in &read(shard).rows {
let best = &row.paths[usize::from(row.best)];
let best = &row.paths[row.best()];
visit(&row.prefix, best.peer(), &best.attrs);
}
}
@@ -548,7 +548,8 @@ impl RibTable {
}
/// Calls `visit` with every route in the table. As
/// [`RibTable::for_each_best`], without stopping the table.
/// [`RibTable::for_each_best`]: the table is not stopped, and `visit`
/// must not call into it.
pub fn for_each_route(&self, mut visit: impl FnMut(RouteRef<'_>)) {
for shard in &self.shards {
for row in &read(shard).rows {
@@ -568,6 +569,19 @@ fn write(shard: &RwLock<Shard>) -> RwLockWriteGuard<'_, Shard> {
shard.write().unwrap_or_else(PoisonError::into_inner)
}
/// Picks one of `count` buckets by a hash, for spreading keys over shards
/// or lock stripes that each hold a hash table keyed by the same hash.
///
/// That table takes a key's slot from the low bits of the hash and, from
/// the top seven, the tag it compares before it looks at the key itself
/// (hashbrown). The bucket is taken from bits in between: taken from
/// either end, the keys of one bucket would all share those bits, and the
/// table inside would lose them.
fn spread(hash: u64, count: usize) -> usize {
let middle = (hash << 7) >> 32;
((middle * count as u64) >> 32) as usize
}
/// Adds to a vector without the doubling that `push` does: a row of a few
/// hundred routes should not hold room for as many again. Grows by a
/// quarter, which keeps adding cheap on average.
@@ -579,6 +593,10 @@ fn push_tight<T>(vec: &mut Vec<T>, value: T) {
}
impl Shard {
fn tidy_journals(&mut self) {
self.journals.tidy(&self.rows, &self.hasher);
}
fn announce(
&mut self,
hash: u64,
@@ -597,9 +615,8 @@ impl Shard {
let path = Path::new(preference, peer, attrs);
let row = match entry {
Entry::Vacant(vacant) => {
let changed = self.journal.appeared(prefix);
let touched = self.all_journal.appeared(prefix);
vacant.insert(Row::new(prefix, path, path_id, changed, touched));
let seqs = self.journals.appeared(prefix);
vacant.insert(Row::new(prefix, path, path_id, seqs));
self.routes += 1;
return Outcome {
added: 1,
@@ -610,72 +627,40 @@ impl Shard {
Entry::Occupied(occupied) => occupied.into_mut(),
};
let best = usize::from(row.best);
let best_before = row.paths[best].preference();
let (outcome, best_changed) = match row.position(peer, path_id) {
None => {
row.push(path, path_id);
self.routes += 1;
let added = row.paths.len() - 1;
// Better than the best in the first steps of selection is
// better than everything; worse than it changes nothing. A
// tie has to be settled by the later steps.
let best_changed = if preference > best_before {
row.best = added as u16;
true
} else if preference < best_before {
false
} else if let Some(wins) = select::settles_tie(row, added, &self.peers) {
if wins {
row.best = added as u16;
}
wins
} else {
select::choose(row, &self.peers, &mut self.scratch)
};
let outcome = Outcome {
added: 1,
..Outcome::default()
};
(outcome, best_changed)
(
outcome,
select::after_add(row, &self.peers, &mut self.scratch),
)
}
Some(at) => {
let before = std::mem::replace(&mut row.paths[at], path);
let same = Arc::ptr_eq(&before.attrs, attrs) && before.preference() == preference;
let best_changed = if same {
// Nothing changed at all.
return Outcome {
replaced: 1,
..Outcome::default()
};
} else if at == best {
// The best route itself changed. If it got better in
// the first steps it is still the best, but it is a
// different route now.
if preference <= best_before {
select::choose(row, &self.peers, &mut self.scratch);
}
true
} else if preference > best_before {
row.best = at as u16;
true
} else if preference < best_before && before.preference() < best_before {
// It was not in the running and still is not.
false
} else {
select::choose(row, &self.peers, &mut self.scratch)
};
let outcome = Outcome {
replaced: 1,
..Outcome::default()
};
if Arc::ptr_eq(&before.attrs, attrs) && before.preference() == preference {
// The same route again: nothing changed at all.
return outcome;
}
let best_changed = select::after_replace(
row,
at,
before.preference(),
&self.peers,
&mut self.scratch,
);
(outcome, best_changed)
}
};
if best_changed {
row.changed = self.journal.changed(prefix);
}
row.touched = self.all_journal.changed(prefix);
self.journals.changed(row, best_changed);
Outcome {
best_changed: best_changed.into(),
..outcome
@@ -683,8 +668,8 @@ impl Shard {
}
fn withdraw(&mut self, hash: u64, nlri: &Nlri, peer: PeerIndex) -> Outcome {
let prefix = &nlri.prefix;
let Ok(mut entry) = self.rows.find_entry(hash, |row| row.prefix == *prefix) else {
let prefix = nlri.prefix;
let Ok(mut entry) = self.rows.find_entry(hash, |row| row.prefix == prefix) else {
return Outcome::default();
};
let row = entry.get_mut();
@@ -694,19 +679,15 @@ impl Shard {
self.routes -= 1;
if row.paths.len() == 1 {
entry.remove();
self.journal.disappeared(*prefix);
self.all_journal.disappeared(*prefix);
self.journals.disappeared(prefix);
return Outcome {
removed: 1,
best_changed: 1,
..Outcome::default()
};
}
let best_changed = Self::remove_path(row, at, &self.peers, &mut self.scratch);
if best_changed {
row.changed = self.journal.changed(*prefix);
}
row.touched = self.all_journal.changed(*prefix);
let best_changed = select::after_remove(row, at, &self.peers, &mut self.scratch);
self.journals.changed(row, best_changed);
Outcome {
removed: 1,
best_changed: best_changed.into(),
@@ -714,59 +695,30 @@ impl Shard {
}
}
/// Takes the route at `at` out of a row that has others, and returns
/// whether the best route changed.
fn remove_path(
row: &mut Row,
at: usize,
peers: &[Option<PeerInfo>],
scratch: &mut select::Scratch,
) -> bool {
let best = usize::from(row.best);
let best_before = row.paths[best].preference();
let last = row.paths.len() - 1;
let removed = row.swap_remove(at);
if at == best {
select::choose(row, peers, scratch);
return true;
}
// The last route moved into the gap.
if best == last {
row.best = at as u16;
}
// A route that tied with the best in the first steps may have been
// what kept another from being best, by its MED.
if removed.preference() == best_before {
select::choose(row, peers, scratch)
} else {
false
}
}
fn remove_peer(&mut self, peer: PeerIndex) -> Outcome {
let mut outcome = Outcome::default();
let peers = &self.peers;
let scratch = &mut self.scratch;
let journal = &mut self.journal;
let all_journal = &mut self.all_journal;
let journals = &mut self.journals;
self.rows.retain(|row| {
let mut removed = 0;
let mut best_changed = false;
// A peer that sends several routes for a prefix has several
// here.
while let Some(at) = row.position_of_peer(peer) {
outcome.removed += 1;
removed += 1;
if row.paths.len() == 1 {
outcome.removed += removed;
outcome.best_changed += 1;
journal.disappeared(row.prefix);
all_journal.disappeared(row.prefix);
journals.disappeared(row.prefix);
return false;
}
best_changed |= Self::remove_path(row, at, peers, scratch);
row.touched = all_journal.changed(row.prefix);
best_changed |= select::after_remove(row, at, peers, scratch);
}
if best_changed {
outcome.best_changed += 1;
row.changed = journal.changed(row.prefix);
if removed > 0 {
outcome.removed += removed;
outcome.best_changed += usize::from(best_changed);
journals.changed(row, best_changed);
}
true
});
+98 -12
View File
@@ -19,6 +19,12 @@
//! one changes nothing, and neither needs a look at any other route. Only a
//! tie goes on to [`choose`], which looks at the whole row, and that begins
//! with a pass over the numbers to find the routes still in the running.
//!
//! [`after_add`], [`after_replace`] and [`after_remove`] are what the table
//! calls when a row has changed. Each says whether the best route is now
//! another, and takes the short way where there is one. They rely on, and
//! keep, one fact about a row: its best route has the highest preference
//! of all its routes.
use super::attrs::SelectionKey;
use super::{DEFAULT_LOCAL_PREF, PeerInfo, Row};
@@ -47,14 +53,93 @@ pub(super) fn preference(key: &SelectionKey, peer: &PeerInfo) -> u64 {
#[derive(Default)]
pub(super) struct Scratch {
/// Routes still in the running: (neighbouring AS, MED, index in row).
candidates: Vec<(u32, u32, u16)>,
candidates: Vec<(u32, u32, u32)>,
}
/// The best route changed after a route was added at the end of `row`?
pub(super) fn after_add(row: &mut Row, peers: &[Option<PeerInfo>], scratch: &mut Scratch) -> bool {
let added = row.paths.len() - 1;
let preference = row.paths[added].preference();
let best = row.paths[row.best()].preference();
// Better than the best in the first steps of selection is better than
// everything; worse than it changes nothing. A tie has to be settled by
// the later steps.
if preference > best {
row.set_best(added);
true
} else if preference < best {
false
} else if let Some(wins) = settles_tie(row, added, peers) {
if wins {
row.set_best(added);
}
wins
} else {
choose(row, peers, scratch)
}
}
/// The best route changed after the route at `at` was replaced by another
/// from the same peer? `before` is the preference of the route it replaced.
/// The two are known to differ.
pub(super) fn after_replace(
row: &mut Row,
at: usize,
before: u64,
peers: &[Option<PeerInfo>],
scratch: &mut Scratch,
) -> bool {
let preference = row.paths[at].preference();
if at == row.best() {
// The best route itself changed. If it got better in the first
// steps it is still the best, but it is a different route now.
if preference <= before {
choose(row, peers, scratch);
}
return true;
}
let best = row.paths[row.best()].preference();
if preference > best {
row.set_best(at);
true
} else if preference < best && before < best {
// It was not in the running and still is not.
false
} else {
choose(row, peers, scratch)
}
}
/// Takes the route at `at` out of `row`, which has others, and returns
/// whether the best route changed.
pub(super) fn after_remove(
row: &mut Row,
at: usize,
peers: &[Option<PeerInfo>],
scratch: &mut Scratch,
) -> bool {
let best = row.best();
let best_preference = row.paths[best].preference();
let last = row.paths.len() - 1;
let removed = row.swap_remove(at);
if at == best {
choose(row, peers, scratch);
return true;
}
// The last route moved into the gap.
if best == last {
row.set_best(at);
}
// A route that tied with the best in the first steps may have been
// what kept another from being best, by its MED.
removed.preference() == best_preference && choose(row, peers, scratch)
}
/// Chooses the best route of `row` from all its routes, records it in the
/// row, and returns whether it is another route than before.
///
/// The row must have at least one route.
pub(super) fn choose(row: &mut Row, peers: &[Option<PeerInfo>], scratch: &mut Scratch) -> bool {
fn choose(row: &mut Row, peers: &[Option<PeerInfo>], scratch: &mut Scratch) -> bool {
let before = row.best;
let top = row
.paths
@@ -70,7 +155,7 @@ pub(super) fn choose(row: &mut Row, peers: &[Option<PeerInfo>], scratch: &mut Sc
let key = path.attrs.key();
// A route without a MED is taken to have the lowest (RFC 4271
// §9.1.2.2 c).
candidates.push((key.neighbor_as, key.med.unwrap_or(0), index as u16));
candidates.push((key.neighbor_as, key.med.unwrap_or(0), index as u32));
}
}
@@ -80,7 +165,7 @@ pub(super) fn choose(row: &mut Row, peers: &[Option<PeerInfo>], scratch: &mut Sc
// within each neighbouring AS only the routes with its lowest MED stay
// in, and the steps after are an order over what is left.
candidates.sort_unstable();
let mut best: Option<(TieBreak, u16)> = None;
let mut best: Option<(TieBreak, u32)> = None;
let mut group = None;
for &(neighbor_as, med, index) in candidates.iter() {
// Sorted by AS and then MED: the first of each AS has its lowest.
@@ -94,7 +179,7 @@ pub(super) fn choose(row: &mut Row, peers: &[Option<PeerInfo>], scratch: &mut Sc
if med != lowest {
continue;
}
let tie_break = TieBreak::of(row, index, peers);
let tie_break = TieBreak::of(row, index as usize, peers);
if best.as_ref().is_none_or(|(least, _)| tie_break < *least) {
best = Some((tie_break, index));
}
@@ -114,16 +199,16 @@ pub(super) fn choose(row: &mut Row, peers: &[Option<PeerInfo>], scratch: &mut Sc
/// best is not one of those, the best is still the best of all the others.
/// So the better of the two is the best of all. This is the usual case:
/// most routes have no MED.
pub(super) fn settles_tie(row: &Row, added: usize, peers: &[Option<PeerInfo>]) -> Option<bool> {
fn settles_tie(row: &Row, added: usize, peers: &[Option<PeerInfo>]) -> Option<bool> {
let new = row.paths[added].attrs.key();
let best = row.paths[usize::from(row.best)].attrs.key();
let best = row.paths[row.best()].attrs.key();
if new.med.unwrap_or(0) != 0 {
return None;
}
if best.neighbor_as == new.neighbor_as && best.med.unwrap_or(0) != 0 {
return None;
}
Some(TieBreak::of(row, added as u16, peers) < TieBreak::of(row, row.best, peers))
Some(TieBreak::of(row, added, peers) < TieBreak::of(row, row.best(), peers))
}
/// Steps 5 to 7, then the peer's number, and then the path identifier for
@@ -139,9 +224,9 @@ struct TieBreak {
}
impl TieBreak {
fn of(row: &Row, index: u16, peers: &[Option<PeerInfo>]) -> Self {
let peer = row.paths[usize::from(index)].peer();
let path_id = row.path_id(usize::from(index));
fn of(row: &Row, index: usize, peers: &[Option<PeerInfo>]) -> Self {
let peer = row.paths[index].peer();
let path_id = row.path_id(index);
match peers.get(peer.as_usize()).copied().flatten() {
Some(info) => Self {
internal: info.ibgp,
@@ -150,7 +235,8 @@ impl TieBreak {
peer: peer.0,
path_id,
},
// A peer on its way out; anything known is preferred to it.
// Not expected: a peer is forgotten only once its routes are
// gone. If it happened, anything known would be preferred.
None => Self {
internal: true,
router_id: u32::MAX,
+68 -3
View File
@@ -123,7 +123,7 @@ impl Fixture {
prefix: *prefix,
path_id,
};
self.table.announce_paths(self.peers[peer], &attrs, &[nlri])
self.table.announce(self.peers[peer], &attrs, &[nlri])
}
pub(super) fn withdraw_path(&self, peer: usize, prefix: &IpPrefix, path_id: u32) -> Outcome {
@@ -131,7 +131,7 @@ impl Fixture {
prefix: *prefix,
path_id,
};
self.table.withdraw_paths(self.peers[peer], &[nlri])
self.table.withdraw(self.peers[peer], &[nlri])
}
/// The best route for `prefix`: which peer (by order added) it is from
@@ -391,7 +391,7 @@ fn removing_a_peer_takes_its_routes_and_frees_its_number() {
let prefixes: Vec<IpPrefix> = (0..200u32)
.map(|i| {
IpPrefix::new(
bgp_primitives::constants::AddressFamilyId::Ipv4,
bgp_primitives::constants::IpVersion::Ipv4,
(i << 8).to_be_bytes(),
24,
)
@@ -692,3 +692,68 @@ proptest! {
}
}
}
/// Shards and lock stripes are picked by bits of the hash that the hash
/// table inside does not use, so that keys which share a shard still
/// differ where that table looks.
#[test]
fn keys_of_one_shard_differ_in_the_bits_their_table_uses() {
let hasher = RandomState::default();
let shards = 64;
let mut counts = vec![0usize; shards];
let mut tags = vec![std::collections::HashSet::new(); shards];
for n in 0..64_000u32 {
let hash = hasher.hash_one(n);
let shard = spread(hash, shards);
counts[shard] += 1;
// The seven bits hashbrown keeps as a tag.
tags[shard].insert(hash >> 57);
}
for shard in 0..shards {
assert!(
(700..1300).contains(&counts[shard]),
"shard {shard} has {} of 64000 keys",
counts[shard]
);
assert_eq!(tags[shard].len(), 128, "shard {shard}");
}
assert_eq!(spread(u64::MAX, 3), 2);
assert_eq!(spread(0, 3), 0);
}
/// A peer that sends several routes for a prefix can send more than fit a
/// sixteen-bit index. The best route was kept as one, and wrapped.
#[test]
fn a_row_can_hold_more_routes_than_a_sixteen_bit_index_counts() {
let store = AttrStore::new();
let worse = store
.intern(&Route::via(&[1, 2]).attributes(), None)
.unwrap();
let better = store.intern(&Route::via(&[1]).attributes(), None).unwrap();
let info = external(1);
let peers = [Some(info)];
let peer = PeerIndex(0);
let path =
|attrs: &Arc<AttrSet>| Path::new(select::preference(attrs.key(), &info), peer, attrs);
let seqs = journal::Journals::new(journal::Limits::default()).appeared(prefix("10.0.0.0/8"));
let mut row = Row::new(prefix("10.0.0.0/8"), path(&worse), 1, seqs);
let mut scratch = select::Scratch::default();
let count = usize::from(u16::MAX) + 10;
for path_id in 2..=count as u32 {
row.push(path(&worse), path_id);
}
// Equal routes from one peer: the lowest identifier is the best.
assert!(!select::after_add(&mut row, &peers, &mut scratch));
assert_eq!(row.best(), 0);
row.push(path(&better), count as u32 + 1);
assert!(select::after_add(&mut row, &peers, &mut scratch));
assert_eq!(row.best(), count);
assert_eq!(row.route(row.best()).path_id, count as u32 + 1);
// With the better one gone, the choice among all the rest is the
// lowest identifier again.
assert!(select::after_remove(&mut row, count, &peers, &mut scratch));
assert_eq!(row.route(row.best()).path_id, 1);
}
+12 -15
View File
@@ -14,7 +14,7 @@ use bgp_packet::message::{
Message, MpReachNlriPathAttribute, MpUnreachNlriPathAttribute, MultiExitDiscPathAttribute,
NextHopPathAttribute, NlriNextHop, PathAttribute, UpdateMessage,
};
use bgp_primitives::constants::{AddressFamilyId, Afi, Safi};
use bgp_primitives::constants::{AfiSafi, IpVersion};
use bgp_primitives::ip_prefix::{IpPrefix, Nlri};
use bgp_rib::{LocalRib, LocalRibEvent, LocalRoute, RibStopped};
use tokio::sync::broadcast::error::RecvError;
@@ -179,11 +179,11 @@ pub fn build_announce(
let mut announced_routes = vec![];
match (route.prefix.address_family(), next_hop) {
(AddressFamilyId::Ipv4, IpAddr::V4(nh)) => {
(IpVersion::Ipv4, IpAddr::V4(nh)) => {
path_attributes.push(PathAttribute::NextHop(NextHopPathAttribute(nh)));
announced_routes.push(route.prefix.into());
}
(AddressFamilyId::Ipv6, IpAddr::V6(_)) => {}
(IpVersion::Ipv6, IpAddr::V6(_)) => {}
_ => return None,
}
@@ -197,10 +197,9 @@ pub fn build_announce(
attrs.communities.clone(),
)));
}
if let (AddressFamilyId::Ipv6, IpAddr::V6(nh)) = (route.prefix.address_family(), next_hop) {
if let (IpVersion::Ipv6, IpAddr::V6(nh)) = (route.prefix.address_family(), next_hop) {
path_attributes.push(PathAttribute::MpReachNlri(MpReachNlriPathAttribute {
afi: Afi::IPV6,
safi: Safi::UNICAST,
family: AfiSafi::IPV6_UNICAST,
next_hop: NlriNextHop::Ipv6(nh),
prefixes: vec![route.prefix.into()],
}));
@@ -224,16 +223,15 @@ pub fn build_announce(
/// Builds the UPDATE withdrawing `prefix`.
pub fn build_withdraw(prefix: &IpPrefix) -> UpdateMessage {
match prefix.address_family() {
AddressFamilyId::Ipv4 => UpdateMessage {
IpVersion::Ipv4 => UpdateMessage {
withdrawn_routes: vec![Nlri::from(*prefix)],
path_attributes: vec![],
announced_routes: vec![],
},
AddressFamilyId::Ipv6 => UpdateMessage {
IpVersion::Ipv6 => UpdateMessage {
withdrawn_routes: vec![],
path_attributes: vec![PathAttribute::MpUnreachNlri(MpUnreachNlriPathAttribute {
afi: Afi::IPV6,
safi: Safi::UNICAST,
family: AfiSafi::IPV6_UNICAST,
prefixes: vec![Nlri::from(*prefix)],
})],
announced_routes: vec![],
@@ -300,15 +298,14 @@ mod tests {
// Serialise and parse back to make sure the wire encoding is sound.
let ctx = ParserContext {
four_octet_asn: Some(true),
four_octet_asn: true,
ibgp: false,
add_path: Vec::new(),
};
let msg = Message::Update(update.clone());
let mut buf = BytesMut::new();
msg.to_wire(&ctx, &mut buf).unwrap();
let (_, parsed) = Message::from_wire(&ctx, &buf).unwrap();
let Message::Update(parsed) = parsed else {
let Message::Update(parsed) = Message::decode(&ctx, &buf).unwrap() else {
panic!("expected UPDATE")
};
assert_eq!(parsed, update);
@@ -359,13 +356,13 @@ mod tests {
// What the peer reads from the wire.
let ctx = ParserContext {
four_octet_asn: Some(false),
four_octet_asn: false,
ibgp: false,
add_path: Vec::new(),
};
let mut buf = BytesMut::new();
Message::Update(update).to_wire(&ctx, &mut buf).unwrap();
let (_, Message::Update(seen)) = Message::from_wire(&ctx, &buf).unwrap() else {
let Message::Update(seen) = Message::decode(&ctx, &buf).unwrap() else {
panic!("expected UPDATE")
};
assert_eq!(path(&seen, false), Some(vec![23456; 2]), "AS_TRANS");
+1 -1
View File
@@ -355,7 +355,7 @@ impl<R: AsyncRead + Unpin> Session<R> {
}
Action::Admit(negotiated) => {
self.ctx = ParserContext {
four_octet_asn: Some(negotiated.four_octet_asn),
four_octet_asn: negotiated.four_octet_asn,
ibgp: negotiated.peer_asn == self.config.local_asn,
add_path: negotiated.add_path_receive.iter().collect(),
};
+5 -12
View File
@@ -15,7 +15,6 @@ use bgp_packet::message::{PathAttribute, UpdateMessage};
use bgp_packet::open::{Capability, OpenMessage};
use bgp_primitives::constants::AfiSafi;
use bgp_primitives::ip_prefix::IpPrefix;
use bgp_primitives::parser::ParserContext;
use bytes::BytesMut;
/// The address families this speaker can exchange routes for. Supporting a
@@ -82,15 +81,11 @@ impl FamilySet {
/// These families as Multiprotocol capabilities in wire form: the Data
/// of an Unsupported Capability NOTIFICATION (RFC 5492 §5).
pub fn capabilities_wire(&self) -> Vec<u8> {
let ctx = ParserContext::default();
let mut out = BytesMut::new();
for family in self.iter() {
// A Multiprotocol capability always encodes; if that ever
// changed, a NOTIFICATION with less Data is still correct.
if Capability::multiprotocol(family)
.to_wire(&ctx, &mut out)
.is_err()
{
if Capability::multiprotocol(family).to_wire(&mut out).is_err() {
return vec![];
}
}
@@ -112,8 +107,8 @@ impl FamilySet {
}
update.path_attributes.retain(|attribute| {
let (family, prefixes) = match attribute {
PathAttribute::MpReachNlri(mp) => (mp.family(), mp.prefixes.len()),
PathAttribute::MpUnreachNlri(mp) => (mp.family(), mp.prefixes.len()),
PathAttribute::MpReachNlri(mp) => (mp.family, mp.prefixes.len()),
PathAttribute::MpUnreachNlri(mp) => (mp.family, mp.prefixes.len()),
_ => return true,
};
let keep = self.contains(family);
@@ -257,8 +252,7 @@ mod tests {
path_attributes: vec![
PathAttribute::Origin(OriginPathAttribute::IGP),
PathAttribute::MpReachNlri(MpReachNlriPathAttribute {
afi: Afi::IPV6,
safi: Safi::UNICAST,
family: AfiSafi::IPV6_UNICAST,
next_hop: NlriNextHop::Ipv6(Ipv6Addr::LOCALHOST),
prefixes: vec![
prefix("2001:db8:1::/48").into(),
@@ -266,8 +260,7 @@ mod tests {
],
}),
PathAttribute::MpUnreachNlri(MpUnreachNlriPathAttribute {
afi: Afi::IPV6,
safi: Safi::UNICAST,
family: AfiSafi::IPV6_UNICAST,
prefixes: vec![prefix("2001:db8:3::/48").into()],
}),
],
+3 -4
View File
@@ -425,7 +425,7 @@ mod tests {
MpReachNlriPathAttribute, NlriNextHop, NotificationMessage, PathAttribute,
RouteRefreshMessage,
};
use bgp_primitives::constants::{Afi, AfiSafi, Safi};
use bgp_primitives::constants::AfiSafi;
use bgp_primitives::ip_prefix::IpPrefix;
use super::*;
@@ -664,7 +664,7 @@ mod tests {
/// offered to send and the other to receive.
#[test]
fn add_path_is_in_use_where_one_side_sends_and_the_other_receives() {
use AddPathSendRecv::{Both, Receive, Send};
use bgp_packet::open::SendReceive::{Both, Receive, Send};
let negotiated = |ours: Option<AddPathSendRecv>,
theirs: Vec<(AfiSafi, AddPathSendRecv)>| {
let (mut fsm, first) = Fsm::new(FsmConfig {
@@ -747,8 +747,7 @@ mod tests {
let update = UpdateMessage {
withdrawn_routes: vec![],
path_attributes: vec![PathAttribute::MpReachNlri(MpReachNlriPathAttribute {
afi: Afi::IPV6,
safi: Safi::UNICAST,
family: AfiSafi::IPV6_UNICAST,
next_hop: NlriNextHop::Ipv6(std::net::Ipv6Addr::LOCALHOST),
prefixes: vec![IpPrefix::try_from("2001:db8::/32").unwrap().into()],
})],
+3 -3
View File
@@ -29,11 +29,11 @@ pub(crate) fn apply(
let mut outcome = Outcome::default();
if !update.withdrawn_routes.is_empty() {
outcome += table.withdraw_paths(peer, &update.withdrawn_routes);
outcome += table.withdraw(peer, &update.withdrawn_routes);
}
for attribute in &update.path_attributes {
if let PathAttribute::MpUnreachNlri(mp) = attribute {
outcome += table.withdraw_paths(peer, &mp.prefixes);
outcome += table.withdraw(peer, &mp.prefixes);
}
}
@@ -47,7 +47,7 @@ pub(crate) fn apply(
};
let mut announce = |nlri: &[Nlri], next_hop| -> Result<(), EncodeError> {
if !nlri.is_empty() {
outcome += table.announce_paths(peer, &attributes(next_hop)?, nlri);
outcome += table.announce(peer, &attributes(next_hop)?, nlri);
}
Ok(())
};
+27 -10
View File
@@ -273,7 +273,8 @@ impl PeerNames {
/// the route is on its way out of the table.
pub(crate) fn view(&self, route: &Route) -> Option<ReceivedRouteView> {
let peer = self.0.get(&route.peer)?;
let (attributes, next_hop) = route.attrs.attributes();
// What a session stored was accepted by the decoder, and parses.
let (attributes, next_hop) = route.attrs.attributes().unwrap_or_default();
Some(ReceivedRouteView {
peer_addr: peer.addr.to_string(),
peer_asn: peer.asn,
@@ -365,6 +366,10 @@ pub async fn local_routes(
/// number. The table can hold hundreds of millions.
const DEFAULT_ROUTE_LIMIT: usize = 1000;
/// The most routes [`received_routes`] returns, whatever is asked for:
/// each is copied out of the table and written as JSON.
const MAX_ROUTE_LIMIT: usize = 100_000;
#[derive(Debug, Deserialize)]
pub struct RouteQuery {
/// At most this many routes.
@@ -376,14 +381,26 @@ pub struct RouteQuery {
pub async fn received_routes(
State(state): State<AppState>,
Query(query): Query<RouteQuery>,
) -> Json<Vec<ReceivedRouteView>> {
) -> Result<Json<Vec<ReceivedRouteView>>, StatusCode> {
let limit = query
.limit
.unwrap_or(DEFAULT_ROUTE_LIMIT)
.min(MAX_ROUTE_LIMIT);
let names = PeerNames::of(&state.registry);
let mut routes: Vec<ReceivedRouteView> = state
.rib
.some_routes(query.limit.unwrap_or(DEFAULT_ROUTE_LIMIT))
.iter()
.filter_map(|route| names.view(route))
.collect();
routes.sort_by(|a, b| (&a.prefix, &a.peer_addr).cmp(&(&b.prefix, &b.peer_addr)));
Json(routes)
// Copying, parsing and sorting many routes takes long enough to keep
// off the threads that sessions run on.
let routes = tokio::task::spawn_blocking(move || {
let mut routes: Vec<ReceivedRouteView> = state
.rib
.some_routes(limit)
.iter()
.filter_map(|route| names.view(route))
.collect();
routes.sort_by(|a, b| (&a.prefix, &a.peer_addr).cmp(&(&b.prefix, &b.peer_addr)));
routes
})
.await
// Only if making the views panicked.
.map_err(|_| StatusCode::INTERNAL_SERVER_ERROR)?;
Ok(Json(routes))
}
+1 -5
View File
@@ -29,7 +29,7 @@ impl Session {
fn context(self) -> ParserContext {
ParserContext {
four_octet_asn: Some(self.four_octet_asn),
four_octet_asn: self.four_octet_asn,
ibgp: self.ibgp,
add_path: [AfiSafi::IPV4_UNICAST, AfiSafi::IPV6_UNICAST]
.into_iter()
@@ -62,10 +62,6 @@ pub fn decode_and_check(session: Session, wire: &[u8]) {
Err(EncodeError::TooLong { .. }) if !repairs.is_empty() => return,
Err(e) => panic!("an accepted message cannot be written: {e}"),
}
assert_eq!(
usize::from(msg.wire_len(&ctx).expect("and has a length")),
rewritten.len()
);
// An UPDATE the decoder had to repair (RFC 7606) is not what was on the
// wire, and need not read back as itself.
if repairs.is_empty() {