Commit 4e75b714df6 for nodejs
commit 4e75b714df611f0721b19f4d9dcdf85e694cf5a2
Author: Matteo Collina <hello@matteocollina.com>
Date: Sat Oct 17 08:20:05 2026 +0200
stream: trim TransformStream construction costs
Constructing a TransformStream allocated five closures for the sink and
source algorithms of its two sides, and each side adopted the start
promise through a wrapper promise plus a thenable job, so a stream
without a start() left about three kilobytes pending in the microtask
queue until the started steps ran. Per-stream creation bursts spend
most of their time copying that graph through the scavenger.
The sink and source algorithms are now shared functions that reach the
transform stream through a field on their controller state (the
controller is passed to the close, abort and cancel algorithms for
that), and the post-start steps of both sides are delivered by one
reaction chain on a shared promise, taking the same microtask hops as
the spec's start promise adoption: three after construction for a
non-thenable start result, two after the adopting promise settles for a
thenable one. The readable and writable transfer state records are
materialized on first transfer instead of per stream.
Signed-off-by: Matteo Collina <hello@matteocollina.com>
PR-URL: https://github.com/nodejs/node/pull/66643
Reviewed-By: James M Snell <jasnell@gmail.com>
Reviewed-By: Mattias Buelens <mattias@buelens.com>
Reviewed-By: Yagiz Nizipli <yagiz@nizipli.com>
diff --git a/lib/internal/webstreams/readablestream.js b/lib/internal/webstreams/readablestream.js
index 1b88b867216..f19337f5b6b 100644
--- a/lib/internal/webstreams/readablestream.js
+++ b/lib/internal/webstreams/readablestream.js
@@ -111,6 +111,7 @@ const {
extractSizeAlgorithm,
getNonWritablePropertyDescriptor,
isBrandCheck,
+ kDeferredStart,
kEmptyQueue,
kParkedAlgorithmResult,
kPendingPromise,
@@ -521,10 +522,11 @@ class ReadableStream {
[kTransfer]() {
if (!isReadableStream(this))
throw new ERR_INVALID_THIS('ReadableStream');
+ const transfer = readableStreamTransferState(this);
if (this.locked) {
- this[kState].transfer.port1?.close();
- this[kState].transfer.port1 = undefined;
- this[kState].transfer.port2 = undefined;
+ transfer.port1?.close();
+ transfer.port1 = undefined;
+ transfer.port2 = undefined;
throw new DOMException(
'Cannot transfer a locked ReadableStream',
'DataCloneError');
@@ -533,15 +535,13 @@ class ReadableStream {
const {
writable,
promise,
- } = lazyTransfer().newCrossRealmWritableSink(
- this,
- this[kState].transfer.port1);
+ } = lazyTransfer().newCrossRealmWritableSink(this, transfer.port1);
- this[kState].transfer.writable = writable;
- this[kState].transfer.promise = promise;
+ transfer.writable = writable;
+ transfer.promise = promise;
return {
- data: { port: this[kState].transfer.port2 },
+ data: { port: transfer.port2 },
deserializeInfo:
'internal/webstreams/readablestream:TransferredReadableStream',
};
@@ -549,8 +549,9 @@ class ReadableStream {
[kTransferList]() {
const { port1, port2 } = new MessageChannel();
- this[kState].transfer.port1 = port1;
- this[kState].transfer.port2 = port2;
+ const transfer = readableStreamTransferState(this);
+ transfer.port1 = port1;
+ transfer.port2 = port2;
return [ port2 ];
}
@@ -1415,10 +1416,15 @@ class ReadableStreamState {
state = 'readable';
storedError = undefined;
controller = undefined;
- transfer = new ReadableStreamTransferState();
+ // Materialized on transfer, see readableStreamTransferState().
+ transfer = undefined;
}
ObjectSetPrototypeOf(ReadableStreamState.prototype, null);
+function readableStreamTransferState(stream) {
+ return stream[kState].transfer ??= new ReadableStreamTransferState();
+}
+
function createReadableStreamState() {
return new ReadableStreamState();
}
@@ -2961,7 +2967,7 @@ function readableStreamDefaultControllerError(controller, error) {
function readableStreamDefaultControllerCancelSteps(controller, reason) {
resetQueue(controller);
- const result = controller[kState].cancelAlgorithm(reason);
+ const result = controller[kState].cancelAlgorithm(reason, controller);
readableStreamDefaultControllerClearAlgorithms(controller);
return result;
}
@@ -3019,10 +3025,16 @@ function setupReadableStreamDefaultController(
started: false,
sizeAlgorithm,
stream,
+ // The TransformStream this controller is the source of, if any; its
+ // shared source algorithms reach the stream through it.
+ transformStream: undefined,
};
stream[kState].controller = controller;
const startResult = startAlgorithm();
+ if (startResult === kDeferredStart)
+ return;
+
// A non-thenable start result guarantees fulfillment, and no .then
// lookup on it is observable.
const startFulfilled = startResult === null ||
@@ -3036,12 +3048,7 @@ function setupReadableStreamDefaultController(
return;
}
- const started = () => {
- controller[kState].started = true;
- assert(!controller[kState].pulling);
- assert(!controller[kState].pullAgain);
- readableStreamDefaultControllerCallPullIfNeeded(controller);
- };
+ const started = () => readableStreamDefaultControllerStarted(controller);
if (startFulfilled) {
// The post-start step runs at the exact microtask position the
@@ -3058,6 +3065,15 @@ function setupReadableStreamDefaultController(
(error) => readableStreamDefaultControllerError(controller, error));
}
+// The post-start steps; transform streams deliver them directly (see
+// transformStreamStart).
+function readableStreamDefaultControllerStarted(controller) {
+ controller[kState].started = true;
+ assert(!controller[kState].pulling);
+ assert(!controller[kState].pullAgain);
+ readableStreamDefaultControllerCallPullIfNeeded(controller);
+}
+
function setupReadableStreamDefaultControllerFromSource(
stream,
source,
@@ -4033,6 +4049,7 @@ module.exports = {
readableStreamDefaultControllerError,
readableStreamDefaultControllerCancelSteps,
readableStreamDefaultControllerPullSteps,
+ readableStreamDefaultControllerStarted,
setupReadableStreamDefaultController,
setupReadableStreamDefaultControllerFromSource,
readableByteStreamControllerClose,
diff --git a/lib/internal/webstreams/transformstream.js b/lib/internal/webstreams/transformstream.js
index 81f60b284f5..b8f9034e0ff 100644
--- a/lib/internal/webstreams/transformstream.js
+++ b/lib/internal/webstreams/transformstream.js
@@ -4,6 +4,7 @@ const {
FunctionPrototypeCall,
ObjectDefineProperties,
ObjectSetPrototypeOf,
+ Promise,
PromisePrototypeThen,
PromiseReject,
PromiseResolve,
@@ -48,6 +49,7 @@ const {
createPromiseCallback1Param,
createRawCallback2Params,
customInspect,
+ deferredStartAlgorithm,
extractHighWaterMark,
extractSizeAlgorithm,
getNonWritablePropertyDescriptor,
@@ -68,11 +70,14 @@ const {
readableStreamDefaultControllerError,
readableStreamDefaultControllerGetDesiredSize,
readableStreamDefaultControllerHasBackpressure,
+ readableStreamDefaultControllerStarted,
} = require('internal/webstreams/readablestream');
const {
createWritableStream,
writableStreamDefaultControllerErrorIfNeeded,
+ writableStreamDefaultControllerStarted,
+ writableStreamDefaultControllerStartFailed,
} = require('internal/webstreams/writablestream');
const assert = require('internal/assert');
@@ -159,14 +164,8 @@ class TransformStream {
extractHighWaterMark(writableHighWaterMark, 1);
const actualWritableSize = extractSizeAlgorithm(writableSize);
- // Without a start() the start promise is already resolved by the time
- // the readable and writable sides adopt it, so the shared resolved
- // promise stands in for the record.
- const startPromise = start !== undefined ? PromiseWithResolvers() : undefined;
-
initializeTransformStream(
this,
- startPromise,
actualWritableHighWaterMark,
actualWritableSize,
actualReadableHighWaterMark,
@@ -174,13 +173,11 @@ class TransformStream {
setupTransformStreamDefaultControllerFromTransformer(this, transformer);
- if (start !== undefined) {
- startPromise.resolve(
- FunctionPrototypeCall(
- start,
- transformer,
- this[kState].controller));
- }
+ transformStreamStart(
+ this,
+ start === undefined ?
+ undefined :
+ FunctionPrototypeCall(start, transformer, this[kState].controller));
}
/**
@@ -380,38 +377,35 @@ function defaultTransformAlgorithm(chunk, controller) {
transformStreamDefaultControllerEnqueue(controller, chunk);
}
-function resolvedStartAlgorithm() {
- return kResolvedPromise;
-}
-
function initializeTransformStream(
stream,
- startPromise,
writableHighWaterMark,
writableSizeAlgorithm,
readableHighWaterMark,
readableSizeAlgorithm) {
- const startAlgorithm = startPromise === undefined ?
- resolvedStartAlgorithm :
- () => startPromise.promise;
-
+ // The sink and source algorithms are shared functions that reach the
+ // transform stream through their controller, and the post-start step
+ // is delivered by transformStreamStart, so no per-stream closures or
+ // start promises are allocated here.
const writable = createWritableStream(
- startAlgorithm,
- (chunk) => transformStreamDefaultSinkWriteAlgorithm(stream, chunk),
- () => transformStreamDefaultSinkCloseAlgorithm(stream),
- (reason) => transformStreamDefaultSinkAbortAlgorithm(stream, reason),
+ deferredStartAlgorithm,
+ transformStreamDefaultSinkWriteAlgorithm,
+ transformStreamDefaultSinkCloseAlgorithm,
+ transformStreamDefaultSinkAbortAlgorithm,
writableHighWaterMark,
writableSizeAlgorithm,
);
+ writable[kState].controller[kState].transformStream = stream;
const readable = createReadableStream(
- startAlgorithm,
- () => transformStreamDefaultSourcePullAlgorithm(stream),
- (reason) => transformStreamDefaultSourceCancelAlgorithm(stream, reason),
+ deferredStartAlgorithm,
+ transformStreamDefaultSourcePullAlgorithm,
+ transformStreamDefaultSourceCancelAlgorithm,
readableHighWaterMark,
readableSizeAlgorithm,
);
+ readable[kState].controller[kState].transformStream = stream;
const state = new TransformStreamState();
state.readable = readable;
@@ -421,6 +415,43 @@ function initializeTransformStream(
transformStreamSetBackpressure(stream, true);
}
+// Delivers the post-start step of both sides. The spec resolves the start
+// promise with the transformer's start result at the end of construction
+// and each side then adopts it through a wrapper promise, so with a
+// non-thenable start result the started steps run three microtasks after
+// construction, and with a thenable one two microtasks after the adopting
+// promise settles. The same hops are taken as reactions on one shared
+// promise instead of a promise record plus two wrapper promises.
+function transformStreamStart(stream, startResult) {
+ const state = stream[kState];
+ const started = () => {
+ writableStreamDefaultControllerStarted(state.writable[kState].controller);
+ readableStreamDefaultControllerStarted(state.readable[kState].controller);
+ };
+ if (startResult === null ||
+ (typeof startResult !== 'object' && typeof startResult !== 'function')) {
+ PromisePrototypeThen(
+ PromisePrototypeThen(
+ PromisePrototypeThen(kResolvedPromise, undefined, undefined),
+ undefined,
+ undefined),
+ started);
+ return;
+ }
+ const startPromise = new Promise((resolve) => resolve(startResult));
+ PromisePrototypeThen(
+ PromisePrototypeThen(startPromise, undefined, undefined),
+ started,
+ (error) => {
+ writableStreamDefaultControllerStartFailed(
+ state.writable[kState].controller,
+ error);
+ readableStreamDefaultControllerError(
+ state.readable[kState].controller,
+ error);
+ });
+}
+
function transformStreamError(stream, error) {
const {
readable,
@@ -609,7 +640,8 @@ function transformStreamDefaultControllerTerminate(controller) {
new ERR_INVALID_STATE.TypeError('TransformStream has been terminated'));
}
-function transformStreamDefaultSinkWriteAlgorithm(stream, chunk) {
+function transformStreamDefaultSinkWriteAlgorithm(chunk, writableController) {
+ const stream = writableController[kState].transformStream;
const state = stream[kState];
const {
writable,
@@ -656,7 +688,10 @@ function transformStreamDefaultSinkWriteAlgorithm(stream, chunk) {
return transformStreamDefaultControllerPerformTransform(controller, chunk);
}
-async function transformStreamDefaultSinkAbortAlgorithm(stream, reason) {
+async function transformStreamDefaultSinkAbortAlgorithm(
+ reason,
+ writableController) {
+ const stream = writableController[kState].transformStream;
const {
controller,
readable,
@@ -690,7 +725,8 @@ async function transformStreamDefaultSinkAbortAlgorithm(stream, reason) {
return controller[kState].finishPromise;
}
-function transformStreamDefaultSinkCloseAlgorithm(stream) {
+function transformStreamDefaultSinkCloseAlgorithm(writableController) {
+ const stream = writableController[kState].transformStream;
const {
readable,
controller,
@@ -720,7 +756,8 @@ function transformStreamDefaultSinkCloseAlgorithm(stream) {
return controller[kState].finishPromise;
}
-function transformStreamDefaultSourcePullAlgorithm(stream) {
+function transformStreamDefaultSourcePullAlgorithm(readableController) {
+ const stream = readableController[kState].transformStream;
const state = stream[kState];
assert(state.backpressure);
transformStreamSetBackpressure(stream, false);
@@ -732,7 +769,10 @@ function transformStreamDefaultSourcePullAlgorithm(stream) {
return kParkedAlgorithmResult;
}
-function transformStreamDefaultSourceCancelAlgorithm(stream, reason) {
+function transformStreamDefaultSourceCancelAlgorithm(
+ reason,
+ readableController) {
+ const stream = readableController[kState].transformStream;
const {
controller,
writable,
diff --git a/lib/internal/webstreams/util.js b/lib/internal/webstreams/util.js
index 581e3540c68..bc6818c11ac 100644
--- a/lib/internal/webstreams/util.js
+++ b/lib/internal/webstreams/util.js
@@ -357,6 +357,15 @@ const kResolvedPromise = PromiseResolve();
// (see the transform stream source pull algorithm).
const kParkedAlgorithmResult = Symbol('kParkedAlgorithmResult');
+// Start algorithm of internal streams whose creator delivers the
+// post-start step itself (see transformStreamStart): the controller setup
+// leaves `started` unset when the start result is this sentinel.
+const kDeferredStart = Symbol('kDeferredStart');
+
+function deferredStartAlgorithm() {
+ return kDeferredStart;
+}
+
// Wires the (possibly non-thenable) result of an underlying algorithm
// callback to its fulfilled/rejected continuations. A non-thenable result
// means fulfillment is guaranteed and no then() lookup is observable, so
@@ -461,6 +470,7 @@ module.exports = {
createRawCallback2Params,
customInspect,
defaultSizeAlgorithm,
+ deferredStartAlgorithm,
dequeueValue,
enqueueValueWithSize,
extractHighWaterMark,
@@ -468,6 +478,7 @@ module.exports = {
getNonWritablePropertyDescriptor,
isBrandCheck,
isPromisePending,
+ kDeferredStart,
kEmptyQueue,
kParkedAlgorithmResult,
kPendingPromise,
diff --git a/lib/internal/webstreams/writablestream.js b/lib/internal/webstreams/writablestream.js
index 31f2a1f875d..7b6a7327944 100644
--- a/lib/internal/webstreams/writablestream.js
+++ b/lib/internal/webstreams/writablestream.js
@@ -66,6 +66,7 @@ const {
getNonWritablePropertyDescriptor,
isBrandCheck,
isPromisePending,
+ kDeferredStart,
kEmptyQueue,
kState,
kType,
@@ -303,10 +304,11 @@ class WritableStream {
[kTransfer]() {
if (!isWritableStream(this))
throw new ERR_INVALID_THIS('WritableStream');
+ const transfer = writableStreamTransferState(this);
if (this.locked) {
- this[kState].transfer.port1?.close();
- this[kState].transfer.port1 = undefined;
- this[kState].transfer.port2 = undefined;
+ transfer.port1?.close();
+ transfer.port1 = undefined;
+ transfer.port2 = undefined;
throw new DOMException(
'Cannot transfer a locked WritableStream',
'DataCloneError');
@@ -315,15 +317,13 @@ class WritableStream {
const {
readable,
promise,
- } = lazyTransfer().newCrossRealmReadableStream(
- this,
- this[kState].transfer.port1);
+ } = lazyTransfer().newCrossRealmReadableStream(this, transfer.port1);
- this[kState].transfer.readable = readable;
- this[kState].transfer.promise = promise;
+ transfer.readable = readable;
+ transfer.promise = promise;
return {
- data: { port: this[kState].transfer.port2 },
+ data: { port: transfer.port2 },
deserializeInfo:
'internal/webstreams/writablestream:TransferredWritableStream',
};
@@ -331,8 +331,9 @@ class WritableStream {
[kTransferList]() {
const { port1, port2 } = new MessageChannel();
- this[kState].transfer.port1 = port1;
- this[kState].transfer.port2 = port2;
+ const transfer = writableStreamTransferState(this);
+ transfer.port1 = port1;
+ transfer.port2 = port2;
return [ port2 ];
}
@@ -511,7 +512,7 @@ class WritableStreamDefaultController {
}
[kAbort](reason) {
- const result = this[kState].abortAlgorithm(reason);
+ const result = this[kState].abortAlgorithm(reason, this);
writableStreamDefaultControllerClearAlgorithms(this);
return result;
}
@@ -605,10 +606,15 @@ class WritableStreamState {
// no request storage.
writeRequests = kEmptyQueue;
writer = undefined;
- transfer = new WritableStreamTransferState();
+ // Materialized on transfer, see writableStreamTransferState().
+ transfer = undefined;
}
ObjectSetPrototypeOf(WritableStreamState.prototype, null);
+function writableStreamTransferState(stream) {
+ return stream[kState].transfer ??= new WritableStreamTransferState();
+}
+
function createWritableStreamState() {
return new WritableStreamState();
}
@@ -1219,7 +1225,7 @@ function writableStreamDefaultControllerProcessClose(controller) {
writableStreamMarkCloseRequestInFlight(stream);
dequeueValue(controller);
assert(!queue.length);
- const sinkClosePromise = closeAlgorithm();
+ const sinkClosePromise = closeAlgorithm(controller);
writableStreamDefaultControllerClearAlgorithms(controller);
PromisePrototypeThen(
sinkClosePromise,
@@ -1370,6 +1376,9 @@ function setupWritableStreamDefaultController(
sizeAlgorithm,
started: false,
stream,
+ // The TransformStream this controller is the sink of, if any; its
+ // shared sink algorithms reach the stream through it.
+ transformStream: undefined,
writeAlgorithm,
writeFulfilled: undefined,
writeRejected: undefined,
@@ -1379,13 +1388,10 @@ function setupWritableStreamDefaultController(
writableStreamUpdateBackpressure(controller, stream[kState]);
const startResult = startAlgorithm();
+ if (startResult === kDeferredStart)
+ return;
- const started = () => {
- assert(stream[kState].state === 'writable' ||
- stream[kState].state === 'erroring');
- controller[kState].started = true;
- writableStreamDefaultControllerAdvanceQueueIfNeeded(controller);
- };
+ const started = () => writableStreamDefaultControllerStarted(controller);
if (startResult === null ||
(typeof startResult !== 'object' && typeof startResult !== 'function')) {
@@ -1401,12 +1407,25 @@ function setupWritableStreamDefaultController(
PromisePrototypeThen(
new Promise((r) => r(startResult)),
started,
- (error) => {
- assert(stream[kState].state === 'writable' ||
- stream[kState].state === 'erroring');
- controller[kState].started = true;
- writableStreamDealWithRejection(stream, error);
- });
+ (error) => writableStreamDefaultControllerStartFailed(controller, error));
+}
+
+// The post-start steps; transform streams deliver them directly (see
+// transformStreamStart).
+function writableStreamDefaultControllerStarted(controller) {
+ const { stream } = controller[kState];
+ assert(stream[kState].state === 'writable' ||
+ stream[kState].state === 'erroring');
+ controller[kState].started = true;
+ writableStreamDefaultControllerAdvanceQueueIfNeeded(controller);
+}
+
+function writableStreamDefaultControllerStartFailed(controller, error) {
+ const { stream } = controller[kState];
+ assert(stream[kState].state === 'writable' ||
+ stream[kState].state === 'erroring');
+ controller[kState].started = true;
+ writableStreamDealWithRejection(stream, error);
}
module.exports = {
@@ -1459,6 +1478,8 @@ module.exports = {
writableStreamDefaultControllerClose,
writableStreamDefaultControllerClearAlgorithms,
writableStreamDefaultControllerAdvanceQueueIfNeeded,
+ writableStreamDefaultControllerStarted,
+ writableStreamDefaultControllerStartFailed,
setupWritableStreamDefaultControllerFromSink,
setupWritableStreamDefaultController,
createWritableStream,