From 0530d580a6f24bd28afb77436bf26ec8c016932b Mon Sep 17 00:00:00 2001 From: tsctx <91457664+tsctx@users.noreply.github.com> Date: Sat, 20 Jan 2024 22:48:48 +0900 Subject: [PATCH] stream: fix cloned webstreams not being unref correctly --- lib/internal/webstreams/readablestream.js | 6 +++- lib/internal/webstreams/transfer.js | 36 +++++++++++++------ lib/internal/webstreams/writablestream.js | 8 +++-- test/parallel/test-webstreams-clone-unref.js | 16 +++++++++ .../test-whatwg-webstreams-transfer.js | 11 ++++++ 5 files changed, 63 insertions(+), 14 deletions(-) create mode 100644 test/parallel/test-webstreams-clone-unref.js diff --git a/lib/internal/webstreams/readablestream.js b/lib/internal/webstreams/readablestream.js index 62dfd8a288ab4c..d4526011bec46d 100644 --- a/lib/internal/webstreams/readablestream.js +++ b/lib/internal/webstreams/readablestream.js @@ -607,7 +607,11 @@ class ReadableStream { const transfer = lazyTransfer(); setupReadableStreamDefaultControllerFromSource( this, - new transfer.CrossRealmTransformReadableSource(port), + // The MessagePort is set to be referenced when reading. + // After two MessagePorts are closed, there is a problem with + // lingering promise not being properly resolved. + // https://github.com/nodejs/node/issues/51486 + new transfer.CrossRealmTransformReadableSource(port, true), 0, () => 1); } } diff --git a/lib/internal/webstreams/transfer.js b/lib/internal/webstreams/transfer.js index 136b0d81a99464..9835e6ab272c98 100644 --- a/lib/internal/webstreams/transfer.js +++ b/lib/internal/webstreams/transfer.js @@ -102,10 +102,11 @@ function InternalCloneableDOMException() { InternalCloneableDOMException[kDeserialize] = () => {}; class CrossRealmTransformReadableSource { - constructor(port) { + constructor(port, unref) { this[kState] = { port, controller: undefined, + unref, }; port.onmessage = ({ data }) => { @@ -143,6 +144,8 @@ class CrossRealmTransformReadableSource { error); port.close(); }; + + port.unref(); } start(controller) { @@ -150,6 +153,10 @@ class CrossRealmTransformReadableSource { } async pull() { + if (this[kState].unref) { + this[kState].unref = false; + this[kState].port.ref(); + } this[kState].port.postMessage({ type: 'pull' }); } @@ -170,11 +177,12 @@ class CrossRealmTransformReadableSource { } class CrossRealmTransformWritableSink { - constructor(port) { + constructor(port, unref) { this[kState] = { port, controller: undefined, backpressurePromise: createDeferredPromise(), + unref, }; port.onmessage = ({ data }) => { @@ -211,6 +219,7 @@ class CrossRealmTransformWritableSink { port.close(); }; + port.unref(); } start(controller) { @@ -218,6 +227,10 @@ class CrossRealmTransformWritableSink { } async write(chunk) { + if (this[kState].unref) { + this[kState].unref = false; + this[kState].port.ref(); + } if (this[kState].backpressurePromise === undefined) { this[kState].backpressurePromise = { promise: PromiseResolve(), @@ -262,12 +275,12 @@ class CrossRealmTransformWritableSink { } function newCrossRealmReadableStream(writable, port) { - const readable = - new ReadableStream( - new CrossRealmTransformReadableSource(port)); + // MessagePort should always be unref. + // There is a problem with the process not terminating. + // https://github.com/nodejs/node/issues/44985 + const readable = new ReadableStream(new CrossRealmTransformReadableSource(port, false)); - const promise = - readableStreamPipeTo(readable, writable, false, false, false); + const promise = readableStreamPipeTo(readable, writable, false, false, false); setPromiseHandled(promise); @@ -278,12 +291,15 @@ function newCrossRealmReadableStream(writable, port) { } function newCrossRealmWritableSink(readable, port) { - const writable = - new WritableStream( - new CrossRealmTransformWritableSink(port)); + // MessagePort should always be unref. + // There is a problem with the process not terminating. + // https://github.com/nodejs/node/issues/44985 + const writable = new WritableStream(new CrossRealmTransformWritableSink(port, false)); const promise = readableStreamPipeTo(readable, writable, false, false, false); + setPromiseHandled(promise); + return { writable, promise, diff --git a/lib/internal/webstreams/writablestream.js b/lib/internal/webstreams/writablestream.js index eea99f321d50c6..121afc0e181ec6 100644 --- a/lib/internal/webstreams/writablestream.js +++ b/lib/internal/webstreams/writablestream.js @@ -263,8 +263,6 @@ class WritableStream { this[kState].transfer.readable = readable; this[kState].transfer.promise = promise; - setPromiseHandled(this[kState].transfer.promise); - return { data: { port: this[kState].transfer.port2 }, deserializeInfo: @@ -283,7 +281,11 @@ class WritableStream { const transfer = lazyTransfer(); setupWritableStreamDefaultControllerFromSink( this, - new transfer.CrossRealmTransformWritableSink(port), + // The MessagePort is set to be referenced when reading. + // After two MessagePorts are closed, there is a problem with + // lingering promise not being properly resolved. + // https://github.com/nodejs/node/issues/51486 + new transfer.CrossRealmTransformWritableSink(port, true), 1, () => 1); } diff --git a/test/parallel/test-webstreams-clone-unref.js b/test/parallel/test-webstreams-clone-unref.js new file mode 100644 index 00000000000000..88a9cebd9c3046 --- /dev/null +++ b/test/parallel/test-webstreams-clone-unref.js @@ -0,0 +1,16 @@ +'use strict'; + +require('../common'); +const { ok } = require('node:assert'); + +// This test verifies that cloned ReadableStream and WritableStream instances +// do not keep the process alive. The test fails if it timesout (it should just +// exit immediately) + +const rs1 = new ReadableStream(); +const ws1 = new WritableStream(); + +const [rs2, ws2] = structuredClone([rs1, ws1], { transfer: [rs1, ws1] }); + +ok(rs2 instanceof ReadableStream); +ok(ws2 instanceof WritableStream); diff --git a/test/parallel/test-whatwg-webstreams-transfer.js b/test/parallel/test-whatwg-webstreams-transfer.js index 01cfaa02ad075e..7be01c339652c0 100644 --- a/test/parallel/test-whatwg-webstreams-transfer.js +++ b/test/parallel/test-whatwg-webstreams-transfer.js @@ -464,12 +464,23 @@ const theData = 'hello'; tracker.verify(); }); + // We create an interval to keep the event loop alive while + // we wait for the stream read to complete. The reason this is needed is because there's + // otherwise nothing to keep the worker thread event loop alive long enough to actually + // complete the read from the stream. Under the covers the ReadableStream uses an + // unref'd MessagePort to communicate with the main thread. Because the MessagePort + // is unref'd, it's existence would not keep the thread alive on its own. There was previously + // a bug where this MessagePort was ref'd which would block the thread and main thread + // from terminating at all unless the stream was consumed/closed. + const i = setInterval(() => {}, 1000); + parentPort.onmessage = tracker.calls(({ data }) => { assert(isReadableStream(data)); const reader = data.getReader(); reader.read().then(tracker.calls((result) => { assert(!result.done); assert(result.value instanceof Uint8Array); + clearInterval(i); })); parentPort.close(); });