Weak reference processing (#564)

This PR revamps the reference processor, and schedules reference processing work. This PR is one step towards #544: it gets the old reference processor working again.

There is some more work needed for weak reference processing. So `no_reference_types` is `true` and by default, we do not process weak references.

Changes
* Rework on `ReferenceProcessor`:
  * retain soft references properly, depending on whether the collection is an emergency collection or not.
  * add a separate reference enqueue step. For references that need to be enqueue'd (by Java semantics), the reference processor will store them in a buffer, and call to the binding at the end of a GC to enqueue the references. This makes sure the references are valid during the GC.
  * reference table is now an `HashSet` rather than `Vec`. We used to assume weak references are added to MMTk once they are created, so for one reference, it will be added once. Now we allow adding weak references during GC, and a weak reference may be traced multiple times in a GC, and added multiple times to MMTk. Using `HashSet` can deduplicate the references.
  * Remove unnecessary use of `UnsafeCell` for `ReferenceProcessorSync`.
  * Remove unnecessary `unforwarded_references`: I suspect it was intended to deal with meta-circularity in JikesRVM/Java MMTk (https://github.com/JikesRVM/JikesRVM/blob/5072f19761115d987b6ee162f49a03522d36c697/MMTk/ext/vm/jikesrvm/org/jikesrvm/mm/mmtk/ReferenceProcessor.java#L96), and is no longer needed for MMTk core. For MMTk core, any Rust struct and field is 'untraced' unless we specifically 'trace' it.
  * Add a `allow_new_candidate` field to avoid adding new candidates after we finish processing references. See the comments for the field for details.
  * `process_reference()` is moved from `ReferenceGlue` (in VMBinding traits) to `ReferenceProcessor`.
  * `enqueue_references()` is added to `ReferenceGlue`.
* Add a few work packets for weak reference processing. They are simply wrapping the reference processor's methods, and are single threaded.
* Add a few more work buckets for weak reference processing: `SoftRefClosure`, `WeakRefClosure`, `PhantomRefClosure`, `FinalRefClosure`, `FinalizableForwarding`.
* Schedule the weak reference processing work properly.
* Rename `ProcessWeakRef` (which call `VMBinding::VMCollection::process_weak_refs()`) to `VMProcessWeakRef`.
* `add_soft/weak/phantom_candidate()` now no longer needs a `referent` argument.

examples/mmtk.h

@@ -107,13 +107,13 @@ extern void* mmtk_starting_heap_address();
 extern void* mmtk_last_heap_address();
 
 // Add a reference to the list of weak references
-extern void mmtk_add_weak_candidate(void* ref, void* referent);
+extern void mmtk_add_weak_candidate(void* ref);
 
 // Add a reference to the list of soft references
-extern void mmtk_add_soft_candidate(void* ref, void* referent);
+extern void mmtk_add_soft_candidate(void* ref);
 
 // Add a reference to the list of phantom references
-extern void mmtk_add_phantom_candidate(void* ref, void* referent);
+extern void mmtk_add_phantom_candidate(void* ref);
 
 // Generic hook to allow benchmarks to be harnessed
 extern void mmtk_harness_begin(void* tls);

src/memory_manager.rs

@@ -416,49 +416,34 @@ pub fn modify_check<VM: VMBinding>(mmtk: &MMTK<VM>, object: ObjectReference) {
     mmtk.plan.modify_check(object);
 }
 
-/// Add a reference to the list of weak references.
+/// Add a reference to the list of weak references. A binding may
+/// call this either when a weak reference is created, or when a weak reference is traced during GC.
 ///
 /// Arguments:
 /// * `mmtk`: A reference to an MMTk instance.
 /// * `reff`: The weak reference to add.
-/// * `referent`: The object that the reference points to.
-pub fn add_weak_candidate<VM: VMBinding>(
-    mmtk: &MMTK<VM>,
-    reff: ObjectReference,
-    referent: ObjectReference,
-) {
-    mmtk.reference_processors
-        .add_weak_candidate::<VM>(reff, referent);
+pub fn add_weak_candidate<VM: VMBinding>(mmtk: &MMTK<VM>, reff: ObjectReference) {
+    mmtk.reference_processors.add_weak_candidate::<VM>(reff);
 }
 
-/// Add a reference to the list of soft references.
+/// Add a reference to the list of soft references. A binding may
+/// call this either when a weak reference is created, or when a weak reference is traced during GC.
 ///
 /// Arguments:
 /// * `mmtk`: A reference to an MMTk instance.
 /// * `reff`: The soft reference to add.
-/// * `referent`: The object that the reference points to.
-pub fn add_soft_candidate<VM: VMBinding>(
-    mmtk: &MMTK<VM>,
-    reff: ObjectReference,
-    referent: ObjectReference,
-) {
-    mmtk.reference_processors
-        .add_soft_candidate::<VM>(reff, referent);
+pub fn add_soft_candidate<VM: VMBinding>(mmtk: &MMTK<VM>, reff: ObjectReference) {
+    mmtk.reference_processors.add_soft_candidate::<VM>(reff);
 }
 
-/// Add a reference to the list of phantom references.
+/// Add a reference to the list of phantom references. A binding may
+/// call this either when a weak reference is created, or when a weak reference is traced during GC.
 ///
 /// Arguments:
 /// * `mmtk`: A reference to an MMTk instance.
 /// * `reff`: The phantom reference to add.
-/// * `referent`: The object that the reference points to.
-pub fn add_phantom_candidate<VM: VMBinding>(
-    mmtk: &MMTK<VM>,
-    reff: ObjectReference,
-    referent: ObjectReference,
-) {
-    mmtk.reference_processors
-        .add_phantom_candidate::<VM>(reff, referent);
+pub fn add_phantom_candidate<VM: VMBinding>(mmtk: &MMTK<VM>, reff: ObjectReference) {
+    mmtk.reference_processors.add_phantom_candidate::<VM>(reff);
 }
 
 /// Generic hook to allow benchmarks to be harnessed. We do a full heap

src/mmtk.rs

@@ -101,6 +101,11 @@ impl<VM: VMBinding> MMTK<VM> {
     pub fn get_plan(&self) -> &dyn Plan<VM = VM> {
         self.plan.as_ref()
     }
+
+    #[inline(always)]
+    pub fn get_options(&self) -> &Options {
+        &self.options
+    }
 }
 
 impl<VM: VMBinding> Default for MMTK<VM> {

src/plan/markcompact/global.rs

@@ -97,10 +97,6 @@ impl<VM: VMBinding> Plan for MarkCompact<VM> {
         scheduler.work_buckets[WorkBucketStage::Prepare]
             .add(Prepare::<MarkCompactGCWorkContext<VM>>::new(self));
 
-        // VM-specific weak ref processing
-        scheduler.work_buckets[WorkBucketStage::RefClosure]
-            .add(ProcessWeakRefs::<MarkingProcessEdges<VM>>::new());
-
         scheduler.work_buckets[WorkBucketStage::CalculateForwarding]
             .add(CalculateForwardingAddress::<VM>::new(&self.mc_space));
         // do another trace to update references
@@ -111,17 +107,41 @@ impl<VM: VMBinding> Plan for MarkCompact<VM> {
         scheduler.work_buckets[WorkBucketStage::Release]
             .add(Release::<MarkCompactGCWorkContext<VM>>::new(self));
 
+        // Reference processing
+        if !*self.base().options.no_reference_types {
+            use crate::util::reference_processor::{
+                PhantomRefProcessing, SoftRefProcessing, WeakRefProcessing,
+            };
+            scheduler.work_buckets[WorkBucketStage::SoftRefClosure]
+                .add(SoftRefProcessing::<MarkingProcessEdges<VM>>::new());
+            scheduler.work_buckets[WorkBucketStage::WeakRefClosure]
+                .add(WeakRefProcessing::<MarkingProcessEdges<VM>>::new());
+            scheduler.work_buckets[WorkBucketStage::PhantomRefClosure]
+                .add(PhantomRefProcessing::<MarkingProcessEdges<VM>>::new());
+
+            // VM-specific weak ref processing
+            scheduler.work_buckets[WorkBucketStage::WeakRefClosure]
+                .add(VMProcessWeakRefs::<MarkingProcessEdges<VM>>::new());
+
+            use crate::util::reference_processor::RefForwarding;
+            scheduler.work_buckets[WorkBucketStage::RefForwarding]
+                .add(RefForwarding::<ForwardingProcessEdges<VM>>::new());
+
+            use crate::util::reference_processor::RefEnqueue;
+            scheduler.work_buckets[WorkBucketStage::Release].add(RefEnqueue::<VM>::new());
+        }
+
         // Finalization
         if !*self.base().options.no_finalizer {
             use crate::util::finalizable_processor::{Finalization, ForwardFinalization};
             // finalization
             // treat finalizable objects as roots and perform a closure (marking)
             // must be done before calculating forwarding pointers
-            scheduler.work_buckets[WorkBucketStage::RefClosure]
+            scheduler.work_buckets[WorkBucketStage::FinalRefClosure]
                 .add(Finalization::<MarkingProcessEdges<VM>>::new());
             // update finalizable object references
             // must be done before compacting
-            scheduler.work_buckets[WorkBucketStage::RefForwarding]
+            scheduler.work_buckets[WorkBucketStage::FinalizableForwarding]
                 .add(ForwardFinalization::<ForwardingProcessEdges<VM>>::new());
         }
 

src/plan/plan_constraints.rs

@@ -30,6 +30,8 @@ pub struct PlanConstraints {
     pub needs_linear_scan: bool,
     pub needs_concurrent_workers: bool,
     pub generate_gc_trace: bool,
+    /// Some policies do object forwarding after the first liveness transitive closure, such as mark compact.
+    /// For plans that use those policies, they should set this as true.
     pub needs_forward_after_liveness: bool,
 }
 

src/scheduler/gc_work.rs

@@ -121,8 +121,6 @@ impl<C: GCWorkContext + 'static> GCWork<C::VM> for Release<C> {
         for w in &mmtk.scheduler.workers_shared {
             w.local_work_bucket.add(ReleaseCollector);
         }
-        // TODO: Process weak references properly
-        mmtk.reference_processors.clear();
     }
 }
 
@@ -250,15 +248,15 @@ impl<VM: VMBinding> CoordinatorWork<VM> for EndOfGC {}
 /// processing of those weakrefs may be more complex. For such case, we delegate to the
 /// VM binding to process weak references.
 #[derive(Default)]
-pub struct ProcessWeakRefs<E: ProcessEdgesWork>(PhantomData<E>);
+pub struct VMProcessWeakRefs<E: ProcessEdgesWork>(PhantomData<E>);
 
-impl<E: ProcessEdgesWork> ProcessWeakRefs<E> {
+impl<E: ProcessEdgesWork> VMProcessWeakRefs<E> {
     pub fn new() -> Self {
         Self(PhantomData)
     }
 }
 
-impl<E: ProcessEdgesWork> GCWork<E::VM> for ProcessWeakRefs<E> {
+impl<E: ProcessEdgesWork> GCWork<E::VM> for VMProcessWeakRefs<E> {
     fn do_work(&mut self, worker: &mut GCWorker<E::VM>, _mmtk: &'static MMTK<E::VM>) {
         trace!("ProcessWeakRefs");
         <E::VM as VMBinding>::VMCollection::process_weak_refs::<E>(worker);
@@ -384,6 +382,13 @@ impl<VM: VMBinding> ProcessEdgesBase<VM> {
 }
 
 /// Scan & update a list of object slots
+//
+// Note: be very careful when using this trait. process_node() will push objects
+// to the buffer, and it is expected that at the end of the operation, flush()
+// is called to create new scan work from the buffered objects. If flush()
+// is not called, we may miss the objects in the GC and have dangling pointers.
+// FIXME: We possibly want to enforce Drop on this trait, and require calling
+// flush() in Drop.
 pub trait ProcessEdgesWork:
     Send + 'static + Sized + DerefMut + Deref<Target = ProcessEdgesBase<Self::VM>>
 {

src/scheduler/scheduler.rs

@@ -57,10 +57,14 @@ impl<VM: VMBinding> GCWorkScheduler<VM> {
             WorkBucketStage::Unconstrained => WorkBucket::new(true, worker_monitor.clone()),
             WorkBucketStage::Prepare => WorkBucket::new(false, worker_monitor.clone()),
             WorkBucketStage::Closure => WorkBucket::new(false, worker_monitor.clone()),
-            WorkBucketStage::RefClosure => WorkBucket::new(false, worker_monitor.clone()),
+            WorkBucketStage::SoftRefClosure => WorkBucket::new(false, worker_monitor.clone()),
+            WorkBucketStage::WeakRefClosure => WorkBucket::new(false, worker_monitor.clone()),
+            WorkBucketStage::FinalRefClosure => WorkBucket::new(false, worker_monitor.clone()),
+            WorkBucketStage::PhantomRefClosure => WorkBucket::new(false, worker_monitor.clone()),
             WorkBucketStage::CalculateForwarding => WorkBucket::new(false, worker_monitor.clone()),
             WorkBucketStage::SecondRoots => WorkBucket::new(false, worker_monitor.clone()),
             WorkBucketStage::RefForwarding => WorkBucket::new(false, worker_monitor.clone()),
+            WorkBucketStage::FinalizableForwarding => WorkBucket::new(false, worker_monitor.clone()),
             WorkBucketStage::Compact => WorkBucket::new(false, worker_monitor.clone()),
             WorkBucketStage::Release => WorkBucket::new(false, worker_monitor.clone()),
             WorkBucketStage::Final => WorkBucket::new(false, worker_monitor.clone()),
@@ -78,7 +82,7 @@ impl<VM: VMBinding> GCWorkScheduler<VM> {
                     let should_open = scheduler.are_buckets_drained(&cur_stages)
                         && scheduler.all_workers_parked();
                     // Additional check before the `RefClosure` bucket opens.
-                    if should_open && s == WorkBucketStage::RefClosure {
+                    if should_open && s == crate::scheduler::work_bucket::LAST_CLOSURE_BUCKET {
                         if let Some(closure_end) = scheduler.closure_end.lock().unwrap().as_ref() {
                             if closure_end() {
                                 // Don't open `RefClosure` if `closure_end` added more works to `Closure`.
@@ -92,10 +96,14 @@ impl<VM: VMBinding> GCWorkScheduler<VM> {
             };
 
             open_next(Closure);
-            open_next(RefClosure);
+            open_next(SoftRefClosure);
+            open_next(WeakRefClosure);
+            open_next(FinalRefClosure);
+            open_next(PhantomRefClosure);
             open_next(CalculateForwarding);
             open_next(SecondRoots);
             open_next(RefForwarding);
+            open_next(FinalizableForwarding);
             open_next(Compact);
             open_next(Release);
             open_next(Final);
@@ -183,10 +191,6 @@ impl<VM: VMBinding> GCWorkScheduler<VM> {
         // Prepare global/collectors/mutators
         self.work_buckets[WorkBucketStage::Prepare].add(Prepare::<C>::new(plan));
 
-        // VM-specific weak ref processing
-        self.work_buckets[WorkBucketStage::RefClosure]
-            .add(ProcessWeakRefs::<C::ProcessEdgesWorkType>::new());
-
         // Release global/collectors/mutators
         self.work_buckets[WorkBucketStage::Release].add(Release::<C>::new(plan));
 
@@ -205,15 +209,41 @@ impl<VM: VMBinding> GCWorkScheduler<VM> {
                 .add(ScheduleSanityGC::<C::PlanType>::new(plan));
         }
 
+        // Reference processing
+        if !*plan.base().options.no_reference_types {
+            use crate::util::reference_processor::{
+                PhantomRefProcessing, SoftRefProcessing, WeakRefProcessing,
+            };
+            self.work_buckets[WorkBucketStage::SoftRefClosure]
+                .add(SoftRefProcessing::<C::ProcessEdgesWorkType>::new());
+            self.work_buckets[WorkBucketStage::WeakRefClosure]
+                .add(WeakRefProcessing::<C::ProcessEdgesWorkType>::new());
+            self.work_buckets[WorkBucketStage::PhantomRefClosure]
+                .add(PhantomRefProcessing::<C::ProcessEdgesWorkType>::new());
+
+            // VM-specific weak ref processing
+            self.work_buckets[WorkBucketStage::WeakRefClosure]
+                .add(VMProcessWeakRefs::<C::ProcessEdgesWorkType>::new());
+
+            use crate::util::reference_processor::RefForwarding;
+            if plan.constraints().needs_forward_after_liveness {
+                self.work_buckets[WorkBucketStage::RefForwarding]
+                    .add(RefForwarding::<C::ProcessEdgesWorkType>::new());
+            }
+
+            use crate::util::reference_processor::RefEnqueue;
+            self.work_buckets[WorkBucketStage::Release].add(RefEnqueue::<VM>::new());
+        }
+
         // Finalization
         if !*plan.base().options.no_finalizer {
             use crate::util::finalizable_processor::{Finalization, ForwardFinalization};
             // finalization
-            self.work_buckets[WorkBucketStage::RefClosure]
+            self.work_buckets[WorkBucketStage::FinalRefClosure]
                 .add(Finalization::<C::ProcessEdgesWorkType>::new());
             // forward refs
             if plan.constraints().needs_forward_after_liveness {
-                self.work_buckets[WorkBucketStage::RefForwarding]
+                self.work_buckets[WorkBucketStage::FinalizableForwarding]
                     .add(ForwardFinalization::<C::ProcessEdgesWorkType>::new());
             }
         }

src/scheduler/work_bucket.rs

@@ -154,13 +154,17 @@ pub enum WorkBucketStage {
     Unconstrained,
     Prepare,
     Closure,
-    // TODO: We only support final reference at the moment. If we have references of multiple strengths,
-    // we may need more than one buckets for each reference strength.
-    RefClosure,
+    SoftRefClosure,
+    WeakRefClosure,
+    FinalRefClosure,
+    PhantomRefClosure,
     CalculateForwarding,
     SecondRoots,
     RefForwarding,
+    FinalizableForwarding,
     Compact,
     Release,
     Final,
 }
+
+pub const LAST_CLOSURE_BUCKET: WorkBucketStage = WorkBucketStage::PhantomRefClosure;

src/util/address.rs

@@ -453,6 +453,8 @@ mod tests {
 pub struct ObjectReference(usize);
 
 impl ObjectReference {
+    pub const NULL: ObjectReference = ObjectReference(0);
+
     /// converts the ObjectReference to an Address
     #[inline(always)]
     pub fn to_address(self) -> Address {

src/util/options.rs

@@ -331,7 +331,11 @@ options! {
     // Should finalization be disabled?
     no_finalizer:          bool                 [env_var: true, command_line: true]  [always_valid] = false,
     // Should reference type processing be disabled?
-    no_reference_types:    bool                 [env_var: true, command_line: true]  [always_valid] = false,
+    // If reference type processing is disabled, no weak reference processing work is scheduled,
+    // and we expect a binding to treat weak references as strong references.
+    // We disable weak reference processing by default, as we are still working on it. This will be changed to `false`
+    // once weak reference processing is implemented properly.
+    no_reference_types:    bool                 [env_var: true, command_line: true]  [always_valid] = true,
     // The zeroing approach to use for new object allocations. Affects each plan differently. (not supported)
     nursery_zeroing:       NurseryZeroingOptions[env_var: true, command_line: true]  [always_valid] = NurseryZeroingOptions::Temporal,
     // How frequent (every X bytes) should we do a stress GC?