Skip to content

stream: reduce per-chunk overhead in streams and webstreams #64312

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
anonrig wants to merge 2 commits into
base: main
Choose a base branch
from
+413 −71
Open

stream: reduce per-chunk overhead in streams and webstreams #64312

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
6f77105
stream: hoist repeated loads in readable paths
anonrig Jul 5, 2026
c01723a
stream: use ring buffer for WHATWG stream queues
anonrig Jul 5, 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
107 changes: 61 additions & 46 deletions lib/internal/streams/readable.js
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -679,15 +679,17 @@ Readable.prototype.read = function(n) {

// If we're doing read(0) to trigger a readable event, but we
// already have a bunch of data in the buffer, then just trigger
// the 'readable' event and move on.
// the 'readable' event and move on. `state.length` cannot change
// within this block, so it is loaded once instead of three times.
const stateLength = state.length;
if (n === 0 &&
(state[kState] & kNeedReadable) !== 0 &&
((state.highWaterMark !== 0 ?
state.length >= state.highWaterMark :
state.length > 0) ||
stateLength >= state.highWaterMark :
stateLength > 0) ||
(state[kState] & kEnded) !== 0)) {
debug('read: emitReadable');
if (state.length === 0 && (state[kState] & kEnded) !== 0)
if (stateLength === 0 && (state[kState] & kEnded) !== 0)
endReadable(this);
else
emitReadable(this);
Expand Down Expand Up @@ -1628,8 +1630,14 @@ Readable._fromList = fromList;
// This function is designed to be inlinable, so please take care when making
// changes to the function body.
function fromList(n, state) {
// `state.length` cannot change while this function runs (only the
// caller updates it, after this returns) and the chunk lengths feeding
// the copy loops cannot change across the copy calls, so every
// repeated property load below is hoisted into a local.
const stateLength = state.length;

// nothing buffered.
if (state.length === 0)
if (stateLength === 0)
return null;

let idx = state.bufferIndex;
Expand All @@ -1641,7 +1649,7 @@ function fromList(n, state) {
if ((state[kState] & kObjectMode) !== 0) {
ret = buf[idx];
buf[idx++] = null;
} else if (!n || n >= state.length) {
} else if (!n || n >= stateLength) {
// Read it all, truncate the list.
if ((state[kState] & kDecoder) !== 0) {
ret = '';
Expand All @@ -1655,61 +1663,68 @@ function fromList(n, state) {
ret = buf[idx];
buf[idx++] = null;
} else {
ret = Buffer.allocUnsafe(state.length);
ret = Buffer.allocUnsafe(stateLength);

let i = 0;
while (idx < len) {
TypedArrayPrototypeSet(ret, buf[idx], i);
i += buf[idx].length;
const data = buf[idx];
TypedArrayPrototypeSet(ret, data, i);
i += data.length;
buf[idx++] = null;
}
}
} else if (n < buf[idx].length) {
// `slice` is the same for buffers and strings.
ret = buf[idx].slice(0, n);
buf[idx] = buf[idx].slice(n);
} else if (n === buf[idx].length) {
// First chunk is a perfect match.
ret = buf[idx];
buf[idx++] = null;
} else if ((state[kState] & kDecoder) !== 0) {
ret = '';
while (idx < len) {
const str = buf[idx];
if (n > str.length) {
ret += str;
n -= str.length;
buf[idx++] = null;
} else {
if (n === str.length) {
} else {
const first = buf[idx];
const firstLength = first.length;
if (n < firstLength) {
// `slice` is the same for buffers and strings.
ret = first.slice(0, n);
buf[idx] = first.slice(n);
} else if (n === firstLength) {
// First chunk is a perfect match.
ret = first;
buf[idx++] = null;
} else if ((state[kState] & kDecoder) !== 0) {
ret = '';
while (idx < len) {
const str = buf[idx];
const strLength = str.length;
if (n > strLength) {
ret += str;
n -= strLength;
buf[idx++] = null;
} else {
ret += str.slice(0, n);
buf[idx] = str.slice(n);
if (n === strLength) {
ret += str;
buf[idx++] = null;
} else {
ret += str.slice(0, n);
buf[idx] = str.slice(n);
}
break;
}
break;
}
}
} else {
ret = Buffer.allocUnsafe(n);

const retLen = n;
while (idx < len) {
const data = buf[idx];
if (n > data.length) {
TypedArrayPrototypeSet(ret, data, retLen - n);
n -= data.length;
buf[idx++] = null;
} else {
if (n === data.length) {
} else {
ret = Buffer.allocUnsafe(n);

const retLen = n;
while (idx < len) {
const data = buf[idx];
const dataLength = data.length;
if (n > dataLength) {
TypedArrayPrototypeSet(ret, data, retLen - n);
n -= dataLength;
buf[idx++] = null;
} else {
TypedArrayPrototypeSet(ret, new FastBuffer(data.buffer, data.byteOffset, n), retLen - n);
buf[idx] = new FastBuffer(data.buffer, data.byteOffset + n, data.length - n);
if (n === dataLength) {
TypedArrayPrototypeSet(ret, data, retLen - n);
buf[idx++] = null;
} else {
TypedArrayPrototypeSet(ret, new FastBuffer(data.buffer, data.byteOffset, n), retLen - n);
buf[idx] = new FastBuffer(data.buffer, data.byteOffset + n, dataLength - n);
}
break;
}
break;
}
}
}
Expand Down
32 changes: 19 additions & 13 deletions lib/internal/webstreams/readablestream.js
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -113,9 +113,11 @@ const {
extractSizeAlgorithm,
getNonWritablePropertyDescriptor,
isBrandCheck,
kEmptyQueue,
kState,
kType,
lazyTransfer,
materializeQueue,
nonOpCancel,
nonOpPull,
nonOpStart,
Expand Down Expand Up @@ -2495,6 +2497,11 @@ function readableStreamDefaultControllerEnqueue(controller, chunk) {
reader[kType] === 'ReadableStreamDefaultReader' &&
reader[kState].readRequests.length) {
readableStreamFulfillReadRequest(stream, chunk, false);
} else if (controllerState.sizeAlgorithm === defaultSizeAlgorithm) {
// The internal default size algorithm is never observable by user
// code, always returns 1, and cannot throw: enqueue with the
// constant instead of calling it.
enqueueValueWithSize(controller, chunk, 1);
} else {
try {
const chunkSize =
Expand Down Expand Up @@ -2652,7 +2659,7 @@ function setupReadableStreamDefaultController(
pulling: false,
pullFulfilled: undefined,
pullRejected: undefined,
queue: [],
queue: kEmptyQueue,
queueTotalSize: 0,
started: false,
sizeAlgorithm,
Expand Down Expand Up @@ -3112,14 +3119,13 @@ function readableByteStreamControllerEnqueueChunkToQueue(
buffer,
byteOffset,
byteLength) {
ArrayPrototypePush(
controller[kState].queue,
{
buffer,
byteOffset,
byteLength,
});
controller[kState].queueTotalSize += byteLength;
const state = controller[kState];
materializeQueue(state).push({
buffer,
byteOffset,
byteLength,
});
state.queueTotalSize += byteLength;
}

function readableByteStreamControllerEnqueueDetachedPullIntoToQueue(
Expand Down Expand Up @@ -3174,7 +3180,7 @@ function readableByteStreamControllerFillPullIntoDescriptorFromQueue(
} = controller[kState];

while (totalBytesToCopyRemaining) {
const headOfQueue = queue[0];
const headOfQueue = queue.peek();
const bytesToCopy = MathMin(
totalBytesToCopyRemaining,
headOfQueue.byteLength);
Expand All @@ -3192,7 +3198,7 @@ function readableByteStreamControllerFillPullIntoDescriptorFromQueue(
headOfQueue.byteOffset,
bytesToCopy);
if (headOfQueue.byteLength === bytesToCopy) {
ArrayPrototypeShift(queue);
queue.shift();
} else {
headOfQueue.byteOffset += bytesToCopy;
headOfQueue.byteLength -= bytesToCopy;
Expand Down Expand Up @@ -3405,7 +3411,7 @@ function readableByteStreamControllerFillReadRequestFromQueue(controller, readRe
buffer,
byteOffset,
byteLength,
} = ArrayPrototypeShift(queue);
} = queue.shift();

controller[kState].queueTotalSize -= byteLength;
readableByteStreamControllerHandleQueueDrain(controller);
Expand Down Expand Up @@ -3498,7 +3504,7 @@ function setupReadableByteStreamController(
pullRejected: undefined,
started: false,
stream,
queue: [],
queue: kEmptyQueue,
queueTotalSize: 0,
highWaterMark,
pullAlgorithm,
Expand Down
Loading
Loading