diff --git a/.github/workflows/flutter-build.yml b/.github/workflows/flutter-build.yml index eb2f14f30..c55e3f10b 100644 --- a/.github/workflows/flutter-build.yml +++ b/.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 diff --git a/build.py b/build.py index 7df00f341..1270b739f 100755 --- a/build.py +++ b/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 diff --git a/libs/scrap/Cargo.toml b/libs/scrap/Cargo.toml index c29647ad1..185c38f30 100644 --- a/libs/scrap/Cargo.toml +++ b/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 diff --git a/libs/scrap/src/common/drm_reader.rs b/libs/scrap/src/common/drm_reader.rs index e058d2c11..737616aff 100644 --- a/libs/scrap/src/common/drm_reader.rs +++ b/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 { 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"); + } +} diff --git a/libs/scrap/src/common/drmtap_dl.rs b/libs/scrap/src/common/drmtap_dl.rs index 6bb74eee6..3f043e905 100644 --- a/libs/scrap/src/common/drmtap_dl.rs +++ b/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, + /// 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, 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 = lib.get(b"drmtap_plane_rotation").ok().map(|s| *s); + let crtc_refresh: Option = + 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), }) } diff --git a/src/ipc/drm.rs b/src/ipc/drm.rs index 3f543454a..b129e1109 100644 --- a/src/ipc/drm.rs +++ b/src/ipc/drm.rs @@ -264,6 +264,11 @@ static DRM_DISPLAY_CACHE: std::sync::Mutex> = 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::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, Vec) { let render_node = reader.render_node().unwrap_or_default(); let mut undriven = Vec::new(); - let displays: Vec = 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 = 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 = 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 { + 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, crtc_rx: std::sync::mpsc::Receiver<(String, u32, bool)>, stop: std::sync::Arc, frames_gated: std::sync::Arc, + demand_fps: std::sync::Arc, 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> = if frames_gated.load(Ordering::Relaxed) + let frame_due = std::time::Instant::now() >= next_frame; + let mut grabbed: Option> = 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 { + 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::*; diff --git a/src/server/drm_capturer.rs b/src/server/drm_capturer.rs index 656efa80f..a50123c25 100644 --- a/src/server/drm_capturer.rs +++ b/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>, + // 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, +} + +/// 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, + last_wait: Duration, + busy: Option, + 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, 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) -> 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::, usize)>>(); @@ -494,8 +574,18 @@ impl Drop for IpcDrmCapturer { impl TraitCapturer for IpcDrmCapturer { fn frame<'a>(&'a mut self, timeout: Duration) -> io::Result> { + // 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,