From c455f65414e2cc5c0e6c8b9453f9263fb9bbf383 Mon Sep 17 00:00:00 2001 From: Matteo Collina Date: Sat, 8 Aug 2026 17:00:33 +0200 Subject: [PATCH 1/4] benchmark: apply highWaterMark in webstreams pipe-to The highWaterMark values were passed as properties of the underlying source and sink dictionaries, where they are ignored: a queuing strategy's highWaterMark is read from the constructors' second argument. Every configuration therefore measured the identical workload at the default highWaterMark of 1, which also explains the historically high run-to-run variance of this benchmark family. Pass the strategies as the constructors' second argument and cover the default (1) alongside buffered (1024, 4096) configurations. Signed-off-by: Matteo Collina --- benchmark/webstreams/pipe-to.js | 10 ++++------ 1 file changed, 4 insertions(+), 6 deletions(-) diff --git a/benchmark/webstreams/pipe-to.js b/benchmark/webstreams/pipe-to.js index 38324cd20822..e902f67a9887 100644 --- a/benchmark/webstreams/pipe-to.js +++ b/benchmark/webstreams/pipe-to.js @@ -7,8 +7,8 @@ const { const bench = common.createBenchmark(main, { n: [5e5], - highWaterMarkR: [512, 1024, 2048, 4096], - highWaterMarkW: [512, 1024, 2048, 4096], + highWaterMarkR: [1, 1024, 4096], + highWaterMarkW: [1, 1024, 4096], }); @@ -16,7 +16,6 @@ async function main({ n, highWaterMarkR, highWaterMarkW }) { const b = Buffer.alloc(1024); let i = 0; const rs = new ReadableStream({ - highWaterMark: highWaterMarkR, pull: function(controller) { if (i++ < n) { controller.enqueue(b); @@ -24,12 +23,11 @@ async function main({ n, highWaterMarkR, highWaterMarkW }) { controller.close(); } }, - }); + }, { highWaterMark: highWaterMarkR }); const ws = new WritableStream({ - highWaterMark: highWaterMarkW, write(chunk, controller) {}, close() { bench.end(n); }, - }); + }, { highWaterMark: highWaterMarkW }); bench.start(); rs.pipeTo(ws); From 0ff1bfc362249fc26e1de3b57c2aee3b8991ddd9 Mon Sep 17 00:00:00 2001 From: Matteo Collina Date: Sat, 8 Aug 2026 17:00:48 +0200 Subject: [PATCH 2/4] stream: cut promise churn in webstreams hot paths Three related reductions on the per-chunk paths: Wrap user sink.write and source.pull callbacks without coercing their result into a promise. When the callback returns a non-thenable (the common synchronous case), fulfillment is guaranteed and no then() lookup is observable, so the fulfilled reaction is enqueued through a single shared resolved promise at the exact microtask position the coerced promise's reaction would have had, skipping the implicit async-wrapper promise per chunk. Thenable results go through PromiseResolve(), which matches the spec's "a promise resolved with" conversion (identity for native promises). Park pipeTo's pump on backpressure by installing a record that duck-types the writer's lazily-materialized [[readyPromise]] record and whose resolve function is the pump continuation itself. Backpressure clearing then resumes the pump directly instead of materializing a fresh promise record plus reaction per flip, and the pump no longer schedules a microtask per batch. writableStreamUpdateBackpressure publishes the new backpressure state before resolving the ready record so the pump observes the updated value. Replace queueMicrotask() on the pipeTo and tee chunk-forwarding paths with a reaction on the shared resolved promise, which enqueues the continuation at the same position without the per-call scheduling overhead. pipe-to improves by 8-14% across all benchmark configurations, with readable-read and tee also improving in spot runs. Signed-off-by: Matteo Collina --- lib/internal/webstreams/readablestream.js | 79 ++++++++++++++++++----- lib/internal/webstreams/util.js | 45 ++++++++++++- lib/internal/webstreams/writablestream.js | 27 ++++++-- 3 files changed, 126 insertions(+), 25 deletions(-) diff --git a/lib/internal/webstreams/readablestream.js b/lib/internal/webstreams/readablestream.js index e1e80eb953c0..ad192ff846ea 100644 --- a/lib/internal/webstreams/readablestream.js +++ b/lib/internal/webstreams/readablestream.js @@ -101,6 +101,7 @@ const { cloneAsUint8Array, copyArrayBuffer, createPromiseCallback1Param, + createRawCallback1Param, customInspect, defaultSizeAlgorithm, dequeueValue, @@ -110,6 +111,7 @@ const { getNonWritablePropertyDescriptor, isBrandCheck, kEmptyQueue, + kResolvedPromise, kState, kType, lazyTransfer, @@ -121,6 +123,7 @@ const { resetQueue, resolvedRecord, setPromiseHandled, + thenAlgorithmResult, } = require('internal/webstreams/util'); const { @@ -137,7 +140,6 @@ const { writableStreamDefaultWriterRelease, writableStreamDefaultWriterWriteWithRequest, writerClosedPromise, - writerReadyPromise, } = require('internal/webstreams/writablestream'); const { Buffer } = require('buffer'); @@ -1674,11 +1676,31 @@ function readableStreamPipeTo( // the chunk travels through `pendingChunk`. let pendingChunk; let readRequest; + let readyHook; // Ready promise rejection is handled by the destination-errored // watcher. function ignoreReadyRejection() {} + // Parks the pump on the destination's backpressure by installing a + // record that duck-types the writer's lazily-materialized + // [[readyPromise]] record: writableStreamUpdateBackpressure resolves it + // when backpressure clears (after publishing the new backpressure + // state), which re-enters the pump directly instead of rotating a + // fresh promise record plus reaction per flip. The pipe holds the only + // reference to the writer, so the record is never observable as a real + // ready promise; the erroring/release paths probe `promise` via + // isPromisePending() and call `reject`, so it carries a real + // forever-pending promise and a no-op reject. + function parkOnReady() { + readyHook ??= { + promise: PromiseWithResolvers().promise, + resolve: pump, + reject: ignoreReadyRejection, + }; + writer[kState].ready = readyHook; + } + function forwardChunk() { const chunk = pendingChunk; pendingChunk = undefined; @@ -1690,10 +1712,7 @@ function readableStreamPipeTo( if (shuttingDown) return; if (dest[kState].backpressure) { - PromisePrototypeThen( - writerReadyPromise(writer).promise, - pump, - ignoreReadyRejection); + parkOnReady(); return; } @@ -1738,9 +1757,18 @@ function readableStreamPipeTo( return; } - // Yield to microtask queue between batches to allow events/signals - // to fire - queueMicrotask(pump); + // Park on backpressure directly: the ready hook resumes the pump + // when a completed write clears it. + if (dest[kState].backpressure) { + parkOnReady(); + return; + } + + // Yield to the microtask queue between batches so completed-write + // reactions and events/signals fire; a shared resolved promise + // enqueues the continuation at the same position as queueMicrotask + // without the per-batch scheduling overhead. + PromisePrototypeThen(kResolvedPromise, pump); return; } @@ -1752,7 +1780,7 @@ function readableStreamPipeTo( // synchronous write during enqueue(). See WHATWG Streams spec // "ReadableStreamPipeTo" step 15's "chunk steps". pendingChunk = chunk; - queueMicrotask(forwardChunk); + PromisePrototypeThen(kResolvedPromise, forwardChunk); }, [kClose]() {}, [kError]() {}, @@ -1867,7 +1895,7 @@ function readableStreamDefaultTee(stream, cloneForBranch2) { // The microtask is required by the spec (ReadableStreamTee's // "chunk steps" queue one). pendingChunk = value; - queueMicrotask(forwardChunk); + PromisePrototypeThen(kResolvedPromise, forwardChunk); }, [kClose]() { // The `process.nextTick()` is not part of the spec. @@ -2020,7 +2048,7 @@ function readableByteStreamTee(stream) { defaultReadRequest ??= { [kChunk](chunk) { pendingChunk = chunk; - queueMicrotask(forwardChunk); + PromisePrototypeThen(kResolvedPromise, forwardChunk); }, [kClose]() { reading = false; @@ -2709,8 +2737,18 @@ function readableStreamDefaultControllerPull(controller) { controller[kState].pullRejected = (error) => readableStreamDefaultControllerError(controller, error); } - PromisePrototypeThen( - controller[kState].pullAlgorithm(controller), + // The pull algorithm may be a raw callback (a wrapped user source.pull + // returns its result uncoerced; a synchronous throw surfaces here) or an + // internal algorithm that always returns a promise; thenAlgorithmResult + // handles both. + let result; + try { + result = controller[kState].pullAlgorithm(controller); + } catch (error) { + result = PromiseReject(error); + } + thenAlgorithmResult( + result, controller[kState].pullFulfilled, controller[kState].pullRejected); } @@ -2836,7 +2874,7 @@ function setupReadableStreamDefaultControllerFromSource( FunctionPrototypeBind(start, source, controller) : nonOpStart; const pullAlgorithm = pull ? - createPromiseCallback1Param('source.pull', pull, source) : + createRawCallback1Param('source.pull', pull, source) : nonOpPull; const cancelAlgorithm = cancel ? createPromiseCallback1Param('source.cancel', cancel, source) : @@ -3529,8 +3567,15 @@ function readableByteStreamControllerCallPullIfNeeded(controller) { controller[kState].pullRejected = (error) => readableByteStreamControllerError(controller, error); } - PromisePrototypeThen( - controller[kState].pullAlgorithm(controller), + // See readableStreamDefaultControllerPull for the raw-callback contract. + let result; + try { + result = controller[kState].pullAlgorithm(controller); + } catch (error) { + result = PromiseReject(error); + } + thenAlgorithmResult( + result, controller[kState].pullFulfilled, controller[kState].pullRejected); } @@ -3710,7 +3755,7 @@ function setupReadableByteStreamControllerFromSource( FunctionPrototypeBind(start, source, controller) : nonOpStart; const pullAlgorithm = pull ? - createPromiseCallback1Param('source.pull', pull, source) : + createRawCallback1Param('source.pull', pull, source) : nonOpPull; const cancelAlgorithm = cancel ? createPromiseCallback1Param('source.cancel', cancel, source) : diff --git a/lib/internal/webstreams/util.js b/lib/internal/webstreams/util.js index 8bc4c02be31e..b3a8e7827e38 100644 --- a/lib/internal/webstreams/util.js +++ b/lib/internal/webstreams/util.js @@ -338,6 +338,40 @@ function createPromiseCallbackNoParams(name, fn, thisArg) { return async () => FunctionPrototypeCall(fn, thisArg); } +// Raw variants that skip the async wrapper's implicit result promise. +// Consumers of a raw callback invoke it inside try/catch and route the +// result through thenAlgorithmResult() below. +function createRawCallback1Param(name, fn, thisArg) { + validateFunction(fn, name); + return (arg) => FunctionPrototypeCall(fn, thisArg, arg); +} + +function createRawCallback2Params(name, fn, thisArg) { + validateFunction(fn, name); + return (arg1, arg2) => FunctionPrototypeCall(fn, thisArg, arg1, arg2); +} + +// A single shared, forever-resolved promise used to enqueue a reaction at +// the next microtask checkpoint without allocating a fresh promise. +const kResolvedPromise = PromiseResolve(); + +// Wires the (possibly non-thenable) result of an underlying algorithm +// callback to its fulfilled/rejected continuations. A non-thenable result +// means fulfillment is guaranteed and no then() lookup is observable, so +// the fulfillment step is enqueued directly at the exact microtask +// position the coerced promise's reaction would have had, skipping the +// per-chunk promise allocation. For thenable results PromiseResolve() +// matches the spec's "a promise resolved with" conversion (identity for +// native promises). +function thenAlgorithmResult(result, onFulfilled, onRejected) { + if (result === null || + (typeof result !== 'object' && typeof result !== 'function')) { + PromisePrototypeThen(kResolvedPromise, onFulfilled); + } else { + PromisePrototypeThen(PromiseResolve(result), onFulfilled, onRejected); + } +} + function createPromiseCallback1Param(name, fn, thisArg) { validateFunction(fn, name); return async (arg) => FunctionPrototypeCall(fn, thisArg, arg); @@ -386,11 +420,14 @@ async function nonOpFlush() {} function nonOpStart() {} -async function nonOpPull() {} +// nonOpPull and nonOpWrite are raw callbacks (see createRawCallback*): +// their non-thenable return takes the allocation-free fast path in +// thenAlgorithmResult(). +function nonOpPull() {} async function nonOpCancel() {} -async function nonOpWrite() {} +function nonOpWrite() {} let transfer; function lazyTransfer() { @@ -411,6 +448,8 @@ module.exports = { createPromiseCallbackNoParams, createPromiseCallback1Param, createPromiseCallback2Params, + createRawCallback1Param, + createRawCallback2Params, customInspect, defaultSizeAlgorithm, dequeueValue, @@ -421,6 +460,7 @@ module.exports = { isBrandCheck, isPromisePending, kEmptyQueue, + kResolvedPromise, kState, kType, lazyTransfer, @@ -435,4 +475,5 @@ module.exports = { resetQueue, resolvedRecord, setPromiseHandled, + thenAlgorithmResult, }; diff --git a/lib/internal/webstreams/writablestream.js b/lib/internal/webstreams/writablestream.js index 10b7dbcf277c..5c73f0fe757e 100644 --- a/lib/internal/webstreams/writablestream.js +++ b/lib/internal/webstreams/writablestream.js @@ -56,7 +56,7 @@ const { Queue, createPromiseCallbackNoParams, createPromiseCallback1Param, - createPromiseCallback2Params, + createRawCallback2Params, customInspect, defaultSizeAlgorithm, dequeueValue, @@ -78,6 +78,7 @@ const { resetQueue, resolvedRecord, setPromiseHandled, + thenAlgorithmResult, } = require('internal/webstreams/util'); const { @@ -769,7 +770,12 @@ function writableStreamUpdateBackpressure(controller, streamState) { const backpressure = controllerState.highWaterMark - controllerState.queueTotalSize <= 0; const writer = streamState.writer; - if (writer !== undefined && streamState.backpressure !== backpressure) { + const changed = streamState.backpressure !== backpressure; + // The state field is published before the ready record is resolved so + // that a ready resolve hook (pipeTo's pump continuation) observes the + // new value. + streamState.backpressure = backpressure; + if (writer !== undefined && changed) { if (backpressure) { // The spec replaces [[readyPromise]] with a fresh pending promise; // dropping the cache lets the next observation derive it. @@ -778,7 +784,6 @@ function writableStreamUpdateBackpressure(controller, streamState) { writer[kState].ready?.resolve(); } } - streamState.backpressure = backpressure; } function writableStreamStartErroring(stream, reason) { @@ -1197,8 +1202,18 @@ function writableStreamDefaultControllerProcessWrite(controller, chunk) { }; } - PromisePrototypeThen( - writeAlgorithm(chunk, controller), + // The write algorithm may be a raw callback (a wrapped user sink.write + // returns its result uncoerced; a synchronous throw surfaces here) or an + // internal algorithm that always returns a promise; thenAlgorithmResult + // handles both. + let result; + try { + result = writeAlgorithm(chunk, controller); + } catch (error) { + result = PromiseReject(error); + } + thenAlgorithmResult( + result, controller[kState].writeFulfilled, controller[kState].writeRejected); } @@ -1323,7 +1338,7 @@ function setupWritableStreamDefaultControllerFromSink( FunctionPrototypeBind(start, sink, controller) : nonOpStart; const writeAlgorithm = write ? - createPromiseCallback2Params('sink.write', write, sink) : + createRawCallback2Params('sink.write', write, sink) : nonOpWrite; const closeAlgorithm = close ? createPromiseCallbackNoParams('sink.close', close, sink) : From e30f373bb556f3d8c6e40b89fbfffa8a6fa091c5 Mon Sep 17 00:00:00 2001 From: Matteo Collina Date: Sat, 8 Aug 2026 19:32:35 +0200 Subject: [PATCH 3/4] stream: consolidate non-op algorithm callbacks The start, pull, and write non-op algorithms are all raw callbacks with an identical empty body now, so a single shared nonOpCallback replaces nonOpStart, nonOpPull, and nonOpWrite. Signed-off-by: Matteo Collina --- lib/internal/webstreams/readablestream.js | 21 ++++++++++----------- lib/internal/webstreams/util.js | 16 ++++++---------- lib/internal/webstreams/writablestream.js | 7 +++---- 3 files changed, 19 insertions(+), 25 deletions(-) diff --git a/lib/internal/webstreams/readablestream.js b/lib/internal/webstreams/readablestream.js index ad192ff846ea..d102a2aaaf94 100644 --- a/lib/internal/webstreams/readablestream.js +++ b/lib/internal/webstreams/readablestream.js @@ -116,9 +116,8 @@ const { kType, lazyTransfer, materializeQueue, + nonOpCallback, nonOpCancel, - nonOpPull, - nonOpStart, rejectedHandledRecord, resetQueue, resolvedRecord, @@ -1457,7 +1456,7 @@ function readableStreamFromIterable(iterable) { if (iterator === null || (typeof iterator !== 'object' && typeof iterator !== 'function')) { throw new ERR_INVALID_STATE.TypeError('The iterator method must return an object'); } - const startAlgorithm = nonOpStart; + const startAlgorithm = nonOpCallback; async function pullAlgorithm() { const iterResult = await iterator.next(); @@ -1939,9 +1938,9 @@ function readableStreamDefaultTee(stream, cloneForBranch2) { } branch1 = - createReadableStream(nonOpStart, pullAlgorithm, cancel1Algorithm); + createReadableStream(nonOpCallback, pullAlgorithm, cancel1Algorithm); branch2 = - createReadableStream(nonOpStart, pullAlgorithm, cancel2Algorithm); + createReadableStream(nonOpCallback, pullAlgorithm, cancel2Algorithm); PromisePrototypeThen( readerClosedPromise(reader).promise, @@ -2227,9 +2226,9 @@ function readableByteStreamTee(stream) { } branch1 = - createReadableByteStream(nonOpStart, pull1Algorithm, cancel1Algorithm); + createReadableByteStream(nonOpCallback, pull1Algorithm, cancel1Algorithm); branch2 = - createReadableByteStream(nonOpStart, pull2Algorithm, cancel2Algorithm); + createReadableByteStream(nonOpCallback, pull2Algorithm, cancel2Algorithm); forwardReaderError(reader); @@ -2872,10 +2871,10 @@ function setupReadableStreamDefaultControllerFromSource( const cancel = source?.cancel; const startAlgorithm = start ? FunctionPrototypeBind(start, source, controller) : - nonOpStart; + nonOpCallback; const pullAlgorithm = pull ? createRawCallback1Param('source.pull', pull, source) : - nonOpPull; + nonOpCallback; const cancelAlgorithm = cancel ? createPromiseCallback1Param('source.cancel', cancel, source) : nonOpCancel; @@ -3753,10 +3752,10 @@ function setupReadableByteStreamControllerFromSource( const autoAllocateChunkSize = source?.autoAllocateChunkSize; const startAlgorithm = start ? FunctionPrototypeBind(start, source, controller) : - nonOpStart; + nonOpCallback; const pullAlgorithm = pull ? createRawCallback1Param('source.pull', pull, source) : - nonOpPull; + nonOpCallback; const cancelAlgorithm = cancel ? createPromiseCallback1Param('source.cancel', cancel, source) : nonOpCancel; diff --git a/lib/internal/webstreams/util.js b/lib/internal/webstreams/util.js index b3a8e7827e38..05439a25dcb5 100644 --- a/lib/internal/webstreams/util.js +++ b/lib/internal/webstreams/util.js @@ -418,17 +418,14 @@ function setPromiseHandled(promise) { async function nonOpFlush() {} -function nonOpStart() {} - -// nonOpPull and nonOpWrite are raw callbacks (see createRawCallback*): -// their non-thenable return takes the allocation-free fast path in +// Shared non-op for the start/pull/write algorithm callbacks, which all +// follow the raw-callback contract (see createRawCallback*): the +// non-thenable return takes the allocation-free fast path in // thenAlgorithmResult(). -function nonOpPull() {} +function nonOpCallback() {} async function nonOpCancel() {} -function nonOpWrite() {} - let transfer; function lazyTransfer() { if (transfer === undefined) @@ -465,11 +462,10 @@ module.exports = { kType, lazyTransfer, materializeQueue, + nonOpCallback, nonOpCancel, nonOpFlush, - nonOpPull, - nonOpStart, - nonOpWrite, + peekQueueValue, rejectedHandledRecord, resetQueue, diff --git a/lib/internal/webstreams/writablestream.js b/lib/internal/webstreams/writablestream.js index 5c73f0fe757e..1e9ca02cfe96 100644 --- a/lib/internal/webstreams/writablestream.js +++ b/lib/internal/webstreams/writablestream.js @@ -70,9 +70,8 @@ const { kState, kType, lazyTransfer, + nonOpCallback, nonOpCancel, - nonOpStart, - nonOpWrite, peekQueueValue, rejectedHandledRecord, resetQueue, @@ -1336,10 +1335,10 @@ function setupWritableStreamDefaultControllerFromSink( const abort = sink?.abort; const startAlgorithm = start ? FunctionPrototypeBind(start, sink, controller) : - nonOpStart; + nonOpCallback; const writeAlgorithm = write ? createRawCallback2Params('sink.write', write, sink) : - nonOpWrite; + nonOpCallback; const closeAlgorithm = close ? createPromiseCallbackNoParams('sink.close', close, sink) : nonOpCancel; From 0195e8e2cdaa204b910144f9b8fff909f2d9a6e6 Mon Sep 17 00:00:00 2001 From: Matteo Collina Date: Sat, 8 Aug 2026 19:48:51 +0200 Subject: [PATCH 4/4] stream: decouple transform backpressure changes The spec's [[backpressureChangePromise]] is only ever observed by the transform source pull algorithm (settles when backpressure next becomes true) and by a sink write arriving under backpressure (settles when it next becomes false), both internal. Replace the promise record with direct continuation delivery: a parked pull is completed by enqueueing the readable controller's pull-fulfilled step on the shared resolved promise, and a parked write by a cached per-stream continuation that resolves the sink promise with the perform-transform promise, so adoption reproduces the previous derived-chain settle depth exactly. Each delivery lands at the same microtask position as the old record's reaction. transformStreamDefaultControllerPerformTransform now mirrors the reference implementation's promiseCall().then(undefined, rejection steps) directly instead of running an async wrapper pair per chunk: the transformer.transform callback is wrapped raw, a non-thenable result reuses the shared resolved promise, and a synchronous throw is delivered through a rejected promise, keeping the error timing inside a reaction. A pipe-through benchmark is added since the suite had no transform-throughput row. pipeThrough passthrough improves by ~8% and a writer-driven transform loop by ~14%; the pipe-to and read families are unchanged. Signed-off-by: Matteo Collina --- benchmark/webstreams/pipe-through.js | 38 +++++ lib/internal/webstreams/transformstream.js | 153 +++++++++++++++------ lib/internal/webstreams/util.js | 9 ++ 3 files changed, 156 insertions(+), 44 deletions(-) create mode 100644 benchmark/webstreams/pipe-through.js diff --git a/benchmark/webstreams/pipe-through.js b/benchmark/webstreams/pipe-through.js new file mode 100644 index 000000000000..8af088f4eed1 --- /dev/null +++ b/benchmark/webstreams/pipe-through.js @@ -0,0 +1,38 @@ +'use strict'; +const common = require('../common.js'); +const { + ReadableStream, + TransformStream, +} = require('node:stream/web'); + +const bench = common.createBenchmark(main, { + n: [5e5], + kind: ['default', 'transform'], +}); + +async function main({ n, kind }) { + const b = Buffer.alloc(64); + let i = 0; + const rs = new ReadableStream({ + pull(controller) { + if (i++ < n) { + controller.enqueue(b); + } else { + controller.close(); + } + }, + }); + const ts = kind === 'default' ? + new TransformStream() : + new TransformStream({ + transform(chunk, controller) { controller.enqueue(chunk); }, + }); + + const reader = rs.pipeThrough(ts).getReader(); + bench.start(); + for (;;) { + const { done } = await reader.read(); + if (done) break; + } + bench.end(n); +} diff --git a/lib/internal/webstreams/transformstream.js b/lib/internal/webstreams/transformstream.js index 535c783a3a31..30b7b1c8fac1 100644 --- a/lib/internal/webstreams/transformstream.js +++ b/lib/internal/webstreams/transformstream.js @@ -5,6 +5,8 @@ const { ObjectDefineProperties, ObjectSetPrototypeOf, PromisePrototypeThen, + PromiseReject, + PromiseResolve, PromiseWithResolvers, Symbol, SymbolToStringTag, @@ -44,12 +46,14 @@ const { const { createPromiseCallback1Param, - createPromiseCallback2Params, + createRawCallback2Params, customInspect, extractHighWaterMark, extractSizeAlgorithm, getNonWritablePropertyDescriptor, isBrandCheck, + kParkedAlgorithmResult, + kResolvedPromise, kState, kType, nonOpCancel, @@ -258,7 +262,10 @@ function InternalTransferredTransformStream() { readable: undefined, writable: undefined, backpressure: undefined, - backpressureChange: undefined, + pullPending: false, + pendingWrite: undefined, + pendingWriteChunk: undefined, + writeContinuation: undefined, controller: undefined, }; } @@ -348,7 +355,9 @@ const isTransformStream = const isTransformStreamDefaultController = isBrandCheck('TransformStreamDefaultController'); -async function defaultTransformAlgorithm(chunk, controller) { +// Raw callback (see createRawCallback*): invoked inside the try/catch of +// transformStreamDefaultControllerPerformTransform. +function defaultTransformAlgorithm(chunk, controller) { transformStreamDefaultControllerEnqueue(controller, chunk); } @@ -385,7 +394,12 @@ function initializeTransformStream( writable, controller: undefined, backpressure: undefined, - backpressureChange: undefined, + // Continuation slots replacing the spec's + // [[backpressureChangePromise]]; see transformStreamSetBackpressure. + pullPending: false, + pendingWrite: undefined, + pendingWriteChunk: undefined, + writeContinuation: undefined, }; transformStreamSetBackpressure(stream, true); @@ -422,24 +436,30 @@ 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). 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; -} - +// 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. function transformStreamSetBackpressure(stream, backpressure) { const state = stream[kState]; assert(state.backpressure !== backpressure); - const backpressureChange = state.backpressureChange; - if (backpressureChange !== undefined) { - state.backpressureChange = undefined; - backpressureChange.resolve(); - } 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.pendingWrite !== undefined) { + PromisePrototypeThen(kResolvedPromise, state.writeContinuation); + } } function setupTransformStreamDefaultController( @@ -456,6 +476,7 @@ function setupTransformStreamDefaultController( transformAlgorithm, flushAlgorithm, cancelAlgorithm, + performTransformRejected: undefined, }; stream[kState].controller = controller; } @@ -468,7 +489,7 @@ function setupTransformStreamDefaultControllerFromTransformer( const flush = transformer?.flush; const cancel = transformer?.cancel; const transformAlgorithm = transform ? - createPromiseCallback2Params('transformer.transform', transform, transformer) : + createRawCallback2Params('transformer.transform', transform, transformer) : defaultTransformAlgorithm; const flushAlgorithm = flush ? createPromiseCallback1Param('transformer.flush', flush, transformer) : @@ -521,18 +542,40 @@ function transformStreamDefaultControllerError(controller, error) { transformStreamError(controller[kState].stream, error); } -async function transformStreamDefaultControllerPerformTransform(controller, chunk) { +// 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; try { - const transformAlgorithm = controller[kState].transformAlgorithm; - if (transformAlgorithm === undefined) { - // Algorithms were cleared by a concurrent cancel/abort/close. - return; - } - return await transformAlgorithm(chunk, controller); + result = 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) { @@ -553,26 +596,42 @@ function transformStreamDefaultControllerTerminate(controller) { } function transformStreamDefaultSinkWriteAlgorithm(stream, chunk) { + const state = stream[kState]; const { writable, controller, - } = stream[kState]; + } = state; assert(writable[kState].state === 'writable'); - 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( + if (state.backpressure) { + // Park the chunk and one promise record; the backpressure -> false + // flip delivers the cached continuation (see + // transformStreamSetBackpressure) at the same microtask position as + // the old [[backpressureChangePromise]] reaction. The continuation + // resolves the sink promise with the perform-transform promise, so + // adoption reproduces the old derived-chain settle depth exactly. + // The writable dispatches a single write at a time, so one pending + // slot suffices. + assert(state.pendingWrite === undefined); + const pendingWrite = PromiseWithResolvers(); + state.pendingWrite = pendingWrite; + state.pendingWriteChunk = chunk; + state.writeContinuation ??= () => { + const pending = state.pendingWrite; + const pendingChunk = state.pendingWriteChunk; + state.pendingWrite = undefined; + state.pendingWriteChunk = undefined; + const writableState = state.writable[kState]; + if (writableState.state === 'erroring') { + pending.reject(writableState.storedError); + return; + } + assert(writableState.state === 'writable'); + pending.resolve( + transformStreamDefaultControllerPerformTransform( controller, - chunk); - }); + pendingChunk)); + }; + return pendingWrite.promise; } return transformStreamDefaultControllerPerformTransform(controller, chunk); } @@ -642,9 +701,15 @@ function transformStreamDefaultSinkCloseAlgorithm(stream) { } function transformStreamDefaultSourcePullAlgorithm(stream) { - assert(stream[kState].backpressure); + const state = stream[kState]; + assert(state.backpressure); transformStreamSetBackpressure(stream, false); - return transformStreamBackpressureChangePromise(stream); + // 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; } function transformStreamDefaultSourceCancelAlgorithm(stream, reason) { diff --git a/lib/internal/webstreams/util.js b/lib/internal/webstreams/util.js index 05439a25dcb5..9598796f35c8 100644 --- a/lib/internal/webstreams/util.js +++ b/lib/internal/webstreams/util.js @@ -355,6 +355,12 @@ function createRawCallback2Params(name, fn, thisArg) { // the next microtask checkpoint without allocating a fresh promise. const kResolvedPromise = PromiseResolve(); +// Returned by an internal algorithm to signal that it parked the +// operation and takes responsibility for delivering the fulfilled (or +// rejected) continuation itself later, instead of settling a promise +// (see the transform stream source pull algorithm). +const kParkedAlgorithmResult = { __proto__: null }; + // Wires the (possibly non-thenable) result of an underlying algorithm // callback to its fulfilled/rejected continuations. A non-thenable result // means fulfillment is guaranteed and no then() lookup is observable, so @@ -364,6 +370,8 @@ const kResolvedPromise = PromiseResolve(); // matches the spec's "a promise resolved with" conversion (identity for // native promises). function thenAlgorithmResult(result, onFulfilled, onRejected) { + if (result === kParkedAlgorithmResult) + return; if (result === null || (typeof result !== 'object' && typeof result !== 'function')) { PromisePrototypeThen(kResolvedPromise, onFulfilled); @@ -457,6 +465,7 @@ module.exports = { isBrandCheck, isPromisePending, kEmptyQueue, + kParkedAlgorithmResult, kResolvedPromise, kState, kType,