Allow synchronous proxying of async JS functions

.circleci/config.yml

@@ -969,6 +969,7 @@ jobs:
             core0.test_pthread_join_and_asyncify
             core0.test_async_ccall_promise_jspi*
             core0.test_cubescript_jspi
+            core0.test_poll_blocking_jspi
             "
       # Run some basic tests with the minimum version of node that we currently
       # support in the generated code.

src/closure-externs/closure-externs.js

@@ -268,3 +268,5 @@ Symbol.dispose;
 var os = {};
 
 AudioWorkletProcessor.parameterDescriptors;
+
+Promise.withResolvers = function() {};

src/jsifier.mjs

@@ -446,20 +446,28 @@ function(${args}) {
         error(`JS library error: invalid proxying mode '${symbol}__proxy: ${proxyingMode}' specified`);
       }
       if (SHARED_MEMORY && proxyingMode != 'none') {
-        const sync = proxyingMode === 'sync';
         if (PTHREADS) {
           snippet = modifyJSFunction(snippet, (args, body, async_, oneliner) => {
             if (oneliner) {
               body = `return ${body}`;
             }
+            let sync = '0'
+            if (proxyingMode === 'sync') {
+              const isAsyncFunction = LibraryManager.library[symbol + '__async'];
+              if (isAsyncFunction) {
+                sync = '2';
+              } else {
+                sync = '1';
+              }
+            }
             const rtnType = sig && sig.length ? sig[0] : null;
             const proxyFunc =
               MEMORY64 && rtnType == 'p' ? 'proxyToMainThreadPtr' : 'proxyToMainThread';
             deps.push('$' + proxyFunc);
             return `
 ${async_}function(${args}) {
 if (ENVIRONMENT_IS_PTHREAD)
-  return ${proxyFunc}(${proxiedFunctionTable.length}, 0, ${+sync}${args ? ', ' : ''}${args});
+  return ${proxyFunc}(${proxiedFunctionTable.length}, 0, ${sync}${args ? ', ' : ''}${args});
 ${body}
 }\n`;
           });

src/lib/libpthread.js

@@ -982,8 +982,9 @@ var LibraryPThread = {
 
   _emscripten_receive_on_main_thread_js__deps: [
     '$proxyToMainThread',
+    '_emscripten_run_js_on_main_thread_done',
     '$proxiedJSCallArgs'],
-  _emscripten_receive_on_main_thread_js: (funcIndex, emAsmAddr, callingThread, bufSize, args) => {
+  _emscripten_receive_on_main_thread_js: (funcIndex, emAsmAddr, callingThread, bufSize, args, ctx, ctxArgs) => {
     // Sometimes we need to backproxy events to the calling thread (e.g.
     // HTML5 DOM events handlers such as
     // emscripten_set_mousemove_callback()), so keep track in a globally
@@ -1022,6 +1023,11 @@ var LibraryPThread = {
     PThread.currentProxiedOperationCallerThread = callingThread;
     var rtn = func(...proxiedJSCallArgs);
     PThread.currentProxiedOperationCallerThread = 0;
+    if (ctx) {
+      rtn.then((rtn) => __emscripten_run_js_on_main_thread_done(ctx, ctxArgs, rtn));
+      return;
+    }
+
 #if MEMORY64
     // In memory64 mode some proxied functions return bigint/pointer but
     // our return type is i53/double.
@@ -1160,14 +1166,11 @@ var LibraryPThread = {
     }
   },
 
-  // Asynchronous version dlsync_threads. Always run on the main thread.
-  // 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_dlsync_threads_async__deps: ['_emscripten_proxy_dlsync_async', '$makePromise'],
-  _emscripten_dlsync_threads_async: (caller, callback, ctx) => {
+  // Asynchronous version dlsync. Always run on the main thread.
+  // This work happens asynchronously.
+  $dlsyncAsync__deps: ['_emscripten_proxy_dlsync_async', '$makePromise'],
+  $dlsyncAsync: async () => {
+    const caller = PThread.currentProxiedOperationCallerThread;
 #if PTHREADS_DEBUG
     dbg("_emscripten_dlsync_threads_async caller=" + ptrToString(caller));
 #endif
@@ -1202,27 +1205,27 @@ var LibraryPThread = {
 #if PTHREADS_DEBUG
     dbg(`_emscripten_dlsync_threads_async: 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(() => {
-      PThread.outstandingPromises = {};
+    await Promise.all(promises);
+
+    PThread.outstandingPromises = {};
 #if PTHREADS_DEBUG
-      dbg('_emscripten_dlsync_threads_async done: calling callback');
+    dbg('_emscripten_dlsync_threads_async done');
 #endif
-      {{{ makeDynCall('vp', 'callback') }}}(ctx);
-    });
   },
 
-  // Synchronous version dlsync_threads. This is only needed for the case then
+  // Synchronous version of dlsync_threads. This is only needed for the case then
   // the main thread call dlopen and in that case we have not choice but to
   // synchronously block the main thread until all other threads are in sync.
   // When `dlopen` is called from a worker, the worker itself is blocked but
   // the operation its waiting on (on the main thread) can be async.
-  _emscripten_dlsync_threads__deps: ['_emscripten_proxy_dlsync'],
-  _emscripten_dlsync_threads: () => {
-#if ASSERTIONS
-    assert(!ENVIRONMENT_IS_PTHREAD, 'Internal Error! _emscripten_dlsync_threads() can only ever be called from main thread');
-#endif
+  _dlsync__deps: ['_emscripten_proxy_dlsync', '$dlsyncAsync'],
+  _dlsync__async: true,
+  _dlsync__proxy: 'sync',
+  _dlsync: () => {
+    const callingThread = PThread.currentProxiedOperationCallerThread;
+    if (callingThread) {
+      return dlsyncAsync()
+    }
     for (const ptr of Object.keys(PThread.pthreads)) {
       const pthread_ptr = Number(ptr);
       if (!PThread.finishedThreads.has(pthread_ptr)) {

src/lib/libsigs.js

@@ -262,6 +262,7 @@ sigs = {
   __syscall_newfstatat__sig: 'iippi',
   __syscall_openat__sig: 'iipip',
   __syscall_pipe__sig: 'ip',
+  __syscall_poll__sig: 'ipii',
   __syscall_readlinkat__sig: 'iippp',
   __syscall_recvfrom__sig: 'iippipp',
   __syscall_recvmsg__sig: 'iipiiii',
@@ -281,6 +282,7 @@ sigs = {
   _dlopen_js__sig: 'pp',
   _dlsym_catchup_js__sig: 'ppi',
   _dlsym_js__sig: 'pppp',
+  _dlsync__sig: 'v',
   _embind_create_inheriting_constructor__sig: 'pppp',
   _embind_finalize_value_array__sig: 'vp',
   _embind_finalize_value_object__sig: 'vp',
@@ -313,8 +315,6 @@ sigs = {
   _embind_register_void__sig: 'vpp',
   _emscripten_create_wasm_worker__sig: 'ipi',
   _emscripten_dlopen_js__sig: 'vpppp',
-  _emscripten_dlsync_threads__sig: 'v',
-  _emscripten_dlsync_threads_async__sig: 'vppp',
   _emscripten_fetch_get_response_headers__sig: 'pipp',
   _emscripten_fetch_get_response_headers_length__sig: 'pi',
   _emscripten_fs_load_embedded_files__sig: 'vp',
@@ -329,7 +329,7 @@ sigs = {
   _emscripten_notify_mailbox_postmessage__sig: 'vpp',
   _emscripten_push_main_loop_blocker__sig: 'vppp',
   _emscripten_push_uncounted_main_loop_blocker__sig: 'vppp',
-  _emscripten_receive_on_main_thread_js__sig: 'dippip',
+  _emscripten_receive_on_main_thread_js__sig: 'dippippp',
   _emscripten_runtime_keepalive_clear__sig: 'v',
   _emscripten_system__sig: 'ip',
   _emscripten_thread_cleanup__sig: 'vp',
@@ -379,7 +379,6 @@ sigs = {
   _mmap_js__sig: 'ipiiijpp',
   _msync_js__sig: 'ippiiij',
   _munmap_js__sig: 'ippiiij',
-  _poll_js__sig: 'ipiipp',
   _setitimer_js__sig: 'iid',
   _timegm_js__sig: 'jp',
   _tzset_js__sig: 'vpppp',

src/lib/libsyscall.js

@@ -551,15 +551,17 @@ var SyscallsLibrary = {
     var stream = SYSCALLS.getStreamFromFD(fd);
     return 0; // we can't do anything synchronously; the in-memory FS is already synced to
   },
-  _poll_js__proxy: 'none',
-  _poll_js__deps: [
+  __syscall_poll__proxy: 'sync',
+  __syscall_poll__async: true,
+  __syscall_poll: (fds, nfds, timeout) => {
+#if PTHREADS || JSPI
 #if PTHREADS
-    '_emscripten_proxy_poll_finish',
+    const isAsyncContext = PThread.currentProxiedOperationCallerThread;
+#else
+    const isAsyncContext = true;
 #endif
-  ],
-  _poll_js: (fds, nfds, timeout, ctx, arg) => {
-#if PTHREADS
     // Enable event handlers only when the poll call is proxied from a worker.
+    var { promise, resolve, reject } = Promise.withResolvers();
     var cleanupFuncs = [];
     var notifyDone = false;
     function asyncPollComplete(count) {
@@ -571,7 +573,7 @@ var SyscallsLibrary = {
       dbg('asyncPollComplete', count);
 #endif
       cleanupFuncs.forEach(cb => cb());
-      __emscripten_proxy_poll_finish(ctx, arg, count);
+      resolve(count);
     }
     function makeNotifyCallback(stream, pollfd) {
       var cb = (flags) => {
@@ -595,17 +597,18 @@ var SyscallsLibrary = {
       return cb;
     }
 
-    if (ctx) {
+    if (isAsyncContext) {
 #if RUNTIME_DEBUG
       dbg('async poll start');
 #endif
       if (timeout > 0) {
-        setTimeout(() => {
+        var t = setTimeout(() => {
 #if RUNTIME_DEBUG
-          dbg('poll: timeout');
+          dbg('poll: timeout', timeout);
 #endif
           asyncPollComplete(0);
         }, timeout);
+        cleanupFuncs.push(() => clearTimeout(t));
       }
     }
 #endif
@@ -619,8 +622,8 @@ var SyscallsLibrary = {
       var stream = FS.getStream(fd);
       if (stream) {
         if (stream.stream_ops.poll) {
-#if PTHREADS
-          if (ctx && timeout) {
+#if PTHREADS || JSPI
+          if (isAsyncContext && timeout) {
             flags = stream.stream_ops.poll(stream, timeout, makeNotifyCallback(stream, pollfd));
           } else
 #endif
@@ -634,12 +637,12 @@ var SyscallsLibrary = {
       {{{ makeSetValue('pollfd', C_STRUCTS.pollfd.revents, 'flags', 'i16') }}};
     }
 
-#if PTHREADS
-    if (ctx) {
+#if PTHREADS || JSPI
+    if (isAsyncContext) {
       if (count || !timeout) {
         asyncPollComplete(count);
       }
-      return 0;
+      return promise;
     }
 #endif
 

system/lib/libc/dynlink.c

@@ -83,8 +83,6 @@ static thread_local struct dlevent* thread_local_tail = &main_event;
 static pthread_mutex_t write_lock = PTHREAD_MUTEX_INITIALIZER;
 static thread_local bool skip_dlsync = false;
 
-static void dlsync();
-
 static void do_write_lock() {
   // Once we have the lock we want to avoid automatic code sync as that would
   // result in a deadlock.
@@ -161,7 +159,7 @@ static void load_library_done(struct dso* p) {
   new_dlevent(p, -1);
 #ifdef _REENTRANT
   // Block until all other threads have loaded this module.
-  dlsync();
+  _dlsync();
 #endif
   // TODO: figure out some way to tell when its safe to free p->file_data.  Its
   // not safe to do here because some threads could have been alseep then when
@@ -431,35 +429,6 @@ int _emscripten_proxy_dlsync(pthread_t target_thread) {
   }
   return result;
 }
-
-static void done_sync_all(em_proxying_ctx* ctx) {
-  dbg("done_sync_all");
-  emscripten_proxy_finish(ctx);
-}
-
-static void run_dlsync_async(em_proxying_ctx* ctx, void* arg) {
-  pthread_t calling_thread = (pthread_t)arg;
-  dbg("main_thread_dlsync calling=%p", calling_thread);
-  _emscripten_dlsync_threads_async(calling_thread, done_sync_all, ctx);
-}
-
-static void dlsync() {
-  // Call dlsync process.  This call will block until all threads are in sync.
-  // This gets called after a shared library is loaded by a worker.
-  dbg("dlsync main=%p", emscripten_main_runtime_thread_id());
-  if (emscripten_is_main_runtime_thread()) {
-    // dlsync was called on the main thread.  In this case we have no choice by
-    // to run the blocking version of emscripten_dlsync_threads.
-    _emscripten_dlsync_threads();
-  } else {
-    // Otherwise we block here while the asynchronous version runs in the main
-    // thread.
-    em_proxying_queue* q = emscripten_proxy_get_system_queue();
-    int success = emscripten_proxy_sync_with_ctx(
-      q, emscripten_main_runtime_thread_id(), run_dlsync_async, pthread_self());
-    assert(success);
-  }
-}
 #endif // _REENTRANT
 
 static void dlopen_onsuccess(struct dso* dso, void* user_data) {
@@ -683,7 +652,7 @@ void* __dlsym(void* restrict p, const char* restrict s, void* restrict ra) {
     new_dlevent(p, sym_index);
 #ifdef _REENTRANT
     // Block until all other threads have loaded this module.
-    dlsync();
+    _dlsync();
 #endif
   }
   dbg("__dlsym done dso:%p res:%p", p, res);

system/lib/libc/emscripten_internal.h

@@ -94,18 +94,7 @@ void* _dlsym_catchup_js(struct dso* handle, int sym_index);
 
 int _setitimer_js(int which, double timeout);
 
-// Synchronous version of "dlsync_threads".  Called only on the main thread.
-// Runs _emscripten_dlsync_self on each of the threads that are running at
-// the time of the call.
-void _emscripten_dlsync_threads();
-
-// Asynchronous version of "dlsync_threads".  Called only on the main thread.
-// Runs _emscripten_dlsync_self on each of the 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_dlsync_threads_async(pthread_t calling_thread,
-                                      void (*callback)(em_proxying_ctx*),
-                                      em_proxying_ctx* ctx);
+void _dlsync();
 
 #ifdef _GNU_SOURCE
 void __call_sighandler(sighandler_t handler, int sig);