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

Compare commits

base: rayhaan/ferrix:b0dd3579868d602c5f36c9e5de9c53d96b75d488
rayhaan/ferrix:main rayhaan/ferrix:speaker-cleanup rayhaan/ferrix:netsec_2026
rayhaan/ferrix:v0.2.0
...
compare: rayhaan/ferrix:c70a223d0d74fa13f32b3637bdabd8b6b4e1425c
rayhaan/ferrix:main rayhaan/ferrix:speaker-cleanup rayhaan/ferrix:netsec_2026
rayhaan/ferrix:v0.2.0
12 Commits
b0dd357986 ... c70a223d0d
Author SHA1 Message Date
Rayhaan JaufeerallyandClaude Fable 5.1 c70a223d0d rib: a reader counts the table once a poll, not once a shard
ci / check (push) Canceled after 0s
Details
ci / docs (push) Canceled after 0s
Details
ci / interop (push) Canceled after 0s
Details
ci / fuzz (push) Canceled after 0s
Details 
A router with many peers has a reader of the table's journal for each
session that is sent learned routes, seventy on ddln, and every change
to the table wakes them all. Each pass of a reader went round every
shard, and for each shard took a count on the table (an Arc clone) to
hold it while the shard was read, and gave it back after: two atomic
adds on one cache line, by seventy readers, 256 shards a pass. A
profile of ddln taking a table from the other router had the reader's
poll as the top function at two to three cores, and in the Linux VM
with 64 such sessions half of all CPU time was in the two atomic add
instructions. The sessions writing to the table waited behind it: the
table came in at 58,000 routes a second.

The count is now taken once a poll. And each journal keeps its latest
sequence number in an atomic the table holds the other end of, so that
a reader can tell a shard with nothing new for it from one with
something without the shard's lock, and pass over it: on a quiet
table, which is what the readers see most of the time, a pass is a
load a shard and no lock. An exporter that is sent little of a table,
as one to a peer at an exchange is, no longer makes a verdict of
nothing for a prefix the peer does not hold, which cost a map insert
and a lookup on the way out, and tells a route that came down or
across (RFC 9234) from its attributes' bytes before parsing them.

Testing: the mesh benchmark can have any number of collector sessions
of the first router (BENCH_COLLECTORS): peers in the sense of RFC 9234,
sent learned routes, with the feeders' routes marked as having come
across from a peer, so that each has a reader following the table
while it fills and passes nearly all of it over, as a router with many
peers at exchanges has. In the Linux VM with 64 of them, three rounds
alternating with the previous commit, the routers' exchange of their
tables takes 11 s instead of 20 s, at 86 s of CPU time instead of
155 s; without them it is unchanged. The table's tests, the server
tests, the integration tests, the kernel tests and every lab phase
pass.

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
 2026-10-10 17:12:37 +02:00
Rayhaan JaufeerallyandClaude Fable 5.1 a15efaa26c export: give the leave to gather back once the table is gathered 
The last change held a session's leave to gather a table, of which
there are four, until its table had gone out. A peer that reads at the
pace of its receive window takes hours over a table, and two such
peers held two of the four for that long, keeping other sessions from
gathering theirs: they waited for a leave before their first UPDATE.
The leave is for the memory a gather takes; what is still held once
the table is gathered is let go of as it goes out, as it was before
the gather was reworked.

The mesh benchmark's new peer can also come once the first router has
its own peers' routes and the other router's are still arriving
(BENCH_RECEIVER_EARLY=2), as a router's sessions come up after a
restart while its peers' tables arrive; the walk of a full table while
the table fills costs the arriving routes nothing measurable in the
VM.

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
 2026-10-10 15:43:48 +02:00
Rayhaan JaufeerallyandClaude Fable 5.1 624d20f965 Release 0.2.6 
Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
 2026-10-10 15:30:07 +02:00
Rayhaan JaufeerallyandClaude Fable 5.1 b5ca42244d export: a lighter gather, and UPDATEs to the session by the piece 
Two things kept an exporter from its peer. A table being gathered to
go out anew kept a map of every prefix heard, to let a later word on a
prefix supersede an earlier one, and dressed every set of attributes
before the first UPDATE went: a map insert and a parse-and-encode a
route that the peer waited on. And the session's queue took UPDATEs
one by one, a thousand at most, so the exporter stood at the queue
once a hundred kilobytes were in it, behind a writer that takes a
megabyte: the two took turns instead of running together.

The journal gives a table out shard by shard, each prefix once, and a
change meanwhile comes after the table as news of its own, so there is
no second word to keep track of while gathering: the map is gone, and
a prefix the peer is not to have is simply not gathered, as what it
held of it is withdrawn once the table is through. Each set is dressed
as its turn to go comes, so the first UPDATE goes once the table is
gathered; only when the gathered routes alone would be over the
peer's limit are the sets dressed first, to know which cannot be and
take their routes with them, so that none goes if too many would. The
UPDATEs for a set now go to the session in pieces of 64 KiB or so,
each of whole messages, and the queue takes 64 of those: a few
megabytes ahead of the writer, not a hundred kilobytes. The session
counts the messages in a piece by their headers.

Testing: the byte-for-byte test of the encoded UPDATEs now encodes
thirty thousand routes and checks the pieces are of whole messages
and of the size meant. The server tests, the integration tests, the
kernel tests and every lab phase pass. In the Linux VM, three rounds
alternating with the previous commit, a new peer's first UPDATE comes
1.8 s after its session is up instead of 3.1 s, and it has its four
million routes in 3.8 s instead of 4.1 s; the smaller of the two
routers' tables crosses in 3.0 s instead of 4.2 s and the larger in
5.0 s instead of 5.2 s, at the same CPU time.

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
 2026-10-10 15:28:40 +02:00
Rayhaan JaufeerallyandClaude Fable 5.1 7425b101ed export: gather a table by attributes, then stream it 
A table given out anew to a session, at its start or when the journal
had let changes go, was gathered as a verdict for each prefix, each
with the route dressed for the session, and sent once the table was
all there. Dressing a route parses its attributes and encodes them for
the session, so a table of five million routes took seconds before
the first UPDATE went, on one worker thread without a yield. And the
routes that share attributes went out in separate UPDATEs after all:
they were grouped by the dressed form, of which only so many are kept,
and the routes of a set are spread over the whole table, so by the
time the next was reached the form had been let go of and made anew.
The mesh benchmark's new peer got four million routes in 3.8 million
UPDATEs; on a RIPE RIS table, grouped whole, a full table is 160,000
UPDATEs and 21 MB, and grouped in slices of 4,096 routes it is 845,000
UPDATEs and 72 MB, so slices are not the answer either.

The table is now gathered by what dresses its routes: the attribute
set as the table has it, who the route is from, and its IP version.
Nothing is dressed while gathering, which is a map insert a route.
Once the table is all there, each set is dressed once, with the thread
lent to the other tasks every few thousand, and then the sets go out
one after the other, each as few UPDATEs as its routes fit in, with
the peer reading the first while the rest are sent. A prefix heard of
again while the table is read supersedes what was heard before: each
word on a prefix is numbered, and only the latest goes. Path
identifiers are given at gathering, in the order the table has a
prefix's routes, so that they are as before. What was announced before
and not heard of again is withdrawn after the table, as before. The
exporter never blocks the thread on a table: the steady state's batch
is small, and the gathered table is gone through with yields.

Testing: a new test gives a session a table of 300 routes sharing
three sets of attributes, spread over the shards, and checks they go
in three UPDATEs. The mesh benchmark has a new phase in which a peer
takes the first router's table once the mesh is complete, timing the
first UPDATE and the whole; its feeders now negotiate no hold timer,
since they say nothing after their tables, and `BENCH_LOG` prints the
routers' logs. A new example, packing, counts the UPDATEs and octets
a RIS table takes grouped whole and in slices. The server tests, the
integration tests, the kernel tests and every lab phase pass. In the
Linux VM, three rounds alternating with the previous commit, the new
peer's first UPDATE comes 2.9 s after its session is up instead of
5.3 s, and it has all four million routes in 4.0 s instead of 8.0 s,
in 667,000 UPDATEs instead of 3.76 million; the whole run takes
22.4 s of CPU time instead of 26.2 s. The routers' exchange of their
tables, which does not go through this path once their sessions are
up, is unchanged within its noise.

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
 2026-10-10 15:16:18 +02:00
Rayhaan JaufeerallyandClaude Fable 5.1 92b9bdd3c5 export: keep the peer reading while the next batch is dressed 
An exporter following the table took up to 65,536 changes at a time,
dressed them all, and only then sent them; and before each batch it
slept the 50 ms that lets a run of changes be gathered, whether or not
the table was already far ahead of it. The session's queue holds about
a thousand UPDATEs, so once the batch went out the exporter waited on
the peer reading it, and while the next batch was dressed the peer had
nothing to read. A profile of two routers exchanging their tables,
sampled in 25 ms windows, showed the two stages taking turns: the
exporters at two cores while the sessions were at none, then the other
way round, with a gap of 50 to 75 ms at the start of each cycle for the
sleep. Each task ran at about half a core.

A batch is now 4,096 changes, so that what one batch sends is being
read by the peer while the next is being dressed, and the gathering
sleep is skipped while the reader is behind the table: a run is in hand
already. A reader says whether its last poll caught up.

Testing: the server tests, the integration tests, the kernel tests and
every lab phase pass. On the mesh benchmark in the Linux VM, three
rounds each alternating with the previous commit, the larger table
(4.9 million routes) is across in 5.6 s instead of 6.7 s and the
smaller (2.5 million) in 4.4 s instead of 6.4 s; the whole exchange
takes 7.3 s of wall-clock time instead of 8.6 s, and 18.9 s of CPU time
instead of 20.5 s.

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
 2026-10-10 14:19:02 +02:00
Rayhaan JaufeerallyandClaude Fable 5.1 381eabd886 export: encode a route's attributes once a session 
The exporter built each UPDATE as a message: the attributes of the
route's set, parsed from the table, were cloned, had the next hop
added, were sorted, and were handed to the session, which encoded them
to the wire with the rest of the message. For a table whose attribute
sets each cover a few routes that is an encoding of the same octets for
nearly every route, and a copy of the message's bytes on the way to the
writer besides. A profile of a production router taking the other
router's table with ADD-PATH had the exporters at 55% of the CPU time,
their dressing of routes at 23%, and memcpy at 12%.

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

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

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
 2026-10-10 13:47:47 +02:00
Rayhaan JaufeerallyandClaude Fable 5.1 f1aca37ebe rib: a short row grows by doubling 
A prefix's row of paths is given room for as many paths as there are
peers, and from there grows by a quarter at a time, one path at a time
while it is short. A peer sending every path it has (ADD-PATH) gives a
prefix several paths from one peer, so a row goes past the peers' count
from the first few routes and was then reallocated for each path.

A row shorter than the first allocation's room doubles instead, and
grows by a quarter only from there. Measured on the mesh benchmark, the
reallocations in the table's insert path fall by half; the memory held
is as before, as the rows that reach the quarter-growth are the ones
that were already allocated for it.

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
 2026-10-10 13:45:15 +02:00
Rayhaan JaufeerallyandClaude Fable 5.1 cbc1dc7e44 bench: two routers exchanging their tables with ADD-PATH 
The full-table benchmark measures one router taking in tables from its
peers. What two routers of one AS do with each other, each sending the
other every path it has (ADD-PATH both ways with all_paths, as a mesh
of routers that each hold the network's whole view is configured), was
measured nowhere: the exporter following the table's journal while the
sessions fill the table, at the size of the Internet. In production
that exchange took 70 seconds for a table that arrives from the
Internet in 20.

A new benchmark, mesh, starts two servers in one AS with distinct
router IDs, peers them over IPv6 loopback with ADD-PATH both ways and
all_paths, and feeds the first from four synthetic peers and the second
from two over IPv4 loopback (BENCH_PEERS, BENCH_PEERS_B,
BENCH_SCALE_DOWN). It prints how long each router takes to hold its
own peers' routes and then the other's, with the rate of the latter.
BENCH_DEADLINE prints the sessions' state if a router is not there in
time.

The benchmarks now allocate with jemalloc, as the daemon does, except
when memory is being counted: the C library's allocator put its own
locks in the profiles. Two routers on one host have each other's
addresses as next hops, loopback addresses, which the import check
refuses as it refuses a next hop that is this machine; the permissive
import policy, which only tests use and no configuration can ask for,
now takes them.

Testing: the next-hop test covers both policies; the benchmark runs on
macOS and in the Linux VM under perf.

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
 2026-10-10 13:45:15 +02:00
Rayhaan JaufeerallyandClaude Fable 5.1 d0f092ff06 session: write to the peer from a task of its own 
A session read from and wrote to its socket in one loop: while it was
writing, it was not reading. Writing waits when the peer's receive
window is closed, and the peer's window closes when the peer is not
reading, which it is not while it is writing to us and waiting on our
window in turn. Two routers sending each other every path at once, as
they do with ADD-PATH both ways after a restart, each stopped reading
within seconds and waited on the other until the send hold timer ended
both sessions eight minutes later. The status page's TCP row showed it:
both windows zero, the send queue full, the receive queue unread. The
batching of writes in the previous change made a session spend far
longer writing and so made this certain; it was possible before.

Now each session has a writer task. The session encodes its messages
into batches and hands them to the task through a bounded queue without
ever waiting: a batch the queue has no room for is kept, and while any
is kept the session takes no more messages from the rest of the server
(which then waits, as before, in its own task) but goes on reading the
peer. The writer task's failure, or a write that makes no progress for
the send hold time, comes back to the session as a local failure and
ends it. When the session ends, what is still to go out, a NOTIFICATION
among it, is handed over and the writer shuts our side once it has
written it.

Testing: a new integration test has one peer's table in the server and
a second peer send a table of its own while reading nothing of what the
server sends it, and checks that the server takes the second table all
the same; against the previous commit the test hangs until its timeout.
The send hold timer test now reads the writer's failure. The session
tests, the integration tests, the kernel tests and every lab phase
pass.

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
 2026-10-10 12:22:06 +02:00
Rayhaan JaufeerallyandClaude Fable 5.1 ce1b68ca59 bgp: jemalloc's background thread on Linux only 
The last change built jemalloc with its background thread on, so that
freed pages go back to the kernel on time. jemalloc has that thread on
Linux only: on macOS every start of the daemon, and every test that
starts one, printed "option background_thread currently supports
pthread only".

The feature now comes from a Linux-only dependency entry of the daemon
crate, so the macOS build asks for nothing jemalloc cannot do. The
Linux build is as before: measured in the VM with four peers and a
collector, resident memory falls after the load without any setting
in the environment.

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
 2026-10-10 12:06:39 +02:00
Rayhaan JaufeerallyandClaude Fable 5.1 9276bc9f7a session: write messages in batches, with TCP_NODELAY 
A session wrote every message to its socket as it came: one write and,
with Nagle's algorithm, mostly one segment for each UPDATE, a hundred or
two octets. On the IPv6 cross-connect between the two routers, whose
round trip is 50 microseconds, a million routes took 42 seconds to go
out: the status page's new TCP row showed the connection busy for that
long with neither window nor buffer limiting it, a congestion window
stuck at ten segments, and a smoothed round-trip time of 18 ms. That is
the receiver's delayed acknowledgements, 40 ms at a time, on a stream
of small segments, holding the sender to a few segments a round trip.

Now a session gathers what is waiting for it, up to the outbound queue's
length, in one wake, encodes the messages into one buffer, and writes
the buffer when it reaches 64 KiB or when the actions of the wake are
done; the socket has TCP_NODELAY, so a batch's last, partial segment is
not held back for the acknowledgement of the one before. The
NOTIFICATION that ends a session, including one sent after a panic, is
flushed on its own.

Measured in the Linux VM with four peers' tables going to a collector
over a veth pair, 151 MiB: 121,850 segments instead of 313,773, the
connection busy 0.8 s instead of 2.0 s, and the congestion window grown
to 47 segments instead of 13. Over loopback the transit benchmark is
unchanged at 2.5 s for the second neighbor to have the table: there the
receiver's parsing sets the pace, not the segments.

Testing: the session tests, including the one that sends a NOTIFICATION
after a panic; the integration tests; the kernel tests and every lab
phase.

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
 2026-10-10 11:14:08 +02:00
22 changed files with 1912 additions and 219 deletions
Expand all files Collapse all files

No files matched your search

Cargo.lock
Generated
+13 -10
View File
@@ -157,7 +157,7 @@ checksum = "ac07cdecf99051d9a5238b80f35af32cdeba5b336e55d957b318b50137e18da5"
[[package]]
name = "bgp"
version = "0.2.5"
version = "0.2.6"
dependencies = [
"bgp-server",
"bgp-web",
@@ -172,7 +172,7 @@ dependencies = [
[[package]]
name = "bgp-bench"
version = "0.2.5"
version = "0.2.6"
dependencies = [
"bgp-integration-tests",
"bgp-packet",
@@ -182,12 +182,14 @@ dependencies = [
"bytes",
"flate2",
"stats_alloc",
"tikv-jemallocator",
"tokio",
"tracing-subscriber",
]
[[package]]
name = "bgp-fib"
version = "0.2.5"
version = "0.2.6"
dependencies = [
"bgp-packet",
"bgp-primitives",
@@ -205,7 +207,7 @@ dependencies = [
[[package]]
name = "bgp-integration-tests"
version = "0.2.5"
version = "0.2.6"
dependencies = [
"bgp-fib",
"bgp-packet",
@@ -224,7 +226,7 @@ dependencies = [
[[package]]
name = "bgp-packet"
version = "0.2.5"
version = "0.2.6"
dependencies = [
"bgp-primitives",
"bitfield",
@@ -240,7 +242,7 @@ dependencies = [
[[package]]
name = "bgp-primitives"
version = "0.2.5"
version = "0.2.6"
dependencies = [
"bytes",
"nom",
@@ -252,7 +254,7 @@ dependencies = [
[[package]]
name = "bgp-rib"
version = "0.2.5"
version = "0.2.6"
dependencies = [
"bgp-packet",
"bgp-primitives",
@@ -267,7 +269,7 @@ dependencies = [
[[package]]
name = "bgp-rpki"
version = "0.2.5"
version = "0.2.6"
dependencies = [
"bgp-primitives",
"bytes",
@@ -280,7 +282,7 @@ dependencies = [
[[package]]
name = "bgp-server"
version = "0.2.5"
version = "0.2.6"
dependencies = [
"bgp-fib",
"bgp-packet",
@@ -288,6 +290,7 @@ dependencies = [
"bgp-rib",
"bgp-rpki",
"bytes",
"foldhash",
"futures",
"nix",
"prometheus",
@@ -302,7 +305,7 @@ dependencies = [
[[package]]
name = "bgp-web"
version = "0.2.5"
version = "0.2.6"
dependencies = [
"axum",
"bgp-fib",
Cargo.toml
+2 -2
View File
@@ -12,7 +12,7 @@ edition = "2024"
homepage = "https://rayhaan.ch"
license = "Apache-2.0"
repository = "https://git.rayhaan.net/rayhaan/ferrix"
version = "0.2.5"
version = "0.2.6"
# Every crate opts in with `[lints] workspace = true`. CI turns warnings into
# errors, so a lint added here is enforced.
@@ -65,7 +65,7 @@ tokio-stream = { version = "0.1.19", features = ["sync"] }
reqwest = { version = "0.13.5", default-features = false, features = ["json"] }
# The daemon's memory allocator; see bin/bgp/src/main.rs. With its
# statistics, which the status page and the metrics show.
tikv-jemallocator = { version = "0.7.0", features = ["stats", "background_threads"] }
tikv-jemallocator = { version = "0.7.0", features = ["stats"] }
tikv-jemalloc-ctl = { version = "0.7.0", features = ["stats"] }
# `cargo build --profile profiling`: the release build, with the line tables
bin/bgp/Cargo.toml
+6
View File
@@ -23,6 +23,12 @@ tracing-subscriber.workspace = true
[target.'cfg(not(target_env = "msvc"))'.dependencies]
tikv-jemallocator.workspace = true
# jemalloc's background thread returns freed memory to the kernel on time
# (see the memory figures on the status page); jemalloc has it on Linux
# only, and warns at every start elsewhere if asked for it.
[target.'cfg(target_os = "linux")'.dependencies]
tikv-jemallocator = { workspace = true, features = ["background_threads"] }
[features]
# jemalloc's heap profiler, for scripts/profile/full-table.sh with
# HEAP=1: not for a production build, where it costs memory and time.
crates/bench/Cargo.toml
+8
View File
@@ -16,11 +16,15 @@ bytes.workspace = true
flate2 = "1"
stats_alloc = { version = "0.1.10", optional = true }
[target.'cfg(not(target_env = "msvc"))'.dependencies]
tikv-jemallocator.workspace = true
[dev-dependencies]
bgp-integration-tests.workspace = true
bgp-rib.workspace = true
bgp-server.workspace = true
tokio.workspace = true
tracing-subscriber.workspace = true
[features]
# Counts every allocation, to report how much memory a table holds. The
@@ -42,3 +46,7 @@ harness = false
[lints]
workspace = true
[[bench]]
name = "mesh"
harness = false
crates/bench/benches/mesh.rs
+469
View File
@@ -0,0 +1,469 @@
//! Two routers of one AS, each taking in full tables from its own peers
//! and sending the other every path (ADD-PATH both ways, `all_paths`), as
//! a mesh of routers that each hold the network's whole view do. How
//! long until each router has the other's routes is what this measures:
//! the exporter reading the table's journal while sessions fill the
//! table, which nothing else exercises at this size.
//!
//! Then a new peer takes the first router's table, as a collector or a
//! router coming back does: how soon its first UPDATE arrives and how
//! long the table takes are measured too. It is a peer of the routers'
//! AS over IPv4, so what it gets is the first router's IPv4 routes from
//! its own peers: the other router's are not reflected to it, and IPv6
//! routes need an IPv6 session.
//!
//! Run with `cargo bench -p bgp-bench --bench mesh`. `BENCH_PEERS`
//! (default 4) feed the first router and `BENCH_PEERS_B` (default 2) the
//! second; `BENCH_SCALE_DOWN` shrinks the table. The routers peer over
//! IPv6 loopback and the feeders come over IPv4 loopback; the new peer
//! comes from 127.0.0.2, which Linux has and macOS does not, so that
//! part is skipped on macOS. With `BENCH_RECEIVER_EARLY=1` the new peer
//! comes as the feeders start instead, so that its exporter goes
//! through the table while the table is being filled, as a router's
//! do after a restart. `BENCH_COLLECTORS` (default 0) more peers of the
//! first router, from 127.0.1.0/24, are peers in the sense of RFC 9234
//! told of learned routes, and the feeders' routes are marked as having
//! come across from a peer, so none of them goes: as many exporters
//! following the table while it fills and passing nearly all of it
//! over, as a router with many peers at exchanges has.
use std::net::{Ipv4Addr, Ipv6Addr, SocketAddr};
use std::sync::Arc;
use std::sync::atomic::{AtomicU64, Ordering};
use std::time::{Duration, Instant};
use bgp_bench::{Prefixes, Shape, announcements, encoded, env};
use bgp_integration_tests::peer_config;
use bgp_integration_tests::test_peer::TestPeer;
use bgp_integration_tests::test_server::TestServer;
use bgp_packet::attributes::PathAttribute;
use bgp_packet::open::{AddPathEntry, Capability, SendReceive};
use bgp_server::config::PeerConfig;
use bgp_server::family::SUPPORTED;
use bgp_server::policy::{ExportPolicy, ImportPolicy, PeerRole};
use bgp_server::session::SessionRegistry;
use tokio::io::{AsyncReadExt, AsyncWriteExt, BufReader};
/// An AS number that is in none of the synthetic paths.
const ASN: u32 = 4_200_000_000;
/// Where the peer that takes the whole table connects from.
const RECEIVER: Ipv4Addr = Ipv4Addr::new(127, 0, 0, 2);
/// Where the collectors connect from: this and the addresses after it.
fn collector_address(n: u32) -> Ipv4Addr {
Ipv4Addr::new(127, 0, 1, n as u8)
}
/// The collectors' AS numbers start here.
const COLLECTOR_ASN: u32 = 65000;
/// What the routers send each other, and the receiver: every path.
fn every_path() -> Arc<ExportPolicy> {
Arc::new(ExportPolicy {
learned_routes: true,
all_paths: true,
..ExportPolicy::default()
})
}
fn free_port() -> u16 {
std::net::TcpListener::bind((Ipv6Addr::LOCALHOST, 0))
.expect("a port")
.local_addr()
.expect("an address")
.port()
}
/// A router of the mesh, listening for feeders on IPv4 loopback and for
/// the other router on IPv6 loopback at `port`, which it dials at
/// `other`.
async fn router(
name: &'static str,
id: u8,
port: u16,
other: u16,
with_receiver: bool,
collectors: u32,
) -> TestServer {
TestServer::start_with(ASN, |c| {
// Two routers of one AS must not share a BGP identifier.
c.router_id = Ipv4Addr::new(192, 0, 2, id);
c.cluster_id = c.router_id;
c.listen_endpoints = vec![
SocketAddr::new(Ipv4Addr::LOCALHOST.into(), 0),
SocketAddr::new(Ipv6Addr::LOCALHOST.into(), port),
];
c.peers = vec![PeerConfig {
add_path: Some(SendReceive::Both),
import: ImportPolicy::permissive(),
export: every_path(),
..peer_config(
name,
SocketAddr::new(Ipv6Addr::LOCALHOST.into(), other),
ASN,
)
}];
if with_receiver {
c.peers.push(PeerConfig {
add_path: Some(SendReceive::Send),
export: every_path(),
..peer_config("receiver", SocketAddr::new(RECEIVER.into(), 179), ASN)
});
}
for n in 1..=collectors {
c.peers.push(PeerConfig {
role: Some(PeerRole::Peer),
export: Arc::new(ExportPolicy {
learned_routes: true,
all_paths: false,
..ExportPolicy::default()
}),
..peer_config(
"collector",
SocketAddr::new(collector_address(n).into(), 179),
COLLECTOR_ASN + n,
)
});
}
c.connect_retry = Duration::from_millis(200);
})
.await
.expect("a router")
}
/// Connects `count` feeders to `router`, numbered from `first`, and
/// returns their sessions with the bytes each is to send.
async fn feeders(
router: &TestServer,
prefixes: &Prefixes,
shape: Shape,
first: u32,
count: u32,
marked: bool,
) -> Vec<(TestPeer, Vec<u8>)> {
let mut sessions = Vec::new();
for peer in first..first + count {
let mut updates = announcements(prefixes, shape, peer);
if marked {
// As having come across from a peer (RFC 9234): not for the
// collectors, which are peers too.
for update in &mut updates {
update
.path_attributes
.push(PathAttribute::OnlyToCustomer(64512 + peer));
}
}
let stream: Vec<u8> = encoded(&updates)
.iter()
.flat_map(|m| m.iter().copied())
.collect();
let mut session = TestPeer::connect(router.addr).await.expect("a connection");
// No hold timer: a feeder has nothing to say after its table.
session
.handshake(64512 + peer, Ipv4Addr::new(10, 0, 0, peer as u8), 0)
.await
.expect("a session");
sessions.push((session, stream));
}
sessions
}
fn main() {
// The routers' logs, with `BENCH_LOG` as a `RUST_LOG` filter.
if let Ok(filter) = std::env::var("BENCH_LOG") {
tracing_subscriber::fmt()
.with_env_filter(filter)
.with_writer(std::io::stderr)
.init();
}
let peers_a = env("BENCH_PEERS", 4) as u32;
let peers_b = env("BENCH_PEERS_B", 2) as u32;
let shape = Shape::INTERNET.scaled_down(env("BENCH_SCALE_DOWN", 1));
let per_peer = shape.prefixes();
println!(
"== Two routers of AS{ASN}, every path both ways: A fed by {peers_a} peers, B by {peers_b}, {} prefixes each",
per_peer
);
let prefixes = Prefixes::generate(shape);
let runtime = tokio::runtime::Builder::new_multi_thread()
.enable_all()
.build()
.expect("a runtime");
runtime.block_on(async {
let (port_a, port_b) = (free_port(), free_port());
let collectors = env("BENCH_COLLECTORS", 0) as u32;
let a = router("b", 1, port_a, port_b, true, collectors).await;
let b = router("a", 2, port_b, port_a, false, 0).await;
a.wait_for_sessions(1).await.expect("the mesh session at A");
b.wait_for_sessions(1).await.expect("the mesh session at B");
// The collectors are there before the feeders, as a router's
// peers are before its tables arrive after a restart.
let mut draining = Vec::new();
for n in 1..=collectors {
let socket = tokio::net::TcpSocket::new_v4().expect("a socket");
if let Err(e) = socket.bind(SocketAddr::new(collector_address(n).into(), 0)) {
println!(
"collectors: skipped, {} is not ours ({e})",
collector_address(n)
);
break;
}
let stream = socket.connect(a.addr).await.expect("A listens");
let mut session = TestPeer::from_stream(stream);
session
.handshake(COLLECTOR_ASN + n, Ipv4Addr::new(192, 0, 3, n as u8), 0)
.await
.expect("A takes a collector's session");
let (mut reader, writer) = session.into_stream().into_split();
draining.push((
writer,
tokio::spawn(async move {
let mut sink = vec![0u8; 1 << 16];
while matches!(reader.read(&mut sink).await, Ok(n) if n > 0) {}
}),
));
}
if collectors > 0 {
a.wait_for_sessions(1 + draining.len())
.await
.expect("the collectors' sessions at A");
}
let feed_a = feeders(&a, &prefixes, shape, 1, peers_a, collectors > 0).await;
let feed_b = feeders(&b, &prefixes, shape, 1 + peers_a, peers_b, false).await;
let own_a = per_peer * peers_a as usize;
let own_b = per_peer * peers_b as usize;
let receiver_routes = shape.ipv4_prefixes * peers_a as usize;
let started = Instant::now();
// Early: as the feeders start, or (2) once A has its own peers'
// routes and B's are still coming, as a router's sessions come
// up after a restart while its peers' tables arrive.
let when = env("BENCH_RECEIVER_EARLY", 0);
let early_receiver = (when > 0).then(|| {
let (addr, registry, rib) = (a.addr, a.registry.clone(), Arc::clone(&a.rib));
tokio::spawn(async move {
while when == 2 && rib.len() < own_a {
tokio::time::sleep(Duration::from_millis(5)).await;
}
resync(addr, registry, receiver_routes).await
})
});
let mut sending = Vec::new();
for (mut session, stream) in feed_a.into_iter().chain(feed_b) {
sending.push(tokio::spawn(async move {
session.send_raw(&stream).await.expect("the router reads");
session
}));
}
let mut a_own = None;
let mut b_own = None;
let mut a_all = None;
let mut b_all = None;
while a_all.is_none() || b_all.is_none() {
tokio::time::sleep(Duration::from_millis(5)).await;
let (at_a, at_b) = (a.rib.len(), b.rib.len());
let now = started.elapsed();
if a_own.is_none() && at_a >= own_a {
a_own = Some(now);
}
if b_own.is_none() && at_b >= own_b {
b_own = Some(now);
}
if a_all.is_none() && at_a >= own_a + own_b {
a_all = Some(now);
}
if b_all.is_none() && at_b >= own_a + own_b {
b_all = Some(now);
}
if now > Duration::from_secs(env("BENCH_DEADLINE", 600) as u64) {
let sessions = |router: &TestServer| -> Vec<String> {
router
.registry
.list()
.iter()
.map(|s| {
format!(
"{} routes {} rejected {}",
s.peer,
s.counters.routes(),
s.counters.rejected()
)
})
.collect()
};
panic!(
"not all routes arrived: A has {at_a} {:?}, B has {at_b} {:?}",
sessions(&a),
sessions(&b)
);
}
}
let row = |what: &str, when: Option<Duration>, routes: usize| {
let when = when.unwrap_or_default();
println!(
"{what:<52} {:>8.3} s {:.2} M routes/s",
when.as_secs_f64(),
routes as f64 / when.as_secs_f64() / 1e6
);
};
row("A has its own peers' routes", a_own, own_a);
row("B has its own peers' routes", b_own, own_b);
row(
"A has B's routes too (B's exporter, A's session)",
a_all,
own_b,
);
row(
"B has A's routes too (A's exporter, B's session)",
b_all,
own_a,
);
// The feeders' sessions stay up until the end: a feeder going
// would have its routes withdrawn from both tables meanwhile.
let mut feeders = Vec::new();
for session in sending {
feeders.push(session.await.expect("a feeder is done"));
}
match early_receiver {
Some(receiver) => receiver.await.expect("the receiver is done"),
None => resync(a.addr, a.registry.clone(), receiver_routes).await,
}
drop(feeders);
drop(draining);
});
}
/// A new peer takes the table of the router at `addr`, whose sessions
/// are in `registry`: `routes` routes with every path. How long until
/// the first UPDATE, and until all of it is sent.
async fn resync(addr: SocketAddr, registry: SessionRegistry, routes: usize) {
let socket = tokio::net::TcpSocket::new_v4().expect("a socket");
if let Err(e) = socket.bind(SocketAddr::new(RECEIVER.into(), 0)) {
println!("A new peer taking A's routes: skipped, {RECEIVER} is not ours ({e})");
return;
}
let stream = socket.connect(addr).await.expect("A listens");
let mut session = TestPeer::from_stream(stream);
let mut capabilities = vec![Capability::FourByteAsn { asn: ASN }];
capabilities.extend(SUPPORTED.into_iter().map(Capability::multiprotocol));
capabilities.push(Capability::AddPath {
entries: SUPPORTED
.iter()
.map(|family| AddPathEntry {
afi: family.afi,
safi: family.safi,
send_recv: SendReceive::Receive,
})
.collect(),
});
session
.handshake_with(ASN, Ipv4Addr::new(192, 0, 2, 9), 90, capabilities)
.await
.expect("A takes the receiver's session");
let started = Instant::now();
let (reader, mut writer) = session.into_stream().into_split();
// KEEPALIVEs, so that the session outlives the hold time.
let keepalives = tokio::spawn(async move {
let mut keepalive = [0xffu8; 19];
keepalive[16..19].copy_from_slice(&[0, 19, 4]);
loop {
tokio::time::sleep(Duration::from_secs(20)).await;
if writer.write_all(&keepalive).await.is_err() {
break;
}
}
});
// When the first UPDATE arrived, in microseconds; 0 until it has.
let first = Arc::new(AtomicU64::new(0));
let updates = Arc::new(AtomicU64::new(0));
let counting = tokio::spawn({
let (first, updates) = (Arc::clone(&first), Arc::clone(&updates));
async move {
let mut reader = BufReader::with_capacity(1 << 20, reader);
let mut header = [0u8; 19];
let mut body = vec![0u8; 4096];
while reader.read_exact(&mut header).await.is_ok() {
let length = usize::from(u16::from_be_bytes([header[16], header[17]]));
if reader.read_exact(&mut body[..length - 19]).await.is_err() {
break;
}
match header[18] {
// An UPDATE (RFC 4271 §4.1).
2 => {
let _ = first.compare_exchange(
0,
started.elapsed().as_micros() as u64,
Ordering::Relaxed,
Ordering::Relaxed,
);
let n = updates.fetch_add(1, Ordering::Relaxed) + 1;
if n % 1_000_000 == 0 {
println!(
" the receiver has read {n} UPDATEs after {:.2} s",
started.elapsed().as_secs_f64()
);
}
}
3 => println!(
"the receiver was sent a NOTIFICATION, error {} subcode {}, after {} UPDATEs",
body[0],
body[1],
updates.load(Ordering::Relaxed)
),
_ => {}
}
}
println!(
"the receiver's connection closed after {} UPDATEs",
updates.load(Ordering::Relaxed)
);
}
});
let deadline = Duration::from_secs(env("BENCH_DEADLINE", 600) as u64);
let advertised = loop {
tokio::time::sleep(Duration::from_millis(5)).await;
let advertised = registry
.list()
.iter()
.find(|s| s.peer.addr.ip() == RECEIVER)
.map_or(0, |s| s.counters.advertised());
if advertised as usize >= routes {
break advertised;
}
if started.elapsed() > deadline {
let sessions: Vec<String> = registry
.list()
.iter()
.map(|s| {
format!(
"{} {:?} export {:?} add_path_send {:?} advertised {}",
s.peer,
s.name,
s.export,
s.add_path_send,
s.counters.advertised()
)
})
.collect();
panic!(
"the receiver has {advertised} of {routes} routes and read {} UPDATEs; sessions at A: {sessions:#?}",
updates.load(Ordering::Relaxed)
);
}
};
let took = started.elapsed();
counting.abort();
keepalives.abort();
let first = Duration::from_micros(first.load(Ordering::Relaxed));
println!(
"{:<52} {:>8.3} s {:.2} M routes/s first UPDATE after {:.0} ms, {} UPDATEs",
"A new peer has A's IPv4 routes (A's exporter)",
took.as_secs_f64(),
advertised as f64 / took.as_secs_f64() / 1e6,
first.as_secs_f64() * 1e3,
updates.load(Ordering::Relaxed),
);
}
crates/bench/examples/packing.rs
+71
View File
@@ -0,0 +1,71 @@
//! How many UPDATEs, and how many octets, a table takes when the routes
//! that share attributes are grouped only within slices of the table
//! taken in shard order, for a few slice sizes: what streaming a table
//! out in slices costs against grouping it whole.
use std::collections::HashMap;
use std::hash::{BuildHasher, RandomState};
use std::path::PathBuf;
use bgp_bench::mrt::Dump;
fn main() {
let path = std::env::var("BENCH_MRT")
.map(PathBuf::from)
.unwrap_or_else(|_| "crates/bench/data/rrc03-bview.20260915.0000.gz".into());
let dump = Dump::read(&path).expect("a readable dump");
let peers = dump.largest_peers(usize::MAX);
let full: Vec<usize> = peers
.iter()
.copied()
.filter(|&p| dump.routes[p].len() > 800_000)
.collect();
println!("{} peers with a full table", full.len());
let hasher = RandomState::new();
for (what, chosen) in [
("one full table", vec![full[0]]),
("every full table, every path", full.clone()),
] {
// (hash of prefix, attribute set, prefix octets) in shard order.
let mut routes: Vec<(u64, u32, usize)> = chosen
.iter()
.flat_map(|&p| dump.routes[p].iter())
.map(|(prefix, set)| {
(
hasher.hash_one(prefix),
*set,
1 + usize::from(prefix.length()).div_ceil(8) + 4,
)
})
.collect();
routes.sort_unstable_by_key(|r| r.0);
println!("== {what}: {} routes", routes.len());
for slice in [4_096usize, 16_384, 65_536, 262_144, usize::MAX] {
let mut updates = 0usize;
let mut octets = 0usize;
for chunk in routes.chunks(slice.min(routes.len())) {
let mut groups: HashMap<u32, usize> = HashMap::new();
for &(_, set, prefix_octets) in chunk {
*groups.entry(set).or_insert(0) += prefix_octets;
}
for (set, prefix_octets) in groups {
// One UPDATE per 4096 octets of prefixes, roughly.
let attrs = dump.attribute_sets[set as usize].len();
let messages = prefix_octets.div_ceil(4000 - attrs.min(3000)).max(1);
updates += messages;
octets += messages * (23 + attrs) + prefix_octets;
}
}
println!(
" slice {:>9}: {:>9} UPDATEs, {:>5.1} MB on the wire",
if slice == usize::MAX {
"whole".to_string()
} else {
slice.to_string()
},
updates,
octets as f64 / 1e6
);
}
}
}
crates/bench/src/lib.rs
+7
View File
@@ -17,6 +17,13 @@
pub mod memory;
pub mod mrt;
// The benchmarks allocate as the daemon does (bin/bgp/src/main.rs), with
// jemalloc; the C library's allocator would put its own locks in the
// profile. Not when memory is counted: that needs the counting one.
#[cfg(all(not(feature = "count-memory"), not(target_env = "msvc")))]
#[global_allocator]
static ALLOCATOR: tikv_jemallocator::Jemalloc = tikv_jemallocator::Jemalloc;
use std::collections::HashSet;
use std::net::{Ipv4Addr, Ipv6Addr};
crates/integration-tests/tests/deadlock.rs
+92
View File
@@ -0,0 +1,92 @@
//! A peer that sends us a table while not reading what we send it must
//! not stop us reading it: two routers sending each other their tables at
//! once would otherwise each wait on the other for ever.
use std::net::Ipv4Addr;
use std::sync::Arc;
use std::time::Duration;
use bgp_integration_tests::announcement::Announcement;
use bgp_integration_tests::prefix;
use bgp_integration_tests::test_peer::TestPeer;
use bgp_integration_tests::test_server::TestServer;
use bgp_server::policy::ExportPolicy;
use eyre::Result;
const LOCAL_ASN: u32 = 64601;
const ROUTES: u32 = 100_000;
fn nth(n: u32) -> bgp_primitives::ip_prefix::IpPrefix {
let [a, b, c, _] = (0x0a00_0000u32 + n * 256).to_be_bytes();
prefix(&format!("{a}.{b}.{c}.0/24"))
}
#[tokio::test]
async fn a_peer_that_does_not_read_is_still_read() -> Result<()> {
let server = TestServer::start_with(LOCAL_ASN, |c| {
c.export = Arc::new(ExportPolicy {
learned_routes: true,
all_paths: false,
..ExportPolicy::default()
});
})
.await?;
// One peer's table, for the server to pass on.
let mut first = server
.established(64701, Ipv4Addr::new(10, 0, 0, 1))
.await?;
// A route to an UPDATE, each with attributes of its own, as a real
// table has them: passing it on is then megabytes, not a few hundred
// UPDATEs that the socket buffers would take whole.
for i in 0..ROUTES {
first
.announce(
Announcement::of([nth(i).into()], &[64701, 64510])
.with_communities(&[((i >> 16) as u16 + 1, i as u16)]),
)
.await?;
}
server.wait_for_routes(ROUTES as usize).await?;
// A second peer, which the server starts sending that table to as
// soon as it is up, sends a table of its own and reads nothing back:
// the server's writes to it fill its socket and stop. The server has
// to go on reading it all the same. The peer's receive buffer is
// small, as a real link's window would be, so that this happens with
// the first few hundred kilobytes and not after the loopback's
// megabytes.
let socket = tokio::net::TcpSocket::new_v4()?;
socket.set_recv_buffer_size(16 * 1024)?;
let stream = socket.connect(server.addr).await?;
let mut second = TestPeer::from_stream(stream);
second
.handshake(64702, Ipv4Addr::new(10, 0, 0, 2), 90)
.await?;
server.wait_for_sessions(2).await?;
// Sent over a couple of seconds, so that the server's export to the
// peer is well under way while the peer is still sending.
let sending = async {
for start in (0..ROUTES).step_by(100) {
second
.announce(Announcement::of(
(start..start + 100).map(|i| nth(i).into()),
&[64702, 64510],
))
.await?;
if start % 2000 == 0 {
tokio::time::sleep(Duration::from_millis(40)).await;
}
}
Ok::<_, eyre::Report>(())
};
tokio::time::timeout(Duration::from_secs(30), sending)
.await
.expect("the server reads the peer's table though the peer reads nothing")?;
tokio::time::timeout(
Duration::from_secs(30),
server.wait_for_routes(2 * ROUTES as usize),
)
.await
.expect("every route is in the table")?;
Ok(())
}
crates/packet/src/message.rs
+2 -2
View File
@@ -30,7 +30,7 @@ pub const BGP_HEADER_LEN: u16 = 19;
/// Maximum BGP message length per RFC 4271 §4.1.
pub const BGP_MAX_MSG_LEN: u16 = 4096;
const BGP_MARKER: [u8; 16] = [0xFF; 16];
pub const BGP_MARKER: [u8; 16] = [0xFF; 16];
/// BGP message type codes (RFC 4271 §4.1, RFC 2918).
mod msg_type {
@@ -358,7 +358,7 @@ impl UpdateMessage {
/// Writes prefixes as an UPDATE names them: for each, its path identifier
/// if the session uses them for the family (RFC 7911 §3), its length in
/// bits, and the bytes that length covers.
pub(crate) fn write_nlri(nlri: &[Nlri], with_path_id: bool, out: &mut BytesMut) {
pub fn write_nlri(nlri: &[Nlri], with_path_id: bool, out: &mut BytesMut) {
for n in nlri {
if with_path_id {
out.put_u32(n.path_id);
crates/rib/src/table/attrs.rs
+16
View File
@@ -209,6 +209,22 @@ impl AttrSet {
plain
}
/// Whether the route carries the OTC attribute (RFC 9234): it came
/// down from a provider or across from a peer, and is to go on down
/// only. Read off the stored bytes without building anything else,
/// for those who ask it of every route: a session to a peer or a
/// provider asks it of most of the table, to pass over them.
pub fn is_only_to_customer(&self) -> bool {
let mut wire = &*self.wire;
while let Ok((attribute, rest)) = split_attribute(wire) {
if attribute.type_code == attr_type::ONLY_TO_CUSTOMER {
return true;
}
wire = rest;
}
false
}
/// Whether the route carries the BLACKHOLE community (RFC 7999), which
/// asks for the packets to its prefix to be dropped. Read off the
/// stored bytes without building anything else, for those who ask it
crates/rib/src/table/journal.rs
+60 -8
View File
@@ -113,6 +113,10 @@ struct Journal {
/// Where each reader has read to, by reader slot. Written by the
/// readers themselves while they hold the shard for reading.
cursors: Vec<AtomicU64>,
/// `seq`, for reading without the shard's lock: the table holds the
/// other end, so that a reader can tell a shard with nothing new for
/// it from one with something without stopping the shard's writers.
latest: Arc<AtomicU64>,
/// How many reader slots are in use.
readers: usize,
limits: Limits,
@@ -125,6 +129,7 @@ impl Journal {
Self {
kind,
limits,
latest: Arc::new(AtomicU64::new(0)),
seq: 0,
floor: 0,
entries: VecDeque::new(),
@@ -139,6 +144,7 @@ impl Journal {
/// number for the row to remember.
fn changed(&mut self, prefix: IpPrefix) -> u64 {
self.seq += 1;
self.latest.store(self.seq, Ordering::Release);
if self.readers > 0 {
self.entries.push_back(Entry {
seq: self.seq,
@@ -161,6 +167,7 @@ impl Journal {
/// Notes that `prefix` lost its last route and has no row any more.
fn disappeared(&mut self, prefix: IpPrefix) {
self.seq += 1;
self.latest.store(self.seq, Ordering::Release);
if self.readers > 0 {
self.entries.push_back(Entry {
seq: self.seq,
@@ -253,6 +260,16 @@ pub(super) enum Kind {
All,
}
impl Kind {
/// Its place in what the table keeps for each shard's two journals.
pub(super) fn index(self) -> usize {
match self {
Self::Best => 0,
Self::All => 1,
}
}
}
/// The sequence numbers a row remembers, one for each journal: of the
/// latest change of its best route, and of any of its routes.
#[derive(Debug, Clone, Copy)]
@@ -285,6 +302,12 @@ impl Journals {
}
}
/// The latest sequence number of each journal, by [`Kind::index`],
/// as something to read without the shard's lock.
pub(super) fn latest(&self) -> [Arc<AtomicU64>; 2] {
[Arc::clone(&self.best.latest), Arc::clone(&self.all.latest)]
}
fn of(&self, kind: Kind) -> &Journal {
match kind {
Kind::Best => &self.best,
@@ -598,7 +621,11 @@ impl Reader {
pub fn poll(&mut self, budget: usize, mut visit: impl FnMut(Event<'_>)) -> Progress {
// A call that may give nothing out would never get anywhere.
let budget = budget.max(1);
let shards = self.table.shards.len();
// Held once for the call: a count taken on the table for every
// shard of every pass, by every reader, was the most of what a
// router with many peers did.
let table = Arc::clone(&self.table);
let shards = table.shards.len();
let mut events = 0;
loop {
match self.phase {
@@ -621,7 +648,7 @@ impl Reader {
self.phase = Phase::Follow { next: 0, upto: 0 };
continue;
}
events += self.scan(next, upto, &mut visit);
events += self.scan(&table, next, upto, &mut visit);
self.phase = Phase::Resync {
next: next + 1,
announced: true,
@@ -654,7 +681,14 @@ impl Reader {
self.phase = Phase::Follow { next: shard, upto };
return self.progress(events, false);
}
match self.drain(shard, upto, &mut visit) {
// Nothing new in this shard, by the number that is
// read without its lock: it is left alone, and so
// are the sessions writing to it.
if self.cursors[shard] == self.table.latest_seq(shard, self.kind) {
shard += 1;
continue;
}
match self.drain(&table, shard, upto, &mut visit) {
Drained::Lagged => {
lagged = true;
break;
@@ -706,6 +740,11 @@ impl Reader {
Progress { events, caught_up }
}
/// Whether the last [`Reader::poll`] read everything there was.
pub fn caught_up(&self) -> bool {
self.caught_up
}
/// Waits until there may be something to read. Returns at once if the
/// last [`Reader::poll`] did not catch up.
pub async fn changed(&mut self) {
@@ -760,8 +799,13 @@ impl Reader {
/// as of `upto` on the table's clock, in one hold of the shard: what
/// changed after that time is in the journal after that place, whether
/// or not the rows given out show it already.
fn scan(&mut self, index: usize, upto: u64, visit: &mut impl FnMut(Event<'_>)) -> usize {
let table = Arc::clone(&self.table);
fn scan(
&mut self,
table: &RibTable,
index: usize,
upto: u64,
visit: &mut impl FnMut(Event<'_>),
) -> usize {
let shard = read(&table.shards[index]);
for row in &shard.rows {
visit(self.event_for(row));
@@ -797,9 +841,12 @@ impl Reader {
// many changes were made with it.
let mut earliest = None;
for (index, shard) in self.table.shards.iter().enumerate() {
let cursor = self.cursors[index];
if cursor == self.table.latest_seq(index, self.kind) {
continue;
}
let shard = read(shard);
let journal = shard.journals.of(self.kind);
let cursor = self.cursors[index];
if cursor < journal.floor || cursor == journal.seq {
continue;
}
@@ -817,8 +864,13 @@ impl Reader {
/// Reads a shard's journal from this reader's place, no further than
/// the changes made by `upto` on the table's clock.
fn drain(&mut self, index: usize, upto: u64, visit: &mut impl FnMut(Event<'_>)) -> Drained {
let table = Arc::clone(&self.table);
fn drain(
&mut self,
table: &RibTable,
index: usize,
upto: u64,
visit: &mut impl FnMut(Event<'_>),
) -> Drained {
let shard = read(&table.shards[index]);
let journal = shard.journals.of(self.kind);
let mut cursor = self.cursors[index];
crates/rib/src/table/mod.rs
+19 -1
View File
@@ -317,6 +317,9 @@ struct Shard {
/// takes `&self`.
pub struct RibTable {
shards: Box<[RwLock<Shard>]>,
/// The latest sequence number of each shard's two journals, kept by
/// the journals themselves, for reading without the shard's lock.
latest: Box<[[Arc<AtomicU64>; 2]]>,
hasher: RandomState,
attrs: AttrStore,
/// Which peer numbers are in use. Changing the peer list also takes
@@ -355,19 +358,23 @@ impl RibTable {
/// sizes. Tests use small ones to get there quickly.
fn with_journal_limits(shards: usize, limits: journal::Limits) -> Self {
let hasher = RandomState::default();
let mut latest = Vec::with_capacity(shards.max(1));
Self {
shards: (0..shards.max(1))
.map(|_| {
let journals = journal::Journals::new(limits);
latest.push(journals.latest());
RwLock::new(Shard {
rows: HashTable::new(),
routes: 0,
peers: Vec::new(),
hasher: hasher.clone(),
scratch: select::Scratch::default(),
journals: journal::Journals::new(limits),
journals,
})
})
.collect(),
latest: latest.into_boxed_slice(),
hasher,
attrs: AttrStore::new(),
peers: Mutex::default(),
@@ -651,6 +658,12 @@ impl RibTable {
}
}
/// The latest sequence number of shard `shard`'s journal of `kind`,
/// without the shard's lock.
fn latest_seq(&self, shard: usize, kind: journal::Kind) -> u64 {
self.latest[shard][kind.index()].load(std::sync::atomic::Ordering::Acquire)
}
/// The time on the clock up to which every change is complete in every
/// shard: everything before the earliest one still under way.
pub(super) fn settled(&self) -> u64 {
@@ -873,6 +886,11 @@ fn push_tight<T>(vec: &mut Vec<T>, value: T, expected: usize) {
let first = expected.min(FIRST_ROOM);
let more = if vec.capacity() < first {
first - vec.len()
} else if vec.capacity() < FIRST_ROOM {
// A peer that sends several routes for a prefix (ADD-PATH)
// takes a row past the peers' count: doubled while small,
// and from there by a quarter.
vec.capacity()
} else {
(vec.len() / 4).max(1)
};
crates/server/Cargo.toml
+1
View File
@@ -8,6 +8,7 @@ repository.workspace = true
version.workspace = true
[dependencies]
foldhash.workspace = true
bgp-packet.workspace = true
bgp-primitives.workspace = true
bgp-rib.workspace = true
crates/server/src/advertise.rs
+609 -150
View File
File diff suppressed because it is too large. Load diff
crates/server/src/config.rs
+1
View File
@@ -646,6 +646,7 @@ fn parse_import(import: &proto::ImportPolicy) -> Result<ImportPolicy, ImportProb
accept_aspa_invalid: import.accept_aspa_invalid,
accept_as_sets: import.accept_as_sets,
accept_third_party_next_hop: import.accept_third_party_next_hop,
accept_loopback_next_hop: false,
accept_blackholes: import.accept_blackholes,
own_prefixes: Arc::default(),
originated: Arc::default(),
crates/server/src/connection.rs
+226 -29
View File
@@ -31,7 +31,9 @@ use prometheus::IntGauge;
use tokio::io::{AsyncRead, AsyncReadExt, AsyncWriteExt};
use tokio::net::TcpStream;
use tokio::net::tcp::{OwnedReadHalf, OwnedWriteHalf};
use tokio::sync::mpsc::error::TrySendError;
use tokio::sync::{mpsc, watch};
use tokio::task::JoinHandle;
use tokio::time::{Instant, Interval, MissedTickBehavior, Sleep, interval_at, sleep, timeout};
use tracing::{debug, error, info, warn};
@@ -53,12 +55,26 @@ use crate::rejections::{PeerRejections, RejectionLog};
use crate::restart::{self, Agreed, GracefulRestartConfig, Leave, Restarts};
use crate::routes::{self, Held};
use crate::rpki::{Checks, Subject, Validator};
use crate::session::{PeerId, SessionCounters, SessionGuard, SessionHandle, SessionRegistry};
use crate::session::{
Outbound, PeerId, SessionCounters, SessionGuard, SessionHandle, SessionRegistry,
};
use crate::throttle::Throttle;
/// Number of outbound messages that may be queued for a session before
/// senders are back-pressured.
const OUTBOUND_QUEUE_CAPACITY: usize = 1024;
/// How many items may be queued for a session before senders wait: each
/// a message, or a piece of up to [`export::ENCODED_PIECE`] octets of
/// UPDATEs from the exporter, so a few megabytes of a table at most.
const OUTBOUND_QUEUE_CAPACITY: usize = 64;
/// How many octets of messages are gathered before they are written to
/// the socket, when more are waiting. A full table is a million small
/// UPDATEs; written one at a time, each was a segment of its own, and a
/// receiver that acknowledges late held the sender at a few segments a
/// round trip. Gathered, they go as full segments in one write.
const WRITE_BATCH_BYTES: usize = 64 * 1024;
/// How many batches may wait for the writer task at a time: a megabyte
/// or so a session, beyond what the socket's own buffer takes.
const WRITER_QUEUE: usize = 16;
/// The least time a write to the peer may make no progress before the
/// session is given up: the Send Hold Timer of RFC 9687, which is this or
@@ -358,7 +374,7 @@ struct Established {
/// The peer's number in the routing table.
rib_peer: PeerIndex,
/// Messages the rest of the server queued for the peer.
outbound: mpsc::Receiver<Message>,
outbound: mpsc::Receiver<Outbound>,
/// Where the rest of the server asks for the session to end, and why.
ending: mpsc::UnboundedReceiver<SessionError>,
/// The routes learned from the peer, which go when the session does.
@@ -595,6 +611,11 @@ struct Validation {
enum Wake {
/// Something for the state machine.
Event(Event),
/// The writer has room for a batch that was waiting.
WriterRoom(mpsc::OwnedPermit<BytesMut>),
/// Messages from the rest of the server to send, as many as were
/// waiting, to be written together.
Outbound(Vec<Outbound>),
/// The validation data has changed; these routes of the peer's are
/// now invalid.
Revalidate(Vec<Nlri>),
@@ -628,8 +649,17 @@ struct Session {
said: Said,
fsm: Fsm,
frames: FrameReader<OwnedReadHalf>,
writer: OwnedWriteHalf,
/// Reused for every message written.
/// Batches of messages on their way to the writer task, which writes
/// them to the peer; see [`write_to_peer`].
writer: mpsc::Sender<BytesMut>,
writer_task: JoinHandle<()>,
/// Batches the writer had no room for yet, oldest first. While there
/// are any, no more messages are taken from the rest of the server;
/// the peer is still read.
pending: VecDeque<BytesMut>,
/// What the writer task says if it fails.
write_failed: mpsc::Receiver<SessionError>,
/// Messages encoded and not yet made a batch of.
write_buf: BytesMut,
/// How messages from the peer are decoded: the width of AS numbers and
/// where prefixes carry path identifiers are known once the OPENs are
@@ -642,7 +672,8 @@ struct Session {
hold_timer: Option<Pin<Box<Sleep>>>,
keepalive_timer: Option<Interval>,
/// How long a write may make no progress; see [`MIN_SEND_HOLD_TIME`].
send_hold_time: Duration,
/// The writer task watches it.
send_hold: watch::Sender<Duration>,
/// Whether a NOTIFICATION of ours is on its way to the peer.
notified: bool,
/// This connection's slot in the collision table, once admitted.
@@ -671,7 +702,15 @@ impl Session {
peer: PeerSettings,
) -> io::Result<(Self, Vec<Action>)> {
let local_addr = stream.local_addr()?;
// Messages are written in batches of full segments (see
// `WRITE_BATCH_BYTES`); a batch's last, partial segment is not to
// wait for the acknowledgement of the one before it.
let _ = stream.set_nodelay(true);
let (reader, writer) = stream.into_split();
let (batches, queue) = mpsc::channel(WRITER_QUEUE);
let (failed, write_failed) = mpsc::channel(1);
let (send_hold, hold) = watch::channel(MIN_SEND_HOLD_TIME);
let writer_task = tokio::spawn(write_to_peer(writer, queue, hold, failed));
let (fsm, first_actions) = Fsm::new(FsmConfig {
router_id: server.router_id,
local_asn: server.local_asn,
@@ -696,13 +735,16 @@ impl Session {
rejections: server.rejections.of_peer(peer_addr.into(), &label),
fsm,
frames: FrameReader::new(reader),
writer,
writer: batches,
writer_task,
pending: VecDeque::new(),
write_failed,
write_buf: BytesMut::new(),
ctx: ParserContext::default(),
send_ctx: ParserContext::default(),
hold_timer: None,
keepalive_timer: None,
send_hold_time: MIN_SEND_HOLD_TIME,
send_hold,
notified: false,
collision: None,
established: None,
@@ -728,10 +770,18 @@ impl Session {
error!(peer_addr = %socket_text(self.peer_addr), err = %error, "Closing session after a panic");
if let Some(notification) = error.notification() {
let notification = Message::Notification(notification);
// Contained as well: whatever panicked may do so again.
if !matches!(contained(self.send(&notification)).await, Ok(Ok(()))) {
// Whatever was gathered to be written when the panic came is
// not to go; the NOTIFICATION is, now. Contained as well:
// whatever panicked may do so again.
self.write_buf.clear();
self.pending.clear();
if !matches!(
contained(async { self.send(&notification) }).await,
Ok(Ok(()))
) {
warn!(peer_addr = %socket_text(self.peer_addr), "Could not send NOTIFICATION after a panic");
}
self.flush();
}
error
}
@@ -741,12 +791,25 @@ impl Session {
async fn close(mut self) {
// A new connection from the peer must not find this one in its way.
self.collision = None;
// Whatever is still to go out, a NOTIFICATION among it, is given to
// the writer, which closes our side once it has written it.
self.flush();
let hand_over = async {
for batch in std::mem::take(&mut self.pending) {
if self.writer.send(batch).await.is_err() {
break;
}
}
};
let _ = timeout(CLOSE_LINGER, hand_over).await;
let (closed, _) = mpsc::channel(1);
drop(std::mem::replace(&mut self.writer, closed));
let _ = timeout(CLOSE_LINGER, &mut self.writer_task).await;
if !self.notified {
return;
}
// Our side is done sending; what the peer still sends is read and
// thrown away until it closes its side too.
let _ = self.writer.shutdown().await;
let _ = timeout(CLOSE_LINGER, self.frames.discard_until_closed()).await;
}
@@ -758,6 +821,32 @@ impl Session {
}
actions = match self.next_event().await? {
Wake::Event(event) => self.fsm.handle(event),
Wake::Outbound(messages) => {
let mut actions = Vec::new();
for item in messages {
match item {
Outbound::Message(msg) => {
actions.extend(self.fsm.handle(Event::Outbound(msg)));
}
// Encoded by the exporter for this session as
// it is now; one for a session that is no
// longer established is void.
Outbound::Encoded(update) => {
if self.established.is_some() {
self.send_encoded(update);
}
}
}
}
actions
}
Wake::WriterRoom(permit) => {
if let Some(batch) = self.pending.pop_front() {
permit.send(batch);
}
self.push_pending();
vec![]
}
Wake::Revalidate(suspects) => self.revalidate(&suspects),
Wake::Swept(outcome) => self.swept(&outcome),
Wake::StaleTimeout => {
@@ -794,7 +883,15 @@ impl Session {
},
() = expired(&mut self.hold_timer) => Event::HoldTimerExpired,
() = ticked(&mut self.keepalive_timer) => Event::KeepaliveTimerExpired,
wake = from_server(&mut self.established) => return Ok(wake),
Some(error) = self.write_failed.recv() => Event::LocalFailure(error),
permit = self.writer.clone().reserve_owned(), if !self.pending.is_empty() => {
return Ok(match permit {
Ok(permit) => Wake::WriterRoom(permit),
// The writer task is gone; its word is on its way.
Err(_) => return Ok(Wake::Event(Event::LocalFailure(SessionError::SendHoldTimerExpired(*self.send_hold.borrow())))),
});
}
wake = from_server(&mut self.established, self.pending.is_empty()) => return Ok(wake),
() = killed(&mut self.collision) => Event::CollisionLost,
leave = leaving(&mut self.leave) => Event::Leave(leave),
};
@@ -826,12 +923,20 @@ impl Session {
}
/// Carries out `actions` in order. Returns the session's result once it
/// is over.
/// is over. What the actions send is written once they are all done,
/// or as each batch of `WRITE_BATCH_BYTES` fills.
async fn execute(&mut self, actions: Vec<Action>) -> Option<Result<(), SessionError>> {
let result = self.execute_all(actions).await;
// Whatever came of the actions goes to the writer as one batch.
self.flush();
result
}
async fn execute_all(&mut self, actions: Vec<Action>) -> Option<Result<(), SessionError>> {
let mut queue: VecDeque<Action> = actions.into();
while let Some(action) = queue.pop_front() {
match action {
Action::Send(msg) => match self.send(&msg).await {
Action::Send(msg) => match self.send(&msg) {
Ok(()) => {}
// The session is ending anyway and the Close that
// follows says why; a NOTIFICATION is best effort.
@@ -941,7 +1046,8 @@ impl Session {
..self.ctx.clone()
};
let hold_time = Duration::from_secs(negotiated.hold_time_secs.into());
self.send_hold_time = MIN_SEND_HOLD_TIME.max(2 * hold_time);
self.send_hold
.send_replace(MIN_SEND_HOLD_TIME.max(2 * hold_time));
self.collision = self.server.collisions.admit(
self.server.router_id,
@@ -1037,7 +1143,7 @@ impl Session {
negotiated: &Negotiated,
peer_id: PeerId,
rib_peer: PeerIndex,
outbound: mpsc::Sender<Message>,
outbound: mpsc::Sender<Outbound>,
) -> SessionHandle {
let peer = &self.peer;
let mut handle = SessionHandle::new(
@@ -1285,7 +1391,9 @@ impl Session {
vec![]
}
async fn send(&mut self, msg: &Message) -> Result<(), SessionError> {
/// Puts `msg` in the write buffer, and makes a batch of the buffer if
/// it has a batch's worth; the rest goes with [`Session::flush`].
fn send(&mut self, msg: &Message) -> Result<(), SessionError> {
if let (Message::Update(update), Some(established)) = (msg, &self.established) {
let (withdrawn, announced) = prefix_counts(update);
debug!(peer = %established.peer_id, withdrawn, announced, "Sending UPDATE");
@@ -1294,17 +1402,66 @@ impl Session {
info!(peer_addr = %socket_text(self.peer_addr), error = %notification.error, "Sending NOTIFICATION, closing session");
self.notified = true;
}
self.write_buf.clear();
msg.to_wire(&self.send_ctx, &mut self.write_buf)?;
match timeout(self.send_hold_time, self.writer.write_all(&self.write_buf)).await {
Ok(written) => written?,
Err(_) => return Err(SessionError::SendHoldTimerExpired(self.send_hold_time)),
let before = self.write_buf.len();
if let Err(e) = msg.to_wire(&self.send_ctx, &mut self.write_buf) {
self.write_buf.truncate(before);
return Err(e.into());
}
MESSAGES_SENT
.with_label_values(&[message_type_name(msg)])
.inc();
if self.write_buf.len() >= WRITE_BATCH_BYTES {
self.flush();
}
Ok(())
}
/// Puts a piece of UPDATEs that are already encoded in the write
/// buffer.
fn send_encoded(&mut self, piece: BytesMut) {
// Counted by their headers: a piece is of whole messages.
let mut messages = 0;
let mut at = 0;
while at + 18 <= piece.len() {
at += usize::from(u16::from_be_bytes([piece[at + 16], piece[at + 17]])).max(19);
messages += 1;
}
self.write_buf.extend_from_slice(&piece);
MESSAGES_SENT
.with_label_values(&["update"])
.inc_by(messages);
if self.write_buf.len() >= WRITE_BATCH_BYTES {
self.flush();
}
}
/// Makes a batch of the write buffer and gives it to the writer, or
/// keeps it until the writer has room. Never waits: a session that
/// waited to write would not be reading, and a peer doing the same
/// would wait on it for ever.
fn flush(&mut self) {
if self.write_buf.is_empty() {
return;
}
// Split off, not copied: the buffer's bytes go as they are.
self.pending.push_back(self.write_buf.split());
self.push_pending();
}
/// Gives the writer what it has room for, oldest first.
fn push_pending(&mut self) {
while let Some(batch) = self.pending.pop_front() {
match self.writer.try_send(batch) {
Ok(()) => {}
Err(TrySendError::Full(batch)) => {
self.pending.push_front(batch);
return;
}
// The writer task is gone; its word is on its way.
Err(TrySendError::Closed(_)) => return,
}
}
}
}
// ---------------------------------------------------------------------------
@@ -1353,7 +1510,7 @@ fn sweep(
/// to send to the peer, a request to end the session, word that the
/// validation data has changed, the outcome of a sweep of stale routes,
/// or the time being up for the routes kept from the peer's last session.
async fn from_server(established: &mut Option<Established>) -> Wake {
async fn from_server(established: &mut Option<Established>, may_take_outbound: bool) -> Wake {
let Some(Established {
outbound,
ending,
@@ -1379,7 +1536,16 @@ async fn from_server(established: &mut Option<Established>) -> Wake {
};
tokio::select! {
Some(error) = ending.recv() => Wake::Event(Event::LocalFailure(error)),
Some(msg) = outbound.recv() => Wake::Event(Event::Outbound(msg)),
Some(msg) = outbound.recv(), if may_take_outbound => {
// And whatever else is waiting, to go out together.
let mut messages = vec![msg];
while messages.len() < OUTBOUND_QUEUE_CAPACITY
&& let Ok(more) = outbound.try_recv()
{
messages.push(more);
}
Wake::Outbound(messages)
}
Some(suspects) = invalidated => Wake::Revalidate(suspects),
Some(outcome) = swept.reported.recv() => Wake::Swept(outcome),
() = stale_timeout => Wake::StaleTimeout,
@@ -1390,6 +1556,33 @@ async fn from_server(established: &mut Option<Established>) -> Wake {
}
}
/// Writes a session's batches to the peer, as its own task, so that the
/// session's reading never waits on its writing. Two routers sending each
/// other their tables at once would otherwise each stop reading while
/// held up writing to the other's closed window, and neither would read
/// again until the send hold timer ended both sessions. A write that
/// makes no progress for the send hold time, or fails, ends the session
/// through `failed`; the remaining batches are dropped with it. Once the
/// session has nothing more to send, our side of the connection is shut.
async fn write_to_peer(
mut writer: OwnedWriteHalf,
mut queue: mpsc::Receiver<BytesMut>,
send_hold: watch::Receiver<Duration>,
failed: mpsc::Sender<SessionError>,
) {
while let Some(batch) = queue.recv().await {
let hold_time = *send_hold.borrow();
let error = match timeout(hold_time, writer.write_all(&batch)).await {
Ok(Ok(())) => continue,
Ok(Err(e)) => e.into(),
Err(_) => SessionError::SendHoldTimerExpired(hold_time),
};
let _ = failed.try_send(error);
return;
}
let _ = writer.shutdown().await;
}
/// Resolves with how the daemon is leaving, once it is. Cancel-safe.
async fn leaving(leave: &mut watch::Receiver<Option<Leave>>) -> Leave {
let leave = match leave.wait_for(Option::is_some).await {
@@ -1672,18 +1865,22 @@ mod tests {
PeerSettings::unconfigured(90, Arc::default(), None, GracefulRestartConfig::default()),
)
.unwrap();
session.send_hold_time = Duration::from_millis(100);
session.send_hold.send_replace(Duration::from_millis(100));
// Until the socket's buffers are full, writes succeed.
// Until the socket's buffers are full, writes go through; then
// the writer gives up, and says so.
let filler = Message::Notification(NotificationMessage {
error: BgpError::Cease(CeaseSubcode::AdminShutdown),
data: vec![0; 4000],
});
let stuck = tokio::time::timeout(Duration::from_secs(30), async {
loop {
if let Err(e) = session.send(&filler).await {
session.send(&filler).unwrap();
session.flush();
if let Ok(e) = session.write_failed.try_recv() {
return e;
}
tokio::time::sleep(Duration::from_millis(1)).await;
}
})
.await
crates/server/src/export.rs
+259 -2
View File
@@ -33,12 +33,15 @@
use std::net::{IpAddr, Ipv4Addr};
use bgp_packet::message::{
AsPathAttribute, AsPathSegment, LocalPrefPathAttribute, MpReachNlriPathAttribute,
AsPathAttribute, AsPathSegment, BGP_MARKER, LocalPrefPathAttribute, MpReachNlriPathAttribute,
MpUnreachNlriPathAttribute, NextHopPathAttribute, NlriNextHop, PathAttribute, UpdateMessage,
attr_type, write_nlri,
};
use bgp_primitives::constants::{AfiSafi, IpVersion};
use bgp_primitives::ip_prefix::Nlri;
use bgp_primitives::parser::ParserContext;
use bgp_rib::table::{DEFAULT_LOCAL_PREF, PeerIndex};
use bytes::{BufMut, Bytes, BytesMut};
use crate::advertise::{Towards, ipv6_next_hop};
use crate::marks::{self, Control, Named, communities_of, well_known_allow};
@@ -399,9 +402,141 @@ fn nlri_octets(routes: IpVersion) -> usize {
4 + 1 + usize::from(routes.bits()) / 8
}
/// A route's attributes as they go out on one session, encoded once and
/// kept: what every UPDATE announcing a route with them carries, less the
/// next hop, which goes with the prefixes. `at_next_hop` and
/// `at_mp_reach` are where in `bytes` a NEXT_HOP (type 3) and an
/// MP_REACH_NLRI (type 14) go, so that the attributes stay in order of
/// type code.
#[derive(Debug)]
pub struct EncodedAttributes {
pub bytes: Bytes,
pub at_next_hop: usize,
pub at_mp_reach: usize,
}
impl EncodedAttributes {
/// Encodes `attributes`, which are `finish`ed, for a session with
/// `four_octet_asn`. `None` if they cannot be encoded.
pub fn encode(attributes: &[PathAttribute], four_octet_asn: bool) -> Option<Self> {
let ctx = ParserContext {
four_octet_asn,
ibgp: false,
add_path: Vec::new(),
};
let mut out = BytesMut::with_capacity(256);
let mut at_next_hop = None;
let mut at_mp_reach = None;
for attribute in attributes {
let code = attribute.type_code();
if code > attr_type::NEXT_HOP && at_next_hop.is_none() {
at_next_hop = Some(out.len());
}
if code > attr_type::MP_REACH_NLRI && at_mp_reach.is_none() {
at_mp_reach = Some(out.len());
}
attribute.to_wire(&ctx, &mut out).ok()?;
}
Some(Self {
at_next_hop: at_next_hop.unwrap_or(out.len()),
at_mp_reach: at_mp_reach.unwrap_or(out.len()),
bytes: out.freeze(),
})
}
pub fn octets(&self) -> usize {
self.bytes.len()
}
}
/// What the next hop takes, in its own attribute or with the prefixes,
/// and a little to spare.
const NEXT_HOP_ROOM: usize = 64;
/// How many octets of UPDATEs go to a session in one piece: what the
/// session hands its writer at a time. A piece holds whole messages,
/// so it runs over by at most one.
pub(crate) const ENCODED_PIECE: usize = 64 * 1024;
/// The UPDATE messages, complete with their headers, that announce
/// `routes`, all of IP version `version`, with the attributes `encoded`
/// and `next_hop`: as few as they fit in, in pieces of
/// [`ENCODED_PIECE`] octets or so, each of whole messages. `ctx` is the
/// session's, and says whether path identifiers go with the family's
/// prefixes (RFC 7911).
///
/// What [`announcements`] builds as messages, built as bytes: the
/// attributes are not parsed, cloned or encoded again for each message.
pub(crate) fn encoded_announcements(
encoded: &EncodedAttributes,
next_hop: NlriNextHop,
version: IpVersion,
routes: &[Nlri],
ctx: &ParserContext,
) -> Vec<BytesMut> {
let room = UPDATE_ROOM.saturating_sub(encoded.octets() + NEXT_HOP_ROOM);
let at_once = (room / nlri_octets(version)).max(1);
let in_body = matches!(next_hop, NlriNextHop::Ipv4(_)) && version == IpVersion::Ipv4;
let with_path_id = ctx.add_path.contains(&AfiSafi::unicast(version));
let mut pieces = Vec::new();
let mut out = BytesMut::with_capacity(ENCODED_PIECE.min(routes.len() * 64 + 4096));
for routes in routes.chunks(at_once) {
{
out.put_slice(&BGP_MARKER);
let length_at = out.len();
out.put_u16(0);
// The type of an UPDATE (RFC 4271 §4.1).
out.put_u8(2);
// No withdrawn routes.
out.put_u16(0);
let attributes_at = out.len();
out.put_u16(0);
if in_body {
let NlriNextHop::Ipv4(next_hop) = next_hop else {
unreachable!("in_body means an IPv4 next hop");
};
out.put_slice(&encoded.bytes[..encoded.at_next_hop]);
PathAttribute::NextHop(NextHopPathAttribute(next_hop))
.to_wire(ctx, &mut out)
.expect("a next hop encodes");
out.put_slice(&encoded.bytes[encoded.at_next_hop..]);
} else {
out.put_slice(&encoded.bytes[..encoded.at_mp_reach]);
PathAttribute::MpReachNlri(MpReachNlriPathAttribute {
family: AfiSafi::unicast(version),
next_hop,
prefixes: routes.to_vec(),
})
.to_wire(ctx, &mut out)
.expect("a chunk that fits encodes");
out.put_slice(&encoded.bytes[encoded.at_mp_reach..]);
}
let attributes_len = out.len() - attributes_at - 2;
out[attributes_at..attributes_at + 2]
.copy_from_slice(&(attributes_len as u16).to_be_bytes());
if in_body {
write_nlri(routes, with_path_id, &mut out);
}
let length = out.len() - length_at + BGP_MARKER.len();
out[length_at..length_at + 2].copy_from_slice(&(length as u16).to_be_bytes());
}
if out.len() >= ENCODED_PIECE {
pieces.push(out.split());
}
}
if !out.is_empty() {
pieces.push(out);
}
pieces
}
/// The UPDATEs that announce `routes`, all of IP version `version`, as
/// `outgoing`: as few as they fit in. `attributes_octets` is how long
/// `outgoing`'s attributes are on this session's wire.
///
/// What [`encoded_announcements`] does, built as messages: kept as what
/// its bytes are checked against.
#[cfg(test)]
pub(crate) fn announcements(
outgoing: &Outgoing,
version: IpVersion,
@@ -484,7 +619,7 @@ mod tests {
use std::net::Ipv4Addr;
use bgp_packet::message::{
CommunitiesPathAttribute, LargeCommunitiesPathAttribute, LargeCommunity,
CommunitiesPathAttribute, LargeCommunitiesPathAttribute, LargeCommunity, Message,
MultiExitDiscPathAttribute, OriginPathAttribute, PathAttributeFlags,
};
use bgp_primitives::ip_prefix::IpPrefix;
@@ -549,6 +684,128 @@ mod tests {
PathAttribute::Communities(CommunitiesPathAttribute(communities.to_vec()))
}
/// What `encoded_announcements` writes is, byte for byte, what the
/// messages `announcements` builds encode to: with the next hop in
/// its own attribute (IPv4) and in MP_REACH_NLRI (IPv6), with and
/// without path identifiers, to two-octet and four-octet AS speakers
/// (so with the attributes around NEXT_HOP, MP_REACH_NLRI and
/// AS4_PATH in order), and over as many messages when the routes do
/// not fit in one.
#[test]
fn encoded_announcements_are_the_messages_on_the_wire() {
use std::net::Ipv6Addr;
for four_octet_asn in [true, false] {
for with_path_id in [false, true] {
for version in [IpVersion::Ipv4, IpVersion::Ipv6] {
let towards = Towards {
four_octet_asn,
..towards(None, false)
};
let next_hop = match version {
IpVersion::Ipv4 => NlriNextHop::Ipv4(Ipv4Addr::new(192, 0, 2, 1)),
IpVersion::Ipv6 => NlriNextHop::Ipv6WithLl {
global: Ipv6Addr::new(0x2001, 0xdb8, 0, 0, 0, 0, 0, 1),
link_local: Ipv6Addr::new(0xfe80, 0, 0, 0, 0, 0, 0, 1),
},
};
let attributes = vec![
PathAttribute::Origin(OriginPathAttribute::IGP),
AsPathAttribute::from_asns(vec![4_200_000_000, 3356, 64700]),
PathAttribute::MultiExitDisc(MultiExitDiscPathAttribute(10)),
communities(&[(3356, 100), (64700, 1)]),
PathAttribute::LargeCommunities(LargeCommunitiesPathAttribute {
communities: vec![LargeCommunity {
global_admin: 64700,
data_1: 1,
data_2: 2,
}],
}),
];
let routes: Vec<Nlri> = (0..30_000u32)
.map(|i| Nlri {
prefix: match version {
IpVersion::Ipv4 => IpPrefix::new(
version,
Ipv4Addr::from(0x0a00_0000 + i * 256).octets(),
24,
),
IpVersion::Ipv6 => IpPrefix::new(
version,
Ipv6Addr::new(0x2001, 0xdb8, i as u16, 0, 0, 0, 0, 0).octets(),
48,
),
}
.unwrap(),
path_id: if with_path_id { i + 1 } else { 0 },
})
.collect();
let ctx = ParserContext {
four_octet_asn,
ibgp: false,
add_path: if with_path_id {
vec![AfiSafi::unicast(version)]
} else {
Vec::new()
},
};
let mut finished = attributes.clone();
finish(&mut finished, &towards);
let encoded = EncodedAttributes::encode(&finished, four_octet_asn).unwrap();
let bytes = encoded_announcements(&encoded, next_hop, version, &routes, &ctx);
let outgoing = Outgoing {
attributes,
next_hop,
};
let messages =
announcements(&outgoing, version, &routes, &towards, encoded.octets());
let expected: Vec<BytesMut> = messages
.into_iter()
.map(|update| {
let mut out = BytesMut::new();
Message::Update(update).to_wire(&ctx, &mut out).unwrap();
out
})
.collect();
let what = format!(
"{version:?}, path ids {with_path_id}, four-octet {four_octet_asn}"
);
assert!(expected.len() > 1, "{what}: more than one message");
for update in &expected {
assert!(update.len() <= 4096, "{what}: a message fits");
}
// In pieces of whole messages, each of about the size
// meant, all but the last at least that.
assert!(bytes.len() > 1, "{what}: more than one piece");
for (n, piece) in bytes.iter().enumerate() {
assert!(
n == bytes.len() - 1
|| (ENCODED_PIECE..ENCODED_PIECE + 4096).contains(&piece.len()),
"{what}: piece {n} is {} octets",
piece.len()
);
let mut at = 0;
while at < piece.len() {
let length =
usize::from(u16::from_be_bytes([piece[at + 16], piece[at + 17]]));
at += length;
}
assert_eq!(at, piece.len(), "{what}: piece {n} is of whole messages");
}
let got: Vec<u8> = bytes
.iter()
.flat_map(|piece| piece.iter().copied())
.collect();
let want: Vec<u8> = expected.iter().flat_map(|m| m.iter().copied()).collect();
assert_eq!(got, want, "{what}: the same octets");
}
}
}
}
struct Case<'a> {
policy: &'a ExportPolicy,
role: Option<PeerRole>,
crates/server/src/policy.rs
+23 -9
View File
@@ -91,6 +91,10 @@ pub struct ImportPolicy {
/// who never asked for it. A route server's routes always have its
/// clients as next hops, and are taken either way.
pub accept_third_party_next_hop: bool,
/// Take a next hop that is a loopback address, or one of our own: for
/// routers that share a host, as those of a test do. Not a setting;
/// only [`ImportPolicy::permissive`] has it.
pub accept_loopback_next_hop: bool,
/// Take routes with the BLACKHOLE community for single addresses, and
/// have the kernel drop the traffic to them (RFC 7999).
pub accept_blackholes: bool,
@@ -199,6 +203,7 @@ impl Default for ImportPolicy {
accept_aspa_invalid: false,
accept_as_sets: false,
accept_third_party_next_hop: false,
accept_loopback_next_hop: false,
accept_blackholes: false,
own_prefixes: Arc::default(),
originated: Arc::default(),
@@ -228,6 +233,7 @@ impl ImportPolicy {
accept_aspa_invalid: true,
accept_as_sets: true,
accept_third_party_next_hop: true,
accept_loopback_next_hop: true,
accept_blackholes: false,
..Self::default()
}
@@ -727,10 +733,10 @@ impl ImportPolicy {
let mut is_peer = false;
for address in named {
let usable = !address.is_unspecified()
&& !address.is_loopback()
&& !address.is_multicast()
&& address != IpAddr::V4(Ipv4Addr::BROADCAST)
&& address != from.local_address;
&& (self.accept_loopback_next_hop
|| (!address.is_loopback() && address != from.local_address));
if !usable {
return Err(Rejection::NextHopUnusable);
}
@@ -1548,13 +1554,7 @@ mod tests {
);
// Nowhere at all, whatever the policy.
let lenient = ImportPolicy::permissive();
for nowhere in [
"0.0.0.0",
"127.0.0.1",
"224.0.0.5",
"255.255.255.255",
"192.0.2.1",
] {
for nowhere in ["0.0.0.0", "224.0.0.5", "255.255.255.255"] {
for policy in [&policy, &lenient] {
assert_eq!(
check(policy, &v4(nowhere), &EXTERNAL),
@@ -1563,6 +1563,20 @@ mod tests {
);
}
}
// This machine: nowhere for traffic to go, unless the policy is
// the permissive one, for routers that share a host.
for ourselves in ["127.0.0.1", "192.0.2.1"] {
assert_eq!(
check(&policy, &v4(ourselves), &EXTERNAL),
Err(Rejection::NextHopUnusable),
"{ourselves}"
);
assert_eq!(
check(&lenient, &v4(ourselves), &EXTERNAL),
Ok(()),
"{ourselves}"
);
}
// Where it may be someone else: said so, a route server, our own AS.
assert_eq!(check(&lenient, &v4("192.0.2.3"), &EXTERNAL), Ok(()));
for from in [
crates/server/src/session.rs
+22 -3
View File
@@ -17,6 +17,7 @@ use std::time::{Duration, SystemTime};
use bgp_packet::message::Message;
use bgp_primitives::ip_prefix::Nlri;
use bgp_rib::table::PeerIndex;
use bytes::BytesMut;
use tokio::sync::{broadcast, mpsc, watch};
use crate::address::{Endpoint, PeerAddress, socket_text};
@@ -199,7 +200,7 @@ pub struct SessionHandle {
pub reflector_client: bool,
/// The peer's BGP Identifier.
pub peer_router_id: Ipv4Addr,
outbound: mpsc::Sender<Message>,
outbound: mpsc::Sender<Outbound>,
/// Where the session is asked to end, if it listens for that.
ending: Option<mpsc::UnboundedSender<SessionError>>,
/// Which of the RPKI's checks the peer's routes are subject to, if
@@ -219,7 +220,7 @@ impl SessionHandle {
local_addr: SocketAddr,
negotiated: &Negotiated,
rib_peer: PeerIndex,
outbound: mpsc::Sender<Message>,
outbound: mpsc::Sender<Outbound>,
) -> Self {
Self {
peer,
@@ -312,10 +313,28 @@ impl SessionHandle {
/// space if the session's writer is behind.
pub async fn send(&self, msg: Message) -> Result<(), SessionClosed> {
self.outbound
.send(msg)
.send(Outbound::Message(msg))
.await
.map_err(|_| SessionClosed(self.peer))
}
/// Sends the peer a piece of one or more UPDATEs that are already
/// encoded, headers and all, as the exporter builds them.
pub async fn send_encoded(&self, piece: BytesMut) -> Result<(), SessionClosed> {
self.outbound
.send(Outbound::Encoded(piece))
.await
.map_err(|_| SessionClosed(self.peer))
}
}
/// What the rest of the server gives a session to send: a message, or a
/// piece of one or more UPDATEs already encoded, headers and all, which
/// goes to the peer as it is once the session is established.
#[derive(Debug)]
pub enum Outbound {
Message(Message),
Encoded(BytesMut),
}
#[derive(Debug, Clone)]
docs/content/docs/getting-started/installation.md
+1 -1
View File
@@ -55,7 +55,7 @@ To build and install the daemon:
bgp --version
```
The command displays the installed version, for example `bgp 0.2.5`.
The command displays the installed version, for example `bgp 0.2.6`.
## Building a Static Executable
docs/content/docs/getting-started/introduction.md
+2 -2
View File
@@ -55,11 +55,11 @@ Ferrix provides the following capabilities:
## Release Status
> [!IMPORTANT]
> Ferrix 0.2.5 is an early release, in production use on one small
> Ferrix 0.2.6 is an early release, in production use on one small
> network. Ferrix is tested against FRR, BIRD, and Routinator in an
> automated interoperability lab and against real Linux kernels.
Ferrix 0.2.5 does not support the following features:
Ferrix 0.2.6 does not support the following features:
- Per-client best path selection for route servers, and ADD-PATH for
advertised routes.
scripts/profile/full-table.sh
+3
View File
@@ -195,6 +195,9 @@ if [ "$collector" = 1 ]; then
last=$now_held
done
echo "collector: $(advertised) routes advertised after $(since "$start") s"
collector_addr=$(status peers '.[] | select(.name == "collector") | .peer_addr')
run curl -sf "http://127.0.0.1:8080/api/tcp?peer=$collector_addr" \
| jq -r '.connection | "collector tcp: busy \(.busy_ms / 1000 | . * 10 | round / 10) s, rwnd-limited \(.limited_by_receive_window_share * 100 | round)%, rtt \(.rtt_ms) ms, cwnd \(.cwnd_segments), sent \(.bytes_sent / 1048576 | round) MiB in \(.segments_out) segments"'
fi
memory_report() {
rss=$(awk '/VmRSS/ { printf "%.0f", $2 / 1024 }' "/proc/$daemon_pid/status")