Upgrade rayon-core to 1.10.1 am: acae64b67d am: bae007f73a am: 6529ff1800 Original change: https://android-review.googlesource.com/c/platform/external/rust/crates/rayon-core/+/2345307 Change-Id: I5d69a47690602209295e9019e4d6f7278844bfd5 Signed-off-by: Automerger Merge Worker <[email protected]>
diff --git a/.cargo_vcs_info.json b/.cargo_vcs_info.json index c8aa4e4..975ebdb 100644 --- a/.cargo_vcs_info.json +++ b/.cargo_vcs_info.json
@@ -1,5 +1,6 @@ { "git": { - "sha1": "ebcb09b1dc53211c6b5abdf4dc5b40e4bcd0a965" - } -} + "sha1": "8e48eae53e969f328ef41f8d6977bb8cb7084bec" + }, + "path_in_vcs": "rayon-core" +} \ No newline at end of file
diff --git a/Android.bp b/Android.bp index 65b0527..54c4506 100644 --- a/Android.bp +++ b/Android.bp
@@ -42,14 +42,17 @@ host_supported: true, crate_name: "rayon_core", cargo_env_compat: true, - cargo_pkg_version: "1.9.1", + cargo_pkg_version: "1.10.1", srcs: ["src/lib.rs"], - edition: "2018", + edition: "2021", rustlibs: [ "libcrossbeam_channel", "libcrossbeam_deque", "libcrossbeam_utils", - "liblazy_static", "libnum_cpus", ], + apex_available: [ + "//apex_available:platform", + "//apex_available:anyapex", + ], }
diff --git a/Cargo.toml b/Cargo.toml index 8b110fd..da520f7 100644 --- a/Cargo.toml +++ b/Cargo.toml
@@ -3,26 +3,35 @@ # When uploading crates to the registry Cargo will automatically # "normalize" Cargo.toml files for maximal compatibility # with all versions of Cargo and also rewrite `path` dependencies -# to registry (e.g., crates.io) dependencies +# to registry (e.g., crates.io) dependencies. # -# If you believe there's an error in this file please file an -# issue against the rust-lang/cargo repository. If you're -# editing this file be aware that the upstream Cargo.toml -# will likely look very different (and much more reasonable) +# If you are reading this file be aware that the original Cargo.toml +# will likely look very different (and much more reasonable). +# See Cargo.toml.orig for the original contents. [package] -edition = "2018" +edition = "2021" +rust-version = "1.56" name = "rayon-core" -version = "1.9.1" -authors = ["Niko Matsakis <[email protected]>", "Josh Stone <[email protected]>"] +version = "1.10.1" +authors = [ + "Niko Matsakis <[email protected]>", + "Josh Stone <[email protected]>", +] build = "build.rs" links = "rayon-core" description = "Core APIs for Rayon" documentation = "https://docs.rs/rayon/" readme = "README.md" -keywords = ["parallel", "thread", "concurrency", "join", "performance"] +keywords = [ + "parallel", + "thread", + "concurrency", + "join", + "performance", +] categories = ["concurrency"] -license = "Apache-2.0/MIT" +license = "MIT OR Apache-2.0" repository = "https://github.com/rayon-rs/rayon" [[test]] @@ -49,20 +58,19 @@ [[test]] name = "scoped_threadpool" path = "tests/scoped_threadpool.rs" + [dependencies.crossbeam-channel] version = "0.5.0" [dependencies.crossbeam-deque] -version = "0.8.0" +version = "0.8.1" [dependencies.crossbeam-utils] version = "0.8.0" -[dependencies.lazy_static] -version = "1" - [dependencies.num_cpus] version = "1.2" + [dev-dependencies.rand] version = "0.8" @@ -71,5 +79,6 @@ [dev-dependencies.scoped-tls] version = "1.0" + [target."cfg(unix)".dev-dependencies.libc] version = "0.2"
diff --git a/Cargo.toml.orig b/Cargo.toml.orig index d31ed22..9e45992 100644 --- a/Cargo.toml.orig +++ b/Cargo.toml.orig
@@ -1,13 +1,14 @@ [package] name = "rayon-core" -version = "1.9.1" # reminder to update html_root_url attribute +version = "1.10.1" authors = ["Niko Matsakis <[email protected]>", "Josh Stone <[email protected]>"] description = "Core APIs for Rayon" -license = "Apache-2.0/MIT" +license = "MIT OR Apache-2.0" repository = "https://github.com/rayon-rs/rayon" documentation = "https://docs.rs/rayon/" -edition = "2018" +rust-version = "1.56" +edition = "2021" links = "rayon-core" build = "build.rs" readme = "README.md" @@ -17,9 +18,8 @@ # Some dependencies may not be their latest version, in order to support older rustc. [dependencies] num_cpus = "1.2" -lazy_static = "1" crossbeam-channel = "0.5.0" -crossbeam-deque = "0.8.0" +crossbeam-deque = "0.8.1" crossbeam-utils = "0.8.0" [dev-dependencies]
diff --git a/METADATA b/METADATA index 7814725..58639ba 100644 --- a/METADATA +++ b/METADATA
@@ -1,3 +1,7 @@ +# This project was upgraded with external_updater. +# Usage: tools/external_updater/updater.sh update rust/crates/rayon-core +# For more info, check https://cs.android.com/android/platform/superproject/+/master:tools/external_updater/README.md + name: "rayon-core" description: "Core APIs for Rayon" third_party { @@ -7,13 +11,13 @@ } url { type: ARCHIVE - value: "https://static.crates.io/crates/rayon-core/rayon-core-1.9.1.crate" + value: "https://static.crates.io/crates/rayon-core/rayon-core-1.10.1.crate" } - version: "1.9.1" + version: "1.10.1" license_type: NOTICE last_upgrade_date { - year: 2021 - month: 5 - day: 19 + year: 2022 + month: 12 + day: 13 } }
diff --git a/README.md b/README.md index 0c49362..7bf2d9b 100644 --- a/README.md +++ b/README.md
@@ -8,4 +8,4 @@ [Rayon Docs]: https://docs.rs/rayon/ -Rayon-core currently requires `rustc 1.36.0` or greater. +Rayon-core currently requires `rustc 1.56.0` or greater.
diff --git a/src/broadcast/mod.rs b/src/broadcast/mod.rs new file mode 100644 index 0000000..452aa71 --- /dev/null +++ b/src/broadcast/mod.rs
@@ -0,0 +1,148 @@ +use crate::job::{ArcJob, StackJob}; +use crate::registry::{Registry, WorkerThread}; +use crate::scope::ScopeLatch; +use std::fmt; +use std::marker::PhantomData; +use std::sync::Arc; + +mod test; + +/// 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. +/// Any attempts to use `join`, `scope`, or parallel iterators will then operate +/// within that threadpool. When the call has completed on each thread, returns +/// a vector containing all of their return values. +/// +/// For more information, see the [`ThreadPool::broadcast()`][m] method. +/// +/// [m]: struct.ThreadPool.html#method.broadcast +pub fn broadcast<OP, R>(op: OP) -> Vec<R> +where + OP: Fn(BroadcastContext<'_>) -> R + Sync, + R: Send, +{ + // We assert that current registry has not terminated. + unsafe { broadcast_in(op, &Registry::current()) } +} + +/// Spawns an asynchronous task on every thread in this thread-pool. This task +/// will run in the implicit, global scope, which means that it may outlast the +/// current stack frame -- therefore, it cannot capture any references onto the +/// stack (you will likely need a `move` closure). +/// +/// For more information, see the [`ThreadPool::spawn_broadcast()`][m] method. +/// +/// [m]: struct.ThreadPool.html#method.spawn_broadcast +pub fn spawn_broadcast<OP>(op: OP) +where + OP: Fn(BroadcastContext<'_>) + Send + Sync + 'static, +{ + // We assert that current registry has not terminated. + unsafe { spawn_broadcast_in(op, &Registry::current()) } +} + +/// Provides context to a closure called by `broadcast`. +pub struct BroadcastContext<'a> { + worker: &'a WorkerThread, + + /// Make sure to prevent auto-traits like `Send` and `Sync`. + _marker: PhantomData<&'a mut dyn Fn()>, +} + +impl<'a> BroadcastContext<'a> { + pub(super) fn with<R>(f: impl FnOnce(BroadcastContext<'_>) -> R) -> R { + let worker_thread = WorkerThread::current(); + assert!(!worker_thread.is_null()); + f(BroadcastContext { + worker: unsafe { &*worker_thread }, + _marker: PhantomData, + }) + } + + /// Our index amongst the broadcast threads (ranges from `0..self.num_threads()`). + #[inline] + pub fn index(&self) -> usize { + self.worker.index() + } + + /// The number of threads receiving the broadcast in the thread pool. + /// + /// # Future compatibility note + /// + /// Future versions of Rayon might vary the number of threads over time, but + /// this method will always return the number of threads which are actually + /// receiving your particular `broadcast` call. + #[inline] + pub fn num_threads(&self) -> usize { + self.worker.registry().num_threads() + } +} + +impl<'a> fmt::Debug for BroadcastContext<'a> { + fn fmt(&self, fmt: &mut fmt::Formatter<'_>) -> fmt::Result { + fmt.debug_struct("BroadcastContext") + .field("index", &self.index()) + .field("num_threads", &self.num_threads()) + .field("pool_id", &self.worker.registry().id()) + .finish() + } +} + +/// Execute `op` on every thread in the pool. It will be executed on each +/// thread when they have nothing else to do locally, before they try to +/// steal work from other threads. This function will not return until all +/// threads have completed the `op`. +/// +/// Unsafe because `registry` must not yet have terminated. +pub(super) unsafe fn broadcast_in<OP, R>(op: OP, registry: &Arc<Registry>) -> Vec<R> +where + OP: Fn(BroadcastContext<'_>) -> R + Sync, + R: Send, +{ + let f = move |injected: bool| { + debug_assert!(injected); + BroadcastContext::with(&op) + }; + + let n_threads = registry.num_threads(); + let current_thread = WorkerThread::current().as_ref(); + let latch = ScopeLatch::with_count(n_threads, current_thread); + let jobs: Vec<_> = (0..n_threads).map(|_| StackJob::new(&f, &latch)).collect(); + let job_refs = jobs.iter().map(|job| 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() +} + +/// Execute `op` on every thread in the pool. It will be executed on each +/// thread when they have nothing else to do locally, before they try to +/// steal work from other threads. This function returns immediately after +/// injecting the jobs. +/// +/// Unsafe because `registry` must not yet have terminated. +pub(super) unsafe fn spawn_broadcast_in<OP>(op: OP, registry: &Arc<Registry>) +where + OP: Fn(BroadcastContext<'_>) + Send + Sync + 'static, +{ + let job = ArcJob::new({ + let registry = Arc::clone(registry); + move || { + registry.catch_unwind(|| BroadcastContext::with(&op)); + registry.terminate(); // (*) permit registry to terminate now + } + }); + + let n_threads = registry.num_threads(); + let job_refs = (0..n_threads).map(|_| { + // Ensure that registry cannot terminate until this job has executed + // on each thread. This ref is decremented at the (*) above. + registry.increment_terminate_count(); + + ArcJob::as_static_job_ref(&job) + }); + + registry.inject_broadcast(job_refs); +}
diff --git a/src/broadcast/test.rs b/src/broadcast/test.rs new file mode 100755 index 0000000..a765cb0 --- /dev/null +++ b/src/broadcast/test.rs
@@ -0,0 +1,216 @@ +#![cfg(test)] + +use crate::ThreadPoolBuilder; +use std::sync::atomic::{AtomicUsize, Ordering}; +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] +fn spawn_broadcast_global() { + let (tx, rx) = crossbeam_channel::unbounded(); + 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] +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] +fn spawn_broadcast_pool() { + let (tx, rx) = crossbeam_channel::unbounded(); + 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] +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] +fn spawn_broadcast_self() { + let (tx, rx) = crossbeam_channel::unbounded(); + 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] +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] +fn spawn_broadcast_mutual() { + let (tx, rx) = crossbeam_channel::unbounded(); + 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] +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] +fn spawn_broadcast_mutual_sleepy() { + let (tx, rx) = crossbeam_channel::unbounded(); + 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] +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] +fn spawn_broadcast_panic_one() { + let (tx, rx) = crossbeam_channel::unbounded(); + let (panic_tx, panic_rx) = crossbeam_channel::unbounded(); + 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] +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] +fn spawn_broadcast_panic_many() { + let (tx, rx) = crossbeam_channel::unbounded(); + let (panic_tx, panic_rx) = crossbeam_channel::unbounded(); + 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] +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)); + }); + } +}
diff --git a/src/compile_fail/rc_return.rs b/src/compile_fail/rc_return.rs index 164f8ce..93e3a60 100644 --- a/src/compile_fail/rc_return.rs +++ b/src/compile_fail/rc_return.rs
@@ -2,9 +2,7 @@ use std::rc::Rc; -fn main() { - rayon_core::join(|| Rc::new(22), || ()); //~ ERROR -} +rayon_core::join(|| Rc::new(22), || ()); //~ ERROR ``` */ mod left {} @@ -13,9 +11,7 @@ use std::rc::Rc; -fn main() { - rayon_core::join(|| (), || Rc::new(23)); //~ ERROR -} +rayon_core::join(|| (), || Rc::new(23)); //~ ERROR ``` */ mod right {}
diff --git a/src/compile_fail/rc_upvar.rs b/src/compile_fail/rc_upvar.rs index 62895bf..d8aebcf 100644 --- a/src/compile_fail/rc_upvar.rs +++ b/src/compile_fail/rc_upvar.rs
@@ -2,10 +2,8 @@ use std::rc::Rc; -fn main() { - let r = Rc::new(22); - rayon_core::join(|| r.clone(), || r.clone()); - //~^ ERROR -} +let r = Rc::new(22); +rayon_core::join(|| r.clone(), || r.clone()); +//~^ ERROR ``` */
diff --git a/src/job.rs b/src/job.rs index a71f1b0..b7a3dae 100644 --- a/src/job.rs +++ b/src/job.rs
@@ -4,6 +4,7 @@ use std::any::Any; use std::cell::UnsafeCell; use std::mem; +use std::sync::Arc; pub(super) enum JobResult<T> { None, @@ -20,7 +21,7 @@ /// Unsafe: this may be called from a different thread than the one /// which scheduled the job, so the implementer must ensure the /// appropriate traits are met, whether `Send`, `Sync`, or both. - unsafe fn execute(this: *const Self); + unsafe fn execute(this: *const ()); } /// Effectively a Job trait object. Each JobRef **must** be executed @@ -45,17 +46,15 @@ where T: Job, { - let fn_ptr: unsafe fn(*const T) = <T as Job>::execute; - // erase types: JobRef { pointer: data as *const (), - execute_fn: mem::transmute(fn_ptr), + execute_fn: <T as Job>::execute, } } #[inline] - pub(super) unsafe fn execute(&self) { + pub(super) unsafe fn execute(self) { (self.execute_fn)(self.pointer) } } @@ -108,18 +107,11 @@ F: FnOnce(bool) -> R + Send, R: Send, { - unsafe fn execute(this: *const Self) { - fn call<R>(func: impl FnOnce(bool) -> R) -> impl FnOnce() -> R { - move || func(true) - } - - let this = &*this; + unsafe fn execute(this: *const ()) { + let this = &*(this as *const Self); let abort = unwind::AbortIfPanic; let func = (*this.func.get()).take().unwrap(); - (*this.result.get()) = match unwind::halt_unwinding(call(func)) { - Ok(x) => JobResult::Ok(x), - Err(x) => JobResult::Panic(x), - }; + (*this.result.get()) = JobResult::call(func); this.latch.set(); mem::forget(abort); } @@ -135,25 +127,30 @@ where BODY: FnOnce() + Send, { - job: UnsafeCell<Option<BODY>>, + job: BODY, } impl<BODY> HeapJob<BODY> where BODY: FnOnce() + Send, { - pub(super) fn new(func: BODY) -> Self { - HeapJob { - job: UnsafeCell::new(Some(func)), - } + pub(super) fn new(job: BODY) -> Box<Self> { + Box::new(HeapJob { job }) } /// Creates a `JobRef` from this job -- note that this hides all /// 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(self: Box<Self>) -> JobRef { - let this: *const Self = mem::transmute(self); - JobRef::new(this) + pub(super) unsafe fn into_job_ref(self: Box<Self>) -> JobRef { + JobRef::new(Box::into_raw(self)) + } + + /// Creates a static `JobRef` from this job. + pub(super) fn into_static_job_ref(self: Box<Self>) -> JobRef + where + BODY: 'static, + { + unsafe { self.into_job_ref() } } } @@ -161,14 +158,63 @@ where BODY: FnOnce() + Send, { - unsafe fn execute(this: *const Self) { - let this: Box<Self> = mem::transmute(this); - let job = (*this.job.get()).take().unwrap(); - job(); + unsafe fn execute(this: *const ()) { + let this = Box::from_raw(this as *mut Self); + (this.job)(); + } +} + +/// Represents a job stored in an `Arc` -- like `HeapJob`, but may +/// be turned into multiple `JobRef`s and called multiple times. +pub(super) struct ArcJob<BODY> +where + BODY: Fn() + Send + Sync, +{ + job: BODY, +} + +impl<BODY> ArcJob<BODY> +where + BODY: Fn() + Send + Sync, +{ + pub(super) fn new(job: BODY) -> Arc<Self> { + Arc::new(ArcJob { job }) + } + + /// Creates a `JobRef` from this job -- note that this hides all + /// 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<Self>) -> JobRef { + JobRef::new(Arc::into_raw(Arc::clone(this))) + } + + /// Creates a static `JobRef` from this job. + pub(super) fn as_static_job_ref(this: &Arc<Self>) -> JobRef + where + BODY: 'static, + { + unsafe { Self::as_job_ref(this) } + } +} + +impl<BODY> Job for ArcJob<BODY> +where + BODY: Fn() + Send + Sync, +{ + unsafe fn execute(this: *const ()) { + let this = Arc::from_raw(this as *mut Self); + (this.job)(); } } impl<T> JobResult<T> { + fn call(func: impl FnOnce(bool) -> T) -> Self { + match unwind::halt_unwinding(|| func(true)) { + Ok(x) => JobResult::Ok(x), + Err(x) => JobResult::Panic(x), + } + } + /// Convert the `JobResult` for a job that has finished (and hence /// its JobResult is populated) into its return value. /// @@ -204,10 +250,11 @@ } impl Job for JobFifo { - unsafe fn execute(this: *const Self) { + unsafe fn execute(this: *const ()) { // We "execute" a queue by executing its first job, FIFO. + let this = &*(this as *const Self); loop { - match (*this).inner.steal() { + match this.inner.steal() { Steal::Success(job_ref) => break job_ref.execute(), Steal::Empty => panic!("FIFO is empty"), Steal::Retry => {}
diff --git a/src/latch.rs b/src/latch.rs index 1d573b7..0909293 100644 --- a/src/latch.rs +++ b/src/latch.rs
@@ -189,7 +189,7 @@ fn set(&self) { let cross_registry; - let registry = if self.cross { + let registry: &Registry = if self.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 @@ -200,7 +200,9 @@ } else { // If this is not a "cross-registry" spin-latch, then the // thread which is performing `set` is itself ensuring - // that the registry stays alive. + // 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. self.registry }; let target_worker_index = self.target_worker_index; @@ -284,9 +286,14 @@ impl CountLatch { #[inline] pub(super) fn new() -> CountLatch { + Self::with_count(1) + } + + #[inline] + pub(super) fn with_count(n: usize) -> CountLatch { CountLatch { core_latch: CoreLatch::new(), - counter: AtomicUsize::new(1), + counter: AtomicUsize::new(n), } } @@ -335,10 +342,10 @@ impl CountLockLatch { #[inline] - pub(super) fn new() -> CountLockLatch { + pub(super) fn with_count(n: usize) -> CountLockLatch { CountLockLatch { lock_latch: LockLatch::new(), - counter: AtomicUsize::new(1), + counter: AtomicUsize::new(n), } }
diff --git a/src/lib.rs b/src/lib.rs index 2df24b2..b31a2d7 100644 --- a/src/lib.rs +++ b/src/lib.rs
@@ -44,7 +44,6 @@ //! conflicting requirements will need to be resolved before the build will //! succeed. -#![doc(html_root_url = "https://docs.rs/rayon-core/1.9")] #![deny(missing_debug_implementations)] #![deny(missing_docs)] #![deny(unreachable_pub)] @@ -63,6 +62,7 @@ #[macro_use] mod private; +mod broadcast; mod job; mod join; mod latch; @@ -76,6 +76,7 @@ mod compile_fail; mod test; +pub use self::broadcast::{broadcast, spawn_broadcast, BroadcastContext}; pub use self::join::{join, join_context}; pub use self::registry::ThreadBuilder; pub use self::scope::{in_place_scope, scope, Scope}; @@ -87,6 +88,17 @@ use self::registry::{CustomSpawn, DefaultSpawn, ThreadSpawn}; +/// Returns the maximum number of threads that Rayon supports in a single thread-pool. +/// +/// If a higher thread count is requested by calling `ThreadPoolBuilder::num_threads` or by setting +/// the `RAYON_NUM_THREADS` environment variable, then it will be reduced to this maximum. +/// +/// The value may vary between different targets, and is subject to change in new Rayon versions. +pub fn max_num_threads() -> usize { + // We are limited by the bits available in the sleep counter's `AtomicUsize`. + crate::sleep::THREADS_MAX +} + /// Returns the number of threads in the current registry. If this /// code is executing within a Rayon thread-pool, then this will be /// the number of threads for the thread-pool of the current @@ -174,6 +186,7 @@ /// /// [`ThreadPoolBuilder`]: struct.ThreadPoolBuilder.html #[deprecated(note = "Use `ThreadPoolBuilder`")] +#[derive(Default)] pub struct Configuration { builder: ThreadPoolBuilder, } @@ -258,7 +271,7 @@ /// The threads in this pool will start by calling `wrapper`, which should /// do initialization and continue by calling `ThreadBuilder::run()`. /// - /// [`crossbeam::scope`]: https://docs.rs/crossbeam/0.7/crossbeam/fn.scope.html + /// [`crossbeam::scope`]: https://docs.rs/crossbeam/0.8/crossbeam/fn.scope.html /// /// # Examples /// @@ -328,7 +341,7 @@ /// if the pool is leaked. Furthermore, the global thread pool doesn't terminate /// until the entire process exits! /// - /// [`crossbeam::scope`]: https://docs.rs/crossbeam/0.7/crossbeam/fn.scope.html + /// [`crossbeam::scope`]: https://docs.rs/crossbeam/0.8/crossbeam/fn.scope.html /// /// # Examples /// @@ -373,6 +386,39 @@ /// Ok(()) /// } /// ``` + /// + /// This can also be used for a pool of scoped threads like [`crossbeam::scope`], + /// or [`std::thread::scope`] introduced in Rust 1.63, which is encapsulated in + /// [`build_scoped`](#method.build_scoped). + /// + /// [`std::thread::scope`]: https://doc.rust-lang.org/std/thread/fn.scope.html + /// + /// ``` + /// # use rayon_core as rayon; + /// fn main() -> Result<(), rayon::ThreadPoolBuildError> { + /// std::thread::scope(|scope| { + /// let pool = rayon::ThreadPoolBuilder::new() + /// .spawn_handler(|thread| { + /// let mut builder = std::thread::Builder::new(); + /// if let Some(name) = thread.name() { + /// builder = builder.name(name.to_string()); + /// } + /// if let Some(size) = thread.stack_size() { + /// builder = builder.stack_size(size); + /// } + /// builder.spawn_scoped(scope, || { + /// // Add any scoped initialization here, then run! + /// thread.run() + /// })?; + /// Ok(()) + /// }) + /// .build()?; + /// + /// pool.install(|| println!("Hello from my custom scoped thread!")); + /// Ok(()) + /// }) + /// } + /// ``` pub fn spawn_handler<F>(self, spawn: F) -> ThreadPoolBuilder<CustomSpawn<F>> where F: FnMut(ThreadBuilder) -> io::Result<()>, @@ -462,7 +508,7 @@ /// **Old environment variable:** `RAYON_NUM_THREADS` is a one-to-one /// replacement of the now deprecated `RAYON_RS_NUM_CPUS` environment /// variable. If both variables are specified, `RAYON_NUM_THREADS` will - /// be prefered. + /// be preferred. pub fn num_threads(mut self, num_threads: usize) -> Self { self.num_threads = num_threads; self @@ -515,7 +561,7 @@ /// to true, however, workers will prefer to execute in a /// *breadth-first* fashion -- that is, they will search for jobs at /// the *bottom* of their local deque. (At present, workers *always* - /// steal from the bottom of other worker's deques, regardless of + /// steal from the bottom of other workers' deques, regardless of /// the setting of this flag.) /// /// If you think of the tasks as a tree, where a parent task @@ -738,15 +784,6 @@ } #[allow(deprecated)] -impl Default for Configuration { - fn default() -> Self { - Configuration { - builder: Default::default(), - } - } -} - -#[allow(deprecated)] impl fmt::Debug for Configuration { fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { self.builder.fmt(f)
diff --git a/src/log.rs b/src/log.rs index e1ff827..7b6daf0 100644 --- a/src/log.rs +++ b/src/log.rs
@@ -93,6 +93,9 @@ /// A job was removed from the global queue. JobUninjected { worker: usize }, + /// A job was broadcasted to N threads. + JobBroadcast { count: usize }, + /// When announcing a job, this was the value of the counters we observed. /// /// No effect on thread state, just a debugging event. @@ -124,15 +127,15 @@ let (sender, receiver) = crossbeam_channel::unbounded(); - if env_log.starts_with("tail:") { - let filename = env_log["tail:".len()..].to_string(); + if let Some(filename) = env_log.strip_prefix("tail:") { + let filename = filename.to_string(); ::std::thread::spawn(move || { Self::tail_logger_thread(num_workers, filename, 10_000, receiver) }); } else if env_log == "all" { ::std::thread::spawn(move || Self::all_logger_thread(num_workers, receiver)); - } else if env_log.starts_with("profile:") { - let filename = env_log["profile:".len()..].to_string(); + } else if let Some(filename) = env_log.strip_prefix("profile:") { + let filename = filename.to_string(); ::std::thread::spawn(move || { Self::profile_logger_thread(num_workers, filename, 10_000, receiver) }); @@ -140,9 +143,9 @@ panic!("RAYON_LOG should be 'tail:<file>' or 'profile:<file>'"); } - return Logger { + Logger { sender: Some(sender), - }; + } } fn disabled() -> Logger { @@ -175,19 +178,12 @@ let timeout = std::time::Duration::from_secs(30); loop { - loop { - match receiver.recv_timeout(timeout) { - Ok(event) => { - if let Event::Flush = event { - break; - } else { - events.push(event); - } - } - - Err(_) => break, + while let Ok(event) = receiver.recv_timeout(timeout) { + if let Event::Flush = event { + break; } + events.push(event); if events.len() == capacity { break; } @@ -219,31 +215,25 @@ let mut skipped = false; loop { - loop { - match receiver.recv_timeout(timeout) { - Ok(event) => { - if let Event::Flush = event { - // We ignore Flush events in tail mode -- - // we're really just looking for - // deadlocks. - continue; - } else { - if events.len() == capacity { - let event = events.pop_front().unwrap(); - state.simulate(&event); - skipped = true; - } - - events.push_back(event); - } + while let Ok(event) = receiver.recv_timeout(timeout) { + if let Event::Flush = event { + // We ignore Flush events in tail mode -- + // we're really just looking for + // deadlocks. + continue; + } else { + if events.len() == capacity { + let event = events.pop_front().unwrap(); + state.simulate(&event); + skipped = true; } - Err(_) => break, + events.push_back(event); } } if skipped { - write!(writer, "...\n").unwrap(); + writeln!(writer, "...").unwrap(); skipped = false; } @@ -417,7 +407,7 @@ } } - write!(w, "\n")?; + writeln!(w)?; Ok(()) } }
diff --git a/src/registry.rs b/src/registry.rs index 4156b90..279e298 100644 --- a/src/registry.rs +++ b/src/registry.rs
@@ -8,7 +8,6 @@ ErrorKind, ExitHandler, PanicHandler, StartHandler, ThreadPoolBuildError, ThreadPoolBuilder, }; use crossbeam_deque::{Injector, Steal, Stealer, Worker}; -use std::any::Any; use std::cell::Cell; use std::collections::hash_map::DefaultHasher; use std::fmt; @@ -16,10 +15,8 @@ use std::io; use std::mem; use std::ptr; -#[allow(deprecated)] -use std::sync::atomic::ATOMIC_USIZE_INIT; use std::sync::atomic::{AtomicUsize, Ordering}; -use std::sync::{Arc, Once}; +use std::sync::{Arc, Mutex, Once}; use std::thread; use std::usize; @@ -29,6 +26,7 @@ name: Option<String>, stack_size: Option<usize>, worker: Worker<JobRef>, + stealer: Stealer<JobRef>, registry: Arc<Registry>, index: usize, } @@ -41,7 +39,7 @@ /// Gets the string that was specified by `ThreadPoolBuilder::name()`. pub fn name(&self) -> Option<&str> { - self.name.as_ref().map(String::as_str) + self.name.as_deref() } /// Gets the value that was specified by `ThreadPoolBuilder::stack_size()`. @@ -52,7 +50,7 @@ /// Executes the main loop for this thread. This will not return until the /// thread pool is dropped. pub fn run(self) { - unsafe { main_loop(self.worker, self.registry, self.index) } + unsafe { main_loop(self.worker, self.stealer, self.registry, self.index) } } } @@ -135,6 +133,7 @@ thread_infos: Vec<ThreadInfo>, sleep: Sleep, injected_jobs: Injector<JobRef>, + broadcasts: Mutex<Vec<Worker<JobRef>>>, panic_handler: Option<Box<PanicHandler>>, start_handler: Option<Box<StartHandler>>, exit_handler: Option<Box<ExitHandler>>, @@ -214,7 +213,9 @@ where S: ThreadSpawn, { - let n_threads = builder.get_num_threads(); + // Soft-limit the number of threads that we can actually support. + let n_threads = Ord::min(builder.get_num_threads(), crate::max_num_threads()); + let breadth_first = builder.get_breadth_first(); let (workers, stealers): (Vec<_>, Vec<_>) = (0..n_threads) @@ -230,12 +231,21 @@ }) .unzip(); + let (broadcasts, broadcast_stealers): (Vec<_>, Vec<_>) = (0..n_threads) + .map(|_| { + let worker = Worker::new_fifo(); + let stealer = worker.stealer(); + (worker, stealer) + }) + .unzip(); + let logger = Logger::new(n_threads); let registry = Arc::new(Registry { logger: logger.clone(), thread_infos: stealers.into_iter().map(ThreadInfo::new).collect(), sleep: Sleep::new(logger, n_threads), injected_jobs: Injector::new(), + broadcasts: Mutex::new(broadcasts), terminate_count: AtomicUsize::new(1), panic_handler: builder.take_panic_handler(), start_handler: builder.take_start_handler(), @@ -245,12 +255,13 @@ // If we return early or panic, make sure to terminate existing threads. let t1000 = Terminator(®istry); - for (index, worker) in workers.into_iter().enumerate() { + for (index, (worker, stealer)) in workers.into_iter().zip(broadcast_stealers).enumerate() { let thread = ThreadBuilder { name: builder.get_thread_name(index), stack_size: builder.get_stack_size(), - registry: registry.clone(), + registry: Arc::clone(®istry), worker, + stealer, index, }; if let Err(e) = builder.get_spawn_handler().spawn(thread) { @@ -261,17 +272,18 @@ // Returning normally now, without termination. mem::forget(t1000); - Ok(registry.clone()) + Ok(registry) } pub(super) fn current() -> Arc<Registry> { unsafe { let worker_thread = WorkerThread::current(); - if worker_thread.is_null() { - global_registry().clone() + let registry = if worker_thread.is_null() { + global_registry() } else { - (*worker_thread).registry.clone() - } + &(*worker_thread).registry + }; + Arc::clone(registry) } } @@ -319,19 +331,14 @@ self.thread_infos.len() } - pub(super) fn handle_panic(&self, err: Box<dyn Any + Send>) { - match self.panic_handler { - Some(ref handler) => { - // If the customizable panic handler itself panics, - // then we abort. - let abort_guard = unwind::AbortIfPanic; + pub(super) fn catch_unwind(&self, f: impl FnOnce()) { + if let Err(err) = unwind::halt_unwinding(f) { + // If there is no handler, or if that handler itself panics, then we abort. + let abort_guard = unwind::AbortIfPanic; + if let Some(ref handler) = self.panic_handler { handler(err); mem::forget(abort_guard); } - None => { - // Default panic handler aborts. - let _ = unwind::AbortIfPanic; // let this drop. - } } } @@ -375,7 +382,7 @@ } /// Push a job into the "external jobs" queue; it will be taken by - /// whatever worker has nothing to do. Use this is you know that + /// whatever worker has nothing to do. Use this if you know that /// you are not on a worker of this registry. pub(super) fn inject(&self, injected_jobs: &[JobRef]) { self.log(|| JobsInjected { @@ -422,6 +429,40 @@ } } + /// Push a job into each thread's own "external jobs" queue; it will be + /// executed only on that thread, when it has nothing else to do locally, + /// before it tries to steal other work. + /// + /// **Panics** if not given exactly as many jobs as there are threads. + pub(super) fn inject_broadcast(&self, injected_jobs: impl ExactSizeIterator<Item = JobRef>) { + assert_eq!(self.num_threads(), injected_jobs.len()); + self.log(|| JobBroadcast { + count: self.num_threads(), + }); + { + let broadcasts = self.broadcasts.lock().unwrap(); + + // It should not be possible for `state.terminate` to be true + // here. It is only set to true when the user creates (and + // drops) a `ThreadPool`; and, in that case, they cannot be + // calling `inject_broadcast()` later, since they dropped their + // `ThreadPool`. + debug_assert_ne!( + self.terminate_count.load(Ordering::Acquire), + 0, + "inject_broadcast() sees state.terminate as true" + ); + + assert_eq!(broadcasts.len(), injected_jobs.len()); + for (worker, job_ref) in broadcasts.iter().zip(injected_jobs) { + worker.push(job_ref); + } + } + for i in 0..self.num_threads() { + self.sleep.notify_worker_latch_is_set(i); + } + } + /// 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 @@ -591,6 +632,9 @@ /// the "worker" half of our local deque worker: Worker<JobRef>, + /// the "stealer" half of the worker's broadcast deque + stealer: Stealer<JobRef>, + /// local queue used for `spawn_fifo` indirection fifo: JobFifo, @@ -686,9 +730,20 @@ if popped_job.is_some() { self.log(|| JobPopped { worker: self.index }); + return popped_job; } - popped_job + loop { + match self.stealer.steal() { + Steal::Success(job) => return Some(job), + Steal::Empty => return None, + Steal::Retry => {} + } + } + } + + fn has_injected_job(&self) -> bool { + !self.stealer.is_empty() || self.registry.has_injected_job() } /// Wait until the latch is set. Try to keep busy by popping and @@ -728,7 +783,7 @@ } else { self.registry .sleep - .no_work_found(&mut idle_state, latch, || self.registry.has_injected_job()) + .no_work_found(&mut idle_state, latch, || self.has_injected_job()) } } @@ -796,15 +851,22 @@ /// //////////////////////////////////////////////////////////////////////// -unsafe fn main_loop(worker: Worker<JobRef>, registry: Arc<Registry>, index: usize) { +unsafe fn main_loop( + worker: Worker<JobRef>, + stealer: Stealer<JobRef>, + registry: Arc<Registry>, + index: usize, +) { let worker_thread = &WorkerThread { worker, + stealer, fifo: JobFifo::new(), index, rng: XorShift64Star::new(), - registry: registry.clone(), + registry, }; WorkerThread::set_current(worker_thread); + let registry = &*worker_thread.registry; // let registry know we are ready to do work registry.thread_infos[index].primed.set(); @@ -816,13 +878,7 @@ // Inform a user callback that we started a thread. if let Some(ref handler) = registry.start_handler { - let registry = registry.clone(); - match unwind::halt_unwinding(|| handler(index)) { - Ok(()) => {} - Err(err) => { - registry.handle_panic(err); - } - } + registry.catch_unwind(|| handler(index)); } let my_terminate_latch = ®istry.thread_infos[index].terminate; @@ -845,13 +901,7 @@ // Inform a user callback that we exited a thread. if let Some(ref handler) = registry.exit_handler { - let registry = registry.clone(); - match unwind::halt_unwinding(|| handler(index)) { - Ok(()) => {} - Err(err) => { - registry.handle_panic(err); - } - } + registry.catch_unwind(|| handler(index)); // We're already exiting the thread, there's nothing else to do. } } @@ -893,8 +943,7 @@ let mut seed = 0; while seed == 0 { let mut hasher = DefaultHasher::new(); - #[allow(deprecated)] - static COUNTER: AtomicUsize = ATOMIC_USIZE_INIT; + static COUNTER: AtomicUsize = AtomicUsize::new(0); hasher.write_usize(COUNTER.fetch_add(1, Ordering::Relaxed)); seed = hasher.finish(); }
diff --git a/src/scope/mod.rs b/src/scope/mod.rs index 543aa26..25cda83 100644 --- a/src/scope/mod.rs +++ b/src/scope/mod.rs
@@ -5,7 +5,8 @@ //! [`in_place_scope()`]: fn.in_place_scope.html //! [`join()`]: ../join/join.fn.html -use crate::job::{HeapJob, JobFifo}; +use crate::broadcast::BroadcastContext; +use crate::job::{ArcJob, HeapJob, JobFifo, JobRef}; use crate::latch::{CountLatch, CountLockLatch, Latch}; use crate::registry::{global_registry, in_worker, Registry, WorkerThread}; use crate::unwind; @@ -38,7 +39,7 @@ fifos: Vec<JobFifo>, } -enum ScopeLatch { +pub(super) enum ScopeLatch { /// A latch for scopes created on a rayon thread which will participate in work- /// stealing while it waits for completion. This thread is not necessarily part /// of the same registry as the scope itself! @@ -127,7 +128,7 @@ /// Task execution potentially starts as soon as `spawn()` is called. /// The task will end sometime before `scope()` returns. Note that the /// *closure* given to scope may return much earlier. In general -/// the lifetime of a scope created like `scope(body) goes something like this: +/// the lifetime of a scope created like `scope(body)` goes something like this: /// /// - Scope begins when `scope(body)` is called /// - Scope body `body()` is invoked @@ -241,7 +242,7 @@ /// }); /// /// // That closure is fine, but now we can't use `ok` anywhere else, -/// // since it is owend by the previous task: +/// // since it is owned by the previous task: /// // s.spawn(|_| println!("ok: {:?}", ok)); /// }); /// ``` @@ -495,7 +496,7 @@ /// the stack (if those variables outlive the scope) or /// communicate through shared channels. /// - /// (The intention is to eventualy integrate with Rust futures to + /// (The intention is to eventually integrate with Rust futures to /// support spawns of functions that compute a value.) /// /// # Examples @@ -538,18 +539,37 @@ where BODY: FnOnce(&Scope<'scope>) + Send + 'scope, { - self.base.increment(); - unsafe { - let job_ref = Box::new(HeapJob::new(move || { - self.base.execute_job(move || body(self)) - })) - .as_job_ref(); + let scope_ptr = ScopePtr(self); + let job = HeapJob::new(move || { + // SAFETY: this job will execute before the scope ends. + let scope = unsafe { scope_ptr.as_ref() }; + scope.base.execute_job(move || body(scope)) + }); + let job_ref = self.base.heap_job_ref(job); - // Since `Scope` implements `Sync`, we can't be sure that we're still in a - // thread of this pool, so we can't just push to the local worker thread. - // Also, this might be an in-place scope. - self.base.registry.inject_or_push(job_ref); - } + // Since `Scope` implements `Sync`, we can't be sure that we're still in a + // thread of this pool, so we can't just push to the local worker thread. + // Also, this might be an in-place scope. + self.base.registry.inject_or_push(job_ref); + } + + /// Spawns a job into every thread of the fork-join scope `self`. This job will + /// 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<BODY>(&self, body: BODY) + where + BODY: Fn(&Scope<'scope>, BroadcastContext<'_>) + Send + Sync + 'scope, + { + let scope_ptr = ScopePtr(self); + let job = ArcJob::new(move || { + // SAFETY: this job will execute before the scope ends. + let scope = unsafe { scope_ptr.as_ref() }; + let body = &body; + let func = move || BroadcastContext::with(move |ctx| body(scope, ctx)); + scope.base.execute_job(func); + }); + self.base.inject_broadcast(job) } } @@ -579,24 +599,44 @@ where BODY: FnOnce(&ScopeFifo<'scope>) + Send + 'scope, { - self.base.increment(); - unsafe { - let job_ref = Box::new(HeapJob::new(move || { - self.base.execute_job(move || body(self)) - })) - .as_job_ref(); + let scope_ptr = ScopePtr(self); + let job = HeapJob::new(move || { + // SAFETY: this job will execute before the scope ends. + let scope = unsafe { scope_ptr.as_ref() }; + scope.base.execute_job(move || body(scope)) + }); + let job_ref = self.base.heap_job_ref(job); - // If we're in the pool, use our scope's private fifo for this thread to execute - // in a locally-FIFO order. Otherwise, just use the pool's global injector. - match self.base.registry.current_thread() { - Some(worker) => { - let fifo = &self.fifos[worker.index()]; - worker.push(fifo.push(job_ref)); - } - None => self.base.registry.inject(&[job_ref]), + // If we're in the pool, use our scope's private fifo for this thread to execute + // in a locally-FIFO order. Otherwise, just use the pool's global injector. + match self.base.registry.current_thread() { + Some(worker) => { + let fifo = &self.fifos[worker.index()]; + // SAFETY: this job will execute before the scope ends. + unsafe { worker.push(fifo.push(job_ref)) }; } + None => self.base.registry.inject(&[job_ref]), } } + + /// 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 + /// 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<BODY>(&self, body: BODY) + where + BODY: Fn(&ScopeFifo<'scope>, BroadcastContext<'_>) + Send + Sync + 'scope, + { + let scope_ptr = ScopePtr(self); + let job = ArcJob::new(move || { + // SAFETY: this job will execute before the scope ends. + let scope = unsafe { scope_ptr.as_ref() }; + let body = &body; + let func = move || BroadcastContext::with(move |ctx| body(scope, ctx)); + scope.base.execute_job(func); + }); + self.base.inject_broadcast(job) + } } impl<'scope> ScopeBase<'scope> { @@ -619,6 +659,29 @@ self.job_completed_latch.increment(); } + fn heap_job_ref<FUNC>(&self, job: Box<HeapJob<FUNC>>) -> JobRef + where + FUNC: FnOnce() + Send + 'scope, + { + unsafe { + self.increment(); + job.into_job_ref() + } + } + + fn inject_broadcast<FUNC>(&self, job: Arc<ArcJob<FUNC>>) + where + FUNC: Fn() + Send + Sync + 'scope, + { + let n_threads = self.registry.num_threads(); + let job_refs = (0..n_threads).map(|_| unsafe { + self.increment(); + ArcJob::as_job_ref(&job) + }); + + self.registry.inject_broadcast(job_refs); + } + /// Executes `func` as a job, either aborting or executing as /// appropriate. fn complete<FUNC, R>(&self, owner: Option<&WorkerThread>, func: FUNC) -> R @@ -687,14 +750,18 @@ impl ScopeLatch { fn new(owner: Option<&WorkerThread>) -> Self { + Self::with_count(1, owner) + } + + pub(super) fn with_count(count: usize, owner: Option<&WorkerThread>) -> Self { match owner { Some(owner) => ScopeLatch::Stealing { - latch: CountLatch::new(), + latch: CountLatch::with_count(count), registry: Arc::clone(owner.registry()), worker_index: owner.index(), }, None => ScopeLatch::Blocking { - latch: CountLockLatch::new(), + latch: CountLockLatch::with_count(count), }, } } @@ -706,18 +773,7 @@ } } - fn set(&self) { - match self { - ScopeLatch::Stealing { - latch, - registry, - worker_index, - } => latch.set_and_tickle_one(registry, *worker_index), - ScopeLatch::Blocking { latch } => latch.set(), - } - } - - fn wait(&self, owner: Option<&WorkerThread>) { + pub(super) fn wait(&self, owner: Option<&WorkerThread>) { match self { ScopeLatch::Stealing { latch, @@ -734,6 +790,19 @@ } } +impl Latch for ScopeLatch { + fn set(&self) { + match self { + ScopeLatch::Stealing { + latch, + registry, + worker_index, + } => latch.set_and_tickle_one(registry, *worker_index), + ScopeLatch::Blocking { latch } => latch.set(), + } + } +} + impl<'scope> fmt::Debug for Scope<'scope> { fn fmt(&self, fmt: &mut fmt::Formatter<'_>) -> fmt::Result { fmt.debug_struct("Scope") @@ -769,3 +838,22 @@ } } } + +/// Used to capture a scope `&Self` pointer in jobs, without faking a lifetime. +/// +/// Unsafe code is still required to dereference the pointer, but that's fine in +/// scope jobs that are guaranteed to execute before the scope ends. +struct ScopePtr<T>(*const T); + +// SAFETY: !Send for raw pointers is not for safety, just as a lint +unsafe impl<T: Sync> Send for ScopePtr<T> {} + +// SAFETY: !Sync for raw pointers is not for safety, just as a lint +unsafe impl<T: Sync> Sync for ScopePtr<T> {} + +impl<T> ScopePtr<T> { + // Helper to avoid disjoint captures of `scope_ptr.0` + unsafe fn as_ref(&self) -> &T { + &*self.0 + } +}
diff --git a/src/scope/test.rs b/src/scope/test.rs index 471d78e..00dd18c 100644 --- a/src/scope/test.rs +++ b/src/scope/test.rs
@@ -6,7 +6,7 @@ use std::cmp; use std::iter::once; use std::sync::atomic::{AtomicUsize, Ordering}; -use std::sync::Mutex; +use std::sync::{Barrier, Mutex}; use std::vec; #[test] @@ -42,7 +42,7 @@ 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); + divide_and_conquer_seq(counter_s, 1024); let p = counter_p.load(Ordering::SeqCst); let s = counter_s.load(Ordering::SeqCst); @@ -75,7 +75,7 @@ } impl<T: Send> Tree<T> { - fn iter<'s>(&'s self) -> vec::IntoIter<&'s T> { + fn iter(&self) -> vec::IntoIter<&T> { once(&self.value) .chain(self.children.iter().flat_map(Tree::iter)) .collect::<Vec<_>>() // seems like it shouldn't be needed... but prevents overflow @@ -245,7 +245,7 @@ let mut x = false; match unwind::halt_unwinding(|| { scope(|s| { - s.spawn(|_| x = true); // spanwed job should still execute despite later panic + s.spawn(|_| x = true); // spawned job should still execute despite later panic panic!("Hello, world!"); }); }) { @@ -293,7 +293,7 @@ #[test] fn lifo_order() { - // In the absense of stealing, `scope()` runs its `spawn()` jobs in LIFO order. + // In the absence of stealing, `scope()` runs its `spawn()` jobs in LIFO order. let vec = test_order!(scope => spawn); let expected: Vec<i32> = (0..100).rev().collect(); // LIFO -> reversed assert_eq!(vec, expected); @@ -301,7 +301,7 @@ #[test] fn fifo_order() { - // In the absense of stealing, `scope_fifo()` runs its `spawn_fifo()` jobs in 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<i32> = (0..100).collect(); // FIFO -> natural order assert_eq!(vec, expected); @@ -335,7 +335,7 @@ #[test] fn nested_lifo_order() { - // In the absense of stealing, `scope()` runs its `spawn()` jobs in 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<i32> = (0..100).rev().collect(); // LIFO -> reversed assert_eq!(vec, expected); @@ -343,7 +343,7 @@ #[test] fn nested_fifo_order() { - // In the absense of stealing, `scope_fifo()` runs its `spawn_fifo()` jobs in 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<i32> = (0..100).collect(); // FIFO -> natural order assert_eq!(vec, expected); @@ -513,3 +513,89 @@ 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] +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] +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] +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/src/sleep/README.md b/src/sleep/README.md index c62c397..55426c8 100644 --- a/src/sleep/README.md +++ b/src/sleep/README.md
@@ -75,7 +75,7 @@ * When a thread awakens a sleeping thread: decrement the sleeping thread counter. * Subtle point: the thread that *awakens* the sleeping thread decrements the counter, not the thread that is *sleeping*. This is because there is a delay - between siganling a thread to wake and the thread actually waking: + between signaling a thread to wake and the thread actually waking: decrementing the counter when awakening the thread means that other threads that may be posting work will see the up-to-date value that much faster. * When a thread finds work, exiting the idle state: decrement the inactive @@ -137,7 +137,7 @@ above, remembering the `final_value` of the JEC. It does one more search for work. * If no work is found, the thread atomically: * Checks the JEC to see that it has not changed from `final_value`. - * If it has, then the thread goes back to searchin for work. We reset to + * If it has, then the thread goes back to searching for work. We reset to just before we got sleepy, so that we will do one more search before attending to sleep again (rather than searching for many rounds). * Increments the number of sleeping threads by 1.
diff --git a/src/sleep/counters.rs b/src/sleep/counters.rs index 879d32e..f2a3de3 100644 --- a/src/sleep/counters.rs +++ b/src/sleep/counters.rs
@@ -53,14 +53,20 @@ } /// Number of bits used for the thread counters. -const THREADS_BITS: usize = 10; +#[cfg(target_pointer_width = "64")] +const THREADS_BITS: usize = 16; + +#[cfg(target_pointer_width = "32")] +const THREADS_BITS: usize = 8; /// Bits to shift to select the sleeping threads /// (used with `select_bits`). +#[allow(clippy::erasing_op)] const SLEEPING_SHIFT: usize = 0 * THREADS_BITS; /// Bits to shift to select the inactive threads /// (used with `select_bits`). +#[allow(clippy::identity_op)] const INACTIVE_SHIFT: usize = 1 * THREADS_BITS; /// Bits to shift to select the JEC @@ -68,7 +74,7 @@ const JEC_SHIFT: usize = 2 * THREADS_BITS; /// Max value for the thread counters. -const THREADS_MAX: usize = (1 << THREADS_BITS) - 1; +pub(crate) const THREADS_MAX: usize = (1 << THREADS_BITS) - 1; /// Constant that can be added to add one sleeping thread. const ONE_SLEEPING: usize = 1;
diff --git a/src/sleep/mod.rs b/src/sleep/mod.rs index 2aa262c..af7225a 100644 --- a/src/sleep/mod.rs +++ b/src/sleep/mod.rs
@@ -11,6 +11,7 @@ use std::usize; mod counters; +pub(crate) use self::counters::THREADS_MAX; use self::counters::{AtomicCounters, JobsEventCounter}; /// The `Sleep` struct is embedded into each registry. It governs the waking and sleeping @@ -62,6 +63,7 @@ impl Sleep { pub(super) fn new(logger: Logger, n_threads: usize) -> Sleep { + assert!(n_threads <= THREADS_MAX); Sleep { logger, worker_sleep_states: (0..n_threads).map(|_| Default::default()).collect(), @@ -157,7 +159,7 @@ debug_assert!(!*is_blocked); // Our latch was signalled. We should wake back up fully as we - // wil have some stuff to do. + // will have some stuff to do. if !latch.fall_asleep() { self.logger.log(|| ThreadSleepInterruptedByLatch { worker: worker_index,
diff --git a/src/spawn/mod.rs b/src/spawn/mod.rs index 0006103..ae1f211 100644 --- a/src/spawn/mod.rs +++ b/src/spawn/mod.rs
@@ -73,7 +73,7 @@ F: FnOnce() + Send + 'static, { // We assert that this does not hold any references (we know - // this because of the `'static` bound in the inferface); + // this because of the `'static` bound in the interface); // moreover, we assert that the code below is not supposed to // be able to panic, and hence the data won't leak but will be // enqueued into some deque for later execution. @@ -91,19 +91,14 @@ // executed. This ref is decremented at the (*) below. registry.increment_terminate_count(); - Box::new(HeapJob::new({ - let registry = registry.clone(); + HeapJob::new({ + let registry = Arc::clone(registry); move || { - match unwind::halt_unwinding(func) { - Ok(()) => {} - Err(err) => { - registry.handle_panic(err); - } - } + registry.catch_unwind(func); registry.terminate(); // (*) permit registry to terminate now } - })) - .as_job_ref() + }) + .into_static_job_ref() } /// Fires off a task into the Rayon threadpool in the "static" or @@ -148,7 +143,7 @@ F: FnOnce() + Send + 'static, { // We assert that this does not hold any references (we know - // this because of the `'static` bound in the inferface); + // this because of the `'static` bound in the interface); // moreover, we assert that the code below is not supposed to // be able to panic, and hence the data won't leak but will be // enqueued into some deque for later execution.
diff --git a/src/spawn/test.rs b/src/spawn/test.rs index 9c59754..761fafc 100644 --- a/src/spawn/test.rs +++ b/src/spawn/test.rs
@@ -166,7 +166,7 @@ #[test] fn lifo_order() { - // In the absense of stealing, `spawn()` jobs on a thread will run in 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<i32> = (0..100).rev().collect(); // LIFO -> reversed assert_eq!(vec, expected); @@ -174,7 +174,7 @@ #[test] fn fifo_order() { - // In the absense of stealing, `spawn_fifo()` jobs on a thread will run in 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<i32> = (0..100).collect(); // FIFO -> natural order assert_eq!(vec, expected);
diff --git a/src/test.rs b/src/test.rs index 015d3ec..46d63a7 100644 --- a/src/test.rs +++ b/src/test.rs
@@ -1,7 +1,5 @@ #![cfg(test)] -#[allow(deprecated)] -use crate::Configuration; use crate::{ThreadPoolBuildError, ThreadPoolBuilder}; use std::sync::atomic::{AtomicUsize, Ordering}; use std::sync::{Arc, Barrier}; @@ -22,8 +20,8 @@ // Wait for all the threads in the pool plus the one running tests. let barrier = Arc::new(Barrier::new(n_threads + 1)); - let b = barrier.clone(); - let nc = n_called.clone(); + let b = Arc::clone(&barrier); + let nc = Arc::clone(&n_called); let start_handler = move |_| { nc.fetch_add(1, Ordering::SeqCst); b.wait(); @@ -48,8 +46,8 @@ // Wait for all the threads in the pool plus the one running tests. let barrier = Arc::new(Barrier::new(n_threads + 1)); - let b = barrier.clone(); - let nc = n_called.clone(); + let b = Arc::clone(&barrier); + let nc = Arc::clone(&n_called); let exit_handler = move |_| { nc.fetch_add(1, Ordering::SeqCst); b.wait(); @@ -85,9 +83,9 @@ panic!("ensure panic handler is called when exiting"); }; - let sb = start_barrier.clone(); - let eb = exit_barrier.clone(); - let nc = n_called.clone(); + 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 { @@ -121,7 +119,6 @@ } #[test] -#[allow(deprecated)] fn check_config_build() { let pool = ThreadPoolBuilder::new().num_threads(22).build().unwrap(); assert_eq!(pool.current_num_threads(), 22); @@ -144,7 +141,7 @@ let thread_name = move |i| format!("thread_name_{}", i); // Ensure we can call all public methods on Configuration - Configuration::new() + crate::Configuration::new() .thread_name(thread_name) .num_threads(5) .panic_handler(panic_handler)
diff --git a/src/thread_pool/mod.rs b/src/thread_pool/mod.rs index 5edaedc..0fc06dd 100644 --- a/src/thread_pool/mod.rs +++ b/src/thread_pool/mod.rs
@@ -3,12 +3,11 @@ //! //! [`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; -#[allow(deprecated)] -use crate::Configuration; use crate::{scope, Scope}; use crate::{scope_fifo, ScopeFifo}; use crate::{ThreadPoolBuildError, ThreadPoolBuilder}; @@ -57,7 +56,7 @@ #[deprecated(note = "Use `ThreadPoolBuilder::build`")] #[allow(deprecated)] /// Deprecated in favor of `ThreadPoolBuilder::build`. - pub fn new(configuration: Configuration) -> Result<ThreadPool, Box<dyn Error>> { + pub fn new(configuration: crate::Configuration) -> Result<ThreadPool, Box<dyn Error>> { Self::build(configuration.into_builder()).map_err(Box::from) } @@ -111,6 +110,57 @@ self.registry.in_worker(|_, _| op()) } + /// Executes `op` within every thread in the threadpool. Any attempts to use + /// `join`, `scope`, or parallel iterators will then operate within that + /// threadpool. + /// + /// Broadcasts are executed on each thread after they have exhausted their + /// local work queue, before they attempt work-stealing from other threads. + /// The goal of that strategy is to run everywhere in a timely manner + /// *without* being too disruptive to current work. There may be alternative + /// broadcast styles added in the future for more or less aggressive + /// injection, if the need arises. + /// + /// # Warning: thread-local data + /// + /// Because `op` is executing within the Rayon thread-pool, + /// thread-local data from the current thread will not be + /// accessible. + /// + /// # Panics + /// + /// If `op` should panic on one or more threads, exactly one panic + /// will be propagated, only after all threads have completed + /// (or panicked) their own `op`. + /// + /// # Examples + /// + /// ``` + /// # use rayon_core as rayon; + /// use std::sync::atomic::{AtomicUsize, Ordering}; + /// + /// fn main() { + /// let pool = rayon::ThreadPoolBuilder::new().num_threads(5).build().unwrap(); + /// + /// // The argument gives context, including the index of each thread. + /// let v: Vec<usize> = pool.broadcast(|ctx| ctx.index() * ctx.index()); + /// assert_eq!(v, &[0, 1, 4, 9, 16]); + /// + /// // The closure can reference the local stack + /// let count = AtomicUsize::new(0); + /// pool.broadcast(|_| count.fetch_add(1, Ordering::Relaxed)); + /// assert_eq!(count.into_inner(), 5); + /// } + /// ``` + pub fn broadcast<OP, R>(&self, op: OP) -> Vec<R> + where + OP: Fn(BroadcastContext<'_>) -> R + Sync, + R: Send, + { + // We assert that `self.registry` has not terminated. + unsafe { broadcast::broadcast_in(op, &self.registry) } + } + /// Returns the (current) number of threads in the thread pool. /// /// # Future compatibility note @@ -277,6 +327,18 @@ // We assert that `self.registry` has not terminated. unsafe { spawn::spawn_fifo_in(op, &self.registry) } } + + /// Spawns an asynchronous task on every thread in this thread-pool. This task + /// will run in the implicit, global scope, which means that it may outlast the + /// current stack frame -- therefore, it cannot capture any references onto the + /// stack (you will likely need a `move` closure). + pub fn spawn_broadcast<OP>(&self, op: OP) + where + OP: Fn(BroadcastContext<'_>) + Send + Sync + 'static, + { + // We assert that `self.registry` has not terminated. + unsafe { broadcast::spawn_broadcast_in(op, &self.registry) } + } } impl Drop for ThreadPool {
diff --git a/src/thread_pool/test.rs b/src/thread_pool/test.rs index 1510e37..ac750a6 100644 --- a/src/thread_pool/test.rs +++ b/src/thread_pool/test.rs
@@ -4,8 +4,6 @@ use std::sync::mpsc::channel; use std::sync::{Arc, Mutex}; -#[allow(deprecated)] -use crate::Configuration; use crate::{join, Scope, ScopeFifo, ThreadPool, ThreadPoolBuilder}; #[test] @@ -28,7 +26,7 @@ // do some work on these threads join_a_lot(22); - thread_pool.registry.clone() + Arc::clone(&thread_pool.registry) }); assert_eq!(registry.num_threads(), 22); } @@ -53,7 +51,7 @@ { // once we exit this block, thread-pool will be dropped let thread_pool = ThreadPoolBuilder::new().num_threads(22).build().unwrap(); - registry = thread_pool.registry.clone(); + 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)); @@ -67,7 +65,7 @@ /// Creates a start/exit handler that increments an atomic counter. fn count_handler() -> (Arc<AtomicUsize>, impl Fn(usize)) { let count = Arc::new(AtomicUsize::new(0)); - (count.clone(), move |_| { + (Arc::clone(&count), move |_| { count.fetch_add(1, Ordering::SeqCst); }) } @@ -197,7 +195,7 @@ #[test] #[allow(deprecated)] fn check_thread_pool_new() { - let pool = ThreadPool::new(Configuration::new().num_threads(22)).unwrap(); + let pool = ThreadPool::new(crate::Configuration::new().num_threads(22)).unwrap(); assert_eq!(pool.current_num_threads(), 22); }
diff --git a/tests/double_init_fail.rs b/tests/double_init_fail.rs index ea06bf0..b3ddbeb 100644 --- a/tests/double_init_fail.rs +++ b/tests/double_init_fail.rs
@@ -4,7 +4,7 @@ #[test] fn double_init_fail() { let result1 = ThreadPoolBuilder::new().build_global(); - assert_eq!(result1.unwrap(), ()); + assert!(result1.is_ok()); let err = ThreadPoolBuilder::new().build_global().unwrap_err(); assert!(err.source().is_none()); assert_eq!(