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; }