From 5e8671107e0b0a80cb21bdb6936334a8f069e550 Mon Sep 17 00:00:00 2001 From: Matteo Collina Date: Fri, 28 Aug 2026 20:18:51 +0200 Subject: [PATCH 1/2] stream: keep webstream stream states in fast-mode objects MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The per-stream state records were built as object literals with __proto__: null, which V8 creates as dictionary-mode objects: roughly 50x slower to allocate, and every subsequent property load on them is a dictionary lookup. These records back every hot path, so both stream construction and per-chunk field accesses were paying for it. Replace the literals with classes whose prototype has a null prototype, so instances stay in fast mode while Object.prototype remains excluded from the lookup chain. Every field ever assigned is declared up front so the shape never transitions. Also add benchmark/webstreams/lifecycle.js covering the short-lived stream pattern (create, few chunks, close) that first exposed this. confidence improvement accuracy (*) (**) (***) webstreams/creation.js kind='ReadableStream' n=50000 *** 138.13 % ±14.29% ±19.14% ±25.17% webstreams/creation.js kind='TransformStream' n=50000 *** 133.64 % ±8.61% ±11.46% ±14.94% webstreams/creation.js kind='WritableStream' n=50000 *** 204.00 % ±7.70% ±10.26% ±13.37% webstreams/lifecycle.js kind='pipe-through' n=50000 *** 96.41 % ±3.41% ±4.58% ±6.03% webstreams/lifecycle.js kind='readable' n=50000 *** 80.04 % ±3.10% ±4.15% ±5.45% webstreams/pipe-through.js kind='default' n=500000 *** 91.16 % ±2.56% ±3.43% ±4.52% webstreams/pipe-to.js highWaterMarkW=1 highWaterMarkR=1 n=500000 *** 110.48 % ±2.91% ±3.89% ±5.11% webstreams/readable-read.js type='normal' n=100000 *** 34.63 % ±4.79% ±6.38% ±8.31% Signed-off-by: Matteo Collina --- benchmark/webstreams/lifecycle.js | 69 +++++++++++++++++++++ lib/internal/webstreams/readablestream.js | 42 ++++++++----- lib/internal/webstreams/transformstream.js | 71 ++++++++++++---------- lib/internal/webstreams/writablestream.js | 65 +++++++++++--------- 4 files changed, 172 insertions(+), 75 deletions(-) create mode 100644 benchmark/webstreams/lifecycle.js diff --git a/benchmark/webstreams/lifecycle.js b/benchmark/webstreams/lifecycle.js new file mode 100644 index 000000000000..05d188eed149 --- /dev/null +++ b/benchmark/webstreams/lifecycle.js @@ -0,0 +1,69 @@ +'use strict'; +const common = require('../common.js'); +const { + ReadableStream, + WritableStream, + TransformStream, +} = require('node:stream/web'); + +const bench = common.createBenchmark(main, { + n: [5e4], + kind: ['readable', 'pipe-to', 'pipe-through'], +}); + +const chunk = Buffer.alloc(1024); + +function makeSource() { + let i = 0; + return { + pull(controller) { + if (i++ < 4) + controller.enqueue(chunk); + else + controller.close(); + }, + }; +} + +async function readable(n) { + bench.start(); + for (let i = 0; i < n; i++) { + const reader = new ReadableStream(makeSource()).getReader(); + while (!(await reader.read()).done); + } + bench.end(n); +} + +async function pipeTo(n) { + bench.start(); + for (let i = 0; i < n; i++) { + await new ReadableStream(makeSource()) + .pipeTo(new WritableStream({ write() {} })); + } + bench.end(n); +} + +async function pipeThrough(n) { + bench.start(); + for (let i = 0; i < n; i++) { + const reader = new ReadableStream(makeSource()) + .pipeThrough(new TransformStream()) + .getReader(); + while (!(await reader.read()).done); + } + bench.end(n); +} + +function main({ n, kind }) { + switch (kind) { + case 'readable': + readable(n); + break; + case 'pipe-to': + pipeTo(n); + break; + case 'pipe-through': + pipeThrough(n); + break; + } +} diff --git a/lib/internal/webstreams/readablestream.js b/lib/internal/webstreams/readablestream.js index 78313bd03c1a..d4e3d7809a25 100644 --- a/lib/internal/webstreams/readablestream.js +++ b/lib/internal/webstreams/readablestream.js @@ -1419,22 +1419,34 @@ const isReadableStreamBYOBReader = // ---- ReadableStream Implementation +// The state records are classes whose prototype chain ends at null +// instead of `__proto__: null` object literals: the literals fall back +// to dictionary-mode objects in V8 (~50x slower to create, and every +// later property load is a dictionary lookup), while class instances +// stay in fast mode with the same protection against Object.prototype +// pollution. Every field ever assigned is declared so the shape never +// transitions. +class ReadableStreamTransferState { + writable = undefined; + port1 = undefined; + port2 = undefined; + promise = undefined; +} +ObjectSetPrototypeOf(ReadableStreamTransferState.prototype, null); + +class ReadableStreamState { + closedPromise = undefined; + disturbed = false; + reader = undefined; + state = 'readable'; + storedError = undefined; + controller = undefined; + transfer = new ReadableStreamTransferState(); +} +ObjectSetPrototypeOf(ReadableStreamState.prototype, null); + function createReadableStreamState() { - return { - __proto__: null, - closedPromise: undefined, - disturbed: false, - reader: undefined, - state: 'readable', - storedError: undefined, - transfer: { - __proto__: null, - writable: undefined, - port1: undefined, - port2: undefined, - promise: undefined, - }, - }; + return new ReadableStreamState(); } function readableStreamFromIterable(iterable) { diff --git a/lib/internal/webstreams/transformstream.js b/lib/internal/webstreams/transformstream.js index 30b7b1c8fac1..2591a7ddcd98 100644 --- a/lib/internal/webstreams/transformstream.js +++ b/lib/internal/webstreams/transformstream.js @@ -253,21 +253,39 @@ ObjectDefineProperties(TransformStream.prototype, { [SymbolToStringTag]: getNonWritablePropertyDescriptor(TransformStream.name), }); +// A class with a null prototype chain instead of a `__proto__: null` +// literal: the literal produces a dictionary-mode object (slow to +// create, slow property loads), the class instance stays in fast mode +// with the same protection against Object.prototype pollution. +class TransformStreamState { + readable = undefined; + writable = undefined; + controller = undefined; + backpressure = undefined; + // Continuation slots replacing the spec's + // [[backpressureChangePromise]]; see transformStreamSetBackpressure. + pullPending = false; + pendingWrite = undefined; + pendingWriteChunk = undefined; + writeContinuation = undefined; +} +ObjectSetPrototypeOf(TransformStreamState.prototype, null); + +class TransformStreamControllerState { + stream = undefined; + transformAlgorithm = undefined; + flushAlgorithm = undefined; + cancelAlgorithm = undefined; + performTransformRejected = undefined; + finishPromise = undefined; +} +ObjectSetPrototypeOf(TransformStreamControllerState.prototype, null); + function InternalTransferredTransformStream() { ObjectSetPrototypeOf(this, TransformStream.prototype); markTransferMode(this, false, true); this[kType] = 'TransformStream'; - this[kState] = { - __proto__: null, - readable: undefined, - writable: undefined, - backpressure: undefined, - pullPending: false, - pendingWrite: undefined, - pendingWriteChunk: undefined, - writeContinuation: undefined, - controller: undefined, - }; + this[kState] = new TransformStreamState(); } ObjectSetPrototypeOf(InternalTransferredTransformStream.prototype, TransformStream.prototype); @@ -388,19 +406,10 @@ function initializeTransformStream( readableSizeAlgorithm, ); - stream[kState] = { - __proto__: null, - readable, - writable, - controller: undefined, - backpressure: undefined, - // Continuation slots replacing the spec's - // [[backpressureChangePromise]]; see transformStreamSetBackpressure. - pullPending: false, - pendingWrite: undefined, - pendingWriteChunk: undefined, - writeContinuation: undefined, - }; + const state = new TransformStreamState(); + state.readable = readable; + state.writable = writable; + stream[kState] = state; transformStreamSetBackpressure(stream, true); } @@ -470,14 +479,12 @@ function setupTransformStreamDefaultController( cancelAlgorithm) { assert(isTransformStream(stream)); assert(stream[kState].controller === undefined); - controller[kState] = { - __proto__: null, - stream, - transformAlgorithm, - flushAlgorithm, - cancelAlgorithm, - performTransformRejected: undefined, - }; + const controllerState = new TransformStreamControllerState(); + controllerState.stream = stream; + controllerState.transformAlgorithm = transformAlgorithm; + controllerState.flushAlgorithm = flushAlgorithm; + controllerState.cancelAlgorithm = cancelAlgorithm; + controller[kState] = controllerState; stream[kState].controller = controller; } diff --git a/lib/internal/webstreams/writablestream.js b/lib/internal/webstreams/writablestream.js index 1e9ca02cfe96..da6364befe64 100644 --- a/lib/internal/webstreams/writablestream.js +++ b/lib/internal/webstreams/writablestream.js @@ -596,35 +596,44 @@ const isWritableStreamDefaultWriter = const isWritableStreamDefaultController = isBrandCheck('WritableStreamDefaultController'); +// Classes with a null prototype chain instead of `__proto__: null` +// literals: the literals produce dictionary-mode objects (slow to +// create, slow property loads), class instances stay in fast mode with +// the same protection against Object.prototype pollution. Every field +// ever assigned is declared so the shape never transitions. +class WritableStreamTransferState { + readable = undefined; + port1 = undefined; + port2 = undefined; + promise = undefined; +} +ObjectSetPrototypeOf(WritableStreamTransferState.prototype, null); + +class WritableStreamState { + closedPromise = undefined; + closeRequest = kNilRequest; + // Mirrors "closeRequest or inFlightCloseRequest is pending"; kept as a + // flag because the predicate runs several times per chunk on the write + // hot path. + closeQueuedOrInFlight = false; + inFlightWriteRequest = kNilRequest; + inFlightCloseRequest = kNilRequest; + pendingAbortRequest = kNilPendingAbortRequest; + backpressure = false; + controller = undefined; + state = 'writable'; + storedError = undefined; + // Ring-buffer request queue, materialized lazily on the first pending + // write (see writableStreamAddWriteRequest) so construction allocates + // no request storage. + writeRequests = kEmptyQueue; + writer = undefined; + transfer = new WritableStreamTransferState(); +} +ObjectSetPrototypeOf(WritableStreamState.prototype, null); + function createWritableStreamState() { - return { - __proto__: null, - closedPromise: undefined, - closeRequest: kNilRequest, - // Mirrors "closeRequest or inFlightCloseRequest is pending"; kept as a - // flag because the predicate runs several times per chunk on the write - // hot path. - closeQueuedOrInFlight: false, - inFlightWriteRequest: kNilRequest, - inFlightCloseRequest: kNilRequest, - pendingAbortRequest: kNilPendingAbortRequest, - backpressure: false, - controller: undefined, - state: 'writable', - storedError: undefined, - // Ring-buffer request queue, materialized lazily on the first pending - // write (see writableStreamAddWriteRequest) so construction allocates - // no request storage. - writeRequests: kEmptyQueue, - writer: undefined, - transfer: { - __proto__: null, - readable: undefined, - port1: undefined, - port2: undefined, - promise: undefined, - }, - }; + return new WritableStreamState(); } function isWritableStreamLocked(stream) { From 36d534f4b7e6e79c0c6aef8c2c8424ab65e85eb5 Mon Sep 17 00:00:00 2001 From: Matteo Collina Date: Fri, 28 Aug 2026 20:19:01 +0200 Subject: [PATCH 2/2] stream: avoid promise allocation for parked transform writes A transform sink write arriving under backpressure parked the chunk together with a PromiseWithResolvers record whose promise was returned to the writable controller and later resolved with the perform-transform promise. The writable's write reactions already exist before the write algorithm runs, so the parked write can instead return the parked-result sentinel and have the continuation wire the perform-transform promise directly to those reactions, dropping the per-chunk promise record and the thenable adoption hop. Failures while erroring are delivered in a microtask, preserving the old rejection position. Signed-off-by: Matteo Collina --- lib/internal/webstreams/transformstream.js | 44 ++++++++++++---------- 1 file changed, 25 insertions(+), 19 deletions(-) diff --git a/lib/internal/webstreams/transformstream.js b/lib/internal/webstreams/transformstream.js index 2591a7ddcd98..7f0bec847642 100644 --- a/lib/internal/webstreams/transformstream.js +++ b/lib/internal/webstreams/transformstream.js @@ -265,7 +265,7 @@ class TransformStreamState { // Continuation slots replacing the spec's // [[backpressureChangePromise]]; see transformStreamSetBackpressure. pullPending = false; - pendingWrite = undefined; + pendingWriteParked = false; pendingWriteChunk = undefined; writeContinuation = undefined; } @@ -466,7 +466,7 @@ function transformStreamSetBackpressure(stream, backpressure) { kResolvedPromise, state.readable[kState].controller[kState].pullFulfilled); } - } else if (state.pendingWrite !== undefined) { + } else if (state.pendingWriteParked) { PromisePrototypeThen(kResolvedPromise, state.writeContinuation); } } @@ -610,35 +610,41 @@ function transformStreamDefaultSinkWriteAlgorithm(stream, chunk) { } = state; assert(writable[kState].state === 'writable'); 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; + // 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 pending = state.pendingWrite; const pendingChunk = state.pendingWriteChunk; - state.pendingWrite = undefined; + state.pendingWriteParked = false; state.pendingWriteChunk = undefined; const writableState = state.writable[kState]; + const writableControllerState = writableState.controller[kState]; if (writableState.state === 'erroring') { - pending.reject(writableState.storedError); + const error = writableState.storedError; + PromisePrototypeThen( + kResolvedPromise, + () => writableControllerState.writeRejected(error)); return; } assert(writableState.state === 'writable'); - pending.resolve( + PromisePrototypeThen( transformStreamDefaultControllerPerformTransform( controller, - pendingChunk)); + pendingChunk), + writableControllerState.writeFulfilled, + writableControllerState.writeRejected); }; - return pendingWrite.promise; + return kParkedAlgorithmResult; } return transformStreamDefaultControllerPerformTransform(controller, chunk); }