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

@@ -1069,6 +1069,83 @@ 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
+  $newNativePromise__deps: ['$promiseMap', '$nextPromiseId'],
+  $newNativePromise: function(func, args) {
+    return new Promise((resolve, reject) => {
+      var promiseId = nextPromiseId;
+      nextPromiseId += 1;
+      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();
+  },
+
+  // Called on the main thread to syncronize the code loaded on all threads.
+  // This work happens asyncronously. The `callback` is called once this work
+  // is completely, passing the 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 module
+    promises.push(newNativePromise(__emscripten_thread_sync_code_async, [0]));
+
+    // We then create a sequence of promises, one per thread, that resolve once
+    // each thread has performed its sync using _emscripten_proxy_sync_code.
+    for (const ptr of Object.keys(PThread.pthreads)) {
+      const pthread_ptr = Number(ptr);
+      if (pthread_ptr !== caller) {
+        promises.push(newNativePromise(__emscripten_proxy_sync_code, [pthread_ptr]));
+      }
+    }
+
+    // Once all promises are resolved then we 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

system/lib/libc/dynlink.c

@@ -21,6 +21,8 @@
 #include <dynlink.h>
 
 #include <emscripten/console.h>
+#include <emscripten/threading.h>
+#include <emscripten/proxying.h>
 
 //#define DYLINK_DEBUG
 
@@ -29,15 +31,15 @@ struct async_data {
   em_arg_callback_func onerror;
   void* user_data;
 };
-typedef void (*dlopen_callback_func)(struct dso*, struct async_data* user_data);
+typedef void (*dlopen_callback_func)(struct dso*, void* user_data);
 
 void _dlinit(struct dso* main_dso_handle);
 void* _dlopen_js(struct dso* handle);
 void* _dlsym_js(struct dso* handle, const char* symbol);
 void _emscripten_dlopen_js(struct dso* handle,
                            dlopen_callback_func onsuccess,
                            dlopen_callback_func onerror,
-                           struct async_data* user_data);
+                           void* user_data);
 void __dl_vseterr(const char*, va_list);
 
 static struct dso * _Atomic head, * _Atomic tail;
@@ -105,7 +107,8 @@ static struct dso* load_library_start(const char* name, int flags) {
   return p;
 }
 
-static void dlopen_js_onsuccess(struct dso* dso, struct async_data* data) {
+static void dlopen_onsuccess(struct dso* dso, void* user_data) {
+  struct async_data* data = (struct async_data*)user_data;
 #ifdef DYLINK_DEBUG
   _emscripten_errf("%p: dlopen_js_onsuccess: dso=%p mem_addr=%p mem_size=%zu",
                    pthread_self(),
@@ -119,7 +122,8 @@ static void dlopen_js_onsuccess(struct dso* dso, struct async_data* data) {
   free(data);
 }
 
-static void dlopen_js_onerror(struct dso* dso, struct async_data* data) {
+static void dlopen_onerror(struct dso* dso, void* user_data) {
+  struct async_data* data = (struct async_data*)user_data;
 #ifdef DYLINK_DEBUG
   _emscripten_errf("%p: dlopen_js_onerror: dso=%p", pthread_self(), dso);
 #endif
@@ -148,6 +152,153 @@ static void ensure_init() {
   pthread_rwlock_unlock(&lock);
 }
 
+#ifdef _REENTRANT
+// These functions are defined in JS.
+
+// Signal the completion of a given promise that was passed into
+// emscripten_proxy_sync_code.
+void emscripten_promise_resolve(int promise_id, void* value);
+void emscripten_promise_reject(int promise_id);
+
+void _emscripten_thread_sync_code_async(struct dso* dso, int promise_id);
+
+static void sync_one_onsuccess(struct dso* dso, void* user_data) {
+  int promise_id = (intptr_t)user_data;
+  // Load the next dso in the list
+  thread_local_tail = dso;
+  _emscripten_thread_sync_code_async(dso->next, promise_id);
+}
+
+static void sync_one_onerror(struct dso* dso, void* user_data) {
+  int promise_id = (intptr_t)user_data;
+  pthread_rwlock_unlock(&lock);
+  emscripten_promise_reject(promise_id);
+}
+
+void _emscripten_thread_sync_code_async(struct dso* dso, int promise_id) {
+  _emscripten_errf("_emscripten_thread_sync_code_async %p promise_id=%d", dso, promise_id);
+  ensure_init();
+  pthread_rwlock_rdlock(&lock);
+  if (!thread_local_tail) {
+    thread_local_tail = head;
+  }
+  if (!dso) {
+    dso = thread_local_tail->next;
+  }
+  if (!dso) {
+    pthread_rwlock_unlock(&lock);
+    emscripten_promise_resolve(promise_id, NULL);
+    return;
+  }
+
+  // Unlock happens once all DSO have been loaded, or one of them fails
+  // with sync_one_onerror.
+  _emscripten_dlopen_js(
+    dso, sync_one_onsuccess, sync_one_onerror, (void*)promise_id);
+}
+
+void _emscripten_thread_sync_code() {
+  // Should only ever be called from a background thread.
+  assert(!emscripten_is_main_runtime_thread());
+  ensure_init();
+  if (thread_local_tail == tail) {
+#ifdef DYLINK_DEBUG
+    _emscripten_errf("%p: emscripten_thread_sync_code: already in sync", pthread_self());
+#endif
+    return;
+  }
+  pthread_rwlock_rdlock(&lock);
+  if (!thread_local_tail) {
+    thread_local_tail = head;
+  }
+  while (thread_local_tail->next) {
+    struct dso* p = thread_local_tail->next;
+#ifdef DYLINK_DEBUG
+    fprintf(stderr,
+            "%p: emscripten_thread_sync_code: %s mem_addr=%p mem_size=%zu "
+            "table_addr=%p table_size=%d",
+            pthread_self(),
+            p->name,
+            p->mem_addr,
+            p->mem_size,
+            p->table_addr,
+            p->table_size);
+#endif
+    void* success = _dlopen_js(p);
+    if (!success) {
+      // TODO(sbc): Do a better job of handling errors here.
+      _emscripten_errf("emscripten_thread_sync_code failed: %s", dlerror());
+      abort();
+    }
+    thread_local_tail = p;
+  }
+  pthread_rwlock_unlock(&lock);
+#ifdef DYLINK_DEBUG
+  _emscripten_errf("%p: emscripten_thread_sync_code done", pthread_self());
+#endif
+}
+
+static void do_thread_sync(void* arg) {
+  int promise_id = (intptr_t)arg;
+#ifdef DYLINK_DEBUG
+  _emscripten_errf("%p: do_thread_sync: %d", pthread_self(), promise_id);
+#endif
+  _emscripten_thread_sync_code();
+}
+
+// Called once _emscripten_proxy_sync_code completes
+static void done_thread_sync(void* arg) {
+  int promise_id = (intptr_t)arg;
+#ifdef DYLINK_DEBUG
+  _emscripten_errf("%p: done_thread_sync: promise_id=%d", pthread_self(), promise_id);
+#endif
+  emscripten_promise_resolve(promise_id, NULL);
+}
+
+int _emscripten_proxy_sync_code(pthread_t target_thread, int promise_id) {
+  em_proxying_queue* q = emscripten_proxy_get_system_queue();
+  return emscripten_proxy_async_with_callback(
+    q, target_thread, do_thread_sync, NULL, done_thread_sync, (void*)promise_id);
+}
+
+static void done_sync_all(em_proxying_ctx* ctx) {
+#ifdef DYLINK_DEBUG
+  _emscripten_errf("%p: done_sync_all", pthread_self());
+#endif
+  emscripten_proxy_finish(ctx);
+}
+
+// This function is defined in JS.  It is an async function that runs
+// emscripten_thread_sync_code of each of threads that are running at the
+// time of the call.  Once this is done the callback is called with the given
+// em_proxying_ctx.
+void _emscripten_sync_all_threads(pthread_t calling_thread,
+                                  void (*callback)(em_proxying_ctx*),
+                                  em_proxying_ctx* ctx);
+
+static void main_thread_sync_all(em_proxying_ctx* ctx, void* arg) {
+  pthread_t calling_thread = (pthread_t)arg;
+#ifdef DYLINK_DEBUG
+  _emscripten_errf("%p: main_thread_sync_all calling=%p", pthread_self(), calling_thread);
+#endif
+  _emscripten_sync_all_threads(calling_thread, done_sync_all, ctx);
+}
+
+static void sync_code_all_threads() {
+  // Call `main_thread_sync_all` on the main thread and block until its
+  // complete. This gets called after a shared library is loaded by a worker.
+  pthread_t main_thread = emscripten_main_browser_thread_id();
+#ifdef DYLINK_DEBUG
+  _emscripten_errf("%p: sync_code_all_threads main=%p", pthread_self(), main_thread);
+#endif
+  assert(pthread_self() != main_thread);
+  em_proxying_queue* q = emscripten_proxy_get_system_queue();
+  int success = emscripten_proxy_sync_with_ctx(
+    q, main_thread, main_thread_sync_all, pthread_self());
+  assert(success);
+}
+#endif // _REENTRANT
+
 void* dlopen(const char* file, int flags) {
   ensure_init();
   if (!file) {
@@ -191,6 +342,10 @@ void* dlopen(const char* file, int flags) {
   load_library_done(p);
 end:
   pthread_rwlock_unlock(&lock);
+#ifdef _REENTRANT
+  // Block until all other threads have loaded this module.
+  sync_code_all_threads();
+#endif
   pthread_setcancelstate(cs, 0);
   return p;
 }
@@ -219,8 +374,8 @@ void emscripten_dlopen(const char* filename, int flags, void* user_data,
 #ifdef DYLINK_DEBUG
   _emscripten_errf("%p: calling emscripten_dlopen_js %p", pthread_self(), p);
 #endif
-  // Unlock happens in dlopen_js_onsuccess/dlopen_js_onerror
-  _emscripten_dlopen_js(p, dlopen_js_onsuccess, dlopen_js_onerror, d);
+  // Unlock happens in dlopen_onsuccess/dlopen_onerror
+  _emscripten_dlopen_js(p, dlopen_onsuccess, dlopen_onerror, d);
 }
 
 void* __dlsym(void* restrict p, const char* restrict s, void* restrict ra) {
@@ -246,57 +401,3 @@ int dladdr(const void* addr, Dl_info* info) {
   info->dli_saddr = NULL;
   return 1;
 }
-
-#ifdef _REENTRANT
-void _emscripten_thread_sync_code() {
-  // This function is called from emscripten_yeild which itself is called
-  // whenever we block on a futex.  We need to check to avoid infinite
-  // recursion when taking the lock below.
-  static thread_local bool syncing = false;
-  if (syncing) {
-    return;
-  }
-  syncing = true;
-  ensure_init();
-  if (thread_local_tail == tail) {
-#ifdef DYLINK_DEBUG
-    _emscripten_errf("%p: emscripten_thread_sync_code: already in sync", pthread_self());
-#endif
-    goto done;
-  }
-  pthread_rwlock_rdlock(&lock);
-  if (!thread_local_tail) {
-    thread_local_tail = head;
-  }
-  while (thread_local_tail->next) {
-    struct dso* p = thread_local_tail->next;
-#ifdef DYLINK_DEBUG
-    _emscripten_errf("%p: emscripten_thread_sync_code: %s mem_addr=%p "
-                     "mem_size=%zu table_addr=%p table_size=%zu",
-                     pthread_self(),
-                     p->name,
-                     p->mem_addr,
-                     p->mem_size,
-                     p->table_addr,
-                     p->table_size);
-#endif
-    void* success = _dlopen_js(p);
-    if (!success) {
-      // If any on the libraries fails to load here then we give up.
-      // TODO(sbc): Ideally this would never happen and we could/should
-      // abort, but on the main thread (where we don't have sync xhr) its
-      // often not possible to syncronously load side module.
-      _emscripten_errf("emscripten_thread_sync_code failed: %s", dlerror());
-      break;
-    }
-    thread_local_tail = p;
-  }
-  pthread_rwlock_unlock(&lock);
-#ifdef DYLINK_DEBUG
-  _emscripten_errf("%p: emscripten_thread_sync_code done", pthread_self());
-#endif
-
-done:
-  syncing = false;
-}
-#endif