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) {