From 35c5144394c1b93784867d330f694fa7c8f480e3 Mon Sep 17 00:00:00 2001 From: "Celina G. Val" Date: Wed, 11 Jun 2025 10:43:59 -0700 Subject: Move rayon-core to rustc_thread_pool files as is This commit literally copied the directory rayon-core from revision `5fadf44`. Link: https://github.com/rust-lang/rustc-rayon/tree/5fadf44/rayon-core --- compiler/rustc_thread_pool/src/join/mod.rs | 202 +++++++++++++++++++++++++++++ 1 file changed, 202 insertions(+) create mode 100644 compiler/rustc_thread_pool/src/join/mod.rs (limited to 'compiler/rustc_thread_pool/src/join/mod.rs') diff --git a/compiler/rustc_thread_pool/src/join/mod.rs b/compiler/rustc_thread_pool/src/join/mod.rs new file mode 100644 index 00000000000..032eec9c4c8 --- /dev/null +++ b/compiler/rustc_thread_pool/src/join/mod.rs @@ -0,0 +1,202 @@ +use crate::job::StackJob; +use crate::latch::SpinLatch; +use crate::registry::{self, WorkerThread}; +use crate::tlv::{self, Tlv}; +use crate::unwind; +use std::any::Any; + +use crate::FnContext; + +#[cfg(test)] +mod test; + +/// Takes two closures and *potentially* runs them in parallel. It +/// returns a pair of the results from those closures. +/// +/// Conceptually, calling `join()` is similar to spawning two threads, +/// one executing each of the two closures. However, the +/// implementation is quite different and incurs very low +/// overhead. The underlying technique is called "work stealing": the +/// Rayon runtime uses a fixed pool of worker threads and attempts to +/// only execute code in parallel when there are idle CPUs to handle +/// it. +/// +/// When `join` is called from outside the thread pool, the calling +/// thread will block while the closures execute in the pool. When +/// `join` is called within the pool, the calling thread still actively +/// participates in the thread pool. It will begin by executing closure +/// A (on the current thread). While it is doing that, it will advertise +/// closure B as being available for other threads to execute. Once closure A +/// has completed, the current thread will try to execute closure B; +/// if however closure B has been stolen, then it will look for other work +/// while waiting for the thief to fully execute closure B. (This is the +/// typical work-stealing strategy). +/// +/// # Examples +/// +/// This example uses join to perform a quick-sort (note this is not a +/// particularly optimized implementation: if you **actually** want to +/// sort for real, you should prefer [the `par_sort` method] offered +/// by Rayon). +/// +/// [the `par_sort` method]: ../rayon/slice/trait.ParallelSliceMut.html#method.par_sort +/// +/// ```rust +/// # use rayon_core as rayon; +/// let mut v = vec![5, 1, 8, 22, 0, 44]; +/// quick_sort(&mut v); +/// assert_eq!(v, vec![0, 1, 5, 8, 22, 44]); +/// +/// fn quick_sort(v: &mut [T]) { +/// if v.len() > 1 { +/// let mid = partition(v); +/// let (lo, hi) = v.split_at_mut(mid); +/// rayon::join(|| quick_sort(lo), +/// || quick_sort(hi)); +/// } +/// } +/// +/// // Partition rearranges all items `<=` to the pivot +/// // item (arbitrary selected to be the last item in the slice) +/// // to the first half of the slice. It then returns the +/// // "dividing point" where the pivot is placed. +/// fn partition(v: &mut [T]) -> usize { +/// let pivot = v.len() - 1; +/// let mut i = 0; +/// for j in 0..pivot { +/// if v[j] <= v[pivot] { +/// v.swap(i, j); +/// i += 1; +/// } +/// } +/// v.swap(i, pivot); +/// i +/// } +/// ``` +/// +/// # Warning about blocking I/O +/// +/// The assumption is that the closures given to `join()` are +/// CPU-bound tasks that do not perform I/O or other blocking +/// operations. If you do perform I/O, and that I/O should block +/// (e.g., waiting for a network request), the overall performance may +/// be poor. Moreover, if you cause one closure to be blocked waiting +/// on another (for example, using a channel), that could lead to a +/// deadlock. +/// +/// # Panics +/// +/// No matter what happens, both closures will always be executed. If +/// a single closure panics, whether it be the first or second +/// closure, that panic will be propagated and hence `join()` will +/// panic with the same panic value. If both closures panic, `join()` +/// will panic with the panic value from the first closure. +pub fn join(oper_a: A, oper_b: B) -> (RA, RB) +where + A: FnOnce() -> RA + Send, + B: FnOnce() -> RB + Send, + RA: Send, + RB: Send, +{ + #[inline] + fn call(f: impl FnOnce() -> R) -> impl FnOnce(FnContext) -> R { + move |_| f() + } + + join_context(call(oper_a), call(oper_b)) +} + +/// Identical to `join`, except that the closures have a parameter +/// that provides context for the way the closure has been called, +/// especially indicating whether they're executing on a different +/// thread than where `join_context` was called. This will occur if +/// the second job is stolen by a different thread, or if +/// `join_context` was called from outside the thread pool to begin +/// with. +pub fn join_context(oper_a: A, oper_b: B) -> (RA, RB) +where + A: FnOnce(FnContext) -> RA + Send, + B: FnOnce(FnContext) -> RB + Send, + RA: Send, + RB: Send, +{ + #[inline] + fn call_a(f: impl FnOnce(FnContext) -> R, injected: bool) -> impl FnOnce() -> R { + move || f(FnContext::new(injected)) + } + + #[inline] + fn call_b(f: impl FnOnce(FnContext) -> R) -> impl FnOnce(bool) -> R { + move |migrated| f(FnContext::new(migrated)) + } + + registry::in_worker(|worker_thread, injected| unsafe { + let tlv = tlv::get(); + // 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_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; + } + } + + // Restore the TLV since we might have run some jobs overwriting it when waiting for job b. + tlv::set(tlv); + + (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, + 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) +} -- cgit 1.4.1-3-g733a5 From 0b9b1df0064396708a5e5ca27fd010ae3ad3a305 Mon Sep 17 00:00:00 2001 From: "Celina G. Val" Date: Wed, 11 Jun 2025 11:12:32 -0700 Subject: Fix format and tidy for code moved from rayon --- compiler/rustc_thread_pool/src/broadcast/mod.rs | 17 +- compiler/rustc_thread_pool/src/broadcast/test.rs | 263 --------- compiler/rustc_thread_pool/src/broadcast/tests.rs | 264 +++++++++ compiler/rustc_thread_pool/src/job.rs | 26 +- compiler/rustc_thread_pool/src/join/mod.rs | 17 +- compiler/rustc_thread_pool/src/join/test.rs | 150 ----- compiler/rustc_thread_pool/src/join/tests.rs | 151 +++++ compiler/rustc_thread_pool/src/latch.rs | 54 +- compiler/rustc_thread_pool/src/lib.rs | 58 +- compiler/rustc_thread_pool/src/private.rs | 2 +- compiler/rustc_thread_pool/src/registry.rs | 99 ++-- compiler/rustc_thread_pool/src/scope/mod.rs | 34 +- compiler/rustc_thread_pool/src/scope/test.rs | 622 --------------------- compiler/rustc_thread_pool/src/scope/tests.rs | 607 ++++++++++++++++++++ compiler/rustc_thread_pool/src/sleep/README.md | 2 +- compiler/rustc_thread_pool/src/sleep/counters.rs | 8 +- compiler/rustc_thread_pool/src/sleep/mod.rs | 26 +- compiler/rustc_thread_pool/src/spawn/mod.rs | 17 +- compiler/rustc_thread_pool/src/spawn/test.rs | 255 --------- compiler/rustc_thread_pool/src/spawn/tests.rs | 246 ++++++++ compiler/rustc_thread_pool/src/test.rs | 200 ------- compiler/rustc_thread_pool/src/tests.rs | 197 +++++++ compiler/rustc_thread_pool/src/thread_pool/mod.rs | 15 +- compiler/rustc_thread_pool/src/thread_pool/test.rs | 418 -------------- .../rustc_thread_pool/src/thread_pool/tests.rs | 416 ++++++++++++++ compiler/rustc_thread_pool/src/tlv.rs | 3 +- compiler/rustc_thread_pool/src/worker_local.rs | 11 +- .../rustc_thread_pool/tests/double_init_fail.rs | 8 +- .../rustc_thread_pool/tests/init_zero_threads.rs | 5 +- .../rustc_thread_pool/tests/scoped_threadpool.rs | 4 +- .../tests/stack_overflow_crash.rs | 25 +- 31 files changed, 2041 insertions(+), 2179 deletions(-) delete mode 100644 compiler/rustc_thread_pool/src/broadcast/test.rs create mode 100644 compiler/rustc_thread_pool/src/broadcast/tests.rs delete mode 100644 compiler/rustc_thread_pool/src/join/test.rs create mode 100644 compiler/rustc_thread_pool/src/join/tests.rs delete mode 100644 compiler/rustc_thread_pool/src/scope/test.rs create mode 100644 compiler/rustc_thread_pool/src/scope/tests.rs delete mode 100644 compiler/rustc_thread_pool/src/spawn/test.rs create mode 100644 compiler/rustc_thread_pool/src/spawn/tests.rs delete mode 100644 compiler/rustc_thread_pool/src/test.rs create mode 100644 compiler/rustc_thread_pool/src/tests.rs delete mode 100644 compiler/rustc_thread_pool/src/thread_pool/test.rs create mode 100644 compiler/rustc_thread_pool/src/thread_pool/tests.rs (limited to 'compiler/rustc_thread_pool/src/join/mod.rs') diff --git a/compiler/rustc_thread_pool/src/broadcast/mod.rs b/compiler/rustc_thread_pool/src/broadcast/mod.rs index 442891f2d28..c2b0d47f829 100644 --- a/compiler/rustc_thread_pool/src/broadcast/mod.rs +++ b/compiler/rustc_thread_pool/src/broadcast/mod.rs @@ -1,10 +1,11 @@ -use crate::job::{ArcJob, StackJob}; -use crate::latch::{CountLatch, LatchRef}; -use crate::registry::{Registry, WorkerThread}; use std::fmt; use std::marker::PhantomData; use std::sync::Arc; +use crate::job::{ArcJob, StackJob}; +use crate::latch::{CountLatch, LatchRef}; +use crate::registry::{Registry, WorkerThread}; + mod test; /// Executes `op` within every thread in the current threadpool. If this is @@ -53,10 +54,7 @@ impl<'a> BroadcastContext<'a> { pub(super) fn with(f: impl FnOnce(BroadcastContext<'_>) -> R) -> R { let worker_thread = WorkerThread::current(); assert!(!worker_thread.is_null()); - f(BroadcastContext { - worker: unsafe { &*worker_thread }, - _marker: PhantomData, - }) + f(BroadcastContext { worker: unsafe { &*worker_thread }, _marker: PhantomData }) } /// Our index amongst the broadcast threads (ranges from `0..self.num_threads()`). @@ -108,9 +106,8 @@ where let current_thread = WorkerThread::current().as_ref(); let tlv = crate::tlv::get(); let latch = CountLatch::with_count(n_threads, current_thread); - let jobs: Vec<_> = (0..n_threads) - .map(|_| StackJob::new(tlv, &f, LatchRef::new(&latch))) - .collect(); + let jobs: Vec<_> = + (0..n_threads).map(|_| StackJob::new(tlv, &f, LatchRef::new(&latch))).collect(); let job_refs = jobs.iter().map(|job| job.as_job_ref()); registry.inject_broadcast(job_refs); diff --git a/compiler/rustc_thread_pool/src/broadcast/test.rs b/compiler/rustc_thread_pool/src/broadcast/test.rs deleted file mode 100644 index 00ab4ad7fe4..00000000000 --- a/compiler/rustc_thread_pool/src/broadcast/test.rs +++ /dev/null @@ -1,263 +0,0 @@ -#![cfg(test)] - -use crate::ThreadPoolBuilder; -use std::sync::atomic::{AtomicUsize, Ordering}; -use std::sync::mpsc::channel; -use std::sync::Arc; -use std::{thread, time}; - -#[test] -fn broadcast_global() { - let v = crate::broadcast(|ctx| ctx.index()); - assert!(v.into_iter().eq(0..crate::current_num_threads())); -} - -#[test] -#[cfg_attr(any(target_os = "emscripten", target_family = "wasm"), ignore)] -fn spawn_broadcast_global() { - let (tx, rx) = channel(); - crate::spawn_broadcast(move |ctx| tx.send(ctx.index()).unwrap()); - - let mut v: Vec<_> = rx.into_iter().collect(); - v.sort_unstable(); - assert!(v.into_iter().eq(0..crate::current_num_threads())); -} - -#[test] -#[cfg_attr(any(target_os = "emscripten", target_family = "wasm"), ignore)] -fn broadcast_pool() { - let pool = ThreadPoolBuilder::new().num_threads(7).build().unwrap(); - let v = pool.broadcast(|ctx| ctx.index()); - assert!(v.into_iter().eq(0..7)); -} - -#[test] -#[cfg_attr(any(target_os = "emscripten", target_family = "wasm"), ignore)] -fn spawn_broadcast_pool() { - let (tx, rx) = channel(); - let pool = ThreadPoolBuilder::new().num_threads(7).build().unwrap(); - pool.spawn_broadcast(move |ctx| tx.send(ctx.index()).unwrap()); - - let mut v: Vec<_> = rx.into_iter().collect(); - v.sort_unstable(); - assert!(v.into_iter().eq(0..7)); -} - -#[test] -#[cfg_attr(any(target_os = "emscripten", target_family = "wasm"), ignore)] -fn broadcast_self() { - let pool = ThreadPoolBuilder::new().num_threads(7).build().unwrap(); - let v = pool.install(|| crate::broadcast(|ctx| ctx.index())); - assert!(v.into_iter().eq(0..7)); -} - -#[test] -#[cfg_attr(any(target_os = "emscripten", target_family = "wasm"), ignore)] -fn spawn_broadcast_self() { - let (tx, rx) = channel(); - let pool = ThreadPoolBuilder::new().num_threads(7).build().unwrap(); - pool.spawn(|| crate::spawn_broadcast(move |ctx| tx.send(ctx.index()).unwrap())); - - let mut v: Vec<_> = rx.into_iter().collect(); - v.sort_unstable(); - assert!(v.into_iter().eq(0..7)); -} - -#[test] -#[cfg_attr(any(target_os = "emscripten", target_family = "wasm"), ignore)] -fn broadcast_mutual() { - let count = AtomicUsize::new(0); - let pool1 = ThreadPoolBuilder::new().num_threads(3).build().unwrap(); - let pool2 = ThreadPoolBuilder::new().num_threads(7).build().unwrap(); - pool1.install(|| { - pool2.broadcast(|_| { - pool1.broadcast(|_| { - count.fetch_add(1, Ordering::Relaxed); - }) - }) - }); - assert_eq!(count.into_inner(), 3 * 7); -} - -#[test] -#[cfg_attr(any(target_os = "emscripten", target_family = "wasm"), ignore)] -fn spawn_broadcast_mutual() { - let (tx, rx) = channel(); - let pool1 = Arc::new(ThreadPoolBuilder::new().num_threads(3).build().unwrap()); - let pool2 = ThreadPoolBuilder::new().num_threads(7).build().unwrap(); - pool1.spawn({ - let pool1 = Arc::clone(&pool1); - move || { - pool2.spawn_broadcast(move |_| { - let tx = tx.clone(); - pool1.spawn_broadcast(move |_| tx.send(()).unwrap()) - }) - } - }); - assert_eq!(rx.into_iter().count(), 3 * 7); -} - -#[test] -#[cfg_attr(any(target_os = "emscripten", target_family = "wasm"), ignore)] -fn broadcast_mutual_sleepy() { - let count = AtomicUsize::new(0); - let pool1 = ThreadPoolBuilder::new().num_threads(3).build().unwrap(); - let pool2 = ThreadPoolBuilder::new().num_threads(7).build().unwrap(); - pool1.install(|| { - thread::sleep(time::Duration::from_secs(1)); - pool2.broadcast(|_| { - thread::sleep(time::Duration::from_secs(1)); - pool1.broadcast(|_| { - thread::sleep(time::Duration::from_millis(100)); - count.fetch_add(1, Ordering::Relaxed); - }) - }) - }); - assert_eq!(count.into_inner(), 3 * 7); -} - -#[test] -#[cfg_attr(any(target_os = "emscripten", target_family = "wasm"), ignore)] -fn spawn_broadcast_mutual_sleepy() { - let (tx, rx) = channel(); - let pool1 = Arc::new(ThreadPoolBuilder::new().num_threads(3).build().unwrap()); - let pool2 = ThreadPoolBuilder::new().num_threads(7).build().unwrap(); - pool1.spawn({ - let pool1 = Arc::clone(&pool1); - move || { - thread::sleep(time::Duration::from_secs(1)); - pool2.spawn_broadcast(move |_| { - let tx = tx.clone(); - thread::sleep(time::Duration::from_secs(1)); - pool1.spawn_broadcast(move |_| { - thread::sleep(time::Duration::from_millis(100)); - tx.send(()).unwrap(); - }) - }) - } - }); - assert_eq!(rx.into_iter().count(), 3 * 7); -} - -#[test] -#[cfg_attr(not(panic = "unwind"), ignore)] -fn broadcast_panic_one() { - let count = AtomicUsize::new(0); - let pool = ThreadPoolBuilder::new().num_threads(7).build().unwrap(); - let result = crate::unwind::halt_unwinding(|| { - pool.broadcast(|ctx| { - count.fetch_add(1, Ordering::Relaxed); - if ctx.index() == 3 { - panic!("Hello, world!"); - } - }) - }); - assert_eq!(count.into_inner(), 7); - assert!(result.is_err(), "broadcast panic should propagate!"); -} - -#[test] -#[cfg_attr(not(panic = "unwind"), ignore)] -fn spawn_broadcast_panic_one() { - let (tx, rx) = channel(); - let (panic_tx, panic_rx) = channel(); - let pool = ThreadPoolBuilder::new() - .num_threads(7) - .panic_handler(move |e| panic_tx.send(e).unwrap()) - .build() - .unwrap(); - pool.spawn_broadcast(move |ctx| { - tx.send(()).unwrap(); - if ctx.index() == 3 { - panic!("Hello, world!"); - } - }); - drop(pool); // including panic_tx - assert_eq!(rx.into_iter().count(), 7); - assert_eq!(panic_rx.into_iter().count(), 1); -} - -#[test] -#[cfg_attr(not(panic = "unwind"), ignore)] -fn broadcast_panic_many() { - let count = AtomicUsize::new(0); - let pool = ThreadPoolBuilder::new().num_threads(7).build().unwrap(); - let result = crate::unwind::halt_unwinding(|| { - pool.broadcast(|ctx| { - count.fetch_add(1, Ordering::Relaxed); - if ctx.index() % 2 == 0 { - panic!("Hello, world!"); - } - }) - }); - assert_eq!(count.into_inner(), 7); - assert!(result.is_err(), "broadcast panic should propagate!"); -} - -#[test] -#[cfg_attr(not(panic = "unwind"), ignore)] -fn spawn_broadcast_panic_many() { - let (tx, rx) = channel(); - let (panic_tx, panic_rx) = channel(); - let pool = ThreadPoolBuilder::new() - .num_threads(7) - .panic_handler(move |e| panic_tx.send(e).unwrap()) - .build() - .unwrap(); - pool.spawn_broadcast(move |ctx| { - tx.send(()).unwrap(); - if ctx.index() % 2 == 0 { - panic!("Hello, world!"); - } - }); - drop(pool); // including panic_tx - assert_eq!(rx.into_iter().count(), 7); - assert_eq!(panic_rx.into_iter().count(), 4); -} - -#[test] -#[cfg_attr(any(target_os = "emscripten", target_family = "wasm"), ignore)] -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)); - }); - } -} - -#[test] -fn broadcast_after_spawn_broadcast() { - let (tx, rx) = channel(); - - // Queue a non-blocking spawn_broadcast. - crate::spawn_broadcast(move |ctx| tx.send(ctx.index()).unwrap()); - - // This blocking broadcast runs after all prior broadcasts. - crate::broadcast(|_| {}); - - // The spawn_broadcast **must** have run by now on all threads. - let mut v: Vec<_> = rx.try_iter().collect(); - v.sort_unstable(); - assert!(v.into_iter().eq(0..crate::current_num_threads())); -} - -#[test] -fn broadcast_after_spawn() { - let (tx, rx) = channel(); - - // Queue a regular spawn on a thread-local deque. - crate::registry::in_worker(move |_, _| { - crate::spawn(move || tx.send(22).unwrap()); - }); - - // Broadcast runs after the local deque is empty. - crate::broadcast(|_| {}); - - // The spawn **must** have run by now. - assert_eq!(22, rx.try_recv().unwrap()); -} diff --git a/compiler/rustc_thread_pool/src/broadcast/tests.rs b/compiler/rustc_thread_pool/src/broadcast/tests.rs new file mode 100644 index 00000000000..fac8b8ad466 --- /dev/null +++ b/compiler/rustc_thread_pool/src/broadcast/tests.rs @@ -0,0 +1,264 @@ +#![cfg(test)] + +use std::sync::Arc; +use std::sync::atomic::{AtomicUsize, Ordering}; +use std::sync::mpsc::channel; +use std::{thread, time}; + +use crate::ThreadPoolBuilder; + +#[test] +fn broadcast_global() { + let v = crate::broadcast(|ctx| ctx.index()); + assert!(v.into_iter().eq(0..crate::current_num_threads())); +} + +#[test] +#[cfg_attr(any(target_os = "emscripten", target_family = "wasm"), ignore)] +fn spawn_broadcast_global() { + let (tx, rx) = channel(); + crate::spawn_broadcast(move |ctx| tx.send(ctx.index()).unwrap()); + + let mut v: Vec<_> = rx.into_iter().collect(); + v.sort_unstable(); + assert!(v.into_iter().eq(0..crate::current_num_threads())); +} + +#[test] +#[cfg_attr(any(target_os = "emscripten", target_family = "wasm"), ignore)] +fn broadcast_pool() { + let pool = ThreadPoolBuilder::new().num_threads(7).build().unwrap(); + let v = pool.broadcast(|ctx| ctx.index()); + assert!(v.into_iter().eq(0..7)); +} + +#[test] +#[cfg_attr(any(target_os = "emscripten", target_family = "wasm"), ignore)] +fn spawn_broadcast_pool() { + let (tx, rx) = channel(); + let pool = ThreadPoolBuilder::new().num_threads(7).build().unwrap(); + pool.spawn_broadcast(move |ctx| tx.send(ctx.index()).unwrap()); + + let mut v: Vec<_> = rx.into_iter().collect(); + v.sort_unstable(); + assert!(v.into_iter().eq(0..7)); +} + +#[test] +#[cfg_attr(any(target_os = "emscripten", target_family = "wasm"), ignore)] +fn broadcast_self() { + let pool = ThreadPoolBuilder::new().num_threads(7).build().unwrap(); + let v = pool.install(|| crate::broadcast(|ctx| ctx.index())); + assert!(v.into_iter().eq(0..7)); +} + +#[test] +#[cfg_attr(any(target_os = "emscripten", target_family = "wasm"), ignore)] +fn spawn_broadcast_self() { + let (tx, rx) = channel(); + let pool = ThreadPoolBuilder::new().num_threads(7).build().unwrap(); + pool.spawn(|| crate::spawn_broadcast(move |ctx| tx.send(ctx.index()).unwrap())); + + let mut v: Vec<_> = rx.into_iter().collect(); + v.sort_unstable(); + assert!(v.into_iter().eq(0..7)); +} + +#[test] +#[cfg_attr(any(target_os = "emscripten", target_family = "wasm"), ignore)] +fn broadcast_mutual() { + let count = AtomicUsize::new(0); + let pool1 = ThreadPoolBuilder::new().num_threads(3).build().unwrap(); + let pool2 = ThreadPoolBuilder::new().num_threads(7).build().unwrap(); + pool1.install(|| { + pool2.broadcast(|_| { + pool1.broadcast(|_| { + count.fetch_add(1, Ordering::Relaxed); + }) + }) + }); + assert_eq!(count.into_inner(), 3 * 7); +} + +#[test] +#[cfg_attr(any(target_os = "emscripten", target_family = "wasm"), ignore)] +fn spawn_broadcast_mutual() { + let (tx, rx) = channel(); + let pool1 = Arc::new(ThreadPoolBuilder::new().num_threads(3).build().unwrap()); + let pool2 = ThreadPoolBuilder::new().num_threads(7).build().unwrap(); + pool1.spawn({ + let pool1 = Arc::clone(&pool1); + move || { + pool2.spawn_broadcast(move |_| { + let tx = tx.clone(); + pool1.spawn_broadcast(move |_| tx.send(()).unwrap()) + }) + } + }); + assert_eq!(rx.into_iter().count(), 3 * 7); +} + +#[test] +#[cfg_attr(any(target_os = "emscripten", target_family = "wasm"), ignore)] +fn broadcast_mutual_sleepy() { + let count = AtomicUsize::new(0); + let pool1 = ThreadPoolBuilder::new().num_threads(3).build().unwrap(); + let pool2 = ThreadPoolBuilder::new().num_threads(7).build().unwrap(); + pool1.install(|| { + thread::sleep(time::Duration::from_secs(1)); + pool2.broadcast(|_| { + thread::sleep(time::Duration::from_secs(1)); + pool1.broadcast(|_| { + thread::sleep(time::Duration::from_millis(100)); + count.fetch_add(1, Ordering::Relaxed); + }) + }) + }); + assert_eq!(count.into_inner(), 3 * 7); +} + +#[test] +#[cfg_attr(any(target_os = "emscripten", target_family = "wasm"), ignore)] +fn spawn_broadcast_mutual_sleepy() { + let (tx, rx) = channel(); + let pool1 = Arc::new(ThreadPoolBuilder::new().num_threads(3).build().unwrap()); + let pool2 = ThreadPoolBuilder::new().num_threads(7).build().unwrap(); + pool1.spawn({ + let pool1 = Arc::clone(&pool1); + move || { + thread::sleep(time::Duration::from_secs(1)); + pool2.spawn_broadcast(move |_| { + let tx = tx.clone(); + thread::sleep(time::Duration::from_secs(1)); + pool1.spawn_broadcast(move |_| { + thread::sleep(time::Duration::from_millis(100)); + tx.send(()).unwrap(); + }) + }) + } + }); + assert_eq!(rx.into_iter().count(), 3 * 7); +} + +#[test] +#[cfg_attr(not(panic = "unwind"), ignore)] +fn broadcast_panic_one() { + let count = AtomicUsize::new(0); + let pool = ThreadPoolBuilder::new().num_threads(7).build().unwrap(); + let result = crate::unwind::halt_unwinding(|| { + pool.broadcast(|ctx| { + count.fetch_add(1, Ordering::Relaxed); + if ctx.index() == 3 { + panic!("Hello, world!"); + } + }) + }); + assert_eq!(count.into_inner(), 7); + assert!(result.is_err(), "broadcast panic should propagate!"); +} + +#[test] +#[cfg_attr(not(panic = "unwind"), ignore)] +fn spawn_broadcast_panic_one() { + let (tx, rx) = channel(); + let (panic_tx, panic_rx) = channel(); + let pool = ThreadPoolBuilder::new() + .num_threads(7) + .panic_handler(move |e| panic_tx.send(e).unwrap()) + .build() + .unwrap(); + pool.spawn_broadcast(move |ctx| { + tx.send(()).unwrap(); + if ctx.index() == 3 { + panic!("Hello, world!"); + } + }); + drop(pool); // including panic_tx + assert_eq!(rx.into_iter().count(), 7); + assert_eq!(panic_rx.into_iter().count(), 1); +} + +#[test] +#[cfg_attr(not(panic = "unwind"), ignore)] +fn broadcast_panic_many() { + let count = AtomicUsize::new(0); + let pool = ThreadPoolBuilder::new().num_threads(7).build().unwrap(); + let result = crate::unwind::halt_unwinding(|| { + pool.broadcast(|ctx| { + count.fetch_add(1, Ordering::Relaxed); + if ctx.index() % 2 == 0 { + panic!("Hello, world!"); + } + }) + }); + assert_eq!(count.into_inner(), 7); + assert!(result.is_err(), "broadcast panic should propagate!"); +} + +#[test] +#[cfg_attr(not(panic = "unwind"), ignore)] +fn spawn_broadcast_panic_many() { + let (tx, rx) = channel(); + let (panic_tx, panic_rx) = channel(); + let pool = ThreadPoolBuilder::new() + .num_threads(7) + .panic_handler(move |e| panic_tx.send(e).unwrap()) + .build() + .unwrap(); + pool.spawn_broadcast(move |ctx| { + tx.send(()).unwrap(); + if ctx.index() % 2 == 0 { + panic!("Hello, world!"); + } + }); + drop(pool); // including panic_tx + assert_eq!(rx.into_iter().count(), 7); + assert_eq!(panic_rx.into_iter().count(), 4); +} + +#[test] +#[cfg_attr(any(target_os = "emscripten", target_family = "wasm"), ignore)] +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)); + }); + } +} + +#[test] +fn broadcast_after_spawn_broadcast() { + let (tx, rx) = channel(); + + // Queue a non-blocking spawn_broadcast. + crate::spawn_broadcast(move |ctx| tx.send(ctx.index()).unwrap()); + + // This blocking broadcast runs after all prior broadcasts. + crate::broadcast(|_| {}); + + // The spawn_broadcast **must** have run by now on all threads. + let mut v: Vec<_> = rx.try_iter().collect(); + v.sort_unstable(); + assert!(v.into_iter().eq(0..crate::current_num_threads())); +} + +#[test] +fn broadcast_after_spawn() { + let (tx, rx) = channel(); + + // Queue a regular spawn on a thread-local deque. + crate::registry::in_worker(move |_, _| { + crate::spawn(move || tx.send(22).unwrap()); + }); + + // Broadcast runs after the local deque is empty. + crate::broadcast(|_| {}); + + // The spawn **must** have run by now. + assert_eq!(22, rx.try_recv().unwrap()); +} diff --git a/compiler/rustc_thread_pool/src/job.rs b/compiler/rustc_thread_pool/src/job.rs index 394c7576b2c..3241914ba81 100644 --- a/compiler/rustc_thread_pool/src/job.rs +++ b/compiler/rustc_thread_pool/src/job.rs @@ -1,13 +1,14 @@ -use crate::latch::Latch; -use crate::tlv; -use crate::tlv::Tlv; -use crate::unwind; -use crossbeam_deque::{Injector, Steal}; use std::any::Any; use std::cell::UnsafeCell; use std::mem; use std::sync::Arc; +use crossbeam_deque::{Injector, Steal}; + +use crate::latch::Latch; +use crate::tlv::Tlv; +use crate::{tlv, unwind}; + pub(super) enum JobResult { None, Ok(T), @@ -29,7 +30,7 @@ pub(super) trait Job { /// Effectively a Job trait object. Each JobRef **must** be executed /// exactly once, or else data may leak. /// -/// Internally, we store the job's data in a `*const ()` pointer. The +/// Internally, we store the job's data in a `*const ()` pointer. The /// true type is something like `*const StackJob<...>`, but we hide /// it. We also carry the "execute fn" from the `Job` trait. pub(super) struct JobRef { @@ -48,10 +49,7 @@ impl JobRef { T: Job, { // erase types: - JobRef { - pointer: data as *const (), - execute_fn: ::execute, - } + JobRef { pointer: data as *const (), execute_fn: ::execute } } /// Returns an opaque handle that can be saved and compared, @@ -69,7 +67,7 @@ impl JobRef { /// A job that will be owned by a stack slot. This means that when it /// executes it need not free any heap data, the cleanup occurs when -/// the stack frame is later popped. The function parameter indicates +/// the stack frame is later popped. The function parameter indicates /// `true` if the job was stolen -- executed on a different thread. pub(super) struct StackJob where @@ -248,13 +246,11 @@ pub(super) struct JobFifo { impl JobFifo { pub(super) fn new() -> Self { - JobFifo { - inner: Injector::new(), - } + JobFifo { inner: Injector::new() } } pub(super) unsafe fn push(&self, job_ref: JobRef) -> JobRef { - // A little indirection ensures that spawns are always prioritized in FIFO order. The + // A little indirection ensures that spawns are always prioritized in FIFO order. The // jobs in a thread's deque may be popped from the back (LIFO) or stolen from the front // (FIFO), but either way they will end up popping from the front of this queue. self.inner.push(job_ref); diff --git a/compiler/rustc_thread_pool/src/join/mod.rs b/compiler/rustc_thread_pool/src/join/mod.rs index 032eec9c4c8..798a8347d79 100644 --- a/compiler/rustc_thread_pool/src/join/mod.rs +++ b/compiler/rustc_thread_pool/src/join/mod.rs @@ -1,11 +1,10 @@ +use std::any::Any; + use crate::job::StackJob; use crate::latch::SpinLatch; use crate::registry::{self, WorkerThread}; use crate::tlv::{self, Tlv}; -use crate::unwind; -use std::any::Any; - -use crate::FnContext; +use crate::{FnContext, unwind}; #[cfg(test)] mod test; @@ -22,7 +21,7 @@ mod test; /// it. /// /// When `join` is called from outside the thread pool, the calling -/// thread will block while the closures execute in the pool. When +/// thread will block while the closures execute in the pool. When /// `join` is called within the pool, the calling thread still actively /// participates in the thread pool. It will begin by executing closure /// A (on the current thread). While it is doing that, it will advertise @@ -80,13 +79,13 @@ mod test; /// CPU-bound tasks that do not perform I/O or other blocking /// operations. If you do perform I/O, and that I/O should block /// (e.g., waiting for a network request), the overall performance may -/// be poor. Moreover, if you cause one closure to be blocked waiting +/// be poor. Moreover, if you cause one closure to be blocked waiting /// on another (for example, using a channel), that could lead to a /// deadlock. /// /// # Panics /// -/// No matter what happens, both closures will always be executed. If +/// No matter what happens, both closures will always be executed. If /// a single closure panics, whether it be the first or second /// closure, that panic will be propagated and hence `join()` will /// panic with the same panic value. If both closures panic, `join()` @@ -109,7 +108,7 @@ where /// Identical to `join`, except that the closures have a parameter /// that provides context for the way the closure has been called, /// especially indicating whether they're executing on a different -/// thread than where `join_context` was called. This will occur if +/// thread than where `join_context` was called. This will occur if /// the second job is stolen by a different thread, or if /// `join_context` was called from outside the thread pool to begin /// with. @@ -148,7 +147,7 @@ where }; // 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 + // 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. diff --git a/compiler/rustc_thread_pool/src/join/test.rs b/compiler/rustc_thread_pool/src/join/test.rs deleted file mode 100644 index 03f4ab4478d..00000000000 --- a/compiler/rustc_thread_pool/src/join/test.rs +++ /dev/null @@ -1,150 +0,0 @@ -//! Tests for the join code. - -use super::*; -use crate::ThreadPoolBuilder; -use rand::distr::StandardUniform; -use rand::{Rng, SeedableRng}; -use rand_xorshift::XorShiftRng; - -fn quick_sort(v: &mut [T]) { - if v.len() <= 1 { - return; - } - - let mid = partition(v); - let (lo, hi) = v.split_at_mut(mid); - join(|| quick_sort(lo), || quick_sort(hi)); -} - -fn partition(v: &mut [T]) -> usize { - let pivot = v.len() - 1; - let mut i = 0; - for j in 0..pivot { - if v[j] <= v[pivot] { - v.swap(i, j); - i += 1; - } - } - v.swap(i, pivot); - i -} - -fn seeded_rng() -> XorShiftRng { - let mut seed = ::Seed::default(); - (0..).zip(seed.as_mut()).for_each(|(i, x)| *x = i); - XorShiftRng::from_seed(seed) -} - -#[test] -fn sort() { - let rng = seeded_rng(); - let mut data: Vec = rng.sample_iter(&StandardUniform).take(6 * 1024).collect(); - let mut sorted_data = data.clone(); - sorted_data.sort(); - quick_sort(&mut data); - assert_eq!(data, sorted_data); -} - -#[test] -#[cfg_attr(any(target_os = "emscripten", target_family = "wasm"), ignore)] -fn sort_in_pool() { - let rng = seeded_rng(); - let mut data: Vec = rng.sample_iter(&StandardUniform).take(12 * 1024).collect(); - - let pool = ThreadPoolBuilder::new().build().unwrap(); - let mut sorted_data = data.clone(); - sorted_data.sort(); - pool.install(|| quick_sort(&mut data)); - assert_eq!(data, sorted_data); -} - -#[test] -#[should_panic(expected = "Hello, world!")] -fn panic_propagate_a() { - join(|| panic!("Hello, world!"), || ()); -} - -#[test] -#[should_panic(expected = "Hello, world!")] -fn panic_propagate_b() { - join(|| (), || panic!("Hello, world!")); -} - -#[test] -#[should_panic(expected = "Hello, world!")] -fn panic_propagate_both() { - join(|| panic!("Hello, world!"), || panic!("Goodbye, world!")); -} - -#[test] -#[cfg_attr(not(panic = "unwind"), ignore)] -fn panic_b_still_executes() { - let mut x = false; - match unwind::halt_unwinding(|| join(|| panic!("Hello, world!"), || x = true)) { - Ok(_) => panic!("failed to propagate panic from closure A,"), - Err(_) => assert!(x, "closure b failed to execute"), - } -} - -#[test] -#[cfg_attr(any(target_os = "emscripten", target_family = "wasm"), ignore)] -fn join_context_both() { - // If we're not in a pool, both should be marked stolen as they're injected. - let (a_migrated, b_migrated) = join_context(|a| a.migrated(), |b| b.migrated()); - assert!(a_migrated); - assert!(b_migrated); -} - -#[test] -#[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. - let pool = ThreadPoolBuilder::new().num_threads(1).build().unwrap(); - let (a_migrated, b_migrated) = - pool.install(|| join_context(|a| a.migrated(), |b| b.migrated())); - assert!(!a_migrated); - assert!(!b_migrated); -} - -#[test] -#[cfg_attr(any(target_os = "emscripten", target_family = "wasm"), ignore)] -fn join_context_second() { - use std::sync::Barrier; - - // If we're already in a 2-thread pool, the second job should be stolen. - let barrier = Barrier::new(2); - let pool = ThreadPoolBuilder::new().num_threads(2).build().unwrap(); - let (a_migrated, b_migrated) = pool.install(|| { - join_context( - |a| { - barrier.wait(); - a.migrated() - }, - |b| { - barrier.wait(); - b.migrated() - }, - ) - }); - assert!(!a_migrated); - assert!(b_migrated); -} - -#[test] -#[cfg_attr(any(target_os = "emscripten", target_family = "wasm"), ignore)] -fn join_counter_overflow() { - const MAX: u32 = 500_000; - - let mut i = 0; - let mut j = 0; - let pool = ThreadPoolBuilder::new().num_threads(2).build().unwrap(); - - // Hammer on join a bunch of times -- used to hit overflow debug-assertions - // in JEC on 32-bit targets: https://github.com/rayon-rs/rayon/issues/797 - for _ in 0..MAX { - pool.join(|| i += 1, || j += 1); - } - - assert_eq!(i, MAX); - assert_eq!(j, MAX); -} diff --git a/compiler/rustc_thread_pool/src/join/tests.rs b/compiler/rustc_thread_pool/src/join/tests.rs new file mode 100644 index 00000000000..9df99072c3a --- /dev/null +++ b/compiler/rustc_thread_pool/src/join/tests.rs @@ -0,0 +1,151 @@ +//! Tests for the join code. + +use rand::distr::StandardUniform; +use rand::{Rng, SeedableRng}; +use rand_xorshift::XorShiftRng; + +use super::*; +use crate::ThreadPoolBuilder; + +fn quick_sort(v: &mut [T]) { + if v.len() <= 1 { + return; + } + + let mid = partition(v); + let (lo, hi) = v.split_at_mut(mid); + join(|| quick_sort(lo), || quick_sort(hi)); +} + +fn partition(v: &mut [T]) -> usize { + let pivot = v.len() - 1; + let mut i = 0; + for j in 0..pivot { + if v[j] <= v[pivot] { + v.swap(i, j); + i += 1; + } + } + v.swap(i, pivot); + i +} + +fn seeded_rng() -> XorShiftRng { + let mut seed = ::Seed::default(); + (0..).zip(seed.as_mut()).for_each(|(i, x)| *x = i); + XorShiftRng::from_seed(seed) +} + +#[test] +fn sort() { + let rng = seeded_rng(); + let mut data: Vec = rng.sample_iter(&StandardUniform).take(6 * 1024).collect(); + let mut sorted_data = data.clone(); + sorted_data.sort(); + quick_sort(&mut data); + assert_eq!(data, sorted_data); +} + +#[test] +#[cfg_attr(any(target_os = "emscripten", target_family = "wasm"), ignore)] +fn sort_in_pool() { + let rng = seeded_rng(); + let mut data: Vec = rng.sample_iter(&StandardUniform).take(12 * 1024).collect(); + + let pool = ThreadPoolBuilder::new().build().unwrap(); + let mut sorted_data = data.clone(); + sorted_data.sort(); + pool.install(|| quick_sort(&mut data)); + assert_eq!(data, sorted_data); +} + +#[test] +#[should_panic(expected = "Hello, world!")] +fn panic_propagate_a() { + join(|| panic!("Hello, world!"), || ()); +} + +#[test] +#[should_panic(expected = "Hello, world!")] +fn panic_propagate_b() { + join(|| (), || panic!("Hello, world!")); +} + +#[test] +#[should_panic(expected = "Hello, world!")] +fn panic_propagate_both() { + join(|| panic!("Hello, world!"), || panic!("Goodbye, world!")); +} + +#[test] +#[cfg_attr(not(panic = "unwind"), ignore)] +fn panic_b_still_executes() { + let mut x = false; + match unwind::halt_unwinding(|| join(|| panic!("Hello, world!"), || x = true)) { + Ok(_) => panic!("failed to propagate panic from closure A,"), + Err(_) => assert!(x, "closure b failed to execute"), + } +} + +#[test] +#[cfg_attr(any(target_os = "emscripten", target_family = "wasm"), ignore)] +fn join_context_both() { + // If we're not in a pool, both should be marked stolen as they're injected. + let (a_migrated, b_migrated) = join_context(|a| a.migrated(), |b| b.migrated()); + assert!(a_migrated); + assert!(b_migrated); +} + +#[test] +#[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. + let pool = ThreadPoolBuilder::new().num_threads(1).build().unwrap(); + let (a_migrated, b_migrated) = + pool.install(|| join_context(|a| a.migrated(), |b| b.migrated())); + assert!(!a_migrated); + assert!(!b_migrated); +} + +#[test] +#[cfg_attr(any(target_os = "emscripten", target_family = "wasm"), ignore)] +fn join_context_second() { + use std::sync::Barrier; + + // If we're already in a 2-thread pool, the second job should be stolen. + let barrier = Barrier::new(2); + let pool = ThreadPoolBuilder::new().num_threads(2).build().unwrap(); + let (a_migrated, b_migrated) = pool.install(|| { + join_context( + |a| { + barrier.wait(); + a.migrated() + }, + |b| { + barrier.wait(); + b.migrated() + }, + ) + }); + assert!(!a_migrated); + assert!(b_migrated); +} + +#[test] +#[cfg_attr(any(target_os = "emscripten", target_family = "wasm"), ignore)] +fn join_counter_overflow() { + const MAX: u32 = 500_000; + + let mut i = 0; + let mut j = 0; + let pool = ThreadPoolBuilder::new().num_threads(2).build().unwrap(); + + // Hammer on join a bunch of times -- used to hit overflow debug-assertions + // in JEC on 32-bit targets: https://github.com/rayon-rs/rayon/issues/797 + for _ in 0..MAX { + pool.join(|| i += 1, || j += 1); + } + + assert_eq!(i, MAX); + assert_eq!(j, MAX); +} diff --git a/compiler/rustc_thread_pool/src/latch.rs b/compiler/rustc_thread_pool/src/latch.rs index 8903942a8ce..f2f806e0184 100644 --- a/compiler/rustc_thread_pool/src/latch.rs +++ b/compiler/rustc_thread_pool/src/latch.rs @@ -28,7 +28,7 @@ use crate::registry::{Registry, WorkerThread}; /// - Once `probe()` returns true, all memory effects from the `set()` /// are visible (in other words, the set should synchronize-with /// the probe). -/// - Once `set()` occurs, the next `probe()` *will* observe it. This +/// - Once `set()` occurs, the next `probe()` *will* observe it. This /// typically requires a seq-cst ordering. See [the "tickle-then-get-sleepy" scenario in the sleep /// README](/src/sleep/README.md#tickle-then-get-sleepy) for details. pub(super) trait Latch { @@ -78,9 +78,7 @@ pub(super) struct CoreLatch { impl CoreLatch { #[inline] fn new() -> Self { - Self { - state: AtomicUsize::new(0), - } + Self { state: AtomicUsize::new(0) } } /// Invoked by owning thread as it prepares to sleep. Returns true @@ -88,9 +86,7 @@ impl CoreLatch { /// latch was set in the meantime. #[inline] pub(super) fn get_sleepy(&self) -> bool { - self.state - .compare_exchange(UNSET, SLEEPY, Ordering::SeqCst, Ordering::Relaxed) - .is_ok() + self.state.compare_exchange(UNSET, SLEEPY, Ordering::SeqCst, Ordering::Relaxed).is_ok() } /// Invoked by owning thread as it falls asleep sleep. Returns @@ -98,9 +94,7 @@ impl CoreLatch { /// was set in the meantime. #[inline] pub(super) fn fall_asleep(&self) -> bool { - self.state - .compare_exchange(SLEEPY, SLEEPING, Ordering::SeqCst, Ordering::Relaxed) - .is_ok() + self.state.compare_exchange(SLEEPY, SLEEPING, Ordering::SeqCst, Ordering::Relaxed).is_ok() } /// Invoked by owning thread as it falls asleep sleep. Returns @@ -110,8 +104,7 @@ impl CoreLatch { pub(super) fn wake_up(&self) { if !self.probe() { let _ = - self.state - .compare_exchange(SLEEPING, UNSET, Ordering::SeqCst, Ordering::Relaxed); + self.state.compare_exchange(SLEEPING, UNSET, Ordering::SeqCst, Ordering::Relaxed); } } @@ -166,15 +159,12 @@ impl<'r> SpinLatch<'r> { } } - /// Creates a new spin latch for cross-threadpool blocking. Notably, we + /// Creates a new spin latch for cross-threadpool blocking. Notably, we /// need to make sure the registry is kept alive after setting, so we can /// safely call the notification. #[inline] pub(super) fn cross(thread: &'r WorkerThread) -> SpinLatch<'r> { - SpinLatch { - cross: true, - ..SpinLatch::new(thread) - } + SpinLatch { cross: true, ..SpinLatch::new(thread) } } #[inline] @@ -235,10 +225,7 @@ pub(super) struct LockLatch { impl LockLatch { #[inline] pub(super) fn new() -> LockLatch { - LockLatch { - m: Mutex::new(false), - v: Condvar::new(), - } + LockLatch { m: Mutex::new(false), v: Condvar::new() } } /// Block until latch is set, then resets this lock latch so it can be reused again. @@ -288,9 +275,7 @@ pub(super) struct OnceLatch { impl OnceLatch { #[inline] pub(super) fn new() -> OnceLatch { - Self { - core_latch: CoreLatch::new(), - } + Self { core_latch: CoreLatch::new() } } /// Set the latch, then tickle the specific worker thread, @@ -372,9 +357,7 @@ impl CountLatch { registry: Arc::clone(owner.registry()), worker_index: owner.index(), }, - None => CountLatchKind::Blocking { - latch: LockLatch::new(), - }, + None => CountLatchKind::Blocking { latch: LockLatch::new() }, }, } } @@ -387,11 +370,7 @@ impl CountLatch { pub(super) fn wait(&self, owner: Option<&WorkerThread>) { match &self.kind { - CountLatchKind::Stealing { - latch, - registry, - worker_index, - } => unsafe { + CountLatchKind::Stealing { latch, registry, worker_index } => unsafe { let owner = owner.expect("owner thread"); debug_assert_eq!(registry.id(), owner.registry().id()); debug_assert_eq!(*worker_index, owner.index()); @@ -409,11 +388,7 @@ impl Latch for CountLatch { // NOTE: Once we call `set` on the internal `latch`, // the target may proceed and invalidate `this`! match (*this).kind { - CountLatchKind::Stealing { - ref latch, - ref registry, - worker_index, - } => { + CountLatchKind::Stealing { ref latch, ref registry, worker_index } => { let registry = Arc::clone(registry); if CoreLatch::set(latch) { registry.notify_worker_latch_is_set(worker_index); @@ -433,10 +408,7 @@ pub(super) struct LatchRef<'a, L> { impl LatchRef<'_, L> { pub(super) fn new(inner: &L) -> LatchRef<'_, L> { - LatchRef { - inner, - marker: PhantomData, - } + LatchRef { inner, marker: PhantomData } } } diff --git a/compiler/rustc_thread_pool/src/lib.rs b/compiler/rustc_thread_pool/src/lib.rs index 72064547e52..179d63ed668 100644 --- a/compiler/rustc_thread_pool/src/lib.rs +++ b/compiler/rustc_thread_pool/src/lib.rs @@ -57,20 +57,17 @@ //! //! While we strive to keep `rayon-core` semver-compatible, it's still //! possible to arrive at this situation if different crates have overly -//! restrictive tilde or inequality requirements for `rayon-core`. The +//! restrictive tilde or inequality requirements for `rayon-core`. The //! conflicting requirements will need to be resolved before the build will //! succeed. #![warn(rust_2018_idioms)] use std::any::Any; -use std::env; use std::error::Error; -use std::fmt; -use std::io; use std::marker::PhantomData; use std::str::FromStr; -use std::thread; +use std::{env, fmt, io, thread}; #[macro_use] mod private; @@ -92,20 +89,18 @@ mod test; pub mod tlv; -pub use self::broadcast::{broadcast, spawn_broadcast, BroadcastContext}; -pub use self::join::{join, join_context}; -pub use self::registry::ThreadBuilder; -pub use self::registry::{mark_blocked, mark_unblocked, Registry}; -pub use self::scope::{in_place_scope, scope, Scope}; -pub use self::scope::{in_place_scope_fifo, scope_fifo, ScopeFifo}; -pub use self::spawn::{spawn, spawn_fifo}; -pub use self::thread_pool::current_thread_has_pending_tasks; -pub use self::thread_pool::current_thread_index; -pub use self::thread_pool::ThreadPool; -pub use self::thread_pool::{yield_local, yield_now, Yield}; pub use worker_local::WorkerLocal; +pub use self::broadcast::{BroadcastContext, broadcast, spawn_broadcast}; +pub use self::join::{join, join_context}; use self::registry::{CustomSpawn, DefaultSpawn, ThreadSpawn}; +pub use self::registry::{Registry, ThreadBuilder, mark_blocked, mark_unblocked}; +pub use self::scope::{Scope, ScopeFifo, in_place_scope, in_place_scope_fifo, scope, scope_fifo}; +pub use self::spawn::{spawn, spawn_fifo}; +pub use self::thread_pool::{ + ThreadPool, Yield, current_thread_has_pending_tasks, current_thread_index, yield_local, + yield_now, +}; /// Returns the maximum number of threads that Rayon supports in a single thread-pool. /// @@ -282,7 +277,7 @@ where } /// Initializes the global thread pool. This initialization is - /// **optional**. If you do not call this function, the thread pool + /// **optional**. If you do not call this function, the thread pool /// will be automatically initialized with the default /// configuration. Calling `build_global` is not recommended, except /// in two scenarios: @@ -290,7 +285,7 @@ where /// - You wish to change the default configuration. /// - You are running a benchmark, in which case initializing may /// yield slightly more consistent results, since the worker threads - /// will already be ready to go even in the first iteration. But + /// will already be ready to go even in the first iteration. But /// this cost is minimal. /// /// Initialization of the global thread pool happens exactly @@ -490,26 +485,16 @@ impl ThreadPoolBuilder { if self.num_threads > 0 { self.num_threads } else { - let default = || { - thread::available_parallelism() - .map(|n| n.get()) - .unwrap_or(1) - }; - - match env::var("RAYON_NUM_THREADS") - .ok() - .and_then(|s| usize::from_str(&s).ok()) - { + let default = || thread::available_parallelism().map(|n| n.get()).unwrap_or(1); + + match env::var("RAYON_NUM_THREADS").ok().and_then(|s| usize::from_str(&s).ok()) { Some(x @ 1..) => return x, Some(0) => return default(), _ => {} } // Support for deprecated `RAYON_RS_NUM_CPUS`. - match env::var("RAYON_RS_NUM_CPUS") - .ok() - .and_then(|s| usize::from_str(&s).ok()) - { + match env::var("RAYON_RS_NUM_CPUS").ok().and_then(|s| usize::from_str(&s).ok()) { Some(x @ 1..) => x, _ => default(), } @@ -723,9 +708,7 @@ impl ThreadPoolBuilder { impl Configuration { /// Creates and return a valid rayon thread pool configuration, but does not initialize it. pub fn new() -> Configuration { - Configuration { - builder: ThreadPoolBuilder::new(), - } + Configuration { builder: ThreadPoolBuilder::new() } } /// Deprecated in favor of `ThreadPoolBuilder::build`. @@ -905,10 +888,7 @@ pub struct FnContext { impl FnContext { #[inline] fn new(migrated: bool) -> Self { - FnContext { - migrated, - _marker: PhantomData, - } + FnContext { migrated, _marker: PhantomData } } } diff --git a/compiler/rustc_thread_pool/src/private.rs b/compiler/rustc_thread_pool/src/private.rs index c85e77b9cbb..5d4f4a8c2ca 100644 --- a/compiler/rustc_thread_pool/src/private.rs +++ b/compiler/rustc_thread_pool/src/private.rs @@ -1,5 +1,5 @@ //! The public parts of this private module are used to create traits -//! that cannot be implemented outside of our own crate. This way we +//! that cannot be implemented outside of our own crate. This way we //! can feel free to extend those traits without worrying about it //! being a breaking change for other implementations. diff --git a/compiler/rustc_thread_pool/src/registry.rs b/compiler/rustc_thread_pool/src/registry.rs index 781b6827b82..2848556aab6 100644 --- a/compiler/rustc_thread_pool/src/registry.rs +++ b/compiler/rustc_thread_pool/src/registry.rs @@ -1,23 +1,20 @@ +use std::cell::Cell; +use std::collections::hash_map::DefaultHasher; +use std::hash::Hasher; +use std::sync::atomic::{AtomicUsize, Ordering}; +use std::sync::{Arc, Mutex, Once}; +use std::{fmt, io, mem, ptr, thread}; + +use crossbeam_deque::{Injector, Steal, Stealer, Worker}; + use crate::job::{JobFifo, JobRef, StackJob}; use crate::latch::{AsCoreLatch, CoreLatch, Latch, LatchRef, LockLatch, OnceLatch, SpinLatch}; use crate::sleep::Sleep; use crate::tlv::Tlv; -use crate::unwind; use crate::{ AcquireThreadHandler, DeadlockHandler, ErrorKind, ExitHandler, PanicHandler, - ReleaseThreadHandler, StartHandler, ThreadPoolBuildError, ThreadPoolBuilder, Yield, + ReleaseThreadHandler, StartHandler, ThreadPoolBuildError, ThreadPoolBuilder, Yield, unwind, }; -use crossbeam_deque::{Injector, Steal, Stealer, Worker}; -use std::cell::Cell; -use std::collections::hash_map::DefaultHasher; -use std::fmt; -use std::hash::Hasher; -use std::io; -use std::mem; -use std::ptr; -use std::sync::atomic::{AtomicUsize, Ordering}; -use std::sync::{Arc, Mutex, Once}; -use std::thread; /// Thread builder used for customization via /// [`ThreadPoolBuilder::spawn_handler`](struct.ThreadPoolBuilder.html#method.spawn_handler). @@ -193,9 +190,7 @@ fn set_global_registry(registry: F) -> Result<&'static Arc, ThreadP where F: FnOnce() -> Result, ThreadPoolBuildError>, { - let mut result = Err(ThreadPoolBuildError::new( - ErrorKind::GlobalPoolAlreadyInitialized, - )); + let mut result = Err(ThreadPoolBuildError::new(ErrorKind::GlobalPoolAlreadyInitialized)); THE_REGISTRY_SET.call_once(|| { result = registry().map(|registry: Arc| { @@ -222,25 +217,23 @@ fn default_global_registry() -> Result, ThreadPoolBuildError> { // is stubbed out, and we won't have to change anything if they do add real threading. let unsupported = matches!(&result, Err(e) if e.is_unsupported()); if unsupported && WorkerThread::current().is_null() { - let builder = ThreadPoolBuilder::new() - .num_threads(1) - .spawn_handler(|thread| { - // Rather than starting a new thread, we're just taking over the current thread - // *without* running the main loop, so we can still return from here. - // The WorkerThread is leaked, but we never shutdown the global pool anyway. - let worker_thread = Box::leak(Box::new(WorkerThread::from(thread))); - let registry = &*worker_thread.registry; - let index = worker_thread.index; - - unsafe { - WorkerThread::set_current(worker_thread); - - // let registry know we are ready to do work - Latch::set(®istry.thread_infos[index].primed); - } + let builder = ThreadPoolBuilder::new().num_threads(1).spawn_handler(|thread| { + // Rather than starting a new thread, we're just taking over the current thread + // *without* running the main loop, so we can still return from here. + // The WorkerThread is leaked, but we never shutdown the global pool anyway. + let worker_thread = Box::leak(Box::new(WorkerThread::from(thread))); + let registry = &*worker_thread.registry; + let index = worker_thread.index; + + unsafe { + WorkerThread::set_current(worker_thread); + + // let registry know we are ready to do work + Latch::set(®istry.thread_infos[index].primed); + } - Ok(()) - }); + Ok(()) + }); let fallback_result = Registry::new(builder); if fallback_result.is_ok() { @@ -273,11 +266,7 @@ impl Registry { let (workers, stealers): (Vec<_>, Vec<_>) = (0..n_threads) .map(|_| { - let worker = if breadth_first { - Worker::new_fifo() - } else { - Worker::new_lifo() - }; + let worker = if breadth_first { Worker::new_fifo() } else { Worker::new_lifo() }; let stealer = worker.stealer(); (worker, stealer) @@ -341,7 +330,7 @@ impl Registry { } } - /// Returns the number of threads in the current registry. This + /// Returns the number of threads in the current registry. This /// is better than `Registry::current().num_threads()` because it /// avoids incrementing the `Arc`. pub(super) fn current_num_threads() -> usize { @@ -359,11 +348,7 @@ impl Registry { pub(super) fn current_thread(&self) -> Option<&WorkerThread> { unsafe { let worker = WorkerThread::current().as_ref()?; - if worker.registry().id() == self.id() { - Some(worker) - } else { - None - } + if worker.registry().id() == self.id() { Some(worker) } else { None } } } @@ -371,9 +356,7 @@ impl Registry { pub(super) fn id(&self) -> RegistryId { // We can rely on `self` not to change since we only ever create // registries that are boxed up in an `Arc` (see `new()` above). - RegistryId { - addr: self as *const Self as usize, - } + RegistryId { addr: self as *const Self as usize } } pub(super) fn num_threads(&self) -> usize { @@ -391,7 +374,7 @@ impl Registry { } } - /// Waits for the worker threads to get up and running. This is + /// Waits for the worker threads to get up and running. This is /// meant to be used for benchmarking purposes, primarily, so that /// you can get more consistent numbers by having everything /// "ready to go". @@ -512,7 +495,7 @@ impl Registry { /// If already in a worker-thread of this registry, just execute `op`. /// Otherwise, inject `op` in this thread-pool. Either way, block until `op` /// completes and return its return value. If `op` panics, that panic will - /// be propagated as well. The second argument indicates `true` if injection + /// be propagated as well. The second argument indicates `true` if injection /// was performed, `false` if executed directly. pub(super) fn in_worker(&self, op: OP) -> R where @@ -844,9 +827,7 @@ impl WorkerThread { // The job might have injected local work, so go back to the outer loop. continue 'outer; } else { - self.registry - .sleep - .no_work_found(&mut idle_state, latch, &self) + self.registry.sleep.no_work_found(&mut idle_state, latch, &self) } } @@ -880,9 +861,7 @@ impl WorkerThread { // deques, and finally to injected jobs from the // outside. The idea is to finish what we started before // we take on something new. - self.take_local_job() - .or_else(|| self.steal()) - .or_else(|| self.registry.pop_injected_job()) + self.take_local_job().or_else(|| self.steal()).or_else(|| self.registry.pop_injected_job()) } pub(super) fn yield_now(&self) -> Yield { @@ -984,10 +963,10 @@ unsafe fn main_loop(thread: ThreadBuilder) { registry.release_thread(); } -/// If already in a worker-thread, just execute `op`. Otherwise, +/// If already in a worker-thread, just execute `op`. Otherwise, /// execute `op` in the default thread-pool. Either way, block until /// `op` completes and return its return value. If `op` panics, that -/// panic will be propagated as well. The second argument indicates +/// panic will be propagated as well. The second argument indicates /// `true` if injection was performed, `false` if executed directly. pub(super) fn in_worker(op: OP) -> R where @@ -1026,9 +1005,7 @@ impl XorShift64Star { seed = hasher.finish(); } - XorShift64Star { - state: Cell::new(seed), - } + XorShift64Star { state: Cell::new(seed) } } fn next(&self) -> u64 { diff --git a/compiler/rustc_thread_pool/src/scope/mod.rs b/compiler/rustc_thread_pool/src/scope/mod.rs index 364b322baad..95a4e0b7a18 100644 --- a/compiler/rustc_thread_pool/src/scope/mod.rs +++ b/compiler/rustc_thread_pool/src/scope/mod.rs @@ -5,19 +5,19 @@ //! [`in_place_scope()`]: fn.in_place_scope.html //! [`join()`]: ../join/join.fn.html +use std::any::Any; +use std::marker::PhantomData; +use std::mem::ManuallyDrop; +use std::sync::Arc; +use std::sync::atomic::{AtomicPtr, Ordering}; +use std::{fmt, ptr}; + use crate::broadcast::BroadcastContext; use crate::job::{ArcJob, HeapJob, JobFifo, JobRef}; use crate::latch::{CountLatch, Latch}; -use crate::registry::{global_registry, in_worker, Registry, WorkerThread}; +use crate::registry::{Registry, WorkerThread, global_registry, in_worker}; use crate::tlv::{self, Tlv}; use crate::unwind; -use std::any::Any; -use std::fmt; -use std::marker::PhantomData; -use std::mem::ManuallyDrop; -use std::ptr; -use std::sync::atomic::{AtomicPtr, Ordering}; -use std::sync::Arc; #[cfg(test)] mod test; @@ -53,7 +53,7 @@ struct ScopeBase<'scope> { job_completed_latch: CountLatch, /// 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 + /// 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 /// the closures are only *moved* across threads to be executed. #[allow(clippy::type_complexity)] @@ -179,9 +179,9 @@ struct ScopeBase<'scope> { /// they were spawned. So in this example, absent any stealing, we can /// expect `s.2` to execute before `s.1`, and `t.2` before `t.1`. Other /// threads always steal from the other end of the deque, like FIFO -/// order. The idea is that "recent" tasks are most likely to be fresh +/// order. The idea is that "recent" tasks are most likely to be fresh /// in the local CPU's cache, while other threads can steal older -/// "stale" tasks. For an alternate approach, consider +/// "stale" tasks. For an alternate approach, consider /// [`scope_fifo()`] instead. /// /// [`scope_fifo()`]: fn.scope_fifo.html @@ -353,7 +353,7 @@ where /// /// Under `scope_fifo()`, the spawns are prioritized in a FIFO order on /// the thread from which they were spawned, as opposed to `scope()`'s -/// LIFO. So in this example, we can expect `s.1` to execute before +/// LIFO. So in this example, we can expect `s.1` to execute before /// `s.2`, and `t.1` before `t.2`. Other threads also steal tasks in /// FIFO order, as usual. Overall, this has roughly the same order as /// the now-deprecated [`breadth_first`] option, except the effect is @@ -469,7 +469,7 @@ impl<'scope> Scope<'scope> { } /// Spawns a job into the fork-join scope `self`. This job will - /// execute sometime before the fork-join scope completes. The + /// execute sometime before the fork-join scope completes. The /// job is specified as a closure, and this closure receives its /// own reference to the scope `self` as argument. This can be /// used to inject new jobs into `self`. @@ -539,7 +539,7 @@ impl<'scope> Scope<'scope> { } /// Spawns a job into every thread of the fork-join scope `self`. This job will - /// execute on each thread sometime before the fork-join scope completes. The + /// execute on each thread sometime before the fork-join scope completes. The /// job is specified as a closure, and this closure receives its own reference /// to the scope `self` as argument, as well as a `BroadcastContext`. pub fn spawn_broadcast(&self, body: BODY) @@ -567,7 +567,7 @@ impl<'scope> ScopeFifo<'scope> { } /// Spawns a job into the fork-join scope `self`. This job will - /// execute sometime before the fork-join scope completes. The + /// execute sometime before the fork-join scope completes. The /// job is specified as a closure, and this closure receives its /// own reference to the scope `self` as argument. This can be /// used to inject new jobs into `self`. @@ -575,7 +575,7 @@ impl<'scope> ScopeFifo<'scope> { /// # See also /// /// This method is akin to [`Scope::spawn()`], but with a FIFO - /// priority. The [`scope_fifo` function] has more details about + /// priority. The [`scope_fifo` function] has more details about /// this distinction. /// /// [`Scope::spawn()`]: struct.Scope.html#method.spawn @@ -605,7 +605,7 @@ impl<'scope> ScopeFifo<'scope> { } /// Spawns a job into every thread of the fork-join scope `self`. This job will - /// execute on each thread sometime before the fork-join scope completes. The + /// execute on each thread sometime before the fork-join scope completes. The /// job is specified as a closure, and this closure receives its own reference /// to the scope `self` as argument, as well as a `BroadcastContext`. pub fn spawn_broadcast(&self, body: BODY) diff --git a/compiler/rustc_thread_pool/src/scope/test.rs b/compiler/rustc_thread_pool/src/scope/test.rs deleted file mode 100644 index 4505ba7c4fb..00000000000 --- a/compiler/rustc_thread_pool/src/scope/test.rs +++ /dev/null @@ -1,622 +0,0 @@ -use crate::unwind; -use crate::ThreadPoolBuilder; -use crate::{scope, scope_fifo, Scope, ScopeFifo}; -use rand::{Rng, SeedableRng}; -use rand_xorshift::XorShiftRng; -use std::iter::once; -use std::sync::atomic::{AtomicUsize, Ordering}; -use std::sync::{Barrier, Mutex}; -use std::vec; - -#[test] -fn scope_empty() { - scope(|_| {}); -} - -#[test] -fn scope_result() { - let x = scope(|_| 22); - assert_eq!(x, 22); -} - -#[test] -fn scope_two() { - let counter = &AtomicUsize::new(0); - scope(|s| { - s.spawn(move |_| { - counter.fetch_add(1, Ordering::SeqCst); - }); - s.spawn(move |_| { - counter.fetch_add(10, Ordering::SeqCst); - }); - }); - - let v = counter.load(Ordering::SeqCst); - assert_eq!(v, 11); -} - -#[test] -fn scope_divide_and_conquer() { - let counter_p = &AtomicUsize::new(0); - scope(|s| s.spawn(move |s| divide_and_conquer(s, counter_p, 1024))); - - let counter_s = &AtomicUsize::new(0); - divide_and_conquer_seq(counter_s, 1024); - - let p = counter_p.load(Ordering::SeqCst); - let s = counter_s.load(Ordering::SeqCst); - assert_eq!(p, s); -} - -fn divide_and_conquer<'scope>(scope: &Scope<'scope>, counter: &'scope AtomicUsize, size: usize) { - if size > 1 { - scope.spawn(move |scope| divide_and_conquer(scope, counter, size / 2)); - scope.spawn(move |scope| divide_and_conquer(scope, counter, size / 2)); - } else { - // count the leaves - counter.fetch_add(1, Ordering::SeqCst); - } -} - -fn divide_and_conquer_seq(counter: &AtomicUsize, size: usize) { - if size > 1 { - divide_and_conquer_seq(counter, size / 2); - divide_and_conquer_seq(counter, size / 2); - } else { - // count the leaves - counter.fetch_add(1, Ordering::SeqCst); - } -} - -struct Tree { - value: T, - children: Vec>, -} - -impl Tree { - fn iter(&self) -> vec::IntoIter<&T> { - once(&self.value) - .chain(self.children.iter().flat_map(Tree::iter)) - .collect::>() // seems like it shouldn't be needed... but prevents overflow - .into_iter() - } - - fn update(&mut self, op: OP) - where - OP: Fn(&mut T) + Sync, - T: Send, - { - scope(|s| self.update_in_scope(&op, s)); - } - - fn update_in_scope<'scope, OP>(&'scope mut self, op: &'scope OP, scope: &Scope<'scope>) - where - OP: Fn(&mut T) + Sync, - { - let Tree { - ref mut value, - ref mut children, - } = *self; - scope.spawn(move |scope| { - for child in children { - scope.spawn(move |scope| child.update_in_scope(op, scope)); - } - }); - - op(value); - } -} - -fn random_tree(depth: usize) -> Tree { - assert!(depth > 0); - let mut seed = ::Seed::default(); - (0..).zip(seed.as_mut()).for_each(|(i, x)| *x = i); - let mut rng = XorShiftRng::from_seed(seed); - random_tree1(depth, &mut rng) -} - -fn random_tree1(depth: usize, rng: &mut XorShiftRng) -> Tree { - let children = if depth == 0 { - vec![] - } else { - (0..rng.random_range(0..4)) // somewhere between 0 and 3 children at each level - .map(|_| random_tree1(depth - 1, rng)) - .collect() - }; - - Tree { - value: rng.random_range(0..1_000_000), - children, - } -} - -#[test] -fn update_tree() { - let mut tree: Tree = random_tree(10); - let values: Vec = tree.iter().cloned().collect(); - tree.update(|v| *v += 1); - let new_values: Vec = tree.iter().cloned().collect(); - assert_eq!(values.len(), new_values.len()); - for (&i, &j) in values.iter().zip(&new_values) { - assert_eq!(i + 1, j); - } -} - -/// Check that if you have a chain of scoped tasks where T0 spawns T1 -/// spawns T2 and so forth down to Tn, the stack space should not grow -/// linearly with N. We test this by some unsafe hackery and -/// permitting an approx 10% change with a 10x input change. -#[test] -#[cfg_attr(any(target_os = "emscripten", target_family = "wasm"), ignore)] -fn linear_stack_growth() { - let builder = ThreadPoolBuilder::new().num_threads(1); - let pool = builder.build().unwrap(); - pool.install(|| { - let mut max_diff = Mutex::new(0); - let bottom_of_stack = 0; - scope(|s| the_final_countdown(s, &bottom_of_stack, &max_diff, 5)); - let diff_when_5 = *max_diff.get_mut().unwrap() as f64; - - scope(|s| the_final_countdown(s, &bottom_of_stack, &max_diff, 500)); - let diff_when_500 = *max_diff.get_mut().unwrap() as f64; - - let ratio = diff_when_5 / diff_when_500; - assert!( - ratio > 0.9 && ratio < 1.1, - "stack usage ratio out of bounds: {}", - ratio - ); - }); -} - -fn the_final_countdown<'scope>( - s: &Scope<'scope>, - bottom_of_stack: &'scope i32, - max: &'scope Mutex, - n: usize, -) { - let top_of_stack = 0; - let p = bottom_of_stack as *const i32 as usize; - let q = &top_of_stack as *const i32 as usize; - let diff = if p > q { p - q } else { q - p }; - - let mut data = max.lock().unwrap(); - *data = Ord::max(diff, *data); - - if n > 0 { - s.spawn(move |s| the_final_countdown(s, bottom_of_stack, max, n - 1)); - } -} - -#[test] -#[should_panic(expected = "Hello, world!")] -fn panic_propagate_scope() { - scope(|_| panic!("Hello, world!")); -} - -#[test] -#[should_panic(expected = "Hello, world!")] -fn panic_propagate_spawn() { - scope(|s| s.spawn(|_| panic!("Hello, world!"))); -} - -#[test] -#[should_panic(expected = "Hello, world!")] -fn panic_propagate_nested_spawn() { - scope(|s| s.spawn(|s| s.spawn(|s| s.spawn(|_| panic!("Hello, world!"))))); -} - -#[test] -#[should_panic(expected = "Hello, world!")] -fn panic_propagate_nested_scope_spawn() { - scope(|s| s.spawn(|_| scope(|s| s.spawn(|_| panic!("Hello, world!"))))); -} - -#[test] -#[cfg_attr(not(panic = "unwind"), ignore)] -fn panic_propagate_still_execute_1() { - let mut x = false; - let result = unwind::halt_unwinding(|| { - scope(|s| { - s.spawn(|_| panic!("Hello, world!")); // job A - s.spawn(|_| x = true); // job B, should still execute even though A panics - }); - }); - match result { - Ok(_) => panic!("failed to propagate panic"), - Err(_) => assert!(x, "job b failed to execute"), - } -} - -#[test] -#[cfg_attr(not(panic = "unwind"), ignore)] -fn panic_propagate_still_execute_2() { - let mut x = false; - let result = unwind::halt_unwinding(|| { - scope(|s| { - s.spawn(|_| x = true); // job B, should still execute even though A panics - s.spawn(|_| panic!("Hello, world!")); // job A - }); - }); - match result { - Ok(_) => panic!("failed to propagate panic"), - Err(_) => assert!(x, "job b failed to execute"), - } -} - -#[test] -#[cfg_attr(not(panic = "unwind"), ignore)] -fn panic_propagate_still_execute_3() { - let mut x = false; - let result = unwind::halt_unwinding(|| { - scope(|s| { - s.spawn(|_| x = true); // spawned job should still execute despite later panic - panic!("Hello, world!"); - }); - }); - match result { - Ok(_) => panic!("failed to propagate panic"), - Err(_) => assert!(x, "panic after spawn, spawn failed to execute"), - } -} - -#[test] -#[cfg_attr(not(panic = "unwind"), ignore)] -fn panic_propagate_still_execute_4() { - let mut x = false; - let result = unwind::halt_unwinding(|| { - scope(|s| { - s.spawn(|_| panic!("Hello, world!")); - x = true; - }); - }); - match result { - Ok(_) => panic!("failed to propagate panic"), - Err(_) => assert!(x, "panic in spawn tainted scope"), - } -} - -macro_rules! test_order { - ($scope:ident => $spawn:ident) => {{ - let builder = ThreadPoolBuilder::new().num_threads(1); - let pool = builder.build().unwrap(); - pool.install(|| { - let vec = Mutex::new(vec![]); - $scope(|scope| { - let vec = &vec; - for i in 0..10 { - scope.$spawn(move |scope| { - for j in 0..10 { - scope.$spawn(move |_| { - vec.lock().unwrap().push(i * 10 + j); - }); - } - }); - } - }); - vec.into_inner().unwrap() - }) - }}; -} - -#[test] -#[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. - let vec = test_order!(scope => spawn); - let expected: Vec = (0..100).rev().collect(); // LIFO -> reversed - assert_eq!(vec, expected); -} - -#[test] -#[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. - let vec = test_order!(scope_fifo => spawn_fifo); - let expected: Vec = (0..100).collect(); // FIFO -> natural order - assert_eq!(vec, expected); -} - -macro_rules! test_nested_order { - ($outer_scope:ident => $outer_spawn:ident, - $inner_scope:ident => $inner_spawn:ident) => {{ - let builder = ThreadPoolBuilder::new().num_threads(1); - let pool = builder.build().unwrap(); - pool.install(|| { - let vec = Mutex::new(vec![]); - $outer_scope(|scope| { - let vec = &vec; - for i in 0..10 { - scope.$outer_spawn(move |_| { - $inner_scope(|scope| { - for j in 0..10 { - scope.$inner_spawn(move |_| { - vec.lock().unwrap().push(i * 10 + j); - }); - } - }); - }); - } - }); - vec.into_inner().unwrap() - }) - }}; -} - -#[test] -#[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. - let vec = test_nested_order!(scope => spawn, scope => spawn); - let expected: Vec = (0..100).rev().collect(); // LIFO -> reversed - assert_eq!(vec, expected); -} - -#[test] -#[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. - let vec = test_nested_order!(scope_fifo => spawn_fifo, scope_fifo => spawn_fifo); - let expected: Vec = (0..100).collect(); // FIFO -> natural order - assert_eq!(vec, expected); -} - -#[test] -#[cfg_attr(any(target_os = "emscripten", target_family = "wasm"), ignore)] -fn nested_lifo_fifo_order() { - // LIFO on the outside, FIFO on the inside - let vec = test_nested_order!(scope => spawn, scope_fifo => spawn_fifo); - let expected: Vec = (0..10) - .rev() - .flat_map(|i| (0..10).map(move |j| i * 10 + j)) - .collect(); - assert_eq!(vec, expected); -} - -#[test] -#[cfg_attr(any(target_os = "emscripten", target_family = "wasm"), ignore)] -fn nested_fifo_lifo_order() { - // FIFO on the outside, LIFO on the inside - let vec = test_nested_order!(scope_fifo => spawn_fifo, scope => spawn); - let expected: Vec = (0..10) - .flat_map(|i| (0..10).rev().map(move |j| i * 10 + j)) - .collect(); - assert_eq!(vec, expected); -} - -macro_rules! spawn_push { - ($scope:ident . $spawn:ident, $vec:ident, $i:expr) => {{ - $scope.$spawn(move |_| $vec.lock().unwrap().push($i)); - }}; -} - -/// Test spawns pushing a series of numbers, interleaved -/// such that negative values are using an inner scope. -macro_rules! test_mixed_order { - ($outer_scope:ident => $outer_spawn:ident, - $inner_scope:ident => $inner_spawn:ident) => {{ - let builder = ThreadPoolBuilder::new().num_threads(1); - let pool = builder.build().unwrap(); - pool.install(|| { - let vec = Mutex::new(vec![]); - $outer_scope(|outer_scope| { - let vec = &vec; - spawn_push!(outer_scope.$outer_spawn, vec, 0); - $inner_scope(|inner_scope| { - spawn_push!(inner_scope.$inner_spawn, vec, -1); - spawn_push!(outer_scope.$outer_spawn, vec, 1); - spawn_push!(inner_scope.$inner_spawn, vec, -2); - spawn_push!(outer_scope.$outer_spawn, vec, 2); - spawn_push!(inner_scope.$inner_spawn, vec, -3); - }); - spawn_push!(outer_scope.$outer_spawn, vec, 3); - }); - vec.into_inner().unwrap() - }) - }}; -} - -#[test] -#[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 - // before they've all been spawned, so they're not perfectly LIFO. - let vec = test_mixed_order!(scope => spawn, scope => spawn); - let expected = vec![-3, 2, -2, 1, -1, 3, 0]; - assert_eq!(vec, expected); -} - -#[test] -#[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); - let expected = vec![-1, 0, -2, 1, -3, 2, 3]; - assert_eq!(vec, expected); -} - -#[test] -#[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 - // before they've all been spawned, so they're not perfectly LIFO. - let vec = test_mixed_order!(scope => spawn, scope_fifo => spawn_fifo); - let expected = vec![-1, 2, -2, 1, -3, 3, 0]; - assert_eq!(vec, expected); -} - -#[test] -#[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); - let expected = vec![-3, 0, -2, 1, -1, 2, 3]; - assert_eq!(vec, expected); -} - -#[test] -fn static_scope() { - static COUNTER: AtomicUsize = AtomicUsize::new(0); - - let mut range = 0..100; - let sum = range.clone().sum(); - let iter = &mut range; - - COUNTER.store(0, Ordering::Relaxed); - scope(|s: &Scope<'static>| { - // While we're allowed the locally borrowed iterator, - // the spawns must be static. - for i in iter { - s.spawn(move |_| { - COUNTER.fetch_add(i, Ordering::Relaxed); - }); - } - }); - - assert_eq!(COUNTER.load(Ordering::Relaxed), sum); -} - -#[test] -fn static_scope_fifo() { - static COUNTER: AtomicUsize = AtomicUsize::new(0); - - let mut range = 0..100; - let sum = range.clone().sum(); - let iter = &mut range; - - COUNTER.store(0, Ordering::Relaxed); - scope_fifo(|s: &ScopeFifo<'static>| { - // While we're allowed the locally borrowed iterator, - // the spawns must be static. - for i in iter { - s.spawn_fifo(move |_| { - COUNTER.fetch_add(i, Ordering::Relaxed); - }); - } - }); - - assert_eq!(COUNTER.load(Ordering::Relaxed), sum); -} - -#[test] -fn mixed_lifetime_scope() { - fn increment<'slice, 'counter>(counters: &'slice [&'counter AtomicUsize]) { - scope(move |s: &Scope<'counter>| { - // We can borrow 'slice here, but the spawns can only borrow 'counter. - for &c in counters { - s.spawn(move |_| { - c.fetch_add(1, Ordering::Relaxed); - }); - } - }); - } - - let counter = AtomicUsize::new(0); - increment(&[&counter; 100]); - assert_eq!(counter.into_inner(), 100); -} - -#[test] -fn mixed_lifetime_scope_fifo() { - fn increment<'slice, 'counter>(counters: &'slice [&'counter AtomicUsize]) { - scope_fifo(move |s: &ScopeFifo<'counter>| { - // We can borrow 'slice here, but the spawns can only borrow 'counter. - for &c in counters { - s.spawn_fifo(move |_| { - c.fetch_add(1, Ordering::Relaxed); - }); - } - }); - } - - let counter = AtomicUsize::new(0); - increment(&[&counter; 100]); - assert_eq!(counter.into_inner(), 100); -} - -#[test] -fn scope_spawn_broadcast() { - let sum = AtomicUsize::new(0); - let n = scope(|s| { - s.spawn_broadcast(|_, ctx| { - sum.fetch_add(ctx.index(), Ordering::Relaxed); - }); - crate::current_num_threads() - }); - assert_eq!(sum.into_inner(), n * (n - 1) / 2); -} - -#[test] -fn scope_fifo_spawn_broadcast() { - let sum = AtomicUsize::new(0); - let n = scope_fifo(|s| { - s.spawn_broadcast(|_, ctx| { - sum.fetch_add(ctx.index(), Ordering::Relaxed); - }); - crate::current_num_threads() - }); - assert_eq!(sum.into_inner(), n * (n - 1) / 2); -} - -#[test] -fn scope_spawn_broadcast_nested() { - let sum = AtomicUsize::new(0); - let n = scope(|s| { - s.spawn_broadcast(|s, _| { - s.spawn_broadcast(|_, ctx| { - sum.fetch_add(ctx.index(), Ordering::Relaxed); - }); - }); - crate::current_num_threads() - }); - assert_eq!(sum.into_inner(), n * n * (n - 1) / 2); -} - -#[test] -#[cfg_attr(any(target_os = "emscripten", target_family = "wasm"), ignore)] -fn scope_spawn_broadcast_barrier() { - let barrier = Barrier::new(8); - let pool = ThreadPoolBuilder::new().num_threads(7).build().unwrap(); - pool.in_place_scope(|s| { - s.spawn_broadcast(|_, _| { - barrier.wait(); - }); - barrier.wait(); - }); -} - -#[test] -#[cfg_attr(any(target_os = "emscripten", target_family = "wasm"), ignore)] -fn scope_spawn_broadcast_panic_one() { - let count = AtomicUsize::new(0); - let pool = ThreadPoolBuilder::new().num_threads(7).build().unwrap(); - let result = crate::unwind::halt_unwinding(|| { - pool.scope(|s| { - s.spawn_broadcast(|_, ctx| { - count.fetch_add(1, Ordering::Relaxed); - if ctx.index() == 3 { - panic!("Hello, world!"); - } - }); - }); - }); - assert_eq!(count.into_inner(), 7); - assert!(result.is_err(), "broadcast panic should propagate!"); -} - -#[test] -#[cfg_attr(any(target_os = "emscripten", target_family = "wasm"), ignore)] -fn scope_spawn_broadcast_panic_many() { - let count = AtomicUsize::new(0); - let pool = ThreadPoolBuilder::new().num_threads(7).build().unwrap(); - let result = crate::unwind::halt_unwinding(|| { - pool.scope(|s| { - s.spawn_broadcast(|_, ctx| { - count.fetch_add(1, Ordering::Relaxed); - if ctx.index() % 2 == 0 { - panic!("Hello, world!"); - } - }); - }); - }); - assert_eq!(count.into_inner(), 7); - assert!(result.is_err(), "broadcast panic should propagate!"); -} diff --git a/compiler/rustc_thread_pool/src/scope/tests.rs b/compiler/rustc_thread_pool/src/scope/tests.rs new file mode 100644 index 00000000000..2df3bc67e29 --- /dev/null +++ b/compiler/rustc_thread_pool/src/scope/tests.rs @@ -0,0 +1,607 @@ +use std::iter::once; +use std::sync::atomic::{AtomicUsize, Ordering}; +use std::sync::{Barrier, Mutex}; +use std::vec; + +use rand::{Rng, SeedableRng}; +use rand_xorshift::XorShiftRng; + +use crate::{Scope, ScopeFifo, ThreadPoolBuilder, scope, scope_fifo, unwind}; + +#[test] +fn scope_empty() { + scope(|_| {}); +} + +#[test] +fn scope_result() { + let x = scope(|_| 22); + assert_eq!(x, 22); +} + +#[test] +fn scope_two() { + let counter = &AtomicUsize::new(0); + scope(|s| { + s.spawn(move |_| { + counter.fetch_add(1, Ordering::SeqCst); + }); + s.spawn(move |_| { + counter.fetch_add(10, Ordering::SeqCst); + }); + }); + + let v = counter.load(Ordering::SeqCst); + assert_eq!(v, 11); +} + +#[test] +fn scope_divide_and_conquer() { + let counter_p = &AtomicUsize::new(0); + scope(|s| s.spawn(move |s| divide_and_conquer(s, counter_p, 1024))); + + let counter_s = &AtomicUsize::new(0); + divide_and_conquer_seq(counter_s, 1024); + + let p = counter_p.load(Ordering::SeqCst); + let s = counter_s.load(Ordering::SeqCst); + assert_eq!(p, s); +} + +fn divide_and_conquer<'scope>(scope: &Scope<'scope>, counter: &'scope AtomicUsize, size: usize) { + if size > 1 { + scope.spawn(move |scope| divide_and_conquer(scope, counter, size / 2)); + scope.spawn(move |scope| divide_and_conquer(scope, counter, size / 2)); + } else { + // count the leaves + counter.fetch_add(1, Ordering::SeqCst); + } +} + +fn divide_and_conquer_seq(counter: &AtomicUsize, size: usize) { + if size > 1 { + divide_and_conquer_seq(counter, size / 2); + divide_and_conquer_seq(counter, size / 2); + } else { + // count the leaves + counter.fetch_add(1, Ordering::SeqCst); + } +} + +struct Tree { + value: T, + children: Vec>, +} + +impl Tree { + fn iter(&self) -> vec::IntoIter<&T> { + once(&self.value) + .chain(self.children.iter().flat_map(Tree::iter)) + .collect::>() // seems like it shouldn't be needed... but prevents overflow + .into_iter() + } + + fn update(&mut self, op: OP) + where + OP: Fn(&mut T) + Sync, + T: Send, + { + scope(|s| self.update_in_scope(&op, s)); + } + + fn update_in_scope<'scope, OP>(&'scope mut self, op: &'scope OP, scope: &Scope<'scope>) + where + OP: Fn(&mut T) + Sync, + { + let Tree { ref mut value, ref mut children } = *self; + scope.spawn(move |scope| { + for child in children { + scope.spawn(move |scope| child.update_in_scope(op, scope)); + } + }); + + op(value); + } +} + +fn random_tree(depth: usize) -> Tree { + assert!(depth > 0); + let mut seed = ::Seed::default(); + (0..).zip(seed.as_mut()).for_each(|(i, x)| *x = i); + let mut rng = XorShiftRng::from_seed(seed); + random_tree1(depth, &mut rng) +} + +fn random_tree1(depth: usize, rng: &mut XorShiftRng) -> Tree { + let children = if depth == 0 { + vec![] + } else { + (0..rng.random_range(0..4)) // somewhere between 0 and 3 children at each level + .map(|_| random_tree1(depth - 1, rng)) + .collect() + }; + + Tree { value: rng.random_range(0..1_000_000), children } +} + +#[test] +fn update_tree() { + let mut tree: Tree = random_tree(10); + let values: Vec = tree.iter().cloned().collect(); + tree.update(|v| *v += 1); + let new_values: Vec = tree.iter().cloned().collect(); + assert_eq!(values.len(), new_values.len()); + for (&i, &j) in values.iter().zip(&new_values) { + assert_eq!(i + 1, j); + } +} + +/// Check that if you have a chain of scoped tasks where T0 spawns T1 +/// spawns T2 and so forth down to Tn, the stack space should not grow +/// linearly with N. We test this by some unsafe hackery and +/// permitting an approx 10% change with a 10x input change. +#[test] +#[cfg_attr(any(target_os = "emscripten", target_family = "wasm"), ignore)] +fn linear_stack_growth() { + let builder = ThreadPoolBuilder::new().num_threads(1); + let pool = builder.build().unwrap(); + pool.install(|| { + let mut max_diff = Mutex::new(0); + let bottom_of_stack = 0; + scope(|s| the_final_countdown(s, &bottom_of_stack, &max_diff, 5)); + let diff_when_5 = *max_diff.get_mut().unwrap() as f64; + + scope(|s| the_final_countdown(s, &bottom_of_stack, &max_diff, 500)); + let diff_when_500 = *max_diff.get_mut().unwrap() as f64; + + let ratio = diff_when_5 / diff_when_500; + assert!(ratio > 0.9 && ratio < 1.1, "stack usage ratio out of bounds: {}", ratio); + }); +} + +fn the_final_countdown<'scope>( + s: &Scope<'scope>, + bottom_of_stack: &'scope i32, + max: &'scope Mutex, + n: usize, +) { + let top_of_stack = 0; + let p = bottom_of_stack as *const i32 as usize; + let q = &top_of_stack as *const i32 as usize; + let diff = if p > q { p - q } else { q - p }; + + let mut data = max.lock().unwrap(); + *data = Ord::max(diff, *data); + + if n > 0 { + s.spawn(move |s| the_final_countdown(s, bottom_of_stack, max, n - 1)); + } +} + +#[test] +#[should_panic(expected = "Hello, world!")] +fn panic_propagate_scope() { + scope(|_| panic!("Hello, world!")); +} + +#[test] +#[should_panic(expected = "Hello, world!")] +fn panic_propagate_spawn() { + scope(|s| s.spawn(|_| panic!("Hello, world!"))); +} + +#[test] +#[should_panic(expected = "Hello, world!")] +fn panic_propagate_nested_spawn() { + scope(|s| s.spawn(|s| s.spawn(|s| s.spawn(|_| panic!("Hello, world!"))))); +} + +#[test] +#[should_panic(expected = "Hello, world!")] +fn panic_propagate_nested_scope_spawn() { + scope(|s| s.spawn(|_| scope(|s| s.spawn(|_| panic!("Hello, world!"))))); +} + +#[test] +#[cfg_attr(not(panic = "unwind"), ignore)] +fn panic_propagate_still_execute_1() { + let mut x = false; + let result = unwind::halt_unwinding(|| { + scope(|s| { + s.spawn(|_| panic!("Hello, world!")); // job A + s.spawn(|_| x = true); // job B, should still execute even though A panics + }); + }); + match result { + Ok(_) => panic!("failed to propagate panic"), + Err(_) => assert!(x, "job b failed to execute"), + } +} + +#[test] +#[cfg_attr(not(panic = "unwind"), ignore)] +fn panic_propagate_still_execute_2() { + let mut x = false; + let result = unwind::halt_unwinding(|| { + scope(|s| { + s.spawn(|_| x = true); // job B, should still execute even though A panics + s.spawn(|_| panic!("Hello, world!")); // job A + }); + }); + match result { + Ok(_) => panic!("failed to propagate panic"), + Err(_) => assert!(x, "job b failed to execute"), + } +} + +#[test] +#[cfg_attr(not(panic = "unwind"), ignore)] +fn panic_propagate_still_execute_3() { + let mut x = false; + let result = unwind::halt_unwinding(|| { + scope(|s| { + s.spawn(|_| x = true); // spawned job should still execute despite later panic + panic!("Hello, world!"); + }); + }); + match result { + Ok(_) => panic!("failed to propagate panic"), + Err(_) => assert!(x, "panic after spawn, spawn failed to execute"), + } +} + +#[test] +#[cfg_attr(not(panic = "unwind"), ignore)] +fn panic_propagate_still_execute_4() { + let mut x = false; + let result = unwind::halt_unwinding(|| { + scope(|s| { + s.spawn(|_| panic!("Hello, world!")); + x = true; + }); + }); + match result { + Ok(_) => panic!("failed to propagate panic"), + Err(_) => assert!(x, "panic in spawn tainted scope"), + } +} + +macro_rules! test_order { + ($scope:ident => $spawn:ident) => {{ + let builder = ThreadPoolBuilder::new().num_threads(1); + let pool = builder.build().unwrap(); + pool.install(|| { + let vec = Mutex::new(vec![]); + $scope(|scope| { + let vec = &vec; + for i in 0..10 { + scope.$spawn(move |scope| { + for j in 0..10 { + scope.$spawn(move |_| { + vec.lock().unwrap().push(i * 10 + j); + }); + } + }); + } + }); + vec.into_inner().unwrap() + }) + }}; +} + +#[test] +#[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. + let vec = test_order!(scope => spawn); + let expected: Vec = (0..100).rev().collect(); // LIFO -> reversed + assert_eq!(vec, expected); +} + +#[test] +#[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. + let vec = test_order!(scope_fifo => spawn_fifo); + let expected: Vec = (0..100).collect(); // FIFO -> natural order + assert_eq!(vec, expected); +} + +macro_rules! test_nested_order { + ($outer_scope:ident => $outer_spawn:ident, + $inner_scope:ident => $inner_spawn:ident) => {{ + let builder = ThreadPoolBuilder::new().num_threads(1); + let pool = builder.build().unwrap(); + pool.install(|| { + let vec = Mutex::new(vec![]); + $outer_scope(|scope| { + let vec = &vec; + for i in 0..10 { + scope.$outer_spawn(move |_| { + $inner_scope(|scope| { + for j in 0..10 { + scope.$inner_spawn(move |_| { + vec.lock().unwrap().push(i * 10 + j); + }); + } + }); + }); + } + }); + vec.into_inner().unwrap() + }) + }}; +} + +#[test] +#[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. + let vec = test_nested_order!(scope => spawn, scope => spawn); + let expected: Vec = (0..100).rev().collect(); // LIFO -> reversed + assert_eq!(vec, expected); +} + +#[test] +#[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. + let vec = test_nested_order!(scope_fifo => spawn_fifo, scope_fifo => spawn_fifo); + let expected: Vec = (0..100).collect(); // FIFO -> natural order + assert_eq!(vec, expected); +} + +#[test] +#[cfg_attr(any(target_os = "emscripten", target_family = "wasm"), ignore)] +fn nested_lifo_fifo_order() { + // LIFO on the outside, FIFO on the inside + let vec = test_nested_order!(scope => spawn, scope_fifo => spawn_fifo); + let expected: Vec = (0..10).rev().flat_map(|i| (0..10).map(move |j| i * 10 + j)).collect(); + assert_eq!(vec, expected); +} + +#[test] +#[cfg_attr(any(target_os = "emscripten", target_family = "wasm"), ignore)] +fn nested_fifo_lifo_order() { + // FIFO on the outside, LIFO on the inside + let vec = test_nested_order!(scope_fifo => spawn_fifo, scope => spawn); + let expected: Vec = (0..10).flat_map(|i| (0..10).rev().map(move |j| i * 10 + j)).collect(); + assert_eq!(vec, expected); +} + +macro_rules! spawn_push { + ($scope:ident . $spawn:ident, $vec:ident, $i:expr) => {{ + $scope.$spawn(move |_| $vec.lock().unwrap().push($i)); + }}; +} + +/// Test spawns pushing a series of numbers, interleaved +/// such that negative values are using an inner scope. +macro_rules! test_mixed_order { + ($outer_scope:ident => $outer_spawn:ident, + $inner_scope:ident => $inner_spawn:ident) => {{ + let builder = ThreadPoolBuilder::new().num_threads(1); + let pool = builder.build().unwrap(); + pool.install(|| { + let vec = Mutex::new(vec![]); + $outer_scope(|outer_scope| { + let vec = &vec; + spawn_push!(outer_scope.$outer_spawn, vec, 0); + $inner_scope(|inner_scope| { + spawn_push!(inner_scope.$inner_spawn, vec, -1); + spawn_push!(outer_scope.$outer_spawn, vec, 1); + spawn_push!(inner_scope.$inner_spawn, vec, -2); + spawn_push!(outer_scope.$outer_spawn, vec, 2); + spawn_push!(inner_scope.$inner_spawn, vec, -3); + }); + spawn_push!(outer_scope.$outer_spawn, vec, 3); + }); + vec.into_inner().unwrap() + }) + }}; +} + +#[test] +#[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 + // before they've all been spawned, so they're not perfectly LIFO. + let vec = test_mixed_order!(scope => spawn, scope => spawn); + let expected = vec![-3, 2, -2, 1, -1, 3, 0]; + assert_eq!(vec, expected); +} + +#[test] +#[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); + let expected = vec![-1, 0, -2, 1, -3, 2, 3]; + assert_eq!(vec, expected); +} + +#[test] +#[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 + // before they've all been spawned, so they're not perfectly LIFO. + let vec = test_mixed_order!(scope => spawn, scope_fifo => spawn_fifo); + let expected = vec![-1, 2, -2, 1, -3, 3, 0]; + assert_eq!(vec, expected); +} + +#[test] +#[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); + let expected = vec![-3, 0, -2, 1, -1, 2, 3]; + assert_eq!(vec, expected); +} + +#[test] +fn static_scope() { + static COUNTER: AtomicUsize = AtomicUsize::new(0); + + let mut range = 0..100; + let sum = range.clone().sum(); + let iter = &mut range; + + COUNTER.store(0, Ordering::Relaxed); + scope(|s: &Scope<'static>| { + // While we're allowed the locally borrowed iterator, + // the spawns must be static. + for i in iter { + s.spawn(move |_| { + COUNTER.fetch_add(i, Ordering::Relaxed); + }); + } + }); + + assert_eq!(COUNTER.load(Ordering::Relaxed), sum); +} + +#[test] +fn static_scope_fifo() { + static COUNTER: AtomicUsize = AtomicUsize::new(0); + + let mut range = 0..100; + let sum = range.clone().sum(); + let iter = &mut range; + + COUNTER.store(0, Ordering::Relaxed); + scope_fifo(|s: &ScopeFifo<'static>| { + // While we're allowed the locally borrowed iterator, + // the spawns must be static. + for i in iter { + s.spawn_fifo(move |_| { + COUNTER.fetch_add(i, Ordering::Relaxed); + }); + } + }); + + assert_eq!(COUNTER.load(Ordering::Relaxed), sum); +} + +#[test] +fn mixed_lifetime_scope() { + fn increment<'slice, 'counter>(counters: &'slice [&'counter AtomicUsize]) { + scope(move |s: &Scope<'counter>| { + // We can borrow 'slice here, but the spawns can only borrow 'counter. + for &c in counters { + s.spawn(move |_| { + c.fetch_add(1, Ordering::Relaxed); + }); + } + }); + } + + let counter = AtomicUsize::new(0); + increment(&[&counter; 100]); + assert_eq!(counter.into_inner(), 100); +} + +#[test] +fn mixed_lifetime_scope_fifo() { + fn increment<'slice, 'counter>(counters: &'slice [&'counter AtomicUsize]) { + scope_fifo(move |s: &ScopeFifo<'counter>| { + // We can borrow 'slice here, but the spawns can only borrow 'counter. + for &c in counters { + s.spawn_fifo(move |_| { + c.fetch_add(1, Ordering::Relaxed); + }); + } + }); + } + + let counter = AtomicUsize::new(0); + increment(&[&counter; 100]); + assert_eq!(counter.into_inner(), 100); +} + +#[test] +fn scope_spawn_broadcast() { + let sum = AtomicUsize::new(0); + let n = scope(|s| { + s.spawn_broadcast(|_, ctx| { + sum.fetch_add(ctx.index(), Ordering::Relaxed); + }); + crate::current_num_threads() + }); + assert_eq!(sum.into_inner(), n * (n - 1) / 2); +} + +#[test] +fn scope_fifo_spawn_broadcast() { + let sum = AtomicUsize::new(0); + let n = scope_fifo(|s| { + s.spawn_broadcast(|_, ctx| { + sum.fetch_add(ctx.index(), Ordering::Relaxed); + }); + crate::current_num_threads() + }); + assert_eq!(sum.into_inner(), n * (n - 1) / 2); +} + +#[test] +fn scope_spawn_broadcast_nested() { + let sum = AtomicUsize::new(0); + let n = scope(|s| { + s.spawn_broadcast(|s, _| { + s.spawn_broadcast(|_, ctx| { + sum.fetch_add(ctx.index(), Ordering::Relaxed); + }); + }); + crate::current_num_threads() + }); + assert_eq!(sum.into_inner(), n * n * (n - 1) / 2); +} + +#[test] +#[cfg_attr(any(target_os = "emscripten", target_family = "wasm"), ignore)] +fn scope_spawn_broadcast_barrier() { + let barrier = Barrier::new(8); + let pool = ThreadPoolBuilder::new().num_threads(7).build().unwrap(); + pool.in_place_scope(|s| { + s.spawn_broadcast(|_, _| { + barrier.wait(); + }); + barrier.wait(); + }); +} + +#[test] +#[cfg_attr(any(target_os = "emscripten", target_family = "wasm"), ignore)] +fn scope_spawn_broadcast_panic_one() { + let count = AtomicUsize::new(0); + let pool = ThreadPoolBuilder::new().num_threads(7).build().unwrap(); + let result = crate::unwind::halt_unwinding(|| { + pool.scope(|s| { + s.spawn_broadcast(|_, ctx| { + count.fetch_add(1, Ordering::Relaxed); + if ctx.index() == 3 { + panic!("Hello, world!"); + } + }); + }); + }); + assert_eq!(count.into_inner(), 7); + assert!(result.is_err(), "broadcast panic should propagate!"); +} + +#[test] +#[cfg_attr(any(target_os = "emscripten", target_family = "wasm"), ignore)] +fn scope_spawn_broadcast_panic_many() { + let count = AtomicUsize::new(0); + let pool = ThreadPoolBuilder::new().num_threads(7).build().unwrap(); + let result = crate::unwind::halt_unwinding(|| { + pool.scope(|s| { + s.spawn_broadcast(|_, ctx| { + count.fetch_add(1, Ordering::Relaxed); + if ctx.index() % 2 == 0 { + panic!("Hello, world!"); + } + }); + }); + }); + assert_eq!(count.into_inner(), 7); + assert!(result.is_err(), "broadcast panic should propagate!"); +} diff --git a/compiler/rustc_thread_pool/src/sleep/README.md b/compiler/rustc_thread_pool/src/sleep/README.md index e79efd15ca9..1e11da55f4a 100644 --- a/compiler/rustc_thread_pool/src/sleep/README.md +++ b/compiler/rustc_thread_pool/src/sleep/README.md @@ -182,7 +182,7 @@ This is possible because the C++ memory model typically offers guarantees of the form "if you see the access A, then you must see those other accesses" -- but it doesn't guarantee that you will see the access A (i.e., if you think of processors with independent caches, you may be operating on very out of date -cache state). +cache state). ## Using seq-cst fences to prevent deadlock diff --git a/compiler/rustc_thread_pool/src/sleep/counters.rs b/compiler/rustc_thread_pool/src/sleep/counters.rs index 05941becd1c..f2682028b96 100644 --- a/compiler/rustc_thread_pool/src/sleep/counters.rs +++ b/compiler/rustc_thread_pool/src/sleep/counters.rs @@ -89,9 +89,7 @@ const ONE_JEC: usize = 1 << JEC_SHIFT; impl AtomicCounters { #[inline] pub(super) fn new() -> AtomicCounters { - AtomicCounters { - value: AtomicUsize::new(0), - } + AtomicCounters { value: AtomicUsize::new(0) } } /// Load and return the current value of the various counters. @@ -230,9 +228,7 @@ impl Counters { fn increment_jobs_counter(self) -> Counters { // We can freely add to JEC because it occupies the most significant bits. // Thus it doesn't overflow into the other counters, just wraps itself. - Counters { - word: self.word.wrapping_add(ONE_JEC), - } + Counters { word: self.word.wrapping_add(ONE_JEC) } } #[inline] diff --git a/compiler/rustc_thread_pool/src/sleep/mod.rs b/compiler/rustc_thread_pool/src/sleep/mod.rs index 7d88ece2107..bee7c82c450 100644 --- a/compiler/rustc_thread_pool/src/sleep/mod.rs +++ b/compiler/rustc_thread_pool/src/sleep/mod.rs @@ -1,14 +1,16 @@ //! Code that decides when workers should go to sleep. See README.md //! for an overview. -use crate::latch::CoreLatch; -use crate::registry::WorkerThread; -use crate::DeadlockHandler; -use crossbeam_utils::CachePadded; use std::sync::atomic::Ordering; use std::sync::{Condvar, Mutex}; use std::thread; +use crossbeam_utils::CachePadded; + +use crate::DeadlockHandler; +use crate::latch::CoreLatch; +use crate::registry::WorkerThread; + mod counters; pub(crate) use self::counters::THREADS_MAX; use self::counters::{AtomicCounters, JobsEventCounter}; @@ -125,11 +127,7 @@ impl Sleep { pub(super) fn start_looking(&self, worker_index: usize) -> IdleState { self.counters.add_inactive_thread(); - IdleState { - worker_index, - rounds: 0, - jobs_counter: JobsEventCounter::DUMMY, - } + IdleState { worker_index, rounds: 0, jobs_counter: JobsEventCounter::DUMMY } } #[inline] @@ -165,9 +163,7 @@ impl Sleep { #[cold] fn announce_sleepy(&self) -> JobsEventCounter { - self.counters - .increment_jobs_event_counter_if(JobsEventCounter::is_active) - .jobs_counter() + self.counters.increment_jobs_event_counter_if(JobsEventCounter::is_active).jobs_counter() } #[cold] @@ -258,7 +254,7 @@ impl Sleep { } /// Notify the given thread that it should wake up (if it is - /// sleeping). When this method is invoked, we typically know the + /// sleeping). When this method is invoked, we typically know the /// thread is asleep, though in rare cases it could have been /// awoken by (e.g.) new work having been posted. pub(super) fn notify_worker_latch_is_set(&self, target_worker_index: usize) { @@ -307,9 +303,7 @@ impl Sleep { // Read the counters and -- if sleepy workers have announced themselves // -- announce that there is now work available. The final value of `counters` // with which we exit the loop thus corresponds to a state when - let counters = self - .counters - .increment_jobs_event_counter_if(JobsEventCounter::is_sleepy); + let counters = self.counters.increment_jobs_event_counter_if(JobsEventCounter::is_sleepy); let num_awake_but_idle = counters.awake_but_idle_threads(); let num_sleepers = counters.sleeping_threads(); diff --git a/compiler/rustc_thread_pool/src/spawn/mod.rs b/compiler/rustc_thread_pool/src/spawn/mod.rs index 034df30dcfb..f1679a98234 100644 --- a/compiler/rustc_thread_pool/src/spawn/mod.rs +++ b/compiler/rustc_thread_pool/src/spawn/mod.rs @@ -1,9 +1,10 @@ +use std::mem; +use std::sync::Arc; + use crate::job::*; use crate::registry::Registry; use crate::tlv::Tlv; use crate::unwind; -use std::mem; -use std::sync::Arc; /// Puts the task into the Rayon threadpool's job queue in the "static" /// or "global" scope. Just like a standard thread, this task is not @@ -28,9 +29,9 @@ use std::sync::Arc; /// other threads may steal tasks at any time. However, they are /// generally prioritized in a LIFO order on the thread from which /// they were spawned. Other threads always steal from the other end of -/// the deque, like FIFO order. The idea is that "recent" tasks are +/// the deque, like FIFO order. The idea is that "recent" tasks are /// most likely to be fresh in the local CPU's cache, while other -/// threads can steal older "stale" tasks. For an alternate approach, +/// threads can steal older "stale" tasks. For an alternate approach, /// consider [`spawn_fifo()`] instead. /// /// [`spawn_fifo()`]: fn.spawn_fifo.html @@ -39,7 +40,7 @@ use std::sync::Arc; /// /// If this closure should panic, the resulting panic will be /// propagated to the panic handler registered in the `ThreadPoolBuilder`, -/// if any. See [`ThreadPoolBuilder::panic_handler()`][ph] for more +/// if any. See [`ThreadPoolBuilder::panic_handler()`][ph] for more /// details. /// /// [ph]: struct.ThreadPoolBuilder.html#method.panic_handler @@ -103,7 +104,7 @@ where } /// Fires off a task into the Rayon threadpool in the "static" or -/// "global" scope. Just like a standard thread, this task is not +/// "global" scope. Just like a standard thread, this task is not /// tied to the current stack frame, and hence it cannot hold any /// references other than those with `'static` lifetime. If you want /// to spawn a task that references stack data, use [the `scope_fifo()` @@ -124,7 +125,7 @@ where /// /// If this closure should panic, the resulting panic will be /// propagated to the panic handler registered in the `ThreadPoolBuilder`, -/// if any. See [`ThreadPoolBuilder::panic_handler()`][ph] for more +/// if any. See [`ThreadPoolBuilder::panic_handler()`][ph] for more /// details. /// /// [ph]: struct.ThreadPoolBuilder.html#method.panic_handler @@ -152,7 +153,7 @@ where let job_ref = spawn_job(func, registry); // If we're in the pool, use our thread's private fifo for this thread to execute - // in a locally-FIFO order. Otherwise, just use the pool's global injector. + // in a locally-FIFO order. Otherwise, just use the pool's global injector. match registry.current_thread() { Some(worker) => worker.push_fifo(job_ref), None => registry.inject(job_ref), diff --git a/compiler/rustc_thread_pool/src/spawn/test.rs b/compiler/rustc_thread_pool/src/spawn/test.rs deleted file mode 100644 index b7a0535aabf..00000000000 --- a/compiler/rustc_thread_pool/src/spawn/test.rs +++ /dev/null @@ -1,255 +0,0 @@ -use crate::scope; -use std::any::Any; -use std::sync::mpsc::channel; -use std::sync::Mutex; - -use super::{spawn, spawn_fifo}; -use crate::ThreadPoolBuilder; - -#[test] -#[cfg_attr(any(target_os = "emscripten", target_family = "wasm"), ignore)] -fn spawn_then_join_in_worker() { - let (tx, rx) = channel(); - scope(move |_| { - spawn(move || tx.send(22).unwrap()); - }); - assert_eq!(22, rx.recv().unwrap()); -} - -#[test] -#[cfg_attr(any(target_os = "emscripten", target_family = "wasm"), ignore)] -fn spawn_then_join_outside_worker() { - let (tx, rx) = channel(); - spawn(move || tx.send(22).unwrap()); - assert_eq!(22, rx.recv().unwrap()); -} - -#[test] -#[cfg_attr(not(panic = "unwind"), ignore)] -fn panic_fwd() { - let (tx, rx) = channel(); - - let tx = Mutex::new(tx); - let panic_handler = move |err: Box| { - let tx = tx.lock().unwrap(); - if let Some(&msg) = err.downcast_ref::<&str>() { - if msg == "Hello, world!" { - tx.send(1).unwrap(); - } else { - tx.send(2).unwrap(); - } - } else { - tx.send(3).unwrap(); - } - }; - - let builder = ThreadPoolBuilder::new().panic_handler(panic_handler); - - builder - .build() - .unwrap() - .spawn(move || panic!("Hello, world!")); - - assert_eq!(1, rx.recv().unwrap()); -} - -/// Test what happens when the thread-pool is dropped but there are -/// still active asynchronous tasks. We expect the thread-pool to stay -/// alive and executing until those threads are complete. -#[test] -#[cfg_attr(any(target_os = "emscripten", target_family = "wasm"), ignore)] -fn termination_while_things_are_executing() { - let (tx0, rx0) = channel(); - let (tx1, rx1) = channel(); - - // Create a thread-pool and spawn some code in it, but then drop - // our reference to it. - { - let thread_pool = ThreadPoolBuilder::new().build().unwrap(); - thread_pool.spawn(move || { - let data = rx0.recv().unwrap(); - - // At this point, we know the "main" reference to the - // `ThreadPool` has been dropped, but there are still - // active threads. Launch one more. - spawn(move || { - tx1.send(data).unwrap(); - }); - }); - } - - tx0.send(22).unwrap(); - let v = rx1.recv().unwrap(); - assert_eq!(v, 22); -} - -#[test] -#[cfg_attr(not(panic = "unwind"), ignore)] -fn custom_panic_handler_and_spawn() { - let (tx, rx) = channel(); - - // Create a parallel closure that will send panics on the - // channel; since the closure is potentially executed in parallel - // with itself, we have to wrap `tx` in a mutex. - let tx = Mutex::new(tx); - let panic_handler = move |e: Box| { - tx.lock().unwrap().send(e).unwrap(); - }; - - // Execute an async that will panic. - let builder = ThreadPoolBuilder::new().panic_handler(panic_handler); - builder.build().unwrap().spawn(move || { - panic!("Hello, world!"); - }); - - // Check that we got back the panic we expected. - let error = rx.recv().unwrap(); - if let Some(&msg) = error.downcast_ref::<&str>() { - assert_eq!(msg, "Hello, world!"); - } else { - panic!("did not receive a string from panic handler"); - } -} - -#[test] -#[cfg_attr(not(panic = "unwind"), ignore)] -fn custom_panic_handler_and_nested_spawn() { - let (tx, rx) = channel(); - - // Create a parallel closure that will send panics on the - // channel; since the closure is potentially executed in parallel - // with itself, we have to wrap `tx` in a mutex. - let tx = Mutex::new(tx); - let panic_handler = move |e| { - tx.lock().unwrap().send(e).unwrap(); - }; - - // Execute an async that will (eventually) panic. - const PANICS: usize = 3; - let builder = ThreadPoolBuilder::new().panic_handler(panic_handler); - builder.build().unwrap().spawn(move || { - // launch 3 nested spawn-asyncs; these should be in the same - // thread-pool and hence inherit the same panic handler - for _ in 0..PANICS { - spawn(move || { - panic!("Hello, world!"); - }); - } - }); - - // Check that we get back the panics we expected. - for _ in 0..PANICS { - let error = rx.recv().unwrap(); - if let Some(&msg) = error.downcast_ref::<&str>() { - assert_eq!(msg, "Hello, world!"); - } else { - panic!("did not receive a string from panic handler"); - } - } -} - -macro_rules! test_order { - ($outer_spawn:ident, $inner_spawn:ident) => {{ - let builder = ThreadPoolBuilder::new().num_threads(1); - let pool = builder.build().unwrap(); - let (tx, rx) = channel(); - pool.install(move || { - for i in 0..10 { - let tx = tx.clone(); - $outer_spawn(move || { - for j in 0..10 { - let tx = tx.clone(); - $inner_spawn(move || { - tx.send(i * 10 + j).unwrap(); - }); - } - }); - } - }); - rx.iter().collect::>() - }}; -} - -#[test] -#[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. - let vec = test_order!(spawn, spawn); - let expected: Vec = (0..100).rev().collect(); // LIFO -> reversed - assert_eq!(vec, expected); -} - -#[test] -#[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. - let vec = test_order!(spawn_fifo, spawn_fifo); - let expected: Vec = (0..100).collect(); // FIFO -> natural order - assert_eq!(vec, expected); -} - -#[test] -#[cfg_attr(any(target_os = "emscripten", target_family = "wasm"), ignore)] -fn lifo_fifo_order() { - // LIFO on the outside, FIFO on the inside - let vec = test_order!(spawn, spawn_fifo); - let expected: Vec = (0..10) - .rev() - .flat_map(|i| (0..10).map(move |j| i * 10 + j)) - .collect(); - assert_eq!(vec, expected); -} - -#[test] -#[cfg_attr(any(target_os = "emscripten", target_family = "wasm"), ignore)] -fn fifo_lifo_order() { - // FIFO on the outside, LIFO on the inside - let vec = test_order!(spawn_fifo, spawn); - let expected: Vec = (0..10) - .flat_map(|i| (0..10).rev().map(move |j| i * 10 + j)) - .collect(); - assert_eq!(vec, expected); -} - -macro_rules! spawn_send { - ($spawn:ident, $tx:ident, $i:expr) => {{ - let tx = $tx.clone(); - $spawn(move || tx.send($i).unwrap()); - }}; -} - -/// Test mixed spawns pushing a series of numbers, interleaved such -/// such that negative values are using the second kind of spawn. -macro_rules! test_mixed_order { - ($pos_spawn:ident, $neg_spawn:ident) => {{ - let builder = ThreadPoolBuilder::new().num_threads(1); - let pool = builder.build().unwrap(); - let (tx, rx) = channel(); - pool.install(move || { - spawn_send!($pos_spawn, tx, 0); - spawn_send!($neg_spawn, tx, -1); - spawn_send!($pos_spawn, tx, 1); - spawn_send!($neg_spawn, tx, -2); - spawn_send!($pos_spawn, tx, 2); - spawn_send!($neg_spawn, tx, -3); - spawn_send!($pos_spawn, tx, 3); - }); - rx.iter().collect::>() - }}; -} - -#[test] -#[cfg_attr(any(target_os = "emscripten", target_family = "wasm"), ignore)] -fn mixed_lifo_fifo_order() { - let vec = test_mixed_order!(spawn, spawn_fifo); - let expected = vec![3, -1, 2, -2, 1, -3, 0]; - assert_eq!(vec, expected); -} - -#[test] -#[cfg_attr(any(target_os = "emscripten", target_family = "wasm"), ignore)] -fn mixed_fifo_lifo_order() { - let vec = test_mixed_order!(spawn_fifo, spawn); - let expected = vec![0, -3, 1, -2, 2, -1, 3]; - assert_eq!(vec, expected); -} diff --git a/compiler/rustc_thread_pool/src/spawn/tests.rs b/compiler/rustc_thread_pool/src/spawn/tests.rs new file mode 100644 index 00000000000..8a70d2faf9c --- /dev/null +++ b/compiler/rustc_thread_pool/src/spawn/tests.rs @@ -0,0 +1,246 @@ +use std::any::Any; +use std::sync::Mutex; +use std::sync::mpsc::channel; + +use super::{spawn, spawn_fifo}; +use crate::{ThreadPoolBuilder, scope}; + +#[test] +#[cfg_attr(any(target_os = "emscripten", target_family = "wasm"), ignore)] +fn spawn_then_join_in_worker() { + let (tx, rx) = channel(); + scope(move |_| { + spawn(move || tx.send(22).unwrap()); + }); + assert_eq!(22, rx.recv().unwrap()); +} + +#[test] +#[cfg_attr(any(target_os = "emscripten", target_family = "wasm"), ignore)] +fn spawn_then_join_outside_worker() { + let (tx, rx) = channel(); + spawn(move || tx.send(22).unwrap()); + assert_eq!(22, rx.recv().unwrap()); +} + +#[test] +#[cfg_attr(not(panic = "unwind"), ignore)] +fn panic_fwd() { + let (tx, rx) = channel(); + + let tx = Mutex::new(tx); + let panic_handler = move |err: Box| { + let tx = tx.lock().unwrap(); + if let Some(&msg) = err.downcast_ref::<&str>() { + if msg == "Hello, world!" { + tx.send(1).unwrap(); + } else { + tx.send(2).unwrap(); + } + } else { + tx.send(3).unwrap(); + } + }; + + let builder = ThreadPoolBuilder::new().panic_handler(panic_handler); + + builder.build().unwrap().spawn(move || panic!("Hello, world!")); + + assert_eq!(1, rx.recv().unwrap()); +} + +/// Test what happens when the thread-pool is dropped but there are +/// still active asynchronous tasks. We expect the thread-pool to stay +/// alive and executing until those threads are complete. +#[test] +#[cfg_attr(any(target_os = "emscripten", target_family = "wasm"), ignore)] +fn termination_while_things_are_executing() { + let (tx0, rx0) = channel(); + let (tx1, rx1) = channel(); + + // Create a thread-pool and spawn some code in it, but then drop + // our reference to it. + { + let thread_pool = ThreadPoolBuilder::new().build().unwrap(); + thread_pool.spawn(move || { + let data = rx0.recv().unwrap(); + + // At this point, we know the "main" reference to the + // `ThreadPool` has been dropped, but there are still + // active threads. Launch one more. + spawn(move || { + tx1.send(data).unwrap(); + }); + }); + } + + tx0.send(22).unwrap(); + let v = rx1.recv().unwrap(); + assert_eq!(v, 22); +} + +#[test] +#[cfg_attr(not(panic = "unwind"), ignore)] +fn custom_panic_handler_and_spawn() { + let (tx, rx) = channel(); + + // Create a parallel closure that will send panics on the + // channel; since the closure is potentially executed in parallel + // with itself, we have to wrap `tx` in a mutex. + let tx = Mutex::new(tx); + let panic_handler = move |e: Box| { + tx.lock().unwrap().send(e).unwrap(); + }; + + // Execute an async that will panic. + let builder = ThreadPoolBuilder::new().panic_handler(panic_handler); + builder.build().unwrap().spawn(move || { + panic!("Hello, world!"); + }); + + // Check that we got back the panic we expected. + let error = rx.recv().unwrap(); + if let Some(&msg) = error.downcast_ref::<&str>() { + assert_eq!(msg, "Hello, world!"); + } else { + panic!("did not receive a string from panic handler"); + } +} + +#[test] +#[cfg_attr(not(panic = "unwind"), ignore)] +fn custom_panic_handler_and_nested_spawn() { + let (tx, rx) = channel(); + + // Create a parallel closure that will send panics on the + // channel; since the closure is potentially executed in parallel + // with itself, we have to wrap `tx` in a mutex. + let tx = Mutex::new(tx); + let panic_handler = move |e| { + tx.lock().unwrap().send(e).unwrap(); + }; + + // Execute an async that will (eventually) panic. + const PANICS: usize = 3; + let builder = ThreadPoolBuilder::new().panic_handler(panic_handler); + builder.build().unwrap().spawn(move || { + // launch 3 nested spawn-asyncs; these should be in the same + // thread-pool and hence inherit the same panic handler + for _ in 0..PANICS { + spawn(move || { + panic!("Hello, world!"); + }); + } + }); + + // Check that we get back the panics we expected. + for _ in 0..PANICS { + let error = rx.recv().unwrap(); + if let Some(&msg) = error.downcast_ref::<&str>() { + assert_eq!(msg, "Hello, world!"); + } else { + panic!("did not receive a string from panic handler"); + } + } +} + +macro_rules! test_order { + ($outer_spawn:ident, $inner_spawn:ident) => {{ + let builder = ThreadPoolBuilder::new().num_threads(1); + let pool = builder.build().unwrap(); + let (tx, rx) = channel(); + pool.install(move || { + for i in 0..10 { + let tx = tx.clone(); + $outer_spawn(move || { + for j in 0..10 { + let tx = tx.clone(); + $inner_spawn(move || { + tx.send(i * 10 + j).unwrap(); + }); + } + }); + } + }); + rx.iter().collect::>() + }}; +} + +#[test] +#[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. + let vec = test_order!(spawn, spawn); + let expected: Vec = (0..100).rev().collect(); // LIFO -> reversed + assert_eq!(vec, expected); +} + +#[test] +#[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. + let vec = test_order!(spawn_fifo, spawn_fifo); + let expected: Vec = (0..100).collect(); // FIFO -> natural order + assert_eq!(vec, expected); +} + +#[test] +#[cfg_attr(any(target_os = "emscripten", target_family = "wasm"), ignore)] +fn lifo_fifo_order() { + // LIFO on the outside, FIFO on the inside + let vec = test_order!(spawn, spawn_fifo); + let expected: Vec = (0..10).rev().flat_map(|i| (0..10).map(move |j| i * 10 + j)).collect(); + assert_eq!(vec, expected); +} + +#[test] +#[cfg_attr(any(target_os = "emscripten", target_family = "wasm"), ignore)] +fn fifo_lifo_order() { + // FIFO on the outside, LIFO on the inside + let vec = test_order!(spawn_fifo, spawn); + let expected: Vec = (0..10).flat_map(|i| (0..10).rev().map(move |j| i * 10 + j)).collect(); + assert_eq!(vec, expected); +} + +macro_rules! spawn_send { + ($spawn:ident, $tx:ident, $i:expr) => {{ + let tx = $tx.clone(); + $spawn(move || tx.send($i).unwrap()); + }}; +} + +/// Test mixed spawns pushing a series of numbers, interleaved such +/// such that negative values are using the second kind of spawn. +macro_rules! test_mixed_order { + ($pos_spawn:ident, $neg_spawn:ident) => {{ + let builder = ThreadPoolBuilder::new().num_threads(1); + let pool = builder.build().unwrap(); + let (tx, rx) = channel(); + pool.install(move || { + spawn_send!($pos_spawn, tx, 0); + spawn_send!($neg_spawn, tx, -1); + spawn_send!($pos_spawn, tx, 1); + spawn_send!($neg_spawn, tx, -2); + spawn_send!($pos_spawn, tx, 2); + spawn_send!($neg_spawn, tx, -3); + spawn_send!($pos_spawn, tx, 3); + }); + rx.iter().collect::>() + }}; +} + +#[test] +#[cfg_attr(any(target_os = "emscripten", target_family = "wasm"), ignore)] +fn mixed_lifo_fifo_order() { + let vec = test_mixed_order!(spawn, spawn_fifo); + let expected = vec![3, -1, 2, -2, 1, -3, 0]; + assert_eq!(vec, expected); +} + +#[test] +#[cfg_attr(any(target_os = "emscripten", target_family = "wasm"), ignore)] +fn mixed_fifo_lifo_order() { + let vec = test_mixed_order!(spawn_fifo, spawn); + let expected = vec![0, -3, 1, -2, 2, -1, 3]; + assert_eq!(vec, expected); +} diff --git a/compiler/rustc_thread_pool/src/test.rs b/compiler/rustc_thread_pool/src/test.rs deleted file mode 100644 index 25b8487f73b..00000000000 --- a/compiler/rustc_thread_pool/src/test.rs +++ /dev/null @@ -1,200 +0,0 @@ -#![cfg(test)] - -use crate::{ThreadPoolBuildError, ThreadPoolBuilder}; -use std::sync::atomic::{AtomicUsize, Ordering}; -use std::sync::{Arc, Barrier}; - -#[test] -#[cfg_attr(any(target_os = "emscripten", target_family = "wasm"), ignore)] -fn worker_thread_index() { - let pool = ThreadPoolBuilder::new().num_threads(22).build().unwrap(); - assert_eq!(pool.current_num_threads(), 22); - assert_eq!(pool.current_thread_index(), None); - let index = pool.install(|| pool.current_thread_index().unwrap()); - assert!(index < 22); -} - -#[test] -#[cfg_attr(any(target_os = "emscripten", target_family = "wasm"), ignore)] -fn start_callback_called() { - let n_threads = 16; - let n_called = Arc::new(AtomicUsize::new(0)); - // Wait for all the threads in the pool plus the one running tests. - let barrier = Arc::new(Barrier::new(n_threads + 1)); - - let b = Arc::clone(&barrier); - let nc = Arc::clone(&n_called); - let start_handler = move |_| { - nc.fetch_add(1, Ordering::SeqCst); - b.wait(); - }; - - let conf = ThreadPoolBuilder::new() - .num_threads(n_threads) - .start_handler(start_handler); - let _ = conf.build().unwrap(); - - // Wait for all the threads to have been scheduled to run. - barrier.wait(); - - // The handler must have been called on every started thread. - assert_eq!(n_called.load(Ordering::SeqCst), n_threads); -} - -#[test] -#[cfg_attr(any(target_os = "emscripten", target_family = "wasm"), ignore)] -fn exit_callback_called() { - let n_threads = 16; - let n_called = Arc::new(AtomicUsize::new(0)); - // Wait for all the threads in the pool plus the one running tests. - let barrier = Arc::new(Barrier::new(n_threads + 1)); - - let b = Arc::clone(&barrier); - let nc = Arc::clone(&n_called); - let exit_handler = move |_| { - nc.fetch_add(1, Ordering::SeqCst); - b.wait(); - }; - - let conf = ThreadPoolBuilder::new() - .num_threads(n_threads) - .exit_handler(exit_handler); - { - let _ = conf.build().unwrap(); - // Drop the pool so it stops the running threads. - } - - // Wait for all the threads to have been scheduled to run. - barrier.wait(); - - // The handler must have been called on every exiting thread. - assert_eq!(n_called.load(Ordering::SeqCst), n_threads); -} - -#[test] -#[cfg_attr(any(target_os = "emscripten", target_family = "wasm"), ignore)] -fn handler_panics_handled_correctly() { - let n_threads = 16; - let n_called = Arc::new(AtomicUsize::new(0)); - // Wait for all the threads in the pool plus the one running tests. - let start_barrier = Arc::new(Barrier::new(n_threads + 1)); - let exit_barrier = Arc::new(Barrier::new(n_threads + 1)); - - let start_handler = move |_| { - panic!("ensure panic handler is called when starting"); - }; - let exit_handler = move |_| { - panic!("ensure panic handler is called when exiting"); - }; - - let sb = Arc::clone(&start_barrier); - let eb = Arc::clone(&exit_barrier); - let nc = Arc::clone(&n_called); - let panic_handler = move |_| { - let val = nc.fetch_add(1, Ordering::SeqCst); - if val < n_threads { - sb.wait(); - } else { - eb.wait(); - } - }; - - let conf = ThreadPoolBuilder::new() - .num_threads(n_threads) - .start_handler(start_handler) - .exit_handler(exit_handler) - .panic_handler(panic_handler); - { - let _ = conf.build().unwrap(); - - // Wait for all the threads to start, panic in the start handler, - // and been taken care of by the panic handler. - start_barrier.wait(); - - // Drop the pool so it stops the running threads. - } - - // Wait for all the threads to exit, panic in the exit handler, - // and been taken care of by the panic handler. - exit_barrier.wait(); - - // The panic handler must have been called twice on every thread. - assert_eq!(n_called.load(Ordering::SeqCst), 2 * n_threads); -} - -#[test] -#[cfg_attr(any(target_os = "emscripten", target_family = "wasm"), ignore)] -fn check_config_build() { - let pool = ThreadPoolBuilder::new().num_threads(22).build().unwrap(); - assert_eq!(pool.current_num_threads(), 22); -} - -/// Helper used by check_error_send_sync to ensure ThreadPoolBuildError is Send + Sync -fn _send_sync() {} - -#[test] -fn check_error_send_sync() { - _send_sync::(); -} - -#[allow(deprecated)] -#[test] -#[cfg_attr(any(target_os = "emscripten", target_family = "wasm"), ignore)] -fn configuration() { - let start_handler = move |_| {}; - let exit_handler = move |_| {}; - let panic_handler = move |_| {}; - let thread_name = move |i| format!("thread_name_{}", i); - - // Ensure we can call all public methods on Configuration - crate::Configuration::new() - .thread_name(thread_name) - .num_threads(5) - .panic_handler(panic_handler) - .stack_size(4e6 as usize) - .breadth_first() - .start_handler(start_handler) - .exit_handler(exit_handler) - .build() - .unwrap(); -} - -#[test] -#[cfg_attr(any(target_os = "emscripten", target_family = "wasm"), ignore)] -fn default_pool() { - ThreadPoolBuilder::default().build().unwrap(); -} - -/// Test that custom spawned threads get their `WorkerThread` cleared once -/// the pool is done with them, allowing them to be used with rayon again -/// later. e.g. WebAssembly want to have their own pool of available threads. -#[test] -#[cfg_attr(any(target_os = "emscripten", target_family = "wasm"), ignore)] -fn cleared_current_thread() -> Result<(), ThreadPoolBuildError> { - let n_threads = 5; - let mut handles = vec![]; - let pool = ThreadPoolBuilder::new() - .num_threads(n_threads) - .spawn_handler(|thread| { - let handle = std::thread::spawn(move || { - thread.run(); - - // Afterward, the current thread shouldn't be set anymore. - assert_eq!(crate::current_thread_index(), None); - }); - handles.push(handle); - Ok(()) - }) - .build()?; - assert_eq!(handles.len(), n_threads); - - pool.install(|| assert!(crate::current_thread_index().is_some())); - drop(pool); - - // Wait for all threads to make their assertions and exit - for handle in handles { - handle.join().unwrap(); - } - - Ok(()) -} diff --git a/compiler/rustc_thread_pool/src/tests.rs b/compiler/rustc_thread_pool/src/tests.rs new file mode 100644 index 00000000000..3082f11a167 --- /dev/null +++ b/compiler/rustc_thread_pool/src/tests.rs @@ -0,0 +1,197 @@ +#![cfg(test)] + +use std::sync::atomic::{AtomicUsize, Ordering}; +use std::sync::{Arc, Barrier}; + +use crate::{ThreadPoolBuildError, ThreadPoolBuilder}; + +#[test] +#[cfg_attr(any(target_os = "emscripten", target_family = "wasm"), ignore)] +fn worker_thread_index() { + let pool = ThreadPoolBuilder::new().num_threads(22).build().unwrap(); + assert_eq!(pool.current_num_threads(), 22); + assert_eq!(pool.current_thread_index(), None); + let index = pool.install(|| pool.current_thread_index().unwrap()); + assert!(index < 22); +} + +#[test] +#[cfg_attr(any(target_os = "emscripten", target_family = "wasm"), ignore)] +fn start_callback_called() { + let n_threads = 16; + let n_called = Arc::new(AtomicUsize::new(0)); + // Wait for all the threads in the pool plus the one running tests. + let barrier = Arc::new(Barrier::new(n_threads + 1)); + + let b = Arc::clone(&barrier); + let nc = Arc::clone(&n_called); + let start_handler = move |_| { + nc.fetch_add(1, Ordering::SeqCst); + b.wait(); + }; + + let conf = ThreadPoolBuilder::new().num_threads(n_threads).start_handler(start_handler); + let _ = conf.build().unwrap(); + + // Wait for all the threads to have been scheduled to run. + barrier.wait(); + + // The handler must have been called on every started thread. + assert_eq!(n_called.load(Ordering::SeqCst), n_threads); +} + +#[test] +#[cfg_attr(any(target_os = "emscripten", target_family = "wasm"), ignore)] +fn exit_callback_called() { + let n_threads = 16; + let n_called = Arc::new(AtomicUsize::new(0)); + // Wait for all the threads in the pool plus the one running tests. + let barrier = Arc::new(Barrier::new(n_threads + 1)); + + let b = Arc::clone(&barrier); + let nc = Arc::clone(&n_called); + let exit_handler = move |_| { + nc.fetch_add(1, Ordering::SeqCst); + b.wait(); + }; + + let conf = ThreadPoolBuilder::new().num_threads(n_threads).exit_handler(exit_handler); + { + let _ = conf.build().unwrap(); + // Drop the pool so it stops the running threads. + } + + // Wait for all the threads to have been scheduled to run. + barrier.wait(); + + // The handler must have been called on every exiting thread. + assert_eq!(n_called.load(Ordering::SeqCst), n_threads); +} + +#[test] +#[cfg_attr(any(target_os = "emscripten", target_family = "wasm"), ignore)] +fn handler_panics_handled_correctly() { + let n_threads = 16; + let n_called = Arc::new(AtomicUsize::new(0)); + // Wait for all the threads in the pool plus the one running tests. + let start_barrier = Arc::new(Barrier::new(n_threads + 1)); + let exit_barrier = Arc::new(Barrier::new(n_threads + 1)); + + let start_handler = move |_| { + panic!("ensure panic handler is called when starting"); + }; + let exit_handler = move |_| { + panic!("ensure panic handler is called when exiting"); + }; + + let sb = Arc::clone(&start_barrier); + let eb = Arc::clone(&exit_barrier); + let nc = Arc::clone(&n_called); + let panic_handler = move |_| { + let val = nc.fetch_add(1, Ordering::SeqCst); + if val < n_threads { + sb.wait(); + } else { + eb.wait(); + } + }; + + let conf = ThreadPoolBuilder::new() + .num_threads(n_threads) + .start_handler(start_handler) + .exit_handler(exit_handler) + .panic_handler(panic_handler); + { + let _ = conf.build().unwrap(); + + // Wait for all the threads to start, panic in the start handler, + // and been taken care of by the panic handler. + start_barrier.wait(); + + // Drop the pool so it stops the running threads. + } + + // Wait for all the threads to exit, panic in the exit handler, + // and been taken care of by the panic handler. + exit_barrier.wait(); + + // The panic handler must have been called twice on every thread. + assert_eq!(n_called.load(Ordering::SeqCst), 2 * n_threads); +} + +#[test] +#[cfg_attr(any(target_os = "emscripten", target_family = "wasm"), ignore)] +fn check_config_build() { + let pool = ThreadPoolBuilder::new().num_threads(22).build().unwrap(); + assert_eq!(pool.current_num_threads(), 22); +} + +/// Helper used by check_error_send_sync to ensure ThreadPoolBuildError is Send + Sync +fn _send_sync() {} + +#[test] +fn check_error_send_sync() { + _send_sync::(); +} + +#[allow(deprecated)] +#[test] +#[cfg_attr(any(target_os = "emscripten", target_family = "wasm"), ignore)] +fn configuration() { + let start_handler = move |_| {}; + let exit_handler = move |_| {}; + let panic_handler = move |_| {}; + let thread_name = move |i| format!("thread_name_{}", i); + + // Ensure we can call all public methods on Configuration + crate::Configuration::new() + .thread_name(thread_name) + .num_threads(5) + .panic_handler(panic_handler) + .stack_size(4e6 as usize) + .breadth_first() + .start_handler(start_handler) + .exit_handler(exit_handler) + .build() + .unwrap(); +} + +#[test] +#[cfg_attr(any(target_os = "emscripten", target_family = "wasm"), ignore)] +fn default_pool() { + ThreadPoolBuilder::default().build().unwrap(); +} + +/// Test that custom spawned threads get their `WorkerThread` cleared once +/// the pool is done with them, allowing them to be used with rayon again +/// later. e.g. WebAssembly want to have their own pool of available threads. +#[test] +#[cfg_attr(any(target_os = "emscripten", target_family = "wasm"), ignore)] +fn cleared_current_thread() -> Result<(), ThreadPoolBuildError> { + let n_threads = 5; + let mut handles = vec![]; + let pool = ThreadPoolBuilder::new() + .num_threads(n_threads) + .spawn_handler(|thread| { + let handle = std::thread::spawn(move || { + thread.run(); + + // Afterward, the current thread shouldn't be set anymore. + assert_eq!(crate::current_thread_index(), None); + }); + handles.push(handle); + Ok(()) + }) + .build()?; + assert_eq!(handles.len(), n_threads); + + pool.install(|| assert!(crate::current_thread_index().is_some())); + drop(pool); + + // Wait for all threads to make their assertions and exit + for handle in handles { + handle.join().unwrap(); + } + + Ok(()) +} diff --git a/compiler/rustc_thread_pool/src/thread_pool/mod.rs b/compiler/rustc_thread_pool/src/thread_pool/mod.rs index 65af6d7106e..ce8783cf0d6 100644 --- a/compiler/rustc_thread_pool/src/thread_pool/mod.rs +++ b/compiler/rustc_thread_pool/src/thread_pool/mod.rs @@ -3,18 +3,17 @@ //! //! [`ThreadPool`]: struct.ThreadPool.html -use crate::broadcast::{self, BroadcastContext}; -use crate::join; -use crate::registry::{Registry, ThreadSpawn, WorkerThread}; -use crate::scope::{do_in_place_scope, do_in_place_scope_fifo}; -use crate::spawn; -use crate::{scope, Scope}; -use crate::{scope_fifo, ScopeFifo}; -use crate::{ThreadPoolBuildError, ThreadPoolBuilder}; use std::error::Error; use std::fmt; use std::sync::Arc; +use crate::broadcast::{self, BroadcastContext}; +use crate::registry::{Registry, ThreadSpawn, WorkerThread}; +use crate::scope::{do_in_place_scope, do_in_place_scope_fifo}; +use crate::{ + Scope, ScopeFifo, ThreadPoolBuildError, ThreadPoolBuilder, join, scope, scope_fifo, spawn, +}; + mod test; /// Represents a user created [thread-pool]. diff --git a/compiler/rustc_thread_pool/src/thread_pool/test.rs b/compiler/rustc_thread_pool/src/thread_pool/test.rs deleted file mode 100644 index 88b36282d48..00000000000 --- a/compiler/rustc_thread_pool/src/thread_pool/test.rs +++ /dev/null @@ -1,418 +0,0 @@ -#![cfg(test)] - -use std::sync::atomic::{AtomicUsize, Ordering}; -use std::sync::mpsc::channel; -use std::sync::{Arc, Mutex}; - -use crate::{join, Scope, ScopeFifo, ThreadPool, ThreadPoolBuilder}; - -#[test] -#[should_panic(expected = "Hello, world!")] -fn panic_propagate() { - let thread_pool = ThreadPoolBuilder::new().build().unwrap(); - thread_pool.install(|| { - panic!("Hello, world!"); - }); -} - -#[test] -#[cfg_attr(any(target_os = "emscripten", target_family = "wasm"), ignore)] -fn workers_stop() { - let registry; - - { - // once we exit this block, thread-pool will be dropped - let thread_pool = ThreadPoolBuilder::new().num_threads(22).build().unwrap(); - registry = thread_pool.install(|| { - // do some work on these threads - join_a_lot(22); - - Arc::clone(&thread_pool.registry) - }); - assert_eq!(registry.num_threads(), 22); - } - - // once thread-pool is dropped, registry should terminate, which - // should lead to worker threads stopping - registry.wait_until_stopped(); -} - -fn join_a_lot(n: usize) { - if n > 0 { - join(|| join_a_lot(n - 1), || join_a_lot(n - 1)); - } -} - -#[test] -#[cfg_attr(any(target_os = "emscripten", target_family = "wasm"), ignore)] -fn sleeper_stop() { - use std::{thread, time}; - - let registry; - - { - // once we exit this block, thread-pool will be dropped - let thread_pool = ThreadPoolBuilder::new().num_threads(22).build().unwrap(); - registry = Arc::clone(&thread_pool.registry); - - // Give time for at least some of the thread pool to fall asleep. - thread::sleep(time::Duration::from_secs(1)); - } - - // once thread-pool is dropped, registry should terminate, which - // should lead to worker threads stopping - registry.wait_until_stopped(); -} - -/// Creates a start/exit handler that increments an atomic counter. -fn count_handler() -> (Arc, impl Fn(usize)) { - let count = Arc::new(AtomicUsize::new(0)); - (Arc::clone(&count), move |_| { - count.fetch_add(1, Ordering::SeqCst); - }) -} - -/// Wait until a counter is no longer shared, then return its value. -fn wait_for_counter(mut counter: Arc) -> usize { - use std::{thread, time}; - - for _ in 0..60 { - counter = match Arc::try_unwrap(counter) { - Ok(counter) => return counter.into_inner(), - Err(counter) => { - thread::sleep(time::Duration::from_secs(1)); - counter - } - }; - } - - // That's too long! - panic!("Counter is still shared!"); -} - -#[test] -#[cfg_attr(any(target_os = "emscripten", target_family = "wasm"), ignore)] -fn failed_thread_stack() { - // Note: we first tried to force failure with a `usize::MAX` stack, but - // macOS and Windows weren't fazed, or at least didn't fail the way we want. - // They work with `isize::MAX`, but 32-bit platforms may feasibly allocate a - // 2GB stack, so it might not fail until the second thread. - let stack_size = ::std::isize::MAX as usize; - - let (start_count, start_handler) = count_handler(); - let (exit_count, exit_handler) = count_handler(); - let builder = ThreadPoolBuilder::new() - .num_threads(10) - .stack_size(stack_size) - .start_handler(start_handler) - .exit_handler(exit_handler); - - let pool = builder.build(); - assert!(pool.is_err(), "thread stack should have failed!"); - - // With such a huge stack, 64-bit will probably fail on the first thread; - // 32-bit might manage the first 2GB, but certainly fail the second. - let start_count = wait_for_counter(start_count); - assert!(start_count <= 1); - assert_eq!(start_count, wait_for_counter(exit_count)); -} - -#[test] -#[cfg_attr(not(panic = "unwind"), ignore)] -fn panic_thread_name() { - let (start_count, start_handler) = count_handler(); - let (exit_count, exit_handler) = count_handler(); - let builder = ThreadPoolBuilder::new() - .num_threads(10) - .start_handler(start_handler) - .exit_handler(exit_handler) - .thread_name(|i| { - if i >= 5 { - panic!(); - } - format!("panic_thread_name#{}", i) - }); - - let pool = crate::unwind::halt_unwinding(|| builder.build()); - assert!(pool.is_err(), "thread-name panic should propagate!"); - - // Assuming they're created in order, threads 0 through 4 should have - // been started already, and then terminated by the panic. - assert_eq!(5, wait_for_counter(start_count)); - assert_eq!(5, wait_for_counter(exit_count)); -} - -#[test] -#[cfg_attr(any(target_os = "emscripten", target_family = "wasm"), ignore)] -fn self_install() { - let pool = ThreadPoolBuilder::new().num_threads(1).build().unwrap(); - - // If the inner `install` blocks, then nothing will actually run it! - assert!(pool.install(|| pool.install(|| true))); -} - -#[test] -#[cfg_attr(any(target_os = "emscripten", target_family = "wasm"), ignore)] -fn mutual_install() { - let pool1 = ThreadPoolBuilder::new().num_threads(1).build().unwrap(); - let pool2 = ThreadPoolBuilder::new().num_threads(1).build().unwrap(); - - let ok = pool1.install(|| { - // This creates a dependency from `pool1` -> `pool2` - pool2.install(|| { - // This creates a dependency from `pool2` -> `pool1` - pool1.install(|| { - // If they blocked on inter-pool installs, there would be no - // threads left to run this! - true - }) - }) - }); - assert!(ok); -} - -#[test] -#[cfg_attr(any(target_os = "emscripten", target_family = "wasm"), ignore)] -fn mutual_install_sleepy() { - use std::{thread, time}; - - let pool1 = ThreadPoolBuilder::new().num_threads(1).build().unwrap(); - let pool2 = ThreadPoolBuilder::new().num_threads(1).build().unwrap(); - - let ok = pool1.install(|| { - // This creates a dependency from `pool1` -> `pool2` - pool2.install(|| { - // Give `pool1` time to fall asleep. - thread::sleep(time::Duration::from_secs(1)); - - // This creates a dependency from `pool2` -> `pool1` - pool1.install(|| { - // Give `pool2` time to fall asleep. - thread::sleep(time::Duration::from_secs(1)); - - // If they blocked on inter-pool installs, there would be no - // threads left to run this! - true - }) - }) - }); - assert!(ok); -} - -#[test] -#[allow(deprecated)] -#[cfg_attr(any(target_os = "emscripten", target_family = "wasm"), ignore)] -fn check_thread_pool_new() { - let pool = ThreadPool::new(crate::Configuration::new().num_threads(22)).unwrap(); - assert_eq!(pool.current_num_threads(), 22); -} - -macro_rules! test_scope_order { - ($scope:ident => $spawn:ident) => {{ - let builder = ThreadPoolBuilder::new().num_threads(1); - let pool = builder.build().unwrap(); - pool.install(|| { - let vec = Mutex::new(vec![]); - pool.$scope(|scope| { - let vec = &vec; - for i in 0..10 { - scope.$spawn(move |_| { - vec.lock().unwrap().push(i); - }); - } - }); - vec.into_inner().unwrap() - }) - }}; -} - -#[test] -#[cfg_attr(any(target_os = "emscripten", target_family = "wasm"), ignore)] -fn scope_lifo_order() { - let vec = test_scope_order!(scope => spawn); - let expected: Vec = (0..10).rev().collect(); // LIFO -> reversed - assert_eq!(vec, expected); -} - -#[test] -#[cfg_attr(any(target_os = "emscripten", target_family = "wasm"), ignore)] -fn scope_fifo_order() { - let vec = test_scope_order!(scope_fifo => spawn_fifo); - let expected: Vec = (0..10).collect(); // FIFO -> natural order - assert_eq!(vec, expected); -} - -macro_rules! test_spawn_order { - ($spawn:ident) => {{ - let builder = ThreadPoolBuilder::new().num_threads(1); - let pool = &builder.build().unwrap(); - let (tx, rx) = channel(); - pool.install(move || { - for i in 0..10 { - let tx = tx.clone(); - pool.$spawn(move || { - tx.send(i).unwrap(); - }); - } - }); - rx.iter().collect::>() - }}; -} - -#[test] -#[cfg_attr(any(target_os = "emscripten", target_family = "wasm"), ignore)] -fn spawn_lifo_order() { - let vec = test_spawn_order!(spawn); - let expected: Vec = (0..10).rev().collect(); // LIFO -> reversed - assert_eq!(vec, expected); -} - -#[test] -#[cfg_attr(any(target_os = "emscripten", target_family = "wasm"), ignore)] -fn spawn_fifo_order() { - let vec = test_spawn_order!(spawn_fifo); - let expected: Vec = (0..10).collect(); // FIFO -> natural order - assert_eq!(vec, expected); -} - -#[test] -#[cfg_attr(any(target_os = "emscripten", target_family = "wasm"), ignore)] -fn nested_scopes() { - // Create matching scopes for every thread pool. - fn nest<'scope, OP>(pools: &[ThreadPool], scopes: Vec<&Scope<'scope>>, op: OP) - where - OP: FnOnce(&[&Scope<'scope>]) + Send, - { - if let Some((pool, tail)) = pools.split_first() { - pool.scope(move |s| { - // This move reduces the reference lifetimes by variance to match s, - // but the actual scopes are still tied to the invariant 'scope. - let mut scopes = scopes; - scopes.push(s); - nest(tail, scopes, op) - }) - } else { - (op)(&scopes) - } - } - - let pools: Vec<_> = (0..10) - .map(|_| ThreadPoolBuilder::new().num_threads(1).build().unwrap()) - .collect(); - - let counter = AtomicUsize::new(0); - nest(&pools, vec![], |scopes| { - for &s in scopes { - s.spawn(|_| { - // Our 'scope lets us borrow the counter in every pool. - counter.fetch_add(1, Ordering::Relaxed); - }); - } - }); - assert_eq!(counter.into_inner(), pools.len()); -} - -#[test] -#[cfg_attr(any(target_os = "emscripten", target_family = "wasm"), ignore)] -fn nested_fifo_scopes() { - // Create matching fifo scopes for every thread pool. - fn nest<'scope, OP>(pools: &[ThreadPool], scopes: Vec<&ScopeFifo<'scope>>, op: OP) - where - OP: FnOnce(&[&ScopeFifo<'scope>]) + Send, - { - if let Some((pool, tail)) = pools.split_first() { - pool.scope_fifo(move |s| { - // This move reduces the reference lifetimes by variance to match s, - // but the actual scopes are still tied to the invariant 'scope. - let mut scopes = scopes; - scopes.push(s); - nest(tail, scopes, op) - }) - } else { - (op)(&scopes) - } - } - - let pools: Vec<_> = (0..10) - .map(|_| ThreadPoolBuilder::new().num_threads(1).build().unwrap()) - .collect(); - - let counter = AtomicUsize::new(0); - nest(&pools, vec![], |scopes| { - for &s in scopes { - s.spawn_fifo(|_| { - // Our 'scope lets us borrow the counter in every pool. - counter.fetch_add(1, Ordering::Relaxed); - }); - } - }); - assert_eq!(counter.into_inner(), pools.len()); -} - -#[test] -#[cfg_attr(any(target_os = "emscripten", target_family = "wasm"), ignore)] -fn in_place_scope_no_deadlock() { - let pool = ThreadPoolBuilder::new().num_threads(1).build().unwrap(); - let (tx, rx) = channel(); - let rx_ref = ℞ - pool.in_place_scope(move |s| { - // With regular scopes this closure would never run because this scope op - // itself would block the only worker thread. - s.spawn(move |_| { - tx.send(()).unwrap(); - }); - rx_ref.recv().unwrap(); - }); -} - -#[test] -#[cfg_attr(any(target_os = "emscripten", target_family = "wasm"), ignore)] -fn in_place_scope_fifo_no_deadlock() { - let pool = ThreadPoolBuilder::new().num_threads(1).build().unwrap(); - let (tx, rx) = channel(); - let rx_ref = ℞ - pool.in_place_scope_fifo(move |s| { - // With regular scopes this closure would never run because this scope op - // itself would block the only worker thread. - s.spawn_fifo(move |_| { - tx.send(()).unwrap(); - }); - rx_ref.recv().unwrap(); - }); -} - -#[test] -fn yield_now_to_spawn() { - let (tx, rx) = channel(); - - // Queue a regular spawn. - crate::spawn(move || tx.send(22).unwrap()); - - // The single-threaded fallback mode (for wasm etc.) won't - // get a chance to run the spawn if we never yield to it. - crate::registry::in_worker(move |_, _| { - crate::yield_now(); - }); - - // The spawn **must** have started by now, but we still might have to wait - // for it to finish if a different thread stole it first. - assert_eq!(22, rx.recv().unwrap()); -} - -#[test] -fn yield_local_to_spawn() { - let (tx, rx) = channel(); - - // Queue a regular spawn. - crate::spawn(move || tx.send(22).unwrap()); - - // The single-threaded fallback mode (for wasm etc.) won't - // get a chance to run the spawn if we never yield to it. - crate::registry::in_worker(move |_, _| { - crate::yield_local(); - }); - - // The spawn **must** have started by now, but we still might have to wait - // for it to finish if a different thread stole it first. - assert_eq!(22, rx.recv().unwrap()); -} diff --git a/compiler/rustc_thread_pool/src/thread_pool/tests.rs b/compiler/rustc_thread_pool/src/thread_pool/tests.rs new file mode 100644 index 00000000000..42c99565088 --- /dev/null +++ b/compiler/rustc_thread_pool/src/thread_pool/tests.rs @@ -0,0 +1,416 @@ +#![cfg(test)] + +use std::sync::atomic::{AtomicUsize, Ordering}; +use std::sync::mpsc::channel; +use std::sync::{Arc, Mutex}; + +use crate::{Scope, ScopeFifo, ThreadPool, ThreadPoolBuilder, join}; + +#[test] +#[should_panic(expected = "Hello, world!")] +fn panic_propagate() { + let thread_pool = ThreadPoolBuilder::new().build().unwrap(); + thread_pool.install(|| { + panic!("Hello, world!"); + }); +} + +#[test] +#[cfg_attr(any(target_os = "emscripten", target_family = "wasm"), ignore)] +fn workers_stop() { + let registry; + + { + // once we exit this block, thread-pool will be dropped + let thread_pool = ThreadPoolBuilder::new().num_threads(22).build().unwrap(); + registry = thread_pool.install(|| { + // do some work on these threads + join_a_lot(22); + + Arc::clone(&thread_pool.registry) + }); + assert_eq!(registry.num_threads(), 22); + } + + // once thread-pool is dropped, registry should terminate, which + // should lead to worker threads stopping + registry.wait_until_stopped(); +} + +fn join_a_lot(n: usize) { + if n > 0 { + join(|| join_a_lot(n - 1), || join_a_lot(n - 1)); + } +} + +#[test] +#[cfg_attr(any(target_os = "emscripten", target_family = "wasm"), ignore)] +fn sleeper_stop() { + use std::{thread, time}; + + let registry; + + { + // once we exit this block, thread-pool will be dropped + let thread_pool = ThreadPoolBuilder::new().num_threads(22).build().unwrap(); + registry = Arc::clone(&thread_pool.registry); + + // Give time for at least some of the thread pool to fall asleep. + thread::sleep(time::Duration::from_secs(1)); + } + + // once thread-pool is dropped, registry should terminate, which + // should lead to worker threads stopping + registry.wait_until_stopped(); +} + +/// Creates a start/exit handler that increments an atomic counter. +fn count_handler() -> (Arc, impl Fn(usize)) { + let count = Arc::new(AtomicUsize::new(0)); + (Arc::clone(&count), move |_| { + count.fetch_add(1, Ordering::SeqCst); + }) +} + +/// Wait until a counter is no longer shared, then return its value. +fn wait_for_counter(mut counter: Arc) -> usize { + use std::{thread, time}; + + for _ in 0..60 { + counter = match Arc::try_unwrap(counter) { + Ok(counter) => return counter.into_inner(), + Err(counter) => { + thread::sleep(time::Duration::from_secs(1)); + counter + } + }; + } + + // That's too long! + panic!("Counter is still shared!"); +} + +#[test] +#[cfg_attr(any(target_os = "emscripten", target_family = "wasm"), ignore)] +fn failed_thread_stack() { + // Note: we first tried to force failure with a `usize::MAX` stack, but + // macOS and Windows weren't fazed, or at least didn't fail the way we want. + // They work with `isize::MAX`, but 32-bit platforms may feasibly allocate a + // 2GB stack, so it might not fail until the second thread. + let stack_size = ::std::isize::MAX as usize; + + let (start_count, start_handler) = count_handler(); + let (exit_count, exit_handler) = count_handler(); + let builder = ThreadPoolBuilder::new() + .num_threads(10) + .stack_size(stack_size) + .start_handler(start_handler) + .exit_handler(exit_handler); + + let pool = builder.build(); + assert!(pool.is_err(), "thread stack should have failed!"); + + // With such a huge stack, 64-bit will probably fail on the first thread; + // 32-bit might manage the first 2GB, but certainly fail the second. + let start_count = wait_for_counter(start_count); + assert!(start_count <= 1); + assert_eq!(start_count, wait_for_counter(exit_count)); +} + +#[test] +#[cfg_attr(not(panic = "unwind"), ignore)] +fn panic_thread_name() { + let (start_count, start_handler) = count_handler(); + let (exit_count, exit_handler) = count_handler(); + let builder = ThreadPoolBuilder::new() + .num_threads(10) + .start_handler(start_handler) + .exit_handler(exit_handler) + .thread_name(|i| { + if i >= 5 { + panic!(); + } + format!("panic_thread_name#{}", i) + }); + + let pool = crate::unwind::halt_unwinding(|| builder.build()); + assert!(pool.is_err(), "thread-name panic should propagate!"); + + // Assuming they're created in order, threads 0 through 4 should have + // been started already, and then terminated by the panic. + assert_eq!(5, wait_for_counter(start_count)); + assert_eq!(5, wait_for_counter(exit_count)); +} + +#[test] +#[cfg_attr(any(target_os = "emscripten", target_family = "wasm"), ignore)] +fn self_install() { + let pool = ThreadPoolBuilder::new().num_threads(1).build().unwrap(); + + // If the inner `install` blocks, then nothing will actually run it! + assert!(pool.install(|| pool.install(|| true))); +} + +#[test] +#[cfg_attr(any(target_os = "emscripten", target_family = "wasm"), ignore)] +fn mutual_install() { + let pool1 = ThreadPoolBuilder::new().num_threads(1).build().unwrap(); + let pool2 = ThreadPoolBuilder::new().num_threads(1).build().unwrap(); + + let ok = pool1.install(|| { + // This creates a dependency from `pool1` -> `pool2` + pool2.install(|| { + // This creates a dependency from `pool2` -> `pool1` + pool1.install(|| { + // If they blocked on inter-pool installs, there would be no + // threads left to run this! + true + }) + }) + }); + assert!(ok); +} + +#[test] +#[cfg_attr(any(target_os = "emscripten", target_family = "wasm"), ignore)] +fn mutual_install_sleepy() { + use std::{thread, time}; + + let pool1 = ThreadPoolBuilder::new().num_threads(1).build().unwrap(); + let pool2 = ThreadPoolBuilder::new().num_threads(1).build().unwrap(); + + let ok = pool1.install(|| { + // This creates a dependency from `pool1` -> `pool2` + pool2.install(|| { + // Give `pool1` time to fall asleep. + thread::sleep(time::Duration::from_secs(1)); + + // This creates a dependency from `pool2` -> `pool1` + pool1.install(|| { + // Give `pool2` time to fall asleep. + thread::sleep(time::Duration::from_secs(1)); + + // If they blocked on inter-pool installs, there would be no + // threads left to run this! + true + }) + }) + }); + assert!(ok); +} + +#[test] +#[allow(deprecated)] +#[cfg_attr(any(target_os = "emscripten", target_family = "wasm"), ignore)] +fn check_thread_pool_new() { + let pool = ThreadPool::new(crate::Configuration::new().num_threads(22)).unwrap(); + assert_eq!(pool.current_num_threads(), 22); +} + +macro_rules! test_scope_order { + ($scope:ident => $spawn:ident) => {{ + let builder = ThreadPoolBuilder::new().num_threads(1); + let pool = builder.build().unwrap(); + pool.install(|| { + let vec = Mutex::new(vec![]); + pool.$scope(|scope| { + let vec = &vec; + for i in 0..10 { + scope.$spawn(move |_| { + vec.lock().unwrap().push(i); + }); + } + }); + vec.into_inner().unwrap() + }) + }}; +} + +#[test] +#[cfg_attr(any(target_os = "emscripten", target_family = "wasm"), ignore)] +fn scope_lifo_order() { + let vec = test_scope_order!(scope => spawn); + let expected: Vec = (0..10).rev().collect(); // LIFO -> reversed + assert_eq!(vec, expected); +} + +#[test] +#[cfg_attr(any(target_os = "emscripten", target_family = "wasm"), ignore)] +fn scope_fifo_order() { + let vec = test_scope_order!(scope_fifo => spawn_fifo); + let expected: Vec = (0..10).collect(); // FIFO -> natural order + assert_eq!(vec, expected); +} + +macro_rules! test_spawn_order { + ($spawn:ident) => {{ + let builder = ThreadPoolBuilder::new().num_threads(1); + let pool = &builder.build().unwrap(); + let (tx, rx) = channel(); + pool.install(move || { + for i in 0..10 { + let tx = tx.clone(); + pool.$spawn(move || { + tx.send(i).unwrap(); + }); + } + }); + rx.iter().collect::>() + }}; +} + +#[test] +#[cfg_attr(any(target_os = "emscripten", target_family = "wasm"), ignore)] +fn spawn_lifo_order() { + let vec = test_spawn_order!(spawn); + let expected: Vec = (0..10).rev().collect(); // LIFO -> reversed + assert_eq!(vec, expected); +} + +#[test] +#[cfg_attr(any(target_os = "emscripten", target_family = "wasm"), ignore)] +fn spawn_fifo_order() { + let vec = test_spawn_order!(spawn_fifo); + let expected: Vec = (0..10).collect(); // FIFO -> natural order + assert_eq!(vec, expected); +} + +#[test] +#[cfg_attr(any(target_os = "emscripten", target_family = "wasm"), ignore)] +fn nested_scopes() { + // Create matching scopes for every thread pool. + fn nest<'scope, OP>(pools: &[ThreadPool], scopes: Vec<&Scope<'scope>>, op: OP) + where + OP: FnOnce(&[&Scope<'scope>]) + Send, + { + if let Some((pool, tail)) = pools.split_first() { + pool.scope(move |s| { + // This move reduces the reference lifetimes by variance to match s, + // but the actual scopes are still tied to the invariant 'scope. + let mut scopes = scopes; + scopes.push(s); + nest(tail, scopes, op) + }) + } else { + (op)(&scopes) + } + } + + let pools: Vec<_> = + (0..10).map(|_| ThreadPoolBuilder::new().num_threads(1).build().unwrap()).collect(); + + let counter = AtomicUsize::new(0); + nest(&pools, vec![], |scopes| { + for &s in scopes { + s.spawn(|_| { + // Our 'scope lets us borrow the counter in every pool. + counter.fetch_add(1, Ordering::Relaxed); + }); + } + }); + assert_eq!(counter.into_inner(), pools.len()); +} + +#[test] +#[cfg_attr(any(target_os = "emscripten", target_family = "wasm"), ignore)] +fn nested_fifo_scopes() { + // Create matching fifo scopes for every thread pool. + fn nest<'scope, OP>(pools: &[ThreadPool], scopes: Vec<&ScopeFifo<'scope>>, op: OP) + where + OP: FnOnce(&[&ScopeFifo<'scope>]) + Send, + { + if let Some((pool, tail)) = pools.split_first() { + pool.scope_fifo(move |s| { + // This move reduces the reference lifetimes by variance to match s, + // but the actual scopes are still tied to the invariant 'scope. + let mut scopes = scopes; + scopes.push(s); + nest(tail, scopes, op) + }) + } else { + (op)(&scopes) + } + } + + let pools: Vec<_> = + (0..10).map(|_| ThreadPoolBuilder::new().num_threads(1).build().unwrap()).collect(); + + let counter = AtomicUsize::new(0); + nest(&pools, vec![], |scopes| { + for &s in scopes { + s.spawn_fifo(|_| { + // Our 'scope lets us borrow the counter in every pool. + counter.fetch_add(1, Ordering::Relaxed); + }); + } + }); + assert_eq!(counter.into_inner(), pools.len()); +} + +#[test] +#[cfg_attr(any(target_os = "emscripten", target_family = "wasm"), ignore)] +fn in_place_scope_no_deadlock() { + let pool = ThreadPoolBuilder::new().num_threads(1).build().unwrap(); + let (tx, rx) = channel(); + let rx_ref = ℞ + pool.in_place_scope(move |s| { + // With regular scopes this closure would never run because this scope op + // itself would block the only worker thread. + s.spawn(move |_| { + tx.send(()).unwrap(); + }); + rx_ref.recv().unwrap(); + }); +} + +#[test] +#[cfg_attr(any(target_os = "emscripten", target_family = "wasm"), ignore)] +fn in_place_scope_fifo_no_deadlock() { + let pool = ThreadPoolBuilder::new().num_threads(1).build().unwrap(); + let (tx, rx) = channel(); + let rx_ref = ℞ + pool.in_place_scope_fifo(move |s| { + // With regular scopes this closure would never run because this scope op + // itself would block the only worker thread. + s.spawn_fifo(move |_| { + tx.send(()).unwrap(); + }); + rx_ref.recv().unwrap(); + }); +} + +#[test] +fn yield_now_to_spawn() { + let (tx, rx) = channel(); + + // Queue a regular spawn. + crate::spawn(move || tx.send(22).unwrap()); + + // The single-threaded fallback mode (for wasm etc.) won't + // get a chance to run the spawn if we never yield to it. + crate::registry::in_worker(move |_, _| { + crate::yield_now(); + }); + + // The spawn **must** have started by now, but we still might have to wait + // for it to finish if a different thread stole it first. + assert_eq!(22, rx.recv().unwrap()); +} + +#[test] +fn yield_local_to_spawn() { + let (tx, rx) = channel(); + + // Queue a regular spawn. + crate::spawn(move || tx.send(22).unwrap()); + + // The single-threaded fallback mode (for wasm etc.) won't + // get a chance to run the spawn if we never yield to it. + crate::registry::in_worker(move |_, _| { + crate::yield_local(); + }); + + // The spawn **must** have started by now, but we still might have to wait + // for it to finish if a different thread stole it first. + assert_eq!(22, rx.recv().unwrap()); +} diff --git a/compiler/rustc_thread_pool/src/tlv.rs b/compiler/rustc_thread_pool/src/tlv.rs index ce22f7aa0ce..b5f63479e2f 100644 --- a/compiler/rustc_thread_pool/src/tlv.rs +++ b/compiler/rustc_thread_pool/src/tlv.rs @@ -1,7 +1,8 @@ //! Allows access to the Rayon's thread local value //! which is preserved when moving jobs across threads -use std::{cell::Cell, ptr}; +use std::cell::Cell; +use std::ptr; thread_local!(pub static TLV: Cell<*const ()> = const { Cell::new(ptr::null()) }); diff --git a/compiler/rustc_thread_pool/src/worker_local.rs b/compiler/rustc_thread_pool/src/worker_local.rs index 85d51687c19..d108c91f9ee 100644 --- a/compiler/rustc_thread_pool/src/worker_local.rs +++ b/compiler/rustc_thread_pool/src/worker_local.rs @@ -1,8 +1,9 @@ -use crate::registry::{Registry, WorkerThread}; use std::fmt; use std::ops::Deref; use std::sync::Arc; +use crate::registry::{Registry, WorkerThread}; + #[repr(align(64))] #[derive(Debug)] struct CacheAligned(T); @@ -27,9 +28,7 @@ impl WorkerLocal { pub fn new T>(mut initial: F) -> WorkerLocal { let registry = Registry::current(); WorkerLocal { - locals: (0..registry.num_threads()) - .map(|i| CacheAligned(initial(i))) - .collect(), + locals: (0..registry.num_threads()).map(|i| CacheAligned(initial(i))).collect(), registry, } } @@ -62,9 +61,7 @@ impl WorkerLocal> { impl fmt::Debug for WorkerLocal { fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { - f.debug_struct("WorkerLocal") - .field("registry", &self.registry.id()) - .finish() + f.debug_struct("WorkerLocal").field("registry", &self.registry.id()).finish() } } diff --git a/compiler/rustc_thread_pool/tests/double_init_fail.rs b/compiler/rustc_thread_pool/tests/double_init_fail.rs index 15915304ddc..71ed425bb32 100644 --- a/compiler/rustc_thread_pool/tests/double_init_fail.rs +++ b/compiler/rustc_thread_pool/tests/double_init_fail.rs @@ -1,6 +1,7 @@ -use rayon_core::ThreadPoolBuilder; use std::error::Error; +use rayon_core::ThreadPoolBuilder; + #[test] #[cfg_attr(any(target_os = "emscripten", target_family = "wasm"), ignore)] fn double_init_fail() { @@ -8,8 +9,5 @@ fn double_init_fail() { assert!(result1.is_ok()); let err = ThreadPoolBuilder::new().build_global().unwrap_err(); assert!(err.source().is_none()); - assert_eq!( - err.to_string(), - "The global thread pool has already been initialized.", - ); + assert_eq!(err.to_string(), "The global thread pool has already been initialized.",); } diff --git a/compiler/rustc_thread_pool/tests/init_zero_threads.rs b/compiler/rustc_thread_pool/tests/init_zero_threads.rs index 3c1ad251c7e..c1770e57f3c 100644 --- a/compiler/rustc_thread_pool/tests/init_zero_threads.rs +++ b/compiler/rustc_thread_pool/tests/init_zero_threads.rs @@ -3,8 +3,5 @@ use rayon_core::ThreadPoolBuilder; #[test] #[cfg_attr(any(target_os = "emscripten", target_family = "wasm"), ignore)] fn init_zero_threads() { - ThreadPoolBuilder::new() - .num_threads(0) - .build_global() - .unwrap(); + ThreadPoolBuilder::new().num_threads(0).build_global().unwrap(); } diff --git a/compiler/rustc_thread_pool/tests/scoped_threadpool.rs b/compiler/rustc_thread_pool/tests/scoped_threadpool.rs index 932147179f5..8cc2c859c0c 100644 --- a/compiler/rustc_thread_pool/tests/scoped_threadpool.rs +++ b/compiler/rustc_thread_pool/tests/scoped_threadpool.rs @@ -10,9 +10,7 @@ scoped_tls::scoped_thread_local!(static LOCAL: Local); #[cfg_attr(any(target_os = "emscripten", target_family = "wasm"), ignore)] fn missing_scoped_tls() { LOCAL.set(&Local(42), || { - let pool = ThreadPoolBuilder::new() - .build() - .expect("thread pool created"); + let pool = ThreadPoolBuilder::new().build().expect("thread pool created"); // `LOCAL` is not set in the pool. pool.install(|| { diff --git a/compiler/rustc_thread_pool/tests/stack_overflow_crash.rs b/compiler/rustc_thread_pool/tests/stack_overflow_crash.rs index a6494069212..c7a880de8bb 100644 --- a/compiler/rustc_thread_pool/tests/stack_overflow_crash.rs +++ b/compiler/rustc_thread_pool/tests/stack_overflow_crash.rs @@ -1,10 +1,9 @@ -use rayon_core::ThreadPoolBuilder; - use std::env; -use std::process::{Command, ExitStatus, Stdio}; - #[cfg(target_os = "linux")] use std::os::unix::process::ExitStatusExt; +use std::process::{Command, ExitStatus, Stdio}; + +use rayon_core::ThreadPoolBuilder; fn force_stack_overflow(depth: u32) { let mut buffer = [0u8; 1024 * 1024]; @@ -18,13 +17,7 @@ fn force_stack_overflow(depth: u32) { #[cfg(unix)] fn disable_core() { unsafe { - libc::setrlimit( - libc::RLIMIT_CORE, - &libc::rlimit { - rlim_cur: 0, - rlim_max: 0, - }, - ); + libc::setrlimit(libc::RLIMIT_CORE, &libc::rlimit { rlim_cur: 0, rlim_max: 0 }); } } @@ -50,10 +43,7 @@ fn stack_overflow_crash() { #[cfg(any(unix, windows))] assert_eq!(status.code(), overflow_code()); #[cfg(target_os = "linux")] - assert!(matches!( - status.signal(), - Some(libc::SIGABRT | libc::SIGSEGV) - )); + assert!(matches!(status.signal(), Some(libc::SIGABRT | libc::SIGSEGV))); // Now run with a larger stack and verify correct operation. let status = run_ignored("run_with_large_stack"); @@ -86,10 +76,7 @@ fn run_with_large_stack() { } fn run_with_stack(stack_size_in_mb: usize) { - let pool = ThreadPoolBuilder::new() - .stack_size(stack_size_in_mb * 1024 * 1024) - .build() - .unwrap(); + let pool = ThreadPoolBuilder::new().stack_size(stack_size_in_mb * 1024 * 1024).build().unwrap(); pool.install(|| { #[cfg(unix)] disable_core(); -- cgit 1.4.1-3-g733a5 From 4aa62ea9e9015621969a0f505abf7a6894e99e9e Mon Sep 17 00:00:00 2001 From: "Celina G. Val" Date: Wed, 11 Jun 2025 11:45:06 -0700 Subject: Use `rustc_thread_pool` instead of `rustc-rayon-core` --- Cargo.lock | 39 ++++++++++++++-------- compiler/rustc_data_structures/Cargo.toml | 2 +- compiler/rustc_data_structures/src/sync.rs | 2 +- .../rustc_data_structures/src/sync/parallel.rs | 12 +++---- compiler/rustc_interface/Cargo.toml | 2 +- compiler/rustc_interface/src/util.rs | 8 ++--- compiler/rustc_middle/Cargo.toml | 2 +- compiler/rustc_middle/src/ty/context/tls.rs | 2 +- compiler/rustc_query_system/Cargo.toml | 2 +- compiler/rustc_query_system/src/query/job.rs | 10 +++--- compiler/rustc_thread_pool/Cargo.toml | 14 +++----- .../src/compile_fail/quicksort_race1.rs | 2 +- .../src/compile_fail/quicksort_race2.rs | 2 +- .../src/compile_fail/quicksort_race3.rs | 2 +- .../src/compile_fail/rc_return.rs | 4 +-- .../rustc_thread_pool/src/compile_fail/rc_upvar.rs | 2 +- .../src/compile_fail/scope_join_bad.rs | 4 +-- compiler/rustc_thread_pool/src/join/mod.rs | 2 +- compiler/rustc_thread_pool/src/lib.rs | 12 +++---- compiler/rustc_thread_pool/src/scope/mod.rs | 16 ++++----- compiler/rustc_thread_pool/src/spawn/mod.rs | 2 +- compiler/rustc_thread_pool/src/thread_pool/mod.rs | 10 +++--- .../rustc_thread_pool/tests/double_init_fail.rs | 2 +- .../rustc_thread_pool/tests/init_zero_threads.rs | 2 +- compiler/rustc_thread_pool/tests/scope_join.rs | 2 +- .../rustc_thread_pool/tests/scoped_threadpool.rs | 2 +- compiler/rustc_thread_pool/tests/simple_panic.rs | 2 +- .../tests/stack_overflow_crash.rs | 2 +- src/tools/tidy/src/deps.rs | 2 +- 29 files changed, 86 insertions(+), 81 deletions(-) (limited to 'compiler/rustc_thread_pool/src/join/mod.rs') diff --git a/Cargo.lock b/Cargo.lock index 93abab8469a..ba3a10bad95 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -2977,6 +2977,15 @@ dependencies = [ "getrandom 0.3.3", ] +[[package]] +name = "rand_xorshift" +version = "0.4.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "513962919efc330f829edb2535844d1b912b0fbe2ca165d613e4e8788bb05a5a" +dependencies = [ + "rand_core 0.9.3", +] + [[package]] name = "rand_xoshiro" version = "0.7.0" @@ -3176,16 +3185,6 @@ dependencies = [ "tikv-jemalloc-sys", ] -[[package]] -name = "rustc-rayon-core" -version = "0.5.1" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "2f42932dcd3bcbe484b38a3ccf79b7906fac41c02d408b5b1bac26da3416efdb" -dependencies = [ - "crossbeam-deque", - "crossbeam-utils", -] - [[package]] name = "rustc-semver" version = "1.1.0" @@ -3554,7 +3553,6 @@ dependencies = [ "parking_lot", "portable-atomic", "rustc-hash 2.1.1", - "rustc-rayon-core", "rustc-stable-hash", "rustc_arena", "rustc_graphviz", @@ -3562,6 +3560,7 @@ dependencies = [ "rustc_index", "rustc_macros", "rustc_serialize", + "rustc_thread_pool", "smallvec", "stacker", "tempfile", @@ -3908,7 +3907,6 @@ dependencies = [ name = "rustc_interface" version = "0.0.0" dependencies = [ - "rustc-rayon-core", "rustc_abi", "rustc_ast", "rustc_ast_lowering", @@ -3947,6 +3945,7 @@ dependencies = [ "rustc_span", "rustc_symbol_mangling", "rustc_target", + "rustc_thread_pool", "rustc_trait_selection", "rustc_traits", "rustc_ty_utils", @@ -4074,7 +4073,6 @@ dependencies = [ "either", "gsgdt", "polonius-engine", - "rustc-rayon-core", "rustc_abi", "rustc_apfloat", "rustc_arena", @@ -4098,6 +4096,7 @@ dependencies = [ "rustc_session", "rustc_span", "rustc_target", + "rustc_thread_pool", "rustc_type_ir", "smallvec", "thin-vec", @@ -4344,7 +4343,6 @@ version = "0.0.0" dependencies = [ "hashbrown", "parking_lot", - "rustc-rayon-core", "rustc_abi", "rustc_ast", "rustc_attr_data_structures", @@ -4359,6 +4357,7 @@ dependencies = [ "rustc_serialize", "rustc_session", "rustc_span", + "rustc_thread_pool", "smallvec", "tracing", ] @@ -4520,6 +4519,18 @@ dependencies = [ "tracing", ] +[[package]] +name = "rustc_thread_pool" +version = "0.0.0" +dependencies = [ + "crossbeam-deque", + "crossbeam-utils", + "libc", + "rand 0.9.1", + "rand_xorshift", + "scoped-tls", +] + [[package]] name = "rustc_tools_util" version = "0.4.2" diff --git a/compiler/rustc_data_structures/Cargo.toml b/compiler/rustc_data_structures/Cargo.toml index f6a02011618..17204883fb0 100644 --- a/compiler/rustc_data_structures/Cargo.toml +++ b/compiler/rustc_data_structures/Cargo.toml @@ -14,7 +14,6 @@ indexmap = "2.4.0" jobserver_crate = { version = "0.1.28", package = "jobserver" } measureme = "12.0.1" rustc-hash = "2.0.0" -rustc-rayon-core = { version = "0.5.0" } rustc-stable-hash = { version = "0.1.0", features = ["nightly"] } rustc_arena = { path = "../rustc_arena" } rustc_graphviz = { path = "../rustc_graphviz" } @@ -22,6 +21,7 @@ rustc_hashes = { path = "../rustc_hashes" } rustc_index = { path = "../rustc_index", package = "rustc_index" } rustc_macros = { path = "../rustc_macros" } rustc_serialize = { path = "../rustc_serialize" } +rustc_thread_pool = { path = "../rustc_thread_pool" } smallvec = { version = "1.8.1", features = ["const_generics", "union", "may_dangle"] } stacker = "0.1.17" tempfile = "3.2" diff --git a/compiler/rustc_data_structures/src/sync.rs b/compiler/rustc_data_structures/src/sync.rs index b28c333d860..68527dde1f3 100644 --- a/compiler/rustc_data_structures/src/sync.rs +++ b/compiler/rustc_data_structures/src/sync.rs @@ -23,7 +23,7 @@ //! | `RwLock` | `RefCell` | `parking_lot::RwLock` | //! | `MTLock` [^1] | `T` | `Lock` | //! | | | | -//! | `ParallelIterator` | `Iterator` | `rayon::iter::ParallelIterator` | +//! | `ParallelIterator` | `Iterator` | `rustc_thread_pool::iter::ParallelIterator` | //! //! [^1]: `MTLock` is similar to `Lock`, but the serial version avoids the cost //! of a `RefCell`. This is appropriate when interior mutability is not diff --git a/compiler/rustc_data_structures/src/sync/parallel.rs b/compiler/rustc_data_structures/src/sync/parallel.rs index ab65c7f3a6b..ff4b60a1031 100644 --- a/compiler/rustc_data_structures/src/sync/parallel.rs +++ b/compiler/rustc_data_structures/src/sync/parallel.rs @@ -96,7 +96,7 @@ macro_rules! parallel { pub fn spawn(func: impl FnOnce() + DynSend + 'static) { if mode::is_dyn_thread_safe() { let func = FromDyn::from(func); - rayon_core::spawn(|| { + rustc_thread_pool::spawn(|| { (func.into_inner())(); }); } else { @@ -107,11 +107,11 @@ pub fn spawn(func: impl FnOnce() + DynSend + 'static) { // This function only works when `mode::is_dyn_thread_safe()`. pub fn scope<'scope, OP, R>(op: OP) -> R where - OP: FnOnce(&rayon_core::Scope<'scope>) -> R + DynSend, + OP: FnOnce(&rustc_thread_pool::Scope<'scope>) -> R + DynSend, R: DynSend, { let op = FromDyn::from(op); - rayon_core::scope(|s| FromDyn::from(op.into_inner()(s))).into_inner() + rustc_thread_pool::scope(|s| FromDyn::from(op.into_inner()(s))).into_inner() } #[inline] @@ -124,7 +124,7 @@ where let oper_a = FromDyn::from(oper_a); let oper_b = FromDyn::from(oper_b); let (a, b) = parallel_guard(|guard| { - rayon_core::join( + rustc_thread_pool::join( move || guard.run(move || FromDyn::from(oper_a.into_inner()())), move || guard.run(move || FromDyn::from(oper_b.into_inner()())), ) @@ -158,7 +158,7 @@ fn par_slice( let (left, right) = items.split_at_mut(items.len() / 2); let mut left = state.for_each.derive(left); let mut right = state.for_each.derive(right); - rayon_core::join(move || par_rec(*left, state), move || par_rec(*right, state)); + rustc_thread_pool::join(move || par_rec(*left, state), move || par_rec(*right, state)); } } @@ -241,7 +241,7 @@ pub fn par_map, R: DynSend, C: FromIterato pub fn broadcast(op: impl Fn(usize) -> R + DynSync) -> Vec { if mode::is_dyn_thread_safe() { let op = FromDyn::from(op); - let results = rayon_core::broadcast(|context| op.derive(op(context.index()))); + let results = rustc_thread_pool::broadcast(|context| op.derive(op(context.index()))); results.into_iter().map(|r| r.into_inner()).collect() } else { vec![op(0)] diff --git a/compiler/rustc_interface/Cargo.toml b/compiler/rustc_interface/Cargo.toml index ff28dbeaee6..a72a7958787 100644 --- a/compiler/rustc_interface/Cargo.toml +++ b/compiler/rustc_interface/Cargo.toml @@ -5,7 +5,6 @@ edition = "2024" [dependencies] # tidy-alphabetical-start -rustc-rayon-core = { version = "0.5.0" } rustc_ast = { path = "../rustc_ast" } rustc_ast_lowering = { path = "../rustc_ast_lowering" } rustc_ast_passes = { path = "../rustc_ast_passes" } @@ -43,6 +42,7 @@ rustc_session = { path = "../rustc_session" } rustc_span = { path = "../rustc_span" } rustc_symbol_mangling = { path = "../rustc_symbol_mangling" } rustc_target = { path = "../rustc_target" } +rustc_thread_pool = { path = "../rustc_thread_pool" } rustc_trait_selection = { path = "../rustc_trait_selection" } rustc_traits = { path = "../rustc_traits" } rustc_ty_utils = { path = "../rustc_ty_utils" } diff --git a/compiler/rustc_interface/src/util.rs b/compiler/rustc_interface/src/util.rs index 8bdc24d47d9..8a7d6117265 100644 --- a/compiler/rustc_interface/src/util.rs +++ b/compiler/rustc_interface/src/util.rs @@ -208,7 +208,7 @@ pub(crate) fn run_in_thread_pool_with_globals< let proxy_ = Arc::clone(&proxy); let proxy__ = Arc::clone(&proxy); - let builder = rayon_core::ThreadPoolBuilder::new() + let builder = rustc_thread_pool::ThreadPoolBuilder::new() .thread_name(|_| "rustc".to_string()) .acquire_thread_handler(move || proxy_.acquire_thread()) .release_thread_handler(move || proxy__.release_thread()) @@ -218,7 +218,7 @@ pub(crate) fn run_in_thread_pool_with_globals< // locals to it. The new thread runs the deadlock handler. let current_gcx2 = current_gcx2.clone(); - let registry = rayon_core::Registry::current(); + let registry = rustc_thread_pool::Registry::current(); let session_globals = rustc_span::with_session_globals(|session_globals| { session_globals as *const SessionGlobals as usize }); @@ -265,7 +265,7 @@ pub(crate) fn run_in_thread_pool_with_globals< builder .build_scoped( // Initialize each new worker thread when created. - move |thread: rayon_core::ThreadBuilder| { + move |thread: rustc_thread_pool::ThreadBuilder| { // Register the thread for use with the `WorkerLocal` type. registry.register(); @@ -274,7 +274,7 @@ pub(crate) fn run_in_thread_pool_with_globals< }) }, // Run `f` on the first thread in the thread pool. - move |pool: &rayon_core::ThreadPool| { + move |pool: &rustc_thread_pool::ThreadPool| { pool.install(|| f(current_gcx.into_inner(), proxy)) }, ) diff --git a/compiler/rustc_middle/Cargo.toml b/compiler/rustc_middle/Cargo.toml index 43c1af642dd..edd0af6e4f5 100644 --- a/compiler/rustc_middle/Cargo.toml +++ b/compiler/rustc_middle/Cargo.toml @@ -9,7 +9,6 @@ bitflags = "2.4.1" either = "1.5.0" gsgdt = "0.1.2" polonius-engine = "0.13.0" -rustc-rayon-core = { version = "0.5.0" } rustc_abi = { path = "../rustc_abi" } rustc_apfloat = "0.2.0" rustc_arena = { path = "../rustc_arena" } @@ -33,6 +32,7 @@ rustc_serialize = { path = "../rustc_serialize" } rustc_session = { path = "../rustc_session" } rustc_span = { path = "../rustc_span" } rustc_target = { path = "../rustc_target" } +rustc_thread_pool = { path = "../rustc_thread_pool" } rustc_type_ir = { path = "../rustc_type_ir" } smallvec = { version = "1.8.1", features = ["union", "may_dangle"] } thin-vec = "0.2.12" diff --git a/compiler/rustc_middle/src/ty/context/tls.rs b/compiler/rustc_middle/src/ty/context/tls.rs index 5fc80bc7936..fa9995898ac 100644 --- a/compiler/rustc_middle/src/ty/context/tls.rs +++ b/compiler/rustc_middle/src/ty/context/tls.rs @@ -36,7 +36,7 @@ impl<'a, 'tcx> ImplicitCtxt<'a, 'tcx> { } // Import the thread-local variable from Rayon, which is preserved for Rayon jobs. -use rayon_core::tlv::TLV; +use rustc_thread_pool::tlv::TLV; #[inline] fn erase(context: &ImplicitCtxt<'_, '_>) -> *const () { diff --git a/compiler/rustc_query_system/Cargo.toml b/compiler/rustc_query_system/Cargo.toml index 7db06953aeb..3d2d879a764 100644 --- a/compiler/rustc_query_system/Cargo.toml +++ b/compiler/rustc_query_system/Cargo.toml @@ -6,7 +6,6 @@ edition = "2024" [dependencies] # tidy-alphabetical-start parking_lot = "0.12" -rustc-rayon-core = { version = "0.5.0" } rustc_abi = { path = "../rustc_abi" } rustc_ast = { path = "../rustc_ast" } rustc_attr_data_structures = { path = "../rustc_attr_data_structures" } @@ -21,6 +20,7 @@ rustc_macros = { path = "../rustc_macros" } rustc_serialize = { path = "../rustc_serialize" } rustc_session = { path = "../rustc_session" } rustc_span = { path = "../rustc_span" } +rustc_thread_pool = { path = "../rustc_thread_pool" } smallvec = { version = "1.8.1", features = ["union", "may_dangle"] } tracing = "0.1" # tidy-alphabetical-end diff --git a/compiler/rustc_query_system/src/query/job.rs b/compiler/rustc_query_system/src/query/job.rs index 1e79bd461d2..7e61f5026da 100644 --- a/compiler/rustc_query_system/src/query/job.rs +++ b/compiler/rustc_query_system/src/query/job.rs @@ -236,7 +236,7 @@ impl QueryLatch { // If this detects a deadlock and the deadlock handler wants to resume this thread // we have to be in the `wait` call. This is ensured by the deadlock handler // getting the self.info lock. - rayon_core::mark_blocked(); + rustc_thread_pool::mark_blocked(); let proxy = qcx.jobserver_proxy(); proxy.release_thread(); waiter.condvar.wait(&mut info); @@ -251,9 +251,9 @@ impl QueryLatch { let mut info = self.info.lock(); debug_assert!(!info.complete); info.complete = true; - let registry = rayon_core::Registry::current(); + let registry = rustc_thread_pool::Registry::current(); for waiter in info.waiters.drain(..) { - rayon_core::mark_unblocked(®istry); + rustc_thread_pool::mark_unblocked(®istry); waiter.condvar.notify_one(); } } @@ -507,7 +507,7 @@ fn remove_cycle( /// all active queries for cycles before finally resuming all the waiters at once. pub fn break_query_cycles( query_map: QueryMap, - registry: &rayon_core::Registry, + registry: &rustc_thread_pool::Registry, ) { let mut wakelist = Vec::new(); // It is OK per the comments: @@ -543,7 +543,7 @@ pub fn break_query_cycles( // we wake the threads up as otherwise Rayon could detect a deadlock if a thread we // resumed fell asleep and this thread had yet to mark the remaining threads as unblocked. for _ in 0..wakelist.len() { - rayon_core::mark_unblocked(registry); + rustc_thread_pool::mark_unblocked(registry); } for waiter in wakelist.into_iter() { diff --git a/compiler/rustc_thread_pool/Cargo.toml b/compiler/rustc_thread_pool/Cargo.toml index f7a3d1306b4..d0bd065c457 100644 --- a/compiler/rustc_thread_pool/Cargo.toml +++ b/compiler/rustc_thread_pool/Cargo.toml @@ -1,25 +1,19 @@ [package] -name = "rustc-rayon-core" -version = "0.5.1" +name = "rustc_thread_pool" +version = "0.0.0" authors = ["Niko Matsakis ", "Josh Stone "] description = "Core APIs for Rayon - fork for rustc" license = "MIT OR Apache-2.0" -repository = "https://github.com/rust-lang/rustc-rayon" -documentation = "https://docs.rs/rustc-rayon-core/" rust-version = "1.63" edition = "2021" readme = "README.md" keywords = ["parallel", "thread", "concurrency", "join", "performance"] categories = ["concurrency"] -[lib] -name = "rayon_core" - -# Some dependencies may not be their latest version, in order to support older rustc. [dependencies] -crossbeam-deque = "0.8.1" -crossbeam-utils = "0.8.0" +crossbeam-deque = "0.8" +crossbeam-utils = "0.8" [dev-dependencies] rand = "0.9" diff --git a/compiler/rustc_thread_pool/src/compile_fail/quicksort_race1.rs b/compiler/rustc_thread_pool/src/compile_fail/quicksort_race1.rs index 5615033895a..1f7a7b0b429 100644 --- a/compiler/rustc_thread_pool/src/compile_fail/quicksort_race1.rs +++ b/compiler/rustc_thread_pool/src/compile_fail/quicksort_race1.rs @@ -7,7 +7,7 @@ fn quick_sort(v: &mut [T]) { let mid = partition(v); let (lo, _hi) = v.split_at_mut(mid); - rayon_core::join(|| quick_sort(lo), || quick_sort(lo)); //~ ERROR + rustc_thred_pool::join(|| quick_sort(lo), || quick_sort(lo)); //~ ERROR } fn partition(v: &mut [T]) -> usize { diff --git a/compiler/rustc_thread_pool/src/compile_fail/quicksort_race2.rs b/compiler/rustc_thread_pool/src/compile_fail/quicksort_race2.rs index 020589c29a8..71b695dd855 100644 --- a/compiler/rustc_thread_pool/src/compile_fail/quicksort_race2.rs +++ b/compiler/rustc_thread_pool/src/compile_fail/quicksort_race2.rs @@ -7,7 +7,7 @@ fn quick_sort(v: &mut [T]) { let mid = partition(v); let (lo, _hi) = v.split_at_mut(mid); - rayon_core::join(|| quick_sort(lo), || quick_sort(v)); //~ ERROR + rustc_thred_pool::join(|| quick_sort(lo), || quick_sort(v)); //~ ERROR } fn partition(v: &mut [T]) -> usize { diff --git a/compiler/rustc_thread_pool/src/compile_fail/quicksort_race3.rs b/compiler/rustc_thread_pool/src/compile_fail/quicksort_race3.rs index 16fbf3b824d..8484cebaae8 100644 --- a/compiler/rustc_thread_pool/src/compile_fail/quicksort_race3.rs +++ b/compiler/rustc_thread_pool/src/compile_fail/quicksort_race3.rs @@ -7,7 +7,7 @@ fn quick_sort(v: &mut [T]) { let mid = partition(v); let (_lo, hi) = v.split_at_mut(mid); - rayon_core::join(|| quick_sort(hi), || quick_sort(hi)); //~ ERROR + rustc_thred_pool::join(|| quick_sort(hi), || quick_sort(hi)); //~ ERROR } fn partition(v: &mut [T]) -> usize { diff --git a/compiler/rustc_thread_pool/src/compile_fail/rc_return.rs b/compiler/rustc_thread_pool/src/compile_fail/rc_return.rs index 93e3a603849..509c8d62ad1 100644 --- a/compiler/rustc_thread_pool/src/compile_fail/rc_return.rs +++ b/compiler/rustc_thread_pool/src/compile_fail/rc_return.rs @@ -2,7 +2,7 @@ use std::rc::Rc; -rayon_core::join(|| Rc::new(22), || ()); //~ ERROR +rustc_thred_pool::join(|| Rc::new(22), || ()); //~ ERROR ``` */ mod left {} @@ -11,7 +11,7 @@ mod left {} use std::rc::Rc; -rayon_core::join(|| (), || Rc::new(23)); //~ ERROR +rustc_thred_pool::join(|| (), || Rc::new(23)); //~ ERROR ``` */ mod right {} diff --git a/compiler/rustc_thread_pool/src/compile_fail/rc_upvar.rs b/compiler/rustc_thread_pool/src/compile_fail/rc_upvar.rs index d8aebcfcbf2..a27b3c8c39f 100644 --- a/compiler/rustc_thread_pool/src/compile_fail/rc_upvar.rs +++ b/compiler/rustc_thread_pool/src/compile_fail/rc_upvar.rs @@ -3,7 +3,7 @@ use std::rc::Rc; let r = Rc::new(22); -rayon_core::join(|| r.clone(), || r.clone()); +rustc_thred_pool::join(|| r.clone(), || r.clone()); //~^ ERROR ``` */ diff --git a/compiler/rustc_thread_pool/src/compile_fail/scope_join_bad.rs b/compiler/rustc_thread_pool/src/compile_fail/scope_join_bad.rs index 75e4c5ca6c0..6e700a483b1 100644 --- a/compiler/rustc_thread_pool/src/compile_fail/scope_join_bad.rs +++ b/compiler/rustc_thread_pool/src/compile_fail/scope_join_bad.rs @@ -3,7 +3,7 @@ fn bad_scope(f: F) where F: FnOnce(&i32) + Send, { - rayon_core::scope(|s| { + rustc_thred_pool::scope(|s| { let x = 22; s.spawn(|_| f(&x)); //~ ERROR `x` does not live long enough }); @@ -13,7 +13,7 @@ fn good_scope(f: F) where F: FnOnce(&i32) + Send, { let x = 22; - rayon_core::scope(|s| { + rustc_thred_pool::scope(|s| { s.spawn(|_| f(&x)); }); } diff --git a/compiler/rustc_thread_pool/src/join/mod.rs b/compiler/rustc_thread_pool/src/join/mod.rs index 798a8347d79..e48d17f16a3 100644 --- a/compiler/rustc_thread_pool/src/join/mod.rs +++ b/compiler/rustc_thread_pool/src/join/mod.rs @@ -41,7 +41,7 @@ mod test; /// [the `par_sort` method]: ../rayon/slice/trait.ParallelSliceMut.html#method.par_sort /// /// ```rust -/// # use rayon_core as rayon; +/// # use rustc_thred_pool as rayon; /// let mut v = vec![5, 1, 8, 22, 0, 44]; /// quick_sort(&mut v); /// assert_eq!(v, vec![0, 1, 5, 8, 22, 44]); diff --git a/compiler/rustc_thread_pool/src/lib.rs b/compiler/rustc_thread_pool/src/lib.rs index 179d63ed668..f1d466b4948 100644 --- a/compiler/rustc_thread_pool/src/lib.rs +++ b/compiler/rustc_thread_pool/src/lib.rs @@ -152,14 +152,14 @@ enum ErrorKind { /// The following creates a thread pool with 22 threads. /// /// ```rust -/// # use rayon_core as rayon; +/// # use rustc_thred_pool as rayon; /// let pool = rayon::ThreadPoolBuilder::new().num_threads(22).build().unwrap(); /// ``` /// /// To instead configure the global thread pool, use [`build_global()`]: /// /// ```rust -/// # use rayon_core as rayon; +/// # use rustc_thred_pool as rayon; /// rayon::ThreadPoolBuilder::new().num_threads(22).build_global().unwrap(); /// ``` /// @@ -315,7 +315,7 @@ impl ThreadPoolBuilder { /// A scoped pool may be useful in combination with scoped thread-local variables. /// /// ``` - /// # use rayon_core as rayon; + /// # use rustc_thred_pool as rayon; /// /// scoped_tls::scoped_thread_local!(static POOL_DATA: Vec); /// @@ -382,7 +382,7 @@ impl ThreadPoolBuilder { /// A minimal spawn handler just needs to call `run()` from an independent thread. /// /// ``` - /// # use rayon_core as rayon; + /// # use rustc_thred_pool as rayon; /// fn main() -> Result<(), rayon::ThreadPoolBuildError> { /// let pool = rayon::ThreadPoolBuilder::new() /// .spawn_handler(|thread| { @@ -400,7 +400,7 @@ impl ThreadPoolBuilder { /// any errors from the thread builder. /// /// ``` - /// # use rayon_core as rayon; + /// # use rustc_thred_pool as rayon; /// fn main() -> Result<(), rayon::ThreadPoolBuildError> { /// let pool = rayon::ThreadPoolBuilder::new() /// .spawn_handler(|thread| { @@ -429,7 +429,7 @@ impl ThreadPoolBuilder { /// [`std::thread::scope`]: https://doc.rust-lang.org/std/thread/fn.scope.html /// /// ``` - /// # use rayon_core as rayon; + /// # use rustc_thred_pool as rayon; /// fn main() -> Result<(), rayon::ThreadPoolBuildError> { /// std::thread::scope(|scope| { /// let pool = rayon::ThreadPoolBuilder::new() diff --git a/compiler/rustc_thread_pool/src/scope/mod.rs b/compiler/rustc_thread_pool/src/scope/mod.rs index 95a4e0b7a18..82b3d053dcb 100644 --- a/compiler/rustc_thread_pool/src/scope/mod.rs +++ b/compiler/rustc_thread_pool/src/scope/mod.rs @@ -84,7 +84,7 @@ struct ScopeBase<'scope> { /// it would be less efficient than the real implementation: /// /// ```rust -/// # use rayon_core as rayon; +/// # use rustc_thred_pool as rayon; /// pub fn join(oper_a: A, oper_b: B) -> (RA, RB) /// where A: FnOnce() -> RA + Send, /// B: FnOnce() -> RB + Send, @@ -125,7 +125,7 @@ struct ScopeBase<'scope> { /// To see how and when tasks are joined, consider this example: /// /// ```rust -/// # use rayon_core as rayon; +/// # use rustc_thred_pool as rayon; /// // point start /// rayon::scope(|s| { /// s.spawn(|s| { // task s.1 @@ -193,7 +193,7 @@ struct ScopeBase<'scope> { /// spawned task. /// /// ```rust -/// # use rayon_core as rayon; +/// # use rustc_thred_pool as rayon; /// let ok: Vec = vec![1, 2, 3]; /// rayon::scope(|s| { /// let bad: Vec = vec![4, 5, 6]; @@ -217,7 +217,7 @@ struct ScopeBase<'scope> { /// in this case including both `ok` *and* `bad`: /// /// ```rust -/// # use rayon_core as rayon; +/// # use rustc_thred_pool as rayon; /// let ok: Vec = vec![1, 2, 3]; /// rayon::scope(|s| { /// let bad: Vec = vec![4, 5, 6]; @@ -238,7 +238,7 @@ struct ScopeBase<'scope> { /// is a borrow of `ok` and capture *that*: /// /// ```rust -/// # use rayon_core as rayon; +/// # use rustc_thred_pool as rayon; /// let ok: Vec = vec![1, 2, 3]; /// rayon::scope(|s| { /// let bad: Vec = vec![4, 5, 6]; @@ -260,7 +260,7 @@ struct ScopeBase<'scope> { /// of individual variables: /// /// ```rust -/// # use rayon_core as rayon; +/// # use rustc_thred_pool as rayon; /// let ok: Vec = vec![1, 2, 3]; /// rayon::scope(|s| { /// let bad: Vec = vec![4, 5, 6]; @@ -312,7 +312,7 @@ where /// [`scope()`]: fn.scope.html /// /// ```rust -/// # use rayon_core as rayon; +/// # use rustc_thred_pool as rayon; /// // point start /// rayon::scope_fifo(|s| { /// s.spawn_fifo(|s| { // task s.1 @@ -487,7 +487,7 @@ impl<'scope> Scope<'scope> { /// # Examples /// /// ```rust - /// # use rayon_core as rayon; + /// # use rustc_thred_pool as rayon; /// let mut value_a = None; /// let mut value_b = None; /// let mut value_c = None; diff --git a/compiler/rustc_thread_pool/src/spawn/mod.rs b/compiler/rustc_thread_pool/src/spawn/mod.rs index f1679a98234..92b89ed5948 100644 --- a/compiler/rustc_thread_pool/src/spawn/mod.rs +++ b/compiler/rustc_thread_pool/src/spawn/mod.rs @@ -50,7 +50,7 @@ use crate::unwind; /// This code creates a Rayon task that increments a global counter. /// /// ```rust -/// # use rayon_core as rayon; +/// # use rustc_thred_pool as rayon; /// use std::sync::atomic::{AtomicUsize, Ordering, ATOMIC_USIZE_INIT}; /// /// static GLOBAL_COUNTER: AtomicUsize = ATOMIC_USIZE_INIT; diff --git a/compiler/rustc_thread_pool/src/thread_pool/mod.rs b/compiler/rustc_thread_pool/src/thread_pool/mod.rs index ce8783cf0d6..c8644ecf9a9 100644 --- a/compiler/rustc_thread_pool/src/thread_pool/mod.rs +++ b/compiler/rustc_thread_pool/src/thread_pool/mod.rs @@ -28,7 +28,7 @@ mod test; /// ## Creating a ThreadPool /// /// ```rust -/// # use rayon_core as rayon; +/// # use rustc_thred_pool as rayon; /// let pool = rayon::ThreadPoolBuilder::new().num_threads(8).build().unwrap(); /// ``` /// @@ -88,10 +88,10 @@ impl ThreadPool { /// meantime. For example /// /// ```rust - /// # use rayon_core as rayon; + /// # use rustc_thred_pool as rayon; /// fn main() { /// rayon::ThreadPoolBuilder::new().num_threads(1).build_global().unwrap(); - /// let pool = rayon_core::ThreadPoolBuilder::default().build().unwrap(); + /// let pool = rustc_thred_pool::ThreadPoolBuilder::default().build().unwrap(); /// let do_it = || { /// print!("one "); /// pool.install(||{}); @@ -123,7 +123,7 @@ impl ThreadPool { /// ## Using `install()` /// /// ```rust - /// # use rayon_core as rayon; + /// # use rustc_thred_pool as rayon; /// fn main() { /// let pool = rayon::ThreadPoolBuilder::new().num_threads(8).build().unwrap(); /// let n = pool.install(|| fib(20)); @@ -172,7 +172,7 @@ impl ThreadPool { /// # Examples /// /// ``` - /// # use rayon_core as rayon; + /// # use rustc_thred_pool as rayon; /// use std::sync::atomic::{AtomicUsize, Ordering}; /// /// fn main() { diff --git a/compiler/rustc_thread_pool/tests/double_init_fail.rs b/compiler/rustc_thread_pool/tests/double_init_fail.rs index 71ed425bb32..ef190099293 100644 --- a/compiler/rustc_thread_pool/tests/double_init_fail.rs +++ b/compiler/rustc_thread_pool/tests/double_init_fail.rs @@ -1,6 +1,6 @@ use std::error::Error; -use rayon_core::ThreadPoolBuilder; +use rustc_thred_pool::ThreadPoolBuilder; #[test] #[cfg_attr(any(target_os = "emscripten", target_family = "wasm"), ignore)] diff --git a/compiler/rustc_thread_pool/tests/init_zero_threads.rs b/compiler/rustc_thread_pool/tests/init_zero_threads.rs index c1770e57f3c..1f7e299e3e9 100644 --- a/compiler/rustc_thread_pool/tests/init_zero_threads.rs +++ b/compiler/rustc_thread_pool/tests/init_zero_threads.rs @@ -1,4 +1,4 @@ -use rayon_core::ThreadPoolBuilder; +use rustc_thred_pool::ThreadPoolBuilder; #[test] #[cfg_attr(any(target_os = "emscripten", target_family = "wasm"), ignore)] diff --git a/compiler/rustc_thread_pool/tests/scope_join.rs b/compiler/rustc_thread_pool/tests/scope_join.rs index 9d88133bc5b..0bd33d086cf 100644 --- a/compiler/rustc_thread_pool/tests/scope_join.rs +++ b/compiler/rustc_thread_pool/tests/scope_join.rs @@ -4,7 +4,7 @@ where F: FnOnce() + Send, G: FnOnce() + Send, { - rayon_core::scope(|s| { + rustc_thred_pool::scope(|s| { s.spawn(|_| g()); f(); }); diff --git a/compiler/rustc_thread_pool/tests/scoped_threadpool.rs b/compiler/rustc_thread_pool/tests/scoped_threadpool.rs index 8cc2c859c0c..e4b0f6c41e1 100644 --- a/compiler/rustc_thread_pool/tests/scoped_threadpool.rs +++ b/compiler/rustc_thread_pool/tests/scoped_threadpool.rs @@ -1,5 +1,5 @@ use crossbeam_utils::thread; -use rayon_core::ThreadPoolBuilder; +use rustc_thred_pool::ThreadPoolBuilder; #[derive(PartialEq, Eq, Debug)] struct Local(i32); diff --git a/compiler/rustc_thread_pool/tests/simple_panic.rs b/compiler/rustc_thread_pool/tests/simple_panic.rs index 2564482a47e..16896e36fa0 100644 --- a/compiler/rustc_thread_pool/tests/simple_panic.rs +++ b/compiler/rustc_thread_pool/tests/simple_panic.rs @@ -1,4 +1,4 @@ -use rayon_core::join; +use rustc_thred_pool::join; #[test] #[should_panic(expected = "should panic")] diff --git a/compiler/rustc_thread_pool/tests/stack_overflow_crash.rs b/compiler/rustc_thread_pool/tests/stack_overflow_crash.rs index c7a880de8bb..49c9ca1d75e 100644 --- a/compiler/rustc_thread_pool/tests/stack_overflow_crash.rs +++ b/compiler/rustc_thread_pool/tests/stack_overflow_crash.rs @@ -3,7 +3,7 @@ use std::env; use std::os::unix::process::ExitStatusExt; use std::process::{Command, ExitStatus, Stdio}; -use rayon_core::ThreadPoolBuilder; +use rustc_thred_pool::ThreadPoolBuilder; fn force_stack_overflow(depth: u32) { let mut buffer = [0u8; 1024 * 1024]; diff --git a/src/tools/tidy/src/deps.rs b/src/tools/tidy/src/deps.rs index fdca7a7a40e..7cdfe79a40b 100644 --- a/src/tools/tidy/src/deps.rs +++ b/src/tools/tidy/src/deps.rs @@ -356,6 +356,7 @@ const PERMITTED_RUSTC_DEPENDENCIES: &[&str] = &[ "rand", "rand_chacha", "rand_core", + "rand_xorshift", // dependency for doc-tests in rustc_thread_pool "rand_xoshiro", "redox_syscall", "regex", @@ -364,7 +365,6 @@ const PERMITTED_RUSTC_DEPENDENCIES: &[&str] = &[ "rustc-demangle", "rustc-hash", "rustc-literal-escaper", - "rustc-rayon-core", "rustc-stable-hash", "rustc_apfloat", "rustix", -- cgit 1.4.1-3-g733a5 From f52c6eee02fb9a9cfe203ce95c4968c2835c034b Mon Sep 17 00:00:00 2001 From: "Celina G. Val" Date: Wed, 11 Jun 2025 12:32:09 -0700 Subject: Another round of tidy / warning fixes --- Cargo.toml | 2 +- compiler/rustc_data_structures/src/sync.rs | 2 -- compiler/rustc_thread_pool/src/broadcast/mod.rs | 8 ++--- .../src/compile_fail/quicksort_race1.rs | 2 +- .../src/compile_fail/quicksort_race2.rs | 2 +- .../src/compile_fail/quicksort_race3.rs | 2 +- .../src/compile_fail/rc_return.rs | 4 +-- .../rustc_thread_pool/src/compile_fail/rc_upvar.rs | 2 +- .../src/compile_fail/scope_join_bad.rs | 4 +-- compiler/rustc_thread_pool/src/job.rs | 30 ++++++++++--------- compiler/rustc_thread_pool/src/join/mod.rs | 6 ++-- compiler/rustc_thread_pool/src/latch.rs | 32 ++++++++++---------- compiler/rustc_thread_pool/src/lib.rs | 15 +++++----- compiler/rustc_thread_pool/src/registry.rs | 34 +++++++++++----------- compiler/rustc_thread_pool/src/scope/mod.rs | 26 ++++++++--------- compiler/rustc_thread_pool/src/sleep/mod.rs | 6 ++-- compiler/rustc_thread_pool/src/spawn/mod.rs | 10 +++---- compiler/rustc_thread_pool/src/thread_pool/mod.rs | 14 ++++----- .../rustc_thread_pool/tests/double_init_fail.rs | 4 ++- .../rustc_thread_pool/tests/init_zero_threads.rs | 4 ++- compiler/rustc_thread_pool/tests/scope_join.rs | 4 ++- .../rustc_thread_pool/tests/scoped_threadpool.rs | 4 ++- compiler/rustc_thread_pool/tests/simple_panic.rs | 4 ++- .../tests/stack_overflow_crash.rs | 4 ++- 24 files changed, 120 insertions(+), 105 deletions(-) (limited to 'compiler/rustc_thread_pool/src/join/mod.rs') diff --git a/Cargo.toml b/Cargo.toml index c4d2a06f4cb..e08f14d2101 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -60,7 +60,7 @@ exclude = [ "obj", ] -[profile.release.package.rustc-rayon-core] +[profile.release.package.rustc_thread_pool] # The rustc fork of Rayon has deadlock detection code which intermittently # causes overflows in the CI (see https://github.com/rust-lang/rust/issues/90227) # so we turn overflow checks off for now. diff --git a/compiler/rustc_data_structures/src/sync.rs b/compiler/rustc_data_structures/src/sync.rs index 68527dde1f3..3881f3c2aa8 100644 --- a/compiler/rustc_data_structures/src/sync.rs +++ b/compiler/rustc_data_structures/src/sync.rs @@ -22,8 +22,6 @@ //! | | | `parking_lot::Mutex` | //! | `RwLock` | `RefCell` | `parking_lot::RwLock` | //! | `MTLock` [^1] | `T` | `Lock` | -//! | | | | -//! | `ParallelIterator` | `Iterator` | `rustc_thread_pool::iter::ParallelIterator` | //! //! [^1]: `MTLock` is similar to `Lock`, but the serial version avoids the cost //! of a `RefCell`. This is appropriate when interior mutability is not diff --git a/compiler/rustc_thread_pool/src/broadcast/mod.rs b/compiler/rustc_thread_pool/src/broadcast/mod.rs index c2b0d47f829..9545c4b15d8 100644 --- a/compiler/rustc_thread_pool/src/broadcast/mod.rs +++ b/compiler/rustc_thread_pool/src/broadcast/mod.rs @@ -6,7 +6,7 @@ use crate::job::{ArcJob, StackJob}; use crate::latch::{CountLatch, LatchRef}; use crate::registry::{Registry, WorkerThread}; -mod test; +mod tests; /// Executes `op` within every thread in the current threadpool. If this is /// called from a non-Rayon thread, it will execute in the global threadpool. @@ -103,18 +103,18 @@ where }; let n_threads = registry.num_threads(); - let current_thread = WorkerThread::current().as_ref(); + let current_thread = unsafe { WorkerThread::current().as_ref() }; let tlv = crate::tlv::get(); let latch = CountLatch::with_count(n_threads, current_thread); let jobs: Vec<_> = (0..n_threads).map(|_| StackJob::new(tlv, &f, LatchRef::new(&latch))).collect(); - let job_refs = jobs.iter().map(|job| job.as_job_ref()); + let job_refs = jobs.iter().map(|job| unsafe { job.as_job_ref() }); registry.inject_broadcast(job_refs); // Wait for all jobs to complete, then collect the results, maybe propagating a panic. latch.wait(current_thread); - jobs.into_iter().map(|job| job.into_result()).collect() + jobs.into_iter().map(|job| unsafe { job.into_result() }).collect() } /// Execute `op` on every thread in the pool. It will be executed on each diff --git a/compiler/rustc_thread_pool/src/compile_fail/quicksort_race1.rs b/compiler/rustc_thread_pool/src/compile_fail/quicksort_race1.rs index 1f7a7b0b429..f6dbc769699 100644 --- a/compiler/rustc_thread_pool/src/compile_fail/quicksort_race1.rs +++ b/compiler/rustc_thread_pool/src/compile_fail/quicksort_race1.rs @@ -7,7 +7,7 @@ fn quick_sort(v: &mut [T]) { let mid = partition(v); let (lo, _hi) = v.split_at_mut(mid); - rustc_thred_pool::join(|| quick_sort(lo), || quick_sort(lo)); //~ ERROR + rustc_thread_pool::join(|| quick_sort(lo), || quick_sort(lo)); //~ ERROR } fn partition(v: &mut [T]) -> usize { diff --git a/compiler/rustc_thread_pool/src/compile_fail/quicksort_race2.rs b/compiler/rustc_thread_pool/src/compile_fail/quicksort_race2.rs index 71b695dd855..ccd737a700d 100644 --- a/compiler/rustc_thread_pool/src/compile_fail/quicksort_race2.rs +++ b/compiler/rustc_thread_pool/src/compile_fail/quicksort_race2.rs @@ -7,7 +7,7 @@ fn quick_sort(v: &mut [T]) { let mid = partition(v); let (lo, _hi) = v.split_at_mut(mid); - rustc_thred_pool::join(|| quick_sort(lo), || quick_sort(v)); //~ ERROR + rustc_thread_pool::join(|| quick_sort(lo), || quick_sort(v)); //~ ERROR } fn partition(v: &mut [T]) -> usize { diff --git a/compiler/rustc_thread_pool/src/compile_fail/quicksort_race3.rs b/compiler/rustc_thread_pool/src/compile_fail/quicksort_race3.rs index 8484cebaae8..6acdf084433 100644 --- a/compiler/rustc_thread_pool/src/compile_fail/quicksort_race3.rs +++ b/compiler/rustc_thread_pool/src/compile_fail/quicksort_race3.rs @@ -7,7 +7,7 @@ fn quick_sort(v: &mut [T]) { let mid = partition(v); let (_lo, hi) = v.split_at_mut(mid); - rustc_thred_pool::join(|| quick_sort(hi), || quick_sort(hi)); //~ ERROR + rustc_thread_pool::join(|| quick_sort(hi), || quick_sort(hi)); //~ ERROR } fn partition(v: &mut [T]) -> usize { diff --git a/compiler/rustc_thread_pool/src/compile_fail/rc_return.rs b/compiler/rustc_thread_pool/src/compile_fail/rc_return.rs index 509c8d62ad1..165c685aba1 100644 --- a/compiler/rustc_thread_pool/src/compile_fail/rc_return.rs +++ b/compiler/rustc_thread_pool/src/compile_fail/rc_return.rs @@ -2,7 +2,7 @@ use std::rc::Rc; -rustc_thred_pool::join(|| Rc::new(22), || ()); //~ ERROR +rustc_thread_pool::join(|| Rc::new(22), || ()); //~ ERROR ``` */ mod left {} @@ -11,7 +11,7 @@ mod left {} use std::rc::Rc; -rustc_thred_pool::join(|| (), || Rc::new(23)); //~ ERROR +rustc_thread_pool::join(|| (), || Rc::new(23)); //~ ERROR ``` */ mod right {} diff --git a/compiler/rustc_thread_pool/src/compile_fail/rc_upvar.rs b/compiler/rustc_thread_pool/src/compile_fail/rc_upvar.rs index a27b3c8c39f..6dc9ead48a0 100644 --- a/compiler/rustc_thread_pool/src/compile_fail/rc_upvar.rs +++ b/compiler/rustc_thread_pool/src/compile_fail/rc_upvar.rs @@ -3,7 +3,7 @@ use std::rc::Rc; let r = Rc::new(22); -rustc_thred_pool::join(|| r.clone(), || r.clone()); +rustc_thread_pool::join(|| r.clone(), || r.clone()); //~^ ERROR ``` */ diff --git a/compiler/rustc_thread_pool/src/compile_fail/scope_join_bad.rs b/compiler/rustc_thread_pool/src/compile_fail/scope_join_bad.rs index 6e700a483b1..e65abfc3c1e 100644 --- a/compiler/rustc_thread_pool/src/compile_fail/scope_join_bad.rs +++ b/compiler/rustc_thread_pool/src/compile_fail/scope_join_bad.rs @@ -3,7 +3,7 @@ fn bad_scope(f: F) where F: FnOnce(&i32) + Send, { - rustc_thred_pool::scope(|s| { + rustc_thread_pool::scope(|s| { let x = 22; s.spawn(|_| f(&x)); //~ ERROR `x` does not live long enough }); @@ -13,7 +13,7 @@ fn good_scope(f: F) where F: FnOnce(&i32) + Send, { let x = 22; - rustc_thred_pool::scope(|s| { + rustc_thread_pool::scope(|s| { s.spawn(|_| f(&x)); }); } diff --git a/compiler/rustc_thread_pool/src/job.rs b/compiler/rustc_thread_pool/src/job.rs index 3241914ba81..e6e84ac2320 100644 --- a/compiler/rustc_thread_pool/src/job.rs +++ b/compiler/rustc_thread_pool/src/job.rs @@ -61,7 +61,7 @@ impl JobRef { #[inline] pub(super) unsafe fn execute(self) { - (self.execute_fn)(self.pointer) + unsafe { (self.execute_fn)(self.pointer) } } } @@ -97,7 +97,7 @@ where } pub(super) unsafe fn as_job_ref(&self) -> JobRef { - JobRef::new(self) + unsafe { JobRef::new(self) } } pub(super) unsafe fn run_inline(self, stolen: bool) -> R { @@ -116,12 +116,16 @@ where R: Send, { unsafe fn execute(this: *const ()) { - let this = &*(this as *const Self); + let this = unsafe { &*(this as *const Self) }; tlv::set(this.tlv); let abort = unwind::AbortIfPanic; - let func = (*this.func.get()).take().unwrap(); - (*this.result.get()) = JobResult::call(func); - Latch::set(&this.latch); + let func = unsafe { (*this.func.get()).take().unwrap() }; + unsafe { + (*this.result.get()) = JobResult::call(func); + } + unsafe { + Latch::set(&this.latch); + } mem::forget(abort); } } @@ -152,7 +156,7 @@ where /// lifetimes, so it is up to you to ensure that this JobRef /// doesn't outlive any data that it closes over. pub(super) unsafe fn into_job_ref(self: Box) -> JobRef { - JobRef::new(Box::into_raw(self)) + unsafe { JobRef::new(Box::into_raw(self)) } } /// Creates a static `JobRef` from this job. @@ -169,7 +173,7 @@ where BODY: FnOnce() + Send, { unsafe fn execute(this: *const ()) { - let this = Box::from_raw(this as *mut Self); + let this = unsafe { Box::from_raw(this as *mut Self) }; tlv::set(this.tlv); (this.job)(); } @@ -196,7 +200,7 @@ where /// lifetimes, so it is up to you to ensure that this JobRef /// doesn't outlive any data that it closes over. pub(super) unsafe fn as_job_ref(this: &Arc) -> JobRef { - JobRef::new(Arc::into_raw(Arc::clone(this))) + unsafe { JobRef::new(Arc::into_raw(Arc::clone(this))) } } /// Creates a static `JobRef` from this job. @@ -213,7 +217,7 @@ where BODY: Fn() + Send + Sync, { unsafe fn execute(this: *const ()) { - let this = Arc::from_raw(this as *mut Self); + let this = unsafe { Arc::from_raw(this as *mut Self) }; (this.job)(); } } @@ -254,17 +258,17 @@ impl JobFifo { // jobs in a thread's deque may be popped from the back (LIFO) or stolen from the front // (FIFO), but either way they will end up popping from the front of this queue. self.inner.push(job_ref); - JobRef::new(self) + unsafe { JobRef::new(self) } } } impl Job for JobFifo { unsafe fn execute(this: *const ()) { // We "execute" a queue by executing its first job, FIFO. - let this = &*(this as *const Self); + let this = unsafe { &*(this as *const Self) }; loop { match this.inner.steal() { - Steal::Success(job_ref) => break job_ref.execute(), + Steal::Success(job_ref) => break unsafe { job_ref.execute() }, Steal::Empty => panic!("FIFO is empty"), Steal::Retry => {} } diff --git a/compiler/rustc_thread_pool/src/join/mod.rs b/compiler/rustc_thread_pool/src/join/mod.rs index e48d17f16a3..f285362c19b 100644 --- a/compiler/rustc_thread_pool/src/join/mod.rs +++ b/compiler/rustc_thread_pool/src/join/mod.rs @@ -7,7 +7,7 @@ use crate::tlv::{self, Tlv}; use crate::{FnContext, unwind}; #[cfg(test)] -mod test; +mod tests; /// Takes two closures and *potentially* runs them in parallel. It /// returns a pair of the results from those closures. @@ -41,7 +41,7 @@ mod test; /// [the `par_sort` method]: ../rayon/slice/trait.ParallelSliceMut.html#method.par_sort /// /// ```rust -/// # use rustc_thred_pool as rayon; +/// # use rustc_thread_pool as rayon; /// let mut v = vec![5, 1, 8, 22, 0, 44]; /// quick_sort(&mut v); /// assert_eq!(v, vec![0, 1, 5, 8, 22, 44]); @@ -192,7 +192,7 @@ unsafe fn join_recover_from_panic( err: Box, tlv: Tlv, ) -> ! { - worker_thread.wait_until(job_b_latch); + unsafe { 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); diff --git a/compiler/rustc_thread_pool/src/latch.rs b/compiler/rustc_thread_pool/src/latch.rs index f2f806e0184..49ba62d3bea 100644 --- a/compiler/rustc_thread_pool/src/latch.rs +++ b/compiler/rustc_thread_pool/src/latch.rs @@ -116,7 +116,7 @@ impl CoreLatch { /// latch code. #[inline] unsafe fn set(this: *const Self) -> bool { - let old_state = (*this).state.swap(SET, Ordering::AcqRel); + let old_state = unsafe { (*this).state.swap(SET, Ordering::AcqRel) }; old_state == SLEEPING } @@ -185,13 +185,13 @@ impl<'r> Latch for SpinLatch<'r> { unsafe fn set(this: *const Self) { let cross_registry; - let registry: &Registry = if (*this).cross { + let registry: &Registry = if unsafe { (*this).cross } { // Ensure the registry stays alive while we notify it. // Otherwise, it would be possible that we set the spin // latch and the other thread sees it and exits, causing // the registry to be deallocated, all before we get a // chance to invoke `registry.notify_worker_latch_is_set`. - cross_registry = Arc::clone((*this).registry); + cross_registry = Arc::clone(unsafe { (*this).registry }); &cross_registry } else { // If this is not a "cross-registry" spin-latch, then the @@ -199,12 +199,12 @@ impl<'r> Latch for SpinLatch<'r> { // that the registry stays alive. However, that doesn't // include this *particular* `Arc` handle if the waiting // thread then exits, so we must completely dereference it. - (*this).registry + unsafe { (*this).registry } }; - let target_worker_index = (*this).target_worker_index; + let target_worker_index = unsafe { (*this).target_worker_index }; // NOTE: Once we `set`, the target may proceed and invalidate `this`! - if CoreLatch::set(&(*this).core_latch) { + if unsafe { CoreLatch::set(&(*this).core_latch) } { // Subtle: at this point, we can no longer read from // `self`, because the thread owning this spin latch may // have awoken and deallocated the latch. Therefore, we @@ -249,9 +249,9 @@ impl LockLatch { impl Latch for LockLatch { #[inline] unsafe fn set(this: *const Self) { - let mut guard = (*this).m.lock().unwrap(); + let mut guard = unsafe { (*this).m.lock().unwrap() }; *guard = true; - (*this).v.notify_all(); + unsafe { (*this).v.notify_all() }; } } @@ -286,7 +286,7 @@ impl OnceLatch { registry: &Registry, target_worker_index: usize, ) { - if CoreLatch::set(&(*this).core_latch) { + if unsafe { CoreLatch::set(&(*this).core_latch) } { registry.notify_worker_latch_is_set(target_worker_index); } } @@ -384,17 +384,17 @@ impl CountLatch { impl Latch for CountLatch { #[inline] unsafe fn set(this: *const Self) { - if (*this).counter.fetch_sub(1, Ordering::SeqCst) == 1 { + if unsafe { (*this).counter.fetch_sub(1, Ordering::SeqCst) == 1 } { // NOTE: Once we call `set` on the internal `latch`, // the target may proceed and invalidate `this`! - match (*this).kind { - CountLatchKind::Stealing { ref latch, ref registry, worker_index } => { + match unsafe { &(*this).kind } { + CountLatchKind::Stealing { latch, registry, worker_index } => { let registry = Arc::clone(registry); - if CoreLatch::set(latch) { - registry.notify_worker_latch_is_set(worker_index); + if unsafe { CoreLatch::set(latch) } { + registry.notify_worker_latch_is_set(*worker_index); } } - CountLatchKind::Blocking { ref latch } => LockLatch::set(latch), + CountLatchKind::Blocking { latch } => unsafe { LockLatch::set(latch) }, } } } @@ -426,6 +426,6 @@ impl Deref for LatchRef<'_, L> { impl Latch for LatchRef<'_, L> { #[inline] unsafe fn set(this: *const Self) { - L::set((*this).inner); + unsafe { L::set((*this).inner) }; } } diff --git a/compiler/rustc_thread_pool/src/lib.rs b/compiler/rustc_thread_pool/src/lib.rs index f1d466b4948..34252d919e3 100644 --- a/compiler/rustc_thread_pool/src/lib.rs +++ b/compiler/rustc_thread_pool/src/lib.rs @@ -61,6 +61,7 @@ //! conflicting requirements will need to be resolved before the build will //! succeed. +#![cfg_attr(test, allow(unused_crate_dependencies))] #![warn(rust_2018_idioms)] use std::any::Any; @@ -85,7 +86,7 @@ mod unwind; mod worker_local; mod compile_fail; -mod test; +mod tests; pub mod tlv; @@ -152,14 +153,14 @@ enum ErrorKind { /// The following creates a thread pool with 22 threads. /// /// ```rust -/// # use rustc_thred_pool as rayon; +/// # use rustc_thread_pool as rayon; /// let pool = rayon::ThreadPoolBuilder::new().num_threads(22).build().unwrap(); /// ``` /// /// To instead configure the global thread pool, use [`build_global()`]: /// /// ```rust -/// # use rustc_thred_pool as rayon; +/// # use rustc_thread_pool as rayon; /// rayon::ThreadPoolBuilder::new().num_threads(22).build_global().unwrap(); /// ``` /// @@ -315,7 +316,7 @@ impl ThreadPoolBuilder { /// A scoped pool may be useful in combination with scoped thread-local variables. /// /// ``` - /// # use rustc_thred_pool as rayon; + /// # use rustc_thread_pool as rayon; /// /// scoped_tls::scoped_thread_local!(static POOL_DATA: Vec); /// @@ -382,7 +383,7 @@ impl ThreadPoolBuilder { /// A minimal spawn handler just needs to call `run()` from an independent thread. /// /// ``` - /// # use rustc_thred_pool as rayon; + /// # use rustc_thread_pool as rayon; /// fn main() -> Result<(), rayon::ThreadPoolBuildError> { /// let pool = rayon::ThreadPoolBuilder::new() /// .spawn_handler(|thread| { @@ -400,7 +401,7 @@ impl ThreadPoolBuilder { /// any errors from the thread builder. /// /// ``` - /// # use rustc_thred_pool as rayon; + /// # use rustc_thread_pool as rayon; /// fn main() -> Result<(), rayon::ThreadPoolBuildError> { /// let pool = rayon::ThreadPoolBuilder::new() /// .spawn_handler(|thread| { @@ -429,7 +430,7 @@ impl ThreadPoolBuilder { /// [`std::thread::scope`]: https://doc.rust-lang.org/std/thread/fn.scope.html /// /// ``` - /// # use rustc_thred_pool as rayon; + /// # use rustc_thread_pool as rayon; /// fn main() -> Result<(), rayon::ThreadPoolBuildError> { /// std::thread::scope(|scope| { /// let pool = rayon::ThreadPoolBuilder::new() diff --git a/compiler/rustc_thread_pool/src/registry.rs b/compiler/rustc_thread_pool/src/registry.rs index 2848556aab6..03a01aa29d2 100644 --- a/compiler/rustc_thread_pool/src/registry.rs +++ b/compiler/rustc_thread_pool/src/registry.rs @@ -533,16 +533,16 @@ impl Registry { |injected| { let worker_thread = WorkerThread::current(); assert!(injected && !worker_thread.is_null()); - op(&*worker_thread, true) + op(unsafe { &*worker_thread }, true) }, LatchRef::new(l), ); - self.inject(job.as_job_ref()); + self.inject(unsafe { job.as_job_ref() }); self.release_thread(); job.latch.wait_and_reset(); // Make sure we can use the same latch again next time. self.acquire_thread(); - job.into_result() + unsafe { job.into_result() } }) } @@ -561,13 +561,13 @@ impl Registry { |injected| { let worker_thread = WorkerThread::current(); assert!(injected && !worker_thread.is_null()); - op(&*worker_thread, true) + op(unsafe { &*worker_thread }, true) }, latch, ); - self.inject(job.as_job_ref()); - current_thread.wait_until(&job.latch); - job.into_result() + self.inject(unsafe { job.as_job_ref() }); + unsafe { current_thread.wait_until(&job.latch) }; + unsafe { job.into_result() } } /// Increments the terminate counter. This increment should be @@ -759,7 +759,7 @@ impl WorkerThread { #[inline] pub(super) unsafe fn push_fifo(&self, job: JobRef) { - self.push(self.fifo.push(job)); + unsafe { self.push(self.fifo.push(job)) }; } #[inline] @@ -798,7 +798,7 @@ impl WorkerThread { pub(super) unsafe fn wait_until(&self, latch: &L) { let latch = latch.as_core_latch(); if !latch.probe() { - self.wait_until_cold(latch); + unsafe { self.wait_until_cold(latch) }; } } @@ -815,7 +815,7 @@ impl WorkerThread { // Check for local work *before* we start marking ourself idle, // especially to avoid modifying shared sleep state. if let Some(job) = self.take_local_job() { - self.execute(job); + unsafe { self.execute(job) }; continue; } @@ -823,7 +823,7 @@ impl WorkerThread { while !latch.probe() { if let Some(job) = self.find_work() { self.registry.sleep.work_found(); - self.execute(job); + unsafe { self.execute(job) }; // The job might have injected local work, so go back to the outer loop. continue 'outer; } else { @@ -846,13 +846,13 @@ impl WorkerThread { let index = self.index; registry.acquire_thread(); - self.wait_until(®istry.thread_infos[index].terminate); + unsafe { self.wait_until(®istry.thread_infos[index].terminate) }; // Should not be any work left in our queue. debug_assert!(self.take_local_job().is_none()); // Let registry know we are done - Latch::set(®istry.thread_infos[index].stopped); + unsafe { Latch::set(®istry.thread_infos[index].stopped) }; } fn find_work(&self) -> Option { @@ -886,7 +886,7 @@ impl WorkerThread { #[inline] pub(super) unsafe fn execute(&self, job: JobRef) { - job.execute(); + unsafe { job.execute() }; } /// Try to steal a single job and return it. @@ -932,12 +932,12 @@ impl WorkerThread { unsafe fn main_loop(thread: ThreadBuilder) { let worker_thread = &WorkerThread::from(thread); - WorkerThread::set_current(worker_thread); + unsafe { WorkerThread::set_current(worker_thread) }; let registry = &*worker_thread.registry; let index = worker_thread.index; // let registry know we are ready to do work - Latch::set(®istry.thread_infos[index].primed); + unsafe { Latch::set(®istry.thread_infos[index].primed) }; // Worker threads should not panic. If they do, just abort, as the // internal state of the threadpool is corrupted. Note that if @@ -949,7 +949,7 @@ unsafe fn main_loop(thread: ThreadBuilder) { registry.catch_unwind(|| handler(index)); } - worker_thread.wait_until_out_of_work(); + unsafe { worker_thread.wait_until_out_of_work() }; // Normal termination, do not abort. mem::forget(abort_guard); diff --git a/compiler/rustc_thread_pool/src/scope/mod.rs b/compiler/rustc_thread_pool/src/scope/mod.rs index 82b3d053dcb..55e58b3509d 100644 --- a/compiler/rustc_thread_pool/src/scope/mod.rs +++ b/compiler/rustc_thread_pool/src/scope/mod.rs @@ -20,7 +20,7 @@ use crate::tlv::{self, Tlv}; use crate::unwind; #[cfg(test)] -mod test; +mod tests; /// Represents a fork-join scope which can be used to spawn any number of tasks. /// See [`scope()`] for more information. @@ -84,7 +84,7 @@ struct ScopeBase<'scope> { /// it would be less efficient than the real implementation: /// /// ```rust -/// # use rustc_thred_pool as rayon; +/// # use rustc_thread_pool as rayon; /// pub fn join(oper_a: A, oper_b: B) -> (RA, RB) /// where A: FnOnce() -> RA + Send, /// B: FnOnce() -> RB + Send, @@ -125,7 +125,7 @@ struct ScopeBase<'scope> { /// To see how and when tasks are joined, consider this example: /// /// ```rust -/// # use rustc_thred_pool as rayon; +/// # use rustc_thread_pool as rayon; /// // point start /// rayon::scope(|s| { /// s.spawn(|s| { // task s.1 @@ -193,7 +193,7 @@ struct ScopeBase<'scope> { /// spawned task. /// /// ```rust -/// # use rustc_thred_pool as rayon; +/// # use rustc_thread_pool as rayon; /// let ok: Vec = vec![1, 2, 3]; /// rayon::scope(|s| { /// let bad: Vec = vec![4, 5, 6]; @@ -217,7 +217,7 @@ struct ScopeBase<'scope> { /// in this case including both `ok` *and* `bad`: /// /// ```rust -/// # use rustc_thred_pool as rayon; +/// # use rustc_thread_pool as rayon; /// let ok: Vec = vec![1, 2, 3]; /// rayon::scope(|s| { /// let bad: Vec = vec![4, 5, 6]; @@ -238,7 +238,7 @@ struct ScopeBase<'scope> { /// is a borrow of `ok` and capture *that*: /// /// ```rust -/// # use rustc_thred_pool as rayon; +/// # use rustc_thread_pool as rayon; /// let ok: Vec = vec![1, 2, 3]; /// rayon::scope(|s| { /// let bad: Vec = vec![4, 5, 6]; @@ -260,7 +260,7 @@ struct ScopeBase<'scope> { /// of individual variables: /// /// ```rust -/// # use rustc_thred_pool as rayon; +/// # use rustc_thread_pool as rayon; /// let ok: Vec = vec![1, 2, 3]; /// rayon::scope(|s| { /// let bad: Vec = vec![4, 5, 6]; @@ -312,7 +312,7 @@ where /// [`scope()`]: fn.scope.html /// /// ```rust -/// # use rustc_thred_pool as rayon; +/// # use rustc_thread_pool as rayon; /// // point start /// rayon::scope_fifo(|s| { /// s.spawn_fifo(|s| { // task s.1 @@ -487,7 +487,7 @@ impl<'scope> Scope<'scope> { /// # Examples /// /// ```rust - /// # use rustc_thred_pool as rayon; + /// # use rustc_thread_pool as rayon; /// let mut value_a = None; /// let mut value_b = None; /// let mut value_c = None; @@ -686,7 +686,7 @@ impl<'scope> ScopeBase<'scope> { where FUNC: FnOnce(), { - let _: Option<()> = Self::execute_job_closure(this, func); + let _: Option<()> = unsafe { Self::execute_job_closure(this, func) }; } /// Executes `func` as a job in scope. Adjusts the "job completed" @@ -699,11 +699,11 @@ impl<'scope> ScopeBase<'scope> { let result = match unwind::halt_unwinding(func) { Ok(r) => Some(r), Err(err) => { - (*this).job_panicked(err); + unsafe { (*this).job_panicked(err) }; None } }; - Latch::set(&(*this).job_completed_latch); + unsafe { Latch::set(&(*this).job_completed_latch) }; result } @@ -778,6 +778,6 @@ unsafe impl Sync for ScopePtr {} impl ScopePtr { // Helper to avoid disjoint captures of `scope_ptr.0` unsafe fn as_ref(&self) -> &T { - &*self.0 + unsafe { &*self.0 } } } diff --git a/compiler/rustc_thread_pool/src/sleep/mod.rs b/compiler/rustc_thread_pool/src/sleep/mod.rs index bee7c82c450..a9cdf68cc7e 100644 --- a/compiler/rustc_thread_pool/src/sleep/mod.rs +++ b/compiler/rustc_thread_pool/src/sleep/mod.rs @@ -31,7 +31,7 @@ struct SleepData { impl SleepData { /// Checks if the conditions for a deadlock holds and if so calls the deadlock handler #[inline] - pub fn deadlock_check(&self, deadlock_handler: &Option>) { + pub(super) fn deadlock_check(&self, deadlock_handler: &Option>) { if self.active_threads == 0 && self.blocked_threads > 0 { (deadlock_handler.as_ref().unwrap())(); } @@ -102,7 +102,7 @@ impl Sleep { /// Mark a Rayon worker thread as blocked. This triggers the deadlock handler /// if no other worker thread is active #[inline] - pub fn mark_blocked(&self, deadlock_handler: &Option>) { + pub(super) fn mark_blocked(&self, deadlock_handler: &Option>) { let mut data = self.data.lock().unwrap(); debug_assert!(data.active_threads > 0); debug_assert!(data.blocked_threads < data.worker_count); @@ -115,7 +115,7 @@ impl Sleep { /// Mark a previously blocked Rayon worker thread as unblocked #[inline] - pub fn mark_unblocked(&self) { + pub(super) fn mark_unblocked(&self) { let mut data = self.data.lock().unwrap(); debug_assert!(data.active_threads < data.worker_count); debug_assert!(data.blocked_threads > 0); diff --git a/compiler/rustc_thread_pool/src/spawn/mod.rs b/compiler/rustc_thread_pool/src/spawn/mod.rs index 92b89ed5948..040a02bfa67 100644 --- a/compiler/rustc_thread_pool/src/spawn/mod.rs +++ b/compiler/rustc_thread_pool/src/spawn/mod.rs @@ -50,7 +50,7 @@ use crate::unwind; /// This code creates a Rayon task that increments a global counter. /// /// ```rust -/// # use rustc_thred_pool as rayon; +/// # use rustc_thread_pool as rayon; /// use std::sync::atomic::{AtomicUsize, Ordering, ATOMIC_USIZE_INIT}; /// /// static GLOBAL_COUNTER: AtomicUsize = ATOMIC_USIZE_INIT; @@ -80,7 +80,7 @@ where // be able to panic, and hence the data won't leak but will be // enqueued into some deque for later execution. let abort_guard = unwind::AbortIfPanic; // just in case we are wrong, and code CAN panic - let job_ref = spawn_job(func, registry); + let job_ref = unsafe { spawn_job(func, registry) }; registry.inject_or_push(job_ref); mem::forget(abort_guard); } @@ -150,16 +150,16 @@ where // be able to panic, and hence the data won't leak but will be // enqueued into some deque for later execution. let abort_guard = unwind::AbortIfPanic; // just in case we are wrong, and code CAN panic - let job_ref = spawn_job(func, registry); + let job_ref = unsafe { spawn_job(func, registry) }; // If we're in the pool, use our thread's private fifo for this thread to execute // in a locally-FIFO order. Otherwise, just use the pool's global injector. match registry.current_thread() { - Some(worker) => worker.push_fifo(job_ref), + Some(worker) => unsafe { worker.push_fifo(job_ref) }, None => registry.inject(job_ref), } mem::forget(abort_guard); } #[cfg(test)] -mod test; +mod tests; diff --git a/compiler/rustc_thread_pool/src/thread_pool/mod.rs b/compiler/rustc_thread_pool/src/thread_pool/mod.rs index c8644ecf9a9..3294e2a77cb 100644 --- a/compiler/rustc_thread_pool/src/thread_pool/mod.rs +++ b/compiler/rustc_thread_pool/src/thread_pool/mod.rs @@ -14,7 +14,7 @@ use crate::{ Scope, ScopeFifo, ThreadPoolBuildError, ThreadPoolBuilder, join, scope, scope_fifo, spawn, }; -mod test; +mod tests; /// Represents a user created [thread-pool]. /// @@ -28,7 +28,7 @@ mod test; /// ## Creating a ThreadPool /// /// ```rust -/// # use rustc_thred_pool as rayon; +/// # use rustc_thread_pool as rayon; /// let pool = rayon::ThreadPoolBuilder::new().num_threads(8).build().unwrap(); /// ``` /// @@ -88,10 +88,10 @@ impl ThreadPool { /// meantime. For example /// /// ```rust - /// # use rustc_thred_pool as rayon; + /// # use rustc_thread_pool as rayon; /// fn main() { /// rayon::ThreadPoolBuilder::new().num_threads(1).build_global().unwrap(); - /// let pool = rustc_thred_pool::ThreadPoolBuilder::default().build().unwrap(); + /// let pool = rustc_thread_pool::ThreadPoolBuilder::default().build().unwrap(); /// let do_it = || { /// print!("one "); /// pool.install(||{}); @@ -123,7 +123,7 @@ impl ThreadPool { /// ## Using `install()` /// /// ```rust - /// # use rustc_thred_pool as rayon; + /// # use rustc_thread_pool as rayon; /// fn main() { /// let pool = rayon::ThreadPoolBuilder::new().num_threads(8).build().unwrap(); /// let n = pool.install(|| fib(20)); @@ -172,7 +172,7 @@ impl ThreadPool { /// # Examples /// /// ``` - /// # use rustc_thred_pool as rayon; + /// # use rustc_thread_pool as rayon; /// use std::sync::atomic::{AtomicUsize, Ordering}; /// /// fn main() { @@ -401,7 +401,7 @@ impl ThreadPool { } pub(crate) fn wait_until_stopped(self) { - let registry = self.registry.clone(); + let registry = Arc::clone(&self.registry); drop(self); registry.wait_until_stopped(); } diff --git a/compiler/rustc_thread_pool/tests/double_init_fail.rs b/compiler/rustc_thread_pool/tests/double_init_fail.rs index ef190099293..85e509518d4 100644 --- a/compiler/rustc_thread_pool/tests/double_init_fail.rs +++ b/compiler/rustc_thread_pool/tests/double_init_fail.rs @@ -1,6 +1,8 @@ +#![allow(unused_crate_dependencies)] + use std::error::Error; -use rustc_thred_pool::ThreadPoolBuilder; +use rustc_thread_pool::ThreadPoolBuilder; #[test] #[cfg_attr(any(target_os = "emscripten", target_family = "wasm"), ignore)] diff --git a/compiler/rustc_thread_pool/tests/init_zero_threads.rs b/compiler/rustc_thread_pool/tests/init_zero_threads.rs index 1f7e299e3e9..261493fcb7b 100644 --- a/compiler/rustc_thread_pool/tests/init_zero_threads.rs +++ b/compiler/rustc_thread_pool/tests/init_zero_threads.rs @@ -1,4 +1,6 @@ -use rustc_thred_pool::ThreadPoolBuilder; +#![allow(unused_crate_dependencies)] + +use rustc_thread_pool::ThreadPoolBuilder; #[test] #[cfg_attr(any(target_os = "emscripten", target_family = "wasm"), ignore)] diff --git a/compiler/rustc_thread_pool/tests/scope_join.rs b/compiler/rustc_thread_pool/tests/scope_join.rs index 0bd33d086cf..83468da81c0 100644 --- a/compiler/rustc_thread_pool/tests/scope_join.rs +++ b/compiler/rustc_thread_pool/tests/scope_join.rs @@ -1,10 +1,12 @@ +#![allow(unused_crate_dependencies)] + /// Test that one can emulate join with `scope`: fn pseudo_join(f: F, g: G) where F: FnOnce() + Send, G: FnOnce() + Send, { - rustc_thred_pool::scope(|s| { + rustc_thread_pool::scope(|s| { s.spawn(|_| g()); f(); }); diff --git a/compiler/rustc_thread_pool/tests/scoped_threadpool.rs b/compiler/rustc_thread_pool/tests/scoped_threadpool.rs index e4b0f6c41e1..295da650e88 100644 --- a/compiler/rustc_thread_pool/tests/scoped_threadpool.rs +++ b/compiler/rustc_thread_pool/tests/scoped_threadpool.rs @@ -1,5 +1,7 @@ +#![allow(unused_crate_dependencies)] + use crossbeam_utils::thread; -use rustc_thred_pool::ThreadPoolBuilder; +use rustc_thread_pool::ThreadPoolBuilder; #[derive(PartialEq, Eq, Debug)] struct Local(i32); diff --git a/compiler/rustc_thread_pool/tests/simple_panic.rs b/compiler/rustc_thread_pool/tests/simple_panic.rs index 16896e36fa0..b35b4d632d2 100644 --- a/compiler/rustc_thread_pool/tests/simple_panic.rs +++ b/compiler/rustc_thread_pool/tests/simple_panic.rs @@ -1,4 +1,6 @@ -use rustc_thred_pool::join; +#![allow(unused_crate_dependencies)] + +use rustc_thread_pool::join; #[test] #[should_panic(expected = "should panic")] diff --git a/compiler/rustc_thread_pool/tests/stack_overflow_crash.rs b/compiler/rustc_thread_pool/tests/stack_overflow_crash.rs index 49c9ca1d75e..805b6d8ee3f 100644 --- a/compiler/rustc_thread_pool/tests/stack_overflow_crash.rs +++ b/compiler/rustc_thread_pool/tests/stack_overflow_crash.rs @@ -1,9 +1,11 @@ +#![allow(unused_crate_dependencies)] + use std::env; #[cfg(target_os = "linux")] use std::os::unix::process::ExitStatusExt; use std::process::{Command, ExitStatus, Stdio}; -use rustc_thred_pool::ThreadPoolBuilder; +use rustc_thread_pool::ThreadPoolBuilder; fn force_stack_overflow(depth: u32) { let mut buffer = [0u8; 1024 * 1024]; -- cgit 1.4.1-3-g733a5