From 7a8be5a08f73cd5914261157e248618d523b09fa Mon Sep 17 00:00:00 2001 From: Danil Silantyev Date: Thu, 8 Oct 2026 03:55:28 +0500 Subject: [PATCH 1/6] perf(desktop): recycle X11 capture buffers --- crates/rds-desktop/src/capture/x11.rs | 50 ++++++++++++++++++++++++++- crates/rds-desktop/src/session.rs | 5 +++ 2 files changed, 54 insertions(+), 1 deletion(-) diff --git a/crates/rds-desktop/src/capture/x11.rs b/crates/rds-desktop/src/capture/x11.rs index 7e722589..2bf87ebf 100644 --- a/crates/rds-desktop/src/capture/x11.rs +++ b/crates/rds-desktop/src/capture/x11.rs @@ -54,6 +54,9 @@ pub struct X11Capturer { /// Sticky flag: the first `changed` call must be true (the screen /// existed before the damage object did). dirty: bool, + /// One owned frame buffer returned by the producer after encoding. The + /// next capture fills it in place, avoiding allocator churn at 60 fps. + recycled: Option, } impl X11Capturer { @@ -122,6 +125,7 @@ impl X11Capturer { shm, damage, dirty: true, + recycled: None, }) } @@ -154,6 +158,15 @@ impl X11Capturer { }) } + /// Return an encoded frame's owned storage for reuse on the next + /// capture. A scaled frame has a different size and is ignored. + pub(crate) fn recycle(&mut self, data: Bytes) { + let expected = usize::from(self.width) * usize::from(self.height) * 4; + if self.shm.is_some() && data.len() == expected { + self.recycled = Some(data); + } + } + /// DAMAGE (XFixes ≥ 4): one object on the root window reporting /// `NON_EMPTY` transitions — enough for a changed/not-changed bit. /// `None` where the extension is absent. @@ -196,7 +209,7 @@ impl Capturer for X11Capturer { width: u32::from(self.width), height: u32::from(self.height), stride: u32::from(self.width) * 4, - data: Bytes::copy_from_slice(&shm.map[..shm.len]), + data: copy_capture(&mut self.recycled, &shm.map[..shm.len]), }); } Err(e) => { @@ -268,6 +281,20 @@ impl Capturer for X11Capturer { } } +/// The X11 segment is overwritten by the next request, so capture must +/// copy into owned storage. Reuse only uniquely owned, correctly sized data; +/// a retained snapshot must remain immutable. +fn copy_capture(recycled: &mut Option, source: &[u8]) -> Bytes { + if let Some(bytes) = recycled.take() + && let Ok(mut buffer) = bytes.try_into_mut() + && buffer.len() == source.len() + { + buffer.copy_from_slice(source); + return buffer.freeze(); + } + Bytes::copy_from_slice(source) +} + impl Drop for X11Capturer { fn drop(&mut self) { if let Some(dmg) = self.damage.take() { @@ -294,6 +321,27 @@ pub fn capabilities() -> Result { mod tests { use super::*; + #[test] + fn capture_storage_reuses_unique_buffers_and_preserves_retained_snapshots() { + let first = Bytes::from(vec![1u8; 1024]); + let pointer = first.as_ptr(); + let mut recycled = Some(first); + let next = copy_capture(&mut recycled, &[2u8; 1024]); + assert_eq!(next.as_ptr(), pointer, "unique buffer must be reused"); + assert!(recycled.is_none()); + assert_eq!(next.as_ref(), &[2u8; 1024]); + + let retained = next.clone(); + recycled = Some(next); + let changed = copy_capture(&mut recycled, &[3u8; 1024]); + assert_ne!(changed.as_ptr(), retained.as_ptr()); + assert_eq!(retained.as_ref(), &[2u8; 1024]); + assert_eq!(changed.as_ref(), &[3u8; 1024]); + + recycled = Some(changed); + assert_eq!(copy_capture(&mut recycled, &[4u8; 16]).as_ref(), &[4u8; 16]); + } + /// Explicit native fixture: never silently succeeds without capture. #[test] #[ignore = "requires a dedicated Xvfb server; repaints its root"] diff --git a/crates/rds-desktop/src/session.rs b/crates/rds-desktop/src/session.rs index a8a2bb33..1405902f 100644 --- a/crates/rds-desktop/src/session.rs +++ b/crates/rds-desktop/src/session.rs @@ -1771,6 +1771,11 @@ mod x11 { ); let (width, height) = (raw.width, raw.height); let encoded = self.encoder.encode(&raw); + // The encoder has finished borrowing the capture. Return a + // native-sized frame buffer to MIT-SHM capture so the next + // frame reuses its allocation. Downscaled frames are naturally + // rejected by the capturer's size check. + self.capturer.recycle(raw.data); self.skipped = encoded.as_ref().is_ok_and(|frame| frame.data.is_empty()); self.last_work = work_started.elapsed(); match encoded { From 5fdf3757d111f0b75669e1882b2e3e3be14642d4 Mon Sep 17 00:00:00 2001 From: Danil Silantyev Date: Thu, 8 Oct 2026 04:12:24 +0500 Subject: [PATCH 2/6] perf(desktop): borrow completed X11 frames through encoding --- crates/rds-desktop/src/capture/x11.rs | 155 +++++++++++++---------- crates/rds-desktop/src/codec/openh264.rs | 60 +++++++-- crates/rds-desktop/src/lib.rs | 23 ++++ crates/rds-desktop/src/scaling.rs | 9 +- crates/rds-desktop/src/session.rs | 38 +++--- 5 files changed, 186 insertions(+), 99 deletions(-) diff --git a/crates/rds-desktop/src/capture/x11.rs b/crates/rds-desktop/src/capture/x11.rs index 2bf87ebf..33ebe6c6 100644 --- a/crates/rds-desktop/src/capture/x11.rs +++ b/crates/rds-desktop/src/capture/x11.rs @@ -4,7 +4,7 @@ //! //! MIT-SHM removes the dominant capture cost: the pixmap is written //! into a shared segment the client mmaps, so a 1080p frame costs one -//! completion event plus a memcpy instead of ~8 MiB serialized through +//! reply and a synchronous borrowed view instead of ~8 MiB serialized through //! the X socket. `CreateSegment` (SHM ≥ 1.2 fd passing) is preferred — //! the server allocates and owns nothing persists on the client side. //! `ext-image-copy-capture` and the DRM/KMS tap remain scheduled behind @@ -15,6 +15,7 @@ //! module (workspace convention: FFI paths allow `unsafe_code` locally). #![allow(unsafe_code)] +use std::borrow::Cow; use std::os::fd::OwnedFd; use bytes::Bytes; @@ -27,7 +28,7 @@ use x11rb::protocol::shm::{self, ConnectionExt as _, Seg}; use x11rb::protocol::xproto::{ConnectionExt, GetImageReply, ImageFormat}; use x11rb::rust_connection::RustConnection; -use crate::{Capturer, DesktopError, RawFrame}; +use crate::{BgraFrame, Capturer, DesktopError, RawFrame}; /// MIT-SHM segment the server fills in place — the reply is a /// completion event, not a serialized pixmap. @@ -54,9 +55,6 @@ pub struct X11Capturer { /// Sticky flag: the first `changed` call must be true (the screen /// existed before the damage object did). dirty: bool, - /// One owned frame buffer returned by the producer after encoding. The - /// next capture fills it in place, avoiding allocator churn at 60 fps. - recycled: Option, } impl X11Capturer { @@ -125,7 +123,6 @@ impl X11Capturer { shm, damage, dirty: true, - recycled: None, }) } @@ -158,15 +155,6 @@ impl X11Capturer { }) } - /// Return an encoded frame's owned storage for reuse on the next - /// capture. A scaled frame has a different size and is ignored. - pub(crate) fn recycle(&mut self, data: Bytes) { - let expected = usize::from(self.width) * usize::from(self.height) * 4; - if self.shm.is_some() && data.len() == expected { - self.recycled = Some(data); - } - } - /// DAMAGE (XFixes ≥ 4): one object on the root window reporting /// `NON_EMPTY` transitions — enough for a changed/not-changed bit. /// `None` where the extension is absent. @@ -183,11 +171,29 @@ impl X11Capturer { } } -impl Capturer for X11Capturer { - fn capture(&mut self) -> Result { +impl X11Capturer { + /// The synchronous callback is the lifetime boundary: the next capture + /// cannot overwrite the mapping until conversion/encode returns. Owned + /// capture consumers retain the separate snapshot-producing trait method. + pub(crate) fn capture_with( + &mut self, + process: impl FnOnce(BgraFrame<'_>) -> T, + ) -> Result { + let (width, height) = (u32::from(self.width), u32::from(self.height)); + let pixels = self.capture_pixels()?; + Ok(process(BgraFrame { + width, + height, + stride: width * 4, + data: pixels.as_ref(), + })) + } + + fn capture_pixels(&mut self) -> Result, DesktopError> { if let Some(shm) = &self.shm { - // `reply()` waits for the ShmCompletion event the server - // sends after filling the segment. + // The GetImage reply arrives after the server has finished + // writing the segment. No new request can reuse it while the + // callback holds the capturer's exclusive borrow. let filled = self .conn .shm_get_image( @@ -205,12 +211,11 @@ impl Capturer for X11Capturer { .and_then(|c| c.reply().map_err(|e| DesktopError::Capture(e.to_string()))); match filled { Ok(_) => { - return Ok(RawFrame { - width: u32::from(self.width), - height: u32::from(self.height), - stride: u32::from(self.width) * 4, - data: copy_capture(&mut self.recycled, &shm.map[..shm.len]), - }); + let shm = self + .shm + .as_ref() + .expect("successful capture retains its segment"); + return Ok(Cow::Borrowed(&shm.map[..shm.len])); } Err(e) => { tracing::warn!("MIT-SHM capture failed ({e}); falling back to GetImage"); @@ -243,11 +248,19 @@ impl Capturer for X11Capturer { self.height ))); } + Ok(Cow::Owned(reply.data)) + } +} + +impl Capturer for X11Capturer { + fn capture(&mut self) -> Result { + let (width, height) = (u32::from(self.width), u32::from(self.height)); + let pixels = self.capture_pixels()?; Ok(RawFrame { - width: u32::from(self.width), - height: u32::from(self.height), - stride: u32::from(self.width) * 4, - data: Bytes::from(reply.data), + width, + height, + stride: width * 4, + data: Bytes::from(pixels.into_owned()), }) } @@ -281,20 +294,6 @@ impl Capturer for X11Capturer { } } -/// The X11 segment is overwritten by the next request, so capture must -/// copy into owned storage. Reuse only uniquely owned, correctly sized data; -/// a retained snapshot must remain immutable. -fn copy_capture(recycled: &mut Option, source: &[u8]) -> Bytes { - if let Some(bytes) = recycled.take() - && let Ok(mut buffer) = bytes.try_into_mut() - && buffer.len() == source.len() - { - buffer.copy_from_slice(source); - return buffer.freeze(); - } - Bytes::copy_from_slice(source) -} - impl Drop for X11Capturer { fn drop(&mut self) { if let Some(dmg) = self.damage.take() { @@ -321,27 +320,6 @@ pub fn capabilities() -> Result { mod tests { use super::*; - #[test] - fn capture_storage_reuses_unique_buffers_and_preserves_retained_snapshots() { - let first = Bytes::from(vec![1u8; 1024]); - let pointer = first.as_ptr(); - let mut recycled = Some(first); - let next = copy_capture(&mut recycled, &[2u8; 1024]); - assert_eq!(next.as_ptr(), pointer, "unique buffer must be reused"); - assert!(recycled.is_none()); - assert_eq!(next.as_ref(), &[2u8; 1024]); - - let retained = next.clone(); - recycled = Some(next); - let changed = copy_capture(&mut recycled, &[3u8; 1024]); - assert_ne!(changed.as_ptr(), retained.as_ptr()); - assert_eq!(retained.as_ref(), &[2u8; 1024]); - assert_eq!(changed.as_ref(), &[3u8; 1024]); - - recycled = Some(changed); - assert_eq!(copy_capture(&mut recycled, &[4u8; 16]).as_ref(), &[4u8; 16]); - } - /// Explicit native fixture: never silently succeeds without capture. #[test] #[ignore = "requires a dedicated Xvfb server; repaints its root"] @@ -395,6 +373,55 @@ mod tests { &[0x56, 0x34, 0x12], "captured pixel did not match the native paint" ); + // The producer borrows the completed mapping directly; the owned + // snapshot above must survive another capture and a different paint. + let mapped = cap.shm.as_ref().unwrap().map.as_ptr(); + cap.conn + .change_gc( + gc, + &x11rb::protocol::xproto::ChangeGCAux::new().foreground(0xAB_CD_EF), + ) + .unwrap() + .check() + .unwrap(); + cap.conn + .poly_fill_rectangle( + cap.root, + gc, + &[Rectangle { + x: 0, + y: 0, + width: 100, + height: 100, + }], + ) + .unwrap() + .check() + .unwrap(); + cap.capture_with(|borrowed| { + assert_eq!( + borrowed.data.as_ptr(), + mapped, + "borrowed capture copied pixels" + ); + assert_eq!(&borrowed.data[pixel..pixel + 3], &[0xEF, 0xCD, 0xAB]); + }) + .unwrap(); + assert_eq!(&frame.data[pixel..pixel + 3], &[0x56, 0x34, 0x12]); + + // A server without usable SHM still provides the same borrowed input + // contract from its owned GetImage reply; no extra staging copy. + let shm = cap.shm.take().unwrap(); + cap.conn.shm_detach(shm.seg).unwrap().check().unwrap(); + drop(shm); + cap.capture_with(|borrowed| { + assert_eq!(&borrowed.data[pixel..pixel + 3], &[0xEF, 0xCD, 0xAB]); + }) + .unwrap(); + assert_eq!( + &cap.capture().unwrap().data[pixel..pixel + 3], + &[0xEF, 0xCD, 0xAB] + ); cap.conn.free_gc(gc).unwrap().check().unwrap(); } } diff --git a/crates/rds-desktop/src/codec/openh264.rs b/crates/rds-desktop/src/codec/openh264.rs index d4056e35..13e8ee11 100644 --- a/crates/rds-desktop/src/codec/openh264.rs +++ b/crates/rds-desktop/src/codec/openh264.rs @@ -13,7 +13,7 @@ use openh264::formats::{BGRA8Source, RGB8Source, RGBSource, YUVBuffer, YUVSource use openh264::{Error as OhError, OpenH264API}; use rds_core::Codec; -use crate::{Decoder, DesktopError, EncodedFrame, Encoder, RawFrame}; +use crate::{BgraFrame, Decoder, DesktopError, EncodedFrame, Encoder, RawFrame}; // A large recovery picture can leave timestamp rate control in debt for // seconds at its low-rate floor. Preserve occasional delta progress while @@ -84,7 +84,7 @@ impl H264Encoder { impl Encoder for H264Encoder { fn encode(&mut self, frame: &RawFrame) -> Result { - self.encode_timed(frame, self.started.elapsed().as_millis() as u64) + self.encode_bgra(frame.into()) } fn request_idr(&mut self) { self.want_idr = true; @@ -99,10 +99,26 @@ impl Encoder for H264Encoder { } impl H264Encoder { + #[cfg(test)] fn encode_timed( &mut self, frame: &RawFrame, now_ms: u64, + ) -> Result { + self.encode_bgra_timed(frame.into(), now_ms) + } + + pub(crate) fn encode_bgra( + &mut self, + frame: BgraFrame<'_>, + ) -> Result { + self.encode_bgra_timed(frame, self.started.elapsed().as_millis() as u64) + } + + fn encode_bgra_timed( + &mut self, + frame: BgraFrame<'_>, + now_ms: u64, ) -> Result { let dimensions = (frame.width, frame.height); let target = self @@ -340,14 +356,14 @@ impl Decoder for H264Decoder { pub fn bgra_to_i420(frame: &RawFrame) -> YUVBuffer { let stride = frame.stride as usize; if stride.is_multiple_of(4) && frame.data.len() >= stride * frame.height as usize { - return YUVBuffer::from_bgra8_source(StridedBgra(frame)); + return YUVBuffer::from_bgra8_source(StridedBgra(frame.into())); } - bgra_to_i420_scalar(frame) + bgra_to_i420_scalar(frame.into()) } /// `RawFrame` as an `openh264` BGRA source, carrying its real stride so /// padded capture buffers need no intermediate copy. -struct StridedBgra<'a>(&'a RawFrame); +struct StridedBgra<'a>(BgraFrame<'a>); impl RGBSource for StridedBgra<'_> { fn dimensions(&self) -> (usize, usize) { @@ -367,7 +383,7 @@ impl RGB8Source for StridedBgra<'_> { } fn rgb8_data(&self) -> &[u8] { - &self.0.data + self.0.data } fn pixel_stride(&self) -> usize { @@ -383,7 +399,7 @@ impl BGRA8Source for StridedBgra<'_> {} /// Scalar fallback for strides that are not whole pixels — also the /// test reference for the SIMD path's output. -fn bgra_to_i420_scalar(frame: &RawFrame) -> YUVBuffer { +fn bgra_to_i420_scalar(frame: BgraFrame<'_>) -> YUVBuffer { let (w, h) = (frame.width as usize, frame.height as usize); let stride = frame.stride as usize; let mut yuv = vec![0u8; w * h * 3 / 2]; @@ -424,6 +440,34 @@ mod tests { } } + #[test] + fn borrowed_input_preserves_owned_bitstream_and_recovery() { + let mut owned = H264Encoder::new(1_000_000, 30.0).unwrap(); + let mut borrowed = H264Encoder::new(1_000_000, 30.0).unwrap(); + let mut decoder = H264Decoder::new().unwrap(); + for sequence in 0..8 { + let mut raw = frame(); + raw.data = Bytes::from(vec![sequence as u8 * 20; raw.data.len()]); + if sequence == 4 { + owned.request_idr(); + borrowed.request_idr(); + } + let a = owned.encode_timed(&raw, sequence * 34).unwrap(); + let b = borrowed + .encode_bgra_timed((&raw).into(), sequence * 34) + .unwrap(); + assert_eq!( + a.data, b.data, + "borrowed frame changed bitstream at {sequence}" + ); + assert_eq!(a.keyframe, b.keyframe); + if sequence == 0 || sequence == 4 { + assert!(a.keyframe, "explicit recovery did not emit IDR"); + } + assert!(decoder.decode(&b).unwrap().is_some()); + } + } + /// The keyframe flag must describe the emitted bitstream: IDR on /// frame 0 (fresh encoder) and after `request_idr`, deltas between. #[test] @@ -633,7 +677,7 @@ mod tests { data: Bytes::from(data), }; let fast = bgra_to_i420(&raw); - let slow = bgra_to_i420_scalar(&raw); + let slow = bgra_to_i420_scalar((&raw).into()); assert_eq!(fast.dimensions(), slow.dimensions()); for (a, b) in fast.y().iter().zip(slow.y()) { assert!((i32::from(*a) - i32::from(*b)).abs() <= 2, "Y diverges"); diff --git a/crates/rds-desktop/src/lib.rs b/crates/rds-desktop/src/lib.rs index c2151d5c..d09d5232 100644 --- a/crates/rds-desktop/src/lib.rs +++ b/crates/rds-desktop/src/lib.rs @@ -58,6 +58,29 @@ pub struct RawFrame { pub data: Bytes, } +/// Borrowed BGRA input for synchronous capture/conversion/encode work. A +/// native mapping can supply pixels without first allocating an owned frame. +#[cfg(any(feature = "x11", test))] +#[derive(Clone, Copy)] +pub(crate) struct BgraFrame<'a> { + pub width: u32, + pub height: u32, + pub stride: u32, + pub data: &'a [u8], +} + +#[cfg(any(feature = "x11", test))] +impl<'a> From<&'a RawFrame> for BgraFrame<'a> { + fn from(frame: &'a RawFrame) -> Self { + Self { + width: frame.width, + height: frame.height, + stride: frame.stride, + data: &frame.data, + } + } +} + /// Local software receive limit: up to 8K pixels, no side above 8192. /// Check before allocating BGRA output; native codec internal memory is separate. pub(crate) fn frame_bytes(width: usize, height: usize) -> Option { diff --git a/crates/rds-desktop/src/scaling.rs b/crates/rds-desktop/src/scaling.rs index 5922c232..47e724df 100644 --- a/crates/rds-desktop/src/scaling.rs +++ b/crates/rds-desktop/src/scaling.rs @@ -1,12 +1,9 @@ //! Owned low-latency BGRA downscaling for a configured serving profile. The //! display capabilities keep original coordinates; only video dimensions change. -use crate::{DesktopError, RawFrame}; +use crate::{BgraFrame, DesktopError, RawFrame}; use bytes::Bytes; -pub(crate) fn downscale(frame: RawFrame, height: Option) -> Result { - let Some(height) = height.filter(|height| *height < frame.height) else { - return Ok(frame); - }; +pub(crate) fn downscale(frame: BgraFrame<'_>, height: u32) -> Result { let height = height & !1; let width = ((u64::from(frame.width) * u64::from(height) / u64::from(frame.height)) as u32) & !1; @@ -57,7 +54,7 @@ mod tests { stride: 72, data: Bytes::from(bytes), }; - let resized = downscale(frame, Some(4)).unwrap(); + let resized = downscale((&frame).into(), 4).unwrap(); assert_eq!((resized.width, resized.height, resized.stride), (8, 4, 32)); assert_eq!(&resized.data[..], &vec![12; 128]); } diff --git a/crates/rds-desktop/src/session.rs b/crates/rds-desktop/src/session.rs index 1405902f..23896981 100644 --- a/crates/rds-desktop/src/session.rs +++ b/crates/rds-desktop/src/session.rs @@ -1746,20 +1746,6 @@ mod x11 { let capture_ts_ms = clock.now_ms(); self.skipped = false; let work_started = Instant::now(); - let raw = match self.capturer.capture() { - Ok(f) => f, - Err(e) => { - tracing::warn!("capture failed: {e}"); - return None; - } - }; - let raw = match crate::scaling::downscale(raw, self.output_height) { - Ok(raw) => raw, - Err(error) => { - tracing::warn!(%error,"capture scaling failed"); - return None; - } - }; if controls.idr.swap(false, Ordering::Relaxed) { self.encoder.request_idr(); } @@ -1769,13 +1755,23 @@ mod x11 { .load(Ordering::Relaxed) .min(u64::from(u32::MAX)) as u32, ); - let (width, height) = (raw.width, raw.height); - let encoded = self.encoder.encode(&raw); - // The encoder has finished borrowing the capture. Return a - // native-sized frame buffer to MIT-SHM capture so the next - // frame reuses its allocation. Downscaled frames are naturally - // rejected by the capturer's size check. - self.capturer.recycle(raw.data); + let encoder = &mut self.encoder; + let output_height = self.output_height; + let captured = self.capturer.capture_with(|raw| { + let scaled = output_height + .filter(|height| *height < raw.height) + .map(|height| crate::scaling::downscale(raw, height)) + .transpose()?; + let raw = scaled.as_ref().map(crate::BgraFrame::from).unwrap_or(raw); + Ok::<_, DesktopError>((raw.width, raw.height, encoder.encode_bgra(raw))) + }); + let (width, height, encoded) = match captured { + Ok(Ok(frame)) => frame, + Ok(Err(error)) | Err(error) => { + tracing::warn!(%error, "capture or scaling failed"); + return None; + } + }; self.skipped = encoded.as_ref().is_ok_and(|frame| frame.data.is_empty()); self.last_work = work_started.elapsed(); match encoded { From 3ccc5563cc89fc0231897e31c75900697b2126be Mon Sep 17 00:00:00 2001 From: Danil Silantyev Date: Thu, 8 Oct 2026 04:12:24 +0500 Subject: [PATCH 3/6] docs(desktop): record borrowed capture CPU measurements --- docs/architecture.md | 11 ++- docs/native-viewer.md | 5 ++ .../rds-capture-conversion-20261008.rs | 82 +++++++++++++++++++ .../rds-desktop-cpu-20261008-data.json | 81 ++++++++++++++++++ docs/reports/rds-desktop-cpu-20261008.md | 60 ++++++++++++++ docs/roadmap.md | 5 ++ 6 files changed, 241 insertions(+), 3 deletions(-) create mode 100644 docs/reports/rds-capture-conversion-20261008.rs create mode 100644 docs/reports/rds-desktop-cpu-20261008-data.json create mode 100644 docs/reports/rds-desktop-cpu-20261008.md diff --git a/docs/architecture.md b/docs/architecture.md index 00070bcf..eba41ac3 100644 --- a/docs/architecture.md +++ b/docs/architecture.md @@ -524,6 +524,11 @@ opens a stream and never writes its tag cannot stall the routing of the streams queued behind it, and a claimed inbox whose consumer dropped is reclaimable by the next `uni_streams` call. +The X11 producer synchronously borrows a completed MIT-SHM mapping through +conversion/encode, avoiding an owned native-resolution BGRA staging copy. +Owned capture consumers still receive independent snapshots; the next X11 +capture cannot reuse the segment until the synchronous callback returns. + Desktop media: capture → BGRA→I420 → H.264 (OpenH264 baseline, no B-frames; hw encoders behind a trait) → per-frame uni stream with a `FrameHeader` `{seq, keyframe, capture_ts_ms, send_ts_ms}`. Freshness is enforced at @@ -536,9 +541,9 @@ is still sending* is reset mid-write (MoQ-style) — a stale delta's tail only consumes path capacity the fresher frame needs. The reset breaks the client's delta chain, so the producer's IDR flag is re-armed and queued deltas are drained. The encoder's `keyframe` flag is read off the -emitted NAL units (forced IDRs, periodic IDRs at the configured -~8s `intra_frame_period`, and encoder rebuilds on a >15% bitrate change -all mark real IDRs) — never assumed from a schedule. The receiver drops +emitted NAL units — never assumed from a schedule. Periodic IDRs are disabled; +new sessions, geometry changes and explicit reference recovery request an IDR. +Native bitrate updates preserve references instead of rebuilding the encoder. The receiver drops anything below a "next expected seq" watermark, decodes with a session-local decoder (a shared one would cross-contaminate reference chains). A bounded completion-order queue preserves an unfinished reference diff --git a/docs/native-viewer.md b/docs/native-viewer.md index bfaa5425..4d7485c6 100644 --- a/docs/native-viewer.md +++ b/docs/native-viewer.md @@ -322,6 +322,11 @@ damage readiness wakes via the X socket rather than a fixed 25 ms poll delay. An idle wake starts capture immediately without counting skipped idle slots or sub-slot scheduling jitter as encoder starvation. This prevents an idle desktop from reducing its bitrate solely because it resumed after waiting. +The X11 software producer borrows the completed MIT-SHM frame during synchronous +conversion/encode, avoiding the native-size BGRA allocation and staging copy. +Downscaling reads this view directly and allocates only the scaled output; +owned capture consumers retain independent snapshots. See the +[CPU measurement and qualification limits](reports/rds-desktop-cpu-20261008.md). The software producer treats `max_fps` as a ceiling and accounts for measured capture/conversion/codec work when scheduling its next frame. Fixed capture CPU cost cannot reduce network bitrate simply by exceeding a 60 FPS interval. diff --git a/docs/reports/rds-capture-conversion-20261008.rs b/docs/reports/rds-capture-conversion-20261008.rs new file mode 100644 index 00000000..b5860618 --- /dev/null +++ b/docs/reports/rds-capture-conversion-20261008.rs @@ -0,0 +1,82 @@ +// Frozen paired benchmark input for rds-desktop-cpu-20261008.md. +use bytes::Bytes; +use openh264::formats::{BGRA8Source, RGB8Source, RGBSource, YUVBuffer, YUVSource}; +use std::{hint::black_box, time::Instant}; +#[derive(Clone, Copy)] +pub(crate) struct BgraFrame<'a> { + pub width: u32, + pub height: u32, + pub stride: u32, + pub data: &'a [u8], +} +struct StridedBgra<'a>(BgraFrame<'a>); + +impl RGBSource for StridedBgra<'_> { + fn dimensions(&self) -> (usize, usize) { + (self.0.width as usize, self.0.height as usize) + } + + fn pixel_f32(&self, x: usize, y: usize) -> (f32, f32, f32) { + let o = y * self.0.stride as usize + x * 4; + let px = &self.0.data[o..o + 4]; + (px[2] as f32, px[1] as f32, px[0] as f32) + } +} + +impl RGB8Source for StridedBgra<'_> { + fn dimensions_padded(&self) -> (usize, usize) { + (self.0.stride as usize / 4, self.0.height as usize) + } + + fn rgb8_data(&self) -> &[u8] { + self.0.data + } + + fn pixel_stride(&self) -> usize { + 4 + } + + fn rgb_channel_offsets(&self) -> (usize, usize, usize) { + (2, 1, 0) + } +} + +impl BGRA8Source for StridedBgra<'_> {} + +fn main() { + let source: Vec = (0..1920 * 1080 * 4).map(|i| (i % 251) as u8).collect(); + let rounds = 600usize; + let raw = BgraFrame { + width: 1920, + height: 1080, + stride: 7680, + data: &source, + }; + let mut yuv = YUVBuffer::from_bgra8_source(StridedBgra(raw)); + for pair in 0..5 { + for copied in if pair % 2 == 0 { + [true, false] + } else { + [false, true] + } { + let started = Instant::now(); + for _ in 0..rounds { + if copied { + let owned = Bytes::copy_from_slice(black_box(&source)); + let input = BgraFrame { + data: &owned, + ..raw + }; + yuv.read_bgra8(StridedBgra(input)); + } else { + yuv.read_bgra8(StridedBgra(black_box(raw))); + } + black_box(yuv.y()); + } + println!( + "{{\"pair\":{pair},\"copied\":{copied},\"rounds\":{rounds},\"elapsed_ns\":{}}}", + started.elapsed().as_nanos() + ); + } + } +} diff --git a/docs/reports/rds-desktop-cpu-20261008-data.json b/docs/reports/rds-desktop-cpu-20261008-data.json new file mode 100644 index 00000000..29b017ea --- /dev/null +++ b/docs/reports/rds-desktop-cpu-20261008-data.json @@ -0,0 +1,81 @@ +{ + "schema_version": 1, + "scope": "isolated full-HD capture copy plus BGRA-to-I420 conversion; excludes X11 request, encoder, transport, decoder and viewer", + "environment": { + "os": "Linux x86_64", + "rust": "1.98.1", + "bytes": "1.12.1", + "openh264": "0.9.8", + "profile": "rustc -C opt-level=3; release dependency artifacts" + }, + "width": 1920, + "height": 1080, + "stride": 7680, + "source_pattern": "byte index modulo 251", + "alternating_order": true, + "samples": [ + { + "pair": 0, + "copied": true, + "rounds": 600, + "elapsed_ns": 2454133391 + }, + { + "pair": 0, + "copied": false, + "rounds": 600, + "elapsed_ns": 1108534954 + }, + { + "pair": 1, + "copied": false, + "rounds": 600, + "elapsed_ns": 1049389560 + }, + { + "pair": 1, + "copied": true, + "rounds": 600, + "elapsed_ns": 1989168959 + }, + { + "pair": 2, + "copied": true, + "rounds": 600, + "elapsed_ns": 2055466733 + }, + { + "pair": 2, + "copied": false, + "rounds": 600, + "elapsed_ns": 1080034741 + }, + { + "pair": 3, + "copied": false, + "rounds": 600, + "elapsed_ns": 1126424350 + }, + { + "pair": 3, + "copied": true, + "rounds": 600, + "elapsed_ns": 2092071067 + }, + { + "pair": 4, + "copied": true, + "rounds": 600, + "elapsed_ns": 2181969117 + }, + { + "pair": 4, + "copied": false, + "rounds": 600, + "elapsed_ns": 1134398298 + } + ], + "copied_median_ms": 3.486785111666667, + "borrowed_median_ms": 1.8475582566666666, + "stage_reduction_percent": 47.01255748497955 +} diff --git a/docs/reports/rds-desktop-cpu-20261008.md b/docs/reports/rds-desktop-cpu-20261008.md new file mode 100644 index 00000000..e904ce3f --- /dev/null +++ b/docs/reports/rds-desktop-cpu-20261008.md @@ -0,0 +1,60 @@ +# Desktop CPU work — 2026-10-08 + +W6.1 software-capture follow-up. The X11 producer now converts/encodes a +synchronous borrowed view of the completed MIT-SHM mapping. Native-resolution +capture no longer allocates and copies an 8,294,400-byte BGRA staging frame. +Downscaling reads the same view and allocates only its smaller output. The +public owned `Capturer::capture` and `Encoder::encode` contracts remain intact; +plain GetImage supplies the view from its owned reply without another copy. +No new dependency, unsafe block, codec setting, FPS cap or wire change is added. + +The exclusive capturer borrow spans the entire callback. No subsequent +ShmGetImage request can overwrite its segment before synchronous conversion and +encode return; encoded bytes retain their separate ownership. Native tests +check mapping-address identity, different paints, retained owned snapshots and +the plain GetImage fallback. Codec tests compare owned and borrowed input byte +for byte across explicit IDR recovery; rate-control and reference tests remain. + +## Measurement + +The [frozen Rust input](rds-capture-conversion-20261008.rs) runs five alternating +pairs of 600 Full HD conversions, reusing the I420 destination in both modes. +The copied mode calls the prior `Bytes::copy_from_slice` immediately before the +same stride-aware converter; borrowed mode reads the source directly. The +[raw data](rds-desktop-cpu-20261008-data.json) retains every pair. + +| Isolated stage | Median per frame | +| --- | ---: | +| Owned capture copy + conversion | 3.487 ms | +| Borrowed capture + conversion | 1.848 ms | + +This stage takes 47.0% less elapsed time in this shared Linux VM measurement. It is not a +47% reduction in whole-agent CPU, end-to-end latency or physical display time. +The synthetic source is fixed, the run measures elapsed time, and concurrent +host work is not controlled. Live software encode remains the dominant cost. + +Reproduce with Rust 1.98.1 and this checkout's locked release dependency +artifacts (substitute the actual Cargo artifact hashes): + +```sh +rustc --edition=2024 -C opt-level=3 docs/reports/rds-capture-conversion-20261008.rs \ + --extern bytes=target/release/deps/libbytes-.rlib \ + --extern openh264=target/release/deps/libopenh264-.rlib \ + -L dependency=target/release/deps -o /tmp/rds-capture-conversion +/tmp/rds-capture-conversion +``` + +An allocation-recycling-only candidate did not improve the paired copy run and +is absent from the final implementation. Disabling scene-change detection is +also ineffective: OpenH264's screen-content `ParamValidation` forces it back on. +[Upstream configuration validation](https://github.com/cisco/openh264/blob/master/codec/encoder/core/src/encoder_ext.cpp) +and the exact openh264-sys2 0.9.8 source confirm that behavior. Medium codec +complexity remains; lowering it would be a separate quality/bitrate decision. + +## Qualification boundary + +The macOS x11 feature lane checks the codec and scaling code; native X11 work +must pass on Linux under a dedicated Xvfb display. Required PR checks cover +both supported OSes and Linux native fixtures. Installed runtime receipts and +private identities belong to the estate. This increment does not close the +broader hardware-codec, geometry, quality or glass-to-glass gates. diff --git a/docs/roadmap.md b/docs/roadmap.md index 65cabc6f..7fa7d204 100644 --- a/docs/roadmap.md +++ b/docs/roadmap.md @@ -12,6 +12,11 @@ The X11 serving increment removes fixed idle polling and supports explicit software downscaling. This advances W6.1–W6.4/W6.6/W6.7; it does not close the v0.3 hardware-codec, geometry, quality or glass-to-glass acceptance gates. +The [borrowed X11 capture increment](reports/rds-desktop-cpu-20261008.md) +removes the native-resolution staging copy before synchronous software encode. +Its isolated conversion measurement advances W6.1 CPU work; it retains the +existing codec quality, FPS and recovery contracts and does not close those gates. + The explicit [validated-payload receipt increment](reports/rds-payload-receipts-20261005.md) advances W6.8/WS5 delivery: bounded sender admission follows an exact complete-read proof instead of a potentially delayed FIN acknowledgement. Legacy behavior and From 87a2be493c78c9ae6be8d7b807dbb2227de7beef Mon Sep 17 00:00:00 2001 From: Danil Silantyev Date: Thu, 8 Oct 2026 04:24:23 +0500 Subject: [PATCH 4/6] fix(desktop): confirm completed superseded payload receipts --- crates/rds-desktop/src/session.rs | 96 ++++++++++++++++++++++++++++++- 1 file changed, 93 insertions(+), 3 deletions(-) diff --git a/crates/rds-desktop/src/session.rs b/crates/rds-desktop/src/session.rs index 23896981..2c700ebd 100644 --- a/crates/rds-desktop/src/session.rs +++ b/crates/rds-desktop/src/session.rs @@ -2025,9 +2025,10 @@ async fn send_frame_inner( return SendOutcome::Failed; } }; - if matches!(outcome, SendOutcome::Sent | SendOutcome::Done) - && let (Some(receipts), Some(hasher)) = (&payload_receipts, payload_hasher) - { + // Superseded also means the current payload was fully written: its + // successor waits behind the current reference. Only Abandoned returned + // early above. Install every completed payload's proof before exposing FIN. + if let (Some(receipts), Some(hasher)) = (&payload_receipts, payload_hasher) { let digest = *hasher.finalize().as_bytes(); if !receipts.set_digest(produced.header.seq, digest) { tracing::debug!( @@ -2697,6 +2698,95 @@ mod tests { assert_eq!(decoded, [0, 1, 5, 6]); } + #[tokio::test(flavor = "multi_thread", worker_threads = 2)] + async fn completed_supersession_installs_its_payload_receipt_digest() { + for backend in [rds_net::Backend::Iroh, rds_net::Backend::Noq] { + tokio::time::timeout(Duration::from_secs(15), async { + let config = rds_net::EndpointConfig { + backend, + discovery: false, + bind_addrs: vec!["127.0.0.1:0".parse().unwrap()], + ..Default::default() + }; + let client = rds_net::bind_endpoint(config.clone()).await.unwrap(); + let server = rds_net::bind_endpoint(config).await.unwrap(); + let (a, b) = tokio::join!(client.connect(server.addr(), rds_core::ALPN), async { + server.accept().await.unwrap().await + }); + let (a, b) = (a.unwrap(), b.unwrap()); + let admission = Arc::new(AtomicU64::new(2)); + let payload = Bytes::from(vec![7; 32 * 1024 * 1024]); + let expected = *blake3::hash(&payload).as_bytes(); + let mut first = produced(0, true); + first.payload = payload; + let first = AdmittedFrame { + produced: first, + generation: 0, + key: None, + permit: CapturePermit(admission.clone()), + }; + let (tx, mut rx) = mpsc::channel(1); + let mut acknowledgements = JoinSet::new(); + let proofs = crate::receipts::Receipts::new(); + let delivery = FrameDelivery { + repair: crate::media_repair::MediaRepair::new().0, + idr: Arc::new(AtomicBool::new(false)), + feedback: Arc::new(DeliveryFeedback::default()), + latest_key_seq: Arc::new(AtomicU64::new(u64::MAX)), + payload_receipts: Some(proofs.clone()), + }; + let (outcome, ()) = tokio::join!( + send_frame_inner( + &a, + rds_core::UniHello::Desktop, + first, + &mut rx, + &mut acknowledgements, + delivery + ), + async { + let mut stream = b.accept_uni().await.unwrap(); + let _: rds_core::UniHello = read_frame(&mut stream).await.unwrap(); + let header: FrameHeader = read_frame(&mut stream).await.unwrap(); + assert_eq!(header.seq, 0); + // Keep the large original payload blocked until the + // producer event has selected its completed-supersession + // path. The receiver then drains it exactly once. + tx.send(AdmittedFrame { + produced: produced(1, false), + generation: 0, + key: None, + permit: CapturePermit(admission.clone()), + }) + .await + .unwrap(); + tokio::time::sleep(Duration::from_millis(20)).await; + let bytes = stream.read_to_end(32 * 1024 * 1024).await.unwrap(); + assert_eq!(*blake3::hash(&bytes).as_bytes(), expected); + assert!( + proofs.confirm(0, &expected, false), + "a completed superseded frame rejected its exact receipt" + ); + } + ); + let SendOutcome::Superseded(next) = outcome else { + panic!("fixture did not take the completed-supersession path"); + }; + assert_eq!(next.produced.header.seq, 1); + drop(next); + assert!(matches!( + acknowledgements.join_next().await.unwrap().unwrap(), + (0, FrameReceipt::Delivered) + )); + assert_eq!(admission.load(Ordering::Acquire), 0); + a.close(0u32.into(), b"test done"); + tokio::join!(client.close(), server.close()); + }) + .await + .expect("completed supersession hung"); + } + } + #[tokio::test(flavor = "multi_thread", worker_threads = 2)] async fn cancelled_frame_writer_resets_stream_and_preserves_connection() { for backend in [rds_net::Backend::Iroh, rds_net::Backend::Noq] { From b7a1f733d8ba2736de1bce9e0e3feff97817bee9 Mon Sep 17 00:00:00 2001 From: Danil Silantyev Date: Thu, 8 Oct 2026 04:24:23 +0500 Subject: [PATCH 5/6] docs(desktop): retain supersession receipt regression evidence --- docs/reports/rds-desktop-cpu-20261008.md | 15 +++++++++++++++ 1 file changed, 15 insertions(+) diff --git a/docs/reports/rds-desktop-cpu-20261008.md b/docs/reports/rds-desktop-cpu-20261008.md index e904ce3f..61cfd3c4 100644 --- a/docs/reports/rds-desktop-cpu-20261008.md +++ b/docs/reports/rds-desktop-cpu-20261008.md @@ -51,6 +51,21 @@ also ineffective: OpenH264's screen-content `ParamValidation` forces it back on. and the exact openh264-sys2 0.9.8 source confirm that behavior. Medium codec complexity remains; lowering it would be a separate quality/bitrate decision. +## Completed payload receipt regression + +The incremental hashing change exposed a missing completion case: when the +producer supplied a successor during a blocked write, `send_payload` finished +the original reference and returned `Superseded(next)`. The sender finalized +receipt digests only for `Sent` and `Done`, so a fully read superseded frame +rejected its exact proof and held sender admission until timeout. + +Every completed write now installs its digest before FIN. An abandoned delta +returns earlier and retains its reset behavior. A real loopback regression +blocks a 32 MiB keyframe, admits a successor, drains the original byte-exact, +confirms its proof and requires delivery plus release of both admission slots. +The original code failed the exact-receipt assertion on Iroh; corrected code +passes on both Iroh and Noq. No additional payload scan or wire field is added. + ## Qualification boundary The macOS x11 feature lane checks the codec and scaling code; native X11 work From 33fe309df050a2643e7f3cfe9ea11da116231236 Mon Sep 17 00:00:00 2001 From: Danil Silantyev Date: Thu, 8 Oct 2026 04:48:40 +0500 Subject: [PATCH 6/6] test(net): fence contemporary standby proof before recovery timing --- .../src/backends/noq/path_blackhole_tests.rs | 26 +++++++++++++++++++ docs/reports/rds-desktop-cpu-20261008.md | 18 +++++++++++++ 2 files changed, 44 insertions(+) diff --git a/crates/rds-net/src/backends/noq/path_blackhole_tests.rs b/crates/rds-net/src/backends/noq/path_blackhole_tests.rs index edad6e25..4158ccd6 100644 --- a/crates/rds-net/src/backends/noq/path_blackhole_tests.rs +++ b/crates/rds-net/src/backends/noq/path_blackhole_tests.rs @@ -474,6 +474,32 @@ async fn exercise(automatic: bool) { .await .is_err() ); + if automatic { + // Warm validation proves the standby before the failure. Retirement + // additionally requires an ACK during the pending work interval. + // Fence that real protocol precondition before measuring the two- + // second policy/stream recovery gate; otherwise a one-second probe + // plus the next selection tick races the same two-second deadline. + remote_secondary.ping().unwrap(); + tokio::time::timeout(Duration::from_secs(2), async { + loop { + let failed = b + .congestion_state(noq::PathId::ZERO) + .and_then(crate::ack_progress::snapshot); + let sibling = b + .congestion_state(remote_secondary.id()) + .and_then(crate::ack_progress::snapshot); + if failed.zip(sibling).is_some_and(|(failed, sibling)| { + sibling.can_replace(failed, remote_secondary.stats().rtt) + }) { + break; + } + tokio::time::sleep(Duration::from_millis(5)).await; + } + }) + .await + .expect("standby did not acknowledge during the failed work interval"); + } let (_relay_sender, relay_dead) = tokio::sync::watch::channel(0u64); let driver = if automatic { let telemetry = super::telemetry::Telemetry::new(&b); diff --git a/docs/reports/rds-desktop-cpu-20261008.md b/docs/reports/rds-desktop-cpu-20261008.md index 61cfd3c4..0090dd57 100644 --- a/docs/reports/rds-desktop-cpu-20261008.md +++ b/docs/reports/rds-desktop-cpu-20261008.md @@ -73,3 +73,21 @@ must pass on Linux under a dedicated Xvfb display. Required PR checks cover both supported OSes and Linux native fixtures. Installed runtime receipts and private identities belong to the estate. This increment does not close the broader hardware-codec, geometry, quality or glass-to-glass gates. + +## CI recovery-fixture precondition + +The first final macOS CI run failed the existing Noq ACK-starvation fixture at +its two-second reliable-byte recovery deadline. Its warm standby proof preceded +the pending-work interval; retirement requires sibling acknowledgement during +that interval. Initial idle probes and the one-second selection tick could race +the same two-second deadline. A local original-code repeat passed, retaining the +CI failure as scheduling-sensitive evidence rather than a deterministic product +regression. + +The fixture now sends a real standby PING after the failed STREAM transmission +and waits for `can_replace` evidence before starting its policy/stream recovery +measurement. The two-second recovery limit and production policy are unchanged; +precondition setup has its own two-second failure bound. This tests recovery +from a contemporaneously proved sibling; it does not measure the combined delay +for initially idle probing plus recovery. Whole installed failover qualification +remains separate.