stream: trim per-pipe and per-tee costs in webstreams

@@ -876,6 +876,31 @@ class DefaultReadRequest {   get promise() { return this[kState].promise; } } +// Read (into) request record for the internal consumers (pipeTo, tee).+// The step functions are per-consumer closures, but the record is a class+// instance rather than an object literal: a literal with computed symbol+// keys is rebuilt through the runtime on every evaluation, which costs+// microseconds per pipe or tee (and per read in the byte tee's BYOB path).+class StepsReadRequest {+  constructor(chunkSteps, closeSteps, errorSteps) {+    this.chunkSteps = chunkSteps;+    this.closeSteps = closeSteps;+    this.errorSteps = errorSteps;+  }++  [kChunk](chunk) {+    this.chunkSteps(chunk);+  }++  [kClose](chunk) {+    this.closeSteps(chunk);+  }++  [kError](error) {+    this.errorSteps(error);+  }+}+ class ReadIntoRequest {   constructor() {     this[kState] = PromiseWithResolvers();@@ -1053,53 +1078,15 @@ class ReadableStreamBYOBReader {    *   done : boolean,    * }>}    */-  async read(view, options = kEmptyObject) {-    if (!isReadableStreamBYOBReader(this))-      throw new ERR_INVALID_THIS('ReadableStreamBYOBReader');-    validateBuffer(view, 'view');-    validateObject(options, 'options', kValidateObjectAllowObjectsAndNull);--    const viewByteLength = ArrayBufferViewGetByteLength(view);-    const viewBuffer = ArrayBufferViewGetBuffer(view);--    if (isSharedArrayBuffer(viewBuffer)) {-      throw new ERR_INVALID_ARG_VALUE(-        'view',-        view,-        'must not be backed by a SharedArrayBuffer',-      );-    }--    const viewBufferByteLength = ArrayBufferPrototypeGetByteLength(viewBuffer);--    if (viewByteLength === 0 || viewBufferByteLength === 0) {-      throw new ERR_INVALID_STATE.TypeError(-        'View or Viewed ArrayBuffer is zero-length or detached');-    }--    // Supposed to assert here that the view's buffer is not-    // detached, but there's no API available to use to check that.--    const min = options?.min ?? 1;-    validateNumber(min, 'options.min');-    if (!NumberIsInteger(min))-      throw new ERR_INVALID_ARG_VALUE('options.min', min, 'must be an integer');-    if (min <= 0)-      throw new ERR_INVALID_ARG_VALUE('options.min', min, 'must be greater than 0');-    if (!isDataView(view)) {-      if (min > TypedArrayPrototypeGetLength(view)) {-        throw new ERR_OUT_OF_RANGE('options.min', '<= view.length', min);-      }-    } else if (min > viewByteLength) {-      throw new ERR_OUT_OF_RANGE('options.min', '<= view.byteLength', min);-    }--    if (this[kState].stream === undefined) {-      throw new ERR_INVALID_STATE.TypeError('The reader is not attached to a stream');+  read(view, options = kEmptyObject) {+    // Returns the read request's promise directly, as the spec does,+    // instead of adopting it through an async wrapper (an extra promise+    // and two microtask hops per read); argument errors become rejections.+    try {+      return readableStreamBYOBReaderReadView(this, view, options);+    } catch (error) {+      return PromiseReject(error);     }-    const readIntoRequest = new ReadIntoRequest();-    readableStreamBYOBReaderRead(this, view, min, readIntoRequest);-    return readIntoRequest.promise;   }    releaseLock() {@@ -1804,17 +1791,16 @@ function readableStreamPipeTo(      // Slow path: park a lazily materialized read request. Close and     // error are handled by the source watchers.-    readRequest ??= {-      [kChunk](chunk) {+    readRequest ??= new StepsReadRequest(+      (chunk) => {         // Per spec, pipeTo must queue a microtask for the write to avoid         // synchronous write during enqueue(). See WHATWG Streams spec         // "ReadableStreamPipeTo" step 15's "chunk steps".         pendingChunk = chunk;         PromisePrototypeThen(kResolvedPromise, forwardChunk);       },-      [kClose]() {},-      [kError]() {},-    };+      nonOpCallback,+      nonOpCallback);     readableStreamDefaultReaderRead(reader, readRequest);   } @@ -1931,17 +1917,20 @@ function readableStreamDefaultTee(stream, cloneForBranch2) {     }   } -  async function pullAlgorithm() {+  // A non-thenable pull result reaches the controller's pull-fulfilled+  // step at the same microtask position as the async wrapper's promise+  // did, without the implicit promise per pull.+  function pullAlgorithm() {     if (reading) return;     reading = true;-    readRequest ??= {-      [kChunk](value) {+    readRequest ??= new StepsReadRequest(+      (value) => {         // The microtask is required by the spec (ReadableStreamTee's         // "chunk steps" queue one).         pendingChunk = value;         PromisePrototypeThen(kResolvedPromise, forwardChunk);       },-      [kClose]() {+      () => {         // The `process.nextTick()` is not part of the spec.         // This approach was needed to avoid a race condition working with esm         // Further information, see: https://github.com/nodejs/node/issues/39758@@ -1955,10 +1944,9 @@ function readableStreamDefaultTee(stream, cloneForBranch2) {             cancelPromise.resolve();         });       },-      [kError]() {+      () => {         reading = false;-      },-    };+      });     readableStreamDefaultReaderRead(reader, readRequest);   } @@ -2089,12 +2077,12 @@ function readableByteStreamTee(stream) {       forwardReaderError(reader);     } -    defaultReadRequest ??= {-      [kChunk](chunk) {+    defaultReadRequest ??= new StepsReadRequest(+      (chunk) => {         pendingChunk = chunk;         PromisePrototypeThen(kResolvedPromise, forwardChunk);       },-      [kClose]() {+      () => {         reading = false;          if (!canceled1) {@@ -2113,10 +2101,9 @@ function readableByteStreamTee(stream) {           cancelDeferred.resolve();         }       },-      [kError]() {+      () => {         reading = false;-      },-    };+      });     readableStreamDefaultReaderRead(reader, defaultReadRequest);   } @@ -2129,8 +2116,8 @@ function readableByteStreamTee(stream) {      const byobBranch = forBranch2 === true ? branch2 : branch1;     const otherBranch = forBranch2 === false ? branch2 : branch1;-    const readIntoRequest = {-      [kChunk](chunk) {+    const readIntoRequest = new StepsReadRequest(+      (chunk) => {         queueMicrotask(() => {           readAgainForBranch1 = false;           readAgainForBranch2 = false;@@ -2180,7 +2167,7 @@ function readableByteStreamTee(stream) {           }         });       },-      [kClose](chunk) {+      (chunk) => {         reading = false;          const byobCanceled = forBranch2 === true ? canceled2 : canceled1;@@ -2213,17 +2200,16 @@ function readableByteStreamTee(stream) {           cancelDeferred.resolve();         }       },-      [kError]() {+      () => {         reading = false;-      },-    };+      });     readableStreamBYOBReaderRead(reader, view, 1, readIntoRequest);   }    function pull1Algorithm() {     if (reading) {       readAgainForBranch1 = true;-      return PromiseResolve();+      return;     }     reading = true; @@ -2233,13 +2219,12 @@ function readableByteStreamTee(stream) {     } else {       pullWithBYOBReader(byobRequest[kState].view, false);     }-    return PromiseResolve();   }    function pull2Algorithm() {     if (reading) {       readAgainForBranch2 = true;-      return PromiseResolve();+      return;     }     reading = true; @@ -2249,7 +2234,6 @@ function readableByteStreamTee(stream) {     } else {       pullWithBYOBReader(byobRequest[kState].view, true);     }-    return PromiseResolve();   }    function cancel1Algorithm(reason) {@@ -2581,6 +2565,56 @@ function readableStreamReaderGenericRelease(reader) {   reader[kState].stream = undefined; } +// The argument validation half of ReadableStreamBYOBReader.prototype.read().+function readableStreamBYOBReaderReadView(reader, view, options) {+  if (!isReadableStreamBYOBReader(reader))+    throw new ERR_INVALID_THIS('ReadableStreamBYOBReader');+  validateBuffer(view, 'view');+  validateObject(options, 'options', kValidateObjectAllowObjectsAndNull);++  const viewByteLength = ArrayBufferViewGetByteLength(view);+  const viewBuffer = ArrayBufferViewGetBuffer(view);++  if (isSharedArrayBuffer(viewBuffer)) {+    throw new ERR_INVALID_ARG_VALUE(+      'view',+      view,+      'must not be backed by a SharedArrayBuffer',+    );+  }++  const viewBufferByteLength = ArrayBufferPrototypeGetByteLength(viewBuffer);++  if (viewByteLength === 0 || viewBufferByteLength === 0) {+    throw new ERR_INVALID_STATE.TypeError(+      'View or Viewed ArrayBuffer is zero-length or detached');+  }++  // Supposed to assert here that the view's buffer is not+  // detached, but there's no API available to use to check that.++  const min = options?.min ?? 1;+  validateNumber(min, 'options.min');+  if (!NumberIsInteger(min))+    throw new ERR_INVALID_ARG_VALUE('options.min', min, 'must be an integer');+  if (min <= 0)+    throw new ERR_INVALID_ARG_VALUE('options.min', min, 'must be greater than 0');+  if (!isDataView(view)) {+    if (min > TypedArrayPrototypeGetLength(view)) {+      throw new ERR_OUT_OF_RANGE('options.min', '<= view.length', min);+    }+  } else if (min > viewByteLength) {+    throw new ERR_OUT_OF_RANGE('options.min', '<= view.byteLength', min);+  }++  if (reader[kState].stream === undefined) {+    throw new ERR_INVALID_STATE.TypeError('The reader is not attached to a stream');+  }+  const readIntoRequest = new ReadIntoRequest();+  readableStreamBYOBReaderRead(reader, view, min, readIntoRequest);+  return readIntoRequest.promise;+}+ function readableStreamBYOBReaderRead(reader, view, min, readIntoRequest) {   const {     stream,@@ -2761,6 +2795,11 @@ function readableStreamDefaultControllerCallPullIfNeeded(controller) { // that have already established the predicate from state in scope (the // enqueue path above) can skip re-running it. function readableStreamDefaultControllerPull(controller) {+  // A source without pull() has nothing to run: skip the pulling/pullAgain+  // bookkeeping, the reaction closures and the microtask that would only+  // clear the flag again. Push-style sources hit this on every read.+  if (controller[kState].pullAlgorithm === nonOpCallback)+    return;   if (controller[kState].pulling) {     controller[kState].pullAgain = true;     return;@@ -3590,6 +3629,9 @@ function readableByteStreamControllerShiftPendingPullInto(controller) { function readableByteStreamControllerCallPullIfNeeded(controller) {   if (!readableByteStreamControllerShouldCallPull(controller))     return;+  // See readableStreamDefaultControllerPull.+  if (controller[kState].pullAlgorithm === nonOpCallback)+    return;   if (controller[kState].pulling) {     controller[kState].pullAgain = true;     return;

@@ -159,7 +159,10 @@ class TransformStream {       extractHighWaterMark(writableHighWaterMark, 1);     const actualWritableSize = extractSizeAlgorithm(writableSize); -    const startPromise = PromiseWithResolvers();+    // Without a start() the start promise is already resolved by the time+    // the readable and writable sides adopt it, so the shared resolved+    // promise stands in for the record.+    const startPromise = start !== undefined ? PromiseWithResolvers() : undefined;      initializeTransformStream(       this,@@ -177,8 +180,6 @@ class TransformStream {           start,           transformer,           this[kState].controller));-    } else {-      startPromise.resolve();     }   } @@ -378,6 +379,10 @@ function defaultTransformAlgorithm(chunk, controller) {   transformStreamDefaultControllerEnqueue(controller, chunk); } +function resolvedStartAlgorithm() {+  return kResolvedPromise;+}+ function initializeTransformStream(   stream,   startPromise,@@ -386,7 +391,9 @@ function initializeTransformStream(   readableHighWaterMark,   readableSizeAlgorithm) { -  const startAlgorithm = () => startPromise.promise;+  const startAlgorithm = startPromise === undefined ?+    resolvedStartAlgorithm :+    () => startPromise.promise;    const writable = createWritableStream(     startAlgorithm,