stream: fold desired-size check into WHATWG backpressure update

lib/internal/webstreams/readablestream.js

@@ -101,6 +101,7 @@ const {
   ArrayBufferViewGetByteLength,
   ArrayBufferViewGetByteOffset,
   AsyncIterator,
+  Queue,
   canCopyArrayBuffer,
   cloneAsUint8Array,
   copyArrayBuffer,
@@ -2196,9 +2197,9 @@ function readableStreamCancel(stream, reason) {
     reader,
   } = stream[kState];
   if (reader !== undefined && readableStreamHasBYOBReader(stream)) {
-    for (let n = 0; n < reader[kState].readIntoRequests.length; n++)
-      reader[kState].readIntoRequests[n][kClose]();
-    reader[kState].readIntoRequests = [];
+    const readIntoRequests = reader[kState].readIntoRequests;
+    while (readIntoRequests.length)
+      readIntoRequests.shift()[kClose]();
   }
 
   return PromisePrototypeThen(
@@ -2220,9 +2221,9 @@ function readableStreamClose(stream) {
   reader[kState].close?.resolve();
 
   if (readableStreamHasDefaultReader(stream)) {
-    for (let n = 0; n < reader[kState].readRequests.length; n++)
-      reader[kState].readRequests[n][kClose]();
-    reader[kState].readRequests = [];
+    const readRequests = reader[kState].readRequests;
+    while (readRequests.length)
+      readRequests.shift()[kClose]();
   }
 }
 
@@ -2250,14 +2251,14 @@ function readableStreamError(stream, error) {
   }
 
   if (readableStreamHasDefaultReader(stream)) {
-    for (let n = 0; n < reader[kState].readRequests.length; n++)
-      reader[kState].readRequests[n][kError](error);
-    reader[kState].readRequests = [];
+    const readRequests = reader[kState].readRequests;
+    while (readRequests.length)
+      readRequests.shift()[kError](error);
   } else {
     assert(readableStreamHasBYOBReader(stream));
-    for (let n = 0; n < reader[kState].readIntoRequests.length; n++)
-      reader[kState].readIntoRequests[n][kError](error);
-    reader[kState].readIntoRequests = [];
+    const readIntoRequests = reader[kState].readIntoRequests;
+    while (readIntoRequests.length)
+      readIntoRequests.shift()[kError](error);
   }
 }
 
@@ -2301,7 +2302,7 @@ function readableStreamFulfillReadRequest(stream, chunk, done) {
     reader,
   } = stream[kState];
   assert(reader[kState].readRequests.length);
-  const readRequest = ArrayPrototypeShift(reader[kState].readRequests);
+  const readRequest = reader[kState].readRequests.shift();
 
   // TODO(@jasnell): It's not clear under what exact conditions done
   // will be true here. The spec requires this check but none of the
@@ -2319,7 +2320,7 @@ function readableStreamFulfillReadIntoRequest(stream, chunk, done) {
     reader,
   } = stream[kState];
   assert(reader[kState].readIntoRequests.length);
-  const readIntoRequest = ArrayPrototypeShift(reader[kState].readIntoRequests);
+  const readIntoRequest = reader[kState].readIntoRequests.shift();
   if (done)
     readIntoRequest[kClose](chunk);
   else
@@ -2329,15 +2330,21 @@ function readableStreamFulfillReadIntoRequest(stream, chunk, done) {
 function readableStreamAddReadRequest(stream, readRequest) {
   assert(readableStreamHasDefaultReader(stream));
   assert(stream[kState].state === 'readable');
-  ArrayPrototypePush(stream[kState].reader[kState].readRequests, readRequest);
+  const readerState = stream[kState].reader[kState];
+  let readRequests = readerState.readRequests;
+  if (readRequests === kEmptyQueue)
+    readRequests = readerState.readRequests = new Queue();
+  readRequests.push(readRequest);
 }
 
 function readableStreamAddReadIntoRequest(stream, readIntoRequest) {
   assert(readableStreamHasBYOBReader(stream));
   assert(stream[kState].state !== 'errored');
-  ArrayPrototypePush(
-    stream[kState].reader[kState].readIntoRequests,
-    readIntoRequest);
+  const readerState = stream[kState].reader[kState];
+  let readIntoRequests = readerState.readIntoRequests;
+  if (readIntoRequests === kEmptyQueue)
+    readIntoRequests = readerState.readIntoRequests = new Queue();
+  readIntoRequests.push(readIntoRequest);
 }
 
 function readableStreamReaderGenericCancel(reader, reason) {
@@ -2409,10 +2416,9 @@ function readableStreamDefaultReaderRelease(reader) {
 }
 
 function readableStreamDefaultReaderErrorReadRequests(reader, e) {
-  for (let n = 0; n < reader[kState].readRequests.length; ++n) {
-    reader[kState].readRequests[n][kError](e);
-  }
-  reader[kState].readRequests = [];
+  const readRequests = reader[kState].readRequests;
+  while (readRequests.length)
+    readRequests.shift()[kError](e);
 }
 
 function readableStreamBYOBReaderRelease(reader) {
@@ -2424,10 +2430,9 @@ function readableStreamBYOBReaderRelease(reader) {
 }
 
 function readableStreamBYOBReaderErrorReadIntoRequests(reader, e) {
-  for (let n = 0; n < reader[kState].readIntoRequests.length; ++n) {
-    reader[kState].readIntoRequests[n][kError](e);
-  }
-  reader[kState].readIntoRequests = [];
+  const readIntoRequests = reader[kState].readIntoRequests;
+  while (readIntoRequests.length)
+    readIntoRequests.shift()[kError](e);
 }
 
 function readableStreamReaderGenericRelease(reader) {
@@ -2499,14 +2504,18 @@ function setupReadableStreamBYOBReader(reader, stream) {
   if (!isReadableByteStreamController(controller))
     throw new ERR_INVALID_ARG_VALUE('stream', stream, 'must be a byte stream');
   readableStreamReaderGenericInitialize(reader, stream);
-  reader[kState].readIntoRequests = [];
+  // The read-request queues use the same ring buffer as [[queue]], drained
+  // from a moving head rather than with ArrayPrototypeShift. Start from the
+  // shared immutable empty queue so acquiring a reader allocates no request
+  // storage until a read actually parks.
+  reader[kState].readIntoRequests = kEmptyQueue;
 }
 
 function setupReadableStreamDefaultReader(reader, stream) {
   if (isReadableStreamLocked(stream))
     throw new ERR_INVALID_STATE.TypeError('ReadableStream is locked');
   readableStreamReaderGenericInitialize(reader, stream);
-  reader[kState].readRequests = [];
+  reader[kState].readRequests = kEmptyQueue;
 }
 
 function readableStreamDefaultControllerClose(controller) {
@@ -3151,7 +3160,7 @@ function readableByteStreamControllerEnqueue(controller, chunk) {
       }
       const transferredView =
         new Uint8Array(transferredBuffer, byteOffset, byteLength);
-      const readRequest = ArrayPrototypeShift(readRequests);
+      const readRequest = readRequests.shift();
       readRequest[kChunk](transferredView);
     }
   } else if (readableStreamHasBYOBReader(stream)) {
@@ -3524,7 +3533,7 @@ function readableByteStreamControllerProcessReadRequestsUsingQueue(controller) {
     }
     readableByteStreamControllerFillReadRequestFromQueue(
       controller,
-      ArrayPrototypeShift(reader[kState].readRequests),
+      reader[kState].readRequests.shift(),
     );
   }
 }

lib/internal/webstreams/util.js

@@ -404,6 +404,7 @@ module.exports = {
   ArrayBufferViewGetByteLength,
   ArrayBufferViewGetByteOffset,
   AsyncIterator,
+  Queue,
   canCopyArrayBuffer,
   cloneAsUint8Array,
   copyArrayBuffer,

lib/internal/webstreams/writablestream.js

@@ -1,8 +1,6 @@
 'use strict';
 
 const {
-  ArrayPrototypePush,
-  ArrayPrototypeShift,
   FunctionPrototypeBind,
   FunctionPrototypeCall,
   ObjectDefineProperties,
@@ -55,6 +53,7 @@ const {
 } = require('internal/worker/js_transferable');
 
 const {
+  Queue,
   createPromiseCallbackNoParams,
   createPromiseCallback1Param,
   createPromiseCallback2Params,
@@ -613,7 +612,10 @@ function createWritableStreamState() {
     controller: undefined,
     state: 'writable',
     storedError: undefined,
-    writeRequests: [],
+    // Ring-buffer request queue, materialized lazily on the first pending
+    // write (see writableStreamAddWriteRequest) so construction allocates
+    // no request storage.
+    writeRequests: kEmptyQueue,
     writer: undefined,
     transfer: {
       __proto__: null,
@@ -755,13 +757,15 @@ function writableStreamClose(stream) {
   return promise;
 }
 
-function writableStreamUpdateBackpressure(stream, backpressure) {
-  const streamState = stream[kState];
-  assert(streamState.state === 'writable');
-  assert(!streamState.closeQueuedOrInFlight);
-  const {
-    writer,
-  } = streamState;
+// Callers invoke this only in the 'writable' state with no close queued or
+// in flight (the two conditions the spec's assertions check), so the
+// backpressure value is derived here from the desired size instead of being
+// computed and passed in by each caller.
+function writableStreamUpdateBackpressure(controller, streamState) {
+  const controllerState = controller[kState];
+  const backpressure =
+    controllerState.highWaterMark - controllerState.queueTotalSize <= 0;
+  const writer = streamState.writer;
   if (writer !== undefined && streamState.backpressure !== backpressure) {
     if (backpressure) {
       // The spec replaces [[readyPromise]] with a fresh pending promise;
@@ -823,7 +827,7 @@ function writableStreamRejectCloseAndClosedPromiseIfNeeded(stream) {
 function writableStreamMarkFirstWriteRequestInFlight(stream) {
   assert(stream[kState].inFlightWriteRequest.promise === undefined);
   assert(stream[kState].writeRequests.length);
-  const writeRequest = ArrayPrototypeShift(stream[kState].writeRequests);
+  const writeRequest = stream[kState].writeRequests.shift();
   stream[kState].inFlightWriteRequest = writeRequest;
 }
 
@@ -901,9 +905,9 @@ function writableStreamFinishErroring(stream) {
   stream[kState].state = 'errored';
   stream[kState].controller[kError]();
   const storedError = stream[kState].storedError;
-  for (let n = 0; n < stream[kState].writeRequests.length; n++)
-    stream[kState].writeRequests[n].reject(storedError);
-  stream[kState].writeRequests = [];
+  const writeRequests = stream[kState].writeRequests;
+  while (writeRequests.length)
+    writeRequests.shift().reject(storedError);
 
   if (stream[kState].pendingAbortRequest.abort.promise === undefined) {
     writableStreamRejectCloseAndClosedPromiseIfNeeded(stream);
@@ -952,7 +956,11 @@ function writableStreamAddWriteRequest(stream) {
   // PromiseWithResolvers() already returns a { promise, resolve, reject }
   // record, so push it as-is instead of rebuilding an identical object.
   const writeRequest = PromiseWithResolvers();
-  ArrayPrototypePush(stream[kState].writeRequests, writeRequest);
+  const streamState = stream[kState];
+  let writeRequests = streamState.writeRequests;
+  if (writeRequests === kEmptyQueue)
+    writeRequests = streamState.writeRequests = new Queue();
+  writeRequests.push(writeRequest);
   return writeRequest.promise;
 }
 
@@ -1094,9 +1102,7 @@ function writableStreamDefaultControllerWrite(controller, chunk, chunkSize) {
   const streamState = stream[kState];
   if (!streamState.closeQueuedOrInFlight &&
       streamState.state === 'writable') {
-    writableStreamUpdateBackpressure(
-      stream,
-      writableStreamDefaultControllerGetBackpressure(controller));
+    writableStreamUpdateBackpressure(controller, streamState);
   }
   writableStreamDefaultControllerAdvanceQueueIfNeeded(controller);
 }
@@ -1122,9 +1128,7 @@ function writableStreamDefaultControllerProcessWrite(controller, chunk) {
       dequeueValue(controller);
       if (!streamState.closeQueuedOrInFlight &&
           state === 'writable') {
-        writableStreamUpdateBackpressure(
-          stream,
-          writableStreamDefaultControllerGetBackpressure(controller));
+        writableStreamUpdateBackpressure(controller, streamState);
       }
       writableStreamDefaultControllerAdvanceQueueIfNeeded(controller);
     };
@@ -1223,10 +1227,6 @@ function writableStreamDefaultControllerClearAlgorithms(controller) {
   controller[kState].sizeAlgorithm = undefined;
 }
 
-function writableStreamDefaultControllerGetBackpressure(controller) {
-  return writableStreamDefaultControllerGetDesiredSize(controller) <= 0;
-}
-
 function writableStreamDefaultControllerAdvanceQueueIfNeeded(controller) {
   const {
     queue,
@@ -1311,9 +1311,7 @@ function setupWritableStreamDefaultController(
   };
   stream[kState].controller = controller;
 
-  writableStreamUpdateBackpressure(
-    stream,
-    writableStreamDefaultControllerGetBackpressure(controller));
+  writableStreamUpdateBackpressure(controller, stream[kState]);
 
   const startResult = startAlgorithm();
 
@@ -1396,7 +1394,6 @@ module.exports = {
   writableStreamDefaultControllerError,
   writableStreamDefaultControllerClose,
   writableStreamDefaultControllerClearAlgorithms,
-  writableStreamDefaultControllerGetBackpressure,
   writableStreamDefaultControllerAdvanceQueueIfNeeded,
   setupWritableStreamDefaultControllerFromSink,
   setupWritableStreamDefaultController,