From a17d5be56a6b99248da37e2c9201e74abf7d68a8 Mon Sep 17 00:00:00 2001 From: Mike Solar Date: Wed, 19 Aug 2026 04:49:15 +0800 Subject: [PATCH] fix(render): assign batch slots in the worker's acquisition order MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A claim mixing audio and video tickets is delivered as the video message first and the audio message second, and the worker pops one free-ring slot per ticket in that message order, checking each pop against the assignment. The dispatcher however assigned slots in the scheduler's interleaved frame order, so every audio ticket inside a mixed batch mismatched, and each mismatch consumed a worker slot without recycling it — cascading into the 'slot assignment mismatch' flood and failed frames during playback. Slot assignment now partitions the claim: video tickets first, then audio. The mixed_audio_video integration test forces mixed claims (queue depth > slot count with immediate releases) and fails with the exact production signature when the fix is reverted. --- .../oak-worker/tests/procpool_integration.rs | 58 +++++++++++++++++++ crates/oakrender/src/procpool.rs | 26 +++++++-- 2 files changed, 78 insertions(+), 6 deletions(-) diff --git a/crates/oak-worker/tests/procpool_integration.rs b/crates/oak-worker/tests/procpool_integration.rs index 2c628654a..c1f8185a8 100644 --- a/crates/oak-worker/tests/procpool_integration.rs +++ b/crates/oak-worker/tests/procpool_integration.rs @@ -418,6 +418,64 @@ fn audio_tickets_roundtrip_through_shm_slots() { dispatcher.shutdown(); } +/// A claim batch that mixes audio and video tickets must assign slots in +/// the worker's acquisition order (the video message is processed first, +/// the audio message second). Interleaved assignment scrambled the +/// worker's free ring and flooded "slot assignment mismatch" failures +/// during playback — this is the regression guard. +#[test] +fn mixed_audio_video_batch_keeps_slot_assignment_order() { + let _guard = lock_test(); + // One worker, four slots: the first four posts dispatch singly (post + // pumps once itself), the remaining posts queue behind the busy + // slots. As completions are released below, claims gather up to + // `batch_size` queued tickets — mixed audio/video batches. + let dispatcher = ProcessDispatcher::new(config(1, 4)).expect("dispatcher config"); + dispatcher.start().expect("worker starts"); + + let results = Arc::new(Mutex::new(Vec::new())); + for i in 0..8 { + submit(&dispatcher, &results, 1, None); + submit_audio( + &dispatcher, + &results, + 1, + Rational::new(i, 24), + Rational::new(1, 24), + ); + } + // Drain completions, releasing each slot immediately so the queued + // tickets keep flowing into new (mixed) claims. + let deadline = Instant::now() + Duration::from_secs(60); + let mut total = 0usize; + loop { + dispatcher.poll(); + let done: Vec = results + .lock() + .unwrap_or_else(|e| e.into_inner()) + .drain(..) + .collect(); + total += done.len(); + for result in &done { + match result { + Ok(TicketPayload::ShmFrame(frame)) => dispatcher.release_frame(frame), + Ok(TicketPayload::ShmAudio(audio)) => dispatcher.release_audio_frame(audio), + Err(e) => panic!("mixed-batch ticket failed: {e}"), + _ => panic!("unexpected payload variant"), + } + } + if total >= 16 { + break; + } + if Instant::now() > deadline { + panic!("timeout waiting for the mixed batch ({total}/16)"); + } + std::thread::sleep(Duration::from_millis(2)); + } + + 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. diff --git a/crates/oakrender/src/procpool.rs b/crates/oakrender/src/procpool.rs index c3d388989..913b4b1e4 100644 --- a/crates/oakrender/src/procpool.rs +++ b/crates/oakrender/src/procpool.rs @@ -1199,13 +1199,27 @@ impl ProcessDispatcher { let Some(batch) = inner.scheduler.claim_batch(worker, credit, max_bytes) else { return; }; - let mut video_tickets = Vec::with_capacity(batch.frames.len()); - let mut audio_tickets: Vec = Vec::new(); - for req in &batch.frames { + // Slot assignment order MUST match the worker's acquisition + // order: the batch is delivered as the video message first and + // the audio message second, and the worker pops one slot per + // ticket in that message order, checking each pop against the + // assignment. Assigning in the scheduler's interleaved frame + // order scrambles the free ring (every audio ticket in a mixed + // batch mismatched, and each mismatch leaked a slot — the + // "slot assignment mismatch" flood). Two passes: video first. + let (video_reqs, audio_reqs): (Vec<_>, Vec<_>) = + batch.frames.iter().partition(|r| { + !inner + .tickets + .get(&r.payload) + .is_some_and(|pt| pt.audio.is_some()) + }); + let mut video_tickets = Vec::with_capacity(video_reqs.len()); + let mut audio_tickets: Vec = Vec::with_capacity(audio_reqs.len()); + for req in video_reqs.into_iter().chain(audio_reqs) { let ticket = req.payload; - let slot = match inner.workers[worker].free_slots.pop_front() { - Some(s) => s, - None => break, // credit accounting drifted; stop cleanly + let Some(slot) = inner.workers[worker].free_slots.pop_front() else { + break; // credit accounting drifted; stop cleanly }; inner.workers[worker].outstanding.insert(ticket, slot); let Some(pt) = inner.tickets.get(&ticket) else {