Merge #992

992: Fix a sleep race with broadcast jobs r=cuviper a=cuviper

The sleep code checks `has_injected_jobs` as the final step before
blocking the thread, but this was only looking at the normal pool-
injected jobs. With `broadcast`, there's now a way to inject jobs
directly on each thread, which they should also check individually
before going to sleep -- and now they do!


Co-authored-by: Josh Stone <[email protected]>

Author: bors[bot]

RELEASES.md

@@ -1,3 +1,7 @@
+# Release rayon-core 1.10.1 (2022-11-18)
+
+- Fixed a race condition with threads going to sleep while a broadcast starts.
+
 # Release rayon 1.6.0 / rayon-core 1.10.0 (2022-11-18)
 
 - The minimum supported `rustc` is now 1.56.

ci/compat-Cargo.lock

@@ -1,6 +1,6 @@
 [package]
 name = "rayon-core"
-version = "1.10.0"
+version = "1.10.1"
 authors = ["Niko Matsakis <[email protected]>",
            "Josh Stone <[email protected]>"]
 description = "Core APIs for Rayon"

rayon-core/Cargo.toml

@@ -200,3 +200,17 @@ fn spawn_broadcast_panic_many() {
     assert_eq!(rx.into_iter().count(), 7);
     assert_eq!(panic_rx.into_iter().count(), 4);
 }
+
+#[test]
+fn broadcast_sleep_race() {
+    let test_duration = time::Duration::from_secs(1);
+    let pool = ThreadPoolBuilder::new().num_threads(7).build().unwrap();
+    let start = time::Instant::now();
+    while start.elapsed() < test_duration {
+        pool.broadcast(|ctx| {
+            // A slight spread of sleep duration increases the chance that one
+            // of the threads will race in the pool's idle sleep afterward.
+            thread::sleep(time::Duration::from_micros(ctx.index() as u64));
+        });
+    }
+}

rayon-core/src/broadcast/test.rs

@@ -742,6 +742,10 @@ impl WorkerThread {
         }
     }
 
+    fn has_injected_job(&self) -> bool {
+        !self.stealer.is_empty() || self.registry.has_injected_job()
+    }
+
     /// Wait until the latch is set. Try to keep busy by popping and
     /// stealing tasks as necessary.
     #[inline]
@@ -779,7 +783,7 @@ impl WorkerThread {
             } else {
                 self.registry
                     .sleep
-                    .no_work_found(&mut idle_state, latch, || self.registry.has_injected_job())
+                    .no_work_found(&mut idle_state, latch, || self.has_injected_job())
             }
         }