Parallel concurrent set processing

mmtk/Cargo.toml

@@ -12,7 +12,7 @@ edition = "2021"
 # Metadata for the Ruby repository
 [package.metadata.ci-repos.ruby]
 repo = "mmtk/ruby" # This is used by actions/checkout, so the format is "owner/repo", not URL.
-rev = "a11526830e747dc68f250a38bbebf01c02bf6462"
+rev = "f3f893c6fb8cbe08c3bc4c68d4cd0deceae42461"
 
 [lib]
 name = "mmtk_ruby"

mmtk/cbindgen.toml

@@ -47,3 +47,4 @@ include = ["HiddenHeader"]
 "MIN_OBJ_ALIGN" = "MMTK_MIN_OBJ_ALIGN"
 "HiddenHeader" = "MMTk_HiddenHeader"
 "HIDDEN_SIZE_MASK" = "MMTK_HIDDEN_SIZE_MASK"
+"ConcurrentSetStats" = "MMTk_ConcurrentSetStats"

mmtk/src/abi.rs

@@ -305,6 +305,14 @@ pub struct RubyBindingOptions {
     pub suffix_size: usize,
 }
 
+#[repr(C)]
+#[derive(Clone, Default)]
+pub struct ConcurrentSetStats {
+    pub live: usize,
+    pub moved: usize,
+    pub deleted: usize,
+}
+
 #[repr(C)]
 #[derive(Clone)]
 pub struct RubyUpcalls {
@@ -353,8 +361,9 @@ pub struct RubyUpcalls {
     pub get_cc_refinement_table_size: extern "C" fn() -> usize,
     pub update_cc_refinement_table: extern "C" fn(),
     // Get tables for specialized processing
+    pub get_fstring_table_obj: extern "C" fn() -> ObjectReference,
     pub get_global_symbols_table: extern "C" fn() -> *mut st_table,
-    // Detailed table info queries and operations
+    // Detailed st_table info queries and operations
     pub st_get_num_entries: extern "C" fn(table: *const st_table) -> usize,
     pub st_get_size_info: extern "C" fn(
         table: *const st_table,
@@ -372,6 +381,15 @@ pub struct RubyUpcalls {
     ) -> usize,
     pub st_update_bins_range:
         extern "C" fn(table: *mut st_table, begin: libc::size_t, end: libc::size_t) -> usize,
+    // Detailed concurrent_set info queries and operations
+    pub concurrent_set_get_num_entries: extern "C" fn(set: ObjectReference) -> usize,
+    pub concurrent_set_get_capacity: extern "C" fn(set: ObjectReference) -> usize,
+    pub concurrent_set_update_entries_range: extern "C" fn(
+        set: ObjectReference,
+        begin: usize,
+        end: usize,
+        stats: *mut ConcurrentSetStats,
+    ),
 }
 
 unsafe impl Sync for RubyUpcalls {}

mmtk/src/binding.rs

@@ -62,6 +62,7 @@ pub struct RubyBinding {
     pub wb_unprotected_objects: Mutex<HashSet<ObjectReference>>,
     pub st_entries_chunk_size: usize,
     pub st_bins_chunk_size: usize,
+    pub concurrent_set_chunk_size: usize,
 }
 
 unsafe impl Sync for RubyBinding {}
@@ -89,9 +90,12 @@ impl RubyBinding {
 
         let st_entries_chunk_size = env_default::<usize>("RUBY_MMTK_ENTRIES_CHUNK_SIZE", 1024);
         let st_bins_chunk_size = env_default::<usize>("RUBY_MMTK_BINS_CHUNK_SIZE", 4096);
+        let concurrent_set_chunk_size =
+            env_default::<usize>("RUBY_MMTK_CONCURRENT_SET_CHUNK_SIZE", 1024);
 
         debug!("st_entries_chunk_size: {st_entries_chunk_size}");
         debug!("st_bins_chunk_size: {st_bins_chunk_size}");
+        debug!("concurrent_set_chunk_size: {concurrent_set_chunk_size}");
 
         Self {
             mmtk,
@@ -105,6 +109,7 @@ impl RubyBinding {
             wb_unprotected_objects: Default::default(),
             st_entries_chunk_size,
             st_bins_chunk_size,
+            concurrent_set_chunk_size,
         }
     }
 

mmtk/src/weak_proc.rs

@@ -1,17 +1,18 @@
-use std::sync::{Arc, Mutex};
+use std::sync::Mutex;
 
 use mmtk::{
     scheduler::{GCWork, GCWorker, WorkBucketStage},
     util::ObjectReference,
     vm::ObjectTracerContext,
 };
 
-use crate::{
-    abi::{st_table, GCThreadTLS},
-    extra_assert, is_mmtk_object_safe, upcalls,
-    utils::AfterAll,
-    Ruby,
-};
+use crate::{abi::GCThreadTLS, extra_assert, is_mmtk_object_safe, upcalls, Ruby};
+
+pub mod concurrent_set_parallel;
+pub mod st_table_parallel;
+
+/// Set this to true to use chunked processing optimization for the fstring table.
+const SPECIALIZE_FSTRING_TABLE_PROCESSING: bool = true;
 
 /// Set this to true to use chunked processing optimization for the global symbols table.
 const SPECIALIZE_GLOBAL_SYMBOLS_TABLE_PROCESSING: bool = true;
@@ -74,16 +75,26 @@ impl WeakProcessor {
             // global symbols table specialized
             Box::new(UpdateFinalizerAndObjIdTables) as _,
             Box::new(UpdateGenericFieldsTbl) as _,
-            Box::new(UpdateFrozenStringsTable) as _,
             Box::new(UpdateCCRefinementTable) as _,
             // END: Weak tables
             Box::new(UpdateWbUnprotectedObjectsList) as _,
         ]);
 
         let forward = crate::mmtk().get_plan().current_gc_may_move_object();
 
+        if SPECIALIZE_FSTRING_TABLE_PROCESSING {
+            concurrent_set_parallel::process_weak_concurrent_set_chunked(
+                "fstring",
+                (upcalls().get_fstring_table_obj)(),
+                worker,
+            );
+        } else {
+            worker.scheduler().work_buckets[WorkBucketStage::VMRefClosure]
+                .add_boxed(Box::new(UpdateFrozenStringsTable) as _);
+        }
+
         if SPECIALIZE_GLOBAL_SYMBOLS_TABLE_PROCESSING {
-            Self::process_weak_table_chunked(
+            st_table_parallel::process_weak_table_chunked(
                 "global symbols",
                 (upcalls().get_global_symbols_table)(),
                 false,
@@ -96,78 +107,6 @@ impl WeakProcessor {
                 .add_boxed(Box::new(UpdateGlobalSymbolsTable) as _);
         }
     }
-
-    pub fn process_weak_table_chunked(
-        name: &'static str,
-        table: *mut st_table,
-        weak_keys: bool,
-        weak_values: bool,
-        forward: bool,
-        worker: &mut GCWorker<Ruby>,
-    ) {
-        let mut entries_start = 0;
-        let mut entries_bound = 0;
-        let mut bins_num = 0;
-        (upcalls().st_get_size_info)(table, &mut entries_start, &mut entries_bound, &mut bins_num);
-        let num_entries = (upcalls().st_get_num_entries)(table);
-        debug!(
-            "name: {name}, entries_start: {entries_start}, entries_bound: {entries_bound}, bins_num: {bins_num}, num_entries: {num_entries}"
-        );
-
-        let table_name_ptr = name.as_ptr();
-        let table_name_len = name.len();
-
-        probe!(
-            mmtk_ruby,
-            initial_weak_table_stats,
-            entries_start,
-            entries_bound,
-            bins_num,
-            num_entries,
-            table_name_ptr,
-            table_name_len,
-        );
-
-        let entries_chunk_size = crate::binding().st_entries_chunk_size;
-        let bins_chunk_size = crate::binding().st_bins_chunk_size;
-
-        let after_all = Arc::new(AfterAll::new(WorkBucketStage::VMRefClosure));
-
-        let entries_packets = (entries_start..entries_bound)
-            .step_by(entries_chunk_size)
-            .map(|begin| {
-                let end = (begin + entries_chunk_size).min(entries_bound);
-                let after_all = after_all.clone();
-                Box::new(UpdateTableEntriesParallel {
-                    name,
-                    table,
-                    begin,
-                    end,
-                    weak_keys,
-                    weak_values,
-                    forward,
-                    after_all,
-                }) as _
-            })
-            .collect::<Vec<_>>();
-        after_all.count_up(entries_packets.len());
-
-        let bins_packets = (0..bins_num)
-            .step_by(entries_chunk_size)
-            .map(|begin| {
-                let end = (begin + bins_chunk_size).min(bins_num);
-                Box::new(UpdateTableBinsParallel {
-                    name: name.to_string(),
-                    table,
-                    begin,
-                    end,
-                }) as _
-            })
-            .collect::<Vec<_>>();
-        after_all.add_packets(bins_packets);
-
-        worker.scheduler().work_buckets[WorkBucketStage::VMRefClosure].bulk_add(entries_packets);
-    }
 }
 
 struct ProcessObjFreeCandidates;
@@ -321,89 +260,6 @@ define_global_table_processor!(UpdateCCRefinementTable, {
 
 ///////// END: Simple table updating work packets ////////
 
-struct UpdateTableEntriesParallel {
-    name: &'static str,
-    table: *mut st_table,
-    begin: usize,
-    end: usize,
-    weak_keys: bool,
-    weak_values: bool,
-    forward: bool,
-    after_all: Arc<AfterAll>,
-}
-
-unsafe impl Send for UpdateTableEntriesParallel {}
-
-impl UpdateTableEntriesParallel {}
-
-impl GCWork<Ruby> for UpdateTableEntriesParallel {
-    fn do_work(&mut self, worker: &mut GCWorker<Ruby>, _mmtk: &'static mmtk::MMTK<Ruby>) {
-        debug!("Updating entries of {} table", self.name);
-        let deleted_entries = (upcalls().st_update_entries_range)(
-            self.table,
-            self.begin,
-            self.end,
-            self.weak_keys,
-            self.weak_values,
-            self.forward,
-        );
-        debug!("Done updating entries of {} table", self.name);
-        let table_name = self.name.as_ptr();
-        let table_name_len = self.name.len();
-        probe!(
-            mmtk_ruby,
-            update_table_entries_parallel,
-            self.begin,
-            self.end,
-            deleted_entries,
-            table_name,
-            table_name_len
-        );
-
-        let is_last = self.after_all.count_down(worker);
-        if is_last {
-            let num_entries = (upcalls().st_get_num_entries)(self.table);
-            probe!(
-                mmtk_ruby,
-                final_weak_table_stats,
-                num_entries,
-                table_name,
-                table_name_len
-            )
-        }
-    }
-}
-
-struct UpdateTableBinsParallel {
-    name: String,
-    table: *mut st_table,
-    begin: usize,
-    end: usize,
-}
-
-unsafe impl Send for UpdateTableBinsParallel {}
-
-impl UpdateTableBinsParallel {}
-
-impl GCWork<Ruby> for UpdateTableBinsParallel {
-    fn do_work(&mut self, _worker: &mut GCWorker<Ruby>, _mmtk: &'static mmtk::MMTK<Ruby>) {
-        debug!("Updating bins of {} table", self.name);
-        let deleted_bins = (upcalls().st_update_bins_range)(self.table, self.begin, self.end);
-        debug!("Done updating bins of {} table", self.name);
-        let table_name = self.name.as_ptr();
-        let table_name_len = self.name.len();
-        probe!(
-            mmtk_ruby,
-            update_table_bins_parallel,
-            self.begin,
-            self.end,
-            deleted_bins,
-            table_name,
-            table_name_len
-        );
-    }
-}
-
 struct UpdateWbUnprotectedObjectsList;
 
 impl GCWork<Ruby> for UpdateWbUnprotectedObjectsList {