Decrement proxying queue refcounts in finally blocks (#16571)

This ensures that the refcounts are correctly decremented even if a proxied task
throws an exception, although it won't protect again other bad things like
dropped work that could happen in that case.

Author: tlively

src/library_pthread.js

@@ -268,10 +268,7 @@ var LibraryPThread = {
         }
 
         if (cmd === 'processProxyingQueue') {
-          // TODO: Must post message to main Emscripten thread in PROXY_TO_WORKER mode.
-          _emscripten_proxy_execute_queue(d['queue']);
-          // Decrement the ref count
-          Atomics.sub(HEAP32, d['queue'] >> 2, 1);
+          executeNotifiedProxyingQueue(d['queue']);
         } else if (cmd === 'spawnThread') {
           spawnThread(d);
         } else if (cmd === 'cleanupThread') {
@@ -1047,19 +1044,25 @@ var LibraryPThread = {
     return {{{ makeDynCall('ii', 'ptr') }}}(arg);
   },
 
+  $executeNotifiedProxyingQueue: function(queue) {
+    try {
+      // Only execute the queue if we have a live pthread runtime. We
+      // implement pthread_self to return 0 if there is no live runtime.
+      // TODO: Use `callUserCallback` to correctly handle unwinds, etc. once
+      //       `runtimeExited` is correctly unset on workers.
+      if (_pthread_self()) {
+        _emscripten_proxy_execute_queue(queue);
+      }
+    } finally {
+      // Always decrement the ref count.
+      Atomics.sub(HEAP32, queue >> 2, 1);
+    }
+  },
+
+  _emscripten_notify_proxying_queue__deps: ['$executeNotifiedProxyingQueue'],
   _emscripten_notify_proxying_queue: function(targetThreadId, currThreadId, mainThreadId, queue) {
     if (targetThreadId == currThreadId) {
-      setTimeout(() => {
-        // Only execute the queue if we have a live pthread runtime. We
-        // implement pthread_self to return 0 if there is no live runtime.
-        // TODO: Use `callUserCallback` to correctly handle unwinds, etc. once
-        //       `runtimeExited` is correctly unset on workers.
-        if (_pthread_self()) {
-          _emscripten_proxy_execute_queue(queue);
-        }
-        // Decrement the ref count
-        Atomics.sub(HEAP32, queue >> 2, 1);
-      });
+      setTimeout(() => executeNotifiedProxyingQueue(queue));
     } else if (ENVIRONMENT_IS_PTHREAD) {
       postMessage({'targetThread' : targetThreadId, 'cmd' : 'processProxyingQueue', 'queue' : queue});
     } else {

src/worker.js

@@ -286,11 +286,7 @@ self.onmessage = (e) => {
     } else if (e.data.target === 'setimmediate') {
       // no-op
     } else if (e.data.cmd === 'processProxyingQueue') {
-      if (Module['_pthread_self']()) { // If this thread is actually running?
-        Module['_emscripten_proxy_execute_queue'](e.data.queue);
-      }
-      // Decrement the ref count
-      Atomics.sub(HEAP32, e.data.queue >> 2, 1);
+      executeNotifiedProxyingQueue(e.data.queue);
     } else {
       err('worker.js received unknown command ' + e.data.cmd);
       err(e.data);