1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180
use Configuration; #[cfg(feature = "unstable")] use future::{Future, RayonFuture}; use latch::LockLatch; #[allow(unused_imports)] use log::Event::*; use job::StackJob; use join; use {scope, Scope}; #[cfg(feature = "unstable")] use spawn; use std::sync::Arc; use std::error::Error; use registry::{Registry, WorkerThread}; mod test; pub struct ThreadPool { registry: Arc<Registry>, } impl ThreadPool { /// Constructs a new thread pool with the given configuration. If /// the configuration is not valid, returns a suitable `Err` /// result. See `InitError` for more details. pub fn new(configuration: Configuration) -> Result<ThreadPool, Box<Error>> { let registry = try!(Registry::new(configuration)); Ok(ThreadPool { registry: registry }) } /// Returns a handle to the global thread pool. This is the pool /// that Rayon will use by default when you perform a `join()` or /// `scope()` operation, if no other thread-pool is installed. If /// no global thread-pool has yet been started when this function /// is called, then the global thread-pool will be created (with /// the default configuration). If you wish to configure the /// global thread-pool differently, then you can use [the /// `rayon::initialize()` function][f] to do so. /// /// [f]: fn.initialize.html #[cfg(feature = "unstable")] pub fn global() -> &'static Arc<ThreadPool> { lazy_static! { static ref DEFAULT_THREAD_POOL: Arc<ThreadPool> = Arc::new(ThreadPool { registry: Registry::global() }); } &DEFAULT_THREAD_POOL } /// Executes `op` within the threadpool. Any attempts to use /// `join`, `scope`, or parallel iterators will then operate /// within that threadpool. /// /// # 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, that panic will be propagated. pub fn install<OP, R>(&self, op: OP) -> R where OP: FnOnce() -> R + Send { unsafe { let job_a = StackJob::new(op, LockLatch::new()); self.registry.inject(&[job_a.as_job_ref()]); job_a.latch.wait(); job_a.into_result() } } /// Returns the (current) number of threads in the thread pool. /// /// ### Future compatibility note /// /// Note that unless this thread-pool was created with a /// configuration that specifies the number of threads, then this /// number may vary over time in future versions (see [the /// `num_threads()` method for details][snt]). /// /// [snt]: struct.Configuration.html#method.num_threads pub fn current_num_threads(&self) -> usize { self.registry.num_threads() } /// If called from a Rayon worker thread in this thread-pool, /// returns the index of that thread; if not called from a Rayon /// thread, or called from a Rayon thread that belongs to a /// different thread-pool, returns `None`. /// /// The index for a given thread will not change over the thread's /// lifetime. However, multiple threads may share the same index if /// they are in distinct thread-pools. /// /// ### Future compatibility note /// /// Currently, every thread-pool (including the global /// thread-pool) has a fixed number of threads, but this may /// change in future Rayon versions (see [the `num_threads()` method /// for details][snt]). In that case, the index for a /// thread would not change during its lifetime, but thread /// indices may wind up being reused if threads are terminated and /// restarted. /// /// [snt]: struct.Configuration.html#method.num_threads pub fn current_thread_index(&self) -> Option<usize> { unsafe { let curr = WorkerThread::current(); if curr.is_null() { None } else if (*curr).registry().id() != self.registry.id() { None } else { Some((*curr).index()) } } } /// Execute `oper_a` and `oper_b` in the thread-pool and return /// the results. Equivalent to `self.install(|| join(oper_a, /// oper_b))`. #[cfg(feature = "unstable")] pub fn join<A, B, RA, RB>(&self, oper_a: A, oper_b: B) -> (RA, RB) where A: FnOnce() -> RA + Send, B: FnOnce() -> RB + Send, RA: Send, RB: Send { self.install(|| join(oper_a, oper_b)) } /// Creates a scope that executes within this thread-pool. /// Equivalent to `self.install(|| scope(...))`. /// /// See also: [the `scope()` function][scope]. /// /// [scope]: fn.scope.html #[cfg(feature = "unstable")] pub fn scope<'scope, OP, R>(&self, op: OP) -> R where OP: for<'s> FnOnce(&'s Scope<'scope>) -> R + 'scope + Send, R: Send { self.install(|| scope(op)) } /// Spawns an asynchronous task 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). /// /// See the [`spawn()` method defined on scopes][spawn] for more details. /// /// [spawn]: struct.Scope.html#method.spawn #[cfg(feature = "unstable")] pub fn spawn<OP>(&self, op: OP) where OP: FnOnce() + Send + 'static { // We assert that `self.registry` has not terminated. unsafe { spawn::spawn_in(op, &self.registry) } } /// Spawns an asynchronous future in this thread-pool. See /// `spawn_future()` for more details. #[cfg(feature = "unstable")] pub fn spawn_future<F>(&self, future: F) -> RayonFuture<F::Item, F::Error> where F: Future + Send + 'static { // We assert that `self.registry` has not yet terminated. unsafe { spawn::spawn_future_in(future, self.registry.clone()) } } } impl Drop for ThreadPool { fn drop(&mut self) { self.registry.terminate(); } }