From c5cfdefd64b7d4b00df464d67c60d2f0c6e62a88 Mon Sep 17 00:00:00 2001 From: Matteo Collina Date: Sat, 10 Oct 2026 14:06:26 +0200 Subject: [PATCH 1/2] stream: trim TransformStream construction costs Constructing a TransformStream allocated five closures for the sink and source algorithms of its two sides, and each side adopted the start promise through a wrapper promise plus a thenable job, so a stream without a start() left about three kilobytes pending in the microtask queue until the started steps ran. Per-stream creation bursts spend most of their time copying that graph through the scavenger. The sink and source algorithms are now shared functions that reach the transform stream through a field on their controller state (the controller is passed to the close, abort and cancel algorithms for that), and the post-start steps of both sides are delivered by one reaction chain on a shared promise, taking the same microtask hops as the spec's start promise adoption: three after construction for a non-thenable start result, two after the adopting promise settles for a thenable one. The readable and writable transfer state records are materialized on first transfer instead of per stream. Signed-off-by: Matteo Collina --- lib/internal/webstreams/readablestream.js | 55 +++++++---- lib/internal/webstreams/transformstream.js | 108 ++++++++++++++------- lib/internal/webstreams/util.js | 11 +++ lib/internal/webstreams/writablestream.js | 73 +++++++++----- 4 files changed, 168 insertions(+), 79 deletions(-) diff --git a/lib/internal/webstreams/readablestream.js b/lib/internal/webstreams/readablestream.js index 1b88b8672161..f19337f5b6b7 100644 --- a/lib/internal/webstreams/readablestream.js +++ b/lib/internal/webstreams/readablestream.js @@ -111,6 +111,7 @@ const { extractSizeAlgorithm, getNonWritablePropertyDescriptor, isBrandCheck, + kDeferredStart, kEmptyQueue, kParkedAlgorithmResult, kPendingPromise, @@ -521,10 +522,11 @@ class ReadableStream { [kTransfer]() { if (!isReadableStream(this)) throw new ERR_INVALID_THIS('ReadableStream'); + const transfer = readableStreamTransferState(this); if (this.locked) { - this[kState].transfer.port1?.close(); - this[kState].transfer.port1 = undefined; - this[kState].transfer.port2 = undefined; + transfer.port1?.close(); + transfer.port1 = undefined; + transfer.port2 = undefined; throw new DOMException( 'Cannot transfer a locked ReadableStream', 'DataCloneError'); @@ -533,15 +535,13 @@ class ReadableStream { const { writable, promise, - } = lazyTransfer().newCrossRealmWritableSink( - this, - this[kState].transfer.port1); + } = lazyTransfer().newCrossRealmWritableSink(this, transfer.port1); - this[kState].transfer.writable = writable; - this[kState].transfer.promise = promise; + transfer.writable = writable; + transfer.promise = promise; return { - data: { port: this[kState].transfer.port2 }, + data: { port: transfer.port2 }, deserializeInfo: 'internal/webstreams/readablestream:TransferredReadableStream', }; @@ -549,8 +549,9 @@ class ReadableStream { [kTransferList]() { const { port1, port2 } = new MessageChannel(); - this[kState].transfer.port1 = port1; - this[kState].transfer.port2 = port2; + const transfer = readableStreamTransferState(this); + transfer.port1 = port1; + transfer.port2 = port2; return [ port2 ]; } @@ -1415,10 +1416,15 @@ class ReadableStreamState { state = 'readable'; storedError = undefined; controller = undefined; - transfer = new ReadableStreamTransferState(); + // Materialized on transfer, see readableStreamTransferState(). + transfer = undefined; } ObjectSetPrototypeOf(ReadableStreamState.prototype, null); +function readableStreamTransferState(stream) { + return stream[kState].transfer ??= new ReadableStreamTransferState(); +} + function createReadableStreamState() { return new ReadableStreamState(); } @@ -2961,7 +2967,7 @@ function readableStreamDefaultControllerError(controller, error) { function readableStreamDefaultControllerCancelSteps(controller, reason) { resetQueue(controller); - const result = controller[kState].cancelAlgorithm(reason); + const result = controller[kState].cancelAlgorithm(reason, controller); readableStreamDefaultControllerClearAlgorithms(controller); return result; } @@ -3019,10 +3025,16 @@ function setupReadableStreamDefaultController( started: false, sizeAlgorithm, stream, + // The TransformStream this controller is the source of, if any; its + // shared source algorithms reach the stream through it. + transformStream: undefined, }; stream[kState].controller = controller; const startResult = startAlgorithm(); + if (startResult === kDeferredStart) + return; + // A non-thenable start result guarantees fulfillment, and no .then // lookup on it is observable. const startFulfilled = startResult === null || @@ -3036,12 +3048,7 @@ function setupReadableStreamDefaultController( return; } - const started = () => { - controller[kState].started = true; - assert(!controller[kState].pulling); - assert(!controller[kState].pullAgain); - readableStreamDefaultControllerCallPullIfNeeded(controller); - }; + const started = () => readableStreamDefaultControllerStarted(controller); if (startFulfilled) { // The post-start step runs at the exact microtask position the @@ -3058,6 +3065,15 @@ function setupReadableStreamDefaultController( (error) => readableStreamDefaultControllerError(controller, error)); } +// The post-start steps; transform streams deliver them directly (see +// transformStreamStart). +function readableStreamDefaultControllerStarted(controller) { + controller[kState].started = true; + assert(!controller[kState].pulling); + assert(!controller[kState].pullAgain); + readableStreamDefaultControllerCallPullIfNeeded(controller); +} + function setupReadableStreamDefaultControllerFromSource( stream, source, @@ -4033,6 +4049,7 @@ module.exports = { readableStreamDefaultControllerError, readableStreamDefaultControllerCancelSteps, readableStreamDefaultControllerPullSteps, + readableStreamDefaultControllerStarted, setupReadableStreamDefaultController, setupReadableStreamDefaultControllerFromSource, readableByteStreamControllerClose, diff --git a/lib/internal/webstreams/transformstream.js b/lib/internal/webstreams/transformstream.js index 81f60b284f53..b8f9034e0ff5 100644 --- a/lib/internal/webstreams/transformstream.js +++ b/lib/internal/webstreams/transformstream.js @@ -4,6 +4,7 @@ const { FunctionPrototypeCall, ObjectDefineProperties, ObjectSetPrototypeOf, + Promise, PromisePrototypeThen, PromiseReject, PromiseResolve, @@ -48,6 +49,7 @@ const { createPromiseCallback1Param, createRawCallback2Params, customInspect, + deferredStartAlgorithm, extractHighWaterMark, extractSizeAlgorithm, getNonWritablePropertyDescriptor, @@ -68,11 +70,14 @@ const { readableStreamDefaultControllerError, readableStreamDefaultControllerGetDesiredSize, readableStreamDefaultControllerHasBackpressure, + readableStreamDefaultControllerStarted, } = require('internal/webstreams/readablestream'); const { createWritableStream, writableStreamDefaultControllerErrorIfNeeded, + writableStreamDefaultControllerStarted, + writableStreamDefaultControllerStartFailed, } = require('internal/webstreams/writablestream'); const assert = require('internal/assert'); @@ -159,14 +164,8 @@ class TransformStream { extractHighWaterMark(writableHighWaterMark, 1); const actualWritableSize = extractSizeAlgorithm(writableSize); - // Without a start() the start promise is already resolved by the time - // the readable and writable sides adopt it, so the shared resolved - // promise stands in for the record. - const startPromise = start !== undefined ? PromiseWithResolvers() : undefined; - initializeTransformStream( this, - startPromise, actualWritableHighWaterMark, actualWritableSize, actualReadableHighWaterMark, @@ -174,13 +173,11 @@ class TransformStream { setupTransformStreamDefaultControllerFromTransformer(this, transformer); - if (start !== undefined) { - startPromise.resolve( - FunctionPrototypeCall( - start, - transformer, - this[kState].controller)); - } + transformStreamStart( + this, + start === undefined ? + undefined : + FunctionPrototypeCall(start, transformer, this[kState].controller)); } /** @@ -380,38 +377,35 @@ function defaultTransformAlgorithm(chunk, controller) { transformStreamDefaultControllerEnqueue(controller, chunk); } -function resolvedStartAlgorithm() { - return kResolvedPromise; -} - function initializeTransformStream( stream, - startPromise, writableHighWaterMark, writableSizeAlgorithm, readableHighWaterMark, readableSizeAlgorithm) { - const startAlgorithm = startPromise === undefined ? - resolvedStartAlgorithm : - () => startPromise.promise; - + // The sink and source algorithms are shared functions that reach the + // transform stream through their controller, and the post-start step + // is delivered by transformStreamStart, so no per-stream closures or + // start promises are allocated here. const writable = createWritableStream( - startAlgorithm, - (chunk) => transformStreamDefaultSinkWriteAlgorithm(stream, chunk), - () => transformStreamDefaultSinkCloseAlgorithm(stream), - (reason) => transformStreamDefaultSinkAbortAlgorithm(stream, reason), + deferredStartAlgorithm, + transformStreamDefaultSinkWriteAlgorithm, + transformStreamDefaultSinkCloseAlgorithm, + transformStreamDefaultSinkAbortAlgorithm, writableHighWaterMark, writableSizeAlgorithm, ); + writable[kState].controller[kState].transformStream = stream; const readable = createReadableStream( - startAlgorithm, - () => transformStreamDefaultSourcePullAlgorithm(stream), - (reason) => transformStreamDefaultSourceCancelAlgorithm(stream, reason), + deferredStartAlgorithm, + transformStreamDefaultSourcePullAlgorithm, + transformStreamDefaultSourceCancelAlgorithm, readableHighWaterMark, readableSizeAlgorithm, ); + readable[kState].controller[kState].transformStream = stream; const state = new TransformStreamState(); state.readable = readable; @@ -421,6 +415,43 @@ function initializeTransformStream( transformStreamSetBackpressure(stream, true); } +// Delivers the post-start step of both sides. The spec resolves the start +// promise with the transformer's start result at the end of construction +// and each side then adopts it through a wrapper promise, so with a +// non-thenable start result the started steps run three microtasks after +// construction, and with a thenable one two microtasks after the adopting +// promise settles. The same hops are taken as reactions on one shared +// promise instead of a promise record plus two wrapper promises. +function transformStreamStart(stream, startResult) { + const state = stream[kState]; + const started = () => { + writableStreamDefaultControllerStarted(state.writable[kState].controller); + readableStreamDefaultControllerStarted(state.readable[kState].controller); + }; + if (startResult === null || + (typeof startResult !== 'object' && typeof startResult !== 'function')) { + PromisePrototypeThen( + PromisePrototypeThen( + PromisePrototypeThen(kResolvedPromise, undefined, undefined), + undefined, + undefined), + started); + return; + } + const startPromise = new Promise((resolve) => resolve(startResult)); + PromisePrototypeThen( + PromisePrototypeThen(startPromise, undefined, undefined), + started, + (error) => { + writableStreamDefaultControllerStartFailed( + state.writable[kState].controller, + error); + readableStreamDefaultControllerError( + state.readable[kState].controller, + error); + }); +} + function transformStreamError(stream, error) { const { readable, @@ -609,7 +640,8 @@ function transformStreamDefaultControllerTerminate(controller) { new ERR_INVALID_STATE.TypeError('TransformStream has been terminated')); } -function transformStreamDefaultSinkWriteAlgorithm(stream, chunk) { +function transformStreamDefaultSinkWriteAlgorithm(chunk, writableController) { + const stream = writableController[kState].transformStream; const state = stream[kState]; const { writable, @@ -656,7 +688,10 @@ function transformStreamDefaultSinkWriteAlgorithm(stream, chunk) { return transformStreamDefaultControllerPerformTransform(controller, chunk); } -async function transformStreamDefaultSinkAbortAlgorithm(stream, reason) { +async function transformStreamDefaultSinkAbortAlgorithm( + reason, + writableController) { + const stream = writableController[kState].transformStream; const { controller, readable, @@ -690,7 +725,8 @@ async function transformStreamDefaultSinkAbortAlgorithm(stream, reason) { return controller[kState].finishPromise; } -function transformStreamDefaultSinkCloseAlgorithm(stream) { +function transformStreamDefaultSinkCloseAlgorithm(writableController) { + const stream = writableController[kState].transformStream; const { readable, controller, @@ -720,7 +756,8 @@ function transformStreamDefaultSinkCloseAlgorithm(stream) { return controller[kState].finishPromise; } -function transformStreamDefaultSourcePullAlgorithm(stream) { +function transformStreamDefaultSourcePullAlgorithm(readableController) { + const stream = readableController[kState].transformStream; const state = stream[kState]; assert(state.backpressure); transformStreamSetBackpressure(stream, false); @@ -732,7 +769,10 @@ function transformStreamDefaultSourcePullAlgorithm(stream) { return kParkedAlgorithmResult; } -function transformStreamDefaultSourceCancelAlgorithm(stream, reason) { +function transformStreamDefaultSourceCancelAlgorithm( + reason, + readableController) { + const stream = readableController[kState].transformStream; const { controller, writable, diff --git a/lib/internal/webstreams/util.js b/lib/internal/webstreams/util.js index 581e3540c680..bc6818c11acd 100644 --- a/lib/internal/webstreams/util.js +++ b/lib/internal/webstreams/util.js @@ -357,6 +357,15 @@ const kResolvedPromise = PromiseResolve(); // (see the transform stream source pull algorithm). const kParkedAlgorithmResult = Symbol('kParkedAlgorithmResult'); +// Start algorithm of internal streams whose creator delivers the +// post-start step itself (see transformStreamStart): the controller setup +// leaves `started` unset when the start result is this sentinel. +const kDeferredStart = Symbol('kDeferredStart'); + +function deferredStartAlgorithm() { + return kDeferredStart; +} + // 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 @@ -461,6 +470,7 @@ module.exports = { createRawCallback2Params, customInspect, defaultSizeAlgorithm, + deferredStartAlgorithm, dequeueValue, enqueueValueWithSize, extractHighWaterMark, @@ -468,6 +478,7 @@ module.exports = { getNonWritablePropertyDescriptor, isBrandCheck, isPromisePending, + kDeferredStart, kEmptyQueue, kParkedAlgorithmResult, kPendingPromise, diff --git a/lib/internal/webstreams/writablestream.js b/lib/internal/webstreams/writablestream.js index 31f2a1f875dc..7b6a7327944b 100644 --- a/lib/internal/webstreams/writablestream.js +++ b/lib/internal/webstreams/writablestream.js @@ -66,6 +66,7 @@ const { getNonWritablePropertyDescriptor, isBrandCheck, isPromisePending, + kDeferredStart, kEmptyQueue, kState, kType, @@ -303,10 +304,11 @@ class WritableStream { [kTransfer]() { if (!isWritableStream(this)) throw new ERR_INVALID_THIS('WritableStream'); + const transfer = writableStreamTransferState(this); if (this.locked) { - this[kState].transfer.port1?.close(); - this[kState].transfer.port1 = undefined; - this[kState].transfer.port2 = undefined; + transfer.port1?.close(); + transfer.port1 = undefined; + transfer.port2 = undefined; throw new DOMException( 'Cannot transfer a locked WritableStream', 'DataCloneError'); @@ -315,15 +317,13 @@ class WritableStream { const { readable, promise, - } = lazyTransfer().newCrossRealmReadableStream( - this, - this[kState].transfer.port1); + } = lazyTransfer().newCrossRealmReadableStream(this, transfer.port1); - this[kState].transfer.readable = readable; - this[kState].transfer.promise = promise; + transfer.readable = readable; + transfer.promise = promise; return { - data: { port: this[kState].transfer.port2 }, + data: { port: transfer.port2 }, deserializeInfo: 'internal/webstreams/writablestream:TransferredWritableStream', }; @@ -331,8 +331,9 @@ class WritableStream { [kTransferList]() { const { port1, port2 } = new MessageChannel(); - this[kState].transfer.port1 = port1; - this[kState].transfer.port2 = port2; + const transfer = writableStreamTransferState(this); + transfer.port1 = port1; + transfer.port2 = port2; return [ port2 ]; } @@ -511,7 +512,7 @@ class WritableStreamDefaultController { } [kAbort](reason) { - const result = this[kState].abortAlgorithm(reason); + const result = this[kState].abortAlgorithm(reason, this); writableStreamDefaultControllerClearAlgorithms(this); return result; } @@ -605,10 +606,15 @@ class WritableStreamState { // no request storage. writeRequests = kEmptyQueue; writer = undefined; - transfer = new WritableStreamTransferState(); + // Materialized on transfer, see writableStreamTransferState(). + transfer = undefined; } ObjectSetPrototypeOf(WritableStreamState.prototype, null); +function writableStreamTransferState(stream) { + return stream[kState].transfer ??= new WritableStreamTransferState(); +} + function createWritableStreamState() { return new WritableStreamState(); } @@ -1219,7 +1225,7 @@ function writableStreamDefaultControllerProcessClose(controller) { writableStreamMarkCloseRequestInFlight(stream); dequeueValue(controller); assert(!queue.length); - const sinkClosePromise = closeAlgorithm(); + const sinkClosePromise = closeAlgorithm(controller); writableStreamDefaultControllerClearAlgorithms(controller); PromisePrototypeThen( sinkClosePromise, @@ -1370,6 +1376,9 @@ function setupWritableStreamDefaultController( sizeAlgorithm, started: false, stream, + // The TransformStream this controller is the sink of, if any; its + // shared sink algorithms reach the stream through it. + transformStream: undefined, writeAlgorithm, writeFulfilled: undefined, writeRejected: undefined, @@ -1379,13 +1388,10 @@ function setupWritableStreamDefaultController( writableStreamUpdateBackpressure(controller, stream[kState]); const startResult = startAlgorithm(); + if (startResult === kDeferredStart) + return; - const started = () => { - assert(stream[kState].state === 'writable' || - stream[kState].state === 'erroring'); - controller[kState].started = true; - writableStreamDefaultControllerAdvanceQueueIfNeeded(controller); - }; + const started = () => writableStreamDefaultControllerStarted(controller); if (startResult === null || (typeof startResult !== 'object' && typeof startResult !== 'function')) { @@ -1401,12 +1407,25 @@ function setupWritableStreamDefaultController( PromisePrototypeThen( new Promise((r) => r(startResult)), started, - (error) => { - assert(stream[kState].state === 'writable' || - stream[kState].state === 'erroring'); - controller[kState].started = true; - writableStreamDealWithRejection(stream, error); - }); + (error) => writableStreamDefaultControllerStartFailed(controller, error)); +} + +// The post-start steps; transform streams deliver them directly (see +// transformStreamStart). +function writableStreamDefaultControllerStarted(controller) { + const { stream } = controller[kState]; + assert(stream[kState].state === 'writable' || + stream[kState].state === 'erroring'); + controller[kState].started = true; + writableStreamDefaultControllerAdvanceQueueIfNeeded(controller); +} + +function writableStreamDefaultControllerStartFailed(controller, error) { + const { stream } = controller[kState]; + assert(stream[kState].state === 'writable' || + stream[kState].state === 'erroring'); + controller[kState].started = true; + writableStreamDealWithRejection(stream, error); } module.exports = { @@ -1459,6 +1478,8 @@ module.exports = { writableStreamDefaultControllerClose, writableStreamDefaultControllerClearAlgorithms, writableStreamDefaultControllerAdvanceQueueIfNeeded, + writableStreamDefaultControllerStarted, + writableStreamDefaultControllerStartFailed, setupWritableStreamDefaultControllerFromSink, setupWritableStreamDefaultController, createWritableStream, From 4fe400ba4588d3c656f35cea44bfa3c7ccca8647 Mon Sep 17 00:00:00 2001 From: Matteo Collina Date: Tue, 13 Oct 2026 13:42:17 +0200 Subject: [PATCH 2/2] stream: trim webstreams sink wrappers and BYOB transfers MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The close and abort methods of an underlying sink and the flush and cancel methods of a transformer were each wrapped at construction in an async closure that the controller kept for the stream's lifetime: two per WritableStream and two per TransformStream, each with its context. Store these one-shot methods as-is in the controller state next to the object they belong to, and call them as that object's methods with the spec's argument list through a promiseCall helper that keeps the reference implementation's adoption ticks. The per-chunk methods (write, transform) keep their closures, which make the method a constant for the optimizing compiler on the hot paths; so does the source's cancel wrapper, since the readable stream's construction path is sensitive to any change in its inlining budget. In the byte controller, the transfer of a pull-into descriptor's buffer when it is committed is dropped: nothing can reach that buffer any more, since it was transferred when the descriptor was created and the BYOB request that exposed it is only ever invalidated together with a transfer. The transfer of the first pending descriptor's buffer on enqueue is likewise only needed while a BYOB request exposes it. Allocating an ArrayBuffer costs about half a microsecond, so those two transfers were a large part of a BYOB read. Bytes allocated per stream with user methods: WritableStream 1040 -> 844, TransformStream 1816 -> 1720. The microtask ordering of every user method call is unchanged. Benchmark results (20 runs, vs. the previous commit): webstreams/readable-read.js type='byob' n=100000 *** 17.14 % ±1.80% webstreams/readable-async-iterator.js type='normal' n=100000 3.79 % ±4.49% webstreams/from.js kind='sync' n=1000000 * -2.18 % ±2.02% webstreams/js_transfer.js n=10000 payload='TransformStream' * -1.85 % ±1.58% The other 42 rows are within their confidence intervals; the two rows at the 5% level are the expected number of false positives for 46 comparisons. Signed-off-by: Matteo Collina --- lib/internal/webstreams/readablestream.js | 23 ++- lib/internal/webstreams/transformstream.js | 41 ++++-- lib/internal/webstreams/util.js | 85 +++++++---- lib/internal/webstreams/writablestream.js | 41 ++++-- ...st-whatwg-webstreams-underlying-methods.js | 137 ++++++++++++++++++ 5 files changed, 257 insertions(+), 70 deletions(-) create mode 100644 test/parallel/test-whatwg-webstreams-underlying-methods.js diff --git a/lib/internal/webstreams/readablestream.js b/lib/internal/webstreams/readablestream.js index f19337f5b6b7..e7879ada16a6 100644 --- a/lib/internal/webstreams/readablestream.js +++ b/lib/internal/webstreams/readablestream.js @@ -2388,13 +2388,16 @@ function readableByteStreamControllerConvertPullIntoDescriptor(desc) { if (bytesFilled > byteLength) throw new ERR_INVALID_STATE.RangeError('The buffer size is invalid'); assert(!(bytesFilled % elementSize)); - const transferredBuffer = ArrayBufferPrototypeTransfer(buffer); + // The spec transfers the buffer once more here. Nothing else can reach + // it by now: it was transferred when the descriptor was created, and + // the BYOB request that exposes it has been invalidated, which always + // goes together with a transfer, so the view is built on it as-is. if (ctor === Buffer) { - return Buffer.from(transferredBuffer, byteOffset, bytesFilled / elementSize); + return Buffer.from(buffer, byteOffset, bytesFilled / elementSize); } - return new ctor(transferredBuffer, byteOffset, bytesFilled / elementSize); + return new ctor(buffer, byteOffset, bytesFilled / elementSize); } function isReadableStreamLocked(stream) { @@ -3427,11 +3430,15 @@ function readableByteStreamControllerEnqueue(controller, chunk) { ); } - readableByteStreamControllerInvalidateBYOBRequest(controller); - - firstPendingPullInto.buffer = ArrayBufferPrototypeTransfer( - firstPendingPullInto.buffer, - ); + // The transfer detaches the view the BYOB request exposed; without + // a request the buffer is unreachable and there is nothing to + // detach. + if (controller[kState].byobRequest !== null) { + readableByteStreamControllerInvalidateBYOBRequest(controller); + firstPendingPullInto.buffer = ArrayBufferPrototypeTransfer( + firstPendingPullInto.buffer, + ); + } if (firstPendingPullInto.type === 'none') { readableByteStreamControllerEnqueueDetachedPullIntoToQueue( diff --git a/lib/internal/webstreams/transformstream.js b/lib/internal/webstreams/transformstream.js index b8f9034e0ff5..f38dd90ddc12 100644 --- a/lib/internal/webstreams/transformstream.js +++ b/lib/internal/webstreams/transformstream.js @@ -33,6 +33,7 @@ const { } = require('internal/util'); const { + validateFunction, validateObject, kValidateObjectAllowObjects, kValidateObjectAllowObjectsAndNull, @@ -46,7 +47,6 @@ const { } = require('internal/worker/js_transferable'); const { - createPromiseCallback1Param, createRawCallback2Params, customInspect, deferredStartAlgorithm, @@ -58,8 +58,8 @@ const { kResolvedPromise, kState, kType, - nonOpCancel, - nonOpFlush, + nonOpCallback, + promiseCall1Param, } = require('internal/webstreams/util'); const { @@ -270,6 +270,9 @@ ObjectSetPrototypeOf(TransformStreamState.prototype, null); class TransformStreamControllerState { stream = undefined; + // The user-provided transformer the flush and cancel algorithms are + // methods of. + transformer = undefined; transformAlgorithm = undefined; flushAlgorithm = undefined; cancelAlgorithm = undefined; @@ -514,11 +517,13 @@ function setupTransformStreamDefaultController( controller, transformAlgorithm, flushAlgorithm, - cancelAlgorithm) { + cancelAlgorithm, + transformer) { assert(isTransformStream(stream)); assert(stream[kState].controller === undefined); const controllerState = new TransformStreamControllerState(); controllerState.stream = stream; + controllerState.transformer = transformer; controllerState.transformAlgorithm = transformAlgorithm; controllerState.flushAlgorithm = flushAlgorithm; controllerState.cancelAlgorithm = cancelAlgorithm; @@ -536,25 +541,25 @@ function setupTransformStreamDefaultControllerFromTransformer( const transformAlgorithm = transform ? createRawCallback2Params('transformer.transform', transform, transformer) : defaultTransformAlgorithm; - const flushAlgorithm = flush ? - createPromiseCallback1Param('transformer.flush', flush, transformer) : - nonOpFlush; - const cancelAlgorithm = cancel ? - createPromiseCallback1Param('transformer.cancel', cancel, transformer) : - nonOpCancel; + if (flush) + validateFunction(flush, 'transformer.flush'); + if (cancel) + validateFunction(cancel, 'transformer.cancel'); setupTransformStreamDefaultController( stream, controller, transformAlgorithm, - flushAlgorithm, - cancelAlgorithm); + flush || nonOpCallback, + cancel || nonOpCallback, + transformer); } function transformStreamDefaultControllerClearAlgorithms(controller) { controller[kState].transformAlgorithm = undefined; controller[kState].flushAlgorithm = undefined; controller[kState].cancelAlgorithm = undefined; + controller[kState].transformer = undefined; } function transformStreamDefaultControllerEnqueue(controller, chunk) { @@ -703,7 +708,9 @@ async function transformStreamDefaultSinkAbortAlgorithm( const { promise, resolve, reject } = PromiseWithResolvers(); controller[kState].finishPromise = promise; - const cancelPromise = controller[kState].cancelAlgorithm(reason); + const cancelPromise = promiseCall1Param( + controller[kState].cancelAlgorithm, controller[kState].transformer, + reason); transformStreamDefaultControllerClearAlgorithms(controller); PromisePrototypeThen( @@ -737,7 +744,9 @@ function transformStreamDefaultSinkCloseAlgorithm(writableController) { } const { promise, resolve, reject } = PromiseWithResolvers(); controller[kState].finishPromise = promise; - const flushPromise = controller[kState].flushAlgorithm(controller); + const flushPromise = promiseCall1Param( + controller[kState].flushAlgorithm, controller[kState].transformer, + controller); transformStreamDefaultControllerClearAlgorithms(controller); PromisePrototypeThen( flushPromise, @@ -784,7 +793,9 @@ function transformStreamDefaultSourceCancelAlgorithm( const { promise, resolve, reject } = PromiseWithResolvers(); controller[kState].finishPromise = promise; - const cancelPromise = controller[kState].cancelAlgorithm(reason); + const cancelPromise = promiseCall1Param( + controller[kState].cancelAlgorithm, controller[kState].transformer, + reason); transformStreamDefaultControllerClearAlgorithms(controller); PromisePrototypeThen( diff --git a/lib/internal/webstreams/util.js b/lib/internal/webstreams/util.js index bc6818c11acd..2d9ba0ae4038 100644 --- a/lib/internal/webstreams/util.js +++ b/lib/internal/webstreams/util.js @@ -323,20 +323,14 @@ function enqueueValueWithSize(controller, value, size) { state.queueTotalSize += coercedSize; } -// Arity-specialized variants of the promise-callback wrapper. The generic -// rest-parameter + ReflectApply form allocated an arguments array on every -// invocation; these run on per-chunk hot paths (pull/write/transform), so -// each known call-site arity gets its own wrapper. The exact number of -// arguments passed through to the user callback is observable and must be -// preserved. -function createPromiseCallbackNoParams(name, fn, thisArg) { - validateFunction(fn, name); - 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. +// The per-chunk callbacks of a user-provided source, sink or transformer +// (pull, write, transform) are wrapped in a closure whose captured method +// stays a constant for the optimizing compiler. Each known call-site +// arity gets its own wrapper, since a rest-parameter form would allocate +// an arguments array per invocation and the exact number of arguments +// passed to the user callback is observable. 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); @@ -347,6 +341,46 @@ function createRawCallback2Params(name, fn, thisArg) { return (arg1, arg2) => FunctionPrototypeCall(fn, thisArg, arg1, arg2); } +// Async wrapper of a one-shot callback (the source cancel method). +function createPromiseCallback1Param(name, fn, thisArg) { + validateFunction(fn, name); + return async (arg) => FunctionPrototypeCall(fn, thisArg, arg); +} + +// The one-shot methods of a sink or transformer (close, abort, flush, +// cancel) are instead stored as-is in the controller state next to the +// object they belong to and called the way the reference implementation's +// promiseCall() does: a synchronous throw becomes a rejection and the +// result is adopted through a fresh promise, so a returned promise settles +// the caller two microtasks later and a plain value settles it at once. +function promiseCall1Param(fn, thisArg, arg) { + let result; + try { + result = FunctionPrototypeCall(fn, thisArg, arg); + } catch (error) { + return PromiseReject(error); + } + return promiseResolvedWith(result); +} + +function promiseCallNoParams(fn, thisArg) { + let result; + try { + result = FunctionPrototypeCall(fn, thisArg); + } catch (error) { + return PromiseReject(error); + } + return promiseResolvedWith(result); +} + +function promiseResolvedWith(value) { + if (value === null || + (typeof value !== 'object' && typeof value !== 'function')) { + return PromiseResolve(value); + } + return new Promise((resolve) => resolve(value)); +} + // A single shared, forever-resolved promise used to enqueue a reaction at // the next microtask checkpoint without allocating a fresh promise. const kResolvedPromise = PromiseResolve(); @@ -385,16 +419,6 @@ function thenAlgorithmResult(result, onFulfilled, onRejected) { } } -function createPromiseCallback1Param(name, fn, thisArg) { - validateFunction(fn, name); - return async (arg) => FunctionPrototypeCall(fn, thisArg, arg); -} - -function createPromiseCallback2Params(name, fn, thisArg) { - validateFunction(fn, name); - return async (arg1, arg2) => FunctionPrototypeCall(fn, thisArg, arg1, arg2); -} - function isPromisePending(promise) { if (promise === undefined) return false; const details = getPromiseDetails(promise); @@ -438,12 +462,11 @@ function setPromiseHandled(promise) { PromisePrototypeThen(promise, undefined, () => {}); } -async function nonOpFlush() {} - // Shared non-op for the start/pull/write algorithm callbacks, which all -// follow the raw-callback contract (see createRawCallback*): the +// follow the raw-callback contract (see createRawCallback*), and for the +// missing one-shot methods of a user-provided sink or transformer: the // non-thenable return takes the allocation-free fast path in -// thenAlgorithmResult(). +// thenAlgorithmResult() and settles promiseCall*() at once. function nonOpCallback() {} async function nonOpCancel() {} @@ -463,9 +486,7 @@ module.exports = { canCopyArrayBuffer, cloneAsUint8Array, copyArrayBuffer, - createPromiseCallbackNoParams, createPromiseCallback1Param, - createPromiseCallback2Params, createRawCallback1Param, createRawCallback2Params, customInspect, @@ -489,9 +510,9 @@ module.exports = { materializeQueue, nonOpCallback, nonOpCancel, - nonOpFlush, - peekQueueValue, + promiseCall1Param, + promiseCallNoParams, rejectedHandledRecord, resetQueue, resolvedRecord, diff --git a/lib/internal/webstreams/writablestream.js b/lib/internal/webstreams/writablestream.js index 7b6a7327944b..882073608040 100644 --- a/lib/internal/webstreams/writablestream.js +++ b/lib/internal/webstreams/writablestream.js @@ -36,6 +36,7 @@ const { } = require('internal/util'); const { + validateFunction, validateObject, kValidateObjectAllowObjects, kValidateObjectAllowObjectsAndNull, @@ -54,8 +55,6 @@ const { const { Queue, - createPromiseCallbackNoParams, - createPromiseCallback1Param, createRawCallback2Params, customInspect, defaultSizeAlgorithm, @@ -72,8 +71,9 @@ const { kType, lazyTransfer, nonOpCallback, - nonOpCancel, peekQueueValue, + promiseCall1Param, + promiseCallNoParams, rejectedHandledRecord, resetQueue, resolvedRecord, @@ -512,7 +512,10 @@ class WritableStreamDefaultController { } [kAbort](reason) { - const result = this[kState].abortAlgorithm(reason, this); + const { abortAlgorithm, underlyingSink } = this[kState]; + const result = underlyingSink === undefined ? + abortAlgorithm(reason, this) : + promiseCall1Param(abortAlgorithm, underlyingSink, reason); writableStreamDefaultControllerClearAlgorithms(this); return result; } @@ -565,6 +568,7 @@ function createWritableStream(start, write, close, abort, highWaterMark = 1, siz abort, highWaterMark, size, + undefined, ); return stream; } @@ -1221,11 +1225,14 @@ function writableStreamDefaultControllerProcessClose(controller) { closeAlgorithm, queue, stream, + underlyingSink, } = controller[kState]; writableStreamMarkCloseRequestInFlight(stream); dequeueValue(controller); assert(!queue.length); - const sinkClosePromise = closeAlgorithm(controller); + const sinkClosePromise = underlyingSink === undefined ? + closeAlgorithm(controller) : + promiseCallNoParams(closeAlgorithm, underlyingSink); writableStreamDefaultControllerClearAlgorithms(controller); PromisePrototypeThen( sinkClosePromise, @@ -1296,6 +1303,7 @@ function writableStreamDefaultControllerClearAlgorithms(controller) { controller[kState].closeAlgorithm = undefined; controller[kState].abortAlgorithm = undefined; controller[kState].sizeAlgorithm = undefined; + controller[kState].underlyingSink = undefined; } function writableStreamDefaultControllerAdvanceQueueIfNeeded(controller) { @@ -1338,21 +1346,20 @@ function setupWritableStreamDefaultControllerFromSink( const writeAlgorithm = write ? createRawCallback2Params('sink.write', write, sink) : nonOpCallback; - const closeAlgorithm = close ? - createPromiseCallbackNoParams('sink.close', close, sink) : - nonOpCancel; - const abortAlgorithm = abort ? - createPromiseCallback1Param('sink.abort', abort, sink) : - nonOpCancel; + if (close) + validateFunction(close, 'sink.close'); + if (abort) + validateFunction(abort, 'sink.abort'); setupWritableStreamDefaultController( stream, controller, startAlgorithm, writeAlgorithm, - closeAlgorithm, - abortAlgorithm, + close || nonOpCallback, + abort || nonOpCallback, highWaterMark, - sizeAlgorithm); + sizeAlgorithm, + sink ?? null); } function setupWritableStreamDefaultController( @@ -1363,7 +1370,8 @@ function setupWritableStreamDefaultController( closeAlgorithm, abortAlgorithm, highWaterMark, - sizeAlgorithm) { + sizeAlgorithm, + underlyingSink) { assert(isWritableStream(stream)); assert(stream[kState].controller === undefined); controller[kState] = { @@ -1379,6 +1387,9 @@ function setupWritableStreamDefaultController( // The TransformStream this controller is the sink of, if any; its // shared sink algorithms reach the stream through it. transformStream: undefined, + // The object the close and abort algorithms are methods of; see + // setupReadableStreamDefaultController. + underlyingSink, writeAlgorithm, writeFulfilled: undefined, writeRejected: undefined, diff --git a/test/parallel/test-whatwg-webstreams-underlying-methods.js b/test/parallel/test-whatwg-webstreams-underlying-methods.js new file mode 100644 index 000000000000..1fb437fec057 --- /dev/null +++ b/test/parallel/test-whatwg-webstreams-underlying-methods.js @@ -0,0 +1,137 @@ +'use strict'; + +const common = require('../common'); +const assert = require('assert'); +const { + ReadableStream, + WritableStream, + TransformStream, +} = require('stream/web'); + +// Underlying source, sink and transformer methods are read once at +// construction, called as methods of the object they were read from with +// exactly the spec's arguments, and validated up front. + +{ + const source = { + start: common.mustCall(function(...args) { + assert.strictEqual(this, source); + assert.strictEqual(args.length, 1); + }), + pull: common.mustCall(function(...args) { + assert.strictEqual(this, source); + assert.strictEqual(args.length, 1); + args[0].enqueue('a'); + }), + cancel: common.mustCall(function(...args) { + assert.strictEqual(this, source); + assert.deepStrictEqual(args, ['why']); + }), + }; + const rs = new ReadableStream(source); + source.pull = common.mustNotCall(); + source.cancel = common.mustNotCall(); + const reader = rs.getReader(); + (async () => { + const { value } = await reader.read(); + assert.strictEqual(value, 'a'); + await reader.cancel('why'); + })().then(common.mustCall()); +} + +{ + const sink = { + start: common.mustCall(function(...args) { + assert.strictEqual(this, sink); + assert.strictEqual(args.length, 1); + }), + write: common.mustCall(function(...args) { + assert.strictEqual(this, sink); + assert.strictEqual(args.length, 2); + assert.strictEqual(args[0], 'a'); + }), + close: common.mustCall(function(...args) { + assert.strictEqual(this, sink); + assert.strictEqual(args.length, 0); + }), + }; + const ws = new WritableStream(sink); + sink.write = common.mustNotCall(); + sink.close = common.mustNotCall(); + const writer = ws.getWriter(); + (async () => { + await writer.write('a'); + await writer.close(); + })().then(common.mustCall()); +} + +{ + const sink = { + abort: common.mustCall(function(...args) { + assert.strictEqual(this, sink); + assert.deepStrictEqual(args, ['why']); + }), + }; + const ws = new WritableStream(sink); + sink.abort = common.mustNotCall(); + ws.abort('why').then(common.mustCall()); +} + +{ + const transformer = { + transform: common.mustCall(function(...args) { + assert.strictEqual(this, transformer); + assert.strictEqual(args.length, 2); + args[1].enqueue(args[0]); + }), + flush: common.mustCall(function(...args) { + assert.strictEqual(this, transformer); + assert.strictEqual(args.length, 1); + }), + }; + const ts = new TransformStream(transformer); + transformer.transform = common.mustNotCall(); + transformer.flush = common.mustNotCall(); + const writer = ts.writable.getWriter(); + const reader = ts.readable.getReader(); + writer.write('a'); + writer.close(); + (async () => { + const { value } = await reader.read(); + assert.strictEqual(value, 'a'); + const { done } = await reader.read(); + assert.strictEqual(done, true); + })().then(common.mustCall()); +} + +{ + const transformer = { + cancel: common.mustCall(function(...args) { + assert.strictEqual(this, transformer); + assert.deepStrictEqual(args, ['why']); + }), + }; + const ts = new TransformStream(transformer); + transformer.cancel = common.mustNotCall(); + ts.readable.cancel('why').then(common.mustCall()); +} + +// A method that is present but not callable is rejected by the constructor. +for (const key of ['pull', 'cancel']) { + assert.throws(() => new ReadableStream({ [key]: 1 }), { + code: 'ERR_INVALID_ARG_TYPE', + }); + assert.throws(() => new ReadableStream({ type: 'bytes', [key]: 1 }), { + code: 'ERR_INVALID_ARG_TYPE', + }); +} +for (const key of ['write', 'close', 'abort']) { + assert.throws(() => new WritableStream({ [key]: 1 }), { + code: 'ERR_INVALID_ARG_TYPE', + }); +} +for (const key of ['transform', 'flush', 'cancel']) { + assert.throws(() => new TransformStream({ [key]: 1 }), { + code: 'ERR_INVALID_ARG_TYPE', + }); +}