[vm, gc] Run parallel GC tasks on mutator threads blocked for the safepoint instead of new thread pool workers.

unit_test_suites.dart --skipTestsThatRequireGit -j8
6:59 -> 6:23 (-8.6%)

TEST=ci
Bug: https://github.com/dart-lang/sdk/issues/55713
Change-Id: I5256175abf4751b3b5e877ed5abdc76a90cd2fb3
Reviewed-on: https://dart-review.googlesource.com/c/sdk/+/381383
Commit-Queue: Ryan Macnak <rmacnak@google.com>
Reviewed-by: Alexander Aprelev <aam@google.com>

runtime/vm/heap/compactor.cc

@@ -122,7 +122,7 @@ struct Partition {
   Page* tail;
 };
 
-class CompactorTask : public ThreadPool::Task {
+class CompactorTask : public SafepointTask {
  public:
   CompactorTask(IsolateGroup* isolate_group,
                 GCCompactor* compactor,
@@ -147,11 +147,14 @@ class CompactorTask : public ThreadPool::Task {
         free_page_(nullptr),
         free_current_(0),
         free_end_(0) {}
+  ~CompactorTask() { barrier_->Release(); }
 
-  void Run();
-  void RunEnteredIsolateGroup();
+  void Run() override;
+  void RunBlockedAtSafepoint() override;
+  void RunMain() override;
 
  private:
+  void RunEnteredIsolateGroup();
   void PlanPage(Page* page);
   void SlidePage(Page* page);
   uword PlanBlock(uword first_object, ForwardingPage* forwarding_page);
@@ -292,24 +295,14 @@ void GCCompactor::Compact(Page* pages, FreeList* freelist, Mutex* pages_lock) {
     RelaxedAtomic<intptr_t> next_sliding_task = {0};
     RelaxedAtomic<intptr_t> next_forwarding_task = {0};
 
-    for (intptr_t task_index = 0; task_index < num_tasks; task_index++) {
-      if (task_index < (num_tasks - 1)) {
-        // Begin compacting on a helper thread.
-        Dart::thread_pool()->Run<CompactorTask>(
-            thread()->isolate_group(), this, barrier, &next_planning_task,
-            &next_setup_task, &next_sliding_task, &next_forwarding_task,
-            num_tasks, partitions, freelist);
-      } else {
-        // Last worker is the main thread.
-        CompactorTask task(thread()->isolate_group(), this, barrier,
-                           &next_planning_task, &next_setup_task,
-                           &next_sliding_task, &next_forwarding_task, num_tasks,
-                           partitions, freelist);
-        task.RunEnteredIsolateGroup();
-        barrier->Sync();
-        barrier->Release();
-      }
+    IntrusiveDList<SafepointTask> tasks;
+    for (intptr_t i = 0; i < num_tasks; i++) {
+      tasks.Append(new CompactorTask(thread()->isolate_group(), this, barrier,
+                                     &next_planning_task, &next_setup_task,
+                                     &next_sliding_task, &next_forwarding_task,
+                                     num_tasks, partitions, freelist));
     }
+    thread()->isolate_group()->safepoint_handler()->RunTasks(&tasks);
   }
 
   // Update inner pointers in typed data views (needs to be done after all
@@ -399,7 +392,6 @@ void GCCompactor::Compact(Page* pages, FreeList* freelist, Mutex* pages_lock) {
 
 void CompactorTask::Run() {
   if (!barrier_->TryEnter()) {
-    barrier_->Release();
     return;
   }
 
@@ -414,7 +406,28 @@ void CompactorTask::Run() {
 
   // This task is done. Notify the original thread.
   barrier_->Sync();
-  barrier_->Release();
+}
+
+void CompactorTask::RunBlockedAtSafepoint() {
+  if (!barrier_->TryEnter()) {
+    return;
+  }
+
+  Thread* thread = Thread::Current();
+  Thread::TaskKind saved_task_kind = thread->task_kind();
+  thread->set_task_kind(Thread::kCompactorTask);
+
+  RunEnteredIsolateGroup();
+
+  thread->set_task_kind(saved_task_kind);
+
+  barrier_->Sync();
+}
+
+void CompactorTask::RunMain() {
+  RunEnteredIsolateGroup();
+
+  barrier_->Sync();
 }
 
 void CompactorTask::RunEnteredIsolateGroup() {

runtime/vm/heap/gc_shared.h

@@ -215,7 +215,7 @@ void MournFinalizerEntry(GCVisitorType* visitor,
   // workers are not, but they bypass safepoint because the main
   // worker is at a safepoint already.
   ASSERT(Thread::Current()->OwnsGCSafepoint() ||
-         Thread::Current()->BypassSafepoints());
+         (Thread::Current()->task_kind() == Thread::kScavengerTask));
 
   if (finalizer.IsNativeFinalizer()) {
     NativeFinalizerPtr native_finalizer =

runtime/vm/heap/incremental_compactor.cc

@@ -99,7 +99,7 @@ struct PrologueState {
   intptr_t freelist_limit;
 };
 
-class PrologueTask : public ThreadPool::Task {
+class PrologueTask : public SafepointTask {
  public:
   PrologueTask(ThreadBarrier* barrier,
                IsolateGroup* isolate_group,
@@ -109,10 +109,10 @@ class PrologueTask : public ThreadPool::Task {
         isolate_group_(isolate_group),
         old_space_(old_space),
         state_(state) {}
+  ~PrologueTask() { barrier_->Release(); }
 
-  void Run() {
+  void Run() override {
     if (!barrier_->TryEnter()) {
-      barrier_->Release();
       return;
     }
 
@@ -126,7 +126,28 @@ class PrologueTask : public ThreadPool::Task {
     Thread::ExitIsolateGroupAsHelper(/*bypass_safepoint=*/true);
 
     barrier_->Sync();
-    barrier_->Release();
+  }
+
+  void RunBlockedAtSafepoint() override {
+    if (!barrier_->TryEnter()) {
+      return;
+    }
+
+    Thread* thread = Thread::Current();
+    Thread::TaskKind saved_task_kind = thread->task_kind();
+    thread->set_task_kind(Thread::kIncrementalCompactorTask);
+
+    RunEnteredIsolateGroup();
+
+    thread->set_task_kind(saved_task_kind);
+
+    barrier_->Sync();
+  }
+
+  void RunMain() override {
+    RunEnteredIsolateGroup();
+
+    barrier_->Sync();
   }
 
   void RunEnteredIsolateGroup() {
@@ -257,20 +278,11 @@ bool GCIncrementalCompactor::SelectEvacuationCandidates(PageSpace* old_space) {
       isolate_group->heap()->new_space()->NumScavengeWorkers();
   RELEASE_ASSERT(num_tasks > 0);
   ThreadBarrier* barrier = new ThreadBarrier(num_tasks, 1);
+  IntrusiveDList<SafepointTask> tasks;
   for (intptr_t i = 0; i < num_tasks; i++) {
-    if (i < (num_tasks - 1)) {
-      // Begin compacting on a helper thread.
-      bool result = Dart::thread_pool()->Run<PrologueTask>(
-          barrier, isolate_group, old_space, &state);
-      ASSERT(result);
-    } else {
-      // Last worker is the main thread.
-      PrologueTask task(barrier, isolate_group, old_space, &state);
-      task.RunEnteredIsolateGroup();
-      barrier->Sync();
-      barrier->Release();
-    }
+    tasks.Append(new PrologueTask(barrier, isolate_group, old_space, &state));
   }
+  isolate_group->safepoint_handler()->RunTasks(&tasks);
 
   for (intptr_t i = PageSpace::kDataFreelist, n = old_space->num_freelists_;
        i < n; i++) {
@@ -627,7 +639,7 @@ class EpilogueState {
   RelaxedAtomic<intptr_t> new_free_size_ = {0};
 };
 
-class EpilogueTask : public ThreadPool::Task {
+class EpilogueTask : public SafepointTask {
  public:
   EpilogueTask(ThreadBarrier* barrier,
                IsolateGroup* isolate_group,
@@ -639,8 +651,9 @@ class EpilogueTask : public ThreadPool::Task {
         old_space_(old_space),
         freelist_(freelist),
         state_(state) {}
+  ~EpilogueTask() { barrier_->Release(); }
 
-  void Run() {
+  void Run() override {
     bool result = Thread::EnterIsolateGroupAsHelper(
         isolate_group_, Thread::kIncrementalCompactorTask,
         /*bypass_safepoint=*/true);
@@ -651,7 +664,24 @@ class EpilogueTask : public ThreadPool::Task {
     Thread::ExitIsolateGroupAsHelper(/*bypass_safepoint=*/true);
 
     barrier_->Sync();
-    barrier_->Release();
+  }
+
+  void RunBlockedAtSafepoint() override {
+    Thread* thread = Thread::Current();
+    Thread::TaskKind saved_task_kind = thread->task_kind();
+    thread->set_task_kind(Thread::kIncrementalCompactorTask);
+
+    RunEnteredIsolateGroup();
+
+    thread->set_task_kind(saved_task_kind);
+
+    barrier_->Sync();
+  }
+
+  void RunMain() override {
+    RunEnteredIsolateGroup();
+
+    barrier_->Sync();
   }
 
   void RunEnteredIsolateGroup() {
@@ -882,21 +912,12 @@ void GCIncrementalCompactor::Evacuate(PageSpace* old_space) {
       isolate_group->heap()->new_space()->NumScavengeWorkers();
   RELEASE_ASSERT(num_tasks > 0);
   ThreadBarrier* barrier = new ThreadBarrier(num_tasks, num_tasks);
+  IntrusiveDList<SafepointTask> tasks;
   for (intptr_t i = 0; i < num_tasks; i++) {
-    // Begin compacting on a helper thread.
-    FreeList* freelist = old_space->DataFreeList(i);
-    if (i < (num_tasks - 1)) {
-      bool result = Dart::thread_pool()->Run<EpilogueTask>(
-          barrier, isolate_group, old_space, freelist, &state);
-      ASSERT(result);
-    } else {
-      // Last worker is the main thread.
-      EpilogueTask task(barrier, isolate_group, old_space, freelist, &state);
-      task.RunEnteredIsolateGroup();
-      barrier->Sync();
-      barrier->Release();
-    }
+    tasks.Append(new EpilogueTask(barrier, isolate_group, old_space,
+                                  old_space->DataFreeList(i), &state));
   }
+  isolate_group->safepoint_handler()->RunTasks(&tasks);
 
   old_space->heap_->new_space()->set_freed_in_words(state.NewFreeSize() >>
                                                     kWordSizeLog2);

runtime/vm/heap/marker.cc

@@ -901,7 +901,7 @@ void GCMarker::ProcessRememberedSet(Thread* thread) {
   store_buffer->PushBlock(writing, StoreBuffer::kIgnoreThreshold);
 }
 
-class ParallelMarkTask : public ThreadPool::Task {
+class ParallelMarkTask : public SafepointTask {
  public:
   ParallelMarkTask(GCMarker* marker,
                    IsolateGroup* isolate_group,
@@ -915,10 +915,10 @@ class ParallelMarkTask : public ThreadPool::Task {
         barrier_(barrier),
         visitor_(visitor),
         num_busy_(num_busy) {}
+  ~ParallelMarkTask() { barrier_->Release(); }
 
-  virtual void Run() {
+  void Run() override {
     if (!barrier_->TryEnter()) {
-      barrier_->Release();
       return;
     }
 
@@ -931,7 +931,28 @@ class ParallelMarkTask : public ThreadPool::Task {
     Thread::ExitIsolateGroupAsHelper(/*bypass_safepoint=*/true);
 
     barrier_->Sync();
-    barrier_->Release();
+  }
+
+  void RunBlockedAtSafepoint() override {
+    if (!barrier_->TryEnter()) {
+      return;
+    }
+
+    Thread* thread = Thread::Current();
+    Thread::TaskKind saved_task_kind = thread->task_kind();
+    thread->set_task_kind(Thread::kMarkerTask);
+
+    RunEnteredIsolateGroup();
+
+    thread->set_task_kind(saved_task_kind);
+
+    barrier_->Sync();
+  }
+
+  void RunMain() override {
+    RunEnteredIsolateGroup();
+
+    barrier_->Sync();
   }
 
   void RunEnteredIsolateGroup() {
@@ -1343,8 +1364,8 @@ void GCMarker::MarkObjects(PageSpace* page_space) {
       ResetSlices();
       // Used to coordinate draining among tasks; all start out as 'busy'.
       RelaxedAtomic<uintptr_t> num_busy = 0;
-      // Phase 1: Iterate over roots and drain marking stack in tasks.
 
+      IntrusiveDList<SafepointTask> tasks;
       for (intptr_t i = 0; i < num_tasks; ++i) {
         SyncMarkingVisitor* visitor = visitors_[i];
         // Visitors may or may not have already been created depending on
@@ -1363,23 +1384,12 @@ void GCMarker::MarkObjects(PageSpace* page_space) {
         // such a visitor's local blocks.
         visitor->Flush(&global_list_);
         // Need to move weak property list too.
-
-        if (i < (num_tasks - 1)) {
-          // Begin marking on a helper thread.
-          bool result = Dart::thread_pool()->Run<ParallelMarkTask>(
-              this, isolate_group_, &old_marking_stack_, barrier, visitor,
-              &num_busy);
-          ASSERT(result);
-        } else {
-          // Last worker is the main thread.
-          visitor->Adopt(&global_list_);
-          ParallelMarkTask task(this, isolate_group_, &old_marking_stack_,
-                                barrier, visitor, &num_busy);
-          task.RunEnteredIsolateGroup();
-          barrier->Sync();
-          barrier->Release();
-        }
+        tasks.Append(new ParallelMarkTask(this, isolate_group_,
+                                          &old_marking_stack_, barrier, visitor,
+                                          &num_busy));
       }
+      visitors_[0]->Adopt(&global_list_);
+      isolate_group_->safepoint_handler()->RunTasks(&tasks);
 
       for (intptr_t i = 0; i < num_tasks; i++) {
         SyncMarkingVisitor* visitor = visitors_[i];

runtime/vm/heap/safepoint.cc

@@ -60,7 +60,7 @@ ForceGrowthSafepointOperationScope::~ForceGrowthSafepointOperationScope() {
 }
 
 SafepointHandler::SafepointHandler(IsolateGroup* isolate_group)
-    : isolate_group_(isolate_group) {
+    : isolate_group_(isolate_group), tasks_() {
   handlers_[SafepointLevel::kGC] =
       new LevelHandler(isolate_group, SafepointLevel::kGC);
   handlers_[SafepointLevel::kGCAndDeopt] =
@@ -70,6 +70,7 @@ SafepointHandler::SafepointHandler(IsolateGroup* isolate_group)
 }
 
 SafepointHandler::~SafepointHandler() {
+  ASSERT(tasks_.IsEmpty());
   for (intptr_t level = 0; level < SafepointLevel::kNumLevels; ++level) {
     ASSERT(handlers_[level]->owner_ == nullptr);
     delete handlers_[level];
@@ -359,12 +360,68 @@ void SafepointHandler::LevelHandler::NotifyWeAreParked(Thread* T) {
 void SafepointHandler::ExitSafepointLocked(Thread* T,
                                            MonitorLocker* tl,
                                            SafepointLevel level) {
+  ASSERT(T == Thread::Current());
   while (T->IsSafepointRequestedLocked(level)) {
     T->SetBlockedForSafepoint(true);
     tl->Wait();
     T->SetBlockedForSafepoint(false);
+
+    MonitorLeaveScope mls(tl);
+    SafepointTask* task = nullptr;
+    {
+      MonitorLocker ml(threads_lock());
+      if (!tasks_.IsEmpty()) {
+        task = tasks_.RemoveFirst();
+      }
+    }
+    if (task != nullptr) {
+      task->RunBlockedAtSafepoint();
+      delete task;
+    }
   }
   T->SetAtSafepoint(false, level);
 }
 
+void SafepointHandler::RunTasks(IntrusiveDList<SafepointTask>* tasks) {
+  ASSERT(Thread::Current()->OwnsSafepoint());
+
+  // Withold one task for the main thread.
+  ASSERT(!tasks->IsEmpty());
+  SafepointTask* main = tasks->RemoveFirst();
+
+  // First use threads blocked at this safepoint.
+  {
+    MonitorLocker tl(threads_lock());
+    ASSERT(tasks_.IsEmpty());
+    for (auto current = isolate_group()->thread_registry()->active_list();
+         current != nullptr; current = current->next()) {
+      if (tasks->IsEmpty()) break;
+
+      MonitorLocker tl(current->thread_lock());
+      if (current->IsBlockedForSafepoint()) {
+        tasks_.Append(tasks->RemoveFirst());
+        tl.Notify();
+      }
+    }
+  }
+
+  // Then use thread pool workers.
+  while (!tasks->IsEmpty()) {
+    bool result = Dart::thread_pool()->Run(tasks->RemoveFirst());
+    ASSERT(result);
+  }
+
+  // Run one task on the main thread.
+  main->RunMain();
+  delete main;
+
+  // Clean up any tasks that took too long to start.
+  {
+    MonitorLocker tl(threads_lock());
+    while (!tasks_.IsEmpty()) {
+      delete tasks_.RemoveFirst();
+    }
+  }
+}
+
 }  // namespace dart

runtime/vm/heap/safepoint.h

@@ -76,6 +76,22 @@ class ForceGrowthSafepointOperationScope : public ThreadStackResource {
   DISALLOW_COPY_AND_ASSIGN(ForceGrowthSafepointOperationScope);
 };
 
+// Subclasses of SafepointTask are able to run on thread blocked at a safepoint.
+class SafepointTask : public ThreadPool::Task,
+                      public IntrusiveDListEntry<SafepointTask> {
+ protected:
+  SafepointTask() {}
+
+ public:
+  virtual ~SafepointTask() {}
+
+  virtual void RunBlockedAtSafepoint() = 0;
+  virtual void RunMain() = 0;
+
+ private:
+  DISALLOW_COPY_AND_ASSIGN(SafepointTask);
+};
+
 // Implements handling of safepoint operations for all threads in an
 // IsolateGroup.
 class SafepointHandler {
@@ -114,6 +130,8 @@ class SafepointHandler {
     return false;
   }
 
+  void RunTasks(IntrusiveDList<SafepointTask>* tasks);
+
  private:
   class LevelHandler {
    public:
@@ -201,6 +219,7 @@ class SafepointHandler {
   IsolateGroup* isolate_group_;
 
   LevelHandler* handlers_[SafepointLevel::kNumLevels];
+  IntrusiveDList<SafepointTask> tasks_;
 
   friend class Isolate;
   friend class IsolateGroup;

runtime/vm/heap/scavenger.cc

@@ -670,7 +670,7 @@ class ScavengerWeakVisitor : public HandleVisitor {
   DISALLOW_COPY_AND_ASSIGN(ScavengerWeakVisitor);
 };
 
-class ParallelScavengerTask : public ThreadPool::Task {
+class ParallelScavengerTask : public SafepointTask {
  public:
   ParallelScavengerTask(IsolateGroup* isolate_group,
                         ThreadBarrier* barrier,
@@ -680,10 +680,10 @@ class ParallelScavengerTask : public ThreadPool::Task {
         barrier_(barrier),
         visitor_(visitor),
         num_busy_(num_busy) {}
+  ~ParallelScavengerTask() { barrier_->Release(); }
 
-  virtual void Run() {
+  void Run() override {
     if (!barrier_->TryEnter()) {
-      barrier_->Release();
       return;
     }
 
@@ -696,7 +696,28 @@ class ParallelScavengerTask : public ThreadPool::Task {
     Thread::ExitIsolateGroupAsHelper(/*bypass_safepoint=*/true);
 
     barrier_->Sync();
-    barrier_->Release();
+  }
+
+  void RunBlockedAtSafepoint() override {
+    if (!barrier_->TryEnter()) {
+      return;
+    }
+
+    Thread* thread = Thread::Current();
+    Thread::TaskKind saved_task_kind = thread->task_kind();
+    thread->set_task_kind(Thread::kScavengerTask);
+
+    RunEnteredIsolateGroup();
+
+    thread->set_task_kind(saved_task_kind);
+
+    barrier_->Sync();
+  }
+
+  void RunMain() override {
+    RunEnteredIsolateGroup();
+
+    barrier_->Sync();
   }
 
   void RunEnteredIsolateGroup() {
@@ -2039,28 +2060,21 @@ intptr_t Scavenger::ParallelScavenge(SemiSpace* from) {
   ThreadBarrier* barrier = new ThreadBarrier(num_tasks, 1);
   RelaxedAtomic<uintptr_t> num_busy = 0;
 
+  IsolateGroup* isolate_group = heap_->isolate_group();
+
   ParallelScavengerVisitor** visitors =
       new ParallelScavengerVisitor*[num_tasks];
+  IntrusiveDList<SafepointTask> tasks;
   for (intptr_t i = 0; i < num_tasks; i++) {
     FreeList* freelist = heap_->old_space()->DataFreeList(i);
-    visitors[i] = new ParallelScavengerVisitor(
-        heap_->isolate_group(), this, from, freelist, &promotion_stack_);
-    if (i < (num_tasks - 1)) {
-      // Begin scavenging on a helper thread.
-      bool result = Dart::thread_pool()->Run<ParallelScavengerTask>(
-          heap_->isolate_group(), barrier, visitors[i], &num_busy);
-      ASSERT(result);
-    } else {
-      // Last worker is the main thread.
-      ParallelScavengerTask task(heap_->isolate_group(), barrier, visitors[i],
-                                 &num_busy);
-      task.RunEnteredIsolateGroup();
-      barrier->Sync();
-      barrier->Release();
-    }
+    visitors[i] = new ParallelScavengerVisitor(isolate_group, this, from,
+                                               freelist, &promotion_stack_);
+    tasks.Append(new ParallelScavengerTask(isolate_group, barrier, visitors[i],
+                                           &num_busy));
   }
+  isolate_group->safepoint_handler()->RunTasks(&tasks);
 
-  StoreBuffer* store_buffer = heap_->isolate_group()->store_buffer();
+  StoreBuffer* store_buffer = isolate_group->store_buffer();
   for (intptr_t i = 0; i < num_tasks; i++) {
     ParallelScavengerVisitor* visitor = visitors[i];
     visitor->Finalize(store_buffer);

runtime/vm/thread.cc

@@ -396,11 +396,6 @@ void Thread::EnterIsolate(Isolate* isolate) {
     ASSERT(thread->scheduled_dart_mutator_isolate_ == isolate);
     ASSERT(thread->isolate() == isolate);
     ASSERT(thread->isolate_group() == isolate->group());
-    {
-      // Descheduled isolates are reloadable (if nothing else prevents it).
-      RawReloadParticipationScope enable_reload(thread);
-      thread->ExitSafepoint();
-    }
   } else {
     thread = AddActiveThread(group, isolate, /*is_dart_mutator*/ true,
                              /*bypass_safepoint=*/false);
@@ -411,6 +406,14 @@ void Thread::EnterIsolate(Isolate* isolate) {
 
   isolate->scheduled_mutator_thread_ = thread;
   ResumeDartMutatorThreadInternal(thread);
+
+  if (is_resumable) {
+    // Descheduled isolates are reloadable (if nothing else prevents it).
+    RawReloadParticipationScope enable_reload(thread);
+    thread->ExitSafepoint();
+  }
+
+  ASSERT(!thread->IsAtSafepoint());
 }
 
 static bool ShouldSuspend(bool isolate_shutdown, Thread* thread) {
@@ -556,7 +559,6 @@ void Thread::SuspendDartMutatorThreadInternal(Thread* thread,
 }
 
 void Thread::ResumeThreadInternal(Thread* thread) {
-  ASSERT(!thread->IsAtSafepoint());
   ASSERT(thread->isolate_group() != nullptr);
   ASSERT(thread->execution_state() == Thread::kThreadInNative);
   ASSERT(thread->vm_tag() == VMTag::kInvalidTagId ||

runtime/vm/thread.h

@@ -487,6 +487,7 @@ class Thread : public ThreadState {
   }
 
   TaskKind task_kind() const { return task_kind_; }
+  void set_task_kind(TaskKind kind) { task_kind_ = kind; }
 
   // Retrieves and clears the stack overflow flags.  These are set by
   // the generated code before the slow path runtime routine for a

runtime/vm/thread_pool.h

@@ -45,6 +45,7 @@ class ThreadPool {
   bool Run(Args&&... args) {
     return RunImpl(std::unique_ptr<Task>(new T(std::forward<Args>(args)...)));
   }
+  bool Run(Task* task) { return RunImpl(std::unique_ptr<Task>(task)); }
 
   // Returns `true` if the current thread is running on the [this] thread pool.
   bool CurrentThreadIsWorker();