stream: keep webstream stream states in fast-mode objects

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;
+  }
+}

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) {

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);
 }

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) {