Merge PR #1302

# Conflicts:
#	crates/preview/src/tests.rs

crates/preview/src/pool.rs

@@ -30,6 +30,11 @@ struct Entry<S, R> {
 
 struct Shared<S, R> {
     queue: Mutex<Vec<Entry<S, R>>>,
+    /// 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<Vec<S>>,
     cv: Condvar,
     shutdown: AtomicBool,
     /// The worker threads that have not ended yet, and the signal that one did.
@@ -75,6 +80,7 @@ impl<S: Copy + Eq + Hash + Send + 'static, R: Send + 'static> JobPool<S, R> {
         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"))]
@@ -122,6 +128,8 @@ impl<S: Copy + Eq + Hash + Send + 'static, R: Send + 'static> JobPool<S, R> {
                     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() {
@@ -230,6 +238,7 @@ impl<S: Copy + Eq + Hash + Send + 'static, R: Send + 'static> JobPool<S, R> {
         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;
@@ -240,14 +249,21 @@ impl<S: Copy + Eq + Hash + Send + 'static, R: Send + 'static> JobPool<S, R> {
 
 /// Pop the best job (highest priority, newest first). Blocks when `wait` until one is queued or
 /// the pool shuts down.
-fn take<S, R>(shared: &Shared<S, R>, wait: bool) -> Option<Entry<S, R>> {
+fn take<S: PartialEq + Copy, R>(shared: &Shared<S, R>, wait: bool) -> Option<Entry<S, R>> {
     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;
@@ -256,6 +272,28 @@ fn take<S, R>(shared: &Shared<S, R>, wait: bool) -> Option<Entry<S, R>> {
     }
 }
 
+/// The job for `slot` ended: a job queued for it may start.
+fn finished<S: PartialEq, R>(shared: &Shared<S, R>, 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, R>, S);
+
+impl<S: PartialEq, R> 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<S, R>);

crates/preview/src/tests.rs

@@ -391,4 +391,78 @@ fn pool_wait_keeps_the_finished_job_for_try_recv() {
     assert!(p.wait(std::time::Duration::ZERO), "kept until taken");
     assert_eq!(p.try_recv().map(|d| d.result), Some(7));
     assert!(p.try_recv().is_none());
+/// 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<u8, u32> = 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<std::sync::mpsc::Receiver<()>>| -> crate::pool::Job<u32> {
+        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<u8, u32> = 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));
 }