perf(editor): decode imported music with bounded parallelism

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;
+}

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::<Vec<_>>()
+        };
+        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::<MusicTracks>();
+            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 {

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::<Vec<_>>()
+    };
+    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