Revert "[wasm] Remove finisher task"

This reverts commit ac2fb66b.

Reason for revert: Flakily crashes on several bots:
https://ci.chromium.org/p/v8/builders/luci.v8.ci/V8%20Win32/18524
https://ci.chromium.org/p/v8/builders/luci.v8.ci/V8%20Win64%20-%20msvc/6824
https://ci.chromium.org/p/v8/builders/luci.v8.ci/V8%20Linux64%20-%20internal%20snapshot/19766

Original change's description:
> [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/1335553
> Reviewed-by: Andreas Haas <ahaas@chromium.org>
> Commit-Queue: Clemens Hammacher <clemensh@chromium.org>
> Cr-Commit-Position: refs/heads/master@{#58630}

TBR=ahaas@chromium.org,clemensh@chromium.org

Change-Id: I6b332b66adaec8f713fb31f4c8517cae7ebb4645
No-Presubmit: true
No-Tree-Checks: true
No-Try: true
Bug: v8:7921, v8:8423
Cq-Include-Trybots: luci.v8.try:v8_linux64_tsan_rel
Reviewed-on: https://chromium-review.googlesource.com/c/1400420Reviewed-by: Michael Achenbach <machenbach@chromium.org>
Commit-Queue: Michael Achenbach <machenbach@chromium.org>
Cr-Commit-Position: refs/heads/master@{#58634}

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()->AbortCompileJobsOnIsolate(isolate);
+    isolate->wasm_engine()->DeleteCompileJobsOnIsolate(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()->AbortCompileJobsOnIsolate(this);
+  wasm_engine()->DeleteCompileJobsOnIsolate(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 AbortCompilation();
+  void CancelAndWait();
 
   void SetError(uint32_t func_index, const ResultBase& error_result);
 

src/wasm/module-compiler.h

@@ -65,11 +65,8 @@ 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 : public std::enable_shared_from_this<AsyncCompileJob> {
+class AsyncCompileJob {
  public:
   AsyncCompileJob(Isolate* isolate, const WasmFeatures& enabled_features,
                   std::unique_ptr<byte[]> bytes_copy, size_t length,
@@ -95,6 +92,7 @@ class AsyncCompileJob : public std::enable_shared_from_this<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,10 +103,8 @@ void StreamingDecoder::Finish() {
 void StreamingDecoder::Abort() {
   TRACE_STREAMING("Abort\n");
   if (!ok()) return;  // Failed already.
-  // 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();
+  processor_->OnAbort();
+  Fail();
 }
 
 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_->AbortCompilation();
+  compilation_state_->CancelAndWait();
   code_manager_->FreeNativeModule(this);
 }
 

src/wasm/wasm-engine.cc

@@ -24,7 +24,7 @@ WasmEngine::WasmEngine()
 
 WasmEngine::~WasmEngine() {
   // All AsyncCompileJobs have been canceled.
-  DCHECK(async_compile_jobs_.empty());
+  DCHECK(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());
 
-  std::shared_ptr<AsyncCompileJob> job = CreateAsyncCompileJob(
+  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) {
-  std::shared_ptr<AsyncCompileJob> job =
+  AsyncCompileJob* job =
       CreateAsyncCompileJob(isolate, enabled, std::unique_ptr<byte[]>(nullptr),
                             0, context, std::move(resolver));
   return job->CreateStreamingDecoder();
@@ -278,49 +278,48 @@ CodeTracer* WasmEngine::GetCodeTracer() {
   return code_tracer_.get();
 }
 
-std::shared_ptr<AsyncCompileJob> WasmEngine::CreateAsyncCompileJob(
+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) {
-  std::shared_ptr<AsyncCompileJob> job =
-      std::make_shared<AsyncCompileJob>(isolate, enabled, std::move(bytes_copy),
-                                        length, context, std::move(resolver));
+  AsyncCompileJob* job =
+      new AsyncCompileJob(isolate, enabled, std::move(bytes_copy), length,
+                          context, std::move(resolver));
+  // Pass ownership to the unique_ptr in {jobs_}.
   base::MutexGuard guard(&mutex_);
-  async_compile_jobs_.insert({job.get(), job});
+  jobs_[job] = std::unique_ptr<AsyncCompileJob>(job);
   return job;
 }
 
-void WasmEngine::RemoveCompileJob(AsyncCompileJob* job) {
+std::unique_ptr<AsyncCompileJob> WasmEngine::RemoveCompileJob(
+    AsyncCompileJob* job) {
   base::MutexGuard guard(&mutex_);
-  auto item = async_compile_jobs_.find(job);
-  DCHECK(item != async_compile_jobs_.end());
-  async_compile_jobs_.erase(item);
+  auto item = jobs_.find(job);
+  DCHECK(item != jobs_.end());
+  std::unique_ptr<AsyncCompileJob> result = std::move(item->second);
+  jobs_.erase(item);
+  return result;
 }
 
 bool WasmEngine::HasRunningCompileJob(Isolate* isolate) {
   base::MutexGuard guard(&mutex_);
   DCHECK_EQ(1, isolates_.count(isolate));
-  for (auto& entry : async_compile_jobs_) {
+  for (auto& entry : jobs_) {
     if (entry.first->isolate() == isolate) return true;
   }
   return false;
 }
 
-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));
-      }
+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;
     }
   }
-  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.
-  void RemoveCompileJob(AsyncCompileJob* job);
+  std::unique_ptr<AsyncCompileJob> RemoveCompileJob(AsyncCompileJob* job);
 
   // Returns true if at least one AsyncCompileJob that belongs to the given
   // Isolate is currently running.
   bool HasRunningCompileJob(Isolate* isolate);
 
-  // Aborts all AsyncCompileJobs that belong to the given Isolate. All
+  // Deletes 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 AbortCompileJobsOnIsolate(Isolate* isolate);
+  void DeleteCompileJobsOnIsolate(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:
-  std::shared_ptr<AsyncCompileJob> CreateAsyncCompileJob(
+  AsyncCompileJob* CreateAsyncCompileJob(
       Isolate* isolate, const WasmFeatures& enabled,
       std::unique_ptr<byte[]> bytes_copy, size_t length,
       Handle<Context> context,
@@ -172,11 +172,9 @@ class V8_EXPORT_PRIVATE WasmEngine {
   //////////////////////////////////////////////////////////////////////////////
   // Protected by {mutex_}:
 
-  // 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_;
+  // 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_;
 
   std::unique_ptr<CompilationStatistics> compilation_stats_;
   std::unique_ptr<CodeTracer> code_tracer_;