fix(drm): pace the producer to the rate the encoder takes, with the panel refresh as floor (#16423)

* fix(drm): pace the producer to the rate the encoder takes, with the panel refresh as floor

The root producer slept a fixed 33 ms after every grab, so every DRM stream topped
out at 30 fps whatever the client asked (rustdesk#16175). Measured on a
1920x1080@60 i915 box with the client at Custom 60 and QoS settled at 60: 30
grabs, 30 converted, 29.8 sent. A shorter fixed interval reaches the rate but
oversamples: at 8 ms the same box grabbed and converted 117/s for 60 sent.

The consumer now tells the producer the rate it encodes at, in the ack byte it
already sends per converted frame: 0x01 as before, or 0x80|fps. Every producer
counts ack bytes and never read their value, so an older service sees exactly
the same credit and keeps its 33 ms; the JSON handshake and the ack count and
timing do not change. The rate is the encoder's spf (what frame() is called
with), lowered to what the encoder achieves when it is busy longer than spf per
frame: the time from one frame() call to the next, less what the call spent
waiting for a frame, so a late producer cannot talk its own demand down.

The producer replaces the post-work sleep with a deadline: the encoder's period
above 30 fps, the old 33 ms from 30 down to 15, and half the encoder's period
below 15, never faster than the panel (the exact CRTC refresh from libdrmtap
0.5.9's drmtap_crtc_refresh, re-read once a second; with an older library the
whole-hertz refresh the enumeration already read, so no extra connector probe).
The worker still wakes at least every 33 ms, because each tick also reads the
cursor and a slow frame rate must not slow the cursor down. Without a demand (an
older consumer, or before the first frame() call) the old 33 ms sleep stays.
MAX_STALLED and the WouldBlock retry keep their 5 s budget, pinned by a const
assertion.

Tests: the ack byte round trip and that it is one byte on the wire, the drain
(seeded byte streams: every byte one credit, the newest demand wins), the pacing
interval table (panel floor, below 30 and below 15, a 23.976 Hz panel), the panel
period rounded up, the deadline and its overrun, the wake that keeps the cursor
at 33 ms, the demand from spf and from a busy encoder, the busy time without the
wait (pure and through frame()), the demand log once per connection, the
whole-hertz fallback, and the reader's refresh answer. Each rule has a mutation
that turns its test red.

* ci: run the drm pacing tests in the drm step

drm_pacing_tests (src/ipc/drm.rs) and crtc_refresh_tests (libs/scrap
drm_reader.rs) match none of the existing filters.

* test(drm): keep three tests, on outputs and not on internal state

Per AGENTS.md (Tests), keep one to three tests, on outputs and not on internal
state: the demand ack as it goes on the wire, the producer interval for a
demand and a panel, and the exact refresh the reader returns.

* test(drm): the ack test goes through the producer drain

Per review: send_frame_ack(60) read back through drain_frame_acks gives one
credit and a demand of 60, so dropping the demand read in the producer turns
the test red. The interval test also pins drm_panel_period(148352, 2475) to
16_683_294 ns.

* fix(drm): time the encoder only after a frame() call that returned a frame

Since #16450 frame() drops a repeat as WouldBlock. The time after such a
call is video_service sleeping out its spf, and after a timeout it is
about zero once the wait is subtracted. Neither follows an encode, so
neither is a sample of the encoder pace now. Uses returned_at from #16450;
no new state.

* fix(drm): ask for the old 33 ms tick while the screen is still

Since #16450 a still screen is grabbed at the demand and dropped in
frame(), twice as often as master at 60. After 100 ms without a new
frame the demand is capped at 30, the old producer tick, and the next
new frame lifts it. 100 ms is three of those ticks, so a producer paced
at 30 in motion never reads as still. Uses returned_at; no new state.

* test(drm): a dropped repeat or timeout is no encoder time; a still screen asks for the old tick

Per review: the demand frame() reports stays with a 50 ms encoder through
a still screen of repeats and timeouts, and a still screen asks for 30
after STILL_AFTER until a new frame brings back 60.

* fix(drm): start the encoder average over when the rate goes up

Per review: the samples hold the sleep video_service paces itself with, so
after a rate increase (custom fps 1 -> 60, or QoS recovering) an average
taken at the old rate kept the demand low and the producer slow. A
shorter spf than the last now starts the average over and drops the
sample that spans the change. The demand is computed in one method that
takes its time as input, the still cap included.

* test(drm): drop the two pacing tests of the last round

Per review: they read shared.demand_fps and slept for real. The three
tests stay.

* build: pin libdrmtap 0.5.12 (rustdesk-org/libdrmtap 5936a7a88)

The panel floor of the pacing reads drmtap_crtc_refresh(), added in
libdrmtap 0.5.9. The mirror carries v0.5.12 since 2026-10-13, so the pin
moves there.

The rest of the range from 0.5.8 changes no struct in drmtap.h: 0.5.10
reads its environment through secure_getenv and loads no GL in a setuid,
setgid or file-capability process, which rustdesk does not run; 0.5.11
makes drmtap_deswizzle, also the CPU fallback of a tiled grab, refuse a
destination stride narrower than its rows instead of writing past it,
and exports drmtap_frame_owns_data, unused here; 0.5.12 changes only
docs and the version.

.github/workflows/flutter-build.yml

@@ -2344,10 +2344,14 @@ jobs:
               cargo test --locked --lib --features drm peer_abs_sample_tests
             run_drm_tests "rustdesk input_map_tests" \
               cargo test --locked --lib --features drm input_map_tests
+            run_drm_tests "rustdesk drm_pacing_tests" \
+              cargo test --locked --lib --features drm drm_pacing_tests
             run_drm_tests "scrap hotspot_provenance_tests" \
               cargo test --locked -p scrap --features drm hotspot_provenance_tests
             run_drm_tests "scrap hotspot_guess_tests" \
               cargo test --locked -p scrap --features drm hotspot_guess_tests
+            run_drm_tests "scrap crtc_refresh_tests" \
+              cargo test --locked -p scrap --features drm crtc_refresh_tests
             run_drm_tests "scrap plane_rotation_tests" \
               cargo test --locked -p scrap --features drm plane_rotation_tests
             cargo build --locked --lib --features "$FEATURES" --release

build.py

@@ -390,9 +390,9 @@ def ffi_bindgen_function_refactor():
 # The commit is fetched directly by sha, so no branch or tag name takes part in the build: see
 # build_libdrmtap_so(). This is the SINGLE source of truth for the pin, deliberately not duplicated in
 # any workflow, so a bump is one edit here (plus the informational version comment in
-# libs/scrap/Cargo.toml). This commit is libdrmtap v0.5.8.
+# libs/scrap/Cargo.toml). This commit is libdrmtap v0.5.12.
 LIBDRMTAP_REPO_PINNED = 'https://github.com/rustdesk-org/libdrmtap'
-LIBDRMTAP_SHA_PINNED = '95d4d74549631aa5c39461300acfd2e106583cc9'
+LIBDRMTAP_SHA_PINNED = '5936a7a88b94aa43516a3dd0eb509d2383f31e2d'
 LIBDRMTAP_REPO = os.environ.get('DRMTAP_REPO', LIBDRMTAP_REPO_PINNED)
 LIBDRMTAP_SHA = os.environ.get('DRMTAP_SHA', LIBDRMTAP_SHA_PINNED)
 # Every way of getting a different .so than the pin needs the same explicit opt-in. Otherwise the

libs/scrap/Cargo.toml

@@ -14,7 +14,7 @@ wayland = ["gstreamer", "gstreamer-app", "gstreamer-video", "dbus", "tracing", "
 # `drm` is a pure runtime-dlopen backend: rustdesk loads `libdrmtap.so.0` at runtime (`drmtap_dl.rs`)
 # and NEVER link-time links it, so the graceful PipeWire fallback when the .so or EGL is absent is
 # preserved and the drm build pulls in no libdrm/seccomp/cap/EGL link-time deps. The .so is pinned by
-# `DRMTAP_SHA` in build.py, which fetches that exact commit (libdrmtap v0.5.8). We deliberately do
+# `DRMTAP_SHA` in build.py, which fetches that exact commit (libdrmtap v0.5.12). We deliberately do
 # NOT depend on the `libdrmtap-sys` crate: its build.rs statically compiles the whole libdrmtap C tree
 # and a CAP_SYS_ADMIN helper and emits `-ldrm -lseccomp -lcap`, which would defeat the dlopen model.
 # Depends on `wayland`: the three drm modules live inside the `#[cfg(feature = "wayland")]` arm of

libs/scrap/src/common/drm_reader.rs

@@ -309,6 +309,8 @@ pub struct DisplaySnapshot {
     pub y: i32,
     pub width: u32,
     pub height: u32,
+    /// The kernel's whole-hertz vrefresh: 59.94 reads 60. `DrmReader::crtc_refresh` is exact.
+    pub refresh_hz: u32,
     pub active: bool,
 }
 
@@ -545,6 +547,17 @@ impl DrmReader {
         self.last_plane_rotation
     }
 
+    /// The refresh of the captured CRTC in hertz as the reduced fraction `(num, den)` of its
+    /// current mode, read now. `None` when the library predates the call (0.5.9) or the CRTC has
+    /// no mode.
+    pub fn crtc_refresh(&mut self) -> Option<(u64, u64)> {
+        let f = self.lib.crtc_refresh?;
+        let (mut num, mut den) = (0u64, 0u64);
+        // SAFETY: self.ctx is a live context and both outputs outlive the call.
+        let rc = unsafe { f(self.ctx, &mut num, &mut den) };
+        (rc == 0 && num != 0 && den != 0).then_some((num, den))
+    }
+
     fn read_plane_rotation(&mut self) -> Option<u32> {
         let f = self.lib.plane_rotation?;
         let mut rotation: u32 = 0;
@@ -761,6 +774,7 @@ impl DrmReader {
                         y: raw[i].y as i32,
                         width: raw[i].width,
                         height: raw[i].height,
+                        refresh_hz: raw[i].refresh_hz,
                         active: raw[i].active != 0,
                     }
                 })
@@ -1108,3 +1122,63 @@ mod plane_rotation_tests {
         assert_eq!(r.plane_rotation(), Some(0x8), "dma-buf export");
     }
 }
+
+#[cfg(test)]
+mod crtc_refresh_tests {
+    use super::*;
+    use std::sync::atomic::{AtomicI32, AtomicU64, Ordering};
+
+    static RC: AtomicI32 = AtomicI32::new(0);
+    static NUM: AtomicU64 = AtomicU64::new(0);
+    static DEN: AtomicU64 = AtomicU64::new(0);
+
+    unsafe extern "C" fn crtc_refresh(_: *mut drmtap_ctx, num: *mut u64, den: *mut u64) -> c_int {
+        *num = NUM.load(Ordering::SeqCst);
+        *den = DEN.load(Ordering::SeqCst);
+        RC.load(Ordering::SeqCst)
+    }
+    unsafe extern "C" fn grab_mapped(_: *mut drmtap_ctx, _: *mut drmtap_frame_info) -> c_int {
+        -1
+    }
+    unsafe extern "C" fn frame_release(_: *mut drmtap_ctx, _: *mut drmtap_frame_info) {}
+    unsafe extern "C" fn grab_desc(
+        _: *mut drmtap_ctx,
+        _: *mut drmtap_dmabuf_desc,
+        _: *mut drmtap_frame_info,
+    ) -> c_int {
+        -1
+    }
+    unsafe extern "C" fn plane_rotation(_: *mut drmtap_ctx, _: *mut u32) -> c_int {
+        -1
+    }
+
+    fn reader(with_call: bool) -> DrmReader {
+        let mut lib = DrmtapLib::fake(grab_mapped, frame_release, grab_desc, plane_rotation);
+        if with_call {
+            lib.crtc_refresh = Some(crtc_refresh);
+        }
+        DrmReader {
+            lib: Box::leak(Box::new(lib)),
+            ctx: std::ptr::null_mut(),
+            buf: Vec::new(),
+            last_provenance: -1,
+            last_plane_rotation: None,
+        }
+    }
+
+    /// rustdesk#16175: the exact refresh paces the producer; an older library has no answer.
+    #[test]
+    fn the_crtc_refresh_is_a_fraction_or_nothing() {
+        assert_eq!(reader(false).crtc_refresh(), None, "a library before 0.5.9");
+        let mut r = reader(true);
+        RC.store(0, Ordering::SeqCst);
+        NUM.store(148352, Ordering::SeqCst);
+        DEN.store(2475, Ordering::SeqCst);
+        assert_eq!(r.crtc_refresh(), Some((148352, 2475)));
+        RC.store(-hbb_common::libc::ENODATA, Ordering::SeqCst);
+        assert_eq!(r.crtc_refresh(), None, "a CRTC with no mode");
+        RC.store(0, Ordering::SeqCst);
+        DEN.store(0, Ordering::SeqCst);
+        assert_eq!(r.crtc_refresh(), None, "a zero denominator is no answer");
+    }
+}

libs/scrap/src/common/drmtap_dl.rs

@@ -146,6 +146,10 @@ type FnRenderNode = unsafe extern "C" fn(*mut drmtap_ctx) -> *const c_char;
 /// means the compositor can only have rotated in software; `-ENOENT` with no plane bound;
 /// `-EINVAL` on a null argument.
 type FnPlaneRotation = unsafe extern "C" fn(*mut drmtap_ctx, *mut u32) -> c_int;
+/// `drmtap_crtc_refresh`, added in libdrmtap 0.5.9: the refresh of the captured CRTC in hertz as
+/// the reduced fraction `*num / *den` of its current mode. 0 on success; `-ENODATA` for a CRTC
+/// with no mode; `-ENOENT` with none to pick; `-EINVAL` on a null argument.
+type FnCrtcRefresh = unsafe extern "C" fn(*mut drmtap_ctx, *mut u64, *mut u64) -> c_int;
 type FnConvertDmabuf =
     unsafe extern "C" fn(*mut drmtap_ctx, *const drmtap_dmabuf_desc, *mut drmtap_frame_info) -> c_int;
 
@@ -171,6 +175,9 @@ pub struct DrmtapLib {
     /// Optional: it only exists from libdrmtap 0.5.8. Absent means the library cannot say whether
     /// the plane rotated the scanout, and the consumer keeps the pre-0.5.8 rule for that case.
     pub plane_rotation: Option<FnPlaneRotation>,
+    /// Optional: it only exists from libdrmtap 0.5.9. Absent means the producer takes the
+    /// whole-hertz refresh of the enumeration as its pacing floor.
+    pub crtc_refresh: Option<FnCrtcRefresh>,
     pub version: (c_int, c_int, c_int),
 }
 
@@ -221,6 +228,7 @@ impl DrmtapLib {
             convert_dmabuf,
             render_node: None,
             plane_rotation: Some(plane_rotation),
+            crtc_refresh: None,
             version: (0, 5, 8),
         }
     }
@@ -353,6 +361,8 @@ impl DrmtapLib {
                 lib.get(b"drmtap_cursor_hotspot_valid").ok().map(|s| *s);
             let plane_rotation: Option<FnPlaneRotation> =
                 lib.get(b"drmtap_plane_rotation").ok().map(|s| *s);
+            let crtc_refresh: Option<FnCrtcRefresh> =
+                lib.get(b"drmtap_crtc_refresh").ok().map(|s| *s);
             // Log the load only now that every required symbol resolved: this fn still returns None on a missing one.
             let loaded_from = real
                 .as_ref()
@@ -404,6 +414,14 @@ impl DrmtapLib {
                      software will be captured upside down."
                 );
             }
+            if (minor, patch) >= (5, 9) && crtc_refresh.is_none() {
+                log::warn!(
+                    "libdrmtap at {loaded_from} reports v{major}.{minor}.{patch} but is missing \
+                     drmtap_crtc_refresh: it is a stale or pre-release build. Check what the \
+                     soname symlink points at. Capture paces to the whole-hertz refresh, so a \
+                     59.94 Hz panel is sampled as if it were 60."
+                );
+            }
             Some(DrmtapLib {
                 _lib: lib,
                 open,
@@ -420,6 +438,7 @@ impl DrmtapLib {
                 convert_dmabuf,
                 render_node,
                 plane_rotation,
+                crtc_refresh,
                 version: (major, minor, patch),
             })
         }

src/ipc/drm.rs

@@ -264,6 +264,11 @@ static DRM_DISPLAY_CACHE: std::sync::Mutex<Vec<DrmDisplayInfo>> = std::sync::Mut
 /// Bumped only when a change altered `DRM_DISPLAY_CACHE`; Release orders it after the cache write.
 static DRM_DISPLAY_GENERATION: std::sync::atomic::AtomicU64 = std::sync::atomic::AtomicU64::new(0);
 
+/// The whole-hertz refresh of every driven CRTC the last enumeration saw, by (device, crtc), for
+/// a producer whose library cannot read the exact one: it then needs no connector probe of its own.
+static DRM_WHOLE_HZ: std::sync::Mutex<std::collections::BTreeMap<(String, u32), u32>> =
+    std::sync::Mutex::new(std::collections::BTreeMap::new());
+
 /// Displays this reader serves, plus the identity (`device:connector`) of each undriven output.
 fn drm_displays_from_reader(
     reader: &mut scrap::drm_reader::DrmReader,
@@ -271,8 +276,13 @@ fn drm_displays_from_reader(
 ) -> (Vec<DrmDisplayInfo>, Vec<String>) {
     let render_node = reader.render_node().unwrap_or_default();
     let mut undriven = Vec::new();
-    let displays: Vec<DrmDisplayInfo> = reader
-        .displays()
+    let snapshots = reader.displays();
+    if let Ok(mut hz) = DRM_WHOLE_HZ.lock() {
+        for d in snapshots.iter().filter(|d| d.active && d.crtc_id != 0) {
+            hz.insert((device.to_owned(), d.crtc_id), d.refresh_hz);
+        }
+    }
+    let displays: Vec<DrmDisplayInfo> = snapshots
         .into_iter()
         // Only outputs bound to a CRTC: a CONNECTED-but-unbound connector enumerates with
         // `crtc_id == 0`, and `open(crtc=0)` auto-selects the FIRST ACTIVE CRTC and streams ITS frames.
@@ -967,9 +977,20 @@ async fn handle_drm_conn(stream: Connection) -> ResultType<()> {
     let worker_stop = stop.clone();
     let frames_gated = Arc::new(AtomicBool::new(false));
     let worker_gate = frames_gated.clone();
+    let demand_fps = Arc::new(std::sync::atomic::AtomicU8::new(0));
+    let worker_demand = demand_fps.clone();
     std::thread::Builder::new()
         .name("drm-capture".into())
-        .spawn(move || drm_capture_worker(frame_tx, crtc_rx, worker_stop, worker_gate, wake))
+        .spawn(move || {
+            drm_capture_worker(
+                frame_tx,
+                crtc_rx,
+                worker_stop,
+                worker_gate,
+                worker_demand,
+                wake,
+            )
+        })
         .map_err(|err| anyhow::anyhow!("could not spawn the drm capture worker: {err}"))?;
 
     let displays = match frame_rx.recv().await {
@@ -1009,10 +1030,13 @@ async fn handle_drm_conn(stream: Connection) -> ResultType<()> {
     let mut seen_gen = DRM_DISPLAY_GENERATION.load(Ordering::Acquire);
     const DRM_FRAME_CREDIT: i32 = 2;
     let mut credit: i32 = DRM_FRAME_CREDIT;
+    let mut demand: u8 = 0;
+    let mut demand_logged: Option<(u8, std::time::Instant)> = None;
     let mut credit_since = std::time::Instant::now();
     let mut held_frame: Option<DrmProducerMsg> = None;
     loop {
-        conn.drain_frame_acks(&mut credit, DRM_FRAME_CREDIT)?;
+        conn.drain_frame_acks(&mut credit, DRM_FRAME_CREDIT, &mut demand)?;
+        note_drm_demand(&demand_fps, demand, target_crtc, &mut demand_logged);
         // While gated the worker does not grab, so it cannot advance its own MAX_STALLED watchdog: a
         // consumer that stops acking without closing the socket would otherwise hold this connection,
         // its worker thread and the privileged DRM context open indefinitely.
@@ -1104,7 +1128,8 @@ async fn handle_drm_conn(stream: Connection) -> ResultType<()> {
             }
             msg = frame_rx.try_recv().ok();
         }
-        conn.drain_frame_acks(&mut credit, DRM_FRAME_CREDIT)?;
+        conn.drain_frame_acks(&mut credit, DRM_FRAME_CREDIT, &mut demand)?;
+        note_drm_demand(&demand_fps, demand, target_crtc, &mut demand_logged);
         if credit <= 0 {
             held_frame = latest_frame;
             continue;
@@ -1149,18 +1174,109 @@ async fn handle_drm_conn(stream: Connection) -> ResultType<()> {
     Ok(())
 }
 
+/// The producer's tick when the consumer has not said its rate, and the WouldBlock retry period.
+const FRAME_INTERVAL: std::time::Duration = std::time::Duration::from_millis(33);
+/// Bound continuous no-frame (WouldBlock) time so a wedged device ends the stream (~5 s).
+const MAX_STALLED: u32 = 150;
+const _: () = assert!(
+    FRAME_INTERVAL.as_millis() * MAX_STALLED as u128 >= 4_500
+        && FRAME_INTERVAL.as_millis() * MAX_STALLED as u128 <= 5_500
+);
+/// A frame ack is one byte and every producer counts bytes, never their value, so a byte that
+/// also carries the consumer's encode rate is still exactly one credit to an older producer.
+const DRM_ACK_PLAIN: u8 = 0x01;
+const DRM_ACK_FPS_FLAG: u8 = 0x80;
+/// video_qos's ceiling; it fits the seven low bits.
+const DRM_MAX_DEMAND_FPS: u8 = 120;
+
+fn drm_ack_byte(fps: u8) -> u8 {
+    if fps == 0 {
+        DRM_ACK_PLAIN
+    } else {
+        DRM_ACK_FPS_FLAG | fps.min(DRM_MAX_DEMAND_FPS)
+    }
+}
+
+fn drm_ack_demand(b: u8) -> Option<u8> {
+    let fps = b & !DRM_ACK_FPS_FLAG;
+    (b & DRM_ACK_FPS_FLAG != 0 && (1..=DRM_MAX_DEMAND_FPS).contains(&fps)).then_some(fps)
+}
+
+/// One period of a panel refreshing at `num/den` Hz, rounded up so pacing never outruns it.
+fn drm_panel_period(num: u64, den: u64) -> std::time::Duration {
+    if num == 0 || den == 0 {
+        return std::time::Duration::ZERO;
+    }
+    let nanos = (u128::from(den) * 1_000_000_000).div_ceil(u128::from(num));
+    std::time::Duration::from_nanos(u64::try_from(nanos).unwrap_or(u64::MAX))
+}
+
+/// Grab period for a consumer encoding at `demand_fps`, never faster than the panel: its own
+/// period above 30 fps, the old 33 ms tick from 30 down to 15, and below 15 half its period, so a
+/// slow consumer's frame is at most half its period old.
+fn drm_pacing_interval(demand_fps: u8, panel: std::time::Duration) -> std::time::Duration {
+    if demand_fps == 0 {
+        return FRAME_INTERVAL;
+    }
+    let want = std::time::Duration::from_secs(1) / u32::from(demand_fps);
+    want.min((want / 2).max(FRAME_INTERVAL)).max(panel)
+}
+
+/// When the worker's next tick starts. Without a demand it is the old sleep after the work;
+/// with one it is a deadline phased on the previous tick, and `now` after an overrun.
+fn drm_next_tick(
+    prev: std::time::Instant,
+    now: std::time::Instant,
+    demand_fps: u8,
+    panel: std::time::Duration,
+) -> std::time::Instant {
+    if demand_fps == 0 {
+        return now + FRAME_INTERVAL;
+    }
+    (prev + drm_pacing_interval(demand_fps, panel)).max(now)
+}
+
+/// Hands the consumer's rate to the worker. The measured part of it moves with almost every frame:
+/// this connection logs its first rate, then a different one at most every ten seconds.
+fn note_drm_demand(
+    shared: &std::sync::atomic::AtomicU8,
+    demand: u8,
+    crtc: u32,
+    logged: &mut Option<(u8, std::time::Instant)>,
+) {
+    shared.store(demand, std::sync::atomic::Ordering::Relaxed);
+    if drm_demand_log_due(*logged, demand, std::time::Instant::now()) {
+        log::debug!("drm: crtc {crtc}: the consumer encodes at {demand} fps");
+        *logged = Some((demand, std::time::Instant::now()));
+    }
+}
+
+fn drm_demand_log_due(
+    logged: Option<(u8, std::time::Instant)>,
+    demand: u8,
+    now: std::time::Instant,
+) -> bool {
+    demand != 0
+        && logged.map_or(true, |(v, at)| {
+            v != demand && now.saturating_duration_since(at) >= std::time::Duration::from_secs(10)
+        })
+}
+
+/// When the worker wakes: for the next frame, and at least every FRAME_INTERVAL, because each
+/// tick also reads the cursor and a slow frame rate must not slow the cursor down.
+fn drm_worker_wake(now: std::time::Instant, next_frame: std::time::Instant) -> std::time::Instant {
+    next_frame.min(now + FRAME_INTERVAL)
+}
+
 fn drm_capture_worker(
     frame_tx: tokio::sync::mpsc::Sender<DrmProducerMsg>,
     crtc_rx: std::sync::mpsc::Receiver<(String, u32, bool)>,
     stop: std::sync::Arc<std::sync::atomic::AtomicBool>,
     frames_gated: std::sync::Arc<std::sync::atomic::AtomicBool>,
+    demand_fps: std::sync::Arc<std::sync::atomic::AtomicU8>,
     wake: bool,
 ) {
     use std::sync::atomic::Ordering;
-    use std::time::Duration;
-    const FRAME_INTERVAL: Duration = Duration::from_millis(33);
-    // Bound continuous no-frame (WouldBlock) time so a wedged device ends the stream (~5 s).
-    const MAX_STALLED: u32 = 150;
 
     let t_conn = std::time::Instant::now();
 
@@ -1207,6 +1323,8 @@ fn drm_capture_worker(
         "drm: capture reader for crtc {target_crtc} opened in {:?}",
         t_open.elapsed()
     );
+    let mut panel = drm_read_panel_period(&mut reader, &target_device, target_crtc, None);
+    let mut panel_read_at = std::time::Instant::now();
 
     static DRM_CONN_EPOCH: std::sync::atomic::AtomicU32 = std::sync::atomic::AtomicU32::new(0);
     let conn_epoch = DRM_CONN_EPOCH.fetch_add(1, Ordering::Relaxed);
@@ -1216,8 +1334,11 @@ fn drm_capture_worker(
     let mut last_cursor_id: u64 = 0;
     let mut stalled: u32 = 0;
     let mut logged_first = false;
+    let mut next_frame = std::time::Instant::now();
     while !stop.load(Ordering::Relaxed) {
-        let mut grabbed: Option<std::io::Result<DrmProducerMsg>> = if frames_gated.load(Ordering::Relaxed)
+        let frame_due = std::time::Instant::now() >= next_frame;
+        let mut grabbed: Option<std::io::Result<DrmProducerMsg>> = if !frame_due
+            || frames_gated.load(Ordering::Relaxed)
         {
             // `stalled` is left untouched because the device is healthy -- the task bounds this
             // state itself (CREDIT_STALL) since our watchdog cannot advance.
@@ -1294,6 +1415,8 @@ fn drm_capture_worker(
                     break;
                 }
                 std::thread::sleep(FRAME_INTERVAL);
+                // Or the first good grab after the stall finds its deadline gone and grabs twice.
+                next_frame = std::time::Instant::now();
                 continue;
             }
             Some(Err(err)) if use_dmabuf && err.kind() == std::io::ErrorKind::Unsupported => {
@@ -1333,10 +1456,49 @@ fn drm_capture_worker(
             }
         }
 
-        std::thread::sleep(FRAME_INTERVAL);
+        if panel_read_at.elapsed() >= std::time::Duration::from_secs(1) {
+            panel = drm_read_panel_period(&mut reader, &target_device, target_crtc, Some(panel));
+            panel_read_at = std::time::Instant::now();
+        }
+        let now = std::time::Instant::now();
+        // A frame tick the credit held back is retried one pacing interval later.
+        if frame_due {
+            next_frame = drm_next_tick(next_frame, now, demand_fps.load(Ordering::Relaxed), panel);
+        }
+        std::thread::sleep(drm_worker_wake(now, next_frame).saturating_duration_since(now));
     }
 }
 
+/// The panel period of `crtc`, exact when the library can say (libdrmtap 0.5.9, one GETCRTC), else
+/// the whole-hertz refresh the last enumeration saw; zero when neither is known, which leaves the
+/// pacing to the demand.
+fn drm_read_panel_period(
+    reader: &mut scrap::drm_reader::DrmReader,
+    device: &str,
+    crtc: u32,
+    prev: Option<std::time::Duration>,
+) -> std::time::Duration {
+    let (period, how) = match reader.crtc_refresh() {
+        Some((num, den)) => (drm_panel_period(num, den), format!("{num}/{den} Hz")),
+        None => {
+            let hz = drm_whole_hz(device, crtc);
+            (drm_panel_period(u64::from(hz), 1), format!("{hz} Hz (whole hertz)"))
+        }
+    };
+    if prev != Some(period) {
+        log::debug!("drm: crtc {crtc} refreshes at {how}: grabs never outpace {period:?}");
+    }
+    period
+}
+
+fn drm_whole_hz(device: &str, crtc: u32) -> u32 {
+    DRM_WHOLE_HZ
+        .lock()
+        .ok()
+        .and_then(|hz| hz.get(&(device.to_owned(), crtc)).copied())
+        .unwrap_or(0)
+}
+
 /// Ancillary-fd transport for `_drm`: `Framed`/`BytesCodec` cannot carry an SCM_RIGHTS cmsg, so the
 /// messages and raw bodies use a 4-byte big-endian length + payload, with any fd bound to the first
     /// byte. The reverse-direction frame acks are bare bytes, not framed.
@@ -1565,7 +1727,7 @@ impl DrmConn {
         drm_send_frame(&self.stream, &payload, pass_fd).await
     }
 
-    pub async fn send_frame_ack(&self) -> ResultType<()> {
+    pub async fn send_frame_ack(&self, demand_fps: u8) -> ResultType<()> {
         let deadline =
             tokio::time::Instant::now() + std::time::Duration::from_millis(DRM_SEND_TIMEOUT_MS);
         loop {
@@ -1575,7 +1737,7 @@ impl DrmConn {
                     "drm: _drm frame-ack was not accepted within {DRM_SEND_TIMEOUT_MS}ms; closing"
                 ),
             }
-            match self.stream.try_write(&[1u8]) {
+            match self.stream.try_write(&[drm_ack_byte(demand_fps)]) {
                 Ok(n) if n > 0 => return Ok(()),
                 Ok(_) => bail!("drm: _drm frame-ack write returned 0 (peer closed)"),
                 Err(e) if e.kind() == std::io::ErrorKind::WouldBlock => continue,
@@ -1584,7 +1746,7 @@ impl DrmConn {
         }
     }
 
-    pub fn drain_frame_acks(&self, credit: &mut i32, max: i32) -> ResultType<()> {
+    pub fn drain_frame_acks(&self, credit: &mut i32, max: i32, demand: &mut u8) -> ResultType<()> {
         let mut buf = [0u8; 64];
         // BOUNDED: "until WouldBlock" is the peer's promise; a continuous writer would pin us.
         const MAX_ACK_READS: usize = 64;
@@ -1593,6 +1755,9 @@ impl DrmConn {
                 Ok(0) => bail!("drm: _drm frame-ack peer closed"),
                 Ok(n) => {
                     *credit = (*credit + n as i32).min(max);
+                    if let Some(d) = buf[..n].iter().rev().find_map(|&b| drm_ack_demand(b)) {
+                        *demand = d;
+                    }
                     if *credit >= max {
                         return Ok(());
                     }
@@ -1676,6 +1841,45 @@ impl DrmConn {
     }
 }
 
+#[cfg(test)]
+mod drm_pacing_tests {
+    use super::*;
+    use hbb_common::tokio;
+
+    /// An ack is one credit, as an older producer counts it, and a new one reads the demand from it.
+    #[tokio::test]
+    async fn a_demand_ack_is_one_credit_and_carries_the_demand() {
+        let (a, b) = tokio::net::UnixStream::pair().unwrap();
+        let tx = DrmConn::new(a);
+        let rx = DrmConn::new(b);
+        tx.send_frame_ack(60).await.unwrap();
+        rx.wait_readable().await.unwrap();
+        let (mut credit, mut demand) = (0, 0);
+        rx.drain_frame_acks(&mut credit, 2, &mut demand).unwrap();
+        assert_eq!((credit, demand), (1, 60));
+    }
+
+    #[test]
+    fn the_pacing_interval_follows_the_demand_with_the_panel_as_floor() {
+        let p60 = drm_panel_period(60, 1);
+        let p2998 = drm_panel_period(131375, 4382);
+        assert_eq!(
+            drm_panel_period(148352, 2475),
+            std::time::Duration::from_nanos(16_683_294),
+            "59.94 Hz, rounded up"
+        );
+        let cases = [
+            (0u8, p60, FRAME_INTERVAL, "no demand: the old tick"),
+            (60, p60, p60, "60 on a 60 Hz panel"),
+            (120, p60, p60, "more than the panel shows: the panel"),
+            (60, p2998, p2998, "a 29.98 Hz panel, exactly"),
+        ];
+        for (demand, panel, want, why) in cases {
+            assert_eq!(drm_pacing_interval(demand, panel), want, "{why}");
+        }
+    }
+}
+
 #[cfg(test)]
 mod drm_conn_tests {
     use super::*;

src/server/drm_capturer.rs

@@ -27,6 +27,11 @@ const BODY_READ_TIMEOUT: Duration = Duration::from_secs(5);
 /// A repeat of the frame returned last goes out anyway once this much time has passed since
 /// then: a frame whose encode failed reaches the encoder again, even if nothing moves.
 const REPEAT_EVERY: Duration = Duration::from_secs(1);
+/// After this long without a new frame the screen reads as still, and the producer is asked for at
+/// most STILL_FPS, its old 33 ms tick: grabbing a still screen faster only feeds frame() repeats to
+/// drop. Three of those ticks, so a producer paced at STILL_FPS in motion never reads as still.
+const STILL_AFTER: Duration = Duration::from_millis(100);
+const STILL_FPS: u8 = 30;
 
 struct FrameSlot {
     // Row stride is `pixels.len() / height`, possibly padded; the format and the plane rotation
@@ -84,6 +89,79 @@ struct Shared {
     // it: a receive thread that has seen the transform sees this too. `None` means this stream
     // never measures. Read once per shape arrival, never on the frame path.
     cal_context: Mutex<Option<CalContext>>,
+    // The rate the encoder takes frames at, carried on every ack so the producer grabs no faster.
+    // 0 until the first frame() call, which keeps the producer on its old 33 ms tick.
+    demand_fps: std::sync::atomic::AtomicU8,
+    pace: Mutex<EncoderPace>,
+}
+
+/// The encoder's own time per frame: from a frame() call that returned a frame to the next call,
+/// less what the call spent waiting for it, smoothed. Work inside frame() (a rotation) counts; the
+/// wait does not, so a late producer cannot talk its own demand down. A call that returned no frame
+/// (a dropped repeat, a timeout) left the encoder nothing to do, so the time after it is no sample.
+/// The samples also hold the sleep video_service paces itself with, so a shorter spf starts over.
+#[derive(Default)]
+struct EncoderPace {
+    last_call: Option<Instant>,
+    last_wait: Duration,
+    busy: Option<Duration>,
+    spf: Duration,
+}
+
+impl EncoderPace {
+    /// The rate to ask the producer for at a frame() call at `now` with `spf`, the last frame
+    /// having been returned at `returned_at`.
+    fn demand(&mut self, now: Instant, returned_at: Instant, spf: Duration) -> u8 {
+        if spf < self.spf {
+            // The sample that spans the change holds the old sleep too.
+            self.busy = None;
+            self.last_call = None;
+        }
+        self.spf = spf;
+        if let Some(prev) = self.last_call.filter(|&prev| returned_at > prev) {
+            let sample = now
+                .saturating_duration_since(prev)
+                .saturating_sub(self.last_wait)
+                .min(Duration::from_secs(1));
+            self.busy = Some(self.busy.map_or(sample, |avg| (avg * 7 + sample) / 8));
+        }
+        self.last_call = Some(now);
+        self.last_wait = Duration::ZERO;
+        let demand = demand_fps_of(spf, self.busy);
+        if now.saturating_duration_since(returned_at) >= STILL_AFTER {
+            demand.min(STILL_FPS)
+        } else {
+            demand
+        }
+    }
+}
+
+/// Records how long a frame() call waited for a frame, on every way out of the wait.
+struct WaitStamp<'a>(&'a Mutex<EncoderPace>, Instant);
+
+impl Drop for WaitStamp<'_> {
+    fn drop(&mut self) {
+        if let Ok(mut pace) = self.0.lock() {
+            pace.last_wait = self.1.elapsed();
+        }
+    }
+}
+
+/// The rate to grab at for an encoder that asks for `spf` and is busy `busy` per frame: what it
+/// asks for, or what it achieves when that is less.
+fn demand_fps_of(spf: Duration, busy: Option<Duration>) -> u8 {
+    let max = f64::from(super::video_qos::MAX_FPS);
+    let asked = if spf.is_zero() {
+        max
+    } else {
+        // f32 makes 1/60 read 59.99999: round, or 60 would be asked as 59.
+        (1.0 / spf.as_secs_f64()).round().clamp(1.0, max)
+    };
+    let achieved = match busy {
+        Some(o) if o > spf => (1.0 / o.as_secs_f64()).ceil().clamp(1.0, max),
+        _ => max,
+    };
+    asked.min(achieved) as u8
 }
 
 pub struct IpcDrmCapturer {
@@ -390,6 +468,8 @@ impl IpcDrmCapturer {
             cv: Condvar::new(),
             cursor_transform: std::sync::atomic::AtomicI32::new(TRANSFORM_PENDING),
             cal_context: Mutex::new(None),
+            demand_fps: std::sync::atomic::AtomicU8::new(0),
+            pace: Mutex::new(EncoderPace::default()),
         });
         let stop = Arc::new(AtomicBool::new(false));
         let (tx, rx) = std::sync::mpsc::channel::<ResultType<(Vec<DrmDisplayInfo>, usize)>>();
@@ -494,8 +574,18 @@ impl Drop for IpcDrmCapturer {
 
 impl TraitCapturer for IpcDrmCapturer {
     fn frame<'a>(&'a mut self, timeout: Duration) -> io::Result<Frame<'a>> {
+        // video_service passes its spf as the timeout: that, and how busy it is between calls, is
+        // the rate the producer is asked for.
+        let demand =
+            self.shared
+                .pace
+                .lock()
+                .unwrap()
+                .demand(Instant::now(), self.returned_at, timeout);
+        self.shared.demand_fps.store(demand, Ordering::Relaxed);
         let deadline = Instant::now() + timeout;
         let repeat = {
+            let waited = WaitStamp(&self.shared.pace, Instant::now());
             let mut slot = self.shared.slot.lock().unwrap();
             loop {
                 if slot.latest.is_some() || slot.ended.is_some() {
@@ -511,6 +601,7 @@ impl TraitCapturer for IpcDrmCapturer {
             }
             if let Some((w, h, fmt, plane_rotation, buf)) = slot.latest.take() {
                 drop(slot);
+                drop(waited);
                 // A layout change bumps the generation and is otherwise invisible here (mode
                 // and framebuffer keep their size). Rebuild for the new transform; not counted
                 // against health: the layout moved, the display did not fail.
@@ -823,7 +914,7 @@ async fn recv_thread(
                 // `recv_fd` closes at the end of this iteration, AFTER convert imported it.
                 // Ack so the producer RELEASES ONE SEND CREDIT and forwards the next; this bounds
                 // the socket to a couple of in-flight frames instead of a stale backlog.
-                if let Err(err) = conn.send_frame_ack().await {
+                if let Err(err) = conn.send_frame_ack(shared.demand_fps.load(Ordering::Relaxed)).await {
                     break format!("frame ack: {err}");
                 }
             }
@@ -877,7 +968,7 @@ async fn recv_thread(
                     Ok(Err(err)) => break format!("frame body: {err}"),
                 }
                 // Ack this CPU frame too (flow control; see the dma-buf arm above).
-                if let Err(err) = conn.send_frame_ack().await {
+                if let Err(err) = conn.send_frame_ack(shared.demand_fps.load(Ordering::Relaxed)).await {
                     break format!("frame ack: {err}");
                 }
             }
@@ -2411,6 +2502,8 @@ mod drm_capturer_tests {
                 cv: Condvar::new(),
                 cursor_transform: std::sync::atomic::AtomicI32::new(0),
                 cal_context: Mutex::new(None),
+                demand_fps: std::sync::atomic::AtomicU8::new(0),
+                pace: Mutex::new(EncoderPace::default()),
             }),
             stop: Arc::new(AtomicBool::new(false)),
             display: 0,