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>

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,

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,

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,

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,