stream: reduce allocations on WHATWG streams hot paths

Pure-JavaScript optimizations to lib/internal/webstreams/*:

- Reuse the pull and write promise-reaction closures per controller
  instead of allocating two fresh closures per chunk. They are created
  lazily on the first pull/write so construction-only workloads never
  pay for them.
- Add the buffered fast path to the ReadableStream async iterator,
  mirroring the one ReadableStreamDefaultReader.read() already has:
  when data is queued in a default controller, dequeue directly and
  skip the read request, the deferred promise, and the promise
  chaining on the following call.
- Run the post-start step through queueMicrotask() when the start
  algorithm result is not an object (fulfillment is guaranteed and no
  .then lookup is observable), instead of wrapping it in
  new Promise((r) => r(result)) plus a two-closure reaction. Object
  and thenable results keep the promise path since their adoption
  timing and .then lookups are observable.
- Specialize the promise-callback wrappers for user algorithms by
  arity (0/1/2), replacing the rest-parameter + ReflectApply form that
  allocated an arguments array per invocation. The exact number of
  arguments each user callback observes is preserved.
- Share immutable nil records for the writable stream closeRequest,
  inFlightWriteRequest, inFlightCloseRequest and pendingAbortRequest
  resets; these records are only ever replaced wholesale. Push the
  PromiseWithResolvers() record directly as the write request rather
  than rebuilding an identical object.
- Remove dead per-instance allocations: the never-read close record in
  the writable stream state, the placeholder close/ready records that
  reader/writer setup unconditionally replaces, the per-stream
  () => 1 size algorithm closures, and the kControllerErrorFunction
  placeholder plus bound function (now a prototype method; byte
  streams keep their historical no-op behavior there).

Benchmark results vs main, same-day builds, benchmark/compare.js
--runs 10 (statistically significant rows):

  creation WritableStream (n=500k)              +97.1% ***
  creation TransformStream (n=500k)             +50.3% ***
  creation ReadableStream (n=500k)              +13.9% ***
  creation ReadableStreamBYOBReader (n=500k)    +13.0% ***
  creation ReadableStreamDefaultReader (n=500k)  +9.8% **
  readable-async-iterator                       +38.1% ***
  pipe-to (16 hwm configs)                +2.6..+4.5% (all positive)
  js_transfer WS / TS / RS              +7.7% / +6.3% / +3.0%
  readable-read normal/byob, read-buffered      parity (n.s.)

WPT streams/compression/encoding results are identical to main
(1403/338/3822 subtests passed, same 8 expected failures by name),
and all webstreams-related parallel tests pass.

Assisted-by: Claude Fable 5 <noreply@anthropic.com>

Author: mcollina

lib/internal/webstreams/readablestream.js

@@ -104,8 +104,9 @@ const {
   canCopyArrayBuffer,
   cloneAsUint8Array,
   copyArrayBuffer,
-  createPromiseCallback,
+  createPromiseCallback1,
   customInspect,
+  defaultSizeAlgorithm,
   dequeueValue,
   enqueueValueWithSize,
   extractHighWaterMark,
@@ -252,7 +253,6 @@ class ReadableStream {
     this[kState] = createReadableStreamState();
 
     this[kIsClosedPromise] = PromiseWithResolvers();
-    this[kControllerErrorFunction] = () => {};
 
     // The spec requires handling of the strategy first
     // here. Specifically, if getting the size and
@@ -294,6 +294,15 @@ class ReadableStream {
     return this[kState].state === 'readable';
   }
 
+  [kControllerErrorFunction](error) {
+    // Used by the internal stream interop (addAbortSignal). Historically
+    // only default controllers were wired here; byte stream controllers
+    // keep the previous no-op behavior.
+    const controller = this[kState].controller;
+    if (isReadableStreamDefaultController(controller))
+      controller.error(error);
+  }
+
   /**
    * @readonly
    * @type {boolean}
@@ -535,9 +544,38 @@ class ReadableStream {
           state.current = PromiseResolve();
           started = true;
         }
-        state.current = state.current !== undefined ?
-          PromisePrototypeThen(state.current, nextSteps, nextSteps) :
-          nextSteps();
+        if (state.current !== undefined) {
+          state.current =
+            PromisePrototypeThen(state.current, nextSteps, nextSteps);
+          return state.current;
+        }
+        // No read is in flight. Mirror the buffered fast path of
+        // ReadableStreamDefaultReader.read(): when data is already queued
+        // in a default controller, resolve immediately without allocating
+        // a read request. The result settles synchronously, so leaving
+        // state.current undefined matches the state the slow path reaches
+        // once its read request callbacks have settled.
+        const stream = reader[kState].stream;
+        if (!state.done && stream !== undefined) {
+          const controller = stream[kState].controller;
+          if (stream[kState].state === 'readable' &&
+              isReadableStreamDefaultController(controller) &&
+              controller[kState].queue.length > 0) {
+            stream[kState].disturbed = true;
+            const chunk = dequeueValue(controller);
+
+            if (controller[kState].closeRequested &&
+                !controller[kState].queue.length) {
+              readableStreamDefaultControllerClearAlgorithms(controller);
+              readableStreamClose(stream);
+            } else {
+              readableStreamDefaultControllerCallPullIfNeeded(controller);
+            }
+
+            return PromiseResolve({ done: false, value: chunk });
+          }
+        }
+        state.current = nextSteps();
         return state.current;
       },
 
@@ -611,7 +649,7 @@ class ReadableStream {
       // lingering promise not being properly resolved.
       // https://github.com/nodejs/node/issues/51486
       new transfer.CrossRealmTransformReadableSource(port, true),
-      0, () => 1);
+      0, defaultSizeAlgorithm);
   }
 }
 
@@ -835,11 +873,8 @@ class ReadableStreamDefaultReader {
     this[kState] = {
       readRequests: [],
       stream: undefined,
-      close: {
-        promise: undefined,
-        resolve: undefined,
-        reject: undefined,
-      },
+      // The record is unconditionally replaced during setup.
+      close: undefined,
     };
     setupReadableStreamDefaultReader(this, stream);
   }
@@ -949,11 +984,8 @@ class ReadableStreamBYOBReader {
     this[kState] = {
       stream: undefined,
       readIntoRequests: [],
-      close: {
-        promise: undefined,
-        resolve: undefined,
-        reject: undefined,
-      },
+      // The record is unconditionally replaced during setup.
+      close: undefined,
     };
     setupReadableStreamBYOBReader(this, stream);
   }
@@ -1287,7 +1319,7 @@ function InternalReadableStream(start, pull, cancel, highWaterMark, size) {
 ObjectSetPrototypeOf(InternalReadableStream.prototype, ReadableStream.prototype);
 ObjectSetPrototypeOf(InternalReadableStream, ReadableStream);
 
-function createReadableStream(start, pull, cancel, highWaterMark = 1, size = () => 1) {
+function createReadableStream(start, pull, cancel, highWaterMark = 1, size = defaultSizeAlgorithm) {
   const stream = new InternalReadableStream(start, pull, cancel, highWaterMark, size);
 
   // For spec compliance the InternalReadableStream must be a ReadableStream
@@ -2475,16 +2507,24 @@ function readableStreamDefaultControllerCallPullIfNeeded(controller) {
   }
   assert(!controller[kState].pullAgain);
   controller[kState].pulling = true;
-  PromisePrototypeThen(
-    controller[kState].pullAlgorithm(controller),
-    () => {
+  if (controller[kState].pullFulfilled === undefined) {
+    // The pull reaction closures only capture the controller, so they are
+    // created once on the first pull and reused for every subsequent pull
+    // instead of allocating two fresh closures per chunk.
+    controller[kState].pullFulfilled = () => {
       controller[kState].pulling = false;
       if (controller[kState].pullAgain) {
         controller[kState].pullAgain = false;
         readableStreamDefaultControllerCallPullIfNeeded(controller);
       }
-    },
-    (error) => readableStreamDefaultControllerError(controller, error));
+    };
+    controller[kState].pullRejected =
+      (error) => readableStreamDefaultControllerError(controller, error);
+  }
+  PromisePrototypeThen(
+    controller[kState].pullAlgorithm(controller),
+    controller[kState].pullFulfilled,
+    controller[kState].pullRejected);
 }
 
 function readableStreamDefaultControllerClearAlgorithms(controller) {
@@ -2547,17 +2587,33 @@ function setupReadableStreamDefaultController(
     pullAgain: false,
     pullAlgorithm,
     pulling: false,
+    pullFulfilled: undefined,
+    pullRejected: undefined,
     queue: [],
     queueTotalSize: 0,
     started: false,
     sizeAlgorithm,
     stream,
   };
   stream[kState].controller = controller;
-  stream[kControllerErrorFunction] = FunctionPrototypeBind(controller.error, controller);
 
   const startResult = startAlgorithm();
 
+  if (startResult === null ||
+      (typeof startResult !== 'object' && typeof startResult !== 'function')) {
+    // Non-thenable start result: fulfillment is guaranteed and no .then
+    // lookup on the result is observable, so run the post-start step
+    // directly at the exact microtask position the promise reaction
+    // would have had, skipping two promise allocations.
+    queueMicrotask(() => {
+      controller[kState].started = true;
+      assert(!controller[kState].pulling);
+      assert(!controller[kState].pullAgain);
+      readableStreamDefaultControllerCallPullIfNeeded(controller);
+    });
+    return;
+  }
+
   PromisePrototypeThen(
     new Promise((r) => r(startResult)),
     () => {
@@ -2582,10 +2638,10 @@ function setupReadableStreamDefaultControllerFromSource(
     FunctionPrototypeBind(start, source, controller) :
     nonOpStart;
   const pullAlgorithm = pull ?
-    createPromiseCallback('source.pull', pull, source) :
+    createPromiseCallback1('source.pull', pull, source) :
     nonOpPull;
   const cancelAlgorithm = cancel ?
-    createPromiseCallback('source.cancel', cancel, source) :
+    createPromiseCallback1('source.cancel', cancel, source) :
     nonOpCancel;
 
   setupReadableStreamDefaultController(
@@ -3236,16 +3292,23 @@ function readableByteStreamControllerCallPullIfNeeded(controller) {
   }
   assert(!controller[kState].pullAgain);
   controller[kState].pulling = true;
-  PromisePrototypeThen(
-    controller[kState].pullAlgorithm(controller),
-    () => {
+  if (controller[kState].pullFulfilled === undefined) {
+    // See readableStreamDefaultControllerCallPullIfNeeded: created once,
+    // reused for every pull.
+    controller[kState].pullFulfilled = () => {
       controller[kState].pulling = false;
       if (controller[kState].pullAgain) {
         controller[kState].pullAgain = false;
         readableByteStreamControllerCallPullIfNeeded(controller);
       }
-    },
-    (error) => readableByteStreamControllerError(controller, error));
+    };
+    controller[kState].pullRejected =
+      (error) => readableByteStreamControllerError(controller, error);
+  }
+  PromisePrototypeThen(
+    controller[kState].pullAlgorithm(controller),
+    controller[kState].pullFulfilled,
+    controller[kState].pullRejected);
 }
 
 function readableByteStreamControllerError(controller, error) {
@@ -3367,6 +3430,8 @@ function setupReadableByteStreamController(
     closeRequested: false,
     pullAgain: false,
     pulling: false,
+    pullFulfilled: undefined,
+    pullRejected: undefined,
     started: false,
     stream,
     queue: [],
@@ -3381,6 +3446,18 @@ function setupReadableByteStreamController(
 
   const startResult = startAlgorithm();
 
+  if (startResult === null ||
+      (typeof startResult !== 'object' && typeof startResult !== 'function')) {
+    // See setupReadableStreamDefaultController.
+    queueMicrotask(() => {
+      controller[kState].started = true;
+      assert(!controller[kState].pulling);
+      assert(!controller[kState].pullAgain);
+      readableByteStreamControllerCallPullIfNeeded(controller);
+    });
+    return;
+  }
+
   PromisePrototypeThen(
     new Promise((r) => r(startResult)),
     () => {
@@ -3405,10 +3482,10 @@ function setupReadableByteStreamControllerFromSource(
     FunctionPrototypeBind(start, source, controller) :
     nonOpStart;
   const pullAlgorithm = pull ?
-    createPromiseCallback('source.pull', pull, source, controller) :
+    createPromiseCallback1('source.pull', pull, source) :
     nonOpPull;
   const cancelAlgorithm = cancel ?
-    createPromiseCallback('source.cancel', cancel, source) :
+    createPromiseCallback1('source.cancel', cancel, source) :
     nonOpCancel;
 
   if (autoAllocateChunkSize === 0) {

lib/internal/webstreams/transformstream.js

@@ -43,7 +43,8 @@ const {
 } = require('internal/worker/js_transferable');
 
 const {
-  createPromiseCallback,
+  createPromiseCallback1,
+  createPromiseCallback2,
   customInspect,
   extractHighWaterMark,
   extractSizeAlgorithm,
@@ -462,13 +463,13 @@ function setupTransformStreamDefaultControllerFromTransformer(
   const flush = transformer?.flush;
   const cancel = transformer?.cancel;
   const transformAlgorithm = transform ?
-    createPromiseCallback('transformer.transform', transform, transformer) :
+    createPromiseCallback2('transformer.transform', transform, transformer) :
     defaultTransformAlgorithm;
   const flushAlgorithm = flush ?
-    createPromiseCallback('transformer.flush', flush, transformer) :
+    createPromiseCallback1('transformer.flush', flush, transformer) :
     nonOpFlush;
   const cancelAlgorithm = cancel ?
-    createPromiseCallback('transformer.cancel', cancel, transformer) :
+    createPromiseCallback1('transformer.cancel', cancel, transformer) :
     nonOpCancel;
 
   setupTransformStreamDefaultController(

lib/internal/webstreams/util.js

@@ -7,10 +7,10 @@ const {
   ArrayPrototypePush,
   ArrayPrototypeShift,
   AsyncIteratorPrototype,
+  FunctionPrototypeCall,
   MathMax,
   NumberIsNaN,
   PromisePrototypeThen,
-  ReflectApply,
   ReflectGet,
   Symbol,
   Uint8Array,
@@ -69,8 +69,12 @@ function extractHighWaterMark(value, defaultHWM) {
   return coercedValue;
 }
 
+// The default size algorithm is never exposed to user code, so a single
+// shared function avoids one closure allocation per stream.
+const defaultSizeAlgorithm = () => 1;
+
 function extractSizeAlgorithm(size) {
-  if (size === undefined) return () => 1;
+  if (size === undefined) return defaultSizeAlgorithm;
   validateFunction(size, 'strategy.size');
   return size;
 }
@@ -169,9 +173,25 @@ function enqueueValueWithSize(controller, value, size) {
   controller[kState].queueTotalSize += size;
 }
 
-function createPromiseCallback(name, fn, thisArg) {
+// Arity-specialized variants of the promise-callback wrapper. The generic
+// rest-parameter + ReflectApply form allocated an arguments array on every
+// invocation; these run on per-chunk hot paths (pull/write/transform), so
+// each known call-site arity gets its own wrapper. The exact number of
+// arguments passed through to the user callback is observable and must be
+// preserved.
+function createPromiseCallback0(name, fn, thisArg) {
+  validateFunction(fn, name);
+  return async () => FunctionPrototypeCall(fn, thisArg);
+}
+
+function createPromiseCallback1(name, fn, thisArg) {
+  validateFunction(fn, name);
+  return async (arg) => FunctionPrototypeCall(fn, thisArg, arg);
+}
+
+function createPromiseCallback2(name, fn, thisArg) {
   validateFunction(fn, name);
-  return async (...args) => ReflectApply(fn, thisArg, args);
+  return async (arg1, arg2) => FunctionPrototypeCall(fn, thisArg, arg1, arg2);
 }
 
 function isPromisePending(promise) {
@@ -213,8 +233,11 @@ module.exports = {
   canCopyArrayBuffer,
   cloneAsUint8Array,
   copyArrayBuffer,
-  createPromiseCallback,
+  createPromiseCallback0,
+  createPromiseCallback1,
+  createPromiseCallback2,
   customInspect,
+  defaultSizeAlgorithm,
   dequeueValue,
   enqueueValueWithSize,
   extractHighWaterMark,