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