From dcfba07803c6662c09572121400a7ac9a39150bb Mon Sep 17 00:00:00 2001 From: enricobuehler Date: Sun, 9 Aug 2026 12:49:53 +0200 Subject: [PATCH] refactor(pf-zerocopy): split the worker rails out of the zerocopy vocabulary so a second worker can reuse them MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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. --- crates/pf-zerocopy/src/imp/client.rs | 258 ++-------- crates/pf-zerocopy/src/imp/ipc.rs | 680 +++++++++++++++++++++++++++ crates/pf-zerocopy/src/imp/mod.rs | 5 + crates/pf-zerocopy/src/imp/proto.rs | 279 +---------- crates/pf-zerocopy/src/imp/worker.rs | 45 +- 5 files changed, 771 insertions(+), 496 deletions(-) create mode 100644 crates/pf-zerocopy/src/imp/ipc.rs diff --git a/crates/pf-zerocopy/src/imp/client.rs b/crates/pf-zerocopy/src/imp/client.rs index ea11aba6..563cb84d 100644 --- a/crates/pf-zerocopy/src/imp/client.rs +++ b/crates/pf-zerocopy/src/imp/client.rs @@ -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> = 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 = 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 -/// " (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> = OnceLock::new(); - -fn self_exe() -> Option> { - 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/` — 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 { - 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 { - 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/` — 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/` — 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::(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::(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::(self.shared.sock.as_fd(), &mut self.rbuf) { + match ipc::recv::(self.shared.sock.as_fd(), &mut self.rbuf) { Ok((Reply::Modifiers { modifiers }, _)) => modifiers, Ok((other, _)) => { tracing::warn!(?other, "unexpected zerocopy worker reply to Modifiers"); @@ -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::(self.shared.sock.as_fd(), &mut self.rbuf) { + let reply = match ipc::recv::(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 { #[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 " (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, ) -> (RemoteImporter, thread::JoinHandle>) { - 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::(worker.as_fd(), &mut buf) { + while let Ok((req, fd)) = ipc::recv::(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 } } diff --git a/crates/pf-zerocopy/src/imp/ipc.rs b/crates/pf-zerocopy/src/imp/ipc.rs new file mode 100644 index 00000000..89fc8de4 --- /dev/null +++ b/crates/pf-zerocopy/src/imp/ipc.rs @@ -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::()) 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) -> io::Result<()> { + let tv = match timeout { + Some(d) => libc::timeval { + tv_sec: d.as_secs() as libc::time_t, + tv_usec: d.subsec_micros() as libc::suseconds_t, + }, + None => libc::timeval { + tv_sec: 0, + tv_usec: 0, + }, + }; + // SAFETY: `setsockopt(SO_RCVTIMEO)` reads `size_of::()` bytes from `&tv`, a live + // stack `timeval` that outlives this synchronous call; `sock` is the caller's live socket fd. + // Nothing is retained or written through Rust pointers. + let r = unsafe { + libc::setsockopt( + sock.as_raw_fd(), + libc::SOL_SOCKET, + libc::SO_RCVTIMEO, + &tv as *const libc::timeval as *const libc::c_void, + std::mem::size_of::() as libc::socklen_t, + ) + }; + if r != 0 { + return Err(io::Error::last_os_error()); + } + Ok(()) +} + +/// Send one message (+ optionally one fd as `SCM_RIGHTS`) as a single SEQPACKET datagram — the +/// single-descriptor fast path over [`send_fds`] (a borrowed one-element slice; no fd list is +/// built). +pub fn send( + sock: BorrowedFd, + msg: &T, + pass_fd: Option, +) -> 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(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::()) 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( + sock: BorrowedFd, + buf: &mut Vec, +) -> io::Result<(T, Option)> { + 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( + sock: BorrowedFd, + buf: &mut Vec, +) -> io::Result<(T, Vec)> { + 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 = 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::() { + 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 +/// `" (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 { + 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/` — 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> = 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/`. `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> = 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 = 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::(), + "CMSG_SPACE({MAX_FDS} fds) = {} > store {}", + cmsg_space(MAX_FDS), + std::mem::size_of::() + ); + } + + #[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::(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::(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 = planes.iter().map(|(pr, _)| pr.as_fd()).collect(); + send_fds(a.as_fd(), &msg("planes"), &fds).unwrap(); + } + let (got, fds) = recv_fds::(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 = 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 = pipes.iter().map(|(pr, _)| pr.as_fd()).collect(); + send_fds(a.as_fd(), &msg("full"), &fds[..MAX_FDS]).unwrap(); + let (_, got) = recv_fds::(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::(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::(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::(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 " (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::(host.as_fd(), &mut buf).unwrap(); + assert_eq!(got, "pong"); + assert!(fds.is_empty()); + child.wait().unwrap(); + } +} diff --git a/crates/pf-zerocopy/src/imp/mod.rs b/crates/pf-zerocopy/src/imp/mod.rs index f6d38174..b3e5a1c3 100644 --- a/crates/pf-zerocopy/src/imp/mod.rs +++ b/crates/pf-zerocopy/src/imp/mod.rs @@ -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; diff --git a/crates/pf-zerocopy/src/imp/proto.rs b/crates/pf-zerocopy/src/imp/proto.rs index ccc774db..9812621e 100644 --- a/crates/pf-zerocopy/src/imp/proto.rs +++ b/crates/pf-zerocopy/src/imp/proto.rs @@ -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, usize)>, } -/// A CLOEXEC `SOCK_SEQPACKET` socketpair — `(host_end, worker_end)`. -pub fn socketpair_seqpacket() -> io::Result<(OwnedFd, OwnedFd)> { - let mut fds = [0i32; 2]; - // SAFETY: `socketpair` writes two fds into `fds`, a live 2-element stack array matching the - // API contract; it reads no other Rust memory. The result is checked before the fds are used, - // and each returned fd is fresh (owned by no other wrapper), so the two `OwnedFd::from_raw_fd` - // each take sole ownership of a distinct, valid descriptor — no alias, no double-close. - unsafe { - if libc::socketpair( - libc::AF_UNIX, - libc::SOCK_SEQPACKET | libc::SOCK_CLOEXEC, - 0, - fds.as_mut_ptr(), - ) != 0 - { - return Err(io::Error::last_os_error()); - } - Ok((OwnedFd::from_raw_fd(fds[0]), OwnedFd::from_raw_fd(fds[1]))) - } -} - -/// Set (or clear) the receive timeout: a blocked [`recv`] then fails with -/// `ErrorKind::WouldBlock`. Used by the host so a hung worker can't wedge the capture thread. -pub fn set_recv_timeout(sock: BorrowedFd, timeout: Option) -> io::Result<()> { - let tv = match timeout { - Some(d) => libc::timeval { - tv_sec: d.as_secs() as libc::time_t, - tv_usec: d.subsec_micros() as libc::suseconds_t, - }, - None => libc::timeval { - tv_sec: 0, - tv_usec: 0, - }, - }; - // SAFETY: `setsockopt(SO_RCVTIMEO)` reads `size_of::()` bytes from `&tv`, a live - // stack `timeval` that outlives this synchronous call; `sock` is the caller's live socket fd. - // Nothing is retained or written through Rust pointers. - let r = unsafe { - libc::setsockopt( - sock.as_raw_fd(), - libc::SOL_SOCKET, - libc::SO_RCVTIMEO, - &tv as *const libc::timeval as *const libc::c_void, - std::mem::size_of::() as libc::socklen_t, - ) - }; - if r != 0 { - return Err(io::Error::last_os_error()); - } - Ok(()) -} - -/// Send one message (+ optionally one fd as `SCM_RIGHTS`) as a single SEQPACKET datagram. -/// Atomic per message, so concurrent senders on the same socket (the capture thread's imports, -/// the encode thread's releases) need no lock. `MSG_NOSIGNAL` turns a dead peer into `EPIPE` -/// instead of `SIGPIPE`. -pub fn send( - sock: BorrowedFd, - msg: &T, - pass_fd: Option, -) -> io::Result<()> { - let body = - serde_json::to_vec(msg).map_err(|e| io::Error::new(io::ErrorKind::InvalidData, e))?; - debug_assert!( - !body.is_empty(), - "zero-length messages are reserved for EOF" - ); - if body.len() > MAX_MSG { - return Err(io::Error::new( - io::ErrorKind::InvalidData, - "zerocopy proto message too large", - )); - } - let mut iov = libc::iovec { - iov_base: body.as_ptr() as *mut libc::c_void, - iov_len: body.len(), - }; - // Control buffer for one fd: CMSG_SPACE(4) = 24 bytes on 64-bit; [u64; 4] gives 32 bytes at - // the 8-byte alignment `cmsghdr` requires. - let mut cmsg_store = [0u64; 4]; - // SAFETY: `mhdr` is a plain-old-data C struct for which all-zero is a valid value. - let mut mhdr: libc::msghdr = unsafe { std::mem::zeroed() }; - mhdr.msg_iov = &mut iov; - mhdr.msg_iovlen = 1; - if let Some(fd) = pass_fd { - mhdr.msg_control = cmsg_store.as_mut_ptr() as *mut libc::c_void; - // SAFETY: `CMSG_SPACE`/`CMSG_LEN` are pure size computations (no memory access). - // `CMSG_FIRSTHDR(&mhdr)` returns a pointer into `cmsg_store` (non-null: msg_controllen - // ≥ one cmsghdr), which is live, 8-aligned, and large enough (32 ≥ CMSG_SPACE(4) = 24) - // for the header fields and the 4-byte fd written via `CMSG_DATA`; `write_unaligned` - // handles the data area's byte alignment. All writes stay within `cmsg_store`, which - // outlives the synchronous `sendmsg` below. - unsafe { - mhdr.msg_controllen = libc::CMSG_SPACE(4) as _; - let c = libc::CMSG_FIRSTHDR(&mhdr); - (*c).cmsg_level = libc::SOL_SOCKET; - (*c).cmsg_type = libc::SCM_RIGHTS; - (*c).cmsg_len = libc::CMSG_LEN(4) as _; - std::ptr::write_unaligned(libc::CMSG_DATA(c) as *mut i32, fd.as_raw_fd()); - } - } - // SAFETY: `sock` is the caller's live socket; `mhdr` points at the live `iov` (over `body`, - // which outlives the call) and — when an fd is passed — at `cmsg_store` (ditto). `sendmsg` - // only reads these buffers. The kernel dups the fd into the message; our `BorrowedFd` stays - // owned by the caller. - let n = unsafe { libc::sendmsg(sock.as_raw_fd(), &mhdr, libc::MSG_NOSIGNAL) }; - if n < 0 { - return Err(io::Error::last_os_error()); - } - if n as usize != body.len() { - return Err(io::Error::new( - io::ErrorKind::WriteZero, - "short sendmsg on SEQPACKET socket", - )); - } - Ok(()) -} - -/// Receive one message (+ up to one `SCM_RIGHTS` fd). `buf` is a caller-owned scratch buffer -/// (grown to [`MAX_MSG`] once, then reused frame to frame). Errors: -/// `UnexpectedEof` = the peer is gone; `WouldBlock` = the [`set_recv_timeout`] expired. -pub fn recv( - sock: BorrowedFd, - buf: &mut Vec, -) -> io::Result<(T, Option)> { - buf.resize(MAX_MSG, 0); - let mut iov = libc::iovec { - iov_base: buf.as_mut_ptr() as *mut libc::c_void, - iov_len: buf.len(), - }; - let mut cmsg_store = [0u64; 4]; - // SAFETY: `mhdr` is a plain-old-data C struct for which all-zero is a valid value. - let mut mhdr: libc::msghdr = unsafe { std::mem::zeroed() }; - mhdr.msg_iov = &mut iov; - mhdr.msg_iovlen = 1; - mhdr.msg_control = cmsg_store.as_mut_ptr() as *mut libc::c_void; - mhdr.msg_controllen = std::mem::size_of_val(&cmsg_store) as _; - // SAFETY: `sock` is the caller's live socket. `recvmsg` writes at most `iov_len` bytes into - // `buf` (live for the call) and at most `msg_controllen` control bytes into `cmsg_store` - // (live, 8-aligned). `MSG_CMSG_CLOEXEC` makes any received fd CLOEXEC atomically. - let n = unsafe { libc::recvmsg(sock.as_raw_fd(), &mut mhdr, libc::MSG_CMSG_CLOEXEC) }; - if n < 0 { - return Err(io::Error::last_os_error()); - } - if n == 0 { - return Err(io::Error::new( - io::ErrorKind::UnexpectedEof, - "zerocopy proto peer closed", - )); - } - // Collect a passed fd (if any) BEFORE any early return below, so it can't leak. - let mut got_fd: Option = None; - // SAFETY: `CMSG_FIRSTHDR`/`CMSG_NXTHDR` walk the control area the kernel just wrote inside - // `cmsg_store` (bounded by the updated `mhdr.msg_controllen`), returning either null or a - // pointer to a complete `cmsghdr` within it — each dereference reads kernel-initialized - // fields in bounds. For an `SCM_RIGHTS` cmsg the data area holds whole `i32` fds; we read the - // first via `read_unaligned`. The kernel gave us ownership of that fd (it is a fresh - // descriptor in our table), so `OwnedFd::from_raw_fd` takes sole ownership — any previously - // collected `got_fd` is dropped (closed) first, so nothing leaks even with multiple cmsgs. - unsafe { - let mut c = libc::CMSG_FIRSTHDR(&mhdr); - while !c.is_null() { - if (*c).cmsg_level == libc::SOL_SOCKET && (*c).cmsg_type == libc::SCM_RIGHTS { - let fd = std::ptr::read_unaligned(libc::CMSG_DATA(c) as *const i32); - if fd >= 0 { - got_fd = Some(OwnedFd::from_raw_fd(fd)); - } - } - c = libc::CMSG_NXTHDR(&mhdr, c); - } - } - if mhdr.msg_flags & libc::MSG_TRUNC != 0 { - return Err(io::Error::new( - io::ErrorKind::InvalidData, - "zerocopy proto message truncated", - )); - } - let msg = serde_json::from_slice(&buf[..n as usize]) - .map_err(|e| io::Error::new(io::ErrorKind::InvalidData, e))?; - Ok((msg, got_fd)) -} - #[cfg(test)] mod tests { use super::*; - use std::io::{Read, Write}; + use 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::(b.as_fd(), &mut buf).unwrap(); + ipc::send(a.as_fd(), &req, None).unwrap(); + let (got, fd) = ipc::recv::(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::(a.as_fd(), &mut buf).unwrap(); + ipc::send(b.as_fd(), &reply, None).unwrap(); + let (got, fd) = ipc::recv::(a.as_fd(), &mut buf).unwrap(); assert_eq!(got, reply); assert!(fd.is_none()); } - - #[test] - fn passes_an_fd() { - let (a, b) = socketpair_seqpacket().unwrap(); - let mut buf = Vec::new(); - // A pipe stands in for a dmabuf: pass the read end, write through the original write end, - // and read the bytes back through the RECEIVED fd. - let (mut pr, mut pw) = std::io::pipe().unwrap(); - send(a.as_fd(), &Request::ClearCache, Some(pr.as_fd())).unwrap(); - let (got, fd) = recv::(b.as_fd(), &mut buf).unwrap(); - assert_eq!(got, Request::ClearCache); - let fd = fd.expect("fd should have been passed"); - pw.write_all(b"hello").unwrap(); - drop(pw); - let mut file = std::fs::File::from(fd); - let mut s = String::new(); - file.read_to_string(&mut s).unwrap(); - assert_eq!(s, "hello"); - // The original read end still works independently of the passed dup. - let mut nothing = [0u8; 1]; - assert_eq!(pr.read(&mut nothing).unwrap(), 0); - } - - #[test] - fn eof_when_peer_closes() { - let (a, b) = socketpair_seqpacket().unwrap(); - drop(a); - let mut buf = Vec::new(); - let err = recv::(b.as_fd(), &mut buf).unwrap_err(); - assert_eq!(err.kind(), io::ErrorKind::UnexpectedEof); - } - - #[test] - fn send_to_dead_peer_is_epipe_not_sigpipe() { - let (a, b) = socketpair_seqpacket().unwrap(); - drop(b); - let err = send(a.as_fd(), &Request::ClearCache, None).unwrap_err(); - // MSG_NOSIGNAL: a dead peer surfaces as EPIPE (BrokenPipe), never a process-killing signal. - assert_eq!(err.kind(), io::ErrorKind::BrokenPipe); - } - - #[test] - fn recv_timeout_fires() { - let (a, _b) = socketpair_seqpacket().unwrap(); - set_recv_timeout(a.as_fd(), Some(Duration::from_millis(50))).unwrap(); - let mut buf = Vec::new(); - let err = recv::(a.as_fd(), &mut buf).unwrap_err(); - assert!( - matches!( - err.kind(), - io::ErrorKind::WouldBlock | io::ErrorKind::TimedOut - ), - "unexpected error kind: {err:?}" - ); - } } diff --git a/crates/pf-zerocopy/src/imp/worker.rs b/crates/pf-zerocopy/src/imp/worker.rs index 4168b67d..a439384c 100644 --- a/crates/pf-zerocopy/src/imp/worker.rs +++ b/crates/pf-zerocopy/src/imp/worker.rs @@ -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::(sock.as_fd(), &mut buf) { + let (req, fd) = match ipc::recv::(sock.as_fd(), &mut buf) { Ok(v) => v, Err(e) if e.kind() == io::ErrorKind::UnexpectedEof => return Ok(()), Err(e) => return Err(e).context("worker recv"), @@ -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 { - 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, std::thread::JoinHandle>, ) { - 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::(host.as_fd(), &mut buf).unwrap(); + ipc::send(host.as_fd(), &Request::Modifiers { fourcc: 42 }, None).unwrap(); + let (reply, _) = ipc::recv::(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::(host.as_fd(), &mut buf).unwrap(); + ipc::send(host.as_fd(), &import_req(1, false), None).unwrap(); + let (reply, _) = ipc::recv::(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::(host.as_fd(), &mut buf).unwrap(); + ipc::send(host.as_fd(), &import_req(1, false), None).unwrap(); + let (reply, _) = ipc::recv::(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::(host.as_fd(), &mut buf).unwrap(); + ipc::send(host.as_fd(), &import_req(3, true), Some(pr.as_fd())).unwrap(); + let (reply, _) = ipc::recv::(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::(host.as_fd(), &mut buf).unwrap(); + ipc::send(host.as_fd(), &import_req(0xfeed, false), None).unwrap(); + let (reply, _) = ipc::recv::(host.as_fd(), &mut buf).unwrap(); assert_eq!(reply, Reply::NeedFd); // A failed import is an Err reply, not a dead worker. - proto::send(host.as_fd(), &import_req(0xbad, false), None).unwrap(); - let (reply, _) = proto::recv::(host.as_fd(), &mut buf).unwrap(); + ipc::send(host.as_fd(), &import_req(0xbad, false), None).unwrap(); + let (reply, _) = ipc::recv::(host.as_fd(), &mut buf).unwrap(); match reply { Reply::Err { message } => assert!(message.contains("scripted failure")), other => panic!("unexpected reply {other:?}"), } // Fire-and-forget ops reach the backend without replies. - proto::send(host.as_fd(), &Request::Release { id: 0 }, None).unwrap(); - proto::send(host.as_fd(), &Request::ClearCache, None).unwrap(); + 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);