1+ use std:: sync:: atomic:: AtomicUsize ;
2+ use std:: sync:: atomic:: Ordering ;
13use std:: sync:: Mutex ;
24
35use mmtk:: {
@@ -9,10 +11,13 @@ use mmtk::{
911use crate :: { abi:: GCThreadTLS , upcalls, Ruby } ;
1012
1113pub struct WeakProcessor {
14+ non_parallel_obj_free_candidates : Mutex < Vec < ObjectReference > > ,
15+ parallel_obj_free_candidates : Vec < Mutex < Vec < ObjectReference > > > ,
16+ parallel_obj_free_candidates_counter : AtomicUsize ,
17+
1218 /// Objects that needs `obj_free` called when dying.
1319/// If it is a bottleneck, replace it with a lock-free data structure,
1420/// or add candidates in batch.
15- obj_free_candidates : Mutex < Vec < ObjectReference > > ,
1621weak_references : Mutex < Vec < ObjectReference > > ,
1722}
1823
@@ -25,32 +30,59 @@ impl Default for WeakProcessor {
2530impl WeakProcessor {
2631 pub fn new ( ) -> Self {
2732 Self {
28- obj_free_candidates : Mutex :: new ( Vec :: new ( ) ) ,
33+ non_parallel_obj_free_candidates : Mutex :: new ( Vec :: new ( ) ) ,
34+ parallel_obj_free_candidates : vec ! [ Mutex :: new( Vec :: new( ) ) ] ,
35+ parallel_obj_free_candidates_counter : AtomicUsize :: new ( 0 ) ,
2936 weak_references : Mutex :: new ( Vec :: new ( ) ) ,
3037 }
3138 }
3239
33- /// Add an object as a candidate for `obj_free`.
34- ///
35- /// Multiple mutators can call it concurrently, so it has `&self`.
36- pub fn add_obj_free_candidate ( & self , object : ObjectReference ) {
37- let mut obj_free_candidates = self . obj_free_candidates . lock ( ) . unwrap ( ) ;
38- obj_free_candidates. push ( object) ;
40+ pub fn init_parallel_obj_free_candidates ( & mut self , num_workers : usize ) {
41+ debug_assert_eq ! ( self . parallel_obj_free_candidates. len( ) , 1 ) ;
42+
43+ for _ in 1 ..num_workers {
44+ self . parallel_obj_free_candidates
45+ . push ( Mutex :: new ( Vec :: new ( ) ) ) ;
46+ }
3947 }
4048
41- /// Add many objects as candidates for `obj_free`.
49+ /// Add an object as a candidate for `obj_free`.
4250///
4351/// Multiple mutators can call it concurrently, so it has `&self`.
44- pub fn add_obj_free_candidates ( & self , objects : & [ ObjectReference ] ) {
45- let mut obj_free_candidates = self . obj_free_candidates . lock ( ) . unwrap ( ) ;
46- for object in objects. iter ( ) . copied ( ) {
47- obj_free_candidates. push ( object) ;
52+ pub fn add_obj_free_candidate ( & self , object : ObjectReference , can_parallel_free : bool ) {
53+ if can_parallel_free {
54+ // Newly allocated objects are placed in parallel_obj_free_candidates using
55+ // round-robin. This may not be ideal for load balancing.
56+ let idx = self
57+ . parallel_obj_free_candidates_counter
58+ . fetch_add ( 1 , Ordering :: Relaxed )
59+ % self . parallel_obj_free_candidates . len ( ) ;
60+
61+ self . parallel_obj_free_candidates [ idx]
62+ . lock ( )
63+ . unwrap ( )
64+ . push ( object) ;
65+ } else {
66+ self . non_parallel_obj_free_candidates
67+ . lock ( )
68+ . unwrap ( )
69+ . push ( object) ;
4870 }
4971 }
5072
5173 pub fn get_all_obj_free_candidates ( & self ) -> Vec < ObjectReference > {
52- let mut obj_free_candidates = self . obj_free_candidates . lock ( ) . unwrap ( ) ;
53- std:: mem:: take ( obj_free_candidates. as_mut ( ) )
74+ // let mut obj_free_candidates = self.obj_free_candidates.lock().unwrap();
75+ let mut all_obj_free_candidates = self
76+ . non_parallel_obj_free_candidates
77+ . lock ( )
78+ . unwrap ( )
79+ . to_vec ( ) ;
80+
81+ for candidates_mutex in & self . parallel_obj_free_candidates {
82+ all_obj_free_candidates. extend ( candidates_mutex. lock ( ) . unwrap ( ) . to_vec ( ) ) ;
83+ }
84+
85+ std:: mem:: take ( all_obj_free_candidates. as_mut ( ) )
5486 }
5587
5688 pub fn add_weak_reference ( & self , object : ObjectReference ) {
@@ -63,7 +95,18 @@ impl WeakProcessor {
6395 worker : & mut GCWorker < Ruby > ,
6496 _tracer_context : impl ObjectTracerContext < Ruby > ,
6597 ) {
66- worker. add_work ( WorkBucketStage :: VMRefClosure , ProcessObjFreeCandidates ) ;
98+ worker. add_work (
99+ WorkBucketStage :: VMRefClosure ,
100+ ProcessNonParallelObjFreeCanadidates { } ,
101+ ) ;
102+
103+ for index in 0 ..self . parallel_obj_free_candidates . len ( ) {
104+ worker. add_work (
105+ WorkBucketStage :: VMRefClosure ,
106+ ProcessParallelObjFreeCandidates { index } ,
107+ ) ;
108+ }
109+
67110 worker. add_work ( WorkBucketStage :: VMRefClosure , ProcessWeakReferences ) ;
68111
69112 worker. add_work ( WorkBucketStage :: Prepare , UpdateFinalizerObjIdTables ) ;
@@ -80,36 +123,50 @@ impl WeakProcessor {
80123 }
81124}
82125
83- struct ProcessObjFreeCandidates ;
126+ fn process_obj_free_candidates ( obj_free_candidates : & mut Vec < ObjectReference > ) {
127+ // Process obj_free
128+ let mut new_candidates = Vec :: new ( ) ;
129+
130+ for object in obj_free_candidates. iter ( ) . copied ( ) {
131+ if object. is_reachable ( ) {
132+ // Forward and add back to the candidate list.
133+ let new_object = object. forward ( ) ;
134+ trace ! ( "Forwarding obj_free candidate: {object} -> {new_object}" ) ;
135+ new_candidates. push ( new_object) ;
136+ } else {
137+ ( upcalls ( ) . call_obj_free ) ( object) ;
138+ }
139+ }
140+
141+ * obj_free_candidates = new_candidates;
142+ }
143+
144+ struct ProcessParallelObjFreeCandidates {
145+ index : usize ,
146+ }
84147
85- impl GCWork < Ruby > for ProcessObjFreeCandidates {
148+ impl GCWork < Ruby > for ProcessParallelObjFreeCandidates {
86149 fn do_work ( & mut self , _worker : & mut GCWorker < Ruby > , _mmtk : & ' static mmtk:: MMTK < Ruby > ) {
87- // If it blocks, it is a bug.
88- let mut obj_free_candidates = crate :: binding ( )
89- . weak_proc
90- . obj_free_candidates
150+ let mut obj_free_candidates = crate :: binding ( ) . weak_proc . parallel_obj_free_candidates
151+ [ self . index ]
91152 . try_lock ( )
92- . expect ( "It's GC time. No mutators should hold this lock at this time." ) ;
93-
94- let n_cands = obj_free_candidates. len ( ) ;
153+ . expect ( "Lock for parallel_obj_free_candidates should not be held" ) ;
95154
96- debug ! ( "Total: {n_cands} candidates" ) ;
155+ process_obj_free_candidates ( & mut obj_free_candidates) ;
156+ }
157+ }
97158
98- // Process obj_free
99- let mut new_candidates = Vec :: new ( ) ;
159+ struct ProcessNonParallelObjFreeCanadidates ;
100160
101- for object in obj_free_candidates. iter ( ) . copied ( ) {
102- if object. is_reachable ( ) {
103- // Forward and add back to the candidate list.
104- let new_object = object. forward ( ) ;
105- trace ! ( "Forwarding obj_free candidate: {object} -> {new_object}" ) ;
106- new_candidates. push ( new_object) ;
107- } else {
108- ( upcalls ( ) . call_obj_free ) ( object) ;
109- }
110- }
161+ impl GCWork < Ruby > for ProcessNonParallelObjFreeCanadidates {
162+ fn do_work ( & mut self , _worker : & mut GCWorker < Ruby > , _mmtk : & ' static mmtk:: MMTK < Ruby > ) {
163+ let mut obj_free_candidates = crate :: binding ( )
164+ . weak_proc
165+ . non_parallel_obj_free_candidates
166+ . try_lock ( )
167+ . expect ( "Lock for non_parallel_obj_free_candidates should not be held" ) ;
111168
112- * obj_free_candidates = new_candidates ;
169+ process_obj_free_candidates ( & mut obj_free_candidates ) ;
113170 }
114171}
115172
0 commit comments