Bug 1522945: Dequeue OffThreadPromiseTasks one at a time, to support reentrant draining. r=luke

`OffThreadPromiseRuntimeState::internalDrain` assumes that if
`internalDispatchQueue_` is empty but `live_` is not, then there must be
`OffThreadPromiseTasks` still in flight that it should block for. To support
reentrant calls to `internalDrain`, we need to make two changes:

- First, `internalDrain` needs to dequeue jobs one at a time, rather than
  swapping out the entire queue, which makes the queue look empty when there are
  still jobs yet to be run.

- Second, `OffThreadPromiseTask::run` needs to remove each task from `live_`
  *before* calling its `resolve` method, so that if the last task in the queue
  reenters `internalDrain`, that won't mistake the final entry in `live_` for
  something it must block on.

Since there are still `OffThreadPromiseTasks` that get registered in `live_`
but then get destructed without being run (for example, in `WasmJS.cpp`'s
`ResolveResponse_OnFulfilled`), we cannot assume that the `run` method will
unregister all tasks; the destructor still needs to check if unregistration is
necessary. So we factor out unregistration into its own method.

Differential Revision: https://phabricator.services.mozilla.com/D17704

--HG--
extra : moz-landing-system : lando

js/src/builtin/Promise.cpp

@@ -4924,8 +4924,7 @@ OffThreadPromiseTask::~OffThreadPromiseTask() {
   MOZ_ASSERT(state.initialized());
 
   if (registered_) {
-    LockGuard<Mutex> lock(state.mutex_);
-    state.live_.remove(this);
+    unregister(state);
   }
 }
 
@@ -4947,12 +4946,25 @@ bool OffThreadPromiseTask::init(JSContext* cx) {
   return true;
 }
 
+void OffThreadPromiseTask::unregister(OffThreadPromiseRuntimeState& state) {
+  MOZ_ASSERT(registered_);
+  LockGuard<Mutex> lock(state.mutex_);
+  state.live_.remove(this);
+  registered_ = false;
+}
+
 void OffThreadPromiseTask::run(JSContext* cx,
                                MaybeShuttingDown maybeShuttingDown) {
   MOZ_ASSERT(cx->runtime() == runtime_);
   MOZ_ASSERT(CurrentThreadCanAccessRuntime(runtime_));
   MOZ_ASSERT(registered_);
-  MOZ_ASSERT(runtime_->offThreadPromiseState.ref().initialized());
+
+  // Remove this task from live_ before calling `resolve`, so that if `resolve`
+  // itself drains the queue reentrantly, the queue will not think this task is
+  // yet to be queued and block waiting for it.
+  OffThreadPromiseRuntimeState& state = runtime_->offThreadPromiseState.ref();
+  MOZ_ASSERT(state.initialized());
+  unregister(state);
 
   if (maybeShuttingDown == JS::Dispatchable::NotShuttingDown) {
     // We can't leave a pending exception when returning to the caller so do
@@ -5058,8 +5070,8 @@ void OffThreadPromiseRuntimeState::internalDrain(JSContext* cx) {
   MOZ_ASSERT(usingInternalDispatchQueue());
   MOZ_ASSERT(!internalDispatchQueueClosed_);
 
-  while (true) {
-    DispatchableFifo dispatchQueue;
+  for (;;) {
+    JS::Dispatchable* d;
     {
       LockGuard<Mutex> lock(mutex_);
 
@@ -5068,18 +5080,17 @@ void OffThreadPromiseRuntimeState::internalDrain(JSContext* cx) {
         return;
       }
 
+      // There are extant live OffThreadPromiseTasks. If none are in the queue,
+      // block until one of them finishes and enqueues a dispatchable.
       while (internalDispatchQueue_.empty()) {
         internalDispatchQueueAppended_.wait(lock);
       }
 
-      mozilla::Swap(dispatchQueue, internalDispatchQueue_);
-      MOZ_ASSERT(internalDispatchQueue_.empty());
+      d = internalDispatchQueue_.popCopyFront();
     }
 
     // Don't call run() with mutex_ held to avoid deadlock.
-    for (JS::Dispatchable* d : dispatchQueue) {
-      d->run(cx, JS::Dispatchable::NotShuttingDown);
-    }
+    d->run(cx, JS::Dispatchable::NotShuttingDown);
   }
 }
 

js/src/builtin/Promise.h

@@ -423,6 +423,8 @@ class MOZ_NON_TEMPORARY_CLASS PromiseLookup final {
   }
 };
 
+class OffThreadPromiseRuntimeState;
+
 // [SMDOC] OffThreadPromiseTask: an off-main-thread task that resolves a promise
 //
 // An OffThreadPromiseTask is an abstract base class holding a JavaScript
@@ -498,6 +500,8 @@ class OffThreadPromiseTask : public JS::Dispatchable {
   void operator=(const OffThreadPromiseTask&) = delete;
   OffThreadPromiseTask(const OffThreadPromiseTask&) = delete;
 
+  void unregister(OffThreadPromiseRuntimeState& state);
+
  protected:
   OffThreadPromiseTask(JSContext* cx, Handle<PromiseObject*> promise);
 

js/src/jit-test/tests/debug/onNewScript-wasm-01.js

@@ -0,0 +1,29 @@
+// |jit-test| skip-if: !wasmDebuggingIsSupported()
+// Draining the job queue from an onNewScript hook reporting a streamed wasm
+// module should not deadlock.
+
+ignoreUnhandledRejections();
+
+try {
+    WebAssembly.compileStreaming();
+} catch (err) {
+    assertEq(String(err).indexOf("not supported with --no-threads") !== -1, true);
+    quit();
+}
+
+load(libdir + "asserts.js");
+
+var g = newGlobal({newCompartment: true});
+
+var source = new g.Uint8Array(wasmTextToBinary('(module (func unreachable))'));
+g.source = source;
+
+var dbg = new Debugger(g);
+dbg.allowWasmBinarySource = true;
+dbg.onNewScript = function (s, g) {
+    drainJobQueue();
+};
+
+g.eval('WebAssembly.instantiateStreaming(source);');
+
+drainJobQueue();

js/src/jit-test/tests/debug/onNewScript-wasm-02.js

@@ -0,0 +1,37 @@
+// |jit-test| skip-if: !wasmDebuggingIsSupported()
+// Draining the job queue from an onNewScript hook reporting a streamed wasm
+// module should not deadlock.
+
+ignoreUnhandledRejections();
+
+try {
+    WebAssembly.compileStreaming();
+} catch (err) {
+    assertEq(String(err).indexOf("not supported with --no-threads") !== -1, true);
+    quit();
+}
+
+var g = newGlobal({newCompartment: true});
+
+var source = new g.Uint8Array(wasmTextToBinary('(module (func unreachable))'));
+g.source = source;
+
+var dbg = new Debugger(g);
+dbg.allowWasmBinarySource = true;
+dbg.onNewScript = function (s, g) {
+  drainJobQueue();
+};
+
+// For the old code to deadlock, OffThreadPromiseRuntimeState::internalDrain
+// needs to get two Dispatchables at once when it swaps queues. A call to
+// instantiateStreaming enqueues a job that will kick off a helper thread, so to
+// make sure that two helper threads have had time to complete and enqueue their
+// Dispatchables, we put a delay in the job queue after the helper thread
+// launches.
+g.eval(`
+  WebAssembly.instantiateStreaming(source);
+  WebAssembly.instantiateStreaming(source);
+  Promise.resolve().then(() => sleep(0.1));
+`);
+
+drainJobQueue();