Merge pull request #796 from vangel76/fix-stream-video-open

#794: open movies in place instead of reading them whole

# Conflicts:
#	crates/media/src/lib.rs

crates/media/src/layered.rs

@@ -40,6 +40,12 @@ fn is_pdf_like(path: &str, bytes: &[u8]) -> bool {
     effectcraft_pdf::sniff(bytes).is_some() && (effectcraft_pdf::EXTENSIONS.contains(&ext.as_str()) || bytes.starts_with(b"%PDF"))
 }
 
+/// Whether `bytes` (the first 64 KiB suffice) are a Photoshop or vector file [`probe`] handles.
+#[cfg_attr(not(any(unix, windows)), allow(dead_code))]
+pub(crate) fn sniff(path: &str, bytes: &[u8]) -> bool {
+    effectcraft_psd::is_psd(bytes) || is_svg(path, bytes) || is_pdf_like(path, bytes)
+}
+
 /// Footage for a Photoshop or vector file (`None` for other formats).
 pub(crate) fn probe(path: &str, bytes: &[u8]) -> Result<Option<Footage>> {
     if effectcraft_psd::is_psd(bytes) {

crates/media/src/lib.rs

@@ -24,6 +24,7 @@ pub use layered::vector_doc;
 mod pool;
 mod probe;
 mod webp_anim;
+mod stream;
 
 pub use pool::{DEFAULT_BUDGET, MediaPool, PoolStats};
 pub use probe::{DEFAULT_SEQUENCE_RATE, probe, probe_bytes, probe_model};

crates/media/src/pool.rs

@@ -579,9 +579,15 @@ impl Inner {
         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)
+            // Files on disk are read in place (the index, then samples on demand) rather than
+            // whole; in-memory files (`add_bytes`) and wasm32 builds open from bytes.
+            let in_memory = lock(&self.files).contains_key(&**path);
+            let in_place = if in_memory { None } else { crate::stream::open_source(std::path::Path::new(&**path)) };
+            let opened = in_place.unwrap_or_else(|| {
+                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),

crates/media/src/probe.rs

@@ -15,6 +15,10 @@ pub const DEFAULT_SEQUENCE_RATE: FrameRate = FrameRate::FPS_30;
 /// (`effectcraft_engine::sequence`) and probes their first frame here.
 pub fn probe(path: impl AsRef<Path>) -> Result<Footage> {
     let path = path.as_ref();
+    // A movie is read in place (its index), not loaded whole.
+    if let Some(src) = crate::stream::probe_source(path) {
+        return Ok(footage_from_info(&path.to_string_lossy(), src?.info()));
+    }
     let bytes = std::fs::read(path).map_err(|e| MediaError::Io(format!("{}: {e}", path.display())))?;
     if crate::MODEL_EXTENSIONS.contains(&ext_of(path).as_str()) {
         let dir = path.parent().map(Path::to_path_buf).unwrap_or_default();

crates/media/src/stream.rs

@@ -0,0 +1,111 @@
+//! Movie and audio files read in place: the container's index now, single samples on demand.
+//!
+//! Reading a file whole before opening it made a clip cost its full size in memory (twice while
+//! probing: the file's bytes plus the demuxer's copy), so a 45-minute movie took gigabytes just to
+//! import. FilmCraft's reader openers (MP4/MOV, Matroska/WebM, MXF, Ogg, MPEG-PS/TS) open from a
+//! random-access [`ByteReader`]; [`FileReader`] serves them positional reads from the open file.
+//! Formats decoded in one go (stills, WAV, FLAC, MP3…) still read the whole file, as before.
+//! Builds without a native file system (wasm32) keep using in-memory bytes.
+
+use std::path::Path;
+
+use filmcraft_media::SharedSource;
+
+use crate::Result;
+
+/// A file on disk read in place: it stays open and only the requested ranges are read.
+#[cfg(any(unix, windows))]
+#[derive(Debug)]
+pub(crate) struct FileReader {
+    file: std::fs::File,
+    len: u64,
+}
+
+#[cfg(any(unix, windows))]
+impl FileReader {
+    pub(crate) fn open(path: &Path) -> std::io::Result<Self> {
+        let file = std::fs::File::open(path)?;
+        let m = file.metadata()?;
+        if !m.is_file() {
+            return Err(std::io::Error::new(std::io::ErrorKind::InvalidInput, format!("{} is not a file", path.display())));
+        }
+        Ok(Self { file, len: m.len() })
+    }
+}
+
+#[cfg(any(unix, windows))]
+impl filmcraft_media::ByteReader for FileReader {
+    fn len(&self) -> u64 {
+        self.len
+    }
+
+    /// Positional reads: threads decoding different parts of the file share no cursor.
+    fn read_at(&self, offset: u64, buf: &mut [u8]) -> std::io::Result<()> {
+        #[cfg(unix)]
+        {
+            std::os::unix::fs::FileExt::read_exact_at(&self.file, buf, offset)
+        }
+        #[cfg(windows)]
+        {
+            let (mut done, mut at) = (0usize, offset);
+            while let Some(rest) = buf.get_mut(done..).filter(|r| !r.is_empty()) {
+                match std::os::windows::fs::FileExt::seek_read(&self.file, rest, at) {
+                    Ok(0) => return Err(std::io::Error::new(std::io::ErrorKind::UnexpectedEof, "read past end of file")),
+                    Ok(n) => {
+                        done = done.saturating_add(n);
+                        at = at.saturating_add(n as u64);
+                    }
+                    Err(e) if e.kind() == std::io::ErrorKind::Interrupted => {}
+                    Err(e) => return Err(e),
+                }
+            }
+            Ok(())
+        }
+    }
+}
+
+#[cfg(any(unix, windows))]
+fn name_of(path: &Path) -> String {
+    path.file_name().map(|n| n.to_string_lossy().to_string()).unwrap_or_default()
+}
+
+/// Open a movie/audio file for the media pool, reading only what its container needs; files the
+/// reader openers don't handle are read whole and opened from memory, exactly as before. `None`
+/// when the file can't be opened in place (the caller reads it whole, which reports the error).
+#[cfg(any(unix, windows))]
+pub(crate) fn open_source(path: &Path) -> Option<Result<SharedSource>> {
+    let reader = FileReader::open(path).ok()?;
+    let reader: filmcraft_media::SharedReader = std::sync::Arc::new(reader);
+    let opened = filmcraft_media::reader::open_reader(&name_of(path), reader, &filmcraft_codecs::reader_openers(), &filmcraft_codecs::openers());
+    Some(opened.map_err(Into::into))
+}
+
+#[cfg(not(any(unix, windows)))]
+pub(crate) fn open_source(_path: &Path) -> Option<Result<SharedSource>> {
+    None
+}
+
+/// Probing in place: the opened container when `path` is a movie/audio container a reader opener
+/// handles and nothing probed before containers (stills, layered and vector documents, 3D models)
+/// claims it. `None`: probe the whole file's bytes as before.
+#[cfg(any(unix, windows))]
+pub(crate) fn probe_source(path: &Path) -> Option<Result<SharedSource>> {
+    let ext = crate::ext_of(path);
+    if crate::STILL_EXTENSIONS.contains(&ext.as_str()) || crate::MODEL_EXTENSIONS.contains(&ext.as_str()) {
+        return None;
+    }
+    let reader = FileReader::open(path).ok()?;
+    let reader: filmcraft_media::SharedReader = std::sync::Arc::new(reader);
+    let head = filmcraft_media::reader::read_range(&*reader, 0, filmcraft_media::reader::HEAD_LEN).ok()?;
+    let name = path.to_string_lossy();
+    if crate::layered::sniff(&name, &head) || image::guess_format(&head).is_ok() {
+        return None;
+    }
+    let name = name_of(path);
+    filmcraft_codecs::reader_openers().into_iter().find_map(|open| open(&name, &head, &reader)).map(|r| r.map_err(Into::into))
+}
+
+#[cfg(not(any(unix, windows)))]
+pub(crate) fn probe_source(_path: &Path) -> Option<Result<SharedSource>> {
+    None
+}

crates/media/tests/open_in_place.rs

@@ -0,0 +1,101 @@
+//! Opening a long movie reads its index, not the whole file: probing it and decoding a frame and
+//! some audio must not grow peak memory by anything near the file's size.
+//!
+//! The large movie is a small ffmpeg-made MP4 followed by a 256 MiB `free` box (a sparse file, so
+//! it costs no disk writes). Reading it whole, as before, grew peak RSS by its size; read in place,
+//! the box is skipped by its header. The only test in this binary, so peak RSS is its own.
+//! Skips (passes with a note) without ffmpeg or off Linux (peak RSS from `/proc/self/status`).
+
+use std::io::Write;
+use std::path::{Path, PathBuf};
+use std::process::Command;
+
+use effectcraft_media::{MediaPool, probe};
+use effectcraft_time::Tick;
+
+const PADDING: u64 = 256 << 20;
+
+fn fixtures() -> PathBuf {
+    Path::new(env!("CARGO_MANIFEST_DIR")).join("../../target/fixtures/media")
+}
+
+/// A 2 s 320×240 H.264 + AAC movie (written to a temp name and renamed, so concurrent test
+/// processes never see a partial file). `None` without ffmpeg.
+fn small_mp4() -> Option<PathBuf> {
+    let out = fixtures().join("in_place_small.mp4");
+    if out.exists() {
+        return Some(out);
+    }
+    if !Command::new("ffmpeg").arg("-version").output().is_ok_and(|o| o.status.success()) {
+        return None;
+    }
+    std::fs::create_dir_all(fixtures()).unwrap();
+    let tmp = fixtures().join(format!("in_place_tmp_{}.mp4", std::process::id()));
+    let st = Command::new("ffmpeg")
+        .args(["-hide_banner", "-loglevel", "error", "-y", "-f", "lavfi", "-i", "testsrc2=size=320x240:rate=30:duration=2"])
+        .args(["-f", "lavfi", "-i", "sine=frequency=440:sample_rate=48000:duration=2"])
+        .args(["-c:v", "libx264", "-pix_fmt", "yuv420p", "-c:a", "aac", "-shortest"])
+        .arg(&tmp)
+        .status()
+        .unwrap();
+    assert!(st.success(), "ffmpeg failed");
+    std::fs::rename(&tmp, &out).unwrap();
+    Some(out)
+}
+
+/// `/proc/self/status` field `key` in KiB.
+fn status_kib(key: &str) -> Option<u64> {
+    let s = std::fs::read_to_string("/proc/self/status").ok()?;
+    let line = s.lines().find(|l| l.starts_with(key))?;
+    line.split_whitespace().nth(1)?.parse().ok()
+}
+
+/// Peak RSS so far in bytes, after resetting it to the current RSS where the kernel allows that.
+fn reset_peak() -> Option<u64> {
+    let _ = std::fs::write("/proc/self/clear_refs", "5");
+    status_kib("VmHWM:").map(|k| k * 1024)
+}
+
+#[test]
+fn a_long_movie_is_read_in_place_not_whole() {
+    if !cfg!(target_os = "linux") {
+        eprintln!("peak RSS is read from /proc: skipping");
+        return;
+    }
+    let Some(small) = small_mp4() else {
+        eprintln!("ffmpeg not found: skipping");
+        return;
+    };
+    // The same movie with 256 MiB of padding after its boxes.
+    let large = fixtures().join(format!("in_place_large_{}.mp4", std::process::id()));
+    std::fs::copy(&small, &large).unwrap();
+    let mut f = std::fs::OpenOptions::new().append(true).open(&large).unwrap();
+    f.write_all(&(8 + PADDING as u32).to_be_bytes()).unwrap();
+    f.write_all(b"free").unwrap();
+    f.set_len(std::fs::metadata(&small).unwrap().len() + 8 + PADDING).unwrap();
+    drop(f);
+
+    // Warm up on the small file (code paths, allocator, decoder tables).
+    let small_footage = probe(&small).unwrap();
+    let pool = MediaPool::new();
+    pool.set_read_ahead(false);
+    let small_frame = pool.frame_at(&small_footage, Tick::ZERO).unwrap();
+    let small_audio = pool.audio_samples(&small_footage, Tick::ZERO, 4800, 48_000);
+
+    let before = reset_peak().unwrap();
+    let footage = probe(&large).unwrap();
+    let frame = pool.frame_at(&footage, Tick::ZERO).unwrap();
+    let audio = pool.audio_samples(&footage, Tick::ZERO, 4800, 48_000);
+    let after = status_kib("VmHWM:").unwrap() * 1024;
+    drop(pool);
+    let _ = std::fs::remove_file(&large);
+
+    let growth = after.saturating_sub(before);
+    eprintln!("peak RSS growth: {} KiB", growth >> 10);
+    assert!(growth < 64 << 20, "probing and decoding a {} MiB movie grew peak RSS by {} MiB", PADDING >> 20, growth >> 20);
+    // Same movie, same results.
+    assert_eq!(footage, effectcraft_project::Footage { path: footage.path.clone(), ..small_footage });
+    assert_eq!(frame, small_frame);
+    assert_eq!(audio, small_audio);
+    assert!(audio.iter().any(|s| s.abs() > 0.1), "the tone decodes");
+}