diff --git a/docs/runtime/networking/udp.mdx b/docs/runtime/networking/udp.mdx index 8ca87d969294..930d300e8457 100644 --- a/docs/runtime/networking/udp.mdx +++ b/docs/runtime/networking/udp.mdx @@ -51,6 +51,10 @@ const client = await Bun.udpSocket({}); client.send("Hello!", server.port, "127.0.0.1"); ``` +Bun reads at most 32 datagrams from a socket in one iteration of the event loop, as Node.js does. The operating system holds the rest in the receive buffer of the socket, and Bun reads them in the next iterations. Timers and other sockets keep running when a peer sends without pause. + +When the receive buffer is full, the operating system drops new datagrams and does not tell Bun. To keep up with a fast sender, keep the `data` callback short and move long work to a [`Worker`](/runtime/workers). + ### Connections UDP has no concept of a connection, but many UDP exchanges (especially as a client) involve only one peer. diff --git a/packages/bun-usockets/src/bsd.c b/packages/bun-usockets/src/bsd.c index dcffbcfc8f5a..e19f91182c3b 100644 --- a/packages/bun-usockets/src/bsd.c +++ b/packages/bun-usockets/src/bsd.c @@ -148,7 +148,7 @@ int bsd_sendmmsg(LIBUS_SOCKET_DESCRIPTOR fd, struct udp_sendbuf* sendbuf, int fl int bsd_recvmmsg(LIBUS_SOCKET_DESCRIPTOR fd, struct udp_recvbuf *recvbuf, int flags, int max_packets) { if (max_packets > LIBUS_UDP_RECV_COUNT) max_packets = LIBUS_UDP_RECV_COUNT; #if defined(_WIN32) - for (int i = 0; i < LIBUS_UDP_RECV_COUNT; i++) { + for (int i = 0; i < max_packets; i++) { while (1) { socklen_t addr_len = sizeof(struct sockaddr_storage); ssize_t ret = recvfrom(fd, recvbuf->buf + (size_t) i * LIBUS_UDP_MAX_SIZE, @@ -169,7 +169,7 @@ int bsd_recvmmsg(LIBUS_SOCKET_DESCRIPTOR fd, struct udp_recvbuf *recvbuf, int fl break; } } - return LIBUS_UDP_RECV_COUNT; + return max_packets; #elif defined(__APPLE__) if (Bun__doesMacOSVersionSupportSendRecvMsgX()) { while (1) { diff --git a/packages/bun-usockets/src/internal/networking/bsd.h b/packages/bun-usockets/src/internal/networking/bsd.h index e269a5ca81c5..62c7a6a8f30a 100644 --- a/packages/bun-usockets/src/internal/networking/bsd.h +++ b/packages/bun-usockets/src/internal/networking/bsd.h @@ -57,6 +57,10 @@ struct bsd_addr_t { }; #define LIBUS_UDP_RECV_COUNT (LIBUS_RECV_BUFFER_LENGTH / LIBUS_UDP_MAX_SIZE) +/* The most datagrams that one readable event of a UDP socket hands over. It is + * libuv's count, and like libuv's it counts datagrams, not receive calls: + * https://github.com/libuv/libuv/blob/895cd04bf7cadd2f8993c0d6fc7e55eaaec60bc9/src/unix/udp.c#L309-L312 */ +#define LIBUS_UDP_MAX_RECV_PER_EVENT 32 #ifdef __APPLE__ /* diff --git a/packages/bun-usockets/src/loop.c b/packages/bun-usockets/src/loop.c index 2e254029f256..6b4e8d1d6be9 100644 --- a/packages/bun-usockets/src/loop.c +++ b/packages/bun-usockets/src/loop.c @@ -1024,13 +1024,22 @@ void us_internal_dispatch_ready_poll(struct us_poll_t *p, int error, int eof, in #else const int run_recv = events & LIBUS_SOCKET_READABLE; #endif + /* Reading until EAGAIN lets the peer decide when this loop ends: a + * sender at or above the rate on_data drains at, or one that + * answers what on_data sends, keeps the queue non-empty, and no + * timer, immediate or other poll runs until it stops. The poll is + * level-triggered on every backend, so what is left raises the + * next event. */ + int recv_budget = LIBUS_UDP_MAX_RECV_PER_EVENT; if (run_recv && !u->closed) { - do { struct udp_recvbuf recvbuf; bsd_udp_setup_recvbuf(&recvbuf, u->loop->data.recv_buf, LIBUS_RECV_BUFFER_LENGTH); - int npackets = bsd_recvmmsg(us_poll_fd(p), &recvbuf, MSG_DONTWAIT, u->shared_fd ? 1 : LIBUS_UDP_RECV_COUNT); + int max_packets = u->shared_fd ? 1 : LIBUS_UDP_RECV_COUNT; + if (max_packets > recv_budget) max_packets = recv_budget; + int npackets = bsd_recvmmsg(us_poll_fd(p), &recvbuf, MSG_DONTWAIT, max_packets); if (npackets > 0) { + recv_budget -= npackets; u->on_data(u, &recvbuf, npackets); } else { if (npackets == LIBUS_SOCKET_ERROR) { @@ -1077,7 +1086,7 @@ void us_internal_dispatch_ready_poll(struct us_poll_t *p, int error, int eof, in break; } - } while (!u->closed); + } while (!u->closed && recv_budget > 0); } if (events & LIBUS_SOCKET_WRITABLE && !u->closed) { @@ -1100,8 +1109,9 @@ void us_internal_dispatch_ready_poll(struct us_poll_t *p, int error, int eof, in * EAGAIN (which means the error queue is already drained, * leaving a residual EPOLLERR). Otherwise the socket stays * open so the user can keep sending/receiving after a - * transient ICMP error. */ - if (error && !recv_error_surfaced && !recv_would_block_only && !u->closed) { + * transient ICMP error. A read that stopped on its budget never + * got as far as either answer: the next event decides. */ + if (error && !recv_error_surfaced && !recv_would_block_only && recv_budget > 0 && !u->closed) { us_udp_socket_close(u); } #else diff --git a/test/_util/loop-iterations.ts b/test/_util/loop-iterations.ts new file mode 100644 index 000000000000..2f95c1830de1 --- /dev/null +++ b/test/_util/loop-iterations.ts @@ -0,0 +1,68 @@ +// Helpers for fixtures that measure what one iteration of the event loop does: +// how many datagrams, reads or connections a socket hands over before the loop +// goes on to its timers, immediates and other sockets. For processes that run +// with bunEnv, which makes bun:internal-for-testing available. +import { getEventLoopStats } from "bun:internal-for-testing"; + +/** + * Counts events by the iteration of the event loop they happen in. Call + * `count()` from the callback to measure. `perIteration` has one entry for each + * iteration that counted something, in order. + */ +export function iterationCounter() { + const counts = new Map(); + let total = 0; + return { + count(events = 1) { + const { iteration } = getEventLoopStats(); + counts.set(iteration, (counts.get(iteration) ?? 0) + events); + total += events; + }, + get total() { + return total; + }, + summary() { + const perIteration = [...counts.values()]; + return { total, max: Math.max(0, ...perIteration), perIteration }; + }, + }; +} + +/** + * Lets the event loop iterate until `done()` holds. Resolves to false when it + * still does not hold after `seconds`, so that the caller can report what it + * has and not hang. + */ +export async function iterateUntil(done: () => boolean, seconds = 10) { + const deadline = performance.now() + seconds * 1000; + while (!done()) { + if (performance.now() > deadline) return false; + await new Promise(resolve => setImmediate(resolve)); + } + return true; +} + +/** + * Runs `scenario` until a run is `usable`, at most `attempts` times and within + * `seconds` in all, and resolves to that run. A scenario that depends on what + * the kernel had queued before the loop polled cannot promise it on every + * platform and under every load, so a run that did not get there is set up + * again. When no run is usable the last one is the result, for the test to + * fail on. + * + * The scenario gets the seconds that are left of the budget. It passes them to + * `iterateUntil`, so that a run whose datagrams never arrive ends with the + * budget and not 20 deadlines later. + */ +export async function firstUsable( + scenario: (secondsLeft: number) => Promise, + usable: (run: T) => boolean, + { attempts = 20, seconds = 20 } = {}, +) { + const deadline = performance.now() + seconds * 1000; + for (let attempt = 1; ; attempt++) { + const secondsLeft = Math.max(0, (deadline - performance.now()) / 1000); + const run = await scenario(secondsLeft); + if (usable(run) || attempt === attempts || performance.now() > deadline) return { ...run, attempt }; + } +} diff --git a/test/js/bun/udp/dgram-cluster-recv-budget-fixture.ts b/test/js/bun/udp/dgram-cluster-recv-budget-fixture.ts new file mode 100644 index 000000000000..5c08dd364576 --- /dev/null +++ b/test/js/bun/udp/dgram-cluster-recv-budget-fixture.ts @@ -0,0 +1,58 @@ +// A worker reads a cluster-shared descriptor one datagram per receive call, so +// that the workers share the load. The bound of a readable event counts +// datagrams, not receive calls: a backlog of 100 still arrives 32 to an +// iteration of the event loop. The worker queues the backlog on its own socket +// and reports the datagrams of every iteration. The primary prints that report. +import cluster from "node:cluster"; +import dgram from "node:dgram"; +import { iterateUntil, iterationCounter } from "../../../_util/loop-iterations"; + +if (cluster.isPrimary) { + const worker = cluster.fork(); + // Fail fast instead of hanging the harness. + const watchdog = setTimeout(() => { + worker.process.kill("SIGKILL"); + console.error("the worker did not report"); + process.exit(1); + }, 15_000); + watchdog.unref(); + worker.on("message", report => { + console.log(JSON.stringify(report)); + worker.send("stop"); + }); + const code = await new Promise(resolve => worker.on("exit", resolve)); + if (code !== 0) throw new Error(`worker exited with ${code}`); +} else { + const socket = dgram.createSocket("udp4"); + let received = iterationCounter(); + socket.on("message", () => received.count()); + process.on("message", message => { + if (message !== "stop") return; + socket.close(); + cluster.worker!.disconnect(); + }); + + socket.bind({ port: 0, exclusive: false }, async () => { + // The shared descriptor binds INADDR_ANY. Sending to 0.0.0.0 is not + // portably deliverable, loopback is. + const port = socket.address().port; + const sender = await Bun.udpSocket({ hostname: "127.0.0.1", port: 0, socket: { data() {}, error() {} } }); + const packets: (string | number)[] = []; + for (let i = 0; i < 100; i++) packets.push("x", port, "127.0.0.1"); + + // The bound shows only when the backlog was queued before the loop polled + // the socket. A burst that arrived in full without that is sent again. + for (let attempt = 1; ; attempt++) { + received = iterationCounter(); + const sent = sender.sendMany(packets); + if (sent !== 100) throw new Error(`sendMany accepted ${sent} of 100`); + const finished = await iterateUntil(() => received.total === 100); + const run = { finished, ...received.summary(), attempt }; + if (!finished || run.max >= 32 || attempt === 20) { + sender.close(); + process.send!(run); + return; + } + } + }); +} diff --git a/test/js/bun/udp/dgram.test.ts b/test/js/bun/udp/dgram.test.ts index 2623ec39681c..b1cffc4aa4f5 100644 --- a/test/js/bun/udp/dgram.test.ts +++ b/test/js/bun/udp/dgram.test.ts @@ -2,7 +2,7 @@ import { describe, expect, jest, test } from "bun:test"; import { createSocket } from "dgram"; import { Worker } from "node:worker_threads"; -import { bunEnv, bunExe, bunRun, disableAggressiveGCScope, isWindows } from "harness"; +import { bunEnv, bunExe, bunRun, disableAggressiveGCScope, isLinux, isWindows } from "harness"; import path from "path"; import { nodeDataCases } from "./testdata"; @@ -261,6 +261,16 @@ describe.skipIf(isWindows)("cluster", () => { expect(stdout).toStartWith("ok: all 4 workers adopted and released the shared descriptor "); expect(exitCode).toBe(0); }, 40_000); + + // A worker reads a shared descriptor one datagram per receive call. The + // bound of a readable event counts datagrams, not calls, so a backlog + // arrives 32 to an iteration of the event loop, as it does in Node. + test("a shared socket hands over at most 32 datagrams per readable event", async () => { + const { stdout, stderr, exitCode } = await runClusterFixture("dgram-cluster-recv-budget-fixture.ts", 20_000); + expect(stderr).toBe(""); + expect(JSON.parse(stdout)).toMatchObject({ finished: true, total: 100, max: 32 }); + expect(exitCode).toBe(0); + }); }); describe("after close()", () => { @@ -915,6 +925,56 @@ for (const [kind, bind] of Object.entries(icmpBindModes)) { }); } +// One readable event hands over at most 32 datagrams, as in Node, and the rest +// waits in the kernel for the next iteration of the event loop. Each scenario +// of the fixture queues its backlog before the loop polls and reports the +// datagrams of every iteration. +describe.concurrent("a readable event hands over at most 32 datagrams", () => { + const fixture = path.join(import.meta.dir, "udp-recv-budget-fixture.ts"); + + test("of a backlog of 100", async () => { + const result = await bunRun([fixture, "dgram-backlog"]); + expect(result).toSpawn(); + expect(JSON.parse(result.stdout)).toMatchObject({ finished: true, total: 100, max: 32 }); + }); + + test("of each socket", async () => { + const result = await bunRun([fixture, "dgram-two-sockets"]); + expect(result).toSpawn(); + expect(JSON.parse(result.stdout)).toMatchObject({ finished: true, total: 200, max: 64, maxOfEach: [32, 32] }); + }); + + // An event with EPOLLERR keeps its socket open when the receive failed with + // the error or ran dry. An event that stops at its bound saw neither. The + // flag can be stale then: a send from an earlier callback of the same + // iteration takes the pending error. The socket has to stay open. + describe.skipIf(!isLinux)("and keeps the socket open when a send took the pending error", () => { + // A descriptor bun did not create has no error queue to find a report on. + test("of an adopted descriptor", async () => { + const result = await bunRun([fixture, "residual-adopted"]); + expect(result).toSpawn(); + expect(JSON.parse(result.stdout)).toMatchObject({ + residual: true, + finished: true, + closed: false, + total: 40, + max: 32, + }); + }); + + // The kernel queues the report only when the receive buffer has room. + test("of a socket with a full receive buffer", async () => { + const result = await bunRun([fixture, "residual-full-buffer"]); + expect(result).toSpawn(); + const run = JSON.parse(result.stdout); + expect(run).toMatchObject({ residual: true, finished: true, closed: false, max: 32 }); + // More than one event's worth arrived, and the buffer did overflow. + expect(run.total).toBeGreaterThan(32); + expect(run.total).toBeLessThan(run.sent); + }); + }); +}); + // A worker that calls process.exit() from the FIRST 'message' of a batch // leaves a TerminationException pending for the rest of that poll dispatch: // the remaining on_data iterations, the drain, and any recv error must all diff --git a/test/js/bun/udp/udp-recv-budget-fixture.ts b/test/js/bun/udp/udp-recv-budget-fixture.ts new file mode 100644 index 000000000000..af0f7e8f667e --- /dev/null +++ b/test/js/bun/udp/udp-recv-budget-fixture.ts @@ -0,0 +1,210 @@ +// One readable event of a UDP socket hands over at most 32 datagrams. What is +// left in the kernel queue arrives with the next iteration of the event loop. +// Each scenario queues a backlog, counts the datagrams of every iteration, and +// prints one line of JSON. +// +// Spawned by udp_socket.test.ts and dgram.test.ts with the scenario name. +import { exposedInternals } from "bun:internal-for-testing"; +import { createSocket, type Socket } from "node:dgram"; +import { firstUsable, iterateUntil, iterationCounter } from "../../../_util/loop-iterations"; + +const { _createSocketHandle, kStateSymbol } = exposedInternals["internal/dgram"]; + +const HOST = "127.0.0.1"; +const quiet = { data() {}, error() {} }; + +// One sendmmsg. Over loopback on Linux the burst is in the receive queue when +// the call returns. +function queue(sender: Bun.udp.Socket<"buffer">, port: number, count: number) { + const packets: (string | number)[] = []; + for (let i = 0; i < count; i++) packets.push("x", port, HOST); + const sent = sender.sendMany(packets); + if (sent !== count) throw new Error(`sendMany accepted ${sent} of ${count}`); +} + +function bound(socket: Socket) { + return new Promise((resolve, reject) => { + socket.once("error", reject); + socket.bind(0, HOST, () => { + socket.off("error", reject); + resolve(socket.address().port); + }); + }); +} + +type Run = { finished: boolean; total: number; max: number; perIteration: number[] }; + +// A backlog of `count` on one socket. The run is usable when all of it arrived +// and one iteration had the whole bound to read. +const reachedTheBound = (count: number) => (run: Run) => run.finished && run.total === count && run.max >= 32; + +// Linux. The poll reports EPOLLERR for a pending ICMP error, and a send from an +// earlier callback of the same iteration takes that error before the socket's +// own event runs. The error queue is empty and recvmmsg returns only data, so +// the event ends on its bound with no answer about the error. The socket has +// to stay open and deliver the rest. +// +// "adopted": a descriptor bun did not create has no IP_RECVERR, so the kernel +// never queues a report. "full-buffer": a socket bun created, whose receive +// buffer has no room left for the report. +async function residual(kind: "adopted" | "full-buffer", seconds: number) { + const received = iterationCounter(); + let lastDelivery = 0; + let tookTheError = false; + + const peer = await Bun.udpSocket({ hostname: HOST, port: 0, socket: quiet }); + const nudge = await Bun.udpSocket({ hostname: HOST, port: 0, socket: quiet }); + + // Created first and made readable first, so the loop runs this callback + // before the event of `socket`. + const first = await Bun.udpSocket({ + hostname: HOST, + port: 0, + socket: { + data() { + socket.send("y", error => { + tookTheError = (error as NodeJS.ErrnoException | null)?.code === "ECONNREFUSED"; + }); + }, + error() {}, + }, + }); + + const socket = kind === "adopted" ? createSocket("udp4") : createSocket({ type: "udp4", recvBufferSize: 64 * 1024 }); + socket.on("error", () => {}); + socket.on("message", () => { + received.count(); + lastDelivery = performance.now(); + }); + if (kind === "adopted") { + const wrap = _createSocketHandle(HOST, 0, "udp4"); + if (typeof wrap === "number") throw new Error(`_createSocketHandle failed: ${wrap}`); + const { promise, resolve } = Promise.withResolvers(); + socket.once("listening", resolve); + socket.bind({ fd: wrap.fd }); + await promise; + } else { + await bound(socket); + } + await new Promise(resolve => socket.connect(peer.port, HOST, () => resolve())); + const native = socket[kStateSymbol].handle.socket; + const port = socket.address().port; + + // Every socket starts writable. Let those events pass, so that the order of + // the two events below is the order they are raised in. + for (let i = 0; i < 4; i++) await new Promise(resolve => setImmediate(resolve)); + + // All of this before the loop polls again. 400 is more than the 64 KiB + // receive buffer holds, so how many arrive is the kernel's answer. + const sent = kind === "adopted" ? 40 : 400; + nudge.send("go", first.port, HOST); + queue(peer, port, sent); + peer.close(); + socket.send("x"); + + const finished = await iterateUntil(() => { + if (native.closed || received.total === sent) return true; + return kind === "full-buffer" && received.total > 0 && performance.now() - lastDelivery > 300; + }, seconds); + const closed = native.closed; + socket.close(); + first.close(); + nudge.close(); + return { residual: tookTheError, closed, sent, finished, ...received.summary() }; +} + +const scenarios = { + // 100 datagrams queued on a Bun.udpSocket. + backlog: () => + firstUsable(async seconds => { + const received = iterationCounter(); + const receiver = await Bun.udpSocket({ hostname: HOST, port: 0, socket: { data: () => received.count() } }); + const sender = await Bun.udpSocket({ hostname: HOST, port: 0, socket: quiet }); + queue(sender, receiver.port, 100); + const finished = await iterateUntil(() => received.total === 100, seconds); + receiver.close(); + sender.close(); + return { finished, ...received.summary() }; + }, reachedTheBound(100)), + + // The handler of the first datagram queues 100 more on its own socket. The + // event that has read 7 by then has 25 left: not 32, and not 4 more batches. + refill: () => + firstUsable(async seconds => { + const received = iterationCounter(); + const sender = await Bun.udpSocket({ hostname: HOST, port: 0, socket: quiet }); + const receiver = await Bun.udpSocket({ + hostname: HOST, + port: 0, + socket: { + data(socket) { + received.count(); + if (received.total === 1) queue(sender, socket.port, 100); + }, + }, + }); + queue(sender, receiver.port, 7); + const finished = await iterateUntil(() => received.total === 107, seconds); + receiver.close(); + sender.close(); + return { finished, ...received.summary() }; + }, reachedTheBound(107)), + + // 100 datagrams queued on a node:dgram socket. + "dgram-backlog": () => + firstUsable(async seconds => { + const received = iterationCounter(); + const receiver = createSocket("udp4"); + receiver.on("message", () => received.count()); + const port = await bound(receiver); + const sender = await Bun.udpSocket({ hostname: HOST, port: 0, socket: quiet }); + queue(sender, port, 100); + const finished = await iterateUntil(() => received.total === 100, seconds); + receiver.close(); + sender.close(); + return { finished, ...received.summary() }; + }, reachedTheBound(100)), + + // The bound belongs to one socket and one event: two sockets with 100 each + // deliver 32 each in the same iteration. + "dgram-two-sockets": () => + firstUsable( + async seconds => { + const received = iterationCounter(); + const each = [iterationCounter(), iterationCounter()]; + const sender = await Bun.udpSocket({ hostname: HOST, port: 0, socket: quiet }); + const receivers: Socket[] = []; + const ports: number[] = []; + for (const counter of each) { + const receiver = createSocket("udp4"); + receiver.on("message", () => { + counter.count(); + received.count(); + }); + receivers.push(receiver); + ports.push(await bound(receiver)); + } + for (const port of ports) queue(sender, port, 100); + const finished = await iterateUntil(() => received.total === 200, seconds); + for (const receiver of receivers) receiver.close(); + sender.close(); + return { finished, maxOfEach: each.map(counter => counter.summary().max), ...received.summary() }; + }, + run => run.finished && run.total === 200 && run.max >= 64 && run.maxOfEach.every(max => max >= 32), + ), + + "residual-adopted": () => + firstUsable( + seconds => residual("adopted", seconds), + run => run.residual && run.finished, + ), + "residual-full-buffer": () => + firstUsable( + seconds => residual("full-buffer", seconds), + run => run.residual && run.finished, + ), +}; + +const scenario = scenarios[process.argv[2] as keyof typeof scenarios]; +if (!scenario) throw new Error(`unknown scenario: ${process.argv[2]}`); +console.log(JSON.stringify(await scenario())); diff --git a/test/js/bun/udp/udp_socket.test.ts b/test/js/bun/udp/udp_socket.test.ts index 24c3939a767c..e24dff3aabc4 100644 --- a/test/js/bun/udp/udp_socket.test.ts +++ b/test/js/bun/udp/udp_socket.test.ts @@ -4,6 +4,7 @@ import { describe, expect, test } from "bun:test"; import { bunEnv, bunExe, + bunRun, disableAggressiveGCScope, expectRssDeltaBelow, isIPv6, @@ -15,6 +16,8 @@ import { closeSync, openSync } from "node:fs"; import path from "node:path"; import { dataCases, dataTypes } from "./testdata"; +const recvBudgetFixture = path.join(import.meta.dir, "udp-recv-budget-fixture.ts"); + describe("udpSocket()", () => { test.each(["setTTL", "setMulticastTTL"])( "%s does not crash when socket is closed during argument coercion", @@ -673,6 +676,28 @@ describe("udpSocket()", () => { expect(exitCode).toBe(0); }); + // One readable event hands over at most 32 datagrams, the count libuv uses, + // and the rest waits in the kernel for the next iteration of the event loop. + // Without a bound a peer that keeps the queue non-empty keeps the loop inside + // that one event, and no timer, immediate or other socket runs. Each scenario + // of the fixture queues its backlog before the loop polls and reports the + // datagrams of every iteration. + describe.concurrent("a readable event hands over at most 32 datagrams", () => { + test("of a backlog of 100", async () => { + const result = await bunRun([recvBudgetFixture, "backlog"]); + expect(result).toSpawn(); + expect(JSON.parse(result.stdout)).toMatchObject({ finished: true, total: 100, max: 32 }); + }); + + // The first batch of the event is short, so the last one has to ask for + // what is left of the 32 and not for a full batch. + test("when the data handler queues more on its own socket", async () => { + const result = await bunRun([recvBudgetFixture, "refill"]); + expect(result).toSpawn(); + expect(JSON.parse(result.stdout)).toMatchObject({ finished: true, total: 107, max: 32 }); + }); + }); + // sendMany() iterates the input array and may run user JS (array index // getters, port `valueOf()`, address `toString()`). That user JS can // connect or disconnect the socket; sendMany must snapshot the connection diff --git a/test/js/node/quic/quic-endpoint.test.ts b/test/js/node/quic/quic-endpoint.test.ts index edaced224456..b2905e0d6e52 100644 --- a/test/js/node/quic/quic-endpoint.test.ts +++ b/test/js/node/quic/quic-endpoint.test.ts @@ -310,3 +310,48 @@ describe("endpoint.close() while a session is live", () => { expect({ announced, resolved }).toEqual({ announced: 1, resolved: true }); }); }); + +// The endpoint's socket is a UDP socket of the loop like any other: one readable +// event hands over at most 32 packets, and the rest waits for the next +// iteration. Without a bound a transfer holds the loop for as long as the +// packets of the peer keep the socket non-empty. The fixture receives the body +// and stalls once, so that an iteration starts with a backlog. +describe("a readable event of the endpoint's socket", () => { + test("hands over at most 32 packets", async () => { + const port = Promise.withResolvers(); + let report = Promise.withResolvers<{ max: number; stalled: boolean; packets: number }>(); + await using receiver = Bun.spawn({ + cmd: [bunExe(), join(import.meta.dir, "quic-recv-budget-fixture.ts")], + env: bunEnv, + stderr: "pipe", + ipc(message: { port: number } | { max: number; stalled: boolean; packets: number }) { + if ("port" in message) port.resolve(message.port); + else report.resolve(message); + }, + }); + const died = receiver.exited.then(async code => { + throw new Error(`fixture exited with code ${code} before reporting:\n${await receiver.stderr.text()}`); + }); + died.catch(() => {}); + + const client = await connect(`127.0.0.1:${await Promise.race([port.promise, died])}`, { + alpn: "quic-test", + verifyPeer: "manual", + }); + client.closed.catch(() => {}); + await client.opened; + + // A body whose backlog never got to 32 packets says nothing about the + // bound, so it is sent again. + let run = { max: 0, stalled: false, packets: 0 }; + for (let attempt = 0; attempt < 5 && run.max < 32; attempt++) { + report = Promise.withResolvers(); + const stream = await client.createBidirectionalStream({ body: Buffer.alloc(1024 * 1024, "busy") }); + stream.closed.catch(() => {}); + run = await Promise.race([report.promise, died]); + } + client.destroy(); + + expect(run).toMatchObject({ stalled: true, max: 32 }); + }); +}); diff --git a/test/js/node/quic/quic-recv-budget-fixture.ts b/test/js/node/quic/quic-recv-budget-fixture.ts new file mode 100644 index 000000000000..99227a05fa9b --- /dev/null +++ b/test/js/node/quic/quic-recv-budget-fixture.ts @@ -0,0 +1,57 @@ +// Receives stream bodies and reports over IPC the largest number of packets +// the endpoint read in one iteration of the event loop. Spawned by +// quic-endpoint.test.ts, which connects and sends the bodies. +// +// A sender only gets ahead of a receiver that is busy, so the receiver stalls +// once while the body arrives: the sender fills its window in that time, and +// the iteration after the stall starts with a backlog on the socket. +import { getEventLoopStats } from "bun:internal-for-testing"; +import { createPrivateKey } from "node:crypto"; +import { readFileSync } from "node:fs"; +import { join } from "node:path"; +import { listen } from "node:quic"; + +const keysDir = join(import.meta.dir, "..", "test", "fixtures", "keys"); +const key = createPrivateKey(readFileSync(join(keysDir, "agent1-key.pem"))); +const cert = readFileSync(join(keysDir, "agent1-cert.pem")); + +const STALL_AFTER_PACKETS = 200; +const STALL_MS = 50; + +const endpoint = await listen( + session => { + session.closed.catch(() => {}); + session.onstream = async stream => { + stream.closed.catch(() => {}); + const first = Number(endpoint.stats.packetsReceived); + let seen = first; + let iteration = getEventLoopStats().iteration; + let max = 0; + let stalled = false; + let sampling = true; + setImmediate(function sample() { + if (!sampling) return; + const packets = Number(endpoint.stats.packetsReceived); + const now = getEventLoopStats().iteration; + // An immediate runs once per iteration, so this is what one iteration read. + if (now - iteration === 1) max = Math.max(max, packets - seen); + seen = packets; + iteration = now; + if (!stalled && packets - first >= STALL_AFTER_PACKETS) { + stalled = true; + Bun.sleepSync(STALL_MS); + } + setImmediate(sample); + }); + try { + for await (const _ of stream); + } catch { + // The test destroys its session once it has the report. + } + sampling = false; + process.send!({ max, stalled, packets: Number(endpoint.stats.packetsReceived) - first }); + }; + }, + { sni: { "*": { keys: [key], certs: [cert] } }, alpn: ["quic-test"] }, +); +process.send!({ port: endpoint.address.port });