rustc_thread_pool/sleep/
counters.rs

1use std::sync::atomic::{AtomicUsize, Ordering};
2
3pub(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)]
19pub(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)]
27pub(super) struct JobsEventCounter(usize);
28
29impl JobsEventCounter {
30    pub(super) const DUMMY: JobsEventCounter = JobsEventCounter(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")]
57const THREADS_BITS: usize = 16;
58
59#[cfg(target_pointer_width = "32")]
60const THREADS_BITS: usize = 8;
61
62/// Bits to shift to select the sleeping threads
63/// (used with `select_bits`).
64#[allow(clippy::erasing_op)]
65const SLEEPING_SHIFT: usize = 0 * THREADS_BITS;
66
67/// Bits to shift to select the inactive threads
68/// (used with `select_bits`).
69#[allow(clippy::identity_op)]
70const INACTIVE_SHIFT: usize = 1 * THREADS_BITS;
71
72/// Bits to shift to select the JEC
73/// (use JOBS_BITS).
74const JEC_SHIFT: usize = 2 * THREADS_BITS;
75
76/// Max value for the thread counters.
77pub(crate) const THREADS_MAX: usize = (1 << THREADS_BITS) - 1;
78
79/// Constant that can be added to add one sleeping thread.
80const ONE_SLEEPING: usize = 1;
81
82/// Constant that can be added to add one inactive thread.
83/// An inactive thread is either idle, sleepy, or sleeping.
84const ONE_INACTIVE: usize = 1 << INACTIVE_SHIFT;
85
86/// Constant that can be added to add one to the JEC.
87const ONE_JEC: usize = 1 << JEC_SHIFT;
88
89impl AtomicCounters {
90    #[inline]
91    pub(super) fn new() -> AtomicCounters {
92        AtomicCounters { value: AtomicUsize::new(0) }
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        Ord::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]
212fn select_thread(word: usize, shift: usize) -> usize {
213    (word >> shift) & THREADS_MAX
214}
215
216#[inline]
217fn select_jec(word: usize) -> usize {
218    word >> JEC_SHIFT
219}
220
221impl 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 { word: self.word.wrapping_add(ONE_JEC) }
232    }
233
234    #[inline]
235    pub(super) fn jobs_counter(self) -> JobsEventCounter {
236        JobsEventCounter(select_jec(self.word))
237    }
238
239    /// The number of threads that are not actively
240    /// executing work. They may be idle, sleepy, or asleep.
241    #[inline]
242    pub(super) fn inactive_threads(self) -> usize {
243        select_thread(self.word, INACTIVE_SHIFT)
244    }
245
246    #[inline]
247    pub(super) fn awake_but_idle_threads(self) -> usize {
248        debug_assert!(
249            self.sleeping_threads() <= self.inactive_threads(),
250            "sleeping threads: {} > raw idle threads {}",
251            self.sleeping_threads(),
252            self.inactive_threads()
253        );
254        self.inactive_threads() - self.sleeping_threads()
255    }
256
257    #[inline]
258    pub(super) fn sleeping_threads(self) -> usize {
259        select_thread(self.word, SLEEPING_SHIFT)
260    }
261}
262
263impl std::fmt::Debug for Counters {
264    fn fmt(&self, fmt: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
265        let word = format!("{:016x}", self.word);
266        fmt.debug_struct("Counters")
267            .field("word", &word)
268            .field("jobs", &self.jobs_counter().0)
269            .field("inactive", &self.inactive_threads())
270            .field("sleeping", &self.sleeping_threads())
271            .finish()
272    }
273}
rustc_thread_pool/sleep/counters.rs

rustc_thread_pool/sleep/
counters.rs
