diff --git a/crates/pf-capture/src/linux/mod.rs b/crates/pf-capture/src/linux/mod.rs index ca340e5c..e5e56880 100644 --- a/crates/pf-capture/src/linux/mod.rs +++ b/crates/pf-capture/src/linux/mod.rs @@ -1310,21 +1310,26 @@ mod pipewire { /// (which Mutter delivers as metadata-only "corrupted" buffers) still refresh the position. fn update_cursor_meta(cursor: &mut CursorState, spa_buf: *mut spa::sys::spa_buffer) { // SAFETY: `spa_buf` is the live buffer we still hold (dequeued, not yet requeued). - // `spa_buffer_find_meta_data` scans its metadata array for a `SPA_META_Cursor` of at least - // `size_of::()` bytes and returns a pointer into that buffer's metadata - // (or null), valid until requeue. The size argument matches the struct the result is cast to. - let cur = unsafe { - spa::sys::spa_buffer_find_meta_data( - spa_buf, - spa::sys::SPA_META_Cursor, - std::mem::size_of::(), - ) as *const spa::sys::spa_meta_cursor - }; - if cur.is_null() { + // `spa_buffer_find_meta` returns the `spa_meta` (type + byte `size` + `data` pointer) for + // `SPA_META_Cursor`, or null. We take `find_meta` rather than `find_meta_data` specifically + // to obtain the region's real `size`: the bitmap offset, pixel offset and stride read below + // are ALL producer-written, and without a bound against the actual region they drive + // out-of-bounds pointer arithmetic and an oversized `slice::from_raw_parts` — an OOB read + // that SIGSEGVs inside the PipeWire `.process` callback (a segfault `catch_unwind` cannot + // catch). Every offset below is validated against `region_size` with checked arithmetic, + // mirroring the fd-length guard the main frame path already applies to xdg-desktop-portal-wlr. + let meta = + unsafe { spa::sys::spa_buffer_find_meta(spa_buf, spa::sys::SPA_META_Cursor as u32) }; + if meta.is_null() { return; } - // SAFETY: `cur` is non-null and points to a `spa_meta_cursor` of at least its own size - // inside the held buffer (guaranteed by the size arg above), so every field read is in bounds. + // SAFETY: `meta` is non-null and points into the held buffer's metadata array. + let (region_size, data) = unsafe { ((*meta).size as usize, (*meta).data as *const u8) }; + if data.is_null() || region_size < std::mem::size_of::() { + return; + } + let cur = data as *const spa::sys::spa_meta_cursor; + // SAFETY: `region_size >= size_of::()` checked above, so every field is in bounds. let (id, pos_x, pos_y, hot_x, hot_y, bmp_off) = unsafe { ( (*cur).id, @@ -1347,13 +1352,18 @@ mod pipewire { // Position-only update — keep the cached bitmap. return; } - // SAFETY: `bitmap_offset` is a byte offset from `cur` to a `spa_meta_bitmap`, which the - // producer placed inside the same meta region it sized for this cursor (>= the size we - // requested). The resulting pointer is in bounds and aligned for `spa_meta_bitmap`. - let bmp = - unsafe { (cur as *const u8).add(bmp_off as usize) as *const spa::sys::spa_meta_bitmap }; - // SAFETY: `bmp` is the in-bounds, aligned `spa_meta_bitmap` pointer computed just above; the - // producer fully initialized this header, so reading its scalar fields is sound. + let bmp_off = bmp_off as usize; + // The `spa_meta_bitmap` header must fit entirely inside the region before we read it — + // `bitmap_offset` is producer-controlled and otherwise reads past the metadata. + match bmp_off.checked_add(std::mem::size_of::()) { + Some(end) if end <= region_size => {} + _ => return, + } + // SAFETY: `bmp_off + size_of::() <= region_size` (checked directly above), + // so the header is fully in bounds; the producer places it aligned as before. + let bmp = unsafe { data.add(bmp_off) as *const spa::sys::spa_meta_bitmap }; + // SAFETY: `bmp` is the in-bounds `spa_meta_bitmap` header validated just above; reading its + // scalar fields is sound. let (vfmt, bw, bh, stride, pix_off) = unsafe { ( (*bmp).format, @@ -1369,10 +1379,27 @@ mod pipewire { } let row = bw as usize * 4; let stride = if stride < row { row } else { stride }; - let span = stride * (bh as usize - 1) + row; - // SAFETY: the bitmap pixels live at `bmp + pix_off` for `span` bytes, within the - // producer-sized meta region. `span` is the exact extent the strided copy below reads. - let src = unsafe { std::slice::from_raw_parts((bmp as *const u8).add(pix_off), span) }; + // `span` is the exact byte extent the strided loop reads: `stride·(bh-1) + row`. Compute it + // with checked arithmetic (a producer stride near `i32::MAX` would otherwise overflow) and + // require the whole pixel block `[bmp_off + pix_off, +span)` to lie inside the region before + // fabricating the slice — this is the check whose absence made the read go out of bounds. + let span = match stride + .checked_mul(bh as usize - 1) + .and_then(|v| v.checked_add(row)) + { + Some(s) => s, + None => return, + }; + match bmp_off + .checked_add(pix_off) + .and_then(|v| v.checked_add(span)) + { + Some(end) if end <= region_size => {} + _ => return, + } + // SAFETY: `bmp_off + pix_off + span <= region_size` (checked directly above), so the slice + // is fully within the producer's meta region; `span` is exactly the strided loop's extent. + let src = unsafe { std::slice::from_raw_parts(data.add(bmp_off + pix_off), span) }; let mut rgba = vec![0u8; bw as usize * bh as usize * 4]; for y in 0..bh as usize { for x in 0..bw as usize { @@ -2164,36 +2191,37 @@ mod pipewire { } }) .process(|stream, ud| { - // PipeWire dispatches this from a C trampoline with no catch_unwind; a - // panic crossing that FFI boundary would abort the whole host. Contain it. + // Latest-frame-only (OBS pattern): Mutter delivers buffers in bursts and recycles its + // pool; an older queued buffer carries a STALE frame. Drain all queued buffers, requeue + // the older ones, keep only the newest. This dequeue/requeue runs OUTSIDE the + // `catch_unwind` below — they are non-panicking C FFI pointer ops, and `newest` is + // requeued exactly once AFTER the panic-containing region. Previously the whole thing was + // inside the catch, so a caught panic (in `update_cursor_meta`/`consume_frame`) stranded + // `newest` forever, permanently shrinking the stream's fixed pool until capture wedged. + // SAFETY: `stream` is the live stream PipeWire passes into this `.process` callback on the + // loop thread; `dequeue_raw_buffer` returns a stream-owned `*mut pw_buffer` or null + // (null-checked), single-threaded so no concurrent access. + let mut newest = unsafe { stream.dequeue_raw_buffer() }; + if newest.is_null() { + return; + } + let mut drained = 1u32; + loop { + // SAFETY: same stream/loop-thread contract; returns the next stream-owned buffer or null. + let next = unsafe { stream.dequeue_raw_buffer() }; + if next.is_null() { + break; + } + // SAFETY: `newest` was dequeued from this stream and not yet requeued; we immediately + // overwrite it, so the requeued pointer is never touched again. + unsafe { stream.queue_raw_buffer(newest) }; + newest = next; + drained += 1; + } + // PipeWire dispatches from a C trampoline with no catch_unwind; a panic crossing that FFI + // boundary would abort the whole host. Contain the inspect/consume work — the only Rust + // code here that can panic — and requeue `newest` unconditionally after it. let outcome = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| { - // Latest-frame-only (OBS pattern): Mutter delivers buffers in bursts and - // recycles its pool; an older queued buffer carries a STALE frame. Drain all - // queued buffers, requeue the older ones, keep only the newest. - // SAFETY: `stream` is the live stream PipeWire passes into this `.process` callback on - // the loop thread, where `pw_stream_dequeue_buffer` is the documented call. It returns - // a `*mut pw_buffer` owned by the stream (or null when the queue is drained), - // null-checked before any use. The loop is single-threaded, so no concurrent access. - let mut newest = unsafe { stream.dequeue_raw_buffer() }; - if newest.is_null() { - return; - } - let mut drained = 1u32; - loop { - // SAFETY: same stream/loop-thread contract as the dequeue above; each call returns - // the next stream-owned `*mut pw_buffer` or null (null-checked before use). - let next = unsafe { stream.dequeue_raw_buffer() }; - if next.is_null() { - break; - } - // SAFETY: `newest` is a non-null `*mut pw_buffer` previously dequeued from this same - // stream and not yet requeued; `pw_stream_queue_buffer` hands ownership back to the - // stream. We immediately overwrite `newest = next`, so the requeued pointer is never - // touched again (no use-after-requeue). Loop thread, single-threaded. - unsafe { stream.queue_raw_buffer(newest) }; - newest = next; - drained += 1; - } // SAFETY: `newest` is the non-null buffer we still own (dequeued, not requeued); // `.buffer` is a `*mut spa_buffer` field libpipewire populated. This is a single field // load through a valid pointer — no mutation or aliasing. @@ -2272,19 +2300,18 @@ mod pipewire { "capture: skipped a stale CORRUPTED/cursor buffer (GNOME)" ); } - // SAFETY: `newest` is the non-null buffer we own (dequeued, never requeued on this - // skip path); hand it back to the stream exactly once and return without touching it - // again. Loop thread inside `.process`. - unsafe { stream.queue_raw_buffer(newest) }; + // Skip this stale/cursor buffer — `newest` is requeued unconditionally below. return; } consume_frame(ud, spa_buf); - // SAFETY: `consume_frame` has finished reading `spa_buf` (and the `datas` borrows derived - // from `newest`), so requeuing the owned `newest` exactly once here is sound — no - // use-after-requeue. Loop thread inside `.process`. - unsafe { stream.queue_raw_buffer(newest) }; })); + // Hand `newest` back to the stream exactly once, on EVERY path — normal, corrupted-skip, + // or a caught panic in the closure above. This single requeue is what keeps the fixed + // buffer pool from draining. + // SAFETY: all reads of `spa_buf`/`newest` (update_cursor_meta, consume_frame) completed + // inside the closure above; `newest` was dequeued from this stream and not yet requeued. + unsafe { stream.queue_raw_buffer(newest) }; if outcome.is_err() { // In the per-frame `.process` callback: a deterministic panic (e.g. a bad // format) would fire this every frame, so power-of-two throttle it — enough to diff --git a/crates/pf-capture/src/windows/idd_push.rs b/crates/pf-capture/src/windows/idd_push.rs index c34fac88..089d1a09 100644 --- a/crates/pf-capture/src/windows/idd_push.rs +++ b/crates/pf-capture/src/windows/idd_push.rs @@ -1341,6 +1341,12 @@ impl IddPushCapturer { self.out_ring.clear(); // the output format changed → rebuild lazily at the new format self.video_conv = None; // converters are sized + HDR-specific → rebuild at the new mode self.hdr_p010_conv = None; + // The PyroWave CSC is mode-baked too (BgraToYuvPlanes picks different SDR vs HDR shaders + // and R8/R8G8 vs R16/R16G16 outputs). Without this, a display_hdr flip (Downgrade point D: + // client_10bit=true but HDR couldn't enable at open) reused the stale SDR converter against + // the freshly HDR-formatted pyro ring — every frame corrupted. `ensure_pyro_conv` only + // builds when None, so it must be reset here like its siblings. + self.pyro_conv = None; self.pyro_ring.clear(); // PyroWave two-plane ring is sized → rebuild at the new mode self.pyro_last = None; self.out_idx = 0; diff --git a/crates/pf-inject/src/inject/linux/steam_gadget.rs b/crates/pf-inject/src/inject/linux/steam_gadget.rs index e328640d..c734b1bb 100644 --- a/crates/pf-inject/src/inject/linux/steam_gadget.rs +++ b/crates/pf-inject/src/inject/linux/steam_gadget.rs @@ -17,7 +17,7 @@ use anyhow::{bail, Context, Result}; use std::mem::size_of; use std::os::fd::RawFd; -use std::sync::atomic::{AtomicBool, Ordering}; +use std::sync::atomic::{AtomicBool, AtomicU64, Ordering}; use std::sync::{Arc, Mutex}; use std::thread::JoinHandle; @@ -196,6 +196,41 @@ impl Drop for GadgetFd { } } +/// The signal used to break a worker thread out of a blocking raw_gadget ioctl at teardown. +/// `EVENT_FETCH`/`EP_WRITE` are `wait_event_interruptible` in the kernel with no timeout and no +/// `O_NONBLOCK` honouring, and closing the fd cannot wake a thread already inside the ioctl (the +/// in-flight syscall holds a reference to the struct file). A signal is the only reliable lever: +/// delivered with a no-op, non-`SA_RESTART` handler it forces the ioctl to return `EINTR`, after +/// which the loop's top-of-iteration `running` check exits. `SIGUSR1` is unused elsewhere in this +/// process; the handler is a no-op, so a stray `SIGUSR1` becomes harmless rather than fatal. +const WAKE_SIGNAL: libc::c_int = libc::SIGUSR1; + +/// Install the no-op `WAKE_SIGNAL` handler exactly once. Crucially `sa_flags = 0` (no `SA_RESTART`) +/// so a delivered signal makes the interruptible ioctl return `EINTR` instead of auto-restarting. +fn install_wake_handler() { + static ONCE: std::sync::Once = std::sync::Once::new(); + ONCE.call_once(|| { + extern "C" fn noop(_: libc::c_int) {} + // SAFETY: installing a well-formed `sigaction` with an empty mask and a valid no-op handler + // for a single signal; touches only this process's disposition for `WAKE_SIGNAL`. + unsafe { + let mut sa: libc::sigaction = std::mem::zeroed(); + sa.sa_sigaction = noop as usize; + libc::sigemptyset(&mut sa.sa_mask); + sa.sa_flags = 0; + libc::sigaction(WAKE_SIGNAL, &sa, std::ptr::null_mut()); + } + }); +} + +/// Lets `Drop` wake a specific worker thread parked in a blocking ioctl. `tid` is the thread's +/// `pthread_self()` (0 until it starts); `done` is set right before the thread returns, so `Drop` +/// stops signalling a thread that has already exited. +struct Waker { + tid: Arc, + done: Arc, +} + /// A virtual Steam Deck presented over the USB gadget subsystem. Dropping it stops the threads and /// closes the gadget (the kernel tears down the device). pub struct SteamDeckGadget { @@ -203,6 +238,7 @@ pub struct SteamDeckGadget { feedback: Arc>, running: Arc, threads: Vec>, + wakers: Vec, _fd: Arc, seq: u32, } @@ -243,6 +279,18 @@ impl SteamDeckGadget { let ctrl_ep = Arc::new(std::sync::atomic::AtomicI32::new(-1)); let configured = Arc::new(AtomicBool::new(false)); + // The teardown wake path (see `WAKE_SIGNAL`) needs the handler installed before any thread + // can park in a blocking ioctl. + install_wake_handler(); + let ctrl_waker = Waker { + tid: Arc::new(AtomicU64::new(0)), + done: Arc::new(AtomicBool::new(false)), + }; + let stream_waker = Waker { + tid: Arc::new(AtomicU64::new(0)), + done: Arc::new(AtomicBool::new(false)), + }; + // Control thread: enumerate + answer every control transfer. let control = { let fd = fd.clone(); @@ -250,10 +298,15 @@ impl SteamDeckGadget { let ctrl_ep = ctrl_ep.clone(); let configured = configured.clone(); let feedback = feedback.clone(); + let tid = ctrl_waker.tid.clone(); + let done = ctrl_waker.done.clone(); std::thread::Builder::new() .name("pf-deck-gadget-ctrl".into()) .spawn(move || { - control_loop(fd, running, ctrl_ep, configured, feedback, serial, unit_id) + // SAFETY: `pthread_self` is always valid on the calling thread. + tid.store(unsafe { libc::pthread_self() } as u64, Ordering::SeqCst); + control_loop(fd, running, ctrl_ep, configured, feedback, serial, unit_id); + done.store(true, Ordering::SeqCst); }) .context("spawn gadget control thread")? }; @@ -264,9 +317,16 @@ impl SteamDeckGadget { let ctrl_ep = ctrl_ep.clone(); let configured = configured.clone(); let report = report.clone(); + let tid = stream_waker.tid.clone(); + let done = stream_waker.done.clone(); std::thread::Builder::new() .name("pf-deck-gadget-stream".into()) - .spawn(move || stream_loop(fd, running, ctrl_ep, configured, report)) + .spawn(move || { + // SAFETY: `pthread_self` is always valid on the calling thread. + tid.store(unsafe { libc::pthread_self() } as u64, Ordering::SeqCst); + stream_loop(fd, running, ctrl_ep, configured, report); + done.store(true, Ordering::SeqCst); + }) .context("spawn gadget stream thread")? }; @@ -275,6 +335,7 @@ impl SteamDeckGadget { feedback, running, threads: vec![control, stream], + wakers: vec![ctrl_waker, stream_waker], _fd: fd, seq: 0, }) @@ -302,6 +363,32 @@ impl SteamDeckGadget { impl Drop for SteamDeckGadget { fn drop(&mut self) { self.running.store(false, Ordering::SeqCst); + // The control thread spends steady state parked in a blocking `EVENT_FETCH` ioctl that only + // tests `running` at the top of its loop, so clearing the flag is not enough — it must be + // signalled out of the syscall (see `WAKE_SIGNAL`). Without this the join below can hang the + // caller (the session input thread, via `PadSlots::sweep`) indefinitely. Retry until each + // thread reports done, to cover the race where the signal lands just before the thread + // re-enters the ioctl; bounded (~1 s) so a genuinely stuck thread can't wedge teardown either. + for _ in 0..200 { + let mut all_done = true; + for w in &self.wakers { + if w.done.load(Ordering::SeqCst) { + continue; + } + all_done = false; + let tid = w.tid.load(Ordering::SeqCst); + if tid != 0 { + // SAFETY: the thread is joinable and not yet joined (join runs after this loop), + // so `tid` names a live pthread; `pthread_kill` on a finished-but-unjoined thread + // is defined (returns ESRCH), never UB. + unsafe { libc::pthread_kill(tid as libc::pthread_t, WAKE_SIGNAL) }; + } + } + if all_done { + break; + } + std::thread::sleep(std::time::Duration::from_millis(5)); + } for t in self.threads.drain(..) { let _ = t.join(); } diff --git a/crates/pf-inject/src/inject/windows/dualsense_windows.rs b/crates/pf-inject/src/inject/windows/dualsense_windows.rs index 60c7d9d9..078e83f7 100644 --- a/crates/pf-inject/src/inject/windows/dualsense_windows.rs +++ b/crates/pf-inject/src/inject/windows/dualsense_windows.rs @@ -299,13 +299,20 @@ pub(super) fn create_swdevice(p: &SwDeviceProfile) -> Result<(HSWDEVICE, Option< let event = unsafe { CreateEventW(None, true, false, PCWSTR::null())? }; // `result` starts as E_FAIL, NOT S_OK: if the wait below times out, a zero-initialised HRESULT // would read as success and mask the failure (found by the 2026-07 driver-health audit). - let mut ctx = SwCreateCtx { + // HEAP-allocated, deliberately: `sw_create_cb` writes `result` + up to 127 u16 of instance id + // through this pointer and then `SetEvent`s. The wait below is bounded (10 s), so on a wedged-PnP + // timeout the callback may still be PENDING — a stack context would be popped and a late callback + // would corrupt whatever the input thread put there next, and SetEvent a closed/recycled handle. + // On the timeout path we therefore LEAK the box and leave the event open (a one-off ~264 B + one + // HANDLE, only on that rare path) so a late callback always writes to live memory. + let ctx = Box::into_raw(Box::new(SwCreateCtx { event, result: E_FAIL, instance_id: [0; 128], - }; - // SAFETY: info + the buffers + ctx outlive the call (we wait on the event before returning); - // windows-rs returns the HSWDEVICE (the C out-param) as the Result value. + })); + // SAFETY: info + the buffers outlive the call; `ctx` is a live heap allocation that outlives every + // path below (reclaimed only where the callback provably ran). windows-rs returns the HSWDEVICE + // (the C out-param) as the Result value. let hsw = match unsafe { SwDeviceCreate( w!("punktfunk"), @@ -313,13 +320,15 @@ pub(super) fn create_swdevice(p: &SwDeviceProfile) -> Result<(HSWDEVICE, Option< &info, None, Some(sw_create_cb), - Some(&mut ctx as *mut SwCreateCtx as *const c_void), + Some(ctx as *const c_void), ) } { Ok(h) => h, Err(e) => { - // SAFETY: event is valid. + // SAFETY: the call failed, so no callback was registered and `ctx` is ours to reclaim; + // `event` is valid and unreferenced. unsafe { + drop(Box::from_raw(ctx)); let _ = CloseHandle(event); } return Err(anyhow!("SwDeviceCreate failed: {e}")); @@ -328,17 +337,22 @@ pub(super) fn create_swdevice(p: &SwDeviceProfile) -> Result<(HSWDEVICE, Option< // Block until PnP finishes enumerating (the callback signals), then check its result. // SAFETY: event is valid. let wait = unsafe { WaitForSingleObject(event, 10_000) }; - // SAFETY: event is valid. - unsafe { - let _ = CloseHandle(event); - } if wait != WAIT_OBJECT_0 { + // Timed out: the callback may still fire. Intentionally leak `ctx` AND leave `event` open so + // its eventual write + SetEvent target live memory/handle rather than freed ones. // SAFETY: hsw is the handle SwDeviceCreate returned. unsafe { SwDeviceClose(hsw) }; return Err(anyhow!( "SwDeviceCreate enumeration callback never fired (10s) — PnP may be wedged" )); } + // The callback ran (it is what signalled the event), so nothing else will touch `ctx`/`event`. + // SAFETY: `ctx` came from `Box::into_raw` above and is reclaimed exactly once here; `event` is + // valid and no longer referenced by a pending callback. + let ctx = unsafe { + let _ = CloseHandle(event); + Box::from_raw(ctx) + }; if ctx.result.is_err() { // SAFETY: hsw is the handle SwDeviceCreate returned. unsafe { SwDeviceClose(hsw) }; diff --git a/crates/pf-inject/src/inject/windows/dualshock4_windows.rs b/crates/pf-inject/src/inject/windows/dualshock4_windows.rs index 84c65e46..836d23a0 100644 --- a/crates/pf-inject/src/inject/windows/dualshock4_windows.rs +++ b/crates/pf-inject/src/inject/windows/dualshock4_windows.rs @@ -62,7 +62,7 @@ impl Ds4WinPad { std::ptr::write_unaligned(base as *mut u32, SHM_MAGIC); } let inst = format!("pf_ds4_{index}"); - let (hsw, instance_id) = match create_swdevice(&SwDeviceProfile { + let (hsw, instance_id) = create_swdevice(&SwDeviceProfile { instance: &inst, container_tag: 0x5046_4453, // "PFDS" container_index: index, @@ -70,13 +70,13 @@ impl Ds4WinPad { usb_vid_pid: "VID_054C&PID_09CC", usb_mi: None, description: "punktfunk Virtual DualShock 4", - }) { - Ok((h, id)) => (Some(h), id), - Err(e) => { - tracing::warn!(error = %format!("{e:#}"), "SwDeviceCreate failed; DualShock 4 devnode unavailable"); - (None, None) - } - }; + })?; // Propagate, do NOT swallow — see below. + let (hsw, instance_id) = (Some(hsw), instance_id); + // Swallowing a create failure here (the previous behaviour) latched the pad slot to + // `Some(pad)` with no live devnode: `PadSlots::ensure` short-circuits on `is_some()` and + // `gate.on_success()` cleared the backoff, so the create-gate that exists precisely to + // self-heal a transient PnP failure never retried. The game saw no controller for the whole + // session unless the client unplugged the pad. Matches the XUSB sibling, which propagates. let _sw = hsw.map(super::gamepad_raii::SwDevice::new); // Bounded eager delivery — for the DS4 this is what closes the identity race: the driver // must read `device_type = 1` from the delivered DATA section before hidclass asks it for diff --git a/crates/pf-inject/src/inject/windows/gamepad_windows.rs b/crates/pf-inject/src/inject/windows/gamepad_windows.rs index b30d017e..e1c5d97f 100644 --- a/crates/pf-inject/src/inject/windows/gamepad_windows.rs +++ b/crates/pf-inject/src/inject/windows/gamepad_windows.rs @@ -82,12 +82,16 @@ fn create_swdevice(index: u8) -> Result<(HSWDEVICE, Option)> { let event = unsafe { CreateEventW(None, true, false, PCWSTR::null())? }; // `result` starts as E_FAIL, NOT S_OK: if the wait below times out, a zero-initialised HRESULT // would read as success and mask the failure (found by the 2026-07 driver-health audit). - let mut ctx = SwCreateCtx { + // HEAP-allocated for the same reason as the DualSense sibling: the callback writes through this + // pointer and SetEvents, and the wait below is bounded — a stack context would be popped while a + // late callback still holds it. On the timeout path the box is deliberately leaked and the event + // left open so a late write/SetEvent always targets live memory/handle. + let ctx = Box::into_raw(Box::new(SwCreateCtx { event, result: E_FAIL, instance_id: [0; 128], - }; - // SAFETY: info + buffers + ctx outlive the call (we wait on the event before returning). + })); + // SAFETY: info + buffers outlive the call; `ctx` is a live heap allocation outliving every path. let hsw = match unsafe { SwDeviceCreate( w!("punktfunk"), @@ -95,13 +99,14 @@ fn create_swdevice(index: u8) -> Result<(HSWDEVICE, Option)> { &info, None, Some(sw_create_cb), - Some(&mut ctx as *mut SwCreateCtx as *const c_void), + Some(ctx as *const c_void), ) } { Ok(h) => h, Err(e) => { - // SAFETY: event is valid. + // SAFETY: the call failed, so no callback is pending and `ctx` is ours to reclaim. unsafe { + drop(Box::from_raw(ctx)); let _ = CloseHandle(event); } return Err(anyhow!("SwDeviceCreate(pf_xusb) failed: {e}")); @@ -109,17 +114,20 @@ fn create_swdevice(index: u8) -> Result<(HSWDEVICE, Option)> { }; // SAFETY: event valid; block until PnP finishes enumerating, then check the callback result. let wait = unsafe { WaitForSingleObject(event, 10_000) }; - // SAFETY: event is valid. - unsafe { - let _ = CloseHandle(event); - } if wait != WAIT_OBJECT_0 { + // Timed out — intentionally leak `ctx` and leave `event` open (see above). // SAFETY: hsw is the handle SwDeviceCreate returned. unsafe { SwDeviceClose(hsw) }; return Err(anyhow!( "SwDeviceCreate(pf_xusb) enumeration callback never fired (10s) — PnP may be wedged" )); } + // The callback ran (it signalled the event), so nothing else will touch `ctx`/`event`. + // SAFETY: `ctx` came from `Box::into_raw` and is reclaimed exactly once here. + let ctx = unsafe { + let _ = CloseHandle(event); + Box::from_raw(ctx) + }; if ctx.result.is_err() { // SAFETY: hsw is the handle SwDeviceCreate returned. unsafe { SwDeviceClose(hsw) }; diff --git a/crates/pf-inject/src/inject/windows/steam_deck_windows.rs b/crates/pf-inject/src/inject/windows/steam_deck_windows.rs index 14e215e1..8994af40 100644 --- a/crates/pf-inject/src/inject/windows/steam_deck_windows.rs +++ b/crates/pf-inject/src/inject/windows/steam_deck_windows.rs @@ -66,7 +66,7 @@ impl DeckWinPad { std::ptr::write_unaligned(base as *mut u32, SHM_MAGIC); } let inst = format!("pf_deck_{index}"); - let (hsw, instance_id) = match create_swdevice(&SwDeviceProfile { + let (hsw, instance_id) = create_swdevice(&SwDeviceProfile { instance: &inst, container_tag: 0x5046_4453, // "PFDS" container_index: index, @@ -77,13 +77,8 @@ impl DeckWinPad { // spike's run-1 failure). usb_mi: Some(2), description: "punktfunk Virtual Steam Deck", - }) { - Ok((h, i)) => (Some(h), i), - Err(e) => { - tracing::warn!(error = %format!("{e:#}"), "SwDeviceCreate failed; Steam Deck devnode unavailable"); - (None, None) - } - }; + })?; // Propagate — swallowing latched the slot to a pad with no devnode (see the DS4 twin). + let (hsw, instance_id) = (Some(hsw), instance_id); let _sw = hsw.map(super::gamepad_raii::SwDevice::new); // Bounded eager delivery — the driver must read `device_type = 3` before hidclass asks // it for descriptors, or the pad would enumerate with the default DualSense identity. diff --git a/crates/pf-zerocopy/src/imp/client.rs b/crates/pf-zerocopy/src/imp/client.rs index 3183b3e5..e260dc6a 100644 --- a/crates/pf-zerocopy/src/imp/client.rs +++ b/crates/pf-zerocopy/src/imp/client.rs @@ -21,7 +21,7 @@ use std::path::{Path, PathBuf}; use std::process::{Child, Command}; use std::sync::atomic::{AtomicBool, Ordering}; use std::sync::{Arc, Mutex, OnceLock}; -use std::time::Duration; +use std::time::{Duration, Instant}; /// Handshake budget: EGL + CUDA bring-up is ~200 ms; a cold driver load can take seconds. const HANDSHAKE_TIMEOUT: Duration = Duration::from_secs(20); @@ -64,11 +64,27 @@ impl Drop for Shared { /// Children whose worker hasn't exited yet at `RemoteImporter` drop time (it exits on socket /// EOF, i.e. after the last in-flight frame drops). Swept on every spawn and every drop so /// workers don't linger as zombies for more than one capture generation. -static REAPER: Mutex> = Mutex::new(Vec::new()); +static REAPER: Mutex> = Mutex::new(Vec::new()); + +/// How long past `REPLY_TIMEOUT` a parked worker may linger before it is force-killed. A worker +/// wedged INSIDE a driver call never observes socket EOF, so `try_wait` alone would keep it (and +/// its CUcontext + BufferPool — order hundreds of MB of VRAM) forever. +const REAPER_KILL_DEADLINE: Duration = Duration::from_secs(20); fn sweep_reaper() { let mut list = REAPER.lock().unwrap(); - list.retain_mut(|c| !matches!(c.try_wait(), Ok(Some(_)))); + let now = Instant::now(); + list.retain_mut(|(c, parked)| { + if matches!(c.try_wait(), Ok(Some(_))) { + return false; // exited on its own → reaped + } + if now.duration_since(*parked) > REAPER_KILL_DEADLINE { + let _ = c.kill(); + let _ = c.wait(); + return false; // wedged past the deadline → force-killed + reaped + } + true + }); } /// Fd pinned to this process's own executable image, opened (once, lazily) via the @@ -455,7 +471,7 @@ impl Drop for RemoteImporter { // gone; park the rest for the next sweep. if let Some(mut child) = self.child.take() { if !matches!(child.try_wait(), Ok(Some(_))) { - REAPER.lock().unwrap().push(child); + REAPER.lock().unwrap().push((child, Instant::now())); } } sweep_reaper(); diff --git a/crates/pf-zerocopy/src/imp/cuda.rs b/crates/pf-zerocopy/src/imp/cuda.rs index 4e9a0e3f..4b80cc04 100644 --- a/crates/pf-zerocopy/src/imp/cuda.rs +++ b/crates/pf-zerocopy/src/imp/cuda.rs @@ -1003,7 +1003,13 @@ impl RegisteredTexture { // SAFETY: `self.resource` is the valid `CUgraphicsResource` from a successful `register_gl` // (its only constructor), so the wrappers forward to the live table; the caller holds the // GL+CUDA contexts current (the registration's contract). `cuGraphicsMapResources` maps - // `count == 1` resource via `&mut self.resource` (a live field) on the default stream; + // `count == 1` resource via `&mut self.resource` (a live field). It is issued on + // `copy_stream()` — NOT the NULL stream — because map's only ordering guarantee is that + // prior GL work completes before subsequent CUDA work issued IN THE STREAM PASSED TO IT; + // the copy below runs on `copy_stream()` (a `CU_STREAM_NON_BLOCKING` stream, exempt from + // implicit NULL-stream ordering), so mapping on NULL left the copy free to race the GL + // de-tile/CSC that produced this texture (glFlush only, no fence) — intermittent torn or + // stale frames under GPU load. Map, copy, and unmap now all share `copy_stream()`. // `cuGraphicsSubResourceGetMappedArray` writes the mapped `CUarray` into the live local // `array` (index 0, mip 0). On failure we unmap and bail (balanced). `©` is a live // local `CUDA_MEMCPY2D` outliving the synchronous `copy_blocking`: `srcArray` is valid @@ -1012,12 +1018,12 @@ impl RegisteredTexture { // we always unmap afterward (even on error), keeping the map/unmap pair balanced. unsafe { ck( - cuGraphicsMapResources(1, &mut self.resource, std::ptr::null_mut()), + cuGraphicsMapResources(1, &mut self.resource, copy_stream()), "cuGraphicsMapResources", )?; let mut array: CUarray = std::ptr::null_mut(); if cuGraphicsSubResourceGetMappedArray(&mut array, self.resource, 0, 0) != 0 { - let _ = cuGraphicsUnmapResources(1, &mut self.resource, std::ptr::null_mut()); + let _ = cuGraphicsUnmapResources(1, &mut self.resource, copy_stream()); bail!("cuGraphicsSubResourceGetMappedArray failed"); } let copy = CUDA_MEMCPY2D { @@ -1031,7 +1037,7 @@ impl RegisteredTexture { ..Default::default() }; let res = copy_blocking(©, "cuMemcpy2DAsync_v2"); - let _ = cuGraphicsUnmapResources(1, &mut self.resource, std::ptr::null_mut()); + let _ = cuGraphicsUnmapResources(1, &mut self.resource, copy_stream()); res } } @@ -1058,12 +1064,12 @@ impl RegisteredTexture { // so the map/unmap pair stays balanced and the array outlives the copy. unsafe { ck( - cuGraphicsMapResources(1, &mut self.resource, std::ptr::null_mut()), + cuGraphicsMapResources(1, &mut self.resource, copy_stream()), "cuGraphicsMapResources", )?; let mut array: CUarray = std::ptr::null_mut(); if cuGraphicsSubResourceGetMappedArray(&mut array, self.resource, 0, 0) != 0 { - let _ = cuGraphicsUnmapResources(1, &mut self.resource, std::ptr::null_mut()); + let _ = cuGraphicsUnmapResources(1, &mut self.resource, copy_stream()); bail!("cuGraphicsSubResourceGetMappedArray failed"); } let copy = CUDA_MEMCPY2D { @@ -1077,7 +1083,7 @@ impl RegisteredTexture { ..Default::default() }; let res = copy_blocking(©, "cuMemcpy2DAsync_v2(plane)"); - let _ = cuGraphicsUnmapResources(1, &mut self.resource, std::ptr::null_mut()); + let _ = cuGraphicsUnmapResources(1, &mut self.resource, copy_stream()); res } } diff --git a/crates/pf-zerocopy/src/imp/egl.rs b/crates/pf-zerocopy/src/imp/egl.rs index 8e8a9060..951c450b 100644 --- a/crates/pf-zerocopy/src/imp/egl.rs +++ b/crates/pf-zerocopy/src/imp/egl.rs @@ -691,6 +691,15 @@ impl EglImporter { width: u32, height: u32, ) -> Result { + // Even dimensions only: the UV copy walks `height.div_ceil(2)` chroma rows (the correct NV12 + // count), but the pooled UV plane is sized at `height/2` rows — for an odd height those + // disagree by one row and the copy writes a full `uv_pitch` past the allocation (OOB device + // write / CUDA_ERROR_ILLEGAL_ADDRESS that poisons the shared context). Reject here, matching + // the guards `Nv12Blit::new`/`Yuv444Blit::new` already carry. + anyhow::ensure!( + width % 2 == 0 && height % 2 == 0, + "LINEAR NV12 needs even dimensions (got {width}x{height})" + ); cuda::make_current()?; if self .linear_nv12_pool diff --git a/crates/pf-zerocopy/src/imp/vulkan.rs b/crates/pf-zerocopy/src/imp/vulkan.rs index bcc76318..ae427588 100644 --- a/crates/pf-zerocopy/src/imp/vulkan.rs +++ b/crates/pf-zerocopy/src/imp/vulkan.rs @@ -193,10 +193,17 @@ impl VkBridge { /// Import `fd` (dup'd internally; Vulkan owns the dup) as a transfer-src buffer of `size`. unsafe fn import_src(&mut self, fd: i32, size: u64) -> Result<()> { + use std::os::fd::{AsRawFd, FromRawFd, IntoRawFd, OwnedFd}; let dup = libc::dup(fd); if dup < 0 { bail!("dup(dmabuf fd)"); } + // Own the dup so every early return BEFORE Vulkan consumes it (at `allocate_memory` success) + // closes it. `SrcBuf` holds raw handles with no Drop and is only populated on the success + // path, so each fallible step below must also destroy the buffer it created — otherwise a + // failed import (which the worker survives and the caller retries every frame) leaks a + // VkBuffer + VkDeviceMemory + fd per frame for the worker's whole lifetime. + let dup = OwnedFd::from_raw_fd(dup); let mut ext_info = vk::ExternalMemoryBufferCreateInfo::default() .handle_types(vk::ExternalMemoryHandleTypeFlags::DMA_BUF_EXT); let buffer = self @@ -212,41 +219,55 @@ impl VkBridge { .push_next(&mut ext_info), None, ) - .context("create import buffer")?; + .context("create import buffer")?; // `dup` drops → closes on failure let mut fd_props = vk::MemoryFdPropertiesKHR::default(); - self.ext_fd - .get_memory_fd_properties( - vk::ExternalMemoryHandleTypeFlags::DMA_BUF_EXT, - dup, - &mut fd_props, - ) - .context("vkGetMemoryFdPropertiesKHR")?; + if let Err(e) = self.ext_fd.get_memory_fd_properties( + vk::ExternalMemoryHandleTypeFlags::DMA_BUF_EXT, + dup.as_raw_fd(), + &mut fd_props, + ) { + self.device.destroy_buffer(buffer, None); + return Err(e).context("vkGetMemoryFdPropertiesKHR"); + } let reqs = self.device.get_buffer_memory_requirements(buffer); - let mem_type = self.memory_type( + let mem_type = match self.memory_type( reqs.memory_type_bits & fd_props.memory_type_bits, vk::MemoryPropertyFlags::empty(), - )?; + ) { + Ok(t) => t, + Err(e) => { + self.device.destroy_buffer(buffer, None); + return Err(e); + } + }; + // Vulkan takes ownership of the fd on a SUCCESSFUL import: hand over the raw fd now, and on + // failure close it ourselves (matching the original contract) plus destroy the buffer. + let raw = dup.into_raw_fd(); let mut import = vk::ImportMemoryFdInfoKHR::default() .handle_type(vk::ExternalMemoryHandleTypeFlags::DMA_BUF_EXT) - .fd(dup); // Vulkan takes ownership of `dup` on success + .fd(raw); let mut dedicated = vk::MemoryDedicatedAllocateInfo::default().buffer(buffer); - let memory = self - .device - .allocate_memory( - &vk::MemoryAllocateInfo::default() - .allocation_size(reqs.size.max(size)) - .memory_type_index(mem_type) - .push_next(&mut import) - .push_next(&mut dedicated), - None, - ) - .map_err(|e| { - libc::close(dup); // failed import does not consume the fd - anyhow!("import dmabuf memory: {e}") - })?; - self.device - .bind_buffer_memory(buffer, memory, 0) - .context("bind import memory")?; + let memory = match self.device.allocate_memory( + &vk::MemoryAllocateInfo::default() + .allocation_size(reqs.size.max(size)) + .memory_type_index(mem_type) + .push_next(&mut import) + .push_next(&mut dedicated), + None, + ) { + Ok(m) => m, + Err(e) => { + libc::close(raw); // failed import does not consume the fd + self.device.destroy_buffer(buffer, None); + return Err(anyhow!("import dmabuf memory: {e}")); + } + }; + if let Err(e) = self.device.bind_buffer_memory(buffer, memory, 0) { + // `memory` owns the imported fd — freeing it releases the fd too. + self.device.free_memory(memory, None); + self.device.destroy_buffer(buffer, None); + return Err(e).context("bind import memory"); + } self.src_cache.insert( fd, SrcBuf { @@ -263,11 +284,11 @@ impl VkBridge { if self.dst.as_ref().is_some_and(|d| d.size >= size) { return Ok(()); } - if let Some(old) = self.dst.take() { - self.device.destroy_buffer(old.buffer, None); - self.device.free_memory(old.memory, None); - // old.cuda drops its mapping with it - } + // Build the replacement FULLY before retiring the old one. Previously the old dst was + // destroyed and `self.dst` nulled up front, so a failed rebuild both dropped the working + // buffer AND leaked every object the partial rebuild created (`buffer`/`memory` are raw ash + // handles with no Drop, and `VkBridge::drop` only frees the live `self.dst`). Now every + // fallible step unwinds locally, and the swap happens only on full success. let mut ext_info = vk::ExternalMemoryBufferCreateInfo::default() .handle_types(vk::ExternalMemoryHandleTypeFlags::OPAQUE_FD); let buffer = self @@ -285,35 +306,63 @@ impl VkBridge { .context("create export buffer")?; let reqs = self.device.get_buffer_memory_requirements(buffer); let mem_type = - self.memory_type(reqs.memory_type_bits, vk::MemoryPropertyFlags::DEVICE_LOCAL)?; + match self.memory_type(reqs.memory_type_bits, vk::MemoryPropertyFlags::DEVICE_LOCAL) { + Ok(t) => t, + Err(e) => { + self.device.destroy_buffer(buffer, None); + return Err(e); + } + }; let mut export = vk::ExportMemoryAllocateInfo::default() .handle_types(vk::ExternalMemoryHandleTypeFlags::OPAQUE_FD); let mut dedicated = vk::MemoryDedicatedAllocateInfo::default().buffer(buffer); - let memory = self - .device - .allocate_memory( - &vk::MemoryAllocateInfo::default() - .allocation_size(reqs.size) - .memory_type_index(mem_type) - .push_next(&mut export) - .push_next(&mut dedicated), - None, - ) - .context("allocate exportable memory")?; - self.device - .bind_buffer_memory(buffer, memory, 0) - .context("bind export memory")?; - let opaque_fd = self - .ext_fd - .get_memory_fd( - &vk::MemoryGetFdInfoKHR::default() - .memory(memory) - .handle_type(vk::ExternalMemoryHandleTypeFlags::OPAQUE_FD), - ) - .context("vkGetMemoryFdKHR")?; + let memory = match self.device.allocate_memory( + &vk::MemoryAllocateInfo::default() + .allocation_size(reqs.size) + .memory_type_index(mem_type) + .push_next(&mut export) + .push_next(&mut dedicated), + None, + ) { + Ok(m) => m, + Err(e) => { + self.device.destroy_buffer(buffer, None); + return Err(e).context("allocate exportable memory"); + } + }; + if let Err(e) = self.device.bind_buffer_memory(buffer, memory, 0) { + self.device.free_memory(memory, None); + self.device.destroy_buffer(buffer, None); + return Err(e).context("bind export memory"); + } + let opaque_fd = match self.ext_fd.get_memory_fd( + &vk::MemoryGetFdInfoKHR::default() + .memory(memory) + .handle_type(vk::ExternalMemoryHandleTypeFlags::OPAQUE_FD), + ) { + Ok(f) => f, + Err(e) => { + self.device.free_memory(memory, None); + self.device.destroy_buffer(buffer, None); + return Err(e).context("vkGetMemoryFdKHR"); + } + }; // CUDA imports (and on success owns) the exported fd. Size must match the allocation. - let cuda = cuda::ExternalDmabuf::import_owned_fd(opaque_fd, reqs.size) - .context("cuImportExternalMemory(OPAQUE_FD from Vulkan)")?; + // `import_owned_fd` closes `opaque_fd` on its own failure, so only the Vulkan objects unwind. + let cuda = match cuda::ExternalDmabuf::import_owned_fd(opaque_fd, reqs.size) { + Ok(c) => c, + Err(e) => { + self.device.free_memory(memory, None); + self.device.destroy_buffer(buffer, None); + return Err(e).context("cuImportExternalMemory(OPAQUE_FD from Vulkan)"); + } + }; + // Full success: retire the previous buffer now, then publish the new one. + if let Some(old) = self.dst.take() { + self.device.destroy_buffer(old.buffer, None); + self.device.free_memory(old.memory, None); + // old.cuda drops its mapping with it + } tracing::info!(size, "Vulkan→CUDA exportable staging buffer ready"); self.dst = Some(DstBuf { buffer, @@ -544,9 +593,19 @@ impl VkBridge { self.device .queue_submit(self.queue, &[submit], self.fence) .context("queue submit")?; - self.device + // Exception-safe wait: a TIMEOUT/DEVICE_LOST must not `?` out with the submission still + // executing — `self.cmd` and `self.fence` are reused every frame, and the caller retries + // on the SAME bridge (and `ensure_dst` later destroys `dst.buffer` assuming no in-flight + // work references it). Drain the GPU and reset the fence before propagating so the shared + // cmd/fence return clean. + if let Err(e) = self + .device .wait_for_fences(&[self.fence], true, 1_000_000_000) - .context("fence wait")?; + { + let _ = self.device.device_wait_idle(); + let _ = self.device.reset_fences(&[self.fence]); + return Err(e).context("fence wait"); + } self.device .reset_fences(&[self.fence]) .context("reset fence")?; @@ -639,9 +698,19 @@ impl VkBridge { self.device .queue_submit(self.queue, &[submit], self.fence) .context("queue submit")?; - self.device + // Exception-safe wait: a TIMEOUT/DEVICE_LOST must not `?` out with the submission still + // executing — `self.cmd` and `self.fence` are reused every frame, and the caller retries + // on the SAME bridge (and `ensure_dst` later destroys `dst.buffer` assuming no in-flight + // work references it). Drain the GPU and reset the fence before propagating so the shared + // cmd/fence return clean. + if let Err(e) = self + .device .wait_for_fences(&[self.fence], true, 1_000_000_000) - .context("fence wait")?; + { + let _ = self.device.device_wait_idle(); + let _ = self.device.reset_fences(&[self.fence]); + return Err(e).context("fence wait"); + } self.device .reset_fences(&[self.fence]) .context("reset fence")?;