diff --git a/quest/m0/ci/README.md b/quest/m0/ci/README.md index 700b5b7d160..6bb6748c468 100644 --- a/quest/m0/ci/README.md +++ b/quest/m0/ci/README.md @@ -134,7 +134,6 @@ The historical samples locate candidates; they are not controlled benchmarks. - [First shared FFI release](/quest/m0/ci/nightly-ffi-reuse.md) - condition: the backport merges and the next `moq-ffi-v*` tag runs `Release FFI`; verify the wrappers - [Fewer BTreeMap instantiations](/quest/m0/ci/btree-keys.md) - moq-net shares BTreeMap key and value types, kept only if its CPU time drops - [Halve the slowest tests' CPU](/quest/m0/ci/slow-tests.md) - the slowest 1% of tests take under half their 289 CPU-seconds, with coverage kept -- [Recovery clips once](/quest/m0/ci/recovery-clips.md) - `recovery_test`'s FFmpeg clips encode once per run, not once per process - [Padding runs](/quest/m0/ci/quic-padding.md) - moq-quic parses a PADDING run as one frame, keeping the padding stats' meaning - [tracing-test filter](/quest/m0/ci/tracing-filter.md) - moq-tokio's tests stop forcing every crate to format at TRACE - [Stats oversize](/quest/m0/ci/stats-oversize.md) - moq-stats refuses an oversized plain snapshot before parsing it back diff --git a/quest/m0/ci/recovery-clips.md b/quest/m0/ci/recovery-clips.md deleted file mode 100644 index 9548d429ddd..00000000000 --- a/quest/m0/ci/recovery-clips.md +++ /dev/null @@ -1,20 +0,0 @@ -# [S] Recovery test clips encode once per run - -## Goal - -`recovery_test`'s FFmpeg clips, and the reference frames decoded from them, -are produced once per test run instead of once per test process, with every -case and its coverage kept. FFmpeg encode and decode together take 35 to 40% -of its CPU today; the PR reports how much of that each part saved. - -## Plan - -Found by [moq-dev/moq#6987](https://github.com/moq-dev/moq/pull/6987)'s -profile of the slowest tests. Pick one of: committing the clips (about 320 KB -each), generating them in a nextest setup script, or caching them under a -lock keyed by the FFmpeg version. Reusing the encoded clip alone leaves the -per-process reference decode, so share the decoded reference data the same -way or report the encode-only saving. Measure it the way that PR measured its cohort, on the same machine before -and after. - -Public API: none. diff --git a/rs/moq-mux/src/container/ts/recovery_test.rs b/rs/moq-mux/src/container/ts/recovery_test.rs index 77fc7f0ed2a..51f8ee99063 100644 --- a/rs/moq-mux/src/container/ts/recovery_test.rs +++ b/rs/moq-mux/src/container/ts/recovery_test.rs @@ -158,14 +158,73 @@ impl Encode { "mpegts", "pipe:1", ]); - let output = std::process::Command::new("ffmpeg") - .args(&args) - .output() + cached(&args, &[]) + } +} + +/// Run FFmpeg on `input` and return what it writes, failing on any error it logs. +fn ffmpeg(args: &[&str], input: &[u8]) -> Result, String> { + use std::io::Write as _; + let mut child = std::process::Command::new("ffmpeg") + .args(args) + .stdin(std::process::Stdio::piped()) + .stdout(std::process::Stdio::piped()) + .stderr(std::process::Stdio::piped()) + .spawn() + .expect("ffmpeg is in the dev shell"); + let mut stdin = child.stdin.take().unwrap(); + let output = std::thread::scope(|scope| { + // FFmpeg may stop reading at its first error, which the status reports. + scope.spawn(move || stdin.write_all(input)); + child.wait_with_output().unwrap() + }); + let stderr = String::from_utf8_lossy(&output.stderr); + if !output.status.success() || !stderr.is_empty() { + return Err(format!("FFmpeg failed ({}): {stderr}", output.status)); + } + Ok(output.stdout) +} + +/// [`ffmpeg`], run once per FFmpeg build for the same arguments and input. Nextest runs each +/// test in a process of its own, so the clips and their reference decodes are kept beside the +/// test binary, and the first process to need one makes it while the rest wait. +fn cached(args: &[&str], input: &[u8]) -> Vec { + use std::hash::{Hash, Hasher}; + static FFMPEG: std::sync::OnceLock<(std::path::PathBuf, String)> = std::sync::OnceLock::new(); + let (dir, build) = FFMPEG.get_or_init(|| { + // The binary `Command` runs, named by its resolved path (in the Nix dev shell, a store + // path that changes with any of its inputs) and when it last changed. + let name = format!("ffmpeg{}", std::env::consts::EXE_SUFFIX); + let path = std::env::split_paths(&std::env::var_os("PATH").unwrap_or_default()) + .map(|dir| dir.join(&name)) + .find(|path| path.is_file()) .expect("ffmpeg is in the dev shell"); - let stderr = String::from_utf8_lossy(&output.stderr); - assert!(output.status.success() && stderr.is_empty(), "ffmpeg: {stderr}"); - output.stdout + let path = std::fs::canonicalize(path).unwrap(); + let modified = std::fs::metadata(&path).unwrap().modified().unwrap(); + let exe = std::env::current_exe().unwrap(); + let dir = exe.parent().unwrap().with_file_name("recovery-ffmpeg"); + std::fs::create_dir_all(&dir).unwrap(); + (dir, format!("{} {modified:?}", path.display())) + }); + let mut hasher = std::hash::DefaultHasher::new(); + (build, args, input).hash(&mut hasher); + let path = dir.join(format!("{:016x}", hasher.finish())); + + let lock = (std::fs::OpenOptions::new().create(true).truncate(false).write(true)) + .open(path.with_extension("lock")) + .unwrap(); + lock.lock().unwrap(); + match std::fs::read(&path) { + Ok(output) => return output, + Err(err) if err.kind() == std::io::ErrorKind::NotFound => {} + Err(err) => panic!("{}: {err}", path.display()), } + let output = ffmpeg(args, input).unwrap(); + // Renamed into place, so a process killed mid-write leaves no entry behind. + let partial = path.with_extension("partial"); + std::fs::write(&partial, &output).unwrap(); + std::fs::rename(&partial, &path).unwrap(); + output } /// A generated clip, and the encodes its pictures come from. @@ -176,7 +235,7 @@ struct Clip { } impl Profile { - /// The clip, generated once per process. + /// The clip, loaded once per process. fn clip(self) -> &'static Clip { static CLIPS: [std::sync::OnceLock; 8] = [const { std::sync::OnceLock::new() }; 8]; CLIPS[self as usize].get_or_init(|| { @@ -1078,7 +1137,11 @@ struct Decoded { /// Decode `ts` with software FFmpeg, its audio too unless `video_only`, failing on any error it /// logs. Pictures keep their own size, so a resolution change hashes each at its own. fn decode(ts: &[u8], video_only: bool) -> Result { - use std::io::Write as _; + parse(&ffmpeg(&decoder(video_only), ts)?, video_only) +} + +/// The FFmpeg arguments [`decode`] runs with. +fn decoder(video_only: bool) -> Vec<&'static str> { let mut args = vec![ "-nostdin", "-hide_banner", @@ -1106,28 +1169,16 @@ fn decode(ts: &[u8], video_only: bool) -> Result { args.extend(["-map", "0:a:0", "-c:a", "pcm_s16le"]); } args.extend(["-f", "framemd5", "pipe:1"]); - let mut child = std::process::Command::new("ffmpeg") - .args(&args) - .stdin(std::process::Stdio::piped()) - .stdout(std::process::Stdio::piped()) - .stderr(std::process::Stdio::piped()) - .spawn() - .expect("ffmpeg is in the dev shell"); - let mut stdin = child.stdin.take().unwrap(); - let input = ts.to_vec(); - let feed = std::thread::spawn(move || stdin.write_all(&input)); - let output = child.wait_with_output().unwrap(); - // FFmpeg may stop reading at its first error, which the status reports. - let _ = feed.join().unwrap(); - let stderr = String::from_utf8_lossy(&output.stderr); - if !output.status.success() || !stderr.is_empty() { - return Err(format!("FFmpeg failed ({}): {stderr}", output.status)); - } + args +} + +/// What [`decoder`] wrote. +fn parse(output: &[u8], video_only: bool) -> Result { let mut decoded = Decoded { pictures: Vec::new(), samples: 0, }; - for line in String::from_utf8(output.stdout).unwrap().lines() { + for line in std::str::from_utf8(output).unwrap().lines() { if line.starts_with("#tb 0:") { assert_eq!(line, "#tb 0: 1/90000", "pictures are stamped at 90 kHz"); } @@ -1162,7 +1213,7 @@ impl Reference { // Each encode decoded on its own, from its start, keeping the pictures the clip took. let source: Vec<(u64, String)> = (profile.clip().parts.iter()) .flat_map(|(ts, range)| { - let pictures = decode(ts, false).unwrap().pictures; + let pictures = parse(&cached(&decoder(false), ts), false).unwrap().pictures; pictures.into_iter().filter(|(pts, _)| range.contains(pts)) }) .collect();