stream: trim per-stream costs in webstreams

@@ -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,15 +1829,34 @@ function readableStreamPipeTo(    pump(); -  watchErrored(source, readerClosedPromise(reader).promise, (error) => {+  function onSourceErrored(error) {     if (!preventAbort) {       return shutdownWithAnAction(         () => writableStreamAbort(dest, error),         true,         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)); } 

@@ -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;

@@ -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,33 +552,18 @@ 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,     abort,     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');