diff --git a/src/js/node/worker_threads.ts b/src/js/node/worker_threads.ts index 7262da0f30f7..b73155835474 100644 --- a/src/js/node/worker_threads.ts +++ b/src/js/node/worker_threads.ts @@ -5,8 +5,10 @@ type WebWorker = InstanceType; const EventEmitter = require("node:events"); const { SafeMap } = require("internal/primordials"); -const Readable = require("internal/streams/readable"); -const Writable = require("internal/streams/writable"); +// Readable/Writable are loaded lazily: this module is preloaded in every +// node:worker_threads worker, and the streams graph is several ms cold. +let Readable; +let Writable; const { throwNotImplemented, warnNotImplementedOnce } = require("internal/shared"); const { validateString, @@ -26,8 +28,6 @@ function normalizeWorkerName(rawName) { return ""; } -const { isAbsolute: pathIsAbsolute } = require("node:path"); - // node's filename validation for non-eval workers: absolute or "./"/"../"-relative // paths and file: URL objects; bare specifiers and string URLs are rejected. function validateWorkerFilename(filename) { @@ -41,7 +41,7 @@ function validateWorkerFilename(filename) { // throws the canonical ERR_INVALID_ARG_TYPE with the exact node message. return filename; } - if (pathIsAbsolute(filename) || /^\.\.?[\\/]/.test(filename)) { + if (require("node:path").isAbsolute(filename) || /^\.\.?[\\/]/.test(filename)) { return filename; } let message = @@ -334,6 +334,7 @@ const BUN_WORKER_MESSAGING_KEY = "@@bunWorkerThreadsMessaging"; // Readable fed by a control MessagePort (worker.stdout/stderr on the parent, // process.stdin in the worker). The peer posts arrays of Buffers; null signals EOF. function makePortReadable(port) { + Readable ??= require("internal/streams/readable"); let attached = false; let ended = false; function onMessage(payload) { @@ -380,6 +381,7 @@ function makePortReadable(port) { // Writable that forwards chunks over a control MessagePort (worker.stdin on the // parent, process.stdout/stderr in the worker). final() posts null as EOF. function makePortWritable(port) { + Writable ??= require("internal/streams/writable"); // Reader-side acks complete the in-flight writev. The listener refs the // event loop; release that immediately — the port is re-ref'd only while a // batch is awaiting its ack, so unflushed data keeps the writer alive @@ -435,44 +437,57 @@ function makePortWritable(port) { }); } +// Install a self-replacing accessor: the first get() builds the value and +// rewrites the slot to a plain writable data property, so later reads are +// direct and identity-stable; set() does the same rewrite so assignment works. +function defineLazy(obj, name, enumerable, make) { + Object.defineProperty(obj, name, { + configurable: true, + enumerable, + get() { + const value = make(); + Object.defineProperty(obj, name, { value, writable: true, configurable: true, enumerable }); + return value; + }, + set(value) { + Object.defineProperty(obj, name, { value, writable: true, configurable: true, enumerable }); + }, + }); +} + function setupWorkerStdio(stdio) { + const proc: any = process; const { stdin, stdout, stderr } = stdio; - if (stdout) { - Object.defineProperty(process, "stdout", { - value: makePortWritable(stdout), - writable: true, - configurable: true, - enumerable: true, - }); - } - if (stderr) { - Object.defineProperty(process, "stderr", { - value: makePortWritable(stderr), - writable: true, - configurable: true, - enumerable: true, - }); - } + // Shadow the native PropertyCallback slots via plain [[Set]] first: defineProperty + // on one reifies the process static table and invokes constructStdout/Stderr/Stdin, + // cold-loading node:stream just to build fd streams we then discard. + proc.stdin = undefined; + if (stdout) proc.stdout = undefined; + if (stderr) proc.stderr = undefined; + // Lazy: the streams graph + Console constructor are several ms cold, and most + // workers never touch process.stdout/stderr/stdin directly. + if (stdout) defineLazy(proc, "stdout", true, () => makePortWritable(stdout)); + if (stderr) defineLazy(proc, "stderr", true, () => makePortWritable(stderr)); // node always replaces a worker's process.stdin: port-backed when { stdin: true }, // otherwise an immediately-EOF'd stream — never the process-wide fd 0, which // would race the main thread (and hang on a TTY). - Object.defineProperty(process, "stdin", { - value: stdin + defineLazy(proc, "stdin", true, () => + stdin ? makePortReadable(stdin) - : new Readable({ + : new (Readable ??= require("internal/streams/readable"))({ read() { this.push(null); }, }), - writable: true, - configurable: true, - enumerable: true, - }); + ); // node routes console.log through process.stdout/stderr; Bun's global console // writes the fd directly, so rebind it to the captured streams when present. + // Capture via require("node:console") first so the module registry caches the + // native object (which carries .Console/.write); otherwise a later user require + // would evaluate `export default console` through this getter and cache a bare instance. if (stdout || stderr) { - const { Console } = require("node:console"); - globalThis.console = new Console(process.stdout, process.stderr); + const nativeConsole = require("node:console") as typeof globalThis.console & { Console: any }; + defineLazy(globalThis, "console", false, () => new nativeConsole.Console(process.stdout, process.stderr)); } } @@ -1199,7 +1214,7 @@ class Worker extends EventEmitter { getHeapSnapshot(options: unknown) { const stringPromise = this.#worker.getHeapSnapshot(options); - return stringPromise.then(s => new HeapSnapshotStream(s)); + return stringPromise.then(s => makeHeapSnapshotStream(s)); } getHeapStatistics() { @@ -1351,21 +1366,24 @@ class Worker extends EventEmitter { } } -class HeapSnapshotStream extends Readable { - #json: string | undefined; - - constructor(json: string) { - super(); - this.#json = json; - } - - _read() { - if (this.#json !== undefined) { - this.push(this.#json); - this.push(null); - this.#json = undefined; +let _HeapSnapshotStream; +function makeHeapSnapshotStream(json: string) { + Readable ??= require("internal/streams/readable"); + _HeapSnapshotStream ??= class HeapSnapshotStream extends Readable { + #json: string | undefined; + constructor(json: string) { + super(); + this.#json = json; } - } + _read() { + if (this.#json !== undefined) { + this.push(this.#json); + this.push(null); + this.#json = undefined; + } + } + }; + return new _HeapSnapshotStream(json); } export default { diff --git a/test/js/node/worker_threads/worker-stdio-lazy.test.ts b/test/js/node/worker_threads/worker-stdio-lazy.test.ts new file mode 100644 index 000000000000..6867911f1953 --- /dev/null +++ b/test/js/node/worker_threads/worker-stdio-lazy.test.ts @@ -0,0 +1,63 @@ +import { expect, test } from "bun:test"; +import { once } from "node:events"; +import { Worker } from "node:worker_threads"; + +// Regression: the stdio rebind used to run eagerly in every worker's preload, +// cold-loading node:stream + Console and reifying the process static table on +// every spawn even when the worker never touched stdio. +test("worker process.stdio and console are installed as lazy accessors", async () => { + const worker = new Worker( + ` + const { parentPort } = require("worker_threads"); + const d = (obj, k) => { + const desc = Object.getOwnPropertyDescriptor(obj, k); + return desc ? (desc.get ? "accessor" : "value") : "none"; + }; + const before = { + stdout: d(process, "stdout"), + stderr: d(process, "stderr"), + stdin: d(process, "stdin"), + console: d(globalThis, "console"), + }; + // first read materializes the stream; second read must be the same object + const s1 = process.stdout; + const s2 = process.stdout; + parentPort.postMessage({ + before, + afterStdout: d(process, "stdout"), + sameInstance: s1 === s2, + isWritable: typeof s1.write === "function", + }); + `, + { eval: true }, + ); + const [msg] = await once(worker, "message"); + expect(msg).toEqual({ + before: { stdout: "accessor", stderr: "accessor", stdin: "accessor", console: "accessor" }, + afterStdout: "value", + sameInstance: true, + isWritable: true, + }); + await worker.terminate(); +}); + +// node:console is `export default globalThis.console`; evaluating it after the +// lazy getter is installed would cache a plain Console instance that lacks +// .Console/.write, so the preload primes the module registry first. +test("require('node:console') in a worker still exposes Console and write", async () => { + const worker = new Worker( + ` + const { parentPort } = require("worker_threads"); + const c = require("node:console"); + parentPort.postMessage({ + hasConsoleCtor: typeof c.Console === "function", + hasWrite: typeof c.write === "function", + hasAsyncIterator: typeof c[Symbol.asyncIterator] === "function", + }); + `, + { eval: true }, + ); + const [msg] = await once(worker, "message"); + expect(msg).toEqual({ hasConsoleCtor: true, hasWrite: true, hasAsyncIterator: true }); + await worker.terminate(); +});