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,10 +3441,12 @@ 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. Perform ! [$CopyDataBlockBytes$](|pullIntoDescriptor|'s [=pull-into
-     descriptor/buffer=].\[[ArrayBufferData]], |destStart|,
-     |headOfQueue|'s [=readable byte stream queue entry/buffer=].\[[ArrayBufferData]],
-     |headOfQueue|'s [=readable byte stream queue entry/byte offset=], |bytesToCopy|).
+  1. If ! [$SafeCopyDataBlockBytes$](|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 false,
+   1. Let |e| be a {{TypeError}} exception.
+   1. Perform ! [$ReadableByteStreamControllerError$](|controller|, |e|).
+   1. Return false.
   1. If |headOfQueue|'s [=readable byte stream queue entry/byte length=] is |bytesToCopy|,
    1. [=list/Remove=] |queue|[0].
   1. Otherwise,
@@ -3544,17 +3552,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 +3692,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 +3716,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 +3739,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 +6850,24 @@ The following abstract operations are a grab-bag of utilities.
  1. Return ? [$StructuredDeserialize$](|serialized|, [=the current Realm=]).
 </div>
 
+<div algorithm>
+ <dfn abstract-op lt="SafeCopyDataBlockBytes">SafeCopyDataBlockBytes(|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. Perform ! [$CopyDataBlockBytes$](|toBuffer|.\[[ArrayBufferData]], |toIndex|,
+    |fromBuffer|.\[[ArrayBufferData]], |fromIndex|, |count|).
+ 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 { CopyDataBlockBytes, IsDetachedBuffer } = require('./ecmascript');
 
 exports.IsNonNegativeNumber = v => {
   if (typeof v !== 'number') {
@@ -20,3 +21,23 @@ exports.CloneAsUint8Array = O => {
   const buffer = O.buffer.slice(O.byteOffset, O.byteOffset + O.byteLength);
   return new Uint8Array(buffer);
 };
+
+exports.SafeCopyDataBlockBytes = (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;
+  }
+  CopyDataBlockBytes(toBuffer, toIndex, fromBuffer, fromIndex, count);
+  return true;
+};

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

@@ -4,9 +4,9 @@ const assert = require('assert');
 const { promiseResolvedWith, promiseRejectedWith, newPromise, resolvePromise, rejectPromise, uponPromise,
         setPromiseIsHandledToTrue, waitForAllPromise, transformPromiseWith, uponFulfillment, uponRejection } =
   require('../helpers/webidl.js');
-const { CanTransferArrayBuffer, Call, CopyDataBlockBytes, CreateArrayFromList, GetIterator, GetMethod, IsDetachedBuffer,
+const { CanTransferArrayBuffer, Call, CreateArrayFromList, GetIterator, GetMethod, IsDetachedBuffer,
         IteratorComplete, IteratorNext, IteratorValue, TransferArrayBuffer, typeIsObject } = require('./ecmascript.js');
-const { CloneAsUint8Array, IsNonNegativeNumber } = require('./miscellaneous.js');
+const { CloneAsUint8Array, IsNonNegativeNumber, SafeCopyDataBlockBytes } = 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,7 +1447,14 @@ function ReadableByteStreamControllerFillPullIntoDescriptorFromQueue(controller,
     const bytesToCopy = Math.min(totalBytesToCopyRemaining, headOfQueue.byteLength);
 
     const destStart = pullIntoDescriptor.byteOffset + pullIntoDescriptor.bytesFilled;
-    CopyDataBlockBytes(pullIntoDescriptor.buffer, destStart, headOfQueue.buffer, headOfQueue.byteOffset, bytesToCopy);
+
+    if (SafeCopyDataBlockBytes(pullIntoDescriptor.buffer, destStart, headOfQueue.buffer, headOfQueue.byteOffset,
+                               bytesToCopy) === false) {
+      // This should never happen. Please report an issue if it does! https://github.com/whatwg/streams/issues
+      const e = new TypeError('Invalid buffer');
+      ReadableByteStreamControllerError(controller, e);
+      return false;
+    }
 
     if (headOfQueue.byteLength === bytesToCopy) {
       queue.shift();
@@ -1532,23 +1543,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 +1683,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 +1703,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 +1730,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) {