diff --git a/rayon-core/Cargo.toml b/rayon-core/Cargo.toml
index 3f57bef15..3ace30ac1 100644
--- a/rayon-core/Cargo.toml
+++ b/rayon-core/Cargo.toml
@@ -22,6 +22,7 @@ num_cpus = "1.2"
 crossbeam-channel = "0.5.0"
 crossbeam-deque = "0.8.1"
 crossbeam-utils = "0.8.0"
+smallvec = "1.11.0"
 
 [dev-dependencies]
 rand = "0.8"
diff --git a/rayon-core/src/broadcast/mod.rs b/rayon-core/src/broadcast/mod.rs
index f9cfc47ac..cf3f15046 100644
--- a/rayon-core/src/broadcast/mod.rs
+++ b/rayon-core/src/broadcast/mod.rs
@@ -4,6 +4,7 @@ use crate::registry::{Registry, WorkerThread};
 use crate::scope::ScopeLatch;
 use std::fmt;
 use std::marker::PhantomData;
+use std::sync::atomic::{AtomicBool, Ordering};
 use std::sync::Arc;
 
 mod test;
@@ -100,13 +101,22 @@ where
     OP: Fn(BroadcastContext<'_>) -> R + Sync,
     R: Send,
 {
+    let current_thread = WorkerThread::current();
+    let current_thread_addr = current_thread as usize;
+    let started = &AtomicBool::new(false);
     let f = move |injected: bool| {
         debug_assert!(injected);
+
+        // Mark as started if we are on the thread that initiated the broadcast.
+        if current_thread_addr == WorkerThread::current() as usize {
+            started.store(true, Ordering::Relaxed);
+        }
+
         BroadcastContext::with(&op)
     };
 
     let n_threads = registry.num_threads();
-    let current_thread = WorkerThread::current().as_ref();
+    let current_thread = current_thread.as_ref();
     let tlv = crate::tlv::get();
     let latch = ScopeLatch::with_count(n_threads, current_thread);
     let jobs: Vec<_> = (0..n_threads)
@@ -116,8 +126,16 @@ where
 
     registry.inject_broadcast(job_refs);
 
+    let current_thread_job_id = current_thread
+        .and_then(|worker| (registry.id() == worker.registry.id()).then(|| worker))
+        .map(|worker| jobs[worker.index].as_job_ref().id());
+
     // Wait for all jobs to complete, then collect the results, maybe propagating a panic.
-    latch.wait(current_thread);
+    latch.wait(
+        current_thread,
+        || started.load(Ordering::Relaxed),
+        |job| Some(job.id()) == current_thread_job_id,
+    );
     jobs.into_iter().map(|job| job.into_result()).collect()
 }
 
@@ -133,7 +151,7 @@ where
 {
     let job = ArcJob::new({
         let registry = Arc::clone(registry);
-        move || {
+        move |_| {
             registry.catch_unwind(|| BroadcastContext::with(&op));
             registry.terminate(); // (*) permit registry to terminate now
         }
diff --git a/rayon-core/src/broadcast/test.rs b/rayon-core/src/broadcast/test.rs
index 3ae11f7f6..792b6329a 100644
--- a/rayon-core/src/broadcast/test.rs
+++ b/rayon-core/src/broadcast/test.rs
@@ -63,6 +63,7 @@ fn spawn_broadcast_self() {
 }
 
 #[test]
+#[ignore]
 #[cfg_attr(any(target_os = "emscripten", target_family = "wasm"), ignore)]
 fn broadcast_mutual() {
     let count = AtomicUsize::new(0);
@@ -97,6 +98,7 @@ fn spawn_broadcast_mutual() {
 }
 
 #[test]
+#[ignore]
 #[cfg_attr(any(target_os = "emscripten", target_family = "wasm"), ignore)]
 fn broadcast_mutual_sleepy() {
     let count = AtomicUsize::new(0);
diff --git a/rayon-core/src/job.rs b/rayon-core/src/job.rs
index 394c7576b..e93b22fd4 100644
--- a/rayon-core/src/job.rs
+++ b/rayon-core/src/job.rs
@@ -26,6 +26,11 @@ pub(super) trait Job {
     unsafe fn execute(this: *const ());
 }
 
+#[derive(PartialEq, Eq, Hash, Copy, Clone, Debug)]
+pub(super) struct JobRefId {
+    pointer: usize,
+}
+
 /// Effectively a Job trait object. Each JobRef **must** be executed
 /// exactly once, or else data may leak.
 ///
@@ -54,11 +59,11 @@ impl JobRef {
         }
     }
 
-    /// Returns an opaque handle that can be saved and compared,
-    /// without making `JobRef` itself `Copy + Eq`.
     #[inline]
-    pub(super) fn id(&self) -> impl Eq {
-        (self.pointer, self.execute_fn)
+    pub(super) fn id(&self) -> JobRefId {
+        JobRefId {
+            pointer: self.pointer as usize,
+        }
     }
 
     #[inline]
@@ -102,8 +107,13 @@ where
         JobRef::new(self)
     }
 
-    pub(super) unsafe fn run_inline(self, stolen: bool) -> R {
-        self.func.into_inner().unwrap()(stolen)
+    pub(super) unsafe fn run_inline(&self, stolen: bool) {
+        let func = (*self.func.get()).take().unwrap();
+        (*self.result.get()) = match unwind::halt_unwinding(|| func(stolen)) {
+            Ok(x) => JobResult::Ok(x),
+            Err(x) => JobResult::Panic(x),
+        };
+        Latch::set(&self.latch);
     }
 
     pub(super) unsafe fn into_result(self) -> R {
@@ -136,7 +146,7 @@ where
 /// (Probably `StackJob` should be refactored in a similar fashion.)
 pub(super) struct HeapJob<BODY>
 where
-    BODY: FnOnce() + Send,
+    BODY: FnOnce(JobRefId) + Send,
 {
     job: BODY,
     tlv: Tlv,
@@ -144,7 +154,7 @@ where
 
 impl<BODY> HeapJob<BODY>
 where
-    BODY: FnOnce() + Send,
+    BODY: FnOnce(JobRefId) + Send,
 {
     pub(super) fn new(tlv: Tlv, job: BODY) -> Box<Self> {
         Box::new(HeapJob { job, tlv })
@@ -168,12 +178,13 @@ where
 
 impl<BODY> Job for HeapJob<BODY>
 where
-    BODY: FnOnce() + Send,
+    BODY: FnOnce(JobRefId) + Send,
 {
     unsafe fn execute(this: *const ()) {
+        let pointer = this as usize;
         let this = Box::from_raw(this as *mut Self);
         tlv::set(this.tlv);
-        (this.job)();
+        (this.job)(JobRefId { pointer });
     }
 }
 
@@ -181,14 +192,14 @@ where
 /// be turned into multiple `JobRef`s and called multiple times.
 pub(super) struct ArcJob<BODY>
 where
-    BODY: Fn() + Send + Sync,
+    BODY: Fn(JobRefId) + Send + Sync,
 {
     job: BODY,
 }
 
 impl<BODY> ArcJob<BODY>
 where
-    BODY: Fn() + Send + Sync,
+    BODY: Fn(JobRefId) + Send + Sync,
 {
     pub(super) fn new(job: BODY) -> Arc<Self> {
         Arc::new(ArcJob { job })
@@ -212,11 +223,12 @@ where
 
 impl<BODY> Job for ArcJob<BODY>
 where
-    BODY: Fn() + Send + Sync,
+    BODY: Fn(JobRefId) + Send + Sync,
 {
     unsafe fn execute(this: *const ()) {
+        let pointer = this as usize;
         let this = Arc::from_raw(this as *mut Self);
-        (this.job)();
+        (this.job)(JobRefId { pointer });
     }
 }
 
diff --git a/rayon-core/src/join/mod.rs b/rayon-core/src/join/mod.rs
index 032eec9c4..6d25b8386 100644
--- a/rayon-core/src/join/mod.rs
+++ b/rayon-core/src/join/mod.rs
@@ -1,9 +1,10 @@
+use crate::job::JobRef;
 use crate::job::StackJob;
 use crate::latch::SpinLatch;
-use crate::registry::{self, WorkerThread};
-use crate::tlv::{self, Tlv};
+use crate::registry;
+use crate::tlv;
 use crate::unwind;
-use std::any::Any;
+use std::sync::atomic::{AtomicBool, Ordering};
 
 use crate::FnContext;
 
@@ -135,68 +136,41 @@ where
         // Create virtual wrapper for task b; this all has to be
         // done here so that the stack frame can keep it all live
         // long enough.
-        let job_b = StackJob::new(tlv, call_b(oper_b), SpinLatch::new(worker_thread));
+        let job_b_started = AtomicBool::new(false);
+        let job_b = StackJob::new(
+            tlv,
+            |migrated| {
+                job_b_started.store(true, Ordering::Relaxed);
+                call_b(oper_b)(migrated)
+            },
+            SpinLatch::new(worker_thread),
+        );
         let job_b_ref = job_b.as_job_ref();
         let job_b_id = job_b_ref.id();
         worker_thread.push(job_b_ref);
 
         // Execute task a; hopefully b gets stolen in the meantime.
         let status_a = unwind::halt_unwinding(call_a(oper_a, injected));
-        let result_a = match status_a {
-            Ok(v) => v,
-            Err(err) => join_recover_from_panic(worker_thread, &job_b.latch, err, tlv),
-        };
-
-        // Now that task A has finished, try to pop job B from the
-        // local stack.  It may already have been popped by job A; it
-        // may also have been stolen. There may also be some tasks
-        // pushed on top of it in the stack, and we will have to pop
-        // those off to get to it.
-        while !job_b.latch.probe() {
-            if let Some(job) = worker_thread.take_local_job() {
-                if job_b_id == job.id() {
-                    // Found it! Let's run it.
-                    //
-                    // Note that this could panic, but it's ok if we unwind here.
 
-                    // Restore the TLV since we might have run some jobs overwriting it when waiting for job b.
-                    tlv::set(tlv);
-
-                    let result_b = job_b.run_inline(injected);
-                    return (result_a, result_b);
-                } else {
-                    worker_thread.execute(job);
-                }
-            } else {
-                // Local deque is empty. Time to steal from other
-                // threads.
-                worker_thread.wait_until(&job_b.latch);
-                debug_assert!(job_b.latch.probe());
-                break;
-            }
-        }
+        // Wait for job B or execute it if it's in the local queue.
+        worker_thread.wait_for_jobs::<_, false>(
+            &job_b.latch,
+            || job_b_started.load(Ordering::Relaxed),
+            |job| job.id() == job_b_id,
+            |job: JobRef| {
+                debug_assert_eq!(job.id(), job_b_id);
+                job_b.run_inline(injected);
+            },
+        );
 
         // Restore the TLV since we might have run some jobs overwriting it when waiting for job b.
         tlv::set(tlv);
 
+        let result_a = match status_a {
+            Ok(v) => v,
+            Err(err) => unwind::resume_unwinding(err),
+        };
+
         (result_a, job_b.into_result())
     })
 }
-
-/// If job A panics, we still cannot return until we are sure that job
-/// B is complete. This is because it may contain references into the
-/// enclosing stack frame(s).
-#[cold] // cold path
-unsafe fn join_recover_from_panic(
-    worker_thread: &WorkerThread,
-    job_b_latch: &SpinLatch<'_>,
-    err: Box<dyn Any + Send>,
-    tlv: Tlv,
-) -> ! {
-    worker_thread.wait_until(job_b_latch);
-
-    // Restore the TLV since we might have run some jobs overwriting it when waiting for job b.
-    tlv::set(tlv);
-
-    unwind::resume_unwinding(err)
-}
diff --git a/rayon-core/src/join/test.rs b/rayon-core/src/join/test.rs
index b303dbc81..af3597a59 100644
--- a/rayon-core/src/join/test.rs
+++ b/rayon-core/src/join/test.rs
@@ -97,6 +97,7 @@ fn join_context_both() {
 }
 
 #[test]
+#[ignore]
 #[cfg_attr(any(target_os = "emscripten", target_family = "wasm"), ignore)]
 fn join_context_neither() {
     // If we're already in a 1-thread pool, neither job should be stolen.
diff --git a/rayon-core/src/latch.rs b/rayon-core/src/latch.rs
index de4327234..ceef5e967 100644
--- a/rayon-core/src/latch.rs
+++ b/rayon-core/src/latch.rs
@@ -177,11 +177,6 @@ impl<'r> SpinLatch<'r> {
             ..SpinLatch::new(thread)
         }
     }
-
-    #[inline]
-    pub(super) fn probe(&self) -> bool {
-        self.core_latch.probe()
-    }
 }
 
 impl<'r> AsCoreLatch for SpinLatch<'r> {
diff --git a/rayon-core/src/registry.rs b/rayon-core/src/registry.rs
index 15ceb6b0c..19bb018f9 100644
--- a/rayon-core/src/registry.rs
+++ b/rayon-core/src/registry.rs
@@ -10,6 +10,7 @@ use crate::{
     ReleaseThreadHandler, StartHandler, ThreadPoolBuildError, ThreadPoolBuilder, Yield,
 };
 use crossbeam_deque::{Injector, Steal, Stealer, Worker};
+use smallvec::SmallVec;
 use std::cell::Cell;
 use std::collections::hash_map::DefaultHasher;
 use std::fmt;
@@ -840,14 +841,83 @@ impl WorkerThread {
     /// stealing tasks as necessary.
     #[inline]
     pub(super) unsafe fn wait_until<L: AsCoreLatch + ?Sized>(&self, latch: &L) {
+        self.wait_or_steal_until(latch, false)
+    }
+
+    /// Wait until the latch is set. Executes local jobs if `is_job` is true for them and
+    /// `all_jobs_started` still returns false.
+    #[inline]
+    pub(super) unsafe fn wait_for_jobs<L: AsCoreLatch + ?Sized, const BROADCAST_JOBS: bool>(
+        &self,
+        latch: &L,
+        mut all_jobs_started: impl FnMut() -> bool,
+        mut is_job: impl FnMut(&JobRef) -> bool,
+        mut execute_job: impl FnMut(JobRef),
+    ) {
+        let mut jobs = SmallVec::<[JobRef; 8]>::new();
+        let mut broadcast_jobs = SmallVec::<[JobRef; 8]>::new();
+
+        // Make sure all jobs have started.
+        while !all_jobs_started() {
+            if let Some(job) = self.worker.pop() {
+                if is_job(&job) {
+                    // Found a job, let's run it.
+                    execute_job(job);
+                } else {
+                    jobs.push(job);
+                }
+            } else {
+                if BROADCAST_JOBS {
+                    let broadcast_job = loop {
+                        match self.stealer.steal() {
+                            Steal::Success(job) => break Some(job),
+                            Steal::Empty => break None,
+                            Steal::Retry => {}
+                        }
+                    };
+                    if let Some(job) = broadcast_job {
+                        if is_job(&job) {
+                            // Found a job, let's run it.
+                            self.execute(job);
+                        } else {
+                            broadcast_jobs.push(job);
+                        }
+                    }
+                }
+                break;
+            }
+        }
+
+        // Restore the jobs that we weren't looking for.
+        for job in jobs.into_iter().rev() {
+            self.worker.push(job);
+        }
+        if BROADCAST_JOBS {
+            let broadcasts = self.registry.broadcasts.lock().unwrap();
+            for job in broadcast_jobs.into_iter() {
+                broadcasts[self.index].push(job);
+            }
+        }
+
+        // Wait for the jobs to finish.
+        self.wait_until(latch);
+        debug_assert!(latch.as_core_latch().probe());
+    }
+
+    #[inline]
+    pub(super) unsafe fn wait_or_steal_until<L: AsCoreLatch + ?Sized>(
+        &self,
+        latch: &L,
+        steal: bool,
+    ) {
         let latch = latch.as_core_latch();
         if !latch.probe() {
-            self.wait_until_cold(latch);
+            self.wait_until_cold(latch, steal);
         }
     }
 
     #[cold]
-    unsafe fn wait_until_cold(&self, latch: &CoreLatch) {
+    unsafe fn wait_until_cold(&self, latch: &CoreLatch, steal: bool) {
         // the code below should swallow all panics and hence never
         // unwind; but if something does wrong, we want to abort,
         // because otherwise other code in rayon may assume that the
@@ -857,15 +927,17 @@ impl WorkerThread {
 
         let mut idle_state = self.registry.sleep.start_looking(self.index, latch);
         while !latch.probe() {
-            if let Some(job) = self.find_work() {
-                self.registry.sleep.work_found(idle_state);
-                self.execute(job);
-                idle_state = self.registry.sleep.start_looking(self.index, latch);
-            } else {
-                self.registry
-                    .sleep
-                    .no_work_found(&mut idle_state, latch, &self)
+            if steal {
+                if let Some(job) = self.find_work() {
+                    self.registry.sleep.work_found(idle_state);
+                    self.execute(job);
+                    idle_state = self.registry.sleep.start_looking(self.index, latch);
+                    continue;
+                }
             }
+            self.registry
+                .sleep
+                .no_work_found(&mut idle_state, latch, &self, steal);
         }
 
         // If we were sleepy, we are not anymore. We "found work" --
@@ -988,7 +1060,7 @@ unsafe fn main_loop(thread: ThreadBuilder) {
         terminate_addr: my_terminate_latch.as_core_latch().addr(),
     });
     registry.acquire_thread();
-    worker_thread.wait_until(my_terminate_latch);
+    worker_thread.wait_or_steal_until(my_terminate_latch, true);
 
     // Should not be any work left in our queue.
     debug_assert!(worker_thread.take_local_job().is_none());
diff --git a/rayon-core/src/scope/mod.rs b/rayon-core/src/scope/mod.rs
index 1b74f274d..ff6cf6834 100644
--- a/rayon-core/src/scope/mod.rs
+++ b/rayon-core/src/scope/mod.rs
@@ -6,18 +6,19 @@
 //! [`join()`]: ../join/join.fn.html
 
 use crate::broadcast::BroadcastContext;
-use crate::job::{ArcJob, HeapJob, JobFifo, JobRef};
+use crate::job::{ArcJob, HeapJob, JobFifo, JobRef, JobRefId};
 use crate::latch::{CountLatch, CountLockLatch, Latch};
 use crate::registry::{global_registry, in_worker, Registry, WorkerThread};
 use crate::tlv::{self, Tlv};
 use crate::unwind;
 use std::any::Any;
+use std::collections::HashSet;
 use std::fmt;
 use std::marker::PhantomData;
 use std::mem::ManuallyDrop;
 use std::ptr;
 use std::sync::atomic::{AtomicPtr, Ordering};
-use std::sync::Arc;
+use std::sync::{Arc, Mutex};
 
 #[cfg(test)]
 mod test;
@@ -60,7 +61,7 @@ pub(super) enum ScopeLatch {
     Blocking { latch: CountLockLatch },
 }
 
-struct ScopeBase<'scope> {
+pub(super) struct ScopeBase<'scope> {
     /// thread registry where `scope()` was executed or where `in_place_scope()`
     /// should spawn jobs.
     registry: Arc<Registry>,
@@ -72,6 +73,12 @@ struct ScopeBase<'scope> {
     /// latch to track job counts
     job_completed_latch: ScopeLatch,
 
+    /// Jobs that have been spawned, but not yet started.
+    pending_jobs: Mutex<HashSet<JobRefId>>,
+
+    /// The worker which will wait on scope completion, if any.
+    worker: Option<usize>,
+
     /// You can think of a scope as containing a list of closures to execute,
     /// all of which outlive `'scope`.  They're not actually required to be
     /// `Sync`, but it's still safe to let the `Scope` implement `Sync` because
@@ -544,13 +551,20 @@ impl<'scope> Scope<'scope> {
         BODY: FnOnce(&Scope<'scope>) + Send + 'scope,
     {
         let scope_ptr = ScopePtr(self);
-        let job = HeapJob::new(self.base.tlv, move || unsafe {
+        let job = HeapJob::new(self.base.tlv, move |id| unsafe {
             // SAFETY: this job will execute before the scope ends.
             let scope = scope_ptr.as_ref();
+
+            // Mark this job as started.
+            scope.base.pending_jobs.lock().unwrap().remove(&id);
+
             ScopeBase::execute_job(&scope.base, move || body(scope))
         });
         let job_ref = self.base.heap_job_ref(job);
 
+        // Mark this job as pending.
+        self.base.pending_jobs.lock().unwrap().insert(job_ref.id());
+
         // Since `Scope` implements `Sync`, we can't be sure that we're still in a
         // thread of this pool, so we can't just push to the local worker thread.
         // Also, this might be an in-place scope.
@@ -566,13 +580,21 @@ impl<'scope> Scope<'scope> {
         BODY: Fn(&Scope<'scope>, BroadcastContext<'_>) + Send + Sync + 'scope,
     {
         let scope_ptr = ScopePtr(self);
-        let job = ArcJob::new(move || unsafe {
+        let job = ArcJob::new(move |id| unsafe {
             // SAFETY: this job will execute before the scope ends.
             let scope = scope_ptr.as_ref();
             let body = &body;
+
+            let current_index = WorkerThread::current().as_ref().map(|worker| worker.index);
+            if current_index == scope.base.worker {
+                // Mark this job as started on the scope's worker thread
+                scope.base.pending_jobs.lock().unwrap().remove(&id);
+            }
+
             let func = move || BroadcastContext::with(move |ctx| body(scope, ctx));
             ScopeBase::execute_job(&scope.base, func)
         });
+
         self.base.inject_broadcast(job)
     }
 }
@@ -604,23 +626,24 @@ impl<'scope> ScopeFifo<'scope> {
         BODY: FnOnce(&ScopeFifo<'scope>) + Send + 'scope,
     {
         let scope_ptr = ScopePtr(self);
-        let job = HeapJob::new(self.base.tlv, move || unsafe {
+        let job = HeapJob::new(self.base.tlv, move |id| unsafe {
             // SAFETY: this job will execute before the scope ends.
             let scope = scope_ptr.as_ref();
+
+            // Mark this job as started.
+            scope.base.pending_jobs.lock().unwrap().remove(&id);
+
             ScopeBase::execute_job(&scope.base, move || body(scope))
         });
         let job_ref = self.base.heap_job_ref(job);
 
-        // If we're in the pool, use our scope's private fifo for this thread to execute
-        // in a locally-FIFO order. Otherwise, just use the pool's global injector.
-        match self.base.registry.current_thread() {
-            Some(worker) => {
-                let fifo = &self.fifos[worker.index()];
-                // SAFETY: this job will execute before the scope ends.
-                unsafe { worker.push(fifo.push(job_ref)) };
-            }
-            None => self.base.registry.inject(job_ref),
-        }
+        // Mark this job as pending.
+        self.base.pending_jobs.lock().unwrap().insert(job_ref.id());
+
+        // Since `Scope` implements `Sync`, we can't be sure that we're still in a
+        // thread of this pool, so we can't just push to the local worker thread.
+        // Also, this might be an in-place scope.
+        self.base.registry.inject_or_push(job_ref);
     }
 
     /// Spawns a job into every thread of the fork-join scope `self`. This job will
@@ -632,13 +655,21 @@ impl<'scope> ScopeFifo<'scope> {
         BODY: Fn(&ScopeFifo<'scope>, BroadcastContext<'_>) + Send + Sync + 'scope,
     {
         let scope_ptr = ScopePtr(self);
-        let job = ArcJob::new(move || unsafe {
+        let job = ArcJob::new(move |id| unsafe {
             // SAFETY: this job will execute before the scope ends.
             let scope = scope_ptr.as_ref();
             let body = &body;
+
+            let current_index = WorkerThread::current().as_ref().map(|worker| worker.index);
+            if current_index == scope.base.worker {
+                // Mark this job as started on the scope's worker thread
+                scope.base.pending_jobs.lock().unwrap().remove(&id);
+            }
+
             let func = move || BroadcastContext::with(move |ctx| body(scope, ctx));
             ScopeBase::execute_job(&scope.base, func)
         });
+
         self.base.inject_broadcast(job)
     }
 }
@@ -655,7 +686,9 @@ impl<'scope> ScopeBase<'scope> {
             registry: Arc::clone(registry),
             panic: AtomicPtr::new(ptr::null_mut()),
             job_completed_latch: ScopeLatch::new(owner),
+            pending_jobs: Mutex::new(HashSet::new()),
             marker: PhantomData,
+            worker: owner.map(|owner| owner.index),
             tlv: tlv::get(),
         }
     }
@@ -666,7 +699,7 @@ impl<'scope> ScopeBase<'scope> {
 
     fn heap_job_ref<FUNC>(&self, job: Box<HeapJob<FUNC>>) -> JobRef
     where
-        FUNC: FnOnce() + Send + 'scope,
+        FUNC: FnOnce(JobRefId) + Send + 'scope,
     {
         unsafe {
             self.increment();
@@ -674,10 +707,22 @@ impl<'scope> ScopeBase<'scope> {
         }
     }
 
-    fn inject_broadcast<FUNC>(&self, job: Arc<ArcJob<FUNC>>)
+    fn inject_broadcast<FUNC>(&self, mut job: Arc<ArcJob<FUNC>>)
     where
-        FUNC: Fn() + Send + Sync + 'scope,
+        FUNC: Fn(JobRefId) + Send + Sync + 'scope,
     {
+        if self.worker.is_some() {
+            unsafe {
+                // Get an id of the job
+                let raw = Arc::into_raw(job);
+                let id = JobRef::new(raw).id();
+                job = Arc::from_raw(raw);
+
+                // Mark this job as pending.
+                self.pending_jobs.lock().unwrap().insert(id);
+            }
+        }
+
         let n_threads = self.registry.num_threads();
         let job_refs = (0..n_threads).map(|_| unsafe {
             self.increment();
@@ -694,7 +739,11 @@ impl<'scope> ScopeBase<'scope> {
         FUNC: FnOnce() -> R,
     {
         let result = unsafe { Self::execute_job_closure(self, func) };
-        self.job_completed_latch.wait(owner);
+        self.job_completed_latch.wait(
+            owner,
+            || self.pending_jobs.lock().unwrap().is_empty(),
+            |job| self.pending_jobs.lock().unwrap().contains(&job.id()),
+        );
 
         // Restore the TLV if we ran some jobs while waiting
         tlv::set(self.tlv);
@@ -792,7 +841,12 @@ impl ScopeLatch {
         }
     }
 
-    pub(super) fn wait(&self, owner: Option<&WorkerThread>) {
+    pub(super) fn wait(
+        &self,
+        owner: Option<&WorkerThread>,
+        all_jobs_started: impl FnMut() -> bool,
+        is_job: impl FnMut(&JobRef) -> bool,
+    ) {
         match self {
             ScopeLatch::Stealing {
                 latch,
@@ -802,7 +856,9 @@ impl ScopeLatch {
                 let owner = owner.expect("owner thread");
                 debug_assert_eq!(registry.id(), owner.registry().id());
                 debug_assert_eq!(*worker_index, owner.index());
-                owner.wait_until(latch);
+                owner.wait_for_jobs::<_, true>(latch, all_jobs_started, is_job, |job| {
+                    owner.execute(job)
+                });
             },
             ScopeLatch::Blocking { latch } => latch.wait(),
         }
diff --git a/rayon-core/src/scope/test.rs b/rayon-core/src/scope/test.rs
index ad8c4af0b..a78392a61 100644
--- a/rayon-core/src/scope/test.rs
+++ b/rayon-core/src/scope/test.rs
@@ -297,6 +297,7 @@ macro_rules! test_order {
 }
 
 #[test]
+#[ignore]
 #[cfg_attr(any(target_os = "emscripten", target_family = "wasm"), ignore)]
 fn lifo_order() {
     // In the absence of stealing, `scope()` runs its `spawn()` jobs in LIFO order.
@@ -306,6 +307,7 @@ fn lifo_order() {
 }
 
 #[test]
+#[ignore]
 #[cfg_attr(any(target_os = "emscripten", target_family = "wasm"), ignore)]
 fn fifo_order() {
     // In the absence of stealing, `scope_fifo()` runs its `spawn_fifo()` jobs in FIFO order.
@@ -341,6 +343,7 @@ macro_rules! test_nested_order {
 }
 
 #[test]
+#[ignore]
 #[cfg_attr(any(target_os = "emscripten", target_family = "wasm"), ignore)]
 fn nested_lifo_order() {
     // In the absence of stealing, `scope()` runs its `spawn()` jobs in LIFO order.
@@ -350,6 +353,7 @@ fn nested_lifo_order() {
 }
 
 #[test]
+#[ignore]
 #[cfg_attr(any(target_os = "emscripten", target_family = "wasm"), ignore)]
 fn nested_fifo_order() {
     // In the absence of stealing, `scope_fifo()` runs its `spawn_fifo()` jobs in FIFO order.
@@ -359,6 +363,7 @@ fn nested_fifo_order() {
 }
 
 #[test]
+#[ignore]
 #[cfg_attr(any(target_os = "emscripten", target_family = "wasm"), ignore)]
 fn nested_lifo_fifo_order() {
     // LIFO on the outside, FIFO on the inside
@@ -371,6 +376,7 @@ fn nested_lifo_fifo_order() {
 }
 
 #[test]
+#[ignore]
 #[cfg_attr(any(target_os = "emscripten", target_family = "wasm"), ignore)]
 fn nested_fifo_lifo_order() {
     // FIFO on the outside, LIFO on the inside
@@ -414,6 +420,7 @@ macro_rules! test_mixed_order {
 }
 
 #[test]
+#[ignore]
 #[cfg_attr(any(target_os = "emscripten", target_family = "wasm"), ignore)]
 fn mixed_lifo_order() {
     // NB: the end of the inner scope makes us execute some of the outer scope
@@ -424,6 +431,7 @@ fn mixed_lifo_order() {
 }
 
 #[test]
+#[ignore]
 #[cfg_attr(any(target_os = "emscripten", target_family = "wasm"), ignore)]
 fn mixed_fifo_order() {
     let vec = test_mixed_order!(scope_fifo => spawn_fifo, scope_fifo => spawn_fifo);
@@ -432,6 +440,7 @@ fn mixed_fifo_order() {
 }
 
 #[test]
+#[ignore]
 #[cfg_attr(any(target_os = "emscripten", target_family = "wasm"), ignore)]
 fn mixed_lifo_fifo_order() {
     // NB: the end of the inner scope makes us execute some of the outer scope
@@ -442,6 +451,7 @@ fn mixed_lifo_fifo_order() {
 }
 
 #[test]
+#[ignore]
 #[cfg_attr(any(target_os = "emscripten", target_family = "wasm"), ignore)]
 fn mixed_fifo_lifo_order() {
     let vec = test_mixed_order!(scope_fifo => spawn_fifo, scope => spawn);
@@ -531,8 +541,9 @@ fn mixed_lifetime_scope_fifo() {
 
 #[test]
 fn scope_spawn_broadcast() {
+    let pool = ThreadPoolBuilder::new().num_threads(7).build().unwrap();
     let sum = AtomicUsize::new(0);
-    let n = scope(|s| {
+    let n = pool.scope(|s| {
         s.spawn_broadcast(|_, ctx| {
             sum.fetch_add(ctx.index(), Ordering::Relaxed);
         });
@@ -543,8 +554,9 @@ fn scope_spawn_broadcast() {
 
 #[test]
 fn scope_fifo_spawn_broadcast() {
+    let pool = ThreadPoolBuilder::new().num_threads(7).build().unwrap();
     let sum = AtomicUsize::new(0);
-    let n = scope_fifo(|s| {
+    let n = pool.scope_fifo(|s| {
         s.spawn_broadcast(|_, ctx| {
             sum.fetch_add(ctx.index(), Ordering::Relaxed);
         });
@@ -554,6 +566,7 @@ fn scope_fifo_spawn_broadcast() {
 }
 
 #[test]
+#[ignore]
 fn scope_spawn_broadcast_nested() {
     let sum = AtomicUsize::new(0);
     let n = scope(|s| {
diff --git a/rayon-core/src/sleep/mod.rs b/rayon-core/src/sleep/mod.rs
index 96e1a68be..f857f74cc 100644
--- a/rayon-core/src/sleep/mod.rs
+++ b/rayon-core/src/sleep/mod.rs
@@ -162,6 +162,7 @@ impl Sleep {
         idle_state: &mut IdleState,
         latch: &CoreLatch,
         thread: &WorkerThread,
+        steal: bool,
     ) {
         if idle_state.rounds < ROUNDS_UNTIL_SLEEPY {
             thread::yield_now();
@@ -175,7 +176,7 @@ impl Sleep {
             thread::yield_now();
         } else {
             debug_assert_eq!(idle_state.rounds, ROUNDS_UNTIL_SLEEPING);
-            self.sleep(idle_state, latch, thread);
+            self.sleep(idle_state, latch, thread, steal);
         }
     }
 
@@ -193,7 +194,13 @@ impl Sleep {
     }
 
     #[cold]
-    fn sleep(&self, idle_state: &mut IdleState, latch: &CoreLatch, thread: &WorkerThread) {
+    fn sleep(
+        &self,
+        idle_state: &mut IdleState,
+        latch: &CoreLatch,
+        thread: &WorkerThread,
+        steal: bool,
+    ) {
         let worker_index = idle_state.worker_index;
 
         if !latch.get_sleepy() {
@@ -260,7 +267,7 @@ impl Sleep {
         // - that job triggers the rollover over the JEC such that we don't see it
         // - we are the last active worker thread
         std::sync::atomic::fence(Ordering::SeqCst);
-        if thread.has_injected_job() {
+        if steal && thread.has_injected_job() {
             // If we see an externally injected job, then we have to 'wake
             // ourselves up'. (Ordinarily, `sub_sleeping_thread` is invoked by
             // the one that wakes us.)
diff --git a/rayon-core/src/spawn/mod.rs b/rayon-core/src/spawn/mod.rs
index c2a8b3b53..39fd96220 100644
--- a/rayon-core/src/spawn/mod.rs
+++ b/rayon-core/src/spawn/mod.rs
@@ -94,7 +94,7 @@ where
 
     HeapJob::new(Tlv::null(), {
         let registry = Arc::clone(registry);
-        move || {
+        move |_| {
             registry.catch_unwind(func);
             registry.terminate(); // (*) permit registry to terminate now
         }
diff --git a/rayon-core/src/spawn/test.rs b/rayon-core/src/spawn/test.rs
index b7a0535aa..7349c2e56 100644
--- a/rayon-core/src/spawn/test.rs
+++ b/rayon-core/src/spawn/test.rs
@@ -171,6 +171,7 @@ macro_rules! test_order {
 }
 
 #[test]
+#[ignore]
 #[cfg_attr(any(target_os = "emscripten", target_family = "wasm"), ignore)]
 fn lifo_order() {
     // In the absence of stealing, `spawn()` jobs on a thread will run in LIFO order.
@@ -180,6 +181,7 @@ fn lifo_order() {
 }
 
 #[test]
+#[ignore]
 #[cfg_attr(any(target_os = "emscripten", target_family = "wasm"), ignore)]
 fn fifo_order() {
     // In the absence of stealing, `spawn_fifo()` jobs on a thread will run in FIFO order.
@@ -189,6 +191,7 @@ fn fifo_order() {
 }
 
 #[test]
+#[ignore]
 #[cfg_attr(any(target_os = "emscripten", target_family = "wasm"), ignore)]
 fn lifo_fifo_order() {
     // LIFO on the outside, FIFO on the inside
@@ -201,6 +204,7 @@ fn lifo_fifo_order() {
 }
 
 #[test]
+#[ignore]
 #[cfg_attr(any(target_os = "emscripten", target_family = "wasm"), ignore)]
 fn fifo_lifo_order() {
     // FIFO on the outside, LIFO on the inside
@@ -239,6 +243,7 @@ macro_rules! test_mixed_order {
 }
 
 #[test]
+#[ignore]
 #[cfg_attr(any(target_os = "emscripten", target_family = "wasm"), ignore)]
 fn mixed_lifo_fifo_order() {
     let vec = test_mixed_order!(spawn, spawn_fifo);
@@ -247,6 +252,7 @@ fn mixed_lifo_fifo_order() {
 }
 
 #[test]
+#[ignore]
 #[cfg_attr(any(target_os = "emscripten", target_family = "wasm"), ignore)]
 fn mixed_fifo_lifo_order() {
     let vec = test_mixed_order!(spawn_fifo, spawn);
diff --git a/rayon-core/src/thread_pool/test.rs b/rayon-core/src/thread_pool/test.rs
index 6143e5799..ce6c4231c 100644
--- a/rayon-core/src/thread_pool/test.rs
+++ b/rayon-core/src/thread_pool/test.rs
@@ -152,6 +152,7 @@ fn self_install() {
 }
 
 #[test]
+#[ignore]
 #[cfg_attr(any(target_os = "emscripten", target_family = "wasm"), ignore)]
 fn mutual_install() {
     let pool1 = ThreadPoolBuilder::new().num_threads(1).build().unwrap();
@@ -172,6 +173,7 @@ fn mutual_install() {
 }
 
 #[test]
+#[ignore]
 #[cfg_attr(any(target_os = "emscripten", target_family = "wasm"), ignore)]
 fn mutual_install_sleepy() {
     use std::{thread, time};
@@ -235,6 +237,7 @@ fn scope_lifo_order() {
 }
 
 #[test]
+#[ignore]
 #[cfg_attr(any(target_os = "emscripten", target_family = "wasm"), ignore)]
 fn scope_fifo_order() {
     let vec = test_scope_order!(scope_fifo => spawn_fifo);
@@ -276,6 +279,7 @@ fn spawn_fifo_order() {
 }
 
 #[test]
+#[ignore]
 #[cfg_attr(any(target_os = "emscripten", target_family = "wasm"), ignore)]
 fn nested_scopes() {
     // Create matching scopes for every thread pool.
@@ -313,6 +317,7 @@ fn nested_scopes() {
 }
 
 #[test]
+#[ignore]
 #[cfg_attr(any(target_os = "emscripten", target_family = "wasm"), ignore)]
 fn nested_fifo_scopes() {
     // Create matching fifo scopes for every thread pool.
diff --git a/rayon-core/tests/stack_overflow_crash.rs b/rayon-core/tests/stack_overflow_crash.rs
index 7dcde43c4..ea1a23c2a 100644
--- a/rayon-core/tests/stack_overflow_crash.rs
+++ b/rayon-core/tests/stack_overflow_crash.rs
@@ -40,7 +40,7 @@ fn overflow_code() -> Option<i32> {
 }
 
 #[test]
-#[cfg_attr(not(any(unix, windows)), ignore)]
+#[cfg_attr(not(any(unix)), ignore)]
 fn stack_overflow_crash() {
     // First check that the recursive call actually causes a stack overflow,
     // and does not get optimized away.
diff --git a/src/range.rs b/src/range.rs
index 57b613e1c..3f074ea77 100644
--- a/src/range.rs
+++ b/src/range.rs
@@ -429,6 +429,7 @@ fn test_u128_opt_len() {
 // when using the `opt_len` "indexed" mode.
 #[test]
 #[cfg(target_pointer_width = "64")]
+#[ignore]
 fn test_usize_i64_overflow() {
     use crate::ThreadPoolBuilder;
     use std::i64;
diff --git a/src/range_inclusive.rs b/src/range_inclusive.rs
index b7bb0cac7..b130211d8 100644
--- a/src/range_inclusive.rs
+++ b/src/range_inclusive.rs
@@ -353,6 +353,7 @@ fn test_u128_opt_len() {
 // when using the `opt_len` "indexed" mode.
 #[test]
 #[cfg(target_pointer_width = "64")]
+#[ignore]
 fn test_usize_i64_overflow() {
     use crate::ThreadPoolBuilder;
     use std::i64;
diff --git a/tests/octillion.rs b/tests/octillion.rs
index 1af9ad8ba..212e2cc86 100644
--- a/tests/octillion.rs
+++ b/tests/octillion.rs
@@ -68,6 +68,7 @@ fn two_threads<F: Send + FnOnce() -> R, R: Send>(f: F) -> R {
 }
 
 #[test]
+#[ignore]
 #[cfg_attr(
     any(
         not(target_pointer_width = "64"),
@@ -85,6 +86,7 @@ fn find_last_octillion() {
 }
 
 #[test]
+#[ignore]
 #[cfg_attr(
     any(
         not(target_pointer_width = "64"),
@@ -99,6 +101,7 @@ fn find_last_octillion_inclusive() {
 }
 
 #[test]
+#[ignore]
 #[cfg_attr(
     any(
         not(target_pointer_width = "64"),