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..f0eb687be129d1 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, @@ -755,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; @@ -823,7 +827,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 +905,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 +956,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; } @@ -1094,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); } @@ -1122,9 +1128,7 @@ function writableStreamDefaultControllerProcessWrite(controller, chunk) { dequeueValue(controller); if (!streamState.closeQueuedOrInFlight && state === 'writable') { - writableStreamUpdateBackpressure( - stream, - writableStreamDefaultControllerGetBackpressure(controller)); + writableStreamUpdateBackpressure(controller, streamState); } writableStreamDefaultControllerAdvanceQueueIfNeeded(controller); }; @@ -1223,10 +1227,6 @@ function writableStreamDefaultControllerClearAlgorithms(controller) { controller[kState].sizeAlgorithm = undefined; } -function writableStreamDefaultControllerGetBackpressure(controller) { - return writableStreamDefaultControllerGetDesiredSize(controller) <= 0; -} - function writableStreamDefaultControllerAdvanceQueueIfNeeded(controller) { const { queue, @@ -1311,9 +1311,7 @@ function setupWritableStreamDefaultController( }; stream[kState].controller = controller; - writableStreamUpdateBackpressure( - stream, - writableStreamDefaultControllerGetBackpressure(controller)); + writableStreamUpdateBackpressure(controller, stream[kState]); const startResult = startAlgorithm(); @@ -1396,7 +1394,6 @@ module.exports = { writableStreamDefaultControllerError, writableStreamDefaultControllerClose, writableStreamDefaultControllerClearAlgorithms, - writableStreamDefaultControllerGetBackpressure, writableStreamDefaultControllerAdvanceQueueIfNeeded, setupWritableStreamDefaultControllerFromSink, setupWritableStreamDefaultController,