refactor(host/W6.2): extract the Linux zero-copy GPU plumbing into the pf-zerocopy leaf crate
linux/zerocopy/* (CUDA context/buffers + EGL/Vulkan dmabuf import + the isolated import worker) and linux/dmabuf_fence.rs move wholesale into crates/pf-zerocopy, so the coming pf-frame vocabulary crate (FramePayload::Cuda owns a DeviceBuffer) and the pf-encode/pf-capture subsystem crates can reach the GPU plumbing without the host orchestrator in between (plan §W6). Content stays Linux-only; the crate compiles to an empty lib elsewhere, so dependents carry a plain dependency. drm_fourcc deliberately does NOT move: it consumes the frame vocabulary (PixelFormat), which sits ABOVE pf-zerocopy — it lives with capture for now and moves into pf-frame next. cuda's ffi re-export bumps pub(crate)->pub (the raw CUdeviceptr vocabulary is consumed across the crate boundary by the encode backends). A crate::zerocopy shim module keeps every existing path valid until capture/encode themselves move out. Verified: Linux clippy -D warnings (pf-zerocopy --all-targets + host nvenc,vulkan-encode,pyrowave --all-targets) + 17/17 pf-zerocopy tests + 321/321 host tests; Windows clippy nvenc,amf-qsv --all-targets Finished exit 0. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
@@ -0,0 +1,480 @@
|
||||
//! 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<()> {
|
||||
// The host execs this worker through its pinned exe fd (`client::self_exe`), so the kernel
|
||||
// derives our comm from the exec path's basename — a meaningless fd number. Rename so
|
||||
// `top`/`pkill` see the worker.
|
||||
// SAFETY: `PR_SET_NAME` copies at most 16 bytes from the given pointer; the C-string literal
|
||||
// is valid, NUL-terminated, and short enough. No pointer is retained past the call.
|
||||
unsafe {
|
||||
libc::prctl(libc::PR_SET_NAME, c"pf-zerocopy".as_ptr());
|
||||
}
|
||||
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::Tiled444 => self.importer.import_yuv444(
|
||||
&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",
|
||||
]
|
||||
);
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user