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,