diff --git a/Cargo.lock b/Cargo.lock index cd0f4afc72..09c7326201 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -4714,6 +4714,7 @@ dependencies = [ "aws-lc-rs", "bytes", "cpu-time", + "criterion", "fastbloom", "getrandom 0.4.3", "hex-literal", diff --git a/quest/m1/quic/coverage/README.md b/quest/m1/quic/coverage/README.md index 5fc0a32b00..116507d341 100644 --- a/quest/m1/quic/coverage/README.md +++ b/quest/m1/quic/coverage/README.md @@ -18,7 +18,7 @@ Decided in the 2026-10-16 plan with the maintainer: (#5711 scheduler, #5690 reliable reset, #5735 receive memory, #5688 shard, #5674 peer limits) merge. #5690 and #5688 merged while this was planned, so they take the merged-first comparison in - [the benchmarks](/quest/m1/quic/coverage/bench.md) and the + [the benchmarks](/rs/moq-quic/benches/common/mod.rs) and the characterization tests are written against b30289127, then rebased. The net lands first so each rewrite can compare against it: benchmarks via `just bench` against the commit that lands them (Criterion only compares cases present at the base), tests by @@ -42,7 +42,6 @@ and the harness defaults to Cubic while production runs BBR3. ## Required -- [Benchmarks](/quest/m1/quic/coverage/bench.md) - Criterion sweeps of stream scheduling, receive and reset, endpoint routing, and the wire codec, on the baseline - [Characterization tests](/quest/m1/quic/coverage/characterize.md) - golden scheduling order, and every uncovered frame, reset, limit, and endpoint path, on today's harness - [Seeded simulation](/quest/m1/quic/coverage/sim.md) - a deterministic, impaired network in `Pair`, BBR3 as the harness default, and property tests over random workloads diff --git a/quest/m1/quic/coverage/bench.md b/quest/m1/quic/coverage/bench.md deleted file mode 100644 index e1b885065f..0000000000 --- a/quest/m1/quic/coverage/bench.md +++ /dev/null @@ -1,38 +0,0 @@ -# [M] Benchmark moq-quic - -## Goal - -`rs/moq-quic/benches/` holds Criterion benchmarks that each QUIC rewrite -compares against the code it replaced. They run in the nightly Criterion -smoke with the other crates. - -## Plan - -Decided 2026-10-16. Land this first in the epic, so the open rewrites can -compare against it. - -`just bench ` only compares cases that exist at ``, so the base -is the commit that lands this suite, not b30289127. Landed before a -rewrite, that commit is still pre-rewrite: the rewrite rebases and runs -`just bench `. For a rewrite that merges first, compare its -parent with this suite cherry-picked onto it against its merge commit with -the same suite, and record those numbers in this PR. - -- Drive a sans-IO `Pair` (or a bench-local equivalent built from the public - API) with no-op packet protection, ported from quinn `perf`'s - `noprotection.rs`, so only protocol CPU is measured. Use only API present - on the baseline and kept by the open rewrites, or the base comparison - cannot build. -- Cases, each swept over both of its axes so a cost that grows with the - table shows up as a slope: - - **Stream send and scheduling**: stream count x priority levels, and - bytes per write. This is what #5711 rewrites. - - **Receive, reassembly, and reset**: out-of-order ingest across many - partial streams, and reset and STOP_SENDING churn. Covers #5690 and - #5735. - - **Endpoint routing**: datagram dispatch over connections x CIDs per - connection. Covers #5688. - - **Wire codec and ACKs**: frame and packet encode and decode, and ACK - processing over large range sets. -- The existing `#[ignore]` print-only benchmarks (`bbr3/cleanup_bench.rs`, - `benchmark_receive_credit`) stay as they are. diff --git a/rs/moq-quic/Cargo.toml b/rs/moq-quic/Cargo.toml index f5a5f77616..c5cacd318a 100644 --- a/rs/moq-quic/Cargo.toml +++ b/rs/moq-quic/Cargo.toml @@ -87,8 +87,11 @@ web-time = "1" [dev-dependencies] arbitrary = { version = "1.5", features = ["derive"] } assert_matches = "1.1" +criterion = { workspace = true } hex-literal = "1" rcgen = { workspace = true, features = ["pem", "ring"] } +# The benches build their own TLS configs, which the crate only re-exports behind `rustls`. +rustls = { version = "0.23.5", default-features = false, features = ["std", "ring"] } stats_alloc = "0.1.10" tracing-subscriber = { version = "0.3.0", default-features = false, features = ["env-filter", "fmt", "ansi", "time", "local-time"] } wasm-bindgen-test = "0.3.45" @@ -103,6 +106,22 @@ name = "fuzz-seeds" path = "examples/fuzz-seeds.rs" required-features = ["fuzz"] +[[bench]] +name = "streams" +harness = false + +[[bench]] +name = "receive" +harness = false + +[[bench]] +name = "endpoint" +harness = false + +[[bench]] +name = "wire" +harness = false + # quinn's workspace lints, kept so upstream code stays clean against them. [lints.rust] elided_lifetimes_in_paths = "warn" diff --git a/rs/moq-quic/benches/common/mod.rs b/rs/moq-quic/benches/common/mod.rs new file mode 100644 index 0000000000..deaed2ff87 --- /dev/null +++ b/rs/moq-quic/benches/common/mod.rs @@ -0,0 +1,908 @@ +//! A sans-IO client/server harness for the benchmarks, built only from the public API so the +//! suite compiles on both sides of a rewrite. +//! +//! Packets are not protected and the congestion window is unlimited, so a benchmark measures +//! protocol CPU rather than AES or a controller's model. Time is virtual: it only advances to +//! the next timer once no datagram is in flight. Cases are timed in thread CPU time, so a busy +//! host descheduling the benchmark does not inflate them. + +use std::{ + any::Any, + collections::{HashMap, VecDeque}, + net::{Ipv6Addr, SocketAddr}, + sync::{Arc, Mutex}, + time::Duration, +}; + +use bytes::{BufMut, Bytes, BytesMut}; +use cpu_time::ThreadTime; +use criterion::{ + Criterion, + measurement::{Measurement, ValueFormatter, WallTime}, +}; +use moq_clock::Instant; +use moq_quic::{ + ClientConfig, ConnectError, Connection, ConnectionHandle, ConnectionId, ConnectionIdGenerator, + DatagramEvent, Endpoint, EndpointConfig, Event, RandomConnectionIdGenerator, ReadError, + ServerConfig, Side, StreamEvent, StreamId, TransportConfig, TransportError, VarInt, + coding::Codec, + congestion::{Controller, ControllerFactory}, + crypto::{ + self, CryptoError, ExportKeyingMaterialError, HeaderKey, KeyPair, Keys, PacketKey, + UnsupportedVersion, + rustls::{QuicClientConfig, QuicServerConfig}, + }, + transport_parameters::TransportParameters, +}; +use rustls::pki_types::{CertificateDer, PrivateKeyDer}; + +/// The most datagrams one `poll_transmit` may batch, as a GSO-capable socket would take. +const MAX_DATAGRAMS: usize = 10; +/// The window the unlimited controller reports: far beyond anything a benchmark puts in flight. +const WINDOW: u64 = 1 << 30; +/// AEAD tag space the protocol reserves in every packet, which the no-op key leaves zeroed. +const TAG_LEN: usize = 16; +/// How far virtual time may jump to a timer while settling before the pair counts as idle. +const SETTLE: Duration = Duration::from_secs(1); +/// Virtual time between forged iterations: well past two PTOs on a zero-latency path, after which +/// a connection forgets the packets it declared lost. +const TICK: Duration = Duration::from_millis(100); +/// How far virtual time may jump to a timer before a stuck exchange fails loudly. +const STALL: Duration = Duration::from_secs(10); + +/// How a peer's application handles each stream the other side opens. +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +pub(crate) enum Sink { + /// Read everything to the end or the reset. + Read, + /// Stop sending as soon as the stream opens. + Stop, +} + +/// What a peer's application has received so far. +#[derive(Clone, Copy, Debug, Default)] +pub(crate) struct Received { + pub(crate) bytes: u64, + pub(crate) finished: u64, + pub(crate) reset: u64, + pub(crate) stopped: u64, + pub(crate) datagrams: u64, +} + +/// Options for [`Pair::new`]. +pub(crate) struct Setup { + /// Client endpoints to connect, each with one connection and its own address. + pub(crate) clients: usize, + /// The client's connection ID length. Zero makes the client advertise an + /// `active_connection_id_limit` of 2, otherwise 5, which bounds the CIDs the server issues. + pub(crate) client_cid_len: usize, +} + +impl Default for Setup { + fn default() -> Self { + Self { + clients: 1, + client_cid_len: 8, + } + } +} + +/// One endpoint and its connections, with an application that drains whatever arrives. +pub(crate) struct Peer { + pub(crate) endpoint: Endpoint, + pub(crate) addr: SocketAddr, + pub(crate) conns: HashMap, + pub(crate) sink: Sink, + pub(crate) received: Received, + /// Wall time spent in [`Peer::process`], to time one side of an exchange. + pub(crate) busy: Duration, + inbound: VecDeque<(SocketAddr, BytesMut)>, + outbound: Vec<(SocketAddr, BytesMut)>, + buf: Vec, +} + +impl Peer { + fn new(endpoint: Endpoint, addr: SocketAddr) -> Self { + Self { + endpoint, + addr, + conns: HashMap::new(), + sink: Sink::Read, + received: Received::default(), + busy: Duration::ZERO, + inbound: VecDeque::new(), + outbound: Vec::new(), + buf: Vec::new(), + } + } + + /// The only connection, for a peer that has exactly one. + pub(crate) fn conn(&mut self) -> &mut Connection { + assert_eq!(self.conns.len(), 1, "peer has more than one connection"); + self.conns.values_mut().next().unwrap() + } + + /// Handles every queued datagram, fires due timers, runs the application, and transmits. + fn process(&mut self, now: Instant) { + let start = ThreadTime::now(); + let mut buf = std::mem::take(&mut self.buf); + + for (remote, data) in std::mem::take(&mut self.inbound) { + buf.clear(); + match self + .endpoint + .handle(now, remote, None, None, data, &mut buf) + { + Some(DatagramEvent::NewConnection(incoming)) => { + buf.clear(); + match self.endpoint.accept(incoming, now, &mut buf, None) { + Ok((ch, conn)) => { + self.conns.insert(ch, conn); + } + Err(err) => panic!("accept failed: {:?}", err.cause), + } + } + Some(DatagramEvent::ConnectionEvent(ch, event)) => { + self.conns + .get_mut(&ch) + .expect("routed to a connection") + .handle_event(event); + } + Some(DatagramEvent::Response(transmit)) => { + split(&mut self.outbound, &transmit, &buf); + } + None => {} + } + } + + for (&ch, conn) in self.conns.iter_mut() { + if conn.poll_timeout().is_some_and(|at| at <= now) { + conn.handle_timeout(now); + } + while let Some(event) = conn.poll_endpoint_events() { + if let Some(event) = self.endpoint.handle_event(ch, event) { + conn.handle_event(event); + } + } + while let Some(event) = conn.poll() { + match event { + Event::ConnectionLost { reason } => panic!("connection lost: {reason}"), + Event::Stream(StreamEvent::Opened { dir }) => { + while let Some(id) = conn.streams().accept(dir) { + drain(conn, id, self.sink, &mut self.received); + } + } + Event::Stream(StreamEvent::Readable { id }) => { + drain(conn, id, self.sink, &mut self.received) + } + // The application answers STOP_SENDING the way a writer does: it gives up. + Event::Stream(StreamEvent::Stopped { id, error_code }) => { + let _ = conn.send_stream(id).reset(error_code); + } + Event::DatagramReceived => { + while conn.datagrams().recv().is_some() { + self.received.datagrams += 1; + } + } + _ => {} + } + } + buf.clear(); + while let Some(transmit) = conn.poll_transmit(now, MAX_DATAGRAMS, &mut buf) { + split(&mut self.outbound, &transmit, &buf); + buf.clear(); + } + } + + self.buf = buf; + self.busy += start.elapsed(); + } + + fn next_timeout(&self) -> Option { + self.conns.values().filter_map(|c| c.poll_timeout()).min() + } +} + +/// Reads or stops one stream, recording what the application saw. +fn drain(conn: &mut Connection, id: StreamId, sink: Sink, received: &mut Received) { + let mut stream = conn.recv_stream(id); + if sink == Sink::Stop { + if stream.stop(VarInt::from_u32(0)).is_ok() { + received.stopped += 1; + } + return; + } + // A stream already stopped, finished, or reset is gone. + let Ok(mut chunks) = stream.read(true) else { + return; + }; + loop { + match chunks.next(usize::MAX) { + Ok(Some(chunk)) => received.bytes += chunk.bytes.len() as u64, + Ok(None) => { + received.finished += 1; + break; + } + Err(ReadError::Blocked) => break, + Err(ReadError::Reset(_)) => { + received.reset += 1; + break; + } + // Unreachable today, but keeps the suite compiling against a rewrite's new variants. + #[allow(unreachable_patterns)] + Err(err) => panic!("read failed: {err}"), + } + } + let _ = chunks.finalize(); +} + +/// Splits a possibly segmented transmit into one datagram each. +fn split(out: &mut Vec<(SocketAddr, BytesMut)>, transmit: &moq_quic::Transmit, buf: &[u8]) { + let data = &buf[..transmit.size]; + let segment = transmit.segment_size.unwrap_or(data.len()); + for datagram in data.chunks(segment) { + out.push((transmit.destination, BytesMut::from(datagram))); + } +} + +/// A server and its connected clients, exchanging datagrams over a zero-latency path. +pub(crate) struct Pair { + pub(crate) now: Instant, + pub(crate) server: Peer, + pub(crate) clients: Vec, + /// Every connection ID the server's endpoint generated, in order. + pub(crate) server_cids: Arc>>, + /// Drops every Nth client datagram, so the server holds data past each hole until the + /// retransmission fills it. Reordering instead would have to stay within the packet number + /// window the sender encoded for, a few dozen packets. + pub(crate) lose_every: Option, + sent: u64, + by_addr: HashMap, +} + +impl Pair { + /// Connects `setup.clients` clients to one server and settles every handshake. + pub(crate) fn new(setup: Setup) -> Self { + let server_cids = Arc::new(Mutex::new(Vec::new())); + let log = server_cids.clone(); + // Fixed seeds make packet number skips and key update timing the same on every run, and + // on both sides of a comparison. + let mut server_endpoint = EndpointConfig::default(); + server_endpoint + .rng_seed(Some([0xff; 32])) + .cid_generator(Arc::new(move || { + Box::new(Recording { + inner: RandomConnectionIdGenerator::new(8), + log: log.clone(), + }) + })); + let server = Endpoint::new(Arc::new(server_endpoint), Some(server_config()), false); + let server = Peer::new(server, SocketAddr::new(Ipv6Addr::LOCALHOST.into(), 4433)); + + let mut pair = Self { + now: Instant::ZERO + Duration::from_secs(3600), + server, + clients: Vec::new(), + server_cids, + lose_every: None, + sent: 0, + by_addr: HashMap::new(), + }; + + let client_config = client_config(); + for index in 0..setup.clients { + let mut config = EndpointConfig::default(); + let cid_len = setup.client_cid_len; + let mut seed = [0; 32]; + seed[..8].copy_from_slice(&(index as u64).to_le_bytes()); + config.rng_seed(Some(seed)).cid_generator(Arc::new(move || { + Box::new(RandomConnectionIdGenerator::new(cid_len)) + })); + let mut endpoint = Endpoint::new(Arc::new(config), None, false); + let port = 10_000 + u16::try_from(index).expect("too many clients"); + let addr = SocketAddr::new(Ipv6Addr::LOCALHOST.into(), port); + let (ch, conn) = endpoint + .connect( + pair.now, + client_config.clone(), + pair.server.addr, + "localhost", + ) + .expect("connect"); + let mut client = Peer::new(endpoint, addr); + client.conns.insert(ch, conn); + pair.by_addr.insert(addr, index); + pair.clients.push(client); + } + + pair.drive_until(|pair| { + pair.server.conns.len() == setup.clients + && pair + .server + .conns + .values() + .chain(pair.clients.iter().flat_map(|c| c.conns.values())) + .all(|c| !c.is_handshaking()) + }); + pair.settle(); + pair + } + + /// The only client's connection. + pub(crate) fn client(&mut self) -> &mut Connection { + assert_eq!(self.clients.len(), 1, "pair has more than one client"); + self.clients[0].conn() + } + + /// Runs every peer once and delivers what they sent, returning whether anything moved. + pub(crate) fn step(&mut self) -> bool { + let mut moved = false; + for client in &mut self.clients { + client.process(self.now); + for (destination, datagram) in client.outbound.drain(..) { + assert_eq!(destination, self.server.addr); + moved = true; + self.sent += 1; + if self.lose_every.is_some_and(|n| self.sent.is_multiple_of(n)) { + continue; + } + self.server.inbound.push_back((client.addr, datagram)); + } + } + + self.server.process(self.now); + for (destination, datagram) in self.server.outbound.drain(..) { + let client = &mut self.clients[self.by_addr[&destination]]; + client.inbound.push_back((self.server.addr, datagram)); + moved = true; + } + moved + } + + /// Exchanges datagrams, jumping to the next timer whenever the path is idle, until `done`. + pub(crate) fn drive_until(&mut self, mut done: impl FnMut(&mut Self) -> bool) { + loop { + let moved = self.step(); + if done(self) { + return; + } + if !moved { + let next = self.next_timeout(); + let received = self.server.received; + let next = next.unwrap_or_else(|| panic!("stalled with no timer: {received:?}")); + assert!( + next <= self.now + STALL, + "stalled with no timer due within {STALL:?}: {received:?}" + ); + self.now = self.now.max(next); + } + } + } + + /// Exchanges datagrams until nothing is in flight and no timer is due soon. + pub(crate) fn settle(&mut self) { + loop { + if self.step() { + continue; + } + match self.next_timeout() { + Some(next) if next <= self.now + SETTLE => self.now = self.now.max(next), + _ => return, + } + } + } + + fn next_timeout(&self) -> Option { + self.clients + .iter() + .chain(std::iter::once(&self.server)) + .filter_map(Peer::next_timeout) + .min() + } +} + +/// Builds client 1-RTT packets by hand, for a server connection whose real client has gone quiet. +/// +/// With no packet or header protection, a short header packet is its plaintext: the first byte, +/// the server's connection ID, a 4-byte packet number, the frames, and zeroed tag space. +pub(crate) struct Forge { + dcid: ConnectionId, + next: u64, +} + +impl Forge { + /// Forges for the pair's only server connection, numbering well past the real client's. + pub(crate) fn new(pair: &Pair) -> Self { + Self { + dcid: pair.server_cids.lock().unwrap()[0], + next: 1 << 20, + } + } + + /// Wraps `frames` in the next packet. + pub(crate) fn packet(&mut self, frames: &[u8]) -> BytesMut { + let mut packet = BytesMut::with_capacity(1 + self.dcid.len() + 4 + frames.len() + TAG_LEN); + packet.put_u8(0x40 | 0x03); + packet.put_slice(&self.dcid); + packet.put_u32(self.next as u32); + packet.put_slice(frames); + packet.put_bytes(0, TAG_LEN); + self.next += 1; + packet + } +} + +/// Appends a DATAGRAM frame carrying `data`. +pub(crate) fn datagram_frame(out: &mut BytesMut, data: &[u8]) { + VarInt::from_u32(0x31).encode(out); + VarInt::from_u64(data.len() as u64).unwrap().encode(out); + out.put_slice(data); +} + +/// Appends an ACK frame for `ranges`, inclusive and sorted from the highest packet down. +pub(crate) fn ack_frame(out: &mut BytesMut, ranges: &[(u64, u64)]) { + let varint = |out: &mut BytesMut, x: u64| VarInt::from_u64(x).unwrap().encode(out); + let (first_start, largest) = ranges[0]; + varint(out, 0x02); + varint(out, largest); + varint(out, 0); // ACK delay + varint(out, ranges.len() as u64 - 1); + varint(out, largest - first_start); + let mut previous = first_start; + for &(start, end) in &ranges[1..] { + varint(out, previous - end - 2); + varint(out, end - start); + previous = start; + } +} + +/// Recovers the full packet numbers of server 1-RTT datagrams sent to a client, in order. +pub(crate) struct ServerPackets { + client_cid_len: usize, + largest: Option, +} + +impl ServerPackets { + pub(crate) fn new(client_cid_len: usize) -> Self { + Self { + client_cid_len, + largest: None, + } + } + + /// The packet number of one short header datagram, per RFC 9000 appendix A.3. + pub(crate) fn number(&mut self, datagram: &[u8]) -> u64 { + assert_eq!(datagram[0] & 0x80, 0, "expected a short header packet"); + let len = usize::from(datagram[0] & 0x03) + 1; + let start = 1 + self.client_cid_len; + let truncated = datagram[start..start + len] + .iter() + .fold(0u64, |pn, &byte| (pn << 8) | u64::from(byte)); + + let expected = self.largest.map_or(0, |pn| pn + 1); + let window = 1u64 << (len * 8); + let candidate = (expected & !(window - 1)) | truncated; + let pn = if candidate + window / 2 <= expected { + candidate + window + } else if candidate > expected + window / 2 && candidate >= window { + candidate - window + } else { + candidate + }; + assert!( + self.largest.is_none_or(|largest| pn > largest), + "packet numbers went backwards" + ); + self.largest = Some(pn); + pn + } +} + +/// Delivers one forged datagram to the pair's only server connection. +pub(crate) fn deliver(pair: &mut Pair, datagram: BytesMut) { + let remote = pair.clients[0].addr; + let server = &mut pair.server; + server.buf.clear(); + match server + .endpoint + .handle(pair.now, remote, None, None, datagram, &mut server.buf) + { + Some(DatagramEvent::ConnectionEvent(ch, event)) => { + server.conns.get_mut(&ch).unwrap().handle_event(event) + } + _ => panic!("forged datagram was not routed to the connection"), + } +} + +/// Lets virtual time pass on the pair's only server connection and fires its due timers. +/// +/// A forged exchange never waits on a timer, so without this, state that expires with time, such +/// as packets remembered as lost, would pile up across iterations. +pub(crate) fn tick(pair: &mut Pair) { + pair.now += TICK; + let conn = pair.server.conn(); + if conn.poll_timeout().is_some_and(|at| at <= pair.now) { + conn.handle_timeout(pair.now); + } +} + +/// Transmits from the pair's only server connection until its datagram send buffer is back to +/// `empty` free bytes, jumping virtual time over pacing delays, and returns the datagrams. +pub(crate) fn flush(pair: &mut Pair, empty: usize) -> Vec { + let mut out = Vec::new(); + let server = &mut pair.server; + let conn = server.conns.values_mut().next().unwrap(); + loop { + server.buf.clear(); + while let Some(transmit) = conn.poll_transmit(pair.now, MAX_DATAGRAMS, &mut server.buf) { + split(&mut out, &transmit, &server.buf); + server.buf.clear(); + } + let free = conn.datagrams().send_buffer_space(); + if free == empty { + return out.into_iter().map(|(_, datagram)| datagram).collect(); + } + // Pacing is the only reason to stop short. + let next = conn.poll_timeout().expect("paced with no timer"); + assert!( + next > pair.now, + "datagrams stuck with {free} of {empty} bytes free" + ); + pair.now = next; + conn.handle_timeout(pair.now); + } +} + +/// Room for the largest burst of datagrams a benchmark queues or receives at once. +const DATAGRAM_BUFFER: usize = 64 << 20; + +fn transport() -> Arc { + let mut transport = TransportConfig::default(); + transport + .congestion_controller_factory(Arc::new(Unlimited)) + // Pinned, so black hole detection cannot shrink packets after a benchmark's losses. + .initial_mtu(1452) + .min_mtu(1452) + .max_idle_timeout(None) + .max_concurrent_uni_streams(VarInt::from_u32(4096)) + .max_concurrent_bidi_streams(VarInt::from_u32(4096)) + .stream_receive_window(VarInt::from_u32(16 << 20)) + .receive_window(VarInt::MAX) + .send_window(1 << 30) + .datagram_receive_buffer_size(Some(DATAGRAM_BUFFER)) + .datagram_send_buffer_size(DATAGRAM_BUFFER); + Arc::new(transport) +} + +fn identity() -> (CertificateDer<'static>, PrivateKeyDer<'static>) { + let key = rcgen::generate_simple_self_signed(vec!["localhost".into()]).unwrap(); + ( + key.cert.der().clone(), + PrivateKeyDer::Pkcs8(key.signing_key.serialize_der().into()), + ) +} + +fn provider() -> Arc { + Arc::new(rustls::crypto::ring::default_provider()) +} + +fn server_config() -> Arc { + let (cert, key) = IDENTITY.clone_key(); + let tls = rustls::ServerConfig::builder_with_provider(provider()) + .with_protocol_versions(&[&rustls::version::TLS13]) + .unwrap() + .with_no_client_auth() + .with_single_cert(vec![cert], key) + .unwrap(); + let crypto = QuicServerConfig::try_from(tls).unwrap(); + let mut config = ServerConfig::with_crypto(Arc::new(NoProtectionServer(Arc::new(crypto)))); + config.transport_config(transport()); + Arc::new(config) +} + +fn client_config() -> ClientConfig { + let mut roots = rustls::RootCertStore::empty(); + roots.add(IDENTITY.0.clone()).unwrap(); + let tls = rustls::ClientConfig::builder_with_provider(provider()) + .with_protocol_versions(&[&rustls::version::TLS13]) + .unwrap() + .with_root_certificates(roots) + .with_no_client_auth(); + let crypto = QuicClientConfig::try_from(tls).unwrap(); + let mut config = ClientConfig::new(Arc::new(NoProtectionClient(Arc::new(crypto)))); + config.transport_config(transport()); + config +} + +/// One certificate shared by every pair, since generating it dominates a small setup. +static IDENTITY: std::sync::LazyLock = std::sync::LazyLock::new(|| { + let (cert, key) = identity(); + Identity(cert, key) +}); + +struct Identity(CertificateDer<'static>, PrivateKeyDer<'static>); + +impl Identity { + fn clone_key(&self) -> (CertificateDer<'static>, PrivateKeyDer<'static>) { + (self.0.clone(), self.1.clone_key()) + } +} + +/// A server connection ID generator that records every ID, so benchmarks can address them. +struct Recording { + inner: RandomConnectionIdGenerator, + log: Arc>>, +} + +impl ConnectionIdGenerator for Recording { + fn generate_cid(&mut self) -> ConnectionId { + let cid = self.inner.generate_cid(); + self.log.lock().unwrap().push(cid); + cid + } + + fn cid_len(&self) -> usize { + self.inner.cid_len() + } + + fn cid_lifetime(&self) -> Option { + self.inner.cid_lifetime() + } +} + +/// A congestion controller that never limits, so benchmarks measure the protocol, not a model. +struct Unlimited; + +impl Controller for Unlimited { + fn on_congestion_event( + &mut self, + _now: Instant, + _sent: Instant, + _is_persistent_congestion: bool, + _is_ecn: bool, + _lost_bytes: u64, + _largest_lost: u64, + _space: moq_quic::SpaceId, + ) { + } + + fn on_mtu_update(&mut self, _new_mtu: u16) {} + + fn window(&self) -> u64 { + WINDOW + } + + fn clone_box(&self) -> Box { + Box::new(Self) + } + + fn initial_window(&self) -> u64 { + WINDOW + } + + fn into_any(self: Box) -> Box { + self + } +} + +impl ControllerFactory for Unlimited { + fn build(self: Arc, _now: Instant, _current_mtu: u16) -> Box { + Box::new(Self) + } +} + +// No-op packet protection, ported from quinn's `perf` `noprotection.rs`: the TLS handshake runs +// for real, but every packet and header key leaves the bytes alone. + +struct NoProtectionClient(Arc); + +impl crypto::ClientConfig for NoProtectionClient { + fn start_session( + self: Arc, + version: u32, + server_name: &str, + params: &TransportParameters, + ) -> Result, ConnectError> { + let inner = self.0.clone().start_session(version, server_name, params)?; + Ok(Box::new(NoProtectionSession(inner))) + } +} + +struct NoProtectionServer(Arc); + +impl crypto::ServerConfig for NoProtectionServer { + fn initial_keys( + &self, + version: u32, + dst_cid: ConnectionId, + ) -> Result { + self.0.initial_keys(version, dst_cid).map(unprotect) + } + + fn retry_tag(&self, version: u32, orig_dst_cid: ConnectionId, packet: &[u8]) -> [u8; 16] { + self.0.retry_tag(version, orig_dst_cid, packet) + } + + fn start_session( + self: Arc, + version: u32, + params: &TransportParameters, + ) -> Box { + Box::new(NoProtectionSession( + self.0.clone().start_session(version, params), + )) + } +} + +struct NoProtectionSession(Box); + +impl crypto::Session for NoProtectionSession { + fn initial_keys(&self, dst_cid: ConnectionId, side: Side) -> Keys { + unprotect(self.0.initial_keys(dst_cid, side)) + } + + fn handshake_data(&self) -> Option> { + self.0.handshake_data() + } + + fn peer_identity(&self) -> Option> { + self.0.peer_identity() + } + + fn early_crypto(&self) -> Option<(Box, Box)> { + let (header, packet) = self.0.early_crypto()?; + Some(( + Box::new(NoHeader(header.sample_size())), + Box::new(NoPacket(packet.tag_len())), + )) + } + + fn early_data_accepted(&self) -> Option { + self.0.early_data_accepted() + } + + fn is_handshaking(&self) -> bool { + self.0.is_handshaking() + } + + fn read_handshake(&mut self, buf: &[u8]) -> Result { + self.0.read_handshake(buf) + } + + fn transport_parameters(&self) -> Result, TransportError> { + self.0.transport_parameters() + } + + fn write_handshake(&mut self, buf: &mut Vec) -> Option { + self.0.write_handshake(buf).map(unprotect) + } + + fn next_1rtt_keys(&mut self) -> Option>> { + let keys = self.0.next_1rtt_keys()?; + Some(KeyPair { + local: Box::new(NoPacket(keys.local.tag_len())), + remote: Box::new(NoPacket(keys.remote.tag_len())), + }) + } + + fn is_valid_retry(&self, orig_dst_cid: ConnectionId, header: &[u8], payload: &[u8]) -> bool { + self.0.is_valid_retry(orig_dst_cid, header, payload) + } + + fn export_keying_material( + &self, + output: &mut [u8], + label: &[u8], + context: &[u8], + ) -> Result<(), ExportKeyingMaterialError> { + self.0.export_keying_material(output, label, context) + } +} + +fn unprotect(keys: Keys) -> Keys { + Keys { + header: KeyPair { + local: Box::new(NoHeader(keys.header.local.sample_size())), + remote: Box::new(NoHeader(keys.header.remote.sample_size())), + }, + packet: KeyPair { + local: Box::new(NoPacket(keys.packet.local.tag_len())), + remote: Box::new(NoPacket(keys.packet.remote.tag_len())), + }, + } +} + +/// Header protection that leaves the header in the clear. +struct NoHeader(usize); + +impl HeaderKey for NoHeader { + fn decrypt(&self, _pn_offset: usize, _packet: &mut [u8]) {} + + fn encrypt(&self, _pn_offset: usize, _packet: &mut [u8]) {} + + fn sample_size(&self) -> usize { + self.0 + } +} + +/// Packet protection that zeroes the tag on the way out and strips it on the way in. +/// +/// Its limits never trigger a key update, which a forged packet could not follow. +struct NoPacket(usize); + +impl PacketKey for NoPacket { + fn encrypt(&self, _packet: u64, buf: &mut [u8], _header_len: usize) { + let tag = buf.len() - self.0; + buf[tag..].fill(0); + } + + fn decrypt( + &self, + _packet: u64, + _header: &[u8], + payload: &mut BytesMut, + ) -> Result<(), CryptoError> { + let len = payload.len().checked_sub(self.0).ok_or(CryptoError)?; + payload.truncate(len); + Ok(()) + } + + fn tag_len(&self) -> usize { + self.0 + } + + fn confidentiality_limit(&self) -> u64 { + u64::MAX + } + + fn integrity_limit(&self) -> u64 { + u64::MAX + } +} + +/// The Criterion config every target runs with: thread CPU time, not elapsed time. +/// +/// The suite is single-threaded, so this thread's CPU time is the protocol's, and a loaded host +/// preempting the benchmark does not count toward it. +pub(crate) fn config() -> Criterion { + Criterion::default().with_measurement(ThreadCpu) +} + +/// Measures this thread's CPU time. +pub(crate) struct ThreadCpu; + +impl Measurement for ThreadCpu { + type Intermediate = ThreadTime; + type Value = Duration; + + fn start(&self) -> ThreadTime { + ThreadTime::now() + } + + fn end(&self, start: ThreadTime) -> Duration { + start.elapsed() + } + + fn add(&self, a: &Duration, b: &Duration) -> Duration { + *a + *b + } + + fn zero(&self) -> Duration { + Duration::ZERO + } + + fn to_f64(&self, value: &Duration) -> f64 { + value.as_nanos() as f64 + } + + fn formatter(&self) -> &dyn ValueFormatter { + // Durations, so they format the way elapsed time does. + static WALL: WallTime = WallTime; + WALL.formatter() + } +} + +/// A buffer of `len` bytes, for payloads whose content does not matter. +pub(crate) fn payload(len: usize) -> Bytes { + Bytes::from(vec![0xab; len]) +} diff --git a/rs/moq-quic/benches/endpoint.rs b/rs/moq-quic/benches/endpoint.rs new file mode 100644 index 0000000000..6f1178812f --- /dev/null +++ b/rs/moq-quic/benches/endpoint.rs @@ -0,0 +1,92 @@ +//! Endpoint routing: dispatching short header datagrams to the connection that owns their CID. +//! +//! Only `Endpoint::handle` is timed; the routed event is dropped before the connection sees it, +//! so the same datagrams replay every iteration. + +use std::hint::black_box; + +use bytes::{BufMut, BytesMut}; +use criterion::{BatchSize, BenchmarkId, Criterion, Throughput, criterion_group, criterion_main}; +use moq_quic::DatagramEvent; + +#[allow(dead_code)] +mod common; + +use common::{Pair, Setup, ThreadCpu}; + +/// Datagrams dispatched per iteration, drawn across every CID the server issued. +const BATCH: usize = 1024; + +/// Connections against CIDs per connection. A client with zero-length CIDs advertises an +/// `active_connection_id_limit` of 2, otherwise 5, and the server issues that many. +fn bench_route(c: &mut Criterion) { + let mut group = c.benchmark_group("endpoint_route"); + group.throughput(Throughput::Elements(BATCH as u64)); + for connections in [16, 256, 4096] { + for cids in [2, 5] { + let mut setup = None; + group.bench_function( + BenchmarkId::new(format!("connections_{connections}"), format!("cids_{cids}")), + |b| { + let (pair, datagrams) = setup.get_or_insert_with(|| route(connections, cids)); + let remote = pair.clients[0].addr; + let now = pair.now; + let mut buf = Vec::new(); + b.iter_batched( + || datagrams.clone(), + |datagrams| { + for datagram in datagrams { + let event = pair + .server + .endpoint + .handle(now, remote, None, None, datagram, &mut buf); + assert!(matches!(event, Some(DatagramEvent::ConnectionEvent(..)))); + black_box(event); + } + }, + BatchSize::SmallInput, + ); + }, + ); + } + } + group.finish(); +} + +/// Connects the clients and forges one 1-RTT datagram per issued CID, in a scattered order. +fn route(connections: usize, cids: usize) -> (Pair, Vec) { + let pair = Pair::new(Setup { + clients: connections, + client_cid_len: if cids == 2 { 0 } else { 8 }, + }); + let issued = pair.server_cids.lock().unwrap().clone(); + assert_eq!( + issued.len(), + connections * cids, + "server issued an unexpected number of CIDs" + ); + + // A fixed-seed LCG, so both sides of a comparison route the same scattered sequence. + let mut state = 0x2545_f491_4f6c_dd1du64; + let datagrams = (0..BATCH) + .map(|_| { + state = state + .wrapping_mul(6364136223846793005) + .wrapping_add(1442695040888963407); + let cid = issued[(state >> 33) as usize % issued.len()]; + let mut datagram = BytesMut::with_capacity(1200); + datagram.put_u8(0x40 | 0x03); + datagram.put_slice(&cid); + datagram.resize(1200, 0); + datagram + }) + .collect(); + (pair, datagrams) +} + +criterion_group! { + name = benches; + config = common::config(); + targets = bench_route +} +criterion_main!(benches); diff --git a/rs/moq-quic/benches/receive.rs b/rs/moq-quic/benches/receive.rs new file mode 100644 index 0000000000..1cb84fc1a4 --- /dev/null +++ b/rs/moq-quic/benches/receive.rs @@ -0,0 +1,140 @@ +//! Receive, reassembly, and reset: what a server pays to ingest, and to abandon, many streams. + +use std::time::Duration; + +use criterion::{BenchmarkId, Criterion, Throughput, criterion_group, criterion_main}; +use moq_quic::{Dir, VarInt}; + +#[allow(dead_code)] +mod common; + +use common::{Pair, Setup, Sink, ThreadCpu, payload}; + +/// Stream data per packet, so a stream of `n` packets spans about `n` datagrams. +const PACKET: usize = 1100; + +/// Out-of-order ingest: every eighth client datagram is lost, so the server buffers data past a +/// hole in many partial streams at once until each retransmission arrives. +/// +/// Only the server's side of the exchange is timed. +fn bench_loss(c: &mut Criterion) { + let mut group = c.benchmark_group("receive_loss"); + for streams in [1, 16, 256] { + for packets in [2, 16] { + let mut pair = None; + group.throughput(Throughput::Bytes((streams * packets * PACKET) as u64)); + group.bench_function( + BenchmarkId::new(format!("streams_{streams}"), format!("packets_{packets}")), + |b| { + let pair = pair.get_or_insert_with(|| { + let mut pair = Pair::new(Setup::default()); + pair.lose_every = Some(8); + pair + }); + let data = payload(packets * PACKET); + b.iter_custom(|iters| { + let mut busy = Duration::ZERO; + for _ in 0..iters { + pair.server.busy = Duration::ZERO; + let target = pair.server.received.finished + streams as u64; + let conn = pair.client(); + for _ in 0..streams { + let id = conn.streams().open(Dir::Uni).expect("stream limit"); + let mut stream = conn.send_stream(id); + assert_eq!(stream.write(&data).unwrap(), data.len()); + stream.finish().unwrap(); + } + pair.drive_until(|pair| pair.server.received.finished >= target); + pair.settle(); + busy += pair.server.busy; + } + busy + }); + }, + ); + } + } + group.finish(); +} + +/// Sender resets: streams carry `bytes` each, of which one burst leaves before every stream is +/// reset, and the exchange runs until the server has seen every reset. +fn bench_reset(c: &mut Criterion) { + let mut group = c.benchmark_group("receive_reset"); + for streams in [16, 256, 1024] { + for bytes in [0, 16384] { + let mut pair = None; + group.throughput(Throughput::Elements(streams as u64)); + group.bench_function( + BenchmarkId::new(format!("streams_{streams}"), format!("bytes_{bytes}")), + |b| { + let pair = pair.get_or_insert_with(|| Pair::new(Setup::default())); + let data = payload(bytes); + b.iter(|| { + let target = pair.server.received.reset + streams as u64; + let conn = pair.client(); + let ids: Vec<_> = (0..streams) + .map(|_| { + let id = conn.streams().open(Dir::Uni).expect("stream limit"); + assert_eq!(conn.send_stream(id).write(&data).unwrap(), bytes); + id + }) + .collect(); + pair.step(); + let conn = pair.client(); + for id in ids { + conn.send_stream(id).reset(VarInt::from_u32(1)).unwrap(); + } + pair.drive_until(|pair| pair.server.received.reset >= target); + pair.settle(); + }); + }, + ); + } + } + group.finish(); +} + +/// Receiver stops: the server sends STOP_SENDING for every stream as it opens, and the exchange +/// runs until the client has reset and forgotten them all. +fn bench_stop(c: &mut Criterion) { + let mut group = c.benchmark_group("receive_stop"); + for streams in [16, 256, 1024] { + for bytes in [1100, 16384] { + let mut pair = None; + group.throughput(Throughput::Elements(streams as u64)); + group.bench_function( + BenchmarkId::new(format!("streams_{streams}"), format!("bytes_{bytes}")), + |b| { + let pair = pair.get_or_insert_with(|| { + let mut pair = Pair::new(Setup::default()); + pair.server.sink = Sink::Stop; + pair + }); + let data = payload(bytes); + b.iter(|| { + let target = pair.server.received.stopped + streams as u64; + let conn = pair.client(); + for _ in 0..streams { + let id = conn.streams().open(Dir::Uni).expect("stream limit"); + assert_eq!(conn.send_stream(id).write(&data).unwrap(), bytes); + } + pair.drive_until(|pair| { + pair.server.received.stopped >= target + && pair.client().streams().send_streams() == 0 + }); + pair.settle(); + }); + }, + ); + } + } + group.finish(); +} + +criterion_group! { + name = benches; + config = common::config(); + targets = bench_loss, bench_reset, bench_stop +} +criterion_main!(benches); diff --git a/rs/moq-quic/benches/streams.rs b/rs/moq-quic/benches/streams.rs new file mode 100644 index 0000000000..35fc49536f --- /dev/null +++ b/rs/moq-quic/benches/streams.rs @@ -0,0 +1,87 @@ +//! Stream send and scheduling: one client writes many streams to a server that reads them all. +//! +//! Each iteration opens fresh unidirectional streams on a long-lived connection, writes and +//! finishes them, then exchanges packets until the server has read every stream to its end and +//! the path is idle, so the cost covers the scheduler, packetization, the receiver, and the ACKs +//! that free it all. + +use criterion::{BenchmarkId, Criterion, Throughput, criterion_group, criterion_main}; +use moq_quic::Dir; + +#[allow(dead_code)] +mod common; + +use common::{Pair, Setup, ThreadCpu, payload}; + +/// Opens `streams` streams across `priorities` levels, writes `writes` writes of `size` bytes to +/// each, and drives until the server has read them all. +fn send(pair: &mut Pair, streams: usize, priorities: usize, writes: usize, size: usize) { + let data = payload(size); + let target = pair.server.received.finished + streams as u64; + let conn = pair.client(); + for index in 0..streams { + let id = conn.streams().open(Dir::Uni).expect("stream limit"); + let mut stream = conn.send_stream(id); + // `as _` follows the priority type, which a scheduler rewrite widens. + stream.set_priority((index % priorities) as _).unwrap(); + for _ in 0..writes { + assert_eq!(stream.write(&data).unwrap(), size, "write was cut short"); + } + stream.finish().unwrap(); + } + pair.drive_until(|pair| pair.server.received.finished >= target); + pair.settle(); +} + +/// Stream count against priority levels, with a few packets per stream so they interleave. +fn bench_priority(c: &mut Criterion) { + const BYTES: usize = 4096; + let mut group = c.benchmark_group("send_priority"); + for streams in [1, 16, 256, 1024] { + for priorities in [1, 8, 64] { + if priorities > streams { + continue; + } + let mut pair = None; + group.throughput(Throughput::Bytes((streams * BYTES) as u64)); + group.bench_function( + BenchmarkId::new( + format!("streams_{streams}"), + format!("priorities_{priorities}"), + ), + |b| { + let pair = pair.get_or_insert_with(|| Pair::new(Setup::default())); + b.iter(|| send(pair, streams, priorities, 1, BYTES)) + }, + ); + } + } + group.finish(); +} + +/// Bytes per write against stream count, writing a fixed 64 KiB to each stream. +fn bench_write_size(c: &mut Criterion) { + const BYTES: usize = 64 << 10; + let mut group = c.benchmark_group("send_write_size"); + for streams in [1, 64] { + for size in [64, 1024, 16384] { + let mut pair = None; + group.throughput(Throughput::Bytes((streams * BYTES) as u64)); + group.bench_function( + BenchmarkId::new(format!("streams_{streams}"), format!("bytes_{size}")), + |b| { + let pair = pair.get_or_insert_with(|| Pair::new(Setup::default())); + b.iter(|| send(pair, streams, 1, BYTES / size, size)) + }, + ); + } + } + group.finish(); +} + +criterion_group! { + name = benches; + config = common::config(); + targets = bench_priority, bench_write_size +} +criterion_main!(benches); diff --git a/rs/moq-quic/benches/wire.rs b/rs/moq-quic/benches/wire.rs new file mode 100644 index 0000000000..cf167d371b --- /dev/null +++ b/rs/moq-quic/benches/wire.rs @@ -0,0 +1,207 @@ +//! Wire codec and ACKs: packet and frame encode, decode, and ACK processing on one server +//! connection whose real client goes quiet after the handshake, so every client packet is forged. +//! +//! DATAGRAM frames carry the payload because they are never retransmitted: whatever an +//! iteration leaves unacknowledged is simply lost, and the next iteration starts clean. + +use std::time::Duration; + +use cpu_time::ThreadTime; + +use bytes::BytesMut; +use criterion::{BenchmarkId, Criterion, Throughput, criterion_group, criterion_main}; + +#[allow(dead_code)] +mod common; + +use common::{ + Forge, Pair, ServerPackets, Setup, ThreadCpu, ack_frame, datagram_frame, deliver, flush, + payload, tick, +}; + +/// Frame bytes that fit one 1452-byte datagram after the header and tag, with room to spare +/// for only one more small frame, so N frames of `ROOM / N` bytes pack exactly N per packet. +const ROOM: usize = 1360; + +/// A DATAGRAM payload that makes a frame of `ROOM / frames` bytes: a type byte and a two-byte +/// length precede it. +fn datagram(frames: usize) -> bytes::Bytes { + payload(ROOM / frames - 3) +} + +/// A quiet pair, a forger for its client, and a parser for its server's packet numbers. +fn quiet() -> (Pair, Forge, ServerPackets) { + let setup = Setup::default(); + let parser = ServerPackets::new(setup.client_cid_len); + let pair = Pair::new(setup); + let forge = Forge::new(&pair); + (pair, forge, parser) +} + +/// Queues enough DATAGRAM frames on the server for about `packets` packets of `frames` each, +/// returning the free send buffer space once they are all sent. +/// +/// A pacing burst can end mid-packet, so a flight runs a packet or two over. +fn queue(pair: &mut Pair, packets: usize, frames: usize) -> usize { + let data = datagram(frames); + let conn = pair.server.conn(); + // A forged packet the server rejected would close it quietly. + assert!(!conn.is_closed(), "the server closed the connection"); + let mut datagrams = conn.datagrams(); + let empty = datagrams.send_buffer_space(); + for _ in 0..packets * frames { + datagrams.send(data.clone(), false).unwrap(); + } + empty +} + +/// The packet numbers of the server's `datagrams`. +fn numbers(parser: &mut ServerPackets, datagrams: &[BytesMut]) -> Vec { + datagrams.iter().map(|d| parser.number(d)).collect() +} + +/// Acknowledges the packets in `pns` in one forged ACK, splitting it into about `ranges` ranges +/// by leaving evenly spaced packets out, which the server then declares lost. +fn ack(forge: &mut Forge, pns: &[u64], ranges: usize) -> BytesMut { + let skip = pns.len() / ranges; + let acked = pns + .iter() + .enumerate() + .filter(|&(index, _)| ranges == 1 || index == 0 || index % skip != 0) + .map(|(_, &pn)| pn); + + // Contiguous runs, highest first; packet numbers the server skipped split runs too. + let mut runs: Vec<(u64, u64)> = Vec::new(); + for pn in acked { + match runs.last_mut() { + Some((_, end)) if *end + 1 == pn => *end = pn, + _ => runs.push((pn, pn)), + } + } + runs.reverse(); + + let mut frames = BytesMut::new(); + ack_frame(&mut frames, &runs); + forge.packet(&frames) +} + +/// Packet encode: the server packs queued DATAGRAM frames, frames per packet against packets. +/// Only `poll_transmit` is timed; a forged ACK then clears the flight. +fn bench_encode(c: &mut Criterion) { + let mut group = c.benchmark_group("wire_encode"); + for frames in [1, 16] { + for packets in [64, 1024] { + let mut state = None; + group.throughput(Throughput::Elements(packets as u64)); + group.bench_function( + BenchmarkId::new(format!("frames_{frames}"), format!("packets_{packets}")), + |b| { + let (pair, forge, parser) = state.get_or_insert_with(quiet); + b.iter_custom(|iters| { + let mut busy = Duration::ZERO; + for _ in 0..iters { + let empty = queue(pair, packets, frames); + let start = ThreadTime::now(); + let datagrams = flush(pair, empty); + busy += start.elapsed(); + + let sent = numbers(parser, &datagrams); + deliver(pair, ack(forge, &sent, 1)); + tick(pair); + } + busy + }); + }, + ); + } + } + group.finish(); +} + +/// Packet and frame decode: the server ingests forged packets of DATAGRAM frames, frames per +/// packet against packets. Only the ingest is timed. +fn bench_decode(c: &mut Criterion) { + let mut group = c.benchmark_group("wire_decode"); + for frames in [1, 16] { + for packets in [64, 1024] { + let mut state = None; + group.throughput(Throughput::Elements(packets as u64)); + group.bench_function( + BenchmarkId::new(format!("frames_{frames}"), format!("packets_{packets}")), + |b| { + let (pair, forge, _) = state.get_or_insert_with(quiet); + let data = datagram(frames); + let mut body = BytesMut::new(); + for _ in 0..frames { + datagram_frame(&mut body, &data); + } + b.iter_custom(|iters| { + let mut busy = Duration::ZERO; + for _ in 0..iters { + let forged: Vec<_> = + (0..packets).map(|_| forge.packet(&body)).collect(); + let start = ThreadTime::now(); + for packet in forged { + deliver(pair, packet); + } + busy += start.elapsed(); + + let conn = pair.server.conn(); + let mut received = 0; + while conn.datagrams().recv().is_some() { + received += 1; + } + assert_eq!(received, packets * frames, "server dropped datagrams"); + + // Send the ACKs, which no forged packet acknowledges back. + let empty = conn.datagrams().send_buffer_space(); + flush(pair, empty); + tick(pair); + } + busy + }); + }, + ); + } + } + group.finish(); +} + +/// ACK processing: one forged ACK covers a flight of server packets in a number of ranges, +/// in-flight packets against ranges. Only the ACK's ingest is timed. +fn bench_ack(c: &mut Criterion) { + let mut group = c.benchmark_group("wire_ack"); + for packets in [256, 2048] { + for ranges in [1, 16, 128] { + let mut state = None; + group.throughput(Throughput::Elements(packets as u64)); + group.bench_function( + BenchmarkId::new(format!("packets_{packets}"), format!("ranges_{ranges}")), + |b| { + let (pair, forge, parser) = state.get_or_insert_with(quiet); + b.iter_custom(|iters| { + let mut busy = Duration::ZERO; + for _ in 0..iters { + let empty = queue(pair, packets, 1); + let sent = numbers(parser, &flush(pair, empty)); + let ack = ack(forge, &sent, ranges); + let start = ThreadTime::now(); + deliver(pair, ack); + busy += start.elapsed(); + tick(pair); + } + busy + }); + }, + ); + } + } + group.finish(); +} + +criterion_group! { + name = benches; + config = common::config(); + targets = bench_encode, bench_decode, bench_ack +} +criterion_main!(benches); diff --git a/sh/gh/nightly.sh b/sh/gh/nightly.sh index cee489c822..43f3bb4659 100755 --- a/sh/gh/nightly.sh +++ b/sh/gh/nightly.sh @@ -34,10 +34,11 @@ case "${1:-}" in # One invocation unifies this release build instead of rebuilding shared crates # independently for each target. Keep the explicit targets: '*' would add coverage. run cargo bench --locked --no-fail-fast -p moq-relay -p moq-stats -p moq-mux -p moq-rtmp -p moq-uring -p moq-pattern \ - --features moq-stats/bench \ + -p moq-quic --features moq-stats/bench \ --bench http_idle --bench auth --bench producer_tick --bench decode --bench aggregate \ --bench catalog --bench ts_schedule --bench export --bench containers --bench matching \ - --bench chunk_interleaving --bench worker_turns -- --test + --bench chunk_interleaving --bench worker_turns \ + --bench streams --bench receive --bench endpoint --bench wire -- --test run just rs test -p moq-archive -p moq-mux -p moq-hls --no-fail-fast \ -E 'test(benchmark_timeline_fetch_fanout) | test(benchmark_damage_pid_fanout) | test(benchmark_pes_staging) | test(benchmark_reference_handoff_fanout)' \ --run-ignored only --success-output immediate diff --git a/sh/gh/nightly.test.py b/sh/gh/nightly.test.py index d2a96d1d8f..2fbcb8eae6 100644 --- a/sh/gh/nightly.test.py +++ b/sh/gh/nightly.test.py @@ -51,6 +51,8 @@ def test_release_benchmarks_share_one_invocation(self): self.assertIn("-p moq-pattern", shared[0]) self.assertIn("--bench matching", shared[0]) self.assertIn("--bench containers", shared[0]) + self.assertIn("-p moq-quic", shared[0]) + self.assertIn("--bench wire", shared[0]) self.assertIn("just rs bench-gst-catalog -- --quick", calls) self.assertTrue(calls[-1].startswith("just rs test -p moq-cli"))