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
a shared memory region. 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 <[email protected]>

Author: ktock

src/lib/libpthread.js

@@ -88,6 +88,11 @@ var LibraryPThread = {
                    'exit',
 #if PTHREADS_DEBUG || ASSERTIONS
                    '$ptrToString',
+#endif
+#if PROXY_TO_PTHREAD
+                   '$addThreadToActiveSelectCallbacks',
+                   '$removeThreadFromActiveSelectCallbacks',
+                   '$activeSelectCallbacks',
 #endif
                    ],
   $PThread: {
@@ -577,6 +582,9 @@ var LibraryPThread = {
 #endif
 #if ASSERTIONS
     assert(worker);
+#endif
+#if PROXY_TO_PTHREAD
+    removeThreadFromActiveSelectCallbacks(pthread_ptr);
 #endif
     PThread.returnWorkerToPool(worker);
   },
@@ -621,6 +629,49 @@ var LibraryPThread = {
   $registerTLSInit: (tlsInitFunc) => PThread.tlsInitFunctions.push(tlsInitFunc),
 #endif
 
+#if PROXY_TO_PTHREAD
+  // On the main worker, activeSelectCallbacks records the set of callbacks
+  // that are allowed to update the shared region. Any callback not in this
+  // set (i.e. when !isActiveSelectCallback) must not update the region.
+  //
+  // 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 region.
+  // However the thread worker will not read that value and just ignore it.
+  //
+  // activeSelectCallbacks records multiple callback lists one per thread
+  // worker so that each worker can manage its own set of active callbacks
+  // independently.
+  $activeSelectCallbacks: {},
+  $addThreadToActiveSelectCallbacks__deps: ['malloc'],
+  $addThreadToActiveSelectCallbacks: (pthread_ptr) => {
+    activeSelectCallbacks[pthread_ptr] = {
+      buf: _malloc(8),
+      callbacks: [],
+    };
+  },
+  $removeThreadFromActiveSelectCallbacks: (pthread_ptr) => {
+    delete activeSelectCallbacks[pthread_ptr];
+  },
+  $getActiveSelectCallbacks: (pthread_ptr) => {
+    return activeSelectCallbacks[pthread_ptr];
+  },
+  $deactivateSelectCallbacks: (pthread_ptr) => {
+    activeSelectCallbacks[pthread_ptr].callbacks = [];
+  },
+  $activateSelectCallback: (pthread_ptr, cb) => {
+    activeSelectCallbacks[pthread_ptr].callbacks.push(cb);
+  },
+  $isActiveSelectCallback: (pthread_ptr, cb) => {
+    return (activeSelectCallbacks[pthread_ptr] != null) &&
+      (activeSelectCallbacks[pthread_ptr].callbacks.indexOf(cb) != -1);
+  },
+#endif
+
   $spawnThread: (threadParams) => {
 #if ASSERTIONS
     assert(!ENVIRONMENT_IS_PTHREAD, 'Internal Error! spawnThread() can only ever be called from main application thread!');
@@ -648,6 +699,9 @@ var LibraryPThread = {
         arg: threadParams.arg,
         pthread_ptr: threadParams.pthread_ptr,
     };
+#if PROXY_TO_PTHREAD
+    addThreadToActiveSelectCallbacks(threadParams.pthread_ptr);
+#endif
 #if OFFSCREENCANVAS_SUPPORT
     // Note that we do not need to quote these names because they are only used
     // in this file, and not from the external worker.js.

src/lib/libsyscall.js

@@ -607,7 +607,35 @@ 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
+    var waitPtr = _malloc(8);
+    var result = newselectInner(nfds, readfds, writefds, exceptfds, timeout, waitPtr);
+    if ((result != 0) || ((timeout) && (SYSCALLS.getTimeoutInMillis(timeout) == 0))) {
+      _free(waitPtr);
+      return result;
+    }
+    var fdRegion = {{{ makeGetValue('waitPtr', 0, '*') }}};
+    Atomics.wait(HEAP32 , fdRegion >> 2, -1);
+    var fd = Atomics.load(HEAP32 , fdRegion >> 2);
+    var flags = Atomics.load(HEAP32 , fdRegion >> 2 + 1);
+    _free(waitPtr);
+    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);
+#endif
+  },
+#if PROXY_TO_PTHREAD
+  $newselectInner__deps: ['$PThread', '$deactivateSelectCallbacks', '$getActiveSelectCallbacks', '$activateSelectCallback', '$isActiveSelectCallback'],
+#endif
+  $newselectInner__proxy: 'sync',
+  $newselectInner: (nfds, readfds, writefds, exceptfds, timeout, waitPtr) => {
     // readfds are supported,
     // writefds checks socket open status
     // exceptfds are supported, although on web, such exceptional conditions never arise in web sockets
@@ -633,6 +661,34 @@ var SyscallsLibrary = {
       timeoutInMillis = SYSCALLS.getTimeoutInMillis(timeout);
     }
 
+#if PROXY_TO_PTHREAD
+    const pthread_ptr = PThread.currentProxiedOperationCallerThread;
+    deactivateSelectCallbacks(pthread_ptr); // deactivate all old callbacks
+    var makeNotifyCallback = (fd) => null;
+    if (timeoutInMillis != 0) {
+      var info = getActiveSelectCallbacks(pthread_ptr);
+      {{{ makeSetValue('waitPtr', 0, 'info.buf', '*') }}};
+      Atomics.store(HEAP32, info.buf >> 2, -1); // Initialize the shared region
+      makeNotifyCallback = (fd) => {
+        var cb = (flags) => {
+          if (!isActiveSelectCallback(pthread_ptr, cb)) {
+            return; // This callback is no longer active.
+          }
+          deactivateSelectCallbacks(pthread_ptr); // Only the first event is notified.
+          Atomics.store(HEAP32, info.buf >> 2 + 1, flags);
+          Atomics.store(HEAP32, info.buf >> 2, fd);
+          Atomics.notify(HEAP32, info.buf >> 2);
+        }
+        activateSelectCallback(pthread_ptr, 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 +700,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.
+      deactivateSelectCallbacks(pthread_ptr);
+    }
+#else
     fdSet.commit(fd, flags);
+#endif
 
     return fdSet.getTotal();
   },