Reland "[compiler, heap] Create LocalHeap outside of ExecuteJob"

This is a reland of 44708a5b

Original change's description:
> [compiler, heap] Create LocalHeap outside of ExecuteJob
>
> Create LocalHeap directly in the Task or in GetOptimizedCodeNow and
> pass its reference as argument to ExecuteJob. This allows us to create
> LocalHeap differently for the main and background thread, e.g. by
> passing an additional argument to the constructor in the future.
> It will be required in the future anyways when the main thread will
> have its own LocalHeap/LocalIsolate.
>
> Extending the scope of LocalHeap, also made
> HandleBase::IsDereferenceAllowed more precise and uncovered two
> potential issues: heap accesses in
> OptimizingCompileDispatcher::CompileNext and PipelineImpl::AssembleCode
> with --code-comments.
>
> LocalHeap can now be created in the parked state. Also fixed a data
> race with LocalHeap's destructor publishing write barrier entries
> without holding the lock.
>
> Bug: v8:10315
> Change-Id: I9226972601a07b87108cd66efbbb6a0d118af58d
> Reviewed-on: https://chromium-review.googlesource.com/c/v8/v8/+/2460818
> Commit-Queue: Georg Neis <neis@chromium.org>
> Reviewed-by: Leszek Swirski <leszeks@chromium.org>
> Reviewed-by: Santiago Aboy Solanes <solanes@chromium.org>
> Reviewed-by: Georg Neis <neis@chromium.org>
> Reviewed-by: Ulan Degenbaev <ulan@chromium.org>
> Cr-Commit-Position: refs/heads/master@{#70521}

Bug: v8:10315
Change-Id: I4c459fd6dfb98d47fc9941c0dc6864bf5a1d2d3e
Reviewed-on: https://chromium-review.googlesource.com/c/v8/v8/+/2474788Reviewed-by: Ulan Degenbaev <ulan@chromium.org>
Reviewed-by: Santiago Aboy Solanes <solanes@chromium.org>
Reviewed-by: Georg Neis <neis@chromium.org>
Reviewed-by: Leszek Swirski <leszeks@chromium.org>
Commit-Queue: Dominik Inführ <dinfuehr@chromium.org>
Cr-Commit-Position: refs/heads/master@{#70560}

src/codegen/compiler.cc

@@ -36,6 +36,7 @@
 #include "src/heap/heap-inl.h"
 #include "src/heap/local-factory-inl.h"
 #include "src/heap/local-heap-inl.h"
+#include "src/heap/local-heap.h"
 #include "src/init/bootstrapper.h"
 #include "src/interpreter/interpreter.h"
 #include "src/logging/log-inl.h"
@@ -340,12 +341,13 @@ CompilationJob::Status OptimizedCompilationJob::PrepareJob(Isolate* isolate) {
 }
 
 CompilationJob::Status OptimizedCompilationJob::ExecuteJob(
-    RuntimeCallStats* stats) {
+    RuntimeCallStats* stats, LocalIsolate* local_isolate) {
   DisallowHeapAccess no_heap_access;
   // Delegate to the underlying implementation.
   DCHECK_EQ(state(), State::kReadyToExecute);
   ScopedTimer t(&time_taken_to_execute_);
-  return UpdateState(ExecuteJobImpl(stats), State::kReadyToFinalize);
+  return UpdateState(ExecuteJobImpl(stats, local_isolate),
+                     State::kReadyToFinalize);
 }
 
 CompilationJob::Status OptimizedCompilationJob::FinalizeJob(Isolate* isolate) {
@@ -951,10 +953,21 @@ bool GetOptimizedCodeNow(OptimizedCompilationJob* job, Isolate* isolate,
   TRACE_EVENT0(TRACE_DISABLED_BY_DEFAULT("v8.compile"),
                "V8.OptimizeNonConcurrent");
 
-  if (!PrepareJobWithHandleScope(job, isolate, compilation_info) ||
-      job->ExecuteJob(isolate->counters()->runtime_call_stats()) !=
-          CompilationJob::SUCCEEDED ||
-      job->FinalizeJob(isolate) != CompilationJob::SUCCEEDED) {
+  if (!PrepareJobWithHandleScope(job, isolate, compilation_info)) {
+    CompilerTracer::TraceAbortedJob(isolate, compilation_info);
+    return false;
+  }
+
+  {
+    LocalIsolate local_isolate(isolate);
+    if (job->ExecuteJob(isolate->counters()->runtime_call_stats(),
+                        &local_isolate)) {
+      CompilerTracer::TraceAbortedJob(isolate, compilation_info);
+      return false;
+    }
+  }
+
+  if (job->FinalizeJob(isolate) != CompilationJob::SUCCEEDED) {
     CompilerTracer::TraceAbortedJob(isolate, compilation_info);
     return false;
   }
@@ -1560,6 +1573,7 @@ void BackgroundCompileTask::Run() {
     DCHECK(info_->flags().is_toplevel());
 
     LocalIsolate isolate(isolate_for_local_isolate_);
+    UnparkedScope unparked_scope(isolate.heap());
     LocalHandleScope handle_scope(&isolate);
 
     info_->ast_value_factory()->Internalize(&isolate);

src/codegen/compiler.h

@@ -332,7 +332,8 @@ class OptimizedCompilationJob : public CompilationJob {
 
   // Executes the compile job. Can be called on a background thread if
   // can_execute_on_background_thread() returns true.
-  V8_WARN_UNUSED_RESULT Status ExecuteJob(RuntimeCallStats* stats);
+  V8_WARN_UNUSED_RESULT Status
+  ExecuteJob(RuntimeCallStats* stats, LocalIsolate* local_isolate = nullptr);
 
   // Finalizes the compile job. Must be called on the main thread.
   V8_WARN_UNUSED_RESULT Status FinalizeJob(Isolate* isolate);
@@ -357,7 +358,8 @@ class OptimizedCompilationJob : public CompilationJob {
  protected:
   // Overridden by the actual implementation.
   virtual Status PrepareJobImpl(Isolate* isolate) = 0;
-  virtual Status ExecuteJobImpl(RuntimeCallStats* stats) = 0;
+  virtual Status ExecuteJobImpl(RuntimeCallStats* stats,
+                                LocalIsolate* local_heap) = 0;
   virtual Status FinalizeJobImpl(Isolate* isolate) = 0;
 
  private:

src/codegen/optimized-compilation-info.h

@@ -302,11 +302,8 @@ class V8_EXPORT_PRIVATE OptimizedCompilationInfo final {
   // 1) PersistentHandles created via PersistentHandlesScope inside of
   //    CompilationHandleScope
   // 2) Owned by OptimizedCompilationInfo
-  // 3) Owned by JSHeapBroker
-  // 4) Owned by the broker's LocalHeap
-  // 5) Back to the broker for a brief moment (after tearing down the
-  //   LocalHeap as part of exiting LocalHeapScope)
-  // 6) Back to OptimizedCompilationInfo when exiting the LocalHeapScope.
+  // 3) Owned by the broker's LocalHeap when entering the LocalHeapScope.
+  // 4) Back to OptimizedCompilationInfo when exiting the LocalHeapScope.
   //
   // In normal execution it gets destroyed when PipelineData gets destroyed.
   // There is a special case in GenerateCodeForTesting where the JSHeapBroker

src/compiler-dispatcher/optimizing-compile-dispatcher.cc

@@ -8,6 +8,8 @@
 #include "src/codegen/compiler.h"
 #include "src/codegen/optimized-compilation-info.h"
 #include "src/execution/isolate.h"
+#include "src/execution/local-isolate.h"
+#include "src/heap/local-heap.h"
 #include "src/init/v8.h"
 #include "src/logging/counters.h"
 #include "src/logging/log.h"
@@ -56,6 +58,7 @@ class OptimizingCompileDispatcher::CompileTask : public CancelableTask {
  private:
   // v8::Task overrides.
   void RunInternal() override {
+    LocalIsolate local_isolate(isolate_);
     DisallowHeapAllocation no_allocation;
     DisallowHandleAllocation no_handles;
     DisallowHandleDereference no_deref;
@@ -76,8 +79,8 @@ class OptimizingCompileDispatcher::CompileTask : public CancelableTask {
             dispatcher_->recompilation_delay_));
       }
 
-      dispatcher_->CompileNext(dispatcher_->NextInput(true),
-                               runtime_call_stats_scope.Get());
+      dispatcher_->CompileNext(dispatcher_->NextInput(&local_isolate, true),
+                               runtime_call_stats_scope.Get(), &local_isolate);
     }
     {
       base::MutexGuard lock_guard(&dispatcher_->ref_count_mutex_);
@@ -106,7 +109,7 @@ OptimizingCompileDispatcher::~OptimizingCompileDispatcher() {
 }
 
 OptimizedCompilationJob* OptimizingCompileDispatcher::NextInput(
-    bool check_if_flushing) {
+    LocalIsolate* local_isolate, bool check_if_flushing) {
   base::MutexGuard access_input_queue_(&input_queue_mutex_);
   if (input_queue_length_ == 0) return nullptr;
   OptimizedCompilationJob* job = input_queue_[InputQueueIndex(0)];
@@ -115,6 +118,7 @@ OptimizedCompilationJob* OptimizingCompileDispatcher::NextInput(
   input_queue_length_--;
   if (check_if_flushing) {
     if (mode_ == FLUSH) {
+      UnparkedScope scope(local_isolate->heap());
       AllowHandleDereference allow_handle_dereference;
       DisposeCompilationJob(job, true);
       return nullptr;
@@ -124,11 +128,12 @@ OptimizedCompilationJob* OptimizingCompileDispatcher::NextInput(
 }
 
 void OptimizingCompileDispatcher::CompileNext(OptimizedCompilationJob* job,
-                                              RuntimeCallStats* stats) {
+                                              RuntimeCallStats* stats,
+                                              LocalIsolate* local_isolate) {
   if (!job) return;
 
   // The function may have already been optimized by OSR.  Simply continue.
-  CompilationJob::Status status = job->ExecuteJob(stats);
+  CompilationJob::Status status = job->ExecuteJob(stats, local_isolate);
   USE(status);  // Prevent an unused-variable error.
 
   {

src/compiler-dispatcher/optimizing-compile-dispatcher.h

@@ -18,6 +18,7 @@
 namespace v8 {
 namespace internal {
 
+class LocalHeap;
 class OptimizedCompilationJob;
 class RuntimeCallStats;
 class SharedFunctionInfo;
@@ -58,8 +59,10 @@ class V8_EXPORT_PRIVATE OptimizingCompileDispatcher {
   enum ModeFlag { COMPILE, FLUSH };
 
   void FlushOutputQueue(bool restore_function_code);
-  void CompileNext(OptimizedCompilationJob* job, RuntimeCallStats* stats);
-  OptimizedCompilationJob* NextInput(bool check_if_flushing = false);
+  void CompileNext(OptimizedCompilationJob* job, RuntimeCallStats* stats,
+                   LocalIsolate* local_isolate);
+  OptimizedCompilationJob* NextInput(LocalIsolate* local_isolate,
+                                     bool check_if_flushing = false);
 
   inline int InputQueueIndex(int i) {
     int result = (i + input_queue_shift_) % input_queue_capacity_;

src/compiler/js-heap-broker.cc

@@ -2427,7 +2427,6 @@ JSHeapBroker::JSHeapBroker(Isolate* isolate, Zone* broker_zone,
       tracing_enabled_(tracing_enabled),
       is_concurrent_inlining_(is_concurrent_inlining),
       code_kind_(code_kind),
-      local_heap_(base::nullopt),
       feedback_(zone()),
       bytecode_analyses_(zone()),
       property_access_infos_(zone()),
@@ -2472,20 +2471,23 @@ std::string JSHeapBroker::Trace() const {
   return oss.str();
 }
 
-void JSHeapBroker::InitializeLocalHeap(OptimizedCompilationInfo* info) {
-  set_persistent_handles(info->DetachPersistentHandles());
+void JSHeapBroker::InitializeLocalHeap(OptimizedCompilationInfo* info,
+                                       LocalHeap* local_heap) {
   set_canonical_handles(info->DetachCanonicalHandles());
-  DCHECK(!local_heap_);
-  local_heap_.emplace(isolate_->heap(), std::move(ph_));
+  DCHECK_NULL(local_heap_);
+  local_heap_ = local_heap;
+  DCHECK_NOT_NULL(local_heap_);
+  local_heap_->AttachPersistentHandles(info->DetachPersistentHandles());
 }
 
 void JSHeapBroker::TearDownLocalHeap(OptimizedCompilationInfo* info) {
   DCHECK_NULL(ph_);
   DCHECK(local_heap_);
-  ph_ = local_heap_->DetachPersistentHandles();
-  local_heap_.reset();
+  std::unique_ptr<PersistentHandles> ph =
+      local_heap_->DetachPersistentHandles();
+  local_heap_ = nullptr;
   info->set_canonical_handles(DetachCanonicalHandles());
-  info->set_persistent_handles(DetachPersistentHandles());
+  info->set_persistent_handles(std::move(ph));
 }
 
 void JSHeapBroker::StopSerializing() {

src/compiler/js-heap-broker.h

@@ -117,7 +117,8 @@ class V8_EXPORT_PRIVATE JSHeapBroker {
   BrokerMode mode() const { return mode_; }
   // Initialize the local heap with the persistent and canonical handles
   // provided by {info}.
-  void InitializeLocalHeap(OptimizedCompilationInfo* info);
+  void InitializeLocalHeap(OptimizedCompilationInfo* info,
+                           LocalHeap* local_heap);
   // Tear down the local heap and pass the persistent and canonical handles
   // provided back to {info}. {info} is responsible for disposing of them.
   void TearDownLocalHeap(OptimizedCompilationInfo* info);
@@ -227,9 +228,7 @@ class V8_EXPORT_PRIVATE JSHeapBroker {
   bool IsSerializedForCompilation(const SharedFunctionInfoRef& shared,
                                   const FeedbackVectorRef& feedback) const;
 
-  LocalHeap* local_heap() {
-    return local_heap_.has_value() ? &(*local_heap_) : nullptr;
-  }
+  LocalHeap* local_heap() { return local_heap_; }
 
   // Return the corresponding canonical persistent handle for {object}. Create
   // one if it does not exist.
@@ -361,7 +360,7 @@ class V8_EXPORT_PRIVATE JSHeapBroker {
   bool const is_concurrent_inlining_;
   CodeKind const code_kind_;
   std::unique_ptr<PersistentHandles> ph_;
-  base::Optional<LocalHeap> local_heap_;
+  LocalHeap* local_heap_ = nullptr;
   std::unique_ptr<CanonicalHandlesMap> canonical_handles_;
   unsigned trace_indentation_ = 0;
   PerIsolateCompilerCache* compiler_cache_ = nullptr;
@@ -453,14 +452,16 @@ class OffHeapBytecodeArray final : public interpreter::AbstractBytecodeArray {
 
 // Scope that unparks the LocalHeap, if:
 //   a) We have a JSHeapBroker,
-//   b) Said JSHeapBroker has a LocalHeap, and
-//   c) Said LocalHeap has been parked.
+//   b) Said JSHeapBroker has a LocalHeap,
+//   c) Said LocalHeap has been parked and
+//   d) The given condition evaluates to true.
 // Used, for example, when printing the graph with --trace-turbo with a
 // previously parked LocalHeap.
 class UnparkedScopeIfNeeded {
  public:
-  explicit UnparkedScopeIfNeeded(JSHeapBroker* broker) {
-    if (broker != nullptr) {
+  explicit UnparkedScopeIfNeeded(JSHeapBroker* broker,
+                                 bool extra_condition = true) {
+    if (broker != nullptr && extra_condition) {
       LocalHeap* local_heap = broker->local_heap();
       if (local_heap != nullptr && local_heap->IsParked()) {
         unparked_scope.emplace(local_heap);

src/compiler/pipeline.cc

@@ -83,6 +83,7 @@
 #include "src/diagnostics/code-tracer.h"
 #include "src/diagnostics/disassembler.h"
 #include "src/execution/isolate-inl.h"
+#include "src/heap/local-heap.h"
 #include "src/init/bootstrapper.h"
 #include "src/logging/counters.h"
 #include "src/objects/shared-function-info.h"
@@ -765,12 +766,14 @@ class PipelineRunScope {
   RuntimeCallTimerScope runtime_call_timer_scope;
 };
 
-// LocalHeapScope encapsulates the liveness of the brokers's LocalHeap.
+// LocalHeapScope encapsulates the phase where persistent handles are attached
+// to the LocalHeap.
 class LocalHeapScope {
  public:
-  explicit LocalHeapScope(JSHeapBroker* broker, OptimizedCompilationInfo* info)
+  explicit LocalHeapScope(JSHeapBroker* broker, OptimizedCompilationInfo* info,
+                          LocalHeap* local_heap)
       : broker_(broker), info_(info) {
-    broker_->InitializeLocalHeap(info_);
+    broker_->InitializeLocalHeap(info_, local_heap);
     info_->tick_counter().AttachLocalHeap(broker_->local_heap());
   }
 
@@ -1030,7 +1033,8 @@ class PipelineCompilationJob final : public OptimizedCompilationJob {
 
  protected:
   Status PrepareJobImpl(Isolate* isolate) final;
-  Status ExecuteJobImpl(RuntimeCallStats* stats) final;
+  Status ExecuteJobImpl(RuntimeCallStats* stats,
+                        LocalIsolate* local_isolate) final;
   Status FinalizeJobImpl(Isolate* isolate) final;
 
   // Registers weak object to optimized code dependencies.
@@ -1177,12 +1181,13 @@ PipelineCompilationJob::Status PipelineCompilationJob::PrepareJobImpl(
 }
 
 PipelineCompilationJob::Status PipelineCompilationJob::ExecuteJobImpl(
-    RuntimeCallStats* stats) {
+    RuntimeCallStats* stats, LocalIsolate* local_isolate) {
   // Ensure that the RuntimeCallStats table is only available during execution
   // and not during finalization as that might be on a different thread.
   PipelineJobScope scope(&data_, stats);
-  {
-    LocalHeapScope local_heap_scope(data_.broker(), data_.info());
+  LocalHeapScope local_heap_scope(data_.broker(), data_.info(),
+                                  local_isolate->heap());
+
   if (data_.broker()->is_concurrent_inlining()) {
     if (!pipeline_.CreateGraph()) {
       return AbortOptimization(BailoutReason::kGraphBuildingFailed);
@@ -1190,8 +1195,6 @@ PipelineCompilationJob::Status PipelineCompilationJob::ExecuteJobImpl(
   }
 
   // We selectively Unpark inside OptimizeGraph*.
-    ParkedScope parked_scope(data_.broker()->local_heap());
-
   bool success;
   if (compilation_info_.code_kind() == CodeKind::TURBOPROP) {
     success = pipeline_.OptimizeGraphForMidTier(linkage_);
@@ -1199,7 +1202,6 @@ PipelineCompilationJob::Status PipelineCompilationJob::ExecuteJobImpl(
     success = pipeline_.OptimizeGraph(linkage_);
   }
   if (!success) return FAILED;
-  }
 
   pipeline_.AssembleCode(linkage_);
 
@@ -1283,7 +1285,8 @@ class WasmHeapStubCompilationJob final : public OptimizedCompilationJob {
 
  protected:
   Status PrepareJobImpl(Isolate* isolate) final;
-  Status ExecuteJobImpl(RuntimeCallStats* stats) final;
+  Status ExecuteJobImpl(RuntimeCallStats* stats,
+                        LocalIsolate* local_isolate) final;
   Status FinalizeJobImpl(Isolate* isolate) final;
 
  private:
@@ -1318,7 +1321,7 @@ CompilationJob::Status WasmHeapStubCompilationJob::PrepareJobImpl(
 }
 
 CompilationJob::Status WasmHeapStubCompilationJob::ExecuteJobImpl(
-    RuntimeCallStats* stats) {
+    RuntimeCallStats* stats, LocalIsolate* local_isolate) {
   std::unique_ptr<PipelineStatistics> pipeline_statistics;
   if (FLAG_turbo_stats || FLAG_turbo_stats_nvp) {
     pipeline_statistics.reset(new PipelineStatistics(
@@ -2471,6 +2474,7 @@ void PipelineImpl::Serialize() {
 
 bool PipelineImpl::CreateGraph() {
   PipelineData* data = this->data_;
+  UnparkedScopeIfNeeded unparked_scope(data->broker());
 
   data->BeginPhaseKind("V8.TFGraphCreation");
 
@@ -3022,13 +3026,15 @@ MaybeHandle<Code> Pipeline::GenerateCodeForTesting(
   }
 
   {
-    LocalHeapScope local_heap_scope(data.broker(), info);
+    LocalIsolate local_isolate(isolate);
+    LocalHeapScope local_heap_scope(data.broker(), info, local_isolate.heap());
     if (data.broker()->is_concurrent_inlining()) {
       if (!pipeline.CreateGraph()) return MaybeHandle<Code>();
     }
     // We selectively Unpark inside OptimizeGraph.
-    ParkedScope parked_scope(data.broker()->local_heap());
     if (!pipeline.OptimizeGraph(&linkage)) return MaybeHandle<Code>();
+
+    pipeline.AssembleCode(&linkage);
   }
 
   const bool will_retire_broker = out_broker == nullptr;
@@ -3040,7 +3046,6 @@ MaybeHandle<Code> Pipeline::GenerateCodeForTesting(
         info->DetachPersistentHandles(), info->DetachCanonicalHandles());
   }
 
-  pipeline.AssembleCode(&linkage);
   Handle<Code> code;
   if (pipeline.FinalizeCode(will_retire_broker).ToHandle(&code) &&
       pipeline.CommitDependencies(code)) {
@@ -3478,6 +3483,8 @@ void PipelineImpl::AssembleCode(Linkage* linkage,
   data->BeginPhaseKind("V8.TFCodeGeneration");
   data->InitializeCodeGenerator(linkage, std::move(buffer));
 
+  UnparkedScopeIfNeeded unparked_scope(data->broker(), FLAG_code_comments);
+
   Run<AssembleCodePhase>();
   if (data->info()->trace_turbo_json()) {
     TurboJsonFile json_of(data->info(), std::ios_base::app);

src/compiler/wasm-compiler.cc

@@ -7657,7 +7657,7 @@ Handle<Code> CompileCWasmEntry(Isolate* isolate, const wasm::FunctionSig* sig,
           CodeKind::C_WASM_ENTRY, std::move(name_buffer),
           AssemblerOptions::Default(isolate)));
 
-  CHECK_NE(job->ExecuteJob(isolate->counters()->runtime_call_stats()),
+  CHECK_NE(job->ExecuteJob(isolate->counters()->runtime_call_stats(), nullptr),
            CompilationJob::FAILED);
   CHECK_NE(job->FinalizeJob(isolate), CompilationJob::FAILED);
 

src/heap/concurrent-allocator.cc

@@ -9,6 +9,7 @@
 #include "src/handles/persistent-handles.h"
 #include "src/heap/concurrent-allocator-inl.h"
 #include "src/heap/local-heap-inl.h"
+#include "src/heap/local-heap.h"
 #include "src/heap/marking.h"
 #include "src/heap/memory-chunk.h"
 
@@ -18,6 +19,7 @@ namespace internal {
 void StressConcurrentAllocatorTask::RunInternal() {
   Heap* heap = isolate_->heap();
   LocalHeap local_heap(heap);
+  UnparkedScope unparked_scope(&local_heap);
 
   const int kNumIterations = 2000;
   const int kSmallObjectSize = 10 * kTaggedSize;

src/heap/local-heap.cc

@@ -27,7 +27,7 @@ LocalHeap* LocalHeap::Current() { return current_local_heap; }
 LocalHeap::LocalHeap(Heap* heap,
                      std::unique_ptr<PersistentHandles> persistent_handles)
     : heap_(heap),
-      state_(ThreadState::Running),
+      state_(ThreadState::Parked),
       safepoint_requested_(false),
       allocation_failed_(false),
       prev_(nullptr),
@@ -36,34 +36,35 @@ LocalHeap::LocalHeap(Heap* heap,
       persistent_handles_(std::move(persistent_handles)),
       marking_barrier_(new MarkingBarrier(this)),
       old_space_allocator_(this, heap->old_space()) {
-  heap_->safepoint()->AddLocalHeap(this);
-  if (persistent_handles_) {
-    persistent_handles_->Attach(this);
-  }
-  DCHECK_NULL(current_local_heap);
-  current_local_heap = this;
-  // TODO(ulan): Ensure that LocalHeap cannot be created without --local-heaps.
+  heap_->safepoint()->AddLocalHeap(this, [this] {
     if (FLAG_local_heaps) {
       WriteBarrier::SetForThread(marking_barrier_.get());
       if (heap_->incremental_marking()->IsMarking()) {
-      marking_barrier_->Activate(heap_->incremental_marking()->IsCompacting());
+        marking_barrier_->Activate(
+            heap_->incremental_marking()->IsCompacting());
+      }
     }
+  });
+
+  if (persistent_handles_) {
+    persistent_handles_->Attach(this);
   }
+  DCHECK_NULL(current_local_heap);
+  current_local_heap = this;
 }
 
 LocalHeap::~LocalHeap() {
-  // TODO(ulan): Ensure that LocalHeap cannot be created without --local-heaps.
+  // Park thread since removing the local heap could block.
+  EnsureParkedBeforeDestruction();
+
+  heap_->safepoint()->RemoveLocalHeap(this, [this] {
+    old_space_allocator_.FreeLinearAllocationArea();
+
     if (FLAG_local_heaps) {
       marking_barrier_->Publish();
       WriteBarrier::ClearForThread(marking_barrier_.get());
     }
-  // Give up LAB before parking thread
-  old_space_allocator_.FreeLinearAllocationArea();
-
-  // Park thread since removing the local heap could block.
-  EnsureParkedBeforeDestruction();
-
-  heap_->safepoint()->RemoveLocalHeap(this);
+  });
 
   DCHECK_EQ(current_local_heap, this);
   current_local_heap = nullptr;
@@ -77,6 +78,13 @@ void LocalHeap::EnsurePersistentHandles() {
   }
 }
 
+void LocalHeap::AttachPersistentHandles(
+    std::unique_ptr<PersistentHandles> persistent_handles) {
+  DCHECK_NULL(persistent_handles_);
+  persistent_handles_ = std::move(persistent_handles);
+  persistent_handles_->Attach(this);
+}
+
 std::unique_ptr<PersistentHandles> LocalHeap::DetachPersistentHandles() {
   if (persistent_handles_) persistent_handles_->Detach();
   return std::move(persistent_handles_);
@@ -116,6 +124,7 @@ void LocalHeap::Unpark() {
 }
 
 void LocalHeap::EnsureParkedBeforeDestruction() {
+  if (IsParked()) return;
   base::MutexGuard guard(&state_mutex_);
   state_ = ThreadState::Parked;
   state_change_.NotifyAll();

src/heap/local-heap.h

@@ -22,6 +22,15 @@ class Heap;
 class Safepoint;
 class LocalHandles;
 
+// LocalHeap is used by the GC to track all threads with heap access in order to
+// stop them before performing a collection. LocalHeaps can be either Parked or
+// Running and are in Parked mode when initialized.
+//   Running: Thread is allowed to access the heap but needs to give the GC the
+//            chance to run regularly by manually invoking Safepoint(). The
+//            thread can be parked using ParkedScope.
+//   Parked:  Heap access is not allowed, so the GC will not stop this thread
+//            for a collection. Useful when threads do not need heap access for
+//            some time or for blocking operations like locking a mutex.
 class V8_EXPORT_PRIVATE LocalHeap {
  public:
   explicit LocalHeap(
@@ -70,6 +79,8 @@ class V8_EXPORT_PRIVATE LocalHeap {
     return kNullMaybeHandle;
   }
 
+  void AttachPersistentHandles(
+      std::unique_ptr<PersistentHandles> persistent_handles);
   std::unique_ptr<PersistentHandles> DetachPersistentHandles();
 #ifdef DEBUG
   bool ContainsPersistentHandle(Address* location);

src/heap/safepoint.cc

@@ -114,23 +114,6 @@ SafepointScope::SafepointScope(Heap* heap) : safepoint_(heap->safepoint()) {
 
 SafepointScope::~SafepointScope() { safepoint_->LeaveSafepointScope(); }
 
-void GlobalSafepoint::AddLocalHeap(LocalHeap* local_heap) {
-  base::MutexGuard guard(&local_heaps_mutex_);
-  if (local_heaps_head_) local_heaps_head_->prev_ = local_heap;
-  local_heap->prev_ = nullptr;
-  local_heap->next_ = local_heaps_head_;
-  local_heaps_head_ = local_heap;
-}
-
-void GlobalSafepoint::RemoveLocalHeap(LocalHeap* local_heap) {
-  base::MutexGuard guard(&local_heaps_mutex_);
-  if (local_heap->next_) local_heap->next_->prev_ = local_heap->prev_;
-  if (local_heap->prev_)
-    local_heap->prev_->next_ = local_heap->next_;
-  else
-    local_heaps_head_ = local_heap->next_;
-}
-
 bool GlobalSafepoint::ContainsLocalHeap(LocalHeap* local_heap) {
   base::MutexGuard guard(&local_heaps_mutex_);
   LocalHeap* current = local_heaps_head_;

src/heap/safepoint.h

@@ -62,8 +62,36 @@ class GlobalSafepoint {
   void EnterSafepointScope();
   void LeaveSafepointScope();
 
-  void AddLocalHeap(LocalHeap* local_heap);
-  void RemoveLocalHeap(LocalHeap* local_heap);
+  template <typename Callback>
+  void AddLocalHeap(LocalHeap* local_heap, Callback callback) {
+    // Safepoint holds this lock in order to stop threads from starting or
+    // stopping.
+    base::MutexGuard guard(&local_heaps_mutex_);
+
+    // Additional code protected from safepoint
+    callback();
+
+    // Add list to doubly-linked list
+    if (local_heaps_head_) local_heaps_head_->prev_ = local_heap;
+    local_heap->prev_ = nullptr;
+    local_heap->next_ = local_heaps_head_;
+    local_heaps_head_ = local_heap;
+  }
+
+  template <typename Callback>
+  void RemoveLocalHeap(LocalHeap* local_heap, Callback callback) {
+    base::MutexGuard guard(&local_heaps_mutex_);
+
+    // Additional code protected from safepoint
+    callback();
+
+    // Remove list from doubly-linked list
+    if (local_heap->next_) local_heap->next_->prev_ = local_heap->prev_;
+    if (local_heap->prev_)
+      local_heap->prev_->next_ = local_heap->next_;
+    else
+      local_heaps_head_ = local_heap->next_;
+  }
 
   Barrier barrier_;
   Heap* heap_;

test/cctest/heap/test-concurrent-allocation.cc

@@ -52,6 +52,7 @@ class ConcurrentAllocationThread final : public v8::base::Thread {
 
   void Run() override {
     LocalHeap local_heap(heap_);
+    UnparkedScope unparked_scope(&local_heap);
 
     for (int i = 0; i < kNumIterations; i++) {
       Address address = local_heap.AllocateRawOrFail(
@@ -119,6 +120,7 @@ class LargeObjectConcurrentAllocationThread final : public v8::base::Thread {
 
   void Run() override {
     LocalHeap local_heap(heap_);
+    UnparkedScope unparked_scope(&local_heap);
     const size_t kLargeObjectSize = kMaxRegularHeapObjectSize * 2;
 
     for (int i = 0; i < kNumIterations; i++) {
@@ -186,6 +188,7 @@ class ConcurrentBlackAllocationThread final : public v8::base::Thread {
 
   void Run() override {
     LocalHeap local_heap(heap_);
+    UnparkedScope unparked_scope(&local_heap);
 
     for (int i = 0; i < kNumIterations; i++) {
       if (i == kWhiteIterations) {
@@ -265,6 +268,7 @@ class ConcurrentWriteBarrierThread final : public v8::base::Thread {
 
   void Run() override {
     LocalHeap local_heap(heap_);
+    UnparkedScope unparked_scope(&local_heap);
     fixed_array_.set(0, value_);
   }
 
@@ -326,6 +330,7 @@ class ConcurrentRecordRelocSlotThread final : public v8::base::Thread {
 
   void Run() override {
     LocalHeap local_heap(heap_);
+    UnparkedScope unparked_scope(&local_heap);
     int mode_mask = RelocInfo::EmbeddedObjectModeMask();
     for (RelocIterator it(code_, mode_mask); !it.done(); it.next()) {
       DCHECK(RelocInfo::IsEmbeddedObjectMode(it.rinfo()->rmode()));

test/cctest/heap/test-heap.cc

@@ -7184,6 +7184,7 @@ TEST(GarbageCollectionWithLocalHeap) {
   Heap* heap = CcTest::i_isolate()->heap();
 
   LocalHeap local_heap(heap);
+  UnparkedScope unparked_scope(&local_heap);
   CcTest::CollectGarbage(OLD_SPACE);
   { ParkedScope parked_scope(&local_heap); }
   CcTest::CollectGarbage(OLD_SPACE);

test/cctest/test-concurrent-descriptor-array.cc

@@ -34,6 +34,7 @@ class ConcurrentSearchThread final : public v8::base::Thread {
 
   void Run() override {
     LocalHeap local_heap(heap_, std::move(ph_));
+    UnparkedScope unparked_scope(&local_heap);
     LocalHandleScope scope(&local_heap);
 
     for (int i = 0; i < kNumHandles; i++) {

test/cctest/test-concurrent-prototype.cc

@@ -33,6 +33,7 @@ class ConcurrentSearchThread final : public v8::base::Thread {
 
   void Run() override {
     LocalHeap local_heap(heap_, std::move(ph_));
+    UnparkedScope unparked_scope(&local_heap);
     LocalHandleScope scope(&local_heap);
 
     for (int i = 0; i < kNumHandles; i++) {

test/cctest/test-concurrent-script-context-table.cc

@@ -35,6 +35,7 @@ class ScriptContextTableAccessUsedThread final : public v8::base::Thread {
 
   void Run() override {
     LocalHeap local_heap(heap_, std::move(ph_));
+    UnparkedScope unparked_scope(&local_heap);
     LocalHandleScope scope(&local_heap);
 
     sema_started_->Signal();
@@ -67,6 +68,7 @@ class AccessScriptContextTableThread final : public v8::base::Thread {
 
   void Run() override {
     LocalHeap local_heap(heap_, std::move(ph_));
+    UnparkedScope unparked_scope(&local_heap);
     LocalHandleScope scope(&local_heap);
 
     sema_started_->Signal();

test/cctest/test-concurrent-transition-array.cc

@@ -35,6 +35,7 @@ class ConcurrentSearchThread : public v8::base::Thread {
 
   void Run() override {
     LocalHeap local_heap(heap_, std::move(ph_));
+    UnparkedScope scope(&local_heap);
 
     background_thread_started_->Signal();
 
@@ -70,6 +71,7 @@ class ConcurrentSearchOnOutdatedAccessorThread final
 
   void Run() override {
     LocalHeap local_heap(heap_, std::move(ph_));
+    UnparkedScope scope(&local_heap);
 
     TransitionsAccessor accessor(CcTest::i_isolate(), map_, true);
     background_thread_started_->Signal();

test/cctest/test-local-handles.cc

@@ -36,6 +36,7 @@ class LocalHandlesThread final : public v8::base::Thread {
 
   void Run() override {
     LocalHeap local_heap(heap_);
+    UnparkedScope unparked_scope(&local_heap);
     LocalHandleScope scope(&local_heap);
 
     static constexpr int kNumHandles =
@@ -103,6 +104,7 @@ TEST(CreateLocalHandlesWithoutLocalHandleScope) {
 
   {
     LocalHeap local_heap(isolate->heap());
+    UnparkedScope scope(&local_heap);
     handle(Smi::FromInt(17), &local_heap);
   }
 }
@@ -123,6 +125,7 @@ TEST(DereferenceLocalHandle) {
   }
   {
     LocalHeap local_heap(isolate->heap(), std::move(phs));
+    UnparkedScope unparked_scope(&local_heap);
     LocalHandleScope scope(&local_heap);
     Handle<HeapNumber> local_number = handle(*ph, &local_heap);
     CHECK_EQ(42, local_number->value());

test/cctest/test-persistent-handles.cc

@@ -42,6 +42,7 @@ class PersistentHandlesThread final : public v8::base::Thread {
 
   void Run() override {
     LocalHeap local_heap(heap_, std::move(ph_));
+    UnparkedScope unparked_scope(&local_heap);
     LocalHandleScope scope(&local_heap);
 
     for (int i = 0; i < kNumHandles; i++) {
@@ -129,6 +130,7 @@ TEST(DereferencePersistentHandle) {
   }
   {
     LocalHeap local_heap(isolate->heap(), std::move(phs));
+    UnparkedScope scope(&local_heap);
     CHECK_EQ(42, ph->value());
     DisallowHandleDereference disallow_scope;
     CHECK_EQ(42, ph->value());
@@ -140,7 +142,6 @@ TEST(NewPersistentHandleFailsWhenParked) {
   Isolate* isolate = CcTest::i_isolate();
 
   LocalHeap local_heap(isolate->heap());
-  ParkedScope scope(&local_heap);
   // Fail here in debug mode: Persistent handles can't be created if local heap
   // is parked
   local_heap.NewPersistentHandle(Smi::FromInt(1));
@@ -151,7 +152,6 @@ TEST(NewPersistentHandleFailsWhenParkedExplicit) {
   Isolate* isolate = CcTest::i_isolate();
 
   LocalHeap local_heap(isolate->heap(), isolate->NewPersistentHandles());
-  ParkedScope scope(&local_heap);
   // Fail here in debug mode: Persistent handles can't be created if local heap
   // is parked
   local_heap.NewPersistentHandle(Smi::FromInt(1));

test/unittests/compiler-dispatcher/optimizing-compile-dispatcher-unittest.cc

@@ -10,7 +10,9 @@
 #include "src/codegen/compiler.h"
 #include "src/codegen/optimized-compilation-info.h"
 #include "src/execution/isolate.h"
+#include "src/execution/local-isolate.h"
 #include "src/handles/handles.h"
+#include "src/heap/local-heap.h"
 #include "src/objects/objects-inl.h"
 #include "src/parsing/parse-info.h"
 #include "test/unittests/test-helpers.h"
@@ -42,7 +44,8 @@ class BlockingCompilationJob : public OptimizedCompilationJob {
   // OptimiziedCompilationJob implementation.
   Status PrepareJobImpl(Isolate* isolate) override { UNREACHABLE(); }
 
-  Status ExecuteJobImpl(RuntimeCallStats* stats) override {
+  Status ExecuteJobImpl(RuntimeCallStats* stats,
+                        LocalIsolate* local_isolate) override {
     blocking_.SetValue(true);
     semaphore_.Wait();
     blocking_.SetValue(false);

test/unittests/heap/local-factory-unittest.cc

@@ -62,7 +62,8 @@ class LocalFactoryTest : public TestWithIsolateAndZone {
                 isolate(), true, construct_language_mode(FLAG_use_strict),
                 REPLMode::kNo),
             &state_),
-        local_isolate_(isolate()) {
+        local_isolate_(isolate()),
+        unparked_scope_(local_isolate_.heap()) {
     FLAG_concurrent_allocation = true;
   }
 
@@ -114,6 +115,7 @@ class LocalFactoryTest : public TestWithIsolateAndZone {
   UnoptimizedCompileState state_;
   ParseInfo parse_info_;
   LocalIsolate local_isolate_;
+  UnparkedScope unparked_scope_;
   Handle<String> source_string_;
   Handle<Script> script_;
 };

test/unittests/heap/safepoint-unittest.cc

@@ -43,7 +43,6 @@ class ParkedThread final : public v8::base::Thread {
     LocalHeap local_heap(heap_);
 
     if (mutex_) {
-      ParkedScope scope(&local_heap);
       base::MutexGuard guard(mutex_);
     }
   }
@@ -100,6 +99,7 @@ class RunningThread final : public v8::base::Thread {
 
   void Run() override {
     LocalHeap local_heap(heap_);
+    UnparkedScope unparked_scope(&local_heap);
 
     for (int i = 0; i < kRuns; i++) {
       counter_->fetch_add(1);
@@ -148,6 +148,7 @@ TEST_F(SafepointTest, SkipLocalHeapOfThisThread) {
   EnsureFlagLocalHeapsEnabled();
   Heap* heap = i_isolate()->heap();
   LocalHeap local_heap(heap);
+  UnparkedScope unparked_scope(&local_heap);
   {
     SafepointScope scope(heap);
     local_heap.Safepoint();