mmtk/scheduler/
worker.rs

1use super::stat::WorkerLocalStat;
2use super::work_bucket::*;
3use super::*;
4use crate::mmtk::MMTK;
5use crate::util::copy::GCWorkerCopyContext;
6use crate::util::heap::layout::heap_parameters::MAX_SPACES;
7use crate::util::opaque_pointer::*;
8use crate::util::ObjectReference;
9use crate::vm::{Collection, GCThreadContext, VMBinding};
10use atomic::Atomic;
11use atomic_refcell::{AtomicRef, AtomicRefCell, AtomicRefMut};
12use crossbeam::deque::{self, Stealer};
13use crossbeam::queue::ArrayQueue;
14use std::sync::atomic::Ordering;
15use std::sync::{Arc, Mutex};
16
17/// Represents the ID of a GC worker thread.
18pub type ThreadId = usize;
19
20thread_local! {
21    /// Current worker's ordinal
22    static WORKER_ORDINAL: Atomic<ThreadId> = const { Atomic::new(ThreadId::MAX) };
23}
24
25/// Get current worker ordinal. Return `None` if the current thread is not a worker.
26pub fn current_worker_ordinal() -> Option<ThreadId> {
27    let ordinal = WORKER_ORDINAL.with(|x| x.load(Ordering::Relaxed));
28    if ordinal == ThreadId::MAX {
29        None
30    } else {
31        Some(ordinal)
32    }
33}
34
35/// The struct has one instance per worker, but is shared between workers via the scheduler
36/// instance.  This structure is used for communication between workers, e.g. adding designated
37/// work packets, stealing work packets from other workers, and collecting per-worker statistics.
38pub struct GCWorkerShared<VM: VMBinding> {
39    /// Worker-local statistics data.
40    stat: AtomicRefCell<WorkerLocalStat<VM>>,
41    /// Accumulated bytes for live objects in this GC. When each worker scans
42    /// objects, we increase the live bytes. We get this value from each worker
43    /// at the end of a GC, and reset this counter.
44    /// The live bytes are stored in an array. The index is the index from the space descriptor.
45    pub live_bytes_per_space: AtomicRefCell<[usize; MAX_SPACES]>,
46    /// A queue of GCWork that can only be processed by the owned thread.
47    pub designated_work: ArrayQueue<Box<dyn GCWork<VM>>>,
48    /// Handle for stealing packets from the current worker
49    pub stealer: Option<Stealer<Box<dyn GCWork<VM>>>>,
50}
51
52impl<VM: VMBinding> GCWorkerShared<VM> {
53    /// Create a new `GCWorkerShared` instance, optionally with a `stealer` handle that other
54    /// workers can use to steal work packets from this worker's local queue.
55    pub fn new(stealer: Option<Stealer<Box<dyn GCWork<VM>>>>) -> Self {
56        Self {
57            stat: Default::default(),
58            live_bytes_per_space: AtomicRefCell::new([0; MAX_SPACES]),
59            designated_work: ArrayQueue::new(16),
60            stealer,
61        }
62    }
63
64    pub(crate) fn increase_live_bytes(
65        live_bytes_per_space: &mut [usize; MAX_SPACES],
66        object: ObjectReference,
67    ) {
68        use crate::mmtk::VM_MAP;
69        use crate::vm::object_model::ObjectModel;
70
71        // The live bytes of the object
72        let bytes = VM::VMObjectModel::get_current_size(object);
73        // Get the space index from descriptor
74        let space_descriptor = VM_MAP.get_descriptor_for_address(object.to_raw_address());
75        if space_descriptor != crate::util::heap::space_descriptor::SpaceDescriptor::UNINITIALIZED {
76            let space_index = space_descriptor.get_index();
77            debug_assert!(
78                space_index < MAX_SPACES,
79                "Space index {} is not in the range of [0, {})",
80                space_index,
81                MAX_SPACES
82            );
83            // Accumulate the live bytes for the index
84            live_bytes_per_space[space_index] += bytes;
85        }
86    }
87}
88
89/// A GC worker.  This part is privately owned by a worker thread.
90pub struct GCWorker<VM: VMBinding> {
91    /// The VM-specific thread-local state of the GC thread.
92    pub tls: VMWorkerThread,
93    /// The ordinal of the worker, numbered from 0 to the number of workers minus one.
94    pub ordinal: ThreadId,
95    /// The reference to the scheduler.
96    scheduler: Arc<GCWorkScheduler<VM>>,
97    /// The copy context, used to implement copying GC.
98    copy: GCWorkerCopyContext<VM>,
99    /// The reference to the MMTk instance.
100    pub mmtk: &'static MMTK<VM>,
101    /// Reference to the shared part of the GC worker.  It is used for synchronization.
102    pub shared: Arc<GCWorkerShared<VM>>,
103    /// Local work packet queue.
104    pub local_work_buffer: deque::Worker<Box<dyn GCWork<VM>>>,
105}
106
107unsafe impl<VM: VMBinding> Sync for GCWorkerShared<VM> {}
108unsafe impl<VM: VMBinding> Send for GCWorkerShared<VM> {}
109
110// Error message for borrowing `GCWorkerShared::stat`.
111const STAT_BORROWED_MSG: &str = "GCWorkerShared.stat is already borrowed.  This may happen if \
112    the mutator calls harness_begin or harness_end while the GC is running.";
113
114impl<VM: VMBinding> GCWorkerShared<VM> {
115    /// Immutably borrow this worker's local statistics.
116    pub fn borrow_stat(&self) -> AtomicRef<'_, WorkerLocalStat<VM>> {
117        self.stat.try_borrow().expect(STAT_BORROWED_MSG)
118    }
119
120    /// Mutably borrow this worker's local statistics.
121    pub fn borrow_stat_mut(&self) -> AtomicRefMut<'_, WorkerLocalStat<VM>> {
122        self.stat.try_borrow_mut().expect(STAT_BORROWED_MSG)
123    }
124}
125
126/// A special error type that indicate a worker should exit.
127/// This may happen if the VM needs to fork and asks workers to exit.
128#[derive(Debug)]
129pub(crate) struct WorkerShouldExit;
130
131/// The result type of `GCWorker::pool`.
132/// Too many functions return `Option<Box<dyn GCWork<VM>>>`.  In most cases, when `None` is
133/// returned, the caller should try getting work packets from another place.  To avoid confusion,
134/// we use `Err(WorkerShouldExit)` to clearly indicate that the worker should exit immediately.
135pub(crate) type PollResult<VM> = Result<Box<dyn GCWork<VM>>, WorkerShouldExit>;
136
137impl<VM: VMBinding> GCWorker<VM> {
138    pub(crate) fn new(
139        mmtk: &'static MMTK<VM>,
140        ordinal: ThreadId,
141        scheduler: Arc<GCWorkScheduler<VM>>,
142        shared: Arc<GCWorkerShared<VM>>,
143        local_work_buffer: deque::Worker<Box<dyn GCWork<VM>>>,
144    ) -> Self {
145        Self {
146            tls: VMWorkerThread(VMThread::UNINITIALIZED),
147            ordinal,
148            // We will set this later
149            copy: GCWorkerCopyContext::new_non_copy(),
150            scheduler,
151            mmtk,
152            shared,
153            local_work_buffer,
154        }
155    }
156
157    const LOCALLY_CACHED_WORK_PACKETS: usize = 16;
158
159    /// Add a boxed work packet to the work queue, in the given bucket.
160    /// Like [`GCWorker::add_work`], but the work packet is already boxed.
161    pub fn add_boxed_work(&mut self, bucket: WorkBucketStage, work: Box<dyn GCWork<VM>>) {
162        if !self.scheduler().work_buckets[bucket].is_open()
163            || self.local_work_buffer.len() >= Self::LOCALLY_CACHED_WORK_PACKETS
164        {
165            self.scheduler.work_buckets[bucket].add_boxed(work);
166            return;
167        }
168        self.local_work_buffer.push(work);
169    }
170
171    /// Add a work packet to the work queue.
172    /// If the bucket is open, the packet will be pushed to the local queue, otherwise it will be
173    /// pushed to the global bucket.
174    pub fn add_work(&mut self, bucket: WorkBucketStage, work: impl GCWork<VM>) {
175        if !self.scheduler().work_buckets[bucket].is_open()
176            || self.local_work_buffer.len() >= Self::LOCALLY_CACHED_WORK_PACKETS
177        {
178            self.scheduler.work_buckets[bucket].add(work);
179            return;
180        }
181        self.local_work_buffer.push(Box::new(work));
182    }
183
184    /// Get the scheduler. There is only one scheduler per MMTk instance.
185    pub fn scheduler(&self) -> &GCWorkScheduler<VM> {
186        &self.scheduler
187    }
188
189    /// Get a mutable reference of the copy context for this worker.
190    pub fn get_copy_context_mut(&mut self) -> &mut GCWorkerCopyContext<VM> {
191        &mut self.copy
192    }
193
194    /// Poll a ready-to-execute work packet in the following order:
195    ///
196    /// 1. Any packet that should be processed only by this worker.
197    /// 2. Poll from the local work queue.
198    /// 3. Poll from open global work-buckets
199    /// 4. Steal from other workers
200    fn poll(&mut self) -> PollResult<VM> {
201        if let Some(work) = self.shared.designated_work.pop() {
202            return Ok(work);
203        }
204
205        if let Some(work) = self.local_work_buffer.pop() {
206            return Ok(work);
207        }
208
209        self.scheduler().poll(self)
210    }
211
212    /// Entry point of the worker thread.
213    ///
214    /// This function will resolve thread affinity, if it has been specified by the user.
215    ///
216    /// Each worker will keep polling and executing work packets in a loop.  It runs until the
217    /// worker is requested to exit.  Currently a worker may exit after
218    /// [`crate::mmtk::MMTK::prepare_to_fork`] is called.
219    ///
220    /// Arguments:
221    /// * `tls`: The VM-specific thread-local storage for this GC worker thread.
222    /// * `mmtk`: A reference to an MMTk instance.
223    pub fn run(mut self: Box<Self>, tls: VMWorkerThread, mmtk: &'static MMTK<VM>) {
224        probe!(mmtk, gcworker_run);
225        debug!(
226            "Worker started. ordinal: {}, {}",
227            self.ordinal,
228            crate::util::rust_util::debug_process_thread_id(),
229        );
230        WORKER_ORDINAL.with(|x| x.store(self.ordinal, Ordering::SeqCst));
231        self.scheduler.resolve_affinity(self.ordinal);
232        self.tls = tls;
233        self.copy = crate::plan::create_gc_worker_context(tls, mmtk);
234        loop {
235            // Instead of having work_start and work_end tracepoints, we have
236            // one tracepoint before polling for more work and one tracepoint
237            // before executing the work.
238            // This allows measuring the distribution of both the time needed
239            // poll work (between work_poll and work), and the time needed to
240            // execute work (between work and next work_poll).
241            // If we have work_start and work_end, we cannot measure the first
242            // poll.
243            probe!(mmtk, work_poll);
244            let Ok(mut work) = self.poll() else {
245                // The worker is asked to exit.  Break from the loop.
246                break;
247            };
248            // probe! expands to an empty block on unsupported platforms
249            #[allow(unused_variables)]
250            let typename = work.get_type_name();
251
252            #[cfg(feature = "bpftrace_workaround")]
253            // Workaround a problem where bpftrace script cannot see the work packet names,
254            // by force loading from the packet name.
255            // See the "Known issues" section in `tools/tracing/timeline/README.md`
256            std::hint::black_box(unsafe { *(typename.as_ptr()) });
257
258            probe!(mmtk, work, typename.as_ptr(), typename.len());
259            debug_assert!(
260                self.scheduler().is_worker_active(self.ordinal),
261                "Worker {} is executing {} while inactive.",
262                self.ordinal,
263                typename
264            );
265            work.do_work_with_stat(&mut self, mmtk);
266        }
267        debug!(
268            "Worker exiting. ordinal: {}, {}",
269            self.ordinal,
270            crate::util::rust_util::debug_process_thread_id(),
271        );
272        probe!(mmtk, gcworker_exit);
273
274        mmtk.scheduler.surrender_gc_worker(self);
275    }
276}
277
278/// Stateful part of [`WorkerGroup`].
279enum WorkerCreationState<VM: VMBinding> {
280    /// The initial state.  `GCWorker` structs have not been created and GC worker threads have not
281    /// been spawn.
282    Initial {
283        /// The local work queues for to-be-created workers.
284        local_work_queues: Vec<deque::Worker<Box<dyn GCWork<VM>>>>,
285    },
286    /// All worker threads are spawn and running.  `GCWorker` structs have been transferred to
287    /// worker threads.
288    Spawned,
289    /// Worker threads are stopping, or have already stopped, for forking. Instances of `GCWorker`
290    /// structs are collected here to be reused when GC workers are respawn.
291    Surrendered {
292        /// `GCWorker` instances not currently owned by active GC worker threads.  Once GC workers
293        /// are respawn, they will take ownership of these `GCWorker` instances.
294        // Note: Clippy warns about `Vec<Box<T>>` because `Vec<T>` is already in the heap.
295        // However, the purpose of this `Vec` is allowing GC worker threads to give their
296        // `Box<GCWorker<VM>>` instances back to this pool.  Therefore, the `Box` is necessary.
297        #[allow(clippy::vec_box)]
298        workers: Vec<Box<GCWorker<VM>>>,
299    },
300}
301
302/// A worker group to manage all the GC workers.
303pub(crate) struct WorkerGroup<VM: VMBinding> {
304    /// Shared worker data
305    pub workers_shared: Vec<Arc<GCWorkerShared<VM>>>,
306    /// The stateful part.  `None` means state transition is underway.
307    state: Mutex<Option<WorkerCreationState<VM>>>,
308}
309
310/// We have to persuade Rust that `WorkerGroup` is safe to share because the compiler thinks one
311/// worker can refer to another worker via the path "worker -> scheduler -> worker_group ->
312/// `Surrendered::workers` -> worker" which is cyclic reference and unsafe.
313unsafe impl<VM: VMBinding> Sync for WorkerGroup<VM> {}
314
315impl<VM: VMBinding> WorkerGroup<VM> {
316    /// Create a WorkerGroup
317    pub fn new(num_workers: usize) -> Arc<Self> {
318        let local_work_queues = (0..num_workers)
319            .map(|_| deque::Worker::new_fifo())
320            .collect::<Vec<_>>();
321
322        let workers_shared = (0..num_workers)
323            .map(|i| {
324                Arc::new(GCWorkerShared::<VM>::new(Some(
325                    local_work_queues[i].stealer(),
326                )))
327            })
328            .collect::<Vec<_>>();
329
330        Arc::new(Self {
331            workers_shared,
332            state: Mutex::new(Some(WorkerCreationState::Initial { local_work_queues })),
333        })
334    }
335
336    /// Spawn GC worker threads for the first time.
337    pub fn initial_spawn(&self, tls: VMThread, mmtk: &'static MMTK<VM>) {
338        let mut state = self.state.lock().unwrap();
339
340        let WorkerCreationState::Initial { local_work_queues } = state.take().unwrap() else {
341            panic!("GCWorker structs have already been created");
342        };
343
344        let workers = self.create_workers(local_work_queues, mmtk);
345        self.spawn(workers, tls);
346
347        *state = Some(WorkerCreationState::Spawned);
348    }
349
350    /// Respawn GC threads after stopping for forking.
351    pub fn respawn(&self, tls: VMThread) {
352        let mut state = self.state.lock().unwrap();
353
354        let WorkerCreationState::Surrendered { workers } = state.take().unwrap() else {
355            panic!("GCWorker structs have not been created, yet.");
356        };
357
358        self.spawn(workers, tls);
359
360        *state = Some(WorkerCreationState::Spawned)
361    }
362
363    /// Create `GCWorker` instances.
364    #[allow(clippy::vec_box)] // See `WorkerCreationState::Surrendered`.
365    fn create_workers(
366        &self,
367        local_work_queues: Vec<deque::Worker<Box<dyn GCWork<VM>>>>,
368        mmtk: &'static MMTK<VM>,
369    ) -> Vec<Box<GCWorker<VM>>> {
370        debug!("Creating GCWorker instances...");
371
372        assert_eq!(self.workers_shared.len(), local_work_queues.len());
373
374        // Each `GCWorker` instance corresponds to a `GCWorkerShared` at the same index.
375        let workers = (local_work_queues.into_iter())
376            .zip(self.workers_shared.iter())
377            .enumerate()
378            .map(|(ordinal, (queue, shared))| {
379                Box::new(GCWorker::new(
380                    mmtk,
381                    ordinal,
382                    mmtk.scheduler.clone(),
383                    shared.clone(),
384                    queue,
385                ))
386            })
387            .collect::<Vec<_>>();
388
389        debug!("Created {} GCWorker instances.", workers.len());
390        workers
391    }
392
393    /// Spawn all the worker threads
394    #[allow(clippy::vec_box)] // See `WorkerCreationState::Surrendered`.
395    fn spawn(&self, workers: Vec<Box<GCWorker<VM>>>, tls: VMThread) {
396        debug!(
397            "Spawning GC workers.  {}",
398            crate::util::rust_util::debug_process_thread_id(),
399        );
400
401        // We transfer the ownership of each `GCWorker` instance to a GC thread.
402        for worker in workers {
403            VM::VMCollection::spawn_gc_thread(tls, GCThreadContext::<VM>::Worker(worker));
404        }
405
406        debug!(
407            "Spawned {} worker threads.  {}",
408            self.worker_count(),
409            crate::util::rust_util::debug_process_thread_id(),
410        );
411    }
412
413    /// Prepare the buffer for workers to surrender their `GCWorker` structs.
414    pub fn prepare_surrender_buffer(&self) {
415        let mut state = self.state.lock().unwrap();
416        assert!(matches!(*state, Some(WorkerCreationState::Spawned)));
417
418        *state = Some(WorkerCreationState::Surrendered {
419            workers: Vec::with_capacity(self.worker_count()),
420        })
421    }
422
423    /// Return the `GCWorker` struct to the worker group.
424    /// This function returns `true` if all workers returned their `GCWorker` structs.
425    pub fn surrender_gc_worker(&self, worker: Box<GCWorker<VM>>) -> bool {
426        let mut state = self.state.lock().unwrap();
427        let WorkerCreationState::Surrendered { ref mut workers } = state.as_mut().unwrap() else {
428            panic!("GCWorker structs have not been created, yet.");
429        };
430        let ordinal = worker.ordinal;
431        workers.push(worker);
432        trace!(
433            "Worker {} surrendered. ({}/{})",
434            ordinal,
435            workers.len(),
436            self.worker_count()
437        );
438        workers.len() == self.worker_count()
439    }
440
441    /// Get the number of workers in the group
442    pub fn worker_count(&self) -> usize {
443        self.workers_shared.len()
444    }
445
446    /// Return true if there're any pending designated work
447    pub fn has_designated_work(&self) -> bool {
448        self.workers_shared
449            .iter()
450            .any(|w| !w.designated_work.is_empty())
451    }
452
453    /// Get the live bytes data from the worker, and clear the local data.
454    pub fn get_and_clear_worker_live_bytes(&self) -> [usize; MAX_SPACES] {
455        let mut ret = [0; MAX_SPACES];
456        self.workers_shared.iter().for_each(|w| {
457            let mut live_bytes_per_space = w.live_bytes_per_space.borrow_mut();
458            for (idx, val) in live_bytes_per_space.iter_mut().enumerate() {
459                ret[idx] += *val;
460                *val = 0;
461            }
462        });
463        ret
464    }
465}
mmtk/scheduler/worker.rs

mmtk/scheduler/
worker.rs
