Report POLLHUP/POLLERR readiness for pipe and socket hangups (emscripten-core#27298)

Makes poll/epoll surface the hangup and error conditions that Linux does
when the far end of a pipe or a connecting socket goes away. Without
these edges a reader (or writer) blocked in poll/epoll (e.g. mio's
`is_write_closed()`/`is_error()`) never wakes and hangs.

Pipe (`libpipefs.js`): on Linux a `pipe(2)` signals both directions of
hangup, which PIPEFS never modelled — `poll()` returned readiness only
while unread bytes remained and `close()` just decremented a shared
refcount without waking anyone.

* Once every write end is closed, poll on the read end reports
`POLLHUP|POLLIN` (read returns EOF). We track open write ends via a
`writerCount` (incremented in `dup`); when the last one closes we set
`writeClosed`, `notifyListeners(POLLHUP | POLLRDNORM | POLLIN)` on the
read end's wait-queue, and OR `POLLHUP|POLLIN` into `poll()`.
* Symmetrically, once every read end is closed, poll on the write end
reports `POLLERR` (a further write would get `EPIPE`), while staying
writable per Linux. Mirrored via `readerCount`/`readClosed` and a
`pipe.writeNode` wait-queue.

The old `refcnt` field is dropped — it always equalled `readerCount +
writerCount`, so buckets are freed once both reach 0.

Socket (`libsockfs_node.js`): a refused connect resolves with
`sock.error = ECONNREFUSED`, but `poll()` only reported `POLLOUT`, so
`is_write_closed()` and `is_error()` were both false under
NODERAWSOCKETS. We now report `POLLOUT | POLLERR | POLLHUP` on a pending
error, matching Linux's connect-failure readiness. `POLLOUT|POLLERR`
satisfies epoll's `is_write_closed()` mapping. The `SO_ERROR`
read-and-clear semantics are untouched.

Before (pipe read end after writer close): `poll()` returned `0` once
the buffer drained.
After: `poll()` returns `POLLHUP | POLLIN`.

Before (refused connect): `poll()` revents `0x11` (`POLLIN|POLLHUP` from
the CLOSED transition, no `POLLERR`).
After: `0x1d` (`POLLIN|POLLOUT|POLLERR|POLLHUP`).

Test coverage:

* `test/core/test_pipe_pollhup.c` — writes, closes the writer, and
asserts `POLLHUP|POLLIN` both while data is buffered and after draining
to EOF; also closes the reader and asserts the write end reports
`POLLOUT|POLLERR`.
* Extends `test/sockets/test_tcp_refused.c` to `poll()` for
`POLLERR|POLLHUP` *before* reading (and thereby clearing) `SO_ERROR`.

_Made with AI assistance under my review_

Author: guybedford

src/lib/libpipefs.js

@@ -17,9 +17,15 @@ addToLibrary({
     createPipe() {
       var pipe = {
         buckets: [],
-        // refcnt 2 because pipe has a read end and a write end. We need to be
-        // able to read from the read end after write end is closed.
-        refcnt : 2,
+        // Open write ends. When it drops to 0 the reader sees EOF and poll must
+        // report POLLHUP (Linux semantics). Buckets are freed once both counts
+        // reach 0.
+        writerCount: 1,
+        writeClosed: false,
+        // Open read ends. When it drops to 0 the writer sees POLLERR (a further
+        // write would get EPIPE).
+        readerCount: 1,
+        readClosed: false,
         timestamp: new Date(),
       };
 
@@ -36,8 +42,10 @@ addToLibrary({
 
       rNode.pipe = pipe;
       wNode.pipe = pipe;
-      // The read end's node carries the poll wait-queue; writes wake it.
+      // The read end's node carries the reader poll wait-queue (writes wake it);
+      // the write end's node carries the writer wait-queue (read-end close wakes it).
       pipe.readNode = rNode;
+      pipe.writeNode = wNode;
 
       var readableStream = FS.createStream({
         path: rName,
@@ -86,17 +94,35 @@ addToLibrary({
         var pipe = stream.node.pipe;
 
         if ((stream.flags & {{{ cDefs.O_ACCMODE }}}) === {{{ cDefs.O_WRONLY }}}) {
-          return ({{{ cDefs.POLLWRNORM }}} | {{{ cDefs.POLLOUT }}});
+          // Linux keeps the write end writable (the write itself fails with
+          // EPIPE) while also signalling POLLERR once every read end is closed.
+          var mask = {{{ cDefs.POLLWRNORM }}} | {{{ cDefs.POLLOUT }}};
+          if (pipe.readClosed) {
+            mask |= {{{ cDefs.POLLERR }}};
+          }
+          return mask;
         }
+        var mask = 0;
         for (var bucket of pipe.buckets) {
           if (bucket.offset - bucket.roffset > 0) {
-            return ({{{ cDefs.POLLRDNORM }}} | {{{ cDefs.POLLIN }}});
+            mask = {{{ cDefs.POLLRDNORM }}} | {{{ cDefs.POLLIN }}};
+            break;
           }
         }
-        return 0;
+        // With every write end closed the read end is at EOF: readable (read
+        // returns 0) and hung up.
+        if (pipe.writeClosed) {
+          mask |= {{{ cDefs.POLLHUP }}} | {{{ cDefs.POLLIN }}};
+        }
+        return mask;
       },
       dup(stream) {
-        stream.node.pipe.refcnt++;
+        var pipe = stream.node.pipe;
+        if ((stream.flags & {{{ cDefs.O_ACCMODE }}}) === {{{ cDefs.O_WRONLY }}}) {
+          pipe.writerCount++;
+        } else {
+          pipe.readerCount++;
+        }
       },
       ioctl(stream, request, argp) {
         if (request == {{{ cDefs.FIONREAD }}}) {
@@ -251,8 +277,20 @@ addToLibrary({
       },
       close(stream) {
         var pipe = stream.node.pipe;
-        pipe.refcnt--;
-        if (pipe.refcnt === 0) {
+        // When the last write end closes, wake any poll/epoll waiter on the read
+        // end with POLLHUP so a reader blocked on the writer dropping unblocks.
+        if ((stream.flags & {{{ cDefs.O_ACCMODE }}}) === {{{ cDefs.O_WRONLY }}}) {
+          if (--pipe.writerCount === 0) {
+            pipe.writeClosed = true;
+            pipe.readNode.notifyListeners({{{ cDefs.POLLHUP }}} | {{{ cDefs.POLLRDNORM }}} | {{{ cDefs.POLLIN }}});
+          }
+        } else if (--pipe.readerCount === 0) {
+          // Mirror: when the last read end closes, wake any poll/epoll waiter on
+          // the write end with POLLERR (a further write would get EPIPE).
+          pipe.readClosed = true;
+          pipe.writeNode.notifyListeners({{{ cDefs.POLLERR }}} | {{{ cDefs.POLLWRNORM }}} | {{{ cDefs.POLLOUT }}});
+        }
+        if (pipe.readerCount === 0 && pipe.writerCount === 0) {
           pipe.buckets = null;
         }
       }

src/lib/libsockfs_node.js

@@ -336,8 +336,10 @@ var NodeSockFSLibrary = {
         mask |= ({{{ cDefs.POLLRDNORM }}} | {{{ cDefs.POLLIN }}});
       }
       if (sock.error) {
-        // Mark writable on error so SO_ERROR can be read.
-        mask |= {{{ cDefs.POLLOUT }}};
+        // A pending socket error (e.g. a refused connect) is Linux's
+        // POLLERR|POLLHUP, plus writable so SO_ERROR can be read. POLLOUT|POLLERR
+        // also satisfies epoll's is_write_closed() mapping.
+        mask |= {{{ cDefs.POLLOUT }}} | {{{ cDefs.POLLERR }}} | {{{ cDefs.POLLHUP }}};
       } else if (sock.connection && sock.state === {{{ SOCK_STATE_CONNECTED }}} && !sock.writeBlocked) {
         mask |= {{{ cDefs.POLLOUT }}};
       }

test/codesize/test_codesize_hello_dylink_all.json

@@ -1,7 +1,7 @@
 {
-  "a.out.js": 268425,
+  "a.out.js": 268621,
   "a.out.nodebug.wasm": 587978,
-  "total": 856403,
+  "total": 856599,
   "sent": [
     "IMG_Init",
     "IMG_Load",

test/core/test_pipe_pollhup.c

@@ -0,0 +1,80 @@
+/*
+ * 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.
+ */
+
+// Once every write end of a pipe is closed, poll on the read end must report
+// POLLHUP (and POLLIN, since read returns EOF), matching Linux. Symmetrically,
+// once every read end is closed, poll on the write end must report POLLERR
+// (while staying writable, the write itself failing with EPIPE).
+
+#include <poll.h>
+#include <assert.h>
+#include <stdio.h>
+#include <unistd.h>
+#include <string.h>
+
+int poll_events(int fd, short events) {
+  struct pollfd pfd = {fd, events, 0};
+  assert(poll(&pfd, 1, 0) >= 0);
+  return pfd.revents;
+}
+
+int poll_read(int fd) {
+  return poll_events(fd, POLLIN);
+}
+
+int poll_write(int fd) {
+  return poll_events(fd, POLLOUT);
+}
+
+int main() {
+  const char *t = "test\n";
+  int p[2];
+
+  assert(pipe(p) == 0);
+
+  // Nothing written yet, writer still open: not readable, no hangup.
+  assert(poll_read(p[0]) == 0);
+
+  // Both ends open: write end is writable, no error.
+  int wevents = poll_write(p[1]);
+  assert(wevents & POLLOUT);
+  assert(!(wevents & POLLERR));
+
+  write(p[1], t, strlen(t));
+
+  // Data pending, writer still open: readable, no hangup.
+  int revents = poll_read(p[0]);
+  assert(revents & POLLIN);
+  assert(!(revents & POLLHUP));
+
+  // Close the write end. Data is still buffered, so POLLIN and POLLHUP.
+  close(p[1]);
+  revents = poll_read(p[0]);
+  assert(revents & POLLIN);
+  assert(revents & POLLHUP);
+
+  // Drain the buffer. Read end at EOF with writer gone: POLLIN and POLLHUP.
+  char buf[16];
+  assert(read(p[0], buf, sizeof(buf)) == (ssize_t)strlen(t));
+  revents = poll_read(p[0]);
+  assert(revents & POLLIN);
+  assert(revents & POLLHUP);
+
+  close(p[0]);
+
+  // Mirror case: close the read end and poll the write end for POLLERR.
+  assert(pipe(p) == 0);
+  close(p[0]);
+  wevents = poll_write(p[1]);
+  // Still writable per Linux, but now signalling an error.
+  assert(wevents & POLLOUT);
+  assert(wevents & POLLERR);
+  close(p[1]);
+
+  printf("done\n");
+  return 0;
+}

test/sockets/test_tcp_refused.c

@@ -13,6 +13,7 @@
 #include <errno.h>
 #include <fcntl.h>
 #include <netinet/in.h>
+#include <poll.h>
 #include <stdio.h>
 #include <stdlib.h>
 #include <string.h>
@@ -45,10 +46,17 @@ void main_loop(void) {
   select(64, NULL, &fdw, NULL, &tv);
 
   if (FD_ISSET(fd, &fdw)) {
+    // Poll before reading SO_ERROR (which clears the pending error). A refused
+    // connect is Linux POLLERR|POLLHUP (plus writable).
+    struct pollfd pfd = {fd, POLLIN | POLLOUT, 0};
+    assert(poll(&pfd, 1, 0) == 1);
+    printf("poll revents 0x%x\n", pfd.revents);
+    assert(pfd.revents & POLLERR);
+    assert(pfd.revents & POLLHUP);
+
     int err = 0;
     socklen_t l = sizeof(err);
     getsockopt(fd, SOL_SOCKET, SO_ERROR, &err, &l);
-    if (err == 0) return; // not resolved yet
     printf("connect resolved with errno %d (%s)\n", err, strerror(err));
     assert(err == ECONNREFUSED && "expected ECONNREFUSED");
     test_success();

test/test_core.py

@@ -9735,6 +9735,9 @@ def test_pipe_select(self, args):
       self.require_pthreads()
     self.do_runf('core/test_pipe_select.c', cflags=args)
 
+  def test_pipe_pollhup(self):
+    self.do_runf('core/test_pipe_pollhup.c', 'done\n')
+
   @also_without_bigint
   def test_jslib_i64_params(self):
     # Tests the defineI64Param and receiveI64ParamAsI53 helpers that are