rayon_core/lib.rs
1//! Rayon-core houses the core stable APIs of Rayon.
2//!
3//! These APIs have been mirrored in the Rayon crate and it is recommended to use these from there.
4//!
5//! [`join()`] is used to take two closures and potentially run them in parallel.
6//! - It will run in parallel if task B gets stolen before task A can finish.
7//! - It will run sequentially if task A finishes before task B is stolen and can continue on task B.
8//!
9//! [`scope()`] creates a scope in which you can run any number of parallel tasks.
10//! These tasks can spawn nested tasks and scopes, but given the nature of work stealing, the order of execution can not be guaranteed.
11//! The scope will exist until all tasks spawned within the scope have been completed.
12//!
13//! [`spawn()`] add a task into the 'static' or 'global' scope, or a local scope created by the [`scope()`] function.
14//!
15//! [`ThreadPool`] can be used to create your own thread pools (using [`ThreadPoolBuilder`]) or to customize the global one.
16//! Tasks spawned within the pool (using [`install()`][tpinstall], [`join()`][tpjoin], etc.) will be added to a deque,
17//! where it becomes available for work stealing from other threads in the local thread pool.
18//!
19//! [tpinstall]: ThreadPool::install()
20//! [tpjoin]: ThreadPool::join()
21//!
22//! # Global fallback when threading is unsupported
23//!
24//! Rayon uses `std` APIs for threading, but some targets have incomplete implementations that
25//! always return `Unsupported` errors. The WebAssembly `wasm32-unknown-unknown` and `wasm32-wasi`
26//! targets are notable examples of this. Rather than panicking on the unsupported error when
27//! creating the implicit global thread pool, Rayon configures a fallback mode instead.
28//!
29//! This fallback mode mostly functions as if it were using a single-threaded "pool", like setting
30//! `RAYON_NUM_THREADS=1`. For example, `join` will execute its two closures sequentially, since
31//! there is no other thread to share the work. However, since the pool is not running independent
32//! of the main thread, non-blocking calls like `spawn` may not execute at all, unless a lower-
33//! priority call like `broadcast` gives them an opening. The fallback mode does not try to emulate
34//! anything like thread preemption or `async` task switching, but `yield_now` or `yield_local`
35//! can also volunteer execution time.
36//!
37//! Explicit `ThreadPoolBuilder` methods always report their error without any fallback.
38//!
39//! # Restricting multiple versions
40//!
41//! In order to ensure proper coordination between thread pools, and especially
42//! to make sure there's only one global thread pool, `rayon-core` is actively
43//! restricted from building multiple versions of itself into a single target.
44//! You may see a build error like this in violation:
45//!
46//! ```text
47//! error: native library `rayon-core` is being linked to by more
48//! than one package, and can only be linked to by one package
49//! ```
50//!
51//! While we strive to keep `rayon-core` semver-compatible, it's still
52//! possible to arrive at this situation if different crates have overly
53//! restrictive tilde or inequality requirements for `rayon-core`. The
54//! conflicting requirements will need to be resolved before the build will
55//! succeed.
56
57#![deny(missing_debug_implementations)]
58#![deny(missing_docs)]
59#![deny(unreachable_pub)]
60#![warn(rust_2018_idioms)]
61
62use std::any::Any;
63use std::env;
64use std::error::Error;
65use std::fmt;
66use std::io;
67use std::marker::PhantomData;
68use std::str::FromStr;
69use std::thread;
70
71#[macro_use]
72mod private;
73
74mod broadcast;
75mod job;
76mod join;
77mod latch;
78mod registry;
79mod scope;
80mod sleep;
81mod spawn;
82mod thread_pool;
83mod unwind;
84
85mod compile_fail;
86mod test;
87
88pub use self::broadcast::{broadcast, spawn_broadcast, BroadcastContext};
89pub use self::join::{join, join_context};
90pub use self::registry::ThreadBuilder;
91pub use self::scope::{in_place_scope, scope, Scope};
92pub use self::scope::{in_place_scope_fifo, scope_fifo, ScopeFifo};
93pub use self::spawn::{spawn, spawn_fifo};
94pub use self::thread_pool::current_thread_has_pending_tasks;
95pub use self::thread_pool::current_thread_index;
96pub use self::thread_pool::ThreadPool;
97pub use self::thread_pool::{yield_local, yield_now, Yield};
98
99#[cfg(not(feature = "web_spin_lock"))]
100use std::sync;
101
102#[cfg(feature = "web_spin_lock")]
103use wasm_sync as sync;
104
105use self::registry::{CustomSpawn, DefaultSpawn, ThreadSpawn};
106
107/// Returns the maximum number of threads that Rayon supports in a single thread pool.
108///
109/// If a higher thread count is requested by calling `ThreadPoolBuilder::num_threads` or by setting
110/// the `RAYON_NUM_THREADS` environment variable, then it will be reduced to this maximum.
111///
112/// The value may vary between different targets, and is subject to change in new Rayon versions.
113pub fn max_num_threads() -> usize {
114 // We are limited by the bits available in the sleep counter's `AtomicUsize`.
115 crate::sleep::THREADS_MAX
116}
117
118/// Returns the number of threads in the current registry. If this
119/// code is executing within a Rayon thread pool, then this will be
120/// the number of threads for the thread pool of the current
121/// thread. Otherwise, it will be the number of threads for the global
122/// thread pool.
123///
124/// This can be useful when trying to judge how many times to split
125/// parallel work (the parallel iterator traits use this value
126/// internally for this purpose).
127///
128/// # Future compatibility note
129///
130/// Note that unless this thread pool was created with a
131/// builder that specifies the number of threads, then this
132/// number may vary over time in future versions (see [the
133/// `num_threads()` method for details][snt]).
134///
135/// [snt]: ThreadPoolBuilder::num_threads
136pub fn current_num_threads() -> usize {
137 crate::registry::Registry::current_num_threads()
138}
139
140/// Error when initializing a thread pool.
141#[derive(Debug)]
142pub struct ThreadPoolBuildError {
143 kind: ErrorKind,
144}
145
146#[derive(Debug)]
147enum ErrorKind {
148 GlobalPoolAlreadyInitialized,
149 CurrentThreadAlreadyInPool,
150 IOError(io::Error),
151}
152
153/// Used to create a new [`ThreadPool`] or to configure the global rayon thread pool.
154/// ## Creating a ThreadPool
155/// The following creates a thread pool with 22 threads.
156///
157/// ```ignore-wasm
158/// # use rayon_core as rayon;
159/// let pool = rayon::ThreadPoolBuilder::new().num_threads(22).build().unwrap();
160/// ```
161///
162/// To instead configure the global thread pool, use [`build_global()`]:
163///
164/// ```ignore-wasm
165/// # use rayon_core as rayon;
166/// rayon::ThreadPoolBuilder::new().num_threads(22).build_global().unwrap();
167/// ```
168///
169/// [`build_global()`]: Self::build_global()
170pub struct ThreadPoolBuilder<S = DefaultSpawn> {
171 /// The number of threads in the rayon thread pool.
172 /// If zero will use the RAYON_NUM_THREADS environment variable.
173 /// If RAYON_NUM_THREADS is invalid or zero will use the default.
174 num_threads: usize,
175
176 /// The thread we're building *from* will also be part of the pool.
177 use_current_thread: bool,
178
179 /// Custom closure, if any, to handle a panic that we cannot propagate
180 /// anywhere else.
181 panic_handler: Option<Box<PanicHandler>>,
182
183 /// Closure to compute the name of a thread.
184 get_thread_name: Option<Box<dyn FnMut(usize) -> String>>,
185
186 /// The stack size for the created worker threads
187 stack_size: Option<usize>,
188
189 /// Closure invoked on worker-thread start.
190 start_handler: Option<Box<StartHandler>>,
191
192 /// Closure invoked on worker-thread exit.
193 exit_handler: Option<Box<ExitHandler>>,
194
195 /// Closure invoked to spawn threads.
196 spawn_handler: S,
197
198 /// If false, worker threads will execute spawned jobs in a
199 /// "depth-first" fashion. If true, they will do a "breadth-first"
200 /// fashion. Depth-first is the default.
201 breadth_first: bool,
202}
203
204/// Contains the rayon thread pool configuration. Use [`ThreadPoolBuilder`] instead.
205#[deprecated(note = "Use `ThreadPoolBuilder`")]
206#[derive(Default)]
207pub struct Configuration {
208 builder: ThreadPoolBuilder,
209}
210
211/// The type for a panic-handling closure. Note that this same closure
212/// may be invoked multiple times in parallel.
213type PanicHandler = dyn Fn(Box<dyn Any + Send>) + Send + Sync;
214
215/// The type for a closure that gets invoked when a thread starts. The
216/// closure is passed the index of the thread on which it is invoked.
217/// Note that this same closure may be invoked multiple times in parallel.
218type StartHandler = dyn Fn(usize) + Send + Sync;
219
220/// The type for a closure that gets invoked when a thread exits. The
221/// closure is passed the index of the thread on which it is invoked.
222/// Note that this same closure may be invoked multiple times in parallel.
223type ExitHandler = dyn Fn(usize) + Send + Sync;
224
225// NB: We can't `#[derive(Default)]` because `S` is left ambiguous.
226impl Default for ThreadPoolBuilder {
227 fn default() -> Self {
228 ThreadPoolBuilder {
229 num_threads: 0,
230 use_current_thread: false,
231 panic_handler: None,
232 get_thread_name: None,
233 stack_size: None,
234 start_handler: None,
235 exit_handler: None,
236 spawn_handler: DefaultSpawn,
237 breadth_first: false,
238 }
239 }
240}
241
242impl ThreadPoolBuilder {
243 /// Creates and returns a valid rayon thread pool builder, but does not initialize it.
244 pub fn new() -> Self {
245 Self::default()
246 }
247}
248
249/// Note: the `S: ThreadSpawn` constraint is an internal implementation detail for the
250/// default spawn and those set by [`spawn_handler`](#method.spawn_handler).
251impl<S> ThreadPoolBuilder<S>
252where
253 S: ThreadSpawn,
254{
255 /// Creates a new `ThreadPool` initialized using this configuration.
256 pub fn build(self) -> Result<ThreadPool, ThreadPoolBuildError> {
257 ThreadPool::build(self)
258 }
259
260 /// Initializes the global thread pool. This initialization is
261 /// **optional**. If you do not call this function, the thread pool
262 /// will be automatically initialized with the default
263 /// configuration. Calling `build_global` is not recommended, except
264 /// in two scenarios:
265 ///
266 /// - You wish to change the default configuration.
267 /// - You are running a benchmark, in which case initializing may
268 /// yield slightly more consistent results, since the worker threads
269 /// will already be ready to go even in the first iteration. But
270 /// this cost is minimal.
271 ///
272 /// Initialization of the global thread pool happens exactly
273 /// once. Once started, the configuration cannot be
274 /// changed. Therefore, if you call `build_global` a second time, it
275 /// will return an error. An `Ok` result indicates that this
276 /// is the first initialization of the thread pool.
277 pub fn build_global(self) -> Result<(), ThreadPoolBuildError> {
278 let registry = registry::init_global_registry(self)?;
279 registry.wait_until_primed();
280 Ok(())
281 }
282}
283
284impl ThreadPoolBuilder {
285 /// Creates a scoped `ThreadPool` initialized using this configuration.
286 ///
287 /// This is a convenience function for building a pool using [`std::thread::scope`]
288 /// to spawn threads in a [`spawn_handler`].
289 /// The threads in this pool will start by calling `wrapper`, which should
290 /// do initialization and continue by calling `ThreadBuilder::run()`.
291 ///
292 /// [`spawn_handler`]: Self::spawn_handler()
293 ///
294 /// # Examples
295 ///
296 /// A scoped pool may be useful in combination with scoped thread-local variables.
297 ///
298 /// ```ignore-wasm
299 /// # use rayon_core as rayon;
300 ///
301 /// scoped_tls::scoped_thread_local!(static POOL_DATA: Vec<i32>);
302 ///
303 /// fn main() -> Result<(), rayon::ThreadPoolBuildError> {
304 /// let pool_data = vec![1, 2, 3];
305 ///
306 /// // We haven't assigned any TLS data yet.
307 /// assert!(!POOL_DATA.is_set());
308 ///
309 /// rayon::ThreadPoolBuilder::new()
310 /// .build_scoped(
311 /// // Borrow `pool_data` in TLS for each thread.
312 /// |thread| POOL_DATA.set(&pool_data, || thread.run()),
313 /// // Do some work that needs the TLS data.
314 /// |pool| pool.install(|| assert!(POOL_DATA.is_set())),
315 /// )?;
316 ///
317 /// // Once we've returned, `pool_data` is no longer borrowed.
318 /// drop(pool_data);
319 /// Ok(())
320 /// }
321 /// ```
322 pub fn build_scoped<W, F, R>(self, wrapper: W, with_pool: F) -> Result<R, ThreadPoolBuildError>
323 where
324 W: Fn(ThreadBuilder) + Sync, // expected to call `run()`
325 F: FnOnce(&ThreadPool) -> R,
326 {
327 std::thread::scope(|scope| {
328 let pool = self
329 .spawn_handler(|thread| {
330 let mut builder = std::thread::Builder::new();
331 if let Some(name) = thread.name() {
332 builder = builder.name(name.to_string());
333 }
334 if let Some(size) = thread.stack_size() {
335 builder = builder.stack_size(size);
336 }
337 builder.spawn_scoped(scope, || wrapper(thread))?;
338 Ok(())
339 })
340 .build()?;
341 Ok(with_pool(&pool))
342 })
343 }
344}
345
346impl<S> ThreadPoolBuilder<S> {
347 /// Sets a custom function for spawning threads.
348 ///
349 /// Note that the threads will not exit until after the pool is dropped. It
350 /// is up to the caller to wait for thread termination if that is important
351 /// for any invariants. For instance, threads created in [`std::thread::scope`]
352 /// will be joined before that scope returns, and this will block indefinitely
353 /// if the pool is leaked. Furthermore, the global thread pool doesn't terminate
354 /// until the entire process exits!
355 ///
356 /// # Examples
357 ///
358 /// A minimal spawn handler just needs to call `run()` from an independent thread.
359 ///
360 /// ```ignore-wasm
361 /// # use rayon_core as rayon;
362 /// fn main() -> Result<(), rayon::ThreadPoolBuildError> {
363 /// let pool = rayon::ThreadPoolBuilder::new()
364 /// .spawn_handler(|thread| {
365 /// std::thread::spawn(|| thread.run());
366 /// Ok(())
367 /// })
368 /// .build()?;
369 ///
370 /// pool.install(|| println!("Hello from my custom thread!"));
371 /// Ok(())
372 /// }
373 /// ```
374 ///
375 /// The default spawn handler sets the name and stack size if given, and propagates
376 /// any errors from the thread builder.
377 ///
378 /// ```ignore-wasm
379 /// # use rayon_core as rayon;
380 /// fn main() -> Result<(), rayon::ThreadPoolBuildError> {
381 /// let pool = rayon::ThreadPoolBuilder::new()
382 /// .spawn_handler(|thread| {
383 /// let mut b = std::thread::Builder::new();
384 /// if let Some(name) = thread.name() {
385 /// b = b.name(name.to_owned());
386 /// }
387 /// if let Some(stack_size) = thread.stack_size() {
388 /// b = b.stack_size(stack_size);
389 /// }
390 /// b.spawn(|| thread.run())?;
391 /// Ok(())
392 /// })
393 /// .build()?;
394 ///
395 /// pool.install(|| println!("Hello from my fully custom thread!"));
396 /// Ok(())
397 /// }
398 /// ```
399 ///
400 /// This can also be used for a pool of scoped threads like [`crossbeam::scope`],
401 /// or [`std::thread::scope`] introduced in Rust 1.63, which is encapsulated in
402 /// [`build_scoped`].
403 ///
404 /// [`crossbeam::scope`]: https://docs.rs/crossbeam/0.8/crossbeam/fn.scope.html
405 /// [`build_scoped`]: Self::build_scoped()
406 ///
407 /// ```ignore-wasm
408 /// # use rayon_core as rayon;
409 /// fn main() -> Result<(), rayon::ThreadPoolBuildError> {
410 /// std::thread::scope(|scope| {
411 /// let pool = rayon::ThreadPoolBuilder::new()
412 /// .spawn_handler(|thread| {
413 /// let mut builder = std::thread::Builder::new();
414 /// if let Some(name) = thread.name() {
415 /// builder = builder.name(name.to_string());
416 /// }
417 /// if let Some(size) = thread.stack_size() {
418 /// builder = builder.stack_size(size);
419 /// }
420 /// builder.spawn_scoped(scope, || {
421 /// // Add any scoped initialization here, then run!
422 /// thread.run()
423 /// })?;
424 /// Ok(())
425 /// })
426 /// .build()?;
427 ///
428 /// pool.install(|| println!("Hello from my custom scoped thread!"));
429 /// Ok(())
430 /// })
431 /// }
432 /// ```
433 pub fn spawn_handler<F>(self, spawn: F) -> ThreadPoolBuilder<CustomSpawn<F>>
434 where
435 F: FnMut(ThreadBuilder) -> io::Result<()>,
436 {
437 ThreadPoolBuilder {
438 spawn_handler: CustomSpawn::new(spawn),
439 // ..self
440 num_threads: self.num_threads,
441 use_current_thread: self.use_current_thread,
442 panic_handler: self.panic_handler,
443 get_thread_name: self.get_thread_name,
444 stack_size: self.stack_size,
445 start_handler: self.start_handler,
446 exit_handler: self.exit_handler,
447 breadth_first: self.breadth_first,
448 }
449 }
450
451 /// Returns a reference to the current spawn handler.
452 fn get_spawn_handler(&mut self) -> &mut S {
453 &mut self.spawn_handler
454 }
455
456 /// Get the number of threads that will be used for the thread
457 /// pool. See `num_threads()` for more information.
458 fn get_num_threads(&self) -> usize {
459 if self.num_threads > 0 {
460 self.num_threads
461 } else {
462 let default = || {
463 thread::available_parallelism()
464 .map(|n| n.get())
465 .unwrap_or(1)
466 };
467
468 match env::var("RAYON_NUM_THREADS")
469 .ok()
470 .and_then(|s| usize::from_str(&s).ok())
471 {
472 Some(x @ 1..) => return x,
473 Some(0) => return default(),
474 _ => {}
475 }
476
477 // Support for deprecated `RAYON_RS_NUM_CPUS`.
478 match env::var("RAYON_RS_NUM_CPUS")
479 .ok()
480 .and_then(|s| usize::from_str(&s).ok())
481 {
482 Some(x @ 1..) => x,
483 _ => default(),
484 }
485 }
486 }
487
488 /// Get the thread name for the thread with the given index.
489 fn get_thread_name(&mut self, index: usize) -> Option<String> {
490 let f = self.get_thread_name.as_mut()?;
491 Some(f(index))
492 }
493
494 /// Sets a closure which takes a thread index and returns
495 /// the thread's name.
496 pub fn thread_name<F>(mut self, closure: F) -> Self
497 where
498 F: FnMut(usize) -> String + 'static,
499 {
500 self.get_thread_name = Some(Box::new(closure));
501 self
502 }
503
504 /// Sets the number of threads to be used in the rayon thread pool.
505 ///
506 /// If you specify a non-zero number of threads using this
507 /// function, then the resulting thread pools are guaranteed to
508 /// start at most this number of threads.
509 ///
510 /// If `num_threads` is 0, or you do not call this function, then
511 /// the Rayon runtime will select the number of threads
512 /// automatically. At present, this is based on the
513 /// `RAYON_NUM_THREADS` environment variable (if set),
514 /// or the number of logical CPUs (otherwise).
515 /// In the future, however, the default behavior may
516 /// change to dynamically add or remove threads as needed.
517 ///
518 /// **Future compatibility warning:** Given the default behavior
519 /// may change in the future, if you wish to rely on a fixed
520 /// number of threads, you should use this function to specify
521 /// that number. To reproduce the current default behavior, you
522 /// may wish to use [`std::thread::available_parallelism`]
523 /// to query the number of CPUs dynamically.
524 ///
525 /// **Old environment variable:** `RAYON_NUM_THREADS` is a one-to-one
526 /// replacement of the now deprecated `RAYON_RS_NUM_CPUS` environment
527 /// variable. If both variables are specified, `RAYON_NUM_THREADS` will
528 /// be preferred.
529 pub fn num_threads(mut self, num_threads: usize) -> Self {
530 self.num_threads = num_threads;
531 self
532 }
533
534 /// Use the current thread as one of the threads in the pool.
535 ///
536 /// The current thread is guaranteed to be at index 0, and since the thread is not managed by
537 /// rayon, the spawn and exit handlers do not run for that thread.
538 ///
539 /// Note that the current thread won't run the main work-stealing loop, so jobs spawned into
540 /// the thread pool will generally not be picked up automatically by this thread unless you
541 /// yield to rayon in some way, like via [`yield_now()`], [`yield_local()`], or [`scope()`].
542 ///
543 /// # Local thread pools
544 ///
545 /// Using this in a local thread pool means the registry will be leaked. In future versions
546 /// there might be a way of cleaning up the current-thread state.
547 pub fn use_current_thread(mut self) -> Self {
548 self.use_current_thread = true;
549 self
550 }
551
552 /// Returns a copy of the current panic handler.
553 fn take_panic_handler(&mut self) -> Option<Box<PanicHandler>> {
554 self.panic_handler.take()
555 }
556
557 /// Normally, whenever Rayon catches a panic, it tries to
558 /// propagate it to someplace sensible, to try and reflect the
559 /// semantics of sequential execution. But in some cases,
560 /// particularly with the `spawn()` APIs, there is no
561 /// obvious place where we should propagate the panic to.
562 /// In that case, this panic handler is invoked.
563 ///
564 /// If no panic handler is set, the default is to abort the
565 /// process, under the principle that panics should not go
566 /// unobserved.
567 ///
568 /// If the panic handler itself panics, this will abort the
569 /// process. To prevent this, wrap the body of your panic handler
570 /// in a call to `std::panic::catch_unwind()`.
571 pub fn panic_handler<H>(mut self, panic_handler: H) -> Self
572 where
573 H: Fn(Box<dyn Any + Send>) + Send + Sync + 'static,
574 {
575 self.panic_handler = Some(Box::new(panic_handler));
576 self
577 }
578
579 /// Get the stack size of the worker threads
580 fn get_stack_size(&self) -> Option<usize> {
581 self.stack_size
582 }
583
584 /// Sets the stack size of the worker threads
585 pub fn stack_size(mut self, stack_size: usize) -> Self {
586 self.stack_size = Some(stack_size);
587 self
588 }
589
590 /// **(DEPRECATED)** Suggest to worker threads that they execute
591 /// spawned jobs in a "breadth-first" fashion.
592 ///
593 /// Typically, when a worker thread is idle or blocked, it will
594 /// attempt to execute the job from the *top* of its local deque of
595 /// work (i.e., the job most recently spawned). If this flag is set
596 /// to true, however, workers will prefer to execute in a
597 /// *breadth-first* fashion -- that is, they will search for jobs at
598 /// the *bottom* of their local deque. (At present, workers *always*
599 /// steal from the bottom of other workers' deques, regardless of
600 /// the setting of this flag.)
601 ///
602 /// If you think of the tasks as a tree, where a parent task
603 /// spawns its children in the tree, then this flag loosely
604 /// corresponds to doing a breadth-first traversal of the tree,
605 /// whereas the default would be to do a depth-first traversal.
606 ///
607 /// **Note that this is an "execution hint".** Rayon's task
608 /// execution is highly dynamic and the precise order in which
609 /// independent tasks are executed is not intended to be
610 /// guaranteed.
611 ///
612 /// This `breadth_first()` method is now deprecated per [RFC #1],
613 /// and in the future its effect may be removed. Consider using
614 /// [`scope_fifo()`] for a similar effect.
615 ///
616 /// [RFC #1]: https://github.com/rayon-rs/rfcs/blob/main/accepted/rfc0001-scope-scheduling.md
617 #[deprecated(note = "use `scope_fifo` and `spawn_fifo` for similar effect")]
618 pub fn breadth_first(mut self) -> Self {
619 self.breadth_first = true;
620 self
621 }
622
623 fn get_breadth_first(&self) -> bool {
624 self.breadth_first
625 }
626
627 /// Takes the current thread start callback, leaving `None`.
628 fn take_start_handler(&mut self) -> Option<Box<StartHandler>> {
629 self.start_handler.take()
630 }
631
632 /// Sets a callback to be invoked on thread start.
633 ///
634 /// The closure is passed the index of the thread on which it is invoked.
635 /// Note that this same closure may be invoked multiple times in parallel.
636 /// If this closure panics, the panic will be passed to the panic handler.
637 /// If that handler returns, then startup will continue normally.
638 pub fn start_handler<H>(mut self, start_handler: H) -> Self
639 where
640 H: Fn(usize) + Send + Sync + 'static,
641 {
642 self.start_handler = Some(Box::new(start_handler));
643 self
644 }
645
646 /// Returns a current thread exit callback, leaving `None`.
647 fn take_exit_handler(&mut self) -> Option<Box<ExitHandler>> {
648 self.exit_handler.take()
649 }
650
651 /// Sets a callback to be invoked on thread exit.
652 ///
653 /// The closure is passed the index of the thread on which it is invoked.
654 /// Note that this same closure may be invoked multiple times in parallel.
655 /// If this closure panics, the panic will be passed to the panic handler.
656 /// If that handler returns, then the thread will exit normally.
657 pub fn exit_handler<H>(mut self, exit_handler: H) -> Self
658 where
659 H: Fn(usize) + Send + Sync + 'static,
660 {
661 self.exit_handler = Some(Box::new(exit_handler));
662 self
663 }
664}
665
666#[allow(deprecated)]
667impl Configuration {
668 /// Creates and return a valid rayon thread pool configuration, but does not initialize it.
669 pub fn new() -> Configuration {
670 Configuration {
671 builder: ThreadPoolBuilder::new(),
672 }
673 }
674
675 /// Deprecated in favor of `ThreadPoolBuilder::build`.
676 pub fn build(self) -> Result<ThreadPool, Box<dyn Error + 'static>> {
677 self.builder.build().map_err(Box::from)
678 }
679
680 /// Deprecated in favor of `ThreadPoolBuilder::thread_name`.
681 pub fn thread_name<F>(mut self, closure: F) -> Self
682 where
683 F: FnMut(usize) -> String + 'static,
684 {
685 self.builder = self.builder.thread_name(closure);
686 self
687 }
688
689 /// Deprecated in favor of `ThreadPoolBuilder::num_threads`.
690 pub fn num_threads(mut self, num_threads: usize) -> Configuration {
691 self.builder = self.builder.num_threads(num_threads);
692 self
693 }
694
695 /// Deprecated in favor of `ThreadPoolBuilder::panic_handler`.
696 pub fn panic_handler<H>(mut self, panic_handler: H) -> Configuration
697 where
698 H: Fn(Box<dyn Any + Send>) + Send + Sync + 'static,
699 {
700 self.builder = self.builder.panic_handler(panic_handler);
701 self
702 }
703
704 /// Deprecated in favor of `ThreadPoolBuilder::stack_size`.
705 pub fn stack_size(mut self, stack_size: usize) -> Self {
706 self.builder = self.builder.stack_size(stack_size);
707 self
708 }
709
710 /// Deprecated in favor of `ThreadPoolBuilder::breadth_first`.
711 pub fn breadth_first(mut self) -> Self {
712 self.builder = self.builder.breadth_first();
713 self
714 }
715
716 /// Deprecated in favor of `ThreadPoolBuilder::start_handler`.
717 pub fn start_handler<H>(mut self, start_handler: H) -> Configuration
718 where
719 H: Fn(usize) + Send + Sync + 'static,
720 {
721 self.builder = self.builder.start_handler(start_handler);
722 self
723 }
724
725 /// Deprecated in favor of `ThreadPoolBuilder::exit_handler`.
726 pub fn exit_handler<H>(mut self, exit_handler: H) -> Configuration
727 where
728 H: Fn(usize) + Send + Sync + 'static,
729 {
730 self.builder = self.builder.exit_handler(exit_handler);
731 self
732 }
733
734 /// Returns a ThreadPoolBuilder with identical parameters.
735 fn into_builder(self) -> ThreadPoolBuilder {
736 self.builder
737 }
738}
739
740impl ThreadPoolBuildError {
741 fn new(kind: ErrorKind) -> ThreadPoolBuildError {
742 ThreadPoolBuildError { kind }
743 }
744
745 fn is_unsupported(&self) -> bool {
746 matches!(&self.kind, ErrorKind::IOError(e) if e.kind() == io::ErrorKind::Unsupported)
747 }
748}
749
750const GLOBAL_POOL_ALREADY_INITIALIZED: &str =
751 "The global thread pool has already been initialized.";
752
753const CURRENT_THREAD_ALREADY_IN_POOL: &str =
754 "The current thread is already part of another thread pool.";
755
756impl Error for ThreadPoolBuildError {
757 #[allow(deprecated)]
758 fn description(&self) -> &str {
759 match self.kind {
760 ErrorKind::GlobalPoolAlreadyInitialized => GLOBAL_POOL_ALREADY_INITIALIZED,
761 ErrorKind::CurrentThreadAlreadyInPool => CURRENT_THREAD_ALREADY_IN_POOL,
762 ErrorKind::IOError(ref e) => e.description(),
763 }
764 }
765
766 fn source(&self) -> Option<&(dyn Error + 'static)> {
767 match &self.kind {
768 ErrorKind::GlobalPoolAlreadyInitialized | ErrorKind::CurrentThreadAlreadyInPool => None,
769 ErrorKind::IOError(e) => Some(e),
770 }
771 }
772}
773
774impl fmt::Display for ThreadPoolBuildError {
775 fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
776 match &self.kind {
777 ErrorKind::CurrentThreadAlreadyInPool => CURRENT_THREAD_ALREADY_IN_POOL.fmt(f),
778 ErrorKind::GlobalPoolAlreadyInitialized => GLOBAL_POOL_ALREADY_INITIALIZED.fmt(f),
779 ErrorKind::IOError(e) => e.fmt(f),
780 }
781 }
782}
783
784/// Deprecated in favor of `ThreadPoolBuilder::build_global`.
785#[deprecated(note = "use `ThreadPoolBuilder::build_global`")]
786#[allow(deprecated)]
787pub fn initialize(config: Configuration) -> Result<(), Box<dyn Error>> {
788 config.into_builder().build_global().map_err(Box::from)
789}
790
791impl<S> fmt::Debug for ThreadPoolBuilder<S> {
792 fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
793 let ThreadPoolBuilder {
794 ref num_threads,
795 ref use_current_thread,
796 ref get_thread_name,
797 ref panic_handler,
798 ref stack_size,
799 ref start_handler,
800 ref exit_handler,
801 spawn_handler: _,
802 ref breadth_first,
803 } = *self;
804
805 // Just print `Some(<closure>)` or `None` to the debug
806 // output.
807 struct ClosurePlaceholder;
808 impl fmt::Debug for ClosurePlaceholder {
809 fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
810 f.write_str("<closure>")
811 }
812 }
813 let get_thread_name = get_thread_name.as_ref().map(|_| ClosurePlaceholder);
814 let panic_handler = panic_handler.as_ref().map(|_| ClosurePlaceholder);
815 let start_handler = start_handler.as_ref().map(|_| ClosurePlaceholder);
816 let exit_handler = exit_handler.as_ref().map(|_| ClosurePlaceholder);
817
818 f.debug_struct("ThreadPoolBuilder")
819 .field("num_threads", num_threads)
820 .field("use_current_thread", use_current_thread)
821 .field("get_thread_name", &get_thread_name)
822 .field("panic_handler", &panic_handler)
823 .field("stack_size", &stack_size)
824 .field("start_handler", &start_handler)
825 .field("exit_handler", &exit_handler)
826 .field("breadth_first", &breadth_first)
827 .finish()
828 }
829}
830
831#[allow(deprecated)]
832impl fmt::Debug for Configuration {
833 fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
834 self.builder.fmt(f)
835 }
836}
837
838/// Provides the calling context to a closure called by `join_context`.
839#[derive(Debug)]
840pub struct FnContext {
841 migrated: bool,
842
843 /// disable `Send` and `Sync`, just for a little future-proofing.
844 _marker: PhantomData<*mut ()>,
845}
846
847impl FnContext {
848 #[inline]
849 fn new(migrated: bool) -> Self {
850 FnContext {
851 migrated,
852 _marker: PhantomData,
853 }
854 }
855}
856
857impl FnContext {
858 /// Returns `true` if the closure was called from a different thread
859 /// than it was provided from.
860 #[inline]
861 pub fn migrated(&self) -> bool {
862 self.migrated
863 }
864}