stream: commit pull-into descriptors after filling from queue

Fixes: #56044
PR-URL: #56072
Reviewed-By: Matteo Collina <matteo.collina@gmail.com>
Reviewed-By: Matthew Aitken <maitken033380023@gmail.com>

Author: MattiasBuelens

lib/internal/webstreams/readablestream.js

@@ -94,6 +94,7 @@ const {
  ArrayBufferViewGetByteLength,
  ArrayBufferViewGetByteOffset,
  AsyncIterator,
+ canCopyArrayBuffer,
  cloneAsUint8Array,
  copyArrayBuffer,
  createPromiseCallback,
@@ -2552,6 +2553,15 @@ function readableByteStreamControllerCommitPullIntoDescriptor(stream, desc) {
 }
 }
 
+functionreadableByteStreamControllerCommitPullIntoDescriptors(stream,descriptors){
+for(leti=0;i<descriptors.length;++i){
+readableByteStreamControllerCommitPullIntoDescriptor(
+stream,
+descriptors[i],
+);
+}
+}
+
 functionreadableByteStreamControllerInvalidateBYOBRequest(controller){
 if(controller[kState].byobRequest===null)
 return;
@@ -2758,11 +2768,11 @@ function readableByteStreamControllerRespondInClosedState(controller, desc) {
  stream,
 }=controller[kState];
 if(readableStreamHasBYOBReader(stream)){
-while(readableStreamGetNumReadIntoRequests(stream)>0){
-readableByteStreamControllerCommitPullIntoDescriptor(
-stream,
-readableByteStreamControllerShiftPendingPullInto(controller));
+constfilledPullIntos=[];
+for(leti=0;i<readableStreamGetNumReadIntoRequests(stream);++i){
+ArrayPrototypePush(filledPullIntos,readableByteStreamControllerShiftPendingPullInto(controller));
 }
+readableByteStreamControllerCommitPullIntoDescriptors(stream,filledPullIntos);
 }
 }
 
@@ -2843,8 +2853,9 @@ function readableByteStreamControllerEnqueue(controller, chunk) {
 transferredBuffer,
 byteOffset,
 byteLength);
-readableByteStreamControllerProcessPullIntoDescriptorsUsingQueue(
+constfilledPullIntos=readableByteStreamControllerProcessPullIntoDescriptorsUsingQueue(
 controller);
+readableByteStreamControllerCommitPullIntoDescriptors(stream,filledPullIntos);
 }else{
 assert(!isReadableStreamLocked(stream));
 readableByteStreamControllerEnqueueChunkToQueue(
@@ -2937,6 +2948,7 @@ function readableByteStreamControllerFillPullIntoDescriptorFromQueue(
 constmaxAlignedBytes=maxBytesFilled-(maxBytesFilled%elementSize);
 lettotalBytesToCopyRemaining=maxBytesToCopy;
 letready=false;
+assert(!ArrayBufferPrototypeGetDetached(buffer));
 assert(bytesFilled<minimumFill);
 if(maxAlignedBytes>=minimumFill){
 totalBytesToCopyRemaining=maxAlignedBytes-bytesFilled;
@@ -2952,12 +2964,12 @@ function readableByteStreamControllerFillPullIntoDescriptorFromQueue(
 totalBytesToCopyRemaining,
 headOfQueue.byteLength);
 constdestStart=byteOffset+desc.bytesFilled;
-constarrayBufferByteLength=ArrayBufferPrototypeGetByteLength(buffer);
-if(arrayBufferByteLength-destStart<bytesToCopy){
-thrownewERR_INVALID_STATE.RangeError(
-'view ArrayBuffer size is invalid');
-}
-assert(arrayBufferByteLength-destStart>=bytesToCopy);
+assert(canCopyArrayBuffer(
+buffer,
+destStart,
+headOfQueue.buffer,
+headOfQueue.byteOffset,
+bytesToCopy));
 copyArrayBuffer(
 buffer,
 destStart,
@@ -2991,26 +3003,30 @@ function readableByteStreamControllerProcessPullIntoDescriptorsUsingQueue(
 const{
  closeRequested,
  pendingPullIntos,
- stream,
 }=controller[kState];
 assert(!closeRequested);
+constfilledPullIntos=[];
 while(pendingPullIntos.length){
 if(!controller[kState].queueTotalSize)
-return;
+break;
 constdesc=pendingPullIntos[0];
 if(readableByteStreamControllerFillPullIntoDescriptorFromQueue(
 controller,
 desc)){
 readableByteStreamControllerShiftPendingPullInto(controller);
-readableByteStreamControllerCommitPullIntoDescriptor(stream,desc);
+ArrayPrototypePush(filledPullIntos,desc);
 }
 }
+returnfilledPullIntos;
 }
 
 functionreadableByteStreamControllerRespondInReadableState(
 controller,
 bytesWritten,
 desc){
+const{
+ stream,
+}=controller[kState];
 const{
  buffer,
  bytesFilled,
@@ -3031,9 +3047,10 @@ function readableByteStreamControllerRespondInReadableState(
 controller,
 desc,
 );
-readableByteStreamControllerProcessPullIntoDescriptorsUsingQueue(
+constfilledPullIntos=readableByteStreamControllerProcessPullIntoDescriptorsUsingQueue(
 controller,
 );
+readableByteStreamControllerCommitPullIntoDescriptors(stream,filledPullIntos);
 return;
 }
 
@@ -3059,10 +3076,10 @@ function readableByteStreamControllerRespondInReadableState(
 ArrayBufferPrototypeGetByteLength(remainder));
 }
 desc.bytesFilled-=remainderSize;
-readableByteStreamControllerCommitPullIntoDescriptor(
-controller[kState].stream,
-desc);
-readableByteStreamControllerProcessPullIntoDescriptorsUsingQueue(controller);
+constfilledPullIntos=readableByteStreamControllerProcessPullIntoDescriptorsUsingQueue(controller);
+
+readableByteStreamControllerCommitPullIntoDescriptor(stream,desc);
+readableByteStreamControllerCommitPullIntoDescriptors(stream,filledPullIntos);
 }
 
 functionreadableByteStreamControllerRespondWithNewView(controller,view){

lib/internal/webstreams/util.js

@@ -1,6 +1,8 @@
 'use strict';
 
 const{
+ ArrayBufferPrototypeGetByteLength,
+ ArrayBufferPrototypeGetDetached,
  ArrayBufferPrototypeSlice,
  ArrayPrototypePush,
  ArrayPrototypeShift,
@@ -107,6 +109,14 @@ function cloneAsUint8Array(view) {
 );
 }
 
+functioncanCopyArrayBuffer(toBuffer,toIndex,fromBuffer,fromIndex,count){
+returntoBuffer!==fromBuffer&&
+!ArrayBufferPrototypeGetDetached(toBuffer)&&
+!ArrayBufferPrototypeGetDetached(fromBuffer)&&
+toIndex+count<=ArrayBufferPrototypeGetByteLength(toBuffer)&&
+fromIndex+count<=ArrayBufferPrototypeGetByteLength(fromBuffer);
+}
+
 functionisBrandCheck(brand){
 return(value)=>{
 returnvalue!=null&&
@@ -261,6 +271,7 @@ module.exports = {
  ArrayBufferViewGetByteLength,
  ArrayBufferViewGetByteOffset,
  AsyncIterator,
+ canCopyArrayBuffer,
  createPromiseCallback,
  cloneAsUint8Array,
  copyArrayBuffer,