diff --git a/packages/bun-usockets/src/eventing/epoll_kqueue.c b/packages/bun-usockets/src/eventing/epoll_kqueue.c index 068515494743..d59eb47fe22e 100644 --- a/packages/bun-usockets/src/eventing/epoll_kqueue.c +++ b/packages/bun-usockets/src/eventing/epoll_kqueue.c @@ -23,8 +23,12 @@ #include #if defined(LIBUS_USE_EPOLL) || defined(LIBUS_USE_KQUEUE) -void Bun__internal_dispatch_ready_poll(void* loop, void* poll); -// void Bun__internal_dispatch_ready_poll(void* loop, void* poll) {} +void Bun__internal_dispatch_ready_poll(void* loop, void* poll, const void* event); + +#ifdef LIBUS_USE_EPOLL +/* kevent64-only; us_internal_collect_ready_polls ignores it on epoll */ +#define KEVENT_FLAG_IMMEDIATE 0 +#endif #ifndef WIN32 /* Cannot include this one on Windows */ @@ -245,127 +249,97 @@ struct us_loop_t *us_create_loop(void *hint, void (*wakeup_cb)(struct us_loop_t return loop; } -/* Shared dispatch loop for both us_loop_run and us_loop_run_bun_tick */ -static void us_internal_dispatch_ready_polls(struct us_loop_t *loop) { -#ifdef LIBUS_USE_EPOLL - for (loop->current_ready_poll = 0; loop->current_ready_poll < loop->num_ready_polls; loop->current_ready_poll++) { - struct us_poll_t *poll = GET_READY_POLL(loop, loop->current_ready_poll); - if (LIKELY(poll)) { - if (CLEAR_POINTER_TAG(poll) != poll) { - Bun__internal_dispatch_ready_poll(loop, poll); - continue; - } - int events = loop->ready_polls[loop->current_ready_poll].events; - /* Normalize to 0/1 like the kqueue path's EV_ERROR: the value is - * forwarded as a libus close code, and a raw EPOLLERR (8) would - * read as errno 8 (ENOEXEC) in the JS error path. */ - const int error = !!(events & EPOLLERR); - /* A read-side FIN is EPOLLIN + recv()==0; EPOLLHUP means both directions - * are down and is level-triggered, so tag it for the dispatch to close. */ - const int eof = (events & EPOLLHUP) ? LIBUS_POLL_HANGUP : 0; - events &= us_poll_events(poll); - if (events || error || eof) { - us_internal_dispatch_ready_poll(poll, error, eof, events); - } - } - } -#else - /* Kqueue delivers each filter (READ, WRITE, TIMER, etc.) as a separate kevent, - * so the same fd/poll can appear twice in ready_polls. We coalesce them into a - * single set of flags per poll before dispatching, matching epoll's behavior - * where each fd appears once with a combined bitmask. */ - struct kevent_flags { - uint8_t readable : 1; - uint8_t writable : 1; - uint8_t error : 1; - uint8_t eof : 1; - uint8_t send_eof : 1; - uint8_t send_eof_err : 1; - uint8_t eof_err : 1; - uint8_t skip : 1; - }; - - _Static_assert(sizeof(struct kevent_flags) == 1, "kevent_flags must be 1 byte"); - struct kevent_flags coalesced[LIBUS_MAX_READY_POLLS]; /* no zeroing needed — every index is written in the first pass */ - - /* First pass: decode kevents and coalesce same-poll entries */ - for (int i = 0; i < loop->num_ready_polls; i++) { - struct us_poll_t *poll = GET_READY_POLL(loop, i); - if (!poll || CLEAR_POINTER_TAG(poll) != poll) { - coalesced[i] = (struct kevent_flags){ .skip = 1 }; - continue; - } - - const int16_t filter = loop->ready_polls[i].filter; - const uint16_t flags = loop->ready_polls[i].flags; - struct kevent_flags bits = { +#ifdef LIBUS_USE_KQUEUE +enum { + KEVENT_READABLE = 1 << 0, + KEVENT_WRITABLE = 1 << 1, + KEVENT_ERROR = 1 << 2, + /* EV_EOF on EVFILT_READ: the peer's FIN */ + KEVENT_EOF = 1 << 3, + /* EV_EOF on EVFILT_WRITE: SS_CANTSENDMORE (peer gone, or our own shutdown) - + * not a read EOF (libuv kqueue.c ignores it there too) */ + KEVENT_SEND_EOF = 1 << 4, + /* kevent(2): with EV_EOF, fflags carries the socket error. Nonzero alongside a + * write-filter EV_EOF means the connection died hard, not just that our own + * shutdown() set SS_CANTSENDMORE. */ + KEVENT_SEND_EOF_ERR = 1 << 5, + /* Same on the read filter: a reset, which epoll reports as EPOLLERR. */ + KEVENT_EOF_ERR = 1 << 6, +}; + +static unsigned int us_internal_kevent_bits(const struct kevent64_s *ev) { + const int read_eof = (ev->flags & EV_EOF) && ev->filter == EVFILT_READ; + const int write_eof = (ev->flags & EV_EOF) && ev->filter == EVFILT_WRITE; #if defined(__APPLE__) - .readable = (filter == EVFILT_READ || filter == EVFILT_MACHPORT), + return ((ev->filter == EVFILT_READ || ev->filter == EVFILT_MACHPORT) ? KEVENT_READABLE : 0) #else - .readable = (filter == EVFILT_READ || filter == EVFILT_USER), + return ((ev->filter == EVFILT_READ || ev->filter == EVFILT_USER) ? KEVENT_READABLE : 0) +#endif + | (ev->filter == EVFILT_WRITE ? KEVENT_WRITABLE : 0) + | ((ev->flags & EV_ERROR) ? KEVENT_ERROR : 0) + | (read_eof ? KEVENT_EOF : 0) + | (write_eof ? KEVENT_SEND_EOF : 0) + | (write_eof && ev->fflags != 0 ? KEVENT_SEND_EOF_ERR : 0) + | (read_eof && ev->fflags != 0 ? KEVENT_EOF_ERR : 0); +} #endif - .writable = (filter == EVFILT_WRITE), - .error = !!(flags & EV_ERROR), - /* EV_EOF on EVFILT_READ is the peer's FIN; on EVFILT_WRITE it is SS_CANTSENDMORE (peer gone, or our own shutdown) - not a read EOF (libuv kqueue.c ignores it there too). */ - .eof = (flags & EV_EOF) && filter == EVFILT_READ, - .send_eof = (flags & EV_EOF) && filter == EVFILT_WRITE, - /* kevent(2): with EV_EOF, fflags carries the socket error. Nonzero - * alongside a write-filter EV_EOF means the connection died hard, - * not just that our own shutdown() set SS_CANTSENDMORE. */ - .send_eof_err = (flags & EV_EOF) && filter == EVFILT_WRITE && loop->ready_polls[i].fflags != 0, - /* Same on the read filter: a reset, which epoll reports as EPOLLERR. */ - .eof_err = (flags & EV_EOF) && filter == EVFILT_READ && loop->ready_polls[i].fflags != 0, - }; - - /* Look backward for a prior entry with the same poll to coalesce into. - * Kqueue returns at most 2 kevents per fd (READ + WRITE). */ - int merged = 0; - for (int j = i - 1; j >= 0; j--) { - if (!coalesced[j].skip && GET_READY_POLL(loop, j) == poll) { - coalesced[j].readable |= bits.readable; - coalesced[j].writable |= bits.writable; - coalesced[j].error |= bits.error; - coalesced[j].eof |= bits.eof; - coalesced[j].send_eof |= bits.send_eof; - coalesced[j].send_eof_err |= bits.send_eof_err; - coalesced[j].eof_err |= bits.eof_err; - coalesced[i] = (struct kevent_flags){ .skip = 1 }; - merged = 1; - break; - } - } - if (!merged) { - coalesced[i] = bits; - } - } - /* Second pass: dispatch everything in order — tagged pointers and coalesced events */ - for (loop->current_ready_poll = 0; loop->current_ready_poll < loop->num_ready_polls; loop->current_ready_poll++) { - struct us_poll_t *poll = GET_READY_POLL(loop, loop->current_ready_poll); +/* Dispatches the collected entries that have not been dispatched yet, which are + * always exactly [current_ready_poll, num_ready_polls). A handler may tick the + * loop again before it returns (anything that waits on a promise synchronously, + * a debugger pause, ...); that nested tick first dispatches the rest of this + * batch (see us_internal_dispatch_enclosing_ready_polls) and then collects and + * dispatches its own into the same array, so the cursor is advanced before each + * dispatch and re-read after it: when a nested tick has run, the condition below + * finds its batch fully consumed and this one ends. */ +static void us_internal_dispatch_ready_polls(struct us_loop_t *loop) { + while (loop->current_ready_poll < loop->num_ready_polls) { + const int i = loop->current_ready_poll++; + struct us_poll_t *poll = GET_READY_POLL(loop, i); if (!poll) continue; /* Tagged pointers (FilePoll) go through Bun's own dispatch */ if (CLEAR_POINTER_TAG(poll) != poll) { - Bun__internal_dispatch_ready_poll(loop, poll); + Bun__internal_dispatch_ready_poll(loop, poll, &loop->ready_polls[i]); continue; } - struct kevent_flags bits = coalesced[loop->current_ready_poll]; - if (bits.skip) continue; - - int events = (bits.readable ? LIBUS_SOCKET_READABLE : 0) - | (bits.writable ? LIBUS_SOCKET_WRITABLE : 0); +#ifdef LIBUS_USE_EPOLL + int events = loop->ready_polls[i].events; + /* Normalize to 0/1 like the kqueue path's EV_ERROR: the value is + * forwarded as a libus close code, and a raw EPOLLERR (8) would + * read as errno 8 (ENOEXEC) in the JS error path. */ + const int error = !!(events & EPOLLERR); + /* A read-side FIN is EPOLLIN + recv()==0; EPOLLHUP means both directions + * are down and is level-triggered, so tag it for the dispatch to close. */ + const int eof = (events & EPOLLHUP) ? LIBUS_POLL_HANGUP : 0; +#else + /* Kqueue delivers each filter (READ, WRITE) as a separate kevent, so the + * same poll can appear once more later in this batch. Fold that entry + * into this one and blank it, so the poll is dispatched once with the + * combined flags like epoll's single bitmask per fd. */ + unsigned int bits = us_internal_kevent_bits(&loop->ready_polls[i]); + for (int k = i + 1; k < loop->num_ready_polls; k++) { + if (GET_READY_POLL(loop, k) == poll) { + bits |= us_internal_kevent_bits(&loop->ready_polls[k]); + SET_READY_POLL(loop, k, NULL); + /* Kqueue returns at most 2 kevents per fd (READ + WRITE). */ + break; + } + } - int error = bits.error; - int eof = bits.eof; - if (bits.send_eof || bits.eof_err) { + int events = ((bits & KEVENT_READABLE) ? LIBUS_SOCKET_READABLE : 0) + | ((bits & KEVENT_WRITABLE) ? LIBUS_SOCKET_WRITABLE : 0); + int error = !!(bits & KEVENT_ERROR); + int eof = !!(bits & KEVENT_EOF); + if (bits & (KEVENT_SEND_EOF | KEVENT_EOF_ERR)) { int type = us_internal_poll_type(poll); if (type == POLL_TYPE_SOCKET || type == POLL_TYPE_SEMI_SOCKET) { /* Write side dead without our own shutdown() (peer reset / * connect refused), or a read-filter EV_EOF carrying the * socket error in fflags: both are epoll's EPOLLERR. */ - if (bits.send_eof && !bits.send_eof_err && !bits.eof_err && type == POLL_TYPE_SOCKET && - ((struct us_socket_t *) poll)->flags.is_paused) { + if ((bits & KEVENT_SEND_EOF) && !(bits & (KEVENT_SEND_EOF_ERR | KEVENT_EOF_ERR)) && + type == POLL_TYPE_SOCKET && ((struct us_socket_t *) poll)->flags.is_paused) { /* fflags==0 with reads paused is AF_UNIX's graceful peer * close (TCP only gets SS_CANTSENDMORE from RST, which * carries the error): the receive buffer survives, so @@ -379,20 +353,52 @@ static void us_internal_dispatch_ready_polls(struct us_loop_t *loop) { /* Our own shutdown() sets SS_CANTSENDMORE, so a write-side * EV_EOF alone proves nothing here; a socket error in fflags * (either filter) is the peer dying hard: EPOLLERR parity. */ - if (bits.send_eof_err || bits.eof_err) { + if (bits & (KEVENT_SEND_EOF_ERR | KEVENT_EOF_ERR)) { error = 1; } } } +#endif events &= us_poll_events(poll); if (events || error || eof) { us_internal_dispatch_ready_poll(poll, error, eof, events); } } +} + +/* Waits for events (up to `timeout`; NULL waits indefinitely) and makes them the + * batch to dispatch. This reuses ready_polls, so the previous batch must have + * been dispatched in full - see us_internal_dispatch_enclosing_ready_polls. */ +static void us_internal_collect_ready_polls(struct us_loop_t *loop, const struct timespec *timeout, unsigned int kevent_flags) { +#ifdef LIBUS_USE_EPOLL + (void) kevent_flags; + loop->num_ready_polls = bun_epoll_pwait2(loop->fd, loop->ready_polls, LIBUS_MAX_READY_POLLS, timeout); +#else + loop->num_ready_polls = bun_kevent64_wait(loop->fd, loop->ready_polls, LIBUS_MAX_READY_POLLS, kevent_flags, timeout); #endif + if (loop->num_ready_polls < 0) { + loop->num_ready_polls = 0; + } + loop->current_ready_poll = 0; +} + +/* A tick that starts while an enclosing tick is still dispatching (one of its + * handlers is driving the loop and has not returned) first dispatches what that + * tick collected but did not get to: the wait that follows reuses ready_polls, + * and for a one-shot poll the kernel will not report the event a second time. + * Returns whether there was anything, in which case the caller must not park - + * those handlers may already have produced what it is being ticked for. */ +static int us_internal_dispatch_enclosing_ready_polls(struct us_loop_t *loop) { + if (loop->current_ready_poll >= loop->num_ready_polls) { + return 0; + } + us_internal_dispatch_ready_polls(loop); + return 1; } +static const struct timespec zero_timeout = {0, 0}; + /* If the kernel filled our entire buffer, more events are likely already queued. * Re-poll non-blocking and dispatch again before running pre/post callbacks, so a * single tick covers all pending I/O instead of one 1024-event slice per roundtrip. @@ -401,16 +407,8 @@ static void us_internal_dispatch_ready_polls(struct us_loop_t *loop) { static void us_internal_drain_ready_polls(struct us_loop_t *loop) { int drain_count = 48; while (UNLIKELY(loop->num_ready_polls == LIBUS_MAX_READY_POLLS) && --drain_count != 0 && loop->num_polls > 0) { -#ifdef LIBUS_USE_EPOLL - static const struct timespec zero = {0, 0}; - loop->num_ready_polls = bun_epoll_pwait2(loop->fd, loop->ready_polls, LIBUS_MAX_READY_POLLS, &zero); -#else - do { - loop->num_ready_polls = kevent64(loop->fd, NULL, 0, loop->ready_polls, LIBUS_MAX_READY_POLLS, KEVENT_FLAG_IMMEDIATE, NULL); - } while (IS_EINTR(loop->num_ready_polls)); -#endif - if (loop->num_ready_polls <= 0) { - loop->num_ready_polls = 0; + us_internal_collect_ready_polls(loop, &zero_timeout, KEVENT_FLAG_IMMEDIATE); + if (loop->num_ready_polls == 0) { break; } us_internal_dispatch_ready_polls(loop); @@ -443,14 +441,13 @@ void us_loop_run(struct us_loop_t *loop) { struct timespec sweep_ts; const struct timespec *timeout = us_internal_clamp_to_sweep(loop, NULL, &sweep_ts); + unsigned int kevent_flags = 0; + if (us_internal_dispatch_enclosing_ready_polls(loop)) { + timeout = &zero_timeout; + kevent_flags = KEVENT_FLAG_IMMEDIATE; + } - /* Fetch ready polls */ -#ifdef LIBUS_USE_EPOLL - loop->num_ready_polls = bun_epoll_pwait2(loop->fd, loop->ready_polls, LIBUS_MAX_READY_POLLS, timeout); -#else - loop->num_ready_polls = bun_kevent64_wait(loop->fd, loop->ready_polls, LIBUS_MAX_READY_POLLS, 0, timeout); -#endif - + us_internal_collect_ready_polls(loop, timeout, kevent_flags); us_internal_dispatch_ready_polls(loop); us_internal_drain_ready_polls(loop); us_internal_sweep_if_due(loop); @@ -489,6 +486,9 @@ void us_loop_run_bun_tick(struct us_loop_t *loop, const struct timespec* timeout struct timespec sweep_ts; timeout = us_internal_clamp_to_sweep(loop, timeout, &sweep_ts); + if (us_internal_dispatch_enclosing_ready_polls(loop)) { + timeout = &zero_timeout; + } const unsigned int had_wakeups = __atomic_exchange_n(&loop->pending_wakeups, 0, __ATOMIC_ACQUIRE); const int will_idle_inside_event_loop = had_wakeups == 0 && (!timeout || (timeout->tv_nsec != 0 || timeout->tv_sec != 0)); @@ -514,23 +514,16 @@ void us_loop_run_bun_tick(struct us_loop_t *loop, const struct timespec* timeout } } - /* Fetch ready polls */ -#ifdef LIBUS_USE_EPOLL - /* A zero timespec already has a fast path in ep_poll (fs/eventpoll.c): - * it sets timed_out=1 (line 1952) and returns before any scheduler - * interaction (line 1975). No equivalent of KEVENT_FLAG_IMMEDIATE needed. */ - loop->num_ready_polls = bun_epoll_pwait2(loop->fd, loop->ready_polls, LIBUS_MAX_READY_POLLS, timeout); -#else - loop->num_ready_polls = bun_kevent64_wait(loop->fd, loop->ready_polls, LIBUS_MAX_READY_POLLS, - /* When we won't idle (pending wakeups or zero timeout), use KEVENT_FLAG_IMMEDIATE. - * In XNU's kqueue_scan (bsd/kern/kern_event.c): - * - KEVENT_FLAG_IMMEDIATE: returns immediately after kqueue_process() (line 8031) - * - Zero timespec without the flag: falls through to assert_wait_deadline (line 8039) - * and thread_block (line 8048), doing a full context switch cycle (~14us) even - * though the deadline is already in the past. */ - will_idle_inside_event_loop ? 0 : KEVENT_FLAG_IMMEDIATE, - timeout); -#endif + /* When we won't idle (pending wakeups or zero timeout), pass KEVENT_FLAG_IMMEDIATE. + * In XNU's kqueue_scan (bsd/kern/kern_event.c): + * - KEVENT_FLAG_IMMEDIATE: returns immediately after kqueue_process() (line 8031) + * - Zero timespec without the flag: falls through to assert_wait_deadline (line 8039) + * and thread_block (line 8048), doing a full context switch cycle (~14us) even + * though the deadline is already in the past. + * epoll needs no equivalent: a zero timespec already has a fast path in ep_poll + * (fs/eventpoll.c), which sets timed_out=1 (line 1952) and returns before any + * scheduler interaction (line 1975). */ + us_internal_collect_ready_polls(loop, timeout, will_idle_inside_event_loop ? 0 : KEVENT_FLAG_IMMEDIATE); /* Before anything can allocate again. */ if (handed_off) diff --git a/packages/bun-usockets/src/internal/eventing/epoll_kqueue.h b/packages/bun-usockets/src/internal/eventing/epoll_kqueue.h index daad8a330e96..93ac8d161183 100644 --- a/packages/bun-usockets/src/internal/eventing/epoll_kqueue.h +++ b/packages/bun-usockets/src/internal/eventing/epoll_kqueue.h @@ -82,10 +82,9 @@ struct us_loop_t { /* Number of non-fallthrough polls in the loop */ int num_polls; - /* Number of ready polls this iteration */ + /* ready_polls[current_ready_poll .. num_ready_polls) are collected but not + * dispatched yet; the cursor is advanced before an entry is dispatched */ int num_ready_polls; - - /* Current index in list of ready polls */ int current_ready_poll; /* Loop's own file descriptor */ diff --git a/src/io/posix_event_loop.rs b/src/io/posix_event_loop.rs index 1d9cfc2782be..6fc3c54ed5cf 100644 --- a/src/io/posix_event_loop.rs +++ b/src/io/posix_event_loop.rs @@ -1456,9 +1456,13 @@ impl Pollable { } } -// `current_ready_poll`/`ready_polls` only exist on the POSIX uws loop layout; -// on Windows the libuv loop drives readiness, so this entry point is never -// linked there. Restrict to the platforms where the fields are present. +// Called by the epoll/kqueue loop's dispatch; on Windows the libuv loop drives +// readiness and this entry point is never linked. +#[cfg(any(target_os = "linux", target_os = "android"))] +type ReadyEvent = bun_sys::linux::epoll_event; +#[cfg(any(target_os = "macos", target_os = "freebsd"))] +type ReadyEvent = KQueueEvent; + #[cfg(any( target_os = "linux", target_os = "android", @@ -1467,11 +1471,12 @@ impl Pollable { ))] #[unsafe(no_mangle)] /// # Safety -/// uWS C callback: `loop_` is the live per-thread `us_loop_t`; `tagged_pointer` -/// was registered via `Pollable::init` in `register_with_fd`. +/// uWS C callback: `tagged_pointer` was registered via `Pollable::init` in +/// `register_with_fd`; `event` is the loop's ready-poll entry being dispatched. unsafe extern "C" fn Bun__internal_dispatch_ready_poll( - loop_: *mut Loop, + _loop: *mut Loop, tagged_pointer: *mut c_void, + event: *const ReadyEvent, ) { let tag = Pollable::from(tagged_pointer); @@ -1485,14 +1490,10 @@ unsafe extern "C" fn Bun__internal_dispatch_ready_poll( return; } - // SAFETY: `loop_` is the live uws loop. Do *not* materialize `&mut *loop_` - // here — `on_update` (via `__bun_run_file_poll`) re-enters the loop and conjures - // a fresh `&mut Loop` through `EventLoopCtx::platform_event_loop()`; a - // protected `&mut Loop` spanning that call would be SB-UB. Take a short-lived - // `&*loop_` only to copy the POD event onto the stack (the `BackRef`-style - // accessor returns by value), then drop the borrow before dispatching so the - // handler is free to form its own `&mut Loop`. - let ev = unsafe { &*loop_ }.current_ready_event(); + // Copied out: the handler may tick the loop, which reuses the array `event` + // points into. + // SAFETY: points at a live `ready_polls` entry for the duration of this call. + let ev = unsafe { *event }; #[cfg(any(target_os = "macos", target_os = "freebsd"))] file_poll.on_kqueue_event(&ev); diff --git a/src/spawn/process.rs b/src/spawn/process.rs index e870e2e42734..20e8a677fe27 100644 --- a/src/spawn/process.rs +++ b/src/spawn/process.rs @@ -69,20 +69,8 @@ pub use bun_spawn_sys::{ /// tracked process exits and stays readable until the fd is closed, so a /// plain level-triggered watch is sufficient: when the event fires we /// `wait4(WNOHANG)`, and on success we close the pidfd (which removes it -/// from epoll). -/// -/// `EPOLLONESHOT` is actively harmful here: the kernel disarms the fd the -/// instant `epoll_wait` returns it — before user-space has dispatched it. -/// If a poll callback then re-enters `us_loop_run_bun_tick` (e.g. -/// `expect(p).resolves` → `waitForPromise` → `autoTick`, or any other -/// `waitForPromise` path), the inner tick overwrites the shared -/// `loop->ready_polls`/`num_ready_polls`/`current_ready_poll` and the outer -/// dispatch silently skips its remaining events. A dropped one-shot pidfd -/// event is unrecoverable: the fd is disarmed with no re-arm path, so the -/// process's `'exit'` arrives only when the next unrelated timer wakes the -/// loop. Level-triggered makes a dropped slot harmless — the next -/// `epoll_wait` just returns it again. `rewatch_posix` still re-registers -/// defensively if `wait4` returns 0, which is a harmless `CTL_MOD`. +/// from epoll). `rewatch_posix` re-registers if `wait4` returns 0, which is a +/// harmless `CTL_MOD`. /// /// macOS/FreeBSD watch the pid via `EVFILT_PROC` + `NOTE_EXIT`, which is /// inherently once-per-process — keep `EV_ONESHOT` there so the kernel diff --git a/src/uws_sys/Loop.rs b/src/uws_sys/Loop.rs index e94ff26ad119..03d4db3a3416 100644 --- a/src/uws_sys/Loop.rs +++ b/src/uws_sys/Loop.rs @@ -29,10 +29,9 @@ pub struct PosixLoop { /// Number of non-fallthrough polls in the loop pub num_polls: i32, - /// Number of ready polls this iteration + /// `ready_polls[current_ready_poll..num_ready_polls]` are collected but not + /// dispatched yet; the cursor is advanced before an entry is dispatched. pub num_ready_polls: i32, - - /// Current index in list of ready polls pub(crate) current_ready_poll: i32, /// Loop's own file descriptor @@ -123,20 +122,6 @@ impl PosixLoop { self.internal_loop_data.iteration_nr } - /// Copy out the ready-poll event at `current_ready_poll`. - /// - /// Safe back-reference accessor consolidating the C-dispatch - /// `(*loop_).ready_polls[(*loop_).current_ready_poll]` raw-deref pattern - /// into one short-lived `&self` borrow. `EventType` is POD (`epoll_event` - /// / `kevent64_s` / `kevent` — all `Copy` in `libc`), so the by-value - /// return is a stack copy the caller may borrow across re-entrant handler - /// dispatch without aliasing the loop. - #[inline] - pub fn current_ready_event(&self) -> EventType { - let idx = usize::try_from(self.current_ready_poll).expect("int cast"); - self.ready_polls[idx] - } - pub fn inc(&mut self) { bun_core::scoped_log!(Loop, "inc {} + 1 = {}", self.num_polls, self.num_polls + 1); self.num_polls += 1; diff --git a/test/js/bun/net/nested-event-loop-fixture.ts b/test/js/bun/net/nested-event-loop-fixture.ts new file mode 100644 index 000000000000..00cf0e84d779 --- /dev/null +++ b/test/js/bun/net/nested-event-loop-fixture.ts @@ -0,0 +1,214 @@ +// Spawned by socket.test.ts as `bun test `: it has to run under the +// test runner because expect(promise).resolves waits by driving the event loop +// synchronously, which is what nests event-loop ticks inside a socket's data +// callback while the dispatch for that socket is still on the stack. +import { expect, test } from "bun:test"; +import fs from "node:fs"; +import net from "node:net"; +import os from "node:os"; +import path from "node:path"; + +test("a socket closed inside its data callback survives nested event-loop ticks until the dispatch returns", async () => { + using server = Bun.listen({ + hostname: "127.0.0.1", + port: 0, + socket: { + open(socket) { + socket.write("x"); + }, + data() {}, + }, + }); + + for (let i = 0; i < 8; i++) { + const returned = Promise.withResolvers(); + const churn: Promise[] = []; + await Bun.connect({ + hostname: "127.0.0.1", + port: server.port, + socket: { + data(socket) { + // Closing moves the socket to the loop's closed list; it may only be + // freed once this callback (and the dispatch that called it) is done. + socket.terminate(); + // Nested ticks: timers, I/O and the loop's post phase all run here. + expect(new Promise(resolve => setTimeout(resolve, 5))).resolves.toBeUndefined(); + // Allocations of the same size class as the closed socket, so a + // prematurely freed block is likely to be handed out again before + // the outer dispatch looks at it. + for (let j = 0; j < 16; j++) { + churn.push( + Bun.connect({ hostname: "127.0.0.1", port: server.port, socket: { data() {} } }).then(s => s.terminate()), + ); + } + returned.resolve(); + }, + }, + }); + await returned.promise; + await Promise.all(churn); + // Let the outer dispatch unwind and the loop reach its post phase. + await new Promise(resolve => setImmediate(resolve)); + } +}); + +// Two client sockets become readable in the same poll of the loop (the server +// writes to both while this thread is busy). A's data handler then waits, with +// nested event-loop ticks, for B's data handler to have run. B's readiness was +// collected by the outer tick before A's handler started; the nested ticks must +// still deliver it, or A waits for an event the loop already has in hand. +test("an event collected by the outer tick is delivered to a nested tick", async () => { + const accepted: any[] = []; + const bothAccepted = Promise.withResolvers(); + using server = Bun.listen({ + hostname: "127.0.0.1", + port: 0, + socket: { + open(socket) { + accepted.push(socket); + if (accepted.length === 2) bothAccepted.resolve(); + }, + data() {}, + }, + }); + + const order: string[] = []; + const gotB = Promise.withResolvers(); + const aDone = Promise.withResolvers(); + const a = await Bun.connect({ + hostname: "127.0.0.1", + port: server.port, + socket: { + data() { + order.push("a:start"); + // The deadline turns "never delivered" into a failure of this test + // rather than a hang of the whole file; `aDone` settles either way. + const deadline = new Promise((_, reject) => + setTimeout(() => reject(new Error("b's data was not delivered to the nested tick")), 2000), + ); + try { + expect(Promise.race([gotB.promise, deadline])).resolves.toBeUndefined(); + order.push("a:end"); + aDone.resolve(); + } catch (e) { + aDone.reject(e); + } + }, + }, + }); + const b = await Bun.connect({ + hostname: "127.0.0.1", + port: server.port, + socket: { + data() { + order.push("b"); + gotB.resolve(); + }, + }, + }); + await bothAccepted.promise; + + accepted[0].write("a"); + accepted[0].flush(); + accepted[1].write("b"); + accepted[1].flush(); + // Stay busy until both writes have certainly arrived, so the next poll of + // the loop reports both sockets at once. + Bun.sleepSync(100); + + await aDone.promise; + // Which of the two is dispatched first is up to the kernel; either way a's + // wait has to end with b already delivered. + expect(order.indexOf("b")).toBeGreaterThanOrEqual(0); + expect(order.indexOf("b")).toBeLessThan(order.indexOf("a:end")); + a.terminate(); + b.terminate(); +}); + +// Same as above with child-process pipes, which are polled one-shot: unlike a +// socket, the kernel does not report the pipe again to the nested tick, so this +// only holds if the nested tick dispatches what the outer one collected. +test("an event collected by the outer tick is delivered to a nested tick (child process pipes)", async () => { + // Same shape as above with pipe reads instead of sockets: both children + // answer at once, so one poll of the loop collects both stdout reads, and + // whichever is handled first waits (in a nested tick) for the other. + const child = `process.stdout.write("r"); process.stdin.on("data", () => { process.stdout.write("x"); });`; + const spawn = () => + Bun.spawn({ cmd: [process.execPath, "-e", child], stdin: "pipe", stdout: "pipe", stderr: "inherit" }); + await using a = spawn(); + await using b = spawn(); + const ra = a.stdout.getReader(); + const rb = b.stdout.getReader(); + // Both children are up once they have said "r". + expect(new TextDecoder().decode((await ra.read()).value)).toBe("r"); + expect(new TextDecoder().decode((await rb.read()).value)).toBe("r"); + + const order: string[] = []; + const got = { a: Promise.withResolvers(), b: Promise.withResolvers() }; + const handled = (me: "a" | "b", other: "a" | "b") => () => { + order.push(me + ":start"); + got[me].resolve(); + const deadline = new Promise((_, reject) => + setTimeout(() => reject(new Error(other + "'s data was not delivered to the nested tick")), 2000), + ); + expect(Promise.race([got[other].promise, deadline])).resolves.toBeUndefined(); + order.push(me + ":end"); + }; + const done = Promise.all([ra.read().then(handled("a", "b")), rb.read().then(handled("b", "a"))]); + + a.stdin.write("go"); + a.stdin.flush(); + b.stdin.write("go"); + b.stdin.flush(); + // Stay busy until both children have certainly answered, so the next poll + // of the loop reports both pipes at once. + Bun.sleepSync(200); + + await done; + const [first, second] = order[0] === "a:start" ? ["a", "b"] : ["b", "a"]; + expect(order).toEqual([first + ":start", second + ":start", second + ":end", first + ":end"]); + a.kill(); + b.kill(); +}); + +test("closing a pipe server from its connection handler while more accepts are pending, then ticking", async () => { + // Several clients connect at once so one poll of the loop reports several + // pending accepts; the first connection handler closes the server and then + // waits in a nested tick. The remaining accepts are dispatched against a + // listener that is going away underneath them. + const name = + process.platform === "win32" + ? String.fromCharCode(92, 92, 46, 92) + "pipe" + String.fromCharCode(92) + "nested-close-" + process.pid + : path.join(os.tmpdir(), "nested-close-" + process.pid + ".sock"); + let accepted = 0; + const closed = Promise.withResolvers(); + const server = net.createServer(c => { + accepted++; + c.on("error", () => {}); + c.destroy(); + if (accepted === 1) { + server.close(() => closed.resolve()); + expect(new Promise(resolve => setTimeout(resolve, 20))).resolves.toBeUndefined(); + } + }); + const unlink = () => { + if (process.platform !== "win32") fs.rmSync(name, { force: true }); + }; + unlink(); + await new Promise(resolve => server.listen(name, resolve)); + const clients = Array.from( + { length: 8 }, + () => + new Promise(resolve => { + const c = net.connect(name); + c.on("error", () => resolve()); + c.on("close", () => resolve()); + }), + ); + // Let all eight connects land before the loop polls again. + Bun.sleepSync(100); + await Promise.all(clients); + await closed.promise; + unlink(); + expect(accepted).toBeGreaterThanOrEqual(1); +}); diff --git a/test/js/bun/net/socket.test.ts b/test/js/bun/net/socket.test.ts index 6619e16894c5..56c49ff157eb 100644 --- a/test/js/bun/net/socket.test.ts +++ b/test/js/bun/net/socket.test.ts @@ -4662,3 +4662,25 @@ describe.concurrent("a socket closed by data() while its peer's reset is being d expect(exitCode).toBe(0); }); }); + +describe.concurrent("socket handlers that re-enter the event loop before returning", () => { + // The fixture runs under `bun test` so that expect(promise).resolves can drive + // nested event-loop ticks from inside a data callback. It covers a socket that + // is closed by its own data() and must stay allocated until that dispatch has + // returned, and an event the outer tick already collected that the nested tick + // has to deliver. Windows: #40023. + it.skipIf(isWindows)("keeps the socket alive and the collected events visible", async () => { + await using proc = Bun.spawn({ + cmd: [bunExe(), "test", fileURLToPath(new URL("./nested-event-loop-fixture.ts", import.meta.url))], + env: bunEnv, + stdout: "pipe", + stderr: "pipe", + }); + const [stdout, stderr, exitCode] = await Promise.all([proc.stdout.text(), proc.stderr.text(), proc.exited]); + // stdout carries only the runner's version banner; results go to stderr. + expect(stdout).toMatch(/^bun test v\S+ \(\S+\)\n$/); + expect(stderr).toContain(" 4 pass"); + expect(proc.signalCode).toBeNull(); + expect(exitCode).toBe(0); + }); +}); diff --git a/test/js/bun/spawn/spawn.test.ts b/test/js/bun/spawn/spawn.test.ts index 2868b74f4a56..909d63caeb7b 100644 --- a/test/js/bun/spawn/spawn.test.ts +++ b/test/js/bun/spawn/spawn.test.ts @@ -807,6 +807,60 @@ describe("should not hang", () => { } }); +// Two children answer at once, so one poll of the event loop collects both +// stdout pipes. The read handler that runs first calls Bun.build() with an async +// plugin setup(), which Bun waits for by ticking the event loop from inside that +// handler, and setup() needs the other child's output: the nested tick has to +// deliver an event the outer tick already collected - pipe polls are one-shot, +// so nothing will report it again. Windows: #40023. +it.skipIf(isWindows)( + "a stdout chunk the loop already collected is delivered while another stdout handler waits in Bun.build()", + async () => { + const script = /* js */ ` + const child = 'process.stdout.write("r"); process.stdin.on("data", () => { process.stdout.write("x"); });'; + const spawn = () => Bun.spawn({ cmd: [process.execPath, "-e", child], stdin: "pipe", stdout: "pipe", stderr: "inherit" }); + const a = spawn(), b = spawn(); + const ra = a.stdout.getReader(), rb = b.stdout.getReader(); + await ra.read(); + await rb.read(); + const got = { a: Promise.withResolvers(), b: Promise.withResolvers() }; + const order = []; + const handled = (me, other) => async () => { + order.push(me + ":start"); + got[me].resolve(); + if (order.length === 1) { + // Only the wait for setup() matters; the entrypoint does not exist. + await Bun.build({ + entrypoints: ["./does-not-exist.ts"], + plugins: [{ + name: "waits-for-" + other, + async setup() { + const lost = await Promise.race([got[other].promise, new Promise(r => setTimeout(r, 3000, true).unref())]); + if (lost) { console.log(order.join(" ") + " - " + other + " was never delivered"); process.exit(1); } + }, + }], + }).catch(() => {}); + } + order.push(me + ":end"); + }; + const done = Promise.all([ra.read().then(handled("a", "b")), rb.read().then(handled("b", "a"))]); + a.stdin.write("go"); a.stdin.flush(); + b.stdin.write("go"); b.stdin.flush(); + // Stay busy until both children have certainly answered. + Bun.sleepSync(200); + await done; + // Which pipe is dispatched first is up to the kernel. + const first = order[0][0]; + console.log(order.map(s => (s[0] === first ? "1" : "2") + s.slice(1)).join(" ")); + a.kill(); b.kill(); + `; + await using proc = spawn({ cmd: [bunExe(), "-e", script], env: bunEnv, stdout: "pipe", stderr: "inherit" }); + const [stdout, exitCode] = await Promise.all([proc.stdout.text(), proc.exited]); + expect(stdout).toBe("1:start 2:start 2:end 1:end\n"); + expect(exitCode).toBe(0); + }, +); + describe("unref() + .exited with nothing else ref'd (Windows)", () => { // Windows: with only an unref'd uv_process_t left, uv_run() used to skip its // body and never dequeue the IOCP exit packet, so these children busy-spun