mmtk/scheduler/
scheduler.rs

1use self::worker::PollResult;
2
3use super::gc_work::ScheduleCollection;
4use super::stat::SchedulerStat;
5use super::work_bucket::*;
6use super::worker::{GCWorker, ThreadId, WorkerGroup};
7use super::worker_goals::{WorkerGoal, WorkerGoals};
8use super::worker_monitor::{LastParkedResult, WorkerMonitor};
9use super::*;
10use crate::mmtk::MMTK;
11use crate::plan::tracing::gc_work::weakref::{
12    VMForwardWeakRefs, VMPostForwarding, VMProcessWeakRefs,
13};
14use crate::util::opaque_pointer::*;
15use crate::util::options::AffinityKind;
16use crate::vm::Collection;
17use crate::vm::VMBinding;
18use crate::Plan;
19use crossbeam::deque::Steal;
20use enum_map::{Enum, EnumMap};
21use std::collections::HashMap;
22use std::sync::Arc;
23use std::time::Instant;
24
25pub struct GCWorkScheduler<VM: VMBinding> {
26    /// Work buckets
27    pub work_buckets: EnumMap<WorkBucketStage, WorkBucket<VM>>,
28    /// Workers
29    pub(crate) worker_group: Arc<WorkerGroup<VM>>,
30    /// For synchronized communication between workers and with mutators.
31    pub(crate) worker_monitor: Arc<WorkerMonitor>,
32    /// How to assign the affinity of each GC thread. Specified by the user.
33    affinity: AffinityKind,
34}
35
36// FIXME: GCWorkScheduler should be naturally Sync, but we cannot remove this `impl` yet.
37// Some subtle interaction between ObjectRememberingBarrier, Mutator and some GCWork instances
38// makes the compiler think WorkBucket is not Sync.
39unsafe impl<VM: VMBinding> Sync for GCWorkScheduler<VM> {}
40
41impl<VM: VMBinding> GCWorkScheduler<VM> {
42    pub fn new(num_workers: usize, affinity: AffinityKind) -> Arc<Self> {
43        let worker_monitor: Arc<WorkerMonitor> = Arc::new(WorkerMonitor::new(num_workers));
44        let worker_group = WorkerGroup::new(num_workers);
45
46        // Create work buckets for workers.
47        let mut work_buckets = EnumMap::from_fn(|stage: WorkBucketStage| {
48            WorkBucket::new(stage, worker_monitor.clone())
49        });
50
51        // Set the open condition of each bucket.
52        {
53            let mut open_stages: Vec<WorkBucketStage> = vec![WorkBucketStage::FIRST_STW_STAGE];
54            let stages = (0..WorkBucketStage::LENGTH).map(WorkBucketStage::from_usize);
55            for stage in stages {
56                if stage.is_sequentially_opened() {
57                    let cur_stages = open_stages.clone();
58                    // Other work packets will be opened after previous stages are done
59                    // (i.e their buckets are drained and all workers parked).
60                    work_buckets[stage].set_open_condition(
61                        move |scheduler: &GCWorkScheduler<VM>| {
62                            debug!(
63                                "Check if {:?} can be opened? These needs to be drained: {:?}",
64                                stage, cur_stages
65                            );
66                            scheduler.are_buckets_drained(&cur_stages)
67                        },
68                    );
69                    open_stages.push(stage);
70                }
71            }
72        }
73
74        Arc::new(Self {
75            work_buckets,
76            worker_group,
77            worker_monitor,
78            affinity,
79        })
80    }
81
82    pub fn num_workers(&self) -> usize {
83        self.worker_group.as_ref().worker_count()
84    }
85
86    pub fn set_active_workers(&self, active_workers: usize) {
87        self.worker_monitor.set_active_workers(active_workers);
88    }
89
90    pub fn num_active_workers(&self) -> usize {
91        self.worker_monitor.active_workers()
92    }
93
94    pub(crate) fn is_worker_active(&self, ordinal: usize) -> bool {
95        self.worker_monitor.is_worker_active(ordinal)
96    }
97
98    /// Create GC threads for the first time.  It will also create the `GCWorker` instances.
99    ///
100    /// Currently GC threads only include worker threads, and we currently have only one worker
101    /// group.  We may add more worker groups in the future.
102    pub fn spawn_gc_threads(self: &Arc<Self>, mmtk: &'static MMTK<VM>, tls: VMThread) {
103        self.worker_group.initial_spawn(tls, mmtk);
104    }
105
106    /// Ask all GC workers to exit for forking.
107    pub fn stop_gc_threads_for_forking(self: &Arc<Self>) {
108        self.worker_group.prepare_surrender_buffer();
109
110        debug!("A mutator is requesting GC threads to stop for forking...");
111        self.worker_monitor.make_request(WorkerGoal::StopForFork);
112    }
113
114    /// Ask all GC workers to exit permanently.
115    pub fn shutdown_gc_threads(self: &Arc<Self>) {
116        self.worker_group.prepare_surrender_buffer();
117
118        info!("A mutator is requesting GC threads to shut down...");
119        self.worker_monitor.make_request(WorkerGoal::Shutdown);
120    }
121
122    /// Surrender the `GCWorker` struct of a GC worker when it exits.
123    pub fn surrender_gc_worker(&self, worker: Box<GCWorker<VM>>) {
124        let all_surrendered = self.worker_group.surrender_gc_worker(worker);
125
126        if all_surrendered {
127            debug!(
128                "All {} workers surrendered.",
129                self.worker_group.worker_count()
130            );
131            self.worker_monitor.on_all_workers_exited();
132        }
133    }
134
135    /// Respawn GC threads after forking.  This will reuse the `GCWorker` instances of stopped
136    /// workers.  `tls` is the VM thread that requests GC threads to be re-spawn, and will be
137    /// passed down to [`crate::vm::Collection::spawn_gc_thread`].
138    pub fn respawn_gc_threads_after_forking(self: &Arc<Self>, tls: VMThread) {
139        self.worker_group.respawn(tls)
140    }
141
142    /// Resolve the affinity of a thread.
143    pub fn resolve_affinity(&self, thread: ThreadId) {
144        self.affinity.resolve_affinity(thread);
145    }
146
147    /// Request a GC to be scheduled.  Called by mutator via `GCTrigger`.
148    pub(crate) fn request_schedule_collection(&self) {
149        debug!("A mutator is sending GC-scheduling request to workers...");
150        self.worker_monitor.make_request(WorkerGoal::Gc);
151    }
152
153    /// Add the `ScheduleCollection` packet.  Called by the last parked worker.
154    fn add_schedule_collection_packet(&self) {
155        // We are still holding the mutex `WorkerMonitor::sync`.  Do not notify now.
156        probe!(mmtk, add_schedule_collection_packet);
157        self.work_buckets[WorkBucketStage::Unconstrained].add_no_notify(ScheduleCollection);
158    }
159
160    /// Schedule all the common work packets
161    pub fn schedule_common_work<C: GCWorkContext<VM = VM>>(&self, plan: &'static C::PlanType) {
162        use crate::scheduler::gc_work::*;
163        // Stop & scan mutators (mutator scanning can happen before STW)
164        self.work_buckets[WorkBucketStage::Unconstrained].add(StopMutators::<C>::new());
165
166        // Prepare global/collectors/mutators
167        self.work_buckets[WorkBucketStage::Prepare].add(Prepare::<C>::new(plan));
168
169        // Release global/collectors/mutators
170        self.work_buckets[WorkBucketStage::Release].add(Release::<C>::new(plan));
171
172        // Analysis GC work
173        #[cfg(feature = "analysis")]
174        {
175            use crate::util::analysis::GcHookWork;
176            self.work_buckets[WorkBucketStage::Unconstrained].add(GcHookWork);
177        }
178
179        // Sanity
180        #[cfg(feature = "sanity")]
181        {
182            use crate::util::sanity::sanity_checker::ScheduleSanityGC;
183            self.work_buckets[WorkBucketStage::Final]
184                .add(ScheduleSanityGC::<C::PlanType>::new(plan));
185        }
186
187        // Reference processing
188        if !*plan.base().options.no_reference_types {
189            use crate::util::reference_processor::{
190                PhantomRefProcessing, SoftRefProcessing, WeakRefProcessing,
191            };
192            self.work_buckets[WorkBucketStage::SoftRefClosure]
193                .add(SoftRefProcessing::<C::DefaultTrace>::new());
194            self.work_buckets[WorkBucketStage::WeakRefClosure].add(WeakRefProcessing::<VM>::new());
195            self.work_buckets[WorkBucketStage::PhantomRefClosure]
196                .add(PhantomRefProcessing::<VM>::new());
197
198            use crate::util::reference_processor::RefForwarding;
199            if plan.constraints().needs_forward_after_liveness {
200                self.work_buckets[WorkBucketStage::RefForwarding]
201                    .add(RefForwarding::<C::DefaultTrace>::new());
202            }
203
204            use crate::util::reference_processor::RefEnqueue;
205            self.work_buckets[WorkBucketStage::Release].add(RefEnqueue::<VM>::new());
206        }
207
208        // Finalization
209        if !*plan.base().options.no_finalizer {
210            use crate::util::finalizable_processor::{Finalization, ForwardFinalization};
211            // finalization
212            self.work_buckets[WorkBucketStage::FinalRefClosure]
213                .add(Finalization::<C::DefaultTrace>::new());
214            // forward refs
215            if plan.constraints().needs_forward_after_liveness {
216                self.work_buckets[WorkBucketStage::FinalizableForwarding]
217                    .add(ForwardFinalization::<C::DefaultTrace>::new());
218            }
219        }
220
221        // We add the VM-specific weak ref processing work regardless of MMTK-side options,
222        // including Options::no_finalizer and Options::no_reference_types.
223        //
224        // VMs need weak reference handling to function properly.  The VM may treat weak references
225        // as strong references, but it is not appropriate to simply disable weak reference
226        // handling from MMTk's side.  The VM, however, may choose to do nothing in
227        // `Collection::process_weak_refs` if appropriate.
228        //
229        // It is also not sound for MMTk core to turn off weak
230        // reference processing or finalization alone, because (1) not all VMs have the notion of
231        // weak references or finalizers, so it may not make sence, and (2) the VM may
232        // processing them together.
233
234        // VM-specific weak ref processing
235        // The `VMProcessWeakRefs` work packet is set as the sentinel so that it is executed when
236        // the `VMRefClosure` bucket is drained.  The VM binding may spawn new work packets into
237        // the `VMRefClosure` bucket, and request another `VMProcessWeakRefs` work packet to be
238        // executed again after this bucket is drained again.  Strictly speaking, the first
239        // `VMProcessWeakRefs` packet can be an ordinary packet (doesn't have to be a sentinel)
240        // because there are no other packets in the bucket.  We set it as sentinel for
241        // consistency.
242        self.work_buckets[WorkBucketStage::VMRefClosure]
243            .set_sentinel(Box::new(VMProcessWeakRefs::<C::DefaultTrace>::new()));
244
245        if plan.constraints().needs_forward_after_liveness {
246            // VM-specific weak ref forwarding
247            self.work_buckets[WorkBucketStage::VMRefForwarding]
248                .add(VMForwardWeakRefs::<C::DefaultTrace>::new());
249        }
250
251        self.work_buckets[WorkBucketStage::Release].add(VMPostForwarding::<VM>::default());
252    }
253
254    fn are_buckets_drained(&self, buckets: &[WorkBucketStage]) -> bool {
255        buckets
256            .iter()
257            .all(|&b| !self.work_buckets[b].is_enabled() || self.work_buckets[b].is_drained())
258    }
259
260    pub fn debug_assert_all_stw_buckets_empty(&self) {
261        debug_assert!(self
262            .work_buckets
263            .values()
264            .filter(|bucket| bucket.get_stage().is_stw())
265            .all(|bucket| {
266                if !bucket.is_empty() {
267                    warn!(
268                        "Work bucket {:?} is not empty but it is expected to be empty!",
269                        bucket.get_stage()
270                    );
271                    warn!("Queue: {:?}", bucket.get_queue().debug_dump_packets());
272                    false
273                } else {
274                    true
275                }
276            }))
277    }
278
279    /// Schedule "sentinel" work packets for all open buckets.
280    pub(crate) fn schedule_sentinels(&self) -> bool {
281        let mut new_packets = false;
282        for (id, work_bucket) in self.work_buckets.iter() {
283            if work_bucket.is_open() && work_bucket.maybe_schedule_sentinel() {
284                trace!("Scheduled sentinel packet into {:?}", id);
285                new_packets = true;
286            }
287        }
288        new_packets
289    }
290
291    /// Open buckets if their conditions are met.
292    ///
293    /// This function should only be called after all the workers are parked.
294    /// No workers will be waked up by this function. The caller is responsible for that.
295    ///
296    /// Return true if there're any non-empty buckets updated.
297    pub(crate) fn update_buckets(&self) -> bool {
298        debug!("update_buckets");
299        let mut buckets_updated = false;
300        let mut new_packets = false;
301        for i in 0..WorkBucketStage::LENGTH {
302            let id = WorkBucketStage::from_usize(i);
303            if id.is_always_open() {
304                continue;
305            }
306            let bucket = &self.work_buckets[id];
307            if !bucket.is_enabled() {
308                debug!("Work bucket {:?} is disabled. Skip.", id);
309                continue;
310            }
311            debug!("Checking if {:?} can be opened...", id);
312            let bucket_opened = bucket.update(self);
313            buckets_updated = buckets_updated || bucket_opened;
314            if bucket_opened {
315                probe!(mmtk, bucket_opened, id);
316                new_packets = new_packets || !bucket.is_drained();
317                if new_packets {
318                    // Quit the loop. There are already new packets in the newly opened buckets.
319                    trace!("Found new packets at stage {:?}.  Break.", id);
320                    break;
321                }
322                new_packets = new_packets || bucket.maybe_schedule_sentinel();
323                if new_packets {
324                    // Quit the loop. A sentinel packet is added to the newly opened buckets.
325                    trace!("Sentinel is scheduled at stage {:?}.  Break.", id);
326                    break;
327                }
328            }
329        }
330        buckets_updated && new_packets
331    }
332
333    pub fn close_all_stw_buckets(&self) {
334        self.work_buckets.iter().for_each(|(id, bkt)| {
335            if id.is_stw() {
336                bkt.close();
337            }
338        });
339    }
340
341    pub fn reset_state(&self) {
342        self.work_buckets.iter().for_each(|(id, bkt)| {
343            if id.is_stw() && !id.is_first_stw_stage() {
344                bkt.close();
345            }
346        });
347    }
348
349    pub fn debug_assert_all_stw_buckets_closed(&self) {
350        if cfg!(debug_assertions) {
351            self.work_buckets.iter().for_each(|(id, bkt)| {
352                if id.is_stw() {
353                    assert!(!bkt.is_open());
354                }
355            });
356        }
357    }
358
359    /// Check if all the work buckets are empty
360    pub(crate) fn assert_all_open_buckets_are_empty(&self) {
361        let mut error_example = None;
362        for (id, bucket) in self.work_buckets.iter() {
363            if bucket.is_enabled() && bucket.is_open() && !bucket.is_empty() {
364                error!("Work bucket {:?} is not drained!", id);
365                error!("Queue: {:?}", bucket.get_queue().debug_dump_packets());
366                // This error can be hard to reproduce.
367                // If an error happens in the release build where logs are turned off,
368                // we should show at least one abnormal bucket in the panic message
369                // so that we still have some information for debugging.
370                error_example = Some(id);
371            }
372        }
373        if let Some(id) = error_example {
374            panic!("Some open buckets (such as {:?}) are not empty.", id);
375        }
376    }
377
378    /// Get a schedulable work packet without retry.
379    fn poll_schedulable_work_once(&self, worker: &GCWorker<VM>) -> Steal<Box<dyn GCWork<VM>>> {
380        let mut should_retry = false;
381        // Try find a packet that can be processed only by this worker.
382        if let Some(w) = worker.shared.designated_work.pop() {
383            return Steal::Success(w);
384        }
385        // Try get a packet from a work bucket.
386        for work_bucket in self.work_buckets.values() {
387            match work_bucket.poll(&worker.local_work_buffer) {
388                Steal::Success(w) => return Steal::Success(w),
389                Steal::Retry => should_retry = true,
390                _ => {}
391            }
392        }
393        // Try steal some packets from any worker
394        for (id, worker_shared) in self.worker_group.workers_shared.iter().enumerate() {
395            if id == worker.ordinal {
396                continue;
397            }
398            match worker_shared.stealer.as_ref().unwrap().steal() {
399                Steal::Success(w) => return Steal::Success(w),
400                Steal::Retry => should_retry = true,
401                _ => {}
402            }
403        }
404        if should_retry {
405            Steal::Retry
406        } else {
407            Steal::Empty
408        }
409    }
410
411    /// Get a schedulable work packet.
412    fn poll_schedulable_work(&self, worker: &GCWorker<VM>) -> Option<Box<dyn GCWork<VM>>> {
413        // Loop until we successfully get a packet.
414        loop {
415            match self.poll_schedulable_work_once(worker) {
416                Steal::Success(w) => {
417                    return Some(w);
418                }
419                Steal::Retry => {
420                    std::thread::yield_now();
421                    continue;
422                }
423                Steal::Empty => {
424                    return None;
425                }
426            }
427        }
428    }
429
430    /// Called by workers to get a schedulable work packet.
431    /// Park the worker if there're no available packets.
432    pub(crate) fn poll(&self, worker: &GCWorker<VM>) -> PollResult<VM> {
433        if let Some(work) = self.poll_schedulable_work(worker) {
434            return Ok(work);
435        }
436        self.poll_slow(worker)
437    }
438
439    fn poll_slow(&self, worker: &GCWorker<VM>) -> PollResult<VM> {
440        loop {
441            // Retry polling
442            if let Some(work) = self.poll_schedulable_work(worker) {
443                return Ok(work);
444            }
445
446            let ordinal = worker.ordinal;
447            self.worker_monitor
448                .park_and_wait(ordinal, |goals| self.on_last_parked(worker, goals))?;
449        }
450    }
451
452    /// Called when the last worker parked.  `goal` allows this function to inspect and change the
453    /// current goal.
454    fn on_last_parked(&self, worker: &GCWorker<VM>, goals: &mut WorkerGoals) -> LastParkedResult {
455        let Some(ref current_goal) = goals.current() else {
456            // There is no goal.  Find a request to respond to.
457            return self.respond_to_requests(worker, goals);
458        };
459
460        match current_goal {
461            WorkerGoal::Gc => {
462                // We are in the progress of GC.
463
464                // In stop-the-world GC, mutators cannot request for GC while GC is in progress.
465                // When we support concurrent GC, we should remove this assertion.
466                assert!(
467                    !goals.debug_is_requested(WorkerGoal::Gc),
468                    "GC request sent to WorkerMonitor while GC is still in progress."
469                );
470
471                // We are in the middle of GC, and the last GC worker parked.
472                trace!("The last worker parked during GC.  Try to find more work to do...");
473
474                // During GC, if all workers parked, all open buckets must have been drained.
475                self.assert_all_open_buckets_are_empty();
476
477                // Find more work for workers to do.
478                let found_more_work = self.find_more_work_for_workers();
479
480                if found_more_work {
481                    LastParkedResult::WakeAll
482                } else {
483                    // GC finished.
484                    let concurrent_work_scheduled = self.on_gc_finished(worker);
485
486                    // Clear the current goal
487                    goals.on_current_goal_completed();
488
489                    if concurrent_work_scheduled {
490                        // It was the initial mark pause and scheduled concurrent work.
491                        // Wake up all GC workers to do concurrent work.
492                        LastParkedResult::WakeAll
493                    } else {
494                        // It was an STW GC or the final mark pause of a concurrent GC.
495                        // Respond to another goal.
496                        self.respond_to_requests(worker, goals)
497                    }
498                }
499            }
500            WorkerGoal::StopForFork | WorkerGoal::Shutdown => {
501                panic!(
502                    "Worker {} parked again when it is asked to exit.",
503                    worker.ordinal
504                )
505            }
506        }
507    }
508
509    /// Respond to a worker reqeust.
510    fn respond_to_requests(
511        &self,
512        worker: &GCWorker<VM>,
513        goals: &mut WorkerGoals,
514    ) -> LastParkedResult {
515        assert!(goals.current().is_none());
516
517        let Some(goal) = goals.poll_next_goal() else {
518            // No requests.  Park this worker, too.
519            return LastParkedResult::ParkSelf;
520        };
521
522        match goal {
523            WorkerGoal::Gc => {
524                trace!("A mutator requested a GC to be scheduled.");
525
526                // We set the eBPF trace point here so that bpftrace scripts can start recording
527                // work packet events before the `ScheduleCollection` work packet starts.
528                probe!(mmtk, gc_start);
529
530                {
531                    let mut gc_start_time = worker.mmtk.state.gc_start_time.borrow_mut();
532                    assert!(gc_start_time.is_none(), "GC already started?");
533                    *gc_start_time = Some(Instant::now());
534                }
535
536                self.add_schedule_collection_packet();
537                LastParkedResult::WakeSelf
538            }
539            WorkerGoal::StopForFork | WorkerGoal::Shutdown => {
540                trace!("A mutator requested {:?}", goal);
541                LastParkedResult::WakeAll
542            }
543        }
544    }
545
546    /// Find more work for workers to do.  Return true if more work is available.
547    fn find_more_work_for_workers(&self) -> bool {
548        if self.worker_group.has_designated_work() {
549            trace!("Some workers have designated work.");
550            return true;
551        }
552
553        // See if any bucket has a sentinel.
554        if self.schedule_sentinels() {
555            trace!("Some sentinels are scheduled.");
556            return true;
557        }
558
559        // Try to open new buckets.
560        if self.update_buckets() {
561            trace!("Some buckets are opened.");
562            return true;
563        }
564
565        // If all of the above failed, it means GC has finished.
566        false
567    }
568
569    /// Called when GC has finished, i.e. when all work packets have been executed.
570    ///
571    /// Return `true` if any concurrent work packets have been scheduled.
572    fn on_gc_finished(&self, worker: &GCWorker<VM>) -> bool {
573        // All GC workers must have parked by now.
574        debug_assert!(!self.worker_group.has_designated_work());
575        self.debug_assert_all_stw_buckets_empty();
576
577        // Close all work buckets to prepare for the next GC.
578        self.close_all_stw_buckets();
579        self.debug_assert_all_stw_buckets_closed();
580
581        let mmtk = worker.mmtk;
582
583        // Tell GC trigger that GC ended - this happens before we resume mutators.
584        mmtk.gc_trigger.policy.on_pause_end(mmtk);
585
586        // All other workers are parked, so it is safe to access the Plan instance mutably.
587        probe!(mmtk, plan_end_of_gc_begin);
588        let plan_mut: &mut dyn Plan<VM = VM> = unsafe { mmtk.get_plan_mut() };
589        // This also tells the GC trigger whether the GC cycle has ended (see
590        // `Plan::end_of_pause`).
591        plan_mut.end_of_pause(mmtk, worker.tls);
592        probe!(mmtk, plan_end_of_gc_end);
593
594        // Compute the elapsed time of the GC.
595        let start_time = {
596            let mut gc_start_time = worker.mmtk.state.gc_start_time.borrow_mut();
597            gc_start_time.take().expect("GC not started yet?")
598        };
599        let elapsed = start_time.elapsed();
600
601        info!(
602            "End of GC ({}/{} pages, took {:.2} ms)",
603            mmtk.get_plan().get_reserved_pages(),
604            mmtk.get_plan().get_total_pages(),
605            elapsed.as_secs_f64() * 1000.0
606        );
607
608        // USDT tracepoint for the end of GC.
609        probe!(mmtk, gc_end);
610
611        if *mmtk.get_options().count_live_bytes_in_gc {
612            // Aggregate the live bytes
613            let live_bytes = mmtk
614                .scheduler
615                .worker_group
616                .get_and_clear_worker_live_bytes();
617            let mut live_bytes_in_last_gc = mmtk.state.live_bytes_in_last_gc.borrow_mut();
618            *live_bytes_in_last_gc = mmtk.aggregate_live_bytes_in_last_gc(live_bytes);
619            // Logging
620            for (space_name, &stats) in live_bytes_in_last_gc.iter() {
621                info!(
622                    "{} = {} pages ({:.1}% live)",
623                    space_name,
624                    stats.used_pages,
625                    stats.live_bytes as f64 * 100.0 / stats.used_bytes as f64,
626                );
627            }
628        }
629
630        mmtk.state
631            .set_used_pages_after_last_gc(mmtk.get_plan().get_used_pages());
632
633        #[cfg(feature = "extreme_assertions")]
634        if crate::util::slot_logger::should_check_duplicate_slots(mmtk.get_plan()) {
635            // reset the logging info at the end of each GC
636            mmtk.slot_logger.reset();
637        }
638
639        // Reset the triggering information.
640        mmtk.state.reset_collection_trigger();
641
642        let concurrent_work_scheduled = self.schedule_concurrent_packets();
643        self.debug_assert_all_stw_buckets_closed();
644
645        // Set to NotInGC after everything, and right before resuming mutators.
646        if concurrent_work_scheduled {
647            mmtk.state.gc_status.set_in_concurrent_gc();
648            // Going to start concurrent GC. Set the number of active workers to the configured number of concurrent threads.
649            self.set_active_workers(*mmtk.options.concurrent_threads);
650            debug!(
651                "Concurrent work started. Active worker count set from concurrent_threads={}.",
652                *mmtk.options.concurrent_threads
653            );
654        } else {
655            mmtk.state.gc_status.set_not_in_gc();
656        }
657        if mmtk.stats.get_gathering_stats() {
658            mmtk.stats.end_gc();
659        }
660        <VM as VMBinding>::VMCollection::resume_mutators(worker.tls);
661
662        concurrent_work_scheduled
663    }
664
665    pub fn enable_stat(&self) {
666        for worker in &self.worker_group.workers_shared {
667            let worker_stat = worker.borrow_stat();
668            worker_stat.enable();
669        }
670    }
671
672    pub fn statistics(&self) -> HashMap<String, String> {
673        let mut summary = SchedulerStat::default();
674        for worker in &self.worker_group.workers_shared {
675            let worker_stat = worker.borrow_stat();
676            summary.merge(&worker_stat);
677        }
678        summary.harness_stat()
679    }
680
681    pub fn notify_mutators_paused(&self, mmtk: &'static MMTK<VM>) {
682        mmtk.state.gc_status.set_in_pause();
683        // The number of active workers should be either the configured number of threads or the configured number of concurrent threads.
684        assert!(
685            mmtk.scheduler.num_active_workers() == *mmtk.options.threads
686                || mmtk.scheduler.num_active_workers() == *mmtk.options.concurrent_threads
687        );
688        // Pause started, use all GC threads.
689        mmtk.scheduler.set_active_workers(*mmtk.options.threads);
690
691        let first_stw_bucket = &self.work_buckets[WorkBucketStage::FIRST_STW_STAGE];
692        debug_assert!(!first_stw_bucket.is_open());
693        // Note: This is the only place where a bucket is opened without having all workers parked.
694        // We usually require all workers to park before opening new buckets because otherwise
695        // packets will be executed out of order.  However, since `Prepare` is the first STW
696        // bucket, and all subsequent buckets require all workers to park before opening, workers
697        // cannot execute work packets out of order.  This is not generally true if we are not
698        // opening the first STW bucket.  In the future, we should redesign the opening condition
699        // of work buckets to make the synchronization more robust,
700        first_stw_bucket.open();
701        self.worker_monitor.notify_work_available(true);
702    }
703
704    pub(super) fn schedule_concurrent_packets(&self) -> bool {
705        let concurrent_bucket = &self.work_buckets[WorkBucketStage::Concurrent];
706        if !concurrent_bucket.is_empty() {
707            concurrent_bucket.set_enabled(true);
708            concurrent_bucket.open();
709            true
710        } else {
711            concurrent_bucket.set_enabled(false);
712            concurrent_bucket.close();
713            false
714        }
715    }
716}