From 9e55b6dfcf2f4797bda1b3efd7d7c474d93b81ee Mon Sep 17 00:00:00 2001 From: vedanth Date: Sun, 25 Oct 2026 16:14:39 +0000 Subject: [PATCH] perf(editor): bound parallel imported music decoding --- .../imported-music-playback-benchmark.rs | 152 ++++++++++++++++++ crates/editor/src/audio.rs | 83 ++++++++++ crates/editor/src/segments.rs | 58 +++++-- 3 files changed, 278 insertions(+), 15 deletions(-) create mode 100644 crates/editor/examples/imported-music-playback-benchmark.rs diff --git a/crates/editor/examples/imported-music-playback-benchmark.rs b/crates/editor/examples/imported-music-playback-benchmark.rs new file mode 100644 index 00000000000..da8a1250350 --- /dev/null +++ b/crates/editor/examples/imported-music-playback-benchmark.rs @@ -0,0 +1,152 @@ +use cap_editor::{AudioOutput, EditorFrameOutput, EditorInstance, PlaybackEvent}; +use cap_project::{ + AudioTrackSegment, ClipConfiguration, ProjectConfiguration, RecordingMeta, TimelineSegment, XY, +}; +use std::{ + path::PathBuf, + sync::Arc, + time::{Duration, Instant}, +}; + +#[tokio::main(flavor = "multi_thread", worker_threads = 2)] +async fn main() { + ffmpeg::init().unwrap(); + let root = PathBuf::from(std::env::args_os().nth(1).expect("fixture directory")); + let count: usize = std::env::args() + .nth(2) + .unwrap_or("2".into()) + .parse() + .unwrap(); + assert!((1..=2).contains(&count)); + let mut project = ProjectConfiguration { + timeline: Some( + serde_json::from_value(serde_json::json!({"segments":[],"zoomSegments":[]})).unwrap(), + ), + ..Default::default() + }; + project.camera.hide = true; + project.cursor.hide = true; + project.clips = vec![ClipConfiguration { + index: 0, + ..Default::default() + }]; + let timeline = project.timeline.as_mut().unwrap(); + timeline.segments = vec![TimelineSegment { + recording_clip: 0, + start: 0.0, + end: 1200.0, + timescale: 1.0, + ..Default::default() + }]; + for path in ["music-a.m4a", "music-b.m4a"].into_iter().take(count) { + timeline.audio_segments.push(AudioTrackSegment { + start: 0.0, + end: 1200.0, + track: 0, + path: path.into(), + name: None, + enabled: true, + trim_start: 0.0, + volume_db: -12.0, + fade_in: 0.0, + fade_out: 0.0, + duration: Some(3600.0), + }); + } + let delayed_cold = std::env::args().nth(3).as_deref() == Some("delayed-cold"); + let playback_project = project.clone(); + if delayed_cold { + project.timeline.as_mut().unwrap().audio_segments.clear(); + } + project.write(&root).unwrap(); + let mut meta: RecordingMeta = serde_json::from_value(serde_json::json!({ + "platform":"Linux","pretty_name":"Imported music playback benchmark","sharing":null, + "segments":[{"display":{"path":"display.mp4","fps":1,"start_time":0.0}}], + "cursors":{},"status":{"status":"Complete"} + })) + .unwrap(); + meta.project_path = root.clone(); + meta.save_for_project().unwrap(); + let (audio_tx, mut audio_rx) = tokio::sync::mpsc::unbounded_channel(); + let output = Arc::new(AudioOutput::new_headless(Box::new(move |samples, _| { + if samples.iter().any(|sample| sample.abs() > 0.00001) { + let _ = audio_tx.send(Instant::now()); + } + }))); + let (frame_tx, mut frame_rx) = tokio::sync::mpsc::unbounded_channel(); + let opened = Instant::now(); + let editor = EditorInstance::new_with_audio_output( + root, + |_| {}, + Box::new(move |frame, _| { + let number = match frame { + EditorFrameOutput::Rgba(frame) => frame.frame_number, + EditorFrameOutput::Nv12(frame) => frame.frame_number, + #[cfg(target_os = "macos")] + EditorFrameOutput::Surface(frame) => frame.frame_number, + }; + let _ = frame_tx.send((number, Instant::now())); + }), + None, + output, + ) + .await + .unwrap(); + let open_seconds = opened.elapsed().as_secs_f64(); + if delayed_cold { + tokio::time::sleep(Duration::from_millis(250)).await; + editor.project_config.0.send(playback_project).unwrap(); + tokio::time::sleep(Duration::from_millis(250)).await; + } + for phase in [ + if delayed_cold { + "delayed-cold" + } else { + "immediate" + }, + "warm", + ] { + while audio_rx.try_recv().is_ok() {} + while frame_rx.try_recv().is_ok() {} + let start_frame = if phase == "warm" { 900 } else { 0 }; + let started = Instant::now(); + let mut handle = tokio::time::timeout( + Duration::from_secs(30), + editor.start_playback_with_handle(15, XY::new(640, 360), Some(start_frame)), + ) + .await + .unwrap() + .unwrap(); + let start_return_seconds = started.elapsed().as_secs_f64(); + let first_audio = tokio::time::timeout(Duration::from_secs(20), audio_rx.recv()) + .await + .unwrap() + .unwrap(); + let (observed_frame, first_frame) = tokio::time::timeout(Duration::from_secs(20), async { + loop { + let (number, timestamp) = frame_rx.recv().await.unwrap(); + if number >= start_frame && number < start_frame + 150 { + break (number, timestamp); + } + } + }) + .await + .unwrap(); + println!( + "MUSIC_PLAYBACK {}", + serde_json::json!({"tracks":count,"phase":phase,"start_frame":start_frame,"observed_frame":observed_frame,"open_seconds":open_seconds,"start_return_seconds":start_return_seconds,"first_audio_seconds":first_audio.saturating_duration_since(started).as_secs_f64(),"first_frame_seconds":first_frame.saturating_duration_since(started).as_secs_f64(),"open_to_audio_seconds":first_audio.saturating_duration_since(opened).as_secs_f64()}) + ); + handle.stop(); + tokio::time::timeout(Duration::from_secs(5), async { + loop { + if matches!(*handle.receive_event().await, PlaybackEvent::Stop) { + break; + } + } + }) + .await + .unwrap(); + tokio::time::sleep(Duration::from_millis(250)).await; + } + editor.dispose().await; +} diff --git a/crates/editor/src/audio.rs b/crates/editor/src/audio.rs index 7f846172b49..f793e41d67c 100644 --- a/crates/editor/src/audio.rs +++ b/crates/editor/src/audio.rs @@ -4740,6 +4740,89 @@ mod tests { assert!((left_at_second(&stream, 1) - expected(10_000)).abs() < 0.02); } + #[test] + fn parallel_music_load_matches_serial_mix_and_preserves_failed_cache_entries() { + ffmpeg::init().unwrap(); + let dir = tempfile::tempdir().unwrap(); + let mut segments = Vec::new(); + for (index, values) in [ + [2_000, -4_000, 6_000], + [-3_000, 5_000, -7_000], + [1_000, 8_000, -2_000], + ] + .iter() + .enumerate() + { + let path = format!("music-{index}.wav"); + write_step_wav(&dir.path().join(&path), values); + let mut segment = music_track_segment(&path, 0.0, 1.5, 0.125, 0.25); + segment.trim_start = 0.5; + segments.push(segment); + } + let project = music_project(segments.clone()); + let mut reference = MusicTracks::new(); + for segment in &segments { + reference.extend(crate::load_music_tracks_uncached( + &music_project(vec![segment.clone()]), + dir.path(), + )); + } + let mut cache = MusicTracks::new(); + let actual = crate::load_music_tracks(&project, dir.path(), &mut cache); + assert_eq!(actual.len(), 3); + let repeated = crate::load_music_tracks(&project, dir.path(), &mut cache); + for (path, data) in &actual { + assert!(Arc::ptr_eq(data, &repeated[path])); + } + let render = |tracks| { + let mut renderer = AudioRenderer::new(vec![]).with_music(tracks); + render_export_audio(&mut renderer, &project, 30, 45) + .into_iter() + .map(f32::to_bits) + .collect::>() + }; + assert_eq!(render(actual.clone()), render(reference)); + for retained in [1, 2] { + let mut partial = actual + .iter() + .take(retained) + .map(|(path, data)| (path.clone(), Arc::clone(data))) + .collect::(); + let retained_entries = partial.clone(); + let loaded = crate::load_music_tracks(&project, dir.path(), &mut partial); + assert_eq!(render(loaded), render(actual.clone())); + for (path, data) in &retained_entries { + assert!(Arc::ptr_eq(data, &partial[path])); + } + } + let corrupt_path = "corrupt.wav"; + std::fs::write(dir.path().join(corrupt_path), b"invalid audio").unwrap(); + let mut corrupt_segments = segments.clone(); + corrupt_segments.push(music_track_segment(corrupt_path, 0.0, 1.5, 0.0, 0.0)); + let mut corrupt_cache = MusicTracks::new(); + let surviving = crate::load_music_tracks( + &music_project(corrupt_segments), + dir.path(), + &mut corrupt_cache, + ); + assert_eq!(surviving.len(), 3); + assert!(!corrupt_cache.contains_key(corrupt_path)); + assert_eq!(render(surviving), render(actual.clone())); + for segment in &mut segments { + std::fs::remove_file(dir.path().join(&segment.path)).unwrap(); + segment.trim_start = 0.0; + } + let failed = crate::load_music_tracks(&music_project(segments), dir.path(), &mut cache); + assert!(failed.is_empty()); + for (path, data) in &actual { + assert!(Arc::ptr_eq(data, &cache[path])); + } + let restored = crate::load_music_tracks(&project, dir.path(), &mut cache); + for (path, data) in &actual { + assert!(Arc::ptr_eq(data, &restored[path])); + } + } + static CACHE_RELEASE: std::sync::RwLock<()> = std::sync::RwLock::new(()); struct ProgressiveFixture { diff --git a/crates/editor/src/segments.rs b/crates/editor/src/segments.rs index 60953e588f5..5a485f19d94 100644 --- a/crates/editor/src/segments.rs +++ b/crates/editor/src/segments.rs @@ -61,29 +61,57 @@ pub fn load_music_tracks( .or_insert((trim_start, trim_end)); } + let mut missing = Vec::new(); for (path, (source_start, source_end)) in ranges { if let Some(data) = cache.get(path) && data.covers_source_range(source_start, source_end) { result.insert(path.to_string(), Arc::clone(data)); - continue; + } else { + missing.push((path, source_start, source_end)); } + } - let resolved = resolve_music_path(project_path, path); - match AudioData::from_file_range(&resolved, source_start, source_end) { - Ok(data) => { - let data = Arc::new(data); - cache.insert(path.to_string(), Arc::clone(&data)); - result.insert(path.to_string(), data); - } - Err(error) => { - warn!( - path = %resolved.display(), - error, - "Failed to load timeline audio track; skipping" - ); + let decode = |ranges: &[(&str, usize, usize)]| { + ranges + .iter() + .filter_map(|&(path, source_start, source_end)| { + let resolved = resolve_music_path(project_path, path); + match AudioData::from_file_range(&resolved, source_start, source_end) { + Ok(data) => Some((path.to_string(), Arc::new(data))), + Err(error) => { + warn!( + path = %resolved.display(), + error, + "Failed to load timeline audio track; skipping" + ); + None + } + } + }) + .collect::>() + }; + let decoded = if missing.len() < 2 { + decode(&missing) + } else { + std::thread::scope(|scope| { + let (left, right) = missing.split_at(missing.len().div_ceil(2)); + match std::thread::Builder::new().spawn_scoped(scope, || decode(left)) { + Ok(worker) => { + let mut decoded = decode(right); + match worker.join() { + Ok(other) => decoded.extend(other), + Err(panic) => std::panic::resume_unwind(panic), + } + decoded + } + Err(_) => decode(&missing), } - } + }) + }; + for (path, data) in decoded { + cache.insert(path.clone(), Arc::clone(&data)); + result.insert(path, data); } result