diff --git a/crates/media/src/pool.rs b/crates/media/src/pool.rs index f8651cd6..3df9ec45 100644 --- a/crates/media/src/pool.rs +++ b/crates/media/src/pool.rs @@ -2,7 +2,7 @@ use std::collections::{BTreeMap, HashMap, HashSet}; use std::sync::atomic::{AtomicBool, AtomicU64, Ordering}; -use std::sync::{Arc, Condvar, Mutex}; +use std::sync::{Arc, Condvar, Mutex, OnceLock}; use effectcraft_project::{AlphaMode, Footage, FootageKind, ItemId}; use effectcraft_raster::{Image, Px}; @@ -53,8 +53,25 @@ struct Entry { /// version 1 files of sources longer than about 12:36 at 48 kHz were silent from there on (#172). const CONFORM_VERSION: u32 = 2; +/// Footage longer than this (24 hours) is read from its decoder rather than conformed: its +/// duration comes from the file's header, and a damaged one claiming days of audio had the +/// conform thread allocate all of it at once (#408). +const MAX_CONFORM_SECONDS: i64 = 24 * 3600; + +/// The sample frames of `footage`'s audio at `rate` Hz that a conformed file holds, `None` past +/// [`MAX_CONFORM_SECONDS`]. +#[cfg_attr(target_arch = "wasm32", allow(dead_code))] +fn conform_frames(footage: &Footage, rate: u32) -> Option { + let frames = footage.duration.to_units_floor(i64::from(rate)).max(0); + if frames > MAX_CONFORM_SECONDS.saturating_mul(i64::from(rate)) { + return None; + } + usize::try_from(frames).ok() +} + /// The source time of sample frame `at` at `rate` Hz. (`at` × ticks per second overflows `i64` /// past about 12:36 at 48 kHz.) +#[cfg_attr(target_arch = "wasm32", allow(dead_code))] fn sample_time(at: usize, rate: u32) -> Tick { Tick::from_units(i64::try_from(at).unwrap_or(i64::MAX), i64::from(rate)) } @@ -135,6 +152,8 @@ pub struct PoolStats { /// Frames currently cached and their total size in bytes. pub frames: usize, pub bytes: usize, + /// Movie and audio files opened (each path once, however many callers ask for it at once). + pub opened: u64, } #[derive(Default)] @@ -146,9 +165,15 @@ struct CacheState { last: HashMap, i64>, } +/// A movie/audio source by path, opened by the first caller while the others wait for it +/// (`None`: failed to open; not retried until `forget`). +type SourceSlot = Arc>>; + struct Inner { - /// Opened movie/audio sources by path (`None`: failed to open; not retried until `forget`). - sources: Mutex, Option>>, + /// Opened movie/audio sources by path. + sources: Mutex, SourceSlot>>, + /// Files opened into `sources` (see [`PoolStats::opened`]). + opened: AtomicU64, /// In-memory files registered with `add_bytes` (web builds, tests). files: Mutex>>, /// Parsed 3D models by path (`None`: failed to load; not retried until `forget`). @@ -228,6 +253,7 @@ impl MediaPool { hits: AtomicU64::new(0), misses: AtomicU64::new(0), prefetched: AtomicU64::new(0), + opened: AtomicU64::new(0), conform: Mutex::default(), conforming: Mutex::default(), }), @@ -259,6 +285,7 @@ impl MediaPool { prefetched: i.prefetched.load(Ordering::Relaxed), frames: c.lru.map.len(), bytes: c.lru.bytes, + opened: i.opened.load(Ordering::Relaxed), } } @@ -336,30 +363,21 @@ impl MediaPool { Some(out) } - /// Conform `footage`'s audio at `rate` into `file` on a worker thread (once). + /// Conform `frames` sample frames of `footage`'s audio at `rate` into `file` on a worker + /// thread (once). #[cfg(not(target_arch = "wasm32"))] - fn conform(&self, footage: &Footage, file: std::path::PathBuf, rate: u32) { + fn conform(&self, footage: &Footage, file: std::path::PathBuf, frames: usize, rate: u32) { if !lock(&self.inner.conforming).insert(file.clone()) { return; } let (pool, f) = (self.clone(), footage.clone()); let spawned = std::thread::Builder::new().name("ec-conform-audio".into()).spawn(move || { - let frames = f.duration.to_units_floor(rate as i64).max(0) as usize; - let mut bytes = Vec::with_capacity(frames * 8); - let mut at = 0usize; - while at < frames { - let n = (frames - at).min(1 << 16); - let t = sample_time(at, rate); - for v in pool.decode_audio(&f, t, n, rate) { - bytes.extend_from_slice(&v.to_le_bytes()); - } - at += n; - } - let ok = file.parent().is_some_and(|d| std::fs::create_dir_all(d).is_ok()) && { - let tmp = file.with_extension("tmp"); - std::fs::write(&tmp, &bytes).is_ok() && std::fs::rename(&tmp, &file).is_ok() - }; + let tmp = file.with_extension("tmp"); + let ok = file.parent().is_some_and(|d| std::fs::create_dir_all(d).is_ok()) + && pool.write_conformed(&f, &tmp, frames, rate).is_ok() + && std::fs::rename(&tmp, &file).is_ok(); if !ok { + let _ = std::fs::remove_file(&tmp); log::warn!("media: could not write conformed audio {}", file.display()); } lock(&pool.inner.conforming).remove(&file); @@ -369,6 +387,23 @@ impl MediaPool { } } + /// Write `frames` sample frames of `footage`'s audio at `rate` to `path` (raw interleaved + /// stereo `f32`) a chunk at a time: the whole track was held in memory before it was written. + #[cfg(not(target_arch = "wasm32"))] + fn write_conformed(&self, footage: &Footage, path: &std::path::Path, frames: usize, rate: u32) -> std::io::Result<()> { + use std::io::Write; + let mut out = std::io::BufWriter::new(std::fs::File::create(path)?); + let mut at = 0usize; + while at < frames { + let n = (frames - at).min(1 << 16); + for v in self.decode_audio(footage, sample_time(at, rate), n, rate) { + out.write_all(&v.to_le_bytes())?; + } + at += n; + } + out.flush() + } + /// `frames` stereo sample frames of `footage`'s audio starting at source time `start`, at /// `rate` Hz, interleaved (L R L R …). Mono is duplicated to both channels; channels beyond /// the first two are dropped. Silence where there is no audio (or before the media starts). @@ -378,7 +413,9 @@ impl MediaPool { return vec![0.0; frames * 2]; } #[cfg(not(target_arch = "wasm32"))] - if let Some(file) = self.conformed_path(&footage.path, rate) { + if let Some(conform) = conform_frames(footage, rate) + && let Some(file) = self.conformed_path(&footage.path, rate) + { let s0 = start.to_units_floor(rate as i64); if file.exists() && !lock(&self.inner.conforming).contains(&file) { let skip = (-s0).max(0) as usize; @@ -391,7 +428,7 @@ impl MediaPool { return out; } } else { - self.conform(footage, file, rate); + self.conform(footage, file, conform, rate); } } self.decode_audio(footage, start, frames, rate) @@ -478,24 +515,28 @@ impl Inner { std::fs::read(path).map(Into::into).map_err(|e| MediaError::Io(format!("{path}: {e}"))) } - /// The opened movie/audio source for `path`. + /// The opened movie/audio source for `path`. The first caller opens it, outside the map's + /// lock (reading and parsing a large file takes a moment; FLAC, MP3, Ogg and AIFF files are + /// decoded whole); callers asking for the same file meanwhile wait for that instead of opening + /// it again: a preview's frame workers, audio feeder, conform and waveform threads each + /// decoded a long FLAC in full at once (#408). fn source(&self, path: &Arc) -> Option { - if let Some(s) = lock(&self.sources).get(path) { - return s.clone(); - } - // open outside the lock (reading and parsing a large file takes a moment) - let opened = self.read(path).and_then(|bytes| { - let name = std::path::Path::new(&**path).file_name().map(|n| n.to_string_lossy().to_string()).unwrap_or_default(); - filmcraft_codecs::open_bytes(&name, bytes).map_err(MediaError::from) - }); - let opened = match opened { - Ok(s) => Some(s), - Err(e) => { - log::warn!("media: cannot open {path}: {e}"); - None + let slot = lock(&self.sources).entry(path.clone()).or_default().clone(); + slot.get_or_init(|| { + self.opened.fetch_add(1, Ordering::Relaxed); + let opened = self.read(path).and_then(|bytes| { + let name = std::path::Path::new(&**path).file_name().map(|n| n.to_string_lossy().to_string()).unwrap_or_default(); + filmcraft_codecs::open_bytes(&name, bytes).map_err(MediaError::from) + }); + match opened { + Ok(s) => Some(s), + Err(e) => { + log::warn!("media: cannot open {path}: {e}"); + None + } } - }; - lock(&self.sources).entry(path.clone()).or_insert(opened).clone() + }) + .clone() } /// The frame index (in the footage's interpreted rate) shown at source time `t`, after @@ -747,6 +788,37 @@ mod tests { assert!((sample_time(36_372_480, 48_000).seconds() - 757.76).abs() < 1e-9); } + /// #408: footage whose header claims a month of audio isn't conformed: the conform thread + /// allocated the whole conformed track up front (a terabyte here), and the failed allocation + /// aborted the app. Its audio is read from the decoder instead (silence: this file doesn't + /// decode). + #[test] + fn a_header_claiming_a_month_of_audio_is_not_conformed() { + let dir = std::path::Path::new(env!("CARGO_MANIFEST_DIR")).join("../../target/test-conform-month"); + let _ = std::fs::remove_dir_all(&dir); + std::fs::create_dir_all(&dir).unwrap(); + let path = dir.join("month.flac"); + std::fs::write(&path, b"fLaC, not really").unwrap(); + let footage = Footage { + path: path.to_string_lossy().into(), + kind: FootageKind::Audio, + duration: Tick::from_seconds_f64(30.0 * 86_400.0), + has_audio: true, + ..Default::default() + }; + assert_eq!(conform_frames(&footage, 48_000), None); + assert_eq!(conform_frames(&Footage { duration: Tick::from_seconds_f64(60.0), ..footage.clone() }, 48_000), Some(2_880_000)); + let pool = MediaPool::new(); + let conformed = dir.join("conformed"); + pool.set_conform_folder(Some(conformed.clone())); + let s = pool.audio_samples(&footage, Tick::from_seconds_f64(3600.0), 1024, 48_000); + assert_eq!(s.len(), 2048); + assert!(s.iter().all(|v| *v == 0.0)); + assert!(lock(&pool.inner.conforming).is_empty(), "no conform thread started"); + assert!(!conformed.exists(), "nothing was written"); + let _ = std::fs::remove_dir_all(&dir); + } + #[test] fn frame_index_loops_and_clamps() { let r = FrameRate::FPS_30; diff --git a/crates/media/tests/footage.rs b/crates/media/tests/footage.rs index b473337c..ed113c35 100644 --- a/crates/media/tests/footage.rs +++ b/crates/media/tests/footage.rs @@ -545,3 +545,109 @@ fn exr_still_is_linear_float() { let b = img.data[1]; assert!(b[3] == 0.5 && b[0] > 0.85 && b[0] < 1.0, "{b:?}"); } + +/// A FLAC of `secs` seconds at `rate` Hz with `channels` channels of a 440 Hz sine (first two: +/// −24 dBFS RMS, the rest quieter), samples of `fmt` (`s16`, or `s32` with 24 significant bits). +fn sine_flac(secs: u32, rate: u32, channels: u32, fmt: &str) -> Option { + let name = format!("sine_{secs}s_{rate}_{channels}ch_{fmt}.flac"); + let src = format!("sine=frequency=440:sample_rate={rate}:duration={secs}"); + let pan = format!("pan={channels}c|{}", (0..channels).map(|c| format!("c{c}={}*c0", if c < 2 { "1" } else { "0.5" })).collect::>().join("|")); + let mut args = vec!["-f", "lavfi", "-i", &src, "-af", &pan, "-sample_fmt", fmt]; + if fmt == "s32" { + args.extend(["-bits_per_raw_sample", "24"]); + } + args.extend(["-c:a", "flac"]); + ffmpeg_fixture(&name, &args) +} + +/// #408 robustness: FLAC footage of any rate, depth and channel count reads correctly as the +/// preview feeder reads it, one device block at a time at the mix rate, through its end and past +/// it: two minutes of 24-bit, 6-channel, 96 kHz audio, and mono 8 kHz, 11.025 kHz, 192 kHz and +/// 8-channel files. +#[test] +fn flac_audio_of_any_shape_reads_in_device_blocks() { + for (secs, rate, channels, fmt) in [(120, 96_000, 6, "s32"), (5, 8_000, 1, "s16"), (5, 11_025, 2, "s16"), (5, 192_000, 2, "s32"), (5, 48_000, 8, "s16")] { + let Some(p) = sine_flac(secs, rate, channels, fmt) else { return }; + let f = probe(&p).expect("probe"); + assert!(f.has_audio && f.kind == FootageKind::Audio, "{}", p.display()); + assert!((f.duration.seconds() - f64::from(secs)).abs() < 0.01, "{}: {}", p.display(), f.duration.seconds()); + let pool = MediaPool::new(); + let mix = 48_000i64; + let total = f.duration.to_units_floor(mix); + let (mut at, mut sum, mut n, mut worst) = (0i64, 0.0f64, 0usize, std::time::Duration::ZERO); + let t0 = Instant::now(); + while at < total + 4096 { + let t = Instant::now(); + let s = pool.audio_samples(&f, Tick::from_units(at, mix), 1024, 48_000); + if at > 0 { + worst = worst.max(t.elapsed()); + } + assert_eq!(s.len(), 2048); + assert!(s.iter().all(|v| v.is_finite() && v.abs() <= 1.0), "{} at {at}", p.display()); + if at + 1024 <= total { + sum += s.iter().map(|v| f64::from(*v) * f64::from(*v)).sum::(); + n += s.len(); + } else if at >= total { + assert!(s.iter().all(|v| *v == 0.0), "{}: silence past the end", p.display()); + } + at += 1024; + } + // ffmpeg's sine source has an amplitude of 1/8, on both channels (mono is duplicated). + let rms = (sum / n.max(1) as f64).sqrt(); + assert!((rms - 0.125 / 2f64.sqrt()).abs() < 0.003, "{}: rms {rms}", p.display()); + eprintln!("{}: read {secs} s in {:?}, slowest block after the first {worst:?}", p.display(), t0.elapsed()); + assert_eq!(pool.stats().opened, 1, "opened once, not per block"); + } +} + +/// #408: the first reads of a FLAC from many threads at once (a preview's frame workers, audio +/// feeder, conform and waveform threads) decode it once: each decoded the whole file, so memory +/// and time grew with the number of threads. +#[test] +fn concurrent_first_reads_open_a_file_once() { + let Some(p) = sine_flac(60, 44_100, 2, "s16") else { return }; + let f = probe(&p).expect("probe"); + let pool = MediaPool::new(); + let start = Arc::new(std::sync::Barrier::new(8)); + let reads: Vec<_> = (0..8) + .map(|_| { + let (pool, f, start) = (pool.clone(), f.clone(), start.clone()); + std::thread::spawn(move || { + start.wait(); + pool.audio_samples(&f, Tick(TICKS_PER_SECOND), 1024, 48_000) + }) + }) + .collect(); + let reads: Vec> = reads.into_iter().map(|t| t.join().expect("read")).collect(); + assert_eq!(pool.stats().opened, 1, "the file was opened once"); + assert!(reads.iter().all(|r| *r == reads[0]) && reads[0].iter().any(|v| *v != 0.0)); +} + +/// Settings ▸ Disk ▸ Conformed Audio Folder: a FLAC's conformed file, written a chunk at a time, +/// reads back exactly as the decoder reads. +#[test] +fn conformed_flac_audio_reads_back_like_the_decoder() { + let Some(p) = sine_flac(5, 44_100, 2, "s16") else { return }; + let f = probe(&p).expect("probe"); + let folder = fixtures().join(format!("conformed-{}", std::process::id())); + let _ = std::fs::remove_dir_all(&folder); + let pool = MediaPool::new(); + let at = Tick::from_seconds_f64(2.5); + let decoded = pool.audio_samples(&f, at, 4800, 48_000); + pool.set_conform_folder(Some(folder.clone())); + let file = pool.conformed_path(&p.to_string_lossy(), 48_000).expect("a conformed path"); + pool.audio_samples(&f, at, 4800, 48_000); + let t0 = Instant::now(); + while !file.exists() { + assert!(t0.elapsed().as_secs() < 60, "the conformed file was never written"); + std::thread::sleep(std::time::Duration::from_millis(20)); + } + // (Renamed into place once complete.) Raw interleaved stereo f32 at 48 kHz. + let bytes = std::fs::read(&file).expect("conformed file"); + assert_eq!(bytes.len(), 5 * 48_000 * 8); + let s0 = at.to_units_floor(48_000) as usize * 8; + let conformed: Vec = bytes[s0..s0 + 4800 * 8].as_chunks::<4>().0.iter().map(|b| f32::from_le_bytes(*b)).collect(); + assert_eq!(conformed, decoded); + assert_eq!(pool.audio_samples(&f, at, 4800, 48_000), decoded); + let _ = std::fs::remove_dir_all(&folder); +} diff --git a/crates/ui-egui/src/audio.rs b/crates/ui-egui/src/audio.rs index dbc37b5d..7972e694 100644 --- a/crates/ui-egui/src/audio.rs +++ b/crates/ui-egui/src/audio.rs @@ -211,23 +211,23 @@ impl AudioPlayback { let span = PlaySpan::new(start, wa, wb, looping, rate); let feed = Arc::new(AudioFeed::default()); let stop = Arc::new(AtomicBool::new(false)); - // Prime ~50 ms so the device does not start on an empty queue. - let mut cursor = span.start; - let prime = (rate as usize / 20).div_ceil(BLOCK); - for _ in 0..prime { - cursor = feed_block(&src, comp, &feed, &span, cursor, rate, mix); - } if cfg!(target_arch = "wasm32") { - let mut p = AudioPlayback { feed: feed.clone(), device, stop, thread: None, feeder: Some((src, comp, cursor, mix)), rate, span }; + // The device takes samples only in `pump`, which fills the queue first. + let mut p = AudioPlayback { feed: feed.clone(), device, stop, thread: None, feeder: Some((src, comp, span.start, mix)), rate, span }; p.device.start(feed)?; p.pump(); return Ok(p); } + // All mixing happens on the feeder thread, the first block too: reading footage for the + // first time opens it (a FLAC or MP3 file decodes whole), which froze the UI for as long + // as that took (#408). The device plays silence until the first block arrives, and the + // clock doesn't count it. let thread = { let (feed, stop) = (feed.clone(), stop.clone()); std::thread::Builder::new() .name("ec-audio-feed".into()) .spawn(move || { + let mut cursor = span.start; let target = queue_target(rate); while !stop.load(Ordering::Relaxed) { if feed.queued_frames() < target && !feed.finished.load(Ordering::Relaxed) { @@ -485,6 +485,80 @@ mod tests { assert_eq!(f.take_peaks(), [0.0, 0.0]); } + /// #408: starting a preview with sound returns at once even when reading the footage takes + /// long (the first read of a long FLAC decodes all of it): the first block was mixed on the + /// calling (UI) thread. The sound arrives from the feeder thread, and the clock waits for it. + #[test] + fn starting_audio_does_not_wait_for_slow_footage() { + use effectcraft_engine::color::Label; + use effectcraft_engine::project::{Comp, Footage, FootageKind, ItemKind, LayerSource, Project, build}; + use effectcraft_engine::render::FootageSource; + use effectcraft_engine::render::cache::LayerCache; + use std::time::{Duration, Instant}; + + /// Footage audio that takes `delay` to read the first time (then a constant 0.5). + struct Slow { + delay: Duration, + opened: Mutex, + } + impl FootageSource for Slow { + fn frame(&self, _: ItemId, _: &Footage, _: Tick) -> Option> { + None + } + fn audio(&self, _: ItemId, _: &Footage, _: Tick, frames: usize, _: u32) -> Option> { + let mut opened = self.opened.lock().unwrap_or_else(std::sync::PoisonError::into_inner); + if !*opened { + std::thread::sleep(self.delay); + *opened = true; + } + Some(vec![0.5; frames * 2]) + } + } + /// An output that only takes the feed. + struct Device; + impl AudioDevice for Device { + fn sample_rate(&self) -> u32 { + 48000 + } + fn start(&mut self, _: Arc) -> Result<(), String> { + Ok(()) + } + fn stop(&mut self) {} + } + + let mut p = Project::default(); + let footage = + Footage { path: "long.flac".into(), kind: FootageKind::Audio, duration: Tick::from_seconds_f64(600.0), has_audio: true, ..Default::default() }; + let item = p.add_item("long.flac", Label::SeaFoam, None, ItemKind::Footage(footage)); + let mut comp = Comp::new(64, 36, FrameRate::new(30, 1), Tick::from_seconds_f64(10.0)); + let layer = build::layer(&mut p, &comp, "long.flac", LayerSource::Footage { item }, (0, 0), None); + comp.layers = vec![layer]; + let cid = p.add_item("Comp", Label::Sandstone, None, ItemKind::Comp(comp.into())); + let src = RenderSource { + project: Arc::new(p), + footage: Arc::new(Slow { delay: Duration::from_secs(2), opened: Mutex::new(false) }), + expr: None, + layer_cache: Arc::new(LayerCache::default()), + gpu: None, + gpu_display: false, + disk: None, + }; + let t = Instant::now(); + let mut playback = + AudioPlayback::start(Box::new(Device), src, cid, Tick::ZERO, Tick::ZERO, Tick::from_seconds_f64(10.0), true, 0).expect("the preview starts"); + assert!(t.elapsed() < Duration::from_secs(1), "start waited {:?} for the footage", t.elapsed()); + assert_eq!(playback.clock(), 0, "the clock waits for the sound"); + // The feeder thread delivers it once the footage is open. + while playback.feed.queued_frames() == 0 { + assert!(t.elapsed() < Duration::from_secs(30), "the feeder never delivered"); + std::thread::sleep(Duration::from_millis(10)); + } + let mut out = [0.0f32; 8]; + playback.feed.pull(&mut out); + assert_eq!(out, [0.5; 8]); + playback.stop(); + } + #[test] fn meter_decays_and_holds_peaks() { let mut m = Meter::default();