diff options
Diffstat (limited to 'vendor/rayon-core/src/spawn')
| -rw-r--r-- | vendor/rayon-core/src/spawn/mod.rs | 163 | ||||
| -rw-r--r-- | vendor/rayon-core/src/spawn/test.rs | 255 |
2 files changed, 0 insertions, 418 deletions
diff --git a/vendor/rayon-core/src/spawn/mod.rs b/vendor/rayon-core/src/spawn/mod.rs deleted file mode 100644 index 1aa9edb..0000000 --- a/vendor/rayon-core/src/spawn/mod.rs +++ /dev/null @@ -1,163 +0,0 @@ -use crate::job::*; -use crate::registry::Registry; -use crate::unwind; -use std::mem; -use std::sync::Arc; - -/// Fires off a task into the Rayon threadpool in the "static" or -/// "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()` -/// function][scope] to create a scope. -/// -/// [scope]: fn.scope.html -/// -/// Since tasks spawned with this function cannot hold references into -/// the enclosing stack frame, you almost certainly want to use a -/// `move` closure as their argument (otherwise, the closure will -/// typically hold references to any variables from the enclosing -/// function that you happen to use). -/// -/// This API assumes that the closure is executed purely for its -/// side-effects (i.e., it might send messages, modify data protected -/// by a mutex, or some such thing). -/// -/// There is no guaranteed order of execution for spawns, given that -/// 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 -/// most likely to be fresh in the local CPU's cache, while other -/// threads can steal older "stale" tasks. For an alternate approach, -/// consider [`spawn_fifo()`] instead. -/// -/// [`spawn_fifo()`]: fn.spawn_fifo.html -/// -/// # Panic handling -/// -/// 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 -/// details. -/// -/// [ph]: struct.ThreadPoolBuilder.html#method.panic_handler -/// -/// # Examples -/// -/// This code creates a Rayon task that increments a global counter. -/// -/// ```rust -/// # use rayon_core as rayon; -/// use std::sync::atomic::{AtomicUsize, Ordering, ATOMIC_USIZE_INIT}; -/// -/// static GLOBAL_COUNTER: AtomicUsize = ATOMIC_USIZE_INIT; -/// -/// rayon::spawn(move || { -/// GLOBAL_COUNTER.fetch_add(1, Ordering::SeqCst); -/// }); -/// ``` -pub fn spawn<F>(func: F) -where - F: FnOnce() + Send + 'static, -{ - // We assert that current registry has not terminated. - unsafe { spawn_in(func, &Registry::current()) } -} - -/// Spawns an asynchronous job in `registry.` -/// -/// Unsafe because `registry` must not yet have terminated. -pub(super) unsafe fn spawn_in<F>(func: F, registry: &Arc<Registry>) -where - F: FnOnce() + Send + 'static, -{ - // We assert that this does not hold any references (we know - // 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. - let abort_guard = unwind::AbortIfPanic; // just in case we are wrong, and code CAN panic - let job_ref = spawn_job(func, registry); - registry.inject_or_push(job_ref); - mem::forget(abort_guard); -} - -unsafe fn spawn_job<F>(func: F, registry: &Arc<Registry>) -> JobRef -where - F: FnOnce() + Send + 'static, -{ - // Ensure that registry cannot terminate until this job has - // executed. This ref is decremented at the (*) below. - registry.increment_terminate_count(); - - HeapJob::new({ - let registry = Arc::clone(registry); - move || { - registry.catch_unwind(func); - registry.terminate(); // (*) permit registry to terminate now - } - }) - .into_static_job_ref() -} - -/// Fires off a task into the Rayon threadpool in the "static" or -/// "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()` -/// function](fn.scope_fifo.html) to create a scope. -/// -/// The behavior is essentially the same as [the `spawn` -/// function](fn.spawn.html), except that calls from the same thread -/// will be prioritized in FIFO order. This is similar to the now- -/// deprecated [`breadth_first`] option, except the effect is isolated -/// to relative `spawn_fifo` calls, not all threadpool tasks. -/// -/// For more details on this design, see Rayon [RFC #1]. -/// -/// [`breadth_first`]: struct.ThreadPoolBuilder.html#method.breadth_first -/// [RFC #1]: https://github.com/rayon-rs/rfcs/blob/master/accepted/rfc0001-scope-scheduling.md -/// -/// # Panic handling -/// -/// 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 -/// details. -/// -/// [ph]: struct.ThreadPoolBuilder.html#method.panic_handler -pub fn spawn_fifo<F>(func: F) -where - F: FnOnce() + Send + 'static, -{ - // We assert that current registry has not terminated. - unsafe { spawn_fifo_in(func, &Registry::current()) } -} - -/// Spawns an asynchronous FIFO job in `registry.` -/// -/// Unsafe because `registry` must not yet have terminated. -pub(super) unsafe fn spawn_fifo_in<F>(func: F, registry: &Arc<Registry>) -where - F: FnOnce() + Send + 'static, -{ - // We assert that this does not hold any references (we know - // 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. - let abort_guard = unwind::AbortIfPanic; // just in case we are wrong, and code CAN panic - 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. - match registry.current_thread() { - Some(worker) => worker.push_fifo(job_ref), - None => registry.inject(job_ref), - } - mem::forget(abort_guard); -} - -#[cfg(test)] -mod test; diff --git a/vendor/rayon-core/src/spawn/test.rs b/vendor/rayon-core/src/spawn/test.rs deleted file mode 100644 index b7a0535..0000000 --- a/vendor/rayon-core/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<dyn Any + Send>| { - 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<dyn Any + Send>| { - 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::<Vec<i32>>() - }}; -} - -#[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<i32> = (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<i32> = (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<i32> = (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<i32> = (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::<Vec<i32>>() - }}; -} - -#[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); -} |