From 5e6c052dd70620aae4335e25be06ec8d36e62697 Mon Sep 17 00:00:00 2001 From: Matteo Collina Date: Sun, 20 Sep 2026 11:22:26 +0200 Subject: [PATCH] stream: trim per-pipe and per-tee costs in webstreams An object literal with computed symbol keys is rebuilt through the runtime every time it is evaluated, close to a microsecond each. pipeTo, tee and the byte tee materialized their internal read request as such a literal once per pipe or tee, and the byte tee's BYOB path did so on every read. The requests are now instances of one small class holding the three step functions. A source without pull() ran the pull bookkeeping anyway: two reaction closures and a microtask per read whose only effect was to clear the pulling flag. Push-style sources now skip it. The default tee's pull algorithm was an async function and the byte tee's returned a fresh resolved promise; both now return nothing, which reaches the controller's pull-fulfilled step at the same microtask position without the promise. ReadableStreamBYOBReader.prototype.read() was an async method, so the read request's promise was adopted through a wrapper (an extra promise and two microtask hops per read). It now returns the request's promise directly, as the spec does; argument errors still become rejections. A TransformStream without start() no longer allocates the start promise record: the sides adopt a promise that is already resolved either way. Signed-off-by: Matteo Collina --- lib/internal/webstreams/readablestream.js | 190 +++++++++++++-------- lib/internal/webstreams/transformstream.js | 15 +- 2 files changed, 127 insertions(+), 78 deletions(-) diff --git a/lib/internal/webstreams/readablestream.js b/lib/internal/webstreams/readablestream.js index bf1449fd9042..3af9fd24b66f 100644 --- a/lib/internal/webstreams/readablestream.js +++ b/lib/internal/webstreams/readablestream.js @@ -876,6 +876,31 @@ class DefaultReadRequest { get promise() { return this[kState].promise; } } +// Read (into) request record for the internal consumers (pipeTo, tee). +// The step functions are per-consumer closures, but the record is a class +// instance rather than an object literal: a literal with computed symbol +// keys is rebuilt through the runtime on every evaluation, which costs +// microseconds per pipe or tee (and per read in the byte tee's BYOB path). +class StepsReadRequest { + constructor(chunkSteps, closeSteps, errorSteps) { + this.chunkSteps = chunkSteps; + this.closeSteps = closeSteps; + this.errorSteps = errorSteps; + } + + [kChunk](chunk) { + this.chunkSteps(chunk); + } + + [kClose](chunk) { + this.closeSteps(chunk); + } + + [kError](error) { + this.errorSteps(error); + } +} + class ReadIntoRequest { constructor() { this[kState] = PromiseWithResolvers(); @@ -1053,53 +1078,15 @@ class ReadableStreamBYOBReader { * done : boolean, * }>} */ - async read(view, options = kEmptyObject) { - if (!isReadableStreamBYOBReader(this)) - throw new ERR_INVALID_THIS('ReadableStreamBYOBReader'); - validateBuffer(view, 'view'); - validateObject(options, 'options', kValidateObjectAllowObjectsAndNull); - - const viewByteLength = ArrayBufferViewGetByteLength(view); - const viewBuffer = ArrayBufferViewGetBuffer(view); - - if (isSharedArrayBuffer(viewBuffer)) { - throw new ERR_INVALID_ARG_VALUE( - 'view', - view, - 'must not be backed by a SharedArrayBuffer', - ); - } - - const viewBufferByteLength = ArrayBufferPrototypeGetByteLength(viewBuffer); - - if (viewByteLength === 0 || viewBufferByteLength === 0) { - throw new ERR_INVALID_STATE.TypeError( - 'View or Viewed ArrayBuffer is zero-length or detached'); - } - - // Supposed to assert here that the view's buffer is not - // detached, but there's no API available to use to check that. - - const min = options?.min ?? 1; - validateNumber(min, 'options.min'); - if (!NumberIsInteger(min)) - throw new ERR_INVALID_ARG_VALUE('options.min', min, 'must be an integer'); - if (min <= 0) - throw new ERR_INVALID_ARG_VALUE('options.min', min, 'must be greater than 0'); - if (!isDataView(view)) { - if (min > TypedArrayPrototypeGetLength(view)) { - throw new ERR_OUT_OF_RANGE('options.min', '<= view.length', min); - } - } else if (min > viewByteLength) { - throw new ERR_OUT_OF_RANGE('options.min', '<= view.byteLength', min); - } - - if (this[kState].stream === undefined) { - throw new ERR_INVALID_STATE.TypeError('The reader is not attached to a stream'); + read(view, options = kEmptyObject) { + // Returns the read request's promise directly, as the spec does, + // instead of adopting it through an async wrapper (an extra promise + // and two microtask hops per read); argument errors become rejections. + try { + return readableStreamBYOBReaderReadView(this, view, options); + } catch (error) { + return PromiseReject(error); } - const readIntoRequest = new ReadIntoRequest(); - readableStreamBYOBReaderRead(this, view, min, readIntoRequest); - return readIntoRequest.promise; } releaseLock() { @@ -1804,17 +1791,16 @@ function readableStreamPipeTo( // Slow path: park a lazily materialized read request. Close and // error are handled by the source watchers. - readRequest ??= { - [kChunk](chunk) { + readRequest ??= new StepsReadRequest( + (chunk) => { // Per spec, pipeTo must queue a microtask for the write to avoid // synchronous write during enqueue(). See WHATWG Streams spec // "ReadableStreamPipeTo" step 15's "chunk steps". pendingChunk = chunk; PromisePrototypeThen(kResolvedPromise, forwardChunk); }, - [kClose]() {}, - [kError]() {}, - }; + nonOpCallback, + nonOpCallback); readableStreamDefaultReaderRead(reader, readRequest); } @@ -1931,17 +1917,20 @@ function readableStreamDefaultTee(stream, cloneForBranch2) { } } - async function pullAlgorithm() { + // A non-thenable pull result reaches the controller's pull-fulfilled + // step at the same microtask position as the async wrapper's promise + // did, without the implicit promise per pull. + function pullAlgorithm() { if (reading) return; reading = true; - readRequest ??= { - [kChunk](value) { + readRequest ??= new StepsReadRequest( + (value) => { // The microtask is required by the spec (ReadableStreamTee's // "chunk steps" queue one). pendingChunk = value; PromisePrototypeThen(kResolvedPromise, forwardChunk); }, - [kClose]() { + () => { // The `process.nextTick()` is not part of the spec. // This approach was needed to avoid a race condition working with esm // Further information, see: https://github.com/nodejs/node/issues/39758 @@ -1955,10 +1944,9 @@ function readableStreamDefaultTee(stream, cloneForBranch2) { cancelPromise.resolve(); }); }, - [kError]() { + () => { reading = false; - }, - }; + }); readableStreamDefaultReaderRead(reader, readRequest); } @@ -2089,12 +2077,12 @@ function readableByteStreamTee(stream) { forwardReaderError(reader); } - defaultReadRequest ??= { - [kChunk](chunk) { + defaultReadRequest ??= new StepsReadRequest( + (chunk) => { pendingChunk = chunk; PromisePrototypeThen(kResolvedPromise, forwardChunk); }, - [kClose]() { + () => { reading = false; if (!canceled1) { @@ -2113,10 +2101,9 @@ function readableByteStreamTee(stream) { cancelDeferred.resolve(); } }, - [kError]() { + () => { reading = false; - }, - }; + }); readableStreamDefaultReaderRead(reader, defaultReadRequest); } @@ -2129,8 +2116,8 @@ function readableByteStreamTee(stream) { const byobBranch = forBranch2 === true ? branch2 : branch1; const otherBranch = forBranch2 === false ? branch2 : branch1; - const readIntoRequest = { - [kChunk](chunk) { + const readIntoRequest = new StepsReadRequest( + (chunk) => { queueMicrotask(() => { readAgainForBranch1 = false; readAgainForBranch2 = false; @@ -2180,7 +2167,7 @@ function readableByteStreamTee(stream) { } }); }, - [kClose](chunk) { + (chunk) => { reading = false; const byobCanceled = forBranch2 === true ? canceled2 : canceled1; @@ -2213,17 +2200,16 @@ function readableByteStreamTee(stream) { cancelDeferred.resolve(); } }, - [kError]() { + () => { reading = false; - }, - }; + }); readableStreamBYOBReaderRead(reader, view, 1, readIntoRequest); } function pull1Algorithm() { if (reading) { readAgainForBranch1 = true; - return PromiseResolve(); + return; } reading = true; @@ -2233,13 +2219,12 @@ function readableByteStreamTee(stream) { } else { pullWithBYOBReader(byobRequest[kState].view, false); } - return PromiseResolve(); } function pull2Algorithm() { if (reading) { readAgainForBranch2 = true; - return PromiseResolve(); + return; } reading = true; @@ -2249,7 +2234,6 @@ function readableByteStreamTee(stream) { } else { pullWithBYOBReader(byobRequest[kState].view, true); } - return PromiseResolve(); } function cancel1Algorithm(reason) { @@ -2581,6 +2565,56 @@ function readableStreamReaderGenericRelease(reader) { reader[kState].stream = undefined; } +// The argument validation half of ReadableStreamBYOBReader.prototype.read(). +function readableStreamBYOBReaderReadView(reader, view, options) { + if (!isReadableStreamBYOBReader(reader)) + throw new ERR_INVALID_THIS('ReadableStreamBYOBReader'); + validateBuffer(view, 'view'); + validateObject(options, 'options', kValidateObjectAllowObjectsAndNull); + + const viewByteLength = ArrayBufferViewGetByteLength(view); + const viewBuffer = ArrayBufferViewGetBuffer(view); + + if (isSharedArrayBuffer(viewBuffer)) { + throw new ERR_INVALID_ARG_VALUE( + 'view', + view, + 'must not be backed by a SharedArrayBuffer', + ); + } + + const viewBufferByteLength = ArrayBufferPrototypeGetByteLength(viewBuffer); + + if (viewByteLength === 0 || viewBufferByteLength === 0) { + throw new ERR_INVALID_STATE.TypeError( + 'View or Viewed ArrayBuffer is zero-length or detached'); + } + + // Supposed to assert here that the view's buffer is not + // detached, but there's no API available to use to check that. + + const min = options?.min ?? 1; + validateNumber(min, 'options.min'); + if (!NumberIsInteger(min)) + throw new ERR_INVALID_ARG_VALUE('options.min', min, 'must be an integer'); + if (min <= 0) + throw new ERR_INVALID_ARG_VALUE('options.min', min, 'must be greater than 0'); + if (!isDataView(view)) { + if (min > TypedArrayPrototypeGetLength(view)) { + throw new ERR_OUT_OF_RANGE('options.min', '<= view.length', min); + } + } else if (min > viewByteLength) { + throw new ERR_OUT_OF_RANGE('options.min', '<= view.byteLength', min); + } + + if (reader[kState].stream === undefined) { + throw new ERR_INVALID_STATE.TypeError('The reader is not attached to a stream'); + } + const readIntoRequest = new ReadIntoRequest(); + readableStreamBYOBReaderRead(reader, view, min, readIntoRequest); + return readIntoRequest.promise; +} + function readableStreamBYOBReaderRead(reader, view, min, readIntoRequest) { const { stream, @@ -2761,6 +2795,11 @@ function readableStreamDefaultControllerCallPullIfNeeded(controller) { // that have already established the predicate from state in scope (the // enqueue path above) can skip re-running it. function readableStreamDefaultControllerPull(controller) { + // A source without pull() has nothing to run: skip the pulling/pullAgain + // bookkeeping, the reaction closures and the microtask that would only + // clear the flag again. Push-style sources hit this on every read. + if (controller[kState].pullAlgorithm === nonOpCallback) + return; if (controller[kState].pulling) { controller[kState].pullAgain = true; return; @@ -3590,6 +3629,9 @@ function readableByteStreamControllerShiftPendingPullInto(controller) { function readableByteStreamControllerCallPullIfNeeded(controller) { if (!readableByteStreamControllerShouldCallPull(controller)) return; + // See readableStreamDefaultControllerPull. + if (controller[kState].pullAlgorithm === nonOpCallback) + return; if (controller[kState].pulling) { controller[kState].pullAgain = true; return; diff --git a/lib/internal/webstreams/transformstream.js b/lib/internal/webstreams/transformstream.js index 85c118474894..25771fb6a558 100644 --- a/lib/internal/webstreams/transformstream.js +++ b/lib/internal/webstreams/transformstream.js @@ -159,7 +159,10 @@ class TransformStream { extractHighWaterMark(writableHighWaterMark, 1); const actualWritableSize = extractSizeAlgorithm(writableSize); - const startPromise = PromiseWithResolvers(); + // 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, @@ -177,8 +180,6 @@ class TransformStream { start, transformer, this[kState].controller)); - } else { - startPromise.resolve(); } } @@ -378,6 +379,10 @@ function defaultTransformAlgorithm(chunk, controller) { transformStreamDefaultControllerEnqueue(controller, chunk); } +function resolvedStartAlgorithm() { + return kResolvedPromise; +} + function initializeTransformStream( stream, startPromise, @@ -386,7 +391,9 @@ function initializeTransformStream( readableHighWaterMark, readableSizeAlgorithm) { - const startAlgorithm = () => startPromise.promise; + const startAlgorithm = startPromise === undefined ? + resolvedStartAlgorithm : + () => startPromise.promise; const writable = createWritableStream( startAlgorithm,