diff --git a/crates/oak-app/src/oakui/real.rs b/crates/oak-app/src/oakui/real.rs index 006c20bcc..7223cf185 100644 --- a/crates/oak-app/src/oakui/real.rs +++ b/crates/oak-app/src/oakui/real.rs @@ -268,23 +268,23 @@ struct PreviewWindow { } // --------------------------------------------------------------------------- -// Playback audio prefetch (M15 S3) +// Playback audio prefetch (M15 S3; M16 S2: rendered inline) // --------------------------------------------------------------------------- // -// Real-time audio is pulled by the UI tick; rendering it through the worker -// pool adds one IPC round trip and can wait behind a busy render worker. To -// avoid dropouts the chunks are rendered AHEAD asynchronously (tickets -// complete on the dispatcher's poll) and buffered here; the pull never -// blocks on a worker. `AUDIO_PREFETCH_CHUNKS` ahead ≈ 64 ms of audio at -// 60 fps — enough to cover the delivery latency while keeping at most a -// handful of audio slots in flight (the dispatcher's credit flow control -// caps them per worker). +// Real-time audio is pulled by the UI tick. Through the worker pool it had +// one IPC round trip and could wait behind a busy render worker, so the +// chunks are rendered AHEAD (tickets complete on the dispatcher's poll) and +// buffered here. M16 S2 renders audio on the UI tick itself (the manager's +// audio dispatch is inline — see oak_render::manager), so the prefetch +// window now serves as the scheduling cushion instead of the delivery +// latency: the pull never blocks on a worker. `AUDIO_PREFETCH_CHUNKS` ahead +// ≈ 64 ms of audio at 60 fps — enough to cover the delivery latency while +// keeping at most a handful of audio chunks in flight. -/// How many audio chunks are kept rendered ahead of the playhead (M15 S3). -/// Each chunk is one sequence frame of audio; 4 frames ahead ≈ 66 ms at -/// 60 fps and ≈ 160 ms at 25 fps — enough to cover the delivery latency -/// while keeping at most a handful of audio slots in flight (the -/// dispatcher's credit flow control caps them per worker). +/// How many audio chunks are kept rendered ahead of the playhead (M15 +/// S3; M16 S2: rendered inline on the UI tick, so the window is the +/// scheduling cushion). Each chunk is one sequence frame of audio; +/// 4 frames ahead ≈ 66 ms at 60 fps and ≈ 160 ms at 25 fps. const AUDIO_PREFETCH_CHUNKS: i64 = 4; /// The playback-audio prefetch buffer: rendered chunks ordered by start @@ -1106,13 +1106,14 @@ pub struct RealEngine { impl RealEngine { /// Render one tick's worth of audio at the program playhead and queue - /// it for playback (M12 P1; M15 S3: async worker-pool prefetch). The - /// audio chunks are rendered AHEAD in oak-worker (tickets posted - /// through the process dispatcher; completions arrive on the UI tick's - /// poll) and buffered by [`AudioPrefetch`], so the real-time pull never - /// blocks the UI thread on a busy render worker. Failures degrade to - /// silence; when the manager is down the channel stays empty and - /// playback continues video-only. + /// it for playback (M12 P1; M15 S3: worker-pool prefetch; M16 S2: the + /// manager's audio dispatch is inline, so each chunk renders + /// synchronously on this tick and lands in the channel before the drain + /// below). The chunks are submitted AHEAD of the playhead and buffered + /// by [`AudioPrefetch`], so the real-time pull never blocks on a render + /// worker or an IPC round trip. Failures degrade to silence; when the + /// manager is down the channel stays empty and playback continues + /// video-only. fn pull_audio_tick(&mut self, cx: &mut Context) { let (Some(project), Some(seq)) = (self.project.clone(), self.sequence) else { return; @@ -1157,9 +1158,9 @@ impl RealEngine { } st.next_submit += chunk; } - // Drain completed chunks. For the inline fallback backend the - // submit above already ran them synchronously into the channel; for - // the process backend they arrived on this tick's earlier poll. + // Drain completed chunks. With the M16 S2 inline audio dispatch the + // submit above already ran each chunk synchronously into the channel; + // the drain stays for the worker-pool path (Threads/other backends). let rx = self.audio_rx.lock().unwrap_or_else(|e| e.into_inner()); while let Ok((ts, data)) = rx.try_recv() { st.insert(ts, data); diff --git a/crates/oak-app/src/oakui/renderops.rs b/crates/oak-app/src/oakui/renderops.rs index d8d3285a4..5fe9bd96e 100644 --- a/crates/oak-app/src/oakui/renderops.rs +++ b/crates/oak-app/src/oakui/renderops.rs @@ -870,10 +870,11 @@ pub fn render_audio_range( } } -/// Submit one audio-chunk render for the async playback prefetch (M15 -/// S3): the chunk `[start_ts, start_ts + len_ts)` is rendered through the -/// process dispatcher; the completion copies the samples out of the shm -/// slot, releases it and sends `(start_ts, samples)` on `tx`. Render +/// Submit one audio-chunk render for the playback prefetch (M15 S3; M16 +/// S2: rendered inline on the calling tick — the manager's audio dispatch +/// is the inline dispatcher, so this is a synchronous mix into +/// `TicketPayload::Audio`; the shm-slot branch stays for worker-pool +/// backends). The completion sends `(start_ts, samples)` on `tx`. Render /// errors send silence of the expected length so the playback buffer stays /// aligned (the real-time path must never stall the UI thread on a worker). pub fn submit_audio_chunk( diff --git a/crates/oak-render/src/manager.rs b/crates/oak-render/src/manager.rs index a2ec1978d..4006672b9 100644 --- a/crates/oak-render/src/manager.rs +++ b/crates/oak-render/src/manager.rs @@ -58,9 +58,10 @@ pub enum RenderBackendChoice { pub struct RenderManager { /// Video job dispatch (the process dispatcher, M15). pub dispatch: Arc, - /// Audio job dispatch (M15 S3: the process dispatcher, like video — - /// audio renders in oak-worker so plugin crashes are isolated; the - /// inline fallback lives in the arena, design §3.7). + /// Audio job dispatch (M16 S2: deliberately the inline dispatcher — + /// playback audio renders synchronously on the submitting thread so it + /// never queues behind worker video batches; the worker-pool audio path + /// was the M15 S3 default, design §3.7). pub audio_dispatch: Arc, /// Ticket arena. pub tickets: Arc, @@ -124,13 +125,20 @@ impl RenderManager { RenderBackendChoice::Processes(config) => { let dispatcher = ProcessDispatcher::new(config)?; dispatcher.start()?; - // M15 S3: audio rendering runs in the worker pool too (a - // plugin crash during an audio render must not take down - // the main process — design §3.7). The inline dispatcher - // stays as the fallback when the process dispatcher is - // unavailable (teardown). - let fallback = InlineDispatcher::sync(); - (dispatcher.clone(), dispatcher, Some(fallback)) + // M16 S2: audio rendering deliberately runs INLINE on the + // submitting (UI) thread instead of the worker pool. Video + // batches — especially OFX plugin frames at ~100-500 ms + // each — can occupy the workers far longer than the cpal + // output buffer (~4 chunks, ≈100 ms) holds; an audio batch + // parked behind them underruns and the audio goes silent + // while the video is merely slow. Playback audio is a short + // synchronous range pull, so inline mixing keeps audio + // flowing no matter how stuck the video workers are + // (design §3.7 prefers inline before S3). Trade: an + // audio-side plugin crash now takes down the main process, + // and the mix cost lands on the UI tick. + let inline = InlineDispatcher::sync(); + (dispatcher.clone(), inline, None) } }; let tickets = Arc::new(TicketArena::new_with_audio_fallback( diff --git a/crates/oak-worker/src/framecache.rs b/crates/oak-worker/src/framecache.rs new file mode 100644 index 000000000..5a55be9bb --- /dev/null +++ b/crates/oak-worker/src/framecache.rs @@ -0,0 +1,338 @@ +// Oak Video Editor - Non-Linear Video Editor +// Copyright (C) 2026 Oak Team +// +// This program is free software: you can redistribute it and/or modify +// it under the terms of the GNU General Public License as published by +// the Free Software Foundation, either version 3 of the License, or +// (at your option) any later version. +// +// This program is distributed in the hope that it will be useful, +// but WITHOUT ANY WARRANTY; without even the implied warranty of +// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +// GNU General Public License for more details. +// +// You should have received a copy of the GNU General Public License +// along with this program. If not, see . + +//! LRU byte-budgeted cache of rendered F32 pipeline frames (M16 S2). +//! +//! The worker renders every batch ticket through the CPU eval path — an OFX +//! plugin frame can cost 100-500 ms of full graph re-evaluation with no +//! caching anywhere in the pipeline. Preview traffic is heavily repetitive: +//! the pause frame, scrubbing back over already-rendered time, and the +//! in-flight frames that re-render after a plugin is removed all re-request +//! pixels the worker has already produced. This module memoizes the F32 +//! pipeline frame keyed by the **render-deterministic subset** of the ticket +//! spec, so repeated requests become a memcpy into the shm slot instead of a +//! re-render. +//! +//! Keying. The F32 pipeline bytes depend only on +//! (`time`, `width`, `height`, footage source, montage/effects, viewer graph +//! identity) — the fields [`spec_cache_key`] serializes. `ticket`, `slot`, +//! `format` and `channels` are deliberately excluded: they describe the +//! delivery, not the picture. In particular an F32-slot request and a +//! BGRA8-slot request for the same frame share one cache entry (the F32 +//! pipeline renders at `force_format: F32` regardless of the slot format; +//! the end-of-pipe convert happens after the cache). +//! +//! Invalidations. A [`crate::worker::WorkerSession`] clears the whole cache +//! on every successful `load_graph` — the key set does not include the graph +//! contents, and a fresh snapshot (new project or new undo revision) can +//! change what any viewer identity renders. Media files are assumed +//! immutable for preview, matching the rest of the pipeline. +//! +//! Budget. The default budget is 64 MiB (env `OAK_WORKER_FRAME_CACHE_MB`), +//! capped at 64 entries, LRU-evicted. A single entry is never evicted for +//! size — 1080p F32 frames are 33 MB, 4K are 133 MB, so an oversized frame +//! still gets cached (it is the most likely re-request: the pause frame). + +use std::collections::HashMap; + +use serde::Serialize; + +use crate::ipc::BatchTicketSpec; + +/// The default frame-cache budget in bytes (64 MiB). +const DEFAULT_BUDGET_BYTES: u64 = 64 * 1024 * 1024; +/// The default maximum number of cached frames. +const DEFAULT_MAX_ENTRIES: usize = 64; +/// Env override for the budget, in MiB. +const BUDGET_ENV: &str = "OAK_WORKER_FRAME_CACHE_MB"; + +/// One cached frame: its F32 pipeline bytes and the LRU recency stamp. +struct Entry { + bytes: Vec, + stamp: u64, +} + +/// The cache (see the module docs). Not `Clone`, owned by the +/// [`crate::worker::WorkerSession`], touched only on the worker's single +/// loop thread. +pub struct FrameCache { + entries: HashMap, + budget_bytes: u64, + max_entries: usize, + /// Monotonic LRU clock; bumped on every insert and every hit. + clock: u64, +} + +impl FrameCache { + /// An empty cache with the default budget and entry cap (env + /// `OAK_WORKER_FRAME_CACHE_MB` overrides the budget in MiB; garbage + /// values fall back to the default). + pub fn new() -> Self { + let budget = std::env::var(BUDGET_ENV) + .ok() + .and_then(|v| v.trim().parse::().ok()) + .filter(|mb| mb.is_finite() && *mb > 0.0) + .map(|mb| (mb * 1024.0 * 1024.0).round() as u64) + .unwrap_or(DEFAULT_BUDGET_BYTES); + Self::with_limits(budget, DEFAULT_MAX_ENTRIES) + } + + /// An empty cache with explicit limits (tests). + pub fn with_limits(budget_bytes: u64, max_entries: usize) -> Self { + Self { + entries: HashMap::new(), + budget_bytes, + max_entries, + clock: 0, + } + } + + /// The cached F32 bytes for `key`, or `None`. A hit refreshes the LRU + /// recency. + pub fn get(&mut self, key: &str) -> Option<&[u8]> { + let clock = self.clock; + let entry = self.entries.get_mut(key)?; + self.clock = clock.wrapping_add(1); + entry.stamp = self.clock; + Some(&entry.bytes) + } + + /// Insert `bytes` under `key`, LRU-evicting until within budget and the + /// entry cap. The inserted entry is never evicted for size (a single + /// oversized frame still gets cached); older entries yield to it. + pub fn insert(&mut self, key: String, bytes: Vec) { + if bytes.is_empty() { + return; + } + self.clock = self.clock.wrapping_add(1); + let stamp = self.clock; + self.entries.insert( + key, + Entry { + bytes, + stamp, + }, + ); + self.evict_until_within_limits(); + } + + /// Evict least-recently-used entries until the budget and entry cap + /// hold, keeping at least the single most recent entry (which can be + /// larger than the whole budget). + fn evict_until_within_limits(&mut self) { + loop { + if self.entries.len() <= self.max_entries + && self.entries.values().map(|e| e.bytes.len() as u64).sum::() + <= self.budget_bytes + { + return; + } + if self.entries.len() <= 1 { + return; + } + // Drop the entry with the smallest stamp. + let lru = self + .entries + .iter() + .min_by_key(|(_, e)| e.stamp) + .map(|(k, _)| k.clone()); + if let Some(lru) = lru { + self.entries.remove(&lru); + } else { + return; + } + } + } + + /// Drop every entry (the worker clears on each successful `load_graph`). + pub fn clear(&mut self) { + self.entries.clear(); + } + + /// Number of cached frames (test introspection). + #[cfg(test)] + pub fn len(&self) -> usize { + self.entries.len() + } + + /// Total cached bytes (test introspection). + #[cfg(test)] + pub fn bytes(&self) -> u64 { + self.entries.values().map(|e| e.bytes.len() as u64).sum() + } +} + +impl Default for FrameCache { + fn default() -> Self { + Self::new() + } +} + +/// The render-deterministic ticket-spec subset (see the module docs): the +/// fields whose values determine the F32 pipeline frame, excluding the +/// delivery fields (`ticket`, `slot`, `format`, `channels`). +#[derive(Serialize)] +struct SpecKey { + time_num: i64, + time_den: i64, + width: i32, + height: i32, + footage_file: String, + footage_stream: i32, + montage: Vec, + viewer_node: u64, + project_key: String, +} + +/// The frame-cache key for `spec`: the serialized render-deterministic +/// subset. Serialization cannot fail for these plain fields; a degenerate +/// failure yields the empty key (a cache miss, never a wrong hit). +pub fn spec_cache_key(spec: &BatchTicketSpec) -> String { + let key = SpecKey { + time_num: spec.time_num, + time_den: spec.time_den, + width: spec.width, + height: spec.height, + footage_file: spec.footage_file.clone(), + footage_stream: spec.footage_stream, + montage: spec.montage.clone(), + viewer_node: spec.viewer_node, + project_key: spec.project_key.clone(), + }; + serde_json::to_string(&key).unwrap_or_default() +} + +#[cfg(test)] +mod tests { + use super::*; + + /// A default spec; callers tweak the fields they want to test. + fn spec() -> BatchTicketSpec { + BatchTicketSpec { + ticket: 7, + slot: 2, + time_num: 100, + time_den: 25, + width: 1920, + height: 1080, + format: crate::ipc::SLOT_FORMAT_BGRA8, + channels: 4, + ..Default::default() + } + } + + #[test] + fn key_ignores_delivery_fields() { + let base = spec(); + // Different ticket/slot/format/channels — same picture. + let other = BatchTicketSpec { + ticket: 99, + slot: 17, + format: 0, + channels: 2, + ..base.clone() + }; + assert_eq!(spec_cache_key(&base), spec_cache_key(&other)); + } + + #[test] + fn key_captures_render_fields() { + let base = spec(); + let time_shifted = BatchTicketSpec { + time_num: base.time_num + 1, + ..base.clone() + }; + assert_ne!(spec_cache_key(&base), spec_cache_key(&time_shifted)); + let resized = BatchTicketSpec { + width: 1280, + height: 720, + ..base.clone() + }; + assert_ne!(spec_cache_key(&base), spec_cache_key(&resized)); + } + + #[test] + fn hit_returns_same_bytes_and_refreshes_lru() { + let mut cache = FrameCache::with_limits(1024 * 1024, 64); + let key = "k1".to_string(); + assert!(cache.get(&key).is_none()); + cache.insert(key.clone(), vec![1u8; 64]); + assert_eq!(cache.get(&key), Some(vec![1u8; 64].as_slice())); + // A hit must not drop the entry. + assert_eq!(cache.len(), 1); + } + + #[test] + fn evicts_by_byte_budget_keeping_oversized_single() { + // Budget 100 bytes: two 64-byte frames cannot both fit, but a single + // 200-byte frame must stay cached. + let mut cache = FrameCache::with_limits(100, 64); + cache.insert("a".to_string(), vec![1u8; 64]); + assert_eq!(cache.bytes(), 64); + cache.insert("b".to_string(), vec![2u8; 64]); + // Over budget: the older entry (a) is evicted, b stays. + assert_eq!(cache.len(), 1); + assert!(cache.get("a").is_none()); + assert_eq!(cache.get("b"), Some(vec![2u8; 64].as_slice())); + + // A single entry larger than the whole budget is still cached. + let mut cache = FrameCache::with_limits(100, 64); + cache.insert("big".to_string(), vec![3u8; 200]); + assert_eq!(cache.len(), 1); + assert_eq!(cache.bytes(), 200); + assert_eq!(cache.get("big"), Some(vec![3u8; 200].as_slice())); + } + + #[test] + fn evicts_by_entry_cap() { + let mut cache = FrameCache::with_limits(1 << 20, 2); + cache.insert("a".to_string(), vec![1u8; 8]); + cache.insert("b".to_string(), vec![2u8; 8]); + cache.insert("c".to_string(), vec![3u8; 8]); + assert_eq!(cache.len(), 2); + assert!(cache.get("a").is_none()); + assert!(cache.get("b").is_some()); + assert!(cache.get("c").is_some()); + } + + #[test] + fn lru_eviction_prefers_untouched_entries() { + let mut cache = FrameCache::with_limits(1 << 20, 3); + cache.insert("a".to_string(), vec![1u8; 8]); + cache.insert("b".to_string(), vec![2u8; 8]); + // Touch "a" so it is the most recent; then "b" is the LRU. + let _ = cache.get("a"); + cache.insert("c".to_string(), vec![3u8; 8]); + cache.insert("d".to_string(), vec![4u8; 8]); + assert_eq!(cache.len(), 3); + assert!(cache.get("b").is_none(), "untouched b is evicted first"); + assert!(cache.get("a").is_some()); + assert!(cache.get("c").is_some()); + assert!(cache.get("d").is_some()); + } + + #[test] + fn clear_drops_everything() { + let mut cache = FrameCache::with_limits(1 << 20, 64); + cache.insert("a".to_string(), vec![1u8; 8]); + cache.insert("b".to_string(), vec![2u8; 8]); + cache.clear(); + assert_eq!(cache.len(), 0); + assert_eq!(cache.bytes(), 0); + assert!(cache.get("a").is_none()); + assert!(cache.get("b").is_none()); + } +} diff --git a/crates/oak-worker/src/main.rs b/crates/oak-worker/src/main.rs index 8ef665fa5..f294402dc 100644 --- a/crates/oak-worker/src/main.rs +++ b/crates/oak-worker/src/main.rs @@ -30,6 +30,7 @@ #![deny(unsafe_op_in_unsafe_fn)] #![warn(missing_docs)] +mod framecache; mod ipc; mod worker; diff --git a/crates/oak-worker/src/worker.rs b/crates/oak-worker/src/worker.rs index 5564a7a93..4138a684f 100644 --- a/crates/oak-worker/src/worker.rs +++ b/crates/oak-worker/src/worker.rs @@ -59,6 +59,7 @@ use oak_render::backend::{BackendKind, DisplayRenderer}; use oak_render::eval; use oak_render::ticket::{AudioTicketParams, MontageClip, VideoTicketParams}; +use crate::framecache::FrameCache; use crate::ipc::{ error_message, write_message, AudioTicketSpec, BatchTicketSpec, FrameSlotPool, FrameSlotMeta, HandshakeMsg, LoadGraphMsg, PluginProgressMsg, RenderAudioBatchMsg, RenderBatchMsg, @@ -288,6 +289,10 @@ pub struct WorkerSession { /// Reusable F32 staging buffer (BGRA8 slot conversion; the F32 /// pipeline renders there before the end-of-pipe format convert). f32_scratch: Vec, + /// LRU byte-budgeted memo of rendered F32 frames, keyed by the + /// render-deterministic ticket-spec subset (M16 S2; see + /// [`crate::framecache`]). + frame_cache: FrameCache, } impl WorkerSession { @@ -323,6 +328,7 @@ impl WorkerSession { input_pool: None, graph: None, f32_scratch: Vec::new(), + frame_cache: FrameCache::new(), }) } @@ -582,6 +588,10 @@ impl WorkerSession { id_map, project_copy: 0, }); + // A fresh graph snapshot can change what any viewer + // identity renders; cached pixels from the previous + // graph must not be served (M16 S2 frame cache). + self.frame_cache.clear(); log_error("LoadGraph: oaknode project deserialized"); None } @@ -597,6 +607,9 @@ impl WorkerSession { id_map: std::collections::HashMap::new(), project_copy: pc, }); + // See the full-snapshot branch: any reload + // invalidates the frame cache (M16 S2). + self.frame_cache.clear(); log_error(&format!( "LoadGraph: identity-only snapshot (project_copy {pc})" )); @@ -1055,26 +1068,57 @@ impl WorkerSession { std::slice::from_raw_parts_mut(pool.slot_data(spec.slot as u32), pool.slot_data_bytes()) }; + // M16 S2 frame cache: memoized F32 pipeline bytes keyed by the + // render-deterministic spec subset (ticket/slot/format/channels are + // delivery, not picture — an F32 and a BGRA8 request for the same + // frame share one entry). A hit is a copy (plus the end-of-pipe + // convert for BGRA8 slots); a miss renders as before and memoizes + // the F32 bytes. See [`crate::framecache`]. + let f32_need = (w as usize) * (h as usize) * 16; + let key = crate::framecache::spec_cache_key(spec); + let mut cached = self.frame_cache.get(&key).map(|b| b.to_vec()); + if let Some(c) = &cached { + if c.len() < f32_need { + // Defensive: an entry smaller than this geometry is not for + // this frame (keys are geometry-signed); re-render. + cached = None; + } + } + if !bgra8 { // F32 RGBA: render straight into the slot (no staging copy). - render_f32_into(spec, ¶ms, &self.graph, time, (w, h), &mut dst[..dst_need])?; + match &cached { + Some(c) => dst[..f32_need].copy_from_slice(&c[..f32_need]), + None => { + render_f32_into(spec, ¶ms, &self.graph, time, (w, h), &mut dst[..dst_need])?; + self.frame_cache.insert(key, dst[..f32_need].to_vec()); + } + } } else { - // BGRA8: render the F32 pipeline frame into the session - // scratch, then convert into the slot (the end-of-pipe format - // convert is not an extra frame copy, design §3.1). - let f32_need = (w as usize) * (h as usize) * 16; - if self.f32_scratch.len() < f32_need { - self.f32_scratch.resize(f32_need, 0); + // BGRA8: the F32 pipeline frame comes from the cache or the + // session scratch, then converts into the slot (the end-of-pipe + // format convert is not an extra frame copy, design §3.1). + match &cached { + Some(c) => { + convert_f32_rgba_to_bgra8(&c[..f32_need], &mut dst[..dst_need]); + } + None => { + if self.f32_scratch.len() < f32_need { + self.f32_scratch.resize(f32_need, 0); + } + render_f32_into( + spec, + ¶ms, + &self.graph, + time, + (w, h), + &mut self.f32_scratch[..f32_need], + )?; + self.frame_cache + .insert(key, self.f32_scratch[..f32_need].to_vec()); + convert_f32_rgba_to_bgra8(&self.f32_scratch[..f32_need], &mut dst[..dst_need]); + } } - render_f32_into( - spec, - ¶ms, - &self.graph, - time, - (w, h), - &mut self.f32_scratch[..f32_need], - )?; - convert_f32_rgba_to_bgra8(&self.f32_scratch[..f32_need], &mut dst[..dst_need]); } // Slot meta (fresh each publish).