feat(oakrender): render-process isolation S3 - audio over shm, per-ticket slot formats, tuning

- Audio tickets join the process backend: render_audio_batch wire
  message, workers mix straight into shm slots (SLOT_FORMAT_AUDIO_F32),
  ShmAudio payload with release semantics, crash isolation covers audio
  renders; playback audio uses an async 4-chunk prefetch drained on the
  UI tick (also fixes the sub-60fps chunk truncation bug); oversized
  ranges and dispatcher outages fall back to in-process inline.
- Per-ticket slot formats: force_format is honored (exports request
  F32 slots, dropping the BGRA8 round-trip and its 8-bit quantization);
  segments grow on demand via worker-idle rebuild with generation
  handoff; the scheduler filters over-capacity tickets.
- Adaptive defaults: 128-256MB/worker segment budgets drive slots per
  worker, batch size follows workers/slots; bench_process example
  measures throughput and adjacent-frame completion deltas
  (e.g. 4 workers: 841 fps, 4.6ms mean delta).
This commit is contained in:
2026-08-19 00:42:03 +08:00
parent 194d761ade
commit adf2cef32c
18 changed files with 2156 additions and 185 deletions
Generated
-1
View File
@@ -4924,7 +4924,6 @@ dependencies = [
name = "oakundo"
version = "0.1.0"
dependencies = [
"oakcore-rs",
"thiserror 2.0.20",
]
+13
View File
@@ -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()),
}
}
+4 -1
View File
@@ -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
+324 -16
View File
@@ -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::<Vec<_>>(),
});
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<u32, String> {
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(&params).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(&params, &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<AudioTicketParams, String> {
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<MontageClip> = 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::<Value>(&line)
let typ = serde_json::from_str::<Value>(&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<u8> = Vec::new();
s.handle_render_audio_batch_stream(&batch.to_string(), &mut out)
.unwrap();
let lines: Vec<Value> = 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<f32> = 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<u8> = Vec::new();
s.handle_render_audio_batch_stream(&batch.to_string(), &mut out)
.unwrap();
let lines: Vec<Value> = 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) -------------------------------------
+305 -3
View File
@@ -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<VideoTicketPara
viewer: 1,
time,
force_size: Some((64, 64)),
force_format: Some(PixelFormat::F32),
// No forced format: the dispatcher's default slot format (BGRA8)
// applies — the preview path these tests exercise (M15 S3: an F32
// force would now request an F32 slot instead).
force_format: None,
cache: None,
cache_dir: None,
cache_id: None,
@@ -94,6 +99,7 @@ fn submit(
node_identity: 1,
time: Rational::new(i as i64, 25),
params: params(Rational::new(i as i64, 25), footage),
audio: None,
// Never invoked on the process backend (workers render from
// the wire spec); must still be a valid producer.
produce: Arc::new(|_, _| {
@@ -314,3 +320,299 @@ fn worker_decodes_real_footage_into_slot() {
dispatcher.shutdown();
}
/// Submit an empty-montage audio range pull through the dispatcher
/// (M15 S3): the worker mixes silence into a shm slot and the main
/// process reads it back as [`TicketPayload::ShmAudio`].
fn submit_audio(
dispatcher: &ProcessDispatcher,
results: &Arc<Mutex<Vec<TicketResult>>>,
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();
}
+24 -11
View File
@@ -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.
+210
View File
@@ -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 <http://www.gnu.org/licenses/>.
//! 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/<profile>/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/<profile>/examples/bench_process -> target/<profile>/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<usize> = 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<f64> = 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::<f64>() / 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();
}
+124 -7
View File
@@ -553,6 +553,20 @@ pub fn render_footage_frame(
pub fn render_audio_samples(
params: &crate::ticket::AudioTicketParams,
) -> Result<crate::ticket::TicketPayload> {
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<usize> {
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 &params.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(&params).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(&params).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(&params, &mut dst).unwrap();
let expected: Vec<u8> = 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<f32> = 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(&params, &mut dst).is_err());
}
}
+107 -1
View File
@@ -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<i64>,
}
/// 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<WireMontageClip>,
}
/// `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<AudioTicketSpec>,
}
/// `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 -----------------------------------------
+35 -23
View File
@@ -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<dyn JobDispatch>,
/// 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<dyn JobDispatch>,
/// Ticket arena.
pub tickets: Arc<TicketArena>,
@@ -96,27 +96,33 @@ impl RenderManager {
eval::render_produced_frame(time, params)
.map(crate::ticket::TicketPayload::Video)
});
let (dispatch, audio_dispatch): (Arc<dyn JobDispatch>, Arc<dyn JobDispatch>) =
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<dyn JobDispatch>,
Arc<dyn JobDispatch>,
Option<Arc<dyn JobDispatch>>,
) = 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.
+470 -67
View File
@@ -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<ShmRegionView>,
/// 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<f32> {
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<u64> {
#[cfg(target_os = "macos")]
@@ -390,7 +513,8 @@ pub struct DispatcherConfig {
pub worker_bin: Option<PathBuf>,
/// 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<Child>,
stdin: Option<std::process::ChildStdin>,
shm: Arc<ShmRegionView>,
/// 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<u32>,
/// 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<ShmRegionView>, slots: u32) -> WorkerHandle {
fn shell(
index: usize,
generation: u64,
shm: Arc<ShmRegionView>,
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<VideoTicketParams>,
/// Audio ticket params when this ticket is an audio range pull (M15
/// S3); `None` for video tickets.
audio: Option<Arc<AudioTicketParams>>,
done: Option<Completion>,
}
@@ -533,6 +674,10 @@ struct Inner {
next_ticket: i64,
events_rx: mpsc::Receiver<WorkerEvent>,
events_tx: mpsc::Sender<WorkerEvent>,
/// 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<Arc<ProcessDispatcher>> {
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::<HelloCapsMsg>(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<AudioTicketSpec> = 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<PathBuf> {
)))
}
/// 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]
+79 -17
View File
@@ -86,6 +86,11 @@ pub struct FrameRequest<P> {
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<P: Clone> PreviewScheduler<P> {
/// 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<ClaimedBatch<P>> {
pub fn claim_batch(
&mut self,
worker: usize,
credit: usize,
max_bytes: usize,
) -> Option<ClaimedBatch<P>> {
if worker >= self.workers || credit == 0 {
return None;
}
@@ -208,7 +221,8 @@ impl<P: Clone> PreviewScheduler<P> {
.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<P: Clone> PreviewScheduler<P> {
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<usize> {
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<i64> = 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<i64> = 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<u64> = 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<i64> = 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<u64> = 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<u64> = 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());
}
}
+67 -21
View File
@@ -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<dyn JobDispatch>,
audio_dispatch: Arc<dyn JobDispatch>,
/// 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<Arc<dyn JobDispatch>>,
slots: Mutex<HashMap<TicketId, Arc<TicketSlot>>>,
shutting_down: AtomicBool,
producer: Producer,
@@ -273,11 +286,26 @@ impl TicketArena {
video: Arc<dyn JobDispatch>,
audio: Arc<dyn JobDispatch>,
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<dyn JobDispatch>,
audio: Arc<dyn JobDispatch>,
audio_fallback: Option<Arc<dyn JobDispatch>>,
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<TicketSlot>| {
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
}
+14 -1
View File
@@ -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<T>(m: &Mutex<T>) -> 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<VideoTicketParams>,
/// 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<Arc<AudioTicketParams>>,
/// 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());
+46 -7
View File
@@ -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;
@@ -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-<pid>-<index>`)。
- **槽由主进程统一编址**:render 指令携带目标 slot id,worker 无权自行选槽 → 主进程可以把"预览缓存"直接建在槽上:预渲染帧的槽即缓存,上屏读槽即零拷贝;槽释放 = 缓存淘汰。当前帧(播放头)上屏路径:**shm 槽切片 → `queue.write_texture`**(GPU 上传是用户许可的唯一拷贝)。
- **槽格式**viewer 预览票请求 **BGRA8**(新 force_formatworker 在渲染管线末端 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 @@
音频票同协议走 shmAudioSamples 入槽)。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. 分期
| 期 | 范围 | 验收 |
|---|---|---|
| S1crates only | shm spikemacOS 段上限);协议 v2ProcessDispatcher + 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/scopesscopes 读 BGRA8);播放前向窗口(120 帧可配)喂 PreviewSchedulercpu_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. 风险
+241 -7
View File
@@ -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<i64, ShmFrameRef>,
}
// ---------------------------------------------------------------------------
// 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<super::renderops::RenderedAudio> {
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<mpsc::Receiver<MulticamAngleEvent>>,
/// The sending half of `multicam_rx` (cloned into every worker).
multicam_tx: Mutex<mpsc::Sender<MulticamAngleEvent>>,
/// 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<mpsc::Receiver<(i64, super::renderops::RenderedAudio)>>,
/// The sending half of `audio_rx` (cloned into every audio ticket).
audio_tx: Mutex<mpsc::Sender<(i64, super::renderops::RenderedAudio)>>,
/// The audio prefetch buffer (see [`AudioPrefetch`]).
audio_prefetch: Mutex<AudioPrefetch>,
}
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<Self>) {
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::<FullResEvent>();
let (thumb_tx, thumb_rx) = mpsc::channel::<ThumbEvent>();
let (multicam_tx, multicam_rx) = mpsc::channel::<MulticamAngleEvent>();
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<_>>(),
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");
}
}
+82
View File
@@ -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)