Something went wrong. Try again.
Monorepo for the nibling.run platform
Something went wrong. Try again.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323// Event loop (runtime worker): Node's phases (timers → poll → check) run one guest// callback per browser task, so native microtasks drain between callbacks, and the// nextTick queue drains after every callback. Each phase works on a snapshot taken// at its entry: a timer or immediate queued from inside the phase waits for the next// iteration. Liveness: ref'd timers and immediates, poll completions and in-flight// async syscalls (hold()/release()) keep the loop alive. When none remain after the// entry is done, the loop emits beforeExit; if that schedules nothing, it exits.
// The loop arms itself with the browser's timers. They are captured before guest// globals replace setTimeout/clearTimeout with the loop-backed versions.const nativeSetTimeout = globalThis.setTimeout.bind(globalThis);const nativeClearTimeout = globalThis.clearTimeout.bind(globalThis);
// Async context, stage 1: every callback the loop itself schedules (a timer repeat,// an immediate, a nextTick, a poll-phase completion) captures the frame current when// it was scheduled and runs in it, restoring whatever was current before. See// src/runtime/async_hooks.ts and DESIGN.md's "Async context".import { capture, runInFrame } from "./async_hooks";
function bindFrame<F extends (...args: any[]) => void>(fn: F): F { const frame = capture(); return ((...args: unknown[]) => runInFrame(frame, fn, undefined, args)) as F;}
export interface LoopIO { /** Emit process 'beforeExit' (runs as a guest callback). */ beforeExit(): void; /** The loop is idle and beforeExit scheduled nothing: Node's natural exit. */ exit(): void; /** A guest callback threw. True when an uncaughtException listener took it; * false means the process dies and no further callback may run. */ onUncaught(error: unknown): boolean; /** At poll-phase entry: enqueue completions for events that wait outside the loop * (pending signals in the mailbox). */ beforePoll(): void;}
export interface Timer { readonly id: number; fn: () => void; delay: number; repeat: boolean; ref: boolean; /** Expiry on the Date.now() clock. */ at: number; /** Creation or refresh order: breaks ties between timers with the same expiry. */ seq: number;}
export interface Immediate { fn: () => void; ref: boolean;}
export interface Loop { nextTick(fn: () => void): void; addTimer(fn: () => void, delay: number, repeat: boolean): Timer; timerById(id: number): Timer | undefined; clearTimer(timer: Timer): void; refreshTimer(timer: Timer): void; setTimerRef(timer: Timer, ref: boolean): void; addImmediate(fn: () => void): Immediate; clearImmediate(immediate: Immediate): void; setImmediateRef(immediate: Immediate, ref: boolean): void; enqueueCompletion(fn: () => void): void; /** Run the CJS entry as a loop callback: ticks it queues drain before any microtask. */ runMain(fn: () => void): void; setMainDone(): void; /** Track in-flight async syscalls so an idle-looking loop doesn't exit early. */ hold(): void; release(): void; /** Run a step soon even if nothing is queued, so beforePoll looks again. */ interrupt(): void; /** Milliseconds spent inside a guest callback (active) vs. not (idle), since the * loop started — perf_hooks.eventLoopUtilization's only honest inputs. */ utilization(): { idle: number; active: number }; /** process.getActiveResourcesInfo(): one label per handle this loop itself knows * about. Partial — there's no TCP/TTY/etc. handle table here, just timers. */ activeResourcesInfo(): string[];}
type Phase = "timers" | "poll" | "check";const NEXT_PHASE: Record<Phase, Phase> = { timers: "poll", poll: "check", check: "timers" };
export function installLoop(io: LoopIO): Loop { const nextTicks: Array<() => void> = []; const timers = new Map<number, Timer>(); const immediates = new Set<Immediate>(); const poll: Array<() => void> = []; let nextTimerId = 1; let seq = 0; let holds = 0; let mainDone = false; let halted = false; /** True while a guest callback (and the tick drain after it) is on the stack. */ let inCallback = false; /** beforeExit ran and no loop callback has run since: the next idle step exits. */ let quiet = false; const startedAt = performance.now(); let activeMs = 0;
// The current phase's snapshot, consumed one job per step. let phase: Phase = "check"; let batch: Array<() => void> = [];
// Steps are posted on a MessageChannel: a task with no timer clamp. const channel = new MessageChannel(); let stepPosted = false; let armed: ReturnType<typeof nativeSetTimeout> | null = null; channel.port1.onmessage = () => step();
function post(): void { if (stepPosted || halted) return; stepPosted = true; channel.port2.postMessage(null); }
function drainNextTicks(): void { while (nextTicks.length) nextTicks.shift()!(); }
function runGuest(fn: () => void): void { if (halted) return; inCallback = true; const startedThisRun = performance.now(); try { fn(); drainNextTicks(); } catch (e) { // Callbacks already queued must not run after the process died. if (!io.onUncaught(e)) halted = true; } finally { inCallback = false; activeMs += performance.now() - startedThisRun; } }
function hasRef(): boolean { if (holds > 0 || poll.length) return true; for (const t of timers.values()) if (t.ref) return true; for (const im of immediates) if (im.ref) return true; return false; }
function snapshot(p: Phase): Array<() => void> { if (p === "timers") { const now = Date.now(); const due = [...timers.values()].filter((t) => t.at <= now).sort((a, b) => a.at - b.at || a.seq - b.seq); return due.map((t) => () => { // Cleared or refreshed since the snapshot: not due any more. if (timers.get(t.id) !== t || t.at > now) return; if (t.repeat) { t.at = Date.now() + t.delay; t.seq = seq++; } else { timers.delete(t.id); } runGuest(t.fn); }); } if (p === "poll") { io.beforePoll(); return poll.splice(0).map((fn) => () => runGuest(fn)); } return [...immediates].map((im) => () => { if (!immediates.delete(im)) return; im.ref = false; // Node: a run or cleared Immediate reports hasRef() false runGuest(im.fn); }); }
/** Move to the next phase with work, at most one full iteration ahead. */ function advance(): void { for (let i = 0; i < 3 && batch.length === 0; i++) { phase = NEXT_PHASE[phase]; batch = snapshot(phase); } }
function step(): void { stepPosted = false; if (halted) return; if (nextTicks.length) { // Queued from a microtask: they run once the microtask queue has drained. runGuest(() => {}); wake(); return; } if (batch.length === 0) advance(); const job = batch.shift(); if (job) { quiet = false; job(); wake(); return; } if (!mainDone || hasRef()) { wake(); return; } if (quiet) { io.exit(); return; } runGuest(io.beforeExit); quiet = true; // A step, not an exit: microtasks from the listeners may still schedule work. post(); }
/** Decide when the next step runs: now, when the earliest timer expires, or when * something new arrives (completions and holds call back in here). */ function wake(): void { if (halted) return; if (armed !== null) { nativeClearTimeout(armed); armed = null; } if (nextTicks.length || batch.length || poll.length || immediates.size) return post(); let earliest: Timer | null = null; for (const t of timers.values()) if (!earliest || t.at < earliest.at) earliest = t; // Unref'd timers still fire while something else keeps the loop alive. if (earliest && (hasRef() || !mainDone)) { const wait = earliest.at - Date.now(); if (wait <= 0) return post(); armed = nativeSetTimeout(() => { armed = null; post(); }, wait); return; } if (mainDone && !hasRef()) post(); }
return { nextTick(fn) { nextTicks.push(bindFrame(fn)); // Inside a callback runGuest drains synchronously; from a microtask nothing // else would, so a step drains them after the microtask queue. if (!inCallback) post(); }, addTimer(fn, delay, repeat) { // A repeat reuses the same wrapped fn, so every tick of a setInterval runs in // the frame current when it was created, not when it was (re)scheduled. const t: Timer = { id: nextTimerId++, fn: bindFrame(fn), delay, repeat, ref: true, at: Date.now() + delay, seq: seq++ }; timers.set(t.id, t); wake(); return t; }, timerById(id) { return timers.get(id); }, clearTimer(t) { if (timers.get(t.id) === t) timers.delete(t.id); wake(); }, refreshTimer(t) { t.at = Date.now() + t.delay; t.seq = seq++; // Node re-arms a fired one-shot timer on refresh. timers.set(t.id, t); wake(); }, setTimerRef(t, ref) { t.ref = ref; wake(); }, addImmediate(fn) { const im: Immediate = { fn: bindFrame(fn), ref: true }; immediates.add(im); wake(); return im; }, clearImmediate(im) { immediates.delete(im); im.ref = false; wake(); }, setImmediateRef(im, ref) { im.ref = ref; wake(); }, enqueueCompletion(fn) { poll.push(bindFrame(fn)); wake(); }, runMain(fn) { if (inCallback) throw new Error("runMain: already inside a loop callback"); inCallback = true; try { fn(); drainNextTicks(); } finally { inCallback = false; } }, setMainDone() { mainDone = true; wake(); }, hold() { holds++; }, release() { holds = Math.max(0, holds - 1); wake(); }, interrupt() { post(); }, utilization() { const elapsed = performance.now() - startedAt; return { idle: Math.max(0, elapsed - activeMs), active: activeMs }; }, activeResourcesInfo() { return [ ...Array.from(timers.values(), () => "Timeout"), ...Array.from(immediates, () => "Immediate"), ]; }, };}