stream: optimize webstreams pipeTo

PR-URL: #62079
Reviewed-By: James M Snell <jasnell@gmail.com>
Reviewed-By: Matteo Collina <matteo.collina@gmail.com>

Author: MattiasBuelens

lib/internal/webstreams/readablestream.js

@@ -1532,10 +1532,12 @@ function readableStreamPipeTo(
 }
 
 asyncfunctionstep(){
-if(shuttingDown)
-returntrue;
+if(shuttingDown)returntrue;
 
-awaitwriter[kState].ready.promise;
+if(dest[kState].backpressure){
+awaitwriter[kState].ready.promise;
+if(shuttingDown)returntrue;
+}
 
 constcontroller=source[kState].controller;
 
@@ -1549,7 +1551,6 @@ function readableStreamPipeTo(
 while(controller[kState].queue.length>0){
 if(shuttingDown)returntrue;
 
-source[kState].disturbed=true;
 constchunk=dequeueValue(controller);
 
 if(controller[kState].closeRequested&&!controller[kState].queue.length){
@@ -1563,20 +1564,17 @@ function readableStreamPipeTo(
 setPromiseHandled(state.currentWrite);
 
 // Check backpressure after each write
-if(dest[kState].state==='writable'){
-constdesiredSize=writer.desiredSize;
-if(desiredSize!==null&&desiredSize<=0){
-// Backpressure - stop batch and wait for ready
-break;
-}
+if(dest[kState].backpressure){
+// Backpressure - stop batch and wait for ready
+break;
+}elseif(dest[kState].state!=='writable'||writableStreamCloseQueuedOrInFlight(dest)){
+// Closing or erroring - stop batch and wait for shutdown
+break;
 }
 }
 
 // Trigger pull if needed after batch
-if(source[kState].state==='readable'&&
-!controller[kState].closeRequested){
-readableStreamDefaultControllerCallPullIfNeeded(controller);
-}
+readableStreamDefaultControllerCallPullIfNeeded(controller);
 
 // Check if stream closed during batch
 if(source[kState].state==='closed'){