diff --git a/lib/internal/webstreams/readablestream.js b/lib/internal/webstreams/readablestream.js index a9d57ddeeb1614..4fabeb87eac492 100644 --- a/lib/internal/webstreams/readablestream.js +++ b/lib/internal/webstreams/readablestream.js @@ -119,6 +119,7 @@ const { nonOpCancel, nonOpPull, nonOpStart, + queueLength, rejectedHandledRecord, resetQueue, resolvedRecord, @@ -589,16 +590,16 @@ class ReadableStream { stream[kState].state === 'readable') { const controller = stream[kState].controller; if (isReadableStreamDefaultController(controller)) { - if (controller[kState].queue.length > 0) { + if (queueLength(controller) > 0) { stream[kState].disturbed = true; const chunk = dequeueValue(controller); if (controller[kState].closeRequested && - !controller[kState].queue.length) { + queueLength(controller) === 0) { readableStreamDefaultControllerClearAlgorithms(controller); readableStreamClose(stream); } else { - readableStreamDefaultControllerCallPullIfNeeded(controller); + readableStreamDefaultControllerPullIfNeededAfterDequeue(controller); } return PromiseResolve({ done: false, value: chunk }); @@ -940,15 +941,16 @@ class ReadableStreamDefaultReader { // Promise.resolve() callbacks still run in the microtask queue. if (stream[kState].state === 'readable') { if (isReadableStreamDefaultController(controller)) { - if (controller[kState].queue.length > 0) { + if (queueLength(controller) > 0) { stream[kState].disturbed = true; const chunk = dequeueValue(controller); - if (controller[kState].closeRequested && !controller[kState].queue.length) { + if (controller[kState].closeRequested && + queueLength(controller) === 0) { readableStreamDefaultControllerClearAlgorithms(controller); readableStreamClose(stream); } else { - readableStreamDefaultControllerCallPullIfNeeded(controller); + readableStreamDefaultControllerPullIfNeededAfterDequeue(controller); } return PromiseResolve({ done: false, value: chunk }); @@ -1640,14 +1642,15 @@ function readableStreamPipeTo( // reduces promise allocation overhead. if (source[kState].state === 'readable' && isReadableStreamDefaultController(controller) && - controller[kState].queue.length > 0) { + queueLength(controller) > 0) { - while (controller[kState].queue.length > 0) { + while (queueLength(controller) > 0) { if (shuttingDown) return true; const chunk = dequeueValue(controller); - if (controller[kState].closeRequested && !controller[kState].queue.length) { + if (controller[kState].closeRequested && + queueLength(controller) === 0) { readableStreamDefaultControllerClearAlgorithms(controller); readableStreamClose(source); } @@ -2501,7 +2504,7 @@ function readableStreamDefaultControllerClose(controller) { if (!readableStreamDefaultControllerCanCloseOrEnqueue(controller)) return; controller[kState].closeRequested = true; - if (!controller[kState].queue.length) { + if (queueLength(controller) === 0) { readableStreamDefaultControllerClearAlgorithms(controller); readableStreamClose(controller[kState].stream); } @@ -2522,8 +2525,27 @@ 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 default size algorithm always returns 1, so the call and the + // size validation in enqueueValueWithSize can be skipped entirely. + // 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). + ArrayPrototypePush(controllerState.queue, { value: chunk, size: 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( @@ -2592,6 +2614,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; @@ -2618,6 +2647,19 @@ function readableStreamDefaultControllerCallPullIfNeeded(controller) { controller[kState].pullRejected); } +// ShouldCallPull reduced for a caller that has just dequeued a chunk from a +// readable default controller: the state is known to be readable and the +// queue was non-empty, so no read requests can be parked and only the +// close / started / desired-size arms of the predicate remain. +function readableStreamDefaultControllerPullIfNeededAfterDequeue(controller) { + const state = controller[kState]; + if (!state.closeRequested && + state.started && + state.highWaterMark - state.queueTotalSize > 0) { + readableStreamDefaultControllerPull(controller); + } +} + function readableStreamDefaultControllerClearAlgorithms(controller) { controller[kState].pullAlgorithm = undefined; controller[kState].cancelAlgorithm = undefined; @@ -2643,23 +2685,25 @@ function readableStreamDefaultControllerCancelSteps(controller, reason) { } function readableStreamDefaultControllerPullSteps(controller, readRequest) { - const { - stream, - queue, - } = controller[kState]; - if (queue.length) { + const stream = controller[kState].stream; + if (queueLength(controller) > 0) { const chunk = dequeueValue(controller); - if (controller[kState].closeRequested && !queue.length) { + if (controller[kState].closeRequested && queueLength(controller) === 0) { readableStreamDefaultControllerClearAlgorithms(controller); readableStreamClose(stream); } else { - readableStreamDefaultControllerCallPullIfNeeded(controller); + readableStreamDefaultControllerPullIfNeededAfterDequeue(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( @@ -2681,6 +2725,7 @@ function setupReadableStreamDefaultController( pullFulfilled: undefined, pullRejected: undefined, queue: [], + queueHead: 0, queueTotalSize: 0, started: false, sizeAlgorithm, @@ -3548,6 +3593,7 @@ function setupReadableByteStreamController( started: false, stream, queue: [], + queueHead: 0, queueTotalSize: 0, highWaterMark, pullAlgorithm, diff --git a/lib/internal/webstreams/util.js b/lib/internal/webstreams/util.js index fb5e6a3a8fc7d0..00f21982b36831 100644 --- a/lib/internal/webstreams/util.js +++ b/lib/internal/webstreams/util.js @@ -5,7 +5,7 @@ const { ArrayBufferPrototypeGetDetached, ArrayBufferPrototypeSlice, ArrayPrototypePush, - ArrayPrototypeShift, + ArrayPrototypeSlice, AsyncIteratorPrototype, DataViewPrototypeGetBuffer, DataViewPrototypeGetByteLength, @@ -44,8 +44,6 @@ const { getPromiseDetails, } = internalBinding('util'); -const assert = require('internal/assert'); - const { isDataView, } = require('internal/util/types'); @@ -157,27 +155,50 @@ function isBrandCheck(brand) { // single time and don't assert the existence of the queue fields (both // are unconditionally initialized during controller setup and only ever // replaced wholesale). +// +// The queue is drained from a moving head index instead of with +// ArrayPrototypeShift: shifting is O(queue length), so draining a buffered +// queue one chunk at a time is O(n^2). Reading through `queueHead` makes +// each dequeue O(1); the consumed prefix is dropped in amortized O(1) once +// it grows to at least half of the backing array (and past a small floor, +// so a short oscillating queue never pays a per-dequeue array mutation). +// Callers must use queueLength(), not `queue.length`, for the number of +// live entries. +function queueLength(controller) { + const state = controller[kState]; + return state.queue.length - state.queueHead; +} + function dequeueValue(controller) { const state = controller[kState]; - assert(state.queue.length); + const queue = state.queue; + const head = state.queueHead; const { value, size, - } = ArrayPrototypeShift(state.queue); + } = queue[head]; + queue[head] = undefined; + const newHead = head + 1; state.queueTotalSize = MathMax(0, state.queueTotalSize - size); + if (newHead >= 32 && newHead * 2 >= queue.length) { + state.queue = ArrayPrototypeSlice(queue, newHead); + state.queueHead = 0; + } else { + state.queueHead = newHead; + } return value; } function resetQueue(controller) { const state = controller[kState]; state.queue = []; + state.queueHead = 0; state.queueTotalSize = 0; } function peekQueueValue(controller) { const state = controller[kState]; - assert(state.queue.length); - return state.queue[0].value; + return state.queue[state.queueHead].value; } function enqueueValueWithSize(controller, value, size) { @@ -293,6 +314,7 @@ module.exports = { nonOpStart, nonOpWrite, peekQueueValue, + queueLength, rejectedHandledRecord, resetQueue, resolvedRecord, diff --git a/lib/internal/webstreams/writablestream.js b/lib/internal/webstreams/writablestream.js index 9a6af1aa4a1061..5af6e84a3387df 100644 --- a/lib/internal/webstreams/writablestream.js +++ b/lib/internal/webstreams/writablestream.js @@ -74,6 +74,7 @@ const { nonOpStart, nonOpWrite, peekQueueValue, + queueLength, rejectedHandledRecord, resetQueue, resolvedRecord, @@ -1143,12 +1144,11 @@ function writableStreamDefaultControllerProcessWrite(controller, chunk) { function writableStreamDefaultControllerProcessClose(controller) { const { closeAlgorithm, - queue, stream, } = controller[kState]; writableStreamMarkCloseRequestInFlight(stream); dequeueValue(controller); - assert(!queue.length); + assert(queueLength(controller) === 0); const sinkClosePromise = closeAlgorithm(); writableStreamDefaultControllerClearAlgorithms(controller); PromisePrototypeThen( @@ -1223,7 +1223,6 @@ function writableStreamDefaultControllerGetBackpressure(controller) { function writableStreamDefaultControllerAdvanceQueueIfNeeded(controller) { const { - queue, started, stream, } = controller[kState]; @@ -1235,7 +1234,7 @@ function writableStreamDefaultControllerAdvanceQueueIfNeeded(controller) { return; } - if (!queue.length) + if (queueLength(controller) === 0) return; const value = peekQueueValue(controller); @@ -1294,6 +1293,7 @@ function setupWritableStreamDefaultController( closeAlgorithm, highWaterMark, queue: [], + queueHead: 0, queueTotalSize: 0, abortController: new AbortController(), sizeAlgorithm,