Skip to content

stream: drain WHATWG stream queues from a head index #64354

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
mcollina wants to merge 2 commits into
base: main
Choose a base branch
from
+101 −32
Open

stream: drain WHATWG stream queues from a head index #64354

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
e16a18d
stream: speed up reads and iteration over default WHATWG streams
mcollina Jul 6, 2026
8b43ce8
stream: drain WHATWG stream queues from a head index
mcollina Jul 8, 2026
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
93 changes: 72 additions & 21 deletions lib/internal/webstreams/readablestream.js
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -589,16 +589,20 @@ class ReadableStream {
stream[kState].state === 'readable') {
const controller = stream[kState].controller;
if (isReadableStreamDefaultController(controller)) {
if (controller[kState].queue.length > 0) {
if (controller[kState].queueHead < controller[kState].queue.length) {
stream[kState].disturbed = true;
const chunk = dequeueValue(controller);

if (controller[kState].closeRequested &&
!controller[kState].queue.length) {
controller[kState].queueHead === 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) {

@jasnell jasnell Jul 8, 2026

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

This is repeated more than once and is complicated enough to move into a utility function.

// Reduced ShouldCallPull, as in the read() fast path.
readableStreamDefaultControllerPull(controller);
}

return PromiseResolve({ done: false, value: chunk });
Expand Down Expand Up @@ -940,15 +944,22 @@ 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 (controller[kState].queueHead < controller[kState].queue.length) {

@jasnell jasnell Jul 8, 2026

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Likewise this... worth moving into a utility function

stream[kState].disturbed = true;
const chunk = dequeueValue(controller);

if (controller[kState].closeRequested && !controller[kState].queue.length) {
if (controller[kState].closeRequested &&
controller[kState].queueHead === 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 @@ -1640,14 +1651,15 @@ function readableStreamPipeTo(
// reduces promise allocation overhead.
if (source[kState].state === 'readable' &&
isReadableStreamDefaultController(controller) &&
controller[kState].queue.length > 0) {
controller[kState].queueHead < controller[kState].queue.length) {

while (controller[kState].queue.length > 0) {
while (controller[kState].queueHead < controller[kState].queue.length) {
if (shuttingDown) return true;

const chunk = dequeueValue(controller);

if (controller[kState].closeRequested && !controller[kState].queue.length) {
if (controller[kState].closeRequested &&
controller[kState].queueHead === controller[kState].queue.length) {
readableStreamDefaultControllerClearAlgorithms(controller);
readableStreamClose(source);
}
Expand Down Expand Up @@ -2501,7 +2513,7 @@ function readableStreamDefaultControllerClose(controller) {
if (!readableStreamDefaultControllerCanCloseOrEnqueue(controller))
return;
controller[kState].closeRequested = true;
if (!controller[kState].queue.length) {
if (controller[kState].queueHead === controller[kState].queue.length) {
readableStreamDefaultControllerClearAlgorithms(controller);
readableStreamClose(controller[kState].stream);
}
Expand All @@ -2522,8 +2534,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 +2623,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 @@ -2643,23 +2681,34 @@ function readableStreamDefaultControllerCancelSteps(controller, reason) {
}

function readableStreamDefaultControllerPullSteps(controller, readRequest) {
const {
stream,
queue,
} = controller[kState];
if (queue.length) {
const controllerState = controller[kState];
const stream = controllerState.stream;
// `queue` may be replaced by dequeueValue (queue compaction), so the queue
// and head are read from the state each time rather than bound to a local.
if (controllerState.queueHead < controllerState.queue.length) {
const chunk = dequeueValue(controller);
if (controller[kState].closeRequested && !queue.length) {
if (controllerState.closeRequested &&
controllerState.queueHead === controllerState.queue.length) {
readableStreamDefaultControllerClearAlgorithms(controller);
readableStreamClose(stream);
} else {
readableStreamDefaultControllerCallPullIfNeeded(controller);
} else if (!controllerState.closeRequested &&
controllerState.started &&
controllerState.highWaterMark -
controllerState.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 All @@ -2681,6 +2730,7 @@ function setupReadableStreamDefaultController(
pullFulfilled: undefined,
pullRejected: undefined,
queue: [],
queueHead: 0,
queueTotalSize: 0,
started: false,
sizeAlgorithm,
Expand Down Expand Up @@ -3548,6 +3598,7 @@ function setupReadableByteStreamController(
started: false,
stream,
queue: [],
queueHead: 0,
queueTotalSize: 0,
highWaterMark,
pullAlgorithm,
Expand Down
30 changes: 23 additions & 7 deletions lib/internal/webstreams/util.js
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -5,7 +5,7 @@ const {
ArrayBufferPrototypeGetDetached,
ArrayBufferPrototypeSlice,
ArrayPrototypePush,
ArrayPrototypeShift,
ArrayPrototypeSlice,
AsyncIteratorPrototype,
DataViewPrototypeGetBuffer,
DataViewPrototypeGetByteLength,
Expand Down Expand Up @@ -44,8 +44,6 @@ const {
getPromiseDetails,
} = internalBinding('util');

const assert = require('internal/assert');

const {
isDataView,
} = require('internal/util/types');
Expand Down Expand Up @@ -157,27 +155,45 @@ 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 treat `queue.length - queueHead`, not `queue.length`, as the

@MattiasBuelens MattiasBuelens Jul 8, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Perhaps we should have a helper function for that?

function queueLength(controller) {
  const state = controller[kState];
  return state.queue.length - state.queueHead;
}

Or does that not get inlined as aggressively as we'd want?

// number of live entries.
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) {
Expand Down
10 changes: 6 additions & 4 deletions lib/internal/webstreams/writablestream.js
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1141,14 +1141,14 @@ function writableStreamDefaultControllerProcessWrite(controller, chunk) {
}

function writableStreamDefaultControllerProcessClose(controller) {
const controllerState = controller[kState];
const {
closeAlgorithm,
queue,
stream,
} = controller[kState];
} = controllerState;
writableStreamMarkCloseRequestInFlight(stream);
dequeueValue(controller);
assert(!queue.length);
assert(controllerState.queueHead === controllerState.queue.length);
const sinkClosePromise = closeAlgorithm();
writableStreamDefaultControllerClearAlgorithms(controller);
PromisePrototypeThen(
Expand Down Expand Up @@ -1224,6 +1224,7 @@ function writableStreamDefaultControllerGetBackpressure(controller) {
function writableStreamDefaultControllerAdvanceQueueIfNeeded(controller) {
const {
queue,
queueHead,
started,
stream,
} = controller[kState];
Expand All @@ -1235,7 +1236,7 @@ function writableStreamDefaultControllerAdvanceQueueIfNeeded(controller) {
return;
}

if (!queue.length)
if (queueHead === queue.length)
return;

const value = peekQueueValue(controller);
Expand Down Expand Up @@ -1294,6 +1295,7 @@ function setupWritableStreamDefaultController(
closeAlgorithm,
highWaterMark,
queue: [],
queueHead: 0,
queueTotalSize: 0,
abortController: new AbortController(),
sizeAlgorithm,
Expand Down
Loading