Block in dlopen until all threads have loaded the module

Fixes: #18345

Author: sbc100

emcc.py

@@ -1634,7 +1634,12 @@ def setup_pthreads(target):
   ]
 
   if settings.MAIN_MODULE:
-    settings.REQUIRED_EXPORTS += ['_emscripten_thread_sync_code', '__dl_seterr']
+    settings.REQUIRED_EXPORTS += [
+      '_emscripten_thread_sync_code',
+      '_emscripten_thread_sync_code_async',
+      '_emscripten_proxy_sync_code',
+      '__dl_seterr',
+    ]
 
   settings.DEFAULT_LIBRARY_FUNCS_TO_INCLUDE += [
     '$exitOnMainThread',

src/library_pthread.js

@@ -31,6 +31,7 @@ var LibraryPThread = {
   $PThread__deps: ['_emscripten_thread_init',
                    '$killThread',
                    '$cancelThread', '$cleanupThread', '$zeroMemory',
+                   '$markAsZombie',
                    '$spawnThread',
                    '_emscripten_thread_free_data',
                    'exit',
@@ -55,6 +56,7 @@ var LibraryPThread = {
     // the reverse mapping, each worker has a `pthread_ptr` when its running a
     // pthread.
     pthreads: {},
+    zombieThreads: {},
 #if PTHREADS_DEBUG
     nextWorkerID: 1,
     debugInit: function() {
@@ -267,6 +269,8 @@ var LibraryPThread = {
           spawnThread(d);
         } else if (cmd === 'cleanupThread') {
           cleanupThread(d['thread']);
+        } else if (cmd === 'markAsZombie') {
+          markAsZombie(d['thread']);
         } else if (cmd === 'killThread') {
           killThread(d['thread']);
         } else if (cmd === 'cancelThread') {
@@ -527,6 +531,18 @@ var LibraryPThread = {
     worker.pthread_ptr = 0;
   },
 
+  __emscripten_thread_exit_joinable: function(thread) {
+    // Called when a thread exits and is joinable.  This puts the thread
+    // into zombie state where it can't run anymore work but cannot yet
+    // be cleaned up.
+    if (!ENVIRONMENT_IS_PTHREAD) markAsZombie(thread);
+    else postMessage({ 'cmd': 'markAsZombie', 'thread': thread });
+  },
+
+  $markAsZombie: function(pthread_ptr) {
+    PThread.zombieThreads[pthread_ptr] = true;
+  },
+
   __emscripten_thread_cleanup: function(thread) {
     // Called when a thread needs to be cleaned up so it can be reused.
     // A thread is considered reusable when it either returns from its
@@ -543,6 +559,7 @@ var LibraryPThread = {
     assert(pthread_ptr, 'Internal Error! Null pthread_ptr in cleanupThread!');
 #endif
     var worker = PThread.pthreads[pthread_ptr];
+    delete PThread.zombieThreads[pthread_ptr];
     assert(worker);
     PThread.returnWorkerToPool(worker);
   },
@@ -1069,6 +1086,97 @@ var LibraryPThread = {
 #endif
   },
 
+#if MAIN_MODULE
+  $promiseMap: "new Map();",
+  $nextPromiseId: 1,
+
+  // Create a new promise that can be resolved or rejected by passing
+  // a unique ID to emscripten_promise_resolve/emscripten_promise_reject
+  // TODO(sbc): Should we factor this out and make this API public?
+  $newNativePromise__deps: ['$promiseMap', '$nextPromiseId'],
+  $newNativePromise: function(func, args) {
+    return new Promise((resolve, reject) => {
+      var promiseId = nextPromiseId;
+      nextPromiseId += 1;
+#if RUNTIME_DEBUG
+      dbg('newNativePromise: ' + promiseId);
+#endif
+      promiseMap.set(promiseId, {resolve, reject});
+      // Native promise function take promise ID as last argument
+      args.push(promiseId);
+      func.apply(null, args);
+    });
+  },
+
+  _emscripten_promise_resolve__deps: ['$promiseMap'],
+  _emscripten_promise_resolve__sig: 'vip',
+  _emscripten_promise_resolve: function(id, value) {
+#if RUNTIME_DEBUG
+    err('emscripten_resolve_promise: ' + id);
+#endif
+    assert(promiseMap.has(id));
+    promiseMap.get(id).resolve(value);
+    promiseMap.delete(id);
+  },
+
+  _emscripten_promise_reject__deps: ['$promiseMap'],
+  _emscripten_promise_reject__sig: 'vip',
+  _emscripten_promise_reject: function(id) {
+#if RUNTIME_DEBUG
+    dbg('emscripten_promise_reject: ' + id);
+#endif
+    assert(promiseMap.has(id));
+    promiseMap.get(id).reject();
+    promiseMap.delete(id);
+  },
+
+  // Called on the main thread to synchronize the code loaded on all threads.
+  // This work happens asynchronously. The `callback` is called once this work
+  // is completed, passing the ctx.
+  // TODO(sbc): Should we make a new form of __proxy attribute for JS library
+  // function that run asynchronously like but blocks the caller until they are
+  // done.  Perhaps "sync_with_ctx"?
+  _emscripten_sync_all_threads__sig: 'viii',
+  _emscripten_sync_all_threads__deps: ['_emscripten_proxy_sync_code', '$newNativePromise'],
+  _emscripten_sync_all_threads: function(caller, callback, ctx) {
+#if PTHREADS_DEBUG
+    dbg("_emscripten_sync_all_threads caller=" + ptrToString(caller));
+#endif
+#if ASSERTIONS
+    assert(!ENVIRONMENT_IS_PTHREAD, 'Internal Error! _emscripten_sync_all_threads() can only ever be called from main thread');
+#endif
+
+    let promises = [];
+
+    // This first promise resolves once the main thread has loaded all modules.
+    promises.push(newNativePromise(__emscripten_thread_sync_code_async, []));
+
+    // We then create a sequence of promises, one per thread, that resolve once
+    // each thread has performed its sync using _emscripten_proxy_sync_code.
+    // Any new threads that are created after this call will automaticaly be
+    // in sync because we call `__emscripten_thread_sync_code` in
+    // invokeEntryPoint before the threads entry point is called.
+    for (const ptr of Object.keys(PThread.pthreads)) {
+      const pthread_ptr = Number(ptr);
+      if (pthread_ptr !== caller && !(pthread_ptr in PThread.zombieThreads)) {
+        promises.push(newNativePromise(__emscripten_proxy_sync_code, [pthread_ptr]));
+      }
+    }
+
+#if PTHREADS_DEBUG
+    dbg("_emscripten_sync_all_threads: waiting on " + promises.length + " promises");
+#endif
+    // Once all promises are resolved then we know all threads are in sync and
+    // we can call the callback.
+    Promise.all(promises).then(() => {
+#if PTHREADS_DEBUG
+      dbg("_emscripten_sync_all_threads done: calling callback");
+#endif
+      {{{ makeDynCall('vp', 'callback') }}}(ctx);
+    });
+  },
+#endif
+
   $executeNotifiedProxyingQueue: function(queue) {
     // Set the notification state to processing.
     Atomics.store(HEAP32, queue >> 2, {{{ cDefine('NOTIFICATION_RECEIVED') }}});

system/include/emscripten/threading.h

@@ -286,9 +286,6 @@ int emscripten_pthread_attr_settransferredcanvases(pthread_attr_t *a, const char
 // blocking is not enabled, see ALLOW_BLOCKING_ON_MAIN_THREAD.
 void emscripten_check_blocking_allowed(void);
 
-// Experimental API for syncing loaded code between pthreads.
-void _emscripten_thread_sync_code();
-
 void _emscripten_yield();
 
 #ifdef __cplusplus