From e65f9befafb92366c0e3ec9c9e48de66e9b63148 Mon Sep 17 00:00:00 2001 From: Matteo Collina Date: Sat, 11 Jul 2026 21:54:20 +0200 Subject: [PATCH] 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; }