From 52726af82f0d89c22242af904c9b9642318fb517 Mon Sep 17 00:00:00 2001 From: Matteo Collina Date: Sat, 8 Aug 2026 19:48:51 +0200 Subject: [PATCH 1/3] 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, From d611d445500aef0ac3994317f92bc860d60a53cd Mon Sep 17 00:00:00 2001 From: Matteo Collina Date: Thu, 20 Aug 2026 07:55:35 +0200 Subject: [PATCH 2/3] stream: encode whole chunks in TextEncoderStream The encode-and-enqueue transform walked the chunk code unit by code unit, materializing a single-character string per index and building the output with string concatenation. The only state that crosses chunks is a trailing high (leading) surrogate, and TextEncoder.encode's USVString conversion already replaces every interior lone surrogate with U+FFFD, which is exactly what the spec loop produces. Join a pending high surrogate with the incoming chunk, hold back a new trailing high surrogate, and encode the rest in a single native call. The streaming decode path also reuses a single options object instead of allocating { stream: true } per chunk. An encoding-streams benchmark is added since the suite had no TextEncoderStream/TextDecoderStream row. Encoding improves by ~546% with 1KB string chunks and ~20% with 16-character chunks; decode is unchanged. Signed-off-by: Matteo Collina --- benchmark/webstreams/encoding-streams.js | 39 +++++++++++++++++++ lib/internal/webstreams/encoding.js | 48 +++++++++++------------- 2 files changed, 61 insertions(+), 26 deletions(-) create mode 100644 benchmark/webstreams/encoding-streams.js diff --git a/benchmark/webstreams/encoding-streams.js b/benchmark/webstreams/encoding-streams.js new file mode 100644 index 000000000000..00759bc09eb7 --- /dev/null +++ b/benchmark/webstreams/encoding-streams.js @@ -0,0 +1,39 @@ +'use strict'; +const common = require('../common.js'); +const { + ReadableStream, + TextEncoderStream, + TextDecoderStream, +} = require('node:stream/web'); + +const bench = common.createBenchmark(main, { + n: [1e5], + kind: ['encode', 'decode'], + len: [16, 1024], +}); + +async function main({ n, kind, len }) { + const encoded = new TextEncoder().encode('a'.repeat(len)); + const decoded = 'a'.repeat(len); + let i = 0; + const rs = new ReadableStream({ + pull(controller) { + if (i++ < n) { + controller.enqueue(kind === 'encode' ? decoded : encoded); + } else { + controller.close(); + } + }, + }); + const ts = kind === 'encode' ? + new TextEncoderStream() : + new TextDecoderStream(); + + 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/encoding.js b/lib/internal/webstreams/encoding.js index f316222ccbf0..038b64030a7a 100644 --- a/lib/internal/webstreams/encoding.js +++ b/lib/internal/webstreams/encoding.js @@ -4,6 +4,7 @@ const { ObjectDefineProperties, String, StringPrototypeCharCodeAt, + StringPrototypeSlice, Uint8Array, } = primordials; @@ -31,6 +32,9 @@ const { kEnumerableProperty, } = require('internal/util'); +// Shared per-chunk decode options; decode() only reads the flag. +const kDecodeStreamingOptions = { __proto__: null, stream: true }; + /** * @typedef {import('./readablestream').ReadableStream} ReadableStream * @typedef {import('./writablestream').WritableStream} WritableStream @@ -46,34 +50,26 @@ class TextEncoderStream { this.#transform = new TransformStream({ transform: (chunk, controller) => { // https://encoding.spec.whatwg.org/#encode-and-enqueue-a-chunk + // The only cross-chunk state is a trailing high surrogate; + // encode() replaces interior lone surrogates with U+FFFD exactly + // like the spec's per-code-unit walk. chunk = String(chunk); - let finalChunk = ''; - for (let i = 0; i < chunk.length; i++) { - const item = chunk[i]; - const codeUnit = StringPrototypeCharCodeAt(item, 0); - if (this.#pendingHighSurrogate !== null) { - const highSurrogate = this.#pendingHighSurrogate; - this.#pendingHighSurrogate = null; - if (0xDC00 <= codeUnit && codeUnit <= 0xDFFF) { - finalChunk += highSurrogate + item; - continue; - } - finalChunk += '\uFFFD'; - } - if (0xD800 <= codeUnit && codeUnit <= 0xDBFF) { - this.#pendingHighSurrogate = item; - continue; - } - if (0xDC00 <= codeUnit && codeUnit <= 0xDFFF) { - finalChunk += '\uFFFD'; - continue; - } - finalChunk += item; + if (chunk.length === 0) + return; + if (this.#pendingHighSurrogate !== null) { + chunk = this.#pendingHighSurrogate + chunk; + this.#pendingHighSurrogate = null; } - if (finalChunk) { - const value = this.#handle.encode(finalChunk); - controller.enqueue(value); + const lastCodeUnit = + StringPrototypeCharCodeAt(chunk, chunk.length - 1); + if (0xD800 <= lastCodeUnit && lastCodeUnit <= 0xDBFF) { + this.#pendingHighSurrogate = + StringPrototypeSlice(chunk, -1); + chunk = StringPrototypeSlice(chunk, 0, -1); + if (chunk.length === 0) + return; } + controller.enqueue(this.#handle.encode(chunk)); }, flush: (controller) => { // https://encoding.spec.whatwg.org/#encode-and-flush @@ -137,7 +133,7 @@ class TextDecoderStream { if (chunk === undefined) { throw new ERR_INVALID_ARG_TYPE('chunk', 'string', chunk); } - const value = this.#handle.decode(chunk, { stream: true }); + const value = this.#handle.decode(chunk, kDecodeStreamingOptions); if (value) controller.enqueue(value); }, From 4a0224907c6f8e21c182a1fa8a841a6bc02bc3a1 Mon Sep 17 00:00:00 2001 From: Matteo Collina Date: Thu, 20 Aug 2026 07:55:46 +0200 Subject: [PATCH 3/3] stream: avoid async pull machinery in ReadableStream.from The pull algorithm was an async function that awaited iterator.next() and then the produced value, costing an async-function frame plus two await wrappers and a controller-side reaction per chunk. Rewrite it callback-style: the next() result is adopted exactly like the previous awaits (including the observable .then lookup on plain object values), the reaction steps are created once per stream, and completion is delivered straight to the controller's cached pull reactions using the parked-algorithm-result contract. The iterator's next method is also looked up once at setup, per the spec's GetIteratorDirect. A from benchmark is added since the suite had no ReadableStream.from row. Iterating a stream built from a sync generator improves by ~27% and from an async generator by ~23%; all other rows are unchanged. Signed-off-by: Matteo Collina --- benchmark/webstreams/from.js | 29 ++++++++ lib/internal/webstreams/readablestream.js | 83 ++++++++++++++++++++--- 2 files changed, 104 insertions(+), 8 deletions(-) create mode 100644 benchmark/webstreams/from.js diff --git a/benchmark/webstreams/from.js b/benchmark/webstreams/from.js new file mode 100644 index 000000000000..05eca4079f1d --- /dev/null +++ b/benchmark/webstreams/from.js @@ -0,0 +1,29 @@ +'use strict'; +const common = require('../common.js'); +const { + ReadableStream, +} = require('node:stream/web'); + +const bench = common.createBenchmark(main, { + n: [1e6], + kind: ['sync', 'async'], +}); + +async function main({ n, kind }) { + function* syncGen() { + for (let i = 0; i < n; i++) yield i; + } + + async function* asyncGen() { + for (let i = 0; i < n; i++) yield i; + } + + const reader = ReadableStream.from( + kind === 'sync' ? syncGen() : asyncGen()).getReader(); + bench.start(); + for (;;) { + const { done } = await reader.read(); + if (done) break; + } + bench.end(n); +} diff --git a/lib/internal/webstreams/readablestream.js b/lib/internal/webstreams/readablestream.js index 84f16dc7b4a1..78313bd03c1a 100644 --- a/lib/internal/webstreams/readablestream.js +++ b/lib/internal/webstreams/readablestream.js @@ -112,6 +112,7 @@ const { getNonWritablePropertyDescriptor, isBrandCheck, kEmptyQueue, + kParkedAlgorithmResult, kResolvedPromise, kState, kType, @@ -1446,19 +1447,85 @@ 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'); } + // Per GetIteratorDirect, the next method is looked up once. + const nextMethod = iterator.next; const startAlgorithm = nonOpCallback; - async function pullAlgorithm() { - const iterResult = await iterator.next(); + // Callback-style pull: the reaction steps are reused across chunks and + // completion is delivered to the controller's cached pull reactions + // (the kParkedAlgorithmResult contract). One pull runs at a time, so a + // single slot carries a non-thenable next() result between steps. + let pendingIterResult; + + function rejectPull(error) { + readableStreamDefaultControllerError(stream[kState].controller, error); + } + + function processIterResult(iterResult) { + const controller = stream[kState].controller; if (typeof iterResult !== 'object' || iterResult === null) { - throw new ERR_INVALID_STATE.TypeError( - 'The promise returned by the iterator.next() method must fulfill with an object'); + rejectPull(new ERR_INVALID_STATE.TypeError( + 'The promise returned by the iterator.next() method must fulfill with an object')); + return; } - if (iterResult.done) { - readableStreamDefaultControllerClose(stream[kState].controller); - } else { - readableStreamDefaultControllerEnqueue(stream[kState].controller, await iterResult.value); + try { + if (iterResult.done) { + readableStreamDefaultControllerClose(controller); + } else { + const value = iterResult.value; + if (value !== null && + (typeof value === 'object' || typeof value === 'function')) { + // Adopted like `await iterResult.value`, keeping the observable + // .then lookup on plain objects. + PromisePrototypeThen(PromiseResolve(value), enqueueValue, rejectPull); + return; + } + readableStreamDefaultControllerEnqueue(controller, value); + } + } catch (error) { + rejectPull(error); + return; + } + // pullFulfilled exists: the controller creates it before the pull. + controller[kState].pullFulfilled(); + } + + function enqueueValue(value) { + const controller = stream[kState].controller; + try { + readableStreamDefaultControllerEnqueue(controller, value); + } catch (error) { + rejectPull(error); + return; + } + controller[kState].pullFulfilled(); + } + + function processPendingIterResult() { + const iterResult = pendingIterResult; + pendingIterResult = undefined; + processIterResult(iterResult); + } + + function pullAlgorithm() { + let nextResult; + try { + nextResult = FunctionPrototypeCall(nextMethod, iterator); + } catch (error) { + return PromiseReject(error); + } + if (nextResult !== null && + (typeof nextResult === 'object' || typeof nextResult === 'function')) { + // Mirrors `await iterator.next()`: processIterResult runs at the + // microtask position the await resumed. + PromisePrototypeThen( + PromiseResolve(nextResult), processIterResult, rejectPull); + return kParkedAlgorithmResult; } + // A non-thenable next() result fails validation a microtask later. + pendingIterResult = nextResult; + PromisePrototypeThen(kResolvedPromise, processPendingIterResult); + return kParkedAlgorithmResult; } async function cancelAlgorithm(reason) {