select: enable to handle timeout

This commit enables the select syscall to handle timeout with multiple event
sources when it is called from a worker.

When a thread worker calls __syscall_newselect, it blocks using
emscripten_proxy_sync_with_ctx. When __syscall_newselect is called with zero
timeout, it is unblocked immediately by calling emscripten_proxy_finish
before returning. When it is called with non-zero timeout,
emscripten_proxy_finish is invoked either by the underlying stream
implementation (where an event occurs) or by a setTimeout callback when the
tiemout expires.

In proxying.c, several wrapper functions for proxying-related APIs are added
to allow the JS implementation of newselect to use them.

Signed-off-by: Kohei Tokunaga <ktokunaga.mail@gmail.com>

Author: ktock

src/lib/libsigs.js

@@ -381,6 +381,7 @@ sigs = {
   _mmap_js__sig: 'ipiiijpp',
   _msync_js__sig: 'ippiiij',
   _munmap_js__sig: 'ippiiij',
+  _newselect_js__sig: 'ippipppp',
   _setitimer_js__sig: 'iid',
   _timegm_js__sig: 'jp',
   _tzset_js__sig: 'vpppp',

src/lib/libsyscall.js

@@ -609,12 +609,31 @@ var SyscallsLibrary = {
     FS.chdir(stream.path);
     return 0;
   },
+  __syscall__newselect__proxy: 'none',
+  __syscall__newselect__deps: ['_newselect_js',
+#if PTHREADS
+    '_emscripten_proxy_newselect',
+#endif
+  ],
   __syscall__newselect: (nfds, readfds, writefds, exceptfds, timeout) => {
+#if PTHREADS
+    if (ENVIRONMENT_IS_PTHREAD) {
+      return __emscripten_proxy_newselect(nfds, readfds, writefds, exceptfds, timeout);
+    }
+#endif
+    return __newselect_js(null, null, nfds, readfds, writefds, exceptfds, timeout);
+  },
+  _newselect_js__proxy: 'none',
+#if PTHREADS
+  _newselect_js__deps: ['_emscripten_proxy_newselect_finish'],
+#endif
+  _newselect_js: (ctx, arg, nfds, readfds, writefds, exceptfds, timeout) => {
     // readfds are supported,
     // writefds checks socket open status
     // exceptfds are supported, although on web, such exceptional conditions never arise in web sockets
     //                          and so the exceptfds list will always return empty.
-    // timeout is supported, although on SOCKFS and PIPEFS these are ignored and always treated as 0 - fully async
+    // timeout is supported, although on SOCKFS these are ignored and always treated as 0 - fully async
+    // and PIPEFS supports timeout only when the select is called from a worker.
 #if ASSERTIONS
     assert(nfds <= 64, 'nfds must be less than or equal to 64');  // fd sets have 64 bits // TODO: this could be 1024 based on current musl headers
 #endif
@@ -631,6 +650,36 @@ var SyscallsLibrary = {
       timeoutInMillis = SYSCALLS.getTimeoutInMillis(timeout);
     }
 
+#if PTHREADS
+    var makeNotifyCallback = null;
+    if (ctx != null) {
+      // Enable event handlers only when the select call is proxied from a worker.
+      var cleanupFuncs = [];
+      var notifyDone = false;
+      makeNotifyCallback = (fd) => {
+          var cb = (flags) => {
+              if (notifyDone) {
+                  return;
+              }
+              if (fd >= 0) {
+                  fdSet.setFlags(fd, flags);
+              }
+              notifyDone = true;
+              cleanupFuncs.forEach(cb => cb());
+              fdSet.commit();
+              __emscripten_proxy_newselect_finish({{{ to64('ctx') }}}, {{{ to64('arg') }}}, fdSet.getTotal());
+          }
+          cb.registerCleanupFunc = (f) => {
+              if (f != null) cleanupFuncs.push(f);
+          }
+          return cb;
+      }
+      if (timeoutInMillis > 0) {
+          setTimeout(() => makeNotifyCallback(-1)(0), timeoutInMillis);
+      }
+    }
+#endif
+
     for (var fd = 0; fd < nfds; fd++) {
       var mask = 1 << (fd % 32);
       if (!(check(fd, allLow, allHigh, mask))) {
@@ -642,12 +691,27 @@ var SyscallsLibrary = {
       var flags = SYSCALLS.DEFAULT_POLLMASK;
 
       if (stream.stream_ops.poll) {
-        flags = stream.stream_ops.poll(stream, ((timeoutInMillis < 0) || readfds) ? timeoutInMillis : 0);
+        flags = (() => {
+#if PTHREADS
+          if (makeNotifyCallback != null) {
+            return stream.stream_ops.poll(stream, timeoutInMillis, timeoutInMillis != 0 ? makeNotifyCallback(fd) : null);
+          }
+#endif
+          return stream.stream_ops.poll(stream, ((timeoutInMillis < 0) || readfds) ? timeoutInMillis : 0);
+        })();
       }
 
       fdSet.setFlags(fd, flags);
     }
 
+#if PTHREADS
+    if (makeNotifyCallback != null) {
+      if ((fdSet.getTotal() > 0) || (timeoutInMillis == 0) ) {
+        makeNotifyCallback(-1)(0);
+      }
+      return 0;
+    }
+#endif
 
     fdSet.commit();
 

system/lib/libc/emscripten_internal.h

@@ -151,6 +151,8 @@ EmscriptenDeviceOrientationEvent* _emscripten_get_last_deviceorientation_event()
 EmscriptenDeviceMotionEvent* _emscripten_get_last_devicemotion_event();
 EmscriptenMouseEvent* _emscripten_get_last_mouse_event();
 
+int _newselect_js(void* ctx, void* arg, int n, void *rfds, void *wfds, void *efds, void *tv);
+
 #ifdef __cplusplus
 }
 #endif

system/lib/libc/proxying_select.c

@@ -0,0 +1,43 @@
+/*
+ * Copyright 2021 The Emscripten Authors.  All rights reserved.
+ * Emscripten is available under two separate licenses, the MIT license and the
+ * University of Illinois/NCSA Open Source License.  Both these licenses can be
+ * found in the LICENSE file.
+ */
+
+#include <assert.h>
+#include <emscripten/proxying.h>
+#include <emscripten/threading.h>
+
+#include "emscripten_internal.h"
+
+typedef struct proxied_select_t {
+  int n;
+  void *rfds;
+  void *wfds;
+  void *efds;
+  void *tv;
+  int result;
+} proxied_select_t;
+
+static void call_newselect(em_proxying_ctx* ctx, void* arg) {
+  proxied_select_t* t = arg;
+  _newselect_js(ctx, arg, t->n, t->rfds, t->wfds, t->efds, t->tv);
+}
+
+void _emscripten_proxy_newselect_finish(em_proxying_ctx* ctx, void* arg, int ret) {
+  proxied_select_t* t = arg;
+  t->result = ret;
+  emscripten_proxy_finish(ctx);
+}
+
+int _emscripten_proxy_newselect(int n, void *rfds, void *wfds, void *efds, void *tv) {
+  em_proxying_queue* q = emscripten_proxy_get_system_queue();
+  pthread_t target = emscripten_main_runtime_thread_id();
+  proxied_select_t t = {.n = n, .rfds = rfds, .wfds = wfds, .efds = efds, .tv = tv};
+  if (!emscripten_proxy_sync_with_ctx(q, target, call_newselect, &t)) {
+    assert(false && "emscripten_proxy_sync failed");
+    return -1;
+  }
+  return t.result;
+}

system/lib/standalone/standalone.c

@@ -66,6 +66,10 @@ weak int _munmap_js(
   return -ENOSYS;
 }
 
+weak int _newselect_js(void* ctx, void* arg, int n, void *rfds, void *wfds, void *efds, void *tv) {
+  return -ENOSYS;
+}
+
 // open(), etc. - we just support the standard streams, with no
 // corner case error checking; everything else is not permitted.
 // TODO: full file support for WASI, or an option for it

system/lib/wasmfs/syscalls.cpp

@@ -1780,4 +1780,8 @@ int __syscall__newselect(int nfds,
   return -ENOMEM;
 }
 
+int _newselect_js(void* ctx, void* arg, int n, void *rfds, void *wfds, void *efds, void *tv) {
+  return -ENOSYS;
+}
+
 } // extern "C"

tools/emscripten.py

@@ -1184,6 +1184,8 @@ def create_pointer_conversion_wrappers(metadata):
     '_emscripten_dlsync_self_async': '_p',
     '_emscripten_proxy_dlsync_async': '_pp',
     '_emscripten_wasm_worker_initialize': '_p_',
+    '_emscripten_proxy_newselect': '__pppp',
+    '_emscripten_proxy_newselect_finish': '_pp_',
     '_wasmfs_rename': '_pp',
     '_wasmfs_readlink': '_pp',
     '_wasmfs_truncate': '_p_',

tools/system_libs.py

@@ -1367,6 +1367,7 @@ def get_files(self):
           'sigtimedwait.c',
           'wasi-helpers.c',
           'system.c',
+          'proxying_select.c',
         ])
 
     if settings.RELOCATABLE or settings.MAIN_MODULE: