stream: use the ring buffer for WHATWG stream request queues

The [[queue]] backing the controllers became a ring buffer, but the
read and write request queues (readRequests, readIntoRequests,
writeRequests) were left as plain arrays consumed with
ArrayPrototypeShift, which is O(n) and, even for the single pending
request of the await-each regime, far slower than an indexed head
advance. Back them with the same Queue, materialized lazily from the
shared empty queue so acquiring a reader or constructing a writer
allocates no request storage until a read or write actually parks.

pipe-to: +4.7% to +9.4% (all 16 configs, ***)
readable-read type=byob: +2.3% (**)
parked read loop: +15%, write loop: +11% (local harness)

Follow-up to #64312.

Signed-off-by: Matteo Collina <hello@matteocollina.com>
PR-URL: #64431
Reviewed-By: James M Snell <jasnell@gmail.com>
Reviewed-By: Antoine du Hamel <duhamelantoine1995@gmail.com>

Author: mcollina

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;
 }