mmtk/plan/lxr/
barrier.rs

1//! Read/Write barrier implementations.
2
3use std::sync::Arc;
4
5use atomic::Ordering;
6
7use super::LazySweepingJobsCounter;
8use super::LXR;
9use crate::plan::barriers::BarrierSemantics;
10use crate::plan::concurrent::global::ConcurrentPlan;
11use crate::plan::concurrent::Pause;
12use crate::plan::lxr::gc_work::rc::ProcessDecs;
13use crate::plan::lxr::gc_work::rc::ProcessIncs;
14use crate::plan::lxr::gc_work::rc::EDGE_KIND_MATURE;
15use crate::plan::lxr::gc_work::tracing::ProcessModBufSATB;
16use crate::plan::VectorQueue;
17use crate::scheduler::WorkBucketStage;
18use crate::util::metadata::log_bit::{LOGGED_VALUE, UNLOGGED_VALUE};
19use crate::util::metadata::side_metadata::address_to_meta_address;
20use crate::util::metadata::side_metadata::SideMetadataSpec;
21use crate::util::*;
22use crate::vm::slot::MemorySlice;
23use crate::vm::slot::Slot;
24use crate::vm::*;
25use crate::MMTK;
26
27/// Re-arm the per-object log bits that one mutator's barrier cleared, so the next epoch's first
28/// store to each of those objects reaches the barrier again.
29#[cfg(feature = "lxr_object_log")]
30pub struct RearmLoggedObjects<VM: VMBinding> {
31    objects: Vec<ObjectReference>,
32    _p: std::marker::PhantomData<VM>,
33}
34
35#[cfg(feature = "lxr_object_log")]
36impl<VM: VMBinding> RearmLoggedObjects<VM> {
37    pub fn new(objects: Vec<ObjectReference>) -> Self {
38        Self {
39            objects,
40            _p: std::marker::PhantomData,
41        }
42    }
43}
44
45#[cfg(feature = "lxr_object_log")]
46impl<VM: VMBinding> crate::scheduler::GCWork<VM> for RearmLoggedObjects<VM> {
47    fn do_work(&mut self, _worker: &mut crate::scheduler::GCWorker<VM>, _mmtk: &'static MMTK<VM>) {
48        for obj in &self.objects {
49            VM::VMObjectModel::GLOBAL_LOG_BIT_SPEC.mark_as_unlogged::<VM>(*obj, Ordering::SeqCst);
50        }
51    }
52}
53
54pub struct LXRFieldBarrierSemantics<VM: VMBinding> {
55    mmtk: &'static MMTK<VM>,
56    tls: VMMutatorThread,
57    incs: VectorQueue<VM::VMSlot>,
58    decs: VectorQueue<ObjectReference>,
59    refs: VectorQueue<ObjectReference>,
60    lxr: &'static LXR<VM>,
61    /// Objects logged by [`Self::object_probable_write_slow`], to be re-armed at the end of
62    /// the epoch. See there.
63    #[cfg(feature = "lxr_object_log")]
64    logged_objs: VectorQueue<ObjectReference>,
65}
66
67impl<VM: VMBinding> LXRFieldBarrierSemantics<VM> {
68    const UNLOG_BITS: SideMetadataSpec = *VM::VMObjectModel::GLOBAL_FIELD_UNLOG_BIT_SPEC
69        .as_spec()
70        .extract_side_spec();
71
72    #[allow(unused)]
73    pub fn new(mmtk: &'static MMTK<VM>, tls: VMMutatorThread) -> Self {
74        Self {
75            mmtk,
76            tls,
77            incs: VectorQueue::default(),
78            decs: VectorQueue::default(),
79            refs: VectorQueue::default(),
80            lxr: mmtk.get_plan().downcast_ref::<LXR<VM>>().unwrap(),
81            #[cfg(feature = "lxr_object_log")]
82            logged_objs: VectorQueue::default(),
83        }
84    }
85
86    #[cfg(feature = "lxr_object_log")]
87    #[cold]
88    fn flush_logged_objects(&mut self) {
89        let objects = self.logged_objs.take();
90        if objects.is_empty() {
91            return;
92        }
93        self.mmtk.scheduler.work_buckets[WorkBucketStage::FIRST_STW_STAGE]
94            .add(RearmLoggedObjects::<VM>::new(objects));
95    }
96
97    fn get_slot_logging_state(&self, slot: VM::VMSlot) -> u8 {
98        unsafe { Self::UNLOG_BITS.load(slot.to_address()) }
99    }
100
101    fn attempt_to_log_field(&self, slot: VM::VMSlot) -> bool {
102        loop {
103            // Bailout if logged
104            if self.get_slot_logging_state(slot) == LOGGED_VALUE {
105                return false;
106            }
107            // Attempt to log the slots
108            match Self::UNLOG_BITS.compare_exchange_atomic(
109                slot.to_address(),
110                UNLOGGED_VALUE,
111                LOGGED_VALUE,
112                Ordering::SeqCst,
113                Ordering::SeqCst,
114            ) {
115                Ok(_) => return true,
116                Err(current) => {
117                    if current == LOGGED_VALUE {
118                        return false;
119                    }
120                }
121            }
122            // Failed to log the slot. Spin.
123            std::hint::spin_loop();
124        }
125    }
126
127    fn log_slot_and_get_old_target(&self, slot: VM::VMSlot) -> Result<Option<ObjectReference>, ()> {
128        if self.get_slot_logging_state(slot) == LOGGED_VALUE {
129            return Err(());
130        }
131        let old = slot.load();
132        if self.attempt_to_log_field(slot) {
133            Ok(old)
134        } else {
135            Err(())
136        }
137    }
138
139    fn slow(
140        &mut self,
141        _src: Option<ObjectReference>,
142        slot: VM::VMSlot,
143        old: Option<ObjectReference>,
144    ) {
145        // Reference counting
146        if let Some(old) = old {
147            self.decs.push(old);
148            if self.decs.is_full() {
149                self.flush_decs_and_satb();
150            }
151        }
152        self.incs.push(slot);
153        if self.incs.is_full() {
154            self.flush_incs();
155        }
156    }
157
158    fn enqueue_node(
159        &mut self,
160        src: Option<ObjectReference>,
161        slot: VM::VMSlot,
162        _new: Option<ObjectReference>,
163    ) -> bool {
164        if let Ok(old) = self.log_slot_and_get_old_target(slot) {
165            self.slow(src, slot, old);
166            true
167        } else {
168            false
169        }
170    }
171
172    fn should_create_satb_packets(&self) -> bool {
173        self.lxr.cm_enabled()
174            && (self.lxr.concurrent_work_in_progress()
175                || self.lxr.current_pause() == Some(Pause::FinalMark))
176    }
177
178    #[cold]
179    fn flush_incs(&mut self) {
180        if !self.incs.is_empty() {
181            let incs = self.incs.take();
182            self.lxr.rc.increase_inc_buffer_size(incs.len());
183            self.mmtk.scheduler.work_buckets[WorkBucketStage::RCProcessIncs].add(ProcessIncs::<
184                _,
185                EDGE_KIND_MATURE,
186            >::new(
187                incs, self.lxr
188            ));
189        }
190    }
191
192    #[cold]
193    fn flush_decs_and_satb(&mut self) {
194        if !self.decs.is_empty() {
195            let w = if self.should_create_satb_packets() {
196                let decs = Arc::new(self.decs.take());
197                self.mmtk.scheduler.work_buckets[WorkBucketStage::FinishConcurrentWork]
198                    .add(ProcessModBufSATB::new_arc(decs.clone()));
199                ProcessDecs::new_arc(decs, LazySweepingJobsCounter::new_decs())
200            } else {
201                let decs = self.decs.take();
202                ProcessDecs::new(decs, LazySweepingJobsCounter::new_decs())
203            };
204            if super::LAZY_DECREMENTS {
205                self.mmtk.scheduler.work_buckets[WorkBucketStage::Concurrent]
206                    .add_deferred(Box::new(w));
207            } else {
208                self.mmtk.scheduler.work_buckets[WorkBucketStage::STWRCDecsAndSweep].add(w);
209            }
210        }
211    }
212
213    #[cold]
214    fn flush_weak_refs(&mut self) {
215        if !self.refs.is_empty() {
216            debug_assert!(self.should_create_satb_packets());
217            let nodes = self.refs.take();
218            self.mmtk.scheduler.work_buckets[WorkBucketStage::FinishConcurrentWork]
219                .add(ProcessModBufSATB::new(nodes));
220        }
221    }
222}
223
224impl<VM: VMBinding> BarrierSemantics for LXRFieldBarrierSemantics<VM> {
225    type VM = VM;
226
227    #[cold]
228    fn flush(&mut self) {
229        self.flush_weak_refs();
230        self.flush_incs();
231        self.flush_decs_and_satb();
232        // Ends the coalescing epoch for the objects this mutator logged: each is armed
233        // again, so the next store to it is recorded.
234        #[cfg(feature = "lxr_object_log")]
235        self.flush_logged_objects();
236    }
237
238    fn object_reference_write_slow(
239        &mut self,
240        src: ObjectReference,
241        slot: VM::VMSlot,
242        target: Option<ObjectReference>,
243    ) {
244        self.enqueue_node(Some(src), slot, target);
245    }
246
247    fn memory_region_copy_slow(&mut self, _src: VM::VMMemorySlice, dst: VM::VMMemorySlice) {
248        // Quickly check if all fields are logged. If yes, skip the barrier.
249        let unlog_bits_start = address_to_meta_address(&Self::UNLOG_BITS, dst.start());
250        let unlog_bits_start_aligned = unlog_bits_start.align_down(16);
251        let unlog_bits_end =
252            address_to_meta_address(&Self::UNLOG_BITS, dst.start() + dst.bytes() - 1);
253        let unlog_bits_end_aligned = unlog_bits_end.align_down(16);
254        let mut cursor = unlog_bits_start_aligned;
255        let mut all_logged = true;
256        while cursor <= unlog_bits_end_aligned {
257            if unsafe { cursor.load::<u128>() } != 0 {
258                all_logged = false;
259                break;
260            }
261            cursor += 16usize;
262        }
263        if all_logged {
264            return;
265        }
266
267        for s in dst.iter_slots() {
268            let _succ = self.enqueue_node(None, s, None);
269        }
270    }
271
272    fn load_weak_reference(&mut self, o: ObjectReference) {
273        if !self.lxr.concurrent_work_in_progress() || self.lxr.is_marked(o) {
274            return;
275        }
276        self.refs.push(o);
277        if self.refs.is_full() {
278            self.flush_weak_refs();
279        }
280    }
281
282    fn object_probable_write_slow(&mut self, obj: ObjectReference) {
283        obj.iterate_fields::<VM, _>(self.tls.0, |s| {
284            let _succ = self.enqueue_node(Some(obj), s, None);
285        });
286        // Every field of `obj` is now logged. Also log the object log bit,
287        // so next time we don't hvae to scan the object again. This is a performance optimization.
288        #[cfg(feature = "lxr_object_log")]
289        {
290            VM::VMObjectModel::GLOBAL_LOG_BIT_SPEC.store_atomic::<VM, u8>(
291                obj,
292                LOGGED_VALUE,
293                None,
294                Ordering::SeqCst,
295            );
296            self.logged_objs.push(obj);
297            if self.logged_objs.is_full() {
298                self.flush_logged_objects();
299            }
300        }
301    }
302}
mmtk/plan/lxr/barrier.rs

mmtk/plan/lxr/
barrier.rs
