diff --git a/src/ipc/drm.rs b/src/ipc/drm.rs index e4b84e87c..b2ece0c5f 100644 --- a/src/ipc/drm.rs +++ b/src/ipc/drm.rs @@ -1778,6 +1778,17 @@ const MAX_DRM_JSON_BYTES: usize = 8 * 1024 * 1024; /// a full frame on the CPU path (33 MB at 4K) but it crosses a unix socket, so it is milliseconds in /// practice and this only has to bound a peer that stopped. const DRM_BODY_TIMEOUT_MS: u64 = 5_000; +/// Total budget for one wire SEND. The read side has bounded every wait since the beginning, with +/// the argument written at `next_raw_into`: a peer that stops mid-message pins the other end's +/// thread forever on a readiness wait. The WRITE side had no deadline at all, and the asymmetry +/// matters more here, because the parked task is in the ROOT service: a peer that simply stops +/// reading (a `kill -STOP` on its own `--server`, a ptrace stop, a cgroup freeze) leaves the send +/// blocked inside the forward loop, so the loop top is never reached again -- and the loop top is +/// where the credit stall, the per-frame reauthorization and the topology-generation check live. +/// The connection slot, the worker thread and its DRM context stay pinned until the peer chooses to +/// resume. Generous for the same reason as the read budget: this only has to bound a peer that +/// stopped, not pace a healthy one. +const DRM_SEND_TIMEOUT_MS: u64 = 5_000; /// Cap on a raw body read by `next_raw_into` (CPU-fallback BGRA / cursor RGBA). Covers a 256 MiB 8K /// scanout (`DrmReader` bounds a frame to that) with margin. @@ -1904,7 +1915,16 @@ async fn drm_write_all( mut pass_fd: Option, ) -> ResultType<()> { while !buf.is_empty() { - stream.writable().await?; + // Bounded, and a timeout is a hard error rather than a retry: a partial write has already + // desynchronized the framing and cannot be resumed, which is the same argument + // `next_raw_into` makes for the read side. + match timeout(DRM_SEND_TIMEOUT_MS, stream.writable()).await { + Ok(r) => r?, + Err(_) => bail!( + "drm: peer did not accept the remaining {} byte(s) within {DRM_SEND_TIMEOUT_MS}ms; closing", + buf.len() + ), + } let raw = stream.as_raw_fd(); let chunk = buf; let fd_now = pass_fd; @@ -2022,7 +2042,15 @@ impl DrmConn { /// Uses `&self.stream` directly, so it never conflicts with a concurrent `send_msg`/`recv_msg`. pub async fn send_frame_ack(&self) -> ResultType<()> { loop { - self.stream.writable().await?; + // Same bound as drm_write_all, same reason: an unbounded readiness wait lets a wedged + // peer pin this thread. Here the parked side is the consumer, so the cost is one stalled + // stream rather than a root slot, but the shape is identical and so is the fix. + match timeout(DRM_SEND_TIMEOUT_MS, self.stream.writable()).await { + Ok(r) => r?, + Err(_) => bail!( + "drm: _drm frame-ack was not accepted within {DRM_SEND_TIMEOUT_MS}ms; closing" + ), + } match self.stream.try_write(&[1u8]) { Ok(n) if n > 0 => return Ok(()), // A non-empty write returning 0 means the write half is shut down (the producer is @@ -2040,14 +2068,31 @@ impl DrmConn { /// (a single `try_read` that returns WouldBlock). pub fn drain_frame_acks(&self, credit: &mut i32, max: i32) -> ResultType<()> { let mut buf = [0u8; 64]; - loop { + // BOUNDED, and the bound is the point: this is a synchronous loop on the single-threaded + // `_drm` runtime, so "drain until WouldBlock" is a promise the PEER gets to keep. A peer + // that writes a continuous stream instead of one ack byte per frame keeps the receive queue + // non-empty forever, WouldBlock never happens, and this never returns -- pinning the root + // service's runtime thread at 100% CPU and starving every other stream on it, which on a + // multi-monitor client is one connection wedging its own siblings. + // Credit is capped at `max` anyway, so a full budget is reached long before this cap; there + // is nothing to gain by reading further in one call, and whatever is left stays queued for + // the next pass. + const MAX_ACK_READS: usize = 64; + for _ in 0..MAX_ACK_READS { match self.stream.try_read(&mut buf) { Ok(0) => bail!("drm: _drm frame-ack peer closed"), - Ok(n) => *credit = (*credit + n as i32).min(max), + Ok(n) => { + *credit = (*credit + n as i32).min(max); + // Full budget: nothing further to gain from this pass. + if *credit >= max { + return Ok(()); + } + } Err(e) if e.kind() == std::io::ErrorKind::WouldBlock => return Ok(()), Err(e) => return Err(e.into()), } } + Ok(()) } /// Producer: await until the socket is readable (a frame ack arrived, or the peer errored/closed). diff --git a/src/server/drm_capturer.rs b/src/server/drm_capturer.rs index bec944008..1ff05eb00 100644 --- a/src/server/drm_capturer.rs +++ b/src/server/drm_capturer.rs @@ -215,8 +215,9 @@ const RAPID_REBUILD_MAX: u32 = 6; /// waits 30 s, is advertised online again, fails its four sessions in a few seconds and is demoted /// again, which is PeerInfo churn on a ~35 s cycle for as long as the process lives. Doubling turns /// that into a handful of retries and then near-silence, while keeping the property that made the -/// cooldown recoverable in the first place, because the entry is dropped entirely the moment the -/// display delivers a frame (see `frame()`), not decayed by time. +/// cooldown recoverable in the first place, because a delivered frame zeroes the demote count (see +/// `frame()`), not decayed by time. It zeroes ONLY the streak verdicts: the rebuild cadence and the +/// convert verdict survive on purpose, since a first frame says nothing about either. fn demote_cooldown(demotes: u32) -> Duration { DEMOTE_COOLDOWN * (1u32 << demotes.saturating_sub(1).min(DEMOTE_BACKOFF_MAX_SHIFT)) } @@ -509,8 +510,13 @@ impl TraitCapturer for IpcDrmCapturer { // session. Worse, the set happens on the recv thread and this delete on the // encoder thread, so a convert failure racing a queued frame could destroy // the bit inside the very session that learned it. - // Only `drm_clear_prefer_cpu` (on a topology change, where the mapping really - // can have changed) may clear the convert verdict. + // Nothing clears the convert verdict, and that is deliberate: which GPU exports + // a given monitor is a property of the HOST, so the bit is keyed by connector + // identity (`device:name`) and follows that monitor for the process run, exactly + // as its own doc says. A monitor that moves to another GPU arrives under a new + // identity and starts clean, so there is nothing to invalidate on a topology + // change either. (An earlier revision pointed at `drm_clear_prefer_cpu` here; + // that function no longer exists.) self.got_frame = true; if let Some(key) = &self.connector { if let Some(h) = DRM_DISPLAY_HEALTH.lock().unwrap().get_mut(key) { @@ -1872,7 +1878,8 @@ pub(super) fn get_capturer_info( // Refuse a demoted display UNLESS its demotion has aged past the cooldown for its demote // count, in which case clear the failure streak so the display retries DRM (recoverable). // The demote count itself is KEPT, so a display that fails again waits twice as long; only a - // delivered frame erases it (frame() drops the entry outright). + // delivered frame erases it (frame() zeroes `demotes`; it deliberately leaves the rebuild + // cadence and the convert verdict alone). let mut map = DRM_DISPLAY_HEALTH.lock().unwrap(); if let Some(h) = key.as_ref().and_then(|k| map.get_mut(k)) { if h.zero_frame_streak >= DRM_GRAB_MAX_FAILURES { @@ -2031,8 +2038,14 @@ mod drm_capturer_tests { put_frame(&c, 64, 32); assert!(matches!(c.frame(Duration::from_millis(50)), Ok(_))); - let map = DRM_DISPLAY_HEALTH.lock().unwrap(); - let h = map.get(key).expect("the entry must SURVIVE a delivered frame"); + // Copy the fields out and RELEASE the guard before asserting: DRM_DISPLAY_HEALTH is + // process-wide, and a failing assertion while holding it poisons the mutex for every + // sibling test in the binary -- so the one failure this test exists to report would arrive + // buried under unrelated PoisonErrors. `zero_frame_streak_of` above does the same. + let h = { + let map = DRM_DISPLAY_HEALTH.lock().unwrap(); + *map.get(key).expect("the entry must SURVIVE a delivered frame") + }; assert_eq!(h.zero_frame_streak, 0, "a delivered frame refutes the zero-frame streak"); assert_eq!(h.demotes, 0, "and the demotion count that streak drove"); assert_eq!(