From 91e15d208130a21f995aae0be782cffe47f7ff19 Mon Sep 17 00:00:00 2001 From: Matteo Collina Date: Wed, 29 Jul 2026 14:25:10 +0200 Subject: [PATCH 1/2] 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 --- lib/internal/webstreams/readablestream.js | 50 ++++++++++++--------- test/parallel/test-whatwg-readablestream.js | 7 ++- 2 files changed, 35 insertions(+), 22 deletions(-) diff --git a/lib/internal/webstreams/readablestream.js b/lib/internal/webstreams/readablestream.js index 9b6dc4b4ad216e..a8a88361705851 100644 --- a/lib/internal/webstreams/readablestream.js +++ b/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; } } @@ -2820,7 +2821,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 +2878,7 @@ function readableByteStreamControllerClearAlgorithms(controller) { function readableByteStreamControllerClearPendingPullIntos(controller) { readableByteStreamControllerInvalidateBYOBRequest(controller); - controller[kState].pendingPullIntos = []; + controller[kState].pendingPullIntos = kEmptyQueue; } function readableByteStreamControllerGetDesiredSize(controller) { @@ -2979,7 +2980,7 @@ function readableByteStreamControllerPullInto( type: 'byob', }; if (pendingPullIntos.length) { - ArrayPrototypePush(pendingPullIntos, desc); + pendingPullIntos.push(desc); readableStreamAddReadIntoRequest(stream, readIntoRequest); return; } @@ -3005,17 +3006,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 +3052,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 +3097,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 +3120,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 +3167,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 +3336,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 +3416,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 +3456,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 +3552,6 @@ function readableByteStreamControllerProcessReadRequestsUsingQueue(controller) { function readableByteStreamControllerPullSteps(controller, readRequest) { const { - pendingPullIntos, queueTotalSize, stream, } = controller[kState]; @@ -3559,8 +3570,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 +3620,7 @@ function setupReadableByteStreamController( pullAlgorithm, cancelAlgorithm, autoAllocateChunkSize, - pendingPullIntos: [], + pendingPullIntos: kEmptyQueue, }; stream[kState].controller = controller; diff --git a/test/parallel/test-whatwg-readablestream.js b/test/parallel/test-whatwg-readablestream.js index d1001d51b4e8d1..0105d4bf723b67 100644 --- a/test/parallel/test-whatwg-readablestream.js +++ b/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', }); From 30a58345c4aad1223e7c4dbdc6340d36574a4139 Mon Sep 17 00:00:00 2001 From: Matteo Collina Date: Wed, 29 Jul 2026 14:31:48 +0200 Subject: [PATCH 2/2] stream: reuse the tee read request across chunks Both readableStreamDefaultTee and the byte tee's default-reader pull path allocated a fresh read request object (three closures) plus a fresh forwarding microtask closure for every chunk. Only one read is ever in flight per tee (guarded by the `reading` flag), so a single read request and forwarding function can be shared across all chunks, with the chunk handed over through a captured slot. Add a steady-state tee benchmark covering the default and byte paths. Signed-off-by: Matteo Collina --- benchmark/webstreams/tee.js | 44 +++++++ lib/internal/webstreams/readablestream.js | 149 ++++++++++++---------- 2 files changed, 129 insertions(+), 64 deletions(-) create mode 100644 benchmark/webstreams/tee.js diff --git a/benchmark/webstreams/tee.js b/benchmark/webstreams/tee.js new file mode 100644 index 00000000000000..2b1b4802dd9223 --- /dev/null +++ b/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); +} diff --git a/lib/internal/webstreams/readablestream.js b/lib/internal/webstreams/readablestream.js index a8a88361705851..b2516695761434 100644 --- a/lib/internal/webstreams/readablestream.js +++ b/lib/internal/webstreams/readablestream.js @@ -1805,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. @@ -1921,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); @@ -1928,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; @@ -1996,8 +2018,7 @@ function readableByteStreamTee(stream) { reading = false; }, }; - - readableStreamDefaultReaderRead(reader, readRequest); + readableStreamDefaultReaderRead(reader, defaultReadRequest); } function pullWithBYOBReader(view, forBranch2) {