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

@@ -46,10 +46,6 @@ const {
  DOMException,
 }=internalBinding('messaging');
 
-const{
- markPromiseAsHandled,
-}=internalBinding('util');
-
 const{
  isArrayBufferView,
  isDataView,
@@ -139,7 +135,7 @@ const {
  writableStreamCloseQueuedOrInFlight,
  writableStreamDefaultWriterCloseWithErrorPropagation,
  writableStreamDefaultWriterRelease,
-writableStreamDefaultWriterWrite,
+writableStreamDefaultWriterWriteWithRequest,
  writerClosedPromise,
  writerReadyPromise,
 }=require('internal/webstreams/writablestream');
@@ -1530,8 +1526,38 @@ function readableStreamPipeTo(
 
 constpromise=PromiseWithResolvers();
 
-conststate={
-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.
+constwriteTracker={
+// 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){
+conststall=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){
+conststall=this.stall;
+this.stall=null;
+stall.reject(error);
+}
+},
 };
 
 // The error here can be undefined. The rejected arg
@@ -1548,11 +1574,14 @@ function readableStreamPipeTo(
 promise.resolve();
 }
 
-asyncfunctionwaitForCurrentWrite(){
-constwrite=state.currentWrite;
-awaitwrite;
-if(write!==state.currentWrite)
-awaitwaitForCurrentWrite();
+functionwaitForPendingWrites(){
+if(writeTracker.pending===0){
+returnwriteTracker.failed ?
+PromiseReject(writeTracker.failure) :
+PromiseResolve();
+}
+writeTracker.stall=PromiseWithResolvers();
+returnwriteTracker.stall.promise;
 }
 
 functionshutdownWithAnAction(action,rejected,originalError){
@@ -1561,7 +1590,7 @@ function readableStreamPipeTo(
 if(dest[kState].state==='writable'&&
 !writableStreamCloseQueuedOrInFlight(dest)){
 PromisePrototypeThen(
-waitForCurrentWrite(),
+waitForPendingWrites(),
 complete,
 (error)=>finalize(true,error));
 return;
@@ -1582,7 +1611,7 @@ function readableStreamPipeTo(
 if(dest[kState].state==='writable'&&
 !writableStreamCloseQueuedOrInFlight(dest)){
 PromisePrototypeThen(
-waitForCurrentWrite(),
+waitForPendingWrites(),
 ()=>finalize(rejected,error),
 (error)=>finalize(true,error));
 return;
@@ -1639,25 +1668,46 @@ function readableStreamPipeTo(
 PromisePrototypeThen(promise,action,()=>{});
 }
 
-asyncfunctionstep(){
-if(shuttingDown)returntrue;
+// 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`.
+letpendingChunk;
+letreadRequest;
+
+// Ready promise rejection is handled by the destination-errored
+// watcher.
+functionignoreReadyRejection(){}
+
+functionforwardChunk(){
+constchunk=pendingChunk;
+pendingChunk=undefined;
+writableStreamDefaultWriterWriteWithRequest(writer,chunk,writeTracker);
+pump();
+}
+
+functionpump(){
+if(shuttingDown)return;
 
 if(dest[kState].backpressure){
-awaitwriterReadyPromise(writer).promise;
-if(shuttingDown)returntrue;
+PromisePrototypeThen(
+writerReadyPromise(writer).promise,
+pump,
+ignoreReadyRejection);
+return;
 }
 
 constcontroller=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)returntrue;
+if(shuttingDown)return;
 
 constchunk=dequeueValue(controller);
 
@@ -1668,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);
-markPromiseAsHandled(state.currentWrite);
+writableStreamDefaultWriterWriteWithRequest(writer,chunk,writeTracker);
 
 // Check backpressure after each write
 if(dest[kState].backpressure){
@@ -1686,24 +1735,29 @@ function readableStreamPipeTo(
 
 // Check if stream closed during batch
 if(source[kState].state==='closed'){
-returntrue;
+return;
 }
 
-// Yield to microtask queue between batches to allow events/signals to fire
-returnfalse;
+// Yield to microtask queue between batches to allow events/signals
+// to fire
+queueMicrotask(pump);
+return;
 }
 
-// Slow path: use read request for async reads
-constpromise=PromiseWithResolvers();
-// eslint-disable-next-line no-use-before-define
-readableStreamDefaultReaderRead(reader,newPipeToReadableStreamReadRequest(writer,state,promise));
-
-returnpromise.promise;
-}
-
-asyncfunctionrun(){
-// Run until step resolves as true
-while(!awaitstep());
+// 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){
@@ -1715,7 +1769,7 @@ function readableStreamPipeTo(
 disposable=addAbortListener(signal,abortAlgorithm);
 }
 
-setPromiseHandled(run());
+pump();
 
 watchErrored(source,readerClosedPromise(reader).promise,(error)=>{
 if(!preventAbort){
@@ -1760,33 +1814,6 @@ function readableStreamPipeTo(
 returnpromise.promise;
 }
 
-classPipeToReadableStreamReadRequest{
-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);
-markPromiseAsHandled(this.state.currentWrite);
-this.promise.resolve(false);
-});
-}
-
-[kClose](){
-this.promise.resolve(true);
-}
-
-[kError](error){
-this.promise.reject(error);
-}
-}
-
 functionreadableStreamTee(stream,cloneForBranch2){
 if(isReadableByteStreamController(stream[kState].controller)){
 returnreadableByteStreamTee(stream);

lib/internal/webstreams/writablestream.js

@@ -1001,6 +1001,61 @@ function writableStreamDefaultWriterWrite(writer, chunk) {
 returnpromise;
 }
 
+// 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`.
+functionwritableStreamDefaultWriterWriteWithRequest(writer,chunk,request){
+constwriterState=writer[kState];
+conststream=writerState.stream;
+assert(stream!==undefined);
+conststreamState=stream[kState];
+const{
+ controller,
+}=streamState;
+constchunkSize=writableStreamDefaultControllerGetChunkSize(
+controller,
+chunk);
+if(stream!==writerState.stream){
+request.failed=true;
+request.failure=
+newERR_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=
+newERR_INVALID_STATE.TypeError('WritableStream is closed');
+return;
+}
+
+if(state==='erroring'){
+request.failed=true;
+request.failure=streamState.storedError;
+return;
+}
+
+assert(state==='writable');
+
+letwriteRequests=streamState.writeRequests;
+if(writeRequests===kEmptyQueue)
+writeRequests=streamState.writeRequests=newQueue();
+writeRequests.push(request);
+request.pending++;
+writableStreamDefaultControllerWrite(controller,chunk,chunkSize);
+}
+
 functionwritableStreamDefaultWriterRelease(writer){
 const{
  stream,
@@ -1376,6 +1431,7 @@ module.exports = {
  writableStreamCloseQueuedOrInFlight,
  writableStreamAddWriteRequest,
  writableStreamDefaultWriterWrite,
+ writableStreamDefaultWriterWriteWithRequest,
  writableStreamDefaultWriterRelease,
  writableStreamDefaultWriterGetDesiredSize,
  writableStreamDefaultWriterEnsureReadyPromiseRejected,