From 0ff7249b8f140de66af6a62cd565d91daa08da5d Mon Sep 17 00:00:00 2001 From: Matteo Collina Date: Mon, 6 Jul 2026 10:45:39 +0200 Subject: [PATCH] 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, 51 insertions(+), 9 deletions(-) diff --git a/lib/internal/webstreams/readablestream.js b/lib/internal/webstreams/readablestream.js index c5a2c437d9b78b..d0e3bfafc19fcd 100644 --- a/lib/internal/webstreams/readablestream.js +++ b/lib/internal/webstreams/readablestream.js @@ -599,8 +599,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 }); @@ -949,8 +953,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 }); @@ -2524,13 +2534,28 @@ 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 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); + // constant size instead of calling it or validating the result. + // 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). + materializeQueue(controllerState).pushPair(chunk, 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( @@ -2599,6 +2624,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; @@ -2659,14 +2691,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(