1//! Code that decides when workers should go to sleep. See README.md
2//! for an overview.
3
4use crate::latch::CoreLatch;
5use crate::log::Event::*;
6use crate::log::Logger;
7use crate::registry::WorkerThread;
8use crate::DeadlockHandler;
9use crossbeam_utils::CachePadded;
10use std::sync::atomic::Ordering;
11use std::sync::{Condvar, Mutex};
12use std::thread;
13use std::usize;
14
15mod counters;
16pub(crate) use self::counters::THREADS_MAX;
17use self::counters::{AtomicCounters, JobsEventCounter};
18
19struct SleepData {
20 /// The number of threads in the thread pool.
21 worker_count: usize,
22
23 /// The number of threads in the thread pool which are running and
24 /// aren't blocked in user code or sleeping.
25 active_threads: usize,
26
27 /// The number of threads which are blocked in user code.
28 /// This doesn't include threads blocked by this module.
29 blocked_threads: usize,
30}
31
32impl SleepData {
33 /// Checks if the conditions for a deadlock holds and if so calls the deadlock handler
34 #[inline]
35 pub fn deadlock_check(&self, deadlock_handler: &Option<Box<DeadlockHandler>>) {
36 if self.active_threads == 0 && self.blocked_threads > 0 {
37 (deadlock_handler.as_ref().unwrap())();
38 }
39 }
40}
41
42/// The `Sleep` struct is embedded into each registry. It governs the waking and sleeping
43/// of workers. It has callbacks that are invoked periodically at significant events,
44/// such as when workers are looping and looking for work, when latches are set, or when
45/// jobs are published, and it either blocks threads or wakes them in response to these
46/// events. See the [`README.md`] in this module for more details.
47///
48/// [`README.md`] README.md
49pub(super) struct Sleep {
50 logger: Logger,
51
52 /// One "sleep state" per worker. Used to track if a worker is sleeping and to have
53 /// them block.
54 worker_sleep_states: Vec<CachePadded<WorkerSleepState>>,
55
56 counters: AtomicCounters,
57
58 data: Mutex<SleepData>,
59}
60
61/// An instance of this struct is created when a thread becomes idle.
62/// It is consumed when the thread finds work, and passed by `&mut`
63/// reference for operations that preserve the idle state. (In other
64/// words, producing one of these structs is evidence the thread is
65/// idle.) It tracks state such as how long the thread has been idle.
66pub(super) struct IdleState {
67 /// What is worker index of the idle thread?
68 worker_index: usize,
69
70 /// How many rounds have we been circling without sleeping?
71 rounds: u32,
72
73 /// Once we become sleepy, what was the sleepy counter value?
74 /// Set to `INVALID_SLEEPY_COUNTER` otherwise.
75 jobs_counter: JobsEventCounter,
76}
77
78/// The "sleep state" for an individual worker.
79#[derive(Default)]
80struct WorkerSleepState {
81 /// Set to true when the worker goes to sleep; set to false when
82 /// the worker is notified or when it wakes.
83 is_blocked: Mutex<bool>,
84
85 condvar: Condvar,
86}
87
88const ROUNDS_UNTIL_SLEEPY: u32 = 32;
89const ROUNDS_UNTIL_SLEEPING: u32 = ROUNDS_UNTIL_SLEEPY + 1;
90
91impl Sleep {
92 pub(super) fn new(logger: Logger, n_threads: usize) -> Sleep {
93 assert!(n_threads <= THREADS_MAX);
94 Sleep {
95 logger,
96 worker_sleep_states: (0..n_threads).map(|_| Default::default()).collect(),
97 counters: AtomicCounters::new(),
98 data: Mutex::new(SleepData {
99 worker_count: n_threads,
100 active_threads: n_threads,
101 blocked_threads: 0,
102 }),
103 }
104 }
105
106 /// Mark a Rayon worker thread as blocked. This triggers the deadlock handler
107 /// if no other worker thread is active
108 #[inline]
109 pub fn mark_blocked(&self, deadlock_handler: &Option<Box<DeadlockHandler>>) {
110 let mut data = self.data.lock().unwrap();
111 debug_assert!(data.active_threads > 0);
112 debug_assert!(data.blocked_threads < data.worker_count);
113 debug_assert!(data.active_threads > 0);
114 data.active_threads -= 1;
115 data.blocked_threads += 1;
116
117 data.deadlock_check(deadlock_handler);
118 }
119
120 /// Mark a previously blocked Rayon worker thread as unblocked
121 #[inline]
122 pub fn mark_unblocked(&self) {
123 let mut data = self.data.lock().unwrap();
124 debug_assert!(data.active_threads < data.worker_count);
125 debug_assert!(data.blocked_threads > 0);
126 data.active_threads += 1;
127 data.blocked_threads -= 1;
128 }
129
130 #[inline]
131 pub(super) fn start_looking(&self, worker_index: usize, latch: &CoreLatch) -> IdleState {
132 self.logger.log(|| ThreadIdle {
133 worker: worker_index,
134 latch_addr: latch.addr(),
135 });
136
137 self.counters.add_inactive_thread();
138
139 IdleState {
140 worker_index,
141 rounds: 0,
142 jobs_counter: JobsEventCounter::DUMMY,
143 }
144 }
145
146 #[inline]
147 pub(super) fn work_found(&self, idle_state: IdleState) {
148 self.logger.log(|| ThreadFoundWork {
149 worker: idle_state.worker_index,
150 yields: idle_state.rounds,
151 });
152
153 // If we were the last idle thread and other threads are still sleeping,
154 // then we should wake up another thread.
155 let threads_to_wake = self.counters.sub_inactive_thread();
156 self.wake_any_threads(threads_to_wake as u32);
157 }
158
159 #[inline]
160 pub(super) fn no_work_found(
161 &self,
162 idle_state: &mut IdleState,
163 latch: &CoreLatch,
164 thread: &WorkerThread,
165 ) {
166 if idle_state.rounds < ROUNDS_UNTIL_SLEEPY {
167 thread::yield_now();
168 idle_state.rounds += 1;
169 } else if idle_state.rounds == ROUNDS_UNTIL_SLEEPY {
170 idle_state.jobs_counter = self.announce_sleepy(idle_state.worker_index);
171 idle_state.rounds += 1;
172 thread::yield_now();
173 } else if idle_state.rounds < ROUNDS_UNTIL_SLEEPING {
174 idle_state.rounds += 1;
175 thread::yield_now();
176 } else {
177 debug_assert_eq!(idle_state.rounds, ROUNDS_UNTIL_SLEEPING);
178 self.sleep(idle_state, latch, thread);
179 }
180 }
181
182 #[cold]
183 fn announce_sleepy(&self, worker_index: usize) -> JobsEventCounter {
184 let counters = self
185 .counters
186 .increment_jobs_event_counter_if(JobsEventCounter::is_active);
187 let jobs_counter = counters.jobs_counter();
188 self.logger.log(|| ThreadSleepy {
189 worker: worker_index,
190 jobs_counter: jobs_counter.as_usize(),
191 });
192 jobs_counter
193 }
194
195 #[cold]
196 fn sleep(&self, idle_state: &mut IdleState, latch: &CoreLatch, thread: &WorkerThread) {
197 let worker_index = idle_state.worker_index;
198
199 if !latch.get_sleepy() {
200 self.logger.log(|| ThreadSleepInterruptedByLatch {
201 worker: worker_index,
202 latch_addr: latch.addr(),
203 });
204
205 return;
206 }
207
208 let sleep_state = &self.worker_sleep_states[worker_index];
209 let mut is_blocked = sleep_state.is_blocked.lock().unwrap();
210 debug_assert!(!*is_blocked);
211
212 // Our latch was signalled. We should wake back up fully as we
213 // will have some stuff to do.
214 if !latch.fall_asleep() {
215 self.logger.log(|| ThreadSleepInterruptedByLatch {
216 worker: worker_index,
217 latch_addr: latch.addr(),
218 });
219
220 idle_state.wake_fully();
221 return;
222 }
223
224 loop {
225 let counters = self.counters.load(Ordering::SeqCst);
226
227 // Check if the JEC has changed since we got sleepy.
228 debug_assert!(idle_state.jobs_counter.is_sleepy());
229 if counters.jobs_counter() != idle_state.jobs_counter {
230 // JEC has changed, so a new job was posted, but for some reason
231 // we didn't see it. We should return to just before the SLEEPY
232 // state so we can do another search and (if we fail to find
233 // work) go back to sleep.
234 self.logger.log(|| ThreadSleepInterruptedByJob {
235 worker: worker_index,
236 });
237
238 idle_state.wake_partly();
239 latch.wake_up();
240 return;
241 }
242
243 // Otherwise, let's move from IDLE to SLEEPING.
244 if self.counters.try_add_sleeping_thread(counters) {
245 break;
246 }
247 }
248
249 // Successfully registered as asleep.
250
251 self.logger.log(|| ThreadSleeping {
252 worker: worker_index,
253 latch_addr: latch.addr(),
254 });
255
256 // We have one last check for injected jobs to do. This protects against
257 // deadlock in the very unlikely event that
258 //
259 // - an external job is being injected while we are sleepy
260 // - that job triggers the rollover over the JEC such that we don't see it
261 // - we are the last active worker thread
262 std::sync::atomic::fence(Ordering::SeqCst);
263 if thread.has_injected_job() {
264 // If we see an externally injected job, then we have to 'wake
265 // ourselves up'. (Ordinarily, `sub_sleeping_thread` is invoked by
266 // the one that wakes us.)
267 self.counters.sub_sleeping_thread();
268 } else {
269 {
270 // Decrement the number of active threads and check for a deadlock
271 let mut data = self.data.lock().unwrap();
272 data.active_threads -= 1;
273 data.deadlock_check(&thread.registry.deadlock_handler);
274 }
275
276 // If we don't see an injected job (the normal case), then flag
277 // ourselves as asleep and wait till we are notified.
278 //
279 // (Note that `is_blocked` is held under a mutex and the mutex was
280 // acquired *before* we incremented the "sleepy counter". This means
281 // that whomever is coming to wake us will have to wait until we
282 // release the mutex in the call to `wait`, so they will see this
283 // boolean as true.)
284 thread.registry.release_thread();
285 *is_blocked = true;
286 while *is_blocked {
287 is_blocked = sleep_state.condvar.wait(is_blocked).unwrap();
288 }
289
290 // Drop `is_blocked` now in case `acquire_thread` blocks
291 drop(is_blocked);
292
293 thread.registry.acquire_thread();
294 }
295
296 // Update other state:
297 idle_state.wake_fully();
298 latch.wake_up();
299
300 self.logger.log(|| ThreadAwoken {
301 worker: worker_index,
302 latch_addr: latch.addr(),
303 });
304 }
305
306 /// Notify the given thread that it should wake up (if it is
307 /// sleeping). When this method is invoked, we typically know the
308 /// thread is asleep, though in rare cases it could have been
309 /// awoken by (e.g.) new work having been posted.
310 pub(super) fn notify_worker_latch_is_set(&self, target_worker_index: usize) {
311 self.wake_specific_thread(target_worker_index);
312 }
313
314 /// Signals that `num_jobs` new jobs were injected into the thread
315 /// pool from outside. This function will ensure that there are
316 /// threads available to process them, waking threads from sleep
317 /// if necessary.
318 ///
319 /// # Parameters
320 ///
321 /// - `source_worker_index` -- index of the thread that did the
322 /// push, or `usize::MAX` if this came from outside the thread
323 /// pool -- it is used only for logging.
324 /// - `num_jobs` -- lower bound on number of jobs available for stealing.
325 /// We'll try to get at least one thread per job.
326 #[inline]
327 pub(super) fn new_injected_jobs(
328 &self,
329 source_worker_index: usize,
330 num_jobs: u32,
331 queue_was_empty: bool,
332 ) {
333 // This fence is needed to guarantee that threads
334 // as they are about to fall asleep, observe any
335 // new jobs that may have been injected.
336 std::sync::atomic::fence(Ordering::SeqCst);
337
338 self.new_jobs(source_worker_index, num_jobs, queue_was_empty)
339 }
340
341 /// Signals that `num_jobs` new jobs were pushed onto a thread's
342 /// local deque. This function will try to ensure that there are
343 /// threads available to process them, waking threads from sleep
344 /// if necessary. However, this is not guaranteed: under certain
345 /// race conditions, the function may fail to wake any new
346 /// threads; in that case the existing thread should eventually
347 /// pop the job.
348 ///
349 /// # Parameters
350 ///
351 /// - `source_worker_index` -- index of the thread that did the
352 /// push, or `usize::MAX` if this came from outside the thread
353 /// pool -- it is used only for logging.
354 /// - `num_jobs` -- lower bound on number of jobs available for stealing.
355 /// We'll try to get at least one thread per job.
356 #[inline]
357 pub(super) fn new_internal_jobs(
358 &self,
359 source_worker_index: usize,
360 num_jobs: u32,
361 queue_was_empty: bool,
362 ) {
363 self.new_jobs(source_worker_index, num_jobs, queue_was_empty)
364 }
365
366 /// Common helper for `new_injected_jobs` and `new_internal_jobs`.
367 #[inline]
368 fn new_jobs(&self, source_worker_index: usize, num_jobs: u32, queue_was_empty: bool) {
369 // Read the counters and -- if sleepy workers have announced themselves
370 // -- announce that there is now work available. The final value of `counters`
371 // with which we exit the loop thus corresponds to a state when
372 let counters = self
373 .counters
374 .increment_jobs_event_counter_if(JobsEventCounter::is_sleepy);
375 let num_awake_but_idle = counters.awake_but_idle_threads();
376 let num_sleepers = counters.sleeping_threads();
377
378 self.logger.log(|| JobThreadCounts {
379 worker: source_worker_index,
380 num_idle: num_awake_but_idle as u16,
381 num_sleepers: num_sleepers as u16,
382 });
383
384 if num_sleepers == 0 {
385 // nobody to wake
386 return;
387 }
388
389 // Promote from u16 to u32 so we can interoperate with
390 // num_jobs more easily.
391 let num_awake_but_idle = num_awake_but_idle as u32;
392 let num_sleepers = num_sleepers as u32;
393
394 // If the queue is non-empty, then we always wake up a worker
395 // -- clearly the existing idle jobs aren't enough. Otherwise,
396 // check to see if we have enough idle workers.
397 if !queue_was_empty {
398 let num_to_wake = std::cmp::min(num_jobs, num_sleepers);
399 self.wake_any_threads(num_to_wake);
400 } else if num_awake_but_idle < num_jobs {
401 let num_to_wake = std::cmp::min(num_jobs - num_awake_but_idle, num_sleepers);
402 self.wake_any_threads(num_to_wake);
403 }
404 }
405
406 #[cold]
407 fn wake_any_threads(&self, mut num_to_wake: u32) {
408 if num_to_wake > 0 {
409 for i in 0..self.worker_sleep_states.len() {
410 if self.wake_specific_thread(i) {
411 num_to_wake -= 1;
412 if num_to_wake == 0 {
413 return;
414 }
415 }
416 }
417 }
418 }
419
420 fn wake_specific_thread(&self, index: usize) -> bool {
421 let sleep_state = &self.worker_sleep_states[index];
422
423 let mut is_blocked = sleep_state.is_blocked.lock().unwrap();
424 if *is_blocked {
425 *is_blocked = false;
426 sleep_state.condvar.notify_one();
427
428 // When the thread went to sleep, it will have incremented
429 // this value. When we wake it, its our job to decrement
430 // it. We could have the thread do it, but that would
431 // introduce a delay between when the thread was
432 // *notified* and when this counter was decremented. That
433 // might mislead people with new work into thinking that
434 // there are sleeping threads that they should try to
435 // wake, when in fact there is nothing left for them to
436 // do.
437 self.counters.sub_sleeping_thread();
438
439 // Increment the number of active threads
440 self.data.lock().unwrap().active_threads += 1;
441
442 self.logger.log(|| ThreadNotify { worker: index });
443
444 true
445 } else {
446 false
447 }
448 }
449}
450
451impl IdleState {
452 fn wake_fully(&mut self) {
453 self.rounds = 0;
454 self.jobs_counter = JobsEventCounter::DUMMY;
455 }
456
457 fn wake_partly(&mut self) {
458 self.rounds = ROUNDS_UNTIL_SLEEPY;
459 self.jobs_counter = JobsEventCounter::DUMMY;
460 }
461}
462