From b1e850b3545205e84fbcaa70a138079aad801372 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Fri, 10 Jul 2026 13:14:49 -0700 Subject: [PATCH 1/3] feat(moq-transcode): share one decode across rungs; GPU resize; moq transcode verb Decoding used to be per rung, so N active rungs decoded the same source N times and NVDEC throughput scaled with ladder depth instead of source count. - moq-video grows decode::Frame::resize(w, h): a scaled copy that keeps a CUDA frame on the GPU (a box-filter kernel, vendored as PTX and JIT-compiled by the driver, so builds still need no CUDA toolkit) and resizes CPU frames with a SIMD bilinear convolution. Kernel output is validated against the CPU reference on hardware. - moq-transcode gains a shared live feed (feed.rs): one source subscription and one decoder per broadcast, fanned out to every rung with live demand over a broadcast channel of Arc'd frames. Rungs resize and encode their own copy; a lagging rung skips to the next group instead of stalling the others. Fetches keep their one-shot pipeline (decoder-side scaling), and the CPU scale.rs is gone, subsumed by Frame::resize. - moq-cli gains a feature-gated `transcode` verb wrapping moq-transcode: `moq --client-connect --broadcast transcode [--rung h:bps ...]`. Also fixes a startup race (in the example too): request_broadcast is unroutable until the first session connects, so wait for Connected. Validated on an RTX 3070 Ti: unit suites (multi-rung hardware test rides one NVDEC session into two NVENC sessions) plus a live end-to-end against a local relay (ffmpeg -> moq import -> moq transcode -> two concurrent rung exports, outputs verified with ffprobe; the second rung attached to the already-running shared decode without opening a new decoder). Co-Authored-By: Claude Fable 5 --- Cargo.lock | 3 +- Cargo.toml | 2 + doc/bin/cli.md | 29 ++ rs/CLAUDE.md | 4 +- rs/moq-cli/Cargo.toml | 11 +- rs/moq-cli/src/args.rs | 4 + rs/moq-cli/src/main.rs | 4 + rs/moq-cli/src/transcode.rs | 125 ++++++++ rs/moq-transcode/CHANGELOG.md | 10 + rs/moq-transcode/Cargo.toml | 6 +- rs/moq-transcode/README.md | 7 +- rs/moq-transcode/examples/transcode.rs | 6 +- rs/moq-transcode/src/feed.rs | 195 +++++++++++++ rs/moq-transcode/src/lib.rs | 184 +++++++++++- rs/moq-transcode/src/rung.rs | 195 ++++++++----- rs/moq-transcode/src/scale.rs | 137 --------- rs/moq-video/CHANGELOG.md | 7 + rs/moq-video/Cargo.toml | 8 +- rs/moq-video/src/decode/mod.rs | 42 +++ rs/moq-video/src/frame.rs | 245 +++++++++++++++- rs/moq-video/src/frame/nv12_resize.cu | 75 +++++ rs/moq-video/src/frame/nv12_resize.ptx | 384 +++++++++++++++++++++++++ 22 files changed, 1447 insertions(+), 236 deletions(-) create mode 100644 rs/moq-cli/src/transcode.rs create mode 100644 rs/moq-transcode/src/feed.rs delete mode 100644 rs/moq-transcode/src/scale.rs create mode 100644 rs/moq-video/src/frame/nv12_resize.cu create mode 100644 rs/moq-video/src/frame/nv12_resize.ptx diff --git a/Cargo.lock b/Cargo.lock index 1147f985d..6fa06f346 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -4138,6 +4138,7 @@ dependencies = [ "moq-rtc", "moq-rtmp", "moq-srt", + "moq-transcode", "moq-video", "rustls", "sd-notify", @@ -4490,7 +4491,6 @@ dependencies = [ "anyhow", "bytes", "clap", - "fast_image_resize", "hang", "moq-mux", "moq-native", @@ -4526,6 +4526,7 @@ dependencies = [ "bytes", "cudarc", "dispatch2", + "fast_image_resize", "hang", "libloading 0.8.9", "moq-mux", diff --git a/Cargo.toml b/Cargo.toml index 0b77d7fe3..156524c74 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -88,6 +88,8 @@ moq-vaapi = "0.0.2" # default-features off (the `nvenc` / `vaapi` hardware encoders) so each consumer # opts in: binaries (libmoq, moq-boy, moq-cli) enable them, but a self-compiler can # leave them off to drop the CUDA / libva deps. moq-video itself still defaults them on. +# Hardware codecs are also opt-in here, mirroring moq-video (see above). +moq-transcode = { version = "0.0.1", path = "rs/moq-transcode", default-features = false } moq-video = { version = "0.0.6", path = "rs/moq-video", default-features = false } qmux = { version = "0.3", default-features = false } serde = { version = "1", features = ["derive"] } diff --git a/doc/bin/cli.md b/doc/bin/cli.md index 247f10c70..3747d81fb 100644 --- a/doc/bin/cli.md +++ b/doc/bin/cli.md @@ -160,6 +160,35 @@ ffmpeg -f v4l2 -i /dev/video0 -f mpegts - | \ moq --client-connect https://relay.example.com/anon --broadcast webcam.hang import ts ``` +### Transcode a Broadcast + +The `transcode` verb consumes a broadcast from the relay and publishes a +just-in-time transcoded ladder next to it. The derivative catalog references the +source renditions directly and adds the lower rungs, which are only decoded and +encoded while someone actually watches (or fetches) them. It is gated behind the +`transcode` feature: + +```bash +cargo build --release -p moq-cli --features transcode + +# Publish `cam.hang/transcode.hang` with the default ladder (1080p..240p, +# filtered to rungs strictly below the source): +moq --client-connect https://relay.example.com/anon --broadcast cam.hang transcode + +# Pick the ladder (height:bitrate in bits per second) and pin the codecs: +moq --client-connect https://relay.example.com/anon --broadcast cam.hang transcode --rung 720:2500000 --rung 360:600000 --encoder nvenc --decoder nvdec + +# Publish the derivative somewhere else (the catalog then omits the relative +# source references): +moq --client-connect https://relay.example.com/anon --broadcast cam.hang transcode --output ladder.hang +``` + +On an NVIDIA GPU the pipeline is fully GPU-resident: one shared NVDEC session +decodes the source for all active rungs, a CUDA kernel resizes each rung's copy, +and NVENC encodes it in place, with no CPU copies. Without a GPU it falls back +to openh264 and CPU scaling. Like `capture`, the hardware backends are on by +default; drop the default features for a software-only build. + ### Play a Broadcast Pull a broadcast back out and play it: diff --git a/rs/CLAUDE.md b/rs/CLAUDE.md index 3cd64368e..50c2d38f9 100644 --- a/rs/CLAUDE.md +++ b/rs/CLAUDE.md @@ -27,12 +27,12 @@ Layered roughly transport -> container/format -> media -> apps/bindings. - `moq-mux` (lib): the conversion layer. File/stream formats (`container/`: fmp4, flv, mkv, ts, loc) and codec parsers (`codec/`: h264, h265, av1, vp8/9, opus, aac, ...) <-> hang broadcasts. `Container` trait + generic `Producer`/`Consumer`. Dual catalog (`catalog::hang`, `catalog::msf`). - `moq-audio` (lib): native PCM <-> Opus (`unsafe-libopus`). `AudioProducer`/`AudioConsumer`, `Encoder`/`Decoder`, `AudioFormat`. Optional `capture` feature (cpal microphone), `resample`. - `moq-video` (lib): native video capture, H.264/H.265 encode, and decode; no ffmpeg. Hardware backends (VideoToolbox / Media Foundation / NVENC / VAAPI / NVDEC) with openh264 as the software H.264 fallback; NVDEC frames stay in CUDA memory and feed NVENC zero-copy. `capture::Config`, `encode::{Encoder, Producer, publish_capture}`, `decode::{Consumer, Decoder}`. -- `moq-transcode` (lib): just-in-time live transcoding of hang broadcasts. `run(source, output, config)` publishes a derivative catalog (ladder rungs + relative refs to the source) and encodes each rung only while subscribed/fetched, via `moq-video`. Output groups mirror source group sequences 1:1. +- `moq-transcode` (lib): just-in-time live transcoding of hang broadcasts. `run(source, output, config)` publishes a derivative catalog (ladder rungs + relative refs to the source) and encodes each rung only while subscribed/fetched, via `moq-video`. Live rungs share one decode per source (the `feed` module); output groups mirror source group sequences 1:1. Also a moq-cli verb (`moq ... transcode`, feature-gated). **Apps / binaries** - `moq-relay` (lib+bin): clusterable, media-agnostic relay. axum HTTP API, JWT auth, WebSocket fallback, clustering. Config/TOML merge pattern lives here (see below). -- `moq-cli` (bin, `moq`): the unified media router (`moq `); stdin/stdout media piping. The CLI surface for the gateway library crates below lives here. +- `moq-cli` (bin, `moq`): the unified media router (`moq `, plus the feature-gated `transcode` verb); stdin/stdout media piping. The CLI surface for the gateway library crates below lives here. - `moq-rtc` (lib): WebRTC (WHIP/WHEP) gateway. Bridges browser WebRTC ingest/playback to MoQ broadcasts (str0m ICE/DTLS, A/V sync, NACK). Embeddable axum routers / `Client`; the CLI surface lives in `moq-cli`. - `moq-rtmp` (lib): RTMP / enhanced-RTMP gateway (ingest + egress, `rml_rtmp`, FLV via `moq-mux`). RTMPS (rustls + tokio-rustls) is the optional `tls` feature. - `moq-srt` (lib): bidirectional SRT gateway (MPEG-TS via `srt-tokio` + `moq-mux`). diff --git a/rs/moq-cli/Cargo.toml b/rs/moq-cli/Cargo.toml index 365a78a30..632dabe6e 100644 --- a/rs/moq-cli/Cargo.toml +++ b/rs/moq-cli/Cargo.toml @@ -31,13 +31,17 @@ websocket = ["moq-native/websocket"] # extra install. H.264 is vendored static (openh264), so no system ffmpeg/libav. # Enable with `cargo build -p moq-cli --features capture`. capture = ["dep:moq-video", "dep:moq-audio", "moq-audio/capture"] +# Just-in-time transcoding (`moq ... transcode`). Off by default for the same +# reason as `capture`: it pulls in moq-video, whose default-on `vaapi` feature +# links libva on Linux. Enable with `cargo build -p moq-cli --features transcode`. +transcode = ["dep:moq-transcode", "dep:moq-video"] # NVENC / VAAPI / NVDEC hardware codecs for `capture` (Linux), on by default. Each # just turns on the matching moq-video feature, and only bites when `capture` pulls # in moq-video. For a `capture` build with no CUDA / libva system deps, drop them, e.g. # cargo build -p moq-cli --no-default-features --features "iroh noq websocket capture" -nvenc = ["moq-video?/nvenc"] -vaapi = ["moq-video?/vaapi"] -nvdec = ["moq-video?/nvdec"] +nvenc = ["moq-video?/nvenc", "moq-transcode?/nvenc"] +vaapi = ["moq-video?/vaapi", "moq-transcode?/vaapi"] +nvdec = ["moq-video?/nvdec", "moq-transcode?/nvdec"] [dependencies] anyhow = { version = "1", features = ["backtrace"] } @@ -55,6 +59,7 @@ moq-native = { workspace = true, default-features = false, features = ["aws-lc-r moq-rtc = { workspace = true } moq-rtmp = { workspace = true } moq-srt = { workspace = true } +moq-transcode = { workspace = true, optional = true } moq-video = { workspace = true, optional = true } rustls = { version = "0.23", features = ["aws-lc-rs"], default-features = false } tokio = { workspace = true, features = ["full"] } diff --git a/rs/moq-cli/src/args.rs b/rs/moq-cli/src/args.rs index ddcef0ff4..bc4a6e921 100644 --- a/rs/moq-cli/src/args.rs +++ b/rs/moq-cli/src/args.rs @@ -72,6 +72,10 @@ pub enum Direction { /// Route media OUT OF MoQ to one sink. #[command(alias = "subscribe")] Export(Export), + /// Re-encode `--broadcast` into a lower ladder, published next to it and + /// only encoded while watched (just-in-time). + #[cfg(feature = "transcode")] + Transcode(crate::transcode::Args), } // ------------------------------------------------------------------ import diff --git a/rs/moq-cli/src/main.rs b/rs/moq-cli/src/main.rs index 61eeb450c..bf2d4b5b7 100644 --- a/rs/moq-cli/src/main.rs +++ b/rs/moq-cli/src/main.rs @@ -12,6 +12,8 @@ mod rtc; mod rtmp; mod srt; mod subscribe; +#[cfg(feature = "transcode")] +mod transcode; mod web; use args::{Cli, Direction, Export, ExportSink, Import, ImportSource, MoqSide}; @@ -66,6 +68,8 @@ async fn main() -> anyhow::Result<()> { match cli.direction { Direction::Import(import) => run_import(cli.moq, import, net).await, Direction::Export(export) => run_export(cli.moq, export, net).await, + #[cfg(feature = "transcode")] + Direction::Transcode(args) => transcode::run(cli.moq, args, net).await, } } diff --git a/rs/moq-cli/src/transcode.rs b/rs/moq-cli/src/transcode.rs new file mode 100644 index 000000000..755ea450d --- /dev/null +++ b/rs/moq-cli/src/transcode.rs @@ -0,0 +1,125 @@ +//! The `transcode` verb: consume a source broadcast and publish a just-in-time +//! transcoded ladder next to it. +//! +//! The derivative appears at `/transcode.hang` (or `--output`): its +//! catalog references the source renditions directly and adds the lower rungs, +//! which are only decoded and encoded while someone watches (or fetches) them. +//! On an NVIDIA GPU the whole pipeline is GPU-resident (NVDEC -> CUDA resize -> +//! NVENC); otherwise it falls back to software codecs. + +use anyhow::Context; + +use crate::Net; +use crate::args::MoqSide; +use hang::moq_net; + +/// Ladder and codec options for the `transcode` verb. +#[derive(clap::Args, Clone)] +pub struct Args { + /// The derivative broadcast path. Defaults to `/transcode.hang`. + #[arg(long)] + pub output: Option, + + /// A ladder rung as `height:bitrate` (pixels : bits per second), repeatable, + /// e.g. `--rung 720:2500000 --rung 360:600000`. Rungs at or above the source + /// are dropped at runtime. Defaults to a 1080p..240p ladder. + #[arg(long = "rung", value_parser = parse_rung)] + pub rungs: Vec, + + /// The video encoder: `auto` (hardware first), `hardware`, `software`, or a + /// backend name like `nvenc`. + #[arg(long, default_value = "auto")] + pub encoder: String, + + /// The video decoder: `auto` (hardware first), `hardware`, `software`, or a + /// backend name like `nvdec`. + #[arg(long, default_value = "auto")] + pub decoder: String, +} + +/// Parse a `height:bitrate` rung, e.g. `720:2500000`. +fn parse_rung(arg: &str) -> Result { + let (height, bitrate) = arg + .split_once(':') + .ok_or_else(|| format!("expected height:bitrate, got `{arg}`"))?; + let height: u32 = height.parse().map_err(|e| format!("invalid height `{height}`: {e}"))?; + let bitrate: u64 = bitrate + .parse() + .map_err(|e| format!("invalid bitrate `{bitrate}`: {e}"))?; + Ok(moq_transcode::Rung::new(height, bitrate)) +} + +/// Run the transcoder: subscribe to the source through the relay, publish the +/// derivative back through the same session, and serve rungs until either ends. +pub async fn run(moq: MoqSide, args: Args, net: Net) -> anyhow::Result<()> { + let source_path = moq + .broadcast + .clone() + .filter(|name| !name.is_empty()) + .context("`transcode` requires the source broadcast: pass --broadcast ")?; + let output_path = args + .output + .clone() + .unwrap_or_else(|| format!("{source_path}/transcode.hang")); + + // Publish the derivative through one origin and consume the source through + // another, over a single auto-reconnecting session. + let url = moq + .client + .connect + .clone() + .context("`transcode` requires a relay: pass --client-connect ")?; + let publish = moq_net::Origin::random().produce(); + let remote = moq_net::Origin::random().produce(); + let mut session = net + .client(moq.client.clone())? + .with_publisher(&publish) + .with_subscriber(remote.clone()) + .reconnect(url); + + // Wait for the first session: the origin can't route a broadcast request + // until a connected session registers its handler. + while !matches!(session.status().await?, moq_native::Status::Connected) {} + + // Request the source broadcast; the session subscribes upstream on demand. + let source = remote + .consume() + .request_broadcast(&source_path) + .await + .context("source broadcast unavailable")?; + + let mut config = moq_transcode::Config::default(); + if !args.rungs.is_empty() { + config.rungs = args.rungs.clone(); + } + config.encoder = match args.encoder.as_str() { + "auto" => moq_video::encode::Kind::Auto, + "hardware" => moq_video::encode::Kind::Hardware, + "software" => moq_video::encode::Kind::Software, + name => moq_video::encode::Kind::Named(name.to_string()), + }; + config.decoder = match args.decoder.as_str() { + "auto" => moq_video::decode::Kind::Auto, + "hardware" => moq_video::decode::Kind::Hardware, + "software" => moq_video::decode::Kind::Software, + name => moq_video::decode::Kind::Named(name.to_string()), + }; + // Reference the source renditions relatively when the output nests under + // the source (`a/b` -> `a/b/transcode.hang` is `..`, one `..` per level); + // otherwise the derivative catalog advertises only the rungs. + config.source = output_path.strip_prefix(&format!("{source_path}/")).map(|rest| { + let depth = rest.split('/').count(); + moq_net::PathRelativeOwned::from(vec![".."; depth].join("/")) + }); + + let output = moq_net::broadcast::Info::default().produce(); + let _announce = publish + .publish_broadcast(&output_path, &output) + .context("failed to publish the derivative broadcast")?; + tracing::info!(source = %source_path, output = %output_path, "transcoding"); + + tokio::select! { + res = moq_transcode::run(source, output, config) => Ok(res?), + res = session.closed() => Ok(res?), + } +} diff --git a/rs/moq-transcode/CHANGELOG.md b/rs/moq-transcode/CHANGELOG.md index fe793f44c..f487e24e5 100644 --- a/rs/moq-transcode/CHANGELOG.md +++ b/rs/moq-transcode/CHANGELOG.md @@ -9,6 +9,16 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Added +- Shared live decode: all rungs of a source with live demand now share one + subscription and one decoder (a broadcast feed of decoded frames), instead of + each rung decoding the source independently. NVDEC throughput and upstream + bandwidth now scale with source count, not ladder depth; each rung resizes + its copy on the GPU (`decode::Frame::resize`) and encodes it in place. Group + fetches keep their own one-shot pipeline. +- `moq transcode`: the transcoder is now a moq-cli verb (behind the `transcode` + feature), publishing `/transcode.hang` with a configurable ladder + (`--rung height:bitrate`) and codec pins (`--encoder`, `--decoder`). + - Initial release: just-in-time live transcoding of hang broadcasts. `run(source, output, config)` publishes a derivative catalog (ladder rungs strictly below the source, plus relative references to the source renditions) diff --git a/rs/moq-transcode/Cargo.toml b/rs/moq-transcode/Cargo.toml index 92afd9773..433a6d778 100644 --- a/rs/moq-transcode/Cargo.toml +++ b/rs/moq-transcode/Cargo.toml @@ -30,9 +30,6 @@ nvdec = ["moq-video/nvdec"] [dependencies] bytes = "1" -# CPU I420 scaling between decode and re-encode (SIMD per-plane resize), for -# decoders without a built-in scaler; NVDEC scales in hardware instead. -fast_image_resize = "5" hang = { workspace = true } moq-mux = { workspace = true } moq-net = { workspace = true } @@ -46,5 +43,6 @@ anyhow = "1" clap = { version = "4", features = ["derive"] } # Default features on (unlike the workspace entry) so the example gets a QUIC backend. moq-native = { path = "../moq-native" } -tokio = { workspace = true, features = ["full"] } +# test-util unlocks tokio::time::pause for the deterministic live-feed tests. +tokio = { workspace = true, features = ["full", "test-util"] } url = "2" diff --git a/rs/moq-transcode/README.md b/rs/moq-transcode/README.md index c4f3964bd..c00fb3db8 100644 --- a/rs/moq-transcode/README.md +++ b/rs/moq-transcode/README.md @@ -9,9 +9,10 @@ combined ladder and the transcoder never proxies what it doesn't re-encode. Nothing is encoded until someone asks, Cloudflare-style just-in-time per rung: -- **Subscribe** to a rung and the transcoder subscribes to the source track, - decoding, scaling, and re-encoding group for group until the last subscriber - leaves. +- **Subscribe** to a rung and the transcoder decodes the source live, resizing + and re-encoding group for group until the last subscriber leaves. Every + active rung of a source shares one subscription and one decoder, so decode + cost scales with sources, not ladder depth. - **Fetch** a specific group and the transcoder fetches that same group from the source and transcodes just that group. Output groups mirror source group sequence numbers 1:1, so group N of every rung is the same content as source diff --git a/rs/moq-transcode/examples/transcode.rs b/rs/moq-transcode/examples/transcode.rs index 027d0e6ef..1629e2f95 100644 --- a/rs/moq-transcode/examples/transcode.rs +++ b/rs/moq-transcode/examples/transcode.rs @@ -43,11 +43,15 @@ async fn main() -> anyhow::Result<()> { let remote = moq_net::Origin::random().produce(); let client = moq_native::ClientConfig::default().init()?; - let session = client + let mut session = client .with_publisher(&publish) .with_subscriber(remote.clone()) .reconnect(args.url.clone()); + // Wait for the first session: the origin can't route a broadcast request + // until a connected session registers its handler. + while !matches!(session.status().await?, moq_native::Status::Connected) {} + // Request the source broadcast; the session subscribes upstream on demand. let source = remote .consume() diff --git a/rs/moq-transcode/src/feed.rs b/rs/moq-transcode/src/feed.rs new file mode 100644 index 000000000..20c153373 --- /dev/null +++ b/rs/moq-transcode/src/feed.rs @@ -0,0 +1,195 @@ +//! The shared live decode: one source subscription and one decoder per +//! broadcast, fanned out to every rung with live demand. +//! +//! Rungs used to subscribe and decode independently, so N active rungs decoded +//! the same source N times. NVDEC throughput (and upstream bandwidth) then +//! scaled with ladder depth rather than source count. A [`Feed`] decodes each +//! group once and broadcasts the frames; each rung resizes and encodes its own +//! copy (cheap: GPU frames are refcounted). +//! +//! The decode loop runs only while at least one [`Listener`] exists: the first +//! listener subscribes to the source and opens the decoder, dropping the last +//! one tears both down. An idle broadcast costs nothing, exactly like the +//! per-rung subscriptions this replaces. + +use std::sync::{Arc, Mutex}; + +use hang::catalog::VideoConfig; +use moq_mux::container::Container as _; +use tokio::sync::broadcast; + +use crate::Error; + +/// Frames buffered per listener. A rung that falls further behind than this +/// lags out (skipping to the next group) instead of stalling the other rungs; +/// the decode task itself never blocks on a slow listener. +const CAPACITY: usize = 16; + +/// One item of the shared live feed, in stream order. +#[derive(Clone)] +pub(crate) enum Item { + /// A new source group started; the frames that follow belong to it. + Group(u64), + /// A decoded frame of the current group. Cloning is cheap (`Arc`), and a + /// GPU frame stays on the GPU for every receiving rung. + Frame(Arc), + /// The current group ended cleanly. + End, + /// The source track ended cleanly; no more items follow. + Finished, + /// Never broadcast: synthesized by [`Listener::recv`] when this listener + /// fell behind and the queue rolled over. Abandon the current group and + /// pick up at the next [`Item::Group`]. + Lagged, +} + +/// Handle to the shared live decode of one source track. Cloning shares the +/// same underlying feed. +#[derive(Clone)] +pub(crate) struct Feed { + inner: Arc, +} + +struct Inner { + /// The source media track (subscribed only while listeners exist). + source: moq_net::track::Consumer, + /// The source rendition's catalog entry (codec + container). + config: VideoConfig, + /// Which decoder implementation to use. + decoder: moq_video::decode::Kind, + state: Mutex, +} + +#[derive(Default)] +struct State { + listeners: usize, + /// The running decode session's channel; `None` while idle. + sender: Option>, + task: Option>, +} + +impl Feed { + pub(crate) fn new(source: moq_net::track::Consumer, config: VideoConfig, decoder: moq_video::decode::Kind) -> Self { + Self { + inner: Arc::new(Inner { + source, + config, + decoder, + state: Mutex::new(State::default()), + }), + } + } + + /// Attach to the live feed, starting the shared decode session if idle. + /// A listener attached mid-group sees frames without their [`Item::Group`]; + /// wait for the next boundary before producing. + pub(crate) fn listen(&self) -> Listener { + let mut state = self.inner.state.lock().unwrap(); + state.listeners += 1; + if state.sender.is_none() { + let (sender, _) = broadcast::channel(CAPACITY); + state.task = Some(tokio::spawn(run(self.inner.clone(), sender.clone()))); + state.sender = Some(sender); + } + let receiver = state.sender.as_ref().expect("sender ensured above").subscribe(); + Listener { + feed: self.clone(), + receiver, + } + } +} + +/// One rung's attachment to the shared feed. Dropping it detaches; dropping +/// the last one stops the shared subscription and decoder. +pub(crate) struct Listener { + feed: Feed, + receiver: broadcast::Receiver, +} + +impl Listener { + /// The next feed item, or `None` when the feed died mid-stream (a source or + /// decode error; a clean source end arrives as [`Item::Finished`] first). + pub(crate) async fn recv(&mut self) -> Option { + match self.receiver.recv().await { + Ok(item) => Some(item), + Err(broadcast::error::RecvError::Lagged(_)) => Some(Item::Lagged), + Err(broadcast::error::RecvError::Closed) => None, + } + } +} + +impl Drop for Listener { + fn drop(&mut self) { + let mut state = self.feed.inner.state.lock().unwrap(); + state.listeners -= 1; + if state.listeners == 0 { + // Last listener out: stop decoding and release the subscription. + state.sender = None; + if let Some(task) = state.task.take() { + task.abort(); + } + } + } +} + +/// One decode session: runs until the source ends or errors, then clears +/// itself from the feed state so the next listener starts a fresh session. +async fn run(inner: Arc, sender: broadcast::Sender) { + match decode(&inner, &sender).await { + // Dropping the sender after Finished closes every listener cleanly. + Ok(()) => { + let _ = sender.send(Item::Finished); + } + // Dropping the sender without Finished reads as an error downstream. + Err(err) => tracing::warn!(%err, "shared decode session failed"), + } + + let mut state = inner.state.lock().unwrap(); + // Only clear if this session is still the current one: a full teardown and + // restart may have raced this exit. + if state.sender.as_ref().is_some_and(|s| s.same_channel(&sender)) { + state.sender = None; + state.task = None; + } +} + +/// Subscribe to the source and decode group for group into the channel. +/// Decodes at the stream's native size: the rungs share these frames, so +/// per-rung sizing happens on their side (`Frame::resize`). +async fn decode(inner: &Inner, sender: &broadcast::Sender) -> Result<(), Error> { + let container = moq_mux::catalog::hang::Container::try_from(&inner.config.container)?; + + let mut config = moq_video::decode::Config::new(); + config.kind = inner.decoder.clone(); + let mut decoder = moq_video::decode::Decoder::new(&inner.config, &config)?; + + // The feed serves whichever rungs are active, so there is no single + // downstream subscription to mirror; live-edge defaults fit every rung. + let mut subscriber = inner.source.subscribe(None).await?; + + while let Some(mut group) = subscriber.next_group().await? { + // Sends only fail with zero receivers, which is fine: teardown aborts + // this task at the next await anyway. + let _ = sender.send(Item::Group(group.sequence)); + + let mut first = true; + while let Some(frames) = container.read(&mut group).await? { + for frame in frames { + let timestamp: u64 = frame + .timestamp + .as_micros() + .try_into() + .map_err(|_| moq_net::TimeOverflow)?; + // The first frame of a group is a keyframe by construction. + let keyframe = frame.keyframe || first; + first = false; + + for decoded in decoder.decode(&frame.payload, timestamp, keyframe)? { + let _ = sender.send(Item::Frame(Arc::new(decoded))); + } + } + } + let _ = sender.send(Item::End); + } + Ok(()) +} diff --git a/rs/moq-transcode/src/lib.rs b/rs/moq-transcode/src/lib.rs index 21897af6b..873a76a51 100644 --- a/rs/moq-transcode/src/lib.rs +++ b/rs/moq-transcode/src/lib.rs @@ -7,8 +7,10 @@ //! strings are computed from the ladder, not the bitstream), but nothing is //! encoded until a subscriber actually asks: //! -//! - Subscribing to a rung subscribes to the source track and transcodes live, -//! group for group, stopping when the last subscriber leaves. +//! - Subscribing to a rung attaches it to a shared live decode of the source +//! (one subscription and one decoder per source, no matter how many rungs +//! are active); each rung resizes and encodes its own copy, group for group, +//! stopping when the last subscriber leaves. //! - Fetching a specific group fetches that same group from the source and //! transcodes just that group. Output groups mirror source sequence numbers //! 1:1, so group N of every rung is the same content as source group N. @@ -22,8 +24,8 @@ mod catalog; mod config; mod error; +mod feed; mod rung; -mod scale; pub use config::{Config, Rung}; pub use error::Error; @@ -69,6 +71,14 @@ pub async fn run( let rungs = catalog::resolve_rungs(&config.rungs, &source_name, &source_config)?; tracing::info!(source = %source_name, rungs = rungs.len(), "transcoding"); + // One shared live decode for every rung of this source: N active rungs + // share one subscription and one decoder instead of N. + let feed = feed::Feed::new( + source.track(&source_name)?, + source_config.clone(), + config.decoder.clone(), + ); + // Publish the derivative catalog before any encoder exists, so subscribers // can pick a rung immediately. let entries: Vec<_> = rungs @@ -94,6 +104,7 @@ pub async fn run( Some(info) => { let rung = rung::Rung { source: source.track(&source_name)?, + feed: feed.clone(), broadcast: source.clone(), config: source_config.clone(), encoder: config.encoder.clone(), @@ -201,6 +212,173 @@ mod tests { } } + /// A source like [`source_broadcast`], but the groups arrive over (paused) + /// time instead of all at once, so several rungs can attach to the shared + /// live feed before the first group exists. Returns the broadcast plus the + /// producing task's handle (the track producer lives inside it). + fn source_broadcast_live(groups: u64, frames: u64) -> (Source, tokio::task::JoinHandle<()>) { + let mut broadcast = moq_net::broadcast::Info::default().produce(); + let mut catalog = moq_mux::catalog::Producer::new(&mut broadcast).unwrap(); + + let mut video = hang::catalog::VideoConfig::new(hang::catalog::H264 { + inline: true, + profile: 0x42, + constraints: 0, + level: 30, + }); + video.coded_width = Some(320); + video.coded_height = Some(240); + video.bitrate = Some(1_000_000); + video.framerate = Some(30.0); + catalog.lock().video.insert("video", video).unwrap(); + + let info = moq_net::track::Info::default().with_timescale(hang::container::TIMESCALE); + let mut track = broadcast.create_track("video", info).unwrap(); + + let source = Source { + broadcast, + _catalog: catalog, + // The producing task owns the real track producer; park a clone so + // the struct shape matches `source_broadcast`. + _track: track.clone(), + }; + + let task = tokio::spawn(async move { + let mut encoder = moq_video::encode::Encoder::new(&{ + let mut config = moq_video::encode::Config::new(320, 240, 30); + config.kind = moq_video::encode::Kind::Software; + config + }) + .unwrap(); + let gray = vec![0x80u8; 320 * 240 * 4]; + + for sequence in 0..groups { + // Under `tokio::time::pause` this advances once every task is + // idle, i.e. once all rungs are attached and waiting. + tokio::time::sleep(std::time::Duration::from_millis(100)).await; + let mut group = track.create_group(sequence.into()).unwrap(); + for index in 0..frames { + let timestamp = (sequence * frames + index) * 33_333; + for payload in encoder.encode_rgba(&gray, 320, 240, index == 0).unwrap() { + let frame = hang::container::Frame { + timestamp: moq_net::Timestamp::from_micros(timestamp).unwrap(), + payload, + }; + frame.encode(&mut group).unwrap(); + } + } + group.finish().unwrap(); + } + // Keep the track open until aborted, like a live source. + std::future::pending::<()>().await; + }); + + (source, task) + } + + /// Two rungs subscribed at once ride one shared live decode (the feed): + /// both must produce complete groups mirroring the source sequences. + #[tokio::test] + async fn live_multi_rung() { + tokio::time::pause(); + + let (source, producer_task) = source_broadcast_live(3, 5); + let config = Config { + rungs: vec![Rung::new(120, 100_000), Rung::new(60, 50_000)], + encoder: moq_video::encode::Kind::Software, + decoder: moq_video::decode::Kind::Software, + source: None, + }; + + let output = moq_net::broadcast::Info::default().produce(); + let consumer = output.consume(); + let transcoder = tokio::spawn(run(source.broadcast.consume(), output, config)); + + // Attach both rungs before the first source group exists (paused time: + // the producer's sleep only fires once every rung is parked on the feed). + let mut subscribers = Vec::new(); + for name in ["video/120p", "video/60p"] { + let track = loop { + match consumer.track(name) { + Ok(track) => break track, + Err(moq_net::Error::NotFound) => tokio::task::yield_now().await, + Err(err) => panic!("rung track {name}: {err}"), + } + }; + subscribers.push((name, track.subscribe(None).await.unwrap())); + } + + // Every rung receives a complete group with all 5 source frames. + for (name, subscriber) in &mut subscribers { + let mut group = subscriber.next_group().await.unwrap().unwrap(); + let payload = group.read_frame().await.unwrap().unwrap(); + let frame = hang::container::Frame::decode(payload).unwrap(); + assert!( + frame.payload.starts_with(&[0, 0, 0, 1]) || frame.payload.starts_with(&[0, 0, 1]), + "{name} output is not Annex-B" + ); + let total = group.finished().await.unwrap(); + assert_eq!(total, 5, "{name} dropped frames"); + } + + producer_task.abort(); + transcoder.abort(); + } + + /// The multi-rung live path on real hardware: one shared NVDEC session + /// decodes the source, the GPU box filter resizes per rung, and each rung's + /// NVENC session encodes the CUDA frame in place. Skips without a GPU. + #[tokio::test] + async fn live_multi_rung_hardware() { + if !hardware_available() { + eprintln!("skipping: no hardware decoder + encoder available"); + return; + } + tokio::time::pause(); + + let (source, producer_task) = source_broadcast_live(3, 5); + // 180p and 120p: NVENC rejects tiny frames (80x60 is below its minimum + // encode resolution), so the hardware ladder stays a bit larger than the + // software test's. + let config = Config { + rungs: vec![Rung::new(180, 200_000), Rung::new(120, 100_000)], + encoder: moq_video::encode::Kind::Hardware, + decoder: moq_video::decode::Kind::Hardware, + source: None, + }; + + let output = moq_net::broadcast::Info::default().produce(); + let consumer = output.consume(); + let transcoder = tokio::spawn(run(source.broadcast.consume(), output, config)); + + let mut subscribers = Vec::new(); + for name in ["video/180p", "video/120p"] { + let track = loop { + match consumer.track(name) { + Ok(track) => break track, + Err(moq_net::Error::NotFound) => tokio::task::yield_now().await, + Err(err) => panic!("rung track {name}: {err}"), + } + }; + subscribers.push((name, track.subscribe(None).await.unwrap())); + } + + for (name, subscriber) in &mut subscribers { + let mut group = subscriber.next_group().await.unwrap().unwrap(); + let payload = group.read_frame().await.unwrap().unwrap(); + let frame = hang::container::Frame::decode(payload).unwrap(); + assert!( + frame.payload.starts_with(&[0, 0, 0, 1]) || frame.payload.starts_with(&[0, 0, 1]), + "{name} output is not Annex-B" + ); + let total = group.finished().await.unwrap(); + assert_eq!(total, 5, "{name} dropped frames"); + } + + producer_task.abort(); + transcoder.abort(); + } + /// Whether a hardware decoder AND encoder are usable here (e.g. a Linux box /// with the NVIDIA driver). Probed through the public API so the hardware /// test skips cleanly on GPU-less CI. diff --git a/rs/moq-transcode/src/rung.rs b/rs/moq-transcode/src/rung.rs index 10a1709a7..64e488d10 100644 --- a/rs/moq-transcode/src/rung.rs +++ b/rs/moq-transcode/src/rung.rs @@ -22,7 +22,7 @@ use tokio::sync::Semaphore; use crate::Error; use crate::catalog::Resolved; -use crate::scale::Scaler; +use crate::feed::{Feed, Item}; /// Cap on transcode pipelines a single rung builds concurrently for on-demand /// group fetches. Each pipeline holds a decoder + encoder session, and hardware @@ -36,8 +36,10 @@ const MAX_CONCURRENT_FETCHES: usize = 4; #[derive(Clone)] pub(crate) struct Rung { pub info: Resolved, - /// The source media track (not yet subscribed; demand drives that). + /// The source media track, for group fetches (not yet subscribed). pub source: moq_net::track::Consumer, + /// The shared live decode of the source, for the live path. + pub feed: Feed, /// The source broadcast, to notice it closing while idle. pub broadcast: moq_net::broadcast::Consumer, /// The source rendition's catalog entry (codec + container). @@ -56,6 +58,17 @@ impl Rung { fn container(&self) -> Result { Ok(moq_mux::catalog::hang::Container::try_from(&self.config.container)?) } + + /// An encoder producing this rung's rendition. + fn encode(&self) -> Result { + let mut config = moq_video::encode::Config::new(self.info.width, self.info.height, self.info.framerate); + config.bitrate = Some(self.info.bitrate); + config.kind = self.encoder.clone(); + // Keyframes are forced at every group boundary; the GOP is only a + // backstop against pathologically long source groups. + config.gop = self.info.framerate.saturating_mul(8).max(1); + Ok(moq_video::encode::Encoder::new(&config)?) + } } /// Serve one requested rung track until it closes or the source ends. @@ -77,8 +90,10 @@ pub(crate) async fn serve(rung: Rung, request: moq_net::track::Request) -> Resul result } -/// The live path: wait for demand, mirror the aggregate subscription upstream, -/// and transcode group for group until demand goes away. +/// The live path: wait for demand, attach to the shared decode [`Feed`], and +/// resize + encode its frames group for group until demand goes away. The +/// heavy lifting (subscription, decode) is shared with every other active rung +/// of this source; only the per-rung resize and encode happen here. async fn live(rung: &Rung, producer: &mut moq_net::track::Producer) -> Result<(), Error> { let demand = producer.demand(); loop { @@ -94,28 +109,42 @@ async fn live(rung: &Rung, producer: &mut moq_net::track::Producer) -> Result<() } } - // Mirror the downstream demand upstream (priority, ordering, start). - let subscription = producer.subscription().unwrap_or_default(); - let mut subscriber = rung.source.subscribe(subscription).await?; + // One listener + encoder per demand session: rate control persists + // across groups, while every group still opens with a forced IDR. + // Dropping them on unused releases the shared decode (if last) and the + // encoder session until someone subscribes again. + let mut listener = rung.feed.listen(); + let mut encoder = rung.encode()?; - // One pipeline per demand session: rate control persists across groups, - // while every group still opens with a forced IDR. - let mut pipeline = rung.pipeline()?; - let container = rung.container()?; + // The output group currently being written, if the feed is mid-group. + let mut current: Option = None; + // Whether the next frame opens its output group (forced IDR). + let mut first = true; 'session: loop { - tokio::select! { - group = subscriber.next_group() => { - let Some(mut source) = group? else { - // The source track ended: the derivative ends with it. - producer.finish()?; - return Ok(()); - }; + let item = tokio::select! { + item = listener.recv() => item, + _ = demand.unused() => { + if let Some(mut output) = current.take() { + // Signal downstream that the group is incomplete. + output.abort(moq_net::Error::Cancel)?; + } + break 'session; + } + }; + + match item { + Some(Item::Group(sequence)) => { + if let Some(mut output) = current.take() { + // A group boundary without an end: treat as incomplete. + output.abort(moq_net::Error::Cancel)?; + } + first = true; // Mirror the source sequence so fetches and rendition // switches map 1:1. - let info = moq_net::group::Info { sequence: source.sequence }; - let mut output = match producer.create_group(info) { - Ok(output) => output, + let info = moq_net::group::Info { sequence }; + current = match producer.create_group(info) { + Ok(output) => Some(output), // A fetch task is already serving this sequence (a consumer // fetched a group at the live edge before the live loop // reached it). The fetch is authoritative and its group @@ -125,20 +154,61 @@ async fn live(rung: &Rung, producer: &mut moq_net::track::Producer) -> Result<() // keyframe. Unifying live + fetch into one cache-backed // serving loop (like the relay) would remove the two-writer // race entirely; tracked as a follow-up. - Err(moq_net::Error::Duplicate) => continue, + Err(moq_net::Error::Duplicate) => None, Err(err) => return Err(err.into()), }; - let done = transcode_group(&mut pipeline, &container, &mut source, &mut output, Some(&demand)).await?; - if !done { - // Demand disappeared mid-group; back to waiting. - break 'session; + } + Some(Item::Frame(frame)) => { + // No open group: attached mid-group, skipped a duplicate, or + // recovering from a lag. Wait for the next boundary. + let Some(output) = &mut current else { continue }; + + let keyframe = first; + first = false; + // The feed decodes at the source's native size; size this + // rung's copy here. A GPU frame resizes on the GPU and feeds + // the encoder without touching the CPU. + let encoded = if (frame.width, frame.height) == (rung.info.width, rung.info.height) { + encoder.encode(&frame, keyframe)? + } else { + let scaled = frame.resize(rung.info.width, rung.info.height)?; + encoder.encode(&scaled, keyframe)? + }; + let timestamp = frame.timestamp_us; + write(output, encoded.into_iter().map(|packet| (timestamp, packet)).collect())?; + } + Some(Item::End) => { + if let Some(mut output) = current.take() { + output.finish()?; + } + } + Some(Item::Lagged) => { + // Fell behind the feed: abandon the group and resume at the + // next boundary rather than stalling other rungs. + if let Some(mut output) = current.take() { + output.abort(moq_net::Error::Cancel)?; } } - _ = demand.unused() => break 'session, + Some(Item::Finished) => { + // The source track ended: the derivative ends with it. + if let Some(mut output) = current.take() { + output.abort(moq_net::Error::Cancel)?; + } + producer.finish()?; + return Ok(()); + } + None => { + // The feed died mid-stream (source or decode error). + if let Some(mut output) = current.take() { + let _ = output.abort(moq_net::Error::Cancel); + } + producer.abort(moq_net::Error::Cancel)?; + return Ok(()); + } } } - // Dropping the subscriber releases the upstream subscription (and the - // encoder) until someone subscribes again. + // listener and encoder drop here, releasing the shared decode session + // (when this was the last rung) and the encoder. } } @@ -211,32 +281,23 @@ async fn fetch(rung: Rung, request: moq_net::track::GroupRequest) -> Result<(), Ok(output) => output, Err(err) => return Err(err.into()), }; - transcode_group(&mut pipeline, &container, &mut source, &mut output, None).await?; + transcode_group(&mut pipeline, &container, &mut source, &mut output).await?; Ok(()) } -/// Transcode one source group into one output group. -/// -/// With a `demand` handle (the live path) the group is abandoned as soon as the -/// track goes unused, returning `Ok(false)`; without one (the fetch path) the -/// group always runs to completion and the encoder is drained at the end. +/// Transcode one fetched source group to completion into one output group, +/// draining the encoder at the end. (The live path rides the shared feed +/// instead; see [`live`].) async fn transcode_group( pipeline: &mut Pipeline, container: &moq_mux::catalog::hang::Container, source: &mut moq_net::group::Consumer, output: &mut moq_net::group::Producer, - demand: Option<&moq_net::track::Demand>, -) -> Result { - match transcode_group_inner(pipeline, container, source, output, demand).await { - Ok(done) => { - if done { - output.finish()?; - } else { - // Demand disappeared mid-group: signal downstream that the - // group is incomplete rather than leaving it short-but-finished. - output.abort(moq_net::Error::Cancel)?; - } - Ok(done) +) -> Result<(), Error> { + match transcode_group_inner(pipeline, container, source, output).await { + Ok(()) => { + output.finish()?; + Ok(()) } Err(err) => { let _ = output.abort(moq_net::Error::Cancel); @@ -250,8 +311,7 @@ async fn transcode_group_inner( container: &moq_mux::catalog::hang::Container, source: &mut moq_net::group::Consumer, output: &mut moq_net::group::Producer, - demand: Option<&moq_net::track::Demand>, -) -> Result { +) -> Result<(), Error> { let mut first = true; // The latest presentation time seen, tracked so a one-shot group can stamp any // packets the encoder still holds at the end. `None` until the first frame: @@ -259,18 +319,7 @@ async fn transcode_group_inner( // zero to seed it with (the source's scale isn't known until a frame arrives). let mut last_timestamp: Option = None; - loop { - let frames = match demand { - Some(demand) => tokio::select! { - frames = container.read(source) => frames, - _ = demand.unused() => return Ok(false), - }, - None => container.read(source).await, - }; - let Some(frames) = frames? else { - break; - }; - + while let Some(frames) = container.read(source).await? { for frame in frames { let timestamp = frame.timestamp; // All source frames share one scale, so this `max` never crosses scales. @@ -290,12 +339,12 @@ async fn transcode_group_inner( } } - if let (None, Some(last_timestamp)) = (demand, last_timestamp) { + if let Some(last_timestamp) = last_timestamp { // One-shot group: drain whatever the encoder still buffers, stamping it with // the last presentation time we saw. No frames read means nothing to drain. write(output, pipeline.finish(last_timestamp)?)?; } - Ok(true) + Ok(()) } /// Append encoded packets to the output group in the legacy hang framing. @@ -307,16 +356,15 @@ fn write(output: &mut moq_net::group::Producer, packets: Vec<(moq_net::Timestamp Ok(()) } -/// Decode -> scale -> encode for one rung. +/// Decode -> resize -> encode for one fetched group of one rung. /// /// The decoder is asked to emit frames at the rung's resolution /// (`decode::Config::resize`). A decoder with a hardware scaler (NVDEC) does, /// and its GPU frames feed the encoder in place: the NVDEC -> NVENC path never /// touches the CPU. Frames that come back at any other size (software decode, -/// or a hardware decoder without a scaler) fall back to the CPU [`Scaler`]. +/// or a hardware decoder without a scaler) get `Frame::resize` instead. struct Pipeline { decoder: moq_video::decode::Decoder, - scaler: Scaler, encoder: moq_video::encode::Encoder, width: u32, height: u32, @@ -329,18 +377,9 @@ impl Pipeline { decode.resize = Some((rung.info.width, rung.info.height)); let decoder = moq_video::decode::Decoder::new(&rung.config, &decode)?; - let mut config = moq_video::encode::Config::new(rung.info.width, rung.info.height, rung.info.framerate); - config.bitrate = Some(rung.info.bitrate); - config.kind = rung.encoder.clone(); - // Keyframes are forced at every group boundary; the GOP is only a - // backstop against pathologically long source groups. - config.gop = rung.info.framerate.saturating_mul(8).max(1); - let encoder = moq_video::encode::Encoder::new(&config)?; - Ok(Self { decoder, - scaler: Scaler::new(rung.info.width, rung.info.height), - encoder, + encoder: rung.encode()?, width: rung.info.width, height: rung.info.height, }) @@ -362,9 +401,7 @@ impl Pipeline { // through as-is, keeping a GPU frame on the GPU. self.encoder.encode(&raw, keyframe)? } else { - let (width, height) = (raw.width, raw.height); - let scaled = self.scaler.scale(&raw.into_i420()?, width, height)?; - self.encoder.encode_i420(scaled, self.width, self.height, keyframe)? + self.encoder.encode(&raw.resize(self.width, self.height)?, keyframe)? }; for packet in encoded { packets.push((raw_timestamp, packet)); diff --git a/rs/moq-transcode/src/scale.rs b/rs/moq-transcode/src/scale.rs deleted file mode 100644 index 957104392..000000000 --- a/rs/moq-transcode/src/scale.rs +++ /dev/null @@ -1,137 +0,0 @@ -//! CPU I420 scaling between decode and re-encode. - -use fast_image_resize::images::{Image, ImageRef}; -use fast_image_resize::{FilterType, PixelType, ResizeAlg, ResizeOptions, Resizer}; - -use crate::Error; - -/// Scales tightly-packed I420 frames to a fixed output resolution. -/// -/// Per-plane SIMD resize: Y at full size, U/V at quarter size. The fallback for -/// decoders without a built-in scaler (openh264, VideoToolbox, Media -/// Foundation); NVDEC scales on the GPU during decode instead, so its frames -/// never come through here. -pub(crate) struct Scaler { - resizer: Resizer, - options: ResizeOptions, - width: u32, - height: u32, -} - -impl Scaler { - /// A scaler producing `width` x `height` output (both even). - pub fn new(width: u32, height: u32) -> Self { - Self { - resizer: Resizer::new(), - // Bilinear: the cheapest convolution, fine for live downscaling. - options: ResizeOptions::new().resize_alg(ResizeAlg::Convolution(FilterType::Bilinear)), - width, - height, - } - } - - /// Scale one packed I420 frame (Y then U then V, no row padding, even - /// dimensions) to the output resolution. - pub fn scale(&mut self, src: &[u8], src_width: u32, src_height: u32) -> Result, Error> { - let src_luma = src_width as usize * src_height as usize; - if src.len() < src_luma * 3 / 2 { - return Err(Error::Scale(format!( - "I420 buffer is {} bytes, expected {} for {src_width}x{src_height}", - src.len(), - src_luma * 3 / 2 - ))); - } - - let dst_luma = self.width as usize * self.height as usize; - let mut dst = vec![0u8; dst_luma * 3 / 2]; - let (dst_y, dst_chroma) = dst.split_at_mut(dst_luma); - let (dst_u, dst_v) = dst_chroma.split_at_mut(dst_luma / 4); - - let src_chroma = src_luma / 4; - let (src_w2, src_h2) = (src_width / 2, src_height / 2); - let (dst_w2, dst_h2) = (self.width / 2, self.height / 2); - - plane( - &mut self.resizer, - &self.options, - &src[..src_luma], - src_width, - src_height, - dst_y, - self.width, - self.height, - )?; - plane( - &mut self.resizer, - &self.options, - &src[src_luma..src_luma + src_chroma], - src_w2, - src_h2, - dst_u, - dst_w2, - dst_h2, - )?; - plane( - &mut self.resizer, - &self.options, - &src[src_luma + src_chroma..src_luma + 2 * src_chroma], - src_w2, - src_h2, - dst_v, - dst_w2, - dst_h2, - )?; - - Ok(dst) - } -} - -/// Resize one 8-bit plane. -#[allow(clippy::too_many_arguments)] -fn plane( - resizer: &mut Resizer, - options: &ResizeOptions, - src: &[u8], - src_width: u32, - src_height: u32, - dst: &mut [u8], - dst_width: u32, - dst_height: u32, -) -> Result<(), Error> { - let src = ImageRef::new(src_width, src_height, src, PixelType::U8).map_err(|e| Error::Scale(e.to_string()))?; - let mut dst = - Image::from_slice_u8(dst_width, dst_height, dst, PixelType::U8).map_err(|e| Error::Scale(e.to_string()))?; - resizer - .resize(&src, &mut dst, options) - .map_err(|e| Error::Scale(e.to_string())) -} - -#[cfg(test)] -mod tests { - use super::*; - - #[test] - fn downscale_preserves_flat_planes() { - // A flat frame stays flat through any correct per-plane resize. - let (w, h) = (320u32, 240u32); - let luma = (w * h) as usize; - let mut src = vec![0u8; luma * 3 / 2]; - src[..luma].fill(120); - src[luma..luma + luma / 4].fill(90); - src[luma + luma / 4..].fill(200); - - let mut scaler = Scaler::new(160, 120); - let dst = scaler.scale(&src, w, h).unwrap(); - let dst_luma = 160 * 120; - assert_eq!(dst.len(), dst_luma * 3 / 2); - assert!(dst[..dst_luma].iter().all(|&b| b == 120)); - assert!(dst[dst_luma..dst_luma + dst_luma / 4].iter().all(|&b| b == 90)); - assert!(dst[dst_luma + dst_luma / 4..].iter().all(|&b| b == 200)); - } - - #[test] - fn rejects_short_buffer() { - let mut scaler = Scaler::new(160, 120); - assert!(matches!(scaler.scale(&[0u8; 16], 320, 240), Err(Error::Scale(_)))); - } -} diff --git a/rs/moq-video/CHANGELOG.md b/rs/moq-video/CHANGELOG.md index aa5fe1660..63f52ae44 100644 --- a/rs/moq-video/CHANGELOG.md +++ b/rs/moq-video/CHANGELOG.md @@ -9,6 +9,13 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Added +- `decode::Frame::resize(width, height)`: a scaled copy of a decoded frame, + preserving the timestamp. A CUDA frame (NVDEC output) resizes on the GPU with + a box-filter kernel (vendored PTX, JIT-compiled by the driver; no CUDA + toolkit needed to build) and stays in device memory; CPU frames resize with a + SIMD bilinear convolution. Fans one decoded stream out to several sizes, e.g. + a transcode ladder sharing one decoder. + - H.264 / H.265 hardware decode on Linux via NVIDIA NVDEC, behind the default-on `nvdec` feature. Decoded frames stay in CUDA device memory and feed the NVENC encoder zero-copy through the new diff --git a/rs/moq-video/Cargo.toml b/rs/moq-video/Cargo.toml index 76214843a..180b1c207 100644 --- a/rs/moq-video/Cargo.toml +++ b/rs/moq-video/Cargo.toml @@ -39,6 +39,9 @@ vaapi = ["dep:moq-vaapi"] [dependencies] anyhow = "1" bytes = "1" +# CPU I420 resize (SIMD per-plane convolution), the software half of +# decode::Frame::resize; CUDA frames resize on the GPU instead. +fast_image_resize = "5" # Catalog types (VideoConfig / VideoCodec) the decode consumer reads. hang = { workspace = true } moq-mux = { workspace = true } @@ -61,7 +64,10 @@ yuv = "0.8.14" # links on a GPU-less builder and still starts on machines without the NVIDIA # driver, falling back to the next encoder (see backend::open). `cuda-12020` pins # the CUDA API version so the build needs no CUDA toolkit. -cudarc = { version = "0.19", optional = true, default-features = false, features = ["driver", "fallback-dynamic-loading", "cuda-12020"] } +# The `nvrtc` feature only unlocks the `Ptx` type and `load_module` (both free of +# extra dependencies); libnvrtc itself is never loaded since we ship pre-built PTX +# (see frame/nv12_resize.ptx) that the driver JIT-compiles. +cudarc = { version = "0.19", optional = true, default-features = false, features = ["driver", "fallback-dynamic-loading", "cuda-12020", "nvrtc"] } # Probe for the NVIDIA driver libraries before calling cudarc / the NVENC SDK, # which panic (process-abort under release `panic = "abort"`) if their library is # absent. Lets a GPU-less host fall back to the next encoder instead of crashing. diff --git a/rs/moq-video/src/decode/mod.rs b/rs/moq-video/src/decode/mod.rs index 6195e4446..b161d0c7a 100644 --- a/rs/moq-video/src/decode/mod.rs +++ b/rs/moq-video/src/decode/mod.rs @@ -46,6 +46,48 @@ pub struct Frame { } impl Frame { + /// A copy of this frame scaled to `width` x `height` (both even and + /// non-zero), preserving the timestamp. A GPU frame scales on the GPU (a + /// box filter, correct at any downscale factor) and stays there, so + /// resize -> [`encode`](crate::encode::Encoder::encode) never touches the + /// CPU; a CPU frame scales on the CPU. When one output size is enough, + /// prefer decoding straight to it ([`Config::resize`]), which is free on + /// decoders with a hardware scaler; this method is for fanning one decoded + /// stream out to several sizes. + pub fn resize(&self, width: u32, height: u32) -> Result { + if width == 0 || height == 0 || width % 2 != 0 || height % 2 != 0 { + return Err(Error::Codec(anyhow::anyhow!( + "resize to {width}x{height}: dimensions must be even and non-zero" + ))); + } + + let inner = match &self.inner { + crate::frame::Frame::I420(i420) => crate::frame::Frame::I420(i420.resize(width, height)?), + #[cfg(all(target_os = "linux", feature = "nvdec"))] + crate::frame::Frame::Cuda(cuda) => match cuda.resize(width, height) { + Ok(scaled) => crate::frame::Frame::Cuda(scaled), + // E.g. the driver rejected the vendored PTX: degrade to a CPU + // resize (download once) instead of killing the stream. + Err(err) => { + static WARN_ONCE: std::sync::Once = std::sync::Once::new(); + WARN_ONCE.call_once(|| tracing::warn!(%err, "GPU resize failed; falling back to the CPU")); + crate::frame::Frame::I420(cuda.download_i420()?.resize(width, height)?) + } + }, + // Capture-only surfaces (CVPixelBuffer / D3D11) never appear in + // decoded frames, but stay total: download, then scale on the CPU. + #[allow(unreachable_patterns)] + other => crate::frame::Frame::I420(other.to_i420()?.into_owned().resize(width, height)?), + }; + + Ok(Frame { + timestamp_us: self.timestamp_us, + width, + height, + inner, + }) + } + /// The frame as tightly-packed I420 (YUV 4:2:0, BT.601 limited range): Y /// (`width * height` bytes), then U, then V (`width/2 * height/2` bytes /// each), no row padding. Free for a CPU frame; downloads a GPU frame. diff --git a/rs/moq-video/src/frame.rs b/rs/moq-video/src/frame.rs index a8c3d99e5..969b3f93a 100644 --- a/rs/moq-video/src/frame.rs +++ b/rs/moq-video/src/frame.rs @@ -223,6 +223,42 @@ impl I420 { Ok(Self { width, height, data }) } + /// Resize to `width` x `height` (both even) with a per-plane SIMD bilinear + /// convolution: Y at full size, U/V at quarter size. The CPU half of + /// [`decode::Frame::resize`](crate::decode::Frame::resize). + pub(crate) fn resize(&self, width: u32, height: u32) -> Result { + use fast_image_resize::images::{Image, ImageRef}; + use fast_image_resize::{FilterType, PixelType, ResizeAlg, ResizeOptions, Resizer}; + + let mut resizer = Resizer::new(); + // Bilinear convolution: proper filter support at any downscale factor, + // the cheapest option that doesn't alias. + let options = ResizeOptions::new().resize_alg(ResizeAlg::Convolution(FilterType::Bilinear)); + + let mut plane = |src: &[u8], sw: u32, sh: u32, dst: &mut [u8], dw: u32, dh: u32| -> Result<(), Error> { + let src = ImageRef::new(sw, sh, src, PixelType::U8) + .map_err(|e| Error::Codec(anyhow::anyhow!("resize source: {e}")))?; + let mut dst = Image::from_slice_u8(dw, dh, dst, PixelType::U8) + .map_err(|e| Error::Codec(anyhow::anyhow!("resize destination: {e}")))?; + resizer + .resize(&src, &mut dst, &options) + .map_err(|e| Error::Codec(anyhow::anyhow!("resize: {e}"))) + }; + + let luma = width as usize * height as usize; + let mut data = vec![0u8; Self::len(width, height)]; + let (y_dst, chroma) = data.split_at_mut(luma); + let (u_dst, v_dst) = chroma.split_at_mut(luma / 4); + + plane(self.y(), self.width, self.height, y_dst, width, height)?; + let (sw2, sh2) = (self.width / 2, self.height / 2); + let (dw2, dh2) = (width / 2, height / 2); + plane(self.u(), sw2, sh2, u_dst, dw2, dh2)?; + plane(self.v(), sw2, sh2, v_dst, dw2, dh2)?; + + Ok(Self { width, height, data }) + } + /// Flatten the three planes of a freshly-converted image into one tightly /// packed I420 buffer (Y, then U, then V). fn pack(planar: &YuvPlanarImageMut, width: u32, height: u32) -> Self { @@ -381,13 +417,44 @@ pub(crate) mod macos { #[cfg(all(target_os = "linux", feature = "nvdec"))] pub(crate) mod cuda { - use std::sync::Arc; + use std::sync::{Arc, OnceLock}; - use cudarc::driver::{CudaContext, result}; + use cudarc::driver::{CudaContext, CudaFunction, LaunchConfig, PushKernelArg, result}; use super::I420; use crate::Error; + /// The NV12 box-filter resize kernels, vendored as PTX (see nv12_resize.cu) + /// and JIT-compiled by the driver, so building needs no CUDA toolkit. + const RESIZE_PTX: &str = include_str!("frame/nv12_resize.ptx"); + + /// The loaded resize kernels, one per process (everything runs in the + /// device's primary context, so one module serves every frame). + struct Kernels { + luma: CudaFunction, + chroma: CudaFunction, + } + + fn kernels(ctx: &Arc) -> Result<&'static Kernels, Error> { + static KERNELS: OnceLock> = OnceLock::new(); + KERNELS + .get_or_init(|| { + let module = ctx + .load_module(cudarc::nvrtc::Ptx::from_src(RESIZE_PTX)) + .map_err(|e| format!("load nv12_resize PTX: {e:?}"))?; + Ok(Kernels { + luma: module + .load_function("resize_luma") + .map_err(|e| format!("load resize_luma: {e:?}"))?, + chroma: module + .load_function("resize_chroma") + .map_err(|e| format!("load resize_chroma: {e:?}"))?, + }) + }) + .as_ref() + .map_err(|e| Error::Codec(anyhow::anyhow!("CUDA resize unavailable: {e}"))) + } + /// An owned device allocation. Plain `cuMemAlloc` on purpose: NVENC's /// resource registration rejects stream-ordered pool memory /// (`cuMemAllocAsync`), which is what cudarc's `CudaSlice` uses on any GPU @@ -493,6 +560,81 @@ pub(crate) mod cuda { data, }) } + + /// Resize to `width` x `height` (both even) with the box-filter kernel, + /// staying in device memory. The GPU half of + /// [`decode::Frame::resize`](crate::decode::Frame::resize). + pub(crate) fn resize(&self, width: u32, height: u32) -> Result { + let ctx = &self.buf.ctx; + let kernels = kernels(ctx)?; + + // Destination row pitch aligned to 256 bytes: comfortable coalescing + // and a multiple of 4 as NVENC registration requires. + let pitch = width.next_multiple_of(256); + let dst = Self::alloc(ctx, width, height, pitch)?; + + let stream = ctx.default_stream(); + let block = (16u32, 16, 1); + let grid = |w: u32, h: u32| (w.div_ceil(16), h.div_ceil(16), 1); + let launch_err = |plane: &str, e| Error::Codec(anyhow::anyhow!("CUDA resize {plane}: {e:?}")); + + // Luma plane: one thread per destination pixel. + // + // SAFETY: both buffers are live NV12 allocations of pitch * height * + // 3 / 2 bytes, and the kernels bound every access by the dimensions + // passed alongside the pointers. + unsafe { + stream + .launch_builder(&kernels.luma) + .arg(&self.buf.ptr) + .arg(&self.pitch) + .arg(&self.width) + .arg(&self.height) + .arg(&dst.buf.ptr) + .arg(&pitch) + .arg(&width) + .arg(&height) + .launch(LaunchConfig { + grid_dim: grid(width, height), + block_dim: block, + shared_mem_bytes: 0, + }) + } + .map_err(|e| launch_err("luma", e))?; + + // Chroma plane: one thread per destination UV pair, offset past the + // luma rows in both buffers. + let src_uv = self.buf.ptr + u64::from(self.pitch) * u64::from(self.height); + let dst_uv = dst.buf.ptr + u64::from(pitch) * u64::from(height); + let (src_pw, src_ph) = (self.width / 2, self.height / 2); + let (dst_pw, dst_ph) = (width / 2, height / 2); + // SAFETY: as above; the UV offsets stay inside the same allocations. + unsafe { + stream + .launch_builder(&kernels.chroma) + .arg(&src_uv) + .arg(&self.pitch) + .arg(&src_pw) + .arg(&src_ph) + .arg(&dst_uv) + .arg(&pitch) + .arg(&dst_pw) + .arg(&dst_ph) + .launch(LaunchConfig { + grid_dim: grid(dst_pw, dst_ph), + block_dim: block, + shared_mem_bytes: 0, + }) + } + .map_err(|e| launch_err("chroma", e))?; + + // The frame may head straight to NVENC (which does not order against + // our stream), so wait for the kernels rather than queueing. + stream + .synchronize() + .map_err(|e| Error::Codec(anyhow::anyhow!("CUDA resize sync: {e:?}")))?; + Ok(dst) + } } } @@ -671,3 +813,102 @@ pub(crate) mod d3d11 { } } } + +#[cfg(test)] +mod tests { + use super::I420; + + /// A gradient I420 frame with structure in every plane, so resize bugs + /// (plane swaps, stride mistakes) shift the averages measurably. + fn gradient_i420(width: u32, height: u32) -> I420 { + let (w, h) = (width as usize, height as usize); + let (cw, ch) = (w / 2, h / 2); + let mut data = vec![0u8; I420::len(width, height)]; + let (y, chroma) = data.split_at_mut(w * h); + let (u, v) = chroma.split_at_mut(cw * ch); + for row in 0..h { + for col in 0..w { + y[row * w + col] = ((col * 255) / w) as u8; + } + } + for row in 0..ch { + for col in 0..cw { + u[row * cw + col] = ((row * 255) / ch) as u8; + v[row * cw + col] = (((row + col) * 255) / (ch + cw)) as u8; + } + } + I420 { width, height, data } + } + + /// Mean absolute error between two equal-length planes. + fn mae(a: &[u8], b: &[u8]) -> u64 { + assert_eq!(a.len(), b.len()); + a.iter().zip(b).map(|(x, y)| x.abs_diff(*y) as u64).sum::() / a.len() as u64 + } + + /// The CPU resize follows the source gradients at any downscale factor: a + /// horizontal luma ramp stays a ramp, and the chroma ramps follow too. + #[test] + fn i420_resize_follows_gradients() { + let src = gradient_i420(320, 240); + let dst = src.resize(128, 96).unwrap(); + assert_eq!((dst.width, dst.height), (128, 96)); + + // Reference: the same gradients sampled at the destination geometry. + let expected = gradient_i420(128, 96); + assert!(mae(dst.y(), expected.y()) < 4, "luma ramp drifted"); + assert!(mae(dst.u(), expected.u()) < 4, "u ramp drifted"); + assert!(mae(dst.v(), expected.v()) < 4, "v ramp drifted"); + } + + /// GPU (box filter) and CPU (bilinear convolution) resizes agree on a + /// smooth gradient. Runs on real hardware; skips without the NVIDIA driver. + #[cfg(all(target_os = "linux", feature = "nvdec"))] + #[test] + fn cuda_resize_matches_cpu() { + use std::sync::Arc; + + use cudarc::driver::{CudaContext, result}; + + use super::cuda; + + // Same probe as the codec backends: no driver, no test. + if unsafe { libloading::Library::new("libcuda.so.1") }.is_err() { + return; + } + let Ok(ctx): Result, _> = CudaContext::new(0) else { + return; + }; + + let (w, h) = (322u32, 242u32); // odd-ish sizes: exercise pitch != width + let src_i420 = gradient_i420(w, h); + + // Upload as pitched NV12: Y rows, then interleaved UV rows. + let pitch = 512u32; + let frame = cuda::Frame::alloc(&ctx, w, h, pitch).unwrap(); + let mut host = vec![0u8; pitch as usize * h as usize * 3 / 2]; + for row in 0..h as usize { + let dst = row * pitch as usize; + host[dst..dst + w as usize].copy_from_slice(&src_i420.y()[row * w as usize..(row + 1) * w as usize]); + } + let (cw, ch) = (w as usize / 2, h as usize / 2); + for row in 0..ch { + let dst = (h as usize + row) * pitch as usize; + for col in 0..cw { + host[dst + 2 * col] = src_i420.u()[row * cw + col]; + host[dst + 2 * col + 1] = src_i420.v()[row * cw + col]; + } + } + // SAFETY: the frame's buffer is exactly host.len() bytes. + unsafe { result::memcpy_htod_sync(frame.device_ptr(), &host) }.unwrap(); + + let scaled = frame.resize(160, 120).unwrap(); + let gpu = scaled.download_i420().unwrap(); + let cpu = src_i420.resize(160, 120).unwrap(); + + assert_eq!((gpu.width, gpu.height), (160, 120)); + assert!(mae(gpu.y(), cpu.y()) < 4, "GPU and CPU luma disagree"); + assert!(mae(gpu.u(), cpu.u()) < 4, "GPU and CPU u disagree"); + assert!(mae(gpu.v(), cpu.v()) < 4, "GPU and CPU v disagree"); + } +} diff --git a/rs/moq-video/src/frame/nv12_resize.cu b/rs/moq-video/src/frame/nv12_resize.cu new file mode 100644 index 000000000..083c6f10d --- /dev/null +++ b/rs/moq-video/src/frame/nv12_resize.cu @@ -0,0 +1,75 @@ +// Box-average NV12 resize, the GPU half of decode::Frame::resize. +// +// One thread per destination pixel (luma) or destination UV pair (chroma). +// Each destination pixel averages its full source box, so arbitrary downscale +// factors don't alias the way a fixed 4-tap bilinear would; at 1:1 it degrades +// to a copy. Ladder rungs never upscale, so upscale quality is irrelevant +// (the box degenerates to nearest-neighbor). +// +// Vendored PTX: this file is compiled offline to nv12_resize.ptx (see the +// comment there for the exact command), which is embedded and JIT-compiled by +// the driver at runtime. Building the crate needs no CUDA toolkit, matching +// the dlopen-only design of the NVENC/NVDEC backends. If you edit this file, +// regenerate the PTX next to it. + +extern "C" __global__ void resize_luma( + const unsigned char *src, unsigned int src_pitch, unsigned int src_w, unsigned int src_h, + unsigned char *dst, unsigned int dst_pitch, unsigned int dst_w, unsigned int dst_h) +{ + unsigned int x = blockIdx.x * blockDim.x + threadIdx.x; + unsigned int y = blockIdx.y * blockDim.y + threadIdx.y; + if (x >= dst_w || y >= dst_h) + return; + + // The half-open source box covered by this destination pixel, at least + // one source pixel even when upscaling. + unsigned int x0 = x * src_w / dst_w; + unsigned int x1 = (x + 1) * src_w / dst_w; + if (x1 <= x0) + x1 = x0 + 1; + unsigned int y0 = y * src_h / dst_h; + unsigned int y1 = (y + 1) * src_h / dst_h; + if (y1 <= y0) + y1 = y0 + 1; + + unsigned int sum = 0; + for (unsigned int sy = y0; sy < y1; sy++) + for (unsigned int sx = x0; sx < x1; sx++) + sum += src[sy * src_pitch + sx]; + + unsigned int n = (x1 - x0) * (y1 - y0); + dst[y * dst_pitch + x] = (unsigned char)((sum + n / 2) / n); +} + +// Same box average over the interleaved UV plane. Widths and x are in UV +// *pairs* (2 bytes each); U and V accumulate separately. +extern "C" __global__ void resize_chroma( + const unsigned char *src, unsigned int src_pitch, unsigned int src_w, unsigned int src_h, + unsigned char *dst, unsigned int dst_pitch, unsigned int dst_w, unsigned int dst_h) +{ + unsigned int x = blockIdx.x * blockDim.x + threadIdx.x; + unsigned int y = blockIdx.y * blockDim.y + threadIdx.y; + if (x >= dst_w || y >= dst_h) + return; + + unsigned int x0 = x * src_w / dst_w; + unsigned int x1 = (x + 1) * src_w / dst_w; + if (x1 <= x0) + x1 = x0 + 1; + unsigned int y0 = y * src_h / dst_h; + unsigned int y1 = (y + 1) * src_h / dst_h; + if (y1 <= y0) + y1 = y0 + 1; + + unsigned int sum_u = 0; + unsigned int sum_v = 0; + for (unsigned int sy = y0; sy < y1; sy++) + for (unsigned int sx = x0; sx < x1; sx++) { + sum_u += src[sy * src_pitch + 2 * sx]; + sum_v += src[sy * src_pitch + 2 * sx + 1]; + } + + unsigned int n = (x1 - x0) * (y1 - y0); + dst[y * dst_pitch + 2 * x] = (unsigned char)((sum_u + n / 2) / n); + dst[y * dst_pitch + 2 * x + 1] = (unsigned char)((sum_v + n / 2) / n); +} diff --git a/rs/moq-video/src/frame/nv12_resize.ptx b/rs/moq-video/src/frame/nv12_resize.ptx new file mode 100644 index 000000000..707da2d16 --- /dev/null +++ b/rs/moq-video/src/frame/nv12_resize.ptx @@ -0,0 +1,384 @@ +// Vendored PTX for nv12_resize.cu, JIT-compiled by the NVIDIA driver at +// runtime (no CUDA toolkit needed to build this crate). Targets PTX ISA 8.2 / +// sm_52 so it loads on any driver cudarc's cuda-12020 pin supports (>= 535) +// and any GPU that has NVDEC. Regenerate after editing the .cu: +// nvcc -ptx -arch=compute_52 -o nv12_resize.ptx nv12_resize.cu +// using a CUDA 12.2 nvcc (e.g. from nixpkgs 24.05: cudaPackages_12_2.cuda_nvcc +// with -ccbin gcc12 and -I cuda_cudart's include dir). +// +// Generated by NVIDIA NVVM Compiler +// +// Compiler Build ID: CL-33191640 +// Cuda compilation tools, release 12.2, V12.2.140 +// Based on NVVM 7.0.1 +// + +.version 8.2 +.target sm_52 +.address_size 64 + + // .globl resize_luma + +.visible .entry resize_luma( + .param .u64 resize_luma_param_0, + .param .u32 resize_luma_param_1, + .param .u32 resize_luma_param_2, + .param .u32 resize_luma_param_3, + .param .u64 resize_luma_param_4, + .param .u32 resize_luma_param_5, + .param .u32 resize_luma_param_6, + .param .u32 resize_luma_param_7 +) +{ + .reg .pred %p<14>; + .reg .b32 %r<93>; + .reg .b64 %rd<21>; + + + ld.param.u64 %rd3, [resize_luma_param_0]; + ld.param.u32 %r31, [resize_luma_param_1]; + ld.param.u32 %r32, [resize_luma_param_2]; + ld.param.u32 %r33, [resize_luma_param_3]; + ld.param.u64 %rd2, [resize_luma_param_4]; + ld.param.u32 %r34, [resize_luma_param_5]; + ld.param.u32 %r35, [resize_luma_param_6]; + ld.param.u32 %r36, [resize_luma_param_7]; + cvta.to.global.u64 %rd1, %rd3; + mov.u32 %r37, %ntid.x; + mov.u32 %r38, %ctaid.x; + mov.u32 %r39, %tid.x; + mad.lo.s32 %r1, %r38, %r37, %r39; + mov.u32 %r40, %ntid.y; + mov.u32 %r41, %ctaid.y; + mov.u32 %r42, %tid.y; + mad.lo.s32 %r2, %r41, %r40, %r42; + setp.ge.u32 %p1, %r1, %r35; + setp.ge.u32 %p2, %r2, %r36; + or.pred %p3, %p1, %p2; + @%p3 bra $L__BB0_14; + + mul.lo.s32 %r43, %r1, %r32; + div.u32 %r3, %r43, %r35; + add.s32 %r44, %r43, %r32; + div.u32 %r45, %r44, %r35; + setp.gt.u32 %p4, %r45, %r3; + add.s32 %r4, %r3, 1; + selp.b32 %r5, %r45, %r4, %p4; + mul.lo.s32 %r46, %r2, %r33; + div.u32 %r6, %r46, %r36; + add.s32 %r47, %r46, %r33; + div.u32 %r48, %r47, %r36; + setp.gt.u32 %p5, %r48, %r6; + add.s32 %r49, %r6, 1; + selp.b32 %r7, %r48, %r49, %p5; + setp.gt.u32 %p6, %r7, %r6; + @%p6 bra $L__BB0_3; + bra.uni $L__BB0_2; + +$L__BB0_3: + sub.s32 %r52, %r5, %r3; + not.b32 %r53, %r3; + add.s32 %r8, %r5, %r53; + and.b32 %r9, %r52, 3; + add.s32 %r10, %r3, 2; + add.s32 %r11, %r3, 3; + mov.u32 %r91, 0; + mov.u32 %r83, %r6; + +$L__BB0_4: + setp.le.u32 %p7, %r5, %r3; + @%p7 bra $L__BB0_12; + + setp.eq.s32 %p8, %r9, 0; + mul.lo.s32 %r14, %r83, %r31; + mov.u32 %r85, %r3; + @%p8 bra $L__BB0_9; + + setp.eq.s32 %p9, %r9, 1; + add.s32 %r55, %r3, %r14; + cvt.u64.u32 %rd4, %r55; + add.s64 %rd5, %rd1, %rd4; + ld.global.u8 %r56, [%rd5]; + add.s32 %r91, %r91, %r56; + mov.u32 %r85, %r4; + @%p9 bra $L__BB0_9; + + setp.eq.s32 %p10, %r9, 2; + add.s32 %r57, %r4, %r14; + cvt.u64.u32 %rd6, %r57; + add.s64 %rd7, %rd1, %rd6; + ld.global.u8 %r58, [%rd7]; + add.s32 %r91, %r91, %r58; + mov.u32 %r85, %r10; + @%p10 bra $L__BB0_9; + + add.s32 %r59, %r10, %r14; + cvt.u64.u32 %rd8, %r59; + add.s64 %rd9, %rd1, %rd8; + ld.global.u8 %r60, [%rd9]; + add.s32 %r91, %r91, %r60; + mov.u32 %r85, %r11; + +$L__BB0_9: + setp.lt.u32 %p11, %r8, 3; + @%p11 bra $L__BB0_12; + + add.s32 %r61, %r85, %r14; + add.s32 %r88, %r61, 3; + +$L__BB0_11: + add.s32 %r62, %r14, %r85; + cvt.u64.u32 %rd10, %r62; + add.s64 %rd11, %rd1, %rd10; + ld.global.u8 %r63, [%rd11]; + add.s32 %r64, %r91, %r63; + add.s32 %r65, %r88, -2; + cvt.u64.u32 %rd12, %r65; + add.s64 %rd13, %rd1, %rd12; + ld.global.u8 %r66, [%rd13]; + add.s32 %r67, %r64, %r66; + add.s32 %r68, %r88, -1; + cvt.u64.u32 %rd14, %r68; + add.s64 %rd15, %rd1, %rd14; + ld.global.u8 %r69, [%rd15]; + add.s32 %r70, %r67, %r69; + cvt.u64.u32 %rd16, %r88; + add.s64 %rd17, %rd1, %rd16; + ld.global.u8 %r71, [%rd17]; + add.s32 %r91, %r70, %r71; + add.s32 %r88, %r88, 4; + add.s32 %r85, %r85, 4; + setp.lt.u32 %p12, %r85, %r5; + @%p12 bra $L__BB0_11; + +$L__BB0_12: + add.s32 %r83, %r83, 1; + setp.lt.u32 %p13, %r83, %r7; + @%p13 bra $L__BB0_4; + bra.uni $L__BB0_13; + +$L__BB0_2: + mov.u32 %r91, 0; + +$L__BB0_13: + sub.s32 %r72, %r7, %r6; + sub.s32 %r73, %r5, %r3; + mul.lo.s32 %r74, %r72, %r73; + shr.u32 %r75, %r74, 1; + add.s32 %r76, %r91, %r75; + div.u32 %r77, %r76, %r74; + mad.lo.s32 %r82, %r2, %r34, %r1; + cvt.u64.u32 %rd18, %r82; + cvta.to.global.u64 %rd19, %rd2; + add.s64 %rd20, %rd19, %rd18; + st.global.u8 [%rd20], %r77; + +$L__BB0_14: + ret; + +} + // .globl resize_chroma +.visible .entry resize_chroma( + .param .u64 resize_chroma_param_0, + .param .u32 resize_chroma_param_1, + .param .u32 resize_chroma_param_2, + .param .u32 resize_chroma_param_3, + .param .u64 resize_chroma_param_4, + .param .u32 resize_chroma_param_5, + .param .u32 resize_chroma_param_6, + .param .u32 resize_chroma_param_7 +) +{ + .reg .pred %p<14>; + .reg .b32 %r<128>; + .reg .b64 %rd<37>; + + + ld.param.u64 %rd3, [resize_chroma_param_0]; + ld.param.u32 %r41, [resize_chroma_param_1]; + ld.param.u32 %r42, [resize_chroma_param_2]; + ld.param.u32 %r43, [resize_chroma_param_3]; + ld.param.u64 %rd2, [resize_chroma_param_4]; + ld.param.u32 %r44, [resize_chroma_param_5]; + ld.param.u32 %r45, [resize_chroma_param_6]; + ld.param.u32 %r46, [resize_chroma_param_7]; + cvta.to.global.u64 %rd1, %rd3; + mov.u32 %r47, %ntid.x; + mov.u32 %r48, %ctaid.x; + mov.u32 %r49, %tid.x; + mad.lo.s32 %r1, %r48, %r47, %r49; + mov.u32 %r50, %ntid.y; + mov.u32 %r51, %ctaid.y; + mov.u32 %r52, %tid.y; + mad.lo.s32 %r2, %r51, %r50, %r52; + setp.ge.u32 %p1, %r1, %r45; + setp.ge.u32 %p2, %r2, %r46; + or.pred %p3, %p1, %p2; + @%p3 bra $L__BB1_13; + + mul.lo.s32 %r53, %r1, %r42; + div.u32 %r3, %r53, %r45; + add.s32 %r54, %r53, %r42; + div.u32 %r55, %r54, %r45; + setp.gt.u32 %p4, %r55, %r3; + add.s32 %r4, %r3, 1; + selp.b32 %r5, %r55, %r4, %p4; + mul.lo.s32 %r56, %r2, %r43; + div.u32 %r6, %r56, %r46; + add.s32 %r57, %r56, %r43; + div.u32 %r58, %r57, %r46; + setp.gt.u32 %p5, %r58, %r6; + add.s32 %r59, %r6, 1; + selp.b32 %r7, %r58, %r59, %p5; + setp.gt.u32 %p6, %r7, %r6; + @%p6 bra $L__BB1_3; + bra.uni $L__BB1_2; + +$L__BB1_3: + sub.s32 %r64, %r5, %r3; + not.b32 %r65, %r3; + add.s32 %r8, %r5, %r65; + and.b32 %r9, %r64, 3; + shl.b32 %r10, %r3, 1; + shl.b32 %r11, %r4, 1; + add.s32 %r12, %r3, 2; + shl.b32 %r13, %r12, 1; + add.s32 %r14, %r3, 3; + mov.u32 %r125, 0; + mov.u32 %r113, %r6; + mov.u32 %r124, %r125; + +$L__BB1_4: + setp.le.u32 %p7, %r5, %r3; + @%p7 bra $L__BB1_11; + + setp.eq.s32 %p8, %r9, 0; + mul.lo.s32 %r18, %r113, %r41; + mov.u32 %r116, %r3; + @%p8 bra $L__BB1_9; + + setp.eq.s32 %p9, %r9, 1; + add.s32 %r67, %r10, %r18; + cvt.u64.u32 %rd4, %r67; + add.s64 %rd5, %rd1, %rd4; + ld.global.u8 %r68, [%rd5]; + add.s32 %r124, %r124, %r68; + add.s32 %r69, %r67, 1; + cvt.u64.u32 %rd6, %r69; + add.s64 %rd7, %rd1, %rd6; + ld.global.u8 %r70, [%rd7]; + add.s32 %r125, %r125, %r70; + mov.u32 %r116, %r4; + @%p9 bra $L__BB1_9; + + setp.eq.s32 %p10, %r9, 2; + add.s32 %r71, %r11, %r18; + cvt.u64.u32 %rd8, %r71; + add.s64 %rd9, %rd1, %rd8; + ld.global.u8 %r72, [%rd9]; + add.s32 %r124, %r124, %r72; + add.s32 %r73, %r71, 1; + cvt.u64.u32 %rd10, %r73; + add.s64 %rd11, %rd1, %rd10; + ld.global.u8 %r74, [%rd11]; + add.s32 %r125, %r125, %r74; + mov.u32 %r116, %r12; + @%p10 bra $L__BB1_9; + + add.s32 %r75, %r13, %r18; + cvt.u64.u32 %rd12, %r75; + add.s64 %rd13, %rd1, %rd12; + ld.global.u8 %r76, [%rd13]; + add.s32 %r124, %r124, %r76; + add.s32 %r77, %r75, 1; + cvt.u64.u32 %rd14, %r77; + add.s64 %rd15, %rd1, %rd14; + ld.global.u8 %r78, [%rd15]; + add.s32 %r125, %r125, %r78; + mov.u32 %r116, %r14; + +$L__BB1_9: + setp.lt.u32 %p11, %r8, 3; + @%p11 bra $L__BB1_11; + +$L__BB1_10: + shl.b32 %r79, %r116, 1; + add.s32 %r80, %r79, %r18; + cvt.u64.u32 %rd16, %r80; + add.s64 %rd17, %rd1, %rd16; + ld.global.u8 %r81, [%rd17]; + add.s32 %r82, %r124, %r81; + add.s32 %r83, %r80, 1; + cvt.u64.u32 %rd18, %r83; + add.s64 %rd19, %rd1, %rd18; + ld.global.u8 %r84, [%rd19]; + add.s32 %r85, %r125, %r84; + add.s32 %r86, %r80, 2; + cvt.u64.u32 %rd20, %r86; + add.s64 %rd21, %rd1, %rd20; + ld.global.u8 %r87, [%rd21]; + add.s32 %r88, %r82, %r87; + add.s32 %r89, %r80, 3; + cvt.u64.u32 %rd22, %r89; + add.s64 %rd23, %rd1, %rd22; + ld.global.u8 %r90, [%rd23]; + add.s32 %r91, %r85, %r90; + add.s32 %r92, %r80, 4; + cvt.u64.u32 %rd24, %r92; + add.s64 %rd25, %rd1, %rd24; + ld.global.u8 %r93, [%rd25]; + add.s32 %r94, %r88, %r93; + add.s32 %r95, %r80, 5; + cvt.u64.u32 %rd26, %r95; + add.s64 %rd27, %rd1, %rd26; + ld.global.u8 %r96, [%rd27]; + add.s32 %r97, %r91, %r96; + add.s32 %r98, %r80, 6; + cvt.u64.u32 %rd28, %r98; + add.s64 %rd29, %rd1, %rd28; + ld.global.u8 %r99, [%rd29]; + add.s32 %r124, %r94, %r99; + add.s32 %r100, %r80, 7; + cvt.u64.u32 %rd30, %r100; + add.s64 %rd31, %rd1, %rd30; + ld.global.u8 %r101, [%rd31]; + add.s32 %r125, %r97, %r101; + add.s32 %r116, %r116, 4; + setp.lt.u32 %p12, %r116, %r5; + @%p12 bra $L__BB1_10; + +$L__BB1_11: + add.s32 %r113, %r113, 1; + setp.lt.u32 %p13, %r113, %r7; + @%p13 bra $L__BB1_4; + bra.uni $L__BB1_12; + +$L__BB1_2: + mov.u32 %r124, 0; + mov.u32 %r125, %r124; + +$L__BB1_12: + cvta.to.global.u64 %rd32, %rd2; + sub.s32 %r102, %r7, %r6; + sub.s32 %r103, %r5, %r3; + mul.lo.s32 %r104, %r102, %r103; + shr.u32 %r105, %r104, 1; + add.s32 %r106, %r124, %r105; + div.u32 %r107, %r106, %r104; + shl.b32 %r108, %r1, 1; + mad.lo.s32 %r109, %r2, %r44, %r108; + cvt.u64.u32 %rd33, %r109; + add.s64 %rd34, %rd32, %rd33; + st.global.u8 [%rd34], %r107; + add.s32 %r110, %r125, %r105; + div.u32 %r111, %r110, %r104; + add.s32 %r112, %r109, 1; + cvt.u64.u32 %rd35, %r112; + add.s64 %rd36, %rd32, %rd35; + st.global.u8 [%rd36], %r111; + +$L__BB1_13: + ret; + +} + From 234dbff6afaaac63d08ae7b013d785e1c4d8bb57 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Fri, 10 Jul 2026 13:44:55 -0700 Subject: [PATCH 2/3] fix(moq-video,moq-transcode): review follow-ups for the shared-decode PR - Sort the root Cargo.toml dependencies (cargo-sort, the CI failure) and reattach the default-features comment that drifted onto the wrong entry. - Cache the CPU resizer per thread: I420::resize built a fresh fast_image_resize::Resizer per call, discarding its convolution state on every frame of a software live transcode. - Feed::listen treats a finished-but-not-yet-cleared decode session as idle and starts a fresh one, instead of subscribing to a channel that can only yield Closed (a small teardown race that would abort the rung track). Co-Authored-By: Claude Fable 5 --- Cargo.toml | 10 +++++----- rs/moq-transcode/src/feed.rs | 7 +++++++ rs/moq-video/src/frame.rs | 31 ++++++++++++++++++++++++------- 3 files changed, 36 insertions(+), 12 deletions(-) diff --git a/Cargo.toml b/Cargo.toml index 156524c74..74cae5704 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -83,13 +83,13 @@ moq-rtc = { version = "0.1.2", path = "rs/moq-rtc" } moq-rtmp = { version = "0.1.2", path = "rs/moq-rtmp" } moq-srt = { version = "0.1.1", path = "rs/moq-srt" } moq-token = { version = "0.6", path = "rs/moq-token" } +# default-features off (the hardware codecs) on moq-transcode and moq-video so +# each consumer opts in: binaries (libmoq, moq-boy, moq-cli) enable them, but a +# self-compiler can leave them off to drop the CUDA / libva deps. Both crates +# still default them on when depended on directly. +moq-transcode = { version = "0.0.1", path = "rs/moq-transcode", default-features = false } # Standalone crate (moq-dev/vaapi); vendored from cros-libva + cros-codecs. moq-vaapi = "0.0.2" -# default-features off (the `nvenc` / `vaapi` hardware encoders) so each consumer -# opts in: binaries (libmoq, moq-boy, moq-cli) enable them, but a self-compiler can -# leave them off to drop the CUDA / libva deps. moq-video itself still defaults them on. -# Hardware codecs are also opt-in here, mirroring moq-video (see above). -moq-transcode = { version = "0.0.1", path = "rs/moq-transcode", default-features = false } moq-video = { version = "0.0.6", path = "rs/moq-video", default-features = false } qmux = { version = "0.3", default-features = false } serde = { version = "1", features = ["derive"] } diff --git a/rs/moq-transcode/src/feed.rs b/rs/moq-transcode/src/feed.rs index 20c153373..74fd0aa55 100644 --- a/rs/moq-transcode/src/feed.rs +++ b/rs/moq-transcode/src/feed.rs @@ -86,6 +86,13 @@ impl Feed { pub(crate) fn listen(&self) -> Listener { let mut state = self.inner.state.lock().unwrap(); state.listeners += 1; + // A finished session may not have cleared itself yet (its final lock + // races this one); subscribing to it would only ever yield Closed, so + // treat it as idle and start fresh. + if state.task.as_ref().is_some_and(|task| task.is_finished()) { + state.sender = None; + state.task = None; + } if state.sender.is_none() { let (sender, _) = broadcast::channel(CAPACITY); state.task = Some(tokio::spawn(run(self.inner.clone(), sender.clone()))); diff --git a/rs/moq-video/src/frame.rs b/rs/moq-video/src/frame.rs index 969b3f93a..32cdeb5eb 100644 --- a/rs/moq-video/src/frame.rs +++ b/rs/moq-video/src/frame.rs @@ -227,15 +227,30 @@ impl I420 { /// convolution: Y at full size, U/V at quarter size. The CPU half of /// [`decode::Frame::resize`](crate::decode::Frame::resize). pub(crate) fn resize(&self, width: u32, height: u32) -> Result { + use std::cell::RefCell; + use fast_image_resize::images::{Image, ImageRef}; use fast_image_resize::{FilterType, PixelType, ResizeAlg, ResizeOptions, Resizer}; - let mut resizer = Resizer::new(); + // The resizer caches its convolution state; recreating it per frame on a + // live path would throw that away, so keep one per thread (decode/encode + // loops are single-threaded). + thread_local! { + static RESIZER: RefCell = RefCell::new(Resizer::new()); + } + // Bilinear convolution: proper filter support at any downscale factor, // the cheapest option that doesn't alias. let options = ResizeOptions::new().resize_alg(ResizeAlg::Convolution(FilterType::Bilinear)); - let mut plane = |src: &[u8], sw: u32, sh: u32, dst: &mut [u8], dw: u32, dh: u32| -> Result<(), Error> { + let plane = |resizer: &mut Resizer, + src: &[u8], + sw: u32, + sh: u32, + dst: &mut [u8], + dw: u32, + dh: u32| + -> Result<(), Error> { let src = ImageRef::new(sw, sh, src, PixelType::U8) .map_err(|e| Error::Codec(anyhow::anyhow!("resize source: {e}")))?; let mut dst = Image::from_slice_u8(dw, dh, dst, PixelType::U8) @@ -250,11 +265,13 @@ impl I420 { let (y_dst, chroma) = data.split_at_mut(luma); let (u_dst, v_dst) = chroma.split_at_mut(luma / 4); - plane(self.y(), self.width, self.height, y_dst, width, height)?; - let (sw2, sh2) = (self.width / 2, self.height / 2); - let (dw2, dh2) = (width / 2, height / 2); - plane(self.u(), sw2, sh2, u_dst, dw2, dh2)?; - plane(self.v(), sw2, sh2, v_dst, dw2, dh2)?; + RESIZER.with_borrow_mut(|resizer| { + plane(resizer, self.y(), self.width, self.height, y_dst, width, height)?; + let (sw2, sh2) = (self.width / 2, self.height / 2); + let (dw2, dh2) = (width / 2, height / 2); + plane(resizer, self.u(), sw2, sh2, u_dst, dw2, dh2)?; + plane(resizer, self.v(), sw2, sh2, v_dst, dw2, dh2) + })?; Ok(Self { width, height, data }) } From 4f4b94ec63d9663298e014d99f249f9bc43be6b1 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Sat, 11 Jul 2026 09:49:08 -0700 Subject: [PATCH 3/3] fix(moq-video,moq-transcode): adapt to the moq_net::Timestamp decode refactor Rebase fallout from #2146: decode::Frame::resize preserves `timestamp` (the field `timestamp_us` no longer exists), and the shared feed passes container timestamps through as-is instead of converting to microseconds. Co-Authored-By: Claude Fable 5 --- rs/moq-transcode/src/feed.rs | 6 +----- rs/moq-transcode/src/rung.rs | 2 +- rs/moq-video/src/decode/mod.rs | 2 +- 3 files changed, 3 insertions(+), 7 deletions(-) diff --git a/rs/moq-transcode/src/feed.rs b/rs/moq-transcode/src/feed.rs index 74fd0aa55..2b2724c39 100644 --- a/rs/moq-transcode/src/feed.rs +++ b/rs/moq-transcode/src/feed.rs @@ -182,11 +182,7 @@ async fn decode(inner: &Inner, sender: &broadcast::Sender) -> Result<(), E let mut first = true; while let Some(frames) = container.read(&mut group).await? { for frame in frames { - let timestamp: u64 = frame - .timestamp - .as_micros() - .try_into() - .map_err(|_| moq_net::TimeOverflow)?; + let timestamp = frame.timestamp; // The first frame of a group is a keyframe by construction. let keyframe = frame.keyframe || first; first = false; diff --git a/rs/moq-transcode/src/rung.rs b/rs/moq-transcode/src/rung.rs index 64e488d10..332f9cffa 100644 --- a/rs/moq-transcode/src/rung.rs +++ b/rs/moq-transcode/src/rung.rs @@ -174,7 +174,7 @@ async fn live(rung: &Rung, producer: &mut moq_net::track::Producer) -> Result<() let scaled = frame.resize(rung.info.width, rung.info.height)?; encoder.encode(&scaled, keyframe)? }; - let timestamp = frame.timestamp_us; + let timestamp = frame.timestamp; write(output, encoded.into_iter().map(|packet| (timestamp, packet)).collect())?; } Some(Item::End) => { diff --git a/rs/moq-video/src/decode/mod.rs b/rs/moq-video/src/decode/mod.rs index b161d0c7a..6bc3f7866 100644 --- a/rs/moq-video/src/decode/mod.rs +++ b/rs/moq-video/src/decode/mod.rs @@ -81,7 +81,7 @@ impl Frame { }; Ok(Frame { - timestamp_us: self.timestamp_us, + timestamp: self.timestamp, width, height, inner,