select: enable to handle timeout

This commit enables the select syscall to handle timeout with multiple event
sources when PROXY_TO_PTHREAD is enabled.

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/libsyscall.js

@@ -607,12 +607,30 @@ var SyscallsLibrary = {
     FS.chdir(stream.path);
     return 0;
   },
+#if PROXY_TO_PTHREAD
+  __syscall__newselect__proxy: 'none',
+  __syscall__newselect__deps: ['_emscripten_proxy_newselect'],
   __syscall__newselect: (nfds, readfds, writefds, exceptfds, timeout) => {
+      return __emscripten_proxy_newselect(nfds, readfds, writefds, exceptfds, timeout);
+  },
+#else
+  __syscall__newselect__deps: ['__syscall__newselect_inner'],
+  __syscall__newselect: (nfds, readfds, writefds, exceptfds, timeout) => {
+      return ___syscall__newselect_inner(null, null, nfds, readfds, writefds, exceptfds, timeout);
+  },
+#endif
+#if PROXY_TO_PTHREAD
+  __syscall__newselect_inner__deps: ['_emscripten_proxy_newselect_finish'],
+#endif
+  __syscall__newselect_inner__proxy: 'none',
+  __syscall__newselect_inner__sig: 'ippipppp',
+  __syscall__newselect_inner: (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 PROXY_TO_PTHREAD is enabled.
 #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
@@ -633,6 +651,32 @@ var SyscallsLibrary = {
       timeoutInMillis = SYSCALLS.getTimeoutInMillis(timeout);
     }
 
+#if PROXY_TO_PTHREAD
+    var cleanupFuncs = [];
+    var notifyDone = false;
+    var 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))) {
@@ -644,14 +688,23 @@ var SyscallsLibrary = {
       var flags = SYSCALLS.DEFAULT_POLLMASK;
 
       if (stream.stream_ops.poll) {
+#if PROXY_TO_PTHREAD
+        flags = stream.stream_ops.poll(stream, timeoutInMillis, timeoutInMillis != 0 ? makeNotifyCallback(fd) : null);
+#else
         flags = stream.stream_ops.poll(stream, ((timeoutInMillis < 0) || readfds) ? timeoutInMillis : 0);
+#endif
       }
 
       fdSet.setFlags(fd, flags);
     }
 
-
-    fdSet.commit(fd, flags);
+#if PROXY_TO_PTHREAD
+    if ((fdSet.getTotal() > 0) || (timeoutInMillis == 0) ) {
+      makeNotifyCallback(-1)(0);
+    }
+#else
+    fdSet.commit();
+#endif
 
     return fdSet.getTotal();
   },

system/lib/pthread/proxying.c

@@ -658,3 +658,36 @@ double _emscripten_run_js_on_main_thread(int func_index,
   }
   return 0;
 }
+
+typedef struct proxied_select_t {
+  int n;
+  void *rfds;
+  void *wfds;
+  void *efds;
+  void *tv;
+  int result;
+} proxied_select_t;
+
+int __syscall__newselect_inner(em_proxying_ctx* ctx, void* arg, int n, void *rfds, void *wfds, void *efds, void *tv);
+
+static void call_newselect(em_proxying_ctx* ctx, void* arg) {
+  proxied_select_t* t = arg;
+  __syscall__newselect_inner(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;
+}

tools/emscripten.py

@@ -1106,6 +1106,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_',