diff --git a/.github/workflows/sync-tests.yml b/.github/workflows/sync-tests.yml index 1228d40e528..0e7cc1049ca 100644 --- a/.github/workflows/sync-tests.yml +++ b/.github/workflows/sync-tests.yml @@ -146,7 +146,7 @@ jobs: - name: Timestamp pipeline unit + property tests shell: bash run: | - cargo test --locked -p cap-timestamp -p cap-enc-ffmpeg + cargo test --locked -p cap-timestamp -p cap-enc-ffmpeg -p ffmpeg-hw-device cargo test --locked -p cap-camera-effects --lib cargo test --locked -p cap-recording --lib -- --test-threads=1 # --nocapture so a WARP-adapter notch skip prints instead of diff --git a/crates/enc-ffmpeg/src/video/h264.rs b/crates/enc-ffmpeg/src/video/h264.rs index 0042b4610af..6b2e6d41d77 100644 --- a/crates/enc-ffmpeg/src/video/h264.rs +++ b/crates/enc-ffmpeg/src/video/h264.rs @@ -879,6 +879,14 @@ impl H264Encoder { .map_err(|e| H264EncoderError::VideoToolboxHwInput(e.to_string())) } + /// Whether the selected codec is a hardware encoder, which can fail at + /// runtime after opening (a busy or reset media engine). + pub fn is_hardware(&self) -> bool { + self.encoder + .codec() + .is_some_and(|codec| is_hardware_h264(codec.name())) + } + pub fn conversion_requirements(&self) -> ConversionRequirements { let needs_conversion = self.input_format != self.output_format || self.input_width != self.output_width diff --git a/crates/export/Cargo.toml b/crates/export/Cargo.toml index b1cae02e4ec..9a214783fa2 100644 --- a/crates/export/Cargo.toml +++ b/crates/export/Cargo.toml @@ -3,6 +3,10 @@ name = "cap-export" version = "0.1.0" edition = "2024" +[features] +# Encoder failure injection for the cap-verify suite. Never enabled by the app. +test-hooks = [] + [dependencies] cap-utils = { path = "../utils" } cap-project = { path = "../project" } diff --git a/crates/export/src/estimates.rs b/crates/export/src/estimates.rs index 99bd390dd54..3d245e93164 100644 --- a/crates/export/src/estimates.rs +++ b/crates/export/src/estimates.rs @@ -289,6 +289,8 @@ pub async fn estimate_export( audio_cancellation: None, sample_windows: Some(windows.clone()), sample_timing: Some(timing.clone()), + #[cfg(feature = "test-hooks")] + video_encoder_failure_at: None, }; let progress_editor = editor.clone(); let progress_cancel = cancel.clone(); diff --git a/crates/export/src/lib.rs b/crates/export/src/lib.rs index 86b5514ed3a..59003c58fc6 100644 --- a/crates/export/src/lib.rs +++ b/crates/export/src/lib.rs @@ -254,6 +254,8 @@ impl ExporterBuilder { audio_cancellation, sample_windows: None, sample_timing: None, + #[cfg(feature = "test-hooks")] + video_encoder_failure_at: None, }) } } @@ -440,6 +442,11 @@ impl Mp4ExporterBase { pub fn uses_streaming_audio(&self) -> bool { self.0.streaming_audio.is_some() } + + #[cfg(feature = "test-hooks")] + pub fn fail_video_encoder_at(&mut self, frames: u32) { + self.0.fail_video_encoder_at(frames); + } } pub struct ExporterBase { @@ -456,9 +463,55 @@ pub struct ExporterBase { audio_cancellation: Option, sample_windows: Option, sample_timing: Option>, + #[cfg(feature = "test-hooks")] + video_encoder_failure_at: Option, } impl ExporterBase { + /// Makes the MP4 export's video encoder fail after `frames` frames the + /// way a hardware encoder does at runtime, once, so the software + /// encoder fallback runs on every machine. + #[cfg(feature = "test-hooks")] + pub fn fail_video_encoder_at(&mut self, frames: u32) { + self.video_encoder_failure_at = Some(frames); + } + + /// Reopens what an MP4 export consumes (the staged output and the + /// streaming audio with its cancellation) so the export can start over. + async fn reopen_streaming( + &mut self, + user_cancellation: Option>, + ) -> Result<(), String> { + let Some(user) = user_cancellation else { + return Ok(()); + }; + let control = ExportAudioCancellation::new(user); + let recording = self.recording_meta.clone(); + let studio = self.studio_meta.clone(); + let cancellation = control.user.clone(); + let abort = control.stop.clone(); + let preparation = tokio::task::spawn_blocking(move || { + ExportAudioPreparation::open(&recording, &studio, cancellation, abort) + }) + .await + .map_err(|error| error.to_string())? + .map_err(|error| error.to_string())?; + self.streaming_audio = Some( + preparation + .finish(&self.segments) + .map_err(|error| error.to_string())?, + ); + // The first attempt staged its output, so the restart must too: if + // something created the destination meanwhile, writing in place + // would truncate it. + self.streaming_output = Some( + prepare_streaming_output(&self.output_path, true) + .ok_or("Could not stage the export output to restart on the software encoder")?, + ); + self.audio_cancellation = Some(control); + Ok(()) + } + pub fn total_frames(&self, fps: u32) -> u32 { let duration = cap_rendering::get_duration( &self.recordings, diff --git a/crates/export/src/mp4.rs b/crates/export/src/mp4.rs index 8395ed74090..8c4fb4cae1b 100644 --- a/crates/export/src/mp4.rs +++ b/crates/export/src/mp4.rs @@ -16,7 +16,7 @@ use specta::Type; use std::{ path::PathBuf, sync::{ - Arc, Mutex, + Arc, Mutex, PoisonError, atomic::{AtomicBool, AtomicU64, Ordering}, }, time::Duration, @@ -169,6 +169,7 @@ impl ExportCompression { #[derive(Clone, Default)] struct ExportNv12Mode { stop_after_frames_sent: Option, + software_encoder: bool, record_first_queued_ms_since_pipeline: Option>, nv12_render_startup_breakdown_ms: Option>>>, @@ -294,6 +295,7 @@ impl Mp4ExportSettings { |_| true, ExportNv12Mode { stop_after_frames_sent: Some(1), + software_encoder: false, record_first_queued_ms_since_pipeline: Some(first_ms_enc), nv12_render_startup_breakdown_ms: Some(breakdown_enc), }, @@ -326,7 +328,64 @@ impl Mp4ExportSettings { output_size: (u32, u32), fps: u32, on_progress: impl FnMut(u32) -> bool + Send + 'static, + mut mode: ExportNv12Mode, + ) -> Result { + let on_progress = Arc::new(Mutex::new(on_progress)); + let user_cancellation = base + .audio_cancellation + .as_ref() + .map(|control| control.user.clone()); + loop { + let progress = on_progress.clone(); + let hardware_encoder_failed = Arc::new(AtomicBool::new(false)); + let result = self + .export_nv12_attempt( + &mut base, + output_size, + fps, + move |frame| { + let mut on_progress = + progress.lock().unwrap_or_else(PoisonError::into_inner); + (*on_progress)(frame) + }, + mode.clone(), + hardware_encoder_failed.clone(), + ) + .await; + // A hardware encoder can fail mid-stream when the media engine is + // busy or resets. The MP4 cannot switch codecs partway, so the + // export starts over once with the software encoder. + match result { + Err(error) + if restart_on_software_encoder( + mode.software_encoder, + base.sample_timing.is_some(), + hardware_encoder_failed.load(Ordering::Relaxed), + user_cancellation + .as_ref() + .is_some_and(|cancel| cancel.load(Ordering::Relaxed)), + ) => + { + warn!( + %error, + "Hardware video encoder failed during export, restarting with the software encoder" + ); + base.reopen_streaming(user_cancellation.clone()).await?; + mode.software_encoder = true; + } + result => return result, + } + } + } + + async fn export_nv12_attempt( + self, + base: &mut ExporterBase, + output_size: (u32, u32), + fps: u32, + on_progress: impl FnMut(u32) -> bool + Send + 'static, mode: ExportNv12Mode, + hardware_encoder_failed: Arc, ) -> Result { let pipeline_start = std::time::Instant::now(); let sample_timing = base.sample_timing.clone(); @@ -379,6 +438,9 @@ impl Mp4ExportSettings { let has_audio = has_recording_audio || !music.is_empty(); let record_first_queued_ms = mode.record_first_queued_ms_since_pipeline; + let software_encoder = mode.software_encoder; + #[cfg(feature = "test-hooks")] + let injected_failure_at = base.video_encoder_failure_at.filter(|_| !software_encoder); let nv12_render_startup_breakdown_ms = mode.nv12_render_startup_breakdown_ms; let project_for_audio = base.project_config.clone(); @@ -401,6 +463,7 @@ impl Mp4ExportSettings { } #[cfg(target_os = "macos")] let want_hw_input = !self.optimize_filesize + && !software_encoder && matches!( first_frame.as_ref().map(|frame| &frame.payload), Some(ExportFramePayload::Surface(_)) @@ -442,6 +505,8 @@ impl Mp4ExportSettings { .with_external_conversion(); let builder = if self.optimize_filesize { builder.with_crf(self.compression.crf_value()) + } else if software_encoder { + builder.with_encoder_priority_override(&["libx264"]) } else { builder }; @@ -465,8 +530,17 @@ impl Mp4ExportSettings { encoder.video_mut().set_packet_stats(timing.packet_stats()); } + let hardware_video = encoder.video().is_hardware(); + // The injected failure stands in for a hardware encoder's. + #[cfg(feature = "test-hooks")] + let hardware_video = hardware_video || injected_failure_at.is_some(); + let video_failure = |error: String| { + video_encoder_failure(hardware_video, &hardware_encoder_failed, error) + }; + info!( zero_copy_input = encoder_is_hw, + hardware_video, "Created MP4File encoder (NV12, export settings)" ); @@ -535,7 +609,7 @@ impl Mp4ExportSettings { &mut converted_frame, Duration::MAX, ) - .map_err(|err| err.to_string())?; + .map_err(|err| video_failure(err.to_string()))?; } #[cfg(target_os = "macos")] ExportFramePayload::Surface(surface) => { @@ -545,7 +619,7 @@ impl Mp4ExportSettings { .video_encoder() .wrap_videotoolbox_pixel_buffer(surface.as_pixel_buffer_ptr()) } - .map_err(|err| err.to_string())?; + .map_err(|err| video_failure(err.to_string()))?; frame.set_pts(Some(input.frame_number as i64)); encoder .queue_video_frame_reusable( @@ -553,7 +627,7 @@ impl Mp4ExportSettings { &mut converted_frame, Duration::MAX, ) - .map_err(|err| err.to_string())?; + .map_err(|err| video_failure(err.to_string()))?; } else { fill_nv12_frame_from_surface( &mut reusable_frame, @@ -571,7 +645,7 @@ impl Mp4ExportSettings { &mut converted_frame, Duration::MAX, ) - .map_err(|err| err.to_string())?; + .map_err(|err| video_failure(err.to_string()))?; } } } @@ -579,6 +653,13 @@ impl Mp4ExportSettings { worker.write_ready(&mut encoder)?; } encoded_frames += 1; + #[cfg(feature = "test-hooks")] + if injected_failure_at == Some(encoded_frames) { + return Err(video_failure( + cap_enc_ffmpeg::h264::QueueFrameError::Encode(ffmpeg::Error::External) + .to_string(), + )); + } if let Some(timing) = &sample_timing { timing.record_frame(input.timeline_frame); } @@ -617,7 +698,7 @@ impl Mp4ExportSettings { .map_err(|e| format!("Failed to finish encoding: {e}"))?; if let Err(e) = res.video_finish { - return Err(Mp4PipelineError::Failure(format!("Video encoding failed: {e}"))); + return Err(video_failure(format!("Video encoding failed: {e}"))); } if let Err(e) = res.audio_finish { return Err(Mp4PipelineError::Failure(format!("Audio encoding failed: {e}"))); @@ -735,6 +816,27 @@ impl Mp4ExportSettings { } } +/// A video encoder error that fails this export attempt. A hardware +/// encoder's error also sets `restart`, so the export starts over on libx264. +fn video_encoder_failure(hardware: bool, restart: &AtomicBool, error: String) -> Mp4PipelineError { + if hardware { + restart.store(true, Ordering::Relaxed); + } + Mp4PipelineError::Failure(error) +} + +/// Whether a failed attempt starts over on the software encoder: after a +/// hardware encoder error, once, and not for size-estimate samples or a +/// cancelled export. +fn restart_on_software_encoder( + attempt_was_software: bool, + sampled: bool, + hardware_encoder_failed: bool, + cancelled: bool, +) -> bool { + hardware_encoder_failed && !attempt_was_software && !sampled && !cancelled +} + enum ExportFramePayload { Cpu(SharedNv12Buffer), #[cfg(target_os = "macos")] @@ -1606,6 +1708,40 @@ async fn export_render_to_channel( mod tests { use super::*; + #[test] + fn hardware_encoder_errors_restart_on_software() { + let queue = cap_enc_ffmpeg::h264::QueueFrameError::Encode(ffmpeg::Error::External); + let flush = format!("Video encoding failed: {}", ffmpeg::Error::External); + for error in [queue.to_string(), flush] { + let restart = AtomicBool::new(false); + let failure = video_encoder_failure(true, &restart, error.clone()); + assert_eq!(failure.message(), error); + assert!(restart_on_software_encoder( + false, + false, + restart.load(Ordering::Relaxed), + false + )); + } + } + + #[test] + fn software_encoder_errors_do_not_restart() { + let restart = AtomicBool::new(false); + let error = cap_enc_ffmpeg::h264::QueueFrameError::Encode(ffmpeg::Error::External); + video_encoder_failure(false, &restart, error.to_string()); + assert!(!restart_on_software_encoder( + false, + false, + restart.load(Ordering::Relaxed), + false + )); + // Already on libx264, a size-estimate sample, or cancelled. + assert!(!restart_on_software_encoder(true, false, true, false)); + assert!(!restart_on_software_encoder(false, true, true, false)); + assert!(!restart_on_software_encoder(false, false, true, true)); + } + fn sum_samples(sample_rate: u64, fps: u64, frames: u64) -> u64 { (0..frames) .map(|n| { diff --git a/crates/ffmpeg-hw-device/src/lib.rs b/crates/ffmpeg-hw-device/src/lib.rs index 8691bebe316..0bfd452cb96 100644 --- a/crates/ffmpeg-hw-device/src/lib.rs +++ b/crates/ffmpeg-hw-device/src/lib.rs @@ -1,6 +1,10 @@ use std::{ cell::Cell, ptr::{null, null_mut}, + sync::{ + Arc, + atomic::{AtomicBool, Ordering}, + }, }; use ffmpeg::{ @@ -9,16 +13,18 @@ use ffmpeg::{ sys::{ AV_CODEC_HW_CONFIG_METHOD_HW_DEVICE_CTX, AVBufferRef, AVCodecContext, AVCodecHWConfig, AVHWDeviceType, AVPixelFormat, av_buffer_ref, av_buffer_unref, av_dict_free, av_dict_set, - av_hwdevice_ctx_create, av_hwframe_transfer_data, avcodec_get_hw_config, + av_hwdevice_ctx_create, av_hwframe_transfer_data, avcodec_default_get_format, + avcodec_get_hw_config, }, }; +use tracing::warn; thread_local! { static HW_PIX_FMT: Cell = const { Cell::new(AVPixelFormat::AV_PIX_FMT_NONE) }; } unsafe extern "C" fn get_format( - _: *mut AVCodecContext, + ctx: *mut AVCodecContext, pix_fmts: *const AVPixelFormat, ) -> AVPixelFormat { unsafe { @@ -30,19 +36,50 @@ unsafe extern "C" fn get_format( } if *fmt == HW_PIX_FMT.get() { + set_software_fallback(ctx, false); return *fmt; } fmt = fmt.offset(1); } - AVPixelFormat::AV_PIX_FMT_NONE + // When the hardware format's setup fails (a busy or reset media + // engine), FFmpeg drops it from the list and asks again. Choosing + // nothing here would fail every packet, so take FFmpeg's default + // software format and keep decoding on the CPU. + warn!( + hw_pix_fmt = ?HW_PIX_FMT.get(), + "Hardware decoder setup failed, falling back to software decode" + ); + set_software_fallback(ctx, true); + avcodec_default_get_format(ctx, pix_fmts) + } +} + +/// Records `get_format`'s latest choice in the [`HwDevice`] flag that +/// [`install_get_format`] put in the codec context's `opaque`. +unsafe fn set_software_fallback(ctx: *mut AVCodecContext, software: bool) { + unsafe { + if let Some(flag) = (*ctx).opaque.cast::().cast_const().as_ref() { + flag.store(software, Ordering::Relaxed); + } + } +} + +/// Points the codec context at [`get_format`] and at the flag it records the +/// chosen format in. The flag outlives the context as long as the context is +/// freed before the [`HwDevice`] holding the flag. +unsafe fn install_get_format(context: *mut AVCodecContext, software_fallback: &Arc) { + unsafe { + (*context).get_format = Some(get_format); + (*context).opaque = Arc::as_ptr(software_fallback).cast_mut().cast(); } } pub struct HwDevice { pub device_type: AVHWDeviceType, ctx: *mut AVBufferRef, + software_fallback: Arc, } fn linux_gpu_frames() -> bool { @@ -51,6 +88,12 @@ fn linux_gpu_frames() -> bool { } impl HwDevice { + /// Whether FFmpeg took a software format because the hardware setup + /// failed, so the decoder is running on the CPU despite this device. + pub fn is_software_fallback(&self) -> bool { + self.software_fallback.load(Ordering::Relaxed) + } + pub fn get_hwframe(&self, src: &Video) -> Option