diff --git a/crates/oak-app/src/oakui/real.rs b/crates/oak-app/src/oakui/real.rs index 6a56084be..0840c6375 100644 --- a/crates/oak-app/src/oakui/real.rs +++ b/crates/oak-app/src/oakui/real.rs @@ -1750,19 +1750,36 @@ impl RealEngine { let version = self.preview_generation; let preview_windows = self.preview_windows.clone(); let done: oak_render::ticket::Completion = Box::new(move |result| { - let mut windows = - preview_windows.lock().unwrap_or_else(|e| e.into_inner()); - let window = windows.entry(monitor).or_default(); - match result { - Ok(oak_render::ticket::TicketPayload::ShmFrame(slot)) => { - // The rendered frame is cached in its shm slot until - // the playhead reaches it (cpu_frame) or it falls out - // of the window. - window.slots.insert(frame, slot); + // Generation gate: a completion queued before a window + // rebuild (edit/generation bump) must not land in the new + // window — the stale render would display at this frame + // and its `submitted` entry would reject the fresh + // request. The stale slot's credit goes back instead. + let mut stale_slot = None; + { + let mut windows = + preview_windows.lock().unwrap_or_else(|e| e.into_inner()); + let window = windows.entry(monitor).or_default(); + if window.sequence == node_id && window.generation == version { + match result { + Ok(oak_render::ticket::TicketPayload::ShmFrame(slot)) => { + // The rendered frame is cached in its shm slot + // until the playhead reaches it (cpu_frame) + // or it falls out of the window. + window.slots.insert(frame, slot); + } + _ => { + // Render failed / cancelled: allow a re-request. + window.submitted.remove(&frame); + } + } + } else if let Ok(oak_render::ticket::TicketPayload::ShmFrame(slot)) = result { + stale_slot = Some(slot); } - _ => { - // Render failed / cancelled: allow a re-request. - window.submitted.remove(&frame); + } + if let Some(slot) = stale_slot { + if let Some(m) = RenderManager::global() { + m.release_frame(&slot); } } }); diff --git a/crates/oak-app/src/oakui/renderops.rs b/crates/oak-app/src/oakui/renderops.rs index c0ab56916..4a896b38b 100644 --- a/crates/oak-app/src/oakui/renderops.rs +++ b/crates/oak-app/src/oakui/renderops.rs @@ -718,9 +718,10 @@ pub fn submit_audio_chunk( let channel_layout = 0x3u64; let channels = channel_layout.count_ones().max(1) as i32; let m = RenderManager::global().ok_or_else(|| "render manager is not initialized".to_string())?; - let id = m.tickets.next_id(); - m.tickets.submit_audio_with_id( - id, + // Completion-only (fire-and-forget): use the non-polling submit so + // the arena reaps the entry once the completion lands (the polling + // variant would pin it forever — nobody calls result() on this path). + m.tickets.submit_audio( AudioTicketParams { viewer: seq.identity(), range, diff --git a/crates/oak-codec/src/decoder.rs b/crates/oak-codec/src/decoder.rs index da86d9538..099f27f00 100644 --- a/crates/oak-codec/src/decoder.rs +++ b/crates/oak-codec/src/decoder.rs @@ -136,6 +136,13 @@ pub struct RetrieveVideoParams { pub mode: RenderMode, /// Frame alpha channel is premultiplied. pub alpha_is_premultiplied: bool, + /// Target output size for the scaled frame: `Some((w, h))` lets the + /// decoder's swscale pass convert AND resize in one step (native + /// yuv → RGBA/F32 at the target size), skipping the full-resolution + /// float intermediate (a 4K frame is ~132 MB as F32 RGBA — decoding + /// to a 480px preview through it costs ~260 MB of churn per frame). + /// `None` keeps the native size (the old behavior). + pub target_size: Option<(u32, u32)>, } /// `Decoder::RetrieveAudioStatus` — outcome of an audio retrieve. @@ -708,6 +715,7 @@ mod tests_unimplemented { image_sequence_number: 0, mode: RenderMode::Offline, alpha_is_premultiplied: false, + target_size: None, }; assert!(d.retrieve_video_frame(&p).is_err()); assert!(d.retrieve_video(&p).is_err()); diff --git a/crates/oak-codec/src/ffmpeg.rs b/crates/oak-codec/src/ffmpeg.rs index 7fb5f22f4..ad7a17350 100644 --- a/crates/oak-codec/src/ffmpeg.rs +++ b/crates/oak-codec/src/ffmpeg.rs @@ -332,7 +332,7 @@ impl Decoder for FFmpegDecoder { let f = decoded? .ok_or_else(|| fail("no video frame available at the requested time"))?; - let (w, h, bytes) = state.scale_video_to_f32(f, p.force_range)?; + let (w, h, bytes) = state.scale_video_to_f32(f, p.force_range, p.target_size)?; let frame = copy_rgba_f32_to_frame(w, h, &bytes, p.time)?; Ok(Arc::new(frame)) } @@ -908,10 +908,15 @@ impl DecoderState { /// Scale a decoded frame to float RGBA (F32, 4 channels), returning the /// raw pixel bytes plus dimensions. Mirrors `pre_process_frame` + /// `retrieve_video_frame_internal` scaling with the color-range forcing. + /// `target_size` resizes in the same swscale pass (native → RGBA/F32 at + /// the target size) instead of converting at native size first — the + /// caller's downscale then degenerates to a plain copy, and no + /// full-resolution float intermediate (~132 MB at 4K) ever exists. fn scale_video_to_f32( &mut self, f: ffmpeg::frame::Video, force_range: i32, + target_size: Option<(u32, u32)>, ) -> crate::error::Result<(u32, u32, Vec)> { let video = self .video @@ -929,7 +934,12 @@ impl DecoderState { ffmpeg::color::Range::MPEG }); - let (w, h) = (f.width(), f.height()); + let (src_w, src_h) = (f.width(), f.height()); + // 目标尺寸（None = 原生；0 边回退原生，swscale 不接受 0）。 + let (w, h) = match target_size { + Some((tw, th)) if tw > 0 && th > 0 => (tw, th), + _ => (src_w, src_h), + }; // swscale cannot reliably output float RGBA on every build: // float output is missing from some static FFmpeg swscale @@ -945,7 +955,7 @@ impl DecoderState { } else { (Pixel::RGBA, false) }; - let ctx = get_or_create_scaler(&mut video.scaler, src_format, w, h, out_fmt, w, h)?; + let ctx = get_or_create_scaler(&mut video.scaler, src_format, src_w, src_h, out_fmt, w, h)?; let mut out = ffmpeg::frame::Video::empty(); ctx.run(&f, &mut out).map_err(ffmpeg_err)?; let stride = out.stride(0); @@ -1389,6 +1399,8 @@ fn get_or_create_scaler( None => true, }; if recreate { + // BILINEAR：与渲染侧原 Rust 双线性重采样器等效（POINT 会 + // 明显锯齿）；同尺寸纯格式转换时滤波器不参与运算。 let ctx = scaling::Context::get( src_format, src_width, @@ -1396,7 +1408,7 @@ fn get_or_create_scaler( dst_format, dst_width, dst_height, - scaling::Flags::POINT, + scaling::Flags::BILINEAR, ) .map_err(ffmpeg_err)?; *cache = Some(ScalingCache { @@ -2448,6 +2460,7 @@ mod tests { image_sequence_number: 0, mode: crate::decoder::RenderMode::Offline, alpha_is_premultiplied: false, + target_size: None, } } diff --git a/crates/oak-codec/src/oiio.rs b/crates/oak-codec/src/oiio.rs index c204c5dee..54ab0ec0d 100644 --- a/crates/oak-codec/src/oiio.rs +++ b/crates/oak-codec/src/oiio.rs @@ -242,6 +242,7 @@ mod tests { image_sequence_number: 0, mode: crate::decoder::RenderMode::Offline, alpha_is_premultiplied: false, + target_size: None, } } diff --git a/crates/oak-codec/src/realmedia_tests.rs b/crates/oak-codec/src/realmedia_tests.rs index 1448f257b..537d3d760 100644 --- a/crates/oak-codec/src/realmedia_tests.rs +++ b/crates/oak-codec/src/realmedia_tests.rs @@ -54,6 +54,7 @@ fn video_params(stream: CodecStream, time: Rational) -> RetrieveVideoParams { image_sequence_number: 0, mode: RenderMode::Offline, alpha_is_premultiplied: false, + target_size: None, } } diff --git a/crates/oak-render/src/eval.rs b/crates/oak-render/src/eval.rs index e302e2d4d..c038ea812 100644 --- a/crates/oak-render/src/eval.rs +++ b/crates/oak-render/src/eval.rs @@ -485,14 +485,31 @@ pub fn render_produced_frame( /// Process-wide open decoder sessions, keyed by (filename, stream). /// Sessions are mutex-serialized inside the oakcodec box, so sharing -/// one handle across worker threads is safe. +/// one handle across worker threads is safe. The value carries an LRU +/// tick: the map is capped ([`MAX_CACHED_DECODERS`]) because every +/// session pins an FFmpeg context plus up to two native decoded frames +/// (~50 MB at 4K) — before the cap, scrubbing a footage bin grew the +/// map without bound. static DECODERS: std::sync::OnceLock< - std::sync::Mutex>>, + std::sync::Mutex< + std::collections::HashMap<(String, i32), (Arc, u64)>, + >, > = std::sync::OnceLock::new(); +/// Cap on cached decoder sessions per process (LRU beyond this). 16 +/// covers heavy multi-clip montages without reopen thrash; eviction only +/// drops the map entry — an in-flight render keeps its Arc alive and the +/// session dies with the last reference (Drop releases FFmpeg). +const MAX_CACHED_DECODERS: usize = 16; + +/// LRU tick source for [`DECODERS`]. +static DECODER_TICK: std::sync::atomic::AtomicU64 = std::sync::atomic::AtomicU64::new(1); + fn decoders( -) -> std::sync::MutexGuard<'static, std::collections::HashMap<(String, i32), Arc>> -{ +) -> std::sync::MutexGuard< + 'static, + std::collections::HashMap<(String, i32), (Arc, u64)>, +> { DECODERS .get_or_init(|| std::sync::Mutex::new(std::collections::HashMap::new())) .lock() @@ -501,9 +518,12 @@ fn decoders( /// Open (or reuse) the decoder session for `(filename, stream_index)`. fn open_decoder(filename: &str, stream_index: i32) -> Result> { + let key = (filename.to_string(), stream_index); + let tick = DECODER_TICK.fetch_add(1, std::sync::atomic::Ordering::Relaxed); { - let cache = decoders(); - if let Some(d) = cache.get(&(filename.to_string(), stream_index)) { + let mut cache = decoders(); + if let Some((d, t)) = cache.get_mut(&key) { + *t = tick; return Ok(d.clone()); } } @@ -513,7 +533,19 @@ fn open_decoder(filename: &str, stream_index: i32) -> Result= MAX_CACHED_DECODERS { + let victim = cache + .iter() + .min_by_key(|(_, (_, t))| *t) + .map(|(k, _)| k.clone()); + let Some(victim) = victim else { + break; + }; + cache.remove(&victim); + } + cache.insert(key, (decoder.clone(), tick)); Ok(decoder) } @@ -527,6 +559,7 @@ pub fn render_footage_frame( format: PixelFormat, ) -> Result { let decoder = open_decoder(filename, stream_index)?; + let (w, h) = size; let params = RetrieveVideoParams { stream: CodecStream::with_block(filename.to_string(), stream_index, None), time, @@ -537,6 +570,13 @@ pub fn render_footage_frame( image_sequence_number: 0, mode: RenderMode::Offline, alpha_is_premultiplied: false, + // 直接按目标尺寸出帧：swscale 一次完成格式转换 + 缩放，不再 + // 产生全分辨率 F32 中间帧（4K 预览每帧省 ~260MB 瞬时拷贝）。 + target_size: if w > 0 && h > 0 { + Some((w as u32, h as u32)) + } else { + None + }, }; let decoded = decoder .retrieve_video_frame(¶ms) @@ -545,7 +585,6 @@ pub fn render_footage_frame( let src_w = decoded.width(); let src_h = decoded.height(); let src_linesize = decoded.linesize_bytes(); - let (w, h) = size; if src_w <= 0 || src_h <= 0 || src_linesize <= 0 || !decoded.is_allocated() { return Err(Error::Failed("footage decode: bad decoded frame".into())); } diff --git a/crates/oak-render/src/procpool.rs b/crates/oak-render/src/procpool.rs index aaea81b7d..80bf9fe42 100644 --- a/crates/oak-render/src/procpool.rs +++ b/crates/oak-render/src/procpool.rs @@ -1005,7 +1005,9 @@ impl ProcessDispatcher { .find(|(_, pt)| &pt.key == key) .map(|(id, _)| *id); if let Some(id) = ticket { - if let Some(pt) = inner.tickets.get_mut(&id) { + // 完成即移除（ticket 表只增不减是内存泄漏——播放时 + // 每秒 50-100 个 ticket 全部永久驻留）。 + if let Some(mut pt) = inner.tickets.remove(&id) { if let Some(done) = pt.done.take() { fired.push((done, Err(Error::State))); } @@ -1224,9 +1226,9 @@ impl ProcessDispatcher { let shm = handle.shm.clone(); handle.held.insert(slot as u32); - let pt = inner.tickets.get_mut(&ticket); + let pt = inner.tickets.remove(&ticket); match pt { - Some(pt) => { + Some(mut pt) => { let key = pt.key; inner.scheduler.frame_done(&key); if let Some(done) = pt.done.take() { @@ -1291,7 +1293,7 @@ impl ProcessDispatcher { // The worker acquired the slot but never published it: the // dispatcher (drainer) hands it back to the free pool. self.recycle_slot(inner, worker, slot); - if let Some(pt) = inner.tickets.get_mut(&ticket) { + if let Some(mut pt) = inner.tickets.remove(&ticket) { inner.scheduler.frame_failed(&pt.key); if let Some(done) = pt.done.take() { fired.push((done, Err(Error::Failed(format!("render failed: {error}"))))); @@ -1573,7 +1575,7 @@ impl ProcessDispatcher { inner.workers[worker].state = WorkerState::PermanentlyDead; for req in reclaimed { inner.scheduler.cancel_key(&req.key); - if let Some(pt) = inner.tickets.get_mut(&req.payload) { + if let Some(mut pt) = inner.tickets.remove(&req.payload) { if let Some(done) = pt.done.take() { fired.push(( done, @@ -1733,17 +1735,25 @@ impl JobDispatch for ProcessDispatcher { SubmitOutcome::Accepted => {} SubmitOutcome::Replaced(old) => { // A newer request for the same key superseded the old - // pending one: cancel the old ticket's completion. - if let Some(pt) = inner.tickets.get_mut(&old.payload) { + // pending one: cancel the old ticket's completion (and + // reap the entry — the table must not grow unbounded). + if let Some(mut pt) = inner.tickets.remove(&old.payload) { if let Some(done) = pt.done.take() { fired.push((done, Err(Error::State))); } } } SubmitOutcome::InFlight => { - // Already claimed by a worker; the rendered frame is - // still valid for the same params (playback window - // slides re-request frames that are in flight). + // Already claimed by a worker under the same key: the + // worker will deliver the OLD ticket only. Fire the new + // ticket's completion as cancelled so its entry does + // not leak and the caller (playback window) may + // re-request once the in-flight render lands. + if let Some(mut pt) = inner.tickets.remove(&id) { + if let Some(done) = pt.done.take() { + fired.push((done, Err(Error::State))); + } + } } } } @@ -1766,7 +1776,7 @@ impl JobDispatch for ProcessDispatcher { let mut inner = lock(&self.inner); let dropped = inner.scheduler.cancel_sequence(sequence); for request in dropped { - if let Some(pt) = inner.tickets.get_mut(&request.payload) { + if let Some(mut pt) = inner.tickets.remove(&request.payload) { if let Some(done) = pt.done.take() { fired.push((done, Err(Error::State))); } @@ -1866,12 +1876,15 @@ impl JobDispatch for ProcessDispatcher { } w.stdin = None; } - // Every ticket still open completes with cancellation. + // Every ticket still open completes with cancellation; the map + // is dropped with the dispatcher (clear for hygiene — leaked + // entries pin shm region views). for (_, pt) in inner.tickets.iter_mut() { if let Some(done) = pt.done.take() { fired.push((done, Err(Error::State))); } } + inner.tickets.clear(); } for (done, result) in fired { done(result); diff --git a/crates/oak-render/src/ticket.rs b/crates/oak-render/src/ticket.rs index 8190c9b1c..0541a0ddf 100644 --- a/crates/oak-render/src/ticket.rs +++ b/crates/oak-render/src/ticket.rs @@ -231,6 +231,13 @@ struct TicketSlot { /// whose render completed must recycle its shm slot — the consumer /// never sees the `ShmFrame` payload). dispatch: Arc, + /// The caller polls `wait()`/`result()` after completion (the sync + /// render path): the arena entry must survive `finish()` until + /// `result()` reaps it. Fire-and-forget tickets (completion-only: + /// playback window, autocacher, manager) are reaped at the next + /// `allocate()` once finished — the arena map must not grow + /// unbounded (playback posts 50-100 tickets/sec). + poll_after: bool, } impl TicketSlot { @@ -348,7 +355,12 @@ impl TicketArena { fn allocate(&self, slot: Arc) -> TicketId { let id = slot.id; - lock(&self.slots).insert(id, slot); + let mut slots = lock(&self.slots); + slots.insert(id, slot); + // Reap finished fire-and-forget tickets (their completions were + // delivered at finish(); nobody will poll them). Poll-path slots + // survive until `result()` reaps them. + slots.retain(|_, s| !(s.is_finished() && !s.poll_after)); id } @@ -365,24 +377,29 @@ impl TicketArena { /// on cancellation (with `Error::State`). The job posts as a Seek /// single-frame request (M15 S2; see /// [`TicketArena::submit_playback`] for the pre-render window). + /// The caller is expected to poll `wait()`/`result()` (the sync + /// render path); the arena entry survives until `result()` reaps it. pub fn submit_video_with_id( &self, id: TicketId, params: VideoTicketParams, done: Completion, ) -> TicketId { - self.submit_video_impl(id, params, done, JobSchedule::seek()) + self.submit_video_impl(id, params, done, JobSchedule::seek(), true) } /// The shared video-ticket submit path: register the slot, post the /// job with `schedule` (M15 S2), and deliver `Error::State` - /// immediately when the backend is gone. + /// immediately when the backend is gone. `poll_after` marks entries + /// the caller reaps via `result()` (sync path) rather than the + /// completion (fire-and-forget — reaped once finished). fn submit_video_impl( &self, id: TicketId, params: VideoTicketParams, done: Completion, schedule: JobSchedule, + poll_after: bool, ) -> TicketId { let meta = TicketMeta { kind: Some(ticket_kind::VIDEO), @@ -404,6 +421,7 @@ impl TicketArena { completion: Mutex::new(Some(done)), result: Mutex::new(None), dispatch: self.dispatch.clone(), + poll_after, }); self.allocate(slot.clone()); @@ -446,6 +464,7 @@ impl TicketArena { params, done, JobSchedule::playback(frame, distance, version), + false, ) } @@ -468,24 +487,38 @@ impl TicketArena { params: VideoTicketParams, done: Completion, ) -> TicketId { - self.submit_video_impl(id, params, done, JobSchedule::background()) + self.submit_video_impl(id, params, done, JobSchedule::background(), false) } /// Submit a video ticket; completion fires exactly once, including /// on cancellation (with `Error::State`). pub fn submit_video(&self, params: VideoTicketParams, done: Completion) -> TicketId { let id = self.next_id(); - self.submit_video_with_id(id, params, done) + self.submit_video_impl(id, params, done, JobSchedule::seek(), false) } /// Submit an audio ticket (range pull; C++ render_audio) with a /// caller-reserved id (allocated by [`TicketArena::next_id`]). The - /// completion fires exactly once. + /// completion fires exactly once. The caller is expected to poll + /// `wait()`/`result()` (the sync path); the arena entry survives + /// until `result()` reaps it. pub fn submit_audio_with_id( &self, id: TicketId, params: AudioTicketParams, done: Completion, + ) -> TicketId { + self.submit_audio_impl(id, params, done, true) + } + + /// The shared audio-ticket submit path (`poll_after` 语义同 + /// [`TicketArena::submit_video_impl`]). + fn submit_audio_impl( + &self, + id: TicketId, + params: AudioTicketParams, + done: Completion, + poll_after: bool, ) -> TicketId { let range = params.range; let meta = TicketMeta { @@ -506,6 +539,7 @@ impl TicketArena { completion: Mutex::new(Some(done)), result: Mutex::new(None), dispatch: self.audio_dispatch.clone(), + poll_after, }); self.allocate(slot.clone()); @@ -562,7 +596,7 @@ impl TicketArena { /// completion fires exactly once. pub fn submit_audio(&self, params: AudioTicketParams, done: Completion) -> TicketId { let id = self.next_id(); - self.submit_audio_with_id(id, params, done) + self.submit_audio_impl(id, params, done, false) } /// True when the ticket has finished. @@ -602,7 +636,9 @@ impl TicketArena { /// The ticket result, when finished (clone; unknown/unfinished ids give /// `None`). pub fn result(&self, id: TicketId) -> Option { - lock(&self.slots).get(&id).and_then(|s| s.result()) + // Terminal read: the poll path's only observation point — reap + // the entry with it (the arena map must not grow unbounded). + lock(&self.slots).remove(&id).and_then(|s| s.result()) } /// Ticket metadata query (C++ property()). @@ -716,6 +752,9 @@ mod tests { "exactly once" ); + // is_finished 必须在 result() 之前查：result() 是终止性读取 + // （连同条目一起回收，arena 表不能无限增长）。 + assert!(arena.is_finished(id)); let res = arena.result(id).unwrap().unwrap(); assert_eq!( match res { @@ -724,7 +763,6 @@ mod tests { }, (4, 4) ); - assert!(arena.is_finished(id)); video.shutdown(); } diff --git a/crates/oak-render/tests/ticket_worker_test.rs b/crates/oak-render/tests/ticket_worker_test.rs index 546b9ae8d..1f4a1a4b8 100644 --- a/crates/oak-render/tests/ticket_worker_test.rs +++ b/crates/oak-render/tests/ticket_worker_test.rs @@ -99,10 +99,12 @@ fn ticket_completion_once_success() { "exactly once" ); + // is_finished 必须在 result() 之前查：result() 是终止性读取 + // （连同条目一起回收，arena 表不能无限增长）。 + assert!(arena.is_finished(id)); let res = arena.result(id).unwrap().unwrap(); assert_eq!(res_video(&res).size(), (8, 4)); assert_eq!(res_video(&res).format(), oak_core::PixelFormat::F32); - assert!(arena.is_finished(id)); d.shutdown(); }