[heap] Introduce safepoint mechanism

Add safepoint mechanism to stop concurrent threads and bring them to a
safepoint. Threads are stopped before the safepoint and after e.g. the
GC resumed again. Each thread needs to be stopped in a safepoint, such
that all roots can be iterated safely.

Running threads need to be cooperative and are required to perform
regular safepoint polls.

Bug: v8:10315
Change-Id: I47f07e7d2ef5bc5adbba6b9e8e79a1f0f45b97ad
Reviewed-on: https://chromium-review.googlesource.com/c/v8/v8/+/2102578
Commit-Queue: Dominik Inführ <dinfuehr@chromium.org>
Reviewed-by: Ulan Degenbaev <ulan@chromium.org>
Cr-Commit-Position: refs/heads/master@{#66727}

BUILD.gn

@@ -2402,6 +2402,8 @@ v8_source_set("v8_base_without_compiler") {
     "src/heap/read-only-heap.cc",
     "src/heap/read-only-heap.h",
     "src/heap/remembered-set.h",
+    "src/heap/safepoint.cc",
+    "src/heap/safepoint.h",
     "src/heap/scavenge-job.cc",
     "src/heap/scavenge-job.h",
     "src/heap/scavenger-inl.h",

src/heap/heap.cc

@@ -50,6 +50,7 @@
 #include "src/heap/objects-visiting.h"
 #include "src/heap/read-only-heap.h"
 #include "src/heap/remembered-set.h"
+#include "src/heap/safepoint.h"
 #include "src/heap/scavenge-job.h"
 #include "src/heap/scavenger-inl.h"
 #include "src/heap/stress-marking-observer.h"
@@ -202,6 +203,7 @@ Heap::Heap()
       memory_pressure_level_(MemoryPressureLevel::kNone),
       global_pretenuring_feedback_(kInitialFeedbackCapacity),
       local_heaps_head_(nullptr),
+      safepoint_(new Safepoint(this)),
       external_string_table_(this) {
   // Ensure old_generation_size_ is a multiple of kPageSize.
   DCHECK_EQ(0, max_old_generation_size_ & (Page::kPageSize - 1));

src/heap/heap.h

@@ -83,6 +83,7 @@ class Page;
 class PagedSpace;
 class ReadOnlyHeap;
 class RootVisitor;
+class Safepoint;
 class ScavengeJob;
 class Scavenger;
 class ScavengerCollector;
@@ -624,6 +625,8 @@ class Heap {
   V8_EXPORT_PRIVATE bool ContainsLocalHeap(LocalHeap* local_heap);
   V8_EXPORT_PRIVATE bool ContainsAnyLocalHeap();
 
+  Safepoint* safepoint() { return safepoint_.get(); }
+
   V8_EXPORT_PRIVATE double MonotonicallyIncreasingTimeInMs();
 
   void RecordStats(HeapStats* stats, bool take_snapshot = false);
@@ -2171,6 +2174,8 @@ class Heap {
   base::Mutex local_heaps_mutex_;
   LocalHeap* local_heaps_head_;
 
+  std::unique_ptr<Safepoint> safepoint_;
+
   bool is_current_gc_forced_ = false;
 
   ExternalStringTable external_string_table_;
@@ -2238,6 +2243,7 @@ class Heap {
   friend class Page;
   friend class PagedSpace;
   friend class ReadOnlyRoots;
+  friend class Safepoint;
   friend class Scavenger;
   friend class ScavengerCollector;
   friend class Space;

src/heap/local-heap.cc

@@ -4,6 +4,7 @@
 
 #include "src/heap/local-heap.h"
 #include "src/heap/heap.h"
+#include "src/heap/safepoint.h"
 
 namespace v8 {
 namespace internal {
@@ -16,12 +17,18 @@ LocalHeap::LocalHeap(Heap* heap)
   heap_->AddLocalHeap(this);
 }
 
-LocalHeap::~LocalHeap() { heap_->RemoveLocalHeap(this); }
+LocalHeap::~LocalHeap() {
+  // Park thread since removing the local heap could block.
+  EnsureParkedBeforeDestruction();
+
+  heap_->RemoveLocalHeap(this);
+}
 
 void LocalHeap::Park() {
   base::MutexGuard guard(&state_mutex_);
   CHECK(state_ == ThreadState::Running);
   state_ = ThreadState::Parked;
+  state_change_.NotifyAll();
 }
 
 void LocalHeap::Unpark() {
@@ -30,6 +37,12 @@ void LocalHeap::Unpark() {
   state_ = ThreadState::Running;
 }
 
+void LocalHeap::EnsureParkedBeforeDestruction() {
+  base::MutexGuard guard(&state_mutex_);
+  state_ = ThreadState::Parked;
+  state_change_.NotifyAll();
+}
+
 void LocalHeap::RequestSafepoint() {
   safepoint_requested_.store(true, std::memory_order_relaxed);
 }
@@ -40,6 +53,7 @@ bool LocalHeap::IsSafepointRequested() {
 
 void LocalHeap::Safepoint() {
   if (IsSafepointRequested()) {
+    ClearSafepointRequested();
     EnterSafepoint();
   }
 }
@@ -48,7 +62,7 @@ void LocalHeap::ClearSafepointRequested() {
   safepoint_requested_.store(false, std::memory_order_relaxed);
 }
 
-void LocalHeap::EnterSafepoint() { UNIMPLEMENTED(); }
+void LocalHeap::EnterSafepoint() { heap_->safepoint()->EnterFromThread(this); }
 
 }  // namespace internal
 }  // namespace v8

src/heap/local-heap.h

@@ -14,6 +14,7 @@ namespace v8 {
 namespace internal {
 
 class Heap;
+class Safepoint;
 
 class LocalHeap {
  public:
@@ -26,7 +27,7 @@ class LocalHeap {
 
   // Frequently invoked by local thread to check whether safepoint was requested
   // from the main thread.
-  void Safepoint();
+  V8_EXPORT_PRIVATE void Safepoint();
 
  private:
   enum class ThreadState {
@@ -39,21 +40,41 @@ class LocalHeap {
     Safepoint
   };
 
-  void Park();
-  void Unpark();
+  V8_EXPORT_PRIVATE void Park();
+  V8_EXPORT_PRIVATE void Unpark();
+  void EnsureParkedBeforeDestruction();
+
   bool IsSafepointRequested();
   void ClearSafepointRequested();
+
   void EnterSafepoint();
 
   Heap* heap_;
+
   base::Mutex state_mutex_;
-  base::ConditionVariable state_condvar_;
+  base::ConditionVariable state_change_;
   ThreadState state_;
+
   std::atomic<bool> safepoint_requested_;
+
   LocalHeap* prev_;
   LocalHeap* next_;
 
   friend class Heap;
+  friend class Safepoint;
+  friend class ParkedScope;
+};
+
+class ParkedScope {
+ public:
+  explicit ParkedScope(LocalHeap* local_heap) : local_heap_(local_heap) {
+    local_heap_->Park();
+  }
+
+  ~ParkedScope() { local_heap_->Unpark(); }
+
+ private:
+  LocalHeap* local_heap_;
 };
 
 }  // namespace internal

src/heap/safepoint.cc

+// Copyright 2020 the V8 project authors. All rights reserved.
+// Use of this source code is governed by a BSD-style license that can be
+// found in the LICENSE file.
+
+#include "src/heap/safepoint.h"
+#include "src/heap/heap.h"
+#include "src/heap/local-heap.h"
+
+namespace v8 {
+namespace internal {
+
+Safepoint::Safepoint(Heap* heap) : heap_(heap) {}
+
+void Safepoint::StopThreads() {
+  heap_->local_heaps_mutex_.Lock();
+
+  barrier_.Arm();
+
+  for (LocalHeap* current = heap_->local_heaps_head_; current;
+       current = current->next_) {
+    current->RequestSafepoint();
+  }
+
+  for (LocalHeap* current = heap_->local_heaps_head_; current;
+       current = current->next_) {
+    current->state_mutex_.Lock();
+
+    while (current->state_ == LocalHeap::ThreadState::Running) {
+      current->state_change_.Wait(&current->state_mutex_);
+    }
+  }
+}
+
+void Safepoint::ResumeThreads() {
+  for (LocalHeap* current = heap_->local_heaps_head_; current;
+       current = current->next_) {
+    current->state_mutex_.Unlock();
+  }
+
+  barrier_.Disarm();
+
+  heap_->local_heaps_mutex_.Unlock();
+}
+
+void Safepoint::EnterFromThread(LocalHeap* local_heap) {
+  {
+    base::MutexGuard guard(&local_heap->state_mutex_);
+    local_heap->state_ = LocalHeap::ThreadState::Safepoint;
+    local_heap->state_change_.NotifyAll();
+  }
+
+  barrier_.Wait();
+
+  {
+    base::MutexGuard guard(&local_heap->state_mutex_);
+    local_heap->state_ = LocalHeap::ThreadState::Running;
+  }
+}
+
+void Safepoint::Barrier::Arm() {
+  base::MutexGuard guard(&mutex_);
+  CHECK(!armed_);
+  armed_ = true;
+}
+
+void Safepoint::Barrier::Disarm() {
+  base::MutexGuard guard(&mutex_);
+  CHECK(armed_);
+  armed_ = false;
+  cond_.NotifyAll();
+}
+
+void Safepoint::Barrier::Wait() {
+  base::MutexGuard guard(&mutex_);
+  while (armed_) {
+    cond_.Wait(&mutex_);
+  }
+}
+
+SafepointScope::SafepointScope(Heap* heap) : safepoint_(heap->safepoint()) {
+  safepoint_->StopThreads();
+}
+
+SafepointScope::~SafepointScope() { safepoint_->ResumeThreads(); }
+
+}  // namespace internal
+}  // namespace v8

src/heap/safepoint.h

+// Copyright 2020 the V8 project authors. All rights reserved.
+// Use of this source code is governed by a BSD-style license that can be
+// found in the LICENSE file.
+
+#ifndef V8_HEAP_SAFEPOINT_H_
+#define V8_HEAP_SAFEPOINT_H_
+
+#include "src/base/platform/condition-variable.h"
+#include "src/base/platform/mutex.h"
+
+namespace v8 {
+namespace internal {
+
+class Heap;
+class LocalHeap;
+
+class Safepoint {
+ public:
+  explicit Safepoint(Heap* heap);
+
+  // Enter the safepoint from a thread
+  void EnterFromThread(LocalHeap* local_heap);
+
+ private:
+  class Barrier {
+    base::Mutex mutex_;
+    base::ConditionVariable cond_;
+    bool armed_;
+
+   public:
+    Barrier() : armed_(false) {}
+
+    void Arm();
+    void Disarm();
+    void Wait();
+  };
+
+  void StopThreads();
+  void ResumeThreads();
+
+  Barrier barrier_;
+  Heap* heap_;
+
+  friend class SafepointScope;
+};
+
+class SafepointScope {
+ public:
+  V8_EXPORT_PRIVATE explicit SafepointScope(Heap* heap);
+  V8_EXPORT_PRIVATE ~SafepointScope();
+
+ private:
+  Safepoint* safepoint_;
+};
+
+}  // namespace internal
+}  // namespace v8
+
+#endif  // V8_HEAP_SAFEPOINT_H_

test/unittests/BUILD.gn

@@ -176,6 +176,7 @@ v8_source_set("unittests_sources") {
     "heap/memory-reducer-unittest.cc",
     "heap/object-stats-unittest.cc",
     "heap/off-thread-factory-unittest.cc",
+    "heap/safepoint-unittest.cc",
     "heap/slot-set-unittest.cc",
     "heap/spaces-unittest.cc",
     "heap/unmapper-unittest.cc",

test/unittests/heap/safepoint-unittest.cc

+// Copyright 2020 the V8 project authors. All rights reserved.
+// Use of this source code is governed by a BSD-style license that can be
+// found in the LICENSE file.
+
+#include "src/heap/safepoint.h"
+#include "src/base/platform/mutex.h"
+#include "src/base/platform/platform.h"
+#include "src/heap/heap.h"
+#include "src/heap/local-heap.h"
+#include "test/unittests/test-utils.h"
+#include "testing/gtest/include/gtest/gtest.h"
+
+namespace v8 {
+namespace internal {
+
+using SafepoinTest = TestWithIsolate;
+
+TEST_F(SafepoinTest, ReachSafepointWithoutLocalHeaps) {
+  Heap* heap = i_isolate()->heap();
+  bool run = false;
+  {
+    SafepointScope scope(heap);
+    run = true;
+  }
+  CHECK(run);
+}
+
+class ParkedThread final : public v8::base::Thread {
+ public:
+  ParkedThread(Heap* heap, base::Mutex* mutex)
+      : v8::base::Thread(base::Thread::Options("ThreadWithLocalHeap")),
+        heap_(heap),
+        mutex_(mutex) {}
+
+  void Run() override {
+    LocalHeap local_heap(heap_);
+
+    if (mutex_) {
+      ParkedScope scope(&local_heap);
+      base::MutexGuard guard(mutex_);
+    }
+  }
+
+  Heap* heap_;
+  base::Mutex* mutex_;
+};
+
+TEST_F(SafepoinTest, StopParkedThreads) {
+  Heap* heap = i_isolate()->heap();
+
+  int safepoints = 0;
+
+  const int kThreads = 10;
+  const int kRuns = 5;
+
+  for (int run = 0; run < kRuns; run++) {
+    base::Mutex mutex;
+    std::vector<ParkedThread*> threads;
+
+    mutex.Lock();
+
+    for (int i = 0; i < kThreads; i++) {
+      ParkedThread* thread =
+          new ParkedThread(heap, i % 2 == 0 ? &mutex : nullptr);
+      CHECK(thread->Start());
+      threads.push_back(thread);
+    }
+
+    {
+      SafepointScope scope(heap);
+      safepoints++;
+    }
+    mutex.Unlock();
+
+    for (ParkedThread* thread : threads) {
+      thread->Join();
+      delete thread;
+    }
+  }
+
+  CHECK_EQ(safepoints, kRuns);
+}
+
+static const int kRuns = 10000;
+
+class RunningThread final : public v8::base::Thread {
+ public:
+  RunningThread(Heap* heap, std::atomic<int>* counter)
+      : v8::base::Thread(base::Thread::Options("ThreadWithLocalHeap")),
+        heap_(heap),
+        counter_(counter) {}
+
+  void Run() override {
+    LocalHeap local_heap(heap_);
+
+    for (int i = 0; i < kRuns; i++) {
+      counter_->fetch_add(1);
+      if (i % 100 == 0) local_heap.Safepoint();
+    }
+  }
+
+  Heap* heap_;
+  std::atomic<int>* counter_;
+};
+
+TEST_F(SafepoinTest, StopRunningThreads) {
+  Heap* heap = i_isolate()->heap();
+
+  const int kThreads = 10;
+  const int kRuns = 5;
+  const int kSafepoints = 3;
+  int safepoint_count = 0;
+
+  for (int run = 0; run < kRuns; run++) {
+    std::atomic<int> counter(0);
+    std::vector<RunningThread*> threads;
+
+    for (int i = 0; i < kThreads; i++) {
+      RunningThread* thread = new RunningThread(heap, &counter);
+      CHECK(thread->Start());
+      threads.push_back(thread);
+    }
+
+    for (int i = 0; i < kSafepoints; i++) {
+      SafepointScope scope(heap);
+      safepoint_count++;
+    }
+
+    for (RunningThread* thread : threads) {
+      thread->Join();
+      delete thread;
+    }
+  }
+
+  CHECK_EQ(safepoint_count, kRuns * kSafepoints);
+}
+
+}  // namespace internal
+}  // namespace v8