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>

crates/bench/Cargo.toml

@@ -24,6 +24,7 @@ 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

crates/bench/benches/mesh.rs

@@ -5,26 +5,50 @@
 //! 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.
+//! 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.
 
 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::open::SendReceive;
+use bgp_packet::open::{AddPathEntry, Capability, SendReceive};
 use bgp_server::config::PeerConfig;
+use bgp_server::family::SUPPORTED;
 use bgp_server::policy::{ExportPolicy, ImportPolicy};
+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);
+
+/// 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")
@@ -36,7 +60,13 @@ fn free_port() -> u16 {
 /// 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) -> TestServer {
+async fn router(
+    name: &'static str,
+    id: u8,
+    port: u16,
+    other: u16,
+    with_receiver: bool,
+) -> 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);
@@ -48,17 +78,20 @@ async fn router(name: &'static str, id: u8, port: u16, other: u16) -> TestServer
         c.peers = vec![PeerConfig {
             add_path: Some(SendReceive::Both),
             import: ImportPolicy::permissive(),
-            export: Arc::new(ExportPolicy {
-                learned_routes: true,
-                all_paths: true,
-                ..ExportPolicy::default()
-            }),
+            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)
+            });
+        }
         c.connect_retry = Duration::from_millis(200);
     })
     .await
@@ -81,8 +114,9 @@ async fn feeders(
             .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), 90)
+            .handshake(64512 + peer, Ipv4Addr::new(10, 0, 0, peer as u8), 0)
             .await
             .expect("a session");
         sessions.push((session, stream));
@@ -91,6 +125,13 @@ async fn feeders(
 }
 
 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));
@@ -107,8 +148,8 @@ fn main() {
         .expect("a runtime");
     runtime.block_on(async {
         let (port_a, port_b) = (free_port(), free_port());
-        let a = router("b", 1, port_a, port_b).await;
-        let b = router("a", 2, port_b, port_a).await;
+        let a = router("b", 1, port_a, port_b, true).await;
+        let b = router("a", 2, port_b, port_a, false).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");
 
@@ -188,8 +229,146 @@ fn main() {
             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 {
-            drop(session.await);
+            feeders.push(session.await.expect("a feeder is done"));
         }
+        resync(&a, shape.ipv4_prefixes * peers_a as usize).await;
+        drop(feeders);
     });
 }
+
+/// A new peer takes router `a`'s table, `routes` routes with every path:
+/// how long until the first UPDATE, and until all of it is sent.
+async fn resync(a: &TestServer, 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(a.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 = a
+            .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> = a
+                .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

@@ -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
+            );
+        }
+    }
+}