Skip to content

stream: speed up WHATWG web streams #65273

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
Open

stream: speed up WHATWG web streams #65273

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
9f69ca1
stream: speed up WHATWG web streams
anonrig Aug 13, 2026
c796760
typings: add webstreams internalBinding types
anonrig Aug 13, 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
84 changes: 76 additions & 8 deletions lib/internal/webstreams/readablestream.js
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -109,7 +109,9 @@ const {
extractSizeAlgorithm,
getNonWritablePropertyDescriptor,
isBrandCheck,
isNonThenable,
kEmptyQueue,
promiseFromAlgorithmResult,
kState,
kType,
lazyTransfer,
Expand Down Expand Up @@ -250,6 +252,22 @@ class ReadableStream {
*/
constructor(source = kEmptyObject, strategy = kEmptyObject) {
markTransferMode(this, false, true);
// The empty-argument constructor is the creation.js / `new
// ReadableStream()` hot path: skip validateObject and strategy/source
// extraction when both arguments are the shared default sentinel.
if (source === kEmptyObject && strategy === kEmptyObject) {

@jasnell jasnell Aug 14, 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.

In this case, we ought to be able to fully elide the creation of the standard controller entirely. The result here is a completely useless stream whose reads will never resolve. It's a degenerate case that is likely quite unlikely, but if it happens, we may as not waste the additional allocations.

this[kState] = createReadableStreamState();
setupReadableStreamDefaultController(
this,
// eslint-disable-next-line no-use-before-define
new ReadableStreamDefaultController(kSkipThrow),
nonOpStart,
nonOpPull,
nonOpCancel,
1,
defaultSizeAlgorithm);
return;
}
validateObject(source, 'source', kValidateObjectAllowObjects);
validateObject(strategy, 'strategy', kValidateObjectAllowObjectsAndNull);
this[kState] = createReadableStreamState();
Expand Down Expand Up @@ -1689,6 +1707,11 @@ function readableStreamPipeTo(

const controller = source[kState].controller;

if (source[kState].state === 'readable' &&
isReadableStreamDefaultController(controller)) {
readableStreamDefaultControllerFillSync(controller);
}

// Fast path: batch reads when data is buffered in a default controller.
// This avoids parking read requests and reduces promise allocation
// overhead.
Expand Down Expand Up @@ -2699,12 +2722,54 @@ function readableStreamDefaultControllerPull(controller) {
controller[kState].pullRejected =
(error) => readableStreamDefaultControllerError(controller, error);
}
const result = controller[kState].pullAlgorithm(controller);
if (isNonThenable(result)) {

@jasnell jasnell Aug 14, 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 should be clearly documented as a non-standard and potentially breaking behavior. The pull algorithm is defined by the spec as "a promise-returning algorithm". The elimination of the microtask continuation likely makes this semver-major.

queueMicrotask(controller[kState].pullFulfilled);
return;
}
PromisePrototypeThen(
controller[kState].pullAlgorithm(controller),
promiseFromAlgorithmResult(result),
controller[kState].pullFulfilled,
controller[kState].pullRejected);
}

// pipeTo-only: fill a default controller's queue by repeatedly invoking a
// synchronous pull. Not used by the spec pull-fulfillment path (tee and
// WPT require one pull per microtask there).
function readableStreamDefaultControllerFillSync(controller) {
const state = controller[kState];
if (state.pulling || state.pullAlgorithm === undefined)
return;
if (state.pullFulfilled === undefined) {
state.pullFulfilled = () => {
state.pulling = false;
if (state.pullAgain) {
state.pullAgain = false;
readableStreamDefaultControllerCallPullIfNeeded(controller);
}
};
state.pullRejected =
(error) => readableStreamDefaultControllerError(controller, error);
}
while (readableStreamDefaultControllerShouldCallPull(controller)) {
state.pulling = true;
const before = state.queue.length;
const result = state.pullAlgorithm(controller);
if (isNonThenable(result)) {
state.pulling = false;
state.pullAgain = false;
if (state.queue.length === before && !state.closeRequested)
return;
continue;
}
PromisePrototypeThen(
promiseFromAlgorithmResult(result),
state.pullFulfilled,
state.pullRejected);
return;
}
}

function readableStreamDefaultControllerClearAlgorithms(controller) {
controller[kState].pullAlgorithm = undefined;
controller[kState].cancelAlgorithm = undefined;
Expand All @@ -2726,7 +2791,7 @@ function readableStreamDefaultControllerCancelSteps(controller, reason) {
resetQueue(controller);
const result = controller[kState].cancelAlgorithm(reason);
readableStreamDefaultControllerClearAlgorithms(controller);
return result;
return promiseFromAlgorithmResult(result);
}

function readableStreamDefaultControllerPullSteps(controller, readRequest) {
Expand Down Expand Up @@ -2787,8 +2852,7 @@ function setupReadableStreamDefaultController(

const startResult = startAlgorithm();

if (startResult === null ||
(typeof startResult !== 'object' && typeof startResult !== 'function')) {
if (isNonThenable(startResult)) {
// Non-thenable start result: fulfillment is guaranteed and no .then
// lookup on the result is observable, so run the post-start step
// directly at the exact microtask position the promise reaction
Expand Down Expand Up @@ -3519,8 +3583,13 @@ function readableByteStreamControllerCallPullIfNeeded(controller) {
controller[kState].pullRejected =
(error) => readableByteStreamControllerError(controller, error);
}
const result = controller[kState].pullAlgorithm(controller);
if (isNonThenable(result)) {
queueMicrotask(controller[kState].pullFulfilled);

@jasnell jasnell Aug 14, 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.

Care should be taken here. If the pullFulfilled happens to throw for whatever reason, the error is going to be propagated differently than in the promise case below. Not a significant issue since pullfulfilled really shouldn't throw but the case needs to be carefully evaluated.

return;
}
PromisePrototypeThen(
controller[kState].pullAlgorithm(controller),
promiseFromAlgorithmResult(result),
controller[kState].pullFulfilled,
controller[kState].pullRejected);
}
Expand All @@ -3542,7 +3611,7 @@ function readableByteStreamControllerCancelSteps(controller, reason) {
resetQueue(controller);
const result = controller[kState].cancelAlgorithm(reason);
readableByteStreamControllerClearAlgorithms(controller);
return result;
return promiseFromAlgorithmResult(result);
}

// Dequeues the first chunk of the byte queue as a Uint8Array view,
Expand Down Expand Up @@ -3664,8 +3733,7 @@ function setupReadableByteStreamController(

const startResult = startAlgorithm();

if (startResult === null ||
(typeof startResult !== 'object' && typeof startResult !== 'function')) {
if (isNonThenable(startResult)) {
// See setupReadableStreamDefaultController.
queueMicrotask(() => {
controller[kState].started = true;
Expand Down
31 changes: 20 additions & 11 deletions lib/internal/webstreams/transformstream.js
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -54,6 +54,7 @@ const {
kType,
nonOpCancel,
nonOpFlush,
delayedAlgorithmResult,
} = require('internal/webstreams/util');

const {
Expand Down Expand Up @@ -123,9 +124,14 @@ class TransformStream {
writableStrategy = kEmptyObject,
readableStrategy = kEmptyObject) {
markTransferMode(this, false, true);
validateObject(transformer, 'transformer', kValidateObjectAllowObjects);
validateObject(writableStrategy, 'writableStrategy', kValidateObjectAllowObjectsAndNull);
validateObject(readableStrategy, 'readableStrategy', kValidateObjectAllowObjectsAndNull);
if (transformer !== kEmptyObject)
validateObject(transformer, 'transformer', kValidateObjectAllowObjects);
if (writableStrategy !== kEmptyObject) {
validateObject(writableStrategy, 'writableStrategy', kValidateObjectAllowObjectsAndNull);
}
if (readableStrategy !== kEmptyObject) {
validateObject(readableStrategy, 'readableStrategy', kValidateObjectAllowObjectsAndNull);
}
const readableType = transformer?.readableType;
const writableType = transformer?.writableType;
const start = transformer?.start;
Expand Down Expand Up @@ -348,7 +354,7 @@ const isTransformStream =
const isTransformStreamDefaultController =
isBrandCheck('TransformStreamDefaultController');

async function defaultTransformAlgorithm(chunk, controller) {
function defaultTransformAlgorithm(chunk, controller) {
transformStreamDefaultControllerEnqueue(controller, chunk);
}

Expand Down Expand Up @@ -589,15 +595,16 @@ async function transformStreamDefaultSinkAbortAlgorithm(stream, reason) {

const { promise, resolve, reject } = PromiseWithResolvers();
controller[kState].finishPromise = promise;
const cancelPromise = controller[kState].cancelAlgorithm(reason);
const cancelPromise =
delayedAlgorithmResult(controller[kState].cancelAlgorithm(reason));
transformStreamDefaultControllerClearAlgorithms(controller);

PromisePrototypeThen(
cancelPromise,
() => {
if (readable[kState].state === 'errored')
if (readable[kState].state === 'errored') {
reject(readable[kState].storedError);
else {
} else {
readableStreamDefaultControllerError(readable[kState].controller, reason);
resolve();
}
Expand All @@ -622,7 +629,8 @@ function transformStreamDefaultSinkCloseAlgorithm(stream) {
}
const { promise, resolve, reject } = PromiseWithResolvers();
controller[kState].finishPromise = promise;
const flushPromise = controller[kState].flushAlgorithm(controller);
const flushPromise =
delayedAlgorithmResult(controller[kState].flushAlgorithm(controller));
transformStreamDefaultControllerClearAlgorithms(controller);
PromisePrototypeThen(
flushPromise,
Expand Down Expand Up @@ -659,15 +667,16 @@ function transformStreamDefaultSourceCancelAlgorithm(stream, reason) {

const { promise, resolve, reject } = PromiseWithResolvers();
controller[kState].finishPromise = promise;
const cancelPromise = controller[kState].cancelAlgorithm(reason);
const cancelPromise =
delayedAlgorithmResult(controller[kState].cancelAlgorithm(reason));
transformStreamDefaultControllerClearAlgorithms(controller);

PromisePrototypeThen(
cancelPromise,
() => {
if (writable[kState].state === 'errored')
if (writable[kState].state === 'errored') {
reject(writable[kState].storedError);
else {
} else {
writableStreamDefaultControllerErrorIfNeeded(
writable[kState].controller,
reason);
Expand Down
74 changes: 59 additions & 15 deletions lib/internal/webstreams/util.js
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -4,7 +4,6 @@ const {
Array,
ArrayBufferPrototypeGetByteLength,
ArrayBufferPrototypeGetDetached,
ArrayBufferPrototypeSlice,
AsyncIteratorPrototype,
DataViewPrototypeGetBuffer,
DataViewPrototypeGetByteLength,
Expand All @@ -20,7 +19,6 @@ const {
TypedArrayPrototypeGetBuffer,
TypedArrayPrototypeGetByteLength,
TypedArrayPrototypeGetByteOffset,
Uint8Array,
} = primordials;

const {
Expand All @@ -33,6 +31,11 @@ const {
copyArrayBuffer,
} = internalBinding('buffer');

const {
isNonThenable,
cloneAsUint8Array: nativeCloneAsUint8Array,
} = internalBinding('webstreams');

const {
inspect,
} = require('util');
Expand Down Expand Up @@ -128,12 +131,7 @@ function ArrayBufferViewGetByteOffset(view) {
}

function cloneAsUint8Array(view) {
const buffer = ArrayBufferViewGetBuffer(view);
const byteOffset = ArrayBufferViewGetByteOffset(view);
const byteLength = ArrayBufferViewGetByteLength(view);
return new Uint8Array(
ArrayBufferPrototypeSlice(buffer, byteOffset, byteOffset + byteLength),
);
return nativeCloneAsUint8Array(view);
}

function canCopyArrayBuffer(toBuffer, toIndex, fromBuffer, fromIndex, count) {
Expand Down Expand Up @@ -333,19 +331,43 @@ function enqueueValueWithSize(controller, value, size) {
// each known call-site arity gets its own wrapper. The exact number of
// arguments passed through to the user callback is observable and must be
// preserved.
//
// These are intentionally not `async` functions. An `async` wrapper always
// allocates a Promise even when the user callback is synchronous and
// returns a non-thenable; callers use `isNonThenable()` (or
// `PromisePrototypeThen` for thenables) to settle the result, which matches
// the spec's promise-returning conversion without the extra allocation.
function createPromiseCallbackNoParams(name, fn, thisArg) {
validateFunction(fn, name);
return async () => FunctionPrototypeCall(fn, thisArg);
return () => {
try {

@jasnell jasnell Aug 14, 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.

Promise.try / PromiseTry?

return FunctionPrototypeCall(fn, thisArg);
} catch (error) {
return PromiseReject(error);
}
};
}

function createPromiseCallback1Param(name, fn, thisArg) {
validateFunction(fn, name);
return async (arg) => FunctionPrototypeCall(fn, thisArg, arg);
return (arg) => {
try {
return FunctionPrototypeCall(fn, thisArg, arg);
} catch (error) {
return PromiseReject(error);
}
};
}

function createPromiseCallback2Params(name, fn, thisArg) {
validateFunction(fn, name);
return async (arg1, arg2) => FunctionPrototypeCall(fn, thisArg, arg1, arg2);
return (arg1, arg2) => {
try {
return FunctionPrototypeCall(fn, thisArg, arg1, arg2);
} catch (error) {
return PromiseReject(error);
}
};
}

function isPromisePending(promise) {
Expand All @@ -354,6 +376,25 @@ function isPromisePending(promise) {
return details?.[0] === kPending;
}

// Convert a promise-returning algorithm's raw result into a Promise. A
// non-thenable (the common sync-callback case) becomes the shared
// resolved promise; a user thenable is wrapped so Promise.prototype.then
// can be called on it.
function promiseFromAlgorithmResult(result) {
if (isNonThenable(result))
return PromiseResolve();
return PromiseResolve(result);
}

@jasnell jasnell Aug 14, 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.

Are you certain that Proxy is handled correctly here? Worth a test and verification


// Cancel/flush/abort only: insert an extra microtask so "upon fulfillment"
// of an already-settled user promise runs after start-settlement reactions
// queued during construction. Pull/write must not use this.
function delayedAlgorithmResult(result) {
if (isNonThenable(result))
return PromiseResolve();
return PromisePrototypeThen(PromiseResolve(), () => result);
}

// Shared shapes for lazily-materialized { promise, resolve, reject }
// records whose settlement is already known.
function resolvedRecord() {
Expand Down Expand Up @@ -382,15 +423,15 @@ function setPromiseHandled(promise) {
PromisePrototypeThen(promise, undefined, () => {});
}

async function nonOpFlush() {}
function nonOpFlush() {}

function nonOpStart() {}

async function nonOpPull() {}
function nonOpPull() {}

async function nonOpCancel() {}
function nonOpCancel() {}

async function nonOpWrite() {}
function nonOpWrite() {}

@jasnell jasnell Aug 14, 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.

Just have a single no-op function. There's no reason to create multiple functions that do nothing.


let transfer;
function lazyTransfer() {
Expand All @@ -407,6 +448,7 @@ module.exports = {
Queue,
canCopyArrayBuffer,
cloneAsUint8Array,
isNonThenable,
copyArrayBuffer,
createPromiseCallbackNoParams,
createPromiseCallback1Param,
Expand All @@ -427,6 +469,8 @@ module.exports = {
materializeQueue,
nonOpCancel,
nonOpFlush,
promiseFromAlgorithmResult,
delayedAlgorithmResult,
nonOpPull,
nonOpStart,
nonOpWrite,
Expand Down
Loading
Loading