diff --git a/Cargo.lock b/Cargo.lock index 63ef8e406f..2cdad8410a 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -4390,6 +4390,7 @@ version = "0.4.17" dependencies = [ "anyhow", "bytes", + "criterion", "futures", "gst-plugin-version-helper", "gstreamer", @@ -4470,6 +4471,7 @@ dependencies = [ "base64 0.23.1", "bytes", "crc", + "criterion", "futures", "h264-parser", "hang", @@ -4613,6 +4615,7 @@ dependencies = [ name = "moq-pattern" version = "0.1.2" dependencies = [ + "criterion", "moq-pattern", "serde", "serde_json", @@ -4705,6 +4708,7 @@ dependencies = [ "anyhow", "byteorder", "bytes", + "criterion", "futures", "hang", "hmac 0.13.0", diff --git a/js/json/bench/codec.ts b/js/json/bench/codec.ts new file mode 100644 index 0000000000..1abc04c5a9 --- /dev/null +++ b/js/json/bench/codec.ts @@ -0,0 +1,34 @@ +/** Sweep snapshot codec cost over document size and changed keys. */ +import { Decoder } from "../src/snapshot/decoder.ts"; +import { Encoder } from "../src/snapshot/encoder.ts"; + +const updates = 256; +console.log("keys\tchanged\tcompression\tms/update"); +for (const keys of [1, 10, 100, 1000]) { + for (const changed of new Set([1, Math.min(10, keys)])) { + for (const compression of ["none", "deflate"] as const) { + const samples: number[] = []; + for (let sample = 0; sample < 3; sample++) { + const config = { deltaRatio: 100, compression }; + const encoder = new Encoder>(config); + const decoder = new Decoder>(config); + let value = Object.fromEntries(Array.from({ length: keys }, (_, n) => [`k${n}`, 0])); + const start = performance.now(); + for (let n = 1; n <= updates; n++) { + value = { ...value }; + for (let key = 0; key < changed; key++) value[`k${key}`] = n; + const frame = encoder.update(value); + if (frame) { + if (frame.keyframe) decoder.snapshot(frame.payload); + else decoder.delta(frame.payload); + frame.commit(); + } + if (decoder.decode()?.k0 !== n) throw new Error("codec benchmark lost an update"); + } + samples.push((performance.now() - start) / updates); + } + samples.sort((a, b) => a - b); + console.log(`${keys}\t${changed}\t${compression}\t${samples[1]?.toFixed(6)}`); + } + } +} diff --git a/js/justfile b/js/justfile index 65944996eb..03fadd9333 100644 --- a/js/justfile +++ b/js/justfile @@ -48,6 +48,11 @@ bench name: bun install --frozen-lockfile bun net/bench/{{ name }}.ts +# Time Effect create and close, which pays for the creation stack its diagnostics print. +signals-bench: + bun install --frozen-lockfile + bun signals/bench/effect.ts + # Build all packages. build: bun run --filter='*' build diff --git a/js/net/bench/frames.ts b/js/net/bench/frames.ts index bff942ce72..eb2ac881bf 100644 --- a/js/net/bench/frames.ts +++ b/js/net/bench/frames.ts @@ -13,6 +13,9 @@ import * as Path from "../src/path.ts"; import { Reader, Stream } from "../src/stream.ts"; import * as Varint from "../src/varint.ts"; +// Keep protocol diagnostics out of the measured CSV on stdout. +console.debug = console.error; + const frameSizes = [16, 100, 1000]; const chunkSizes = [1200, 16 * 1024, 64 * 1024]; const framesPerGroup = 50; diff --git a/js/signals/bench/effect.ts b/js/signals/bench/effect.ts new file mode 100644 index 0000000000..ef7738e451 --- /dev/null +++ b/js/signals/bench/effect.ts @@ -0,0 +1,59 @@ +/** Time creating and closing an Effect, bare and with a tracked signal, against a bare `new Error()`. */ +import { Effect, Signal } from "../src/index.ts"; + +const runs = 20_000; +const reps = 9; +const signal = new Signal(0); +let checksum = 0; +let tracked = 0; + +const ops: [string, () => void | Promise][] = [ + // The creation stack each Effect captures for its diagnostics; the floor for the rows below. + [ + "new-error", + () => { + checksum += new Error("created here:").message.length; + }, + ], + [ + "create-close", + () => { + new Effect().close(); + checksum++; + }, + ], + [ + "create-get-close", + async () => { + const effect = new Effect((effect) => { + tracked += effect.get(signal) + 1; + }); + // The callback runs in a microtask, so yield once to let it track the signal before closing. + await Promise.resolve(); + effect.close(); + }, + ], +]; + +// Nanoseconds per op for the fastest of several reps, since a slower one measured the machine. +async function time(body: () => void | Promise): Promise { + let best = Number.POSITIVE_INFINITY; + for (let rep = 0; rep < reps; rep++) { + const start = performance.now(); + for (let i = 0; i < runs; i++) { + // Awaiting a sync row would add a microtask hop it doesn't pay. + const pending = body(); + if (pending instanceof Promise) await pending; + } + best = Math.min(best, ((performance.now() - start) * 1e6) / runs); + } + return best; +} + +// Time every op once first and discard it, so the JIT has seen all of them and the first row isn't timing warmup. +for (const [, body] of ops) await time(body); + +console.log("op,ns_per_op"); +for (const [op, body] of ops) console.log(`${op},${(await time(body)).toFixed(0)}`); +if (checksum === 0) throw new Error("benchmark did no work"); +if (tracked === 0) throw new Error("create-get-close never ran its effect"); diff --git a/rs/kio/Cargo.toml b/rs/kio/Cargo.toml index 0bee684259..0c09f113b6 100644 --- a/rs/kio/Cargo.toml +++ b/rs/kio/Cargo.toml @@ -39,5 +39,9 @@ harness = false name = "channel" harness = false +[[bench]] +name = "contended" +harness = false + [lints.rust] unexpected_cfgs = { level = "warn", check-cfg = ['cfg(loom)'] } diff --git a/rs/kio/benches/contended.rs b/rs/kio/benches/contended.rs new file mode 100644 index 0000000000..d54ae34fea --- /dev/null +++ b/rs/kio/benches/contended.rs @@ -0,0 +1,334 @@ +//! Cross-core contention on one kio channel. +//! +//! `channel.rs` drives every consumer from one thread, so the channel mutex never +//! changes hands between cores. Here one producer (the bench thread) feeds consumers +//! spread round-robin over 1, 4, or 16 worker threads, swept over the consumer count, +//! so a cost that grows with fan-out or with cores shows up as a slope. Throughput is +//! per consumer: a flat time per element across threads means no contention. +//! +//! - `contended_notify`: the producer writes a `u64`, waking every consumer; each +//! observes the new value and parks again, as in `channel_notify_cycle`. +//! - `contended_repoll`: consumers re-poll a value that has not changed, as when a +//! deadline or a second channel wakes their task. Workers run free, with no +//! producer and no per-pass sync. +//! - `contended_log`: a `GroupState`-shaped log. The producer appends a frame to a +//! `VecDeque`; each consumer reads its cursor's frame, cloning it out under the +//! lock, and parks again. +//! +//! The notify and log cycles include releasing and joining the workers each +//! iteration, a fixed cost that matters only at a few consumers per thread. +//! +//! Run with `cargo bench -p kio --bench contended`. + +use std::{ + collections::VecDeque, + hint::black_box, + sync::{ + Arc, + atomic::{AtomicU64, AtomicUsize, Ordering}, + }, + task::{Poll, Waker}, + thread, + time::{Duration, Instant}, +}; + +use criterion::{BenchmarkId, Criterion, Throughput, criterion_group, criterion_main}; +use kio::{Consumer, Producer, Waiter}; + +/// Worker threads the consumers are spread over. +const THREADS: [usize; 3] = [1, 4, 16]; + +/// Consumers on the one channel, from a lone reader up to a large broadcast. +const CONSUMERS: [usize; 4] = [1, 16, 256, 4096]; + +/// Frames the log keeps before evicting its oldest, like a group's cache. +const LOG_FRAMES: usize = 64; + +/// One pass of a worker over its consumers. +type Work = Box; + +/// Worker threads that each run a number of passes when the driver releases a round. +struct Crew { + control: Arc, + threads: Vec>, +} + +#[derive(Default)] +struct Control { + /// Rounds released so far, or [`STOP`]. + round: AtomicU64, + /// Passes each worker runs in the current round. + passes: AtomicU64, + /// Workers done with the current round. + done: AtomicUsize, +} + +const STOP: u64 = u64::MAX; + +impl Crew { + fn new(work: Vec) -> Self { + let control = Arc::new(Control::default()); + let threads = work + .into_iter() + .map(|mut work| { + let control = control.clone(); + thread::spawn(move || { + let mut seen = 0; + loop { + seen = spin(|| { + let round = control.round.load(Ordering::Acquire); + (round != seen).then_some(round) + }); + if seen == STOP { + return; + } + for _ in 0..control.passes.load(Ordering::Relaxed) { + work(); + } + control.done.fetch_add(1, Ordering::AcqRel); + } + }) + }) + .collect(); + Self { control, threads } + } + + /// Run `passes` passes on every worker and wait for all of them. + fn run(&self, passes: u64) { + self.control.done.store(0, Ordering::Relaxed); + self.control.passes.store(passes, Ordering::Relaxed); + self.control.round.fetch_add(1, Ordering::Release); + let threads = self.threads.len(); + spin(|| { + // A panicked worker never reports done, so fail instead of spinning forever. + assert!(!self.threads.iter().any(|t| t.is_finished()), "worker panicked"); + (self.control.done.load(Ordering::Acquire) == threads).then_some(()) + }); + } +} + +impl Drop for Crew { + fn drop(&mut self) { + self.control.round.store(STOP, Ordering::Release); + for thread in self.threads.drain(..) { + let result = thread.join(); + // Already unwinding from `run`'s check: a second panic would abort. + if !thread::panicking() { + result.expect("worker panicked"); + } + } + } +} + +/// Busy-wait for `f`, yielding once spinning has gone on long enough that the +/// machine may be oversubscribed. +fn spin(mut f: impl FnMut() -> Option) -> R { + let mut spins = 0u32; + loop { + if let Some(r) = f() { + return r; + } + if spins < 1_000 { + spins += 1; + std::hint::spin_loop(); + } else { + thread::yield_now(); + } + } +} + +/// One consumer, its task's waiter, and the last value (or log index) it read. +struct Reader { + consumer: Consumer, + waiter: Waiter, + seen: u64, +} + +/// `consumers` readers parked on `producer`, dealt round-robin into `threads` sets. +fn readers(producer: &Producer, consumers: usize, threads: usize) -> Vec>> { + let mut sets: Vec>> = (0..threads).map(|_| Vec::new()).collect(); + for i in 0..consumers { + let reader = Reader { + consumer: producer.consume(), + waiter: Waiter::new(Waker::noop().clone()), + seen: 0, + }; + let parked = reader.consumer.poll(&reader.waiter, |_| Poll::<()>::Pending); + assert!(parked.is_pending()); + drop(parked); + sets[i % threads].push(reader); + } + sets +} + +/// Every swept (threads, consumers) pair with at least one consumer per thread. +fn sweep() -> impl Iterator { + THREADS + .into_iter() + .flat_map(|threads| CONSUMERS.into_iter().map(move |consumers| (threads, consumers))) + .filter(|(threads, consumers)| consumers >= threads) +} + +fn bench_notify(c: &mut Criterion) { + let mut group = c.benchmark_group("contended_notify"); + for (threads, consumers) in sweep() { + group.throughput(Throughput::Elements(consumers as u64)); + let id = BenchmarkId::new(format!("threads_{threads}"), consumers); + group.bench_function(id, |b| { + let producer = Producer::new(0u64); + let work = readers(&producer, consumers, threads) + .into_iter() + .map(|mut set| -> Work { + Box::new(move || { + for reader in &mut set { + let seen = reader.seen; + let changed = |state: &kio::Ref<'_, u64>| { + if **state != seen { + Poll::Ready(**state) + } else { + Poll::Pending + } + }; + let Poll::Ready(Ok(value)) = reader.consumer.poll(&reader.waiter, changed) else { + panic!("a published value must be ready"); + }; + reader.seen = black_box(value); + let changed = |state: &kio::Ref<'_, u64>| { + if **state != value { + Poll::Ready(()) + } else { + Poll::Pending + } + }; + assert!(reader.consumer.poll(&reader.waiter, changed).is_pending()); + } + }) + }) + .collect(); + let crew = Crew::new(work); + + let mut next = 0; + b.iter_custom(|iters| { + let start = Instant::now(); + for _ in 0..iters { + next += 1; + *producer.write().ok().expect("open") = next; + crew.run(1); + } + start.elapsed() + }); + }); + } + group.finish(); +} + +fn bench_repoll(c: &mut Criterion) { + let mut group = c.benchmark_group("contended_repoll"); + for (threads, consumers) in sweep() { + group.throughput(Throughput::Elements(consumers as u64)); + let id = BenchmarkId::new(format!("threads_{threads}"), consumers); + group.bench_function(id, |b| { + let producer = Producer::new(0u64); + let work = readers(&producer, consumers, threads) + .into_iter() + .map(|set| -> Work { + Box::new(move || { + for reader in &set { + let changed = |state: &kio::Ref<'_, u64>| { + if **state != reader.seen { + Poll::Ready(()) + } else { + Poll::Pending + } + }; + assert!(reader.consumer.poll(&reader.waiter, changed).is_pending()); + } + }) + }) + .collect(); + let crew = Crew::new(work); + + b.iter_custom(|iters| { + let start = Instant::now(); + crew.run(iters); + start.elapsed() + }); + }); + } + group.finish(); +} + +/// A frame payload. Like `Bytes`, cloning it bumps a refcount shared by every reader. +type Frame = Arc; + +/// The shape of `GroupState`'s frame cache: appended at the back, read by absolute index. +#[derive(Default)] +struct Log { + frames: VecDeque, + /// Absolute index of `frames[0]`. + offset: u64, +} + +fn bench_log(c: &mut Criterion) { + let mut group = c.benchmark_group("contended_log"); + for (threads, consumers) in sweep() { + group.throughput(Throughput::Elements(consumers as u64)); + let id = BenchmarkId::new(format!("threads_{threads}"), consumers); + group.bench_function(id, |b| { + let producer = Producer::new(Log::default()); + let work = readers(&producer, consumers, threads) + .into_iter() + .map(|mut set| -> Work { + Box::new(move || { + for reader in &mut set { + let cursor = reader.seen; + let next = |log: &kio::Ref<'_, Log>| match log.frames.get((cursor - log.offset) as usize) { + Some(frame) => Poll::Ready(frame.clone()), + None => Poll::Pending, + }; + let Poll::Ready(Ok(frame)) = reader.consumer.poll(&reader.waiter, next) else { + panic!("an appended frame must be ready"); + }; + black_box(frame); + reader.seen += 1; + + let cursor = reader.seen; + let next = |log: &kio::Ref<'_, Log>| match log.frames.get((cursor - log.offset) as usize) { + Some(_) => Poll::Ready(()), + None => Poll::Pending, + }; + assert!(reader.consumer.poll(&reader.waiter, next).is_pending()); + } + }) + }) + .collect(); + let crew = Crew::new(work); + + let mut index = 0; + b.iter_custom(|iters| { + let start = Instant::now(); + for _ in 0..iters { + { + let mut log = producer.write().ok().expect("open"); + log.frames.push_back(Arc::new(index)); + if log.frames.len() > LOG_FRAMES { + log.frames.pop_front(); + log.offset += 1; + } + } + index += 1; + crew.run(1); + } + start.elapsed() + }); + }); + } + group.finish(); +} + +criterion_group! { + name = benches; + config = Criterion::default().warm_up_time(Duration::from_secs(1)).measurement_time(Duration::from_secs(3)); + targets = bench_notify, bench_repoll, bench_log +} +criterion_main!(benches); diff --git a/rs/moq-gst/Cargo.toml b/rs/moq-gst/Cargo.toml index 96efeb0bbb..9b3db82860 100644 --- a/rs/moq-gst/Cargo.toml +++ b/rs/moq-gst/Cargo.toml @@ -32,5 +32,10 @@ url = { workspace = true } gst-plugin-version-helper = "0.8" [dev-dependencies] +criterion = { workspace = true } futures = { workspace = true } tokio = { workspace = true, features = ["test-util"] } + +[[bench]] +name = "catalog" +harness = false diff --git a/rs/moq-gst/benches/catalog.rs b/rs/moq-gst/benches/catalog.rs new file mode 100644 index 0000000000..bbf333f888 --- /dev/null +++ b/rs/moq-gst/benches/catalog.rs @@ -0,0 +1,50 @@ +//! Per-pad EOS work across total renditions x touched pads, while the first catalog is gated. + +use criterion::{BenchmarkId, Criterion, Throughput, criterion_group, criterion_main}; +use gst::prelude::*; + +fn eos(c: &mut Criterion) { + gst::init().unwrap(); + gstmoq::plugin_register_static().unwrap(); + let caps = gst::Caps::builder("audio/x-opus") + .field("rate", 48_000i32) + .field("channels", 2i32) + .build(); + let mut group = c.benchmark_group("moqsink/eos"); + for renditions in [1, 16, 64] { + let sink = gst::ElementFactory::make("moqsink") + .property("url", "tcp://127.0.0.1:1") + .property("broadcast", "bench") + .build() + .unwrap(); + let pads = (0..renditions) + .map(|_| sink.request_pad_simple("sink_%u").unwrap()) + .collect::>(); + // Keep one reservation unresolved: the lookup must see current configs without cloning the + // full catalog or publishing a snapshot, and EOS must not finalize the whole element. + let _pending = sink.request_pad_simple("sink_%u").unwrap(); + sink.set_state(gst::State::Paused).unwrap(); + for pad in &pads { + assert!(pad.send_event(gst::event::StreamStart::new(pad.name().as_str()))); + assert!(pad.send_event(gst::event::Caps::new(&caps))); + } + for touched in [1, 8, 64] { + group.throughput(Throughput::Elements(touched)); + group.bench_function(BenchmarkId::new(format!("renditions={renditions}"), touched), |b| { + b.iter(|| { + for index in 0..touched as usize { + let pad = &pads[index % pads.len()]; + assert!(pad.send_event(gst::event::FlushStart::new())); + assert!(pad.send_event(gst::event::FlushStop::new(false))); + assert!(pad.send_event(gst::event::Eos::new())); + } + }); + }); + } + sink.set_state(gst::State::Null).unwrap(); + } + group.finish(); +} + +criterion_group!(benches, eos); +criterion_main!(benches); diff --git a/rs/moq-mux/Cargo.toml b/rs/moq-mux/Cargo.toml index ac76acf26a..6565994f91 100644 --- a/rs/moq-mux/Cargo.toml +++ b/rs/moq-mux/Cargo.toml @@ -42,6 +42,15 @@ web-async = { workspace = true } webm-iterable = "0.7" [dev-dependencies] +criterion = { workspace = true } futures = { workspace = true } tokio = { workspace = true, features = ["test-util"] } tracing-test = { workspace = true, features = ["no-env-filter"] } + +[[bench]] +name = "export" +harness = false + +[[bench]] +name = "containers" +harness = false diff --git a/rs/moq-mux/benches/containers.rs b/rs/moq-mux/benches/containers.rs new file mode 100644 index 0000000000..5a625f68f5 --- /dev/null +++ b/rs/moq-mux/benches/containers.rs @@ -0,0 +1,261 @@ +//! Complete fMP4/CMAF and Annex-B imports and exports, with setup outside timing. +use std::time::Duration; + +use bytes::{Bytes, BytesMut}; +use criterion::{BatchSize, BenchmarkId, Criterion, Throughput, criterion_group, criterion_main}; +use moq_mux::catalog::{CatalogFormat, Stream}; +use moq_mux::codec::h264; +use moq_mux::container::{Frame, Kind}; +use moq_net::Timestamp; + +// A real baseline SPS, also exercised by the H.264 import regression. +const SPS: &[u8] = &[ + 0x67, 0x42, 0xc0, 0x1f, 0xda, 0x01, 0x40, 0x16, 0xe9, 0xb8, 0x08, 0x08, 0x0a, 0x00, 0x00, 0x07, 0xd0, 0x00, 0x01, + 0xd4, 0xc0, 0x80, +]; +const PPS: &[u8] = &[0x68, 0xce, 0x3c, 0x80]; + +fn payload(size: usize) -> Bytes { + let mut payload = BytesMut::new(); + for nal in [SPS, PPS, &[0x65, 0x88, 0x84, 0x21]] { + payload.extend_from_slice(&[0, 0, 0, 1]); + payload.extend_from_slice(nal); + } + // Nonzero padding cannot accidentally add an Annex-B start code. + payload.resize(size, 0xff); + payload.freeze() +} + +fn timestamp(frame: usize) -> Timestamp { + Timestamp::from_millis(frame as u64 * 20).unwrap() +} + +struct Fixture { + _published: Box, + _origin: moq_net::origin::Producer, + driver: tokio::task::JoinHandle, + requests: tokio::task::JoinHandle<()>, +} + +impl Drop for Fixture { + fn drop(&mut self) { + self.driver.abort(); + self.requests.abort(); + } +} + +async fn prepare(tracks: usize, frames: usize, payload: &Bytes) -> (moq_mux::Source, impl Stream, Fixture) { + let mut broadcast = moq_net::broadcast::Info::new().produce(); + let consumer = broadcast.consume(); + let mut catalog = moq_mux::catalog::Producer::new(&mut broadcast, Default::default()).unwrap(); + let mut publishers = Vec::new(); + for index in 0..tracks { + let track = broadcast + .create_track( + format!("video{index}"), + catalog.track_info(hang::catalog::PRIORITY.video), + ) + .unwrap(); + let config = h264::config(payload).unwrap(); + let mut publisher = catalog + .video(track, moq_mux::catalog::hang::Container::Legacy(Kind::Video), config) + .unwrap(); + for index in 0..frames { + publisher + .write(Frame { + timestamp: timestamp(index), + duration: Some(Timestamp::from_millis(20).unwrap()), + payload: payload.clone(), + keyframe: true, + }) + .unwrap(); + } + publisher.finish().unwrap(); + publishers.push(publisher); + } + catalog.finish().unwrap(); + let (origin, driver) = moq_net::origin::Producer::new(Default::default()); + let driver = tokio::spawn(moq_net::time::run(driver)); + let dynamic = origin.dynamic("", Default::default()).unwrap(); + let requests = tokio::spawn(async move { + while let Ok(request) = dynamic.requested_broadcast().await { + request.accept(consumer.clone()); + } + }); + let source = moq_mux::Source::new(origin.consume(), "bench"); + let stream = source.catalog::<()>(CatalogFormat::Hang).await.unwrap(); + let fixture = Fixture { + _published: Box::new((broadcast, catalog, publishers)), + _origin: origin, + driver, + requests, + }; + (source, stream, fixture) +} + +async fn export(format: &str, source: moq_mux::Source, stream: impl Stream, collect: bool) -> Bytes { + let mut data = BytesMut::new(); + if format == "cmaf" { + let mut export = moq_mux::container::fmp4::Export::new(source, stream) + .with_max_age(Duration::from_secs(30)) + .with_fragment_duration(Duration::ZERO); + while let Some(chunk) = export.next().await.unwrap() { + if collect { + data.extend_from_slice(&chunk); + } else { + std::hint::black_box(chunk); + } + } + } else { + let mut export = h264::Export::new(source, stream).with_max_age(Duration::from_secs(30)); + while let Some(chunk) = export.next().await.unwrap() { + if collect { + data.extend_from_slice(&chunk); + } else { + std::hint::black_box(chunk); + } + } + } + data.freeze() +} + +fn cmaf_import() -> (moq_mux::catalog::Producer, moq_mux::container::fmp4::Import) { + let mut broadcast = moq_net::broadcast::Info::new().produce(); + let catalog = moq_mux::catalog::Producer::new(&mut broadcast, Default::default()).unwrap(); + let import = moq_mux::container::fmp4::Import::new(broadcast, catalog.reserve()); + (catalog, import) +} + +fn import_cmaf( + (catalog, mut import): (moq_mux::catalog::Producer, moq_mux::container::fmp4::Import), + data: &Bytes, +) -> (moq_mux::catalog::Producer, moq_mux::container::fmp4::Import) { + import.decode(data).unwrap(); + import.finish().unwrap(); + (catalog, import) +} + +struct Annexb { + _broadcast: moq_net::broadcast::Producer, + catalog: moq_mux::catalog::Producer, + import: h264::Import, + split: h264::Split, +} + +impl Annexb { + fn new() -> Self { + let mut broadcast = moq_net::broadcast::Info::new().produce(); + let catalog = moq_mux::catalog::Producer::new(&mut broadcast, Default::default()).unwrap(); + let track = broadcast + .create_track("video", catalog.track_info(hang::catalog::PRIORITY.video)) + .unwrap(); + let import = h264::Import::new(track, catalog.reserve(), Default::default()).unwrap(); + Self { + _broadcast: broadcast, + catalog, + import, + split: h264::Split::new(), + } + } + + fn decode(&mut self, payload: &Bytes, frames: usize) { + for index in 0..frames { + let pts = timestamp(index); + let mut complete = self.split.decode(payload, pts).unwrap(); + complete.extend(self.split.flush(pts).unwrap()); + self.import.decode(complete).unwrap(); + } + self.import.finish().unwrap(); + } +} + +fn containers(c: &mut Criterion) { + let runtime = tokio::runtime::Builder::new_current_thread() + .enable_all() + .build() + .unwrap(); + for format in ["cmaf", "annexb"] { + for tracks in if format == "cmaf" { &[1, 4][..] } else { &[1][..] } { + for frames in [1, 16, 64] { + for size in [256, 4096] { + let payload = payload(size); + let (source, stream, fixture) = runtime.block_on(prepare(*tracks, frames, &payload)); + let data = runtime.block_on(export(format, source, stream, true)); + drop(fixture); + if format == "cmaf" { + let (catalog, mut import) = cmaf_import(); + import.decode(&data).unwrap(); + import.finish().unwrap(); + assert_eq!(catalog.snapshot().video.renditions.len(), *tracks); + use mp4_atom::DecodeMaybe; + let mut cursor = std::io::Cursor::new(data.as_ref()); + let mut fragments = 0; + while let Some(atom) = mp4_atom::Any::decode_maybe(&mut cursor).unwrap() { + if matches!(atom, mp4_atom::Any::Moof(_)) { + fragments += 1; + } + } + assert_eq!(fragments, tracks * frames); + } else { + let mut import = Annexb::new(); + import.decode(&payload, frames); + assert_eq!(import.catalog.snapshot().video.renditions.len(), 1); + } + assert!(!data.is_empty()); + if format == "annexb" { + assert_eq!(data.len(), size * frames); + } + for unit in ["frames", "bytes"] { + let throughput = if unit == "frames" { + Throughput::Elements((*tracks * frames) as u64) + } else { + Throughput::Bytes(data.len() as u64) + }; + for direction in ["import", "export"] { + let mut group = c.benchmark_group(format!("{format}-{direction}-{unit}")); + group.throughput(throughput.clone()); + group.bench_function( + BenchmarkId::new(format!("{tracks}tracks-{frames}frames"), size), + |b| { + if direction == "import" { + if format == "cmaf" { + b.iter_batched( + cmaf_import, + |import| import_cmaf(import, std::hint::black_box(&data)), + BatchSize::PerIteration, + ); + } else { + b.iter_batched( + Annexb::new, + |mut import| { + import.decode(std::hint::black_box(&payload), frames); + import + }, + BatchSize::PerIteration, + ); + } + } else { + b.iter_batched( + || runtime.block_on(prepare(*tracks, frames, &payload)), + |(source, stream, fixture)| { + std::hint::black_box( + runtime.block_on(export(format, source, stream, false)), + ); + fixture + }, + BatchSize::PerIteration, + ); + } + }, + ); + group.finish(); + } + } + } + } + } + } +} + +criterion_group!(benches, containers); +criterion_main!(benches); diff --git a/rs/moq-mux/benches/export.rs b/rs/moq-mux/benches/export.rs new file mode 100644 index 0000000000..0690a62d33 --- /dev/null +++ b/rs/moq-mux/benches/export.rs @@ -0,0 +1,167 @@ +//! Container export cost swept over independently varied tracks and queued frames. +use std::time::Duration; + +use bytes::Bytes; +use criterion::{BatchSize, BenchmarkId, Criterion, Throughput, criterion_group, criterion_main}; +use hang::catalog::{AAC, AudioConfig}; +use moq_mux::catalog::hang::Container; +use moq_mux::container::{Frame, Kind}; +use moq_net::Timestamp; + +/// What keeps a prepared broadcast served; dropping it stops the origin's tasks. +struct Fixture { + /// The broadcast, its catalog and its tracks, which a reader needs kept open. + _published: Box, + _origin: moq_net::origin::Producer, + driver: tokio::task::JoinHandle, + requests: tokio::task::JoinHandle<()>, +} + +impl Drop for Fixture { + fn drop(&mut self) { + self.driver.abort(); + self.requests.abort(); + } +} + +/// A finished broadcast of `tracks` audio tracks with `frames` queued frames each. +async fn prepare(tracks: usize, frames: usize) -> (moq_mux::Source, Fixture) { + let mut broadcast = moq_net::broadcast::Info::new().produce(); + let consumer = broadcast.consume(); + let mut catalog = moq_mux::catalog::Producer::new(&mut broadcast, Default::default()).unwrap(); + let mut publishers = Vec::new(); + for index in 0..tracks { + let track = broadcast + .create_track( + format!("audio{index}"), + catalog.track_info(hang::catalog::PRIORITY.audio), + ) + .unwrap(); + let mut config = AudioConfig::new(AAC { profile: 2 }, 44100, 2); + config.description = Some(Bytes::from_static(&[0x12, 0x10])); + publishers.push(catalog.audio(track, Container::Legacy(Kind::Audio), config).unwrap()); + } + for track in &mut publishers { + for index in 0..frames { + track + .write(Frame { + timestamp: Timestamp::from_millis(index as u64 * 20).unwrap(), + duration: None, + payload: Bytes::from_static(b"audio-frame"), + keyframe: index == 0, + }) + .unwrap(); + } + track.finish().unwrap(); + } + catalog.finish().unwrap(); + let (origin, driver) = moq_net::origin::Producer::new(Default::default()); + let driver = tokio::spawn(moq_net::time::run(driver)); + let dynamic = origin.dynamic("", Default::default()).unwrap(); + let requests = tokio::spawn(async move { + while let Ok(request) = dynamic.requested_broadcast().await { + request.accept(consumer.clone()); + } + }); + let source = moq_mux::Source::new(origin.consume(), "bench"); + let fixture = Fixture { + _published: Box::new((broadcast, catalog, publishers)), + _origin: origin, + driver, + requests, + }; + (source, fixture) +} + +/// An export of one container format, opened but not yet drained. Boxed in setup, outside the timed drain. +enum Export { + Flv(Box), + Mkv(Box>), + Ts(Box), +} + +/// Open an export, subscribing its catalog through the origin. +async fn open(format: &str, source: moq_mux::Source) -> Export { + if format == "flv" { + Export::Flv(Box::new( + moq_mux::container::flv::Export::new(source) + .await + .unwrap() + .with_multitrack(true), + )) + } else if format == "ts" { + Export::Ts(Box::new(moq_mux::container::ts::Export::new(source).await.unwrap())) + } else { + let stream = source.catalog(Default::default()).await.unwrap(); + Export::Mkv(Box::new( + moq_mux::container::mkv::Export::new(source, stream).with_fragment_duration(Duration::ZERO), + )) + } +} + +async fn export(export: Export) -> usize { + let mut chunks = 0; + match export { + Export::Flv(mut export) => { + while let Some(chunk) = export.next().await.unwrap() { + std::hint::black_box(chunk); + chunks += 1; + } + } + Export::Ts(mut export) => { + while let Some(chunk) = export.next().await.unwrap() { + std::hint::black_box(chunk); + chunks += 1; + } + } + Export::Mkv(mut export) => { + while let Some(chunk) = export.next().await.unwrap() { + std::hint::black_box(chunk); + chunks += 1; + } + } + } + chunks +} + +fn bench_export(c: &mut Criterion) { + let runtime = tokio::runtime::Builder::new_current_thread() + .enable_all() + .build() + .unwrap(); + let mut group = c.benchmark_group("container-export"); + group.sample_size(10); + group.measurement_time(Duration::from_secs(1)); + for format in ["flv", "mkv", "ts"] { + for tracks in [1, 4, 16] { + for frames in [1, 16, 64] { + group.throughput(Throughput::Elements((tracks * frames) as u64)); + group.bench_function(BenchmarkId::new(format, format!("{tracks}x{frames}")), |b| { + // Publishing the broadcast and opening the export is setup; only exporting it is measured. + b.iter_batched( + || { + runtime.block_on(async { + let (source, fixture) = prepare(tracks, frames).await; + (open(format, source).await, fixture) + }) + }, + |(opened, fixture)| { + let chunks = runtime.block_on(export(opened)); + if format == "ts" { + assert!(chunks > 0); + } else { + assert_eq!(chunks, 1 + tracks * frames); + } + fixture + }, + BatchSize::PerIteration, + ); + }); + } + } + } + group.finish(); +} + +criterion_group!(benches, bench_export); +criterion_main!(benches); diff --git a/rs/moq-net/benches/group.rs b/rs/moq-net/benches/group.rs index 19520bf205..0cfbcc0499 100644 --- a/rs/moq-net/benches/group.rs +++ b/rs/moq-net/benches/group.rs @@ -1,25 +1,34 @@ //! Delivery-path benchmarks for the group and track models. //! -//! The point of interest is small frames: today each frame in a group is a -//! `frame::Producer` owning its own `kio` channel plus a couple of `Arc`s, so a -//! group with thousands of tiny frames allocates thousands of tiny control -//! objects. These benchmarks write and read many small frames so that cost shows -//! up as wall-clock time, giving a before/after for reshaping frames into plain -//! data. +//! Small frames expose control and frame-slot allocation costs. The write sweep +//! includes the initial deque capacity and its first growth boundary, as well as +//! groups with thousands of tiny frames. //! //! `track_recv_groups` covers the layer above: how much it costs to hand out one //! cached group, swept over cache depth so a per-delivery scan shows up as a slope. //! It also covers the steady-state path after a subscriber has reached the edge. //! +//! `track_stamp_groups` covers stamping a group's first frame, which checks it +//! against its neighbors' starts, swept over cache depth so a scan shows as a slope: +//! `append` among stamped groups, `sparse` among groups created but not yet stamped. +//! //! Run with `cargo bench -p moq-net`. -use std::{hint::black_box, time::Duration}; +use std::{ + hint::black_box, + time::{Duration, Instant}, +}; use bytes::Bytes; use criterion::{BatchSize, BenchmarkId, Criterion, Throughput, criterion_group, criterion_main}; use futures::FutureExt; use moq_net::{Timestamp, broadcast, frame, group, track}; +/// A millisecond track: the frames written here carry timestamps. +fn timed() -> track::Info { + track::Info::default().with_timescale(moq_net::Timescale::MILLI) +} + /// A small, fixed payload shared across every frame. Cloning a `Bytes` is a /// refcount bump (no allocation), so the benchmark isolates the per-frame control /// overhead rather than payload allocation. @@ -28,7 +37,7 @@ const PAYLOAD: usize = 64; /// Frame counts to sweep. 8192 is the largest legal group (`MAX_GROUP_FRAMES`); /// the 8193rd write returns `GroupTooLarge`. The top end is that full group of /// tiny frames. -const COUNTS: [usize; 3] = [512, 2_048, 8_192]; +const COUNTS: [usize; 5] = [4, 5, 512, 2_048, 8_192]; /// Cached group counts to sweep for the track-level delivery benchmarks. A track /// publishing one group per frame at the default 5s retention sits in the hundreds, @@ -50,7 +59,7 @@ struct Ctx { /// Build a fresh, empty group via the public producer path. fn fresh_group() -> Ctx { let broadcast = broadcast::Producer::new(broadcast::Info::default()); - let track = broadcast.create_track("bench", None).unwrap(); + let track = broadcast.create_track("bench", timed()).unwrap(); let group = track.append_group().unwrap(); Ctx { _broadcast: broadcast, @@ -198,7 +207,7 @@ struct TrackCtx { /// Build a track holding N cached groups, each with a single small frame. fn filled_track(n: usize, payload: &Bytes) -> TrackCtx { let broadcast = broadcast::Producer::new(broadcast::Info::default()); - let track = broadcast.create_track("bench", None).unwrap(); + let track = broadcast.create_track("bench", timed()).unwrap(); for _ in 0..n { let mut group = track.append_group().unwrap(); group.write_frame(Timestamp::ZERO, payload.clone()).unwrap(); @@ -214,7 +223,7 @@ fn filled_track(n: usize, payload: &Bytes) -> TrackCtx { fn replay() -> track::Subscription { track::Subscription::default() .with_max_age(Duration::MAX) - .with_start(track::Position::group(0)) + .with_groups(0..) } /// Drain N cached groups from a track, in arrival order and in sequence order. @@ -317,5 +326,64 @@ fn bench_roundtrip(c: &mut Criterion) { g.finish(); } -criterion_group!(benches, bench_write, bench_read, bench_roundtrip, bench_track_recv); +/// Append and stamp groups on a track already holding N cached groups. +/// +/// A group's first frame is checked against the nearest stamped starts on either side, +/// once per group. Each fresh track takes at most N more groups, so the cache stays +/// within 2N for the whole measurement. +fn bench_stamp(c: &mut Criterion) { + let payload = Bytes::from(vec![0u8; PAYLOAD]); + let mut g = c.benchmark_group("track_stamp_groups"); + for &n in &DEPTHS { + g.bench_with_input(BenchmarkId::new("append", n), &n, |b, &n| { + b.iter_custom(|iters| { + let mut total = Duration::ZERO; + let mut left = iters; + while left > 0 { + let ctx = filled_track(n, &payload); + let batch = left.min(n as u64); + let start = Instant::now(); + for _ in 0..batch { + let mut group = ctx.track.append_group().unwrap(); + group.write_frame(Timestamp::ZERO, payload.clone()).unwrap(); + group.finish().unwrap(); + } + total += start.elapsed(); + left -= batch; + } + total + }); + }); + // N groups created before any is stamped, then stamped in order: each one's upper + // neighbor is a run of unstamped groups, which a slot scan walks. + g.bench_with_input(BenchmarkId::new("sparse", n), &n, |b, &n| { + b.iter_custom(|iters| { + let mut total = Duration::ZERO; + let mut left = iters; + while left > 0 { + let ctx = filled_track(0, &payload); + let mut groups: Vec<_> = (0..n).map(|_| ctx.track.append_group().unwrap()).collect(); + let batch = left.min(n as u64); + let start = Instant::now(); + for group in groups.iter_mut().take(batch as usize) { + group.write_frame(Timestamp::ZERO, payload.clone()).unwrap(); + } + total += start.elapsed(); + left -= batch; + } + total + }); + }); + } + g.finish(); +} + +criterion_group!( + benches, + bench_write, + bench_read, + bench_roundtrip, + bench_track_recv, + bench_stamp +); criterion_main!(benches); diff --git a/rs/moq-net/benches/origin.rs b/rs/moq-net/benches/origin.rs index 7ce79e8600..b1d1d64035 100644 --- a/rs/moq-net/benches/origin.rs +++ b/rs/moq-net/benches/origin.rs @@ -1,5 +1,6 @@ //! Fan-out benchmarks for the origin: its route table, the announce cursors -//! watching it, request resolution, and the handoff between local sources. +//! watching it, request resolution, the handoff between local sources, and the +//! delivery through a front. //! //! Every shape is swept over both axes, publishers (routes) and subscribers //! (cursors), so a cost that grows with the size of the table rather than with @@ -8,13 +9,32 @@ //! down to its prefix and beneath it. Competing routes at that prefix are //! ranked once per change, then each cursor selects its first visible entry. //! +//! An equal-cost pool is swept the same way, over its members and the paths it +//! already serves, and over those paths and the cursors watching the pool. +//! +//! A route swap on one front is swept over its tracks and the copies each track +//! still holds from earlier routes, and a front retiring over the fronts around it. +//! Readers draining a swapped track are swept over their count, the groups every +//! route delivered, and the routes each reader still holds a copy of. +//! //! Run with `cargo bench -p moq-net --bench origin`. +// Custom timings measure the real time the work took, not the ambient clock. +#![allow(clippy::disallowed_methods)] + +use std::pin::Pin; +use std::sync::{Arc, Mutex}; +use std::task::{Context, Poll, Wake, Waker}; use std::time::Duration; use criterion::{BatchSize, BenchmarkId, Criterion, Throughput, criterion_group, criterion_main}; use futures::FutureExt; -use moq_net::{Hop, Hops, Pattern, Patterns, Timestamp, announce, broadcast, kio, origin}; +use moq_net::{Hop, Hops, Pattern, Patterns, Timestamp, announce, broadcast, kio, origin, track}; + +/// A millisecond track: the frames written here carry timestamps. +fn timed() -> track::Info { + track::Info::default().with_timescale(moq_net::Timescale::MILLI) +} /// `(publishers, subscribers)` shapes for the fan-out benchmarks. const SHAPES: [(usize, usize); 3] = [(100, 10), (1_000, 100), (1_000, 1_000)]; @@ -81,6 +101,35 @@ fn bench_announce(c: &mut Criterion) { group.finish(); } +/// Cancel unseen additions while every cursor stalls on its initial replay. +fn bench_pending_churn(c: &mut Criterion) { + let mut group = c.benchmark_group("origin/pending_churn"); + for routes in [10, 1_000] { + for readers in [1, 100] { + group.bench_function(format!("{routes}routes_{readers}readers"), |b| { + let (producer, _driver) = origin::Producer::new(origin::Config::default()); + let _routes: Vec<_> = (0..routes) + .map(|i| producer.publish(format!("z/{i}"), origin::Route::default()).unwrap()) + .collect(); + let mut cursors: Vec<_> = (0..readers).map(|_| producer.consume().announced()).collect(); + let mut sequence = 0u64; + b.iter(|| { + sequence += 1; + drop( + producer + .publish(format!("a/{sequence}"), origin::Route::default()) + .unwrap(), + ); + }); + for cursor in &mut cursors { + assert_eq!(std::iter::from_fn(|| cursor.try_next()).count(), routes); + } + }); + } + } + group.finish(); +} + /// A peer carrying `publishers` broadcasts withdraws them all before the /// driver's clock moves again, then `subscribers` cursors drain the retractions. /// Each retraction quarantines its prefix until a fresh clock reading, so a cost @@ -228,7 +277,10 @@ fn bench_announce_duplicate(c: &mut Criterion) { for (duplicates, subscribers) in CONTENDED { let id = BenchmarkId::from_parameter(format!("{duplicates}d_{subscribers}s")); group.bench_function(id, |b| { - let (producer, _driver) = origin::Producer::new(origin::Config::default()); + // No hold, so an iteration times the ranking rather than the origin batching it. + let mut config = origin::Config::default(); + config.update_hold = Duration::ZERO; + let (producer, _driver) = origin::Producer::new(config); let consumer = producer.consume(); // Every peer announces the one path, each under its own hop chain so // the entries are distinct routes rather than one re-priced in place. @@ -264,52 +316,63 @@ fn bench_announce_duplicate(c: &mut Criterion) { /// honor each cursor's scope rather than reuse one global winner. fn bench_reprice_duplicate(c: &mut Criterion) { let mut group = c.benchmark_group("origin/reprice_duplicate"); - for (duplicates, subscribers) in CONTENDED { - let id = BenchmarkId::from_parameter(format!("{duplicates}d_{subscribers}s")); - group.bench_function(id, |b| { - let (producer, _driver) = origin::Producer::new(origin::Config::default()); - let _routes: Vec<_> = (1..=duplicates) - .map(|peer| { - producer - .scope( - "", - &Patterns::from(Pattern::subtree(&format!("{PATH}/p{}", (peer - 1) % 2)).unwrap()), - ) - .unwrap() - .dynamic(PATH, peer_route(peer as u64, INCUMBENT_COST)) - .unwrap() - }) - .collect(); - let challenger = duplicates as u64 + 1; - let route = producer - .dynamic(PATH, peer_route(challenger, INCUMBENT_COST + 1)) - .unwrap(); - let mut cursors: Vec = (0..subscribers) - .map(|peer| { - producer - .consume() - .scope( - "", - &Patterns::from(Pattern::subtree(&format!("{PATH}/p{}", peer % 2)).unwrap()), - ) - .unwrap() - .with_hidden(true) - .announced() - }) - .collect(); - for cursor in &mut cursors { - while cursor.next().now_or_never().flatten().is_some() {} - } - b.iter(|| { - for cost in [INCUMBENT_COST - 1, INCUMBENT_COST + 1] { - route.update(peer_route(challenger, cost)).unwrap(); - for cursor in &mut cursors { - let update = cursor.next().now_or_never().flatten().expect("winner changed"); - assert_eq!(update.route.cost, moq_net::origin::Cost::new(cost.min(INCUMBENT_COST))); - } + for incumbent_cost in [INCUMBENT_COST, origin::Cost::MAX.warm - 1] { + for (duplicates, subscribers) in CONTENDED { + let id = BenchmarkId::from_parameter(format!("{duplicates}d_{subscribers}s_cost{incumbent_cost}")); + group.bench_function(id, |b| { + // No hold, so an iteration times the ranking rather than the origin batching it. + let mut config = origin::Config::default(); + config.update_hold = Duration::ZERO; + let (producer, _driver) = origin::Producer::new(config); + let _routes: Vec<_> = (1..=duplicates) + .map(|peer| { + producer + .scope( + "", + &Patterns::from(Pattern::subtree(&format!("{PATH}/p{}", (peer - 1) % 2)).unwrap()), + ) + .unwrap() + .dynamic(PATH, peer_route(peer as u64, incumbent_cost)) + .unwrap() + }) + .collect(); + let challenger = duplicates as u64 + 1; + let route = producer + .dynamic(PATH, peer_route(challenger, incumbent_cost + 1)) + .unwrap(); + let mut cursors: Vec = (0..subscribers) + .map(|peer| { + producer + .consume() + .scope( + "", + &Patterns::from(Pattern::subtree(&format!("{PATH}/p{}", peer % 2)).unwrap()), + ) + .unwrap() + .with_hidden(true) + .announced() + }) + .collect(); + for cursor in &mut cursors { + while cursor.next().now_or_never().flatten().is_some() {} } + b.iter(|| { + for cost in [incumbent_cost - 1, incumbent_cost + 1] { + route.update(peer_route(challenger, cost)).unwrap(); + for cursor in &mut cursors { + // The winner is re-priced in place. + let update = cursor.next().now_or_never().flatten().expect("winner changed"); + assert_eq!( + update.kind, + announce::Kind::Updated, + "expected an update: got {update:?}" + ); + assert_eq!(update.route.cost, moq_net::origin::Cost::new(cost.min(incumbent_cost))); + } + } + }); }); - }); + } } group.finish(); } @@ -410,7 +473,11 @@ fn bench_subscribe(c: &mut Criterion) { b.iter(|| { let mut cursor = fleet.consumer.announced(); let mut replayed = 0; - while cursor.next().now_or_never().flatten().is_some() { + while let Some(announce::Update { + kind: announce::Kind::Announced, + .. + }) = cursor.next().now_or_never().flatten() + { replayed += 1; } assert_eq!(replayed, publishers); @@ -458,6 +525,310 @@ fn bench_request(c: &mut Criterion) { group.finish(); } +/// Records which request a wake belongs to, so only woken requests are polled again. +struct Woken { + id: usize, + list: Arc>>, +} + +impl Wake for Woken { + fn wake(self: Arc) { + self.list.lock().unwrap().push(self.id); + } +} + +/// Waiting requests, polled only once woken, the way an executor polls tasks. +struct Waiting { + requests: Vec>>, + wakers: Vec, + woken: Arc>>, +} + +impl Waiting { + /// Every request starts woken, so the first drain polls each once. + fn new(requests: Vec) -> Self { + let woken = Arc::new(Mutex::new((0..requests.len()).collect())); + let wakers = (0..requests.len()) + .map(|id| { + Waker::from(Arc::new(Woken { + id, + list: woken.clone(), + })) + }) + .collect(); + Self { + requests: requests.into_iter().map(Box::pin).collect(), + wakers, + woken, + } + } + + fn is_woken(&self) -> bool { + !self.woken.lock().unwrap().is_empty() + } + + /// Polls the woken requests until none are, asserting each is still waiting. + fn drain(&mut self) { + loop { + let mut ids = std::mem::take(&mut *self.woken.lock().unwrap()); + if ids.is_empty() { + return; + } + ids.sort_unstable(); + ids.dedup(); + for id in ids { + let mut cx = Context::from_waker(&self.wakers[id]); + assert!(self.requests[id].as_mut().poll(&mut cx).is_pending()); + } + } + } +} + +/// Waiting requests have already received an unroutable verdict. Repricing a +/// covering route wakes them; repricing an unrelated route must leave them asleep. +/// Shared paths isolate requester fan-out from the cost of minting one front per path. +/// Only woken requests are polled, so harness polling does not scale with requesters. +fn bench_unroutable_churn(c: &mut Criterion) { + let mut group = c.benchmark_group("origin/unroutable_churn"); + group.sample_size(20); + for covering in [false, true] { + for requesters in [0, 1, 32, 256] { + for distinct in [false, true] { + if distinct && requesters <= 1 { + continue; + } + for changes in [1, 32] { + let scope = if covering { "covering" } else { "unrelated" }; + let paths = if distinct { "distinct" } else { "shared" }; + let id = BenchmarkId::new(scope, format!("{requesters}r_{changes}c_{paths}")); + group.throughput(Throughput::Elements(changes)); + group.bench_function(id, |b| { + let (producer, mut driver) = origin::Producer::new(origin::Config::default()); + let consumer = producer.consume(); + let server = producer.dynamic("room", origin::Route::default()).unwrap(); + let unrelated = producer.dynamic("other", origin::Route::default()).unwrap(); + let mut waiting = Waiting::new( + (0..requesters) + .map(|i| { + consumer.routed_broadcast(format!("room/job-{}", if distinct { i } else { 0 })) + }) + .collect(), + ); + let waiter = kio::Waiter::noop(); + let now = moq_net::time::Instant::now(); + // Drive the first requests to refusal before measuring any route churn. + waiting.drain(); + driver.poll(now, &waiter).unwrap(); + let expected = if distinct { + requesters + } else { + usize::from(requesters > 0) + }; + for _ in 0..expected { + let Poll::Ready(Ok(request)) = server.poll_requested_broadcast(&waiter) else { + panic!("every distinct path reaches the covering handler"); + }; + request.reject(moq_net::Error::Unroutable); + } + driver.poll(now, &waiter).unwrap(); + waiting.drain(); + assert!(server.poll_requested_broadcast(&waiter).is_pending()); + let changed = if covering { &server } else { &unrelated }; + let mut cost = 1; + b.iter(|| { + for _ in 0..changes { + cost = 3 - cost; + changed.update(origin::Route::default().with_cost(cost)).unwrap(); + assert!(covering || !waiting.is_woken(), "unrelated churn wakes no request"); + waiting.drain(); + driver.poll(now, &waiter).unwrap(); + for _ in 0..if covering { expected } else { 0 } { + let Poll::Ready(Ok(request)) = server.poll_requested_broadcast(&waiter) else { + panic!("covering churn re-asks every distinct path"); + }; + request.reject(moq_net::Error::Unroutable); + } + assert!(server.poll_requested_broadcast(&waiter).is_pending()); + driver.poll(now, &waiter).unwrap(); + assert!(covering || !waiting.is_woken(), "unrelated churn wakes no request"); + waiting.drain(); + } + }); + }); + } + } + } + } + group.finish(); +} + +/// A front minted, resolved, and retired once its holder leaves, among `fronts` +/// others held open. A retiring front leaves the table it shares with them, so a +/// cost that grows with the table rather than with the one path shows up as a slope. +fn bench_retire(c: &mut Criterion) { + let mut group = c.benchmark_group("origin/retire"); + for fronts in [100, 1_000, 10_000] { + group.bench_function(BenchmarkId::from_parameter(format!("{fronts}f")), |b| { + let mut fleet = fanout(fronts + 1, 0); + let waiter = kio::Waiter::noop(); + let held: Vec<_> = (0..fronts) + .map(|i| fleet.consumer.request_broadcast(format!("room/{i}"))) + .collect(); + fleet.driver.poll(moq_net::time::Instant::now(), &waiter).unwrap(); + let _held: Vec<_> = held + .into_iter() + .map(|pending| pending.now_or_never().expect("resolved once driven").expect("local")) + .collect(); + let path = format!("room/{fronts}"); + b.iter(|| { + let pending = fleet.consumer.request_broadcast(&path); + fleet.driver.poll(moq_net::time::Instant::now(), &waiter).unwrap(); + drop(pending.now_or_never().expect("resolved once driven").expect("local")); + // Nothing holds it and a local source keeps no track, so it retires here. + fleet.driver.poll(moq_net::time::Instant::now(), &waiter).unwrap(); + }); + }); + } + group.finish(); +} + +/// `(members, paths)` shapes for an equal-cost pool: how many advertisers claim +/// one prefix, against how many paths beneath it are already being served. +const POOL: [usize; 3] = [4, 32, 256]; +const POOL_PATHS: [usize; 3] = [100, 1_000, 10_000]; + +/// Announce cursors watching the pool, against [`POOL_PATHS`]. +const POOL_CURSORS: [usize; 3] = [1, 100, 1_000]; + +/// Hop ids for pool members, clear of the origin's own. +const POOL_HOP: u64 = 1_000; + +/// An origin where `members` equal-cost advertisers claim `pool`, already +/// serving `paths` requested paths beneath it, spread across the pool by the +/// hash on each path, and watched by `cursors` announce cursors. Every handle +/// is held so the fronts stay up. +struct Pool { + producer: origin::Producer, + driver: origin::Driver, + _members: Vec, + _producers: Vec, + _consumers: Vec, + cursors: Vec, +} + +impl Pool { + /// Discard what the cursors were delivered, so each iteration measures only what it adds. + fn drain(&mut self) { + for cursor in &mut self.cursors { + while cursor.next().now_or_never().flatten().is_some() {} + } + } +} + +fn pool(members: usize, paths: usize, cursors: usize) -> Pool { + let (producer, mut driver) = origin::Producer::new(origin::Config::new(Hop::new(1).unwrap())); + let consumer = producer.consume(); + let cursors = (0..cursors).map(|_| consumer.announced()).collect(); + let members: Vec<_> = (0..members) + .map(|i| producer.dynamic("pool", pool_route(POOL_HOP + i as u64)).unwrap()) + .collect(); + let waiter = kio::Waiter::noop(); + + let requests: Vec<_> = (0..paths) + .map(|i| consumer.request_broadcast(format!("pool/job-{i}"))) + .collect(); + driver.poll(moq_net::time::Instant::now(), &waiter).unwrap(); + let mut producers = Vec::with_capacity(paths); + for member in &members { + while let Poll::Ready(request) = member.poll_requested_broadcast(&waiter) { + let broadcast = broadcast::Info::new().produce(); + request.unwrap().accept(&broadcast); + producers.push(broadcast); + } + } + assert_eq!(producers.len(), paths, "every path reached a member"); + driver.poll(moq_net::time::Instant::now(), &waiter).unwrap(); + let consumers = requests + .into_iter() + .map(|request| request.now_or_never().expect("resolved once driven").expect("served")) + .collect(); + + let mut pool = Pool { + producer, + driver, + _members: members, + _producers: producers, + _consumers: consumers, + cursors, + }; + pool.drain(); + pool +} + +/// A pool member's claim: one hop, at the same cost as every other member. +fn pool_route(hop: u64) -> origin::Route { + peer_route(hop, 3) +} + +/// A member joining and leaving a pool that already serves `paths` paths. +/// +/// Every front under the prefix watches the routes covering it, so each one +/// re-selects on both changes, and each selection scans the pool: the cost is +/// paths times members by design, which this sweep makes visible. The fronts the +/// newcomer outranks request through it, and fall back when it leaves before +/// answering. +fn bench_pool_churn(c: &mut Criterion) { + let mut group = c.benchmark_group("origin/pool_churn"); + // The widest shapes take a noticeable fraction of a second per iteration. + group.sample_size(10); + for members in POOL { + for paths in POOL_PATHS { + let id = BenchmarkId::from_parameter(format!("{members}m_{paths}p")); + group.bench_function(id, |b| { + let mut pool = pool(members, paths, 0); + let waiter = kio::Waiter::noop(); + let hop = POOL_HOP + members as u64; + b.iter(|| { + let joined = pool.producer.dynamic("pool", pool_route(hop)).unwrap(); + pool.driver.poll(moq_net::time::Instant::now(), &waiter).unwrap(); + drop(joined); + pool.driver.poll(moq_net::time::Instant::now(), &waiter).unwrap(); + }); + }); + } + } + group.finish(); +} + +/// [`bench_pool_churn`] at 32 members, watched by `cursors` announce cursors. +/// +/// A front whose path the newcomer takes checks whether every cursor on the +/// prefix already restarted for it before renewing the prefix, so the moved +/// fronts times the cursors is the slope to watch. +fn bench_pool_cursors(c: &mut Criterion) { + let mut group = c.benchmark_group("origin/pool_cursors"); + group.sample_size(10); + let members = 32; + for paths in POOL_PATHS { + for cursors in POOL_CURSORS { + let id = BenchmarkId::from_parameter(format!("{paths}p_{cursors}c")); + group.bench_function(id, |b| { + let mut pool = pool(members, paths, cursors); + let waiter = kio::Waiter::noop(); + let hop = POOL_HOP + members as u64; + b.iter(|| { + let joined = pool.producer.dynamic("pool", pool_route(hop)).unwrap(); + pool.driver.poll(moq_net::time::Instant::now(), &waiter).unwrap(); + drop(joined); + pool.driver.poll(moq_net::time::Instant::now(), &waiter).unwrap(); + pool.drain(); + }); + }); + } + } + group.finish(); +} + /// Publisher handoff at one path: a subscriber is reading from one local /// source when a second announces at the same path and takes over (newest /// wins). Measured from the standby's attach to the subscriber receiving its @@ -484,7 +855,7 @@ fn bench_handoff(c: &mut Criterion) { let mut total = Duration::ZERO; for _ in 0..iterations { let incumbent = producer.publish("room/live", origin::Route::default()).unwrap(); - let track = incumbent.create_track("video", None).unwrap(); + let track = incumbent.create_track("video", timed()).unwrap(); let mut first = track.append_group().unwrap(); first.write_frame(Timestamp::ZERO, b"one".as_ref()).unwrap(); first.finish().unwrap(); @@ -495,14 +866,16 @@ fn bench_handoff(c: &mut Criterion) { let started = std::time::Instant::now(); let standby = producer.create_broadcast("room/live").unwrap(); - let track = standby.create_track("video", None).unwrap(); + let track = standby.create_track("video", timed()).unwrap(); let mut second = track.create_group(moq_net::group::Info { sequence: 1 }).unwrap(); second.write_frame(Timestamp::ZERO, b"two".as_ref()).unwrap(); second.finish().unwrap(); // Announcing is what makes the standby a route the front can take. standby.announce(origin::Route::default()).unwrap(); - subscription.recv_group().await.unwrap().expect("standby group"); + // The open subscription follows the route swap onto the standby. + let group = subscription.recv_group().await.unwrap().expect("standby group"); total += started.elapsed(); + assert_eq!(group.sequence, 1, "standby group"); drop(subscription); incumbent.close(); @@ -518,9 +891,412 @@ fn bench_handoff(c: &mut Criterion) { group.finish(); } +/// `(tracks, readers per track)` shapes for [`bench_relay`]. +const RELAY: [(usize, usize); 4] = [(1, 1), (1, 100), (100, 1), (10, 100)]; + +/// Frames per group in [`bench_relay`]. +const RELAY_FRAMES: usize = 10; + +/// Steady-state delivery through a front: one group of [`RELAY_FRAMES`] frames per +/// track, read in full by every reader of the broadcast the front serves. Readers +/// drain the serving route's shared group cache directly, so measure both the track +/// and reader axes. +fn bench_relay(c: &mut Criterion) { + let mut group = c.benchmark_group("origin/relay"); + for (tracks, readers) in RELAY { + group.throughput(Throughput::Elements((tracks * readers * RELAY_FRAMES) as u64)); + group.bench_function(BenchmarkId::from_parameter(format!("{tracks}t_{readers}r")), |b| { + let runtime = tokio::runtime::Builder::new_current_thread() + .enable_all() + .build() + .unwrap(); + let (producer, driver) = origin::Producer::new(origin::Config::default()); + runtime.spawn(moq_net::time::run(driver)); + let broadcast = producer.publish("room/live", origin::Route::default()).unwrap(); + let sources: Vec<_> = (0..tracks) + .map(|i| broadcast.create_track(format!("{i}"), timed()).unwrap()) + .collect(); + let mut subscriptions = runtime.block_on(async { + let resolved = producer.consume().request_broadcast("room/live").await.unwrap(); + let mut subscriptions = Vec::new(); + for i in 0..tracks { + let track = resolved.track(&format!("{i}")).unwrap(); + for _ in 0..readers { + subscriptions.push(track.subscribe(None).await.unwrap()); + } + } + subscriptions + }); + + b.iter_custom(|iterations| { + runtime.block_on(async { + let started = std::time::Instant::now(); + for _ in 0..iterations { + for source in &sources { + let mut group = source.append_group().unwrap(); + for _ in 0..RELAY_FRAMES { + group.write_frame(Timestamp::ZERO, b"frame".as_ref()).unwrap(); + } + group.finish().unwrap(); + } + for subscription in &mut subscriptions { + let mut group = subscription.recv_group().await.unwrap().expect("group"); + for _ in 0..RELAY_FRAMES { + group.read_frame().await.unwrap().expect("frame"); + } + } + } + started.elapsed() + }) + }); + }); + } + group.finish(); +} + +/// Re-poll stalled groups through a front, including their subscription budget. +/// Sweep tracks and readers separately to expose unrelated table scans. +fn bench_parked(c: &mut Criterion) { + let mut group = c.benchmark_group("origin/parked"); + for tracks in [1, 100] { + for readers in [1, 100] { + group.throughput(Throughput::Elements((tracks * readers) as u64)); + group.bench_function(BenchmarkId::from_parameter(format!("{tracks}t_{readers}r")), |b| { + let runtime = tokio::runtime::Builder::new_current_thread() + .enable_all() + .build() + .unwrap(); + let (producer, driver) = origin::Producer::new(origin::Config::default()); + runtime.spawn(moq_net::time::run(driver)); + let broadcast = producer.publish("room/live", origin::Route::default()).unwrap(); + let sources: Vec<_> = (0..tracks) + .map(|i| broadcast.create_track(format!("{i}"), None).unwrap()) + .collect(); + let _writers: Vec<_> = sources.iter().map(|source| source.append_group().unwrap()).collect(); + let (subscriptions, mut groups) = runtime.block_on(async { + let resolved = producer.consume().request_broadcast("room/live").await.unwrap(); + let mut subscriptions = Vec::new(); + let mut groups = Vec::new(); + for i in 0..tracks { + let track = resolved.track(&format!("{i}")).unwrap(); + for _ in 0..readers { + let mut sub = track.subscribe(None).await.unwrap(); + groups.push(sub.recv_group().await.unwrap().unwrap()); + subscriptions.push(sub); + } + } + (subscriptions, groups) + }); + let waiter = kio::Waiter::noop(); + b.iter(|| { + for group in &mut groups { + assert!(group.poll_read_frame(&waiter).is_pending()); + } + }); + drop(subscriptions); + }); + } + } + group.finish(); +} + +/// `(tracks, copies)` for [`bench_copy_walk`]: tracks one front is serving, and +/// route copies each of those tracks still holds when the route swaps. +/// +/// The single-copy points from 32 to 128 tracks show the per-track slope alone. +const COPY_WALK: [(usize, usize); 8] = [(1, 1), (1, 32), (32, 1), (64, 1), (128, 1), (8, 32), (32, 8), (64, 16)]; + +const COPY_PATH: &str = "room/live"; + +/// How long a splice may take before the bench treats it as stuck. +const COPY_WAIT: Duration = Duration::from_secs(5); + +/// One generation at [`COPY_PATH`], held so its copies stay subscribed. +struct Generation { + broadcast: broadcast::Producer, + tracks: Vec, +} + +impl Generation { + /// Create a hidden generation and leave a finished group on every track, so a + /// reader that does look has something to resume. + fn build(producer: &origin::Producer, tracks: usize) -> Self { + let broadcast = producer.create_broadcast(COPY_PATH).unwrap(); + let mut held = Vec::with_capacity(tracks); + for i in 0..tracks { + let track = broadcast.create_track(format!("{i}"), timed()).unwrap(); + let mut written = track.append_group().unwrap(); + written.write_frame(Timestamp::ZERO, b"f".as_ref()).unwrap(); + written.finish().unwrap(); + held.push(track); + } + Self { + broadcast, + tracks: held, + } + } + + /// Route the front to this generation without waiting. + fn route(&self) { + self.broadcast.announce(origin::Route::default()).unwrap(); + } + + /// Route the front to this generation and wait until every track is spliced. + /// + /// The wait ends on demand, which flips when the front subscribes the new + /// copy, before its splice. That bounds the splice only on a current-thread + /// runtime, where every local query is ready within the front poll that + /// splices it. + async fn announce(&self) { + self.route(); + for track in &self.tracks { + tokio::time::timeout(COPY_WAIT, track.demand().used()) + .await + .expect("route swap did not splice") + .expect("replacement track closed"); + } + } +} + +/// A front serving [`COPY_PATH`], with `copies` route copies on each of its tracks. +struct CopyWalk { + _producer: origin::Producer, + _subscribers: Vec, + generations: Vec, + /// The generation the timed swap announces, built during setup. + next: Option, +} + +/// Build a front with `copies` already spliced onto each track. +/// +/// Subscribers are never polled. A poll drops a replaced copy once its groups +/// are delivered, and this measures the walk over the copies a swap still has. +/// The same is why every generation is kept: dropping one ends its copy. +async fn copy_walk(tracks: usize, copies: usize) -> CopyWalk { + assert!(copies >= 1, "a front holds at least the serving copy"); + let (producer, driver) = origin::Producer::new(origin::Config::default()); + tokio::spawn(moq_net::time::run(driver)); + let first = Generation::build(&producer, tracks); + first.route(); + let resolved = tokio::time::timeout(COPY_WAIT, producer.consume().request_broadcast(COPY_PATH)) + .await + .expect("front did not resolve") + .expect("front refused"); + let mut subscribers = Vec::with_capacity(tracks); + for i in 0..tracks { + let track = resolved.track(&format!("{i}")).unwrap(); + let subscriber = tokio::time::timeout(COPY_WAIT, track.subscribe(None)) + .await + .expect("track did not splice") + .expect("subscribe failed"); + subscribers.push(subscriber); + } + let mut generations = Vec::with_capacity(copies + 1); + generations.push(first); + for _ in 1..copies { + let generation = Generation::build(&producer, tracks); + generation.announce().await; + generations.push(generation); + } + CopyWalk { + next: Some(Generation::build(&producer, tracks)), + _producer: producer, + _subscribers: subscribers, + generations, + } +} + +impl CopyWalk { + /// Swap in the prebuilt newer route. + /// + /// A local track's info is already known, so the front subscribes, walks + /// every track, and each reader walks its copies in one driver turn. + async fn swap(&mut self) { + let generation = self.next.take().expect("one swap per setup"); + generation.announce().await; + self.generations.push(generation); + } +} + +/// A route swap on one front. Setup holds `copies` per track and builds the +/// next generation, untimed; the announce and splice are timed. Warm-up sizes +/// the iteration count from wall time, which includes that setup, so a large +/// front does not multiply it by a fast routine. +/// +/// The runtime must stay current-thread; see [`Generation::announce`]. +fn bench_copy_walk(c: &mut Criterion) { + let mut group = c.benchmark_group("origin/copy_walk"); + group.sample_size(10); + for (tracks, copies) in COPY_WALK { + group.throughput(Throughput::Elements(tracks as u64)); + group.bench_function(BenchmarkId::from_parameter(format!("{tracks}t_{copies}c")), |b| { + let runtime = tokio::runtime::Builder::new_current_thread() + .enable_all() + .build() + .unwrap(); + b.iter_batched_ref( + || runtime.block_on(copy_walk(tracks, copies)), + |rig| runtime.block_on(rig.swap()), + BatchSize::PerIteration, + ); + }); + } + group.finish(); +} + +/// `(readers, retained, routes)` for [`bench_resume_overlap`]: readers of one track, +/// the groups every route holds, and the routes it swapped through. +const RESUME_OVERLAP: [(usize, usize, usize); 15] = [ + (1, 1, 1), + (1, 64, 1), + (32, 64, 1), + (1, 1, 2), + (1, 64, 2), + (1, 1024, 2), + (8, 1, 2), + (8, 64, 2), + (8, 1024, 2), + (32, 64, 2), + (256, 64, 2), + (32, 1024, 2), + (1, 64, 8), + (1, 64, 32), + (32, 64, 8), +]; + +/// A track that swapped through its routes while its readers' local cap held every +/// group back, so each reader keeps a copy per route and every copy holds the same +/// groups, each replacement in reverse arrival order. +struct ResumeOverlap { + _producer: origin::Producer, + _generations: Vec<(broadcast::Producer, track::Producer)>, + readers: Vec, +} + +enum ResumeReader { + Arrival(track::Subscriber), + Sequence(track::Ordered), +} + +impl ResumeReader { + fn poll_group(&mut self, waiter: &kio::Waiter) -> Poll>> { + match self { + Self::Arrival(reader) => reader.poll_recv_group(waiter), + Self::Sequence(reader) => reader.poll_next_group(waiter), + } + } +} + +async fn resume_overlap(readers: usize, retained: usize, routes: usize, ordered: bool) -> ResumeOverlap { + let (producer, driver) = origin::Producer::new(origin::Config::default()); + tokio::spawn(moq_net::time::run(driver)); + let generation = |order: &mut dyn Iterator| { + let broadcast = producer.create_broadcast(COPY_PATH).unwrap(); + let track = broadcast.create_track("0", timed()).unwrap(); + for sequence in order { + let mut group = track + .create_group(moq_net::group::Info { + sequence: sequence as u64, + }) + .unwrap(); + group.write_frame(Timestamp::ZERO, b"f".as_ref()).unwrap(); + group.finish().unwrap(); + } + broadcast.announce(origin::Route::default()).unwrap(); + (broadcast, track) + }; + let mut generations = vec![generation(&mut (0..retained))]; + let resolved = tokio::time::timeout(COPY_WAIT, producer.consume().request_broadcast(COPY_PATH)) + .await + .expect("front did not resolve") + .expect("front refused"); + let track = resolved.track("0").unwrap(); + // A budget covering every group, so none is written off as late. + let subscription = track::Subscription::default().with_max_age(Duration::from_secs(3600)); + let mut subscribers = Vec::with_capacity(readers); + for _ in 0..readers { + let mut subscriber = tokio::time::timeout(COPY_WAIT, track.subscribe(subscription.clone())) + .await + .expect("track did not splice") + .expect("subscribe failed"); + // Held back until the drain, so every route's copy is kept for its unread groups. + subscriber.set_groups(..0); + subscribers.push(subscriber); + } + let waiter = kio::Waiter::noop(); + for _ in 1..routes { + generations.push(generation(&mut (0..retained).rev())); + let (_, latest) = generations.last().unwrap(); + tokio::time::timeout(COPY_WAIT, latest.demand().used()) + .await + .expect("route swap did not splice") + .expect("replacement track closed"); + // Each reader follows the swap, subscribing to the new copy. + for subscriber in &mut subscribers { + assert!( + subscriber.poll_recv_group(&waiter).is_pending(), + "the cap holds every group" + ); + } + } + for subscriber in &mut subscribers { + subscriber.set_groups(..); + } + let readers = subscribers + .into_iter() + .map(|subscriber| match ordered { + true => ResumeReader::Sequence(subscriber.ordered()), + false => ResumeReader::Arrival(subscriber), + }) + .collect(); + ResumeOverlap { + _producer: producer, + _generations: generations, + readers, + } +} + +/// Readers draining a track whose routes all delivered every group: each group is +/// handed out once, and every other copy's duplicate is skipped against what the reader +/// handed out. The slopes over retained groups and routes are that skipping's cost. +fn bench_resume_overlap(c: &mut Criterion) { + let mut group = c.benchmark_group("origin/resume_overlap"); + group.sample_size(10); + for (order, ordered) in [("arrival", false), ("sequence", true)] { + for (readers, retained, routes) in RESUME_OVERLAP { + group.throughput(Throughput::Elements((readers * retained) as u64)); + group.bench_function( + BenchmarkId::new(order, format!("{readers}r_{retained}g_{routes}c")), + |b| { + let runtime = tokio::runtime::Builder::new_current_thread() + .enable_all() + .build() + .unwrap(); + let waiter = kio::Waiter::noop(); + b.iter_batched_ref( + || runtime.block_on(resume_overlap(readers, retained, routes, ordered)), + |rig| { + for reader in &mut rig.readers { + let mut delivered = 0; + while let Poll::Ready(Ok(Some(_))) = reader.poll_group(&waiter) { + delivered += 1; + } + assert_eq!(delivered, retained, "every group once"); + } + }, + BatchSize::PerIteration, + ); + }, + ); + } + } + group.finish(); +} + criterion_group!( benches, bench_announce, + bench_pending_churn, bench_withdraw, bench_announce_mounted, bench_announce_fleet, @@ -530,6 +1306,14 @@ criterion_group!( bench_serve_idle, bench_subscribe, bench_request, - bench_handoff + bench_unroutable_churn, + bench_retire, + bench_pool_churn, + bench_pool_cursors, + bench_handoff, + bench_relay, + bench_parked, + bench_copy_walk, + bench_resume_overlap ); criterion_main!(benches); diff --git a/rs/moq-net/benches/session.rs b/rs/moq-net/benches/session.rs index 0a47f54cad..50ddb21208 100644 --- a/rs/moq-net/benches/session.rs +++ b/rs/moq-net/benches/session.rs @@ -11,6 +11,12 @@ //! route table they occupy, not a publisher writing into its own cache. //! //! `SESSION_ALLOCS=1` prints allocations per viewer-group instead of timing. +//! Cache-clock advancement stays outside the timed region. +//! `session_delivery_group_cost` sweeps publishers and viewers independently for SUBSCRIBE +//! and FETCH. +//! `SESSION_EXPIRY_ONLY=1` uses the standalone 30s-expiry pool and reports groups +//! written plus charged cache bytes every 10s during long profiling runs. Each round +//! advances the cache clock by a second, so the expiry window spans 30 rounds. //! //! `session_join_*` times a new viewer connecting, resolving a broadcast, and //! receiving its latest group, swept over what the relays already announce. @@ -20,11 +26,23 @@ //! one small frame per track per tick. Throughput counts tracks, so a flat //! elements/s across a sweep means the cost per track stays constant. //! +//! `session_idle` times one second of idle viewer sessions, swept over sessions. +//! Sweep held consumers per session too: only held consumers request PROBE or transport estimates. +//! Nothing else happens, so it measures the sessions' own work. Paused time skips waiting. +//! +//! `session_churn_*` times one request opened and closed by a session that already +//! holds others open, swept over sessions and requests held per session. On +//! moq-transport 14 every close is credited back through the MAX_REQUEST_ID window, +//! so a slope here is that bookkeeping growing with the sessions or the requests. +//! //! Everything runs on one current-thread runtime, so the only work measured is //! the protocol and model code, never scheduling across threads. //! //! Run with `cargo bench -p moq-net --bench session`. +// Custom timings measure the real time the work took, not the ambient clock. +#![allow(clippy::disallowed_methods)] + #[path = "../tests/support/mod.rs"] mod support; @@ -38,6 +56,11 @@ use moq_net::{Hop, Timestamp, Version, broadcast, cache, group, origin, track}; use support::harness::{MockConnectOptions, MockPair, connect_mock}; use support::mock::MockStats; +/// A millisecond track: the frames written here carry timestamps. +fn timed() -> track::Info { + track::Info::default().with_timescale(moq_net::Timescale::MILLI) +} + /// The lite draft production negotiates, the next one (opt-in), and the newest /// IETF draft. const VERSIONS: [&str; 3] = ["moq-lite-06", "moq-lite-07-wip", "moq-transport-22"]; @@ -75,8 +98,17 @@ struct Shape { watch: usize, /// Payload bytes per frame. frame: usize, + /// Frames in a complete group. + group_frames: usize, /// Write one frame per round, as a live source does, instead of a whole group. paced: bool, + /// Whole groups each watched broadcast writes per round, so a serve loop finds a + /// backlog ready: the group rate a round stands in for. + burst: usize, + /// Cached groups per track before a join, whose older groups the live budget skips. + history: usize, + /// Read each group via a fresh FETCH rather than a live subscription. + fetch: bool, } impl Shape { @@ -89,33 +121,63 @@ impl Shape { viewers: 16, watch: 1, frame: 64, + group_frames: FRAMES, paced: false, + burst: 1, + history: 1, + fetch: false, }; + /// What each viewer subscribes with: a burst is a backlog, which the default + /// subscription writes off as late. + fn subscription(&self) -> Option { + (self.burst > 1).then(|| track::Subscription::default().with_max_age(Duration::from_secs(30))) + } + fn total(&self) -> usize { self.publishers * self.broadcasts } /// Frames each group gets per round. fn frames(&self) -> usize { - if self.paced { 1 } else { FRAMES } + if self.paced { 1 } else { self.group_frames } } /// Payload bytes every viewer reads in one round. fn expected(&self) -> usize { - self.viewers * self.watch * self.frames() * self.frame + self.viewers * self.watch * self.burst * self.frames() * self.frame } fn id(&self, version: &str) -> String { format!( - "{version}/relays={}/publishers={}/broadcasts={}/viewers={}/watch={}/frame={}{}", + "{version}/relays={}/publishers={}/broadcasts={}/viewers={}/watch={}/frame={}{}{}{}{}{}", self.relays, self.publishers, self.broadcasts, self.viewers, self.watch, self.frame, - if self.paced { "/paced" } else { "" }, + if self.group_frames != FRAMES { + format!("/frames={}", self.group_frames) + } else { + String::new() + }, + if self.paced { + "/paced".to_string() + } else { + String::new() + }, + if self.burst > 1 { + format!("/burst={}", self.burst) + } else { + String::new() + }, + if self.fetch { "/fetch" } else { "" }, + if self.history > 1 { + format!("/history={}", self.history) + } else { + String::new() + }, ) } } @@ -163,12 +225,13 @@ fn runtime() -> tokio::runtime::Runtime { .unwrap() } -fn write_group(track: &track::Producer, payload: &Bytes, frames: usize) { +fn write_group(track: &track::Producer, payload: &Bytes, frames: usize) -> u64 { let mut group = track.append_group().unwrap(); for _ in 0..frames { group.write_frame(Timestamp::ZERO, payload.clone()).unwrap(); } group.finish().unwrap(); + group.sequence } /// Read one whole group, returning its payload bytes so skipped work can't look @@ -197,6 +260,8 @@ struct Cluster { _origins: Vec, _pairs: Vec, next_hop: u64, + /// The relay and publisher caches; each viewer owns its own. + pools: Vec, } impl Cluster { @@ -209,11 +274,13 @@ impl Cluster { _origins: Vec::new(), _pairs: Vec::new(), next_hop: 0, + pools: Vec::new(), }; for _ in 0..shape.relays { - let relay = this.origin(RELAY_CACHE); + let (relay, pool) = this.origin(RELAY_CACHE); this.relays.push(relay); + this.pools.push(pool); } // Peer every pair the way a cluster dial does: each side publishes its @@ -232,12 +299,13 @@ impl Cluster { } for publisher in 0..shape.publishers { - let origin = this.origin(CLIENT_CACHE); + let (origin, pool) = this.origin(CLIENT_CACHE); + this.pools.push(pool); for index in 0..shape.broadcasts { let broadcast = origin .publish(path(publisher * shape.broadcasts + index), Default::default()) .unwrap(); - this.tracks.push(broadcast.create_track(TRACK, None).unwrap()); + this.tracks.push(broadcast.create_track(TRACK, timed()).unwrap()); this._broadcasts.push(broadcast); } @@ -251,18 +319,23 @@ impl Cluster { this } - fn origin(&mut self, capacity: u64) -> origin::Producer { + fn origin(&mut self, capacity: u64) -> (origin::Producer, cache::Pool) { self.next_hop += 1; let mut config = origin::Config::new(Hop::new(self.next_hop).unwrap()); - config.pool = cache::Pool::new(cache::Config::default().with_capacity(capacity)); + let pool = cache::Pool::new(if std::env::var_os("SESSION_EXPIRY_ONLY").is_some() { + cache::Config::default().with_expiry(cache::DEFAULT_EXPIRY) + } else { + cache::Config::default().with_capacity(capacity) + }); + config.pool = pool.clone(); let (producer, driver) = origin::Producer::new(config); tokio::spawn(support::harness::run(driver)); - producer + (producer, pool) } /// Connect a viewer session to `relay`. async fn join(&mut self, relay: usize) -> Viewer { - let origin = self.origin(CLIENT_CACHE); + let (origin, pool) = self.origin(CLIENT_CACHE); let mut options = MockConnectOptions::new(self.version); // Hidden included, as the dash's `.stats/` scope sees its subtree. options.server_publish = Some(self.relays[relay].consume().with_hidden(true)); @@ -271,25 +344,35 @@ impl Cluster { Viewer { subscribers: Vec::new(), + fetches: Vec::new(), _pair: pair, origin, + pool, } } } struct Viewer { subscribers: Vec, + fetches: Vec<(usize, track::Consumer)>, _pair: MockPair, origin: origin::Producer, + pool: cache::Pool, } impl Viewer { /// Resolve `path` and subscribe to each of its `tracks`. - async fn watch(&mut self, path: &str, tracks: &[impl AsRef]) { + async fn watch(&mut self, path: &str, tracks: &[impl AsRef], subscription: Option) { let broadcast = self.origin.consume().routed_broadcast(path).await.unwrap(); for track in tracks { - self.subscribers - .push(broadcast.track(track.as_ref()).unwrap().subscribe(None).await.unwrap()); + self.subscribers.push( + broadcast + .track(track.as_ref()) + .unwrap() + .subscribe(subscription.clone()) + .await + .unwrap(), + ); } } } @@ -304,10 +387,13 @@ struct Room { shape: Shape, /// Frames written so far, so paced rounds know where each group starts and ends. written: usize, + /// Groups finished so far across every watched broadcast. + groups: usize, /// The group each watched broadcast is writing, when paced. writing: Vec>, /// The group each viewer's subscriber is reading, when paced, flattened in viewer order. reading: Vec>, + cache_now: moq_net::time::Instant, } impl Room { @@ -316,6 +402,11 @@ impl Room { shape.watch <= shape.total(), "a viewer can't watch more broadcasts than exist" ); + assert!(!shape.paced || shape.burst == 1, "a paced round writes one frame"); + assert!( + !shape.fetch || (!shape.paced && shape.burst == 1), + "fetch rounds read one finished group" + ); let mut cluster = Cluster::new(version, shape).await; // Viewer v watches a contiguous window starting at v * watch, so viewers @@ -333,7 +424,12 @@ impl Room { // publisher it watches whenever there is more than one. let mut viewer = cluster.join((viewer + 1) % shape.relays).await; for broadcast in broadcasts { - viewer.watch(&path(broadcast), &[TRACK]).await; + if shape.fetch { + let source = viewer.origin.consume().routed_broadcast(path(broadcast)).await.unwrap(); + viewer.fetches.push((broadcast, source.track(TRACK).unwrap())); + } else { + viewer.watch(&path(broadcast), &[TRACK], shape.subscription()).await; + } } viewers.push(viewer); } @@ -350,38 +446,88 @@ impl Room { payload: Bytes::from(vec![0; shape.frame]), shape, written: 0, + groups: 0, writing: Vec::new(), reading: Vec::new(), + cache_now: moq_net::time::Instant::now(), }; room.writing.resize_with(room.watched.len(), || None); room.reading.resize_with(shape.viewers * shape.watch, || None); + + // First measure the complete batch without eviction. The delivery benchmark must + // fit its working set; cache-pressure loss is a different workload. + let capacities: Vec<_> = room.pools().map(cache::Pool::capacity).collect(); + for pool in room.pools() { + pool.resize(None); + } // Warm every path so the timed rounds skip first-group setup. - for _ in 0..FRAMES / shape.frames() { + for _ in 0..shape.group_frames / shape.frames() { + room.tick(); + room.round().await; + } + for (pool, capacity) in room.pools().zip(capacities) { + // Keep the old and current batch resident during the handover, including + // bookkeeping rather than estimating a group from payload bytes alone. + pool.resize(capacity.map(|capacity| capacity.max(pool.used() * 2))); + } + // Exercise turnover as well as the empty cache, including Criterion's one-pass + // nightly smoke mode. + for _ in 0..4 { + room.tick(); room.round().await; } room } + fn pools(&self) -> impl Iterator { + let viewers = self.viewers.iter().map(|viewer| &viewer.pool); + self.cluster.pools.iter().chain(viewers) + } + + /// Unpaced rounds otherwise collapse many complete batches into one cache tick, + /// where yesterday's read outranks today's unread write. Sample a new recency + /// epoch per round without sleeping or changing protocol time. + fn tick(&mut self) { + self.cache_now += Duration::from_secs(1); + for pool in self.pools() { + pool.gc(self.cache_now); + } + } + async fn round(&mut self) { if self.shape.paced { return self.paced_round().await; } + let mut newest = vec![0; self.cluster.tracks.len()]; for &broadcast in &self.watched { - write_group(&self.cluster.tracks[broadcast], &self.payload, FRAMES); + for _ in 0..self.shape.burst { + newest[broadcast] = + write_group(&self.cluster.tracks[broadcast], &self.payload, self.shape.group_frames); + } } let mut bytes = 0; - for viewer in &mut self.viewers { - for subscriber in &mut viewer.subscribers { - bytes += read_group(subscriber).await; + // Round-robin, so no subscription's backlog waits on another's being read. + for _ in 0..self.shape.burst { + for viewer in &mut self.viewers { + for (broadcast, track) in &viewer.fetches { + let mut group = track.fetch_group(newest[*broadcast], None).await.unwrap(); + while let Some(frame) = group.read_frame().await.unwrap() { + bytes += frame.payload.len(); + } + } + for subscriber in &mut viewer.subscribers { + bytes += read_group(subscriber).await; + } } } assert_eq!(bytes, self.shape.expected()); + self.groups += self.shape.burst * self.watched.len(); } /// One frame into every watched group and one frame out to every viewer, so each - /// hop serves a group frame by frame across [`FRAMES`] rounds. + /// hop serves a group frame by frame across its configured frame count. async fn paced_round(&mut self) { - let last = self.written % FRAMES == FRAMES - 1; + let last = self.written % self.shape.group_frames == self.shape.group_frames - 1; self.written += 1; for (writing, &broadcast) in self.writing.iter_mut().zip(&self.watched) { @@ -392,6 +538,9 @@ impl Room { *writing = None; } } + if last { + self.groups += self.watched.len(); + } let mut bytes = 0; let subscribers = self.viewers.iter_mut().flat_map(|viewer| &mut viewer.subscribers); @@ -409,12 +558,29 @@ impl Room { assert_eq!(bytes, self.shape.expected()); } + /// Time `iters` rounds, excluding each round's cache tick: its cost grows with the + /// number of pools, not with the delivery being measured. async fn measure(&mut self, iters: u64) -> Duration { - let start = Instant::now(); + let mut elapsed = Duration::ZERO; + let report = std::env::var_os("SESSION_EXPIRY_ONLY").is_some(); + let began = Instant::now(); + let mut next_report = Duration::from_secs(10); for _ in 0..iters { + self.tick(); + let start = Instant::now(); self.round().await; + elapsed += start.elapsed(); + if report && began.elapsed() >= next_report { + eprintln!( + "cache sample: elapsed={:.1}s groups={} charged={}", + began.elapsed().as_secs_f64(), + self.groups, + self.pools().map(cache::Pool::used).sum::() + ); + next_report += Duration::from_secs(10); + } } - start.elapsed() + elapsed } } @@ -437,6 +603,46 @@ fn delivery(c: &mut Criterion, name: &str, shapes: impl IntoIterator) { let rt = runtime(); let shapes: Vec<_> = shapes.into_iter().collect(); @@ -450,7 +656,15 @@ fn join(c: &mut Criterion, name: &str, shapes: impl IntoIterator) let cluster = Cluster::new(version, *shape).await; let payload = Bytes::from(vec![0; shape.frame]); for track in &cluster.tracks { - write_group(track, &payload, FRAMES); + for sequence in 0..shape.history { + let mut group = track.append_group().unwrap(); + for _ in 0..shape.group_frames { + group + .write_frame(Timestamp::from_millis(sequence as u64).unwrap(), payload.clone()) + .unwrap(); + } + group.finish().unwrap(); + } } cluster }) @@ -468,10 +682,10 @@ fn join(c: &mut Criterion, name: &str, shapes: impl IntoIterator) let broadcast = hosted[iter % hosted.len()]; let start = Instant::now(); let mut viewer = cluster.join(relay).await; - viewer.watch(&path(broadcast), &[TRACK]).await; + viewer.watch(&path(broadcast), &[TRACK], None).await; let bytes = read_group(&mut viewer.subscribers[0]).await; elapsed += start.elapsed(); - assert_eq!(bytes, FRAMES * shape.frame); + assert_eq!(bytes, shape.group_frames * shape.frame); // Teardown is a leaving viewer's cost, not a joining one's. drop(viewer); tokio::task::yield_now().await; @@ -547,7 +761,7 @@ impl DashRoom { let path = format!(".stats/{project}/node/{node}"); let broadcast = cluster.relays[node].publish(path.as_str(), Default::default()).unwrap(); for name in &names { - tracks.push(broadcast.create_track(name.as_str(), None).unwrap()); + tracks.push(broadcast.create_track(name.as_str(), timed()).unwrap()); } broadcasts.push(broadcast); paths.push(path); @@ -556,7 +770,7 @@ impl DashRoom { let mut dash = cluster.join(0).await; for path in &paths { - dash.watch(path, &names).await; + dash.watch(path, &names, None).await; } let mut room = Self { @@ -630,7 +844,7 @@ fn allocations() { rt.block_on(room.measure(ROUNDS as u64)); COUNTING.store(false, Ordering::Relaxed); - let groups = ROUNDS * shape.viewers * shape.watch * shape.frames() / FRAMES; + let groups = ROUNDS * shape.viewers * shape.watch * shape.frames() / shape.group_frames; let allocations = ALLOCATIONS.load(Ordering::Relaxed) as f64 / groups as f64; println!("{}: {allocations:.1} allocations per viewer-group", shape.id(version)); } @@ -638,6 +852,176 @@ fn allocations() { } } +/// Versions for the churn sweep: the one draft with a MAX_REQUEST_ID window and the +/// newest draft without it. Lite is left out: a reopened subscription there never +/// reaches the relay, so there is no request round trip to time. +const CHURN_VERSIONS: [&str; 2] = ["moq-transport-14", "moq-transport-22"]; + +/// Sessions on one relay, each holding `held` subscriptions open while it churns one more. +#[derive(Clone, Copy)] +struct Churn { + sessions: usize, + held: usize, +} + +impl Churn { + fn id(&self, version: &str) -> String { + format!("{version}/sessions={}/held={}", self.sessions, self.held) + } +} + +/// One relay publishing a broadcast with `held` shared tracks plus one churn track per session. +struct ChurnRoom { + _cluster: Cluster, + _held: Vec, + _viewers: Vec, + /// Each session's view of the broadcast, to open the churned subscription on. + broadcasts: Vec, + /// The relay's side of each session's churn track, to wait out its close. + churn: Vec, +} + +impl ChurnRoom { + async fn new(version: &str, shape: Churn) -> Self { + let mut cluster = Cluster::new( + version, + Shape { + relays: 1, + publishers: 1, + viewers: 0, + ..Shape::BASE + }, + ) + .await; + let published = &cluster._broadcasts[0]; + let held: Vec<_> = (0..shape.held).map(|i| format!("held{i}")).collect(); + let tracks: Vec<_> = held + .iter() + .map(|name| published.create_track(name.as_str(), None).unwrap()) + .collect(); + let churn: Vec<_> = (0..shape.sessions) + .map(|i| published.create_track(format!("churn{i}"), None).unwrap()) + .collect(); + + let mut viewers = Vec::new(); + let mut broadcasts = Vec::new(); + for _ in 0..shape.sessions { + let mut viewer = cluster.join(0).await; + viewer.watch(&path(0), &held, None).await; + broadcasts.push(viewer.origin.consume().routed_broadcast(path(0)).await.unwrap()); + viewers.push(viewer); + } + Self { + _cluster: cluster, + _held: tracks, + _viewers: viewers, + broadcasts, + churn, + } + } + + /// Open and close one subscription per iteration, round-robin over the sessions. + async fn measure(&self, iters: u64) -> Duration { + let start = Instant::now(); + for iter in 0..iters as usize { + let session = iter % self.broadcasts.len(); + let track = self.broadcasts[session].track(&format!("churn{session}")).unwrap(); + let subscriber = track.subscribe(None).await.unwrap(); + // The relay saw the request, rather than a cached track answering it locally. + self.churn[session].demand().used().await.unwrap(); + drop(subscriber); + drop(track); + self.churn[session].demand().unused().await.unwrap(); + } + start.elapsed() + } +} + +fn churn(c: &mut Criterion, name: &str, shapes: impl IntoIterator) { + let rt = runtime(); + let shapes: Vec<_> = shapes.into_iter().collect(); + let mut group = c.benchmark_group(format!("session_churn_{name}")); + group.throughput(Throughput::Elements(1)); + for version in CHURN_VERSIONS { + for shape in &shapes { + let mut room = None; + group.bench_function(BenchmarkId::from_parameter(shape.id(version)), |b| { + let room = room.get_or_insert_with(|| rt.block_on(ChurnRoom::new(version, *shape))); + b.iter_custom(|iters| rt.block_on(room.measure(iters))) + }); + } + } + group.finish(); +} + +/// Idle sessions to one relay, holding PROBE or their local transport estimate. +struct IdleRoom { + _viewers: Vec<(MockPair, Vec)>, +} + +impl IdleRoom { + async fn new(version: &str, sessions: usize, consumers: usize) -> Self { + let (relay, driver) = origin::Producer::new(origin::Config::new(Hop::new(1).unwrap())); + tokio::spawn(support::harness::run(driver)); + let mut viewers = Vec::new(); + for _ in 0..sessions { + let mut options = MockConnectOptions::new(version.parse().unwrap()); + options.server_publish = Some(relay.consume()); + // A QUIC backend reports both; Lite serves PROBE, IETF samples on demand. + options.stats = MockStats { + estimated_send_rate: Some(1_000_000), + rtt: Some(Duration::from_millis(40)), + }; + let pair = connect_mock(options).await; + let mut probes = Vec::new(); + for _ in 0..consumers { + let mut probe = if version.parse::().unwrap().is_lite() { + pair.client.recv_bandwidth().unwrap() + } else { + pair.server.send_bandwidth().unwrap() + }; + probe.changed().await.unwrap(); + probes.push(probe); + } + viewers.push((pair, probes)); + } + Self { _viewers: viewers } + } + + /// Let a second of protocol time pass per iteration. + async fn measure(&self, iters: u64) -> Duration { + let start = Instant::now(); + for _ in 0..iters { + tokio::time::sleep(Duration::from_secs(1)).await; + } + start.elapsed() + } +} + +fn idle(c: &mut Criterion) { + let mut group = c.benchmark_group("session_idle"); + for version in ["moq-lite-06", "moq-transport-16", "moq-transport-22"] { + for sessions in [1, 64, 1024] { + for consumers in [0, 1, 8] { + group.throughput(Throughput::Elements(sessions as u64)); + // Its own runtime, so its paused clock and sessions end with the shape. + let rt = runtime(); + let mut room = None; + group.bench_function(format!("{version}/sessions={sessions}/consumers={consumers}"), |b| { + let room = room.get_or_insert_with(|| { + rt.block_on(async { + tokio::time::pause(); + IdleRoom::new(version, sessions, consumers).await + }) + }); + b.iter_custom(|iters| rt.block_on(room.measure(iters))) + }); + } + } + } + group.finish(); +} + /// Withdraw a batch across a full mesh, sweeping both peers and prefixes. /// Paused protocol time lets the quiet-point timeout drain pending work without /// including a real wait in the measured duration. @@ -677,7 +1061,7 @@ fn withdrawal(c: &mut Criterion) { while let Ok(Some(update)) = tokio::time::timeout(Duration::from_secs(1), announced.next()).await { - assert_eq!(update.kind, moq_net::announce::Kind::Retracted); + assert_eq!(update.kind, moq_net::announce::Kind::Retracted, "{update:?}"); retracted += 1; } let elapsed = start.elapsed(); @@ -747,6 +1131,20 @@ fn session(c: &mut Criterion) { }), ); delivery(c, "frame", [64, 1024, 16 * 1024].map(|frame| Shape { frame, ..base })); + // One session serving a backlog on each of its subscriptions: the serve loop that a + // `kio::coop::Budget` bounds per poll. + delivery( + c, + "burst", + [(1, 1), (1, 8), (1, 64), (8, 1), (8, 8), (16, 16), (16, 64)].map(|(watch, burst)| Shape { + publishers: 1, + broadcasts: 16, + viewers: 1, + watch, + burst, + ..base + }), + ); delivery( c, "relays", @@ -760,14 +1158,40 @@ fn session(c: &mut Criterion) { join( c, "broadcasts", - [1, 64, 1024].map(|broadcasts| Shape { - publishers: 1, - broadcasts, - ..base + [1, 2, 8].into_iter().flat_map(|relays| { + [1, 64, 1024].map(move |broadcasts| Shape { + publishers: 1, + broadcasts, + relays, + ..base + }) + }), + ); + // A live join skips a cached prefix. Sweep publishers against that prefix so + // resolving one start cannot hide a scan of unrelated publishers or group IDs. + join( + c, + "stale", + [1, 8, 32].into_iter().flat_map(|publishers| { + [1, 64, 1024].map(move |history| Shape { + publishers, + history, + ..base + }) }), ); + join(c, "relays", [1, 2, 8].map(|relays| Shape { relays, ..base })); + churn(c, "sessions", [1, 16, 128].map(|sessions| Churn { sessions, held: 16 })); + churn( + c, + "held", + [1, 4, 16] + .into_iter() + .flat_map(|sessions| [1, 64, 1024].map(move |held| Churn { sessions, held })), + ); + // Production (2026-09): ~34 nodes x >=5 projects x ~24 tracks. let prod = Dash { relays: 35, @@ -810,61 +1234,5 @@ fn session(c: &mut Criterion) { ); } -/// Viewer sessions to one relay, each holding a PROBE stream and otherwise idle. -struct IdleRoom { - _viewers: Vec<(MockPair, moq_net::bandwidth::Consumer)>, -} - -impl IdleRoom { - async fn new(sessions: usize) -> Self { - let (relay, driver) = origin::Producer::new(origin::Config::new(Hop::new(1).unwrap())); - tokio::spawn(support::harness::run(driver)); - let mut viewers = Vec::new(); - for _ in 0..sessions { - let mut options = MockConnectOptions::new("moq-lite-06".parse().unwrap()); - options.server_publish = Some(relay.consume()); - // A QUIC backend reports both, so the relay serves PROBE. - options.stats = MockStats { - estimated_send_rate: Some(1_000_000), - rtt: Some(Duration::from_millis(40)), - }; - let pair = connect_mock(options).await; - let mut probe = pair.client.recv_bandwidth().unwrap(); - probe.changed().await.unwrap(); - viewers.push((pair, probe)); - } - Self { _viewers: viewers } - } - - /// Let a second of protocol time pass per iteration. - async fn measure(&self, iters: u64) -> Duration { - let start = Instant::now(); - for _ in 0..iters { - tokio::time::sleep(Duration::from_secs(1)).await; - } - start.elapsed() - } -} - -fn idle(c: &mut Criterion) { - let mut group = c.benchmark_group("session_idle"); - for sessions in [1, 64, 1024] { - group.throughput(Throughput::Elements(sessions as u64)); - // Its own runtime, so its paused clock and its sessions end with the shape. - let rt = runtime(); - let mut room = None; - group.bench_function(BenchmarkId::from_parameter(format!("sessions={sessions}")), |b| { - let room = room.get_or_insert_with(|| { - rt.block_on(async { - tokio::time::pause(); - IdleRoom::new(sessions).await - }) - }); - b.iter_custom(|iters| rt.block_on(room.measure(iters))) - }); - } - group.finish(); -} - -criterion_group!(benches, session, withdrawal, idle); +criterion_group!(benches, session, group_cost, fetch_frames, withdrawal, idle); criterion_main!(benches); diff --git a/rs/moq-net/benches/stats.rs b/rs/moq-net/benches/stats.rs index 299b7fe9d7..cd9b61d380 100644 --- a/rs/moq-net/benches/stats.rs +++ b/rs/moq-net/benches/stats.rs @@ -64,5 +64,27 @@ fn report(c: &mut Criterion) { group.finish(); } -criterion_group!(benches, report); +fn session_report(c: &mut Criterion) { + let mut group = c.benchmark_group("stats/session-report"); + for labels in [1, 100, 1_000] { + for sessions in [1, 10] { + let registry = stats::Registry::new(stats::Config::new()); + let handle = registry.tier(stats::Tier::default()); + let contexts: Vec<_> = (0..labels) + .flat_map(|label| { + let handle = handle.clone(); + (0..sessions).map(move |_| handle.session(format!("project/key/{label}"))) + }) + .collect(); + let mut report = stats::Report::default(); + group.bench_function(BenchmarkId::from_parameter(format!("{labels}x{sessions}")), |b| { + b.iter(|| registry.report(&mut report)); + }); + std::hint::black_box(contexts); + } + } + group.finish(); +} + +criterion_group!(benches, report, session_report); criterion_main!(benches); diff --git a/rs/moq-net/benches/track.rs b/rs/moq-net/benches/track.rs index 637157a266..f19888672b 100644 --- a/rs/moq-net/benches/track.rs +++ b/rs/moq-net/benches/track.rs @@ -7,6 +7,9 @@ //! `track_aborted_scan` isolates the other half of delivery: how much a cached //! prefix of aborted groups costs the scan that has to walk past it. //! +//! `track_gc` is the cache's clock-driven half: the per-poll [`cache::Pool::gc`] +//! call every origin driver makes, and the due expiry pass it runs over every track. +//! //! `track_subscriber_churn` measures viewers joining and leaving beside steady ones: //! each departure wakes the producer's aggregate poll, which walks the subscription //! list, so departed entries left in the list show up as a slope over churn. @@ -16,17 +19,54 @@ //! `track_subscriber_join` measures a burst of viewers joining one track: each join //! registers in the subscription list, so any per-join walk of it shows up as a slope. //! +//! `track_parked_read` appends past a backlog of reads parked on their drift budget: +//! each append must wake only the reads it can expire, so waking the backlog shows up +//! as a slope over it. +//! +//! `track_sparse_read` hands out groups from a feed with every other sequence missing, +//! holding a fixed number of gaps the reader still remembers: each hand-out must cost the +//! same however many gaps it holds, so a walk over them shows up as a slope. +//! +//! `track_gap_flush` forgets every remembered gap in one hand-out, each below an aborted +//! group still cached: the successor they share must be found once, not once per gap. +//! +//! `track_empty_run` drains a run of groups that finished without a frame across readers, +//! with the groups after it past the budget or within it: each empty group is handed out +//! like any other, so a walk over the rest of the run per group shows up as a slope over it. +//! +//! `track_fanout_frame` writes frames into an open group while every subscriber is parked +//! on the next one, re-polling whichever reader the write woke: a frame wakes no track +//! reader, so waking them shows up as a slope over the subscribers. +//! +//! `track_range_demand` moves one reader between ranges nobody else holds, over readers +//! holding a given number of distinct overlapping ranges: both axes must stay flat, since +//! a change touches only the moved range, never the rest of the audience's. +//! //! Run with `cargo bench -p moq-net --bench track`. +// Custom timings measure the real time the work took, not the ambient clock. +#![allow(clippy::disallowed_methods)] + use std::hint::black_box; use std::sync::{Arc, Barrier}; use std::task::Poll; -use std::time::{Duration, Instant}; +use std::time::{Duration, Instant as WallInstant}; + +use moq_net::time::Instant; use bytes::Bytes; use criterion::{BatchSize, BenchmarkId, Criterion, Throughput, criterion_group, criterion_main}; use moq_net::{Error, Timestamp, broadcast, cache, track}; +/// A millisecond track: the frames written here carry timestamps. +/// +/// Unbounded publisher retention, so a subscriber's budget is never clamped below what it asks. +fn timed() -> track::Info { + track::Info::default() + .with_timescale(moq_net::Timescale::MILLI) + .with_max_age(Duration::MAX) +} + /// Fanout sizes spanning a direct viewer, a small room, and a large room. const FANOUT: [usize; 4] = [1, 8, 64, 512]; @@ -40,20 +80,23 @@ const PAYLOAD: usize = 64; const CACHE_CAPACITY: u64 = 64 * 1024; /// Steady viewers that stay subscribed while others churn. -const STEADY: [usize; 3] = [1, 8, 64]; +const STEADY: [usize; 4] = [1, 100, 1000, 10_000]; /// Viewers that join and leave, one at a time, during one measured run. -const CHURN: [usize; 3] = [8, 64, 512]; - -/// Viewers that join at once and stay, during one measured run. -const JOIN: [usize; 3] = [64, 1024, 16384]; +const CHURN: [usize; 3] = [1, 8, 64]; /// Viewers that joined and left before the churn, leaving the list room for that many. const PEAK: [usize; 3] = [0, 1024, 16384]; +/// Distinct ranges the steady readers hold between them, capped at the readers. +const DISTINCT: [usize; 4] = [1, 100, 1000, 10_000]; + /// Concurrent publishers sharing one relay-style cache pool. const WRITERS: [usize; 4] = [1, 2, 4, 8]; +/// Tracks sharing one expiring pool: the table a due collection pass walks. +const SWEPT: [usize; 4] = [1, 64, 1_024, 16_384]; + /// Keeps the ownership chain alive around the track and its subscribers. struct Fanout { _broadcast: broadcast::Producer, @@ -71,7 +114,7 @@ impl Fanout { .with_expiry(cache::DEFAULT_EXPIRY); info.pool = cache::Pool::new(config); let broadcast = broadcast::Producer::new(info); - let track = broadcast.create_track("bench", None).unwrap(); + let track = broadcast.create_track("bench", timed()).unwrap(); let mut subscribers: Vec<_> = (0..subscribers).map(|_| track.subscribe(None)).collect(); let waiters: Vec<_> = (0..subscribers.len()).map(|_| kio::Waiter::noop()).collect(); @@ -144,6 +187,45 @@ impl AbortedScan { } } +/// Live tracks with one cached group each, sharing an expiring pool. +/// +/// Each track's only group is its live latest, which expiry never reclaims, so +/// every pass walks the same table instead of emptying it. +struct Swept { + _broadcast: broadcast::Producer, + _tracks: Vec, + pool: cache::Pool, + now: Instant, +} + +impl Swept { + fn new(tracks: usize) -> Self { + let mut info = broadcast::Info::default(); + info.pool = cache::Pool::new(cache::Config::default().with_expiry(cache::DEFAULT_EXPIRY)); + let pool = info.pool.clone(); + let broadcast = broadcast::Producer::new(info); + let payload = Bytes::from_static(&[0; PAYLOAD]); + let tracks = (0..tracks) + .map(|i| { + let track = broadcast.create_track(format!("bench{i}"), timed()).unwrap(); + let mut group = track.append_group().unwrap(); + group.write_frame(Timestamp::ZERO, payload.clone()).unwrap(); + group.finish().unwrap(); + track + }) + .collect(); + // The first call starts the pool's clock and runs its first pass. + let now = Instant::now(); + pool.gc(now); + Self { + _broadcast: broadcast, + _tracks: tracks, + pool, + now, + } + } +} + /// Steady viewers on one track, with the producer's aggregate poll caught up. struct Churn { _broadcast: broadcast::Producer, @@ -182,6 +264,30 @@ impl Churn { } } +fn bench_gc(c: &mut Criterion) { + let mut group = c.benchmark_group("track_gc"); + for tracks in SWEPT { + // Between passes, which is every origin poll: should stay flat as tracks grow. + group.throughput(Throughput::Elements(1)); + group.bench_with_input(BenchmarkId::new("idle", tracks), &tracks, |b, &tracks| { + let swept = Swept::new(tracks); + b.iter(|| black_box(swept.pool.gc(swept.now))); + }); + // A due pass, stepping one sweep interval per call. + group.throughput(Throughput::Elements(tracks as u64)); + group.bench_with_input(BenchmarkId::new("sweep", tracks), &tracks, |b, &tracks| { + let swept = Swept::new(tracks); + let interval = cache::DEFAULT_EXPIRY / 2; + let mut now = swept.now; + b.iter(|| { + now += interval; + black_box(swept.pool.gc(now)) + }); + }); + } + group.finish(); +} + fn bench_fanout(c: &mut Criterion) { let mut group = c.benchmark_group("track_fanout_group"); for subscribers in FANOUT { @@ -198,6 +304,77 @@ fn bench_fanout(c: &mut Criterion) { group.finish(); } +/// Frames written per group in `track_fanout_frame`, after which the group rolls and every +/// subscriber takes the next one: enough that the roll's cost per frame stays small. +const FRAMES_PER_GROUP: usize = 256; + +/// Queues its reader's index when woken, so only woken readers are re-polled. +struct Fired { + index: usize, + queue: Arc>>, +} + +impl futures::task::ArcWake for Fired { + fn wake_by_ref(fired: &Arc) { + fired.queue.lock().unwrap().push(fired.index); + } +} + +fn bench_fanout_frame(c: &mut Criterion) { + let mut group = c.benchmark_group("track_fanout_frame"); + for subscribers in FANOUT { + group.bench_with_input( + BenchmarkId::from_parameter(subscribers), + &subscribers, + |b, &subscribers| { + let mut fanout = Fanout::new(subscribers); + let queue = Arc::new(std::sync::Mutex::new(Vec::new())); + let waiters: Vec<_> = (0..subscribers) + .map(|index| { + let fired = Arc::new(Fired { + index, + queue: queue.clone(), + }); + kio::Waiter::new(futures::task::waker(fired)) + }) + .collect(); + let mut open: Option = None; + let mut written = FRAMES_PER_GROUP; + // Only the frame write and the re-polls it causes are timed, not the group roll. + b.iter_custom(|iterations| { + let mut elapsed = Duration::ZERO; + for _ in 0..iterations { + if written == FRAMES_PER_GROUP { + if let Some(open) = open.take() { + open.finish().unwrap(); + } + open = Some(fanout.track.append_group().unwrap()); + written = 0; + for (subscriber, waiter) in fanout.subscribers.iter_mut().zip(&waiters) { + assert!(matches!(subscriber.poll_recv_group(waiter), Poll::Ready(Ok(Some(_))))); + assert!(subscriber.poll_recv_group(waiter).is_pending()); + } + queue.lock().unwrap().clear(); + } + let group = open.as_mut().unwrap(); + let start = Instant::now(); + group.write_frame(Timestamp::ZERO, fanout.payload.clone()).unwrap(); + // A woken reader re-polls, as its task would. + let woken = std::mem::take(&mut *queue.lock().unwrap()); + for index in woken { + black_box(fanout.subscribers[index].poll_recv_group(&waiters[index]).is_pending()); + } + elapsed += start.elapsed(); + written += 1; + } + elapsed + }); + }, + ); + } + group.finish(); +} + /// Write `iterations` single-frame groups across independent tracks sharing one pool. fn parallel_write(pool: &cache::Pool, writers: usize, iterations: u64) -> Duration { let barrier = Arc::new(Barrier::new(writers + 1)); @@ -211,7 +388,7 @@ fn parallel_write(pool: &cache::Pool, writers: usize, iterations: u64) -> Durati let mut info = broadcast::Info::default(); info.pool = pool; let broadcast = broadcast::Producer::new(info); - let track = broadcast.create_track("bench", None).unwrap(); + let track = broadcast.create_track("bench", timed()).unwrap(); let payload = Bytes::from_static(&[0; PAYLOAD]); // Arrive once setup is done, then park until the main thread has stamped the clock. barrier.wait(); @@ -228,7 +405,7 @@ fn parallel_write(pool: &cache::Pool, writers: usize, iterations: u64) -> Durati // The first wait returns once every writer has built its producer, keeping setup out // of the interval. The second releases them, so no write lands before the stamp. barrier.wait(); - let start = Instant::now(); + let start = WallInstant::now(); barrier.wait(); for handle in handles { handle.join().unwrap(); @@ -274,11 +451,8 @@ fn bench_subscriber_churn(c: &mut Criterion) { BenchmarkId::new(format!("steady_{steady}"), churn), &churn, |b, &churn| { - b.iter_batched_ref( - || Churn::new(steady, 0), - |setup| setup.run(churn), - BatchSize::SmallInput, - ); + let mut setup = Churn::new(steady, 0); + b.iter(|| setup.run(churn)); }, ); } @@ -292,70 +466,461 @@ fn bench_subscriber_churn_after_peak(c: &mut Criterion) { group.throughput(Throughput::Elements(CHURN as u64)); for peak in PEAK { group.bench_with_input(BenchmarkId::from_parameter(peak), &peak, |b, &peak| { - b.iter_batched_ref(|| Churn::new(1, peak), |setup| setup.run(CHURN), BatchSize::SmallInput); + let mut setup = Churn::new(1, peak); + b.iter(|| setup.run(CHURN)); }); } group.finish(); } +/// One join and its aggregate poll, excluding the departure between iterations. fn bench_subscriber_join(c: &mut Criterion) { let mut group = c.benchmark_group("track_subscriber_join"); - for join in JOIN { - group.throughput(Throughput::Elements(join as u64)); - group.bench_with_input(BenchmarkId::from_parameter(join), &join, |b, &join| { - b.iter_batched( - || { - let broadcast = broadcast::Info::default().produce(); - let track = broadcast.create_track("bench", None).unwrap(); - (broadcast, track) - }, - |(broadcast, track)| { - let viewers: Vec<_> = (0..join).map(|_| track.subscribe(None)).collect(); - (broadcast, track, viewers) - }, - BatchSize::SmallInput, - ); + group.throughput(Throughput::Elements(1)); + for steady in STEADY { + group.bench_function(BenchmarkId::from_parameter(steady), |b| { + let mut setup = Churn::new(steady, 0); + b.iter_custom(|iterations| { + let mut elapsed = Duration::ZERO; + for _ in 0..iterations { + let start = Instant::now(); + let viewer = setup.track.subscribe(None); + assert!(black_box(setup.track.poll_subscription_changed(&setup.waiter)).is_pending()); + elapsed += start.elapsed(); + drop(viewer); + assert!(setup.track.poll_subscription_changed(&setup.waiter).is_pending()); + } + elapsed + }); }); } group.finish(); } -/// Parked reads re-judge their reach, including the successor's abort waiter. -/// Sweep readers and cached groups so added per-reader scans show up as a slope. -fn bench_parked_read(c: &mut Criterion) { - let mut group = c.benchmark_group("track_parked_read"); - for cached in [8, 64, 512] { - for readers in [1, 8, 64] { - let broadcast = broadcast::Info::default().produce(); - let track = broadcast.create_track("bench", None).unwrap(); - let mut head = track.append_group().unwrap(); - head.write_frame(Timestamp::ZERO, Bytes::from_static(b"head")).unwrap(); - for _ in 1..cached { - let mut next = track.append_group().unwrap(); - next.write_frame(Timestamp::ZERO, Bytes::from_static(b"next")).unwrap(); - next.finish().unwrap(); - } - let waiters: Vec<_> = (0..readers).map(|_| kio::Waiter::noop()).collect(); - let mut held: Vec<_> = waiters - .iter() - .map(|waiter| { - let mut sub = track.subscribe(track::Subscription::default().with_max_age(Duration::from_secs(1))); - let Poll::Ready(Ok(Some(mut head))) = sub.poll_recv_group(waiter) else { - panic!("head is cached"); - }; - assert!(matches!(head.poll_read_frame(waiter), Poll::Ready(Ok(Some(_))))); - head - }) - .collect(); - group.throughput(Throughput::Elements(readers as u64)); - group.bench_function(BenchmarkId::new(format!("cached_{cached}"), readers), |b| { +/// Preference changes and an extreme holder leaving, over the already held audience. +/// +/// Each held reader asks for distinct values, so the aggregate's tables grow with the audience. +fn bench_subscriber_preferences(c: &mut Criterion) { + let ordinary = track::Subscription::default().with_groups(100..200); + let held = |i: usize| { + let i = i as u64; + track::Subscription::default() + .with_priority((i % 255) as u8) + .with_max_age(Duration::from_micros(i)) + .with_groups(100 + i..200 + i) + }; + let extreme = track::Subscription::default() + .with_priority(255) + .with_max_age(Duration::from_secs(10)); + let mut group = c.benchmark_group("track_subscriber_preferences"); + for steady in STEADY { + for touched in CHURN { + group.throughput(Throughput::Elements(2 * touched as u64)); + group.bench_function(BenchmarkId::new(format!("update/{steady}"), touched), |b| { + let mut setup = Churn::new(steady, 0); + for (i, subscriber) in setup._steady.iter_mut().enumerate() { + subscriber.update(held(i)).unwrap(); + } + let mut viewer = setup.track.subscribe(ordinary.clone()); + assert!(setup.track.poll_subscription_changed(&setup.waiter).is_ready()); b.iter(|| { - for (head, waiter) in held.iter_mut().zip(&waiters) { - assert!(black_box(head.poll_read_frame(waiter)).is_pending()); + for _ in 0..touched { + viewer.update(extreme.clone()).unwrap(); + assert!(black_box(setup.track.poll_subscription_changed(&setup.waiter)).is_ready()); + viewer.update(ordinary.clone()).unwrap(); + assert!(black_box(setup.track.poll_subscription_changed(&setup.waiter)).is_ready()); } }); }); } + group.throughput(Throughput::Elements(1)); + group.bench_function(BenchmarkId::new("extreme_leave", steady), |b| { + let mut setup = Churn::new(steady, 0); + for (i, subscriber) in setup._steady.iter_mut().enumerate() { + subscriber.update(held(i)).unwrap(); + } + assert!(setup.track.poll_subscription_changed(&setup.waiter).is_ready()); + b.iter_custom(|iterations| { + let mut elapsed = Duration::ZERO; + for _ in 0..iterations { + let viewer = setup.track.subscribe(extreme.clone()); + assert!(setup.track.poll_subscription_changed(&setup.waiter).is_ready()); + let start = Instant::now(); + drop(viewer); + assert!(black_box(setup.track.poll_subscription_changed(&setup.waiter)).is_ready()); + elapsed += start.elapsed(); + } + elapsed + }); + }); + } + group.finish(); +} + +/// One reader moving between ranges of its own, over readers holding `distinct` ranges. +fn bench_range_demand(c: &mut Criterion) { + // Far past every held range, so each move adds one run and removes another. + let away = |offset: u64| { + let group = (1 << 40) + offset; + track::Subscription::default().with_groups(group..=group) + }; + let mut group = c.benchmark_group("track_range_demand"); + for steady in STEADY { + for distinct in DISTINCT.into_iter().filter(|&distinct| distinct <= steady) { + group.throughput(Throughput::Elements(2)); + group.bench_function(BenchmarkId::new(format!("{steady}r"), distinct), |b| { + let mut setup = Churn::new(steady, 0); + for (i, subscriber) in setup._steady.iter_mut().enumerate() { + // Overlapping, so the held ranges union into one run. + let start = (i % distinct) as u64 * 10; + subscriber + .update(track::Subscription::default().with_groups(start..start + 1000)) + .unwrap(); + } + let mut viewer = setup.track.subscribe(away(0)); + assert!(setup.track.poll_subscription_changed(&setup.waiter).is_ready()); + b.iter(|| { + viewer.update(away(2)).unwrap(); + assert!(black_box(setup.track.poll_subscription_changed(&setup.waiter)).is_ready()); + viewer.update(away(0)).unwrap(); + assert!(black_box(setup.track.poll_subscription_changed(&setup.waiter)).is_ready()); + }); + }); + } + } + group.finish(); +} + +/// Late joins scan a bounded datagram ring; sweep both the backlog and the joining audience. +fn bench_datagram_subscriber_join(c: &mut Criterion) { + let mut group = c.benchmark_group("track_datagram_subscriber_join"); + for buffered in [0, 1, 8, 64] { + for join in [1, 8, 64] { + group.throughput(Throughput::Elements(join as u64)); + group.bench_function(BenchmarkId::new(format!("{buffered}d"), join), |b| { + b.iter_batched( + || { + let broadcast = broadcast::Info::default().produce(); + let mut track = broadcast.create_track("bench", timed()).unwrap(); + for millis in 0..buffered { + track + .append_datagram(Timestamp::from_millis(millis).unwrap(), b"x".as_slice()) + .unwrap(); + } + (broadcast, track) + }, + |(broadcast, track)| { + let preferences = track::Subscription::default().with_max_age(Duration::from_millis(5)); + let viewers: Vec<_> = (0..join).map(|_| track.subscribe(preferences.clone())).collect(); + (broadcast, track, viewers) + }, + BatchSize::SmallInput, + ); + }); + } + } + group.finish(); +} + +/// Queues a parked read as woken, so the bench visits only what an append wakes. +struct Woken { + read: usize, + queue: Arc>>, +} + +impl std::task::Wake for Woken { + fn wake(self: Arc) { + self.wake_by_ref(); + } + + fn wake_by_ref(self: &Arc) { + self.queue.lock().unwrap().push(self.read); + } +} + +/// One append past a backlog of parked reads, one per group like a publisher's parked +/// group streams, re-polling whatever it wakes. Swept over the backlog so a per-append +/// cost that grows with the parked reads shows up as a slope. +fn bench_parked_read(c: &mut Criterion) { + const MICROS_PER_GROUP: u64 = 2500; + let mut group = c.benchmark_group("track_parked_read"); + group.throughput(Throughput::Elements(1)); + for parked in [8, 64, 512] { + group.bench_function(BenchmarkId::from_parameter(parked), |b| { + b.iter_custom(|iterations| { + let broadcast = broadcast::Info::default().produce(); + let track = broadcast.create_track("bench", timed()).unwrap(); + // Cover setup and every measured append so the same backlog stays parked, + // even when Criterion requests more iterations during warm-up. + let max_delay = Duration::from_micros( + iterations + .checked_add(parked as u64) + .unwrap() + .checked_mul(MICROS_PER_GROUP) + .unwrap(), + ); + let mut sub = track.subscribe(track::Subscription::default().with_max_age(max_delay)); + let mut micros = 0; + let mut open = Vec::with_capacity(parked); + let queue = Arc::new(std::sync::Mutex::new(Vec::new())); + let mut reads: Vec<_> = (0..parked) + .map(|read| { + let mut group = track.append_group().unwrap(); + group + .write_frame(Timestamp::from_micros(micros).unwrap(), Bytes::from_static(b"x")) + .unwrap(); + micros += MICROS_PER_GROUP; + // Left open, so its read parks instead of ending. + open.push(group); + + let woken = Woken { + read, + queue: queue.clone(), + }; + let waiter = kio::Waiter::new(std::task::Waker::from(Arc::new(woken))); + let Poll::Ready(Ok(Some(mut read))) = sub.poll_recv_group(&waiter) else { + panic!("the group is cached"); + }; + assert!(matches!(read.poll_read_frame(&waiter), Poll::Ready(Ok(Some(_))))); + assert!(read.poll_read_frame(&waiter).is_pending()); + (waiter, read) + }) + .collect(); + + let mut woken = Vec::new(); + let mut repoll = |reads: &mut Vec<(kio::Waiter, moq_net::group::Consumer)>| { + std::mem::swap(&mut woken, &mut *queue.lock().unwrap()); + for read in woken.drain(..) { + let (waiter, read) = &mut reads[read]; + assert!(black_box(read.poll_read_frame(waiter)).is_pending()); + } + }; + // Each append above woke the reads before it: park them again. + repoll(&mut reads); + + let start = WallInstant::now(); + for _ in 0..iterations { + let mut next = track.append_group().unwrap(); + next.write_frame(Timestamp::from_micros(micros).unwrap(), Bytes::from_static(b"x")) + .unwrap(); + next.finish().unwrap(); + micros += MICROS_PER_GROUP; + repoll(&mut reads); + } + start.elapsed() + }); + }); + } + group.finish(); +} + +/// Hand-outs from a sparse feed holding `gaps` remembered gaps, on a timed track whose +/// budget never writes a gap off. Each iteration hands out the next group, +/// opening a gap at the top, and the lowest missing one, closing the gap at the bottom, so +/// the count stays fixed. Swept over the gaps so a per-hand-out cost that grows with them +/// shows up as a slope. +fn bench_sparse_read(c: &mut Criterion) { + let mut group = c.benchmark_group("track_sparse_read"); + group.throughput(Throughput::Elements(2)); + for gaps in [16u64, 256, 4096] { + group.bench_function(BenchmarkId::new("timed", gaps), |b| { + b.iter_custom(|iterations| { + let broadcast = broadcast::Info::default().produce(); + let track = broadcast.create_track("bench", timed()).unwrap(); + let mut sub = track.subscribe(track::Subscription::default().with_max_age(Duration::from_secs(3600))); + // Microseconds, so no run lasts long enough to cross the budget. + let timestamp = |sequence: u64| Timestamp::from_micros(sequence).unwrap(); + let mut hand_out = |sequence: u64| { + let mut group = track.create_group(moq_net::group::Info { sequence }).unwrap(); + group + .write_frame(timestamp(sequence), Bytes::from_static(b"x")) + .unwrap(); + group.finish().unwrap(); + let Poll::Ready(Ok(Some(read))) = sub.poll_recv_group(&kio::Waiter::noop()) else { + panic!("the group is cached"); + }; + black_box(read); + }; + // Every odd sequence is missing until the bottom gap closes. + for gap in 0..gaps { + hand_out(gap * 2); + } + let start = WallInstant::now(); + for iteration in 0..iterations { + hand_out((gaps + iteration) * 2); + hand_out(iteration * 2 + 1); + } + start.elapsed() + }); + }); + } + group.finish(); +} + +/// A reader remembering `gaps` gaps, each below a handed-out group since aborted but still +/// cached, then a stamped survivor after them all. +struct GapFlush { + _broadcast: broadcast::Producer, + track: track::Producer, + sub: track::Subscriber, + gaps: u64, +} + +impl GapFlush { + fn new(gaps: u64) -> Self { + let broadcast = broadcast::Info::default().produce(); + let track = broadcast.create_track("bench", timed()).unwrap(); + let mut sub = track.subscribe(track::Subscription::default().with_max_age(Duration::from_secs(3600))); + let mut hand_out = |sequence: u64, millis: u64| { + let mut group = track.create_group(moq_net::group::Info { sequence }).unwrap(); + group + .write_frame(Timestamp::from_millis(millis).unwrap(), Bytes::from_static(b"x")) + .unwrap(); + let Poll::Ready(Ok(Some(read))) = sub.poll_recv_group(&kio::Waiter::noop()) else { + panic!("the group is cached"); + }; + black_box(read); + group + }; + // Every odd sequence is missing. + let handed: Vec<_> = (0..gaps).map(|gap| hand_out(gap * 2, gap)).collect(); + hand_out(gaps * 2, gaps).finish().unwrap(); + for group in handed { + group.abort(Error::Cancel).unwrap(); + } + Self { + _broadcast: broadcast, + track, + sub, + gaps, + } + } + + /// Hand out a group two hours on, which puts every remembered gap past the budget. + fn flush(&mut self) { + let mut group = self + .track + .create_group(moq_net::group::Info { + sequence: self.gaps * 2 + 1, + }) + .unwrap(); + group + .write_frame(Timestamp::from_millis(7_200_000).unwrap(), Bytes::from_static(b"x")) + .unwrap(); + let Poll::Ready(Ok(Some(read))) = self.sub.poll_recv_group(&kio::Waiter::noop()) else { + panic!("the group is cached"); + }; + black_box(read); + } +} + +/// One hand-out forgetting every remembered gap at once, each below a group that was +/// aborted but is still cached, swept over the gaps. The groups judged share one +/// successor past the aborted run, so finding it once per gap shows up as a slope. +fn bench_gap_flush(c: &mut Criterion) { + let mut group = c.benchmark_group("track_gap_flush"); + group.sample_size(10); + for gaps in [64u64, 512, 4096] { + group.throughput(Throughput::Elements(gaps)); + group.bench_function(BenchmarkId::from_parameter(gaps), |b| { + b.iter_batched_ref(|| GapFlush::new(gaps), GapFlush::flush, BatchSize::PerIteration); + }); + } + group.finish(); +} + +/// How the groups after an [`EmptyRun`] judge it. +#[derive(Clone, Copy)] +enum Run { + /// Two groups an hour apart in media time follow it, writing off its last group. + Stale, + /// One group follows it, keeping all of it within the budget. + Fresh, +} + +/// Readers on one track holding group 0, a run of groups that finished without a frame, +/// and the groups after it. +struct EmptyRun { + _broadcast: broadcast::Producer, + _track: track::Producer, + readers: Vec, + /// Groups each reader should be handed. + handed: usize, +} + +impl EmptyRun { + /// Media-time budgets span an hour, so setup on the real clock never ages a group out on + /// the wall clock. + fn new(run: u64, readers: usize, mode: Run) -> Self { + let broadcast = broadcast::Info::default().produce(); + let track = broadcast.create_track("bench", timed()).unwrap(); + let append = |millis: Option| { + let mut group = track.append_group().unwrap(); + if let Some(millis) = millis { + group + .write_frame(Timestamp::from_millis(millis).unwrap(), Bytes::from_static(b"x")) + .unwrap(); + } + group.finish().unwrap(); + }; + append(Some(0)); + for _ in 0..run { + append(None); + } + const HOUR: u64 = 3_600_000; + let after: &[u64] = match mode { + Run::Stale => &[2 * HOUR, 4 * HOUR], + Run::Fresh => &[5], + }; + for millis in after { + append(Some(*millis)); + } + let (budget, handed) = match mode { + // The run's last group is written off, but two groups follow it. + Run::Stale => (Duration::from_millis(HOUR), run as usize + 2), + Run::Fresh => (Duration::from_millis(HOUR), run as usize + 2), + }; + let budget = track::Subscription::default().with_max_age(budget); + Self { + readers: (0..readers).map(|_| track.subscribe(budget.clone())).collect(), + _broadcast: broadcast, + _track: track, + handed, + } + } + + fn drain(&mut self) { + for reader in &mut self.readers { + let mut handed = 0; + while let Poll::Ready(Ok(Some(group))) = reader.poll_recv_group(&kio::Waiter::noop()) { + black_box(group); + handed += 1; + } + assert_eq!(handed, self.handed); + } + } +} + +/// Readers draining a run of groups that finished without a frame, swept over the run and +/// the readers, with the groups after it past the budget (`stale`) or within it (`fresh`). +/// A reader's cost must stay linear in the run, so a walk over the rest of the run per +/// group shows up as a slope over it. +fn bench_empty_run(c: &mut Criterion) { + let mut group = c.benchmark_group("track_empty_run"); + group.sample_size(10); + for (name, mode) in [("stale", Run::Stale), ("fresh", Run::Fresh)] { + for readers in [1usize, 8, 64] { + for run in [64u64, 512, 4096] { + group.throughput(Throughput::Elements(run * readers as u64)); + group.bench_function(BenchmarkId::new(format!("{name}/{readers}"), run), |b| { + b.iter_batched_ref( + || EmptyRun::new(run, readers, mode), + EmptyRun::drain, + BatchSize::PerIteration, + ); + }); + } + } } group.finish(); } @@ -363,11 +928,19 @@ fn bench_parked_read(c: &mut Criterion) { criterion_group!( benches, bench_fanout, + bench_fanout_frame, bench_parallel_write, bench_aborted_scan, + bench_gc, bench_subscriber_churn, bench_subscriber_churn_after_peak, bench_subscriber_join, - bench_parked_read + bench_range_demand, + bench_subscriber_preferences, + bench_datagram_subscriber_join, + bench_parked_read, + bench_sparse_read, + bench_gap_flush, + bench_empty_run ); criterion_main!(benches); diff --git a/rs/moq-pattern/Cargo.toml b/rs/moq-pattern/Cargo.toml index bd2f54d7d0..5cdfbd32b4 100644 --- a/rs/moq-pattern/Cargo.toml +++ b/rs/moq-pattern/Cargo.toml @@ -20,6 +20,11 @@ serde = ["dep:serde"] serde = { workspace = true, optional = true } [dev-dependencies] +criterion = { workspace = true } # Enable serde so the crate's own tests cover the optional impls in the merge gate. moq-pattern = { path = ".", features = ["serde"] } serde_json = { workspace = true } + +[[bench]] +name = "matching" +harness = false diff --git a/rs/moq-pattern/benches/matching.rs b/rs/moq-pattern/benches/matching.rs new file mode 100644 index 0000000000..9b34070f13 --- /dev/null +++ b/rs/moq-pattern/benches/matching.rs @@ -0,0 +1,40 @@ +//! Matching a pattern union, swept over union size and path depth. +use criterion::{BenchmarkId, Criterion, Throughput, criterion_group, criterion_main}; +use moq_pattern::{Pattern, Patterns}; + +fn matching(c: &mut Criterion) { + let mut group = c.benchmark_group("pattern-matching"); + for count in [1, 16, 256] { + for depth in [1, 8, 32] { + let tail = vec!["node"; depth - 1].join("/"); + let path = |tenant: String| { + if tail.is_empty() { + tenant + } else { + format!("{tenant}/{tail}") + } + }; + let patterns: Patterns = (0..count) + .map(|index| path(format!("tenant{index:03}")).parse::().unwrap()) + .collect(); + assert_eq!(patterns.len(), count); + for (case, input, expected) in [ + ("last-match", path(format!("tenant{:03}", count - 1)), true), + ("miss", path("other".into()), false), + ] { + assert_eq!(patterns.matches(&input), expected); + group.throughput(Throughput::Elements(1)); + group.bench_function( + BenchmarkId::new(case, format!("{count}patterns-{depth}segments")), + |b| { + b.iter(|| std::hint::black_box(patterns.matches(std::hint::black_box(&input)))); + }, + ); + } + } + } + group.finish(); +} + +criterion_group!(benches, matching); +criterion_main!(benches); diff --git a/rs/moq-rtmp/Cargo.toml b/rs/moq-rtmp/Cargo.toml index fb8563f8d3..020b1f212f 100644 --- a/rs/moq-rtmp/Cargo.toml +++ b/rs/moq-rtmp/Cargo.toml @@ -53,9 +53,14 @@ tokio-rustls = { workspace = true, optional = true } tracing = { workspace = true } [dev-dependencies] +criterion = { workspace = true } # The RTMPS test borrows moq-tokio's self-signed cert generation to build a # `rustls::ServerConfig`. No transport feature: a QUIC backend here would be # unified into every other crate a `--workspace` build selects, which is exactly # what hid moq-tokio's backend-less build from the nightly feature matrix. moq-tokio = { path = "../moq-tokio", default-features = false, features = ["aws-lc-rs"] } tokio = { workspace = true, features = ["test-util"] } + +[[bench]] +name = "chunk_interleaving" +harness = false diff --git a/rs/moq-rtmp/benches/chunk_interleaving.rs b/rs/moq-rtmp/benches/chunk_interleaving.rs new file mode 100644 index 0000000000..0917f00c61 --- /dev/null +++ b/rs/moq-rtmp/benches/chunk_interleaving.rs @@ -0,0 +1,60 @@ +//! Decode one hot CSID while other CSIDs retain incomplete messages. +//! Sweep the table size independently of message size and chunk count. + +#[path = "../src/rml/mod.rs"] +mod rml; + +use criterion::{BenchmarkId, Criterion, Throughput, criterion_group, criterion_main}; +use rml::chunk_io::ChunkDeserializer; +use std::hint::black_box; + +fn message(csid: u16, length: usize, chunk: usize) -> Vec { + let basic = if csid < 64 { + vec![csid as u8] + } else { + vec![0, (csid - 64) as u8] + }; + let mut bytes = basic.clone(); + bytes.extend_from_slice(&[0, 0, 1]); + bytes.extend_from_slice(&(length as u32).to_be_bytes()[1..]); + bytes.extend_from_slice(&[9, 1, 0, 0, 0]); + bytes.extend(std::iter::repeat_n(42, length.min(chunk))); + for start in (chunk..length).step_by(chunk) { + let mut continuation = basic.clone(); + continuation[0] |= 0xc0; + bytes.extend_from_slice(&continuation); + bytes.extend(std::iter::repeat_n(42, chunk.min(length - start))); + } + bytes +} + +fn bench(c: &mut Criterion) { + let mut group = c.benchmark_group("rtmp_chunks"); + for ids in [1, 16, 64, 256] { + for length in [128, 4096, 65536] { + for chunk in [128, 4096] { + let mut decoder = ChunkDeserializer::new(); + decoder.set_max_chunk_size(chunk).unwrap(); + for csid in 3..(ids + 2) { + let bytes = message(csid, chunk * 2, chunk); + let header = if csid < 64 { 12 } else { 13 }; + assert!(decoder.get_next_message(&bytes[..header + chunk]).unwrap().is_none()); + } + let bytes = message(2, length, chunk); + group.throughput(Throughput::Bytes(length as u64)); + group.bench_function(BenchmarkId::new(format!("ids_{ids}/chunk_{chunk}"), length), |b| { + b.iter(|| { + let payload = decoder.get_next_message(black_box(&bytes)).unwrap().unwrap(); + assert_eq!(payload.data.len(), length); + black_box(payload); + assert!(decoder.get_next_message(&[]).unwrap().is_none()); + }); + }); + } + } + } + group.finish(); +} + +criterion_group!(benches, bench); +criterion_main!(benches); diff --git a/rs/moq-stats/Cargo.toml b/rs/moq-stats/Cargo.toml index 398c835eb1..2778ae0c2a 100644 --- a/rs/moq-stats/Cargo.toml +++ b/rs/moq-stats/Cargo.toml @@ -28,3 +28,7 @@ tokio = { workspace = true, features = ["macros", "rt", "test-util", "time"] } [[bench]] name = "decode" harness = false + +[[bench]] +name = "aggregate" +harness = false diff --git a/rs/moq-stats/benches/aggregate.rs b/rs/moq-stats/benches/aggregate.rs new file mode 100644 index 0000000000..d8d4798adf --- /dev/null +++ b/rs/moq-stats/benches/aggregate.rs @@ -0,0 +1,144 @@ +//! Cost of one merged traffic frame from [`moq_stats::aggregate`], swept over live nodes and +//! nodes that have come and gone. +//! +//! Each node reports the same broadcast paths, like a restarted node under a new name. This +//! aggregate keeps every departed node's totals, so it also pays for every departed node on each +//! frame. +//! +//! Run `cargo bench -p moq-stats --bench aggregate` and compare the table across revisions. + +// Custom timings measure the real time the work took, not the ambient clock. +#![allow(clippy::disallowed_methods)] + +use std::time::{Duration, Instant}; + +use moq_net::origin; +use moq_stats::{Role, Tier, Traffic, TrafficFrame, aggregate, traffic_track}; + +/// Broadcast paths each node reports. +const KEYS: usize = 16; + +/// Frames measured per configuration, after one warm-up frame. +const FRAMES: u32 = 64; + +/// A hand-published node stats broadcast with a default-tier publisher traffic track. +struct Node { + /// Held so the node stays announced until dropped. + _source: moq_net::broadcast::Producer, + traffic: moq_json::snapshot::Producer, + frame: TrafficFrame, +} + +impl Node { + fn new(origin: &origin::Producer, name: &str) -> Self { + let source = origin + .create_broadcast(format!(".stats/acme/node/{name}").as_str()) + .unwrap(); + source.announce(origin::Route::default()).unwrap(); + let track = source + .create_track( + traffic_track(&Tier::default(), Role::Publisher, false), + moq_net::track::Info::default().with_timescale(moq_net::Timescale::MILLI), + ) + .unwrap(); + let config = moq_json::snapshot::Config::default().with_delta_ratio(0); + let frame = (0..KEYS) + .map(|key| { + let mut traffic = Traffic::default(); + traffic.subscriptions_started = 1; + (format!("acme/room-{key:02}"), traffic) + }) + .collect(); + Self { + _source: source, + traffic: moq_json::snapshot::Producer::new(track, config), + frame, + } + } + + /// Bump every path's byte counter and publish the whole snapshot. + fn publish(&mut self, bytes: u64) { + for traffic in self.frame.values_mut() { + traffic.bytes += bytes; + } + self.traffic + .update(moq_net::Timed::from(&self.frame).at(moq_net::Timestamp::now())) + .unwrap(); + } +} + +/// Read merged frames until the first path reaches `want` bytes. +async fn read_until(traffic: &mut aggregate::TrafficConsumer, want: u64) { + loop { + let frame = traffic.next().await.unwrap().unwrap(); + if frame.values().next().is_some_and(|t| t.bytes >= want) { + return; + } + } +} + +/// Mean wall time per merged frame with `live` nodes publishing after `departed` nodes left. +async fn measure(live: usize, departed: usize) -> Duration { + let (origin, driver) = origin::Producer::new(origin::Config::default()); + let driver = tokio::spawn(moq_net::time::run(driver)); + let agg = aggregate::Consumer::new(origin.consume(), aggregate::Config::new().with_depth(1)); + let mut traffic = agg.traffic(&Tier::default(), Role::Publisher); + + // Every node contributes 1 byte per path, so the total counts nodes seen. + let mut total = 0; + for index in 0..departed { + let mut node = Node::new(&origin, &format!("gone-{index}")); + node.publish(1); + total += 1; + read_until(&mut traffic, total).await; + } + + let mut nodes: Vec = (0..live) + .map(|index| Node::new(&origin, &format!("live-{index}"))) + .collect(); + for node in &mut nodes { + node.publish(1); + total += 1; + read_until(&mut traffic, total).await; + } + + let mut elapsed = Duration::ZERO; + for frame in 0..=FRAMES { + let node = &mut nodes[frame as usize % live]; + let start = Instant::now(); + node.publish(1); + total += 1; + read_until(&mut traffic, total).await; + // The first frame is the warm-up. + if frame > 0 { + elapsed += start.elapsed(); + } + } + + drop(nodes); + driver.abort(); + elapsed / FRAMES +} + +fn main() { + // Nextest lists all targets as potential test binaries. + if std::env::args().any(|arg| arg == "--list") { + return; + } + let runtime = tokio::runtime::Builder::new_current_thread() + .enable_time() + .start_paused(true) + .build() + .unwrap(); + + println!("live departed grace metadata us/frame"); + for live in [1, 16, 128] { + for departed in [0, 256, 4096] { + // Rows match main's unbounded (`grace` never elapsing) rows with known metadata. + let (label, metadata) = ("never", "known"); + let elapsed = runtime.block_on(measure(live, departed)); + let micros = elapsed.as_secs_f64() * 1e6; + println!("{live:>4} {departed:>8} {label:>5} {metadata:>8} {micros:>11.1}"); + } + } +} diff --git a/rs/moq-uring/Cargo.toml b/rs/moq-uring/Cargo.toml index bce6dc7c93..278efea85c 100644 --- a/rs/moq-uring/Cargo.toml +++ b/rs/moq-uring/Cargo.toml @@ -81,3 +81,7 @@ required-features = ["noq"] [[bench]] name = "session_lite" harness = false + +[[bench]] +name = "worker_turns" +harness = false diff --git a/rs/moq-uring/benches/worker_turns.rs b/rs/moq-uring/benches/worker_turns.rs new file mode 100644 index 0000000000..9d9e21cc14 --- /dev/null +++ b/rs/moq-uring/benches/worker_turns.rs @@ -0,0 +1,75 @@ +//! Worker submission and completion amortization across sockets and burst sizes. + +use criterion::{criterion_group, criterion_main}; + +#[cfg(target_os = "linux")] +fn benchmark(c: &mut criterion::Criterion) { + use std::net::UdpSocket; + + use criterion::{BenchmarkId, Throughput}; + use moq_uring::{Config, Error, Worker, udp}; + + let mut group = c.benchmark_group("worker_turns"); + for sockets in [1, 4, 16] { + for burst in [1, 8, 64] { + let mut worker = match Worker::new(Config::default()) { + Ok(worker) => worker, + Err(Error::Unsupported(reason)) => { + eprintln!("skipping worker benchmark: {reason}"); + return; + } + Err(err) => panic!("worker setup failed: {err}"), + }; + let handle = worker.handle(); + let mut config = udp::Config::default(); + config.gro = false; + config.gso = false; + config.rx_buffer_len = 2048; + config.tx_buffer_len = 2048; + config.rx_buffers_max = 64; + let bind = || UdpSocket::bind("127.0.0.1:0").expect("bind"); + let sender = handle.udp(bind(), config.clone()).expect("sender"); + let receivers: Vec<_> = (0..sockets) + .map(|_| handle.udp(bind(), config.clone()).expect("receiver")) + .collect(); + group.throughput(Throughput::Elements(sockets * burst)); + group.bench_function(BenchmarkId::new(format!("sockets-{sockets}"), burst), |b| { + b.iter(|| { + worker + .block_on(async { + for receiver in &receivers { + for _ in 0..burst { + let mut tx = sender.acquire().await.expect("send buffer"); + tx[..4].copy_from_slice(b"turn"); + tx.send(udp::Transmit { + to: receiver.local_addr().expect("addr"), + len: 4, + segment: 4, + ecn: None, + }) + .expect("send"); + } + } + for receiver in &receivers { + for _ in 0..burst { + assert_eq!(receiver.recv().await.expect("receive").payload(), b"turn"); + } + } + }) + .expect("worker"); + }); + }); + eprintln!( + "worker_turns sockets={sockets} burst={burst}: {:?}", + handle.metrics().snapshot() + ); + } + } + group.finish(); +} + +#[cfg(not(target_os = "linux"))] +fn benchmark(_: &mut criterion::Criterion) {} + +criterion_group!(benches, benchmark); +criterion_main!(benches);