stream: speed up WHATWG web streams

.github/workflows/benchmark.yml

@@ -141,7 +141,7 @@ jobs:
             --arg devTools '[]' \
             --arg benchmarkTools '[]' \
             --run '
-                make -j4 V=1
+                make build-ci -j4 V=1
             '
 
       - name: Run benchmark

lib/internal/webstreams/readablestream.js

@@ -109,7 +109,9 @@ const {
   extractSizeAlgorithm,
   getNonWritablePropertyDescriptor,
   isBrandCheck,
+  isNonThenable,
   kEmptyQueue,
+  promiseFromAlgorithmResult,
   kState,
   kType,
   lazyTransfer,
@@ -250,6 +252,13 @@ class ReadableStream {
    */
   constructor(source = kEmptyObject, strategy = kEmptyObject) {
     markTransferMode(this, false, true);
+    // Empty-argument `new ReadableStream()`: no source, no strategy, and
+    // no controller. Reads never deliver data, so skip those allocations
+    // until getReader/cancel/error first need a default controller.
+    if (source === kEmptyObject && strategy === kEmptyObject) {
+      this[kState] = createReadableStreamState();
+      return;
+    }
     validateObject(source, 'source', kValidateObjectAllowObjects);
     validateObject(strategy, 'strategy', kValidateObjectAllowObjectsAndNull);
     this[kState] = createReadableStreamState();
@@ -299,6 +308,11 @@ class ReadableStream {
     // only default controllers were wired here; byte stream controllers
     // keep the previous no-op behavior.
     const controller = this[kState].controller;
+    if (controller === undefined) {
+      if (this[kState].state === 'readable')
+        readableStreamError(this, error);
+      return;
+    }
     if (isReadableStreamDefaultController(controller))
       controller.error(error);
   }
@@ -349,6 +363,7 @@ class ReadableStream {
       return PromiseReject(
         new ERR_INVALID_STATE.TypeError('ReadableStream is locked'));
     }
+    ensureEmptyDefaultController(this);
     return readableStreamCancel(this, reason);
   }
 
@@ -1689,6 +1704,11 @@ function readableStreamPipeTo(
 
     const controller = source[kState].controller;
 
+    if (source[kState].state === 'readable' &&
+        isReadableStreamDefaultController(controller)) {
+      readableStreamDefaultControllerFillSync(controller);
+    }
+
     // Fast path: batch reads when data is buffered in a default controller.
     // This avoids parking read requests and reduces promise allocation
     // overhead.
@@ -2553,6 +2573,7 @@ function setupReadableStreamBYOBReader(reader, stream) {
 function setupReadableStreamDefaultReader(reader, stream) {
   if (isReadableStreamLocked(stream))
     throw new ERR_INVALID_STATE.TypeError('ReadableStream is locked');
+  ensureEmptyDefaultController(stream);
   readableStreamReaderGenericInitialize(reader, stream);
   reader[kState].readRequests = kEmptyQueue;
 }
@@ -2699,12 +2720,64 @@ function readableStreamDefaultControllerPull(controller) {
     controller[kState].pullRejected =
       (error) => readableStreamDefaultControllerError(controller, error);
   }
+  // Non-thenable results use PromiseResolve().then so a throw in
+  // pullFulfilled becomes an unhandled rejection (same as the Then
+  // path below), not an uncaught exception from queueMicrotask.
+  // That reaction is one microtask, matching Then on a fulfilled
+  // async-wrapper promise. This is not a sync pull-to-pull hop;
+  // pipeTo's FillSync is the only path that skips the microtask
+  // between consecutive user pulls.
+  const result = controller[kState].pullAlgorithm(controller);
+  if (isNonThenable(result)) {
+    PromisePrototypeThen(
+      PromiseResolve(),
+      controller[kState].pullFulfilled,
+      controller[kState].pullRejected);
+    return;
+  }
   PromisePrototypeThen(
-    controller[kState].pullAlgorithm(controller),
+    promiseFromAlgorithmResult(result),
     controller[kState].pullFulfilled,
     controller[kState].pullRejected);
 }
 
+// pipeTo-only: fill a default controller's queue by repeatedly invoking a
+// synchronous pull. Not used by the spec pull-fulfillment path (tee and
+// WPT require one pull per microtask there).
+function readableStreamDefaultControllerFillSync(controller) {
+  const state = controller[kState];
+  if (state.pulling || state.pullAlgorithm === undefined)
+    return;
+  if (state.pullFulfilled === undefined) {
+    state.pullFulfilled = () => {
+      state.pulling = false;
+      if (state.pullAgain) {
+        state.pullAgain = false;
+        readableStreamDefaultControllerCallPullIfNeeded(controller);
+      }
+    };
+    state.pullRejected =
+      (error) => readableStreamDefaultControllerError(controller, error);
+  }
+  while (readableStreamDefaultControllerShouldCallPull(controller)) {
+    state.pulling = true;
+    const before = state.queue.length;
+    const result = state.pullAlgorithm(controller);
+    if (isNonThenable(result)) {
+      state.pulling = false;
+      state.pullAgain = false;
+      if (state.queue.length === before && !state.closeRequested)
+        return;
+      continue;
+    }
+    PromisePrototypeThen(
+      promiseFromAlgorithmResult(result),
+      state.pullFulfilled,
+      state.pullRejected);
+    return;
+  }
+}
+
 function readableStreamDefaultControllerClearAlgorithms(controller) {
   controller[kState].pullAlgorithm = undefined;
   controller[kState].cancelAlgorithm = undefined;
@@ -2726,7 +2799,7 @@ function readableStreamDefaultControllerCancelSteps(controller, reason) {
   resetQueue(controller);
   const result = controller[kState].cancelAlgorithm(reason);
   readableStreamDefaultControllerClearAlgorithms(controller);
-  return result;
+  return promiseFromAlgorithmResult(result);
 }
 
 function readableStreamDefaultControllerPullSteps(controller, readRequest) {
@@ -2759,6 +2832,31 @@ function readableStreamDefaultControllerPullSteps(controller, readRequest) {
     readableStreamDefaultControllerPull(controller);
 }
 
+// Materialize the deferred default controller for `new ReadableStream()`.
+// started is true immediately: start is a no-op and there is no initial pull.
+function ensureEmptyDefaultController(stream) {
+  if (stream[kState].controller !== undefined)
+    return stream[kState].controller;
+  const controller = new ReadableStreamDefaultController(kSkipThrow);
+  controller[kState] = {
+    cancelAlgorithm: nonOpCancel,
+    closeRequested: false,
+    highWaterMark: 1,
+    pullAgain: false,
+    pullAlgorithm: nonOpPull,
+    pulling: false,
+    pullFulfilled: undefined,
+    pullRejected: undefined,
+    queue: kEmptyQueue,
+    queueTotalSize: 0,
+    started: true,
+    sizeAlgorithm: defaultSizeAlgorithm,
+    stream,
+  };
+  stream[kState].controller = controller;
+  return controller;
+}
+
 function setupReadableStreamDefaultController(
   stream,
   controller,
@@ -2787,8 +2885,7 @@ function setupReadableStreamDefaultController(
 
   const startResult = startAlgorithm();
 
-  if (startResult === null ||
-      (typeof startResult !== 'object' && typeof startResult !== 'function')) {
+  if (isNonThenable(startResult)) {
     // 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
@@ -3519,8 +3616,16 @@ function readableByteStreamControllerCallPullIfNeeded(controller) {
     controller[kState].pullRejected =
       (error) => readableByteStreamControllerError(controller, error);
   }
+  const result = controller[kState].pullAlgorithm(controller);
+  if (isNonThenable(result)) {
+    PromisePrototypeThen(
+      PromiseResolve(),
+      controller[kState].pullFulfilled,
+      controller[kState].pullRejected);
+    return;
+  }
   PromisePrototypeThen(
-    controller[kState].pullAlgorithm(controller),
+    promiseFromAlgorithmResult(result),
     controller[kState].pullFulfilled,
     controller[kState].pullRejected);
 }
@@ -3542,7 +3647,7 @@ function readableByteStreamControllerCancelSteps(controller, reason) {
   resetQueue(controller);
   const result = controller[kState].cancelAlgorithm(reason);
   readableByteStreamControllerClearAlgorithms(controller);
-  return result;
+  return promiseFromAlgorithmResult(result);
 }
 
 // Dequeues the first chunk of the byte queue as a Uint8Array view,
@@ -3664,8 +3769,7 @@ function setupReadableByteStreamController(
 
   const startResult = startAlgorithm();
 
-  if (startResult === null ||
-      (typeof startResult !== 'object' && typeof startResult !== 'function')) {
+  if (isNonThenable(startResult)) {
     // See setupReadableStreamDefaultController.
     queueMicrotask(() => {
       controller[kState].started = true;

lib/internal/webstreams/transformstream.js

@@ -54,6 +54,7 @@ const {
   kType,
   nonOpCancel,
   nonOpFlush,
+  delayedAlgorithmResult,
 } = require('internal/webstreams/util');
 
 const {
@@ -123,9 +124,14 @@ class TransformStream {
     writableStrategy = kEmptyObject,
     readableStrategy = kEmptyObject) {
     markTransferMode(this, false, true);
-    validateObject(transformer, 'transformer', kValidateObjectAllowObjects);
-    validateObject(writableStrategy, 'writableStrategy', kValidateObjectAllowObjectsAndNull);
-    validateObject(readableStrategy, 'readableStrategy', kValidateObjectAllowObjectsAndNull);
+    if (transformer !== kEmptyObject)
+      validateObject(transformer, 'transformer', kValidateObjectAllowObjects);
+    if (writableStrategy !== kEmptyObject) {
+      validateObject(writableStrategy, 'writableStrategy', kValidateObjectAllowObjectsAndNull);
+    }
+    if (readableStrategy !== kEmptyObject) {
+      validateObject(readableStrategy, 'readableStrategy', kValidateObjectAllowObjectsAndNull);
+    }
     const readableType = transformer?.readableType;
     const writableType = transformer?.writableType;
     const start = transformer?.start;
@@ -348,7 +354,7 @@ const isTransformStream =
 const isTransformStreamDefaultController =
   isBrandCheck('TransformStreamDefaultController');
 
-async function defaultTransformAlgorithm(chunk, controller) {
+function defaultTransformAlgorithm(chunk, controller) {
   transformStreamDefaultControllerEnqueue(controller, chunk);
 }
 
@@ -589,15 +595,16 @@ async function transformStreamDefaultSinkAbortAlgorithm(stream, reason) {
 
   const { promise, resolve, reject } = PromiseWithResolvers();
   controller[kState].finishPromise = promise;
-  const cancelPromise = controller[kState].cancelAlgorithm(reason);
+  const cancelPromise =
+    delayedAlgorithmResult(controller[kState].cancelAlgorithm(reason));
   transformStreamDefaultControllerClearAlgorithms(controller);
 
   PromisePrototypeThen(
     cancelPromise,
     () => {
-      if (readable[kState].state === 'errored')
+      if (readable[kState].state === 'errored') {
         reject(readable[kState].storedError);
-      else {
+      } else {
         readableStreamDefaultControllerError(readable[kState].controller, reason);
         resolve();
       }
@@ -622,7 +629,8 @@ function transformStreamDefaultSinkCloseAlgorithm(stream) {
   }
   const { promise, resolve, reject } = PromiseWithResolvers();
   controller[kState].finishPromise = promise;
-  const flushPromise = controller[kState].flushAlgorithm(controller);
+  const flushPromise =
+    delayedAlgorithmResult(controller[kState].flushAlgorithm(controller));
   transformStreamDefaultControllerClearAlgorithms(controller);
   PromisePrototypeThen(
     flushPromise,
@@ -659,15 +667,16 @@ function transformStreamDefaultSourceCancelAlgorithm(stream, reason) {
 
   const { promise, resolve, reject } = PromiseWithResolvers();
   controller[kState].finishPromise = promise;
-  const cancelPromise = controller[kState].cancelAlgorithm(reason);
+  const cancelPromise =
+    delayedAlgorithmResult(controller[kState].cancelAlgorithm(reason));
   transformStreamDefaultControllerClearAlgorithms(controller);
 
   PromisePrototypeThen(
     cancelPromise,
     () => {
-      if (writable[kState].state === 'errored')
+      if (writable[kState].state === 'errored') {
         reject(writable[kState].storedError);
-      else {
+      } else {
         writableStreamDefaultControllerErrorIfNeeded(
           writable[kState].controller,
           reason);