[wasm] Remove finisher task

This removes the finisher task and instead finishes compilation units
from the background.
It also changes ownership of the AsyncCompileJob to be shared among all
tasks that still operate on it. The AsyncCompileJob dies when the last
reference dies.

R=ahaas@chromium.org
CC=​​mstarzinger@chromium.org

Bug: v8:7921, v8:8423
Change-Id: Id09378327dfc146459ef41bc97176a8716756ae4
Cq-Include-Trybots: luci.v8.try:v8_linux64_tsan_rel
Reviewed-on: https://chromium-review.googlesource.com/c/1335553Reviewed-by: Andreas Haas <ahaas@chromium.org>
Commit-Queue: Clemens Hammacher <clemensh@chromium.org>
Cr-Commit-Position: refs/heads/master@{#58630}

src/api.cc

@@ -8629,7 +8629,7 @@ int Isolate::ContextDisposedNotification(bool dependant_context) {
   if (!dependant_context) {
     // We left the current context, we can abort all WebAssembly compilations on
     // that isolate.
-    isolate->wasm_engine()->DeleteCompileJobsOnIsolate(isolate);
+    isolate->wasm_engine()->AbortCompileJobsOnIsolate(isolate);
   }
   // TODO(ahaas): move other non-heap activity out of the heap call.
   return isolate->heap()->NotifyContextDisposed(dependant_context);

src/isolate.cc

@@ -2839,7 +2839,7 @@ void Isolate::Deinit() {
 
   debug()->Unload();
 
-  wasm_engine()->DeleteCompileJobsOnIsolate(this);
+  wasm_engine()->AbortCompileJobsOnIsolate(this);
 
   if (concurrent_recompilation_enabled()) {
     optimizing_compile_dispatcher_->Stop();

src/wasm/compilation-environment.h

@@ -93,7 +93,7 @@ class CompilationState {
   using callback_t = std::function<void(CompilationEvent, const ResultBase*)>;
   ~CompilationState();
 
-  void CancelAndWait();
+  void AbortCompilation();
 
   void SetError(uint32_t func_index, const ResultBase& error_result);
 

src/wasm/module-compiler.h

@@ -65,8 +65,11 @@ Address CompileLazy(Isolate*, NativeModule*, uint32_t func_index);
 // allocates on the V8 heap (e.g. creating the module object) must be a
 // foreground task. All other tasks (e.g. decoding and validating, the majority
 // of the work of compilation) can be background tasks.
+// AsyncCompileJobs are stored in shared_ptr by all tasks which need to keep
+// them alive. {std::enable_shared_from_this} allows to regain a shared_ptr from
+// a raw ptr.
 // TODO(wasm): factor out common parts of this with the synchronous pipeline.
-class AsyncCompileJob {
+class AsyncCompileJob : public std::enable_shared_from_this<AsyncCompileJob> {
  public:
   AsyncCompileJob(Isolate* isolate, const WasmFeatures& enabled_features,
                   std::unique_ptr<byte[]> bytes_copy, size_t length,
@@ -92,7 +95,6 @@ class AsyncCompileJob {
   class DecodeModule;            // Step 1  (async)
   class DecodeFail;              // Step 1b (sync)
   class PrepareAndStartCompile;  // Step 2  (sync)
-  class CompileFailed;           // Step 4b (sync)
   class CompileWrappers;         // Step 5  (sync)
   class FinishModule;            // Step 6  (sync)
 

src/wasm/streaming-decoder.cc

@@ -103,8 +103,10 @@ void StreamingDecoder::Finish() {
 void StreamingDecoder::Abort() {
   TRACE_STREAMING("Abort\n");
   if (!ok()) return;  // Failed already.
-  processor_->OnAbort();
-  Fail();
+  // Move the processor out of the unique_ptr field first, to avoid recursive
+  // calls to {OnAbort}.
+  std::unique_ptr<StreamingProcessor> processor = std::move(processor_);
+  processor->OnAbort();
 }
 
 void StreamingDecoder::SetModuleCompiledCallback(

src/wasm/wasm-code-manager.cc

@@ -869,7 +869,7 @@ NativeModule::~NativeModule() {
   TRACE_HEAP("Deleting native module: %p\n", reinterpret_cast<void*>(this));
   // Cancel all background compilation before resetting any field of the
   // NativeModule or freeing anything.
-  compilation_state_->CancelAndWait();
+  compilation_state_->AbortCompilation();
   code_manager_->FreeNativeModule(this);
 }
 

src/wasm/wasm-engine.cc

@@ -24,7 +24,7 @@ WasmEngine::WasmEngine()
 
 WasmEngine::~WasmEngine() {
   // All AsyncCompileJobs have been canceled.
-  DCHECK(jobs_.empty());
+  DCHECK(async_compile_jobs_.empty());
   // All Isolates have been deregistered.
   DCHECK(isolates_.empty());
 }
@@ -212,7 +212,7 @@ void WasmEngine::AsyncCompile(
   std::unique_ptr<byte[]> copy(new byte[bytes.length()]);
   memcpy(copy.get(), bytes.start(), bytes.length());
 
-  AsyncCompileJob* job = CreateAsyncCompileJob(
+  std::shared_ptr<AsyncCompileJob> job = CreateAsyncCompileJob(
       isolate, enabled, std::move(copy), bytes.length(),
       handle(isolate->context(), isolate), std::move(resolver));
   job->Start();
@@ -221,7 +221,7 @@ void WasmEngine::AsyncCompile(
 std::shared_ptr<StreamingDecoder> WasmEngine::StartStreamingCompilation(
     Isolate* isolate, const WasmFeatures& enabled, Handle<Context> context,
     std::shared_ptr<CompilationResultResolver> resolver) {
-  AsyncCompileJob* job =
+  std::shared_ptr<AsyncCompileJob> job =
       CreateAsyncCompileJob(isolate, enabled, std::unique_ptr<byte[]>(nullptr),
                             0, context, std::move(resolver));
   return job->CreateStreamingDecoder();
@@ -278,48 +278,49 @@ CodeTracer* WasmEngine::GetCodeTracer() {
   return code_tracer_.get();
 }
 
-AsyncCompileJob* WasmEngine::CreateAsyncCompileJob(
+std::shared_ptr<AsyncCompileJob> WasmEngine::CreateAsyncCompileJob(
     Isolate* isolate, const WasmFeatures& enabled,
     std::unique_ptr<byte[]> bytes_copy, size_t length, Handle<Context> context,
     std::shared_ptr<CompilationResultResolver> resolver) {
-  AsyncCompileJob* job =
-      new AsyncCompileJob(isolate, enabled, std::move(bytes_copy), length,
-                          context, std::move(resolver));
-  // Pass ownership to the unique_ptr in {jobs_}.
+  std::shared_ptr<AsyncCompileJob> job =
+      std::make_shared<AsyncCompileJob>(isolate, enabled, std::move(bytes_copy),
+                                        length, context, std::move(resolver));
   base::MutexGuard guard(&mutex_);
-  jobs_[job] = std::unique_ptr<AsyncCompileJob>(job);
+  async_compile_jobs_.insert({job.get(), job});
   return job;
 }
 
-std::unique_ptr<AsyncCompileJob> WasmEngine::RemoveCompileJob(
-    AsyncCompileJob* job) {
+void WasmEngine::RemoveCompileJob(AsyncCompileJob* job) {
   base::MutexGuard guard(&mutex_);
-  auto item = jobs_.find(job);
-  DCHECK(item != jobs_.end());
-  std::unique_ptr<AsyncCompileJob> result = std::move(item->second);
-  jobs_.erase(item);
-  return result;
+  auto item = async_compile_jobs_.find(job);
+  DCHECK(item != async_compile_jobs_.end());
+  async_compile_jobs_.erase(item);
 }
 
 bool WasmEngine::HasRunningCompileJob(Isolate* isolate) {
   base::MutexGuard guard(&mutex_);
   DCHECK_EQ(1, isolates_.count(isolate));
-  for (auto& entry : jobs_) {
+  for (auto& entry : async_compile_jobs_) {
     if (entry.first->isolate() == isolate) return true;
   }
   return false;
 }
 
-void WasmEngine::DeleteCompileJobsOnIsolate(Isolate* isolate) {
-  base::MutexGuard guard(&mutex_);
-  DCHECK_EQ(1, isolates_.count(isolate));
-  for (auto it = jobs_.begin(); it != jobs_.end();) {
-    if (it->first->isolate() == isolate) {
-      it = jobs_.erase(it);
-    } else {
-      ++it;
+void WasmEngine::AbortCompileJobsOnIsolate(Isolate* isolate) {
+  // Copy out to a vector first, since abortion can free {AsyncCompileJob}s and
+  // thus modify the {async_compile_jobs_} set.
+  std::vector<std::shared_ptr<AsyncCompileJob>> compile_jobs_to_abort;
+  {
+    base::MutexGuard guard(&mutex_);
+    DCHECK_EQ(1, isolates_.count(isolate));
+    for (auto& job_entry : async_compile_jobs_) {
+      if (job_entry.first->isolate() != isolate) continue;
+      if (auto shared_job = job_entry.second.lock()) {
+        compile_jobs_to_abort.emplace_back(std::move(shared_job));
+      }
     }
   }
+  for (auto job : compile_jobs_to_abort) job->Abort();
 }
 
 void WasmEngine::AddIsolate(Isolate* isolate) {

src/wasm/wasm-engine.h

@@ -131,16 +131,16 @@ class V8_EXPORT_PRIVATE WasmEngine {
   CodeTracer* GetCodeTracer();
 
   // Remove {job} from the list of active compile jobs.
-  std::unique_ptr<AsyncCompileJob> RemoveCompileJob(AsyncCompileJob* job);
+  void RemoveCompileJob(AsyncCompileJob* job);
 
   // Returns true if at least one AsyncCompileJob that belongs to the given
   // Isolate is currently running.
   bool HasRunningCompileJob(Isolate* isolate);
 
-  // Deletes all AsyncCompileJobs that belong to the given Isolate. All
+  // Aborts all AsyncCompileJobs that belong to the given Isolate. All
   // compilation is aborted, no more callbacks will be triggered. This is used
   // for tearing down an isolate, or to clean it up to be reused.
-  void DeleteCompileJobsOnIsolate(Isolate* isolate);
+  void AbortCompileJobsOnIsolate(Isolate* isolate);
 
   // Manage the set of Isolates that use this WasmEngine.
   void AddIsolate(Isolate* isolate);
@@ -155,7 +155,7 @@ class V8_EXPORT_PRIVATE WasmEngine {
   static std::shared_ptr<WasmEngine> GetWasmEngine();
 
  private:
-  AsyncCompileJob* CreateAsyncCompileJob(
+  std::shared_ptr<AsyncCompileJob> CreateAsyncCompileJob(
       Isolate* isolate, const WasmFeatures& enabled,
       std::unique_ptr<byte[]> bytes_copy, size_t length,
       Handle<Context> context,
@@ -172,9 +172,11 @@ class V8_EXPORT_PRIVATE WasmEngine {
   //////////////////////////////////////////////////////////////////////////////
   // Protected by {mutex_}:
 
-  // We use an AsyncCompileJob as the key for itself so that we can delete the
-  // job from the map when it is finished.
-  std::unordered_map<AsyncCompileJob*, std::unique_ptr<AsyncCompileJob>> jobs_;
+  // Keep weak_ptrs to the AsyncCompileJob so we can detect the intermediate
+  // state where the refcount already dropped to zero (and the weak_ptr is
+  // cleared) but the destructor did not run to completion yet.
+  std::unordered_map<AsyncCompileJob*, std::weak_ptr<AsyncCompileJob>>
+      async_compile_jobs_;
 
   std::unique_ptr<CompilationStatistics> compilation_stats_;
   std::unique_ptr<CodeTracer> code_tracer_;