Concurrent immix

Cargo.toml

@@ -47,6 +47,7 @@ portable-atomic = "1.4.3"
 probe = "0.5"
 regex = "1.7.0"
 rustversion = "1.0"
+rayon-core = "=1.12.1" # We can remove this dependency when we use MSRV 1.80+
 spin = "0.9.5"
 static_assertions = "1.1.0"
 strum = "0.27.1"

src/global_state.rs

@@ -49,6 +49,7 @@ pub struct GlobalState {
     pub(crate) malloc_bytes: AtomicUsize,
     /// This stores the live bytes and the used bytes (by pages) for each space in last GC. This counter is only updated in the GC release phase.
     pub(crate) live_bytes_in_last_gc: AtomicRefCell<HashMap<&'static str, LiveBytesStats>>,
+    pub(crate) concurrent_marking_threshold: AtomicUsize,
 }
 
 impl GlobalState {
@@ -206,6 +207,7 @@ impl Default for GlobalState {
             #[cfg(feature = "malloc_counted_size")]
             malloc_bytes: AtomicUsize::new(0),
             live_bytes_in_last_gc: AtomicRefCell::new(HashMap::new()),
+            concurrent_marking_threshold: AtomicUsize::new(0),
         }
     }
 }

src/lib.rs

@@ -32,6 +32,8 @@ extern crate static_assertions;
 extern crate probe;
 
 mod mmtk;
+use std::sync::atomic::AtomicUsize;
+
 pub use mmtk::MMTKBuilder;
 pub(crate) use mmtk::MMAPPER;
 pub use mmtk::MMTK;
@@ -51,3 +53,9 @@ pub mod vm;
 pub use crate::plan::{
     AllocationSemantics, BarrierSelector, Mutator, MutatorContext, ObjectQueue, Plan,
 };
+
+static NUM_CONCURRENT_TRACING_PACKETS: AtomicUsize = AtomicUsize::new(0);
+
+fn concurrent_marking_packets_drained() -> bool {
+    crate::NUM_CONCURRENT_TRACING_PACKETS.load(std::sync::atomic::Ordering::SeqCst) == 0
+}

src/plan/barriers.rs

@@ -21,6 +21,7 @@ pub enum BarrierSelector {
     NoBarrier,
     /// Object remembering barrier is used.
     ObjectBarrier,
+    SATBBarrier,
 }
 
 impl BarrierSelector {
@@ -45,6 +46,19 @@ impl BarrierSelector {
 pub trait Barrier<VM: VMBinding>: 'static + Send + Downcast {
     fn flush(&mut self) {}
 
+    /// Weak reference loading barrier.  A mutator should call this when loading from a weak
+    /// reference field, for example, when executing  `java.lang.ref.Reference.get()` in JVM, or
+    /// loading from a global weak table in CRuby.
+    ///
+    /// Note: Merely loading from a field holding weak reference into a local variable will create a
+    /// strong reference from the stack to the referent, changing its reachablilty from weakly
+    /// reachable to strongly reachable.  Concurrent garbage collectors may need to handle such
+    /// events specially.  See [SATBBarrier::load_weak_reference] for a concrete example.
+    ///
+    /// Arguments:
+    /// *   `referent`: The referent object which the weak reference is pointing to.
+    fn load_weak_reference(&mut self, _referent: ObjectReference) {}
+
     /// Subsuming barrier for object reference write
     fn object_reference_write(
         &mut self,
@@ -92,6 +106,8 @@ pub trait Barrier<VM: VMBinding>: 'static + Send + Downcast {
         self.memory_region_copy_post(src, dst);
     }
 
+    fn object_reference_clone_pre(&mut self, _obj: ObjectReference) {}
+
     /// Full pre-barrier for array copy
     fn memory_region_copy_pre(&mut self, _src: VM::VMMemorySlice, _dst: VM::VMMemorySlice) {}
 
@@ -159,6 +175,10 @@ pub trait BarrierSemantics: 'static + Send {
 
     /// Object will probably be modified
     fn object_probable_write_slow(&mut self, _obj: ObjectReference) {}
+
+    fn load_weak_reference(&mut self, _o: ObjectReference) {}
+
+    fn object_reference_clone_pre(&mut self, _obj: ObjectReference) {}
 }
 
 /// Generic object barrier with a type argument defining it's slow-path behaviour.
@@ -250,3 +270,82 @@ impl<S: BarrierSemantics> Barrier<S::VM> for ObjectBarrier<S> {
         }
     }
 }
+
+pub struct SATBBarrier<S: BarrierSemantics> {
+    semantics: S,
+}
+
+impl<S: BarrierSemantics> SATBBarrier<S> {
+    pub fn new(semantics: S) -> Self {
+        Self { semantics }
+    }
+    fn object_is_unlogged(&self, object: ObjectReference) -> bool {
+        // unsafe { S::UNLOG_BIT_SPEC.load::<S::VM, u8>(object, None) != 0 }
+        S::UNLOG_BIT_SPEC.load_atomic::<S::VM, u8>(object, None, Ordering::SeqCst) != 0
+    }
+}
+
+impl<S: BarrierSemantics> Barrier<S::VM> for SATBBarrier<S> {
+    fn flush(&mut self) {
+        self.semantics.flush();
+    }
+
+    fn load_weak_reference(&mut self, o: ObjectReference) {
+        self.semantics.load_weak_reference(o)
+    }
+
+    fn object_reference_clone_pre(&mut self, obj: ObjectReference) {
+        self.semantics.object_reference_clone_pre(obj);
+    }
+
+    fn object_probable_write(&mut self, obj: ObjectReference) {
+        self.semantics.object_probable_write_slow(obj);
+    }
+
+    fn object_reference_write_pre(
+        &mut self,
+        src: ObjectReference,
+        slot: <S::VM as VMBinding>::VMSlot,
+        target: Option<ObjectReference>,
+    ) {
+        if self.object_is_unlogged(src) {
+            self.semantics
+                .object_reference_write_slow(src, slot, target);
+        }
+    }
+
+    fn object_reference_write_post(
+        &mut self,
+        _src: ObjectReference,
+        _slot: <S::VM as VMBinding>::VMSlot,
+        _target: Option<ObjectReference>,
+    ) {
+        unimplemented!()
+    }
+
+    fn object_reference_write_slow(
+        &mut self,
+        src: ObjectReference,
+        slot: <S::VM as VMBinding>::VMSlot,
+        target: Option<ObjectReference>,
+    ) {
+        self.semantics
+            .object_reference_write_slow(src, slot, target);
+    }
+
+    fn memory_region_copy_pre(
+        &mut self,
+        src: <S::VM as VMBinding>::VMMemorySlice,
+        dst: <S::VM as VMBinding>::VMMemorySlice,
+    ) {
+        self.semantics.memory_region_copy_slow(src, dst);
+    }
+
+    fn memory_region_copy_post(
+        &mut self,
+        _src: <S::VM as VMBinding>::VMMemorySlice,
+        _dst: <S::VM as VMBinding>::VMMemorySlice,
+    ) {
+        unimplemented!()
+    }
+}

src/plan/concurrent/barrier.rs

@@ -0,0 +1,152 @@
+use std::sync::atomic::Ordering;
+
+use super::{concurrent_marking_work::ProcessModBufSATB, Pause};
+use crate::plan::global::PlanTraceObject;
+use crate::{
+    plan::{barriers::BarrierSemantics, concurrent::global::ConcurrentPlan, VectorQueue},
+    scheduler::WorkBucketStage,
+    util::ObjectReference,
+    vm::{
+        slot::{MemorySlice, Slot},
+        VMBinding,
+    },
+    MMTK,
+};
+
+pub struct SATBBarrierSemantics<VM: VMBinding, P: ConcurrentPlan<VM = VM> + PlanTraceObject<VM>> {
+    mmtk: &'static MMTK<VM>,
+    satb: VectorQueue<ObjectReference>,
+    refs: VectorQueue<ObjectReference>,
+    plan: &'static P,
+}
+
+impl<VM: VMBinding, P: ConcurrentPlan<VM = VM> + PlanTraceObject<VM>> SATBBarrierSemantics<VM, P> {
+    pub fn new(mmtk: &'static MMTK<VM>) -> Self {
+        Self {
+            mmtk,
+            satb: VectorQueue::default(),
+            refs: VectorQueue::default(),
+            plan: mmtk.get_plan().downcast_ref::<P>().unwrap(),
+        }
+    }
+
+    fn slow(&mut self, _src: Option<ObjectReference>, _slot: VM::VMSlot, old: ObjectReference) {
+        self.satb.push(old);
+        if self.satb.is_full() {
+            self.flush_satb();
+        }
+    }
+
+    fn enqueue_node(
+        &mut self,
+        src: Option<ObjectReference>,
+        slot: VM::VMSlot,
+        _new: Option<ObjectReference>,
+    ) -> bool {
+        if let Some(old) = slot.load() {
+            self.slow(src, slot, old);
+        }
+        true
+    }
+
+    /// Attempt to atomically log an object.
+    /// Returns true if the object is not logged previously.
+    fn log_object(&self, object: ObjectReference) -> bool {
+        Self::UNLOG_BIT_SPEC.store_atomic::<VM, u8>(object, 0, None, Ordering::SeqCst);
+        true
+    }
+
+    fn flush_satb(&mut self) {
+        if !self.satb.is_empty() {
+            if self.should_create_satb_packets() {
+                let satb = self.satb.take();
+                let bucket = if self.plan.concurrent_work_in_progress() {
+                    WorkBucketStage::Unconstrained
+                } else {
+                    debug_assert_ne!(self.plan.current_pause(), Some(Pause::InitialMark));
+                    WorkBucketStage::Closure
+                };
+                self.mmtk.scheduler.work_buckets[bucket].add(ProcessModBufSATB::<VM, P>::new(satb));
+            } else {
+                let _ = self.satb.take();
+            };
+        }
+    }
+
+    #[cold]
+    fn flush_weak_refs(&mut self) {
+        if !self.refs.is_empty() {
+            // debug_assert!(self.should_create_satb_packets());
+            let nodes = self.refs.take();
+            let bucket = if self.plan.concurrent_work_in_progress() {
+                WorkBucketStage::Unconstrained
+            } else {
+                debug_assert_ne!(self.plan.current_pause(), Some(Pause::InitialMark));
+                WorkBucketStage::Closure
+            };
+            self.mmtk.scheduler.work_buckets[bucket].add(ProcessModBufSATB::<VM, P>::new(nodes));
+        }
+    }
+
+    fn should_create_satb_packets(&self) -> bool {
+        self.plan.concurrent_work_in_progress()
+            || self.plan.current_pause() == Some(Pause::FinalMark)
+    }
+}
+
+impl<VM: VMBinding, P: ConcurrentPlan<VM = VM> + PlanTraceObject<VM>> BarrierSemantics
+    for SATBBarrierSemantics<VM, P>
+{
+    type VM = VM;
+
+    #[cold]
+    fn flush(&mut self) {
+        self.flush_satb();
+        self.flush_weak_refs();
+    }
+
+    fn object_reference_write_slow(
+        &mut self,
+        src: ObjectReference,
+        _slot: <Self::VM as VMBinding>::VMSlot,
+        _target: Option<ObjectReference>,
+    ) {
+        self.object_probable_write_slow(src);
+        self.log_object(src);
+    }
+
+    fn memory_region_copy_slow(
+        &mut self,
+        _src: <Self::VM as VMBinding>::VMMemorySlice,
+        dst: <Self::VM as VMBinding>::VMMemorySlice,
+    ) {
+        for s in dst.iter_slots() {
+            self.enqueue_node(None, s, None);
+        }
+    }
+
+    /// Enqueue the referent during concurrent marking.
+    ///
+    /// Note: During concurrent marking, a collector based on snapshot-at-the-beginning (SATB) will
+    /// not reach objects that were weakly reachable at the time of `InitialMark`.  But if a mutator
+    /// loads from a weak reference field during concurrent marking, it will make the referent
+    /// strongly reachable, yet the referent is still not part of the SATB.  We must conservatively
+    /// enqueue the referent even though its reachability has not yet been established, otherwise it
+    /// (and its children) may be treated as garbage if it happened to be weakly reachable at the
+    /// time of `InitialMark`.
+    fn load_weak_reference(&mut self, o: ObjectReference) {
+        if !self.plan.concurrent_work_in_progress() {
+            return;
+        }
+        self.refs.push(o);
+        if self.refs.is_full() {
+            self.flush_weak_refs();
+        }
+    }
+
+    fn object_probable_write_slow(&mut self, obj: ObjectReference) {
+        obj.iterate_fields::<VM, _>(|s| {
+            self.enqueue_node(Some(obj), s, None);
+        });
+    }
+}