[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.

The last version of this CL was reverted because safepoint_requested_
wasn't initialized (see https://crrev.com/c/2105634).

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

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 {
@@ -11,17 +12,24 @@ namespace internal {
 LocalHeap::LocalHeap(Heap* heap)
     : heap_(heap),
       state_(ThreadState::Running),
+      safepoint_requested_(false),
       prev_(nullptr),
       next_(nullptr) {
   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 +38,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 +54,7 @@ bool LocalHeap::IsSafepointRequested() {
 
 void LocalHeap::Safepoint() {
   if (IsSafepointRequested()) {
+    ClearSafepointRequested();
     EnterSafepoint();
   }
 }
@@ -48,7 +63,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