From e16a18d166a873e1aad99d15bdf0ad760d623993 Mon Sep 17 00:00:00 2001 From: Matteo Collina Date: Mon, 6 Jul 2026 10:45:39 +0200 Subject: [PATCH 1/2] stream: speed up reads and iteration over default WHATWG streams Skip the size algorithm call and its result validation when the size algorithm is the default one, and stop re-running the full ShouldCallPull predicate at per-chunk call sites where its inputs are already established. readable-async-iterator type=normal: +16.6% (***) readable-read-buffered: +13.9% to +27.5% (**/***) pipe-to: +3.7% to +6.0% (15/16 configs significant) Signed-off-by: Matteo Collina --- lib/internal/webstreams/readablestream.js | 60 ++++++++++++++++++++--- 1 file changed, 53 insertions(+), 7 deletions(-) diff --git a/lib/internal/webstreams/readablestream.js b/lib/internal/webstreams/readablestream.js index a9d57ddeeb1614..512a3d35aaa6ed 100644 --- a/lib/internal/webstreams/readablestream.js +++ b/lib/internal/webstreams/readablestream.js @@ -597,8 +597,12 @@ class ReadableStream { !controller[kState].queue.length) { readableStreamDefaultControllerClearAlgorithms(controller); readableStreamClose(stream); - } else { - readableStreamDefaultControllerCallPullIfNeeded(controller); + } else if (!controller[kState].closeRequested && + controller[kState].started && + controller[kState].highWaterMark - + controller[kState].queueTotalSize > 0) { + // Reduced ShouldCallPull, as in the read() fast path. + readableStreamDefaultControllerPull(controller); } return PromiseResolve({ done: false, value: chunk }); @@ -947,8 +951,14 @@ class ReadableStreamDefaultReader { if (controller[kState].closeRequested && !controller[kState].queue.length) { readableStreamDefaultControllerClearAlgorithms(controller); readableStreamClose(stream); - } else { - readableStreamDefaultControllerCallPullIfNeeded(controller); + } else if (!controller[kState].closeRequested && + controller[kState].started && + controller[kState].highWaterMark - + controller[kState].queueTotalSize > 0) { + // ShouldCallPull reduced to the conditions not already + // established on this path: the state is readable and the + // queue was non-empty, so no read requests can be parked. + readableStreamDefaultControllerPull(controller); } return PromiseResolve({ done: false, value: chunk }); @@ -2522,8 +2532,27 @@ function readableStreamDefaultControllerEnqueue(controller, chunk) { reader[kState] !== undefined && reader[kType] === 'ReadableStreamDefaultReader' && reader[kState].readRequests.length) { + // Fulfilling a read request can run user code synchronously (the + // pipeTo read request invokes the sink's write algorithm), so the + // full ShouldCallPull predicate has to be re-evaluated afterwards. readableStreamFulfillReadRequest(stream, chunk, false); + } else if (controllerState.sizeAlgorithm === defaultSizeAlgorithm) { + // The default size algorithm always returns 1, so the call and the + // size validation in enqueueValueWithSize can be skipped entirely. + // No user code runs between the guards at the top of this function + // and this point, so ShouldCallPull reduces to the started flag and + // the desired size (this branch implies no parked read requests, + // ruling out the reader arm of the predicate). + ArrayPrototypePush(controllerState.queue, { value: chunk, size: 1 }); + controllerState.queueTotalSize++; + if (controllerState.started && + controllerState.highWaterMark - controllerState.queueTotalSize > 0) { + readableStreamDefaultControllerPull(controller); + } + return; } else { + // The user-supplied size algorithm may run arbitrary code, so the + // full ShouldCallPull predicate has to be re-evaluated afterwards. try { const chunkSize = FunctionPrototypeCall( @@ -2592,6 +2621,13 @@ function readableStreamDefaultControllerShouldCallPull(controller) { function readableStreamDefaultControllerCallPullIfNeeded(controller) { if (!readableStreamDefaultControllerShouldCallPull(controller)) return; + readableStreamDefaultControllerPull(controller); +} + +// The ShouldCallPull half of CallPullIfNeeded, split out so that callers +// that have already established the predicate from state in scope (the +// enqueue path above) can skip re-running it. +function readableStreamDefaultControllerPull(controller) { if (controller[kState].pulling) { controller[kState].pullAgain = true; return; @@ -2652,14 +2688,24 @@ function readableStreamDefaultControllerPullSteps(controller, readRequest) { if (controller[kState].closeRequested && !queue.length) { readableStreamDefaultControllerClearAlgorithms(controller); readableStreamClose(stream); - } else { - readableStreamDefaultControllerCallPullIfNeeded(controller); + } else if (!controller[kState].closeRequested && + controller[kState].started && + controller[kState].highWaterMark - + controller[kState].queueTotalSize > 0) { + // Reduced ShouldCallPull: the state is known to be readable and the + // queue was non-empty, so no read requests can be parked. + readableStreamDefaultControllerPull(controller); } readRequest[kChunk](chunk); return; } readableStreamAddReadRequest(stream, readRequest); - readableStreamDefaultControllerCallPullIfNeeded(controller); + // Reduced ShouldCallPull: the state is known to be readable, an empty + // queue in that state implies close has not been requested (close with + // an empty queue closes the stream immediately), and the read request + // parked above already satisfies the reader-with-pending-reads arm. + if (controller[kState].started) + readableStreamDefaultControllerPull(controller); } function setupReadableStreamDefaultController( From 8b43ce8b60a060a883a99c1489e8185d5356f611 Mon Sep 17 00:00:00 2001 From: Matteo Collina Date: Wed, 8 Jul 2026 12:24:41 +0200 Subject: [PATCH 2/2] stream: drain WHATWG stream queues from a head index The default readable and writable controller queues were drained with ArrayPrototypeShift, which is O(queue length); reading a buffered queue one chunk at a time is therefore O(n^2). Dequeue through a moving head index instead so each read is O(1), dropping the consumed prefix in amortized O(1) once it reaches half of the backing array (past a small floor, so a short oscillating queue never pays a per-dequeue array mutation). readable-read-buffered: +14.8% to +35.4% (**/***) readable-async-iterator type=normal: +17.8% (***) pipe-to: +12.9% to +18.1% (all 16 configs ***) Signed-off-by: Matteo Collina --- lib/internal/webstreams/readablestream.js | 41 +++++++++++++---------- lib/internal/webstreams/util.js | 30 +++++++++++++---- lib/internal/webstreams/writablestream.js | 10 +++--- 3 files changed, 52 insertions(+), 29 deletions(-) diff --git a/lib/internal/webstreams/readablestream.js b/lib/internal/webstreams/readablestream.js index 512a3d35aaa6ed..eb5f8c83b6e938 100644 --- a/lib/internal/webstreams/readablestream.js +++ b/lib/internal/webstreams/readablestream.js @@ -589,12 +589,12 @@ class ReadableStream { stream[kState].state === 'readable') { const controller = stream[kState].controller; if (isReadableStreamDefaultController(controller)) { - if (controller[kState].queue.length > 0) { + if (controller[kState].queueHead < controller[kState].queue.length) { stream[kState].disturbed = true; const chunk = dequeueValue(controller); if (controller[kState].closeRequested && - !controller[kState].queue.length) { + controller[kState].queueHead === controller[kState].queue.length) { readableStreamDefaultControllerClearAlgorithms(controller); readableStreamClose(stream); } else if (!controller[kState].closeRequested && @@ -944,11 +944,12 @@ class ReadableStreamDefaultReader { // Promise.resolve() callbacks still run in the microtask queue. if (stream[kState].state === 'readable') { if (isReadableStreamDefaultController(controller)) { - if (controller[kState].queue.length > 0) { + if (controller[kState].queueHead < controller[kState].queue.length) { stream[kState].disturbed = true; const chunk = dequeueValue(controller); - if (controller[kState].closeRequested && !controller[kState].queue.length) { + if (controller[kState].closeRequested && + controller[kState].queueHead === controller[kState].queue.length) { readableStreamDefaultControllerClearAlgorithms(controller); readableStreamClose(stream); } else if (!controller[kState].closeRequested && @@ -1650,14 +1651,15 @@ function readableStreamPipeTo( // reduces promise allocation overhead. if (source[kState].state === 'readable' && isReadableStreamDefaultController(controller) && - controller[kState].queue.length > 0) { + controller[kState].queueHead < controller[kState].queue.length) { - while (controller[kState].queue.length > 0) { + while (controller[kState].queueHead < controller[kState].queue.length) { if (shuttingDown) return true; const chunk = dequeueValue(controller); - if (controller[kState].closeRequested && !controller[kState].queue.length) { + if (controller[kState].closeRequested && + controller[kState].queueHead === controller[kState].queue.length) { readableStreamDefaultControllerClearAlgorithms(controller); readableStreamClose(source); } @@ -2511,7 +2513,7 @@ function readableStreamDefaultControllerClose(controller) { if (!readableStreamDefaultControllerCanCloseOrEnqueue(controller)) return; controller[kState].closeRequested = true; - if (!controller[kState].queue.length) { + if (controller[kState].queueHead === controller[kState].queue.length) { readableStreamDefaultControllerClearAlgorithms(controller); readableStreamClose(controller[kState].stream); } @@ -2679,19 +2681,20 @@ function readableStreamDefaultControllerCancelSteps(controller, reason) { } function readableStreamDefaultControllerPullSteps(controller, readRequest) { - const { - stream, - queue, - } = controller[kState]; - if (queue.length) { + const controllerState = controller[kState]; + const stream = controllerState.stream; + // `queue` may be replaced by dequeueValue (queue compaction), so the queue + // and head are read from the state each time rather than bound to a local. + if (controllerState.queueHead < controllerState.queue.length) { const chunk = dequeueValue(controller); - if (controller[kState].closeRequested && !queue.length) { + if (controllerState.closeRequested && + controllerState.queueHead === controllerState.queue.length) { readableStreamDefaultControllerClearAlgorithms(controller); readableStreamClose(stream); - } else if (!controller[kState].closeRequested && - controller[kState].started && - controller[kState].highWaterMark - - controller[kState].queueTotalSize > 0) { + } else if (!controllerState.closeRequested && + controllerState.started && + controllerState.highWaterMark - + controllerState.queueTotalSize > 0) { // Reduced ShouldCallPull: the state is known to be readable and the // queue was non-empty, so no read requests can be parked. readableStreamDefaultControllerPull(controller); @@ -2727,6 +2730,7 @@ function setupReadableStreamDefaultController( pullFulfilled: undefined, pullRejected: undefined, queue: [], + queueHead: 0, queueTotalSize: 0, started: false, sizeAlgorithm, @@ -3594,6 +3598,7 @@ function setupReadableByteStreamController( started: false, stream, queue: [], + queueHead: 0, queueTotalSize: 0, highWaterMark, pullAlgorithm, diff --git a/lib/internal/webstreams/util.js b/lib/internal/webstreams/util.js index fb5e6a3a8fc7d0..708a5e0693107f 100644 --- a/lib/internal/webstreams/util.js +++ b/lib/internal/webstreams/util.js @@ -5,7 +5,7 @@ const { ArrayBufferPrototypeGetDetached, ArrayBufferPrototypeSlice, ArrayPrototypePush, - ArrayPrototypeShift, + ArrayPrototypeSlice, AsyncIteratorPrototype, DataViewPrototypeGetBuffer, DataViewPrototypeGetByteLength, @@ -44,8 +44,6 @@ const { getPromiseDetails, } = internalBinding('util'); -const assert = require('internal/assert'); - const { isDataView, } = require('internal/util/types'); @@ -157,27 +155,45 @@ function isBrandCheck(brand) { // single time and don't assert the existence of the queue fields (both // are unconditionally initialized during controller setup and only ever // replaced wholesale). +// +// The queue is drained from a moving head index instead of with +// ArrayPrototypeShift: shifting is O(queue length), so draining a buffered +// queue one chunk at a time is O(n^2). Reading through `queueHead` makes +// each dequeue O(1); the consumed prefix is dropped in amortized O(1) once +// it grows to at least half of the backing array (and past a small floor, +// so a short oscillating queue never pays a per-dequeue array mutation). +// Callers must treat `queue.length - queueHead`, not `queue.length`, as the +// number of live entries. function dequeueValue(controller) { const state = controller[kState]; - assert(state.queue.length); + const queue = state.queue; + const head = state.queueHead; const { value, size, - } = ArrayPrototypeShift(state.queue); + } = queue[head]; + queue[head] = undefined; + const newHead = head + 1; state.queueTotalSize = MathMax(0, state.queueTotalSize - size); + if (newHead >= 32 && newHead * 2 >= queue.length) { + state.queue = ArrayPrototypeSlice(queue, newHead); + state.queueHead = 0; + } else { + state.queueHead = newHead; + } return value; } function resetQueue(controller) { const state = controller[kState]; state.queue = []; + state.queueHead = 0; state.queueTotalSize = 0; } function peekQueueValue(controller) { const state = controller[kState]; - assert(state.queue.length); - return state.queue[0].value; + return state.queue[state.queueHead].value; } function enqueueValueWithSize(controller, value, size) { diff --git a/lib/internal/webstreams/writablestream.js b/lib/internal/webstreams/writablestream.js index 9a6af1aa4a1061..366c06d2fe214c 100644 --- a/lib/internal/webstreams/writablestream.js +++ b/lib/internal/webstreams/writablestream.js @@ -1141,14 +1141,14 @@ function writableStreamDefaultControllerProcessWrite(controller, chunk) { } function writableStreamDefaultControllerProcessClose(controller) { + const controllerState = controller[kState]; const { closeAlgorithm, - queue, stream, - } = controller[kState]; + } = controllerState; writableStreamMarkCloseRequestInFlight(stream); dequeueValue(controller); - assert(!queue.length); + assert(controllerState.queueHead === controllerState.queue.length); const sinkClosePromise = closeAlgorithm(); writableStreamDefaultControllerClearAlgorithms(controller); PromisePrototypeThen( @@ -1224,6 +1224,7 @@ function writableStreamDefaultControllerGetBackpressure(controller) { function writableStreamDefaultControllerAdvanceQueueIfNeeded(controller) { const { queue, + queueHead, started, stream, } = controller[kState]; @@ -1235,7 +1236,7 @@ function writableStreamDefaultControllerAdvanceQueueIfNeeded(controller) { return; } - if (!queue.length) + if (queueHead === queue.length) return; const value = peekQueueValue(controller); @@ -1294,6 +1295,7 @@ function setupWritableStreamDefaultController( closeAlgorithm, highWaterMark, queue: [], + queueHead: 0, queueTotalSize: 0, abortController: new AbortController(), sizeAlgorithm,