Fix select didn't work with pipes and the timeout argument (#25523)

This PR fixes an issue that the timeout argument for the select syscall
wasn't applied to pipes and it returned immedeately instead of waiting
for events.

Although the select syscall passes the timeout value to the `poll`
method of the stream implementations, this can't handle multiple fds
because a fd can't notify readiness while another is blocking in `poll`.
As a result, using the select syscall with a combination of pipes and
other streams (e.g. PTY) can be problematic.

To address this, this PR implements a callback-based event notification.
Each stream implementation's `poll` invocation receives a callback,
allowing it asynchronously notify the select syscall when it becomes
ready. The select syscall blocks until one of these callbacks is
triggered or the timeout expires. This behviour is enabled only when
PROXY_TO_PTHREAD is enabled to avoid blocking the main worker.

To maintain compatibility with non-pipefs streams, the select syscall
allows stream implementations to ignore the callback and synchronously
return the event status instead. In such cases, the select syscall still
updates the flags accordingly.

Author: ktock

src/lib/libpipefs.js

@@ -21,6 +21,23 @@ addToLibrary({
         // able to read from the read end after write end is closed.
         refcnt : 2,
         timestamp: new Date(),
+#if PTHREADS
+        readableHandlers: [],
+        registerReadableHandler: (callback) => {
+          callback.registerCleanupFunc(() => {
+            const i = pipe.readableHandlers.indexOf(callback);
+            if (i !== -1) pipe.readableHandlers.splice(i, 1);
+          });
+          pipe.readableHandlers.push(callback);
+        },
+        notifyReadableHandlers: () => {
+          while (pipe.readableHandlers.length > 0) {
+            const cb = pipe.readableHandlers.shift();
+            if (cb) cb({{{ cDefs.POLLRDNORM }}} | {{{ cDefs.POLLIN }}});
+          }
+          pipe.readableHandlers = [];
+        }
+#endif
       };
 
       pipe.buckets.push({
@@ -80,7 +97,7 @@ addToLibrary({
           blocks: 0,
         };
       },
-      poll(stream) {
+      poll(stream, timeout, notifyCallback) {
         var pipe = stream.node.pipe;
 
         if ((stream.flags & {{{ cDefs.O_ACCMODE }}}) === {{{ cDefs.O_WRONLY }}}) {
@@ -92,6 +109,9 @@ addToLibrary({
           }
         }
 
+#if PTHREADS
+        if (notifyCallback) pipe.registerReadableHandler(notifyCallback);
+#endif
         return 0;
       },
       dup(stream) {
@@ -204,6 +224,9 @@ addToLibrary({
         if (freeBytesInCurrBuffer >= dataLen) {
           currBucket.buffer.set(data, currBucket.offset);
           currBucket.offset += dataLen;
+#if PTHREADS
+          pipe.notifyReadableHandlers();
+#endif
           return dataLen;
         } else if (freeBytesInCurrBuffer > 0) {
           currBucket.buffer.set(data.subarray(0, freeBytesInCurrBuffer), currBucket.offset);
@@ -235,6 +258,9 @@ addToLibrary({
           newBucket.buffer.set(data);
         }
 
+#if PTHREADS
+        pipe.notifyReadableHandlers();
+#endif
         return dataLen;
       },
       close(stream) {

src/lib/libsigs.js

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

src/lib/libsyscall.js

@@ -104,6 +104,63 @@ var SyscallsLibrary = {
     },
   },
 
+  $parseSelectFDSet__internal: true,
+  $parseSelectFDSet: (readfds, writefds, exceptfds) => {
+    var total = 0;
+
+    var srcReadLow = (readfds ? {{{ makeGetValue('readfds', 0, 'i32') }}} : 0),
+        srcReadHigh = (readfds ? {{{ makeGetValue('readfds', 4, 'i32') }}} : 0);
+    var srcWriteLow = (writefds ? {{{ makeGetValue('writefds', 0, 'i32') }}} : 0),
+        srcWriteHigh = (writefds ? {{{ makeGetValue('writefds', 4, 'i32') }}} : 0);
+    var srcExceptLow = (exceptfds ? {{{ makeGetValue('exceptfds', 0, 'i32') }}} : 0),
+        srcExceptHigh = (exceptfds ? {{{ makeGetValue('exceptfds', 4, 'i32') }}} : 0);
+
+    var dstReadLow = 0,
+        dstReadHigh = 0;
+    var dstWriteLow = 0,
+        dstWriteHigh = 0;
+    var dstExceptLow = 0,
+        dstExceptHigh = 0;
+
+    var check = (fd, low, high, val) => fd < 32 ? (low & val) : (high & val);
+
+    return {
+      allLow: srcReadLow | srcWriteLow | srcExceptLow,
+      allHigh: srcReadHigh | srcWriteHigh | srcExceptHigh,
+      getTotal: () => total,
+      setFlags: (fd, flags) => {
+        var mask = 1 << (fd % 32);
+
+        if ((flags & {{{ cDefs.POLLIN }}}) && check(fd, srcReadLow, srcReadHigh, mask)) {
+          fd < 32 ? (dstReadLow = dstReadLow | mask) : (dstReadHigh = dstReadHigh | mask);
+          total++;
+        }
+        if ((flags & {{{ cDefs.POLLOUT }}}) && check(fd, srcWriteLow, srcWriteHigh, mask)) {
+          fd < 32 ? (dstWriteLow = dstWriteLow | mask) : (dstWriteHigh = dstWriteHigh | mask);
+          total++;
+        }
+        if ((flags & {{{ cDefs.POLLPRI }}}) && check(fd, srcExceptLow, srcExceptHigh, mask)) {
+          fd < 32 ? (dstExceptLow = dstExceptLow | mask) : (dstExceptHigh = dstExceptHigh | mask);
+          total++;
+        }
+      },
+      commit: () => {
+        if (readfds) {
+          {{{ makeSetValue('readfds', '0', 'dstReadLow', 'i32') }}};
+          {{{ makeSetValue('readfds', '4', 'dstReadHigh', 'i32') }}};
+        }
+        if (writefds) {
+          {{{ makeSetValue('writefds', '0', 'dstWriteLow', 'i32') }}};
+          {{{ makeSetValue('writefds', '4', 'dstWriteHigh', 'i32') }}};
+        }
+        if (exceptfds) {
+          {{{ makeSetValue('exceptfds', '0', 'dstExceptLow', 'i32') }}};
+          {{{ makeSetValue('exceptfds', '4', 'dstExceptHigh', 'i32') }}};
+        }
+      }
+    };
+  },
+
   $syscallGetVarargI__internal: true,
   $syscallGetVarargI: () => {
 #if ASSERTIONS
@@ -543,37 +600,88 @@ var SyscallsLibrary = {
     return 0;
   },
   __syscall__newselect__i53abi: true,
+  __syscall__newselect__proxy: 'none',
+  __syscall__newselect__deps: ['_newselect_js',
+#if PTHREADS
+    '_emscripten_proxy_newselect',
+#endif
+  ],
   __syscall__newselect: (nfds, readfds, writefds, exceptfds, timeoutInMillis) => {
+#if PTHREADS
+    if (ENVIRONMENT_IS_PTHREAD) {
+      return __emscripten_proxy_newselect(nfds,
+                                          {{{ to64('readfds') }}},
+                                          {{{ to64('writefds') }}},
+                                          {{{ to64('exceptfds') }}},
+                                          {{{ splitI64('timeoutInMillis') }}});
+    }
+#endif
+      return __newselect_js({{{ to64('0') }}},
+                            {{{ to64('0') }}},
+                            nfds,
+                            {{{ to64('readfds') }}},
+                            {{{ to64('writefds') }}},
+                            {{{ to64('exceptfds') }}},
+                            {{{ splitI64('timeoutInMillis') }}});
+  },
+  _newselect_js__i53abi: true,
+  _newselect_js__proxy: 'none',
+  _newselect_js__deps: ['$parseSelectFDSet',
+#if PTHREADS
+    '_emscripten_proxy_newselect_finish',
+#endif
+  ],
+  _newselect_js: (ctx, arg, nfds, readfds, writefds, exceptfds, timeoutInMillis) => {
     // 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
+#if PTHREADS
+    assert(!ENVIRONMENT_IS_PTHREAD, '_newselect_js must be called in the main thread');
+#endif
 #endif
 
-    var total = 0;
-
-    var srcReadLow = (readfds ? {{{ makeGetValue('readfds', 0, 'i32') }}} : 0),
-        srcReadHigh = (readfds ? {{{ makeGetValue('readfds', 4, 'i32') }}} : 0);
-    var srcWriteLow = (writefds ? {{{ makeGetValue('writefds', 0, 'i32') }}} : 0),
-        srcWriteHigh = (writefds ? {{{ makeGetValue('writefds', 4, 'i32') }}} : 0);
-    var srcExceptLow = (exceptfds ? {{{ makeGetValue('exceptfds', 0, 'i32') }}} : 0),
-        srcExceptHigh = (exceptfds ? {{{ makeGetValue('exceptfds', 4, 'i32') }}} : 0);
-
-    var dstReadLow = 0,
-        dstReadHigh = 0;
-    var dstWriteLow = 0,
-        dstWriteHigh = 0;
-    var dstExceptLow = 0,
-        dstExceptHigh = 0;
+    var fdSet = parseSelectFDSet(readfds, writefds, exceptfds);
 
-    var allLow = srcReadLow | srcWriteLow | srcExceptLow;
-    var allHigh = srcReadHigh | srcWriteHigh | srcExceptHigh;
+    var allLow = fdSet.allLow;
+    var allHigh = fdSet.allHigh;
 
     var check = (fd, low, high, val) => fd < 32 ? (low & val) : (high & val);
 
+#if PTHREADS
+    var makeNotifyCallback = null;
+    if (ctx) {
+      // 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))) {
@@ -585,41 +693,35 @@ var SyscallsLibrary = {
       var flags = SYSCALLS.DEFAULT_POLLMASK;
 
       if (stream.stream_ops.poll) {
-        flags = stream.stream_ops.poll(stream, timeoutInMillis);
+        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);
+        })();
       } else {
 #if ASSERTIONS
         if (timeoutInMillis != 0) warnOnce('non-zero select() timeout not supported: ' + timeoutInMillis)
 #endif
       }
 
-      if ((flags & {{{ cDefs.POLLIN }}}) && check(fd, srcReadLow, srcReadHigh, mask)) {
-        fd < 32 ? (dstReadLow = dstReadLow | mask) : (dstReadHigh = dstReadHigh | mask);
-        total++;
-      }
-      if ((flags & {{{ cDefs.POLLOUT }}}) && check(fd, srcWriteLow, srcWriteHigh, mask)) {
-        fd < 32 ? (dstWriteLow = dstWriteLow | mask) : (dstWriteHigh = dstWriteHigh | mask);
-        total++;
-      }
-      if ((flags & {{{ cDefs.POLLPRI }}}) && check(fd, srcExceptLow, srcExceptHigh, mask)) {
-        fd < 32 ? (dstExceptLow = dstExceptLow | mask) : (dstExceptHigh = dstExceptHigh | mask);
-        total++;
-      }
+      fdSet.setFlags(fd, flags);
     }
 
-    if (readfds) {
-      {{{ makeSetValue('readfds', '0', 'dstReadLow', 'i32') }}};
-      {{{ makeSetValue('readfds', '4', 'dstReadHigh', 'i32') }}};
-    }
-    if (writefds) {
-      {{{ makeSetValue('writefds', '0', 'dstWriteLow', 'i32') }}};
-      {{{ makeSetValue('writefds', '4', 'dstWriteHigh', 'i32') }}};
-    }
-    if (exceptfds) {
-      {{{ makeSetValue('exceptfds', '0', 'dstExceptLow', 'i32') }}};
-      {{{ makeSetValue('exceptfds', '4', 'dstExceptHigh', 'i32') }}};
+#if PTHREADS
+    if (makeNotifyCallback != null) {
+      if ((fdSet.getTotal() > 0) || (timeoutInMillis == 0) ) {
+        makeNotifyCallback(-1)(0);
+      }
+      return 0;
     }
+#endif
+
+    fdSet.commit();
 
-    return total;
+    return fdSet.getTotal();
   },
   _msync_js__i53abi: true,
   _msync_js: (addr, len, prot, flags, fd, offset) => {

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, int64_t timeout);
+
 #ifdef __cplusplus
 }
 #endif

system/lib/libc/proxying_select.c

@@ -0,0 +1,43 @@
+/*
+ * Copyright 2025 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;
+  int64_t timeout;
+  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->timeout);
+}
+
+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, int64_t timeout) {
+  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, .timeout = timeout};
+  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, int64_t timeout) {
+  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

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