Reland "[Atomics.waitAsync] Fix removing multiple nodes when Isolate deinits"

This is a reland of 28ead0540534c8c532fb2a5c92ffd536ae988651

The failure is a test that is sensitive to adding a function
in a FunctionTemplate in d8: https://bugs.chromium.org/p/v8/issues/detail?id=10783

Original change's description:
> [Atomics.waitAsync] Fix removing multiple nodes when Isolate deinits
>
> RemoveNode already nullifies the next_ pointer of FutexWaitListNode,
> and DeleteAsyncNode was trying to retrieve it.
>
> Bug: v8:10239
> Change-Id: I595885de87f433d263eeacfc825a689efd467f5e
> Reviewed-on: https://chromium-review.googlesource.com/c/v8/v8/+/2332812
> Commit-Queue: Marja Hölttä <marja@chromium.org>
> Reviewed-by: Ulan Degenbaev <ulan@chromium.org>
> Reviewed-by: Shu-yu Guo <syg@chromium.org>
> Cr-Commit-Position: refs/heads/master@{#69259}

Bug: v8:10239
Tbr: leszeks@chromium.org
Change-Id: Icec590354886433a0b41c8f9b7af7101b54b7690
Reviewed-on: https://chromium-review.googlesource.com/c/v8/v8/+/2339469
Reviewed-by: Marja Hölttä <marja@chromium.org>
Commit-Queue: Marja Hölttä <marja@chromium.org>
Cr-Commit-Position: refs/heads/master@{#69275}
diff --git a/src/d8/d8.cc b/src/d8/d8.cc
index 1af1baf..f199204 100644
--- a/src/d8/d8.cc
+++ b/src/d8/d8.cc
@@ -1809,6 +1809,19 @@
   worker->Terminate();
 }
 
+void Shell::WorkerTerminateAndWait(
+    const v8::FunctionCallbackInfo<v8::Value>& args) {
+  Isolate* isolate = args.GetIsolate();
+  HandleScope handle_scope(isolate);
+  std::shared_ptr<Worker> worker =
+      GetWorkerFromInternalField(isolate, args.Holder());
+  if (!worker.get()) {
+    return;
+  }
+
+  worker->TerminateAndWaitForThread();
+}
+
 void Shell::QuitOnce(v8::FunctionCallbackInfo<v8::Value>* args) {
   int exit_code = (*args)[0]
                       ->Int32Value(args->GetIsolate()->GetCurrentContext())
@@ -2139,6 +2152,10 @@
       FunctionTemplate::New(isolate, WorkerTerminate, Local<Value>(),
                             worker_signature));
   worker_fun_template->PrototypeTemplate()->Set(
+      isolate, "terminateAndWait",
+      FunctionTemplate::New(isolate, WorkerTerminateAndWait, Local<Value>(),
+                            worker_signature));
+  worker_fun_template->PrototypeTemplate()->Set(
       isolate, "postMessage",
       FunctionTemplate::New(isolate, WorkerPostMessage, Local<Value>(),
                             worker_signature));
@@ -3056,7 +3073,7 @@
   task_runner_->PostTask(std::move(task));
 }
 
-void Worker::WaitForThread() {
+void Worker::TerminateAndWaitForThread() {
   Terminate();
   thread_->Join();
 }
@@ -3917,7 +3934,7 @@
   }
 
   for (auto& worker : workers_copy) {
-    worker->WaitForThread();
+    worker->TerminateAndWaitForThread();
   }
 
   // Now that all workers are terminated, we can re-enable Worker creation.
diff --git a/src/d8/d8.h b/src/d8/d8.h
index 3c680e4..23e7d16 100644
--- a/src/d8/d8.h
+++ b/src/d8/d8.h
@@ -183,7 +183,7 @@
   void Terminate();
   // Terminate and join the thread.
   // This function can be called by any thread.
-  void WaitForThread();
+  void TerminateAndWaitForThread();
 
   // Start running the given worker in another thread.
   static bool StartWorkerThread(std::shared_ptr<Worker> worker);
@@ -429,6 +429,8 @@
       const v8::FunctionCallbackInfo<v8::Value>& args);
   static void WorkerGetMessage(const v8::FunctionCallbackInfo<v8::Value>& args);
   static void WorkerTerminate(const v8::FunctionCallbackInfo<v8::Value>& args);
+  static void WorkerTerminateAndWait(
+      const v8::FunctionCallbackInfo<v8::Value>& args);
   // The OS object on the global object contains methods for performing
   // operating system calls:
   //
diff --git a/src/execution/futex-emulation.cc b/src/execution/futex-emulation.cc
index f22be07..699fd93 100644
--- a/src/execution/futex-emulation.cc
+++ b/src/execution/futex-emulation.cc
@@ -774,8 +774,10 @@
       // NativeContext. Also we don't need to cancel the timeout task, since it
       // will be cancelled by Isolate::Deinit.
       node->timeout_task_id_ = CancelableTaskManager::kInvalidTaskId;
+      auto next = node->next_;
       g_wait_list.Pointer()->RemoveNode(node);
-      node = DeleteAsyncWaiterNode(node);
+      delete node;
+      node = next;
     } else {
       node = node->next_;
     }
diff --git a/test/mjsunit/harmony/atomics-waitasync-worker-shutdown-before-wait-finished-2-waits.js b/test/mjsunit/harmony/atomics-waitasync-worker-shutdown-before-wait-finished-2-waits.js
new file mode 100644
index 0000000..5f46094
--- /dev/null
+++ b/test/mjsunit/harmony/atomics-waitasync-worker-shutdown-before-wait-finished-2-waits.js
@@ -0,0 +1,32 @@
+// 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.
+
+// Flags: --allow-natives-syntax --harmony-sharedarraybuffer --harmony-atomics-waitasync
+
+(function test() {
+  const sab = new SharedArrayBuffer(16);
+  const i32a = new Int32Array(sab);
+  const location = 0;
+
+  (function createWorker() {
+    const script = `onmessage = function(msg) {
+      if (msg.sab) {
+        const i32a = new Int32Array(msg.sab);
+        // Start 2 async waits in the same location.
+        const result1 = Atomics.waitAsync(i32a, ${location}, 0);
+        const result2 = Atomics.waitAsync(i32a, ${location}, 0);
+        postMessage('worker waiting');
+      }
+    }`;
+    const w = new Worker(script, {type : 'string'});
+    w.postMessage({sab: sab});
+    const m = w.getMessage();
+    assertEquals('worker waiting', m);
+    w.terminateAndWait();
+  })();
+
+  const notify_return_value = Atomics.notify(i32a, location, 2);
+  // No waiters got notified, since they got cleaned up before it.
+  assertEquals(0, notify_return_value);
+})();
diff --git a/test/mjsunit/harmony/atomics-waitasync-worker-shutdown-before-wait-finished-2-workers.js b/test/mjsunit/harmony/atomics-waitasync-worker-shutdown-before-wait-finished-2-workers.js
new file mode 100644
index 0000000..c27a141
--- /dev/null
+++ b/test/mjsunit/harmony/atomics-waitasync-worker-shutdown-before-wait-finished-2-workers.js
@@ -0,0 +1,37 @@
+// 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.
+
+// Flags: --allow-natives-syntax --harmony-sharedarraybuffer --harmony-atomics-waitasync
+
+(function test() {
+  const sab = new SharedArrayBuffer(16);
+  const i32a = new Int32Array(sab);
+  const location = 0;
+
+  (function createWorker() {
+    const script = `onmessage = function(msg) {
+      if (msg.sab) {
+        const i32a = new Int32Array(msg.sab);
+        Atomics.waitAsync(i32a, ${location}, 0);
+        postMessage('worker waiting');
+      }
+    }`;
+    // Create 2 workers which wait on the same location.
+    let workers = [];
+    const worker_count = 2;
+    for (let i = 0; i < worker_count; ++i) {
+      workers[i] = new Worker(script, {type : 'string'});
+      workers[i].postMessage({sab: sab});
+      const m = workers[i].getMessage();
+      assertEquals('worker waiting', m);
+    }
+    for (let i = 0; i < worker_count; ++i) {
+      workers[i].terminateAndWait();
+    }
+  })();
+
+  const notify_return_value = Atomics.notify(i32a, location, 2);
+  // No waiters got notified, since they got cleaned up before it.
+  assertEquals(0, notify_return_value);
+})();