mmtk/plan/concurrent/
concurrent_marking_work.rs

1use crate::plan::concurrent::global::ConcurrentPlan;
2use crate::plan::concurrent::Pause;
3use crate::plan::tracing::{PlanTrace, Trace};
4use crate::plan::PlanTraceObject;
5use crate::policy::gc_work::TraceKind;
6use crate::scheduler::{gc_work::RootKind, GCWork, GCWorker, WorkBucketStage};
7use crate::util::{scanning_helper, ObjectReference};
8use crate::vm::slot::Slot;
9use crate::vm::{RootsKind, RootsWorkFactory, VMBinding};
10use crate::MMTK;
11
12use std::collections::VecDeque;
13use std::marker::PhantomData;
14
15pub struct ConcurrentTraceObjects<
16    VM: VMBinding,
17    P: ConcurrentPlan<VM = VM> + PlanTraceObject<VM>,
18    const KIND: TraceKind,
19> {
20    /// initial objects to mark and scan
21    initial_objects: Vec<ObjectReference>,
22    /// `true` if the `initial_objects` are already marked.
23    already_marked: bool,
24    phantom_data: PhantomData<(VM, P)>,
25}
26
27impl<VM: VMBinding, P: ConcurrentPlan<VM = VM> + PlanTraceObject<VM>, const KIND: TraceKind>
28    ConcurrentTraceObjects<VM, P, KIND>
29{
30    const SATB_BUFFER_SIZE: usize = 8192;
31    const CONCURRENT_TRACE_OVERFLOW: usize = Self::SATB_BUFFER_SIZE * 2;
32
33    pub fn new(initial_objects: Vec<ObjectReference>, already_marked: bool) -> Self {
34        Self {
35            initial_objects,
36            already_marked,
37            phantom_data: PhantomData,
38        }
39    }
40}
41
42unsafe impl<VM: VMBinding, P: ConcurrentPlan<VM = VM> + PlanTraceObject<VM>, const KIND: TraceKind>
43    Send for ConcurrentTraceObjects<VM, P, KIND>
44{
45}
46
47impl<VM: VMBinding, P: ConcurrentPlan<VM = VM> + PlanTraceObject<VM>, const KIND: TraceKind>
48    GCWork<VM> for ConcurrentTraceObjects<VM, P, KIND>
49{
50    fn do_work(&mut self, worker: &mut GCWorker<VM>, mmtk: &'static MMTK<VM>) {
51        let tls = worker.tls;
52        let trace = PlanTrace::<P, KIND>::from_mmtk(mmtk);
53
54        // These are initial objects.  They may not have been marked.
55        let initial_objects = std::mem::take(&mut self.initial_objects);
56        let num_initial_objects = initial_objects.len();
57        let mut num_queued_objects = 0;
58
59        // This queue contains marked but not scanned objects.
60        let mut queue = VecDeque::new();
61        if self.already_marked {
62            // The initial objects are already marked.  Put them in the queue.
63            queue.extend(initial_objects);
64        } else {
65            // We scan each object and only enqueue newly visited objects.
66            for object in initial_objects {
67                trace.trace_object(worker, object, &mut |enqueued_object| {
68                    debug_assert_eq!(enqueued_object, object);
69                    queue.push_back(enqueued_object);
70                    num_queued_objects += 1;
71                });
72            }
73        }
74
75        // Loop until the queue is drained.
76        while let Some(object) = queue.pop_back() {
77            scanning_helper::visit_children_non_moving::<VM>(tls, object, &mut |child| {
78                trace.trace_object(worker, child, &mut |enqueued_child| {
79                    debug_assert_eq!(enqueued_child, child);
80                    queue.push_back(enqueued_child);
81                    num_queued_objects += 1;
82                })
83            });
84            trace.post_scan_object(object);
85
86            if queue.len() >= Self::CONCURRENT_TRACE_OVERFLOW {
87                let offloaded_objects = queue.drain(..Self::SATB_BUFFER_SIZE).collect();
88                let w = Self::new(offloaded_objects, true);
89                // If concurrent marking was interrupted, the remaining marking work runs in the
90                // `FinalMark` pause, where the `Concurrent` bucket is disabled.  Keep the overflow
91                // in the pause, or it would only be traced after this pause's `Release`.
92                let stage = if mmtk.get_plan().concurrent().unwrap().current_pause()
93                    == Some(Pause::FinalMark)
94                {
95                    WorkBucketStage::FinishConcurrentWork
96                } else {
97                    WorkBucketStage::Concurrent
98                };
99                worker.add_work(stage, w);
100            }
101        }
102
103        probe!(
104            mmtk,
105            concurrent_trace_objects,
106            num_initial_objects,
107            num_queued_objects
108        );
109    }
110}
111
112pub struct ProcessModBufSATB<
113    VM: VMBinding,
114    P: ConcurrentPlan<VM = VM> + PlanTraceObject<VM>,
115    const KIND: TraceKind,
116> {
117    nodes: Option<Vec<ObjectReference>>,
118    _p: std::marker::PhantomData<(VM, P)>,
119}
120
121unsafe impl<VM: VMBinding, P: ConcurrentPlan<VM = VM> + PlanTraceObject<VM>, const KIND: TraceKind>
122    Send for ProcessModBufSATB<VM, P, KIND>
123{
124}
125
126impl<VM: VMBinding, P: ConcurrentPlan<VM = VM> + PlanTraceObject<VM>, const KIND: TraceKind>
127    ProcessModBufSATB<VM, P, KIND>
128{
129    pub fn new(nodes: Vec<ObjectReference>) -> Self {
130        Self {
131            nodes: Some(nodes),
132            _p: std::marker::PhantomData,
133        }
134    }
135}
136
137impl<VM: VMBinding, P: ConcurrentPlan<VM = VM> + PlanTraceObject<VM>, const KIND: TraceKind>
138    GCWork<VM> for ProcessModBufSATB<VM, P, KIND>
139{
140    fn do_work(&mut self, worker: &mut GCWorker<VM>, mmtk: &'static MMTK<VM>) {
141        let mut w = if let Some(nodes) = self.nodes.take() {
142            if nodes.is_empty() {
143                return;
144            }
145
146            ConcurrentTraceObjects::<VM, P, KIND>::new(
147                nodes, false, // These objects are not marked, yet.
148            )
149        } else {
150            return;
151        };
152        GCWork::do_work(&mut w, worker, mmtk);
153    }
154}
155
156/// A custom implementation of [`RootsWorkFactory`] for concurrent marking.
157///
158/// Slot roots are loaded immediately and represented as root nodes, just like pinning roots.  All
159/// roots are handled by the [`ConcurrentTraceObjects`] work packets.
160pub(crate) struct ConcurrentMarkingRootsWorkFactory<
161    VM: VMBinding,
162    P: ConcurrentPlan<VM = VM> + PlanTraceObject<VM>,
163    const KIND: TraceKind,
164> {
165    pub(crate) mmtk: &'static MMTK<VM>,
166    phantom_data: PhantomData<P>,
167}
168
169impl<VM: VMBinding, P: ConcurrentPlan<VM = VM> + PlanTraceObject<VM>, const KIND: TraceKind> Clone
170    for ConcurrentMarkingRootsWorkFactory<VM, P, KIND>
171{
172    fn clone(&self) -> Self {
173        Self {
174            mmtk: self.mmtk,
175            phantom_data: PhantomData,
176        }
177    }
178}
179
180unsafe impl<VM: VMBinding, P: ConcurrentPlan<VM = VM> + PlanTraceObject<VM>, const KIND: TraceKind>
181    Send for ConcurrentMarkingRootsWorkFactory<VM, P, KIND>
182{
183}
184
185impl<VM: VMBinding, P: ConcurrentPlan<VM = VM> + PlanTraceObject<VM>, const KIND: TraceKind>
186    ConcurrentMarkingRootsWorkFactory<VM, P, KIND>
187{
188    pub(crate) fn new(mmtk: &'static MMTK<VM>) -> Self {
189        Self {
190            mmtk,
191            phantom_data: PhantomData,
192        }
193    }
194
195    fn debug_assert_initial_mark(&self) {
196        let pause = self.mmtk.get_plan().concurrent().unwrap().current_pause();
197
198        debug_assert_eq!(
199            pause,
200            Some(Pause::InitialMark),
201            "Concurrent marking only scans roots during InitialMark."
202        );
203    }
204
205    fn create_and_schedule_root_nodes_work(&mut self, nodes: Vec<ObjectReference>) {
206        let mmtk = self.mmtk;
207        let work_packet = ConcurrentTraceObjects::<VM, P, KIND>::new(nodes, false);
208        mmtk.scheduler.work_buckets[WorkBucketStage::Concurrent].add_no_notify(work_packet);
209    }
210}
211
212impl<VM: VMBinding, P: ConcurrentPlan<VM = VM> + PlanTraceObject<VM>, const KIND: TraceKind>
213    RootsWorkFactory<VM::VMSlot> for ConcurrentMarkingRootsWorkFactory<VM, P, KIND>
214{
215    fn create_process_roots_work_with_root_kind(
216        &mut self,
217        slots: Vec<VM::VMSlot>,
218        _kind: RootKind,
219    ) {
220        probe!(mmtk, roots, RootsKind::NORMAL, slots.len());
221
222        self.debug_assert_initial_mark();
223
224        // We don't divide the `slots` vector into smaller chunks here.  We assume the VM binding
225        // respects the constant `EDGES_WORK_BUFFER_SIZE` and provides lists of slots in reasonable
226        // lengths.  Even if a single `ConcurrentTraceObjects` work packet is too large, it can
227        // still break up the list during tracing using the constant `CONCURRENT_TRACE_OVERFLOW`.
228        let nodes = slots
229            .iter()
230            .flat_map(|slot| slot.load())
231            .collect::<Vec<_>>();
232
233        // Note: During concurrent marking, mutators can overwrite the root slots and make the roots unstable.
234        // Therefore, instead of recording the root slots, we record the loaded root nodes.
235        #[cfg(feature = "sanity")]
236        self.mmtk
237            .sanity_checker
238            .lock()
239            .unwrap()
240            .add_root_nodes(nodes.clone());
241
242        self.create_and_schedule_root_nodes_work(nodes);
243    }
244
245    #[cfg(feature = "non_moving_root")]
246    fn create_process_pinning_roots_work(&mut self, nodes: Vec<ObjectReference>) {
247        probe!(mmtk, roots, RootsKind::PINNING, nodes.len());
248
249        self.debug_assert_initial_mark();
250
251        #[cfg(feature = "sanity")]
252        self.mmtk
253            .sanity_checker
254            .lock()
255            .unwrap()
256            .add_root_nodes(nodes.clone());
257
258        self.create_and_schedule_root_nodes_work(nodes);
259    }
260
261    #[cfg(feature = "non_moving_root")]
262    fn create_process_tpinning_roots_work(&mut self, nodes: Vec<ObjectReference>) {
263        probe!(mmtk, roots, RootsKind::TPINNING, nodes.len());
264
265        self.debug_assert_initial_mark();
266
267        #[cfg(feature = "sanity")]
268        self.mmtk
269            .sanity_checker
270            .lock()
271            .unwrap()
272            .add_root_nodes(nodes.clone());
273
274        self.create_and_schedule_root_nodes_work(nodes);
275    }
276}
mmtk/plan/concurrent/concurrent_marking_work.rs

mmtk/plan/concurrent/
concurrent_marking_work.rs
