From 048f3f7bca2c8c3eded466024a6e430b57b0b37c Mon Sep 17 00:00:00 2001 From: Junyi Ou Date: Thu, 8 Oct 2026 17:10:35 -0400 Subject: [PATCH 1/2] exp(video-streamer): refill quiet live clips with the last picture Experiment behind SessionConfig::fill_interval (DGATEWAY_SHADOW_FILL_MS in Gateway) plus examples/sparse_fill.rs, a timed replay harness. --- .../video-streamer/benches/stream_session.rs | 1 + crates/video-streamer/benches/vpx_reencode.rs | 1 + crates/video-streamer/examples/sparse_fill.rs | 451 ++++++++++++++++++ crates/video-streamer/examples/stream/main.rs | 1 + crates/video-streamer/src/decoder.rs | 27 +- crates/video-streamer/src/normalizer/mod.rs | 97 +++- .../src/normalizer/tests/mod.rs | 11 +- .../src/normalizer/tests/replay.rs | 1 + crates/video-streamer/src/session.rs | 5 + crates/video-streamer/tests/support/mod.rs | 1 + devolutions-gateway/src/streaming.rs | 14 +- 11 files changed, 592 insertions(+), 18 deletions(-) create mode 100644 crates/video-streamer/examples/sparse_fill.rs diff --git a/crates/video-streamer/benches/stream_session.rs b/crates/video-streamer/benches/stream_session.rs index 15597cbb5..e7cffd586 100644 --- a/crates/video-streamer/benches/stream_session.rs +++ b/crates/video-streamer/benches/stream_session.rs @@ -74,6 +74,7 @@ async fn stream_recording(recording: &'static [u8]) -> u64 { let config = SessionConfig { encoder_threads: 1, adaptive_frame_skip: false, + fill_interval: None, }; let server = tokio::spawn(stream_session( start_source, diff --git a/crates/video-streamer/benches/vpx_reencode.rs b/crates/video-streamer/benches/vpx_reencode.rs index 2656de5ae..ecea7221e 100644 --- a/crates/video-streamer/benches/vpx_reencode.rs +++ b/crates/video-streamer/benches/vpx_reencode.rs @@ -66,6 +66,7 @@ fn bench_reencode_first_500_tags(c: &mut Criterion) { video_streamer::SessionConfig { encoder_threads: 1, adaptive_frame_skip: false, + fill_interval: None, }, 500, per_iter_deadline, diff --git a/crates/video-streamer/examples/sparse_fill.rs b/crates/video-streamer/examples/sparse_fill.rs new file mode 100644 index 000000000..eff92ba0a --- /dev/null +++ b/crates/video-streamer/examples/sparse_fill.rs @@ -0,0 +1,451 @@ +//! Experiment: replays a WebM recording into `stream_session` at its real cadence, as a live clip, and reports when +//! output frames reach a client that pulls continuously. +//! +//! Usage: sparse_fill + +#![allow( + clippy::cast_possible_truncation, + clippy::cast_sign_loss, + clippy::cast_precision_loss, + reason = "experiment statistics" +)] + +use std::io::{self, Cursor, Read, Seek, SeekFrom}; +use std::path::PathBuf; +use std::pin::Pin; +use std::sync::{Arc, Mutex}; +use std::task::{Context, Poll}; +use std::time::{Duration, Instant}; + +use anyhow::Context as _; +use bytes::Bytes; +use futures_util::{Sink, Stream}; +use tokio::sync::mpsc; +use video_streamer::{RecordingClip, RecordingEvent, SessionConfig, ShadowProtocolVersion, StartAt, stream_session}; +use webm_iterable::errors::TagIteratorError; +use webm_iterable::matroska_spec::{Block, Master, MatroskaSpec, SimpleBlock}; +use webm_iterable::{WebmIterator, WebmWriter, WriteOptions}; + +struct SourceFrame { + timestamp_ms: u64, + data: Vec, + key_frame: bool, +} + +fn read_input(path: &PathBuf, limit_ms: u64) -> anyhow::Result<(String, u64, u64, Vec)> { + let file = std::fs::File::open(path)?; + let mut tags = WebmIterator::new(io::BufReader::new(file), &[MatroskaSpec::BlockGroup(Master::Start)]); + tags.emit_master_end_when_eof(false); + let (mut codec, mut width, mut height, mut cluster) = (None, None, None, None); + let mut frames = Vec::new(); + for tag in &mut tags { + let tag = match tag { + Ok(tag) => tag, + Err(TagIteratorError::UnexpectedEOF { .. }) => break, + Err(error) => return Err(error.into()), + }; + let (relative, data, key_frame) = match tag { + MatroskaSpec::CodecID(id) => { + codec = Some(id); + continue; + } + MatroskaSpec::PixelWidth(value) => { + width = Some(value); + continue; + } + MatroskaSpec::PixelHeight(value) => { + height = Some(value); + continue; + } + MatroskaSpec::Timestamp(value) => { + cluster = Some(value); + continue; + } + ref simple @ MatroskaSpec::SimpleBlock(_) => { + let block = SimpleBlock::try_from(simple)?; + let data = block.read_frame_data()?.remove(0).data.to_vec(); + (block.timestamp, data, block.keyframe) + } + MatroskaSpec::BlockGroup(Master::Full(children)) => { + let Some(raw @ MatroskaSpec::Block(_)) = + children.iter().find(|tag| matches!(tag, MatroskaSpec::Block(_))) + else { + continue; + }; + let block = Block::try_from(raw)?; + let data = block.read_frame_data()?.remove(0).data.to_vec(); + let key_frame = data.first().is_some_and(|byte| byte & 1 == 0); + (block.timestamp, data, key_frame) + } + _ => continue, + }; + let timestamp_ms = + u64::try_from(i64::try_from(cluster.context("block before cluster")?)? + i64::from(relative))?; + if frames + .first() + .is_some_and(|first: &SourceFrame| timestamp_ms - first.timestamp_ms > limit_ms) + { + break; + } + frames.push(SourceFrame { + timestamp_ms, + data, + key_frame, + }); + } + Ok(( + codec.context("codec")?, + width.context("width")?, + height.context("height")?, + frames, + )) +} + +/// Header bytes, then one cluster per frame. +#[derive(Clone, Default)] +struct SharedBuffer(std::rc::Rc>>); + +impl io::Write for SharedBuffer { + fn write(&mut self, data: &[u8]) -> io::Result { + self.0.borrow_mut().extend_from_slice(data); + Ok(data.len()) + } + + fn flush(&mut self) -> io::Result<()> { + Ok(()) + } +} + +fn build_feed(codec: &str, width: u64, height: u64, frames: &[SourceFrame]) -> anyhow::Result<(Vec, Vec>)> { + let buffer = SharedBuffer::default(); + let take = |buffer: &SharedBuffer| core::mem::take(&mut *buffer.0.borrow_mut()); + let mut header = WebmWriter::new(buffer.clone()); + header.write(&MatroskaSpec::Ebml(Master::Full(vec![ + MatroskaSpec::EbmlVersion(1), + MatroskaSpec::EbmlReadVersion(1), + MatroskaSpec::EbmlMaxIdLength(4), + MatroskaSpec::EbmlMaxSizeLength(8), + MatroskaSpec::DocType("webm".to_owned()), + MatroskaSpec::DocTypeVersion(4), + MatroskaSpec::DocTypeReadVersion(2), + ])))?; + header.write_advanced( + &MatroskaSpec::Segment(Master::Start), + WriteOptions::is_unknown_sized_element(), + )?; + header.write(&MatroskaSpec::Info(Master::Full(vec![MatroskaSpec::TimestampScale( + 1_000_000, + )])))?; + header.write(&MatroskaSpec::Tracks(Master::Full(vec![MatroskaSpec::TrackEntry( + Master::Full(vec![ + MatroskaSpec::TrackNumber(1), + MatroskaSpec::TrackUID(1), + MatroskaSpec::TrackType(1), + MatroskaSpec::CodecID(codec.to_owned()), + MatroskaSpec::Video(Master::Full(vec![ + MatroskaSpec::PixelWidth(width), + MatroskaSpec::PixelHeight(height), + ])), + ]), + )])))?; + let header_bytes = take(&buffer); + + let clusters = frames + .iter() + .map(|frame| { + header.write(&MatroskaSpec::Cluster(Master::Full(vec![ + MatroskaSpec::Timestamp(frame.timestamp_ms), + MatroskaSpec::from(SimpleBlock::new_uncheked( + &frame.data, + 1, + 0, + false, + None, + false, + frame.key_frame, + )), + ])))?; + Ok(take(&buffer)) + }) + .collect::>>()?; + Ok((header_bytes, clusters)) +} + +/// A growing in-memory file. +#[derive(Clone, Default)] +struct LiveFile(Arc>>); + +struct LiveReader { + file: LiveFile, + position: u64, +} + +impl Read for LiveReader { + fn read(&mut self, buffer: &mut [u8]) -> io::Result { + let data = self.file.0.lock().expect("poisoned"); + let mut cursor = Cursor::new(data.as_slice()); + cursor.set_position(self.position); + let read = cursor.read(buffer)?; + self.position += read as u64; + Ok(read) + } +} + +impl Seek for LiveReader { + fn seek(&mut self, position: SeekFrom) -> io::Result { + let length = self.file.0.lock().expect("poisoned").len() as u64; + self.position = match position { + SeekFrom::Start(offset) => offset, + SeekFrom::End(offset) => length.saturating_add_signed(offset), + SeekFrom::Current(offset) => self.position.saturating_add_signed(offset), + }; + Ok(self.position) + } +} + +struct ChannelTransport { + incoming: mpsc::UnboundedReceiver>, + outgoing: mpsc::UnboundedSender<(Instant, Bytes)>, +} + +impl Stream for ChannelTransport { + type Item = Result; + + fn poll_next(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll> { + self.get_mut().incoming.poll_recv(cx) + } +} + +impl Sink for ChannelTransport { + type Error = io::Error; + + fn poll_ready(self: Pin<&mut Self>, _: &mut Context<'_>) -> Poll> { + Poll::Ready(Ok(())) + } + + fn start_send(self: Pin<&mut Self>, item: Bytes) -> Result<(), Self::Error> { + self.outgoing + .send((Instant::now(), item)) + .map_err(|_| io::Error::from(io::ErrorKind::BrokenPipe)) + } + + fn poll_flush(self: Pin<&mut Self>, _: &mut Context<'_>) -> Poll> { + Poll::Ready(Ok(())) + } + + fn poll_close(self: Pin<&mut Self>, _: &mut Context<'_>) -> Poll> { + Poll::Ready(Ok(())) + } +} + +struct OutputFrame { + arrival_ms: f64, + timestamp_ms: u64, + bytes: usize, +} + +fn read_vint(data: &[u8]) -> Option<(u64, usize)> { + let first = *data.first()?; + let length = first.leading_zeros() as usize + 1; + if length > 8 || data.len() < length { + return None; + } + let mut value = u64::from(first) & ((1u64 << (8 - length)) - 1); + for byte in &data[1..length] { + value = (value << 8) | u64::from(*byte); + } + Some((value, length)) +} + +/// Finds Cluster Timestamp and SimpleBlock elements in a chunk of server output. +fn scan_output(chunk: &[u8], cluster: &mut u64, arrival_ms: f64, out: &mut Vec) { + let mut at = 0; + while at < chunk.len() { + let rest = &chunk[at..]; + if rest.starts_with(&[0x1F, 0x43, 0xB6, 0x75]) { + // Cluster of unknown size: step inside. + at += 4 + read_vint(&rest[4..]).map_or(1, |(_, length)| length); + continue; + } + let Some(id_length) = rest.first().map(|byte| byte.leading_zeros() as usize + 1) else { + break; + }; + let Some((size, size_length)) = read_vint(&rest[id_length..]) else { + break; + }; + let body = id_length + size_length; + let Ok(size) = usize::try_from(size) else { break }; + if rest.len() < body + size { + break; + } + let payload = &rest[body..body + size]; + match rest[0] { + 0xE7 if id_length == 1 => { + *cluster = payload.iter().fold(0, |value, byte| (value << 8) | u64::from(*byte)); + } + 0xA3 if id_length == 1 => { + let relative = i16::from_be_bytes([payload[1], payload[2]]); + out.push(OutputFrame { + arrival_ms, + timestamp_ms: cluster.saturating_add_signed(i64::from(relative)), + bytes: size, + }); + } + _ => {} + } + at += body + size; + } +} + +fn percentile(values: &mut [f64], p: f64) -> f64 { + if values.is_empty() { + return f64::NAN; + } + values.sort_by(f64::total_cmp); + values[((values.len() - 1) as f64 * p).round() as usize] +} + +#[tokio::main(flavor = "multi_thread", worker_threads = 4)] +#[expect(clippy::print_stdout, reason = "example output")] +async fn main() -> anyhow::Result<()> { + let args: Vec = std::env::args().collect(); + anyhow::ensure!( + args.len() == 6, + "usage: sparse_fill " + ); + // SAFETY: XMF has no initialization precondition beyond a valid library path. + unsafe { cadeau::xmf::init(&args[1]) }.context("load XMF")?; + let input = PathBuf::from(&args[2]); + let limit_ms = args[3].parse::()? * 1000; + let fill_ms = args[4].parse::()?; + let report = PathBuf::from(&args[5]); + + let (codec, width, height, frames) = read_input(&input, limit_ms)?; + anyhow::ensure!(!frames.is_empty(), "no frames"); + let (header, clusters) = build_feed(&codec, width, height, &frames)?; + let first_ms = frames[0].timestamp_ms; + let source_ms: Vec = frames.iter().map(|frame| frame.timestamp_ms - first_ms).collect(); + let last_ms = *source_ms.last().expect("not empty"); + + let file = LiveFile::default(); + file.0.lock().expect("poisoned").extend_from_slice(&header); + file.0.lock().expect("poisoned").extend_from_slice(&clusters[0]); + + let (event_sender, mut event_receiver) = mpsc::unbounded_channel::>(); + let start = Instant::now(); + let feeder = { + let file = file.clone(); + let source_ms = source_ms.clone(); + tokio::spawn(async move { + for (offset, cluster) in source_ms.iter().zip(&clusters).skip(1) { + tokio::time::sleep_until((start + Duration::from_millis(*offset)).into()).await; + file.0.lock().expect("poisoned").extend_from_slice(cluster); + let _ = event_sender.send(Ok(RecordingEvent::DataAvailable)); + } + // Keep the clip open past the last frame, like a producer that is still connected. + tokio::time::sleep(Duration::from_secs(10)).await; + let _ = event_sender.send(Ok(RecordingEvent::ClipEnded)); + let _ = event_sender.send(Ok(RecordingEvent::SessionEnded)); + }) + }; + + let source_file = file.clone(); + let start_source = move || async move { + let clip = RecordingClip::new(LiveReader { + file: source_file, + position: 0, + }); + let head = futures_util::stream::iter([ + Ok(RecordingEvent::ClipStarted { + sequence: 0, + start_at: StartAt::LiveEdge, + clip, + }), + Ok(RecordingEvent::DataAvailable), + Ok(RecordingEvent::CaughtUp), + ]); + let tail = futures_util::stream::poll_fn(move |cx| event_receiver.poll_recv(cx)); + Ok(futures_util::StreamExt::chain(head, tail)) + }; + + let (client_sender, client_receiver) = mpsc::unbounded_channel(); + let (server_sender, mut server_receiver) = mpsc::unbounded_channel(); + let transport = ChannelTransport { + incoming: client_receiver, + outgoing: server_sender, + }; + client_sender.send(Ok(Bytes::from_static(&[0])))?; + for _ in 0..4 { + client_sender.send(Ok(Bytes::from_static(&[1])))?; + } + + let config = SessionConfig { + encoder_threads: 1, + adaptive_frame_skip: false, + fill_interval: (fill_ms > 0).then(|| Duration::from_millis(fill_ms)), + }; + let server = tokio::spawn(stream_session( + move || async move { start_source().await }, + transport, + config, + ShadowProtocolVersion::V2, + )); + + let mut output = Vec::new(); + let mut cluster = 0; + let mut total_bytes = 0; + while let Some((at, message)) = server_receiver.recv().await { + let _ = client_sender.send(Ok(Bytes::from_static(&[1]))); + let Some((&kind, payload)) = message.split_first() else { + continue; + }; + match kind { + 0x00 => { + total_bytes += payload.len(); + scan_output(payload, &mut cluster, (at - start).as_secs_f64() * 1000.0, &mut output); + } + 0x03 | 0x02 => break, + _ => {} + } + } + drop(client_sender); + server.await??; + feeder.await?; + + // Output timestamps restart at 0 per segment; the first output frame is the first source frame. + let mut wall_gaps: Vec = output + .windows(2) + .map(|pair| pair[1].arrival_ms - pair[0].arrival_ms) + .collect(); + let mut media_gaps: Vec = output + .windows(2) + .map(|pair| pair[1].timestamp_ms.saturating_sub(pair[0].timestamp_ms) as f64) + .collect(); + // Delay between when the source made a picture available and when the client got the frame showing it. + let mut delays: Vec = output + .iter() + .map(|frame| frame.arrival_ms - frame.timestamp_ms as f64) + .collect(); + let source_frames = frames.len(); + let output_frames = output.len(); + let wall_gap_max = percentile(&mut wall_gaps.clone(), 1.0); + let wall_gap_p95 = percentile(&mut wall_gaps, 0.95); + let media_gap_max = percentile(&mut media_gaps, 1.0); + let delay_p50 = percentile(&mut delays.clone(), 0.5); + let delay_max = percentile(&mut delays, 1.0); + let frame_bytes: usize = output.iter().map(|frame| frame.bytes).sum(); + let span_s = last_ms as f64 / 1000.0 + 10.0; + let json = format!( + "{{\"input\":\"{}\",\"fill_ms\":{fill_ms},\"span_s\":{span_s:.1},\"source_frames\":{source_frames},\"output_frames\":{output_frames},\"output_bytes\":{total_bytes},\"frame_bytes\":{frame_bytes},\"kbps\":{:.2},\"wall_gap_p95_ms\":{wall_gap_p95:.0},\"wall_gap_max_ms\":{wall_gap_max:.0},\"media_gap_max_ms\":{media_gap_max:.0},\"delay_p50_ms\":{delay_p50:.0},\"delay_max_ms\":{delay_max:.0},\"source_ms\":{source_ms:?},\"frames\":[{}]}}", + input.display().to_string().replace('\\', "/"), + total_bytes as f64 * 8.0 / span_s / 1000.0, + output + .iter() + .map(|frame| format!("[{:.0},{},{}]", frame.arrival_ms, frame.timestamp_ms, frame.bytes)) + .collect::>() + .join(","), + ); + std::fs::write(&report, &json)?; + println!("{}", json.split(",\"source_ms\"").next().unwrap_or_default()); + Ok(()) +} diff --git a/crates/video-streamer/examples/stream/main.rs b/crates/video-streamer/examples/stream/main.rs index 68209a19a..a893a77e8 100644 --- a/crates/video-streamer/examples/stream/main.rs +++ b/crates/video-streamer/examples/stream/main.rs @@ -72,6 +72,7 @@ async fn main() -> anyhow::Result<()> { SessionConfig { encoder_threads: SessionConfig::default().encoder_threads, adaptive_frame_skip: true, + fill_interval: None, }, || { let (tx, rx) = tokio::sync::oneshot::channel(); diff --git a/crates/video-streamer/src/decoder.rs b/crates/video-streamer/src/decoder.rs index b8fce099e..3f0b9fd02 100644 --- a/crates/video-streamer/src/decoder.rs +++ b/crates/video-streamer/src/decoder.rs @@ -7,14 +7,12 @@ pub(crate) struct Dimensions { pub height: u32, } -pub(crate) struct DecodedFrame<'decoder> { - pub image: VpxImage<'decoder>, - pub dimensions: Dimensions, -} - pub(crate) struct InputDecoder { codec: VpxCodec, threads: u32, + // INVARIANT: `picture` points into a frame buffer owned by `decoder`, which stays valid until the next decode. + // It is cleared before every decode, and it is declared before `decoder` so that it is dropped first. + picture: Option>, decoder: Option, } @@ -23,11 +21,15 @@ impl InputDecoder { Self { codec, threads, + picture: None, decoder: None, } } - pub(crate) fn decode<'decoder>(&'decoder mut self, data: &[u8]) -> anyhow::Result> { + /// Decodes one frame and keeps its picture until the next call. + pub(crate) fn decode(&mut self, data: &[u8]) -> anyhow::Result { + self.picture = None; + if self.decoder.is_none() { self.decoder = Some( VpxDecoder::builder() @@ -50,6 +52,17 @@ impl InputDecoder { dimensions.width > 0 && dimensions.height > 0, "decoder returned invalid frame dimensions" ); - Ok(DecodedFrame { image, dimensions }) + + // SAFETY: Only the lifetime changes. Per the field invariant, the picture is dropped before the decoder + // decodes again or is dropped, so it never outlives the frame buffer it points into. + let image = unsafe { core::mem::transmute::, VpxImage<'static>>(image) }; + self.picture = Some(image); + + Ok(dimensions) + } + + /// The picture of the last decoded frame. + pub(crate) fn picture(&self) -> Option<&VpxImage<'_>> { + self.picture.as_ref() } } diff --git a/crates/video-streamer/src/normalizer/mod.rs b/crates/video-streamer/src/normalizer/mod.rs index 6a6bbcbfc..05fc96fbe 100644 --- a/crates/video-streamer/src/normalizer/mod.rs +++ b/crates/video-streamer/src/normalizer/mod.rs @@ -110,9 +110,26 @@ where let mut worker = tokio::task::spawn_blocking(move || normalize_events(input_receiver, worker_sender, config)); let mut forward = Box::pin(async move { tokio::pin!(source); - while let Some(event) = source.next().await { - if input_sender.send(event).await.is_err() { - break; + // Ticks drive the refill of quiet live clips; with no fill interval, the tick never fires. + let tick_period = config + .fill_interval + .map_or(Duration::from_secs(3600), |interval| interval / 4); + let mut ticks = tokio::time::interval(tick_period.max(Duration::from_millis(10))); + ticks.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip); + loop { + tokio::select! { + event = source.next() => { + let Some(event) = event else { break }; + if input_sender.send(event.map(InputEvent::Recording)).await.is_err() { + break; + } + } + _ = ticks.tick(), if config.fill_interval.is_some() => { + // A busy worker misses the tick, which is fine: it is producing frames. + if let Err(mpsc::error::TrySendError::Closed(_)) = input_sender.try_send(Ok(InputEvent::Tick)) { + break; + } + } } } }); @@ -148,8 +165,13 @@ async fn publish_worker_result( let _ = sender.send(Err(error)).await; } +enum InputEvent { + Recording(RecordingEvent), + Tick, +} + fn normalize_events( - mut receiver: mpsc::Receiver>, + mut receiver: mpsc::Receiver>, sender: mpsc::Sender>, config: SessionConfig, ) -> anyhow::Result<()> { @@ -157,7 +179,16 @@ fn normalize_events( let mut next_segment_sequence = 0; while let Some(event) = receiver.blocking_recv() { - match event.context("recording source failed")? { + let event = match event.context("recording source failed")? { + InputEvent::Recording(event) => event, + InputEvent::Tick => { + if let SessionPhase::InClip(clip) = &mut phase { + clip.fill_if_quiet()?; + } + continue; + } + }; + match event { RecordingEvent::ClipStarted { sequence, start_at, @@ -275,6 +306,8 @@ struct ClipNormalizer { processing_time: Duration, first_frame_timestamp: Option, frames_since_last_encode: u32, + /// Media timestamp of the last source frame and when it was processed, to place refill frames. + last_source_frame: Option<(u64, Instant)>, sender: mpsc::Sender>, config: SessionConfig, } @@ -314,6 +347,7 @@ impl ClipNormalizer { processing_time: Duration::ZERO, first_frame_timestamp: None, frames_since_last_encode: 0, + last_source_frame: None, sender, config, }) @@ -583,8 +617,8 @@ impl ClipNormalizer { let input_decoder = self .input_decoder .get_or_insert_with(|| InputDecoder::new(frame.codec, self.config.encoder_threads)); - let decoded = input_decoder.decode(&frame.data)?; - let dimensions = decoded.dimensions; + let dimensions = input_decoder.decode(&frame.data)?; + self.last_source_frame = Some((frame.timestamp, Instant::now())); let new_segment = next_segment_info( self.output_segment.as_ref().map(|segment| segment.dimensions), dimensions, @@ -611,13 +645,56 @@ impl ClipNormalizer { .checked_add(1) .context("segment sequence overflow")?; } + let picture = self + .input_decoder + .as_ref() + .and_then(InputDecoder::picture) + .context("decoded picture is missing")?; self.output_segment .as_mut() .context("output segment is missing")? - .encode(&decoded.image, frame.timestamp)?; + .encode(picture, frame.timestamp)?; Ok(()) } + /// Repeats the last picture when the live source has been quiet for at least the fill interval. + /// + /// The refill timestamp follows the source clock: the last source timestamp plus the time elapsed since that + /// frame. A source frame that arrives later with a smaller timestamp is moved after the refill. + fn fill_if_quiet(&mut self) -> anyhow::Result<()> { + let Some(interval) = self.config.fill_interval else { + return Ok(()); + }; + if !matches!(self.phase, ClipPhase::Live) || self.sender.is_closed() { + return Ok(()); + } + let Some((source_timestamp, processed_at)) = self.last_source_frame else { + return Ok(()); + }; + let (Some(segment), Some(picture)) = ( + self.output_segment.as_mut(), + self.input_decoder.as_ref().and_then(InputDecoder::picture), + ) else { + return Ok(()); + }; + + let elapsed = processed_at.elapsed(); + if elapsed < interval { + return Ok(()); + } + let interval_ms = u64::try_from(interval.as_millis()).unwrap_or(u64::MAX); + let timestamp = source_timestamp.saturating_add(u64::try_from(elapsed.as_millis()).unwrap_or(u64::MAX)); + if segment + .previous_timestamp + .is_some_and(|previous| timestamp < previous.saturating_add(interval_ms)) + { + return Ok(()); + } + + perf_trace!(timestamp, "Refill quiet live clip"); + segment.encode(picture, timestamp) + } + fn should_skip_encode(&mut self, timestamp: u64) -> bool { let first_timestamp = *self.first_frame_timestamp.get_or_insert(timestamp); let media_advanced_ms = timestamp.saturating_sub(first_timestamp); @@ -874,6 +951,10 @@ impl OutputSegment { } fn encode(&mut self, image: &VpxImage<'_>, timestamp: u64) -> anyhow::Result<()> { + // A refill frame may already cover a source timestamp that arrives late. + let timestamp = self + .previous_timestamp + .map_or(timestamp, |previous| timestamp.max(previous.saturating_add(1))); let origin = *self.origin_timestamp.get_or_insert(timestamp); let relative_timestamp = timestamp.saturating_sub(origin); let duration = self diff --git a/crates/video-streamer/src/normalizer/tests/mod.rs b/crates/video-streamer/src/normalizer/tests/mod.rs index edcca2185..e65d9922c 100644 --- a/crates/video-streamer/src/normalizer/tests/mod.rs +++ b/crates/video-streamer/src/normalizer/tests/mod.rs @@ -198,6 +198,7 @@ where SessionConfig { encoder_threads: 1, adaptive_frame_skip: false, + fill_interval: None, }, 0, ) @@ -495,7 +496,9 @@ fn truncated_clip_tail_does_not_abort_the_following_clip() { ]; let (input_sender, input_receiver) = mpsc::channel(events.len()); for event in events { - input_sender.blocking_send(Ok(event)).expect("queue recording event"); + input_sender + .blocking_send(Ok(InputEvent::Recording(event))) + .expect("queue recording event"); } drop(input_sender); let (output_sender, mut output_receiver) = mpsc::channel(1); @@ -506,6 +509,7 @@ fn truncated_clip_tail_does_not_abort_the_following_clip() { SessionConfig { encoder_threads: 1, adaptive_frame_skip: false, + fill_interval: None, }, ) .expect("normalize reconnecting clips"); @@ -529,7 +533,9 @@ fn corruption_before_an_incomplete_tail_still_fails() { ]; let (input_sender, input_receiver) = mpsc::channel(events.len()); for event in events { - input_sender.blocking_send(Ok(event)).expect("queue recording event"); + input_sender + .blocking_send(Ok(InputEvent::Recording(event))) + .expect("queue recording event"); } drop(input_sender); let (output_sender, _output_receiver) = mpsc::channel(1); @@ -540,6 +546,7 @@ fn corruption_before_an_incomplete_tail_still_fails() { SessionConfig { encoder_threads: 1, adaptive_frame_skip: false, + fill_interval: None, }, ) .expect_err("corruption before the incomplete tail must fail"); diff --git a/crates/video-streamer/src/normalizer/tests/replay.rs b/crates/video-streamer/src/normalizer/tests/replay.rs index 179450056..49e79bf5a 100644 --- a/crates/video-streamer/src/normalizer/tests/replay.rs +++ b/crates/video-streamer/src/normalizer/tests/replay.rs @@ -195,6 +195,7 @@ fn closed_output_stops_scan_at_a_frame_checkpoint() { SessionConfig { encoder_threads: 1, adaptive_frame_skip: false, + fill_interval: None, }, 0, ) diff --git a/crates/video-streamer/src/session.rs b/crates/video-streamer/src/session.rs index 9ed26261e..c02b45562 100644 --- a/crates/video-streamer/src/session.rs +++ b/crates/video-streamer/src/session.rs @@ -2,6 +2,7 @@ use std::error::Error; use std::fmt; use std::future::Future; use std::io::{self, Read, Seek, SeekFrom}; +use std::time::Duration; use bytes::Bytes; use futures_util::{Sink, Stream}; @@ -89,6 +90,9 @@ pub struct SessionConfig { pub encoder_threads: u32, /// When `true`, the encoder skips frames while it falls behind real time, lowering the output frame rate. pub adaptive_frame_skip: bool, + /// When set, a live clip whose source stays quiet repeats its last picture at this interval, so players keep + /// receiving frames. + pub fill_interval: Option, } impl Default for SessionConfig { @@ -96,6 +100,7 @@ impl Default for SessionConfig { Self { encoder_threads: u32::try_from(num_cpus::get()).unwrap_or(1).max(1), adaptive_frame_skip: true, + fill_interval: None, } } } diff --git a/crates/video-streamer/tests/support/mod.rs b/crates/video-streamer/tests/support/mod.rs index 9ecfe1fb2..4e7aded35 100644 --- a/crates/video-streamer/tests/support/mod.rs +++ b/crates/video-streamer/tests/support/mod.rs @@ -410,6 +410,7 @@ pub(crate) async fn spawn_stream_harness_delayed_start( video_streamer::SessionConfig { encoder_threads: u32::try_from(encoder_threads).expect("thread count fits in u32"), adaptive_frame_skip: false, + fill_interval: None, }, when_new_chunk_appended, ) diff --git a/devolutions-gateway/src/streaming.rs b/devolutions-gateway/src/streaming.rs index 3f36edd4e..54425cdef 100644 --- a/devolutions-gateway/src/streaming.rs +++ b/devolutions-gateway/src/streaming.rs @@ -330,7 +330,7 @@ async fn setup_webm_streaming( ); let start_source = move || async move { Ok(recording_event_stream(stream_state)) }; let streaming_result = tokio::select! { - result = stream_session(start_source, websocket_stream, SessionConfig::default(), version) => result, + result = stream_session(start_source, websocket_stream, session_config(), version) => result, () = session_shutdown.wait() => return Ok(()), }; @@ -483,6 +483,18 @@ fn recording_event_stream( }) } +/// Experiment: `DGATEWAY_SHADOW_FILL_MS` sets the refill interval for quiet live clips. +fn session_config() -> SessionConfig { + let fill_interval = std::env::var("DGATEWAY_SHADOW_FILL_MS") + .ok() + .and_then(|value| value.parse::().ok()) + .filter(|&value| value > 0) + .map(Duration::from_millis); + SessionConfig { + fill_interval, + ..SessionConfig::default() + } +} #[cfg(test)] mod tests { use std::fs; From 43a5706938a4d978c7bafe00736dcde3308a1302 Mon Sep 17 00:00:00 2001 From: Junyi Ou Date: Thu, 8 Oct 2026 18:10:41 -0400 Subject: [PATCH 2/2] exp(video-streamer): delay refills and scan pending data first fill_delay (DGATEWAY_SHADOW_FILL_DELAY_MS) makes refills trail the source so a late source frame is not overtaken; refills now read already-written data first; count moved late frames; jitter injection in sparse_fill. --- .../video-streamer/benches/stream_session.rs | 1 + crates/video-streamer/benches/vpx_reencode.rs | 1 + crates/video-streamer/examples/sparse_fill.rs | 24 +++++++++++++++---- crates/video-streamer/examples/stream/main.rs | 1 + crates/video-streamer/src/lib.rs | 2 +- crates/video-streamer/src/normalizer/mod.rs | 23 ++++++++++++++---- .../src/normalizer/tests/mod.rs | 3 +++ .../src/normalizer/tests/replay.rs | 1 + crates/video-streamer/src/session.rs | 9 +++++++ crates/video-streamer/tests/support/mod.rs | 1 + devolutions-gateway/src/streaming.rs | 8 ++++++- 11 files changed, 63 insertions(+), 11 deletions(-) diff --git a/crates/video-streamer/benches/stream_session.rs b/crates/video-streamer/benches/stream_session.rs index e7cffd586..c21706131 100644 --- a/crates/video-streamer/benches/stream_session.rs +++ b/crates/video-streamer/benches/stream_session.rs @@ -75,6 +75,7 @@ async fn stream_recording(recording: &'static [u8]) -> u64 { encoder_threads: 1, adaptive_frame_skip: false, fill_interval: None, + fill_delay: Duration::ZERO, }; let server = tokio::spawn(stream_session( start_source, diff --git a/crates/video-streamer/benches/vpx_reencode.rs b/crates/video-streamer/benches/vpx_reencode.rs index ecea7221e..d3f03a2c4 100644 --- a/crates/video-streamer/benches/vpx_reencode.rs +++ b/crates/video-streamer/benches/vpx_reencode.rs @@ -67,6 +67,7 @@ fn bench_reencode_first_500_tags(c: &mut Criterion) { encoder_threads: 1, adaptive_frame_skip: false, fill_interval: None, + fill_delay: Duration::ZERO, }, 500, per_iter_deadline, diff --git a/crates/video-streamer/examples/sparse_fill.rs b/crates/video-streamer/examples/sparse_fill.rs index eff92ba0a..d472e6a77 100644 --- a/crates/video-streamer/examples/sparse_fill.rs +++ b/crates/video-streamer/examples/sparse_fill.rs @@ -1,7 +1,9 @@ //! Experiment: replays a WebM recording into `stream_session` at its real cadence, as a live clip, and reports when //! output frames reach a client that pulls continuously. //! -//! Usage: sparse_fill +//! Usage: sparse_fill [fill-delay-ms] [jitter-ms] +//! +//! `jitter-ms` delays each source frame by a pseudo-random 0..jitter-ms (order kept), like a congested push link. #![allow( clippy::cast_possible_truncation, @@ -310,8 +312,8 @@ fn percentile(values: &mut [f64], p: f64) -> f64 { async fn main() -> anyhow::Result<()> { let args: Vec = std::env::args().collect(); anyhow::ensure!( - args.len() == 6, - "usage: sparse_fill " + (6..=8).contains(&args.len()), + "usage: sparse_fill [fill-delay-ms] [jitter-ms]" ); // SAFETY: XMF has no initialization precondition beyond a valid library path. unsafe { cadeau::xmf::init(&args[1]) }.context("load XMF")?; @@ -319,6 +321,8 @@ async fn main() -> anyhow::Result<()> { let limit_ms = args[3].parse::()? * 1000; let fill_ms = args[4].parse::()?; let report = PathBuf::from(&args[5]); + let fill_delay_ms = args.get(6).map_or(Ok(0), |value| value.parse::())?; + let jitter_ms = args.get(7).map_or(Ok(0), |value| value.parse::())?; let (codec, width, height, frames) = read_input(&input, limit_ms)?; anyhow::ensure!(!frames.is_empty(), "no frames"); @@ -337,8 +341,16 @@ async fn main() -> anyhow::Result<()> { let file = file.clone(); let source_ms = source_ms.clone(); tokio::spawn(async move { + let mut state = 0x2545_f491_4f6c_dd1du64; + let mut not_before = 0; for (offset, cluster) in source_ms.iter().zip(&clusters).skip(1) { - tokio::time::sleep_until((start + Duration::from_millis(*offset)).into()).await; + state ^= state << 13; + state ^= state >> 7; + state ^= state << 17; + let jitter = if jitter_ms == 0 { 0 } else { state % (jitter_ms + 1) }; + let due = (offset + jitter).max(not_before); + not_before = due; + tokio::time::sleep_until((start + Duration::from_millis(due)).into()).await; file.0.lock().expect("poisoned").extend_from_slice(cluster); let _ = event_sender.send(Ok(RecordingEvent::DataAvailable)); } @@ -383,6 +395,7 @@ async fn main() -> anyhow::Result<()> { encoder_threads: 1, adaptive_frame_skip: false, fill_interval: (fill_ms > 0).then(|| Duration::from_millis(fill_ms)), + fill_delay: Duration::from_millis(fill_delay_ms), }; let server = tokio::spawn(stream_session( move || async move { start_source().await }, @@ -436,8 +449,9 @@ async fn main() -> anyhow::Result<()> { let frame_bytes: usize = output.iter().map(|frame| frame.bytes).sum(); let span_s = last_ms as f64 / 1000.0 + 10.0; let json = format!( - "{{\"input\":\"{}\",\"fill_ms\":{fill_ms},\"span_s\":{span_s:.1},\"source_frames\":{source_frames},\"output_frames\":{output_frames},\"output_bytes\":{total_bytes},\"frame_bytes\":{frame_bytes},\"kbps\":{:.2},\"wall_gap_p95_ms\":{wall_gap_p95:.0},\"wall_gap_max_ms\":{wall_gap_max:.0},\"media_gap_max_ms\":{media_gap_max:.0},\"delay_p50_ms\":{delay_p50:.0},\"delay_max_ms\":{delay_max:.0},\"source_ms\":{source_ms:?},\"frames\":[{}]}}", + "{{\"input\":\"{}\",\"fill_ms\":{fill_ms},\"fill_delay_ms\":{fill_delay_ms},\"jitter_ms\":{jitter_ms},\"late_frames_moved\":{},\"span_s\":{span_s:.1},\"source_frames\":{source_frames},\"output_frames\":{output_frames},\"output_bytes\":{total_bytes},\"frame_bytes\":{frame_bytes},\"kbps\":{:.2},\"wall_gap_p95_ms\":{wall_gap_p95:.0},\"wall_gap_max_ms\":{wall_gap_max:.0},\"media_gap_max_ms\":{media_gap_max:.0},\"delay_p50_ms\":{delay_p50:.0},\"delay_max_ms\":{delay_max:.0},\"source_ms\":{source_ms:?},\"frames\":[{}]}}", input.display().to_string().replace('\\', "/"), + video_streamer::late_frames_moved(), total_bytes as f64 * 8.0 / span_s / 1000.0, output .iter() diff --git a/crates/video-streamer/examples/stream/main.rs b/crates/video-streamer/examples/stream/main.rs index a893a77e8..156f4dc97 100644 --- a/crates/video-streamer/examples/stream/main.rs +++ b/crates/video-streamer/examples/stream/main.rs @@ -73,6 +73,7 @@ async fn main() -> anyhow::Result<()> { encoder_threads: SessionConfig::default().encoder_threads, adaptive_frame_skip: true, fill_interval: None, + fill_delay: core::time::Duration::ZERO, }, || { let (tx, rx) = tokio::sync::oneshot::channel(); diff --git a/crates/video-streamer/src/lib.rs b/crates/video-streamer/src/lib.rs index e0f341d1c..dba2f5c62 100644 --- a/crates/video-streamer/src/lib.rs +++ b/crates/video-streamer/src/lib.rs @@ -41,7 +41,7 @@ pub use streamer::webm_stream; #[rustfmt::skip] pub use session::{ RecordingClip, RecordingEvent, SHADOW_PROTOCOL_V2, SessionConfig, ShadowProtocolVersion, StartAt, - stream_session, + late_frames_moved, stream_session, }; #[cfg(feature = "bench")] diff --git a/crates/video-streamer/src/normalizer/mod.rs b/crates/video-streamer/src/normalizer/mod.rs index 05fc96fbe..742204f30 100644 --- a/crates/video-streamer/src/normalizer/mod.rs +++ b/crates/video-streamer/src/normalizer/mod.rs @@ -29,6 +29,9 @@ const WEBM_TIMESTAMP_SCALE_NS: u64 = 1_000_000; const MAX_WEBM_BLOCK_TIMESTAMP: u64 = 32_767; const MAX_CONSECUTIVE_FRAME_SKIPS: u32 = 1; +/// Experiment counter: frames whose timestamp was moved forward because an earlier output frame already covered it. +pub(crate) static LATE_FRAMES_MOVED: core::sync::atomic::AtomicU64 = core::sync::atomic::AtomicU64::new(0); + #[derive(Clone, Copy, Debug, Eq, PartialEq)] pub(crate) struct SegmentInfo { pub sequence: u64, @@ -668,6 +671,8 @@ impl ClipNormalizer { if !matches!(self.phase, ClipPhase::Live) || self.sender.is_closed() { return Ok(()); } + // A source frame already written but not scanned yet must not be overtaken by a refill. + self.scan_available()?; let Some((source_timestamp, processed_at)) = self.last_source_frame else { return Ok(()); }; @@ -678,7 +683,8 @@ impl ClipNormalizer { return Ok(()); }; - let elapsed = processed_at.elapsed(); + // Refills trail the source by `fill_delay`, so a source frame up to that late still lands before them. + let elapsed = processed_at.elapsed().saturating_sub(self.config.fill_delay); if elapsed < interval { return Ok(()); } @@ -952,9 +958,18 @@ impl OutputSegment { fn encode(&mut self, image: &VpxImage<'_>, timestamp: u64) -> anyhow::Result<()> { // A refill frame may already cover a source timestamp that arrives late. - let timestamp = self - .previous_timestamp - .map_or(timestamp, |previous| timestamp.max(previous.saturating_add(1))); + let timestamp = match self.previous_timestamp { + Some(previous) if timestamp <= previous => { + LATE_FRAMES_MOVED.fetch_add(1, core::sync::atomic::Ordering::Relaxed); + perf_debug!( + timestamp, + previous, + "Moved a late source frame after the previous output frame" + ); + previous.saturating_add(1) + } + _ => timestamp, + }; let origin = *self.origin_timestamp.get_or_insert(timestamp); let relative_timestamp = timestamp.saturating_sub(origin); let duration = self diff --git a/crates/video-streamer/src/normalizer/tests/mod.rs b/crates/video-streamer/src/normalizer/tests/mod.rs index e65d9922c..141eec37e 100644 --- a/crates/video-streamer/src/normalizer/tests/mod.rs +++ b/crates/video-streamer/src/normalizer/tests/mod.rs @@ -199,6 +199,7 @@ where encoder_threads: 1, adaptive_frame_skip: false, fill_interval: None, + fill_delay: Duration::ZERO, }, 0, ) @@ -510,6 +511,7 @@ fn truncated_clip_tail_does_not_abort_the_following_clip() { encoder_threads: 1, adaptive_frame_skip: false, fill_interval: None, + fill_delay: Duration::ZERO, }, ) .expect("normalize reconnecting clips"); @@ -547,6 +549,7 @@ fn corruption_before_an_incomplete_tail_still_fails() { encoder_threads: 1, adaptive_frame_skip: false, fill_interval: None, + fill_delay: Duration::ZERO, }, ) .expect_err("corruption before the incomplete tail must fail"); diff --git a/crates/video-streamer/src/normalizer/tests/replay.rs b/crates/video-streamer/src/normalizer/tests/replay.rs index 49e79bf5a..93747cebe 100644 --- a/crates/video-streamer/src/normalizer/tests/replay.rs +++ b/crates/video-streamer/src/normalizer/tests/replay.rs @@ -196,6 +196,7 @@ fn closed_output_stops_scan_at_a_frame_checkpoint() { encoder_threads: 1, adaptive_frame_skip: false, fill_interval: None, + fill_delay: Duration::ZERO, }, 0, ) diff --git a/crates/video-streamer/src/session.rs b/crates/video-streamer/src/session.rs index c02b45562..106cb563c 100644 --- a/crates/video-streamer/src/session.rs +++ b/crates/video-streamer/src/session.rs @@ -93,6 +93,8 @@ pub struct SessionConfig { /// When set, a live clip whose source stays quiet repeats its last picture at this interval, so players keep /// receiving frames. pub fill_interval: Option, + /// How far refills trail the source clock. A source frame that arrives up to this late is never overtaken. + pub fill_delay: Duration, } impl Default for SessionConfig { @@ -101,6 +103,7 @@ impl Default for SessionConfig { encoder_threads: u32::try_from(num_cpus::get()).unwrap_or(1).max(1), adaptive_frame_skip: true, fill_interval: None, + fill_delay: Duration::ZERO, } } } @@ -139,3 +142,9 @@ where { crate::protocol::stream_segments(transport, start_source, config, version).await } + +/// Experiment: source frames moved after an earlier output frame since the process started. +#[doc(hidden)] +pub fn late_frames_moved() -> u64 { + crate::normalizer::LATE_FRAMES_MOVED.load(core::sync::atomic::Ordering::Relaxed) +} diff --git a/crates/video-streamer/tests/support/mod.rs b/crates/video-streamer/tests/support/mod.rs index 4e7aded35..53c524eff 100644 --- a/crates/video-streamer/tests/support/mod.rs +++ b/crates/video-streamer/tests/support/mod.rs @@ -411,6 +411,7 @@ pub(crate) async fn spawn_stream_harness_delayed_start( encoder_threads: u32::try_from(encoder_threads).expect("thread count fits in u32"), adaptive_frame_skip: false, fill_interval: None, + fill_delay: Duration::ZERO, }, when_new_chunk_appended, ) diff --git a/devolutions-gateway/src/streaming.rs b/devolutions-gateway/src/streaming.rs index 54425cdef..b22ef9f6a 100644 --- a/devolutions-gateway/src/streaming.rs +++ b/devolutions-gateway/src/streaming.rs @@ -483,15 +483,21 @@ fn recording_event_stream( }) } -/// Experiment: `DGATEWAY_SHADOW_FILL_MS` sets the refill interval for quiet live clips. +/// Experiment: `DGATEWAY_SHADOW_FILL_MS` sets the refill interval for quiet live clips, and +/// `DGATEWAY_SHADOW_FILL_DELAY_MS` how far refills trail the source. fn session_config() -> SessionConfig { let fill_interval = std::env::var("DGATEWAY_SHADOW_FILL_MS") .ok() .and_then(|value| value.parse::().ok()) .filter(|&value| value > 0) .map(Duration::from_millis); + let fill_delay = std::env::var("DGATEWAY_SHADOW_FILL_DELAY_MS") + .ok() + .and_then(|value| value.parse::().ok()) + .map_or(Duration::ZERO, Duration::from_millis); SessionConfig { fill_interval, + fill_delay, ..SessionConfig::default() } }