stream: encode whole chunks in TextEncoderStream

The encode-and-enqueue transform walked the chunk code unit by code
unit, materializing a single-character string per index and building
the output with string concatenation. The only state that crosses
chunks is a trailing high (leading) surrogate, and TextEncoder.encode's
USVString conversion already replaces every interior lone surrogate
with U+FFFD, which is exactly what the spec loop produces. Join a
pending high surrogate with the incoming chunk, hold back a new
trailing high surrogate, and encode the rest in a single native call.

The streaming decode path also reuses a single options object instead
of allocating { stream: true } per chunk.

An encoding-streams benchmark is added since the suite had no
TextEncoderStream/TextDecoderStream row. Encoding improves by ~546%
with 1KB string chunks and ~20% with 16-character chunks; decode is
unchanged.

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

Author: mcollina

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

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