Skip to content

fix(recording): keep Windows screen frames in sync #2002

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Draft
hyuhan666-art wants to merge 1 commit into
base: main
Choose a base branch
from
Draft

fix(recording): keep Windows screen frames in sync #2002

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
111 changes: 89 additions & 22 deletions crates/recording/src/output_pipeline/oop_fragmented_m4s_win.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -20,15 +20,15 @@ use std::{
sync::{
Arc,
atomic::AtomicBool,
mpsc::{RecvTimeoutError, SyncSender, sync_channel},
mpsc::{RecvTimeoutError, SyncSender, TryRecvError, TrySendError, sync_channel},
},
thread::JoinHandle,
time::Duration,
};
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")
Expand All @@ -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<ScreenFrameItem>,
timeout: Duration,
) -> Result<Option<LatestScreenFrame>, 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,
Expand Down Expand Up @@ -107,8 +150,6 @@ impl FrameDropTracker {
}
}

type ScreenFrameItem = Option<(screen_capture::ScreenFrame, Duration)>;

struct EncoderState {
video_tx: SyncSender<ScreenFrameItem>,
encoder_handle: Option<JoinHandle<anyhow::Result<()>>>,
Expand Down Expand Up @@ -371,6 +412,8 @@ impl WindowsOOPFragmentedM4SMuxer {
let mut first_timestamp: Option<Duration> = 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;
Expand Down Expand Up @@ -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 => {
Expand All @@ -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);
Expand Down Expand Up @@ -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!(
Expand Down Expand Up @@ -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"));
}
}
Expand Down
92 changes: 80 additions & 12 deletions crates/recording/src/output_pipeline/win.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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")
Expand All @@ -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<ScreenFrameItem>,
timeout: Duration,
) -> Result<Option<LatestScreenFrame>, 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,
Expand Down Expand Up @@ -327,15 +370,28 @@ impl Muxer for WindowsMuxer {
let mut last_timestamp: Option<Duration> = 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");
Expand Down Expand Up @@ -426,19 +482,31 @@ impl Muxer for WindowsMuxer {
let mut last_ffmpeg_frame: Option<ffmpeg::frame::Video> = None;
let mut first_timestamp: Option<Duration> = None;
let mut last_timestamp: Option<Duration> = 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)
}
}
}
Expand Down
Loading