diff --git a/src/js/node/worker_threads.ts b/src/js/node/worker_threads.ts index 5fa6aa8707f8..203c19e193b8 100644 --- a/src/js/node/worker_threads.ts +++ b/src/js/node/worker_threads.ts @@ -342,11 +342,11 @@ function makePortReadable(port, incrementsPortRef) { let startedReading = false; function onMessage(payload) { if (payload === null) { + // The listener (and with it the port's ref) comes off in 'close', once the buffered data is consumed. 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])); diff --git a/src/jsc/bindings/webcore/JSMessagePort.cpp b/src/jsc/bindings/webcore/JSMessagePort.cpp index 3b70232ef476..ed488f02babc 100644 --- a/src/jsc/bindings/webcore/JSMessagePort.cpp +++ b/src/jsc/bindings/webcore/JSMessagePort.cpp @@ -203,12 +203,9 @@ static inline bool setJSMessagePort_onmessageSetter(JSGlobalObject& lexicalGloba vm.writeBarrier(&thisObject, value); ensureStillAliveHere(value); - // node: a callable handler starts the port and keeps the loop alive; assigning anything else - // clears the handler and lets the loop exit again. + // node: a callable handler keeps the loop alive; clearing one is an ordinary listener removal (onDidChangeListenerImpl). if (value.isCallable()) thisObject.wrapped().jsRef(&lexicalGlobalObject); - else - thisObject.wrapped().jsUnref(&lexicalGlobalObject); return true; } @@ -354,7 +351,7 @@ static inline JSC::EncodedJSValue jsMessagePortPrototypeFunction_closeBody(JSC:: UNUSED_PARAM(throwScope); UNUSED_PARAM(callFrame); auto& impl = castedThis->wrapped(); - impl.jsUnref(lexicalGlobalObject); + impl.jsUnref(); RELEASE_AND_RETURN(throwScope, JSValue::encode(toJS(*lexicalGlobalObject, throwScope, [&]() -> decltype(auto) { return impl.close(); }))); } @@ -385,7 +382,7 @@ static inline JSC::EncodedJSValue jsMessagePortPrototypeFunction_unrefBody(JSC:: UNUSED_PARAM(throwScope); UNUSED_PARAM(callFrame); auto& impl = castedThis->wrapped(); - RELEASE_AND_RETURN(throwScope, JSValue::encode(toJS(*lexicalGlobalObject, throwScope, [&]() -> decltype(auto) { return impl.jsUnref(lexicalGlobalObject); }))); + RELEASE_AND_RETURN(throwScope, JSValue::encode(toJS(*lexicalGlobalObject, throwScope, [&]() -> decltype(auto) { return impl.jsUnref(); }))); } JSC_DEFINE_HOST_FUNCTION(jsMessagePortPrototypeFunction_unref, (JSGlobalObject * lexicalGlobalObject, CallFrame* callFrame)) diff --git a/src/jsc/bindings/webcore/MessagePort.cpp b/src/jsc/bindings/webcore/MessagePort.cpp index 1e1681c17d85..bb122335b625 100644 --- a/src/jsc/bindings/webcore/MessagePort.cpp +++ b/src/jsc/bindings/webcore/MessagePort.cpp @@ -224,15 +224,8 @@ void MessagePort::close() // it in hasPendingActivity()); marking our side Closed is sufficient. m_pipe->close(m_side, MessagePortPipe::CloseKind::Explicit); - // Release the self-reference taken by jsRef() (set when .onmessage is - // assigned or .ref() is called from JS). The JS .close() binding calls - // jsUnref() first; stop() and contextDestroyed() do not. - if (m_hasRef) { - m_hasRef = false; - if (auto* context = scriptExecutionContext()) - context->unrefEventLoop(); - deref(); - } + // The JS .close() binding calls jsUnref() first; stop() and contextDestroyed() do not. + releaseJsRef(); // close() can run without a prior jsUnref() (warn-and-close, contextDestroyed()); // clear the listener keepalive so a later listener add can't re-ref the loop. @@ -287,8 +280,7 @@ void MessagePort::peerClosed() dispatchCloseEvent(); // jsUnref() clears both the listener loop-ref (m_isRefd) and the onmessage/ref() // keepalive (m_hasRef), so a listening transferred port stops pinning the loop. - auto* globalObject = defaultGlobalObject(context->globalObject()); - jsUnref(globalObject); + jsUnref(); } TransferredMessagePort MessagePort::disentangle() @@ -301,17 +293,8 @@ TransferredMessagePort MessagePort::disentangle() removeAllEventListeners(); m_hasMessageEventListener = false; - // Release the self-reference taken by jsRef() on the sending side. After - // transfer this object is inert (the receiving side gets a fresh - // MessagePort for the same pipe endpoint) and is no longer a destruction - // observer, so nothing else will ever release a ref taken here. - // The caller (disentanglePorts) holds a RefPtr, so deref() is safe. - if (m_hasRef) { - m_hasRef = false; - if (auto* context = scriptExecutionContext()) - context->unrefEventLoop(); - deref(); - } + // Inert from here on and about to stop observing its context: nothing later could release a jsRef(). + releaseJsRef(); // A transferred port is inert; clear the listener keepalive too so hasRef() // reports false (the disentangle analogue of the close() reset above). @@ -400,6 +383,8 @@ void MessagePort::contextDestroyed() { ASSERT(scriptExecutionContext()); + // With no stop phase before this (ShadowRealm, retired test-isolation global), close() may drop the last reference. + Ref protectedThis { *this }; close(); ActiveDOMObject::contextDestroyed(); } @@ -497,6 +482,7 @@ void MessagePort::onDidChangeListenerImpl(EventTarget& self, const AtomString& e return; auto& port = static_cast(self); + bool hadListeners = port.m_messageEventCount > 0; switch (kind) { case Add: port.m_messageEventCount++; @@ -510,6 +496,9 @@ void MessagePort::onDidChangeListenerImpl(EventTarget& self, const AtomString& e break; } port.updateListenerEventLoopRef(); + // node (setupPortReferencing) unref()s outright when the last 'message' listener goes, .ref() or not. + if (hadListeners && port.m_messageEventCount == 0) + port.releaseJsRef(); } bool MessagePort::addEventListener(const AtomString& eventType, Ref&& listener, const AddEventListenerOptions& options) @@ -576,7 +565,7 @@ void MessagePort::jsRef(JSGlobalObject* lexicalGlobalObject) } } -void MessagePort::jsUnref(JSGlobalObject* lexicalGlobalObject) +void MessagePort::jsUnref() { // Also release the listener loop-ref; otherwise an always-listening transferred // port (a postMessageToThread control port) would pin the event loop forever. @@ -584,11 +573,20 @@ void MessagePort::jsUnref(JSGlobalObject* lexicalGlobalObject) m_isRefd = false; updateListenerEventLoopRef(); } - if (m_hasRef) { - m_hasRef = false; - deref(); - Bun__eventLoop__refKeepAlive(WebCore::clientData(lexicalGlobalObject->vm())->bunVM, -1); - } + releaseJsRef(); +} + +void MessagePort::releaseJsRef() +{ + if (!m_hasRef) + return; + m_hasRef = false; + // The context's VM is the one jsRef() ref'd through the lexical global. + if (auto* context = scriptExecutionContext()) + context->unrefEventLoop(); + // Callers keep using the port afterwards, so this self-ref must not be its last reference. + ASSERT(!hasOneRef()); + deref(); } } // namespace WebCore diff --git a/src/jsc/bindings/webcore/MessagePort.h b/src/jsc/bindings/webcore/MessagePort.h index c08911798f25..e44d8fc990ef 100644 --- a/src/jsc/bindings/webcore/MessagePort.h +++ b/src/jsc/bindings/webcore/MessagePort.h @@ -114,7 +114,7 @@ class MessagePort final : public ActiveDOMObject, public EventTarget, public Thr JSValue tryTakeMessage(JSGlobalObject*, bool& hadMessage); void jsRef(JSGlobalObject*); - void jsUnref(JSGlobalObject*); + void jsUnref(); // Report the actual loop-ref state (matches Node's uv_has_ref), not the intent flag. bool jsHasRef() { return m_hasRef || m_listenerLoopRefActive; } @@ -159,7 +159,9 @@ class MessagePort final : public ActiveDOMObject, public EventTarget, public Thr // Read from the GC thread: a port whose only listener is 'close' must survive // until that event is delivered, or the peer's close is lost to a collection. std::atomic m_hasCloseEventListener { false }; + // jsRef() (.ref() or a callable .onmessage=) holds a self-ref plus an event-loop ref; releaseJsRef() drops both. bool m_hasRef { false }; + void releaseJsRef(); // Whether .ref()/.unref() want this port to keep the loop alive (default refd); // independent of m_hasRef (the .onmessage=/.ref() keepalive). diff --git a/test/js/node/worker_threads/worker_threads.test.ts b/test/js/node/worker_threads/worker_threads.test.ts index 023ea833ffa1..2553654ea7c5 100644 --- a/test/js/node/worker_threads/worker_threads.test.ts +++ b/test/js/node/worker_threads/worker_threads.test.ts @@ -997,6 +997,262 @@ test("hasRef() survives collection of the unreferenced peer", () => { port1.close(); }); +// port1.hasRef() after running `scenario` on a fresh channel. port2 stays reachable +// until the end so that collecting it cannot close port1 underneath the scenario. +async function hasRefAfter(scenario: (port1: MessagePort, port2: MessagePort) => void | Promise) { + const { port1, port2 } = new MessageChannel(); + try { + await scenario(port1, port2); + return port1.hasRef(); + } finally { + port1.close(); + port2.close(); + } +} + +// node's setupPortReferencing (lib/internal/worker/io.js) unref()s a port when its +// last 'message' listener is removed, whether or not ref() was called on it before. +test("removing the last 'message' listener releases an explicit ref() as well", async () => { + const f = () => {}; + const g = () => {}; + const results = { + refThenListenerRemoved: await hasRefAfter(p => { + p.ref(); + p.on("message", f); + p.off("message", f); + }), + listenerThenRefThenRemoved: await hasRefAfter(p => { + p.on("message", f); + p.ref(); + p.off("message", f); + }), + viaRemoveEventListener: await hasRefAfter(p => { + p.ref(); + p.addEventListener("message", f); + p.removeEventListener("message", f); + }), + viaOnceListenerFiring: await hasRefAfter(async (p, peer) => { + p.ref(); + const fired = new Promise(resolve => p.once("message", () => resolve())); + peer.postMessage(1); + await fired; + }), + // None of these remove the last 'message' listener, so nothing is released. + oneOfTwoRemoved: await hasRefAfter(p => { + p.ref(); + p.on("message", f); + p.on("message", g); + p.off("message", f); + }), + neverAddedListenerRemoved: await hasRefAfter(p => { + p.ref(); + p.off("message", f); + }), + closeListenerRemoved: await hasRefAfter(p => { + p.ref(); + p.on("close", f); + p.off("close", f); + }), + // The release is not sticky: a later listener or ref() refs the port again. + listenerAddedAgain: await hasRefAfter(p => { + p.ref(); + p.on("message", f); + p.off("message", f); + p.on("message", g); + }), + refCalledAgain: await hasRefAfter(p => { + p.ref(); + p.on("message", f); + p.off("message", f); + p.ref(); + }), + }; + expect(results).toEqual({ + refThenListenerRemoved: false, + listenerThenRefThenRemoved: false, + viaRemoveEventListener: false, + viaOnceListenerFiring: false, + oneOfTwoRemoved: true, + neverAddedListenerRemoved: true, + closeListenerRemoved: true, + listenerAddedAgain: true, + refCalledAgain: true, + }); +}); + +// bun's removeAllListeners() removes the listeners one by one, so it releases the port +// exactly like off() on each of them would. (node's NodeEventTarget#removeAllListeners +// bypasses its listener hooks and leaves a port with no listeners ref'd; bun does not +// keep the listener-count ref there either, and the explicit ref follows the same rule.) +test("removeAllListeners() releases an explicit ref() like off() does", async () => { + const f = () => {}; + const results = { + byType: await hasRefAfter(p => { + p.ref(); + p.on("message", f); + p.removeAllListeners("message"); + }), + all: await hasRefAfter(p => { + p.ref(); + p.on("message", f); + p.removeAllListeners(); + }), + unrelatedType: await hasRefAfter(p => { + p.ref(); + p.on("message", f); + p.removeAllListeners("close"); + }), + }; + expect(results).toEqual({ byType: false, all: false, unrelatedType: true }); +}); + +// node: assigning a non-object to onmessage removes the handler, and the port is +// unref()'d only if that removed its last 'message' listener. Clearing a handler that +// was never set, or while on() listeners remain, leaves the ref state alone. +// +// A non-callable object is not a removal: the event handler attribute installs it (the +// getter returns it, and it occupies the 'message' listener slot), so the ref state simply +// follows the listener list: ref'd while it is installed, released when it is cleared. +// node agrees on installing one over nothing; it differs on the two rarer transitions +// (it only counts functions, so replacing a function with an object unrefs there and +// clearing an object does not), where bun keeps hasRef() in step with the listener list. +test("onmessage = releases the port only when it removed the last 'message' listener", async () => { + const f = () => {}; + const g = () => {}; + const results = { + handlerCleared: await hasRefAfter(p => { + p.onmessage = f; + p.onmessage = null; + }), + handlerClearedByString: await hasRefAfter(p => { + p.onmessage = f; + p.onmessage = "not a handler" as any; + }), + refdHandlerCleared: await hasRefAfter(p => { + p.ref(); + p.onmessage = f; + p.onmessage = null; + }), + nothingToClear: await hasRefAfter(p => { + p.ref(); + p.onmessage = null; + }), + onListenerRemains: await hasRefAfter(p => { + p.on("message", f); + p.onmessage = g; + p.onmessage = null; + }), + onListenerOnly: await hasRefAfter(p => { + p.on("message", f); + p.onmessage = null; + }), + objectHandlerInstalled: await hasRefAfter(p => { + p.onmessage = {} as any; + }), + functionReplacedByObject: await hasRefAfter(p => { + p.onmessage = f; + p.onmessage = {} as any; + }), + objectHandlerCleared: await hasRefAfter(p => { + p.onmessage = {} as any; + p.onmessage = null; + }), + }; + expect(results).toEqual({ + handlerCleared: false, + handlerClearedByString: false, + refdHandlerCleared: false, + nothingToClear: true, + onListenerRemains: true, + onListenerOnly: true, + objectHandlerInstalled: true, + functionReplacedByObject: true, + objectHandlerCleared: false, + }); +}); + +// End to end: once the listener is gone nothing about the port may hold the process open. +test.concurrent("a ref()'d port whose last 'message' listener was removed lets the process exit", async () => { + await using proc = Bun.spawn({ + cmd: [ + bunExe(), + "-e", + `const { MessageChannel } = require("worker_threads"); + const { port1, port2 } = new MessageChannel(); + // Keep the peer reachable: collecting it would close port1 (and release its refs) by itself. + globalThis.keep = port2; + const f = () => {}; + port1.ref(); + port1.on("message", f); + port1.off("message", f); + console.log("hasRef=" + port1.hasRef()); + // Never fires when the port released the loop; bounds the failure mode (a process + // that stays alive) instead of letting the test run into its timeout. + setTimeout(() => { console.log("still alive"); process.exit(1); }, 2_000).unref();`, + ], + env: bunEnv, + stdout: "pipe", + stderr: "inherit", + }); + const [stdout, exitCode] = await Promise.all([proc.stdout.text(), proc.exited]); + expect(stdout).toBe("hasRef=false\n"); + expect(exitCode).toBe(0); +}); + +// Same rule for parentPort: node's worker exits once the listeners are gone even if +// the script ref()'d the port explicitly. +test("a ref()'d parentPort whose last 'message' listener was removed lets the worker exit", async () => { + const w = new Worker( + `const { parentPort } = require("worker_threads"); + const f = () => {}; + parentPort.ref(); + parentPort.on("message", f); + parentPort.off("message", f); + parentPort.postMessage(parentPort.hasRef()); + // Never fires when the port released the loop; bounds the failure mode (a worker + // that stays alive) with a distinctive exit code. + setTimeout(() => process.exit(7), 2_000).unref();`, + { eval: true }, + ); + const hasRef = once(w, "message"); + const exited = once(w, "exit"); + expect({ hasRef: (await hasRef)[0], exitCode: (await exited)[0] }).toEqual({ hasRef: false, exitCode: 0 }); +}); + +// The worker's process.stdin is fed by a port that is ref()'d once reading starts. Data +// that is still buffered when EOF arrives has to keep the thread alive until it is +// consumed (node's kWaitingStreams), so the stream may only drop its port listener (which +// releases that ref) after it has been drained, not when the EOF message comes in. +test("stdin data buffered behind a paused consumer is still delivered after EOF arrives", async () => { + const w = new Worker( + `const { parentPort } = require("worker_threads"); + const seen = []; + process.stdin.on("data", chunk => { + seen.push(chunk.toString()); + if (seen.length !== 1) return; + process.stdin.pause(); + // Resume once EOF has been received, via unref'd timers only: the buffered data is + // then the one thing entitled to keep this thread alive. A thread that exits at EOF + // never gets here again and reports nothing. + const poll = () => { + if (process.stdin._readableState.ended) process.stdin.resume(); + else setTimeout(poll, 1).unref(); + }; + setTimeout(poll, 1).unref(); + }); + process.stdin.on("end", () => parentPort.postMessage(seen));`, + { eval: true, stdin: true }, + ); + const reports: string[][] = []; + w.on("message", report => reports.push(report)); + const exited = once(w, "exit"); + w.stdin!.write("aaaaaaaaaa"); + w.stdin!.write("bbbbbbbbbb"); + w.stdin!.end(); + const [exitCode] = await exited; + expect({ reports, exitCode }).toEqual({ reports: [["aaaaaaaaaa", "bbbbbbbbbb"]], exitCode: 0 }); +}); + // markAsUncloneable blocks *cloning*, not transfer: a marked port in the transfer // list is moved, so node lets it through and it still works on the far side. test("markAsUncloneable blocks cloning a port but not transferring it", async () => { diff --git a/test/js/web/workers/message-port-context-destroy-leak.test.ts b/test/js/web/workers/message-port-context-destroy-leak.test.ts index 270d44248a85..4a0cd4902cbe 100644 --- a/test/js/web/workers/message-port-context-destroy-leak.test.ts +++ b/test/js/web/workers/message-port-context-destroy-leak.test.ts @@ -79,3 +79,53 @@ test.skipIf(isWindows)( }, 120_000, ); + +// The other half of the contract above: once the wrapper and the peer are collected, the +// self-ref taken by .ref() is the only thing keeping the native port alive. A context that +// is torn down without a stop phase (a collected ShadowRealm global here) reaches +// contextDestroyed() -> close() directly, and close() drops that self-ref part-way +// through, so it has to hold its own reference for the rest of the teardown. +// +// ASAN only: the port lives in a bmalloc heap, which Malloc=1 routes to the system +// allocator so ASAN can see a read of the freed port. +test.skipIf(!isASAN)( + "closing a ref()'d port during context destruction does not free it mid-close", + async () => { + await using proc = Bun.spawn({ + cmd: [ + bunExe(), + "-e", + `const { heapStats } = require("bun:jsc"); + const globals = () => heapStats().objectTypeCounts.GlobalObject; + const baseline = globals(); + let realm = new ShadowRealm(); + realm.evaluate("(() => { const { port1 } = new MessageChannel(); port1.ref(); })()"); + // Collect the wrapper and the peer while the realm is still alive: the ref()'d native + // port now survives on its self-ref alone. + for (let i = 0; i < 5; i++) Bun.gc(true); + // Drop the realm. Destroying its global destroys the context the port is registered on. + // A stale stack slot can keep the realm alive through a collection or two, so collect + // (with a different call in between each time) until its global is actually gone. + realm = null; + let collected = false; + for (let i = 0; i < 50 && !collected; i++) { + Bun.gc(true); + collected = globals() === baseline; + } + console.log(collected ? "PASS" : "the realm's global was never collected");`, + ], + env: { ...bunEnv, ...(isWindows ? {} : { Malloc: "1" }) }, + stdout: "pipe", + stderr: "pipe", + }); + + const [stdout, stderr, exitCode] = await Promise.all([proc.stdout.text(), proc.stderr.text(), proc.exited]); + + expect(stderr).toBe(""); + expect(stdout).toBe("PASS\n"); + expect(exitCode).toBe(0); + }, + // The passing run takes about a second; a failing one has to symbolize an ASAN report + // for the debug binary first, which takes longer than the default timeout. + 30_000, +);