1+ use std:: sync:: atomic:: AtomicUsize ;
2+ use std:: sync:: atomic:: Ordering ;
13use std:: sync:: Mutex ;
24
35use mmtk:: scheduler:: GCWork ;
@@ -11,10 +13,13 @@ use crate::upcalls;
1113use crate :: Ruby ;
1214
1315pub struct WeakProcessor {
16+ non_parallel_obj_free_candidates : Mutex < Vec < ObjectReference > > ,
17+ parallel_obj_free_candidates : Vec < Mutex < Vec < ObjectReference > > > ,
18+ parallel_obj_free_candidates_counter : AtomicUsize ,
19+
1420 /// Objects that needs `obj_free` called when dying.
1521/// If it is a bottleneck, replace it with a lock-free data structure,
1622/// or add candidates in batch.
17- obj_free_candidates : Mutex < Vec < ObjectReference > > ,
1823weak_references : Mutex < Vec < ObjectReference > > ,
1924}
2025
@@ -27,32 +32,59 @@ impl Default for WeakProcessor {
2732impl WeakProcessor {
2833 pub fn new ( ) -> Self {
2934 Self {
30- obj_free_candidates : Mutex :: new ( Vec :: new ( ) ) ,
35+ non_parallel_obj_free_candidates : Mutex :: new ( Vec :: new ( ) ) ,
36+ parallel_obj_free_candidates : vec ! [ Mutex :: new( Vec :: new( ) ) ] ,
37+ parallel_obj_free_candidates_counter : AtomicUsize :: new ( 0 ) ,
3138 weak_references : Mutex :: new ( Vec :: new ( ) ) ,
3239 }
3340 }
3441
35- /// Add an object as a candidate for `obj_free`.
36- ///
37- /// Multiple mutators can call it concurrently, so it has `&self`.
38- pub fn add_obj_free_candidate ( & self , object : ObjectReference ) {
39- let mut obj_free_candidates = self . obj_free_candidates . lock ( ) . unwrap ( ) ;
40- obj_free_candidates. push ( object) ;
42+ pub fn init_parallel_obj_free_candidates ( & mut self , num_workers : usize ) {
43+ debug_assert_eq ! ( self . parallel_obj_free_candidates. len( ) , 1 ) ;
44+
45+ for _ in 1 ..num_workers {
46+ self . parallel_obj_free_candidates
47+ . push ( Mutex :: new ( Vec :: new ( ) ) ) ;
48+ }
4149 }
4250
43- /// Add many objects as candidates for `obj_free`.
51+ /// Add an object as a candidate for `obj_free`.
4452///
4553/// Multiple mutators can call it concurrently, so it has `&self`.
46- pub fn add_obj_free_candidates ( & self , objects : & [ ObjectReference ] ) {
47- let mut obj_free_candidates = self . obj_free_candidates . lock ( ) . unwrap ( ) ;
48- for object in objects. iter ( ) . copied ( ) {
49- obj_free_candidates. push ( object) ;
54+ pub fn add_obj_free_candidate ( & self , object : ObjectReference , can_parallel_free : bool ) {
55+ if can_parallel_free {
56+ // Newly allocated objects are placed in parallel_obj_free_candidates using
57+ // round-robin. This may not be ideal for load balancing.
58+ let idx = self
59+ . parallel_obj_free_candidates_counter
60+ . fetch_add ( 1 , Ordering :: Relaxed )
61+ % self . parallel_obj_free_candidates . len ( ) ;
62+
63+ self . parallel_obj_free_candidates [ idx]
64+ . lock ( )
65+ . unwrap ( )
66+ . push ( object) ;
67+ } else {
68+ self . non_parallel_obj_free_candidates
69+ . lock ( )
70+ . unwrap ( )
71+ . push ( object) ;
5072 }
5173 }
5274
5375 pub fn get_all_obj_free_candidates ( & self ) -> Vec < ObjectReference > {
54- let mut obj_free_candidates = self . obj_free_candidates . lock ( ) . unwrap ( ) ;
55- std:: mem:: take ( obj_free_candidates. as_mut ( ) )
76+ // let mut obj_free_candidates = self.obj_free_candidates.lock().unwrap();
77+ let mut all_obj_free_candidates = self
78+ . non_parallel_obj_free_candidates
79+ . lock ( )
80+ . unwrap ( )
81+ . to_vec ( ) ;
82+
83+ for candidates_mutex in & self . parallel_obj_free_candidates {
84+ all_obj_free_candidates. extend ( candidates_mutex. lock ( ) . unwrap ( ) . to_vec ( ) ) ;
85+ }
86+
87+ std:: mem:: take ( all_obj_free_candidates. as_mut ( ) )
5688 }
5789
5890 pub fn add_weak_reference ( & self , object : ObjectReference ) {
@@ -65,7 +97,18 @@ impl WeakProcessor {
6597 worker : & mut GCWorker < Ruby > ,
6698 _tracer_context : impl ObjectTracerContext < Ruby > ,
6799 ) {
68- worker. add_work ( WorkBucketStage :: VMRefClosure , ProcessObjFreeCandidates ) ;
100+ worker. add_work (
101+ WorkBucketStage :: VMRefClosure ,
102+ ProcessNonParallelObjFreeCanadidates { } ,
103+ ) ;
104+
105+ for index in 0 ..self . parallel_obj_free_candidates . len ( ) {
106+ worker. add_work (
107+ WorkBucketStage :: VMRefClosure ,
108+ ProcessParallelObjFreeCandidates { index } ,
109+ ) ;
110+ }
111+
69112 worker. add_work ( WorkBucketStage :: VMRefClosure , ProcessWeakReferences ) ;
70113
71114 worker. add_work ( WorkBucketStage :: Prepare , UpdateFinalizerObjIdTables ) ;
@@ -82,36 +125,50 @@ impl WeakProcessor {
82125 }
83126}
84127
85- struct ProcessObjFreeCandidates ;
128+ fn process_obj_free_candidates ( obj_free_candidates : & mut Vec < ObjectReference > ) {
129+ // Process obj_free
130+ let mut new_candidates = Vec :: new ( ) ;
131+
132+ for object in obj_free_candidates. iter ( ) . copied ( ) {
133+ if object. is_reachable ( ) {
134+ // Forward and add back to the candidate list.
135+ let new_object = object. forward ( ) ;
136+ trace ! ( "Forwarding obj_free candidate: {object} -> {new_object}" ) ;
137+ new_candidates. push ( new_object) ;
138+ } else {
139+ ( upcalls ( ) . call_obj_free ) ( object) ;
140+ }
141+ }
142+
143+ * obj_free_candidates = new_candidates;
144+ }
145+
146+ struct ProcessParallelObjFreeCandidates {
147+ index : usize ,
148+ }
86149
87- impl GCWork < Ruby > for ProcessObjFreeCandidates {
150+ impl GCWork < Ruby > for ProcessParallelObjFreeCandidates {
88151 fn do_work ( & mut self , _worker : & mut GCWorker < Ruby > , _mmtk : & ' static mmtk:: MMTK < Ruby > ) {
89- // If it blocks, it is a bug.
90- let mut obj_free_candidates = crate :: binding ( )
91- . weak_proc
92- . obj_free_candidates
152+ let mut obj_free_candidates = crate :: binding ( ) . weak_proc . parallel_obj_free_candidates
153+ [ self . index ]
93154 . try_lock ( )
94- . expect ( "It's GC time. No mutators should hold this lock at this time." ) ;
95-
96- let n_cands = obj_free_candidates. len ( ) ;
155+ . expect ( "Lock for parallel_obj_free_candidates should not be held" ) ;
97156
98- debug ! ( "Total: {n_cands} candidates" ) ;
157+ process_obj_free_candidates ( & mut obj_free_candidates) ;
158+ }
159+ }
99160
100- // Process obj_free
101- let mut new_candidates = Vec :: new ( ) ;
161+ struct ProcessNonParallelObjFreeCanadidates ;
102162
103- for object in obj_free_candidates. iter ( ) . copied ( ) {
104- if object. is_reachable ( ) {
105- // Forward and add back to the candidate list.
106- let new_object = object. forward ( ) ;
107- trace ! ( "Forwarding obj_free candidate: {object} -> {new_object}" ) ;
108- new_candidates. push ( new_object) ;
109- } else {
110- ( upcalls ( ) . call_obj_free ) ( object) ;
111- }
112- }
163+ impl GCWork < Ruby > for ProcessNonParallelObjFreeCanadidates {
164+ fn do_work ( & mut self , _worker : & mut GCWorker < Ruby > , _mmtk : & ' static mmtk:: MMTK < Ruby > ) {
165+ let mut obj_free_candidates = crate :: binding ( )
166+ . weak_proc
167+ . non_parallel_obj_free_candidates
168+ . try_lock ( )
169+ . expect ( "Lock for non_parallel_obj_free_candidates should not be held" ) ;
113170
114- * obj_free_candidates = new_candidates ;
171+ process_obj_free_candidates ( & mut obj_free_candidates ) ;
115172 }
116173}
117174
0 commit comments