diff --git a/crates/preview/src/pool.rs b/crates/preview/src/pool.rs index 4dbd35661..285cfd802 100644 --- a/crates/preview/src/pool.rs +++ b/crates/preview/src/pool.rs @@ -30,6 +30,11 @@ struct Entry { struct Shared { queue: Mutex>>, + /// Slots with a job running: their next job waits until it ends, so a slot never has two at + /// once (a slider drag submits faster than renders finish; overlapping renders of one view + /// were mostly thrown away, and took the GPU from the one that is shown). Locked inside + /// `queue`. + busy: Mutex>, cv: Condvar, shutdown: AtomicBool, /// The worker threads that have not ended yet, and the signal that one did. @@ -73,6 +78,7 @@ impl JobPool { JobPool { shared: Arc::new(Shared { queue: Mutex::new(Vec::new()), + busy: Mutex::new(Vec::new()), cv: Condvar::new(), shutdown: AtomicBool::new(false), #[cfg(not(target_arch = "wasm32"))] @@ -119,6 +125,8 @@ impl JobPool { let _ended = Ended(&shared); while let Some(e) = take(&shared, true) { let t0 = std::time::Instant::now(); + // the slot is free again when the job ends, also if it panics + let _free = Free(&shared, e.slot); let result = (e.job)(); let done = Done { slot: e.slot, key: e.key, result, ms: t0.elapsed().as_secs_f64() * 1000.0 }; if tx.send(done).is_err() { @@ -219,6 +227,7 @@ impl JobPool { let mut ran = 0; while ran < n { let Some(e) = take(&self.shared, false) else { break }; + let _free = Free(&self.shared, e.slot); let result = (e.job)(); let _ = self.tx.send(Done { slot: e.slot, key: e.key, result, ms: 0.0 }); ran += 1; @@ -229,14 +238,21 @@ impl JobPool { /// Pop the best job (highest priority, newest first). Blocks when `wait` until one is queued or /// the pool shuts down. -fn take(shared: &Shared, wait: bool) -> Option> { +fn take(shared: &Shared, wait: bool) -> Option> { let mut q = shared.queue.lock().unwrap_or_else(|e| e.into_inner()); loop { if shared.shutdown.load(Ordering::Relaxed) { return None; } - if let Some(i) = q.iter().enumerate().max_by_key(|(_, e)| (e.priority, e.seq)).map(|(i, _)| i) { - return Some(q.swap_remove(i)); + // (a slot with a job running waits for it: see `Shared::busy`) + let best = { + let busy = shared.busy.lock().unwrap_or_else(|e| e.into_inner()); + q.iter().enumerate().filter(|(_, e)| !busy.contains(&e.slot)).max_by_key(|(_, e)| (e.priority, e.seq)).map(|(i, _)| i) + }; + if let Some(i) = best { + let e = q.swap_remove(i); + shared.busy.lock().unwrap_or_else(|e| e.into_inner()).push(e.slot); + return Some(e); } if !wait { return None; @@ -245,6 +261,28 @@ fn take(shared: &Shared, wait: bool) -> Option> { } } +/// The job for `slot` ended: a job queued for it may start. +fn finished(shared: &Shared, slot: &S) { + { + // under the queue lock: a worker is either before its look at `busy` or waiting + let _q = shared.queue.lock().unwrap_or_else(|e| e.into_inner()); + let mut busy = shared.busy.lock().unwrap_or_else(|e| e.into_inner()); + if let Some(i) = busy.iter().position(|s| s == slot) { + busy.swap_remove(i); + } + } + shared.cv.notify_all(); +} + +/// Frees a job's slot when the job ends (see [`finished`]). +struct Free<'a, S: PartialEq, R>(&'a Shared, S); + +impl Drop for Free<'_, S, R> { + fn drop(&mut self) { + finished(self.0, &self.1); + } +} + /// Takes a worker off the live list when its thread ends (also when a job panicked). #[cfg(not(target_arch = "wasm32"))] struct Ended<'a, S, R>(&'a Shared); diff --git a/crates/preview/src/tests.rs b/crates/preview/src/tests.rs index d733ad829..718fa0d20 100644 --- a/crates/preview/src/tests.rs +++ b/crates/preview/src/tests.rs @@ -351,3 +351,79 @@ fn disk_cache_never_touches_foreign_files() { assert!(!disk::is_cache_file("01", "0123456789abcdef0123456789abcdef.jpg.bak")); let _ = std::fs::remove_dir_all(&dir); } + +/// Wait up to `secs` for `f` to hold. +fn eventually(secs: u64, mut f: impl FnMut() -> bool) -> bool { + let t0 = std::time::Instant::now(); + while t0.elapsed() < std::time::Duration::from_secs(secs) { + if f() { + return true; + } + std::thread::sleep(std::time::Duration::from_millis(5)); + } + false +} + +/// A slider drag submits for a view faster than its renders finish. A newer job for a slot used +/// to start beside the running one on another worker: one view's renders piled up on the GPU and +/// most were thrown away. Now a slot runs one job at a time: the newer one waits (replacing an +/// older one still queued), other slots keep going, and it starts when the slot's job ends. +#[test] +fn a_slot_runs_one_job_at_a_time() { + use std::sync::atomic::{AtomicUsize, Ordering}; + let mut p: JobPool = JobPool::new(4); + let (release, released) = std::sync::mpsc::channel::<()>(); + let (started, has_started) = std::sync::mpsc::channel(); + let running = Arc::new(AtomicUsize::new(0)); + let most = Arc::new(AtomicUsize::new(0)); + let job = |n: u32, wait: Option>| -> crate::pool::Job { + let (running, most, started) = (running.clone(), most.clone(), started.clone()); + Box::new(move || { + let now = running.fetch_add(1, Ordering::SeqCst) + 1; + most.fetch_max(now, Ordering::SeqCst); + let _ = started.send(n); + if let Some(w) = wait { + let _ = w.recv_timeout(std::time::Duration::from_secs(30)); + } + running.fetch_sub(1, Ordering::SeqCst); + n + }) + }; + p.submit(1, 1, 9, job(1, Some(released))); + assert_eq!(has_started.recv().unwrap(), 1); + // two newer requests for the busy slot: the second replaces the first, neither starts + p.submit(1, 2, 9, job(2, None)); + p.submit(1, 3, 9, job(3, None)); + // another slot is not held up + p.submit(2, 9, 1, Box::new(|| 9)); + let mut done = Vec::new(); + assert!(eventually(10, || { + done.extend(std::iter::from_fn(|| p.try_recv()).map(|d| d.result)); + done.contains(&9) + })); + std::thread::sleep(std::time::Duration::from_millis(50)); + assert!(has_started.try_recv().is_err(), "a second job of slot 1 started beside the first"); + assert!(p.is_queued(1)); + drop(release); + assert!(eventually(10, || { + done.extend(std::iter::from_fn(|| p.try_recv()).map(|d| d.result)); + done.contains(&3) + })); + assert!(!done.contains(&2), "the replaced job never runs"); + assert_eq!(most.load(Ordering::SeqCst), 1, "slot 1 ran two jobs at once"); +} + +/// A job that panics still frees its slot: the slot's next job runs (on another worker). +#[test] +fn a_panicking_job_frees_its_slot() { + let mut p: JobPool = JobPool::new(2); + p.submit(1, 1, 9, Box::new(|| panic!("simulated render panic"))); + assert!(eventually(10, || !p.is_queued(1) && p.live_workers() == 1)); + p.submit(1, 2, 9, Box::new(|| 2)); + let mut got = None; + assert!(eventually(10, || { + got = got.or_else(|| p.try_recv().map(|d| d.result)); + got.is_some() + })); + assert_eq!(got, Some(2)); +}