diff --git a/src/jsc/event_loop.rs b/src/jsc/event_loop.rs index be5c9fe4a5a3..f464d1773b35 100644 --- a/src/jsc/event_loop.rs +++ b/src/jsc/event_loop.rs @@ -191,6 +191,8 @@ unsafe extern "Rust" { /// `WTFTimer::run` — `timer` is an erased `*mut bun_runtime::timer::WTFTimer`. /// Defined in `bun_runtime::dispatch`. Link-time resolved. fn __bun_run_wtf_timer(timer: *mut (), vm: *mut VirtualMachine); + /// `timer::All::has_due_regular_timer`. Defined in `bun_runtime::dispatch`. + safe fn __bun_has_due_timer() -> bool; /// Tag-specific shutdown release for a queued-but-never-run task. Called /// from `release_queued_tasks_for_shutdown` (after `shutdown_for_exit`, /// before `destructOnExit`) for every entry left in `self.tasks`. @@ -455,6 +457,24 @@ impl EventLoop { let _ = self.tick_concurrent_with_count(); } + /// Mid-tick re-drain: the ref-count/signal/GC maintenance always runs; + /// only the `concurrent_tasks` pop is skipped when a `setImmediate` is + /// pending or a JS timer is due, so `tick()` returns and `auto_tick*` + /// can run them (Node interleaves timers/check between thread-pool + /// completion batches). `immediate_tasks` is same-thread and checked + /// first so a late cross-thread push cannot slip ahead of it; the + /// `concurrent_tasks` guard keeps the clock read off the path when + /// nothing arrived. + fn tick_concurrent_unless_due(&mut self) { + self.tick_concurrent_maintenance(); + if !self.immediate_tasks.is_empty() + || (!self.concurrent_tasks.is_empty() && __bun_has_due_timer()) + { + return; + } + let _ = self.tick_concurrent_pop_batch(); + } + /// Check whether refConcurrently has been called but the change has not yet been applied to the /// underlying event loop's `active` counter pub fn has_pending_refs(&self) -> bool { @@ -474,7 +494,7 @@ impl EventLoop { } } - pub fn tick_concurrent_with_count(&mut self) -> usize { + fn tick_concurrent_maintenance(&mut self) { self.update_counts(); #[cfg(unix)] @@ -489,7 +509,14 @@ impl EventLoop { } self.run_imminent_gc_timer(); + } + + pub fn tick_concurrent_with_count(&mut self) -> usize { + self.tick_concurrent_maintenance(); + self.tick_concurrent_pop_batch() + } + fn tick_concurrent_pop_batch(&mut self) -> usize { let concurrent = self.concurrent_tasks.pop_batch(); let count = concurrent.count; if count == 0 { @@ -620,7 +647,7 @@ impl EventLoop { loop { while self.tick_with_count(ctx) > 0 { - self.tick_concurrent(); + self.tick_concurrent_unless_due(); self.global_ref().handle_rejected_promises(); } if self @@ -631,7 +658,7 @@ impl EventLoop { self.entered_event_loop_count -= 1; return; } - self.tick_concurrent(); + self.tick_concurrent_unless_due(); if self.tasks.readable_length() > 0 { continue; } @@ -639,7 +666,7 @@ impl EventLoop { } while self.tick_with_count(ctx) > 0 { - self.tick_concurrent(); + self.tick_concurrent_unless_due(); } self.global_ref().handle_rejected_promises(); diff --git a/src/runtime/dispatch.rs b/src/runtime/dispatch.rs index 3dbffa4cb078..31a29a22c069 100644 --- a/src/runtime/dispatch.rs +++ b/src/runtime/dispatch.rs @@ -878,6 +878,19 @@ unsafe fn __bun_cancel_pending_immediate( } } +/// Declared `extern "Rust"` in `bun_jsc::event_loop`; `tick()`'s mid-tick +/// yield check. +#[unsafe(no_mangle)] +fn __bun_has_due_timer() -> bool { + let all = crate::jsc_hooks::timer_all(); + if all.is_null() { + return false; + } + // SAFETY: `all` is the live per-thread `All`; single JS thread, and + // `has_due_regular_timer` only peeks (no heap mutation, no JS re-entry). + unsafe { (*all).has_due_regular_timer() } +} + /// `__bun_run_wtf_timer` body — cast the low-tier erased `*mut ()` to the real /// `crate::timer::WTFTimer` and fire it. /// diff --git a/src/runtime/timer/mod.rs b/src/runtime/timer/mod.rs index fc4cb96665e4..d70ffb806816 100644 --- a/src/runtime/timer/mod.rs +++ b/src/runtime/timer/mod.rs @@ -1013,6 +1013,21 @@ impl All { } } + /// `EventLoop::tick()`'s mid-tick yield probe. WTF timers are skipped to + /// avoid the `wtf_timers` lock; they drain at the next `get_timeout`. + pub(crate) fn has_due_regular_timer(&self) -> bool { + let Some(timer) = self.timers.peek() else { + return false; + }; + // SAFETY: `peek` returns a live heap node. + let next = unsafe { &(*timer).next }; + let next = Timespec { + sec: next.sec, + nsec: next.nsec, + }; + !next.greater(&Timespec::now(TimespecMockMode::ForceRealTime)) + } + /// Pop the next due timer. `now` is filled lazily on first call so we /// don't pay for `clock_gettime` when the heap is empty. fn next(&mut self, has_set_now: &mut bool, now: &mut Timespec) -> Option<*mut EventLoopTimer> { diff --git a/test/js/node/timers/node-timers.test.ts b/test/js/node/timers/node-timers.test.ts index 5ba41145a5af..f2f1e2cba226 100644 --- a/test/js/node/timers/node-timers.test.ts +++ b/test/js/node/timers/node-timers.test.ts @@ -245,3 +245,70 @@ describe.each(["with", "without"])("setImmediate %s timers running", mode => { it("should defer microtasks when an exception is thrown in an immediate", async () => { expect(await bunRun(["run", path.join(import.meta.dir, "timers-immediate-exception-fixture.js")])).toSpawn(); }); + +test("chained thread-pool callbacks yield to due timers", async () => { + // crypto.pbkdf2 completes via enqueue_task_concurrent on every platform + // (fs.read on Windows goes through libuv's callback and would not exercise + // the mid-tick re-drain path). + const script = ` + const crypto = require('crypto'); + let order = ''; + setInterval(() => { order += 'T'; }, 1).unref(); + setTimeout(() => { + (function go(i) { + if (i >= 20) { + const runs = order.match(/R+/g) || []; + const maxRun = Math.max(0, ...runs.map(r => r.length)); + console.log(JSON.stringify({ order, maxRun })); + process.exit(0); + } + crypto.pbkdf2('a', 'b', 1, 8, 'sha256', () => { + order += 'R'; + go(i + 1); + }); + // Spin so the 1ms interval is overdue and the thread-pool job has + // landed before the mid-tick re-drain decision. + const s = Date.now(); while (Date.now() - s < 3); + })(0); + }, 5); + `; + await using proc = Bun.spawn({ + cmd: [bunExe(), "-e", script], + env: bunEnv, + stderr: "pipe", + }); + const [stdout, stderr, exitCode] = await Promise.all([proc.stdout.text(), proc.stderr.text(), proc.exited]); + expect(stderr).toBe(""); + const out = JSON.parse(stdout.trim()); + // Due timers must break the chain every iteration; allow small jitter. + expect({ maxRunAtMost3: out.maxRun <= 3, order: out.order }).toEqual({ maxRunAtMost3: true, order: out.order }); + expect(exitCode).toBe(0); +}); + +test("chained thread-pool callbacks yield to pending setImmediate", async () => { + const script = ` + const crypto = require('crypto'); + let order = ''; + setTimeout(() => { + (function go(i) { + if (i >= 20) { console.log(JSON.stringify({ order })); process.exit(0); } + crypto.pbkdf2('a', 'b', 1, 8, 'sha256', () => { + order += 'R'; + if (i === 0) setImmediate(() => { order += 'I'; }); + go(i + 1); + }); + const s = Date.now(); while (Date.now() - s < 3); + })(0); + }, 1); + `; + await using proc = Bun.spawn({ + cmd: [bunExe(), "-e", script], + env: bunEnv, + stderr: "pipe", + }); + const [stdout, stderr, exitCode] = await Promise.all([proc.stdout.text(), proc.stderr.text(), proc.exited]); + expect(stderr).toBe(""); + const out = JSON.parse(stdout.trim()); + expect(out.order.slice(0, 2)).toBe("RI"); + expect(exitCode).toBe(0); +});