From cbd4ba24427a5581936ae91f5a57087791e4f0d4 Mon Sep 17 00:00:00 2001 From: Mike Solar Date: Sat, 29 Aug 2026 05:01:55 +0800 Subject: [PATCH] optimize: video and audio render --- crates/oak-app/src/app.rs | 201 +++++++-- crates/oak-app/src/oakui/audio_thread.rs | 178 ++++++++ crates/oak-app/src/oakui/mod.rs | 1 + crates/oak-app/src/oakui/real.rs | 544 ++++++++++++++++++----- crates/oak-app/src/oakui/renderops.rs | 57 +-- crates/oak-audio/src/manager.rs | 9 + crates/oak-audio/src/outputdevice.rs | 3 + crates/oak-audio/src/previewdevice.rs | 20 + crates/oak-codec/src/ffmpeg.rs | 240 ++++++---- crates/oak-codec/src/hwdecode.rs | 116 ++++- crates/oak-codec/src/realmedia_tests.rs | 79 ++++ crates/oak-common/src/colormath.rs | 484 +++++++++++++------- crates/oak-render/src/eval.rs | 20 +- crates/oak-render/src/manager.rs | 27 ++ crates/oak-render/src/worker.rs | 83 ++++ crates/oak-worker/src/worker.rs | 156 +++++++ 16 files changed, 1776 insertions(+), 442 deletions(-) create mode 100644 crates/oak-app/src/oakui/audio_thread.rs diff --git a/crates/oak-app/src/app.rs b/crates/oak-app/src/app.rs index df7e2e055..c2f62c281 100644 --- a/crates/oak-app/src/app.rs +++ b/crates/oak-app/src/app.rs @@ -105,6 +105,7 @@ pub(crate) mod menu_ids { pub const FOCUS_TIMELINE: usize = ActionId::FocusTimeline.menu_id(); pub const FOCUS_EFFECT_LIBRARY: usize = ActionId::FocusEffectLibrary.menu_id(); pub const FOCUS_MULTICAM: usize = ActionId::FocusMulticam.menu_id(); + pub const RESET_DEFAULT_LAYOUT: usize = ActionId::ResetDefaultLayout.menu_id(); } /// Modal-dialog control ids (see [`ModalEvent::control`]). @@ -636,32 +637,10 @@ impl OakApp { ); }); - // Tune the default split ratios: viewers 60% / timeline 40%, project - // bin 17% of the row. The timeline share leaves room for all four - // tracks (V2/V1 video + A1/A2 audio) plus the ruler and toolbar at - // 1600×900; the viewers keep the remaining ~60%. The program viewer - // (with its audio level strip) is the active tab of its group, so the - // shell opens on the design's visible 素材查看器 | 序列查看器 row rather - // than on the node editor. + // Tune the default split ratios and active tabs — the shared tuning + // block (`tune_default_layout`) that 窗口 → 重置布局 also applies. let mut layout: DockLayout = dock.read(cx).layout().clone(); - layout.resize_split(&NodePath(vec![]), 0.60); - layout.resize_split(&NodePath(vec![0]), 0.17); - // The program viewer's transport row (six transport buttons, the - // timecode, the 安全框/缩放 toggles) plus its 26px meter strip needs - // ~430px at 1600×900 — more than an equal share of the row gives it, - // and the design makes the program monitor the prominent viewer. Tilt - // the source/program and program/inspector boundaries accordingly so - // the transport's trailing toggles are not clipped. - layout.resize_split_child(&NodePath(vec![0]), 1, 0.52); - layout.resize_split_child(&NodePath(vec![0]), 2, 0.62); - if let Some(path) = layout.find_panel(PROGRAM_VIEWER) { - layout.set_tabs_active(&path, PROGRAM_VIEWER); - } - // The project bin is the active tab of its group (the effect library - // sits behind it). - if let Some(path) = layout.find_panel(PROJECT) { - layout.set_tabs_active(&path, PROJECT); - } + tune_default_layout(&mut layout); dock.update(cx, |dock, cx| dock.set_layout(layout, cx)); // --- status bar ---------------------------------------------------- @@ -1104,6 +1083,7 @@ impl OakApp { A::FocusTimeline => self.toggle_panel(TIMELINE, cx), A::FocusEffectLibrary => self.toggle_panel(EFFECT_LIBRARY, cx), A::FocusMulticam => self.toggle_panel(MULTICAM, cx), + A::ResetDefaultLayout => self.reset_default_layout(cx), // --- Tools ----------------------------------------------------- A::Snapping => { let enabled = !self.timeline.read(cx).state.snap_enabled; @@ -1431,6 +1411,53 @@ impl OakApp { } } + /// 窗口 → 重置布局: rebuilds the dock from scratch with every panel at its + /// design position and the default ratios and active tabs restored. + /// Docked panels are removed through the dock's remove flow; floating + /// (tear-off) panels have their window closed for good. + fn reset_default_layout(&mut self, cx: &mut Context) { + let dock = self.dock.clone(); + let ids: Vec = WINDOW_PANELS + .iter() + .map(|(id, _)| *id) + .filter(|id| dock.read(cx).is_panel_visible(*id)) + .collect(); + dock.update(cx, |dock, cx| { + for id in ids { + if dock.is_floating(id) { + dock.close_floating(id, cx); + } else { + let _ = dock.remove_panel(id, cx); + } + } + }); + self.apply_default_layout(cx); + self.rebuild_menu_bar(cx); + cx.notify(); + } + + /// (Re-)seeds every panel at its design position, mirroring the seeding in + /// [`OakApp::new`]: panels are added in the 窗口 menu's order + /// ([`WINDOW_PANELS`]) at their [`default_dock_target`] placement, then + /// the default split ratios and active tabs are applied. + fn apply_default_layout(&mut self, cx: &mut Context) { + let mut panels = Vec::new(); + for (id, _) in WINDOW_PANELS { + if let Some(handle) = self.panel_handle(id, cx) { + panels.push((handle, default_dock_target(id))); + } + } + let dock = self.dock.clone(); + dock.update(cx, |dock, cx| { + for (handle, target) in panels { + let _ = dock.add_panel(handle, target, cx); + } + let mut layout = dock.layout().clone(); + tune_default_layout(&mut layout); + dock.set_layout(layout, cx); + }); + } + /// Switches the UI language live: updates the [`i18n`] global, rebuilds /// the menu bar (so the menu labels and the language checkmark move /// immediately), and repaints the whole shell. @@ -2602,6 +2629,29 @@ fn panel_bit(id: PanelId) -> u16 { 1u16 << (id.raw() as u16) } +/// Applies the design's default split ratios and active tabs to `layout`, +/// shared by [`OakApp::new`] and the 窗口 → 重置布局 action. +/// +/// Viewers 60% / timeline 40%, project bin 17% of the row — the timeline +/// share leaves room for all four tracks (V2/V1 video + A1/A2 audio) plus +/// the ruler and toolbar at 1600×900, the viewers keep the remaining ~60%. +/// The program viewer's transport row plus its 26px meter strip needs ~430px +/// at 1600×900, so the source/program and program/inspector boundaries are +/// tilted (0.52 / 0.62) to keep the transport's trailing toggles unclipped. +/// The program viewer and project bin end up active in their groups. +fn tune_default_layout(layout: &mut DockLayout) { + layout.resize_split(&NodePath(vec![]), 0.60); + layout.resize_split(&NodePath(vec![0]), 0.17); + layout.resize_split_child(&NodePath(vec![0]), 1, 0.52); + layout.resize_split_child(&NodePath(vec![0]), 2, 0.62); + if let Some(path) = layout.find_panel(PROGRAM_VIEWER) { + layout.set_tabs_active(&path, PROGRAM_VIEWER); + } + if let Some(path) = layout.find_panel(PROJECT) { + layout.set_tabs_active(&path, PROJECT); + } +} + /// The design's default dock position for each panel, mirroring the seeding /// in [`OakApp::new`] — 项目 | 素材查看器 | 序列查看器+节点编辑器 | /// 检查器+历史记录 row, timeline full width at the bottom. Used to re-open a @@ -3321,6 +3371,107 @@ mod tests { assert_eq!(cx.read(|app| inspector_checked(app)), Some(true)); } + /// 窗口 → 重置布局 tears the current dock arrangement down and rebuilds + /// every panel at its design position with the default ratios: panels + /// dismissed via the 窗口 menu come back, floating (tear-off) panels' + /// windows are closed for good, and tilted splits are restored. + #[gpui::test] + async fn reset_layout_restores_the_default_workspace(cx: &mut TestAppContext) { + let _guard = crate::actions::shortcuts_test_lock() + .lock() + .unwrap_or_else(|e| e.into_inner()); + let _guard = crate::i18n::lang_test_lock() + .lock() + .unwrap_or_else(|e| e.into_inner()); + let (_window, root) = mock_shell(cx); + + // Distort the workspace on every axis: dismiss two panels, tear the + // timeline off into its own window, and tilt the root split to 0.5. + cx.update(|app| root.update(app, |app, cx| app.on_menu(menu_ids::FOCUS_INSPECTOR, cx))); + cx.update(|app| root.update(app, |app, cx| app.on_menu(menu_ids::FOCUS_HISTORY, cx))); + cx.run_until_parked(); + cx.update(|app| { + root.update(app, |app, cx| { + let dock = app.dock.clone(); + dock.update(cx, |dock, cx| { + assert!(dock.float_panel(TIMELINE, cx)); + }); + }); + }); + cx.run_until_parked(); + cx.update(|app| { + root.update(app, |app, cx| { + let mut layout = app.dock.read(cx).layout().clone(); + layout.resize_split(&NodePath(vec![]), 0.5); + let dock = app.dock.clone(); + dock.update(cx, |dock, cx| dock.set_layout(layout, cx)); + }); + }); + cx.run_until_parked(); + + // The distorted state is in place. + assert_eq!( + cx.read(|app| { + let dock = root.read(app).dock.read(app); + ( + dock.is_docked(INSPECTOR), + dock.is_docked(HISTORY), + dock.is_floating(TIMELINE), + dock + .layout() + .split_ratios(&NodePath(vec![])) + .map(|r| r[0]), + ) + }), + (false, false, true, Some(0.5)) + ); + + // 窗口 → 重置布局 restores every panel to its design position. + cx.update(|app| { + root.update(app, |app, cx| app.on_menu(menu_ids::RESET_DEFAULT_LAYOUT, cx)); + }); + cx.run_until_parked(); + + assert!( + cx.read(|app| { + let dock = root.read(app).dock.read(app); + let all_docked = + WINDOW_PANELS.iter().all(|(panel, _)| dock.is_docked(*panel)); + let none_floating = + WINDOW_PANELS.iter().all(|(panel, _)| !dock.is_floating(*panel)); + let root_ratio = dock + .layout() + .split_ratios(&NodePath(vec![])) + .map(|r| r[0]); + all_docked && none_floating && (root_ratio.unwrap_or(0.0) - 0.60).abs() < 1e-4 + }), + "every panel is docked at the default 0.60 split after the reset" + ); + + // The 窗口 menu's checkmarks follow the restored visible-panel set. + let mask = cx.read(|app| open_panels_mask(root.read(app).dock.read(app))); + let mut state = MenuState::new(true); + state.open_panels = mask; + let window_menu = make_menus(state) + .into_iter() + .find(|e| e.title == crate::i18n::tr("menu.window")) + .expect("Window menu exists") + .menu; + for (panel, action) in WINDOW_PANELS { + let item = window_menu + .items + .iter() + .find(|item| item.id == action.menu_id()) + .unwrap_or_else(|| panic!("Window menu is missing panel {}", panel.raw())); + assert_eq!( + item.checked, + Some(true), + "{} is re-checked after the reset", + panel.raw() + ); + } + } + /// Closing a panel via its tab ✕ — the dock's own close flow — also /// refreshes the 窗口 menu: the dock's structural event rebuilds the menu /// bar, so the dismissed panel loses its checkmark without waiting for any diff --git a/crates/oak-app/src/oakui/audio_thread.rs b/crates/oak-app/src/oakui/audio_thread.rs new file mode 100644 index 000000000..bb0750a49 --- /dev/null +++ b/crates/oak-app/src/oakui/audio_thread.rs @@ -0,0 +1,178 @@ +// 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 . + +//! Dedicated audio-render thread (M16 S3): audio chunk renders never run +//! on the UI tick. +//! +//! Playback prefetch submits one audio chunk per frame tick. Those chunks +//! used to be rendered synchronously by the ticket arena's inline audio +//! dispatcher — i.e. inside the UI tick itself: a slow decode (opening a +//! file, hardware probe) froze the UI and delayed the next chunk, starving +//! the audio device (underruns → pops/clicks). This module owns a single +//! background thread that runs the same [`oak_render::eval::render_audio_samples`] +//! path the ticket arena uses; the UI tick only queues the job and returns +//! immediately. +//! +//! The thread starts lazily on the first submit and lives for the process. +//! A single thread is deliberate: the montage decoders are not safe to run +//! concurrently, and one chunk per frame tick stays far below real-time +//! decode throughput. Render errors degrade to silence of the exact length +//! the success path would produce, so the playback buffer never desyncs. + +use std::sync::{mpsc, OnceLock}; + +use oak_render::ticket::{AudioTicketParams, TicketPayload}; + +use super::renderops::RenderedAudio; + +/// One queued audio chunk: render `params`, then send `(start_ts, data)` +/// on `done`. +struct AudioJob { + params: AudioTicketParams, + start_ts: i64, + done: mpsc::Sender<(i64, RenderedAudio)>, +} + +/// The audio-render thread's job queue (set on first use; the sender is +/// kept alive by this `OnceLock` for the whole process). +static TX: OnceLock> = OnceLock::new(); + +/// Queue `params` for rendering on the audio thread; the rendered samples +/// (or aligned silence on failure) are sent on `done` as +/// `(start_ts, data)`. Never blocks — a UI tick must not wait on a worker. +pub fn submit( + params: AudioTicketParams, + start_ts: i64, + done: mpsc::Sender<(i64, RenderedAudio)>, +) -> Result<(), String> { + let tx = ensure_thread(); + tx.send(AudioJob { params, start_ts, done }) + .map_err(|_| "the audio render thread has exited".to_string()) +} + +/// The queue sender, starting the render thread on first use. +fn ensure_thread() -> mpsc::Sender { + if let Some(tx) = TX.get() { + return tx.clone(); + } + let (tx, rx) = mpsc::channel::(); + if TX.set(tx.clone()).is_ok() { + // We won the race and own the receiver; start the thread. The + // loser's `rx` is dropped right here. + std::thread::Builder::new() + .name("oak-audio-render".to_string()) + .spawn(move || render_loop(rx)) + .expect("spawn the audio render thread"); + } + TX.get().expect("audio thread sender was set above").clone() +} + +/// The render loop: one job at a time; failures degrade to silence. +fn render_loop(rx: mpsc::Receiver) { + while let Ok(job) = rx.recv() { + let data = match oak_render::eval::render_audio_samples(&job.params) { + Ok(TicketPayload::Audio(samples)) => RenderedAudio { + data: samples.samples, + sample_rate: samples.sample_rate, + channel_count: samples.channel_count, + }, + _ => silence_for(&job.params), + }; + let _ = job.done.send((job.start_ts, data)); + } +} + +/// Aligned silence for `params`: the same anchored-grid length +/// (`round(out·rate) − round(in·rate)`) the successful path produces, so +/// a failed chunk keeps the playback buffer aligned. +fn silence_for(params: &AudioTicketParams) -> RenderedAudio { + let rate = params.sample_rate.max(1) as f64; + let channels = params.channel_layout.count_ones().max(1) as i32; + let frames = ((params.range.out().to_f64() * rate).round() + - (params.range.in_().to_f64() * rate).round()) + .max(0.0) as usize; + RenderedAudio { + data: vec![0.0; frames * channels as usize], + sample_rate: params.sample_rate.max(1), + channel_count: channels, + } +} + +#[cfg(test)] +mod tests { + use super::*; + use oak_core::{Rational, TimeRange}; + use oak_render::ticket::MontageClip; + + /// Stereo 48 kHz empty-montage params over `[in_sec, out_sec)`. + fn params(in_sec: i64, out_sec: i64) -> AudioTicketParams { + AudioTicketParams { + viewer: 7, + range: TimeRange::new(Rational::new(in_sec, 1), Rational::new(out_sec, 1)), + sample_rate: 48000, + channel_layout: 0x3, + montage: Vec::new(), + } + } + + /// Submitting returns immediately (no blocking render on the caller) + /// and the result arrives asynchronously on the completion channel. + #[test] + fn submit_returns_immediately_and_delivers_async() { + let (done_tx, done_rx) = mpsc::channel(); + let start = std::time::Instant::now(); + submit(params(0, 1), 12, done_tx).expect("submit"); + assert!( + start.elapsed() < std::time::Duration::from_millis(50), + "submit must not block" + ); + + let (start_ts, audio) = done_rx + .recv_timeout(std::time::Duration::from_secs(10)) + .expect("rendered audio delivered"); + assert_eq!(start_ts, 12); + assert_eq!(audio.sample_rate, 48000); + assert_eq!(audio.channel_count, 2); + // Empty montage → pure silence, anchored-grid length (1 s). + assert_eq!(audio.data.len(), 48000 * 2); + assert!(audio.data.iter().all(|&s| s == 0.0)); + } + + /// A clip that cannot be opened (missing file) degrades to silence of + /// the expected anchored-grid length — the playback buffer stays + /// aligned. + #[test] + fn error_produces_silence_of_expected_length() { + let (done_tx, done_rx) = mpsc::channel(); + let mut p = params(0, 1); + p.montage.push(MontageClip { + filename: "/nonexistent/oak-audio-test.wav".to_string(), + stream_index: 0, + in_time: Rational::new(0, 1), + out_time: Rational::new(1, 1), + media_in: Rational::new(0, 1), + gain: 1.0, + effects: Vec::new(), + }); + submit(p, 0, done_tx).expect("submit"); + + let (_, audio) = done_rx + .recv_timeout(std::time::Duration::from_secs(10)) + .expect("silence delivered"); + assert_eq!(audio.data.len(), 48000 * 2); + assert!(audio.data.iter().all(|&s| s == 0.0)); + } +} diff --git a/crates/oak-app/src/oakui/mod.rs b/crates/oak-app/src/oakui/mod.rs index 7f95c12fa..cd0bae8b5 100644 --- a/crates/oak-app/src/oakui/mod.rs +++ b/crates/oak-app/src/oakui/mod.rs @@ -40,6 +40,7 @@ //! * [`timecode`] — timecode / duration / fps / resolution formatting (pure, //! unit tested). +pub mod audio_thread; pub mod component; pub mod displaycolor; pub mod effectchain; diff --git a/crates/oak-app/src/oakui/real.rs b/crates/oak-app/src/oakui/real.rs index d0abcd1ea..ed5f493c6 100644 --- a/crates/oak-app/src/oakui/real.rs +++ b/crates/oak-app/src/oakui/real.rs @@ -241,7 +241,8 @@ enum FullResTarget { // workers, so by the time the playhead reaches a frame its pixels are // already sitting in a shm slot. `cpu_frame` hits this slot cache first // (zero copy — build the display image from the slot bytes, then release -// the slot); the synchronous render path is the miss fallback. Frames +// the slot); a miss falls back to the last displayed proxy frame (the +// synchronous render is the cold-cache fallback only). Frames // that fall out of the window (or whose params were invalidated) release // their slots back to the workers. @@ -268,24 +269,39 @@ struct PreviewWindow { } // --------------------------------------------------------------------------- -// Playback audio prefetch (M15 S3; M16 S2: rendered inline) +// Playback audio prefetch (M15 S3; M16 S3: rendered on the audio thread) // --------------------------------------------------------------------------- // // 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. +// buffered here. M16 S3 renders each chunk on the dedicated audio-render +// thread (see oakui::audio_thread) — never on the UI tick — and delivers it +// into the channel asynchronously, so the prefetch window serves as the +// scheduling cushion: 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; M16 S2: rendered inline on the UI tick, so the window is the +/// S3; M16 S3: rendered on the audio thread, 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; +/// 6 frames ahead ≈ 100 ms at 60 fps and ≈ 240 ms at 25 fps — the +/// output-device queue holds the whole window (chunks are pushed as soon +/// as rendered), so this is also how long a UI stall can last before the +/// device underruns. +const AUDIO_PREFETCH_CHUNKS: i64 = 6; + +/// The playhead frame whose audio is playing at audio-clock time `secs`: +/// `start_frame + (secs - start_secs)` seconds at `rate`. The wall clock +/// anchors play() at the click instant, but the first chunk only reaches +/// the device ~50-150 ms later, so the anchor is taken at the first push — +/// `start_secs` is the output clock at that moment, NOT zero (on a replay +/// the device has already padded underruns onto `output_frames_consumed`). +/// The multiply is split into `num as f64 / den as f64` first so large +/// frame counts cannot overflow i64. +fn audio_target_frame(start_frame: i64, start_secs: f64, secs: f64, rate: FrameRate) -> i64 { + start_frame + ((secs - start_secs) * rate.num as f64 / rate.den as f64).round() as i64 +} /// The playback-audio prefetch buffer: rendered chunks ordered by start /// timestamp, plus the submission cursor. UI-thread-only (guarded by the @@ -298,6 +314,12 @@ struct AudioPrefetch { next_submit: i64, /// Chunk length (sequence frames per tick) the buffer is built with. chunk: i64, + /// Sequence frame of the chunk most recently served to the output + /// device (None until the first pop). A playhead stalled between + /// sequence frames (a 60 Hz tick at a 25 fps rate) presents the same + /// frame again; [`AudioPrefetch::needs_reset`] distinguishes that from + /// a genuine seek so the frame is not re-pushed. + last_consumed: Option, /// Rendered chunks in start-ts order. buffered: VecDeque<(i64, super::renderops::RenderedAudio)>, } @@ -309,14 +331,25 @@ impl AudioPrefetch { front_ts: i64::MIN, next_submit: i64::MIN, chunk: 1, + last_consumed: None, buffered: VecDeque::new(), } } - /// True when `frame` lies inside the submitted window - /// `[front_ts, next_submit)` — the playhead is being served. - fn covers(&self, frame: i64) -> bool { - self.front_ts != i64::MIN && self.front_ts <= frame && frame < self.next_submit + /// True when the playhead jumped beyond what serving can follow — a + /// genuine seek or restart. Serving runs AHEAD of the playhead by up + /// to `AUDIO_PREFETCH_CHUNKS` chunks (rendered chunks are pushed to the + /// output device immediately so its queue rides out UI stalls), so the + /// served cursor normally leads the playhead; only a backward jump past + /// that lead (loop wrap, scrub) or a forward jump past the submitted + /// window means a real seek. The audio-master correction's small + /// re-anchors (a frame or two) stay inside the tolerance. + fn needs_reset(&self, frame: i64) -> bool { + if self.front_ts == i64::MIN { + return true; + } + let served = self.last_consumed.unwrap_or(self.front_ts - self.chunk); + frame < served - (AUDIO_PREFETCH_CHUNKS + 2) || frame >= self.next_submit } /// Reset for a (re)start at `frame`: drop every buffered chunk and @@ -325,6 +358,7 @@ impl AudioPrefetch { self.front_ts = frame; self.next_submit = frame; self.chunk = chunk.max(1); + self.last_consumed = None; self.buffered.clear(); } @@ -345,27 +379,26 @@ impl AudioPrefetch { self.buffered.insert(pos, (ts, data)); } - /// Pop the chunk at `frame` (dropping any stale leading chunks). - /// Returns `None` when the chunk has not been rendered yet. - fn pop_at(&mut self, frame: i64) -> Option { + /// Serve every rendered chunk in start-ts order, returning the ones to + /// push to the output device. Chunks whose sequence time is already + /// behind the playhead arrived too late — they are DROPPED rather than + /// played late, so an underrun/production gap costs content but never + /// accumulates A/V desync. On-time and future chunks are pushed + /// immediately (not one-per-tick): the output device's queue then + /// holds up to the whole prefetch window, so a stalled UI tick can no + /// longer starve the device into crackling. + fn drain_ready(&mut self, playhead: i64) -> Vec { + let mut out = Vec::new(); while let Some((ts, _)) = self.buffered.front() { - if *ts < frame { - let ts = self.buffered.pop_front().unwrap().0; - self.front_ts = ts + self.chunk; - } else { - break; + let ts = *ts; + let (_, data) = self.buffered.pop_front().unwrap(); + if ts >= playhead { + out.push(data); } - } - let (ts, data) = self.buffered.pop_front()?; - if ts == frame { self.front_ts = ts + self.chunk; - Some(data) - } else { - // Gap: the chunk at `frame` is still rendering. Re-insert and - // report nothing (the output device zero-fills this tick). - self.buffered.push_front((ts, data)); - None + self.last_consumed = Some(ts); } + out } } @@ -720,6 +753,19 @@ impl RealClock { self.started = None; } + /// Re-anchors a PLAYING clock at `frame`: subsequent wall-clock ticks + /// advance from here. A bare `transport.seek` while playing is + /// overwritten by the next [`tick`](Self::tick) (it recomputes the + /// playhead from the `started` anchor), so external clock followers + /// (the audio-master correction) must re-anchor instead. + pub fn reanchor_while_playing(&mut self, frame: Frame, length: Frame) { + if self.started.is_none() { + return; + } + self.transport.seek(frame, length); + self.started = Some((Instant::now(), self.transport.frame())); + } + /// Advances the playhead from the wall clock while playing, looping at /// `length` — or pausing on the last frame when `stop_on_last` is set. /// No-op when stopped. The advance is clamped per tick: a @@ -1117,18 +1163,31 @@ pub struct RealEngine { audio_tx: Mutex>, /// The audio prefetch buffer (see [`AudioPrefetch`]). audio_prefetch: Mutex, + /// Audio-mastered playback anchor `(playhead_frame, output_clock_secs)` + /// recorded at the first chunk push of a play run (M12 P1a). While set, + /// the tick loop re-anchors the program playhead onto the frame the + /// output device is actually consuming (`manager.seconds()`), fixing + /// the fixed startup offset and letting playback catch up after + /// underruns. Cleared on pause / seek / stop; `None` falls back to the + /// wall clock. + audio_playback: Option<(i64, f64)>, + /// Cooldown for the underrun self-heal resync in `pull_audio_tick` + /// (without it a resync retriggers every tick while the fresh chunks + /// are still rendering). + last_audio_resync: Option, } impl RealEngine { /// Render one tick's worth of audio at the program playhead and queue - /// 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. + /// it for playback (M12 P1; M15 S3: worker-pool prefetch; M16 S3: each + /// chunk is queued on the dedicated audio-render thread — see + /// [`renderops::submit_audio_chunk`](super::renderops::submit_audio_chunk) + /// — and delivered asynchronously into the channel, drained below, so + /// the render never runs on the UI tick). 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 audio thread 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; @@ -1151,9 +1210,9 @@ impl RealEngine { .audio_prefetch .lock() .unwrap_or_else(|e| e.into_inner()); - // Reset on seek / (re)start: the playhead must lie inside the - // submitted window [front_ts, next_submit). - if !st.covers(frame) { + // Reset on seek / restart (see needs_reset): serving runs ahead of + // the playhead, so only genuine jumps re-arm the buffer. + if st.needs_reset(frame) { st.reset(frame, chunk); } // Submit chunks to cover [next_submit, frame + PREFETCH ahead). In @@ -1173,35 +1232,104 @@ impl RealEngine { } st.next_submit += chunk; } - // 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). + // Drain completed chunks. With the M16 S3 audio thread the submit + // above only queues; the render lands here asynchronously, so the + // drain is the sole delivery path. 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); } drop(rx); - // Push the chunk at the current playhead to the output device. - let Some(buf) = st.pop_at(frame) else { - return; - }; - if buf.sample_rate <= 0 || buf.channel_count <= 0 || buf.data.is_empty() { - return; + // Underrun self-heal: zero-padding advanced the output clock past + // the queued content, so everything audible from here on would + // play late (the classic "audio drifts behind video"). Resync by + // dropping the device queue and re-rendering from the playhead; + // the anchor is re-recorded at the next push. Cooldown prevents a + // resync storm while the fresh chunks are still rendering. + if self.audio_playback.is_some() { + let cooldown_ok = self + .last_audio_resync + .map(|t| t.elapsed() > std::time::Duration::from_secs(1)) + .unwrap_or(true); + let underrun = oak_audio::manager::instance() + .map(|m| m.take_output_underrun_frames()) + .unwrap_or(0); + if cooldown_ok && underrun > 480 { + if let Some(mut manager) = oak_audio::manager::instance() { + let _ = manager.clear_buffered_output(); + } + st.reset(frame, chunk); + self.audio_playback = None; + self.last_audio_resync = Some(std::time::Instant::now()); + return; + } } - // Packed F32; layout: 1ch → mono mask, else stereo. - let layout: u64 = if buf.channel_count == 1 { 0x4 } else { 0x3 }; - let bytes: Vec = buf.data.iter().flat_map(|v| v.to_ne_bytes()).collect(); + // Push every rendered chunk to the output device in ts order (the + // device queue is the cushion that rides out UI stalls); chunks + // whose time already passed are dropped inside drain_ready. + for buf in st.drain_ready(frame) { + if buf.sample_rate <= 0 || buf.channel_count <= 0 || buf.data.is_empty() { + continue; + } + // Packed F32; layout: 1ch → mono mask, else stereo. + let layout: u64 = if buf.channel_count == 1 { 0x4 } else { 0x3 }; + let bytes: Vec = buf.data.iter().flat_map(|v| v.to_ne_bytes()).collect(); + if let Some(mut manager) = oak_audio::manager::instance() { + let _ = manager.push_to_output( + oak_audio::params::AudioParams { + sample_rate: buf.sample_rate, + channel_layout: layout, + format: oak_core::SampleFormat::F32, + }, + &bytes, + &mut [], + ); + // M12 P1a: anchor the audio master clock at the first push of a + // play run. Recording the CURRENT output clock (not 0) keeps the + // anchor honest on a replay — the device has been padding + // underruns onto `output_frames_consumed` since reset, so the + // correction starts from the frame actually audible right now + // instead of teleporting the playhead back to the push position. + drop(manager); + if self.audio_playback.is_none() { + let mut secs = -1.0; + if let Some(manager) = oak_audio::manager::instance() { + if manager.seconds(&mut secs).is_ok() && secs >= 0.0 { + self.audio_playback = Some((frame, secs)); + // Zeros padded before this first push (stream + // startup, or the whole paused interval) are + // not a playback underrun — reset the counter + // so the self-heal doesn't fire spuriously. + let _ = manager.take_output_underrun_frames(); + } + } + } + } + } + } + + /// Stop program playback: clear the play flag and silence the audio + /// output (drop the queued samples and re-anchor the output clock so + /// a later play() starts clean). Shared by [`AppEngine::pause`] and + /// the Stop-on-Last auto-stop — a clock that stopped itself leaves + /// `program_playing` set, and without this the last frame's audio + /// re-pushes forever. + fn stop_program_playback(&mut self, cx: &mut Context) { + self.program_playing = false; + self.audio_playback = None; if let Some(mut manager) = oak_audio::manager::instance() { - let _ = manager.push_to_output( - oak_audio::params::AudioParams { - sample_rate: buf.sample_rate, - channel_layout: layout, - format: oak_core::SampleFormat::F32, - }, - &bytes, - &mut [], - ); + let _ = manager.stop_output(); + let _ = manager.reset_output_clock(); } + // Reset the prefetch at the resting playhead so a later play() + // re-renders it: `last_consumed` would otherwise suppress the + // re-push of the very frame still on screen. + let frame = self.program_clock.read(cx).transport.frame().0; + let mut st = self + .audio_prefetch + .lock() + .unwrap_or_else(|e| e.into_inner()); + st.reset(frame, 1); } /// Builds an engine with no project open. @@ -1255,6 +1383,8 @@ impl RealEngine { audio_rx: Mutex::new(audio_rx), audio_tx: Mutex::new(audio_tx), audio_prefetch: Mutex::new(AudioPrefetch::new()), + audio_playback: None, + last_audio_resync: None, } } @@ -1750,8 +1880,8 @@ impl RealEngine { /// changed; slots that fell behind the playhead are released (credit /// returns to the workers). fn update_preview_window(&mut self, monitor: Monitor, cx: &mut Context) { - // Only during playback: a paused viewer uses the synchronous miss - // path (and the resting full-res fill). + // Only during playback: a paused viewer displays the last proxy + // frame (the resting full-res fill replaces it). let playing = self.clock(monitor).read(cx).transport.is_playing(); if !playing { return; @@ -3332,6 +3462,24 @@ impl EngineGateway for RealEngine { } cx.notify(); }); + if monitor == Monitor::Program { + // A seek must not keep playing audio queued for the old + // playhead: clear the output buffer and reset the prefetch so + // the new position renders from scratch (without the reset the + // dedup guard could suppress the re-push of the seek target). + // The audio master anchor is dropped too — the next play run + // re-anchors at its own first push. + self.audio_playback = None; + if let Some(mut manager) = oak_audio::manager::instance() { + let _ = manager.clear_buffered_output(); + let _ = manager.reset_output_clock(); + } + let mut st = self + .audio_prefetch + .lock() + .unwrap_or_else(|e| e.into_inner()); + st.reset(frame.0, 1); + } self.mirror_program_playhead(cx); self.update_ofx_viewer_time(cx); cx.notify(); @@ -3340,6 +3488,10 @@ impl EngineGateway for RealEngine { fn play(&mut self, monitor: Monitor, cx: &mut Context) { if monitor == Monitor::Program { self.program_playing = true; + // A new play run re-anchors the audio master clock at its first + // chunk push; a stale anchor (previous run, since cleared by + // pause/stop/seek) must not leak into the new one. + self.audio_playback = None; } let clock = self.clock(monitor).clone(); clock.update(cx, |clock, cx| { @@ -3351,14 +3503,14 @@ impl EngineGateway for RealEngine { } fn pause(&mut self, monitor: Monitor, cx: &mut Context) { - if monitor == Monitor::Program { - self.program_playing = false; - } let clock = self.clock(monitor).clone(); clock.update(cx, |clock, cx| { clock.pause(); cx.notify(); }); + if monitor == Monitor::Program { + self.stop_program_playback(cx); + } cx.notify(); } @@ -3394,6 +3546,56 @@ impl EngineGateway for RealEngine { cx.notify(); }); } + // M12 P1a: while the program plays with an audio anchor, re-anchor + // the playhead onto the frame the output device is actually + // consuming. The wall clock anchored play() at the click instant, + // but the first chunk reaches the device 50-150 ms later (tick + // interval + inline mix + cpal open) — a fixed startup offset every + // play run — and underruns pad the output with silence while still + // advancing `output_frames_consumed`, so playback also drifts. + // `manager.seconds()` is the truth: the playhead is set to the + // audible frame (loop playback wraps with the sequence). The strict + // `secs > start_secs` guard is the fail-safe: if the device never + // opens, seconds() stays 0/-1 and playback falls back to the wall + // clock untouched. + if self.program_playing && self.program_clock.read(cx).is_playing() { + if let Some((start_frame, start_secs)) = self.audio_playback { + let mut secs = -1.0; + if let Some(manager) = oak_audio::manager::instance() { + if manager.seconds(&mut secs).is_ok() && secs > start_secs { + let rate = self.program_clock.read(cx).frame_rate(); + let mut target = + audio_target_frame(start_frame, start_secs, secs, rate); + if length.0 > 0 && target >= length.0 { + target %= length.0; + } + let clock = self.program_clock.clone(); + clock.update(cx, |clock, cx| { + // Re-anchor (not transport.seek): the next + // wall-clock tick recomputes from this anchor, + // so the playhead keeps following the output + // device's consumed-sample clock instead of + // snapping back to the click-instant anchor. + clock.reanchor_while_playing(Frame(target), length); + cx.notify(); + }); + } else if secs < 0.0 { + // No running output stream (device closed): drop the + // anchor and stay on the wall clock. + self.audio_playback = None; + } + } else { + self.audio_playback = None; + } + } + } + // A Stop-on-Last clock pauses itself at the last frame; mirror + // that into the engine's play state like pause() (the transport is + // already stopped, so the output must go silent too — otherwise + // the last frame's audio re-pushes and the UI re-renders forever). + if self.program_playing && !self.program_clock.read(cx).is_playing() { + self.stop_program_playback(cx); + } self.mirror_program_playhead(cx); self.update_ofx_viewer_time(cx); self.meter_phase = self.meter_phase.wrapping_add(1); @@ -3619,8 +3821,8 @@ impl AppEngine for RealEngine { gpui_widgets::viewer::clear_gpu_frames(); } // The full-resolution fill replaces the proxy when its frame matches - // the playhead; otherwise the proxy frame is displayed (rendered - // synchronously below on a cache miss, filled by the background + // the playhead; otherwise the proxy frame is displayed (cached + // misses fall back to the last proxy below, filled by the background // worker once the playhead rests). if let Some(image) = cache.entry(monitor).or_default().image_for(frame.0) { return image.clone(); @@ -3636,21 +3838,23 @@ impl AppEngine for RealEngine { }); return image; } - // During playback a cache miss must NOT block the UI thread on a - // synchronous render: the wait starves the tick loop that feeds - // the pre-render window, and the seek-priority ticket steals - // worker capacity from it, so the window never catches up (every - // painted frame blocked in `TicketArena::wait` — the choppy - // playback regression). Show the last displayed frame while the - // window warms up; the workers catch up within a few frames and - // the window then serves every playhead frame. - if self.clock(monitor).read(cx).transport.is_playing() { - if let Some(image) = cache - .get(&monitor) - .and_then(|e| e.proxy.as_ref().map(|p| p.image.clone())) - { - return image; - } + // A cache miss must NOT block the UI thread on a synchronous render + // while the cache has ever displayed a frame: during playback the + // wait starves the tick loop that feeds the pre-render window, and + // the seek-priority ticket steals worker capacity from it, so the + // window never catches up (every painted frame blocked in + // `TicketArena::wait` — the choppy playback regression); after a + // pause or a paused seek the same wait stalls the UI for as long as + // the worker takes to render the resting playhead (the "freeze on + // pause" report). Show the last displayed frame instead — the + // background full-res fill (schedule_full_res on the tick) replaces + // it within a few frames. Only a fully cold cache (nothing ever + // displayed) falls through to the synchronous render below. + if let Some(image) = cache + .get(&monitor) + .and_then(|e| e.proxy.as_ref().map(|p| p.image.clone())) + { + return image; } // Both monitors render through the oakrender ticket arena (falling // back to the synthetic pattern when rendering is unavailable): the @@ -4990,6 +5194,15 @@ impl AppEngine for RealEngine { // The app-side transforms read the process global; the workers pick // the new settings up with the next graph upload. oak_render::color::set_pipeline_color_settings(working, spec); + // The workers derive their pipeline colors from the uploaded project + // snapshot; a settings change alone does not bump the undo-stack + // revision (the manager dedups re-uploads on it), so push an explicit + // resync — the workers re-deserialize the rewritten snapshot, adopt + // the new working space/output spec and drop their cached frames. + if let Some(m) = RenderManager::global() { + let revision = oak_undo::global::index().unwrap_or(0).max(0) as u64; + let _ = m.resync_graph_snapshot(&project, revision); + } // The pipeline colorspace changed: every cached frame (CPU image, // GPU texture, preview slot) was rendered under the old space. super::displaycolor::invalidate(); @@ -8483,9 +8696,8 @@ mod tests { #[test] fn audio_prefetch_orders_and_serves_chunks() { let mut st = AudioPrefetch::new(); - assert!(!st.covers(0), "uninitialized prefetch covers nothing"); + assert!(st.needs_reset(0), "uninitialized prefetch needs a reset"); st.reset(0, 10); - assert!(!st.covers(0), "nothing submitted yet"); st.next_submit = 40; // pretend 4 chunks were submitted (0..40) // Out-of-order arrivals (different workers) are reordered. @@ -8497,16 +8709,25 @@ mod tests { vec![0, 20, 30], "sorted by start ts" ); - // The chunk at the playhead is served. - let got = st.pop_at(0).expect("chunk 0 buffered"); - assert_eq!(got.sample_rate, 48000); - assert_eq!(st.buffered.len(), 2); - // A not-yet-rendered chunk reports nothing. - assert!(st.pop_at(10).is_none()); - // Stale arrivals (a seek raced the render) are dropped. + // drain_ready serves everything rendered, in ts order — on-time + // and future chunks alike (the device queue is the cushion). + let served = st.drain_ready(0); + assert_eq!(served.len(), 3); + assert_eq!(served[0].sample_rate, 48000); + assert_eq!(st.buffered.len(), 0); + assert_eq!(st.last_consumed, Some(30)); + // Stale arrivals (a seek raced the render) are dropped by insert. st.insert(-10, audio_chunk(-10).1); st.insert(100, audio_chunk(100).1); - assert_eq!(st.buffered.len(), 2, "stale chunks dropped"); + assert_eq!(st.buffered.len(), 0, "stale chunks dropped"); + // A chunk that arrives after its playback time passed is dropped, + // not played late (A/V sync beats completeness). + st.reset(0, 1); + st.next_submit = 10; + st.insert(3, audio_chunk(3).1); + let served = st.drain_ready(5); + assert!(served.is_empty(), "late chunk dropped"); + assert_eq!(st.last_consumed, Some(3)); } #[test] @@ -8515,20 +8736,14 @@ mod tests { st.reset(0, 10); st.next_submit = 40; st.insert(10, audio_chunk(10).1); - // A seek far ahead: the playhead is outside [front_ts, next_submit). - assert!(!st.covers(200)); + let _ = st.drain_ready(10); // served cursor now 20 + // A seek far ahead (past the submitted window) resets. + assert!(st.needs_reset(200)); + // A backward seek past the serve-ahead lead resets too. + assert!(st.needs_reset(0), "playhead jumped back behind the served cursor"); st.reset(200, 10); assert_eq!(st.buffered.len(), 0, "old chunks dropped"); - assert!(st.pop_at(200).is_none()); - // A backward seek (the playhead behind the buffer front after it - // advanced) is a reset too. - st.reset(0, 10); - st.next_submit = 40; - st.insert(10, audio_chunk(10).1); - let _ = st.pop_at(10); // front_ts now 20 - assert!(!st.covers(5), "chunk 5 is behind the front"); - st.reset(5, 10); - assert_eq!(st.buffered.len(), 0); + assert!(st.drain_ready(200).is_empty()); } #[test] @@ -8541,6 +8756,119 @@ mod tests { assert_eq!(st.buffered.len(), 1, "duplicates dropped"); } + #[test] + fn audio_prefetch_does_not_repeat_a_stalled_playhead() { + let mut st = AudioPrefetch::new(); + st.reset(0, 1); + st.next_submit = 7; // pretend chunks 0..7 were submitted + st.insert(0, audio_chunk(0).1); + st.insert(1, audio_chunk(1).1); + // Serving runs ahead: both rendered chunks are pushed at once. + assert_eq!(st.drain_ready(0).len(), 2); + assert_eq!(st.last_consumed, Some(1)); + // The next ticks (60 Hz vs 25 fps) still show frame 0: a stalled + // playhead is neither a reset nor a re-push. + assert!(!st.needs_reset(0), "a stalled playhead is not a reset"); + assert!(st.drain_ready(0).is_empty(), "nothing new to serve"); + // Small audio-master corrections stay inside the tolerance. + assert!(!st.needs_reset(-2)); + // A genuine seek (far ahead, or a loop wrap back to 0 after the + // served cursor advanced) still resets. + assert!(st.needs_reset(200), "a seek is a reset"); + st.reset(200, 1); + assert_eq!(st.last_consumed, None, "a reset forgets the old playhead"); + // After serving the new frame the guard re-arms. + st.next_submit = 207; + st.insert(200, audio_chunk(200).1); + assert_eq!(st.drain_ready(200).len(), 1); + assert!(!st.needs_reset(200), "a stall at 200 is not a reset either"); + } + + // ---- M12 P1a audio-mastered playhead ---------------------------------- + + /// The audio-clock → frame conversion: an anchor recorded late (the + /// first chunk reaches the device ~50-150 ms after play()) must land + /// the playhead on the frame actually audible, and drift/underruns must + /// let it catch up. + #[test] + fn audio_target_frame_anchors_and_converts() { + let rate = FrameRate::new(25, 1); + // Startup offset: the anchor was taken 0.1 s after play(); by + // audio-clock 1.1 s exactly 1.0 s has played → 25 frames past the + // anchor frame. + assert_eq!(audio_target_frame(0, 0.1, 1.1, rate), 25); + // Catch-up: the device already consumed audio before the anchor. + assert_eq!(audio_target_frame(100, 2.0, 4.0, rate), 150); + // Fractional seconds round to the nearest frame. + assert_eq!(audio_target_frame(0, 0.0, 0.02, rate), 1, "0.5 s rounds up"); + assert_eq!(audio_target_frame(0, 0.0, 0.018, rate), 0, "0.45 s rounds down"); + // NTSC 29.97: num/den stays a rational, not rounded to 30. + let ntsc = FrameRate::NTSC_2997; + assert_eq!(audio_target_frame(0, 0.0, 1.0, ntsc), 30); + assert_eq!(audio_target_frame(0, 0.0, 10.0, ntsc), 300); + // Large frame counts must not overflow (num/den split before the + // multiply). + assert_eq!(audio_target_frame(1 << 40, 0.0, 1.0, rate), (1 << 40) + 25); + } + + /// The audio-master anchor must not survive without a live output + /// stream: the tick drops it and playback falls back to the wall clock + /// (the fail-safe that keeps playback alive when the device is closed + /// or headless). + #[gpui::test] + async fn audio_master_anchor_falls_back_to_wall_clock(cx: &mut gpui::TestAppContext) { + let engine = cx.update(|cx| cx.new(|cx| RealEngine::create(cx))); + cx.update(|app| { + engine.update(app, |engine, cx| { + // A stale anchor (as if the first chunk was pushed). + engine.audio_playback = Some((0, 0.0)); + engine.play(Monitor::Program, cx); + engine.tick(cx); + assert!( + engine.audio_playback.is_none(), + "no output stream → anchor dropped, wall clock drives" + ); + }); + }); + } + + /// A paused playhead miss (seek while paused, worker still rendering) + /// must serve the last displayed frame instead of blocking the UI on a + /// synchronous render — the "freeze on pause" regression. The engine + /// has no project, so any render path would fail; returning the cached + /// proxy proves the non-blocking fallback. + #[gpui::test] + async fn paused_playhead_miss_serves_last_displayed_frame(cx: &mut gpui::TestAppContext) { + let engine = cx.update(|cx| cx.new(|cx| RealEngine::create(cx))); + let (width, height, samples) = synthetic_frame_samples(Frame(0)); + let scope = analyze_f32_rgba(width, height, &samples); + let image = Arc::new(f32_rgba_to_bgra_image(width, height, &samples)); + cx.update(|app| { + engine.update(app, |engine, cx| { + // The cache already displayed a frame (the proxy). + engine + .cpu_frame_cache + .lock() + .unwrap() + .entry(Monitor::Program) + .or_default() + .proxy = Some(ProxyEntry { + frame: 0, + image: image.clone(), + scope, + }); + // Seek the (paused) playhead to a frame that is NOT cached: + // the display must fall back to the last displayed frame. + engine.request_frame(Monitor::Program, Frame(24), cx); + }); + }); + let got = cx.read(|app| engine.read(app).cpu_frame(Monitor::Program, app)); + assert!( + Arc::ptr_eq(&image, &got), + "a paused miss serves the last displayed frame" + ); + } + // ---- sequence management (engine facade) ------------------------------ /// The engine's create-folder / create-sequence-with-params commands diff --git a/crates/oak-app/src/oakui/renderops.rs b/crates/oak-app/src/oakui/renderops.rs index 98db74d94..e8de19d98 100644 --- a/crates/oak-app/src/oakui/renderops.rs +++ b/crates/oak-app/src/oakui/renderops.rs @@ -897,12 +897,13 @@ pub fn render_audio_range( } /// 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). +/// S3: the chunk is queued on the dedicated audio-render thread — see +/// [`audio_thread`](super::audio_thread) — so the render never runs on the +/// UI tick; a slow decode no longer freezes the UI or delays the next +/// chunk (the old inline-dispatch path underran the audio device). 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( p: &ProjectRef, seq: NodeId, @@ -918,12 +919,7 @@ pub fn submit_audio_chunk( let montage = audio_montage(p, seq, range); let sample_rate = 48000; 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())?; - // 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( + super::audio_thread::submit( AudioTicketParams { viewer: seq.identity(), range, @@ -931,40 +927,9 @@ pub fn submit_audio_chunk( channel_layout, montage, }, - Box::new(move |result| { - let data = match result { - Ok(TicketPayload::Audio(samples)) => RenderedAudio { - data: samples.samples, - sample_rate: samples.sample_rate, - channel_count: samples.channel_count, - }, - Ok(TicketPayload::ShmAudio(audio)) => { - let samples = audio.to_audio_samples(); - if let Some(m) = RenderManager::global() { - m.release_audio_frame(&audio); - } - RenderedAudio { - data: samples.samples, - sample_rate: samples.sample_rate, - channel_count: samples.channel_count, - } - } - _ => { - // Silence of the expected length keeps the output buffer - // aligned (an underrun here just plays zeros). - let seconds = len_ts as f64 * tb.0 as f64 / tb.1 as f64; - let frames = (seconds * sample_rate as f64).round() as usize; - RenderedAudio { - data: vec![0.0; frames * channels as usize], - sample_rate, - channel_count: channels, - } - } - }; - let _ = tx.send((start_ts, data)); - }), - ); - Ok(()) + start_ts, + tx, + ) } // --------------------------------------------------------------------------- diff --git a/crates/oak-audio/src/manager.rs b/crates/oak-audio/src/manager.rs index b46ab3e57..a3252da46 100644 --- a/crates/oak-audio/src/manager.rs +++ b/crates/oak-audio/src/manager.rs @@ -193,6 +193,15 @@ impl ManagerInner { Ok(()) } + /// Read and reset the output underrun counter (frames the callback had + /// to zero-fill because the buffer was empty). The playback engine + /// polls this per tick and resyncs after an underrun: the zero-fill + /// advanced the playback clock past the queued content, so without a + /// resync everything afterwards plays late. + pub fn take_output_underrun_frames(&self) -> i64 { + self.output_buffer.take_underrun_frames() + } + /// Stop the output stream. /// /// `// CPP-PARITY: src/audio/src/audiomanager.cpp:229` (`stop_output` aborts diff --git a/crates/oak-audio/src/outputdevice.rs b/crates/oak-audio/src/outputdevice.rs index 94d4fe549..7250e8cff 100644 --- a/crates/oak-audio/src/outputdevice.rs +++ b/crates/oak-audio/src/outputdevice.rs @@ -131,6 +131,9 @@ impl PortAudioOutput { } } sink_cb.add_output_frames((total / channels_usize) as i64); + // Account the zero-filled tail separately so the engine can + // resync after an underrun instead of drifting out of sync. + sink_cb.add_underrun_frames((total / channels_usize - frames_got) as i64); }; let err_callback = |err: cpal::Error| { eprintln!("output stream error: {err}"); diff --git a/crates/oak-audio/src/previewdevice.rs b/crates/oak-audio/src/previewdevice.rs index 901dbf4d9..faa85bbac 100644 --- a/crates/oak-audio/src/previewdevice.rs +++ b/crates/oak-audio/src/previewdevice.rs @@ -35,6 +35,12 @@ pub struct PreviewAudioDevice { /// Frames consumed by the output callback (playback clock, includes /// underrun zero-fill). output_frames_consumed: AtomicI64, + /// Frames the output callback had to zero-fill (buffer was empty). + /// Kept separate from the playback clock so the engine can detect + /// underruns and resync — zero-fill advances the playback clock past + /// the queued content, so everything queued afterwards would play + /// late (the classic audio-behind-video drift). + underrun_frames: AtomicI64, } struct PreviewAudioDeviceInner { @@ -62,6 +68,7 @@ impl PreviewAudioDevice { notify_callback: None, }), output_frames_consumed: AtomicI64::new(0), + underrun_frames: AtomicI64::new(0), } } @@ -176,6 +183,19 @@ impl PreviewAudioDevice { .fetch_add(frame_count, Ordering::Relaxed); } + /// Account for frames the output callback zero-filled (underrun). + pub fn add_underrun_frames(&self, frame_count: i64) { + if frame_count > 0 { + self.underrun_frames + .fetch_add(frame_count, Ordering::Relaxed); + } + } + + /// Read and reset the underrun counter (the engine polls it per tick). + pub fn take_underrun_frames(&self) -> i64 { + self.underrun_frames.swap(0, Ordering::Relaxed) + } + /// Frames consumed by the output callback. pub fn output_frames_consumed(&self) -> i64 { self.output_frames_consumed.load(Ordering::Relaxed) diff --git a/crates/oak-codec/src/ffmpeg.rs b/crates/oak-codec/src/ffmpeg.rs index d043410aa..d99c18f95 100644 --- a/crates/oak-codec/src/ffmpeg.rs +++ b/crates/oak-codec/src/ffmpeg.rs @@ -250,6 +250,15 @@ impl FFmpegDecoder { .and_then(|s| s.hw_device.map(crate::hwdecode::device_type_name)) .map(|name| name.to_string()) } + + /// Test-only: how many `seek()` calls the open session has performed. + /// Contiguous audio chunks must not increase this (they continue the + /// decode without re-seeking); a non-contiguous chunk must. + #[cfg(test)] + pub(crate) fn audio_seek_count(&self) -> u64 { + let state = self.state.lock().unwrap_or_else(|e| e.into_inner()); + state.as_ref().map(|s| s.audio_seeks).unwrap_or(0) + } } impl Default for FFmpegDecoder { @@ -481,6 +490,10 @@ struct DecoderState { eof: bool, video: Option, audio: Option, + /// Test-only counter of `seek()` calls (proves contiguous audio chunks + /// skip the seek). + #[cfg(test)] + audio_seeks: u64, } enum DecoderInner { @@ -523,6 +536,14 @@ struct ScalingCache { /// Audio decode state: a resampler keyed by (rate, layout). struct AudioDecodeState { resampler: Option<(u32, u64, AudioResampler)>, + /// The first sample of the next contiguous chunk: the sample index + /// (at the current `sample_rate`) one past the last decoded chunk. + /// Set on every successful `retrieve_audio_to`; cleared on every seek + /// (any seek resets the decoder and resampler, breaking continuity). + /// When the next chunk's start sample equals this, the decode can + /// continue without re-seeking — skipping the per-chunk decoder flush + /// and resampler reset that cause boundary artifacts (pops/clicks). + contiguous_end_sample: Option, } /// A swresample conversion context. @@ -641,7 +662,10 @@ impl DecoderState { None }; let audio = if matches!(medium, MediaType::Audio) { - Some(AudioDecodeState { resampler: None }) + Some(AudioDecodeState { + resampler: None, + contiguous_end_sample: None, + }) } else { None }; @@ -661,6 +685,8 @@ impl DecoderState { eof: false, video, audio, + #[cfg(test)] + audio_seeks: 0, }) } @@ -784,7 +810,16 @@ impl DecoderState { DecoderInner::Video(d) => d.flush(), DecoderInner::Audio(d) => d.flush(), } + #[cfg(test)] + { + self.audio_seeks += 1; + } self.eof = false; + // Any seek breaks audio continuity (decoder flush + resampler + // reset): the next audio chunk must not skip its own seek. + if let Some(a) = &mut self.audio { + a.contiguous_end_sample = None; + } Ok(()) } @@ -1137,88 +1172,71 @@ impl DecoderState { dest.fill(0.0); - // Seek to just before the range start. - let start_ts = - oak_rational(self.stream_time_base).time_to_timestamp(Rational::from_double(start_sec)); - self.seek(start_ts)?; - - // Take the cached resampler out (or create one) so the decode loop - // below can borrow `self` freely; it is put back before returning. - let src_layout = channel_layout_from_mask(self.input_channel_layout_mask); - let mut resampler = match self.audio.as_mut().expect("audio session").resampler.take() { - Some((rate, layout, rs)) if rate == sample_rate as u32 && layout == channel_layout => { - rs - } - _ => AudioResampler::get( - self.input_sample_format, - src_layout, - self.input_sample_rate, - Sample::F32(SampleType::Packed), - dst_layout, - sample_rate as u32, - )?, - }; - let stream_time_base = self.stream_time_base; - - let mut next_sample: Option = None; - let dest_frames = (dest.len() / dst_channels) as i64; - - while let Some(frame) = self.next_frame()? { - let DecodedFrame::Audio(audio) = frame else { - continue; - }; - let converted = resample_to_interleaved_f32(&mut resampler, &audio)?; - if converted.is_empty() { - continue; - } - let chunk_samples = (converted.len() / dst_channels) as i64; - let frame_start = match audio.pts() { - Some(pts) => { - let secs = oak_rational(stream_time_base) - .timestamp_to_time(pts) - .to_f64(); - (secs * sample_rate as f64).round() as i64 - } - None => next_sample.unwrap_or(start_sample), - }; - let offset = frame_start - start_sample; - if offset < dest_frames && offset + chunk_samples > 0 { - let copy_start = offset.max(0) as usize; - let copy_end = (offset + chunk_samples).min(dest_frames).max(0) as usize; - if copy_end > copy_start { - let src_off = (copy_start as i64 - offset) as usize * dst_channels; - let n = (copy_end - copy_start) * dst_channels; - let dst_off = copy_start * dst_channels; - if dst_off + n <= dest.len() { - dest[dst_off..dst_off + n] - .copy_from_slice(&converted[src_off..src_off + n]); - } - } - } - next_sample = Some(frame_start + chunk_samples); - if offset + chunk_samples >= dest_frames { - break; - } + // Continuity fast path: when this chunk's start sample equals the + // sample one past the previous chunk's end, the decoder and + // resampler are still positioned exactly there — continue without + // seeking. Skipping the per-chunk seek avoids the decoder flush and + // resampler reset whose boundary artifacts (silence gaps, phase + // resets) were audible as pops/clicks at chunk edges. The caller's + // anchored sample grid (`round(out·rate) − round(in·rate)`) makes + // the next chunk's start exactly the previous chunk's recorded end, + // so exact comparison is correct. + let contiguous = self + .audio + .as_ref() + .map(|a| a.contiguous_end_sample == Some(start_sample)) + .unwrap_or(false); + if !contiguous { + // Seek to just before the range start. + let start_ts = + oak_rational(self.stream_time_base).time_to_timestamp(Rational::from_double(start_sec)); + self.seek(start_ts)?; } - // Put the resampler back into the cache for the next call. - self.audio.as_mut().expect("audio session").resampler = - Some((sample_rate as u32, channel_layout, resampler)); + // The fill+decode+flush body runs in a closure so the success path + // can record the chunk's end sample (continuity) and an error path + // can clear it (the decoder/resampler position is unknown after a + // mid-chunk failure). + let read = (|| -> crate::error::Result { + // Take the cached resampler out (or create one) so the decode loop + // below can borrow `self` freely; it is put back before returning. + let src_layout = channel_layout_from_mask(self.input_channel_layout_mask); + let mut resampler = match self.audio.as_mut().expect("audio session").resampler.take() { + Some((rate, layout, rs)) if rate == sample_rate as u32 && layout == channel_layout => { + rs + } + _ => AudioResampler::get( + self.input_sample_format, + src_layout, + self.input_sample_rate, + Sample::F32(SampleType::Packed), + dst_layout, + sample_rate as u32, + )?, + }; + let stream_time_base = self.stream_time_base; - // Flush any samples still buffered in the resampler (rate conversion - // tail), appending after the last decoded sample. - let resampler = self - .audio - .as_mut() - .expect("audio session") - .resampler - .as_mut() - .map(|r| &mut r.2) - .unwrap(); - if let Some(flush) = flush_resampler_interleaved_f32(resampler)? { - if !flush.is_empty() { - let chunk_samples = (flush.len() / dst_channels) as i64; - let frame_start = next_sample.unwrap_or(start_sample); + let mut next_sample: Option = None; + let dest_frames = (dest.len() / dst_channels) as i64; + + while let Some(frame) = self.next_frame()? { + let DecodedFrame::Audio(audio) = frame else { + continue; + }; + let converted = resample_to_interleaved_f32(&mut resampler, &audio)?; + if converted.is_empty() { + continue; + } + let chunk_samples = (converted.len() / dst_channels) as i64; + let frame_start = match audio.pts() { + Some(pts) => { + let secs = oak_rational(stream_time_base) + .timestamp_to_time(pts) + .to_f64(); + (secs * sample_rate as f64).round() as i64 + } + None => next_sample.unwrap_or(start_sample), + }; let offset = frame_start - start_sample; if offset < dest_frames && offset + chunk_samples > 0 { let copy_start = offset.max(0) as usize; @@ -1229,14 +1247,70 @@ impl DecoderState { let dst_off = copy_start * dst_channels; if dst_off + n <= dest.len() { dest[dst_off..dst_off + n] - .copy_from_slice(&flush[src_off..src_off + n]); + .copy_from_slice(&converted[src_off..src_off + n]); + } + } + } + next_sample = Some(frame_start + chunk_samples); + if offset + chunk_samples >= dest_frames { + break; + } + } + + // Put the resampler back into the cache for the next call. + self.audio.as_mut().expect("audio session").resampler = + Some((sample_rate as u32, channel_layout, resampler)); + + // Flush any samples still buffered in the resampler (rate conversion + // tail), appending after the last decoded sample. + let resampler = self + .audio + .as_mut() + .expect("audio session") + .resampler + .as_mut() + .map(|r| &mut r.2) + .unwrap(); + if let Some(flush) = flush_resampler_interleaved_f32(resampler)? { + if !flush.is_empty() { + let chunk_samples = (flush.len() / dst_channels) as i64; + let frame_start = next_sample.unwrap_or(start_sample); + let offset = frame_start - start_sample; + if offset < dest_frames && offset + chunk_samples > 0 { + let copy_start = offset.max(0) as usize; + let copy_end = (offset + chunk_samples).min(dest_frames).max(0) as usize; + if copy_end > copy_start { + let src_off = (copy_start as i64 - offset) as usize * dst_channels; + let n = (copy_end - copy_start) * dst_channels; + let dst_off = copy_start * dst_channels; + if dst_off + n <= dest.len() { + dest[dst_off..dst_off + n] + .copy_from_slice(&flush[src_off..src_off + n]); + } } } } } - } - Ok(RetrieveAudioStatus::Success) + // Record the end sample: the next chunk starting exactly here + // continues without a seek. + self.audio.as_mut().expect("audio session").contiguous_end_sample = + Some(start_sample + dest_frames); + + Ok(RetrieveAudioStatus::Success) + })(); + + match read { + Ok(status) => Ok(status), + Err(e) => { + // A failed chunk leaves the decoder/resampler position + // unknown: force a fresh seek on the next chunk. + if let Some(a) = &mut self.audio { + a.contiguous_end_sample = None; + } + Err(e) + } + } } /// Conform the open stream's audio into per-channel PCM files, mirroring diff --git a/crates/oak-codec/src/hwdecode.rs b/crates/oak-codec/src/hwdecode.rs index 5603ca5cf..ae3271189 100644 --- a/crates/oak-codec/src/hwdecode.rs +++ b/crates/oak-codec/src/hwdecode.rs @@ -57,17 +57,61 @@ pub static HW_TRANSFERS: std::sync::atomic::AtomicU64 = std::sync::atomic::Atomi /// 0 = force software). Default ON by user mandate. pub const CONFIG_KEY_HARDWARE_DECODING: &str = "HardwareDecoding"; +/// Process-wide negative cache of failed device types (a `static`, not a +/// `const` — a const array is inlined per use site, so writes from +/// `mark_device_unavailable` would never be visible to `device_unavailable`). +/// Creating a hardware device context (`av_hwdevice_ctx_create`) is +/// expensive when the driver is missing and emits FFmpeg's +/// `[VAAPI @ ...] Failed to initialise VAAPI connection` spam every time — +/// on a box without libva that is once per decoder open, per process, with +/// the full log line. The cache marks a device type unavailable after its +/// first failed creation, so every later open skips it (and its log noise) +/// entirely. The slot count is generous: `AVHWDeviceType` values are small +/// non-negative enumerants (< 32), 64 covers them all. +static UNAVAILABLE_DEVICES: [std::sync::atomic::AtomicBool; 64] = + [const { std::sync::atomic::AtomicBool::new(false) }; 64]; + +/// Test-only counter of `open_hw_accel` device-context creation attempts +/// (incremented at the top of the function, before any FFmpeg call). Lets +/// tests prove the negative cache short-circuits before FFmpeg is +/// involved. +#[cfg(test)] +static CREATE_ATTEMPTS: std::sync::atomic::AtomicU64 = std::sync::atomic::AtomicU64::new(0); + +/// Whether `device_type` is known unavailable — a device-context +/// creation failed once earlier in this process. +pub fn device_unavailable(device_type: sys::AVHWDeviceType) -> bool { + UNAVAILABLE_DEVICES + .get(device_type as usize) + .map(|f| f.load(std::sync::atomic::Ordering::Relaxed)) + .unwrap_or(false) +} + +/// Record that `device_type`'s device context failed to create (sticky +/// for the process; the caller falls back to the next candidate). +pub fn mark_device_unavailable(device_type: sys::AVHWDeviceType) { + if let Some(f) = UNAVAILABLE_DEVICES.get(device_type as usize) { + f.store(true, std::sync::atomic::Ordering::Relaxed); + } +} + /// Whether hardware decoding is preferred (the config switch). Default /// ON by user mandate; only an explicit `"false"` turns it off (the /// string accessor, same convention as the app's config helpers — the -/// store's typed `get_bool` only parses pre-typed Bool entries). +/// store's typed `get_bool` only parses pre-typed Bool entries). The +/// `OAK_HWACCEL` environment variable overrides both: `0` force-disables +/// hardware decode entirely (a diagnostic escape hatch on machines where +/// probing the device is slow or noisy). pub fn hardware_decoding_enabled() -> bool { - match oak_common::configstore::ConfigStore::instance() - .get(None, CONFIG_KEY_HARDWARE_DECODING) - { - Ok(value) => value != "false", - Err(_) => true, - } + if let Ok(v) = std::env::var("OAK_HWACCEL") { + return v != "0"; + } + match oak_common::configstore::ConfigStore::instance() + .get(None, CONFIG_KEY_HARDWARE_DECODING) + { + Ok(value) => value != "false", + Err(_) => true, + } } /// The hardware device types to try, most preferred first. On a machine @@ -116,6 +160,18 @@ pub fn open_hw_accel( codec: ffmpeg::Codec, device_type: sys::AVHWDeviceType, ) -> Option<(ffmpeg::codec::decoder::Opened, sys::AVHWDeviceType)> { + // Negative cache: a device type whose context creation failed once + // (no driver, headless box) is never tried again — creation is slow + // and logs `[VAAPI @ ...] Failed to initialise VAAPI connection` per + // attempt. Checked before any FFmpeg call so marked types cost + // nothing. + if device_unavailable(device_type) { + return None; + } + #[cfg(test)] + { + CREATE_ATTEMPTS.fetch_add(1, std::sync::atomic::Ordering::Relaxed); + } let mut context = ffmpeg::codec::Context::from_parameters(params.clone()).ok()?; let mut device: *mut sys::AVBufferRef = std::ptr::null_mut(); // SAFETY: `device` is a valid out-pointer; on success it owns the @@ -130,6 +186,9 @@ pub fn open_hw_accel( ) }; if rc < 0 || device.is_null() { + // The device is unavailable (missing driver / no hardware): mark + // it so later opens skip the attempt and its log noise. + mark_device_unavailable(device_type); return None; } // SAFETY: `hw_device_ctx` takes ownership of the reference; the codec @@ -241,4 +300,47 @@ mod tests { ); assert!(opened.is_some(), "VideoToolbox hwaccel must open for H.264"); } + + /// A device type marked unavailable is short-circuited before any + /// FFmpeg call: `open_hw_accel` returns None without attempting device + /// creation (which is what used to log + /// `[VAAPI @ ...] Failed to initialise VAAPI connection` every time on + /// boxes without the driver). + #[test] + fn negative_cache_skips_marked_device() { + // VDPAU is not a candidate on any supported platform, so marking + // it cannot disturb the platform tests in this process (e.g. the + // macOS VideoToolbox test above). + let dev = sys::AVHWDeviceType::AV_HWDEVICE_TYPE_VDPAU; + mark_device_unavailable(dev); + assert!(device_unavailable(dev)); + + let input = ffmpeg::format::input(&"../oak-app/tests/demo.mp4".to_string()) + .expect("open demo.mp4"); + let fstream = input.stream(0).expect("stream 0"); + let params = fstream.parameters(); + let codec = ffmpeg::decoder::find(params.id()).expect("software h264 codec"); + + CREATE_ATTEMPTS.store(0, std::sync::atomic::Ordering::Relaxed); + let opened = open_hw_accel(¶ms, codec, dev); + assert!(opened.is_none(), "marked device must not open"); + assert_eq!( + CREATE_ATTEMPTS.load(std::sync::atomic::Ordering::Relaxed), + 0, + "marked device must be skipped before any creation attempt" + ); + + // Restore so the (non-candidate) type stays clean for other tests. + UNAVAILABLE_DEVICES[dev as usize].store(false, std::sync::atomic::Ordering::Relaxed); + } + + /// A device type never tried before is not cached as unavailable + /// (only a failed creation marks it). DXVA2 is not a candidate on any + /// supported platform (Windows uses D3D11VA/CUDA), so it is never + /// touched by other tests in this process. + #[test] + fn unmarked_device_type_is_not_unavailable() { + let dev = sys::AVHWDeviceType::AV_HWDEVICE_TYPE_DXVA2; + assert!(!device_unavailable(dev)); + } } diff --git a/crates/oak-codec/src/realmedia_tests.rs b/crates/oak-codec/src/realmedia_tests.rs index 3308a9df8..e9c8e76a0 100644 --- a/crates/oak-codec/src/realmedia_tests.rs +++ b/crates/oak-codec/src/realmedia_tests.rs @@ -179,6 +179,85 @@ fn audio_decode_is_non_empty() { assert!(peak > 0.0, "decoded audio is all silence"); } +/// Contiguous audio chunks (each chunk's start is exactly where the +/// previous chunk ended) must continue the decode without re-seeking: +/// the seek count stays flat, and the concatenated chunks must match a +/// one-shot decode of the same span. A non-contiguous chunk re-seeks. +/// This is the regression test for the boundary pops/clicks caused by the +/// per-chunk `av_seek_frame` + decoder flush + resampler reset. +#[test] +fn contiguous_audio_chunks_skip_seek_and_match_oneshot() { + let d = FFmpegDecoder::new(); + let s = CodecStream::with_block(demo_path().to_string_lossy().into_owned(), 1, None); + d.open(&s).expect("open audio stream"); + + // Chunk [0s, 1s): opens the session (one seek inside retrieve). + let mut c1 = vec![0f32; 48000 * 2]; + d.retrieve_audio( + &mut c1, + &TimeRange::new(Rational::new(0, 1), Rational::new(1, 1)), + 48000, + 0x3, + ) + .expect("chunk [0,1)"); + assert!(c1.iter().any(|&v| v != 0.0), "chunk [0,1) is all silence"); + let seeks_after_first = d.audio_seek_count(); + + // Chunk [1s, 2s): contiguous with the previous one — must not seek. + let mut c2 = vec![0f32; 48000 * 2]; + d.retrieve_audio( + &mut c2, + &TimeRange::new(Rational::new(1, 1), Rational::new(2, 1)), + 48000, + 0x3, + ) + .expect("chunk [1,2)"); + assert!( + d.audio_seek_count() == seeks_after_first, + "contiguous chunk must skip the seek ({} -> {})", + seeks_after_first, + d.audio_seek_count() + ); + assert!(c2.iter().any(|&v| v != 0.0), "chunk [1,2) is all silence"); + + // Chunk [3s, 4s): not contiguous — must seek again. + let mut c4 = vec![0f32; 48000 * 2]; + d.retrieve_audio( + &mut c4, + &TimeRange::new(Rational::new(3, 1), Rational::new(4, 1)), + 48000, + 0x3, + ) + .expect("chunk [3,4)"); + assert!( + d.audio_seek_count() > seeks_after_first, + "non-contiguous chunk must seek" + ); + + // Concatenating [0,1)+[1,2) must equal a one-shot decode of [0,2): + // both read the same decoder continuously from sample 0, so the + // samples must match exactly (the second chunk only differs in that it + // skipped its seek — no flush/resampler reset was involved). + let mut combined = c1; + combined.extend_from_slice(&c2); + let mut oneshot = vec![0f32; 48000 * 2 * 2]; + d.retrieve_audio( + &mut oneshot, + &TimeRange::new(Rational::new(0, 1), Rational::new(2, 1)), + 48000, + 0x3, + ) + .expect("oneshot [0,2)"); + let mut max_diff = 0.0f32; + for (a, b) in combined.iter().zip(oneshot.iter()) { + max_diff = max_diff.max((a - b).abs()); + } + assert!( + max_diff < 0.01, + "contiguous chunks diverge from one-shot decode (max diff {max_diff})" + ); +} + #[test] fn encode_h264_roundtrip_to_tmp() { let out = std::env::temp_dir().join(format!("oakcodec_roundtrip_{}.mp4", std::process::id())); diff --git a/crates/oak-common/src/colormath.rs b/crates/oak-common/src/colormath.rs index b6a6871ec..05f647d50 100644 --- a/crates/oak-common/src/colormath.rs +++ b/crates/oak-common/src/colormath.rs @@ -31,6 +31,15 @@ //! chromatic adaptation between differing white points (D65 sources → the //! ACES D60-ish white of AP1). Unit tests pin the results against the //! published ACES transform values. +//! +//! ## Performance +//! +//! The per-pixel transforms run over full frames on hot paths (decode, +//! the app-side output node). libm `powf` costs ~15 ns per channel — +//! ~375 ms per 1080p frame for a matrix+OETF pass — so the transfer +//! functions are evaluated through 4096-entry LUTs with linear +//! interpolation (max error ≈ 2e-5, far below 10-bit quantization) and +//! large frames are split across scoped threads by row band. /// A 3x3 row-major matrix. pub type Mat3 = [[f32; 3]; 3]; @@ -476,6 +485,138 @@ pub fn hlg_oetf(v: f32) -> f32 { } } +// --------------------------------------------------------------------------- +// Fast transfer evaluation (LUT) and parallel per-pixel transforms +// --------------------------------------------------------------------------- + +/// LUT intervals over the [0, 1] domain (plus the 1.0 endpoint). +const LUT_N: usize = 4096; + +/// Declare one lazily-built transfer LUT over [0, 1]. +macro_rules! tf_lut { + ($name:ident, $f:expr) => { + static $name: std::sync::LazyLock<[f32; LUT_N + 1]> = + std::sync::LazyLock::new(|| { + let f: fn(f32) -> f32 = $f; + let mut t = [0.0f32; LUT_N + 1]; + for (i, e) in t.iter_mut().enumerate() { + *e = f(i as f32 / LUT_N as f32); + } + t + }); + }; +} + +tf_lut!(SRGB_EOTF_LUT, srgb_eotf); +tf_lut!(SRGB_OETF_LUT, srgb_oetf); +tf_lut!(GAMMA22_EOTF_LUT, |v| gamma_eotf(v, 2.2)); +tf_lut!(GAMMA22_OETF_LUT, |v| gamma_oetf(v, 2.2)); +tf_lut!(GAMMA28_EOTF_LUT, |v| gamma_eotf(v, 2.8)); +tf_lut!(PQ_EOTF_LUT, pq_eotf); +tf_lut!(PQ_OETF_LUT, pq_oetf); +tf_lut!(HLG_EOTF_LUT, hlg_eotf); +tf_lut!(HLG_OETF_LUT, hlg_oetf); + +/// Evaluate a transfer function through its [0, 1] LUT (linear +/// interpolation; max error ≈ 2e-5 — far below 10-bit quantization). +/// Negatives mirror like the exact functions; values above 1.0 fall back +/// to the exact function (HDR super-whites are rare and already clamped +/// by the output node). +#[inline] +fn lut_tf(lut: &[f32; LUT_N + 1], exact: fn(f32) -> f32, v: f32) -> f32 { + let (sign, a) = if v < 0.0 { (-1.0, -v) } else { (1.0, v) }; + let out = if a >= 1.0 { + exact(a) + } else { + let x = a * LUT_N as f32; + let i = (x as usize).min(LUT_N - 1); + let frac = x - i as f32; + lut[i] + frac * (lut[i + 1] - lut[i]) + }; + sign * out +} + +/// Fast [`srgb_eotf`]: the linear toe is exact, the power part is LUT'd. +#[inline] +fn srgb_eotf_fast(v: f32) -> f32 { + if v.abs() <= 0.04045 { + v / 12.92 + } else { + lut_tf(&SRGB_EOTF_LUT, srgb_eotf, v) + } +} + +/// Fast [`srgb_oetf`]: the linear toe is exact, the power part is LUT'd. +#[inline] +fn srgb_oetf_fast(v: f32) -> f32 { + if v.abs() <= 0.0031308 { + 12.92 * v + } else { + lut_tf(&SRGB_OETF_LUT, srgb_oetf, v) + } +} + +/// The fast decode-side linearizer for `transfer`. +fn decode_transfer_fn(transfer: SourceTransfer) -> fn(f32) -> f32 { + match transfer { + SourceTransfer::SdrGamma | SourceTransfer::Unknown => srgb_eotf_fast, + SourceTransfer::Gamma22 => |v| lut_tf(&GAMMA22_EOTF_LUT, |x| gamma_eotf(x, 2.2), v), + SourceTransfer::Gamma28 => |v| lut_tf(&GAMMA28_EOTF_LUT, |x| gamma_eotf(x, 2.8), v), + SourceTransfer::Pq => |v| lut_tf(&PQ_EOTF_LUT, pq_eotf, v), + SourceTransfer::Hlg => |v| lut_tf(&HLG_EOTF_LUT, hlg_eotf, v), + SourceTransfer::Linear => |v| v, + } +} + +/// The fast output-side encoder for `transfer`. +fn output_transfer_fn(transfer: OutputTransfer) -> fn(f32) -> f32 { + match transfer { + OutputTransfer::Srgb => srgb_oetf_fast, + OutputTransfer::Gamma22 => |v| lut_tf(&GAMMA22_OETF_LUT, |x| gamma_oetf(x, 2.2), v), + OutputTransfer::Pq => |v| lut_tf(&PQ_OETF_LUT, pq_oetf, v), + OutputTransfer::Hlg => |v| lut_tf(&HLG_OETF_LUT, hlg_oetf, v), + } +} + +/// Split `samples` (whole pixels = 4 floats) into row bands and run `f` +/// on each band, on scoped threads when the frame is big enough to be +/// worth the spawn overhead (a single 1080p frame is ~8 Mpx; the per- +/// pixel work below is memory- and ALU-bound, so bands scale). +fn par_pixels_f32(samples: &mut [f32], f: impl Fn(&mut [f32]) + Sync) { + par_bands(samples, 4, f); +} + +/// The `&mut [u8]` analog of [`par_pixels_f32`] (16 bytes per pixel). +fn par_pixels_bytes(bytes: &mut [u8], f: impl Fn(&mut [u8]) + Sync) { + par_bands(bytes, 16, f); +} + +fn par_bands(buf: &mut [T], align: usize, f: impl Fn(&mut [T]) + Sync) { + let len = buf.len(); + let bands = if len / align >= (1 << 19) { + std::thread::available_parallelism() + .map(|n| n.get()) + .unwrap_or(1) + .min(8) + } else { + 1 + }; + if bands <= 1 { + f(buf); + return; + } + let band_len = (len / bands).div_ceil(align) * align; + let f = &f; + std::thread::scope(|s| { + let mut rest = buf; + while !rest.is_empty() { + let (band, tail) = rest.split_at_mut(band_len.min(rest.len())); + s.spawn(move || f(band)); + rest = tail; + } + }); +} + // --------------------------------------------------------------------------- // Source decode characterization // --------------------------------------------------------------------------- @@ -497,11 +638,13 @@ pub enum SourcePrimaries { /// The transfer characteristic of a decoded source. #[derive(Clone, Copy, Debug, PartialEq, Eq)] pub enum SourceTransfer { - /// The sRGB piecewise EOTF (IEC 61966-2-1), applied at decode. + /// The sRGB piecewise EOTF (IEC 61966-2-1), applied at decode. Also + /// used for BT.709 / BT.2020 SDR video: the BT.709 camera OETF inverse + /// is ≈ gamma 2.2 (sRGB EOTF to within measurement noise), while + /// decoding with the BT.1886 *display* EOTF (2.4) bakes in the ~1.1 + /// system gamma that belongs at final display only — SDR sources + /// round-tripped through the working space came out visibly dark. SdrGamma, - /// BT.1886 display-referred EOTF (pure power 2.4) — the SDR display - /// reference for BT.709 / BT.2020 sources. - Gamma24, /// Pure power 2.2 gamma. Gamma22, /// Pure power 2.8 gamma. @@ -532,13 +675,13 @@ pub fn source_primaries_from_av(color_primaries: i32) -> SourcePrimaries { /// (AVCOL_TRC_* numbering, H.273 ISO codes.) pub fn source_transfer_from_av(color_trc: i32) -> SourceTransfer { match color_trc { - 1 => SourceTransfer::Gamma24, // BT.709 + 1 => SourceTransfer::SdrGamma, // BT.709 (see the enum docs: OETF inverse ≈ 2.2) 4 => SourceTransfer::Gamma22, // gamma 2.2 5 => SourceTransfer::Gamma28, // gamma 2.8 - 6 => SourceTransfer::Gamma24, // SMPTE 170M + 6 => SourceTransfer::SdrGamma, // SMPTE 170M 13 => SourceTransfer::SdrGamma, // sRGB (its EOTF is applied at decode) - 14 => SourceTransfer::Gamma24, // BT.2020 10-bit - 15 => SourceTransfer::Gamma24, // BT.2020 12-bit + 14 => SourceTransfer::SdrGamma, // BT.2020 10-bit + 15 => SourceTransfer::SdrGamma, // BT.2020 12-bit 16 => SourceTransfer::Pq, // SMPTE ST 2084 18 => SourceTransfer::Hlg, // ARIB STD-B67 8 => SourceTransfer::Linear, // linear @@ -563,55 +706,23 @@ impl SourcePrimaries { /// Decode-direction transform: gamma-encoded source RGB (in the source's /// own primaries) → ACEScg linear. `samples` is an F32 RGBA buffer, -/// transformed in place. +/// transformed in place. Linearize and the primaries matrix are fused +/// into a single pass (LUT'd transfer, row-band parallel). pub fn decode_to_acescg( samples: &mut [f32], primaries: SourcePrimaries, transfer: SourceTransfer, ) { - // Stage 1: linearize in the source's own primaries. - match transfer { - SourceTransfer::SdrGamma | SourceTransfer::Unknown => { - for px in samples.chunks_exact_mut(4) { - for c in 0..3 { - px[c] = srgb_eotf(px[c]); - } - } - } - SourceTransfer::Gamma24 | SourceTransfer::Gamma22 | SourceTransfer::Gamma28 => { - let gamma = match transfer { - SourceTransfer::Gamma24 => 2.4, - SourceTransfer::Gamma22 => 2.2, - _ => 2.8, - }; - for px in samples.chunks_exact_mut(4) { - for c in 0..3 { - px[c] = gamma_eotf(px[c], gamma); - } - } - } - SourceTransfer::Pq => { - for px in samples.chunks_exact_mut(4) { - for c in 0..3 { - px[c] = pq_eotf(px[c]); - } - } - } - SourceTransfer::Hlg => { - for px in samples.chunks_exact_mut(4) { - for c in 0..3 { - px[c] = hlg_eotf(px[c]); - } - } - } - SourceTransfer::Linear => {} - } - - // Stage 2: source primaries → AP1 (with chromatic adaptation). + let linearize = decode_transfer_fn(transfer); let matrix = rgb_to_rgb_matrix(primaries.primaries(), PRIMARIES_AP1); - for px in samples.chunks_exact_mut(4) { - apply_mat(matrix, px); - } + par_pixels_f32(samples, move |band| { + for px in band.chunks_exact_mut(4) { + for c in 0..3 { + px[c] = linearize(px[c]); + } + apply_mat(matrix, px); + } + }); } /// Decode-direction transform on a little-endian F32 RGBA byte buffer @@ -626,33 +737,26 @@ pub fn decode_to_acescg_bytes( if bytes.len() < pixels * 16 { return; } - let linearize: fn(f32) -> f32 = match transfer { - SourceTransfer::SdrGamma | SourceTransfer::Unknown => srgb_eotf, - SourceTransfer::Gamma24 => |v| gamma_eotf(v, 2.4), - SourceTransfer::Gamma22 => |v| gamma_eotf(v, 2.2), - SourceTransfer::Gamma28 => |v| gamma_eotf(v, 2.8), - SourceTransfer::Pq => pq_eotf, - SourceTransfer::Hlg => hlg_eotf, - SourceTransfer::Linear => |v| v, - }; + let linearize = decode_transfer_fn(transfer); let matrix = rgb_to_rgb_matrix(primaries.primaries(), PRIMARIES_AP1); - for i in 0..pixels { - let off = i * 16; - let mut px = [ - f32::from_le_bytes(bytes[off..off + 4].try_into().unwrap()), - f32::from_le_bytes(bytes[off + 4..off + 8].try_into().unwrap()), - f32::from_le_bytes(bytes[off + 8..off + 12].try_into().unwrap()), - f32::from_le_bytes(bytes[off + 12..off + 16].try_into().unwrap()), - ]; - for c in 0..3 { - px[c] = linearize(px[c]); + par_pixels_bytes(bytes, move |band| { + for px in band.chunks_exact_mut(16) { + let mut v = [ + f32::from_le_bytes(px[0..4].try_into().unwrap()), + f32::from_le_bytes(px[4..8].try_into().unwrap()), + f32::from_le_bytes(px[8..12].try_into().unwrap()), + f32::from_le_bytes(px[12..16].try_into().unwrap()), + ]; + for c in 0..3 { + v[c] = linearize(v[c]); + } + apply_mat(matrix, &mut v); + px[0..4].copy_from_slice(&v[0].to_le_bytes()); + px[4..8].copy_from_slice(&v[1].to_le_bytes()); + px[8..12].copy_from_slice(&v[2].to_le_bytes()); + px[12..16].copy_from_slice(&v[3].to_le_bytes()); } - apply_mat(matrix, &mut px); - bytes[off..off + 4].copy_from_slice(&px[0].to_le_bytes()); - bytes[off + 4..off + 8].copy_from_slice(&px[1].to_le_bytes()); - bytes[off + 8..off + 12].copy_from_slice(&px[2].to_le_bytes()); - bytes[off + 12..off + 16].copy_from_slice(&px[3].to_le_bytes()); - } + }); } /// Output-direction transform: ACEScg linear → the output colorspace @@ -662,16 +766,19 @@ pub fn decode_to_acescg_bytes( /// AP1 → target-gamut matrix, the RGB channels are clamped to [0, 1] before /// the transfer encoding, so out-of-gamut values are clipped rather than /// wrapped. This applies to HDR targets too (PQ/HLG code 1.0 = 10 000 nits); -/// alpha is untouched. +/// alpha is untouched. Matrix, clamp and encoding are fused into a single +/// pass (LUT'd transfer, row-band parallel). pub fn acescg_to_output(samples: &mut [f32], spec: OutputColorSpec) { let matrix = rgb_to_rgb_matrix(PRIMARIES_AP1, spec.gamut.primaries()); - for px in samples.chunks_exact_mut(4) { - apply_mat(matrix, px); - for c in 0..3 { - px[c] = px[c].clamp(0.0, 1.0); + let encode = output_transfer_fn(spec.transfer); + par_pixels_f32(samples, move |band| { + for px in band.chunks_exact_mut(4) { + apply_mat(matrix, px); + for c in 0..3 { + px[c] = encode(px[c].clamp(0.0, 1.0)); + } } - } - apply_transfer_oetf(samples, spec.transfer); + }); } /// Output-direction transform on a little-endian F32 RGBA byte buffer @@ -683,67 +790,38 @@ pub fn acescg_to_output_bytes(bytes: &mut [u8], pixels: usize, spec: OutputColor return; } let matrix = rgb_to_rgb_matrix(PRIMARIES_AP1, spec.gamut.primaries()); - let encode: fn(f32) -> f32 = match spec.transfer { - OutputTransfer::Srgb => srgb_oetf, - OutputTransfer::Gamma22 => |v| gamma_oetf(v, 2.2), - OutputTransfer::Pq => pq_oetf, - OutputTransfer::Hlg => hlg_oetf, - }; - for i in 0..pixels { - let off = i * 16; - let mut px = [ - f32::from_le_bytes(bytes[off..off + 4].try_into().unwrap()), - f32::from_le_bytes(bytes[off + 4..off + 8].try_into().unwrap()), - f32::from_le_bytes(bytes[off + 8..off + 12].try_into().unwrap()), - f32::from_le_bytes(bytes[off + 12..off + 16].try_into().unwrap()), - ]; - apply_mat(matrix, &mut px); - for c in 0..3 { - px[c] = px[c].clamp(0.0, 1.0); + let encode = output_transfer_fn(spec.transfer); + par_pixels_bytes(bytes, move |band| { + for px in band.chunks_exact_mut(16) { + let mut v = [ + f32::from_le_bytes(px[0..4].try_into().unwrap()), + f32::from_le_bytes(px[4..8].try_into().unwrap()), + f32::from_le_bytes(px[8..12].try_into().unwrap()), + f32::from_le_bytes(px[12..16].try_into().unwrap()), + ]; + apply_mat(matrix, &mut v); + for c in 0..3 { + v[c] = encode(v[c].clamp(0.0, 1.0)); + } + px[0..4].copy_from_slice(&v[0].to_le_bytes()); + px[4..8].copy_from_slice(&v[1].to_le_bytes()); + px[8..12].copy_from_slice(&v[2].to_le_bytes()); + px[12..16].copy_from_slice(&v[3].to_le_bytes()); } - for c in 0..3 { - px[c] = encode(px[c]); - } - bytes[off..off + 4].copy_from_slice(&px[0].to_le_bytes()); - bytes[off + 4..off + 8].copy_from_slice(&px[1].to_le_bytes()); - bytes[off + 8..off + 12].copy_from_slice(&px[2].to_le_bytes()); - bytes[off + 12..off + 16].copy_from_slice(&px[3].to_le_bytes()); - } + }); } /// Apply just the transfer encoding of `transfer` (linear → code). In place -/// on the RGB channels. +/// on the RGB channels (LUT'd, row-band parallel). pub fn apply_transfer_oetf(samples: &mut [f32], transfer: OutputTransfer) { - match transfer { - OutputTransfer::Srgb => { - for px in samples.chunks_exact_mut(4) { - for c in 0..3 { - px[c] = srgb_oetf(px[c]); - } + let encode = output_transfer_fn(transfer); + par_pixels_f32(samples, move |band| { + for px in band.chunks_exact_mut(4) { + for c in 0..3 { + px[c] = encode(px[c]); } } - OutputTransfer::Gamma22 => { - for px in samples.chunks_exact_mut(4) { - for c in 0..3 { - px[c] = gamma_oetf(px[c], 2.2); - } - } - } - OutputTransfer::Pq => { - for px in samples.chunks_exact_mut(4) { - for c in 0..3 { - px[c] = pq_oetf(px[c]); - } - } - } - OutputTransfer::Hlg => { - for px in samples.chunks_exact_mut(4) { - for c in 0..3 { - px[c] = hlg_oetf(px[c]); - } - } - } - } + }); } /// The presentation transform: working space → the display target colorspace @@ -772,21 +850,23 @@ pub fn working_to_display_target( /// Hlg → HLG), then convert RGB → XYZ through the gamut's own matrix. All /// output gamuts share the D65 white point, so no chromatic adaptation is /// needed; `rgb_to_xyz_matrix` already normalizes the white point to unit -/// luminance. +/// luminance. Linearize and matrix are fused into one LUT'd parallel pass. pub fn output_spec_to_xyz_d65(samples: &mut [f32], spec: OutputColorSpec) { - let linearize: fn(f32) -> f32 = match spec.transfer { - OutputTransfer::Srgb => srgb_eotf, - OutputTransfer::Gamma22 => |v| gamma_eotf(v, 2.2), - OutputTransfer::Pq => pq_eotf, - OutputTransfer::Hlg => hlg_eotf, - }; + let linearize = decode_transfer_fn(match spec.transfer { + OutputTransfer::Srgb => SourceTransfer::SdrGamma, + OutputTransfer::Gamma22 => SourceTransfer::Gamma22, + OutputTransfer::Pq => SourceTransfer::Pq, + OutputTransfer::Hlg => SourceTransfer::Hlg, + }); let matrix = rgb_to_xyz_matrix(spec.gamut.primaries()); - for px in samples.chunks_exact_mut(4) { - for c in 0..3 { - px[c] = linearize(px[c]); + par_pixels_f32(samples, move |band| { + for px in band.chunks_exact_mut(4) { + for c in 0..3 { + px[c] = linearize(px[c]); + } + apply_mat(matrix, px); } - apply_mat(matrix, px); - } + }); } // --------------------------------------------------------------------------- @@ -1050,13 +1130,13 @@ mod tests { assert_eq!(source_primaries_from_av(9), SourcePrimaries::Bt2020); assert_eq!(source_primaries_from_av(12), SourcePrimaries::DisplayP3); assert_eq!(source_primaries_from_av(2), SourcePrimaries::Unknown); - assert_eq!(source_transfer_from_av(1), SourceTransfer::Gamma24); + assert_eq!(source_transfer_from_av(1), SourceTransfer::SdrGamma); assert_eq!(source_transfer_from_av(4), SourceTransfer::Gamma22); assert_eq!(source_transfer_from_av(5), SourceTransfer::Gamma28); - assert_eq!(source_transfer_from_av(6), SourceTransfer::Gamma24); + assert_eq!(source_transfer_from_av(6), SourceTransfer::SdrGamma); assert_eq!(source_transfer_from_av(13), SourceTransfer::SdrGamma); - assert_eq!(source_transfer_from_av(14), SourceTransfer::Gamma24); - assert_eq!(source_transfer_from_av(15), SourceTransfer::Gamma24); + assert_eq!(source_transfer_from_av(14), SourceTransfer::SdrGamma); + assert_eq!(source_transfer_from_av(15), SourceTransfer::SdrGamma); assert_eq!(source_transfer_from_av(16), SourceTransfer::Pq); assert_eq!(source_transfer_from_av(18), SourceTransfer::Hlg); assert_eq!(source_transfer_from_av(8), SourceTransfer::Linear); @@ -1121,29 +1201,93 @@ mod tests { } #[test] - fn decode_gamma24_dispatch() { - // BT.1886 decode: code 0.5 linearizes via pure power 2.4 (≈ 0.1895, - // not sRGB's piecewise ≈ 0.2140), then the BT.709 → AP1 matrix. - let mut samples = [0.5f32, 0.5, 0.5, 1.0]; - decode_to_acescg(&mut samples, SourcePrimaries::Bt709, SourceTransfer::Gamma24); - let m = rgb_to_rgb_matrix(PRIMARIES_SRGB, PRIMARIES_AP1); - let expected = mat_vec(m, [gamma_eotf(0.5, 2.4); 3]); - for i in 0..3 { - assert!( - approx(samples[i], expected[i], 1e-6), - "Gamma24 decode channel {i}: {} vs {}", - samples[i], - expected[i] - ); + fn lut_fast_transfer_matches_exact() { + // The LUT'd fast paths track the exact functions over the whole + // [0, 1] domain (and mirror negatives); far below 10-bit quanta. + let cases: &[(&std::sync::LazyLock<[f32; LUT_N + 1]>, fn(f32) -> f32, &str)] = &[ + (&SRGB_EOTF_LUT, srgb_eotf, "srgb_eotf"), + (&SRGB_OETF_LUT, srgb_oetf, "srgb_oetf"), + (&GAMMA22_EOTF_LUT, |v| gamma_eotf(v, 2.2), "gamma22_eotf"), + (&GAMMA22_OETF_LUT, |v| gamma_oetf(v, 2.2), "gamma22_oetf"), + (&GAMMA28_EOTF_LUT, |v| gamma_eotf(v, 2.8), "gamma28_eotf"), + (&PQ_EOTF_LUT, pq_eotf, "pq_eotf"), + (&PQ_OETF_LUT, pq_oetf, "pq_oetf"), + (&HLG_EOTF_LUT, hlg_eotf, "hlg_eotf"), + (&HLG_OETF_LUT, hlg_oetf, "hlg_oetf"), + ]; + for &(lut, exact, name) in cases { + for k in 0..=1000 { + let v = k as f32 / 1000.0; + for v in [v, -v] { + let fast = lut_tf(lut, exact, v); + let want = exact(v); + assert!( + (fast - want).abs() < 2e-4, + "{name}({v}): fast {fast} vs exact {want}" + ); + } + } } - // ... and the bytes variant picks the same curve. + // Above 1.0 the exact function is used. + assert_eq!(lut_tf(&SRGB_OETF_LUT, srgb_oetf, 2.0), srgb_oetf(2.0)); + // The sRGB wrappers keep their linear toes exact. + assert_eq!(srgb_eotf_fast(0.02), 0.02 / 12.92); + assert_eq!(srgb_oetf_fast(0.001), 12.92 * 0.001); + } + + #[test] + fn parallel_transform_matches_scalar_reference() { + // A frame above the band threshold exercises the scoped-thread + // path; it must agree with the scalar math pixel for pixel. + let pixels = 1 << 20; + let mut samples: Vec = (0..pixels * 4) + .map(|i: usize| ((i.wrapping_mul(2654435761)) % 1000) as f32 / 999.0) + .collect(); + let reference: Vec = samples + .chunks_exact(4) + .flat_map(|px| { + let mut v = [srgb_eotf(px[0]), srgb_eotf(px[1]), srgb_eotf(px[2]), px[3]]; + apply_mat(rgb_to_rgb_matrix(PRIMARIES_SRGB, PRIMARIES_AP1), &mut v); + v + }) + .collect(); + decode_to_acescg(&mut samples, SourcePrimaries::Bt709, SourceTransfer::SdrGamma); + for (i, (&got, &want)) in samples.iter().zip(reference.iter()).enumerate() { + assert!(approx(got, want, 1e-4), "pixel {i}: {got} vs {want}"); + } + } + + #[test] + fn bt709_decode_output_round_trip_is_identity() { + // SDR video (BT.709 / BT.2020 NCL tags) decodes with the sRGB EOTF + // (OETF inverse ≈ 2.2 — see the SourceTransfer docs), so an SDR + // frame round-tripped through the ACEScg working space and back to + // sRGB output must come back unchanged. Decoding with the BT.1886 + // display EOTF (2.4) instead made this round trip visibly dark + // (the ~1.1 system gamma belongs at final display only). + for code in [0.05f32, 0.18, 0.5, 0.75, 1.0] { + let mut samples = [code, code, code, 1.0]; + decode_to_acescg(&mut samples, SourcePrimaries::Bt709, SourceTransfer::SdrGamma); + acescg_to_output(&mut samples, OutputColorSpec::default()); + for i in 0..3 { + assert!( + approx(samples[i], code, 1e-4), + "BT.709 round trip code {code} channel {i}: {}", + samples[i] + ); + } + } + // ... and the bytes variant picks the same curve (sRGB piecewise, + // not a pure power): code 0.5 → srgb_eotf(0.5) ≈ 0.2140. let mut bytes = [0u8; 16]; bytes[0..4].copy_from_slice(&0.5f32.to_le_bytes()); bytes[4..8].copy_from_slice(&0.5f32.to_le_bytes()); bytes[8..12].copy_from_slice(&0.5f32.to_le_bytes()); - decode_to_acescg_bytes(&mut bytes, 1, SourcePrimaries::Bt709, SourceTransfer::Gamma24); + decode_to_acescg_bytes(&mut bytes, 1, SourcePrimaries::Bt709, SourceTransfer::SdrGamma); let r = f32::from_le_bytes(bytes[0..4].try_into().unwrap()); - assert!(approx(r, expected[0], 1e-6), "bytes Gamma24 R: {r}"); + let m = rgb_to_rgb_matrix(PRIMARIES_SRGB, PRIMARIES_AP1); + let expected = mat_vec(m, [srgb_eotf(0.5); 3]); + assert!(approx(r, expected[0], 1e-6), "bytes SdrGamma R: {r}"); } #[test] diff --git a/crates/oak-render/src/eval.rs b/crates/oak-render/src/eval.rs index a654213aa..6f51bf00a 100644 --- a/crates/oak-render/src/eval.rs +++ b/crates/oak-render/src/eval.rs @@ -1198,7 +1198,18 @@ fn audio_layout(params: &crate::ticket::AudioTicketParams) -> Result<(i32, u64, if seconds <= 0.0 || seconds > 3600.0 { return Err(Error::Invalid); } - let total_frames = (seconds * rate as f64).round() as usize; + + // Anchor each chunk to the absolute sample grid (`round(out·rate) - + // round(in·rate)`) instead of rounding the duration: at fractional + // frame rates (29.97 fps → 1601.6 samples/frame) a duration-round + // would emit 1602 samples for every chunk and accumulate ~12 extra + // samples per second, slowly desyncing audio from video. Per-chunk + // anchoring keeps the total exact and matches the decode side + // (`FFmpegDecoder::retrieve_audio_to` fills `round(out·rate) - + // round(in·rate)` samples). + let total_frames = ((params.range.out().to_f64() * rate as f64).round() + - (params.range.in_().to_f64() * rate as f64).round()) + .max(0.0) as usize; Ok((rate, params.channel_layout, channels, total_frames)) } @@ -1230,8 +1241,11 @@ fn mix_audio_montage( if out_time <= in_time { continue; } - let start_frame = ((in_time - params.range.in_()).to_f64() * rate as f64) as usize; - let end_frame = ((out_time - params.range.in_()).to_f64() * rate as f64) as usize; + // Round to the absolute sample grid like the decode side (which + // anchors at `round(media_start·rate)`): truncation would shift a + // clip's mix by up to one sample per chunk. + let start_frame = ((in_time - params.range.in_()).to_f64() * rate as f64).round() as usize; + let end_frame = ((out_time - params.range.in_()).to_f64() * rate as f64).round() as usize; if start_frame >= total_frames { continue; } diff --git a/crates/oak-render/src/manager.rs b/crates/oak-render/src/manager.rs index 4006672b9..3f37ac357 100644 --- a/crates/oak-render/src/manager.rs +++ b/crates/oak-render/src/manager.rs @@ -232,6 +232,33 @@ impl RenderManager { Ok(()) } + /// M16 S1 graph mode: force the workers to re-load the current project + /// snapshot even when the undo-stack revision is unchanged. The + /// color-settings dialog writes project settings directly (no undo + /// command), so the revision-based dedup in [`set_graph_snapshot`] would + /// never re-send the snapshot — workers would keep rendering under the + /// colors they loaded at project-load time. The snapshot file is + /// rewritten (see [`GraphSnapshotStore::acquire_rewrite`]) and + /// `load_graph` re-broadcast to every live worker (the dispatcher's + /// re-send has no dedup). + pub fn resync_graph_snapshot( + &self, + project: &std::sync::Mutex, + revision: u64, + ) -> Result<()> { + if self.stopping.load(Ordering::Acquire) { + return Ok(()); // teardown: no re-arm after the drain + } + let path = self.snapshots.acquire_rewrite(project, revision)?; + if let Some(old) = lock(&self.current_snapshot).replace(path.clone()) { + self.snapshots.release(&old); + } + self.dispatch.set_graph_snapshot(Some(path)); + let uuid = lock(project).uuid.clone(); + *lock(&self.current_key) = Some((uuid, revision)); + Ok(()) + } + /// M16 S1 graph mode: drop the current snapshot (project closed). The /// protocol has no clear message, so alive workers keep their loaded /// graph; new/restarted workers no longer load it and the snapshot file diff --git a/crates/oak-render/src/worker.rs b/crates/oak-render/src/worker.rs index a939ee066..932c4f099 100644 --- a/crates/oak-render/src/worker.rs +++ b/crates/oak-render/src/worker.rs @@ -352,6 +352,51 @@ impl GraphSnapshotStore { Ok(path_str) } + /// Like [`acquire`], but always rewrites the snapshot file even when the + /// (uuid, revision) key is already present. Used to force the workers to + /// re-deserialize a project whose settings changed without an undo-stack + /// revision bump (color-settings resync): the dedup in [`acquire`] would + /// otherwise hand the workers the stale file and `load_graph` would never + /// re-broadcast. Atomic staging and reference counting are identical to + /// [`acquire`]. + pub fn acquire_rewrite( + &self, + project: &std::sync::Mutex, + revision: u64, + ) -> Result { + let (uuid, xml) = { + let g = lock(project); + let xml = oak_node::serializer::save(&g) + .map_err(|e| Error::Failed(format!("save graph snapshot: {e}")))?; + (g.uuid.clone(), xml) + }; + let path = self.dir.join(format!("graph-{uuid}-{revision}.xml")); + let path_str = path.to_string_lossy().into_owned(); + // Atomic staging: temp file + rename (see [`acquire`]). + let tmp = self + .dir + .join(format!("graph-{uuid}-{revision}.{}.tmp", std::process::id())); + std::fs::write(&tmp, &xml) + .map_err(|e| Error::Failed(format!("write snapshot temp: {e}")))?; + if let Err(e) = std::fs::rename(&tmp, &path) { + let _ = std::fs::remove_file(&tmp); + return Err(Error::Failed(format!("rename snapshot: {e}"))); + } + let mut entries = lock(&self.entries); + if let Some(entry) = entries.get_mut(&path_str) { + entry.refs += 1; + } else { + entries.insert( + path_str.clone(), + SnapshotEntry { + refs: 1, + cached: false, + }, + ); + } + Ok(path_str) + } + /// Drop one reference. The FILE IS NOT UNLINKED: a worker may still /// hold the path for a late `load_graph` (M16 S1), and per-project /// filenames mean a stale snapshot is never confused with a live one. @@ -608,4 +653,42 @@ mod tests { "cleanup removes the snapshot directory" ); } + + #[test] + fn acquire_rewrite_forces_file_rewrite_on_same_key() { + let store = GraphSnapshotStore::new(); + let project = oak_node::project::Project::new(); + // First snapshot: the default project (working space ACEScg). + let p1 = store.acquire(&project, 1).unwrap(); + let before = std::fs::read_to_string(&p1).unwrap(); + assert!( + std::path::Path::new(&p1).exists(), + "snapshot file written" + ); + assert_eq!(store.refs(&p1), 1); + + // A settings change with no revision bump: plain acquire must NOT + // rewrite the file (the dedup the resync exists to bypass)... + { + let mut guard = project.lock().unwrap_or_else(|e| e.into_inner()); + guard.set_working_color_space(oak_common::colormath::WorkingColorSpace::SrgbLegacy); + } + let p2 = store.acquire(&project, 1).unwrap(); + assert_eq!(p1, p2, "same (uuid, revision) key reuses the file"); + assert_eq!(std::fs::read_to_string(&p1).unwrap(), before, "acquire never rewrites"); + store.release(&p2); + + // ...while acquire_rewrite rewrites it in place at the same key. + let p3 = store.acquire_rewrite(&project, 1).unwrap(); + assert_eq!(p1, p3, "rewrite keeps the same path token"); + let after = std::fs::read_to_string(&p1).unwrap(); + assert_ne!(after, before, "rewrite replaces the snapshot content"); + assert!( + after.contains("srgb_legacy"), + "rewritten snapshot carries the new setting: {after}" + ); + store.release(&p3); + store.release(&p1); + store.cleanup(); + } } diff --git a/crates/oak-worker/src/worker.rs b/crates/oak-worker/src/worker.rs index 4e884e86b..8225685aa 100644 --- a/crates/oak-worker/src/worker.rs +++ b/crates/oak-worker/src/worker.rs @@ -1047,8 +1047,40 @@ impl WorkerSession { }) } + /// Refresh the process-global pipeline color settings from the loaded + /// project snapshot (the source of truth for what this worker renders). + /// Returns true when the settings changed — the caller must drop the + /// frame cache, whose F32 bytes were produced under the old settings + /// (M16 S2). The resync keeps the global current via `load_graph`; this + /// covers tickets in flight before that IPC lands, and mirrors + /// [`handle_load_graph`]'s adopt step. + fn sync_pipeline_color_from_graph(&mut self) -> bool { + let Some(graph) = &self.graph else { return false; }; + let Some(project) = &graph.project else { return false; }; + let (working, output) = { + let guard = project.lock().unwrap_or_else(|e| e.into_inner()); + (guard.working_color_space(), guard.output_color_spec()) + }; + if oak_render::color::pipeline_working_space() == working + && oak_render::color::pipeline_output_spec() == output + { + return false; + } + oak_render::color::set_pipeline_color_settings(working, output); + true + } + /// Render `spec` into the slot's data block and fill the slot meta. fn render_spec_pixels(&mut self, spec: &BatchTicketSpec, pool: &FrameSlotPool) -> Result<(), String> { + // Derive the pipeline colors from the loaded project on every render: + // eval's decode linearization and the output node below read the + // process global, which the resync (load_graph) keeps current — but a + // ticket in flight before that IPC lands must not render under stale + // colors. A change invalidates the F32 frame cache: cached bytes were + // produced under the previous settings (M16 S2). + if self.sync_pipeline_color_from_graph() { + self.frame_cache.clear(); + } let w = spec.width; let h = spec.height; if w <= 0 || h <= 0 { @@ -1806,6 +1838,130 @@ mod tests { let _ = std::fs::remove_file(&bad); } + #[test] + fn load_graph_adopts_pipeline_colors_from_snapshot() { + use oak_common::colormath::{OutputColorSpec, WorkingColorSpace}; + // Reset the process global to something different from the snapshot's + // settings so the adopt step is observable. + oak_render::color::set_pipeline_color_settings( + WorkingColorSpace::SrgbLegacy, + OutputColorSpec::default(), + ); + let project = oak_node::project::Project::new(); + { + let mut guard = project.lock().unwrap_or_else(|e| e.into_inner()); + guard.set_working_color_space(WorkingColorSpace::AcesCg); + } + let xml = oak_node::serializer::save(&project.lock().unwrap_or_else(|e| e.into_inner())) + .expect("serialize project"); + let path = std::env::temp_dir().join("oak_worker_main_test_acescg.ove"); + std::fs::write(&path, &xml).unwrap(); + + let mut s = WorkerSession::create("none").unwrap(); + let resp = s.handle_line( + &json!({ "type": "load_graph", "path": path.display().to_string() }).to_string(), + ); + assert!(resp.is_none(), "unexpected error: {resp:?}"); + assert_eq!( + oak_render::color::pipeline_working_space(), + WorkingColorSpace::AcesCg, + "load_graph must adopt the snapshot's working space" + ); + let _ = std::fs::remove_file(&path); + // Restore the default global so parallel tests are not disturbed. + oak_render::color::set_pipeline_color_settings( + WorkingColorSpace::default(), + OutputColorSpec::default(), + ); + } + + #[test] + fn sync_pipeline_color_from_graph_restores_stale_global() { + use oak_common::colormath::{OutputColorSpec, WorkingColorSpace}; + let project = oak_node::project::Project::new(); + { + let mut guard = project.lock().unwrap_or_else(|e| e.into_inner()); + guard.set_working_color_space(WorkingColorSpace::AcesCg); + } + let xml = oak_node::serializer::save(&project.lock().unwrap_or_else(|e| e.into_inner())) + .expect("serialize project"); + let path = std::env::temp_dir().join("oak_worker_main_test_sync.ove"); + std::fs::write(&path, &xml).unwrap(); + let mut s = WorkerSession::create("none").unwrap(); + let resp = s.handle_line( + &json!({ "type": "load_graph", "path": path.display().to_string() }).to_string(), + ); + assert!(resp.is_none(), "unexpected error: {resp:?}"); + + // Simulate the app committing new project settings: the process + // global flips immediately, but the resync (load_graph re-broadcast) + // reaches this worker later. A render ticket in that window must not + // run under the stale colors. + oak_render::color::set_pipeline_color_settings( + WorkingColorSpace::SrgbLegacy, + oak_render::color::pipeline_output_spec(), + ); + assert!( + s.sync_pipeline_color_from_graph(), + "stale global must be refreshed from the loaded project" + ); + assert_eq!( + oak_render::color::pipeline_working_space(), + WorkingColorSpace::AcesCg, + "sync must restore the snapshot's working space" + ); + assert!( + !s.sync_pipeline_color_from_graph(), + "no change means no frame-cache invalidation" + ); + let _ = std::fs::remove_file(&path); + oak_render::color::set_pipeline_color_settings( + WorkingColorSpace::default(), + OutputColorSpec::default(), + ); + } + + #[test] + fn acescg_pipeline_round_trip_preserves_srgb_encoded_midgray() { + // The ACEScg overexposure bug: F32 bytes produced by the legacy sRGB + // pass-through (already sRGB-encoded) are re-encoded by the ACEScg + // output transform, applying the sRGB OETF a second time — 0.5 + // becomes srgb_oetf(0.5) ≈ 0.735. The correct pipeline linearizes + // the code (sRGB EOTF) into ACEScg and then encodes for output, so + // the mid-gray comes back near 0.5. + use oak_common::colormath::{ + acescg_to_output_bytes, decode_to_acescg_bytes, srgb_oetf, OutputColorSpec, OutputGamut, + OutputTransfer, SourcePrimaries, SourceTransfer, WorkingColorSpace, + }; + let spec = OutputColorSpec { + gamut: OutputGamut::Srgb, + transfer: OutputTransfer::Srgb, + }; + + // Buggy path: treating the already-encoded code as ACEScg linear. + let mut double_encoded = [0.5f32, 0.5, 0.5, 1.0]; + oak_common::colormath::working_to_display_target( + &mut double_encoded, + WorkingColorSpace::AcesCg, + spec, + ); + assert!( + (double_encoded[0] - srgb_oetf(0.5)).abs() < 1e-4, + "mis-encoding 0.5 as ACEScg-linear must yield srgb_oetf(0.5) ≈ 0.735, got {}", + double_encoded[0] + ); + + // Correct path: decode (sRGB EOTF) → ACEScg → output encode. + let mut bytes = [0.5f32, 0.5, 0.5, 1.0].map(f32::to_le_bytes).concat(); + decode_to_acescg_bytes(&mut bytes, 1, SourcePrimaries::Bt709, SourceTransfer::SdrGamma); + acescg_to_output_bytes(&mut bytes, 1, spec); + let out = f32::from_le_bytes(bytes[0..4].try_into().unwrap()); + assert!( + (out - 0.5).abs() < 1e-3, + "correct pipeline must preserve mid-gray, got {out}" + ); + } + #[test] fn render_frame_without_pool_reports_error_with_ticket() { let mut s = WorkerSession::create("none").unwrap();