refactor(pf-zerocopy): split the worker rails out of the zerocopy vocabulary so a second worker can reuse them
The encode worker (design/gpu-priority-capability-worker.md) needs exactly what the zerocopy worker
already has — SEQPACKET framing, fds as SCM_RIGHTS, a pinned-exe spawn that survives an on-disk
replacement, and a reaper that never blocks session teardown on a wedged child — but it must NOT
inherit the zerocopy protocol. Its messages are its own and version independently.
So `imp/proto.rs` keeps the vocabulary (PROTO_VERSION, ImportKind, Request, Reply, BufferDesc) and
all transport moves to `imp/ipc.rs`, reachable as `pf_zerocopy::ipc`. No behaviour change for the
zerocopy worker: client.rs now calls `ipc::self_exe()`/`ipc::spawn_worker()` and keeps the same fd-3
dup2 slot, PR_SET_PDEATHSIG, kill-then-reap-outside-the-lock, bounded reap with a D-state re-park,
and per-generation zombie sweep it had before.
Two real changes underneath the move:
* The cmsg store was sized for exactly one fd (CMSG_SPACE(4) = 24 B). A multi-planar dmabuf can
carry up to four, so it is now CMSG_SPACE(4*4); `send_fds`/`recv_fds` take a slice while `send`
and `recv` keep their single-fd shapes as the fast path. An over-long fd list is rejected with
io::Error rather than asserting — that is how MAX_MSG overflow is already handled — and the
receive cap is enforced by the kernel through msg_controllen, so a 5-fd peer trips MSG_CTRUNC.
* The old recv loop read only the FIRST i32 of each SCM_RIGHTS control message. Nothing sends two
fds yet so it never fired, but every descriptor after the first in a multi-fd message would have
leaked into the process. It now reads all of them.
Spawn takes the executable path as a parameter instead of assuming /proc/self/exe. The zerocopy
worker keeps self-exec; the encode worker passes its own binary, which must be a separate FILE and
never a subcommand — a shared inode shares the file capability.
This commit is contained in:
@@ -1,27 +1,27 @@
|
||||
//! 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.
|
||||
//! `design/zerocopy-worker-isolation.md`): spawns the `zerocopy-worker` subprocess on the shared
|
||||
//! [`super::ipc`] rails, mirrors the [`super::egl::EglImporter`] entry points over the
|
||||
//! [`super::proto`] vocabulary, 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 super::ipc;
|
||||
use super::proto::{BufferDesc, ImportKind, Reply, Request, PROTO_VERSION};
|
||||
use anyhow::{bail, Context, Result};
|
||||
use std::collections::{HashMap, HashSet};
|
||||
use std::fs::File;
|
||||
use std::io;
|
||||
use std::os::fd::{AsFd, AsRawFd, BorrowedFd, OwnedFd};
|
||||
use std::os::unix::process::CommandExt;
|
||||
use std::path::{Path, PathBuf};
|
||||
use std::process::{Child, Command};
|
||||
use std::os::fd::{AsFd, BorrowedFd, OwnedFd};
|
||||
use std::path::Path;
|
||||
use std::process::Child;
|
||||
use std::sync::atomic::{AtomicBool, Ordering};
|
||||
use std::sync::{Arc, Mutex, OnceLock};
|
||||
use std::time::{Duration, Instant};
|
||||
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);
|
||||
@@ -79,110 +79,6 @@ fn close_mapping(m: &Mapping) {
|
||||
}
|
||||
}
|
||||
|
||||
/// 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, Instant)>> = Mutex::new(Vec::new());
|
||||
|
||||
/// How long past `REPLY_TIMEOUT` a parked worker may linger before it is force-killed. A worker
|
||||
/// wedged INSIDE a driver call never observes socket EOF, so `try_wait` alone would keep it (and
|
||||
/// its CUcontext + BufferPool — order hundreds of MB of VRAM) forever.
|
||||
const REAPER_KILL_DEADLINE: Duration = Duration::from_secs(20);
|
||||
|
||||
fn sweep_reaper() {
|
||||
// Partition under the lock; kill/reap OUTSIDE it. A worker wedged inside a driver ioctl sits
|
||||
// in D state and ignores SIGKILL — the old blocking `wait()` under the global mutex would
|
||||
// then park every later `spawn()` and `drop()` behind a process that may never die.
|
||||
let mut expired: Vec<Child> = Vec::new();
|
||||
{
|
||||
let mut list = REAPER.lock().unwrap();
|
||||
let now = Instant::now();
|
||||
let mut i = 0;
|
||||
while i < list.len() {
|
||||
if matches!(list[i].0.try_wait(), Ok(Some(_))) {
|
||||
list.swap_remove(i); // exited on its own → reaped
|
||||
} else if now.duration_since(list[i].1) > REAPER_KILL_DEADLINE {
|
||||
expired.push(list.swap_remove(i).0);
|
||||
} else {
|
||||
i += 1;
|
||||
}
|
||||
}
|
||||
}
|
||||
for mut c in expired {
|
||||
let _ = c.kill();
|
||||
// Bounded reap (~100 ms of polls): a SIGKILL'd process reaps near-instantly unless it is
|
||||
// in D state — then park it again (re-killing later is harmless) so a future sweep reaps
|
||||
// it once the driver unwedges, instead of blocking anyone here forever.
|
||||
let mut reaped = false;
|
||||
for _ in 0..10 {
|
||||
if matches!(c.try_wait(), Ok(Some(_))) {
|
||||
reaped = true;
|
||||
break;
|
||||
}
|
||||
std::thread::sleep(Duration::from_millis(10));
|
||||
}
|
||||
if !reaped {
|
||||
tracing::warn!(
|
||||
pid = c.id(),
|
||||
"zerocopy worker ignored SIGKILL (likely wedged in a driver call, D state) — \
|
||||
parked for a later sweep"
|
||||
);
|
||||
REAPER.lock().unwrap().push((c, Instant::now()));
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// Fd pinned to this process's own executable image, opened (once, lazily) via the
|
||||
/// `/proc/self/exe` magic link. The link names the running image's *inode*, not its path, so it
|
||||
/// resolves even after the installed binary was replaced or deleted — and exec'ing the fd (via
|
||||
/// [`fd_exec_path`]) then still runs byte-for-byte the build this process is. `current_exe()`
|
||||
/// instead readlinks to a path: after a package upgrade under a running host that path is
|
||||
/// "<path> (deleted)" and spawning it fails ENOENT — every capture then silently fell back to
|
||||
/// the CPU copy — and even while the path exists it may hold a newer build whose worker
|
||||
/// protocol mismatches this process.
|
||||
static SELF_EXE: OnceLock<Option<File>> = OnceLock::new();
|
||||
|
||||
fn self_exe() -> Option<BorrowedFd<'static>> {
|
||||
SELF_EXE
|
||||
.get_or_init(|| {
|
||||
let f = match File::open("/proc/self/exe") {
|
||||
Ok(f) => f,
|
||||
Err(e) => {
|
||||
tracing::warn!(
|
||||
error = %e,
|
||||
"cannot pin /proc/self/exe — worker spawns use the current_exe() path, \
|
||||
which breaks if this binary is replaced on disk"
|
||||
);
|
||||
return None;
|
||||
}
|
||||
};
|
||||
if f.as_raw_fd() != 3 {
|
||||
return Some(f);
|
||||
}
|
||||
// Fd 3 is the slot the spawn hands the worker its socket on (the `dup2` in
|
||||
// `spawn_exe`) — pinned there, the child would clobber it before exec resolves
|
||||
// `/proc/self/fd/3`. Re-number: 3 stays occupied by `f` during the clone, so the
|
||||
// duplicate cannot land on it.
|
||||
match f.try_clone() {
|
||||
Ok(clone) => Some(clone),
|
||||
Err(e) => {
|
||||
tracing::warn!(error = %e, "re-numbering the pinned exe fd off fd 3 failed");
|
||||
None
|
||||
}
|
||||
}
|
||||
})
|
||||
.as_ref()
|
||||
.map(|f| f.as_fd())
|
||||
}
|
||||
|
||||
/// `/proc/self/fd/<n>` — an exec'able path to `fd`'s inode. The kernel resolves it at exec time
|
||||
/// inside the forked child, whose fd table is a copy of ours (close-on-exec applies only once
|
||||
/// the exec succeeds), so it names the pinned inode no matter what sits at the file's original
|
||||
/// path by then.
|
||||
fn fd_exec_path(fd: BorrowedFd<'_>) -> PathBuf {
|
||||
PathBuf::from(format!("/proc/self/fd/{}", fd.as_raw_fd()))
|
||||
}
|
||||
|
||||
/// 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 {
|
||||
@@ -196,13 +92,17 @@ pub struct RemoteImporter {
|
||||
|
||||
impl RemoteImporter {
|
||||
/// Spawn the worker from this host binary and complete the readiness handshake. The worker
|
||||
/// is exec'd through the pinned `SELF_EXE` fd, so it is always the exact image this
|
||||
/// is exec'd through the pinned [`ipc::self_exe`] fd, so it is always the exact image this
|
||||
/// process runs — even after the installed binary was replaced mid-flight. An `Err` here
|
||||
/// means "no isolated zero-copy available" — callers fall back to the CPU path, exactly like
|
||||
/// an in-process `EglImporter::new()` failure.
|
||||
///
|
||||
/// Self-exec is right *here* — host and worker are the same build by construction, so the
|
||||
/// version check is a formality. It is the one thing the capability-carrying encode worker
|
||||
/// must NOT copy: a shared inode shares the file capability.
|
||||
pub fn spawn() -> Result<RemoteImporter> {
|
||||
match self_exe() {
|
||||
Some(fd) => Self::spawn_exe(&fd_exec_path(fd)),
|
||||
match ipc::self_exe() {
|
||||
Some(exe) => Self::spawn_exe(&exe.exec_path()),
|
||||
None => Self::spawn_exe(
|
||||
&std::env::current_exe().context("resolve /proc/self/exe for the worker")?,
|
||||
),
|
||||
@@ -211,45 +111,10 @@ impl RemoteImporter {
|
||||
|
||||
/// [`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);
|
||||
// `exe` is normally an opaque `/proc/self/fd/<n>` — keep `ps` output meaningful.
|
||||
cmd.arg0("punktfunk-host");
|
||||
cmd.arg("zerocopy-worker").arg("--fd").arg("3");
|
||||
let raw = worker_end.as_raw_fd();
|
||||
let parent = std::process::id() as libc::pid_t;
|
||||
// SAFETY: `pre_exec` runs between fork and exec, so only async-signal-safe calls are
|
||||
// allowed — `prctl`, `getppid`, `dup2` and `fcntl` all are, and the closure captures only
|
||||
// `Copy` ints (no allocation, no locks; the error paths use `from_raw_os_error`, which
|
||||
// does not allocate). PR_SET_PDEATHSIG makes the kernel SIGKILL the worker when the host
|
||||
// dies — without it a crashed host left the worker holding its CUcontext + BufferPool
|
||||
// (order hundreds of MB of VRAM) indefinitely. The `getppid` check closes the standard
|
||||
// race: if the host died between fork and the prctl, the signal is never delivered, so
|
||||
// refuse to exec instead. `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 {
|
||||
cmd.pre_exec(move || {
|
||||
if libc::prctl(libc::PR_SET_PDEATHSIG, libc::SIGKILL) != 0 {
|
||||
return Err(io::Error::last_os_error());
|
||||
}
|
||||
if libc::getppid() != parent {
|
||||
return Err(io::Error::from_raw_os_error(libc::ESRCH));
|
||||
}
|
||||
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
|
||||
// `exe` is normally an opaque `/proc/self/fd/<n>` — the argv[0] keeps `ps` meaningful.
|
||||
let (host_end, child) =
|
||||
ipc::spawn_worker(exe, "punktfunk-host", &["zerocopy-worker", "--fd", "3"])
|
||||
.context("spawn zerocopy-worker")?;
|
||||
Self::from_socket(host_end, Some(child))
|
||||
}
|
||||
|
||||
@@ -266,11 +131,11 @@ impl RemoteImporter {
|
||||
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))?;
|
||||
ipc::set_recv_timeout(importer.shared.sock.as_fd(), Some(HANDSHAKE_TIMEOUT))?;
|
||||
let ready = ipc::recv::<Reply>(importer.shared.sock.as_fd(), &mut importer.rbuf);
|
||||
ipc::set_recv_timeout(importer.shared.sock.as_fd(), Some(REPLY_TIMEOUT))?;
|
||||
match ready {
|
||||
Ok((Reply::Ready { version }, _)) if version == proto::PROTO_VERSION => {
|
||||
Ok((Reply::Ready { version }, _)) if version == PROTO_VERSION => {
|
||||
tracing::info!(
|
||||
pid = importer.child.as_ref().map(|c| c.id()),
|
||||
"zero-copy GPU import isolated in a worker process"
|
||||
@@ -281,7 +146,7 @@ impl RemoteImporter {
|
||||
importer.mark_dead();
|
||||
bail!(
|
||||
"zerocopy worker protocol mismatch (worker v{version}, host v{})",
|
||||
proto::PROTO_VERSION
|
||||
PROTO_VERSION
|
||||
)
|
||||
}
|
||||
Ok((Reply::InitErr { message }, _)) => {
|
||||
@@ -315,7 +180,7 @@ impl RemoteImporter {
|
||||
if self.dead() {
|
||||
return Vec::new();
|
||||
}
|
||||
if let Err(e) = proto::send(
|
||||
if let Err(e) = ipc::send(
|
||||
self.shared.sock.as_fd(),
|
||||
&Request::Modifiers { fourcc },
|
||||
None,
|
||||
@@ -324,7 +189,7 @@ impl RemoteImporter {
|
||||
self.mark_dead();
|
||||
return Vec::new();
|
||||
}
|
||||
match proto::recv::<Reply>(self.shared.sock.as_fd(), &mut self.rbuf) {
|
||||
match ipc::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");
|
||||
@@ -439,11 +304,11 @@ impl RemoteImporter {
|
||||
stride: plane.stride,
|
||||
has_fd,
|
||||
};
|
||||
if let Err(e) = proto::send(self.shared.sock.as_fd(), &req, pass) {
|
||||
if let Err(e) = ipc::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) {
|
||||
let reply = match ipc::recv::<Reply>(self.shared.sock.as_fd(), &mut self.rbuf) {
|
||||
Ok((reply, _)) => reply,
|
||||
Err(e) => {
|
||||
self.mark_dead();
|
||||
@@ -503,7 +368,7 @@ impl RemoteImporter {
|
||||
// captured `shared` Arc is what keeps the mapping + socket alive until
|
||||
// the last frame drops. A retired mapping (its generation renegotiated
|
||||
// away) closes here with its last reference.
|
||||
let _ = proto::send(shared.sock.as_fd(), &Request::Release { id }, None);
|
||||
let _ = ipc::send(shared.sock.as_fd(), &Request::Release { id }, None);
|
||||
let mut g = shared.mappings.lock().unwrap();
|
||||
if let Some(entry) = g.get_mut(&id) {
|
||||
entry.refs = entry.refs.saturating_sub(1);
|
||||
@@ -542,7 +407,7 @@ impl RemoteImporter {
|
||||
});
|
||||
}
|
||||
if !self.dead() {
|
||||
if let Err(e) = proto::send(self.shared.sock.as_fd(), &Request::ClearCache, None) {
|
||||
if let Err(e) = ipc::send(self.shared.sock.as_fd(), &Request::ClearCache, None) {
|
||||
tracing::warn!(error = %e, "zerocopy worker ClearCache failed");
|
||||
self.mark_dead();
|
||||
}
|
||||
@@ -557,10 +422,10 @@ impl Drop for RemoteImporter {
|
||||
// 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, Instant::now()));
|
||||
ipc::park_child(child);
|
||||
}
|
||||
}
|
||||
sweep_reaper();
|
||||
ipc::sweep_reaper();
|
||||
}
|
||||
}
|
||||
|
||||
@@ -619,11 +484,12 @@ fn open_mapping(desc: &BufferDesc) -> Result<Mapping> {
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::*;
|
||||
use std::os::fd::AsRawFd;
|
||||
use std::thread;
|
||||
|
||||
fn handshake_server(reply: Reply) -> OwnedFd {
|
||||
let (host, worker) = proto::socketpair_seqpacket().unwrap();
|
||||
proto::send(worker.as_fd(), &reply, None).unwrap();
|
||||
let (host, worker) = ipc::socketpair_seqpacket().unwrap();
|
||||
ipc::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.
|
||||
@@ -634,7 +500,7 @@ mod tests {
|
||||
#[test]
|
||||
fn handshake_ready_and_version_gate() {
|
||||
let host = handshake_server(Reply::Ready {
|
||||
version: proto::PROTO_VERSION,
|
||||
version: PROTO_VERSION,
|
||||
});
|
||||
let imp = RemoteImporter::from_socket(host, None).unwrap();
|
||||
assert!(!imp.dead());
|
||||
@@ -656,7 +522,7 @@ mod tests {
|
||||
|
||||
#[test]
|
||||
fn handshake_eof_is_an_error() {
|
||||
let (host, worker) = proto::socketpair_seqpacket().unwrap();
|
||||
let (host, worker) = ipc::socketpair_seqpacket().unwrap();
|
||||
drop(worker);
|
||||
assert!(RemoteImporter::from_socket(host, None).is_err());
|
||||
}
|
||||
@@ -683,36 +549,6 @@ mod tests {
|
||||
assert!(format!("{err:#}").contains("handshake"), "{err:#}");
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn pinned_fd_exec_survives_on_disk_replacement() {
|
||||
// The 2026-07-10 canary regression: a package upgrade replaced the installed binary and
|
||||
// every worker spawn ENOENT'd (`current_exe()` readlinked to "<path> (deleted)"). The
|
||||
// pinned-fd mechanism must keep exec'ing the original image after the file is gone: pin
|
||||
// a copy of /bin/sh, delete it, then run it through the fd path.
|
||||
let copy = std::env::temp_dir().join(format!("pf-zerocopy-exe-pin-{}", std::process::id()));
|
||||
std::fs::copy("/bin/sh", ©).unwrap();
|
||||
let pinned = File::open(©).unwrap();
|
||||
std::fs::remove_file(©).unwrap();
|
||||
// Retry ETXTBSY: `fs::copy`'s write fd leaks into other tests' concurrently-forked
|
||||
// children until their execs clear it (CLOEXEC applies only at exec), and exec'ing a
|
||||
// file someone holds open for writing is refused. A harness artifact of copy-then-exec,
|
||||
// not the mechanism under test — production pins a read-only fd on a binary nobody
|
||||
// write-opens.
|
||||
let status = loop {
|
||||
match Command::new(fd_exec_path(pinned.as_fd()))
|
||||
.arg("-c")
|
||||
.arg("exit 42")
|
||||
.status()
|
||||
{
|
||||
Err(e) if e.raw_os_error() == Some(libc::ETXTBSY) => {
|
||||
std::thread::sleep(Duration::from_millis(10))
|
||||
}
|
||||
other => break other.expect("exec via /proc/self/fd of a deleted file"),
|
||||
}
|
||||
};
|
||||
assert_eq!(status.code(), Some(42));
|
||||
}
|
||||
|
||||
/// A request as the scripted peer saw it, paired with the identity (`st_ino`) of the
|
||||
/// descriptor that actually arrived via SCM_RIGHTS — the `has_fd` boolean in the JSON body is
|
||||
/// a *claim*; the received fd is the mechanism the whole worker design rests on, so tests
|
||||
@@ -723,11 +559,11 @@ mod tests {
|
||||
fn scripted_server(
|
||||
replies: Vec<Reply>,
|
||||
) -> (RemoteImporter, thread::JoinHandle<Vec<SeenRequest>>) {
|
||||
let (host, worker) = proto::socketpair_seqpacket().unwrap();
|
||||
proto::send(
|
||||
let (host, worker) = ipc::socketpair_seqpacket().unwrap();
|
||||
ipc::send(
|
||||
worker.as_fd(),
|
||||
&Reply::Ready {
|
||||
version: proto::PROTO_VERSION,
|
||||
version: PROTO_VERSION,
|
||||
},
|
||||
None,
|
||||
)
|
||||
@@ -736,7 +572,7 @@ mod tests {
|
||||
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) {
|
||||
while let Ok((req, fd)) = ipc::recv::<Request>(worker.as_fd(), &mut buf) {
|
||||
let needs_reply = matches!(req, Request::Modifiers { .. } | Request::Import { .. });
|
||||
let ino = fd
|
||||
.as_ref()
|
||||
@@ -744,7 +580,7 @@ mod tests {
|
||||
seen.push((req, ino));
|
||||
if needs_reply {
|
||||
match replies.next() {
|
||||
Some(r) => proto::send(worker.as_fd(), &r, None).unwrap(),
|
||||
Some(r) => ipc::send(worker.as_fd(), &r, None).unwrap(),
|
||||
None => break, // close → client sees a dead worker
|
||||
}
|
||||
}
|
||||
|
||||
@@ -0,0 +1,680 @@
|
||||
//! Worker-IPC rails, shared by every punktfunk worker subprocess (design:
|
||||
//! `design/zerocopy-worker-isolation.md`, generalized in
|
||||
//! `design/gpu-priority-capability-worker-implementation-plan.md` §2/WP0). Two halves, both
|
||||
//! deliberately free of any worker's *vocabulary* — the message enums live with their worker and
|
||||
//! stay independently versioned:
|
||||
//!
|
||||
//! - **Framing.** A `SOCK_SEQPACKET` unix socketpair — reliable, ordered, message-framed (one
|
||||
//! `sendmsg` = one message) — carrying serde bodies, with descriptors riding as `SCM_RIGHTS`
|
||||
//! control data. Zero-length messages are reserved: `recvmsg` returning 0 on a SEQPACKET socket
|
||||
//! is EOF (the peer died/closed), and a serialized message is never empty, so the two can't be
|
||||
//! confused.
|
||||
//! - **Process rails.** Spawn a worker on a *pinned* executable with its socket end on fd 3, kill
|
||||
//! the host's children with it (`PR_SET_PDEATHSIG`), and reap them without ever blocking a
|
||||
//! caller behind a process wedged in a driver ioctl.
|
||||
//!
|
||||
//! The zerocopy worker execs this process's own image ([`self_exe`]); the encode worker
|
||||
//! (`design/gpu-priority-capability-worker.md`) is a **separate file** — it must never share an
|
||||
//! inode with `punktfunk-host`, because a shared inode shares the file capability — so it passes
|
||||
//! its own resolved path to [`spawn_worker`] instead.
|
||||
|
||||
// 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::Serialize;
|
||||
use std::fs::File;
|
||||
use std::io;
|
||||
use std::os::fd::{AsRawFd, BorrowedFd, FromRawFd, OwnedFd, RawFd};
|
||||
use std::os::unix::process::CommandExt;
|
||||
use std::path::{Path, PathBuf};
|
||||
use std::process::{Child, Command};
|
||||
use std::sync::{Mutex, OnceLock};
|
||||
use std::time::{Duration, Instant};
|
||||
|
||||
/// 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;
|
||||
|
||||
/// Descriptors one message may carry. Four, because a multi-planar dmabuf can hold one fd per
|
||||
/// plane; the single-fd case ([`send`]/[`recv`]) is the hot path and stays allocation-free.
|
||||
pub const MAX_FDS: usize = 4;
|
||||
|
||||
/// Backing store for the `SCM_RIGHTS` control data, `u64` so it carries the 8-byte alignment
|
||||
/// `cmsghdr` requires. `MAX_FDS` fds need `CMSG_SPACE(4 * 4) = 32` bytes on 64-bit Linux (a
|
||||
/// 16-byte `cmsghdr` + 16 bytes of fds); 64 is double that, so the slack absorbs any platform
|
||||
/// whose header is larger — the `cmsg_store_is_large_enough` test asserts it for real.
|
||||
type CmsgStore = [u64; 8];
|
||||
|
||||
/// Control bytes the kernel may use for `n` descriptors. (`CMSG_SPACE` is pure size arithmetic —
|
||||
/// the `unsafe` is libc's signature, not a contract.)
|
||||
fn cmsg_space(n: usize) -> usize {
|
||||
// SAFETY: `CMSG_SPACE` performs alignment arithmetic on its argument and touches no memory.
|
||||
unsafe { libc::CMSG_SPACE((n * std::mem::size_of::<RawFd>()) as u32) as 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 calling 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 — the
|
||||
/// single-descriptor fast path over [`send_fds`] (a borrowed one-element slice; no fd list is
|
||||
/// built).
|
||||
pub fn send<T: Serialize>(
|
||||
sock: BorrowedFd,
|
||||
msg: &T,
|
||||
pass_fd: Option<BorrowedFd>,
|
||||
) -> io::Result<()> {
|
||||
match pass_fd {
|
||||
Some(fd) => send_fds(sock, msg, &[fd]),
|
||||
None => send_fds(sock, msg, &[]),
|
||||
}
|
||||
}
|
||||
|
||||
/// Send one message plus up to [`MAX_FDS`] descriptors as a single SEQPACKET datagram. Atomic per
|
||||
/// message, so concurrent senders on the same socket (e.g. the capture thread's imports and the
|
||||
/// encode thread's releases) need no lock. `MSG_NOSIGNAL` turns a dead peer into `EPIPE` instead
|
||||
/// of `SIGPIPE`. More than [`MAX_FDS`] is a caller bug, refused like an over-long body rather than
|
||||
/// panicking.
|
||||
pub fn send_fds<T: Serialize>(sock: BorrowedFd, msg: &T, fds: &[BorrowedFd]) -> io::Result<()> {
|
||||
if fds.len() > MAX_FDS {
|
||||
return Err(io::Error::new(
|
||||
io::ErrorKind::InvalidInput,
|
||||
format!(
|
||||
"worker ipc: {} fds in one message (max {MAX_FDS})",
|
||||
fds.len()
|
||||
),
|
||||
));
|
||||
}
|
||||
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,
|
||||
"worker ipc message too large",
|
||||
));
|
||||
}
|
||||
let mut iov = libc::iovec {
|
||||
iov_base: body.as_ptr() as *mut libc::c_void,
|
||||
iov_len: body.len(),
|
||||
};
|
||||
let mut cmsg_store: CmsgStore = [0; 8];
|
||||
// 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 !fds.is_empty() {
|
||||
let bytes = (fds.len() * std::mem::size_of::<RawFd>()) as u32;
|
||||
debug_assert!(cmsg_space(fds.len()) <= std::mem::size_of_val(&cmsg_store));
|
||||
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 — the store holds
|
||||
// `CMSG_SPACE(MAX_FDS * 4)` and `fds.len() <= MAX_FDS` was checked above — for the header
|
||||
// fields plus one 4-byte fd per element written through `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(bytes) 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(bytes) as _;
|
||||
let data = libc::CMSG_DATA(c) as *mut RawFd;
|
||||
for (i, fd) in fds.iter().enumerate() {
|
||||
std::ptr::write_unaligned(data.add(i), 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 fds are passed — at `cmsg_store` (ditto). `sendmsg`
|
||||
// only reads these buffers. The kernel dups the fds into the message; our `BorrowedFd`s stay
|
||||
// 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) — the single-descriptor fast path over
|
||||
/// [`recv_fds`]. Any further descriptors the peer attached are dropped, i.e. closed, so a
|
||||
/// protocol mix-up cannot leak fds into a caller that only knows about one.
|
||||
pub fn recv<T: DeserializeOwned>(
|
||||
sock: BorrowedFd,
|
||||
buf: &mut Vec<u8>,
|
||||
) -> io::Result<(T, Option<OwnedFd>)> {
|
||||
let (msg, fds) = recv_fds(sock, buf)?;
|
||||
Ok((msg, fds.into_iter().next()))
|
||||
}
|
||||
|
||||
/// Receive one message plus its `SCM_RIGHTS` descriptors (up to [`MAX_FDS`], in the order the
|
||||
/// sender listed them). `buf` is a caller-owned scratch buffer (grown to [`MAX_MSG`] once, then
|
||||
/// reused message to message); the returned `Vec` does not allocate when no fd arrived, which is
|
||||
/// the steady state under an fd-identity cache. Errors: `UnexpectedEof` = the peer is gone;
|
||||
/// `WouldBlock` = the [`set_recv_timeout`] expired; `InvalidData` = a truncated body or more
|
||||
/// descriptors than [`MAX_FDS`] (the kernel drops the excess and flags `MSG_CTRUNC`).
|
||||
pub fn recv_fds<T: DeserializeOwned>(
|
||||
sock: BorrowedFd,
|
||||
buf: &mut Vec<u8>,
|
||||
) -> io::Result<(T, Vec<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: CmsgStore = [0; 8];
|
||||
// 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;
|
||||
// Exactly MAX_FDS worth of control space (not the store's full size): a peer that attaches
|
||||
// more gets its excess dropped by the kernel and the message flagged `MSG_CTRUNC`, which is
|
||||
// checked below — the cap is enforced by the kernel rather than trusted from the peer.
|
||||
debug_assert!(cmsg_space(MAX_FDS) <= std::mem::size_of_val(&cmsg_store));
|
||||
mhdr.msg_controllen = cmsg_space(MAX_FDS) 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, and at least that large — asserted above). `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,
|
||||
"worker ipc peer closed",
|
||||
));
|
||||
}
|
||||
// Collect the passed fds (if any) BEFORE any early return below, so they can't leak — an
|
||||
// error path drops the `Vec`, which closes every one of them.
|
||||
let mut got: Vec<OwnedFd> = Vec::new();
|
||||
// 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 `RawFd`s, `cmsg_len -
|
||||
// CMSG_LEN(0)` bytes of them, all inside that same complete cmsg, so every `read_unaligned`
|
||||
// at index < that count is in bounds. The kernel gave us ownership of each fd (they are fresh
|
||||
// descriptors in our table), so each `OwnedFd::from_raw_fd` takes sole ownership of a distinct
|
||||
// descriptor — no alias, no double-close, and nothing dropped on the floor even with multiple
|
||||
// cmsgs or multiple fds per cmsg.
|
||||
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 payload = ((*c).cmsg_len as usize).saturating_sub(libc::CMSG_LEN(0) as usize);
|
||||
let data = libc::CMSG_DATA(c) as *const RawFd;
|
||||
for i in 0..payload / std::mem::size_of::<RawFd>() {
|
||||
let fd = std::ptr::read_unaligned(data.add(i));
|
||||
if fd >= 0 {
|
||||
got.push(OwnedFd::from_raw_fd(fd));
|
||||
}
|
||||
}
|
||||
}
|
||||
c = libc::CMSG_NXTHDR(&mhdr, c);
|
||||
}
|
||||
}
|
||||
if mhdr.msg_flags & libc::MSG_CTRUNC != 0 {
|
||||
return Err(io::Error::new(
|
||||
io::ErrorKind::InvalidData,
|
||||
format!("worker ipc message carried more than {MAX_FDS} descriptors"),
|
||||
));
|
||||
}
|
||||
if mhdr.msg_flags & libc::MSG_TRUNC != 0 {
|
||||
return Err(io::Error::new(
|
||||
io::ErrorKind::InvalidData,
|
||||
"worker ipc 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))
|
||||
}
|
||||
|
||||
/// An executable pinned by an open fd, exec'able through [`PinnedExe::exec_path`]. The fd names
|
||||
/// the file's *inode*, not its path, so it resolves even after the binary was replaced or deleted
|
||||
/// — and exec'ing it then still runs byte-for-byte the build that was pinned. `current_exe()`
|
||||
/// instead readlinks to a path: after a package upgrade under a running host that path is
|
||||
/// `"<path> (deleted)"` and spawning it fails ENOENT — every capture then silently fell back to the
|
||||
/// CPU copy (the 2026-07-10 canary regression) — and even while the path exists it may hold a
|
||||
/// newer build whose worker protocol mismatches this process.
|
||||
pub struct PinnedExe(File);
|
||||
|
||||
impl PinnedExe {
|
||||
/// Pin the executable at `path`.
|
||||
pub fn open(path: &Path) -> io::Result<PinnedExe> {
|
||||
let f = File::open(path)?;
|
||||
if f.as_raw_fd() != 3 {
|
||||
return Ok(PinnedExe(f));
|
||||
}
|
||||
// Fd 3 is the slot the spawn hands the worker its socket on (the `dup2` in
|
||||
// [`spawn_worker`]) — pinned there, the child would clobber it before exec resolves
|
||||
// `/proc/self/fd/3`. Re-number: 3 stays occupied by `f` during the clone, so the
|
||||
// duplicate cannot land on it.
|
||||
let clone = f.try_clone().map_err(|e| {
|
||||
io::Error::new(
|
||||
e.kind(),
|
||||
format!("re-numbering the pinned exe fd off fd 3 failed: {e}"),
|
||||
)
|
||||
})?;
|
||||
Ok(PinnedExe(clone))
|
||||
}
|
||||
|
||||
/// `/proc/self/fd/<n>` — an exec'able path to the pinned inode. The kernel resolves it at exec
|
||||
/// time inside the forked child, whose fd table is a copy of ours (close-on-exec applies only
|
||||
/// once the exec succeeds), so it names the pinned inode no matter what sits at the file's
|
||||
/// original path by then.
|
||||
pub fn exec_path(&self) -> PathBuf {
|
||||
PathBuf::from(format!("/proc/self/fd/{}", self.0.as_raw_fd()))
|
||||
}
|
||||
}
|
||||
|
||||
/// This process's own executable image, pinned once (lazily) via the `/proc/self/exe` magic link
|
||||
/// — see [`PinnedExe`] for why the fd and not the path. `None` when it could not be opened, in
|
||||
/// which case callers fall back to `current_exe()` and inherit that trap.
|
||||
///
|
||||
/// Only for a worker that is the same file as its host by construction (the zerocopy worker
|
||||
/// re-execs this binary). The encode worker must stay a **separate file** — it carries a file
|
||||
/// capability the host must never have — so it pins its own resolved path with [`PinnedExe::open`].
|
||||
pub fn self_exe() -> Option<&'static PinnedExe> {
|
||||
static SELF_EXE: OnceLock<Option<PinnedExe>> = OnceLock::new();
|
||||
SELF_EXE
|
||||
.get_or_init(|| match PinnedExe::open(Path::new("/proc/self/exe")) {
|
||||
Ok(p) => Some(p),
|
||||
Err(e) => {
|
||||
tracing::warn!(
|
||||
error = %e,
|
||||
"cannot pin /proc/self/exe — worker spawns use the current_exe() path, \
|
||||
which breaks if this binary is replaced on disk"
|
||||
);
|
||||
None
|
||||
}
|
||||
})
|
||||
.as_ref()
|
||||
}
|
||||
|
||||
/// Spawn a worker process on `exe` (normally a [`PinnedExe::exec_path`]) and hand it one end of a
|
||||
/// fresh SEQPACKET socketpair on **fd 3**; the host end is returned alongside the child.
|
||||
///
|
||||
/// `argv0` is what `ps` shows — worth setting, because `exe` is normally an opaque
|
||||
/// `/proc/self/fd/<n>`. `args` follow it.
|
||||
pub fn spawn_worker(exe: &Path, argv0: &str, args: &[&str]) -> io::Result<(OwnedFd, Child)> {
|
||||
sweep_reaper();
|
||||
let (host_end, worker_end) = socketpair_seqpacket()?;
|
||||
let mut cmd = Command::new(exe);
|
||||
cmd.arg0(argv0);
|
||||
cmd.args(args);
|
||||
let raw = worker_end.as_raw_fd();
|
||||
let parent = std::process::id() as libc::pid_t;
|
||||
// SAFETY: `pre_exec` runs between fork and exec, so only async-signal-safe calls are
|
||||
// allowed — `prctl`, `getppid`, `dup2` and `fcntl` all are, and the closure captures only
|
||||
// `Copy` ints (no allocation, no locks; the error paths use `from_raw_os_error`, which
|
||||
// does not allocate). PR_SET_PDEATHSIG makes the kernel SIGKILL the worker when the host
|
||||
// dies — without it a crashed host left the worker holding its driver state (order hundreds
|
||||
// of MB of VRAM) indefinitely. The `getppid` check closes the standard race: if the host died
|
||||
// between fork and the prctl, the signal is never delivered, so refuse to exec instead.
|
||||
// `dup2(raw, 3)` installs the socket at the fd number the worker 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 {
|
||||
cmd.pre_exec(move || {
|
||||
if libc::prctl(libc::PR_SET_PDEATHSIG, libc::SIGKILL) != 0 {
|
||||
return Err(io::Error::last_os_error());
|
||||
}
|
||||
if libc::getppid() != parent {
|
||||
return Err(io::Error::from_raw_os_error(libc::ESRCH));
|
||||
}
|
||||
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()?;
|
||||
drop(worker_end); // the child holds its own copy now
|
||||
Ok((host_end, child))
|
||||
}
|
||||
|
||||
/// Children whose worker hasn't exited yet at teardown time (a worker 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 generation.
|
||||
static REAPER: Mutex<Vec<(Child, Instant)>> = Mutex::new(Vec::new());
|
||||
|
||||
/// How long past the caller's reply timeout a parked worker may linger before it is force-killed.
|
||||
/// A worker wedged INSIDE a driver call never observes socket EOF, so `try_wait` alone would keep
|
||||
/// it (and its driver state — order hundreds of MB of VRAM) forever.
|
||||
const REAPER_KILL_DEADLINE: Duration = Duration::from_secs(20);
|
||||
|
||||
/// Hand a still-running worker to the reaper. Call from a `Drop` after the socket is (about to
|
||||
/// be) closed: the worker sees EOF and exits, and the next [`sweep_reaper`] collects it.
|
||||
pub fn park_child(child: Child) {
|
||||
REAPER.lock().unwrap().push((child, Instant::now()));
|
||||
}
|
||||
|
||||
/// Reap exited workers; force-kill the ones parked past the kill deadline (20 s).
|
||||
pub fn sweep_reaper() {
|
||||
// Partition under the lock; kill/reap OUTSIDE it. A worker wedged inside a driver ioctl sits
|
||||
// in D state and ignores SIGKILL — the old blocking `wait()` under the global mutex would
|
||||
// then park every later spawn and drop behind a process that may never die.
|
||||
let mut expired: Vec<Child> = Vec::new();
|
||||
{
|
||||
let mut list = REAPER.lock().unwrap();
|
||||
let now = Instant::now();
|
||||
let mut i = 0;
|
||||
while i < list.len() {
|
||||
if matches!(list[i].0.try_wait(), Ok(Some(_))) {
|
||||
list.swap_remove(i); // exited on its own → reaped
|
||||
} else if now.duration_since(list[i].1) > REAPER_KILL_DEADLINE {
|
||||
expired.push(list.swap_remove(i).0);
|
||||
} else {
|
||||
i += 1;
|
||||
}
|
||||
}
|
||||
}
|
||||
for mut c in expired {
|
||||
let _ = c.kill();
|
||||
// Bounded reap (~100 ms of polls): a SIGKILL'd process reaps near-instantly unless it is
|
||||
// in D state — then park it again (re-killing later is harmless) so a future sweep reaps
|
||||
// it once the driver unwedges, instead of blocking anyone here forever.
|
||||
let mut reaped = false;
|
||||
for _ in 0..10 {
|
||||
if matches!(c.try_wait(), Ok(Some(_))) {
|
||||
reaped = true;
|
||||
break;
|
||||
}
|
||||
std::thread::sleep(Duration::from_millis(10));
|
||||
}
|
||||
if !reaped {
|
||||
tracing::warn!(
|
||||
pid = c.id(),
|
||||
"worker ignored SIGKILL (likely wedged in a driver call, D state) — \
|
||||
parked for a later sweep"
|
||||
);
|
||||
park_child(c);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::*;
|
||||
use serde::Deserialize;
|
||||
use std::io::{Read, Write};
|
||||
use std::os::fd::AsFd;
|
||||
|
||||
/// A stand-in payload: the transport is generic over the serde body, and the workers' own
|
||||
/// message enums live with the workers — nothing here may depend on one.
|
||||
#[derive(Serialize, Deserialize, Debug, PartialEq)]
|
||||
struct Msg {
|
||||
tag: String,
|
||||
n: u32,
|
||||
}
|
||||
|
||||
fn msg(tag: &str) -> Msg {
|
||||
Msg {
|
||||
tag: tag.into(),
|
||||
n: 7,
|
||||
}
|
||||
}
|
||||
|
||||
/// A pipe whose read end is passed over the socket, carrying `payload` for the receiver to
|
||||
/// read back — the only way to prove the *descriptor* crossed, not just the claim that it did.
|
||||
fn marked_pipe(payload: &[u8]) -> (std::io::PipeReader, std::io::PipeWriter) {
|
||||
let (pr, mut pw) = std::io::pipe().unwrap();
|
||||
pw.write_all(payload).unwrap();
|
||||
(pr, pw)
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn cmsg_store_is_large_enough() {
|
||||
assert!(
|
||||
cmsg_space(MAX_FDS) <= std::mem::size_of::<CmsgStore>(),
|
||||
"CMSG_SPACE({MAX_FDS} fds) = {} > store {}",
|
||||
cmsg_space(MAX_FDS),
|
||||
std::mem::size_of::<CmsgStore>()
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn round_trip_no_fd() {
|
||||
let (a, b) = socketpair_seqpacket().unwrap();
|
||||
let mut buf = Vec::new();
|
||||
send(a.as_fd(), &msg("hello"), None).unwrap();
|
||||
let (got, fd) = recv::<Msg>(b.as_fd(), &mut buf).unwrap();
|
||||
assert_eq!(got, msg("hello"));
|
||||
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(), &msg("one"), Some(pr.as_fd())).unwrap();
|
||||
let (got, fd) = recv::<Msg>(b.as_fd(), &mut buf).unwrap();
|
||||
assert_eq!(got, msg("one"));
|
||||
let fd = fd.expect("fd should have been passed");
|
||||
pw.write_all(b"hello").unwrap();
|
||||
drop(pw);
|
||||
let mut file = 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 round_trip_three_fds() {
|
||||
// The multi-planar case (WP0): one message carrying one fd per plane. Each pipe is
|
||||
// pre-loaded with a distinct byte, so reading through the RECEIVED descriptors proves
|
||||
// both that all three crossed and that they arrived in the sender's order.
|
||||
let (a, b) = socketpair_seqpacket().unwrap();
|
||||
let mut buf = Vec::new();
|
||||
let planes = [marked_pipe(b"Y"), marked_pipe(b"U"), marked_pipe(b"V")];
|
||||
{
|
||||
let fds: Vec<BorrowedFd> = planes.iter().map(|(pr, _)| pr.as_fd()).collect();
|
||||
send_fds(a.as_fd(), &msg("planes"), &fds).unwrap();
|
||||
}
|
||||
let (got, fds) = recv_fds::<Msg>(b.as_fd(), &mut buf).unwrap();
|
||||
assert_eq!(got, msg("planes"));
|
||||
assert_eq!(fds.len(), 3, "all three descriptors must cross");
|
||||
// Drop the write ends so each read sees EOF after its byte.
|
||||
drop(planes);
|
||||
let read_back: Vec<String> = fds
|
||||
.into_iter()
|
||||
.map(|fd| {
|
||||
let mut s = String::new();
|
||||
File::from(fd).read_to_string(&mut s).unwrap();
|
||||
s
|
||||
})
|
||||
.collect();
|
||||
assert_eq!(read_back, vec!["Y", "U", "V"]);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn max_fds_round_trip_and_one_more_is_refused() {
|
||||
let (a, b) = socketpair_seqpacket().unwrap();
|
||||
let mut buf = Vec::new();
|
||||
let pipes: Vec<_> = (0..MAX_FDS + 1).map(|_| std::io::pipe().unwrap()).collect();
|
||||
let fds: Vec<BorrowedFd> = pipes.iter().map(|(pr, _)| pr.as_fd()).collect();
|
||||
send_fds(a.as_fd(), &msg("full"), &fds[..MAX_FDS]).unwrap();
|
||||
let (_, got) = recv_fds::<Msg>(b.as_fd(), &mut buf).unwrap();
|
||||
assert_eq!(got.len(), MAX_FDS);
|
||||
// One over the cap is refused at the sender (a caller bug), not panicked on.
|
||||
let err = send_fds(a.as_fd(), &msg("over"), &fds).unwrap_err();
|
||||
assert_eq!(err.kind(), io::ErrorKind::InvalidInput);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn recv_keeps_the_first_fd_and_closes_the_rest() {
|
||||
// A peer that attaches more descriptors than the single-fd caller expects must not leak
|
||||
// them: `recv` returns the first and drops (closes) the others.
|
||||
let (a, b) = socketpair_seqpacket().unwrap();
|
||||
let mut buf = Vec::new();
|
||||
let (first, second) = (marked_pipe(b"1"), marked_pipe(b"2"));
|
||||
send_fds(a.as_fd(), &msg("two"), &[first.0.as_fd(), second.0.as_fd()]).unwrap();
|
||||
let (_, fd) = recv::<Msg>(b.as_fd(), &mut buf).unwrap();
|
||||
let fd = fd.expect("the first fd is returned");
|
||||
drop(first);
|
||||
let mut s = String::new();
|
||||
File::from(fd).read_to_string(&mut s).unwrap();
|
||||
assert_eq!(s, "1");
|
||||
// The second pipe's receiver-side dup was closed with the returned Vec's tail, so the
|
||||
// writer sees EPIPE once the local read end goes too.
|
||||
drop(second.0);
|
||||
let mut pw = second.1;
|
||||
assert_eq!(
|
||||
pw.write_all(b"x").unwrap_err().kind(),
|
||||
io::ErrorKind::BrokenPipe
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn eof_when_peer_closes() {
|
||||
let (a, b) = socketpair_seqpacket().unwrap();
|
||||
drop(a);
|
||||
let mut buf = Vec::new();
|
||||
let err = recv::<Msg>(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(), &msg("gone"), 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::<Msg>(a.as_fd(), &mut buf).unwrap_err();
|
||||
assert!(
|
||||
matches!(
|
||||
err.kind(),
|
||||
io::ErrorKind::WouldBlock | io::ErrorKind::TimedOut
|
||||
),
|
||||
"unexpected error kind: {err:?}"
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn pinned_fd_exec_survives_on_disk_replacement() {
|
||||
// The 2026-07-10 canary regression: a package upgrade replaced the installed binary and
|
||||
// every worker spawn ENOENT'd (`current_exe()` readlinked to "<path> (deleted)"). The
|
||||
// pinned-fd mechanism must keep exec'ing the original image after the file is gone: pin
|
||||
// a copy of /bin/sh, delete it, then run it through the fd path.
|
||||
let copy = std::env::temp_dir().join(format!("pf-zerocopy-exe-pin-{}", std::process::id()));
|
||||
std::fs::copy("/bin/sh", ©).unwrap();
|
||||
let pinned = PinnedExe::open(©).unwrap();
|
||||
std::fs::remove_file(©).unwrap();
|
||||
// Retry ETXTBSY: `fs::copy`'s write fd leaks into other tests' concurrently-forked
|
||||
// children until their execs clear it (CLOEXEC applies only at exec), and exec'ing a
|
||||
// file someone holds open for writing is refused. A harness artifact of copy-then-exec,
|
||||
// not the mechanism under test — production pins a read-only fd on a binary nobody
|
||||
// write-opens.
|
||||
let status = loop {
|
||||
match Command::new(pinned.exec_path())
|
||||
.arg("-c")
|
||||
.arg("exit 42")
|
||||
.status()
|
||||
{
|
||||
Err(e) if e.raw_os_error() == Some(libc::ETXTBSY) => {
|
||||
std::thread::sleep(Duration::from_millis(10))
|
||||
}
|
||||
other => break other.expect("exec via /proc/self/fd of a deleted file"),
|
||||
}
|
||||
};
|
||||
assert_eq!(status.code(), Some(42));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn spawn_worker_hands_the_socket_on_fd_3() {
|
||||
// `sh` echoes back through fd 3 — the inheritance slot every worker reads its socket
|
||||
// from. Receiving it here proves the dup2 landed and CLOEXEC was cleared on the copy.
|
||||
let (host, mut child) = spawn_worker(
|
||||
Path::new("/bin/sh"),
|
||||
"pf-test-worker",
|
||||
&["-c", r#"printf '"pong"' >&3"#],
|
||||
)
|
||||
.unwrap();
|
||||
let mut buf = Vec::new();
|
||||
let (got, fds) = recv_fds::<String>(host.as_fd(), &mut buf).unwrap();
|
||||
assert_eq!(got, "pong");
|
||||
assert!(fds.is_empty());
|
||||
child.wait().unwrap();
|
||||
}
|
||||
}
|
||||
@@ -14,6 +14,11 @@
|
||||
pub mod client;
|
||||
pub mod cuda;
|
||||
pub mod egl;
|
||||
// Worker-subprocess IPC rails (SEQPACKET framing ± `SCM_RIGHTS`, pinned-exe spawn, reaping),
|
||||
// generic over the message body so every punktfunk worker shares them — the zerocopy one here,
|
||||
// the capability-carrying encode worker in `pf-encode`. Vocabulary stays per-worker (`proto` is
|
||||
// only this one's).
|
||||
pub mod ipc;
|
||||
pub mod proto;
|
||||
pub mod vkslot;
|
||||
pub mod vulkan;
|
||||
|
||||
@@ -1,32 +1,20 @@
|
||||
//! Wire protocol between the PipeWire capture thread and the isolated zero-copy GPU-import
|
||||
//! Wire *vocabulary* 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`]).
|
||||
//! `design/zerocopy-worker-isolation.md`) — the message types and this protocol's version, and
|
||||
//! nothing else. The transport they ride on ([`super::ipc`]: SEQPACKET framing, `SCM_RIGHTS`,
|
||||
//! spawn/reap) is shared with the other workers and is deliberately generic over the body type;
|
||||
//! each worker's vocabulary stays its own and versions independently.
|
||||
//!
|
||||
//! 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.
|
||||
//! 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`]).
|
||||
|
||||
// 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 `EglImporter` entry points.
|
||||
#[derive(Serialize, Deserialize, Debug, Clone, Copy, PartialEq, Eq)]
|
||||
pub enum ImportKind {
|
||||
@@ -118,197 +106,17 @@ pub struct BufferDesc {
|
||||
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 crate::imp::ipc;
|
||||
use std::os::fd::AsFd;
|
||||
|
||||
/// The vocabulary survives the wire in both directions. (The framing itself — fds, EOF,
|
||||
/// timeouts, the descriptor cap — is exercised in [`ipc`]; this only pins the message types.)
|
||||
#[test]
|
||||
fn round_trip_no_fd() {
|
||||
let (a, b) = socketpair_seqpacket().unwrap();
|
||||
fn round_trip_both_directions() {
|
||||
let (a, b) = ipc::socketpair_seqpacket().unwrap();
|
||||
let mut buf = Vec::new();
|
||||
let req = Request::Import {
|
||||
key: 0xdead_beef_u64,
|
||||
@@ -321,8 +129,8 @@ mod tests {
|
||||
stride: 5120 * 4,
|
||||
has_fd: false,
|
||||
};
|
||||
send(a.as_fd(), &req, None).unwrap();
|
||||
let (got, fd) = recv::<Request>(b.as_fd(), &mut buf).unwrap();
|
||||
ipc::send(a.as_fd(), &req, None).unwrap();
|
||||
let (got, fd) = ipc::recv::<Request>(b.as_fd(), &mut buf).unwrap();
|
||||
assert_eq!(got, req);
|
||||
assert!(fd.is_none());
|
||||
|
||||
@@ -336,64 +144,9 @@ mod tests {
|
||||
uv: Some((vec![2u8; 64], 5632)),
|
||||
}),
|
||||
};
|
||||
send(b.as_fd(), &reply, None).unwrap();
|
||||
let (got, fd) = recv::<Reply>(a.as_fd(), &mut buf).unwrap();
|
||||
ipc::send(b.as_fd(), &reply, None).unwrap();
|
||||
let (got, fd) = ipc::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:?}"
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
@@ -14,7 +14,8 @@
|
||||
|
||||
use super::cuda::{self, CUdeviceptr, DeviceBuffer};
|
||||
use super::egl::{DmabufPlane, EglImporter};
|
||||
use super::proto::{self, BufferDesc, ImportKind, Reply, Request};
|
||||
use super::ipc;
|
||||
use super::proto::{BufferDesc, ImportKind, Reply, Request, PROTO_VERSION};
|
||||
use anyhow::{bail, Context, Result};
|
||||
use std::collections::{HashMap, VecDeque};
|
||||
use std::io;
|
||||
@@ -27,7 +28,7 @@ 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
|
||||
// The host execs this worker through its pinned exe fd (`ipc::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
|
||||
@@ -72,7 +73,7 @@ fn run(sock: OwnedFd) -> Result<()> {
|
||||
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(
|
||||
let _ = ipc::send(
|
||||
sock.as_fd(),
|
||||
&Reply::InitErr {
|
||||
message: format!("{e:#}"),
|
||||
@@ -82,10 +83,10 @@ fn run(sock: OwnedFd) -> Result<()> {
|
||||
return Ok(());
|
||||
}
|
||||
};
|
||||
proto::send(
|
||||
ipc::send(
|
||||
sock.as_fd(),
|
||||
&Reply::Ready {
|
||||
version: proto::PROTO_VERSION,
|
||||
version: PROTO_VERSION,
|
||||
},
|
||||
None,
|
||||
)
|
||||
@@ -124,7 +125,7 @@ pub(crate) struct ImportReq {
|
||||
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) {
|
||||
let (req, fd) = match ipc::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"),
|
||||
@@ -173,7 +174,7 @@ pub(crate) fn serve(sock: &OwnedFd, backend: &mut dyn ImportBackend) -> Result<(
|
||||
|
||||
/// 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) {
|
||||
match ipc::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"),
|
||||
@@ -434,7 +435,7 @@ mod tests {
|
||||
mpsc::Receiver<String>,
|
||||
std::thread::JoinHandle<Result<()>>,
|
||||
) {
|
||||
let (host, worker) = proto::socketpair_seqpacket().unwrap();
|
||||
let (host, worker) = ipc::socketpair_seqpacket().unwrap();
|
||||
let (tx, rx) = mpsc::channel();
|
||||
let join = std::thread::spawn(move || {
|
||||
let mut backend = MockBackend { calls: tx, next: 0 };
|
||||
@@ -462,8 +463,8 @@ mod tests {
|
||||
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();
|
||||
ipc::send(host.as_fd(), &Request::Modifiers { fourcc: 42 }, None).unwrap();
|
||||
let (reply, _) = ipc::recv::<Reply>(host.as_fd(), &mut buf).unwrap();
|
||||
assert_eq!(
|
||||
reply,
|
||||
Reply::Modifiers {
|
||||
@@ -472,8 +473,8 @@ mod tests {
|
||||
);
|
||||
|
||||
// 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();
|
||||
ipc::send(host.as_fd(), &import_req(1, false), None).unwrap();
|
||||
let (reply, _) = ipc::recv::<Reply>(host.as_fd(), &mut buf).unwrap();
|
||||
match reply {
|
||||
Reply::Frame {
|
||||
id: 0,
|
||||
@@ -481,8 +482,8 @@ mod tests {
|
||||
} => {}
|
||||
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();
|
||||
ipc::send(host.as_fd(), &import_req(1, false), None).unwrap();
|
||||
let (reply, _) = ipc::recv::<Reply>(host.as_fd(), &mut buf).unwrap();
|
||||
assert_eq!(reply, Reply::Frame { id: 1, desc: None });
|
||||
|
||||
// The descriptor itself must cross the socket: an import WITH an fd rides SCM_RIGHTS and
|
||||
@@ -490,26 +491,26 @@ mod tests {
|
||||
// received fd (e.g. `backend.import(&req, None)`) would only be caught here.
|
||||
let (pr, _pw) = std::io::pipe().unwrap();
|
||||
let sent_ino = fd_ino(pr.as_fd().as_raw_fd());
|
||||
proto::send(host.as_fd(), &import_req(3, true), Some(pr.as_fd())).unwrap();
|
||||
let (reply, _) = proto::recv::<Reply>(host.as_fd(), &mut buf).unwrap();
|
||||
ipc::send(host.as_fd(), &import_req(3, true), Some(pr.as_fd())).unwrap();
|
||||
let (reply, _) = ipc::recv::<Reply>(host.as_fd(), &mut buf).unwrap();
|
||||
assert_eq!(reply, Reply::Frame { id: 2, 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();
|
||||
ipc::send(host.as_fd(), &import_req(0xfeed, false), None).unwrap();
|
||||
let (reply, _) = ipc::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();
|
||||
ipc::send(host.as_fd(), &import_req(0xbad, false), None).unwrap();
|
||||
let (reply, _) = ipc::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();
|
||||
ipc::send(host.as_fd(), &Request::Release { id: 0 }, None).unwrap();
|
||||
ipc::send(host.as_fd(), &Request::ClearCache, None).unwrap();
|
||||
|
||||
// Closing the host end terminates serve() cleanly.
|
||||
drop(host);
|
||||
|
||||
Reference in New Issue
Block a user