Something went wrong. Try again.
Monorepo for the nibling.run platform
Something went wrong. Try again.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404// node:worker_threads. A Worker is a kernel thread in this process (DESIGN §Processes,// threads, signals): the kernel starts a runtime worker for it and says when it ends;// the Worker object holds this thread's loop until then. Everything else travels the// port pair the Worker creates (ports.ts): messages both ways, and from the thread its// 'online', its stdout and stderr (written to this thread's process.stdout/stderr, as// Node pipes them) and its uncaught error.import type { HostToken } from "../shared/vurls";import type { KernelToRuntimeInit, RuntimeToKernel, SyscallSpec, ThreadExit } from "../shared/protocol";import { capture, runInFrame, type Context } from "./async_hooks";import { Buffer } from "./buffer";import { uncloneable } from "./clone";import { invalidArgType } from "./errors";import { EventEmitter, type EventEmitterConstructor } from "./events";import type { Loop } from "./loop";import { MessageChannel, MessagePort, exportPorts, importPorts, portPair, type PortRecord } from "./ports";import type { GuestProcess } from "./process";import { unsupported } from "./refuse";import { Readable } from "./stream";
export interface ThreadsDeps { loop: Loop; init: KernelToRuntimeInit; host: HostToken; callSync<K extends keyof SyscallSpec>(op: K, args: SyscallSpec[K]["args"]): SyscallSpec[K]["ret"]; post(msg: RuntimeToKernel, transfer: Transferable[]): void; process(): GuestProcess; resolvePath(p: string): string;}
export interface WorkerThreads { module: Record<string, unknown>; /** This thread's end of its parent's channel; null on the main thread. */ parentPort: MessagePort | null; threadExited(exit: ThreadExit): void;}
interface WorkerOptions { eval?: boolean; workerData?: unknown; argv?: unknown[]; env?: Record<string, unknown> | symbol; transferList?: unknown[]; name?: string; stdout?: boolean; stderr?: boolean; resourceLimits?: Record<string, number>;}
const SHARE_ENV = Symbol.for("nodejs.worker_threads.SHARE_ENV");const EVAL_NAME = "[worker eval]";
function workerPathError(filename: string): TypeError { return Object.assign(new TypeError( "The worker script or module filename must be an absolute path or a relative path starting with './' or '../'." + ` Received "${filename}"`, ), { code: "ERR_WORKER_PATH" });}
export function makeWorkerThreads(deps: ThreadsDeps): WorkerThreads { const { loop, init } = deps; const self = init.thread; const workers = new Map<number, Worker>(); const environmentData = new Map<unknown, unknown>( self ? (importPorts(self.environmentData) as Array<[unknown, unknown]>) : [], ); const untransferable = new WeakSet<object>(); const parentPort = self ? new MessagePort(self.port) : null;
const PROFILER_WHY = "no per-worker V8 heap or profiler from a page";
class Worker extends (EventEmitter as unknown as EventEmitterConstructor) { readonly threadId: number; readonly stdin = null; readonly stdout = new Readable({ read() {} }); readonly stderr = new Readable({ read() {} }); #resourceLimits: Record<string, number>; #port: MessagePort; #pipe: { stdout: boolean; stderr: boolean }; #exit: ThreadExit | null = null; #done = false; #ref = true; #holding = false; #exited: Array<(code: number) => void> = []; #threadName: string; /** new Worker() creates the handle (probed): 'online', 'message' and 'exit' all * run in the frame current here, not in whatever frame happens to be current * when the thread's port wakes this one natively. */ readonly #frame: Context = capture();
constructor(filename: string | URL, options: WorkerOptions = {}) { super(); const isEval = options.eval === true; let entry: string; if (isEval) { if (typeof filename !== "string") throw invalidArgType("filename", "of type string", filename); entry = EVAL_NAME; } else if (filename instanceof URL) { if (filename.protocol !== "file:") throw Object.assign(new TypeError(`The URL must be of scheme file`), { code: "ERR_INVALID_URL_SCHEME" }); entry = decodeURIComponent(filename.pathname); } else if (typeof filename === "string") { if (filename.startsWith("file:")) entry = decodeURIComponent(new URL(filename).pathname); else if (filename.startsWith("/")) entry = filename; else if (filename.startsWith("./") || filename.startsWith("../")) entry = deps.resolvePath(filename); else throw workerPathError(filename); } else { throw invalidArgType("filename", "of type string or an instance of URL", filename); } if (options.env === SHARE_ENV) { throw unsupported("worker_threads", "SHARE_ENV", "a thread's env is a snapshot taken when it starts; the kernel keeps no live environment to share"); } const proc = deps.process(); const source = options.env === undefined || options.env === null ? proc.env : options.env as Record<string, unknown>; const env = Object.fromEntries(Object.entries(source).map(([k, v]) => [k, String(v)])); const argv = (options.argv ?? []).map(String);
this.threadId = deps.callSync("newThread", {}).tid; this.#threadName = typeof options.name === "string" && options.name !== "" ? options.name : "WorkerThread"; this.#resourceLimits = { ...(options.resourceLimits ?? {}) }; this.#pipe = { stdout: options.stdout === true, stderr: options.stderr === true };
const [mine, theirs] = portPair(); this.#port = mine; mine.onRecord = (rec) => runInFrame(this.#frame, () => this.#record(rec)); mine.onWake = () => { if (this.#exit) loop.enqueueCompletion(() => runInFrame(this.#frame, () => this.#tryFinish())); }; // The Worker, not its port, decides whether the loop stays alive. mine.unref(); mine.start();
const { value, natives } = exportPorts( { workerData: options.workerData, environmentData: [...environmentData] }, options.transferList ?? [], ); const shared = value as { workerData: unknown; environmentData: Array<[unknown, unknown]> }; const port = theirs.detach(); const transfer = [port.native, ...natives] as Transferable[]; deps.post({ t: "startThread", entry, argv, env, cwd: proc.cwd(), thread: { tid: this.threadId, name: this.#threadName, eval: isEval ? filename as string : null, workerData: shared.workerData, environmentData: shared.environmentData, port, }, transfer, }, transfer); workers.set(this.threadId, this); this.#updateHold(); }
/** The limits this Worker was given, as Node's getter reads back the constructor * option; Node's own defaults and post-exit `{}` are V8-internal and not modeled. */ get resourceLimits(): Record<string, number> { return { ...this.#resourceLimits }; }
/** The name option this Worker was started with, as the thread sees itself * through worker_threads.threadName; "WorkerThread" is Node's default. */ get threadName(): string { return this.#threadName; }
cpuUsage(): never { throw unsupported("worker_threads", "Worker.cpuUsage()", PROFILER_WHY); }
startCpuProfile(): never { throw unsupported("worker_threads", "Worker.startCpuProfile()", PROFILER_WHY); }
startHeapProfile(): never { throw unsupported("worker_threads", "Worker.startHeapProfile()", PROFILER_WHY); }
postMessage(value: unknown, transferList?: readonly unknown[]): void { if (!this.#done) this.#port.postMessage(value, transferList); }
terminate(): Promise<number | undefined> { if (this.#done) return Promise.resolve(undefined); deps.post({ t: "terminateThread", tid: this.threadId }, []); return new Promise((resolve) => this.#exited.push(resolve)); }
ref(): void { this.#ref = true; this.#updateHold(); }
unref(): void { this.#ref = false; this.#updateHold(); }
getHeapSnapshot(): never { throw unsupported("worker_threads", "Worker.getHeapSnapshot()", "the browser exposes no heap snapshots"); }
getHeapStatistics(): never { throw unsupported("worker_threads", "Worker.getHeapStatistics()", PROFILER_WHY); }
/** @internal The kernel says the thread ended. */ ended(exit: ThreadExit): void { this.#exit = exit; loop.enqueueCompletion(() => runInFrame(this.#frame, () => this.#tryFinish())); }
#record(rec: PortRecord): void { switch (rec.k) { case "online": this.emit("online"); return; case "m": this.emit("message", rec.v); return; case "error": this.emit("error", rec.v); return; case "stdio": { const name = rec.fd === 1 ? "stdout" : "stderr"; if (this.#pipe[name]) this[name].push(Buffer.from(rec.bytes)); else deps.process()[name].write(rec.bytes); return; } case "close": return; } }
/** 'exit' comes after everything the thread sent; a record sent natively may * still be in flight, and its arrival retries. */ #tryFinish(): void { if (!this.#exit || this.#done) return; this.#port.drainNow(); if (this.#port.stalled() && !this.#exit.terminated) return; this.#done = true; workers.delete(this.threadId); this.#updateHold(); this.#port.close(); if (this.#pipe.stdout) this.stdout.push(null); if (this.#pipe.stderr) this.stderr.push(null); const { code } = this.#exit; this.emit("exit", code); for (const resolve of this.#exited.splice(0)) resolve(code); }
#updateHold(): void { const hold = this.#ref && !this.#done; if (hold === this.#holding) return; this.#holding = hold; if (hold) loop.hold(); else loop.release(); } }
/** Node's BroadcastChannel: an EventTarget over the browser's, scoped to this process * so two guests on one origin never hear each other. Open channels hold the loop. */ class BroadcastChannel extends EventTarget { // Node's name/onmessage/onmessageerror are prototype accessors (WebIDL legacy // event-handler attributes), not own instance fields — getters below, backed by // private state, so they show up the same way on the class. #name: string; #native: globalThis.BroadcastChannel | null; #ref = true; #holding = false; #onmessage: ((ev: MessageEvent) => void) | null = null; #onmessageerror: ((ev: MessageEvent) => void) | null = null; readonly #frame: Context;
constructor(name: unknown) { super(); if (name === undefined) throw Object.assign(new TypeError('The "name" argument must be specified'), { code: "ERR_MISSING_ARGS" }); this.#name = String(name); this.#frame = capture(); this.#native = new globalThis.BroadcastChannel(`nib:${deps.host}:${init.pid}:${this.#name}`); this.#native.onmessage = (ev) => loop.enqueueCompletion(() => runInFrame(this.#frame, () => { const out = new MessageEvent("message", { data: ev.data }); this.#onmessage?.(out); this.dispatchEvent(out); })); this.#updateHold(); }
get name(): string { return this.#name; }
get onmessage(): ((ev: MessageEvent) => void) | null { return this.#onmessage; }
set onmessage(fn: ((ev: MessageEvent) => void) | null) { this.#onmessage = typeof fn === "function" ? fn : null; }
get onmessageerror(): ((ev: MessageEvent) => void) | null { return this.#onmessageerror; }
set onmessageerror(fn: ((ev: MessageEvent) => void) | null) { this.#onmessageerror = typeof fn === "function" ? fn : null; }
postMessage(message: unknown): void { if (!this.#native) throw new DOMException("BroadcastChannel is closed.", "InvalidStateError"); this.#native.postMessage(message); }
close(): void { this.#native?.close(); this.#native = null; this.#updateHold(); }
ref(): this { this.#ref = true; this.#updateHold(); return this; }
unref(): this { this.#ref = false; this.#updateHold(); return this; }
#updateHold(): void { const hold = this.#ref && this.#native !== null; if (hold === this.#holding) return; this.#holding = hold; if (hold) loop.hold(); else loop.release(); } }
const module = { isMainThread: self === null, isInternalThread: false, threadId: self?.tid ?? 0, // Node's default display name for an unnamed thread; the main thread's is "". threadName: self ? (self.name || "WorkerThread") : "", workerData: self ? importPorts(self.workerData) : null, parentPort, resourceLimits: {}, SHARE_ENV, Worker, MessageChannel, MessagePort, BroadcastChannel, // The Web Locks API re-export DESIGN calls for; cross-thread locks are the // browser's own, not a kernel reimplementation. locks: (globalThis as unknown as { navigator: { locks: unknown } }).navigator.locks, markAsUntransferable(value: unknown): void { if (value !== null && (typeof value === "object" || typeof value === "function")) untransferable.add(value); }, isMarkedAsUntransferable(value: unknown): boolean { return value !== null && typeof value === "object" && untransferable.has(value); }, /** Node checks this only for a plain object (clone.ts): Array, Map, Date and the * rest already have their own clone path before the check and are unaffected. */ markAsUncloneable(value: unknown): void { if (value !== null && (typeof value === "object" || typeof value === "function")) uncloneable.add(value as object); }, /** Node routes this through the main thread's registry of every live thread in * the process; this kernel only wires a thread to its direct parent/child ports * (ports.ts), so only the same-thread validation can be honest here — reaching * an arbitrary thread id needs the kernel to mediate (a missing primitive, not * implementable from this module alone). */ async postMessageToThread(threadId: unknown, _value: unknown, _transferListOrTimeout?: unknown, _timeout?: unknown): Promise<void> { if (Number(threadId) === module.threadId) { throw Object.assign(new Error("Cannot sent a message to the same thread"), { code: "ERR_WORKER_MESSAGING_SAME_THREAD" }); } throw unsupported("worker_threads", "postMessageToThread()", "no process-wide thread registry: only direct parent/child ports are wired, so an arbitrary thread id can't be routed to"); }, /** The port's next message, taken now from its shared queue, or undefined. A * message sent with transferables rides the native port, so it is here only once * this thread's loop has seen it arrive. */ receiveMessageOnPort(port: unknown): { message: unknown } | undefined { if (!(port instanceof MessagePort)) { throw Object.assign(new TypeError('The "port" argument must be a MessagePort instance'), { code: "ERR_INVALID_ARG_TYPE" }); } return port.receiveSync(); }, getEnvironmentData(key: unknown): unknown { return environmentData.get(key); }, setEnvironmentData(key: unknown, value: unknown): void { if (value === undefined) environmentData.delete(key); else environmentData.set(key, value); }, moveMessagePortToContext(): never { throw unsupported("worker_threads", "moveMessagePortToContext()", "there is no node:vm context to move a port into"); }, };
return { module, parentPort, threadExited(exit) { workers.get(exit.tid)?.ended(exit); }, };}