Allow synchronous proxying of async JS functions

This new mode allows the proxied work itself to be asynchronous (i.e.
return a promise).  The new behaviour is triggered by marking a
functions as both `__proxy: 'sync'` and also `__async: true`.

We can use this new mode to replace the bespoke proxying code that
exists for `dlsync` as well as the `poll` system call.  For the poll
system call this bespoke code was added in #25523 and #25990, but can
now be completely removed.

In addition to this, because the `poll` syscall is now marked as
`__async` it automatically works under JSPI too.

One downside of this new mode is that the proxied work requires the main
thread event loop to run.  Unlike the synchronous proxied work that can
complete even when we are deeply nested.

Fixes: #25970

Author: sbc100

.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);