Commit pull-into descriptors after filling from queue

index.bs

@@ -3331,7 +3331,12 @@ The following abstract operations support the implementation of the
  1. Otherwise, if ! [$ReadableStreamHasBYOBReader$](|stream|) is true,
   1. Perform ! [$ReadableByteStreamControllerEnqueueChunkToQueue$](|controller|,
      |transferredBuffer|, |byteOffset|, |byteLength|).
-  1. Perform ! [$ReadableByteStreamControllerProcessPullIntoDescriptorsUsingQueue$](|controller|).
+  1. Let |filledPullIntos| be the result of performing
+     ! [$ReadableByteStreamControllerProcessPullIntoDescriptorsUsingQueue$](|controller|).
+  1. [=list/For each=] |filledPullInto| of |filledPullIntos|,
+    1. Perform !
+       [$ReadableByteStreamControllerCommitPullIntoDescriptor$](|controller|.[=ReadableByteStreamController/[[stream]]=],
+       |filledPullInto|).
  1. Otherwise,
   1. Assert: ! [$IsReadableStreamLocked$](|stream|) is false.
   1. Perform ! [$ReadableByteStreamControllerEnqueueChunkToQueue$](|controller|,
@@ -3416,6 +3421,7 @@ The following abstract operations support the implementation of the
     |maxBytesToCopy|.
  1. Let |totalBytesToCopyRemaining| be |maxBytesToCopy|.
  1. Let |ready| be false.
+ 1. Assert: ! [$IsDetachedBuffer$](|pullIntoDescriptor|'s [=pull-into descriptor/buffer=]) is false.
  1. Assert: |pullIntoDescriptor|'s [=pull-into descriptor/bytes filled=] < |pullIntoDescriptor|'s
     [=pull-into descriptor/minimum fill=].
  1. Let |remainderBytes| be the remainder after dividing |maxBytesFilled| by |pullIntoDescriptor|'s
@@ -3435,6 +3441,14 @@ The following abstract operations support the implementation of the
      queue entry/byte length=]).
   1. Let |destStart| be |pullIntoDescriptor|'s [=pull-into descriptor/byte offset=] +
      |pullIntoDescriptor|'s [=pull-into descriptor/bytes filled=].
+  1. Assert: ! [$CanCopyDataBlockBytes$](|pullIntoDescriptor|'s [=pull-into descriptor/buffer=],
+     |destStart|, |headOfQueue|'s [=readable byte stream queue entry/buffer=],
+     |headOfQueue|'s [=readable byte stream queue entry/byte offset=], |bytesToCopy|) is true.
+     <p class="warning">If this assertion were to fail (due to a bug in this specification or
+     its implementation), then the next step may read from or write to potentially invalid memory.
+     The user agent should always check this assertion, and stop in an [=implementation-defined=]
+     manner if it fails (e.g. by crashing the process, or by
+     <a abstract-op lt="ReadableByteStreamControllerError">erroring the stream</a>).
   1. Perform ! [$CopyDataBlockBytes$](|pullIntoDescriptor|'s [=pull-into
      descriptor/buffer=].\[[ArrayBufferData]], |destStart|,
      |headOfQueue|'s [=readable byte stream queue entry/buffer=].\[[ArrayBufferData]],
@@ -3544,17 +3558,17 @@ The following abstract operations support the implementation of the
  performs the following steps:
 
  1. Assert: |controller|.[=ReadableByteStreamController/[[closeRequested]]=] is false.
+ 1. Let |filledPullIntos| be a new empty [=list=].
  1. [=While=] |controller|.[=ReadableByteStreamController/[[pendingPullIntos]]=] is not
     [=list/is empty|empty=],
-  1. If |controller|.[=ReadableByteStreamController/[[queueTotalSize]]=] is 0, return.
+  1. If |controller|.[=ReadableByteStreamController/[[queueTotalSize]]=] is 0, then [=break=].
   1. Let |pullIntoDescriptor| be
      |controller|.[=ReadableByteStreamController/[[pendingPullIntos]]=][0].
   1. If ! [$ReadableByteStreamControllerFillPullIntoDescriptorFromQueue$](|controller|,
      |pullIntoDescriptor|) is true,
    1. Perform ! [$ReadableByteStreamControllerShiftPendingPullInto$](|controller|).
-   1. Perform !
-      [$ReadableByteStreamControllerCommitPullIntoDescriptor$](|controller|.[=ReadableByteStreamController/[[stream]]=],
-      |pullIntoDescriptor|).
+   1. [=list/Append=] |pullIntoDescriptor| to |filledPullIntos|.
+ 1. Return |filledPullIntos|.
 </div>
 
 <div algorithm>
@@ -3684,11 +3698,16 @@ The following abstract operations support the implementation of the
     perform ! [$ReadableByteStreamControllerShiftPendingPullInto$](|controller|).
  1. Let |stream| be |controller|.[=ReadableByteStreamController/[[stream]]=].
  1. If ! [$ReadableStreamHasBYOBReader$](|stream|) is true,
-  1. [=While=] ! [$ReadableStreamGetNumReadIntoRequests$](|stream|) > 0,
+  1. Let |filledPullIntos| be a new empty [=list=].
+  1. Let |i| be 0.
+  1. [=While=] |i| < ! [$ReadableStreamGetNumReadIntoRequests$](|stream|),
    1. Let |pullIntoDescriptor| be !
       [$ReadableByteStreamControllerShiftPendingPullInto$](|controller|).
+   1. [=list/Append=] |pullIntoDescriptor| to |filledPullIntos|.
+   1. Set |i| to |i| + 1.
+  1. [=list/For each=] |filledPullInto| of |filledPullIntos|,
    1. Perform ! [$ReadableByteStreamControllerCommitPullIntoDescriptor$](|stream|,
-      |pullIntoDescriptor|).
+      |filledPullInto|).
 </div>
 
 <div algorithm>
@@ -3703,7 +3722,12 @@ The following abstract operations support the implementation of the
  1. If |pullIntoDescriptor|'s [=pull-into descriptor/reader type=] is "`none`",
   1. Perform ? [$ReadableByteStreamControllerEnqueueDetachedPullIntoToQueue$](|controller|,
      |pullIntoDescriptor|).
-  1. Perform ! [$ReadableByteStreamControllerProcessPullIntoDescriptorsUsingQueue$](|controller|).
+  1. Let |filledPullIntos| be the result of performing
+     ! [$ReadableByteStreamControllerProcessPullIntoDescriptorsUsingQueue$](|controller|).
+  1. [=list/For each=] |filledPullInto| of |filledPullIntos|,
+    1. Perform !
+       [$ReadableByteStreamControllerCommitPullIntoDescriptor$](|controller|.[=ReadableByteStreamController/[[stream]]=],
+       |filledPullInto|).
   1. Return.
  1. If |pullIntoDescriptor|'s [=pull-into descriptor/bytes filled=] &lt; |pullIntoDescriptor|'s
     [=pull-into descriptor/minimum fill=], return.
@@ -3721,10 +3745,15 @@ The following abstract operations support the implementation of the
       |remainderSize|).
  1. Set |pullIntoDescriptor|'s [=pull-into descriptor/bytes filled=] to |pullIntoDescriptor|'s
     [=pull-into descriptor/bytes filled=] − |remainderSize|.
+ 1. Let |filledPullIntos| be the result of performing
+    ! [$ReadableByteStreamControllerProcessPullIntoDescriptorsUsingQueue$](|controller|).
  1. Perform !
     [$ReadableByteStreamControllerCommitPullIntoDescriptor$](|controller|.[=ReadableByteStreamController/[[stream]]=],
     |pullIntoDescriptor|).
- 1. Perform ! [$ReadableByteStreamControllerProcessPullIntoDescriptorsUsingQueue$](|controller|).
+ 1. [=list/For each=] |filledPullInto| of |filledPullIntos|,
+   1. Perform !
+      [$ReadableByteStreamControllerCommitPullIntoDescriptor$](|controller|.[=ReadableByteStreamController/[[stream]]=],
+      |filledPullInto|).
 </div>
 
 <div algorithm>
@@ -6827,6 +6856,22 @@ The following abstract operations are a grab-bag of utilities.
  1. Return ? [$StructuredDeserialize$](|serialized|, [=the current Realm=]).
 </div>
 
+<div algorithm>
+ <dfn abstract-op lt="CanCopyDataBlockBytes">CanCopyDataBlockBytes(|toBuffer|, |toIndex|,
+ |fromBuffer|, |fromIndex|, |count|)</dfn> performs the following steps:
+
+ 1. Assert: [$Type$](|toBuffer|) is Object.
+ 1. Assert: |toBuffer| has an \[[ArrayBufferData]] internal slot.
+ 1. Assert: [$Type$](|fromBuffer|) is Object.
+ 1. Assert: |fromBuffer| has an \[[ArrayBufferData]] internal slot.
+ 1. If |toBuffer| is |fromBuffer|, return false.
+ 1. If ! [$IsDetachedBuffer$](|toBuffer|) is true, return false.
+ 1. If ! [$IsDetachedBuffer$](|fromBuffer|) is true, return false.
+ 1. If |toIndex| + |count| > |toBuffer|.\[[ArrayBufferByteLength]], return false.
+ 1. If |fromIndex| + |count| > |fromBuffer|.\[[ArrayBufferByteLength]], return false.
+ 1. Return true.
+</div>
+
 <h2 id="other-specs">Using streams in other specifications</h2>
 
 Much of this standard concerns itself with the internal machinery of streams. Other specifications

reference-implementation/lib/abstract-ops/miscellaneous.js

@@ -1,4 +1,5 @@
 'use strict';
+const { IsDetachedBuffer } = require('./ecmascript');
 
 exports.IsNonNegativeNumber = v => {
   if (typeof v !== 'number') {
@@ -20,3 +21,22 @@ exports.CloneAsUint8Array = O => {
   const buffer = O.buffer.slice(O.byteOffset, O.byteOffset + O.byteLength);
   return new Uint8Array(buffer);
 };
+
+exports.CanCopyDataBlockBytes = (toBuffer, toIndex, fromBuffer, fromIndex, count) => {
+  if (toBuffer === fromBuffer) {
+    return false;
+  }
+  if (IsDetachedBuffer(toBuffer) === true) {
+    return false;
+  }
+  if (IsDetachedBuffer(fromBuffer) === true) {
+    return false;
+  }
+  if (toIndex + count > toBuffer.byteLength) {
+    return false;
+  }
+  if (fromIndex + count > fromBuffer.byteLength) {
+    return false;
+  }
+  return true;
+};

reference-implementation/lib/abstract-ops/readable-streams.js

@@ -6,7 +6,7 @@ const { promiseResolvedWith, promiseRejectedWith, newPromise, resolvePromise, re
   require('../helpers/webidl.js');
 const { CanTransferArrayBuffer, Call, CopyDataBlockBytes, CreateArrayFromList, GetIterator, GetMethod, IsDetachedBuffer,
         IteratorComplete, IteratorNext, IteratorValue, TransferArrayBuffer, typeIsObject } = require('./ecmascript.js');
-const { CloneAsUint8Array, IsNonNegativeNumber } = require('./miscellaneous.js');
+const { CanCopyDataBlockBytes, CloneAsUint8Array, IsNonNegativeNumber } = require('./miscellaneous.js');
 const { EnqueueValueWithSize, ResetQueue } = require('./queue-with-sizes.js');
 const { AcquireWritableStreamDefaultWriter, IsWritableStreamLocked, WritableStreamAbort,
         WritableStreamDefaultWriterCloseWithErrorPropagation, WritableStreamDefaultWriterRelease,
@@ -1360,7 +1360,10 @@ function ReadableByteStreamControllerEnqueue(controller, chunk) {
   } else if (ReadableStreamHasBYOBReader(stream) === true) {
     // TODO: Ideally in this branch detaching should happen only if the buffer is not consumed fully.
     ReadableByteStreamControllerEnqueueChunkToQueue(controller, transferredBuffer, byteOffset, byteLength);
-    ReadableByteStreamControllerProcessPullIntoDescriptorsUsingQueue(controller);
+    const filledPullIntos = ReadableByteStreamControllerProcessPullIntoDescriptorsUsingQueue(controller);
+    for (const filledPullInto of filledPullIntos) {
+      ReadableByteStreamControllerCommitPullIntoDescriptor(controller._stream, filledPullInto);
+    }
   } else {
     assert(IsReadableStreamLocked(stream) === false);
     ReadableByteStreamControllerEnqueueChunkToQueue(controller, transferredBuffer, byteOffset, byteLength);
@@ -1425,6 +1428,7 @@ function ReadableByteStreamControllerFillPullIntoDescriptorFromQueue(controller,
 
   let totalBytesToCopyRemaining = maxBytesToCopy;
   let ready = false;
+  assert(!IsDetachedBuffer(pullIntoDescriptor.buffer));
   assert(pullIntoDescriptor.bytesFilled < pullIntoDescriptor.minimumFill);
   const remainderBytes = maxBytesFilled % pullIntoDescriptor.elementSize;
   const maxAlignedBytes = maxBytesFilled - remainderBytes;
@@ -1443,6 +1447,9 @@ function ReadableByteStreamControllerFillPullIntoDescriptorFromQueue(controller,
     const bytesToCopy = Math.min(totalBytesToCopyRemaining, headOfQueue.byteLength);
 
     const destStart = pullIntoDescriptor.byteOffset + pullIntoDescriptor.bytesFilled;
+
+    assert(CanCopyDataBlockBytes(pullIntoDescriptor.buffer, destStart, headOfQueue.buffer, headOfQueue.byteOffset,
+                                 bytesToCopy));
     CopyDataBlockBytes(pullIntoDescriptor.buffer, destStart, headOfQueue.buffer, headOfQueue.byteOffset, bytesToCopy);
 
     if (headOfQueue.byteLength === bytesToCopy) {
@@ -1532,23 +1539,22 @@ function ReadableByteStreamControllerInvalidateBYOBRequest(controller) {
 function ReadableByteStreamControllerProcessPullIntoDescriptorsUsingQueue(controller) {
   assert(controller._closeRequested === false);
 
+  const filledPullIntos = [];
   while (controller._pendingPullIntos.length > 0) {
     if (controller._queueTotalSize === 0) {
-      return;
+      break;
     }
 
     const pullIntoDescriptor = controller._pendingPullIntos[0];
     assert(pullIntoDescriptor.readerType !== 'none');
 
     if (ReadableByteStreamControllerFillPullIntoDescriptorFromQueue(controller, pullIntoDescriptor) === true) {
       ReadableByteStreamControllerShiftPendingPullInto(controller);
-
-      ReadableByteStreamControllerCommitPullIntoDescriptor(
-        controller._stream,
-        pullIntoDescriptor
-      );
+      filledPullIntos.push(pullIntoDescriptor);
     }
   }
+
+  return filledPullIntos;
 }
 
 function ReadableByteStreamControllerProcessReadRequestsUsingQueue(controller) {
@@ -1673,9 +1679,15 @@ function ReadableByteStreamControllerRespondInClosedState(controller, firstDescr
 
   const stream = controller._stream;
   if (ReadableStreamHasBYOBReader(stream) === true) {
-    while (ReadableStreamGetNumReadIntoRequests(stream) > 0) {
+    const filledPullIntos = [];
+    let i = 0;
+    while (i < ReadableStreamGetNumReadIntoRequests(stream)) {
       const pullIntoDescriptor = ReadableByteStreamControllerShiftPendingPullInto(controller);
-      ReadableByteStreamControllerCommitPullIntoDescriptor(stream, pullIntoDescriptor);
+      filledPullIntos.push(pullIntoDescriptor);
+      ++i;
+    }
+    for (const filledPullInto of filledPullIntos) {
+      ReadableByteStreamControllerCommitPullIntoDescriptor(stream, filledPullInto);
     }
   }
 }
@@ -1687,7 +1699,10 @@ function ReadableByteStreamControllerRespondInReadableState(controller, bytesWri
 
   if (pullIntoDescriptor.readerType === 'none') {
     ReadableByteStreamControllerEnqueueDetachedPullIntoToQueue(controller, pullIntoDescriptor);
-    ReadableByteStreamControllerProcessPullIntoDescriptorsUsingQueue(controller);
+    const filledPullIntos = ReadableByteStreamControllerProcessPullIntoDescriptorsUsingQueue(controller);
+    for (const filledPullInto of filledPullIntos) {
+      ReadableByteStreamControllerCommitPullIntoDescriptor(controller._stream, filledPullInto);
+    }
     return;
   }
 
@@ -1711,9 +1726,12 @@ function ReadableByteStreamControllerRespondInReadableState(controller, bytesWri
   }
 
   pullIntoDescriptor.bytesFilled -= remainderSize;
-  ReadableByteStreamControllerCommitPullIntoDescriptor(controller._stream, pullIntoDescriptor);
+  const filledPullIntos = ReadableByteStreamControllerProcessPullIntoDescriptorsUsingQueue(controller);
 
-  ReadableByteStreamControllerProcessPullIntoDescriptorsUsingQueue(controller);
+  ReadableByteStreamControllerCommitPullIntoDescriptor(controller._stream, pullIntoDescriptor);
+  for (const filledPullInto of filledPullIntos) {
+    ReadableByteStreamControllerCommitPullIntoDescriptor(controller._stream, filledPullInto);
+  }
 }
 
 function ReadableByteStreamControllerRespondInternal(controller, bytesWritten) {