diff --git a/crates/oak-app/src/oakui/real.rs b/crates/oak-app/src/oakui/real.rs index 8afe03951..d4d27f9fb 100644 --- a/crates/oak-app/src/oakui/real.rs +++ b/crates/oak-app/src/oakui/real.rs @@ -1656,12 +1656,15 @@ impl RealEngine { // generation (the old pending/claimed requests are cancelled). let mut windows = self.preview_windows.lock().unwrap_or_else(|e| e.into_inner()); let window = windows.entry(monitor).or_default(); + // Cancel/release calls fire completions synchronously and those + // completions lock `preview_windows`, so they must run AFTER this + // guard is dropped (calling them here self-deadlocks the UI thread). + let mut stale_sequences: Vec = Vec::new(); + let mut stale_keys: Vec<(u64, i64, u64)> = Vec::new(); + let mut stale_slots: Vec = Vec::new(); if window.sequence != node_id || window.generation != self.preview_generation { - m.cancel_preview_sequence(window.sequence); - for slot in window.slots.values() { - m.release_frame(slot); - } - window.slots.clear(); + stale_sequences.push(window.sequence); + stale_slots.extend(std::mem::take(&mut window.slots).into_values()); window.submitted.clear(); window.sequence = node_id; window.generation = self.preview_generation; @@ -1678,7 +1681,7 @@ impl RealEngine { .collect(); for f in stale { if let Some(slot) = window.slots.remove(&f) { - m.release_frame(&slot); + stale_slots.push(slot); } } window @@ -1697,8 +1700,7 @@ impl RealEngine { .filter(|f| *f < keep_from && !window.slots.contains_key(f)) .collect(); for f in stale_pending { - m.dispatch - .cancel_preview_frame(window.sequence, f, window.generation); + stale_keys.push((window.sequence, f, window.generation)); window.submitted.remove(&f); } let new_frames: Vec = (playhead.max(0)..end) @@ -1706,6 +1708,20 @@ impl RealEngine { .collect(); drop(windows); + // Fire the deferred cancels/releases outside the `preview_windows` + // lock (see the comment at the guard above). Cancels come before the + // new submissions below: a sequence cancel drops every pending + // request of that sequence regardless of version. + for sequence in stale_sequences { + m.cancel_preview_sequence(sequence); + } + for (sequence, frame, version) in stale_keys { + m.dispatch.cancel_preview_frame(sequence, frame, version); + } + for slot in &stale_slots { + m.release_frame(slot); + } + for frame in new_frames { let params = match monitor { Monitor::Program => super::renderops::sequence_frame_params( @@ -1793,17 +1809,26 @@ impl RealEngine { /// slots (edit / selection change / project drop / preview-media /// invalidation). fn cancel_preview_windows(&mut self) { - let mut windows = self.preview_windows.lock().unwrap_or_else(|e| e.into_inner()); - let m = RenderManager::global(); - for window in windows.values_mut() { - if let Some(m) = &m { - m.cancel_preview_sequence(window.sequence); - for slot in window.slots.values() { + // Collect the teardown work under the lock, then run it outside: + // `cancel_preview_sequence` fires completions synchronously and those + // completions lock `preview_windows` (self-deadlock otherwise). + let mut pending: Vec<(u64, Vec)> = Vec::new(); + { + let mut windows = self.preview_windows.lock().unwrap_or_else(|e| e.into_inner()); + for window in windows.values_mut() { + let slots: Vec = + std::mem::take(&mut window.slots).into_values().collect(); + pending.push((window.sequence, slots)); + window.submitted.clear(); + } + } + if let Some(m) = RenderManager::global() { + for (sequence, slots) in pending { + m.cancel_preview_sequence(sequence); + for slot in &slots { m.release_frame(slot); } } - window.slots.clear(); - window.submitted.clear(); } } @@ -1811,18 +1836,24 @@ impl RealEngine { /// the old footage's window is stale even though the program window is /// untouched). fn cancel_preview_window(&mut self, monitor: Monitor) { - let mut windows = self.preview_windows.lock().unwrap_or_else(|e| e.into_inner()); - let Some(window) = windows.get_mut(&monitor) else { - return; + // Same lock-order rule as `cancel_preview_windows`: run the cancel + // and releases outside the `preview_windows` guard. + let pending = { + let mut windows = self.preview_windows.lock().unwrap_or_else(|e| e.into_inner()); + let Some(window) = windows.get_mut(&monitor) else { + return; + }; + let slots: Vec = + std::mem::take(&mut window.slots).into_values().collect(); + window.submitted.clear(); + (window.sequence, slots) }; if let Some(m) = RenderManager::global() { - m.cancel_preview_sequence(window.sequence); - for slot in window.slots.values() { + m.cancel_preview_sequence(pending.0); + for slot in &pending.1 { m.release_frame(slot); } } - window.slots.clear(); - window.submitted.clear(); } /// Invalidates every cached/rendered preview frame: the CPU cache is @@ -3159,6 +3190,17 @@ impl AudioMeterDataSource for RealEngine { // AppEngine // --------------------------------------------------------------------------- +impl Drop for RealEngine { + /// The pre-render windows hold worker shm slots; a dropped engine must + /// hand them back. `ShmFrameRef` has no self-release on drop, so without + /// this every closed project (and every test engine) permanently shrank + /// the shared slot pool until later playback windows starved (the + /// full-suite `playback_window_supplies_playhead_frames` failure). + fn drop(&mut self) { + self.cancel_preview_windows(); + } +} + impl AppEngine for RealEngine { type Clock = RealClock; @@ -6579,6 +6621,10 @@ mod tests { // empty for the sequence's birth). #[gpui::test] async fn new_sequence_has_default_two_video_two_audio_tracks(cx: &mut gpui::TestAppContext) { + // Serializes with the other engine tests: `new_project` clears the + // GLOBAL undo stack, and this test asserts on it — running lock-free + // raced a parallel test's undo history. + let _media = media_lock(); let engine = cx.update(|cx| cx.new(|cx| RealEngine::create(cx))); cx.update(|app| engine.update(app, |engine, cx| engine.new_project(cx))); let kinds: Vec = @@ -6899,6 +6945,56 @@ mod tests { let _ = std::fs::remove_file(&media); } + /// Probe: a single interactive seek must render the seeked frame + /// without hanging the UI path (the ruler mouse-down path goes through + /// request_frame + a synchronous cpu_frame render). + #[gpui::test] + async fn interactive_seek_renders_without_hanging(cx: &mut gpui::TestAppContext) { + let _media = media_lock(); + let _worker = WorkerBinGuard::set(); + let engine = cx.update(|cx| cx.new(|cx| RealEngine::create(cx))); + cx.update(|app| engine.update(app, |engine, cx| engine.new_project(cx))); + let media = std::path::Path::new(env!("CARGO_MANIFEST_DIR")).join("tests/demo.mp4"); + cx.update(|app| { + engine.update(app, |engine, cx| { + engine.import_footage(media.clone(), cx).expect("import") + }) + }); + let name = media.file_name().unwrap().to_string_lossy().into_owned(); + let entry = cx.read(|app| { + engine.read(app).roots().into_iter().find(|e| e.name.as_ref() == name) + }).expect("imported footage is listed"); + cx.update(|app| { + engine.update(app, |engine, cx| { + engine.drop_footage(entry.id, TrackKind::Video, 0, Frame(0), cx) + }) + }); + // Interactive seek (not playing): a single synchronous render of the + // target frame. If this hangs, the seek render path deadlocks. + cx.update(|app| { + engine.update(app, |engine, cx| engine.request_frame(Monitor::Program, Frame(24), cx)) + }); + let img = cx.read(|app| engine.read(app).cpu_frame(Monitor::Program, app)); + let bytes = img.as_bytes(0).expect("frame bytes"); + assert!(!bytes.is_empty(), "seeked frame has content"); + + // The reported freeze: play (the preview window fills and holds shm + // slots), THEN an interactive seek — the synchronous render must not + // deadlock against the window's held slots. + cx.update(|app| engine.update(app, |engine, cx| engine.play(Monitor::Program, cx))); + for _ in 0..30 { + cx.update(|app| engine.update(app, |engine, cx| engine.tick(cx))); + std::thread::sleep(Duration::from_millis(10)); + } + cx.update(|app| engine.update(app, |engine, cx| engine.pause(Monitor::Program, cx))); + cx.update(|app| { + engine.update(app, |engine, cx| engine.request_frame(Monitor::Program, Frame(48), cx)) + }); + let img = cx.read(|app| engine.read(app).cpu_frame(Monitor::Program, app)); + let bytes = img.as_bytes(0).expect("seek-after-play frame bytes"); + assert!(!bytes.is_empty(), "seek after playback has content"); + } + /// The production scenario: real 1080p media on the timeline, driving /// the actual `cpu_frame` display path the viewer paints with (not /// just the window internals). The displayed frame must track the diff --git a/crates/oak-app/src/oakui/renderops.rs b/crates/oak-app/src/oakui/renderops.rs index 7817ddb2b..fec225498 100644 --- a/crates/oak-app/src/oakui/renderops.rs +++ b/crates/oak-app/src/oakui/renderops.rs @@ -477,8 +477,14 @@ fn validate_geometry(width: i32, height: i32, tb: (i64, i64)) -> Result<(), Stri fn render_video(params: VideoTicketParams) -> Result { let m = RenderManager::global().ok_or_else(|| "render manager is not initialized".to_string())?; let id = m.tickets.next_id(); + if std::env::var_os("OAK_DEBUG_DISPATCH").is_some() { + eprintln!("renderops: sync render submit arena ticket {}", id.0); + } m.tickets.submit_video_with_id(id, params, Box::new(|_| {})); m.tickets.wait(id).map_err(|e| e.to_string())?; + if std::env::var_os("OAK_DEBUG_DISPATCH").is_some() { + eprintln!("renderops: sync render wait done arena ticket {}", id.0); + } let result = m .tickets .result(id) diff --git a/crates/oak-render/src/procpool.rs b/crates/oak-render/src/procpool.rs index 3794d8a76..2b68360a7 100644 --- a/crates/oak-render/src/procpool.rs +++ b/crates/oak-render/src/procpool.rs @@ -1203,6 +1203,9 @@ impl ProcessDispatcher { Some(h) => h, None => return, }; + if std::env::var_os("OAK_DEBUG_DISPATCH").is_some() { + eprintln!("procpool: worker {worker} frame_ready ticket {ticket} slot {slot}"); + } if handle.outstanding.remove(&ticket).is_none() { return; // late / duplicate / post-restart frame } @@ -1274,6 +1277,9 @@ impl ProcessDispatcher { error: &str, fired: &mut Vec<(Completion, TicketResult)>, ) { + if std::env::var_os("OAK_DEBUG_DISPATCH").is_some() { + eprintln!("procpool: worker {worker} frame_failed ticket {ticket}: {error}"); + } let slot = { let handle = match inner.workers.get_mut(worker) { Some(h) => h, @@ -1338,6 +1344,19 @@ impl ProcessDispatcher { } let max_bytes = inner.workers[worker].slot_bytes; let Some(batch) = inner.scheduler.claim_batch(worker, credit, max_bytes) else { + // Starvation diagnostics (OAK_DEBUG_DISPATCH=1): pending work + // exists but this worker claimed none of it — log why (no + // credit, shard mismatch or oversized slot) instead of + // spinning silently (the seek-starvation hang). + if std::env::var_os("OAK_DEBUG_DISPATCH").is_some() { + let pending = inner.scheduler.pending_len(); + if pending > 0 { + eprintln!( + "procpool: worker {worker} idle with {pending} pending (credit {credit}, slot_bytes {max_bytes}): {:?}", + inner.scheduler.pending_summary() + ); + } + } return; }; // Slot assignment order MUST match the worker's acquisition @@ -1397,6 +1416,10 @@ impl ProcessDispatcher { Value::String(crate::ipc::TYPE_RENDER_BATCH.to_string()), ); } + if std::env::var_os("OAK_DEBUG_DISPATCH").is_some() { + let ids: Vec = msg.tickets.iter().map(|t| t.ticket).collect(); + eprintln!("procpool: worker {worker} sent video batch {} tickets {ids:?}", msg.batch_id); + } if self.send_json(&mut inner.workers[worker], &value).is_err() { inner.workers[worker].state = WorkerState::Dead; return; @@ -1689,6 +1712,23 @@ impl JobDispatch for ProcessDispatcher { payload: id, slot_bytes, }; + if std::env::var_os("OAK_DEBUG_DISPATCH").is_some() { + let pools: Vec = inner + .workers + .iter() + .enumerate() + .map(|(i, w)| { + format!("w{i}: free {} held {} out {} state {:?}", + w.free_slots.len(), w.held.len(), w.outstanding.len(), w.state) + }) + .collect(); + eprintln!( + "procpool: post ticket {id} key ({}, {}, {}) prio {:?} shard {} | {}", + key.sequence, key.frame, key.version, request.priority, + key.frame.rem_euclid(inner.scheduler.workers() as i64), + pools.join(" | ") + ); + } match inner.scheduler.submit(request) { SubmitOutcome::Accepted => {} SubmitOutcome::Replaced(old) => { diff --git a/crates/oak-render/src/scheduler.rs b/crates/oak-render/src/scheduler.rs index a7f87d820..a767da348 100644 --- a/crates/oak-render/src/scheduler.rs +++ b/crates/oak-render/src/scheduler.rs @@ -37,7 +37,11 @@ //! re-dispatch is failure recovery, not stealing. //! - **Flow control.** [`PreviewScheduler::claim_batch`] never claims //! more frames than the caller's `credit` (the worker's free shm -//! slot count): slots are the credit. +//! slot count): slots are the credit. Playback/Background claims +//! leave one credit unused — the per-worker interactive reserve, so +//! a window batch can never drain the last slot and starve a +//! UI-blocking seek (window slots are released by UI-thread +//! consumption; a seek waiting on them deadlocks the UI). //! - **Cancellation.** Frame keys carry a parameter `version`; //! submitting a newer version of a key invalidates the older one, //! and [`PreviewScheduler::cancel_sequence`] drops a whole sequence. @@ -175,17 +179,27 @@ impl PreviewScheduler

{ /// Submit a frame request. An already-pending request with the same /// key is replaced; a key already claimed (in flight) is rejected — /// the dispatcher must cancel/re-version it first. + /// + /// Seek-priority requests (interactive frame / real-time audio) are + /// claimable by ANY worker, not just their interleaved shard: the + /// no-stealing shard rule exists to keep ADJACENT PLAYBACK frames + /// finishing together, but pinning a single urgent frame to one worker + /// starves it whenever that worker's slots are all held by the window + /// (the UI thread then waits on the seek while the window's slot + /// releases run on that same thread — the seek-starvation deadlock). pub fn submit(&mut self, request: FrameRequest

) -> SubmitOutcome

{ if self.claimed.contains_key(&request.key) { return SubmitOutcome::InFlight; } + let any_worker = request.priority == FramePriority::Seek; if let Some(entry) = self.pending.iter_mut().find(|e| e.request.key == request.key) { + entry.any_worker = any_worker; let old = std::mem::replace(&mut entry.request, request); return SubmitOutcome::Replaced(old); } self.pending.push(PendingEntry { request, - any_worker: false, + any_worker, }); SubmitOutcome::Accepted } @@ -196,16 +210,19 @@ impl PreviewScheduler

{ } /// Claim the next batch for `worker`: the worker's interleaved shard - /// (frame number `≡ worker (mod W)`, plus any crash-requeued frames), - /// ordered by priority class / playhead distance / ascending frame, - /// capped at `min(batch_size, credit)`. Requests needing more than + /// (frame number `≡ worker (mod W)`, plus crash-requeued frames and + /// Seek-priority requests — both claimable by any worker), ordered by + /// priority class / playhead distance / ascending frame, capped at + /// `min(batch_size, credit)`. Playback/Background claims additionally + /// leave one credit unused (the per-worker interactive reserve; see + /// [`PreviewScheduler::submit`]). 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). + /// oversized, or only the reserve remains). /// - /// Claimed frames never go to another worker while in flight (no - /// stealing). + /// Claimed Playback frames never go to another worker while in flight + /// (no stealing). pub fn claim_batch( &mut self, worker: usize, @@ -238,7 +255,31 @@ impl PreviewScheduler

{ .then(ra.key.frame.cmp(&rb.key.frame)) .then(ra.key.sequence.cmp(&rb.key.sequence)) }); - indexes.truncate(self.batch_size.min(credit)); + // Per-worker interactive reserve: Playback/Background claims must + // leave one slot free. Window frames are released by UI-thread + // consumption/eviction, so a batch that drains the worker's last + // slot can starve a UI-blocking seek FOREVER; seeks and audio + // (Seek priority) complete on the worker without UI involvement, + // so they may use the last slot. (The global + // `preview_window_capacity` reserve alone did not prevent this: + // its accounting is pool-wide, while slot exhaustion happens per + // worker.) + let cap = self.batch_size.min(credit); + let mut taken: Vec = Vec::with_capacity(cap); + for &i in &indexes { + if taken.len() >= cap { + break; + } + let seek = self.pending[i].request.priority == FramePriority::Seek; + if !seek && taken.len() + 1 >= credit { + continue; // keep the reserve slot free + } + taken.push(i); + } + if taken.is_empty() { + return None; + } + let indexes = taken; let batch_id = self.next_batch_id; self.next_batch_id += 1; @@ -286,6 +327,27 @@ impl PreviewScheduler

{ self.claimed.remove(key).map(|c| c.request) } + /// Pending-request count (dispatcher diagnostics). + pub fn pending_len(&self) -> usize { + self.pending.len() + } + + /// One-line summary of the pending queue for starvation debugging + /// (frame, pinned shard, needed slot bytes, any_worker). + pub fn pending_summary(&self) -> Vec<(i64, i64, usize, bool)> { + self.pending + .iter() + .map(|e| { + ( + e.request.key.frame, + e.request.key.frame.rem_euclid(self.workers as i64), + e.request.slot_bytes, + e.any_worker, + ) + }) + .collect() + } + /// Report a claimed frame as permanently failed (frame_failed; the /// main process paints the fallback). The claim is dropped WITHOUT /// re-dispatch — a render error is not a crash. Returns the request @@ -551,10 +613,47 @@ mod tests { } // Zero credit claims nothing. assert!(s.claim_batch(0, 0, 1024).is_none()); - // Credit 3 claims exactly 3 (free slots are the credit). + // Credit 3 claims 2 playback frames: the last slot stays free as + // the per-worker interactive reserve (window frames are released + // by UI-thread consumption, so a full drain can starve a + // UI-blocking seek). let batch = s.claim_batch(0, 3, 1024).unwrap(); - assert_eq!(batch.frames.len(), 3); - assert_eq!(s.pending_count(), 7); + assert_eq!(batch.frames.len(), 2); + assert_eq!(s.pending_count(), 8); + } + + #[test] + fn playback_claims_keep_the_reserve_but_seeks_may_use_it() { + let mut s: PreviewScheduler = PreviewScheduler::new(1, 100); + for f in 0..4 { + s.submit(req(1, f, FramePriority::Playback)); + } + // One free slot: playback claims nothing (reserve kept)... + assert!(s.claim_batch(0, 1, 1024).is_none()); + assert_eq!(s.pending_count(), 4); + // ...but a Seek (interactive frame / real-time audio) may use it: + // seeks complete on the worker without UI-thread involvement, so + // taking the last slot cannot deadlock the UI. + s.submit(req(1, 100, FramePriority::Seek)); + let batch = s.claim_batch(0, 1, 1024).unwrap(); + assert_eq!(batch.frames.len(), 1); + assert_eq!(batch.frames[0].priority, FramePriority::Seek); + } + + #[test] + fn seek_is_claimable_by_any_worker() { + // The no-stealing shard rule applies to Playback frames (adjacent + // frames finish together); a single urgent seek pinned to a full + // worker would starve even while other workers idle. + let mut s: PreviewScheduler = PreviewScheduler::new(4, 100); + s.submit(req(1, 3, FramePriority::Seek)); // shard 3 + let batch = s.claim_batch(0, 4, 1024).unwrap(); + assert_eq!(batch.frames.len(), 1); + assert_eq!(batch.frames[0].key.frame, 3); + // Playback frames stay pinned to their shard. + s.submit(req(1, 7, FramePriority::Playback)); // shard 3 + assert!(s.claim_batch(0, 4, 1024).is_none()); + assert!(s.claim_batch(3, 4, 1024).is_some()); } #[test] @@ -664,7 +763,7 @@ mod tests { for f in 0..6 { s.submit(req(7, f, FramePriority::Playback)); } - let _ = s.claim_batch(0, 4, 1024).unwrap(); // claims 4 of sequence 7 + let _ = s.claim_batch(0, 4, 1024).unwrap(); // claims 3 of sequence 7 (reserve) 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");