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..7f0bec847642 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; + pendingWriteParked = false; + 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); } @@ -457,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); } } @@ -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; } @@ -603,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); } 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) {