Phase 2: bound video frame cache + stream the export mux
- VideoManager.frame_cache: unbounded HashMap (grew per distinct frame during playback) -> LruCache evicted by a 256MB byte budget. Byte-budget rather than frame count is robust across resolutions (a 4K frame is ~33MB vs ~2MB at 800x600). unload_video pops per-clip keys (LruCache has no retain). - mux_video_and_audio: stream-merge the two inputs by PTS with one pending packet per stream (O(1) memory) instead of collecting every packet into Vecs first (O(duration)). Output is byte-identical. - export AAC: sanitize the planar-f32 path (non-finite -> 0, finite clamped to [-1,1]) like the integer paths, with a one-time warning. A stray NaN/Inf render sample no longer fails the whole export. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
Skyler Lehmkuhl
parent
3d7cff9ad0
commit
c784816615
|
|
@ -671,14 +671,39 @@ fn convert_chunk_to_planar_i16(interleaved: &[f32], channels: u32) -> Vec<Vec<i1
|
|||
planar
|
||||
}
|
||||
|
||||
/// Convert a chunk of interleaved f32 samples to planar f32 format
|
||||
/// Convert a chunk of interleaved f32 samples to planar f32 format.
|
||||
///
|
||||
/// Non-finite samples (NaN/±Inf) are replaced with `0.0` and finite samples are
|
||||
/// clamped to `[-1.0, 1.0]`: the float encoders (e.g. AAC, which takes `fltp`)
|
||||
/// reject a frame outright on "(near) NaN/+-Inf", failing the whole export, so we
|
||||
/// sanitize here exactly as the integer paths already clamp.
|
||||
fn convert_chunk_to_planar_f32(interleaved: &[f32], channels: u32) -> Vec<Vec<f32>> {
|
||||
let num_frames = interleaved.len() / channels as usize;
|
||||
let mut planar = vec![vec![0.0f32; num_frames]; channels as usize];
|
||||
|
||||
let mut non_finite = 0u64;
|
||||
for (i, chunk) in interleaved.chunks(channels as usize).enumerate() {
|
||||
for (ch, &sample) in chunk.iter().enumerate() {
|
||||
planar[ch][i] = sample;
|
||||
planar[ch][i] = if sample.is_finite() {
|
||||
sample.clamp(-1.0, 1.0)
|
||||
} else {
|
||||
non_finite += 1;
|
||||
0.0
|
||||
};
|
||||
}
|
||||
}
|
||||
if non_finite > 0 {
|
||||
// One-time warning: we sanitized rather than failed, but a non-finite
|
||||
// sample reaching here means something upstream (an effect, automation,
|
||||
// or a source decode) produced NaN/Inf — worth chasing if audio is wrong.
|
||||
use std::sync::atomic::{AtomicBool, Ordering};
|
||||
static WARNED: AtomicBool = AtomicBool::new(false);
|
||||
if !WARNED.swap(true, Ordering::Relaxed) {
|
||||
eprintln!(
|
||||
"⚠️ [EXPORT] sanitized {} non-finite (NaN/Inf) audio sample(s) in a chunk — \
|
||||
check effects/automation/source decode",
|
||||
non_finite
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
|
|
|
|||
|
|
@ -458,9 +458,14 @@ pub struct VideoManager {
|
|||
/// Pool of video decoders, one per clip
|
||||
decoders: HashMap<Uuid, Arc<Mutex<VideoDecoder>>>,
|
||||
|
||||
/// Frame cache: (clip_id, timestamp_ms) -> frame
|
||||
/// Stores raw RGBA data for zero-copy rendering
|
||||
frame_cache: HashMap<(Uuid, i64), Arc<VideoFrame>>,
|
||||
/// Frame cache: (clip_id, timestamp_ms) -> frame. Stores decoded RGBA for
|
||||
/// zero-copy rendering. Bounded by a **byte budget** (not a frame count, which
|
||||
/// would be unsafe across resolutions — a 4K frame is ~33MB vs ~2MB at 800x600)
|
||||
/// so playback of arbitrarily long video never grows unbounded.
|
||||
frame_cache: LruCache<(Uuid, i64), Arc<VideoFrame>>,
|
||||
/// Running total of bytes held in `frame_cache` (sum of each frame's RGBA len),
|
||||
/// kept in sync on insert/evict/remove so eviction is O(1) per frame.
|
||||
frame_cache_bytes: usize,
|
||||
|
||||
/// Thumbnail cache: clip_id -> Vec of (timestamp, rgba_data)
|
||||
/// Low-resolution (64px width) thumbnails for scrubbing
|
||||
|
|
@ -470,6 +475,11 @@ pub struct VideoManager {
|
|||
cache_size: usize,
|
||||
}
|
||||
|
||||
/// Byte budget for [`VideoManager::frame_cache`] (decoded full-resolution frames).
|
||||
/// At ~2MB/frame (800x600) this holds ~128 frames; at ~33MB/frame (4K) ~8 — in
|
||||
/// both cases enough for the current frame plus a scrub window, while bounding RAM.
|
||||
const FRAME_CACHE_BYTE_BUDGET: usize = 256 * 1024 * 1024;
|
||||
|
||||
impl VideoManager {
|
||||
/// Create a new video manager with default cache size
|
||||
pub fn new() -> Self {
|
||||
|
|
@ -480,7 +490,8 @@ impl VideoManager {
|
|||
pub fn with_cache_size(cache_size: usize) -> Self {
|
||||
Self {
|
||||
decoders: HashMap::new(),
|
||||
frame_cache: HashMap::new(),
|
||||
frame_cache: LruCache::unbounded(),
|
||||
frame_cache_bytes: 0,
|
||||
thumbnail_cache: HashMap::new(),
|
||||
cache_size,
|
||||
}
|
||||
|
|
@ -533,14 +544,16 @@ impl VideoManager {
|
|||
return Some(Arc::clone(cached_frame));
|
||||
}
|
||||
|
||||
// Get decoder for this clip
|
||||
let decoder_arc = self.decoders.get(clip_id)?;
|
||||
// Get decoder for this clip. Clone the Arc so we don't hold a borrow of
|
||||
// `self.decoders` across the `&mut self` cache insert below.
|
||||
let decoder_arc = Arc::clone(self.decoders.get(clip_id)?);
|
||||
let mut decoder = decoder_arc.lock().ok()?;
|
||||
|
||||
// Decode the frame
|
||||
let rgba_data = decoder.get_frame(timestamp).ok()?;
|
||||
let width = decoder.output_width;
|
||||
let height = decoder.output_height;
|
||||
drop(decoder); // release the lock before touching `self`
|
||||
|
||||
// Create VideoFrame and cache it
|
||||
let frame = Arc::new(VideoFrame {
|
||||
|
|
@ -550,11 +563,29 @@ impl VideoManager {
|
|||
timestamp,
|
||||
});
|
||||
|
||||
self.frame_cache.insert(cache_key, Arc::clone(&frame));
|
||||
self.cache_frame(cache_key, Arc::clone(&frame));
|
||||
|
||||
Some(frame)
|
||||
}
|
||||
|
||||
/// Insert a frame into the byte-budgeted cache, evicting least-recently-used
|
||||
/// frames until the total is within [`FRAME_CACHE_BYTE_BUDGET`].
|
||||
fn cache_frame(&mut self, key: (Uuid, i64), frame: Arc<VideoFrame>) {
|
||||
let bytes = frame.rgba_data.len();
|
||||
if let Some(old) = self.frame_cache.put(key, frame) {
|
||||
self.frame_cache_bytes = self.frame_cache_bytes.saturating_sub(old.rgba_data.len());
|
||||
}
|
||||
self.frame_cache_bytes += bytes;
|
||||
// Keep at least one frame resident even if it alone exceeds the budget.
|
||||
while self.frame_cache_bytes > FRAME_CACHE_BYTE_BUDGET && self.frame_cache.len() > 1 {
|
||||
if let Some((_, evicted)) = self.frame_cache.pop_lru() {
|
||||
self.frame_cache_bytes = self.frame_cache_bytes.saturating_sub(evicted.rgba_data.len());
|
||||
} else {
|
||||
break;
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// Get the decoder Arc for a clip (for external thumbnail generation)
|
||||
/// This allows external code to decode frames without holding the VideoManager lock
|
||||
pub fn get_decoder(&self, clip_id: &Uuid) -> Option<Arc<Mutex<VideoDecoder>>> {
|
||||
|
|
@ -614,8 +645,19 @@ impl VideoManager {
|
|||
pub fn unload_video(&mut self, clip_id: &Uuid) {
|
||||
self.decoders.remove(clip_id);
|
||||
|
||||
// Remove all cached frames for this clip
|
||||
self.frame_cache.retain(|(id, _), _| id != clip_id);
|
||||
// Remove all cached frames for this clip (LruCache has no retain; collect
|
||||
// matching keys, then pop each, keeping the byte total in sync).
|
||||
let keys: Vec<(Uuid, i64)> = self
|
||||
.frame_cache
|
||||
.iter()
|
||||
.filter(|((id, _), _)| id == clip_id)
|
||||
.map(|(k, _)| *k)
|
||||
.collect();
|
||||
for key in keys {
|
||||
if let Some(frame) = self.frame_cache.pop(&key) {
|
||||
self.frame_cache_bytes = self.frame_cache_bytes.saturating_sub(frame.rgba_data.len());
|
||||
}
|
||||
}
|
||||
|
||||
// Remove thumbnails
|
||||
self.thumbnail_cache.remove(clip_id);
|
||||
|
|
@ -624,6 +666,7 @@ impl VideoManager {
|
|||
/// Clear all frame caches (useful for memory management)
|
||||
pub fn clear_frame_cache(&mut self) {
|
||||
self.frame_cache.clear();
|
||||
self.frame_cache_bytes = 0;
|
||||
}
|
||||
}
|
||||
|
||||
|
|
|
|||
|
|
@ -384,90 +384,90 @@ impl ExportOrchestrator {
|
|||
println!("🎵 [MUX] Audio stream - Input TB: {}/{}, Output TB: {}/{}",
|
||||
audio_input_tb.0, audio_input_tb.1, audio_output_tb.0, audio_output_tb.1);
|
||||
|
||||
// Collect all packets with their stream info and timestamps
|
||||
let mut video_packets = Vec::new();
|
||||
for (stream, packet) in video_input.packets() {
|
||||
if stream.index() == video_stream_index {
|
||||
video_packets.push(packet);
|
||||
// Stream-merge the two inputs by PTS, writing each packet as it's read —
|
||||
// O(1) memory (one pending packet per stream) instead of collecting every
|
||||
// packet first, so muxing a long export never grows unbounded.
|
||||
let video_idx = video_stream_index;
|
||||
let audio_idx = audio_stream_index;
|
||||
let mut v_iter = video_input.packets();
|
||||
let mut a_iter = audio_input.packets();
|
||||
|
||||
// Pull the next packet belonging to the desired stream from each input.
|
||||
let mut next_video = move || -> Option<ffmpeg::Packet> {
|
||||
loop {
|
||||
match v_iter.next() {
|
||||
Some((stream, packet)) => {
|
||||
if stream.index() == video_idx {
|
||||
return Some(packet);
|
||||
}
|
||||
}
|
||||
|
||||
let mut audio_packets = Vec::new();
|
||||
for (stream, packet) in audio_input.packets() {
|
||||
if stream.index() == audio_stream_index {
|
||||
audio_packets.push(packet);
|
||||
None => return None,
|
||||
}
|
||||
}
|
||||
|
||||
println!("🎬 [MUX] Collected {} video packets, {} audio packets",
|
||||
video_packets.len(), audio_packets.len());
|
||||
|
||||
// Report first and last timestamps
|
||||
if !video_packets.is_empty() {
|
||||
println!("🎬 [MUX] Video PTS range: {} to {}",
|
||||
video_packets[0].pts().unwrap_or(0),
|
||||
video_packets[video_packets.len()-1].pts().unwrap_or(0));
|
||||
};
|
||||
let mut next_audio = move || -> Option<ffmpeg::Packet> {
|
||||
loop {
|
||||
match a_iter.next() {
|
||||
Some((stream, packet)) => {
|
||||
if stream.index() == audio_idx {
|
||||
return Some(packet);
|
||||
}
|
||||
if !audio_packets.is_empty() {
|
||||
println!("🎵 [MUX] Audio PTS range: {} to {}",
|
||||
audio_packets[0].pts().unwrap_or(0),
|
||||
audio_packets[audio_packets.len()-1].pts().unwrap_or(0));
|
||||
}
|
||||
None => return None,
|
||||
}
|
||||
}
|
||||
};
|
||||
|
||||
// Interleave packets by comparing timestamps in a common time base (use microseconds)
|
||||
let mut v_idx = 0;
|
||||
let mut a_idx = 0;
|
||||
let mut interleave_log_count = 0;
|
||||
let mut pending_v = next_video();
|
||||
let mut pending_a = next_audio();
|
||||
let mut v_count = 0usize;
|
||||
let mut a_count = 0usize;
|
||||
let mut log_count = 0;
|
||||
|
||||
while v_idx < video_packets.len() || a_idx < audio_packets.len() {
|
||||
let write_video = if v_idx >= video_packets.len() {
|
||||
false // No more video
|
||||
} else if a_idx >= audio_packets.len() {
|
||||
true // No more audio, write video
|
||||
} else {
|
||||
// Compare timestamps - convert both to microseconds
|
||||
let v_pts = video_packets[v_idx].pts().unwrap_or(0);
|
||||
let a_pts = audio_packets[a_idx].pts().unwrap_or(0);
|
||||
|
||||
// Convert to microseconds: pts * 1000000 * tb.num / tb.den
|
||||
let v_us = v_pts * 1_000_000 * video_input_tb.0 as i64 / video_input_tb.1 as i64;
|
||||
let a_us = a_pts * 1_000_000 * audio_input_tb.0 as i64 / audio_input_tb.1 as i64;
|
||||
|
||||
v_us <= a_us // Write video if it comes before or at same time as audio
|
||||
loop {
|
||||
// Write whichever pending packet has the earlier PTS (in a common
|
||||
// microsecond base); when one stream is exhausted, drain the other.
|
||||
let write_video = match (&pending_v, &pending_a) {
|
||||
(None, None) => break,
|
||||
(Some(_), None) => true,
|
||||
(None, Some(_)) => false,
|
||||
(Some(v), Some(a)) => {
|
||||
let v_us = v.pts().unwrap_or(0) * 1_000_000 * video_input_tb.0 as i64
|
||||
/ video_input_tb.1 as i64;
|
||||
let a_us = a.pts().unwrap_or(0) * 1_000_000 * audio_input_tb.0 as i64
|
||||
/ audio_input_tb.1 as i64;
|
||||
v_us <= a_us
|
||||
}
|
||||
};
|
||||
|
||||
if write_video {
|
||||
let mut packet = video_packets[v_idx].clone();
|
||||
let mut packet = pending_v.take().unwrap();
|
||||
packet.set_stream(0);
|
||||
packet.rescale_ts(video_input_tb, video_output_tb);
|
||||
|
||||
if interleave_log_count < 10 {
|
||||
println!("🎬 [MUX] Writing V packet {} - PTS={:?}, DTS={:?}, Duration={:?}",
|
||||
v_idx, packet.pts(), packet.dts(), packet.duration());
|
||||
interleave_log_count += 1;
|
||||
if log_count < 10 {
|
||||
println!("🎬 [MUX] Writing V packet - PTS={:?}, DTS={:?}", packet.pts(), packet.dts());
|
||||
log_count += 1;
|
||||
}
|
||||
|
||||
packet.write_interleaved(&mut output)
|
||||
.map_err(|e| format!("Failed to write video packet: {}", e))?;
|
||||
v_idx += 1;
|
||||
v_count += 1;
|
||||
pending_v = next_video();
|
||||
} else {
|
||||
let mut packet = audio_packets[a_idx].clone();
|
||||
let mut packet = pending_a.take().unwrap();
|
||||
packet.set_stream(1);
|
||||
packet.rescale_ts(audio_input_tb, audio_output_tb);
|
||||
|
||||
if interleave_log_count < 10 {
|
||||
println!("🎵 [MUX] Writing A packet {} - PTS={:?}, DTS={:?}, Duration={:?}",
|
||||
a_idx, packet.pts(), packet.dts(), packet.duration());
|
||||
interleave_log_count += 1;
|
||||
if log_count < 10 {
|
||||
println!("🎵 [MUX] Writing A packet - PTS={:?}, DTS={:?}", packet.pts(), packet.dts());
|
||||
log_count += 1;
|
||||
}
|
||||
|
||||
packet.write_interleaved(&mut output)
|
||||
.map_err(|e| format!("Failed to write audio packet: {}", e))?;
|
||||
a_idx += 1;
|
||||
a_count += 1;
|
||||
pending_a = next_audio();
|
||||
}
|
||||
}
|
||||
|
||||
println!("🎬 [MUX] Wrote {} video packets, {} audio packets", v_idx, a_idx);
|
||||
println!("🎬 [MUX] Wrote {} video packets, {} audio packets", v_count, a_count);
|
||||
|
||||
// Write trailer
|
||||
output.write_trailer().map_err(|e| format!("Failed to write trailer: {}", e))?;
|
||||
|
|
|
|||
Loading…
Reference in New Issue