1use crate::job::*;
2use crate::registry::Registry;
3use crate::tlv::Tlv;
4use crate::unwind;
5use std::mem;
6use std::sync::Arc;
7
8/// Fires off a task into the Rayon threadpool in the "static" or
9/// "global" scope. Just like a standard thread, this task is not
10/// tied to the current stack frame, and hence it cannot hold any
11/// references other than those with `'static` lifetime. If you want
12/// to spawn a task that references stack data, use [the `scope()`
13/// function][scope] to create a scope.
14///
15/// [scope]: fn.scope.html
16///
17/// Since tasks spawned with this function cannot hold references into
18/// the enclosing stack frame, you almost certainly want to use a
19/// `move` closure as their argument (otherwise, the closure will
20/// typically hold references to any variables from the enclosing
21/// function that you happen to use).
22///
23/// This API assumes that the closure is executed purely for its
24/// side-effects (i.e., it might send messages, modify data protected
25/// by a mutex, or some such thing).
26///
27/// There is no guaranteed order of execution for spawns, given that
28/// other threads may steal tasks at any time. However, they are
29/// generally prioritized in a LIFO order on the thread from which
30/// they were spawned. Other threads always steal from the other end of
31/// the deque, like FIFO order. The idea is that "recent" tasks are
32/// most likely to be fresh in the local CPU's cache, while other
33/// threads can steal older "stale" tasks. For an alternate approach,
34/// consider [`spawn_fifo()`] instead.
35///
36/// [`spawn_fifo()`]: fn.spawn_fifo.html
37///
38/// # Panic handling
39///
40/// If this closure should panic, the resulting panic will be
41/// propagated to the panic handler registered in the `ThreadPoolBuilder`,
42/// if any. See [`ThreadPoolBuilder::panic_handler()`][ph] for more
43/// details.
44///
45/// [ph]: struct.ThreadPoolBuilder.html#method.panic_handler
46///
47/// # Examples
48///
49/// This code creates a Rayon task that increments a global counter.
50///
51/// ```rust
52/// # use rayon_core as rayon;
53/// use std::sync::atomic::{AtomicUsize, Ordering, ATOMIC_USIZE_INIT};
54///
55/// static GLOBAL_COUNTER: AtomicUsize = ATOMIC_USIZE_INIT;
56///
57/// rayon::spawn(move || {
58/// GLOBAL_COUNTER.fetch_add(1, Ordering::SeqCst);
59/// });
60/// ```
61pub fn spawn<F>(func: F)
62where
63 F: FnOnce() + Send + 'static,
64{
65 // We assert that current registry has not terminated.
66 unsafe { spawn_in(func, &Registry::current()) }
67}
68
69/// Spawns an asynchronous job in `registry.`
70///
71/// Unsafe because `registry` must not yet have terminated.
72pub(super) unsafe fn spawn_in<F>(func: F, registry: &Arc<Registry>)
73where
74 F: FnOnce() + Send + 'static,
75{
76 // We assert that this does not hold any references (we know
77 // this because of the `'static` bound in the interface);
78 // moreover, we assert that the code below is not supposed to
79 // be able to panic, and hence the data won't leak but will be
80 // enqueued into some deque for later execution.
81 let abort_guard: AbortIfPanic = unwind::AbortIfPanic; // just in case we are wrong, and code CAN panic
82 let job_ref: JobRef = spawn_job(func, registry);
83 registry.inject_or_push(job_ref);
84 mem::forget(abort_guard);
85}
86
87unsafe fn spawn_job<F>(func: F, registry: &Arc<Registry>) -> JobRef
88where
89 F: FnOnce() + Send + 'static,
90{
91 // Ensure that registry cannot terminate until this job has
92 // executed. This ref is decremented at the (*) below.
93 registry.increment_terminate_count();
94
95 HeapJobBox>::new(Tlv::null(), {
96 let registry: Arc = Arc::clone(self:registry);
97 move || {
98 registry.catch_unwind(func);
99 registry.terminate(); // (*) permit registry to terminate now
100 }
101 })
102 .into_static_job_ref()
103}
104
105/// Fires off a task into the Rayon threadpool in the "static" or
106/// "global" scope. Just like a standard thread, this task is not
107/// tied to the current stack frame, and hence it cannot hold any
108/// references other than those with `'static` lifetime. If you want
109/// to spawn a task that references stack data, use [the `scope_fifo()`
110/// function](fn.scope_fifo.html) to create a scope.
111///
112/// The behavior is essentially the same as [the `spawn`
113/// function](fn.spawn.html), except that calls from the same thread
114/// will be prioritized in FIFO order. This is similar to the now-
115/// deprecated [`breadth_first`] option, except the effect is isolated
116/// to relative `spawn_fifo` calls, not all threadpool tasks.
117///
118/// For more details on this design, see Rayon [RFC #1].
119///
120/// [`breadth_first`]: struct.ThreadPoolBuilder.html#method.breadth_first
121/// [RFC #1]: https://github.com/rayon-rs/rfcs/blob/master/accepted/rfc0001-scope-scheduling.md
122///
123/// # Panic handling
124///
125/// If this closure should panic, the resulting panic will be
126/// propagated to the panic handler registered in the `ThreadPoolBuilder`,
127/// if any. See [`ThreadPoolBuilder::panic_handler()`][ph] for more
128/// details.
129///
130/// [ph]: struct.ThreadPoolBuilder.html#method.panic_handler
131pub fn spawn_fifo<F>(func: F)
132where
133 F: FnOnce() + Send + 'static,
134{
135 // We assert that current registry has not terminated.
136 unsafe { spawn_fifo_in(func, &Registry::current()) }
137}
138
139/// Spawns an asynchronous FIFO job in `registry.`
140///
141/// Unsafe because `registry` must not yet have terminated.
142pub(super) unsafe fn spawn_fifo_in<F>(func: F, registry: &Arc<Registry>)
143where
144 F: FnOnce() + Send + 'static,
145{
146 // We assert that this does not hold any references (we know
147 // this because of the `'static` bound in the interface);
148 // moreover, we assert that the code below is not supposed to
149 // be able to panic, and hence the data won't leak but will be
150 // enqueued into some deque for later execution.
151 let abort_guard: AbortIfPanic = unwind::AbortIfPanic; // just in case we are wrong, and code CAN panic
152 let job_ref: JobRef = spawn_job(func, registry);
153
154 // If we're in the pool, use our thread's private fifo for this thread to execute
155 // in a locally-FIFO order. Otherwise, just use the pool's global injector.
156 match registry.current_thread() {
157 Some(worker: &WorkerThread) => worker.push_fifo(job_ref),
158 None => registry.inject(injected_job:job_ref),
159 }
160 mem::forget(abort_guard);
161}
162
163#[cfg(test)]
164mod test;
165