diff --git a/src/js/builtins/ProcessObjectInternals.ts b/src/js/builtins/ProcessObjectInternals.ts index 0e3793be2e14..37501f664d0b 100644 --- a/src/js/builtins/ProcessObjectInternals.ts +++ b/src/js/builtins/ProcessObjectInternals.ts @@ -115,6 +115,20 @@ export function getStdioWriteStream( return [stream, underlyingSink]; } +// node:worker_threads worker: process.stdout/stderr write to the parent Worker over a +// MessagePort; process.stdin reads from one for { stdin: true }, else is already ended. +export function getNodeWorkerStdioStream(process: typeof globalThis.process, fd: number, ports: any) { + const stdio = require("internal/worker/stdio"); + if (fd === 0) { + return ports.stdin ? stdio.makePortReadable(ports.stdin, true) : stdio.makeEndedReadable(); + } + const stream = stdio.makePortWritable(fd === 1 ? ports.stdout : ports.stderr); + // A synchronous exit leaves no loop turn for the reader's ack; complete the parked + // writev so buffered chunks are posted before the thread goes away (node's flushSync). + process.on("exit", stream[stdio.kFlushSync]); + return stream; +} + export function getStdinStream( process: typeof globalThis.process, fd: number, diff --git a/src/js/internal/worker/stdio.ts b/src/js/internal/worker/stdio.ts new file mode 100644 index 000000000000..a07790ebc8c6 --- /dev/null +++ b/src/js/internal/worker/stdio.ts @@ -0,0 +1,149 @@ +// Port-backed stdio streams shared by both ends of a node:worker_threads Worker: +// worker.stdout/stderr/stdin on the parent, process.stdout/stderr/stdin in the +// worker (built lazily from BunProcess.cpp via ProcessObjectInternals). +// +// Each stream rides its own MessageChannel with node's flow control +// (lib/internal/worker/io.js): the writer posts an array of chunks and withholds +// the writev callback until the reader acks from _read(). One batch is in +// flight at a time; further writes buffer in the Writable, so write() returns +// false and 'drain' fires only when the consumer catches up. The payload is the +// bare chunk array, EOF is null, and any other message is the ack. +const Readable = require("internal/streams/readable"); +const Writable = require("internal/streams/writable"); + +const kFlushSync = Symbol("kFlushSync"); + +// Readable fed by a MessagePort (worker.stdout/stderr on the parent, process.stdin +// in the worker). The peer posts arrays of Buffers; null signals EOF. +function makePortReadable(port: MessagePort, incrementsPortRef: boolean) { + let ended = false; + let startedReading = false; + function onMessage(event: MessageEvent) { + const payload = event.data; + if (payload === null) { + if (ended === false) { + ended = true; + stream.push(null); + } + port.removeEventListener("message", onMessage); + } else if (ended === false) { + for (let i = 0; i < payload.length; i++) { + stream.push(Buffer.from(payload[i])); + } + } + } + const stream = new Readable({ + read() { + if (startedReading === false && incrementsPortRef) { + startedReading = true; + port.ref(); + } + // Tell the writer we want more data; it completes its in-flight writev + // on receipt (node's STDIO_WANTS_MORE_DATA). + if (ended === false) port.postMessage(true); + }, + }); + // Attach eagerly so the peer's writev is ack'd (via push -> maybeReadMore -> + // _read) even when no one consumes this stream; unref immediately so an + // unconsumed captured stream never pins the loop on its own (node's model). + port.addEventListener("message", onMessage); + port.unref(); + // 'close' covers natural EOF and destroy(); release the read-time ref and + // drop the listener so a destroyed captured stream can't pin an unref'd worker. + stream.on("close", () => { + ended = true; + port.removeEventListener("message", onMessage); + if (startedReading && incrementsPortRef) { + startedReading = false; + port.unref(); + } + }); + // Lets the parent end worker.stdout/stderr when the worker exits abruptly. + stream.endFromOwner = function () { + if (ended === false) { + ended = true; + stream.push(null); + port.removeEventListener("message", onMessage); + } + }; + return stream; +} + +// Writable that forwards chunks over a MessagePort (worker.stdin on the parent, +// process.stdout/stderr in the worker). final() posts null as EOF. +function makePortWritable(port: MessagePort) { + // 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 + // (node's kWaitingStreams) but an idle stream never pins the loop. + let pendingWriteCallback: ((error?: Error | null) => void) | null = null; + function onAck() { + const cb = pendingWriteCallback; + if (cb !== null) { + pendingWriteCallback = null; + port.unref(); + cb(); + } + } + port.addEventListener("message", onAck); + port.unref(); + const stream = new Writable({ + decodeStrings: false, + writev(chunks, cb) { + const payload = new Array(chunks.length); + for (let i = 0; i < chunks.length; i++) { + const { chunk, encoding } = chunks[i]; + payload[i] = typeof chunk === "string" ? Buffer.from(chunk, encoding) : chunk; + } + port.postMessage(payload); + if (process._exiting) { + // No event loop turns remain to deliver an ack; complete synchronously + // so exit-time writes are not lost (node does the same). + cb(); + } else { + // Only one writev is in flight at a time, so the slot can't be occupied. + pendingWriteCallback = cb; + port.ref(); + } + }, + final(cb) { + port.postMessage(null); + cb(); + }, + destroy(err, cb) { + // Discharge an in-flight batch: the reader may never ack a destroyed + // stream, so release the loop ref taken in writev and complete the + // parked callback; drop the ack listener so a late ack can't fire + // into the destroyed stream. + const pending = pendingWriteCallback; + if (pending !== null) { + pendingWriteCallback = null; + port.unref(); + pending(err); + } + port.removeEventListener("message", onAck); + cb(err); + }, + }); + // On a synchronous exit no ack can arrive; completing the parked writev lets + // the Writable clear its buffer through writev, which now completes + // synchronously because process._exiting is set (node's flushSync). + stream[kFlushSync] = onAck; + return stream; +} + +// A node worker's process.stdin without { stdin: true }. +function makeEndedReadable() { + return new Readable({ + read() { + this.push(null); + }, + }); +} + +export default { + kFlushSync, + makePortReadable, + makePortWritable, + makeEndedReadable, +}; diff --git a/src/js/node/worker_threads.ts b/src/js/node/worker_threads.ts index f9d221865a82..bd98b0529f15 100644 --- a/src/js/node/worker_threads.ts +++ b/src/js/node/worker_threads.ts @@ -4,7 +4,6 @@ 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"); const { throwNotImplemented, warnNotImplementedOnce } = require("internal/shared"); const { validateString, @@ -80,6 +79,7 @@ const { 9: _setEntryEvaluatedHook, 10: _isNodeWorker, 11: _setParentPort, + 12: _setStdioPorts, } = $cpp("Worker.cpp", "createNodeWorkerThreadsBinding") as [ unknown, number, @@ -93,6 +93,7 @@ const { (hook: () => void) => void, boolean, (port: MessagePort) => void, + (ports: object) => void, ]; type NodeWorkerOptions = import("node:worker_threads").WorkerOptions; @@ -326,175 +327,24 @@ const BUN_WORKER_MESSAGING_KEY = "@@bunWorkerThreadsMessaging"; // stdio and control ports. const BUN_WORKER_PARENT_PORT_KEY = "@@bunWorkerThreadsParentPort"; -// Captured stdio rides a dedicated MessageChannel per stream with node's flow -// control (lib/internal/worker/io.js): the writer posts an array of chunks -// (STDIO_PAYLOAD) and withholds the writev callback until the reader posts an -// ack (STDIO_WANTS_MORE_DATA) from _read(). One batch is in flight at a time; -// further writes buffer in the Writable, so write() returns false and 'drain' -// fires only when the consumer catches up — end-to-end backpressure. Since -// each stream has its own port (node multiplexes one env port), the payload is -// the bare chunk array, EOF is null, and any other message is the ack. - -const kFlushSync = Symbol("kFlushSync"); - -// 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, incrementsPortRef) { - let ended = false; - let startedReading = false; - function onMessage(payload) { - if (payload === null) { - if (ended === false) { - ended = true; - stream.push(null); - } - port.off("message", onMessage); - } else if (ended === false) { - for (let i = 0; i < payload.length; i++) { - stream.push(Buffer.from(payload[i])); - } - } - } - const stream = new Readable({ - read() { - if (startedReading === false && incrementsPortRef) { - startedReading = true; - port.ref(); - } - // Tell the writer we want more data; it completes its in-flight writev - // on receipt (node's STDIO_WANTS_MORE_DATA). - if (ended === false) port.postMessage(true); - }, - }); - // Attach eagerly so the peer's writev is ack'd (via push -> maybeReadMore -> - // _read) even when no one consumes this stream; unref immediately so an - // unconsumed captured stream never pins the loop on its own (node's model). - port.on("message", onMessage); - port.unref(); - // 'close' covers natural EOF and destroy(); release the read-time ref and - // drop the listener so a destroyed captured stream can't pin an unref'd worker. - stream.on("close", () => { - ended = true; - port.off("message", onMessage); - if (startedReading && incrementsPortRef) { - startedReading = false; - port.unref(); - } - }); - // Lets the parent end worker.stdout/stderr when the worker exits abruptly. - stream.endFromOwner = function () { - if (ended === false) { - ended = true; - stream.push(null); - port.off("message", onMessage); - } - }; - return stream; -} - -// 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) { - // 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 - // (node's kWaitingStreams) but an idle stream never pins the loop. - let pendingWriteCallback: ((error?: Error | null) => void) | null = null; - function onAck() { - const cb = pendingWriteCallback; - if (cb !== null) { - pendingWriteCallback = null; - port.unref(); - cb(); - } - } - port.on("message", onAck); - port.unref(); - const stream = new Writable({ - decodeStrings: false, - writev(chunks, cb) { - const payload = new Array(chunks.length); - for (let i = 0; i < chunks.length; i++) { - const { chunk, encoding } = chunks[i]; - payload[i] = typeof chunk === "string" ? Buffer.from(chunk, encoding) : chunk; - } - port.postMessage(payload); - if (process._exiting) { - // No event loop turns remain to deliver an ack; complete synchronously - // so exit-time writes are not lost (node does the same). - cb(); - } else { - // Only one writev is in flight at a time, so the slot can't be occupied. - pendingWriteCallback = cb; - port.ref(); - } - }, - final(cb) { - port.postMessage(null); - cb(); - }, - destroy(err, cb) { - // Discharge an in-flight batch: the reader may never ack a destroyed - // stream, so release the loop ref taken in writev and complete the - // parked callback; drop the ack listener so a late ack can't fire - // into the destroyed stream. - const pending = pendingWriteCallback; - if (pending !== null) { - pendingWriteCallback = null; - port.unref(); - pending(err); - } - port.off("message", onAck); - cb(err); - }, - }); - // On a synchronous exit no ack can arrive; completing the parked writev lets - // the Writable clear its buffer through writev, which now completes - // synchronously because process._exiting is set (node's flushSync). - stream[kFlushSync] = onAck; - return stream; -} +const { makePortReadable, makePortWritable } = require("internal/worker/stdio"); // The parent always sends stdout and stderr ports; stdin only for { stdin: true }. +// With the ports registered on the global, process.stdin/stdout/stderr's native +// lazy getters build the port-backed streams (never the fd-backed ones a plain +// thread would get); read them here so they exist before user code, as in node. function setupWorkerStdio(stdio) { - const { stdin, stdout, stderr } = stdio; - const stdoutStream = makePortWritable(stdout); - const stderrStream = makePortWritable(stderr); - Object.defineProperty(process, "stdout", { - value: stdoutStream, - writable: true, - configurable: true, - enumerable: true, - }); - Object.defineProperty(process, "stderr", { - value: stderrStream, - writable: true, - configurable: true, - enumerable: true, - }); + _setStdioPorts(stdio); + const stdoutStream = process.stdout; + const stderrStream = process.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 - ? makePortReadable(stdin, true) - : new Readable({ - read() { - this.push(null); - }, - }), - writable: true, - configurable: true, - enumerable: true, - }); + void process.stdin; // node routes console.log through process.stdout/stderr; Bun's global console // writes the fd directly, so rebind it to the port-backed streams. const { Console } = require("node:console"); globalThis.console = new Console(stdoutStream, stderrStream); - process.on("exit", () => { - stdoutStream[kFlushSync](); - stderrStream[kFlushSync](); - }); } // Emulation of Node's JSTransferable protocol (kTransfer/kTransferList/kDeserialize) for @@ -871,12 +721,6 @@ function applyWorkerProcessOverrides() { try { Object.defineProperty(proc, "debugPort", { value: 9229, writable: true, configurable: true, enumerable: true }); } catch {} - // These main-only internals are absent on a worker's process. - for (const k of ["_startProfilerIdleNotifier", "_stopProfilerIdleNotifier", "_debugProcess", "_debugEnd"]) { - try { - delete proc[k]; - } catch {} - } // process.umask(setMask) is unsupported in workers; the getter still works. const realUmask = proc.umask; function umask(mask?: unknown) { @@ -1080,14 +924,6 @@ class Worker extends EventEmitter { // user-supplied value so it can't trigger env sharing on its own. options = { ...options, shareEnv: undefined } as NodeWorkerOptions; } - // node runs its worker bootstrap before user code; preload the - // worker_threads module so process.stdin/stdout/stderr are always rebound, - // even when the worker never requires it. - const userPreload = (options as any).preload; - options = { - ...options, - preload: ["node:worker_threads", ...($isArray(userPreload) ? userPreload : userPreload ? [userPreload] : [])], - } as NodeWorkerOptions; this.#worker = new WebWorker(filename, options as Bun.WorkerOptions, this); // Create the readables eagerly so the worker's writev is ack'd even when // worker.stdout/stderr is never touched; only captured streams ref their diff --git a/src/jsc/bindings/BunClientData.cpp b/src/jsc/bindings/BunClientData.cpp index c538322f8e39..ce81e15532b9 100644 --- a/src/jsc/bindings/BunClientData.cpp +++ b/src/jsc/bindings/BunClientData.cpp @@ -1,6 +1,7 @@ #include "root.h" #include "BunClientData.h" +#include "WorkerMessagingProxy.h" #include "WebCoreJSBuiltins.h" #include "ExtendedDOMClientIsoSubspaces.h" @@ -103,12 +104,13 @@ JSVMClientData::~JSVMClientData() if (vmHandle) Bun__VmHandle__release(std::exchange(vmHandle, nullptr)); } -void JSVMClientData::create(VM* vm, void* bunVM, bool isWorkerVM) +void JSVMClientData::create(VM* vm, void* bunVM, WorkerMessagingProxy* worker) { auto provider = WebCore::createBuiltinsSourceProvider(); JSVMClientData* clientData = new JSVMClientData(*vm, provider); clientData->bunVM = bunVM; - clientData->m_isWorkerVM = isWorkerVM; + clientData->m_isWorkerVM = !!worker; + clientData->m_isNodeWorkerVM = worker && worker->options().kind == WorkerOptions::Kind::Node; clientData->vmHandle = Bun__VmHandle__retain(bunVM); clientData->vmHandleState = Bun__VmHandle__stateAddress(clientData->vmHandle); vm->deferredWorkTimer->onAddPendingWork = [clientData](Ref&& ticket, JSC::DeferredWorkTimer::WorkType kind) -> void { diff --git a/src/jsc/bindings/BunClientData.h b/src/jsc/bindings/BunClientData.h index 0140bf847c35..4e815c5cf419 100644 --- a/src/jsc/bindings/BunClientData.h +++ b/src/jsc/bindings/BunClientData.h @@ -20,6 +20,7 @@ bool takeTerminationOutsideScript(JSC::VM&, JSC::TopExceptionScope&); extern "C" bool Bun__VM__takeTerminationOutsideScript(JSC::JSGlobalObject*); namespace WebCore { +class WorkerMessagingProxy; class EventLoopTask; } // Post through a reference and give it up in one step (a reference taken only to outlive a lock). @@ -147,7 +148,8 @@ class JSVMClientData : public JSC::VM::ClientData { virtual ~JSVMClientData(); - static void create(JSC::VM*, void* bunVM, bool isWorkerVM); + // `worker` is the WorkerMessagingProxy this VM is being created for, or null on the main thread. + static void create(JSC::VM*, void* bunVM, WorkerMessagingProxy* worker); JSHeapData& heapData() { return *m_heapData; } BunBuiltinNames& builtinNames() { return m_builtinNames; } @@ -241,10 +243,13 @@ class JSVMClientData : public JSC::VM::ClientData { SentinelLinkedList> m_clients; bool m_isWorkerVM { false }; + bool m_isNodeWorkerVM { false }; public: // upstream's `&vm != commonVMOrNull()` bool isWorkerVM() const { return m_isWorkerVM; } + // Created by node:worker_threads' Worker (WorkerOptions::Kind::Node), as opposed to the Web Worker constructor. + bool isNodeWorkerVM() const { return m_isNodeWorkerVM; } // VM thread. Unlinking is the client's own (`remove()` in its destructor). void addClient(JSVMClientDataClient& client) { m_clients.append(&client); } }; diff --git a/src/jsc/bindings/BunProcess.cpp b/src/jsc/bindings/BunProcess.cpp index 303e8c1c93d6..62ed78d8a98d 100644 --- a/src/jsc/bindings/BunProcess.cpp +++ b/src/jsc/bindings/BunProcess.cpp @@ -2872,9 +2872,32 @@ enum class BunProcessStdinFdType : int32_t { extern "C" BunProcessStdinFdType Bun__Process__getStdinFdType(void*, int fd); extern "C" void Bun__ForceFileSinkToBeSynchronousForProcessObjectStdio(JSC::JSGlobalObject*, JSC::EncodedJSValue); +// node:worker_threads worker: process.stdout/stderr forward to the parent Worker's +// worker.stdout/stderr over a MessagePort; process.stdin is port-fed for { stdin: true } +// and otherwise an already-ended Readable (never the process-wide fd 0). fd 0/1/2 selects the port. +static JSValue constructNodeWorkerStdioStream(JSC::JSGlobalObject* globalObject, JSC::JSObject* processObject, JSObject* ports, int fd) +{ + auto& vm = JSC::getVM(globalObject); + auto scope = DECLARE_TOP_EXCEPTION_SCOPE(vm); + JSC::JSFunction* getStream = JSC::JSFunction::create(vm, globalObject, processObjectInternalsGetNodeWorkerStdioStreamCodeGenerator(vm), globalObject); + JSC::MarkedArgumentBuffer args; + args.append(processObject); + args.append(JSC::jsNumber(fd)); + args.append(ports); + auto result = JSC::profiledCall(globalObject, ProfilingReason::API, getStream, JSC::getCallData(getStream), globalObject->globalThis(), args); + if (auto* exception = scope.exception()) { + (void)scope.tryClearException(); + Zig::GlobalObject::reportUncaughtExceptionAtEventLoop(globalObject, exception); + return jsUndefined(); + } + return result; +} + static JSValue constructStdioWriteStream(JSC::JSGlobalObject* globalObject, JSC::JSObject* processObject, int fd) { auto& vm = JSC::getVM(globalObject); + if (auto* ports = defaultGlobalObject(globalObject)->nodeWorkerStdioPorts()) + return constructNodeWorkerStdioStream(globalObject, processObject, ports, fd); auto scope = DECLARE_TOP_EXCEPTION_SCOPE(vm); JSC::JSFunction* getStdioWriteStream = JSC::JSFunction::create(vm, globalObject, processObjectInternalsGetStdioWriteStreamCodeGenerator(vm), globalObject); @@ -2938,6 +2961,8 @@ static JSValue constructStderr(VM& vm, JSObject* processObject) static JSValue constructStdin(VM& vm, JSObject* processObject) { auto* globalObject = processObject->globalObject(); + if (auto* ports = defaultGlobalObject(globalObject)->nodeWorkerStdioPorts()) + return constructNodeWorkerStdioStream(globalObject, processObject, ports, 0); auto scope = DECLARE_TOP_EXCEPTION_SCOPE(vm); JSC::JSFunction* getStdinStream = JSC::JSFunction::create(vm, globalObject, processObjectInternalsGetStdinStreamCodeGenerator(vm), globalObject); JSC::MarkedArgumentBuffer args; @@ -4809,8 +4834,6 @@ extern "C" void Process__emitErrorEvent(Zig::GlobalObject* global, EncodedJSValu /* Source for Process.lut.h @begin processObjectTable - _debugEnd Process_stubEmptyFunction Function 0 - _debugProcess Process_stubEmptyFunction Function 0 _eval processGetEval CustomAccessor _getActiveHandles Process_stubFunctionReturningArray Function 0 _getActiveRequests Process_stubFunctionReturningArray Function 0 @@ -4818,8 +4841,6 @@ extern "C" void Process__emitErrorEvent(Zig::GlobalObject* global, EncodedJSValu _linkedBinding Process_stubEmptyFunction Function 0 _preload_modules Process_stubEmptyArray PropertyCallback _rawDebug constructRawDebug PropertyCallback - _startProfilerIdleNotifier Process_stubEmptyFunction Function 0 - _stopProfilerIdleNotifier Process_stubEmptyFunction Function 0 _tickCallback Process_stubEmptyFunction Function 0 abort Process_functionAbort Function 1 allowedNodeEnvironmentFlags constructAllowedNodeEnvironmentFlags PropertyCallback @@ -4939,6 +4960,15 @@ void Process::finishCreation(JSC::VM& vm) putDirect(vm, vm.propertyNames->toStringTagSymbol, jsString(vm, String("process"_s)), 0); putDirect(vm, Identifier::fromString(vm, "_exiting"_s), jsBoolean(false), 0); + + // No-op stubs Node only has on the main thread; a worker_threads Worker's process lacks them. + if (!WebCore::clientData(vm)->isNodeWorkerVM()) { + putDirectNativeFunction(vm, globalObject(), Identifier::fromString(vm, "_debugEnd"_s), 0, Process_stubEmptyFunction, ImplementationVisibility::Public, NoIntrinsic, 0); + putDirectNativeFunction(vm, globalObject(), Identifier::fromString(vm, "_debugProcess"_s), 0, Process_stubEmptyFunction, ImplementationVisibility::Public, NoIntrinsic, 0); + putDirectNativeFunction(vm, globalObject(), Identifier::fromString(vm, "_startProfilerIdleNotifier"_s), 0, Process_stubEmptyFunction, ImplementationVisibility::Public, NoIntrinsic, 0); + putDirectNativeFunction(vm, globalObject(), Identifier::fromString(vm, "_stopProfilerIdleNotifier"_s), 0, Process_stubEmptyFunction, ImplementationVisibility::Public, NoIntrinsic, 0); + } + // Node's addReadOnlyProcessAlias: read-only so `process.noDeprecation = false` // is ignored, but a per-Process property — a Worker must not flip the main // thread. Unflagged it stays an ordinary undefined slot user code can set. diff --git a/src/jsc/bindings/ZigGlobalObject.cpp b/src/jsc/bindings/ZigGlobalObject.cpp index b2507c275056..121c7a6d3c4f 100644 --- a/src/jsc/bindings/ZigGlobalObject.cpp +++ b/src/jsc/bindings/ZigGlobalObject.cpp @@ -511,7 +511,7 @@ extern "C" JSC::JSGlobalObject* Zig__GlobalObject__create(void* console_client, // Every JS VM's RunLoop should use Bun's RunLoop implementation ASSERT(vmPtr->runLoop().kind() == WTF::RunLoop::Kind::Bun); - WebCore::JSVMClientData::create(&vm, Bun__getVM(), /* isWorkerVM */ worker_ptr != nullptr); + WebCore::JSVMClientData::create(&vm, Bun__getVM(), static_cast(worker_ptr)); const auto createGlobalObject = [&]() -> Zig::GlobalObject* { if (executionContextId == std::numeric_limits::max() || executionContextId > 1) [[unlikely]] { @@ -4230,6 +4230,7 @@ void GlobalObject::adoptNapiEnvsForTestIsolation(GlobalObject* oldGlobal) } void GlobalObject::setNodeWorkerEnvironmentData(JSMap* data) { m_nodeWorkerEnvironmentData.set(vm(), this, data); } +void GlobalObject::setNodeWorkerStdioPorts(JSObject* ports) { m_nodeWorkerStdioPorts.set(vm(), this, ports); } void GlobalObject::setNodeWorkerEntryEvaluatedHook(JSObject* hook) { if (hook) diff --git a/src/jsc/bindings/ZigGlobalObject.h b/src/jsc/bindings/ZigGlobalObject.h index 9c5b615cab57..02cea87d7e94 100644 --- a/src/jsc/bindings/ZigGlobalObject.h +++ b/src/jsc/bindings/ZigGlobalObject.h @@ -545,6 +545,9 @@ class GlobalObject : public Bun::GlobalScope { /* setupMainThreadPort's drain callback; run once by WebWorker__entrySettled */ \ /* after entry-module evaluation. Stored here (not on globalThis) so user code can't clobber it. */ \ V(private, WriteBarrier, m_nodeWorkerEntryEvaluatedHook) \ + /* node:worker_threads worker: { stdin?, stdout, stderr } MessagePorts from the parent Worker; */ \ + /* process.stdin/stdout/stderr are built over these lazily (BunProcess.cpp constructStd*). */ \ + V(private, WriteBarrier, m_nodeWorkerStdioPorts) \ \ /* The original, unmodified Error.prepareStackTrace. */ \ /* */ \ @@ -782,6 +785,8 @@ class GlobalObject : public Bun::GlobalScope { JSObject* JSDOMFileConstructor() const { return m_JSDOMFileConstructor.getInitializedOnMainThread(this); } JSMap* nodeWorkerEnvironmentData() { return m_nodeWorkerEnvironmentData.get(); } + JSObject* nodeWorkerStdioPorts() { return m_nodeWorkerStdioPorts.get(); } + void setNodeWorkerStdioPorts(JSObject* ports); void setNodeWorkerEnvironmentData(JSMap* data); // node:worker_threads parentPort — the transferred MessagePort entangled with the parent // Worker's public port. Messages it dispatches are mirrored onto globalEventScope so the diff --git a/src/jsc/bindings/webcore/Worker.cpp b/src/jsc/bindings/webcore/Worker.cpp index a4b39d1b9df8..a2a578d5c713 100644 --- a/src/jsc/bindings/webcore/Worker.cpp +++ b/src/jsc/bindings/webcore/Worker.cpp @@ -27,7 +27,7 @@ #include "config.h" #include "Worker.h" -#include "BunClientData.h" +#include "InternalModuleRegistry.h" #include "ErrorCode.h" #include "ErrorEvent.h" #include "Event.h" @@ -202,6 +202,7 @@ extern "C" void WebWorker__dispatchError(Zig::GlobalObject* globalObject, Worker } JSC_DECLARE_HOST_FUNCTION(jsFunctionSetParentPort); +JSC_DECLARE_HOST_FUNCTION(jsFunctionSetNodeWorkerStdioPorts); JSC_DEFINE_HOST_FUNCTION(jsReceiveMessageOnPort, (JSGlobalObject * lexicalGlobalObject, CallFrame* callFrame)) { @@ -344,7 +345,7 @@ JSValue createNodeWorkerThreadsBinding(Zig::GlobalObject* globalObject) bool isNodeWorker = proxy && proxy->options().kind == WorkerOptions::Kind::Node; - JSObject* array = constructEmptyArray(globalObject, nullptr, 12); + JSObject* array = constructEmptyArray(globalObject, nullptr, 13); RETURN_IF_EXCEPTION(scope, {}); array->putDirectIndex(globalObject, 0, workerData); array->putDirectIndex(globalObject, 1, threadId); @@ -358,9 +359,27 @@ JSValue createNodeWorkerThreadsBinding(Zig::GlobalObject* globalObject) array->putDirectIndex(globalObject, 9, JSFunction::create(vm, globalObject, 1, "setEntryEvaluatedHook"_s, jsFunctionSetEntryEvaluatedHook, ImplementationVisibility::Public, NoIntrinsic)); array->putDirectIndex(globalObject, 10, jsBoolean(isNodeWorker)); array->putDirectIndex(globalObject, 11, JSFunction::create(vm, globalObject, 1, "setParentPort"_s, jsFunctionSetParentPort, ImplementationVisibility::Public, NoIntrinsic)); + array->putDirectIndex(globalObject, 12, JSFunction::create(vm, globalObject, 1, "setStdioPorts"_s, jsFunctionSetNodeWorkerStdioPorts, ImplementationVisibility::Public, NoIntrinsic)); return array; } +// worker_threads (worker side): { stdin?, stdout, stderr } ports from the parent Worker. +// process.stdin/stdout/stderr are created over them on first access (BunProcess.cpp). +JSC_DEFINE_HOST_FUNCTION(jsFunctionSetNodeWorkerStdioPorts, (JSGlobalObject * lexicalGlobalObject, CallFrame* callFrame)) +{ + if (auto* ports = callFrame->argument(0).getObject()) + defaultGlobalObject(lexicalGlobalObject)->setNodeWorkerStdioPorts(ports); + return JSValue::encode(jsUndefined()); +} + +// A node:worker_threads worker loads node:worker_threads before preloads and the entry +// point (it rebinds process stdio and registers parentPort / the postMessageToThread +// port), as Node runs its worker bootstrap before user code. Throws like require() would. +extern "C" void Bun__Worker__loadNodeWorkerThreadsModule(Zig::GlobalObject* globalObject) +{ + globalObject->internalModuleRegistry()->requireId(globalObject, globalObject->vm(), Bun::InternalModuleRegistry::Field::NodeWorkerThreads); +} + // worker_threads (worker side): register the transferred port as this thread's parentPort. JSC_DEFINE_HOST_FUNCTION(jsFunctionSetParentPort, (JSGlobalObject * lexicalGlobalObject, CallFrame* callFrame)) { diff --git a/src/jsc/bindings/webcore/WorkerMessagingProxy.cpp b/src/jsc/bindings/webcore/WorkerMessagingProxy.cpp index 22d8e944916c..1562d0d8e381 100644 --- a/src/jsc/bindings/webcore/WorkerMessagingProxy.cpp +++ b/src/jsc/bindings/webcore/WorkerMessagingProxy.cpp @@ -60,6 +60,7 @@ void* WebWorker__create( bool miniMode, bool unrefByDefault, bool evalMode, + bool isNodeWorker, StringImpl** argvPtr, size_t argvLen, bool defaultExecArgv, @@ -151,6 +152,7 @@ ExceptionOr WorkerMessagingProxy::startWorkerGlobalScope(const String& scr m_options.mini, m_options.unref, m_options.evalMode, + m_options.kind == WorkerOptions::Kind::Node, reinterpret_cast(m_options.argv.begin()), m_options.argv.size(), !m_options.execArgv.has_value(), diff --git a/src/jsc/web_worker.rs b/src/jsc/web_worker.rs index 0c631fb31c4d..3778bc99127d 100644 --- a/src/jsc/web_worker.rs +++ b/src/jsc/web_worker.rs @@ -72,6 +72,9 @@ pub struct WebWorker { execution_context_id: u32, mini: bool, eval_mode: bool, + /// Created by `node:worker_threads`' `Worker` (as opposed to the Web `Worker` + /// constructor): loads `node:worker_threads` before preloads and the entry point. + is_node_worker: bool, store_fd: bool, /// Borrowed from the proxy's `WorkerOptions` (alive as long as the proxy). argv_ptr: *const WTFStringImpl, @@ -160,6 +163,9 @@ unsafe extern "C" { ); safe fn WebWorker__parentContextWillDestroy(proxy: *mut c_void); safe fn WebWorker__entrySettled(global: &JSGlobalObject); + /// Loads `node:worker_threads` in this VM (it rebinds process stdio and + /// registers parentPort). May leave an exception pending. + safe fn Bun__Worker__loadNodeWorkerThreadsModule(global: &JSGlobalObject); safe fn WebWorker__dispatchError( global: &JSGlobalObject, proxy: *mut c_void, @@ -284,6 +290,7 @@ impl WebWorker { mini: bool, default_unref: bool, eval_mode: bool, + is_node_worker: bool, argv_ptr: *const WTFStringImpl, argv_len: usize, inherit_exec_argv: bool, @@ -320,8 +327,7 @@ impl WebWorker { for module in preload_modules { let utf8_slice = module.to_utf8(); // node: builtin specifiers skip the file resolver — the worker-side - // module loader resolves them. Lets node:worker_threads run its - // bootstrap (stdio rebinding) as a preload. + // module loader resolves them. if utf8_slice.slice().starts_with(b"node:") { preloads.push(utf8_slice.slice().to_vec().into_boxed_slice()); continue; @@ -396,6 +402,7 @@ impl WebWorker { execution_context_id: this_context_id, mini, eval_mode, + is_node_worker, store_fd, argv_ptr, argv_len, @@ -842,6 +849,24 @@ impl WebWorker { return self.shutdown(); } + // Node runs its worker bootstrap (parentPort, stdio, process overrides) + // ahead of user code; ours is node:worker_threads' module body. + if self.is_node_worker { + let global = vm.global(); + if let Err(err) = jsc::host_fn::from_js_host_call_generic(global, || { + Bun__Worker__loadNodeWorkerThreadsModule(global) + }) { + let exception = global.take_exception(err); + let _ = vm.as_mut().uncaught_exception(global, exception, false); + if !self.exit_called.load(Ordering::Relaxed) { + vm.as_mut().exit_handler.exit_code = 1; + } + self.flush_logs(vm); + WebWorker__entrySettled(global); + return self.shutdown(); + } + } + // `path` borrows the resolver's process-lifetime string store, the // standalone module graph, or `self.unresolved_specifier` — all of // which outlive the worker VM. `vm.main` stores it as a raw BACKREF diff --git a/src/runtime/bake/BakeGlobalObject.cpp b/src/runtime/bake/BakeGlobalObject.cpp index 766b100614fe..f1bed29daebc 100644 --- a/src/runtime/bake/BakeGlobalObject.cpp +++ b/src/runtime/bake/BakeGlobalObject.cpp @@ -248,7 +248,7 @@ extern "C" GlobalObject* BakeCreateProdGlobal(void* console) vm.heap.acquireAccess(); JSC::JSLockHolder locker(vm); BunVirtualMachine* bunVM = Bun__getVM(); - WebCore::JSVMClientData::create(&vm, bunVM, /* isWorkerVM */ false); + WebCore::JSVMClientData::create(&vm, bunVM, /* worker */ nullptr); JSC::Structure* structure = Bake::GlobalObject::createStructure(vm); Bake::GlobalObject* global = Bake::GlobalObject::create( diff --git a/src/runtime/jsc_hooks.rs b/src/runtime/jsc_hooks.rs index 92d2bfdd6be1..72b9ffcaa347 100644 --- a/src/runtime/jsc_hooks.rs +++ b/src/runtime/jsc_hooks.rs @@ -767,9 +767,7 @@ unsafe fn load_preloads(vm: *mut VirtualMachine) -> bun_jsc::CrateResult<*mut JS .unwrap_or(preload_slice); // node: builtin specifiers bypass the file resolver — JSModuleLoader - // resolves them internally. node:worker_threads is preloaded this way so - // its node-style worker bootstrap (stdio rebinding) runs before user code; - // this also means `bun --import node:*` works like Node's. + // resolves them internally, so `bun --import node:*` works like Node's. let module_name = if normalized.starts_with(b"node:") { bun_core::String::from_bytes(normalized) } else {