From e65f9befafb92366c0e3ec9c9e48de66e9b63148 Mon Sep 17 00:00:00 2001 From: Matteo Collina Date: Sat, 11 Jul 2026 21:54:20 +0200 Subject: [PATCH 1/2] stream: use the ring buffer for WHATWG stream request queues The [[queue]] backing the controllers became a ring buffer, but the read and write request queues (readRequests, readIntoRequests, writeRequests) were left as plain arrays consumed with ArrayPrototypeShift, which is O(n) and, even for the single pending request of the await-each regime, far slower than an indexed head advance. Back them with the same Queue, materialized lazily from the shared empty queue so acquiring a reader or constructing a writer allocates no request storage until a read or write actually parks. pipe-to: +4.7% to +9.4% (all 16 configs, ***) readable-read type=byob: +2.3% (**) parked read loop: +15%, write loop: +11% (local harness) Follow-up to https://github.com/nodejs/node/pull/64312. Signed-off-by: Matteo Collina --- lib/internal/webstreams/readablestream.js | 69 +++++++++++++---------- lib/internal/webstreams/util.js | 1 + lib/internal/webstreams/writablestream.js | 22 +++++--- 3 files changed, 54 insertions(+), 38 deletions(-) diff --git a/lib/internal/webstreams/readablestream.js b/lib/internal/webstreams/readablestream.js index d0e3bfafc19fcd..9b6dc4b4ad216e 100644 --- a/lib/internal/webstreams/readablestream.js +++ b/lib/internal/webstreams/readablestream.js @@ -101,6 +101,7 @@ const { ArrayBufferViewGetByteLength, ArrayBufferViewGetByteOffset, AsyncIterator, + Queue, canCopyArrayBuffer, cloneAsUint8Array, copyArrayBuffer, @@ -2196,9 +2197,9 @@ function readableStreamCancel(stream, reason) { reader, } = stream[kState]; if (reader !== undefined && readableStreamHasBYOBReader(stream)) { - for (let n = 0; n < reader[kState].readIntoRequests.length; n++) - reader[kState].readIntoRequests[n][kClose](); - reader[kState].readIntoRequests = []; + const readIntoRequests = reader[kState].readIntoRequests; + while (readIntoRequests.length) + readIntoRequests.shift()[kClose](); } return PromisePrototypeThen( @@ -2220,9 +2221,9 @@ function readableStreamClose(stream) { reader[kState].close?.resolve(); if (readableStreamHasDefaultReader(stream)) { - for (let n = 0; n < reader[kState].readRequests.length; n++) - reader[kState].readRequests[n][kClose](); - reader[kState].readRequests = []; + const readRequests = reader[kState].readRequests; + while (readRequests.length) + readRequests.shift()[kClose](); } } @@ -2250,14 +2251,14 @@ function readableStreamError(stream, error) { } if (readableStreamHasDefaultReader(stream)) { - for (let n = 0; n < reader[kState].readRequests.length; n++) - reader[kState].readRequests[n][kError](error); - reader[kState].readRequests = []; + const readRequests = reader[kState].readRequests; + while (readRequests.length) + readRequests.shift()[kError](error); } else { assert(readableStreamHasBYOBReader(stream)); - for (let n = 0; n < reader[kState].readIntoRequests.length; n++) - reader[kState].readIntoRequests[n][kError](error); - reader[kState].readIntoRequests = []; + const readIntoRequests = reader[kState].readIntoRequests; + while (readIntoRequests.length) + readIntoRequests.shift()[kError](error); } } @@ -2301,7 +2302,7 @@ function readableStreamFulfillReadRequest(stream, chunk, done) { reader, } = stream[kState]; assert(reader[kState].readRequests.length); - const readRequest = ArrayPrototypeShift(reader[kState].readRequests); + const readRequest = reader[kState].readRequests.shift(); // TODO(@jasnell): It's not clear under what exact conditions done // will be true here. The spec requires this check but none of the @@ -2319,7 +2320,7 @@ function readableStreamFulfillReadIntoRequest(stream, chunk, done) { reader, } = stream[kState]; assert(reader[kState].readIntoRequests.length); - const readIntoRequest = ArrayPrototypeShift(reader[kState].readIntoRequests); + const readIntoRequest = reader[kState].readIntoRequests.shift(); if (done) readIntoRequest[kClose](chunk); else @@ -2329,15 +2330,21 @@ function readableStreamFulfillReadIntoRequest(stream, chunk, done) { function readableStreamAddReadRequest(stream, readRequest) { assert(readableStreamHasDefaultReader(stream)); assert(stream[kState].state === 'readable'); - ArrayPrototypePush(stream[kState].reader[kState].readRequests, readRequest); + const readerState = stream[kState].reader[kState]; + let readRequests = readerState.readRequests; + if (readRequests === kEmptyQueue) + readRequests = readerState.readRequests = new Queue(); + readRequests.push(readRequest); } function readableStreamAddReadIntoRequest(stream, readIntoRequest) { assert(readableStreamHasBYOBReader(stream)); assert(stream[kState].state !== 'errored'); - ArrayPrototypePush( - stream[kState].reader[kState].readIntoRequests, - readIntoRequest); + const readerState = stream[kState].reader[kState]; + let readIntoRequests = readerState.readIntoRequests; + if (readIntoRequests === kEmptyQueue) + readIntoRequests = readerState.readIntoRequests = new Queue(); + readIntoRequests.push(readIntoRequest); } function readableStreamReaderGenericCancel(reader, reason) { @@ -2409,10 +2416,9 @@ function readableStreamDefaultReaderRelease(reader) { } function readableStreamDefaultReaderErrorReadRequests(reader, e) { - for (let n = 0; n < reader[kState].readRequests.length; ++n) { - reader[kState].readRequests[n][kError](e); - } - reader[kState].readRequests = []; + const readRequests = reader[kState].readRequests; + while (readRequests.length) + readRequests.shift()[kError](e); } function readableStreamBYOBReaderRelease(reader) { @@ -2424,10 +2430,9 @@ function readableStreamBYOBReaderRelease(reader) { } function readableStreamBYOBReaderErrorReadIntoRequests(reader, e) { - for (let n = 0; n < reader[kState].readIntoRequests.length; ++n) { - reader[kState].readIntoRequests[n][kError](e); - } - reader[kState].readIntoRequests = []; + const readIntoRequests = reader[kState].readIntoRequests; + while (readIntoRequests.length) + readIntoRequests.shift()[kError](e); } function readableStreamReaderGenericRelease(reader) { @@ -2499,14 +2504,18 @@ function setupReadableStreamBYOBReader(reader, stream) { if (!isReadableByteStreamController(controller)) throw new ERR_INVALID_ARG_VALUE('stream', stream, 'must be a byte stream'); readableStreamReaderGenericInitialize(reader, stream); - reader[kState].readIntoRequests = []; + // The read-request queues use the same ring buffer as [[queue]], drained + // from a moving head rather than with ArrayPrototypeShift. Start from the + // shared immutable empty queue so acquiring a reader allocates no request + // storage until a read actually parks. + reader[kState].readIntoRequests = kEmptyQueue; } function setupReadableStreamDefaultReader(reader, stream) { if (isReadableStreamLocked(stream)) throw new ERR_INVALID_STATE.TypeError('ReadableStream is locked'); readableStreamReaderGenericInitialize(reader, stream); - reader[kState].readRequests = []; + reader[kState].readRequests = kEmptyQueue; } function readableStreamDefaultControllerClose(controller) { @@ -3151,7 +3160,7 @@ function readableByteStreamControllerEnqueue(controller, chunk) { } const transferredView = new Uint8Array(transferredBuffer, byteOffset, byteLength); - const readRequest = ArrayPrototypeShift(readRequests); + const readRequest = readRequests.shift(); readRequest[kChunk](transferredView); } } else if (readableStreamHasBYOBReader(stream)) { @@ -3524,7 +3533,7 @@ function readableByteStreamControllerProcessReadRequestsUsingQueue(controller) { } readableByteStreamControllerFillReadRequestFromQueue( controller, - ArrayPrototypeShift(reader[kState].readRequests), + reader[kState].readRequests.shift(), ); } } diff --git a/lib/internal/webstreams/util.js b/lib/internal/webstreams/util.js index a27c61ed5e5db7..8bc4c02be31e95 100644 --- a/lib/internal/webstreams/util.js +++ b/lib/internal/webstreams/util.js @@ -404,6 +404,7 @@ module.exports = { ArrayBufferViewGetByteLength, ArrayBufferViewGetByteOffset, AsyncIterator, + Queue, canCopyArrayBuffer, cloneAsUint8Array, copyArrayBuffer, diff --git a/lib/internal/webstreams/writablestream.js b/lib/internal/webstreams/writablestream.js index 53b339e47fa798..32b526595e6fbd 100644 --- a/lib/internal/webstreams/writablestream.js +++ b/lib/internal/webstreams/writablestream.js @@ -1,8 +1,6 @@ 'use strict'; const { - ArrayPrototypePush, - ArrayPrototypeShift, FunctionPrototypeBind, FunctionPrototypeCall, ObjectDefineProperties, @@ -55,6 +53,7 @@ const { } = require('internal/worker/js_transferable'); const { + Queue, createPromiseCallbackNoParams, createPromiseCallback1Param, createPromiseCallback2Params, @@ -613,7 +612,10 @@ function createWritableStreamState() { controller: undefined, state: 'writable', storedError: undefined, - writeRequests: [], + // Ring-buffer request queue, materialized lazily on the first pending + // write (see writableStreamAddWriteRequest) so construction allocates + // no request storage. + writeRequests: kEmptyQueue, writer: undefined, transfer: { __proto__: null, @@ -823,7 +825,7 @@ function writableStreamRejectCloseAndClosedPromiseIfNeeded(stream) { function writableStreamMarkFirstWriteRequestInFlight(stream) { assert(stream[kState].inFlightWriteRequest.promise === undefined); assert(stream[kState].writeRequests.length); - const writeRequest = ArrayPrototypeShift(stream[kState].writeRequests); + const writeRequest = stream[kState].writeRequests.shift(); stream[kState].inFlightWriteRequest = writeRequest; } @@ -901,9 +903,9 @@ function writableStreamFinishErroring(stream) { stream[kState].state = 'errored'; stream[kState].controller[kError](); const storedError = stream[kState].storedError; - for (let n = 0; n < stream[kState].writeRequests.length; n++) - stream[kState].writeRequests[n].reject(storedError); - stream[kState].writeRequests = []; + const writeRequests = stream[kState].writeRequests; + while (writeRequests.length) + writeRequests.shift().reject(storedError); if (stream[kState].pendingAbortRequest.abort.promise === undefined) { writableStreamRejectCloseAndClosedPromiseIfNeeded(stream); @@ -952,7 +954,11 @@ function writableStreamAddWriteRequest(stream) { // PromiseWithResolvers() already returns a { promise, resolve, reject } // record, so push it as-is instead of rebuilding an identical object. const writeRequest = PromiseWithResolvers(); - ArrayPrototypePush(stream[kState].writeRequests, writeRequest); + const streamState = stream[kState]; + let writeRequests = streamState.writeRequests; + if (writeRequests === kEmptyQueue) + writeRequests = streamState.writeRequests = new Queue(); + writeRequests.push(writeRequest); return writeRequest.promise; } From 3cb9f689c4865325329c759c9844295da3cee073 Mon Sep 17 00:00:00 2001 From: Matteo Collina Date: Sun, 12 Jul 2026 17:10:48 +0200 Subject: [PATCH 2/2] stream: fold desired-size check into WHATWG backpressure update writableStreamUpdateBackpressure ran two release-mode assertions and took a precomputed backpressure value that every caller derived with a separate GetBackpressure/GetDesiredSize call pair. All callers already hold the 'writable', not-close-queued invariant the assertions checked, so derive the backpressure from the desired size inside the function and drop the assertions and the helper. This runs up to twice per write. writer.write() loop: +10%, pipe-to: +4.6% avg over 16 configs (local harness / benchmark, all configs positive). Signed-off-by: Matteo Collina --- lib/internal/webstreams/writablestream.js | 33 +++++++++-------------- 1 file changed, 12 insertions(+), 21 deletions(-) diff --git a/lib/internal/webstreams/writablestream.js b/lib/internal/webstreams/writablestream.js index 32b526595e6fbd..f0eb687be129d1 100644 --- a/lib/internal/webstreams/writablestream.js +++ b/lib/internal/webstreams/writablestream.js @@ -757,13 +757,15 @@ function writableStreamClose(stream) { return promise; } -function writableStreamUpdateBackpressure(stream, backpressure) { - const streamState = stream[kState]; - assert(streamState.state === 'writable'); - assert(!streamState.closeQueuedOrInFlight); - const { - writer, - } = streamState; +// Callers invoke this only in the 'writable' state with no close queued or +// in flight (the two conditions the spec's assertions check), so the +// backpressure value is derived here from the desired size instead of being +// computed and passed in by each caller. +function writableStreamUpdateBackpressure(controller, streamState) { + const controllerState = controller[kState]; + const backpressure = + controllerState.highWaterMark - controllerState.queueTotalSize <= 0; + const writer = streamState.writer; if (writer !== undefined && streamState.backpressure !== backpressure) { if (backpressure) { // The spec replaces [[readyPromise]] with a fresh pending promise; @@ -1100,9 +1102,7 @@ function writableStreamDefaultControllerWrite(controller, chunk, chunkSize) { const streamState = stream[kState]; if (!streamState.closeQueuedOrInFlight && streamState.state === 'writable') { - writableStreamUpdateBackpressure( - stream, - writableStreamDefaultControllerGetBackpressure(controller)); + writableStreamUpdateBackpressure(controller, streamState); } writableStreamDefaultControllerAdvanceQueueIfNeeded(controller); } @@ -1128,9 +1128,7 @@ function writableStreamDefaultControllerProcessWrite(controller, chunk) { dequeueValue(controller); if (!streamState.closeQueuedOrInFlight && state === 'writable') { - writableStreamUpdateBackpressure( - stream, - writableStreamDefaultControllerGetBackpressure(controller)); + writableStreamUpdateBackpressure(controller, streamState); } writableStreamDefaultControllerAdvanceQueueIfNeeded(controller); }; @@ -1229,10 +1227,6 @@ function writableStreamDefaultControllerClearAlgorithms(controller) { controller[kState].sizeAlgorithm = undefined; } -function writableStreamDefaultControllerGetBackpressure(controller) { - return writableStreamDefaultControllerGetDesiredSize(controller) <= 0; -} - function writableStreamDefaultControllerAdvanceQueueIfNeeded(controller) { const { queue, @@ -1317,9 +1311,7 @@ function setupWritableStreamDefaultController( }; stream[kState].controller = controller; - writableStreamUpdateBackpressure( - stream, - writableStreamDefaultControllerGetBackpressure(controller)); + writableStreamUpdateBackpressure(controller, stream[kState]); const startResult = startAlgorithm(); @@ -1402,7 +1394,6 @@ module.exports = { writableStreamDefaultControllerError, writableStreamDefaultControllerClose, writableStreamDefaultControllerClearAlgorithms, - writableStreamDefaultControllerGetBackpressure, writableStreamDefaultControllerAdvanceQueueIfNeeded, setupWritableStreamDefaultControllerFromSink, setupWritableStreamDefaultController,