diff --git a/Cargo.lock b/Cargo.lock index be4466068..552fbc5bf 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -5130,7 +5130,9 @@ dependencies = [ "futures", "h264-reader", "hang", + "libc", "libloading 0.9.0", + "linux-raw-sys 0.12.1", "moq-mux", "moq-net", "moq-nvenc", diff --git a/Cargo.toml b/Cargo.toml index 593f397e9..8715f5c9d 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -76,6 +76,8 @@ rust-version = "1.91" flate2 = "1.1" hang = { version = "0.20", path = "rs/hang" } kio = { version = "0.5", path = "rs/kio" } +libc = "0.2" +linux-raw-sys = { version = "0.12.1", features = ["ioctl"] } # Permutation testing for the kio primitives (and everything built on them). # Only compiled under `--cfg loom`; see `just rs loom`. loom = { version = "0.7.2", features = ["futures"] } diff --git a/doc/lib/rs/crate/moq-video.md b/doc/lib/rs/crate/moq-video.md index f830f4911..1f5f553c4 100644 --- a/doc/lib/rs/crate/moq-video.md +++ b/doc/lib/rs/crate/moq-video.md @@ -28,7 +28,7 @@ Four role modules, symmetric on both ends of the wire: | `capture` | Camera, display, window, or application frames | AVFoundation + ScreenCaptureKit (macOS), V4L2 + PipeWire (Linux), Media Foundation + DXGI (Windows) | | `encode` | Raw frames to H.264/H.265, published through `moq-mux` | VideoToolbox, Media Foundation, NVENC, VAAPI, openh264 | | `decode` | A subscribed track back to raw frames | VideoToolbox, Media Foundation/DXVA, NVDEC, openh264 | -| `render` | A frame drawn on the GPU, handed back as a `wgpu` texture | wgpu, with a zero-copy Metal import on macOS | +| `render` | A frame drawn on the GPU, handed back as a `wgpu` texture | wgpu, with zero-copy Metal and Vulkan imports | A picture is a `Frame` wherever it crosses the API: a `moq_net::Timestamp` and a `Surface` holding the pixels. Capture and decode produce them, encode and render @@ -66,8 +66,8 @@ cargo add moq-video --features render,pipewire | `capture` | yes | Native device capture (`v4l` and `zune-jpeg` on Linux) | | `nvenc` / `nvdec` | yes | NVIDIA encode/decode on Linux (`cudarc`, `moq-nvenc`) | | `vaapi` | no | Intel/AMD encode on Linux (`moq-vaapi`), unvalidated on hardware | -| `render` | no | `wgpu` and the GPU renderer | -| `pipewire` | no | Wayland/X11 screen capture via xdg-desktop-portal | +| `render` | no | `wgpu`, the GPU renderer, and Linux DMA-BUF support | +| `pipewire` | no | Wayland/X11 screen capture via xdg-desktop-portal and DMA-BUF | `--no-default-features` gives a codec-only build that still encodes and decodes H.264 with openh264 but omits native capture and the Linux GPU dependencies. A @@ -133,9 +133,9 @@ while let Some(frame) = video.read().await? { ## Zero-copy `Surface` is a `#[non_exhaustive]` enum naming what actually holds a frame's -pixels: a `CVPixelBuffer` on macOS, a Direct3D 11 texture on Windows, CUDA memory -on Linux, or plain I420 anywhere. Keeping a decoded frame in the first three -avoids a round trip through system memory on every frame. +pixels: a `CVPixelBuffer` on macOS, a Direct3D 11 texture on Windows, CUDA or a +DMA-BUF on Linux, or plain I420 anywhere. Keeping a frame in one of the native +representations avoids a round trip through system memory on every frame. How far that gets today depends on the platform, so here is the honest matrix rather than a blanket promise: @@ -143,15 +143,16 @@ rather than a blanket promise: | Platform | Decode output | Zero-copy transcode | Zero-copy render | | --- | --- | --- | --- | | macOS | `PixelBuffer` (VideoToolbox) | yes | yes, via `CVMetalTextureCache` | -| Linux | `Cuda` (NVDEC) | yes, straight into NVENC | no, downloaded to I420 first | +| Linux | `Cuda` (NVDEC) | yes, straight into NVENC | decoded CUDA frames: no; packed PipeWire DMA-BUF capture: yes, via Vulkan | | Windows | `Texture` (Media Foundation / DXVA) | yes, through the Direct3D11 video processor | no, downloaded to I420 first | `Frame::resize` stays on the GPU through a `VTPixelTransferSession`, CUDA kernel, or Direct3D11 video processor. Call `Frame::resize_with` with `resize::Acceleration::Cpu` to force a download and CPU resize. A driver that -rejects GPU resizing returns to CPU scaling and warns once. Rendering is -zero-copy on macOS only; the Vulkan and EGL importers that would extend it are -tracked in [#2481](https://github.com/moq-dev/moq/issues/2481). +rejects GPU resizing returns to CPU scaling and warns once. Linux Vulkan can +import packed RGB DMA-BUF screen frames. Multi-plane NV12 import and retiling a +modifier that Vulkan rejects remain tracked in +[#2819](https://github.com/moq-dev/moq/issues/2819). Matching on `Surface` stays portable because every variant has a universal fallback in `Surface::into_i420()`: take the fast path you recognize and let the @@ -182,6 +183,10 @@ The `wgpu` version this was built against is re-exported as `moq_video::render::wgpu`, so you name the exact version rather than guessing at a compatible one. +On Linux, request `wgpu::Features::VULKAN_EXTERNAL_MEMORY_DMA_BUF` when creating +the device to activate the PipeWire DMA-BUF fast path. The renderer still works +without it and falls back to a CPU upload for linear allocations. + `Color` names the matrix and range (BT.601 or BT.709, limited or full), and the shader converts per frame rather than assuming one space. A capture labels what it produced, so a locally captured frame renders correctly with no help from you. diff --git a/rs/moq-native/Cargo.toml b/rs/moq-native/Cargo.toml index 34e39df7a..6b763b2ad 100644 --- a/rs/moq-native/Cargo.toml +++ b/rs/moq-native/Cargo.toml @@ -96,7 +96,7 @@ webpki-roots = "1" [target.'cfg(unix)'.dependencies] # Errno constants for classifying an `accept(2)` failure (`accept::Failure`). # Their numeric values differ across unix platforms, so they can't be spelled out. -libc = "0.2" +libc = { workspace = true } [dev-dependencies] anyhow = "1" diff --git a/rs/moq-video/Cargo.toml b/rs/moq-video/Cargo.toml index 0b6757402..48f3e63ee 100644 --- a/rs/moq-video/Cargo.toml +++ b/rs/moq-video/Cargo.toml @@ -37,13 +37,18 @@ nvdec = ["dep:cudarc", "dep:moq-nvenc", "dep:libloading"] # build: moq-vaapi dlopen's libva, so an enabled-vaapi build needs no libva at # build time and still starts on a libva-less host, where automatic backend # selection falls back to the next encoder like the NVENC backend does. -vaapi = ["dep:moq-vaapi"] +vaapi = ["dmabuf", "dep:moq-vaapi"] +# Linux DMA-BUF surface vocabulary and CPU fallback support. Enabled by every +# backend that can produce or consume one. It pulls no graphics API by itself. +dmabuf = ["dep:libc", "dep:linux-raw-sys"] # GPU rendering of decoded frames (the `render` module): a wgpu pipeline that # converts a frame to RGBA and hands back a texture the caller presents. Off by # default because wgpu pulls a graphics stack (and its backend drivers) that a # relay or a headless publisher has no use for; only an application that actually -# draws video turns it on. +# draws video turns it on. `dmabuf` adds the Linux surface vocabulary; Vulkan +# itself still comes from wgpu. render = [ + "dmabuf", "dep:bytemuck", "dep:wgpu", "dep:objc2-metal", @@ -55,7 +60,7 @@ render = [ # Off by default because the `pipewire` crate links libpipewire-0.3 via pkg-config # at build time (and its bindgen needs libclang), so it only belongs in builds # that actually capture displays. Camera capture (V4L2) needs nothing extra. -pipewire = ["capture", "dep:pipewire", "dep:ashpd"] +pipewire = ["capture", "dmabuf", "dep:pipewire", "dep:ashpd"] [dependencies] anyhow = "1" @@ -106,10 +111,16 @@ ashpd = { version = "0.13", optional = true, default-features = false, features # extra dependencies); libnvrtc itself is never loaded since we ship pre-built PTX # (see frame/nv12_resize.ptx) that the driver JIT-compiles. cudarc = { version = "0.19", optional = true, default-features = false, features = ["driver", "fallback-dynamic-loading", "cuda-12020", "nvrtc"] } +# Mapping a linear DMA-BUF for the universal CPU I420 fallback. Non-linear +# modifiers stay GPU-only and return an honest error instead of treating tiled +# memory as rows. +libc = { workspace = true, optional = true } # Probe for the NVIDIA driver libraries before calling cudarc / the NVENC SDK, # which panic (process-abort under release `panic = "abort"`) if their library is # absent. Lets a GPU-less host fall back to the next encoder instead of crashing. libloading = { version = "0.9", optional = true } +# Architecture-correct DMA_BUF_IOCTL_SYNC request values for CPU access. +linux-raw-sys = { workspace = true, optional = true } moq-nvenc = { workspace = true, optional = true } # Intel/AMD VAAPI hardware encoder, behind the opt-in `vaapi` feature. As of # moq-vaapi 0.0.3 libva is dlopen'd at runtime, so the binary carries no NEEDED diff --git a/rs/moq-video/src/capture/pipewire.rs b/rs/moq-video/src/capture/pipewire.rs index 72eeaf192..96d1bfe30 100644 --- a/rs/moq-video/src/capture/pipewire.rs +++ b/rs/moq-video/src/capture/pipewire.rs @@ -2,8 +2,9 @@ //! //! The ScreenCast portal owns source selection: [`open`] pops the compositor's //! picker dialog, the user chooses a monitor, and the portal hands us a PipeWire -//! fd + node id. A dedicated thread then runs the PipeWire main loop, converting -//! each packed RGB or NV12 frame to CPU [`I420`] and pushing it into the shared +//! fd + node id. A dedicated thread then runs the PipeWire main loop, forwarding +//! DMA-BUF frames without copying when the compositor offers them and converting +//! shared-memory frames to CPU [`I420`] otherwise. It pushes both into the shared //! [`FrameChannel`] (callback-driven like the macOS delegate, not a pull-style //! pump). //! @@ -21,7 +22,7 @@ use std::borrow::Cow; use std::cell::RefCell; -use std::os::fd::OwnedFd; +use std::os::fd::{AsFd, BorrowedFd, OwnedFd}; use std::rc::Rc; use std::sync::{Arc, Mutex}; use std::thread::JoinHandle; @@ -31,15 +32,18 @@ use ashpd::desktop::PersistMode; use ashpd::desktop::screencast::{CursorMode, Screencast, SelectSourcesOptions, SourceType}; use pipewire as pw; use pw::spa; +use spa::buffer::DataType; use spa::param::video::{VideoFormat, VideoInfoRaw}; use super::channel::FrameChannel; use super::pump::Geometry; use super::{Config, FrameStream}; -use crate::frame::{I420, Surface}; +use crate::frame::{DmaBuf, DmaBufFrame, DmaBufPlane, DrmFormat, I420, Surface, wait_dma_buf_readable}; use crate::{Color, Error, Size}; const DEFAULT_FRAMERATE: u32 = 30; +// libspa 0.10 omits this flag from its safe wrapper, but exposes the raw bits. +const CHUNK_FLAG_EMPTY: i32 = 1 << 1; /// The compositor sends the negotiated format right after the stream connects; /// if nothing arrives the session is broken (or the grant was revoked mid-setup). const FORMAT_TIMEOUT: Duration = Duration::from_secs(10); @@ -70,6 +74,7 @@ pub(super) async fn open(config: &Config, device: Option<&str>) -> Result(); + let (return_tx, return_rx) = pw::channel::channel::(); let handle = std::thread::spawn({ let chan = chan.clone(); @@ -81,8 +86,19 @@ pub(super) async fn open(config: &Config, device: Option<&str>) -> Result, geo_tx: Option>>, /// Most recent converted frame, re-emitted while the screen is static. - last: Option, + last: Option, /// Whether a fresh frame arrived since the last pacing tick. fresh: bool, + /// Buffer-pool generation. Returns from superseded pools are discarded. + generation: u64, + /// Explicit DRM modifier from the negotiated DMA-BUF format. + dmabuf_modifier: Option, +} + +/// A retained frame for the static-screen pacing tick. +enum Last { + I420(I420), + DmaBuf(DmaBuf), +} + +impl Last { + fn surface(&self) -> Surface { + match self { + Self::I420(frame) => Surface::I420(frame.clone()), + Self::DmaBuf(frame) => Surface::DmaBuf(frame.clone()), + } + } +} + +#[derive(Clone, Copy)] +struct FrameLayout { + stride: u32, + width: u32, + height: u32, + source_height: u32, +} + +/// A dequeued PipeWire DMA-BUF kept out of the producer's pool until every +/// frame clone drops. Duplicating the fd alone is not enough: it preserves the +/// allocation, but the compositor may overwrite its pixels as soon as the +/// original buffer is queued again. +struct PipeWireDmaBuf { + fd: OwnedFd, + return_tx: pw::channel::Sender, + lease: Lease, + map_offset: u32, + allocation_size: Option, + data_offset: usize, + layout: FrameLayout, + format: DrmFormat, + modifier: u64, + color: Option, +} + +impl Drop for PipeWireDmaBuf { + fn drop(&mut self) { + // A failed send means the stream and its pool have already been destroyed. + // The duplicated fd still closes normally; the stale pointer is never used. + let _ = self.return_tx.send(self.lease); + } +} + +impl DmaBufFrame for PipeWireDmaBuf { + fn export(&self) -> std::io::Result { + self.fd.as_fd().try_clone_to_owned() + } + + fn download_i420(&self) -> Result { + // DRM_FORMAT_MOD_LINEAR is zero. A tiled allocation cannot be interpreted + // as strided rows; its fallback needs a GPU/VPP download instead. + if self.modifier != 0 { + return Err(Error::Codec(anyhow::anyhow!( + "cannot download DMA-BUF modifier {:#x} as linear rows", + self.modifier + ))); + } + + wait_dma_buf_readable(self.fd.as_fd()) + .map_err(|e| Error::Codec(anyhow::anyhow!("waiting for DMA-BUF producer: {e}")))?; + with_dma_buf_read(&self.fd, || { + let allocation_size = self + .allocation_size + .ok_or_else(|| Error::Codec(anyhow::anyhow!("DMA-BUF descriptor does not report a mappable size")))?; + let mapping = Mapping::new(&self.fd, self.map_offset, allocation_size)?; + let data = mapping + .as_slice() + .get(self.data_offset..) + .ok_or_else(|| Error::Codec(anyhow::anyhow!("DMA-BUF chunk starts outside its allocation")))?; + + match self.format { + DrmFormat::NV12 => { + let frame = nv12_to_i420(data, self.layout)?; + Ok(match self.color { + Some(color) => frame.with_color(color), + None => frame, + }) + } + DrmFormat::XRGB8888 | DrmFormat::ARGB8888 => { + I420::from_bgra(data, self.layout.stride, self.layout.width, self.layout.height) + } + DrmFormat::XBGR8888 | DrmFormat::ABGR8888 => { + I420::from_rgba(data, self.layout.stride, self.layout.width, self.layout.height) + } + other => Err(Error::Codec(anyhow::anyhow!( + "cannot download DMA-BUF format {:#x}", + other.as_raw() + ))), + } + }) + } +} + +const DMA_BUF_SYNC_READ: u64 = 1 << 0; +const DMA_BUF_SYNC_END: u64 = 1 << 2; + +#[repr(C)] +struct DmaBufSync { + flags: u64, +} + +/// Brackets CPU reads with the cache-coherency protocol required by DMA-BUF. +fn with_dma_buf_read(fd: &OwnedFd, read: impl FnOnce() -> Result) -> Result { + let sync = DmaBufRead::new(fd)?; + let result = read(); + let end = sync.finish(); + match (result, end) { + (Ok(value), Ok(())) => Ok(value), + (Err(err), _) => Err(err), + (Ok(_), Err(err)) => Err(err), + } +} + +struct DmaBufRead<'a> { + fd: &'a OwnedFd, + finished: bool, +} + +impl<'a> DmaBufRead<'a> { + fn new(fd: &'a OwnedFd) -> Result { + dma_buf_sync(fd, DMA_BUF_SYNC_READ)?; + Ok(Self { fd, finished: false }) + } + + fn finish(mut self) -> Result<(), Error> { + self.finished = true; + dma_buf_sync(self.fd, DMA_BUF_SYNC_READ | DMA_BUF_SYNC_END) + } +} + +impl Drop for DmaBufRead<'_> { + fn drop(&mut self) { + if !self.finished + && let Err(err) = dma_buf_sync(self.fd, DMA_BUF_SYNC_READ | DMA_BUF_SYNC_END) + { + tracing::warn!(%err, "ending DMA-BUF CPU access failed"); + } + } +} + +fn dma_buf_sync(fd: &OwnedFd, flags: u64) -> Result<(), Error> { + let mut sync = DmaBufSync { flags }; + loop { + // SAFETY: this is the DMA-BUF sync ioctl with its matching UAPI payload, + // and `fd` stays open for the duration of the call. + let result = unsafe { + libc::ioctl( + std::os::fd::AsRawFd::as_raw_fd(fd), + linux_raw_sys::ioctl::DMA_BUF_IOCTL_SYNC as libc::Ioctl, + &mut sync, + ) + }; + if result == 0 { + return Ok(()); + } + let err = std::io::Error::last_os_error(); + if err.kind() != std::io::ErrorKind::Interrupted { + return Err(Error::Codec(anyhow::anyhow!("DMA-BUF sync: {err}"))); + } + } +} + +fn dma_buf_allocation_size(fd: BorrowedFd<'_>, map_offset: u32) -> std::io::Result> { + let raw = std::os::fd::AsRawFd::as_raw_fd(&fd); + dma_buf_allocation_size_with_seek(map_offset, |offset, whence| seek_fd(raw, offset, whence)) +} + +fn dma_buf_allocation_size_with_seek( + map_offset: u32, + mut seek: impl FnMut(libc::off_t, libc::c_int) -> std::io::Result, +) -> std::io::Result> { + // DMA-BUF supports SEEK_END specifically so userspace can discover the + // allocation size even though fstat commonly reports zero. + let end = seek(0, libc::SEEK_END)?; + // DMA-BUF also supports SEEK_SET. Reset the shared file description to the + // canonical position before handing the descriptor to another subsystem. + seek(0, libc::SEEK_SET)?; + let Ok(end) = usize::try_from(end) else { + return Ok(None); + }; + Ok(end.checked_sub(map_offset as usize).filter(|size| *size > 0)) +} + +fn seek_fd(fd: std::os::fd::RawFd, offset: libc::off_t, whence: libc::c_int) -> std::io::Result { + loop { + // SAFETY: the caller keeps `fd` open for the duration of this syscall. + let result = unsafe { libc::lseek(fd, offset, whence) }; + if result >= 0 { + return Ok(result); + } + let error = std::io::Error::last_os_error(); + if error.kind() != std::io::ErrorKind::Interrupted { + return Err(error); + } + } +} + +/// Read-only mmap of one linear DMA-BUF allocation. +struct Mapping { + ptr: *mut libc::c_void, + len: usize, +} + +impl Mapping { + fn new(fd: &OwnedFd, offset: u32, len: usize) -> Result { + if len == 0 { + return Err(Error::Codec(anyhow::anyhow!("cannot map an empty DMA-BUF"))); + } + // SAFETY: `fd` stays open for the mapping's lifetime, `len` is non-zero, + // and the kernel validates the PipeWire-provided offset and allocation. + let ptr = unsafe { + libc::mmap( + std::ptr::null_mut(), + len, + libc::PROT_READ, + libc::MAP_SHARED, + std::os::fd::AsRawFd::as_raw_fd(fd), + offset as libc::off_t, + ) + }; + if ptr == libc::MAP_FAILED { + return Err(Error::Codec(anyhow::anyhow!( + "DMA-BUF mmap: {}", + std::io::Error::last_os_error() + ))); + } + Ok(Self { ptr, len }) + } + + fn as_slice(&self) -> &[u8] { + // SAFETY: `ptr` and `len` came from a successful `mmap`, and `self` + // keeps that mapping alive for the returned slice's lifetime. + unsafe { std::slice::from_raw_parts(self.ptr.cast(), self.len) } + } +} + +impl Drop for Mapping { + fn drop(&mut self) { + // SAFETY: this exact pointer and length came from the successful `mmap` + // in `Mapping::new`, and the mapping is released exactly once here. + unsafe { + libc::munmap(self.ptr, self.len); + } + } +} + +/// Deinterleave strided NV12 into the crate's tightly packed I420 layout. +fn nv12_to_i420(data: &[u8], layout: FrameLayout) -> Result { + let (stride, width, height, source_height) = ( + layout.stride as usize, + layout.width as usize, + layout.height as usize, + layout.source_height as usize, + ); + if source_height < height { + return Err(Error::Codec(anyhow::anyhow!( + "NV12 source is shorter than the cropped output" + ))); + } + let y_len = stride + .checked_mul(source_height) + .ok_or_else(|| Error::Codec(anyhow::anyhow!("NV12 luma size overflow")))?; + let uv_rows = height / 2; + let uv_len = uv_rows + .checked_sub(1) + .and_then(|rows| rows.checked_mul(stride)) + .and_then(|offset| offset.checked_add(width)) + .ok_or_else(|| Error::Codec(anyhow::anyhow!("NV12 chroma size overflow")))?; + let frame_len = y_len + .checked_add(uv_len) + .ok_or_else(|| Error::Codec(anyhow::anyhow!("NV12 frame size overflow")))?; + if stride < width || data.len() < frame_len { + return Err(Error::Codec(anyhow::anyhow!( + "NV12 frame is shorter than its declared rows" + ))); + } + + let mut packed = vec![0; width * height * 3 / 2]; + for row in 0..height { + packed[row * width..(row + 1) * width].copy_from_slice(&data[row * stride..row * stride + width]); + } + let uv = &data[y_len..frame_len]; + let packed_uv = width * height; + for row in 0..uv_rows { + packed[packed_uv + row * width..packed_uv + (row + 1) * width] + .copy_from_slice(&uv[row * stride..row * stride + width]); + } + I420::from_nv12(&packed, width as u32, height as u32) +} + +/// Queues a raw PipeWire buffer unless ownership is transferred to a DMA-BUF +/// surface. This mirrors `pipewire::buffer::Buffer` while allowing the surface +/// to return the buffer asynchronously on the PipeWire loop thread. +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +struct Lease { + buffer: usize, + generation: u64, +} + +impl Lease { + fn current(self, generation: u64) -> Option<*mut pw::sys::pw_buffer> { + (self.generation == generation).then_some(self.buffer as *mut pw::sys::pw_buffer) + } +} + +struct Dequeued<'a> { + stream: &'a pw::stream::Stream, + raw: *mut pw::sys::pw_buffer, + queue: bool, +} + +impl<'a> Dequeued<'a> { + unsafe fn new(stream: &'a pw::stream::Stream) -> Option { + // SAFETY: PipeWire permits one dequeue from this process callback. The + // wrapper below returns the pointer to this same stream unless leased. + let raw = unsafe { stream.dequeue_raw_buffer() }; + (!raw.is_null()).then_some(Self { + stream, + raw, + queue: true, + }) + } + + fn datas_mut(&mut self) -> &mut [spa::buffer::Data] { + // SAFETY: `raw` is a non-null buffer dequeued from `stream` and remains + // out of PipeWire's pool while this wrapper owns it. + let buffer = unsafe { (*self.raw).buffer }; + if buffer.is_null() || unsafe { (*buffer).n_datas == 0 || (*buffer).datas.is_null() } { + return &mut []; + } + // SAFETY: PipeWire owns this SPA buffer and guarantees its `datas` array + // has `n_datas` entries until the buffer is queued back. + unsafe { + std::slice::from_raw_parts_mut((*buffer).datas.cast::(), (*buffer).n_datas as usize) + } + } + + fn lease(mut self, generation: u64) -> Lease { + self.queue = false; + Lease { + buffer: self.raw as usize, + generation, + } + } } -/// Connect to the portal's PipeWire node and run the main loop until the stream -/// ends, `quit_rx` fires (the `FrameStream` dropped), or the format changes. -fn run_loop( +impl Drop for Dequeued<'_> { + fn drop(&mut self) { + if self.queue { + // SAFETY: the pointer came from `self.stream` and has not previously + // been returned or transferred to a leased surface. + unsafe { self.stream.queue_raw_buffer(self.raw) }; + } + } +} + +struct CaptureLoop { fd: OwnedFd, node_id: u32, framerate: u32, chan: Arc, state: Rc>, quit_rx: pw::channel::Receiver<()>, -) -> Result<(), Error> { + return_rx: pw::channel::Receiver, + return_tx: pw::channel::Sender, +} + +#[derive(Debug, PartialEq, Eq)] +enum NegotiatedMemory { + Fixating, + SharedMemory, + DmaBuf(u64), +} + +fn negotiated_memory(param: &spa::pod::Pod, format: VideoInfoRaw) -> Option { + let object = param.as_object().ok()?; + let modifier = object.find_prop(spa::utils::Id( + spa::param::format::FormatProperties::VideoModifier.as_raw(), + )); + Some(match modifier { + Some(modifier) if modifier.flags().contains(spa::pod::PodPropFlags::DONT_FIXATE) => NegotiatedMemory::Fixating, + Some(_) => NegotiatedMemory::DmaBuf(format.modifier()), + None => NegotiatedMemory::SharedMemory, + }) +} + +fn fixate_modifier(param: &spa::pod::Pod, modifier: u64) -> Option> { + let (_, value) = spa::pod::deserialize::PodDeserializer::deserialize_any_from(param.as_bytes()).ok()?; + let spa::pod::Value::Object(mut object) = value else { + return None; + }; + let property = object + .properties + .iter_mut() + .find(|property| property.key == spa::param::format::FormatProperties::VideoModifier.as_raw())?; + property.flags.remove(spa::pod::PropertyFlags::from_bits_retain( + spa::sys::SPA_POD_PROP_FLAG_DONT_FIXATE, + )); + property.value = spa::pod::Value::Long(modifier as i64); + spa::pod::serialize::PodSerializer::serialize(std::io::Cursor::new(Vec::new()), &spa::pod::Value::Object(object)) + .ok() + .map(|serialized| serialized.0.into_inner()) +} + +/// Connect to the portal's PipeWire node and run until the stream ends, the +/// consumer drops, or the format changes. +fn run_loop(args: CaptureLoop) -> Result<(), Error> { + let CaptureLoop { + fd, + node_id, + framerate, + chan, + state, + quit_rx, + return_rx, + return_tx, + } = args; pw::init(); let mainloop = pw::main_loop::MainLoopRc::new(None).map_err(|e| err("pipewire main loop", e))?; @@ -282,6 +714,23 @@ fn run_loop( ) .map_err(|e| err("pipewire stream", e))?; + // DMA-BUF surfaces return their dequeued PipeWire buffer here on the loop + // thread. A surface may drop on an encoder or renderer task, where calling + // `pw_stream_queue_buffer` directly would violate PipeWire's threading model. + let _returns = return_rx.attach(mainloop.loop_(), { + let stream = stream.downgrade(); + let state = state.clone(); + move |lease| { + if let Some(stream) = stream.upgrade() + && let Some(raw) = lease.current(state.borrow().generation) + { + // SAFETY: leased surfaces send each pointer exactly once, and every + // current-generation pointer was dequeued from this stream's pool. + unsafe { stream.queue_raw_buffer(raw) }; + } + } + }); + let _listener = stream .add_local_listener::<()>() .state_changed({ @@ -311,7 +760,7 @@ fn run_loop( .param_changed({ let state = state.clone(); let mainloop = mainloop.downgrade(); - move |_, _, id, param| { + move |stream, _, id, param| { let Some(param) = param else { return }; if id != spa::param::ParamType::Format.as_raw() { return; @@ -330,6 +779,38 @@ fn run_loop( tracing::warn!(error = %e, "failed to parse pipewire video format"); return; } + let Some(memory) = negotiated_memory(param, state.format) else { + return; + }; + let dmabuf_modifier = match memory { + NegotiatedMemory::DmaBuf(modifier) => Some(modifier), + NegotiatedMemory::SharedMemory => None, + NegotiatedMemory::Fixating => { + let Some(fixed) = fixate_modifier(param, state.format.modifier()) else { + tracing::warn!("failed to fixate the PipeWire DMA-BUF modifier"); + return; + }; + let Some(param) = spa::pod::Pod::from_bytes(&fixed) else { + return; + }; + let mut params = [param]; + if let Err(e) = stream.update_params(&mut params) { + tracing::warn!(error = %e, "failed to fixate the PipeWire DMA-BUF modifier"); + } + return; + } + }; + state.last = None; + state.dmabuf_modifier = dmabuf_modifier; + let buffers = buffer_offer(state.dmabuf_modifier.is_some()); + if let Some(param) = spa::pod::Pod::from_bytes(&buffers) { + let mut params = [param]; + if let Err(e) = stream.update_params(&mut params) { + tracing::warn!(error = %e, "DMA-BUF buffer negotiation failed; capture may use shared memory"); + } else { + state.generation = state.generation.wrapping_add(1); + } + } let size = state.format.size(); // I420 chroma is 2x2 subsampled; clamp down (drop the last odd @@ -382,44 +863,46 @@ fn run_loop( let state = state.clone(); let chan = chan.clone(); let mainloop = mainloop.downgrade(); + let return_tx = return_tx.clone(); move |stream, _| { let mut state = state.borrow_mut(); let Some((width, height)) = state.geometry else { return }; let source_height = state.format.size().height; let color = state.color; - let Some(mut buffer) = stream.dequeue_buffer() else { + // SAFETY: this is PipeWire's process callback for `stream`; `Dequeued` + // returns the buffer on drop unless a DMA-BUF surface leases it. + let Some(mut buffer) = (unsafe { Dequeued::new(stream) }) else { return; }; let datas = buffer.datas_mut(); - let Some(data) = datas.first_mut() else { return }; - - let maxsize = data.as_raw().maxsize; - let size = clamp_chunk_size(data.chunk().size(), maxsize); - match chunk_kind(size, data.chunk().flags()) { - ChunkKind::Data => {} - ChunkKind::Empty => { - let Ok(frame) = neutral_frame(width, height, color) else { - return; - }; - chan.push(Surface::I420(frame.clone())); - state.last = Some(frame); - state.fresh = true; - return; - } - ChunkKind::Invalid => return, - } - let Some(offset) = normalize_chunk_offset(data.chunk().offset(), maxsize) else { - tracing::warn!("pipewire buffer has zero maximum size"); + let [data] = datas else { + tracing::warn!( + blocks = datas.len(), + "pipewire ignored the negotiated single-block layout" + ); return; }; - // Fall back to the unclamped source width: for an odd-width source - // the real row is one pixel wider than the clamped `width`. The - // packing differs per format, so NV12 cannot assume 4 bytes/pixel. + + let chunk_offset = data.chunk().offset(); + let (map_offset, maxsize) = { + let raw = data.as_raw(); + (raw.mapoffset, raw.maxsize) + }; + let dmabuf = data.type_() == DataType::DmaBuf; + let chunk_size = data.chunk().size(); + let size = if dmabuf { + dma_buf_chunk_size(chunk_size, maxsize) + } else { + clamp_chunk_size(chunk_size, maxsize) + }; let stride = match u32::try_from(data.chunk().stride()) { Ok(stride) if stride > 0 => stride, _ => match state.format.format() { VideoFormat::NV12 => state.format.size().width, - _ => state.format.size().width.saturating_mul(4), + VideoFormat::BGRx | VideoFormat::BGRA | VideoFormat::RGBx | VideoFormat::RGBA => { + state.format.size().width.saturating_mul(4) + } + _ => 0, }, }; let layout = FrameLayout { @@ -428,15 +911,110 @@ fn run_loop( height, source_height, }; - - // The chunk only says how many bytes the producer wrote. Check it - // actually spans every row `convert` will sample, so a short or - // mislabeled buffer is dropped here rather than read past. + match chunk_kind(size, data.chunk().flags(), dmabuf) { + ChunkKind::Data => {} + // An empty screencast chunk means there is no new damage. Keep + // the previous frame for the pacing timer instead of blanking it. + ChunkKind::Empty => return, + ChunkKind::Invalid => return, + } let Some(required) = frame_data_size(state.format.format(), layout) else { tracing::warn!("pipewire frame layout overflows its buffer"); return; }; - if required > size || required > maxsize as usize { + if dmabuf { + let Some(format) = drm_format(state.format.format()) else { + tracing::warn!(format = ?state.format.format(), "unsupported DMA-BUF pixel format"); + return; + }; + if data.fd() < 0 || stride == 0 { + tracing::warn!("DMA-BUF has no valid fd or row stride"); + return; + } + // SAFETY: PipeWire owns this fd while the buffer is dequeued. It is + // duplicated immediately, before the buffer can be queued again. + let fd = unsafe { BorrowedFd::borrow_raw(data.fd()) }; + let Ok(fd) = fd.try_clone_to_owned() else { + return; + }; + let offset = normalize_dma_buf_offset(chunk_offset, maxsize); + if !dma_buf_chunk_contains(required, chunk_size, maxsize) { + tracing::warn!( + required, + available = size, + "DMA-BUF chunk does not contain a complete frame" + ); + return; + } + let allocation_size = match dma_buf_allocation_size(fd.as_fd(), map_offset) { + Ok(Some(size)) if offset.checked_add(required).is_some_and(|end| end <= size) => Some(size), + Ok(Some(size)) => { + tracing::warn!( + required, + available = size, + "DMA-BUF allocation is shorter than its frame layout" + ); + return; + } + Ok(None) => None, + Err(error) => { + tracing::debug!(%error, "could not query DMA-BUF allocation size; CPU fallback is unavailable"); + None + } + }; + let Some(base) = map_offset.checked_add(offset as u32) else { + tracing::warn!("DMA-BUF plane offset overflow"); + return; + }; + let planes = if format == DrmFormat::NV12 { + let Some(uv) = layout + .stride + .checked_mul(layout.source_height) + .and_then(|size| base.checked_add(size)) + else { + tracing::warn!("DMA-BUF chroma plane offset overflow"); + return; + }; + vec![ + DmaBufPlane::new(base, layout.stride), + DmaBufPlane::new(uv, layout.stride), + ] + } else { + vec![DmaBufPlane::new(base, layout.stride)] + }; + let Some(modifier) = state.dmabuf_modifier else { + tracing::warn!("received a DMA-BUF without a negotiated modifier"); + return; + }; + let lease = buffer.lease(state.generation); + let inner = Arc::new(PipeWireDmaBuf { + fd, + return_tx: return_tx.clone(), + lease, + map_offset, + allocation_size, + data_offset: offset, + layout, + format, + modifier, + color, + }); + match DmaBuf::new(format, modifier, layout.width, layout.height, planes, color, inner) { + Ok(frame) => { + chan.push(Surface::DmaBuf(frame.clone())); + state.last = Some(Last::DmaBuf(frame)); + state.fresh = true; + } + Err(e) => tracing::warn!(error = %e, "invalid PipeWire DMA-BUF"), + } + return; + } + let allocation_size = maxsize as usize; + let Some(offset) = normalize_chunk_offset(chunk_offset, maxsize) else { + tracing::warn!("pipewire buffer has zero maximum size"); + return; + }; + if required > size || required > allocation_size { tracing::warn!( required, available = size, @@ -444,10 +1022,8 @@ fn run_loop( ); return; } - - // Without dmabuf modifiers in our format offer the compositor uses - // shared memory, which MAP_BUFFERS mmaps for us; `None` here means - // it forced something we can't read, so give up cleanly. + // MAP_BUFFERS maps MemPtr/MemFd for the CPU fallback. `None` here + // means the producer forced an unsupported buffer representation. let Some(bytes) = data.data() else { tracing::warn!("pipewire buffer is not CPU-mapped; stopping capture"); if let Some(mainloop) = mainloop.upgrade() { @@ -455,17 +1031,16 @@ fn run_loop( } return; }; - let Some(allocation) = bytes.get(..maxsize as usize) else { + let Some(allocation) = bytes.get(..allocation_size) else { return; }; let Some(bytes) = chunk_bytes(allocation, offset, required) else { return; }; - match convert(state.format.format(), bytes.as_ref(), layout, color) { Ok(i420) => { chan.push(Surface::I420(i420.clone())); - state.last = Some(i420); + state.last = Some(Last::I420(i420)); state.fresh = true; } Err(e) => { @@ -481,12 +1056,16 @@ fn run_loop( .register() .map_err(|e| err("pipewire listener", e))?; - // Offer the CPU-convertible RGB layouts; the compositor picks one and - // replies through `param_changed`. No dmabuf modifiers, so buffers stay in - // shared memory (the CPU path, like the other non-macOS backends). - let pod = format_offer(framerate); - let mut params = [spa::pod::Pod::from_bytes(&pod) - .ok_or_else(|| Error::Codec(anyhow::anyhow!("failed to build pipewire format offer")))?]; + // DMA-BUF formats come first so PipeWire prefers them. Each has a matching + // shared-memory offer without a modifier as the required fallback. + let offers = format_offers(framerate); + let mut params = offers + .iter() + .map(|offer| { + spa::pod::Pod::from_bytes(offer) + .ok_or_else(|| Error::Codec(anyhow::anyhow!("failed to build pipewire format offer"))) + }) + .collect::, _>>()?; stream .connect( spa::utils::Direction::Input, @@ -507,7 +1086,7 @@ fn run_loop( return; } if let Some(last) = &state.last { - chan.push(Surface::I420(last.clone())); + chan.push(last.surface()); } } }); @@ -531,50 +1110,91 @@ fn run_loop( Ok(()) } -/// The geometry `convert` needs: the producer's row stride and source height, -/// plus the even-clamped output size. -#[derive(Clone, Copy)] -struct FrameLayout { - stride: u32, - width: u32, - height: u32, - /// Unclamped source height. NV12's chroma plane starts after this many luma - /// rows, which is one more than `height` for an odd-height source. - source_height: u32, +fn normalize_chunk_offset(offset: u32, maxsize: u32) -> Option { + (maxsize != 0).then(|| (offset % maxsize) as usize) } -const CHUNK_FLAG_EMPTY: i32 = 1 << 1; +fn normalize_dma_buf_offset(offset: u32, maxsize: u32) -> usize { + normalize_chunk_offset(offset, maxsize).unwrap_or(offset as usize) +} -#[derive(Debug, PartialEq, Eq)] -enum ChunkKind { - Data, - Empty, - Invalid, +fn clamp_chunk_size(size: u32, maxsize: u32) -> usize { + size.min(maxsize) as usize } -fn chunk_kind(size: usize, flags: spa::buffer::ChunkFlags) -> ChunkKind { - if flags.contains(spa::buffer::ChunkFlags::CORRUPTED) { - ChunkKind::Invalid - } else if flags.bits() & CHUNK_FLAG_EMPTY != 0 { - ChunkKind::Empty - } else if size == 0 { - ChunkKind::Invalid +fn dma_buf_chunk_size(size: u32, maxsize: u32) -> usize { + if maxsize == 0 { + size as usize } else { - ChunkKind::Data + clamp_chunk_size(size, maxsize) } } -fn neutral_frame(width: u32, height: u32, color: Option) -> Result { - let color = color.unwrap_or_else(|| Color::infer(Size::new(width, height))); - let luma = width as usize * height as usize; - let mut data = vec![128; I420::len(width, height)]; - data[..luma].fill(if color.limited() { 16 } else { 0 }); - Ok(I420::new(width, height, data)?.with_color(color)) +fn dma_buf_chunk_contains(required: usize, size: u32, maxsize: u32) -> bool { + size == 0 || required <= dma_buf_chunk_size(size, maxsize) } -/// Parse into a zeroed value before replacing the current format. libspa leaves -/// omitted optional properties untouched, so parsing into the reused value -/// would retain stale color metadata across renegotiation. +fn chunk_bytes(data: &[u8], offset: usize, size: usize) -> Option> { + if offset >= data.len() || size > data.len() { + return None; + } + let end = offset.checked_add(size)?; + if end <= data.len() { + return Some(Cow::Borrowed(&data[offset..end])); + } + + let mut wrapped = Vec::with_capacity(size); + wrapped.extend_from_slice(&data[offset..]); + let head = size - wrapped.len(); + wrapped.extend_from_slice(&data[..head]); + Some(Cow::Owned(wrapped)) +} + +fn frame_data_size(format: VideoFormat, layout: FrameLayout) -> Option { + let stride = layout.stride as usize; + let width = layout.width as usize; + let height = layout.height as usize; + let row_size = match format { + VideoFormat::NV12 => width, + VideoFormat::BGRx | VideoFormat::BGRA | VideoFormat::RGBx | VideoFormat::RGBA => width.checked_mul(4)?, + _ => return None, + }; + if stride < row_size { + return None; + } + + match format { + VideoFormat::NV12 => (layout.source_height as usize) + .checked_add(height / 2)? + .checked_sub(1)? + .checked_mul(stride)? + .checked_add(row_size), + _ => height.checked_sub(1)?.checked_mul(stride)?.checked_add(row_size), + } +} + +#[derive(Debug, PartialEq, Eq)] +enum ChunkKind { + Data, + Empty, + Invalid, +} + +fn chunk_kind(size: usize, flags: spa::buffer::ChunkFlags, dmabuf: bool) -> ChunkKind { + if flags.contains(spa::buffer::ChunkFlags::CORRUPTED) { + ChunkKind::Invalid + } else if flags.bits() & CHUNK_FLAG_EMPTY != 0 { + ChunkKind::Empty + } else if size == 0 && !dmabuf { + ChunkKind::Invalid + } else { + ChunkKind::Data + } +} + +/// Parse into a zeroed value before replacing the current format. libspa leaves +/// omitted optional properties untouched, so parsing into the reused value +/// would retain stale color metadata across renegotiation. fn replace_video_format( current: &mut VideoInfoRaw, parse: impl FnOnce(&mut VideoInfoRaw) -> Result, @@ -671,102 +1291,6 @@ fn format_requires_restart( geometry.is_some_and(|geometry| geometry != (width, height) || color != next_color) } -/// A chunk offset is a ring position within the allocation, so wrap it rather -/// than trusting it to be in range. `None` means the allocation is unusable. -fn normalize_chunk_offset(offset: u32, maxsize: u32) -> Option { - (maxsize != 0).then(|| (offset % maxsize) as usize) -} - -fn clamp_chunk_size(size: u32, maxsize: u32) -> usize { - size.min(maxsize) as usize -} - -fn chunk_bytes(data: &[u8], offset: usize, size: usize) -> Option> { - if offset >= data.len() || size > data.len() { - return None; - } - let end = offset.checked_add(size)?; - if end <= data.len() { - return Some(Cow::Borrowed(&data[offset..end])); - } - - let mut wrapped = Vec::with_capacity(size); - wrapped.extend_from_slice(&data[offset..]); - let head = size - wrapped.len(); - wrapped.extend_from_slice(&data[..head]); - Some(Cow::Owned(wrapped)) -} - -/// Bytes from the chunk start through the span required by `convert`. NV12 stops -/// at the visible width of its final row; packed RGB requires every full stride. -fn frame_data_size(format: VideoFormat, layout: FrameLayout) -> Option { - let stride = layout.stride as usize; - let width = layout.width as usize; - let height = layout.height as usize; - let row_size = match format { - VideoFormat::NV12 => width, - VideoFormat::BGRx | VideoFormat::BGRA | VideoFormat::RGBx | VideoFormat::RGBA => width.checked_mul(4)?, - _ => return None, - }; - if stride < row_size { - return None; - } - - match format { - // Chroma follows every source luma row, then runs at half height. - VideoFormat::NV12 => (layout.source_height as usize) - .checked_add(height / 2)? - .checked_sub(1)? - .checked_mul(stride)? - .checked_add(row_size), - _ => stride.checked_mul(height), - } -} - -/// Deinterleave strided NV12 into the crate's tightly packed I420 layout. -fn nv12_to_i420(data: &[u8], layout: FrameLayout) -> Result { - let (stride, width, height, source_height) = ( - layout.stride as usize, - layout.width as usize, - layout.height as usize, - layout.source_height as usize, - ); - if source_height < height { - return Err(Error::Codec(anyhow::anyhow!( - "NV12 source is shorter than the cropped output" - ))); - } - let y_len = stride - .checked_mul(source_height) - .ok_or_else(|| Error::Codec(anyhow::anyhow!("NV12 luma size overflow")))?; - let uv_rows = height / 2; - let uv_len = uv_rows - .checked_sub(1) - .and_then(|rows| rows.checked_mul(stride)) - .and_then(|offset| offset.checked_add(width)) - .ok_or_else(|| Error::Codec(anyhow::anyhow!("NV12 chroma size overflow")))?; - let frame_len = y_len - .checked_add(uv_len) - .ok_or_else(|| Error::Codec(anyhow::anyhow!("NV12 frame size overflow")))?; - if stride < width || data.len() < frame_len { - return Err(Error::Codec(anyhow::anyhow!( - "NV12 frame is shorter than its declared rows" - ))); - } - - let mut packed = vec![0; I420::len(width as u32, height as u32)]; - for row in 0..height { - packed[row * width..(row + 1) * width].copy_from_slice(&data[row * stride..row * stride + width]); - } - let uv = &data[y_len..frame_len]; - let packed_uv = width * height; - for row in 0..uv_rows { - packed[packed_uv + row * width..packed_uv + (row + 1) * width] - .copy_from_slice(&uv[row * stride..row * stride + width]); - } - I420::from_nv12(&packed, width as u32, height as u32) -} - /// Convert one strided screen frame to tightly-packed I420. fn convert(format: VideoFormat, bytes: &[u8], layout: FrameLayout, color: Option) -> Result { match format { @@ -785,10 +1309,29 @@ fn convert(format: VideoFormat, bytes: &[u8], layout: FrameLayout, color: Option } } -/// Serialize the `EnumFormat` pod offering the layouts we can convert (packed -/// RGB first, then NV12), any size, and a framerate range preferring `framerate`. -fn format_offer(framerate: u32) -> Vec { - let obj = spa::pod::object!( +/// Map the negotiated SPA layout to its DRM fourcc. +fn drm_format(format: VideoFormat) -> Option { + match format { + VideoFormat::NV12 => Some(DrmFormat::NV12), + VideoFormat::BGRx => Some(DrmFormat::XRGB8888), + VideoFormat::BGRA => Some(DrmFormat::ARGB8888), + VideoFormat::RGBx => Some(DrmFormat::XBGR8888), + VideoFormat::RGBA => Some(DrmFormat::ABGR8888), + _ => None, + } +} + +const PIPEWIRE_FORMATS: [VideoFormat; 5] = [ + VideoFormat::BGRx, + VideoFormat::BGRA, + VideoFormat::RGBx, + VideoFormat::RGBA, + VideoFormat::NV12, +]; + +/// Serialize one `EnumFormat` pod for a concrete pixel format. +fn format_offer(framerate: u32, format: VideoFormat, dmabuf: bool) -> Vec { + let mut obj = spa::pod::object!( spa::utils::SpaTypes::ObjectParamFormat, spa::param::ParamType::EnumFormat, spa::pod::property!( @@ -801,18 +1344,7 @@ fn format_offer(framerate: u32) -> Vec { Id, spa::param::format::MediaSubtype::Raw ), - spa::pod::property!( - spa::param::format::FormatProperties::VideoFormat, - Choice, - Enum, - Id, - VideoFormat::BGRx, - VideoFormat::BGRx, - VideoFormat::BGRA, - VideoFormat::RGBx, - VideoFormat::RGBA, - VideoFormat::NV12, - ), + spa::pod::property!(spa::param::format::FormatProperties::VideoFormat, Id, format), spa::pod::property!( spa::param::format::FormatProperties::VideoSize, Choice, @@ -841,79 +1373,284 @@ fn format_offer(framerate: u32) -> Vec { spa::utils::Fraction { num: 1000, denom: 1 } ), ); + if dmabuf { + obj.properties.push(spa::pod::Property { + key: spa::param::format::FormatProperties::VideoModifier.as_raw(), + flags: spa::pod::PropertyFlags::from_bits_retain( + spa::sys::SPA_POD_PROP_FLAG_MANDATORY | spa::sys::SPA_POD_PROP_FLAG_DONT_FIXATE, + ), + value: spa::pod::Value::Choice(spa::pod::ChoiceValue::Long(spa::utils::Choice( + spa::utils::ChoiceFlags::empty(), + spa::utils::ChoiceEnum::Enum { + default: 0, + alternatives: vec![0], + }, + ))), + }); + } spa::pod::serialize::PodSerializer::serialize(std::io::Cursor::new(Vec::new()), &spa::pod::Value::Object(obj)) .expect("serializing a static format pod cannot fail") .0 .into_inner() } +/// Serialize DMA-BUF formats first and shared-memory fallbacks second. +fn format_offers(framerate: u32) -> Vec> { + PIPEWIRE_FORMATS + .into_iter() + .map(|format| format_offer(framerate, format, true)) + .chain( + PIPEWIRE_FORMATS + .into_iter() + .map(|format| format_offer(framerate, format, false)), + ) + .collect() +} + +/// Serialize `SPA_PARAM_Buffers` for the negotiated memory representation. +fn buffer_offer(dmabuf: bool) -> Vec { + let mem_ptr = 1 << DataType::MemPtr.as_raw(); + let mem_fd = 1 << DataType::MemFd.as_raw(); + let dma_buf = 1 << DataType::DmaBuf.as_raw(); + let data_types = if dmabuf { dma_buf } else { mem_fd | mem_ptr }; + + let obj = spa::pod::Object { + type_: spa::utils::SpaTypes::ObjectParamBuffers.as_raw(), + id: spa::param::ParamType::Buffers.as_raw(), + properties: vec![ + spa::pod::Property::new( + spa::sys::SPA_PARAM_BUFFERS_buffers, + spa::pod::Value::Choice(spa::pod::ChoiceValue::Int(spa::utils::Choice( + spa::utils::ChoiceFlags::empty(), + spa::utils::ChoiceEnum::Range { + default: 8, + min: 2, + max: 64, + }, + ))), + ), + spa::pod::Property::new(spa::sys::SPA_PARAM_BUFFERS_blocks, spa::pod::Value::Int(1)), + spa::pod::Property::new( + spa::sys::SPA_PARAM_BUFFERS_dataType, + spa::pod::Value::Choice(spa::pod::ChoiceValue::Int(spa::utils::Choice( + spa::utils::ChoiceFlags::empty(), + spa::utils::ChoiceEnum::Flags { + default: data_types, + flags: Vec::new(), + }, + ))), + ), + ], + }; + spa::pod::serialize::PodSerializer::serialize(std::io::Cursor::new(Vec::new()), &spa::pod::Value::Object(obj)) + .expect("serializing a static buffer pod cannot fail") + .0 + .into_inner() +} + #[cfg(test)] mod tests { use super::*; use crate::capture::Config; - /// The serialized format offer must parse back as a valid pod, carrying every - /// layout `convert` knows how to handle. + /// The serialized format offer must parse back as a valid pod. #[test] fn format_offer_is_valid_pod() { - let bytes = format_offer(30); - let (remaining, value) = spa::pod::deserialize::PodDeserializer::deserialize_any_from(&bytes) - .expect("format offer did not round-trip"); - assert!(remaining.is_empty()); - let spa::pod::Value::Object(object) = value else { - panic!("format offer is not an object"); - }; - let property = object - .properties - .iter() - .find(|property| property.key == spa::param::format::FormatProperties::VideoFormat.as_raw()) - .expect("missing video format property"); - assert_eq!( - property.value, - spa::pod::Value::Choice(spa::pod::ChoiceValue::Id(spa::utils::Choice( - spa::utils::ChoiceFlags::empty(), - spa::utils::ChoiceEnum::Enum { - default: spa::utils::Id(VideoFormat::BGRx.as_raw()), - alternatives: vec![ - spa::utils::Id(VideoFormat::BGRx.as_raw()), - spa::utils::Id(VideoFormat::BGRA.as_raw()), - spa::utils::Id(VideoFormat::RGBx.as_raw()), - spa::utils::Id(VideoFormat::RGBA.as_raw()), - spa::utils::Id(VideoFormat::NV12.as_raw()), - ], - }, - ))) - ); + let offers = format_offers(30); + assert_eq!(offers.len(), PIPEWIRE_FORMATS.len() * 2); + for (index, bytes) in offers.iter().enumerate() { + let (remaining, value) = spa::pod::deserialize::PodDeserializer::deserialize_any_from(bytes) + .expect("format offer did not round-trip"); + assert!(remaining.is_empty()); + let spa::pod::Value::Object(object) = value else { + panic!("format offer is not an object"); + }; + let property = |key| object.properties.iter().find(|property| property.key == key); + let format = PIPEWIRE_FORMATS[index % PIPEWIRE_FORMATS.len()]; + assert_eq!( + property(spa::param::format::FormatProperties::VideoFormat.as_raw()).map(|p| &p.value), + Some(&spa::pod::Value::Id(spa::utils::Id(format.as_raw()))) + ); + let modifier = property(spa::param::format::FormatProperties::VideoModifier.as_raw()); + if index < PIPEWIRE_FORMATS.len() { + let modifier = modifier.expect("DMA-BUF offer has no modifier"); + assert_eq!( + modifier.flags.bits(), + spa::sys::SPA_POD_PROP_FLAG_MANDATORY | spa::sys::SPA_POD_PROP_FLAG_DONT_FIXATE + ); + assert_eq!( + modifier.value, + spa::pod::Value::Choice(spa::pod::ChoiceValue::Long(spa::utils::Choice( + spa::utils::ChoiceFlags::empty(), + spa::utils::ChoiceEnum::Enum { + default: 0, + alternatives: vec![0], + }, + ))) + ); + } else { + assert!(modifier.is_none(), "shared-memory offer contains a modifier"); + } + } + } + + #[test] + fn negotiated_modifier_must_be_present_and_fixed() { + let shared = format_offer(30, VideoFormat::BGRx, false); + let shared = spa::pod::Pod::from_bytes(&shared).unwrap(); + let mut format = VideoInfoRaw::default(); + format.parse(shared).unwrap(); + assert_eq!(negotiated_memory(shared, format), Some(NegotiatedMemory::SharedMemory)); + + let offered = format_offer(30, VideoFormat::BGRx, true); + let offered = spa::pod::Pod::from_bytes(&offered).unwrap(); + format.parse(offered).unwrap(); + assert_eq!(negotiated_memory(offered, format), Some(NegotiatedMemory::Fixating)); + + let fixed = fixate_modifier(offered, format.modifier()).unwrap(); + let fixed = spa::pod::Pod::from_bytes(&fixed).unwrap(); + replace_video_format(&mut format, |format| format.parse(fixed)).unwrap(); + assert_eq!(negotiated_memory(fixed, format), Some(NegotiatedMemory::DmaBuf(0))); + } + + #[test] + fn buffer_offer_is_valid_pod() { + let dma_buf = 1 << DataType::DmaBuf.as_raw(); + let mem_fd = 1 << DataType::MemFd.as_raw(); + let mem_ptr = 1 << DataType::MemPtr.as_raw(); + for (dmabuf, data_types) in [(true, dma_buf), (false, mem_fd | mem_ptr)] { + let bytes = buffer_offer(dmabuf); + let (remaining, value) = spa::pod::deserialize::PodDeserializer::deserialize_any_from(&bytes) + .expect("buffer offer did not round-trip"); + assert!(remaining.is_empty()); + let spa::pod::Value::Object(object) = value else { + panic!("buffer offer is not an object"); + }; + let property = |key| { + &object + .properties + .iter() + .find(|property| property.key == key) + .unwrap_or_else(|| panic!("missing buffer property {key}")) + .value + }; + assert_eq!( + property(spa::sys::SPA_PARAM_BUFFERS_buffers), + &spa::pod::Value::Choice(spa::pod::ChoiceValue::Int(spa::utils::Choice( + spa::utils::ChoiceFlags::empty(), + spa::utils::ChoiceEnum::Range { + default: 8, + min: 2, + max: 64, + }, + ))) + ); + assert_eq!(property(spa::sys::SPA_PARAM_BUFFERS_blocks), &spa::pod::Value::Int(1)); + assert_eq!( + property(spa::sys::SPA_PARAM_BUFFERS_dataType), + &spa::pod::Value::Choice(spa::pod::ChoiceValue::Int(spa::utils::Choice( + spa::utils::ChoiceFlags::empty(), + spa::utils::ChoiceEnum::Flags { + default: data_types, + flags: Vec::new(), + }, + ))) + ); + } } #[test] fn chunk_offset_wraps_to_the_allocation() { assert_eq!(normalize_chunk_offset(18, 16), Some(2)); assert_eq!(normalize_chunk_offset(0, 0), None); + assert_eq!(normalize_dma_buf_offset(18, 16), 2); + assert_eq!(normalize_dma_buf_offset(18, 0), 18); } #[test] fn chunk_size_is_clamped_to_the_allocation() { assert_eq!(clamp_chunk_size(18, 16), 16); assert_eq!(clamp_chunk_size(8, 16), 8); + assert_eq!(dma_buf_chunk_size(8, 16), 8); + assert_eq!(dma_buf_chunk_size(8, 0), 8); + assert!(dma_buf_chunk_contains(16, 0, 0)); + assert!(!dma_buf_chunk_contains(16, 8, 0)); + assert!(dma_buf_chunk_contains(16, 16, 16)); } #[test] - fn chunk_flags_distinguish_neutral_and_invalid_frames() { + fn frame_data_size_covers_every_sampled_row() { + let packed = FrameLayout { + stride: 20, + width: 4, + height: 2, + source_height: 2, + }; + assert_eq!(frame_data_size(VideoFormat::BGRx, packed), Some(36)); + + let nv12 = FrameLayout { + stride: 6, + width: 4, + height: 2, + source_height: 3, + }; + assert_eq!(frame_data_size(VideoFormat::NV12, nv12), Some(22)); + assert_eq!( + frame_data_size(VideoFormat::BGRx, FrameLayout { stride: 15, ..packed }), + None + ); + } + + #[test] + fn wrapped_chunk_is_reassembled() { + let data = [0, 1, 2, 3, 4, 5]; + assert_eq!(chunk_bytes(&data, 1, 3).as_deref(), Some([1, 2, 3].as_slice())); + assert_eq!(chunk_bytes(&data, 4, 4).as_deref(), Some([4, 5, 0, 1].as_slice())); + } + + #[test] + fn chunk_flags_distinguish_empty_and_invalid_frames() { let empty = spa::buffer::ChunkFlags::from_bits_retain(CHUNK_FLAG_EMPTY); - assert_eq!(chunk_kind(0, spa::buffer::ChunkFlags::empty()), ChunkKind::Invalid); - assert_eq!(chunk_kind(1, spa::buffer::ChunkFlags::CORRUPTED), ChunkKind::Invalid); - assert_eq!(chunk_kind(1, empty), ChunkKind::Empty); - assert_eq!(chunk_kind(0, empty), ChunkKind::Empty); - assert_eq!(chunk_kind(1, spa::buffer::ChunkFlags::empty()), ChunkKind::Data); + assert_eq!( + chunk_kind(0, spa::buffer::ChunkFlags::empty(), false), + ChunkKind::Invalid + ); + assert_eq!(chunk_kind(0, spa::buffer::ChunkFlags::empty(), true), ChunkKind::Data); + assert_eq!( + chunk_kind(1, spa::buffer::ChunkFlags::CORRUPTED, false), + ChunkKind::Invalid + ); + assert_eq!(chunk_kind(1, empty, false), ChunkKind::Empty); + assert_eq!(chunk_kind(0, empty, true), ChunkKind::Empty); + assert_eq!(chunk_kind(1, spa::buffer::ChunkFlags::empty(), false), ChunkKind::Data); } #[test] - fn empty_chunk_is_limited_range_black() { - let frame = neutral_frame(4, 2, None).unwrap(); - assert_eq!(frame.y(), &[16; 8]); - assert_eq!(frame.u(), &[128; 2]); - assert_eq!(frame.v(), &[128; 2]); + fn dmabuf_size_comes_from_seek_end_not_stat() { + let mut calls = Vec::new(); + let size = dma_buf_allocation_size_with_seek(1024, |offset, whence| { + calls.push((offset, whence)); + match whence { + libc::SEEK_END => Ok(4096), + libc::SEEK_SET => Ok(0), + _ => panic!("unexpected seek mode {whence}"), + } + }) + .unwrap(); + + // This models a DMA-BUF anon inode: no stat size is available, and the + // allocation length comes exclusively from its SEEK_END operation. + assert_eq!(size, Some(3072)); + assert_eq!(calls, [(0, libc::SEEK_END), (0, libc::SEEK_SET)]); + assert_eq!( + dma_buf_allocation_size_with_seek(4096, |_, whence| match whence { + libc::SEEK_END => Ok(4096), + libc::SEEK_SET => Ok(0), + _ => unreachable!(), + }) + .unwrap(), + None + ); } #[test] @@ -1007,46 +1744,14 @@ mod tests { } #[test] - fn empty_chunk_uses_full_range_black() { - let frame = neutral_frame(4, 2, Some(Color::Bt709Full)).unwrap(); - assert_eq!(frame.y(), &[0; 8]); - assert_eq!(frame.u(), &[128; 2]); - assert_eq!(frame.v(), &[128; 2]); - assert_eq!(frame.color(), Some(Color::Bt709Full)); - } - - /// The required size must reach the last sampled row, but not the padding - /// past it, or a tightly-sized final row would be rejected. - #[test] - fn frame_data_size_covers_every_sampled_row() { - let packed = FrameLayout { - stride: 20, - width: 4, - height: 2, - source_height: 2, - }; - assert_eq!(frame_data_size(VideoFormat::BGRx, packed), Some(40)); - - let nv12 = FrameLayout { - stride: 6, - width: 4, - height: 2, - source_height: 3, + fn stale_pool_lease_is_not_current() { + let buffer = std::ptr::dangling_mut::(); + let lease = Lease { + buffer: buffer as usize, + generation: 2, }; - assert_eq!(frame_data_size(VideoFormat::NV12, nv12), Some(22)); - - // A stride narrower than one row means the layout is nonsense. - assert_eq!( - frame_data_size(VideoFormat::BGRx, FrameLayout { stride: 15, ..packed }), - None - ); - } - - #[test] - fn wrapped_chunk_is_reassembled() { - let data = [0, 1, 2, 3, 4, 5]; - assert_eq!(chunk_bytes(&data, 1, 3).as_deref(), Some([1, 2, 3].as_slice())); - assert_eq!(chunk_bytes(&data, 4, 4).as_deref(), Some([4, 5, 0, 1].as_slice())); + assert_eq!(lease.current(1), None); + assert_eq!(lease.current(2), Some(buffer)); } #[test] @@ -1090,8 +1795,6 @@ mod tests { assert_eq!(frame.v(), &[10, 12]); } - /// An odd-height source has one more luma row than the clamped output, and - /// chroma starts after all of them. #[test] fn nv12_crop_uses_the_source_height_for_chroma() { let data = [ @@ -1115,6 +1818,105 @@ mod tests { assert_eq!(frame.v(), &[10, 12]); } + /// A duplicated fd does not protect the pixels from producer reuse. The + /// dequeued PipeWire buffer must be returned only after the last surface + /// clone drops, and its return crosses back onto the PipeWire loop. + #[test] + fn dmabuf_returns_on_last_drop() { + use std::cell::Cell; + use std::io::Write; + use std::os::fd::OwnedFd; + use std::os::unix::net::UnixStream; + use std::rc::Rc; + + pw::init(); + let mainloop = pw::main_loop::MainLoopRc::new(None).expect("main loop"); + let (return_tx, return_rx) = pw::channel::channel::(); + let returned = Rc::new(Cell::new(None)); + let _returns = return_rx.attach(mainloop.loop_(), { + let mainloop = mainloop.downgrade(); + let returned = returned.clone(); + move |raw| { + returned.set(Some(raw)); + mainloop.upgrade().expect("live loop").quit(); + } + }); + let timer = mainloop.loop_().add_timer({ + let mainloop = mainloop.downgrade(); + move |_| mainloop.upgrade().expect("live loop").quit() + }); + + let (socket, mut peer) = UnixStream::pair().expect("fd pair"); + peer.write_all(&[0]).expect("signal readable"); + let inner = Arc::new(PipeWireDmaBuf { + fd: OwnedFd::from(socket), + return_tx, + lease: Lease { + buffer: 7, + generation: 1, + }, + map_offset: 0, + allocation_size: Some(6), + data_offset: 0, + layout: FrameLayout { + stride: 2, + width: 2, + height: 2, + source_height: 2, + }, + format: DrmFormat::NV12, + 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), + inner, + ) + .expect("DMA-BUF"); + assert_eq!(frame.format(), DrmFormat::NV12); + assert_eq!(frame.modifier(), 0); + assert_eq!((frame.width(), frame.height()), (2, 2)); + assert_eq!(frame.planes()[1], DmaBufPlane::new(4, 2)); + assert_eq!(Surface::DmaBuf(frame.clone()).color(), Some(Color::Bt709Full)); + let export = frame.export().expect("exported fd"); + let clone = frame.clone(); + + drop(frame); + timer + .update_timer(Some(Duration::from_millis(10)), None) + .into_result() + .expect("timer"); + mainloop.run(); + assert_eq!(returned.get(), None, "one live clone still owns the lease"); + + drop(clone); + timer + .update_timer(Some(Duration::from_millis(10)), None) + .into_result() + .expect("timer"); + mainloop.run(); + assert_eq!(returned.get(), None, "one live export still owns the lease"); + + drop(export); + timer + .update_timer(Some(Duration::from_secs(1)), None) + .into_result() + .expect("timeout timer"); + mainloop.run(); + assert_eq!( + returned.get(), + Some(Lease { + buffer: 7, + generation: 1 + }) + ); + } + /// Open the portal, grab a few frames, and check geometry. Ignored because it /// needs a desktop session, PipeWire, and a human clicking the picker dialog: /// `cargo test -p moq-video --features pipewire portal_capture -- --ignored`. diff --git a/rs/moq-video/src/frame.rs b/rs/moq-video/src/frame.rs index 6627a1873..6165eebc1 100644 --- a/rs/moq-video/src/frame.rs +++ b/rs/moq-video/src/frame.rs @@ -13,6 +13,9 @@ //! `PixelBuffer` but has no Direct3D11 path yet. // `render` is deliberately not a doc link: the module sits behind a non-default // feature, so linking it fails the `-D warnings` rustdoc build of a plain build. +//! - `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::I420` is CPU-resident planar I420, for the CPU encode path and //! platforms without a zero-copy capture. //! @@ -22,6 +25,11 @@ use std::borrow::Cow; +#[cfg(all(target_os = "linux", feature = "dmabuf"))] +use std::os::fd::{AsFd, AsRawFd, BorrowedFd, OwnedFd}; +#[cfg(all(target_os = "linux", feature = "dmabuf"))] +use std::sync::Arc; + use bytes::Bytes; use moq_net::Timestamp; @@ -74,6 +82,259 @@ impl Frame { } } +/// A DRM pixel format code carried by a Linux DMA-BUF. +/// +/// The four bytes are the kernel DRM fourcc, kept as a newtype so a stride, +/// PipeWire format id, or another bare integer cannot be passed accidentally. +#[cfg(all(target_os = "linux", feature = "dmabuf"))] +#[derive(Clone, Copy, Debug, PartialEq, Eq, Hash)] +pub struct DrmFormat(u32); + +#[cfg(all(target_os = "linux", feature = "dmabuf"))] +impl DrmFormat { + /// Semi-planar 8-bit 4:2:0 YUV. + pub const NV12: Self = Self::from_bytes(*b"NV12"); + /// Packed BGRx8888 as named by DRM (`XR24`). + pub const XRGB8888: Self = Self::from_bytes(*b"XR24"); + /// Packed BGRA8888 as named by DRM (`AR24`). + pub const ARGB8888: Self = Self::from_bytes(*b"AR24"); + /// Packed RGBx8888 as named by DRM (`XB24`). + pub const XBGR8888: Self = Self::from_bytes(*b"XB24"); + /// Packed RGBA8888 as named by DRM (`AB24`). + pub const ABGR8888: Self = Self::from_bytes(*b"AB24"); + + /// Build a DRM fourcc from its four ASCII bytes. + pub const fn from_bytes(bytes: [u8; 4]) -> Self { + Self(u32::from_le_bytes(bytes)) + } + + /// The integer value used by DRM, Vulkan, EGL, and VAAPI descriptors. + pub const fn as_raw(self) -> u32 { + self.0 + } +} + +/// One plane within a Linux DMA-BUF allocation. +#[cfg(all(target_os = "linux", feature = "dmabuf"))] +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +pub struct DmaBufPlane { + offset: u32, + stride: u32, +} + +#[cfg(all(target_os = "linux", feature = "dmabuf"))] +impl DmaBufPlane { + #[cfg(feature = "pipewire")] + pub(crate) const fn new(offset: u32, stride: u32) -> Self { + Self { offset, stride } + } + + /// Byte offset of this plane from the start of the exported allocation. + pub const fn offset(&self) -> u32 { + self.offset + } + + /// Bytes between adjacent rows in this plane. + pub const fn stride(&self) -> u32 { + self.stride + } +} + +/// An exported Linux DMA-BUF descriptor and its producer lease. +/// +/// Keep this value alive for as long as an external device may read from the +/// descriptor returned by [`as_fd`](Self::as_fd). Dropping it releases the +/// producer's buffer when no other frame or export still owns that lease. +#[cfg(all(target_os = "linux", feature = "dmabuf"))] +pub struct DmaBufExport { + fd: OwnedFd, + inner: Arc, +} + +/// How long to wait on a producer's write fence before giving up. +/// +/// Vulkan does not adopt a DMA-BUF's implicit fence, so a reader has to wait for +/// it here. A screen frame's fence signals within a frame time; anything past +/// this is a wedged compositor, and the caller's CPU fallback beats blocking a +/// render thread forever. +#[cfg(all(target_os = "linux", feature = "dmabuf"))] +const DMA_BUF_FENCE_TIMEOUT: std::time::Duration = std::time::Duration::from_millis(500); + +#[cfg(all(target_os = "linux", feature = "dmabuf"))] +pub(crate) fn wait_dma_buf_readable(fd: BorrowedFd<'_>) -> std::io::Result<()> { + let mut event = libc::pollfd { + fd: fd.as_raw_fd(), + events: libc::POLLIN, + revents: 0, + }; + let deadline = std::time::Instant::now() + DMA_BUF_FENCE_TIMEOUT; + loop { + // A signal restarts the wait against the same deadline rather than + // granting a fresh budget, so the total stall stays bounded. + let remaining = deadline.saturating_duration_since(std::time::Instant::now()); + if remaining.is_zero() { + return Err(std::io::Error::from(std::io::ErrorKind::TimedOut)); + } + // SAFETY: `event` is valid for this call and `fd` remains borrowed until + // the producer's current write fence has completed. + let result = unsafe { + libc::poll( + &mut event, + 1, + remaining.as_millis().min(i32::MAX as u128) as libc::c_int, + ) + }; + if result > 0 && event.revents & libc::POLLIN != 0 { + return Ok(()); + } + if result == 0 { + return Err(std::io::Error::from(std::io::ErrorKind::TimedOut)); + } + if result < 0 { + let error = std::io::Error::last_os_error(); + if error.kind() == std::io::ErrorKind::Interrupted { + continue; + } + return Err(error); + } + return Err(std::io::Error::other(format!( + "DMA-BUF poll returned events {:#x}", + event.revents + ))); + } +} + +#[cfg(all(target_os = "linux", feature = "dmabuf"))] +impl DmaBufExport { + /// Borrow the exported descriptor without separating it from its producer lease. + pub fn as_fd(&self) -> std::os::fd::BorrowedFd<'_> { + std::os::fd::AsFd::as_fd(&self.fd) + } + + pub(crate) fn into_parts(self) -> (OwnedFd, Arc) { + (self.fd, self.inner) + } +} + +#[cfg(all(target_os = "linux", feature = "dmabuf"))] +impl std::os::fd::AsFd for DmaBufExport { + fn as_fd(&self) -> std::os::fd::BorrowedFd<'_> { + std::os::fd::AsFd::as_fd(&self.fd) + } +} + +#[cfg(all(target_os = "linux", feature = "dmabuf"))] +impl std::fmt::Debug for DmaBufExport { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + f.debug_struct("DmaBufExport").finish_non_exhaustive() + } +} + +/// A Linux DMA-BUF surface with an on-demand exported descriptor. +/// +/// Cloning this value retains the producer's surface but opens no file +/// descriptor. [`export`](Self::export) duplicates the descriptor only when a +/// consumer is ready to import it, avoiding one open fd for every buffered +/// frame. Dropping the last clone or [`DmaBufExport`] releases the producer's +/// buffer. +#[cfg(all(target_os = "linux", feature = "dmabuf"))] +#[derive(Clone)] +pub struct DmaBuf { + format: DrmFormat, + modifier: u64, + width: u32, + height: u32, + planes: Vec, + color: Option, + inner: Arc, +} + +#[cfg(all(target_os = "linux", feature = "dmabuf"))] +impl std::fmt::Debug for DmaBuf { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + f.debug_struct("DmaBuf") + .field("format", &self.format) + .field("modifier", &format_args!("{:#x}", self.modifier)) + .field("width", &self.width) + .field("height", &self.height) + .field("planes", &self.planes) + .finish_non_exhaustive() + } +} + +#[cfg(all(target_os = "linux", feature = "dmabuf"))] +impl DmaBuf { + #[cfg(feature = "pipewire")] + 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"))); + } + Ok(Self { + format, + modifier, + width, + height, + planes, + color, + inner, + }) + } + + /// Wait for producer writes, then export the descriptor with its producer lease. + pub fn export(&self) -> std::io::Result { + let fd = self.inner.export()?; + wait_dma_buf_readable(fd.as_fd())?; + Ok(DmaBufExport { + fd, + inner: self.inner.clone(), + }) + } + + /// DRM fourcc describing the plane layout. + pub const fn format(&self) -> DrmFormat { + self.format + } + + /// DRM format modifier describing the allocation's tiling. + pub const fn modifier(&self) -> u64 { + self.modifier + } + + /// Width of the coded allocation in pixels. + pub const fn width(&self) -> u32 { + self.width + } + + /// Height of the coded allocation in pixels. + pub const fn height(&self) -> u32 { + self.height + } + + /// Plane offsets and row strides, in format order. + pub fn planes(&self) -> &[DmaBufPlane] { + &self.planes + } +} + +/// The producer-owned half of a DMA-BUF surface. +/// +/// Kept private to the crate so backend lifetimes and download mechanisms do +/// not become public implementable API. [`DmaBuf`] is the stable consumer seam. +#[cfg(all(target_os = "linux", feature = "dmabuf"))] +pub(crate) trait DmaBufFrame: Send + Sync { + fn export(&self) -> std::io::Result; + fn download_i420(&self) -> Result; +} + /// Where a frame's pixels currently live. /// /// Decoders and capture sources hand these out; encoders and renderers consume @@ -105,6 +366,9 @@ pub enum Surface { /// decoder, consumed in place by the NVENC encoder. #[cfg(all(target_os = "linux", feature = "nvdec"))] Cuda(cuda::Frame), + /// Linux DMA-BUF, exported on access and retained until the last clone drops. + #[cfg(all(target_os = "linux", feature = "dmabuf"))] + DmaBuf(DmaBuf), /// CPU-resident planar I420. I420(I420), } @@ -119,6 +383,8 @@ impl Surface { Surface::Texture(t) => t.width, #[cfg(all(target_os = "linux", feature = "nvdec"))] Surface::Cuda(c) => c.width, + #[cfg(all(target_os = "linux", feature = "dmabuf"))] + Surface::DmaBuf(d) => d.width, Surface::I420(i) => i.width, } } @@ -132,6 +398,8 @@ impl Surface { Surface::Texture(t) => t.height, #[cfg(all(target_os = "linux", feature = "nvdec"))] Surface::Cuda(c) => c.height, + #[cfg(all(target_os = "linux", feature = "dmabuf"))] + Surface::DmaBuf(d) => d.height, Surface::I420(i) => i.height, } } @@ -288,6 +556,8 @@ impl Surface { Surface::Texture(_) => None, #[cfg(all(target_os = "linux", feature = "nvdec"))] Surface::Cuda(_) => None, + #[cfg(all(target_os = "linux", feature = "dmabuf"))] + Surface::DmaBuf(d) => d.color, Surface::I420(i) => i.color(), } } @@ -301,6 +571,8 @@ impl Surface { Surface::Texture(t) => Ok(Cow::Owned(t.download_i420()?)), #[cfg(all(target_os = "linux", feature = "nvdec"))] Surface::Cuda(c) => Ok(Cow::Owned(c.download_i420()?)), + #[cfg(all(target_os = "linux", feature = "dmabuf"))] + Surface::DmaBuf(d) => Ok(Cow::Owned(d.inner.download_i420()?)), Surface::I420(i) => Ok(Cow::Borrowed(i)), } } diff --git a/rs/moq-video/src/lib.rs b/rs/moq-video/src/lib.rs index 86927be58..4165a0b10 100644 --- a/rs/moq-video/src/lib.rs +++ b/rs/moq-video/src/lib.rs @@ -84,6 +84,8 @@ mod mf; 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::{Frame, I420, Surface}; pub use size::Size; diff --git a/rs/moq-video/src/render/dmabuf.rs b/rs/moq-video/src/render/dmabuf.rs new file mode 100644 index 000000000..015447b7b --- /dev/null +++ b/rs/moq-video/src/render/dmabuf.rs @@ -0,0 +1,102 @@ +//! Zero-copy import of packed Linux DMA-BUFs into wgpu's Vulkan backend. +//! +//! wgpu owns the Vulkan image and imported fd once wrapping succeeds. The +//! [`DmaBuf`] itself rides with the submitted render work separately, keeping +//! the dequeued producer buffer out of PipeWire's pool until the GPU is done. + +use wgpu::hal::MemoryFlags; + +use super::source::{Layout, Source}; +use crate::{DmaBuf, DrmFormat, Error, Size}; + +fn err(message: impl std::fmt::Display) -> Error { + Error::Render(anyhow::anyhow!("{message}")) +} + +/// Alias one packed DMA-BUF allocation as a sampled Vulkan texture. +pub(super) fn import(device: &wgpu::Device, buffer: &DmaBuf) -> Result, Error> { + if !device + .features() + .contains(wgpu::Features::VULKAN_EXTERNAL_MEMORY_DMA_BUF) + { + return Ok(None); + } + + let format = match buffer.format() { + DrmFormat::XRGB8888 | DrmFormat::ARGB8888 => wgpu::TextureFormat::Bgra8Unorm, + DrmFormat::XBGR8888 | DrmFormat::ABGR8888 => wgpu::TextureFormat::Rgba8Unorm, + format => return Err(err(format!("cannot import DMA-BUF format {:#x}", format.as_raw()))), + }; + let [plane] = buffer.planes() else { + return Err(err("packed DMA-BUF must have exactly one plane")); + }; + let size = Size::new(buffer.width(), buffer.height()); + let extent = wgpu::Extent3d { + width: size.width, + height: size.height, + depth_or_array_layers: 1, + }; + let descriptor = wgpu::TextureDescriptor { + label: Some("moq-video imported DMA-BUF"), + size: extent, + mip_level_count: 1, + sample_count: 1, + dimension: wgpu::TextureDimension::D2, + format, + usage: wgpu::TextureUsages::TEXTURE_BINDING, + view_formats: &[], + }; + let hal_descriptor = wgpu::hal::TextureDescriptor { + label: descriptor.label, + size: extent, + mip_level_count: 1, + sample_count: 1, + dimension: wgpu::TextureDimension::D2, + format, + usage: wgpu::TextureUses::RESOURCE, + memory_flags: MemoryFlags::empty(), + view_formats: Vec::new(), + }; + + // SAFETY: the guard is only used to import a descriptor into the same + // Vulkan device. It drops before the resulting HAL texture is wrapped. + let Some(hal) = (unsafe { device.as_hal::() }) else { + return Ok(None); + }; + let export = buffer + .export() + .map_err(|e| Error::Render(anyhow::Error::new(e).context("export DMA-BUF")))?; + let (fd, keepalive) = export.into_parts(); + // SAFETY: `fd` is a fresh duplicate of this live DMA-BUF. Export waited for + // producer writes, and the format, modifier, extent, stride, and offset come + // from PipeWire's buffer metadata. Vulkan consumes the duplicate on success + // and wgpu-hal closes it on error. + let texture = unsafe { + hal.texture_from_dmabuf_fd( + fd, + &hal_descriptor, + buffer.modifier(), + plane.stride() as u64, + plane.offset() as u64, + ) + } + .map_err(|e| err(format!("Vulkan DMA-BUF import: {e:?}")))?; + drop(hal); + + // SAFETY: wgpu-hal created `texture` on this device from `hal_descriptor`, + // which exactly matches the public descriptor. Imported pixels are already + // initialized and will first be used as a sampled resource. + let texture = unsafe { + device.create_texture_from_hal::(texture, &descriptor, wgpu::TextureUses::RESOURCE) + }; + let view = texture.create_view(&Default::default()); + + Ok(Some(Source { + layout: Layout::Rgba, + color: None, + plane0: view.clone(), + plane1: view.clone(), + plane2: view, + keepalive: Some(Box::new(keepalive)), + })) +} diff --git a/rs/moq-video/src/render/metal.rs b/rs/moq-video/src/render/metal.rs index 27a6c8b58..73aea95f8 100644 --- a/rs/moq-video/src/render/metal.rs +++ b/rs/moq-video/src/render/metal.rs @@ -157,10 +157,11 @@ impl Import { Ok(Source { layout, - color, + color: Some(color), plane0, plane1, plane2, + keepalive: None, }) } diff --git a/rs/moq-video/src/render/mod.rs b/rs/moq-video/src/render/mod.rs index b9f66bcc2..e80530235 100644 --- a/rs/moq-video/src/render/mod.rs +++ b/rs/moq-video/src/render/mod.rs @@ -26,13 +26,15 @@ //! ## Zero-copy //! //! A hardware-decoded frame is imported by aliasing the decoder's surface as a -//! texture rather than copying it: `CVMetalTextureCache` on macOS, for the -//! `PixelBuffer` variant of [`Surface`](crate::Surface) that capture and a -//! VideoToolbox decode produce. +//! texture rather than copying it: `CVMetalTextureCache` on macOS for the +//! `PixelBuffer` variant of [`Surface`](crate::Surface), and Vulkan external +//! memory on Linux for packed PipeWire DMA-BUFs. // `PixelBuffer` is deliberately not a doc link: the variant is macOS-only, so a // link to it fails the `-D warnings` rustdoc build on every other platform. //! -//! Every other frame, and any import that fails, goes through +//! Linux callers must request +//! [`wgpu::Features::VULKAN_EXTERNAL_MEMORY_DMA_BUF`] when creating the device +//! to enable that import. Every other frame, and any import that fails, goes through //! [`Surface::into_i420`](crate::Surface::into_i420) and a plane upload. That //! path is always available, so which route a frame takes is a question of cost. //! An import path that keeps failing (a driver that cannot do it at all) retires @@ -44,6 +46,9 @@ mod color; mod renderer; mod source; +#[cfg(all(target_os = "linux", feature = "dmabuf"))] +mod dmabuf; + #[cfg(target_os = "macos")] mod metal; diff --git a/rs/moq-video/src/render/renderer.rs b/rs/moq-video/src/render/renderer.rs index 8e521ce39..b82710d5c 100644 --- a/rs/moq-video/src/render/renderer.rs +++ b/rs/moq-video/src/render/renderer.rs @@ -12,6 +12,18 @@ use crate::{Color, Error, Frame, Size}; /// frame rate, so the path is retired and the CPU fallback takes over. const ZERO_COPY_STRIKES: u32 = 3; +#[cfg(all(target_os = "linux", feature = "dmabuf"))] +fn dma_buf_import_timed_out(error: &Error) -> bool { + let Error::Render(error) = error else { + return false; + }; + error.chain().any(|cause| { + cause + .downcast_ref::() + .is_some_and(|error| error.kind() == std::io::ErrorKind::TimedOut) + }) +} + /// Renderer configuration. /// /// `#[non_exhaustive]`: build via [`Config::new`] (or `default()`) and set the @@ -100,6 +112,8 @@ impl Config { pub struct Renderer { device: wgpu::Device, queue: wgpu::Queue, + #[cfg(all(target_os = "linux", feature = "dmabuf"))] + completion: Completion, config: Config, shader: Pipelines, @@ -125,15 +139,78 @@ struct Pipelines { /// everywhere, so it stays validated on every platform either way. #[cfg(target_os = "macos")] nv12: wgpu::RenderPipeline, + #[cfg(all(target_os = "linux", feature = "dmabuf"))] + /// Packed RGB or BGR imported from a Linux DMA-BUF. + rgba: wgpu::RenderPipeline, i420: wgpu::RenderPipeline, } +/// Waits for submitted GPU work before releasing its producer-owned surfaces. +#[cfg(all(target_os = "linux", feature = "dmabuf"))] +struct Completion { + device: wgpu::Device, + tx: Option)>>, + thread: Option>, +} + +#[cfg(all(target_os = "linux", feature = "dmabuf"))] +impl Completion { + fn new(device: &wgpu::Device) -> Result { + let (tx, rx) = std::sync::mpsc::channel(); + let worker_device = device.clone(); + let thread = std::thread::Builder::new() + .name("moq-video-gpu-completion".into()) + .spawn(move || { + while let Ok((submission, keepalive)) = rx.recv() { + if let Err(err) = worker_device.poll(wgpu::PollType::Wait { + submission_index: Some(submission), + timeout: None, + }) { + tracing::warn!(%err, "waiting for imported GPU surface failed"); + } + drop(keepalive); + } + }) + .map_err(|err| Error::Render(anyhow::anyhow!("start GPU completion worker: {err}")))?; + + Ok(Self { + device: device.clone(), + tx: Some(tx), + thread: Some(thread), + }) + } + + fn submit(&self, submission: wgpu::SubmissionIndex, keepalive: Box) { + let tx = self.tx.as_ref().expect("completion sender lives until drop"); + if let Err(err) = tx.send((submission, keepalive)) { + let (submission, keepalive) = err.0; + let _ = self.device.poll(wgpu::PollType::Wait { + submission_index: Some(submission), + timeout: None, + }); + drop(keepalive); + } + } +} + +#[cfg(all(target_os = "linux", feature = "dmabuf"))] +impl Drop for Completion { + fn drop(&mut self) { + drop(self.tx.take()); + if let Some(thread) = self.thread.take() { + let _ = thread.join(); + } + } +} + impl Renderer { /// Build a renderer on an existing `wgpu` device. /// /// The device and queue are the application's: the renderer draws into /// textures that application already owns, so it never creates a device of - /// its own. Both handles are cheap to clone and are kept. + /// its own. Both handles are cheap to clone and are kept. On Linux, request + /// [`wgpu::Features::VULKAN_EXTERNAL_MEMORY_DMA_BUF`] on the device to import + /// PipeWire DMA-BUFs instead of downloading them. pub fn new(device: &wgpu::Device, queue: &wgpu::Queue, config: Config) -> Result { if let Some(size) = config.size { size.validate_nonzero("render output")?; @@ -150,6 +227,8 @@ impl Renderer { Ok(Self { device: device.clone(), queue: queue.clone(), + #[cfg(all(target_os = "linux", feature = "dmabuf"))] + completion: Completion::new(device)?, config, shader, uniform, @@ -167,12 +246,14 @@ impl Renderer { /// call overwrites what you are holding. Present or copy it before rendering /// again. pub fn render(&mut self, frame: &Frame) -> Result { - let source = self.source(frame)?; - let color = self.config.color.unwrap_or(source.color); - if self.color != Some(color) { - self.queue - .write_buffer(&self.uniform, 0, bytemuck::cast_slice(&uniform(color))); - self.color = Some(color); + let mut source = self.source(frame)?; + if let Some(source_color) = source.color { + let color = self.config.color.unwrap_or(source_color); + if self.color != Some(color) { + self.queue + .write_buffer(&self.uniform, 0, bytemuck::cast_slice(&uniform(color))); + self.color = Some(color); + } } let output = self.output(frame.size())?; @@ -206,6 +287,8 @@ impl Renderer { }); let pipeline = match source.layout { + #[cfg(all(target_os = "linux", feature = "dmabuf"))] + Layout::Rgba => &self.shader.rgba, #[cfg(target_os = "macos")] Layout::Nv12 => &self.shader.nv12, Layout::I420 => &self.shader.i420, @@ -236,7 +319,13 @@ impl Renderer { pass.set_bind_group(0, &bind, &[]); pass.draw(0..3, 0..1); } - self.queue.submit([encoder.finish()]); + let submission = self.queue.submit([encoder.finish()]); + if let Some(keepalive) = source.keepalive.take() { + #[cfg(all(target_os = "linux", feature = "dmabuf"))] + self.completion.submit(submission, keepalive); + #[cfg(not(all(target_os = "linux", feature = "dmabuf")))] + drop((submission, keepalive)); + } Ok(output) } @@ -254,6 +343,8 @@ impl Renderer { // failure, so it costs no strike: the CPU path is the answer // for this surface and always will be. Ok(None) => {} + #[cfg(all(target_os = "linux", feature = "dmabuf"))] + Err(err) if dma_buf_import_timed_out(&err) => return Err(err), Err(err) => { self.strikes += 1; self.retired = self.strikes >= ZERO_COPY_STRIKES; @@ -406,6 +497,8 @@ impl Pipelines { }); Ok(Self { + #[cfg(all(target_os = "linux", feature = "dmabuf"))] + rgba: pipeline("rgba"), #[cfg(target_os = "macos")] nv12: pipeline("nv12"), i420: pipeline("i420"), @@ -422,6 +515,18 @@ mod tests { use super::*; use crate::Surface; + #[cfg(all(target_os = "linux", feature = "dmabuf"))] + #[test] + fn dma_buf_fence_timeout_is_terminal_for_the_frame() { + let timed_out = Error::Render(anyhow::Error::new(std::io::Error::from(std::io::ErrorKind::TimedOut))); + let other = Error::Render(anyhow::Error::new(std::io::Error::from( + std::io::ErrorKind::PermissionDenied, + ))); + + assert!(dma_buf_import_timed_out(&timed_out)); + assert!(!dma_buf_import_timed_out(&other)); + } + /// Every test here draws on a real GPU, which a headless CI runner does not /// have (wgpu finds no adapter and `Renderer::new` never gets built). The /// color math itself is covered by [`super::super::color`]'s tests, which diff --git a/rs/moq-video/src/render/shader.wgsl b/rs/moq-video/src/render/shader.wgsl index db8918f6b..a919b38f7 100644 --- a/rs/moq-video/src/render/shader.wgsl +++ b/rs/moq-video/src/render/shader.wgsl @@ -1,5 +1,5 @@ -// YUV 4:2:0 -> RGB, for the two plane layouts the renderer feeds it: NV12 (luma -// plane + interleaved chroma plane) and I420 (three separate planes). +// The renderer feeds this shader packed RGBA, NV12 (luma plus interleaved +// chroma), or I420 (three separate planes). // // Chroma is sampled with a linear filter, so the half-resolution planes are // upsampled by the texture unit rather than in here. Output is gamma-encoded @@ -43,6 +43,11 @@ fn convert(yuv: vec3) -> vec4 { return vec4(clamp(rgb, vec3(0.0), vec3(1.0)), 1.0); } +@fragment +fn rgba(in: Vertex) -> @location(0) vec4 { + return vec4(textureSample(plane0, samp, in.uv).rgb, 1.0); +} + @fragment fn nv12(in: Vertex) -> @location(0) vec4 { let y = textureSample(plane0, samp, in.uv).r; diff --git a/rs/moq-video/src/render/source.rs b/rs/moq-video/src/render/source.rs index 49d045cd1..7d9dc1473 100644 --- a/rs/moq-video/src/render/source.rs +++ b/rs/moq-video/src/render/source.rs @@ -6,6 +6,10 @@ use crate::{Color, Error, Frame, Size, Surface}; /// How the planes of a [`Source`] are arranged, which picks the fragment shader. #[derive(Clone, Copy, Debug, PartialEq, Eq)] pub(super) enum Layout { + /// One packed RGBA or BGRA texture. The fragment shader ignores alpha so + /// both XRGB and ARGB DMA-BUFs produce an opaque frame. + #[cfg(all(target_os = "linux", feature = "dmabuf"))] + Rgba, /// Luma plane plus one interleaved chroma plane. What every hardware /// decoder hands back, so only a zero-copy import produces it: the CPU /// upload path is always I420. Gated to the platforms that have such an @@ -22,10 +26,14 @@ pub(super) enum Layout { /// that one bind group layout serves both shaders. pub(super) struct Source { pub layout: Layout, - pub color: Color, + /// The color conversion for YUV layouts. Packed RGB needs no conversion. + pub color: Option, pub plane0: wgpu::TextureView, pub plane1: wgpu::TextureView, pub plane2: wgpu::TextureView, + /// Producer-owned storage that must remain leased until the submitted draw + /// stops reading it. The renderer moves this to its completion worker. + pub keepalive: Option>, } /// The renderer's per-frame GPU state: the CPU path's plane textures, plus @@ -56,6 +64,8 @@ impl Cache { /// not, which is what the caller counts strikes against. pub fn import(&mut self, device: &wgpu::Device, surface: &Surface) -> Result, Error> { match surface { + #[cfg(all(target_os = "linux", feature = "dmabuf"))] + Surface::DmaBuf(buffer) => super::dmabuf::import(device, buffer), #[cfg(target_os = "macos")] Surface::PixelBuffer(buffer) => { let metal = match &mut self.metal { @@ -97,10 +107,11 @@ impl Cache { // The conversion that produced these samples says which space they // are in where it knows. Only a passthrough (a decode, a camera) // leaves it open, and then the resolution is all there is to go on. - color: i420.color().unwrap_or_else(|| Color::infer(size)), + color: Some(i420.color().unwrap_or_else(|| Color::infer(size))), plane0, plane1, plane2, + keepalive: None, }) } }