diff --git a/crates/oak-cli/src/engine.rs b/crates/oak-cli/src/engine.rs index 2d0e79a25..28643cf11 100644 --- a/crates/oak-cli/src/engine.rs +++ b/crates/oak-cli/src/engine.rs @@ -47,6 +47,7 @@ use oaktimeline::undogeneral::TimelineAddTrackCommand; use oaktimeline::undopointer::TrackPlaceBlockCommand; use oaktimeline::util::NodeRef; use oakrender::manager::RenderManager; +use oakrender::procpool::bgra8_to_rgba8; use oakrender::ticket::{ AudioTicketParams, MontageClip, TicketPayload, VideoTicketParams, }; @@ -628,7 +629,9 @@ pub struct RenderedFrame { } /// Render one frame of the sequence's montage at `time` (seconds -/// rational) into a `(width, height)` F32 frame. +/// rational) into a `(width, height)` F32 frame. M15 S2: the process +/// backend renders a BGRA8 shm slot; the frame is copied out once as +/// RGBA8 u8 (the PPM writer's format 0) and the slot released. pub fn render_frame( seq_id: NodeId, time: Rational, @@ -663,11 +666,35 @@ pub fn render_frame( data: frame.data.clone(), }) } + Ok(TicketPayload::ShmFrame(frame)) => { + let out = shm_to_rendered_frame(frame); + m.release_frame(frame); + Ok(out) + } Ok(TicketPayload::Video(_)) => Err("render produced a non-CPU frame".to_string()), _ => Err("render produced no video frame".to_string()), } } +/// Copy a process-backend shm frame (BGRA8 slot) out into the CLI's +/// RGBA8 u8 frame layout (format 0, 4 channels — the PPM writer's RGB +/// order). Necessary copy: the CLI owns the pixels it writes to disk. +fn shm_to_rendered_frame(frame: &oakrender::procpool::ShmFrameRef) -> RenderedFrame { + let meta = &frame.meta; + let pixels = frame + .shm + .slot_bytes(frame.slot) + .get(..meta.data_size.max(0) as usize) + .unwrap_or_default(); + RenderedFrame { + width: meta.width, + height: meta.height, + format: 0, + linesize: meta.width.max(0) * 4, + data: bgra8_to_rgba8(pixels), + } +} + /// Rendered interleaved f32 audio (the module audio ticket payload). pub struct RenderedAudio { /// Interleaved samples (`frame_count * channel_count` values). diff --git a/crates/oak-worker/README.md b/crates/oak-worker/README.md index ea162c89d..3493d4680 100644 --- a/crates/oak-worker/README.md +++ b/crates/oak-worker/README.md @@ -134,3 +134,13 @@ binary against a created segment to see the real attach path: ```sh target/release/oak-worker --backend cpu <<< '{"type":"shutdown"}' ``` + +## M15 S2 status + +The process-isolated backend is now the **default** `RenderManager` +backend; the in-process render thread pool was deleted (the app drives the +dispatcher from its UI tick and from blocking ticket waits, and the +pre-render window feeds the scheduler ahead of the playhead). The worker +binary is located at `target/debug/oak-worker` next to the main executable +during development (or via `OAK_WORKER_BIN` / `DispatcherConfig::worker_bin`), +and bundled alongside the main binaries by the packager (root `Cargo.toml`). diff --git a/crates/oak-worker/tests/procpool_integration.rs b/crates/oak-worker/tests/procpool_integration.rs index 198bfd652..7ff32af79 100644 --- a/crates/oak-worker/tests/procpool_integration.rs +++ b/crates/oak-worker/tests/procpool_integration.rs @@ -36,7 +36,7 @@ use oakrender::procpool::{ main_heap_frame_copies, reset_main_heap_frame_copies, DispatcherConfig, ProcessDispatcher, }; use oakrender::ticket::{TicketPayload, TicketResult, VideoTicketParams}; -use oakrender::worker::{Job, JobDispatch}; +use oakrender::worker::{Job, JobDispatch, JobSchedule}; /// Serialize every test in this file (shared process environment + /// real child processes). @@ -104,6 +104,7 @@ fn submit( done: Box::new(move |result| { results.lock().unwrap_or_else(|e| e.into_inner()).push(result); }), + schedule: JobSchedule::seek(), }; assert!(dispatcher.post(job), "post accepted while alive"); } diff --git a/crates/oakengine/src/stubs.rs b/crates/oakengine/src/stubs.rs index 12865e473..382ec6d01 100644 --- a/crates/oakengine/src/stubs.rs +++ b/crates/oakengine/src/stubs.rs @@ -12672,6 +12672,31 @@ pub mod render { } Err(e) => e.code(), }, + Some(Ok(TicketPayload::ShmFrame(frame))) => { + // M15 S2 process backend: the frame lives in a worker shm + // slot. Copy it out once as an RGBA8 u8 frame (necessary + // copy — the C ABI consumer owns its buffer), release the + // slot, and hand the frame back. + let meta = &frame.meta; + let pixels = frame + .shm + .slot_bytes(frame.slot) + .get(..meta.data_size.max(0) as usize) + .unwrap_or_default(); + let mut f = oakrender::texture::Frame::new(); + f.width = meta.width; + f.height = meta.height; + f.format = oakcore_rs::PixelFormat::U8; + f.channels = 4; + f.timestamp = oakcore_rs::Rational::new(meta.time_num, meta.time_den); + f.data = oakrender::procpool::bgra8_to_rgba8(pixels); + if let Some(m) = oakrender::manager::RenderManager::global() { + m.release_frame(&frame); + } + // SAFETY: valid out pointer. + unsafe { *out = oakrender::handle::make_owned(f) }; + 0 + } Some(Err(e)) => e.code(), _ => { // SAFETY: valid out pointer. diff --git a/crates/oakrender/COVERAGE.md b/crates/oakrender/COVERAGE.md index 3db227aa4..d6d62cede 100644 --- a/crates/oakrender/COVERAGE.md +++ b/crates/oakrender/COVERAGE.md @@ -17,7 +17,7 @@ |---|---| | `create_instance` / `destroy_instance` / `instance` | `manager::RenderManager::init/shutdown/global` | | `render_frame` / `render_audio`（RenderVideoParams/RenderAudioParams） | `manager` + `ticket::TicketArena::submit_video/submit_audio`（params 结构在 ticket.rs marshalling） | -| `RenderThread`（start/add_ticket/remove_ticket/quit/wait/run） | `worker::WorkerPool`（scoped 线程 + channel；quit/wait → shutdown） | +| `RenderThread`（start/add_ticket/remove_ticket/quit/wait/run） | 已删除（M15 S2 删线程池）；视频经 `procpool::ProcessDispatcher`，单测/音频经 `worker::InlineDispatcher` | | `backend()` / `requested_backend` / `backend_from_string` / `backend_to_string` | `backend::BackendKind` + 字符串互转 | | `get_cacher` | `manager.autocacher` | | `set_project` | `manager`（持有项目身份，不持指针） | @@ -37,11 +37,11 @@ | C++ | Rust 落点 | |---|---| -| `start` / `submit_frame` / `remove_ticket` / `shutdown` / `worker_loop` / `process_job(_attempt)` | `worker::WorkerPool`（线程池路径） | -| `PooledWorker` / `acquire_worker` / `return_worker` / `shutdown_worker` / `clear_graph_cache` | `worker::ProcessPool`（子进程池；复用现有 oakengine_ipc C ABI，Rust 只做客户端） | +| `start` / `submit_frame` / `remove_ticket` / `shutdown` / `worker_loop` / `process_job(_attempt)` | `procpool::ProcessDispatcher`（oak-worker 子进程，M15 默认后端；`worker::WorkerPool` 已删除） | +| `PooledWorker` / `acquire_worker` / `return_worker` / `shutdown_worker` / `clear_graph_cache` | `procpool::ProcessDispatcher` + `WorkerHandle`（spawn/handshake/崩溃重启；pre-M15 `worker::ProcessPool` 桩已删除） | | `write_graph_snapshot` / `cleanup_graph_file` / `add/release_graph_path_ref(_locked)` / `set_graph_path_cached(_locked)` | `worker::GraphSnapshotStore`（图快照文件的引用计数缓存） | | `is_supported` / `prepare_job` / `finish_with_frame` / `cancel_active_process` / `set/clear_active_worker` | `worker` 内部 | -| **决策注记**：进程隔离 worker 保留（崩溃隔离是线上特性）。线程池与进程池并存于 `worker.rs`：`enum WorkerBackend { Threads(WorkerPool), Processes(ProcessPool) }`，选择策略同 C++（config 键）。 | | +| **决策注记**：进程隔离 worker 是唯一视频后端（M15 S2 用户要求删线程池）。`manager::RenderBackendChoice::{Threads,Processes}`：`Threads` = 测试专用同步 inline；`Processes` = 默认。`worker::InlineDispatcher`（sync=音频/测试，queued=测试确定性）。 | | ## 4. Renderer 抽象（renderer.h，后端接口） diff --git a/crates/oakrender/README.md b/crates/oakrender/README.md index 268e01322..cf764f1b4 100644 --- a/crates/oakrender/README.md +++ b/crates/oakrender/README.md @@ -117,16 +117,22 @@ tests/ contract + golden tests (common/ has shared helpers) in float. `oakrender_color_processor_create_transform` resolves the destination transform against the default config's reference role until the oakcommon color-transform bridge lands. -- **Worker process isolation** — landed in M15 S1: `procpool.rs` +- **Worker process isolation** — landed in M15: `procpool.rs` (`ProcessDispatcher`) + `scheduler.rs` + `ipc.rs` drive real oak-worker processes (spawn, handshake, batched renders into main-assigned shm slots, crash restart, zero-copy completions); the end-to-end and crash-isolation tests live in - `crates/oak-worker/tests/procpool_integration.rs`. The frozen pre-M15 - `worker::ProcessPool` facade stub remains for the C ABI. -- **Audio rendering** — audio tickets complete with `Error::Failed` - (the audio graph path is not implemented); `oakrender_ticket_get_samples` - fails explainably. + `crates/oak-worker/tests/procpool_integration.rs`. M15 S2 made the + process backend the `RenderManager` default, removed the in-process + thread pool (`worker::WorkerPool`) and the frozen pre-M15 + `worker::ProcessPool` facade stub, and wired the app's onscreen path + to read the shm slots zero-copy (`worker::InlineDispatcher` supplies + the thread-free audio/test backends). +- **Audio rendering** — audio tickets run on main-process inline + execution (`worker::InlineDispatcher::sync`; design §3.7 — the crash + risk is dominated by video plugins, which live in oak-worker) through + `eval::render_audio_samples`; the S3 work is migrating them to the + shared-memory transport. - **Borrowed caches** — `oakrender_cache_wrap_borrowed` boxes an opaque marker; queries on borrowed caches return `OAKRENDER_E_INVALID` until the C++ interop layer lands. diff --git a/crates/oakrender/src/autocacher.rs b/crates/oakrender/src/autocacher.rs index d4a91a89e..7a70f283b 100644 --- a/crates/oakrender/src/autocacher.rs +++ b/crates/oakrender/src/autocacher.rs @@ -297,7 +297,7 @@ mod tests { use crate::frame::VideoParamsPod; use crate::texture::{Frame, Texture}; - use crate::worker::WorkerPool; + use crate::worker::{InlineDispatcher, JobDispatch}; fn frame_producer() -> crate::ticket::Producer { Arc::new(|_, _| { @@ -311,16 +311,18 @@ mod tests { }) } - fn new_cacher() -> (PreviewAutoCacher, WorkerPool) { - let mut pool = WorkerPool::new(2); - pool.start(); - let arena = Arc::new(TicketArena::new(Arc::new(pool.clone()), frame_producer())); - (PreviewAutoCacher::new(arena), pool) + /// A cacher on a queued inline dispatcher: jobs run only when the test + /// drains it with `InlineDispatcher::run` (deterministic ordering, no + /// worker threads). + fn new_cacher() -> (PreviewAutoCacher, Arc) { + let d = InlineDispatcher::queued(); + let arena = Arc::new(TicketArena::new(d.clone(), frame_producer())); + (PreviewAutoCacher::new(arena), d) } #[test] fn attach_detach_lifecycle() { - let (mut c, mut pool) = new_cacher(); + let (mut c, d) = new_cacher(); assert!(c.attach(0).is_err(), "zero identity rejected"); c.attach(42).unwrap(); assert_eq!(c.copied_project, 42); @@ -332,31 +334,32 @@ mod tests { c.detach(); assert_eq!(c.copied_project, 0); assert!(c.live_jobs().is_empty()); - pool.shutdown(); + d.shutdown(); } #[test] fn single_frame_cancels_previous() { - // The race this asserts ("the previous frame is cancelled") is only - // deterministic when the first job cannot finish before the - // superseding submit lands — produce frames slowly for this test. - let (mut c, mut pool) = new_cacher_slow(); + // The race this asserts ("the previous frame is cancelled") is + // deterministic on the queued inline dispatcher: the first job stays + // queued until `run`, so the superseding submit's cancel lands first. + let (mut c, d) = new_cacher_slow(); c.attach(7); let first = c.single_frame(Rational::new(0, 1)); let second = c.single_frame(Rational::new(1, 1)); assert_ne!(first, second); // The first ticket is cancelled: its completion fired with State. + d.run(); c.arena.wait(first).unwrap(); let r = c.arena.result(first).unwrap(); assert!(r.is_err()); - pool.shutdown(); + d.shutdown(); } /// A cacher whose frames take ~100ms to produce (see - /// [`single_frame_cancels_previous`]). - fn new_cacher_slow() -> (PreviewAutoCacher, WorkerPool) { - let mut pool = WorkerPool::new(2); - pool.start(); + /// [`single_frame_cancels_previous`] — the sleep is irrelevant on the + /// queued dispatcher, kept for the original async-backend semantics). + fn new_cacher_slow() -> (PreviewAutoCacher, Arc) { + let d = InlineDispatcher::queued(); let producer: crate::ticket::Producer = Arc::new(|_, _| { std::thread::sleep(std::time::Duration::from_millis(100)); let mut f = Frame::new(); @@ -367,41 +370,43 @@ mod tests { f.allocate(); Ok(crate::ticket::TicketPayload::Video(Texture::wrap_frame(f))) }); - let arena = Arc::new(TicketArena::new(Arc::new(pool.clone()), producer)); - (PreviewAutoCacher::new(arena), pool) + let arena = Arc::new(TicketArena::new(d.clone(), producer)); + (PreviewAutoCacher::new(arena), d) } #[test] fn ignore_requests_suppresses_jobs() { - let (mut c, mut pool) = new_cacher(); + let (mut c, d) = new_cacher(); c.attach(7); c.ignore_requests = true; c.on_cache_request(7, TimeRange::new(Rational::new(0, 1), Rational::new(5, 1))); assert!(c.live_jobs().is_empty()); - pool.shutdown(); + d.shutdown(); } #[test] fn renders_paused_queues_but_does_not_start() { - let (mut c, mut pool) = new_cacher(); + let (mut c, d) = new_cacher(); c.attach(7); c.renders_paused = true; c.on_cache_request(7, TimeRange::new(Rational::new(0, 1), Rational::new(5, 1))); assert!(c.live_jobs().is_empty(), "paused: no jobs started"); assert_eq!(c.pending_requests().len(), 1); - pool.shutdown(); + d.shutdown(); } #[test] fn cancel_video_tasks_wait_blocks_until_idle() { - let (mut c, mut pool) = new_cacher(); + let (mut c, d) = new_cacher(); c.attach(7); c.force_range(TimeRange::new(Rational::new(0, 1), Rational::new(1, 1))); assert!(c.is_rendering_custom_range() || c.live_jobs().len() == 1); + // Drain so the wait below observes every job finished. + d.run(); c.cancel_video_tasks(true); assert!(c.live_jobs().iter().all(|id| c.arena.is_finished(*id))); assert!(!c.is_rendering_custom_range()); - pool.shutdown(); + d.shutdown(); } struct Probe { @@ -421,7 +426,7 @@ mod tests { #[test] fn events_deliver_progress() { - let (mut c, mut pool) = new_cacher(); + let (mut c, d) = new_cacher(); let probe = Arc::new(Probe { progress: AtomicU32::new(0), stop: AtomicU32::new(0), @@ -432,7 +437,7 @@ mod tests { c.request_stop_proxy_tasks(); assert_eq!(probe.progress.load(Ordering::Relaxed), 1); assert_eq!(probe.stop.load(Ordering::Relaxed), 1); - pool.shutdown(); + d.shutdown(); } struct ProbeEvents { @@ -450,25 +455,23 @@ mod tests { #[test] fn clear_finished_single_frames_removes_done() { - let (mut c, mut pool) = new_cacher(); + let (mut c, d) = new_cacher(); c.attach(7); let id = c.single_frame(Rational::new(0, 1)); + d.run(); c.arena.wait(id).unwrap(); assert_eq!(c.live_jobs().len(), 1, "finished job still tracked"); c.clear_finished_single_frames(); assert!(c.live_jobs().is_empty()); - pool.shutdown(); + d.shutdown(); } #[test] fn noop_events_sink_never_panics() { let mut c = { - let mut pool = WorkerPool::new(1); - pool.start(); - let arena = Arc::new(TicketArena::new(Arc::new(pool.clone()), frame_producer())); - let c = PreviewAutoCacher::new(arena); - pool.shutdown(); - c + let d = InlineDispatcher::sync(); + let arena = Arc::new(TicketArena::new(d, frame_producer())); + PreviewAutoCacher::new(arena) }; c.set_events(Box::new(NoopEvents)); c.report_progress(1.0); diff --git a/crates/oakrender/src/lib.rs b/crates/oakrender/src/lib.rs index 4c62abf3e..58fd940c7 100644 --- a/crates/oakrender/src/lib.rs +++ b/crates/oakrender/src/lib.rs @@ -25,7 +25,8 @@ //! - `cache` — PlaybackCache/FrameHashCache family + disk state //! - `color` — ColorProcessor over `ocio-rs` + LUT library //! - `ticket` — ticket arena with exactly-once completion -//! - `worker` — worker pool + graph snapshot store +//! - `worker` — the JobDispatch seam + thread-free inline dispatcher + +//! graph snapshot store (the in-process thread pool was deleted in M15 S2) //! - `manager` — RenderManager singleton + disk cache //! - `autocacher` — PreviewAutoCacher //! - `eval` — the evaluation seam (RenderHooks) diff --git a/crates/oakrender/src/manager.rs b/crates/oakrender/src/manager.rs index 2688bba10..79ef3828b 100644 --- a/crates/oakrender/src/manager.rs +++ b/crates/oakrender/src/manager.rs @@ -14,8 +14,8 @@ // You should have received a copy of the GNU General Public License // along with this program. If not, see . -//! The render manager: process-wide singleton owning the worker pool, -//! the ticket arena, the auto-cacher, and backend selection +//! The render manager: process-wide singleton owning the process +//! dispatcher, the ticket arena, the auto-cacher, and backend selection //! (C++ `RenderManager`). //! //! The singleton lives behind a `Mutex>>` so `init` / @@ -31,9 +31,9 @@ use crate::autocacher::PreviewAutoCacher; use crate::backend::BackendKind; use crate::error::{Error, Result}; use crate::eval; -use crate::procpool::{DispatcherConfig, ProcessDispatcher}; +use crate::procpool::{DispatcherConfig, ProcessDispatcher, ShmFrameRef}; use crate::ticket::{TicketArena, TicketId}; -use crate::worker::{JobDispatch, WorkerPool}; +use crate::worker::{InlineDispatcher, JobDispatch}; static MANAGER: Mutex>> = Mutex::new(None); @@ -41,23 +41,26 @@ fn lock(m: &Mutex) -> MutexGuard<'_, T> { m.lock().unwrap_or_else(|e| e.into_inner()) } -/// The render backend the manager initializes (M15 S1: the thread pool -/// and the process-isolated dispatcher coexist; S2 makes Processes the -/// default and removes the pool). +/// The render backend the manager initializes (M15 S2). pub enum RenderBackendChoice { - /// In-process thread pool (the default; C++ parity). + /// In-process thread-free dispatch (**test-only** after M15 S2: the + /// internal render thread pool was deleted by mandate; this backend + /// runs jobs synchronously on the calling thread). Kept so manager / + /// integration tests do not spawn oak-worker children. Threads, /// Process-isolated oak-worker pool (crash isolation + shm frames). + /// The M15 S2 default. Processes(DispatcherConfig), } /// The manager. Created by `oakrender_manager_init` (C ABI), accessed /// internally through [`RenderManager::global`]. pub struct RenderManager { - /// Video job dispatch (thread pool or process dispatcher, M15). + /// Video job dispatch (the process dispatcher, M15). pub dispatch: Arc, - /// Audio job dispatch — kept on main-process threads until S3 - /// (design §3.7: crash risk is dominated by video plugins). + /// Audio job dispatch — kept on main-process inline execution until S3 + /// (design §3.7: crash risk is dominated by video plugins, which live + /// in oak-worker). pub audio_dispatch: Arc, /// Ticket arena. pub tickets: Arc, @@ -73,16 +76,16 @@ pub struct RenderManager { } impl RenderManager { - /// Initialize the process-wide manager with the default backend - /// (in-process threads; idempotent; C++ instance() semantics — only - /// the main GUI process does this). + /// Initialize the process-wide manager with the default backend — the + /// process-isolated oak-worker pool (M15 S2 mandate; idempotent; C++ + /// instance() semantics — only the main GUI process does this). pub fn init() -> Result<()> { - Self::init_with_backend(RenderBackendChoice::Threads) + Self::init_with_backend(RenderBackendChoice::Processes(DispatcherConfig::default())) } - /// Initialize the process-wide manager with an explicit backend - /// (M15 S1: `Threads` keeps the C++ parity path, `Processes` spawns - /// the oak-worker pool). + /// Initialize the process-wide manager with an explicit backend. + /// `Threads` is the test-only inline backend (no worker threads, no + /// child processes); `Processes` spawns the oak-worker pool. pub fn init_with_backend(choice: RenderBackendChoice) -> Result<()> { let mut guard = lock(&MANAGER); if guard.is_some() { @@ -96,18 +99,19 @@ impl RenderManager { let (dispatch, audio_dispatch): (Arc, Arc) = match choice { RenderBackendChoice::Threads => { - let mut pool = WorkerPool::new(0); - pool.start(); - let pool = Arc::new(pool); - (pool.clone(), pool) + // Test-only inline backend: synchronous execution on the + // calling thread, shared by video and audio. + let inline = InlineDispatcher::sync(); + (inline.clone(), inline) } RenderBackendChoice::Processes(config) => { let dispatcher = ProcessDispatcher::new(config)?; dispatcher.start()?; - // Audio stays on main-process threads (design §3.7). - let mut audio = WorkerPool::new(2); - audio.start(); - (dispatcher, Arc::new(audio)) + // Audio stays on main-process inline execution until S3 + // (design §3.7): render_audio_samples runs synchronously + // on the posting thread. + let audio = InlineDispatcher::sync(); + (dispatcher, audio) } }; let tickets = Arc::new(TicketArena::new_with_audio( @@ -147,14 +151,35 @@ impl RenderManager { if let Some(manager) = manager { manager.tickets.cancel_all(); // Drain after the cancels so queued completions fire. Both - // dispatches are idempotent (the Threads backend shares one - // Arc for video + audio). + // dispatches are idempotent. manager.dispatch.shutdown(); manager.audio_dispatch.shutdown(); drop(manager); } } + /// Pump the video backend's control plane (M15 S2): the process + /// dispatcher delivers ticket completions from its poll loop, so the + /// UI tick and any blocking wait must call this regularly. No-op on + /// backends that deliver inline. + pub fn poll(&self) { + self.dispatch.poll(); + } + + /// Release a consumed shm frame's slot back to its worker (M15 S2 + /// zero-copy onscreen path: slot release = cache eviction). No-op on + /// backends that hold no slots. + pub fn release_frame(&self, frame: &ShmFrameRef) { + self.dispatch.release_frame(frame); + } + + /// Cancel every pending AND claimed frame of `sequence` (M15 S2 + /// preview-window invalidation); their completions fire with + /// `Error::State`. No-op on backends that schedule no window. + pub fn cancel_preview_sequence(&self, sequence: u64) { + self.dispatch.cancel_preview_sequence(sequence); + } + /// Aggressive-GC toggle (C++ `SetAggressiveGarbageCollection`). pub fn set_aggressive_gc(&self, on: bool) { self.aggressive_gc.store(on, Ordering::Release); @@ -243,19 +268,22 @@ mod tests { #[test] fn init_shutdown_roundtrip() { let _lock = manager_lock(); - // Ensure a clean slate. + // Ensure a clean slate. The manager tests use the test-only inline + // backend (the process backend spawns real oak-worker children). RenderManager::shutdown(); - RenderManager::init().unwrap(); + RenderManager::init_with_backend(RenderBackendChoice::Threads).unwrap(); assert!(RenderManager::global().is_some()); // Idempotence: second init is a state error. assert_eq!( - RenderManager::init().unwrap_err().code(), + RenderManager::init_with_backend(RenderBackendChoice::Threads) + .unwrap_err() + .code(), Error::State.code() ); RenderManager::shutdown(); assert!(RenderManager::global().is_none()); // Re-init works after shutdown (C++ destroy_instance semantics). - RenderManager::init().unwrap(); + RenderManager::init_with_backend(RenderBackendChoice::Threads).unwrap(); RenderManager::shutdown(); } @@ -263,7 +291,7 @@ mod tests { fn aggressive_gc_toggle() { let _lock = manager_lock(); RenderManager::shutdown(); - RenderManager::init().unwrap(); + RenderManager::init_with_backend(RenderBackendChoice::Threads).unwrap(); let m = RenderManager::global().unwrap(); assert!(!m.aggressive_gc()); m.set_aggressive_gc(true); @@ -275,7 +303,7 @@ mod tests { fn cacher_is_lazily_created() { let _lock = manager_lock(); RenderManager::shutdown(); - RenderManager::init().unwrap(); + RenderManager::init_with_backend(RenderBackendChoice::Threads).unwrap(); let m = RenderManager::global().unwrap(); { let g = m.get_cacher(); diff --git a/crates/oakrender/src/procpool.rs b/crates/oakrender/src/procpool.rs index 0a7b86117..a5345fbe1 100644 --- a/crates/oakrender/src/procpool.rs +++ b/crates/oakrender/src/procpool.rs @@ -49,9 +49,12 @@ //! worker dead: its claimed frames are re-queued to the scheduler //! (any healthy worker may claim them), the child is reaped, the //! segment recreated and the process respawned (bounded restarts). -//! - **Coexistence.** S1 keeps the in-process [`crate::worker::WorkerPool`] -//! alive; [`crate::manager::RenderManager`] picks the backend at -//! init. S2 deletes the thread pool. +//! - **S2 model.** The in-process [`crate::worker::WorkerPool`] is +//! gone (M15 S2 mandate); [`crate::manager::RenderManager`] defaults +//! to this backend. The ticket arena also routes **playback-window** +//! frames here via [`JobSchedule::playback`], and the app pumps the +//! control plane from the UI tick ([`ProcessDispatcher::poll`]) and +//! from blocking ticket waits. use std::collections::{HashMap, HashSet, VecDeque}; use std::io::Write as _; @@ -71,7 +74,7 @@ use crate::ipc::{ WireMontageClip, SLOT_FORMAT_BGRA8, TYPE_BATCH_ACCEPTED, TYPE_ERROR, TYPE_FRAME_FAILED, TYPE_FRAME_READY, TYPE_HANDSHAKE, TYPE_HELLO_CAPS, }; -use crate::scheduler::{FrameKey, FramePriority, FrameRequest, PreviewScheduler}; +use crate::scheduler::{FrameKey, FrameRequest, PreviewScheduler, SubmitOutcome}; use crate::ticket::{Completion, TicketPayload, TicketResult, VideoTicketParams}; use crate::worker::{Job, JobDispatch}; @@ -261,6 +264,45 @@ impl std::fmt::Debug for ShmFrameRef { } } +// --------------------------------------------------------------------------- +// Slot pixel conversions (M15 S2) +// --------------------------------------------------------------------------- +// +// The process backend writes BGRA8 into slots (the viewer preview format). +// Long-lived consumers that need the bytes in a different order/format +// convert once after copying out of the slot. + +/// Convert a BGRA8 block into an RGBA8 block (swapping R and B). Used by +/// PNG writers (footage thumbnails) and PPM/CLI output, which require +/// RGB-order buffers. `src.len()` must be a multiple of 4. +pub fn bgra8_to_rgba8(src: &[u8]) -> Vec { + let mut out = Vec::with_capacity(src.len()); + for px in src.chunks_exact(4) { + out.push(px[2]); // R + out.push(px[1]); // G + out.push(px[0]); // B + out.push(px[3]); // A + } + out +} + +/// Convert a BGRA8 block into tightly-packed F32 RGBA samples (`0..=1`). +/// Used by the export/encoder path, which declares F32 input: the worker +/// converts its F32 pipeline output to BGRA8 for the slot (design §3.1), +/// and the export converts back — a necessary conversion at the encoder +/// boundary with 8-bit quantization (S2; per-ticket slot formats are S3 +/// work). +pub fn bgra8_to_f32_rgba(src: &[u8]) -> Vec { + let mut out = Vec::with_capacity(src.len() / 4 * 4); + for px in src.chunks_exact(4) { + out.push(f32::from(px[2]) / 255.0); // R + out.push(f32::from(px[1]) / 255.0); // G + out.push(f32::from(px[0]) / 255.0); // B + out.push(f32::from(px[3]) / 255.0); // A + } + out +} + // --------------------------------------------------------------------------- // Configuration // --------------------------------------------------------------------------- @@ -1128,10 +1170,15 @@ impl ProcessDispatcher { impl JobDispatch for ProcessDispatcher { /// Submit one frame job (the ticket-arena seam). The job joins the - /// scheduler as a Seek-priority request and is dispatched on the - /// next pump; the completion fires with + /// scheduler under its [`JobSchedule`] (Seek single-frame by default, + /// Playback for the pre-render window, Background for exports) and is + /// dispatched on the next pump; the completion fires with /// `TicketPayload::ShmFrame(ShmFrameRef)` — never a pixel buffer. + /// Re-submitting a key that is still pending replaces the old request + /// and cancels its ticket; a key already in flight is left running + /// (its result is still valid for the same params). fn post(&self, job: Job) -> bool { + let mut fired: Vec<(Completion, TicketResult)> = Vec::new(); { let mut inner = lock(&self.inner); if inner.shutting_down { @@ -1139,10 +1186,11 @@ impl JobDispatch for ProcessDispatcher { } let id = inner.next_ticket; inner.next_ticket += 1; + let frame = job.schedule.frame.unwrap_or(id); let key = FrameKey { sequence: job.node_identity, - frame: id, - version: 0, + frame, + version: job.schedule.version, }; inner.tickets.insert( id, @@ -1152,18 +1200,73 @@ impl JobDispatch for ProcessDispatcher { done: Some(job.done), }, ); - inner.scheduler.submit(FrameRequest { + let request = FrameRequest { key, - priority: FramePriority::Seek, - distance: 0, + priority: job.schedule.priority, + distance: job.schedule.distance, payload: id, - }); + }; + match inner.scheduler.submit(request) { + SubmitOutcome::Accepted => {} + SubmitOutcome::Replaced(old) => { + // A newer request for the same key superseded the old + // pending one: cancel the old ticket's completion. + if let Some(pt) = inner.tickets.get_mut(&old.payload) { + if let Some(done) = pt.done.take() { + fired.push((done, Err(Error::State))); + } + } + } + SubmitOutcome::InFlight => { + // Already claimed by a worker; the rendered frame is + // still valid for the same params (playback window + // slides re-request frames that are in flight). + } + } + } + for (done, result) in fired { + done(result); } // Pump once so a live worker picks the frame up immediately. self.poll(); true } + /// Cancel every pending AND claimed request of `sequence` (M15 S2 + /// preview-window invalidation — graph/proxy/resolution/color bump or + /// a sequence switch). Dropped completions fire `Error::State`; + /// frames already dispatched recycle their slots when the late + /// `frame_ready` arrives. + fn cancel_preview_sequence(&self, sequence: u64) { + let mut fired: Vec<(Completion, TicketResult)> = Vec::new(); + { + let mut inner = lock(&self.inner); + let dropped = inner.scheduler.cancel_sequence(sequence); + for request in dropped { + if let Some(pt) = inner.tickets.get_mut(&request.payload) { + if let Some(done) = pt.done.take() { + fired.push((done, Err(Error::State))); + } + } + } + } + for (done, result) in fired { + done(result); + } + } + + /// Pump the control plane (delegates to the inherent poll — the UI + /// tick and blocking ticket waits call this through the trait seam). + fn poll(&self) { + self.poll(); + } + + /// Release a consumed frame's slot (delegates to the inherent + /// release — see [`ProcessDispatcher::release_frame`]). + fn release_frame(&self, frame: &ShmFrameRef) { + self.release_frame(frame); + } + /// Graceful shutdown: `shutdown` messages, a short drain pumping /// completions, then kill stragglers; every ticket still open /// completes with `Error::State`. diff --git a/crates/oakrender/src/scheduler.rs b/crates/oakrender/src/scheduler.rs index e51763aa4..5d4423157 100644 --- a/crates/oakrender/src/scheduler.rs +++ b/crates/oakrender/src/scheduler.rs @@ -62,9 +62,10 @@ pub struct FrameKey { } /// Frame request priority class (lower value = more urgent). -#[derive(Clone, Copy, Debug, PartialEq, Eq, PartialOrd, Ord)] +#[derive(Clone, Copy, Debug, PartialEq, Eq, PartialOrd, Ord, Default)] pub enum FramePriority { /// Seek / current playhead frame (single-frame insert, top priority). + #[default] Seek, /// Playback window frames (ordered by [`FrameRequest::distance`]). Playback, @@ -112,6 +113,19 @@ struct PendingEntry

{ any_worker: bool, } +/// The outcome of [`PreviewScheduler::submit`]. +#[derive(Debug)] +pub enum SubmitOutcome

{ + /// A new request was accepted (nothing was pending under its key). + Accepted, + /// An existing pending request under the same key was replaced; the + /// superseded request is returned so the dispatcher can cancel its + /// ticket completion (M15 S2). + Replaced(FrameRequest

), + /// The key is already claimed (in flight); the request was rejected. + InFlight, +} + /// The scheduler state machine (single-threaded by contract). pub struct PreviewScheduler

{ workers: usize, @@ -155,21 +169,25 @@ impl PreviewScheduler

{ /// Submit a frame request. An already-pending request with the same /// key is replaced; a key already claimed (in flight) is rejected — - /// the dispatcher must cancel/re-version it first. Returns true when - /// the request was accepted. - pub fn submit(&mut self, request: FrameRequest

) -> bool { + /// the dispatcher must cancel/re-version it first. + pub fn submit(&mut self, request: FrameRequest

) -> SubmitOutcome

{ if self.claimed.contains_key(&request.key) { - return false; + return SubmitOutcome::InFlight; } if let Some(entry) = self.pending.iter_mut().find(|e| e.request.key == request.key) { - entry.request = request; - return true; + let old = std::mem::replace(&mut entry.request, request); + return SubmitOutcome::Replaced(old); } self.pending.push(PendingEntry { request, any_worker: false, }); - true + SubmitOutcome::Accepted + } + + /// Whether `key` is currently claimed (in flight). + pub fn is_claimed(&self, key: &FrameKey) -> bool { + self.claimed.contains_key(key) } /// Claim the next batch for `worker`: the worker's interleaved shard @@ -300,14 +318,27 @@ impl PreviewScheduler

{ /// Cancel every pending AND claimed frame of `sequence` (frame-key /// invalidation; the `cancel` wire message covers the worker side). - /// Returns the number of dropped requests. - pub fn cancel_sequence(&mut self, sequence: u64) -> usize { - let before_pending = self.pending.len(); - self.pending.retain(|e| e.request.key.sequence != sequence); - let dropped_pending = before_pending - self.pending.len(); - let before_claimed = self.claimed.len(); - self.claimed.retain(|_, c| c.request.key.sequence != sequence); - dropped_pending + (before_claimed - self.claimed.len()) + /// Returns the dropped requests so the dispatcher can fire their + /// completions with `Error::State`. + pub fn cancel_sequence(&mut self, sequence: u64) -> Vec> { + let mut dropped = Vec::new(); + self.pending.retain(|e| { + if e.request.key.sequence == sequence { + dropped.push(e.request.clone()); + false + } else { + true + } + }); + self.claimed.retain(|_, c| { + if c.request.key.sequence == sequence { + dropped.push(c.request.clone()); + false + } else { + true + } + }); + dropped } /// Pending (unclaimed) request count. @@ -373,7 +404,7 @@ mod tests { fn no_stealing_every_frame_claimed_exactly_once() { let mut s: PreviewScheduler = PreviewScheduler::new(3, 4); for f in 0..40 { - assert!(s.submit(req(1, f, FramePriority::Playback))); + assert!(matches!(s.submit(req(1, f, FramePriority::Playback)), SubmitOutcome::Accepted)); } let claims = claim_all(&mut s); assert_eq!(claims.len(), 40, "every frame claimed"); @@ -527,14 +558,42 @@ mod tests { let mut s: PreviewScheduler = PreviewScheduler::new(1, 4); let r = req(1, 5, FramePriority::Playback); let key = r.key; - s.submit(r); + assert!(matches!(s.submit(r), SubmitOutcome::Accepted)); let _ = s.claim_batch(0, 4).unwrap(); // In flight: rejected. - assert!(!s.submit(req(1, 5, FramePriority::Seek))); + assert!(matches!( + s.submit(req(1, 5, FramePriority::Seek)), + SubmitOutcome::InFlight + )); s.frame_done(&key); // After completion the same key may be requested again (new // version in practice). - assert!(s.submit(req(1, 5, FramePriority::Seek))); + assert!(matches!( + s.submit(req(1, 5, FramePriority::Seek)), + SubmitOutcome::Accepted + )); + } + + #[test] + fn resubmit_replaces_pending_and_returns_the_old_request() { + let mut s: PreviewScheduler = PreviewScheduler::new(1, 4); + let first = req(1, 5, FramePriority::Playback); + let key = first.key; + assert!(matches!(s.submit(first), SubmitOutcome::Accepted)); + // A second submission under the same key replaces the pending entry + // and hands the superseded request back (the dispatcher cancels its + // ticket completion with Error::State). + let second = req(1, 5, FramePriority::Seek); + let second_key = second.key; + match s.submit(second) { + SubmitOutcome::Replaced(old) => { + assert_eq!(old.key, key); + assert_eq!(old.priority, FramePriority::Playback); + } + other => panic!("expected Replaced, got {other:?}"), + } + let _ = s.claim_batch(0, 4).unwrap(); + assert_eq!(s.claimed_worker(&second_key), Some(0)); } #[test] @@ -545,7 +604,8 @@ mod tests { } let _ = s.claim_batch(0, 4).unwrap(); // claims 4 of sequence 7 s.submit(req(8, 0, FramePriority::Playback)); // other sequence - assert_eq!(s.cancel_sequence(7), 6); + let dropped = s.cancel_sequence(7); + assert_eq!(dropped.len(), 6, "all 6 sequence-7 requests dropped"); assert_eq!(s.pending_count(), 1, "sequence 8 untouched"); assert_eq!(s.claimed_count(), 0); } diff --git a/crates/oakrender/src/ticket.rs b/crates/oakrender/src/ticket.rs index 6a74f94a4..2af3b39ac 100644 --- a/crates/oakrender/src/ticket.rs +++ b/crates/oakrender/src/ticket.rs @@ -33,7 +33,7 @@ use oakcore_rs::{Rational, TimeRange}; use crate::error::{Error, Result}; use crate::eval; use crate::texture::Texture; -use crate::worker::JobDispatch; +use crate::worker::{JobDispatch, JobSchedule}; /// One clip of a sequence montage (M12 P0): the facade resolves the /// timeline into an ordered list of clips; the producer decodes each and @@ -194,11 +194,21 @@ struct TicketSlot { delivered: AtomicBool, completion: Mutex>, result: Mutex>>, + /// The dispatch the ticket posted through (M15 S2: a cancelled ticket + /// whose render completed must recycle its shm slot — the consumer + /// never sees the `ShmFrame` payload). + dispatch: Arc, } impl TicketSlot { fn finish(&self, mut result: TicketResult) { if self.cancel.load(Ordering::Acquire) { + // A cancelled ticket still holds its shm slot when the render + // completed: recycle it now, before the payload is replaced by + // `Error::State` and the consumer loses it. + if let Ok(TicketPayload::ShmFrame(frame)) = &result { + self.dispatch.release_frame(frame); + } result = Err(Error::State); } // Publish the result before flipping the state flag: `wait()` only @@ -296,12 +306,27 @@ impl TicketArena { /// Submit a video ticket with a caller-reserved id (allocated by /// [`TicketArena::next_id`]); completion fires exactly once, including - /// on cancellation (with `Error::State`). + /// on cancellation (with `Error::State`). The job posts as a Seek + /// single-frame request (M15 S2; see + /// [`TicketArena::submit_playback`] for the pre-render window). pub fn submit_video_with_id( &self, id: TicketId, params: VideoTicketParams, done: Completion, + ) -> TicketId { + self.submit_video_impl(id, params, done, JobSchedule::seek()) + } + + /// The shared video-ticket submit path: register the slot, post the + /// job with `schedule` (M15 S2), and deliver `Error::State` + /// immediately when the backend is gone. + fn submit_video_impl( + &self, + id: TicketId, + params: VideoTicketParams, + done: Completion, + schedule: JobSchedule, ) -> TicketId { let meta = TicketMeta { kind: Some(ticket_kind::VIDEO), @@ -322,6 +347,7 @@ impl TicketArena { delivered: AtomicBool::new(false), completion: Mutex::new(Some(done)), result: Mutex::new(None), + dispatch: self.dispatch.clone(), }); self.allocate(slot.clone()); @@ -334,6 +360,7 @@ impl TicketArena { params, produce: producer, done: Box::new(move |result| slot_done.finish(result)), + schedule, }; if !self.dispatch.post(job) { // Backend is gone (shutdown raced the submit): deliver now. @@ -342,6 +369,51 @@ impl TicketArena { id } + /// Submit a playback-window frame (M15 S2 pre-render window): the + /// frame joins the scheduler at Playback priority, ordered by + /// `distance` from the playhead and keyed under `version`. `frame` is + /// the sequence frame number (the scheduler key / interleaved shard). + /// The completion fires (from the dispatcher's poll) with + /// `TicketPayload::ShmFrame`. + pub fn submit_playback( + &self, + params: VideoTicketParams, + frame: i64, + distance: i64, + version: u64, + done: Completion, + ) -> TicketId { + let id = self.next_id(); + self.submit_video_impl( + id, + params, + done, + JobSchedule::playback(frame, distance, version), + ) + } + + /// Submit a Background-priority frame (M15 S2 exports/precache): the + /// scheduler renders it whenever no Seek/Playback work is pending. + pub fn submit_video_background( + &self, + params: VideoTicketParams, + done: Completion, + ) -> TicketId { + let id = self.next_id(); + self.submit_video_background_with_id(id, params, done) + } + + /// [`TicketArena::submit_video_background`] with a caller-reserved id + /// (the export loop pre-allocates ids through [`TicketArena::next_id`]). + pub fn submit_video_background_with_id( + &self, + id: TicketId, + params: VideoTicketParams, + done: Completion, + ) -> TicketId { + self.submit_video_impl(id, params, done, JobSchedule::background()) + } + /// Submit a video ticket; completion fires exactly once, including /// on cancellation (with `Error::State`). pub fn submit_video(&self, params: VideoTicketParams, done: Completion) -> TicketId { @@ -376,6 +448,7 @@ impl TicketArena { delivered: AtomicBool::new(false), completion: Mutex::new(Some(done)), result: Mutex::new(None), + dispatch: self.audio_dispatch.clone(), }); self.allocate(slot.clone()); @@ -403,6 +476,7 @@ impl TicketArena { }), produce: producer, done: Box::new(move |result| slot_done.finish(result)), + schedule: JobSchedule::seek(), }; if !self.audio_dispatch.post(job) { slot.finish(Err(Error::State)); @@ -425,12 +499,28 @@ impl TicketArena { } } - /// Blocking wait for completion (C++ wait_for_finished). + /// Blocking wait for completion (C++ wait_for_finished). M15 S2: the + /// process dispatcher delivers completions from its poll loop, so a + /// blocking wait pumps it (the UI tick may be blocked right here). pub fn wait(&self, id: TicketId) -> Result<()> { let slot = lock(&self.slots).get(&id).cloned().ok_or(Error::NotFound)?; let mut state = lock(&slot.state); while !matches!(*state, SlotState::Finished) { - state = slot.cv.wait(state).unwrap_or_else(|e| e.into_inner()); + // Pump the backend before blocking again. The slot lock is + // dropped first: poll() delivers completions that finish this + // very slot (finish() takes the same lock). + drop(state); + self.dispatch.poll(); + state = lock(&slot.state); + if matches!(*state, SlotState::Finished) { + break; + } + let timeout = std::time::Duration::from_millis(5); + let (g, _) = slot + .cv + .wait_timeout(state, timeout) + .unwrap_or_else(|e| e.into_inner()); + state = g; } Ok(()) } @@ -470,8 +560,8 @@ impl TicketArena { } /// Cancel all pending tickets (manager shutdown path). Delivery happens - /// when the pool drains the queued jobs (or the running jobs finish); - /// call [`WorkerPool::shutdown`] afterwards to guarantee all + /// when the backend drains the queued jobs (or the running jobs + /// finish); call the backend's `shutdown` afterwards to guarantee all /// completions have fired. pub fn cancel_all(&self) { self.shutting_down.store(true, Ordering::Release); @@ -494,7 +584,7 @@ mod tests { use crate::frame::VideoParamsPod; use crate::texture::Frame; - use crate::worker::WorkerPool; + use crate::worker::{InlineDispatcher, JobDispatch}; fn small_frame() -> Frame { let mut f = Frame::new(); @@ -510,12 +600,19 @@ mod tests { Arc::new(|_, _| Ok(TicketPayload::Video(Texture::wrap_frame(small_frame())))) } + /// A queued inline dispatcher for video (deterministic cancel-race and + /// shutdown semantics) plus a sync inline dispatcher for audio (the + /// production audio mode). `run` drains the video queue. + fn test_arena(producer: Producer) -> (TicketArena, Arc) { + let video = InlineDispatcher::queued(); + let audio = InlineDispatcher::sync(); + let arena = TicketArena::new_with_audio(video.clone(), audio, producer); + (arena, video) + } + #[test] fn completion_fires_exactly_once_on_success() { - let pool = WorkerPool::new(2); - let mut pool = pool; - pool.start(); - let arena = TicketArena::new(Arc::new(pool.clone()), ok_producer()); + let (arena, video) = test_arena(ok_producer()); let (tx, rx) = mpsc::channel(); let id = arena.submit_video( @@ -536,6 +633,7 @@ mod tests { }), ); + video.run(); arena.wait(id).unwrap(); let ok = rx.recv_timeout(Duration::from_secs(5)).unwrap(); assert!(ok, "completion fired with success"); @@ -553,24 +651,16 @@ mod tests { (4, 4) ); assert!(arena.is_finished(id)); - pool.shutdown(); + video.shutdown(); } #[test] fn completion_fires_exactly_once_on_cancel() { - let pool = WorkerPool::new(1); - let mut pool = pool; - pool.start(); - // Producer blocks until cancelled: exercises the cancel race. - let release = Arc::new(std::sync::atomic::AtomicBool::new(false)); - let release2 = release.clone(); - let producer: Producer = Arc::new(move |_, _| { - while !release2.load(Ordering::Acquire) { - std::thread::sleep(Duration::from_millis(1)); - } - Ok(TicketPayload::Video(Texture::wrap_frame(small_frame()))) - }); - let arena = TicketArena::new(Arc::new(pool.clone()), producer); + // Queued mode: the job is posted but not run yet, so a cancel + // before `run` must deliver State exactly once (the old worker-pool + // test used a blocking producer; the inline queue makes the same + // cancel-before-completion race deterministic without threads). + let (arena, video) = test_arena(ok_producer()); let (tx, rx) = mpsc::channel(); let id = arena.submit_video( @@ -591,9 +681,9 @@ mod tests { }), ); - // Cancel while the job is running, then release the job. + // Cancel before the job runs. arena.cancel(id); - release.store(true, Ordering::Release); + video.run(); arena.wait(id).unwrap(); let res = rx.recv_timeout(Duration::from_secs(5)).unwrap(); @@ -602,39 +692,32 @@ mod tests { rx.recv_timeout(Duration::from_millis(50)).is_err(), "exactly once" ); - pool.shutdown(); + video.shutdown(); } #[test] fn cancel_of_unknown_id_is_ignored() { - let pool = WorkerPool::new(1); - let mut pool = pool; - pool.start(); - let arena = TicketArena::new(Arc::new(pool.clone()), ok_producer()); + let (arena, video) = test_arena(ok_producer()); arena.cancel(TicketId(12345)); assert!(!arena.is_finished(TicketId(12345))); - pool.shutdown(); + video.shutdown(); } #[test] fn wait_unknown_id_errors() { - let pool = WorkerPool::new(1); - let mut pool = pool; - pool.start(); - let arena = TicketArena::new(Arc::new(pool.clone()), ok_producer()); + let (arena, video) = test_arena(ok_producer()); assert_eq!( arena.wait(TicketId(999)).unwrap_err().code(), Error::NotFound.code() ); - pool.shutdown(); + video.shutdown(); } #[test] fn audio_ticket_meta_and_kind() { - let pool = WorkerPool::new(1); - let mut pool = pool; - pool.start(); - let arena = TicketArena::new(Arc::new(pool.clone()), ok_producer()); + // Audio runs on the sync inline backend (the production audio + // mode), so the completion fires during the submit. + let (arena, video) = test_arena(ok_producer()); let (tx, rx) = mpsc::channel(); let range = TimeRange::new(Rational::new(0, 1), Rational::new(10, 1)); @@ -659,15 +742,12 @@ mod tests { assert!(!rx.recv_timeout(Duration::from_secs(5)).unwrap()); // get_time equivalent: audio tickets report range.in. assert_eq!(arena.time(id), Some(range.in_())); - pool.shutdown(); + video.shutdown(); } #[test] fn ticket_ids_are_monotonic() { - let pool = WorkerPool::new(1); - let mut pool = pool; - pool.start(); - let arena = TicketArena::new(Arc::new(pool.clone()), ok_producer()); + let (arena, video) = test_arena(ok_producer()); let a = arena.submit_video( VideoTicketParams { viewer: 1, @@ -700,6 +780,6 @@ mod tests { ); assert!(b.0 > a.0); assert_ne!(a, b); - pool.shutdown(); + video.shutdown(); } } diff --git a/crates/oakrender/src/worker.rs b/crates/oakrender/src/worker.rs index c1176d0c5..992560c71 100644 --- a/crates/oakrender/src/worker.rs +++ b/crates/oakrender/src/worker.rs @@ -14,28 +14,41 @@ // You should have received a copy of the GNU General Public License // along with this program. If not, see . -//! The worker layer (C++ RenderWorkerPool + RenderThread + -//! workerprocess/workerjson): thread pool AND process-isolated pool -//! behind one dispatch seam. +//! The job-dispatch seam (M15 S2): what a render ticket posts through. //! -//! This pass ships the in-process [`WorkerPool`] fully. The -//! process-isolated backend landed in M15 S1 as -//! [`crate::procpool::ProcessDispatcher`] (spawn/handshake/crash-restart -//! of oak-worker binaries over NDJSON + shared memory); both backends -//! implement the [`JobDispatch`] seam the ticket arena posts through. -//! [`ProcessPool`] below is the frozen pre-M15 facade stub kept for C -//! ABI parity. +//! The in-process thread pool was **removed** in M15 S2 (user mandate: +//! "delete the internal render thread pool"); the only video backend is +//! the process-isolated [`crate::procpool::ProcessDispatcher`] +//! (oak-worker children over NDJSON + shared memory). This module keeps +//! the ticket-facing surface: +//! +//! - [`Job`] — one unit of render work plus its scheduler hints. +//! - [`JobDispatch`] — the backend seam the arena posts through, with +//! default no-ops for the process-backend extras (poll / release / +//! preview-window cancellation). +//! - [`InlineDispatcher`] — a thread-free dispatcher that executes jobs +//! on the calling thread. Used as the **audio** backend (audio stays on +//! main-process inline execution until S3 — design §3.7) and by the +//! manager's test-only `Threads` backend and by unit tests. +//! - [`GraphSnapshotStore`] — the graph-snapshot file refcounting cache +//! shared with worker processes. use std::collections::{HashMap, VecDeque}; use std::panic::{catch_unwind, AssertUnwindSafe}; use std::sync::atomic::{AtomicBool, Ordering}; -use std::sync::{Arc, Condvar, Mutex, MutexGuard}; +use std::sync::{Arc, Mutex, MutexGuard}; use oakcore_rs::Rational; use crate::error::{Error, Result}; +use crate::procpool::ShmFrameRef; +use crate::scheduler::FramePriority; use crate::ticket::{Completion, Producer, VideoTicketParams}; +fn lock(m: &Mutex) -> MutexGuard<'_, T> { + m.lock().unwrap_or_else(|e| e.into_inner()) +} + /// A unit of render work (produced by the ticket arena). pub struct Job { /// The graph position this job evaluates. @@ -44,21 +57,63 @@ pub struct Job { pub time: Rational, /// Ticket parameters (size/format overrides). pub params: Arc, - /// Frame producer (arena-installed). + /// Frame producer (arena-installed; the process backend never invokes + /// it — workers render from the wire spec). pub produce: Producer, /// Completion delivery. pub done: Completion, + /// Scheduler hints (M15 S2). Defaults to a Seek single-frame request. + pub schedule: JobSchedule, } -fn lock(m: &Mutex) -> MutexGuard<'_, T> { - m.lock().unwrap_or_else(|e| e.into_inner()) +/// Scheduler hints a posted job carries (M15 S2). The process dispatcher +/// maps these onto [`crate::scheduler::FrameKey`] / priority; the inline +/// dispatcher ignores them. +#[derive(Clone, Debug, Default)] +pub struct JobSchedule { + /// Priority class. Default [`FramePriority::Seek`] (single-frame). + pub priority: FramePriority, + /// Scheduler key frame number. `None` = the ticket id (the Seek + /// single-frame convention). + pub frame: Option, + /// Playhead distance (orders the Playback class). + pub distance: i64, + /// Parameter version (graph/proxy/resolution/color); bumping it + /// invalidates stale requests for the same sequence+frame. + pub version: u64, +} + +impl JobSchedule { + /// A Seek single-frame request (the default for every ticket). + pub fn seek() -> Self { + Self::default() + } + + /// A Background request (exports / precache): rendered whenever the + /// workers have no Seek/Playback work. + pub fn background() -> Self { + Self { + priority: FramePriority::Background, + ..Default::default() + } + } + + /// A Playback-window request at `frame`, ordered by `distance` from + /// the playhead and keyed under `version`. + pub fn playback(frame: i64, distance: i64, version: u64) -> Self { + Self { + priority: FramePriority::Playback, + frame: Some(frame), + distance, + version, + } + } } /// The job-dispatch seam (M15 S1): the ticket arena posts [`Job`]s -/// through this interface without knowing the backend. Implemented by -/// the in-process [`WorkerPool`] (threads) and the process-isolated -/// [`crate::procpool::ProcessDispatcher`] (oak-worker children); S2 -/// removes the thread pool and this seam becomes process-only. +/// through this interface without knowing the backend. Implemented by the +/// process-isolated [`crate::procpool::ProcessDispatcher`] (oak-worker +/// children) and the thread-free [`InlineDispatcher`] (audio / tests). pub trait JobDispatch: Send + Sync { /// Enqueue a job; false when the backend is gone (the arena then /// delivers the completion itself with `Error::State`). @@ -67,197 +122,127 @@ pub trait JobDispatch: Send + Sync { /// Stop accepting work, deliver the queued completions (cancelled) /// and release the backend. Idempotent. fn shutdown(&self); + + /// Pump backend completions (the process dispatcher's poll loop). + /// Default no-op: backends that deliver inline have nothing to pump. + /// The UI tick and blocking ticket waits call this so the process + /// backend's completions are delivered without a dedicated thread. + fn poll(&self) {} + + /// Release a consumed shm frame's slot back to its worker (slot + /// release = cache eviction, design §3.1). Default no-op: only the + /// process backend holds slots. + fn release_frame(&self, _frame: &ShmFrameRef) {} + + /// Cancel every pending AND claimed request of `sequence` (M15 S2 + /// preview-window invalidation); their completions fire with + /// `Error::State`. Default no-op: only the process backend schedules. + fn cancel_preview_sequence(&self, _sequence: u64) {} } -/// Thread-pool backend (C++ RenderThread model). Cheap to clone (all -/// state is behind an `Arc`); the manager and the ticket arena share one -/// pool. -#[derive(Clone)] -pub struct WorkerPool { - inner: Arc, +/// Thread-free job dispatcher (M15 S2). Executes jobs on the calling +/// thread — there are deliberately **no worker threads**: +/// +/// - **Sync mode** ([`InlineDispatcher::sync`]): every `post` runs its +/// job immediately on the caller's thread. This is the production +/// **audio** backend (audio stays on main-process inline execution +/// until S3 — design §3.7: the crash risk is dominated by video +/// plugins, which already live in oak-worker) and the manager's +/// test-only `Threads` backend. +/// - **Queued mode** ([`InlineDispatcher::queued`]): `post` queues the +/// job; the test drains it with [`InlineDispatcher::run`]. This keeps +/// the arena's cancel-race and shutdown semantics deterministic +/// without any threads. +pub struct InlineDispatcher { + inner: Arc, } -struct PoolInner { - workers: usize, +struct InlineInner { queue: Mutex>, - cv: Condvar, + sync: bool, stopping: AtomicBool, - threads: Mutex>>, } -impl WorkerPool { - /// Pool with `workers` threads (0 = hardware concurrency). - pub fn new(workers: usize) -> Self { - let workers = if workers == 0 { - std::thread::available_parallelism() - .map(|n| n.get()) - .unwrap_or(1) - } else { - workers - }; - Self { - inner: Arc::new(PoolInner { - workers, +impl InlineDispatcher { + /// A sync-mode dispatcher: jobs run immediately on the posting thread. + pub fn sync() -> Arc { + Arc::new(Self { + inner: Arc::new(InlineInner { queue: Mutex::new(VecDeque::new()), - cv: Condvar::new(), + sync: true, stopping: AtomicBool::new(false), - threads: Mutex::new(Vec::new()), }), - } + }) } - /// The number of worker threads. - pub fn worker_count(&self) -> usize { - self.inner.workers + /// A queued-mode dispatcher: jobs wait for [`InlineDispatcher::run`]. + pub fn queued() -> Arc { + Arc::new(Self { + inner: Arc::new(InlineInner { + queue: Mutex::new(VecDeque::new()), + sync: false, + stopping: AtomicBool::new(false), + }), + }) } - /// Start threads (idempotent). - pub fn start(&mut self) { - let mut threads = lock(&self.inner.threads); - if !threads.is_empty() { + /// Run every queued job synchronously on the calling thread (queued + /// mode). Jobs posted after `shutdown` are refused by `post`. + pub fn run(&self) { + if self.inner.stopping.load(Ordering::Acquire) { return; } - for _ in 0..self.inner.workers { - let inner = self.inner.clone(); - let handle = std::thread::spawn(move || worker_loop(inner)); - threads.push(handle); + loop { + let job = lock(&self.inner.queue).pop_front(); + let Some(job) = job else { break }; + execute_job(job); } } - /// True when threads are running. - pub fn is_running(&self) -> bool { - !lock(&self.inner.threads).is_empty() + /// The number of queued (not yet run) jobs. + pub fn queued_count(&self) -> usize { + lock(&self.inner.queue).len() } +} - /// Enqueue a job. Returns false when the pool is shut down. - pub fn post(&self, job: Job) -> bool { - // The stopping check and the push share one queue lock: a shutdown - // racing the check would otherwise leave the job queued after every - // worker exited (and after the defensive drain), so its completion - // could never fire. +fn execute_job(job: Job) { + let result = catch_unwind(AssertUnwindSafe(|| (job.produce)(job.time, &job.params))) + .unwrap_or_else(|_| Err(Error::Failed("frame producer panicked".into()))); + (job.done)(result); +} + +impl JobDispatch for InlineDispatcher { + fn post(&self, job: Job) -> bool { + if self.inner.sync { + // Sync mode: run now (refusing only when shutting down). + if self.inner.stopping.load(Ordering::Acquire) { + return false; + } + execute_job(job); + return true; + } let mut queue = lock(&self.inner.queue); if self.inner.stopping.load(Ordering::Acquire) { return false; } queue.push_back(job); - self.inner.cv.notify_one(); true } - /// Stop accepting, drain, join all workers. In-flight job completions - /// fire with cancellation (queued jobs are delivered `Error::State` - /// without running); running jobs are joined so no completion fires - /// after shutdown returns. - pub fn shutdown(&mut self) { - self.shutdown_ref(); - } - - /// [`Self::shutdown`] on a shared reference (the [`JobDispatch`] - /// seam; all state is interior-mutable). Idempotent. - pub fn shutdown_ref(&self) { - // Set the flag and wake the workers while holding the queue lock. - // Workers decide whether to block in `cv.wait` while holding that - // lock, so a flag set outside it could land between a worker's - // predicate check and its wait: the wakeup is lost, the worker - // sleeps forever, and the join below hangs. Serializing store + - // notify with the waiters' lock closes that window. - { - let _guard = lock(&self.inner.queue); - self.inner.stopping.store(true, Ordering::Release); - self.inner.cv.notify_all(); - } - - let threads = std::mem::take(&mut *lock(&self.inner.threads)); - for handle in threads { - let _ = handle.join(); - } - - // Defensive drain: any job that landed between `stopping` and the - // workers' exit (post() refuses them, so this is normally empty). - let mut queue = lock(&self.inner.queue); - while let Some(job) = queue.pop_front() { - deliver_cancelled(job); - } - } -} - -impl JobDispatch for WorkerPool { - fn post(&self, job: Job) -> bool { - WorkerPool::post(self, job) - } - fn shutdown(&self) { - self.shutdown_ref(); - } -} - -fn worker_loop(inner: Arc) { - loop { - let job = { - let mut queue = lock(&inner.queue); - while !inner.stopping.load(Ordering::Acquire) && queue.is_empty() { - queue = inner.cv.wait(queue).unwrap_or_else(|e| e.into_inner()); - } - queue.pop_front() + // Stop accepting and drain the queue with cancellation (queued + // jobs never run after shutdown). + let jobs: Vec = { + let mut queue = lock(&self.inner.queue); + self.inner.stopping.store(true, Ordering::Release); + queue.drain(..).collect() }; - let Some(job) = job else { - return; // stopping and queue drained - }; - if inner.stopping.load(Ordering::Acquire) { - // Shutdown raced this pop: deliver cancellation. - deliver_cancelled(job); - continue; + for job in jobs { + (job.done)(Err(Error::State)); } - let result = catch_unwind(AssertUnwindSafe(|| (job.produce)(job.time, &job.params))) - .unwrap_or_else(|_| Err(Error::Failed("frame producer panicked".into()))); - (job.done)(result); } } -fn deliver_cancelled(job: Job) { - (job.done)(Err(Error::State)); -} - -/// Process-isolated worker backend (C++ RenderWorkerPool + -/// PooledWorker). Child processes talk the oakengine_ipc C ABI; this side -/// is only a client (spawn, dispatch, reap). Not wired in this pass. -pub struct ProcessPool { - workers: usize, -} - -impl ProcessPool { - /// Pool of `workers` child processes. - pub fn new(workers: usize) -> Self { - Self { workers } - } - - /// The configured child count. - pub fn worker_count(&self) -> usize { - self.workers - } - - /// Spawn children and handshake. - pub fn start(&mut self) -> Result<()> { - Err(Error::Failed( - "oakengine_ipc worker-process bridge not implemented in this pass".into(), - )) - } - - /// Dispatch a job to a free child. - pub fn post(&self, _job: Job) -> Result<()> { - Err(Error::Failed( - "oakengine_ipc worker-process bridge not implemented in this pass".into(), - )) - } - - /// Cancel the job running in a child (C++ cancel_active_process). - pub fn cancel_active(&self, _process_slot: usize) {} - - /// Terminate and reap all children; pending jobs complete with - /// cancellation. - pub fn shutdown(&mut self) {} -} - /// Graph snapshot files shared with worker processes (C++ /// write_graph_snapshot + path refcounting): a snapshot is written once /// and reference-counted; the file is unlinked at zero. @@ -355,27 +340,20 @@ impl Default for GraphSnapshotStore { } } -/// The pool the manager runs (config-selected, C++ parity). -pub enum WorkerBackend { - /// In-process threads. - Threads(WorkerPool), - /// Child processes (crash isolation). - Processes(ProcessPool), -} - #[cfg(test)] mod tests { use super::*; - use std::sync::atomic::AtomicUsize; use std::sync::mpsc; use std::time::Duration; use crate::texture::Texture; - fn job(tag: u64, tx: mpsc::Sender, gate: Option>) -> Job { + fn job(tag: u64, tx: mpsc::Sender, gate: Option>) -> Job { let produce: Producer = Arc::new(move |_, _| { if let Some(g) = &gate { - g.fetch_add(1, Ordering::SeqCst); + if g.load(Ordering::Acquire) { + return Err(Error::Failed("gated producer".into())); + } } Ok(crate::ticket::TicketPayload::Video(Texture::dummy())) }); @@ -399,56 +377,58 @@ mod tests { assert!(r.is_ok(), "producer must succeed here"); let _ = tx.send(tag); }), + schedule: JobSchedule::seek(), } } #[test] - fn pool_saturation_all_jobs_complete() { - let mut pool = WorkerPool::new(4); - pool.start(); + fn sync_dispatcher_runs_every_job_immediately() { + let d = InlineDispatcher::sync(); let (tx, rx) = mpsc::channel(); - for i in 0..64u64 { - assert!(pool.post(job(i, tx.clone(), None))); + for i in 0..8u64 { + assert!(d.post(job(i, tx.clone(), None))); } drop(tx); let mut seen = Vec::new(); - while let Ok(tag) = rx.recv_timeout(Duration::from_secs(10)) { + while let Ok(tag) = rx.recv_timeout(Duration::from_secs(5)) { seen.push(tag); } - assert_eq!(seen.len(), 64); - seen.sort_unstable(); - for (i, tag) in seen.iter().enumerate() { - assert_eq!(*tag, i as u64, "every job runs exactly once"); - } - pool.shutdown(); + assert_eq!(seen.len(), 8, "every job ran on the posting thread"); + d.shutdown(); + // Post after shutdown is refused. + let (tx2, _rx2) = mpsc::channel(); + assert!(!d.post(job(9, tx2, None)), "post after shutdown is refused"); } #[test] - fn shutdown_delivers_cancellation_to_queued_jobs() { - // 1 worker + a gate that blocks: jobs 2..N stay queued and must be - // delivered Err(State) at shutdown. - let gate = Arc::new(AtomicUsize::new(0)); - let mut pool = WorkerPool::new(1); - pool.start(); + fn queued_dispatcher_runs_on_demand() { + let d = InlineDispatcher::queued(); let (tx, rx) = mpsc::channel(); for i in 0..8u64 { + assert!(d.post(job(i, tx.clone(), None))); + } + assert_eq!(d.queued_count(), 8, "nothing ran yet"); + d.run(); + assert_eq!(d.queued_count(), 0); + drop(tx); + let mut seen = Vec::new(); + while let Ok(tag) = rx.recv_timeout(Duration::from_secs(5)) { + seen.push(tag); + } + assert_eq!(seen.len(), 8); + d.shutdown(); + } + + #[test] + fn queued_dispatcher_shutdown_delivers_cancellation() { + let d = InlineDispatcher::queued(); + let (tx, rx) = mpsc::channel(); + for _ in 0..4 { let tx = tx.clone(); - let gate = gate.clone(); - let produce: Producer = Arc::new(move |_, _| { - if i == 0 { - // First job blocks until shutdown begins. - let start = std::time::Instant::now(); - while gate.load(Ordering::Acquire) == 0 - && start.elapsed() < Duration::from_secs(5) - { - std::thread::sleep(Duration::from_millis(1)); - } - } - Ok(crate::ticket::TicketPayload::Video(Texture::dummy())) - }); - let job = Job { - node_identity: i, - time: Rational::new(i as i64, 1), + let p: Producer = Arc::new(|_, _| Ok(crate::ticket::TicketPayload::Video(Texture::dummy()))); + d.post(Job { + node_identity: 1, + time: Rational::new(0, 1), params: Arc::new(VideoTicketParams { viewer: 0, time: Rational::new(0, 1), @@ -461,44 +441,28 @@ mod tests { footage: None, montage: Vec::new(), }), - produce, + produce: p, done: Box::new(move |r| { let _ = tx.send(r.is_err()); }), - }; - pool.post(job); + schedule: JobSchedule::seek(), + }); } drop(tx); - gate.store(1, Ordering::Release); - pool.shutdown(); - + d.shutdown(); let mut delivered = Vec::new(); - while let Ok(is_err) = rx.recv_timeout(Duration::from_secs(5)) { - delivered.push(is_err); + while let Ok(err) = rx.recv_timeout(Duration::from_secs(5)) { + delivered.push(err); } - assert_eq!(delivered.len(), 8, "all 8 completions fire"); - assert!( - delivered.iter().filter(|&&e| e).count() >= 7, - "queued jobs complete with cancellation" - ); + assert_eq!(delivered.len(), 4, "all queued completions fire"); + assert!(delivered.iter().all(|&e| e), "queued jobs cancel at shutdown"); } #[test] - fn post_after_shutdown_is_refused() { - let mut pool = WorkerPool::new(1); - pool.start(); - pool.shutdown(); - let (tx, _rx) = mpsc::channel(); - assert!(!pool.post(job(1, tx, None))); - } - - #[test] - fn producer_panic_does_not_kill_worker() { - let mut pool = WorkerPool::new(1); - pool.start(); + fn producer_panic_does_not_kill_the_dispatcher() { + let d = InlineDispatcher::sync(); let (tx, rx) = mpsc::channel(); let tx1 = tx.clone(); - let tx2 = tx.clone(); let boom: Producer = Arc::new(|_, _| panic!("boom")); let ok: Producer = Arc::new(|_, _| Ok(crate::ticket::TicketPayload::Video(Texture::dummy()))); let params = Arc::new(VideoTicketParams { @@ -513,7 +477,7 @@ mod tests { footage: None, montage: Vec::new(), }); - pool.post(Job { + d.post(Job { node_identity: 0, time: Rational::new(0, 1), params: params.clone(), @@ -522,34 +486,25 @@ mod tests { assert!(r.is_err()); let _ = tx1.send(1u64); }), + schedule: JobSchedule::seek(), }); - pool.post(Job { + d.post(Job { node_identity: 1, time: Rational::new(1, 1), params, produce: ok, done: Box::new(move |r| { assert!(r.is_ok()); - let _ = tx2.send(2u64); + let _ = tx.send(2u64); }), + schedule: JobSchedule::seek(), }); let mut got = Vec::new(); while let Ok(v) = rx.recv_timeout(Duration::from_secs(5)) { got.push(v); } - assert_eq!(got.len(), 2, "worker survives a panicking producer"); - pool.shutdown(); - } - - #[test] - fn process_pool_is_documented_stub() { - let mut pp = ProcessPool::new(2); - assert_eq!(pp.worker_count(), 2); - assert!(pp.start().is_err(), "oakengine_ipc bridge pending"); - let (tx, _rx) = mpsc::channel(); - assert!(pp.post(job(1, tx, None)).is_err()); - pp.cancel_active(0); // no-op - pp.shutdown(); // no-op + assert_eq!(got.len(), 2, "the dispatcher survives a panicking producer"); + d.shutdown(); } #[test] diff --git a/crates/oakrender/tests/common/mod.rs b/crates/oakrender/tests/common/mod.rs index a3e2dc05e..3aa2efd05 100644 --- a/crates/oakrender/tests/common/mod.rs +++ b/crates/oakrender/tests/common/mod.rs @@ -30,11 +30,15 @@ pub struct ManagerGuard { } impl ManagerGuard { - /// Initialize the manager and hold the serialization lock. + /// Initialize the manager and hold the serialization lock. Uses the + /// test-only inline backend so no oak-worker children are spawned. pub fn init() -> Self { let guard = MANAGER_LOCK.lock().unwrap_or_else(|e| e.into_inner()); oakrender::manager::RenderManager::shutdown(); - oakrender::manager::RenderManager::init().expect("manager init"); + oakrender::manager::RenderManager::init_with_backend( + oakrender::manager::RenderBackendChoice::Threads, + ) + .expect("manager init"); Self { _guard: guard } } } diff --git a/crates/oakrender/tests/copier_test.rs b/crates/oakrender/tests/copier_test.rs index 2fd8d3a4f..722ea8e34 100644 --- a/crates/oakrender/tests/copier_test.rs +++ b/crates/oakrender/tests/copier_test.rs @@ -30,7 +30,7 @@ use oakrender::error::Error; use oakrender::frame::VideoParamsPod; use oakrender::texture::{Frame, Texture}; use oakrender::ticket::TicketArena; -use oakrender::worker::WorkerPool; +use oakrender::worker::{InlineDispatcher, JobDispatch}; fn frame_producer() -> oakrender::ticket::Producer { Arc::new(|_, _| { @@ -44,14 +44,10 @@ fn frame_producer() -> oakrender::ticket::Producer { }) } -fn cacher() -> (oakrender::autocacher::PreviewAutoCacher, WorkerPool) { - let mut pool = WorkerPool::new(2); - pool.start(); - let arena = Arc::new(TicketArena::new( - Arc::new(pool.clone()), - frame_producer(), - )); - (oakrender::autocacher::PreviewAutoCacher::new(arena), pool) +fn cacher() -> (oakrender::autocacher::PreviewAutoCacher, Arc) { + let d = InlineDispatcher::queued(); + let arena = Arc::new(TicketArena::new(d.clone(), frame_producer())); + (oakrender::autocacher::PreviewAutoCacher::new(arena), d) } /// Deep-copy through oaknode with a valid (non-empty) project identity: @@ -111,24 +107,26 @@ fn autocacher_attach_detach() { } /// cancel_video_tasks(wait=false) returns immediately with jobs -/// cancelled; wait=true blocks until workers are idle. +/// cancelled; wait=true blocks until the dispatcher is idle. #[test] fn cancel_video_tasks_semantics() { - let (mut c, mut pool) = cacher(); + let (mut c, d) = cacher(); c.attach(1); c.force_range(TimeRange::new(Rational::new(0, 1), Rational::new(5, 1))); assert_eq!(c.live_jobs().len(), 1); // wait=false: returns immediately; the ticket may still be draining. c.cancel_video_tasks(false); - // wait=true: blocks until every job finished. + // wait=true: drain the queued jobs first (the inline dispatcher has no + // background threads), then blocks until every job finished. c.force_range(TimeRange::new(Rational::new(5, 1), Rational::new(10, 1))); + d.run(); c.cancel_video_tasks(true); assert!( !c.is_rendering_custom_range(), "all custom-range jobs finished after wait" ); - pool.shutdown(); + d.shutdown(); } /// Change-record marshalling: every ChangeRecord kind survives the diff --git a/crates/oakrender/tests/ticket_worker_test.rs b/crates/oakrender/tests/ticket_worker_test.rs index e013c12b1..ea2de53c7 100644 --- a/crates/oakrender/tests/ticket_worker_test.rs +++ b/crates/oakrender/tests/ticket_worker_test.rs @@ -14,7 +14,10 @@ // You should have received a copy of the GNU General Public License // along with this program. If not, see . -//! Ticket / worker-pool contract tests. +//! Ticket / job-dispatch contract tests (M15 S2: the in-process thread +//! pool is gone; the tests drive the thread-free inline dispatcher, which +//! makes the arena's exactly-once / cancel / shutdown semantics +//! deterministic without any worker threads). mod common; @@ -28,6 +31,7 @@ use oakrender::error::Error; use oakrender::frame::VideoParamsPod; use oakrender::texture::{Frame, Texture}; use oakrender::ticket::{TicketArena, TicketId, VideoTicketParams}; +use oakrender::worker::{GraphSnapshotStore, InlineDispatcher, JobDispatch}; /// Unwrap a video ticket payload for assertions. fn res_video(res: &oakrender::ticket::TicketPayload) -> &oakrender::texture::Texture { @@ -36,7 +40,6 @@ fn res_video(res: &oakrender::ticket::TicketPayload) -> &oakrender::texture::Tex _ => panic!("expected a video payload"), } } -use oakrender::worker::{GraphSnapshotStore, WorkerPool}; fn small_frame() -> Frame { let mut f = Frame::new(); @@ -67,23 +70,19 @@ fn params(time: Rational) -> VideoTicketParams { } } -/// Opens a worker gate on drop (also on panic), so a failing assertion -/// can never leave workers spinning while the manager shuts down. -struct GateRelease(Arc); - -impl Drop for GateRelease { - fn drop(&mut self) { - self.0.store(true, Ordering::Release); - } +/// An arena on a queued inline dispatcher: jobs run only when the test +/// drains it with `InlineDispatcher::run` / `shutdown`. +fn test_arena(producer: oakrender::ticket::Producer) -> (TicketArena, Arc) { + let d = InlineDispatcher::queued(); + let arena = TicketArena::new(d.clone(), producer); + (arena, d) } /// Completion fires exactly once on success; the payload texture has /// the requested size/format. #[test] fn ticket_completion_once_success() { - let mut pool = WorkerPool::new(2); - pool.start(); - let arena = TicketArena::new(Arc::new(pool.clone()), ok_producer()); + let (arena, d) = test_arena(ok_producer()); let (tx, rx) = mpsc::channel(); let id = arena.submit_video( @@ -92,6 +91,7 @@ fn ticket_completion_once_success() { let _ = tx.send(r.is_ok()); }), ); + d.run(); arena.wait(id).unwrap(); assert!(rx.recv_timeout(Duration::from_secs(5)).unwrap()); assert!( @@ -103,24 +103,16 @@ fn ticket_completion_once_success() { assert_eq!(res_video(&res).size(), (8, 4)); assert_eq!(res_video(&res).format(), oakcore_rs::PixelFormat::F32); assert!(arena.is_finished(id)); - pool.shutdown(); + d.shutdown(); } /// Completion fires exactly once on cancel (Error::State), even when -/// cancel races the running job. +/// cancel races the running job. On the queued inline dispatcher the +/// "running" state is the queued-but-not-yet-drained job: a cancel before +/// `run` delivers State deterministically. #[test] fn ticket_completion_once_on_cancel() { - let mut pool = WorkerPool::new(1); - pool.start(); - let release = Arc::new(AtomicBool::new(false)); - let release2 = release.clone(); - let blocking: oakrender::ticket::Producer = Arc::new(move |_, _| { - while !release2.load(Ordering::Acquire) { - std::thread::sleep(Duration::from_millis(1)); - } - Ok(oakrender::ticket::TicketPayload::Video(Texture::wrap_frame(small_frame()))) - }); - let arena = TicketArena::new(Arc::new(pool.clone()), blocking); + let (arena, d) = test_arena(ok_producer()); let (tx, rx) = mpsc::channel(); let id = arena.submit_video( @@ -130,7 +122,7 @@ fn ticket_completion_once_on_cancel() { }), ); arena.cancel(id); - release.store(true, Ordering::Release); + d.run(); arena.wait(id).unwrap(); let res = rx.recv_timeout(Duration::from_secs(5)).unwrap(); assert_eq!(res.unwrap_err().code(), Error::State.code()); @@ -138,28 +130,14 @@ fn ticket_completion_once_on_cancel() { rx.recv_timeout(Duration::from_millis(50)).is_err(), "exactly once" ); - pool.shutdown(); + d.shutdown(); } /// cancel_all during shutdown delivers cancellation to every pending /// ticket; no completion fires after shutdown returns. #[test] fn shutdown_drains_completions() { - let mut pool = WorkerPool::new(1); - pool.start(); - // The producer blocks until released so no job can finish before the - // shutdown (otherwise the timing of which jobs ran is nondeterministic). - let gate = Arc::new(AtomicBool::new(false)); - let blocking: oakrender::ticket::Producer = { - let gate = gate.clone(); - Arc::new(move |_, _| { - while !gate.load(Ordering::Acquire) { - std::thread::sleep(Duration::from_millis(1)); - } - Ok(oakrender::ticket::TicketPayload::Video(Texture::wrap_frame(small_frame()))) - }) - }; - let arena = TicketArena::new(Arc::new(pool.clone()), blocking); + let (arena, d) = test_arena(ok_producer()); let (tx, rx) = mpsc::channel(); let mut ids = Vec::new(); @@ -175,8 +153,7 @@ fn shutdown_drains_completions() { } drop(tx); arena.cancel_all(); - gate.store(true, Ordering::Release); - pool.shutdown(); + d.shutdown(); let mut completions = Vec::new(); while let Ok(err) = rx.recv_timeout(Duration::from_secs(5)) { @@ -191,13 +168,11 @@ fn shutdown_drains_completions() { assert!(ids.iter().all(|id| arena.is_finished(*id))); } -/// Pool saturation: 4 workers × 64 jobs all complete; no job runs -/// twice (arena ids unique). +/// Saturation: 64 jobs all complete on the queued dispatcher after one +/// drain; no job runs twice (arena ids unique). #[test] fn pool_saturation() { - let mut pool = WorkerPool::new(4); - pool.start(); - let arena = TicketArena::new(Arc::new(pool.clone()), ok_producer()); + let (arena, d) = test_arena(ok_producer()); let (tx, rx) = mpsc::channel(); let mut ids = Vec::new(); for i in 0..64u64 { @@ -211,6 +186,7 @@ fn pool_saturation() { ids.push(id); } drop(tx); + d.run(); let mut ok = 0; while let Ok(true) = rx.recv_timeout(Duration::from_secs(10)) { ok += 1; @@ -219,55 +195,19 @@ fn pool_saturation() { ids.sort_by_key(|i| i.0); let unique: std::collections::HashSet<_> = ids.iter().collect(); assert_eq!(unique.len(), 64, "unique arena ids"); - pool.shutdown(); + d.shutdown(); } /// Ticket arena ids are monotonic and never reused within a manager /// lifetime. #[test] fn ticket_id_monotonic() { - let mut pool = WorkerPool::new(1); - pool.start(); - let arena = TicketArena::new(Arc::new(pool.clone()), ok_producer()); + let (arena, d) = test_arena(ok_producer()); let a = arena.submit_video(params(Rational::new(0, 1)), Box::new(|_| {})); let b = arena.submit_video(params(Rational::new(1, 1)), Box::new(|_| {})); let c = arena.submit_video(params(Rational::new(2, 1)), Box::new(|_| {})); assert!(a.0 < b.0 && b.0 < c.0); - pool.shutdown(); -} - -/// Process pool: documented stub until the oakengine_ipc worker binary -/// is wired (see worker.rs). -#[test] -#[ignore = "needs oakengine_ipc worker-process binary"] -fn process_pool_roundtrip() { - let mut pp = oakrender::worker::ProcessPool::new(2); - pp.start().unwrap(); - let (tx, rx) = mpsc::channel(); - let produce: oakrender::ticket::Producer = - Arc::new(|_, _| Ok(oakrender::ticket::TicketPayload::Video(Texture::wrap_frame(small_frame())))); - let job = oakrender::worker::Job { - node_identity: 1, - time: Rational::new(0, 1), - params: Arc::new(params(Rational::new(0, 1))), - produce, - done: Box::new(move |r| { - let _ = tx.send(r.is_ok()); - }), - }; - pp.post(job).unwrap(); - assert!(rx.recv_timeout(Duration::from_secs(5)).unwrap()); - pp.shutdown(); -} - -/// Crash isolation: a child killed mid-job fails that ticket with -/// Error::Failed and the pool stays usable. -#[test] -#[ignore = "needs oakengine_ipc worker-process binary"] -fn process_crash_isolation() { - let mut pp = oakrender::worker::ProcessPool::new(1); - pp.start().unwrap(); - pp.shutdown(); + d.shutdown(); } /// GraphSnapshotStore: acquire twice shares one file; release to zero diff --git a/crates/oaktask/src/render.rs b/crates/oaktask/src/render.rs index 525489828..64b27b1f8 100644 --- a/crates/oaktask/src/render.rs +++ b/crates/oaktask/src/render.rs @@ -29,8 +29,8 @@ //! clip montage (`VideoTicketParams::montage`) resolved from the //! sequence's track lists. Tickets run on the process-wide //! `oakrender::manager::RenderManager` arena when the manager is -//! initialized, otherwise on a private worker pool + arena owned by this -//! render run. +//! initialized, otherwise on a private process dispatcher + arena owned +//! by this render run (M15 S2: the in-process thread pool is gone). //! //! ## Concurrent render loop //! @@ -57,11 +57,12 @@ use oakcommon::videoparams::VideoParams; use oaknode::footage::FootageBehavior; use oaknode::sequence::SequenceBehavior; use oaknode::track::{TrackBehavior, TrackListBehavior, TrackType}; +use oakrender::procpool::{bgra8_to_f32_rgba, DispatcherConfig, ProcessDispatcher, ShmFrameRef}; use oakrender::ticket::{ ticket_kind, AudioTicketParams, MontageClip, TicketArena, TicketId, TicketPayload, TicketResult, VideoTicketParams, }; -use oakrender::worker::WorkerPool; +use oakrender::worker::JobDispatch; use crate::error::{Error, Result}; use crate::nodeops::{ @@ -477,7 +478,10 @@ impl RenderTask { /// Submit one video frame ticket at `time` (mirrors the C++ /// `start_video_ticket`). `dispatch` is the shared completion channel - /// handed to the ticket's finished callback. + /// handed to the ticket's finished callback. M15 S2: export/precache + /// tickets post at Background priority (the scheduler serves them when + /// no Seek/Playback work is pending; credit flow control caps how many + /// are in flight). fn submit_video_ticket( &self, arena: &TicketArena, @@ -487,7 +491,7 @@ impl RenderTask { let params = self.build_video_ticket(time)?; let id = arena.next_id(); let dispatch_ptr = DispatchPtr(dispatch); - arena.submit_video_with_id(id, params, Box::new(move |result| { + arena.submit_video_background_with_id(id, params, Box::new(move |result| { push_finished(id, result, dispatch_ptr); })); Ok(id) @@ -649,19 +653,33 @@ impl RenderTask { let total_slots = slots.len(); // The ticket arena: the process-wide manager arena when the - // manager is initialized, otherwise a private worker pool + arena - // owned by this run (keeps headless/test runs self-contained). - let (arena, mut private_pool) = match oakrender::manager::RenderManager::global() { + // manager is initialized, otherwise a private process dispatcher + + // arena owned by this run (keeps headless/test runs + // self-contained; M15 S2 — the in-process thread pool is gone). + let (arena, mut private_dispatch) = match oakrender::manager::RenderManager::global() { Some(manager) => (manager.tickets.clone(), None), None => { - let mut pool = WorkerPool::new(0); - pool.start(); + let dispatcher = ProcessDispatcher::new(DispatcherConfig::default()) + .map_err(|e| Error::Failed(format!("render worker pool: {e}")))?; + dispatcher + .start() + .map_err(|e| Error::Failed(format!("render worker pool start: {e}")))?; let producer: oakrender::ticket::Producer = Arc::new(|time, params| { oakrender::eval::render_produced_frame(time, params) .map(TicketPayload::Video) }); - let arena = Arc::new(TicketArena::new(Arc::new(pool.clone()), producer)); - (arena, Some(pool)) + let arena = Arc::new(TicketArena::new(dispatcher.clone(), producer)); + (arena, Some(dispatcher)) + } + }; + // M15 S2: the process dispatcher delivers completions from its poll + // loop; the render thread must pump it while it waits (there is no + // UI tick on the export thread). + let pump = || { + if let Some(m) = oakrender::manager::RenderManager::global() { + m.poll(); + } else if let Some(d) = &private_dispatch { + d.poll(); } }; @@ -769,6 +787,27 @@ impl RenderTask { task.emit_progress(progress_counter / total_length); } } + Ok(TicketPayload::ShmFrame(frame)) => { + // M15 S2 process backend: the frame lives in a + // worker shm slot. Copy it out once into a + // texture the encoder consumes (necessary copy — + // the encoder needs an owned F32 buffer), then + // release the slot. + let texture = shm_frame_to_texture(&frame); + if let Err(e) = behavior.frame_downloaded(task, &texture) { + result = Err(e); + break; + } + if let Some(m) = oakrender::manager::RenderManager::global() { + m.release_frame(&frame); + } else if let Some(d) = &private_dispatch { + d.release_frame(&frame); + } + if self.native_progress_signalling { + progress_counter += 1.0; + task.emit_progress(progress_counter / total_length); + } + } Ok(_) => { result = Err(Error::Failed( "Video render ticket delivered a non-video payload".to_string(), @@ -813,20 +852,28 @@ impl RenderTask { // Wait for the next completion (a cancellation or a hook error // aborts the wait; in-flight tickets then finish, waking us). - let mut guard = dispatch_ref - .finished - .lock() - .unwrap_or_else(|e| e.into_inner()); - while guard.is_empty() - && dispatch_ref.running.load(Ordering::SeqCst) > 0 - && !task.is_cancelled() - { - guard = dispatch_ref - .cv - .wait(guard) + // The process dispatcher is pumped so its poll loop delivers the + // completions — never while holding the finished-queue lock + // (pump()'s delivered completions lock the same queue). + loop { + pump(); + let mut guard = dispatch_ref + .finished + .lock() .unwrap_or_else(|e| e.into_inner()); + let done = !guard.is_empty() + || dispatch_ref.running.load(Ordering::SeqCst) == 0 + || task.is_cancelled(); + if done { + break; + } + let (g, _) = dispatch_ref + .cv + .wait_timeout(guard, std::time::Duration::from_millis(5)) + .unwrap_or_else(|e| e.into_inner()); + guard = g; + drop(guard); // release before the next pump } - drop(guard); if dispatch_ref .finished .lock() @@ -850,19 +897,23 @@ impl RenderTask { // Tear down. On cancellation or error, cancel and wait every ticket // still in flight (the C++ abort path), so their completions still // fire exactly once. Then wait until every callback has returned - // and free the completion channel; a private pool is shut down. + // and free the completion channel; a private dispatcher is shut + // down (M15 S2: the process dispatcher, not a thread pool). if result.is_err() || task.is_cancelled() { for &id in &in_flight { arena.cancel(id); let _ = arena.wait(id); } } - dispatch_ref.wait_idle(); + dispatch_ref.wait_idle(&pump); unsafe { drop(Box::from_raw(dispatch)); } - if let Some(mut pool) = private_pool.take() { - pool.shutdown(); + // The pump closure borrows `private_dispatch`; drop it before the + // take below. + drop(pump); + if let Some(d) = private_dispatch.take() { + d.shutdown(); } result @@ -928,10 +979,22 @@ impl RenderDispatch { /// Block until every submitted ticket has fired its callback. Called /// before freeing `self`, so no callback can touch the state afterwards. - fn wait_idle(&self) { - let mut guard = self.finished.lock().unwrap_or_else(|e| e.into_inner()); - while self.running.load(Ordering::SeqCst) > 0 { - guard = self.cv.wait(guard).unwrap_or_else(|e| e.into_inner()); + /// `pump` drives the process dispatcher's poll loop (M15 S2) — never + /// while holding the finished-queue lock (pump's delivered completions + /// lock the same queue). + fn wait_idle(&self, pump: &dyn Fn()) { + loop { + pump(); + let mut guard = self.finished.lock().unwrap_or_else(|e| e.into_inner()); + if self.running.load(Ordering::SeqCst) == 0 { + break; + } + let (g, _) = self + .cv + .wait_timeout(guard, std::time::Duration::from_millis(5)) + .unwrap_or_else(|e| e.into_inner()); + guard = g; + drop(guard); // release before the next pump } } } @@ -965,3 +1028,35 @@ fn push_finished(id: TicketId, result: TicketResult, dispatch: DispatchPtr) { dispatch.running.fetch_sub(1, Ordering::SeqCst); dispatch.cv.notify_all(); } + +/// Copy a process-backend shm frame (BGRA8 slot) out into an F32 CPU +/// texture the encoder consumes (M15 S2). The worker converted its F32 +/// pipeline output to BGRA8 for the slot (design §3.1); the encoder +/// declares F32 input, so the export converts back — a necessary +/// conversion at the encoder boundary with 8-bit quantization (S2). +fn shm_frame_to_texture(frame: &ShmFrameRef) -> oakrender::texture::Texture { + let meta = &frame.meta; + let pixels = frame + .shm + .slot_bytes(frame.slot) + .get(..meta.data_size.max(0) as usize) + .unwrap_or_default(); + let samples = bgra8_to_f32_rgba(pixels); + let mut f = oakrender::texture::Frame::new(); + f.width = meta.width; + f.height = meta.height; + f.format = oakcore_rs::PixelFormat::F32; + f.channels = 4; + f.timestamp = oakcore_rs::Rational::new(meta.time_num, meta.time_den); + f.data = samples + .chunks_exact(4) + .flat_map(|px| { + let mut bytes = [0u8; 16]; + for (i, v) in px.iter().enumerate() { + bytes[i * 4..i * 4 + 4].copy_from_slice(&v.to_le_bytes()); + } + bytes + }) + .collect(); + oakrender::texture::Texture::wrap_frame(f) +} diff --git a/docs/zh/plans/riir/M15-render-process-isolation.md b/docs/zh/plans/riir/M15-render-process-isolation.md index 0fc2331ce..b588cc646 100644 --- a/docs/zh/plans/riir/M15-render-process-isolation.md +++ b/docs/zh/plans/riir/M15-render-process-isolation.md @@ -2,6 +2,7 @@ > 状态：已批准（用户 2026-08-18 提出，作为独立追加任务，不阻塞 M12 其余阶段）。 > 前置调研：见会话调研报告（oak-worker/ipc.rs 传输层已完整、TicketArena 投递口收敛、上屏链路 6 处拷贝点）。 +> 进度：S1 完成（2026-08，协议 v2 + ProcessDispatcher + PreviewScheduler + worker 真实渲染，与线程池并存）；S2 完成（默认 Processes、删除 WorkerPool、oaktask/oak-cli/app 接入、上屏零拷贝、播放预渲染窗口）。S3 待做。 ## 1. 目标（用户原文要求） @@ -68,11 +69,15 @@ 唯一投递口 `TicketArena.pool.post`（ticket.rs:321,390）。消费者 4 个（app renderops、oak-cli、oaktask 导出、oakengine facade）API 不变。`WorkerBackend::{Threads,Processes}` 枚举已预留。oaktask 导出自建私有池改为自建私有 ProcessDispatcher（max_inflight 语义由调度器接管）。**WorkerPool 及其线程在 S2 彻底删除**（用户明确要求；单测需要的同步执行语义由 dispatcher 的 `run_inline` 测试模式提供，不保留生产线程池）。 +> **S2 落地**：`worker::WorkerPool`/`ProcessPool`/`WorkerBackend` 已删除；`worker.rs` 保留 `Job`/`JobDispatch`/`GraphSnapshotStore` 并新增线程无关的 `InlineDispatcher`（`sync` 模式 = 生产音频后端，`queued` 模式 = 测试确定性执行）。`RenderManager::init()` 默认 `Processes`；`init_with_backend(Threads)` 仅供测试（同步 inline）。oaktask 导出/CLI 走 manager 进程后端；`TicketArena::wait` 与 oaktask 渲染循环在等待时泵 `poll()`（进程后端无独立泵线程，全部非阻塞）。 + ### 3.5 上屏零拷贝改造（S2，src/ 侧） 现状拷贝点（调研报告）：ticket result `frame.data.clone()` → linesize 重排 → F32→BGRA8 → 缓存 → atlas write_texture。 改后：ticket 完成返回 `ShmFrameRef{worker, slot, meta}`；viewer 预览帧（BGRA8 槽）直接切片喂 `write_texture`；scopes 分析改为读 BGRA8（精度足够）或另请 F32 票；长期缓存（静止全分辨率单帧槽、缩略图）从 shm 拷出一次（必要拷贝，槽需回收）。断言手段：`renderops` 加拷贝字节计数器，测试断言播放路径帧字节拷贝 = 0（GPU upload 除外）。 +> **S2 落地**：`renderops::RenderedFrame` 变为 `Shm(ShmFrameRef)` / `CpuF32{..}` 枚举；`to_display()` 用槽 BGRA8 字节构造显示图（GPU 上传 staging 拷贝，走 `slot_bytes` 不计入 `main_heap_frame_copies`）；scopes 读 BGRA8（`analyze_bgra8`，8-bit 量化精度损失已在注释说明）；全分辨率/缩略图走 `slot_to_vec`（唯一计数拷贝）后立即 `release_frame`。`real_render_frame_e2e`/`process_backend_preview_path_is_zero_copy` 断言播放路径计数 = 0。 + ### 3.6 OFX 崩溃隔离 插件执行器（oakplugin `install_render_executor`）装在 **worker 进程**；主进程不再链接执行栈（app 只经 ticket API）。测试插件加"崩溃模式"（环境变量触发 raise(SIGSEGV)）→ 验收：主进程存活、受影响帧重派、worker 自动重启、渲染结果仍正确。 @@ -86,7 +91,7 @@ | 期 | 范围 | 验收 | |---|---|---| | S1（crates only） | shm spike（macOS 段上限）；协议 v2；ProcessDispatcher + WorkerHandle + 崩溃重启；worker 侧真实渲染（图快照反序列化 + montage/解码/合成 + 插件执行器）；PreviewScheduler（交织批量认领）；与线程池**并存**（配置切换）。单测 + 集成测试（崩溃隔离/无窃取/均匀性/零拷贝计数）。 | `cargo test -p oakrender -p oak-worker` 全绿；集成测试演示 4 worker 渲 480 帧无重分配、相邻帧完成时间差有界 | -| S2 | RenderManager 默认 Processes；**删除 WorkerPool**；oaktask/oak-cli 接入；src/ 上屏零拷贝（real.rs/renderops.rs/frames.rs）；app 播放前向窗口接入调度器 | `cargo test` 全绿；`cargo run` 播放流畅；拷贝计数=0；kill -SEGV worker 后播放继续 | +| S2（默认切进程 + 删线程池 + 接入） | **完成**：RenderManager 默认 Processes；删除 WorkerPool（`InlineDispatcher` 替代单测同步执行，音频走 sync inline）；oaktask/oak-cli/facade 接入（ShmFrame 消费 + 等待时泵 poll）；UI tick 泵；上屏零拷贝（renderops/real/frames/scopes，scopes 读 BGRA8）；播放前向窗口（120 帧可配）喂 PreviewScheduler，cpu_frame 先命中 shm 槽缓存 | `cargo test` 全绿；`cargo run` 播放流畅；拷贝计数=0（`main_heap_frame_copies`）；kill -SEGV worker 后播放继续（S1 集成测试覆盖） | | S3 | 音频迁移；压测调优（B、槽数、worker 数自适应）；README/docs 收尾 | 性能报告；文档 | ## 5. 风险 diff --git a/src/oakui/frames.rs b/src/oakui/frames.rs index 35327d470..8d00a48fe 100644 --- a/src/oakui/frames.rs +++ b/src/oakui/frames.rs @@ -113,6 +113,28 @@ pub(crate) fn f32_rgba_to_bgra_image(width: u32, height: u32, samples: &[f32]) - RenderImage::new(smallvec::SmallVec::from_elem(image::Frame::new(buffer), 1)) } +/// Wraps a BGRA8 pixel block into the viewers' display image (M15 S2 +/// zero-copy onscreen path). The bytes come straight from a worker's +/// shared-memory slot (already in display order) — this is the +/// GPU-upload staging buffer, the single permitted main-process copy on +/// the preview path (design §3.5). Returns `None` when `bytes` is +/// shorter than `width * height * 4`. +pub(crate) fn bgra_bytes_to_render_image( + width: u32, + height: u32, + bytes: &[u8], +) -> Option { + let need = (width * height * 4) as usize; + if bytes.len() < need { + return None; + } + let buffer = image::RgbaImage::from_raw(width, height, bytes[..need].to_vec())?; + Some(RenderImage::new(smallvec::SmallVec::from_elem( + image::Frame::new(buffer), + 1, + ))) +} + #[cfg(test)] mod tests { use super::*; diff --git a/src/oakui/real.rs b/src/oakui/real.rs index c2ff89aa5..12f7f42f8 100644 --- a/src/oakui/real.rs +++ b/src/oakui/real.rs @@ -59,7 +59,7 @@ //! worker holds the project's `Arc`, so a project drop mid-render is a //! non-event (the drained frame is discarded by the generation check). -use std::collections::{BTreeSet, HashMap, HashSet}; +use std::collections::{BTreeMap, BTreeSet, HashMap, HashSet}; use std::path::{Path, PathBuf}; use std::sync::mpsc; use std::sync::{Arc, Mutex}; @@ -80,15 +80,17 @@ use gpui_widgets::viewer::PlaybackClock; use oaknode::id::NodeId; use oaknode::track::TrackType; +use oakrender::manager::RenderManager; +use oakrender::procpool::{bgra8_to_rgba8, ShmFrameRef}; use oaktimeline::handle::CHandle; use super::engine::{ AppEngine, EngineGateway, ExportSession, LibraryProject, Monitor, Project, ScopeData, Sequence, VideoFormat, }; -use super::frames::{f32_rgba_to_bgra_image, synthetic_frame_samples}; +use super::frames::{bgra_bytes_to_render_image, f32_rgba_to_bgra_image, synthetic_frame_samples}; use super::graphops::{self, ProjectRef}; -use super::scopes::analyze_f32_rgba; +use super::scopes::{analyze_bgra8, analyze_f32_rgba}; use super::transport::TransportState; /// The project name of a blank project before it is saved. @@ -235,6 +237,42 @@ enum FullResTarget { Footage(NodeId), } +// --------------------------------------------------------------------------- +// Playback pre-render window (M15 S2) +// --------------------------------------------------------------------------- +// +// During playback the engine feeds the forward window (default 120 frames, +// config `PlaybackPreRenderFrames`) to the `PreviewScheduler` through the +// process dispatcher: the scheduler interleaves the frames across the +// 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 +// that fall out of the window (or whose params were invalidated) release +// their slots back to the workers. + +/// The default forward pre-render window (frames). +const DEFAULT_PREVIEW_WINDOW_FORWARD: i64 = 120; +/// Config key for the forward pre-render window size (frames). +const CONFIG_KEY_PREVIEW_WINDOW: &str = "PlaybackPreRenderFrames"; + +/// One monitor's playback pre-render state (UI-thread-owned; the completions +/// delivered by the dispatcher's poll run on the UI thread too). +#[derive(Default)] +struct PreviewWindow { + /// The node identity the window renders (sequence / footage). + sequence: u64, + /// The render-params generation the window was built for (bumped on + /// invalidation; a mismatch cancels the old requests and rebuilds). + generation: u64, + /// Frames already submitted to the scheduler (pending or in flight). + submitted: BTreeSet, + /// Rendered frames held in shm slots, keyed by frame number. A frame + /// is consumed by `cpu_frame` (slot released after the display image + /// is built) or released when it falls out of the window. + slots: BTreeMap, +} + /// One background full-resolution render request (built on the UI thread /// at schedule time; the worker thread owns it from there). struct FullResRequest { @@ -323,28 +361,70 @@ fn thumbnail_path(filename: &str) -> PathBuf { // Frame conversion // --------------------------------------------------------------------------- -/// Repack one F32 RGBA rendered frame (rows padded to linesize) into -/// tightly packed samples. Returns `(width, height, samples)` when the -/// frame is well-formed (positive geometry, the pipeline's F32 format). +/// Repack one in-process F32 RGBA rendered frame (rows padded to +/// linesize) into tightly packed samples. Returns `(width, height, +/// samples)` when the frame is the in-process F32 variant (the shm +/// variant is BGRA8 and is read as bytes instead). fn read_f32_frame(frame: &super::renderops::RenderedFrame) -> Option<(u32, u32, Vec)> { - let (width, height, linesize) = (frame.width, frame.height, frame.linesize); - if width <= 0 || height <= 0 || frame.format != super::renderops::PIXEL_FORMAT_F32 { + let super::renderops::RenderedFrame::CpuF32 { + width, + height, + linesize, + data, + } = frame + else { + return None; + }; + if *width <= 0 || *height <= 0 { return None; } - let row_bytes = (width * 4 * 4) as usize; - let linesize = (linesize as usize).max(row_bytes); - if frame.data.len() < linesize * height as usize { + let row_bytes = (*width * 4 * 4) as usize; + let linesize = (*linesize as usize).max(row_bytes); + if data.len() < linesize * *height as usize { return None; } - let mut samples = vec![0.0f32; (width * height * 4) as usize]; - for y in 0..height as usize { - let row = &frame.data[y * linesize..y * linesize + row_bytes]; + let mut samples = vec![0.0f32; (*width * *height * 4) as usize]; + for y in 0..*height as usize { + let row = &data[y * linesize..y * linesize + row_bytes]; for (i, px) in row.chunks_exact(4).enumerate() { let v = f32::from_ne_bytes([px[0], px[1], px[2], px[3]]); - samples[y * (width as usize) * 4 + i] = v; + samples[y * (*width as usize) * 4 + i] = v; + } + } + Some((*width as u32, *height as u32, samples)) +} + +/// Release the shm slot a rendered frame holds, if any (M15 S2: the +/// zero-copy onscreen path builds the display image from the slot bytes, +/// then returns the slot to its worker — slot release = cache eviction). +/// No-op for in-process frames and when the manager is down. +fn release_rendered_frame(rendered: &super::renderops::RenderedFrame) { + if let super::renderops::RenderedFrame::Shm(frame) = rendered { + if let Some(m) = RenderManager::global() { + m.release_frame(frame); + } + } +} + +/// Build an owned viewer image from a rendered frame for a long-lived +/// cache (full-res fills, thumbnails). The shm variant copies the slot +/// bytes out once (`slot_to_vec` — the counted copy path, necessary +/// because the image must outlive the slot), then the caller releases the +/// slot; the in-process variant converts F32→BGRA8. +fn rendered_to_owned_image(rendered: &super::renderops::RenderedFrame) -> Option> { + match rendered { + super::renderops::RenderedFrame::Shm(f) => { + let meta = &f.meta; + let (w, h) = (meta.width.max(0) as u32, meta.height.max(0) as u32); + let pixels = f.shm.slot_to_vec(f.slot); + let data = pixels.get(..meta.data_size.max(0) as usize)?; + bgra_bytes_to_render_image(w, h, data).map(Arc::new) + } + super::renderops::RenderedFrame::CpuF32 { .. } => { + let (w, h, samples) = read_f32_frame(rendered)?; + Some(Arc::new(f32_rgba_to_bgra_image(w, h, &samples))) } } - Some((width as u32, height as u32, samples)) } // --------------------------------------------------------------------------- @@ -675,6 +755,15 @@ pub struct RealEngine { /// project drop); completions tagged with a stale generation are /// discarded by the drain. full_res_generation: u64, + /// M15 S2 playback pre-render state, keyed by monitor (see + /// [`PreviewWindow`]). An `Arc` so the ticket completions (which fire + /// from the dispatcher's poll on the UI thread) can reach the cache; + /// the mutex keeps the engine `Sync`. + preview_windows: Arc>>, + /// Bumped whenever the preview content can change (edit / proxy + /// toggle / selection / project drop); the pre-render window rebuilds + /// against the new generation. + preview_generation: u64, /// The channel background full-res jobs report finished frames through; /// drained on the app tick. The mutex keeps the engine `Sync` (the /// channel is only ever touched on the UI thread). @@ -769,6 +858,8 @@ impl RealEngine { meter_phase: 0, cpu_frame_cache: Mutex::new(HashMap::new()), full_res_generation: 0, + preview_windows: Arc::new(Mutex::new(HashMap::new())), + preview_generation: 0, full_res_rx: Mutex::new(full_res_rx), full_res_tx: Mutex::new(full_res_tx), renderer: Mutex::new(RendererSlot::Untried), @@ -858,10 +949,13 @@ impl RealEngine { /// Renders one program-monitor frame through the oakrender ticket /// arena: builds the sequence's montage at `frame`, renders at the - /// proxy geometry, analyzes the scope samples from the F32 RGBA - /// result, and downconverts to BGRA8. Returns `None` (the caller falls - /// back to the synthetic pattern) when no sequence is open, the render - /// manager is unavailable, or the render itself fails. + /// proxy geometry, and produces the viewer display image plus the + /// scope samples (M15 S2: the process backend delivers a BGRA8 shm + /// slot; the display image is built straight from the slot bytes — + /// the GPU-upload staging copy — and the slot released). Returns + /// `None` (the caller falls back to the synthetic pattern) when no + /// sequence is open, the render manager is unavailable, or the render + /// itself fails. fn render_program_frame(&self, frame: Frame) -> Option<(RenderImage, ScopeData)> { let project = self.project_ref()?.clone(); let seq = self.sequence?; @@ -877,9 +971,9 @@ impl RealEngine { match super::renderops::render_sequence_frame(&project, seq, frame.0, tb, width, height) { Ok(rendered) => { *slot = RendererSlot::Ready; - let (width, height, samples) = read_f32_frame(&rendered)?; - let scope = analyze_f32_rgba(width, height, &samples); - Some((f32_rgba_to_bgra_image(width, height, &samples), scope)) + let out = rendered.to_display(); + release_rendered_frame(&rendered); + out } Err(error) => { println!("[real engine] render_frame failed: {error}"); @@ -902,9 +996,10 @@ impl RealEngine { /// Renders one source-monitor frame through the ticket arena: the /// currently selected footage node decoded at the proxy geometry (same - /// pipeline as the program monitor). Returns `None` (the caller falls - /// back to the synthetic pattern) when no footage is selected, the - /// render manager is unavailable, or the render itself fails. + /// pipeline as the program monitor, M15 S2 shm slot + release). + /// Returns `None` (the caller falls back to the synthetic pattern) + /// when no footage is selected, the render manager is unavailable, or + /// the render itself fails. fn render_source_frame(&self, frame: Frame) -> Option<(RenderImage, ScopeData)> { let project = self.project_ref()?.clone(); let node = self.selected_footage_node()?; @@ -920,9 +1015,9 @@ impl RealEngine { match super::renderops::render_footage_frame(&project, node, frame.0, tb, width, height) { Ok(rendered) => { *slot = RendererSlot::Ready; - let (width, height, samples) = read_f32_frame(&rendered)?; - let scope = analyze_f32_rgba(width, height, &samples); - Some((f32_rgba_to_bgra_image(width, height, &samples), scope)) + let out = rendered.to_display(); + release_rendered_frame(&rendered); + out } Err(error) => { println!("[real engine] source render_frame failed: {error}"); @@ -981,12 +1076,17 @@ impl RealEngine { } }; if let Ok(rendered) = rendered { - if let Some((width, height, samples)) = read_f32_frame(&rendered) { + // M15 S2: the process backend delivers a shm slot — copy the + // pixels out once (counted, long-lived cache) and release the + // slot. + let image = rendered_to_owned_image(&rendered); + release_rendered_frame(&rendered); + if let Some(image) = image { event = Some(FullResEvent { monitor, frame, generation, - image: Arc::new(f32_rgba_to_bgra_image(width, height, &samples)), + image, }); } } @@ -1040,6 +1140,216 @@ impl RealEngine { } } + // ----------------------------------------------------------------------- + // M15 S2: playback pre-render window + // ----------------------------------------------------------------------- + + /// Feeds the playback pre-render window for `monitor` into the + /// scheduler (M15 S2): while playing, the forward window's frames are + /// submitted at Playback priority (ordered by playhead distance), so + /// workers render them ahead of the playhead into shm slots. The + /// window rebuilds when the node or the render-params generation + /// 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). + let playing = self.clock(monitor).read(cx).transport.is_playing(); + if !playing { + return; + } + let Some(project) = self.project.clone() else { return }; + let Some(tb) = self.time_base() else { return }; + let Some((width, height)) = self.proxy_render_size() else { return }; + let Some(node) = (match monitor { + Monitor::Program => self.sequence, + Monitor::Source => self.selected_footage_node(), + }) else { + return; + }; + let node_id = node.identity(); + let playhead = self.clock_frame(monitor, cx).0; + if playhead < 0 { + return; + } + let length = match monitor { + Monitor::Program => self.sequence_length().0, + Monitor::Source => self.source_length().0, + }; + let Some(m) = RenderManager::global() else { return }; + + let forward = config_get_int(CONFIG_KEY_PREVIEW_WINDOW, DEFAULT_PREVIEW_WINDOW_FORWARD) + .clamp(8, 1200); + let end = length.max(playhead).min(playhead + forward); + + // Reset / rebuild when the node changed or an invalidation bumped the + // generation (the old pending/claimed requests are cancelled). + let mut windows = self.preview_windows.lock().unwrap_or_else(|e| e.into_inner()); + let window = windows.entry(monitor).or_default(); + if window.sequence != node_id || window.generation != self.preview_generation { + m.cancel_preview_sequence(window.sequence); + for slot in window.slots.values() { + m.release_frame(slot); + } + window.slots.clear(); + window.submitted.clear(); + window.sequence = node_id; + window.generation = self.preview_generation; + } + // Release slots that fell behind the playhead (frames passed without + // being displayed); the just-displayed frame was already consumed by + // `cpu_frame`. Prune the submitted set the same way. + let keep_from = (playhead - 2).max(0); + let stale: Vec = window + .slots + .keys() + .copied() + .filter(|f| *f < keep_from) + .collect(); + for f in stale { + if let Some(slot) = window.slots.remove(&f) { + m.release_frame(&slot); + } + } + window + .submitted + .retain(|f| *f >= keep_from || (*f >= playhead && *f < end)); + let new_frames: Vec = (playhead.max(0)..end) + .filter(|f| !window.submitted.contains(f)) + .collect(); + drop(windows); + + for frame in new_frames { + let params = match monitor { + Monitor::Program => super::renderops::sequence_frame_params( + &project, + node, + frame, + tb, + width, + height, + ), + Monitor::Source => { + super::renderops::footage_frame_params(&project, node, frame, tb, width, height) + } + }; + let Ok(params) = params else { continue }; + let distance = frame.saturating_sub(playhead).abs(); + let version = self.preview_generation; + let preview_windows = self.preview_windows.clone(); + let done: oakrender::ticket::Completion = Box::new(move |result| { + let mut windows = + preview_windows.lock().unwrap_or_else(|e| e.into_inner()); + let window = windows.entry(monitor).or_default(); + match result { + Ok(oakrender::ticket::TicketPayload::ShmFrame(slot)) => { + // The rendered frame is cached in its shm slot until + // the playhead reaches it (cpu_frame) or it falls out + // of the window. + window.slots.insert(frame, slot); + } + _ => { + // Render failed / cancelled: allow a re-request. + window.submitted.remove(&frame); + } + } + }); + m.tickets.submit_playback(params, frame, distance, version, done); + self.preview_windows + .lock() + .unwrap_or_else(|e| e.into_inner()) + .entry(monitor) + .or_default() + .submitted + .insert(frame); + } + } + + /// Looks a frame up in the pre-rendered playback window's shm slot + /// cache (M15 S2): builds the viewer display image + scope samples + /// straight from the slot bytes (the GPU-upload staging copy) and + /// releases the slot. Returns `None` when the frame is not cached. + fn preview_slot_frame( + &self, + monitor: Monitor, + frame: Frame, + ) -> Option<(Arc, ScopeData)> { + let node = match monitor { + Monitor::Program => self.sequence?, + Monitor::Source => self.selected_footage_node()?, + }; + let node_id = node.identity(); + let mut windows = self.preview_windows.lock().unwrap_or_else(|e| e.into_inner()); + let window = windows.get_mut(&monitor)?; + if window.sequence != node_id || window.generation != self.preview_generation { + return None; + } + let slot = window.slots.remove(&frame.0)?; + let out = { + let meta = &slot.meta; + let (w, h) = (meta.width.max(0) as u32, meta.height.max(0) as u32); + let data = slot + .shm + .slot_bytes(slot.slot) + .get(..meta.data_size.max(0) as usize)?; + let image = bgra_bytes_to_render_image(w, h, data)?; + let scope = analyze_bgra8(w, h, data); + (Arc::new(image), scope) + }; + if let Some(m) = RenderManager::global() { + m.release_frame(&slot); + } + Some(out) + } + + /// Cancels every monitor's pre-render window and releases its held + /// slots (edit / selection change / project drop / preview-media + /// invalidation). + fn cancel_preview_windows(&mut self) { + let mut windows = self.preview_windows.lock().unwrap_or_else(|e| e.into_inner()); + let m = RenderManager::global(); + for window in windows.values_mut() { + if let Some(m) = &m { + m.cancel_preview_sequence(window.sequence); + for slot in window.slots.values() { + m.release_frame(slot); + } + } + window.slots.clear(); + window.submitted.clear(); + } + } + + /// Cancels one monitor's pre-render window (source selection change: + /// the old footage's window is stale even though the program window is + /// untouched). + fn cancel_preview_window(&mut self, monitor: Monitor) { + let mut windows = self.preview_windows.lock().unwrap_or_else(|e| e.into_inner()); + let Some(window) = windows.get_mut(&monitor) else { + return; + }; + if let Some(m) = RenderManager::global() { + m.cancel_preview_sequence(window.sequence); + for slot in window.slots.values() { + m.release_frame(slot); + } + } + window.slots.clear(); + window.submitted.clear(); + } + + /// Invalidates every cached/rendered preview frame: the CPU cache is + /// cleared, the full-res and pre-render-window generations bumped, and + /// the pre-render windows cancelled (pending/claimed requests and held + /// slots released). Shared by edits, undo-stack changes, project drops + /// and preview-media invalidation. + fn invalidate_rendered_frames(&mut self) { + self.cpu_frame_cache.lock().unwrap().clear(); + self.full_res_generation = self.full_res_generation.wrapping_add(1); + self.preview_generation = self.preview_generation.wrapping_add(1); + self.cancel_preview_windows(); + } + /// Attaches cached thumbnails to the bin entries, spawning a background /// generation job for every footage that has none yet. Entries without a /// renderable frame keep the widget's placeholder. @@ -1110,7 +1420,9 @@ impl RealEngine { return None; } // Frame zero: the timebase only scales the timestamp, so any valid - // pair produces time 0. + // pair produces time 0. M15 S2: the process backend delivers a shm + // slot — copy the pixels out once (counted, the PNG must outlive the + // slot), convert BGRA→RGBA, release the slot. let rendered = super::renderops::render_footage_frame( project, node, @@ -1120,11 +1432,24 @@ impl RealEngine { THUMBNAIL_HEIGHT, ) .ok()?; - let (width, height, samples) = read_f32_frame(&rendered)?; - let bytes: Vec = samples - .iter() - .map(|v| (v.clamp(0.0, 1.0) * 255.0).round() as u8) - .collect(); + let (width, height, bytes) = match &rendered { + super::renderops::RenderedFrame::Shm(f) => { + let meta = &f.meta; + let (w, h) = (meta.width.max(0) as u32, meta.height.max(0) as u32); + let pixels = f.shm.slot_to_vec(f.slot); + let data = pixels.get(..meta.data_size.max(0) as usize)?; + (w, h, bgra8_to_rgba8(data)) + } + super::renderops::RenderedFrame::CpuF32 { .. } => { + let (w, h, samples) = read_f32_frame(&rendered)?; + let bytes: Vec = samples + .iter() + .map(|v| (v.clamp(0.0, 1.0) * 255.0).round() as u8) + .collect(); + (w, h, bytes) + } + }; + release_rendered_frame(&rendered); let image = image::RgbaImage::from_raw(width, height, bytes)?; std::fs::create_dir_all(path.parent()?).ok()?; // Write aside then rename so readers never see a partial file. @@ -1149,13 +1474,11 @@ impl RealEngine { } /// Invalidates every monitor's rendered frames (the preview media - /// changed — proxy toggled, generated or deleted): the CPU cache is - /// cleared and the full-res generation bumped so in-flight fills are - /// discarded on arrival. No timeline rebuild is needed — the montage - /// resolution reads the proxy switch on every pull. + /// changed — proxy toggled, generated or deleted). No timeline rebuild + /// is needed — the montage resolution reads the proxy switch on every + /// pull. fn invalidate_preview_frames(&mut self, cx: &mut Context) { - self.cpu_frame_cache.lock().unwrap().clear(); - self.full_res_generation = self.full_res_generation.wrapping_add(1); + self.invalidate_rendered_frames(); cx.notify(); } @@ -1696,11 +2019,10 @@ impl RealEngine { } self.project = None; self.sequence = None; - self.cpu_frame_cache.lock().unwrap().clear(); // The sequence an in-flight full-res job may still be rendering is // gone (the job holds its own project `Arc`, so it stays valid, but // its frame belongs to the dropped project): mark it stale. - self.full_res_generation = self.full_res_generation.wrapping_add(1); + self.invalidate_rendered_frames(); // Same for in-flight thumbnail jobs; the cache is per-project too // (identities are only unique within one graph). self.thumb_generation = self.thumb_generation.wrapping_add(1); @@ -1787,8 +2109,7 @@ impl RealEngine { fn apply_stack_change(&mut self, cx: &mut Context) { self.refresh_sequence_info(); self.rebuild_timeline(); - self.cpu_frame_cache.lock().unwrap().clear(); - self.full_res_generation = self.full_res_generation.wrapping_add(1); + self.invalidate_rendered_frames(); cx.notify(); } @@ -1966,9 +2287,9 @@ impl RealEngine { self.refresh_sequence_info(); self.rebuild_timeline(); // The sequence content changed: cached rendered frames are stale, - // and so are any in-flight full-res renders (M12 P5a). - self.cpu_frame_cache.lock().unwrap().clear(); - self.full_res_generation = self.full_res_generation.wrapping_add(1); + // and so are any in-flight full-res renders and pre-render windows + // (M12 P5a / M15 S2). + self.invalidate_rendered_frames(); cx.notify(); } } @@ -2075,6 +2396,16 @@ impl EngineGateway for RealEngine { } self.mirror_program_playhead(cx); self.meter_phase = self.meter_phase.wrapping_add(1); + // M15 S2: pump the process dispatcher — ticket completions (the + // pre-render window, full-res fills, synchronous renders) are + // delivered from its poll loop, which must run on the UI tick. + // Then feed the playback pre-render windows before draining the + // completion channels. + if let Some(m) = RenderManager::global() { + m.poll(); + } + self.update_preview_window(Monitor::Source, cx); + self.update_preview_window(Monitor::Program, cx); // M12 P1: while the program plays, pull the audio for the // current playhead window and queue it for the output device. if self.program_playing { @@ -2246,6 +2577,17 @@ impl AppEngine for RealEngine { if let Some(image) = cache.entry(monitor).or_default().image_for(frame.0) { return image.clone(); } + // M15 S2: try the pre-rendered playback window first — the frame's + // pixels are already in a worker shm slot (zero copy: build the + // display image from the slot bytes, then release the slot). + if let Some((image, scope)) = self.preview_slot_frame(monitor, frame) { + cache.entry(monitor).or_default().proxy = Some(ProxyEntry { + frame: frame.0, + image: image.clone(), + scope, + }); + return image; + } // Both monitors render through the oakrender ticket arena (falling // back to the synthetic pattern when rendering is unavailable): the // program monitor renders the current sequence, the source monitor @@ -2329,10 +2671,12 @@ impl AppEngine for RealEngine { // The source monitor renders the selected footage node: a new // selection must drop the stale cached frame (the cache key only // tracks the playhead frame), and any in-flight full-res job for - // the old selection is stale. + // the old selection is stale. The source pre-render window (M15 + // S2) rebuilds against the new footage. *self.source_renderer.lock().unwrap() = RendererSlot::Untried; self.cpu_frame_cache.lock().unwrap().remove(&Monitor::Source); self.full_res_generation = self.full_res_generation.wrapping_add(1); + self.cancel_preview_window(Monitor::Source); } cx.notify(); } @@ -3930,6 +4274,35 @@ mod tests { crate::oakui::graphops::test_lock() } + /// Points the render manager's worker resolution at the built + /// oak-worker binary (dev layout `target/debug/oak-worker`) for tests + /// that render through the process backend. The default resolver looks + /// next to the *test* binary (`target/debug/deps`), which holds no + /// oak-worker. Restored on drop; only used inside [`media_lock`] + /// sections so the process env stays serialized. + struct WorkerBinGuard { + prev: Option, + } + impl WorkerBinGuard { + fn set() -> Self { + let prev = std::env::var("OAK_WORKER_BIN").ok(); + let path = std::path::Path::new(env!("CARGO_MANIFEST_DIR")) + .join("target") + .join("debug") + .join("oak-worker"); + std::env::set_var("OAK_WORKER_BIN", path); + Self { prev } + } + } + impl Drop for WorkerBinGuard { + fn drop(&mut self) { + match &self.prev { + Some(p) => std::env::set_var("OAK_WORKER_BIN", p), + None => std::env::remove_var("OAK_WORKER_BIN"), + } + } + } + #[test] fn project_format_dispatches_by_extension() { assert_eq!( @@ -4162,14 +4535,17 @@ mod tests { /// End-to-end CPU render through the same path /// [`RealEngine::render_program_frame`] uses: with the render manager - /// up, rendering an in-memory sequence produces a real F32 frame at the - /// requested proxy geometry, and the samples are well-formed (finite, in - /// range). The sequence starts empty, so the picture is black; with a - /// clip of real media on the video track the same render produces the - /// decoded footage (known content, non-black). + /// up (the default M15 S2 process backend), rendering an in-memory + /// sequence produces a real BGRA8 frame in a shared-memory slot at the + /// requested proxy geometry, the display image builds straight from + /// the slot bytes (zero copy), and the picture is well-formed. The + /// sequence starts empty, so the picture is black; with a clip of real + /// media on the video track the same render produces the decoded + /// footage (known content, non-black). #[test] fn real_render_frame_e2e() { let _media = media_lock(); + let _worker = WorkerBinGuard::set(); if !crate::oakui::renderops::ensure_render_manager() { panic!("the render manager failed to start"); } @@ -4180,21 +4556,23 @@ mod tests { .expect("the sequence has a frame rate"); // The app's proxy size: sequence aspect (default 1920x1080) scaled - // to a 480px long edge, F32 at the sequence's rate. + // to a 480px long edge. let frame = crate::oakui::renderops::render_sequence_frame(&project, seq, 0, tb, 480, 270) .expect("render_frame must produce a frame"); - assert_eq!((frame.width, frame.height), (480, 270)); - assert_eq!(frame.format, crate::oakui::renderops::PIXEL_FORMAT_F32); - assert!(frame.linesize >= 480 * 4 * 4, "linesize covers a full row"); - let (_, _, samples) = read_f32_frame(&frame).expect("well-formed F32 frame"); + assert_eq!((frame.width(), frame.height()), (480, 270)); assert!( - samples.iter().all(|v| v.is_finite() && (0.0..=1.0).contains(v)), - "samples in range" + frame.is_shm(), + "the default process backend delivers shm slots (got format {})", + frame.format() ); + let (image, _scope) = frame.to_display().expect("display image from the slot"); + let bytes = image.as_bytes(0).expect("one frame"); + assert_eq!(bytes.len(), 480 * 270 * 4, "BGRA8 proxy geometry"); assert!( - samples.iter().all(|&v| v == 0.0), + bytes.iter().all(|&b| b == 0), "an empty sequence renders transparent black" ); + release_rendered_frame(&frame); // M12 P0: with a clip of real media on the video track, the same // render must produce the decoded footage (known content, non @@ -4212,20 +4590,22 @@ mod tests { .expect("clip placement"); let frame = crate::oakui::renderops::render_sequence_frame(&project, seq, 0, tb, 480, 270) .expect("render_frame with a clip must produce a frame"); - let (_, _, samples) = read_f32_frame(&frame).expect("well-formed F32 frame"); - let nonzero = samples.iter().filter(|&&v| v != 0.0).count(); + let (image, _scope) = frame.to_display().expect("display image from the slot"); + let bytes = image.as_bytes(0).expect("one frame"); + let nonzero = bytes.chunks(4).filter(|px| px[..3].iter().any(|&c| c != 0)).count(); assert!( nonzero > 0, "the sequence with a footage clip must render non-black pixels" ); // Known content: the test clip's left half is red on frame 0 — - // the center-left pixel must be red-dominant. - let px = |x: usize, y: usize| &samples[(y * 480 + x) * 4..][..4]; + // the center-left pixel must be red-dominant (BGRA bytes: R at 2). + let px = |x: usize, y: usize| &bytes[(y * 480 + x) * 4..][..4]; let center_left = px(120, 135); assert!( - center_left[0] > 0.5 && center_left[1] < 0.4 && center_left[2] < 0.4, + center_left[2] > 127 && center_left[0] < 102 && center_left[1] < 102, "center-left pixel stays red from the decoded clip: {center_left:?}" ); + release_rendered_frame(&frame); // A second frame at a later timestamp renders too. assert!(crate::oakui::renderops::render_sequence_frame(&project, seq, 30, tb, 480, 270).is_ok()); @@ -4237,6 +4617,74 @@ mod tests { let _ = std::fs::remove_file(&media); } + /// M15 S2 acceptance: the app's onscreen path reads the process + /// backend's shm slots without the counted main-process copy. The + /// preview path (`RenderedFrame::to_display`) wraps the slot bytes + /// into the display buffer — the GPU-upload staging copy — and must + /// leave `main_heap_frame_copies == 0`; the long-lived full-res path + /// (`rendered_to_owned_image`) is the one counted copy + /// (`slot_to_vec`), released right after. + #[test] + fn process_backend_preview_path_is_zero_copy() { + let _media = media_lock(); + let _worker = WorkerBinGuard::set(); + use oakrender::manager::{RenderBackendChoice, RenderManager}; + use oakrender::procpool::{ + main_heap_frame_copies, reset_main_heap_frame_copies, DispatcherConfig, + }; + RenderManager::shutdown(); + let config = DispatcherConfig { + worker_bin: Some( + std::path::Path::new(env!("CARGO_MANIFEST_DIR")) + .join("target/debug/oak-worker"), + ), + workers: 1, + slots_per_worker: 2, + width: 64, + height: 64, + batch_size: 2, + ..Default::default() + }; + RenderManager::init_with_backend(RenderBackendChoice::Processes(config)) + .expect("process backend init"); + + let project = graphops::create_project(); + let seq = graphops::create_sequence(&project, "Zero Copy"); + let tb = graphops::sequence_time_base(&graphops::lock(&project).graph, seq).unwrap(); + + reset_main_heap_frame_copies(); + let frame = crate::oakui::renderops::render_sequence_frame(&project, seq, 0, tb, 64, 64) + .expect("render_frame must produce a frame"); + let crate::oakui::renderops::RenderedFrame::Shm(slot) = &frame else { + panic!("the process backend must deliver a shm slot"); + }; + assert_eq!(slot.meta.format, crate::oakui::renderops::SLOT_FORMAT_BGRA8); + let (image, _scope) = frame.to_display().expect("display image from the slot"); + assert_eq!(image.as_bytes(0).expect("one frame").len(), 64 * 64 * 4); + assert_eq!( + main_heap_frame_copies(), + 0, + "the preview path must not copy through slot_to_vec" + ); + release_rendered_frame(&frame); + assert_eq!(main_heap_frame_copies(), 0); + + // The long-lived full-res path is the one counted copy. + let frame = crate::oakui::renderops::render_sequence_frame(&project, seq, 0, tb, 64, 64) + .expect("render_frame must produce a frame"); + let image = rendered_to_owned_image(&frame).expect("owned full-res image"); + assert_eq!(image.as_bytes(0).expect("one frame").len(), 64 * 64 * 4); + assert_eq!( + main_heap_frame_copies(), + 1, + "the long-lived cache path is the counted copy" + ); + release_rendered_frame(&frame); + + RenderManager::shutdown(); + oakundo::global::clear().unwrap(); + } + /// M12 P3 acceptance: importing a media file makes it appear in the /// project browser's real folder tree. #[test] @@ -4489,6 +4937,7 @@ mod tests { #[test] fn full_res_worker_outlives_a_dropped_project() { let _media = media_lock(); + let _worker = WorkerBinGuard::set(); if !crate::oakui::renderops::ensure_render_manager() { panic!("the render manager failed to start"); } @@ -4660,6 +5109,7 @@ mod tests { #[test] fn full_res_worker_renders_real_frame() { let _media = media_lock(); + let _worker = WorkerBinGuard::set(); if !crate::oakui::renderops::ensure_render_manager() { panic!("the render manager failed to start"); } @@ -4721,6 +5171,7 @@ mod tests { #[gpui::test] async fn real_engine_fills_full_res_behind_the_proxy(cx: &mut gpui::TestAppContext) { let _media = media_lock(); + let _worker = WorkerBinGuard::set(); let engine = cx.update(|cx| cx.new(|cx| RealEngine::create(cx))); cx.update(|app| engine.update(app, |engine, cx| engine.new_project(cx))); diff --git a/src/oakui/renderops.rs b/src/oakui/renderops.rs index 2fec36f9b..2eb21cd01 100644 --- a/src/oakui/renderops.rs +++ b/src/oakui/renderops.rs @@ -28,22 +28,31 @@ use std::sync::mpsc; +use gpui::RenderImage; use oakcore_rs::{Rational, TimeRange}; use oaknode::id::NodeId; use oaknode::track::TrackType; use oakrender::manager::RenderManager; +use oakrender::procpool::ShmFrameRef; use oakrender::texture::Texture; use oakrender::ticket::{AudioTicketParams, MontageClip, TicketPayload, VideoTicketParams}; use super::engine::{ExportEvent, ExportSession}; +use super::frames::{bgra_bytes_to_render_image, f32_rgba_to_bgra_image}; use super::graphops::{ clip_behavior, lock, sequence_behavior, track_behavior, track_list_behavior, ProjectRef, }; +use super::scopes::{analyze_bgra8, analyze_f32_rgba, ScopeData}; /// The pixel format the viewers render in (`oakcore_rs::PixelFormat::F32`, /// the pipeline's internal format; the app downconverts to BGRA itself). pub const PIXEL_FORMAT_F32: i32 = 4; +/// The BGRA8 slot wire format the process backend renders into (M15 S2): +/// the worker converts its F32 pipeline output to BGRA8 at the end of the +/// render, so the main process reads display-ready bytes from the slot. +pub const SLOT_FORMAT_BGRA8: i32 = oakrender::ipc::SLOT_FORMAT_BGRA8; + // --------------------------------------------------------------------------- // Render manager // --------------------------------------------------------------------------- @@ -232,19 +241,113 @@ pub fn audio_montage(p: &ProjectRef, seq: NodeId, range: TimeRange) -> Vec, + }, +} + +impl RenderedFrame { /// Width in pixels. - pub width: i32, + pub fn width(&self) -> i32 { + match self { + RenderedFrame::Shm(f) => f.meta.width, + RenderedFrame::CpuF32 { width, .. } => *width, + } + } + /// Height in pixels. - pub height: i32, - /// Pixel format (`oakcore_rs::PixelFormat` as int). - pub format: i32, - /// Bytes per scanline (stride). - pub linesize: i32, - /// Pixel data (at least `linesize * height` bytes). - pub data: Vec, + pub fn height(&self) -> i32 { + match self { + RenderedFrame::Shm(f) => f.meta.height, + RenderedFrame::CpuF32 { height, .. } => *height, + } + } + + /// The pixel format: the BGRA8 slot wire format for the shm variant, + /// `PIXEL_FORMAT_F32` for the in-process variant. + pub fn format(&self) -> i32 { + match self { + RenderedFrame::Shm(_) => SLOT_FORMAT_BGRA8, + RenderedFrame::CpuF32 { .. } => PIXEL_FORMAT_F32, + } + } + + /// True for the process-backend shm variant. + pub fn is_shm(&self) -> bool { + matches!(self, RenderedFrame::Shm(_)) + } + + /// Build the viewer display image plus the scope samples (M15 S2 + /// zero-copy onscreen path). For the shm variant the slot's BGRA8 + /// bytes are wrapped into the display buffer — the GPU-upload staging + /// copy, the single permitted main-process copy on the preview path + /// (design §3.5). The caller releases the slot afterwards. + pub fn to_display(&self) -> Option<(RenderImage, ScopeData)> { + match self { + RenderedFrame::Shm(f) => { + let meta = &f.meta; + let (w, h) = (meta.width.max(0) as u32, meta.height.max(0) as u32); + let pixels = f.shm.slot_bytes(f.slot); + let data = pixels.get(..meta.data_size.max(0) as usize)?; + let image = bgra_bytes_to_render_image(w, h, data)?; + let scope = analyze_bgra8(w, h, data); + Some((image, scope)) + } + RenderedFrame::CpuF32 { + width, + height, + linesize, + data, + } => { + let (w, h) = ((*width).max(0) as u32, (*height).max(0) as u32); + let samples = repack_f32_rows(*width, *height, *linesize, data)?; + Some(( + f32_rgba_to_bgra_image(w, h, &samples), + analyze_f32_rgba(w, h, &samples), + )) + } + } + } +} + +/// Repack one F32 RGBA rendered frame (rows padded to `linesize`) into +/// tightly packed samples. Returns `(width, height, samples)`-style +/// samples only; geometry is validated by the caller. +fn repack_f32_rows(width: i32, height: i32, linesize: i32, data: &[u8]) -> Option> { + if width <= 0 || height <= 0 { + return None; + } + let row_bytes = (width * 4 * 4) as usize; + let linesize = (linesize as usize).max(row_bytes); + if data.len() < linesize * height as usize { + return None; + } + let mut samples = vec![0.0f32; (width * height * 4) as usize]; + for y in 0..height as usize { + let row = &data[y * linesize..y * linesize + row_bytes]; + for (i, px) in row.chunks_exact(4).enumerate() { + let v = f32::from_ne_bytes([px[0], px[1], px[2], px[3]]); + samples[y * (width as usize) * 4 + i] = v; + } + } + Some(samples) } /// The renderer geometry / format validation (the facade's @@ -256,7 +359,9 @@ fn validate_geometry(width: i32, height: i32, tb: (i64, i64)) -> Result<(), Stri Ok(()) } -/// Drive one video ticket synchronously. +/// Drive one video ticket synchronously (M15 S2: returns the payload +/// variant without copying frame bytes — the shm slot is read zero-copy +/// by the caller and released after building the display image). fn render_video(params: VideoTicketParams) -> Result { let m = RenderManager::global().ok_or_else(|| "render manager is not initialized".to_string())?; let id = m.tickets.next_id(); @@ -267,33 +372,31 @@ fn render_video(params: VideoTicketParams) -> Result { .result(id) .ok_or_else(|| "render ticket produced no result".to_string())?; match &result { - Ok(TicketPayload::Video(Texture::Cpu(frame))) => Ok(RenderedFrame { + Ok(TicketPayload::Video(Texture::Cpu(frame))) => Ok(RenderedFrame::CpuF32 { width: frame.width, height: frame.height, - format: frame.format as i32, linesize: frame.linesize_bytes() as i32, data: frame.data.clone(), }), + Ok(TicketPayload::ShmFrame(frame)) => Ok(RenderedFrame::Shm(frame.clone())), Ok(TicketPayload::Video(_)) => Err("render produced a non-CPU frame".to_string()), _ => Err("render produced no video frame".to_string()), } } -/// Render one frame of the sequence's montage at `frame_ts` (a timestamp -/// in the `(tb.1 / tb.0)`-per-frame timebase) into a `(width, height)` -/// F32 frame. -pub fn render_sequence_frame( +/// Build the video ticket params for one sequence-montage frame (M15 S2: +/// shared by the synchronous render and the playback pre-render window). +pub fn sequence_frame_params( p: &ProjectRef, seq: NodeId, frame_ts: i64, tb: (i64, i64), width: i32, height: i32, -) -> Result { +) -> Result { validate_geometry(width, height, tb)?; let time = Rational::new(frame_ts * tb.0, tb.1); - let montage = video_montage(p, seq, time); - render_video(VideoTicketParams { + Ok(VideoTicketParams { viewer: seq.identity(), time, force_size: Some((width, height)), @@ -303,7 +406,54 @@ pub fn render_sequence_frame( cache_id: None, cache_timebase: None, footage: None, - montage, + montage: video_montage(p, seq, time), + }) +} + +/// Render one frame of the sequence's montage at `frame_ts` (a timestamp +/// in the `(tb.1 / tb.0)`-per-frame timebase) into a `(width, height)` +/// frame. +pub fn render_sequence_frame( + p: &ProjectRef, + seq: NodeId, + frame_ts: i64, + tb: (i64, i64), + width: i32, + height: i32, +) -> Result { + render_video(sequence_frame_params(p, seq, frame_ts, tb, width, height)?) +} + +/// Build the video ticket params for one single-footage frame (M15 S2: +/// shared by the synchronous render and the source-monitor pre-render +/// window). +pub fn footage_frame_params( + p: &ProjectRef, + footage: NodeId, + frame_ts: i64, + tb: (i64, i64), + width: i32, + height: i32, +) -> Result { + validate_geometry(width, height, tb)?; + let (filename, stream_index) = { + let g = lock(p); + super::graphops::footage_behavior(&g.graph, footage) + .map(|f| preview_footage_media(f, true)) + .ok_or_else(|| "the node is not footage".to_string())? + }; + let time = Rational::new(frame_ts * tb.0, tb.1); + Ok(VideoTicketParams { + viewer: footage.identity(), + time, + force_size: Some((width, height)), + force_format: None, + cache: None, + cache_dir: None, + cache_id: None, + cache_timebase: None, + footage: Some((filename, stream_index)), + montage: Vec::new(), }) } @@ -317,26 +467,7 @@ pub fn render_footage_frame( width: i32, height: i32, ) -> Result { - validate_geometry(width, height, tb)?; - let (filename, stream_index) = { - let g = lock(p); - super::graphops::footage_behavior(&g.graph, footage) - .map(|f| preview_footage_media(f, true)) - .ok_or_else(|| "the node is not footage".to_string())? - }; - let time = Rational::new(frame_ts * tb.0, tb.1); - render_video(VideoTicketParams { - viewer: footage.identity(), - time, - force_size: Some((width, height)), - force_format: None, - cache: None, - cache_dir: None, - cache_id: None, - cache_timebase: None, - footage: Some((filename, stream_index)), - montage: Vec::new(), - }) + render_video(footage_frame_params(p, footage, frame_ts, tb, width, height)?) } /// Rendered interleaved f32 audio (the module audio ticket payload). diff --git a/src/oakui/scopes.rs b/src/oakui/scopes.rs index b6dbc42fc..3cc7b7472 100644 --- a/src/oakui/scopes.rs +++ b/src/oakui/scopes.rs @@ -77,6 +77,34 @@ pub(crate) fn analyze_f32_rgba(width: u32, height: u32, samples: &[f32]) -> Scop } } +/// Analyzes one BGRA8 frame (the process backend's slot format, M15 S2) +/// into its [`ScopeData`]. The worker converts its F32 pipeline output to +/// BGRA8 at the end of the render, so the scopes read exactly the +/// displayed values with the viewer's 8-bit quantization — precision loss +/// vs the F32 analysis is bounded by 1/255 per channel (acceptable for +/// the scopes; the F32 path stays for the in-process test backend). +/// `bytes` must hold at least `width * height * 4` values. +pub(crate) fn analyze_bgra8(width: u32, height: u32, bytes: &[u8]) -> ScopeData { + let pixels = (width * height) as usize; + let mut luma = Vec::with_capacity(pixels); + let mut chroma = Vec::with_capacity(pixels); + let src = bytes.get(..pixels * 4).unwrap_or_default(); + for px in src.chunks_exact(4) { + let b = f32::from(px[0]) / 255.0; + let g = f32::from(px[1]) / 255.0; + let r = f32::from(px[2]) / 255.0; + let y = KR * r + KG * g + KB * b; + let cb = 0.5 + (b - y) / (2.0 * (1.0 - KB)); + let cr = 0.5 + (r - y) / (2.0 * (1.0 - KR)); + luma.push(y); + chroma.push((cb.clamp(0.0, 1.0), cr.clamp(0.0, 1.0))); + } + ScopeData { + luma: Arc::new(luma), + chroma: Arc::new(chroma), + } +} + #[cfg(test)] mod tests { use super::*;