diff --git a/crates/recording/src/output_pipeline/core.rs b/crates/recording/src/output_pipeline/core.rs index e28e3fc97fa..9045431f630 100644 --- a/crates/recording/src/output_pipeline/core.rs +++ b/crates/recording/src/output_pipeline/core.rs @@ -1991,6 +1991,11 @@ pub struct SharedPauseState { } impl SharedPauseState { + #[cfg(windows)] + pub(crate) fn flag(&self) -> Arc { + self.flag.clone() + } + pub fn new(flag: Arc) -> Self { Self { flag, diff --git a/crates/recording/src/output_pipeline/frame_pacing.rs b/crates/recording/src/output_pipeline/frame_pacing.rs new file mode 100644 index 00000000000..3dd9719bf85 --- /dev/null +++ b/crates/recording/src/output_pipeline/frame_pacing.rs @@ -0,0 +1,267 @@ +//! Repeats the last frame while a Windows video source is stalled, so the +//! encoder keeps a steady track, without taking the timestamp of a frame that +//! is only late. +//! +//! The Windows encoder loops used to repeat the last frame as soon as one +//! frame interval passed without a new one, stamped one interval after it: +//! the slot a slightly late real frame is stamped for. The real frame then +//! landed on the same timestamp and the muxer moved it by one tick, so a +//! camera with ordinary arrival jitter recorded duplicated frames followed by +//! frames shown for 11 µs (16 of 262 in real-capture run 39000141528). + +use std::{ + sync::{ + atomic::{AtomicBool, Ordering}, + mpsc::{Receiver, RecvTimeoutError}, + }, + time::{Duration, Instant}, +}; + +/// Frame intervals of silence before the last frame is repeated. A repeat is +/// stamped one interval after the previous frame and sent once a frame for +/// that slot would be two intervals late. +const STALL_INTERVALS: u32 = 3; + +pub(crate) enum Paced { + Frame(F, Duration), + Repeat(Duration), + End, +} + +pub(crate) struct FramePacer { + interval: Duration, + last: Option, + paused_since: Option, +} + +struct LastFrame { + timestamp: Duration, + arrived: Instant, + repeats: u32, +} + +impl FramePacer { + pub(crate) fn new(frame_rate: u32) -> Self { + Self { + interval: Duration::from_secs_f64(1.0 / f64::from(frame_rate.max(1))), + last: None, + paused_since: None, + } + } + + /// Blocks until a frame arrives, the source has been silent long enough + /// to repeat the last one, or the stream ends. Nothing repeats while + /// `paused`. Frame timestamps exclude paused time, so on resume the last + /// frame counts as having arrived that much later, and repeats continue + /// on its grid two intervals behind real time: a source that sends + /// nothing until the screen changes still gets the resumed span. Waits + /// are cut to half an interval so a pause is timed to within that. + pub(crate) fn next( + &mut self, + rx: &Receiver>, + paused: &AtomicBool, + ) -> Paced { + let slice = self.interval / 2; + loop { + let is_paused = paused.load(Ordering::Acquire); + match (is_paused, self.paused_since) { + (true, None) => self.paused_since = Some(Instant::now()), + (false, Some(since)) => { + if let Some(last) = self.last.as_mut() { + last.arrived += since.elapsed(); + } + self.paused_since = None; + } + _ => {} + } + let due = self + .last + .as_ref() + .map(|last| last.arrived + self.interval * (STALL_INTERVALS + last.repeats)); + let received = match due { + None => rx.recv().map_err(|_| RecvTimeoutError::Disconnected), + Some(_) if is_paused => rx.recv_timeout(slice), + Some(due) => { + rx.recv_timeout(due.saturating_duration_since(Instant::now()).min(slice)) + } + }; + match received { + Ok(Some((frame, timestamp))) => { + // The pause so far happened before this frame; if it + // continues, the next call times it from now. + self.paused_since = None; + self.last = Some(LastFrame { + timestamp, + arrived: Instant::now(), + repeats: 0, + }); + return Paced::Frame(frame, timestamp); + } + Ok(None) | Err(RecvTimeoutError::Disconnected) => return Paced::End, + Err(RecvTimeoutError::Timeout) => { + if is_paused || due.is_some_and(|due| Instant::now() < due) { + continue; + } + let Some(last) = self.last.as_mut() else { + continue; + }; + last.repeats += 1; + return Paced::Repeat(last.timestamp + self.interval * last.repeats); + } + } + } + } +} + +#[cfg(test)] +mod tests { + use super::*; + use std::sync::{Arc, mpsc::sync_channel}; + + // Long intervals so a loaded runner's oversleep (tens of ms) never + // reaches the three-interval stall in the jitter test. + const FPS: u32 = 5; + const INTERVAL: Duration = Duration::from_millis(200); + + fn at(n: u32) -> Duration { + INTERVAL * n + } + + #[test] + fn jittered_frames_pass_through_without_repeats() { + let (tx, rx) = sync_channel(64); + let sender = std::thread::spawn(move || { + for n in 0..10u32 { + // Frames arrive up to three quarters of an interval late. + std::thread::sleep(if n % 2 == 0 { + INTERVAL + INTERVAL * 3 / 4 + } else { + INTERVAL / 4 + }); + tx.send(Some((n, at(n)))).unwrap(); + } + tx.send(None).unwrap(); + }); + let mut pacer = FramePacer::new(FPS); + let paused = AtomicBool::new(false); + let mut seen = Vec::new(); + loop { + match pacer.next(&rx, &paused) { + Paced::Frame(n, ts) => seen.push((n, ts)), + Paced::Repeat(ts) => panic!("repeat at {ts:?} for a frame that was only late"), + Paced::End => break, + } + } + sender.join().unwrap(); + assert_eq!(seen, (0..10).map(|n| (n, at(n))).collect::>()); + } + + #[test] + fn a_stall_repeats_the_last_frame_on_its_grid_behind_real_time() { + let (tx, rx) = sync_channel(4); + let mut pacer = FramePacer::new(FPS); + let paused = AtomicBool::new(false); + tx.send(Some((7u32, at(10)))).unwrap(); + let start = Instant::now(); + assert!(matches!(pacer.next(&rx, &paused), Paced::Frame(7, ts) if ts == at(10))); + let Paced::Repeat(first) = pacer.next(&rx, &paused) else { + panic!("expected a repeat"); + }; + assert_eq!(first, at(11)); + assert!(start.elapsed() >= INTERVAL * STALL_INTERVALS); + let Paced::Repeat(second) = pacer.next(&rx, &paused) else { + panic!("expected a repeat"); + }; + assert_eq!(second, at(12)); + assert!(start.elapsed() >= INTERVAL * (STALL_INTERVALS + 1)); + // The source resumes with a frame captured after the stall. + tx.send(Some((8, at(16)))).unwrap(); + assert!(matches!(pacer.next(&rx, &paused), Paced::Frame(8, ts) if ts == at(16))); + } + + #[test] + fn nothing_repeats_while_paused() { + let (tx, rx) = sync_channel(4); + let mut pacer = FramePacer::new(FPS); + let paused = AtomicBool::new(false); + tx.send(Some((1u32, at(0)))).unwrap(); + assert!(matches!(pacer.next(&rx, &paused), Paced::Frame(1, _))); + paused.store(true, Ordering::Release); + let sender = std::thread::spawn(move || { + std::thread::sleep(INTERVAL * 8); + tx.send(Some((2, at(40)))).unwrap(); + tx.send(None).unwrap(); + }); + assert!(matches!(pacer.next(&rx, &paused), Paced::Frame(2, ts) if ts == at(40))); + assert!(matches!(pacer.next(&rx, &paused), Paced::End)); + sender.join().unwrap(); + } + + #[test] + fn repeats_resume_after_a_pause_on_the_last_frames_grid() { + let (tx, rx) = sync_channel(4); + let mut pacer = FramePacer::new(FPS); + let paused = Arc::new(AtomicBool::new(false)); + tx.send(Some((1u32, at(5)))).unwrap(); + assert!(matches!(pacer.next(&rx, &paused), Paced::Frame(1, _))); + let start = Instant::now(); + paused.store(true, Ordering::Release); + let resumer = { + let paused = paused.clone(); + std::thread::spawn(move || { + std::thread::sleep(INTERVAL * 6); + paused.store(false, Ordering::Release); + }) + }; + // A static screen: nothing arrives after the resume either. + let Paced::Repeat(first) = pacer.next(&rx, &paused) else { + panic!("expected a repeat after the resume"); + }; + assert_eq!(first, at(6)); + assert!(start.elapsed() >= INTERVAL * (6 + STALL_INTERVALS)); + let Paced::Repeat(second) = pacer.next(&rx, &paused) else { + panic!("expected a repeat"); + }; + assert_eq!(second, at(7)); + resumer.join().unwrap(); + drop(tx); + } + + #[test] + fn a_frame_that_arrives_while_paused_ends_the_pause_adjustment() { + let (tx, rx) = sync_channel(4); + let mut pacer = FramePacer::new(FPS); + let paused = AtomicBool::new(false); + tx.send(Some((1u32, at(0)))).unwrap(); + assert!(matches!(pacer.next(&rx, &paused), Paced::Frame(1, _))); + paused.store(true, Ordering::Release); + tx.send(Some((2, at(1)))).unwrap(); + assert!(matches!(pacer.next(&rx, &paused), Paced::Frame(2, _))); + let arrived = Instant::now(); + paused.store(false, Ordering::Release); + std::thread::sleep(INTERVAL * 6); + // Frame 2 has been up for six intervals: its first repeat is overdue. + let Paced::Repeat(first) = pacer.next(&rx, &paused) else { + panic!("expected a repeat"); + }; + assert_eq!(first, at(2)); + assert!(arrived.elapsed() < INTERVAL * 8); + drop(tx); + } + + #[test] + fn the_stream_ends_on_none_or_disconnect() { + let paused = AtomicBool::new(false); + let (tx, rx) = sync_channel::>(1); + tx.send(None).unwrap(); + assert!(matches!( + FramePacer::new(FPS).next(&rx, &paused), + Paced::End + )); + drop(tx); + assert!(matches!( + FramePacer::new(FPS).next(&rx, &paused), + Paced::End + )); + } +} diff --git a/crates/recording/src/output_pipeline/mod.rs b/crates/recording/src/output_pipeline/mod.rs index e8f069a70d2..d5223a33987 100644 --- a/crates/recording/src/output_pipeline/mod.rs +++ b/crates/recording/src/output_pipeline/mod.rs @@ -1,6 +1,8 @@ mod async_camera; mod core; pub mod ffmpeg; +#[cfg(any(windows, test))] +mod frame_pacing; #[cfg(target_os = "macos")] mod macos_fragmented_m4s; #[cfg(target_os = "macos")] diff --git a/crates/recording/src/output_pipeline/oop_fragmented_m4s_win.rs b/crates/recording/src/output_pipeline/oop_fragmented_m4s_win.rs index 4c67f24eb36..d90b19cf7c9 100644 --- a/crates/recording/src/output_pipeline/oop_fragmented_m4s_win.rs +++ b/crates/recording/src/output_pipeline/oop_fragmented_m4s_win.rs @@ -2,6 +2,7 @@ use super::core::{ BlockingThreadFinish, DiskSpaceMonitor, HealthSender, PipelineHealthEvent, SharedHealthSender, resolve_oop_thread_finish, wait_for_blocking_thread_finish, }; +use super::frame_pacing::{FramePacer, Paced}; use super::oop_muxer::{ MuxerSubprocessConfig, MuxerSubprocessError, RespawningMuxerSubprocess, VideoStreamInit, resolve_muxer_binary, @@ -258,6 +259,7 @@ impl WindowsOOPFragmentedM4SMuxer { let base_path = self.base_path.clone(); let video_config = self.video_config; + let pause_flag = self.pause.flag(); let segment_duration = self.segment_duration; let preset = self.preset; let bpp = self.bpp; @@ -363,7 +365,7 @@ impl WindowsOOPFragmentedM4SMuxer { return Err(err); } - let frame_interval = Duration::from_secs_f64(1.0 / video_config.fps() as f64); + let mut pacer = FramePacer::new(video_config.fps()); let mut last_ffmpeg_frame: Option = None; let mut last_timestamp: Option = None; let mut first_timestamp: Option = None; @@ -460,27 +462,30 @@ impl WindowsOOPFragmentedM4SMuxer { let convert_start = std::time::Instant::now(); - let (ffmpeg_frame, timestamp) = match video_rx.recv_timeout(frame_interval) { - Ok(Some((frame, ts))) => match frame.as_ffmpeg() { - Ok(f) => { - last_ffmpeg_frame = Some(reference_video_frame(&f)); - last_timestamp = Some(ts); - (Some(f), ts) - } - Err(e) => { - warn!("Windows OOP failed to convert D3D11 frame: {e:?}"); - match (&last_ffmpeg_frame, last_timestamp) { - (Some(f), Some(last_ts)) => { - let new_ts = last_ts.saturating_add(frame_interval); - last_timestamp = Some(new_ts); - duplicated_frames += 1; - (Some(reference_video_frame(f)), new_ts) - } - _ => (None, Duration::ZERO), + let (ffmpeg_frame, timestamp) = match pacer.next(&video_rx, &pause_flag) { + Paced::Frame(frame, ts) => { + last_timestamp = Some(ts); + match frame.as_ffmpeg() { + Ok(f) => { + last_ffmpeg_frame = Some(reference_video_frame(&f)); + (Some(f), ts) + } + Err(e) => { + warn!("Windows OOP failed to convert D3D11 frame: {e:?}"); + duplicated_frames += 1; + (last_ffmpeg_frame.as_ref().map(reference_video_frame), ts) } } - }, - Ok(None) | Err(RecvTimeoutError::Disconnected) => { + } + Paced::Repeat(ts) => { + let Some(f) = &last_ffmpeg_frame else { + continue; + }; + last_timestamp = Some(ts); + duplicated_frames += 1; + (Some(reference_video_frame(f)), ts) + } + Paced::End => { let drain_start = std::time::Instant::now(); let drain_timeout = Duration::from_millis(500); loop { @@ -516,17 +521,6 @@ impl WindowsOOPFragmentedM4SMuxer { } break; } - Err(RecvTimeoutError::Timeout) => { - match (&last_ffmpeg_frame, last_timestamp) { - (Some(f), Some(last_ts)) => { - let new_ts = last_ts.saturating_add(frame_interval); - last_timestamp = Some(new_ts); - duplicated_frames += 1; - (Some(reference_video_frame(f)), new_ts) - } - _ => continue, - } - } }; let convert_elapsed_ms = convert_start.elapsed().as_millis(); @@ -543,15 +537,6 @@ impl WindowsOOPFragmentedM4SMuxer { } let Some(ffmpeg_frame) = ffmpeg_frame else { - match video_rx.recv() { - Ok(Some((frame, ts))) => { - if let Ok(f) = frame.as_ffmpeg() { - last_ffmpeg_frame = Some(f); - last_timestamp = Some(ts); - } - } - Ok(None) | Err(_) => break, - } continue; }; diff --git a/crates/recording/src/output_pipeline/win.rs b/crates/recording/src/output_pipeline/win.rs index f291851c3b4..cf9003cf2d8 100644 --- a/crates/recording/src/output_pipeline/win.rs +++ b/crates/recording/src/output_pipeline/win.rs @@ -1,6 +1,9 @@ use crate::{ AudioFrame, AudioMuxer, Muxer, TaskPool, VideoFrame, VideoMuxer, - output_pipeline::{HealthSender, PipelineHealthEvent, emit_health}, + output_pipeline::{ + HealthSender, PipelineHealthEvent, emit_health, + frame_pacing::{FramePacer, Paced}, + }, screen_capture, }; use anyhow::{Context, anyhow}; @@ -12,8 +15,8 @@ use std::{ path::PathBuf, sync::{ Arc, Mutex, - atomic::{AtomicBool, Ordering}, - mpsc::{RecvTimeoutError, SyncSender, TrySendError, sync_channel}, + atomic::AtomicBool, + mpsc::{SyncSender, TrySendError, sync_channel}, }, time::Duration, }; @@ -337,70 +340,43 @@ impl Muxer for WindowsMuxer { match encoder { either::Left((mut encoder, mut muxer)) => { trace!("Running native encoder with frame pacing"); - let frame_interval = Duration::from_secs_f64(1.0 / config.frame_rate as f64); + let mut pacer = FramePacer::new(config.frame_rate); let mut last_texture: Option = None; let mut first_timestamp: Option = None; - let mut last_timestamp: Option = None; let mut frame_count: u64 = 0; let mut frames_reused: u64 = 0; let result = encoder.run( Arc::new(AtomicBool::default()), - || { - loop { - match video_rx.recv_timeout(frame_interval) { - Ok(Some((frame, timestamp))) => { - last_texture = Some(frame.texture().clone()); - last_timestamp = Some(timestamp); - } - Ok(None) => { - trace!("End of stream signal received"); - return Ok(None); - } - Err(RecvTimeoutError::Timeout) => { - if pause_flag.load(Ordering::Acquire) { - last_timestamp = None; - continue; - } else if let Some(last_ts) = last_timestamp { - let new_ts = last_ts.saturating_add(frame_interval); - last_timestamp = Some(new_ts); - frames_reused += 1; - if frames_reused.is_multiple_of(30) { - debug!( - frames_reused = frames_reused, - frame_count = frame_count, - "Frame pacing: reusing frames due to slow capture" - ); - } - } - } - Err(RecvTimeoutError::Disconnected) => { - trace!("Channel disconnected"); - return Ok(None); - } + || loop { + let (texture, timestamp) = match pacer.next(&video_rx, &pause_flag) { + Paced::Frame(frame, timestamp) => { + let texture = frame.texture().clone(); + last_texture = Some(texture.clone()); + (texture, timestamp) } - - if let (Some(texture), Some(ts)) = (&last_texture, last_timestamp) { - let normalized_ts = normalize_timestamp(ts, &mut first_timestamp); - frame_count += 1; - let frame_time = duration_to_timespan(normalized_ts); - return Ok(Some((texture.clone(), frame_time))); - } else { - match video_rx.recv() { - Ok(Some((frame, timestamp))) => { - let texture = frame.texture().clone(); - last_texture = Some(texture.clone()); - last_timestamp = Some(timestamp); - let normalized_ts = - normalize_timestamp(timestamp, &mut first_timestamp); - frame_count = 1; - let frame_time = duration_to_timespan(normalized_ts); - return Ok(Some((texture, frame_time))); - } - Ok(None) | Err(_) => return Ok(None), + Paced::Repeat(timestamp) => { + let Some(texture) = &last_texture else { + continue; + }; + frames_reused += 1; + if frames_reused.is_multiple_of(30) { + debug!( + frames_reused = frames_reused, + frame_count = frame_count, + "Frame pacing: reusing frames due to slow capture" + ); } + (texture.clone(), timestamp) } - } + Paced::End => { + trace!("End of stream signal received"); + return Ok(None); + } + }; + let normalized_ts = normalize_timestamp(timestamp, &mut first_timestamp); + frame_count += 1; + return Ok(Some((texture, duration_to_timespan(normalized_ts)))); }, |output_sample| { let Ok(mut output) = output.lock() else { @@ -438,63 +414,33 @@ impl Muxer for WindowsMuxer { } either::Right(mut encoder) => { trace!("Running FFmpeg encoder with frame pacing"); - let frame_interval = Duration::from_secs_f64(1.0 / config.frame_rate as f64); + let mut pacer = FramePacer::new(config.frame_rate); let mut last_ffmpeg_frame: Option = None; let mut first_timestamp: Option = None; - let mut last_timestamp: Option = None; loop { - let (ffmpeg_frame, ts) = match video_rx.recv_timeout(frame_interval) { - Ok(Some((frame, timestamp))) => { - last_timestamp = Some(timestamp); - match frame.as_ffmpeg() { - Ok(f) => { - last_ffmpeg_frame = Some(reference_video_frame(&f)); - (Some(f), timestamp) - } - Err(e) => { - warn!("Failed to convert frame: {e:?}"); - ( - last_ffmpeg_frame - .as_ref() - .map(reference_video_frame), - timestamp, - ) - } - } - } - Ok(None) => break, - Err(RecvTimeoutError::Timeout) => { - if pause_flag.load(Ordering::Acquire) { - last_timestamp = None; - continue; + let (ffmpeg_frame, ts) = match pacer.next(&video_rx, &pause_flag) { + Paced::Frame(frame, timestamp) => match frame.as_ffmpeg() { + Ok(f) => { + last_ffmpeg_frame = Some(reference_video_frame(&f)); + (Some(f), timestamp) } - if let Some(last_ts) = last_timestamp { - let new_ts = last_ts.saturating_add(frame_interval); - last_timestamp = Some(new_ts); + Err(e) => { + warn!("Failed to convert frame: {e:?}"); ( - last_ffmpeg_frame - .as_ref() - .map(reference_video_frame), - new_ts, + last_ffmpeg_frame.as_ref().map(reference_video_frame), + timestamp, ) - } else { - continue; } - } - Err(RecvTimeoutError::Disconnected) => break, + }, + Paced::Repeat(timestamp) => ( + last_ffmpeg_frame.as_ref().map(reference_video_frame), + timestamp, + ), + Paced::End => break, }; let Some(ffmpeg_frame) = ffmpeg_frame else { - match video_rx.recv() { - Ok(Some((frame, timestamp))) => { - last_timestamp = Some(timestamp); - if let Ok(f) = frame.as_ffmpeg() { - last_ffmpeg_frame = Some(f); - } - } - Ok(None) | Err(_) => break, - } continue; }; @@ -927,75 +873,38 @@ impl Muxer for WindowsCameraMuxer { frame_rate ); - let frame_interval = Duration::from_secs_f64(1.0 / frame_rate as f64); + let mut pacer = FramePacer::new(frame_rate); let mut last_frame: Option = None; let mut first_timestamp: Option = None; - let mut last_timestamp: Option = None; let mut frame_count = 0u64; let mut camera_buffers = CameraBuffers::new(); let result = encoder.run( Arc::new(AtomicBool::default()), - || { - loop { - match video_rx.recv_timeout(frame_interval) { - Ok(Some((frame, timestamp))) => { - last_frame = Some(frame); - last_timestamp = Some(timestamp); - } - Ok(None) => { - trace!("End of camera stream signal received"); - return Ok(None); - } - Err(RecvTimeoutError::Timeout) => { - if pause_flag.load(Ordering::Acquire) { - last_timestamp = None; - continue; - } else if let Some(last_ts) = last_timestamp { - let new_ts = last_ts.saturating_add(frame_interval); - last_timestamp = Some(new_ts); - } - } - Err(RecvTimeoutError::Disconnected) => { - trace!("Camera channel disconnected"); - return Ok(None); - } + || loop { + let timestamp = match pacer.next(&video_rx, &pause_flag) { + Paced::Frame(frame, timestamp) => { + last_frame = Some(frame); + timestamp } - - if let (Some(frame), Some(ts)) = (&last_frame, last_timestamp) { - let normalized_ts = - normalize_camera_timestamp(ts, &mut first_timestamp); - frame_count += 1; - if frame_count.is_multiple_of(30) { - debug!( - "Windows camera encoder: processed {} frames", - frame_count - ); - } - let texture = - upload_mf_buffer_to_texture(&d3d_device, frame, &mut camera_buffers)?; - return Ok(Some((texture, duration_to_timespan(normalized_ts)))); - } else { - match video_rx.recv() { - Ok(Some((frame, timestamp))) => { - last_frame = Some(frame.clone()); - last_timestamp = Some(timestamp); - let normalized_ts = normalize_camera_timestamp( - timestamp, - &mut first_timestamp, - ); - frame_count = 1; - let texture = upload_mf_buffer_to_texture( - &d3d_device, - &frame, - &mut camera_buffers, - )?; - return Ok(Some((texture, duration_to_timespan(normalized_ts)))); - } - Ok(None) | Err(_) => return Ok(None), - } + Paced::Repeat(timestamp) => timestamp, + Paced::End => { + trace!("End of camera stream signal received"); + return Ok(None); } + }; + let Some(frame) = &last_frame else { + continue; + }; + let normalized_ts = + normalize_camera_timestamp(timestamp, &mut first_timestamp); + frame_count += 1; + if frame_count.is_multiple_of(30) { + debug!("Windows camera encoder: processed {} frames", frame_count); } + let texture = + upload_mf_buffer_to_texture(&d3d_device, frame, &mut camera_buffers)?; + return Ok(Some((texture, duration_to_timespan(normalized_ts)))); }, |output_sample| { let mut output = match output.lock() { @@ -1047,44 +956,21 @@ impl Muxer for WindowsCameraMuxer { frame_rate ); - let frame_interval = Duration::from_secs_f64(1.0 / frame_rate as f64); + let mut pacer = FramePacer::new(frame_rate); let mut last_frame: Option = None; let mut first_timestamp: Option = None; - let mut last_timestamp: Option = None; let mut frame_count = 0u64; loop { - let (frame_to_encode, ts) = match video_rx.recv_timeout(frame_interval) { - Ok(Some((frame, timestamp))) => { - last_frame = Some(frame.clone()); - last_timestamp = Some(timestamp); - (Some(frame), timestamp) + let ts = match pacer.next(&video_rx, &pause_flag) { + Paced::Frame(frame, timestamp) => { + last_frame = Some(frame); + timestamp } - Ok(None) => break, - Err(RecvTimeoutError::Timeout) => { - if pause_flag.load(Ordering::Acquire) { - last_timestamp = None; - continue; - } - if let Some(last_ts) = last_timestamp { - let new_ts = last_ts.saturating_add(frame_interval); - last_timestamp = Some(new_ts); - (last_frame.clone(), new_ts) - } else { - continue; - } - } - Err(RecvTimeoutError::Disconnected) => break, + Paced::Repeat(timestamp) => timestamp, + Paced::End => break, }; - - let Some(frame) = frame_to_encode else { - match video_rx.recv() { - Ok(Some((frame, timestamp))) => { - last_frame = Some(frame.clone()); - last_timestamp = Some(timestamp); - } - Ok(None) | Err(_) => break, - } + let Some(frame) = last_frame.clone() else { continue; }; diff --git a/crates/recording/src/output_pipeline/win_fragmented_m4s.rs b/crates/recording/src/output_pipeline/win_fragmented_m4s.rs index 303c82ec61a..67278a91340 100644 --- a/crates/recording/src/output_pipeline/win_fragmented_m4s.rs +++ b/crates/recording/src/output_pipeline/win_fragmented_m4s.rs @@ -2,6 +2,7 @@ use super::core::{ BlockingThreadFinish, DiskSpaceMonitor, HealthSender, PipelineHealthEvent, SharedHealthSender, combine_finish_errors, wait_for_blocking_thread_finish, }; +use super::frame_pacing::{FramePacer, Paced}; use super::win::reference_video_frame; use crate::{ AudioFrame, AudioMuxer, Muxer, SharedPauseState, TaskPool, VideoMuxer, @@ -359,6 +360,7 @@ impl WindowsFragmentedM4SMuxer { let video_config = self.video_config; let health_tx = self.health_tx.clone(); let base_path = self.base_path.clone(); + let pause_flag = self.pause.flag(); let encoder_handle = std::thread::Builder::new() .name("win-m4s-segment-encoder".to_string()) .spawn(move || { @@ -368,9 +370,8 @@ impl WindowsFragmentedM4SMuxer { return Err(anyhow!("Failed to send ready signal - receiver dropped")); } - let frame_interval = Duration::from_secs_f64(1.0 / video_config.fps() as f64); + let mut pacer = FramePacer::new(video_config.fps()); let mut last_ffmpeg_frame: Option = None; - let mut last_timestamp: Option = None; let mut first_timestamp: Option = None; let mut slow_convert_count = 0u32; @@ -444,27 +445,26 @@ impl WindowsFragmentedM4SMuxer { let convert_start = std::time::Instant::now(); - let (ffmpeg_frame, timestamp) = match video_rx.recv_timeout(frame_interval) { - Ok(Some((frame, ts))) => match frame.as_ffmpeg() { + let (ffmpeg_frame, timestamp) = match pacer.next(&video_rx, &pause_flag) { + Paced::Frame(frame, ts) => match frame.as_ffmpeg() { Ok(f) => { last_ffmpeg_frame = Some(reference_video_frame(&f)); - last_timestamp = Some(ts); (Some(f), ts) } Err(e) => { warn!("Failed to convert D3D11 frame: {e:?}"); - match (&last_ffmpeg_frame, last_timestamp) { - (Some(f), Some(last_ts)) => { - let new_ts = last_ts.saturating_add(frame_interval); - last_timestamp = Some(new_ts); - duplicated_frames += 1; - (Some(reference_video_frame(f)), new_ts) - } - _ => (None, Duration::ZERO), - } + duplicated_frames += 1; + (last_ffmpeg_frame.as_ref().map(reference_video_frame), ts) } }, - Ok(None) | Err(RecvTimeoutError::Disconnected) => { + Paced::Repeat(ts) => { + let Some(f) = &last_ffmpeg_frame else { + continue; + }; + duplicated_frames += 1; + (Some(reference_video_frame(f)), ts) + } + Paced::End => { let mut drained = 0u64; let drain_start = std::time::Instant::now(); let drain_timeout = Duration::from_millis(500); @@ -503,31 +503,11 @@ impl WindowsFragmentedM4SMuxer { } break; } - Err(RecvTimeoutError::Timeout) => { - match (&last_ffmpeg_frame, last_timestamp) { - (Some(f), Some(last_ts)) => { - let new_ts = last_ts.saturating_add(frame_interval); - last_timestamp = Some(new_ts); - duplicated_frames += 1; - (Some(reference_video_frame(f)), new_ts) - } - _ => continue, - } - } }; let convert_elapsed_ms = convert_start.elapsed().as_millis(); let Some(ffmpeg_frame) = ffmpeg_frame else { - match video_rx.recv() { - Ok(Some((frame, ts))) => { - if let Ok(f) = frame.as_ffmpeg() { - last_ffmpeg_frame = Some(f); - last_timestamp = Some(ts); - } - } - Ok(None) | Err(_) => break, - } continue; }; @@ -828,6 +808,7 @@ impl WindowsFragmentedM4SCameraMuxer { let video_config = self.video_config; let health_tx = self.health_tx.clone(); let base_path = self.base_path.clone(); + let pause_flag = self.pause.flag(); let encoder_handle = std::thread::Builder::new() .name("win-m4s-camera-segment-encoder".to_string()) .spawn(move || { @@ -839,9 +820,8 @@ impl WindowsFragmentedM4SCameraMuxer { )); } - let frame_interval = Duration::from_secs_f64(1.0 / video_config.fps() as f64); + let mut pacer = FramePacer::new(video_config.fps()); let mut last_frame: Option = None; - let mut last_timestamp: Option = None; let mut first_timestamp: Option = None; let mut slow_convert_count = 0u32; @@ -940,13 +920,19 @@ impl WindowsFragmentedM4SCameraMuxer { break; } - let (frame_to_encode, timestamp) = match video_rx.recv_timeout(frame_interval) { - Ok(Some((frame, ts))) => { + let (camera_frame, timestamp) = match pacer.next(&video_rx, &pause_flag) { + Paced::Frame(frame, ts) => { last_frame = Some(frame.clone()); - last_timestamp = Some(ts); - (Some(frame), ts) + (frame, ts) } - Ok(None) | Err(RecvTimeoutError::Disconnected) => { + Paced::Repeat(ts) => { + let Some(frame) = &last_frame else { + continue; + }; + duplicated_frames += 1; + (frame.clone(), ts) + } + Paced::End => { let mut drained = 0u64; let drain_start = std::time::Instant::now(); let drain_timeout = Duration::from_millis(500); @@ -987,28 +973,6 @@ impl WindowsFragmentedM4SCameraMuxer { } break; } - Err(RecvTimeoutError::Timeout) => { - match (&last_frame, last_timestamp) { - (Some(f), Some(last_ts)) => { - let new_ts = last_ts.saturating_add(frame_interval); - last_timestamp = Some(new_ts); - duplicated_frames += 1; - (Some(f.clone()), new_ts) - } - _ => continue, - } - } - }; - - let Some(camera_frame) = frame_to_encode else { - match video_rx.recv() { - Ok(Some((frame, ts))) => { - last_frame = Some(frame); - last_timestamp = Some(ts); - } - Ok(None) | Err(_) => break, - } - continue; }; let normalized_ts = normalize_timestamp(timestamp, &mut first_timestamp);