TheLanc3/rustdesk
1
0
Fork 0
mirror of https://github.com/rustdesk/rustdesk.git synced 2026-09-05 23:51:04 +03:00
Code Issues Packages Projects Releases Wiki Activity

drm: fix three defects in the frame credit gate

Browse Source 
Follow-up to the previous commit, from an adversarial review of it.

- The ack wake-up skipped the coalescing drain. When the socket arm of the
  select won, there was no message to seed the drain loop with, so the channel
  was never polled that iteration: a held frame could be sent while a strictly
  newer one already sat queued, and a queued cursor waited for the next producer
  message. Seed the loop from the channel when we woke on an ack instead.
- The loop could wait while holding a frame it was allowed to send. Credit
  replenished by the top-of-loop drain was not consulted before entering the
  select, so the frame waited for the worker's next message; if capture then
  returned WouldBlock it sat there until the stall teardown. Take whatever is
  queued without blocking in that case and fall through to the send.
- The capture worker no longer had any backpressure. Draining the channel every
  iteration (needed so cursors keep flowing) means a full channel no longer
  parks it, so a consumer converting at a fraction of the capture rate made the
  privileged service keep grabbing frames that were then discarded -- a packed
  copy per frame on the CPU path, a PRIME export on the dma-buf path. The worker
  now skips the grab while the task is holding an undeliverable frame, and keeps
  polling the cursor so the remote pointer stays live. The gate is deliberately
  conditioned on holding a frame, not merely on having no credit: with nothing
  held the task blocks in recv() and cannot observe an ack, so gating there
  would stop the worker feeding it at all.

The comment claiming the bounded channel backpressures the worker is corrected.
This commit is contained in:
Mariano Abad
2026-07-23 22:30:02 -03:00
parent 0b30ea7203
commit b201c7c7e6
1 changed files with 53 additions and 20 deletions
Download Patch File Download Diff File Expand all files Collapse all files

73
src/ipc.rs
View File

@@ -2093,7 +2093,12 @@ async fn handle_drm_conn(stream: Connection) -> ResultType<()> {
let stop = Arc::new(AtomicBool::new(false));
let _stop_guard = DrmStopGuard(stop.clone());
let worker_stop = stop.clone();
std::thread::spawn(move || drm_capture_worker(frame_tx, crtc_rx, worker_stop));
// Set while the task is holding a frame it has no send credit for. The worker then skips the
// scanout grab, which the task would only discard, instead of burning CPU (and a PRIME export
// on the dma-buf path) inside the privileged service for a consumer that is behind.
let frames_gated = Arc::new(AtomicBool::new(false));
let worker_gate = frames_gated.clone();
std::thread::spawn(move || drm_capture_worker(frame_tx, crtc_rx, worker_stop, worker_gate));
// Handshake: the worker sends the display list (from the pre-warmed cache, or a throwaway
// enumeration open if the cache is empty). A closed channel (no Displays) means the reader was
@@ -2155,8 +2160,9 @@ async fn handle_drm_conn(stream: Connection) -> ResultType<()> {
// Flow control (review P1-2): allow at most DRM_FRAME_CREDIT frames in flight on the socket. The
// consumer acks each converted frame (send_frame_ack) and the producer only sends while it has
// credit, so a slow convert bounds the socket FIFO to a couple of frames instead of accumulating
// seconds of stale descriptors (a permanently-behind desktop). The bounded frame_rx backpressures
// the capture worker while we wait for an ack. Cursors and topology updates are not credit-gated.
// seconds of stale descriptors (a permanently-behind desktop). Backpressure on the capture
// worker comes from `frames_gated` (we keep draining the channel for cursors, so a full channel
// no longer stalls it). Cursors and topology updates are not credit-gated.
const DRM_FRAME_CREDIT: i32 = 2;
let mut credit: i32 = DRM_FRAME_CREDIT;
// The newest frame produced while credit was exhausted. Holding it here (latest-wins, exactly
@@ -2167,18 +2173,31 @@ async fn handle_drm_conn(stream: Connection) -> ResultType<()> {
loop {
// Replenish credit from any acks the consumer has finished. Also detects a closed peer.
conn.drain_frame_acks(&mut credit, DRM_FRAME_CREDIT)?;
// Pause the worker's grab only while we are ALREADY holding an unsendable frame. Gating on
// "no credit" alone would deadlock: with nothing held we block in `frame_rx.recv()` below,
// which cannot observe an ack, so a gated worker would stop feeding us entirely. Holding a
// frame is exactly the state in which we also wait on the socket, so acks still wake us.
frames_gated.store(held_frame.is_some() && credit <= 0, Ordering::Relaxed);
// Wait for the next producer message. While a frame is held back we watch the socket too,
// so an arriving ack wakes us promptly instead of only when the next frame shows up; that
// wake yields no message and simply falls through to the send decision below. Both arms are
// cancel-safe (`mpsc::Receiver::recv`, and `wait_readable` is readiness-only).
let first: Option<DrmProducerMsg> = if held_frame.is_some() {
tokio::select! {
biased;
r = conn.wait_readable() => { r?; None }
m = frame_rx.recv() => match m {
Some(m) => Some(m),
None => break,
},
if credit > 0 {
// The drain above already replenished us and we are holding a sendable frame:
// never block here. Waiting would park a frame we are allowed to send right now
// until the worker happens to produce another one -- and if capture then returns
// WouldBlock it would sit there until the stall teardown.
frame_rx.try_recv().ok()
} else {
tokio::select! {
biased;
r = conn.wait_readable() => { r?; None }
m = frame_rx.recv() => match m {
Some(m) => Some(m),
None => break,
},
}
}
} else {
match frame_rx.recv().await {
@@ -2220,7 +2239,10 @@ async fn handle_drm_conn(stream: Connection) -> ResultType<()> {
// (latest-wins by id downstream), so they are forwarded in order and never coalesced away.
// Start from any frame held back for credit, so a newer one supersedes it (latest-wins).
let mut latest_frame: Option<DrmProducerMsg> = held_frame.take();
let mut msg = first;
// When we woke on an ack rather than on a message, `first` is None; seed from the channel
// anyway so anything queued meanwhile still supersedes the held frame and a queued cursor
// still goes out this iteration instead of waiting for the next producer message.
let mut msg = first.or_else(|| frame_rx.try_recv().ok());
while let Some(m) = msg.take() {
match m {
f @ (DrmProducerMsg::Frame { .. } | DrmProducerMsg::FrameCpu { .. }) => {
@@ -2295,6 +2317,7 @@ fn drm_capture_worker(
frame_tx: tokio::sync::mpsc::Sender<DrmProducerMsg>,
crtc_rx: std::sync::mpsc::Receiver<(u32, bool)>,
stop: std::sync::Arc<std::sync::atomic::AtomicBool>,
frames_gated: std::sync::Arc<std::sync::atomic::AtomicBool>,
) {
use std::sync::atomic::Ordering;
use std::time::Duration;
@@ -2373,8 +2396,16 @@ fn drm_capture_worker(
// Grab one frame in the current mode, producing an OWNED message (no borrow of `reader`
// outlives this, so `reader.cursor()` below is free to run). The dma-buf path ships only the
// descriptor + fd; the CPU path copies the packed BGRA once (Bytes::copy_from_slice).
let grabbed: std::io::Result<DrmProducerMsg> = if use_dmabuf {
match reader.grab_desc() {
let grabbed: Option<std::io::Result<DrmProducerMsg>> = if frames_gated.load(Ordering::Relaxed)
{
// The task is holding a frame it has no credit to send, so anything grabbed now would
// only be discarded. Skip the scanout work -- and, on the dma-buf path, a PRIME export
// -- rather than spend it in this privileged process for a consumer that is behind. The
// cursor poll below still runs so the remote pointer stays live, and `stalled` is
// deliberately left untouched: the device is healthy, we are idle by design.
None
} else if use_dmabuf {
Some(match reader.grab_desc() {
Ok((fd, d)) => Ok(DrmProducerMsg::Frame {
desc: DmabufDesc {
buffer_id: (d.fb_id as u64) | ((conn_epoch as u64) << 32),
@@ -2393,19 +2424,21 @@ fn drm_capture_worker(
fd: Some(fd),
}),
Err(err) => Err(err),
}
})
} else {
match reader.grab() {
Some(match reader.grab() {
Ok((buf, w, h)) => Ok(DrmProducerMsg::FrameCpu {
width: w as u32,
height: h as u32,
data: Bytes::copy_from_slice(buf),
}),
Err(err) => Err(err),
}
})
};
match grabbed {
Ok(msg) => {
// Gated: no frame work this tick, fall through to the cursor poll below.
None => {}
Some(Ok(msg)) => {
stalled = 0;
if !logged_first {
logged_first = true;
@@ -2419,7 +2452,7 @@ fn drm_capture_worker(
break;
}
}
Err(err) if err.kind() == std::io::ErrorKind::WouldBlock => {
Some(Err(err)) if err.kind() == std::io::ErrorKind::WouldBlock => {
stalled += 1;
if stalled > MAX_STALLED {
log::info!("drm: capture stalled (no frame); closing _drm connection");
@@ -2428,7 +2461,7 @@ fn drm_capture_worker(
std::thread::sleep(FRAME_INTERVAL);
continue;
}
Err(err) if use_dmabuf && err.kind() == std::io::ErrorKind::Unsupported => {
Some(Err(err)) if use_dmabuf && err.kind() == std::io::ErrorKind::Unsupported => {
// The split export cannot work on this seat/driver (ENOTSUP). Switch this connection
// to the CPU-mapped fallback (pixels over the wire) instead of tearing down or
// rebuild-looping; the reader is already open and usable via grab().
@@ -2439,7 +2472,7 @@ fn drm_capture_worker(
logged_first = false;
continue;
}
Err(err) => {
Some(Err(err)) => {
log::warn!("drm: capture error: {err}; closing _drm connection");
break;
}