Skip to content
Open
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
60 changes: 53 additions & 7 deletions lib/internal/webstreams/readablestream.js
Original file line number Diff line number Diff line change
Expand Up @@ -597,8 +597,12 @@ class ReadableStream {
!controller[kState].queue.length) {
readableStreamDefaultControllerClearAlgorithms(controller);
readableStreamClose(stream);
} else {
readableStreamDefaultControllerCallPullIfNeeded(controller);
} else if (!controller[kState].closeRequested &&
controller[kState].started &&
controller[kState].highWaterMark -
controller[kState].queueTotalSize > 0) {
// Reduced ShouldCallPull, as in the read() fast path.
readableStreamDefaultControllerPull(controller);
}

return PromiseResolve({ done: false, value: chunk });
Expand Down Expand Up @@ -947,8 +951,14 @@ class ReadableStreamDefaultReader {
if (controller[kState].closeRequested && !controller[kState].queue.length) {
readableStreamDefaultControllerClearAlgorithms(controller);
readableStreamClose(stream);
} else {
readableStreamDefaultControllerCallPullIfNeeded(controller);
} else if (!controller[kState].closeRequested &&
controller[kState].started &&
controller[kState].highWaterMark -
controller[kState].queueTotalSize > 0) {
// ShouldCallPull reduced to the conditions not already
// established on this path: the state is readable and the
// queue was non-empty, so no read requests can be parked.
readableStreamDefaultControllerPull(controller);
}

return PromiseResolve({ done: false, value: chunk });
Expand Down Expand Up @@ -2522,8 +2532,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(
Expand Down Expand Up @@ -2592,6 +2621,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;
Expand Down Expand Up @@ -2652,14 +2688,24 @@ function readableStreamDefaultControllerPullSteps(controller, readRequest) {
if (controller[kState].closeRequested && !queue.length) {
readableStreamDefaultControllerClearAlgorithms(controller);
readableStreamClose(stream);
} else {
readableStreamDefaultControllerCallPullIfNeeded(controller);
} else if (!controller[kState].closeRequested &&
controller[kState].started &&
controller[kState].highWaterMark -
controller[kState].queueTotalSize > 0) {
// Reduced ShouldCallPull: the state is known to be readable and the
// queue was non-empty, so no read requests can be parked.
readableStreamDefaultControllerPull(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(
Expand Down
Loading