stream: decouple transform backpressure changes

benchmark/webstreams/pipe-through.js

@@ -0,0 +1,38 @@
+'use strict';
+const common = require('../common.js');
+const {
+  ReadableStream,
+  TransformStream,
+} = require('node:stream/web');
+
+const bench = common.createBenchmark(main, {
+  n: [5e5],
+  kind: ['default', 'transform'],
+});
+
+async function main({ n, kind }) {
+  const b = Buffer.alloc(64);
+  let i = 0;
+  const rs = new ReadableStream({
+    pull(controller) {
+      if (i++ < n) {
+        controller.enqueue(b);
+      } else {
+        controller.close();
+      }
+    },
+  });
+  const ts = kind === 'default' ?
+    new TransformStream() :
+    new TransformStream({
+      transform(chunk, controller) { controller.enqueue(chunk); },
+    });
+
+  const reader = rs.pipeThrough(ts).getReader();
+  bench.start();
+  for (;;) {
+    const { done } = await reader.read();
+    if (done) break;
+  }
+  bench.end(n);
+}

benchmark/webstreams/pipe-to.js

@@ -7,29 +7,27 @@ const {
 
 const bench = common.createBenchmark(main, {
   n: [5e5],
-  highWaterMarkR: [512, 1024, 2048, 4096],
-  highWaterMarkW: [512, 1024, 2048, 4096],
+  highWaterMarkR: [1, 1024, 4096],
+  highWaterMarkW: [1, 1024, 4096],
 });
 
 
 async function main({ n, highWaterMarkR, highWaterMarkW }) {
   const b = Buffer.alloc(1024);
   let i = 0;
   const rs = new ReadableStream({
-    highWaterMark: highWaterMarkR,
     pull: function(controller) {
       if (i++ < n) {
         controller.enqueue(b);
       } else {
         controller.close();
       }
     },
-  });
+  }, { highWaterMark: highWaterMarkR });
   const ws = new WritableStream({
-    highWaterMark: highWaterMarkW,
     write(chunk, controller) {},
     close() { bench.end(n); },
-  });
+  }, { highWaterMark: highWaterMarkW });
 
   bench.start();
   rs.pipeTo(ws);

lib/internal/webstreams/readablestream.js

@@ -101,6 +101,7 @@ const {
   cloneAsUint8Array,
   copyArrayBuffer,
   createPromiseCallback1Param,
+  createRawCallback1Param,
   customInspect,
   defaultSizeAlgorithm,
   dequeueValue,
@@ -110,17 +111,18 @@ const {
   getNonWritablePropertyDescriptor,
   isBrandCheck,
   kEmptyQueue,
+  kResolvedPromise,
   kState,
   kType,
   lazyTransfer,
   materializeQueue,
+  nonOpCallback,
   nonOpCancel,
-  nonOpPull,
-  nonOpStart,
   rejectedHandledRecord,
   resetQueue,
   resolvedRecord,
   setPromiseHandled,
+  thenAlgorithmResult,
 } = require('internal/webstreams/util');
 
 const {
@@ -137,7 +139,6 @@ const {
   writableStreamDefaultWriterRelease,
   writableStreamDefaultWriterWriteWithRequest,
   writerClosedPromise,
-  writerReadyPromise,
 } = require('internal/webstreams/writablestream');
 
 const { Buffer } = require('buffer');
@@ -1455,7 +1456,7 @@ function readableStreamFromIterable(iterable) {
   if (iterator === null || (typeof iterator !== 'object' && typeof iterator !== 'function')) {
     throw new ERR_INVALID_STATE.TypeError('The iterator method must return an object');
   }
-  const startAlgorithm = nonOpStart;
+  const startAlgorithm = nonOpCallback;
 
   async function pullAlgorithm() {
     const iterResult = await iterator.next();
@@ -1674,11 +1675,31 @@ function readableStreamPipeTo(
   // the chunk travels through `pendingChunk`.
   let pendingChunk;
   let readRequest;
+  let readyHook;
 
   // Ready promise rejection is handled by the destination-errored
   // watcher.
   function ignoreReadyRejection() {}
 
+  // Parks the pump on the destination's backpressure by installing a
+  // record that duck-types the writer's lazily-materialized
+  // [[readyPromise]] record: writableStreamUpdateBackpressure resolves it
+  // when backpressure clears (after publishing the new backpressure
+  // state), which re-enters the pump directly instead of rotating a
+  // fresh promise record plus reaction per flip. The pipe holds the only
+  // reference to the writer, so the record is never observable as a real
+  // ready promise; the erroring/release paths probe `promise` via
+  // isPromisePending() and call `reject`, so it carries a real
+  // forever-pending promise and a no-op reject.
+  function parkOnReady() {
+    readyHook ??= {
+      promise: PromiseWithResolvers().promise,
+      resolve: pump,
+      reject: ignoreReadyRejection,
+    };
+    writer[kState].ready = readyHook;
+  }
+
   function forwardChunk() {
     const chunk = pendingChunk;
     pendingChunk = undefined;
@@ -1690,10 +1711,7 @@ function readableStreamPipeTo(
     if (shuttingDown) return;
 
     if (dest[kState].backpressure) {
-      PromisePrototypeThen(
-        writerReadyPromise(writer).promise,
-        pump,
-        ignoreReadyRejection);
+      parkOnReady();
       return;
     }
 
@@ -1738,9 +1756,18 @@ function readableStreamPipeTo(
         return;
       }
 
-      // Yield to microtask queue between batches to allow events/signals
-      // to fire
-      queueMicrotask(pump);
+      // Park on backpressure directly: the ready hook resumes the pump
+      // when a completed write clears it.
+      if (dest[kState].backpressure) {
+        parkOnReady();
+        return;
+      }
+
+      // Yield to the microtask queue between batches so completed-write
+      // reactions and events/signals fire; a shared resolved promise
+      // enqueues the continuation at the same position as queueMicrotask
+      // without the per-batch scheduling overhead.
+      PromisePrototypeThen(kResolvedPromise, pump);
       return;
     }
 
@@ -1752,7 +1779,7 @@ function readableStreamPipeTo(
         // synchronous write during enqueue(). See WHATWG Streams spec
         // "ReadableStreamPipeTo" step 15's "chunk steps".
         pendingChunk = chunk;
-        queueMicrotask(forwardChunk);
+        PromisePrototypeThen(kResolvedPromise, forwardChunk);
       },
       [kClose]() {},
       [kError]() {},
@@ -1867,7 +1894,7 @@ function readableStreamDefaultTee(stream, cloneForBranch2) {
         // The microtask is required by the spec (ReadableStreamTee's
         // "chunk steps" queue one).
         pendingChunk = value;
-        queueMicrotask(forwardChunk);
+        PromisePrototypeThen(kResolvedPromise, forwardChunk);
       },
       [kClose]() {
         // The `process.nextTick()` is not part of the spec.
@@ -1911,9 +1938,9 @@ function readableStreamDefaultTee(stream, cloneForBranch2) {
   }
 
   branch1 =
-    createReadableStream(nonOpStart, pullAlgorithm, cancel1Algorithm);
+    createReadableStream(nonOpCallback, pullAlgorithm, cancel1Algorithm);
   branch2 =
-    createReadableStream(nonOpStart, pullAlgorithm, cancel2Algorithm);
+    createReadableStream(nonOpCallback, pullAlgorithm, cancel2Algorithm);
 
   PromisePrototypeThen(
     readerClosedPromise(reader).promise,
@@ -2020,7 +2047,7 @@ function readableByteStreamTee(stream) {
     defaultReadRequest ??= {
       [kChunk](chunk) {
         pendingChunk = chunk;
-        queueMicrotask(forwardChunk);
+        PromisePrototypeThen(kResolvedPromise, forwardChunk);
       },
       [kClose]() {
         reading = false;
@@ -2199,9 +2226,9 @@ function readableByteStreamTee(stream) {
   }
 
   branch1 =
-    createReadableByteStream(nonOpStart, pull1Algorithm, cancel1Algorithm);
+    createReadableByteStream(nonOpCallback, pull1Algorithm, cancel1Algorithm);
   branch2 =
-    createReadableByteStream(nonOpStart, pull2Algorithm, cancel2Algorithm);
+    createReadableByteStream(nonOpCallback, pull2Algorithm, cancel2Algorithm);
 
   forwardReaderError(reader);
 
@@ -2709,8 +2736,18 @@ function readableStreamDefaultControllerPull(controller) {
     controller[kState].pullRejected =
       (error) => readableStreamDefaultControllerError(controller, error);
   }
-  PromisePrototypeThen(
-    controller[kState].pullAlgorithm(controller),
+  // The pull algorithm may be a raw callback (a wrapped user source.pull
+  // returns its result uncoerced; a synchronous throw surfaces here) or an
+  // internal algorithm that always returns a promise; thenAlgorithmResult
+  // handles both.
+  let result;
+  try {
+    result = controller[kState].pullAlgorithm(controller);
+  } catch (error) {
+    result = PromiseReject(error);
+  }
+  thenAlgorithmResult(
+    result,
     controller[kState].pullFulfilled,
     controller[kState].pullRejected);
 }
@@ -2834,10 +2871,10 @@ function setupReadableStreamDefaultControllerFromSource(
   const cancel = source?.cancel;
   const startAlgorithm = start ?
     FunctionPrototypeBind(start, source, controller) :
-    nonOpStart;
+    nonOpCallback;
   const pullAlgorithm = pull ?
-    createPromiseCallback1Param('source.pull', pull, source) :
-    nonOpPull;
+    createRawCallback1Param('source.pull', pull, source) :
+    nonOpCallback;
   const cancelAlgorithm = cancel ?
     createPromiseCallback1Param('source.cancel', cancel, source) :
     nonOpCancel;
@@ -3529,8 +3566,15 @@ function readableByteStreamControllerCallPullIfNeeded(controller) {
     controller[kState].pullRejected =
       (error) => readableByteStreamControllerError(controller, error);
   }
-  PromisePrototypeThen(
-    controller[kState].pullAlgorithm(controller),
+  // See readableStreamDefaultControllerPull for the raw-callback contract.
+  let result;
+  try {
+    result = controller[kState].pullAlgorithm(controller);
+  } catch (error) {
+    result = PromiseReject(error);
+  }
+  thenAlgorithmResult(
+    result,
     controller[kState].pullFulfilled,
     controller[kState].pullRejected);
 }
@@ -3708,10 +3752,10 @@ function setupReadableByteStreamControllerFromSource(
   const autoAllocateChunkSize = source?.autoAllocateChunkSize;
   const startAlgorithm = start ?
     FunctionPrototypeBind(start, source, controller) :
-    nonOpStart;
+    nonOpCallback;
   const pullAlgorithm = pull ?
-    createPromiseCallback1Param('source.pull', pull, source) :
-    nonOpPull;
+    createRawCallback1Param('source.pull', pull, source) :
+    nonOpCallback;
   const cancelAlgorithm = cancel ?
     createPromiseCallback1Param('source.cancel', cancel, source) :
     nonOpCancel;