From e86fc2ce6069f2ac3e1177aaa57b629cd637f59b Mon Sep 17 00:00:00 2001 From: Richie McIlroy <33632126+richiemcilroy@users.noreply.github.com> Date: Tue, 27 Oct 2026 14:39:26 +0000 Subject: [PATCH 1/9] feat(enc-ffmpeg): add libx264 tuning and detached software encoder --- crates/enc-ffmpeg/src/video/h264.rs | 117 +++++++++++++++++++++++++++- 1 file changed, 114 insertions(+), 3 deletions(-) diff --git a/crates/enc-ffmpeg/src/video/h264.rs b/crates/enc-ffmpeg/src/video/h264.rs index 01db20997a7..bdf1c6f68d1 100644 --- a/crates/enc-ffmpeg/src/video/h264.rs +++ b/crates/enc-ffmpeg/src/video/h264.rs @@ -35,6 +35,30 @@ pub struct H264EncoderBuilder { encoder_priority_override: Option<&'static [&'static str]>, is_export: bool, crf: Option, + software_tuning: Option, +} + +/// libx264 settings for an export split across several encoders running at +/// once: each gets a share of the CPU threads, and `capped_crf` switches rate +/// control to constant quality with the export bitrate as its ceiling. +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +pub struct SoftwareEncoderTuning { + pub threads: usize, + pub capped_crf: Option, +} + +impl SoftwareEncoderTuning { + fn apply(&self, codec: &Codec, options: &mut Dictionary<'static>, max_bitrate: usize) { + if codec.name() != "libx264" { + return; + } + options.set("threads", &self.threads.max(1).to_string()); + if let Some(crf) = self.capped_crf { + options.set("crf", &crf.to_string()); + options.set("maxrate", &max_bitrate.to_string()); + options.set("bufsize", &max_bitrate.saturating_mul(2).to_string()); + } + } } #[derive(Clone, Copy, Debug, PartialEq)] @@ -87,9 +111,15 @@ impl H264EncoderBuilder { encoder_priority_override: None, is_export: false, crf: None, + software_tuning: None, } } + pub fn with_software_tuning(mut self, tuning: SoftwareEncoderTuning) -> Self { + self.software_tuning = Some(tuning); + self + } + pub fn with_preset(mut self, preset: H264Preset) -> Self { self.preset = preset; self @@ -171,8 +201,20 @@ impl H264EncoderBuilder { let mut last_error = None; - for (codec, encoder_options) in candidates { + for (codec, mut encoder_options) in candidates { let codec_name = codec.name().to_string(); + if let Some(tuning) = &self.software_tuning { + tuning.apply( + &codec, + &mut encoder_options, + get_bitrate( + output_width, + output_height, + input_config.frame_rate.0 as f32 / input_config.frame_rate.1.max(1) as f32, + self.bpp, + ), + ); + } if is_hardware_h264(&codec_name) && let Err(reason) = cached_hardware_self_test( @@ -308,6 +350,75 @@ impl H264EncoderBuilder { }) } + /// Opens libx264 without an output file, for packets written through a + /// [`PacketSink`] as stream 0 and muxed into an output later. + pub fn build_software_detached(self) -> Result { + let input_config = self.input_config; + let (raw_width, raw_height) = self + .output_size + .unwrap_or((input_config.width, input_config.height)); + let candidates = get_codec_and_options( + &input_config, + self.preset, + Some(&["libx264"]), + self.is_export, + self.crf, + ); + let (codec, mut encoder_options) = candidates + .into_iter() + .next() + .ok_or(H264EncoderError::CodecNotFound)?; + let (output_width, output_height) = (ensure_even(raw_width), ensure_even(raw_height)); + if let Some(tuning) = &self.software_tuning { + tuning.apply( + &codec, + &mut encoder_options, + get_bitrate( + output_width, + output_height, + input_config.frame_rate.0 as f32 / input_config.frame_rate.1.max(1) as f32, + self.bpp, + ), + ); + } + let OpenedVideoEncoder { + encoder, + converter, + converted_frame_pool, + output_format, + output_width, + output_height, + input_format, + input_width, + input_height, + hw_upload, + } = open_video_encoder( + codec, + encoder_options, + &input_config, + output_width, + output_height, + self.bpp, + self.external_conversion, + self.crf, + )?; + Ok(H264Encoder { + base: EncoderBase::new(0), + encoder, + converter, + output_format, + output_width, + output_height, + input_format, + input_width, + input_height, + converted_frame_pool, + hw_upload, + #[cfg(target_os = "macos")] + videotoolbox_hw: None, + }) + } + /// Builds an `h264_videotoolbox` encoder that consumes IOSurface-backed /// NV12 CVPixelBuffers directly (`AV_PIX_FMT_VIDEOTOOLBOX`), skipping the /// per-frame CPU plane copy a software-input encoder requires. Frames are @@ -1021,12 +1132,12 @@ impl H264Encoder { Ok(()) } - pub fn queue_frame_reusable( + pub fn queue_frame_reusable( &mut self, frame: &mut frame::Video, converted_frame: &mut Option, timestamp: Duration, - output: &mut format::context::Output, + output: &mut S, ) -> Result<(), QueueFrameError> { self.base.update_pts(frame, timestamp, &mut self.encoder); From 771b3456b163f363a47b05e44285f790bd38308c Mon Sep 17 00:00:00 2001 From: Richie McIlroy <33632126+richiemcilroy@users.noreply.github.com> Date: Tue, 27 Oct 2026 14:39:26 +0000 Subject: [PATCH 2/9] feat(export): add turbo mp4 export with parallel chunks --- Cargo.lock | 2 + crates/export/Cargo.toml | 4 + .../export/examples/export-turbo-benchmark.rs | 134 ++ crates/export/src/estimates.rs | 53 +- crates/export/src/lib.rs | 13 + crates/export/src/mp4.rs | 49 +- crates/export/src/mp4/turbo.rs | 1160 +++++++++++++++++ 7 files changed, 1411 insertions(+), 4 deletions(-) create mode 100644 crates/export/examples/export-turbo-benchmark.rs create mode 100644 crates/export/src/mp4/turbo.rs diff --git a/Cargo.lock b/Cargo.lock index b25b55e3989..3fdbfcef062 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1741,6 +1741,7 @@ dependencies = [ name = "cap-export" version = "0.1.0" dependencies = [ + "anyhow", "base64 0.22.1", "cap-editor", "cap-enc-ffmpeg", @@ -1760,6 +1761,7 @@ dependencies = [ "image 0.25.8", "imgref", "inquire", + "libc", "mp4", "rgb", "serde", diff --git a/crates/export/Cargo.toml b/crates/export/Cargo.toml index 9a214783fa2..d24a3fb7355 100644 --- a/crates/export/Cargo.toml +++ b/crates/export/Cargo.toml @@ -43,6 +43,10 @@ inquire = "0.7.5" tokio.workspace = true serde_json.workspace = true tracing-subscriber = { version = "0.3.19", features = ["env-filter"] } +anyhow.workspace = true + +[target.'cfg(unix)'.dev-dependencies] +libc = "0.2" [lints] workspace = true diff --git a/crates/export/examples/export-turbo-benchmark.rs b/crates/export/examples/export-turbo-benchmark.rs new file mode 100644 index 00000000000..9a5110b0392 --- /dev/null +++ b/crates/export/examples/export-turbo-benchmark.rs @@ -0,0 +1,134 @@ +//! Run: cargo run --release -p cap-export --example export-turbo-benchmark -- [WIDTH] [HEIGHT] [FPS] [standard|turbo] + +use std::{ + path::PathBuf, + sync::{Arc, atomic::AtomicBool}, + time::Instant, +}; + +use cap_export::{ + ExporterBase, + mp4::{ExportCompression, Mp4ExportSettings}, +}; +use cap_project::XY; + +/// CPU seconds used and peak resident bytes so far. +#[cfg(unix)] +fn usage() -> (f64, u64) { + let mut usage = std::mem::MaybeUninit::::zeroed(); + unsafe { libc::getrusage(libc::RUSAGE_SELF, usage.as_mut_ptr()) }; + let usage = unsafe { usage.assume_init() }; + let secs = |time: libc::timeval| time.tv_sec as f64 + time.tv_usec as f64 / 1_000_000.0; + #[cfg(target_os = "macos")] + let peak = usage.ru_maxrss as u64; + #[cfg(not(target_os = "macos"))] + let peak = usage.ru_maxrss as u64 * 1024; + (secs(usage.ru_utime) + secs(usage.ru_stime), peak) +} + +#[cfg(not(unix))] +fn usage() -> (f64, u64) { + (0.0, 0) +} + +async fn run( + project_path: PathBuf, + output_path: PathBuf, + size: XY, + fps: u32, + turbo: bool, +) -> anyhow::Result<()> { + let _ = std::fs::remove_file(&output_path); + let cancel = Arc::new(AtomicBool::new(false)); + let started = Instant::now(); + let (cpu_started, _) = usage(); + let base = ExporterBase::builder(project_path) + .with_output_path(output_path.clone()) + .build_for_mp4(cancel) + .await + .map_err(|error| anyhow::anyhow!("{error}"))?; + let total = base.total_frames(fps); + let prepared_ms = started.elapsed().as_secs_f64() * 1000.0; + let settings = Mp4ExportSettings { + fps, + resolution_base: size, + compression: match std::env::var("TURBO_BENCH_COMPRESSION").as_deref() { + Ok("social") => ExportCompression::Social, + Ok("web") => ExportCompression::Web, + Ok("potato") => ExportCompression::Potato, + _ => ExportCompression::Maximum, + }, + custom_bpp: None, + force_ffmpeg_decoder: false, + optimize_filesize: std::env::var_os("TURBO_BENCH_OPTIMIZE").is_some(), + }; + let path = if turbo { + let options = cap_export::mp4::turbo::TurboOptions { + chunks: std::env::var("TURBO_BENCH_CHUNKS") + .ok() + .and_then(|value| value.parse().ok()), + strict: true, + }; + settings + .export_prepared_turbo(base, options, |_| true) + .await + } else { + settings.export_prepared(base, |_| true).await + } + .map_err(|error| anyhow::anyhow!("{error}"))?; + let wall = started.elapsed().as_secs_f64(); + let (cpu, peak) = usage(); + let bytes = std::fs::metadata(&path).map(|meta| meta.len()).unwrap_or(0); + println!( + "{}", + serde_json::json!({ + "mode": if turbo { "turbo" } else { "standard" }, + "frames": total, + "width": size.x, + "height": size.y, + "fps": fps, + "prepareMs": prepared_ms, + "wallSecs": wall, + "exportFps": f64::from(total) / wall, + "cpuSecs": cpu - cpu_started, + "peakRssMb": peak as f64 / 1_048_576.0, + "bytes": bytes, + }) + ); + Ok(()) +} + +fn main() -> anyhow::Result<()> { + tracing_subscriber::fmt() + .with_env_filter( + tracing_subscriber::EnvFilter::try_from_default_env().unwrap_or_else(|_| "warn".into()), + ) + .with_writer(std::io::stderr) + .init(); + + let mut arguments = std::env::args().skip(1); + let project_path = arguments + .next() + .map(PathBuf::from) + .ok_or_else(|| anyhow::anyhow!("Pass a studio .cap project path"))?; + let output_path = arguments + .next() + .map(PathBuf::from) + .ok_or_else(|| anyhow::anyhow!("Pass an output .mp4 path"))?; + let mut number = |default: u32| { + arguments + .next() + .and_then(|value| value.parse().ok()) + .unwrap_or(default) + }; + let size = XY::new(number(1920), number(1080)); + let fps = number(60); + let turbo = std::env::args().any(|argument| argument == "turbo"); + + tokio::runtime::Builder::new_multi_thread() + .worker_threads(2) + .thread_stack_size(16 * 1024 * 1024) + .enable_all() + .build()? + .block_on(run(project_path, output_path, size, fps, turbo)) +} diff --git a/crates/export/src/estimates.rs b/crates/export/src/estimates.rs index 3d245e93164..4d009312a7e 100644 --- a/crates/export/src/estimates.rs +++ b/crates/export/src/estimates.rs @@ -125,6 +125,31 @@ pub async fn estimate_export( project: ProjectConfiguration, settings: ExportSettings, cancel: Arc, + on_estimate: impl FnMut(ExportEstimates) + Send, +) -> Result { + estimate(editor, project, settings, false, cancel, on_estimate).await +} + +/// [`estimate_export`] for an MP4 exported with +/// [`crate::mp4::Mp4ExportSettings::export_prepared_turbo`]: the sample runs +/// through the turbo encoder, so its size and pace are turbo's. +pub async fn estimate_turbo_export( + editor: Arc, + project: ProjectConfiguration, + settings: ExportSettings, + cancel: Arc, + on_estimate: impl FnMut(ExportEstimates) + Send, +) -> Result { + let turbo = matches!(settings, ExportSettings::Mp4(_)); + estimate(editor, project, settings, turbo, cancel, on_estimate).await +} + +async fn estimate( + editor: Arc, + project: ProjectConfiguration, + settings: ExportSettings, + turbo: bool, + cancel: Arc, mut on_estimate: impl FnMut(ExportEstimates) + Send, ) -> Result { if cancel.load(Ordering::Acquire) || editor.export_active.load(Ordering::Acquire) { @@ -177,6 +202,9 @@ pub async fn estimate_export( serde_json::to_vec(&(&project, settings)) .map_err(|error| error.to_string())? .hash(&mut hasher); + if turbo { + turbo.hash(&mut hasher); + } let key = hasher.finish(); if let Ok(cache) = CACHE.lock() && let Some(cached) = cache @@ -245,7 +273,11 @@ pub async fn estimate_export( return Err("Recording sources changed during export estimation".into()); } let setup_seconds = setup_started.elapsed().as_secs_f64(); - let plan = sample_plan(settings); + let plan = if turbo { + turbo_sample_plan(settings) + } else { + sample_plan(settings) + }; let windows = sample_windows( total_frames, plan.window_frames, @@ -287,10 +319,13 @@ pub async fn estimate_export( streaming_audio: None, streaming_output: None, audio_cancellation: None, + user_cancellation: None, sample_windows: Some(windows.clone()), sample_timing: Some(timing.clone()), #[cfg(feature = "test-hooks")] video_encoder_failure_at: None, + #[cfg(feature = "test-hooks")] + turbo_chunk_failure: None, }; let progress_editor = editor.clone(); let progress_cancel = cancel.clone(); @@ -302,6 +337,9 @@ pub async fn estimate_export( let started = Instant::now(); let export = async { match settings { + ExportSettings::Mp4(settings) if turbo => { + crate::mp4::turbo::export_sampled(settings, base, on_progress).await + } ExportSettings::Mp4(settings) => settings.export(base, on_progress).await, ExportSettings::Gif(settings) => settings.export(base, on_progress).await, ExportSettings::Mov(settings) => settings.export(base, on_progress).await, @@ -447,6 +485,19 @@ fn sample_plan(settings: ExportSettings) -> SamplePlan { } } +/// The turbo encoder is libx264 whether or not the file size is optimized, +/// so every turbo sample spans a GOP the way an optimized one does. +fn turbo_sample_plan(settings: ExportSettings) -> SamplePlan { + let keyframe_interval = keyframe_interval_frames(settings); + SamplePlan { + windows: 3, + window_frames: keyframe_interval.unwrap_or(settings.fps() * 2) + X264_WARMUP_FRAMES as u32, + leading_window: false, + timing: FrameTiming::GopAmortized, + keyframe_interval, + } +} + fn keyframe_interval_frames(settings: ExportSettings) -> Option { match settings { ExportSettings::Mp4(settings) => Some( diff --git a/crates/export/src/lib.rs b/crates/export/src/lib.rs index 805ace27777..6c5dc34694f 100644 --- a/crates/export/src/lib.rs +++ b/crates/export/src/lib.rs @@ -190,6 +190,7 @@ impl ExporterBuilder { .map_err(Error::RendererSetup)?, ); + let user_cancellation = cancellation.clone(); let audio_cancellation = if stream_audio { cancellation.map(ExportAudioCancellation::new) } else { @@ -254,10 +255,13 @@ impl ExporterBuilder { streaming_audio, streaming_output, audio_cancellation, + user_cancellation, sample_windows: None, sample_timing: None, #[cfg(feature = "test-hooks")] video_encoder_failure_at: None, + #[cfg(feature = "test-hooks")] + turbo_chunk_failure: None, }) } } @@ -449,6 +453,12 @@ impl Mp4ExporterBase { pub fn fail_video_encoder_at(&mut self, frames: u32) { self.0.fail_video_encoder_at(frames); } + + /// Makes this turbo chunk's encoder fail after a few frames. + #[cfg(feature = "test-hooks")] + pub fn fail_turbo_chunk(&mut self, chunk: usize) { + self.0.turbo_chunk_failure = Some(chunk); + } } pub struct ExporterBase { @@ -463,10 +473,13 @@ pub struct ExporterBase { streaming_audio: Option, streaming_output: Option, audio_cancellation: Option, + user_cancellation: Option>, sample_windows: Option, sample_timing: Option>, #[cfg(feature = "test-hooks")] video_encoder_failure_at: Option, + #[cfg(feature = "test-hooks")] + turbo_chunk_failure: Option, } impl ExporterBase { diff --git a/crates/export/src/mp4.rs b/crates/export/src/mp4.rs index be25ee2e62f..90be5f8c2b6 100644 --- a/crates/export/src/mp4.rs +++ b/crates/export/src/mp4.rs @@ -24,6 +24,8 @@ use std::{ use tokio::sync::Notify; use tracing::{info, trace, warn}; +pub mod turbo; + enum Mp4PipelineError { Interrupted, Failure(String), @@ -209,6 +211,46 @@ impl Mp4ExportSettings { self.export(base.0, on_progress) } + /// Exports with every core: see [`turbo`]. Falls back to the standard + /// export when the turbo pipeline fails for any reason but cancellation. + pub async fn export_prepared_turbo( + self, + base: Mp4ExporterBase, + options: turbo::TurboOptions, + on_progress: impl FnMut(u32) -> bool + Send + 'static, + ) -> Result { + let mut base = base.0; + let on_progress = Arc::new(Mutex::new(on_progress)); + let user_cancellation = base + .audio_cancellation + .as_ref() + .map(|control| control.user.clone()); + match turbo::export(self, &mut base, options, on_progress.clone()).await { + Ok(path) => Ok(path), + Err(turbo::TurboError::Cancelled) => Err("Export cancelled".to_string()), + Err(turbo::TurboError::Failed(error)) if options.strict => Err(error), + Err(turbo::TurboError::Failed(error)) => { + warn!(%error, "Turbo export failed, exporting with the standard pipeline"); + base.reopen_streaming(user_cancellation).await?; + let output_size = ProjectUniforms::get_output_size( + &base.render_constants.options, + &base.project_config, + self.resolution_base, + ); + self.export_nv12( + base, + output_size, + self.fps, + move |frame| { + (on_progress.lock().unwrap_or_else(PoisonError::into_inner))(frame) + }, + ExportNv12Mode::default(), + ) + .await + } + } + } + pub async fn export( self, base: ExporterBase, @@ -755,7 +797,9 @@ impl Mp4ExportSettings { nv12_render_startup_breakdown_ms, audio_cancellation.is_some(), on_progress, - base.project_path.clone(), + base.sample_windows + .is_none() + .then(|| base.project_path.clone()), ) .then(move |v| async move { let result = v.map_err(|error| match &error { @@ -1576,11 +1620,10 @@ async fn export_render_to_channel( startup_breakdown_ms: Option>>>, stop_on_encoder_drop: bool, mut on_progress: impl FnMut(u32) -> bool + Send + 'static, - project_path: PathBuf, + screenshot_project_path: Option, ) -> Result<(), cap_rendering::RenderingError> { let (tx_image_data, mut video_rx) = tokio::sync::mpsc::channel::<(Nv12RenderedFrame, u32)>(8); - let screenshot_project_path = frame_windows.is_none().then_some(project_path); let sampling = frame_windows.is_some(); let render_result = { diff --git a/crates/export/src/mp4/turbo.rs b/crates/export/src/mp4/turbo.rs new file mode 100644 index 00000000000..48f20c9083e --- /dev/null +++ b/crates/export/src/mp4/turbo.rs @@ -0,0 +1,1160 @@ +//! Turbo export: the timeline is split into contiguous chunks that render and +//! encode at the same time, each with its own decoders, render loop and +//! libx264 encoder, so decode, GPU and every CPU core stay busy. Chunk 0 +//! encodes straight into the output like a standard export; later chunks +//! spool their packets to temporary files that are appended in order. +//! +//! libx264 runs in constant quality capped at the export's bitrate. On Apple +//! silicon the single hardware encoder is the standard export's ceiling, +//! while libx264 across all cores is 2-3x faster at matching quality. + +use super::*; +use crate::estimates::SampleTiming; +use cap_editor::SegmentMedia; +use cap_enc_ffmpeg::{PacketSink, h264::SoftwareEncoderTuning}; +use cap_project::{ProjectConfiguration, RecordingMeta, StudioRecordingMeta}; +use cap_rendering::{ProjectRecordingsMeta, RenderVideoConstants}; +use std::{ + io::{BufReader, BufWriter, Read, Write}, + ops::Range, + sync::mpsc, +}; + +/// Shorter chunks spend more of their time opening decoders and layers. +const MIN_CHUNK_SECS: u32 = 4; +const MAX_CHUNKS: usize = 4; +const CORES_PER_CHUNK: usize = 4; + +/// Frames a later chunk renders and discards before its first frame, so +/// state carried from frame to frame (the camera blur's mask smoothing) +/// matches a standard export's at the seam. +fn preroll_frames(fps: u32, start: u32) -> u32 { + (fps / 2).max(16).min(start) +} + +#[derive(Clone, Copy, Debug, Default)] +pub struct TurboOptions { + /// Overrides how many chunks run at once, however short they get. + pub chunks: Option, + /// Returns the turbo pipeline's error instead of exporting again with + /// the standard pipeline. + pub strict: bool, +} + +pub(crate) enum TurboError { + Cancelled, + Failed(String), +} + +impl From for TurboError { + fn from(error: String) -> Self { + Self::Failed(error) + } +} + +#[derive(Debug, PartialEq, Eq)] +struct TurboPlan { + chunks: Vec>, + threads: usize, +} + +fn plan(total_frames: u32, fps: u32, cores: usize, requested: Option) -> TurboPlan { + let count = match requested { + Some(count) => count.clamp(1, MAX_CHUNKS).min(total_frames.max(1) as usize), + None => (cores / CORES_PER_CHUNK) + .clamp(1, MAX_CHUNKS) + .min((total_frames / (fps.max(1) * MIN_CHUNK_SECS)).max(1) as usize), + }; + let length = total_frames / count as u32; + let remainder = total_frames % count as u32; + let mut start = 0; + let chunks = (0..count as u32) + .map(|index| { + let end = start + length + u32::from(index < remainder); + let chunk = start..end; + start = end; + chunk + }) + .collect(); + TurboPlan { + chunks, + threads: (cores / count).max(1), + } +} + +/// libx264 constant quality per bits per pixel and frame rate. Each preset's +/// value matched or beat the hardware encoder's SSIM and PSNR at that preset +/// against a lossless render of real recordings, at a smaller file, at 15, 30 +/// and 60 fps. At 60 fps the hardware encoder spends its higher bitrate on +/// repeated source frames that libx264 makes nearly free, so matching it +/// takes 3 steps more quality; at 15 fps 2 steps less still beats it. +fn constant_quality(bpp: f32, fps: u32) -> u8 { + const POINTS: [(f32, f32); 4] = [(0.04, 22.0), (0.08, 20.0), (0.15, 18.0), (0.3, 15.0)]; + let position = bpp.max(f32::MIN_POSITIVE).log2(); + let crf = POINTS + .windows(2) + .find_map(|pair| { + let (low, high) = (pair[0], pair[1]); + let (from, to) = (low.0.log2(), high.0.log2()); + (position <= to).then(|| { + let t = ((position - from) / (to - from)).clamp(0.0, 1.0); + low.1 + (high.1 - low.1) * t + }) + }) + .unwrap_or(POINTS[POINTS.len() - 1].1); + let fps = fps as f32; + let offset = if fps >= 30.0 { + -3.0 * ((fps - 30.0) / 30.0).min(1.0) + } else { + 2.0 * ((30.0 - fps) / 15.0).min(1.0) + }; + (crf + offset).round() as u8 +} + +struct Progress { + frames: Mutex<(u32, F)>, + stop: Arc, + user: Option>, + cancelled: AtomicBool, + failure: Mutex>, +} + +impl bool> Progress { + fn stopped(&self) -> bool { + self.stop.load(Ordering::Relaxed) + || self + .user + .as_ref() + .is_some_and(|user| user.load(Ordering::Relaxed)) + } + + fn frame_rendered(&self) -> bool { + if self.stopped() { + return false; + } + let mut frames = self.frames.lock().unwrap_or_else(PoisonError::into_inner); + let (rendered, on_progress) = &mut *frames; + let keep_going = on_progress(*rendered); + *rendered += 1; + if !keep_going { + self.cancelled.store(true, Ordering::Relaxed); + self.stop.store(true, Ordering::Relaxed); + } + keep_going + } + + /// Stops every part of the export, keeping the first failure: the parts + /// it stops fail afterwards only because they were stopped. + fn fail(&self, error: String) { + let mut failure = self.failure.lock().unwrap_or_else(PoisonError::into_inner); + if failure.is_none() && !self.stopped() { + *failure = Some(error); + } + self.stop.store(true, Ordering::Relaxed); + } + + fn was_cancelled(&self) -> bool { + self.cancelled.load(Ordering::Relaxed) + || self + .user + .as_ref() + .is_some_and(|user| user.load(Ordering::Relaxed)) + } +} + +/// Everything a chunk's render loop reads, shared by all chunks. +#[derive(Clone)] +struct RenderShared { + constants: Arc, + project: Arc, + recording_meta: Arc, + studio_meta: Arc, + recordings: Arc, + fps: u32, + resolution_base: XY, +} + +#[derive(Clone, Copy)] +struct EncoderSettings { + video_info: VideoInfo, + bpp: f32, + optimize_filesize: bool, + crf: u8, + tuning: SoftwareEncoderTuning, +} + +impl EncoderSettings { + fn builder(&self) -> cap_enc_ffmpeg::h264::H264EncoderBuilder { + let builder = H264Encoder::builder(self.video_info) + .with_bpp(self.bpp) + .with_export_settings() + .with_external_conversion() + .with_encoder_priority_override(&["libx264"]) + .with_software_tuning(self.tuning); + if self.optimize_filesize { + builder.with_crf(self.crf) + } else { + builder + } + } +} + +const STOPPED: &str = "Export cancelled"; + +fn render_chunk bool + Send + 'static>( + shared: RenderShared, + segments: Vec, + frames: cap_rendering::FrameWindows, + preroll: u32, + sender: mpsc::SyncSender, + progress: Arc>, + screenshot_project_path: Option, +) -> std::io::Result>> { + std::thread::Builder::new() + .name("turbo-render".to_string()) + .stack_size(16 * 1024 * 1024) + .spawn(move || { + let runtime = tokio::runtime::Builder::new_current_thread() + .enable_all() + .build() + .map_err(|error| error.to_string())?; + let result = runtime + .block_on(export_render_to_channel( + &shared.constants, + &shared.project, + sender, + &shared.recording_meta, + &shared.studio_meta, + segments, + shared.fps, + shared.resolution_base, + &shared.recordings, + Some(frames), + None, + None, + true, + { + let progress = progress.clone(); + move |frame| { + if frame < preroll { + !progress.stopped() + } else { + progress.frame_rendered() + } + } + }, + screenshot_project_path, + )) + .map_err(|error| match error { + cap_rendering::RenderingError::ImageLoadError(message) + if message == STOPPED => + { + STOPPED.to_string() + } + error => error.to_string(), + }); + if let Err(error) = &result + && error != STOPPED + { + progress.fail(error.clone()); + } + result + }) +} + +/// Copies a rendered frame into the encoder's NV12 input frame. +fn fill_input( + input: &ExportFrame, + frame: &mut ffmpeg::frame::Video, + pts: i64, +) -> Result<(), String> { + match &input.payload { + ExportFramePayload::Cpu(nv12_data) => { + fill_nv12_frame_direct( + frame, + nv12_data, + input.width, + input.height, + input.y_stride, + pts, + ); + Ok(()) + } + #[cfg(target_os = "macos")] + ExportFramePayload::Surface(surface) => { + fill_nv12_frame_from_surface(frame, surface, input.width, input.height, pts) + .ok_or_else(|| "failed to lock NV12 surface for encoding".to_string()) + } + } +} + +/// Packets of one chunk, written as they are encoded and read back once the +/// output reaches the chunk. Timestamps are frames from the chunk's start. +struct Spool { + path: tempfile::TempPath, + writer: BufWriter, + time_base: ffmpeg::Rational, + first_packet: Option>, + error: Option, +} + +struct SpooledChunk { + path: tempfile::TempPath, + first_packet: Option>, +} + +impl Spool { + fn new(time_base: ffmpeg::Rational) -> std::io::Result { + let (file, path) = tempfile::Builder::new() + .prefix("cap-turbo-") + .suffix(".packets") + .tempfile()? + .into_parts(); + Ok(Self { + path, + writer: BufWriter::with_capacity(1 << 20, file), + time_base, + first_packet: None, + error: None, + }) + } + + fn write(&mut self, packet: &ffmpeg::Packet) -> std::io::Result<()> { + let data = packet.data().unwrap_or_default(); + if self.first_packet.is_none() { + self.first_packet = Some(data.to_vec()); + } + let mut header = [0u8; 29]; + header[0..8].copy_from_slice(&packet.pts().unwrap_or(0).to_le_bytes()); + header[8..16].copy_from_slice(&packet.dts().unwrap_or(0).to_le_bytes()); + header[16..24].copy_from_slice(&packet.duration().to_le_bytes()); + header[24] = u8::from(packet.is_key()); + header[25..29].copy_from_slice(&(data.len() as u32).to_le_bytes()); + self.writer.write_all(&header)?; + self.writer.write_all(data) + } + + fn finish(mut self) -> std::io::Result { + if let Some(error) = self.error.take() { + return Err(error); + } + self.writer.flush()?; + Ok(SpooledChunk { + path: self.path, + first_packet: self.first_packet, + }) + } +} + +impl PacketSink for Spool { + fn stream_time_base(&self, _stream_index: usize) -> ffmpeg::Rational { + self.time_base + } + + fn write_packet(&mut self, packet: ffmpeg::Packet) -> Result<(), ffmpeg::Error> { + if self.error.is_some() { + return Err(ffmpeg::Error::External); + } + self.write(&packet).map_err(|error| { + self.error = Some(error); + ffmpeg::Error::External + }) + } +} + +fn read_spooled_packets( + chunk: &SpooledChunk, + mut packet: impl FnMut(ffmpeg::Packet) -> Result<(), TurboError>, +) -> Result<(), TurboError> { + let file = std::fs::File::open(&chunk.path).map_err(|error| error.to_string())?; + let mut reader = BufReader::with_capacity(1 << 20, file); + let mut header = [0u8; 29]; + loop { + match reader.read_exact(&mut header) { + Ok(()) => {} + Err(error) if error.kind() == std::io::ErrorKind::UnexpectedEof => return Ok(()), + Err(error) => return Err(error.to_string().into()), + } + let field = |range: Range| i64::from_le_bytes(header[range].try_into().unwrap()); + let length = u32::from_le_bytes(header[25..29].try_into().unwrap()) as usize; + let mut data = vec![0u8; length]; + reader + .read_exact(&mut data) + .map_err(|error| error.to_string())?; + let mut spooled = ffmpeg::Packet::copy(&data); + spooled.set_pts(Some(field(0..8))); + spooled.set_dts(Some(field(8..16))); + spooled.set_duration(field(16..24)); + if header[24] != 0 { + spooled.set_flags(ffmpeg::packet::Flags::KEY); + } + packet(spooled)?; + } +} + +/// Encodes a chunk into its spool. +fn encode_chunk bool>( + frames: mpsc::Receiver, + settings: EncoderSettings, + progress: &Progress, + preroll: u32, + fail_after: Option, +) -> Result { + let mut encoder = settings + .builder() + .build_software_detached() + .map_err(|error| error.to_string())?; + let mut spool = Spool::new(settings.video_info.time_base).map_err(|error| error.to_string())?; + let mut frame = ffmpeg::frame::Video::new( + ffmpeg::format::Pixel::NV12, + settings.video_info.width, + settings.video_info.height, + ); + let mut converted = None; + for input in frames.iter() { + if progress.stopped() { + return Err(STOPPED.to_string()); + } + let Some(frame_number) = input.frame_number.checked_sub(preroll) else { + continue; + }; + if fail_after == Some(frame_number) { + return Err("Injected turbo chunk failure".to_string()); + } + fill_input(&input, &mut frame, i64::from(frame_number))?; + encoder + .queue_frame_reusable(&mut frame, &mut converted, Duration::MAX, &mut spool) + .map_err(|error| error.to_string())?; + } + if progress.stopped() { + return Err(STOPPED.to_string()); + } + encoder + .flush_to(&mut spool) + .map_err(|error| format!("Video encoding failed: {error}"))?; + spool.finish().map_err(|error| error.to_string()) +} + +/// Writes chunk 0's packets to the output and keeps the first one, whose +/// parameter sets every later chunk has to repeat exactly. +struct FirstPacketSink<'a> { + output: &'a mut ffmpeg::format::context::Output, + first: Option>, +} + +impl PacketSink for FirstPacketSink<'_> { + fn stream_time_base(&self, stream_index: usize) -> ffmpeg::Rational { + self.output.stream_time_base(stream_index) + } + + fn write_packet(&mut self, packet: ffmpeg::Packet) -> Result<(), ffmpeg::Error> { + if self.first.is_none() { + self.first = Some(packet.data().unwrap_or_default().to_vec()); + } + packet.write_interleaved(self.output) + } +} + +/// SPS and PPS NAL units of an Annex B access unit. +fn parameter_sets(access_unit: &[u8]) -> Vec<&[u8]> { + let mut starts = Vec::new(); + let mut index = 0; + while index + 3 <= access_unit.len() { + if access_unit[index] == 0 && access_unit[index + 1] == 0 && access_unit[index + 2] == 1 { + starts.push(index + 3); + index += 3; + } else { + index += 1; + } + } + starts + .iter() + .enumerate() + .filter_map(|(position, &start)| { + let mut end = starts + .get(position + 1) + .map_or(access_unit.len(), |next| next - 3); + while end > start && access_unit[end - 1] == 0 { + end -= 1; + } + let unit = &access_unit[start..end]; + unit.first() + .is_some_and(|header| matches!(header & 0x1f, 7 | 8)) + .then_some(unit) + }) + .collect() +} + +fn same_parameter_sets(first: Option<&[u8]>, other: Option<&[u8]>) -> bool { + match (first, other) { + (Some(first), Some(other)) => { + let sets = parameter_sets(first); + !sets.is_empty() && sets == parameter_sets(other) + } + _ => false, + } +} + +struct AudioOutput { + worker: ExportAudioWorker, +} + +impl AudioOutput { + /// Writes the audio of every frame up to `frame`, waiting for the audio + /// thread only when it falls more than its lag limit behind. + fn write_through( + &mut self, + frame: u32, + output: &mut ffmpeg::format::context::Output, + ) -> Result<(), Mp4PipelineError> { + let worker = &mut self.worker; + while worker.completed <= frame { + let packets = if frame - worker.completed >= worker.lag_limit { + worker.packets.recv().ok() + } else { + match worker.packets.try_recv() { + Ok(packets) => Some(packets), + Err(mpsc::TryRecvError::Empty) => return Ok(()), + Err(mpsc::TryRecvError::Disconnected) => None, + } + }; + let Some(packets) = packets else { + return Err(worker + .thread + .take() + .and_then(|thread| join_audio_thread(thread).err()) + .unwrap_or_else(|| { + Mp4PipelineError::Failure("Export audio thread stopped early".to_string()) + })); + }; + worker.completed += 1; + for packet in packets { + packet + .write_interleaved(output) + .map_err(|error| Mp4PipelineError::Failure(error.to_string()))?; + } + } + Ok(()) + } + + fn finish(self, output: &mut ffmpeg::format::context::Output) -> Result<(), Mp4PipelineError> { + let ExportAudioWorker { + jobs, + packets, + thread, + .. + } = self.worker; + drop(jobs); + for batch in packets.iter() { + for packet in batch { + packet + .write_interleaved(output) + .map_err(|error| Mp4PipelineError::Failure(error.to_string()))?; + } + } + let thread = thread.ok_or_else(|| { + Mp4PipelineError::Failure("Export audio thread stopped early".to_string()) + })?; + join_audio_thread(thread)? + .flush(output) + .map_err(|error| Mp4PipelineError::Failure(format!("Audio encoding failed: {error}"))) + } +} + +struct StitchInput { + path: PathBuf, + settings: EncoderSettings, + total_frames: u32, + first_chunk: mpsc::Receiver, + later_chunks: Vec<(u32, mpsc::Receiver>)>, + audio: Option<(ExportAudioSource, ProjectConfiguration)>, + streaming_audio: Option, + sample_timing: Option>, +} + +fn stitch bool>( + input: StitchInput, + progress: &Progress, +) -> Result<(), Mp4PipelineError> { + let StitchInput { + path, + settings, + total_frames, + first_chunk, + later_chunks, + audio, + mut streaming_audio, + sample_timing, + } = input; + let sampled = sample_timing.is_some(); + let stopped = || { + if progress.stopped() { + Err(Mp4PipelineError::Interrupted) + } else { + Ok(()) + } + }; + let failure = |error: String| Mp4PipelineError::Failure(error); + + let first_frame = first_chunk.recv().ok(); + stopped()?; + let mut path = path; + path.set_extension("mp4"); + let mut output = ffmpeg::format::output(&path).map_err(|error| failure(error.to_string()))?; + let mut encoder = settings + .builder() + .build(&mut output) + .map_err(|error| failure(error.to_string()))?; + if let Some(timing) = &sample_timing { + encoder.set_packet_stats(timing.packet_stats()); + } + let audio_encoder = match &audio { + Some(_) => Some( + cap_enc_ffmpeg::aac::AACEncoder::init_for_export(AudioRenderer::info(), &mut output) + .map_err(|error| failure(error.to_string()))?, + ), + None => None, + }; + if settings.optimize_filesize { + let mut options = ffmpeg::Dictionary::new(); + options.set("movflags", "+faststart"); + output.write_header_with(options).map(|_| ()) + } else { + output.write_header() + } + .map_err(|error| failure(error.to_string()))?; + + let mut audio_output = match (audio, audio_encoder) { + (Some((source, project)), Some(audio_encoder)) => { + let stream_time_base = output.stream_time_base(audio_encoder.stream_index()); + let stop = progress.stop.clone(); + let user = progress.user.clone(); + let mut worker = ExportAudioWorker::spawn( + stream_time_base, + audio_encoder, + source, + ExportAudioTimeline { + project, + fps: settings.video_info.frame_rate.0 as u32, + sampled, + }, + move || { + stop.load(Ordering::Relaxed) + || user + .as_ref() + .is_some_and(|user| user.load(Ordering::Relaxed)) + }, + )?; + if !sampled { + for frame in 0..total_frames { + worker.request(frame, frame); + } + } + Some(AudioOutput { worker }) + } + _ => None, + }; + + let mut frame = ffmpeg::frame::Video::new( + ffmpeg::format::Pixel::NV12, + settings.video_info.width, + settings.video_info.height, + ); + let mut converted = None; + let mut sink = FirstPacketSink { + output: &mut output, + first: None, + }; + let mut written = 0u32; + for input in first_frame + .into_iter() + .chain(std::iter::from_fn(|| first_chunk.recv().ok())) + { + stopped()?; + if sample_timing + .as_ref() + .is_some_and(|timing| timing.is_cancelled()) + { + return Err(Mp4PipelineError::Interrupted); + } + if sampled && let Some(audio) = &mut audio_output { + audio.worker.request(written, input.timeline_frame); + } + fill_input(&input, &mut frame, input.frame_number as i64).map_err(failure)?; + encoder + .queue_frame_reusable(&mut frame, &mut converted, Duration::MAX, &mut sink) + .map_err(|error| failure(error.to_string()))?; + if let Some(audio) = &mut audio_output { + audio.write_through(written, &mut *sink.output)?; + } + if let Some(timing) = &sample_timing { + timing.record_frame(input.timeline_frame); + } + written += 1; + } + stopped()?; + encoder + .flush_to(&mut sink) + .map_err(|error| failure(format!("Video encoding failed: {error}")))?; + let first_packet = sink.first.take(); + + let video_time_base = settings.video_info.time_base; + let stream_time_base = output.stream_time_base(0); + for (start, chunk) in later_chunks { + if written != start { + return Err(failure(format!( + "Turbo chunk starting at frame {start} follows {written} written frames" + ))); + } + let chunk = chunk + .recv() + .map_err(|_| failure("Turbo chunk ended without a result".to_string()))? + .map_err(failure)?; + stopped()?; + if !same_parameter_sets(first_packet.as_deref(), chunk.first_packet.as_deref()) { + return Err(failure(format!( + "Turbo chunk starting at frame {start} has different H.264 parameter sets" + ))); + } + read_spooled_packets(&chunk, |mut packet| { + let offset = i64::from(start); + packet.set_pts(packet.pts().map(|pts| pts + offset)); + packet.set_dts(packet.dts().map(|dts| dts + offset)); + packet.set_stream(0); + packet.rescale_ts(video_time_base, stream_time_base); + packet + .write_interleaved(&mut output) + .map_err(|error| TurboError::Failed(error.to_string()))?; + if let Some(audio) = &mut audio_output { + audio + .write_through(written, &mut output) + .map_err(|error| TurboError::Failed(error.message()))?; + } + written += 1; + Ok(()) + }) + .map_err(|error| match error { + TurboError::Cancelled => Mp4PipelineError::Interrupted, + TurboError::Failed(error) => failure(error), + })?; + } + if !sampled && written != total_frames { + return Err(failure(format!( + "Turbo export wrote {written} of {total_frames} frames" + ))); + } + + match audio_output { + Some(audio) => audio.finish(&mut output)?, + None => { + if let Some(audio) = &mut streaming_audio { + audio.validate_to_end().map_err(Mp4PipelineError::from)?; + } + } + } + stopped()?; + output + .write_trailer() + .map_err(|error| failure(error.to_string())) +} + +/// An export of the base's sample windows through the turbo encoder, for +/// size and time estimates. +pub(crate) async fn export_sampled( + settings: Mp4ExportSettings, + mut base: ExporterBase, + on_progress: impl FnMut(u32) -> bool + Send + 'static, +) -> Result { + match export( + settings, + &mut base, + TurboOptions { + chunks: Some(1), + strict: true, + }, + Arc::new(Mutex::new(on_progress)), + ) + .await + { + Ok(path) => Ok(path), + Err(TurboError::Cancelled) => Err(STOPPED.to_string()), + Err(TurboError::Failed(error)) => Err(error), + } +} + +pub(crate) async fn export bool + Send + 'static>( + settings: Mp4ExportSettings, + base: &mut ExporterBase, + options: TurboOptions, + on_progress: Arc>, +) -> Result { + let fps = settings.fps; + let output_size = ProjectUniforms::get_output_size( + &base.render_constants.options, + &base.project_config, + settings.resolution_base, + ); + let total_frames = base.total_frames(fps); + let cores = std::thread::available_parallelism() + .map(|cores| cores.get()) + .unwrap_or(4); + let sample_windows = base.sample_windows.clone(); + let plan = plan( + total_frames, + fps, + cores, + if sample_windows.is_some() { + Some(1) + } else { + options.chunks + }, + ); + info!( + chunks = plan.chunks.len(), + threads = plan.threads, + total_frames, + "Exporting mp4 with turbo" + ); + + let mut video_info = + VideoInfo::from_raw(RawVideoFormat::Nv12, output_size.0, output_size.1, fps); + video_info.time_base = ffmpeg::Rational::new(1, fps as i32); + let bpp = settings.effective_bpp(); + let encoder_settings = EncoderSettings { + video_info, + bpp, + optimize_filesize: settings.optimize_filesize, + crf: settings.compression.crf_value(), + tuning: SoftwareEncoderTuning { + threads: plan.threads, + capped_crf: (!settings.optimize_filesize).then(|| constant_quality(bpp, fps)), + }, + }; + + let mut extra_segments = Vec::with_capacity(plan.chunks.len().saturating_sub(1)); + for _ in 1..plan.chunks.len() { + extra_segments.push(cap_editor::create_segments_without_audio( + &base.recording_meta, + &base.studio_meta, + settings.force_ffmpeg_decoder, + )); + } + let extra_segments = futures::future::try_join_all(extra_segments).await?; + if crate::reads_recordings_forward(&base.project_config) { + for segment in extra_segments.iter().flatten() { + segment.decoders.prefer_sequential_access(); + } + } + + let output_path = base.output_path.clone(); + let mut streaming_audio = base.streaming_audio.take(); + let audio_control = base.audio_cancellation.take(); + let temporary_output = match base.streaming_output.take() { + Some(staged) => Some(staged), + None => crate::prepare_new_output(&base.output_path).ok().flatten(), + }; + let user_cancellation = base.user_cancellation.clone(); + let stop = audio_control + .as_ref() + .map(|control| control.stop.clone()) + .unwrap_or_default(); + let progress = Arc::new(Progress { + frames: Mutex::new((0, move |frame| { + (on_progress.lock().unwrap_or_else(PoisonError::into_inner))(frame) + })), + stop: stop.clone(), + user: user_cancellation, + cancelled: AtomicBool::new(false), + failure: Mutex::new(None), + }); + + let unused_audio = streaming_audio + .as_mut() + .and_then(|audio| audio.take_unused_sources(&base.project_config)); + let audio_validation = unused_audio.map(|validation| { + let stop = stop.clone(); + tokio::task::spawn_blocking(move || { + let result = validation.validate_to_end().map_err(Mp4PipelineError::from); + if result.is_err() { + stop.store(true, Ordering::Relaxed); + } + result + }) + }); + + let audio_segments = get_audio_segments(&base.segments).await; + let music = load_music_tracks_uncached(&base.project_config, &base.project_path); + let has_recording_audio = audio_segments + .first() + .filter(|_| !base.project_config.audio.mute) + .is_some(); + let audio = (has_recording_audio || !music.is_empty()).then(|| { + let source = match streaming_audio.take() { + Some(audio) => ExportAudioSource::Streaming(audio), + None => { + ExportAudioSource::Renderer(AudioRenderer::new(audio_segments).with_music(music)) + } + }; + (source, base.project_config.clone()) + }); + + let shared = RenderShared { + constants: base.render_constants.clone(), + project: Arc::new(base.project_config.clone()), + recording_meta: Arc::new(base.recording_meta.clone()), + studio_meta: Arc::new(base.studio_meta.clone()), + recordings: base.recordings.clone(), + fps, + resolution_base: settings.resolution_base, + }; + let render_segments = |segments: &[SegmentMedia]| { + segments + .iter() + .map(|segment| RenderSegment { + cursor: segment.cursor.clone(), + keyboard: segment.keyboard.clone(), + decoders: segment.decoders.clone(), + render_display: true, + }) + .collect::>() + }; + + let mut renders = Vec::with_capacity(plan.chunks.len()); + let mut encoders = Vec::with_capacity(plan.chunks.len().saturating_sub(1)); + let mut later_chunks = Vec::with_capacity(plan.chunks.len().saturating_sub(1)); + let (first_tx, first_chunk) = mpsc::sync_channel::(4); + let spawn_failure = |error: std::io::Error| { + stop.store(true, Ordering::Relaxed); + TurboError::Failed(format!("Could not start a turbo export thread: {error}")) + }; + renders.push( + render_chunk( + shared.clone(), + render_segments(&base.segments), + sample_windows + .clone() + .unwrap_or_else(|| cap_rendering::FrameWindows::new(vec![plan.chunks[0].clone()])), + 0, + first_tx, + progress.clone(), + sample_windows.is_none().then(|| base.project_path.clone()), + ) + .map_err(spawn_failure)?, + ); + for (index, (frames, segments)) in plan.chunks.iter().skip(1).zip(&extra_segments).enumerate() { + #[cfg(feature = "test-hooks")] + let fail_after = (base.turbo_chunk_failure == Some(index + 1)).then_some(2); + #[cfg(not(feature = "test-hooks"))] + let fail_after = { + let _ = index; + None + }; + let preroll = preroll_frames(fps, frames.start); + let rendered = frames.start - preroll..frames.end; + let (frame_tx, frame_rx) = mpsc::sync_channel::(4); + let (result_tx, result_rx) = mpsc::sync_channel(1); + let worker_progress = progress.clone(); + encoders.push( + std::thread::Builder::new() + .name("turbo-encode".to_string()) + .spawn(move || { + let result = encode_chunk( + frame_rx, + encoder_settings, + &worker_progress, + preroll, + fail_after, + ); + if let Err(error) = &result + && error != STOPPED + { + worker_progress.fail(error.clone()); + } + let _ = result_tx.send(result); + }) + .map_err(spawn_failure)?, + ); + renders.push( + render_chunk( + shared.clone(), + render_segments(segments), + cap_rendering::FrameWindows::new(vec![rendered]), + preroll, + frame_tx, + progress.clone(), + None, + ) + .map_err(spawn_failure)?, + ); + later_chunks.push((frames.start, result_rx)); + } + + let encoder_output_path = temporary_output + .as_ref() + .map(|path| path.to_path_buf()) + .unwrap_or_else(|| output_path.clone()); + let stitch_input = StitchInput { + path: encoder_output_path, + settings: encoder_settings, + total_frames, + first_chunk, + later_chunks, + audio, + streaming_audio, + sample_timing: base.sample_timing.clone(), + }; + let stitch_progress = progress.clone(); + let result = tokio::task::spawn_blocking(move || { + let stitched = stitch(stitch_input, &stitch_progress); + if let Err( + Mp4PipelineError::Failure(error) + | Mp4PipelineError::AudioSourceFailure { message: error, .. }, + ) = &stitched + { + stitch_progress.fail(error.clone()); + } + let mut errors = vec![stitched.err()]; + for render in renders { + errors.push(match render.join() { + Ok(Ok(())) => None, + Ok(Err(error)) if error == STOPPED => Some(Mp4PipelineError::Interrupted), + Ok(Err(error)) => Some(Mp4PipelineError::Failure(error)), + Err(_) => Some(Mp4PipelineError::Failure( + "Turbo render thread panicked".to_string(), + )), + }); + } + for encoder in encoders { + if encoder.join().is_err() { + errors.push(Some(Mp4PipelineError::Failure( + "Turbo encoder thread panicked".to_string(), + ))); + } + } + errors + }) + .await + .map_err(|error| TurboError::Failed(error.to_string()))?; + let validation = match audio_validation { + Some(task) => task + .await + .map_err(|error| Mp4PipelineError::Failure(error.to_string())) + .and_then(|result| result) + .err(), + None => None, + }; + drop(extra_segments); + + let mut errors = result.into_iter().flatten().chain(validation); + let error = errors.find(|error| !matches!(error, Mp4PipelineError::Interrupted)); + if progress.was_cancelled() { + return Err(TurboError::Cancelled); + } + let first_failure = progress + .failure + .lock() + .unwrap_or_else(PoisonError::into_inner) + .take(); + if let Some(error) = first_failure { + return Err(TurboError::Failed(error)); + } + if let Some(error) = error { + return Err(TurboError::Failed(error.message())); + } + if progress.stopped() { + return Err(TurboError::Failed( + "Turbo export stopped before it finished".to_string(), + )); + } + if let Some(temporary_output) = temporary_output { + temporary_output + .persist_noclobber(&output_path) + .map_err(|error| TurboError::Failed(error.to_string()))?; + } + drop(audio_control); + Ok(output_path) +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn plans_cover_every_frame_in_order() { + for (total, fps, cores, requested) in [ + (2312, 60, 10, None), + (2312, 60, 10, Some(3)), + (100, 30, 16, None), + (7, 30, 16, Some(4)), + (100_000, 60, 64, None), + ] { + let plan = plan(total, fps, cores, requested); + assert!(!plan.chunks.is_empty() && plan.chunks.len() <= MAX_CHUNKS); + assert_eq!(plan.chunks[0].start, 0); + assert_eq!(plan.chunks.last().unwrap().end, total); + for pair in plan.chunks.windows(2) { + assert_eq!(pair[0].end, pair[1].start); + assert!(pair[0].len().abs_diff(pair[1].len()) <= 1); + } + assert!(plan.threads >= 1); + } + } + + #[test] + fn short_exports_stay_in_one_chunk_unless_asked() { + assert_eq!(plan(60 * 7, 60, 16, None).chunks.len(), 1); + assert_eq!(plan(60 * 8, 60, 16, None).chunks.len(), 2); + assert_eq!(plan(60 * 7, 60, 16, Some(3)).chunks.len(), 3); + assert_eq!(plan(2, 60, 16, Some(3)).chunks, vec![0..1, 1..2]); + assert_eq!(plan(0, 60, 16, None).chunks, vec![0..0]); + } + + #[test] + fn preroll_covers_mask_smoothing_without_reaching_before_the_timeline() { + assert_eq!(preroll_frames(60, 1000), 30); + assert_eq!(preroll_frames(15, 1000), 16); + assert_eq!(preroll_frames(30, 10), 10); + assert_eq!(preroll_frames(30, 0), 0); + } + + #[test] + fn constant_quality_follows_the_presets() { + for (compression, crf) in [ + (ExportCompression::Maximum, 15), + (ExportCompression::Social, 18), + (ExportCompression::Web, 20), + (ExportCompression::Potato, 22), + ] { + assert_eq!(constant_quality(compression.bits_per_pixel(), 30), crf); + assert_eq!(constant_quality(compression.bits_per_pixel(), 60), crf - 3); + assert_eq!(constant_quality(compression.bits_per_pixel(), 15), crf + 2); + } + assert_eq!(constant_quality(1.0, 30), 15); + assert_eq!(constant_quality(0.01, 30), 22); + assert_eq!(constant_quality(0.0, 30), 22); + assert_eq!(constant_quality(0.3, 120), 12); + assert!((15..=22).contains(&constant_quality(0.2, 30))); + } + + #[test] + fn parameter_sets_compare_sps_and_pps_only() { + let unit = |nal: &[u8]| [&[0u8, 0, 0, 1][..], nal].concat(); + let sps = [0x67, 0x64, 0x00, 0x28]; + let pps = [0x68, 0xee, 0x3c, 0x80]; + let sei = [0x06, 0x05, 0x01]; + let first = [unit(&sps), unit(&pps), unit(&sei), unit(&[0x65, 1, 2])].concat(); + let other = [ + unit(&sps), + unit(&pps), + unit(&[0x06, 0x09]), + unit(&[0x65, 3]), + ] + .concat(); + assert_eq!(parameter_sets(&first), vec![&sps[..], &pps[..]]); + assert!(same_parameter_sets(Some(&first), Some(&other))); + let changed = [unit(&[0x67, 0x4d, 0x00, 0x28]), unit(&pps)].concat(); + assert!(!same_parameter_sets(Some(&first), Some(&changed))); + assert!(!same_parameter_sets(Some(&first), None)); + assert!(!same_parameter_sets(Some(&[0x65, 1]), Some(&[0x65, 1]))); + } +} From 3032bdeb9eb193c67083c62c7a2c17384b639026 Mon Sep 17 00:00:00 2001 From: Richie McIlroy <33632126+richiemcilroy@users.noreply.github.com> Date: Tue, 27 Oct 2026 14:39:26 +0000 Subject: [PATCH 3/9] test(verify): cover turbo export seams, fallback and cancellation --- crates/verify/tests/common/mod.rs | 45 +++++++- crates/verify/tests/export.rs | 174 +++++++++++++++++++++++++++++- scripts/verify/suites.json | 1 + 3 files changed, 216 insertions(+), 4 deletions(-) diff --git a/crates/verify/tests/common/mod.rs b/crates/verify/tests/common/mod.rs index 78b5c766658..85ca09d7a66 100644 --- a/crates/verify/tests/common/mod.rs +++ b/crates/verify/tests/common/mod.rs @@ -133,6 +133,12 @@ pub fn add_transition(config: &mut ProjectConfiguration, segment_index: u32, dur #[derive(Clone, Copy, Debug, PartialEq, Eq)] pub enum Format { Mp4, + /// MP4 through the turbo export split into `chunks`. With a failing + /// chunk, the export has to fall back to the standard pipeline. + Mp4Turbo { + chunks: usize, + failing_chunk: Option, + }, Gif, Mov, } @@ -279,6 +285,42 @@ pub async fn export( out: &Path, ) -> Result { let force = run.decoder == Decoder::Ffmpeg; + let resolution_base = XY::new(run.resolution.0, run.resolution.1); + if let Format::Mp4Turbo { + chunks, + failing_chunk, + } = run.format + { + let mut base = cap_export::ExporterBase::builder(project.path.clone()) + .with_config(config) + .with_force_ffmpeg_decoder(force) + .with_output_path(out.to_path_buf()) + .build_for_mp4(std::sync::Arc::new(std::sync::atomic::AtomicBool::new( + false, + ))) + .await + .map_err(|e| format!("exporter build: {e}"))?; + if let Some(chunk) = failing_chunk { + base.fail_turbo_chunk(chunk); + } + return cap_export::mp4::Mp4ExportSettings { + fps: run.fps, + resolution_base, + compression: cap_export::mp4::ExportCompression::Social, + custom_bpp: run.custom_bpp, + force_ffmpeg_decoder: force, + optimize_filesize: false, + } + .export_prepared_turbo( + base, + cap_export::mp4::turbo::TurboOptions { + chunks: Some(chunks), + strict: failing_chunk.is_none(), + }, + |_| true, + ) + .await; + } let base = cap_export::ExporterBase::builder(project.path.clone()) .with_config(config) .with_force_ffmpeg_decoder(force) @@ -286,9 +328,8 @@ pub async fn export( .build() .await .map_err(|e| format!("exporter build: {e}"))?; - let resolution_base = XY::new(run.resolution.0, run.resolution.1); match run.format { - Format::Mp4 => { + Format::Mp4 | Format::Mp4Turbo { .. } => { cap_export::mp4::Mp4ExportSettings { fps: run.fps, resolution_base, diff --git a/crates/verify/tests/export.rs b/crates/verify/tests/export.rs index a06af620f84..79d13b63665 100644 --- a/crates/verify/tests/export.rs +++ b/crates/verify/tests/export.rs @@ -21,6 +21,11 @@ fn no_edit(_: &mut ProjectConfiguration) {} fn case_id(fixture: &str, edit: &str, format: Format, fps: u32, decoder: Decoder) -> String { let format = match format { Format::Mp4 => "mp4", + Format::Mp4Turbo { + failing_chunk: None, + .. + } => "mp4-turbo", + Format::Mp4Turbo { .. } => "mp4-turbo-fallback", Format::Gif => "gif", Format::Mov => "mov", }; @@ -75,7 +80,7 @@ fn run_project( let ext = match format { Format::Gif => "gif", Format::Mov => "mov", - Format::Mp4 => "mp4", + Format::Mp4 | Format::Mp4Turbo { .. } => "mp4", }; let run = ExportRun { custom_bpp: None, @@ -104,7 +109,7 @@ fn run_project( &frames, &path, fps, - format == Format::Mp4, + matches!(format, Format::Mp4 | Format::Mp4Turbo { .. }), ); if determinism { let second = dir.path().join(format!("b.{ext}")); @@ -268,6 +273,171 @@ fn basic() { common::assert_passed(failures); } +/// The turbo export split into three chunks, so chunk seams land inside +/// cuts, speed changes, transitions, scene changes and held frames. +#[test] +fn turbo() { + let turbo = Format::Mp4Turbo { + chunks: 3, + failing_chunk: None, + }; + let fallback = Format::Mp4Turbo { + chunks: 3, + failing_chunk: Some(2), + }; + let mut failures: Vec = Vec::new(); + failures.extend(run( + catalog::basic(report::tier()), + &[ + Variant { + edit: "full", + apply: no_edit, + }, + Variant { + edit: "cuts-speeds-reorder", + apply: cuts_speeds_reorder, + }, + Variant { + edit: "transition", + apply: transition, + }, + Variant { + edit: "hide-camera-scene", + apply: hide_camera_scene, + }, + ], + &[(turbo, 30), (turbo, 60)], + &Decoder::all(), + RES, + false, + )); + failures.extend(run( + catalog::vfr_holds(report::tier()), + &[Variant { + edit: "full", + apply: no_edit, + }], + &[(turbo, 60)], + &Decoder::all(), + RES, + false, + )); + failures.extend(run( + catalog::basic(report::tier()), + &[Variant { + edit: "cuts-speeds-reorder", + apply: cuts_speeds_reorder, + }], + &[(fallback, 30)], + &[Decoder::Default], + RES, + false, + )); + failures.extend(turbo_cancellation()); + common::assert_passed(failures); +} + +fn turbo_spool_files() -> Vec { + std::fs::read_dir(std::env::temp_dir()) + .map(|entries| { + entries + .flatten() + .map(|entry| entry.path()) + .filter(|path| { + path.file_name() + .and_then(|name| name.to_str()) + .is_some_and(|name| name.starts_with("cap-turbo-")) + }) + .collect() + }) + .unwrap_or_default() +} + +/// Stopping a turbo export from the progress callback or the user's cancel +/// flag ends every chunk, leaves no output and removes the chunk spools. +fn turbo_cancellation() -> Vec { + use std::sync::{ + Arc, + atomic::{AtomicBool, Ordering}, + }; + let spec = catalog::vfr_holds(report::tier()); + let mut failures = Vec::new(); + for by_flag in [false, true] { + let id = format!( + "export/{}/turbo-cancel-{}", + spec.name, + if by_flag { "flag" } else { "progress" } + ); + if !report::selected(&id) { + continue; + } + let project = common::project(&spec); + let Some(mut case) = common::begin(&id, &project, Decoder::Default) else { + continue; + }; + let dir = common::scratch_dir(&id); + let out = dir.path().join("cancelled.mp4"); + let spools_before = turbo_spool_files(); + let cancel = Arc::new(AtomicBool::new(false)); + let flag = cancel.clone(); + let result = common::runtime().block_on(async { + let base = cap_export::ExporterBase::builder(project.path.clone()) + .with_config(project.config()) + .with_output_path(out.clone()) + .build_for_mp4(cancel) + .await + .map_err(|e| format!("exporter build: {e}"))?; + cap_export::mp4::Mp4ExportSettings { + fps: 60, + resolution_base: XY::new(RES.0, RES.1), + compression: cap_export::mp4::ExportCompression::Social, + custom_bpp: None, + force_ffmpeg_decoder: false, + optimize_filesize: false, + } + .export_prepared_turbo( + base, + cap_export::mp4::turbo::TurboOptions { + chunks: Some(3), + strict: true, + }, + move |frame| { + if frame < 60 { + return true; + } + if by_flag { + flag.store(true, Ordering::Relaxed); + true + } else { + false + } + }, + ) + .await + }); + if result.as_ref().map_err(String::as_str) != Err("Export cancelled") { + case.problem("error", format!("cancelled export returned {result:?}")); + } + if out.exists() { + case.problem("error", "cancelled export left an output file"); + } + let leftover: Vec<_> = turbo_spool_files() + .into_iter() + .filter(|path| !spools_before.contains(path)) + .collect(); + if !leftover.is_empty() { + case.problem( + "error", + format!("cancelled export left spools {leftover:?}"), + ); + } + if let Err(message) = report::settle(case) { + failures.push(message); + } + } + failures +} + #[test] fn gif() { let mut failures: Vec = Vec::new(); diff --git a/scripts/verify/suites.json b/scripts/verify/suites.json index 8283185d0e4..18cae0eaf86 100644 --- a/scripts/verify/suites.json +++ b/scripts/verify/suites.json @@ -29,6 +29,7 @@ "macos": ["export/*/ffmpeg", "export/backward-hold/*/mp430/*"], "windows": [ "export/*/mp460/*", + "export/*/mp4-turbo60/*", "export/backward-hold/*", "export/jitter-camera-*", "export/*/uhd-*" From 4d25abf0f1867e8eb425d6f45163620ee8842490 Mon Sep 17 00:00:00 2001 From: Richie McIlroy <33632126+richiemcilroy@users.noreply.github.com> Date: Tue, 27 Oct 2026 14:39:26 +0000 Subject: [PATCH 4/9] feat(desktop-gpui): add Turbo export toggle to the export page --- apps/desktop-gpui/src/editor_export.rs | 121 ++++++++++++++++++++++--- apps/desktop-gpui/src/store.rs | 2 + 2 files changed, 108 insertions(+), 15 deletions(-) diff --git a/apps/desktop-gpui/src/editor_export.rs b/apps/desktop-gpui/src/editor_export.rs index e3c53ce48ab..16c62ea2a93 100644 --- a/apps/desktop-gpui/src/editor_export.rs +++ b/apps/desktop-gpui/src/editor_export.rs @@ -5,10 +5,10 @@ use std::sync::Arc; use std::sync::atomic::{AtomicBool, Ordering}; use std::time::Duration; -use cap_export::estimates::{ExportEstimates, estimate_export}; +use cap_export::estimates::{ExportEstimates, estimate_export, estimate_turbo_export}; use cap_export::gif::GifExportSettings; use cap_export::mov::MovExportSettings; -use cap_export::mp4::{ExportCompression, Mp4ExportSettings}; +use cap_export::mp4::{ExportCompression, Mp4ExportSettings, turbo::TurboOptions}; use cap_export::preview::{ ExportPreviewSettings, render_preview_with_config, render_preview_with_editor, }; @@ -189,6 +189,7 @@ pub struct ExportUi { pub fps: u32, pub compression: ExportCompression, pub optimize_filesize: bool, + pub turbo: bool, pub cursor_only: bool, pub custom_bpp: Option, pub force_ffmpeg: bool, @@ -262,6 +263,7 @@ impl ExportUi { force_ffmpeg_decoder: false, advanced_open: false, organization_id: None, + turbo: false, }); Self::from_preferences(prefs) } @@ -279,6 +281,7 @@ impl ExportUi { _ => ExportCompression::Maximum, }, optimize_filesize: prefs.optimize_filesize, + turbo: prefs.turbo && turbo_available(), cursor_only: prefs.cursor_only, custom_bpp: prefs.custom_bpp, force_ffmpeg: prefs.force_ffmpeg_decoder, @@ -332,6 +335,7 @@ impl ExportUi { force_ffmpeg_decoder: self.force_ffmpeg, advanced_open: self.advanced_open, organization_id: self.organization_id.clone(), + turbo: self.turbo, }; store::update(|state| state.export = Some(prefs)); } @@ -662,6 +666,7 @@ impl EditorWindow { ui.preview_stats = None; let estimate_cancel = ui.estimate_cancel.clone(); let estimate_settings = ui.estimate_settings(); + let estimate_turbo = ui.turbo; let estimate_instance = instance.clone(); let estimate_project = project.clone(); let (width, height) = ui.resolution.size(); @@ -735,16 +740,28 @@ impl EditorWindow { let result = if let Some(instance) = estimate_instance { let (estimate_tx, estimate_rx) = flume::unbounded(); let estimate_task = gpui_tokio::Tokio::spawn(cx, async move { - estimate_export( - instance, - estimate_project, - estimate_settings, - estimate_cancel, - move |estimate| { - let _ = estimate_tx.send(estimate); - }, - ) - .await + let on_estimate = move |estimate| { + let _ = estimate_tx.send(estimate); + }; + if estimate_turbo { + estimate_turbo_export( + instance, + estimate_project, + estimate_settings, + estimate_cancel, + on_estimate, + ) + .await + } else { + estimate_export( + instance, + estimate_project, + estimate_settings, + estimate_cancel, + on_estimate, + ) + .await + } }); while let Ok(estimate) = estimate_rx.recv_async().await { let _ = this.update(cx, |this, cx| { @@ -821,6 +838,7 @@ impl EditorWindow { let compression = ui.compression; let custom_bpp = ui.custom_bpp; let optimize = ui.optimize_filesize; + let turbo = ui.turbo; let force = ui.force_ffmpeg; let project_path = self.project_path.clone(); let project = self.project.clone(); @@ -935,6 +953,7 @@ impl EditorWindow { compression, custom_bpp, optimize, + turbo, force, save_path.clone(), progress_tx, @@ -1109,6 +1128,7 @@ impl EditorWindow { let compression = ui.compression; let custom_bpp = ui.custom_bpp; let optimize = ui.optimize_filesize; + let turbo = ui.turbo; let force = ui.force_ffmpeg; let project_path = self.project_path.clone(); let project = self.project.clone(); @@ -1178,6 +1198,7 @@ impl EditorWindow { compression, custom_bpp, optimize, + turbo, force, save_path.clone(), progress_tx, @@ -2147,6 +2168,9 @@ impl EditorWindow { .child("Larger file"), ) .child(self.export_optimize_row(ui, cx)) + .when(turbo_available(), |this| { + this.child(self.export_turbo_row(ui, cx)) + }) } fn export_toggle_row( @@ -2206,6 +2230,59 @@ impl EditorWindow { ) } + fn export_turbo_row(&self, ui: &ExportUi, cx: &mut Context) -> impl IntoElement { + let editor = self.theme.editor; + div() + .flex() + .flex_col() + .gap(px(4.)) + .child(self.export_toggle_row( + "export-turbo", + "Turbo export", + "Exports much faster using every CPU core and the GPU", + ui.turbo, + cx.listener(|this, _, window, cx| { + if let Some(ui) = this.export.as_mut() { + ui.turbo = !ui.turbo; + ui.persist(); + } + this.refresh_export_preview(window, cx); + cx.notify(); + }), + )) + .when(ui.turbo, |this| { + this.child( + div() + .p(px(12.)) + .rounded(px(10.)) + .bg(Hsla::from(editor.card_2)) + .flex() + .flex_row() + .items_start() + .gap(px(8.)) + .child( + svg() + .path("icons/triangle-alert.svg") + .size(px(14.)) + .flex_shrink_0() + .mt(px(1.)) + .text_color(Hsla::from(editor.text_2)), + ) + .child( + div() + .flex_1() + .min_w_0() + .text_size(px(11.5)) + .line_height(px(15.)) + .text_color(Hsla::from(editor.text_2)) + .child( + "Uses all of your CPU, the GPU and several times more memory until the export finishes. Other apps may slow down, and your computer may get warm and loud.", + ), + ), + ) + }) + } + fn export_advanced(&self, ui: &ExportUi, cx: &mut Context) -> impl IntoElement { let theme = self.theme; let editor = theme.editor; @@ -2896,6 +2973,13 @@ fn export_stat( }) } +/// Turbo trades the hardware encoder for libx264 on every core, which is +/// only measured faster on macOS so far. +fn turbo_available() -> bool { + cfg!(target_os = "macos") + && std::thread::available_parallelism().is_ok_and(|cores| cores.get() >= 8) +} + fn matches_compression(current: ExportCompression, expected: ExportCompression) -> bool { std::mem::discriminant(¤t) == std::mem::discriminant(&expected) } @@ -2912,6 +2996,7 @@ async fn run_export( compression: ExportCompression, custom_bpp: Option, optimize: bool, + turbo: bool, force: bool, save_path: Option, progress_tx: flume::Sender<(u32, u32)>, @@ -2985,16 +3070,21 @@ async fn run_export( .await } PreparedBase::Mp4(base) => { - Mp4ExportSettings { + let settings = Mp4ExportSettings { fps, resolution_base: resolution, compression, custom_bpp, force_ffmpeg_decoder: force, optimize_filesize: optimize, + }; + if turbo { + settings + .export_prepared_turbo(base, TurboOptions::default(), progress) + .await + } else { + settings.export_prepared(base, progress).await } - .export_prepared(base, progress) - .await } } } @@ -3024,6 +3114,7 @@ mod tests { force_ffmpeg_decoder: false, advanced_open: false, organization_id: None, + turbo: false, }) } diff --git a/apps/desktop-gpui/src/store.rs b/apps/desktop-gpui/src/store.rs index fccc0918df6..d2b83d7c0de 100644 --- a/apps/desktop-gpui/src/store.rs +++ b/apps/desktop-gpui/src/store.rs @@ -148,6 +148,8 @@ pub struct ExportPrefs { pub advanced_open: bool, #[serde(default)] pub organization_id: Option, + #[serde(default)] + pub turbo: bool, } /// `so.cap.desktop`'s app-data dir -- where both stores live. From f4a4a579a754304f90b89c0fe1a15d19976dec4b Mon Sep 17 00:00:00 2001 From: Richie McIlroy <33632126+richiemcilroy@users.noreply.github.com> Date: Tue, 27 Oct 2026 14:54:26 +0000 Subject: [PATCH 5/9] fix(export): keep turbo's staged output owned until it is published --- crates/export/src/mp4.rs | 4 +++- crates/export/src/mp4/turbo.rs | 24 ++++++++++++++++-------- 2 files changed, 19 insertions(+), 9 deletions(-) diff --git a/crates/export/src/mp4.rs b/crates/export/src/mp4.rs index 90be5f8c2b6..1665333af9a 100644 --- a/crates/export/src/mp4.rs +++ b/crates/export/src/mp4.rs @@ -212,7 +212,8 @@ impl Mp4ExportSettings { } /// Exports with every core: see [`turbo`]. Falls back to the standard - /// export when the turbo pipeline fails for any reason but cancellation. + /// export when the turbo pipeline fails, unless it was cancelled or its + /// destination appeared while it ran. pub async fn export_prepared_turbo( self, base: Mp4ExporterBase, @@ -228,6 +229,7 @@ impl Mp4ExportSettings { match turbo::export(self, &mut base, options, on_progress.clone()).await { Ok(path) => Ok(path), Err(turbo::TurboError::Cancelled) => Err("Export cancelled".to_string()), + Err(turbo::TurboError::Publish(error)) => Err(error), Err(turbo::TurboError::Failed(error)) if options.strict => Err(error), Err(turbo::TurboError::Failed(error)) => { warn!(%error, "Turbo export failed, exporting with the standard pipeline"); diff --git a/crates/export/src/mp4/turbo.rs b/crates/export/src/mp4/turbo.rs index 48f20c9083e..47c360d15d4 100644 --- a/crates/export/src/mp4/turbo.rs +++ b/crates/export/src/mp4/turbo.rs @@ -44,6 +44,9 @@ pub struct TurboOptions { pub(crate) enum TurboError { Cancelled, Failed(String), + /// The export finished but could not be moved to its destination, which + /// something else created meanwhile. Exporting again would overwrite it. + Publish(String), } impl From for TurboError { @@ -162,7 +165,6 @@ impl bool> Progress { } } -/// Everything a chunk's render loop reads, shared by all chunks. #[derive(Clone)] struct RenderShared { constants: Arc, @@ -262,7 +264,6 @@ fn render_chunk bool + Send + 'static>( }) } -/// Copies a rendered frame into the encoder's NV12 input frame. fn fill_input( input: &ExportFrame, frame: &mut ffmpeg::frame::Video, @@ -392,7 +393,6 @@ fn read_spooled_packets( } } -/// Encodes a chunk into its spool. fn encode_chunk bool>( frames: mpsc::Receiver, settings: EncoderSettings, @@ -645,6 +645,10 @@ fn stitch bool>( .is_some_and(|user| user.load(Ordering::Relaxed)) }, )?; + // Audio runs ahead of the video on purpose: later chunks' spools + // append far faster than they encode, so audio rendered while + // chunk 0 encodes keeps them from waiting on it. The queued AAC + // packets cost at most 40 KB per second of timeline. if !sampled { for frame in 0..total_frames { worker.request(frame, frame); @@ -735,7 +739,7 @@ fn stitch bool>( }) .map_err(|error| match error { TurboError::Cancelled => Mp4PipelineError::Interrupted, - TurboError::Failed(error) => failure(error), + TurboError::Failed(error) | TurboError::Publish(error) => failure(error), })?; } if !sampled && written != total_frames { @@ -778,7 +782,7 @@ pub(crate) async fn export_sampled( { Ok(path) => Ok(path), Err(TurboError::Cancelled) => Err(STOPPED.to_string()), - Err(TurboError::Failed(error)) => Err(error), + Err(TurboError::Failed(error) | TurboError::Publish(error)) => Err(error), } } @@ -852,7 +856,8 @@ pub(crate) async fn export bool + Send + 'static>( let temporary_output = match base.streaming_output.take() { Some(staged) => Some(staged), None => crate::prepare_new_output(&base.output_path).ok().flatten(), - }; + } + .map(Arc::new); let user_cancellation = base.user_cancellation.clone(); let stop = audio_control .as_ref() @@ -1004,7 +1009,9 @@ pub(crate) async fn export bool + Send + 'static>( sample_timing: base.sample_timing.clone(), }; let stitch_progress = progress.clone(); + let stitch_temporary_output = temporary_output.clone(); let result = tokio::task::spawn_blocking(move || { + let _temporary_output_guard = stitch_temporary_output; let stitched = stitch(stitch_input, &stitch_progress); if let Err( Mp4PipelineError::Failure(error) @@ -1067,9 +1074,10 @@ pub(crate) async fn export bool + Send + 'static>( )); } if let Some(temporary_output) = temporary_output { - temporary_output + Arc::try_unwrap(temporary_output) + .map_err(|_| TurboError::Failed("Export output is still in use".to_string()))? .persist_noclobber(&output_path) - .map_err(|error| TurboError::Failed(error.to_string()))?; + .map_err(|error| TurboError::Publish(error.to_string()))?; } drop(audio_control); Ok(output_path) From bd47cc7cd4e476a25cf8e4fc5dd86400bc548ca4 Mon Sep 17 00:00:00 2001 From: Richie McIlroy <33632126+richiemcilroy@users.noreply.github.com> Date: Tue, 27 Oct 2026 14:54:26 +0000 Subject: [PATCH 6/9] test(verify): cover a destination created during turbo export --- crates/verify/tests/export.rs | 60 +++++++++++++++++++++++++---------- 1 file changed, 43 insertions(+), 17 deletions(-) diff --git a/crates/verify/tests/export.rs b/crates/verify/tests/export.rs index 79d13b63665..701c6c3753f 100644 --- a/crates/verify/tests/export.rs +++ b/crates/verify/tests/export.rs @@ -354,25 +354,32 @@ fn turbo_spool_files() -> Vec { } /// Stopping a turbo export from the progress callback or the user's cancel -/// flag ends every chunk, leaves no output and removes the chunk spools. +/// flag ends every chunk, leaves no output and removes the chunk spools. A +/// destination that appears while turbo runs is left alone: the export fails +/// rather than falling back to a standard export that would overwrite it. fn turbo_cancellation() -> Vec { use std::sync::{ Arc, atomic::{AtomicBool, Ordering}, }; let spec = catalog::vfr_holds(report::tier()); + let decoder = if cfg!(target_os = "windows") { + Decoder::Ffmpeg + } else { + Decoder::Default + }; let mut failures = Vec::new(); - for by_flag in [false, true] { + for mode in ["progress", "flag", "taken"] { let id = format!( - "export/{}/turbo-cancel-{}", + "export/{}/turbo-cancel-{mode}/{}", spec.name, - if by_flag { "flag" } else { "progress" } + decoder.label() ); if !report::selected(&id) { continue; } let project = common::project(&spec); - let Some(mut case) = common::begin(&id, &project, Decoder::Default) else { + let Some(mut case) = common::begin(&id, &project, decoder) else { continue; }; let dir = common::scratch_dir(&id); @@ -380,9 +387,11 @@ fn turbo_cancellation() -> Vec { let spools_before = turbo_spool_files(); let cancel = Arc::new(AtomicBool::new(false)); let flag = cancel.clone(); + let taken = out.clone(); let result = common::runtime().block_on(async { let base = cap_export::ExporterBase::builder(project.path.clone()) .with_config(project.config()) + .with_force_ffmpeg_decoder(decoder == Decoder::Ffmpeg) .with_output_path(out.clone()) .build_for_mp4(cancel) .await @@ -392,34 +401,51 @@ fn turbo_cancellation() -> Vec { resolution_base: XY::new(RES.0, RES.1), compression: cap_export::mp4::ExportCompression::Social, custom_bpp: None, - force_ffmpeg_decoder: false, + force_ffmpeg_decoder: decoder == Decoder::Ffmpeg, optimize_filesize: false, } .export_prepared_turbo( base, cap_export::mp4::turbo::TurboOptions { chunks: Some(3), - strict: true, + strict: mode != "taken", }, move |frame| { if frame < 60 { return true; } - if by_flag { - flag.store(true, Ordering::Relaxed); - true - } else { - false + match mode { + "flag" => { + flag.store(true, Ordering::Relaxed); + true + } + "taken" => { + if frame == 60 { + let _ = std::fs::write(&taken, "taken"); + } + true + } + _ => false, } }, ) .await }); - if result.as_ref().map_err(String::as_str) != Err("Export cancelled") { - case.problem("error", format!("cancelled export returned {result:?}")); - } - if out.exists() { - case.problem("error", "cancelled export left an output file"); + if mode == "taken" { + if result.is_ok() || result.as_ref().map_err(String::as_str) == Err("Export cancelled") + { + case.problem("error", format!("taken destination returned {result:?}")); + } + if std::fs::read(&out).ok().as_deref() != Some(&b"taken"[..]) { + case.problem("error", "export overwrote a destination created meanwhile"); + } + } else { + if result.as_ref().map_err(String::as_str) != Err("Export cancelled") { + case.problem("error", format!("cancelled export returned {result:?}")); + } + if out.exists() { + case.problem("error", "cancelled export left an output file"); + } } let leftover: Vec<_> = turbo_spool_files() .into_iter() From 251a431832a867863a893b20b97366dceb843718 Mon Sep 17 00:00:00 2001 From: Richie McIlroy <33632126+richiemcilroy@users.noreply.github.com> Date: Tue, 27 Oct 2026 15:09:26 +0000 Subject: [PATCH 7/9] fix(rendering): wait for a map callback another renderer is about to run --- crates/rendering/src/frame_pipeline.rs | 91 +++++++++++++++++++++----- 1 file changed, 76 insertions(+), 15 deletions(-) diff --git a/crates/rendering/src/frame_pipeline.rs b/crates/rendering/src/frame_pipeline.rs index 7660f4dc6fb..2b9fb639c81 100644 --- a/crates/rendering/src/frame_pipeline.rs +++ b/crates/rendering/src/frame_pipeline.rs @@ -106,6 +106,25 @@ async fn wait_for_submission( } } +/// The result of a `map_async` whose submission has finished. wgpu runs map +/// callbacks on whichever thread's poll collected them, so when several +/// renderers share a device, another thread can still be about to run this +/// buffer's callback after this thread's own wait returned. +async fn mapped_result( + rx: &mut oneshot::Receiver>, + started: Instant, + frame_number: u32, + timeout: std::time::Duration, +) -> Option> { + match rx.try_recv() { + Ok(result) => return Some(result), + Err(oneshot::error::TryRecvError::Closed) => return None, + Err(oneshot::error::TryRecvError::Empty) => {} + } + let remaining = timeout.checked_sub(readback_waited(started, frame_number))?; + tokio::time::timeout(remaining, rx).await.ok()?.ok() +} + #[derive(Clone, Copy, Debug, Default)] pub struct FinishEncoderTimings { pub wait_previous_duration: std::time::Duration, @@ -1147,17 +1166,18 @@ impl PendingNv12Readback { return Err(self.cancel()); }; - let completed = wait_for_submission( - device, - &submission, - Instant::now(), - self.frame_number, - gpu_buffer_wait_timeout(self.export_deadlines), - ) - .await; - match (completed, rx.try_recv()) { - (true, Ok(Ok(()))) => {} - (_, Ok(Err(error))) => { + let started = Instant::now(); + let timeout = gpu_buffer_wait_timeout(self.export_deadlines); + let completed = + wait_for_submission(device, &submission, started, self.frame_number, timeout).await; + let mapped = if completed { + mapped_result(&mut rx, started, self.frame_number, timeout).await + } else { + rx.try_recv().ok() + }; + match (completed, mapped) { + (true, Some(Ok(()))) => {} + (_, Some(Err(error))) => { self.unmap(); return Err(RenderingError::BufferMapFailed(error)); } @@ -1712,17 +1732,23 @@ impl PendingReadback { recycled: &mut RecycledReadbackFrames, ) -> Result { let start_time = Instant::now(); + let timeout = gpu_buffer_wait_timeout(self.export_deadlines); let completed = wait_for_submission( device, &self.submission, start_time, self.frame_number, - gpu_buffer_wait_timeout(self.export_deadlines), + timeout, ) .await; - match (completed, self.rx.try_recv()) { - (true, Ok(Ok(()))) => {} - (_, Ok(Err(error))) => { + let mapped = if completed { + mapped_result(&mut self.rx, start_time, self.frame_number, timeout).await + } else { + self.rx.try_recv().ok() + }; + match (completed, mapped) { + (true, Some(Ok(()))) => {} + (_, Some(Err(error))) => { self.unmap(); return Err(RenderingError::BufferMapFailed(error)); } @@ -2403,6 +2429,41 @@ pub async fn flush_pending_readback( mod readback_output_tests { use super::*; + #[tokio::test] + async fn map_callback_run_by_another_thread_after_the_wait_still_counts() { + let timeout = std::time::Duration::from_secs(5); + let (tx, mut rx) = oneshot::channel(); + let late = std::thread::spawn(move || { + std::thread::sleep(std::time::Duration::from_millis(50)); + tx.send(Ok(())).expect("map receiver"); + }); + assert!(matches!( + mapped_result(&mut rx, Instant::now(), 0, timeout).await, + Some(Ok(())) + )); + late.join().unwrap(); + + let (tx, mut rx) = oneshot::channel::>(); + drop(tx); + assert!( + mapped_result(&mut rx, Instant::now(), 0, timeout) + .await + .is_none() + ); + + let (_tx, mut rx) = oneshot::channel::>(); + assert!( + mapped_result( + &mut rx, + Instant::now(), + 0, + std::time::Duration::from_millis(20) + ) + .await + .is_none() + ); + } + #[test] fn readback_allocation_reserves_frame_and_metadata_without_changing_length() { let mut data = Vec::new(); From d4870f02a68cb74935e84471332f8d76af7ea8ae Mon Sep 17 00:00:00 2001 From: Richie McIlroy <33632126+richiemcilroy@users.noreply.github.com> Date: Tue, 27 Oct 2026 15:49:14 +0000 Subject: [PATCH 8/9] fix(export): skip the turbo fallback when a file appeared at the destination --- crates/export/src/mp4.rs | 16 ++++++++++++++++ 1 file changed, 16 insertions(+) diff --git a/crates/export/src/mp4.rs b/crates/export/src/mp4.rs index 1665333af9a..0b9901cb4cf 100644 --- a/crates/export/src/mp4.rs +++ b/crates/export/src/mp4.rs @@ -226,6 +226,13 @@ impl Mp4ExportSettings { .audio_cancellation .as_ref() .map(|control| control.user.clone()); + let destination_exists = |path: &std::path::Path| { + std::fs::symlink_metadata(path).map_or_else( + |error| error.kind() != std::io::ErrorKind::NotFound, + |_| true, + ) + }; + let destination_was_free = !destination_exists(&base.output_path); match turbo::export(self, &mut base, options, on_progress.clone()).await { Ok(path) => Ok(path), Err(turbo::TurboError::Cancelled) => Err("Export cancelled".to_string()), @@ -234,6 +241,15 @@ impl Mp4ExportSettings { Err(turbo::TurboError::Failed(error)) => { warn!(%error, "Turbo export failed, exporting with the standard pipeline"); base.reopen_streaming(user_cancellation).await?; + // Turbo staged its output, but a standard export that is not + // staged writes in place and would truncate a file created + // at the destination meanwhile. + if destination_was_free + && base.streaming_output.is_none() + && destination_exists(&base.output_path) + { + return Err(error); + } let output_size = ProjectUniforms::get_output_size( &base.render_constants.options, &base.project_config, From b4c097b3ae9c59cd67c9b8604030d39ac77fce9e Mon Sep 17 00:00:00 2001 From: Richie McIlroy <33632126+richiemcilroy@users.noreply.github.com> Date: Tue, 27 Oct 2026 15:49:14 +0000 Subject: [PATCH 9/9] test(verify): cover a failed turbo export whose destination appeared meanwhile --- crates/verify/tests/export.rs | 38 ++++++++++++++++++----------------- 1 file changed, 20 insertions(+), 18 deletions(-) diff --git a/crates/verify/tests/export.rs b/crates/verify/tests/export.rs index 701c6c3753f..8c89931731b 100644 --- a/crates/verify/tests/export.rs +++ b/crates/verify/tests/export.rs @@ -355,8 +355,9 @@ fn turbo_spool_files() -> Vec { /// Stopping a turbo export from the progress callback or the user's cancel /// flag ends every chunk, leaves no output and removes the chunk spools. A -/// destination that appears while turbo runs is left alone: the export fails -/// rather than falling back to a standard export that would overwrite it. +/// destination that appears while turbo runs is left alone, whether turbo +/// then finishes or fails: the export fails rather than falling back to a +/// standard export that would overwrite it. fn turbo_cancellation() -> Vec { use std::sync::{ Arc, @@ -369,7 +370,7 @@ fn turbo_cancellation() -> Vec { Decoder::Default }; let mut failures = Vec::new(); - for mode in ["progress", "flag", "taken"] { + for mode in ["progress", "flag", "taken", "taken-failed"] { let id = format!( "export/{}/turbo-cancel-{mode}/{}", spec.name, @@ -389,13 +390,16 @@ fn turbo_cancellation() -> Vec { let flag = cancel.clone(); let taken = out.clone(); let result = common::runtime().block_on(async { - let base = cap_export::ExporterBase::builder(project.path.clone()) + let mut base = cap_export::ExporterBase::builder(project.path.clone()) .with_config(project.config()) .with_force_ffmpeg_decoder(decoder == Decoder::Ffmpeg) .with_output_path(out.clone()) .build_for_mp4(cancel) .await .map_err(|e| format!("exporter build: {e}"))?; + if mode == "taken-failed" { + base.fail_turbo_chunk(2); + } cap_export::mp4::Mp4ExportSettings { fps: 60, resolution_base: XY::new(RES.0, RES.1), @@ -408,30 +412,28 @@ fn turbo_cancellation() -> Vec { base, cap_export::mp4::turbo::TurboOptions { chunks: Some(3), - strict: mode != "taken", + strict: !mode.starts_with("taken"), }, move |frame| { + if mode.starts_with("taken") { + if !taken.exists() { + let _ = std::fs::write(&taken, "taken"); + } + return true; + } if frame < 60 { return true; } - match mode { - "flag" => { - flag.store(true, Ordering::Relaxed); - true - } - "taken" => { - if frame == 60 { - let _ = std::fs::write(&taken, "taken"); - } - true - } - _ => false, + if mode == "flag" { + flag.store(true, Ordering::Relaxed); + return true; } + false }, ) .await }); - if mode == "taken" { + if mode.starts_with("taken") { if result.is_ok() || result.as_ref().map_err(String::as_str) == Err("Export cancelled") { case.problem("error", format!("taken destination returned {result:?}"));