vendor/rustc-rayon-core/src/sleep/counters.rs

   1 use std::sync::atomic::{AtomicUsize, Ordering};
   2
   3 pub(super) struct AtomicCounters {
   4     /// Packs together a number of counters. The counters are ordered as
   5     /// follows, from least to most significant bits (here, we assuming
   6     /// that [`THREADS_BITS`] is equal to 10):
   7     ///
   8     /// * Bits 0..10: Stores the number of **sleeping threads**
   9     /// * Bits 10..20: Stores the number of **inactive threads**
  10     /// * Bits 20..: Stores the **job event counter** (JEC)
  11     ///
  12     /// This uses 10 bits ([`THREADS_BITS`]) to encode the number of threads. Note
  13     /// that the total number of bits (and hence the number of bits used for the
  14     /// JEC) will depend on whether we are using a 32- or 64-bit architecture.
  15     value: AtomicUsize,
  16 }
  17
  18 #[derive(Copy, Clone)]
  19 pub(super) struct Counters {
  20     word: usize,
  21 }
  22
  23 /// A value read from the **Jobs Event Counter**.
  24 /// See the [`README.md`](README.md) for more
  25 /// coverage of how the jobs event counter works.
  26 #[derive(Copy, Clone, Debug, PartialEq, PartialOrd)]
  27 pub(super) struct JobsEventCounter(usize);
  28
  29 impl JobsEventCounter {
  30     pub(super) const DUMMY: JobsEventCounter = JobsEventCounter(std::usize::MAX);
  31
  32     #[inline]
  33     pub(super) fn as_usize(self) -> usize {
  34         self.0
  35     }
  36
  37     /// The JEC "is sleepy" if the last thread to increment it was in the
  38     /// process of becoming sleepy. This is indicated by its value being *even*.
  39     /// When new jobs are posted, they check if the JEC is sleepy, and if so
  40     /// they incremented it.
  41     #[inline]
  42     pub(super) fn is_sleepy(self) -> bool {
  43         (self.as_usize() & 1) == 0
  44     }
  45
  46     /// The JEC "is active" if the last thread to increment it was posting new
  47     /// work. This is indicated by its value being *odd*. When threads get
  48     /// sleepy, they will check if the JEC is active, and increment it.
  49     #[inline]
  50     pub(super) fn is_active(self) -> bool {
  51         !self.is_sleepy()
  52     }
  53 }
  54
  55 /// Number of bits used for the thread counters.
  56 #[cfg(target_pointer_width = "64")]
  57 const THREADS_BITS: usize = 16;
  58
  59 #[cfg(target_pointer_width = "32")]
  60 const THREADS_BITS: usize = 8;
  61
  62 /// Bits to shift to select the sleeping threads
  63 /// (used with `select_bits`).
  64 const SLEEPING_SHIFT: usize = 0 * THREADS_BITS;
  65
  66 /// Bits to shift to select the inactive threads
  67 /// (used with `select_bits`).
  68 const INACTIVE_SHIFT: usize = 1 * THREADS_BITS;
  69
  70 /// Bits to shift to select the JEC
  71 /// (use JOBS_BITS).
  72 const JEC_SHIFT: usize = 2 * THREADS_BITS;
  73
  74 /// Max value for the thread counters.
  75 pub(crate) const THREADS_MAX: usize = (1 << THREADS_BITS) - 1;
  76
  77 /// Constant that can be added to add one sleeping thread.
  78 const ONE_SLEEPING: usize = 1;
  79
  80 /// Constant that can be added to add one inactive thread.
  81 /// An inactive thread is either idle, sleepy, or sleeping.
  82 const ONE_INACTIVE: usize = 1 << INACTIVE_SHIFT;
  83
  84 /// Constant that can be added to add one to the JEC.
  85 const ONE_JEC: usize = 1 << JEC_SHIFT;
  86
  87 impl AtomicCounters {
  88     #[inline]
  89     pub(super) fn new() -> AtomicCounters {
  90         AtomicCounters {
  91             value: AtomicUsize::new(0),
  92         }
  93     }
  94
  95     /// Load and return the current value of the various counters.
  96     /// This value can then be given to other method which will
  97     /// attempt to update the counters via compare-and-swap.
  98     #[inline]
  99     pub(super) fn load(&self, ordering: Ordering) -> Counters {
 100         Counters::new(self.value.load(ordering))
 101     }
 102
 103     #[inline]
 104     fn try_exchange(&self, old_value: Counters, new_value: Counters, ordering: Ordering) -> bool {
 105         self.value
 106             .compare_exchange(old_value.word, new_value.word, ordering, Ordering::Relaxed)
 107             .is_ok()
 108     }
 109
 110     /// Adds an inactive thread. This cannot fail.
 111     ///
 112     /// This should be invoked when a thread enters its idle loop looking
 113     /// for work. It is decremented when work is found. Note that it is
 114     /// not decremented if the thread transitions from idle to sleepy or sleeping;
 115     /// so the number of inactive threads is always greater-than-or-equal
 116     /// to the number of sleeping threads.
 117     #[inline]
 118     pub(super) fn add_inactive_thread(&self) {
 119         self.value.fetch_add(ONE_INACTIVE, Ordering::SeqCst);
 120     }
 121
 122     /// Increments the jobs event counter if `increment_when`, when applied to
 123     /// the current value, is true. Used to toggle the JEC from even (sleepy) to
 124     /// odd (active) or vice versa. Returns the final value of the counters, for
 125     /// which `increment_when` is guaranteed to return false.
 126     pub(super) fn increment_jobs_event_counter_if(
 127         &self,
 128         increment_when: impl Fn(JobsEventCounter) -> bool,
 129     ) -> Counters {
 130         loop {
 131             let old_value = self.load(Ordering::SeqCst);
 132             if increment_when(old_value.jobs_counter()) {
 133                 let new_value = old_value.increment_jobs_counter();
 134                 if self.try_exchange(old_value, new_value, Ordering::SeqCst) {
 135                     return new_value;
 136                 }
 137             } else {
 138                 return old_value;
 139             }
 140         }
 141     }
 142
 143     /// Subtracts an inactive thread. This cannot fail. It is invoked
 144     /// when a thread finds work and hence becomes active. It returns the
 145     /// number of sleeping threads to wake up (if any).
 146     ///
 147     /// See `add_inactive_thread`.
 148     #[inline]
 149     pub(super) fn sub_inactive_thread(&self) -> usize {
 150         let old_value = Counters::new(self.value.fetch_sub(ONE_INACTIVE, Ordering::SeqCst));
 151         debug_assert!(
 152             old_value.inactive_threads() > 0,
 153             "sub_inactive_thread: old_value {:?} has no inactive threads",
 154             old_value,
 155         );
 156         debug_assert!(
 157             old_value.sleeping_threads() <= old_value.inactive_threads(),
 158             "sub_inactive_thread: old_value {:?} had {} sleeping threads and {} inactive threads",
 159             old_value,
 160             old_value.sleeping_threads(),
 161             old_value.inactive_threads(),
 162         );
 163
 164         // Current heuristic: whenever an inactive thread goes away, if
 165         // there are any sleeping threads, wake 'em up.
 166         let sleeping_threads = old_value.sleeping_threads();
 167         std::cmp::min(sleeping_threads, 2)
 168     }
 169
 170     /// Subtracts a sleeping thread. This cannot fail, but it is only
 171     /// safe to do if you you know the number of sleeping threads is
 172     /// non-zero (i.e., because you have just awoken a sleeping
 173     /// thread).
 174     #[inline]
 175     pub(super) fn sub_sleeping_thread(&self) {
 176         let old_value = Counters::new(self.value.fetch_sub(ONE_SLEEPING, Ordering::SeqCst));
 177         debug_assert!(
 178             old_value.sleeping_threads() > 0,
 179             "sub_sleeping_thread: old_value {:?} had no sleeping threads",
 180             old_value,
 181         );
 182         debug_assert!(
 183             old_value.sleeping_threads() <= old_value.inactive_threads(),
 184             "sub_sleeping_thread: old_value {:?} had {} sleeping threads and {} inactive threads",
 185             old_value,
 186             old_value.sleeping_threads(),
 187             old_value.inactive_threads(),
 188         );
 189     }
 190
 191     #[inline]
 192     pub(super) fn try_add_sleeping_thread(&self, old_value: Counters) -> bool {
 193         debug_assert!(
 194             old_value.inactive_threads() > 0,
 195             "try_add_sleeping_thread: old_value {:?} has no inactive threads",
 196             old_value,
 197         );
 198         debug_assert!(
 199             old_value.sleeping_threads() < THREADS_MAX,
 200             "try_add_sleeping_thread: old_value {:?} has too many sleeping threads",
 201             old_value,
 202         );
 203
 204         let mut new_value = old_value;
 205         new_value.word += ONE_SLEEPING;
 206
 207         self.try_exchange(old_value, new_value, Ordering::SeqCst)
 208     }
 209 }
 210
 211 #[inline]
 212 fn select_thread(word: usize, shift: usize) -> usize {
 213     ((word >> shift) as usize) & THREADS_MAX
 214 }
 215
 216 #[inline]
 217 fn select_jec(word: usize) -> usize {
 218     (word >> JEC_SHIFT) as usize
 219 }
 220
 221 impl Counters {
 222     #[inline]
 223     fn new(word: usize) -> Counters {
 224         Counters { word }
 225     }
 226
 227     #[inline]
 228     fn increment_jobs_counter(self) -> Counters {
 229         // We can freely add to JEC because it occupies the most significant bits.
 230         // Thus it doesn't overflow into the other counters, just wraps itself.
 231         Counters {
 232             word: self.word.wrapping_add(ONE_JEC),
 233         }
 234     }
 235
 236     #[inline]
 237     pub(super) fn jobs_counter(self) -> JobsEventCounter {
 238         JobsEventCounter(select_jec(self.word))
 239     }
 240
 241     /// The number of threads that are not actively
 242     /// executing work. They may be idle, sleepy, or asleep.
 243     #[inline]
 244     pub(super) fn inactive_threads(self) -> usize {
 245         select_thread(self.word, INACTIVE_SHIFT)
 246     }
 247
 248     #[inline]
 249     pub(super) fn awake_but_idle_threads(self) -> usize {
 250         debug_assert!(
 251             self.sleeping_threads() <= self.inactive_threads(),
 252             "sleeping threads: {} > raw idle threads {}",
 253             self.sleeping_threads(),
 254             self.inactive_threads()
 255         );
 256         self.inactive_threads() - self.sleeping_threads()
 257     }
 258
 259     #[inline]
 260     pub(super) fn sleeping_threads(self) -> usize {
 261         select_thread(self.word, SLEEPING_SHIFT)
 262     }
 263 }
 264
 265 impl std::fmt::Debug for Counters {
 266     fn fmt(&self, fmt: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
 267         let word = format!("{:016x}", self.word);
 268         fmt.debug_struct("Counters")
 269             .field("word", &word)
 270             .field("jobs", &self.jobs_counter().0)
 271             .field("inactive", &self.inactive_threads())
 272             .field("sleeping", &self.sleeping_threads())
 273             .finish()
 274     }
 275 }