Support blocking accept() and recv() on JS sockets under pthreads and ASYNCIFY/JSPI

site/source/docs/porting/networking.rst

@@ -62,15 +62,18 @@ sockets API with real host TCP, UDP and ``AF_UNIX`` stream sockets (see
 Non-blocking sockets (``SOCK_NONBLOCK``, ``fcntl(F_SETFL, O_NONBLOCK)`` or
 ``ioctl(FIONBIO)``) behave as on Linux.
 
-Blocking sockets cannot actually block. An operation on a blocking socket that
-would need to wait (``accept()``, ``recv()``/``read()``) fails with ``EAGAIN``
-instead, a blocking ``connect()`` returns ``0`` before the connection has
-completed, and a blocking ``send()`` never waits: it buffers without limit
-(only a non-blocking socket is bounded by the write buffer's high-water mark
-and reports ``EAGAIN``). Builds with ``ASSERTIONS`` print a warning the first
-time a blocking socket returns ``EAGAIN``. Applications should use non-blocking
-sockets together with ``poll()`` or ``epoll``, which can wait when called from
-a pthread, or when using :ref:`ASYNCIFY`.
+Blocking ``accept()``, ``recv()``, ``recvfrom()`` and ``recvmsg()`` wait when
+called from a pthread (including ``main()`` under :ref:`PROXY_TO_PTHREAD`), or
+when using :ref:`ASYNCIFY` or :ref:`JSPI`, just like ``poll()`` and
+``epoll_wait()``; ``MSG_DONTWAIT`` still returns ``EAGAIN`` without waiting.
+Where no stack can wait (the main thread of a plain build) these fail with
+``EAGAIN`` instead, and builds with ``ASSERTIONS`` print a warning the first
+time that happens. Other blocking operations never wait: a blocking
+``connect()`` returns ``0`` before the connection has completed, and a
+blocking ``send()`` buffers without limit (only a non-blocking socket is
+bounded by the write buffer's high-water mark and reports ``EAGAIN``).
+Applications can otherwise use non-blocking sockets together with ``poll()``
+or ``epoll``.
 
 Full POSIX Sockets over WebSocket Proxy Server
 ==============================================

site/source/docs/tools_reference/settings_reference.rst

@@ -592,7 +592,9 @@ Node.js.
 It is event-driven. Socket readiness comes through the same
 ``emscripten_set_socket_*_callback`` hooks the WebSocket backend uses, so it
 works with existing readiness reactors. It cannot be combined with the
-WebSocket emulation or :ref:`PROXY_POSIX_SOCKETS`.
+WebSocket emulation or :ref:`PROXY_POSIX_SOCKETS`. Blocking ``accept()``
+and ``recv()`` wait (like ``poll()``) from a pthread or under
+:ref:`ASYNCIFY`/JSPI.
 
 It works under -pthread with :ref:`PROXY_TO_PTHREAD`, where main() and every socket
 syscall run on a single worker alongside the node handles and their event

src/lib/libsockfs_node.js

@@ -56,11 +56,7 @@ null;
 
 var NodeSockFSLibrary = {
   // Node plumbing shared by the interface methods below.
-  $nodeSockHelpers__deps: ['$SOCKFS', '$ERRNO_CODES', '$inetPton4', '$inetPton6',
-#if ASSERTIONS
-    '$warnOnce',
-#endif
-  ],
+  $nodeSockHelpers__deps: ['$SOCKFS', '$ERRNO_CODES', '$inetPton4', '$inetPton6'],
   $nodeSockHelpers: {
     // node builtins, resolved once each. getBuiltinModule works in both
     // CommonJS and ESM output, with require as the fallback.
@@ -253,16 +249,6 @@ var NodeSockFSLibrary = {
     connectInProgress(sock) {
       if (sock.stream.flags & {{{ cDefs.O_NONBLOCK }}}) throw new FS.ErrnoError({{{ cDefs.EINPROGRESS }}});
     },
-    // Operations that would block return EAGAIN even on a blocking fd, since
-    // there is no way to block here.
-    wouldBlock(sock) {
-#if ASSERTIONS
-      if (!(sock.stream.flags & {{{ cDefs.O_NONBLOCK }}})) {
-        warnOnce('NODERAWSOCKETS: a blocking socket operation would block, returning EAGAIN instead (blocking I/O is not supported, use O_NONBLOCK with poll/epoll)');
-      }
-#endif
-      return new FS.ErrnoError({{{ cDefs.EAGAIN }}});
-    },
     // The UDP backing object. With a synchronous dgram bindSync available we use
     // a public node:dgram socket (sock.udpPublic); otherwise we fall back to a
     // private udp_wrap handle, which is the only older-node way to get a
@@ -725,7 +711,7 @@ var NodeSockFSLibrary = {
         listensock.error = null;
         throw new FS.ErrnoError(e);
       }
-      if (!listensock.pending.length) throw nodeSockHelpers.wouldBlock(listensock);
+      if (!listensock.pending.length) throw new FS.ErrnoError({{{ cDefs.EAGAIN }}});
       return listensock.pending.shift();
     },
     sendmsg(sock, buffer, offset, length, addr, port) {
@@ -807,7 +793,7 @@ var NodeSockFSLibrary = {
             sock.error = null;
             throw new FS.ErrnoError(derr);
           }
-          throw nodeSockHelpers.wouldBlock(sock);
+          throw new FS.ErrnoError({{{ cDefs.EAGAIN }}});
         }
         // A datagram is atomic: return up to length bytes and drop the rest.
         var dd = dgram.data;
@@ -817,11 +803,19 @@ var NodeSockFSLibrary = {
       }
       var queued = sock.recv_queue[0];
       if (!queued) {
+        // A pending error (poll reports it readable for this) is returned and
+        // cleared here, as Linux does, rather than reporting EAGAIN forever. It
+        // takes precedence over EOF: node emits 'close' right after 'error'.
+        if (sock.error) {
+          var serr = sock.error;
+          sock.error = null;
+          throw new FS.ErrnoError(serr);
+        }
         if (sock.readClosed) return null; // EOF
         if (!sock.connection) {
           throw new FS.ErrnoError({{{ cDefs.ENOTCONN }}});
         }
-        throw nodeSockHelpers.wouldBlock(sock);
+        throw new FS.ErrnoError({{{ cDefs.EAGAIN }}});
       }
       var q = queued.data;
       var bytesRead = Math.min(length, q.length);

src/lib/libsyscall.js

@@ -410,9 +410,107 @@ var SyscallsLibrary = {
     return -{{{ cDefs.ENOSYS }}}; // unsupported feature
 #endif
   },
-  __syscall_accept4__deps: ['$getSocketFromFD', '$writeSockaddr'],
-  __syscall_accept4: (fd, addr, len, flags, u1, u2) => {
+  // Run a receive-side socket syscall body `op(sock)` that may block. The
+  // backend is strictly synchronous: `op` throws EAGAIN when it would block,
+  // whatever the fd's mode. A blocking socket (no O_NONBLOCK, no MSG_DONTWAIT)
+  // then waits for readiness where the calling stack can - a sync-proxied
+  // pthread (PROXY_SYNC_ASYNC) awaits the returned Promise, ASYNCIFY/JSPI
+  // suspends on it - and retries. Every other outcome returns synchronously (a
+  // JSPI Suspending import only suspends on a Promise), so a non-blocking call
+  // stays callable from any stack, and where no stack can wait (the event-loop
+  // thread itself) the EAGAIN surfaces unchanged.
+  $sockCall__internal: true,
+  $sockCall__deps: ['$getSocketFromFD',
+#if PTHREADS || ASYNCIFY
+    '$sockCallAsync', '$sockWouldBlock',
+#endif
+#if ASYNCIFY
+    '$Asyncify',
+#endif
+#if ASSERTIONS
+    '$warnOnce',
+#endif
+  ],
+  $sockCall: (fd, dontWait, op) => {
+#if PTHREADS
+    // A sync-proxied caller awaits a thenable even for an immediate result.
+    if (PThread.currentProxiedOperationCallerThread) return sockCallAsync(fd, dontWait, op);
+#endif
+#if ASYNCIFY == 1
+    // The rewind re-enters here after the wait has already run `op`, so it
+    // must go back through handleAsync rather than run `op` again.
+    if (Asyncify.state === Asyncify.State.Rewinding) return Asyncify.handleAsync(() => {});
+#endif
     var sock = getSocketFromFD(fd);
+#if ASYNCIFY
+    try {
+      return op(sock);
+    } catch (e) {
+      if (!sockWouldBlock(e, sock, dontWait)) throw e;
+      // handleAsync keeps the runtime alive across the suspension (EXIT_RUNTIME
+      // would otherwise tear it down from an event-loop callback while main()
+      // is parked here).
+      return Asyncify.handleAsync(() => sockCallAsync(fd, dontWait, op));
+    }
+#else
+#if ASSERTIONS
+    try {
+      return op(sock);
+    } catch (e) {
+      if (e.name === 'ErrnoError' && e.errno === {{{ cDefs.EAGAIN }}} && !dontWait && !(sock.stream.flags & {{{ cDefs.O_NONBLOCK }}})) {
+        warnOnce('a blocking socket operation would block, returning EAGAIN instead (this stack cannot block: use O_NONBLOCK with poll/epoll, or call from a pthread or with ASYNCIFY/JSPI)');
+      }
+      throw e;
+    }
+#else
+    return op(sock);
+#endif
+#endif
+  },
+#if PTHREADS || ASYNCIFY
+  // Whether a failed `op` on a blocking socket should wait and retry.
+  $sockWouldBlock__internal: true,
+  $sockWouldBlock: (e, sock, dontWait) =>
+    e.name === 'ErrnoError' && e.errno === {{{ cDefs.EAGAIN }}} && !dontWait && !(sock.stream.flags & {{{ cDefs.O_NONBLOCK }}}),
+  // Async sockCall(): run `op`, and on a would-block park on the socket's node
+  // wait-queue until poll() reports something to consume (readable, hung up or
+  // errored; readiness is re-derived on each wake, the wake flags are just the
+  // trigger), then retry. Always parks after an EAGAIN rather than re-checking
+  // readiness first: nothing can have changed since `op` ran, and a backend
+  // whose poll() disagrees with its recv must wait, not spin. Resolves to the
+  // result or -errno: wrapSyscallFunction's catch only covers the synchronous
+  // part of the syscall, so a rejection would escape it.
+  $sockCallAsync__internal: true,
+  $sockCallAsync__deps: ['$getSocketFromFD', '$sockWouldBlock'],
+  $sockCallAsync: async (fd, dontWait, op) => {
+    try {
+      var sock = getSocketFromFD(fd);
+      for (;;) {
+        try {
+          return op(sock);
+        } catch (e) {
+          if (!sockWouldBlock(e, sock, dontWait)) throw e;
+        }
+        await new Promise((resolve) => {
+          var reg = sock.stream.node.addListener(() => {
+            if (!(sock.sock_ops.poll(sock) & {{{ cDefs.POLLIN | cDefs.POLLERR | cDefs.POLLHUP }}})) return;
+            reg.listeners.delete(reg.entry);
+            resolve();
+          });
+        });
+      }
+    } catch (e) {
+      if (e.name !== 'ErrnoError') throw e;
+      return -e.errno;
+    }
+  },
+#endif
+  __syscall_accept4__deps: ['$sockCall', '$writeSockaddr'],
+#if PTHREADS || ASYNCIFY
+  __syscall_accept4__async: true,
+#endif
+  __syscall_accept4: (fd, addr, len, flags, u1, u2) => {
+    return sockCall(fd, false, (sock) => {
     var newsock = sock.sock_ops.accept(sock);
 #if NODERAWSOCKETS
     // Linux: the accepted fd's status flags come only from `flags`, never from
@@ -426,6 +524,7 @@ var SyscallsLibrary = {
 #endif
     }
     return newsock.stream.fd;
+    });
   },
   __syscall_bind__deps: ['$getSocketFromFD', '$getSocketAddress'],
   __syscall_bind: (fd, addr, len, u1, u2, u3) => {
@@ -440,9 +539,12 @@ var SyscallsLibrary = {
     sock.sock_ops.listen(sock, backlog);
     return 0;
   },
-  __syscall_recvfrom__deps: ['$getSocketFromFD', '$writeSockaddr'],
+  __syscall_recvfrom__deps: ['$sockCall', '$writeSockaddr'],
+#if PTHREADS || ASYNCIFY
+  __syscall_recvfrom__async: true,
+#endif
   __syscall_recvfrom: (fd, buf, len, flags, addr, alen) => {
-    var sock = getSocketFromFD(fd);
+    return sockCall(fd, flags & {{{ cDefs.MSG_DONTWAIT }}}, (sock) => {
     var msg = sock.sock_ops.recvmsg(sock, len, flags);
     if (!msg) return 0; // socket is closed
     if (addr) {
@@ -453,6 +555,7 @@ var SyscallsLibrary = {
     }
     HEAPU8.set(msg.buffer, buf);
     return msg.buffer.byteLength;
+    });
   },
   __syscall_sendto__deps: ['$getSocketFromFD', '$getSocketAddress'],
   __syscall_sendto: (fd, buf, len, flags, addr, alen) => {
@@ -529,9 +632,12 @@ var SyscallsLibrary = {
     // write the buffer
     return sock.sock_ops.sendmsg(sock, view, 0, total, addr, port);
   },
-  __syscall_recvmsg__deps: ['$getSocketFromFD', '$writeSockaddr'],
+  __syscall_recvmsg__deps: ['$sockCall', '$writeSockaddr'],
+#if PTHREADS || ASYNCIFY
+  __syscall_recvmsg__async: true,
+#endif
   __syscall_recvmsg: (fd, message, flags, u1, u2, u3) => {
-    var sock = getSocketFromFD(fd);
+    return sockCall(fd, flags & {{{ cDefs.MSG_DONTWAIT }}}, (sock) => {
     var iov = {{{ makeGetValue('message', C_STRUCTS.msghdr.msg_iov, '*') }}};
     var num = {{{ makeGetValue('message', C_STRUCTS.msghdr.msg_iovlen, 'i32') }}};
     // get the total amount of data we can read across all arrays
@@ -587,6 +693,7 @@ var SyscallsLibrary = {
     // MSG_CTRUNC
 
     return bytesRead;
+    });
   },
 #endif // ~PROXY_POSIX_SOCKETS==0
   __syscall_fchdir: (fd) => {

src/settings.js

@@ -420,7 +420,9 @@ var PROXY_POSIX_SOCKETS = false;
 // It is event-driven. Socket readiness comes through the same
 // ``emscripten_set_socket_*_callback`` hooks the WebSocket backend uses, so it
 // works with existing readiness reactors. It cannot be combined with the
-// WebSocket emulation or :ref:`PROXY_POSIX_SOCKETS`.
+// WebSocket emulation or :ref:`PROXY_POSIX_SOCKETS`. Blocking ``accept()``
+// and ``recv()`` wait (like ``poll()``) from a pthread or under
+// :ref:`ASYNCIFY`/JSPI.
 //
 // It works under -pthread with :ref:`PROXY_TO_PTHREAD`, where main() and every socket
 // syscall run on a single worker alongside the node handles and their event

src/struct_info.json

@@ -322,6 +322,7 @@
             "SOCK_CLOEXEC",
             "SOCK_NONBLOCK",
              "MSG_PEEK",
+             "MSG_DONTWAIT",
              "AF_INET",
              "AF_UNSPEC",
              "AF_INET6",

src/struct_info_generated.json

@@ -338,6 +338,7 @@
         "KMOD_RCTRL": 128,
         "KMOD_RSHIFT": 2,
         "MAP_PRIVATE": 2,
+        "MSG_DONTWAIT": 64,
         "MSG_PEEK": 2,
         "NCCS": 32,
         "NI_NAMEREQD": 8,

src/struct_info_generated_wasm64.json

@@ -338,6 +338,7 @@
         "KMOD_RCTRL": 128,
         "KMOD_RSHIFT": 2,
         "MAP_PRIVATE": 2,
+        "MSG_DONTWAIT": 64,
         "MSG_PEEK": 2,
         "NCCS": 32,
         "NI_NAMEREQD": 8,

test/codesize/test_codesize_hello_dylink_all.json

@@ -1,7 +1,7 @@
 {
-  "a.out.js": 270695,
+  "a.out.js": 270739,
   "a.out.nodebug.wasm": 588289,
-  "total": 858984,
+  "total": 859028,
   "sent": [
     "IMG_Init",
     "IMG_Load",

test/sockets/test_nonblock_flags.c

@@ -100,10 +100,6 @@ int main(void) {
   assert(is_nonblock(nonblocking_fd));
 
   assert(accept(nonblocking_fd, NULL, NULL) == -1 && errno == EAGAIN);
-#ifdef __EMSCRIPTEN__
-  // A blocking accept cannot block, so it would-blocks too.
-  assert(accept(blocking_fd, NULL, NULL) == -1 && errno == EAGAIN);
-#endif
 
   check_accept(blocking_fd, (struct sockaddr*)&blocking_addr, sizeof(blocking_addr), 0, 0);
   check_accept(blocking_fd, (struct sockaddr*)&blocking_addr, sizeof(blocking_addr), SOCK_NONBLOCK | SOCK_CLOEXEC, 1);
@@ -128,6 +124,7 @@ int main(void) {
   assert(client_fd >= 0);
   check_connect(client_fd, (struct sockaddr*)&un, sizeof(un));
   close(client_fd);
+  close(accept(unix_fd, NULL, NULL)); // drain that connection from the queue
   check_accept(unix_fd, (struct sockaddr*)&un, sizeof(un), 0, 0);
   check_accept(unix_fd, (struct sockaddr*)&un, sizeof(un), SOCK_NONBLOCK, 1);
   close(unix_fd);