diff --git a/crates/pf-capture/src/windows/idd_push.rs b/crates/pf-capture/src/windows/idd_push.rs index 7a37beec..c34fac88 100644 --- a/crates/pf-capture/src/windows/idd_push.rs +++ b/crates/pf-capture/src/windows/idd_push.rs @@ -1861,6 +1861,7 @@ impl IddPushCapturer { cbcr, fence_handle, fence_value, + ring_gen: self.generation, }), ) } else { @@ -1919,6 +1920,7 @@ impl IddPushCapturer { cbcr: dst_cbcr, fence_handle, fence_value, + ring_gen: self.generation, }), }), cursor: None, diff --git a/crates/pf-encode/src/enc/codec.rs b/crates/pf-encode/src/enc/codec.rs index 0be52010..5bd34720 100644 --- a/crates/pf-encode/src/enc/codec.rs +++ b/crates/pf-encode/src/enc/codec.rs @@ -293,6 +293,16 @@ pub trait Encoder: Send { /// flagged [`EncodedFrame::chunk_aligned`] and the session marks them on the wire. /// Default: no-op (the H.26x backends' bitstreams cannot be cut losslessly). fn set_wire_chunking(&mut self, _shard_payload: usize) {} + /// How many frames the CAPTURER guarantees the encoder may hold in flight before it starts + /// reusing an input texture (`Capturer::pipeline_depth`). Backends that encode the capturer's + /// textures IN PLACE — no `CopyResource` — must not pipeline deeper than this: the capturer + /// rotates its output ring per delivered frame with no regard for encode completion, so a + /// deeper pipeline lets it overwrite a texture mid-encode. That is visual corruption (torn or + /// mixed frames), not UB, so it fails silently and intermittently. + /// + /// Called once by the session glue after the capturer is known; a backend that copies its + /// input, or is synchronous, ignores it. Default: no-op. + fn set_input_ring_depth(&mut self, _depth: usize) {} /// Signal end-of-stream. After this, drain the remaining AUs with [`poll`](Self::poll) /// until it returns `None` — NVENC buffers frames internally even at `delay=0`. fn flush(&mut self) -> Result<()>; diff --git a/crates/pf-encode/src/enc/linux/pyrowave.rs b/crates/pf-encode/src/enc/linux/pyrowave.rs index 43ccd6eb..d2d04169 100644 --- a/crates/pf-encode/src/enc/linux/pyrowave.rs +++ b/crates/pf-encode/src/enc/linux/pyrowave.rs @@ -1176,7 +1176,24 @@ impl Encoder for PyroWaveEncoder { fn submit(&mut self, frame: &CapturedFrame) -> Result<()> { // SAFETY: single-threaded encoder; `encode_frame` records/submits on handles this // struct owns and waits its own fence before touching results. - unsafe { self.encode_frame(frame) } + let r = unsafe { self.encode_frame(frame) }; + if r.is_err() { + // `encode_frame` opens the recording window early and has several fallible steps + // inside it (cursor prep, dmabuf import, format mapping, the CPU-RGB staging path, + // an unsupported-payload bail, and the encode call itself). Every one returns with + // `self.cmd` still RECORDING, and nothing downstream repairs it — there is exactly + // one `begin_command_buffer` in this file and `reset()`/`Drop` never touch `cmd` — + // so the NEXT frame would call `begin` on a recording buffer, which is invalid usage. + // Legal here on every path: the pool carries RESET_COMMAND_BUFFER and the buffer is + // not pending (we never reached the submit, or the submit itself failed). + // SAFETY: `self.cmd` is owned by this encoder and, on these paths, not in flight. + unsafe { + let _ = self + .device + .reset_command_buffer(self.cmd, vk::CommandBufferResetFlags::empty()); + } + } + r } fn caps(&self) -> EncoderCaps { diff --git a/crates/pf-encode/src/enc/linux/vulkan_video.rs b/crates/pf-encode/src/enc/linux/vulkan_video.rs index d99c97b4..e3b51e93 100644 --- a/crates/pf-encode/src/enc/linux/vulkan_video.rs +++ b/crates/pf-encode/src/enc/linux/vulkan_video.rs @@ -37,6 +37,12 @@ const DPB_SLOTS: u32 = 8; /// latency — on-glass validated as rock-solid at 1080p@240, so it is the real-time default; /// backpressure kicks in at the 2nd unread frame. Distinct from `DPB_SLOTS` (reference pool). const RING_DEFAULT: usize = 2; + +/// Ceiling on any blocking GPU fence wait on the encode thread (5 s). Generous against a real +/// encode (single-digit ms even on a loaded GPU) and against a driver hiccup, but finite: this is +/// the thread the stall watchdog's `reset()` runs on, so an unbounded wait would deadlock the very +/// path that recovers the session. Matches the Windows NVENC retrieve-thread budget. +const ENCODE_FENCE_TIMEOUT_NS: u64 = 5_000_000_000; /// AV1 base quantizer index (0..=255) seeded into every frame. CBR rate control overrides it per /// frame; it only matters as the starting point and for the (rate-control-ignored) constant-Q path. const AV1_BASE_Q_IDX: u8 = 128; @@ -868,6 +874,15 @@ impl VulkanVideoEncoder { let (img, mem, view) = self.import_dmabuf(d, cw, ch)?; // Bound the cache; evict oldest (FIFO). A stable PipeWire pool never trips this in steady state // (all imports resident); it only cycles across a pool change (which also rebuilds the session). + // Up to `ring_depth - 1` submitted frames may still be executing against a cached image + // (`enqueue` only drains down to `frames.len()`, and `record_submit` imports before it + // records), so destroying an evicted import here is a GPU-side use-after-free. `Drop` and + // `reset()` both idle the device first; this was the one unguarded destroy. Guarded on the + // length test so the steady-state path — where the cache is resident and never evicts — + // pays nothing. + if self.import_cache.len() >= IMPORT_CACHE_CAP { + let _ = self.device.device_wait_idle(); + } while self.import_cache.len() >= IMPORT_CACHE_CAP { let (_, _, oi, om, ov) = self.import_cache.remove(0); self.device.destroy_image_view(ov, None); @@ -1849,8 +1864,25 @@ impl VulkanVideoEncoder { unsafe fn read_slot(&mut self, slot: usize) -> Result { let dev = self.device.clone(); let f = &self.frames[slot]; - let mut fb = [[0u32; 2]; 1]; - dev.get_query_pool_results(f.query_pool, 0, &mut fb, vk::QueryResultFlags::WAIT)?; + // Ask for the operation status alongside the two feedback words: without it a FAILED encode + // is indistinguishable from a successful one, and its offset/bytes-written are read as if + // they described real bitstream. The status rides as a trailing element (signed: + // `VkQueryResultStatusKHR` is >0 COMPLETE, 0 NOT_READY, <0 error). + let mut fb = [[0i32; 3]; 1]; + dev.get_query_pool_results( + f.query_pool, + 0, + &mut fb, + vk::QueryResultFlags::WAIT | vk::QueryResultFlags::WITH_STATUS_KHR, + )?; + let status = fb[0][2]; + if status <= 0 { + anyhow::bail!( + "vulkan-encode: encode feedback for slot {slot} reports status {status} \ + (not COMPLETE) — dropping the frame rather than shipping its bitstream" + ); + } + let fb = [[fb[0][0] as u32, fb[0][1] as u32]]; // The (offset, bytes-written) pair is driver-reported: validate it against the bitstream // allocation BEFORE mapping, or the `from_raw_parts` below reads outside the buffer and // ships whatever it finds straight onto the wire. Checked in u64 so the add cannot wrap, @@ -1892,8 +1924,25 @@ impl VulkanVideoEncoder { // and free it — that oldest slot is exactly the round-robin `ring` cursor we reuse next. while self.in_flight.len() >= self.frames.len() { let slot = self.in_flight.pop_front().unwrap(); - self.device - .wait_for_fences(&[self.frames[slot].fence], true, u64::MAX)?; + // Bounded, not `u64::MAX`: this runs ON the host encode thread, which is also the + // thread the stall watchdog's `reset()` would run on. An infinite wait against a + // wedged GPU/driver therefore parks the one thread that could recover the session — + // it never errors, never resets, and teardown blocks joining it. Surfacing expiry as + // an error hands control back to the existing recovery path (same convention as the + // pyrowave and Windows NVENC backends). + match self.device.wait_for_fences( + &[self.frames[slot].fence], + true, + ENCODE_FENCE_TIMEOUT_NS, + ) { + Ok(()) => {} + Err(vk::Result::TIMEOUT) => anyhow::bail!( + "vulkan-encode: fence for slot {slot} did not signal within {} ms — GPU or \ + driver wedged; failing the submit so the session can reset", + ENCODE_FENCE_TIMEOUT_NS / 1_000_000 + ), + Err(e) => return Err(e.into()), + } let done = self.read_slot(slot)?; self.pending.push_back(done); } diff --git a/crates/pf-encode/src/enc/windows/nvenc.rs b/crates/pf-encode/src/enc/windows/nvenc.rs index 6c2cd7dc..ef3deb89 100644 --- a/crates/pf-encode/src/enc/windows/nvenc.rs +++ b/crates/pf-encode/src/enc/windows/nvenc.rs @@ -409,6 +409,12 @@ pub struct NvencD3d11Encoder { events: Vec, /// Async mode: the retrieve thread + its channels (`None` = classic same-thread sync retrieve). async_rt: Option, + /// The capturer's `pipeline_depth` (`set_input_ring_depth`). This backend encodes the + /// capturer's textures IN PLACE, so it is a HARD ceiling on async in-flight depth: the + /// capturer rotates its ring per delivered frame regardless of encode completion, so + /// pipelining deeper lets it overwrite a texture mid-encode (torn frames). `None` until the + /// session glue reports it — treated as "unknown, don't pipeline past the env cap". + input_ring_depth: Option, /// `NV_ENC_CAPS_ASYNC_ENCODE_SUPPORT` from the caps probe — gates the async retrieve mode. async_supported: bool, /// (bitstream, mapped input resource to unmap after retrieval, pts_ns, recovery-anchor) per @@ -505,6 +511,7 @@ impl NvencD3d11Encoder { bitstreams: Vec::new(), events: Vec::new(), async_rt: None, + input_ring_depth: None, async_supported: false, pending: VecDeque::new(), frame_idx: 0, @@ -1156,11 +1163,21 @@ impl Encoder for NvencD3d11Encoder { // index, which is non-zero on a mid-session encoder rebuild's first frame. let opening = self.next == 0; // Async backpressure: never hand NVENC an output bitstream that is still in flight, and - // keep in-flight depth within the capturer's texture ring (see `async_inflight_cap`). At - // the cap, block on the OLDEST completion (the retrieve thread is already waiting on its - // event) before submitting more — bounding depth exactly like the sync path's per-tick - // blocking poll, just `cap` deep instead of 1. - while self.async_rt.is_some() && self.pending.len() >= async_inflight_cap() { + // keep in-flight depth within the capturer's texture ring. At the cap, block on the OLDEST + // completion (the retrieve thread is already waiting on its event) before submitting more — + // bounding depth exactly like the sync path's per-tick blocking poll, just `cap` deep + // instead of 1. + // + // The ring term is the one that matters for correctness: `async_inflight_cap()` is only the + // output-bitstream-pool ceiling plus an env knob, and consults NOTHING about the capturer, + // despite this comment previously claiming otherwise. Since this backend encodes the + // capturer's textures in place, exceeding the capturer's declared `pipeline_depth` lets it + // rotate a texture out from under a live encode — torn frames, silently. + let cap = match self.input_ring_depth { + Some(d) => async_inflight_cap().min(d.max(1)), + None => async_inflight_cap(), + }; + while self.async_rt.is_some() && self.pending.len() >= cap { let done = { let rt = self.async_rt.as_mut().expect("checked in loop condition"); rt.done_rx @@ -1336,6 +1353,17 @@ impl Encoder for NvencD3d11Encoder { self.submit(frame) } + fn set_input_ring_depth(&mut self, depth: usize) { + // This backend registers and encodes the capturer's textures in place (no CopyResource), + // so the capturer's ring depth is a hard ceiling on how deep async may pipeline. + self.input_ring_depth = Some(depth); + tracing::debug!( + depth, + env_cap = async_inflight_cap(), + "NVENC: capturer input-ring depth reported — async in-flight bounded by the smaller" + ); + } + fn request_keyframe(&mut self) { self.force_kf = true; } diff --git a/crates/pf-encode/src/enc/windows/pyrowave.rs b/crates/pf-encode/src/enc/windows/pyrowave.rs index a68fd3d9..93001490 100644 --- a/crates/pf-encode/src/enc/windows/pyrowave.rs +++ b/crates/pf-encode/src/enc/windows/pyrowave.rs @@ -43,6 +43,9 @@ const BS_SLACK: usize = 256 * 1024; /// (a desktop-switch device recreate), in which case the stale imports are evicted + destroyed. const IMPORT_CACHE_CAP: usize = 8; +/// Plane-import cache key: the texture's COM address plus the extent it was imported at. +type PlaneKey = (isize, u32, u32); + // --- Vulkan enum values not surfaced by pyrowave-sys' bindgen (only enums *reachable* from the // pyrowave C API are generated; these plain #define / flags-typedef values are stable spec // constants). bindgen renders every reachable Vulkan enum as a `u32` type alias, so these u32 @@ -136,8 +139,12 @@ pub struct PyroWaveEncoder { // Imported plane textures, cached by the out-ring texture's raw pointer (stable per ring slot): // the full-res R8 Y plane and the half-res R8G8 CbCr plane, imported SEPARATELY (a single planar // NV12 import is unreliable on NVIDIA at arbitrary sizes). - y_images: Vec<(isize, pw::pyrowave_image)>, - cbcr_images: Vec<(isize, pw::pyrowave_image)>, + /// The capturer ring generation the cached plane imports below belong to. A recreate bumps it, + /// and every cached import is destroyed — the COM addresses they are keyed on can be recycled + /// by the allocator after a recreate, so identity cannot rest on the pointer alone. + ring_gen: Option, + y_images: Vec<(PlaneKey, pw::pyrowave_image)>, + cbcr_images: Vec<(PlaneKey, pw::pyrowave_image)>, width: u32, height: u32, @@ -268,6 +275,7 @@ impl PyroWaveEncoder { pw_dev, pw_enc, sync: std::ptr::null_mut(), + ring_gen: None, y_images: Vec::new(), cbcr_images: Vec::new(), width, @@ -351,10 +359,16 @@ impl PyroWaveEncoder { /// /// # Safety /// Same contract as [`import_plane`]. + /// Keyed on `(texture address, width, height)` rather than the bare address: the COM pointer + /// carries no reference here, so a released texture's address can be recycled by a later + /// allocation and return an import describing the WRONG surface. Folding the extent in means a + /// recycled address at a different size can never alias. (A recycle at the SAME size is still + /// possible in principle — the complete fix is to key on the capturer's ring generation, which + /// needs that generation plumbed onto `PyroFrameShare`.) unsafe fn cached_plane( - cache: &mut Vec<(isize, pw::pyrowave_image)>, + cache: &mut Vec<(PlaneKey, pw::pyrowave_image)>, make: impl FnOnce() -> Result, - key: isize, + key: PlaneKey, ) -> Result { if let Some((_, img)) = cache.iter().find(|(k, _)| *k == key) { return Ok(*img); @@ -423,6 +437,21 @@ impl PyroWaveEncoder { !self.pw_enc.is_null(), "pyrowave: encode after a failed reset (encoder was destroyed and not rebuilt)" ); + // The plane textures are imported at the encoder's CONFIGURED extent, not the frame's, so a + // capture that changed size would be read under a stale `VkImageCreateInfo`. This is + // reachable without any client Reconfigure: the IDD capturer autonomously recreates its ring + // on a confirmed display-descriptor change (e.g. a fullscreen game mode-setting the virtual + // display). Refuse instead — the session must reopen the encoder at the new mode. Mirrors + // the guard the QSV and AMF backends already carry. + anyhow::ensure!( + frame.width == self.width && frame.height == self.height, + "pyrowave: captured frame {}x{} != encoder {}x{} (the capturer recreated its ring at a \ + new mode — the encoder must be reopened)", + frame.width, + frame.height, + self.width, + self.height + ); let FramePayload::D3d11(d3d) = &frame.payload else { bail!("pyrowave (Windows) needs a D3D11 frame (the capturer must be in pyrowave mode)") }; @@ -431,6 +460,25 @@ impl PyroWaveEncoder { in pyrowave mode (session_plan::output_format must set OutputFormat::pyrowave)", )?; + // Ring recreate ⇒ every cached plane import belongs to textures that no longer exist. Their + // COM addresses can be handed back out by the allocator, so a pointer-keyed hit could return + // an image bound to freed memory. Flush on the generation change rather than relying on the + // address (or the FIFO cap) to notice. + if self.ring_gen != Some(share.ring_gen) { + if self.ring_gen.is_some() { + tracing::info!( + from = ?self.ring_gen, + to = share.ring_gen, + cached = self.y_images.len() + self.cbcr_images.len(), + "pyrowave: capturer recreated its ring — flushing stale plane imports" + ); + } + for (_, img) in self.y_images.drain(..).chain(self.cbcr_images.drain(..)) { + pw::pyrowave_image_destroy(img); + } + self.ring_gen = Some(share.ring_gen); + } + // Import the fence whenever this encoder has no timeline yet — the first frame, OR a fresh // encoder after a client mode-switch rebuild (the capturer passes the persistent handle on // every frame precisely so a rebuilt encoder can re-import it). @@ -465,7 +513,7 @@ impl PyroWaveEncoder { }; let pw_dev = self.pw_dev; let y_img = { - let key = d3d.texture.as_raw() as isize; + let key = (d3d.texture.as_raw() as isize, w, h); let tex = &d3d.texture; Self::cached_plane( &mut self.y_images, @@ -474,7 +522,7 @@ impl PyroWaveEncoder { )? }; let cbcr_img = { - let key = share.cbcr.as_raw() as isize; + let key = (share.cbcr.as_raw() as isize, cw, ch); let tex = &share.cbcr; Self::cached_plane( &mut self.cbcr_images, @@ -976,6 +1024,9 @@ mod tests { cbcr: cbcr_tex, fence_handle: Some(fence_handle.0 as isize), fence_value: 1, + // One synthetic ring for the whole case: a constant generation exercises the + // steady-state cache-hit path (a changing one would flush every frame). + ring_gen: 1, }), }), cursor: None, diff --git a/crates/pf-encode/src/lib.rs b/crates/pf-encode/src/lib.rs index 292eb9db..7bea9d3c 100644 --- a/crates/pf-encode/src/lib.rs +++ b/crates/pf-encode/src/lib.rs @@ -209,6 +209,11 @@ impl Encoder for TrackedEncoder { fn set_wire_chunking(&mut self, shard_payload: usize) { self.inner.set_wire_chunking(shard_payload) } + // Forwarded for the same reason as `set_wire_chunking` above — an unforwarded default here + // would silently leave the in-place backends pipelining past the capturer's ring. + fn set_input_ring_depth(&mut self, depth: usize) { + self.inner.set_input_ring_depth(depth) + } fn poll(&mut self) -> Result> { self.inner.poll() } diff --git a/crates/pf-frame/src/dxgi.rs b/crates/pf-frame/src/dxgi.rs index 60531695..8db1fbbf 100644 --- a/crates/pf-frame/src/dxgi.rs +++ b/crates/pf-frame/src/dxgi.rs @@ -52,6 +52,12 @@ pub struct PyroFrameShare { /// The fence value the capturer signalled after THIS frame's convert. The encoder's Vulkan /// acquire waits on it, so the wavelet read is ordered after the D3D11 CSC. pub fence_value: u64, + /// The capturer's ring generation, bumped every time it recreates its texture ring. The + /// PyroWave encoder caches its plane imports keyed on the texture's COM address, which carries + /// no reference — after a recreate those addresses can be recycled by the allocator, so a + /// cached import may describe a texture that no longer exists. The encoder flushes its import + /// cache whenever this changes, making cache identity independent of allocator behaviour. + pub ring_gen: u32, } /// A GPU-resident captured texture (the Windows zero-copy path: NVENC/AMF/QSV encode it in place; diff --git a/crates/punktfunk-host/src/native/stream.rs b/crates/punktfunk-host/src/native/stream.rs index 2fdb1390..ef739fff 100644 --- a/crates/punktfunk-host/src/native/stream.rs +++ b/crates/punktfunk-host/src/native/stream.rs @@ -1477,6 +1477,9 @@ pub(super) fn virtual_stream(ctx: SessionContext, prepared: Option