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 46aa90a0f4b..a557083e96c 100644 --- a/crates/recording/src/output_pipeline/oop_fragmented_m4s_win.rs +++ b/crates/recording/src/output_pipeline/oop_fragmented_m4s_win.rs @@ -20,7 +20,7 @@ use std::{ sync::{ Arc, atomic::AtomicBool, - mpsc::{RecvTimeoutError, SyncSender, sync_channel}, + mpsc::{RecvTimeoutError, SyncSender, TryRecvError, TrySendError, sync_channel}, }, thread::JoinHandle, time::Duration, @@ -28,7 +28,7 @@ use std::{ use tracing::*; const DEFAULT_MAX_RESPAWNS: u32 = 3; -const DEFAULT_MUXER_BUFFER_SIZE: usize = 240; +const DEFAULT_MUXER_BUFFER_SIZE: usize = 8; fn get_muxer_buffer_size() -> usize { std::env::var("CAP_MUXER_BUFFER_SIZE") @@ -37,6 +37,49 @@ fn get_muxer_buffer_size() -> usize { .unwrap_or(DEFAULT_MUXER_BUFFER_SIZE) } +type ScreenFrameItem = Option<(screen_capture::ScreenFrame, Duration)>; + +struct LatestScreenFrame { + frame: screen_capture::ScreenFrame, + timestamp: Duration, + stale_frames: u64, + end_after_frame: bool, +} + +fn recv_latest_screen_frame( + video_rx: &std::sync::mpsc::Receiver, + timeout: Duration, +) -> Result, RecvTimeoutError> { + let Some((mut frame, mut timestamp)) = video_rx.recv_timeout(timeout)? else { + return Ok(None); + }; + + let mut stale_frames = 0; + let mut end_after_frame = false; + loop { + match video_rx.try_recv() { + Ok(Some((next_frame, next_timestamp))) => { + frame = next_frame; + timestamp = next_timestamp; + stale_frames += 1; + } + Ok(None) => { + end_after_frame = true; + break; + } + Err(TryRecvError::Empty) => break, + Err(TryRecvError::Disconnected) => return Err(RecvTimeoutError::Disconnected), + } + } + + Ok(Some(LatestScreenFrame { + frame, + timestamp, + stale_frames, + end_after_frame, + })) +} + struct FrameDropTracker { drops_in_window: u32, frames_in_window: u32, @@ -107,8 +150,6 @@ impl FrameDropTracker { } } -type ScreenFrameItem = Option<(screen_capture::ScreenFrame, Duration)>; - struct EncoderState { video_tx: SyncSender, encoder_handle: Option>>, @@ -371,6 +412,8 @@ impl WindowsOOPFragmentedM4SMuxer { let mut first_timestamp: Option = None; let mut total_frames = 0u64; let mut duplicated_frames = 0u64; + let mut stale_frames_dropped = 0u64; + let mut pending_end_after_frame = false; let mut slow_convert_count = 0u32; let mut slow_encode_count = 0u32; const SLOW_THRESHOLD_MS: u128 = 5; @@ -446,6 +489,10 @@ impl WindowsOOPFragmentedM4SMuxer { }; loop { + if pending_end_after_frame { + break; + } + match disk_monitor.poll(&base_path, &health_tx) { super::core::DiskSpacePollResult::Exhausted { .. } | super::core::DiskSpacePollResult::Stopped => { @@ -462,26 +509,39 @@ 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(f.clone()); - last_timestamp = Some(ts); - (Some(f), ts) + let (ffmpeg_frame, timestamp) = match recv_latest_screen_frame(&video_rx, frame_interval) { + Ok(Some(latest)) => { + if latest.stale_frames > 0 { + stale_frames_dropped += latest.stale_frames; + if stale_frames_dropped <= 5 || stale_frames_dropped.is_multiple_of(120) { + debug!( + stale_frames = latest.stale_frames, + total_stale_frames = stale_frames_dropped, + "Dropped stale queued screen frames before OOP encoding latest" + ); + } } - 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(f.clone()), new_ts) + pending_end_after_frame = latest.end_after_frame; + match latest.frame.as_ffmpeg() { + Ok(f) => { + last_ffmpeg_frame = Some(f.clone()); + last_timestamp = Some(latest.timestamp); + (Some(f), latest.timestamp) + } + 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(f.clone()), new_ts) + } + _ => (None, Duration::ZERO), } - _ => (None, Duration::ZERO), } } - }, + } Ok(None) | Err(RecvTimeoutError::Disconnected) => { let drain_start = std::time::Instant::now(); let drain_timeout = Duration::from_millis(500); @@ -593,6 +653,7 @@ impl WindowsOOPFragmentedM4SMuxer { debug!( total_frames = total_frames, duplicated_frames = duplicated_frames, + stale_frames_dropped = stale_frames_dropped, slow_converts = slow_convert_count, slow_encodes = slow_encode_count, slow_convert_pct = format!( @@ -695,11 +756,17 @@ impl VideoMuxer for WindowsOOPFragmentedM4SMuxer { }; if let Some(state) = &self.state { - match state.video_tx.send(Some((frame.frame, adjusted_timestamp))) { + match state + .video_tx + .try_send(Some((frame.frame, adjusted_timestamp))) + { Ok(()) => { self.frame_drops.record_frame(); } - Err(_) => { + Err(TrySendError::Full(_)) => { + self.frame_drops.record_drop(); + } + Err(TrySendError::Disconnected(_)) => { return Err(anyhow!("Windows OOP M4S encoder channel disconnected")); } } diff --git a/crates/recording/src/output_pipeline/win.rs b/crates/recording/src/output_pipeline/win.rs index d7c7831c98b..9845695154c 100644 --- a/crates/recording/src/output_pipeline/win.rs +++ b/crates/recording/src/output_pipeline/win.rs @@ -13,13 +13,13 @@ use std::{ sync::{ Arc, Mutex, atomic::{AtomicBool, Ordering}, - mpsc::{RecvTimeoutError, SyncSender, TrySendError, sync_channel}, + mpsc::{RecvTimeoutError, SyncSender, TryRecvError, TrySendError, sync_channel}, }, time::Duration, }; use tracing::*; -const DEFAULT_MUXER_BUFFER_SIZE: usize = 240; +const DEFAULT_MUXER_BUFFER_SIZE: usize = 8; fn get_muxer_buffer_size() -> usize { std::env::var("CAP_MUXER_BUFFER_SIZE") @@ -28,6 +28,49 @@ fn get_muxer_buffer_size() -> usize { .unwrap_or(DEFAULT_MUXER_BUFFER_SIZE) } +type ScreenFrameItem = Option<(screen_capture::ScreenFrame, Duration)>; + +struct LatestScreenFrame { + frame: screen_capture::ScreenFrame, + timestamp: Duration, + stale_frames: u64, + end_after_frame: bool, +} + +fn recv_latest_screen_frame( + video_rx: &std::sync::mpsc::Receiver, + timeout: Duration, +) -> Result, RecvTimeoutError> { + let Some((mut frame, mut timestamp)) = video_rx.recv_timeout(timeout)? else { + return Ok(None); + }; + + let mut stale_frames = 0; + let mut end_after_frame = false; + loop { + match video_rx.try_recv() { + Ok(Some((next_frame, next_timestamp))) => { + frame = next_frame; + timestamp = next_timestamp; + stale_frames += 1; + } + Ok(None) => { + end_after_frame = true; + break; + } + Err(TryRecvError::Empty) => break, + Err(TryRecvError::Disconnected) => return Err(RecvTimeoutError::Disconnected), + } + } + + Ok(Some(LatestScreenFrame { + frame, + timestamp, + stale_frames, + end_after_frame, + })) +} + struct FrameDropTracker { drops_in_window: u32, frames_in_window: u32, @@ -327,15 +370,28 @@ impl Muxer for WindowsMuxer { let mut last_timestamp: Option = None; let mut frame_count: u64 = 0; let mut frames_reused: u64 = 0; + let mut pending_end_after_frame = false; let result = encoder.run( Arc::new(AtomicBool::default()), || { + if pending_end_after_frame { + trace!("End of stream signal received after latest queued frame"); + return Ok(None); + } + loop { - match video_rx.recv_timeout(frame_interval) { - Ok(Some((frame, timestamp))) => { - last_texture = Some(frame.texture().clone()); - last_timestamp = Some(timestamp); + match recv_latest_screen_frame(&video_rx, frame_interval) { + Ok(Some(latest)) => { + if latest.stale_frames > 0 { + debug!( + stale_frames = latest.stale_frames, + "Dropped stale queued screen frames before encoding latest" + ); + } + pending_end_after_frame = latest.end_after_frame; + last_texture = Some(latest.frame.texture().clone()); + last_timestamp = Some(latest.timestamp); } Ok(None) => { trace!("End of stream signal received"); @@ -426,19 +482,31 @@ impl Muxer for WindowsMuxer { let mut last_ffmpeg_frame: Option = None; let mut first_timestamp: Option = None; let mut last_timestamp: Option = None; + let mut pending_end_after_frame = false; loop { - let (ffmpeg_frame, ts) = match video_rx.recv_timeout(frame_interval) { - Ok(Some((frame, timestamp))) => { - last_timestamp = Some(timestamp); - match frame.as_ffmpeg() { + if pending_end_after_frame { + break; + } + + let (ffmpeg_frame, ts) = match recv_latest_screen_frame(&video_rx, frame_interval) { + Ok(Some(latest)) => { + if latest.stale_frames > 0 { + debug!( + stale_frames = latest.stale_frames, + "Dropped stale queued screen frames before encoding latest" + ); + } + pending_end_after_frame = latest.end_after_frame; + last_timestamp = Some(latest.timestamp); + match latest.frame.as_ffmpeg() { Ok(f) => { last_ffmpeg_frame = Some(f.clone()); - (Some(f), timestamp) + (Some(f), latest.timestamp) } Err(e) => { warn!("Failed to convert frame: {e:?}"); - (last_ffmpeg_frame.clone(), timestamp) + (last_ffmpeg_frame.clone(), latest.timestamp) } } } diff --git a/crates/recording/src/output_pipeline/win_fragmented_m4s.rs b/crates/recording/src/output_pipeline/win_fragmented_m4s.rs index 131e7843788..83e484561e1 100644 --- a/crates/recording/src/output_pipeline/win_fragmented_m4s.rs +++ b/crates/recording/src/output_pipeline/win_fragmented_m4s.rs @@ -18,14 +18,14 @@ use std::{ sync::{ Arc, Mutex, atomic::AtomicBool, - mpsc::{RecvTimeoutError, SyncSender, sync_channel}, + mpsc::{RecvTimeoutError, SyncSender, TryRecvError, TrySendError, sync_channel}, }, thread::JoinHandle, time::Duration, }; use tracing::*; -const DEFAULT_MUXER_BUFFER_SIZE: usize = 240; +const DEFAULT_MUXER_BUFFER_SIZE: usize = 8; fn get_muxer_buffer_size() -> usize { std::env::var("CAP_MUXER_BUFFER_SIZE") @@ -34,6 +34,49 @@ fn get_muxer_buffer_size() -> usize { .unwrap_or(DEFAULT_MUXER_BUFFER_SIZE) } +type ScreenFrameItem = Option<(screen_capture::ScreenFrame, Duration)>; + +struct LatestScreenFrame { + frame: screen_capture::ScreenFrame, + timestamp: Duration, + stale_frames: u64, + end_after_frame: bool, +} + +fn recv_latest_screen_frame( + video_rx: &std::sync::mpsc::Receiver, + timeout: Duration, +) -> Result, RecvTimeoutError> { + let Some((mut frame, mut timestamp)) = video_rx.recv_timeout(timeout)? else { + return Ok(None); + }; + + let mut stale_frames = 0; + let mut end_after_frame = false; + loop { + match video_rx.try_recv() { + Ok(Some((next_frame, next_timestamp))) => { + frame = next_frame; + timestamp = next_timestamp; + stale_frames += 1; + } + Ok(None) => { + end_after_frame = true; + break; + } + Err(TryRecvError::Empty) => break, + Err(TryRecvError::Disconnected) => return Err(RecvTimeoutError::Disconnected), + } + } + + Ok(Some(LatestScreenFrame { + frame, + timestamp, + stale_frames, + end_after_frame, + })) +} + struct FrameDropTracker { drops_in_window: u32, frames_in_window: u32, @@ -367,6 +410,8 @@ impl WindowsFragmentedM4SMuxer { let mut slow_encode_count = 0u32; let mut total_frames = 0u64; let mut duplicated_frames = 0u64; + let mut stale_frames_dropped = 0u64; + let mut pending_end_after_frame = false; let mut disk_monitor = DiskSpaceMonitor::new(); let mut disk_exhausted = false; const SLOW_THRESHOLD_MS: u128 = 5; @@ -421,6 +466,10 @@ impl WindowsFragmentedM4SMuxer { }; loop { + if pending_end_after_frame { + break; + } + if matches!( disk_monitor.poll(&base_path, &health_tx), super::core::DiskSpacePollResult::Exhausted { .. } @@ -434,26 +483,44 @@ 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() { - Ok(f) => { - last_ffmpeg_frame = Some(f.clone()); - last_timestamp = Some(ts); - (Some(f), ts) + let (ffmpeg_frame, timestamp) = match recv_latest_screen_frame( + &video_rx, + frame_interval, + ) { + Ok(Some(latest)) => { + if latest.stale_frames > 0 { + stale_frames_dropped += latest.stale_frames; + if stale_frames_dropped <= 5 + || stale_frames_dropped.is_multiple_of(120) + { + debug!( + stale_frames = latest.stale_frames, + total_stale_frames = stale_frames_dropped, + "Dropped stale queued screen frames before encoding latest" + ); + } } - 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(f.clone()), new_ts) + pending_end_after_frame = latest.end_after_frame; + match latest.frame.as_ffmpeg() { + Ok(f) => { + last_ffmpeg_frame = Some(f.clone()); + last_timestamp = Some(latest.timestamp); + (Some(f), latest.timestamp) + } + 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(f.clone()), new_ts) + } + _ => (None, Duration::ZERO), } - _ => (None, Duration::ZERO), } } - }, + } Ok(None) | Err(RecvTimeoutError::Disconnected) => { let mut drained = 0u64; let drain_start = std::time::Instant::now(); @@ -550,6 +617,7 @@ impl WindowsFragmentedM4SMuxer { debug!( total_frames = total_frames, duplicated_frames = duplicated_frames, + stale_frames_dropped = stale_frames_dropped, slow_converts = slow_convert_count, slow_encodes = slow_encode_count, slow_convert_pct = format!( @@ -601,11 +669,17 @@ impl VideoMuxer for WindowsFragmentedM4SMuxer { }; if let Some(state) = &self.state { - match state.video_tx.send(Some((frame.frame, adjusted_timestamp))) { + match state + .video_tx + .try_send(Some((frame.frame, adjusted_timestamp))) + { Ok(()) => { self.frame_drops.record_frame(); } - Err(_) => { + Err(TrySendError::Full(_)) => { + self.frame_drops.record_drop(); + } + Err(TrySendError::Disconnected(_)) => { return Err(anyhow!("Windows M4S encoder channel disconnected")); } } diff --git a/crates/recording/src/output_pipeline/win_segmented.rs b/crates/recording/src/output_pipeline/win_segmented.rs index 1edaf97daf1..d08651fa3b7 100644 --- a/crates/recording/src/output_pipeline/win_segmented.rs +++ b/crates/recording/src/output_pipeline/win_segmented.rs @@ -8,7 +8,7 @@ use std::{ sync::{ Arc, Mutex, atomic::{AtomicBool, Ordering}, - mpsc::{SyncSender, sync_channel}, + mpsc::{RecvError, SyncSender, TryRecvError, sync_channel}, }, thread::JoinHandle, time::Duration, @@ -55,6 +55,46 @@ struct SegmentState { encoder_handle: Option>>, } +struct LatestScreenFrame { + frame: screen_capture::ScreenFrame, + timestamp: Duration, + stale_frames: u64, + end_after_frame: bool, +} + +fn recv_latest_screen_frame( + video_rx: &std::sync::mpsc::Receiver>, +) -> Result, RecvError> { + let Some((mut frame, mut timestamp)) = video_rx.recv()? else { + return Ok(None); + }; + + let mut stale_frames = 0; + let mut end_after_frame = false; + loop { + match video_rx.try_recv() { + Ok(Some((next_frame, next_timestamp))) => { + frame = next_frame; + timestamp = next_timestamp; + stale_frames += 1; + } + Ok(None) => { + end_after_frame = true; + break; + } + Err(TryRecvError::Empty) => break, + Err(TryRecvError::Disconnected) => return Err(RecvError), + } + } + + Ok(Some(LatestScreenFrame { + frame, + timestamp, + stale_frames, + end_after_frame, + })) +} + struct PauseTracker { flag: Arc, paused_at: Option, @@ -531,23 +571,36 @@ impl WindowsSegmentedMuxer { either::Left((mut encoder, mut muxer)) => { trace!("Running native encoder for segment"); let mut first_timestamp: Option = None; + let mut pending_end_after_frame = false; let result = encoder.run( Arc::new(AtomicBool::default()), || { - let Ok(Some((frame, timestamp))) = video_rx.recv() else { + if pending_end_after_frame { + return Ok(None); + } + + let Ok(Some(latest)) = recv_latest_screen_frame(&video_rx) else { trace!("No more frames available for segment"); return Ok(None); }; + if latest.stale_frames > 0 { + debug!( + stale_frames = latest.stale_frames, + "Dropped stale queued segment screen frames before encoding latest" + ); + } + pending_end_after_frame = latest.end_after_frame; + let relative = if let Some(first) = first_timestamp { - timestamp.checked_sub(first).unwrap_or(Duration::ZERO) + latest.timestamp.checked_sub(first).unwrap_or(Duration::ZERO) } else { - first_timestamp = Some(timestamp); + first_timestamp = Some(latest.timestamp); Duration::ZERO }; let frame_time = duration_to_timespan(relative); - Ok(Some((frame.texture().clone(), frame_time))) + Ok(Some((latest.frame.texture().clone(), frame_time))) }, |output_sample| { let mut output = match output_clone.lock() { @@ -590,19 +643,29 @@ impl WindowsSegmentedMuxer { } } either::Right(mut encoder) => { - while let Ok(Some((frame, time))) = video_rx.recv() { + while let Ok(Some(latest)) = recv_latest_screen_frame(&video_rx) { + if latest.stale_frames > 0 { + debug!( + stale_frames = latest.stale_frames, + "Dropped stale queued segment screen frames before encoding latest" + ); + } let Ok(mut output) = output_clone.lock() else { continue; }; - frame + latest + .frame .as_ffmpeg() .context("frame as_ffmpeg") .and_then(|frame| { encoder - .queue_frame(frame, time, &mut output) + .queue_frame(frame, latest.timestamp, &mut output) .context("queue_frame") })?; + if latest.end_after_frame { + break; + } } Ok(()) diff --git a/crates/rendering/src/lib.rs b/crates/rendering/src/lib.rs index 33fed1e170f..8c568059134 100644 --- a/crates/rendering/src/lib.rs +++ b/crates/rendering/src/lib.rs @@ -280,8 +280,18 @@ impl RecordingSegmentDecoders { camera: camera_path, } = segment; + // Windows studio recordings write video packets with recording-epoch timestamps, + // so the finalized tracks already share one timeline that starts at the epoch. + // Applying `latest_start_time` on top (which assumes first-frame-rebased tracks, + // as produced on macOS) double-counts the capture startup delay and makes the + // rendered cursor run ahead of the screen content by that delay. + let recorded_on_windows = matches!( + _recording_meta.platform, + Some(cap_project::Platform::Windows) + ); let latest_start_time = match &meta { StudioRecordingMeta::SingleSegment { .. } => None, + StudioRecordingMeta::MultipleSegments { .. } if recorded_on_windows => None, StudioRecordingMeta::MultipleSegments { inner, .. } => { inner.segments[segment_i].latest_start_time() } diff --git a/scripts/build-desktop-binaries.mjs b/scripts/build-desktop-binaries.mjs index 66af8651a33..506be6962d9 100644 --- a/scripts/build-desktop-binaries.mjs +++ b/scripts/build-desktop-binaries.mjs @@ -92,6 +92,7 @@ async function main() { path.join(repoRoot, "crates", "media"), path.join(repoRoot, "crates", "media-info"), path.join(repoRoot, "crates", "project"), + path.join(repoRoot, "crates", "recording"), path.join(repoRoot, "crates", "rendering"), path.join(repoRoot, "Cargo.lock"), ],