diff --git a/crates/punktfunk-host/src/capture/linux/mod.rs b/crates/punktfunk-host/src/capture/linux/mod.rs index 5a4c0e1..a8dd199 100644 --- a/crates/punktfunk-host/src/capture/linux/mod.rs +++ b/crates/punktfunk-host/src/capture/linux/mod.rs @@ -43,6 +43,12 @@ pub struct PortalCapturer { /// True only while the PipeWire stream is `Streaming`. [`try_latest`](Self::try_latest) reads it /// to distinguish a static desktop (alive, no new buffers) from a dead source (left `Streaming`). streaming: Arc, + /// Poison flag: the zero-copy GPU import is irrecoverably gone for this stream (the import + /// worker died — e.g. it absorbed the driver fault of a crashing compositor — or tiled imports + /// failed repeatedly, where the CPU fallback would de-pad scrambled tiled bytes). Both + /// [`next_frame`](Capturer::next_frame) and [`try_latest`](Self::try_latest) surface it as an + /// error so the session's capture-loss rebuild runs instead of freezing/corrupting. + broken: Arc, /// When the stream first dropped out of `Streaming` with no new frame; used to grace a transient /// renegotiation before declaring the source lost. Cleared whenever a frame arrives or the stream /// is `Streaming`. @@ -130,6 +136,8 @@ struct PwHandles { active: Arc, negotiated: Arc, streaming: Arc, + /// See [`PortalCapturer::broken`]. + broken: Arc, /// This capture will offer LINEAR-dmabuf-only for the VAAPI passthrough (see /// [`PortalCapturer::vaapi_dmabuf`]). vaapi_dmabuf: bool, @@ -146,6 +154,7 @@ impl PwHandles { active: self.active, negotiated: self.negotiated, streaming: self.streaming, + broken: self.broken, stall_since: None, vaapi_dmabuf: self.vaapi_dmabuf, node_id, @@ -178,6 +187,8 @@ fn spawn_pipewire( let negotiated_cb = negotiated.clone(); let streaming = Arc::new(AtomicBool::new(false)); let streaming_cb = streaming.clone(); + let broken = Arc::new(AtomicBool::new(false)); + let broken_cb = broken.clone(); // pipewire's own cross-thread channel: the receiver attaches to the loop and quits it; the // sender lives on the capturer and fires in its `Drop`. Absolute `::pipewire` path — the // inner `mod pipewire` shadows the crate name at this scope. @@ -199,6 +210,7 @@ fn spawn_pipewire( active_cb, negotiated_cb, streaming_cb, + broken_cb, zerocopy, preferred, quit_rx, @@ -212,6 +224,7 @@ fn spawn_pipewire( active, negotiated, streaming, + broken, vaapi_dmabuf, quit: quit_tx, join, @@ -220,48 +233,36 @@ fn spawn_pipewire( impl Capturer for PortalCapturer { fn next_frame(&mut self) -> Result { - // First frame can lag behind format negotiation; later frames arrive at ~fps. - match self.frames.recv_timeout(Duration::from_secs(10)) { - Ok(frame) => Ok(frame), - Err(RecvTimeoutError::Timeout) => { - // Split the two black-screen root causes apart so the operator gets a cause, not - // just a symptom: did the format negotiate (compositor produced no buffers) or - // not (no acceptable format / node never emitted a param)? - if self.negotiated.load(Ordering::Relaxed) { - Err(anyhow!( - "no PipeWire frame within 10s (node {}): format negotiated but no buffers \ - arrived — the compositor produced no frames (virtual output idle/unmapped, \ - or capture never started)", - self.node_id - )) - } else if self.vaapi_dmabuf && !crate::zerocopy::vaapi_dmabuf_forced() { - // The LINEAR-dmabuf-only offer (VAAPI passthrough default) was never accepted. - // Latch the process-wide downgrade so the encode loop's pipeline rebuild - // retries on the CPU offer instead of failing this same negotiation forever. - crate::zerocopy::note_vaapi_dmabuf_failed(); - Err(anyhow!( - "no PipeWire frame within 10s (node {}): the compositor never accepted \ - the LINEAR-dmabuf offer (VAAPI zero-copy) — downgrading this host to the \ - CPU capture path; the pipeline rebuild will renegotiate without dmabuf", - self.node_id - )) - } else { - Err(anyhow!( - "no PipeWire frame within 10s (node {}): format negotiation never \ - completed — the compositor offered no format this consumer accepts \ - (pixel-format/modifier mismatch) or the node never emitted a Format param", - self.node_id - )) - } + // First frame can lag behind format negotiation; later frames arrive at ~fps. Wait in + // short slices so a GPU-import poison (worker death) fails the capture within ~0.5 s + // instead of sitting out the full first-frame budget. + let deadline = std::time::Instant::now() + Duration::from_secs(10); + loop { + if self.broken.load(Ordering::Relaxed) { + return Err(anyhow!( + "zero-copy GPU import lost (node {}): the import worker died or tiled imports \ + failed repeatedly — rebuilding capture", + self.node_id + )); + } + let slice = Duration::from_millis(500) + .min(deadline.saturating_duration_since(std::time::Instant::now())); + match self.frames.recv_timeout(slice) { + Ok(frame) => return Ok(frame), + Err(RecvTimeoutError::Timeout) if std::time::Instant::now() < deadline => continue, + Err(e) => return self.next_frame_timed_out(e), } - Err(RecvTimeoutError::Disconnected) => Err(anyhow!( - "PipeWire capture thread ended before a frame (node {})", - self.node_id - )), } } fn try_latest(&mut self) -> Result> { + if self.broken.load(Ordering::Relaxed) { + return Err(anyhow!( + "zero-copy GPU import lost (node {}): the import worker died or tiled imports \ + failed repeatedly — rebuilding capture", + self.node_id + )); + } // Drain to the newest queued frame without blocking; `None` means the compositor // hasn't produced a new frame since last call (static/idle desktop). let mut latest = None; @@ -304,6 +305,50 @@ impl Capturer for PortalCapturer { } } +impl PortalCapturer { + /// The [`Capturer::next_frame`] budget expired (or the thread ended) — turn it into the + /// diagnosis-bearing error. Split out of the slicing loop above; behavior unchanged. + fn next_frame_timed_out(&self, err: RecvTimeoutError) -> Result { + match err { + RecvTimeoutError::Timeout => { + // Split the two black-screen root causes apart so the operator gets a cause, not + // just a symptom: did the format negotiate (compositor produced no buffers) or + // not (no acceptable format / node never emitted a param)? + if self.negotiated.load(Ordering::Relaxed) { + Err(anyhow!( + "no PipeWire frame within 10s (node {}): format negotiated but no buffers \ + arrived — the compositor produced no frames (virtual output idle/unmapped, \ + or capture never started)", + self.node_id + )) + } else if self.vaapi_dmabuf && !crate::zerocopy::vaapi_dmabuf_forced() { + // The LINEAR-dmabuf-only offer (VAAPI passthrough default) was never accepted. + // Latch the process-wide downgrade so the encode loop's pipeline rebuild + // retries on the CPU offer instead of failing this same negotiation forever. + crate::zerocopy::note_vaapi_dmabuf_failed(); + Err(anyhow!( + "no PipeWire frame within 10s (node {}): the compositor never accepted \ + the LINEAR-dmabuf offer (VAAPI zero-copy) — downgrading this host to the \ + CPU capture path; the pipeline rebuild will renegotiate without dmabuf", + self.node_id + )) + } else { + Err(anyhow!( + "no PipeWire frame within 10s (node {}): format negotiation never \ + completed — the compositor offered no format this consumer accepts \ + (pixel-format/modifier mismatch) or the node never emitted a Format param", + self.node_id + )) + } + } + RecvTimeoutError::Disconnected => Err(anyhow!( + "PipeWire capture thread ended before a frame (node {})", + self.node_id + )), + } + } +} + impl Drop for PortalCapturer { fn drop(&mut self) { // Stop the PipeWire loop and wait for the thread to unwind BEFORE the keepalive (virtual @@ -548,8 +593,15 @@ mod pipewire { /// `Paused`/`Unconnected`/`Error` — the source vanished (compositor torn down on a session /// switch). Read by [`PortalCapturer::try_latest`] to surface a sustained drop as a loss. streaming: Arc, - /// Present when zero-copy is enabled on NVIDIA: imports a dmabuf → CUDA device buffer. - importer: Option, + /// Poison flag (see [`PortalCapturer::broken`]): set here when the GPU import is + /// irrecoverably gone for this stream — the import worker died, or tiled imports failed + /// [`IMPORT_FAIL_POISON`] times in a row. + broken: Arc, + /// Consecutive tiled-import failures (reset on success); see [`IMPORT_FAIL_POISON`]. + import_fail_streak: u32, + /// Present when zero-copy is enabled on NVIDIA: imports a dmabuf → CUDA device buffer, + /// normally via the isolated worker process (`crate::zerocopy::Importer::Remote`). + importer: Option, /// VAAPI zero-copy: hand the raw dmabuf to the encoder (which imports + GPU-CSCs it) instead /// of a CUDA import. Set when zero-copy is on, the EGL→CUDA importer is unavailable, and the /// encoder backend is VAAPI (AMD/Intel). @@ -561,6 +613,12 @@ mod pipewire { dbg_log_n: u64, } + /// Consecutive tiled-import failures (worker alive, e.g. a per-buffer `EGL_BAD_MATCH`) before + /// the stream is poisoned for rebuild. A tiled import failure must NEVER fall through to the + /// CPU mmap path — de-padding tiled bytes as linear produces a scrambled image — so after a + /// short streak of dropped frames the capturer fails loudly and the session renegotiates. + const IMPORT_FAIL_POISON: u32 = 3; + /// Log a frame-drop reason once per process (the process callback runs per frame; a stuck /// pipeline must say why without flooding). fn warn_once(msg: &'static str) { @@ -814,6 +872,11 @@ mod pipewire { if !ud.active.load(Ordering::Relaxed) { return; } + // Poisoned (GPU import lost): the capturer is already surfacing an error to the encode + // loop; skip per-frame work until the rebuild tears this stream down. + if ud.broken.load(Ordering::Relaxed) { + return; + } // SAFETY: `spa_buf` is the `*mut spa_buffer` of the PipeWire buffer we dequeued and still hold for // this `.process` callback (not requeued until after `consume_frame` returns), so it is live. The // block null-checks `spa_buf`, requires `n_datas != 0`, and null-checks the `datas` array pointer @@ -965,6 +1028,8 @@ mod pipewire { }; match imported { Ok(devbuf) => { + ud.import_fail_streak = 0; + crate::zerocopy::note_gpu_import_ok(); static ONCE: std::sync::atomic::AtomicBool = std::sync::atomic::AtomicBool::new(true); if ONCE.swap(false, Ordering::Relaxed) { @@ -990,12 +1055,32 @@ mod pipewire { return; } Err(e) => { - // GPU import unavailable for this buffer kind (e.g. the - // driver rejects LINEAR external-memory import). Disable - // the importer and fall through to the CPU mmap path — - // degraded, not dead. + let dead = importer.dead(); + if dead { + crate::zerocopy::note_gpu_import_death(); + } + if modifier.is_some() { + // Tiled buffer: the CPU fallback below would mmap TILED bytes + // and de-pad them as linear — a scrambled image, worse than no + // frame. Drop the frame instead; on a dead worker (it absorbed a + // driver fault) or a short failure streak, poison the stream so + // the session's capture-loss rebuild renegotiates cleanly. + ud.import_fail_streak += 1; + if dead || ud.import_fail_streak >= IMPORT_FAIL_POISON { + tracing::error!(error = %format!("{e:#}"), dead, + "tiled GPU import lost — failing this capture for rebuild"); + ud.broken.store(true, Ordering::Relaxed); + } else { + tracing::warn!(error = %format!("{e:#}"), + streak = ud.import_fail_streak, + "tiled dmabuf GPU import failed — frame dropped"); + } + return; + } + // LINEAR dmabuf: CPU-mappable, so disable the importer and fall + // through to the CPU mmap path — degraded, not dead. tracing::warn!(error = %format!("{e:#}"), - "dmabuf GPU import failed — falling back to the CPU copy path"); + "LINEAR dmabuf GPU import failed — falling back to the CPU copy path"); gpu_import_broken = true; } } @@ -1138,6 +1223,7 @@ mod pipewire { active: Arc, negotiated: Arc, streaming: Arc, + broken: Arc, zerocopy: bool, preferred: Option<(u32, u32, u32)>, quit_rx: pw::channel::Receiver<()>, @@ -1165,18 +1251,28 @@ mod pipewire { .context("pw connect (default daemon)")?, }; - // Build the EGL→CUDA importer up front; if it fails, log and fall back to the CPU path + // Build the GPU importer up front — normally the ISOLATED worker process + // (design/zerocopy-worker-isolation.md), so a driver fault on a dying compositor's + // dmabuf kills the worker, not this host. If it fails, log and fall back to the CPU path // (we simply won't request dmabuf below). Skipped entirely when the encode backend is // VAAPI: those frames go to the raw-dmabuf passthrough, and building the importer there // would waste a CUDA probe — or worse, on an NVIDIA box forced to PUNKTFUNK_ENCODER=vaapi, - // succeed and produce CUDA payloads the VAAPI encoder must reject. + // succeed and produce CUDA payloads the VAAPI encoder must reject. Also skipped once + // repeated worker deaths latched the import off (a wedged GPU stack must not crash-loop). let backend_is_vaapi = crate::encode::linux_zero_copy_is_vaapi(); - let importer = if zerocopy && !backend_is_vaapi { - match crate::zerocopy::EglImporter::new() { - Ok(i) => Some(i), - Err(e) => { - tracing::warn!(error = %format!("{e:#}"), "zero-copy import unavailable — using CPU path"); - None + let mut importer = if zerocopy && !backend_is_vaapi { + if crate::zerocopy::gpu_import_disabled() { + tracing::warn!( + "zero-copy GPU import disabled after repeated import-worker deaths — using CPU path" + ); + None + } else { + match crate::zerocopy::Importer::new_for_capture() { + Ok(i) => Some(i), + Err(e) => { + tracing::warn!(error = %format!("{e:#}"), "zero-copy import unavailable — using CPU path"); + None + } } } } else { @@ -1194,7 +1290,7 @@ mod pipewire { // CUDA external memory instead. For the VAAPI passthrough path we advertise LINEAR only: // radeonsi/iHD import it and any compositor can allocate it. let mut modifiers = importer - .as_ref() + .as_mut() .map(|i| i.supported_modifiers(crate::zerocopy::drm_fourcc(PixelFormat::Bgrx).unwrap())) .unwrap_or_default(); if (importer.is_some() || vaapi_passthrough) && !modifiers.contains(&0) { @@ -1247,6 +1343,8 @@ mod pipewire { active, negotiated, streaming, + broken, + import_fail_streak: 0, importer, vaapi_passthrough, nv12: crate::zerocopy::nv12_enabled(), @@ -1300,6 +1398,13 @@ mod pipewire { } if ud.info.parse(param).is_ok() { ud.negotiated.store(true, Ordering::Relaxed); + // A (re)negotiation replaces the buffer pool: every cached per-buffer import + // (stored fds in the worker, the Vulkan bridge's per-fd sources) keys on + // buffers that no longer exist — and a recycled fd number/inode must never + // resolve to a stale import. No-op on the first negotiation (empty caches). + if let Some(imp) = ud.importer.as_mut() { + imp.clear_cache(); + } let sz = ud.info.size(); ud.format = map_format(ud.info.format()); ud.modifier = ud.info.modifier(); diff --git a/crates/punktfunk-host/src/linux/zerocopy/client.rs b/crates/punktfunk-host/src/linux/zerocopy/client.rs new file mode 100644 index 0000000..dfa7c95 --- /dev/null +++ b/crates/punktfunk-host/src/linux/zerocopy/client.rs @@ -0,0 +1,609 @@ +//! Host side of the isolated zero-copy GPU import (design: +//! [`design/zerocopy-worker-isolation.md`]): spawns the `zerocopy-worker` subprocess, mirrors the +//! [`super::egl::EglImporter`] entry points over the [`super::proto`] socket, and materializes +//! the worker's pooled CUDA buffers in this process via CUDA IPC (each buffer's handles are +//! opened exactly once and reused as the pool recycles). A worker death — the whole point of the +//! isolation — surfaces as an `Err` with [`RemoteImporter::dead`] set, never as a host fault. + +// Every `unsafe` block in this file carries a `// SAFETY:` proof; enforce it (unsafe-proof program). +#![deny(clippy::undocumented_unsafe_blocks)] + +use super::cuda::{self, CUdeviceptr, DeviceBuffer, CU_IPC_HANDLE_SIZE}; +use super::egl::DmabufPlane; +use super::proto::{self, BufferDesc, ImportKind, Reply, Request}; +use anyhow::{bail, Context, Result}; +use std::collections::{HashMap, HashSet}; +use std::io; +use std::os::fd::{AsFd, AsRawFd, BorrowedFd, OwnedFd}; +use std::path::Path; +use std::process::{Child, Command}; +use std::sync::atomic::{AtomicBool, Ordering}; +use std::sync::{Arc, Mutex}; +use std::time::Duration; + +/// Handshake budget: EGL + CUDA bring-up is ~200 ms; a cold driver load can take seconds. +const HANDSHAKE_TIMEOUT: Duration = Duration::from_secs(20); +/// Per-request budget. An import is a few ms of GPU work; if the worker can't answer in this +/// window it is wedged (GPU fault in progress) and gets treated as dead. +const REPLY_TIMEOUT: Duration = Duration::from_secs(10); + +/// State shared with in-flight frames: the socket (their release messages) and the CUDA IPC +/// mappings (their device pointers). Lives until the LAST in-flight [`DeviceBuffer`] drops, so a +/// mapping is never closed under a frame the encoder still reads — and only then does the socket +/// close, which is what tells an idle worker to exit. +struct Shared { + sock: OwnedFd, + mappings: Mutex>, + dead: AtomicBool, +} + +/// One pooled worker buffer, opened in this process. +#[derive(Clone, Copy)] +struct Mapping { + y: CUdeviceptr, + y_pitch: usize, + uv: Option<(CUdeviceptr, usize)>, + width: u32, + height: u32, +} + +impl Drop for Shared { + fn drop(&mut self) { + // Last reference gone — no DeviceBuffer can still point into these mappings. + for (_, m) in self.mappings.lock().unwrap().drain() { + cuda::ipc_close(m.y); + if let Some((uv, _)) = m.uv { + cuda::ipc_close(uv); + } + } + } +} + +/// Children whose worker hasn't exited yet at `RemoteImporter` drop time (it exits on socket +/// EOF, i.e. after the last in-flight frame drops). Swept on every spawn and every drop so +/// workers don't linger as zombies for more than one capture generation. +static REAPER: Mutex> = Mutex::new(Vec::new()); + +fn sweep_reaper() { + let mut list = REAPER.lock().unwrap(); + list.retain_mut(|c| !matches!(c.try_wait(), Ok(Some(_)))); +} + +/// The remote (isolated) importer — one per capture. Method-for-method mirror of the in-process +/// [`super::egl::EglImporter`] surface the capture thread uses. +pub struct RemoteImporter { + shared: Arc, + child: Option, + /// Reused receive scratch buffer (all replies are read by the single capture thread). + rbuf: Vec, + /// Dmabuf keys (`st_ino`) whose fd the worker already holds — the fd is passed only once. + sent_keys: HashSet, +} + +impl RemoteImporter { + /// Spawn the worker from this host binary and complete the readiness handshake. An `Err` + /// here means "no isolated zero-copy available" — callers fall back to the CPU path, exactly + /// like an in-process `EglImporter::new()` failure. + pub fn spawn() -> Result { + let exe = std::env::current_exe().context("resolve /proc/self/exe for the worker")?; + Self::spawn_exe(&exe) + } + + /// [`Self::spawn`] with an explicit executable (separated for tests). + fn spawn_exe(exe: &Path) -> Result { + sweep_reaper(); + let (host_end, worker_end) = proto::socketpair_seqpacket().context("worker socketpair")?; + let mut cmd = Command::new(exe); + cmd.arg("zerocopy-worker").arg("--fd").arg("3"); + let raw = worker_end.as_raw_fd(); + // SAFETY: `pre_exec` runs between fork and exec, so only async-signal-safe calls are + // allowed — `dup2` and `fcntl` both are, and the closure captures only the `Copy` int + // `raw` (no allocation, no locks). `dup2(raw, 3)` installs the socket at the fd number + // the subcommand expects and clears CLOEXEC on the copy; if the parent's fd already IS 3, + // `dup2(3,3)` would preserve CLOEXEC, so that case clears the flag explicitly instead. + unsafe { + use std::os::unix::process::CommandExt; + cmd.pre_exec(move || { + if raw == 3 { + let flags = libc::fcntl(3, libc::F_GETFD); + if flags < 0 || libc::fcntl(3, libc::F_SETFD, flags & !libc::FD_CLOEXEC) < 0 { + return Err(io::Error::last_os_error()); + } + } else if libc::dup2(raw, 3) < 0 { + return Err(io::Error::last_os_error()); + } + Ok(()) + }); + } + let child = cmd.spawn().context("spawn zerocopy-worker")?; + drop(worker_end); // the child holds its own copy now + Self::from_socket(host_end, Some(child)) + } + + /// Complete the handshake on an already-connected socket (the unit tests drive this against + /// a mock server thread instead of a real subprocess). + fn from_socket(sock: OwnedFd, child: Option) -> Result { + let mut importer = RemoteImporter { + shared: Arc::new(Shared { + sock, + mappings: Mutex::new(HashMap::new()), + dead: AtomicBool::new(false), + }), + child, + rbuf: Vec::new(), + sent_keys: HashSet::new(), + }; + proto::set_recv_timeout(importer.shared.sock.as_fd(), Some(HANDSHAKE_TIMEOUT))?; + let ready = proto::recv::(importer.shared.sock.as_fd(), &mut importer.rbuf); + proto::set_recv_timeout(importer.shared.sock.as_fd(), Some(REPLY_TIMEOUT))?; + match ready { + Ok((Reply::Ready { version }, _)) if version == proto::PROTO_VERSION => { + tracing::info!( + pid = importer.child.as_ref().map(|c| c.id()), + "zero-copy GPU import isolated in a worker process" + ); + Ok(importer) + } + Ok((Reply::Ready { version }, _)) => { + importer.mark_dead(); + bail!( + "zerocopy worker protocol mismatch (worker v{version}, host v{})", + proto::PROTO_VERSION + ) + } + Ok((Reply::InitErr { message }, _)) => { + // The worker exits by itself after reporting; not a death, just "no GPU here". + bail!("zerocopy worker init failed: {message}") + } + Ok((other, _)) => { + importer.mark_dead(); + bail!("unexpected zerocopy worker handshake: {other:?}") + } + Err(e) => { + importer.mark_dead(); + Err(e).context("zerocopy worker handshake (died on startup?)") + } + } + } + + /// True once any exchange failed at the transport level — the worker is gone (or wedged) and + /// every further call fails fast. The capture layer poisons its stream on this. + pub fn dead(&self) -> bool { + self.shared.dead.load(Ordering::Relaxed) + } + + fn mark_dead(&self) { + self.shared.dead.store(true, Ordering::Relaxed); + } + + /// Mirror of [`super::egl::EglImporter::supported_modifiers`] (worker round-trip; empty on + /// any failure, which makes the capture fall back like an importless negotiation). + pub fn supported_modifiers(&mut self, fourcc: u32) -> Vec { + if self.dead() { + return Vec::new(); + } + if let Err(e) = proto::send( + self.shared.sock.as_fd(), + &Request::Modifiers { fourcc }, + None, + ) { + tracing::warn!(error = %e, "zerocopy worker modifier query failed"); + self.mark_dead(); + return Vec::new(); + } + match proto::recv::(self.shared.sock.as_fd(), &mut self.rbuf) { + Ok((Reply::Modifiers { modifiers }, _)) => modifiers, + Ok((other, _)) => { + tracing::warn!(?other, "unexpected zerocopy worker reply to Modifiers"); + self.mark_dead(); + Vec::new() + } + Err(e) => { + tracing::warn!(error = %e, "zerocopy worker modifier reply failed"); + self.mark_dead(); + Vec::new() + } + } + } + + /// Mirror of [`super::egl::EglImporter::import`] (tiled dmabuf → BGRx CUDA buffer). + pub fn import( + &mut self, + plane: &DmabufPlane, + width: u32, + height: u32, + fourcc: u32, + modifier: Option, + ) -> Result { + self.import_impl(plane, ImportKind::Tiled, width, height, fourcc, modifier) + } + + /// Mirror of [`super::egl::EglImporter::import_nv12`]. + pub fn import_nv12( + &mut self, + plane: &DmabufPlane, + width: u32, + height: u32, + fourcc: u32, + modifier: Option, + ) -> Result { + self.import_impl( + plane, + ImportKind::TiledNv12, + width, + height, + fourcc, + modifier, + ) + } + + /// Mirror of [`super::egl::EglImporter::import_linear`] (LINEAR dmabuf → Vulkan bridge). + pub fn import_linear( + &mut self, + plane: &DmabufPlane, + width: u32, + height: u32, + ) -> Result { + self.import_impl(plane, ImportKind::Linear, width, height, 0, None) + } + + fn import_impl( + &mut self, + plane: &DmabufPlane, + kind: ImportKind, + width: u32, + height: u32, + fourcc: u32, + modifier: Option, + ) -> Result { + if self.dead() { + bail!("zerocopy worker is dead"); + } + let key = dmabuf_key(plane.fd)?; + // One retry: a `NeedFd` reply (the worker's fd cache evicted this key) clears our + // "already sent" note so the second attempt carries the fd again. + let mut attempts = 0; + let reply = loop { + attempts += 1; + let has_fd = self.sent_keys.insert(key); + // SAFETY: `plane.fd` is the dmabuf fd of the PipeWire buffer the capture thread still + // holds for this callback (`consume_frame`'s contract), so it is open and stays open + // for this synchronous call; the `BorrowedFd` never outlives it (used only for the + // `send`). + let pass = has_fd.then(|| unsafe { BorrowedFd::borrow_raw(plane.fd) }); + let req = Request::Import { + key, + kind, + width, + height, + fourcc, + modifier, + offset: plane.offset, + stride: plane.stride, + has_fd, + }; + if let Err(e) = proto::send(self.shared.sock.as_fd(), &req, pass) { + self.mark_dead(); + return Err(e).context("zerocopy worker died (send)"); + } + let reply = match proto::recv::(self.shared.sock.as_fd(), &mut self.rbuf) { + Ok((reply, _)) => reply, + Err(e) => { + self.mark_dead(); + return Err(e).context("zerocopy worker died (no reply)"); + } + }; + match reply { + Reply::NeedFd if attempts == 1 => { + self.sent_keys.remove(&key); + continue; + } + Reply::NeedFd => { + self.mark_dead(); + bail!("zerocopy worker still lacks the fd after a resend (desync)"); + } + other => break other, + } + }; + match reply { + Reply::Frame { id, desc } => { + if let Some(desc) = desc { + let mapping = open_mapping(&desc).with_context(|| { + // An unopenable mapping poisons every future frame in this buffer — + // treat it as a dead worker so the capture rebuilds cleanly. + self.mark_dead(); + format!("open CUDA IPC mapping for worker buffer {id}") + })?; + self.shared.mappings.lock().unwrap().insert(id, mapping); + } + let m = self + .shared + .mappings + .lock() + .unwrap() + .get(&id) + .copied() + .ok_or_else(|| { + self.mark_dead(); + anyhow::anyhow!("worker delivered unknown buffer id {id} (desync)") + })?; + let shared = self.shared.clone(); + Ok(DeviceBuffer::remote( + m.y, + m.y_pitch, + m.width, + m.height, + m.uv, + Box::new(move || { + // Fire-and-forget recycle; a dead worker just means EPIPE, ignored. The + // captured `shared` Arc is what keeps the mapping + socket alive until + // the last frame drops. + let _ = proto::send(shared.sock.as_fd(), &Request::Release { id }, None); + }), + )) + } + Reply::Err { message } => bail!("zerocopy worker import failed: {message}"), + other => { + self.mark_dead(); + bail!("unexpected zerocopy worker reply: {other:?}") + } + } + } + + /// The PipeWire stream renegotiated — reset both sides' per-buffer caches. + pub fn clear_cache(&mut self) { + self.sent_keys.clear(); + if !self.dead() { + if let Err(e) = proto::send(self.shared.sock.as_fd(), &Request::ClearCache, None) { + tracing::warn!(error = %e, "zerocopy worker ClearCache failed"); + self.mark_dead(); + } + } + } +} + +impl Drop for RemoteImporter { + fn drop(&mut self) { + // The worker exits on socket EOF, which happens when the last `Shared` reference (this + // importer, or the final in-flight frame on the encode side) drops. Reap what's already + // gone; park the rest for the next sweep. + if let Some(mut child) = self.child.take() { + if !matches!(child.try_wait(), Ok(Some(_))) { + REAPER.lock().unwrap().push(child); + } + } + sweep_reaper(); + } +} + +/// Identity of the dma-buf behind `fd`, stable across frames and across `SCM_RIGHTS` re-numbering: +/// every dma-buf gets a unique inode on the kernel's dmabuf pseudo-fs for its lifetime. Used as +/// the worker's fd-cache key so the fd itself is only passed once. +fn dmabuf_key(fd: i32) -> Result { + // SAFETY: `libc::stat` is plain-old-data for which all-zero is a valid value, so + // `mem::zeroed()` is a sound initializer. `fd` is the caller's live dmabuf fd; `fstat` writes + // into `&mut st`, a live, correctly-sized stack struct that outlives the synchronous call, + // and `st_ino` is read only after the return value is checked. + unsafe { + let mut st: libc::stat = std::mem::zeroed(); + if libc::fstat(fd, &mut st) != 0 { + bail!("fstat(dmabuf fd): {}", io::Error::last_os_error()); + } + Ok(st.st_ino) + } +} + +/// Open a worker buffer's CUDA IPC handles in this process. +fn open_mapping(desc: &BufferDesc) -> Result { + cuda::make_current()?; + let y_handle: [u8; CU_IPC_HANDLE_SIZE] = desc + .y_handle + .as_slice() + .try_into() + .context("worker sent a malformed Y IPC handle")?; + let y = cuda::ipc_open(&y_handle).context("open Y plane IPC handle")?; + let uv = match &desc.uv { + Some((handle, pitch)) => { + let handle: [u8; CU_IPC_HANDLE_SIZE] = handle + .as_slice() + .try_into() + .context("worker sent a malformed UV IPC handle")?; + match cuda::ipc_open(&handle) { + Ok(ptr) => Some((ptr, *pitch)), + Err(e) => { + // Don't leak the Y mapping on a half-open failure. + cuda::ipc_close(y); + return Err(e).context("open UV plane IPC handle"); + } + } + } + None => None, + }; + Ok(Mapping { + y, + y_pitch: desc.y_pitch, + uv, + width: desc.width, + height: desc.height, + }) +} + +#[cfg(test)] +mod tests { + use super::*; + use std::thread; + + fn handshake_server(reply: Reply) -> OwnedFd { + let (host, worker) = proto::socketpair_seqpacket().unwrap(); + proto::send(worker.as_fd(), &reply, None).unwrap(); + // Keep the worker end alive alongside the host end for the test's duration by leaking it + // into the reply thread below? Not needed: the handshake reply is already queued in the + // socket buffer, so the worker end may drop — recv still delivers queued data first. + drop(worker); + host + } + + #[test] + fn handshake_ready_and_version_gate() { + let host = handshake_server(Reply::Ready { + version: proto::PROTO_VERSION, + }); + let imp = RemoteImporter::from_socket(host, None).unwrap(); + assert!(!imp.dead()); + + let host = handshake_server(Reply::Ready { version: 999 }); + assert!(RemoteImporter::from_socket(host, None).is_err()); + } + + #[test] + fn handshake_init_err() { + let host = handshake_server(Reply::InitErr { + message: "no GPU".into(), + }); + let Err(err) = RemoteImporter::from_socket(host, None) else { + panic!("InitErr handshake must fail") + }; + assert!(format!("{err:#}").contains("no GPU"), "{err:#}"); + } + + #[test] + fn handshake_eof_is_an_error() { + let (host, worker) = proto::socketpair_seqpacket().unwrap(); + drop(worker); + assert!(RemoteImporter::from_socket(host, None).is_err()); + } + + #[test] + fn spawning_a_non_worker_fails_cleanly() { + // `true` exits immediately without a handshake → EOF → clean spawn error, the same + // fallback path a GPU-less box takes. + let Err(err) = RemoteImporter::spawn_exe(Path::new("true")) else { + panic!("spawning a non-worker must fail") + }; + assert!(format!("{err:#}").contains("handshake"), "{err:#}"); + } + + /// A scripted peer: answers the handshake, then serves canned replies per request. + fn scripted_server(replies: Vec) -> (RemoteImporter, thread::JoinHandle>) { + let (host, worker) = proto::socketpair_seqpacket().unwrap(); + proto::send( + worker.as_fd(), + &Reply::Ready { + version: proto::PROTO_VERSION, + }, + None, + ) + .unwrap(); + let join = thread::spawn(move || { + let mut buf = Vec::new(); + let mut seen = Vec::new(); + let mut replies = replies.into_iter(); + while let Ok((req, _fd)) = proto::recv::(worker.as_fd(), &mut buf) { + let needs_reply = matches!(req, Request::Modifiers { .. } | Request::Import { .. }); + seen.push(req); + if needs_reply { + match replies.next() { + Some(r) => proto::send(worker.as_fd(), &r, None).unwrap(), + None => break, // close → client sees a dead worker + } + } + } + seen + }); + let imp = RemoteImporter::from_socket(host, None).unwrap(); + (imp, join) + } + + #[test] + fn modifiers_round_trip() { + let (mut imp, join) = scripted_server(vec![Reply::Modifiers { + modifiers: vec![1, 2, 3], + }]); + assert_eq!(imp.supported_modifiers(0x3432_5258), vec![1, 2, 3]); + assert!(!imp.dead()); + drop(imp); + let seen = join.join().unwrap(); + assert_eq!( + seen, + vec![Request::Modifiers { + fourcc: 0x3432_5258 + }] + ); + } + + #[test] + fn need_fd_triggers_one_resend_with_the_fd() { + let (mut imp, join) = scripted_server(vec![ + Reply::Err { + message: "one".into(), + }, + Reply::NeedFd, + Reply::Err { + message: "two".into(), + }, + ]); + let (pr, _pw) = std::io::pipe().unwrap(); + let plane = DmabufPlane { + fd: pr.as_fd().as_raw_fd(), + offset: 0, + stride: 256, + }; + // First import: first sight of the key → fd rides along; the Err reply keeps the key + // marked as sent (the worker cached the fd before failing). + assert!(imp.import(&plane, 64, 64, 1, Some(2)).is_err()); + // Second import: no fd (already sent) → worker answers NeedFd → one retry WITH the fd. + assert!(imp.import(&plane, 64, 64, 1, Some(2)).is_err()); + assert!(!imp.dead(), "NeedFd handling must not mark the worker dead"); + drop(imp); + let fd_flags: Vec = join + .join() + .unwrap() + .iter() + .map(|r| match r { + Request::Import { has_fd, .. } => *has_fd, + other => panic!("unexpected request {other:?}"), + }) + .collect(); + assert_eq!(fd_flags, vec![true, false, true]); + } + + #[test] + fn import_error_reply_keeps_worker_alive_and_death_is_detected() { + let (mut imp, join) = scripted_server(vec![Reply::Err { + message: "EGL_BAD_MATCH".into(), + }]); + // Any pipe works as a stand-in fd for key derivation. + let (pr, _pw) = std::io::pipe().unwrap(); + let plane = DmabufPlane { + fd: pr.as_fd().as_raw_fd(), + offset: 0, + stride: 256, + }; + let Err(err) = imp.import(&plane, 64, 64, 1, Some(2)) else { + panic!("scripted Err reply must fail the import") + }; + assert!(format!("{err:#}").contains("EGL_BAD_MATCH")); + assert!(!imp.dead(), "an Err reply must not mark the worker dead"); + + // The scripted replies are exhausted → the server closes → the next import dies. + let Err(err) = imp.import(&plane, 64, 64, 1, Some(2)) else { + panic!("a closed worker must fail the import") + }; + assert!(format!("{err:#}").contains("died"), "{err:#}"); + assert!(imp.dead()); + drop(imp); + let seen = join.join().unwrap(); + // First import carried the fd (first sight of the key); the retry didn't re-send it. + match (&seen[0], &seen[1]) { + ( + Request::Import { + has_fd: true, + kind: ImportKind::Tiled, + .. + }, + Request::Import { has_fd: false, .. }, + ) => {} + other => panic!("unexpected requests {other:?}"), + } + } +} diff --git a/crates/punktfunk-host/src/linux/zerocopy/cuda.rs b/crates/punktfunk-host/src/linux/zerocopy/cuda.rs index 03a2905..1c32b7e 100644 --- a/crates/punktfunk-host/src/linux/zerocopy/cuda.rs +++ b/crates/punktfunk-host/src/linux/zerocopy/cuda.rs @@ -90,6 +90,21 @@ pub struct CUDA_EXTERNAL_MEMORY_BUFFER_DESC { pub const CU_EXTERNAL_MEMORY_HANDLE_TYPE_OPAQUE_FD: c_uint = 1; +/// `CUipcMemHandle` (cuda.h): an opaque 64-byte struct identifying a device allocation across +/// processes. Produced by `cuIpcGetMemHandle` in the exporting process, consumed by +/// `cuIpcOpenMemHandle` in the importer — passed **by value**, matching the C +/// `struct { char reserved[64]; }`. Plain bytes — safe to ship over a socket. +pub const CU_IPC_HANDLE_SIZE: usize = 64; +#[repr(C)] +#[derive(Clone, Copy)] +pub struct CUipcMemHandle { + pub reserved: [u8; CU_IPC_HANDLE_SIZE], +} + +/// `CUipcMem_flags`: lazily enable peer access on open (the documented flag for +/// `cuIpcOpenMemHandle`; a no-op for a same-device open, which is our only case). +const CU_IPC_MEM_LAZY_ENABLE_PEER_ACCESS: c_uint = 0x1; + /// CUDA Driver API entry points, resolved at runtime from `libcuda.so.1` via `dlopen` rather than /// a link-time `#[link(name = "cuda")]`. This is what lets ONE host binary run on NVIDIA /// (zero-copy via CUDA → NVENC) *and* on AMD/Intel (VAAPI, where the NVIDIA driver — and thus @@ -129,6 +144,9 @@ struct CudaApi { *const CUDA_EXTERNAL_MEMORY_BUFFER_DESC, ) -> CUresult, cuDestroyExternalMemory: unsafe extern "C" fn(CUexternalMemory) -> CUresult, + cuIpcGetMemHandle: unsafe extern "C" fn(*mut CUipcMemHandle, CUdeviceptr) -> CUresult, + cuIpcOpenMemHandle: unsafe extern "C" fn(*mut CUdeviceptr, CUipcMemHandle, c_uint) -> CUresult, + cuIpcCloseMemHandle: unsafe extern "C" fn(CUdeviceptr) -> CUresult, } // SAFETY: every field is a bare `extern "C" fn` address into the leaked, process-lifetime // `libcuda` mapping (`cuda_api` `forget`s the `Library`, so it is never unloaded) — an immutable @@ -192,6 +210,14 @@ fn cuda_api() -> Option<&'static CudaApi> { .get(b"cuExternalMemoryGetMappedBuffer\0") .ok()?, cuDestroyExternalMemory: *lib.get(b"cuDestroyExternalMemory\0").ok()?, + cuIpcGetMemHandle: *lib.get(b"cuIpcGetMemHandle\0").ok()?, + // CUDA 11 renamed the entry point (per-thread-stream ABI split); every modern + // driver exports `_v2`, but accept the unsuffixed one too (same signature). + cuIpcOpenMemHandle: *lib + .get(b"cuIpcOpenMemHandle_v2\0") + .or_else(|_| lib.get(b"cuIpcOpenMemHandle\0")) + .ok()?, + cuIpcCloseMemHandle: *lib.get(b"cuIpcCloseMemHandle\0").ok()?, }; std::mem::forget(lib); // keep libcuda mapped for the fn pointers' lifetime (process) Some(api) @@ -346,6 +372,28 @@ unsafe fn cuDestroyExternalMemory(ext_mem: CUexternalMemory) -> CUresult { None => CU_ERROR_NOT_LOADED, } } +unsafe fn cuIpcGetMemHandle(handle: *mut CUipcMemHandle, dptr: CUdeviceptr) -> CUresult { + match cuda_api() { + Some(a) => (a.cuIpcGetMemHandle)(handle, dptr), + None => CU_ERROR_NOT_LOADED, + } +} +unsafe fn cuIpcOpenMemHandle( + dptr: *mut CUdeviceptr, + handle: CUipcMemHandle, + flags: c_uint, +) -> CUresult { + match cuda_api() { + Some(a) => (a.cuIpcOpenMemHandle)(dptr, handle, flags), + None => CU_ERROR_NOT_LOADED, + } +} +unsafe fn cuIpcCloseMemHandle(dptr: CUdeviceptr) -> CUresult { + match cuda_api() { + Some(a) => (a.cuIpcCloseMemHandle)(dptr), + None => CU_ERROR_NOT_LOADED, + } +} #[inline] fn ck(r: CUresult, what: &str) -> Result<()> { @@ -387,6 +435,55 @@ pub fn read_plane_to_host( Ok(host) } +/// Export a device allocation (from `cuMemAllocPitch`/`cuMemAlloc`) as a cross-process CUDA IPC +/// handle — an opaque 64-byte blob another process opens with [`ipc_open`]. The allocation must +/// stay alive for as long as any importer has it open. The shared context must be current. +pub fn ipc_export(ptr: CUdeviceptr) -> Result<[u8; CU_IPC_HANDLE_SIZE]> { + let mut handle = CUipcMemHandle { + reserved: [0; CU_IPC_HANDLE_SIZE], + }; + // SAFETY: `&mut handle` is a live, correctly-sized stack out-param the driver fills with the + // opaque IPC blob; `ptr` is the caller's live device allocation (by-value integer). The call is + // synchronous and retains no pointer into Rust memory. Wrapper → live table (context current). + unsafe { ck(cuIpcGetMemHandle(&mut handle, ptr), "cuIpcGetMemHandle")? }; + Ok(handle.reserved) +} + +/// Open an IPC handle exported by *another* process ([`ipc_export`]); returns a device pointer +/// valid in this process until [`ipc_close`]. The shared context must be current. +pub fn ipc_open(handle: &[u8; CU_IPC_HANDLE_SIZE]) -> Result { + let h = CUipcMemHandle { reserved: *handle }; + let mut ptr: CUdeviceptr = 0; + // SAFETY: `h` is passed by value (matching the C `CUipcMemHandle` struct ABI); `&mut ptr` is a + // live zero-init stack out-param the driver writes the mapped device address into. Synchronous + // call, distinct locals, no aliasing. Wrapper → live table (context current). + unsafe { + ck( + cuIpcOpenMemHandle(&mut ptr, h, CU_IPC_MEM_LAZY_ENABLE_PEER_ACCESS), + "cuIpcOpenMemHandle", + )? + }; + Ok(ptr) +} + +/// Close a mapping opened with [`ipc_open`] (best-effort teardown; makes the shared context +/// current itself since drops may run off-thread). +pub fn ipc_close(ptr: CUdeviceptr) { + if ptr == 0 { + return; + } + // SAFETY: `ptr` is a device pointer previously returned by `cuIpcOpenMemHandle` (the only + // caller path), closed exactly once by the owning cache. We make the shared context current + // first because this runs from `Drop` on whatever thread holds the last reference. Result + // ignored (best-effort teardown). Wrapper → live table (the mapping exists ⇒ driver present). + unsafe { + if let Some(c) = CONTEXT.get() { + let _ = cuCtxSetCurrent(c.0); + } + let _ = cuIpcCloseMemHandle(ptr); + } +} + /// The shared process-wide CUDA context (created once). Wrapped so it's `Send`/`Sync` to live /// in a `OnceLock`; the raw `CUcontext` is thread-safe to make current from any thread. #[derive(Clone, Copy)] @@ -676,6 +773,7 @@ impl BufferPool { height: self.height, uv: Some((uv_ptr, uv_pitch)), pool: Some(self.inner.clone()), + remote_release: None, }); } let reuse = self.inner.lock().unwrap().free.pop(); @@ -690,6 +788,7 @@ impl BufferPool { height: self.height, uv: None, pool: Some(self.inner.clone()), + remote_release: None, }) } } @@ -706,6 +805,10 @@ pub struct DeviceBuffer { /// `None` for the default 4-byte RGB/BGRx path. When `Some`, [`ptr`] is the Y plane (1 byte/px). pub uv: Option<(CUdeviceptr, usize)>, pool: Option>>, + /// Set for buffers whose device memory is owned by ANOTHER process (the zero-copy import + /// worker, reached via CUDA IPC): drop runs this exactly once (telling the owner to recycle) + /// and must neither free nor pool-recycle the pointers locally. + remote_release: Option>, } impl DeviceBuffer { @@ -719,6 +822,7 @@ impl DeviceBuffer { height, uv: None, pool: None, + remote_release: None, }) } @@ -733,6 +837,7 @@ impl DeviceBuffer { height, uv: Some((uv_ptr, uv_pitch)), pool: None, + remote_release: None, }) } @@ -740,10 +845,38 @@ impl DeviceBuffer { pub fn is_nv12(&self) -> bool { self.uv.is_some() } + + /// Wrap device planes owned by ANOTHER process (opened here via [`ipc_open`]) as a frame + /// buffer. `release` runs exactly once on drop — it tells the owning process to recycle the + /// buffer; nothing is freed or pooled locally (the IPC mapping itself is closed by the cache + /// that opened it, after the last remote buffer referencing it has dropped). + pub fn remote( + ptr: CUdeviceptr, + pitch: usize, + width: u32, + height: u32, + uv: Option<(CUdeviceptr, usize)>, + release: Box, + ) -> DeviceBuffer { + DeviceBuffer { + ptr, + pitch, + width, + height, + uv, + pool: None, + remote_release: Some(release), + } + } } impl Drop for DeviceBuffer { fn drop(&mut self) { + if let Some(release) = self.remote_release.take() { + // Remote (IPC) buffer: the worker owns the memory — just hand it back. + release(); + return; + } if self.ptr == 0 { return; } @@ -988,19 +1121,34 @@ pub fn copy_nv12_to_device( } } +impl RegisteredTexture { + /// Unregister now (idempotent; the later `Drop` then no-ops). Teardown-order helper: the blit + /// destructors call this to release the CUDA registration BEFORE deleting the GL texture it + /// wraps — deleting a still-registered texture leaves the driver holding a registration onto + /// freed GL state, exactly the stale-driver-state class this path once crashed on. + pub fn release(&mut self) { + if self.resource.is_null() { + return; + } + // SAFETY: `self.resource` is non-null (just checked) and is the valid `CUgraphicsResource` + // from `register_gl`, owned exclusively by this `RegisteredTexture`; nulling the field + // right after makes this (and the `Drop` below) unregister it exactly once — no + // use-after-free or double-unregister. We make the shared context current first because a + // release may run during teardown on a thread where it isn't. Wrapper → live table (the + // resource exists ⇒ the driver was present). Result ignored (best-effort teardown). + unsafe { + if let Some(c) = CONTEXT.get() { + let _ = cuCtxSetCurrent(c.0); + } + let _ = cuGraphicsUnregisterResource(self.resource); + } + self.resource = std::ptr::null_mut(); + } +} + impl Drop for RegisteredTexture { fn drop(&mut self) { - if !self.resource.is_null() { - // SAFETY: `self.resource` is non-null (just checked) and is the valid - // `CUgraphicsResource` from `register_gl`, owned exclusively by this `RegisteredTexture` - // and unregistered exactly once here (drop runs once) — no use-after-free or - // double-unregister. `cuGraphicsUnregisterResource` releases the GL↔CUDA registration; - // wrapper → live table (the resource exists ⇒ the driver was present). Result ignored - // (best-effort teardown). - unsafe { - let _ = cuGraphicsUnregisterResource(self.resource); - } - } + self.release(); } } diff --git a/crates/punktfunk-host/src/linux/zerocopy/egl.rs b/crates/punktfunk-host/src/linux/zerocopy/egl.rs index 80d246e..d7fef79 100644 --- a/crates/punktfunk-host/src/linux/zerocopy/egl.rs +++ b/crates/punktfunk-host/src/linux/zerocopy/egl.rs @@ -270,6 +270,27 @@ impl GlBlit { } } +impl Drop for GlBlit { + fn drop(&mut self) { + // Unregister the CUDA graphics resource BEFORE deleting the GL texture it wraps (see + // `Nv12Blit::drop` — same ordering hazard). Previously `GlBlit` had no `Drop` at all, so + // its GL objects leaked on every size change and on importer teardown. + self.registered.release(); + // SAFETY: these GL names were all created by THIS `GlBlit` in `GlBlit::new` on the current + // GL context, still current here (the owning `EglImporter` drops on its single capture + // thread and never releases the context). Each `glDelete*` gets a count of 1 and a `&u32` + // to one live field; the symbols dispatch through libGL to the driver for the current + // context. Each name is deleted exactly once, after its CUDA registration was released. + unsafe { + glDeleteTextures(1, &self.dst_tex); + glDeleteTextures(1, &self.src_tex); + glDeleteFramebuffers(1, &self.fbo); + glDeleteVertexArrays(1, &self.vao); + glDeleteProgram(self.program); + } + } +} + /// Per-size GL machinery to convert a dmabuf EGLImage into an NV12 (BT.709 limited-range) pair — /// the [`GlBlit`] analogue for the `PUNKTFUNK_NV12` path. Two passes share `src_tex`: a full-res Y /// pass into a CUDA-registrable `GL_R8` texture and a half-res UV pass into a `GL_RG8` texture. @@ -417,6 +438,12 @@ impl Nv12Blit { impl Drop for Nv12Blit { fn drop(&mut self) { + // Unregister the CUDA graphics resources BEFORE deleting the GL textures they wrap. + // `Drop::drop` runs before the fields' own drops, so without this the `glDeleteTextures` + // below would destroy `y_tex`/`uv_tex` while still CUDA-registered — leaving the driver a + // registration onto freed GL state (the stale-driver-state class that crashed this path). + self.y_registered.release(); + self.uv_registered.release(); // SAFETY: these GL names (textures/FBOs/VAO/programs) were all created by THIS `Nv12Blit` // in `Nv12Blit::new` on the current GL context, which is still current because the owning // `EglImporter` is dropped on its single capture thread (fields drop before @@ -424,7 +451,8 @@ impl Drop for Nv12Blit { // pointer to that many names: `&self.y_tex`/`&self.vao` are `&u32` to one live field (n=1); // `[self.y_fbo, self.uv_fbo].as_ptr()` points at a 2-element temporary that lives for the // whole `glDeleteFramebuffers` call (n=2 matches). The symbols dispatch through libGL - // (libglvnd) to the driver for the current context. Each name is deleted exactly once. + // (libglvnd) to the driver for the current context. Each name is deleted exactly once, + // after its CUDA registration was released above. unsafe { glDeleteTextures(1, &self.y_tex); glDeleteTextures(1, &self.uv_tex); @@ -637,6 +665,22 @@ impl EglImporter { ) } + /// Drop the Vulkan bridge's cached per-fd import (see [`super::vulkan::VkBridge::forget_fd`]). + /// No-op when the bridge hasn't been built (tiled-only captures). + pub fn forget_linear_fd(&mut self, fd: i32) { + if let Some(vk) = self.vk.as_mut() { + vk.forget_fd(fd); + } + } + + /// Tear down the whole LINEAR-path import cache (the Vulkan bridge and every per-fd source + /// buffer in it). Called when the PipeWire stream renegotiates — the buffer pool the cache + /// keyed on is gone, and a recycled fd number must never resolve to a stale import. The + /// bridge lazily rebuilds on the next LINEAR frame (renegotiations are rare). + pub fn clear_linear_cache(&mut self) { + self.vk = None; + } + /// The DRM format modifiers the NVIDIA EGL stack can import for `fourcc`, via /// `eglQueryDmaBufModifiersEXT`. We advertise these to PipeWire so the compositor allocates /// a dmabuf in a layout we can import. Empty on failure (caller falls back). diff --git a/crates/punktfunk-host/src/linux/zerocopy/mod.rs b/crates/punktfunk-host/src/linux/zerocopy/mod.rs index 4361bf8..f9ff25d 100644 --- a/crates/punktfunk-host/src/linux/zerocopy/mod.rs +++ b/crates/punktfunk-host/src/linux/zerocopy/mod.rs @@ -10,11 +10,14 @@ //! headless EGLDisplay + dmabuf→`EGLImage`→CUDA import). The encoder's CUDA-frame path lives in //! `encode/linux.rs`; the dmabuf negotiation lives in `capture/linux.rs`. +pub mod client; pub mod cuda; pub mod egl; +pub mod proto; pub mod vulkan; +pub mod worker; -use std::sync::atomic::{AtomicBool, Ordering}; +use std::sync::atomic::{AtomicBool, AtomicU32, Ordering}; pub use cuda::DeviceBuffer; pub use egl::{DmabufPlane, EglImporter}; @@ -73,6 +76,127 @@ pub fn nv12_enabled() -> bool { flag_opt("PUNKTFUNK_NV12").unwrap_or(true) } +/// The GPU importer a capture uses — normally the [`client::RemoteImporter`] worker subprocess +/// (design: `design/zerocopy-worker-isolation.md`), so a driver fault on a producer-invalidated +/// dmabuf kills the worker instead of the host. `PUNKTFUNK_ZEROCOPY_INPROC=1` keeps the import +/// in-process (the pre-isolation behavior) for debugging and A/B latency comparison. +pub enum Importer { + Remote(client::RemoteImporter), + InProc(Box), +} + +impl Importer { + /// Build the importer for a capture session, honoring the `PUNKTFUNK_ZEROCOPY_INPROC` + /// escape hatch. An `Err` means "no GPU import available" — callers fall back to the CPU path. + pub fn new_for_capture() -> anyhow::Result { + if flag("PUNKTFUNK_ZEROCOPY_INPROC") { + tracing::warn!( + "PUNKTFUNK_ZEROCOPY_INPROC=1 — GPU import runs IN-PROCESS; a driver fault on a \ + dying compositor's dmabuf can take the whole host down (debug/A-B use only)" + ); + return Ok(Importer::InProc(Box::new(EglImporter::new()?))); + } + Ok(Importer::Remote(client::RemoteImporter::spawn()?)) + } + + pub fn supported_modifiers(&mut self, fourcc: u32) -> Vec { + match self { + Importer::Remote(r) => r.supported_modifiers(fourcc), + Importer::InProc(i) => i.supported_modifiers(fourcc), + } + } + + pub fn import( + &mut self, + plane: &DmabufPlane, + width: u32, + height: u32, + fourcc: u32, + modifier: Option, + ) -> anyhow::Result { + match self { + Importer::Remote(r) => r.import(plane, width, height, fourcc, modifier), + Importer::InProc(i) => i.import(plane, width, height, fourcc, modifier), + } + } + + pub fn import_nv12( + &mut self, + plane: &DmabufPlane, + width: u32, + height: u32, + fourcc: u32, + modifier: Option, + ) -> anyhow::Result { + match self { + Importer::Remote(r) => r.import_nv12(plane, width, height, fourcc, modifier), + Importer::InProc(i) => i.import_nv12(plane, width, height, fourcc, modifier), + } + } + + pub fn import_linear( + &mut self, + plane: &DmabufPlane, + width: u32, + height: u32, + ) -> anyhow::Result { + match self { + Importer::Remote(r) => r.import_linear(plane, width, height), + Importer::InProc(i) => i.import_linear(plane, width, height), + } + } + + /// True once the worker process is gone/wedged (every further call fails fast). Always + /// `false` in-process — an in-process driver fault doesn't return. + pub fn dead(&self) -> bool { + match self { + Importer::Remote(r) => r.dead(), + Importer::InProc(_) => false, + } + } + + /// The PipeWire stream renegotiated its format (the buffer pool is replaced) — drop all + /// per-buffer caches so a recycled fd number can never resolve to a stale import. + pub fn clear_cache(&mut self) { + match self { + Importer::Remote(r) => r.clear_cache(), + Importer::InProc(i) => i.clear_linear_cache(), + } + } +} + +/// Consecutive zero-copy worker deaths without a successful import in between. A short streak is +/// normal (the observed trigger — a compositor crash — kills the worker once, and the rebuilt +/// session's fresh worker succeeds); a sustained streak means the GPU stack itself is wedged and +/// respawning would crash-loop, so [`note_gpu_import_death`] latches [`GPU_IMPORT_DISABLED`] and +/// every later capture negotiates the safe CPU/SHM path instead. +static GPU_IMPORT_DEATH_STREAK: AtomicU32 = AtomicU32::new(0); +static GPU_IMPORT_DISABLED: AtomicBool = AtomicBool::new(false); +const GPU_IMPORT_DEATH_LATCH: u32 = 3; + +/// Record a worker death (transport-level failure). Latches the process-wide disable after +/// [`GPU_IMPORT_DEATH_LATCH`] consecutive deaths. +pub fn note_gpu_import_death() { + let streak = GPU_IMPORT_DEATH_STREAK.fetch_add(1, Ordering::Relaxed) + 1; + if streak >= GPU_IMPORT_DEATH_LATCH && !GPU_IMPORT_DISABLED.swap(true, Ordering::Relaxed) { + tracing::error!( + streak, + "zero-copy GPU import disabled for this host process: the import worker died {streak} \ + times in a row (GPU/driver stack unstable) — captures fall back to the CPU path" + ); + } +} + +/// Record a successful GPU import — resets the death streak (the stack works again). +pub fn note_gpu_import_ok() { + GPU_IMPORT_DEATH_STREAK.store(0, Ordering::Relaxed); +} + +/// True once repeated worker deaths latched the GPU import off (see [`note_gpu_import_death`]). +pub fn gpu_import_disabled() -> bool { + GPU_IMPORT_DISABLED.load(Ordering::Relaxed) +} + /// DRM FourCC for a packed 32-bit format name (little-endian, e.g. `b"XR24"`). const fn fourcc(c: &[u8; 4]) -> u32 { (c[0] as u32) | ((c[1] as u32) << 8) | ((c[2] as u32) << 16) | ((c[3] as u32) << 24) @@ -250,3 +374,23 @@ pub fn nv12_selftest() -> anyhow::Result<()> { bail!("NV12 self-test FAILED (Y={max_y_err:.2} U={max_u_err:.2} V={max_v_err:.2})"); } } + +#[cfg(test)] +mod tests { + use super::*; + + /// Single test owning the process-global latch statics (they are never reset by design). + #[test] + fn gpu_import_death_latch() { + note_gpu_import_death(); + note_gpu_import_ok(); // a successful import resets the streak + note_gpu_import_death(); + note_gpu_import_death(); + assert!( + !gpu_import_disabled(), + "two consecutive deaths must not latch" + ); + note_gpu_import_death(); // third consecutive death + assert!(gpu_import_disabled()); + } +} diff --git a/crates/punktfunk-host/src/linux/zerocopy/proto.rs b/crates/punktfunk-host/src/linux/zerocopy/proto.rs new file mode 100644 index 0000000..c54ce7e --- /dev/null +++ b/crates/punktfunk-host/src/linux/zerocopy/proto.rs @@ -0,0 +1,390 @@ +//! Wire protocol between the PipeWire capture thread and the isolated zero-copy GPU-import +//! worker process (`punktfunk-host zerocopy-worker`; design: +//! [`design/zerocopy-worker-isolation.md`]). Transport is a `SOCK_SEQPACKET` unix socketpair — +//! reliable, ordered, message-framed (one `sendmsg` = one message) — with dmabuf fds riding as +//! `SCM_RIGHTS` control data. Bodies are small serde_json blobs (~200 B/frame); pixels never +//! cross the socket (they move GPU-side via CUDA IPC, see [`super::cuda::ipc_export`]). +//! +//! Zero-length messages are reserved: `recvmsg` returning 0 on a SEQPACKET socket is EOF (the +//! peer died/closed), and every serialized message here is non-empty JSON, so the two can't be +//! confused. + +// Every `unsafe` block in this file carries a `// SAFETY:` proof; enforce it (unsafe-proof program). +#![deny(clippy::undocumented_unsafe_blocks)] + +use serde::de::DeserializeOwned; +use serde::{Deserialize, Serialize}; +use std::io; +use std::os::fd::{AsRawFd, BorrowedFd, FromRawFd, OwnedFd}; +use std::time::Duration; + +/// Bumped on any wire change; the worker echoes it in [`Reply::Ready`] and the host refuses a +/// mismatch. Host and worker are the same binary (`/proc/self/exe`), so this only ever trips on +/// exotic deployment mistakes (a stale binary re-exec'd across an upgrade). +pub const PROTO_VERSION: u32 = 1; + +/// Upper bound for one serialized message (the largest real message — a modifier list — is far +/// below this). A message reported truncated at this size is a protocol error. +pub const MAX_MSG: usize = 64 * 1024; + +/// How a dmabuf should be imported — mirrors the three `EglImporter` entry points. +#[derive(Serialize, Deserialize, Debug, Clone, Copy, PartialEq, Eq)] +pub enum ImportKind { + /// Tiled dmabuf → EGL/GL de-tile blit → BGRx CUDA buffer. + Tiled, + /// Tiled dmabuf → EGL/GL NV12 convert → two-plane CUDA buffer (`PUNKTFUNK_NV12`). + TiledNv12, + /// LINEAR dmabuf → Vulkan bridge → BGRx CUDA buffer (gamescope's only offer). + Linear, +} + +/// host → worker. +#[derive(Serialize, Deserialize, Debug, PartialEq)] +pub enum Request { + /// The EGL-importable DRM modifiers for `fourcc` (startup, before the stream connects — + /// the host advertises these to PipeWire). + Modifiers { fourcc: u32 }, + /// Import one frame. `key` identifies the underlying dmabuf across frames (the host uses + /// the fd's `st_ino` — unique per dma-buf object); the fd itself rides along as + /// `SCM_RIGHTS` only on first sight of `key` (`has_fd`), and the worker keeps its dup. + Import { + key: u64, + kind: ImportKind, + width: u32, + height: u32, + fourcc: u32, + modifier: Option, + offset: u32, + stride: u32, + has_fd: bool, + }, + /// The frame buffer previously delivered as `id` is no longer in use — recycle it into the + /// worker's pool. Fire-and-forget (no reply); may be sent from any host thread. + Release { id: u32 }, + /// The PipeWire stream renegotiated its format: the buffer pool is gone, so drop all cached + /// per-`key` state (stored fds, Vulkan per-fd imports). Fire-and-forget. + ClearCache, +} + +/// worker → host. +#[derive(Serialize, Deserialize, Debug, PartialEq)] +pub enum Reply { + /// Sent once at startup after EGL + CUDA came up. + Ready { + version: u32, + }, + /// Startup failed (no NVIDIA driver, EGL error, …) — the host falls back to the CPU path, + /// exactly like an in-process `EglImporter::new()` failure. + InitErr { + message: String, + }, + Modifiers { + modifiers: Vec, + }, + /// The imported frame is complete (the GPU copy already synced worker-side) in buffer `id`. + /// `desc` rides along the first time `id` is ever delivered — the host opens its CUDA IPC + /// handles once and caches the mapping for every later frame in the same buffer. + Frame { + id: u32, + desc: Option, + }, + /// The worker has no cached fd for the import's `key` (evicted, or the two sides' caches + /// diverged) — the host forgets its "already sent" note and retries once WITH the fd. + NeedFd, + /// This import failed but the worker is alive (e.g. `EGL_BAD_MATCH` on one buffer). + Err { + message: String, + }, +} + +/// CUDA IPC identity of one pooled device buffer (sent once per buffer, then referenced by id). +#[derive(Serialize, Deserialize, Debug, Clone, PartialEq)] +pub struct BufferDesc { + pub width: u32, + pub height: u32, + /// `cuIpcGetMemHandle` blob for the (Y or BGRx) plane — exactly 64 bytes. + pub y_handle: Vec, + pub y_pitch: usize, + /// NV12 only: the interleaved chroma plane's `(handle, pitch)`. + pub uv: Option<(Vec, usize)>, +} + +/// A CLOEXEC `SOCK_SEQPACKET` socketpair — `(host_end, worker_end)`. +pub fn socketpair_seqpacket() -> io::Result<(OwnedFd, OwnedFd)> { + let mut fds = [0i32; 2]; + // SAFETY: `socketpair` writes two fds into `fds`, a live 2-element stack array matching the + // API contract; it reads no other Rust memory. The result is checked before the fds are used, + // and each returned fd is fresh (owned by no other wrapper), so the two `OwnedFd::from_raw_fd` + // each take sole ownership of a distinct, valid descriptor — no alias, no double-close. + unsafe { + if libc::socketpair( + libc::AF_UNIX, + libc::SOCK_SEQPACKET | libc::SOCK_CLOEXEC, + 0, + fds.as_mut_ptr(), + ) != 0 + { + return Err(io::Error::last_os_error()); + } + Ok((OwnedFd::from_raw_fd(fds[0]), OwnedFd::from_raw_fd(fds[1]))) + } +} + +/// Set (or clear) the receive timeout: a blocked [`recv`] then fails with +/// `ErrorKind::WouldBlock`. Used by the host so a hung worker can't wedge the capture thread. +pub fn set_recv_timeout(sock: BorrowedFd, timeout: Option) -> io::Result<()> { + let tv = match timeout { + Some(d) => libc::timeval { + tv_sec: d.as_secs() as libc::time_t, + tv_usec: d.subsec_micros() as libc::suseconds_t, + }, + None => libc::timeval { + tv_sec: 0, + tv_usec: 0, + }, + }; + // SAFETY: `setsockopt(SO_RCVTIMEO)` reads `size_of::()` bytes from `&tv`, a live + // stack `timeval` that outlives this synchronous call; `sock` is the caller's live socket fd. + // Nothing is retained or written through Rust pointers. + let r = unsafe { + libc::setsockopt( + sock.as_raw_fd(), + libc::SOL_SOCKET, + libc::SO_RCVTIMEO, + &tv as *const libc::timeval as *const libc::c_void, + std::mem::size_of::() as libc::socklen_t, + ) + }; + if r != 0 { + return Err(io::Error::last_os_error()); + } + Ok(()) +} + +/// Send one message (+ optionally one fd as `SCM_RIGHTS`) as a single SEQPACKET datagram. +/// Atomic per message, so concurrent senders on the same socket (the capture thread's imports, +/// the encode thread's releases) need no lock. `MSG_NOSIGNAL` turns a dead peer into `EPIPE` +/// instead of `SIGPIPE`. +pub fn send( + sock: BorrowedFd, + msg: &T, + pass_fd: Option, +) -> io::Result<()> { + let body = + serde_json::to_vec(msg).map_err(|e| io::Error::new(io::ErrorKind::InvalidData, e))?; + debug_assert!( + !body.is_empty(), + "zero-length messages are reserved for EOF" + ); + if body.len() > MAX_MSG { + return Err(io::Error::new( + io::ErrorKind::InvalidData, + "zerocopy proto message too large", + )); + } + let mut iov = libc::iovec { + iov_base: body.as_ptr() as *mut libc::c_void, + iov_len: body.len(), + }; + // Control buffer for one fd: CMSG_SPACE(4) = 24 bytes on 64-bit; [u64; 4] gives 32 bytes at + // the 8-byte alignment `cmsghdr` requires. + let mut cmsg_store = [0u64; 4]; + // SAFETY: `mhdr` is a plain-old-data C struct for which all-zero is a valid value. + let mut mhdr: libc::msghdr = unsafe { std::mem::zeroed() }; + mhdr.msg_iov = &mut iov; + mhdr.msg_iovlen = 1; + if let Some(fd) = pass_fd { + mhdr.msg_control = cmsg_store.as_mut_ptr() as *mut libc::c_void; + // SAFETY: `CMSG_SPACE`/`CMSG_LEN` are pure size computations (no memory access). + // `CMSG_FIRSTHDR(&mhdr)` returns a pointer into `cmsg_store` (non-null: msg_controllen + // ≥ one cmsghdr), which is live, 8-aligned, and large enough (32 ≥ CMSG_SPACE(4) = 24) + // for the header fields and the 4-byte fd written via `CMSG_DATA`; `write_unaligned` + // handles the data area's byte alignment. All writes stay within `cmsg_store`, which + // outlives the synchronous `sendmsg` below. + unsafe { + mhdr.msg_controllen = libc::CMSG_SPACE(4) as _; + let c = libc::CMSG_FIRSTHDR(&mhdr); + (*c).cmsg_level = libc::SOL_SOCKET; + (*c).cmsg_type = libc::SCM_RIGHTS; + (*c).cmsg_len = libc::CMSG_LEN(4) as _; + std::ptr::write_unaligned(libc::CMSG_DATA(c) as *mut i32, fd.as_raw_fd()); + } + } + // SAFETY: `sock` is the caller's live socket; `mhdr` points at the live `iov` (over `body`, + // which outlives the call) and — when an fd is passed — at `cmsg_store` (ditto). `sendmsg` + // only reads these buffers. The kernel dups the fd into the message; our `BorrowedFd` stays + // owned by the caller. + let n = unsafe { libc::sendmsg(sock.as_raw_fd(), &mhdr, libc::MSG_NOSIGNAL) }; + if n < 0 { + return Err(io::Error::last_os_error()); + } + if n as usize != body.len() { + return Err(io::Error::new( + io::ErrorKind::WriteZero, + "short sendmsg on SEQPACKET socket", + )); + } + Ok(()) +} + +/// Receive one message (+ up to one `SCM_RIGHTS` fd). `buf` is a caller-owned scratch buffer +/// (grown to [`MAX_MSG`] once, then reused frame to frame). Errors: +/// `UnexpectedEof` = the peer is gone; `WouldBlock` = the [`set_recv_timeout`] expired. +pub fn recv( + sock: BorrowedFd, + buf: &mut Vec, +) -> io::Result<(T, Option)> { + buf.resize(MAX_MSG, 0); + let mut iov = libc::iovec { + iov_base: buf.as_mut_ptr() as *mut libc::c_void, + iov_len: buf.len(), + }; + let mut cmsg_store = [0u64; 4]; + // SAFETY: `mhdr` is a plain-old-data C struct for which all-zero is a valid value. + let mut mhdr: libc::msghdr = unsafe { std::mem::zeroed() }; + mhdr.msg_iov = &mut iov; + mhdr.msg_iovlen = 1; + mhdr.msg_control = cmsg_store.as_mut_ptr() as *mut libc::c_void; + mhdr.msg_controllen = std::mem::size_of_val(&cmsg_store) as _; + // SAFETY: `sock` is the caller's live socket. `recvmsg` writes at most `iov_len` bytes into + // `buf` (live for the call) and at most `msg_controllen` control bytes into `cmsg_store` + // (live, 8-aligned). `MSG_CMSG_CLOEXEC` makes any received fd CLOEXEC atomically. + let n = unsafe { libc::recvmsg(sock.as_raw_fd(), &mut mhdr, libc::MSG_CMSG_CLOEXEC) }; + if n < 0 { + return Err(io::Error::last_os_error()); + } + if n == 0 { + return Err(io::Error::new( + io::ErrorKind::UnexpectedEof, + "zerocopy proto peer closed", + )); + } + // Collect a passed fd (if any) BEFORE any early return below, so it can't leak. + let mut got_fd: Option = None; + // SAFETY: `CMSG_FIRSTHDR`/`CMSG_NXTHDR` walk the control area the kernel just wrote inside + // `cmsg_store` (bounded by the updated `mhdr.msg_controllen`), returning either null or a + // pointer to a complete `cmsghdr` within it — each dereference reads kernel-initialized + // fields in bounds. For an `SCM_RIGHTS` cmsg the data area holds whole `i32` fds; we read the + // first via `read_unaligned`. The kernel gave us ownership of that fd (it is a fresh + // descriptor in our table), so `OwnedFd::from_raw_fd` takes sole ownership — any previously + // collected `got_fd` is dropped (closed) first, so nothing leaks even with multiple cmsgs. + unsafe { + let mut c = libc::CMSG_FIRSTHDR(&mhdr); + while !c.is_null() { + if (*c).cmsg_level == libc::SOL_SOCKET && (*c).cmsg_type == libc::SCM_RIGHTS { + let fd = std::ptr::read_unaligned(libc::CMSG_DATA(c) as *const i32); + if fd >= 0 { + got_fd = Some(OwnedFd::from_raw_fd(fd)); + } + } + c = libc::CMSG_NXTHDR(&mhdr, c); + } + } + if mhdr.msg_flags & libc::MSG_TRUNC != 0 { + return Err(io::Error::new( + io::ErrorKind::InvalidData, + "zerocopy proto message truncated", + )); + } + let msg = serde_json::from_slice(&buf[..n as usize]) + .map_err(|e| io::Error::new(io::ErrorKind::InvalidData, e))?; + Ok((msg, got_fd)) +} + +#[cfg(test)] +mod tests { + use super::*; + use std::io::{Read, Write}; + use std::os::fd::AsFd; + + #[test] + fn round_trip_no_fd() { + let (a, b) = socketpair_seqpacket().unwrap(); + let mut buf = Vec::new(); + let req = Request::Import { + key: 0xdead_beef_u64, + kind: ImportKind::TiledNv12, + width: 5120, + height: 1440, + fourcc: 0x3432_5258, + modifier: Some(0x0300_0000_0000_1234), + offset: 0, + stride: 5120 * 4, + has_fd: false, + }; + send(a.as_fd(), &req, None).unwrap(); + let (got, fd) = recv::(b.as_fd(), &mut buf).unwrap(); + assert_eq!(got, req); + assert!(fd.is_none()); + + let reply = Reply::Frame { + id: 7, + desc: Some(BufferDesc { + width: 5120, + height: 1440, + y_handle: vec![1u8; 64], + y_pitch: 5632, + uv: Some((vec![2u8; 64], 5632)), + }), + }; + send(b.as_fd(), &reply, None).unwrap(); + let (got, fd) = recv::(a.as_fd(), &mut buf).unwrap(); + assert_eq!(got, reply); + assert!(fd.is_none()); + } + + #[test] + fn passes_an_fd() { + let (a, b) = socketpair_seqpacket().unwrap(); + let mut buf = Vec::new(); + // A pipe stands in for a dmabuf: pass the read end, write through the original write end, + // and read the bytes back through the RECEIVED fd. + let (mut pr, mut pw) = std::io::pipe().unwrap(); + send(a.as_fd(), &Request::ClearCache, Some(pr.as_fd())).unwrap(); + let (got, fd) = recv::(b.as_fd(), &mut buf).unwrap(); + assert_eq!(got, Request::ClearCache); + let fd = fd.expect("fd should have been passed"); + pw.write_all(b"hello").unwrap(); + drop(pw); + let mut file = std::fs::File::from(fd); + let mut s = String::new(); + file.read_to_string(&mut s).unwrap(); + assert_eq!(s, "hello"); + // The original read end still works independently of the passed dup. + let mut nothing = [0u8; 1]; + assert_eq!(pr.read(&mut nothing).unwrap(), 0); + } + + #[test] + fn eof_when_peer_closes() { + let (a, b) = socketpair_seqpacket().unwrap(); + drop(a); + let mut buf = Vec::new(); + let err = recv::(b.as_fd(), &mut buf).unwrap_err(); + assert_eq!(err.kind(), io::ErrorKind::UnexpectedEof); + } + + #[test] + fn send_to_dead_peer_is_epipe_not_sigpipe() { + let (a, b) = socketpair_seqpacket().unwrap(); + drop(b); + let err = send(a.as_fd(), &Request::ClearCache, None).unwrap_err(); + // MSG_NOSIGNAL: a dead peer surfaces as EPIPE (BrokenPipe), never a process-killing signal. + assert_eq!(err.kind(), io::ErrorKind::BrokenPipe); + } + + #[test] + fn recv_timeout_fires() { + let (a, _b) = socketpair_seqpacket().unwrap(); + set_recv_timeout(a.as_fd(), Some(Duration::from_millis(50))).unwrap(); + let mut buf = Vec::new(); + let err = recv::(a.as_fd(), &mut buf).unwrap_err(); + assert!( + matches!( + err.kind(), + io::ErrorKind::WouldBlock | io::ErrorKind::TimedOut + ), + "unexpected error kind: {err:?}" + ); + } +} diff --git a/crates/punktfunk-host/src/linux/zerocopy/vulkan.rs b/crates/punktfunk-host/src/linux/zerocopy/vulkan.rs index 274c784..560c110 100644 --- a/crates/punktfunk-host/src/linux/zerocopy/vulkan.rs +++ b/crates/punktfunk-host/src/linux/zerocopy/vulkan.rs @@ -302,6 +302,23 @@ impl VkBridge { Ok(()) } + /// Drop the cached import for `fd` (the PipeWire buffer it wrapped is gone — pool recycle / + /// renegotiation — or the caller is about to store a different dmabuf under the same slot). + /// Without this the cache could serve a stale imported buffer for a reused fd number, or + /// leak an entry per recycled pool buffer. + pub fn forget_fd(&mut self, fd: i32) { + if let Some(s) = self.src_cache.remove(&fd) { + // SAFETY: `s.buffer`/`s.memory` were created by this bridge's `import_src` and are + // exclusively owned by the removed cache entry, so each is destroyed exactly once. + // No GPU work can still reference them: every `import_linear` fence-waits its copy to + // completion before returning, and this runs on the same single owning thread. + unsafe { + self.device.destroy_buffer(s.buffer, None); + self.device.free_memory(s.memory, None); + } + } + } + /// Bridge one LINEAR dmabuf frame into a pooled CUDA buffer: GPU copy dmabuf→exportable, /// then pitched CUDA copy exportable→`pool` buffer. pub fn import_linear( diff --git a/crates/punktfunk-host/src/linux/zerocopy/worker.rs b/crates/punktfunk-host/src/linux/zerocopy/worker.rs new file mode 100644 index 0000000..7d2e933 --- /dev/null +++ b/crates/punktfunk-host/src/linux/zerocopy/worker.rs @@ -0,0 +1,465 @@ +//! The isolated zero-copy GPU-import worker (`punktfunk-host zerocopy-worker`; design: +//! [`design/zerocopy-worker-isolation.md`]). It owns the fragile driver stack — the headless +//! EGLDisplay + GL context, the CUDA context, and the Vulkan bridge — so that a driver fault on a +//! producer-invalidated dmabuf (the `cuGraphicsMapResources` SIGSEGV the F44 Game→Desktop switch +//! reproduced) kills THIS process, not the streaming host. The host observes the dead socket, +//! fails the frame cleanly, and its existing capture-loss rebuild takes over. +//! +//! One worker serves one capture (spawned per `pipewire_thread`). It exits on socket EOF — which +//! only happens after the capturer AND every in-flight frame on the host side are gone, so pooled +//! device memory is never freed under a frame the host still reads. + +// Every `unsafe` block in this file carries a `// SAFETY:` proof; enforce it (unsafe-proof program). +#![deny(clippy::undocumented_unsafe_blocks)] + +use super::cuda::{self, CUdeviceptr, DeviceBuffer}; +use super::egl::{DmabufPlane, EglImporter}; +use super::proto::{self, BufferDesc, ImportKind, Reply, Request}; +use anyhow::{bail, Context, Result}; +use std::collections::{HashMap, VecDeque}; +use std::io; +use std::os::fd::{AsFd, AsRawFd, FromRawFd, OwnedFd}; + +/// Cap on cached per-key dmabuf fds. PipeWire buffer pools are ≤ ~16 buffers; the cap only +/// matters if a misbehaving producer churns buffers without a renegotiation. +const FD_CACHE_CAP: usize = 64; + +/// Entry point for the hidden `zerocopy-worker` subcommand. `args` are the subcommand's own +/// arguments (`--fd N`, default 3 — the socket end the spawning host `dup2`'d in). +pub fn run_from_args(args: &[String]) -> Result<()> { + let fd: i32 = args + .iter() + .skip_while(|a| *a != "--fd") + .nth(1) + .map(|s| s.parse()) + .transpose() + .context("parse --fd")? + .unwrap_or(3); + // SAFETY: the spawning host `dup2`'d its socketpair end onto exactly this fd number before + // exec (the subcommand's contract) and nothing else in this fresh process owns it, so + // `OwnedFd` takes sole ownership and closes it exactly once at exit. + let sock = unsafe { OwnedFd::from_raw_fd(fd) }; + run(sock) +} + +/// Bring up the GPU stack, report readiness, and serve until the host goes away. +fn run(sock: OwnedFd) -> Result<()> { + let importer = match EglImporter::new() { + Ok(i) => i, + Err(e) => { + // Init failure is an ANSWER, not a crash: the host falls back to the CPU path, + // exactly like an in-process `EglImporter::new()` failure. + let _ = proto::send( + sock.as_fd(), + &Reply::InitErr { + message: format!("{e:#}"), + }, + None, + ); + return Ok(()); + } + }; + proto::send( + sock.as_fd(), + &Reply::Ready { + version: proto::PROTO_VERSION, + }, + None, + ) + .context("send Ready")?; + tracing::info!(pid = std::process::id(), "zerocopy import worker ready"); + let mut backend = EglBackend::new(importer); + serve(&sock, &mut backend) +} + +/// What [`serve`] needs from an import implementation — split out so the dispatch loop is +/// unit-testable without a GPU. +pub(crate) trait ImportBackend { + fn modifiers(&mut self, fourcc: u32) -> Vec; + /// Answers with [`Reply::Frame`] (buffer id + [`BufferDesc`] iff first delivery of that id), + /// [`Reply::NeedFd`] (this side lacks the key's fd — host resends it once), or [`Reply::Err`]. + fn import(&mut self, req: &ImportReq, fd: Option) -> Reply; + fn release(&mut self, id: u32); + fn clear_cache(&mut self); +} + +/// The [`Request::Import`] fields, destructured for [`ImportBackend::import`]. +pub(crate) struct ImportReq { + pub key: u64, + pub kind: ImportKind, + pub width: u32, + pub height: u32, + pub fourcc: u32, + pub modifier: Option, + pub offset: u32, + pub stride: u32, + pub has_fd: bool, +} + +/// The request loop. Returns `Ok(())` on host EOF (normal end-of-life); any other socket error +/// propagates (the process exits — the host treats it like a death, which it is). +pub(crate) fn serve(sock: &OwnedFd, backend: &mut dyn ImportBackend) -> Result<()> { + let mut buf = Vec::new(); + loop { + let (req, fd) = match proto::recv::(sock.as_fd(), &mut buf) { + Ok(v) => v, + Err(e) if e.kind() == io::ErrorKind::UnexpectedEof => return Ok(()), + Err(e) => return Err(e).context("worker recv"), + }; + match req { + Request::Modifiers { fourcc } => { + let reply = Reply::Modifiers { + modifiers: backend.modifiers(fourcc), + }; + if send_or_eof(sock, &reply)? { + return Ok(()); + } + } + Request::Import { + key, + kind, + width, + height, + fourcc, + modifier, + offset, + stride, + has_fd, + } => { + let req = ImportReq { + key, + kind, + width, + height, + fourcc, + modifier, + offset, + stride, + has_fd, + }; + let reply = backend.import(&req, fd); + if send_or_eof(sock, &reply)? { + return Ok(()); + } + } + Request::Release { id } => backend.release(id), + Request::ClearCache => backend.clear_cache(), + } + } +} + +/// Send a reply; `Ok(true)` means the host is gone (EPIPE) and the loop should end quietly. +fn send_or_eof(sock: &OwnedFd, reply: &Reply) -> Result { + match proto::send(sock.as_fd(), reply, None) { + Ok(()) => Ok(false), + Err(e) if e.kind() == io::ErrorKind::BrokenPipe => Ok(true), + Err(e) => Err(e).context("worker send"), + } +} + +/// The real backend: the in-process [`EglImporter`] plus the cross-process bookkeeping — +/// per-key dmabuf fds, in-flight frames (held until `Release`), and stable buffer ids. +struct EglBackend { + importer: EglImporter, + /// The dmabuf fd for each host key (`st_ino`), kept because the tiled path re-imports the fd + /// every frame (`eglCreateImage`) and the LINEAR path caches per fd inside the Vulkan bridge. + fds: HashMap, + /// Insertion order of `fds` keys for the LRU cap. + fd_lru: VecDeque, + /// Frames delivered to the host and not yet released — holding the `DeviceBuffer` is what + /// keeps its device memory alive (pool `Arc`s) while the host encodes from it. + inflight: HashMap, + /// Buffer id per device allocation. Valid only within one pool generation: pools never free + /// allocations while alive, so a device VA can't repeat until a size change replaces the pool + /// — at which point [`Self::note_dims`] clears this map (ids themselves are never reused; + /// `next_id` only counts up). + ids: HashMap, + next_id: u32, + /// The (kind, width, height) of the last import — a change means the importer replaced its + /// pool, invalidating the VA→id map (see [`Self::ids`]). + last_shape: Option<(ImportKind, u32, u32)>, +} + +impl EglBackend { + fn new(importer: EglImporter) -> EglBackend { + EglBackend { + importer, + fds: HashMap::new(), + fd_lru: VecDeque::new(), + inflight: HashMap::new(), + ids: HashMap::new(), + next_id: 0, + last_shape: None, + } + } + + /// Store (or replace) the cached fd for `key`, evicting beyond the cap. A replaced or + /// evicted fd is first forgotten by the Vulkan bridge so its per-fd import can't go stale. + fn store_fd(&mut self, key: u64, fd: OwnedFd) { + if let Some(old) = self.fds.insert(key, fd) { + self.importer.forget_linear_fd(old.as_raw_fd()); + self.fd_lru.retain(|k| *k != key); + } + self.fd_lru.push_back(key); + while self.fds.len() > FD_CACHE_CAP { + let Some(oldest) = self.fd_lru.pop_front() else { + break; + }; + if let Some(old) = self.fds.remove(&oldest) { + self.importer.forget_linear_fd(old.as_raw_fd()); + } + } + } + + /// Clear the VA→id map when the importer is about to replace its per-size pool (see + /// [`Self::ids`]). + fn note_dims(&mut self, kind: ImportKind, width: u32, height: u32) { + if self.last_shape != Some((kind, width, height)) { + self.last_shape = Some((kind, width, height)); + self.ids.clear(); + } + } +} + +impl ImportBackend for EglBackend { + fn modifiers(&mut self, fourcc: u32) -> Vec { + self.importer.supported_modifiers(fourcc) + } + + fn import(&mut self, req: &ImportReq, fd: Option) -> Reply { + if let Some(fd) = fd { + self.store_fd(req.key, fd); + } else if req.has_fd { + return Reply::Err { + message: "Import said has_fd but no fd arrived".into(), + }; + } + let Some(raw) = self.fds.get(&req.key).map(|f| f.as_raw_fd()) else { + // We no longer hold this buffer's fd (LRU eviction / cache desync) — ask the host to + // resend it rather than failing the frame. + return Reply::NeedFd; + }; + match self.import_inner(req, raw) { + Ok((id, desc)) => Reply::Frame { id, desc }, + Err(e) => Reply::Err { + message: format!("{e:#}"), + }, + } + } + + fn release(&mut self, id: u32) { + if self.inflight.remove(&id).is_none() { + tracing::warn!(id, "release for a frame not in flight (host/worker desync)"); + } + } + + fn clear_cache(&mut self) { + for (_, fd) in self.fds.drain() { + self.importer.forget_linear_fd(fd.as_raw_fd()); + } + self.fd_lru.clear(); + self.importer.clear_linear_cache(); + } +} + +impl EglBackend { + /// The fallible core of [`ImportBackend::import`], once the fd for `req.key` is resolved. + fn import_inner(&mut self, req: &ImportReq, raw: i32) -> Result<(u32, Option)> { + let plane = DmabufPlane { + fd: raw, + offset: req.offset, + stride: req.stride, + }; + self.note_dims(req.kind, req.width, req.height); + let buf = match req.kind { + ImportKind::Tiled => { + self.importer + .import(&plane, req.width, req.height, req.fourcc, req.modifier)? + } + ImportKind::TiledNv12 => self.importer.import_nv12( + &plane, + req.width, + req.height, + req.fourcc, + req.modifier, + )?, + ImportKind::Linear => self.importer.import_linear(&plane, req.width, req.height)?, + }; + // Assign / look up the buffer's id and export its CUDA IPC identity on first delivery. + cuda::make_current()?; + let (id, desc) = match self.ids.get(&buf.ptr) { + Some(&id) => (id, None), + None => { + let id = self.next_id; + self.next_id = self.next_id.wrapping_add(1); + let y_handle = cuda::ipc_export(buf.ptr)?.to_vec(); + let uv = match buf.uv { + Some((uv_ptr, uv_pitch)) => { + Some((cuda::ipc_export(uv_ptr)?.to_vec(), uv_pitch)) + } + None => None, + }; + self.ids.insert(buf.ptr, id); + ( + id, + Some(BufferDesc { + width: buf.width, + height: buf.height, + y_handle, + y_pitch: buf.pitch, + uv, + }), + ) + } + }; + if self.inflight.insert(id, buf).is_some() { + // A pool never hands out a buffer that hasn't been recycled, so a duplicate id means + // corrupted bookkeeping — fail the import rather than alias two frames. + bail!("buffer id {id} already in flight"); + } + Ok((id, desc)) + } +} + +#[cfg(test)] +mod tests { + use super::*; + use std::sync::mpsc; + + /// Records calls; import behavior is scripted per key. + struct MockBackend { + calls: mpsc::Sender, + next: u32, + } + + impl ImportBackend for MockBackend { + fn modifiers(&mut self, fourcc: u32) -> Vec { + let _ = self.calls.send(format!("modifiers:{fourcc}")); + vec![7, 8, 9] + } + fn import(&mut self, req: &ImportReq, fd: Option) -> Reply { + let _ = self.calls.send(format!( + "import:key={} kind={:?} fd={}", + req.key, + req.kind, + fd.is_some() + )); + if req.key == 0xbad { + return Reply::Err { + message: "scripted failure".into(), + }; + } + if req.key == 0xfeed && !req.has_fd { + return Reply::NeedFd; + } + let id = self.next; + self.next += 1; + let desc = (id == 0).then(|| BufferDesc { + width: req.width, + height: req.height, + y_handle: vec![0u8; 64], + y_pitch: 256, + uv: None, + }); + Reply::Frame { id, desc } + } + fn release(&mut self, id: u32) { + let _ = self.calls.send(format!("release:{id}")); + } + fn clear_cache(&mut self) { + let _ = self.calls.send("clear".into()); + } + } + + fn start_server() -> ( + OwnedFd, + mpsc::Receiver, + std::thread::JoinHandle>, + ) { + let (host, worker) = proto::socketpair_seqpacket().unwrap(); + let (tx, rx) = mpsc::channel(); + let join = std::thread::spawn(move || { + let mut backend = MockBackend { calls: tx, next: 0 }; + serve(&worker, &mut backend) + }); + (host, rx, join) + } + + fn import_req(key: u64, has_fd: bool) -> Request { + Request::Import { + key, + kind: ImportKind::Tiled, + width: 64, + height: 64, + fourcc: 1, + modifier: None, + offset: 0, + stride: 256, + has_fd, + } + } + + #[test] + fn dispatch_and_eof() { + let (host, rx, join) = start_server(); + let mut buf = Vec::new(); + + proto::send(host.as_fd(), &Request::Modifiers { fourcc: 42 }, None).unwrap(); + let (reply, _) = proto::recv::(host.as_fd(), &mut buf).unwrap(); + assert_eq!( + reply, + Reply::Modifiers { + modifiers: vec![7, 8, 9] + } + ); + + // First import delivers the desc; the second (same mock id sequence continues) doesn't. + proto::send(host.as_fd(), &import_req(1, false), None).unwrap(); + let (reply, _) = proto::recv::(host.as_fd(), &mut buf).unwrap(); + match reply { + Reply::Frame { + id: 0, + desc: Some(_), + } => {} + other => panic!("unexpected reply {other:?}"), + } + proto::send(host.as_fd(), &import_req(1, false), None).unwrap(); + let (reply, _) = proto::recv::(host.as_fd(), &mut buf).unwrap(); + assert_eq!(reply, Reply::Frame { id: 1, desc: None }); + + // A missing worker-side fd is a NeedFd reply (host resends), not a failure. + proto::send(host.as_fd(), &import_req(0xfeed, false), None).unwrap(); + let (reply, _) = proto::recv::(host.as_fd(), &mut buf).unwrap(); + assert_eq!(reply, Reply::NeedFd); + + // A failed import is an Err reply, not a dead worker. + proto::send(host.as_fd(), &import_req(0xbad, false), None).unwrap(); + let (reply, _) = proto::recv::(host.as_fd(), &mut buf).unwrap(); + match reply { + Reply::Err { message } => assert!(message.contains("scripted failure")), + other => panic!("unexpected reply {other:?}"), + } + + // Fire-and-forget ops reach the backend without replies. + proto::send(host.as_fd(), &Request::Release { id: 0 }, None).unwrap(); + proto::send(host.as_fd(), &Request::ClearCache, None).unwrap(); + + // Closing the host end terminates serve() cleanly. + drop(host); + join.join().unwrap().unwrap(); + + let calls: Vec = rx.iter().collect(); + assert_eq!( + calls, + vec![ + "modifiers:42", + "import:key=1 kind=Tiled fd=false", + "import:key=1 kind=Tiled fd=false", + "import:key=65261 kind=Tiled fd=false", // 0xfeed + "import:key=2989 kind=Tiled fd=false", // 0xbad + "release:0", + "clear", + ] + ); + } +} diff --git a/crates/punktfunk-host/src/main.rs b/crates/punktfunk-host/src/main.rs index cb05a16..6bfd7de 100644 --- a/crates/punktfunk-host/src/main.rs +++ b/crates/punktfunk-host/src/main.rs @@ -181,6 +181,11 @@ fn real_main() -> Result<()> { // Zero-copy FFI/GPU probe: init the EGL importer + CUDA context (no capture needed). #[cfg(target_os = "linux")] Some("zerocopy-probe") => zerocopy::probe(), + // Hidden: the isolated GPU-import worker the capture path spawns from /proc/self/exe + // (design/zerocopy-worker-isolation.md) — never run by hand; --fd names the inherited + // socketpair end. + #[cfg(target_os = "linux")] + Some("zerocopy-worker") => zerocopy::worker::run_from_args(&args[1..]), // NV12 colour self-test (no display/capture needed): convert a known RGBA pattern to NV12 // on the GPU and compare against a BT.709 limited-range reference. Validates the Tier 2A // `PUNKTFUNK_NV12` convert is colour-correct. Prints PASS/FAIL + max Y/U/V error. diff --git a/design/zerocopy-hardening-handoff.md b/design/zerocopy-hardening-handoff.md index 158b59b..60741c2 100644 --- a/design/zerocopy-hardening-handoff.md +++ b/design/zerocopy-hardening-handoff.md @@ -1,6 +1,15 @@ # Zero-copy capture hardening — issue handoff -> **Status: HANDOFF — issue description only (2026-07-06).** This document describes a reproduced +> **Status: FIXED + validated (2026-07-06).** The fix is implemented and on-glass validated — see +> [`zerocopy-worker-isolation.md`](zerocopy-worker-isolation.md): the GPU import (tiled EGL/GL→CUDA +> *and* LINEAR Vulkan→CUDA) now runs in a per-capture **worker subprocess** (CUDA-IPC frame +> hand-off), so this driver SIGSEGV kills the worker and the host degrades to its capture-loss +> rebuild; plus in-process teardown-order fixes and a poison/latch path replacing the corrupt +> tiled→CPU fallback. Validated on the RTX 5070 Ti/GNOME box: worker path streams at p50 1.30 ms, +> and a `kill -9` of the worker mid-stream is survived + recovered (fresh worker in ~185 ms, +> streaming resumes). The description below is kept as the issue record. +> +> *(Original handoff intro:)* This document describes a reproduced > host **SIGSEGV** in the Linux zero-copy capture path. It deliberately does **not** prescribe a fix — > the next agent plans the implementation. Everything below is observed fact + root-cause analysis; > the "Considerations / open questions" section frames the solution space without committing to one. diff --git a/design/zerocopy-worker-isolation.md b/design/zerocopy-worker-isolation.md new file mode 100644 index 0000000..62d5300 --- /dev/null +++ b/design/zerocopy-worker-isolation.md @@ -0,0 +1,163 @@ +# Zero-copy capture hardening — GPU-import worker isolation + +> **Status: IMPLEMENTED + on-glass validated (2026-07-06).** This is the implementation +> plan + decision record for the crash described in +> [`zerocopy-hardening-handoff.md`](zerocopy-hardening-handoff.md) (host SIGSEGV inside +> `libnvidia-eglcore` via `cuGraphicsMapResources` when the compositor invalidated an imported +> dmabuf mid-map, observed on the Bazzite F44 Game→Desktop switch). Validated on the RTX 5070 Ti / +> GNOME box (.21): the isolated worker carries frames at **p50 1.30 ms** end-to-end (NV12, 1800 +> frames 0-mismatched), and a `kill -9` of the worker mid-stream is survived by the host and +> recovered — poison → `capture lost — rebuilding pipeline in place` → a fresh worker in **~185 ms** +> → streaming resumes (2385 frames, 0 mismatched, one 33 ms blip at the rebuild seam). See §6. + +## 1. The decision: isolate, don't (only) prevent + +The handoff's §9 framed two directions — *prevent the stale resource* vs *isolate the crash*. +The audit (§3 below) shows our per-frame lifetime discipline is already correct: the `EGLImage` +is created and destroyed strictly inside the PipeWire `on_process` callback while the buffer is +held (not requeued), and the CUDA-registered textures are **our own GL render targets**, never +wrappers around producer buffers. The invalidation that crashed the host is **external** — +a compositor crash (or GPU channel wreckage from the surrounding plasmashell/Xwayland core dumps) +yanked the dmabuf's GPU-side state while the driver executed our in-flight GL sampling + CUDA map. +No in-process ordering fix can close that race, and a driver SIGSEGV is not catchable. + +So the fix is **process isolation**: the entire `EglImporter` (tiled dmabuf → EGL/GL → CUDA *and* +LINEAR dmabuf → Vulkan bridge → CUDA) moves into a small per-capture **worker subprocess**. If the +driver faults, the *worker* dies; the host observes a dead socket, fails the frame/capture cleanly, +and the existing capture-loss rebuild path (`gamestream/stream.rs`, `punktfunk1.rs`) takes over — +which is exactly what already happens today on the safe SHM path when a compositor goes away. + +What is deliberately **not** isolated: + +- **SHM/CPU capture** — no GPU import, nothing to contain. +- **VAAPI passthrough** (AMD/Intel) — capture only `dup`s the dmabuf fd; the GPU import happens in + the encoder (Mesa VA, which reports errors rather than faulting; no observed crashes). Out of + scope here. +- **NVENC itself** — libavcodec/NVENC surface errors as return codes; if the GPU is globally + wedged the encoder errors and the session rebuilds. Isolating encode would mean shipping a + session-wide media-pipeline process, far beyond this fix. + +## 2. Architecture + +``` +host process worker process (punktfunk-host zerocopy-worker) +──────────── ─────────────────────────────────────────────── +PipeWire on_process EGLDisplay + GL ctx + CUDA ctx + VkBridge + │ dmabuf fd (held, fence-waited) │ + ├── IMPORT{key,geometry} + fd ──────────────▶│ eglCreateImage → GL blit/NV12 convert + │ (SCM_RIGHTS, first sight per key) │ → cuGraphicsMapResources → copy → unmap + │ │ → pooled CUDA buffer (cuMemAllocPitch) + │◀────────── FRAME{id [, ipc desc]} ─────────┤ exported ONCE via cuIpcGetMemHandle + │ host opens the IPC handle once, │ + │ wraps it as DeviceBuffer │ + ▼ │ +encode thread (NVENC) reads the device ptr │ keeps the DeviceBuffer in-flight + │ DeviceBuffer drop │ + └── RELEASE{id} ────────────────────────────▶│ returns the buffer to its pool +``` + +- **Transport**: a `socketpair(AF_UNIX, SOCK_SEQPACKET)` created before spawn; the child end is + `dup2`'d to fd 3 (`zerocopy-worker --fd 3`). SEQPACKET gives reliable, ordered, message-framed + delivery; dmabuf fds ride as `SCM_RIGHTS`. Messages are small serde_json bodies (~200 B/frame; + negligible at 240 fps). +- **Frame data never crosses the socket.** The worker's `BufferPool` allocations are exported once + each via `cuIpcGetMemHandle`; the host `cuIpcOpenMemHandle`s each exactly once (cached by buffer + id) and reuses the mapping as the pool recycles. Per frame the reply is just `{id}` — the copy + was already synced (`copy_blocking`) worker-side before the reply, so the host reads complete + pixels. The result is the same zero-CPU-touch path as before, plus one socket RTT (~tens of µs). +- **fd caching**: the host keys each PipeWire buffer by its dmabuf `st_ino` (unique per dma-buf + object) and sends the fd only on first sight; the worker keeps the received dup (tiled: for the + per-frame `eglCreateImage`; LINEAR: for the Vulkan `src_cache`). A format renegotiation + (`param_changed`) sends `CLEAR_CACHE`, dropping both sides' caches — this also fixes the + pre-existing LINEAR-path bug where `VkBridge::src_cache` was keyed by raw fd number and never + invalidated across pool recycles (§3, trigger b). Cache desync is self-healing: a worker that no + longer holds a key's fd (LRU eviction) answers `NeedFd` and the host retries once with the fd. +- **Lifetimes**: the worker holds each exported frame as a real `DeviceBuffer` in an in-flight map + until `RELEASE{id}` arrives, so the existing pool `Arc` machinery keeps device memory alive + across pool replacement while the host still reads it. Host-side, every remote `DeviceBuffer` + holds an `Arc` of the client's shared state (socket + IPC-mapping cache), so mappings are closed + only after the last in-flight frame drops. +- **Worker lifetime**: one worker per capture (per `pipewire_thread`), spawned from + `/proc/self/exe`. It exits on socket EOF; the host reaps children via a global sweep list (no + zombies). Host death ⇒ EOF ⇒ worker exit. + +### Failure semantics (the point of the exercise) + +| event | behavior | +|---|---| +| worker init fails (no GPU, EGL error) | handshake reports `init_err` → capture falls back to the CPU/SHM offer, same as `EglImporter::new()` failure today | +| driver SIGSEGV in the worker (the observed crash) | socket EOF → import fails with a *dead-worker* error → the capturer is **poisoned** → `next_frame`/`try_latest` return an error → the session's capture-loss rebuild runs (new capturer, new worker). **The host process survives.** | +| tiled import fails but worker alive (e.g. `EGL_BAD_MATCH` on one frame) | frame dropped; after 3 consecutive failures the capturer poisons → rebuild. It must **never** fall through to the CPU mmap path — mmap of a *tiled* dmabuf de-pads scrambled bytes (a pre-existing fallback bug; the CPU fallback was only ever correct for LINEAR). | +| LINEAR import fails | unchanged: fall back to the CPU mmap path in-stream (a LINEAR dmabuf is mappable), degraded not dead | +| repeated worker deaths | a process-wide latch (`note_gpu_import_death`, 3 consecutive deaths without a successful import between them) disables the GPU importer for the rest of the process — rebuilds renegotiate the SHM offer. Stops a wedged GPU stack from crash-looping the worker while still streaming (CPU path). A successful import resets the streak. | + +### Escape hatch + +`PUNKTFUNK_ZEROCOPY_INPROC=1` keeps the importer in-process (the pre-isolation behavior) for +debugging and A/B latency comparison. Default is the worker. + +## 3. Audit answers for handoff §5 (which triggers are actually reachable) + +- **Compositor crash / restart** — reachable (observed). Contained by the worker. +- **PipeWire buffer-pool recycle / renegotiation**: + - *Tiled EGL path*: **not reachable in code** — the `EGLImage` lives strictly inside + `on_process` while the buffer is held; the CUDA registrations wrap our own persistent GL + textures, not producer buffers. + - *LINEAR Vulkan path*: **reachable** — `VkBridge::src_cache` keyed by raw fd, never + invalidated: a pool teardown + fd-number reuse could serve a stale imported buffer (wrong + frame or driver fault), and old entries leaked. Fixed by st_ino keys + `CLEAR_CACHE` on + renegotiation + an LRU cap. +- **Virtual-output teardown / mode change racing an in-flight map** — same class as compositor + crash (external invalidation, another thread); contained by the worker. +- **Output removal** — ditto. + +## 4. In-process lifetime fixes (also shipped, they harden the worker itself) + +- `Nv12Blit::drop` deleted its GL textures **before** the struct fields dropped, i.e. while + `y_tex`/`uv_tex` were still CUDA-registered. Now `RegisteredTexture::release()` runs first + (unregister → delete), removing a driver-state hazard of exactly the class that crashed. +- `GlBlit` had **no** `Drop` — its GL program/VAO/FBO/textures leaked on every size change and on + importer teardown. Now mirrors `Nv12Blit` (release registrations, then delete GL objects). + +## 5. Residual risks, accepted + +- A worker death while the encode thread still holds an IPC-mapped frame: the exporting process is + gone; the host-side mapping stays open until the `DeviceBuffer` drops. CUDA surfaces this as a + copy error at worst (encode error → session rebuild), not a host fault. +- The VAAPI encoder's in-host VA dmabuf import (Mesa) keeps its current exposure; no NVIDIA-class + faults observed there. +- `cuIpcOpenMemHandle` requires same-device, different-process — both hold by construction. + +## 6. Validation + +- **GPU-less (CI / dev VM)**: protocol unit tests (framing, fd round-trip over a socketpair, + error propagation, dead-worker detection against a mock server, latch behavior); worker-spawn + failure path (spawning a non-worker exe ⇒ clean fallback). +- **On-glass (NVIDIA RTX 5070 Ti + GNOME/Mutter, .21, 2026-07-06)** — steps 1–2 **PASSED**: + 1. streamed `PUNKTFUNK_ZEROCOPY=1` through the worker (`zerocopy import worker ready` → + `zero-copy GPU import isolated in a worker process` → `dmabuf imported to CUDA … nv12=true`), + end-to-end **p50 1.30 ms** (1800 frames, 0 mismatched) — parity with the pre-isolation path; + 2. `kill -9` the worker mid-stream → host **survived**; the next import logged + `tiled GPU import lost — failing this capture for rebuild … Broken pipe … dead=true`, then + `capture lost — rebuilding pipeline in place, rebuild=1`, a **fresh worker (new pid) in + ~185 ms**, and streaming resumed (2385 frames, 0 mismatched; single 33 ms frame at the seam). + The `worker-ready` count was 2 (original + rebuild), confirming the respawn. + Still pending: 3. a real compositor kill/restart mid-stream on a KWin box (the exact original + trigger — a `kill -9` of the worker is a strictly harsher event, so this is corroboration not a + gap); 4. `nv12-selftest` (in-process path untouched). *Note: on a static virtual desktop the + dead-worker detection only fires once a new frame triggers an import — realistic (a running game + produces continuous frames) but it means an idle desktop can sit poisoned-but-quiet briefly.* + +## 7. Files + +- `crates/punktfunk-host/src/linux/zerocopy/proto.rs` — message types + SEQPACKET/SCM_RIGHTS I/O. +- `crates/punktfunk-host/src/linux/zerocopy/worker.rs` — worker main loop (`zerocopy-worker`), + backend trait (testable), EGL/CUDA backend. +- `crates/punktfunk-host/src/linux/zerocopy/client.rs` — `RemoteImporter` (spawn, handshake, IPC + mapping cache, release plumbing, reaping) + the `Importer` enum (Remote | InProc). +- `crates/punktfunk-host/src/linux/zerocopy/cuda.rs` — CUDA IPC entry points; remote-release + `DeviceBuffer`s. +- `crates/punktfunk-host/src/linux/zerocopy/egl.rs` — teardown-order fixes (§4). +- `crates/punktfunk-host/src/capture/linux/mod.rs` — `Importer` wiring, tiled-failure poisoning, + death latch, `CLEAR_CACHE` on renegotiation. +- `crates/punktfunk-host/src/main.rs` — the hidden `zerocopy-worker` subcommand.