stream: cut per-chunk allocations in pipeTo

readableStreamPipeTo allocated, for every chunk written to the
destination, a { promise, resolve, reject } write request record that
it immediately marked as handled, and drove its loop with an async
step()/run() pair whose implicit promises cost one allocation and one
reaction per iteration. The parked-read path additionally allocated a
read request object, a PromiseWithResolvers record, and a microtask
closure per chunk; this is the steady state for pipeThrough, since a
TransformStream's readable side has a high water mark of zero.

Replace the per-write records with a single per-pipe tracker that the
write request queue holds once per pending write and whose
resolve()/reject() methods maintain a pending-write count, drive the
pump loop with plain callbacks instead of async functions, and reuse
one read request and one forwarding function across all chunks, the
same pattern tee uses since c543cfb.

Benchmark results (benchmark/compare.js --runs 20):
webstreams/pipe-to.js +29.9% to +35.8% across all 16 configurations
(all 99.9% confidence); a pipeThrough(TransformStream) passthrough
loop improves ~17%; every other webstreams benchmark is unchanged.

Signed-off-by: Matteo Collina <hello@matteocollina.com>
PR-URL: #64890
Reviewed-By: James M Snell <jasnell@gmail.com>
Reviewed-By: Yagiz Nizipli <yagiz@nizipli.com>

Author: mcollina

lib/internal/webstreams/readablestream.js

@@ -135,7 +135,7 @@ const {
   writableStreamCloseQueuedOrInFlight,
   writableStreamDefaultWriterCloseWithErrorPropagation,
   writableStreamDefaultWriterRelease,
-  writableStreamDefaultWriterWrite,
+  writableStreamDefaultWriterWriteWithRequest,
   writerClosedPromise,
   writerReadyPromise,
 } = require('internal/webstreams/writablestream');
@@ -1526,8 +1526,38 @@ function readableStreamPipeTo(
 
   const promise = PromiseWithResolvers();
 
-  const state = {
-    currentWrite: PromiseResolve(),
+  // One shared write request tracks every chunk written to the
+  // destination, instead of a { promise, resolve, reject } record per
+  // write. `stall` is armed by waitForPendingWrites() during shutdown;
+  // `failed`/`failure` latch a write that could not proceed.
+  const writeTracker = {
+    // Non-undefined: queue entries are discriminated from kNilRequest
+    // by `promise === undefined`.
+    promise: null,
+    pending: 0,
+    failed: false,
+    failure: undefined,
+    stall: null,
+    resolve() {
+      if (--this.pending === 0 && this.stall !== null) {
+        const stall = this.stall;
+        this.stall = null;
+        if (this.failed)
+          stall.reject(this.failure);
+        else
+          stall.resolve();
+      }
+    },
+    reject(error) {
+      this.pending--;
+      this.failed = true;
+      this.failure = error;
+      if (this.stall !== null) {
+        const stall = this.stall;
+        this.stall = null;
+        stall.reject(error);
+      }
+    },
   };
 
   // The error here can be undefined. The rejected arg
@@ -1544,11 +1574,14 @@ function readableStreamPipeTo(
       promise.resolve();
   }
 
-  async function waitForCurrentWrite() {
-    const write = state.currentWrite;
-    await write;
-    if (write !== state.currentWrite)
-      await waitForCurrentWrite();
+  function waitForPendingWrites() {
+    if (writeTracker.pending === 0) {
+      return writeTracker.failed ?
+        PromiseReject(writeTracker.failure) :
+        PromiseResolve();
+    }
+    writeTracker.stall = PromiseWithResolvers();
+    return writeTracker.stall.promise;
   }
 
   function shutdownWithAnAction(action, rejected, originalError) {
@@ -1557,7 +1590,7 @@ function readableStreamPipeTo(
     if (dest[kState].state === 'writable' &&
         !writableStreamCloseQueuedOrInFlight(dest)) {
       PromisePrototypeThen(
-        waitForCurrentWrite(),
+        waitForPendingWrites(),
         complete,
         (error) => finalize(true, error));
       return;
@@ -1578,7 +1611,7 @@ function readableStreamPipeTo(
     if (dest[kState].state === 'writable' &&
         !writableStreamCloseQueuedOrInFlight(dest)) {
       PromisePrototypeThen(
-        waitForCurrentWrite(),
+        waitForPendingWrites(),
         () => finalize(rejected, error),
         (error) => finalize(true, error));
       return;
@@ -1635,25 +1668,46 @@ function readableStreamPipeTo(
       PromisePrototypeThen(promise, action, () => {});
   }
 
-  async function step() {
-    if (shuttingDown) return true;
+  // The pump loop is callback-driven to avoid per-iteration promise
+  // allocations. At most one read is in flight at a time, so one read
+  // request and one forwarding function are reused for every chunk;
+  // the chunk travels through `pendingChunk`.
+  let pendingChunk;
+  let readRequest;
+
+  // Ready promise rejection is handled by the destination-errored
+  // watcher.
+  function ignoreReadyRejection() {}
+
+  function forwardChunk() {
+    const chunk = pendingChunk;
+    pendingChunk = undefined;
+    writableStreamDefaultWriterWriteWithRequest(writer, chunk, writeTracker);
+    pump();
+  }
+
+  function pump() {
+    if (shuttingDown) return;
 
     if (dest[kState].backpressure) {
-      await writerReadyPromise(writer).promise;
-      if (shuttingDown) return true;
+      PromisePrototypeThen(
+        writerReadyPromise(writer).promise,
+        pump,
+        ignoreReadyRejection);
+      return;
     }
 
     const controller = source[kState].controller;
 
     // Fast path: batch reads when data is buffered in a default controller.
-    // This avoids creating PipeToReadableStreamReadRequest objects and
-    // reduces promise allocation overhead.
+    // This avoids parking read requests and reduces promise allocation
+    // overhead.
     if (source[kState].state === 'readable' &&
         isReadableStreamDefaultController(controller) &&
         controller[kState].queue.length > 0) {
 
       while (controller[kState].queue.length > 0) {
-        if (shuttingDown) return true;
+        if (shuttingDown) return;
 
         const chunk = dequeueValue(controller);
 
@@ -1664,8 +1718,7 @@ function readableStreamPipeTo(
 
         // Write the chunk - we're already in a separate microtask from enqueue
         // because we awaited the writer ready promise above.
-        state.currentWrite = writableStreamDefaultWriterWrite(writer, chunk);
-        setPromiseHandled(state.currentWrite);
+        writableStreamDefaultWriterWriteWithRequest(writer, chunk, writeTracker);
 
         // Check backpressure after each write
         if (dest[kState].backpressure) {
@@ -1682,24 +1735,29 @@ function readableStreamPipeTo(
 
       // Check if stream closed during batch
       if (source[kState].state === 'closed') {
-        return true;
+        return;
       }
 
-      // Yield to microtask queue between batches to allow events/signals to fire
-      return false;
+      // Yield to microtask queue between batches to allow events/signals
+      // to fire
+      queueMicrotask(pump);
+      return;
     }
 
-    // Slow path: use read request for async reads
-    const promise = PromiseWithResolvers();
-    // eslint-disable-next-line no-use-before-define
-    readableStreamDefaultReaderRead(reader, new PipeToReadableStreamReadRequest(writer, state, promise));
-
-    return promise.promise;
-  }
-
-  async function run() {
-    // Run until step resolves as true
-    while (!await step());
+    // Slow path: park a lazily materialized read request. Close and
+    // error are handled by the source watchers.
+    readRequest ??= {
+      [kChunk](chunk) {
+        // Per spec, pipeTo must queue a microtask for the write to avoid
+        // synchronous write during enqueue(). See WHATWG Streams spec
+        // "ReadableStreamPipeTo" step 15's "chunk steps".
+        pendingChunk = chunk;
+        queueMicrotask(forwardChunk);
+      },
+      [kClose]() {},
+      [kError]() {},
+    };
+    readableStreamDefaultReaderRead(reader, readRequest);
   }
 
   if (signal !== undefined) {
@@ -1711,7 +1769,7 @@ function readableStreamPipeTo(
     disposable = addAbortListener(signal, abortAlgorithm);
   }
 
-  setPromiseHandled(run());
+  pump();
 
   watchErrored(source, readerClosedPromise(reader).promise, (error) => {
     if (!preventAbort) {
@@ -1756,33 +1814,6 @@ function readableStreamPipeTo(
   return promise.promise;
 }
 
-class PipeToReadableStreamReadRequest {
-  constructor(writer, state, promise) {
-    this.writer = writer;
-    this.state = state;
-    this.promise = promise;
-  }
-
-  [kChunk](chunk) {
-    // Per spec, pipeTo must queue a microtask for the write to avoid
-    // synchronous write during enqueue(). See WHATWG Streams spec
-    // "ReadableStreamPipeTo" step 15's "chunk steps".
-    queueMicrotask(() => {
-      this.state.currentWrite = writableStreamDefaultWriterWrite(this.writer, chunk);
-      setPromiseHandled(this.state.currentWrite);
-      this.promise.resolve(false);
-    });
-  }
-
-  [kClose]() {
-    this.promise.resolve(true);
-  }
-
-  [kError](error) {
-    this.promise.reject(error);
-  }
-}
-
 function readableStreamTee(stream, cloneForBranch2) {
   if (isReadableByteStreamController(stream[kState].controller)) {
     return readableByteStreamTee(stream);

lib/internal/webstreams/writablestream.js

@@ -1001,6 +1001,61 @@ function writableStreamDefaultWriterWrite(writer, chunk) {
   return promise;
 }
 
+// Variant of writableStreamDefaultWriterWrite for pipeTo: the caller
+// provides a shared request object instead of a per-write promise record.
+// `pending` is incremented before the controller write, which can settle
+// requests synchronously when the stream starts erroring; precondition
+// failures are latched on `failed`/`failure`.
+function writableStreamDefaultWriterWriteWithRequest(writer, chunk, request) {
+  const writerState = writer[kState];
+  const stream = writerState.stream;
+  assert(stream !== undefined);
+  const streamState = stream[kState];
+  const {
+    controller,
+  } = streamState;
+  const chunkSize = writableStreamDefaultControllerGetChunkSize(
+    controller,
+    chunk);
+  if (stream !== writerState.stream) {
+    request.failed = true;
+    request.failure =
+      new ERR_INVALID_STATE.TypeError('Mismatched WritableStreams');
+    return;
+  }
+  const {
+    state,
+  } = streamState;
+
+  if (state === 'errored') {
+    request.failed = true;
+    request.failure = streamState.storedError;
+    return;
+  }
+
+  if (streamState.closeQueuedOrInFlight || state === 'closed') {
+    request.failed = true;
+    request.failure =
+      new ERR_INVALID_STATE.TypeError('WritableStream is closed');
+    return;
+  }
+
+  if (state === 'erroring') {
+    request.failed = true;
+    request.failure = streamState.storedError;
+    return;
+  }
+
+  assert(state === 'writable');
+
+  let writeRequests = streamState.writeRequests;
+  if (writeRequests === kEmptyQueue)
+    writeRequests = streamState.writeRequests = new Queue();
+  writeRequests.push(request);
+  request.pending++;
+  writableStreamDefaultControllerWrite(controller, chunk, chunkSize);
+}
+
 function writableStreamDefaultWriterRelease(writer) {
   const {
     stream,
@@ -1376,6 +1431,7 @@ module.exports = {
   writableStreamCloseQueuedOrInFlight,
   writableStreamAddWriteRequest,
   writableStreamDefaultWriterWrite,
+  writableStreamDefaultWriterWriteWithRequest,
   writableStreamDefaultWriterRelease,
   writableStreamDefaultWriterGetDesiredSize,
   writableStreamDefaultWriterEnsureReadyPromiseRejected,