diff --git a/cpp/obs/src/moq-audio-encoder.cpp b/cpp/obs/src/moq-audio-encoder.cpp index 69bd726c425..92fe5b4e6c9 100644 --- a/cpp/obs/src/moq-audio-encoder.cpp +++ b/cpp/obs/src/moq-audio-encoder.cpp @@ -114,7 +114,23 @@ bool Encode(void *data, struct encoder_frame *frame, struct encoder_packet *pack LOG_ERROR("MoQ Opus encode failed: %s", encoded.error().to_string().c_str()); return false; } - enc->packet = std::move(encoded->data); + // libopus readies each frame's packet as soon as it is submitted, and OBS + // takes at most one packet per call. + auto received = enc->codec->receive(); + if (!received) { + LOG_ERROR("MoQ Opus encode failed: %s", received.error().to_string().c_str()); + return false; + } + if (!*received) { + LOG_ERROR("MoQ Opus encoder held a frame's packet"); + return false; + } + enc->packet = std::move((*received)->data); + auto extra = enc->codec->receive(); + if (!extra || *extra) { + LOG_ERROR("MoQ Opus encoder readied more than one packet for one frame"); + return false; + } // One packet per frame, stamped with its first input sample: the OpusHead's // pre-skip already tells the decoder to trim the codec's lookahead. diff --git a/dart/moq_ffi/lib/src/moq.dart b/dart/moq_ffi/lib/src/moq.dart index b1407ed2e86..2013b2dab14 100644 --- a/dart/moq_ffi/lib/src/moq.dart +++ b/dart/moq_ffi/lib/src/moq.dart @@ -9427,9 +9427,11 @@ class FfiConverterMoqAudioConsumer { } abstract class MoqAudioEncoderInterface { - MoqAudioPacket encode({required List pcm}); + void encode({required List pcm}); MoqAudioEncoderFinished finish({required List pcm}); + void flush(); int frameSize(); + MoqAudioPacket? receive(); void setBitrate({required int bitrate}); MoqAudioInit trackInit(); } @@ -9481,16 +9483,14 @@ class MoqAudioEncoder implements MoqAudioEncoderInterface { } } - MoqAudioPacket encode({required List pcm}) { - return rustCallWithLifter( - (status) => uniffi_moq_ffi_fn_method_moqaudioencoder_encode( + void encode({required List pcm}) { + return rustCall((status) { + uniffi_moq_ffi_fn_method_moqaudioencoder_encode( uniffiClonePointer(), FfiConverterSequenceDouble32.lower(pcm), status, - ), - FfiConverterMoqAudioPacket.lift, - moqExceptionErrorHandler, - ); + ); + }, moqExceptionErrorHandler); } MoqAudioEncoderFinished finish({required List pcm}) { @@ -9505,6 +9505,15 @@ class MoqAudioEncoder implements MoqAudioEncoderInterface { ); } + void flush() { + return rustCall((status) { + uniffi_moq_ffi_fn_method_moqaudioencoder_flush( + uniffiClonePointer(), + status, + ); + }, moqExceptionErrorHandler); + } + int frameSize() { return rustCallWithLifter( (status) => uniffi_moq_ffi_fn_method_moqaudioencoder_frame_size( @@ -9516,6 +9525,17 @@ class MoqAudioEncoder implements MoqAudioEncoderInterface { ); } + MoqAudioPacket? receive() { + return rustCallWithLifter( + (status) => uniffi_moq_ffi_fn_method_moqaudioencoder_receive( + uniffiClonePointer(), + status, + ), + FfiConverterOptionalMoqAudioPacket.lift, + moqExceptionErrorHandler, + ); + } + void setBitrate({required int bitrate}) { return rustCall((status) { uniffi_moq_ffi_fn_method_moqaudioencoder_set_bitrate( @@ -17327,6 +17347,53 @@ class FfiConverterOptionalMoqAnnounceEvent { } } +class FfiConverterOptionalMoqAudioPacket { + static MoqAudioPacket? lift(RustBuffer buf) { + return FfiConverterOptionalMoqAudioPacket.read(buf.asUint8List()).value; + } + + static LiftRetVal read(Uint8List buf) { + if (ByteData.view(buf.buffer, buf.offsetInBytes).getInt8(0) == 0) { + return LiftRetVal(null, 1); + } + final result = FfiConverterMoqAudioPacket.read( + Uint8List.view(buf.buffer, buf.offsetInBytes + 1), + ); + return LiftRetVal(result.value, result.bytesRead + 1); + } + + static int allocationSize([MoqAudioPacket? value]) { + if (value == null) { + return 1; + } + return FfiConverterMoqAudioPacket.allocationSize(value) + 1; + } + + static RustBuffer lower(MoqAudioPacket? value) { + if (value == null) { + return toRustBuffer(Uint8List.fromList([0])); + } + final buf = Uint8List( + FfiConverterOptionalMoqAudioPacket.allocationSize(value), + ); + FfiConverterOptionalMoqAudioPacket.write(value, buf); + return toRustBuffer(buf); + } + + static int write(MoqAudioPacket? value, Uint8List buf) { + if (value == null) { + buf[0] = 0; + return 1; + } + buf[0] = 1; + return FfiConverterMoqAudioPacket.write( + value, + Uint8List.view(buf.buffer, buf.offsetInBytes + 1), + ) + + 1; + } +} + class FfiConverterOptionalMoqAudioRead { static MoqAudioRead? lift(RustBuffer buf) { return FfiConverterOptionalMoqAudioRead.read(buf.asUint8List()).value; @@ -19759,10 +19826,10 @@ external Pointer uniffi_moq_ffi_fn_constructor_moqaudioencoder_new( Pointer uniffiStatus, ); -@Native< - RustBuffer Function(Pointer, RustBuffer, Pointer) ->(assetId: _uniffiAssetId) -external RustBuffer uniffi_moq_ffi_fn_method_moqaudioencoder_encode( +@Native, RustBuffer, Pointer)>( + assetId: _uniffiAssetId, +) +external void uniffi_moq_ffi_fn_method_moqaudioencoder_encode( Pointer ptr, RustBuffer pcm, Pointer uniffiStatus, @@ -19777,6 +19844,14 @@ external RustBuffer uniffi_moq_ffi_fn_method_moqaudioencoder_finish( Pointer uniffiStatus, ); +@Native, Pointer)>( + assetId: _uniffiAssetId, +) +external void uniffi_moq_ffi_fn_method_moqaudioencoder_flush( + Pointer ptr, + Pointer uniffiStatus, +); + @Native, Pointer)>( assetId: _uniffiAssetId, ) @@ -19785,6 +19860,14 @@ external int uniffi_moq_ffi_fn_method_moqaudioencoder_frame_size( Pointer uniffiStatus, ); +@Native, Pointer)>( + assetId: _uniffiAssetId, +) +external RustBuffer uniffi_moq_ffi_fn_method_moqaudioencoder_receive( + Pointer ptr, + Pointer uniffiStatus, +); + @Native, Uint64, Pointer)>( assetId: _uniffiAssetId, ) @@ -23282,9 +23365,15 @@ external int uniffi_moq_ffi_checksum_method_moqaudioencoder_encode(); @Native(assetId: _uniffiAssetId) external int uniffi_moq_ffi_checksum_method_moqaudioencoder_finish(); +@Native(assetId: _uniffiAssetId) +external int uniffi_moq_ffi_checksum_method_moqaudioencoder_flush(); + @Native(assetId: _uniffiAssetId) external int uniffi_moq_ffi_checksum_method_moqaudioencoder_frame_size(); +@Native(assetId: _uniffiAssetId) +external int uniffi_moq_ffi_checksum_method_moqaudioencoder_receive(); + @Native(assetId: _uniffiAssetId) external int uniffi_moq_ffi_checksum_method_moqaudioencoder_set_bitrate(); @@ -24146,15 +24235,21 @@ void _checkApiChecksums() { if (uniffi_moq_ffi_checksum_method_moqaudioconsumer_next() != 7650) { throw UniffiInternalError.panicked("UniFFI API checksum mismatch"); } - if (uniffi_moq_ffi_checksum_method_moqaudioencoder_encode() != 56589) { + if (uniffi_moq_ffi_checksum_method_moqaudioencoder_encode() != 12465) { throw UniffiInternalError.panicked("UniFFI API checksum mismatch"); } if (uniffi_moq_ffi_checksum_method_moqaudioencoder_finish() != 24341) { throw UniffiInternalError.panicked("UniFFI API checksum mismatch"); } + if (uniffi_moq_ffi_checksum_method_moqaudioencoder_flush() != 43516) { + throw UniffiInternalError.panicked("UniFFI API checksum mismatch"); + } if (uniffi_moq_ffi_checksum_method_moqaudioencoder_frame_size() != 37264) { throw UniffiInternalError.panicked("UniFFI API checksum mismatch"); } + if (uniffi_moq_ffi_checksum_method_moqaudioencoder_receive() != 46467) { + throw UniffiInternalError.panicked("UniFFI API checksum mismatch"); + } if (uniffi_moq_ffi_checksum_method_moqaudioencoder_set_bitrate() != 10667) { throw UniffiInternalError.panicked("UniFFI API checksum mismatch"); } diff --git a/go/wrapper/audio/audio.go b/go/wrapper/audio/audio.go index 307c3bc00fc..1d719b0ca42 100644 --- a/go/wrapper/audio/audio.go +++ b/go/wrapper/audio/audio.go @@ -207,8 +207,14 @@ func (e *Encoder) TrackInit() (media.AudioInit, error) { return e.inner.TrackIni // SetBitrate retunes the live encoder in bits per second. func (e *Encoder) SetBitrate(value uint64) error { return e.inner.SetBitrate(value) } -// Encode encodes one codec-sized interleaved f32 frame. -func (e *Encoder) Encode(pcm []float32) (Packet, error) { return e.inner.Encode(pcm) } +// Encode submits one codec-sized interleaved f32 frame, whose packet Receive returns. +func (e *Encoder) Encode(pcm []float32) error { return e.inner.Encode(pcm) } + +// Receive pops the oldest ready packet, or nil while the codec holds the rest; call it until nil. +func (e *Encoder) Receive() (*Packet, error) { return e.inner.Receive() } + +// Flush marks the end of input, so Receive returns every packet the codec still holds, without closing the encoder. +func (e *Encoder) Flush() error { return e.inner.Flush() } // Finish pads an optional partial frame, drains the codec, and closes the encoder. func (e *Encoder) Finish(pcm []float32) (EncoderFinish, error) { return e.inner.Finish(pcm) } diff --git a/py/moq-rs/moq/audio.py b/py/moq-rs/moq/audio.py index e01c0f8f100..ce747fd2caf 100644 --- a/py/moq-rs/moq/audio.py +++ b/py/moq-rs/moq/audio.py @@ -223,9 +223,17 @@ def set_bitrate(self, bitrate: int) -> None: """Retune the live encoder in bits per second.""" self._inner.set_bitrate(bitrate) - def encode(self, pcm: list[float]) -> Packet: - """Encode one full PCM frame.""" - return self._inner.encode(pcm) + def encode(self, pcm: list[float]) -> None: + """Submit one full PCM frame, whose packet ``receive`` returns.""" + self._inner.encode(pcm) + + def receive(self) -> Packet | None: + """Pop the oldest ready packet, or ``None`` while the codec holds the rest; call until ``None``.""" + return self._inner.receive() + + def flush(self) -> None: + """Mark the end of input, so ``receive`` returns every packet the codec still holds.""" + self._inner.flush() def finish(self, pcm: list[float] | None = None) -> EncoderFinish: """Pad an optional partial frame and drain the codec.""" diff --git a/py/moq-rs/tests/test_local.py b/py/moq-rs/tests/test_local.py index 4d94a03a30c..8ebef27b316 100644 --- a/py/moq-rs/tests/test_local.py +++ b/py/moq-rs/tests/test_local.py @@ -1548,9 +1548,13 @@ async def test_read_frame_keeps_group_across_cancelled_call(): def test_audio_encoder_returns_namespaced_records(): """The codec-only encoder's results are named under ``moq.audio``.""" encoder = moq.audio.Encoder(moq.audio.EncoderOutput(codec=moq.AudioCodec.opus(), sample_rate=48000, channels=1)) - packet = encoder.encode([0.0] * encoder.frame_size) + encoder.encode([0.0] * encoder.frame_size) + packet = encoder.receive() assert isinstance(packet, moq.audio.Packet) assert packet.data + assert encoder.receive() is None + encoder.flush() + assert encoder.receive() is None finish = encoder.finish() assert isinstance(finish, moq.audio.EncoderFinish) @@ -1566,7 +1570,9 @@ def test_audio_encoder_frame_init_and_bitrate(): assert init.format == moq.media.AudioFormat.OPUS assert init.data[:8] == b"OpusHead" encoder.set_bitrate(32_000) - assert encoder.encode([0.0] * encoder.frame_size).data + encoder.encode([0.0] * encoder.frame_size) + packet = encoder.receive() + assert packet is not None and packet.data with pytest.raises(moq.Error.Audio): moq.audio.Encoder(replace(output, frame_duration=timedelta(milliseconds=15))) diff --git a/quest/m2/README.md b/quest/m2/README.md index 70ef38a0880..22d0013073a 100644 --- a/quest/m2/README.md +++ b/quest/m2/README.md @@ -197,13 +197,13 @@ quests that require them. - [Unbounded groups](/quest/m2/unbounded-groups/README.md) - groups and frames lose their size caps; a group keeps a 32 MB tail, and readers behind it replay through paced flights or refresh - [moq-quic drops inherited surfaces with no consumer](/quest/m2/moq-quic-unused-surface.md) - unused accept, connection, and stream-id surfaces leave moq-quic's public API before the cut - [No sentinel values in public API](/quest/m2/no-sentinels.md) - receive windows, cache durations, reconnect timeouts, and the FFI test clock use `Option` instead of a huge value or zero +- [moq-video codecs split submission from receive](/quest/m2/video-codec-receive.md) - video encoders and decoders submit, then `receive` one output, the shape moq-audio took, so no call allocates a `Vec` of outputs - [Connection state carries each phase's data](/quest/m2/connection-state-phases.md) - moq-tokio's `Observation`, moq-ffi's `MoqConnectionState`, and moq-c's `moq_connection_state` each hold one phase per value, so no field is readable in a phase where it means nothing - [moq-video takes devices, not environment variables](/quest/m2/video-devices.md) - VA-API and V4L2 device paths are config fields, and `moq_video::ENV` never ships - [Gateways share one ingest config and one rate cap](/quest/m2/gateway-ingest.md) - `moq_mux::Ingest` replaces 18 per-gateway setters, and the live rate cap moves to `moq_net::bandwidth` - [JS time values are types, not numbers](/quest/m2/js-time-types.md) - every public JS duration and instant is a `Clock.Duration` or `Clock.Instant` object, so a unit mistake fails to compile - [Small API fixes before the cut](/quest/m2/api-shape/README.md) - one path and one spelling per item, names by role, and no dead or wrong-role public surface across net, transport, media, native, apps, JS, and bindings - [Public constructors spawn nothing, or say why](/quest/m2/spawn-constructors.md) - each public library entry point that spawns returns a driver its caller polls, or is listed as kept -- [moq-audio codecs return zero or more outputs](/quest/m2/audio-codec-seam.md) - `encode` and `decode` return any number of outputs and gain `flush`, as moq-video's do - [Plan: recovery boundaries](/quest/m2/intra-refresh-boundary.md) - the catalog's new `warmup` contract is settled before the cut publishes it - [MoQ on rind](/quest/m1/rind/README.md) - every MoQ crate takes its crypto from rind, a pure-Rust graviola fork, so no build compiles C crypto (held in m1 by an open claim) - [Finishing a producer returns a Finished handle](/quest/m2/finished-handle.md) - `finish(self)` consumes moq-net's and the media crates' producers, so appending after finish doesn't compile diff --git a/quest/m2/audio-codec-seam.md b/quest/m2/audio-codec-seam.md deleted file mode 100644 index 516e03c47ac..00000000000 --- a/quest/m2/audio-codec-seam.md +++ /dev/null @@ -1,26 +0,0 @@ -# [S] moq-audio codecs return zero or more outputs - -## Goal - -moq-audio's `Encoder::encode` and `Decoder::decode` return zero or more -packets or frames, and each has a `flush`, the shape moq-video's codecs -already use. Today `Encoder::encode(&[f32]) -> Result` -(`rs/moq-audio/src/encode/encoder.rs`) and `Decoder::decode(&[u8]) -> -Result` (`rs/moq-audio/src/decode/decoder.rs`) return exactly one, -unchanged since `release`. libopus and PCM keep producing one per call. - -## Plan - -Decided 2026-11-07 (maintainer, /quest-plan from the API sweep of open PRs -and remaining quests): ✅ reshape in this breaking cut (rejected: leaving it -to the m5 backends). Every planned platform backend (AudioToolbox, -MediaCodec, Media Foundation, and WebCodecs in m5's audio encode and decode -quests) emits zero or more outputs per call and needs a flush, so each would -otherwise break `encode`'s one-packet return in its own release. - -Mirror moq-video's names and return types. Prove delayed and multiple -outputs and flushing in the shared encoder and decoder tests with a test -codec that buffers, since no backend in this repository does yet; the m5 -platform backends rely on those tests rather than each owning them. moq-ffi, moq-c, and the wrappers -follow if they expose the codec. Bench the libopus path before and after, -since it allocates per call if the return becomes a `Vec`. diff --git a/quest/m2/video-codec-receive.md b/quest/m2/video-codec-receive.md new file mode 100644 index 00000000000..1f695e86daa --- /dev/null +++ b/quest/m2/video-codec-receive.md @@ -0,0 +1,17 @@ +# [L] moq-video codecs split submission from receive + +## Goal + +moq-video's encoders and decoders take the same send/receive shape moq-audio took in [moq-dev/moq#8270](https://github.com/moq-dev/moq/pull/8270): `encode`/`decode` only submit and return `Result<()>`, `receive` pops one ready output (`Ok(None)` when nothing is ready), and `flush` marks the end of input so `receive` drains what the codec held. +No call allocates a `Vec` just to hand back its outputs. + +## Plan + +- Decided 2026-11-08 (maintainer, on #8270): the FFmpeg/MediaCodec split, rejecting a caller-owned `&mut Vec` buffer and keeping `Vec`. moq-video still returns `Vec` from its backend traits, `encode::Sink`, and the decoder front end, so the two crates disagree until this lands. +- Apply it to every backend (VA-API, NVDEC, Media Foundation, VideoToolbox, MediaCodec, openh264, and the rest under `rs/moq-video/src/{encode,decode}/backend/`), the front ends, and their tests. A backend that readies output synchronously holds it in a reused `VecDeque`. +- Mirror it through moq-ffi (`MoqVideoEncoder.encode`/`flush`/`finish`), its Python, Swift, Go, Dart, and Kotlin wrappers, moq-c (`moq_video_encoder_encode`/`flush`, with a `moq_video_encoder_receive` like `moq_audio_encoder_receive`), and the OBS video encoder. +- Benchmark the encode and decode paths base vs branch; allocations should drop or match. + +## Related + +- [moq-dev/moq#8270](https://github.com/moq-dev/moq/pull/8270) - the moq-audio change whose shape this copies diff --git a/quest/m5/audio-decode-mediacodec.md b/quest/m5/audio-decode-mediacodec.md index e7482483642..544a2034087 100644 --- a/quest/m5/audio-decode-mediacodec.md +++ b/quest/m5/audio-decode-mediacodec.md @@ -28,7 +28,7 @@ behind a new optional audio `mediacodec` feature and the decode seam, on `target - The binding ships in the moq-ffi Android slice, which is how Kotlin and Dart reach it. -Changed 2026-11-07 (maintainer, /quest-plan from the API sweep of open PRs and remaining quests): [audio-codec-seam](/quest/m2/audio-codec-seam.md) makes moq-audio's codecs return zero or more outputs with `flush` before the cut, so this backend adds no break. +Changed 2026-11-07 (maintainer, /quest-plan from the API sweep of open PRs and remaining quests): moq-audio's codecs return zero or more outputs with `flush` before the cut, so this backend adds no break. `decode::Consumer` still places each packet by the one output it returns and refuses any other count, so if this backend holds or splits output, it also teaches the consumer to place that output. ## Related diff --git a/quest/m5/audio-decode-mediafoundation.md b/quest/m5/audio-decode-mediafoundation.md index f8f15d2955b..cb53debe31d 100644 --- a/quest/m5/audio-decode-mediafoundation.md +++ b/quest/m5/audio-decode-mediafoundation.md @@ -22,7 +22,7 @@ ones. Behind the decode seam as the first candidate on `target_os = - Verification runs on a Windows host; the per-PR CI only compiles the platform code (`just rs windows`). -Changed 2026-11-07 (maintainer, /quest-plan from the API sweep of open PRs and remaining quests): [audio-codec-seam](/quest/m2/audio-codec-seam.md) makes moq-audio's codecs return zero or more outputs with `flush` before the cut, so this backend adds no break. +Changed 2026-11-07 (maintainer, /quest-plan from the API sweep of open PRs and remaining quests): moq-audio's codecs return zero or more outputs with `flush` before the cut, so this backend adds no break. `decode::Consumer` still places each packet by the one output it returns and refuses any other count, so if this backend holds or splits output, it also teaches the consumer to place that output. ## Related diff --git a/quest/m5/audio-encode-audiotoolbox.md b/quest/m5/audio-encode-audiotoolbox.md index dd27a266390..b1218f6b2af 100644 --- a/quest/m5/audio-encode-audiotoolbox.md +++ b/quest/m5/audio-encode-audiotoolbox.md @@ -30,13 +30,12 @@ the encode seam as the platform candidate on macOS and iOS. the deficit instead, including the terminal end (found in #4081 review). This backend implements the zero-or-more packet return and `flush` that -[audio-codec-seam](/quest/m2/audio-codec-seam.md) lands in m2 before the cut +moq-audio's encoder took in m2 before the cut (decided 2026-11-07, maintainer, /quest-plan from the API sweep of open PRs and remaining quests), so it adds no break and waits on no other encoder backend; its round-trip test still needs its platform decoder. ## Required -- [moq-audio codecs return zero or more outputs](/quest/m2/audio-codec-seam.md) - the encoder API this backend implements - [AudioToolbox decode](/quest/m3/audio-decode-audiotoolbox.md) - the round-trip regression decodes through it ## Related diff --git a/quest/m5/audio-encode-mediacodec.md b/quest/m5/audio-encode-mediacodec.md index 1ebf6a5cfd9..c04e8d50b6e 100644 --- a/quest/m5/audio-encode-mediacodec.md +++ b/quest/m5/audio-encode-mediacodec.md @@ -24,11 +24,10 @@ behind the `mediacodec` feature and the encode seam. device or emulator through `just rs android-test`, as in the decode quest. This backend implements the zero-or-more packet return and `flush` that -[audio-codec-seam](/quest/m2/audio-codec-seam.md) lands in m2 before the cut +moq-audio's encoder took in m2 before the cut (decided 2026-11-07, maintainer, /quest-plan from the API sweep of open PRs and remaining quests), so it adds no break and waits on no other encoder backend; its round-trip test still needs its platform decoder. ## Required -- [moq-audio codecs return zero or more outputs](/quest/m2/audio-codec-seam.md) - the encoder API this backend implements - [MediaCodec decode](/quest/m5/audio-decode-mediacodec.md) - the round-trip regression decodes through it diff --git a/quest/m5/audio-encode-mediafoundation.md b/quest/m5/audio-encode-mediafoundation.md index 45412ce22b3..2ca85484b43 100644 --- a/quest/m5/audio-encode-mediafoundation.md +++ b/quest/m5/audio-encode-mediafoundation.md @@ -19,12 +19,11 @@ behind the encode seam on Windows. a Windows host, since the per-PR CI compiles only. This backend implements the zero-or-more packet return and `flush` that -[audio-codec-seam](/quest/m2/audio-codec-seam.md) lands in m2 before the cut +moq-audio's encoder took in m2 before the cut (decided 2026-11-07, maintainer, /quest-plan from the API sweep of open PRs and remaining quests), so it adds no break and waits on no other encoder backend; its round-trip test still needs its platform decoder. ## Required -- [moq-audio codecs return zero or more outputs](/quest/m2/audio-codec-seam.md) - the encoder API this backend implements - [Media Foundation decode](/quest/m5/audio-decode-mediafoundation.md) - the round-trip regression decodes through it diff --git a/rs/moq-audio/src/decode/backend/libopus.rs b/rs/moq-audio/src/decode/backend/libopus.rs index 07b03607b05..bac068a456d 100644 --- a/rs/moq-audio/src/decode/backend/libopus.rs +++ b/rs/moq-audio/src/decode/backend/libopus.rs @@ -8,6 +8,8 @@ use unsafe_libopus::{ opus_multistream_decoder_create, opus_multistream_decoder_ctl_impl, opus_multistream_decoder_destroy, varargs, }; +use std::collections::VecDeque; + use super::Backend; use crate::decode::Decoded; use crate::{Error, Layout, opus}; @@ -34,6 +36,8 @@ pub(super) struct Libopus { /// packet most likely carried. `None` until one has. last_frame_size: Option, in_dtx: bool, + /// Outputs submitted but not yet received, which reuse their slots across calls. + ready: VecDeque, } // SAFETY: the decoder is owned exclusively and libopus keeps no thread-local state. @@ -101,6 +105,7 @@ impl Libopus { max_frame_size: (opus::DECODE_RATE as usize * MAX_FRAME_MS) / 1000, last_frame_size: None, in_dtx: false, + ready: VecDeque::new(), }; // Same Q7.8 dB units as OpusHead, and it survives OPUS_RESET_STATE. @@ -150,7 +155,7 @@ impl Backend for Libopus { /// An empty packet conceals as much audio as the last packet that decoded /// held, and is refused before any has. Loss during DTX remains classified /// as DTX, while loss during active audio remains active. - fn decode(&mut self, packet: &[u8]) -> Result { + fn decode(&mut self, packet: &[u8]) -> Result<(), Error> { let channels = self.layout.channels() as usize; // libopus conceals exactly `frame_size` samples for a lost packet, and // the decoder can't know how long it was, so assume the last one's length. @@ -195,7 +200,17 @@ impl Backend for Libopus { let activity = opus::multistream_activity(packet, self.streams, self.in_dtx); self.in_dtx = activity.is_dtx(); - Ok(Decoded { samples: out, activity }) + self.ready.push_back(Decoded { samples: out, activity }); + Ok(()) + } + + fn receive(&mut self) -> Result, Error> { + Ok(self.ready.pop_front()) + } + + /// libopus readies each packet's samples as soon as it is submitted. + fn flush(&mut self) -> Result<(), Error> { + Ok(()) } fn reset(&mut self) -> Result<(), Error> { @@ -205,6 +220,7 @@ impl Backend for Libopus { return Err(opus::error(rc, "OPUS_RESET_STATE")); } self.in_dtx = false; + self.ready.clear(); Ok(()) } @@ -363,7 +379,7 @@ mod tests { let mut pcm = Vec::new(); for packet in &packets { - pcm.extend(decoder.decode(packet).unwrap().samples); + pcm.extend(crate::single(decoder.decode_all(packet).unwrap()).samples); } // Skip the encoder's warmup. let pcm = &pcm[pcm.len() / 2..]; @@ -387,13 +403,13 @@ mod tests { fn surround_silence_is_dtx() { let (head, packets) = surround(6); let mut decoder = Decoder::new(&catalog(head.clone(), 6), &Config::default()).unwrap(); - let mid = decoder.decode(&packets[PACKETS / 2]).unwrap(); + let mid = crate::single(decoder.decode_all(&packets[PACKETS / 2]).unwrap()); assert!(mid.activity.is_active(), "a coded surround frame must stay active"); // Two coupled streams, then two mono, each a 20 ms empty frame. let dtx = [0xfc, 0x00, 0xfc, 0x00, 0xf8, 0x00, 0xf8]; let mut decoder = Decoder::new(&catalog(head, 6), &Config::default()).unwrap(); - let decoded = decoder.decode(&dtx).expect("empty multistream packet"); + let decoded = crate::single(decoder.decode_all(&dtx).expect("empty multistream packet")); assert!(decoded.activity.is_dtx(), "all-DTX surround packet read as active"); assert_eq!(decoded.samples.len() % 6, 0); } diff --git a/rs/moq-audio/src/decode/backend/mod.rs b/rs/moq-audio/src/decode/backend/mod.rs index 588622600da..81ae2ac1f83 100644 --- a/rs/moq-audio/src/decode/backend/mod.rs +++ b/rs/moq-audio/src/decode/backend/mod.rs @@ -27,14 +27,24 @@ mod symphonia; /// An opened decoder: packets in, interleaved `f32` PCM out. /// +/// Output comes back in stream order, but a codec that pipelines (MediaCodec) +/// may hold a packet's samples past later submissions. +/// /// Unwind safe because the published `Decoder` that boxes it is. pub(crate) trait Backend: Send + std::panic::UnwindSafe + std::panic::RefUnwindSafe { - /// Decode one packet into interleaved samples at [`sample_rate`](Self::sample_rate) - /// and [`layout`](Self::layout), untrimmed. - fn decode(&mut self, packet: &[u8]) -> Result; + /// Submit one packet. + fn decode(&mut self, packet: &[u8]) -> Result<(), Error>; + + /// Pop the oldest ready output, interleaved at [`sample_rate`](Self::sample_rate) + /// and [`layout`](Self::layout) and untrimmed, or `None` while the codec holds the rest. + fn receive(&mut self) -> Result, Error>; + + /// Mark the end of input, so [`receive`](Self::receive) returns every output + /// the codec still holds. The next packet submitted starts new input. + fn flush(&mut self) -> Result<(), Error>; - /// Drop codec history after a discontinuity, so the next packet does not - /// predict from audio that is no longer adjacent. + /// Drop codec history, held output included, after a discontinuity, so the + /// next packet does not predict from audio that is no longer adjacent. fn reset(&mut self) -> Result<(), Error>; /// The rate this backend decodes to, which may differ from the catalog's. @@ -142,17 +152,21 @@ fn select(catalog: &AudioConfig, candidates: Vec<&Candidate>) -> Result Result { - Ok(Decoded { - samples: Vec::new(), - activity: Activity::Active, - }) + fn decode(&mut self, _packet: &[u8]) -> Result<(), Error> { + Ok(()) + } + + fn receive(&mut self) -> Result, Error> { + Ok(None) + } + + fn flush(&mut self) -> Result<(), Error> { + Ok(()) } fn reset(&mut self) -> Result<(), Error> { diff --git a/rs/moq-audio/src/decode/backend/pcm.rs b/rs/moq-audio/src/decode/backend/pcm.rs index bead86fb5e0..e7ec07a96d7 100644 --- a/rs/moq-audio/src/decode/backend/pcm.rs +++ b/rs/moq-audio/src/decode/backend/pcm.rs @@ -1,5 +1,7 @@ //! Uncompressed little-endian `f32` PCM, described entirely by the catalog. +use std::collections::VecDeque; + use super::Backend; use crate::decode::Decoded; use crate::{Activity, Error, Layout, pcm}; @@ -10,6 +12,8 @@ pub(super) struct Pcm { sample_rate: u32, layout: Layout, bytes_per_frame: usize, + /// Outputs submitted but not yet received, which reuse their slots across calls. + ready: VecDeque, } impl Pcm { @@ -35,12 +39,13 @@ impl Pcm { sample_rate: catalog.sample_rate, layout: Layout::from_channels(catalog.channel_count)?, bytes_per_frame: pcm::frame_bytes(1, catalog.channel_count)?, + ready: VecDeque::new(), })) } } impl Backend for Pcm { - fn decode(&mut self, packet: &[u8]) -> Result { + fn decode(&mut self, packet: &[u8]) -> Result<(), Error> { if packet.is_empty() || !packet.len().is_multiple_of(self.bytes_per_frame) { return Err(Error::Misaligned { got: packet.len(), @@ -54,13 +59,23 @@ impl Backend for Pcm { .iter() .map(|sample| f32::from_le_bytes(*sample)) .collect(); - Ok(Decoded { + self.ready.push_back(Decoded { samples, activity: Activity::Active, - }) + }); + Ok(()) + } + + fn receive(&mut self) -> Result, Error> { + Ok(self.ready.pop_front()) + } + + fn flush(&mut self) -> Result<(), Error> { + Ok(()) } fn reset(&mut self) -> Result<(), Error> { + self.ready.clear(); Ok(()) } diff --git a/rs/moq-audio/src/decode/backend/symphonia.rs b/rs/moq-audio/src/decode/backend/symphonia.rs index 7bcc8ec448e..9de6e181fb9 100644 --- a/rs/moq-audio/src/decode/backend/symphonia.rs +++ b/rs/moq-audio/src/decode/backend/symphonia.rs @@ -3,6 +3,8 @@ use symphonia_core::codecs::audio::AudioDecoder; +use std::collections::VecDeque; + use super::Backend; use crate::decode::Decoded; use crate::{Activity, Error, Layout, aac}; @@ -14,6 +16,8 @@ pub(super) struct Symphonia { sample_rate: u32, layout: Layout, adts: bool, + /// Outputs submitted but not yet received, which reuse their slots across calls. + ready: VecDeque, } impl Symphonia { @@ -60,12 +64,13 @@ impl Symphonia { sample_rate, layout: Layout::from_channels(channel_count as u32)?, adts: catalog.description.is_none() && !matches!(catalog.container, hang::catalog::Container::Cmaf { .. }), + ready: VecDeque::new(), })) } } impl Backend for Symphonia { - fn decode(&mut self, packet: &[u8]) -> Result { + fn decode(&mut self, packet: &[u8]) -> Result<(), Error> { let packet = if self.adts { let header = moq_mux::codec::aac::adts::Header::parse(packet).map_err(|err| Error::Decode(format!("aac: {err}")))?; @@ -99,14 +104,25 @@ impl Backend for Symphonia { let mut samples = Vec::new(); decoded.copy_to_vec_interleaved(&mut samples); - Ok(Decoded { + self.ready.push_back(Decoded { samples, activity: Activity::Active, - }) + }); + Ok(()) + } + + fn receive(&mut self) -> Result, Error> { + Ok(self.ready.pop_front()) + } + + /// Symphonia readies each packet's samples as soon as it is submitted. + fn flush(&mut self) -> Result<(), Error> { + Ok(()) } fn reset(&mut self) -> Result<(), Error> { self.inner.reset(); + self.ready.clear(); Ok(()) } diff --git a/rs/moq-audio/src/decode/conformance.rs b/rs/moq-audio/src/decode/conformance.rs index 8ffecb4c8c8..77a2f0ea01d 100644 --- a/rs/moq-audio/src/decode/conformance.rs +++ b/rs/moq-audio/src/decode/conformance.rs @@ -201,7 +201,9 @@ async fn every_backend_matches_independent_reference() { let mut pcm = Vec::new(); for packet in &packets { - pcm.extend(decoder.decode(packet).unwrap().samples); + decoder.decode(packet).unwrap(); + pcm.extend(decoder.receive().unwrap().expect("one output per packet").samples); + assert!(decoder.receive().unwrap().is_none(), "one output per packet"); } compare(vector, decoder.name(), &pcm[delay..]); } diff --git a/rs/moq-audio/src/decode/consumer.rs b/rs/moq-audio/src/decode/consumer.rs index 80fee172773..754a2856bac 100644 --- a/rs/moq-audio/src/decode/consumer.rs +++ b/rs/moq-audio/src/decode/consumer.rs @@ -406,7 +406,20 @@ impl Consumer { let rate = self.decoder.sample_rate(); let epoch = *self.epoch.get_or_insert(mux_frame.timestamp); let delay = self.decoder.delay_remaining(); - let decoded = self.decoder.decode(&mux_frame.payload)?; + self.decoder.decode(&mux_frame.payload)?; + // Each packet is placed by its own samples, which a decoder that holds or splits + // its output would hand back somewhere else. + let contract = |name: &str| { + Error::Codec(format!( + "{name} did not ready exactly one output for one packet, and playback places exactly one" + )) + }; + let Some(decoded) = self.decoder.receive()? else { + return Err(contract(self.decoder.name())); + }; + if self.decoder.receive()?.is_some() { + return Err(contract(self.decoder.name())); + } // Anchored only once a packet decodes: a refused one primed nothing. The track's // first packet has no history to warm up from: the codec's pre-skip, trimmed by // the decoder, is all the priming its encoder declared. @@ -1281,7 +1294,7 @@ mod tests { .write(moq_mux::container::Frame { timestamp: Timestamp::from_scale(packet * encoder.frame_size() as u64, 48_000).unwrap(), duration: None, - payload: encoder.encode(&pcm).unwrap().payload, + payload: crate::single(encoder.encode_all(&pcm).unwrap()).payload, keyframe: true, }) .unwrap(); @@ -1822,7 +1835,7 @@ mod tests { let silence = vec![0.0; encoder.frame_size()]; let mut first_dtx = None; for index in 0..40u64 { - let packet = encoder.encode(if index == 0 { &active } else { &silence }).unwrap(); + let packet = crate::single(encoder.encode_all(if index == 0 { &active } else { &silence }).unwrap()); let timestamp = Timestamp::from_scale(index * encoder.frame_size() as u64, 48_000).unwrap(); if first_dtx.is_none() && packet.activity.is_dtx() { first_dtx = Some(timestamp); @@ -2017,11 +2030,11 @@ mod tests { // One coded packet, then a withheld one to follow it. Taken from the same // encoder rather than published in between, so nothing fills the resampler's // chunk between the two. - let active = encoder.encode(&vec![0.5f32; encoder.frame_size()]).unwrap(); + let active = crate::single(encoder.encode_all(&vec![0.5f32; encoder.frame_size()]).unwrap()); assert!(active.activity.is_active()); let silence = vec![0.0f32; encoder.frame_size()]; let dtx = (0..200) - .map(|_| encoder.encode(&silence).unwrap()) + .map(|_| crate::single(encoder.encode_all(&silence).unwrap())) .find(|packet| packet.activity.is_dtx()) .expect("silence should enter Opus DTX"); @@ -2166,7 +2179,7 @@ mod tests { .write(moq_mux::container::Frame { timestamp, duration: None, - payload: encoder.encode(&pcm).unwrap().payload, + payload: crate::single(encoder.encode_all(&pcm).unwrap()).payload, keyframe: true, }) .unwrap(); @@ -2359,7 +2372,7 @@ mod tests { .write(moq_mux::container::Frame { timestamp: Timestamp::from_micros(0).unwrap(), duration: None, - payload: encoder.encode(&vec![0.25f32; encoder.frame_size()]).unwrap().payload, + payload: crate::single(encoder.encode_all(&vec![0.25f32; encoder.frame_size()]).unwrap()).payload, keyframe: true, }) .unwrap(); @@ -2475,7 +2488,7 @@ mod tests { .write(moq_mux::container::Frame { timestamp: Timestamp::from_scale(packet * encoder.frame_size() as u64, 48_000).unwrap(), duration: None, - payload: encoder.encode(&pcm).unwrap().payload, + payload: crate::single(encoder.encode_all(&pcm).unwrap()).payload, keyframe: true, }) .unwrap(); @@ -2538,10 +2551,22 @@ mod tests { .unwrap(); producer.cut(None).unwrap(); }; - write(&mut producer, 0, encoder.encode(&pcm).unwrap().payload); - write(&mut producer, 1, encoder.encode(&pcm).unwrap().payload); + write( + &mut producer, + 0, + crate::single(encoder.encode_all(&pcm).unwrap()).payload, + ); + write( + &mut producer, + 1, + crate::single(encoder.encode_all(&pcm).unwrap()).payload, + ); producer.discontinuity().unwrap(); - write(&mut producer, 2, encoder.encode(&pcm).unwrap().payload); + write( + &mut producer, + 2, + crate::single(encoder.encode_all(&pcm).unwrap()).payload, + ); producer.finish().unwrap(); let first = next_frame(&mut consumer).await.unwrap().expect("first decoded frame"); @@ -2625,7 +2650,7 @@ mod tests { // Every packet counts here; the warmup tests cover trimming. catalog.warmup = None; let pcm = vec![0.0f32; encoder.frame_size() * encoder.codec_channels() as usize]; - let packet = encoder.encode(&pcm).unwrap(); + let packet = crate::single(encoder.encode_all(&pcm).unwrap()); // Re-describe the same rendition as CMAF and publish it that way. let muxer = moq_mux::container::fmp4::Muxer::audio(&catalog).unwrap(); diff --git a/rs/moq-audio/src/decode/decoded.rs b/rs/moq-audio/src/decode/decoded.rs index b20e4546f26..dd14777cd96 100644 --- a/rs/moq-audio/src/decode/decoded.rs +++ b/rs/moq-audio/src/decode/decoded.rs @@ -1,9 +1,10 @@ -//! [`Decoded`]: one packet's worth of PCM on its way out of a codec. +//! [`Decoded`]: one run of PCM on its way out of a codec. use crate::Activity; -/// One decoded Opus/PCM/AAC packet: interleaved `f32` samples at the codec's own -/// rate and layout, plus what the codec was doing when it produced them. +/// One decoder output, a packet's worth for libopus, PCM, and symphonia: +/// interleaved `f32` samples at the codec's own rate and layout, plus what the +/// codec was doing when it produced them. /// /// The low-level counterpart to [`encode::Encoded`](crate::encode::Encoded). /// [`Consumer`](super::Consumer) turns these into [`Frame`](crate::Frame)s in diff --git a/rs/moq-audio/src/decode/decoder.rs b/rs/moq-audio/src/decode/decoder.rs index 462981480fa..6bdb5784fe3 100644 --- a/rs/moq-audio/src/decode/decoder.rs +++ b/rs/moq-audio/src/decode/decoder.rs @@ -105,22 +105,63 @@ impl Decoder { self.delay_remaining } - /// Decode one packet into interleaved `f32` PCM and report its codec activity. + /// Submit one packet, whose interleaved `f32` PCM [`receive`](Self::receive) + /// returns. + /// + /// A codec that pipelines holds a packet's samples past later packets; + /// libopus, PCM, and symphonia ready exactly one output per packet at once. + /// Call [`flush`](Self::flush) after the last packet for whatever is still held. /// /// An empty Opus packet marks one lost packet and conceals as much audio as the /// last packet that decoded held, so a lost 20 ms packet yields 20 ms. It is /// refused with [`Error::Decode`] before any packet has decoded, since there is /// no length to conceal. Loss during DTX remains classified as DTX, while loss /// during active audio remains active. - pub fn decode(&mut self, packet: &[u8]) -> Result { - let mut decoded = self.backend.decode(packet)?; + pub fn decode(&mut self, packet: &[u8]) -> Result<(), Error> { + self.backend.decode(packet) + } + + /// Pop the oldest ready output in stream order, with its codec activity and + /// any startup delay left trimmed off, or `None` while the codec holds the rest. + /// + /// Call until `None` after each [`decode`](Self::decode) or [`flush`](Self::flush). + pub fn receive(&mut self) -> Result, Error> { + let Some(mut output) = self.backend.receive()? else { + return Ok(None); + }; let channels = self.backend.layout().channels() as usize; - let trim = self.delay_remaining.min(decoded.samples.len() / channels); + let trim = self.delay_remaining.min(output.samples.len() / channels); if trim > 0 { - decoded.samples.drain(..trim * channels); + output.samples.drain(..trim * channels); self.delay_remaining -= trim; } - Ok(decoded) + Ok(Some(output)) + } + + /// Mark the end of input, once the stream has ended or before a boundary its + /// output has to respect, so [`receive`](Self::receive) returns every output + /// the codec still holds. Packets decoded afterwards start new input. + pub fn flush(&mut self) -> Result<(), Error> { + self.backend.flush() + } + + /// Submit `packet` and take every output it readied. + #[cfg(test)] + pub(crate) fn decode_all(&mut self, packet: &[u8]) -> Result, Error> { + self.decode(packet)?; + self.ready() + } + + /// Flush and take every output the codec held. + #[cfg(test)] + pub(crate) fn flush_all(&mut self) -> Result, Error> { + self.flush()?; + self.ready() + } + + #[cfg(test)] + fn ready(&mut self) -> Result, Error> { + std::iter::from_fn(|| self.receive().transpose()).collect() } } @@ -155,6 +196,112 @@ pub(crate) mod tests { catalog } + /// Decodes each byte to one mono sample of its value, and pipelines like a + /// platform decoder: it holds every other packet's output and releases it with + /// the next, so submissions alternate between readying no output and two. + struct Pipelined { + held: Option, + ready: std::collections::VecDeque, + } + + /// Frames of startup delay [`Pipelined`] declares. + const PIPELINED_DELAY: usize = 3; + + impl Backend for Pipelined { + fn decode(&mut self, packet: &[u8]) -> Result<(), Error> { + let output = Decoded { + samples: packet.iter().map(|&byte| f32::from(byte)).collect(), + activity: crate::Activity::Active, + }; + match self.held.take() { + Some(held) => self.ready.extend([held, output]), + None => self.held = Some(output), + } + Ok(()) + } + + fn receive(&mut self) -> Result, Error> { + Ok(self.ready.pop_front()) + } + + fn flush(&mut self) -> Result<(), Error> { + self.ready.extend(self.held.take()); + Ok(()) + } + + fn reset(&mut self) -> Result<(), Error> { + self.held = None; + self.ready.clear(); + Ok(()) + } + + fn sample_rate(&self) -> u32 { + 48_000 + } + + fn layout(&self) -> Layout { + Layout::Mono + } + + fn delay(&self) -> usize { + PIPELINED_DELAY + } + + fn name(&self) -> &str { + "pipelined" + } + } + + fn pipelined() -> Decoder { + Decoder { + backend: Box::new(Pipelined { + held: None, + ready: Default::default(), + }), + delay: PIPELINED_DELAY, + delay_remaining: PIPELINED_DELAY, + } + } + + fn samples(outputs: Vec) -> Vec> { + outputs.into_iter().map(|output| output.samples).collect() + } + + /// Held output comes back in order, the startup delay is trimmed across + /// outputs as they emerge rather than as packets arrive, and `flush` returns + /// the tail while leaving the decoder usable. + #[test] + fn held_output_is_trimmed_in_order_and_flushed() { + let mut decoder = pipelined(); + assert!(decoder.decode_all(&[1, 1]).unwrap().is_empty()); + assert_eq!(decoder.delay_remaining(), PIPELINED_DELAY, "nothing emerged to trim"); + + // Three frames of delay take all of the first output and one of the second. + assert_eq!(samples(decoder.decode_all(&[2, 2]).unwrap()), [vec![], vec![2.0]]); + assert_eq!(decoder.delay_remaining(), 0); + + assert!(decoder.decode_all(&[3, 3]).unwrap().is_empty()); + assert_eq!(samples(decoder.flush_all().unwrap()), [vec![3.0, 3.0]]); + assert!(decoder.flush_all().unwrap().is_empty()); + + assert!(decoder.decode_all(&[4]).unwrap().is_empty()); + assert_eq!(samples(decoder.decode_all(&[5]).unwrap()), [vec![4.0], vec![5.0]]); + } + + /// A reset drops held output and reapplies the startup delay to what follows. + #[test] + fn reset_drops_held_output() { + let mut decoder = pipelined(); + assert!(decoder.decode_all(&[1; 4]).unwrap().is_empty()); + assert_eq!(samples(decoder.decode_all(&[2; 4]).unwrap()), [vec![1.0], vec![2.0; 4]]); + assert!(decoder.decode_all(&[3; 4]).unwrap().is_empty()); + + decoder.reset().unwrap(); + assert!(decoder.flush_all().unwrap().is_empty()); + assert!(decoder.decode_all(&[4; 4]).unwrap().is_empty()); + assert_eq!(samples(decoder.flush_all().unwrap()), [vec![4.0]]); + } + /// The published `Decoder` was unwind safe before it boxed a backend; keep it so. #[test] fn decoder_is_unwind_safe() { @@ -172,7 +319,7 @@ pub(crate) mod tests { let decoded: Vec> = AAC_FRAMES .iter() - .map(|frame| decoder.decode(frame).unwrap().samples) + .map(|frame| crate::single(decoder.decode_all(frame).unwrap()).samples) .collect(); // AAC-LC frames are 1024 samples each, whatever the packet size. @@ -231,8 +378,8 @@ pub(crate) mod tests { packet.extend_from_slice(&[0, 0]); } packet.extend_from_slice(frame); - let got = adts.decode(&packet).unwrap().samples; - let expected = raw.decode(frame).unwrap().samples; + let got = crate::single(adts.decode_all(&packet).unwrap()).samples; + let expected = crate::single(raw.decode_all(frame).unwrap()).samples; assert!(got == expected, "ADTS framing changed decoded PCM"); for altered in [2, 3, 6] { let mut invalid = packet.clone(); @@ -262,7 +409,10 @@ pub(crate) mod tests { init: bytes::Bytes::new(), }; let mut decoder = Decoder::new(&catalog, &Config::default()).unwrap(); - assert_eq!(decoder.decode(AAC_FRAMES[0]).unwrap().samples.len(), 1024); + assert_eq!( + crate::single(decoder.decode_all(AAC_FRAMES[0]).unwrap()).samples.len(), + 1024 + ); } /// A packet libopus rejects is that packet's problem, not the @@ -290,7 +440,7 @@ pub(crate) mod tests { let pcm: Vec = (packet * frames..(packet + 1) * frames) .map(|i| (std::f32::consts::TAU * 440.0 * i as f32 / 48_000.0).sin() * 0.5) .collect(); - encoder.encode(&pcm).unwrap().payload + crate::single(encoder.encode_all(&pcm).unwrap()).payload }) .collect() } @@ -318,8 +468,14 @@ pub(crate) mod tests { assert_eq!(decoder.sample_rate(), 48_000, "input rate {input_rate}"); // The pre-skip is 48 kHz samples, trimmed once. - assert_eq!(decoder.decode(&packets[0]).unwrap().samples.len(), 960 - 312); - assert_eq!(decoder.decode(&packets[1]).unwrap().samples.len(), 960); + assert_eq!( + crate::single(decoder.decode_all(&packets[0]).unwrap()).samples.len(), + 960 - 312 + ); + assert_eq!( + crate::single(decoder.decode_all(&packets[1]).unwrap()).samples.len(), + 960 + ); } } @@ -338,12 +494,12 @@ pub(crate) mod tests { assert!(matches!(decoder.decode(&[]), Err(Error::Decode(_)))); for packet in &packets { - decoder.decode(packet).unwrap(); + crate::single(decoder.decode_all(packet).unwrap()); } - assert_eq!(decoder.decode(&[]).unwrap().samples.len(), 960); + assert_eq!(crate::single(decoder.decode_all(&[]).unwrap()).samples.len(), 960); // Concealment doesn't change the length: the next loss is the same size. - assert_eq!(decoder.decode(&[]).unwrap().samples.len(), 960); + assert_eq!(crate::single(decoder.decode_all(&[]).unwrap()).samples.len(), 960); } #[test] @@ -355,11 +511,11 @@ pub(crate) mod tests { let mut decoder = Decoder::new(&opus_catalog(head), &Config::default()).unwrap(); let mut last = Vec::new(); for packet in &packets { - last = decoder.decode(packet).unwrap().samples; + last = crate::single(decoder.decode_all(packet).unwrap()).samples; } // A reset after loss keeps the gain. decoder.reset().unwrap(); - let reset = decoder.decode(&packets[4]).unwrap().samples; + let reset = crate::single(decoder.decode_all(&packets[4]).unwrap()).samples; (rms(&last), rms(&reset)) }; @@ -413,7 +569,10 @@ pub(crate) mod tests { let catalog = hang::catalog::AudioConfig::new(hang::catalog::AudioCodec::Opus, 24_000, 1); let mut decoder = Decoder::new(&catalog, &Config::default()).unwrap(); assert_eq!(decoder.sample_rate(), 48_000); - assert_eq!(decoder.decode(&packets[0]).unwrap().samples.len(), 960); + assert_eq!( + crate::single(decoder.decode_all(&packets[0]).unwrap()).samples.len(), + 960 + ); let catalog = hang::catalog::AudioConfig::new(hang::catalog::AudioCodec::Opus, 48_000, 6); assert!(matches!( diff --git a/rs/moq-audio/src/encode/backend/libopus.rs b/rs/moq-audio/src/encode/backend/libopus.rs index b8fc8a7ffce..8f0fc0d27f4 100644 --- a/rs/moq-audio/src/encode/backend/libopus.rs +++ b/rs/moq-audio/src/encode/backend/libopus.rs @@ -1,5 +1,7 @@ //! Opus through libopus 1.3.1, via [`unsafe_libopus`]. +use std::collections::VecDeque; + use bytes::Bytes; use unsafe_libopus::{ OPUS_APPLICATION_AUDIO, OPUS_GET_BITRATE_REQUEST, OPUS_GET_LOOKAHEAD_REQUEST, OPUS_OK, OPUS_RESET_STATE, @@ -19,6 +21,8 @@ const MAX_PACKET_BYTES: usize = 4_000; pub(super) struct Libopus { inner: *mut OpusEncoder, scratch: Vec, + /// Packets submitted but not yet received, which reuse their slots across calls. + ready: VecDeque, sample_rate: u32, channels: u32, frame_size: usize, @@ -61,6 +65,7 @@ impl Libopus { let mut backend = Self { inner, scratch: vec![0u8; MAX_PACKET_BYTES], + ready: VecDeque::new(), sample_rate, channels, frame_size, @@ -118,7 +123,7 @@ impl Libopus { } impl Backend for Libopus { - fn encode(&mut self, pcm: &[f32]) -> Result { + fn encode(&mut self, pcm: &[f32]) -> Result<(), Error> { // SAFETY: `inner` owns a live OpusEncoder; pcm and scratch slices are // bounded by the lengths we pass, and the front end sized `pcm` to one frame. let n = unsafe { @@ -135,13 +140,24 @@ impl Backend for Libopus { } let payload = Bytes::copy_from_slice(&self.scratch[..n as usize]); let activity = opus::activity(&payload, false); - Ok(Encoded { payload, activity }) + self.ready.push_back(Encoded { payload, activity }); + Ok(()) + } + + fn receive(&mut self) -> Result, Error> { + Ok(self.ready.pop_front()) + } + + /// libopus readies each frame's packet as soon as it is submitted. + fn flush(&mut self) -> Result<(), Error> { + Ok(()) } fn reset(&mut self) { // SAFETY: `inner` owns a live encoder and OPUS_RESET_STATE takes no arguments. let rc = unsafe { opus_encoder_ctl_impl(self.inner, OPUS_RESET_STATE, varargs![]) }; debug_assert_eq!(rc, OPUS_OK, "OPUS_RESET_STATE failed with {rc}"); + self.ready.clear(); } fn set_bitrate(&mut self, bitrate: u64) -> Result<(), Error> { diff --git a/rs/moq-audio/src/encode/backend/mod.rs b/rs/moq-audio/src/encode/backend/mod.rs index 7cd7eb031ba..b12df50800b 100644 --- a/rs/moq-audio/src/encode/backend/mod.rs +++ b/rs/moq-audio/src/encode/backend/mod.rs @@ -24,22 +24,29 @@ mod pcm; #[cfg(test)] pub(crate) mod stub; -/// An opened encoder: one frame of interleaved `f32` PCM in, one packet out. +/// An opened encoder: frames of interleaved `f32` PCM in, packets out. /// /// Input arrives at the config's rate, in the crate's canonical channel order /// for the config's layout; a codec with another native order reorders it here. /// -/// One packet per frame is part of the contract: the producer stamps packets by -/// counting frames. A codec that pipelines output (MediaCodec) needs a `flush` -/// and a zero-or-more return, which changes `Encoder::encode` too, so that lands -/// with the first backend that needs it rather than as an always-empty method. +/// Every frame becomes exactly one packet, in submission order, but a codec that +/// pipelines (MediaCodec, AudioToolbox) may hold it past later submissions: the +/// producer stamps each packet with the frame it pairs with by that order. /// /// Unwind safe because the published `Encoder` that boxes it is. pub(crate) trait Backend: Send + std::panic::UnwindSafe + std::panic::RefUnwindSafe { - /// Encode exactly one frame of the codec's frame size. - fn encode(&mut self, pcm: &[f32]) -> Result; + /// Submit exactly one frame of the codec's frame size. + fn encode(&mut self, pcm: &[f32]) -> Result<(), Error>; - /// Drop codec history so the next frame codes as if it were the first. + /// Pop the oldest packet that is ready, or `None` while the codec holds the rest. + fn receive(&mut self) -> Result, Error>; + + /// Mark the end of input, so [`receive`](Self::receive) returns every packet + /// the codec still holds. The next frame submitted starts new input. + fn flush(&mut self) -> Result<(), Error>; + + /// Drop codec history, held packets included, so the next frame codes as if + /// it were the first. fn reset(&mut self); /// Retune the live encoder to `bitrate` bits per second, a no-op at the current @@ -166,8 +173,16 @@ mod tests { struct Fake(&'static str); impl Backend for Fake { - fn encode(&mut self, _pcm: &[f32]) -> Result { - Ok(Encoded::new(bytes::Bytes::new())) + fn encode(&mut self, _pcm: &[f32]) -> Result<(), Error> { + Ok(()) + } + + fn receive(&mut self) -> Result, Error> { + Ok(None) + } + + fn flush(&mut self) -> Result<(), Error> { + Ok(()) } fn reset(&mut self) {} diff --git a/rs/moq-audio/src/encode/backend/pcm.rs b/rs/moq-audio/src/encode/backend/pcm.rs index 2892cb92cc5..dfcbab6856d 100644 --- a/rs/moq-audio/src/encode/backend/pcm.rs +++ b/rs/moq-audio/src/encode/backend/pcm.rs @@ -1,5 +1,7 @@ //! Uncompressed little-endian `f32` PCM, which needs no codec at all. +use std::collections::VecDeque; + use super::Backend; use crate::encode::{Config, Encoded}; use crate::{Error, Layout, pcm}; @@ -8,6 +10,8 @@ pub(super) const NAME: &str = "pcm"; pub(super) struct Pcm { bitrate: u64, + /// Packets submitted but not yet received, which reuse their slots across calls. + ready: VecDeque, } impl Pcm { @@ -22,20 +26,34 @@ impl Pcm { ))); } let bitrate = pcm::bitrate(config.sample_rate, config.layout.channels())?; - Ok(Box::new(Self { bitrate })) + Ok(Box::new(Self { + bitrate, + ready: VecDeque::new(), + })) } } impl Backend for Pcm { - fn encode(&mut self, pcm: &[f32]) -> Result { + fn encode(&mut self, pcm: &[f32]) -> Result<(), Error> { let mut payload = Vec::with_capacity(std::mem::size_of_val(pcm)); for sample in pcm { payload.extend_from_slice(&sample.to_le_bytes()); } - Ok(Encoded::new(payload.into())) + self.ready.push_back(Encoded::new(payload.into())); + Ok(()) + } + + fn receive(&mut self) -> Result, Error> { + Ok(self.ready.pop_front()) } - fn reset(&mut self) {} + fn flush(&mut self) -> Result<(), Error> { + Ok(()) + } + + fn reset(&mut self) { + self.ready.clear(); + } fn set_bitrate(&mut self, _bitrate: u64) -> Result<(), Error> { Err(Error::Unsupported("pcm bitrate is fixed".into())) diff --git a/rs/moq-audio/src/encode/backend/stub.rs b/rs/moq-audio/src/encode/backend/stub.rs index 52689593377..29eac026215 100644 --- a/rs/moq-audio/src/encode/backend/stub.rs +++ b/rs/moq-audio/src/encode/backend/stub.rs @@ -3,6 +3,7 @@ //! real tiers. use std::cell::RefCell; +use std::collections::VecDeque; use std::sync::{Arc, Weak}; use bytes::Bytes; @@ -49,10 +50,16 @@ impl Drop for Installed { } /// Emits each frame's index as its payload, and cannot retune. +/// +/// Pipelines like a platform encoder: it holds every other frame and releases +/// it with the next, so submissions alternate between readying no packet and two, +/// and a tail waits for [`flush`](Backend::flush). pub(crate) struct Stub { _resource: Arc<()>, bitrate: u64, frames: u64, + held: Option, + ready: VecDeque, } impl Stub { @@ -61,19 +68,36 @@ impl Stub { _resource: INSTALLED.with(|installed| installed.borrow().upgrade().unwrap()), bitrate: config.bitrate.map_or(128_000, |rate| rate.as_bps()), frames: 0, + held: None, + ready: VecDeque::new(), })) } } impl Backend for Stub { - fn encode(&mut self, _pcm: &[f32]) -> Result { - let payload = Bytes::copy_from_slice(&self.frames.to_be_bytes()); + fn encode(&mut self, _pcm: &[f32]) -> Result<(), Error> { + let packet = Encoded::new(Bytes::copy_from_slice(&self.frames.to_be_bytes())); self.frames += 1; - Ok(Encoded::new(payload)) + match self.held.take() { + Some(held) => self.ready.extend([held, packet]), + None => self.held = Some(packet), + } + Ok(()) + } + + fn receive(&mut self) -> Result, Error> { + Ok(self.ready.pop_front()) + } + + fn flush(&mut self) -> Result<(), Error> { + self.ready.extend(self.held.take()); + Ok(()) } fn reset(&mut self) { self.frames = 0; + self.held = None; + self.ready.clear(); } fn set_bitrate(&mut self, bitrate: u64) -> Result<(), Error> { diff --git a/rs/moq-audio/src/encode/encoder.rs b/rs/moq-audio/src/encode/encoder.rs index c33252db14e..76719d1779d 100644 --- a/rs/moq-audio/src/encode/encoder.rs +++ b/rs/moq-audio/src/encode/encoder.rs @@ -288,8 +288,13 @@ fn aac_config(config: &Config) -> Result { /// /// Build one with [`Encoder::new`], feed full PCM frames via /// [`encode`](Self::encode), then pass the trailing partial frame to -/// [`finish`](Self::finish). Publish every packet either call returns and apply -/// the terminal [`Finish::discard_padding`] when the container supports it. +/// [`finish`](Self::finish). Publish every packet either call returns, in order, +/// and apply the terminal [`Finish::discard_padding`] when the container +/// supports it. +/// +/// Every frame becomes exactly one packet, in order, though a codec that +/// pipelines returns it from a later call, so the `n`th packet always starts +/// `n` frames into the stream. pub struct Encoder { backend: Box, config: Config, @@ -412,7 +417,8 @@ impl Encoder { Ok(()) } - /// Drop all codec history so a later epoch cannot emit audio from this one. + /// Drop all codec history, held packets included, so a later epoch cannot + /// emit audio from this one. pub(super) fn reset(&mut self) { self.backend.reset(); self.started = false; @@ -437,12 +443,16 @@ impl Encoder { } } - /// Encode one frame of interleaved `f32` PCM at [`codec_rate`](Self::codec_rate). + /// Submit one frame of interleaved `f32` PCM at [`codec_rate`](Self::codec_rate), + /// whose packet [`receive`](Self::receive) returns. + /// + /// A codec that pipelines holds the packet past later frames; libopus and + /// PCM ready it at once. /// /// `pcm.len()` must equal `frame_size() * codec_channels()`. The /// [`Producer`](super::Producer) handles format conversion and resampling /// before calling this; for direct use, the caller does the same. - pub fn encode(&mut self, pcm: &[f32]) -> Result { + pub fn encode(&mut self, pcm: &[f32]) -> Result<(), Error> { let expected = self.frame_size * self.codec_channels() as usize; if pcm.len() != expected { return Err(Error::Misaligned { @@ -450,16 +460,34 @@ impl Encoder { expected: expected * std::mem::size_of::(), }); } - let encoded = self.backend.encode(pcm)?; + self.backend.encode(pcm)?; self.started = true; - Ok(encoded) + Ok(()) + } + + /// Pop the oldest ready packet, or `None` while the codec holds the rest. + /// + /// Call until `None` after each [`encode`](Self::encode) or [`flush`](Self::flush). + pub fn receive(&mut self) -> Result, Error> { + self.backend.receive() + } + + /// Mark the end of input, so [`receive`](Self::receive) returns every packet + /// the codec still holds. Frames encoded afterwards start new input. + /// + /// Pads nothing, so the codec's lookahead stays inside it: reach for this at a + /// boundary the packets have to respect, and for [`finish`](Self::finish) at + /// the end of the stream. + pub fn flush(&mut self) -> Result<(), Error> { + self.backend.flush() } /// Finish encoding, zero-padding `pcm` as the final partial frame and /// returning every packet needed to drain codec lookahead. /// /// `pcm` is interleaved at [`codec_rate`](Self::codec_rate), may be empty, - /// and must contain at most one frame. Silence added here only drains + /// and must contain at most one frame. The packets start with any not yet + /// [received](Self::receive) from earlier [`encode`](Self::encode) calls. Silence added here only drains /// audio already supplied; [`Finish::discard_padding`] reports how much of /// the decoded tail is artificial and must not count as source duration. /// Consuming the encoder prevents encoding across the artificial terminal @@ -484,7 +512,9 @@ impl Encoder { } let source_frames = pcm.len() / channels; + // Any packets not yet received from earlier calls come first. let mut packets = Vec::new(); + self.receive_into(&mut packets)?; let padding = if pcm.is_empty() { 0 } else { @@ -492,7 +522,8 @@ impl Encoder { frame.extend_from_slice(pcm); frame.resize(frame_samples, 0.0); let padding = (frame_samples - pcm.len()) / channels; - packets.push(self.encode(&frame)?); + self.encode(&frame)?; + self.receive_into(&mut packets)?; padding }; @@ -506,12 +537,18 @@ impl Encoder { let lookahead = self.backend.delay(); let drain = lookahead.saturating_sub(padding); let silence = vec![0.0; frame_samples]; - for _ in 0..drain.div_ceil(self.frame_size) { - packets.push(self.encode(&silence)?); + let silent_frames = drain.div_ceil(self.frame_size); + for _ in 0..silent_frames { + self.encode(&silence)?; + self.receive_into(&mut packets)?; } + self.backend.flush()?; + self.receive_into(&mut packets)?; - let discard_padding = packets - .len() + // Counted in frames submitted here rather than packets returned, which + // include any not yet received from earlier calls. + let drained = usize::from(!pcm.is_empty()) + silent_frames; + let discard_padding = drained .saturating_mul(self.frame_size) .saturating_sub(lookahead) .saturating_sub(source_frames); @@ -522,6 +559,32 @@ impl Encoder { }) } + /// Append every packet ready now, oldest first. + fn receive_into(&mut self, packets: &mut Vec) -> Result<(), Error> { + while let Some(packet) = self.receive()? { + packets.push(packet); + } + Ok(()) + } + + /// Submit `pcm` and take every packet it readied. + #[cfg(test)] + pub(crate) fn encode_all(&mut self, pcm: &[f32]) -> Result, Error> { + self.encode(pcm)?; + let mut packets = Vec::new(); + self.receive_into(&mut packets)?; + Ok(packets) + } + + /// Flush and take every packet the codec held. + #[cfg(test)] + pub(crate) fn flush_all(&mut self) -> Result, Error> { + self.flush()?; + let mut packets = Vec::new(); + self.receive_into(&mut packets)?; + Ok(packets) + } + /// hang catalog entry describing this encoder's output stream. pub fn catalog(&self) -> hang::catalog::AudioConfig { let (rate, channels) = (self.codec_rate(), self.codec_channels()); @@ -592,12 +655,12 @@ mod tests { let frame = sine(440.0, 48_000, 2, enc.frame_size()); for _ in 0..5 { - let pkt = enc.encode(&frame).unwrap(); - let _ = dec.decode(&pkt.payload).unwrap(); + let pkt = crate::single(enc.encode_all(&frame).unwrap()); + let _ = crate::single(dec.decode_all(&pkt.payload).unwrap()); } - let pkt = enc.encode(&frame).unwrap(); - let decoded = dec.decode(&pkt.payload).unwrap(); + let pkt = crate::single(enc.encode_all(&frame).unwrap()); + let decoded = crate::single(dec.decode_all(&pkt.payload).unwrap()); assert_eq!(decoded.samples.len(), frame.len()); let energy_in: f32 = frame.iter().map(|s| s * s).sum(); @@ -631,7 +694,7 @@ mod tests { let mut enc = Encoder::new(&config).unwrap(); assert_eq!(enc.frame_size(), samples); - let packet = enc.encode(&vec![0.0f32; samples]).unwrap(); + let packet = crate::single(enc.encode_all(&vec![0.0f32; samples]).unwrap()); // SAFETY: a non-empty packet from the encoder above; the TOC is its first byte. let coded = unsafe { unsafe_libopus::opus_packet_get_samples_per_frame(packet.payload.as_ptr(), 48_000) }; assert_eq!(coded as usize, samples, "{millis} ms packet"); @@ -668,13 +731,19 @@ mod tests { let mut dec = Decoder::new(&enc.catalog(), &DecodeConfig::default()).unwrap(); let frame = vec![0.0; enc.frame_size() * enc.codec_channels() as usize]; - let first = dec.decode(&enc.encode(&frame).unwrap().payload).unwrap(); + let first = crate::single( + dec.decode_all(&crate::single(enc.encode_all(&frame).unwrap()).payload) + .unwrap(), + ); assert_eq!( first.samples.len(), (enc.frame_size() - enc.backend.delay()) * enc.codec_channels() as usize ); - let second = dec.decode(&enc.encode(&frame).unwrap().payload).unwrap(); + let second = crate::single( + dec.decode_all(&crate::single(enc.encode_all(&frame).unwrap()).payload) + .unwrap(), + ); assert_eq!(second.samples.len(), frame.len()); } @@ -697,7 +766,7 @@ mod tests { }) .unwrap(); let frame = vec![0.0; enc.frame_size()]; - enc.encode(&frame).unwrap(); + crate::single(enc.encode_all(&frame).unwrap()); // Three 120-frame packets are required to push out 312 frames. let packets = enc.finish(&[]).unwrap(); @@ -710,13 +779,13 @@ mod tests { let mut enc = Encoder::new(&Config::new(48_000, Layout::Mono)).unwrap(); let mut old = vec![0.0; enc.frame_size()]; old[enc.frame_size() - 1] = 1.0; - enc.encode(&old).unwrap(); + crate::single(enc.encode_all(&old).unwrap()); enc.reset(); let next = vec![0.0; enc.frame_size()]; - let actual = enc.encode(&next).unwrap(); + let actual = crate::single(enc.encode_all(&next).unwrap()); let mut decoder = Decoder::new(&enc.catalog(), &DecodeConfig::default()).unwrap(); - let decoded = decoder.decode(&actual.payload).unwrap(); + let decoded = crate::single(decoder.decode_all(&actual.payload).unwrap()); let peak = decoded .samples .iter() @@ -771,8 +840,8 @@ mod tests { let mut dec = Decoder::new(&enc.catalog(), &DecodeConfig::default()).unwrap(); let input = sine(440.0, enc.codec_rate(), enc.codec_channels(), enc.frame_size()); - let packet = enc.encode(&input).unwrap(); - let output = dec.decode(&packet.payload).unwrap(); + let packet = crate::single(enc.encode_all(&input).unwrap()); + let output = crate::single(dec.decode_all(&packet.payload).unwrap()); assert_eq!(output.samples, input); } @@ -797,7 +866,10 @@ mod tests { .unwrap(); let mut dec = Decoder::new(&enc.catalog(), &DecodeConfig::default()).unwrap(); let input = sine(440.0, enc.codec_rate(), enc.codec_channels(), enc.frame_size()); - let output = dec.decode(&enc.encode(&input).unwrap().payload).unwrap(); + let output = crate::single( + dec.decode_all(&crate::single(enc.encode_all(&input).unwrap()).payload) + .unwrap(), + ); assert_eq!(output.samples, input, "{layout:?}"); } } @@ -871,7 +943,11 @@ mod tests { let catalog = encoder.catalog(); let mut decoder = Decoder::new(&catalog, &DecodeConfig::default()).unwrap(); let input = [0.1, 0.2, 0.3].repeat(encoder.frame_size()); - let output = decoder.decode(&encoder.encode(&input).unwrap().payload).unwrap(); + let output = crate::single( + decoder + .decode_all(&crate::single(encoder.encode_all(&input).unwrap()).payload) + .unwrap(), + ); assert_eq!(decoder.layout(), Layout::TwoPointOne); assert_eq!(output.samples, input); @@ -1013,16 +1089,66 @@ mod tests { } } - /// The drain pushes the encoder delay out through whole silent frames. + /// The drain pushes the encoder delay out through whole silent frames, after + /// the packet the codec was still holding. #[test] fn aac_finish_drains_the_encoder_delay() { let mut enc = stub(&aac(Layout::Mono)).unwrap(); assert_eq!(enc.folded_delay(), backend::stub::DELAY); - enc.encode(&[0.0; 1024]).unwrap(); + assert!( + enc.encode_all(&[0.0; 1024]).unwrap().is_empty(), + "the stub holds its first frame" + ); // 2112 frames of delay take three 1024-frame packets. let finish = enc.finish(&[]).unwrap(); - assert_eq!(finish.packets().len(), 3); + assert_eq!(indices(finish.packets()), [0, 1, 2, 3]); assert_eq!(finish.discard_padding(), 3 * 1024 - backend::stub::DELAY); } + + /// The stub's frame index in each packet. + fn indices(packets: &[Encoded]) -> Vec { + packets + .iter() + .map(|packet| u64::from_be_bytes(packet.payload[..].try_into().unwrap())) + .collect() + } + + /// A codec that holds frames returns none, then several, and `flush` hands + /// back the rest without padding anything, leaving the encoder usable. + #[test] + fn held_packets_come_back_in_order() { + let mut enc = stub(&aac(Layout::Mono)).unwrap(); + let frame = [0.0; 1024]; + assert!(enc.encode_all(&frame).unwrap().is_empty()); + assert_eq!(indices(&enc.encode_all(&frame).unwrap()), [0, 1]); + assert!(enc.encode_all(&frame).unwrap().is_empty()); + assert_eq!(indices(&enc.flush_all().unwrap()), [2]); + assert!(enc.flush_all().unwrap().is_empty()); + + assert!(enc.encode_all(&frame).unwrap().is_empty()); + assert_eq!(indices(&enc.encode_all(&frame).unwrap()), [3, 4]); + } + + /// A held packet belongs to the epoch that submitted it, so a reset drops it. + #[test] + fn reset_drops_held_packets() { + let mut enc = stub(&aac(Layout::Mono)).unwrap(); + assert!(enc.encode_all(&[0.0; 1024]).unwrap().is_empty()); + enc.reset(); + assert!(enc.flush_all().unwrap().is_empty()); + } + + /// The padding counts the frames the drain submitted, not the held packets it + /// also returns, so a partial tail's padding is the same however much was held. + #[test] + fn finish_padding_ignores_held_packets() { + let mut enc = stub(&aac(Layout::Mono)).unwrap(); + assert!(enc.encode_all(&[0.0; 1024]).unwrap().is_empty()); + + // 100 source frames pad 924; the remaining 1188 frames of delay take two more. + let finish = enc.finish(&[0.0; 100]).unwrap(); + assert_eq!(indices(finish.packets()), [0, 1, 2, 3]); + assert_eq!(finish.discard_padding(), 3 * 1024 - backend::stub::DELAY - 100); + } } diff --git a/rs/moq-audio/src/encode/producer.rs b/rs/moq-audio/src/encode/producer.rs index f9d51c29bec..ef18a928cf3 100644 --- a/rs/moq-audio/src/encode/producer.rs +++ b/rs/moq-audio/src/encode/producer.rs @@ -1,5 +1,7 @@ //! Encode raw PCM and publish it as a moq audio track. +use std::collections::VecDeque; + use bytes::Bytes; use moq_mux::catalog::hang::CatalogExt; @@ -95,6 +97,9 @@ pub struct Producer { grouped: u64, /// Samples emitted since the current anchor (reset by [`reanchor`](Self::reanchor)). frames_produced: u64, + /// Stamps of the frames the codec has been given but has not yet returned a packet for, + /// oldest first: it returns one packet per frame in order, however late. + in_flight: VecDeque, /// Wall-clock anchor in microseconds, taken from the first frame after each /// (re)start. Emitted PTS = `anchor + frames_produced / codec_rate`. `None` /// until the first write so the next frame re-anchors to its timestamp. @@ -253,6 +258,7 @@ impl Reserved { group_samples: registered.group_samples, grouped: 0, frames_produced: 0, + in_flight: VecDeque::new(), anchor_us: None, pending_discontinuity: false, decoder_boundary: true, @@ -372,6 +378,7 @@ impl Producer { && let Ok(encoder) = self.codec.get_mut() { encoder.reset(); + self.in_flight.clear(); } // The resampler holds samples of its own, plus filter state primed by them. // Left alone, `finish` would flush that pre-reset audio onto the track @@ -486,11 +493,32 @@ impl Producer { let encoder = self.codec.get_mut()?; if reset { encoder.reset(); + self.in_flight.clear(); } - let packet = encoder.encode(samples)?; - self.activity = packet.activity; - self.publish(packet, timestamp)?; - self.decoder_boundary = false; + encoder.encode(samples)?; + self.publish_ready(Some(timestamp)) + } + + /// Publish every packet the codec has ready, each stamped with the frame it pairs with: + /// the oldest held stamp, then `submitted`, the frame just encoded, which is queued only + /// when the codec holds its packet, so a codec that holds nothing never fills the queue. + fn publish_ready(&mut self, mut submitted: Option) -> Result<(), Error> { + while let Some(packet) = self.codec.get_mut()?.receive()? { + let Some(timestamp) = self.in_flight.pop_front().or_else(|| submitted.take()) else { + return Err(Error::Codec(format!( + "{} returned more packets than it was given frames", + self.encoder()?.name() + ))); + }; + self.activity = packet.activity; + // The codec and its timestamp queue already advanced past this packet, so no later + // write could stamp what follows: end the producer with the cause. + if let Err(err) = self.publish(packet, timestamp) { + return self.close(Err(err)); + } + self.decoder_boundary = false; + } + self.in_flight.extend(submitted); Ok(()) } @@ -659,6 +687,14 @@ impl Producer { Codec::Failed(cause) => return Err(cause.clone()), } let result = self.drain(err); + self.close(result) + } + + /// Settle the codec as finished or failed with `result`, releasing encoding resources. + fn close(&mut self, result: Result<(), Error>) -> Result<(), Error> { + if !matches!(self.codec, Codec::Live(_)) { + return result; + } self.codec = match &result { Ok(()) => Codec::Finished, Err(cause) => { @@ -669,6 +705,7 @@ impl Producer { } }; self.pending = Vec::new(); + self.in_flight = VecDeque::new(); self.remix = None; self.resampler = None; result @@ -686,6 +723,7 @@ impl Producer { && let Ok(encoder) = self.codec.get_mut() { encoder.reset(); + self.in_flight.clear(); } // Whatever the resampler still holds belongs to this track: its last partial // chunk, plus the audio its filter is running behind on. Dropping it here @@ -698,6 +736,17 @@ impl Producer { // so only the final partial frame reaches the encoder's terminal drain. let anchor_us = self.anchor_us.unwrap_or(0); self.publish_full_frames(anchor_us)?; + // What the codec still holds came from full frames, so it goes out at their stamps, leaving + // the terminal drain's packets to start where the source's partial tail does. + self.codec.get_mut()?.flush()?; + self.publish_ready(None)?; + if !self.in_flight.is_empty() { + return Err(Error::Codec(format!( + "{} kept {} packet(s) past its flush", + self.encoder()?.name(), + self.in_flight.len() + ))); + } // A gap no full frame carried, such as history expiring right before capture ended, // still separates the final partial frame from the epoch before it. if std::mem::take(&mut self.pending_discontinuity) && !self.decoder_boundary && !self.pending.is_empty() { @@ -1210,6 +1259,7 @@ mod tests { producer.finish().unwrap(); assert_eq!(backend.resources(), 0, "retired {index} codecs"); assert_eq!(producer.pending.capacity(), 0); + assert_eq!(producer.in_flight.capacity(), 0); assert!(producer.remix.is_none()); assert!(producer.resampler.is_none()); assert_eq!(producer.bitrate(), bitrate); @@ -1921,6 +1971,59 @@ mod tests { assert_eq!(pts(0).await, vec![0, 0, 0, 20_000]); } + /// A codec that holds frames returns their packets late, and each still goes out at the stamp + /// of the frame it encoded, including the one only `finish` flushes out ahead of the drain. + #[tokio::test] + async fn held_packets_keep_their_frames_stamps() { + 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::new(48_000, Layout::Mono); + let options = Options { + track: Some("audio".to_string()), + config: Config::from_input(crate::encode::Codec::Aac, &input), + ..Options::default() + }; + let stub = crate::encode::backend::stub::install(); + let mut producer = Producer::new(&mut broadcast, catalog, input, &options).unwrap(); + drop(stub); + + let track = consumer + .track("audio") + .unwrap() + .subscribe( + moq_net::track::Subscription::default() + .with_delay(moq_net::track::Delay::default().with_max(Duration::from_secs(1))), + ) + .await + .unwrap(); + let mut reader = + moq_mux::container::Consumer::new(track, moq_mux::container::legacy::Wire(moq_mux::container::Kind::Audio)); + + // The stub holds the third frame, so only `finish` releases it. + producer.write(&pcm_frame(&[0.1; 3 * 1024], 1_000_000)).unwrap(); + producer.finish().unwrap(); + + let mut published = Vec::new(); + while let Some(frame) = packet(&mut reader).await { + let index = u64::from_be_bytes(frame.payload[..].try_into().unwrap()); + published.push((index, frame.timestamp.as_micros())); + } + // Each frame is stamped 2112 frames of priming early, the drain's three silent ones too. + assert_eq!( + published, + vec![ + (0, 956_000), + (1, 977_333), + (2, 998_666), + (3, 1_020_000), + (4, 1_041_333), + (5, 1_062_666), + ] + ); + } + /// The encoder needs no correction for the resampler's own delay: it anchors /// the anchor to the first input timestamp and advances by emitted samples, /// while `Resampler::process` drops its startup silence rather than passing it diff --git a/rs/moq-audio/src/lib.rs b/rs/moq-audio/src/lib.rs index 86f72851803..7c05fb5b127 100644 --- a/rs/moq-audio/src/lib.rs +++ b/rs/moq-audio/src/lib.rs @@ -52,6 +52,13 @@ #[cfg(test)] mod allocation; +/// The one output a codec that holds nothing returns for each call. +#[cfg(test)] +pub(crate) fn single(outputs: Vec) -> T { + let [output] = <[T; 1]>::try_from(outputs).expect("exactly one output"); + output +} + #[cfg(feature = "aac")] mod aac; mod activity; diff --git a/rs/moq-c/README.md b/rs/moq-c/README.md index 873d3374084..e5ec17d8835 100644 --- a/rs/moq-c/README.md +++ b/rs/moq-c/README.md @@ -195,6 +195,8 @@ int32_t moq_audio_encoder_frame_size(uint32_t encoder); int32_t moq_audio_encoder_init(uint32_t encoder, moq_audio_init *dst); int32_t moq_audio_encoder_bitrate(uint32_t encoder, uint64_t bitrate); int32_t moq_audio_encoder_encode(uint32_t encoder, const float *pcm, uintptr_t pcm_len); +int32_t moq_audio_encoder_receive(uint32_t encoder, moq_frame *dst); +int32_t moq_audio_encoder_flush(uint32_t encoder); int32_t moq_audio_encoder_finish(uint32_t encoder, const float *pcm, uintptr_t pcm_len, uint64_t *discard_padding); int32_t moq_audio_encoder_packet(uint32_t encoder, uint32_t index, moq_frame *dst); int32_t moq_audio_encoder_free(uint32_t encoder); diff --git a/rs/moq-c/src/encoder.rs b/rs/moq-c/src/encoder.rs index c1b81ec95e2..d8f43af0cf9 100644 --- a/rs/moq-c/src/encoder.rs +++ b/rs/moq-c/src/encoder.rs @@ -2,7 +2,7 @@ //! //! Each call that produces packets keeps them on the encoder handle, readable by index with //! [moq_audio_encoder_packet] or [moq_video_encoder_packet] until the next such call or the -//! handle is freed. Publish them with [crate::moq_publish_audio] or [crate::moq_publish_video] +//! handle is freed; [moq_audio_encoder_receive] keeps the one packet it writes the same way. Publish them with [crate::moq_publish_audio] or [crate::moq_publish_video] //! and [crate::moq_publish_media_frame]. use std::time::Duration; @@ -69,14 +69,13 @@ impl AudioEncoder { } fn keep(&mut self, packets: impl IntoIterator) -> Result { - self.packets = packets - .into_iter() - .map(|packet| Packet { - payload: packet.payload, - timestamp: None, - keyframe: false, - }) - .collect(); + // Reuses the buffer, since every received packet passes through it. + self.packets.clear(); + self.packets.extend(packets.into_iter().map(|packet| Packet { + payload: packet.payload, + timestamp: None, + keyframe: false, + })); count(&self.packets) } } @@ -226,8 +225,8 @@ pub extern "C" fn moq_audio_encoder_bitrate(encoder: u32, bitrate: u64) -> i32 { /// Encode one frame of interleaved f32 PCM, exactly [moq_audio_encoder_frame_size] samples /// per channel. /// -/// Returns the number of packets produced (one), read with [moq_audio_encoder_packet], or a -/// negative code on failure. +/// Returns zero, or a negative code on failure. Read the frame's packet with +/// [moq_audio_encoder_receive], which a pipelining codec readies only after later frames. /// /// # Safety /// - `pcm` must point at `pcm_len` floats. @@ -236,8 +235,46 @@ pub unsafe extern "C" fn moq_audio_encoder_encode(encoder: u32, pcm: *const f32, ffi::enter(move || { let pcm = unsafe { ffi::parse_slice(pcm, pcm_len)? }; with(encoder, audio, |audio| { - let packet = audio.encoder()?.encode(pcm)?; - audio.keep([packet]) + audio.encoder()?.encode(pcm)?; + Ok(0) + }) + }) +} + +/// Pop the oldest ready packet into `dst`; call until it returns zero after each +/// [moq_audio_encoder_encode] or [moq_audio_encoder_flush]. +/// +/// Returns one when it wrote a packet, zero while the codec holds the rest, or a negative code. +/// `payload` borrows the encoder until its next call or [moq_audio_encoder_free]; `timestamp` +/// is NULL and `keyframe` false, since the codec stamps nothing. +/// +/// # Safety +/// - `dst` must point at a writable [moq_frame]. +#[unsafe(no_mangle)] +pub unsafe extern "C" fn moq_audio_encoder_receive(encoder: u32, dst: *mut moq_frame) -> i32 { + ffi::enter(move || { + let dst = unsafe { dst.as_mut() }.ok_or(Error::InvalidPointer)?; + with(encoder, audio, |audio| { + let Some(ready) = audio.encoder()?.receive()? else { + return Ok(0); + }; + audio.keep([ready])?; + packet(&audio.packets, 0, dst)?; + Ok(1) + }) + }) +} + +/// Mark the end of input, so [moq_audio_encoder_receive] returns every packet the codec still +/// holds, without padding or closing the encoder. +/// +/// Returns zero, or a negative code. +#[unsafe(no_mangle)] +pub extern "C" fn moq_audio_encoder_flush(encoder: u32) -> i32 { + ffi::enter(move || { + with(encoder, audio, |audio| { + audio.encoder()?.flush()?; + Ok(0) }) }) } @@ -274,7 +311,7 @@ pub unsafe extern "C" fn moq_audio_encoder_finish( }) } -/// Read packet `index` from the last [moq_audio_encoder_encode] or [moq_audio_encoder_finish]. +/// Read packet `index` from the last [moq_audio_encoder_finish]. /// /// `payload` borrows the encoder until its next call or [moq_audio_encoder_free]; `timestamp` /// is NULL and `keyframe` false, since the codec stamps nothing. Returns zero, or a negative diff --git a/rs/moq-c/src/test/parity.rs b/rs/moq-c/src/test/parity.rs index 3ad0e72f33b..1fdafea9446 100644 --- a/rs/moq-c/src/test/parity.rs +++ b/rs/moq-c/src/test/parity.rs @@ -595,18 +595,26 @@ fn audio_encoder_without_a_broadcast() { // Stereo by default, so a frame is twice the frame size in interleaved samples. let pcm = vec![0.0f32; frame_size as usize * 2]; + let mut packet: moq_frame = unsafe { std::mem::zeroed() }; let mut packets = 0; for _ in 0..3 { - let count = unsafe { moq_audio_encoder_encode(encoder, pcm.as_ptr(), pcm.len()) }; - assert!(count >= 0, "encode failed: {count}"); - packets += count; + assert_eq!(unsafe { moq_audio_encoder_encode(encoder, pcm.as_ptr(), pcm.len()) }, 0); + assert_eq!(unsafe { moq_audio_encoder_receive(encoder, &mut packet) }, 1); + assert!(packet.payload_size > 0); + assert_eq!(unsafe { moq_audio_encoder_receive(encoder, &mut packet) }, 0); + packets += 1; } + assert_eq!(moq_audio_encoder_flush(encoder), 0); + assert_eq!( + unsafe { moq_audio_encoder_receive(encoder, &mut packet) }, + 0, + "libopus holds no packet" + ); let mut padding = 0; let drained = unsafe { moq_audio_encoder_finish(encoder, pcm.as_ptr(), pcm.len() / 2, &mut padding) }; assert!(drained >= 0, "finish failed: {drained}"); assert!(packets + drained > 0, "the encoder produced packets"); if drained > 0 { - let mut packet: moq_frame = unsafe { std::mem::zeroed() }; assert_eq!(unsafe { moq_audio_encoder_packet(encoder, 0, &mut packet) }, 0); assert!(packet.payload_size > 0); assert!(unsafe { moq_audio_encoder_packet(encoder, drained as u32, &mut packet) } < 0); diff --git a/rs/moq-ffi/src/audio.rs b/rs/moq-ffi/src/audio.rs index 95343feca25..2f23681c101 100644 --- a/rs/moq-ffi/src/audio.rs +++ b/rs/moq-ffi/src/audio.rs @@ -335,13 +335,28 @@ impl MoqAudioEncoder { Ok(()) } - /// Encode one codec-sized interleaved f32 PCM frame. - pub fn encode(&self, pcm: Vec) -> Result { + /// Submit one codec-sized interleaved f32 PCM frame, whose packet `receive` returns. + pub fn encode(&self, pcm: Vec) -> Result<(), MoqError> { let mut guard = self.inner.lock().unwrap(); - let packet = guard.as_mut().ok_or(MoqError::Closed)?.encode(&pcm)?; - Ok(MoqAudioPacket { + guard.as_mut().ok_or(MoqError::Closed)?.encode(&pcm)?; + Ok(()) + } + + /// Pop the oldest ready packet, or none while the codec holds the rest; call until none. + pub fn receive(&self) -> Result, MoqError> { + let mut guard = self.inner.lock().unwrap(); + let packet = guard.as_mut().ok_or(MoqError::Closed)?.receive()?; + Ok(packet.map(|packet| MoqAudioPacket { data: packet.payload.to_vec(), - }) + })) + } + + /// Mark the end of input, so `receive` returns every packet the codec still holds, + /// without padding or closing the encoder. + pub fn flush(&self) -> Result<(), MoqError> { + let mut guard = self.inner.lock().unwrap(); + guard.as_mut().ok_or(MoqError::Closed)?.flush()?; + Ok(()) } /// Drain the encoder, optionally padding one final partial frame. @@ -850,7 +865,11 @@ mod tests { }; let encoder = MoqAudioEncoder::new(output).unwrap(); let samples = encoder.frame_size().unwrap() as usize * 2; - assert!(!encoder.encode(vec![0.0; samples]).unwrap().data.is_empty()); + encoder.encode(vec![0.0; samples]).unwrap(); + assert!(!encoder.receive().unwrap().unwrap().data.is_empty()); + assert!(encoder.receive().unwrap().is_none(), "one packet per frame"); + encoder.flush().unwrap(); + assert!(encoder.receive().unwrap().is_none(), "libopus holds nothing"); encoder.finish(Vec::new()).unwrap(); assert!(matches!(encoder.frame_size(), Err(MoqError::Closed))); } @@ -889,7 +908,8 @@ mod tests { let encoder = MoqAudioEncoder::new(opus_output(None)).unwrap(); encoder.set_bitrate(32_000).unwrap(); let samples = encoder.frame_size().unwrap() as usize * 2; - assert!(!encoder.encode(vec![0.0; samples]).unwrap().data.is_empty()); + encoder.encode(vec![0.0; samples]).unwrap(); + assert!(!encoder.receive().unwrap().unwrap().data.is_empty()); encoder.finish(Vec::new()).unwrap(); assert!(matches!(encoder.set_bitrate(64_000), Err(MoqError::Closed))); assert!(matches!(encoder.track_init(), Err(MoqError::Closed))); diff --git a/rs/moq-gst/src/sink/pad.rs b/rs/moq-gst/src/sink/pad.rs index 33b4013e8d6..91f1cd9113f 100644 --- a/rs/moq-gst/src/sink/pad.rs +++ b/rs/moq-gst/src/sink/pad.rs @@ -1932,9 +1932,11 @@ mod tests { assert_eq!(decoder.layout().channels(), 6); let mut decoded_samples = false; for packet in &packets { - let decoded = decoder.decode(packet).expect("decode the published opus packet"); - assert_eq!(decoded.samples.len() % 6, 0); - decoded_samples |= !decoded.samples.is_empty(); + decoder.decode(packet).expect("decode the published opus packet"); + while let Some(decoded) = decoder.receive().expect("receive the decoded opus packet") { + assert_eq!(decoded.samples.len() % 6, 0); + decoded_samples |= !decoded.samples.is_empty(); + } } assert!(decoded_samples, "the 5.1 packets decoded to silence"); } diff --git a/swift/Sources/Moq/Audio.swift b/swift/Sources/Moq/Audio.swift index 58f8a994061..b5dbeeb213d 100644 --- a/swift/Sources/Moq/Audio.swift +++ b/swift/Sources/Moq/Audio.swift @@ -138,8 +138,14 @@ public enum Audio { /// Retune the live encoder in bits per second. public func setBitrate(_ bitrate: UInt64) throws { try ffi.setBitrate(bitrate: bitrate) } - /// Encode one codec-sized interleaved f32 frame. - public func encode(_ pcm: [Float]) throws -> Packet { try ffi.encode(pcm: pcm) } + /// Submit one codec-sized interleaved f32 frame, whose packet `receive` returns. + public func encode(_ pcm: [Float]) throws { try ffi.encode(pcm: pcm) } + + /// Pop the oldest ready packet, or nil while the codec holds the rest; call until nil. + public func receive() throws -> Packet? { try ffi.receive() } + + /// Mark the end of input, so `receive` returns every packet the codec still holds, without closing the encoder. + public func flush() throws { try ffi.flush() } /// Pad an optional partial frame, drain the codec, and close the encoder. public func finish(_ pcm: [Float] = []) throws -> EncoderFinish { try ffi.finish(pcm: pcm) }