From ebef4895a4be36fe6f8317ce83d025ad2f208df2 Mon Sep 17 00:00:00 2001 From: Yagiz Nizipli Date: Sun, 5 Jul 2026 14:42:17 -0400 Subject: [PATCH 1/2] stream: hoist repeated loads in readable paths fromList() re-read state.length and each chunk's length several times per call and re-loaded buffer[idx] around every copy, and read() loaded state.length three times in its read(0) check; the engine cannot fold these loads across the intervening copy and slice calls. Cache them in locals instead. Ported from Bun's fork of the same functions (src/js/internal/streams/readable.ts fromList/read), which carries these hoists on top of the shared readable-stream lineage. benchmark/compare.js against the unmodified baseline (60-run capture plus an independent 30-run repeat, Welch t-test): streams/readable-unevenread +0.65% (p=4.3e-4) / +0.59% (p=5.9e-3) and streams/pipe.js +0.74% (p=2.0e-4) / +0.52% (p=8.8e-3), with no significant regression across the captured streams benchmarks. Refs: https://github.com/oven-sh/bun/blob/main/src/js/internal/streams/readable.ts Signed-off-by: Yagiz Nizipli Assisted-by: Grok (Grok Build) --- lib/internal/streams/readable.js | 107 ++++++++++++++++++------------- 1 file changed, 61 insertions(+), 46 deletions(-) diff --git a/lib/internal/streams/readable.js b/lib/internal/streams/readable.js index b8c07443e93e2a..b565d1d86b9892 100644 --- a/lib/internal/streams/readable.js +++ b/lib/internal/streams/readable.js @@ -679,15 +679,17 @@ Readable.prototype.read = function(n) { // If we're doing read(0) to trigger a readable event, but we // already have a bunch of data in the buffer, then just trigger - // the 'readable' event and move on. + // the 'readable' event and move on. `state.length` cannot change + // within this block, so it is loaded once instead of three times. + const stateLength = state.length; if (n === 0 && (state[kState] & kNeedReadable) !== 0 && ((state.highWaterMark !== 0 ? - state.length >= state.highWaterMark : - state.length > 0) || + stateLength >= state.highWaterMark : + stateLength > 0) || (state[kState] & kEnded) !== 0)) { debug('read: emitReadable'); - if (state.length === 0 && (state[kState] & kEnded) !== 0) + if (stateLength === 0 && (state[kState] & kEnded) !== 0) endReadable(this); else emitReadable(this); @@ -1635,8 +1637,14 @@ Readable._fromList = fromList; // This function is designed to be inlinable, so please take care when making // changes to the function body. function fromList(n, state) { + // `state.length` cannot change while this function runs (only the + // caller updates it, after this returns) and the chunk lengths feeding + // the copy loops cannot change across the copy calls, so every + // repeated property load below is hoisted into a local. + const stateLength = state.length; + // nothing buffered. - if (state.length === 0) + if (stateLength === 0) return null; let idx = state.bufferIndex; @@ -1648,7 +1656,7 @@ function fromList(n, state) { if ((state[kState] & kObjectMode) !== 0) { ret = buf[idx]; buf[idx++] = null; - } else if (!n || n >= state.length) { + } else if (!n || n >= stateLength) { // Read it all, truncate the list. if ((state[kState] & kDecoder) !== 0) { ret = ''; @@ -1662,61 +1670,68 @@ function fromList(n, state) { ret = buf[idx]; buf[idx++] = null; } else { - ret = Buffer.allocUnsafe(state.length); + ret = Buffer.allocUnsafe(stateLength); let i = 0; while (idx < len) { - TypedArrayPrototypeSet(ret, buf[idx], i); - i += buf[idx].length; + const data = buf[idx]; + TypedArrayPrototypeSet(ret, data, i); + i += data.length; buf[idx++] = null; } } - } else if (n < buf[idx].length) { - // `slice` is the same for buffers and strings. - ret = buf[idx].slice(0, n); - buf[idx] = buf[idx].slice(n); - } else if (n === buf[idx].length) { - // First chunk is a perfect match. - ret = buf[idx]; - buf[idx++] = null; - } else if ((state[kState] & kDecoder) !== 0) { - ret = ''; - while (idx < len) { - const str = buf[idx]; - if (n > str.length) { - ret += str; - n -= str.length; - buf[idx++] = null; - } else { - if (n === str.length) { + } else { + const first = buf[idx]; + const firstLength = first.length; + if (n < firstLength) { + // `slice` is the same for buffers and strings. + ret = first.slice(0, n); + buf[idx] = first.slice(n); + } else if (n === firstLength) { + // First chunk is a perfect match. + ret = first; + buf[idx++] = null; + } else if ((state[kState] & kDecoder) !== 0) { + ret = ''; + while (idx < len) { + const str = buf[idx]; + const strLength = str.length; + if (n > strLength) { ret += str; + n -= strLength; buf[idx++] = null; } else { - ret += str.slice(0, n); - buf[idx] = str.slice(n); + if (n === strLength) { + ret += str; + buf[idx++] = null; + } else { + ret += str.slice(0, n); + buf[idx] = str.slice(n); + } + break; } - break; } - } - } else { - ret = Buffer.allocUnsafe(n); - - const retLen = n; - while (idx < len) { - const data = buf[idx]; - if (n > data.length) { - TypedArrayPrototypeSet(ret, data, retLen - n); - n -= data.length; - buf[idx++] = null; - } else { - if (n === data.length) { + } else { + ret = Buffer.allocUnsafe(n); + + const retLen = n; + while (idx < len) { + const data = buf[idx]; + const dataLength = data.length; + if (n > dataLength) { TypedArrayPrototypeSet(ret, data, retLen - n); + n -= dataLength; buf[idx++] = null; } else { - TypedArrayPrototypeSet(ret, new FastBuffer(data.buffer, data.byteOffset, n), retLen - n); - buf[idx] = new FastBuffer(data.buffer, data.byteOffset + n, data.length - n); + if (n === dataLength) { + TypedArrayPrototypeSet(ret, data, retLen - n); + buf[idx++] = null; + } else { + TypedArrayPrototypeSet(ret, new FastBuffer(data.buffer, data.byteOffset, n), retLen - n); + buf[idx] = new FastBuffer(data.buffer, data.byteOffset + n, dataLength - n); + } + break; } - break; } } } From 9d0477c7082ca895e04de2622e29668495684a89 Mon Sep 17 00:00:00 2001 From: Yagiz Nizipli Date: Sun, 5 Jul 2026 14:42:32 -0400 Subject: [PATCH 2/2] 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 Assisted-by: Grok (Grok Build) --- lib/internal/webstreams/readablestream.js | 32 ++-- lib/internal/webstreams/util.js | 159 ++++++++++++++-- lib/internal/webstreams/writablestream.js | 8 +- .../test-webstreams-queue-wraparound.js | 178 ++++++++++++++++++ 4 files changed, 352 insertions(+), 25 deletions(-) create mode 100644 test/parallel/test-webstreams-queue-wraparound.js diff --git a/lib/internal/webstreams/readablestream.js b/lib/internal/webstreams/readablestream.js index a9d57ddeeb1614..c5a2c437d9b78b 100644 --- a/lib/internal/webstreams/readablestream.js +++ b/lib/internal/webstreams/readablestream.js @@ -113,9 +113,11 @@ const { extractSizeAlgorithm, getNonWritablePropertyDescriptor, isBrandCheck, + kEmptyQueue, kState, kType, lazyTransfer, + materializeQueue, nonOpCancel, nonOpPull, nonOpStart, @@ -2523,6 +2525,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 = @@ -2680,7 +2687,7 @@ function setupReadableStreamDefaultController( pulling: false, pullFulfilled: undefined, pullRejected: undefined, - queue: [], + queue: kEmptyQueue, queueTotalSize: 0, started: false, sizeAlgorithm, @@ -3155,14 +3162,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( @@ -3217,7 +3223,7 @@ function readableByteStreamControllerFillPullIntoDescriptorFromQueue( } = controller[kState]; while (totalBytesToCopyRemaining) { - const headOfQueue = queue[0]; + const headOfQueue = queue.peek(); const bytesToCopy = MathMin( totalBytesToCopyRemaining, headOfQueue.byteLength); @@ -3235,7 +3241,7 @@ function readableByteStreamControllerFillPullIntoDescriptorFromQueue( headOfQueue.byteOffset, bytesToCopy); if (headOfQueue.byteLength === bytesToCopy) { - ArrayPrototypeShift(queue); + queue.shift(); } else { headOfQueue.byteOffset += bytesToCopy; headOfQueue.byteLength -= bytesToCopy; @@ -3451,7 +3457,7 @@ function readableByteStreamControllerDequeueChunk(controller) { buffer, byteOffset, byteLength, - } = ArrayPrototypeShift(controller[kState].queue); + } = controller[kState].queue.shift(); controller[kState].queueTotalSize -= byteLength; readableByteStreamControllerHandleQueueDrain(controller); @@ -3547,7 +3553,7 @@ function setupReadableByteStreamController( pullRejected: undefined, started: false, stream, - queue: [], + queue: kEmptyQueue, queueTotalSize: 0, highWaterMark, pullAlgorithm, diff --git a/lib/internal/webstreams/util.js b/lib/internal/webstreams/util.js index fb5e6a3a8fc7d0..a27c61ed5e5db7 100644 --- a/lib/internal/webstreams/util.js +++ b/lib/internal/webstreams/util.js @@ -1,11 +1,10 @@ 'use strict'; const { + Array, ArrayBufferPrototypeGetByteLength, ArrayBufferPrototypeGetDetached, ArrayBufferPrototypeSlice, - ArrayPrototypePush, - ArrayPrototypeShift, AsyncIteratorPrototype, DataViewPrototypeGetBuffer, DataViewPrototypeGetByteLength, @@ -13,6 +12,7 @@ const { FunctionPrototypeCall, MathMax, NumberIsNaN, + ObjectFreeze, PromisePrototypeThen, PromiseReject, PromiseResolve, @@ -152,6 +152,143 @@ 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 @@ -159,25 +296,23 @@ function isBrandCheck(brand) { // 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) { @@ -188,7 +323,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; } @@ -284,9 +419,11 @@ module.exports = { getNonWritablePropertyDescriptor, isBrandCheck, isPromisePending, + kEmptyQueue, kState, kType, lazyTransfer, + materializeQueue, nonOpCancel, nonOpFlush, nonOpPull, diff --git a/lib/internal/webstreams/writablestream.js b/lib/internal/webstreams/writablestream.js index 9a6af1aa4a1061..53b339e47fa798 100644 --- a/lib/internal/webstreams/writablestream.js +++ b/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, diff --git a/test/parallel/test-webstreams-queue-wraparound.js b/test/parallel/test-webstreams-queue-wraparound.js new file mode 100644 index 00000000000000..15ffc59689b981 --- /dev/null +++ b/test/parallel/test-webstreams-queue-wraparound.js @@ -0,0 +1,178 @@ +'use strict'; + +const common = require('../common'); +const assert = require('assert'); + +// Regression test for the ring-buffer [[queue]] backing store used by the +// WHATWG stream controllers: exercises growth, wrap-around, drain-rewind +// and shrink behavior through the public API only. + +async function testDeepDefaultQueue() { + // Queue 3000 chunks before reading anything: the backing ring must grow + // (and re-linearize) repeatedly, then fully drain (shrink on rewind). + const rs = new ReadableStream({ + start(controller) { + for (let i = 0; i < 3000; i++) + controller.enqueue(i); + controller.close(); + }, + }, { highWaterMark: Infinity }); + + const reader = rs.getReader(); + for (let i = 0; i < 3000; i++) { + const { value, done } = await reader.read(); + assert.strictEqual(done, false); + assert.strictEqual(value, i); + } + const { done } = await reader.read(); + assert.strictEqual(done, true); +} + +async function testInterleavedWraparound() { + // Interleave enqueues and reads so head/tail wrap around the ring many + // times without ever growing it, including repeated empty transitions. + let enqueued = 0; + let controller; + const rs = new ReadableStream({ + start(c) { + controller = c; + }, + }); + + const reader = rs.getReader(); + let expected = 0; + for (let round = 0; round < 500; round++) { + const burst = (round % 3) + 1; + for (let i = 0; i < burst; i++) + controller.enqueue(enqueued++); + for (let i = 0; i < burst; i++) { + const { value, done } = await reader.read(); + assert.strictEqual(done, false); + assert.strictEqual(value, expected++); + } + } + assert.strictEqual(expected, enqueued); +} + +async function testCustomSizesSurviveQueueing() { + // Fractional and zero sizes must round-trip through the queue exactly + // (desiredSize accounting depends on the stored per-chunk size). + const sizes = [0.25, 0, 1.75, 3, 0.5]; + const chunks = ['a', 'b', 'c', 'd', 'e']; + let controller; + const rs = new ReadableStream({ + start(c) { + controller = c; + }, + }, { + highWaterMark: 10, + size(chunk) { + return sizes[chunks.indexOf(chunk)]; + }, + }); + + let expectedDesired = 10; + for (let i = 0; i < chunks.length; i++) { + controller.enqueue(chunks[i]); + expectedDesired -= sizes[i]; + assert.strictEqual(controller.desiredSize, expectedDesired); + } + const reader = rs.getReader(); + for (let i = 0; i < chunks.length; i++) { + const { value } = await reader.read(); + assert.strictEqual(value, chunks[i]); + expectedDesired += sizes[i]; + assert.strictEqual(controller.desiredSize, expectedDesired); + } +} + +async function testByteQueueUnevenReads() { + // Enqueue 7-byte chunks but read through 4-byte BYOB views: the byte + // queue's head record is partially consumed in place across reads. + const total = 7 * 300; + let written = 0; + const rs = new ReadableStream({ + type: 'bytes', + pull(controller) { + if (written >= total) { + controller.close(); + return; + } + const chunk = new Uint8Array(7); + for (let i = 0; i < 7; i++) + chunk[i] = (written + i) % 251; + written += 7; + controller.enqueue(chunk); + }, + }); + + const reader = rs.getReader({ mode: 'byob' }); + let read = 0; + while (read < total) { + const { value, done } = await reader.read(new Uint8Array(4)); + if (done) break; + for (let i = 0; i < value.length; i++) { + assert.strictEqual(value[i], (read + i) % 251, + `byte ${read + i} mismatch`); + } + read += value.length; + } + assert.strictEqual(read, total); +} + +async function testDeepByteQueueBurst() { + // Fill the byte queue with 2500 records before the default reader + // drains it, forcing ring growth and the drain-to-empty rewind. + const rs = new ReadableStream({ + type: 'bytes', + start(controller) { + for (let i = 0; i < 2500; i++) + controller.enqueue(new Uint8Array([i & 0xff])); + controller.close(); + }, + }); + + const reader = rs.getReader(); + for (let i = 0; i < 2500; i++) { + const { value, done } = await reader.read(); + assert.strictEqual(done, false); + assert.strictEqual(value[0], i & 0xff); + } + const { done } = await reader.read(); + assert.strictEqual(done, true); +} + +async function testWritableQueueBackpressureDrain() { + // Buffer thousands of writes behind a slow sink, then let it drain: + // the writable controller queue grows, wraps, and empties while every + // chunk is delivered in order. + const received = []; + let release; + const gate = new Promise((resolve) => { release = resolve; }); + const ws = new WritableStream({ + async write(chunk) { + if (received.length === 0) await gate; + received.push(chunk); + }, + }, { highWaterMark: 1 }); + + const writer = ws.getWriter(); + const writes = []; + for (let i = 0; i < 2000; i++) + writes.push(writer.write(i)); + release(); + await writer.close(); + await Promise.all(writes); + assert.strictEqual(received.length, 2000); + for (let i = 0; i < 2000; i++) + assert.strictEqual(received[i], i); +} + +(async () => { + await testDeepDefaultQueue(); + await testInterleavedWraparound(); + await testCustomSizesSurviveQueueing(); + await testByteQueueUnevenReads(); + await testDeepByteQueueBurst(); + await testWritableQueueBackpressureDrain(); +})().then(common.mustCall());