diff --git a/.github/workflows/nightly.yml b/.github/workflows/nightly.yml index f692005c3d..bb982cfe1c 100644 --- a/.github/workflows/nightly.yml +++ b/.github/workflows/nightly.yml @@ -83,6 +83,10 @@ jobs: env: NEXTEST_PROFILE: ci + - name: Audio grouping relay benchmark + if: ${{ !cancelled() }} + run: nix develop --command just bench-audio + - name: Audio mixer benchmark if: ${{ !cancelled() }} run: nix develop --command just rs bench-playback diff --git a/bench/README.md b/bench/README.md index 9b3b3968aa..2ed1eb44aa 100644 --- a/bench/README.md +++ b/bench/README.md @@ -15,6 +15,20 @@ Run every Criterion target plus the local relay workloads: nix develop --command just bench ``` +Measure audio grouping at 0, 100 and 200 ms with 50 fps, 200-byte frames: + +```bash +nix develop --command just bench-audio +``` + +This sweeps room connections (16, 32) and subscriptions per connection (2, 8) +independently, plus one publisher serving 64 and 200 subscribers. It uses the +normal relay sampler and runs nightly. Each shape reports delivered frames, +p99 latency, relay CPU and RSS; the group-size suffix is the number of frames +after the keyframe (0, 4, 9), so each group carries 1, 5 or 10 frames. These +synthetic payloads measure group overhead; they do not measure codec loss +concealment. + Compare the current tree with another revision: ```bash @@ -43,6 +57,8 @@ interval, and output paths so every runtime receives the same load. - `fanout`: light one-to-many traffic. - `video-heavy`: multicore many-to-many video traffic. - `fanout-heavy`: multicore one-to-many traffic near saturation. +- `audio` and `audio-fanout`: small, 50 fps audio-shaped frames; `bench-audio` + sweeps grouping and load shape through command-line overrides. The runtime matrix rotates execution order between rounds, then reports the median throughput, loss, latency, CPU split, context switches, RSS, and thread diff --git a/bench/relay.sh b/bench/relay.sh index 0a7a3369e0..bc251f4c3a 100755 --- a/bench/relay.sh +++ b/bench/relay.sh @@ -217,6 +217,7 @@ run_workload() { local workload=$3 local runtime=${4:-} local workers=${5:-1} + local -a overrides=("${@:6}") local directory=$RUN/relay/$label/$workload local stats=$directory/load.jsonl local host=$directory/host.jsonl @@ -245,7 +246,7 @@ run_workload() { --startup 2s \ --duration 10s \ --report 500ms \ - --output "$stats" >"$directory/load.log" 2>&1; then + --output "$stats" ${overrides[@]+"${overrides[@]}"} >"$directory/load.log" 2>&1; then printf 'load benchmark failed: %s/%s\n' "$label" "$workload" >&2 cat "$directory/load.log" >&2 return 1 diff --git a/bench/run.sh b/bench/run.sh index db7c0a2c6e..695c517e55 100755 --- a/bench/run.sh +++ b/bench/run.sh @@ -6,10 +6,13 @@ set -euo pipefail # base/current delta. `--runtime` instead compares one multi-threaded Tokio # runtime with the same number of independent Tokio/epoll and io_uring workers # on the current tree. The current load generator drives every relay so the -# workloads stay identical. +# workloads stay identical. `--audio` isolates the audio grouping/load-shape matrix. MODE=${1:-} case $MODE in + --audio) + BASE= + ;; --runtime) BASE= RUNTIME_ROUNDS=${MOQ_BENCH_RUNTIME_ROUNDS:-3} @@ -373,6 +376,42 @@ print_runtime_comparison() { done } +run_audio_comparison() { + printf 'Building relay and audio load generator...\n' + CARGO_TARGET_DIR=$CURRENT_TARGET cargo build --locked --release -p moq-relay -p moq-bench + LOAD_BIN=$CURRENT_TARGET/release/moq-bench + HOST_BIN=$CURRENT_TARGET/release/moq-bench-host + + printf '\nAudio grouping: 50 fps, 200-byte frames, 0 / 100 / 200 ms groups\n' + printf 'shape\trecv-fps\tp99-ms\tCPU\tRSS-MiB\n' + local connections subscribers group_size label + for group_size in 0 4 9; do + for connections in 16 32; do + for subscribers in 2 8; do + label=room-$connections-$subscribers-$group_size + run_workload "$label" "$CURRENT_TARGET/release/moq-relay" audio '' 1 \ + --connections "$connections" --subscribe "$subscribers" --group-size "$group_size" + print_audio_sample "$label" audio + done + done + for connections in 65 201; do + label=fanout-$connections-$group_size + run_workload "$label" "$CURRENT_TARGET/release/moq-relay" audio-fanout '' 1 \ + --connections "$connections" --group-size "$group_size" + print_audio_sample "$label" audio-fanout + done + done +} + +print_audio_sample() { + local label=$1 + local workload=$2 + jq -r --arg shape "$label" '[ + $shape, .recv_fps, .latency_p99_ms, (.cpu_cores // "n/a"), + (if .rss_bytes then .rss_bytes / 1048576 else "n/a" end) + ] | @tsv' "$RUN/relay/$label/$workload/summary.json" +} + run_runtime_comparison() { if [[ $(uname -s) != Linux ]]; then printf 'runtime comparison needs Linux for io_uring and /proc metrics\n' >&2 @@ -429,7 +468,10 @@ run_runtime_comparison() { cd "$ROOT" -if [[ $MODE == --runtime ]]; then +if [[ $MODE == --audio ]]; then + run_audio_comparison + exit 0 +elif [[ $MODE == --runtime ]]; then run_runtime_comparison exit 0 elif [[ -n $BASE ]]; then diff --git a/bench/workloads/audio-fanout.toml b/bench/workloads/audio-fanout.toml new file mode 100644 index 0000000000..db42bdf8f6 --- /dev/null +++ b/bench/workloads/audio-fanout.toml @@ -0,0 +1,5 @@ +connections = 65 +fanout = "audio" +fps = 50 +frame_size = 200 +group_size = 0 diff --git a/bench/workloads/audio.toml b/bench/workloads/audio.toml new file mode 100644 index 0000000000..2e2aa254d0 --- /dev/null +++ b/bench/workloads/audio.toml @@ -0,0 +1,6 @@ +connections = 16 +broadcasts = 1 +subscribe = 2 +fps = 50 +frame_size = 200 +group_size = 0 diff --git a/doc/lib/js/publish.md b/doc/lib/js/publish.md index 8b496b9cea..abe91e70c5 100644 --- a/doc/lib/js/publish.md +++ b/doc/lib/js/publish.md @@ -66,6 +66,19 @@ Disabling a rendition (`muted` on the element) ends the audio timeline with a marker, so a viewer that stays subscribed, or joins during the pause, never plays the audio before it as live. +Each audio frame is its own group by default, so a relay pays a stream and a +group's bookkeeping per 20ms frame. `el.audio.groupDuration` sets a minimum +per group instead, such as `Time.Milli(100)`: the first frame at least 100ms +after the group's first timestamp opens the next group and closes the previous +one. If encoding pauses, the current group stays open until the next frame or +timeline marker. Frames still forward as they are encoded, so grouping does +not buffer them, but loss gets coarser: a viewer that falls behind skips a +whole group, and a lost frame holds back the rest of its group until it is retransmitted. A 60ms +Opus `frameDuration` also cuts the group rate, without code, at the cost of +encoder latency. The synthetic `just bench-audio` workload reports higher p99 +delivery latency with longer groups; immediate forwarding is not a promise of +unchanged end-to-end latency. + `el.video.cut()` asks for a keyframe on top of the `keyframeInterval` cadence, for a resume, a recording cut, or a known tune-in moment. Requests coalesce into the next keyframe, and forced keyframes land at least 500ms apart. diff --git a/doc/lib/rs/moq-audio.md b/doc/lib/rs/moq-audio.md index 352b9f2e2c..c901ec53d6 100644 --- a/doc/lib/rs/moq-audio.md +++ b/doc/lib/rs/moq-audio.md @@ -88,6 +88,19 @@ and no platform encoder is wired in yet, so `Codec::Aac` is refused at construction on every host for now. Linux has no OS encoder, so it will stay that way there. +Each packet is its own group by default, so a relay pays a stream and a +group's bookkeeping per 20 ms Opus frame. `encode::Options::group_duration` +sets a minimum per group instead, for every codec: the packet that reaches it +closes the group. During a pause or after `reset_epoch()`, a partial group +stays open until the next write; `discontinuity()` closes it immediately. +Packets forward as they are encoded rather than waiting for the group to fill, +but loss gets coarser: a subscriber that falls behind skips a whole +group, and a lost packet holds back the rest of its group until it is +retransmitted. A 60 ms Opus `Settings::frame_duration` also cuts the group +rate, without code, at the cost of encoder latency. The synthetic +`just bench-audio` workload reports higher p99 delivery latency with longer +groups; immediate forwarding is not a promise of unchanged end-to-end latency. + Highlights: - **`encode::Control`** advertises the track and opens the microphone only while someone listens. Stop, swap devices, and restart without changing the track subscribers know; read a level meter for the UI. diff --git a/js/publish/src/audio/encoder.test.ts b/js/publish/src/audio/encoder.test.ts index 41cef17080..48a442efb3 100644 --- a/js/publish/src/audio/encoder.test.ts +++ b/js/publish/src/audio/encoder.test.ts @@ -202,20 +202,26 @@ function configured(): Promise { }); } -// An Encoder wired to a fake capture feed, recording each written frame as [timestamp, payload bytes]. -async function setup(baseline = new Baseline(), codec?: Codec) { +// An Encoder wired to a fake capture feed, recording each written frame as [timestamp, payload bytes] +// and how many frames each group carries. +async function setup(baseline = new Baseline(), codec?: Codec, groupDuration?: Time.Milli) { const configuring = configured(); const track = new Moq.Track.Producer("audio").accept(); const written: [number, number][] = []; + const groups: number[] = []; + const appended: ReturnType[] = []; const writes = { onWrite: undefined as (() => void) | undefined }; const appendGroup = track.appendGroup.bind(track); track.appendGroup = () => { const group = appendGroup(); + appended.push(group); + const index = groups.push(0) - 1; const writeFrame = group.writeFrame.bind(group); group.writeFrame = (frame) => { const [timestamp, payload] = Moq.Varint.decode(frame.payload); written.push([timestamp, payload.byteLength]); + groups[index]++; writeFrame(frame); writes.onWrite?.(); }; @@ -245,6 +251,7 @@ async function setup(baseline = new Baseline(), codec?: Codec) { capture: capture as never, enabled, codec, + groupDuration, }); const config = await configuring; @@ -257,6 +264,8 @@ async function setup(baseline = new Baseline(), codec?: Codec) { rendition, feed, written, + groups, + appended, writes, [Symbol.dispose]() { encoder.close(); @@ -385,6 +394,48 @@ test("a push completing several frames keeps the encoder running", async () => { ]); }); +// The first frame at the minimum opens the next group. A timeline break closes the group early, +// so the first frame after it opens a fresh one. +test("a group duration packs frames until the minimum and restarts after a break", async () => { + using _webcodecs = installFakeWebCodecs(); + using env = await setup(new Baseline(), undefined, Time.Milli(100)); + const { enabled, feed, written, groups, appended, writes } = env; + + let index = 0; + const push = async (count: number) => { + for (let i = 0; i < count; i++, index++) { + await feed.push({ timestamp: Time.Micro(20_000 + index * 20_000), channels: [new Float32Array(960)] }); + } + await feed.drain(); + }; + + await push(7); // five written, two held + expect(groups).toEqual([5]); + expect(appended[0].closed.peek()).toBeUndefined(); + await push(2); // the next frame opens a new group and closes the first + expect(groups).toEqual([5, 2]); + expect(appended[0].closed.peek()).toBeNull(); + + const marked = new Promise((resolve) => { + writes.onWrite = resolve; + }); + enabled.set(false); + await marked; + writes.onWrite = undefined; + + const resumed = configured(); + enabled.set(true); + await resumed; + await push(4); // a fresh AudioEncoder, so two written and two held + + expect(written.map(([timestamp]) => timestamp)).toEqual([ + 20_000, 40_000, 60_000, 80_000, 100_000, 120_000, 140_000, 200_000, 200_000, 220_000, + ]); + // 120ms opens the next group, closing the five frames at 20-100ms. The break ends the + // 120-140ms group early, so resumed frames start a fresh group after the marker. + expect(groups).toEqual([5, 2, 1, 2]); +}); + // Chromium stamps Opus output by counting the samples emitted, so every frame DTX suppresses pulls // later audio earlier. A plain-JS caller passing the old knob must not reach the encoder. test("never enables Opus DTX", async () => { diff --git a/js/publish/src/audio/encoder.ts b/js/publish/src/audio/encoder.ts index 618fccd4fd..8b09f6289f 100644 --- a/js/publish/src/audio/encoder.ts +++ b/js/publish/src/audio/encoder.ts @@ -103,6 +103,9 @@ export type EncoderProps = Inputs & { // Codec selection plus encoder settings. Defaults to "opus". codec?: Codec | Signal; + + // The minimum audio carried by each group. Defaults to 0, a group per frame. + groupDuration?: Time.Milli | Signal; }; type EncoderOutput = { @@ -143,6 +146,12 @@ export class Encoder { fade: Signal; /** The live-editable codec selection plus its encoder settings. */ codec: Signal; + /** + * The minimum timestamp span before a frame opens the next group, closing the previous one. 0 puts every frame in its own group. A longer group costs the + * relay fewer streams but makes loss coarser: a viewer that falls behind skips a whole group. A + * negative or non-finite duration refuses the rendition. + */ + groupDuration: Signal; /** * The capture supplying this rendition, or undefined while none is wired. @@ -173,7 +182,8 @@ export class Encoder { #pipeline: Pipeline | undefined; // The current subscription's track and the producer writing into it, or undefined without demand. - #live: { track: Moq.Track.Producer; producer: Container.Legacy.Producer } | undefined; + // `start` is the open group's first timestamp, undefined until a frame opens one. + #live: { track: Moq.Track.Producer; producer: Container.Legacy.Producer; start?: Time.Micro } | undefined; // Where the next frame submitted to the AudioEncoder starts, i.e. the exclusive end of the // newest one, where an epoch's discontinuity marker goes. Cleared once the marker is written. @@ -188,6 +198,9 @@ export class Encoder { // invalid one, so a bad value never reaches the gain. #fade: Time.Milli = FADE; + // The last valid group duration, validated by #runConfig the same way as #fade. + #groupDuration = Time.Micro(0); + // The fatal error an AudioEncoder reported, if any. That instance can never encode again and // reconfiguring it would be a retry, so the rendition stays down for the life of this encoder. #fatal = new Signal(undefined); @@ -224,6 +237,7 @@ export class Encoder { this.volume = Signal.from(props?.volume ?? 1); this.fade = Signal.from(props?.fade ?? FADE); this.codec = Signal.from(props?.codec ?? "opus"); + this.groupDuration = Signal.from(props?.groupDuration ?? Time.Milli(0)); // Only the capture graph has a node to expose. this.#signals.run((effect) => { @@ -385,6 +399,11 @@ export class Encoder { throw new Error(`audio fade must be a finite, non-negative number of ms: ${fade}`); this.#fade = fade; + const groupDuration = effect.get(this.groupDuration); + if (!Number.isFinite(groupDuration) || groupDuration < 0) + throw new Error(`audio group duration must be a finite, non-negative number of ms: ${groupDuration}`); + this.#groupDuration = Time.Micro.fromMilli(groupDuration); + const capture = effect.get(this.in.capture); const captured = capture ? effect.get(capture.out.format) : undefined; if (!effect.get(this.in.enabled) || !captured) { @@ -436,6 +455,7 @@ export class Encoder { if (end === undefined || !live || live.track.closed.peek() !== undefined) return; this.#floor = end; live.producer.discontinuity(end); + live.start = undefined; } // Encode captured audio frames into whichever track producer is live. The broadcast owns the @@ -483,12 +503,18 @@ export class Encoder { bytes: stats.bytes + frame.byteLength, })); - // Each audio frame is a keyframe, so its own group, which the relay can forward - // without waiting for a group boundary. Loss is handled by the codec's PLC. const live = this.#live; if (!live) return; - if (this.#floor !== undefined && frame.timestamp < this.#floor) return; - live.producer.encode(frame, frame.timestamp as Time.Micro, true); + const timestamp = frame.timestamp as Time.Micro; + if (this.#floor !== undefined && timestamp < this.#floor) return; + + // Every audio frame decodes on its own, so any of them can open a group: the + // first one at or past the minimum after the open group's start. Frames forward + // as they are written rather than waiting for the group to fill. A dropped + // group leaves a gap for the codec's PLC. + const keyframe = live.start === undefined || timestamp - live.start >= this.#groupDuration; + if (keyframe) live.start = timestamp; + live.producer.encode(frame, timestamp, keyframe); if (this.#estimator.flush(frame.timestamp, baseline)) { const catalog = this.#out.catalog.peek(); if (catalog) this.#out.catalog.set({ ...catalog, ...this.#estimator.estimate }); diff --git a/justfile b/justfile index c20feb7f2d..d516df66c1 100644 --- a/justfile +++ b/justfile @@ -44,10 +44,13 @@ dev: bench $BASE="": bench/run.sh "$BASE" -# Compare one multi-threaded Tokio runtime with the same number of independent -# Tokio/epoll and io_uring workers; WORKERS defaults to every logical CPU. +# Measure audio group sizes across relay publisher/subscriber and fanout shapes. +bench-audio: + bench/run.sh --audio -# Compare a shared Tokio runtime with independent workers: `just bench-runtime 5 16`. +# Compare one multi-threaded Tokio runtime with independent Tokio/epoll and +# io_uring workers; WORKERS defaults to every logical CPU. +# Example: `just bench-runtime 5 16`. bench-runtime $ROUNDS="3" $WORKERS="": MOQ_BENCH_RUNTIME_ROUNDS="$ROUNDS" MOQ_BENCH_RUNTIME_WORKERS="$WORKERS" bench/run.sh --runtime diff --git a/quest/m1/README.md b/quest/m1/README.md index 45b2059842..a3600a1b19 100644 --- a/quest/m1/README.md +++ b/quest/m1/README.md @@ -134,7 +134,6 @@ blocks. The quests that gated m0 lines moved under them. - [Benchmark comparisons](/quest/m1/performance-comparisons.md) - retained evidence, repeated paired runs, and uncertainty for performance claims - [Relay profiling](/quest/m1/performance-profiles.md) - reproducible CPU and allocation captures under the existing workloads - [Perf](/quest/m1/perf/README.md) - eliminate measured hot-path costs across moq-uring, kio, and the moq-net model -- [Audio group duration](/quest/m1/audio-group-duration.md) - audio publishers in JS and Rust take a minimum group duration, default one frame - [#2924](/quest/m1/2924-moq-relay-tls-rotation-is-not-atomic-across-thread-per.md) - every listener on both runtimes shares one reloadable served identity, so rotation is atomic and generate works with workers - [Benchmark regressions in CI](/quest/m1/bench-ci.md) - PRs get a non-blocking comparison of the Criterion benches they affect, and a nightly trend on main alerts on regressions - [#3126](/quest/m1/3126-moq-bench-every-readme-example-fails-to-parse-and.md) - moq-bench reports per-interval latency percentiles so the ramp leaves the steady state diff --git a/quest/m1/audio-group-duration.md b/quest/m1/audio-group-duration.md deleted file mode 100644 index 373c22595d..0000000000 --- a/quest/m1/audio-group-duration.md +++ /dev/null @@ -1,44 +0,0 @@ -# [M] Audio publishers can pack several frames per group - -## Goal - -`@moq/publish` and `moq-audio` take a minimum audio group duration, so a -relay pays one stream and one group's bookkeeping per N frames instead of per -20 ms frame. The default stays 0, today's one frame per group. - -## Plan - -`js/publish/src/audio/encoder.ts` writes each frame as its own group, and -`rs/moq-audio/src/encode/producer.rs` cuts after every packet. The reporter -measured relay CPU and memory scaling with audio groups (about 50 per second -per track per subscriber). - -Decided (2026-10-04): - -- `groupDuration?: Time.Milli` on Audio.Encoder props and `group_duration` on - moq-audio's encode `Options`, for Opus, AAC, and PCM alike. The frame that - reaches the minimum closes its group; the next frame opens a new one. -- Minimum only. A maximum adds nothing until there is an opportunistic cut - point (silence, a video group start). -- Default 0. A caller raises it, for example when its viewers already buffer - for jitter. A longer Opus `frameDuration` (60 ms) is the no-code - alternative; document both. -- JS builds on the shared `Container.Legacy.Producer` that - `js/publish/src/audio/encoder.ts` already writes through. -- Measure relay cost and loss concealment at 0, 100, and 200 ms with the - existing relay bench, and record the numbers here. Longer groups also make - a viewer's group skipping coarser; note it in the docs. - -Follow-up decision after the measurement: whether the minimum should be -derived from a viewer latency or jitter hint instead of set by hand. Grouping -adds no delay normally, since frames forward within a group, and costs -head-of-line blocking only on loss, which a buffer an RTT deep absorbs. - -## Closes - -- [#4784](https://github.com/moq-dev/moq/issues/4784) - close this issue when the quest finishes - -## Related - -- [Group cost](/quest/m1/perf/group-cost.md) - makes each group cheaper at the relay -- [Watch jitter target](/quest/m0/audio-jitter-target/watch.md) - lists a group per audio frame as a suspect diff --git a/rs/moq-audio/src/encode/producer.rs b/rs/moq-audio/src/encode/producer.rs index dcf9519c96..d92de5745b 100644 --- a/rs/moq-audio/src/encode/producer.rs +++ b/rs/moq-audio/src/encode/producer.rs @@ -40,6 +40,19 @@ pub struct Options { /// can't at all, and at `hang`'s priorities audio outranks video, so it is only /// ever squeezed on a link that can't carry audio alone. pub bandwidth: moq_net::bandwidth::Allocator, + /// The minimum audio carried by each group. The packet that reaches it closes + /// the group, and the next packet opens a new one. + /// + /// Defaults to zero, a group per packet. Each group costs the relay a stream + /// and its bookkeeping, so raising this trades per-group overhead for coarser + /// loss: a viewer that falls behind skips a whole group, and a lost packet + /// stalls the rest of its group until retransmitted. Packets still forward as + /// they are encoded rather than buffered until the group fills. A longer + /// [`Settings::frame_duration`] (up to 60 ms for Opus) also packs more audio + /// per group, at the cost of encoder latency. An unfinished group remains open + /// across a pause or [`Producer::reset_epoch`] until the next write; use + /// [`Producer::discontinuity`] to close it when capture stops. + pub group_duration: std::time::Duration, } impl Default for Options { @@ -48,6 +61,7 @@ impl Default for Options { track: None, settings: Settings::default(), bandwidth: moq_net::bandwidth::Allocator::unlimited(), + group_duration: std::time::Duration::ZERO, } } } @@ -70,6 +84,10 @@ pub struct Producer { track: moq_mux::container::Producer, _ext: std::marker::PhantomData E>, pending: Vec, + /// Codec samples a group carries before the packet reaching it closes the group. + group_samples: u64, + /// Codec samples in the open group. + grouped: u64, /// Samples emitted since the current epoch (reset by [`reset_epoch`](Self::reset_epoch)). frames_produced: u64, /// Wall-clock anchor in microseconds, taken from the first frame after each @@ -164,11 +182,15 @@ impl Reserved { self.track.set(encoder.catalog())?; + let group_samples = options.group_duration.as_nanos() * u128::from(encoder.codec_rate()); + let group_samples = u64::try_from(group_samples.div_ceil(1_000_000_000)).unwrap_or(u64::MAX); + Ok(Registered { encoder, input, remix, resampler, + group_samples, }) } @@ -182,6 +204,8 @@ impl Reserved { track: self.track, _ext: self._ext, pending: Vec::new(), + group_samples: registered.group_samples, + grouped: 0, frames_produced: 0, epoch_us: None, pending_discontinuity: false, @@ -201,6 +225,7 @@ pub(crate) struct Registered { input: Input, remix: Option, resampler: Option, + group_samples: u64, } /// What a capture publication needs while its layout is still undiscovered. @@ -374,7 +399,7 @@ impl Producer { )?; self.frames_produced += self.encoder.frame_size() as u64; self.activity = packet.activity; - Self::publish(&mut self.track, packet, timestamp)?; + self.publish(packet, timestamp)?; self.decoder_boundary = false; } @@ -395,26 +420,29 @@ impl Producer { Ok(Timestamp::from_micros(micros)?) } - fn publish( - track: &mut moq_mux::container::Producer, - encoded: Encoded, - timestamp: Timestamp, - ) -> Result<(), Error> { - // Publish each audio packet as its own moq-lite group: write it as a keyframe, then cut - // (below) so the relay forwards it without waiting for the next. Codecs can recover - // independently after a dropped group. + fn publish(&mut self, encoded: Encoded, timestamp: Timestamp) -> Result<(), Error> { + // Every packet decodes on its own, so a group may start at any of them: the first packet + // after a cut opens one. Codecs recover independently after a dropped group. + let keyframe = self.track.needs_keyframe(); + if keyframe { + self.grouped = 0; + } let mux_frame = MuxFrame { timestamp, payload: encoded.payload, - keyframe: true, + keyframe, duration: None, }; - track.write(mux_frame)?; - // No boundary to give: the next packet bounds this one, and Opus frames have a - // deterministic duration anyway. Cut before observing the flush so a failed observation - // never leaves the group open. - track.cut(None)?; - track.flush(timestamp, Instant::now())?; + self.track.write(mux_frame)?; + self.grouped += self.encoder.frame_size() as u64; + // Cut as soon as the group is long enough rather than when the next packet arrives, so + // the relay sees it finish without waiting. No boundary to give: the next packet bounds + // this one, and codec frames have a deterministic duration anyway. Cut before observing + // the flush so a failed observation never leaves the group open. + if self.grouped >= self.group_samples { + self.track.cut(None)?; + } + self.track.flush(timestamp, Instant::now())?; Ok(()) } @@ -510,7 +538,7 @@ impl Producer { for packet in packets { let timestamp = Self::timestamp(epoch_us, self.frames_produced, delay, codec_rate)?; self.activity = packet.activity; - Self::publish(&mut self.track, packet, timestamp)?; + self.publish(packet, timestamp)?; self.frames_produced += frame_size as u64; } } @@ -577,7 +605,7 @@ mod tests { /// Terminal Opus lookahead samples survive both exact-frame and partial-frame input. #[tokio::test] async fn finish_publishes_the_opus_lookahead_tail() { - for frames in [960, 860] { + for (group_ms, frames) in [(0, 960), (0, 860), (100, 2_880), (100, 2_780)] { let input = Input { format: Format::F32, sample_rate: 48_000, @@ -585,6 +613,7 @@ mod tests { }; let options = Options { track: Some("audio".to_string()), + group_duration: Duration::from_millis(group_ms), settings: Settings { layout: Layout::Mono, bitrate: Some(moq_net::bandwidth::Rate::from_bps(128_000)), @@ -905,6 +934,105 @@ mod tests { pts } + /// The packet that reaches the group duration closes its group, so a minimum that + /// isn't a whole number of packets rounds up. + #[tokio::test] + async fn group_duration_packs_packets_per_group() { + for (group_ms, expected) in [(0, [1, 1]), (30, [2, 2]), (100, [5, 5])] { + let mut broadcast = moq_net::broadcast::Info::new().produce(); + let catalog = moq_mux::catalog::Producer::new(&mut broadcast, moq_mux::catalog::Config::default()).unwrap(); + let consumer = broadcast.consume(); + let input = Input { + format: Format::F32, + sample_rate: 48_000, + layout: Layout::Mono, + }; + let options = Options { + track: Some("audio".to_string()), + group_duration: Duration::from_millis(group_ms), + ..Options::default() + }; + let mut producer = Producer::new(&mut broadcast, catalog, input, &options).unwrap(); + let mut track = consumer + .track("audio") + .unwrap() + .subscribe(moq_net::track::Subscription::default().with_max_delay(Duration::from_secs(1))) + .await + .unwrap(); + + for index in 0..10 { + producer.write(&full_frame(index * 20_000)).unwrap(); + } + + let mut sizes = Vec::new(); + for _ in expected { + let mut group = track.recv_group().await.unwrap().expect("a group"); + let mut frames = 0; + while group.read_frame().await.unwrap().is_some() { + frames += 1; + } + sizes.push(frames); + } + assert_eq!(sizes, expected, "group duration {group_ms} ms"); + } + } + + /// Timeline breaks close a partial group and reset the resumed group's packet count. + #[tokio::test] + async fn group_duration_restarts_after_a_timeline_break() { + for deferred in [false, true] { + let mut broadcast = moq_net::broadcast::Info::new().produce(); + let catalog = moq_mux::catalog::Producer::new(&mut broadcast, moq_mux::catalog::Config::default()).unwrap(); + let consumer = broadcast.consume(); + let input = Input { + format: Format::F32, + sample_rate: 48_000, + layout: Layout::Mono, + }; + let options = Options { + track: Some("audio".to_string()), + group_duration: Duration::from_millis(100), + ..Options::default() + }; + let mut producer = Producer::new(&mut broadcast, catalog, input, &options).unwrap(); + let mut track = consumer.track("audio").unwrap().subscribe(None).await.unwrap(); + + for index in 0..2 { + producer.write(&full_frame(index * 20_000)).unwrap(); + } + let mut partial = track.recv_group().await.unwrap().expect("partial group"); + for _ in 0..2 { + assert!( + partial.read_frame().await.unwrap().is_some(), + "packets forward before the group fills" + ); + } + assert!(partial.poll_read_frame(&moq_net::kio::Waiter::noop()).is_pending()); + + if deferred { + producer.reset_epoch(); + assert!(partial.poll_read_frame(&moq_net::kio::Waiter::noop()).is_pending()); + } else { + producer.discontinuity().unwrap(); + assert!(partial.read_frame().await.unwrap().is_none()); + } + for index in 0..5 { + producer.write(&full_frame(1_000_000 + index * 20_000)).unwrap(); + } + assert!(partial.read_frame().await.unwrap().is_none()); + let mut resumed = track.recv_group().await.unwrap().expect("resumed group"); + assert_eq!( + resumed.sequence, + partial.sequence + 2, + "one marker separates the epochs" + ); + for _ in 0..5 { + assert!(resumed.read_frame().await.unwrap().is_some()); + } + assert!(resumed.read_frame().await.unwrap().is_none()); + } + } + #[tokio::test] async fn epoch_anchors_to_first_frame_timestamp() { // The first frame's timestamp becomes the epoch (regression guard: the