diff --git a/src/js/internal/shared.ts b/src/js/internal/shared.ts index d8b86b61066e..241cefc53796 100644 --- a/src/js/internal/shared.ts +++ b/src/js/internal/shared.ts @@ -435,6 +435,7 @@ export default { once, getLazy, guardCallback, + reportUncaughtException, isInsideNodeModules, resistStopPropagation, diff --git a/src/js/node/worker_threads.ts b/src/js/node/worker_threads.ts index 12f2db6542d8..201809191102 100644 --- a/src/js/node/worker_threads.ts +++ b/src/js/node/worker_threads.ts @@ -12,7 +12,7 @@ type WebWorker = InstanceType & { const EventEmitter = require("node:events"); const { SafeMap } = require("internal/primordials"); -const { throwNotImplemented, warnNotImplementedOnce } = require("internal/shared"); +const { throwNotImplemented, warnNotImplementedOnce, reportUncaughtException } = require("internal/shared"); const { validateString, validateObject, @@ -809,6 +809,21 @@ function moveMessagePortToContext(port, _context) { throwNotImplemented("worker_threads.moveMessagePortToContext"); } +type TerminateCallback = (err: null, exitCode: number) => void; +type PendingTerminate = PromiseWithResolvers & { + // The callbacks of terminate(callback) (DEP0132). + callbacks?: TerminateCallback[]; +}; + +// node also continues its exit handler after a 'message' listener throws: https://github.com/nodejs/node/blob/v26.3.0/lib/internal/event_target.js#L1117-L1119 +function emitGuarded(worker: Worker, name: string, value: unknown) { + try { + worker.emit(name, value); + } catch (err) { + reportUncaughtException(err); + } +} + class Worker extends EventEmitter { #worker: WebWorker; #performance; @@ -823,12 +838,13 @@ class Worker extends EventEmitter { #stdout; #stderr; - // this is used by terminate(); - // either is the exit code if exited, a promise resolving to the exit code, or undefined if we haven't sent .terminate() yet - #onExitPromise: Promise | number | undefined = undefined; + // Made by the first terminate() on a Worker that has not exited. #onClose settles it after 'exit'. + #pendingTerminate: PendingTerminate | undefined = undefined; #urlToRevoke = ""; - // threadId captured for cleaning up the messaging control port on close. - #messagingThreadId: number | undefined = undefined; + // The thread id for the exit handler, where the native one is already -1. + #threadId = -1; + // True from the start of the exit handler: threadId then reads #threadId until the exit state is stored, as node reads a live handle. https://github.com/nodejs/node/blob/v26.3.0/lib/internal/worker.js#L511-L515 + #exiting = false; constructor(filename: string, options: NodeWorkerOptions = {}) { super(); @@ -956,8 +972,8 @@ class Worker extends EventEmitter { } // threadId is only assigned once the WebWorker exists; register the hub-side // control port with the messaging hub now. - this.#messagingThreadId = this.#worker.threadId; - messaging.registerMainThreadPort(this.#messagingThreadId, portToMain); + this.#threadId = this.#worker.threadId; + messaging.registerMainThreadPort(this.#threadId, portToMain); // The transfer is committed - release fds that were transferred but are // not referenced from workerData (nothing will deserialize them). options[kFinalizeJSTransferables]?.(); @@ -1000,7 +1016,8 @@ class Worker extends EventEmitter { } get threadId() { - return this.#worker.threadId; + if (this.#exited) return -1; + return this.#exiting ? this.#threadId : this.#worker.threadId; } get threadName() { @@ -1051,39 +1068,28 @@ class Worker extends EventEmitter { }); } + // node settles the promise from once("exit"), so removeAllListeners() loses it. Here #onClose settles it: https://github.com/nodejs/node/blob/v26.3.0/lib/internal/worker.js#L476-L491 terminate(callback?: unknown) { - if (typeof callback === "function") { + const hasCallback = typeof callback === "function"; + if (hasCallback) { process.emitWarning( "Passing a callback to worker.terminate() is deprecated. It returns a Promise instead.", "DeprecationWarning", "DEP0132", ); - this.#worker.addEventListener("close", event => callback(null, event.code), { once: true }); } - - const onExitPromise = this.#onExitPromise; - // Not a truthy test: after exit #onExitPromise is the exit code, which can be 0; - // falling through would wait on a 'close' event that never fires again. - if (onExitPromise !== undefined) { - // node: terminate() on an already-exited worker resolves with undefined; - // an in-progress terminate (a promise) resolves with the exit code below. - return $isPromise(onExitPromise) ? onExitPromise : Promise.$resolve(undefined); + // node: terminate() on an already-exited worker resolves with undefined. + if (this.#exited) return Promise.$resolve(undefined); + + let pending = this.#pendingTerminate; + if (pending === undefined) { + pending = this.#pendingTerminate = Promise.withResolvers(); + // Keep the event loop alive until the exit, so the promise of an unref()'ed worker resolves too. + this.#worker.ref(); + this.#worker.terminate(); } - - const { resolve, promise } = Promise.withResolvers(); - this.#worker.addEventListener( - "close", - event => { - resolve(event.code); - }, - { once: true }, - ); - // Keep the event loop alive until termination completes so the returned - // promise still resolves even if the worker was unref()'ed. - this.#worker.ref(); - this.#worker.terminate(); - - return (this.#onExitPromise = promise); + if (hasCallback) (pending.callbacks ??= []).push(callback as TerminateCallback); + return pending.promise; } postMessage(...args: [any, any]) { @@ -1157,7 +1163,7 @@ class Worker extends EventEmitter { if (includeObjectsCollectedByMinorGC !== undefined) validateBoolean(includeObjectsCollectedByMinorGC, "options.includeObjectsCollectedByMinorGC"); } - if (this.#exited) { + if (this.#worker.threadId === -1) { return Promise.$reject($ERR_WORKER_NOT_RUNNING("Worker instance not running")); } // Bun has no allocation-sampling heap profiler; yield a valid but empty @@ -1168,17 +1174,48 @@ class Worker extends EventEmitter { } #onClose(e) { - this.#exited = true; + const code: number = e.code; + this.#exiting = true; + try { + this.#exit(code); + } finally { + // A throw from a stdio listener ends #exit, as it ends node's kOnExit. terminate() settles on that path too. + this.#exited = true; + const pending = this.#pendingTerminate; + if (pending !== undefined) { + this.#pendingTerminate = undefined; + const callbacks = pending.callbacks; + if (callbacks !== undefined) { + for (let i = 0; i < callbacks.length; i++) { + try { + callbacks[i](null, code); + } catch (err) { + reportUncaughtException(err); + } + } + } + pending.resolve(code); + } + } + } + + // The order of node's kOnExit: the last messages, the exit state, the stdio EOF, then 'exit'. https://github.com/nodejs/node/blob/v26.3.0/lib/internal/worker.js#L393-L410 + #exit(code: number) { // Revoke the eval blob: URL now that the worker has exited; the // FinalizationRegistry remains only as a GC safety net. if (this.#urlToRevoke) { URL.revokeObjectURL(this.#urlToRevoke); this.#urlToRevoke = ""; } - if (this.#messagingThreadId !== undefined) { - messaging.destroyMainThreadPort(this.#messagingThreadId); - this.#messagingThreadId = undefined; + messaging.destroyMainThreadPort(this.#threadId); + { + let entry; + while ((entry = _receiveMessageOnPort(this.#publicPort)) !== undefined) { + emitGuarded(this, "message", entry.message); + } + this.#publicPort.close(); } + this.#exited = true; // End captured stdio readables when the worker exits, even if it was // terminated before its own streams finished. if (this.#stdout) { @@ -1197,17 +1234,7 @@ class Worker extends EventEmitter { this.#stdin.destroy(); } this.#stdinPort?.close(); - // node delivers everything the worker posted before it exited ahead of - // 'exit' (kOnExit drains the public port), then closes the port. - { - let entry; - while ((entry = _receiveMessageOnPort(this.#publicPort)) !== undefined) { - this.emit("message", entry.message); - } - this.#publicPort.close(); - } - this.#onExitPromise = e.code; - this.emit("exit", e.code); + emitGuarded(this, "exit", code); } #onError(event: ErrorEvent) { diff --git a/test/js/node/worker_threads/worker_threads.test.ts b/test/js/node/worker_threads/worker_threads.test.ts index 809f6376c75c..30948e529cd3 100644 --- a/test/js/node/worker_threads/worker_threads.test.ts +++ b/test/js/node/worker_threads/worker_threads.test.ts @@ -2929,3 +2929,334 @@ describe("no JS entry after a worker's termination has been thrown", () => { }); } }); + +// terminate() called by user code that the parent-side exit handler of a Worker runs. A 'message' +// that the exit handler delivers comes before the exit state is stored: the promise resolves with +// the exit code, after 'exit'. The stdio EOF comes after it: the promise resolves with undefined. +// node v26.3.0 prints the same log for the stdout and stderr fixtures. It delivers a 'message' +// from its exit handler when the parent is held after `new Worker()`, and prints the same log for +// that state. A comment names each difference. +describe("terminate() inside the Worker's exit handler", () => { + // A fixture prints its log when the process exits by itself, so a promise that never settles + // shows as a missing line. + const prelude = ` + const { Worker, MessagePort } = require("node:worker_threads"); + const log = [], warnings = []; + process.on("warning", warning => warnings.push(warning.code)); + process.on("uncaughtException", error => log.push("uncaught " + error.message)); + process.on("exit", () => console.log(JSON.stringify({ log, warnings }))); + let w, id; + const state = () => " threadId=" + (w.threadId === id ? "live" : w.threadId) + " threadName=" + w.threadName; + `; + const construct = (source: string, options: string) => ` + w = new Worker(${JSON.stringify(source)}, { eval: true, name: "name"${options} }); + id = w.threadId; + w.on("exit", code => log.push("exit " + code + state())); + `; + const start = (source: string, options = "") => prelude + construct(source, options); + // Bun only. A 'message' listener starts a port. Without one the public port of the Worker never + // starts, and only the exit handler can deliver what the worker posted. A parent that loses the + // CPU inside `new Worker()` until the worker has ended reaches the same state. + const startWithoutPorts = (source: string, options = "") => `${prelude} + const { addEventListener } = EventTarget.prototype; + EventTarget.prototype.addEventListener = function (type, listener, options) { + if (type === "message" && this instanceof MessagePort) return; + return addEventListener.call(this, type, listener, options); + }; + try { + ${construct(source, options)} + } finally { + EventTarget.prototype.addEventListener = addEventListener; + } + `; + const postsAndEnds = `require("node:worker_threads").parentPort.postMessage(1);`; + // Answers "ready", then stays alive until terminate(). + const idle = `const { parentPort } = require("node:worker_threads"); parentPort.on("message", () => {}); parentPort.postMessage("ready");`; + // Ends without a return to its event loop after the parent wrote to its stdin, so it cannot + // answer the write. + const endsAfterTheWrite = ` + const gate = require("node:worker_threads").workerData; + Atomics.store(gate, 0, 1); + Atomics.notify(gate, 0); + Atomics.wait(gate, 0, 1); + process.exit(0);`; + + const windows = { + // The microtask runs after 'exit': the exit handler delivered the message, not the port. + message: (call: string) => `${startWithoutPorts(postsAndEnds)} + w.on("message", message => { + log.push("message " + message + state()); + queueMicrotask(() => log.push("microtask")); + ${call} + });`, + stdout: (call: string) => `${start("", ", stdout: true")} + w.stdout.on("readable", () => { + if (w.stdout.read() !== null) return; + log.push("stdout EOF" + state()); + ${call} + });`, + stderr: (call: string) => `${start("", ", stderr: true")} + w.stderr.on("readable", () => { + if (w.stderr.read() !== null) return; + log.push("stderr EOF" + state()); + ${call} + });`, + // node does not run this callback: its exit handler does not destroy stdin. + stdin: (call: string) => ` + const gate = new Int32Array(new SharedArrayBuffer(4)); + ${start(endsAfterTheWrite, ", stdin: true, workerData: gate")} + Atomics.wait(gate, 0, 0); + w.stdin.write("x", error => { + log.push("write callback " + error + state()); + ${call} + }); + Atomics.store(gate, 0, 2); + Atomics.notify(gate, 0);`, + }; + const forms = { + promise: `w.terminate().then(value => log.push("promise " + value));`, + callback: `w.terminate((error, code) => log.push("callback " + error + " " + code)).then(value => log.push("promise " + value));`, + asyncDispose: `w[Symbol.asyncDispose]().then(value => log.push("asyncDispose " + value));`, + }; + + async function run(script: string) { + await using proc = Bun.spawn({ + cmd: [bunExe(), "-e", script], + env: bunEnv, + stdout: "pipe", + stderr: "pipe", + }); + const [stdout, stderr, exitCode] = await Promise.all([proc.stdout.text(), proc.stderr.text(), proc.exited]); + return { + printed: stdout.startsWith("{") ? JSON.parse(stdout) : stdout, + // On success stderr has the DEP0132 text, and debug builds add lines to it. + stderr: exitCode === 0 ? "" : stderr, + exitCode, + signalCode: proc.signalCode, + }; + } + const printed = (log: readonly string[], warnings: readonly string[] = []) => ({ + printed: { log, warnings }, + stderr: "", + exitCode: 0, + signalCode: null, + }); + + const live = " threadId=live threadName=name"; + const exited = " threadId=-1 threadName=null"; + test.concurrent.each([ + ["message", "promise", ["message 1" + live, "exit 0" + exited, "microtask", "promise 0"], []], + // node v26.3.0 ignores the callback and prints no warning. node v22 calls it here. + [ + "message", + "callback", + ["message 1" + live, "exit 0" + exited, "callback null 0", "microtask", "promise 0"], + ["DEP0132"], + ], + ["message", "asyncDispose", ["message 1" + live, "exit 0" + exited, "microtask", "asyncDispose undefined"], []], + ["stdout", "promise", ["stdout EOF" + exited, "exit 0" + exited, "promise undefined"], []], + ["stdout", "callback", ["stdout EOF" + exited, "exit 0" + exited, "promise undefined"], ["DEP0132"]], + ["stdout", "asyncDispose", ["stdout EOF" + exited, "exit 0" + exited, "asyncDispose undefined"], []], + ["stderr", "promise", ["stderr EOF" + exited, "exit 0" + exited, "promise undefined"], []], + ["stderr", "callback", ["stderr EOF" + exited, "exit 0" + exited, "promise undefined"], ["DEP0132"]], + ["stderr", "asyncDispose", ["stderr EOF" + exited, "exit 0" + exited, "asyncDispose undefined"], []], + ["stdin", "promise", ["write callback null" + exited, "exit 0" + exited, "promise undefined"], []], + ["stdin", "callback", ["write callback null" + exited, "exit 0" + exited, "promise undefined"], ["DEP0132"]], + ["stdin", "asyncDispose", ["write callback null" + exited, "exit 0" + exited, "asyncDispose undefined"], []], + ] as const)("from %s, %s form", async (window, form, log, warnings) => { + expect(await run(windows[window](forms[form]))).toEqual(printed(log, warnings)); + }); + + test.concurrent("from message: the exit code, one promise for two calls, a call from 'exit'", async () => { + const script = `${startWithoutPorts(postsAndEnds + " process.exit(7);")} + w.on("exit", () => w.terminate().then(value => log.push("from 'exit' " + value))); + w.on("message", () => { + queueMicrotask(() => log.push("microtask")); + const promise = w.terminate(); + // node makes a promise for each call. + log.push("one promise " + (promise === w.terminate())); + promise.then(value => log.push("promise " + value)); + });`; + expect(await run(script)).toEqual( + printed(["one promise true", "exit 7" + exited, "microtask", "from 'exit' undefined", "promise 7"]), + ); + }); + + test.concurrent("from message: the Worker reads as one that runs, but its thread has ended", async () => { + const script = `${startWithoutPorts(postsAndEnds, ", stdin: true")} + w.on("message", message => { + log.push("message " + message + state()); + log.push("stdin destroyed " + w.stdin.destroyed); + try { + w.postMessage(Symbol()); + } catch (error) { + log.push("postMessage " + error.name); + } + w.startHeapProfile().catch(error => log.push("startHeapProfile " + error.code)); + });`; + expect(await run(script)).toEqual( + printed([ + "message 1" + live, + "stdin destroyed false", + "postMessage DataCloneError", + "exit 0" + exited, + "startHeapProfile ERR_WORKER_NOT_RUNNING", + ]), + ); + }); + + // node reports the three errors after 'exit'. + test.concurrent("from message: a listener that throws does not stop the exit handler", async () => { + const posts3 = `const { parentPort } = require("node:worker_threads"); for (let i = 1; i <= 3; i++) parentPort.postMessage(i);`; + const script = `${startWithoutPorts(posts3)} + w.on("message", message => { + log.push("message " + message); + if (message === 1) { + w.terminate().then(value => log.push("promise " + value)); + // Only the public port keeps this process alive, and an exit handler that stops at + // the throw does not close the port. + w.unref(); + } + throw new Error("from message " + message); + });`; + expect(await run(script)).toEqual( + printed([ + "message 1", + "uncaught from message 1", + "message 2", + "uncaught from message 2", + "message 3", + "uncaught from message 3", + "exit 0" + exited, + "promise 0", + ]), + ); + }); + + // node does the same, but it does not settle the promise. + test.concurrent.each(["stdout", "stderr"] as const)( + "from %s EOF: a listener that throws ends the exit handler, and terminate() settles", + async stream => { + const script = `${start(idle, `, ${stream}: true`)} + w.${stream}.on("readable", () => { + if (w.${stream}.read() !== null) return; + log.push("${stream} EOF"); + throw new Error("from ${stream} EOF"); + }); + w.on("message", () => w.terminate().then(value => log.push("promise " + value)));`; + expect(await run(script)).toEqual(printed([`${stream} EOF`, `uncaught from ${stream} EOF`, "promise 1"])); + }, + ); + + test.concurrent("from stdout EOF: an error that nothing handles ends the process with code 1", async () => { + const { printed, stderr, exitCode, signalCode } = await run(` + const { Worker } = require("node:worker_threads"); + const w = new Worker("", { eval: true, stdout: true }); + w.on("exit", code => { + console.log("exit " + code); + process.exit(code); + }); + w.stdout.on("readable", () => { + if (w.stdout.read() === null) throw new Error("from stdout EOF"); + });`); + expect({ printed, exitCode, signalCode }).toEqual({ printed: "", exitCode: 1, signalCode: null }); + expect(stderr).toContain("error: from stdout EOF"); + }); + + test.concurrent("a terminate() before the exit, then one from stdout EOF", async () => { + const script = `${start(idle, ", stdout: true")} + w.stdout.on("readable", () => { + if (w.stdout.read() !== null) return; + log.push("stdout EOF"); + w.terminate().then(value => log.push("from stdout EOF " + value)); + }); + w.on("message", () => w.terminate().then(value => log.push("promise " + value)));`; + expect(await run(script)).toEqual( + printed(["stdout EOF", "exit 1" + exited, "from stdout EOF undefined", "promise 1"]), + ); + }); + + test.concurrent("terminate() adds no listener to the native Worker", async () => { + const script = `${start(idle)} + const { addEventListener } = EventTarget.prototype; + EventTarget.prototype.addEventListener = function (type, listener, options) { + if (type !== "close") return addEventListener.call(this, type, listener, options); + }; + w.on("message", () => w.terminate().then(value => log.push("promise " + value)));`; + expect(await run(script)).toEqual(printed(["exit 1" + exited, "promise 1"])); + }); + + test.concurrent("an exit handler that a throw ended before the exit state: terminate() settles", async () => { + const script = `${start(idle)} + URL.revokeObjectURL = () => { + throw new Error("from revokeObjectURL"); + }; + w.on("message", async () => { + log.push("promise " + (await w.terminate())); + log.push("after the exit " + (await w.terminate())); + });`; + expect(await run(script)).toEqual( + printed(["uncaught from revokeObjectURL", "promise 1", "after the exit undefined"]), + ); + }); + + // The native dispatch does not call the exit handler of a Worker of a disposed Bun.ModuleGraph. + test.concurrent("threadId reads -1 when the thread of a disposed graph's Worker has ended", async () => { + const script = `${prelude} + (async () => { + const graph = new Bun.ModuleGraph({}); + w = graph.run(() => new Worker(${JSON.stringify(idle)}, { eval: true })); + await new Promise(resolve => w.once("message", resolve)); + graph.dispose(); + while (w.threadId !== -1) await new Promise(resolve => setImmediate(resolve)); + log.push("threadId " + w.threadId); + })();`; + expect(await run(script)).toEqual(printed(["threadId -1"])); + }); + + // What terminate() does on a Worker that runs. node v26.3.0 differs in each of these: it + // ignores the callback, it makes a promise and an 'exit' listener for each call, and it loses + // the promise when an 'exit' listener throws or when the listeners are removed. + describe("on a Worker that runs", () => { + test.concurrent("terminate(callback): DEP0132, the callback after 'exit', then the promise", async () => { + const script = `${start(idle)} + w.on("message", () => { + const listeners = w.listenerCount("exit"); + const promise = w.terminate((error, code) => log.push("callback " + error + " " + code)); + log.push("one promise " + (promise === w.terminate())); + log.push("new 'exit' listeners " + (w.listenerCount("exit") - listeners)); + promise.then(value => log.push("promise " + value)); + });`; + expect(await run(script)).toEqual( + printed( + ["one promise true", "new 'exit' listeners 0", "exit 1" + exited, "callback null 1", "promise 1"], + ["DEP0132"], + ), + ); + }); + + test.concurrent("an 'exit' listener and a callback that throw do not lose the rest", async () => { + const script = `${start(idle)} + w.on("exit", () => { throw new Error("from 'exit'"); }); + w.on("message", () => { + w.terminate(() => { throw new Error("from callback"); }); + w.terminate((error, code) => log.push("callback " + error + " " + code)).then(value => log.push("promise " + value)); + });`; + expect(await run(script)).toEqual( + printed( + ["exit 1" + exited, "uncaught from 'exit'", "uncaught from callback", "callback null 1", "promise 1"], + ["DEP0132", "DEP0132"], + ), + ); + }); + + test.concurrent("removeAllListeners() does not lose the promise", async () => { + const script = `${start(idle)} + w.on("message", () => { + w.terminate().then(value => log.push("promise " + value)); + w.removeAllListeners(); + });`; + expect(await run(script)).toEqual(printed(["promise 1"])); + }); + }); +});