diff --git a/benchmark/webstreams/encoding-streams.js b/benchmark/webstreams/encoding-streams.js new file mode 100644 index 000000000000..00759bc09eb7 --- /dev/null +++ b/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); +} diff --git a/benchmark/webstreams/from.js b/benchmark/webstreams/from.js new file mode 100644 index 000000000000..05eca4079f1d --- /dev/null +++ b/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); +} diff --git a/lib/internal/webstreams/encoding.js b/lib/internal/webstreams/encoding.js index f316222ccbf0..038b64030a7a 100644 --- a/lib/internal/webstreams/encoding.js +++ b/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); }, diff --git a/lib/internal/webstreams/readablestream.js b/lib/internal/webstreams/readablestream.js index 84f16dc7b4a1..78313bd03c1a 100644 --- a/lib/internal/webstreams/readablestream.js +++ b/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) {