feat(host): isolate the zero-copy GPU import in a worker process
The tiled EGL/GL→CUDA import crashed the whole host (SIGSEGV inside libnvidia-eglcore via cuGraphicsMapResources) when the compositor invalidated an imported dmabuf mid-map — reproduced on the Bazzite F44 Game→Desktop switch (design/zerocopy-hardening-handoff.md). A driver SIGSEGV is uncatchable in-process, so the whole EglImporter (tiled EGL/GL→CUDA and LINEAR Vulkan→CUDA) now runs in a per-capture `zerocopy-worker` subprocess: dmabuf fds go over a SEQPACKET socketpair (SCM_RIGHTS, sent once per buffer keyed by dmabuf st_ino; NeedFd resend self-heals cache desync), frames come back as CUDA-IPC pooled device buffers (still zero-copy, +one socket RTT/frame). Worker death poisons the capturer so the existing capture-loss rebuild runs — the host survives; 3 consecutive deaths latch the GPU import off (CPU/SHM path). PUNKTFUNK_ZEROCOPY_INPROC=1 keeps the old in-process import for debugging/A-B. Also fixed along the way: a failed *tiled* import no longer falls through to the CPU mmap de-pad (which scrambled tiled bytes; LINEAR keeps the fallback); Nv12Blit dropped its GL textures while still CUDA-registered (unregister now runs first); GlBlit had no Drop at all (GL objects leaked per size change); VkBridge's per-fd src cache is now invalidated on renegotiation/eviction instead of never. Design: design/zerocopy-worker-isolation.md. Unit tests: 14 new (protocol fd-passing, worker dispatch, client handshake/death/NeedFd, death latch). On-glass validated on the RTX 5070 Ti/GNOME box (.21): the worker path streams at p50 1.30 ms (NV12, 1800 frames 0-mismatched, parity with the in-process path), 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). A real KWin compositor-crash repro is still pending (a worker kill -9 is strictly harsher, so it corroborates). Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
This commit is contained in:
@@ -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<AtomicBool>,
|
||||
/// 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<AtomicBool>,
|
||||
/// 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<AtomicBool>,
|
||||
negotiated: Arc<AtomicBool>,
|
||||
streaming: Arc<AtomicBool>,
|
||||
/// See [`PortalCapturer::broken`].
|
||||
broken: Arc<AtomicBool>,
|
||||
/// 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<CapturedFrame> {
|
||||
// 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)",
|
||||
// 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
|
||||
))
|
||||
} 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
|
||||
))
|
||||
));
|
||||
}
|
||||
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<Option<CapturedFrame>> {
|
||||
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<CapturedFrame> {
|
||||
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<AtomicBool>,
|
||||
/// Present when zero-copy is enabled on NVIDIA: imports a dmabuf → CUDA device buffer.
|
||||
importer: Option<crate::zerocopy::EglImporter>,
|
||||
/// 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<AtomicBool>,
|
||||
/// 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<crate::zerocopy::Importer>,
|
||||
/// 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:#}"),
|
||||
"dmabuf GPU import failed — falling back to the CPU copy path");
|
||||
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:#}"),
|
||||
"LINEAR dmabuf GPU import failed — falling back to the CPU copy path");
|
||||
gpu_import_broken = true;
|
||||
}
|
||||
}
|
||||
@@ -1138,6 +1223,7 @@ mod pipewire {
|
||||
active: Arc<AtomicBool>,
|
||||
negotiated: Arc<AtomicBool>,
|
||||
streaming: Arc<AtomicBool>,
|
||||
broken: Arc<AtomicBool>,
|
||||
zerocopy: bool,
|
||||
preferred: Option<(u32, u32, u32)>,
|
||||
quit_rx: pw::channel::Receiver<()>,
|
||||
@@ -1165,20 +1251,30 @@ 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() {
|
||||
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 {
|
||||
None
|
||||
};
|
||||
@@ -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();
|
||||
|
||||
@@ -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<HashMap<u32, Mapping>>,
|
||||
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<Vec<Child>> = 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<Shared>,
|
||||
child: Option<Child>,
|
||||
/// Reused receive scratch buffer (all replies are read by the single capture thread).
|
||||
rbuf: Vec<u8>,
|
||||
/// Dmabuf keys (`st_ino`) whose fd the worker already holds — the fd is passed only once.
|
||||
sent_keys: HashSet<u64>,
|
||||
}
|
||||
|
||||
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<RemoteImporter> {
|
||||
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<RemoteImporter> {
|
||||
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<Child>) -> Result<RemoteImporter> {
|
||||
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::<Reply>(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<u64> {
|
||||
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::<Reply>(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<u64>,
|
||||
) -> Result<DeviceBuffer> {
|
||||
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<u64>,
|
||||
) -> Result<DeviceBuffer> {
|
||||
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<DeviceBuffer> {
|
||||
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<u64>,
|
||||
) -> Result<DeviceBuffer> {
|
||||
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::<Reply>(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<u64> {
|
||||
// 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<Mapping> {
|
||||
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<Reply>) -> (RemoteImporter, thread::JoinHandle<Vec<Request>>) {
|
||||
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::<Request>(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<bool> = 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:?}"),
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -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<CUdeviceptr> {
|
||||
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<Arc<Mutex<PoolInner>>>,
|
||||
/// 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<Box<dyn FnOnce() + Send>>,
|
||||
}
|
||||
|
||||
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<dyn FnOnce() + Send>,
|
||||
) -> 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 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).
|
||||
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) {
|
||||
self.release();
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
@@ -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).
|
||||
|
||||
@@ -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<EglImporter>),
|
||||
}
|
||||
|
||||
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<Importer> {
|
||||
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<u64> {
|
||||
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<u64>,
|
||||
) -> anyhow::Result<DeviceBuffer> {
|
||||
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<u64>,
|
||||
) -> anyhow::Result<DeviceBuffer> {
|
||||
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<DeviceBuffer> {
|
||||
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());
|
||||
}
|
||||
}
|
||||
|
||||
@@ -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<u64>,
|
||||
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<u64>,
|
||||
},
|
||||
/// 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<BufferDesc>,
|
||||
},
|
||||
/// 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<u8>,
|
||||
pub y_pitch: usize,
|
||||
/// NV12 only: the interleaved chroma plane's `(handle, pitch)`.
|
||||
pub uv: Option<(Vec<u8>, 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<Duration>) -> 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::<timeval>()` 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::<libc::timeval>() 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<T: Serialize>(
|
||||
sock: BorrowedFd,
|
||||
msg: &T,
|
||||
pass_fd: Option<BorrowedFd>,
|
||||
) -> 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<T: DeserializeOwned>(
|
||||
sock: BorrowedFd,
|
||||
buf: &mut Vec<u8>,
|
||||
) -> io::Result<(T, Option<OwnedFd>)> {
|
||||
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<OwnedFd> = 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::<Request>(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::<Reply>(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::<Request>(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::<Reply>(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::<Reply>(a.as_fd(), &mut buf).unwrap_err();
|
||||
assert!(
|
||||
matches!(
|
||||
err.kind(),
|
||||
io::ErrorKind::WouldBlock | io::ErrorKind::TimedOut
|
||||
),
|
||||
"unexpected error kind: {err:?}"
|
||||
);
|
||||
}
|
||||
}
|
||||
@@ -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(
|
||||
|
||||
@@ -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<u64>;
|
||||
/// 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<OwnedFd>) -> 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<u64>,
|
||||
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::<Request>(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<bool> {
|
||||
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<u64, OwnedFd>,
|
||||
/// Insertion order of `fds` keys for the LRU cap.
|
||||
fd_lru: VecDeque<u64>,
|
||||
/// 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<u32, DeviceBuffer>,
|
||||
/// 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<CUdeviceptr, u32>,
|
||||
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<u64> {
|
||||
self.importer.supported_modifiers(fourcc)
|
||||
}
|
||||
|
||||
fn import(&mut self, req: &ImportReq, fd: Option<OwnedFd>) -> 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<BufferDesc>)> {
|
||||
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<String>,
|
||||
next: u32,
|
||||
}
|
||||
|
||||
impl ImportBackend for MockBackend {
|
||||
fn modifiers(&mut self, fourcc: u32) -> Vec<u64> {
|
||||
let _ = self.calls.send(format!("modifiers:{fourcc}"));
|
||||
vec![7, 8, 9]
|
||||
}
|
||||
fn import(&mut self, req: &ImportReq, fd: Option<OwnedFd>) -> 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<String>,
|
||||
std::thread::JoinHandle<Result<()>>,
|
||||
) {
|
||||
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::<Reply>(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::<Reply>(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::<Reply>(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::<Reply>(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::<Reply>(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<String> = 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",
|
||||
]
|
||||
);
|
||||
}
|
||||
}
|
||||
@@ -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.
|
||||
|
||||
@@ -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.
|
||||
|
||||
@@ -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.
|
||||
Reference in New Issue
Block a user