Add a parallel/regional Compressor

.github/scripts/ci-test.sh

@@ -12,15 +12,10 @@ if [[ $arch == "x86_64" && $os == "linux" ]]; then
 fi
 
 ALL_PLANS=$(sed -n '/enum PlanSelector/,/}/p' src/util/options.rs | sed -e 's;//.*;;g' -e '/^$/d' -e 's/,//g' | xargs | grep -o '{.*}' | grep -o '\w\+')
-ALL_DISCONTIGUOUS_PLANS=$(echo -n "$ALL_PLANS" | sed '/Compressor/d')
-
-if [[ $arch == "x86" ]]; then
-    ALL_PLANS=$ALL_DISCONTIGUOUS_PLANS
-fi
 
 # At the moment, the Compressor does not work with the mock VM tests.
 # So we skip testing the Compressor entirely.
-ALL_PLANS=$ALL_DISCONTIGUOUS_PLANS
+ALL_PLANS=$(echo -n "$ALL_PLANS" | sed '/Compressor/d')
 
 # Test with mock VM:
 # - Find all the files that start with mock_test_
@@ -34,8 +29,6 @@ find ./src ./tests -type f -name "mock_test_*" | while read -r file; do
     PLANS=$(sed -n 's/^\/\/ *GITHUB-CI: *MMTK_PLAN=//p' $file | tr ',' '\n')
     if [[ $PLANS == 'all' ]]; then
         PLANS=$ALL_PLANS
-    elif [[ $PLANS == 'discontiguous' ]]; then
-        PLANS=$ALL_DISCONTIGUOUS_PLANS
     elif [[ -z $PLANS ]]; then
         PLANS=NoGC
     fi

src/plan/compressor/gc_work.rs

@@ -1,32 +1,33 @@
 use super::global::Compressor;
-use crate::policy::compressor::CompressorSpace;
-use crate::policy::compressor::{TRACE_KIND_FORWARD_ROOT, TRACE_KIND_MARK};
+use crate::policy::compressor::{CompressorSpace, TRACE_KIND_FORWARD_ROOT, TRACE_KIND_MARK};
 use crate::policy::largeobjectspace::LargeObjectSpace;
 use crate::scheduler::gc_work::PlanProcessEdges;
 use crate::scheduler::gc_work::*;
-use crate::scheduler::GCWork;
-use crate::scheduler::GCWorker;
-use crate::scheduler::WorkBucketStage;
-use crate::vm::ActivePlan;
-use crate::vm::Scanning;
-use crate::vm::VMBinding;
+use crate::scheduler::{GCWork, GCWorker, WorkBucketStage};
+use crate::vm::{ActivePlan, Scanning, VMBinding};
 use crate::MMTK;
-use std::marker::PhantomData;
+use std::marker::{PhantomData, Send};
 
-/// Iterate through the heap and calculate the new location of live objects.
-pub struct CalculateForwardingAddress<VM: VMBinding> {
+/// Generate more packets by calling a method on [`CompressorSpace`].
+pub struct GenerateWork<VM: VMBinding, F: Fn(&'static CompressorSpace<VM>) + Send + 'static> {
     compressor_space: &'static CompressorSpace<VM>,
+    f: F,
 }
 
-impl<VM: VMBinding> GCWork<VM> for CalculateForwardingAddress<VM> {
+impl<VM: VMBinding, F: Fn(&'static CompressorSpace<VM>) + Send + 'static> GCWork<VM>
+    for GenerateWork<VM, F>
+{
     fn do_work(&mut self, _worker: &mut GCWorker<VM>, _mmtk: &'static MMTK<VM>) {
-        self.compressor_space.calculate_offset_vector();
+        (self.f)(self.compressor_space);
     }
 }
 
-impl<VM: VMBinding> CalculateForwardingAddress<VM> {
-    pub fn new(compressor_space: &'static CompressorSpace<VM>) -> Self {
-        Self { compressor_space }
+impl<VM: VMBinding, F: Fn(&'static CompressorSpace<VM>) + Send + 'static> GenerateWork<VM, F> {
+    pub fn new(compressor_space: &'static CompressorSpace<VM>, f: F) -> Self {
+        Self {
+            compressor_space,
+            f,
+        }
     }
 }
 
@@ -46,11 +47,6 @@ impl<VM: VMBinding> GCWork<VM> for UpdateReferences<VM> {
         #[cfg(feature = "extreme_assertions")]
         mmtk.slot_logger.reset();
 
-        // We do two passes of transitive closures. We clear the live bytes from the first pass.
-        mmtk.scheduler
-            .worker_group
-            .get_and_clear_worker_live_bytes();
-
         for mutator in VM::VMActivePlan::mutators() {
             mmtk.scheduler.work_buckets[WorkBucketStage::SecondRoots].add(ScanMutatorRoots::<
                 CompressorForwardingWorkContext<VM>,
@@ -68,19 +64,19 @@ impl<VM: VMBinding> UpdateReferences<VM> {
     }
 }
 
-/// Compact live objects based on the previously-calculated forwarding pointers.
-pub struct Compact<VM: VMBinding> {
+/// Reset the allocator and update references in large object space.
+pub struct AfterCompact<VM: VMBinding> {
     compressor_space: &'static CompressorSpace<VM>,
     los: &'static LargeObjectSpace<VM>,
 }
 
-impl<VM: VMBinding> GCWork<VM> for Compact<VM> {
+impl<VM: VMBinding> GCWork<VM> for AfterCompact<VM> {
     fn do_work(&mut self, worker: &mut GCWorker<VM>, _mmtk: &'static MMTK<VM>) {
-        self.compressor_space.compact(worker, self.los);
+        self.compressor_space.after_compact(worker, self.los);
     }
 }
 
-impl<VM: VMBinding> Compact<VM> {
+impl<VM: VMBinding> AfterCompact<VM> {
     pub fn new(
         compressor_space: &'static CompressorSpace<VM>,
         los: &'static LargeObjectSpace<VM>,

src/plan/compressor/global.rs

@@ -1,21 +1,17 @@
 use super::gc_work::CompressorWorkContext;
 use super::gc_work::{
-    CalculateForwardingAddress, Compact, ForwardingProcessEdges, MarkingProcessEdges,
-    UpdateReferences,
+    AfterCompact, ForwardingProcessEdges, GenerateWork, MarkingProcessEdges, UpdateReferences,
 };
 use crate::plan::compressor::mutator::ALLOCATOR_MAPPING;
 use crate::plan::global::CreateGeneralPlanArgs;
 use crate::plan::global::CreateSpecificPlanArgs;
 use crate::plan::global::{BasePlan, CommonPlan};
 use crate::plan::plan_constraints::MAX_NON_LOS_ALLOC_BYTES_COPYING_PLAN;
-use crate::plan::AllocationSemantics;
-use crate::plan::Plan;
-use crate::plan::PlanConstraints;
+use crate::plan::{AllocationSemantics, Plan, PlanConstraints};
 use crate::policy::compressor::CompressorSpace;
 use crate::policy::space::Space;
 use crate::scheduler::gc_work::*;
-use crate::scheduler::GCWorkScheduler;
-use crate::scheduler::WorkBucketStage;
+use crate::scheduler::{GCWorkScheduler, WorkBucketStage};
 use crate::util::alloc::allocators::AllocatorSelector;
 use crate::util::heap::gc_trigger::SpaceStats;
 #[allow(unused_imports)]
@@ -26,7 +22,7 @@ use crate::vm::VMBinding;
 use enum_map::EnumMap;
 use mmtk_macros::{HasSpaces, PlanTraceObject};
 
-/// Compressor implements a stop-the-world and serial implementation of
+/// [`Compressor`] implements a stop-the-world and parallel implementation of
 /// the Compressor, as described in Kermany and Petrank,
 /// [The Compressor: concurrent, incremental, and parallel compaction](https://dl.acm.org/doi/10.1145/1133255.1134023).
 #[derive(HasSpaces, PlanTraceObject)]
@@ -97,13 +93,22 @@ impl<VM: VMBinding> Plan for Compressor<VM> {
         scheduler.work_buckets[WorkBucketStage::Prepare]
             .add(Prepare::<CompressorWorkContext<VM>>::new(self));
 
-        scheduler.work_buckets[WorkBucketStage::CalculateForwarding].add(
-            CalculateForwardingAddress::<VM>::new(&self.compressor_space),
-        );
-        // do another trace to update references
+        scheduler.work_buckets[WorkBucketStage::CalculateForwarding].add(GenerateWork::new(
+            &self.compressor_space,
+            &|space: &'static CompressorSpace<VM>| space.add_offset_vector_tasks(),
+        ));
+
+        // scan roots to update their references
         scheduler.work_buckets[WorkBucketStage::SecondRoots].add(UpdateReferences::<VM>::new());
-        scheduler.work_buckets[WorkBucketStage::Compact]
-            .add(Compact::<VM>::new(&self.compressor_space, &self.common.los));
+
+        scheduler.work_buckets[WorkBucketStage::Compact].add(GenerateWork::new(
+            &self.compressor_space,
+            &|space: &'static CompressorSpace<VM>| space.add_compact_tasks(),
+        ));
+
+        scheduler.work_buckets[WorkBucketStage::Compact].set_sentinel(Box::new(
+            AfterCompact::<VM>::new(&self.compressor_space, &self.common.los),
+        ));
 
         // Release global/collectors/mutators
         scheduler.work_buckets[WorkBucketStage::Release]