diff --git a/quest/m1/gpu-ci.md b/quest/m1/gpu-ci.md index 0311a65090..16df80505f 100644 --- a/quest/m1/gpu-ci.md +++ b/quest/m1/gpu-ci.md @@ -30,15 +30,19 @@ skipping inside the Nix shell. - `just rs nvidia`, a one-line recipe over `sh/rs/nvidia.sh`: symlink only those three libraries (by soname) from `/usr/lib/x86_64-linux-gnu` into a private directory, put that on `LD_LIBRARY_PATH`, and run that selection. Fail when a library is missing - instead of skipping. `just rs vulkan-cuda` (`sh/rs/vulkan-cuda.sh`) puts the - whole host driver directory on the path, which lets host libraries shadow - the Nix ones; fold it into this script and recipe, since its `vulkan_cuda_` - tests are the same kind. Those also need + instead of skipping. `sh/rs/vulkan-cuda.sh`, the NVIDIA branch of + `just rs gpu`, already symlinks those three libraries for its `vulkan_cuda_` + tests; widen it into this script rather than adding a second one. They also need the Vulkan loader to find the host NVIDIA ICD: point it at the ICD manifest and expose the driver libraries it names, or keep them in their own recipe. -- [One external GPU image](/quest/m2/gpu-surface.md) adds `just rs gpu`, - which detects the host's GPU vendors and runs each one's ignored tests. - `just rs nvidia` is its NVIDIA branch, not a second detector. +- `just rs gpu` (added by #4975) detects the host's GPU vendors and runs + each one's ignored tests. `just rs nvidia` is its NVIDIA branch, not a + second detector. +- The selection includes the `vulkan_cuda_` external-image tests in + `rs/moq-video/src/frame/cuda_test.rs`, which have never run on hardware. + `vulkan_cuda_auto_external_encode` must pass its per-plane pixel checks + against the CPU reference for both renditions, RGBA and BGRA, through + NVENC and CUDA external-image import (the verification #4975 deferred here). - Nightly: a job in `.github/workflows/nightly.yml` runs `nix develop --command just rs nvidia` on the self-hosted runner, a recipe and not a script path, like every other workflow step. A self-hosted runner on a public repository must @@ -56,4 +60,3 @@ Public API: none. Wire: none. ## Related - [Video hardware validation](/quest/m3/video-hardware.md) - hardware paths nothing runs yet -- [One external GPU image](/quest/m2/gpu-surface.md) - the vendor-detecting `just rs gpu` recipe diff --git a/quest/m2/README.md b/quest/m2/README.md index 4219fd054d..3069da7a00 100644 --- a/quest/m2/README.md +++ b/quest/m2/README.md @@ -108,7 +108,6 @@ waiting on an upstream release waits in [m4](/quest/m4/README.md). - [moq-transport cluster peers](/quest/m2/ietf-cluster-peers.md) - an extended cluster draft lets moq-transport relays peer again, carrying the cluster route layer - [MSFTS convergence](/quest/m2/msfts-convergence.md) - the demultiplexed TS lane converges on MSFTS where the two still differ: program tables and the ES payload unit - [VAAPI encode and decode](/quest/m2/video-vaapi.md) - H.265 encode and decode, pre-generated bindings, and pooled resize surfaces, including the moq-dev/vaapi release that carries them -- [One external GPU image for every encoder](/quest/m2/gpu-surface.md) - a vendor-neutral `Surface::Vulkan` that `Kind::Auto` routes to whichever backend imports it on its device; NVENC imports internally - [Vulkan Video encode on AMD](/quest/m2/vulkan-encode.md) - H.264 and H.265 from an external Vulkan image on RADV, hand-rolled on ash, proven on an RX 9070 - [VA-API encodes an external Vulkan image](/quest/m2/vaapi-vulkan-import.md) - an explicit-sync DMA-BUF reaches Intel's encoder, proven on Arrow Lake - [A release carries multi-vendor GPU input](/quest/m2/gpu-release.md) - `release` ships all three so a pinned consumer drops its vendor code diff --git a/quest/m2/gpu-health.md b/quest/m2/gpu-health.md index 7479f071b0..8c83301b6a 100644 --- a/quest/m2/gpu-health.md +++ b/quest/m2/gpu-health.md @@ -9,7 +9,7 @@ utilization. It also reports device health (whether the device is usable, lost, or reset) and the GPU model, which moq.pro's admission keys its per-model limits on. The same API works on NVIDIA, AMD, and Intel, keyed by the device's render node `dev_t`, the identity -[`Surface::Vulkan`](/quest/m2/gpu-surface.md) carries, so a caller writes no +`Surface::Vulkan` carries, so a caller writes no vendor code. A value a driver cannot provide is absent, never guessed. Not here: admission policy, thresholds, frame-lateness tracking, or draining. @@ -18,7 +18,7 @@ Those are the caller's. ## Plan Decided 2026-10-05 in moq.pro's quest audit: vendor-neutral now, not -NVIDIA-scoped. Reason: [GPU surface](/quest/m2/gpu-surface.md), +NVIDIA-scoped. Reason: the external GPU surface (#4975), [Vulkan encode](/quest/m2/vulkan-encode.md), and [VA-API import](/quest/m2/vaapi-vulkan-import.md) make AMD and Intel encode first-class, so an NVML-shaped API would be rewritten once they land. @@ -28,7 +28,7 @@ adopt this instead. Decided 2026-10-05: one device identity everywhere, the render node's `dev_t`, shared with [VA-API import](/quest/m2/vaapi-vulkan-import.md). NVIDIA's device UUID maps to its node through `VK_EXT_physical_device_drm`. -That is why this Requires [GPU surface](/quest/m2/gpu-surface.md). +The external GPU surface (#4975) carries that identity. Guidance, to be settled while building: @@ -59,15 +59,11 @@ Guidance, to be settled while building: Test: a unit test of each source's parsing against recorded fixtures, and ignored hardware tests in `just rs gpu`'s per-vendor branches (the recipe -[GPU surface](/quest/m2/gpu-surface.md) adds) that open an encoder and see +#4975 added) that open an encoder and see the session and memory counts move. Public API: a device capacity and health snapshot in moq-video. Wire: none. -## Required - -- [One external GPU image for every encoder](/quest/m2/gpu-surface.md) - the render node identity this keys by, and the `just rs gpu` recipe its tests run in - ## Related - [VA-API encodes an external Vulkan image](/quest/m2/vaapi-vulkan-import.md) - picks a render node by the same `dev_t` diff --git a/quest/m2/gpu-release.md b/quest/m2/gpu-release.md index be4dab0a24..333b10dad3 100644 --- a/quest/m2/gpu-release.md +++ b/quest/m2/gpu-release.md @@ -23,7 +23,6 @@ maintainer at release time: ## Required -- [One external GPU image for every encoder](/quest/m2/gpu-surface.md) - the neutral surface and auto-selection - [Vulkan Video encode on AMD](/quest/m2/vulkan-encode.md) - the AMD encoder - [VA-API encodes an external Vulkan image](/quest/m2/vaapi-vulkan-import.md) - the Intel proof - [GPU capacity and health](/quest/m2/gpu-health.md) - keyed by the same device identity; moq.pro's admission still needs vendor code without it diff --git a/quest/m2/gpu-surface.md b/quest/m2/gpu-surface.md deleted file mode 100644 index 1a71f8d5a0..0000000000 --- a/quest/m2/gpu-surface.md +++ /dev/null @@ -1,113 +0,0 @@ -# [M] One external GPU image for every encoder - -## Goal - -A producer hands moq-video one vendor-neutral `Surface::Vulkan`: an image it -rendered and exported from any Vulkan device, with its timeline. `Kind::Auto` -opens the first encoder backend that can import it on that device, so the -caller holds no vendor code: no CUDA, no backend name, no vendor ID. NVENC -imports the image into CUDA itself, and its output on NVIDIA is unchanged. -`DmaBuf` gets a public constructor, so a producer that only has a DMA-BUF can -reach VA-API too. - -This is the API half of proving moq-video carries a GPU frame from an external -producer to any vendor's encoder without a CPU round trip. The first producer -is moq.pro's CARLA bridge (a Vulkan plugin exporting `OPAQUE_FD` images and -timelines), which today names `nvenc`, builds `Surface::Cuda` -itself, and drives `vulkan::Importer` and `cuda::Converter`. - -## Plan - -Decided 2026-10-05 in a `/quest-plan` interview (paper trail in the PR that -added this quest): - -- The surface is a neutral external image. Rejected: an encoder-owned input - the producer renders into (it flips the producer's protocol), and offering - both. `Surface::Vulkan` is no longer gated on `nvidia` and describes: - - the memory handle: `OPAQUE_FD`, or `DMA_BUF` with a DRM format modifier; - - the format, size, and allocation size. For `OPAQUE_FD`, the exporter's - memory type index too: a Vulkan import must reuse both (VUID 01742), - and an `OPAQUE_FD` handle cannot be queried for its properties. A - `DMA_BUF` importer picks its memory type from the handle's properties; - - the device and driver UUID, and the render node's `dev_t` (decided - 2026-10-05: the one device identity - [VA-API import](/quest/m2/vaapi-vulkan-import.md) and - [GPU health](/quest/m2/gpu-health.md) key by). A device without - `VK_EXT_physical_device_drm` or a render node carries none, and a - backend that needs it refuses the surface rather than guess; - - the timeline: an `OPAQUE_FD` timeline semaphore handle, the value the - producer signals when the image is ready, and the value the consumer - signals when it is done reading. Both handles belong to the slot and live - as long as it does, as `vulkan::Handles` do today. -- Each backend imports the image itself: CUDA for NVENC, - [Vulkan for RADV](/quest/m2/vulkan-encode.md), and - [VA-API through a DMA-BUF](/quest/m2/vaapi-vulkan-import.md). The slot and - completion recycling in `frame/vulkan.rs` stays: a slot returns to the - producer only after the encoder's GPU work on it has finished. -- Selection is automatic by the surface's device. `Kind::Auto` opens the first - backend that can import the surface on its device, matched by device and - driver UUID. `Kind::Named` stays for overrides and tests. Rejected: the - caller names the backend. -- The bar is no CPU round trip. Extra GPU-side copies are allowed, such as a - conversion into the encoder's own NV12 input. A strict direct RGB encode is - not required. -- Per-rendition scaling stays on the GPU: one render feeds several encoders at - different sizes (moq.pro encodes 426x240 and 1280x720 from one 1280x720 - render). Keep NVIDIA at one conversion per capture, or measure what a second - one costs. -- Docs are rustdoc, inline. No doc.moq.dev page. - -Open, found in review: - -- Where `Kind::Auto` learns the device. `Encoder::new` opens its backend - eagerly from `Config` alone, and `Config::probe` feeds a throwaway encoder a - synthetic I420 frame. So no surface exists at selection time, and on a host - with two GPUs `Auto` would open whichever backend comes first in - `HARDWARE`. Options: - - a `Config` field naming the input device by device and driver UUID - (recommended: `probe` and fail-fast open keep working); - - deferring the open to the first frame, which breaks `probe`'s - advertise-before-the-first-frame contract. - - Either is a public API change, so record it here. The refusal test must - cover `Encoder::new` and `probe` as well as a frame. -- Slot identity. Today a `Slot` exists only through the CUDA - `vulkan::Importer::import`, and `Slot::publish` and `Completion` wait on - CUDA. Lift slot identity and completion out of the CUDA importer so a - producer publishes a slot before any backend has imported it. Each backend - caches its import by that slot identity, never by fd, since a fresh or - dup'd fd cannot identify a slot. -- `DmaBuf::new` takes an `Arc`, a trait kept `pub(crate)` so - backend lifetimes stay private. Decide what an external producer passes - instead (an `OwnedFd` plus a release guard, say), and refuse - `download_i420` on an external buffer as `Surface::Vulkan` already refuses - readback. - -What moves: `vulkan::Importer::new(ordinal, capacity)`, `cuda::Converter`, and the -caller's `Surface::Cuda` construction go behind the NVENC backend for this -path. Keep public only what another consumer still needs (NVDEC output stays -`Surface::Cuda`). This is a breaking change on `main`; see -[the release quest](/quest/m2/gpu-release.md) for how it ships. - -Tests: - -- Unit-test selection without a GPU: a device no backend can import is - refused with an error naming the device, never a CPU fallback, at - `Encoder::new` and `probe` as well as at encode. -- The existing `vulkan_cuda_` hardware tests move onto the new surface, - `#[ignore = "requires ..."]`, following [GPU CI](/quest/m1/gpu-ci.md)'s - convention. -- Add one recipe, `just rs gpu`, that detects the host's GPUs by PCI vendor - and runs the ignored tests for each vendor present, and fails when a - detected GPU's driver is missing instead of skipping. `just rs vulkan-cuda` - folds into its NVIDIA branch, which GPU CI's `just rs nvidia` selection - later becomes. The other quests add their vendor's tests to it. - -Public API: `Surface::Vulkan` changes shape and loses its `nvidia` gate; -`DmaBuf::new` becomes public; the CUDA import types leave the caller's path. -Wire: none. - -## Related - -- [GPU CI](/quest/m1/gpu-ci.md) - the NVIDIA selection and the ignored-test convention -- [Linux OBS GPU input](/quest/m3/obs-linux-gpu.md) - another external producer, which hands over a DMA-BUF diff --git a/quest/m2/vaapi-vulkan-import.md b/quest/m2/vaapi-vulkan-import.md index baeae0a2e5..688c0b1831 100644 --- a/quest/m2/vaapi-vulkan-import.md +++ b/quest/m2/vaapi-vulkan-import.md @@ -2,7 +2,7 @@ ## Goal -An external [`Surface::Vulkan`](/quest/m2/gpu-surface.md) exported as a +An external `Surface::Vulkan` exported as a DMA-BUF with a DRM format modifier, from a producer that synchronizes explicitly through a timeline, encodes through VA-API on Intel (iHD) without a CPU round trip. An ignored hardware test proves it on an Arrow Lake iGPU, and @@ -41,7 +41,7 @@ sync_file import, so an explicit-sync producer races it. resize share it so a DMA-BUF moves between them without a copy. Choosing a node per surface makes `device()` per-node, so keep decode, resize, and encode on one node per DMA-BUF. Decided 2026-10-05: the producer puts the - render node's `dev_t` in the [surface](/quest/m2/gpu-surface.md), read from + render node's `dev_t` in the surface, read from its own `VK_EXT_physical_device_drm`, and VA-API opens that node. It is the one device identity [GPU health](/quest/m2/gpu-health.md) keys by too. Rejected: mapping the surface's UUID to a node on the VA-API side, which @@ -61,10 +61,6 @@ first render node: the second-node check moved here from Public API: none beyond the surface quest's. Wire: none. -## Required - -- [One external GPU image for every encoder](/quest/m2/gpu-surface.md) - the surface VA-API imports - ## Related - [Vulkan Video encode on AMD](/quest/m2/vulkan-encode.md) - the AMD half of the same proof diff --git a/quest/m2/vulkan-encode.md b/quest/m2/vulkan-encode.md index 4c1713fefb..ed52c93e10 100644 --- a/quest/m2/vulkan-encode.md +++ b/quest/m2/vulkan-encode.md @@ -2,7 +2,7 @@ ## Goal -moq-video encodes an external [`Surface::Vulkan`](/quest/m2/gpu-surface.md) +moq-video encodes an external `Surface::Vulkan` with Vulkan Video on AMD (RADV), in H.264 and H.265, scaled per rendition on the GPU, with no CPU round trip. An ignored hardware test proves it on an RX 9070, and `Kind::Auto` picks this backend for a surface on an AMD device @@ -73,10 +73,6 @@ only if it costs nothing beyond cargo, per the rule in Public API: a new encoder backend and feature. Wire: none. -## Required - -- [One external GPU image for every encoder](/quest/m2/gpu-surface.md) - the surface this backend imports - ## Related - [VA-API encodes an external Vulkan image](/quest/m2/vaapi-vulkan-import.md) - the Intel half of the same proof diff --git a/quest/m3/obs-linux-gpu.md b/quest/m3/obs-linux-gpu.md index e9d34e33e2..4633b4fdac 100644 --- a/quest/m3/obs-linux-gpu.md +++ b/quest/m3/obs-linux-gpu.md @@ -23,5 +23,5 @@ the bottleneck. ## Related - [Video hardware validation](/quest/m3/video-hardware.md) - native input and encoder acceptance need hardware evidence -- [VA-API encodes an external Vulkan image](/quest/m2/vaapi-vulkan-import.md) - explicit sync for an external DMA-BUF, and the public `DmaBuf` constructor from [One external GPU image](/quest/m2/gpu-surface.md) +- [VA-API encodes an external Vulkan image](/quest/m2/vaapi-vulkan-import.md) - explicit sync for an external DMA-BUF, and the public `DmaBuf` constructor from #4975 - [VAAPI encode and decode](/quest/m2/video-vaapi.md) - H.265 and checked-in bindings. The DMA-BUF encoder import this quest needs is already on main (moq-vaapi 0.1.0). diff --git a/rs/justfile b/rs/justfile index 953a7c32c2..a896b2cbc4 100644 --- a/rs/justfile +++ b/rs/justfile @@ -234,9 +234,9 @@ media-features: test *args: RUSTC_WORKSPACE_WRAPPER="$(command -v clippy-driver || true)" cargo nextest run --locked --all-targets "$@" -# Exercise Vulkan -> CUDA -> NVENC on Linux/NVIDIA hardware. -vulkan-cuda: - sh/rs/vulkan-cuda.sh +# Exercise the ignored hardware tests for every GPU vendor on this Linux host. +gpu: + sh/rs/gpu.sh # The media tasks run against a recorder rather than a device (see # `rs/moq-cli/src/play/fake.rs`), so this needs no speaker or display. diff --git a/rs/moq-video/Cargo.toml b/rs/moq-video/Cargo.toml index a90bdc1a94..b20ff37afd 100644 --- a/rs/moq-video/Cargo.toml +++ b/rs/moq-video/Cargo.toml @@ -45,7 +45,7 @@ mediacodec = ["dep:ndk"] # zero-copy transcode path. Everything is dlopen'd at runtime, so a build with # this on still links GPU-less and starts driverless, falling back to the next # encoder or decoder. -nvidia = ["dep:cudarc", "dep:moq-nvenc", "dep:libloading"] +nvidia = ["dep:ash", "dep:cudarc", "dep:moq-nvenc", "dep:libloading"] # Vendored, statically linked software H.264 encode/decode. Enabled by default, # but separable from native codecs for hardware-only and minimal builds. openh264 = ["dep:openh264", "dep:openh264-sys2"] @@ -164,6 +164,7 @@ objc2-video-toolbox = "0.3.2" ndk = { version = "0.9", optional = true, default-features = false, features = ["media", "api-level-26"] } [target.'cfg(target_os="linux")'.dependencies] +ash = { workspace = true, optional = true } # xdg-desktop-portal ScreenCast and Camera negotiation for the `pipewire` feature # (pure Rust, zbus over DBus). The ScreenCast portal owns source selection: the # user picks a monitor in a compositor dialog, and we get back a PipeWire fd + diff --git a/rs/moq-video/README.md b/rs/moq-video/README.md index ed82678b8e..bf626825a8 100644 --- a/rs/moq-video/README.md +++ b/rs/moq-video/README.md @@ -144,43 +144,33 @@ libvpx comes from the build host through pkg-config (set `VPX_STATIC=1` to link the archive, as the Nix dev shell does). `render` is also opt-in so codec-only consumers do not compile wgpu. -### Vulkan producers on NVIDIA - -`frame::vulkan::Importer` accepts dedicated optimal-tiling -`VK_FORMAT_R8G8B8A8_UNORM` (`Image::rgba8`) or `VK_FORMAT_B8G8R8A8_UNORM` -(`Image::bgra8`) images exported as opaque memory FDs. The producer -also exports a timeline semaphore and supplies the Vulkan physical-device UUID; -imports with another CUDA device, format, layout, allocation shape, or sync -mechanism are refused. Vulkan signals `Timeline::ready` after writes and the -transition to `VK_IMAGE_LAYOUT_GENERAL`. CUDA waits on that value and signals -`Timeline::complete` after all readers queued on the frame stream. - -An imported `Slot` owns the producer's `T`. `Slot::publish` consumes it and -`Completion::wait` returns the same slot only after CUDA completion, so a pool -cannot overwrite an in-flight image. Importer capacity bounds retained slots, -and a dedicated worker drains completion after capture stops or a receiver is -cancelled without blocking the producer thread. If the original application -image is not exportable, copy it on Vulkan into a dedicated exportable slot; -that is one GPU image copy, not zero-copy. There is no CPU mapping, download, or -staging fallback for `Surface::Vulkan`. - -`frame::cuda::Converter` turns a published Vulkan frame into the NV12 -`Surface::Cuda` that NVENC encodes in place. It runs on the GPU in one declared -color space (matrix and range), averages 4:2:0 chroma per 2x2 block, applies no -transfer function, and draws every buffer from a pool sized at construction. -`Converter::reserve` holds one as a `cuda::Slot`; `Slot::convert` fills it with -the captured frame and `Slot::resize` with a smaller rendition. One captured -frame feeding HD and SD therefore holds a fixed number of buffers, and a -producer that outruns its encoder gets `None` from `reserve`, its cue to drop -the frame, rather than unbounded device memory. A slot dropped unfilled, or -consumed by a failed conversion, returns its buffer, so only a real failure is -an error. Open the encoder with `encode::Kind::Named("nvenc")` and the same -`encode::Config::color`: `Kind::Auto` could fall back to a software encoder that -reads the frame back, and the portable `Surface::resize` downloads when the GPU -scaler fails. Everything under `frame::cuda` and `frame::vulkan` runs on the -device or returns an error. - -Run `just rs vulkan-cuda` for the opt-in native hardware exercise. It creates a +### External Vulkan producers + +External Linux Vulkan producers construct `frame::vulkan::Slot` from exported +memory and timeline FDs, an `Image` describing the format, allocation, and +memory handle type, and a guard retaining the producer's allocation. Device +and driver UUIDs identify the exporter; opaque memory also carries its original +memory type index, while DMA-BUF memory carries its DRM modifier and explicit +plane offsets and row pitches, including producer padding. + +Set `encode::Config::input` to the image's device before opening or probing the +encoder. `Kind::Auto` selects an importing backend on that device and refuses +unsupported devices without a CPU fallback. Feed each published image through +`Surface::Vulkan`; NVENC privately imports it into CUDA, shares one full-size +NV12 conversion per capture and color space, and scales each rendition on the +GPU. Pin `Config::color` when renditions cross the SD/HD color-inference boundary. + +`Slot::publish` consumes the producer slot and `Completion::wait` returns it +only after the last reader finishes its GPU work. Publish only what you will +encode: an unconsumed or failed frame fails completion and loses its slot for +good, since nothing signalled its timeline, so drop excess captures before +publishing. Non-exportable application images need a GPU copy into an exportable +slot. External Vulkan images have no CPU mapping or download fallback. + +External DMA-BUF producers use `DmaBuf::new` with an owned FD, `DmaBufLayout`, +and a release guard; their buffers also refuse CPU download. + +Run `just rs gpu` for the opt-in native hardware exercise. It creates a Vulkan image independently of Unreal, imports it once into CUDA, checks repeated slot reuse and held-reader ordering, and tears down through cancellation; a second test converts RGBA and BGRA uploads to NV12, scales them, fills the pool, diff --git a/rs/moq-video/src/capture/pipewire.rs b/rs/moq-video/src/capture/pipewire.rs index 2a7a9f975c..5dccca425c 100644 --- a/rs/moq-video/src/capture/pipewire.rs +++ b/rs/moq-video/src/capture/pipewire.rs @@ -41,7 +41,7 @@ use super::channel::FrameChannel; use super::mode::Request; use super::pump::Geometry; use super::{Config, Stream}; -use crate::frame::{DmaBuf, DmaBufFrame, DmaBufPlane, DrmFormat, I420, Surface, wait_dma_buf_readable}; +use crate::frame::{DmaBuf, DmaBufFrame, DmaBufLayout, DmaBufPlane, DrmFormat, I420, Surface, wait_dma_buf_readable}; use crate::{Color, Error, Size}; const DEFAULT_FRAMERATE: u32 = 30; @@ -1117,7 +1117,16 @@ fn run_loop(args: CaptureLoop) -> Result<(), Error> { modifier, color, }); - match DmaBuf::new(format, modifier, layout.width, layout.height, planes, color, inner) { + match DmaBuf::adopt( + DmaBufLayout { + format, + modifier, + size: Size::new(layout.width, layout.height), + planes, + color, + }, + inner, + ) { Ok(frame) => { chan.push(Surface::DmaBuf(frame.clone())); // Only the pacing timer reads `last`. A camera must not @@ -2087,13 +2096,14 @@ mod tests { modifier: 0, color: Some(Color::Bt709Full), }); - let frame = DmaBuf::new( - DrmFormat::NV12, - 0, - 2, - 2, - vec![DmaBufPlane::new(0, 2), DmaBufPlane::new(4, 2)], - Some(Color::Bt709Full), + let frame = DmaBuf::adopt( + DmaBufLayout { + format: DrmFormat::NV12, + modifier: 0, + size: Size::new(2, 2), + planes: vec![DmaBufPlane::new(0, 2), DmaBufPlane::new(4, 2)], + color: Some(Color::Bt709Full), + }, inner, ) .expect("DMA-BUF"); diff --git a/rs/moq-video/src/encode/backend/mod.rs b/rs/moq-video/src/encode/backend/mod.rs index f6c4d198fe..ca0df34862 100644 --- a/rs/moq-video/src/encode/backend/mod.rs +++ b/rs/moq-video/src/encode/backend/mod.rs @@ -172,6 +172,8 @@ struct Candidate { name: &'static str, codecs: &'static [Codec], open: fn(&Config) -> Result, Error>, + #[cfg(target_os = "linux")] + vulkan: bool, } /// Hardware backends, in priority order. Platform-gated so only the ones that @@ -182,30 +184,40 @@ const HARDWARE: &[Candidate] = &[ name: videotoolbox::NAME, codecs: &[Codec::H264, Codec::H265], open: videotoolbox::VideoToolbox::open, + #[cfg(target_os = "linux")] + vulkan: false, }, #[cfg(target_os = "windows")] Candidate { name: mediafoundation::NAME, codecs: &[Codec::H264, Codec::H265], open: mediafoundation::MediaFoundation::open, + #[cfg(target_os = "linux")] + vulkan: false, }, #[cfg(all(target_os = "android", feature = "mediacodec"))] Candidate { name: mediacodec::NAME, codecs: &[Codec::H264, Codec::H265], open: mediacodec::MediaCodec::open, + #[cfg(target_os = "linux")] + vulkan: false, }, #[cfg(all(target_os = "linux", feature = "nvidia"))] Candidate { name: nvenc::NAME, codecs: &[Codec::H264, Codec::H265], open: nvenc::Nvenc::open, + #[cfg(target_os = "linux")] + vulkan: true, }, #[cfg(all(target_os = "linux", feature = "vaapi"))] Candidate { name: vaapi::NAME, codecs: &[Codec::H264], open: vaapi::Vaapi::open, + #[cfg(target_os = "linux")] + vulkan: false, }, // Last of the Linux hardware encoders: the SoC blocks it drives are the only // hardware on a board that has neither an NVIDIA GPU nor a VAAPI stack, so it @@ -215,6 +227,8 @@ const HARDWARE: &[Candidate] = &[ name: v4l2::NAME, codecs: &[Codec::H264], open: v4l2::V4l2::open, + #[cfg(target_os = "linux")] + vulkan: false, }, ]; @@ -227,6 +241,8 @@ const SOFTWARE: &[Candidate] = &[ name: openh264::NAME, codecs: &[Codec::H264], open: openh264::Openh264::open, + #[cfg(target_os = "linux")] + vulkan: false, }, ]; @@ -239,11 +255,15 @@ const NAMED_ONLY: &[Candidate] = &[ name: probe::NAME, codecs: &[Codec::H264], open: probe::Probe::open, + #[cfg(target_os = "linux")] + vulkan: false, }, Candidate { name: probe::NO_CUT, codecs: &[Codec::H264], open: probe::Probe::open_no_cut, + #[cfg(target_os = "linux")] + vulkan: false, }, ]; @@ -313,6 +333,13 @@ fn select(attempts: Vec, config: &Config) -> Result, E for attempt in attempts { let name = attempt.candidate.name; + #[cfg(target_os = "linux")] + if let Some(device) = config.input + && (!attempt.hardware || !attempt.candidate.vulkan) + { + tried.push(format!("{name}: cannot import external Vulkan {device}")); + continue; + } match (attempt.candidate.open)(config) { Ok(backend) => { // `Auto` returning a software encoder is otherwise invisible except for @@ -348,6 +375,13 @@ fn select(attempts: Vec, config: &Config) -> Result, E // or a backend that does not take this codec. Reporting it as "no usable // encoder (tried: )" tells the caller nothing, and naming what is here is // most of the answer. + #[cfg(target_os = "linux")] + if let Some(device) = config.input { + return Err(Error::NoEncoder(format!( + "no backend can import external Vulkan {device} (tried: {})", + tried.join(", ") + ))); + } if tried.is_empty() { let available = available_names(config.codec); return match &config.kind { @@ -506,6 +540,8 @@ mod tests { name: "stub", codecs: &[Codec::H264], open: Stub::open, + #[cfg(target_os = "linux")] + vulkan: false, }; /// Compiled in but refusing at runtime, the way NVENC does on a host whose @@ -514,6 +550,8 @@ mod tests { name: "driverless", codecs: &[Codec::H264], open: |_| Err(Error::Codec(anyhow::anyhow!("driver libraries not found"))), + #[cfg(target_os = "linux")] + vulkan: false, }; fn config() -> Config { diff --git a/rs/moq-video/src/encode/backend/nvenc.rs b/rs/moq-video/src/encode/backend/nvenc.rs index a77d03f97a..ea1fe1d0cf 100644 --- a/rs/moq-video/src/encode/backend/nvenc.rs +++ b/rs/moq-video/src/encode/backend/nvenc.rs @@ -32,6 +32,7 @@ //! rung of a rendition ladder can be too small for NVENC even though the GPU //! conversion and resize handle it. +use std::num::NonZeroUsize; use std::sync::Arc; use bytes::Bytes; @@ -45,7 +46,7 @@ use moq_nvenc::{Encoder, EncoderInitParams, Session}; use super::super::encoder::{Applied, Codec, Config, Gop, Preset}; use super::{Backend, Encoded, keyframe_annexb}; -use crate::frame::{Surface, interleave_uv}; +use crate::frame::{Surface, cuda, interleave_uv, vulkan}; use crate::{Color, Error, Frame}; pub(crate) const NAME: &str = "nvenc"; @@ -81,6 +82,8 @@ pub(crate) struct Nvenc { _cuda: Arc, timestamp: u64, codec: Codec, + size: crate::Size, + converter: Option, } impl Nvenc { @@ -98,7 +101,7 @@ impl Nvenc { // cudarc 0.19's DriverError is Debug-only (no Display), so format with `{e:?}`. let codec_guid = codec_guid(config.codec); - let cuda = CudaContext::new(0).map_err(|e| Error::Codec(anyhow::anyhow!("CUDA init: {e:?}")))?; + let cuda = input_context(config.input)?; let encoder = Encoder::initialize_with_cuda(cuda.clone()) .map_err(|e| Error::Codec(anyhow::anyhow!("NVENC init: {e}")))?; @@ -224,6 +227,11 @@ impl Nvenc { config.preset, format!("{label}, low-latency tuning, no B-frames, CBR, 1-frame VBV"), ), + converter: config + .input + .map(|_| cuda::Converter::new(cuda.ordinal(), config.resolved_color(), NonZeroUsize::new(8).unwrap())) + .transpose()?, + size: config.size(), _cuda: cuda, timestamp: 0, codec: config.codec, @@ -233,6 +241,38 @@ impl Nvenc { impl Backend for Nvenc { fn encode(&mut self, frame: &Frame, cut: bool) -> Result, Error> { + // The published image shares its full-size conversion across renditions. + // Smaller encoders resize that NV12 buffer, so capture converts only once + // for each declared color space and never downloads its pixels. + let prepared = if let Surface::Vulkan(image) = &frame.surface { + let converter = self.converter.as_ref().ok_or_else(|| { + Error::Unsupported(format!( + "NVENC was not opened for external Vulkan {}", + image.image().device + )) + })?; + let converted = image.converted(converter.color(), || { + // Read before reserving: an exhausted pool must still signal the + // producer's timeline, or the unread frame loses its slot. + converter.import(image)?; + converter + .reserve() + .ok_or_else(|| Error::Unsupported("CUDA conversion pool exhausted".into()))? + .convert(image) + })?; + let sized = if converted.size() == self.size { + converted + } else { + converter + .reserve() + .ok_or_else(|| Error::Unsupported("CUDA resize pool exhausted".into()))? + .resize(&converted, self.size)? + }; + Some(Frame::new(Surface::Cuda(sized), frame.timestamp)) + } else { + None + }; + let frame = prepared.as_ref().unwrap_or(frame); let output = self .session .create_output_bitstream() @@ -362,6 +402,59 @@ fn drain_output(submission: moq_nvenc::Submission) -> Result, Erro Ok(data) } +/// Match the declared Vulkan device before allocating an NVENC session. +fn input_context(input: Option) -> Result, Error> { + let Some(device) = input else { + return CudaContext::new(0).map_err(|e| Error::Codec(anyhow::anyhow!("CUDA init: {e:?}"))); + }; + matching_driver(device)?; + let count = CudaContext::device_count().map_err(|e| Error::Codec(anyhow::anyhow!("CUDA enumerate: {e:?}")))?; + for ordinal in 0..count { + let physical = cudarc::driver::result::device::get(ordinal) + .map_err(|e| Error::Codec(anyhow::anyhow!("CUDA device {ordinal}: {e:?}")))?; + let uuid = cudarc::driver::result::device::get_uuid(physical) + .map_err(|e| Error::Codec(anyhow::anyhow!("CUDA UUID: {e:?}")))?; + if uuid.bytes.map(|byte| u8::from_ne_bytes(byte.to_ne_bytes())) == device.device_uuid { + return CudaContext::new(ordinal as usize) + .map_err(|e| Error::Codec(anyhow::anyhow!("CUDA input device {ordinal}: {e:?}"))); + } + } + Err(Error::Unsupported(format!( + "no CUDA device matches external Vulkan {device}" + ))) +} + +/// Opaque memory belongs to the exporting Vulkan driver as well as its device. +fn matching_driver(device: vulkan::Device) -> Result<(), Error> { + use ash::vk; + // SAFETY: load the system Vulkan implementation and retain it through the query. + let entry = unsafe { ash::Entry::load() } + .map_err(|e| Error::Unsupported(format!("Vulkan unavailable for {device}: {e}")))?; + let app = vk::ApplicationInfo::default().api_version(vk::API_VERSION_1_1); + // SAFETY: the create info has no borrowed pointers beyond this call. + let instance = unsafe { entry.create_instance(&vk::InstanceCreateInfo::default().application_info(&app), None) } + .map_err(|e| Error::Unsupported(format!("Vulkan instance for {device}: {e}")))?; + let result = (|| { + // SAFETY: the instance and queried physical devices stay alive throughout. + for physical in unsafe { instance.enumerate_physical_devices() } + .map_err(|e| Error::Unsupported(format!("Vulkan devices: {e}")))? + { + let mut id = vk::PhysicalDeviceIDProperties::default(); + let mut properties = vk::PhysicalDeviceProperties2::default().push_next(&mut id); + unsafe { instance.get_physical_device_properties2(physical, &mut properties) }; + if id.device_uuid == device.device_uuid && id.driver_uuid == device.driver_uuid { + return Ok(()); + } + } + Err(Error::Unsupported(format!( + "no Vulkan driver matches external {device}" + ))) + })(); + // SAFETY: no children or outstanding device work were created by this query. + unsafe { instance.destroy_instance(None) }; + result +} + /// Whether cudarc's CUDA driver library can be opened without panicking. fn cuda_driver_present() -> bool { // libcuda is the CUDA driver API; matches cudarc's "cuda" search. diff --git a/rs/moq-video/src/encode/encoder.rs b/rs/moq-video/src/encode/encoder.rs index b4024ca49e..2e73331780 100644 --- a/rs/moq-video/src/encode/encoder.rs +++ b/rs/moq-video/src/encode/encoder.rs @@ -158,7 +158,7 @@ impl Applied { } /// Encoder configuration. `width` / `height` / `framerate` are the encoded -/// output; input frames must already be at this resolution. +/// output; external Vulkan frames are scaled by the backend on their GPU. /// /// `#[non_exhaustive]`: build via [`Config::new`] and set the optional fields, /// so future knobs don't break callers. @@ -188,6 +188,10 @@ pub struct Config { /// Set it only when feeding frames the crate did not convert and whose space /// you know from elsewhere. pub color: Option, + /// External Vulkan input device, required before opening a GPU-only encoder. + /// `None` retains selection for CPU and native decoder/capture surfaces. + #[cfg(target_os = "linux")] + pub input: Option, } impl Config { @@ -204,6 +208,8 @@ impl Config { kind: Kind::Auto, preset: Preset::default(), color: None, + #[cfg(target_os = "linux")] + input: None, } } @@ -323,6 +329,8 @@ pub struct Encoder { /// rather than applied immediately because the caller decides a group /// boundary before it has the frame that opens it. pending_cut: bool, + #[cfg(target_os = "linux")] + input: Option, /// Keeps direct use bound to the constructing thread, regardless of backend. _thread_bound: PhantomData>, } @@ -347,6 +355,8 @@ impl Encoder { applied, color: config.resolved_color(), pending_cut: false, + #[cfg(target_os = "linux")] + input: config.input, _thread_bound: PhantomData, }) } @@ -362,7 +372,7 @@ impl Encoder { &self.applied } - /// The resolution this encoder emits, which every frame fed to it must match. + /// The resolution this encoder emits; external Vulkan inputs may be larger. pub fn size(&self) -> Size { self.size } @@ -450,15 +460,31 @@ impl Encoder { /// A GPU surface feeds a hardware encoder on the same device directly /// (NVDEC -> NVENC never leaves the GPU, a `CVPixelBuffer` goes straight to /// VideoToolbox); anything else falls back to a CPU I420 upload. The frame must - /// already be at the encoder's resolution: scale it with + /// already be at the encoder's resolution, except external Vulkan inputs + /// which the backend scales on the GPU. Scale other surfaces with /// [`Frame::resize`](crate::Frame::resize), which /// [`decode::Config::scale_hint`](crate::decode::Config::scale_hint) lets a /// hardware decoder make a no-op. pub fn encode(&mut self, frame: &Frame) -> Result, Error> { // A transposed frame is why this compares the shape rather than a byte // count: 240x320 and 320x240 hold the same number of bytes. + #[cfg(target_os = "linux")] + let external = if let crate::Surface::Vulkan(image) = &frame.surface { + let device = image.image().device; + if self.input != Some(device) { + return Err(Error::Unsupported(format!( + "external Vulkan {device} does not match encoder input {:?}", + self.input + ))); + } + true + } else { + false + }; + #[cfg(not(target_os = "linux"))] + let external = false; let size = frame.size(); - if size != self.size { + if !external && size != self.size { return Err(Error::Codec(anyhow::anyhow!( "frame {size} does not match encoder {}", self.size @@ -1107,6 +1133,8 @@ mod tests { applied: Applied::default(), color: config.resolved_color(), pending_cut: false, + #[cfg(target_os = "linux")] + input: None, _thread_bound: PhantomData, } } @@ -1562,3 +1590,67 @@ mod tests { } } } + +#[cfg(all(test, target_os = "linux"))] +mod external_tests { + use super::*; + use crate::frame::vulkan::{Device, Format, Handles, Image, Memory, Slot, Timeline}; + use std::os::unix::net::UnixStream; + + fn device() -> Device { + Device { + device_uuid: [0xaa; 16], + driver_uuid: [0xbb; 16], + render_node: None, + } + } + + #[tokio::test] + async fn unknown_external_device_is_refused_at_open_and_probe() { + let mut config = Config::new(320, 240, Rate::new(30, 1).unwrap()); + config.input = Some(device()); + let error = match Encoder::new(&config) { + Ok(_) => panic!("opened an unrelated device"), + Err(error) => error, + }; + assert!(error.to_string().contains(&device().to_string()), "{error}"); + let error = config.probe().await.unwrap_err(); + assert!(error.to_string().contains(&device().to_string()), "{error}"); + config.kind = Kind::Software; + assert!(matches!(Encoder::new(&config), Err(Error::NoEncoder(_)))); + } + + #[test] + fn refuses_external_frame_on_an_encoder_without_that_device() { + let config = { + let mut config = Config::new(320, 240, Rate::new(30, 1).unwrap()); + config.kind = Kind::Named("probe".into()); + config + }; + let mut encoder = Encoder::new(&config).unwrap(); + let (memory, timeline) = UnixStream::pair().unwrap(); + let slot = Slot::new( + Handles { + memory: memory.into(), + timeline: timeline.into(), + }, + Image { + device: device(), + memory: Memory::OpaqueFd { memory_type: 0 }, + size: config.size(), + allocation_size: 4096, + format: Format::Rgba8, + }, + (), + ) + .unwrap(); + let (image, _) = slot.publish(Timeline::new(1, 2).unwrap()).unwrap(); + let frame = Frame::new( + crate::Surface::Vulkan(image), + moq_net::Timestamp::from_micros(0).unwrap(), + ); + let error = encoder.encode(&frame).unwrap_err(); + assert!(error.to_string().contains(&device().to_string()), "{error}"); + assert!(frame.surface.to_i420().is_err()); + } +} diff --git a/rs/moq-video/src/frame.rs b/rs/moq-video/src/frame.rs index 2f82cdffaa..18c4ecc532 100644 --- a/rs/moq-video/src/frame.rs +++ b/rs/moq-video/src/frame.rs @@ -16,10 +16,9 @@ //! - `Surface::DmaBuf` is a Linux DRM allocation, produced by PipeWire capture. //! The Vulkan renderer imports supported packed formats directly, while CPU //! consumers map linear allocations only. -//! - `Surface::Vulkan` is a Linux/NVIDIA Vulkan RGBA or BGRA image imported into -//! CUDA with an explicit timeline semaphore. It deliberately has no CPU -//! download fallback: a `cuda::Converter` turns it into a `Surface::Cuda` on -//! the GPU, and consumers return its producer slot after CUDA completion. +//! - `Surface::Vulkan` is an external Linux Vulkan image with an explicit +//! timeline semaphore. Encoders import and convert it on its device, with +//! no CPU download fallback, then return its producer slot after GPU completion. //! - `Surface::HardwareBuffer` is an Android `AHardwareBuffer`, produced by the //! MediaCodec decoder rendering into an `ImageReader`. A GPU consumer imports //! it as a GL or Vulkan image; `into_i420` reads the planes back instead. @@ -28,7 +27,7 @@ //! //! A backend that consumes a GPU surface takes the frame as-is; a CPU encoder //! asks for I420 via [`Surface::into_i420`], which downloads GPU frames that -//! permit readback. Vulkan/CUDA surfaces intentionally refuse that fallback. +//! permit readback. External Vulkan surfaces intentionally refuse that fallback. use std::borrow::Cow; @@ -131,12 +130,8 @@ pub struct DmaBufPlane { #[cfg(all(target_os = "linux", feature = "dmabuf"))] impl DmaBufPlane { - // The producers that build a real one: PipeWire capture and, since it can - // hand its decode surfaces out, the VA-API decoder. Plus the importer's own - // tests, which build them without a producer to check how a layout is split - // up, and so need `render`. - #[cfg(any(feature = "pipewire", feature = "vaapi", all(feature = "render", test)))] - pub(crate) const fn new(offset: u32, stride: u32) -> Self { + /// Describe a plane's byte offset and row stride. + pub const fn new(offset: u32, stride: u32) -> Self { Self { offset, stride } } @@ -282,27 +277,33 @@ impl std::fmt::Debug for DmaBuf { #[cfg(all(target_os = "linux", feature = "dmabuf"))] impl DmaBuf { - // The two producers: PipeWire capture, and the VA-API decoder describing a - // picture it is handing out rather than downloading. - #[cfg(any(feature = "pipewire", feature = "vaapi"))] - pub(crate) fn new( - format: DrmFormat, - modifier: u64, - width: u32, - height: u32, - planes: Vec, - color: Option, - inner: Arc, - ) -> Result { - Size::new(width, height).validate("DMA-BUF")?; - if planes.is_empty() { - return Err(Error::Codec(anyhow::anyhow!("DMA-BUF has no planes"))); + /// Retain an external DMA-BUF and its producer's release guard. + /// + /// Clones share the descriptor and guard; exports duplicate the FD as needed. + /// External buffers refuse CPU download, including linear allocations. + pub fn new(fd: OwnedFd, layout: DmaBufLayout, owner: T) -> Result { + Self::adopt(layout, Arc::new(ExternalDmaBuf { fd, _owner: owner })) + } + + pub(crate) fn adopt(layout: DmaBufLayout, inner: Arc) -> Result { + let DmaBufLayout { + format, + modifier, + size, + planes, + color, + } = layout; + size.validate("DMA-BUF")?; + if planes.is_empty() || planes.len() > 4 || planes.iter().any(|plane| plane.stride == 0) { + return Err(Error::Unsupported( + "DMA-BUF requires one to four planes with non-zero strides".into(), + )); } Ok(Self { format, modifier, - width, - height, + width: size.width, + height: size.height, planes, color, inner, @@ -359,6 +360,55 @@ impl DmaBuf { } } +/// The format, tiling, and plane layout of an external DMA-BUF. +#[cfg(all(target_os = "linux", feature = "dmabuf"))] +#[derive(Clone, Debug)] +#[non_exhaustive] +pub struct DmaBufLayout { + /// DRM fourcc describing the pixel format. + pub format: DrmFormat, + /// DRM modifier describing memory tiling; defaults to linear. + pub modifier: u64, + /// Visible image dimensions. + pub size: Size, + /// Plane offsets and row strides, in format order. + pub planes: Vec, + /// Color space of YUV samples, when the producer knows it. + pub color: Option, +} + +#[cfg(all(target_os = "linux", feature = "dmabuf"))] +impl DmaBufLayout { + /// Describe linear planes and set optional tiling and color fields afterward. + pub fn new(format: DrmFormat, size: Size, planes: Vec) -> Self { + Self { + format, + modifier: 0, + size, + planes, + color: None, + } + } +} + +#[cfg(all(target_os = "linux", feature = "dmabuf"))] +struct ExternalDmaBuf { + fd: OwnedFd, + _owner: T, +} + +#[cfg(all(target_os = "linux", feature = "dmabuf"))] +impl DmaBufFrame for ExternalDmaBuf { + fn export(&self) -> std::io::Result { + self.fd.try_clone() + } + fn download_i420(&self) -> Result { + Err(Error::Unsupported( + "external DMA-BUF surfaces have no CPU download fallback".into(), + )) + } +} + /// The producer-owned half of a DMA-BUF surface. /// /// Kept private to the crate so backend lifetimes and download mechanisms do @@ -397,12 +447,11 @@ pub enum Surface { #[cfg(target_os = "windows")] Texture(d3d11::Texture), /// Zero-copy GPU buffer (Linux CUDA NV12). Produced by the NVDEC decoder or - /// a [`cuda::Converter`], consumed in place by the NVENC encoder. + /// the GPU color converter, consumed in place by the NVENC encoder. #[cfg(all(target_os = "linux", feature = "nvidia"))] Cuda(cuda::Frame), - /// Vulkan RGBA8 / BGRA8 image imported into CUDA with explicit GPU - /// synchronization. A [`cuda::Converter`] turns it into `Cuda` on the GPU. - #[cfg(all(target_os = "linux", feature = "nvidia"))] + /// External Vulkan RGBA8 / BGRA8 image with explicit GPU synchronization. + #[cfg(target_os = "linux")] Vulkan(vulkan::Frame), /// Linux DMA-BUF, exported on access and retained until the last clone drops. #[cfg(all(target_os = "linux", feature = "dmabuf"))] @@ -425,7 +474,7 @@ impl Surface { Surface::Texture(t) => t.width, #[cfg(all(target_os = "linux", feature = "nvidia"))] Surface::Cuda(c) => c.width, - #[cfg(all(target_os = "linux", feature = "nvidia"))] + #[cfg(target_os = "linux")] Surface::Vulkan(v) => v.width(), #[cfg(all(target_os = "linux", feature = "dmabuf"))] Surface::DmaBuf(d) => d.width, @@ -444,7 +493,7 @@ impl Surface { Surface::Texture(t) => t.height, #[cfg(all(target_os = "linux", feature = "nvidia"))] Surface::Cuda(c) => c.height, - #[cfg(all(target_os = "linux", feature = "nvidia"))] + #[cfg(target_os = "linux")] Surface::Vulkan(v) => v.height(), #[cfg(all(target_os = "linux", feature = "dmabuf"))] Surface::DmaBuf(d) => d.height, @@ -487,7 +536,7 @@ impl Surface { /// which is what you usually want since it carries the timestamp across too. /// /// A GPU scaler that a driver refuses falls back to downloading and scaling - /// on the CPU where the surface permits readback. Vulkan/CUDA surfaces fail + /// on the CPU where the surface permits readback. External Vulkan surfaces fail /// instead because their contract forbids CPU pixel access. pub fn resize(&self, size: Size, config: &crate::resize::Config) -> Result { // Counts as a use on builds where every GPU arm is compiled out. @@ -524,10 +573,10 @@ impl Surface { Surface::I420(cuda.download_i420()?.resize(size)?) } }, - #[cfg(all(target_os = "linux", feature = "nvidia"))] + #[cfg(target_os = "linux")] Surface::Vulkan(_) => { return Err(Error::Unsupported( - "Vulkan/CUDA surfaces require a GPU consumer and cannot be resized or downloaded".into(), + "External Vulkan surfaces require a GPU consumer and cannot be resized or downloaded".into(), )); } #[cfg(target_os = "windows")] @@ -569,7 +618,7 @@ impl Surface { /// /// Free for `Surface::I420`; downloads native GPU surfaces that permit /// readback, so it is the universal arm of a `match` on every other - /// platform. A Vulkan/CUDA surface returns [`Error::Unsupported`] because + /// platform. A external Vulkan surface returns [`Error::Unsupported`] because /// its contract deliberately exposes no CPU pixel path. pub fn into_i420(self) -> Result { match self { @@ -582,7 +631,7 @@ impl Surface { /// Convert to owned, tightly packed RGBA8 pixels on the CPU. /// /// Native GPU surfaces that permit readback are downloaded first; CPU I420 is - /// converted directly. Vulkan/CUDA surfaces return [`Error::Unsupported`]. + /// converted directly. External Vulkan surfaces return [`Error::Unsupported`]. /// The conversion honors [`color`](Self::color) and otherwise falls back to /// [`Color::infer`]. pub fn to_rgba(&self, config: &crate::convert::Config) -> Result { @@ -640,7 +689,7 @@ impl Surface { Surface::Texture(texture) => texture.color, #[cfg(all(target_os = "linux", feature = "nvidia"))] Surface::Cuda(c) => c.color(), - #[cfg(all(target_os = "linux", feature = "nvidia"))] + #[cfg(target_os = "linux")] Surface::Vulkan(_) => None, #[cfg(all(target_os = "linux", feature = "dmabuf"))] Surface::DmaBuf(d) => d.color, @@ -653,7 +702,7 @@ impl Surface { /// A CPU I420 view, downloading a GPU frame only if necessary. /// /// Borrowed for `Surface::I420`, owned for a GPU surface that permits - /// readback, and unsupported for Vulkan/CUDA. The borrowing counterpart to + /// readback, and unsupported for external Vulkan. The borrowing counterpart to /// [`into_i420`](Self::into_i420), for a /// caller that cannot give up the surface: a publisher's preview frame is /// shared with every rendition's encoder, so its `Arc` never has a refcount @@ -666,9 +715,9 @@ impl Surface { Surface::Texture(t) => Ok(Cow::Owned(t.download_i420()?)), #[cfg(all(target_os = "linux", feature = "nvidia"))] Surface::Cuda(c) => Ok(Cow::Owned(c.download_i420()?)), - #[cfg(all(target_os = "linux", feature = "nvidia"))] + #[cfg(target_os = "linux")] Surface::Vulkan(_) => Err(Error::Unsupported( - "Vulkan/CUDA surfaces have no CPU mapping or download fallback".into(), + "External Vulkan surfaces have no CPU mapping or download fallback".into(), )), #[cfg(all(target_os = "linux", feature = "dmabuf"))] Surface::DmaBuf(d) => Ok(Cow::Owned(d.inner.download_i420()?)), @@ -1924,10 +1973,14 @@ pub mod apple { } } -#[cfg(all(target_os = "linux", feature = "nvidia"))] +#[cfg(target_os = "linux")] #[path = "frame/vulkan.rs"] pub mod vulkan; +#[cfg(all(target_os = "linux", feature = "nvidia"))] +#[path = "frame/cuda_vulkan.rs"] +mod cuda_vulkan; + #[cfg(all(target_os = "linux", feature = "nvidia"))] #[path = "frame/cuda.rs"] pub mod cuda; @@ -3250,3 +3303,45 @@ mod tests { assert!(mae(gpu.v(), cpu.v()) < 4, "GPU and CPU v disagree"); } } + +#[cfg(all(test, target_os = "linux", feature = "dmabuf"))] +mod external_dmabuf_tests { + use super::*; + use std::sync::atomic::{AtomicUsize, Ordering}; + + #[test] + fn external_buffer_retains_lease_and_refuses_cpu_download() { + struct Guard(Arc); + impl Drop for Guard { + fn drop(&mut self) { + self.0.fetch_add(1, Ordering::SeqCst); + } + } + let released = Arc::new(AtomicUsize::new(0)); + let layout = DmaBufLayout::new( + DrmFormat::NV12, + Size::new(4, 2), + vec![DmaBufPlane::new(0, 4), DmaBufPlane::new(8, 4)], + ); + let buffer = DmaBuf::new( + std::fs::File::open("/dev/zero").unwrap().into(), + layout, + Guard(released.clone()), + ) + .unwrap(); + assert!(Surface::DmaBuf(buffer.clone()).to_i420().is_err()); + let export = buffer.export().unwrap(); + drop(buffer); + assert_eq!(released.load(Ordering::SeqCst), 0); + drop(export); + assert_eq!(released.load(Ordering::SeqCst), 1); + } + + #[test] + fn external_buffer_refuses_empty_or_zero_stride_planes() { + for planes in [vec![], vec![DmaBufPlane::new(0, 0)]] { + let layout = DmaBufLayout::new(DrmFormat::NV12, Size::new(4, 2), planes); + assert!(DmaBuf::new(std::fs::File::open("/dev/zero").unwrap().into(), layout, ()).is_err()); + } + } +} diff --git a/rs/moq-video/src/frame/cuda.rs b/rs/moq-video/src/frame/cuda.rs index 4540246958..54b7cbaa2f 100644 --- a/rs/moq-video/src/frame/cuda.rs +++ b/rs/moq-video/src/frame/cuda.rs @@ -1,27 +1,16 @@ //! Linux CUDA device memory: the NV12 [`Frame`] behind `Surface::Cuda`, which -//! NVDEC or a [`Converter`] produces and NVENC consumes in place. +//! NVDEC or a `Converter` produces and NVENC consumes in place. //! //! # The GPU-only path //! -//! A producer that must keep raw pixels off the CPU (a game engine publishing -//! its render target) stays inside this module and `frame::vulkan`: import the -//! image ([`vulkan::Importer`]), reserve a pool buffer ([`Converter::reserve`]) -//! and convert into it ([`Slot::convert`]), scale into another -//! ([`Slot::resize`]), and encode it with an -//! [`encode::Encoder`](crate::encode::Encoder) opened as -//! `Kind::Named("nvenc")`, which registers a `Surface::Cuda` with NVENC in -//! place. Every operation on that path runs on the device or fails: nothing -//! here downloads, uploads, or converts on the CPU, and a request the device -//! cannot serve (a mismatched device, an odd size) is an error rather than a -//! fallback. A full pool is neither: `reserve` returns `None`, and the producer -//! drops the frame. +//! External producers publish `Surface::Vulkan` with `encode::Config::input` +//! naming their device. NVENC imports and converts those images here, caching +//! one full-size conversion per capture and color space, then scaling each +//! rendition on the GPU. Those images are never downloaded. //! -//! The portable `Surface` methods are a different contract. `Surface::resize` -//! keeps a stream alive by downloading when the GPU scaler fails, and a -//! software encoder reads a `Surface::Cuda` back through `Surface::to_i420`; -//! both exist for the transcode path, whose frames come from NVDEC and may go -//! anywhere. A GPU-only producer never calls them and never opens an encoder -//! with `Kind::Auto`, whose openh264 fallback would read the frame back. +//! Native `Surface::Cuda` frames from NVDEC also encode in place. Their portable +//! `Surface` methods permit CPU readback for software transcoding, while an +//! external Vulkan image refuses that fallback. //! //! NVENC takes the converted frame as a registered `cuMemAlloc` NV12 buffer //! (`NV_ENC_INPUT_RESOURCE_TYPE_CUDADEVICEPTR`), the registration the NVDEC @@ -31,7 +20,7 @@ //! its own conversion, so a kernel that writes NV12 straight from the image is //! the one pass this path needs: no staging copy in either direction. //! -//! What the hardware says (RTX 3070 Ti, driver 595.91.07, `just rs vulkan-cuda`): +//! What the hardware says (RTX 3070 Ti, driver 595.91.07, `just rs gpu`): //! the kernels match a CPU reference exactly, and an `nsys --trace=cuda` capture //! of `tests::vulkan_cuda_three_view_workload` records no `cuMemcpy` in either //! direction. Registering a converted buffer with NVENC costs around 70us and @@ -47,7 +36,7 @@ use std::sync::{Arc, Mutex, OnceLock}; use cudarc::driver::{CudaContext, CudaFunction, LaunchConfig, PushKernelArg, result}; use super::pool::{self, Pool}; -use super::{I420, vulkan}; +use super::{I420, cuda_vulkan, vulkan}; use crate::{Color, Error, Size}; /// The NV12 box-filter resize kernels, vendored as PTX (see nv12_resize.cu) @@ -136,7 +125,7 @@ impl pool::Alloc for Device { } } -/// A frame's allocation: a buffer leased from a [`Converter`]'s pool, handed +/// A frame's allocation: a buffer leased from a `Converter`'s pool, handed /// back for reuse on drop, or a plain one freed on drop. enum Buffer { Pooled(pool::Lease), @@ -154,7 +143,7 @@ impl std::ops::Deref for Buffer { } } -/// A GPU NV12 frame in CUDA device memory: NVDEC's or a [`Converter`]'s +/// A GPU NV12 frame in CUDA device memory: NVDEC's or a `Converter`'s /// output and NVENC's zero-copy input. One buffer holds both planes at a /// shared row `pitch`: `height` luma rows, then `height / 2` interleaved-UV /// rows. Cloning bumps refcounts (no pixel copy), which keeps decode -> encode @@ -241,7 +230,7 @@ impl Frame { Size::new(self.width, self.height) } - /// The color space of the samples: the one a [`Converter`] wrote, or `None` + /// The color space of the samples: the one a `Converter` wrote, or `None` /// for a decoded frame that merely passed through. pub fn color(&self) -> Option { self.color @@ -300,7 +289,7 @@ impl Frame { /// [`Surface::resize`](crate::Surface::resize), which falls back to the CPU /// when this fails. A pooled frame draws the copy from its own pool, so a /// full pool is one such failure; a GPU-only producer scales through - /// [`Slot::resize`] instead. + /// the encoder's pooled rendition scaler instead. pub(crate) fn resize(&self, size: Size) -> Result { size.validate("resize to")?; let dst = match &*self.buf { @@ -405,7 +394,7 @@ const CONVERT_PTX: &str = include_str!("rgba_to_nv12.ptx"); /// bitstream, and the frames it returns report that space through /// [`Frame::color`] so the two cannot silently disagree. Convert once at the /// captured size and resize the result for smaller renditions, each into a -/// [`Slot`] from [`reserve`](Self::reserve); every buffer, converted or scaled, +/// `Slot` from `reserve`; every buffer, converted or scaled, /// comes from the pool sized by `capacity`, so a producer that outruns its /// encoder gets `None` from `reserve` instead of unbounded device memory. Drop /// the frames the encoder has finished with (it retains its own clone until @@ -415,11 +404,12 @@ const CONVERT_PTX: &str = include_str!("rgba_to_nv12.ptx"); /// transfer function is applied on the way to Y'CbCr, so an sRGB-encoded 8-bit /// render target is not gamma-encoded twice. Chroma is 4:2:0, each sample the /// average of its 2x2 block. -pub struct Converter { +pub(crate) struct Converter { ctx: Arc, color: Color, kernel: CudaFunction, pool: Arc>, + importer: cuda_vulkan::Importer, } impl std::fmt::Debug for Converter { @@ -436,7 +426,7 @@ impl Converter { /// Open CUDA device `ordinal`, load the conversion kernel, and size the /// buffer pool. Two live frames per captured frame (the converted one and /// one scaled copy) plus one of slack is a reasonable `capacity`. - pub fn new(ordinal: usize, color: Color, capacity: NonZeroUsize) -> Result { + pub(crate) fn new(ordinal: usize, color: Color, capacity: NonZeroUsize) -> Result { let ctx = CudaContext::new(ordinal) .map_err(|e| Error::Unsupported(format!("CUDA device {ordinal} is unavailable: {e:?}")))?; let kernel = ctx @@ -444,6 +434,7 @@ impl Converter { .and_then(|module| module.load_function("rgba_to_nv12")) .map_err(|e| Error::Unsupported(format!("CUDA color conversion unavailable: {e:?}")))?; Ok(Self { + importer: cuda_vulkan::Importer::new(ordinal)?, pool: Arc::new(Pool::new(Device(ctx.clone()), capacity)), ctx, color, @@ -456,27 +447,35 @@ impl Converter { self.color } + /// Import `frame` as its reader, so dropping it returns the producer slot + /// even when no pool buffer is free to convert it. + pub(crate) fn import(&self, frame: &vulkan::Frame) -> Result<(), Error> { + self.importer.import(frame).map(drop) + } + /// Hold one pool buffer for the next frame, or `None` while every buffer is /// live: back-pressure, so drop the frame and try again after the encoder /// releases one. - pub fn reserve(&self) -> Option { + pub(crate) fn reserve(&self) -> Option { Some(Slot { reservation: self.pool.reserve()?, ctx: self.ctx.clone(), color: self.color, kernel: self.kernel.clone(), + importer: self.importer.clone(), }) } } -/// One buffer held from a [`Converter`]'s pool, filled by converting or +/// One buffer held from a `Converter`'s pool, filled by converting or /// scaling a frame into it. Dropped unfilled, or on a failed fill, it goes /// back to the pool. -pub struct Slot { +pub(crate) struct Slot { reservation: pool::Reservation, ctx: Arc, color: Color, kernel: CudaFunction, + importer: cuda_vulkan::Importer, } impl std::fmt::Debug for Slot { @@ -497,7 +496,8 @@ impl Slot { /// does not order against CUDA streams. Fails without touching the CPU when /// the image lives on another device or has an odd dimension (4:2:0 chroma /// needs even ones). - pub fn convert(self, frame: &vulkan::Frame) -> Result { + pub(crate) fn convert(self, frame: &vulkan::Frame) -> Result { + let frame = self.importer.import(frame)?; if frame.cuda_context().ordinal() != self.ctx.ordinal() { return Err(Error::Unsupported(format!( "Vulkan image on CUDA device {} cannot be converted on device {}", @@ -510,7 +510,7 @@ impl Slot { let dst = Frame::pooled(self.reservation, size, Some(self.color))?; let weights = self.color.coefficients(); - let bgra = u32::from(frame.channels() == vulkan::Channels::Bgra); + let bgra = u32::from(frame.format() == vulkan::Format::Bgra8); let stream = frame.cuda_stream(); // One thread per 2x2 block: the kernel writes four luma samples and one @@ -552,7 +552,7 @@ impl Slot { /// box-filter kernel, in the same color space. Fails without touching the /// CPU when `frame` lives on another device or the driver refuses the /// kernel. - pub fn resize(self, frame: &Frame, size: Size) -> Result { + pub(crate) fn resize(self, frame: &Frame, size: Size) -> Result { if frame.buf.ctx.ordinal() != self.ctx.ordinal() { return Err(Error::Unsupported(format!( "CUDA frame on device {} cannot be scaled on device {}", diff --git a/rs/moq-video/src/frame/cuda_test.rs b/rs/moq-video/src/frame/cuda_test.rs index 26fef6d811..0e11167735 100644 --- a/rs/moq-video/src/frame/cuda_test.rs +++ b/rs/moq-video/src/frame/cuda_test.rs @@ -3,12 +3,12 @@ use std::time::Duration; use super::*; use crate::frame::vulkan::tests::Producer; -use crate::frame::vulkan::{Channels, Image, ImportError, Importer, Timeline}; +use crate::frame::vulkan::{Format, Slot as VulkanSlot, Timeline}; use crate::{Frame as VideoFrame, Surface}; /// A gradient with structure on both axes, so a pitch, plane, or channel-order -/// bug moves the picture. Bytes are in `channels` order with alpha 255. -fn gradient(size: Size, channels: Channels, shift: usize) -> Vec { +/// bug moves the picture. Bytes are in `format` order with alpha 255. +fn gradient(size: Size, format: Format, shift: usize) -> Vec { let (w, h) = (size.width as usize, size.height as usize); let mut pixels = vec![0u8; w * h * 4]; for y in 0..h { @@ -19,9 +19,9 @@ fn gradient(size: Size, channels: Channels, shift: usize) -> Vec { ((x + y + shift) * 255 / (w + h)) as u8, ); let i = (y * w + x) * 4; - pixels[i..i + 4].copy_from_slice(&match channels { - Channels::Rgba => [r, g, b, 255], - Channels::Bgra => [b, g, r, 255], + pixels[i..i + 4].copy_from_slice(&match format { + Format::Rgba8 => [r, g, b, 255], + Format::Bgra8 => [b, g, r, 255], }); } } @@ -99,7 +99,7 @@ fn nvenc(size: Size, color: Color) -> crate::encode::Encoder { crate::encode::Encoder::new(&config).expect("open NVENC") } -/// Real hardware only, and loud about it: the opt-in `just rs vulkan-cuda` +/// Real hardware only, and loud about it: the opt-in `just rs gpu` /// recipe runs this, and a machine without the GPU path fails rather than /// reporting a pass it did not earn. /// @@ -119,17 +119,13 @@ async fn vulkan_cuda_convert_resize_encode() { let sd = Size::new(160, 96); let color = Color::Bt709Limited; let mut producer = Producer::new(size).expect("no Vulkan NVIDIA device: this opt-in test needs one"); - let importer = Importer::new(0, NonZeroUsize::new(1).unwrap()).expect("CUDA importer"); let converter = Converter::new(0, color, NonZeroUsize::new(3).unwrap()).expect("CUDA converter"); assert_eq!(converter.color(), color); // Round one, RGBA: conversion, resize, and the pool bound. - let image = Image::rgba8(producer.uuid, size, producer.allocation_size).unwrap(); - let slot = importer - .import(producer.export(), image, ()) - .map_err(ImportError::into_parts) - .expect("import RGBA image"); - let rgba = gradient(size, Channels::Rgba, 0); + let image = producer.contract(Format::Rgba8); + let slot = VulkanSlot::new(producer.export(), image, ()).expect("import RGBA image"); + let rgba = gradient(size, Format::Rgba8, 0); let expected = reference(&rgba, size, color); producer.upload(&rgba, None, 1); let (frame, completion) = slot.publish(Timeline::new(1, 2).unwrap()).unwrap(); @@ -201,19 +197,13 @@ async fn vulkan_cuda_convert_resize_encode() { eprintln!("pool bound held at capacity 3"); drop((converted, scaled, frame)); - let slot = tokio::time::timeout(std::time::Duration::from_secs(2), completion.wait()) - .await - .expect("CUDA completion timed out") - .expect("completion worker stopped"); + let slot = completion.wait().await.expect("completion worker stopped"); drop(slot); // Round two, BGRA: the same picture in the other channel order converts to // the same samples, and both renditions encode through NVENC in place. - let image = Image::bgra8(producer.uuid, size, producer.allocation_size).unwrap(); - let mut slot = importer - .import(producer.export(), image, ()) - .map_err(ImportError::into_parts) - .expect("import BGRA image"); + let image = producer.contract(Format::Bgra8); + let mut slot = VulkanSlot::new(producer.export(), image, ()).expect("import BGRA image"); let mut hd = nvenc(size, color); let mut sd_encoder = nvenc(sd, color); assert_eq!(hd.name(), "nvenc"); @@ -228,7 +218,7 @@ async fn vulkan_cuda_convert_resize_encode() { for i in 0..8u64 { let ready = 2 * i + 3; - let bgra = gradient(size, Channels::Bgra, i as usize * 4); + let bgra = gradient(size, Format::Bgra8, i as usize * 4); producer.upload(&bgra, Some(ready - 1), ready); let (frame, completion) = slot.publish(Timeline::new(ready, ready + 1).unwrap()).unwrap(); @@ -238,7 +228,7 @@ async fn vulkan_cuda_convert_resize_encode() { .convert(&frame) .expect("convert BGRA"); if i == 0 { - let rgba = gradient(size, Channels::Rgba, 0); + let rgba = gradient(size, Format::Rgba8, 0); let actual = converted.download_i420().unwrap(); let reference = reference(&rgba, size, color); assert_eq!(actual.y(), reference.y(), "BGRA luma differs from RGBA"); @@ -270,7 +260,7 @@ async fn vulkan_cuda_convert_resize_encode() { } for out in decoder.decode(packets[0].payload.clone(), timestamp, i == 0).unwrap() { decoded = Some(out.surface.to_i420().unwrap().into_owned()); - expected = Some(reference(&gradient(size, Channels::Rgba, i as usize * 4), size, color)); + expected = Some(reference(&gradient(size, Format::Rgba8, i as usize * 4), size, color)); } let packets = sd_encoder @@ -280,10 +270,7 @@ async fn vulkan_cuda_convert_resize_encode() { assert_eq!(packets[0].timestamp, timestamp); sd_packets += packets.len(); - slot = tokio::time::timeout(std::time::Duration::from_secs(2), completion.wait()) - .await - .expect("CUDA completion timed out") - .expect("completion worker stopped"); + slot = completion.wait().await.expect("completion worker stopped"); } assert_eq!(sd_packets, 8); @@ -301,7 +288,7 @@ async fn vulkan_cuda_convert_resize_encode() { assert!(mae(decoded.u(), expected.u()) < 8, "decoded u corrupt"); assert!(mae(decoded.v(), expected.v()) < 8, "decoded v corrupt"); - drop((hd, sd_encoder, slot, importer, converter)); + drop((hd, sd_encoder, slot, converter)); eprintln!("encoders, importer, and converter torn down"); } @@ -377,9 +364,8 @@ async fn vulkan_cuda_three_view_workload() { // One picture per tick phase, so the encoders see real motion instead of a // frozen frame no P-slice has to code. - let pictures: Vec> = (0..8).map(|i| gradient(size, Channels::Rgba, i * 37)).collect(); + let pictures: Vec> = (0..8).map(|i| gradient(size, Format::Rgba8, i * 37)).collect(); - let importer = Importer::new(0, NonZeroUsize::new(VIEWS).unwrap()).expect("CUDA importer"); // Two live buffers per view (the converted frame and its scaled copy) plus // one of slack, the sizing the `Converter` docs recommend. let converter = Converter::new(0, color, NonZeroUsize::new(2 * VIEWS + 1).unwrap()).expect("CUDA converter"); @@ -387,11 +373,8 @@ async fn vulkan_cuda_three_view_workload() { let mut views = Vec::new(); for _ in 0..VIEWS { let mut producer = Producer::new(size).expect("no Vulkan NVIDIA device: this opt-in test needs one"); - let image = Image::rgba8(producer.uuid, size, producer.allocation_size).unwrap(); - let slot = importer - .import(producer.export(), image, ()) - .map_err(ImportError::into_parts) - .expect("import view"); + let image = producer.contract(Format::Rgba8); + let slot = VulkanSlot::new(producer.export(), image, ()).expect("import view"); producer.upload(&pictures[0], None, 1); views.push((producer, Some(slot), nvenc(size, color), nvenc(sd, color), 1u64)); } @@ -448,10 +431,7 @@ async fn vulkan_cuda_three_view_workload() { bytes.1 += packets.iter().map(|p| p.payload.len()).sum::(); let (wall, cpu) = (std::time::Instant::now(), cpu_now()); - let returned = tokio::time::timeout(Duration::from_secs(2), completion.wait()) - .await - .expect("CUDA completion timed out") - .expect("completion worker stopped"); + let returned = completion.wait().await.expect("completion worker stopped"); complete.wall.push(wall.elapsed()); complete.cpu.push(cpu_now() - cpu); @@ -498,3 +478,134 @@ async fn vulkan_cuda_three_view_workload() { (bytes.0 + bytes.1) as f64 * 8.0 / elapsed.as_secs_f64() / 1e6, ); } + +/// Mean per-plane error a decoded NVENC rendition may show against the CPU reference. +const ROUNDTRIP_TOLERANCE: u64 = 8; + +/// The CPU reference for an RGBA gradient at `shift`, converted and then scaled to `output`. +fn expected(size: Size, shift: usize, color: Color, output: Size) -> I420 { + let full = reference(&gradient(size, Format::Rgba8, shift), size, color); + if output == size { + full + } else { + full.resize(output).unwrap() + } +} + +/// Each capture moves the gradient half a width, so successive pictures differ +/// by more than twice the round-trip tolerance and a stale one cannot pass. +fn capture_shift(size: Size, capture: u64) -> usize { + capture as usize * size.width as usize / 2 +} + +/// The pictures this test feeds are far enough apart for the tolerance to tell +/// them apart, at every rendition size. CPU only, so it guards the hardware +/// test's premise on every run. +#[test] +fn auto_external_captures_are_distinguishable() { + let size = Size::new(320, 192); + for output in [size, Size::new(160, 96)] { + for capture in 1..6 { + let previous = expected(size, capture_shift(size, capture - 1), Color::Bt709Limited, output); + let current = expected(size, capture_shift(size, capture), Color::Bt709Limited, output); + for (plane, a, b) in [ + ("y", current.y(), previous.y()), + ("u", current.u(), previous.u()), + ("v", current.v(), previous.v()), + ] { + let distance = mae(a, b); + assert!( + distance >= 2 * ROUNDTRIP_TOLERANCE, + "{output} capture {capture}: {plane} differs from the previous capture by only {distance}" + ); + } + } + } +} + +/// Producers declare only their device; both renditions import, convert, and scale inside NVENC. +/// +/// Every decoded rendition of every capture, in both channel orders, matches +/// the CPU conversion and resize of what the producer uploaded, so stale, +/// black, or channel-swapped output fails. +#[tokio::test] +#[ignore = "requires a Linux NVIDIA GPU with Vulkan/CUDA external memory and NVENC"] +async fn vulkan_cuda_auto_external_encode() { + let size = Size::new(320, 192); + let color = Color::Bt709Limited; + let mut producer = Producer::new(size).expect("NVIDIA Vulkan device required by opted-in test"); + let device = producer.contract(Format::Rgba8).device; + let decode = crate::decode::Config { + kind: crate::decode::Kind::Software, + ..crate::decode::Config::new() + }; + let mut renditions = Vec::new(); + for output in [size, Size::new(160, 96)] { + let mut config = crate::encode::Config::new(output.width, output.height, crate::Rate::new(30, 1).unwrap()); + config.input = Some(device); + config.color = Some(color); + config + .probe() + .await + .expect("probe the declared input device before capture"); + let encoder = crate::encode::Encoder::new(&config).expect("automatic external-image selection"); + assert_eq!(encoder.name(), "nvenc"); + let decoder = crate::decode::backend::open(crate::decode::backend::Codec::H264, &decode).unwrap(); + renditions.push((encoder, decoder)); + } + + let mut capture = 0u64; + for format in [Format::Rgba8, Format::Bgra8] { + let mut slot = VulkanSlot::new(producer.export(), producer.contract(format), ()).unwrap(); + for _ in 0..3 { + let ready = capture * 2 + 1; + let shift = capture_shift(size, capture); + producer.upload( + &gradient(size, format, shift), + (capture > 0).then_some(ready - 1), + ready, + ); + let (image, completion) = slot.publish(Timeline::new(ready, ready + 1).unwrap()).unwrap(); + let timestamp = moq_net::Timestamp::from_micros(capture * 33_333).unwrap(); + let frame = VideoFrame::new(Surface::Vulkan(image), timestamp); + for (encoder, decoder) in &mut renditions { + let output = encoder.size(); + encoder.cut().unwrap(); + let encoded = encoder + .encode(&frame) + .expect("external image encoded entirely on the GPU"); + assert_eq!(encoded.len(), 1, "one access unit for {output} capture {capture}"); + let mut decoded = Vec::new(); + for packet in encoded { + decoded.extend( + decoder + .decode(packet.payload, packet.timestamp, packet.keyframe) + .unwrap(), + ); + } + assert_eq!(decoded.len(), 1, "one picture for {output} capture {capture}"); + let actual = decoded[0].surface.to_i420().unwrap(); + assert_eq!(actual.size(), output); + let expected = expected(size, shift, color, output); + for (plane, actual, expected) in [ + ("y", actual.y(), expected.y()), + ("u", actual.u(), expected.u()), + ("v", actual.v(), expected.v()), + ] { + let error = mae(actual, expected); + eprintln!("{format:?} {output} capture {capture} {plane} mae={error}"); + assert!( + error < ROUNDTRIP_TOLERANCE, + "{format:?} {output} capture {capture}: decoded {plane} differs from the reference by {error}" + ); + } + } + drop(frame); + slot = completion + .wait() + .await + .expect("producer slot returned after both renditions"); + capture += 1; + } + } +} diff --git a/rs/moq-video/src/frame/cuda_vulkan.rs b/rs/moq-video/src/frame/cuda_vulkan.rs new file mode 100644 index 0000000000..46561a23b4 --- /dev/null +++ b/rs/moq-video/src/frame/cuda_vulkan.rs @@ -0,0 +1,360 @@ +//! CUDA imports and completion for neutral external Vulkan slots. +use super::vulkan::{self, Format, Handles, Image, uuid}; +use crate::{Error, Size}; +use cudarc::driver::sys; +use cudarc::driver::{CudaContext, CudaStream}; +use std::fmt; +use std::os::fd::AsRawFd; +use std::sync::{Arc, Mutex, mpsc}; + +/// Imports producer-owned Vulkan image slots into one CUDA device. +#[derive(Clone)] +pub(crate) struct Importer { + backend: Arc, +} + +impl fmt::Debug for Importer { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + f.debug_struct("Importer") + .field("device", &self.backend.ctx.ordinal()) + .finish_non_exhaustive() + } +} + +impl Importer { + /// Open CUDA device `ordinal` and its completion worker. + pub(crate) fn new(ordinal: usize) -> Result { + let ctx = CudaContext::new(ordinal) + .map_err(|e| Error::Unsupported(format!("CUDA device {ordinal} is unavailable: {e:?}")))?; + let (reap, receiver) = mpsc::channel::(); + let thread = std::thread::Builder::new() + .name("moq-video-vulkan-cuda".into()) + .spawn(move || { + while let Ok(job) = receiver.recv() { + let Reap::Synchronize(stream, completion, signal_queued) = job; + // A device loss returns an error here instead of stranding the + // producer. This is the dedicated completion worker, never the + // producer's render thread. A failed slot is released instead of + // being reused with an unsignalled Vulkan semaphore. + let reusable = signal_queued && stream.synchronize().is_ok(); + completion.complete(reusable); + } + }) + .map_err(|e| Error::Codec(anyhow::anyhow!("start Vulkan/CUDA completion worker: {e}")))?; + let thread_id = thread.thread().id(); + Ok(Self { + backend: Arc::new(Backend { + ctx, + reap: Mutex::new(Some(reap)), + thread: Mutex::new(Some((thread_id, thread))), + }), + }) + } + + pub(crate) fn import(&self, frame: &vulkan::Frame) -> Result { + let imported = frame.import(|| Imported::new(self.backend.clone(), frame.handles()?, frame.image().clone()))?; + frame.read(imported.clone())?; + Ok(Frame { + imported, + _frame: frame.clone(), + }) + } +} + +struct Backend { + ctx: Arc, + reap: Mutex>>, + thread: Mutex)>>, +} + +impl Backend { + fn send(&self, job: Reap) { + if let Some(sender) = self.reap.lock().expect("completion sender poisoned").as_ref() { + // A job owns a Slot, which owns this Backend, so the receiver cannot + // disappear before this send. + let _ = sender.send(job); + } + } +} + +impl Drop for Backend { + fn drop(&mut self) { + self.reap.lock().expect("completion sender poisoned").take(); + if let Some((thread_id, thread)) = self.thread.lock().expect("completion worker poisoned").take() + && std::thread::current().id() != thread_id + { + let _ = thread.join(); + } + } +} + +enum Reap { + Synchronize(Arc, Box, bool), +} + +/// One imported image retained alongside its published producer frame. +pub(crate) struct Frame { + imported: Arc, + _frame: vulkan::Frame, +} +impl Frame { + /// Image width in pixels. + #[cfg(test)] + pub(crate) fn width(&self) -> u32 { + self.imported.image.size.width + } + + /// Image height in pixels. + #[cfg(test)] + pub(crate) fn height(&self) -> u32 { + self.imported.image.size.height + } + + /// Image size in pixels. + pub(crate) fn size(&self) -> Size { + self.imported.image.size + } + + /// The pixel format the producer declared when importing the image. + pub(crate) fn format(&self) -> Format { + self.imported.image.format + } + + /// The level-0 array behind the surface, for the tests' readback only. + #[cfg(test)] + pub(crate) fn cuda_array(&self) -> sys::CUarray { + let mut array = std::ptr::null_mut(); + // SAFETY: the imported image has exactly one mip level and is alive. + unsafe { sys::cuMipmappedArrayGetLevel(&mut array, self.imported.mipmap, 0) } + .result() + .expect("Vulkan image mip level"); + array + } + + /// The surface object over the image, for a kernel reading it in place. + pub(crate) fn cuda_surface(&self) -> sys::CUsurfObject { + self.imported.surface + } + + /// The stream every CUDA reader of this image queues on, so the completion + /// signal queued after the last reader lands behind their work. + pub(crate) fn cuda_stream(&self) -> &Arc { + &self.imported.stream + } + + /// The context that owns the imported image. + pub(crate) fn cuda_context(&self) -> &Arc { + &self.imported.backend.ctx + } +} + +impl vulkan::Reader for Imported { + fn wait(&self, value: u64) -> Result<(), Error> { + self.wait(value) + } + fn finish(&self, value: u64, completion: Box) { + let signal_queued = self.signal(value).is_ok(); + self.backend + .send(Reap::Synchronize(self.stream.clone(), completion, signal_queued)); + } +} +pub(crate) struct Imported { + backend: Arc, + stream: Arc, + image: Image, + memory: sys::CUexternalMemory, + semaphore: sys::CUexternalSemaphore, + mipmap: sys::CUmipmappedArray, + surface: sys::CUsurfObject, +} + +// CUDA's external handles are explicitly safe to use from threads after making +// their context current. `CudaContext` and `CudaStream` already carry the same +// guarantees; cudarc's raw pointer aliases cannot express them. +unsafe impl Send for Imported {} +unsafe impl Sync for Imported {} + +impl Imported { + fn new(backend: Arc, handles: Handles, image: Image) -> Result { + if !matches!(&image.memory, vulkan::Memory::OpaqueFd { .. }) { + return Err(Error::Unsupported("CUDA accepts only OPAQUE_FD Vulkan images".into())); + } + backend.ctx.bind_to_thread().map_err(cuda("bind CUDA context"))?; + let actual = backend.ctx.uuid().map_err(cuda("read CUDA device UUID"))?; + // `CUuuid.bytes` is `[c_char; 16]`, which is `i8` on x86_64 and `u8` on + // aarch64. Reinterpret per element: an `as u8` cast fails clippy's + // `unnecessary_cast` where `c_char` is already `u8`. + let actual = actual.bytes.map(|b| u8::from_ne_bytes(b.to_ne_bytes())); + if actual != image.device.device_uuid { + return Err(Error::Unsupported(format!( + "Vulkan device UUID {} does not match CUDA device UUID {}", + uuid(image.device.device_uuid), + uuid(actual) + ))); + } + + let stream = backend.ctx.new_stream().map_err(cuda("create CUDA stream"))?; + let memory_desc = sys::CUDA_EXTERNAL_MEMORY_HANDLE_DESC { + type_: sys::CUexternalMemoryHandleType::CU_EXTERNAL_MEMORY_HANDLE_TYPE_OPAQUE_FD, + handle: sys::CUDA_EXTERNAL_MEMORY_HANDLE_DESC_st__bindgen_ty_1 { + fd: handles.memory.as_raw_fd(), + }, + size: image.allocation_size, + flags: 1, // CUDA_EXTERNAL_MEMORY_DEDICATED + reserved: [0; 16], + }; + let mut memory = std::ptr::null_mut(); + // SAFETY: the descriptor names a live owned fd and its exact allocation + // size. CUDA takes fd ownership only after successful import. + unsafe { sys::cuImportExternalMemory(&mut memory, &memory_desc) } + .result() + .map_err(cuda("import Vulkan image memory"))?; + std::mem::forget(handles.memory); + + let semaphore_desc = sys::CUDA_EXTERNAL_SEMAPHORE_HANDLE_DESC { + type_: sys::CUexternalSemaphoreHandleType::CU_EXTERNAL_SEMAPHORE_HANDLE_TYPE_TIMELINE_SEMAPHORE_FD, + handle: sys::CUDA_EXTERNAL_SEMAPHORE_HANDLE_DESC_st__bindgen_ty_1 { + fd: handles.timeline.as_raw_fd(), + }, + flags: 0, + reserved: [0; 16], + }; + let mut semaphore = std::ptr::null_mut(); + // SAFETY: the descriptor names a Vulkan timeline semaphore exported as an + // opaque fd. CUDA takes fd ownership only after successful import. + if let Err(error) = unsafe { sys::cuImportExternalSemaphore(&mut semaphore, &semaphore_desc) }.result() { + // SAFETY: memory was imported above and has no mappings yet. + let _ = unsafe { sys::cuDestroyExternalMemory(memory) }; + return Err(cuda("import Vulkan timeline semaphore")(error)); + } + std::mem::forget(handles.timeline); + + let array_desc = sys::CUDA_EXTERNAL_MEMORY_MIPMAPPED_ARRAY_DESC { + offset: 0, + arrayDesc: sys::CUDA_ARRAY3D_DESCRIPTOR { + Width: image.size.width as usize, + Height: image.size.height as usize, + Depth: 0, + Format: sys::CUarray_format::CU_AD_FORMAT_UNSIGNED_INT8, + NumChannels: 4, + Flags: sys::CUDA_ARRAY3D_SURFACE_LDST, + }, + numLevels: 1, + reserved: [0; 16], + }; + let mut mipmap = std::ptr::null_mut(); + // SAFETY: the Vulkan allocation is dedicated to an optimal-tiling RGBA8 + // image matching this descriptor. + if let Err(error) = + unsafe { sys::cuExternalMemoryGetMappedMipmappedArray(&mut mipmap, memory, &array_desc) }.result() + { + // SAFETY: both handles were imported and no work references them. + let _ = unsafe { sys::cuDestroyExternalSemaphore(semaphore) }; + let _ = unsafe { sys::cuDestroyExternalMemory(memory) }; + return Err(cuda("map Vulkan image in CUDA")(error)); + } + + let mut array = std::ptr::null_mut(); + // SAFETY: the imported image has exactly one mip level. + if let Err(error) = unsafe { sys::cuMipmappedArrayGetLevel(&mut array, mipmap, 0) }.result() { + // SAFETY: no work references these newly-created handles. + let _ = unsafe { sys::cuMipmappedArrayDestroy(mipmap) }; + let _ = unsafe { sys::cuDestroyExternalSemaphore(semaphore) }; + let _ = unsafe { sys::cuDestroyExternalMemory(memory) }; + return Err(cuda("get Vulkan image mip level")(error)); + } + + // A surface object is how a kernel reads the array in place; the array + // was mapped with `CUDA_ARRAY3D_SURFACE_LDST` for exactly this. + let surface_desc = sys::CUDA_RESOURCE_DESC { + resType: sys::CUresourcetype::CU_RESOURCE_TYPE_ARRAY, + res: sys::CUDA_RESOURCE_DESC_st__bindgen_ty_1 { + array: sys::CUDA_RESOURCE_DESC_st__bindgen_ty_1__bindgen_ty_1 { hArray: array }, + }, + flags: 0, + }; + let mut surface = 0; + // SAFETY: the descriptor names the live level-0 array mapped above. + if let Err(error) = unsafe { sys::cuSurfObjectCreate(&mut surface, &surface_desc) }.result() { + // SAFETY: no work references these newly-created handles. + let _ = unsafe { sys::cuMipmappedArrayDestroy(mipmap) }; + let _ = unsafe { sys::cuDestroyExternalSemaphore(semaphore) }; + let _ = unsafe { sys::cuDestroyExternalMemory(memory) }; + return Err(cuda("create surface over Vulkan image")(error)); + } + + Ok(Self { + backend, + stream, + image, + memory, + semaphore, + mipmap, + surface, + }) + } + + fn wait(&self, value: u64) -> Result<(), Error> { + self.backend.ctx.bind_to_thread().map_err(cuda("bind CUDA context"))?; + let params = timeline_wait(value); + // SAFETY: both semaphore and stream are live and belong to this context. + unsafe { sys::cuWaitExternalSemaphoresAsync(&self.semaphore, ¶ms, 1, self.stream.cu_stream()) } + .result() + .map_err(cuda("queue Vulkan timeline wait")) + } + + fn signal(&self, value: u64) -> Result<(), Error> { + self.backend.ctx.bind_to_thread().map_err(cuda("bind CUDA context"))?; + let params = timeline_signal(value); + // SAFETY: both semaphore and stream are live and belong to this context. + unsafe { sys::cuSignalExternalSemaphoresAsync(&self.semaphore, ¶ms, 1, self.stream.cu_stream()) } + .result() + .map_err(cuda("queue Vulkan timeline signal")) + } +} + +impl Drop for Imported { + fn drop(&mut self) { + if self.backend.ctx.bind_to_thread().is_ok() { + // Completion returned the slot only after stream synchronization, so no + // queued work can still reference these handles. + let _ = unsafe { sys::cuSurfObjectDestroy(self.surface) }; + let _ = unsafe { sys::cuMipmappedArrayDestroy(self.mipmap) }; + let _ = unsafe { sys::cuDestroyExternalMemory(self.memory) }; + let _ = unsafe { sys::cuDestroyExternalSemaphore(self.semaphore) }; + } + } +} + +fn timeline_wait(value: u64) -> sys::CUDA_EXTERNAL_SEMAPHORE_WAIT_PARAMS { + sys::CUDA_EXTERNAL_SEMAPHORE_WAIT_PARAMS { + params: sys::CUDA_EXTERNAL_SEMAPHORE_WAIT_PARAMS_st__bindgen_ty_1 { + fence: sys::CUDA_EXTERNAL_SEMAPHORE_WAIT_PARAMS_st__bindgen_ty_1__bindgen_ty_1 { value }, + nvSciSync: sys::CUDA_EXTERNAL_SEMAPHORE_WAIT_PARAMS_st__bindgen_ty_1__bindgen_ty_2 { reserved: 0 }, + keyedMutex: sys::CUDA_EXTERNAL_SEMAPHORE_WAIT_PARAMS_st__bindgen_ty_1__bindgen_ty_3 { + key: 0, + timeoutMs: 0, + }, + reserved: [0; 10], + }, + flags: 0, + reserved: [0; 16], + } +} + +fn timeline_signal(value: u64) -> sys::CUDA_EXTERNAL_SEMAPHORE_SIGNAL_PARAMS { + sys::CUDA_EXTERNAL_SEMAPHORE_SIGNAL_PARAMS { + params: sys::CUDA_EXTERNAL_SEMAPHORE_SIGNAL_PARAMS_st__bindgen_ty_1 { + fence: sys::CUDA_EXTERNAL_SEMAPHORE_SIGNAL_PARAMS_st__bindgen_ty_1__bindgen_ty_1 { value }, + nvSciSync: sys::CUDA_EXTERNAL_SEMAPHORE_SIGNAL_PARAMS_st__bindgen_ty_1__bindgen_ty_2 { reserved: 0 }, + keyedMutex: sys::CUDA_EXTERNAL_SEMAPHORE_SIGNAL_PARAMS_st__bindgen_ty_1__bindgen_ty_3 { key: 0 }, + reserved: [0; 12], + }, + flags: 0, + reserved: [0; 16], + } +} + +fn cuda(action: &'static str) -> impl FnOnce(cudarc::driver::result::DriverError) -> Error { + move |error| Error::Codec(anyhow::anyhow!("{action}: {error:?}")) +} diff --git a/rs/moq-video/src/frame/vaapi.rs b/rs/moq-video/src/frame/vaapi.rs index 569421bcbe..e685533b13 100644 --- a/rs/moq-video/src/frame/vaapi.rs +++ b/rs/moq-video/src/frame/vaapi.rs @@ -26,7 +26,7 @@ use moq_vaapi::dmabuf::{DmaBuf as VaapiDmaBuf, Plane}; use moq_vaapi::vpp::Processor; use moq_vaapi::{Matrix, VA_FOURCC_NV12}; -use super::{DmaBuf, DmaBufExport, DmaBufFrame, DmaBufPlane, DrmFormat, I420}; +use super::{DmaBuf, DmaBufExport, DmaBufFrame, DmaBufLayout, DmaBufPlane, DrmFormat, I420}; use crate::{Color, Error, Size}; /// Environment variable naming the render node, such as `/dev/dri/renderD129`, that [`device`] returns. @@ -259,13 +259,14 @@ pub(crate) fn adopt(frame: ExportedFrame, color: Option) -> anyhow::Resul .collect(); let modifier = object.drm_format_modifier; - DmaBuf::new( - DrmFormat::NV12, - modifier, - width, - height, - planes, - color, + DmaBuf::adopt( + DmaBufLayout { + format: DrmFormat::NV12, + modifier, + size: Size::new(width, height), + planes, + color, + }, Arc::new(Exported { frame: Mutex::new(frame), color, @@ -350,7 +351,7 @@ pub(crate) mod testing { use moq_vaapi::{Display, Image, Surface as VaSurface, UsageHint, VA_FOURCC_BGRX, VA_RT_FORMAT_RGB32}; - use super::super::{DmaBuf, DmaBufFrame, DmaBufPlane, DrmFormat, I420}; + use super::super::{DmaBuf, DmaBufFrame, DmaBufLayout, DmaBufPlane, DrmFormat, I420}; use crate::{Error, Size}; /// A driver surface exported as a DMA-BUF, standing in for a PipeWire buffer. @@ -390,13 +391,14 @@ pub(crate) mod testing { pub(crate) fn unimportable_dmabuf(pixels: I420) -> DmaBuf { let fd = OwnedFd::from(std::fs::File::open("/dev/null").expect("open /dev/null")); let (width, height) = (pixels.width(), pixels.height()); - DmaBuf::new( - DrmFormat::NV12, - 0x00ff_ffff_ffff_fffe, - width, - height, - vec![DmaBufPlane::new(0, width), DmaBufPlane::new(width * height, width)], - None, + DmaBuf::adopt( + DmaBufLayout { + format: DrmFormat::NV12, + modifier: 0x00ff_ffff_ffff_fffe, + size: Size::new(width, height), + planes: vec![DmaBufPlane::new(0, width), DmaBufPlane::new(width * height, width)], + color: None, + }, Arc::new(Unimportable { fd, pixels }), ) .expect("a valid description") @@ -446,13 +448,14 @@ pub(crate) mod testing { assert_eq!(layer.drm_format, DrmFormat::XRGB8888.as_raw()); let plane = DmaBufPlane::new(layer.offset[0], layer.pitch[0]); let object = exported.objects.remove(0); - DmaBuf::new( - DrmFormat::XRGB8888, - object.drm_format_modifier, - width, - height, - vec![plane], - None, + DmaBuf::adopt( + DmaBufLayout { + format: DrmFormat::XRGB8888, + modifier: object.drm_format_modifier, + size, + planes: vec![plane], + color: None, + }, Arc::new(Allocated { _surface: surface, fd: object.fd, diff --git a/rs/moq-video/src/frame/vulkan.rs b/rs/moq-video/src/frame/vulkan.rs index 0d858e495f..2af1d27b0f 100644 --- a/rs/moq-video/src/frame/vulkan.rs +++ b/rs/moq-video/src/frame/vulkan.rs @@ -1,275 +1,169 @@ -//! Vulkan images imported into CUDA on Linux/NVIDIA. +//! External Vulkan images and their producer-owned reuse slots. //! -//! A [`Slot`] is an exportable Vulkan allocation imported once. Publishing it -//! consumes the slot and [`Completion::wait`] returns it only after every CUDA -//! operation queued through the frame's stream has finished. A producer cannot -//! accidentally overwrite an image while a consumer still reads it. -//! -//! The image is packed RGBA or BGRA, which no encoder takes: a -//! [`cuda::Converter`](super::cuda::Converter) turns a published [`Frame`] into -//! the NV12 [`cuda::Frame`](super::cuda::Frame) NVENC encodes in place. +//! A producer exports a dedicated image and timeline semaphore once, then +//! publishes the slot for each capture. Encoders import it on its declared +//! device and return the slot only after their GPU reads have completed. +use std::any::{Any, TypeId}; +use std::collections::HashMap; use std::fmt; -use std::num::NonZeroUsize; -use std::os::fd::{AsRawFd, OwnedFd}; -use std::sync::atomic::{AtomicUsize, Ordering}; -use std::sync::{Arc, Mutex, mpsc}; - -use cudarc::driver::sys; -use cudarc::driver::{CudaContext, CudaStream}; +use std::os::fd::OwnedFd; +use std::sync::{Arc, Mutex}; use crate::{Error, Size}; +/// The physical device and driver that own an external image. +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +pub struct Device { + /// Vulkan physical-device UUID. + pub device_uuid: [u8; 16], + /// Vulkan driver UUID, required to import opaque Vulkan allocations. + pub driver_uuid: [u8; 16], + /// Render node `dev_t`, absent when the exporter has no DRM render node. + pub render_node: Option, +} + +impl fmt::Display for Device { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + write!(f, "device {} driver {}", uuid(self.device_uuid), uuid(self.driver_uuid)) + } +} + +/// One DRM modifier memory plane's allocation offset and row pitch. +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +pub struct Plane { + /// Byte offset from the start of the exported allocation. + pub offset: u64, + /// Byte pitch reported by the producer, including any row padding. + pub row_pitch: u64, +} + +/// The memory handle's import contract. +#[derive(Clone, Debug, PartialEq, Eq)] +pub enum Memory { + /// Dedicated optimal-tiling memory, imported with the exporter's memory type. + OpaqueFd { + /// Vulkan memory type index used for the original allocation. + memory_type: u32, + }, + /// A DMA-BUF image, imported with its DRM format modifier. + DmaBuf { + /// DRM format modifier describing the allocation's tiling. + modifier: u64, + /// Explicit memory-plane layouts, in the DRM modifier's plane order. + /// Importers check the plane count against the modifier's properties. + planes: Vec, + }, +} + /// The Vulkan handles exported for one reusable image slot. pub struct Handles { - /// A dedicated `VK_EXTERNAL_MEMORY_HANDLE_TYPE_OPAQUE_FD_BIT` allocation. + /// Memory exported with the handle type declared by [`Image::memory`]. pub memory: OwnedFd, - /// A `VK_EXTERNAL_SEMAPHORE_HANDLE_TYPE_OPAQUE_FD_BIT` timeline semaphore. + /// An `OPAQUE_FD` timeline semaphore. pub timeline: OwnedFd, } impl Handles { - /// Group the exported memory and timeline semaphore handles for an image. - pub fn new(memory: OwnedFd, timeline: OwnedFd) -> Self { - Self { memory, timeline } + #[cfg_attr(not(feature = "nvidia"), allow(dead_code))] + fn try_clone(&self) -> Result { + Ok(Self { + memory: self.memory.try_clone().map_err(|e| Error::Codec(e.into()))?, + timeline: self.timeline.try_clone().map_err(|e| Error::Codec(e.into()))?, + }) } } -/// The byte order of an imported image's four 8-bit channels. +/// The Vulkan format of an image's packed pixels. #[derive(Clone, Copy, Debug, PartialEq, Eq)] -pub enum Channels { +pub enum Format { /// `VK_FORMAT_R8G8B8A8_UNORM`. - Rgba, - /// `VK_FORMAT_B8G8R8A8_UNORM`, what a swapchain or an Unreal render target - /// usually holds. - Bgra, -} - -/// The Vulkan image contract accepted by CUDA. -#[derive(Clone, Copy, Debug, PartialEq, Eq)] + Rgba8, + /// `VK_FORMAT_B8G8R8A8_UNORM`. + Bgra8, +} + +/// An external dedicated Vulkan image, in `GENERAL` layout while being read. +/// +/// The producer creates a 2D image with one mip, one array layer, one sample, +/// no create flags, exclusive sharing, and usage `TRANSFER_DST | SAMPLED | +/// STORAGE`. It binds dedicated memory at offset zero. `OpaqueFd` images use +/// optimal tiling; `DmaBuf` images use DRM modifier tiling with explicit memory +/// planes. [`Format`] names the exact UNORM format. Importers recreate +/// this contract, including usage, rather than infer image creation parameters. +#[derive(Clone, Debug, PartialEq, Eq)] pub struct Image { - device_uuid: [u8; 16], - size: Size, - allocation_size: u64, - channels: Channels, + /// Device and driver that allocated the image. + pub device: Device, + /// Memory handle type and its import parameters. + pub memory: Memory, + /// Visible image size. + pub size: Size, + /// Complete allocation size, rather than the number of visible pixel bytes. + pub allocation_size: u64, + /// Packed pixel format. + pub format: Format, } impl Image { - /// Describe a dedicated optimal-tiling `VK_FORMAT_R8G8B8A8_UNORM` image. - /// - /// The image must be in `VK_IMAGE_LAYOUT_GENERAL` while CUDA owns the slot. - /// `allocation_size` is the complete `VkDeviceMemory` allocation size from - /// Vulkan, not `width * height * 4`. - pub fn rgba8(device_uuid: [u8; 16], size: Size, allocation_size: u64) -> Result { - Self::new(device_uuid, size, allocation_size, Channels::Rgba) - } - - /// Describe a dedicated optimal-tiling `VK_FORMAT_B8G8R8A8_UNORM` image, - /// under the same contract as [`rgba8`](Self::rgba8). - pub fn bgra8(device_uuid: [u8; 16], size: Size, allocation_size: u64) -> Result { - Self::new(device_uuid, size, allocation_size, Channels::Bgra) - } - - fn new(device_uuid: [u8; 16], size: Size, allocation_size: u64, channels: Channels) -> Result { - size.validate_nonzero("Vulkan/CUDA image")?; - if allocation_size == 0 { + fn validate(&self) -> Result<(), Error> { + self.size.validate_nonzero("external Vulkan image")?; + if self.allocation_size == 0 { return Err(Error::Unsupported( - "Vulkan/CUDA allocation size must be non-zero".into(), + "external Vulkan allocation size must be non-zero".into(), )); } - Ok(Self { - device_uuid, - size, - allocation_size, - channels, - }) - } - - /// Vulkan physical-device UUID required to match the CUDA device. - pub const fn device_uuid(&self) -> [u8; 16] { - self.device_uuid - } - - /// Visible image size. - pub const fn size(&self) -> Size { - self.size - } - - /// Complete size of the dedicated Vulkan memory allocation. - pub const fn allocation_size(&self) -> u64 { - self.allocation_size - } - - /// The channel order of the image's pixels. - pub const fn channels(&self) -> Channels { - self.channels + match &self.memory { + Memory::OpaqueFd { memory_type } if *memory_type >= 32 => { + return Err(Error::Unsupported("Vulkan memory type index must be below 32".into())); + } + Memory::DmaBuf { planes, .. } + if (planes.is_empty() + || planes.len() > 4 + || planes + .iter() + .any(|plane| plane.row_pitch == 0 || plane.offset >= self.allocation_size)) => + { + return Err(Error::Unsupported("Vulkan DMA-BUF requires one to four explicit memory planes with non-zero row pitches and offsets within its allocation".into())); + } + _ => {} + } + Ok(()) } } -/// Monotonic values for one Vulkan-to-CUDA-to-Vulkan handoff. +/// Monotonic values for one producer-to-consumer-to-producer handoff. #[derive(Clone, Copy, Debug, PartialEq, Eq)] pub struct Timeline { - /// Value Vulkan signals after finishing writes and the layout transition. + /// Value the producer signals after finishing writes and the layout transition. pub ready: u64, - /// Value CUDA signals after the last queued reader finishes. + /// Value the consumer signals after the last GPU reader finishes. pub complete: u64, } impl Timeline { - /// Create one handoff. `complete` must be greater than `ready`. + /// Create one handoff with a completion value greater than its ready value. pub fn new(ready: u64, complete: u64) -> Result { if complete <= ready { return Err(Error::Unsupported(format!( - "Vulkan/CUDA completion value {complete} must be greater than ready value {ready}" + "Vulkan completion value {complete} must exceed ready value {ready}" ))); } Ok(Self { ready, complete }) } } -/// Imports a bounded number of reusable Vulkan image slots into one CUDA device. -#[derive(Clone)] -pub struct Importer { - backend: Arc, -} - -impl fmt::Debug for Importer { - fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { - f.debug_struct("Importer") - .field("device", &self.backend.ctx.ordinal()) - .field("capacity", &self.backend.capacity) - .finish_non_exhaustive() - } -} - -impl Importer { - /// Open CUDA device `ordinal` and cap the number of imported image slots. - pub fn new(ordinal: usize, capacity: NonZeroUsize) -> Result { - let ctx = CudaContext::new(ordinal) - .map_err(|e| Error::Unsupported(format!("CUDA device {ordinal} is unavailable: {e:?}")))?; - let (reap, receiver) = mpsc::channel::(); - let thread = std::thread::Builder::new() - .name("moq-video-vulkan-cuda".into()) - .spawn(move || { - while let Ok(job) = receiver.recv() { - let Reap::Synchronize(stream, completion, signal_queued) = job; - // A device loss returns an error here instead of stranding the - // producer. This is the dedicated completion worker, never the - // producer's render thread. A failed slot is released instead of - // being reused with an unsignalled Vulkan semaphore. - let reusable = signal_queued && stream.synchronize().is_ok(); - completion.complete(reusable); - } - }) - .map_err(|e| Error::Codec(anyhow::anyhow!("start Vulkan/CUDA completion worker: {e}")))?; - let thread_id = thread.thread().id(); - Ok(Self { - backend: Arc::new(Backend { - ctx, - capacity: capacity.get(), - imported: AtomicUsize::new(0), - reap: Mutex::new(Some(reap)), - thread: Mutex::new(Some((thread_id, thread))), - }), - }) - } - - /// Import an exportable Vulkan image and attach its producer-owned slot. - /// - /// `owner` is returned inside [`Slot`] and remains retained until CUDA - /// completion. If import fails, it is returned in [`ImportError`]. The opaque - /// FDs are consumed by CUDA on success and closed on failure. - pub fn import( - &self, - handles: Handles, - image: Image, - owner: T, - ) -> Result, ImportError> { - if self - .backend - .imported - .fetch_update(Ordering::AcqRel, Ordering::Acquire, |count| { - (count < self.backend.capacity).then_some(count + 1) - }) - .is_err() - { - return Err(ImportError::new( - Error::Unsupported(format!( - "Vulkan/CUDA importer capacity {} exhausted", - self.backend.capacity - )), - owner, - )); - } - - match Imported::new(self.backend.clone(), handles, image) { - Ok(imported) => Ok(Slot { - imported: Arc::new(imported), - owner, - last_complete: 0, - }), - Err(error) => { - self.backend.imported.fetch_sub(1, Ordering::AcqRel); - Err(ImportError::new(error, owner)) - } - } - } -} - -struct Backend { - ctx: Arc, - capacity: usize, - imported: AtomicUsize, - reap: Mutex>>, - thread: Mutex)>>, -} - -impl Backend { - fn send(&self, job: Reap) { - if let Some(sender) = self.reap.lock().expect("completion sender poisoned").as_ref() { - // A job owns a Slot, which owns this Backend, so the receiver cannot - // disappear before this send. - let _ = sender.send(job); - } - } -} - -impl Drop for Backend { - fn drop(&mut self) { - self.reap.lock().expect("completion sender poisoned").take(); - if let Some((thread_id, thread)) = self.thread.lock().expect("completion worker poisoned").take() - && std::thread::current().id() != thread_id - { - let _ = thread.join(); - } - } -} - -enum Reap { - Synchronize(Arc, Box, bool), -} - -trait Complete: Send + Sync { - fn complete(self: Box, reusable: bool); -} - -struct ReturnSlot { - slot: Slot, - sender: tokio::sync::oneshot::Sender>, -} - -impl Complete for ReturnSlot { - fn complete(self: Box, reusable: bool) { - let Self { slot, sender } = *self; - if reusable { - let _ = sender.send(slot); - } - } +#[cfg_attr(not(feature = "nvidia"), allow(dead_code))] +struct Allocation { + handles: Handles, + image: Image, + // Imports belong to a slot, not to an FD that can be duplicated or reused. + imports: Mutex>>, } -/// One imported Vulkan image plus its producer-owned slot. +/// One external Vulkan allocation and the producer value that keeps it alive. pub struct Slot { - imported: Arc, + allocation: Arc, owner: T, last_complete: u64, } @@ -277,13 +171,32 @@ pub struct Slot { impl fmt::Debug for Slot { fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { f.debug_struct("Slot") - .field("image", &self.imported.image) + .field("image", &self.allocation.image) .field("last_complete", &self.last_complete) .finish_non_exhaustive() } } impl Slot { + /// Own exported handles and retain their producer without opening a backend. + /// + /// The producer guard must keep the Vulkan image, memory, and semaphore alive + /// and safely finish or cancel its own queued writes before destroying them. + /// An unread frame has no imported backend through which to wait for those + /// writes, so its failed completion releases the guard for producer teardown. + pub fn new(handles: Handles, image: Image, owner: T) -> Result { + image.validate()?; + Ok(Self { + allocation: Arc::new(Allocation { + handles, + image, + imports: Mutex::new(HashMap::new()), + }), + owner, + last_complete: 0, + }) + } + /// Access the producer-owned value while the slot is idle. pub fn owner(&self) -> &T { &self.owner @@ -294,42 +207,42 @@ impl Slot { &mut self.owner } - /// Publish this slot after Vulkan has queued its `ready` signal. + /// Publish after queueing the ready signal, returning a completion handle. /// - /// CUDA waits asynchronously. Dropping the last [`Frame`] queues the - /// `complete` signal after all consumer work on the frame stream, then the - /// completion worker returns this exact slot. + /// Publish only what you will encode. A frame that no backend reads (one a + /// publisher drops under backpressure, say) fails completion and loses its + /// slot for good: no consumer signalled its timeline, so the producer must + /// export a new slot to replace it. Drop excess captures before publishing + /// instead. The guard must safely finish or cancel queued producer writes + /// before teardown, as required by [`Slot::new`]. pub fn publish(self, timeline: Timeline) -> Result<(Frame, Completion), PublishError> { - if timeline.ready <= self.last_complete { + if timeline.complete <= timeline.ready || timeline.ready <= self.last_complete { return Err(PublishError::new( Error::Unsupported(format!( - "Vulkan/CUDA ready value {} must be greater than previous completion {}", - timeline.ready, self.last_complete + "invalid Vulkan handoff {timeline:?} after completion {}", + self.last_complete )), self, )); } - - if let Err(error) = self.imported.wait(timeline.ready) { - return Err(PublishError::new(error, self)); - } - let (sender, receiver) = tokio::sync::oneshot::channel(); - let imported = self.imported.clone(); + let allocation = self.allocation.clone(); let completion: Box = Box::new(ReturnSlot { slot: Slot { - imported: self.imported, - owner: self.owner, last_complete: timeline.complete, + ..self }, sender, }); Ok(( Frame { inner: Arc::new(FrameInner { - imported, - complete: timeline.complete, + allocation, + timeline, + reader: Mutex::new(None), completion: Some(completion), + #[cfg(feature = "nvidia")] + converted: Mutex::new(Vec::new()), }), }, Completion { receiver }, @@ -337,7 +250,7 @@ impl Slot { } } -/// A Vulkan image being read through CUDA. +/// A published external image, retained until every consumer drops its clone. #[derive(Clone)] pub struct Frame { inner: Arc, @@ -346,8 +259,8 @@ pub struct Frame { impl fmt::Debug for Frame { fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { f.debug_struct("Frame") - .field("image", &self.inner.imported.image) - .field("complete", &self.inner.complete) + .field("image", &self.inner.allocation.image) + .field("timeline", &self.inner.timeline) .finish_non_exhaustive() } } @@ -355,79 +268,126 @@ impl fmt::Debug for Frame { impl Frame { /// Image width in pixels. pub fn width(&self) -> u32 { - self.inner.imported.image.size.width + self.image().size.width } - /// Image height in pixels. pub fn height(&self) -> u32 { - self.inner.imported.image.size.height + self.image().size.height } - /// Image size in pixels. pub fn size(&self) -> Size { - self.inner.imported.image.size + self.image().size } - - /// The channel order the producer declared when importing the image. - pub fn channels(&self) -> Channels { - self.inner.imported.image.channels + /// Image and device import contract. + pub fn image(&self) -> &Image { + &self.inner.allocation.image } - - /// The level-0 array behind the surface, for the tests' readback only. - #[cfg(test)] - pub(crate) fn cuda_array(&self) -> sys::CUarray { - let mut array = std::ptr::null_mut(); - // SAFETY: the imported image has exactly one mip level and is alive. - unsafe { sys::cuMipmappedArrayGetLevel(&mut array, self.inner.imported.mipmap, 0) } - .result() - .expect("Vulkan image mip level"); - array + /// Packed pixel format. + pub fn format(&self) -> Format { + self.image().format } - /// The surface object over the image, for a kernel reading it in place. - pub(crate) fn cuda_surface(&self) -> sys::CUsurfObject { - self.inner.imported.surface + #[cfg(feature = "nvidia")] + pub(crate) fn handles(&self) -> Result { + self.inner.allocation.handles.try_clone() } - /// The stream every CUDA reader of this image queues on, so the completion - /// signal queued after the last reader lands behind their work. - pub(crate) fn cuda_stream(&self) -> &Arc { - &self.inner.imported.stream + #[cfg(feature = "nvidia")] + pub(crate) fn import( + &self, + create: impl FnOnce() -> Result, + ) -> Result, Error> { + let mut imports = self.inner.allocation.imports.lock().expect("Vulkan imports poisoned"); + if let Some(imported) = imports.get(&TypeId::of::()) { + return Ok(imported.clone().downcast().expect("Vulkan import type")); + } + let imported = Arc::new(create()?); + imports.insert(TypeId::of::(), imported.clone()); + Ok(imported) + } + + #[cfg(feature = "nvidia")] + pub(crate) fn read(&self, imported: Arc) -> Result<(), Error> { + let mut reader = self.inner.reader.lock().expect("Vulkan reader poisoned"); + if let Some(current) = reader.as_ref() { + if !Arc::ptr_eq(current, &imported) { + return Err(Error::Unsupported( + "external image cannot be read through multiple import backends".into(), + )); + } + } else { + imported.wait(self.inner.timeline.ready)?; + *reader = Some(imported); + } + Ok(()) } - /// The context that owns the imported image. - pub(crate) fn cuda_context(&self) -> &Arc { - &self.inner.imported.backend.ctx + #[cfg(feature = "nvidia")] + pub(crate) fn converted( + &self, + color: crate::Color, + convert: impl FnOnce() -> Result, + ) -> Result { + let mut converted = self.inner.converted.lock().expect("Vulkan conversions poisoned"); + if let Some((_, frame)) = converted.iter().find(|(existing, _)| *existing == color) { + return Ok(frame.clone()); + } + let frame = convert()?; + converted.push((color, frame.clone())); + Ok(frame) } } +pub(crate) trait Reader: Send + Sync { + #[cfg_attr(not(feature = "nvidia"), allow(dead_code))] + fn wait(&self, value: u64) -> Result<(), Error>; + fn finish(&self, value: u64, completion: Box); +} + struct FrameInner { - imported: Arc, - complete: u64, + allocation: Arc, + timeline: Timeline, + reader: Mutex>>, completion: Option>, + #[cfg(feature = "nvidia")] + converted: Mutex>, } impl Drop for FrameInner { fn drop(&mut self) { - let completion = self.completion.take().expect("Vulkan/CUDA completion missing"); - // Queueing is asynchronous. A device error releases the slot after the - // worker drains the stream, but never returns it for unsafe Vulkan reuse. - let signal_queued = self.imported.signal(self.complete).is_ok(); - self.imported.backend.send(Reap::Synchronize( - self.imported.stream.clone(), - completion, - signal_queued, - )); + let completion = self.completion.take().expect("Vulkan completion missing"); + if let Some(reader) = self.reader.get_mut().expect("Vulkan reader poisoned").take() { + reader.finish(self.timeline.complete, completion); + } else { + completion.complete(false); + } + } +} + +pub(crate) trait Complete: Send + Sync { + fn complete(self: Box, reusable: bool); +} + +struct ReturnSlot { + slot: Slot, + sender: tokio::sync::oneshot::Sender>, +} + +impl Complete for ReturnSlot { + fn complete(self: Box, reusable: bool) { + if reusable { + let _ = self.sender.send(self.slot); + } } } -/// Resolves to the producer slot after CUDA has signalled completion. +/// Resolves to the producer slot after the GPU has signalled completion. pub struct Completion { receiver: tokio::sync::oneshot::Receiver>, } impl Completion { - /// Return the slot if CUDA has completed, without blocking or polling CUDA. + /// Return the slot if the GPU has completed, without blocking or polling the GPU. pub fn try_wait(&mut self) -> Result>, CompletionError> { match self.receiver.try_recv() { Ok(slot) => Ok(Some(slot)), @@ -442,44 +402,11 @@ impl Completion { } } -/// CUDA failed before the producer slot became safe to reuse. +/// the GPU failed before the producer slot became safe to reuse. #[derive(Clone, Copy, Debug, thiserror::Error)] -#[error("Vulkan/CUDA completion failed; the producer slot was released")] +#[error("Vulkan GPU completion failed; the producer slot was released")] pub struct CompletionError; -/// An import failure that returns the producer-owned value. -pub struct ImportError { - error: Error, - owner: T, -} - -impl ImportError { - fn new(error: Error, owner: T) -> Self { - Self { error, owner } - } - - /// Split the error from the producer-owned value. - pub fn into_parts(self) -> (Error, T) { - (self.error, self.owner) - } -} - -impl fmt::Debug for ImportError { - fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { - f.debug_struct("ImportError") - .field("error", &self.error) - .finish_non_exhaustive() - } -} - -impl fmt::Display for ImportError { - fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { - fmt::Display::fmt(&self.error, f) - } -} - -impl std::error::Error for ImportError {} - /// A publish failure that returns the still-idle slot. pub struct PublishError { error: Error, @@ -513,209 +440,167 @@ impl fmt::Display for PublishError { impl std::error::Error for PublishError {} -struct Imported { - backend: Arc, - stream: Arc, - image: Image, - memory: sys::CUexternalMemory, - semaphore: sys::CUexternalSemaphore, - mipmap: sys::CUmipmappedArray, - surface: sys::CUsurfObject, -} - -// CUDA's external handles are explicitly safe to use from threads after making -// their context current. `CudaContext` and `CudaStream` already carry the same -// guarantees; cudarc's raw pointer aliases cannot express them. -unsafe impl Send for Imported {} -unsafe impl Sync for Imported {} - -impl Imported { - fn new(backend: Arc, handles: Handles, image: Image) -> Result { - backend.ctx.bind_to_thread().map_err(cuda("bind CUDA context"))?; - let actual = backend.ctx.uuid().map_err(cuda("read CUDA device UUID"))?; - // `CUuuid.bytes` is `[c_char; 16]`, which is `i8` on x86_64 and `u8` on - // aarch64. Reinterpret per element: an `as u8` cast fails clippy's - // `unnecessary_cast` where `c_char` is already `u8`. - let actual = actual.bytes.map(|b| u8::from_ne_bytes(b.to_ne_bytes())); - if actual != image.device_uuid { - return Err(Error::Unsupported(format!( - "Vulkan device UUID {} does not match CUDA device UUID {}", - uuid(image.device_uuid), - uuid(actual) - ))); - } - - let stream = backend.ctx.new_stream().map_err(cuda("create CUDA stream"))?; - let memory_desc = sys::CUDA_EXTERNAL_MEMORY_HANDLE_DESC { - type_: sys::CUexternalMemoryHandleType::CU_EXTERNAL_MEMORY_HANDLE_TYPE_OPAQUE_FD, - handle: sys::CUDA_EXTERNAL_MEMORY_HANDLE_DESC_st__bindgen_ty_1 { - fd: handles.memory.as_raw_fd(), - }, - size: image.allocation_size, - flags: 1, // CUDA_EXTERNAL_MEMORY_DEDICATED - reserved: [0; 16], - }; - let mut memory = std::ptr::null_mut(); - // SAFETY: the descriptor names a live owned fd and its exact allocation - // size. CUDA takes fd ownership only after successful import. - unsafe { sys::cuImportExternalMemory(&mut memory, &memory_desc) } - .result() - .map_err(cuda("import Vulkan image memory"))?; - std::mem::forget(handles.memory); - - let semaphore_desc = sys::CUDA_EXTERNAL_SEMAPHORE_HANDLE_DESC { - type_: sys::CUexternalSemaphoreHandleType::CU_EXTERNAL_SEMAPHORE_HANDLE_TYPE_TIMELINE_SEMAPHORE_FD, - handle: sys::CUDA_EXTERNAL_SEMAPHORE_HANDLE_DESC_st__bindgen_ty_1 { - fd: handles.timeline.as_raw_fd(), - }, - flags: 0, - reserved: [0; 16], - }; - let mut semaphore = std::ptr::null_mut(); - // SAFETY: the descriptor names a Vulkan timeline semaphore exported as an - // opaque fd. CUDA takes fd ownership only after successful import. - if let Err(error) = unsafe { sys::cuImportExternalSemaphore(&mut semaphore, &semaphore_desc) }.result() { - // SAFETY: memory was imported above and has no mappings yet. - let _ = unsafe { sys::cuDestroyExternalMemory(memory) }; - return Err(cuda("import Vulkan timeline semaphore")(error)); - } - std::mem::forget(handles.timeline); - - let array_desc = sys::CUDA_EXTERNAL_MEMORY_MIPMAPPED_ARRAY_DESC { - offset: 0, - arrayDesc: sys::CUDA_ARRAY3D_DESCRIPTOR { - Width: image.size.width as usize, - Height: image.size.height as usize, - Depth: 0, - Format: sys::CUarray_format::CU_AD_FORMAT_UNSIGNED_INT8, - NumChannels: 4, - Flags: sys::CUDA_ARRAY3D_SURFACE_LDST, - }, - numLevels: 1, - reserved: [0; 16], - }; - let mut mipmap = std::ptr::null_mut(); - // SAFETY: the Vulkan allocation is dedicated to an optimal-tiling RGBA8 - // image matching this descriptor. - if let Err(error) = - unsafe { sys::cuExternalMemoryGetMappedMipmappedArray(&mut mipmap, memory, &array_desc) }.result() - { - // SAFETY: both handles were imported and no work references them. - let _ = unsafe { sys::cuDestroyExternalSemaphore(semaphore) }; - let _ = unsafe { sys::cuDestroyExternalMemory(memory) }; - return Err(cuda("map Vulkan image in CUDA")(error)); - } +pub(crate) fn uuid(bytes: [u8; 16]) -> String { + bytes.iter().map(|byte| format!("{byte:02x}")).collect() +} - let mut array = std::ptr::null_mut(); - // SAFETY: the imported image has exactly one mip level. - if let Err(error) = unsafe { sys::cuMipmappedArrayGetLevel(&mut array, mipmap, 0) }.result() { - // SAFETY: no work references these newly-created handles. - let _ = unsafe { sys::cuMipmappedArrayDestroy(mipmap) }; - let _ = unsafe { sys::cuDestroyExternalSemaphore(semaphore) }; - let _ = unsafe { sys::cuDestroyExternalMemory(memory) }; - return Err(cuda("get Vulkan image mip level")(error)); - } +#[cfg(all(test, feature = "nvidia"))] +#[path = "vulkan_test.rs"] +pub(crate) mod tests; - // A surface object is how a kernel reads the array in place; the array - // was mapped with `CUDA_ARRAY3D_SURFACE_LDST` for exactly this. - let surface_desc = sys::CUDA_RESOURCE_DESC { - resType: sys::CUresourcetype::CU_RESOURCE_TYPE_ARRAY, - res: sys::CUDA_RESOURCE_DESC_st__bindgen_ty_1 { - array: sys::CUDA_RESOURCE_DESC_st__bindgen_ty_1__bindgen_ty_1 { hArray: array }, +#[cfg(test)] +mod unit_tests { + use super::*; + use std::os::unix::net::UnixStream; + + fn image() -> Image { + Image { + device: Device { + device_uuid: [0xaa; 16], + driver_uuid: [0xbb; 16], + render_node: None, }, - flags: 0, - }; - let mut surface = 0; - // SAFETY: the descriptor names the live level-0 array mapped above. - if let Err(error) = unsafe { sys::cuSurfObjectCreate(&mut surface, &surface_desc) }.result() { - // SAFETY: no work references these newly-created handles. - let _ = unsafe { sys::cuMipmappedArrayDestroy(mipmap) }; - let _ = unsafe { sys::cuDestroyExternalSemaphore(semaphore) }; - let _ = unsafe { sys::cuDestroyExternalMemory(memory) }; - return Err(cuda("create surface over Vulkan image")(error)); + memory: Memory::OpaqueFd { memory_type: 1 }, + size: Size::new(4, 2), + allocation_size: 4096, + format: Format::Rgba8, } - - Ok(Self { - backend, - stream, - image, - memory, - semaphore, - mipmap, - surface, - }) } - fn wait(&self, value: u64) -> Result<(), Error> { - self.backend.ctx.bind_to_thread().map_err(cuda("bind CUDA context"))?; - let params = timeline_wait(value); - // SAFETY: both semaphore and stream are live and belong to this context. - unsafe { sys::cuWaitExternalSemaphoresAsync(&self.semaphore, ¶ms, 1, self.stream.cu_stream()) } - .result() - .map_err(cuda("queue Vulkan timeline wait")) + fn slot() -> Slot<()> { + let (memory, timeline) = UnixStream::pair().unwrap(); + Slot::new( + Handles { + memory: memory.into(), + timeline: timeline.into(), + }, + image(), + (), + ) + .unwrap() + } + + #[test] + fn rejects_invalid_image_contracts() { + let mut contract = image(); + contract.size.width = 0; + assert!(contract.validate().is_err()); + contract.size = Size::new(3, 5); + assert!(contract.validate().is_ok()); + contract.allocation_size = 0; + assert!(contract.validate().is_err()); + contract = image(); + contract.memory = Memory::OpaqueFd { memory_type: 32 }; + assert!(contract.validate().is_err()); + } + + #[test] + fn refuses_dma_buf_without_explicit_valid_plane_layouts() { + for planes in [ + vec![], + vec![Plane { + offset: 0, + row_pitch: 0, + }], + vec![Plane { + offset: 4096, + row_pitch: 32, + }], + vec![ + Plane { + offset: 0, + row_pitch: 32 + }; + 5 + ], + ] { + let mut contract = image(); + contract.memory = Memory::DmaBuf { modifier: 0, planes }; + assert!(contract.validate().is_err()); + } + let mut padded = image(); + padded.memory = Memory::DmaBuf { + modifier: 0, + planes: vec![Plane { + offset: 128, + row_pitch: 64, + }], + }; + assert!(padded.validate().is_ok()); } - fn signal(&self, value: u64) -> Result<(), Error> { - self.backend.ctx.bind_to_thread().map_err(cuda("bind CUDA context"))?; - let params = timeline_signal(value); - // SAFETY: both semaphore and stream are live and belong to this context. - unsafe { sys::cuSignalExternalSemaphoresAsync(&self.semaphore, ¶ms, 1, self.stream.cu_stream()) } - .result() - .map_err(cuda("queue Vulkan timeline signal")) + #[test] + fn validates_timeline_even_when_fields_are_set_directly() { + assert!(Timeline::new(4, 4).is_err()); + assert!(Timeline::new(5, 4).is_err()); + assert!(slot().publish(Timeline { ready: 4, complete: 4 }).is_err()); + assert!(slot().publish(Timeline { ready: 0, complete: 1 }).is_err()); } -} -impl Drop for Imported { - fn drop(&mut self) { - if self.backend.ctx.bind_to_thread().is_ok() { - // Completion returned the slot only after stream synchronization, so no - // queued work can still reference these handles. - let _ = unsafe { sys::cuSurfObjectDestroy(self.surface) }; - let _ = unsafe { sys::cuMipmappedArrayDestroy(self.mipmap) }; - let _ = unsafe { sys::cuDestroyExternalMemory(self.memory) }; - let _ = unsafe { sys::cuDestroyExternalSemaphore(self.semaphore) }; + #[tokio::test] + async fn unconsumed_image_retains_guard_until_last_clone_and_is_never_recycled() { + use std::sync::atomic::{AtomicUsize, Ordering}; + struct Guard(Arc); + impl Drop for Guard { + fn drop(&mut self) { + self.0.fetch_add(1, Ordering::SeqCst); + } } - self.backend.imported.fetch_sub(1, Ordering::AcqRel); - } -} - -fn timeline_wait(value: u64) -> sys::CUDA_EXTERNAL_SEMAPHORE_WAIT_PARAMS { - sys::CUDA_EXTERNAL_SEMAPHORE_WAIT_PARAMS { - params: sys::CUDA_EXTERNAL_SEMAPHORE_WAIT_PARAMS_st__bindgen_ty_1 { - fence: sys::CUDA_EXTERNAL_SEMAPHORE_WAIT_PARAMS_st__bindgen_ty_1__bindgen_ty_1 { value }, - nvSciSync: sys::CUDA_EXTERNAL_SEMAPHORE_WAIT_PARAMS_st__bindgen_ty_1__bindgen_ty_2 { reserved: 0 }, - keyedMutex: sys::CUDA_EXTERNAL_SEMAPHORE_WAIT_PARAMS_st__bindgen_ty_1__bindgen_ty_3 { - key: 0, - timeoutMs: 0, + let released = Arc::new(AtomicUsize::new(0)); + let (memory, timeline) = UnixStream::pair().unwrap(); + let slot = Slot::new( + Handles { + memory: memory.into(), + timeline: timeline.into(), }, - reserved: [0; 10], - }, - flags: 0, - reserved: [0; 16], - } -} - -fn timeline_signal(value: u64) -> sys::CUDA_EXTERNAL_SEMAPHORE_SIGNAL_PARAMS { - sys::CUDA_EXTERNAL_SEMAPHORE_SIGNAL_PARAMS { - params: sys::CUDA_EXTERNAL_SEMAPHORE_SIGNAL_PARAMS_st__bindgen_ty_1 { - fence: sys::CUDA_EXTERNAL_SEMAPHORE_SIGNAL_PARAMS_st__bindgen_ty_1__bindgen_ty_1 { value }, - nvSciSync: sys::CUDA_EXTERNAL_SEMAPHORE_SIGNAL_PARAMS_st__bindgen_ty_1__bindgen_ty_2 { reserved: 0 }, - keyedMutex: sys::CUDA_EXTERNAL_SEMAPHORE_SIGNAL_PARAMS_st__bindgen_ty_1__bindgen_ty_3 { key: 0 }, - reserved: [0; 12], - }, - flags: 0, - reserved: [0; 16], + image(), + Guard(released.clone()), + ) + .unwrap(); + let (frame, mut completion) = slot.publish(Timeline::new(1, 2).unwrap()).unwrap(); + let held = frame.clone(); + drop(frame); + assert_eq!(released.load(Ordering::SeqCst), 0); + assert!(completion.try_wait().unwrap().is_none()); + drop(held); + assert!(completion.wait().await.is_err()); + assert_eq!(released.load(Ordering::SeqCst), 1); + } + + #[cfg(feature = "nvidia")] + #[tokio::test] + async fn caches_imports_by_slot_and_returns_only_after_last_reader() { + struct Import; + impl Reader for Import { + fn wait(&self, _: u64) -> Result<(), Error> { + Ok(()) + } + fn finish(&self, _: u64, completion: Box) { + completion.complete(true); + } + } + let (frame, mut completion) = slot().publish(Timeline::new(1, 2).unwrap()).unwrap(); + let imported = frame.import(|| Ok(Import)).unwrap(); + let duplicate_handles = frame.handles().unwrap(); + drop(duplicate_handles); + assert!(Arc::ptr_eq( + &imported, + &frame.import::(|| panic!("imported twice")).unwrap() + )); + frame.read(imported.clone()).unwrap(); + let held = frame.clone(); + drop(frame); + assert!(completion.try_wait().unwrap().is_none()); + drop(held); + let slot = completion.wait().await.unwrap(); + let (_, completion) = slot.publish(Timeline::new(2, 3).unwrap()).err().unwrap().into_parts(); + let (frame, completion) = completion.publish(Timeline::new(3, 4).unwrap()).unwrap(); + assert!(Arc::ptr_eq( + &imported, + &frame.import::(|| panic!("reimported slot")).unwrap() + )); + frame.read(imported).unwrap(); + drop(frame); + assert!(completion.wait().await.is_ok()); } } - -fn cuda(action: &'static str) -> impl FnOnce(cudarc::driver::result::DriverError) -> Error { - move |error| Error::Codec(anyhow::anyhow!("{action}: {error:?}")) -} - -fn uuid(bytes: [u8; 16]) -> String { - bytes.iter().map(|byte| format!("{byte:02x}")).collect() -} - -#[cfg(test)] -#[path = "vulkan_test.rs"] -pub(crate) mod tests; diff --git a/rs/moq-video/src/frame/vulkan_test.rs b/rs/moq-video/src/frame/vulkan_test.rs index 4769472835..eda9cd53a4 100644 --- a/rs/moq-video/src/frame/vulkan_test.rs +++ b/rs/moq-video/src/frame/vulkan_test.rs @@ -1,5 +1,4 @@ use std::ffi::CStr; -use std::num::NonZeroUsize; use std::os::fd::{FromRawFd, OwnedFd}; use ash::vk; @@ -7,20 +6,6 @@ use cudarc::driver::sys; use super::*; -#[test] -fn rejects_invalid_image_contracts() { - assert!(Image::rgba8([0; 16], Size::new(0, 2), 4096).is_err()); - assert!(Image::rgba8([0; 16], Size::new(3, 5), 4096).is_ok()); - assert!(Image::rgba8([0; 16], Size::new(2, 2), 0).is_err()); -} - -#[test] -fn rejects_non_monotonic_timeline_values() { - assert!(Timeline::new(4, 4).is_err()); - assert!(Timeline::new(5, 4).is_err()); - assert_eq!(Timeline::new(4, 5).unwrap(), Timeline { ready: 4, complete: 5 }); -} - /// Real hardware only: a native Vulkan producer clears one exportable image to /// distinct identities, CUDA imports it once, and the same producer slot makes /// sixteen synchronized round trips. The one CUDA-to-host copy per iteration is @@ -28,23 +13,13 @@ fn rejects_non_monotonic_timeline_values() { #[tokio::test] #[ignore = "requires a Linux NVIDIA GPU with Vulkan/CUDA external-memory support"] async fn vulkan_cuda_slot_reuse_and_teardown() { - if unsafe { libloading::Library::new("libcuda.so.1") }.is_err() { - eprintln!("skipping: libcuda.so.1 unavailable"); - return; - } - let Some(mut producer) = Producer::new(Size::new(64, 32)) else { - eprintln!("skipping: no compatible Vulkan NVIDIA device"); - return; - }; + let mut producer = Producer::new(Size::new(64, 32)).expect("NVIDIA Vulkan device required by opted-in test"); - let importer = Importer::new(0, NonZeroUsize::new(1).unwrap()).expect("CUDA importer"); - let image = Image::rgba8(producer.uuid, producer.size, producer.allocation_size).unwrap(); + let importer = super::super::cuda_vulkan::Importer::new(0).expect("CUDA importer"); + let image = producer.contract(Format::Rgba8); let (released, release) = tokio::sync::oneshot::channel(); let owner = Owner(Some(released)); - let mut slot = importer - .import(producer.export(), image, owner) - .map_err(ImportError::into_parts) - .expect("import Vulkan image into CUDA"); + let mut slot = Slot::new(producer.export(), image, owner).expect("import Vulkan image into CUDA"); for identity in 1u8..=16 { let ready = u64::from(identity) * 2 - 1; @@ -54,7 +29,9 @@ async fn vulkan_cuda_slot_reuse_and_teardown() { let (frame, mut completion) = slot.publish(Timeline::new(ready, complete).unwrap()).unwrap(); let held = frame.clone(); - let pixels = readback(&frame); + let imported = importer.import(&frame).expect("CUDA image import"); + let pixels = readback(&imported); + drop(imported); eprintln!("cuda validation readback identity={identity} bytes={}", pixels.len()); assert_eq!(&pixels[..4], &[identity, 255 - identity, identity / 2, 255]); @@ -64,10 +41,7 @@ async fn vulkan_cuda_slot_reuse_and_teardown() { "a held reader returned the producer slot" ); drop(held); - slot = tokio::time::timeout(std::time::Duration::from_secs(2), completion.wait()) - .await - .expect("CUDA completion timed out") - .expect("completion worker stopped"); + slot = completion.wait().await.expect("completion worker stopped"); eprintln!("cuda complete={complete}; producer slot returned"); } @@ -77,12 +51,10 @@ async fn vulkan_cuda_slot_reuse_and_teardown() { producer.clear(33, Some(32), ready); let (frame, completion) = slot.publish(Timeline::new(ready, 34).unwrap()).unwrap(); drop(completion); + drop(importer.import(&frame).expect("CUDA image import")); drop(frame); drop(importer); - tokio::time::timeout(std::time::Duration::from_secs(2), release) - .await - .expect("cancelled completion did not drain") - .expect("owner release sender dropped"); + release.await.expect("owner release sender dropped"); eprintln!("cancelled consumer drained; imported resources and producer owner released"); } @@ -97,7 +69,7 @@ impl Drop for Owner { } } -fn readback(frame: &Frame) -> Vec { +fn readback(frame: &super::super::cuda_vulkan::Frame) -> Vec { let width = frame.width() as usize; let height = frame.height() as usize; let mut pixels = vec![0u8; width * height * 4]; @@ -147,6 +119,8 @@ pub(crate) struct Producer { staging_memory: vk::DeviceMemory, staging_ptr: *mut u8, pub(crate) uuid: [u8; 16], + pub(crate) driver_uuid: [u8; 16], + pub(crate) memory_type: u32, pub(crate) size: Size, pub(crate) allocation_size: u64, first: bool, @@ -322,6 +296,8 @@ impl Producer { staging_memory, staging_ptr, uuid, + driver_uuid: id.driver_uuid, + memory_type, size, allocation_size: requirements.size, first: true, @@ -329,6 +305,22 @@ impl Producer { }) } + pub(crate) fn contract(&self, format: Format) -> Image { + Image { + device: Device { + device_uuid: self.uuid, + driver_uuid: self.driver_uuid, + render_node: None, + }, + memory: Memory::OpaqueFd { + memory_type: self.memory_type, + }, + size: self.size, + allocation_size: self.allocation_size, + format, + } + } + pub(crate) fn export(&self) -> Handles { let memory_fd = ash::khr::external_memory_fd::Device::new(&self.instance, &self.device); let semaphore_fd = ash::khr::external_semaphore_fd::Device::new(&self.instance, &self.device); @@ -351,7 +343,12 @@ impl Producer { .expect("export Vulkan timeline semaphore") }; // SAFETY: Vulkan returned fresh owned opaque fds to the caller. - unsafe { Handles::new(OwnedFd::from_raw_fd(memory), OwnedFd::from_raw_fd(timeline)) } + unsafe { + Handles { + memory: OwnedFd::from_raw_fd(memory), + timeline: OwnedFd::from_raw_fd(timeline), + } + } } /// Clear the whole image to a color derived from `identity`. @@ -511,3 +508,19 @@ impl Drop for Producer { } } } + +/// The GPU recipe must distinguish a driverless GPU from a GPU-less host. +#[test] +#[ignore = "requires the host PCI GPU's Vulkan driver; run just rs gpu"] +fn vulkan_gpu_device() { + let vendor = std::env::var("MOQ_GPU_VENDOR").expect("just rs gpu supplies the PCI vendor"); + let vendor = u32::from_str_radix(vendor.trim_start_matches("0x"), 16).unwrap(); + let entry = unsafe { ash::Entry::load() }.expect("Vulkan loader missing"); + let instance = unsafe { entry.create_instance(&vk::InstanceCreateInfo::default(), None) }.expect("Vulkan instance"); + let devices = unsafe { instance.enumerate_physical_devices() }.expect("Vulkan devices"); + let found = devices + .into_iter() + .any(|physical| unsafe { instance.get_physical_device_properties(physical) }.vendor_id == vendor); + unsafe { instance.destroy_instance(None) }; + assert!(found, "no Vulkan driver exposes detected PCI GPU vendor {vendor:#x}"); +} diff --git a/rs/moq-video/src/lib.rs b/rs/moq-video/src/lib.rs index 43f23fb156..9f85712630 100644 --- a/rs/moq-video/src/lib.rs +++ b/rs/moq-video/src/lib.rs @@ -103,7 +103,7 @@ mod v4l2; pub use color::Color; pub use error::Error; #[cfg(all(target_os = "linux", feature = "dmabuf"))] -pub use frame::{DmaBuf, DmaBufExport, DmaBufPlane, DrmFormat}; +pub use frame::{DmaBuf, DmaBufExport, DmaBufLayout, DmaBufPlane, DrmFormat}; pub use frame::{Frame, I420, Surface}; pub use output::Output; pub use rate::{MAX_FRAMES_PER_SECOND, Rate, RateError}; diff --git a/sh/rs/gpu.sh b/sh/rs/gpu.sh new file mode 100755 index 0000000000..72e8b95698 --- /dev/null +++ b/sh/rs/gpu.sh @@ -0,0 +1,35 @@ +#!/usr/bin/env bash +# Hardware tests are opt-in; a detected GPU must have its driver and pass its tests. +set -euo pipefail +cd "$(git rev-parse --show-toplevel)" + +declare -A vendors=() +for device in /sys/bus/pci/devices/*; do + [[ -f "$device/class" && $(<"$device/class") == 0x03* ]] || continue + vendor=$(<"$device/vendor") + case "$vendor" in + 0x10de | 0x1002 | 0x8086) ;; + *) + # A BMC or virtual display adapter has no tests; a supported GPU still must. + echo "Skipping unsupported display controller $vendor at $device" >&2 + continue + ;; + esac + [[ -L "$device/driver" ]] || { + echo "GPU $device has no kernel driver" >&2 + exit 1 + } + vendors[$vendor]=1 +done +[[ ${#vendors[@]} -gt 0 ]] || { + echo 'No supported PCI GPU detected' >&2 + exit 1 +} + +for vendor in "${!vendors[@]}"; do + MOQ_GPU_VENDOR="$vendor" just rs test -p moq-video --run-ignored only -E 'test(vulkan_gpu_device)' + case "$vendor" in + 0x10de) sh/rs/vulkan-cuda.sh ;; + 0x1002 | 0x8086) echo "GPU $vendor: Vulkan driver verified; encoder hardware tests arrive with its backend" ;; + esac +done diff --git a/sh/rs/vulkan-cuda.sh b/sh/rs/vulkan-cuda.sh index bcdc12944c..fbd5441107 100755 --- a/sh/rs/vulkan-cuda.sh +++ b/sh/rs/vulkan-cuda.sh @@ -12,6 +12,17 @@ if [[ -z "$driver" ]]; then echo "libcuda.so.1 is not installed" >&2 exit 1 fi -driver_dir=$(dirname "$driver") +# Expose only NVIDIA libraries; adding the whole host directory can override +# the devshell's newer libc and libm with incompatible host copies. +driver_dir=$(mktemp -d) +trap 'rm -rf "$driver_dir"' EXIT +for library in libcuda.so.1 libnvidia-encode.so.1 libnvidia-ptxjitcompiler.so.1; do + path=$(/usr/sbin/ldconfig -p | awk -v name="$library" '$1 == name {print $NF; exit}') + [[ -n "$path" ]] || { + echo "$library is not installed" >&2 + exit 1 + } + ln -s "$path" "$driver_dir/$library" +done export LD_LIBRARY_PATH="$driver_dir${LD_LIBRARY_PATH:+:$LD_LIBRARY_PATH}" -cargo nextest run --locked -p moq-video --run-ignored only -E 'test(/^frame::(vulkan|cuda)::tests::vulkan_cuda_/)' +just rs test -p moq-video --run-ignored only -E 'test(/^frame::(vulkan|cuda)::tests::vulkan_cuda_/)'