stream: avoid async pull machinery in ReadableStream.from

The pull algorithm was an async function that awaited iterator.next()
and then the produced value, costing an async-function frame plus two
await wrappers and a controller-side reaction per chunk. Rewrite it
callback-style: the next() result is adopted exactly like the previous
awaits (including the observable .then lookup on plain object values),
the reaction steps are created once per stream, and completion is
delivered straight to the controller's cached pull reactions using the
parked-algorithm-result contract. The iterator's next method is also
looked up once at setup, per the spec's GetIteratorDirect.

A from benchmark is added since the suite had no ReadableStream.from
row. Iterating a stream built from a sync generator improves by ~27%
and from an async generator by ~23%; all other rows are unchanged.

Signed-off-by: Matteo Collina <hello@matteocollina.com>

Author: mcollina

benchmark/webstreams/from.js

@@ -0,0 +1,29 @@
+'use strict';
+const common = require('../common.js');
+const {
+  ReadableStream,
+} = require('node:stream/web');
+
+const bench = common.createBenchmark(main, {
+  n: [1e6],
+  kind: ['sync', 'async'],
+});
+
+async function main({ n, kind }) {
+  function* syncGen() {
+    for (let i = 0; i < n; i++) yield i;
+  }
+
+  async function* asyncGen() {
+    for (let i = 0; i < n; i++) yield i;
+  }
+
+  const reader = ReadableStream.from(
+    kind === 'sync' ? syncGen() : asyncGen()).getReader();
+  bench.start();
+  for (;;) {
+    const { done } = await reader.read();
+    if (done) break;
+  }
+  bench.end(n);
+}

lib/internal/webstreams/readablestream.js

@@ -112,6 +112,7 @@ const {
   getNonWritablePropertyDescriptor,
   isBrandCheck,
   kEmptyQueue,
+  kParkedAlgorithmResult,
   kResolvedPromise,
   kState,
   kType,
@@ -1446,19 +1447,85 @@ function readableStreamFromIterable(iterable) {
   if (iterator === null || (typeof iterator !== 'object' && typeof iterator !== 'function')) {
     throw new ERR_INVALID_STATE.TypeError('The iterator method must return an object');
   }
+  // Per GetIteratorDirect, the next method is looked up once.
+  const nextMethod = iterator.next;
   const startAlgorithm = nonOpCallback;
 
-  async function pullAlgorithm() {
-    const iterResult = await iterator.next();
+  // Callback-style pull: the reaction steps are reused across chunks and
+  // completion is delivered to the controller's cached pull reactions
+  // (the kParkedAlgorithmResult contract). One pull runs at a time, so a
+  // single slot carries a non-thenable next() result between steps.
+  let pendingIterResult;
+
+  function rejectPull(error) {
+    readableStreamDefaultControllerError(stream[kState].controller, error);
+  }
+
+  function processIterResult(iterResult) {
+    const controller = stream[kState].controller;
     if (typeof iterResult !== 'object' || iterResult === null) {
-      throw new ERR_INVALID_STATE.TypeError(
-        'The promise returned by the iterator.next() method must fulfill with an object');
+      rejectPull(new ERR_INVALID_STATE.TypeError(
+        'The promise returned by the iterator.next() method must fulfill with an object'));
+      return;
     }
-    if (iterResult.done) {
-      readableStreamDefaultControllerClose(stream[kState].controller);
-    } else {
-      readableStreamDefaultControllerEnqueue(stream[kState].controller, await iterResult.value);
+    try {
+      if (iterResult.done) {
+        readableStreamDefaultControllerClose(controller);
+      } else {
+        const value = iterResult.value;
+        if (value !== null &&
+            (typeof value === 'object' || typeof value === 'function')) {
+          // Adopted like `await iterResult.value`, keeping the observable
+          // .then lookup on plain objects.
+          PromisePrototypeThen(PromiseResolve(value), enqueueValue, rejectPull);
+          return;
+        }
+        readableStreamDefaultControllerEnqueue(controller, value);
+      }
+    } catch (error) {
+      rejectPull(error);
+      return;
+    }
+    // pullFulfilled exists: the controller creates it before the pull.
+    controller[kState].pullFulfilled();
+  }
+
+  function enqueueValue(value) {
+    const controller = stream[kState].controller;
+    try {
+      readableStreamDefaultControllerEnqueue(controller, value);
+    } catch (error) {
+      rejectPull(error);
+      return;
+    }
+    controller[kState].pullFulfilled();
+  }
+
+  function processPendingIterResult() {
+    const iterResult = pendingIterResult;
+    pendingIterResult = undefined;
+    processIterResult(iterResult);
+  }
+
+  function pullAlgorithm() {
+    let nextResult;
+    try {
+      nextResult = FunctionPrototypeCall(nextMethod, iterator);
+    } catch (error) {
+      return PromiseReject(error);
+    }
+    if (nextResult !== null &&
+        (typeof nextResult === 'object' || typeof nextResult === 'function')) {
+      // Mirrors `await iterator.next()`: processIterResult runs at the
+      // microtask position the await resumed.
+      PromisePrototypeThen(
+        PromiseResolve(nextResult), processIterResult, rejectPull);
+      return kParkedAlgorithmResult;
     }
+    // A non-thenable next() result fails validation a microtask later.
+    pendingIterResult = nextResult;
+    PromisePrototypeThen(kResolvedPromise, processPendingIterResult);
+    return kParkedAlgorithmResult;
   }
 
   async function cancelAlgorithm(reason) {