stream: use the ring buffer for pending BYOB pull-into descriptors

The byte controller's [[pendingPullIntos]] list was still a plain array
consumed with ArrayPrototypeShift, while every other per-chunk queue in
the WHATWG streams implementation has moved to the Queue ring buffer.
BYOB reads push and shift one descriptor per read, and Array.prototype
shift has real per-call cost even at length 1.

Back the descriptor list with the same lazily materialized Queue used
for the request queues, so constructing a byte stream still allocates
no descriptor storage.

Signed-off-by: Matteo Collina <hello@matteocollina.com>
PR-URL: https://github.com/nodejs/node/pull/64818
Reviewed-By: Yagiz Nizipli <yagiz@nizipli.com>
Reviewed-By: Paolo Insogna <paolo@cowtech.it>
Reviewed-By: Aviv Keller <me@aviv.sh>
Reviewed-By: Gürgün Dayıoğlu <hey@gurgun.day>

benchmark/webstreams/tee.js

@@ -0,0 +1,44 @@
+'use strict';
+const common = require('../common.js');
+const { ReadableStream } = require('node:stream/web');
+
+const bench = common.createBenchmark(main, {
+  n: [1e5],
+  type: ['normal', 'bytes'],
+});
+
+async function main({ n, type }) {
+  let i = 0;
+  const source = type === 'bytes' ?
+    {
+      type: 'bytes',
+      pull(controller) {
+        if (i++ < n) controller.enqueue(new Uint8Array(16));
+        else controller.close();
+      },
+    } :
+    {
+      pull(controller) {
+        if (i++ < n) controller.enqueue('a');
+        else controller.close();
+      },
+    };
+
+  const rs = new ReadableStream(source);
+  const [branch1, branch2] = rs.tee();
+  const reader1 = branch1.getReader();
+  const reader2 = branch2.getReader();
+  let reads = 0;
+
+  bench.start();
+  for (;;) {
+    const [result1, result2] = await Promise.all([
+      reader1.read(),
+      reader2.read(),
+    ]);
+    if (result1.done || result2.done) break;
+    reads++;
+  }
+  bench.end(reads);
+  console.assert(reads === n);
+}

lib/internal/webstreams/readablestream.js

@@ -7,7 +7,6 @@ const {
   ArrayBufferPrototypeSlice,
   ArrayBufferPrototypeTransfer,
   ArrayPrototypePush,
-  ArrayPrototypeShift,
   DataView,
   FunctionPrototypeBind,
   FunctionPrototypeCall,
@@ -1258,7 +1257,7 @@ class ReadableByteStreamController {
         byteOffset,
         bytesFilled,
         byteLength,
-      } = this[kState].pendingPullIntos[0];
+      } = this[kState].pendingPullIntos.peek();
       const view =
         new Uint8Array(
           buffer,
@@ -1341,9 +1340,11 @@ class ReadableByteStreamController {
       pendingPullIntos,
     } = this[kState];
     if (pendingPullIntos.length > 0) {
-      const firstPendingPullInto = pendingPullIntos[0];
+      const firstPendingPullInto = pendingPullIntos.peek();
       firstPendingPullInto.type = 'none';
-      this[kState].pendingPullIntos = [firstPendingPullInto];
+      const queue = new Queue();
+      queue.push(firstPendingPullInto);
+      this[kState].pendingPullIntos = queue;
     }
   }
 
@@ -1804,29 +1805,42 @@ function readableStreamDefaultTee(stream, cloneForBranch2) {
   let branch2;
   const cancelPromise = PromiseWithResolvers();
 
+  // At most one read is ever in flight (`reading` guards pullAlgorithm),
+  // so one read request object and one forwarding microtask function are
+  // reused for every chunk; the chunk travels through `pendingChunk`.
+  // The request is materialized lazily on the first pull so that tee()
+  // itself stays allocation-light.
+  let pendingChunk;
+  let readRequest;
+  function forwardChunk() {
+    reading = false;
+    const value1 = pendingChunk;
+    let value2 = pendingChunk;
+    pendingChunk = undefined;
+    if (!canceled2 && cloneForBranch2) {
+      value2 = structuredClone(value2);
+    }
+    if (!canceled1) {
+      readableStreamDefaultControllerEnqueue(
+        branch1[kState].controller,
+        value1);
+    }
+    if (!canceled2) {
+      readableStreamDefaultControllerEnqueue(
+        branch2[kState].controller,
+        value2);
+    }
+  }
+
   async function pullAlgorithm() {
     if (reading) return;
     reading = true;
-    const readRequest = {
+    readRequest ??= {
       [kChunk](value) {
-        queueMicrotask(() => {
-          reading = false;
-          const value1 = value;
-          let value2 = value;
-          if (!canceled2 && cloneForBranch2) {
-            value2 = structuredClone(value2);
-          }
-          if (!canceled1) {
-            readableStreamDefaultControllerEnqueue(
-              branch1[kState].controller,
-              value1);
-          }
-          if (!canceled2) {
-            readableStreamDefaultControllerEnqueue(
-              branch2[kState].controller,
-              value2);
-          }
-        });
+        // The microtask is required by the spec (ReadableStreamTee's
+        // "chunk steps" queue one).
+        pendingChunk = value;
+        queueMicrotask(forwardChunk);
       },
       [kClose]() {
         // The `process.nextTick()` is not part of the spec.
@@ -1920,6 +1934,55 @@ function readableByteStreamTee(stream) {
     );
   }
 
+  // As in readableStreamDefaultTee, only one read is ever in flight, so
+  // the default-reader read request and its forwarding microtask are
+  // shared across all chunks.
+  let pendingChunk;
+  function forwardChunk() {
+    readAgainForBranch1 = false;
+    readAgainForBranch2 = false;
+    const chunk1 = pendingChunk;
+    let chunk2 = pendingChunk;
+    pendingChunk = undefined;
+
+    if (!canceled1 && !canceled2) {
+      try {
+        chunk2 = cloneAsUint8Array(chunk1);
+      } catch (error) {
+        readableByteStreamControllerError(
+          branch1[kState].controller,
+          error,
+        );
+        readableByteStreamControllerError(
+          branch2[kState].controller,
+          error,
+        );
+        cancelDeferred.resolve(readableStreamCancel(stream, error));
+        return;
+      }
+    }
+    if (!canceled1) {
+      readableByteStreamControllerEnqueue(
+        branch1[kState].controller,
+        chunk1,
+      );
+    }
+    if (!canceled2) {
+      readableByteStreamControllerEnqueue(
+        branch2[kState].controller,
+        chunk2,
+      );
+    }
+    reading = false;
+
+    if (readAgainForBranch1) {
+      pull1Algorithm();
+    } else if (readAgainForBranch2) {
+      pull2Algorithm();
+    }
+  }
+
+  let defaultReadRequest;
   function pullWithDefaultReader() {
     if (isReadableStreamBYOBReader(reader)) {
       readableStreamBYOBReaderRelease(reader);
@@ -1927,50 +1990,10 @@ function readableByteStreamTee(stream) {
       forwardReaderError(reader);
     }
 
-    const readRequest = {
+    defaultReadRequest ??= {
       [kChunk](chunk) {
-        queueMicrotask(() => {
-          readAgainForBranch1 = false;
-          readAgainForBranch2 = false;
-          const chunk1 = chunk;
-          let chunk2 = chunk;
-
-          if (!canceled1 && !canceled2) {
-            try {
-              chunk2 = cloneAsUint8Array(chunk);
-            } catch (error) {
-              readableByteStreamControllerError(
-                branch1[kState].controller,
-                error,
-              );
-              readableByteStreamControllerError(
-                branch2[kState].controller,
-                error,
-              );
-              cancelDeferred.resolve(readableStreamCancel(stream, error));
-              return;
-            }
-          }
-          if (!canceled1) {
-            readableByteStreamControllerEnqueue(
-              branch1[kState].controller,
-              chunk1,
-            );
-          }
-          if (!canceled2) {
-            readableByteStreamControllerEnqueue(
-              branch2[kState].controller,
-              chunk2,
-            );
-          }
-          reading = false;
-
-          if (readAgainForBranch1) {
-            pull1Algorithm();
-          } else if (readAgainForBranch2) {
-            pull2Algorithm();
-          }
-        });
+        pendingChunk = chunk;
+        queueMicrotask(forwardChunk);
       },
       [kClose]() {
         reading = false;
@@ -1995,8 +2018,7 @@ function readableByteStreamTee(stream) {
         reading = false;
       },
     };
-
-    readableStreamDefaultReaderRead(reader, readRequest);
+    readableStreamDefaultReaderRead(reader, defaultReadRequest);
   }
 
   function pullWithBYOBReader(view, forBranch2) {
@@ -2820,7 +2842,7 @@ function readableByteStreamControllerClose(controller) {
   }
 
   if (pendingPullIntos.length) {
-    const firstPendingPullInto = pendingPullIntos[0];
+    const firstPendingPullInto = pendingPullIntos.peek();
     if (firstPendingPullInto.bytesFilled % firstPendingPullInto.elementSize !== 0) {
       const error = new ERR_INVALID_STATE.TypeError('Partial read');
       readableByteStreamControllerError(controller, error);
@@ -2877,7 +2899,7 @@ function readableByteStreamControllerClearAlgorithms(controller) {
 
 function readableByteStreamControllerClearPendingPullIntos(controller) {
   readableByteStreamControllerInvalidateBYOBRequest(controller);
-  controller[kState].pendingPullIntos = [];
+  controller[kState].pendingPullIntos = kEmptyQueue;
 }
 
 function readableByteStreamControllerGetDesiredSize(controller) {
@@ -2979,7 +3001,7 @@ function readableByteStreamControllerPullInto(
     type: 'byob',
   };
   if (pendingPullIntos.length) {
-    ArrayPrototypePush(pendingPullIntos, desc);
+    pendingPullIntos.push(desc);
     readableStreamAddReadIntoRequest(stream, readIntoRequest);
     return;
   }
@@ -3005,17 +3027,28 @@ function readableByteStreamControllerPullInto(
       return;
     }
   }
-  ArrayPrototypePush(pendingPullIntos, desc);
+  materializePendingPullIntos(controller[kState]).push(desc);
   readableStreamAddReadIntoRequest(stream, readIntoRequest);
   readableByteStreamControllerCallPullIfNeeded(controller);
 }
 
+// Pending pull-into descriptor queues start out as (and are reset to)
+// the shared immutable empty queue so that constructing a byte stream
+// never allocates descriptor storage; the two push sites that can
+// observe an empty queue materialize a real Queue on first use.
+function materializePendingPullIntos(state) {
+  const pendingPullIntos = state.pendingPullIntos;
+  if (pendingPullIntos === kEmptyQueue)
+    return state.pendingPullIntos = new Queue();
+  return pendingPullIntos;
+}
+
 function readableByteStreamControllerRespondInternal(controller, bytesWritten) {
   const {
     stream,
     pendingPullIntos,
   } = controller[kState];
-  const desc = pendingPullIntos[0];
+  const desc = pendingPullIntos.peek();
   readableByteStreamControllerInvalidateBYOBRequest(controller);
   if (stream[kState].state === 'closed') {
     if (bytesWritten)
@@ -3040,7 +3073,7 @@ function readableByteStreamControllerRespond(controller, bytesWritten) {
     stream,
   } = controller[kState];
   assert(pendingPullIntos.length);
-  const desc = pendingPullIntos[0];
+  const desc = pendingPullIntos.peek();
 
   if (stream[kState].state === 'closed') {
     if (bytesWritten !== 0)
@@ -3085,7 +3118,7 @@ function readableByteStreamControllerFillHeadPullIntoDescriptor(
     pendingPullIntos,
     byobRequest,
   } = controller[kState];
-  assert(!pendingPullIntos.length || pendingPullIntos[0] === desc);
+  assert(!pendingPullIntos.length || pendingPullIntos.peek() === desc);
   assert(byobRequest === null);
   desc.bytesFilled += size;
 }
@@ -3108,7 +3141,7 @@ function readableByteStreamControllerEnqueue(controller, chunk) {
   const transferredBuffer = ArrayBufferPrototypeTransfer(buffer);
 
   if (pendingPullIntos.length) {
-    const firstPendingPullInto = pendingPullIntos[0];
+    const firstPendingPullInto = pendingPullIntos.peek();
 
     if (ArrayBufferPrototypeGetDetached(firstPendingPullInto.buffer)) {
       throw new ERR_INVALID_STATE.TypeError(
@@ -3155,7 +3188,7 @@ function readableByteStreamControllerEnqueue(controller, chunk) {
     } else {
       assert(!queue.length);
       if (pendingPullIntos.length) {
-        assert(pendingPullIntos[0].type === 'default');
+        assert(pendingPullIntos.peek().type === 'default');
         readableByteStreamControllerShiftPendingPullInto(controller);
       }
       const transferredView =
@@ -3324,7 +3357,7 @@ function readableByteStreamControllerProcessPullIntoDescriptorsUsingQueue(
   while (pendingPullIntos.length) {
     if (!controller[kState].queueTotalSize)
       break;
-    const desc = pendingPullIntos[0];
+    const desc = pendingPullIntos.peek();
     if (readableByteStreamControllerFillPullIntoDescriptorFromQueue(
       controller,
       desc)) {
@@ -3404,7 +3437,7 @@ function readableByteStreamControllerRespondWithNewView(controller, view) {
   } = controller[kState];
   assert(pendingPullIntos.length);
 
-  const desc = pendingPullIntos[0];
+  const desc = pendingPullIntos.peek();
   assert(stream[kState].state !== 'errored');
 
   const viewByteLength = ArrayBufferViewGetByteLength(view);
@@ -3444,7 +3477,7 @@ function readableByteStreamControllerRespondWithNewView(controller, view) {
 
 function readableByteStreamControllerShiftPendingPullInto(controller) {
   assert(controller[kState].byobRequest === null);
-  return ArrayPrototypeShift(controller[kState].pendingPullIntos);
+  return controller[kState].pendingPullIntos.shift();
 }
 
 function readableByteStreamControllerCallPullIfNeeded(controller) {
@@ -3540,7 +3573,6 @@ function readableByteStreamControllerProcessReadRequestsUsingQueue(controller) {
 
 function readableByteStreamControllerPullSteps(controller, readRequest) {
   const {
-    pendingPullIntos,
     queueTotalSize,
     stream,
   } = controller[kState];
@@ -3559,8 +3591,7 @@ function readableByteStreamControllerPullSteps(controller, readRequest) {
   if (autoAllocateChunkSize !== undefined) {
     try {
       const buffer = new ArrayBuffer(autoAllocateChunkSize);
-      ArrayPrototypePush(
-        pendingPullIntos,
+      materializePendingPullIntos(controller[kState]).push(
         {
           buffer,
           bufferByteLength: autoAllocateChunkSize,
@@ -3610,7 +3641,7 @@ function setupReadableByteStreamController(
     pullAlgorithm,
     cancelAlgorithm,
     autoAllocateChunkSize,
-    pendingPullIntos: [],
+    pendingPullIntos: kEmptyQueue,
   };
   stream[kState].controller = controller;
 

test/parallel/test-whatwg-readablestream.js

@@ -37,7 +37,8 @@ const {
 } = require('internal/webstreams/readablestream');
 
 const {
-  kState
+  kState,
+  Queue,
 } = require('internal/webstreams/util');
 
 const {
@@ -1581,7 +1582,9 @@ class Source {
     start(c) { controller = c; }
   });
 
-  controller[kState].pendingPullIntos = [{}];
+  const pendingPullIntos = new Queue();
+  pendingPullIntos.push({});
+  controller[kState].pendingPullIntos = pendingPullIntos;
   assert.throws(() => readableByteStreamControllerRespond(controller, 0), {
     code: 'ERR_INVALID_ARG_VALUE',
   });