Something went wrong. Try again.
Monorepo for the nibling.run platform
Something went wrong. Try again.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472// node:worker_threads ports. A pair is one native MessageChannel plus a shared queue per// direction (a growable SharedArrayBuffer). A message without transferables is cloned// into the queue (clone.ts) and the native port only rings the receiver's doorbell; a// message with transferables, or holding what only a native clone can carry, rides the// native port itself. Every message takes the next number in its direction's sequence,// so the receiver delivers both paths in send order. The queue is what lets// receiveMessageOnPort take a message synchronously, from a thread that never yields to// its event loop (esbuild-wasm's sync API blocks in Atomics.wait, then reads its port).//// Records, not bare values, travel the pair: user messages and, on a Worker's channel,// the thread's lifecycle (online, stdio, its uncaught error) and port close.import type { PortDescriptor } from "../shared/protocol";import { capture, runInFrame, type Context } from "./async_hooks";import { NeedsNative, deserialize, kTransferOnly, serialize } from "./clone";import { EventEmitter, type EventEmitterConstructor } from "./events";import type { Loop } from "./loop";
export type PortRecord = | { k: "m"; v: unknown } | { k: "close" } | { k: "online" } | { k: "stdio"; fd: 1 | 2; bytes: Uint8Array } | { k: "error"; v: unknown };
/** What crosses the native port: a doorbell for the queue, or a whole record. */type Wire = { bell: true } | { seq: number; rec: PortRecord };
// ---- the shared queue: one direction of a pair ----// Int32 header [0] lock [1] head [2] tail, then records [u32 length][u32 seq][bytes]// from byte 16. The writer compacts, then grows, when a record does not fit; the reader// resets both ends when it empties the queue. Critical sections are copies, so the// lock spins with a short wait.
const DATA = 16;const INITIAL = 64 * 1024;const MAX = 256 * 1024 * 1024;
export function newQueue(): SharedArrayBuffer { return new SharedArrayBuffer(DATA + INITIAL, { maxByteLength: DATA + MAX });}
function lock(h: Int32Array): void { while (Atomics.compareExchange(h, 0, 0, 1) !== 0) Atomics.wait(h, 0, 1, 1);}
function unlock(h: Int32Array): void { Atomics.store(h, 0, 0); Atomics.notify(h, 0, 1);}
function push(sab: SharedArrayBuffer, seq: number, bytes: Uint8Array): void { const h = new Int32Array(sab, 0, 4); lock(h); try { let head = h[1]!; let tail = h[2]!; const need = 8 + bytes.length; if (tail + need > sab.byteLength - DATA && head > 0) { new Uint8Array(sab, DATA).copyWithin(0, head, tail); tail -= head; head = 0; } if (tail + need > sab.byteLength - DATA) { const want = DATA + Math.max(2 * (sab.byteLength - DATA), tail + need); if (want > sab.maxByteLength) throw new RangeError(`ERR_NIB_PORT_FULL: a port's queue cannot hold another ${bytes.length} bytes`); sab.grow(want); } const view = new DataView(sab, DATA); view.setUint32(tail, bytes.length, true); view.setUint32(tail + 4, seq, true); new Uint8Array(sab, DATA + tail + 8, bytes.length).set(bytes); h[1] = head; h[2] = tail + need; } finally { unlock(h); }}
/** The record at the head when it is number `seq`, else null (empty, or `seq` is a * native record still in flight). */function shift(sab: SharedArrayBuffer, seq: number): Uint8Array | null { const h = new Int32Array(sab, 0, 4); lock(h); try { const head = h[1]!; const tail = h[2]!; if (head === tail) return null; const view = new DataView(sab, DATA); if (view.getUint32(head + 4, true) !== seq) return null; const len = view.getUint32(head, true); // slice() copies out of shared memory, which TextDecoder will not read. const bytes = new Uint8Array(sab, DATA + head + 8, len).slice(); const next = head + 8 + len; h[1] = next === tail ? 0 : next; if (next === tail) h[2] = 0; return bytes; } finally { unlock(h); }}
function queued(sab: SharedArrayBuffer): boolean { const h = new Int32Array(sab, 0, 4); return Atomics.load(h, 1) !== Atomics.load(h, 2);}
// ---- the runtime's loop, set once at boot ----
let loop: Loop | null = null;export function installPorts(l: Loop): void { loop = l;}
type NativePort = globalThis.MessagePort;const BELL: Wire = { bell: true };
/** Chromium's DataCloneError names the call it came from; Node's does not. */function nodeCloneError(e: unknown): unknown { if (e instanceof DOMException && e.name === "DataCloneError") { return new DOMException(e.message.replace(/^Failed to execute 'postMessage' on '\w+': /, ""), "DataCloneError"); } return e;}
export class MessagePort extends (EventEmitter as unknown as EventEmitterConstructor) { declare [kTransferOnly]: true; #native: NativePort | null; #rx: SharedArrayBuffer; #tx: SharedArrayBuffer; #rxSeq: number; #txSeq: number; /** Records that came natively, by sequence number, until their turn. */ #inbox = new Map<number, PortRecord>(); /** Records taken from the queues and not yet delivered. */ #pending: PortRecord[] = []; #started = false; #closed = false; #ref = true; #holding = false; #drainQueued = false; /** addEventListener's registry, keyed by type then the original listener, so * removeEventListener finds the EventEmitter listener it wraps (DESIGN: ports had * no EventTarget API; this is the thin layer over the existing on()/emit()). */ #wrapped = new Map<string, Map<EventListenerOrEventListenerObject, (...args: unknown[]) => void>>(); #onmessage: ((ev: MessageEvent) => void) | null = null; #onmessageerror: ((ev: MessageEvent) => void) | null = null; /** A Worker's end of its channel: every record goes here, lifecycle records ahead * of the messages taken with them, as Node drains a worker's internal port before * its public one. */ onRecord: ((rec: PortRecord) => void) | null = null; /** The frame this port was created in: worker_threads.ts overrides delivery via * onRecord (and captures its own, Worker-scoped frame there), so this only governs * a plain MessageChannel's own 'message' event. */ readonly #frame: Context;
/** @internal Ports come from MessageChannel or arrive in a message. */ constructor(desc: PortDescriptor) { super(); Object.defineProperty(this, kTransferOnly, { value: true }); this.#frame = capture(); this.#native = desc.native; this.#rx = desc.rx; this.#tx = desc.tx; this.#rxSeq = desc.rxSeq; this.#txSeq = desc.txSeq; this.#native.onmessage = (ev: MessageEvent<Wire>) => this.#arrived(ev.data); // Node's ports start, and hold the loop, while they have a 'message' listener. this.on("newListener", (event: string | symbol) => { if (event === "message" && this.listenerCount("message") === 0) { this.#ref = true; this.start(); } }); this.on("removeListener", (event: string | symbol) => { if (event === "message" && this.listenerCount("message") === 0) { this.#started = false; this.#ref = false; this.#updateHold(); } }); }
postMessage(value: unknown, transferList?: readonly unknown[] | { transfer?: readonly unknown[] }): void { const transfer = Array.isArray(transferList) ? transferList : (transferList as { transfer?: unknown[] } | undefined)?.transfer ?? []; if (this.#closed) return; this.send({ k: "m", v: value }, transfer); }
/** @internal Send one record, by the queue when it can go there. */ send(rec: PortRecord, transfer: readonly unknown[] = []): void { if (!this.#native) return; if (transfer.length === 0) { let bytes: Uint8Array | null = null; try { bytes = serialize(rec, { errorProps: rec.k === "error" }); } catch (e) { if (!(e instanceof NeedsNative)) throw e; } if (bytes) { push(this.#tx, this.#txSeq, bytes); this.#txSeq++; this.#native.postMessage(BELL); return; } } const { value, natives } = exportPorts(rec, transfer); try { this.#native.postMessage({ seq: this.#txSeq, rec: value as PortRecord } satisfies Wire, natives as Transferable[]); } catch (e) { throw nodeCloneError(e); } this.#txSeq++; }
start(): void { if (this.#closed) return; this.#started = true; this.#updateHold(); this.#queueDrain(); }
close(): void { if (this.#closed) return; this.send({ k: "close" }); this.#shut(); }
ref(): this { this.#ref = true; this.#updateHold(); return this; }
unref(): this { this.#ref = false; this.#updateHold(); return this; }
hasRef(): boolean { return this.#holding; }
/** The EventTarget half of a port (Node's MessagePort is both): wraps a listener * to receive a MessageEvent, backed by the same on()/emit() the Worker's internal * record delivery and the 'newListener' auto-start already use. */ addEventListener(type: string, listener: EventListenerOrEventListenerObject | null, options?: boolean | AddEventListenerOptions): void { if (listener === null || (typeof listener !== "function" && typeof listener.handleEvent !== "function")) return; let byType = this.#wrapped.get(type); if (!byType) { byType = new Map(); this.#wrapped.set(type, byType); } if (byType.has(listener)) return; // DOM: adding the same listener twice is a no-op. const once = typeof options === "object" && options?.once === true; const call = typeof listener === "function" ? listener : listener.handleEvent.bind(listener); const wrapped = (data: unknown) => { if (once) this.removeEventListener(type, listener); call(new MessageEvent(type, { data })); }; byType.set(listener, wrapped); this.on(type, wrapped); }
removeEventListener(type: string, listener: EventListenerOrEventListenerObject | null): void { const wrapped = listener === null ? undefined : this.#wrapped.get(type)?.get(listener); if (!wrapped) return; this.#wrapped.get(type)!.delete(listener!); this.off(type, wrapped); }
dispatchEvent(event: Event): boolean { this.emit(event.type, (event as MessageEvent).data); return true; }
/** `port.onmessage = fn` is Node's sugar for addEventListener("message", fn): * one slot, replaced on reassignment, which is also why it starts the port (the * same 'newListener' hook that a plain on("message", ...) trips). */ get onmessage(): ((ev: MessageEvent) => void) | null { return this.#onmessage; }
set onmessage(fn: ((ev: MessageEvent) => void) | null) { if (this.#onmessage) this.removeEventListener("message", this.#onmessage as EventListener); this.#onmessage = typeof fn === "function" ? fn : null; if (this.#onmessage) this.addEventListener("message", this.#onmessage as EventListener); }
get onmessageerror(): ((ev: MessageEvent) => void) | null { return this.#onmessageerror; }
set onmessageerror(fn: ((ev: MessageEvent) => void) | null) { if (this.#onmessageerror) this.removeEventListener("messageerror", this.#onmessageerror as EventListener); this.#onmessageerror = typeof fn === "function" ? fn : null; if (this.#onmessageerror) this.addEventListener("messageerror", this.#onmessageerror as EventListener); }
/** @internal receiveMessageOnPort: the next message, taken now, or undefined. */ receiveSync(): { message: unknown } | undefined { for (;;) { const rec = this.#pending.shift() ?? this.#take(); if (!rec) return undefined; if (rec.k === "m") return { message: rec.v }; this.#deliver(rec); } }
/** @internal Hand this end to another thread: it stops here. */ detach(): PortDescriptor { const native = this.#native; if (!native) throw new DOMException("MessagePort was already transferred", "DataCloneError"); native.onmessage = null; this.#native = null; this.#started = false; this.#closed = true; this.#updateHold(); return { nibPort: true, native, rx: this.#rx, tx: this.#tx, rxSeq: this.#rxSeq, txSeq: this.#txSeq }; }
/** @internal Records are waiting on a native one still in flight. */ stalled(): boolean { return queued(this.#rx) || this.#inbox.size > 0; }
/** @internal Deliver everything available now; the Worker calls it when its thread * has ended, before 'exit'. */ drainNow(): void { for (let rec = this.#pending.shift() ?? this.#take(); rec; rec = this.#pending.shift() ?? this.#take()) this.#deliver(rec); }
#arrived(wire: Wire): void { if (!("bell" in wire)) this.#inbox.set(wire.seq, wire.rec); this.#queueDrain(); this.onWake?.(); }
/** @internal Called on every arrival: a Worker waiting to emit 'exit' retries. */ onWake: (() => void) | null = null;
#take(): PortRecord | undefined { const native = this.#inbox.get(this.#rxSeq); if (native) { this.#inbox.delete(this.#rxSeq); this.#rxSeq++; return importPorts(native) as PortRecord; } const bytes = shift(this.#rx, this.#rxSeq); if (!bytes) return undefined; this.#rxSeq++; return deserialize(bytes) as PortRecord; }
#queueDrain(): void { if (this.#drainQueued || !this.#started || !loop) return; this.#drainQueued = true; loop.enqueueCompletion(() => { this.#drainQueued = false; this.#drainOne(); }); }
/** One record per loop callback, so ticks and microtasks run between messages. */ #drainOne(): void { if (!this.#started) return; if (this.#pending.length === 0) { const batch: PortRecord[] = []; for (let rec = this.#take(); rec; rec = this.#take()) batch.push(rec); // Lifecycle records first: Node drains a worker's internal port before its public one. this.#pending = this.onRecord ? [...batch.filter((r) => r.k !== "m"), ...batch.filter((r) => r.k === "m")] : batch; } const rec = this.#pending.shift(); if (rec) this.#deliver(rec); // Later arrivals ring their own doorbell. if (this.#pending.length) this.#queueDrain(); }
#deliver(rec: PortRecord): void { if (this.onRecord) return this.onRecord(rec); switch (rec.k) { case "m": runInFrame(this.#frame, () => this.emit("message", rec.v)); return; case "close": this.#shut(); return; default: return; } }
#shut(): void { if (this.#closed) return; this.#closed = true; this.#started = false; this.#updateHold(); this.#native?.close(); this.#native = null; loop?.enqueueCompletion(() => this.emit("close")); }
#updateHold(): void { const hold = this.#started && this.#ref && !this.#closed; if (hold === this.#holding || !loop) return; this.#holding = hold; if (hold) loop.hold(); else loop.release(); }}
/** A connected pair, as `new MessageChannel()` gives the guest. */export function portPair(): [MessagePort, MessagePort] { const ch = new globalThis.MessageChannel(); const a = newQueue(); const b = newQueue(); return [ new MessagePort({ nibPort: true, native: ch.port1, rx: a, tx: b, rxSeq: 0, txSeq: 0 }), new MessagePort({ nibPort: true, native: ch.port2, rx: b, tx: a, rxSeq: 0, txSeq: 0 }), ];}
export class MessageChannel { readonly port1: MessagePort; readonly port2: MessagePort; constructor() { [this.port1, this.port2] = portPair(); }}
/** Replace transferred node ports in `value` with descriptors, and list the native * objects a native postMessage must transfer. */export function exportPorts(value: unknown, transfer: readonly unknown[]): { value: unknown; natives: unknown[] } { const swapped = new Map<object, PortDescriptor>(); const natives = transfer.map((t) => { if (!(t instanceof MessagePort)) return t; const desc = t.detach(); swapped.set(t, desc); return desc.native; }); return { value: swapped.size ? mapPlain(value, (o) => swapped.get(o)) : value, natives };}
/** Turn descriptors in a natively received value back into node ports. */export function importPorts(value: unknown): unknown { return mapPlain(value, (o) => ((o as Partial<PortDescriptor>).nibPort === true ? new MessagePort(o as PortDescriptor) : undefined));}
/** Copy arrays and plain objects in `value`, replacing each object `swap` answers for. */function mapPlain(value: unknown, swap: (o: object) => unknown): unknown { const memo = new Map<object, unknown>(); const walk = (v: unknown): unknown => { if (v === null || typeof v !== "object") return v; const replaced = swap(v); if (replaced !== undefined) return replaced; if (memo.has(v)) return memo.get(v); if (Array.isArray(v)) { const out: unknown[] = []; memo.set(v, out); for (const x of v) out.push(walk(x)); return out; } const proto = Object.getPrototypeOf(v); if (proto !== Object.prototype && proto !== null) return v; const out: Record<string, unknown> = {}; memo.set(v, out); for (const [k, x] of Object.entries(v)) out[k] = walk(x); return out; }; return walk(value);}