From 3b6914b1b236a96314be4c2b070e61475a686d4a Mon Sep 17 00:00:00 2001 From: Mariano Abad Date: Mon, 20 Jul 2026 23:53:22 -0300 Subject: [PATCH] feat(drm): phase-2 split, pass the dma-buf fd instead of the converted frame move the egl detile and rgba pack out of the root --service and into the unprivileged --server. the root now calls only drmtap_open + drmtap_grab_desc and exports a raw dma-buf fd; the fd rides the _drm channel over SCM_RIGHTS with a small descriptor (geometry, per-plane offsets/pitches, modifier, hdr) instead of the full rgba frame, dropping the per-frame copy. the --server imports the fd with drmtap_open_render + drmtap_convert_dmabuf, keyed by the import-once egl cache, and the render context is created and dropped on the recv thread. the _drm transport moves off Framed (which cannot carry a fd) to a bespoke sendmsg/recvmsg framing (DrmConn) that attaches one SCM_RIGHTS cmsg only when a fd is present and rejects a truncated ancillary message. the split symbols are bound optionally so an older libdrmtap still loads the cpu path, and the whole thing degrades to the cpu BGRA path or PipeWire when no render node is available. pins libdrmtap-sys to =0.4.13 with the Cargo.lock checksum. folds in the DP-MST, ldconfig-restart and per-display PipeWire-fallback review fixes and a udev hotplug refresh. --- .github/workflows/flutter-build.yml | 22 +- Cargo.lock | 11 + build.py | 23 +- libs/scrap/Cargo.toml | 15 +- libs/scrap/src/common/drm_reader.rs | 141 ++++- libs/scrap/src/common/drm_render.rs | 195 +++++++ libs/scrap/src/common/drmtap_dl.rs | 69 +++ libs/scrap/src/common/mod.rs | 2 + src/ipc.rs | 783 ++++++++++++++++++++++++++-- src/server/drm_capturer.rs | 183 ++++++- src/server/wayland.rs | 31 +- 11 files changed, 1403 insertions(+), 72 deletions(-) create mode 100644 libs/scrap/src/common/drm_render.rs diff --git a/.github/workflows/flutter-build.yml b/.github/workflows/flutter-build.yml index 01db303a5..bcb8c7e61 100644 --- a/.github/workflows/flutter-build.yml +++ b/.github/workflows/flutter-build.yml @@ -1702,9 +1702,17 @@ jobs: # exported so build.py reuses the exact same source. We clone + build # the .so here and hand it to build.py via DRMTAP_PREBUILT_DIR, because # a later build step in this container disturbs the working tree. - # rustdesk-org/libdrmtap main tracks the current release (0.4.8+). + # DRMTAP_REF is an EXACT release tag (vX.Y.Z), NOT a branch: the bundled + # .so version must match the `libdrmtap-sys = "=0.4.13"` crate pin in + # libs/scrap/Cargo.toml (the loader only checks ABI-major). Bump both together. export DRMTAP_REPO="https://github.com/rustdesk-org/libdrmtap" - export DRMTAP_REF="main" + export DRMTAP_REF="v0.4.13" + # Guard: refuse a loose/branch ref so a moving `main` can never silently + # regress the pin. Only a vX.Y.Z tag is accepted. + case "$DRMTAP_REF" in + v[0-9]*.[0-9]*.[0-9]*) ;; + *) echo "FATAL: DRMTAP_REF must be a pinned vX.Y.Z tag, got '$DRMTAP_REF'"; exit 1;; + esac git config --global --add safe.directory '*' || true rm -rf third_party/libdrmtap git clone --depth 1 --branch "$DRMTAP_REF" "$DRMTAP_REPO" third_party/libdrmtap @@ -1723,6 +1731,16 @@ jobs: [ -n "$(find "$DRMTAP_PREBUILT_DIR" -name 'libdrmtap.so.0.*' -type f)" ] \ || { echo "FATAL: prebuilt libdrmtap.so missing"; exit 1; } ls -l "$DRMTAP_PREBUILT_DIR" + # Lock-freshness guard, BEFORE the `--locked` build (which itself fails on a + # checksum mismatch): prove Cargo.lock is in sync with Cargo.toml and actually + # contains the pinned libdrmtap-sys (the integrator must have run + # `cargo update -p libdrmtap-sys --precise 0.4.13`). This turns a stale/loose lock + # into an explicit, readable failure instead of a confusing error deep in the build. + # The `=0.4.13` requirement in libs/scrap/Cargo.toml + `--locked` already force the + # exact resolved version + its checksum; this only confirms the lock was regenerated. + cargo update --locked --dry-run + cargo tree --locked --features drm -p libdrmtap-sys -i >/dev/null \ + || { echo "FATAL: Cargo.lock does not contain the pinned libdrmtap-sys (run: cargo update -p libdrmtap-sys --precise 0.4.13)"; exit 1; } cargo build --locked --lib $JOBS --features hwcodec,flutter,unix-file-copy-paste,drm --release python3 ./build.py --flutter --drm --skip-cargo for name in rustdesk-unattended-wayland*??.deb; do diff --git a/Cargo.lock b/Cargo.lock index 23cf35cbe..38ff7bc21 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -4485,6 +4485,16 @@ dependencies = [ "pkg-config", ] +[[package]] +name = "libdrmtap-sys" +version = "0.4.13" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e6c52136bd2bf94bb1071f3926eff327148faddc704338f3fca473e1d1974723" +dependencies = [ + "cc", + "pkg-config", +] + [[package]] name = "libgit2-sys" version = "0.14.2+1.5.1" @@ -7610,6 +7620,7 @@ dependencies = [ "hwcodec", "jni", "lazy_static", + "libdrmtap-sys", "log", "ndk 0.7.0", "ndk-context", diff --git a/build.py b/build.py index e9e27db2d..09a06654e 100755 --- a/build.py +++ b/build.py @@ -334,11 +334,14 @@ def ffi_bindgen_function_refactor(): # libdrmtap is fetched at build time by cloning the rustdesk-org fork at a pinned # ref — the same way rustdesk sources its other native build deps (vcpkg, -# flutter_rust_bridge, ...), rather than carrying a git submodule. The ref can be -# a branch or a tag; rustdesk-org/libdrmtap main tracks the current release. -# Override the repo/ref via env (DRMTAP_REPO / DRMTAP_REF) for local testing or another fork. +# flutter_rust_bridge, ...), rather than carrying a git submodule. The ref is an +# EXACT release tag (vX.Y.Z), NOT a moving branch: the bundled runtime .so version must +# match the `libdrmtap-sys = "=0.4.13"` crate pinned in libs/scrap/Cargo.toml (the dlopen +# loader only checks ABI-major, so a minor skew between the two artifacts would go undetected). +# Bump this tag and the crate pin together. Override the repo/ref via env (DRMTAP_REPO / +# DRMTAP_REF) for local testing or another fork. LIBDRMTAP_REPO = os.environ.get('DRMTAP_REPO', 'https://github.com/rustdesk-org/libdrmtap') -LIBDRMTAP_REF = os.environ.get('DRMTAP_REF', 'main') +LIBDRMTAP_REF = os.environ.get('DRMTAP_REF', 'v0.4.13') def _single_real_so(paths, where): @@ -390,11 +393,23 @@ def append_drm_ldconfig_postinst(): # The DRM package installs libdrmtap.so under a private dir; register it with the # dynamic linker so the in-process dlopen("libdrmtap.so.0") resolves. Only the DRM # package calls this, so the stock package's postinst stays byte-identical to upstream. + # + # This block is appended AFTER the stock postinst, which has already run + # `systemctl start rustdesk`. On a FRESH install that ordering is a trap: the root + # service's DRM pre-warm can dlopen("libdrmtap.so.0") BEFORE this ldconfig has + # populated the linker cache, the dlopen fails, and that failure is cached in the + # DRMTAP_LIB OnceLock for the life of the process — so DRM stays disabled until a + # manual restart. So immediately after ldconfig we `try-restart` the unit: it re-runs + # the pre-warm against the now-resolvable soname. `try-restart` is a no-op when the + # unit is not running, so it never spuriously starts the service. with open('tmpdeb/DEBIAN/postinst', 'a') as f: f.write( '\n' 'if [ "$1" = configure ] && [ -d /usr/lib/rustdesk ]; then\n' '\tldconfig /usr/lib/rustdesk 2>/dev/null || ldconfig 2>/dev/null || true\n' + '\tif command -v systemctl >/dev/null 2>&1; then\n' + '\t\tsystemctl try-restart rustdesk 2>/dev/null || true\n' + '\tfi\n' 'fi\n' ) diff --git a/libs/scrap/Cargo.toml b/libs/scrap/Cargo.toml index 68a7ae4f3..7ff6c2898 100644 --- a/libs/scrap/Cargo.toml +++ b/libs/scrap/Cargo.toml @@ -11,7 +11,11 @@ edition = "2018" [features] wayland = ["gstreamer", "gstreamer-app", "gstreamer-video", "dbus", "tracing", "zbus"] -drm = [] +# `drm` pulls in the exact-pinned `libdrmtap-sys` crate (see the Linux target table below). rustdesk +# still dlopen's `libdrmtap.so.0` at runtime (`drmtap_dl.rs`) rather than link-time linking it, so the +# graceful PipeWire fallback when the .so/EGL is absent is preserved; the crate is pinned purely so its +# committed Cargo.lock checksum supply-chain-pins the vendored source that build.py bundles. +drm = ["dep:libdrmtap-sys"] mediacodec = ["ndk"] linux-pkg-config = ["dep:pkg-config"] hwcodec = ["dep:hwcodec"] @@ -59,6 +63,15 @@ gstreamer = { version = "0.16", optional = true } gstreamer-app = { version = "0.16", features = ["v1_10"], optional = true } gstreamer-video = { version = "0.16", optional = true } zbus = { version = "3.15", optional = true } +# libdrmtap ABI, pinned EXACTLY (the leading `=` blocks ^0.4.x semver drift). This is the ONLY +# supply-chain pin for libdrmtap: the committed Cargo.lock `checksum` for this exact version pins the +# bytes, so `cargo build --locked` refuses a tampered/republished 0.4.13. It must move in lockstep +# with build.py's `DRMTAP_REF = v0.4.13` (the bundled runtime .so) — the loader only checks ABI-major. +# NOTE for the integrator: after this edit the lockfile MUST be regenerated + committed with +# cargo update -p libdrmtap-sys --precise 0.4.13 +# so Cargo.lock gains the `[[package]] libdrmtap-sys 0.4.13 checksum=` block (this stage does +# not run cargo). `=0.4.13` alone is a version requirement, not a byte pin — the checksum is the pin. +libdrmtap-sys = { version = "=0.4.13", optional = true } [dependencies.hwcodec] git = "https://github.com/rustdesk-org/hwcodec" diff --git a/libs/scrap/src/common/drm_reader.rs b/libs/scrap/src/common/drm_reader.rs index 3d5d02548..4b2e0cbad 100644 --- a/libs/scrap/src/common/drm_reader.rs +++ b/libs/scrap/src/common/drm_reader.rs @@ -10,12 +10,13 @@ // before opening. The DRM_DEVICE env is intentionally NOT consulted here. use super::drmtap_dl::{ - self, drmtap_config, drmtap_ctx, drmtap_cursor_info, drmtap_display, drmtap_frame_info, - DrmtapLib, + self, drmtap_config, drmtap_ctx, drmtap_cursor_info, drmtap_display, drmtap_dmabuf_desc, + drmtap_frame_info, DrmtapLib, }; use hbb_common::log; use std::ffi::CString; use std::io; +use std::os::fd::{FromRawFd, OwnedFd}; // Largest scanout we will copy; also bounds w*4*h against overflow. 16384 covers // 8K+ with headroom; anything larger is rejected as a bogus/hostile geometry. @@ -215,6 +216,142 @@ impl DrmReader { } } + /// True if the loaded libdrmtap exposes the split-capture export entry point + /// (`drmtap_grab_desc`, libdrmtap >= 0.4.9). When false the caller must use + /// the CPU-mapped `grab()` path (an older `.so`). + pub fn supports_grab_desc(&self) -> bool { + self.lib.grab_desc.is_some() + } + + /// Zero-copy EXPORT grab for the split-capture path (root `--service`). Calls + /// `drmtap_grab_desc`, which fills a `drmtap_dmabuf_desc` (the scanout dma-buf + /// fd + the full plane layout + HDR metadata) WITHOUT mapping, detiling or + /// copying any pixels — so on this path the root process never loads + /// libEGL/libGLESv2 (the EGL convert now lives in the unprivileged `--server`). + /// + /// The scanout `dma_buf_fd` is dup'd into an `OwnedFd` BEFORE the frame is + /// released, so we keep an independently-owned reference to the buffer that + /// survives `drmtap_frame_release` (the dma-buf refcount keeps the memory + /// alive while the peer also holds a reference). The descriptor is validated + /// on METADATA ONLY (no pixel access on the export side): the fourcc gate + /// (kept from `grab()`), `MAX_DIM`, and `num_planes` in `1..=4`. + /// + /// Returns the owned fd + the validated descriptor with `dma_buf_fd` reset to + /// `-1` (the `OwnedFd` owns the fd now; the descriptor's local int must never + /// be closed or re-used). Errno mapping mirrors `grab()`: EAGAIN/EBUSY/EINTR + /// -> WouldBlock (retry); ENOTSUP -> a distinct `Unsupported` error (this + /// seat/driver produced pixels but no transferable dma-buf) so the caller + /// falls back to the mapped/PipeWire path instead of a per-frame rebuild loop; + /// any other errno -> hard error. Errors when `grab_desc` is unbound (old .so). + pub fn grab_desc(&mut self) -> io::Result<(OwnedFd, drmtap_dmabuf_desc)> { + let grab_desc = self.lib.grab_desc.ok_or_else(|| { + io::Error::new( + io::ErrorKind::Unsupported, + "libdrmtap too old: drmtap_grab_desc unavailable (need >= 0.4.9)", + ) + })?; + // SAFETY: self.ctx is a valid context; desc/frame are zeroed before the + // call and the frame is released on every return path (after the dup). + unsafe { + let mut desc: drmtap_dmabuf_desc = std::mem::zeroed(); + let mut frame: drmtap_frame_info = std::mem::zeroed(); + let ret = grab_desc(self.ctx, &mut desc, &mut frame); + if ret < 0 { + let errno = -ret; + if errno == hbb_common::libc::EAGAIN + || errno == hbb_common::libc::EBUSY + || errno == hbb_common::libc::EINTR + { + return Err(io::ErrorKind::WouldBlock.into()); + } + if errno == hbb_common::libc::ENOTSUP { + // Pixels exist but there is no transferable dma-buf on this + // seat/driver: the split export can never work here. A distinct + // Unsupported error so the caller degrades (CPU-mapped/PipeWire) + // rather than tight-looping a rebuild. + return Err(io::Error::new( + io::ErrorKind::Unsupported, + "drmtap_grab_desc: no transferable dma-buf (ENOTSUP)", + )); + } + return Err(io::Error::new( + io::ErrorKind::Other, + format!("drmtap_grab_desc failed: errno {errno}"), + )); + } + // The canonical fd is `desc.dma_buf_fd` (what split_capture.c sends); + // `frame` also owns it and `frame_release` will close the library's + // copy. A negative fd here means no new scanout this grab -> retry. + let raw_fd = if desc.dma_buf_fd >= 0 { + desc.dma_buf_fd + } else { + frame.dma_buf_fd + }; + if raw_fd < 0 { + (self.lib.frame_release)(self.ctx, &mut frame); + return Err(io::ErrorKind::WouldBlock.into()); + } + // ---- METADATA-ONLY validation (no pixel access on the export side) ---- + let w = desc.width; + let h = desc.height; + if w == 0 || h == 0 || w > MAX_DIM || h > MAX_DIM { + (self.lib.frame_release)(self.ctx, &mut frame); + return Err(io::Error::new( + io::ErrorKind::Other, + format!("DRM scanout geometry {w}x{h} out of range"), + )); + } + // fourcc gate (see grab()): reject a scanout the converter could not + // present as BGRA. 0/unknown is allowed here — an older .so may not set + // it, and the converter reads `frame.format` authoritatively per frame. + const DRM_FORMAT_XRGB8888: u32 = 0x3432_5258; // 'XR24' + const DRM_FORMAT_ARGB8888: u32 = 0x3432_5241; // 'AR24' + if desc.format != 0 + && desc.format != DRM_FORMAT_XRGB8888 + && desc.format != DRM_FORMAT_ARGB8888 + { + log::warn!( + "DRM scanout fourcc {:#010x} is not BGRA-compatible; falling back", + desc.format + ); + (self.lib.frame_release)(self.ctx, &mut frame); + return Err(io::Error::new( + io::ErrorKind::Other, + "unsupported DRM scanout format", + )); + } + // num_planes must index offsets/pitches (0 is treated as 1 per the ABI). + let planes = if desc.num_planes == 0 { 1 } else { desc.num_planes }; + if planes > 4 { + (self.lib.frame_release)(self.ctx, &mut frame); + return Err(io::Error::new( + io::ErrorKind::Other, + format!("DRM scanout num_planes {} out of range (1..=4)", desc.num_planes), + )); + } + // dup the fd into an OwnedFd BEFORE releasing the frame: after release + // the library may recycle its handle, but our dup (an independent fd on + // the same open dma-buf) keeps the buffer alive for the peer. + let dup_fd = hbb_common::libc::dup(raw_fd); + if dup_fd < 0 { + let e = io::Error::last_os_error(); + (self.lib.frame_release)(self.ctx, &mut frame); + return Err(e); + } + let owned = OwnedFd::from_raw_fd(dup_fd); + // Release now that the fd is safely dup'd (split_capture.c releases only + // after the send; we release after the dup, which is equivalent because + // the dup holds its own reference to the dma-buf). + (self.lib.frame_release)(self.ctx, &mut frame); + // Normalize num_planes and blank the descriptor's local fd int: the + // OwnedFd owns the fd, and the wire descriptor carries `has_fd` + the + // ancillary fd, never this integer. + desc.num_planes = planes; + desc.dma_buf_fd = -1; + Ok((owned, desc)) + } + } + /// Read the hardware cursor plane. Returns a hidden sentinel when the plane /// reports the cursor invisible, the real shape when visible, or None on a /// read error / unsupported cursor. Ported from the old drm.rs update_cursor. diff --git a/libs/scrap/src/common/drm_render.rs b/libs/scrap/src/common/drm_render.rs new file mode 100644 index 000000000..eb5be020a --- /dev/null +++ b/libs/scrap/src/common/drm_render.rs @@ -0,0 +1,195 @@ +// Unprivileged (`--server`) render-side converter for the split DRM/KMS capture +// path. This is the OTHER half of the split introduced with libdrmtap >= 0.4.9: +// the root `--service` now only EXPORTS a scanout dma-buf fd + a small metadata +// descriptor (see `drm_reader::grab_desc`), and THIS side imports that fd and does +// the EGL detile / RGBA convert. Because the convert lives here, libEGL/libGLESv2 +// are dlopen'd in the UNPRIVILEGED process, never in the privileged root service. +// +// A `RenderConverter` wraps a `drmtap_open_render(NULL)` render-node context and +// converts one imported dma-buf per `convert()` call. The EGL context and the +// import-once EGLImage cache it holds are THREAD-LOCAL inside libdrmtap: the +// context MUST be created, used (`convert`), and closed (`drop`) on the SAME +// thread (the consumer's `recv_thread`). Dropping it off-thread would strand the +// cached EGLImages — the exact leak class behind the 0.4.8 OOM regression. The raw +// ctx pointer makes `RenderConverter` !Send/!Sync, which enforces that at the type +// level. + +use super::drmtap_dl::{self, drmtap_ctx, drmtap_dmabuf_desc, drmtap_frame_info, DrmtapLib}; +use super::Pixfmt; +use hbb_common::log; +use std::io; +use std::os::fd::RawFd; + +// DRM fourccs of the 32-bit linear formats libdrmtap's convert can emit. XRGB/ARGB +// are little-endian B,G,R,{X,A} in memory == our `Pixfmt::BGRA`; XBGR/ABGR are +// R,G,B,{X,A} == `Pixfmt::RGBA`. libdrmtap normalizes the EGL path to XRGB8888, but +// we read `frame.format` per frame so a CPU-fallback convert that keeps the source +// channel order is still presented correctly (not hardcoded BGRA). +const DRM_FORMAT_XRGB8888: u32 = 0x3432_5258; // 'XR24' +const DRM_FORMAT_ARGB8888: u32 = 0x3432_5241; // 'AR24' +const DRM_FORMAT_XBGR8888: u32 = 0x3432_4258; // 'XB24' +const DRM_FORMAT_ABGR8888: u32 = 0x3432_4241; // 'AB24' + +// Same geometry / size guards as the export side (`drm_reader`), applied to the +// convert OUTPUT so a malformed `frame_info` cannot make us build an out-of-range +// slice from the context-owned pointer. 16384 covers 8K+; 256 MiB covers an 8K +// BGRA frame (7680x4320x4 ~= 127 MiB) with margin. +const MAX_DIM: u32 = 16384; +const MAX_FRAME_BYTES: usize = 256 * 1024 * 1024; + +/// An unprivileged DRM render-node convert context (`drmtap_open_render`). Imports a +/// scanout dma-buf (received over SCM_RIGHTS) and EGL-detiles it to linear pixels. +/// Deliberately !Send/!Sync (the raw ctx pointer): the context and libdrmtap's +/// thread-local EGL state must stay on ONE thread for the context's whole life +/// (create + convert + close). +pub struct RenderConverter { + lib: &'static DrmtapLib, + ctx: *mut drmtap_ctx, +} + +impl RenderConverter { + /// Open an unprivileged DRM render-node convert context (`drmtap_open_render(NULL)` + /// auto-selects a render node — it opens no KMS card, spawns no helper, and needs + /// no elevated capability). Returns `None` when libdrmtap is unavailable, the split + /// convert symbols are missing (a pre-0.4.9 `.so`), or no render node could be + /// opened (a locked-down seat with no `/dev/dri/renderD*` access) — the caller then + /// degrades to the CPU-mapped / PipeWire path. MUST be called on the thread that + /// will later `convert()` and drop it. + pub fn open_render() -> Option { + let lib = drmtap_dl::get()?; + // The converter needs BOTH split symbols; bail (so the caller degrades) if either + // is absent, rather than open a ctx we could never convert with. + let open_render = lib.open_render?; + if lib.convert_dmabuf.is_none() { + log::info!( + "libdrmtap exposes drmtap_open_render but not drmtap_convert_dmabuf; \ + cannot convert dma-buf frames (old .so)" + ); + return None; + } + // SAFETY: `open_render` is a resolved C entry point; NULL requests auto-selection + // of a render node. + let ctx = unsafe { open_render(std::ptr::null()) }; + if ctx.is_null() { + log::info!("drmtap_open_render(NULL) failed; no usable DRM render node"); + return None; + } + log::info!("drm: opened unprivileged render-node convert context"); + Some(RenderConverter { lib, ctx }) + } + + /// Import + convert one scanout dma-buf. `desc` is the descriptor rebuilt from the + /// wire `DmabufDesc`; `received_fd` is the fd number this process obtained via + /// SCM_RIGHTS (or `-1` for an import-once cache hit, where libdrmtap reuses the + /// EGLImage it already holds for `desc.fb_id`). The fd is written into + /// `desc.dma_buf_fd` before the call (LOAD-BEARING: the integer the exporter + /// serialized was process-local and never crossed the wire). + /// + /// On success returns a borrow of the CONTEXT-OWNED linear pixels plus the frame + /// width/height and the `Pixfmt` read from `frame.format`. The slice covers + /// `stride * height` bytes, so the caller can recover the (possibly padded) row + /// stride as `data.len() / height`. It is valid ONLY until the next `convert()` + /// (or drop) — do NOT free it and do NOT call `drmtap_frame_release` on it + /// (libdrmtap owns it). The `&mut self` borrow keeps the slice from outliving the + /// next convert; copy it out (into the latest-wins slot) before the next call. + pub fn convert( + &mut self, + desc: &mut drmtap_dmabuf_desc, + received_fd: RawFd, + ) -> io::Result<(&[u8], u32, u32, Pixfmt)> { + let convert_dmabuf = self.lib.convert_dmabuf.ok_or_else(|| { + io::Error::new( + io::ErrorKind::Unsupported, + "libdrmtap too old: drmtap_convert_dmabuf unavailable (need >= 0.4.9)", + ) + })?; + // Overwrite the descriptor's fd with the one THIS process received (split_capture.c + // does the same at recv time). -1 means "reuse the cached import for `fb_id`". + desc.dma_buf_fd = received_fd; + // SAFETY: self.ctx is a valid render context; `desc` points to a fully-initialized + // descriptor; `frame` is zeroed before the call. libdrmtap OWNS the returned + // `frame.data` (no release/free from this side, per drmtap.h). + unsafe { + let mut frame: drmtap_frame_info = std::mem::zeroed(); + let ret = convert_dmabuf(self.ctx, &*desc as *const drmtap_dmabuf_desc, &mut frame); + if ret < 0 { + let errno = -ret; + // Transient contention (device busy, interrupted syscall) -> retry rather + // than tear the stream down. + if errno == hbb_common::libc::EAGAIN + || errno == hbb_common::libc::EBUSY + || errno == hbb_common::libc::EINTR + { + return Err(io::ErrorKind::WouldBlock.into()); + } + return Err(io::Error::new( + io::ErrorKind::Other, + format!("drmtap_convert_dmabuf failed: errno {errno}"), + )); + } + if frame.data.is_null() || frame.width == 0 || frame.height == 0 || frame.stride == 0 { + return Err(io::Error::new( + io::ErrorKind::Other, + "drmtap_convert_dmabuf produced an empty frame", + )); + } + let w = frame.width; + let h = frame.height; + let stride = frame.stride as usize; + // Guard the slice we are about to build from a hostile/garbage `frame_info`: + // reject an insane geometry or a stride below 32bpp (would under-size the row + // and, read as BGRA downstream, disclose adjacent memory). + if w > MAX_DIM || h > MAX_DIM || stride < (w as usize) * 4 { + return Err(io::Error::new( + io::ErrorKind::Other, + format!( + "drmtap_convert_dmabuf bad geometry {w}x{h} stride {stride} fourcc {:#010x}", + frame.format + ), + )); + } + let len = match stride.checked_mul(h as usize) { + Some(sz) if sz > 0 && sz <= MAX_FRAME_BYTES => sz, + other => { + return Err(io::Error::new( + io::ErrorKind::Other, + format!("drmtap_convert_dmabuf frame size out of range ({other:?} bytes)"), + )); + } + }; + // Channel order from the ACTUAL convert output (do NOT hardcode BGRA): the EGL + // path normalizes to XRGB8888 (BGRA), but reading it keeps any other emitted + // order labeled correctly for the encoder. + let pixfmt = match frame.format { + DRM_FORMAT_XRGB8888 | DRM_FORMAT_ARGB8888 => Pixfmt::BGRA, + DRM_FORMAT_XBGR8888 | DRM_FORMAT_ABGR8888 => Pixfmt::RGBA, + // Unset by an older convert -> libdrmtap's normalized BGRA. + 0 => Pixfmt::BGRA, + other => { + log::debug!( + "drm: convert output fourcc {other:#010x} unrecognized; presenting as BGRA" + ); + Pixfmt::BGRA + } + }; + // Borrow the context-owned pixels. The returned lifetime is tied to `&mut self` + // (elision), so the borrow cannot outlive the next `convert()` that would + // overwrite these bytes. + let data = std::slice::from_raw_parts(frame.data as *const u8, len); + Ok((data, w, h, pixfmt)) + } + } +} + +impl Drop for RenderConverter { + fn drop(&mut self) { + if !self.ctx.is_null() { + // SAFETY: ctx came from drmtap_open_render and is non-null. This MUST run on the + // same thread that created and used it (thread-local EGL state + cached imports); + // guaranteed because the !Send ctx pointer keeps the whole `RenderConverter` on + // the owning `recv_thread`, where it is also dropped. + unsafe { (self.lib.close)(self.ctx) }; + self.ctx = std::ptr::null_mut(); + } + } +} diff --git a/libs/scrap/src/common/drmtap_dl.rs b/libs/scrap/src/common/drmtap_dl.rs index 8d052ee98..c49f5f836 100644 --- a/libs/scrap/src/common/drmtap_dl.rs +++ b/libs/scrap/src/common/drmtap_dl.rs @@ -68,6 +68,46 @@ pub struct drmtap_frame_info { pub _priv: *mut c_void, } +// Descriptor of an externally-supplied scanout DMA-BUF (the split-capture +// contract). Mirrors `drmtap_dmabuf_desc` in libdrmtap include/drmtap.h EXACTLY +// (field order + widths); a mismatch mis-reads CCS/HDR scanouts. The privileged +// exporter fills it in one call via `drmtap_grab_desc`; the unprivileged +// converter receives it over IPC, overwrites `dma_buf_fd` with the fd it got via +// SCM_RIGHTS, and passes it to `drmtap_convert_dmabuf`. +#[repr(C)] +#[derive(Clone, Copy)] +pub struct drmtap_dmabuf_desc { + pub dma_buf_fd: c_int, // scanout DMA-BUF; -1 for an already-imported fb_id + pub width: u32, + pub height: u32, + pub format: u32, // DRM fourcc of the scanout + pub modifier: u64, // DRM format modifier (tiling/compression) + pub fb_id: u32, // import-once cache key; 0 disables caching + pub num_planes: u32, // used entries in offsets/pitches (1..4); 0 => 1 + pub offsets: [u32; 4], // per-plane byte offsets (CCS main+aux+clear-color) + pub pitches: [u32; 4], // per-plane strides; pitches[0] = main stride + pub hdr_eotf: u32, // DRMTAP_EOTF_* (SDR=0, PQ=2, HLG=3) + pub hdr_max_nits: u32, // mastering/content peak luminance cd/m2; 0=unknown +} + +impl Default for drmtap_dmabuf_desc { + fn default() -> Self { + Self { + dma_buf_fd: -1, + width: 0, + height: 0, + format: 0, + modifier: 0, + fb_id: 0, + num_planes: 0, + offsets: [0; 4], + pitches: [0; 4], + hdr_eotf: 0, + hdr_max_nits: 0, + } + } +} + #[repr(C)] pub struct drmtap_cursor_info { pub x: i32, @@ -91,6 +131,14 @@ type FnGrabMapped = unsafe extern "C" fn(*mut drmtap_ctx, *mut drmtap_frame_info type FnFrameRelease = unsafe extern "C" fn(*mut drmtap_ctx, *mut drmtap_frame_info); type FnGetCursor = unsafe extern "C" fn(*mut drmtap_ctx, *mut drmtap_cursor_info) -> c_int; type FnCursorRelease = unsafe extern "C" fn(*mut drmtap_ctx, *mut drmtap_cursor_info); +// Split-capture entry points (libdrmtap >= 0.4.9). Bound OPTIONALLY (see below). +// `grab_desc` runs on the privileged export side; `open_render`/`convert_dmabuf` +// on the unprivileged converter side. +type FnGrabDesc = + unsafe extern "C" fn(*mut drmtap_ctx, *mut drmtap_dmabuf_desc, *mut drmtap_frame_info) -> c_int; +type FnOpenRender = unsafe extern "C" fn(*const c_char) -> *mut drmtap_ctx; +type FnConvertDmabuf = + unsafe extern "C" fn(*mut drmtap_ctx, *const drmtap_dmabuf_desc, *mut drmtap_frame_info) -> c_int; /// The dlopen'd libdrmtap with its resolved entry points. The `Library` is kept /// alive for the process lifetime (this lives in a `OnceLock`), so the raw fn @@ -104,6 +152,15 @@ pub struct DrmtapLib { pub frame_release: FnFrameRelease, pub get_cursor: FnGetCursor, pub cursor_release: FnCursorRelease, + // Split-capture symbols (present only on libdrmtap >= 0.4.9). `None` on an + // older .so; callers gate on `Some(..)` and fall back to the mapped path. + // Root needs `grab_desc`; the unprivileged converter needs + // `open_render` + `convert_dmabuf`. + pub grab_desc: Option, + pub open_render: Option, + pub convert_dmabuf: Option, + // Parsed (major, minor, patch) from `drmtap_version()`, for feature gating. + pub version: (c_int, c_int, c_int), } // SAFETY: the resolved fn pointers are plain C entry points with no interior @@ -152,6 +209,14 @@ impl DrmtapLib { let frame_release: FnFrameRelease = *lib.get(b"drmtap_frame_release").ok()?; let get_cursor: FnGetCursor = *lib.get(b"drmtap_get_cursor").ok()?; let cursor_release: FnCursorRelease = *lib.get(b"drmtap_cursor_release").ok()?; + // Split-capture symbols are bound OPTIONALLY (not through the `.ok()?` + // chain above): a pre-0.4.9 .so lacks them, and forcing them here would + // fail the WHOLE load and silently disable DRM. Each side checks the + // symbol it needs before taking the split path. + let grab_desc: Option = lib.get(b"drmtap_grab_desc").ok().map(|s| *s); + let open_render: Option = lib.get(b"drmtap_open_render").ok().map(|s| *s); + let convert_dmabuf: Option = + lib.get(b"drmtap_convert_dmabuf").ok().map(|s| *s); Some(DrmtapLib { _lib: lib, open, @@ -161,6 +226,10 @@ impl DrmtapLib { frame_release, get_cursor, cursor_release, + grab_desc, + open_render, + convert_dmabuf, + version: (major, minor, patch), }) } } diff --git a/libs/scrap/src/common/mod.rs b/libs/scrap/src/common/mod.rs index edfe52cee..1efed1176 100644 --- a/libs/scrap/src/common/mod.rs +++ b/libs/scrap/src/common/mod.rs @@ -20,6 +20,8 @@ cfg_if! { pub mod drmtap_dl; #[cfg(all(target_os = "linux", feature = "drm"))] pub mod drm_reader; + #[cfg(all(target_os = "linux", feature = "drm"))] + pub mod drm_render; pub use self::linux::*; pub use self::wayland::set_map_err; pub use self::x11::PixelBuffer; diff --git a/src/ipc.rs b/src/ipc.rs index 35f3b8955..bb652aae6 100644 --- a/src/ipc.rs +++ b/src/ipc.rs @@ -68,6 +68,8 @@ use serde_derive::{Deserialize, Serialize}; use std::cell::Cell; #[cfg(any(target_os = "linux", target_os = "macos"))] use std::os::unix::fs::PermissionsExt; +#[cfg(all(target_os = "linux", feature = "drm"))] +use std::os::fd::{AsFd, AsRawFd, BorrowedFd, FromRawFd, OwnedFd, RawFd}; use std::{ collections::HashMap, sync::atomic::{AtomicBool, Ordering}, @@ -487,17 +489,32 @@ pub enum Data { // client replies `DrmStart{display}`, then the service streams `DrmFrame` + send_raw(BGRA) and // `DrmCursor` + send_raw(RGBA). A frame/cursor header is ALWAYS immediately followed by exactly // one `send_raw()` payload (the same header-then-raw pairing as `FileBlockFromCM`). This keeps - // the header extensible: a future zero-copy `DrmFrameDmabuf { fd, stride, modifier, .. }` slots - // in as a sibling variant without changing the transport. + // the header extensible. The zero-copy `DrmFrameDmabuf(DmabufDesc)` sibling below carries only a + // small JSON metadata descriptor; the scanout dma-buf fd rides an SCM_RIGHTS ancillary message on + // the same `DrmConn` send (see `DrmConn::send_msg`), so it has NO trailing `send_raw()` body. /// Client -> service: begin streaming the chosen display. #[cfg(all(target_os = "linux", feature = "drm"))] DrmStart { display: i32 }, /// Service -> client: the enumerated DRM displays (sent once, before frames). #[cfg(all(target_os = "linux", feature = "drm"))] DrmDisplayList(Vec), + /// Service -> client: the connector topology changed mid-stream (a monitor hotplug/unplug/modeset, + /// observed by the service's udev DRM-uevent listener). Carries the freshly-enumerated list so the + /// consumer can swap its sticky positive availability cache off the hot path, WITHOUT re-probing + /// `_drm` (which would trip the enumeration restart loop). Interleaved with frames on the same + /// stream; carries no `send_raw()` body and no fd. + #[cfg(all(target_os = "linux", feature = "drm"))] + DrmDisplaysChanged(Vec), /// Service -> client: a frame header; the packed BGRA pixels follow via `send_raw()`. + /// CPU-fallback path (old .so, no render node): pixels cross the wire. #[cfg(all(target_os = "linux", feature = "drm"))] DrmFrame { width: u32, height: u32 }, + /// Service -> client: a zero-copy dma-buf frame descriptor. The scanout fd is NOT a field; when + /// `desc.has_fd` it rides an SCM_RIGHTS ancillary message on the same `DrmConn::send_msg`, and + /// there is NO trailing `send_raw()` body. The unprivileged `--server` imports the fd and does + /// the EGL detile/convert itself (see `DmabufDesc`). + #[cfg(all(target_os = "linux", feature = "drm"))] + DrmFrameDmabuf(DmabufDesc), /// Service -> client: a hardware-cursor header; the RGBA pixels follow via `send_raw()`. #[cfg(all(target_os = "linux", feature = "drm"))] DrmCursor { @@ -513,7 +530,7 @@ pub enum Data { /// form of `scrap::drm_reader::DisplaySnapshot`; the server augments it with the Wayland /// logical geometry/scale, which needs the user session. #[cfg(all(target_os = "linux", feature = "drm"))] -#[derive(Debug, Serialize, Deserialize, Clone)] +#[derive(Debug, Serialize, Deserialize, Clone, PartialEq, Eq)] pub struct DrmDisplayInfo { pub name: String, pub crtc_id: u32, @@ -524,6 +541,41 @@ pub struct DrmDisplayInfo { pub active: bool, } +/// Serializable metadata descriptor of a scanout dma-buf, shipped over `_drm` as the JSON payload of +/// `Data::DrmFrameDmabuf`. It mirrors `scrap::drm_reader::drmtap_dmabuf_desc` field-for-field EXCEPT +/// the process-local `dma_buf_fd` (which never serializes — it rides SCM_RIGHTS ancillary), and adds +/// `buffer_id` (the producer's stable pool key) and `has_fd` (whether this message's `send_msg` +/// carries the fd, vs an import-once cache hit that omits it). The converter rebuilds a +/// `drmtap_dmabuf_desc` from these fields and overwrites its `dma_buf_fd` with the received fd. +#[cfg(all(target_os = "linux", feature = "drm"))] +#[derive(Debug, Serialize, Deserialize, Clone)] +pub struct DmabufDesc { + /// Producer-side stable pool key (e.g. fb_id + a connection epoch). Distinct from `fb_id`, which + /// is libdrmtap's import-once cache key. + pub buffer_id: u64, + pub width: u32, + pub height: u32, + /// DRM fourcc of the scanout. + pub format: u32, + /// DRM format modifier (tiling/compression). + pub modifier: u64, + /// KMS framebuffer id — libdrmtap's import-once cache key. 0 disables caching for this frame. + pub fb_id: u32, + /// Used entries in `offsets`/`pitches` (1..4); 0 is treated as 1. + pub num_planes: u32, + /// Per-plane byte offsets into the dma-buf (CCS main + aux + clear-color). + pub offsets: [u32; 4], + /// Per-plane strides in bytes; `pitches[0]` is the main-surface stride. + pub pitches: [u32; 4], + /// DRMTAP_EOTF_* (SDR=0, PQ=2, HLG=3). PQ triggers the HDR->SDR tone-map on convert. + pub hdr_eotf: u32, + /// Content/mastering peak luminance (cd/m2); 0 = unknown. + pub hdr_max_nits: u32, + /// True: this message's `send_msg` attaches the dma-buf fd in an SCM_RIGHTS cmsg. False: an + /// import-once cache hit for `fb_id` — no fd attached, converter reuses its cached EGLImage. + pub has_fd: bool, +} + #[tokio::main(flavor = "current_thread")] pub async fn start(postfix: &str) -> ResultType<()> { let mut incoming = new_listener(postfix).await?; @@ -1505,10 +1557,14 @@ pub(crate) fn drm_ipc_path() -> String { /// Connect (from the user `--server`) to the root service's `_drm` capture channel. Uses the /// derived `drm_ipc_path()` rather than `Config::ipc_path` since `_drm` is not a hbb_common -/// service postfix (Option 2 isolation — no shared-lib change). +/// service postfix (Option 2 isolation — no shared-lib change). Returns a [`DrmConn`] (bespoke +/// SCM_RIGHTS framing) rather than the `Framed<_, BytesCodec>` `ConnectionTmpl`: the `_drm` channel +/// must carry the scanout dma-buf fd as ancillary data, which the codec cannot do (see `DrmConn`). #[cfg(all(target_os = "linux", feature = "drm"))] -pub(crate) async fn connect_drm(ms_timeout: u64) -> ResultType> { - connect_with_path(ms_timeout, &drm_ipc_path()).await +pub(crate) async fn connect_drm(ms_timeout: u64) -> ResultType { + let path = drm_ipc_path(); + let stream = timeout(ms_timeout, tokio::net::UnixStream::connect(&path)).await??; + Ok(DrmConn::new(stream)) } /// Bind the `_drm` listener. Unlike `new_listener`, this does not route through hbb_common's @@ -1542,8 +1598,20 @@ async fn new_drm_listener() -> ResultType { enum DrmProducerMsg { /// Enumerated displays, sent once before any frame so the task can answer the handshake. Displays(Vec), - /// A captured frame header + its packed BGRA pixels. + /// A captured frame (split/zero-copy path): the serializable dma-buf descriptor plus the (owned) + /// scanout fd to hand to the peer via SCM_RIGHTS. The worker always produces a real `fd` here; the + /// async task's `ExportLedger` decides whether to actually attach it (`desc.has_fd`) or elide it as + /// an import-once cache hit. The `OwnedFd` is closed once the send has dup'd it into the peer (or + /// immediately, when elided). Frame { + desc: DmabufDesc, + fd: Option, + }, + /// A captured frame (CPU-mapped fallback path): a full packed-BGRA frame body. Used when the + /// loaded libdrmtap predates the split API (no `drmtap_grab_desc`) or the seat has no transferable + /// dma-buf (ENOTSUP). Forwarded as `Data::DrmFrame{width,height}` + `send_raw(BGRA)`, exactly like + /// the pre-split protocol, so an unprivileged converter is never required. + FrameCpu { width: u32, height: u32, data: Bytes, @@ -1570,11 +1638,119 @@ impl Drop for DrmStopGuard { } } +/// Producer-side fd-elision ledger (root `--service`, one per `_drm` connection). Decides, per +/// exported frame, whether the scanout dma-buf fd must ride an SCM_RIGHTS cmsg (`has_fd = true`) or +/// can be elided as an import-once cache hit (`has_fd = false`) because the peer's converter already +/// imported that `fb_id`. Keyed by `fb_id -> (modifier, dims)`; a change in any of those (a resize, +/// a modifier/tiling change, or a recycled fb_id that also changed geometry) forces a real fd, and a +/// modeset/hotplug that invalidates the CRTC ends the connection (so a reconnect starts with a fresh, +/// empty ledger — matching the peer's fresh, empty converter cache). +/// +/// SAFETY / CORRECTNESS: eliding relies solely on `(fb_id, modifier, dims)` uniquely identifying a +/// buffer, but the kernel can recycle an `fb_id` onto a *different* buffer with identical geometry +/// and modifier; eliding then would serve a stale EGLImage. libdrmtap's own import cache keys on +/// `fb_id + dma-buf inode` and can re-import ONLY when it is handed a real fd. Because always sending +/// the fd is cheap (the converter still imports once per `fb_id` and closes the surplus fd) and is +/// strictly safe, `DRM_FD_ELISION` defaults to `false` for v1 (always send). The ledger is fully +/// wired so flipping the const on enables elision once the recycled-fb_id case is validated. +#[cfg(all(target_os = "linux", feature = "drm"))] +const DRM_FD_ELISION: bool = false; + +#[cfg(all(target_os = "linux", feature = "drm"))] +struct SeenBuf { + modifier: u64, + dims: (u32, u32), + epoch: u32, +} + +#[cfg(all(target_os = "linux", feature = "drm"))] +struct ExportLedger { + seen: HashMap, + order: std::collections::VecDeque, // insertion order, for evict-oldest + epoch: u32, +} + +#[cfg(all(target_os = "linux", feature = "drm"))] +impl ExportLedger { + // Grow-once, hard-capped (preallocated model): a hostile/buggy peer or a fb_id churn cannot grow + // this unbounded; oldest keys are evicted so a real fd is simply re-sent for them later. + const MAX_LEDGER: usize = 32; + + fn new() -> Self { + Self { + seen: HashMap::new(), + order: std::collections::VecDeque::new(), + epoch: 0, + } + } + + /// Returns true if this frame's fd must be attached (new/changed/recycled buffer, caching + /// disabled, or elision off), false if the converter already holds `fb_id` imported. + fn should_send_fd(&mut self, desc: &DmabufDesc) -> bool { + // fb_id == 0 disables caching for that frame; elision-off always sends. + if !DRM_FD_ELISION || desc.fb_id == 0 { + return true; + } + let ident = SeenBuf { + modifier: desc.modifier, + dims: (desc.width, desc.height), + epoch: self.epoch, + }; + if let Some(prev) = self.seen.get(&desc.fb_id) { + if prev.modifier == ident.modifier + && prev.dims == ident.dims + && prev.epoch == ident.epoch + { + return false; // import-once cache hit: elide the fd + } + } else { + // New key: record insertion order and evict the oldest if at capacity. + if self.order.len() >= Self::MAX_LEDGER { + if let Some(old) = self.order.pop_front() { + self.seen.remove(&old); + } + } + self.order.push_back(desc.fb_id); + } + self.seen.insert(desc.fb_id, ident); + true + } +} + +/// Build a [`DrmConn`] from an already-authorized `_drm` `Connection` (root `--service` side). The +/// parity `Connection` wraps a tokio `UnixStream` but exposes no way to move it out, so we `dup()` +/// its fd into a fresh, independently-owned tokio `UnixStream` for the bespoke SCM_RIGHTS framing. +/// A dup gives a NEW fd number, which registers as its own epoll entry in tokio's reactor (reusing +/// the same fd number would double-register); the caller drops the parity `Connection` afterwards, +/// closing ITS fd, while the dup keeps the socket alive via the shared open file description. +#[cfg(all(target_os = "linux", feature = "drm"))] +fn dup_to_drm_conn(stream: &Connection) -> ResultType { + let raw = stream.inner.get_ref().as_raw_fd(); + let dup = unsafe { hbb_common::libc::dup(raw) }; + if dup < 0 { + return Err(std::io::Error::last_os_error().into()); + } + // SAFETY: `dup` is a freshly dup'd, owned fd for a connected SOCK_STREAM unix socket. + let std_stream = unsafe { std::os::unix::net::UnixStream::from_raw_fd(dup) }; + std_stream.set_nonblocking(true)?; + let tokio_stream = tokio::net::UnixStream::from_std(std_stream)?; + Ok(DrmConn::new(tokio_stream)) +} + /// Cached DRM display enumeration. The pre-warm populates it and each capture open refreshes it, so /// a consumer's handshake can send the display list without first paying a DRM enumeration open. #[cfg(all(target_os = "linux", feature = "drm"))] static DRM_DISPLAY_CACHE: std::sync::Mutex> = std::sync::Mutex::new(Vec::new()); +/// Monotonic generation bumped by the udev DRM-uevent listener ONLY when a connector-topology change +/// actually altered `DRM_DISPLAY_CACHE` (a monitor hotplug/unplug/modeset). Each live `handle_drm_conn` +/// forward loop watches this (one atomic load per frame) and, on a bump, pushes a `DrmDisplaysChanged` +/// with the fresh list to its consumer — the cheap live-refresh path that avoids a consumer re-probe. +/// `Release`/`Acquire` order it after the cache write so a reader that sees the new generation also sees +/// the new cache (the cache `Mutex` re-synchronizes the contents regardless). +#[cfg(all(target_os = "linux", feature = "drm"))] +static DRM_DISPLAY_GENERATION: std::sync::atomic::AtomicU64 = std::sync::atomic::AtomicU64::new(0); + /// Snapshot a reader's enumerated displays as the IPC `DrmDisplayInfo` form. `displays()` lists all /// device outputs regardless of the reader's target CRTC, so a capture reader can refresh the cache. #[cfg(all(target_os = "linux", feature = "drm"))] @@ -1605,6 +1781,110 @@ fn drm_displays_from_reader(reader: &mut scrap::drm_reader::DrmReader) -> Vec bool { + let mut is_drm = false; + let mut is_change = false; + for rec in msg.split(|&b| b == 0) { + if rec == b"SUBSYSTEM=drm" { + is_drm = true; + } else if rec == b"ACTION=change" || rec == b"HOTPLUG=1" { + is_change = true; + } + } + is_drm && is_change +} + +/// Listen for DRM connector hotplug/modeset uevents and refresh the display cache when the topology +/// actually changes. Uses a raw `NETLINK_KOBJECT_UEVENT` socket (the same hotplug stream udev consumes) +/// so no libudev dependency is added; the root `--service` already runs privileged and joining the +/// kernel-uevent multicast group needs no extra cap. On a real change it re-enumerates (off any hot +/// path — this is a dedicated thread, so the blocking `open`/`displays` is fine), and only when the +/// enumerated set differs does it swap `DRM_DISPLAY_CACHE` and bump `DRM_DISPLAY_GENERATION`; live +/// `handle_drm_conn` loops then push the fresh list to their consumers. Best-effort: if the socket is +/// unavailable it logs and returns, and DRM capture still works (a consumer reconnect re-reads the +/// fresh list) — just without the mid-session live refresh. +#[cfg(all(target_os = "linux", feature = "drm"))] +fn drm_udev_listener() { + use hbb_common::libc; + use std::sync::atomic::Ordering; + + let sock = unsafe { + libc::socket( + libc::AF_NETLINK, + libc::SOCK_DGRAM | libc::SOCK_CLOEXEC, + libc::NETLINK_KOBJECT_UEVENT, + ) + }; + if sock < 0 { + log::info!( + "drm: udev uevent socket unavailable ({}); hotplug refresh disabled", + std::io::Error::last_os_error() + ); + return; + } + // Own the fd so it is closed on every return / unwind path. + let _owned = unsafe { OwnedFd::from_raw_fd(sock) }; + let mut addr: libc::sockaddr_nl = unsafe { std::mem::zeroed() }; + addr.nl_family = libc::AF_NETLINK as u16; + // Group 1 = kernel-originated uevents (udev re-broadcasts on group 2); pid 0 => kernel assigns. + addr.nl_groups = 1; + let rc = unsafe { + libc::bind( + sock, + &addr as *const libc::sockaddr_nl as *const libc::sockaddr, + std::mem::size_of::() as libc::socklen_t, + ) + }; + if rc < 0 { + log::info!( + "drm: udev uevent bind failed ({}); hotplug refresh disabled", + std::io::Error::last_os_error() + ); + return; + } + log::info!("drm: udev DRM-uevent listener started"); + // Fixed-size receive buffer (preallocated model): a uevent is well under 8 KiB; a rare larger + // datagram is truncated by `recv` and simply re-enumerates on the next matching event. + let mut buf = [0u8; 8192]; + loop { + let n = unsafe { libc::recv(sock, buf.as_mut_ptr() as *mut libc::c_void, buf.len(), 0) }; + if n <= 0 { + let err = std::io::Error::last_os_error(); + if n < 0 && err.kind() == std::io::ErrorKind::Interrupted { + continue; + } + log::info!("drm: udev uevent recv ended ({err}); hotplug refresh stopped"); + break; + } + if !uevent_is_drm_change(&buf[..n as usize]) { + continue; + } + // Re-enumerate and diff. Only a real change swaps the cache + bumps the generation, so a + // uevent that does not alter the captured topology stays silent (no consumer churn). + if let Some(mut r) = scrap::drm_reader::DrmReader::open(None, 0) { + let fresh = drm_displays_from_reader(&mut r); + let changed = { + let mut cache = DRM_DISPLAY_CACHE.lock().unwrap(); + if *cache != fresh { + *cache = fresh; + true + } else { + false + } + }; + if changed { + DRM_DISPLAY_GENERATION.fetch_add(1, Ordering::Release); + log::info!("drm: connector topology changed (udev); display cache refreshed"); + } + } + } +} + /// Best-effort warm-up at listener start: loads libdrmtap, initializes EGL, enumerates displays into /// the cache, and maps the first framebuffer once. Moves that one-time cost (which otherwise lands /// on the first consumer and can push the first frame past the client's initial-frame timeout) off @@ -1616,7 +1896,16 @@ fn drm_prewarm() { Some(mut r) => { let displays = drm_displays_from_reader(&mut r); let n = displays.len(); - let _ = r.grab(); // force the first framebuffer map / import + // Warm the first framebuffer export. On the split path, grab_desc() exports a dma-buf fd + // WITHOUT loading libEGL/libGLESv2 into the root service (the convert now runs in the + // unprivileged --server); only an old .so (no grab_desc) still force-maps via grab(). + if r.supports_grab_desc() { + if let Ok((fd, _desc)) = r.grab_desc() { + drop(fd); // close the warm-up fd; we only wanted to prime the device/import path + } + } else { + let _ = r.grab(); + } *DRM_DISPLAY_CACHE.lock().unwrap() = displays; log::info!("drm: pre-warm ok ({n} displays) in {:?}", t.elapsed()); } @@ -1640,6 +1929,10 @@ pub async fn start_drm() { // Warm libdrmtap/EGL + enumeration off-thread so the first consumer does not pay that // one-time cost on its critical path. std::thread::spawn(drm_prewarm); + // Watch for connector hotplug/modeset uevents so a mid-session topology change refreshes + // the display cache and is pushed to live consumers (best-effort; own thread since it + // blocks on recv and re-enumeration is a blocking `!Send` open). + std::thread::spawn(drm_udev_listener); loop { match incoming.next().await { Some(Ok(stream)) => { @@ -1673,7 +1966,7 @@ pub async fn start_drm() { /// messages to the wire. On any error / disconnect it returns; the `DrmStopGuard` plus dropping the /// channels tears the worker down, and the client falls back to PipeWire/portal. #[cfg(all(target_os = "linux", feature = "drm"))] -async fn handle_drm_conn(mut stream: Connection) -> ResultType<()> { +async fn handle_drm_conn(stream: Connection) -> ResultType<()> { use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering}; use std::sync::Arc; @@ -1707,6 +2000,14 @@ async fn handle_drm_conn(mut stream: Connection) -> ResultType<()> { } let _conn_guard = DrmConnGuard; + // Move the authorized `_drm` stream onto the bespoke SCM_RIGHTS framing (see `DrmConn`). ALL + // further traffic — display list, `DrmStart`, frame descriptors + their ancillary fd, and the + // cursor / CPU-fallback bodies — goes through `conn` so no `Framed` read buffer ever competes with + // a `recvmsg` for the fd. The parity `Connection` (used only for the authorization above) is + // dropped here, closing its fd; the dup inside `conn` keeps the socket alive. + let mut conn = dup_to_drm_conn(&stream)?; + drop(stream); + // worker -> task: display list, frames, cursor (bounded = backpressure). let (frame_tx, mut frame_rx) = tokio::sync::mpsc::channel::(2); // task -> worker: the chosen CRTC, sent once after the client's DrmStart. @@ -1726,14 +2027,16 @@ async fn handle_drm_conn(mut stream: Connection) -> ResultType<()> { return Ok(()); } }; - stream.send(&Data::DrmDisplayList(displays.clone())).await?; + conn.send_msg(&Data::DrmDisplayList(displays.clone()), None).await?; - // Wait for the client to choose a display before streaming. + // Wait for the client to choose a display before streaming. `recv_msg_timeout2` gates only the + // wait for the first byte, so a timeout leaves the stream at a clean frame boundary. let display_idx = loop { - match stream.next_timeout(10_000).await? { - Some(Data::DrmStart { display }) => break display, - Some(_) => continue, - None => return Ok(()), + match conn.recv_msg_timeout2(10_000).await { + Some(Ok((Data::DrmStart { display }, _fd))) => break display, + Some(Ok((_, _fd))) => continue, // ignore unexpected messages; drop any stray fd + Some(Err(e)) => return Err(e), + None => return Ok(()), // timed out: client never chose a display } }; // Resolve the chosen display's CRTC. `displays` here is already filtered to @@ -1760,16 +2063,43 @@ async fn handle_drm_conn(mut stream: Connection) -> ResultType<()> { } // Forward frames + cursor updates until the worker ends or the client disconnects (a wire send - // error on a dropped client propagates out and tears the worker down via the guard). + // error on a dropped client propagates out and tears the worker down via the guard). The + // per-connection `ExportLedger` decides, for the zero-copy path, whether each frame's fd must ride + // an SCM_RIGHTS cmsg or can be elided as an import-once cache hit. + let mut ledger = ExportLedger::new(); + // Live hotplug: the udev listener bumps DRM_DISPLAY_GENERATION when the connector topology changes. + // Seed from the value current at handshake (the list already sent reflects it) and, whenever it + // moves, push the fresh list to this consumer. Piggybacked on the frame cadence so it costs only one + // atomic load per frame; a genuinely idle stream tears down after MAX_STALLED and the consumer + // reconnects to a fresh list anyway. + let mut seen_gen = DRM_DISPLAY_GENERATION.load(Ordering::Acquire); while let Some(msg) = frame_rx.recv().await { + let gen = DRM_DISPLAY_GENERATION.load(Ordering::Acquire); + if gen != seen_gen { + seen_gen = gen; + let fresh = DRM_DISPLAY_CACHE.lock().unwrap().clone(); + if !fresh.is_empty() { + conn.send_msg(&Data::DrmDisplaysChanged(fresh), None).await?; + } + } match msg { - DrmProducerMsg::Frame { + DrmProducerMsg::Frame { mut desc, fd } => { + // The worker always supplies a real fd; the ledger decides whether to attach it. + let send_fd = fd.is_some() && ledger.should_send_fd(&desc); + desc.has_fd = send_fd; + let borrowed = if send_fd { fd.as_ref().map(|f| f.as_fd()) } else { None }; + conn.send_msg(&Data::DrmFrameDmabuf(desc), borrowed).await?; + // `fd` (OwnedFd) is closed here whether or not it was attached (the cmsg dup'd it into + // the peer). Closing immediately bounds our fd usage to ~1 in flight per frame. + } + DrmProducerMsg::FrameCpu { width, height, data, } => { - stream.send(&Data::DrmFrame { width, height }).await?; - stream.send_raw(data).await?; + // CPU-mapped fallback: pixels cross the wire, exactly like the pre-split protocol. + conn.send_msg(&Data::DrmFrame { width, height }, None).await?; + conn.send_raw(data).await?; } DrmProducerMsg::Cursor { id, @@ -1779,16 +2109,18 @@ async fn handle_drm_conn(mut stream: Connection) -> ResultType<()> { hoty, colors, } => { - stream - .send(&Data::DrmCursor { + conn.send_msg( + &Data::DrmCursor { id, width, height, hotx, hoty, - }) - .await?; - stream.send_raw(Bytes::from(colors)).await?; + }, + None, + ) + .await?; + conn.send_raw(Bytes::from(colors)).await?; } DrmProducerMsg::Displays(_) => {} } @@ -1862,28 +2194,66 @@ fn drm_capture_worker( t_open.elapsed() ); + // A per-connection buffer-pool epoch so `buffer_id` is unique across connections even for the same + // fb_id (the consumer may key a pool by buffer_id). + 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); + + // Prefer the zero-copy split export (root does NO EGL / convert / copy). If the loaded libdrmtap + // predates the split API, or grab_desc later reports ENOTSUP (no transferable dma-buf on this + // seat), fall back to the CPU-mapped path for this connection (pixels cross the wire). + let mut use_dmabuf = reader.supports_grab_desc(); + let mut last_cursor_id: u64 = 0; let mut stalled: u32 = 0; let mut logged_first = false; while !stop.load(Ordering::Relaxed) { - match reader.grab() { - Ok((buf, w, h)) => { + // 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 = if use_dmabuf { + match reader.grab_desc() { + Ok((fd, d)) => Ok(DrmProducerMsg::Frame { + desc: DmabufDesc { + buffer_id: (d.fb_id as u64) | ((conn_epoch as u64) << 32), + width: d.width, + height: d.height, + format: d.format, + modifier: d.modifier, + fb_id: d.fb_id, + num_planes: d.num_planes, + offsets: d.offsets, + pitches: d.pitches, + hdr_eotf: d.hdr_eotf, + hdr_max_nits: d.hdr_max_nits, + has_fd: true, // the async task's ExportLedger may downgrade this + }, + fd: Some(fd), + }), + Err(err) => Err(err), + } + } else { + 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) => { stalled = 0; if !logged_first { logged_first = true; log::debug!( - "drm: first frame {w}x{h} for crtc {target_crtc} in {:?}", - t_conn.elapsed() + "drm: first frame for crtc {target_crtc} in {:?} ({} path)", + t_conn.elapsed(), + if use_dmabuf { "dma-buf" } else { "cpu" } ); } - if frame_tx - .blocking_send(DrmProducerMsg::Frame { - width: w as u32, - height: h as u32, - data: Bytes::copy_from_slice(buf), - }) - .is_err() - { + if frame_tx.blocking_send(msg).is_err() { break; } } @@ -1896,6 +2266,17 @@ fn drm_capture_worker( std::thread::sleep(FRAME_INTERVAL); continue; } + 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(). + log::warn!( + "drm: grab_desc unsupported ({err}); switching to CPU-mapped fallback for this connection" + ); + use_dmabuf = false; + logged_first = false; + continue; + } Err(err) => { log::warn!("drm: capture error: {err}; closing _drm connection"); break; @@ -1997,6 +2378,336 @@ where } } +/// Ancillary-fd transport for the `_drm` channel. +/// +/// `ConnectionTmpl`'s `Framed<_, BytesCodec>` cannot carry (nor collect) an SCM_RIGHTS control +/// message: tokio's `AsyncRead` never does a `recvmsg` with a control buffer, so a fd sent alongside +/// a `Framed` byte-frame is silently dropped on receive, and interleaving a raw `sendmsg` with the +/// codec desyncs its internal read buffer. So the WHOLE `_drm` channel moves onto this bespoke +/// length-prefixed `sendmsg`/`recvmsg` framing, owning the raw `tokio::net::UnixStream` directly: +/// handshake (`DrmDisplayList`/`DrmStart`), frame descriptors, and the CPU-fallback/cursor bodies all +/// go through it so no `Framed` read buffer ever competes with a `recvmsg`. +/// +/// Framing: each frame is a 4-byte big-endian length prefix + payload. `send_msg`/`recv_msg` carry a +/// JSON `Data`; `send_raw`/`next_raw` carry an opaque body. The dma-buf fd (when present) rides an +/// SCM_RIGHTS cmsg bound to the frame's first (prefix) byte, so reading the prefix with a control +/// buffer reliably collects it (`MSG_CTRUNC` is rejected). Reads use exact-length loops so they never +/// cross a frame boundary and thus never discard a following frame's ancillary fd. +#[cfg(all(target_os = "linux", feature = "drm"))] +pub struct DrmConn { + /// The raw stream. Obtained from `connect_drm` (client) or the accepted `_drm` listener stream + /// (service). All framing is done by hand on this fd; there is no `Framed` codec. + stream: tokio::net::UnixStream, + /// Grow-once accumulation buffer for `recv_msg`/`next_raw` length-prefixed reads (preallocated + /// model: it grows to the largest frame seen and is then reused, never per-frame reallocated). + read_buf: Vec, +} + +/// Cap on a JSON `Data` message read by `recv_msg` (headers/handshake are tiny; this only bounds a +/// hostile/oversized length prefix). Distinct from the raw-body cap because a body can be a whole +/// CPU-fallback frame. +#[cfg(all(target_os = "linux", feature = "drm"))] +const MAX_DRM_JSON_BYTES: usize = 8 * 1024 * 1024; +/// Cap on a raw body read by `next_raw` (CPU-fallback BGRA / cursor RGBA). Covers a 256 MiB 8K +/// scanout (`DrmReader` bounds a frame to that) with margin. +#[cfg(all(target_os = "linux", feature = "drm"))] +const MAX_DRM_RAW_BYTES: usize = 512 * 1024 * 1024; +/// Control-buffer capacity for one SCM_RIGHTS cmsg carrying a single fd. `CMSG_SPACE(sizeof(int))` is +/// 24 bytes on our targets; 64 gives headroom and the `align(8)` matches `cmsghdr` alignment. +#[cfg(all(target_os = "linux", feature = "drm"))] +const DRM_CMSG_CAP: usize = 64; + +/// Aligned storage for the SCM_RIGHTS control buffer (`msg_control` must be `cmsghdr`-aligned). +#[cfg(all(target_os = "linux", feature = "drm"))] +#[repr(align(8))] +struct DrmCmsgBuf([u8; DRM_CMSG_CAP]); + +/// One non-blocking `sendmsg`: writes `buf` and, when `pass_fd` is `Some`, attaches exactly one +/// SCM_RIGHTS cmsg carrying that fd. The cmsg is attached ONLY when a fd is present (a -1 fd in an +/// SCM_RIGHTS cmsg fails the whole call). Returns bytes sent, or a `WouldBlock`/other io error. +/// +/// SAFETY: `fd` must be a valid open socket fd; `buf` a valid readable slice; `pass_fd` (if any) a +/// valid open fd. The ancillary data is delivered by the kernel with the first byte of `buf`. +#[cfg(all(target_os = "linux", feature = "drm"))] +unsafe fn drm_sendmsg(fd: RawFd, buf: &[u8], pass_fd: Option) -> std::io::Result { + use hbb_common::libc; + let mut iov = libc::iovec { + iov_base: buf.as_ptr() as *mut libc::c_void, + iov_len: buf.len(), + }; + let mut msg: libc::msghdr = std::mem::zeroed(); + msg.msg_iov = &mut iov; + msg.msg_iovlen = 1; + let mut cbuf = DrmCmsgBuf([0u8; DRM_CMSG_CAP]); + if let Some(sfd) = pass_fd { + msg.msg_control = cbuf.0.as_mut_ptr() as *mut libc::c_void; + msg.msg_controllen = libc::CMSG_SPACE(std::mem::size_of::() as u32) as _; + let cmsg = libc::CMSG_FIRSTHDR(&msg); + // Sized above so CMSG_FIRSTHDR is non-null; guard anyway to avoid UB on any platform quirk. + if cmsg.is_null() { + return Err(std::io::Error::new( + std::io::ErrorKind::Other, + "drm: CMSG_FIRSTHDR null", + )); + } + (*cmsg).cmsg_level = libc::SOL_SOCKET; + (*cmsg).cmsg_type = libc::SCM_RIGHTS; + (*cmsg).cmsg_len = libc::CMSG_LEN(std::mem::size_of::() as u32) as _; + let sfd_c: libc::c_int = sfd; + std::ptr::copy_nonoverlapping( + &sfd_c as *const libc::c_int as *const u8, + libc::CMSG_DATA(cmsg), + std::mem::size_of::(), + ); + } + let n = libc::sendmsg(fd, &msg, libc::MSG_NOSIGNAL); + if n < 0 { + Err(std::io::Error::last_os_error()) + } else { + Ok(n as usize) + } +} + +/// One non-blocking `recvmsg` into `buf` with a control buffer. Collects at most one SCM_RIGHTS fd +/// (any surplus fds are closed); rejects a truncated cmsg (`MSG_CTRUNC`) as a hard error after closing +/// whatever it parsed. Returns (bytes read, fd). Received fds are `O_CLOEXEC` (`MSG_CMSG_CLOEXEC`). +/// +/// SAFETY: `fd` must be a valid open socket fd; `buf` a valid writable slice. +#[cfg(all(target_os = "linux", feature = "drm"))] +unsafe fn drm_recvmsg(fd: RawFd, buf: &mut [u8]) -> std::io::Result<(usize, Option)> { + use hbb_common::libc; + let mut iov = libc::iovec { + iov_base: buf.as_mut_ptr() as *mut libc::c_void, + iov_len: buf.len(), + }; + let mut cbuf = DrmCmsgBuf([0u8; DRM_CMSG_CAP]); + let mut msg: libc::msghdr = std::mem::zeroed(); + msg.msg_iov = &mut iov; + msg.msg_iovlen = 1; + msg.msg_control = cbuf.0.as_mut_ptr() as *mut libc::c_void; + msg.msg_controllen = cbuf.0.len() as _; + let n = libc::recvmsg(fd, &mut msg, libc::MSG_CMSG_CLOEXEC); + if n < 0 { + return Err(std::io::Error::last_os_error()); + } + // Walk the cmsgs; keep the first SCM_RIGHTS fd, close any extras. Each parsed int is wrapped in an + // OwnedFd immediately so it is always closed on drop (no fd leak on any error path below). + let mut got: Option = None; + let mut cmsg = libc::CMSG_FIRSTHDR(&msg); + while !cmsg.is_null() { + if (*cmsg).cmsg_level == libc::SOL_SOCKET && (*cmsg).cmsg_type == libc::SCM_RIGHTS { + let data = libc::CMSG_DATA(cmsg); + let hdr = libc::CMSG_LEN(0) as usize; + let payload = ((*cmsg).cmsg_len as usize).saturating_sub(hdr); + let count = payload / std::mem::size_of::(); + for i in 0..count { + let mut rawfd: libc::c_int = -1; + std::ptr::copy_nonoverlapping( + data.add(i * std::mem::size_of::()), + &mut rawfd as *mut libc::c_int as *mut u8, + std::mem::size_of::(), + ); + if rawfd >= 0 { + let owned = OwnedFd::from_raw_fd(rawfd); + if got.is_none() { + got = Some(owned); + } // else: surplus fd, dropped here -> closed + } + } + } + cmsg = libc::CMSG_NXTHDR(&msg, cmsg); + } + // A truncated control message means the kernel dropped fd(s) that did not fit: fail rather than + // proceed with a missing/partial fd (drop `got` so anything parsed is closed first). + if msg.msg_flags & libc::MSG_CTRUNC != 0 { + drop(got); + return Err(std::io::Error::new( + std::io::ErrorKind::Other, + "drm: truncated SCM_RIGHTS control message (MSG_CTRUNC)", + )); + } + Ok((n as usize, got)) +} + +/// Write all of `buf` to `stream`, attaching `pass_fd` (if any) to the FIRST byte (the kernel binds +/// SCM_RIGHTS ancillary to the first data byte of the `sendmsg` that carried it). Loops on +/// `WouldBlock` via `writable()`; the fd is attached only until the first `sendmsg` sends >= 1 byte. +#[cfg(all(target_os = "linux", feature = "drm"))] +async fn drm_write_all( + stream: &tokio::net::UnixStream, + mut buf: &[u8], + mut pass_fd: Option, +) -> ResultType<()> { + while !buf.is_empty() { + stream.writable().await?; + let raw = stream.as_raw_fd(); + let chunk = buf; + let fd_now = pass_fd; + match stream.try_io(tokio::io::Interest::WRITABLE, || unsafe { + drm_sendmsg(raw, chunk, fd_now) + }) { + Ok(0) => bail!("drm: socket write returned 0 (peer closed)"), + Ok(n) => { + pass_fd = None; // ancillary delivered with these bytes; do not re-send it + buf = &buf[n..]; + } + Err(ref e) if e.kind() == std::io::ErrorKind::WouldBlock => continue, + Err(e) => return Err(e.into()), + } + } + Ok(()) +} + +/// Write one length-prefixed frame: a 4-byte big-endian length + payload, with `pass_fd` (if any) +/// riding the prefix's first byte. +#[cfg(all(target_os = "linux", feature = "drm"))] +async fn drm_send_frame( + stream: &tokio::net::UnixStream, + payload: &[u8], + pass_fd: Option, +) -> ResultType<()> { + if payload.len() > u32::MAX as usize { + bail!("drm: frame too large ({} bytes)", payload.len()); + } + let prefix = (payload.len() as u32).to_be_bytes(); + // The fd rides the prefix (its first byte); the payload carries no ancillary. + drm_write_all(stream, &prefix, pass_fd).await?; + drm_write_all(stream, payload, None).await?; + Ok(()) +} + +/// Read exactly `buf.len()` bytes from `stream`. When `want_cmsg` is true, the FIRST read uses a +/// control buffer to collect an SCM_RIGHTS fd (which the sender bound to the frame's first byte); +/// subsequent reads within the same frame are plain. Returns the collected fd, if any. +#[cfg(all(target_os = "linux", feature = "drm"))] +async fn drm_read_full( + stream: &tokio::net::UnixStream, + buf: &mut [u8], + want_cmsg: bool, +) -> ResultType> { + use hbb_common::libc; + let mut off = 0usize; + let mut got: Option = None; + while off < buf.len() { + stream.readable().await?; + let raw = stream.as_raw_fd(); + // Only the first read of a frame carries the fd (bound to byte 0); after that, plain reads. + let use_cmsg = want_cmsg && got.is_none(); + let n = { + let dst: &mut [u8] = &mut buf[off..]; + match stream.try_io(tokio::io::Interest::READABLE, move || unsafe { + if use_cmsg { + drm_recvmsg(raw, dst) + } else { + let m = libc::read(raw, dst.as_mut_ptr() as *mut libc::c_void, dst.len()); + if m < 0 { + Err(std::io::Error::last_os_error()) + } else { + Ok((m as usize, None)) + } + } + }) { + Ok((0, _fd)) => bail!("drm: socket closed by peer"), + Ok((m, fd)) => { + if let Some(f) = fd { + if got.is_none() { + got = Some(f); + } + } + m + } + Err(ref e) if e.kind() == std::io::ErrorKind::WouldBlock => continue, + Err(e) => return Err(e.into()), + } + }; + off += n; + } + Ok(got) +} + +#[cfg(all(target_os = "linux", feature = "drm"))] +impl DrmConn { + /// Take ownership of an already-connected/accepted raw `_drm` stream. + pub fn new(stream: tokio::net::UnixStream) -> Self { + Self { + stream, + read_buf: Vec::new(), + } + } + + /// Send one `Data` message (JSON, length-prefixed). When `fd` is `Some`, attach exactly one + /// SCM_RIGHTS cmsg carrying that fd on the SAME frame as the payload (a -1 in an SCM_RIGHTS cmsg + /// fails the whole call, so the cmsg is attached ONLY when a fd is present). `fd` is borrowed so + /// the caller keeps ownership and closes it after the send has dup'd it into the peer. + pub async fn send_msg(&mut self, data: &Data, fd: Option>) -> ResultType<()> { + let payload = serde_json::to_vec(data)?; + let pass_fd = fd.map(|f| f.as_raw_fd()); + drm_send_frame(&self.stream, &payload, pass_fd).await + } + + /// Receive one `Data` message plus any dma-buf fd delivered via SCM_RIGHTS. Reads the 4-byte + /// length prefix (with a `CMSG_SPACE(size_of::())` control buffer that collects the fd bound + /// to the frame's first byte, rejecting `MSG_CTRUNC`), then the payload into the reusable + /// `read_buf`. Returns the decoded `Data` and an `OwnedFd` iff one arrived. + pub async fn recv_msg(&mut self) -> ResultType<(Data, Option)> { + let mut prefix = [0u8; 4]; + let fd = drm_read_full(&self.stream, &mut prefix, true).await?; + let len = u32::from_be_bytes(prefix) as usize; + if len > MAX_DRM_JSON_BYTES { + // `fd` (if any) is closed on drop. + bail!("drm: message length {len} exceeds cap {MAX_DRM_JSON_BYTES}"); + } + if self.read_buf.len() < len { + self.read_buf.resize(len, 0); + } + // Disjoint field borrows: &self.stream (read) + &mut self.read_buf (dest). No fd on the body. + drm_read_full(&self.stream, &mut self.read_buf[..len], false).await?; + let data: Data = serde_json::from_slice(&self.read_buf[..len])?; + Ok((data, fd)) + } + + /// Cancel-safe timeout wrapper around `recv_msg`, mirroring `ConnectionTmpl::next_timeout2`, so a + /// dropped consumer re-checks its `stop` flag between frames. `None` on timeout. The timeout gates + /// ONLY the wait for the first byte (`readable()` consumes nothing), so a fired timeout leaves the + /// stream at a clean frame boundary and never strands a partial frame or its fd. + pub async fn recv_msg_timeout2( + &mut self, + ms_timeout: u64, + ) -> Option)>> { + // Bind the readiness result to a `let` so the borrowed `readable()` future temporary is dropped + // at the `;` (releasing `&self.stream`) BEFORE `recv_msg()` takes `&mut self` in an arm. + let ready = timeout(ms_timeout, self.stream.readable()).await; + match ready { + Err(_) => None, // timed out at a frame boundary + Ok(Err(e)) => Some(Err(e.into())), + Ok(Ok(())) => Some(self.recv_msg().await), + } + } + + /// Send a raw length-prefixed body (cursor pixels, CPU-fallback BGRA). Parity with + /// `ConnectionTmpl::send_raw`, over the same manual framing (never carries an fd). + pub async fn send_raw(&mut self, data: Bytes) -> ResultType<()> { + drm_send_frame(&self.stream, &data, None).await + } + + /// Receive a raw length-prefixed body. Parity with `ConnectionTmpl::next_raw`. A raw body never + /// carries an fd; a stray fd (protocol desync) is collected by `drm_read_full` and dropped/closed. + pub async fn next_raw(&mut self) -> ResultType { + let mut prefix = [0u8; 4]; + if drm_read_full(&self.stream, &mut prefix, true).await?.is_some() { + log::warn!("drm: unexpected fd on a raw-body frame; dropping"); + } + let len = u32::from_be_bytes(prefix) as usize; + if len > MAX_DRM_RAW_BYTES { + bail!("drm: raw body length {len} exceeds cap {MAX_DRM_RAW_BYTES}"); + } + let mut out = bytes::BytesMut::new(); + out.resize(len, 0); + drm_read_full(&self.stream, &mut out[..], false).await?; + Ok(out) + } +} + #[tokio::main(flavor = "current_thread")] pub async fn get_config(name: &str) -> ResultType> { get_config_async(name, 1_000).await diff --git a/src/server/drm_capturer.rs b/src/server/drm_capturer.rs index 04d80e1c7..5457c7921 100644 --- a/src/server/drm_capturer.rs +++ b/src/server/drm_capturer.rs @@ -1,22 +1,34 @@ // Server-side (`--server`, unprivileged) consumer of the root `--service`'s DRM/KMS capture stream. // -// The architecture pivot moved the scanout read into the root service; this process no longer -// links or dlopens libdrmtap. It connects to the service's `_drm` channel, learns the display -// geometry from the service, and pulls packed-BGRA frames. This mirrors the Windows -// `portable_service` CapturerPortable split (a privileged process captures, this process presents), -// but over rustdesk's own IPC instead of shared memory. +// The phase-2 split moved only the privileged EXPORT (open + grab the scanout dma-buf fd) into the +// root service; the EGL detile / RGBA convert now runs HERE, in the unprivileged process. So this +// process DOES dlopen libdrmtap again (its unprivileged render half: `drmtap_open_render` + +// `drmtap_convert_dmabuf`), holding one render-node context on the receive thread. It connects to +// the service's `_drm` channel, learns the display geometry, then on each frame receives a small +// dma-buf descriptor + the scanout fd (over SCM_RIGHTS) and converts it to linear pixels locally. +// This mirrors the Windows `portable_service` CapturerPortable split (a privileged process captures, +// this process presents), but over rustdesk's own IPC and with only the fd (not the pixels) crossing +// the socket. A CPU-fallback path is kept: an older `.so` or a seat with no transferable dma-buf +// makes the service send `DrmFrame` + packed-BGRA over the wire, which this side stores as-is. // // `TraitCapturer::frame()` is synchronous (the encoder loop calls it) while the IPC receive is // async, so a dedicated background thread runs the receive loop and keeps only the newest frame // (latest-wins, so a slow encoder never backs the socket up). `frame()` returns that frame as a // borrowed `PixelBuffer`, `WouldBlock` when nothing new arrived within the timeout, and a hard // `Err` once the stream ends (the caller then rebuilds the capturer or falls back to PipeWire). +// +// The render context (`RenderConverter`) is created ONCE on the receive thread and dropped there on +// exit (NOT in `IpcDrmCapturer::Drop`): libdrmtap's EGL state + import-once EGLImage cache are +// thread-local, so both convert and close must run on the same thread. use crate::ipc::{connect_drm, Data, DrmDisplayInfo}; use hbb_common::{anyhow::anyhow, log, message_proto::DisplayInfo, tokio, ResultType}; +use scrap::drm_render::RenderConverter; +use scrap::drmtap_dl::drmtap_dmabuf_desc; use scrap::{Frame, Pixfmt, PixelBuffer, TraitCapturer}; use std::collections::BTreeMap; use std::io; +use std::os::fd::{AsRawFd, RawFd}; use std::sync::atomic::{AtomicBool, Ordering}; use std::sync::{Arc, Condvar, Mutex}; use std::time::{Duration, Instant}; @@ -26,8 +38,12 @@ use std::time::{Duration, Instant}; const HANDSHAKE_TIMEOUT_MS: u64 = 3000; struct FrameSlot { - // (width, height, packed-BGRA) of the newest frame not yet consumed by `frame()`; latest-wins. - latest: Option<(usize, usize, Vec)>, + // (width, height, pixel format, packed pixels) of the newest frame not yet consumed by + // `frame()`; latest-wins. The pixel format is carried per frame because the split convert path + // reads it from the actual convert output (XRGB8888 -> BGRA, XBGR8888 -> RGBA) rather than + // assuming BGRA; the CPU-fallback path stores BGRA. The row stride is recoverable from + // `pixels.len() / height` (the convert output may carry a padded stride). + latest: Option<(usize, usize, Pixfmt, Vec)>, // Set once the stream ends so `frame()` returns a hard error (triggers a capturer rebuild). ended: Option, } @@ -47,6 +63,10 @@ pub struct IpcDrmCapturer { cur: Vec, cur_w: usize, cur_h: usize, + // Pixel format of `cur`, taken from the frame stored in the slot (BGRA on the CPU-fallback path; + // BGRA/RGBA per the convert output on the dma-buf path). Honored by `frame()` instead of a + // hardcoded BGRA so an EGL-less / source-order convert is not shipped with red/blue swapped. + cur_fmt: Pixfmt, // Whether this capturer ever delivered a frame. Used to distinguish a stream that fails to // produce ANY frame (a permanent grab failure — unsupported scanout on that CRTC) from a normal // teardown, so DRM can fall back to PipeWire for that display instead of rebuilding it forever. @@ -117,6 +137,7 @@ impl IpcDrmCapturer { cur: Vec::new(), cur_w: 0, cur_h: 0, + cur_fmt: Pixfmt::BGRA, got_frame: false, }, displays, @@ -149,11 +170,12 @@ impl TraitCapturer for IpcDrmCapturer { slot = guard; } // Deliver a pending frame before surfacing an end, so the last frame is not dropped. - if let Some((w, h, buf)) = slot.latest.take() { + if let Some((w, h, fmt, buf)) = slot.latest.take() { drop(slot); self.cur = buf; self.cur_w = w; self.cur_h = h; + self.cur_fmt = fmt; if !self.got_frame { // First frame of this session: DRM capture works for this display, clear its // failure streak. @@ -187,7 +209,7 @@ impl TraitCapturer for IpcDrmCapturer { } Ok(Frame::PixelBuffer(PixelBuffer::new( &self.cur, - Pixfmt::BGRA, + self.cur_fmt, self.cur_w, self.cur_h, ))) @@ -212,42 +234,114 @@ async fn recv_thread( return; } }; - let displays = match conn.next_timeout(HANDSHAKE_TIMEOUT_MS).await { - Ok(Some(Data::DrmDisplayList(v))) => v, - Ok(other) => { + let displays = match conn.recv_msg_timeout2(HANDSHAKE_TIMEOUT_MS).await { + Some(Ok((Data::DrmDisplayList(v), _fd))) => v, + Some(Ok((other, _fd))) => { let _ = tx.send(Err(anyhow!("expected DrmDisplayList, got {:?}", other))); return; } - Err(err) => { + Some(Err(err)) => { let _ = tx.send(Err(err)); return; } + None => { + let _ = tx.send(Err(anyhow!("timed out waiting for DrmDisplayList"))); + return; + } }; - if let Err(err) = conn.send(&Data::DrmStart { display }).await { + if let Err(err) = conn.send_msg(&Data::DrmStart { display }, None).await { let _ = tx.send(Err(err)); return; } let _ = tx.send(Ok(displays)); + // Open the unprivileged render-node convert context ONCE, on THIS thread; it is dropped on this + // same thread when the loop exits (its EGL state + import-once cache are thread-local). `None` + // means no usable render node (a locked-down seat, or an old `.so` without the split symbols): the + // CPU-fallback `DrmFrame` path still works, but a `DrmFrameDmabuf` we cannot convert ends the + // stream so the caller falls back (PipeWire per-display). + let mut converter = RenderConverter::open_render(); + if converter.is_none() { + log::info!( + "drm: no render-node convert context (drmtap_open_render failed or old .so); \ + only the CPU-fallback frame path will work on this stream" + ); + } + // Stream until stopped or the connection ends. Poll the header read with a short timeout (rather - // than blocking indefinitely on `next()`) so a dropped capturer re-checks `stop` and tears down - // promptly even when the producer has stalled (no frames arriving). A header is always followed - // immediately by its `next_raw()` body, so only the header read needs the poll. + // than blocking indefinitely) so a dropped capturer re-checks `stop` and tears down promptly even + // when the producer has stalled (no frames arriving). A dma-buf frame carries its fd inline on the + // header (no body); a CPU-fallback frame and a cursor each carry a `next_raw()` body immediately + // after their header, so only the header read needs the poll. let end_reason = loop { if stop.load(Ordering::SeqCst) { break "stopped".to_owned(); } - let msg = match conn.next_timeout2(200).await { + // The decoded `Data` plus any SCM_RIGHTS fd that rode this frame (the scanout dma-buf fd). + let (msg, recv_fd) = match conn.recv_msg_timeout2(200).await { None => continue, // timeout: re-check stop at the loop top - Some(Ok(Some(d))) => d, - Some(Ok(None)) => break "desynchronized frame".to_owned(), + Some(Ok(pair)) => pair, Some(Err(err)) => break format!("recv: {err}"), }; match msg { + // Zero-copy split path: a dma-buf descriptor + (usually) the scanout fd. Import + EGL + // detile/convert to linear pixels HERE, then copy them latest-wins into the slot. That + // copy out of the context-owned convert buffer is the ONE remaining pixel copy in the + // whole pipeline (only the fd + this small descriptor crossed the socket). + Data::DrmFrameDmabuf(desc) => { + let conv = match converter.as_mut() { + Some(c) => c, + None => break "no DRM render node; cannot convert dma-buf frame".to_owned(), + }; + // The fd number valid in THIS process: the received fd when the producer attached + // one, or -1 for an import-once cache hit (libdrmtap reuses the EGLImage it holds for + // `fb_id`). `has_fd` set but no fd delivered is a protocol desync. + let received_fd: RawFd = if desc.has_fd { + match recv_fd.as_ref() { + Some(f) => f.as_raw_fd(), + None => { + break "dma-buf frame set has_fd but carried no SCM_RIGHTS fd".to_owned() + } + } + } else { + -1 + }; + // Rebuild the libdrmtap descriptor from the wire fields; `convert` overwrites its + // `dma_buf_fd` with `received_fd` (the exporter's local int is meaningless here). + let mut ddesc = drmtap_dmabuf_desc { + dma_buf_fd: -1, + width: desc.width, + height: desc.height, + format: desc.format, + modifier: desc.modifier, + fb_id: desc.fb_id, + num_planes: desc.num_planes, + offsets: desc.offsets, + pitches: desc.pitches, + hdr_eotf: desc.hdr_eotf, + hdr_max_nits: desc.hdr_max_nits, + }; + match conv.convert(&mut ddesc, received_fd) { + Ok((data, w, h, fmt)) => { + let mut slot = shared.slot.lock().unwrap(); + slot.latest = Some((w as usize, h as usize, fmt, data.to_vec())); + shared.cv.notify_one(); + } + // Transient convert contention: skip this frame (latest-wins keeps the newest), + // do not tear the stream down. + Err(err) if err.kind() == io::ErrorKind::WouldBlock => {} + Err(err) => break format!("convert: {err}"), + } + // `recv_fd` (the OwnedFd, if any) is dropped/closed at the end of this iteration, AFTER + // convert has imported it (the EGLImage import holds its own reference to the buffer). + } + // CPU-fallback path (old `.so` / no transferable dma-buf): the producer packed BGRA and + // sent it over the wire after the header. Store it as-is (BGRA); no convert needed. Data::DrmFrame { width, height } => match conn.next_raw().await { Ok(raw) => { let mut slot = shared.slot.lock().unwrap(); - slot.latest = Some((width as usize, height as usize, raw.to_vec())); + slot.latest = + Some((width as usize, height as usize, Pixfmt::BGRA, raw.to_vec())); shared.cv.notify_one(); } Err(err) => break format!("frame body: {err}"), @@ -272,10 +366,24 @@ async fn recv_thread( ), Err(err) => break format!("cursor body: {err}"), }, + // Live hotplug: the service pushed a fresh display list after a connector-topology change. + // Swap it into the sticky positive availability cache directly (no re-probe over `_drm`, so + // this never trips the wayland::clear() re-probe restart loop). A subsequent + // get_display_infos()/get_primary_index() then reports the fresh geometry. + Data::DrmDisplaysChanged(list) => { + if !list.is_empty() { + swap_available_displays(list); + } + } _ => {} // ignore any unexpected control message } }; log::info!("drm capture stream ended: {end_reason}"); + // Drop the render context on THIS thread (its EGL state + cached imports are thread-local; a + // cross-thread close would strand them — the 0.4.8 EGL-leak/OOM class). Explicit so it releases + // before the post-loop cleanup rather than at some later scope exit, and NEVER in + // `IpcDrmCapturer::Drop` (which runs on the encoder thread). + drop(converter); // Drop only THIS stream's cursor entry so a torn-down monitor does not erase the cursor state of // other still-active streams. remove_drm_cursor(display); @@ -372,9 +480,11 @@ fn query_displays() -> ResultType> { #[tokio::main(flavor = "current_thread")] async fn query_displays_async() -> ResultType> { let mut conn = connect_drm(1000).await?; - match conn.next_timeout(HANDSHAKE_TIMEOUT_MS).await? { - Some(Data::DrmDisplayList(v)) => Ok(v), - other => Err(anyhow!("expected DrmDisplayList, got {:?}", other)), + match conn.recv_msg_timeout2(HANDSHAKE_TIMEOUT_MS).await { + Some(Ok((Data::DrmDisplayList(v), _fd))) => Ok(v), + Some(Ok((other, _fd))) => Err(anyhow!("expected DrmDisplayList, got {:?}", other)), + Some(Err(err)) => Err(err), + None => Err(anyhow!("timed out waiting for DrmDisplayList")), } } @@ -567,9 +677,17 @@ fn match_wayland_display<'a>( /// Normalize a connector name for cross-source matching: DRM inserts a single-letter type /// discriminator that the compositor drops ("HDMI-A-1" -> "HDMI-1", "DVI-D-1" -> "DVI-1"); names /// like "DP-1" / "eDP-1" pass through unchanged. +/// +/// The middle component is only folded when it is a single *letter* (a type discriminator: the "A" +/// in HDMI-A, the "D" in DVI-D). A single *digit* middle component is NOT a discriminator but a +/// DisplayPort MST port index: "DP-1-2" is sink 2 downstream of DP connector 1 and is a DISTINCT +/// output from "DP-2". Folding it (the old `parts[1].len() == 1` guard did) aliased the MST sink onto +/// a real "DP-2", so primary selection and geometry augmentation attached the wrong logical position +/// and scale. The `is_ascii_alphabetic` predicate preserves "DP-1-2" verbatim while still folding the +/// letter discriminators. fn normalize_connector(name: &str) -> String { let parts: Vec<&str> = name.split('-').collect(); - if parts.len() == 3 && parts[1].len() == 1 { + if parts.len() == 3 && parts[1].len() == 1 && parts[1].chars().all(|c| c.is_ascii_alphabetic()) { format!("{}-{}", parts[0], parts[2]) } else { name.to_string() @@ -581,6 +699,21 @@ pub(super) fn clear() { *DRM_STATE.lock().unwrap() = ProbeState::Unknown; } +/// Swap the sticky positive availability cache to a freshly-enumerated display list, driven by a +/// service-pushed `DrmDisplaysChanged` hotplug signal on a live stream. This is the off-hot-path cache +/// refresh that keeps mid-session hotplug geometry fresh WITHOUT the blocking `_drm` re-probe that +/// `wayland::clear()` deliberately avoids (that re-probe blocks the async enumeration executor long +/// enough to trip "deadline has elapsed" and spiral into a restart loop). It only replaces an already +/// `Available` verdict — never flips `Unknown`/`Unavailable` to `Available` — so a stray signal cannot +/// force DRM on; establishing availability stays the job of the probe path. +fn swap_available_displays(list: Vec) { + let mut st = DRM_STATE.lock().unwrap(); + if matches!(&*st, ProbeState::Available(_)) { + log::info!("drm: hotplug refresh -> {} display(s)", list.len()); + *st = ProbeState::Available(list); + } +} + fn display_info_from_drm(d: &DrmDisplayInfo) -> DisplayInfo { let original_resolution = super::display_service::get_original_resolution(&d.name, d.width as usize, d.height as usize); diff --git a/src/server/wayland.rs b/src/server/wayland.rs index 7a5b16cd0..e4758271c 100644 --- a/src/server/wayland.rs +++ b/src/server/wayland.rs @@ -329,6 +329,19 @@ pub fn clear() { *PIPEWIRE_INITIALIZED.write().unwrap() = false; } +/// Initialize the PipeWire/portal capture path from the plain (sync) video thread, so a DRM display +/// that cannot be captured can fall through to PipeWire for THAT display. `ensure_inited` short-circuits +/// to the DRM branch whenever DRM is globally available, so it never runs `check_init`; this helper +/// drives the same async portal ScreenCast init directly (mirroring `ensure_inited`'s pattern). Needed +/// because `is_available()` is a GLOBAL verdict — it stays true for the still-working DRM outputs — so +/// without a per-display fallback a single failed/demoted DRM display would restart-loop the video +/// service instead of degrading to PipeWire only for itself. +#[cfg(feature = "drm")] +#[tokio::main(flavor = "current_thread")] +async fn ensure_pipewire_inited() -> ResultType<()> { + check_init().await +} + pub(super) fn get_capturer_for_display( display_idx: usize, ) -> ResultType { @@ -336,10 +349,24 @@ pub(super) fn get_capturer_for_display( bail!("Do not call this function if not wayland"); } // DRM/KMS capture path: build the capturer straight from the service `_drm` stream, bypassing - // the PipeWire CAP_DISPLAY_INFO machinery entirely. + // the PipeWire CAP_DISPLAY_INFO machinery entirely. `is_available()` is a GLOBAL verdict, so a + // per-display DRM failure (an ungrabbable/demoted CRTC, or — after the phase-2 split — a + // render-node-absent seat or a convert failure on the unprivileged side) must NOT propagate out + // and restart-loop this per-display video service. Instead fall THROUGH to PipeWire for just this + // display; the other DRM outputs keep streaming over DRM. #[cfg(feature = "drm")] if super::drm_capturer::is_available() { - return super::drm_capturer::get_capturer_info(display_idx); + match super::drm_capturer::get_capturer_info(display_idx) { + Ok(info) => return Ok(info), + Err(e) => { + log::warn!( + "drm capturer for display {} unavailable ({:#}); falling back to PipeWire", + display_idx, + e + ); + ensure_pipewire_inited()?; + } + } } let cap_map = CAP_DISPLAY_INFO.read().unwrap(); if let Some(addr) = cap_map.get(&display_idx) {