stream: use ring buffer for WHATWG stream queues

The [[queue]] backing every default readable/writable controller was a
plain array of { value, size } wrappers consumed with
ArrayPrototypeShift, so each buffered chunk allocated a wrapper object
and each dequeue moved (or forced the engine to re-linearize) the
remaining elements; the byte controller queue paid the same shift cost
for its chunk descriptor records.

Replace the array with a power-of-two ring buffer. Default controller
queues store each entry as (value, size) in two consecutive slots, so
the per-chunk wrapper allocation disappears; the byte controller keeps
its descriptor records (they are mutated in place at the head) in
single slots. Controllers start from (and are reset to) a shared
immutable empty queue, so constructing a stream allocates no queue
storage until a chunk is actually buffered. Enqueues measured by the
internal default size algorithm (never observable by user code, always
returns 1, cannot throw) skip the algorithm call and its try/catch
entirely.

The layout mirrors what Bun/WebKit use for the same spec structure:
[[queue]] as a ring-buffer deque (WTF::Deque in Bun's
src/jsc/bindings/webcore/streams/StreamQueue.h), the pure-JS ring
buffer in Bun's src/js/internal/fifo.ts, and the trivial-size-algorithm
bypass in
src/jsc/bindings/webcore/streams/JSReadableStreamDefaultController.cpp.

benchmark/compare.js against the unmodified baseline (30-run capture
plus an independent 15-run repeat, Welch t-test, all p < 1e-5):
webstreams/pipe-to.js +12-17% across all sixteen high-water-mark
configurations, readable-read-buffered +20% (bufferSize=1) to +49%
(bufferSize=1000), readable-async-iterator +21%. No stable significant
regression across the rest of the webstreams suite: the creation.js and
readable-read.js deltas seen in the full-suite capture disappear in
isolated 60-run rechecks.

Refs: https://github.com/oven-sh/bun/blob/main/src/js/internal/fifo.ts
Signed-off-by: Yagiz Nizipli <yagiz@nizipli.com>

Author: anonrig

lib/internal/webstreams/readablestream.js

@@ -113,9 +113,11 @@ const {
   extractSizeAlgorithm,
   getNonWritablePropertyDescriptor,
   isBrandCheck,
+  kEmptyQueue,
   kState,
   kType,
   lazyTransfer,
+  materializeQueue,
   nonOpCancel,
   nonOpPull,
   nonOpStart,
@@ -2495,6 +2497,11 @@ function readableStreamDefaultControllerEnqueue(controller, chunk) {
       reader[kType] === 'ReadableStreamDefaultReader' &&
       reader[kState].readRequests.length) {
     readableStreamFulfillReadRequest(stream, chunk, false);
+  } else if (controllerState.sizeAlgorithm === defaultSizeAlgorithm) {
+    // The internal default size algorithm is never observable by user
+    // code, always returns 1, and cannot throw: enqueue with the
+    // constant instead of calling it.
+    enqueueValueWithSize(controller, chunk, 1);
   } else {
     try {
       const chunkSize =
@@ -2652,7 +2659,7 @@ function setupReadableStreamDefaultController(
     pulling: false,
     pullFulfilled: undefined,
     pullRejected: undefined,
-    queue: [],
+    queue: kEmptyQueue,
     queueTotalSize: 0,
     started: false,
     sizeAlgorithm,
@@ -3112,14 +3119,13 @@ function readableByteStreamControllerEnqueueChunkToQueue(
   buffer,
   byteOffset,
   byteLength) {
-  ArrayPrototypePush(
-    controller[kState].queue,
-    {
-      buffer,
-      byteOffset,
-      byteLength,
-    });
-  controller[kState].queueTotalSize += byteLength;
+  const state = controller[kState];
+  materializeQueue(state).push({
+    buffer,
+    byteOffset,
+    byteLength,
+  });
+  state.queueTotalSize += byteLength;
 }
 
 function readableByteStreamControllerEnqueueDetachedPullIntoToQueue(
@@ -3174,7 +3180,7 @@ function readableByteStreamControllerFillPullIntoDescriptorFromQueue(
   } = controller[kState];
 
   while (totalBytesToCopyRemaining) {
-    const headOfQueue = queue[0];
+    const headOfQueue = queue.peek();
     const bytesToCopy = MathMin(
       totalBytesToCopyRemaining,
       headOfQueue.byteLength);
@@ -3192,7 +3198,7 @@ function readableByteStreamControllerFillPullIntoDescriptorFromQueue(
       headOfQueue.byteOffset,
       bytesToCopy);
     if (headOfQueue.byteLength === bytesToCopy) {
-      ArrayPrototypeShift(queue);
+      queue.shift();
     } else {
       headOfQueue.byteOffset += bytesToCopy;
       headOfQueue.byteLength -= bytesToCopy;
@@ -3405,7 +3411,7 @@ function readableByteStreamControllerFillReadRequestFromQueue(controller, readRe
     buffer,
     byteOffset,
     byteLength,
-  } = ArrayPrototypeShift(queue);
+  } = queue.shift();
 
   controller[kState].queueTotalSize -= byteLength;
   readableByteStreamControllerHandleQueueDrain(controller);
@@ -3498,7 +3504,7 @@ function setupReadableByteStreamController(
     pullRejected: undefined,
     started: false,
     stream,
-    queue: [],
+    queue: kEmptyQueue,
     queueTotalSize: 0,
     highWaterMark,
     pullAlgorithm,

lib/internal/webstreams/util.js

@@ -1,15 +1,15 @@
 'use strict';
 
 const {
+  Array,
   ArrayBufferPrototypeGetByteLength,
   ArrayBufferPrototypeGetDetached,
   ArrayBufferPrototypeSlice,
-  ArrayPrototypePush,
-  ArrayPrototypeShift,
   AsyncIteratorPrototype,
   FunctionPrototypeCall,
   MathMax,
   NumberIsNaN,
+  ObjectFreeze,
   PromisePrototypeThen,
   PromiseReject,
   PromiseResolve,
@@ -134,32 +134,167 @@ function isBrandCheck(brand) {
   };
 }
 
+// Backing store for the spec's [[queue]]: a power-of-two ring buffer
+// instead of a plain array. Entries are pushed at the tail and consumed
+// at the head, so a plain array either moves every element or forces the
+// engine to re-linearize on each shift, and the default controllers would
+// additionally have to allocate a { value, size } wrapper object per
+// chunk just to keep the pair together. Default readable/writable
+// controller queues store each entry as (value, size) in two consecutive
+// slots via the *Pair methods; the readable byte controller queue stores
+// its chunk descriptor records in single slots via push/shift/peek. A
+// given instance only ever uses one of the two access patterns, so
+// head/tail stay aligned to the entry stride.
+class Queue {
+  constructor(listLength = 8) {
+    this.head = 0;
+    this.tail = 0;
+    // Number of logical entries currently in the queue: (value, size)
+    // pairs for default controller queues, descriptor records for byte
+    // controller queues.
+    this.length = 0;
+    this.capacityMask = listLength - 1;
+    this.list = new Array(listLength);
+    this.dequeuedSize = 0;
+  }
+
+  // Single-slot entries (readable byte controller chunk records).
+
+  push(entry) {
+    const tail = this.tail;
+    this.list[tail] = entry;
+    this.tail = (tail + 1) & this.capacityMask;
+    this.length++;
+    if (this.tail === this.head)
+      this.grow();
+  }
+
+  shift() {
+    const head = this.head;
+    const list = this.list;
+    const entry = list[head];
+    list[head] = undefined;
+    this.head = (head + 1) & this.capacityMask;
+    if (--this.length === 0)
+      this.rewind();
+    return entry;
+  }
+
+  peek() {
+    return this.list[this.head];
+  }
+
+  // Two-slot (value, size) entries (default controller queues). The
+  // stride is always 2 and capacities are even, so `tail + 1`/`head + 1`
+  // never need to wrap.
+
+  pushPair(value, size) {
+    const tail = this.tail;
+    const list = this.list;
+    list[tail] = value;
+    list[tail + 1] = size;
+    this.tail = (tail + 2) & this.capacityMask;
+    this.length++;
+    if (this.tail === this.head)
+      this.grow();
+  }
+
+  // Returns the dequeued value; the size of the same entry is left in
+  // `this.dequeuedSize` so that callers can update [[queueTotalSize]]
+  // without a per-entry wrapper object having to exist.
+  shiftPair() {
+    const head = this.head;
+    const list = this.list;
+    const value = list[head];
+    this.dequeuedSize = list[head + 1];
+    list[head] = undefined;
+    list[head + 1] = undefined;
+    this.head = (head + 2) & this.capacityMask;
+    if (--this.length === 0)
+      this.rewind();
+    return value;
+  }
+
+  peekPairValue() {
+    return this.list[this.head];
+  }
+
+  // The ring is completely full (the post-push tail caught up with the
+  // head): double the capacity, re-linearizing from the head so index
+  // arithmetic stays trivial.
+  grow() {
+    const list = this.list;
+    const capacity = list.length;
+    const head = this.head;
+    if (head !== 0) {
+      const relinearized = new Array(capacity * 2);
+      let n = 0;
+      for (let i = head; i < capacity; i++)
+        relinearized[n++] = list[i];
+      for (let i = 0; i < head; i++)
+        relinearized[n++] = list[i];
+      this.list = relinearized;
+      this.head = 0;
+    } else {
+      list.length = capacity * 2;
+    }
+    this.tail = capacity;
+    this.capacityMask = (capacity * 2) - 1;
+  }
+
+  // The queue just became empty: restart at slot 0 so shallow queues net
+  // sequential slot access, and drop the enlarged backing store after a
+  // large burst has fully drained.
+  rewind() {
+    this.head = 0;
+    this.tail = 0;
+    if (this.list.length > 1024) {
+      this.list.length = 8;
+      this.capacityMask = 0b111;
+    }
+  }
+}
+
+// Controllers start out with (and are reset to) this shared immutable
+// empty queue, so constructing a stream never allocates queue storage;
+// a real Queue is materialized by the enqueue paths on first use. All
+// dequeue/peek paths are guarded by `.length` (or the equivalent
+// [[queueTotalSize]]) checks, so they can never observe the sentinel in
+// a mutating way; it never stores entries, so it gets a zero-length
+// backing list.
+const kEmptyQueue = ObjectFreeze(new Queue(0));
+
+function materializeQueue(state) {
+  const queue = state.queue;
+  if (queue === kEmptyQueue)
+    return state.queue = new Queue();
+  return queue;
+}
+
 // The queue helpers below run once per chunk on the hot paths of every
 // default readable/writable stream, so they load the controller state a
 // single time and don't assert the existence of the queue fields (both
 // are unconditionally initialized during controller setup and only ever
 // replaced wholesale).
 function dequeueValue(controller) {
   const state = controller[kState];
-  assert(state.queue.length);
-  const {
-    value,
-    size,
-  } = ArrayPrototypeShift(state.queue);
-  state.queueTotalSize = MathMax(0, state.queueTotalSize - size);
+  const queue = state.queue;
+  assert(queue.length);
+  const value = queue.shiftPair();
+  state.queueTotalSize = MathMax(0, state.queueTotalSize - queue.dequeuedSize);
   return value;
 }
 
 function resetQueue(controller) {
   const state = controller[kState];
-  state.queue = [];
+  state.queue = kEmptyQueue;
   state.queueTotalSize = 0;
 }
 
 function peekQueueValue(controller) {
   const state = controller[kState];
   assert(state.queue.length);
-  return state.queue[0].value;
+  return state.queue.peekPairValue();
 }
 
 function enqueueValueWithSize(controller, value, size) {
@@ -170,7 +305,7 @@ function enqueueValueWithSize(controller, value, size) {
       coercedSize === Infinity) {
     throw new ERR_INVALID_ARG_VALUE.RangeError('size', size);
   }
-  ArrayPrototypePush(state.queue, { value, size: coercedSize });
+  materializeQueue(state).pushPair(value, coercedSize);
   state.queueTotalSize += coercedSize;
 }
 
@@ -266,9 +401,11 @@ module.exports = {
   getNonWritablePropertyDescriptor,
   isBrandCheck,
   isPromisePending,
+  kEmptyQueue,
   kState,
   kType,
   lazyTransfer,
+  materializeQueue,
   nonOpCancel,
   nonOpFlush,
   nonOpPull,

lib/internal/webstreams/writablestream.js

@@ -67,6 +67,7 @@ const {
   getNonWritablePropertyDescriptor,
   isBrandCheck,
   isPromisePending,
+  kEmptyQueue,
   kState,
   kType,
   lazyTransfer,
@@ -1177,6 +1178,11 @@ function writableStreamDefaultControllerGetChunkSize(controller, chunk) {
     return 1;
   }
 
+  // The internal default size algorithm is never observable by user
+  // code, always returns 1, and cannot throw: skip the call.
+  if (sizeAlgorithm === defaultSizeAlgorithm)
+    return 1;
+
   try {
     return FunctionPrototypeCall(
       sizeAlgorithm,
@@ -1293,7 +1299,7 @@ function setupWritableStreamDefaultController(
     abortAlgorithm,
     closeAlgorithm,
     highWaterMark,
-    queue: [],
+    queue: kEmptyQueue,
     queueTotalSize: 0,
     abortController: new AbortController(),
     sizeAlgorithm,