cppgc: Parallelize weakness processing for WeakMember

This CL handles weakcallbacks for resetting WeakMember smart pointers in parallel.

Tracing data for browsing nytimes.com:
CppGC.AtomicWeak
4.628 ms
CppGC.ConcurrentWeakCallback
0.157 ms

Bug: v8:12796
Change-Id: I1e7aa9d27e22985072a8fc332376ac3bd84a3720
Reviewed-on: https://chromium-review.googlesource.com/c/v8/v8/+/3692433Reviewed-by: Michael Lippautz <mlippautz@chromium.org>
Auto-Submit: Jianxiao Lu <jianxiao.lu@intel.com>
Commit-Queue: Jianxiao Lu <jianxiao.lu@intel.com>
Cr-Commit-Position: refs/heads/main@{#81036}

src/heap/cppgc/marker.cc

@@ -325,12 +325,53 @@ void MarkerBase::FinishMarking(MarkingConfig::StackState stack_state) {
   LeaveAtomicPause();
 }
 
+class WeakCallbackJobTask final : public cppgc::JobTask {
+ public:
+  WeakCallbackJobTask(MarkerBase* marker,
+                      MarkingWorklists::WeakCallbackWorklist* callback_worklist,
+                      LivenessBroker& broker)
+      : marker_(marker),
+        callback_worklist_(callback_worklist),
+        broker_(broker) {}
+
+  void Run(JobDelegate* delegate) override {
+    StatsCollector::EnabledConcurrentScope stats_scope(
+        marker_->heap().stats_collector(),
+        StatsCollector::kConcurrentWeakCallback);
+    MarkingWorklists::WeakCallbackWorklist::Local local(callback_worklist_);
+    MarkingWorklists::WeakCallbackItem item;
+    while (local.Pop(&item)) {
+      item.callback(broker_, item.parameter);
+    }
+  }
+
+  size_t GetMaxConcurrency(size_t worker_count) const override {
+    return std::min(static_cast<size_t>(1), callback_worklist_->Size());
+  }
+
+ private:
+  MarkerBase* marker_;
+  MarkingWorklists::WeakCallbackWorklist* callback_worklist_;
+  LivenessBroker& broker_;
+};
+
 void MarkerBase::ProcessWeakness() {
   DCHECK_EQ(MarkingConfig::MarkingType::kAtomic, config_.marking_type);
 
   StatsCollector::EnabledScope stats_scope(heap().stats_collector(),
                                            StatsCollector::kAtomicWeak);
 
+  LivenessBroker broker = LivenessBrokerFactory::Create();
+  std::unique_ptr<cppgc::JobHandle> job_handle{nullptr};
+  if (heap().marking_support() ==
+      cppgc::Heap::MarkingType::kIncrementalAndConcurrent) {
+    job_handle = platform_->PostJob(
+        cppgc::TaskPriority::kUserBlocking,
+        std::make_unique<WeakCallbackJobTask>(
+            this, marking_worklists_.parallel_weak_callback_worklist(),
+            broker));
+  }
+
   heap().GetWeakPersistentRegion().Trace(&visitor());
   // Processing cross-thread handles requires taking the process lock.
   g_process_mutex.Get().AssertHeld();
@@ -338,7 +379,6 @@ void MarkerBase::ProcessWeakness() {
   heap().GetWeakCrossThreadPersistentRegion().Trace(&visitor());
 
   // Call weak callbacks on objects that may now be pointing to dead objects.
-  LivenessBroker broker = LivenessBrokerFactory::Create();
 #if defined(CPPGC_YOUNG_GENERATION)
   if (heap().generational_gc_supported()) {
     auto& remembered_set = heap().remembered_set();
@@ -369,6 +409,17 @@ void MarkerBase::ProcessWeakness() {
 #endif  // defined(CPPGC_YOUNG_GENERATION)
   }
 
+  if (job_handle) {
+    job_handle->Join();
+  } else {
+    MarkingWorklists::WeakCallbackItem item;
+    MarkingWorklists::WeakCallbackWorklist::Local& local =
+        mutator_marking_state_.parallel_weak_callback_worklist();
+    while (local.Pop(&item)) {
+      item.callback(broker, item.parameter);
+    }
+  }
+
   // Weak callbacks should not add any new objects for marking.
   DCHECK(marking_worklists_.marking_worklist()->IsEmpty());
 }

src/heap/cppgc/marking-state.h

@@ -140,6 +140,7 @@ class BasicMarkingState : public MarkingStateBase {
     MarkingStateBase::Publish();
     previously_not_fully_constructed_worklist_.Publish();
     weak_callback_worklist_.Publish();
+    parallel_weak_callback_worklist_.Publish();
     write_barrier_worklist_.Publish();
     concurrent_marking_bailout_worklist_.Publish();
     discovered_ephemeron_pairs_worklist_.Publish();
@@ -154,6 +155,10 @@ class BasicMarkingState : public MarkingStateBase {
   MarkingWorklists::WeakCallbackWorklist::Local& weak_callback_worklist() {
     return weak_callback_worklist_;
   }
+  MarkingWorklists::WeakCallbackWorklist::Local&
+  parallel_weak_callback_worklist() {
+    return parallel_weak_callback_worklist_;
+  }
   MarkingWorklists::WriteBarrierWorklist::Local& write_barrier_worklist() {
     return write_barrier_worklist_;
   }
@@ -202,6 +207,8 @@ class BasicMarkingState : public MarkingStateBase {
   MarkingWorklists::PreviouslyNotFullyConstructedWorklist::Local
       previously_not_fully_constructed_worklist_;
   MarkingWorklists::WeakCallbackWorklist::Local weak_callback_worklist_;
+  MarkingWorklists::WeakCallbackWorklist::Local
+      parallel_weak_callback_worklist_;
   MarkingWorklists::WriteBarrierWorklist::Local write_barrier_worklist_;
   MarkingWorklists::ConcurrentMarkingBailoutWorklist::Local
       concurrent_marking_bailout_worklist_;
@@ -230,6 +237,8 @@ BasicMarkingState::BasicMarkingState(HeapBase& heap,
       previously_not_fully_constructed_worklist_(
           marking_worklists.previously_not_fully_constructed_worklist()),
       weak_callback_worklist_(marking_worklists.weak_callback_worklist()),
+      parallel_weak_callback_worklist_(
+          marking_worklists.parallel_weak_callback_worklist()),
       write_barrier_worklist_(marking_worklists.write_barrier_worklist()),
       concurrent_marking_bailout_worklist_(
           marking_worklists.concurrent_marking_bailout_worklist()),
@@ -258,7 +267,7 @@ void BasicMarkingState::RegisterWeakReferenceIfNeeded(
   if (!header.IsInConstruction<AccessMode::kAtomic>() &&
       header.IsMarked<AccessMode::kAtomic>())
     return;
-  RegisterWeakCallback(weak_callback, parameter);
+  parallel_weak_callback_worklist_.Push({weak_callback, parameter});
 }
 
 void BasicMarkingState::RegisterWeakCallback(WeakCallback callback,

src/heap/cppgc/marking-worklists.cc

@@ -16,6 +16,7 @@ void MarkingWorklists::ClearForTesting() {
   previously_not_fully_constructed_worklist_.Clear();
   write_barrier_worklist_.Clear();
   weak_callback_worklist_.Clear();
+  parallel_weak_callback_worklist_.Clear();
   concurrent_marking_bailout_worklist_.Clear();
   discovered_ephemeron_pairs_worklist_.Clear();
   ephemeron_pairs_for_processing_worklist_.Clear();

src/heap/cppgc/marking-worklists.h

@@ -101,6 +101,9 @@ class MarkingWorklists {
   WeakCallbackWorklist* weak_callback_worklist() {
     return &weak_callback_worklist_;
   }
+  WeakCallbackWorklist* parallel_weak_callback_worklist() {
+    return &parallel_weak_callback_worklist_;
+  }
   ConcurrentMarkingBailoutWorklist* concurrent_marking_bailout_worklist() {
     return &concurrent_marking_bailout_worklist_;
   }
@@ -125,7 +128,10 @@ class MarkingWorklists {
   PreviouslyNotFullyConstructedWorklist
       previously_not_fully_constructed_worklist_;
   WriteBarrierWorklist write_barrier_worklist_;
+  // Hold weak callbacks which can only invoke on main thread.
   WeakCallbackWorklist weak_callback_worklist_;
+  // Hold weak callbacks which can invoke on main or worker thread.
+  WeakCallbackWorklist parallel_weak_callback_worklist_;
   ConcurrentMarkingBailoutWorklist concurrent_marking_bailout_worklist_;
   EphemeronPairsWorklist discovered_ephemeron_pairs_worklist_;
   EphemeronPairsWorklist ephemeron_pairs_for_processing_worklist_;

src/heap/cppgc/stats-collector.h

@@ -60,7 +60,8 @@ namespace internal {
 
 #define CPPGC_FOR_ALL_HISTOGRAM_CONCURRENT_SCOPES(V) \
   V(ConcurrentMark)                                  \
-  V(ConcurrentSweep)
+  V(ConcurrentSweep)                                 \
+  V(ConcurrentWeakCallback)
 
 #define CPPGC_FOR_ALL_CONCURRENT_SCOPES(V) V(ConcurrentMarkProcessEphemerons)