usockets(kqueue): stop a non-reading socket's read knote from waking the loop on data

packages/bun-usockets/src/eventing/epoll_kqueue.c

@@ -18,6 +18,7 @@
 #include "libusockets.h"
 #include "internal/internal.h"
 #include "internal/fault_inject.h"
+#include <limits.h>
 #include <stdlib.h>
 #include <time.h>
 #if defined(LIBUS_USE_EPOLL) || defined(LIBUS_USE_KQUEUE)
@@ -581,13 +582,16 @@ static int kqueue_is_socket_poll(struct us_poll_t *p) {
  * stays registered while the socket does not poll for reads (paused, half-open after the
  * peer's FIN, shut down with reads off, parked as low priority), re-added with EV_CLEAR.
  * It is kqueue's stand-in for epoll's implicit EPOLLHUP/EPOLLERR: the peer's FIN or RST
- * still reaches the dispatcher as eof/error (us_poll_events masks the readable bit out),
- * and EV_CLEAR keeps unread data or a consumed EOF from re-firing every tick. Nothing else
+ * still reaches the dispatcher as eof/error (us_poll_events masks the readable bit out).
+ * NOTE_LOWAT with an unreachable low-water mark keeps arriving data from waking the loop
+ * while reads are off (the socket filter reports EOF and so_error before it consults the
+ * mark; xnu clamps the mark to the receive buffer size, so there it can fire once more when
+ * the buffer fills), and EV_CLEAR keeps a consumed EOF from re-firing. Nothing else
  * reports them: the one-shot write filter is consumed by the first, immediate, writable
  * event, so a reset of a paused socket went unreported until resume(), unlike on epoll
- * and libuv. EV_ADD on an existing knote updates its udata but keeps its flags, so each
- * switch between the two modes deletes the knote and adds a new one; us_poll_resize relies
- * on the same rule to move the udata without changing the mode. */
+ * and libuv. EV_ADD on an existing knote updates its udata but keeps its flags (and would
+ * reset the sentinel's NOTE_LOWAT), so each switch between the two modes, and the udata
+ * move in us_poll_resize, deletes the knote and adds a new one. */
 int kqueue_change(int kqfd, int fd, int old_events, int new_events, void *user_data, int keep_read_knote) {
     struct kevent64_s change_list[3];
     int change_length = 0;
@@ -596,7 +600,11 @@ int kqueue_change(int kqfd, int fd, int old_events, int new_events, void *user_d
     if (is_readable != (old_events & LIBUS_SOCKET_READABLE)) {
         if (keep_read_knote) {
             EV_SET64(&change_list[change_length++], fd, EVFILT_READ, EV_DELETE, 0, 0, 0, 0, 0);
-            EV_SET64(&change_list[change_length++], fd, EVFILT_READ, EV_ADD | (is_readable ? 0 : EV_CLEAR), 0, 0, (uint64_t)(void*)user_data, 0, 0);
+            if (is_readable) {
+                EV_SET64(&change_list[change_length++], fd, EVFILT_READ, EV_ADD, 0, 0, (uint64_t)(void*)user_data, 0, 0);
+            } else {
+                EV_SET64(&change_list[change_length++], fd, EVFILT_READ, EV_ADD | EV_CLEAR, NOTE_LOWAT, INT_MAX, (uint64_t)(void*)user_data, 0, 0);
+            }
         } else {
             EV_SET64(&change_list[change_length++], fd, EVFILT_READ, is_readable ? EV_ADD : EV_DELETE, 0, 0, (uint64_t)(void*)user_data, 0, 0);
         }
@@ -649,10 +657,10 @@ struct us_poll_t *us_poll_resize(struct us_poll_t *p, struct us_loop_t *loop, un
     new_p->state.poll_type = us_internal_poll_type(new_p);
     us_poll_change(new_p, loop, events);
 #else
-    /* Re-add both filters to move their udata to new_p, whether or not they are polled: the
-     * EV_CLEAR read knote of a socket that is not reading (see kqueue_change) has to follow
-     * the relocation too, and EV_ADD keeps the mode of a knote that already exists. */
-    kqueue_change(loop->fd, new_p->state.fd, 0, LIBUS_SOCKET_WRITABLE | LIBUS_SOCKET_READABLE, new_p, 0);
+    /* Move the udata to new_p: re-register the polled filters, and for a socket poll re-create
+     * its read knote in its current mode (it has one whether or not it reads, see kqueue_change). */
+    const int is_socket = kqueue_is_socket_poll(new_p);
+    kqueue_change(loop->fd, new_p->state.fd, is_socket ? (~events & LIBUS_SOCKET_READABLE) : 0, events, new_p, is_socket);
 #endif
     /* This is needed for epoll also (us_change_poll doesn't update the old poll) */
     us_internal_loop_update_pending_ready_polls(loop, p, new_p, events, events);

src/js/internal-for-testing.ts

@@ -695,6 +695,8 @@ export const getEventLoopStats: () => {
   numPolls: number;
   loopActive: boolean;
   eventLoopAlive: boolean;
+  /** usockets/libuv loop iterations so far (us_internal_loop_pre count). */
+  iteration: number;
 } = $newRustFunction("event_loop.rs", "getActiveTasks", 0);
 
 export const hostedGitInfo = {

src/jsc/event_loop.rs

@@ -1331,7 +1331,7 @@ pub fn get_active_tasks(global_object: &JSGlobalObject, _frame: &CallFrame) -> J
     // fields and call &-methods on it for the duration of this host fn.
     let vm_ref = global_object.bun_vm();
     let event_loop = vm_ref.event_loop_shared();
-    let result = JSValue::create_empty_object(global_object, 8);
+    let result = JSValue::create_empty_object(global_object, 9);
     result.put(
         global_object,
         b"activeTasks",
@@ -1382,6 +1382,12 @@ pub fn get_active_tasks(global_object: &JSGlobalObject, _frame: &CallFrame) -> J
         b"numPolls",
         JSValue::js_number(num_polls as f64),
     );
+    result.put(
+        global_object,
+        b"iteration",
+        // SAFETY: usockets_loop() returns the live process-global loop.
+        JSValue::js_number(unsafe { (*event_loop.usockets_loop()).iteration_number() } as f64),
+    );
     Ok(result)
 }
 

test/js/bun/net/socket.test.ts

@@ -4087,6 +4087,98 @@ describe("allowHalfOpen socket whose peer resets behind pending writes", () => {
   });
 });
 
+// A paused socket must not wake the event loop for data it is not going to read. epoll and
+// AFD do not report data without readable interest; kqueue keeps a read knote registered on a
+// non-reading socket so the peer's FIN/RST still arrive, and that knote used to fire once per
+// incoming segment. The paused socket lives in a child so nothing else turns its loop: it
+// samples the usockets loop iteration counter, we send SEGMENTS one-byte writes from here,
+// then it samples again. A loop that wakes per segment reports ~SEGMENTS iterations; one that
+// does not reports the handful caused by our two control lines.
+it("a paused socket does not wake the event loop for every segment its peer sends", async () => {
+  const SEGMENTS = 200;
+  await using child = spawn({
+    cmd: [
+      bunExe(),
+      "-e",
+      `
+      const { getEventLoopStats } = require("bun:internal-for-testing");
+      let socket;
+      const server = Bun.listen({
+        hostname: "127.0.0.1",
+        port: 0,
+        socket: {
+          open(s) {
+            socket = s;
+            s.pause();
+            process.stdout.write("port " + server.port + "\\n");
+          },
+          data() {
+            process.stdout.write("data while paused\\n");
+          },
+          close() {},
+          drain() {},
+        },
+      });
+      process.stdout.write("port " + server.port + "\\n");
+      let before;
+      for await (const line of console) {
+        if (line === "start") {
+          before = getEventLoopStats().iteration;
+          process.stdout.write("started\\n");
+        } else if (line === "stop") {
+          process.stdout.write("iterations " + (getEventLoopStats().iteration - before) + "\\n");
+          server.stop(true);
+          process.exit(0);
+        }
+      }
+      `,
+    ],
+    env: bunEnv,
+    stdin: "pipe",
+    stdout: "pipe",
+    stderr: "inherit",
+  });
+  const reader = child.stdout.getReader();
+  let buffered = "";
+  async function line() {
+    while (!buffered.includes("\n")) {
+      const { value, done } = await reader.read();
+      if (done) return buffered;
+      buffered += new TextDecoder().decode(value);
+    }
+    const i = buffered.indexOf("\n");
+    const out = buffered.slice(0, i);
+    buffered = buffered.slice(i + 1);
+    return out;
+  }
+  const port = Number((await line()).split(" ")[1]);
+  const peer = await Bun.connect({
+    hostname: "127.0.0.1",
+    port,
+    socket: { data() {}, open() {}, close() {}, drain() {} },
+  });
+  expect(await line()).toBe(`port ${port}`); // open() ran: the socket is paused
+  child.stdin.write("start\n");
+  await child.stdin.flush();
+  expect(await line()).toBe("started");
+  for (let i = 0; i < SEGMENTS; i++) {
+    peer.write("x");
+    peer.flush();
+    // Separate segments need separate event-loop turns on our side; this paces the sender,
+    // it is not waiting for a condition in the child.
+    await new Promise<void>(resolve => setTimeout(resolve, 1));
+  }
+  child.stdin.write("stop\n");
+  await child.stdin.flush();
+  const result = await line();
+  peer.end();
+  expect(result).toMatch(/^iterations \d+$/);
+  // Two stdin lines and process bookkeeping account for a few turns; per-segment wakeups
+  // would put this near SEGMENTS.
+  expect(Number(result.split(" ")[1])).toBeLessThan(SEGMENTS / 4);
+  expect(await child.exited).toBe(0);
+});
+
 // A paused socket polls for nothing. epoll reports the reset anyway (EPOLLERR cannot be
 // masked); kqueue only reports it through the read knote that epoll_kqueue.c keeps
 // registered while reads are off. Before that, the pause left a one-shot writable event