diff --git a/Cargo.lock b/Cargo.lock index a90fba29b..cc1ed17d8 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -4924,7 +4924,6 @@ dependencies = [ name = "oakundo" version = "0.1.0" dependencies = [ - "oakcore-rs", "thiserror 2.0.20", ] diff --git a/crates/oak-cli/src/engine.rs b/crates/oak-cli/src/engine.rs index 28643cf11..7b86a89d5 100644 --- a/crates/oak-cli/src/engine.rs +++ b/crates/oak-cli/src/engine.rs @@ -733,6 +733,19 @@ pub fn render_audio( sample_rate: samples.sample_rate, channel_count: samples.channel_count, }), + // M15 S3 process backend: the audio sits in a worker shm slot; copy + // it out and release the slot (the bytes must outlive the slot). + Ok(TicketPayload::ShmAudio(audio)) => { + let samples = audio.to_audio_samples(); + if let Some(m) = RenderManager::global() { + m.release_audio_frame(&audio); + } + Ok(RenderedAudio { + data: samples.samples, + sample_rate: samples.sample_rate, + channel_count: samples.channel_count, + }) + } _ => Err("audio render produced no samples".to_string()), } } diff --git a/crates/oak-worker/README.md b/crates/oak-worker/README.md index 3493d4680..f5ec9c7b8 100644 --- a/crates/oak-worker/README.md +++ b/crates/oak-worker/README.md @@ -23,7 +23,10 @@ dylib is needed at build or run time. snapshot through `oaknode::serializer`, and `render_frame` / `render_batch` render through `oakrender::eval` (generated frames, footage decode via oakcodec/ffmpeg, montage compositing) directly into - the main-assigned shared-memory slots. + the main-assigned shared-memory slots. M15 S3 adds `render_audio_batch` + (audio range pulls mixed by `oakrender::eval::render_audio_samples_into` + into `SLOT_FORMAT_AUDIO_F32` slots — interleaved f32 — so audio plugin + crashes take down this process, not the editor). - `src/ipc.rs` is a shim re-exporting `oakrender::ipc` (M15 S1): the NDJSON protocol and the shared-memory frame-slot transport moved to the oakrender crate so both ends of the pipe link one copy (the diff --git a/crates/oak-worker/src/worker.rs b/crates/oak-worker/src/worker.rs index 17a502f2c..d8b015b02 100644 --- a/crates/oak-worker/src/worker.rs +++ b/crates/oak-worker/src/worker.rs @@ -51,13 +51,13 @@ use serde_json::{json, Value}; use oakcore_rs::{PixelFormat, Rational}; use oakrender::backend::{BackendKind, DisplayRenderer}; use oakrender::eval; -use oakrender::ticket::{MontageClip, VideoTicketParams}; +use oakrender::ticket::{AudioTicketParams, MontageClip, VideoTicketParams}; use crate::ipc::{ - error_message, write_message, BatchTicketSpec, FrameSlotPool, FrameSlotMeta, HandshakeMsg, - LoadGraphMsg, RenderBatchMsg, RenderFrameMsg, SharedMemoryRegion, ShmMode, TYPE_CANCEL, - TYPE_HANDSHAKE, TYPE_LOAD_GRAPH, TYPE_RENDER_BATCH, TYPE_RENDER_FRAME, TYPE_SHUTDOWN, - SLOT_FORMAT_BGRA8, + error_message, write_message, AudioTicketSpec, BatchTicketSpec, FrameSlotPool, FrameSlotMeta, + HandshakeMsg, LoadGraphMsg, RenderAudioBatchMsg, RenderBatchMsg, RenderFrameMsg, + SharedMemoryRegion, ShmMode, TYPE_CANCEL, TYPE_HANDSHAKE, TYPE_LOAD_GRAPH, TYPE_RENDER_AUDIO_BATCH, + TYPE_RENDER_BATCH, TYPE_RENDER_FRAME, TYPE_SHUTDOWN, SLOT_FORMAT_AUDIO_F32, SLOT_FORMAT_BGRA8, }; use crate::{log_error, PROTOCOL_VERSION}; @@ -391,7 +391,7 @@ impl WorkerSession { Some(json!({ "type": crate::ipc::TYPE_HELLO_CAPS, "protocol_version": PROTOCOL_VERSION, - "formats": [PixelFormat::F32 as i32, SLOT_FORMAT_BGRA8], + "formats": [PixelFormat::F32 as i32, SLOT_FORMAT_BGRA8, SLOT_FORMAT_AUDIO_F32], "max_slot_bytes": hs.slot_data_bytes, })) } @@ -721,6 +721,164 @@ impl WorkerSession { } } + /// `render_audio_batch` (protocol v2, M15 S3): claim confirmation + /// followed by in-order mixing of every audio range pull into its + /// main-assigned shm slot (interleaved f32, wire format + /// [`SLOT_FORMAT_AUDIO_F32`]). Responses stream to `out`: one + /// `batch_accepted`, then one `frame_ready` or `frame_failed` per + /// ticket — the same claim/credit/frame_ready flow as + /// [`Self::handle_render_batch_stream`]. Crashes the process + /// deliberately when the crash-mode environment asks for it (the + /// audio crash-isolation test hook; the audio slot geometry check is + /// below the video one in `handle_line`). + fn handle_render_audio_batch_stream( + &mut self, + line: &str, + out: &mut impl Write, + ) -> io::Result<()> { + let batch: RenderAudioBatchMsg = match serde_json::from_str(line) { + Ok(b) => b, + Err(_) => { + return write_message(out, &error_message("invalid render_audio_batch message", None)) + } + }; + + let accepted = json!({ + "type": crate::ipc::TYPE_BATCH_ACCEPTED, + "batch_id": batch.batch_id, + "tickets": batch.tickets.iter().map(|t| t.ticket).collect::>(), + }); + write_message(out, &accepted)?; + out.flush()?; + + for spec in &batch.tickets { + self.maybe_crash_for_testing(spec.ticket); + + let response = match self.render_audio_ticket_to_slot(spec) { + Ok(slot) => json!({ + "type": crate::ipc::TYPE_FRAME_READY, + "ticket": spec.ticket, + "slot": slot, + }), + Err(e) => { + log_error(&format!( + "render_audio_batch: ticket {} failed: {e}", + spec.ticket + )); + json!({ + "type": crate::ipc::TYPE_FRAME_FAILED, + "ticket": spec.ticket, + "error": e, + }) + } + }; + write_message(out, &response)?; + out.flush()?; + } + Ok(()) + } + + /// Mix one audio range pull into its main-assigned slot (acquire, + /// mix, publish). Returns the published slot index. + fn render_audio_ticket_to_slot(&mut self, spec: &AudioTicketSpec) -> Result { + let pool = match self.output_pool.clone() { + Some(p) if p.is_valid() => p, + _ => return Err("no shared-memory pool attached".to_string()), + }; + + let acquired = self + .acquire_slot(&pool) + .ok_or_else(|| "no free shm slot (shutdown or timeout)".to_string())?; + if acquired != spec.slot as u32 { + return Err(format!( + "slot assignment mismatch: acquired {acquired}, assigned {}", + spec.slot + )); + } + let slot = acquired; + + let params = self.audio_ticket_params(spec)?; + let need = eval::audio_samples_byte_len(¶ms).map_err(|e| e.to_string())?; + if need > pool.slot_data_bytes() { + // The slot was acquired but never published; main recycles it on + // frame_failed (see render_ticket_to_slot). + return Err(format!( + "audio range needs {need} bytes, slot holds {}", + pool.slot_data_bytes() + )); + } + + // SAFETY: `slot` was acquired above; the block is live shared memory + // of the attached pool. + let dst = unsafe { + std::slice::from_raw_parts_mut( + pool.slot_data(spec.slot as u32), + pool.slot_data_bytes(), + ) + }; + eval::render_audio_samples_into(¶ms, &mut dst[..need]).map_err(|e| e.to_string())?; + + // SAFETY: slot in range of the attached pool. + unsafe { + let meta = &mut *pool.meta(spec.slot as u32); + *meta = FrameSlotMeta::default(); + meta.id = spec.ticket; + meta.time_num = spec.time_num; + meta.time_den = spec.time_den; + // Audio slots reuse the video meta fields: `width` carries the + // sample rate, `channel_count`/`linesize` describe the interleaved + // layout, `data_size` is the sample bytes (SLOT_FORMAT_AUDIO_F32). + meta.width = params.sample_rate; + meta.height = 0; + meta.format = SLOT_FORMAT_AUDIO_F32; + meta.channel_count = params.channel_layout.count_ones().max(1) as i32; + meta.linesize = meta.channel_count * 4; + meta.data_size = need as i32; + } + + let published = unsafe { pool.publish(slot) }; + if !published { + Err("ready ring full".to_string()) + } else { + Ok(slot) + } + } + + /// Map a wire audio ticket spec to the eval producer's audio params. + fn audio_ticket_params(&self, spec: &AudioTicketSpec) -> Result { + if spec.time_den <= 0 || spec.duration_den <= 0 || spec.sample_rate <= 0 { + return Err(format!( + "bad audio geometry: {}x{}, rate {}", + spec.time_den, spec.duration_den, spec.sample_rate + )); + } + let start = Rational::new(spec.time_num, spec.time_den); + let duration = Rational::new(spec.duration_num, spec.duration_den); + let montage: Vec = spec + .montage + .iter() + .map(|c| MontageClip { + filename: c.filename.clone(), + stream_index: c.stream_index, + in_time: Rational::new(c.in_num, c.in_den), + out_time: Rational::new(c.out_num, c.out_den), + media_in: Rational::new(c.media_in_num, c.media_in_den), + gain: c.gain, + }) + .collect(); + Ok(AudioTicketParams { + viewer: self + .graph + .as_ref() + .map(|g| g.project_copy) + .unwrap_or(0), + range: oakcore_rs::TimeRange::new(start, start + duration), + sample_rate: spec.sample_rate, + channel_layout: spec.channel_layout, + montage, + }) + } + /// Render `spec` into the slot's data block and fill the slot meta. fn render_spec_pixels(&mut self, spec: &BatchTicketSpec, pool: &FrameSlotPool) -> Result<(), String> { let w = spec.width; @@ -955,18 +1113,29 @@ pub fn worker_main(backend: &str) -> i32 { } // Protocol v2: render_batch streams its responses (one // batch_accepted + one frame_ready/frame_failed per ticket). - if serde_json::from_str::(&line) + let typ = serde_json::from_str::(&line) .ok() - .and_then(|v| v.get("type").and_then(Value::as_str).map(str::to_string)) - .as_deref() - == Some(TYPE_RENDER_BATCH) - { - if let Err(e) = session.handle_render_batch_stream(&line, &mut out) { - log_error(&format!("failed to serve render_batch: {e}")); - exit_code = 1; - break; + .and_then(|v| v.get("type").and_then(Value::as_str).map(str::to_string)); + match typ.as_deref() { + Some(TYPE_RENDER_BATCH) => { + if let Err(e) = session.handle_render_batch_stream(&line, &mut out) { + log_error(&format!("failed to serve render_batch: {e}")); + exit_code = 1; + break; + } + continue; } - continue; + // M15 S3: render_audio_batch streams the same way (audio range + // pulls into shm slots, interleaved f32). + Some(TYPE_RENDER_AUDIO_BATCH) => { + if let Err(e) = session.handle_render_audio_batch_stream(&line, &mut out) { + log_error(&format!("failed to serve render_audio_batch: {e}")); + exit_code = 1; + break; + } + continue; + } + _ => {} } if let Some(response) = session.handle_line(&line) { if let Err(e) = write_message(&mut out, &response) { @@ -1430,6 +1599,145 @@ mod tests { assert!(!unsafe { parent_pool.consume(&mut consumed) }); } + #[test] + fn render_audio_batch_stream_mixes_silence_into_slot() { + // M15 S3: an empty-montage audio range pull (1/24 s at 48 kHz + // stereo = 2000 frames x 2 ch = 16000 bytes) renders total silence + // into the assigned slot and reports frame_ready with the audio + // slot meta. + let mut s = WorkerSession::create("none").unwrap(); + let slot_bytes = 2000 * 2 * 4; + let (hs, out_region, _in) = parent_side(4, slot_bytes as i64, false); + let resp = s.handle_line(&hs.to_string()).expect("hello_caps response"); + assert_eq!(resp["type"], crate::ipc::TYPE_HELLO_CAPS); + + let batch = json!({ + "type": "render_audio_batch", + "batch_id": 3, + "tickets": [ + { + "ticket": 11, + "slot": 0, + "time_num": 0, + "time_den": 1, + "duration_num": 1, + "duration_den": 24, + "sample_rate": 48000, + "channel_layout": 0x3, + "channels": 2, + }, + { + "ticket": 12, + "slot": 1, + "time_num": 0, + "time_den": 1, + "duration_num": 1, + "duration_den": 24, + "sample_rate": 0, + "channel_layout": 0x3, + "channels": 2, + }, + ], + }); + let mut out: Vec = Vec::new(); + s.handle_render_audio_batch_stream(&batch.to_string(), &mut out) + .unwrap(); + let lines: Vec = String::from_utf8(out) + .unwrap() + .lines() + .map(|l| serde_json::from_str(l).unwrap()) + .collect(); + assert_eq!(lines.len(), 3, "accepted + one reply per ticket"); + assert_eq!(lines[0]["type"], "batch_accepted"); + assert_eq!(lines[0]["batch_id"], 3); + assert_eq!(lines[1]["type"], "frame_ready"); + assert_eq!(lines[1]["ticket"], 11); + assert_eq!(lines[1]["slot"], 0); + assert_eq!(lines[2]["type"], "frame_failed"); + assert_eq!(lines[2]["ticket"], 12); + + // The rendered slot holds the audio meta + silent samples. + let parent_pool = unsafe { FrameSlotPool::attach(out_region.data()) }; + let mut consumed = 0; + assert!(unsafe { parent_pool.consume(&mut consumed) }); + assert_eq!(consumed, 0); + let meta = unsafe { &*parent_pool.meta_const(consumed) }; + assert_eq!(meta.id, 11); + assert_eq!(meta.format, SLOT_FORMAT_AUDIO_F32); + assert_eq!(meta.width, 48000, "width carries the sample rate"); + assert_eq!(meta.height, 0); + assert_eq!(meta.channel_count, 2); + assert_eq!(meta.linesize, 2 * 4); + assert_eq!(meta.data_size, slot_bytes as i32); + let data = + unsafe { std::slice::from_raw_parts(parent_pool.slot_data_const(consumed), slot_bytes) }; + assert!(data.iter().all(|&b| b == 0), "empty montage is silence"); + // The samples parse back as 2000 stereo frames. + let parsed: Vec = data + .chunks_exact(4) + .map(|c| f32::from_le_bytes([c[0], c[1], c[2], c[3]])) + .collect(); + assert_eq!(parsed.len(), 2000 * 2); + assert!(parsed.iter().all(|&v| v == 0.0)); + unsafe { parent_pool.release(consumed) }; + + // The failed ticket (bad sample rate) acquired slot 1 but never + // published it. + assert!(!unsafe { parent_pool.consume(&mut consumed) }); + } + + #[test] + fn render_audio_batch_rejects_oversized_range() { + // A range longer than the slot can hold must fail with frame_failed + // (never a buffer overflow into the next slot). + let mut s = WorkerSession::create("none").unwrap(); + // Slot sized for 1/48 s of stereo audio (2000 bytes). + let (hs, out_region, _in) = parent_side(4, 2000, false); + let resp = s.handle_line(&hs.to_string()).expect("hello_caps response"); + assert_eq!(resp["type"], crate::ipc::TYPE_HELLO_CAPS); + + let batch = json!({ + "type": "render_audio_batch", + "batch_id": 4, + "tickets": [ + { + "ticket": 21, + "slot": 0, + "time_num": 0, + "time_den": 1, + "duration_num": 1, + "duration_den": 24, + "sample_rate": 48000, + "channel_layout": 0x3, + "channels": 2, + }, + ], + }); + let mut out: Vec = Vec::new(); + s.handle_render_audio_batch_stream(&batch.to_string(), &mut out) + .unwrap(); + let lines: Vec = String::from_utf8(out) + .unwrap() + .lines() + .map(|l| serde_json::from_str(l).unwrap()) + .collect(); + assert_eq!(lines[0]["type"], "batch_accepted"); + assert_eq!(lines[1]["type"], "frame_failed"); + assert_eq!(lines[1]["ticket"], 21); + assert!( + lines[1]["error"] + .as_str() + .unwrap() + .contains("needs 16000 bytes"), + "oversized range reported: {}", + lines[1]["error"] + ); + // Slot 0 acquired but never published. + let parent_pool = unsafe { FrameSlotPool::attach(out_region.data()) }; + let mut consumed = 0; + assert!(!unsafe { parent_pool.consume(&mut consumed) }); + } + // ---- oak-worker's in-process session tests (M14 R2: folded from the // ---- former src/session.rs mirror; the facade's production session // ---- tests above cover the rest) ------------------------------------- diff --git a/crates/oak-worker/tests/procpool_integration.rs b/crates/oak-worker/tests/procpool_integration.rs index 7ff32af79..2c628654a 100644 --- a/crates/oak-worker/tests/procpool_integration.rs +++ b/crates/oak-worker/tests/procpool_integration.rs @@ -30,12 +30,14 @@ use std::sync::{Arc, Mutex}; use std::time::{Duration, Instant}; -use oakcore_rs::{PixelFormat, Rational}; +use oakcore_rs::{PixelFormat, Rational, TimeRange}; use oakrender::ipc::SLOT_FORMAT_BGRA8; use oakrender::procpool::{ main_heap_frame_copies, reset_main_heap_frame_copies, DispatcherConfig, ProcessDispatcher, }; -use oakrender::ticket::{TicketPayload, TicketResult, VideoTicketParams}; +use oakrender::ticket::{ + AudioTicketParams, TicketPayload, TicketResult, VideoTicketParams, +}; use oakrender::worker::{Job, JobDispatch, JobSchedule}; /// Serialize every test in this file (shared process environment + @@ -69,7 +71,10 @@ fn params(time: Rational, footage: Option<(String, i32)>) -> Arc>>, + viewer: u64, + start: Rational, + duration: Rational, +) { + let range = TimeRange::new(start, start + duration); + let audio = Arc::new(AudioTicketParams { + viewer, + range, + sample_rate: 48000, + channel_layout: 0x3, + montage: Vec::new(), + }); + let results = results.clone(); + let job = Job { + node_identity: viewer, + time: start, + params: Arc::new(VideoTicketParams { + viewer, + time: start, + force_size: None, + force_format: None, + cache: None, + cache_dir: None, + cache_id: None, + cache_timebase: None, + footage: None, + montage: Vec::new(), + }), + audio: Some(audio), + // Never invoked on the process backend (the worker renders audio + // from the wire spec); must still be a valid producer. + produce: Arc::new(|_, _| { + Err(oakrender::error::Error::Failed( + "process backend does not use the in-process producer".into(), + )) + }), + 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"); +} + +/// Audio tickets flow through the worker pool into shm slots (M15 S3): +/// empty-montage ranges render as interleaved f32 silence, the main +/// process reads them back with `ShmAudioRef::samples()` (zero copy — +/// the copy counter stays 0) and releases the slots. +#[test] +fn audio_tickets_roundtrip_through_shm_slots() { + let _guard = lock_test(); + let dispatcher = ProcessDispatcher::new(config(2, 4)).expect("dispatcher config"); + dispatcher.start().expect("workers start"); + + reset_main_heap_frame_copies(); + + let results = Arc::new(Mutex::new(Vec::new())); + // 1/24 s at 48 kHz stereo = 2000 sample frames x 2 ch = 16000 bytes — + // fits the 64x64 BGRA8 slot (16384 bytes). + for i in 0..4 { + submit_audio(&dispatcher, &results, 1, Rational::new(i, 24), Rational::new(1, 24)); + } + pump_until(&dispatcher, &results, 4); + + let mut seen_worker = [false; 2]; + for result in results.lock().unwrap().drain(..) { + let payload = result.expect("audio rendered"); + let TicketPayload::ShmAudio(audio) = payload else { + panic!("audio tickets must deliver ShmAudio payloads"); + }; + assert_eq!(audio.sample_rate, 48000); + assert_eq!(audio.channel_count, 2); + assert_eq!(audio.meta.format, oakrender::ipc::SLOT_FORMAT_AUDIO_F32); + assert_eq!(audio.meta.channel_count, 2); + assert_eq!(audio.meta.linesize, 2 * 4); + assert_eq!(audio.meta.data_size, 2000 * 2 * 4); + // Empty montage: total silence, parsed back as f32. + let samples = audio.samples(); + assert_eq!(samples.len(), 2000 * 2); + assert!(samples.iter().all(|&v| v == 0.0), "empty montage is silence"); + seen_worker[audio.worker as usize] = true; + dispatcher.release_audio_frame(&audio); + } + assert!(seen_worker[0] && seen_worker[1], "both workers rendered audio"); + // Samples are read via the slot mapping and parsed into a Vec — never + // through the counted `slot_to_vec` copy path. + assert_eq!(main_heap_frame_copies(), 0); + + dispatcher.shutdown(); +} + +/// An audio render crashing mid-mix (SIGSEGV hook) must not take down the +/// main process: the audio ticket is re-queued, the worker restarted and +/// the samples still arrive. +#[test] +fn audio_crash_isolation_restarts_worker_and_audio_still_renders() { + let _guard = lock_test(); + + let marker = std::env::temp_dir().join(format!( + "oak-procpool-audio-crash-marker-{}", + std::process::id() + )); + let _ = std::fs::remove_file(&marker); + std::env::set_var("OAK_WORKER_CRASH_ON_TICKET", "1"); + std::env::set_var("OAK_WORKER_CRASH_MARKER", &marker); + struct EnvGuard; + impl Drop for EnvGuard { + fn drop(&mut self) { + std::env::remove_var("OAK_WORKER_CRASH_ON_TICKET"); + std::env::remove_var("OAK_WORKER_CRASH_MARKER"); + } + } + let _env_guard = EnvGuard; + + let dispatcher = ProcessDispatcher::new(config(1, 4)).expect("dispatcher config"); + dispatcher.start().expect("worker starts"); + + let results = Arc::new(Mutex::new(Vec::new())); + // The dispatcher's first ticket id is 1 — exactly the crash ticket + // (an audio ticket this time). + for i in 0..4 { + submit_audio(&dispatcher, &results, 1, Rational::new(i, 24), Rational::new(1, 24)); + } + pump_until(&dispatcher, &results, 4); + + assert!( + dispatcher.restarts_of(0) >= 1, + "crashed audio worker must be restarted (restarts={})", + dispatcher.restarts_of(0) + ); + let mut crashed_ticket_seen = false; + for result in results.lock().unwrap().drain(..) { + let payload = result.expect("audio rendered despite the worker crash"); + let TicketPayload::ShmAudio(audio) = payload else { + panic!("ShmAudio payload"); + }; + if audio.meta.id == 1 { + crashed_ticket_seen = true; + assert_eq!(audio.sample_rate, 48000); + assert_eq!(audio.meta.data_size, 2000 * 2 * 4); + } + dispatcher.release_audio_frame(&audio); + } + assert!(crashed_ticket_seen, "ticket 1 delivered after the restart"); + assert!(marker.exists(), "the crash hook fired exactly once"); + let _ = std::fs::remove_file(&marker); + + dispatcher.shutdown(); +} + +/// An audio range too large for a practical shm slot (> 64 MB, i.e. a +/// multi-minute export) is refused by the dispatcher's `post`, so the +/// arena can fall back to main-process inline rendering (design §3.7) +/// instead of forcing a giant shared-memory segment. +#[test] +fn oversized_audio_ticket_is_refused_by_process_backend() { + let _guard = lock_test(); + let dispatcher = ProcessDispatcher::new(config(1, 4)).expect("dispatcher config"); + dispatcher.start().expect("worker starts"); + + // ~3 min of 48 kHz stereo > 64 MB: post refuses it (false). + let duration = Rational::new(175, 1); // 175 s + let results = Arc::new(Mutex::new(Vec::new())); + let results2 = results.clone(); + let audio = Arc::new(AudioTicketParams { + viewer: 1, + range: TimeRange::new(Rational::new(0, 1), duration), + sample_rate: 48000, + channel_layout: 0x3, + montage: Vec::new(), + }); + let job = Job { + node_identity: 1, + time: Rational::new(0, 1), + params: Arc::new(VideoTicketParams { + viewer: 1, + time: Rational::new(0, 1), + force_size: None, + force_format: None, + cache: None, + cache_dir: None, + cache_id: None, + cache_timebase: None, + footage: None, + montage: Vec::new(), + }), + audio: Some(audio), + produce: Arc::new(|_, _| { + Err(oakrender::error::Error::Failed( + "process backend does not use the in-process producer".into(), + )) + }), + done: Box::new(move |_| { + results2.lock().unwrap_or_else(|e| e.into_inner()).push(()); + }), + schedule: JobSchedule::seek(), + }; + assert!( + !dispatcher.post(job), + "oversized audio must be refused so the arena falls back inline" + ); + assert_eq!(results.lock().unwrap().len(), 0, "no completion fires"); + + dispatcher.shutdown(); +} +/// Per-ticket slot formats (M15 S3): a forced-F32 ticket gets an F32 slot +/// (the segment grows on demand) while the default BGRA8 tickets keep +/// BGRA8 slots — the export path's F32 request no longer round-trips +/// through BGRA8. +#[test] +fn f32_ticket_gets_f32_slot_and_bgra8_stays_bgra8() { + let _guard = lock_test(); + let dispatcher = ProcessDispatcher::new(config(1, 4)).expect("dispatcher config"); + dispatcher.start().expect("worker starts"); + + // 1. Default BGRA8 preview ticket first. + let results = Arc::new(Mutex::new(Vec::new())); + submit(&dispatcher, &results, 1, None); + pump_until(&dispatcher, &results, 1); + let bg = results.lock().unwrap().pop().unwrap().expect("frame rendered"); + let TicketPayload::ShmFrame(frame) = bg else { + panic!("ShmFrame payload"); + }; + assert_eq!(frame.meta.format, SLOT_FORMAT_BGRA8); + assert_eq!(frame.meta.data_size, 64 * 64 * 4); + dispatcher.release_frame(&frame); + + // 2. Forced-F32 ticket: the segment grows on demand and the slot holds + // F32 RGBA (16 bytes per pixel). + let results2 = Arc::new(Mutex::new(Vec::new())); + { + let results2 = results2.clone(); + let job = Job { + node_identity: 1, + time: Rational::new(0, 1), + params: Arc::new(VideoTicketParams { + viewer: 1, + time: Rational::new(0, 1), + force_size: Some((64, 64)), + force_format: Some(PixelFormat::F32), + cache: None, + cache_dir: None, + cache_id: None, + cache_timebase: None, + footage: None, + montage: Vec::new(), + }), + audio: None, + produce: Arc::new(|_, _| { + Err(oakrender::error::Error::Failed( + "process backend does not use the in-process producer".into(), + )) + }), + done: Box::new(move |result| { + results2.lock().unwrap_or_else(|e| e.into_inner()).push(result); + }), + schedule: JobSchedule::seek(), + }; + assert!(dispatcher.post(job), "f32 post accepted"); + } + pump_until(&dispatcher, &results2, 1); + let f32res = results2.lock().unwrap().pop().unwrap().expect("f32 frame rendered"); + let TicketPayload::ShmFrame(frame) = f32res else { + panic!("ShmFrame payload"); + }; + assert_eq!(frame.meta.format, PixelFormat::F32 as i32); + assert_eq!(frame.meta.data_size, 64 * 64 * 16); + // The F32 bytes are zero (transparent black pipeline output). + let pixels = frame.shm.slot_bytes(frame.slot); + assert!( + pixels[..frame.meta.data_size as usize].iter().all(|&b| b == 0), + "generated F32 frame is transparent black" + ); + dispatcher.release_frame(&frame); + + // 3. A later default BGRA8 ticket still lands as BGRA8 in the grown + // segment (the wire format is per ticket). + submit(&dispatcher, &results, 1, None); + pump_until(&dispatcher, &results, 1); + let bg2 = results.lock().unwrap().pop().unwrap().expect("frame rendered"); + let TicketPayload::ShmFrame(frame) = bg2 else { + panic!("ShmFrame payload"); + }; + assert_eq!(frame.meta.format, SLOT_FORMAT_BGRA8); + assert_eq!(frame.meta.data_size, 64 * 64 * 4); + dispatcher.release_frame(&frame); + + dispatcher.shutdown(); +} diff --git a/crates/oakrender/README.md b/crates/oakrender/README.md index cf764f1b4..f25894fe6 100644 --- a/crates/oakrender/README.md +++ b/crates/oakrender/README.md @@ -70,16 +70,18 @@ src/ color.rs ColorProcessor over ocio-rs + default config + LUT library manager.rs RenderManager singleton + lifecycle + disk cache ticket.rs Ticket arena, params, exactly-once completion delivery - worker.rs Worker pool + frozen pre-M15 ProcessPool facade stub + - graph snapshot store + worker.rs JobDispatch seam + thread-free InlineDispatcher (audio + fallback / test backend) + graph snapshot store scheduler.rs M15 PreviewScheduler: interleaved shard claiming, - priority lanes (seek/playback/background), crash reclaim + priority lanes (seek/playback/background), crash reclaim, + per-request slot-bytes capacity filtering (S3) ipc.rs M15 render-worker IPC (moved from oak-worker): NDJSON control protocol (v1 + v2 messages) + the POSIX shared-memory frame-slot transport both pipe ends link procpool.rs M15 ProcessDispatcher: spawn/handshake oak-worker processes, main-assigned slot batches, crash detection + - restart, zero-copy ShmFrameRef completions + restart, zero-copy ShmFrameRef / ShmAudioRef completions, + grow-on-demand segment geometry (S3) autocacher.rs PreviewAutoCacher eval.rs RenderHooks impl: the CPU evaluation seam backend.rs wgpu device/queue/texture management + DisplayRenderer @@ -126,13 +128,24 @@ tests/ contract + golden tests (common/ has shared helpers) 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. + to read the shm slots zero-copy. M15 S3 added **audio over shm** + (`render_audio_batch` → `TicketPayload::ShmAudio`, with the + main-process inline `InlineDispatcher` as the fallback when the + process dispatcher is unavailable), **per-ticket slot formats** + (forced-F32 tickets get F32 slots; the export path reads them + directly, eliminating the BGRA8→F32 round trip; segments grow on + demand), and **adaptive pool tuning** (`default_slots_per_worker` / + `default_batch_size` / `default_worker_count` policies plus the + `bench_process` example that reports 1080p throughput and + adjacent-frame completion deltas). +- **Audio rendering** — M15 S3 migrated audio tickets onto the process + dispatcher (`render_audio_batch`; the worker mixes via + `eval::render_audio_samples_into` into `SLOT_FORMAT_AUDIO_F32` slots; + the main process reads `ShmAudioRef` and releases). The app's + real-time pull (`pull_audio_tick`) renders chunks **ahead** + asynchronously (prefetch depth 4 ≈ 66 ms) so it never blocks on a + busy worker; `worker::InlineDispatcher::sync` remains the fallback + (design §3.7). - **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/examples/bench_process.rs b/crates/oakrender/examples/bench_process.rs new file mode 100644 index 000000000..8a9f79944 --- /dev/null +++ b/crates/oakrender/examples/bench_process.rs @@ -0,0 +1,210 @@ +// Oak Video Editor - Non-Linear Video Editor +// Copyright (C) 2026 Oak Team +// +// This program is free software: you can redistribute it and/or modify +// it under the terms of the GNU General Public License as published by +// the Free Software Foundation, either version 3 of the License, or +// (at your option) any later version. +// +// This program is distributed in the hope that it will be useful, +// but WITHOUT ANY WARRANTY; without even the implied warranty of +// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +// GNU General Public License for more details. +// +// You should have received a copy of the GNU General Public License +// along with this program. If not, see . + +//! M15 S3 process-pool frame-throughput benchmark (design §3.2 scheduling +//! metrics): renders `N` 1080p BGRA8 generated frames through the real +//! oak-worker pool and reports +//! +//! - **throughput**: total frames / wall time (frames per second); +//! - **adjacent-frame completion delta**: for each pair of adjacent +//! frame numbers `(f, f+1)`, `|completion(f+1) - completion(f)|` — +//! the design doc's "相邻帧完成时间差", which the interleaved +//! batch-claim scheduler keeps bounded because adjacent frames land on +//! different workers. Reported as max / mean / p95 / count. +//! +//! Run from the repo root: +//! +//! ```sh +//! cargo run --release -p oakrender --example bench_process [frames] [workers] +//! ``` +//! +//! `frames` defaults to 240, `workers` to the adaptive +//! [`oakrender::procpool::default_worker_count`] policy. The oak-worker +//! binary is located next to the build output (`target//oak-worker`), +//! or via `$OAK_WORKER_BIN`. + +use std::path::PathBuf; +use std::sync::{Arc, Mutex}; +use std::time::{Duration, Instant}; + +use oakcore_rs::Rational; +use oakrender::ipc::SLOT_FORMAT_BGRA8; +use oakrender::procpool::{DispatcherConfig, ProcessDispatcher}; +use oakrender::ticket::{TicketPayload, TicketResult, VideoTicketParams}; +use oakrender::worker::{Job, JobDispatch, JobSchedule}; + +/// Locate the oak-worker binary: `$OAK_WORKER_BIN`, else the sibling of +/// the current executable's `examples/` directory, else `oak-worker` on +/// `PATH`. +fn worker_bin() -> PathBuf { + if let Ok(p) = std::env::var("OAK_WORKER_BIN") { + return PathBuf::from(p); + } + if let Ok(exe) = std::env::current_exe() { + // target//examples/bench_process -> target//oak-worker + if let Some(examples) = exe.parent() { + if let Some(profile) = examples.parent() { + let candidate = + profile.join(format!("oak-worker{}", std::env::consts::EXE_SUFFIX)); + if candidate.exists() { + return candidate; + } + } + } + } + PathBuf::from("oak-worker") +} + +fn main() { + let frames: usize = std::env::args() + .nth(1) + .and_then(|s| s.parse().ok()) + .unwrap_or(240); + let workers: Option = std::env::args().nth(2).and_then(|s| s.parse().ok()); + let width: i32 = 1920; + let height: i32 = 1080; + + let config = DispatcherConfig { + worker_bin: Some(worker_bin()), + workers: workers.unwrap_or(0), + slots_per_worker: 8, + width, + height, + slot_format: SLOT_FORMAT_BGRA8, + batch_size: 0, + graph_snapshot: None, + handshake_timeout_ms: 30_000, + }; + let dispatcher = ProcessDispatcher::new(config).expect("dispatcher config"); + dispatcher.start().expect("workers start + handshake"); + let worker_count = dispatcher.worker_count(); + println!( + "oak-worker pool: {worker_count} worker(s), {frames} x {width}x{height} BGRA8 frames" + ); + + // One completion record per frame: (frame number, wall-clock completion). + let results = Arc::new(Mutex::new(Vec::<(i64, Instant)>::new())); + let start = Instant::now(); + let dispatcher_for_closure = dispatcher.clone(); + for i in 0..frames { + let results = results.clone(); + let dc = dispatcher_for_closure.clone(); + let frame = i as i64; + let job = Job { + node_identity: 1, + time: Rational::new(frame, 25), + params: Arc::new(VideoTicketParams { + viewer: 1, + time: Rational::new(frame, 25), + force_size: Some((width, height)), + force_format: None, + cache: None, + cache_dir: None, + cache_id: None, + cache_timebase: None, + footage: None, + montage: Vec::new(), + }), + audio: None, + // Never invoked on the process backend (workers render from the + // wire spec); must still be a valid producer. + produce: Arc::new(|_, _| { + Err(oakrender::error::Error::Failed( + "process backend does not use the in-process producer".into(), + )) + }), + done: Box::new(move |result: TicketResult| match result { + Ok(TicketPayload::ShmFrame(f)) => { + results + .lock() + .unwrap_or_else(|e| e.into_inner()) + .push((f.meta.id, Instant::now())); + // Slot release = credit (cache eviction): without it the + // worker's free slots never come back and the pool + // stalls at slots-per-worker frames. + dc.release_frame(&f); + } + Ok(TicketPayload::ShmAudio(a)) => { + dc.release_audio_frame(&a); + } + Ok(other) => { + eprintln!("unexpected payload: {other:?}"); + } + Err(e) => { + eprintln!("frame {frame} failed: {e}"); + } + }), + // Playback priority: the pre-render window schedule (seek/playback + // prioritization is what the scheduler benchmark measures). + schedule: JobSchedule::playback(frame, 0, 0), + }; + if !dispatcher.post(job) { + eprintln!("post refused at frame {frame}"); + break; + } + } + + // Pump until every completion has landed. + let deadline = Instant::now() + Duration::from_secs(120); + loop { + dispatcher.poll(); + let done = results.lock().unwrap_or_else(|e| e.into_inner()).len(); + if done >= frames { + break; + } + if Instant::now() > deadline { + eprintln!("timeout: {done}/{frames} completions"); + break; + } + std::thread::sleep(Duration::from_millis(2)); + } + let elapsed = start.elapsed(); + + let mut entries: Vec<(i64, Instant)> = results.lock().unwrap_or_else(|e| e.into_inner()).drain(..).collect(); + entries.sort_by_key(|(id, _)| *id); + let completed = entries.len(); + let throughput = completed as f64 / elapsed.as_secs_f64(); + + // Adjacent-frame completion delta: |t(f+1) - t(f)| over pairs that + // completed in order. With interleaved claiming these stay small. + let mut deltas: Vec = Vec::new(); + for w in entries.windows(2) { + if w[1].0 == w[0].0 + 1 { + deltas.push((w[1].1 - w[0].1).as_secs_f64().abs()); + } + } + deltas.sort_by(|a, b| a.partial_cmp(b).unwrap()); + + let report = |name: &str, value: String| { + println!("{name:<38} {value}"); + }; + report("frames completed", completed.to_string()); + report("total wall time", format!("{:.2} s", elapsed.as_secs_f64())); + report("throughput", format!("{throughput:.1} fps ({:.1} ms/frame)", 1000.0 / throughput)); + if !deltas.is_empty() { + let mean = deltas.iter().sum::() / deltas.len() as f64; + let p95 = deltas[((deltas.len() as f64 * 0.95) as usize).min(deltas.len() - 1)]; + report("adjacent-frame delta (pairs)", deltas.len().to_string()); + report(" max", format!("{:.3} ms", deltas.last().unwrap() * 1000.0)); + report(" mean", format!("{:.3} ms", mean * 1000.0)); + report(" p95", format!("{:.3} ms", p95 * 1000.0)); + } else { + report("adjacent-frame delta", "no adjacent pairs completed".to_string()); + } + report("main-heap frame copies", oakrender::procpool::main_heap_frame_copies().to_string()); + + dispatcher.shutdown(); +} diff --git a/crates/oakrender/src/eval.rs b/crates/oakrender/src/eval.rs index 1dcc8367d..62fe1bd95 100644 --- a/crates/oakrender/src/eval.rs +++ b/crates/oakrender/src/eval.rs @@ -553,6 +553,20 @@ pub fn render_footage_frame( pub fn render_audio_samples( params: &crate::ticket::AudioTicketParams, ) -> Result { + let (rate, layout, channels, total_frames) = audio_layout(params)?; + let mut acc = vec![0.0f32; total_frames.saturating_mul(channels as usize)]; + mix_audio_montage(params, rate, channels, total_frames, &mut acc)?; + Ok(crate::ticket::TicketPayload::Audio(crate::ticket::AudioSamples { + samples: acc, + sample_rate: rate, + channel_layout: layout, + channel_count: channels, + })) +} + +/// The output layout an audio render produces: `(sample_rate, +/// channel_layout, channel_count, total_sample_frames)`. +fn audio_layout(params: &crate::ticket::AudioTicketParams) -> Result<(i32, u64, i32, usize)> { let rate = params.sample_rate.max(1); let channels = params.channel_layout.count_ones().max(1) as i32; let duration = params.range.out() - params.range.in_(); @@ -565,8 +579,30 @@ pub fn render_audio_samples( return Err(Error::Invalid); } let total_frames = (seconds * rate as f64).round() as usize; - let mut acc = vec![0.0f32; total_frames.saturating_mul(channels as usize)]; + Ok((rate, params.channel_layout, channels, total_frames)) +} +/// The byte length (interleaved f32, little-endian) an audio render of +/// `params` writes into a shm slot — the worker's slot-geometry check +/// (M15 S3). Mirrors [`render_audio_samples_into`]'s layout math. +pub fn audio_samples_byte_len(params: &crate::ticket::AudioTicketParams) -> Result { + let (_rate, _layout, channels, total_frames) = audio_layout(params)?; + Ok(total_frames + .saturating_mul(channels as usize) + .saturating_mul(4)) +} + +/// Mix the audio montage into `acc` (`total_frames * channels` samples, +/// zero-initialized by the caller). Shared by the heap +/// [`render_audio_samples`] and the shm-slot [`render_audio_samples_into`] +/// paths so the decode/mix logic exists once. +fn mix_audio_montage( + params: &crate::ticket::AudioTicketParams, + rate: i32, + channels: i32, + total_frames: usize, + acc: &mut [f32], +) -> Result<()> { for clip in ¶ms.montage { // Overlap of the clip with the requested range. let in_time = params.range.in_().max(clip.in_time); @@ -603,13 +639,32 @@ pub fn render_audio_samples( acc[start_frame * channels as usize + i] += buf[i] * clip.gain; } } + Ok(()) +} - Ok(crate::ticket::TicketPayload::Audio(crate::ticket::AudioSamples { - samples: acc, - sample_rate: rate, - channel_layout: params.channel_layout, - channel_count: channels, - })) +/// Render the audio montage over `params.range` directly into `dst` as +/// little-endian f32 bytes (M15 S3 worker seam): the render worker passes +/// a shared-memory slot slice as `dst`, so the samples land in the slot +/// with no staging allocation. `dst.len()` must hold +/// `frame_count * channels * 4` bytes. +pub fn render_audio_samples_into( + params: &crate::ticket::AudioTicketParams, + dst: &mut [u8], +) -> Result<()> { + let (rate, _layout, channels, total_frames) = audio_layout(params)?; + let need = total_frames + .saturating_mul(channels as usize) + .saturating_mul(4); + if dst.len() < need { + return Err(Error::NoMem); + } + let mut acc = vec![0.0f32; total_frames.saturating_mul(channels as usize)]; + mix_audio_montage(params, rate, channels, total_frames, &mut acc)?; + // Interleaved f32 -> little-endian bytes in the slot. + for (out, sample) in dst[..need].chunks_exact_mut(4).zip(&acc) { + out.copy_from_slice(&sample.to_le_bytes()); + } + Ok(()) } /// Bilinear scale an F32-RGBA image (row-major with per-row strides). @@ -1048,4 +1103,66 @@ mod tests { } } use std::sync::Arc; + + // ---- Audio (M12 P1 / M15 S3) ----------------------------------------- + + fn audio_params(range: TimeRange) -> crate::ticket::AudioTicketParams { + crate::ticket::AudioTicketParams { + viewer: 1, + range, + sample_rate: 48000, + channel_layout: 0x3, + montage: Vec::new(), + } + } + + #[test] + fn render_audio_samples_produces_silence_for_empty_montage() { + // M12 P1: an empty montage renders total silence at the requested + // layout. + let params = audio_params(TimeRange::new(Rational::new(0, 1), Rational::new(1, 24))); + match render_audio_samples(¶ms).unwrap() { + crate::ticket::TicketPayload::Audio(samples) => { + // 1/24 s at 48 kHz = 2000 sample frames, stereo. + assert_eq!(samples.sample_rate, 48000); + assert_eq!(samples.channel_count, 2); + assert_eq!(samples.samples.len(), 2000 * 2); + assert!(samples.samples.iter().all(|&v| v == 0.0), "silence"); + } + other => panic!("expected Audio payload, got {other:?}"), + } + } + + #[test] + fn render_audio_samples_into_matches_heap_path_byte_for_byte() { + // M15 S3: the shm-slot writer must produce exactly the same + // little-endian f32 bytes as the heap path, so a worker's slot and + // the in-process fallback agree for the same montage. + let params = audio_params(TimeRange::new(Rational::new(0, 1), Rational::new(1, 48))); + let heap = match render_audio_samples(¶ms).unwrap() { + crate::ticket::TicketPayload::Audio(samples) => samples, + other => panic!("expected Audio payload, got {other:?}"), + }; + let mut dst = vec![0u8; heap.samples.len() * 4]; + render_audio_samples_into(¶ms, &mut dst).unwrap(); + let expected: Vec = heap + .samples + .iter() + .flat_map(|v| v.to_le_bytes()) + .collect(); + assert_eq!(dst, expected); + // And the into-path output parses back into the same samples. + let parsed: Vec = dst + .chunks_exact(4) + .map(|c| f32::from_le_bytes([c[0], c[1], c[2], c[3]])) + .collect(); + assert_eq!(parsed, heap.samples); + } + + #[test] + fn render_audio_samples_into_rejects_small_buffer() { + let params = audio_params(TimeRange::new(Rational::new(0, 1), Rational::new(1, 24))); + let mut dst = [0u8; 8]; // far too small for 2000x2 f32 samples + assert!(render_audio_samples_into(¶ms, &mut dst).is_err()); + } } diff --git a/crates/oakrender/src/ipc.rs b/crates/oakrender/src/ipc.rs index 37db21cc0..ab1e55b27 100644 --- a/crates/oakrender/src/ipc.rs +++ b/crates/oakrender/src/ipc.rs @@ -116,6 +116,9 @@ pub const TYPE_RENDER_BATCH: &str = "render_batch"; pub const TYPE_BATCH_ACCEPTED: &str = "batch_accepted"; /// `"frame_failed"` (protocol v2). pub const TYPE_FRAME_FAILED: &str = "frame_failed"; +/// `"render_audio_batch"` (protocol v2, M15 S3): a batch of audio range +/// pulls rendered into the same shm slot transport as video frames. +pub const TYPE_RENDER_AUDIO_BATCH: &str = "render_audio_batch"; /// Wire-format slot format for 8-bit BGRA frames (M15 S1). The viewer /// preview path requests BGRA8 so the worker converts its F32 pipeline @@ -125,6 +128,15 @@ pub const TYPE_FRAME_FAILED: &str = "frame_failed"; /// slot wire format, not a pipeline format. pub const SLOT_FORMAT_BGRA8: i32 = 100; +/// Wire-format slot format for interleaved f32 audio samples (M15 S3). +/// An audio slot reuses [`FrameSlotMeta`]: `format` is this marker, +/// `channel_count` is the channel count, `linesize` is +/// `channels * 4` (bytes per sample frame), `data_size` is the total +/// sample bytes and `width` carries the output sample rate (Hz). The +/// value lives outside the `PixelFormat` enum range like +/// [`SLOT_FORMAT_BGRA8`]. +pub const SLOT_FORMAT_AUDIO_F32: i32 = 101; + /// `handshake` — field-for-field equivalent of `oak_ipc_handshake` /// (ipc.h). Wire field names match the C++ serializer. #[derive(Serialize, Deserialize, Default, Debug, Clone)] @@ -323,6 +335,49 @@ pub struct BatchAcceptedMsg { pub tickets: Vec, } +/// One audio range pull inside a [`RenderAudioBatchMsg`] (M15 S3) — the +/// audio counterpart of [`BatchTicketSpec`]: the main process assigns the +/// destination `slot`, the worker mixes the montage over `[time, +/// time + duration)` at `sample_rate`/`channel_layout` into interleaved +/// f32 and writes it into the slot (wire format +/// [`SLOT_FORMAT_AUDIO_F32`]). +#[derive(Serialize, Deserialize, Default, Debug, Clone, PartialEq)] +#[serde(default)] +pub struct AudioTicketSpec { + /// Ticket id (correlates with frame_ready / frame_failed). + pub ticket: i64, + /// Destination slot index in the worker->main output pool. + pub slot: i32, + /// Range start numerator. + pub time_num: i64, + /// Range start denominator. + pub time_den: i64, + /// Range length numerator. + pub duration_num: i64, + /// Range length denominator. + pub duration_den: i64, + /// Output sample rate (Hz). + pub sample_rate: i32, + /// Output channel layout mask. + pub channel_layout: u64, + /// Channel count (derived from the layout; written into the slot meta). + pub channels: i32, + /// Sequence montage (ordered topmost-last; empty = silence). + pub montage: Vec, +} + +/// `render_audio_batch` (main->worker, M15 S3) — a batch of audio range +/// pulls rendered in order, sharing the claim/credit/frame_ready flow of +/// [`RenderBatchMsg`]. +#[derive(Serialize, Deserialize, Default, Debug, Clone, PartialEq)] +#[serde(default)] +pub struct RenderAudioBatchMsg { + /// Batch id (correlates with batch_accepted). + pub batch_id: i64, + /// The audio tickets, rendered in order. + pub tickets: Vec, +} + /// `frame_failed` (worker->main) — one ticket failed to render; the main /// process falls back (purple frame) and owns the slot again. #[derive(Serialize, Deserialize, Default, Debug, Clone, PartialEq)] @@ -1338,15 +1393,66 @@ mod tests { assert_eq!(tagged, failed); } + #[test] + fn render_audio_batch_wire_roundtrip() { + // M15 S3: the audio batch message reuses the claim/credit flow of + // render_batch with a dedicated ticket spec. + let batch = RenderAudioBatchMsg { + batch_id: 5, + tickets: vec![AudioTicketSpec { + ticket: 41, + slot: 0, + time_num: 0, + time_den: 48000, + duration_num: 1600, + duration_den: 48000, + sample_rate: 48000, + channel_layout: 0x3, + channels: 2, + montage: vec![WireMontageClip { + filename: "a.mp4".into(), + stream_index: 0, + in_num: 0, + in_den: 24, + out_num: 48, + out_den: 24, + media_in_num: 10, + media_in_den: 24, + gain: 0.5, + }], + }], + }; + let value = serde_json::to_value(&batch).unwrap(); + assert_eq!(value["batch_id"], 5); + assert_eq!(value["tickets"][0]["ticket"], 41); + assert_eq!(value["tickets"][0]["slot"], 0); + assert_eq!(value["tickets"][0]["sample_rate"], 48000); + assert_eq!(value["tickets"][0]["channel_layout"], 3); + assert_eq!(value["tickets"][0]["channels"], 2); + assert_eq!(value["tickets"][0]["duration_num"], 1600); + assert_eq!(value["tickets"][0]["montage"][0]["filename"], "a.mp4"); + let round: RenderAudioBatchMsg = serde_json::from_value(value).unwrap(); + assert_eq!(round, batch); + // Defaults: a bare audio ticket parses (missing fields default). + let bare: AudioTicketSpec = serde_json::from_str(r#"{"ticket":1,"slot":2}"#).unwrap(); + assert_eq!(bare.ticket, 1); + assert_eq!(bare.slot, 2); + assert_eq!(bare.sample_rate, 0); + assert!(bare.montage.is_empty()); + } + #[test] fn v2_type_constants_are_stable_wire_names() { assert_eq!(TYPE_HELLO_CAPS, "hello_caps"); assert_eq!(TYPE_RENDER_BATCH, "render_batch"); assert_eq!(TYPE_BATCH_ACCEPTED, "batch_accepted"); assert_eq!(TYPE_FRAME_FAILED, "frame_failed"); + assert_eq!(TYPE_RENDER_AUDIO_BATCH, "render_audio_batch"); assert_eq!(TYPE_SHUTDOWN, "shutdown"); - // BGRA8 slot format stays outside the PixelFormat enum range. + // BGRA8 slot format stays outside the PixelFormat enum range; the + // audio slot format follows it (M15 S3). assert_eq!(SLOT_FORMAT_BGRA8, 100); + assert_eq!(SLOT_FORMAT_AUDIO_F32, 101); } // ---- Shared-memory transport ----------------------------------------- diff --git a/crates/oakrender/src/manager.rs b/crates/oakrender/src/manager.rs index 79ef3828b..47a215722 100644 --- a/crates/oakrender/src/manager.rs +++ b/crates/oakrender/src/manager.rs @@ -31,7 +31,7 @@ use crate::autocacher::PreviewAutoCacher; use crate::backend::BackendKind; use crate::error::{Error, Result}; use crate::eval; -use crate::procpool::{DispatcherConfig, ProcessDispatcher, ShmFrameRef}; +use crate::procpool::{DispatcherConfig, ProcessDispatcher, ShmAudioRef, ShmFrameRef}; use crate::ticket::{TicketArena, TicketId}; use crate::worker::{InlineDispatcher, JobDispatch}; @@ -58,9 +58,9 @@ pub enum RenderBackendChoice { pub struct RenderManager { /// Video job dispatch (the process dispatcher, M15). pub dispatch: Arc, - /// 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). + /// Audio job dispatch (M15 S3: the process dispatcher, like video — + /// audio renders in oak-worker so plugin crashes are isolated; the + /// inline fallback lives in the arena, design §3.7). pub audio_dispatch: Arc, /// Ticket arena. pub tickets: Arc, @@ -96,27 +96,33 @@ impl RenderManager { eval::render_produced_frame(time, params) .map(crate::ticket::TicketPayload::Video) }); - let (dispatch, audio_dispatch): (Arc, Arc) = - match choice { - RenderBackendChoice::Threads => { - // 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 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( + let (dispatch, audio_dispatch, audio_fallback): ( + Arc, + Arc, + Option>, + ) = match choice { + RenderBackendChoice::Threads => { + // Test-only inline backend: synchronous execution on the + // calling thread, shared by video and audio. + let inline = InlineDispatcher::sync(); + (inline.clone(), inline, None) + } + RenderBackendChoice::Processes(config) => { + let dispatcher = ProcessDispatcher::new(config)?; + dispatcher.start()?; + // M15 S3: audio rendering runs in the worker pool too (a + // plugin crash during an audio render must not take down + // the main process — design §3.7). The inline dispatcher + // stays as the fallback when the process dispatcher is + // unavailable (teardown). + let fallback = InlineDispatcher::sync(); + (dispatcher.clone(), dispatcher, Some(fallback)) + } + }; + let tickets = Arc::new(TicketArena::new_with_audio_fallback( dispatch.clone(), audio_dispatch.clone(), + audio_fallback, producer, )); *guard = Some(Arc::new(RenderManager { @@ -173,6 +179,12 @@ impl RenderManager { self.dispatch.release_frame(frame); } + /// Release a consumed shm audio frame's slot (M15 S3; see + /// [`RenderManager::release_frame`]). + pub fn release_audio_frame(&self, frame: &ShmAudioRef) { + self.dispatch.release_audio_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. diff --git a/crates/oakrender/src/procpool.rs b/crates/oakrender/src/procpool.rs index a5345fbe1..91ce58736 100644 --- a/crates/oakrender/src/procpool.rs +++ b/crates/oakrender/src/procpool.rs @@ -69,13 +69,16 @@ use serde_json::{json, Value}; use crate::error::{Error, Result}; use crate::ipc::{ - write_message, BatchAcceptedMsg, FrameFailedMsg, FrameReadyMsg, FrameSlotMeta, FrameSlotPool, - HandshakeMsg, HelloCapsMsg, RenderBatchMsg, BatchTicketSpec, SharedMemoryRegion, ShmMode, - WireMontageClip, SLOT_FORMAT_BGRA8, TYPE_BATCH_ACCEPTED, TYPE_ERROR, TYPE_FRAME_FAILED, - TYPE_FRAME_READY, TYPE_HANDSHAKE, TYPE_HELLO_CAPS, + write_message, AudioTicketSpec, BatchAcceptedMsg, BatchTicketSpec, FrameFailedMsg, FrameReadyMsg, + FrameSlotMeta, FrameSlotPool, HandshakeMsg, HelloCapsMsg, RenderAudioBatchMsg, RenderBatchMsg, + SharedMemoryRegion, ShmMode, WireMontageClip, SLOT_FORMAT_BGRA8, TYPE_BATCH_ACCEPTED, + TYPE_ERROR, TYPE_FRAME_FAILED, TYPE_FRAME_READY, TYPE_HANDSHAKE, TYPE_HELLO_CAPS, + TYPE_RENDER_AUDIO_BATCH, }; use crate::scheduler::{FrameKey, FrameRequest, PreviewScheduler, SubmitOutcome}; -use crate::ticket::{Completion, TicketPayload, TicketResult, VideoTicketParams}; +use crate::ticket::{ + AudioSamples, AudioTicketParams, Completion, TicketPayload, TicketResult, VideoTicketParams, +}; use crate::worker::{Job, JobDispatch}; /// Protocol version spoken by the dispatcher (v1 base; v2 messages are @@ -85,7 +88,17 @@ pub const DISPATCH_PROTOCOL_VERSION: i32 = 1; /// Restart attempts per worker before its tickets fail permanently. const MAX_RESTARTS: u32 = 5; -/// Default slots per worker segment (design §3.1: 8 slots starting). +/// Maximum audio bytes a process-backend audio ticket may occupy in a shm +/// slot (M15 S3). Larger ranges (long exports) are refused by `post` so +/// the arena falls back to main-process inline rendering — a several- +/// minute export audio buffer does not need (and should not force) a +/// giant shared-memory segment. ~64 MB ≈ 2.9 min of 48 kHz stereo. +const MAX_AUDIO_SLOT_BYTES: usize = 64 * 1024 * 1024; + +/// Legacy fixed default slots per worker (design §3.1 "8 slots starting"). +/// The M15 S3 adaptive [`default_slots_per_worker`] policy supersedes it +/// for auto-configured dispatchers; kept as the documented starting point +/// and the cap for small frames. pub const DEFAULT_SLOTS_PER_WORKER: u32 = 8; /// Frame bytes copied into main-process heap buffers. The playback path @@ -264,6 +277,81 @@ impl std::fmt::Debug for ShmFrameRef { } } +/// Zero-copy handle to rendered audio in a worker segment (M15 S3): what +/// an audio ticket completion carries on the process backend. The samples +/// live in the shm slot as little-endian interleaved f32 (wire format +/// [`crate::ipc::SLOT_FORMAT_AUDIO_F32`]); the consumer reads them with +/// [`ShmAudioRef::samples`] and releases the slot through +/// [`ProcessDispatcher::release_audio_frame`] when done. `sample_rate` / +/// `channel_layout` are carried from the ticket params (they are not part +/// of the slot meta POD). +#[derive(Clone)] +pub struct ShmAudioRef { + /// Worker index owning the segment. + pub worker: u32, + /// Slot index in that segment. + pub slot: u32, + /// Slot metadata (format = `SLOT_FORMAT_AUDIO_F32`). + pub meta: ShmFrameMeta, + /// The segment view (keeps the mapping alive). + pub shm: Arc, + /// Output sample rate (Hz; from the ticket params). + pub sample_rate: i32, + /// Output channel layout mask (from the ticket params). + pub channel_layout: u64, + /// Channel count (also in the slot meta). + pub channel_count: i32, +} + +impl ShmAudioRef { + /// View the same slot as a generic [`ShmFrameRef`] (slot release paths + /// that are shared with video frames). + pub fn frame_ref(&self) -> ShmFrameRef { + ShmFrameRef { + worker: self.worker, + slot: self.slot, + meta: self.meta.clone(), + shm: self.shm.clone(), + } + } + + /// Copy the interleaved f32 samples out of the slot (the counted + /// copy path — audio bytes must outlive the slot to reach the output + /// device / encoder). + pub fn samples(&self) -> Vec { + let bytes = self.shm.slot_bytes(self.slot); + let valid = bytes + .get(..self.meta.data_size.max(0) as usize) + .unwrap_or(&[]); + valid + .chunks_exact(4) + .map(|c| f32::from_le_bytes([c[0], c[1], c[2], c[3]])) + .collect() + } + + /// The decoded [`AudioSamples`] (sample rate / layout from the ticket + /// params, not the slot). + pub fn to_audio_samples(&self) -> AudioSamples { + AudioSamples { + samples: self.samples(), + sample_rate: self.sample_rate, + channel_layout: self.channel_layout, + channel_count: self.channel_count, + } + } +} + +impl std::fmt::Debug for ShmAudioRef { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + f.debug_struct("ShmAudioRef") + .field("worker", &self.worker) + .field("slot", &self.slot) + .field("sample_rate", &self.sample_rate) + .field("channel_count", &self.channel_count) + .finish() + } +} + // --------------------------------------------------------------------------- // Slot pixel conversions (M15 S2) // --------------------------------------------------------------------------- @@ -342,6 +430,41 @@ pub fn default_worker_count(slots_per_worker: u32, slot_bytes: usize) -> usize { by_cores.min(by_mem).max(1) } +/// Slot-count policy when a segment grows (M15 S3 grow-on-demand): cap +/// the per-worker segment memory at `GROWN_SEGMENT_BUDGET`, never drop +/// below 2 slots (enough to keep a worker flowing), never exceed the +/// current count. +pub fn default_slots_for_bytes(slot_bytes: usize, current_slots: u32) -> u32 { + /// Per-worker segment budget for a grown segment (256 MiB). + const GROWN_SEGMENT_BUDGET: usize = 256 * 1024 * 1024; + let by_mem = (GROWN_SEGMENT_BUDGET / slot_bytes.max(1)).max(2) as u32; + by_mem.min(current_slots).max(2) +} + +/// Default slots per worker segment (M15 S3 adaptive policy): the +/// segment is sized so per-worker shared memory stays within +/// `DEFAULT_SEGMENT_BUDGET` (128 MiB), bounded to `[2, 8]`. Small preview +/// frames (BGRA8 1080p ≈ 8.3 MB) get the full 8 slots (~66 MB); F32 1080p +/// (≈ 33 MB) drops to 3; F32 4K (≈ 132 MB) to 2. The worker-count policy +/// then bounds the whole pool against RAM/4. +pub fn default_slots_per_worker(slot_bytes: usize) -> u32 { + const DEFAULT_SEGMENT_BUDGET: usize = 128 * 1024 * 1024; + const MIN_SLOTS: u32 = 2; + const MAX_SLOTS: u32 = 8; + ((DEFAULT_SEGMENT_BUDGET / slot_bytes.max(1)).max(MIN_SLOTS as usize) as u32) + .clamp(MIN_SLOTS, MAX_SLOTS) +} + +/// Default batch size B (M15 S3 adaptive policy): the design figure +/// `120 / workers` (a full playback pre-render window split across the +/// pool), capped at the per-worker slot count — credit caps a batch at +/// the free slots anyway, so a B larger than the slots just wastes a +/// claim round trip. +pub fn default_batch_size(workers: usize, slots: u32) -> usize { + let design = (120 / workers.max(1)).max(1); + design.min(slots.max(1) as usize) +} + /// Physical memory in bytes (macOS `hw.memsize`, Linux `sysconf`). fn physical_memory_bytes() -> Option { #[cfg(target_os = "macos")] @@ -390,7 +513,8 @@ pub struct DispatcherConfig { pub worker_bin: Option, /// Worker process count. `0` = the [`default_worker_count`] policy. pub workers: usize, - /// Output slots per worker segment. `0` = 8. + /// Output slots per worker segment. `0` = the + /// [`default_slots_per_worker`] policy (adaptive to the frame size). pub slots_per_worker: u32, /// Frame width of the segment geometry. `0` = 1920. pub width: i32, @@ -399,7 +523,8 @@ pub struct DispatcherConfig { /// Slot wire format: an `oakcore_rs::PixelFormat` int or /// [`SLOT_FORMAT_BGRA8`]. Default BGRA8 (the viewer preview path). pub slot_format: i32, - /// Batch size `B`. `0` = `max(1, 120 / workers)`. + /// Batch size `B`. `0` = the [`default_batch_size`] policy (adaptive + /// to workers and slots). pub batch_size: usize, /// Graph snapshot path sent to every worker via `load_graph` after /// the handshake (`None` = no graph). @@ -426,12 +551,10 @@ impl Default for DispatcherConfig { impl DispatcherConfig { fn normalize(&self) -> DispatcherConfig { + // Only the geometry defaults are resolved here; the adaptive + // policies (workers / slots / batch size) are resolved in + // `ProcessDispatcher::new` where slot_bytes is known. let mut c = self.clone(); - c.slots_per_worker = if c.slots_per_worker == 0 { - DEFAULT_SLOTS_PER_WORKER - } else { - c.slots_per_worker - }; c.width = if c.width == 0 { 1920 } else { c.width }; c.height = if c.height == 0 { 1080 } else { c.height }; c @@ -476,6 +599,9 @@ struct WorkerHandle { child: Option, stdin: Option, shm: Arc, + /// Current per-slot data capacity (grows on demand, M15 S3: a ticket + /// requesting F32 or a larger frame rebuilds the segment first). + slot_bytes: usize, /// FIFO mirror of the shm free ring's contents (the credit). free_slots: VecDeque, /// Dispatched ticket -> assigned slot (awaiting frame_ready). @@ -488,10 +614,20 @@ struct WorkerHandle { restarts: u32, spawned_at: Instant, accepted_batches: u64, + /// True between a segment grow (M15 S3) and the worker's hello_caps + /// re-attach: the dispatcher must not send new batches while the worker + /// is still attached to the old pool. + reconfiguring: bool, } impl WorkerHandle { - fn shell(index: usize, generation: u64, shm: Arc, slots: u32) -> WorkerHandle { + fn shell( + index: usize, + generation: u64, + shm: Arc, + slots: u32, + slot_bytes: usize, + ) -> WorkerHandle { WorkerHandle { index, generation, @@ -499,6 +635,7 @@ impl WorkerHandle { child: None, stdin: None, shm, + slot_bytes, free_slots: (0..slots).collect(), outstanding: HashMap::new(), held: HashSet::new(), @@ -508,6 +645,7 @@ impl WorkerHandle { restarts: 0, spawned_at: Instant::now(), accepted_batches: 0, + reconfiguring: false, } } } @@ -519,6 +657,9 @@ impl WorkerHandle { struct PendingTicket { key: FrameKey, params: Arc, + /// Audio ticket params when this ticket is an audio range pull (M15 + /// S3); `None` for video tickets. + audio: Option>, done: Option, } @@ -533,6 +674,10 @@ struct Inner { next_ticket: i64, events_rx: mpsc::Receiver, events_tx: mpsc::Sender, + /// Segment rebuild generation (M15 S3 grow-on-demand geometry): bumped + /// on every per-worker segment resize so re-created segments never + /// reuse the name of a live mapping. + seg_generation: u64, started: bool, shutting_down: bool, } @@ -550,18 +695,28 @@ pub struct ProcessDispatcher { impl ProcessDispatcher { /// Build a dispatcher from `config` (does not spawn; call - /// [`ProcessDispatcher::start`]). + /// [`ProcessDispatcher::start`]). The adaptive policies resolve here + /// (M15 S3): slots scale with the frame's slot bytes, workers with + /// cores/RAM, batch size with workers and slots. pub fn new(config: DispatcherConfig) -> Result> { let config = config.normalize(); - let slots = config.slots_per_worker; let slot_bytes = slot_bytes_for(config.width, config.height, config.slot_format); + let slots = if config.slots_per_worker == 0 { + default_slots_per_worker(slot_bytes) + } else { + config.slots_per_worker + }; let workers = if config.workers == 0 { default_worker_count(slots, slot_bytes) } else { config.workers }; + let batch_size = if config.batch_size == 0 { + default_batch_size(workers, slots) + } else { + config.batch_size + }; let bin = resolve_worker_bin(&config)?; - let batch_size = config.batch_size; let (events_tx, events_rx) = mpsc::channel(); Ok(Arc::new(ProcessDispatcher { inner: Mutex::new(Inner { @@ -575,6 +730,7 @@ impl ProcessDispatcher { next_ticket: 1, events_rx, events_tx, + seg_generation: 0, started: false, shutting_down: false, }), @@ -697,6 +853,12 @@ impl ProcessDispatcher { unsafe { handle.shm.pool().release(frame.slot) }; } + /// Release a consumed audio frame's slot (M15 S3) — the audio + /// counterpart of [`ProcessDispatcher::release_frame`]. + pub fn release_audio_frame(&self, frame: &ShmAudioRef) { + self.release_frame(&frame.frame_ref()); + } + /// Cancel one frame request (pending or in flight). The completion /// fires with `Error::State` exactly once; a late frame_ready for an /// in-flight cancel recycles the slot silently. @@ -780,7 +942,9 @@ impl ProcessDispatcher { // 3. Interleaved batch claims + dispatch (free slots = credit). for i in 0..inner.workers.len() { - if matches!(inner.workers[i].state, WorkerState::Alive) { + if matches!(inner.workers[i].state, WorkerState::Alive) + && !inner.workers[i].reconfiguring + { self.dispatch_to(inner, i); } } @@ -805,18 +969,10 @@ impl ProcessDispatcher { match typ { TYPE_HANDSHAKE => { // The worker's startup handshake: answer with the shm - // geometry (protocol v1 flow). + // geometry (protocol v1 flow). A mid-session handshake is a + // segment grow (M15 S3): the worker re-attaches the new pool. handle.startup_seen = true; - let hs = HandshakeMsg { - protocol_version: DISPATCH_PROTOCOL_VERSION, - shm_key: handle.shm.key().to_string(), - input_shm_key: String::new(), - input_slots: 0, - output_slots: handle.shm.slot_count() as i32, - slot_data_bytes: handle.shm.slot_data_bytes() as i64, - input_slot_data_bytes: 0, - }; - if self.send_json(handle, &hs.to_json()).is_err() { + if self.send_json(handle, &handshake_for(handle)).is_err() { handle.state = WorkerState::Dead; } } @@ -824,6 +980,9 @@ impl ProcessDispatcher { if let Ok(caps) = serde_json::from_value::(msg) { handle.caps = Some(caps); handle.state = WorkerState::Alive; + // A re-attach after a segment grow is complete: the + // dispatcher may send batches again. + handle.reconfiguring = false; // One load_graph right after the first handshake. if !handle.graph_sent { if let Some(path) = inner.config.graph_snapshot.clone() { @@ -913,15 +1072,34 @@ impl ProcessDispatcher { let key = pt.key; inner.scheduler.frame_done(&key); if let Some(done) = pt.done.take() { - fired.push(( - done, - Ok(TicketPayload::ShmFrame(ShmFrameRef { - worker: worker as u32, - slot: slot as u32, - meta, - shm, - })), - )); + // M15 S3: audio tickets complete with the shm audio + // payload (the consumer reads the slot and releases it); + // video tickets keep the ShmFrame payload. + if let Some(audio) = &pt.audio { + let params = audio.clone(); + fired.push(( + done, + Ok(TicketPayload::ShmAudio(ShmAudioRef { + worker: worker as u32, + slot: slot as u32, + meta, + shm, + sample_rate: params.sample_rate, + channel_layout: params.channel_layout, + channel_count: params.channel_layout.count_ones().max(1) as i32, + })), + )); + } else { + fired.push(( + done, + Ok(TicketPayload::ShmFrame(ShmFrameRef { + worker: worker as u32, + slot: slot as u32, + meta, + shm, + })), + )); + } } else { // Cancelled while in flight: recycle the slot now. self.recycle_slot(inner, worker, slot as u32); @@ -977,10 +1155,38 @@ impl ProcessDispatcher { if credit == 0 { return; } - let Some(batch) = inner.scheduler.claim_batch(worker, credit) else { + // Grow-on-demand (M15 S3): if a pending request for this worker + // needs a bigger slot than the segment provides, and the worker + // has no in-flight frames, rebuild its segment first (the worker + // re-attaches on a fresh handshake). While the worker is busy the + // oversized request simply stays pending — claim_batch filters it + // by max_bytes, so it is served after the drain. + let grow = { + let handle = &inner.workers[worker]; + if handle.outstanding.is_empty() { + inner + .scheduler + .max_pending_bytes_for_worker(worker, handle.slot_bytes) + } else { + None + } + }; + if let Some(need) = grow { + if let Err(e) = self.rebuild_segment(inner, worker, need) { + eprintln!("procpool: worker {worker} segment grow to {need} B failed: {e}"); + } + // Stop here: the worker is re-attaching to the new segment + // (hello_caps pending). Dispatch resumes on the next pump + // once `reconfiguring` clears — sending a batch now would + // race the pool swap. + return; + } + let max_bytes = inner.workers[worker].slot_bytes; + let Some(batch) = inner.scheduler.claim_batch(worker, credit, max_bytes) else { return; }; - let mut wire_tickets = Vec::with_capacity(batch.frames.len()); + let mut video_tickets = Vec::with_capacity(batch.frames.len()); + let mut audio_tickets: Vec = Vec::new(); for req in &batch.frames { let ticket = req.payload; let slot = match inner.workers[worker].free_slots.pop_front() { @@ -991,32 +1197,61 @@ impl ProcessDispatcher { let Some(pt) = inner.tickets.get(&ticket) else { continue; }; - wire_tickets.push(build_ticket_spec( - ticket, - slot, - &pt.params, - inner.config.slot_format, - )); + if let Some(audio) = &pt.audio { + audio_tickets.push(build_audio_ticket_spec(ticket, slot, audio)); + } else { + video_tickets.push(build_ticket_spec( + ticket, + slot, + &pt.params, + inner.config.slot_format, + )); + } } - let msg = RenderBatchMsg { - batch_id: batch.batch_id as i64, - tickets: wire_tickets, - }; - // The `type` tag is added by hand: [`RenderBatchMsg`] only - // carries the payload fields (it is the parse-side struct). - let mut value = match serde_json::to_value(&msg) { - Ok(v) => v, - Err(_) => return, - }; - if let Some(obj) = value.as_object_mut() { - obj.insert( - "type".to_string(), - Value::String(crate::ipc::TYPE_RENDER_BATCH.to_string()), - ); + // A single claim may mix audio and video (different scheduler + // keys in one batch); they are delivered as two messages under + // the same batch id, claimed by the worker in order. + if !video_tickets.is_empty() { + let msg = RenderBatchMsg { + batch_id: batch.batch_id as i64, + tickets: video_tickets, + }; + // The `type` tag is added by hand: the parse-side structs only + // carry the payload fields. + let mut value = match serde_json::to_value(&msg) { + Ok(v) => v, + Err(_) => return, + }; + if let Some(obj) = value.as_object_mut() { + obj.insert( + "type".to_string(), + Value::String(crate::ipc::TYPE_RENDER_BATCH.to_string()), + ); + } + if self.send_json(&mut inner.workers[worker], &value).is_err() { + inner.workers[worker].state = WorkerState::Dead; + return; + } } - if self.send_json(&mut inner.workers[worker], &value).is_err() { - inner.workers[worker].state = WorkerState::Dead; - return; + if !audio_tickets.is_empty() { + let msg = RenderAudioBatchMsg { + batch_id: batch.batch_id as i64, + tickets: audio_tickets, + }; + let mut value = match serde_json::to_value(&msg) { + Ok(v) => v, + Err(_) => return, + }; + if let Some(obj) = value.as_object_mut() { + obj.insert( + "type".to_string(), + Value::String(TYPE_RENDER_AUDIO_BATCH.to_string()), + ); + } + if self.send_json(&mut inner.workers[worker], &value).is_err() { + inner.workers[worker].state = WorkerState::Dead; + return; + } } } } @@ -1098,7 +1333,7 @@ impl ProcessDispatcher { }) .map_err(|e| Error::Failed(format!("spawn reader thread: {e}")))?; - let mut handle = WorkerHandle::shell(index, generation, shm, inner.slots); + let mut handle = WorkerHandle::shell(index, generation, shm, inner.slots, inner.slot_bytes); handle.child = Some(child); handle.stdin = stdin; handle.spawned_at = Instant::now(); @@ -1166,6 +1401,59 @@ impl ProcessDispatcher { inner.workers[worker].state = WorkerState::Dead; } } + + /// Grow a worker's segment to `need_bytes` per slot (M15 S3 grow-on- + /// demand geometry, design §3.1 "段按需扩容或重建"): create a fresh + /// segment under a new key (the old mapping stays alive for consumers + /// still holding [`ShmFrameRef`]s into it — they release as stale + /// refs), re-point the handle, reseed the free slots and have the + /// worker re-attach through a fresh handshake. The caller guarantees + /// `outstanding` is empty (no frame is mid-render in the old pool). + /// The worker's hello_caps clears `reconfiguring`, unblocking dispatch. + fn rebuild_segment(&self, inner: &mut Inner, worker: usize, need_bytes: usize) -> Result<()> { + let current_slots = inner.workers[worker].shm.slot_count(); + let slots = default_slots_for_bytes(need_bytes, current_slots); + let generation = inner.seg_generation; + inner.seg_generation += 1; + let base = SharedMemoryRegion::make_key(std::process::id() as i64, worker as i32); + let key = format!( + "{base}-g{}-s{generation}", + inner.workers[worker].generation + ); + let shm = ShmRegionView::create(&key, slots, need_bytes)?; + { + let handle = &mut inner.workers[worker]; + handle.shm = shm; + handle.slot_bytes = need_bytes; + handle.free_slots = (0..slots).collect(); + handle.held.clear(); + // No new batches until the worker re-attaches the new pool. + handle.reconfiguring = true; + } + let hs = { + let handle = &inner.workers[worker]; + handshake_for(handle) + }; + if self.send_json(&mut inner.workers[worker], &hs).is_err() { + inner.workers[worker].state = WorkerState::Dead; + } + Ok(()) + } +} + +/// The handshake reply the dispatcher sends a worker (startup and M15 S3 +/// segment-grow re-attach): the worker's current shm geometry. +fn handshake_for(handle: &WorkerHandle) -> Value { + HandshakeMsg { + protocol_version: DISPATCH_PROTOCOL_VERSION, + shm_key: handle.shm.key().to_string(), + input_shm_key: String::new(), + input_slots: 0, + output_slots: handle.shm.slot_count() as i32, + slot_data_bytes: handle.shm.slot_data_bytes() as i64, + input_slot_data_bytes: 0, + } + .to_json() } impl JobDispatch for ProcessDispatcher { @@ -1184,6 +1472,20 @@ impl JobDispatch for ProcessDispatcher { if inner.shutting_down { return false; } + // M15 S3: audio ranges larger than a practical shm slot (export + // of many minutes of audio) are refused here so the arena falls + // back to main-process inline rendering (design §3.7) — the + // process backend stays for the real-time chunks and short + // ranges that fit a segment. + if let Some(audio) = &job.audio { + let too_large = match crate::eval::audio_samples_byte_len(audio) { + Ok(bytes) => bytes > MAX_AUDIO_SLOT_BYTES, + Err(_) => true, // invalid range: let the inline path report it + }; + if too_large { + return false; + } + } let id = inner.next_ticket; inner.next_ticket += 1; let frame = job.schedule.frame.unwrap_or(id); @@ -1192,11 +1494,22 @@ impl JobDispatch for ProcessDispatcher { frame, version: job.schedule.version, }; + // M15 S3: per-request slot geometry. Audio tickets need the + // sample bytes of their range; video tickets need the frame + // size x the ticket's wire format (force_format honored). + let slot_bytes = match &job.audio { + Some(audio) => crate::eval::audio_samples_byte_len(audio).unwrap_or(0), + None => { + let (w, h) = job.params.render_size(); + slot_bytes_for(w, h, ticket_wire_format(&job.params, inner.config.slot_format)) + } + }; inner.tickets.insert( id, PendingTicket { key, params: job.params, + audio: job.audio, done: Some(job.done), }, ); @@ -1205,6 +1518,7 @@ impl JobDispatch for ProcessDispatcher { priority: job.schedule.priority, distance: job.schedule.distance, payload: id, + slot_bytes, }; match inner.scheduler.submit(request) { SubmitOutcome::Accepted => {} @@ -1267,6 +1581,12 @@ impl JobDispatch for ProcessDispatcher { self.release_frame(frame); } + /// Release a consumed audio frame's slot (M15 S3; delegates to the + /// inherent release). + fn release_audio_frame(&self, frame: &ShmAudioRef) { + self.release_audio_frame(frame); + } + /// Graceful shutdown: `shutdown` messages, a short drain pumping /// completions, then kill stragglers; every ticket still open /// completes with `Error::State`. @@ -1357,6 +1677,15 @@ fn resolve_worker_bin(config: &DispatcherConfig) -> Result { ))) } +/// The wire slot format a video ticket requests: the ticket's forced +/// PixelFormat when set (F32 for exports / full-resolution / scopes, M15 +/// S3 — the worker then writes F32 straight into the slot and the export +/// reads it back with no BGRA8 round trip), else the dispatcher's default +/// slot format (BGRA8 for the viewer preview path). +fn ticket_wire_format(params: &VideoTicketParams, config_format: i32) -> i32 { + params.force_format.map(|f| f as i32).unwrap_or(config_format) +} + /// Map ticket params to the wire ticket spec (main assigns `slot`). fn build_ticket_spec( ticket: i64, @@ -1391,7 +1720,7 @@ fn build_ticket_spec( time_den: params.time.denominator(), width, height, - format: slot_format, + format: ticket_wire_format(params, slot_format), channels: 4, footage_file, footage_stream, @@ -1399,6 +1728,39 @@ fn build_ticket_spec( } } +/// Map audio ticket params to the wire audio ticket spec (M15 S3; main +/// assigns `slot`). +fn build_audio_ticket_spec(ticket: i64, slot: u32, params: &AudioTicketParams) -> AudioTicketSpec { + let duration = params.range.out() - params.range.in_(); + let montage = params + .montage + .iter() + .map(|c| WireMontageClip { + filename: c.filename.clone(), + stream_index: c.stream_index, + in_num: c.in_time.numerator(), + in_den: c.in_time.denominator(), + out_num: c.out_time.numerator(), + out_den: c.out_time.denominator(), + media_in_num: c.media_in.numerator(), + media_in_den: c.media_in.denominator(), + gain: c.gain, + }) + .collect(); + AudioTicketSpec { + ticket, + slot: slot as i32, + time_num: params.range.in_().numerator(), + time_den: params.range.in_().denominator(), + duration_num: duration.numerator(), + duration_den: duration.denominator(), + sample_rate: params.sample_rate, + channel_layout: params.channel_layout, + channels: params.channel_layout.count_ones().max(1) as i32, + montage, + } +} + #[cfg(test)] mod tests { use super::*; @@ -1426,10 +1788,51 @@ mod tests { #[test] fn config_normalization_defaults() { let c = DispatcherConfig::default().normalize(); - assert_eq!(c.slots_per_worker, DEFAULT_SLOTS_PER_WORKER); + // Geometry defaults resolve here; the adaptive counts stay 0 (auto) + // and resolve in `ProcessDispatcher::new` where slot_bytes is known. assert_eq!(c.width, 1920); assert_eq!(c.height, 1080); assert_eq!(c.slot_format, SLOT_FORMAT_BGRA8); + assert_eq!(c.slots_per_worker, 0); + assert_eq!(c.workers, 0); + assert_eq!(c.batch_size, 0); + } + + #[test] + fn slots_policy_adapts_to_slot_size() { + // BGRA8 1080p: the full 8 slots (~66 MB per worker segment). + let bgra8_1080p = slot_bytes_for(1920, 1080, SLOT_FORMAT_BGRA8); + assert_eq!(default_slots_per_worker(bgra8_1080p), 8); + // F32 1080p: drops to 4 (~133 MB per worker segment). + let f32_1080p = slot_bytes_for(1920, 1080, 4); + assert_eq!(default_slots_per_worker(f32_1080p), 4); + // F32 4K: 2 slots (the floor). + let f32_4k = slot_bytes_for(3840, 2160, 4); + assert_eq!(default_slots_per_worker(f32_4k), 2); + // Tiny slots: the cap at 8. + assert_eq!(default_slots_per_worker(16), 8); + } + + #[test] + fn batch_size_policy_scales_with_workers_and_slots() { + // 4 workers x 8 slots: the design 120/4 = 30 caps at the 8 slots. + assert_eq!(default_batch_size(4, 8), 8); + // 1 worker x 8 slots: 120/1 = 120 caps at 8. + assert_eq!(default_batch_size(1, 8), 8); + // 2 workers x 4 slots: 120/2 = 60 caps at 4. + assert_eq!(default_batch_size(2, 4), 4); + // 8 workers x 8 slots: 120/8 = 15 caps at 8. + assert_eq!(default_batch_size(8, 8), 8); + } + + #[test] + fn grown_segment_slots_stay_bounded() { + // Growing a segment keeps a sane slot count: 8.3 MB slots keep 8; + // 33 MB slots keep 8 (still within the 256 MiB grown budget); + // absurd sizes clamp at 2. + assert_eq!(default_slots_for_bytes(8_300_000, 8), 8); + assert_eq!(default_slots_for_bytes(33_000_000, 8), 8); + assert_eq!(default_slots_for_bytes(1 << 30, 8), 2); } #[test] diff --git a/crates/oakrender/src/scheduler.rs b/crates/oakrender/src/scheduler.rs index 5d4423157..a7f87d820 100644 --- a/crates/oakrender/src/scheduler.rs +++ b/crates/oakrender/src/scheduler.rs @@ -86,6 +86,11 @@ pub struct FrameRequest

{ pub distance: i64, /// Caller payload. pub payload: P, + /// Slot bytes this request needs (frame size x wire format; M15 S3). + /// `claim_batch` skips requests whose bytes exceed the worker's current + /// slot capacity, so the dispatcher can grow the segment first (grow- + /// on-demand geometry, design §3.1). + pub slot_bytes: usize, } /// A batch of frames one worker claimed. @@ -193,12 +198,20 @@ impl PreviewScheduler

{ /// Claim the next batch for `worker`: the worker's interleaved shard /// (frame number `≡ worker (mod W)`, plus any crash-requeued frames), /// ordered by priority class / playhead distance / ascending frame, - /// capped at `min(batch_size, credit)`. Returns `None` when nothing - /// is claimable (`credit == 0`, unknown worker, empty shard). + /// capped at `min(batch_size, credit)`. Requests needing more than + /// `max_bytes` of slot space are skipped (they stay pending until the + /// dispatcher grows the segment). Returns `None` when nothing + /// claimable (`credit == 0`, unknown worker, empty shard, all + /// oversized). /// /// Claimed frames never go to another worker while in flight (no /// stealing). - pub fn claim_batch(&mut self, worker: usize, credit: usize) -> Option> { + pub fn claim_batch( + &mut self, + worker: usize, + credit: usize, + max_bytes: usize, + ) -> Option> { if worker >= self.workers || credit == 0 { return None; } @@ -208,7 +221,8 @@ impl PreviewScheduler

{ .iter() .enumerate() .filter(|(_, e)| { - e.any_worker || e.request.key.frame.rem_euclid(workers as i64) == worker as i64 + e.request.slot_bytes <= max_bytes + && (e.any_worker || e.request.key.frame.rem_euclid(workers as i64) == worker as i64) }) .map(|(i, _)| i) .collect(); @@ -346,6 +360,25 @@ impl PreviewScheduler

{ self.pending.len() } + /// The largest `slot_bytes` among pending requests claimable by + /// `worker` (its shard plus crash-requeued frames) that exceeds + /// `current`, if any. The dispatcher uses this to grow a worker's + /// segment before the next claim (M15 S3 grow-on-demand geometry). + pub fn max_pending_bytes_for_worker(&self, worker: usize, current: usize) -> Option { + if worker >= self.workers { + return None; + } + let workers = self.workers; + self.pending + .iter() + .filter(|e| { + e.any_worker || e.request.key.frame.rem_euclid(workers as i64) == worker as i64 + }) + .map(|e| e.request.slot_bytes) + .max() + .filter(|&m| m > current) + } + /// Claimed (in-flight) request count. pub fn claimed_count(&self) -> usize { self.claimed.len() @@ -376,6 +409,7 @@ mod tests { priority: prio, distance: frame, payload: frame as u64, + slot_bytes: 0, } } @@ -386,7 +420,7 @@ mod tests { loop { let mut progress = false; for w in 0..s.workers() { - while let Some(batch) = s.claim_batch(w, 1024) { + while let Some(batch) = s.claim_batch(w, 1024, usize::MAX) { for f in &batch.frames { out.push((w, f.key.frame)); } @@ -448,9 +482,9 @@ mod tests { s.submit(req(1, f, FramePriority::Playback)); } // Both workers claim their shards first. - let batch0 = s.claim_batch(0, 8).unwrap(); + let batch0 = s.claim_batch(0, 8, 1024).unwrap(); assert_eq!(batch0.frames.len(), 4); // frames 0,2,4,6 - let batch1 = s.claim_batch(1, 8).unwrap(); + let batch1 = s.claim_batch(1, 8, 1024).unwrap(); assert_eq!(batch1.frames.len(), 4); // frames 1,3,5,7 // Worker 0 crashes: its frames come back... @@ -461,7 +495,7 @@ mod tests { assert_eq!(s.crash_requeued(), 4); // ...and worker 1 (NOT their shard) can claim them all. - let batch = s.claim_batch(1, 8).unwrap(); + let batch = s.claim_batch(1, 8, 1024).unwrap(); assert_eq!(batch.frames.len(), 4); let mut frames: Vec = batch.frames.iter().map(|f| f.key.frame).collect(); frames.sort_unstable(); @@ -483,7 +517,7 @@ mod tests { } s.submit(req(1, 100, FramePriority::Seek)); - let batch = s.claim_batch(0, 100).unwrap(); + let batch = s.claim_batch(0, 100, 1024).unwrap(); let order: Vec<(FramePriority, i64)> = batch .frames .iter() @@ -516,9 +550,9 @@ mod tests { s.submit(req(1, f, FramePriority::Playback)); } // Zero credit claims nothing. - assert!(s.claim_batch(0, 0).is_none()); + assert!(s.claim_batch(0, 0, 1024).is_none()); // Credit 3 claims exactly 3 (free slots are the credit). - let batch = s.claim_batch(0, 3).unwrap(); + let batch = s.claim_batch(0, 3, 1024).unwrap(); assert_eq!(batch.frames.len(), 3); assert_eq!(s.pending_count(), 7); } @@ -529,19 +563,47 @@ mod tests { for f in 0..10 { s.submit(req(1, f, FramePriority::Playback)); } - let batch = s.claim_batch(0, 100).unwrap(); + let batch = s.claim_batch(0, 100, 1024).unwrap(); assert_eq!(batch.frames.len(), 4, "batch size B caps the claim"); // Ascending frame order inside the batch. let frames: Vec = batch.frames.iter().map(|f| f.key.frame).collect(); assert_eq!(frames, vec![0, 1, 2, 3]); } + #[test] + fn claim_batch_skips_requests_exceeding_slot_capacity() { + // M15 S3 grow-on-demand: a request needing more bytes than the + // worker's current slot capacity stays pending (the dispatcher can + // grow the segment and claim it later); smaller requests still flow. + let mut s: PreviewScheduler = PreviewScheduler::new(1, 100); + for f in 0..4 { + let mut r = req(1, f, FramePriority::Playback); + r.slot_bytes = if f == 1 { 33_000_000 } else { 8_300_000 }; + s.submit(r); + } + // max_bytes 8_300_000: frame 1 (33 MB) is skipped. + let batch = s.claim_batch(0, 100, 8_300_000).unwrap(); + let frames: Vec = batch.frames.iter().map(|f| f.key.frame).collect(); + assert_eq!(frames, vec![0, 2, 3]); + assert_eq!(s.pending_count(), 1, "the oversized frame stays pending"); + assert_eq!( + s.max_pending_bytes_for_worker(0, 8_300_000), + Some(33_000_000), + "the dispatcher sees the growth need" + ); + // After the segment grows, the oversized frame is claimable. + let batch = s.claim_batch(0, 100, 33_000_000).unwrap(); + assert_eq!(batch.frames.len(), 1); + assert_eq!(batch.frames[0].key.frame, 1); + assert_eq!(s.pending_count(), 0); + } + #[test] fn done_and_failed_drop_the_claim() { let mut s: PreviewScheduler = PreviewScheduler::new(1, 4); s.submit(req(1, 0, FramePriority::Playback)); s.submit(req(1, 1, FramePriority::Playback)); - let batch = s.claim_batch(0, 4).unwrap(); + let batch = s.claim_batch(0, 4, 1024).unwrap(); assert_eq!(batch.frames.len(), 2); let k0 = batch.frames[0].key; let k1 = batch.frames[1].key; @@ -559,7 +621,7 @@ mod tests { let r = req(1, 5, FramePriority::Playback); let key = r.key; assert!(matches!(s.submit(r), SubmitOutcome::Accepted)); - let _ = s.claim_batch(0, 4).unwrap(); + let _ = s.claim_batch(0, 4, 1024).unwrap(); // In flight: rejected. assert!(matches!( s.submit(req(1, 5, FramePriority::Seek)), @@ -592,7 +654,7 @@ mod tests { } other => panic!("expected Replaced, got {other:?}"), } - let _ = s.claim_batch(0, 4).unwrap(); + let _ = s.claim_batch(0, 4, 1024).unwrap(); assert_eq!(s.claimed_worker(&second_key), Some(0)); } @@ -602,7 +664,7 @@ mod tests { for f in 0..6 { s.submit(req(7, f, FramePriority::Playback)); } - let _ = s.claim_batch(0, 4).unwrap(); // claims 4 of sequence 7 + let _ = s.claim_batch(0, 4, 1024).unwrap(); // claims 4 of sequence 7 s.submit(req(8, 0, FramePriority::Playback)); // other sequence let dropped = s.cancel_sequence(7); assert_eq!(dropped.len(), 6, "all 6 sequence-7 requests dropped"); @@ -623,6 +685,6 @@ mod tests { fn unknown_worker_claims_nothing() { let mut s: PreviewScheduler = PreviewScheduler::new(2, 4); s.submit(req(1, 0, FramePriority::Playback)); - assert!(s.claim_batch(2, 4).is_none()); + assert!(s.claim_batch(2, 4, 1024).is_none()); } } diff --git a/crates/oakrender/src/ticket.rs b/crates/oakrender/src/ticket.rs index 2af3b39ac..902cc5f8a 100644 --- a/crates/oakrender/src/ticket.rs +++ b/crates/oakrender/src/ticket.rs @@ -132,6 +132,11 @@ pub enum TicketPayload { /// process backend): zero copy — the consumer reads the pixels from /// the mapping and releases the slot through the dispatcher. ShmFrame(crate::procpool::ShmFrameRef), + /// Rendered audio living in a worker's shared-memory slot (M15 S3): + /// interleaved f32 in `SLOT_FORMAT_AUDIO_F32` slots, consumed with + /// [`crate::procpool::ShmAudioRef::samples`] and released through the + /// dispatcher — the audio counterpart of `ShmFrame`. + ShmAudio(crate::procpool::ShmAudioRef), } /// Completion payload: the rendered texture/samples or the failure @@ -209,6 +214,9 @@ impl TicketSlot { if let Ok(TicketPayload::ShmFrame(frame)) = &result { self.dispatch.release_frame(frame); } + if let Ok(TicketPayload::ShmAudio(audio)) = &result { + self.dispatch.release_audio_frame(audio); + } result = Err(Error::State); } // Publish the result before flipping the state flag: `wait()` only @@ -255,6 +263,11 @@ pub struct TicketArena { next: AtomicU64, dispatch: Arc, audio_dispatch: Arc, + /// M15 S3: main-process inline audio fallback, used when the process + /// dispatcher is unavailable (shutting down) — design §3.7. Audio + /// rendering normally runs in oak-worker; the inline fallback keeps + /// playback alive during teardown. + audio_fallback: Option>, slots: Mutex>>, shutting_down: AtomicBool, producer: Producer, @@ -273,11 +286,26 @@ impl TicketArena { video: Arc, audio: Arc, producer: Producer, + ) -> Self { + Self::new_with_audio_fallback(video, audio, None, producer) + } + + /// Arena with separate video/audio backends plus a main-process inline + /// audio fallback (M15 S3): when `audio` (the process dispatcher) + /// refuses a job, it is re-posted to `audio_fallback` (an + /// [`crate::worker::InlineDispatcher::sync`]); both gone, the ticket + /// cancels. + pub fn new_with_audio_fallback( + video: Arc, + audio: Arc, + audio_fallback: Option>, + producer: Producer, ) -> Self { Self { next: AtomicU64::new(1), dispatch: video, audio_dispatch: audio, + audio_fallback, slots: Mutex::new(HashMap::new()), shutting_down: AtomicBool::new(false), producer, @@ -358,6 +386,7 @@ impl TicketArena { node_identity: params.viewer, time: params.time, params, + audio: None, produce: producer, done: Box::new(move |result| slot_done.finish(result)), schedule, @@ -454,32 +483,49 @@ impl TicketArena { // The audio producer captures the montage + output params; the // video-params field of the job is a dummy (unused by the audio - // path). + // path). M15 S3: `audio` carries the params to the process + // dispatcher, which renders the range in oak-worker via + // render_audio_batch; the producer is the inline-fallback path. let ap = Arc::new(params); let viewer = ap.viewer; - let producer: Producer = Arc::new(move |_, _| eval::render_audio_samples(&ap)); - let slot_done = slot.clone(); - let job = crate::worker::Job { - node_identity: viewer, - time: range.in_(), - params: Arc::new(VideoTicketParams { - viewer, + let make_job = |slot_done: Arc| { + let ap_job = ap.clone(); + let ap_prod = ap.clone(); + let producer: Producer = Arc::new(move |_, _| eval::render_audio_samples(&ap_prod)); + crate::worker::Job { + node_identity: viewer, time: range.in_(), - force_size: None, - force_format: None, - cache: None, - cache_dir: None, - cache_id: None, - cache_timebase: None, - footage: None, - montage: Vec::new(), - }), - produce: producer, - done: Box::new(move |result| slot_done.finish(result)), - schedule: JobSchedule::seek(), + params: Arc::new(VideoTicketParams { + viewer, + time: range.in_(), + force_size: None, + force_format: None, + cache: None, + cache_dir: None, + cache_id: None, + cache_timebase: None, + footage: None, + montage: Vec::new(), + }), + audio: Some(ap_job), + produce: producer, + done: Box::new(move |result| slot_done.finish(result)), + schedule: JobSchedule::seek(), + } }; + let job = make_job(slot.clone()); if !self.audio_dispatch.post(job) { - slot.finish(Err(Error::State)); + // The process dispatcher is unavailable (shutting down): fall + // back to main-process inline audio rendering (design §3.7) so + // playback/export audio keeps flowing during teardown. + if let Some(fallback) = &self.audio_fallback { + let job = make_job(slot.clone()); + if !fallback.post(job) { + slot.finish(Err(Error::State)); + } + } else { + slot.finish(Err(Error::State)); + } } id } diff --git a/crates/oakrender/src/worker.rs b/crates/oakrender/src/worker.rs index 992560c71..5124d6d8e 100644 --- a/crates/oakrender/src/worker.rs +++ b/crates/oakrender/src/worker.rs @@ -43,7 +43,7 @@ use oakcore_rs::Rational; use crate::error::{Error, Result}; use crate::procpool::ShmFrameRef; use crate::scheduler::FramePriority; -use crate::ticket::{Completion, Producer, VideoTicketParams}; +use crate::ticket::{AudioTicketParams, Completion, Producer, VideoTicketParams}; fn lock(m: &Mutex) -> MutexGuard<'_, T> { m.lock().unwrap_or_else(|e| e.into_inner()) @@ -57,6 +57,11 @@ pub struct Job { pub time: Rational, /// Ticket parameters (size/format overrides). pub params: Arc, + /// Audio ticket parameters when this is an audio range pull (M15 S3): + /// the process dispatcher renders it through the worker's + /// `render_audio_batch` path into a shm slot; the in-process inline + /// fallback executes the producer instead. `None` for video jobs. + pub audio: Option>, /// Frame producer (arena-installed; the process backend never invokes /// it — workers render from the wire spec). pub produce: Producer, @@ -134,6 +139,10 @@ pub trait JobDispatch: Send + Sync { /// process backend holds slots. fn release_frame(&self, _frame: &ShmFrameRef) {} + /// Release a consumed shm audio frame's slot (M15 S3). Default no-op: + /// only the process backend holds slots. + fn release_audio_frame(&self, _frame: &crate::procpool::ShmAudioRef) {} + /// 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. @@ -372,6 +381,7 @@ mod tests { footage: None, montage: Vec::new(), }), + audio: None, produce, done: Box::new(move |r| { assert!(r.is_ok(), "producer must succeed here"); @@ -441,6 +451,7 @@ mod tests { footage: None, montage: Vec::new(), }), + audio: None, produce: p, done: Box::new(move |r| { let _ = tx.send(r.is_err()); @@ -481,6 +492,7 @@ mod tests { node_identity: 0, time: Rational::new(0, 1), params: params.clone(), + audio: None, produce: boom, done: Box::new(move |r| { assert!(r.is_err()); @@ -492,6 +504,7 @@ mod tests { node_identity: 1, time: Rational::new(1, 1), params, + audio: None, produce: ok, done: Box::new(move |r| { assert!(r.is_ok()); diff --git a/crates/oaktask/src/render.rs b/crates/oaktask/src/render.rs index 64b27b1f8..275fbf27b 100644 --- a/crates/oaktask/src/render.rs +++ b/crates/oaktask/src/render.rs @@ -57,7 +57,9 @@ 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::procpool::{ + bgra8_to_f32_rgba, DispatcherConfig, ProcessDispatcher, ShmAudioRef, ShmFrameRef, +}; use oakrender::ticket::{ ticket_kind, AudioTicketParams, MontageClip, TicketArena, TicketId, TicketPayload, TicketResult, VideoTicketParams, @@ -668,7 +670,17 @@ impl RenderTask { oakrender::eval::render_produced_frame(time, params) .map(TicketPayload::Video) }); - let arena = Arc::new(TicketArena::new(dispatcher.clone(), producer)); + // M15 S3: the private dispatcher routes audio through the + // worker pool too; the inline dispatcher is the fallback + // when the dispatcher refuses a job (oversized audio ranges + // / teardown). + let inline = oakrender::worker::InlineDispatcher::sync(); + let arena = Arc::new(TicketArena::new_with_audio_fallback( + dispatcher.clone(), + dispatcher.clone(), + Some(inline), + producer, + )); (arena, Some(dispatcher)) } }; @@ -762,6 +774,22 @@ impl RenderTask { break; } } + Ok(TicketPayload::ShmAudio(audio)) => { + // M15 S3 process backend: the audio lives in a + // worker shm slot. Copy the samples out (the + // encoder needs an owned f32 buffer), hand them + // to the behavior, then release the slot. + let samples = audio.to_audio_samples(); + if let Err(e) = behavior.audio_downloaded(task, &samples) { + result = Err(e); + break; + } + if let Some(m) = oakrender::manager::RenderManager::global() { + m.release_audio_frame(&audio); + } else if let Some(d) = &private_dispatch { + d.release_audio_frame(&audio); + } + } Ok(_) => { result = Err(Error::Failed( "Audio render ticket delivered a non-audio payload".to_string(), @@ -1029,11 +1057,11 @@ fn push_finished(id: TicketId, result: TicketResult, dispatch: DispatchPtr) { 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). +/// Copy a process-backend shm frame out into an F32 CPU texture the +/// encoder consumes (M15 S2/S3). F32 slots (a forced F32 export ticket) +/// are read straight out — their bytes are already f32 RGBA little-endian, +/// no conversion; BGRA8 slots (the default preview path) convert once with +/// [`bgra8_to_f32_rgba`]. fn shm_frame_to_texture(frame: &ShmFrameRef) -> oakrender::texture::Texture { let meta = &frame.meta; let pixels = frame @@ -1041,6 +1069,17 @@ fn shm_frame_to_texture(frame: &ShmFrameRef) -> oakrender::texture::Texture { .slot_bytes(frame.slot) .get(..meta.data_size.max(0) as usize) .unwrap_or_default(); + if meta.format == oakcore_rs::PixelFormat::F32 as i32 { + // F32 slot: the encoder gets the pipeline samples with no round trip. + 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 = pixels.to_vec(); + return oakrender::texture::Texture::wrap_frame(f); + } let samples = bgra8_to_f32_rgba(pixels); let mut f = oakrender::texture::Frame::new(); f.width = meta.width; diff --git a/docs/zh/plans/riir/M15-render-process-isolation.md b/docs/zh/plans/riir/M15-render-process-isolation.md index b588cc646..749c3e8b4 100644 --- a/docs/zh/plans/riir/M15-render-process-isolation.md +++ b/docs/zh/plans/riir/M15-render-process-isolation.md @@ -2,7 +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 待做。 +> 进度：S1 完成（2026-08，协议 v2 + ProcessDispatcher + PreviewScheduler + worker 真实渲染，与线程池并存）；S2 完成（默认 Processes、删除 WorkerPool、oaktask/oak-cli/app 接入、上屏零拷贝、播放预渲染窗口）；S3 完成（2026-08-19：音频票走共享内存 + 播放异步预取 + 崩溃隔离覆盖音频、按票槽格式 F32 + 段按需扩容、worker/槽/批自适应策略 + 基准 bench_process）。 ## 1. 目标（用户原文要求） @@ -37,6 +37,7 @@ - **每 worker 一个 shm 段**（主进程 create、worker attach，复用 `SharedMemoryRegion` 双模式与 `FrameSlotPool`，key 规范沿用 `olive-rw--`）。 - **槽由主进程统一编址**：render 指令携带目标 slot id，worker 无权自行选槽 → 主进程可以把"预览缓存"直接建在槽上：预渲染帧的槽即缓存，上屏读槽即零拷贝；槽释放 = 缓存淘汰。当前帧（播放头）上屏路径：**shm 槽切片 → `queue.write_texture`**（GPU 上传是用户许可的唯一拷贝）。 - **槽格式**：viewer 预览票请求 **BGRA8**（新 force_format；worker 在渲染管线末端 F32→BGRA8 转换后写入槽——格式转换不是拷贝）；导出/全分辨率/scopes 票请求 **F32 RGBA**。槽大小 = 该段服务过的最大帧（64 对齐），段按需 ftruncate 扩容或重建。 + - **S3 落地**：槽格式按**票**指定（`force_format` 为 F32 的票得 F32 槽，worker 直写 F32；默认票保持 BGRA8）。段几何**按需增长**：调度器按请求所需 `slot_bytes` 过滤认领（`claim_batch(max_bytes)`），dispatcher 在 worker 空闲（无在飞帧）时重建更大段并通过新 handshake 让 worker 重挂（`reconfiguring` 门控），旧段随 `ShmFrameRef` 引用自然存活。导出票（encoder 请求 F32）直接读 F32 槽，消除 BGRA8→F32 回转。 - **槽数**：每 worker 8 槽起步（决定单 worker 在飞帧数；内存 = N × 8 × 8.3MB(BGRA8 1080p)）。 - ⚠️ **Spike 必验**：macOS POSIX `shm_open` 单段大小上限（目标 ≥ 512MB）。不达标则回退"临时文件 + `mmap(MAP_SHARED)`"（接口封装在 `SharedMemoryRegion` 内，加 backend 枚举，协议不变）。Linux 用 POSIX shm 即可。 @@ -86,13 +87,21 @@ 音频票同协议走 shm（AudioSamples 入槽）。S1/S2 可先保持主进程音频路径（崩溃风险主要来自视频插件），S3 迁移。 +> **S3 落地**：音频票走同一进程池。新增 `render_audio_batch` 消息（v2 增量）与 `AudioTicketSpec`；worker 侧经 `oakrender::eval::render_audio_samples_into` 直接混音入槽（`SLOT_FORMAT_AUDIO_F32 = 101`，复用 `FrameSlotMeta`：`width` 带采样率、`channel_count`/`linesize` 描述交错布局、`data_size` 为采样字节）。主进程经 `TicketPayload::ShmAudio(ShmAudioRef)` 读槽后 `release`。播放路径（`real.rs pull_audio_tick`）改为**异步预取**（`AUDIO_PREFETCH_CHUNKS = 4` 块 ≈ 66ms 提前量；ticket 完成经 poll 回调进通道，按 start_ts 排序入缓冲，实时拉取永不阻塞 UI 线程）；`submit_audio_chunk` 渲染失败补静音保持对齐。超过 `MAX_AUDIO_SLOT_BYTES`（64MB，约 3 分钟 48kHz 立体声）的音频范围（如长导出）由 dispatcher `post` 拒绝，arena 回退主进程内联渲染。dispatcher 不可用时同样回退内联（arena 的 `audio_fallback`，`InlineDispatcher::sync`），详见注释。 + ## 4. 分期 | 期 | 范围 | 验收 | |---|---|---| | 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（`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 收尾 | 性能报告；文档 | +| S3 | 音频迁移；压测调优（B、槽数、worker 数自适应）；README/docs 收尾 | 性能报告；文档 | +> **S3 验收（2026-08-19 完成）**： +> - 音频：`render_audio_batch` 协议 + worker 混音入槽 + 主进程 `ShmAudioRef` 读槽释放；播放异步预取（66ms 提前量、不阻塞 UI）；dispatcher 不可用时内联回退。集成测试覆盖 shm round-trip（4 块静音逐字节校验）与音频崩溃隔离（SIGSEGV 钩子，重启后仍出结果）。 +> - 按票槽格式：F32 票得 F32 槽（导出路径消除 BGRA8→F32 回转，`shm_frame_to_texture` 直读 F32）；段按需重建（`default_slots_for_bytes` 控制新段槽数，256MB/worker 预算）。 +> - 自适应策略：`default_slots_per_worker`（128MB/worker 段预算，BGRA8 1080p→8 槽、F32 1080p→4、F32 4K→2）、`default_batch_size`（`min(120/W, slots)`）、`default_worker_count`（核数/RAM/槽容量）。 +> - 基准：`cargo run --release -p oakrender --example bench_process [frames] [workers]` 量 1080p BGRA8 生成帧吞吐与相邻帧完成时间差（见下）。 +> - `cargo test` 全绿；`cargo check --workspace` 通过。 ## 5. 风险 diff --git a/src/oakui/real.rs b/src/oakui/real.rs index 82e81d056..26a06f01c 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::{BTreeMap, BTreeSet, HashMap, HashSet}; +use std::collections::{BTreeMap, BTreeSet, HashMap, HashSet, VecDeque}; use std::path::{Path, PathBuf}; use std::sync::mpsc; use std::sync::{Arc, Mutex}; @@ -274,6 +274,108 @@ struct PreviewWindow { slots: BTreeMap, } +// --------------------------------------------------------------------------- +// Playback audio prefetch (M15 S3) +// --------------------------------------------------------------------------- +// +// Real-time audio is pulled by the UI tick; rendering it through the worker +// pool adds one IPC round trip and can wait behind a busy render worker. To +// avoid dropouts the chunks are rendered AHEAD asynchronously (tickets +// complete on the dispatcher's poll) and buffered here; the pull never +// blocks on a worker. `AUDIO_PREFETCH_CHUNKS` ahead ≈ 64 ms of audio at +// 60 fps — enough to cover the delivery latency while keeping at most a +// handful of audio slots in flight (the dispatcher's credit flow control +// caps them per worker). + +/// How many audio chunks are kept rendered ahead of the playhead (M15 S3). +/// Each chunk is one sequence frame of audio; 4 frames ahead ≈ 66 ms at +/// 60 fps and ≈ 160 ms at 25 fps — enough to cover the delivery latency +/// while keeping at most a handful of audio slots in flight (the +/// dispatcher's credit flow control caps them per worker). +const AUDIO_PREFETCH_CHUNKS: i64 = 4; + +/// The playback-audio prefetch buffer: rendered chunks ordered by start +/// timestamp, plus the submission cursor. UI-thread-only (guarded by the +/// engine's `audio_prefetch` mutex so the engine stays `Sync`). +struct AudioPrefetch { + /// Sequence frame of the first buffered chunk (or `next_submit` when + /// the buffer is empty). + front_ts: i64, + /// Sequence frame of the next chunk to submit. + next_submit: i64, + /// Chunk length (sequence frames per tick) the buffer is built with. + chunk: i64, + /// Rendered chunks in start-ts order. + buffered: VecDeque<(i64, super::renderops::RenderedAudio)>, +} + +impl AudioPrefetch { + /// An empty, uninitialized prefetch. + fn new() -> Self { + Self { + front_ts: i64::MIN, + next_submit: i64::MIN, + chunk: 1, + buffered: VecDeque::new(), + } + } + + /// True when `frame` lies inside the submitted window + /// `[front_ts, next_submit)` — the playhead is being served. + fn covers(&self, frame: i64) -> bool { + self.front_ts != i64::MIN && self.front_ts <= frame && frame < self.next_submit + } + + /// Reset for a (re)start at `frame`: drop every buffered chunk and + /// restart the submission cursor there (a seek or a new project). + fn reset(&mut self, frame: i64, chunk: i64) { + self.front_ts = frame; + self.next_submit = frame; + self.chunk = chunk.max(1); + self.buffered.clear(); + } + + /// Insert a rendered chunk; stale arrivals (outside the submitted + /// window — a seek raced the render) are dropped. + fn insert(&mut self, ts: i64, data: super::renderops::RenderedAudio) { + if ts < self.front_ts || ts >= self.next_submit { + return; + } + if self.buffered.iter().any(|(t, _)| *t == ts) { + return; + } + let pos = self + .buffered + .iter() + .position(|(t, _)| *t > ts) + .unwrap_or(self.buffered.len()); + self.buffered.insert(pos, (ts, data)); + } + + /// Pop the chunk at `frame` (dropping any stale leading chunks). + /// Returns `None` when the chunk has not been rendered yet. + fn pop_at(&mut self, frame: i64) -> Option { + while let Some((ts, _)) = self.buffered.front() { + if *ts < frame { + let ts = self.buffered.pop_front().unwrap().0; + self.front_ts = ts + self.chunk; + } else { + break; + } + } + let (ts, data) = self.buffered.pop_front()?; + if ts == frame { + self.front_ts = ts + self.chunk; + Some(data) + } else { + // Gap: the chunk at `frame` is still rendering. Re-insert and + // report nothing (the output device zero-fills this tick). + self.buffered.push_front((ts, data)); + None + } + } +} + /// One background full-resolution render request (built on the UI thread /// at schedule time; the worker thread owns it from there). struct FullResRequest { @@ -912,12 +1014,26 @@ pub struct RealEngine { multicam_rx: Mutex>, /// The sending half of `multicam_rx` (cloned into every worker). multicam_tx: Mutex>, + /// M15 S3: async playback-audio prefetch — the process dispatcher + /// renders audio chunks ahead (completions arrive on the UI tick's + /// poll); the channel delivers `(start_ts, samples)` and the prefetch + /// state reorders them for the real-time pull. + audio_rx: Mutex>, + /// The sending half of `audio_rx` (cloned into every audio ticket). + audio_tx: Mutex>, + /// The audio prefetch buffer (see [`AudioPrefetch`]). + audio_prefetch: Mutex, } impl RealEngine { /// Render one tick's worth of audio at the program playhead and queue - /// it for playback (M12 P1). Failures are silent: playback continues - /// video-only. + /// it for playback (M12 P1; M15 S3: async worker-pool prefetch). The + /// audio chunks are rendered AHEAD in oak-worker (tickets posted + /// through the process dispatcher; completions arrive on the UI tick's + /// poll) and buffered by [`AudioPrefetch`], so the real-time pull never + /// blocks the UI thread on a busy render worker. Failures degrade to + /// silence; when the manager is down the channel stays empty and + /// playback continues video-only. fn pull_audio_tick(&mut self, cx: &mut Context) { let (Some(project), Some(seq)) = (self.project.clone(), self.sequence) else { return; @@ -925,14 +1041,46 @@ impl RealEngine { let Some(tb) = self.time_base() else { return; }; - let fps = self.frame_rate(); let frame = self.clock_frame(Monitor::Program, cx).0; if frame < 0 { return; } - // ~1/60 s of sequence per tick. - let chunk = ((fps.num as f64 / fps.den as f64) / 60.0).max(0.001) as i64; - let Ok(buf) = super::renderops::render_audio_range(&project, seq, frame, chunk, tb) else { + // One sequence frame of audio per chunk, keyed by the playhead + // frame. A per-tick wall-clock heuristic (~1/60 s) truncates to zero + // frames for sub-60 fps sequences, so render exactly one frame per + // playhead frame instead — the output device consumes one frame's + // audio per frame advance regardless of the rate. + let chunk: i64 = 1; + + let mut st = self.audio_prefetch.lock().unwrap_or_else(|e| e.into_inner()); + // Reset on seek / (re)start: the playhead must lie inside the + // submitted window [front_ts, next_submit). + if !st.covers(frame) { + st.reset(frame, chunk); + } + // Submit chunks to cover [next_submit, frame + PREFETCH ahead). In + // steady state the window moves by one chunk per tick, so exactly + // one new ticket is posted; the rest are already buffered. + let tx = self.audio_tx.lock().unwrap_or_else(|e| e.into_inner()).clone(); + let target = frame + chunk * AUDIO_PREFETCH_CHUNKS; + while st.next_submit < target { + let ts = st.next_submit; + let p = project.clone(); + if super::renderops::submit_audio_chunk(&p, seq, ts, chunk, tb, tx.clone()).is_err() { + break; + } + st.next_submit += chunk; + } + // Drain completed chunks. For the inline fallback backend the + // submit above already ran them synchronously into the channel; for + // the process backend they arrived on this tick's earlier poll. + let rx = self.audio_rx.lock().unwrap_or_else(|e| e.into_inner()); + while let Ok((ts, data)) = rx.try_recv() { + st.insert(ts, data); + } + drop(rx); + // Push the chunk at the current playhead to the output device. + let Some(buf) = st.pop_at(frame) else { return; }; if buf.sample_rate <= 0 || buf.channel_count <= 0 || buf.data.is_empty() { @@ -960,6 +1108,7 @@ impl RealEngine { let (full_res_tx, full_res_rx) = mpsc::channel::(); let (thumb_tx, thumb_rx) = mpsc::channel::(); let (multicam_tx, multicam_rx) = mpsc::channel::(); + let (audio_tx, audio_rx) = mpsc::channel::<(i64, super::renderops::RenderedAudio)>(); Self { project: None, sequence: None, @@ -996,6 +1145,9 @@ impl RealEngine { multicam_frames: Arc::new(Mutex::new(MulticamFrameCache::default())), multicam_rx: Mutex::new(multicam_rx), multicam_tx: Mutex::new(multicam_tx), + audio_rx: Mutex::new(audio_rx), + audio_tx: Mutex::new(audio_tx), + audio_prefetch: Mutex::new(AudioPrefetch::new()), } } @@ -2297,6 +2449,12 @@ impl RealEngine { } self.project = None; self.sequence = None; + // The audio prefetch belongs to the dropped project's sequence time: + // invalidate it so the next playback restarts the submission cursor. + self.audio_prefetch + .lock() + .unwrap_or_else(|e| e.into_inner()) + .reset(0, 1); // 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. @@ -5840,4 +5998,80 @@ mod tests { std::thread::sleep(Duration::from_millis(10)); } } + + // ---- M15 S3 audio prefetch ------------------------------------------ + + /// A lightweight `RenderedAudio` stand-in (the prefetch logic only + /// reads the fields, never renders). + fn audio_chunk(start: i64) -> (i64, crate::oakui::renderops::RenderedAudio) { + ( + start, + crate::oakui::renderops::RenderedAudio { + data: vec![0.0; 800], + sample_rate: 48000, + channel_count: 2, + }, + ) + } + + #[test] + fn audio_prefetch_orders_and_serves_chunks() { + let mut st = AudioPrefetch::new(); + assert!(!st.covers(0), "uninitialized prefetch covers nothing"); + st.reset(0, 10); + assert!(!st.covers(0), "nothing submitted yet"); + st.next_submit = 40; // pretend 4 chunks were submitted (0..40) + + // Out-of-order arrivals (different workers) are reordered. + st.insert(20, audio_chunk(20).1); + st.insert(0, audio_chunk(0).1); + st.insert(30, audio_chunk(30).1); + assert_eq!( + st.buffered.iter().map(|(t, _)| *t).collect::>(), + vec![0, 20, 30], + "sorted by start ts" + ); + // The chunk at the playhead is served. + let got = st.pop_at(0).expect("chunk 0 buffered"); + assert_eq!(got.sample_rate, 48000); + assert_eq!(st.buffered.len(), 2); + // A not-yet-rendered chunk reports nothing. + assert!(st.pop_at(10).is_none()); + // Stale arrivals (a seek raced the render) are dropped. + st.insert(-10, audio_chunk(-10).1); + st.insert(100, audio_chunk(100).1); + assert_eq!(st.buffered.len(), 2, "stale chunks dropped"); + } + + #[test] + fn audio_prefetch_resets_on_seek() { + let mut st = AudioPrefetch::new(); + st.reset(0, 10); + st.next_submit = 40; + st.insert(10, audio_chunk(10).1); + // A seek far ahead: the playhead is outside [front_ts, next_submit). + assert!(!st.covers(200)); + st.reset(200, 10); + assert_eq!(st.buffered.len(), 0, "old chunks dropped"); + assert!(st.pop_at(200).is_none()); + // A backward seek (the playhead behind the buffer front after it + // advanced) is a reset too. + st.reset(0, 10); + st.next_submit = 40; + st.insert(10, audio_chunk(10).1); + let _ = st.pop_at(10); // front_ts now 20 + assert!(!st.covers(5), "chunk 5 is behind the front"); + st.reset(5, 10); + assert_eq!(st.buffered.len(), 0); + } + + #[test] + fn audio_prefetch_drops_duplicate_arrivals() { + let mut st = AudioPrefetch::new(); + st.reset(0, 10); + st.next_submit = 30; + st.insert(10, audio_chunk(10).1); + st.insert(10, audio_chunk(10).1); + assert_eq!(st.buffered.len(), 1, "duplicates dropped"); + } } diff --git a/src/oakui/renderops.rs b/src/oakui/renderops.rs index 8e6b17aa6..01c887497 100644 --- a/src/oakui/renderops.rs +++ b/src/oakui/renderops.rs @@ -620,10 +620,92 @@ pub fn render_audio_range( sample_rate: samples.sample_rate, channel_count: samples.channel_count, }), + // M15 S3 process backend: the audio sits in a worker shm slot; copy + // the samples out and release the slot (the bytes must outlive it). + Ok(TicketPayload::ShmAudio(audio)) => { + let samples = audio.to_audio_samples(); + if let Some(m) = RenderManager::global() { + m.release_audio_frame(&audio); + } + Ok(RenderedAudio { + data: samples.samples, + sample_rate: samples.sample_rate, + channel_count: samples.channel_count, + }) + } _ => Err("audio render produced no samples".to_string()), } } +/// Submit one audio-chunk render for the async playback prefetch (M15 +/// S3): the chunk `[start_ts, start_ts + len_ts)` is rendered through the +/// process dispatcher; the completion copies the samples out of the shm +/// slot, releases it and sends `(start_ts, samples)` on `tx`. Render +/// errors send silence of the expected length so the playback buffer stays +/// aligned (the real-time path must never stall the UI thread on a worker). +pub fn submit_audio_chunk( + p: &ProjectRef, + seq: NodeId, + start_ts: i64, + len_ts: i64, + tb: (i64, i64), + tx: mpsc::Sender<(i64, RenderedAudio)>, +) -> Result<(), String> { + let range = TimeRange::new( + Rational::new(start_ts * tb.0, tb.1), + Rational::new((start_ts + len_ts) * tb.0, tb.1), + ); + let montage = audio_montage(p, seq, range); + let sample_rate = 48000; + let channel_layout = 0x3u64; + let channels = channel_layout.count_ones().max(1) as i32; + let m = RenderManager::global().ok_or_else(|| "render manager is not initialized".to_string())?; + let id = m.tickets.next_id(); + m.tickets.submit_audio_with_id( + id, + AudioTicketParams { + viewer: seq.identity(), + range, + sample_rate, + channel_layout, + montage, + }, + Box::new(move |result| { + let data = match result { + Ok(TicketPayload::Audio(samples)) => RenderedAudio { + data: samples.samples, + sample_rate: samples.sample_rate, + channel_count: samples.channel_count, + }, + Ok(TicketPayload::ShmAudio(audio)) => { + let samples = audio.to_audio_samples(); + if let Some(m) = RenderManager::global() { + m.release_audio_frame(&audio); + } + RenderedAudio { + data: samples.samples, + sample_rate: samples.sample_rate, + channel_count: samples.channel_count, + } + } + _ => { + // Silence of the expected length keeps the output buffer + // aligned (an underrun here just plays zeros). + let seconds = len_ts as f64 * tb.0 as f64 / tb.1 as f64; + let frames = (seconds * sample_rate as f64).round() as usize; + RenderedAudio { + data: vec![0.0; frames * channels as usize], + sample_rate, + channel_count: channels, + } + } + }; + let _ = tx.send((start_ts, data)); + }), + ); + Ok(()) +} + // --------------------------------------------------------------------------- // Export (the facade's `oakengine_task_create_export` + sync run over the // module export task)