Skip to content

[v24.x] Revert parked transform write changes #66913

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: v24.x-staging
Choose a base branch
from
+43 −110
Open

[v24.x] Revert parked transform write changes #66913

Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
6491a18
Revert "stream: avoid promise allocation for parked transform writes"
mcollina Oct 22, 2026
beeec2e
Revert "stream: decouple transform backpressure changes"
mcollina Oct 21, 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
153 changes: 43 additions & 110 deletions lib/internal/webstreams/transformstream.js
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -5,8 +5,6 @@ const {
ObjectDefineProperties,
ObjectSetPrototypeOf,
PromisePrototypeThen,
PromiseReject,
PromiseResolve,
PromiseWithResolvers,
Symbol,
SymbolToStringTag,
Expand Down Expand Up @@ -46,14 +44,12 @@ const {

const {
createPromiseCallback1Param,
createRawCallback2Params,
createPromiseCallback2Params,
customInspect,
extractHighWaterMark,
extractSizeAlgorithm,
getNonWritablePropertyDescriptor,
isBrandCheck,
kParkedAlgorithmResult,
kResolvedPromise,
kState,
kType,
nonOpCancel,
Expand Down Expand Up @@ -261,12 +257,7 @@ class TransformStreamState {
writable = undefined;
controller = undefined;
backpressure = undefined;
// Continuation slots replacing the spec's
// [[backpressureChangePromise]]; see transformStreamSetBackpressure.
pullPending = false;
pendingWriteParked = false;
pendingWriteChunk = undefined;
writeContinuation = undefined;
backpressureChange = undefined;
}
ObjectSetPrototypeOf(TransformStreamState.prototype, null);

Expand Down Expand Up @@ -372,9 +363,7 @@ const isTransformStream =
const isTransformStreamDefaultController =
isBrandCheck('TransformStreamDefaultController');

// Raw callback (see createRawCallback*): invoked inside the try/catch of
// transformStreamDefaultControllerPerformTransform.
function defaultTransformAlgorithm(chunk, controller) {
async function defaultTransformAlgorithm(chunk, controller) {
transformStreamDefaultControllerEnqueue(controller, chunk);
}

Expand Down Expand Up @@ -444,30 +433,24 @@ function transformStreamUnblockWrite(stream) {
// The spec's [[backpressureChangePromise]] is only ever observed by the
// source pull algorithm (settles when backpressure next becomes true) and
// by a sink write arriving while backpressure is set (settles when
// backpressure next becomes false). Both observers are internal, so the
// promise record is replaced by continuation slots: a parked pull is
// completed by delivering the readable controller's pull-fulfilled step,
// and a parked write by the cached write continuation (see
// transformStreamDefaultSinkWriteAlgorithm). Each is enqueued on the
// shared resolved promise at the exact microtask position the old
// record's reaction would have had.
// backpressure next becomes false). Instead of allocating a fresh promise
// record on every flip, the record is materialized lazily on first
// observation and dropped once settled; flips nobody is waiting on
// allocate nothing.
function transformStreamBackpressureChangePromise(stream) {
const state = stream[kState];
return (state.backpressureChange ??= PromiseWithResolvers()).promise;
}

function transformStreamSetBackpressure(stream, backpressure) {
const state = stream[kState];
assert(state.backpressure !== backpressure);
state.backpressure = backpressure;
if (backpressure) {
if (state.pullPending) {
state.pullPending = false;
// The pull-fulfilled step exists: a pull parked it (see
// transformStreamDefaultSourcePullAlgorithm), and the readable
// controller creates it before invoking the pull algorithm.
PromisePrototypeThen(
kResolvedPromise,
state.readable[kState].controller[kState].pullFulfilled);
}
} else if (state.pendingWriteParked) {
PromisePrototypeThen(kResolvedPromise, state.writeContinuation);
const backpressureChange = state.backpressureChange;
if (backpressureChange !== undefined) {
state.backpressureChange = undefined;
backpressureChange.resolve();
}
state.backpressure = backpressure;
}

function setupTransformStreamDefaultController(
Expand Down Expand Up @@ -495,7 +478,7 @@ function setupTransformStreamDefaultControllerFromTransformer(
const flush = transformer?.flush;
const cancel = transformer?.cancel;
const transformAlgorithm = transform ?
createRawCallback2Params('transformer.transform', transform, transformer) :
createPromiseCallback2Params('transformer.transform', transform, transformer) :
defaultTransformAlgorithm;
const flushAlgorithm = flush ?
createPromiseCallback1Param('transformer.flush', flush, transformer) :
Expand Down Expand Up @@ -548,40 +531,18 @@ function transformStreamDefaultControllerError(controller, error) {
transformStreamError(controller[kState].stream, error);
}

// Mirrors the reference implementation's
// `promiseCall(transformAlgorithm, ...).then(undefined, rejectionSteps)`:
// the returned promise settles one microtask after the (coerced) result
// does, and a rejection errors the transform stream before propagating.
// The raw transform callback plus the shared resolved promise for
// non-thenable results replace the previous async wrapper's two implicit
// promises per chunk.
function transformStreamDefaultControllerPerformTransform(controller, chunk) {
const controllerState = controller[kState];
const transformAlgorithm = controllerState.transformAlgorithm;
if (transformAlgorithm === undefined) {
// Algorithms were cleared by a concurrent cancel/abort/close.
return kResolvedPromise;
}
let result;
async function transformStreamDefaultControllerPerformTransform(controller, chunk) {
try {
result = transformAlgorithm(chunk, controller);
const transformAlgorithm = controller[kState].transformAlgorithm;
if (transformAlgorithm === undefined) {
// Algorithms were cleared by a concurrent cancel/abort/close.
return;
}
return await transformAlgorithm(chunk, controller);
} catch (error) {
result = PromiseReject(error);
}
if (result === null ||
(typeof result !== 'object' && typeof result !== 'function')) {
result = kResolvedPromise;
} else {
result = PromiseResolve(result);
}
controllerState.performTransformRejected ??= (error) => {
transformStreamError(controller[kState].stream, error);
throw error;
};
return PromisePrototypeThen(
result,
undefined,
controllerState.performTransformRejected);
}
}

function transformStreamDefaultControllerTerminate(controller) {
Expand All @@ -602,48 +563,26 @@ function transformStreamDefaultControllerTerminate(controller) {
}

function transformStreamDefaultSinkWriteAlgorithm(stream, chunk) {
const state = stream[kState];
const {
writable,
controller,
} = state;
} = stream[kState];
assert(writable[kState].state === 'writable');
if (state.backpressure) {
// Park the chunk; the backpressure -> false flip delivers the cached
// continuation (see transformStreamSetBackpressure) at the same
// microtask position as the old [[backpressureChangePromise]]
// reaction. The continuation completes the parked write by wiring
// the perform-transform promise directly to the writable
// controller's write reactions (they exist: the controller creates
// them before invoking the write algorithm), replacing the promise
// record the old code allocated and resolved per parked chunk. The
// writable dispatches a single write at a time, so one pending slot
// suffices.
assert(!state.pendingWriteParked);
state.pendingWriteParked = true;
state.pendingWriteChunk = chunk;
state.writeContinuation ??= () => {
const pendingChunk = state.pendingWriteChunk;
state.pendingWriteParked = false;
state.pendingWriteChunk = undefined;
const writableState = state.writable[kState];
const writableControllerState = writableState.controller[kState];
if (writableState.state === 'erroring') {
const error = writableState.storedError;
PromisePrototypeThen(
kResolvedPromise,
() => writableControllerState.writeRejected(error));
return;
}
assert(writableState.state === 'writable');
PromisePrototypeThen(
transformStreamDefaultControllerPerformTransform(
if (stream[kState].backpressure) {
const backpressureChange = transformStreamBackpressureChangePromise(stream);
return PromisePrototypeThen(
backpressureChange,
() => {
const {
writable,
} = stream[kState];
if (writable[kState].state === 'erroring')
throw writable[kState].storedError;
assert(writable[kState].state === 'writable');
return transformStreamDefaultControllerPerformTransform(
controller,
pendingChunk),
writableControllerState.writeFulfilled,
writableControllerState.writeRejected);
};
return kParkedAlgorithmResult;
chunk);
});
}
return transformStreamDefaultControllerPerformTransform(controller, chunk);
}
Expand Down Expand Up @@ -713,15 +652,9 @@ function transformStreamDefaultSinkCloseAlgorithm(stream) {
}

function transformStreamDefaultSourcePullAlgorithm(stream) {
const state = stream[kState];
assert(state.backpressure);
assert(stream[kState].backpressure);
transformStreamSetBackpressure(stream, false);
// Park the pull: the next backpressure -> true flip delivers the
// pull-fulfilled step (see transformStreamSetBackpressure). The old
// [[backpressureChangePromise]] this replaces was only ever resolved,
// so the parked pull needs no rejection delivery.
state.pullPending = true;
return kParkedAlgorithmResult;
return transformStreamBackpressureChangePromise(stream);
}

function transformStreamDefaultSourceCancelAlgorithm(stream, reason) {
Expand Down
Loading