From 906ea04390ffb9222b8f33275595f15352cdd930 Mon Sep 17 00:00:00 2001 From: Matteo Collina Date: Wed, 16 Sep 2026 02:35:43 +0200 Subject: [PATCH] stream: trim per-stream costs in webstreams Short-lived streams (create, a few chunks, close) pay a fixed cost per stream that dominates once the per-chunk path is lean. Streams created internally (transform stream sides, tee branches, ReadableStream.from, transferred streams) were built by wrapper constructors that swapped the prototype of every instance and then assigned an own, enumerable `constructor` property to look like a public stream. Each internal stream therefore had its own hidden class and `Object.keys(stream)` reported `['constructor']`. The public constructors now accept the internal construction sentinel and leave controller setup to the caller, so every ReadableStream and WritableStream shares one hidden class and no per-instance prototype swap or own property is needed. The queue ring buffer grew after a push filled it, so the initial 8-slot ring held only three (value, size) pairs and a four-chunk stream reallocated every time. Growing before the push lets the ring hold four pairs. pipeTo observed the source's closed promise with two reactions and, on teardown, let the reader and writer release paths probe and reject promise records that only the pipe could have observed. One reaction pair now watches the source, and finalize drops the records before release. Signed-off-by: Matteo Collina --- lib/internal/webstreams/readablestream.js | 163 ++++++++---------- lib/internal/webstreams/util.js | 16 +- lib/internal/webstreams/writablestream.js | 74 +++----- ...whatwg-webstreams-internal-construction.js | 53 ++++++ 4 files changed, 156 insertions(+), 150 deletions(-) create mode 100644 test/parallel/test-whatwg-webstreams-internal-construction.js diff --git a/lib/internal/webstreams/readablestream.js b/lib/internal/webstreams/readablestream.js index ae2ec7972ae3..bf1449fd9042 100644 --- a/lib/internal/webstreams/readablestream.js +++ b/lib/internal/webstreams/readablestream.js @@ -253,6 +253,13 @@ class ReadableStream { */ constructor(source = kEmptyObject, strategy = kEmptyObject) { markTransferMode(this, false, true); + // Internal construction (tee, transform streams, adapters, transfer): + // the caller sets up the controller, so every ReadableStream shares + // one hidden class and no per-instance prototype swap is needed. + if (source === kSkipThrow) { + this[kState] = createReadableStreamState(); + return; + } validateObject(source, 'source', kValidateObjectAllowObjects); validateObject(strategy, 'strategy', kValidateObjectAllowObjectsAndNull); this[kState] = createReadableStreamState(); @@ -718,22 +725,8 @@ ObjectDefineProperties(ReadableStream, { from: kEnumerableProperty, }); -function InternalTransferredReadableStream() { - ObjectSetPrototypeOf(this, ReadableStream.prototype); - markTransferMode(this, false, true); - this[kType] = 'ReadableStream'; - this[kState] = createReadableStreamState(); -} - -ObjectSetPrototypeOf(InternalTransferredReadableStream.prototype, ReadableStream.prototype); -ObjectSetPrototypeOf(InternalTransferredReadableStream, ReadableStream); - function TransferredReadableStream() { - const stream = new InternalTransferredReadableStream(); - - stream.constructor = ReadableStream; - - return stream; + return new ReadableStream(kSkipThrow); } TransferredReadableStream.prototype[kDeserialize] = () => {}; @@ -1350,57 +1343,29 @@ ObjectDefineProperties(ReadableByteStreamController.prototype, { [SymbolToStringTag]: getNonWritablePropertyDescriptor(ReadableByteStreamController.name), }); -function InternalReadableStream(start, pull, cancel, highWaterMark, size) { - ObjectSetPrototypeOf(this, ReadableStream.prototype); - markTransferMode(this, false, true); - this[kType] = 'ReadableStream'; - this[kState] = createReadableStreamState(); - const controller = new ReadableStreamDefaultController(kSkipThrow); +function createReadableStream(start, pull, cancel, highWaterMark = 1, size = defaultSizeAlgorithm) { + const stream = new ReadableStream(kSkipThrow); setupReadableStreamDefaultController( - this, - controller, + stream, + new ReadableStreamDefaultController(kSkipThrow), start, pull, cancel, highWaterMark, size); -} - -ObjectSetPrototypeOf(InternalReadableStream.prototype, ReadableStream.prototype); -ObjectSetPrototypeOf(InternalReadableStream, ReadableStream); - -function createReadableStream(start, pull, cancel, highWaterMark = 1, size = defaultSizeAlgorithm) { - const stream = new InternalReadableStream(start, pull, cancel, highWaterMark, size); - - // For spec compliance the InternalReadableStream must be a ReadableStream - stream.constructor = ReadableStream; return stream; } -function InternalReadableByteStream(start, pull, cancel) { - ObjectSetPrototypeOf(this, ReadableStream.prototype); - markTransferMode(this, false, true); - this[kType] = 'ReadableStream'; - this[kState] = createReadableStreamState(); - const controller = new ReadableByteStreamController(kSkipThrow); +function createReadableByteStream(start, pull, cancel) { + const stream = new ReadableStream(kSkipThrow); setupReadableByteStreamController( - this, - controller, + stream, + new ReadableByteStreamController(kSkipThrow), start, pull, cancel, 0, undefined); -} - -ObjectSetPrototypeOf(InternalReadableByteStream.prototype, ReadableStream.prototype); -ObjectSetPrototypeOf(InternalReadableByteStream, ReadableStream); - -function createReadableByteStream(start, pull, cancel) { - const stream = new InternalReadableByteStream(start, pull, cancel); - - // For spec compliance the InternalReadableByteStream must be a ReadableStream - stream.constructor = ReadableStream; return stream; } @@ -1630,6 +1595,13 @@ function readableStreamPipeTo( // tells us that the promise must be rejected even // when error is undefine. function finalize(rejected, error) { + // The pipe is the only observer of the reader's and writer's promise + // records (including the ready hook installed by parkOnReady), and + // it is done with them: dropping them lets release skip the + // pending-promise probes and the rejections nothing would handle. + writer[kState].ready = undefined; + writer[kState].close = undefined; + reader[kState].close = undefined; writableStreamDefaultWriterRelease(writer); readableStreamReaderGenericRelease(reader); if (signal !== undefined) @@ -1727,12 +1699,6 @@ function readableStreamPipeTo( PromisePrototypeThen(promise, undefined, action); } - function watchClosed(stream, promise, action) { - if (stream[kState].state === 'closed') - action(); - else - PromisePrototypeThen(promise, action, () => {}); - } // The pump loop is callback-driven to avoid per-iteration promise // allocations. At most one read is in flight at a time, so one read @@ -1863,7 +1829,7 @@ function readableStreamPipeTo( pump(); - watchErrored(source, readerClosedPromise(reader).promise, (error) => { + function onSourceErrored(error) { if (!preventAbort) { return shutdownWithAnAction( () => writableStreamAbort(dest, error), @@ -1871,7 +1837,26 @@ function readableStreamPipeTo( error); } shutdown(true, error); - }); + } + + function onSourceClosed() { + if (!preventClose) { + return shutdownWithAnAction( + () => writableStreamDefaultWriterCloseWithErrorPropagation(writer)); + } + shutdown(); + } + + // The spec installs the source-errored watcher before the dest-errored + // one and the source-closed watcher last; a source that is already + // errored is handled before the dest watcher is installed, and an + // already-closed source after it, as before. + if (source[kState].state === 'errored') { + onSourceErrored(source[kState].storedError); + } else if (source[kState].state !== 'closed') { + PromisePrototypeThen( + readerClosedPromise(reader).promise, onSourceClosed, onSourceErrored); + } watchErrored(dest, writerClosedPromise(writer).promise, (error) => { if (!preventCancel) { @@ -1883,13 +1868,8 @@ function readableStreamPipeTo( shutdown(true, error); }); - watchClosed(source, readerClosedPromise(reader).promise, () => { - if (!preventClose) { - return shutdownWithAnAction( - () => writableStreamDefaultWriterCloseWithErrorPropagation(writer)); - } - shutdown(); - }); + if (source[kState].state === 'closed') + onSourceClosed(); if (writableStreamCloseQueuedOrInFlight(dest) || dest[kState].state === 'closed') { @@ -2899,29 +2879,27 @@ function setupReadableStreamDefaultController( const startResult = startAlgorithm(); + const started = () => { + controller[kState].started = true; + assert(!controller[kState].pulling); + assert(!controller[kState].pullAgain); + readableStreamDefaultControllerCallPullIfNeeded(controller); + }; + if (startResult === null || (typeof startResult !== 'object' && typeof startResult !== 'function')) { // Non-thenable start result: fulfillment is guaranteed and no .then - // lookup on the result is observable, so run the post-start step - // directly at the exact microtask position the promise reaction - // would have had, skipping two promise allocations. - queueMicrotask(() => { - controller[kState].started = true; - assert(!controller[kState].pulling); - assert(!controller[kState].pullAgain); - readableStreamDefaultControllerCallPullIfNeeded(controller); - }); + // lookup on the result is observable, so the post-start step runs at + // the exact microtask position the promise reaction would have had. + queueMicrotask(started); return; } + // The wrapper promise matches the reference implementation's + // promiseResolvedWith(), whose extra microtask hops WPT relies on. PromisePrototypeThen( new Promise((r) => r(startResult)), - () => { - controller[kState].started = true; - assert(!controller[kState].pulling); - assert(!controller[kState].pullAgain); - readableStreamDefaultControllerCallPullIfNeeded(controller); - }, + started, (error) => readableStreamDefaultControllerError(controller, error)); } @@ -3783,26 +3761,23 @@ function setupReadableByteStreamController( const startResult = startAlgorithm(); + const started = () => { + controller[kState].started = true; + assert(!controller[kState].pulling); + assert(!controller[kState].pullAgain); + readableByteStreamControllerCallPullIfNeeded(controller); + }; + + // See setupReadableStreamDefaultController. if (startResult === null || (typeof startResult !== 'object' && typeof startResult !== 'function')) { - // See setupReadableStreamDefaultController. - queueMicrotask(() => { - controller[kState].started = true; - assert(!controller[kState].pulling); - assert(!controller[kState].pullAgain); - readableByteStreamControllerCallPullIfNeeded(controller); - }); + queueMicrotask(started); return; } PromisePrototypeThen( new Promise((r) => r(startResult)), - () => { - controller[kState].started = true; - assert(!controller[kState].pulling); - assert(!controller[kState].pullAgain); - readableByteStreamControllerCallPullIfNeeded(controller); - }, + started, (error) => readableByteStreamControllerError(controller, error)); } diff --git a/lib/internal/webstreams/util.js b/lib/internal/webstreams/util.js index 9a93a2b17d41..0c54a7f37593 100644 --- a/lib/internal/webstreams/util.js +++ b/lib/internal/webstreams/util.js @@ -179,12 +179,12 @@ class Queue { // Single-slot entries (readable byte controller chunk records). push(entry) { + if (this.length === this.list.length) + this.grow(); const tail = this.tail; this.list[tail] = entry; this.tail = (tail + 1) & this.capacityMask; this.length++; - if (this.tail === this.head) - this.grow(); } shift() { @@ -207,14 +207,14 @@ class Queue { // never need to wrap. pushPair(value, size) { + if (this.length * 2 === this.list.length) + this.grow(); const tail = this.tail; const list = this.list; list[tail] = value; list[tail + 1] = size; this.tail = (tail + 2) & this.capacityMask; this.length++; - if (this.tail === this.head) - this.grow(); } // Returns the dequeued value; the size of the same entry is left in @@ -237,9 +237,11 @@ class Queue { return this.list[this.head]; } - // The ring is completely full (the post-push tail caught up with the - // head): double the capacity, re-linearizing from the head so index - // arithmetic stays trivial. + // The ring is completely full (the tail has caught up with the head, so + // the next push would overwrite the oldest entry): double the capacity, + // re-linearizing from the head so index arithmetic stays trivial. + // Growing before the push rather than after it lets the initial 8-slot + // ring hold four (value, size) pairs without reallocating. grow() { const list = this.list; const capacity = list.length; diff --git a/lib/internal/webstreams/writablestream.js b/lib/internal/webstreams/writablestream.js index 362a68fc9db9..73d7ed0fbf8a 100644 --- a/lib/internal/webstreams/writablestream.js +++ b/lib/internal/webstreams/writablestream.js @@ -183,6 +183,13 @@ class WritableStream { */ constructor(sink = kEmptyObject, strategy = kEmptyObject) { markTransferMode(this, false, true); + // Internal construction (transform streams, adapters, transfer): + // the caller sets up the controller, so every WritableStream shares + // one hidden class and no per-instance prototype swap is needed. + if (sink === kSkipThrow) { + this[kState] = createWritableStreamState(); + return; + } validateObject(sink, 'sink', kValidateObjectAllowObjects); validateObject(strategy, 'strategy', kValidateObjectAllowObjectsAndNull); const type = sink?.type; @@ -351,22 +358,8 @@ ObjectDefineProperties(WritableStream.prototype, { [SymbolToStringTag]: getNonWritablePropertyDescriptor(WritableStream.name), }); -function InternalTransferredWritableStream() { - ObjectSetPrototypeOf(this, WritableStream.prototype); - markTransferMode(this, false, true); - this[kType] = 'WritableStream'; - this[kState] = createWritableStreamState(); -} - -ObjectSetPrototypeOf(InternalTransferredWritableStream.prototype, WritableStream.prototype); -ObjectSetPrototypeOf(InternalTransferredWritableStream, WritableStream); - function TransferredWritableStream() { - const stream = new InternalTransferredWritableStream(); - - stream.constructor = WritableStream; - - return stream; + return new WritableStream(kSkipThrow); } TransferredWritableStream.prototype[kDeserialize] = () => {}; @@ -559,16 +552,11 @@ ObjectDefineProperties(WritableStreamDefaultController.prototype, { [SymbolToStringTag]: getNonWritablePropertyDescriptor(WritableStreamDefaultController.name), }); -function InternalWritableStream(start, write, close, abort, highWaterMark, size) { - ObjectSetPrototypeOf(this, WritableStream.prototype); - markTransferMode(this, false, true); - this[kType] = 'WritableStream'; - this[kState] = createWritableStreamState(); - - const controller = new WritableStreamDefaultController(kSkipThrow); +function createWritableStream(start, write, close, abort, highWaterMark = 1, size = defaultSizeAlgorithm) { + const stream = new WritableStream(kSkipThrow); setupWritableStreamDefaultController( - this, - controller, + stream, + new WritableStreamDefaultController(kSkipThrow), start, write, close, @@ -576,16 +564,6 @@ function InternalWritableStream(start, write, close, abort, highWaterMark, size) highWaterMark, size, ); -} - -ObjectSetPrototypeOf(InternalWritableStream.prototype, WritableStream.prototype); -ObjectSetPrototypeOf(InternalWritableStream, WritableStream); - -function createWritableStream(start, write, close, abort, highWaterMark = 1, size = defaultSizeAlgorithm) { - const stream = new InternalWritableStream(start, write, close, abort, highWaterMark, size); - - // For spec compliance the InternalWritableStream must be a WritableStream - stream.constructor = WritableStream; return stream; } @@ -1401,29 +1379,27 @@ function setupWritableStreamDefaultController( const startResult = startAlgorithm(); + const started = () => { + assert(stream[kState].state === 'writable' || + stream[kState].state === 'erroring'); + controller[kState].started = true; + writableStreamDefaultControllerAdvanceQueueIfNeeded(controller); + }; + if (startResult === null || (typeof startResult !== 'object' && typeof startResult !== 'function')) { // Non-thenable start result: fulfillment is guaranteed and no .then - // lookup on the result is observable, so run the post-start step - // directly at the exact microtask position the promise reaction - // would have had, skipping two promise allocations. - queueMicrotask(() => { - assert(stream[kState].state === 'writable' || - stream[kState].state === 'erroring'); - controller[kState].started = true; - writableStreamDefaultControllerAdvanceQueueIfNeeded(controller); - }); + // lookup on the result is observable, so the post-start step runs at + // the exact microtask position the promise reaction would have had. + queueMicrotask(started); return; } + // The wrapper promise matches the reference implementation's + // promiseResolvedWith(), whose extra microtask hops WPT relies on. PromisePrototypeThen( new Promise((r) => r(startResult)), - () => { - assert(stream[kState].state === 'writable' || - stream[kState].state === 'erroring'); - controller[kState].started = true; - writableStreamDefaultControllerAdvanceQueueIfNeeded(controller); - }, + started, (error) => { assert(stream[kState].state === 'writable' || stream[kState].state === 'erroring'); diff --git a/test/parallel/test-whatwg-webstreams-internal-construction.js b/test/parallel/test-whatwg-webstreams-internal-construction.js new file mode 100644 index 000000000000..6ff0140c9f10 --- /dev/null +++ b/test/parallel/test-whatwg-webstreams-internal-construction.js @@ -0,0 +1,53 @@ +'use strict'; + +require('../common'); +const assert = require('assert'); +const { + ReadableStream, + WritableStream, + TransformStream, +} = require('stream/web'); + +// Streams created by internal code paths (transform stream sides, tee +// branches, ReadableStream.from, transferred streams) are plain +// ReadableStream/WritableStream instances: same prototype and no own +// properties beyond what the public constructors create. + +function check(stream, Class) { + assert.ok(stream instanceof Class); + assert.strictEqual(Object.getPrototypeOf(stream), Class.prototype); + assert.strictEqual(stream.constructor, Class); + assert.deepStrictEqual(Object.keys(stream), Object.keys(new Class())); + assert.strictEqual( + Object.getOwnPropertyDescriptor(stream, 'constructor'), undefined); +} + +{ + const { readable, writable } = new TransformStream(); + check(readable, ReadableStream); + check(writable, WritableStream); +} + +{ + const [branch1, branch2] = new ReadableStream().tee(); + check(branch1, ReadableStream); + check(branch2, ReadableStream); +} + +{ + const [branch1, branch2] = new ReadableStream({ type: 'bytes' }).tee(); + check(branch1, ReadableStream); + check(branch2, ReadableStream); +} + +check(ReadableStream.from([]), ReadableStream); + +{ + const readable = new ReadableStream(); + const writable = new WritableStream(); + const transferred = structuredClone( + { readable, writable }, + { transfer: [readable, writable] }); + check(transferred.readable, ReadableStream); + check(transferred.writable, WritableStream); +}