From a9a4675b2319a818929f2f5a28271fd866ba00e2 Mon Sep 17 00:00:00 2001 From: FedyaLight Date: Wed, 11 Nov 2026 20:56:06 +0300 Subject: [PATCH] Inspect media files in parallel with one ffprobe call each Inspection was sequential and started three ffprobe processes for each video file without an ISO header: the format and stream headers, then the same video header twice, for the timecode and for the timing walk. Each start analyses the opening packets, about 0.3 s for Canon XF-AVC. One call now serves all three. The cached timing walk is guarded by the file's revision from before its header was read, so a file replaced meanwhile never caches another version's frame rate. The video header query did not request the format tags, where MXF keeps its timecode, so Canon and other MXF clips arrived without container timecode and inspection searched every audio track for LTC instead. The timecode is now read. Files are inspected on four bounded workers, like fingerprinting, and the outcomes keep input order. Progress counts are published in order, and a cancellation while files are in flight is reported as such. Inspecting eight Canon C70 clips and five other files went from 21.2 s to 0.56 s (2.1 s with one worker), with the timecode of all eight clips (#17). --- crates/align-decode/src/ff.rs | 164 ++++++++++++++-------- crates/align-decode/src/pipeline.rs | 205 ++++++++++++++++++++++++---- 2 files changed, 285 insertions(+), 84 deletions(-) diff --git a/crates/align-decode/src/ff.rs b/crates/align-decode/src/ff.rs index 5fb6455..1a6797e 100644 --- a/crates/align-decode/src/ff.rs +++ b/crates/align-decode/src/ff.rs @@ -77,8 +77,10 @@ pub fn ffmpeg_bin() -> Option { resolve_bin("ALIGN_FFMPEG", "ffmpeg") } -/// Header-only inspect. No sample data is read (audio-only invariant). -pub fn inspect(path: &Path) -> Result { +/// Format and stream headers as ffprobe JSON. Each ffprobe start analyses +/// the opening packets (about 0.3 s for 4:2:2 10-bit camera video), so +/// inspection asks once and parses this for audio and video alike. +fn probe_json(path: &Path) -> Result, DecodeError> { let bin = ffprobe_bin().ok_or_else(|| DecodeError::FfmpegMissing("ffprobe not found".into()))?; let out = command(bin) @@ -99,7 +101,12 @@ pub fn inspect(path: &Path) -> Result { path.display() ))); } - parse_ffprobe_json(&out.stdout) + Ok(out.stdout) +} + +/// Header-only inspect. No sample data is read (audio-only invariant). +pub fn inspect(path: &Path) -> Result { + parse_ffprobe_json(&probe_json(path)?) .ok_or_else(|| DecodeError::Incomplete(format!("ffprobe JSON for {}", path.display()))) } @@ -165,16 +172,22 @@ pub fn parse_ffprobe_json(json: &[u8]) -> Option { duration_seconds, audio_streams, has_video, - video: None, // filled by inspect_video() below, not the pure parser + video: None, // filled by inspect_full() from the same JSON }) } /// Header inspect + video timing walk. Pure-header fast path when the /// container has no video track. pub fn inspect_full(path: &Path) -> Result { - let mut report = inspect(path)?; + // Taken before the header is read: the cached timing walk includes the + // header's frame rate. + let revision = media_revision(path); + let json = probe_json(path)?; + let mut report = parse_ffprobe_json(&json) + .ok_or_else(|| DecodeError::Incomplete(format!("ffprobe JSON for {}", path.display())))?; if report.has_video { - report.video = inspect_video(path); + report.video = + parse_video_stream(&json).and_then(|header| video_probe(path, &header, revision)); } Ok(report) } @@ -186,7 +199,8 @@ struct FfVideoStream { timecode: Option, } -/// First video stream's header facts + optional `timecode` tag. +/// First video stream's header facts + optional `timecode` tag. MXF keeps +/// the timecode in the format tags, which need `-show_format`. fn parse_video_stream(json: &[u8]) -> Option { let v: serde_json::Value = serde_json::from_slice(json).ok()?; let stream = v.get("streams")?.as_array()?.iter().find(|s| { @@ -271,6 +285,43 @@ impl PacketTiming { /// the shared [`align_core::classify`]. Early-exits once variability is /// proven; constant-rate classification requires the full walk. pub fn video_timing(path: &Path) -> Result { + cached_video_timing(path, media_revision(path), |bin| { + let header = command(bin) + .args([ + "-v", + "error", + "-print_format", + "json", + "-show_streams", + "-select_streams", + "v:0", + ]) + .arg(path) + .output() + .map_err(DecodeError::Io)?; + Ok(parse_video_stream(&header.stdout) + .as_ref() + .and_then(header_frame_duration)) + }) +} + +/// Nominal rate from the header, snapped through the broadcast table. +fn header_frame_duration(header: &FfVideoStream) -> Option { + align_core::canonical_frame_duration(header.avg_fps?, None) +} + +fn media_revision(path: &Path) -> Option { + align_core::cache::media_revision(path).ok() +} + +/// The walk is cached per file revision; `frame_duration` runs only when +/// it is not. `revision` is the file's revision from before its header was +/// read, so a header and a walk of different versions are never cached. +fn cached_video_timing( + path: &Path, + revision: Option, + frame_duration: impl FnOnce(&Path) -> Result, DecodeError>, +) -> Result { let bin = ffprobe_bin().ok_or_else(|| DecodeError::FfmpegMissing("ffprobe not found".into()))?; let decoder = bin.canonicalize().ok().and_then(|bin| { @@ -280,18 +331,16 @@ pub fn video_timing(path: &Path) -> Result Result, ) -> Result { - use align_core::canonical_frame_duration; use std::io::BufRead; let mut child = command(bin) @@ -325,26 +374,6 @@ fn video_timing_uncached( .map_err(DecodeError::Io)?; let stdout = child.stdout.take().ok_or(DecodeError::UnsupportedOutput)?; - // Nominal rate + dimensions from the header call. - let header_json = command(bin) - .args([ - "-v", - "error", - "-print_format", - "json", - "-show_streams", - "-select_streams", - "v:0", - ]) - .arg(path) - .output() - .map_err(DecodeError::Io)?; - let header = parse_video_stream(&header_json.stdout); - let frame_duration = header - .as_ref() - .and_then(|h| h.avg_fps) - .and_then(|fps| canonical_frame_duration(fps, None)); - let mut timing = PacketTiming::default(); let read_result = (|| { for line in std::io::BufReader::new(stdout).lines() { @@ -382,32 +411,18 @@ fn video_timing_uncached( } /// Full video probe: header facts + timing walk + timecode tag. -/// `None` when the container has no usable video stream. -pub fn inspect_video(path: &Path) -> Option { - let header_json = command(ffprobe_bin()?) - .args([ - "-v", - "error", - "-print_format", - "json", - "-show_streams", - "-select_streams", - "v:0", - ]) - .arg(path) - .output() - .ok()?; - if !header_json.status.success() { - return None; - } - let header = parse_video_stream(&header_json.stdout)?; - let timing = video_timing(path).ok()?; +fn video_probe( + path: &Path, + header: &FfVideoStream, + revision: Option, +) -> Option { + let timing = cached_video_timing(path, revision, |_| Ok(header_frame_duration(header))).ok()?; Some(VideoProbe { width: header.width, height: header.height, frame_duration: timing.frame_duration, mode: timing.mode, - source_timecode: timecode_from_header(&header), + source_timecode: timecode_from_header(header), }) } @@ -416,7 +431,7 @@ pub fn inspect_video(path: &Path) -> Option { /// `1/30`), then the label validates as NDF/DF. `None` when the rate is /// unknown or the label is forbidden — never guessed. Pure (no binary). fn timecode_from_header(header: &FfVideoStream) -> Option { - let duration = align_core::canonical_frame_duration(header.avg_fps?, None)?; + let duration = header_frame_duration(header)?; align_core::SourceTimecode::from_label(header.timecode.as_deref()?, duration) } pub struct AudioPipe { @@ -877,4 +892,43 @@ mod tests { assert!((peak - 1.0).abs() < 0.01, "peak={peak}"); let _ = std::fs::remove_dir_all(&dir); } + + /// MXF stores its start timecode with the material package, which + /// ffprobe reports in the format tags rather than on the video stream. + #[test] + fn mxf_timecode_comes_from_the_format_tags() { + let (Some(ffmpeg), Some(_)) = (ffmpeg_bin(), ffprobe_bin()) else { + return; + }; + let lists = |kind: &str| { + command(&ffmpeg) + .args(["-hide_banner", kind]) + .output() + .map(|out| String::from_utf8_lossy(&out.stdout).into_owned()) + .unwrap_or_default() + }; + if !lists("-muxers").contains(" mxf ") || !lists("-encoders").contains(" mpeg2video ") { + eprintln!("SKIP: FFmpeg without MXF or MPEG-2 support"); + return; + } + let dir = tempfile::tempdir().unwrap(); + let clip = dir.path().join("A001C001.MXF"); + let ok = command(&ffmpeg) + .args(["-v", "error", "-y", "-f", "lavfi", "-i"]) + .arg("testsrc2=size=320x240:rate=25") + .args(["-f", "lavfi", "-i", "sine=r=48000", "-t", "1"]) + .args(["-c:v", "mpeg2video", "-c:a", "pcm_s16le"]) + .args(["-timecode", "10:20:30:12"]) + .arg(&clip) + .status() + .is_ok_and(|s| s.success()); + assert!(ok, "FFmpeg could not write the MXF fixture"); + let video = inspect_full(&clip).unwrap().video.expect("video probe"); + let timecode = video.source_timecode.expect("container timecode"); + assert_eq!(timecode.text, "10:20:30:12"); + assert_eq!( + video.frame_duration, + Some(align_core::MediaTime::new(1, 25)) + ); + } } diff --git a/crates/align-decode/src/pipeline.rs b/crates/align-decode/src/pipeline.rs index 2af4ab5..b326cd7 100644 --- a/crates/align-decode/src/pipeline.rs +++ b/crates/align-decode/src/pipeline.rs @@ -6,8 +6,8 @@ //! (via [`BackendWindowProvider`]) → graph solve → [`align_core::SyncResult`]. //! //! Important implementation choices: -//! - fingerprinting and refinement use four bounded workers with indexed -//! deterministic results; +//! - inspection, fingerprinting and refinement use four bounded workers with +//! indexed deterministic results; //! - ClipIDs hash `path\0duration-micros`, using the same units across backends; //! - spanned-file, BWF/iXML/container/LTC timecode, FCP 7 XML and FCPXML //! evidence join the same waveform-first graph without overriding a @@ -27,6 +27,12 @@ use sha2::{Digest, Sha256}; use crate::backend::{BackendKind, MediaBackend, default_backend}; use crate::provider::BackendWindowProvider; +/// Files inspected at once. Inspection waits on ffprobe processes and header +/// reads, and the first time it sees an MXF or MTS file it reads the whole +/// file to classify frame timing. The bound matches the fingerprint stage's +/// four concurrent readers of external media. +const INSPECT_WORKERS: usize = 4; + // ------------------------------------------------------------ types #[derive(Clone, Debug, PartialEq)] @@ -316,8 +322,9 @@ impl Pipeline { } let mut clips = Vec::new(); let mut warnings = Vec::new(); - for url in &urls { - match self.inspect_one(url) { + let inspected = self.inspect_all(&urls, None, &std::sync::atomic::AtomicBool::new(false)); + for (url, outcome) in urls.iter().zip(inspected) { + match outcome { Ok(clip) => clips.push(clip), Err(e) => warnings.push(SyncWarning { url: url.clone(), @@ -348,8 +355,8 @@ impl Pipeline { /// Full pipeline: media files/dirs (+ optional one timeline) → /// [`align_core::SyncResult`]. /// - /// Fingerprint decode and fine refine run on up to 4 workers with - /// deterministic indexed results; `progress` + /// Inspection, fingerprint decode and fine refine run on up to 4 workers + /// with deterministic indexed results; `progress` /// is shared across those threads, `cancel` is polled per clip, /// candidate and job. pub fn synchronize( @@ -398,30 +405,15 @@ impl Pipeline { // ---- inspect let mut clips = Vec::new(); let mut warnings = Vec::new(); - for (i, url) in urls.iter().enumerate() { - if cancelled(cancel) { - return Err(PipelineError::Cancelled); - } - if let Some(p) = &progress { - p(PipelineProgress::basic( - Phase::Inspect, - i, - urls.len(), - Some(url.clone()), - )); - } - match self.inspect_one(url) { + let inspected = self.inspect_all(&urls, progress, cancel); + // Files already being inspected finish (or fail) after a + // cancellation; it must not turn into a missing-media error. + if cancelled(cancel) { + return Err(PipelineError::Cancelled); + } + for (url, outcome) in urls.iter().zip(inspected) { + match outcome { Ok(clip) => { - if let Some(p) = &progress { - let mut event = PipelineProgress::basic( - Phase::Inspect, - clips.len(), - urls.len(), - Some(url.clone()), - ); - event.discovered = Some(clip.clone()); - p(event); - } // Variable video timing does not affect audio sync, // but constant-rate interchange needs a CFR transcode. if let Some(video) = &clip.video { @@ -440,6 +432,7 @@ impl Pipeline { } clips.push(clip); } + Err(PipelineError::Cancelled) => return Err(PipelineError::Cancelled), Err(e) => warnings.push(SyncWarning { url: url.clone(), message: e.to_string(), @@ -743,6 +736,38 @@ impl Pipeline { }) } + /// Inspects files on bounded workers. Outcomes keep input order, so clips + /// and warnings never depend on which file finishes first; progress + /// reports each file as it starts and as it finishes. + fn inspect_all( + &self, + urls: &[PathBuf], + progress: Option<&(dyn Fn(PipelineProgress) + Send + Sync)>, + cancel: &std::sync::atomic::AtomicBool, + ) -> Vec> { + // Counting and publishing under one lock keeps the completed count + // monotonic for listeners such as the progress bar. + let done = std::sync::Mutex::new(0_usize); + let publish = |finished: bool, url: &PathBuf, discovered: Option| { + let Some(p) = progress else { return }; + let mut done = done.lock().unwrap_or_else(|poisoned| poisoned.into_inner()); + *done += usize::from(finished); + let mut event = + PipelineProgress::basic(Phase::Inspect, *done, urls.len(), Some(url.clone())); + event.discovered = discovered; + p(event); + }; + align_core::par_map(urls, INSPECT_WORKERS, |url| { + if cancelled(cancel) { + return Err(PipelineError::Cancelled); + } + publish(false, url, None); + let outcome = self.inspect_one(url); + publish(true, url, outcome.as_ref().ok().cloned()); + outcome + }) + } + fn inspect_one(&self, url: &Path) -> Result { let probe = self.backend.inspect(url).map_err(PipelineError::Decode)?; let micros = (probe.duration_seconds * 1_000_000.0).round() as i64; @@ -1525,6 +1550,128 @@ mod tests { } } + /// Inspection that fails every file after `delay`; nothing else runs. + struct FailingInspect { + delay: std::time::Duration, + } + + impl MediaBackend for FailingInspect { + fn kind(&self) -> BackendKind { + BackendKind::Portable + } + + fn inspect(&self, _: &Path) -> Result { + std::thread::sleep(self.delay); + Err(crate::DecodeError::Incomplete("unreadable".into())) + } + + fn decode_mono_8k( + &self, + _: &Path, + _: AudioAnalysisSource, + _: &mut dyn FnMut(&[f32]) -> Result<(), crate::DecodeError>, + ) -> Result<(), crate::DecodeError> { + unreachable!() + } + + fn decode_window_16k( + &self, + _: &Path, + _: f64, + _: f64, + _: AudioAnalysisSource, + ) -> Result<(f64, Vec), crate::DecodeError> { + unreachable!() + } + + fn decode_native( + &self, + _: &Path, + _: usize, + _: Option<(f64, Option)>, + _: &mut dyn FnMut(crate::backend::NativeBlock) -> Result<(), crate::DecodeError>, + ) -> Result { + unreachable!() + } + + fn decode_native_i32( + &self, + _: &Path, + _: usize, + _: Option<(f64, Option)>, + _: &mut dyn FnMut(crate::backend::NativeBlockI32) -> Result<(), crate::DecodeError>, + ) -> Result { + unreachable!() + } + } + + fn failing_inspection(tag: &str, files: usize, delay_ms: u64) -> (PathBuf, Pipeline) { + let dir = fixture_dir(tag); + for index in 0..files { + std::fs::write(dir.join(format!("{index:03}.wav")), b"").unwrap(); + } + let pipe = Pipeline { + backend: Box::new(FailingInspect { + delay: std::time::Duration::from_millis(delay_ms), + }), + cache: FingerprintCache::with_backend(Some(dir.join(".cache")), "failing-inspect"), + }; + (dir, pipe) + } + + /// Cancelling while every worker is busy used to end as "no media" once + /// those files failed. + #[test] + fn cancelling_during_inspection_is_reported_as_cancelled() { + let (dir, pipe) = failing_inspection("cancel-inspection", INSPECT_WORKERS, 400); + let cancel = std::sync::atomic::AtomicBool::new(false); + let result = std::thread::scope(|scope| { + scope.spawn(|| { + std::thread::sleep(std::time::Duration::from_millis(100)); + cancel.store(true, std::sync::atomic::Ordering::Relaxed); + }); + pipe.synchronize( + &[PipelineInput::Media(dir.clone())], + &[], + &PipelineOptions::default(), + None, + &cancel, + ) + }); + assert!( + matches!(result, Err(PipelineError::Cancelled)), + "{result:?}" + ); + let _ = std::fs::remove_dir_all(&dir); + } + + /// Workers finish in any order; the completed count the progress bar + /// shows must still only grow. + #[test] + fn inspection_progress_never_goes_backwards() { + let (dir, pipe) = failing_inspection("inspection-progress", 400, 0); + let counts = std::sync::Mutex::new(Vec::new()); + let progress = |event: PipelineProgress| { + if event.phase == Phase::Inspect { + counts.lock().unwrap().push(event.completed); + } + }; + let _ = pipe.synchronize( + &[PipelineInput::Media(dir.clone())], + &[], + &PipelineOptions::default(), + Some(&progress), + &std::sync::atomic::AtomicBool::new(false), + ); + let counts = counts.into_inner().unwrap(); + assert!(counts.len() > 400); + assert!( + counts.windows(2).all(|pair| pair[0] <= pair[1]), + "{counts:?}" + ); + let _ = std::fs::remove_dir_all(&dir); + } + #[test] fn fixed_project_copy_repairs_every_sequence_and_preserves_source() { let dir = fixture_dir("fixed-project-copy");