select: Enable timeout on select when PROXY_TO_PTHREAD

This commit enables the select syscall to handle timeout with multiple event
sources. PROXY_TO_PTHREAD is needed to prevent blocking the main worker.

When a thread worker invokes the select syscall with non-zero timeout and no
fd is ready, it blocks using Atmoics.wait until it receives a readiness
notification. On the main worker, the underlying stream implementation can
trigger readiness using Atomics.notify through a callback. A notification
also issued automatically once the specified timeout expires.

Communication between the thread worker and the main worker occurs via
shared memory regions. To prevent a select invocation being unblocked by the
callbacks of a previous invocation, all active callbacks are tracked in a
list. See the comments in activeSelectCallbacks for details.

Usage of the notification callback is optional. If the stream implementation
doesn't support it, the "poll" method can still synchronously return the
event status.

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

Author: ktock

src/lib/libsyscall.js

@@ -167,6 +167,43 @@ var SyscallsLibrary = {
         }
       };
     },
+
+#if PROXY_TO_PTHREAD
+    // Pointers to regions which will be shared to the main worker for
+    // notification purpose.
+    selectFDRegion: undefined,
+    selectFlagsRegion: undefined,
+
+    // On the main worker, activeSelectCallbacks records the set of callbacks
+    // that are allowed to update the shared regions and notify the thread
+    // worker. Any callback not in this set (i.e. when !isActiveSelectCallback)
+    // must not update the regions.
+    //
+    // Each select syscall invocation must call deactivateSelectCallbacks to
+    // reset this set, ensuring that callbacks from previous invocations don't
+    // affect the current one.
+    //
+    // If a callback executes after the thread worker has already returned (due
+    // to a timeout, a readiness notification or other exceptional conditions)
+    // but before the next deactivation, it may still update the shared regions.
+    // However the thread worker will no longer read from those regions and will
+    // ignore such updates.
+    //
+    // activeSelectCallbacks records multiple callback lists one per thread
+    // worker so that each worker can manage its own set of active callbacks
+    // independently.
+    activeSelectCallbacks: {},
+    activateSelectCallback(i, cb) {
+      if (SYSCALLS.activeSelectCallbacks[i] == null) SYSCALLS.activeSelectCallbacks[i] = [];
+      SYSCALLS.activeSelectCallbacks[i].push(cb);
+    },
+    deactivateSelectCallbacks(i) {
+      SYSCALLS.activeSelectCallbacks[i] = [];
+    },
+    isActiveSelectCallback(i, cb) {
+      return (SYSCALLS.activeSelectCallbacks[i] != null) && (SYSCALLS.activeSelectCallbacks[i].indexOf(cb) != -1);
+    },
+#endif
   },
 
   $syscallGetVarargI__internal: true,
@@ -607,7 +644,36 @@ var SyscallsLibrary = {
     FS.chdir(stream.path);
     return 0;
   },
+  __syscall__newselect__deps: ['$newselectInner','malloc','free'],
+  __syscall__newselect__proxy: 'none',
   __syscall__newselect: (nfds, readfds, writefds, exceptfds, timeout) => {
+#if PROXY_TO_PTHREAD
+    if (SYSCALLS.selectFDRegion == undefined) {
+      SYSCALLS.selectFDRegion = _malloc(4) >> 2;
+      SYSCALLS.selectFlagsRegion = _malloc(4) >> 2;
+    }
+    var fdIdx = SYSCALLS.selectFDRegion;
+    var flagsIdx = SYSCALLS.selectFlagsRegion;
+
+    var result = newselectInner(nfds, readfds, writefds, exceptfds, timeout, fdIdx, flagsIdx);
+    if ((result != 0) || ((timeout) && (SYSCALLS.getTimeoutInMillis(timeout) == 0))) {
+      return result;
+    }
+
+    Atomics.wait(HEAP32 , fdIdx, -1);
+    var fd = HEAP32[fdIdx];
+    var flags = HEAP32[flagsIdx];
+    if (fd < 0) return 0;
+    var fdSet = SYSCALLS.parseSelectFDSet(readfds, writefds, exceptfds);
+    fdSet.setFlags(fd, flags);
+    fdSet.commit();
+    return fdSet.getTotal();
+#else
+    return newselectInner(nfds, readfds, writefds, exceptfds, timeout, -1, -1);
+#endif
+  },
+  $newselectInner__proxy: 'sync',
+  $newselectInner: (nfds, readfds, writefds, exceptfds, timeout, fdIdx, flagsIdx) => {
     // readfds are supported,
     // writefds checks socket open status
     // exceptfds are supported, although on web, such exceptional conditions never arise in web sockets
@@ -633,6 +699,32 @@ var SyscallsLibrary = {
       timeoutInMillis = SYSCALLS.getTimeoutInMillis(timeout);
     }
 
+#if PROXY_TO_PTHREAD
+    SYSCALLS.deactivateSelectCallbacks(fdIdx); // deactivate all old callbacks
+    var makeNotifyCallback = (fd) => null;
+    if (timeoutInMillis != 0) {
+      Atomics.store(HEAP32, fdIdx, -1); // Initialize the shared region
+      Atomics.store(HEAP32, flagsIdx, -1); // Initialize the shared region
+      makeNotifyCallback = (fd) => {
+        var cb = (flags) => {
+          if (!SYSCALLS.isActiveSelectCallback(fdIdx, cb)) {
+            return; // This callback is no longer active.
+          }
+          SYSCALLS.deactivateSelectCallbacks(fdIdx); // Only the first event is notified.
+          Atomics.store(HEAP32, flagsIdx, flags);
+          Atomics.store(HEAP32, fdIdx, fd);
+          Atomics.notify(HEAP32, fdIdx);
+        }
+        SYSCALLS.activateSelectCallback(fdIdx, cb);
+        return cb;
+      }
+      if (timeoutInMillis > 0) {
+        var cb = makeNotifyCallback(-2);
+        setTimeout(() => cb(0), timeoutInMillis);
+      }
+    }
+#endif
+
     for (var fd = 0; fd < nfds; fd++) {
       var mask = 1 << (fd % 32);
       if (!(check(fd, allLow, allHigh, mask))) {
@@ -644,14 +736,26 @@ var SyscallsLibrary = {
       var flags = SYSCALLS.DEFAULT_POLLMASK;
 
       if (stream.stream_ops.poll) {
+#if PROXY_TO_PTHREAD
+        flags = stream.stream_ops.poll(stream, timeoutInMillis, makeNotifyCallback(fd));
+#else
         flags = stream.stream_ops.poll(stream, ((timeoutInMillis < 0) || readfds) ? timeoutInMillis : 0);
+#endif
       }
 
       fdSet.setFlags(fd, flags);
     }
 
 
+#if PROXY_TO_PTHREAD
+    if ((fdSet.getTotal() > 0) || (timeoutInMillis == 0) ) {
+      fdSet.commit(fd, flags);
+      // No wait will happen in the caller. Deactivate all callbacks.
+      SYSCALLS.deactivateSelectCallbacks(fdIdx);
+    }
+#else
     fdSet.commit(fd, flags);
+#endif
 
     return fdSet.getTotal();
   },