stream: use the ring buffer for WHATWG stream request queues

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,
@@ -823,7 +825,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 +903,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 +954,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;
 }