diff --git a/Cargo.lock b/Cargo.lock index ab0e8608..9e97a9a2 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -3118,6 +3118,8 @@ dependencies = [ "pf-zerocopy", "punktfunk-core", "pyrowave-sys", + "serde", + "serde_json", "tracing", "tracing-subscriber", "windows 0.62.2 (registry+https://github.com/rust-lang/crates.io-index)", @@ -3624,6 +3626,15 @@ dependencies = [ "zeroize", ] +[[package]] +name = "punktfunk-encode-worker" +version = "0.26.0" +dependencies = [ + "pf-encode", + "tracing", + "tracing-subscriber", +] + [[package]] name = "punktfunk-host" version = "0.26.0" diff --git a/Cargo.toml b/Cargo.toml index 8a6b99d4..d67d3e51 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -4,6 +4,9 @@ members = [ "crates/punktfunk-core", "crates/punktfunk-host", "crates/punktfunk-host/vendor/usbip-sim", + # The capability-carrying PyroWave encode worker. A SEPARATE binary by design — never a + # hardlink of, or a subcommand of, punktfunk-host (design/gpu-priority-capability-worker.md). + "crates/punktfunk-encode-worker", "crates/punktfunk-tray", "crates/pf-bitstream", "crates/pf-bitstream/vendor/cros-codecs", diff --git a/crates/pf-encode/Cargo.toml b/crates/pf-encode/Cargo.toml index 6ffed5c6..eece77c3 100644 --- a/crates/pf-encode/Cargo.toml +++ b/crates/pf-encode/Cargo.toml @@ -34,7 +34,15 @@ pf-capture = { path = "../pf-capture" } # Software H.264 (openh264, BSD-2) — the GPU-less encode path on both platforms. openh264 = "0.9" +[target.'cfg(target_os = "linux")'.dev-dependencies] +# The encode-worker protocol tests measure what an AU costs as a serde_json body — the reason the +# access units ride a memfd instead (enc/linux/worker.rs). +serde_json = "1" + [target.'cfg(target_os = "linux")'.dependencies] +# The `punktfunk-encode-worker` protocol (enc/linux/worker.rs). The framing is pf-zerocopy's +# `ipc`, which is generic over the serde body; the message enums live here and version separately. +serde = { version = "1", features = ["derive"] } # libavcodec (NVENC libav + VAAPI backends). `ffmpeg-sys-next` auto-detects the FFmpeg version, so # this pin tracks the crate's own major (which shadows FFmpeg's): 9 = FFmpeg 9 (libavcodec 63, # libavutil 61). Arch shipped FFmpeg 9 on 2026-08-08 and every soname moved with it; the packaged diff --git a/crates/pf-encode/src/enc/linux/pyrowave.rs b/crates/pf-encode/src/enc/linux/pyrowave.rs index 8547930e..8e3b6c08 100644 --- a/crates/pf-encode/src/enc/linux/pyrowave.rs +++ b/crates/pf-encode/src/enc/linux/pyrowave.rs @@ -599,6 +599,15 @@ pub struct PyroWaveEncoder { /// Session-fixed negotiated chroma: 4:4:4 = full-res RG8 chroma plane + per-pixel CSC /// (`rgb2yuv444.comp`) + `Chroma444` pyrowave objects. chroma444: bool, + /// What the global-priority ladder in `open_inner` actually produced, kept so it can be + /// REPORTED rather than only logged. `punktfunk-encode-worker` sends it back to the host in + /// its handshake, which is the process that owns the log pipeline and knows which binary to + /// name — see [`super::worker::PriorityOutcome`]. + priority: super::worker::PriorityOutcome, + /// `VkPhysicalDeviceProperties::deviceName` of the device this encoder opened. Sanity for the + /// same handshake: on a multi-GPU host, "which GPU is the worker on" is otherwise invisible + /// from the host process. + device_name: String, /// Per-frame bitstream budget (hard CBR): `bitrate / (8 * fps)`. frame_budget: usize, /// `PUNKTFUNK_PERF`: the synchronous encode's own duration, which is the quantity the @@ -679,6 +688,18 @@ impl PyroWaveEncoder { ); } + /// What the global-priority ladder produced for this encoder — the quantity + /// `punktfunk-encode-worker` reports back so the host can log the grant (or the INERT refusal) + /// once, naming the right binary. + pub(crate) fn priority_outcome(&self) -> super::worker::PriorityOutcome { + self.priority + } + + /// The Vulkan device this encoder opened on. + pub(crate) fn device_name(&self) -> &str { + &self.device_name + } + pub fn open( width: u32, height: u32, @@ -686,7 +707,54 @@ impl PyroWaveEncoder { bitrate_bps: u64, chroma: crate::ChromaFormat, ) -> Result { - if !chroma.is_444() && (width % 2 != 0 || height % 2 != 0) { + // The in-process path reads the intent from ITS OWN environment, exactly as it always + // has, and owns the INERT warn. (`punktfunk-encode-worker` takes both from its parent — + // see `open_in_worker`.) + let intent = std::env::var("PYROWAVE_QUEUE_PRIORITY").ok(); + Self::open_checked( + width, + height, + fps, + bitrate_bps, + chroma.is_444(), + intent.as_deref(), + true, + ) + } + + /// [`Self::open`] as `punktfunk-encode-worker` runs it. + /// + /// Two things differ, and only two — the encoder itself is opened by the identical code path, + /// which is what keeps the worker/in-process A/B honest: + /// + /// * `intent` arrives **explicitly** from the host's handshake rather than from this process's + /// environment (which the worker strips of `PYROWAVE_QUEUE_PRIORITY` at startup), so one + /// operator knob cannot come to mean two different things across the process boundary; + /// * the INERT warn is **left to the host**. It is the process with the log pipeline, and its + /// wording has to name the worker binary — the historical text says "CAP_SYS_NICE on the + /// host binary", which after 0.26.0-1 would send an operator to do the one thing that + /// breaks every KDE session. + pub(crate) fn open_in_worker( + width: u32, + height: u32, + fps: u32, + bitrate_bps: u64, + chroma444: bool, + intent: Option<&str>, + ) -> Result { + Self::open_checked(width, height, fps, bitrate_bps, chroma444, intent, false) + } + + fn open_checked( + width: u32, + height: u32, + fps: u32, + bitrate_bps: u64, + chroma444: bool, + intent: Option<&str>, + warn_inert: bool, + ) -> Result { + if !chroma444 && (width % 2 != 0 || height % 2 != 0) { bail!("pyrowave 4:2:0 needs even dimensions (got {width}x{height})"); } // Checked against the chroma actually being opened, NOT hardcoded 4:4:4. The 4:2:0 block @@ -697,11 +765,11 @@ impl PyroWaveEncoder { // (its own bounds `assert` is compiled out by the Release vendored build). // `validate_dimensions` rejects the impossible-at-any-chroma modes earlier; this is the // 4:4:4-specific half plus defence in depth for the lab override. - if !crate::pyrowave_mode_fits_rdo(width, height, chroma.is_444()) { + if !crate::pyrowave_mode_fits_rdo(width, height, chroma444) { bail!( "pyrowave {} at {width}x{height} exceeds the rate controller's 16-bit block \ index (see pyrowave-sys patches/0002 note) — lower the resolution", - if chroma.is_444() { "4:4:4" } else { "4:2:0" } + if chroma444 { "4:4:4" } else { "4:2:0" } ); } // SAFETY: `open_inner` only issues Vulkan/pyrowave calls whose preconditions it @@ -713,12 +781,27 @@ impl PyroWaveEncoder { height, fps.max(1), bitrate_bps.max(1_000_000), - chroma.is_444(), + chroma444, + intent, + warn_inert, ) } } - unsafe fn open_inner(w: u32, h: u32, fps: u32, bitrate: u64, chroma444: bool) -> Result { + /// `intent` is the raw `PYROWAVE_QUEUE_PRIORITY` value (`None` = unset ⇒ the default ladder), + /// resolved by the CALLER: in-process from this process's environment, in the worker from the + /// host's handshake. `warn_inert` decides whether THIS process emits the "every class refused" + /// warning — see [`Self::open_in_worker`]. + #[allow(clippy::too_many_arguments)] + unsafe fn open_inner( + w: u32, + h: u32, + fps: u32, + bitrate: u64, + chroma444: bool, + intent: Option<&str>, + warn_inert: bool, + ) -> Result { let entry = ash::Entry::load().context("load vulkan loader")?; let mut hold = DeviceHold { @@ -860,8 +943,7 @@ impl PyroWaveEncoder { // Granite takes the inherit branch and the patch has never done anything here. This // is the Linux half. Must be pushed BEFORE the count/as_ptr wiring below, exactly // like queue_family_foreign above. - let gp_candidates = - queue_priority_candidates(std::env::var("PYROWAVE_QUEUE_PRIORITY").ok().as_deref()); + let gp_candidates = queue_priority_candidates(intent); // Enable whichever alias the driver advertises (KHR = the promoted name), mirroring // pf-zerocopy's VkBridge probe so the two can never disagree about the spelling. let gp_ext = @@ -943,13 +1025,18 @@ impl PyroWaveEncoder { // created with. (The extension itself stays enabled and that is correct: it // IS enabled on the device, it just carries no request.) hold._queue_ci[0].p_next = std::ptr::null(); - if !gp_candidates.is_empty() && gp.is_some() { + if !gp_candidates.is_empty() && gp.is_some() && warn_inert { // MEASURED on .21 (RTX 5070 Ti, NVIDIA 610.43.02, 2026-08-08), and it is // not a vendor quirk: an unprivileged host is refused EVERY class, and the // same binary with `cap_sys_nice+ep` is granted REALTIME on the first // attempt. So this arm is the normal state of a packaged host today, the // lever is inert until the capability ships, and the message has to say // which capability rather than leave an operator guessing. + // + // `warn_inert` is false in `punktfunk-encode-worker`: it reports the + // outcome to its parent, which logs the same sentence naming the WORKER + // binary. Sending an operator to `setcap` the host — which this wording + // does — is precisely the 0.26.0-1 incident. tracing::warn!( "pyrowave: every global queue priority class was refused — encoding \ at default priority. The GPU-preemption lever is INERT without \ @@ -962,9 +1049,33 @@ impl PyroWaveEncoder { .context("create device")? } }; - Ok((pd, family, device, foreign_qfi)) + // The ladder's outcome, made reportable. `queue_priority_candidates` only ever yields + // REALTIME or HIGH, so the `Some(_)` arm is exact rather than a fallback (ash models + // the class as a newtype, not a Rust enum, so this cannot be a `match` on constants). + let priority = match chosen { + Some(c) if c == vk::QueueGlobalPriorityKHR::REALTIME => { + super::worker::PriorityOutcome::Granted(super::worker::GrantedClass::Realtime) + } + Some(_) => { + super::worker::PriorityOutcome::Granted(super::worker::GrantedClass::High) + } + // Exactly the condition the INERT warn above fires on: something was asked for, + // the extension was there, and every class came back refused. + None if !gp_candidates.is_empty() && gp.is_some() => { + super::worker::PriorityOutcome::Refused + } + None => super::worker::PriorityOutcome::NotRequested, + }; + let device_name = instance + .get_physical_device_properties(pd) + .device_name_as_c_str() + .ok() + .and_then(|s| s.to_str().ok()) + .unwrap_or("unknown") + .to_string(); + Ok((pd, family, device, foreign_qfi, priority, device_name)) })(); - let (pd, family, device, foreign_qfi) = match selected { + let (pd, family, device, foreign_qfi, priority, device_name) = match selected { Ok(v) => v, Err(e) => { instance.destroy_instance(None); @@ -1013,6 +1124,8 @@ impl PyroWaveEncoder { height: h, fps, chroma444, + priority, + device_name, frame_budget: budget_for(bitrate, fps), perf_us: Vec::new(), perf_logged_at: None, diff --git a/crates/pf-encode/src/enc/linux/pyrowave_remote.rs b/crates/pf-encode/src/enc/linux/pyrowave_remote.rs new file mode 100644 index 00000000..4b79525f --- /dev/null +++ b/crates/pf-encode/src/enc/linux/pyrowave_remote.rs @@ -0,0 +1,1422 @@ +//! Host half of `punktfunk-encode-worker` (design: `design/gpu-priority-capability-worker.md` §3; +//! plan §2/WP2). The worker half and the vocabulary they share are [`super::worker`]. +//! +//! [`RemotePyroWave`] is an `Encoder` that forwards a PyroWave session to the capability-carrying +//! worker process, and it sits **under** `TrackedEncoder` — session accounting, the encode-stall +//! watchdog, encoder recovery and the forwarding rot-guard all apply to it unchanged. +//! +//! ## The ladder — no rung may kill a negotiated session +//! +//! A PyroWave open is only ever reached by a session that already negotiated PyroWave, so a hard +//! error here is a dead stream, not a fallback to another codec. Every rung therefore ends at the +//! **in-process encoder exactly as today**, at default GPU priority, with one warning: +//! +//! | rung | where | outcome | +//! |---|---|---| +//! | `PUNKTFUNK_ENCODE_WORKER=off` | [`resolve_worker_path`] | in-process, one info line | +//! | binary not found | [`resolve_worker_path`] | in-process, one warn | +//! | spawn failed | [`open_preferring_worker`] | in-process, one warn | +//! | handshake timed out / worker died starting | [`spawn_link`] | in-process, one warn | +//! | proto or workspace-version skew | [`spawn_link`] | in-process, one warn | +//! | the worker could not open its encoder | [`spawn_link`] | in-process, one warn | +//! | a non-dmabuf frame arrived | [`RemotePyroWave::submit`] | in-process for the session, one warn | +//! | the worker failed a frame | [`RemotePyroWave::submit`] | in-process for the session, one warn | +//! | socket EOF mid-session | [`RemotePyroWave::reset`] | one respawn, then in-process | +//! +//! The first six happen before any frame and return the bare in-process encoder — no proxy, no +//! per-frame cost on the path that is still the overwhelming majority of hosts. The last three +//! happen inside a live proxy, which keeps its own in-process encoder from then on. +//! +//! Every fallback line ends with the same clause — *"encoding in-process at default GPU +//! priority"* — so one `grep` finds them all, whichever rung fired. + +use super::worker::{self, FromWorker, PriorityOutcome, ToWorker, WireCursor}; +use crate::pyrowave_wire::{stream_chunk_step, AuChunker}; +use crate::{AuChunk, ChromaFormat, EncodedFrame, Encoder, EncoderCaps}; +use anyhow::{bail, Context, Result}; +use pf_frame::{CapturedFrame, FramePayload}; +use pf_zerocopy::ipc; +use std::collections::{HashSet, VecDeque}; +use std::fs::File; +use std::io; +use std::os::fd::{AsFd, AsRawFd, BorrowedFd, OwnedFd}; +use std::os::unix::fs::FileExt; +use std::path::{Path, PathBuf}; +use std::process::Child; +use std::time::Duration; + +/// The installed name every packaging channel writes — and the ONLY file that may carry +/// `cap_sys_nice=ep`. Never a hardlink of `punktfunk-host` and never a subcommand of it: a shared +/// inode shares the file capability, which makes the host unidentifiable to KWin and kills every +/// KDE desktop session (0.26.0-1). +const WORKER_BIN: &str = "punktfunk-encode-worker"; + +/// Handshake budget — the same one the zerocopy worker's GPU bring-up gets. It covers a Vulkan +/// instance + device create and the pyrowave object build; a cold driver load can take seconds. +/// Blowing it means the driver is wedged, in which case the in-process encoder would wedge too, +/// so a generous budget costs nothing that the fallback would have saved. +const HANDSHAKE_TIMEOUT: Duration = Duration::from_secs(20); + +/// Per-request budget. A PyroWave encode is 2–5 ms and the encoder's own fence wait is capped at +/// 5 s, so a worker silent for twice that is wedged inside a driver call, not slow. +const REPLY_TIMEOUT: Duration = Duration::from_secs(10); + +/// Where the worker binary is, or why we are not using one. +#[derive(Debug, PartialEq, Eq)] +pub(crate) enum WorkerPath { + /// `PUNKTFUNK_ENCODE_WORKER=off` — the debug escape hatch that makes the worker/in-process + /// A/B a one-line change. + Off, + Found(PathBuf), + /// Not beside the host binary and not on `PATH` (a source build, or a partial install). + Missing, +} + +/// Resolve the worker binary: `PUNKTFUNK_ENCODE_WORKER` → alongside `/proc/self/exe` → `PATH`. +/// +/// The env override is **load-bearing on NixOS**, not a convenience: a file capability cannot live +/// on a read-only store path, so the module exposes the worker through `security.wrappers` and +/// points this variable at the wrapper (an ambient grant is fine *there* — the worker is not a KWin +/// client). Pure and fully injected so the table can be tested without mutating process env, which +/// races `getenv` in parallel tests. +fn resolve_worker_path_in( + env: Option<&str>, + exe_dir: Option<&Path>, + path_var: Option<&str>, + exists: &dyn Fn(&Path) -> bool, +) -> WorkerPath { + if let Some(v) = env.map(str::trim).filter(|v| !v.is_empty()) { + if v.eq_ignore_ascii_case("off") { + return WorkerPath::Off; + } + // Deliberately NOT existence-checked. An operator who names a path is entitled to a + // failure that names it back — the spawn rung's warn carries the path, where a silent + // fall-through to "beside the host binary" would hide the typo behind a working stream. + return WorkerPath::Found(PathBuf::from(v)); + } + if let Some(p) = exe_dir.map(|d| d.join(WORKER_BIN)).filter(|p| exists(p)) { + return WorkerPath::Found(p); + } + if let Some(p) = path_var + .into_iter() + .flat_map(|v| v.split(':')) + .filter(|d| !d.is_empty()) + .map(|d| Path::new(d).join(WORKER_BIN)) + .find(|p| exists(p)) + { + return WorkerPath::Found(p); + } + WorkerPath::Missing +} + +/// [`resolve_worker_path_in`] against the real process. +pub(crate) fn resolve_worker_path() -> WorkerPath { + // `current_exe()` readlinks `/proc/self/exe`, which reads `" (deleted)"` after the + // binary was replaced under a running host — harmless HERE, because only the parent directory + // is used and the suffix lands on the file name. (The worker we then resolve is pinned by fd + // before it is exec'd, which is where that trap actually bites.) + let exe_dir = std::env::current_exe().ok(); + let path_var = std::env::var("PATH").ok(); + resolve_worker_path_in( + std::env::var("PUNKTFUNK_ENCODE_WORKER").ok().as_deref(), + exe_dir.as_deref().and_then(Path::parent), + path_var.as_deref(), + &|p| p.is_file(), + ) +} + +/// The negotiated session's parameters, kept so the in-process fallback can be opened at ANY point +/// mid-session with exactly what the worker was opened with. +#[derive(Clone, Copy)] +struct Params { + width: u32, + height: u32, + fps: u32, + chroma: ChromaFormat, +} + +/// A per-frame failure and what it means for the session. +enum Fail { + /// The transport broke: the worker is gone or wedged. Surfaces as an `Err` from `submit` so + /// the host's existing encoder-rebuild path runs — [`RemotePyroWave::reset`] is where the + /// single respawn attempt lives. + Dead(anyhow::Error), + /// The worker is alive and said no to this frame (an unimportable dmabuf, a frame that is not + /// the session's mode). Pins the session in-process, because the recovery machinery for those + /// causes lives there: the raw-dmabuf degrade latch is a process-wide static in the HOST, and + /// an import failure noted inside the worker dies with it. + Encode(String), +} + +/// A live worker: its socket, its child, and the two caches that keep the steady state free of +/// descriptors. +#[derive(Debug)] +struct Link { + sock: OwnedFd, + child: Option, + /// The worker's AU return buffer (a memfd), received once in `Ready`. Every AU is `pwrite`n at + /// offset 0 there and `pread` back here — see [`super::worker`] for why the bytes cannot ride + /// in the message body. `None` only between the spawn and the handshake, because the `Link` + /// exists that early so its `Drop` reaps the child on every failure path. + au_buf: Option, + rbuf: Vec, + /// Dmabuf keys whose fd the worker already holds; the fd crosses only on first sight. + sent_keys: HashSet, + /// The cursor bitmap `serial` the worker has pixels for. A moving pointer re-sends position + /// only, exactly as the in-process blend re-uses its uploaded texture. + cursor_serial: Option, +} + +impl Drop for Link { + fn drop(&mut self) { + // The worker exits on socket EOF, which is `self.sock` dropping right after this. Hand the + // child to the shared reaper rather than waiting on it: a worker wedged inside a driver + // ioctl sits in D state and ignores SIGKILL, and session teardown must never block behind + // a process that may never die (plan §4 R3). + if let Some(child) = self.child.take() { + ipc::park_child(child); + } + } +} + +impl Link { + /// Send one request and take its single reply. Every host→worker message has exactly one + /// reply, so the two sides cannot desync into "whose turn is it". + fn request(&mut self, msg: &ToWorker, fds: &[BorrowedFd]) -> Result { + worker::send_eintr(self.sock.as_fd(), msg, fds).context("send to the encode worker")?; + let (reply, _) = worker::recv_eintr::( + self.sock.as_fd(), + &mut self.rbuf, + Some(REPLY_TIMEOUT), + ) + .context("no reply from the encode worker")?; + Ok(reply) + } + + /// Encode one frame across the socket. The reply doubles as the buffer-release signal, which + /// is exactly `Encoder::submit`'s existing lifetime contract — the caller already holds the + /// frame alive until its AU comes back. + fn encode(&mut self, frame: &CapturedFrame) -> Result { + let FramePayload::Dmabuf(d) = &frame.payload else { + // Unreachable: `submit` routes every non-dmabuf payload in-process before reaching here. + return Err(Fail::Encode("not a dmabuf payload".into())); + }; + let key = dmabuf_key(d.fd.as_fd()).map_err(|e| Fail::Dead(e.into()))?; + // The upload is built ONCE and re-sent on a `NeedFd` retry: the worker drops every + // descriptor of a frame it refuses, so a retry that dropped the cursor would blend a stale + // pointer for the rest of that bitmap's life. + // An EMPTY bitmap uploads too, deliberately: the invariant the worker checks is "the host + // has sent pixels for every serial it asks me to blend", and an empty overlay is a real, + // handled state (`prep_cursor` takes its no-cursor arm on one). Special-casing it here + // would make "no upload" mean two different things — new-and-empty, or already-sent — and + // the worker could no longer tell a desync from a blank pointer. + let upload = match frame.cursor.as_ref() { + Some(c) if self.cursor_serial != Some(c.serial) => Some( + worker::cursor_upload(&c.rgba) + .map_err(|e| Fail::Dead(anyhow::Error::from(e).context("stage the cursor")))?, + ), + _ => None, + }; + let cursor = frame.cursor.as_ref().map(|c| WireCursor { + x: c.x, + y: c.y, + w: c.w, + h: c.h, + serial: c.serial, + hot_x: c.hot_x, + hot_y: c.hot_y, + visible: c.visible, + upload: upload.as_ref().map(|(_, n)| *n), + }); + + let mut attempts = 0; + let reply = loop { + attempts += 1; + let has_fd = self.sent_keys.insert(key); + let msg = ToWorker::Frame { + key, + has_fd, + fourcc: d.fourcc, + modifier: d.modifier, + offset: d.offset, + stride: d.stride, + plane1: d.plane1, + width: frame.width, + height: frame.height, + pts_ns: frame.pts_ns, + format: frame.format.into(), + cursor: cursor.clone(), + }; + // Descriptor ORDER is the protocol: dmabuf first (iff first sight), cursor second. + let mut fds: Vec = Vec::new(); + if has_fd { + fds.push(d.fd.as_fd()); + } + if let Some((f, _)) = upload.as_ref() { + fds.push(f.as_fd()); + } + match self.request(&msg, &fds).map_err(Fail::Dead)? { + // The worker's fd cache evicted this key (or the two diverged): forget our + // "already sent" note and retry ONCE, with the fd. + FromWorker::NeedFd if attempts == 1 => { + self.sent_keys.remove(&key); + continue; + } + FromWorker::NeedFd => { + return Err(Fail::Dead(anyhow::anyhow!( + "the encode worker still lacks the dmabuf fd after a resend (desync)" + ))) + } + other => break other, + } + }; + match reply { + FromWorker::Au { + len, + pts_ns, + keyframe, + chunk_aligned, + encode_us, + .. + } => { + let mut data = vec![0u8; len]; + self.au_buf + .as_ref() + .ok_or_else(|| { + Fail::Dead(anyhow::anyhow!("no AU return buffer (no handshake)")) + })? + .read_exact_at(&mut data, 0) + .map_err(|e| { + Fail::Dead(anyhow::Error::from(e).context("read the AU return buffer")) + })?; + if let Some(c) = frame.cursor.as_ref() { + if upload.is_some() { + self.cursor_serial = Some(c.serial); + } + } + tracing::trace!(len, encode_us, "pyrowave: AU from the encode worker"); + Ok(EncodedFrame { + data, + pts_ns, + keyframe, + recovery_anchor: false, + chunk_aligned, + }) + } + FromWorker::EncodeErr { message } => Err(Fail::Encode(message)), + other => Err(Fail::Dead(anyhow::anyhow!( + "unexpected encode worker reply to a frame: {other:?}" + ))), + } + } +} + +/// The dmabuf's identity across frames: its inode. dma-buf objects live on one anonymous inode +/// filesystem and the number is unique per object, which is what makes the fd-identity cache +/// possible — the same key the zerocopy worker's importer uses. +fn dmabuf_key(fd: BorrowedFd) -> io::Result { + // SAFETY: `libc::stat` is plain-old-data for which all-zero is a valid value; `fstat` writes + // into the live, correctly-sized `&mut st` and only reads `fd`, which the caller keeps open + // for the duration. `st_ino` is read only after the return value is checked. + unsafe { + let mut st: libc::stat = std::mem::zeroed(); + if libc::fstat(fd.as_raw_fd(), &mut st) != 0 { + return Err(io::Error::last_os_error()); + } + Ok(st.st_ino as u64) + } +} + +/// What a completed handshake told us about the worker. +#[derive(Debug)] +struct Handshake { + link: Link, + caps: EncoderCaps, + priority: PriorityOutcome, + device: String, +} + +/// Spawn a worker on `exe` and complete the handshake. Any `Err` is a ladder rung: the caller +/// warns once and encodes in-process. +fn spawn_link(exe: &Path, p: &Params, bitrate_bps: u64) -> Result { + // Pin the WORKER's inode, never `self_exe()`: this binary is a different file from the host by + // construction (it carries the capability the host must never have). Pinning also means a + // package upgrade landing between here and the exec still runs the build we resolved. + let pinned = + ipc::PinnedExe::open(exe).with_context(|| format!("open {} for exec", exe.display()))?; + let (sock, child) = ipc::spawn_worker(&pinned.exec_path(), WORKER_BIN, &["--fd", "3"]) + .with_context(|| format!("spawn {}", exe.display()))?; + // Built before the handshake ON PURPOSE: its `Drop` is what hands the child to the reaper, so + // every `?` in `handshake` reaps rather than leaving a worker behind. + handshake( + Link { + sock, + child: Some(child), + au_buf: None, + rbuf: Vec::new(), + sent_keys: HashSet::new(), + cursor_serial: None, + }, + p, + bitrate_bps, + ) +} + +/// The handshake itself, split from the spawn so the ladder's rungs are testable against a plain +/// socket — and, more to the point, so the tests exercise THIS code rather than a copy of it that +/// can drift from it. +fn handshake(mut link: Link, p: &Params, bitrate_bps: u64) -> Result { + let hello = ToWorker::Hello { + proto: worker::PROTO_VERSION, + workspace_version: worker::WORKSPACE_VERSION.to_string(), + drm_node: std::env::var("PUNKTFUNK_RENDER_NODE").ok(), + width: p.width, + height: p.height, + fps: p.fps, + bitrate_bps, + chroma444: p.chroma.is_444(), + // Resolved HERE and forwarded explicitly. The worker strips this variable from its own + // environment, so the operator's knob cannot silently mean something different across + // the process boundary. + priority_intent: std::env::var("PYROWAVE_QUEUE_PRIORITY").ok(), + }; + worker::send_eintr(link.sock.as_fd(), &hello, &[]).context("send Hello")?; + let (ready, fds) = worker::recv_eintr::( + link.sock.as_fd(), + &mut link.rbuf, + Some(HANDSHAKE_TIMEOUT), + ) + .context("encode worker handshake (died on startup?)")?; + match ready { + FromWorker::Ready { + proto, + workspace_version, + priority, + device, + chroma444, + blends_cursor, + } => { + // Load-bearing, unlike the zerocopy worker's formality of a check: host and worker are + // different FILES here, so a channel that shipped them out of lockstep is a real + // deployment state — and it must degrade to the in-process encoder, not to a session + // that cannot decode its own peer. + if proto != worker::PROTO_VERSION || workspace_version != worker::WORKSPACE_VERSION { + bail!( + "encode worker version skew: worker proto {proto} v{workspace_version}, \ + host proto {} v{} — host and worker must ship lockstep", + worker::PROTO_VERSION, + worker::WORKSPACE_VERSION + ); + } + let au_buf = fds + .into_iter() + .next() + .context("Ready carried no AU return buffer")?; + link.au_buf = Some(File::from(au_buf)); + Ok(Handshake { + link, + caps: EncoderCaps { + // The REAL opened values, not a guess: a hardcoded default mis-reports a + // 4:4:4 open and fires the session glue's spurious "chroma disagrees with the + // negotiated Welcome" warn. + blends_cursor, + chroma_444: chroma444, + ..EncoderCaps::default() + }, + priority, + device, + }) + } + FromWorker::InitErr { message } => { + bail!("encode worker could not open its encoder: {message}") + } + other => bail!("unexpected encode worker handshake: {other:?}"), + } +} + +/// Open a PyroWave session, preferring the capability-carrying worker. +/// +/// This is the ONE seam the Linux `open_video` PyroWave arms go through. It is deliberately not +/// wired into the Windows arm: that platform has no such worker, and pointing it at a Linux-only +/// binary would be a fallback rung firing on every Windows session. +pub(crate) fn open_preferring_worker( + width: u32, + height: u32, + fps: u32, + bitrate_bps: u64, + chroma: ChromaFormat, +) -> Result> { + let params = Params { + width, + height, + fps, + chroma, + }; + // Every rung below ends here — the in-process encoder, exactly as it opened before this + // worker existed. + let inline = || -> Result> { + super::pyrowave::PyroWaveEncoder::open(width, height, fps, bitrate_bps, chroma) + .map(|e| Box::new(e) as Box) + }; + let path = match resolve_worker_path() { + WorkerPath::Off => { + tracing::info!( + "pyrowave: PUNKTFUNK_ENCODE_WORKER=off — encoding in-process at default GPU priority" + ); + return inline(); + } + WorkerPath::Missing => { + tracing::warn!( + worker = WORKER_BIN, + "pyrowave: the encode worker was not found beside the host binary or on PATH — \ + encoding in-process at default GPU priority (the GPU-preemption lever needs the \ + capability-carrying worker; set PUNKTFUNK_ENCODE_WORKER to its path)" + ); + return inline(); + } + WorkerPath::Found(p) => p, + }; + let hs = match spawn_link(&path, ¶ms, bitrate_bps) { + Ok(hs) => hs, + Err(e) => { + tracing::warn!( + worker = %path.display(), + error = %format!("{e:#}"), + "pyrowave: the encode worker did not come up — encoding in-process at default \ + GPU priority" + ); + return inline(); + } + }; + match hs.priority { + PriorityOutcome::Granted(class) => tracing::info!( + priority = ?class, + device = %hs.device, + worker = %path.display(), + "pyrowave: encoding in the capability-carrying worker at an elevated global queue \ + priority (the encode dispatch preempts a GPU-bound game where the driver honors it)" + ), + PriorityOutcome::Refused => tracing::warn!( + device = %hs.device, + worker = %path.display(), + "pyrowave: every global queue priority class was refused — encoding at default \ + priority. The GPU-preemption lever is INERT without CAP_SYS_NICE on the encode \ + WORKER binary (never on punktfunk-host — a capability there makes the host \ + unidentifiable to KWin and kills desktop streaming); PYROWAVE_QUEUE_PRIORITY=off \ + silences this" + ), + PriorityOutcome::NotRequested => tracing::info!( + device = %hs.device, + worker = %path.display(), + "pyrowave: encoding in the capability-carrying worker, no queue priority requested" + ), + } + Ok(Box::new(RemotePyroWave { + link: Some(hs.link), + inline: None, + params, + bitrate_bps, + worker_path: path, + caps: hs.caps, + wire_chunk: None, + pending: VecDeque::new(), + chunker: None, + respawn_used: false, + })) +} + +/// A PyroWave session encoded in `punktfunk-encode-worker`, with the in-process encoder underneath +/// it as the floor. See the module docs for the ladder. +pub(crate) struct RemotePyroWave { + /// The live worker, or `None` once this session is pinned in-process. + link: Option, + /// The in-process encoder, opened lazily on any mid-session rung. Once open it serves the rest + /// of the session — the worker is not re-attempted per frame. + inline: Option, + params: Params, + /// The live rate, so a fallback opens the in-process encoder at the bitrate ABR last set + /// rather than the one the session started with. + bitrate_bps: u64, + worker_path: PathBuf, + caps: EncoderCaps, + /// The datagram-aligned boundary, mirrored HERE as well as forwarded. It has to cross (it + /// changes the AU bytes), and it has to be kept (it decides this proxy's own chunked-poll + /// answers) — see [`Self::poll_chunk`]. + wire_chunk: Option, + /// AUs the worker returned and the caller has not polled yet. Empty in in-process mode, where + /// the inline encoder owns its own queue — except for whatever was still here when the + /// fallback fired, which [`Self::poll_whole`] drains first. + pending: VecDeque, + /// The AU currently being handed out in streamed chunks — the proxy's, in BOTH modes, so + /// exactly one chunker can ever be open (see [`Self::poll_chunk`]). + chunker: Option, + /// One respawn per session. After that the in-process encoder is the answer: a worker that + /// dies twice is a worker that will keep dying, and burning the host's five-reset budget on it + /// costs the session. + respawn_used: bool, +} + +impl RemotePyroWave { + /// Drop the worker and encode in-process for the rest of the session, with the one warn every + /// mid-session rung owes. + fn pin_inline(&mut self, reason: &str) { + if self.link.take().is_some() { + tracing::warn!( + worker = %self.worker_path.display(), + reason, + "pyrowave: leaving the encode worker — encoding in-process at default GPU \ + priority for the rest of this session" + ); + } + } + + /// The in-process encoder, opened on demand at the session's CURRENT parameters and with the + /// state the caller set through the trait replayed onto it. + fn inline_mut(&mut self) -> Result<&mut super::pyrowave::PyroWaveEncoder> { + if self.inline.is_none() { + let mut e = super::pyrowave::PyroWaveEncoder::open( + self.params.width, + self.params.height, + self.params.fps, + self.bitrate_bps, + self.params.chroma, + ) + .context("open the in-process PyroWave encoder after leaving the worker")?; + // Replay: the boundary changes the AU BYTES, so a fallback that forgot it would ship + // dense AUs flagged as datagram-aligned. (The bitrate needs no replay — it is an open + // parameter above.) + if let Some(shard) = self.wire_chunk { + e.set_wire_chunking(shard); + } + self.inline = Some(e); + } + Ok(self.inline.as_mut().expect("just opened")) + } + + /// One whole AU from whichever half is live, worker leftovers first. + fn poll_whole(&mut self) -> Result> { + if let Some(f) = self.pending.pop_front() { + return Ok(Some(f)); + } + match self.inline.as_mut() { + Some(e) => e.poll(), + None => Ok(None), + } + } +} + +impl Encoder for RemotePyroWave { + fn submit(&mut self, frame: &CapturedFrame) -> Result<()> { + // A CPU-backed frame can genuinely reach this encoder — a 4:4:4 PyroWave session with + // zero-copy off takes the host's `force_cpu_for_nvenc_444` arm, and the process-wide + // raw-dmabuf degrade latch flips later sessions to CPU delivery — and it must never cross + // the socket: 1080p BGRA is ~8 MB, i.e. ~480 MB/s at 60 fps. The in-process encoder + // uploads it straight into its own device, which is what the CPU arm of `submit_frame` + // has always done, so this is just one more rung of the same ladder. + if self.link.is_some() && !matches!(frame.payload, FramePayload::Dmabuf(_)) { + self.pin_inline( + "capture delivered a non-dmabuf frame, which the worker path cannot take", + ); + } + if self.link.is_some() { + match self + .link + .as_mut() + .expect("checked just above") + .encode(frame) + { + Ok(au) => { + self.pending.push_back(au); + return Ok(()); + } + Err(Fail::Dead(e)) => { + // The worker is gone. Surface it: the host's encoder-rebuild path runs, and + // `reset` below is where the single respawn attempt lives. + self.link = None; + return Err(e.context("pyrowave encode worker")); + } + Err(Fail::Encode(message)) => { + // The worker is alive but refused this frame. Re-run it in-process — the + // recovery machinery for every cause of this lives on THIS side (the + // raw-dmabuf degrade latch is a host-process static), so the caller gets the + // in-process outcome, latch included, instead of a frame dropped in a + // subprocess. + self.pin_inline(&format!("the worker failed a frame: {message}")); + } + } + } + self.inline_mut()?.submit(frame) + } + + fn submit_indexed(&mut self, frame: &CapturedFrame, wire_index: u32) -> Result<()> { + // Mirrors the in-process impl, which takes the trait default: every PyroWave AU is a + // keyframe, so there is no per-frame reference bookkeeping to pin to the wire index. + let _ = wire_index; + self.submit(frame) + } + + fn caps(&self) -> EncoderCaps { + // Once in-process, the live encoder is authoritative; before that, the values the worker + // reported for the encoder it really opened. + match self.inline.as_ref() { + Some(e) => e.caps(), + None => self.caps, + } + } + + fn request_keyframe(&mut self) { + // Intra-only: every AU is already a keyframe (mirrors the in-process impl's default). + } + + fn set_hdr_meta(&mut self, _meta: Option) { + // PyroWave carries no VUI/SEI grade — the colour contract is the CSC shader's. + } + + fn invalidate_ref_frames(&mut self, _first_frame: i64, _last_frame: i64) -> bool { + // No references to invalidate. + false + } + + fn set_pipelined(&mut self, _on: bool) -> bool { + // No pipelined-retrieve mode; the encode is synchronous by design. + false + } + + fn poll(&mut self) -> Result> { + // Trait contract (PW6): each AU is drained through ONE method. Same wording as the + // in-process impl, because it is the same caller bug. + if self.chunker.is_some() { + bail!("pyrowave: poll() on an AU already being drained through poll_chunk"); + } + self.poll_whole() + } + + fn supports_chunked_poll(&self) -> bool { + // Exactly the in-process answer: the cut rule is a pure function of the boundary, and the + // boundary is mirrored here. + stream_chunk_step(self.wire_chunk).is_some() + } + + fn poll_chunk(&mut self) -> Result> { + // The streamed-AU cut needs NO protocol. `submit` is synchronous on both sides of the + // socket, so an AU that reached `pending` is complete by construction and the identical + // `AuChunker` runs here — sub-frame send latency survives at zero wire cost. The chunker + // is the proxy's in BOTH modes (the fallback's `poll_chunk` is never called), so exactly + // one cursor can ever be open. + if let Some(c) = self.chunker.as_mut() { + if let Some(chunk) = c.next() { + return Ok(Some(chunk)); + } + self.chunker = None; + } + let Some(f) = self.poll_whole()? else { + return Ok(None); + }; + match stream_chunk_step(self.wire_chunk) { + Some(step) => Ok(self.chunker.insert(AuChunker::new(f, step)).next()), + None => Ok(Some(AuChunk::whole(f))), + } + } + + fn reset(&mut self) -> bool { + // A rebuild forfeits every in-flight frame, including an AU only half-handed-out — drop + // the cursor first so the next `poll_chunk` cannot splice the tail of a dead AU onto a + // fresh one. + self.chunker = None; + self.pending.clear(); + if let Some(link) = self.link.as_mut() { + // A message, not a respawn, and deliberately: the expensive, capability-dependent + // thing is the priority-elevated DEVICE, and an in-worker reset keeps it. A respawn + // would re-run the whole global-priority ladder mid-session and could silently land on + // a different class than the one this session has been measured at — changing the very + // quantity the worker exists to protect — and it would pay the full Vulkan (and, until + // WP-D, FFmpeg) load inside the stall watchdog's recovery window. The in-process + // `reset` it forwards to is already the bounded, wedge-aware rebuild: it re-waits the + // in-flight fences under a 5 s cap and reports failure rather than destroying a + // pyrowave encoder under live GPU work. And if the worker is itself wedged, this + // request times out and falls through to the rung below — so respawn stays reachable + // as the failure path without being the policy. + match link.request(&ToWorker::Reset, &[]) { + Ok(FromWorker::Ack { ok }) => return ok, + Ok(other) => { + tracing::warn!(?other, "pyrowave: unexpected encode worker reply to Reset"); + self.link = None; + } + Err(e) => { + tracing::warn!( + worker = %self.worker_path.display(), + error = %format!("{e:#}"), + "pyrowave: the encode worker died mid-session — rebuilding the encoder" + ); + self.link = None; + } + } + } + // One respawn, and only while nothing has fallen back yet. + if self.link.is_none() && self.inline.is_none() && !self.respawn_used { + self.respawn_used = true; + match spawn_link(&self.worker_path, &self.params, self.bitrate_bps) { + Ok(hs) => { + self.caps = hs.caps; + self.link = Some(hs.link); + if let Some(shard) = self.wire_chunk { + // Same replay the in-process fallback does, for the same reason. + self.set_wire_chunking(shard); + } + tracing::info!( + worker = %self.worker_path.display(), + priority = ?hs.priority, + "pyrowave: respawned the encode worker after a mid-session death" + ); + return true; + } + Err(e) => tracing::warn!( + worker = %self.worker_path.display(), + error = %format!("{e:#}"), + "pyrowave: the encode worker would not respawn — encoding in-process at \ + default GPU priority for the rest of this session" + ), + } + } + let already_open = self.inline.is_some(); + match self.inline_mut() { + // A freshly opened in-process encoder IS the rebuild the caller asked for. + Ok(e) => { + if already_open { + e.reset() + } else { + true + } + } + Err(e) => { + tracing::error!( + error = %format!("{e:#}"), + "pyrowave: no encoder left after the worker went away — the session cannot \ + recover" + ); + false + } + } + } + + fn reconfigure_bitrate(&mut self, bps: u64) -> bool { + // Kept regardless of which half is live: a later in-process fallback opens at the rate ABR + // actually settled on, not the one the session started with. + self.bitrate_bps = bps; + if let Some(link) = self.link.as_mut() { + match link.request(&ToWorker::Reconfigure { bitrate_bps: bps }, &[]) { + Ok(FromWorker::Ack { ok }) => return ok, + Ok(other) => { + tracing::warn!( + ?other, + "pyrowave: unexpected encode worker reply to Reconfigure" + ) + } + Err(e) => tracing::warn!( + error = %format!("{e:#}"), + "pyrowave: the encode worker did not accept a bitrate retarget" + ), + } + // Not a lie to the ABR controller: report failure and let it use its rebuild path, + // which lands on `reset` above. + return false; + } + match self.inline.as_mut() { + Some(e) => e.reconfigure_bitrate(bps), + // Nothing is open yet, and the new rate is now the one the open will use. + None => true, + } + } + + fn applied_bitrate_bps(&self) -> Option { + // Mirrors the in-process impl (the trait default): PyroWave applies the requested rate as + // a per-frame byte budget with no internal clamp to report. + None + } + + fn set_wire_chunking(&mut self, shard_payload: usize) { + // The same sanity floor as the in-process impl, applied HERE so the mirrored state and the + // worker's can never disagree about whether chunking is on. + if shard_payload < 64 { + return; + } + self.wire_chunk = Some(shard_payload); + if let Some(link) = self.link.as_mut() { + // This one really does have to cross: it changes the packetize boundary and the rate + // budget, i.e. the AU BYTES. Only the streamed-AU CUT stays host-side. + if let Err(e) = link.request(&ToWorker::SetWireChunking { shard_payload }, &[]) { + tracing::warn!( + error = %format!("{e:#}"), + "pyrowave: the encode worker did not accept the datagram-aligned boundary" + ); + } + // Deliberately NOT word-for-word the in-process line: the worker emits that one + // itself (to inherited stderr), and two identical sentences from two processes read + // as a bug. This is the host-ring copy — the web console's Logs tab only ever sees + // this process — and it says what actually happened here. + tracing::info!( + shard_payload, + "pyrowave: datagram-aligned packetization forwarded to the encode worker \ + (partial-frame loss mode)" + ); + } + if let Some(e) = self.inline.as_mut() { + e.set_wire_chunking(shard_payload); + } + } + + fn set_send_spread_us(&mut self, _us: u32) { + // Only the direct-NVENC split arbitration consumes this; PyroWave never splits. + } + + fn set_input_ring_depth(&mut self, _depth: usize) { + // The encoder imports the capture dmabuf and CSCs it into its own images, so the + // capturer's ring depth constrains nothing here. + } + + fn flush(&mut self) -> Result<()> { + // Nothing is ever in flight ACROSS the socket: `submit` returns only once the AU is in + // `pending`, so there is no worker-side backlog a flush could drain — unlike the + // in-process encoder, whose `submit`/`poll` split really does leave a fence unwaited. + match self.inline.as_mut() { + Some(e) => e.flush(), + None => Ok(()), + } + } +} + +#[cfg(test)] +mod tests { + use super::worker::GrantedClass; + use super::*; + use pf_frame::{CursorOverlay, DmabufFrame, PixelFormat}; + use std::os::fd::AsFd; + use std::sync::Arc; + + fn params() -> Params { + Params { + width: 1920, + height: 1080, + fps: 60, + chroma: ChromaFormat::Yuv420, + } + } + + /// A `Link` wired to a socket the test drives itself — the shape pf-zerocopy's importer tests + /// use, and the only way to exercise the per-frame path without a GPU. + fn mock_link(sock: OwnedFd, au_buf: File) -> Link { + Link { + sock, + child: None, + au_buf: Some(au_buf), + rbuf: Vec::new(), + sent_keys: HashSet::new(), + cursor_serial: None, + } + } + + /// [`handshake`] driven against a socket instead of a spawned child — the process rungs get + /// their own tests; this one exercises what the worker SAYS. + fn handshake_on(sock: OwnedFd) -> Result { + handshake(mock_link(sock, File::from(memfd())), ¶ms(), 40_000_000) + } + + /// A dmabuf-shaped frame whose "dmabuf" is a memfd: the host half only fstats the descriptor + /// and passes it, so this exercises the real key/cache/`SCM_RIGHTS` path. + fn frame(fd: OwnedFd, cursor: Option) -> CapturedFrame { + CapturedFrame { + width: 1920, + height: 1080, + pts_ns: 42, + format: PixelFormat::Bgrx, + payload: FramePayload::Dmabuf(DmabufFrame { + fd, + fourcc: 0x3432_5258, + modifier: 0, + plane1: None, + offset: 0, + stride: 1920 * 4, + }), + cursor, + } + } + + fn memfd() -> OwnedFd { + let (f, _) = worker::cursor_upload(&[]).unwrap(); + OwnedFd::from(f) + } + + #[test] + fn path_resolution_table() { + let here = Path::new("/opt/punktfunk/bin"); + let installed = here.join(WORKER_BIN); + let on_path = Path::new("/usr/bin").join(WORKER_BIN); + let exists = { + let (a, b) = (installed.clone(), on_path.clone()); + move |p: &Path| p == a || p == b + }; + + // `off`, in every spelling the row promises — the debug escape hatch. + for v in ["off", "OFF", " Off "] { + assert_eq!( + resolve_worker_path_in(Some(v), Some(here), Some("/usr/bin"), &exists), + WorkerPath::Off + ); + } + // An explicit path wins over both discoveries — the NixOS wrapper case… + assert_eq!( + resolve_worker_path_in( + Some("/run/wrappers/bin/punktfunk-encode-worker"), + Some(here), + Some("/usr/bin"), + &exists + ), + WorkerPath::Found("/run/wrappers/bin/punktfunk-encode-worker".into()) + ); + // …and it is NOT existence-checked, so a typo surfaces as a spawn failure naming the path + // instead of silently falling through to a worker that happens to be installed. + assert_eq!( + resolve_worker_path_in( + Some("/nope/pf-worker"), + Some(here), + Some("/usr/bin"), + &exists + ), + WorkerPath::Found("/nope/pf-worker".into()) + ); + // Empty/whitespace reads as unset, not as a path. + assert_eq!( + resolve_worker_path_in(Some(" "), Some(here), Some("/usr/bin"), &exists), + WorkerPath::Found(installed.clone()) + ); + // Beside the host binary beats PATH. + assert_eq!( + resolve_worker_path_in(None, Some(here), Some("/usr/bin"), &exists), + WorkerPath::Found(installed) + ); + // Then PATH, entry by entry, skipping empties. + assert_eq!( + resolve_worker_path_in( + None, + Some(Path::new("/nowhere")), + Some(":/nope:/usr/bin"), + &exists + ), + WorkerPath::Found(on_path) + ); + // Nothing anywhere: the "missing binary" rung. + assert_eq!( + resolve_worker_path_in(None, Some(Path::new("/nowhere")), Some("/nope"), &exists), + WorkerPath::Missing + ); + assert_eq!( + resolve_worker_path_in(None, None, None, &exists), + WorkerPath::Missing + ); + } + + /// Ladder rung: the binary exists and runs but is not a worker. `/bin/false` exits at once, so + /// the handshake reads EOF — the same rung a worker that dies during Vulkan bring-up takes. + #[test] + fn a_worker_that_exits_immediately_is_a_handshake_failure() { + let err = spawn_link(Path::new("/bin/false"), ¶ms(), 40_000_000).unwrap_err(); + let text = format!("{err:#}"); + assert!( + text.contains("handshake"), + "the rung must name the handshake: {text}" + ); + } + + /// Ladder rung: a binary that does not exist at all — an operator-set `PUNKTFUNK_ENCODE_WORKER` + /// typo, or a half-installed package. + #[test] + fn a_missing_binary_is_a_spawn_failure() { + let err = spawn_link( + Path::new("/nonexistent/punktfunk-encode-worker"), + ¶ms(), + 40_000_000, + ) + .unwrap_err(); + assert!(format!("{err:#}").contains("punktfunk-encode-worker")); + } + + /// Ladder rung: the worker started, spoke, and could not open its encoder (no Vulkan 1.3 + /// device, a missing feature). Not a death — an ANSWER, and the host encodes in-process, + /// where the same cause will either reproduce or turn out to have been the worker's own + /// environment. + #[test] + fn an_init_error_fails_the_handshake_without_looking_like_a_death() { + let (host, peer) = ipc::socketpair_seqpacket().unwrap(); + let server = std::thread::spawn(move || { + let mut buf = Vec::new(); + let _ = worker::recv_eintr::(peer.as_fd(), &mut buf, None); + worker::send_eintr( + peer.as_fd(), + &FromWorker::InitErr { + message: "GPU lacks pyrowave-required Vulkan features".into(), + }, + &[], + ) + .unwrap(); + }); + let err = handshake_on(host).unwrap_err(); + server.join().unwrap(); + let text = format!("{err:#}"); + assert!( + text.contains("could not open its encoder") && text.contains("Vulkan features"), + "the rung must carry the worker's own diagnosis: {text}" + ); + } + + /// Ladder rung: a dead worker gets **one** respawn from `reset`, then the session is + /// in-process for good. Driven with a dead socket and `/bin/false` as the worker binary, so + /// the respawn attempt is real and fails; what is pinned here is the budget, which is the part + /// that must not drift — a proxy that retried every `reset` would burn the host's five-reset + /// recovery budget on a worker that is not coming back and end the session. + #[test] + fn reset_respawns_the_worker_at_most_once() { + let (host, peer) = ipc::socketpair_seqpacket().unwrap(); + drop(peer); // the worker is already gone + let mut proxy = RemotePyroWave { + link: Some(mock_link(host, File::from(memfd()))), + inline: None, + params: params(), + bitrate_bps: 40_000_000, + // Spawns, then exits without a handshake — the respawn attempt runs for real and + // fails, which is the case this budget exists for. + worker_path: PathBuf::from("/bin/false"), + caps: EncoderCaps::default(), + wire_chunk: None, + pending: VecDeque::new(), + chunker: None, + respawn_used: false, + }; + // The reply never comes: the link dies, the one respawn is spent and fails, and the + // session is in-process from here. + let _ = proxy.reset(); + assert!(proxy.respawn_used, "the one respawn must have been spent"); + assert!(proxy.link.is_none(), "/bin/false cannot become a worker"); + // Every later reset stays in-process — the budget is spent, not renewed. + let _ = proxy.reset(); + assert!(proxy.respawn_used); + assert!(proxy.link.is_none()); + } + + /// Ladder rung: a CPU-backed frame. It genuinely reaches this encoder (a 4:4:4 session with + /// zero-copy off, or after the raw-dmabuf degrade latch fires), and ~8 MB per frame must never + /// cross the socket — so the session leaves the worker for good. Asserts the CLASSIFICATION + /// (the link is dropped) and not the submit result, which depends on whether the box running + /// the test has a Vulkan device to open the in-process encoder on. + #[test] + fn a_cpu_frame_leaves_the_worker_for_the_rest_of_the_session() { + let (host, peer) = ipc::socketpair_seqpacket().unwrap(); + let mut proxy = RemotePyroWave { + link: Some(mock_link(host, File::from(memfd()))), + inline: None, + params: Params { + width: 64, + height: 64, + fps: 60, + chroma: ChromaFormat::Yuv420, + }, + bitrate_bps: 5_000_000, + worker_path: PathBuf::from("/usr/bin/punktfunk-encode-worker"), + caps: EncoderCaps::default(), + wire_chunk: None, + pending: VecDeque::new(), + chunker: None, + respawn_used: false, + }; + let cpu = CapturedFrame { + width: 64, + height: 64, + pts_ns: 0, + format: PixelFormat::Bgrx, + payload: FramePayload::Cpu(vec![0u8; 64 * 64 * 4]), + cursor: None, + }; + let _ = proxy.submit(&cpu); + assert!( + proxy.link.is_none(), + "a non-dmabuf payload must pin the session in-process" + ); + // Nothing was ever asked of the worker: the rung fires before the socket is touched. + drop(proxy); + let mut buf = Vec::new(); + assert_eq!( + worker::recv_eintr::( + peer.as_fd(), + &mut buf, + Some(Duration::from_millis(200)) + ) + .unwrap_err() + .kind(), + io::ErrorKind::UnexpectedEof + ); + } + + /// Ladder rung: proto/workspace skew. Host and worker are different files, so this is a real + /// deployment state and must land on the in-process encoder rather than a broken session. + #[test] + fn version_skew_fails_the_handshake() { + for (proto, version) in [ + ( + worker::PROTO_VERSION + 1, + worker::WORKSPACE_VERSION.to_string(), + ), + (worker::PROTO_VERSION, "0.0.1-stale".to_string()), + ] { + let (host, peer) = ipc::socketpair_seqpacket().unwrap(); + let server = std::thread::spawn(move || { + let mut buf = Vec::new(); + let _ = worker::recv_eintr::(peer.as_fd(), &mut buf, None); + worker::send_eintr( + peer.as_fd(), + &FromWorker::Ready { + proto, + workspace_version: version, + priority: PriorityOutcome::Granted(GrantedClass::Realtime), + device: "mock".into(), + chroma444: false, + blends_cursor: true, + }, + &[], + ) + .unwrap(); + }); + let err = handshake_on(host).unwrap_err(); + server.join().unwrap(); + assert!( + format!("{err:#}").contains("version skew"), + "unexpected error: {err:#}" + ); + } + } + + /// The per-frame path end to end, host side: the fd crosses ONCE, the cursor bitmap crosses + /// only when its serial changes, and the AU comes back through the memfd rather than the + /// message body. + #[test] + fn frames_pass_the_fd_once_and_the_au_comes_back_through_the_buffer() { + let (host, peer) = ipc::socketpair_seqpacket().unwrap(); + let au_buf = File::from(memfd()); + let au = vec![0x7Eu8; 300_000]; + let expect = au.clone(); + let server_buf = au_buf.try_clone().unwrap(); + let server = std::thread::spawn(move || { + let mut buf = Vec::new(); + let mut seen: Vec<(bool, Option)> = Vec::new(); + for _ in 0..3 { + let (msg, fds) = + worker::recv_eintr::(peer.as_fd(), &mut buf, None).unwrap(); + let ToWorker::Frame { + key, + has_fd, + cursor, + .. + } = msg + else { + panic!("expected a frame"); + }; + let want = usize::from(has_fd) + + usize::from(cursor.as_ref().is_some_and(|c| c.upload.is_some())); + assert_eq!(fds.len(), want, "descriptor count must match the flags"); + seen.push((has_fd, cursor.and_then(|c| c.upload))); + server_buf.write_all_at(&expect, 0).unwrap(); + worker::send_eintr( + peer.as_fd(), + &FromWorker::Au { + key, + len: expect.len(), + pts_ns: 42, + keyframe: true, + chunk_aligned: false, + encode_us: 4400, + }, + &[], + ) + .unwrap(); + } + seen + }); + + let mut link = mock_link(host, au_buf); + let cursor = |serial: u64| { + Some(CursorOverlay { + x: 1, + y: 2, + w: 32, + h: 32, + rgba: Arc::new(vec![9u8; 32 * 32 * 4]), + serial, + hot_x: 0, + hot_y: 0, + visible: true, + }) + }; + // The SAME buffer three times (one memfd, re-passed), with the cursor bitmap changing once. + let dmabuf = memfd(); + for c in [cursor(1), cursor(1), cursor(2)] { + let f = frame(dmabuf.try_clone().unwrap(), c); + let got = match link.encode(&f) { + Ok(au) => au, + Err(Fail::Dead(e)) => panic!("worker died: {e:#}"), + Err(Fail::Encode(m)) => panic!("encode error: {m}"), + }; + assert_eq!(got.data, au, "the AU must arrive byte-for-byte"); + assert_eq!(got.pts_ns, 42); + assert!(got.keyframe); + } + let seen = server.join().unwrap(); + assert_eq!( + seen, + vec![ + (true, Some(32 * 32 * 4)), + (false, None), + (false, Some(32 * 32 * 4)) + ], + "the dmabuf fd crosses once; the cursor crosses only on a serial change" + ); + } + + /// A `NeedFd` (the worker evicted the key) is answered by ONE retry carrying the fd again. + #[test] + fn need_fd_resends_the_descriptor_once() { + let (host, peer) = ipc::socketpair_seqpacket().unwrap(); + let au_buf = File::from(memfd()); + let server_buf = au_buf.try_clone().unwrap(); + let server = std::thread::spawn(move || { + let mut buf = Vec::new(); + let mut flags = Vec::new(); + for i in 0..2 { + let (msg, fds) = + worker::recv_eintr::(peer.as_fd(), &mut buf, None).unwrap(); + let ToWorker::Frame { key, has_fd, .. } = msg else { + panic!("expected a frame"); + }; + flags.push((has_fd, fds.len())); + let reply = if i == 0 { + FromWorker::NeedFd + } else { + server_buf.write_all_at(&[1, 2, 3], 0).unwrap(); + FromWorker::Au { + key, + len: 3, + pts_ns: 0, + keyframe: true, + chunk_aligned: false, + encode_us: 1, + } + }; + worker::send_eintr(peer.as_fd(), &reply, &[]).unwrap(); + } + flags + }); + let mut link = mock_link(host, au_buf); + // Pretend the key already crossed, so the first attempt sends no descriptor. + let dmabuf = memfd(); + link.sent_keys.insert(dmabuf_key(dmabuf.as_fd()).unwrap()); + let f = frame(dmabuf, None); + let au = match link.encode(&f) { + Ok(au) => au, + Err(Fail::Dead(e)) => panic!("worker died: {e:#}"), + Err(Fail::Encode(m)) => panic!("encode error: {m}"), + }; + assert_eq!(au.data, vec![1, 2, 3]); + assert_eq!(server.join().unwrap(), vec![(false, 0), (true, 1)]); + } + + /// Ladder rung: the worker dies mid-stream. The transport error must classify as `Dead` (the + /// host's rebuild path, then one respawn) and never as a per-frame encode error. + #[test] + fn a_worker_that_dies_mid_stream_is_a_transport_death() { + let (host, peer) = ipc::socketpair_seqpacket().unwrap(); + drop(peer); + let mut link = mock_link(host, File::from(memfd())); + let f = frame(memfd(), None); + match link.encode(&f) { + Err(Fail::Dead(_)) => {} + Err(Fail::Encode(m)) => panic!("a dead socket must not read as an encode error: {m}"), + Ok(_) => panic!("a dead socket must not produce an AU"), + } + } + + /// …and a worker that is alive and refuses the frame classifies as `Encode`, which pins the + /// session in-process so the raw-dmabuf degrade latch (a HOST-process static) still fires. + #[test] + fn a_refused_frame_is_an_encode_error() { + let (host, peer) = ipc::socketpair_seqpacket().unwrap(); + let server = std::thread::spawn(move || { + let mut buf = Vec::new(); + let _ = worker::recv_eintr::(peer.as_fd(), &mut buf, None).unwrap(); + worker::send_eintr( + peer.as_fd(), + &FromWorker::EncodeErr { + message: "unsupported dmabuf fourcc".into(), + }, + &[], + ) + .unwrap(); + }); + let mut link = mock_link(host, File::from(memfd())); + let f = frame(memfd(), None); + match link.encode(&f) { + Err(Fail::Encode(m)) => assert!(m.contains("fourcc")), + Err(Fail::Dead(e)) => panic!("a live worker's refusal must not read as death: {e:#}"), + Ok(_) => panic!("a refusal must not produce an AU"), + } + server.join().unwrap(); + } + + /// WP7.7 guard, applied to the proxy: every `Encoder` trait method must be explicitly written + /// here. The proxy has TWO backends under it, so an unforwarded default is worse than the + /// `TrackedEncoder` case it copies — it would silently disable a feature only in the + /// worker-backed half of the ladder, i.e. on exactly the hosts that got the new code path. + /// Source-text parse, same as `tracked_encoder_forwards_every_trait_method`. + #[test] + fn the_proxy_writes_every_trait_method() { + fn item_block<'a>(src: &'a str, marker: &str) -> &'a str { + let start = src + .find(marker) + .unwrap_or_else(|| panic!("marker {marker:?} not found — update this guard")); + let body = &src[start..]; + let end = body + .find("\n}") + .unwrap_or_else(|| panic!("no column-0 close brace after {marker:?}")); + &body[..end] + } + fn fn_names(block: &str) -> std::collections::BTreeSet<&str> { + block + .lines() + .map(str::trim_start) + .filter(|l| !l.starts_with("//")) + .filter_map(|l| l.strip_prefix("fn ")) + .map(|rest| { + rest.split(|c: char| !(c.is_ascii_alphanumeric() || c == '_')) + .next() + .expect("split yields at least one item") + }) + .collect() + } + let trait_fns = fn_names(item_block( + include_str!("../codec.rs"), + "pub trait Encoder: Send {", + )); + let impl_fns = fn_names(item_block( + include_str!("pyrowave_remote.rs"), + "impl Encoder for RemotePyroWave {", + )); + assert!( + trait_fns.len() >= 12, + "only {} trait methods parsed — the extraction markers have rotted", + trait_fns.len() + ); + let missing: Vec<_> = trait_fns.difference(&impl_fns).collect(); + assert!( + missing.is_empty(), + "Encoder methods NOT written by RemotePyroWave: {missing:?} — an unforwarded default \ + silently disables the feature for every worker-backed session." + ); + assert_eq!(trait_fns, impl_fns); + } +} diff --git a/crates/pf-encode/src/enc/linux/worker.rs b/crates/pf-encode/src/enc/linux/worker.rs new file mode 100644 index 00000000..14ce0f31 --- /dev/null +++ b/crates/pf-encode/src/enc/linux/worker.rs @@ -0,0 +1,971 @@ +//! `punktfunk-encode-worker` — the vocabulary both halves speak, and the worker half itself +//! (design: `design/gpu-priority-capability-worker.md` §3; plan §2/WP1). The host half is +//! [`super::pyrowave_remote`]. +//! +//! **Why this process exists at all.** PyroWave encodes on the same shader cores a game +//! saturates, and the only lever that preempts it is an elevated `VK_KHR_global_priority` queue — +//! which the driver grants only to a process holding `CAP_SYS_NICE`. `punktfunk-host` may never +//! hold one: KWin identifies a client by `readlink /proc//exe`, the kernel refuses that +//! readlink to a reader whose effective set is not a superset of the target's PERMITTED set, and +//! a capped host is therefore unidentifiable — 0.26.0-1 killed every KDE desktop session that +//! way. So the capability lives here, in a leaf that fronts nothing: no Wayland, no D-Bus, no +//! network, one socket to its parent. +//! +//! 🛑 This worker is a **separate executable file**, never a hardlink of the host and never a +//! subcommand of it (unlike the zerocopy worker, which deliberately re-execs the host image). A +//! shared inode shares the file capability, which silently re-creates 0.26.0-1. +//! +//! ## Shape +//! +//! One worker per PyroWave session, spawned at encoder open on the shared [`ipc`] rails (SEQPACKET +//! framing, `SCM_RIGHTS`, fd-3 inheritance, pinned-exe spawn, the zombie sweep). Strict +//! request/response: **every** host→worker message gets exactly one reply, so the two sides can +//! never desync into "whose turn is it". +//! +//! ## Where the bytes go (and why they are not in the JSON) +//! +//! [`ipc::MAX_MSG`] is 64 KiB and the bodies are serde_json, which renders a `Vec` as one +//! decimal number per byte. A PyroWave AU is `bitrate / (8 × fps)` — 83 KB at 1080p60/40 Mb/s, +//! ~830 KB at 4K — so an inline `bytes` field is not a slow path, it is *unrepresentable*, and +//! base64 would still need ~17 datagrams and ~0.8 ms of codec per frame against a +1.0 ms +//! whole-IPC-hop budget (plan §4 R1). The AU therefore rides a **memfd** the worker creates once +//! and `pwrite`s at offset 0 every frame; the host `pread`s exactly `len` bytes out of it. The fd +//! crosses once, in [`FromWorker::Ready`]. A memfd grows on write, so there is no capacity +//! negotiation and no regrow protocol — a bitrate retarget is invisible to it. +//! +//! Cursor bitmaps take the same route for the same reason (256×256 RGBA = 256 KiB > `MAX_MSG`), +//! except they are rare enough (only when the pointer *image* changes) that a fresh memfd rides +//! along with the frame instead of a persistent one. +//! +//! Frame pixels never cross at all: the dmabuf fd is passed on first sight of its `key` and the +//! worker caches it, so the steady state passes **zero** descriptors (the PipeWire pool recycles a +//! small buffer set). + +// Every `unsafe` block in this file carries a `// SAFETY:` proof; enforce it (unsafe-proof program). +#![deny(clippy::undocumented_unsafe_blocks)] + +use anyhow::{Context, Result}; +use pf_frame::{CapturedFrame, CursorOverlay, DmabufFrame, FramePayload, PixelFormat}; +use pf_zerocopy::ipc; +use serde::de::DeserializeOwned; +use serde::{Deserialize, Serialize}; +use std::collections::{HashMap, VecDeque}; +use std::ffi::CStr; +use std::fs::File; +use std::io; +use std::os::fd::{AsFd, BorrowedFd, FromRawFd, OwnedFd}; +use std::os::unix::fs::FileExt; +use std::sync::Arc; +use std::time::{Duration, Instant}; + +/// Bumped on any wire change. Unlike the zerocopy worker — the same binary as its host by +/// construction — host and worker are **different files** here, so this check is load-bearing: +/// a package that shipped them out of lockstep must degrade to the in-process encoder, never to +/// a dead session. +pub(crate) const PROTO_VERSION: u32 = 1; + +/// The workspace version this half was compiled from. A protocol can be unchanged while the +/// *encoder* moves (a vendored-codec bump, a CSC shader change), and the two halves must still be +/// one build — so the handshake compares this too. `env!` resolves at compile time of THIS crate, +/// so a stale worker binary carries its own older string even though both link the same source. +pub(crate) const WORKSPACE_VERSION: &str = env!("CARGO_PKG_VERSION"); + +/// Cached dmabuf fds. PipeWire pools are ≤ ~16 buffers; the cap only matters if a producer churns +/// buffers without a renegotiation, and an eviction is recoverable ([`FromWorker::NeedFd`]). +const FD_CACHE_CAP: usize = 64; + +/// The largest cursor bitmap that can matter: the encoder clamps to a 256×256 RGBA texture +/// (`pyrowave.rs::CURSOR_MAX`), so uploading more would be bytes the blend cannot read. +const CURSOR_UPLOAD_MAX: usize = 256 * 256 * 4; + +// --------------------------------------------------------------------------- +// Vocabulary +// --------------------------------------------------------------------------- + +/// What the `VK_KHR_global_priority` ladder produced — i.e. whether the capability is doing +/// anything. Reported to the host so exactly ONE process logs it: the worker's own +/// `tracing` goes to inherited stderr, but the host is the process with the log pipeline (the +/// ring the web console serves), and the in-process INERT warn's wording ("CAP_SYS_NICE on the +/// host binary") would now actively mislead — the capability belongs on the worker. +#[derive(Serialize, Deserialize, Debug, Clone, Copy, PartialEq, Eq)] +pub(crate) enum PriorityOutcome { + /// A class was granted — the lever is live. + Granted(GrantedClass), + /// A class was requested, the extension is there, and every class was refused: the lever is + /// INERT. This is the normal state of an *uncapped* worker. + Refused, + /// Nothing was asked for (`PYROWAVE_QUEUE_PRIORITY=off`) or the device advertises no + /// global-priority extension — not a problem, and never warned about. + NotRequested, +} + +/// The granted `VkQueueGlobalPriorityKHR` class, wire-side. +#[derive(Serialize, Deserialize, Debug, Clone, Copy, PartialEq, Eq)] +pub(crate) enum GrantedClass { + Realtime, + High, +} + +/// [`pf_frame::PixelFormat`] on the wire. A hand-written mirror rather than a serde derive on the +/// original: pf-frame carries no serde dependency, and the exhaustive `match` in both directions +/// makes a new capture format a COMPILE error here instead of a silently mis-described frame. +#[derive(Serialize, Deserialize, Debug, Clone, Copy, PartialEq, Eq)] +pub(crate) enum WireFormat { + Bgrx, + Rgbx, + Bgra, + Rgba, + Rgb, + Bgr, + Rgb10a2, + Nv12, + P010, + Yuv444, + X2Rgb10, + X2Bgr10, +} + +impl From for WireFormat { + fn from(f: PixelFormat) -> WireFormat { + match f { + PixelFormat::Bgrx => WireFormat::Bgrx, + PixelFormat::Rgbx => WireFormat::Rgbx, + PixelFormat::Bgra => WireFormat::Bgra, + PixelFormat::Rgba => WireFormat::Rgba, + PixelFormat::Rgb => WireFormat::Rgb, + PixelFormat::Bgr => WireFormat::Bgr, + PixelFormat::Rgb10a2 => WireFormat::Rgb10a2, + PixelFormat::Nv12 => WireFormat::Nv12, + PixelFormat::P010 => WireFormat::P010, + PixelFormat::Yuv444 => WireFormat::Yuv444, + PixelFormat::X2Rgb10 => WireFormat::X2Rgb10, + PixelFormat::X2Bgr10 => WireFormat::X2Bgr10, + } + } +} + +impl From for PixelFormat { + fn from(f: WireFormat) -> PixelFormat { + match f { + WireFormat::Bgrx => PixelFormat::Bgrx, + WireFormat::Rgbx => PixelFormat::Rgbx, + WireFormat::Bgra => PixelFormat::Bgra, + WireFormat::Rgba => PixelFormat::Rgba, + WireFormat::Rgb => PixelFormat::Rgb, + WireFormat::Bgr => PixelFormat::Bgr, + WireFormat::Rgb10a2 => PixelFormat::Rgb10a2, + WireFormat::Nv12 => PixelFormat::Nv12, + WireFormat::P010 => PixelFormat::P010, + WireFormat::Yuv444 => PixelFormat::Yuv444, + WireFormat::X2Rgb10 => PixelFormat::X2Rgb10, + WireFormat::X2Bgr10 => PixelFormat::X2Bgr10, + } + } +} + +/// [`pf_frame::CursorOverlay`] minus its pixels — cursor-as-metadata, the way the CSC consumes it. +/// `upload` is the pixel channel: `Some(len)` means a fresh memfd carrying `len` bytes of straight +/// -alpha RGBA rides with this frame (the bitmap `serial` changed); `None` means "reuse the bitmap +/// you cached for `serial`", which is every frame of a pointer that is merely moving. +#[derive(Serialize, Deserialize, Debug, Clone, PartialEq, Eq)] +pub(crate) struct WireCursor { + pub x: i32, + pub y: i32, + pub w: u32, + pub h: u32, + pub serial: u64, + pub hot_x: u32, + pub hot_y: u32, + pub visible: bool, + pub upload: Option, +} + +/// host → worker. Every variant has exactly one reply. +#[derive(Serialize, Deserialize, Debug, PartialEq)] +pub(crate) enum ToWorker { + /// Open the encoder. Answered with [`FromWorker::Ready`] (which carries the AU memfd) or + /// [`FromWorker::InitErr`]. + /// + /// `priority_intent` is the raw `PYROWAVE_QUEUE_PRIORITY` value as the HOST resolved it + /// (`None` = unset ⇒ the default REALTIME→HIGH ladder). Forwarded **explicitly** rather than + /// read from the worker's environment, and the worker strips the variable from its own env + /// before opening, so one knob cannot mean two things across the process boundary. + Hello { + proto: u32, + workspace_version: String, + /// The host's `PUNKTFUNK_RENDER_NODE` (`None` = unset). Log-only, exactly as it is + /// in-process: the device selection deliberately ignores every node anchor (see + /// `pyrowave.rs::select_physical_device` — two "fixes" were withdrawn). Carried so a + /// wrong-device field report shows the host's anchor beside the worker's pick. + drm_node: Option, + width: u32, + height: u32, + fps: u32, + bitrate_bps: u64, + chroma444: bool, + priority_intent: Option, + }, + /// Encode one frame. The dmabuf fd rides as `SCM_RIGHTS` only on first sight of `key` + /// (`has_fd`); a cursor upload, when present, is the fd AFTER it. Answered with + /// [`FromWorker::Au`], [`FromWorker::NeedFd`] or [`FromWorker::EncodeErr`]. + Frame { + key: u64, + has_fd: bool, + fourcc: u32, + modifier: u64, + offset: u32, + stride: u32, + plane1: Option<(u32, u32)>, + width: u32, + height: u32, + pts_ns: u64, + format: WireFormat, + cursor: Option, + }, + /// `Encoder::set_wire_chunking` — the datagram-aligned packetization boundary (plan §4.4). + /// This has to cross: it changes the AU BYTES (the windowed `build_au` framing) and the rate + /// budget, not just how the host hands them out. The streamed-AU *cutting* stays host-side. + SetWireChunking { shard_payload: usize }, + /// `Encoder::reconfigure_bitrate` — an in-place rate retarget. + Reconfigure { bitrate_bps: u64 }, + /// `Encoder::reset` — the stall watchdog's in-place rebuild, run INSIDE the worker so the + /// priority-elevated device survives it (see [`super::pyrowave_remote::RemotePyroWave::reset`] + /// for why this is a message and not a respawn). + Reset, +} + +/// worker → host. +#[derive(Serialize, Deserialize, Debug, PartialEq)] +pub(crate) enum FromWorker { + /// The encoder is open. Carries the AU memfd as its single `SCM_RIGHTS` descriptor. + /// + /// ⚠ `proto` and `workspace_version` are the first two fields and must never be renamed: they + /// are how a version-skewed pair diagnoses itself instead of failing obscurely. + Ready { + proto: u32, + workspace_version: String, + priority: PriorityOutcome, + device: String, + /// The chroma the encoder REALLY opened, and whether it blends the cursor — i.e. + /// `EncoderCaps` as only the opened encoder knows it. The proxy must not guess: a + /// hardcoded default mis-reports a 4:4:4 open and fires the session glue's spurious + /// "chroma disagrees with the negotiated Welcome" warn. + chroma444: bool, + blends_cursor: bool, + }, + /// The open failed (no Vulkan 1.3 device, missing features, …) — an ANSWER, not a crash: the + /// host falls back to the in-process encoder, which will fail the same way if the cause is + /// real and succeed if the cause was the worker's own environment. + InitErr { message: String }, + /// One access unit, complete, at offset 0 of the AU memfd. **Doubles as the buffer-release + /// signal**: it maps 1:1 onto `Encoder::submit`'s lifetime contract (the caller already holds + /// the frame alive until its AU comes back from `poll`), so the host loop needs no change. + Au { + key: u64, + len: usize, + pts_ns: u64, + keyframe: bool, + chunk_aligned: bool, + encode_us: u32, + }, + /// No cached fd for this `key` (evicted, or the caches diverged) — the host forgets its + /// "already sent" note and retries the frame once, with the fd. + NeedFd, + /// This frame failed but the worker is alive. + EncodeErr { message: String }, + /// Reply to [`ToWorker::SetWireChunking`] / [`ToWorker::Reconfigure`] / [`ToWorker::Reset`]. + Ack { ok: bool }, +} + +// --------------------------------------------------------------------------- +// Framing helpers — EINTR, and the deadline it must not defeat +// --------------------------------------------------------------------------- + +/// [`ipc::recv_fds`] that survives a signal. +/// +/// With `SO_RCVTIMEO` armed the kernel returns **EINTR**, not `ERESTARTSYS`, so `SA_RESTART` does +/// not save the caller — any signal delivered to a thread blocked in `recv` surfaces as an error. +/// pf-zerocopy's importer maps *any* recv error to "the worker died", which is right for a +/// once-per-capture handshake and wrong for a per-frame AU: one stray signal would drop a healthy +/// session to the in-process fallback. So retry here. +/// +/// The retry re-arms with the REMAINING budget rather than the full one — a signal arriving every +/// 100 ms would otherwise reset the clock forever and a real hang would never time out. +/// `budget = None` means "block until the host speaks or closes" (the worker's own serve loop). +pub(crate) fn recv_eintr( + sock: BorrowedFd, + buf: &mut Vec, + budget: Option, +) -> io::Result<(T, Vec)> { + let deadline = budget.map(|d| Instant::now() + d); + loop { + if let Some(deadline) = deadline { + let left = deadline.saturating_duration_since(Instant::now()); + if left.is_zero() { + return Err(io::Error::new( + io::ErrorKind::TimedOut, + "encode worker did not answer within its budget", + )); + } + ipc::set_recv_timeout(sock, Some(left))?; + } + match ipc::recv_fds::(sock, buf) { + Err(e) if e.kind() == io::ErrorKind::Interrupted => continue, + other => return other, + } + } +} + +/// [`ipc::send_fds`] that survives a signal. A small body on a socket whose peer is actively +/// reading does not block, so this normally retries never; it exists so that "normally" is not +/// load-bearing. +pub(crate) fn send_eintr( + sock: BorrowedFd, + msg: &T, + fds: &[BorrowedFd], +) -> io::Result<()> { + loop { + match ipc::send_fds(sock, msg, fds) { + Err(e) if e.kind() == io::ErrorKind::Interrupted => continue, + other => return other, + } + } +} + +/// An anonymous RAM-backed file for the bulk channels. Grows on `pwrite`, so callers never size it. +fn memfd(name: &CStr) -> io::Result { + // SAFETY: `memfd_create` reads a NUL-terminated name (a live `CStr` for the duration of the + // call) and returns a fresh descriptor or -1; it retains no pointer. The result is checked + // before use, and the returned fd is owned by nobody else, so `File::from_raw_fd` takes sole + // ownership and closes it exactly once. + let fd = unsafe { libc::memfd_create(name.as_ptr(), libc::MFD_CLOEXEC) }; + if fd < 0 { + return Err(io::Error::last_os_error()); + } + // SAFETY: `fd` is the fresh, valid descriptor just created and checked above. + Ok(unsafe { File::from_raw_fd(fd) }) +} + +/// Build the memfd carrying one cursor bitmap, clamped to what the blend can actually sample. +pub(crate) fn cursor_upload(rgba: &[u8]) -> io::Result<(File, usize)> { + let n = rgba.len().min(CURSOR_UPLOAD_MAX); + let f = memfd(c"pf-encode-cursor")?; + f.write_all_at(&rgba[..n], 0)?; + Ok((f, n)) +} + +// --------------------------------------------------------------------------- +// The worker half +// --------------------------------------------------------------------------- + +/// `punktfunk-encode-worker` entry point. `args` are the process's own arguments after argv[0] +/// (`--fd N`, default 3 — the socket end the spawning host `dup2`'d in). +pub fn run_from_args(args: &[String]) -> Result<()> { + // Core dumps ON, and FIRST — the opposite of the host's posture, deliberately. `PR_SET_DUMPABLE` + // is cleared by the kernel whenever a process gains a file capability, which also suppresses + // core dumps and makes `/proc//environ` unreadable. This process fronts nothing (no + // Wayland, no D-Bus, no network), so nothing is protected by that suppression and a crash in a + // GPU driver is exactly what we want a core for. It does NOT make us identifiable to KWin — + // the 0.26.0-1 matrix measured that dumpable is not the gate, the PERMITTED set is — and it + // does not need to: this process never speaks Wayland. + // SAFETY: `prctl(PR_SET_DUMPABLE, 1)` takes integers by value, touches no Rust memory and + // affects only this process. + unsafe { + libc::prctl(libc::PR_SET_DUMPABLE, 1); + } + // The host execs us via a pinned `/proc/self/fd/`, so the kernel derives our comm from a + // meaningless fd number. Rename so `top`/`pkill`/a coredump path see the worker. + // SAFETY: `PR_SET_NAME` copies at most 16 bytes from the given pointer; the C-string literal is + // valid, NUL-terminated and short enough, and no pointer is retained past the call. + unsafe { + libc::prctl(libc::PR_SET_NAME, c"pf-encode-wk".as_ptr()); + } + sanitize_env(); + // Real teeth, and the second half of what the capability buys: `setpriority` is a silent no-op + // without `CAP_SYS_NICE`/`RLIMIT_NICE`, which is why the in-host encode thread's nice(-10) has + // never actually applied on a packaged Linux host. Here it applies. The worker is + // single-threaded, so this IS the encode thread. + pf_frame::thread_qos::boost_thread_priority(true); + + let fd: i32 = args + .iter() + .skip_while(|a| *a != "--fd") + .nth(1) + .map(|s| s.parse()) + .transpose() + .context("parse --fd")? + .unwrap_or(3); + // Refuse anything that cannot be the spawning host's socket: a negative fd is UB inside + // `OwnedFd` (its niche), and 0–2 would make the worker close one of its own stdio streams on + // exit. Then confirm the number really holds a socket — this binary is installed and runnable + // by hand, and adopting an arbitrary inherited fd would close it behind its real owner. + anyhow::ensure!(fd >= 3, "--fd must be >= 3 (got {fd})"); + // SAFETY: `libc::stat` is plain-old-data for which all-zero is a valid value, so `mem::zeroed` + // is a sound initializer; `fstat` writes into the live, correctly-sized `&mut st` and only + // reads `fd`. `st_mode` is read only after the return value is checked. + let is_socket = unsafe { + let mut st: libc::stat = std::mem::zeroed(); + libc::fstat(fd, &mut st) == 0 && (st.st_mode & libc::S_IFMT) == libc::S_IFSOCK + }; + anyhow::ensure!( + is_socket, + "--fd {fd} is not an open socket (this binary is spawned by punktfunk-host, not run by hand)" + ); + // SAFETY: the spawning host `dup2`'d its socketpair end onto exactly this fd number before + // exec (the worker's contract, just verified to be an open socket ≥ 3) and nothing else in + // this fresh process owns it, so `OwnedFd` takes sole ownership and closes it once at exit. + let sock = unsafe { OwnedFd::from_raw_fd(fd) }; + run(sock) +} + +/// Drop the environment variables this process must not act on. +/// +/// Deliberately a DENYLIST, not an allowlist. The obvious "clear everything but a handful of +/// names" is wrong here: the Vulkan loader discovers its ICDs through the environment +/// (`VK_ICD_FILENAMES`/`VK_DRIVER_FILES`/`XDG_DATA_DIRS`), so a strict allowlist would leave the +/// worker with no GPU exactly on NixOS — the one channel where this worker's env override is +/// load-bearing. What must go is the punktfunk state that would make one knob mean two things: +/// the priority intent (it arrives explicitly in `Hello`) and the worker path itself (nothing here +/// spawns a worker, and a stale value in a core dump is just noise). +fn sanitize_env() { + // Single-threaded — this runs before anything in this process creates a thread, which is the + // one situation where mutating the environment is sound (the `getenv` race the house rule + // about `set_var` is about needs a second thread). + for k in ["PYROWAVE_QUEUE_PRIORITY", "PUNKTFUNK_ENCODE_WORKER"] { + std::env::remove_var(k); + } +} + +/// Handshake, then serve until the host goes away. +fn run(sock: OwnedFd) -> Result<()> { + let mut buf = Vec::new(); + // No timeout on the worker's own receives: the host owns the clock (it arms `SO_RCVTIMEO` on + // its end), and a worker that gave up on its own would look exactly like a crash. + let (hello, _) = recv_eintr::(sock.as_fd(), &mut buf, None).context("recv Hello")?; + let ToWorker::Hello { + proto, + workspace_version, + drm_node, + width, + height, + fps, + bitrate_bps, + chroma444, + priority_intent, + } = hello + else { + anyhow::bail!("first message was not Hello"); + }; + if proto != PROTO_VERSION || workspace_version != WORKSPACE_VERSION { + // Answer, don't crash: the host prints one warn naming both builds and encodes in-process. + let _ = send_eintr( + sock.as_fd(), + &FromWorker::InitErr { + message: format!( + "version skew: worker proto {PROTO_VERSION} v{WORKSPACE_VERSION}, \ + host proto {proto} v{workspace_version}" + ), + }, + &[], + ); + return Ok(()); + } + + let enc = match super::pyrowave::PyroWaveEncoder::open_in_worker( + width, + height, + fps, + bitrate_bps, + chroma444, + priority_intent.as_deref(), + ) { + Ok(e) => e, + Err(e) => { + let _ = send_eintr( + sock.as_fd(), + &FromWorker::InitErr { + message: format!("{e:#}"), + }, + &[], + ); + return Ok(()); + } + }; + let au_buf = memfd(c"pf-encode-au").context("create the AU return buffer")?; + let caps = crate::Encoder::caps(&enc); + let ready = FromWorker::Ready { + proto: PROTO_VERSION, + workspace_version: WORKSPACE_VERSION.to_string(), + priority: enc.priority_outcome(), + device: enc.device_name().to_string(), + chroma444: caps.chroma_444, + blends_cursor: caps.blends_cursor, + }; + send_eintr(sock.as_fd(), &ready, &[au_buf.as_fd()]).context("send Ready")?; + tracing::info!( + pid = std::process::id(), + device = %enc.device_name(), + priority = ?enc.priority_outcome(), + host_render_node = ?drm_node, + "punktfunk-encode-worker ready" + ); + serve(&sock, enc, &au_buf) +} + +/// The request loop. `Ok(())` on host EOF (normal end-of-life — the host dropped its proxy); +/// any other socket error propagates and the process exits, which the host reads as a death, +/// because it is one. +fn serve(sock: &OwnedFd, mut enc: super::pyrowave::PyroWaveEncoder, au_buf: &File) -> Result<()> { + use crate::Encoder as _; + let mut buf = Vec::new(); + let mut fds: HashMap = HashMap::new(); + // Insertion order, for the eviction the cap implies. + let mut fd_order: VecDeque = VecDeque::new(); + // The cursor bitmap the host last uploaded, by `serial` — a moving pointer re-sends only its + // position, exactly like the in-process path re-uses its uploaded texture. + let mut cursor_rgba: Option<(u64, Arc>)> = None; + loop { + let (msg, got) = match recv_eintr::(sock.as_fd(), &mut buf, None) { + Ok(v) => v, + Err(e) if e.kind() == io::ErrorKind::UnexpectedEof => return Ok(()), + Err(e) => return Err(e).context("worker recv"), + }; + let reply = match msg { + ToWorker::Hello { .. } => FromWorker::EncodeErr { + message: "duplicate Hello".into(), + }, + ToWorker::SetWireChunking { shard_payload } => { + enc.set_wire_chunking(shard_payload); + FromWorker::Ack { ok: true } + } + ToWorker::Reconfigure { bitrate_bps } => FromWorker::Ack { + ok: enc.reconfigure_bitrate(bitrate_bps), + }, + ToWorker::Reset => FromWorker::Ack { ok: enc.reset() }, + ToWorker::Frame { + key, + has_fd, + fourcc, + modifier, + offset, + stride, + plane1, + width, + height, + pts_ns, + format, + cursor, + } => { + // Descriptor order is the sender's: the dmabuf (iff `has_fd`), then the cursor + // upload (iff the bitmap changed). Taken before any early return so an unexpected + // extra descriptor is closed with the `Vec` rather than leaked. + let mut got = got.into_iter(); + let dmabuf = if has_fd { got.next() } else { None }; + let cursor_fd = cursor.as_ref().and_then(|c| c.upload.map(|_| got.next())); + if let Some(fd) = dmabuf { + if fds.insert(key, fd).is_none() { + fd_order.push_back(key); + } + while fd_order.len() > FD_CACHE_CAP { + if let Some(old) = fd_order.pop_front() { + fds.remove(&old); + } + } + } + match encode_one( + &mut enc, + au_buf, + &fds, + &mut cursor_rgba, + FrameReq { + key, + fourcc, + modifier, + offset, + stride, + plane1, + width, + height, + pts_ns, + format, + cursor, + }, + cursor_fd.flatten(), + ) { + Ok(reply) => reply, + Err(e) => FromWorker::EncodeErr { + message: format!("{e:#}"), + }, + } + } + }; + match send_eintr(sock.as_fd(), &reply, &[]) { + Ok(()) => {} + // The host vanished between our recv and our send — the same end-of-life as EOF. + Err(e) if e.kind() == io::ErrorKind::BrokenPipe => return Ok(()), + Err(e) => return Err(e).context("worker send"), + } + } +} + +/// [`ToWorker::Frame`] minus the descriptors, so [`encode_one`] takes one argument per concept. +struct FrameReq { + key: u64, + fourcc: u32, + modifier: u64, + offset: u32, + stride: u32, + plane1: Option<(u32, u32)>, + width: u32, + height: u32, + pts_ns: u64, + format: WireFormat, + cursor: Option, +} + +/// Rebuild the `CapturedFrame`, encode it synchronously, and write the AU into `au_buf`. +fn encode_one( + enc: &mut super::pyrowave::PyroWaveEncoder, + au_buf: &File, + fds: &HashMap, + cursor_rgba: &mut Option<(u64, Arc>)>, + req: FrameReq, + cursor_fd: Option, +) -> Result { + use crate::Encoder as _; + let Some(cached) = fds.get(&req.key) else { + return Ok(FromWorker::NeedFd); + }; + // A dup per frame, not a borrow: `DmabufFrame` owns its fd (the encoder's import path dups it + // again for Vulkan and drops the rest), while the cache must keep holding the original so the + // steady state passes no descriptors at all. One `dup`/`close` pair per frame is µs. + let fd = cached.try_clone().context("dup the cached dmabuf fd")?; + + let cursor = match req.cursor { + Some(c) => { + match (c.upload, cursor_fd) { + (Some(len), Some(f)) => { + let mut px = vec![0u8; len]; + File::from(f) + .read_exact_at(&mut px, 0) + .context("read the cursor upload")?; + *cursor_rgba = Some((c.serial, Arc::new(px))); + } + // An announced upload whose descriptor did not arrive. Rare (it takes a kernel + // refusal of the `SCM_RIGHTS`), and the reason it is an ERROR rather than a + // shrug: the host marks the serial "sent" on a successful AU, so blending + // nothing here would leave the pointer INVISIBLE for the rest of that bitmap's + // life, silently. Failing the frame drops the session onto the in-process + // encoder instead, which is a rung with a warning attached. + (Some(_), None) => anyhow::bail!("cursor upload announced but no descriptor came"), + (None, _) => {} + } + // Likewise a serial we hold no pixels for: the host only omits the upload for a serial + // it has seen acknowledged, so a miss is a desync, not a frame to guess at. + let Some(rgba) = cursor_rgba + .as_ref() + .filter(|(serial, _)| *serial == c.serial) + .map(|(_, px)| px.clone()) + else { + anyhow::bail!("no cursor bitmap cached for serial {}", c.serial); + }; + Some(CursorOverlay { + x: c.x, + y: c.y, + w: c.w, + h: c.h, + rgba, + serial: c.serial, + hot_x: c.hot_x, + hot_y: c.hot_y, + visible: c.visible, + }) + } + None => None, + }; + let frame = CapturedFrame { + width: req.width, + height: req.height, + pts_ns: req.pts_ns, + format: req.format.into(), + payload: FramePayload::Dmabuf(DmabufFrame { + fd, + fourcc: req.fourcc, + modifier: req.modifier, + plane1: req.plane1, + offset: req.offset, + stride: req.stride, + }), + cursor, + }; + // submit→poll in one breath: this backend's encode is synchronous at depth 1, so the AU is + // ready when `poll` returns and `frame` (with its fd) is alive across both halves — the + // trait's lifetime contract, honored on this side of the socket too. + let t0 = Instant::now(); + enc.submit(&frame)?; + let Some(au) = enc.poll()? else { + anyhow::bail!("encoder returned no AU for a submitted frame"); + }; + let encode_us = t0.elapsed().as_micros() as u32; + au_buf + .write_all_at(&au.data, 0) + .context("write the AU into the return buffer")?; + Ok(FromWorker::Au { + key: req.key, + len: au.data.len(), + pts_ns: au.pts_ns, + keyframe: au.keyframe, + chunk_aligned: au.chunk_aligned, + encode_us, + }) +} + +#[cfg(test)] +mod tests { + use super::*; + use std::os::fd::AsFd; + + fn hello() -> ToWorker { + ToWorker::Hello { + proto: PROTO_VERSION, + workspace_version: WORKSPACE_VERSION.to_string(), + drm_node: Some("/dev/dri/renderD128".into()), + width: 3840, + height: 2160, + fps: 60, + bitrate_bps: 400_000_000, + chroma444: true, + priority_intent: Some("realtime".into()), + } + } + + /// The vocabulary survives the wire in both directions, descriptors included. (The framing — + /// EOF, timeouts, the descriptor cap — is pf-zerocopy's `ipc` tests' job; this pins the + /// message types and the fd ORDER the frame path depends on.) + #[test] + fn proto_round_trip_both_directions() { + let (a, b) = ipc::socketpair_seqpacket().unwrap(); + let mut buf = Vec::new(); + ipc::send(a.as_fd(), &hello(), None).unwrap(); + let (got, fds) = ipc::recv_fds::(b.as_fd(), &mut buf).unwrap(); + assert_eq!(got, hello()); + assert!(fds.is_empty()); + + let frame = ToWorker::Frame { + key: 0xdead_beef, + has_fd: true, + fourcc: 0x3432_5258, + modifier: 0x0300_0000_0000_1234, + offset: 0, + stride: 3840 * 4, + plane1: None, + width: 3840, + height: 2160, + pts_ns: 1_234_567_890, + format: WireFormat::Bgrx, + cursor: Some(WireCursor { + x: 10, + y: 20, + w: 32, + h: 32, + serial: 7, + hot_x: 1, + hot_y: 2, + visible: true, + upload: Some(32 * 32 * 4), + }), + }; + // Two descriptors, in the order the receiver destructures them: dmabuf, then cursor. + let (dma, cur) = (memfd(c"t-dma").unwrap(), memfd(c"t-cur").unwrap()); + ipc::send_fds(a.as_fd(), &frame, &[dma.as_fd(), cur.as_fd()]).unwrap(); + let (got, fds) = ipc::recv_fds::(b.as_fd(), &mut buf).unwrap(); + assert_eq!(got, frame); + assert_eq!(fds.len(), 2); + + let ready = FromWorker::Ready { + proto: PROTO_VERSION, + workspace_version: WORKSPACE_VERSION.to_string(), + priority: PriorityOutcome::Granted(GrantedClass::Realtime), + device: "NVIDIA GeForce RTX 5070 Ti".into(), + chroma444: true, + blends_cursor: true, + }; + ipc::send(b.as_fd(), &ready, Some(dma.as_fd())).unwrap(); + let (got, fd) = ipc::recv::(a.as_fd(), &mut buf).unwrap(); + assert_eq!(got, ready); + assert!(fd.is_some(), "Ready carries the AU return buffer"); + + for reply in [ + FromWorker::Au { + key: 1, + len: 830_000, + pts_ns: 5, + keyframe: true, + chunk_aligned: true, + encode_us: 4400, + }, + FromWorker::NeedFd, + FromWorker::Ack { ok: true }, + FromWorker::EncodeErr { + message: "boom".into(), + }, + ] { + ipc::send(b.as_fd(), &reply, None).unwrap(); + let (got, _) = ipc::recv::(a.as_fd(), &mut buf).unwrap(); + assert_eq!(got, reply); + } + } + + /// An AU never rides in the JSON body, and this is why: the smallest per-frame budget the + /// encoder will ever use is already `MAX_MSG`, and serde_json renders a byte as up to four + /// characters. Pinned as a test so nobody "simplifies" the memfd away. + #[test] + fn an_inline_au_would_not_fit_a_message() { + // 1080p60 at a modest 40 Mb/s — well inside the shipped range. + let au = vec![0xABu8; 40_000_000 / (8 * 60)]; + let body = serde_json::to_vec(&au).unwrap(); + assert!( + body.len() > ipc::MAX_MSG, + "a {}-byte AU serialized to {} bytes, which would (wrongly) fit MAX_MSG {}", + au.len(), + body.len(), + ipc::MAX_MSG + ); + } + + /// A body over [`ipc::MAX_MSG`] is refused at the sender rather than truncated on the wire — + /// the property the memfd channel exists to respect. + #[test] + fn oversized_messages_are_refused_not_truncated() { + let (a, _b) = ipc::socketpair_seqpacket().unwrap(); + let ToWorker::Hello { + proto, + drm_node, + width, + height, + fps, + bitrate_bps, + chroma444, + priority_intent, + .. + } = hello() + else { + unreachable!("hello() builds a Hello"); + }; + let huge = ToWorker::Hello { + proto, + // Over `MAX_MSG` on its own — enum variants take no functional-update syntax, so the + // rest is destructured above rather than `..hello()`. + workspace_version: "x".repeat(ipc::MAX_MSG), + drm_node, + width, + height, + fps, + bitrate_bps, + chroma444, + priority_intent, + }; + let err = ipc::send(a.as_fd(), &huge, None).unwrap_err(); + assert_eq!(err.kind(), io::ErrorKind::InvalidData); + } + + /// Every `PixelFormat` maps to a wire tag and back unchanged. The `match`es are exhaustive, so + /// a new capture format is a compile error; this catches a mis-typed ARM in either direction. + #[test] + fn pixel_formats_round_trip() { + for f in [ + PixelFormat::Bgrx, + PixelFormat::Rgbx, + PixelFormat::Bgra, + PixelFormat::Rgba, + PixelFormat::Rgb, + PixelFormat::Bgr, + PixelFormat::Rgb10a2, + PixelFormat::Nv12, + PixelFormat::P010, + PixelFormat::Yuv444, + PixelFormat::X2Rgb10, + PixelFormat::X2Bgr10, + ] { + assert_eq!(PixelFormat::from(WireFormat::from(f)), f); + } + } + + /// The bulk channel: a memfd written by one holder of the descriptor is readable at offset 0 + /// by another, and it grows on write with no explicit sizing. That is the whole mechanism the + /// AU return depends on. + #[test] + fn memfd_round_trips_bytes_across_a_descriptor() { + let f = memfd(c"pf-encode-test").unwrap(); + let au = vec![0x5Au8; 900_000]; + f.write_all_at(&au, 0).unwrap(); + let (a, b) = ipc::socketpair_seqpacket().unwrap(); + let mut buf = Vec::new(); + ipc::send(a.as_fd(), &FromWorker::NeedFd, Some(f.as_fd())).unwrap(); + let (_, fd) = ipc::recv::(b.as_fd(), &mut buf).unwrap(); + let mut back = vec![0u8; au.len()]; + File::from(fd.unwrap()).read_exact_at(&mut back, 0).unwrap(); + assert_eq!(back, au); + } + + /// A cursor bitmap is clamped to what the 256×256 blend texture can sample — a larger one is + /// truncated, exactly as `prep_cursor`'s `bytes.min(rgba.len())` copy already truncates it. + #[test] + fn cursor_upload_clamps_to_the_blend_texture() { + let (_, n) = cursor_upload(&vec![0u8; CURSOR_UPLOAD_MAX * 4]).unwrap(); + assert_eq!(n, CURSOR_UPLOAD_MAX); + let (_, n) = cursor_upload(&vec![0u8; 64 * 64 * 4]).unwrap(); + assert_eq!(n, 64 * 64 * 4); + } + + /// EINTR must not read as a dead worker. A `SIGURG` (default-ignored, so the test process + /// survives it) delivered to a thread parked in `recv` with `SO_RCVTIMEO` armed returns EINTR + /// — `SA_RESTART` does not apply to a timeout-armed socket — and the retry must swallow it and + /// still deliver the message that arrives afterwards. + #[test] + fn recv_survives_a_signal() { + let (a, b) = ipc::socketpair_seqpacket().unwrap(); + let b = std::sync::Arc::new(b); + let waiter = { + let b = b.clone(); + std::thread::spawn(move || { + let mut buf = Vec::new(); + recv_eintr::(b.as_fd(), &mut buf, Some(Duration::from_secs(10))) + }) + }; + // Give the thread time to park in `recvmsg`, then interrupt it repeatedly while the + // message is still not there. + std::thread::sleep(Duration::from_millis(50)); + for _ in 0..5 { + // SAFETY: `pthread_kill` takes the live thread's id by value and a signal number; + // SIGURG's default disposition is "ignore", so delivery cannot kill the process. + unsafe { + libc::pthread_kill( + std::os::unix::thread::JoinHandleExt::as_pthread_t(&waiter), + libc::SIGURG, + ); + } + std::thread::sleep(Duration::from_millis(10)); + } + ipc::send(a.as_fd(), &FromWorker::Ack { ok: true }, None).unwrap(); + let (got, _) = waiter.join().unwrap().expect("EINTR must not surface"); + assert_eq!(got, FromWorker::Ack { ok: true }); + } + + /// …and the retry must not defeat the deadline: a socket nobody ever writes to still times + /// out, signals or no signals. + #[test] + fn recv_still_times_out() { + let (a, _b) = ipc::socketpair_seqpacket().unwrap(); + let mut buf = Vec::new(); + let err = recv_eintr::(a.as_fd(), &mut buf, Some(Duration::from_millis(80))) + .unwrap_err(); + assert!( + matches!( + err.kind(), + io::ErrorKind::WouldBlock | io::ErrorKind::TimedOut + ), + "unexpected error kind: {err:?}" + ); + } +} diff --git a/crates/pf-encode/src/lib.rs b/crates/pf-encode/src/lib.rs index 984bddfd..adabe2e0 100644 --- a/crates/pf-encode/src/lib.rs +++ b/crates/pf-encode/src/lib.rs @@ -358,8 +358,17 @@ fn open_video_backend_linux( if codec == Codec::PyroWave { #[cfg(feature = "pyrowave")] { - return pyrowave::PyroWaveEncoder::open(width, height, fps, bitrate_bps, chroma) - .map(|e| (Box::new(e) as Box, "pyrowave")); + // Through the worker seam, not straight at the encoder: PyroWave's GPU-priority lever + // needs CAP_SYS_NICE, which only `punktfunk-encode-worker` may carry. Every rung of + // that ladder ends at this exact in-process open — see `pyrowave_remote`. + return pyrowave_remote::open_preferring_worker( + width, + height, + fps, + bitrate_bps, + chroma, + ) + .map(|e| (e, "pyrowave")); } #[cfg(not(feature = "pyrowave"))] anyhow::bail!( @@ -517,14 +526,16 @@ fn open_video_backend_linux( // The lab override forces the wavelet stream onto a session negotiated for // another codec — that session's chroma may be HEVC-4:4:4, which the // pyrowave encoder doesn't do yet, so pin the override to 4:2:0. - pyrowave::PyroWaveEncoder::open( + // Same worker seam as the negotiated arm above: the lab override is where the + // A/B is measured, so it must not be the one path that skips the worker. + pyrowave_remote::open_preferring_worker( width, height, fps, bitrate_bps, ChromaFormat::Yuv420, ) - .map(|e| (Box::new(e) as Box, "pyrowave")) + .map(|e| (e, "pyrowave")) } #[cfg(not(feature = "pyrowave"))] { @@ -2035,6 +2046,18 @@ mod vk_util; #[cfg(all(target_os = "linux", feature = "pyrowave"))] #[path = "enc/linux/pyrowave.rs"] mod pyrowave; +// `punktfunk-encode-worker` (design/gpu-priority-capability-worker.md): the capability-carrying +// process that owns the priority-elevated PyroWave device, because `punktfunk-host` may never hold +// a file capability (0.26.0-1 — a capped host is unidentifiable to KWin and loses desktop +// streaming). `worker` is the vocabulary plus the worker's own run loop, and it is `pub` for +// exactly one caller: the ~30-line `main` of the separate `punktfunk-encode-worker` binary. +// `pyrowave_remote` is the host-side proxy and its fallback ladder. +#[cfg(all(target_os = "linux", feature = "pyrowave"))] +#[path = "enc/linux/pyrowave_remote.rs"] +mod pyrowave_remote; +#[cfg(all(target_os = "linux", feature = "pyrowave"))] +#[path = "enc/linux/worker.rs"] +pub mod worker; // The Windows PyroWave encoder — NV12 zero-copy D3D11→Vulkan via pyrowave's own compat device // (design/pyrowave-windows-host-zerocopy.md). Same module name as the Linux one (per-platform // `#[path]`, mutually-exclusive cfg) so `crate::pyrowave::*` is flat on both. diff --git a/crates/punktfunk-encode-worker/Cargo.toml b/crates/punktfunk-encode-worker/Cargo.toml new file mode 100644 index 00000000..56d03943 --- /dev/null +++ b/crates/punktfunk-encode-worker/Cargo.toml @@ -0,0 +1,39 @@ +# The capability-carrying PyroWave encode worker (design/gpu-priority-capability-worker.md). +# +# 🛑 This is a SEPARATE BINARY on purpose, and it must stay one. It is the only Punktfunk file that +# may carry `cap_sys_nice=ep`; `punktfunk-host` carries no capability on any channel, ever, because +# a capped process cannot be identified by KWin (`cap_ptrace_access_check` refuses +# `/proc//exe` to a reader whose effective set is not a superset of the target's PERMITTED +# set) and therefore never gets `zkde_screencast_unstable_v1` — that was 0.26.0-1, and it killed +# every KDE desktop session. A hardlink to the host, or a hidden host subcommand, shares the inode +# and so shares the capability: same incident, silently. Never do either. +# +# All the logic lives in `pf-encode::worker`; this crate is the file, not the code. +[package] +name = "punktfunk-encode-worker" +version.workspace = true +edition = "2021" +rust-version.workspace = true +license = "MIT OR Apache-2.0" +description = "punktfunk PyroWave encode worker: owns the priority-elevated Vulkan device so the host never carries a capability." +publish = false + +[[bin]] +name = "punktfunk-encode-worker" +path = "src/main.rs" + +[dependencies] +# `features = ["pyrowave"]` unconditionally rather than through a feature of this crate: packaging +# builds this with `cargo build -p punktfunk-encode-worker` and default features, and a worker +# built without the codec would be a binary that hands every session straight back to the +# in-process fallback while still carrying the capability. (No extra cost to the workspace: +# punktfunk-host already has `default = ["pyrowave"]`, so a workspace build resolves it anyway.) +pf-encode = { path = "../pf-encode", features = ["pyrowave"] } +tracing = "0.1" +# The worker's stderr is inherited from the host that spawned it, so its lines land in the same +# journal — but it is a fresh process with no subscriber of its own, and without one the +# device-pick and priority-ladder lines would go nowhere. +tracing-subscriber = { version = "0.3", features = ["env-filter"] } + +[lints] +workspace = true diff --git a/crates/punktfunk-encode-worker/src/main.rs b/crates/punktfunk-encode-worker/src/main.rs new file mode 100644 index 00000000..16c5312a --- /dev/null +++ b/crates/punktfunk-encode-worker/src/main.rs @@ -0,0 +1,43 @@ +//! `punktfunk-encode-worker` — spawned by `punktfunk-host` for the duration of one PyroWave +//! session, never run by hand. It reads its socket from the inherited fd 3 and speaks only to the +//! parent that spawned it: no Wayland, no D-Bus, no network, no plugins. +//! +//! Everything it does lives in [`pf_encode::worker`]; this file exists so the capability has a +//! **file of its own** (see this crate's Cargo.toml for why that is not negotiable). + +fn main() -> std::process::ExitCode { + // Stderr, inherited from the host, so the worker's lines land in the host's journal next to + // the session that spawned it. `RUST_LOG` is inherited too, so raising the host's level + // raises the worker's. + let filter = + tracing_subscriber::EnvFilter::try_from_default_env().unwrap_or_else(|_| "info".into()); + tracing_subscriber::fmt() + .with_env_filter(filter) + .with_writer(std::io::stderr) + .init(); + + #[cfg(target_os = "linux")] + { + let args: Vec = std::env::args().skip(1).collect(); + match pf_encode::worker::run_from_args(&args) { + Ok(()) => std::process::ExitCode::SUCCESS, + Err(e) => { + // The host reads the dead socket long before it could read this, so the message is + // for a human running `journalctl` — say enough to place the failure. + tracing::error!(error = %format!("{e:#}"), "punktfunk-encode-worker exiting"); + std::process::ExitCode::FAILURE + } + } + } + // Linux-only by construction: the worker exists for `VK_KHR_global_priority` under + // `CAP_SYS_NICE`, and the Windows host raises its GPU scheduling priority through WDDM + // instead (`D3DKMTSetProcessSchedulingPriorityClass`). Packaging never installs this + // elsewhere; the arm exists so a workspace build stays green on every platform. + #[cfg(not(target_os = "linux"))] + { + tracing::error!( + "punktfunk-encode-worker is a Linux-only helper and has nothing to do here" + ); + std::process::ExitCode::FAILURE + } +}