usockets: bound the accepts of one listener readiness event

bench/snippets/http-connection-per-request.mjs

@@ -0,0 +1,93 @@
+// A server for connection-per-request load: each client opens a connection, sends one
+// request and closes it. `ab` does this when it runs without -k.
+//
+//   bun http-connection-per-request.mjs [Bun.serve | Bun.listen | node:http | node:net]
+//   node http-connection-per-request.mjs [node:http | node:net]
+//   ab -n 100000 -c 64 http://127.0.0.1:3000/
+//
+// Each second with requests it prints the count and the CPU time of the server per request.
+import http from "node:http";
+import net from "node:net";
+
+const kind = process.argv[2] ?? (typeof Bun === "undefined" ? "node:http" : "Bun.serve");
+const port = Number(process.env.PORT ?? 3000);
+const body = "Hello, World!";
+const response = `HTTP/1.1 200 OK\r\nContent-Type: text/plain\r\nContent-Length: ${body.length}\r\nConnection: close\r\n\r\n${body}`;
+
+let requests = 0;
+
+// The head of a request can arrive in more than one chunk. The raw servers answer when it is
+// complete, as the HTTP servers do. `head` is null after the answer.
+function complete(head, chunk) {
+  head += chunk.toString("latin1");
+  return head.includes("\r\n\r\n") ? null : head;
+}
+
+switch (kind) {
+  case "Bun.serve":
+    Bun.serve({
+      port,
+      fetch() {
+        requests++;
+        return new Response(body);
+      },
+    });
+    break;
+  case "Bun.listen":
+    Bun.listen({
+      hostname: "0.0.0.0",
+      port,
+      socket: {
+        open(socket) {
+          socket.data = "";
+        },
+        data(socket, chunk) {
+          if (socket.data === null) return;
+          socket.data = complete(socket.data, chunk);
+          if (socket.data !== null) return;
+          requests++;
+          socket.end(response);
+        },
+      },
+    });
+    break;
+  case "node:http":
+    http
+      .createServer((req, res) => {
+        requests++;
+        res.end(body);
+      })
+      .listen(port);
+    break;
+  case "node:net":
+    net
+      .createServer(socket => {
+        let head = "";
+        socket.on("error", () => {});
+        socket.on("data", chunk => {
+          if (head === null) return;
+          head = complete(head, chunk);
+          if (head !== null) return;
+          requests++;
+          socket.end(response);
+        });
+      })
+      .listen(port);
+    break;
+  default:
+    throw new Error(`unknown server "${kind}"`);
+}
+
+console.log(`${kind} on port ${port}`);
+
+let last = { requests, cpu: process.cpuUsage() };
+setInterval(() => {
+  const cpu = process.cpuUsage();
+  const count = requests - last.requests;
+  if (count) {
+    const micros = cpu.user - last.cpu.user + (cpu.system - last.cpu.system);
+    console.log(`${count} requests, ${(micros / count).toFixed(1)} us of CPU per request`);
+  }
+  // Read after the report, so that the next interval does not pay for this one's log line.
+  last = { requests, cpu: process.cpuUsage() };
+}, 1000);

packages/bun-usockets/src/loop.c

@@ -334,6 +334,12 @@ void us_internal_timer_sweep(struct us_loop_t *loop) {
  * easier on CPU */
 static const int MAX_LOW_PRIO_SOCKETS_PER_LOOP_ITERATION = 5;
 
+/* The handlers of an accepted socket run inside the accept loop, so a client they answer can be
+ * back in the queue before the next accept(). Without a bound the loop does not end while such
+ * clients keep coming, and nothing else on the event loop runs. The poll of the listener is
+ * level-triggered: it reports what is still queued on the next iteration. */
+static const int MAX_ACCEPTS_PER_READINESS_EVENT = 32;
+
 void us_internal_handle_low_priority_sockets(struct us_loop_t *loop) {
     struct us_internal_loop_data_t *loop_data = &loop->data;
     struct us_socket_t *s;
@@ -519,6 +525,7 @@ void us_internal_dispatch_ready_poll(struct us_poll_t *p, int error, int eof, in
 
                     /* Todo: stop timer if any */
 
+                    int accepted = 0;
                     do {
                         struct us_poll_t *accepted_p = us_create_poll(loop, 0, sizeof(struct us_socket_t) - sizeof(struct us_poll_t) + listen_socket->socket_ext_size);
                         us_poll_init(accepted_p, client_fd, POLL_TYPE_SOCKET);
@@ -578,7 +585,7 @@ void us_internal_dispatch_ready_poll(struct us_poll_t *p, int error, int eof, in
                             break;
                         }
 
-                    } while ((client_fd = bsd_accept_socket(us_poll_fd(p), &addr)) != LIBUS_SOCKET_ERROR);
+                    } while (++accepted < MAX_ACCEPTS_PER_READINESS_EVENT && (client_fd = bsd_accept_socket(us_poll_fd(p), &addr)) != LIBUS_SOCKET_ERROR);
                 }
             }
         break;