diff --git a/crates/rds-desktop/src/capture/x11.rs b/crates/rds-desktop/src/capture/x11.rs index 7e722589..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. @@ -170,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( @@ -192,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: Bytes::copy_from_slice(&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"); @@ -230,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()), }) } @@ -347,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 a8a2bb33..2c700ebd 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,8 +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); + 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 { @@ -2024,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!( @@ -2696,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] { 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/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..0090dd57 --- /dev/null +++ b/docs/reports/rds-desktop-cpu-20261008.md @@ -0,0 +1,93 @@ +# 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. + +## 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 +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. 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