stream: reduce webstreams encoding and iteration overhead

benchmark/webstreams/encoding-streams.js

@@ -0,0 +1,39 @@
+'use strict';
+const common = require('../common.js');
+const {
+  ReadableStream,
+  TextEncoderStream,
+  TextDecoderStream,
+} = require('node:stream/web');
+
+const bench = common.createBenchmark(main, {
+  n: [1e5],
+  kind: ['encode', 'decode'],
+  len: [16, 1024],
+});
+
+async function main({ n, kind, len }) {
+  const encoded = new TextEncoder().encode('a'.repeat(len));
+  const decoded = 'a'.repeat(len);
+  let i = 0;
+  const rs = new ReadableStream({
+    pull(controller) {
+      if (i++ < n) {
+        controller.enqueue(kind === 'encode' ? decoded : encoded);
+      } else {
+        controller.close();
+      }
+    },
+  });
+  const ts = kind === 'encode' ?
+    new TextEncoderStream() :
+    new TextDecoderStream();
+
+  const reader = rs.pipeThrough(ts).getReader();
+  bench.start();
+  for (;;) {
+    const { done } = await reader.read();
+    if (done) break;
+  }
+  bench.end(n);
+}

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

benchmark/webstreams/pipe-through.js

@@ -0,0 +1,38 @@
+'use strict';
+const common = require('../common.js');
+const {
+  ReadableStream,
+  TransformStream,
+} = require('node:stream/web');
+
+const bench = common.createBenchmark(main, {
+  n: [5e5],
+  kind: ['default', 'transform'],
+});
+
+async function main({ n, kind }) {
+  const b = Buffer.alloc(64);
+  let i = 0;
+  const rs = new ReadableStream({
+    pull(controller) {
+      if (i++ < n) {
+        controller.enqueue(b);
+      } else {
+        controller.close();
+      }
+    },
+  });
+  const ts = kind === 'default' ?
+    new TransformStream() :
+    new TransformStream({
+      transform(chunk, controller) { controller.enqueue(chunk); },
+    });
+
+  const reader = rs.pipeThrough(ts).getReader();
+  bench.start();
+  for (;;) {
+    const { done } = await reader.read();
+    if (done) break;
+  }
+  bench.end(n);
+}

lib/internal/webstreams/encoding.js

@@ -4,6 +4,7 @@ const {
   ObjectDefineProperties,
   String,
   StringPrototypeCharCodeAt,
+  StringPrototypeSlice,
   Uint8Array,
 } = primordials;
 
@@ -31,6 +32,9 @@ const {
   kEnumerableProperty,
 } = require('internal/util');
 
+// Shared per-chunk decode options; decode() only reads the flag.
+const kDecodeStreamingOptions = { __proto__: null, stream: true };
+
 /**
  * @typedef {import('./readablestream').ReadableStream} ReadableStream
  * @typedef {import('./writablestream').WritableStream} WritableStream
@@ -46,34 +50,26 @@ class TextEncoderStream {
     this.#transform = new TransformStream({
       transform: (chunk, controller) => {
         // https://encoding.spec.whatwg.org/#encode-and-enqueue-a-chunk
+        // The only cross-chunk state is a trailing high surrogate;
+        // encode() replaces interior lone surrogates with U+FFFD exactly
+        // like the spec's per-code-unit walk.
         chunk = String(chunk);
-        let finalChunk = '';
-        for (let i = 0; i < chunk.length; i++) {
-          const item = chunk[i];
-          const codeUnit = StringPrototypeCharCodeAt(item, 0);
-          if (this.#pendingHighSurrogate !== null) {
-            const highSurrogate = this.#pendingHighSurrogate;
-            this.#pendingHighSurrogate = null;
-            if (0xDC00 <= codeUnit && codeUnit <= 0xDFFF) {
-              finalChunk += highSurrogate + item;
-              continue;
-            }
-            finalChunk += '\uFFFD';
-          }
-          if (0xD800 <= codeUnit && codeUnit <= 0xDBFF) {
-            this.#pendingHighSurrogate = item;
-            continue;
-          }
-          if (0xDC00 <= codeUnit && codeUnit <= 0xDFFF) {
-            finalChunk += '\uFFFD';
-            continue;
-          }
-          finalChunk += item;
+        if (chunk.length === 0)
+          return;
+        if (this.#pendingHighSurrogate !== null) {
+          chunk = this.#pendingHighSurrogate + chunk;
+          this.#pendingHighSurrogate = null;
         }
-        if (finalChunk) {
-          const value = this.#handle.encode(finalChunk);
-          controller.enqueue(value);
+        const lastCodeUnit =
+          StringPrototypeCharCodeAt(chunk, chunk.length - 1);
+        if (0xD800 <= lastCodeUnit && lastCodeUnit <= 0xDBFF) {
+          this.#pendingHighSurrogate =
+            StringPrototypeSlice(chunk, -1);
+          chunk = StringPrototypeSlice(chunk, 0, -1);
+          if (chunk.length === 0)
+            return;
         }
+        controller.enqueue(this.#handle.encode(chunk));
       },
       flush: (controller) => {
         // https://encoding.spec.whatwg.org/#encode-and-flush
@@ -137,7 +133,7 @@ class TextDecoderStream {
         if (chunk === undefined) {
           throw new ERR_INVALID_ARG_TYPE('chunk', 'string', chunk);
         }
-        const value = this.#handle.decode(chunk, { stream: true });
+        const value = this.#handle.decode(chunk, kDecodeStreamingOptions);
         if (value)
           controller.enqueue(value);
       },

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