From 3c62da3b8eeefa01d01e9ddf4b12ab99fc259f9c Mon Sep 17 00:00:00 2001 From: enricobuehler Date: Tue, 28 Jul 2026 12:52:42 +0200 Subject: [PATCH] fix(zerocopy/client): renegotiation retires the old generation's IPC mappings MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Shared::mappings only ever grew: clear_cache reset sent_keys and told the worker to drop its fd cache, but nothing closed the host-side CUDA IPC mappings for the previous pool generation. A session that renegotiates repeatedly (mode changes, HDR toggles, client reconnects) accumulated a pool's worth of stale mappings each time, each pinning a host VA reservation to peer memory the worker had already freed. Mappings now carry a refcount and a retired flag. clear_cache closes every unreferenced mapping immediately and marks the rest retired; a retired mapping closes when its last in-flight DeviceBuffer releases. The worker's half of the contract: ClearCache also forgets its VA→id map, so anything delivered after the boundary gets a fresh id WITH its descriptor — without that, a same-shape renegotiation (worker keeps its pool) would re-deliver an old id whose host mapping was just closed, and the host would misread it as a desync. Ids never repeat (next_id only counts up), so fresh ids cannot collide with the graveyard. (R6 from design/pf-zerocopy-sweep-handoff.md.) Co-Authored-By: Claude Fable 5 --- crates/pf-zerocopy/src/imp/client.rs | 83 ++++++++++++++++++++++------ crates/pf-zerocopy/src/imp/worker.rs | 12 +++- 2 files changed, 75 insertions(+), 20 deletions(-) diff --git a/crates/pf-zerocopy/src/imp/client.rs b/crates/pf-zerocopy/src/imp/client.rs index 97a82f3d..b25e3510 100644 --- a/crates/pf-zerocopy/src/imp/client.rs +++ b/crates/pf-zerocopy/src/imp/client.rs @@ -35,7 +35,7 @@ const REPLY_TIMEOUT: Duration = Duration::from_secs(10); /// close, which is what tells an idle worker to exit. struct Shared { sock: OwnedFd, - mappings: Mutex>, + mappings: Mutex>, dead: AtomicBool, } @@ -49,18 +49,36 @@ struct Mapping { height: u32, } +/// A [`Mapping`] plus its lifecycle: how many in-flight [`DeviceBuffer`]s still point into it, +/// and whether its pool generation was retired by a renegotiation ([`RemoteImporter::clear_cache`]). +/// Retired-but-referenced entries linger as a graveyard and close when their last frame releases +/// — without this, every renegotiation (mode change, HDR toggle, client reconnect) permanently +/// pinned a pool's worth of host VA reservations to peer memory the worker had already freed. +/// Worker buffer ids are never reused (its `next_id` only counts up), so retired entries can +/// share the map with the next generation's. +struct MapEntry { + m: Mapping, + refs: u32, + retired: bool, +} + impl Drop for Shared { fn drop(&mut self) { - // Last reference gone — no DeviceBuffer can still point into these mappings. - for (_, m) in self.mappings.lock().unwrap().drain() { - cuda::ipc_close(m.y); - if let Some((uv, _)) = m.uv { - cuda::ipc_close(uv); - } + // Last reference gone — no DeviceBuffer can still point into these mappings (current + // generation or graveyard alike). + for (_, e) in self.mappings.lock().unwrap().drain() { + close_mapping(&e.m); } } } +fn close_mapping(m: &Mapping) { + cuda::ipc_close(m.y); + if let Some((uv, _)) = m.uv { + cuda::ipc_close(uv); + } +} + /// 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. @@ -414,19 +432,24 @@ impl RemoteImporter { self.mark_dead(); format!("open CUDA IPC mapping for worker buffer {id}") })?; - self.shared.mappings.lock().unwrap().insert(id, mapping); + self.shared.mappings.lock().unwrap().insert( + id, + MapEntry { + m: mapping, + refs: 0, + retired: false, + }, + ); } - let m = self - .shared - .mappings - .lock() - .unwrap() - .get(&id) - .copied() - .ok_or_else(|| { + let m = { + let mut g = self.shared.mappings.lock().unwrap(); + let entry = g.get_mut(&id).ok_or_else(|| { self.mark_dead(); anyhow::anyhow!("worker delivered unknown buffer id {id} (desync)") })?; + entry.refs += 1; + entry.m + }; let shared = self.shared.clone(); Ok(DeviceBuffer::remote( m.y, @@ -439,8 +462,17 @@ impl RemoteImporter { Box::new(move || { // Fire-and-forget recycle; a dead worker just means EPIPE, ignored. The // captured `shared` Arc is what keeps the mapping + socket alive until - // the last frame drops. + // the last frame drops. A retired mapping (its generation renegotiated + // away) closes here with its last reference. let _ = proto::send(shared.sock.as_fd(), &Request::Release { id }, None); + let mut g = shared.mappings.lock().unwrap(); + if let Some(entry) = g.get_mut(&id) { + entry.refs = entry.refs.saturating_sub(1); + if entry.retired && entry.refs == 0 { + let entry = g.remove(&id).expect("entry exists"); + close_mapping(&entry.m); + } + } }), )) } @@ -452,9 +484,24 @@ impl RemoteImporter { } } - /// The PipeWire stream renegotiated — reset both sides' per-buffer caches. + /// The PipeWire stream renegotiated — reset both sides' per-buffer caches, and retire the + /// outgoing generation's CUDA IPC mappings: the worker replaces its pool, so these host-side + /// mappings pin VA reservations to peer memory that is about to be (or already was) freed. + /// Unreferenced ones close now; ones still under an in-flight frame close with its release. pub fn clear_cache(&mut self) { self.sent_keys.clear(); + { + let mut g = self.shared.mappings.lock().unwrap(); + g.retain(|_, entry| { + if entry.refs == 0 { + close_mapping(&entry.m); + false + } else { + entry.retired = true; + true + } + }); + } if !self.dead() { if let Err(e) = proto::send(self.shared.sock.as_fd(), &Request::ClearCache, None) { tracing::warn!(error = %e, "zerocopy worker ClearCache failed"); diff --git a/crates/pf-zerocopy/src/imp/worker.rs b/crates/pf-zerocopy/src/imp/worker.rs index 5e51ef26..4168b67d 100644 --- a/crates/pf-zerocopy/src/imp/worker.rs +++ b/crates/pf-zerocopy/src/imp/worker.rs @@ -194,8 +194,9 @@ struct EglBackend { inflight: HashMap, /// Buffer id per device allocation. Valid only within one pool generation: pools never free /// allocations while alive, so a device VA can't repeat until a size change replaces the pool - /// — at which point [`Self::note_dims`] clears this map (ids themselves are never reused; - /// `next_id` only counts up). + /// — at which point [`Self::note_dims`] clears this map, as does `ClearCache` (the host + /// retires its IPC mappings on that boundary). Ids themselves are never reused; `next_id` + /// only counts up. ids: HashMap, next_id: u32, /// The (kind, width, height) of the last import — a change means the importer replaced its @@ -282,6 +283,13 @@ impl ImportBackend for EglBackend { } self.fd_lru.clear(); self.importer.clear_linear_cache(); + // ClearCache is the generation boundary the HOST retires its CUDA IPC mappings on: it + // closes every mapping the renegotiated-away generation opened (keeping only ones still + // under an in-flight frame, until they release). Forget the VA→id map so anything + // delivered after this point gets a fresh id WITH its descriptor — re-delivering an old + // id would reference a mapping the host just closed. Ids never repeat (`next_id` only + // counts up), so fresh ids can't collide with the host's graveyard. + self.ids.clear(); } }