diff --git a/lib/internal/webstreams/transformstream.js b/lib/internal/webstreams/transformstream.js index 81f60b284f53..d07010dd2f15 100644 --- a/lib/internal/webstreams/transformstream.js +++ b/lib/internal/webstreams/transformstream.js @@ -5,8 +5,6 @@ const { ObjectDefineProperties, ObjectSetPrototypeOf, PromisePrototypeThen, - PromiseReject, - PromiseResolve, PromiseWithResolvers, Symbol, SymbolToStringTag, @@ -46,13 +44,12 @@ const { const { createPromiseCallback1Param, - createRawCallback2Params, + createPromiseCallback2Params, customInspect, extractHighWaterMark, extractSizeAlgorithm, getNonWritablePropertyDescriptor, isBrandCheck, - kParkedAlgorithmResult, kResolvedPromise, kState, kType, @@ -262,12 +259,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); @@ -374,9 +366,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); } @@ -452,30 +442,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( @@ -503,7 +487,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) : @@ -556,40 +540,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) { @@ -610,48 +572,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); } @@ -721,15 +661,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) {