stream: encode whole chunks in TextEncoderStream · nodejs/node@1d6ec2d · GitHub
Skip to content

Commit 1d6ec2d

Browse files
mcollinaaduh95
mcollina
authored and
aduh95
committed
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> PR-URL: #65414 Reviewed-By: Gürgün Dayıoğlu <hey@gurgun.day> Reviewed-By: Yagiz Nizipli <yagiz@nizipli.com> Reviewed-By: Mattias Buelens <mattias@buelens.com>
1 parent 9e227ef commit 1d6ec2d

5 files changed

Lines changed: 166 additions & 35 deletions

File tree

‎benchmark/webstreams/encoding-streams.js‎

Lines changed: 39 additions & 0 deletions

‎benchmark/webstreams/from.js‎

Lines changed: 29 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,29 @@
1+
'use strict';
2+
const common = require('../common.js');
3+
const {
4+
ReadableStream,
5+
} = require('node:stream/web');
6+
7+
const bench = common.createBenchmark(main, {
8+
n: [1e6],
9+
kind: ['sync', 'async'],
10+
});
11+
12+
async function main({ n, kind }) {
13+
function* syncGen() {
14+
for (let i = 0; i < n; i++) yield i;
15+
}
16+
17+
async function* asyncGen() {
18+
for (let i = 0; i < n; i++) yield i;
19+
}
20+
21+
const reader = ReadableStream.from(
22+
kind === 'sync' ? syncGen() : asyncGen()).getReader();
23+
bench.start();
24+
for (;;) {
25+
const { done } = await reader.read();
26+
if (done) break;
27+
}
28+
bench.end(n);
29+
}

‎lib/internal/webstreams/encoding.js‎

Lines changed: 22 additions & 26 deletions
Original file line numberDiff line numberDiff line change
@@ -4,6 +4,7 @@ const {
44
ObjectDefineProperties,
55
String,
66
StringPrototypeCharCodeAt,
7+
StringPrototypeSlice,
78
Uint8Array,
89
} = primordials;
910

@@ -31,6 +32,9 @@ const {
3132
kEnumerableProperty,
3233
} = require('internal/util');
3334

35+
// Shared per-chunk decode options; decode() only reads the flag.
36+
const kDecodeStreamingOptions = { __proto__: null, stream: true };
37+
3438
/**
3539
* @typedef {import('./readablestream').ReadableStream} ReadableStream
3640
* @typedef {import('./writablestream').WritableStream} WritableStream
@@ -46,34 +50,26 @@ class TextEncoderStream {
4650
this.#transform = new TransformStream({
4751
transform: (chunk, controller) => {
4852
// https://encoding.spec.whatwg.org/#encode-and-enqueue-a-chunk
53+
// The only cross-chunk state is a trailing high surrogate;
54+
// encode() replaces interior lone surrogates with U+FFFD exactly
55+
// like the spec's per-code-unit walk.
4956
chunk = String(chunk);
50-
let finalChunk = '';
51-
for (let i = 0; i < chunk.length; i++) {
52-
const item = chunk[i];
53-
const codeUnit = StringPrototypeCharCodeAt(item, 0);
54-
if (this.#pendingHighSurrogate !== null) {
55-
const highSurrogate = this.#pendingHighSurrogate;
56-
this.#pendingHighSurrogate = null;
57-
if (0xDC00 <= codeUnit && codeUnit <= 0xDFFF) {
58-
finalChunk += highSurrogate + item;
59-
continue;
60-
}
61-
finalChunk += '\uFFFD';
62-
}
63-
if (0xD800 <= codeUnit && codeUnit <= 0xDBFF) {
64-
this.#pendingHighSurrogate = item;
65-
continue;
66-
}
67-
if (0xDC00 <= codeUnit && codeUnit <= 0xDFFF) {
68-
finalChunk += '\uFFFD';
69-
continue;
70-
}
71-
finalChunk += item;
57+
if (chunk.length === 0)
58+
return;
59+
if (this.#pendingHighSurrogate !== null) {
60+
chunk = this.#pendingHighSurrogate + chunk;
61+
this.#pendingHighSurrogate = null;
7262
}
73-
if (finalChunk) {
74-
const value = this.#handle.encode(finalChunk);
75-
controller.enqueue(value);
63+
const lastCodeUnit =
64+
StringPrototypeCharCodeAt(chunk, chunk.length - 1);
65+
if (0xD800 <= lastCodeUnit && lastCodeUnit <= 0xDBFF) {
66+
this.#pendingHighSurrogate =
67+
StringPrototypeSlice(chunk, -1);
68+
chunk = StringPrototypeSlice(chunk, 0, -1);
69+
if (chunk.length === 0)
70+
return;
7671
}
72+
controller.enqueue(this.#handle.encode(chunk));
7773
},
7874
flush: (controller) => {
7975
// https://encoding.spec.whatwg.org/#encode-and-flush
@@ -137,7 +133,7 @@ class TextDecoderStream {
137133
if (chunk === undefined) {
138134
throw new ERR_INVALID_ARG_TYPE('chunk', 'string', chunk);
139135
}
140-
const value = this.#handle.decode(chunk, { stream: true });
136+
const value = this.#handle.decode(chunk, kDecodeStreamingOptions);
141137
if (value)
142138
controller.enqueue(value);
143139
},

‎lib/internal/webstreams/readablestream.js‎

Lines changed: 75 additions & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -112,6 +112,7 @@ const {
112112
getNonWritablePropertyDescriptor,
113113
isBrandCheck,
114114
kEmptyQueue,
115+
kParkedAlgorithmResult,
115116
kResolvedPromise,
116117
kState,
117118
kType,
@@ -1446,19 +1447,85 @@ function readableStreamFromIterable(iterable) {
14461447
if (iterator === null || (typeof iterator !== 'object' && typeof iterator !== 'function')) {
14471448
throw new ERR_INVALID_STATE.TypeError('The iterator method must return an object');
14481449
}
1450+
// Per GetIteratorDirect, the next method is looked up once.
1451+
const nextMethod = iterator.next;
14491452
const startAlgorithm = nonOpCallback;
14501453

1451-
async function pullAlgorithm() {
1452-
const iterResult = await iterator.next();
1454+
// Callback-style pull: the reaction steps are reused across chunks and
1455+
// completion is delivered to the controller's cached pull reactions
1456+
// (the kParkedAlgorithmResult contract). One pull runs at a time, so a
1457+
// single slot carries a non-thenable next() result between steps.
1458+
let pendingIterResult;
1459+
1460+
function rejectPull(error) {
1461+
readableStreamDefaultControllerError(stream[kState].controller, error);
1462+
}
1463+
1464+
function processIterResult(iterResult) {
1465+
const controller = stream[kState].controller;
14531466
if (typeof iterResult !== 'object' || iterResult === null) {
1454-
throw new ERR_INVALID_STATE.TypeError(
1455-
'The promise returned by the iterator.next() method must fulfill with an object');
1467+
rejectPull(new ERR_INVALID_STATE.TypeError(
1468+
'The promise returned by the iterator.next() method must fulfill with an object'));
1469+
return;
14561470
}
1457-
if (iterResult.done) {
1458-
readableStreamDefaultControllerClose(stream[kState].controller);
1459-
} else {
1460-
readableStreamDefaultControllerEnqueue(stream[kState].controller, await iterResult.value);
1471+
try {
1472+
if (iterResult.done) {
1473+
readableStreamDefaultControllerClose(controller);
1474+
} else {
1475+
const value = iterResult.value;
1476+
if (value !== null &&
1477+
(typeof value === 'object' || typeof value === 'function')) {
1478+
// Adopted like `await iterResult.value`, keeping the observable
1479+
// .then lookup on plain objects.
1480+
PromisePrototypeThen(PromiseResolve(value), enqueueValue, rejectPull);
1481+
return;
1482+
}
1483+
readableStreamDefaultControllerEnqueue(controller, value);
1484+
}
1485+
} catch (error) {
1486+
rejectPull(error);
1487+
return;
1488+
}
1489+
// pullFulfilled exists: the controller creates it before the pull.
1490+
controller[kState].pullFulfilled();
1491+
}
1492+
1493+
function enqueueValue(value) {
1494+
const controller = stream[kState].controller;
1495+
try {
1496+
readableStreamDefaultControllerEnqueue(controller, value);
1497+
} catch (error) {
1498+
rejectPull(error);
1499+
return;
1500+
}
1501+
controller[kState].pullFulfilled();
1502+
}
1503+
1504+
function processPendingIterResult() {
1505+
const iterResult = pendingIterResult;
1506+
pendingIterResult = undefined;
1507+
processIterResult(iterResult);
1508+
}
1509+
1510+
function pullAlgorithm() {
1511+
let nextResult;
1512+
try {
1513+
nextResult = FunctionPrototypeCall(nextMethod, iterator);
1514+
} catch (error) {
1515+
return PromiseReject(error);
1516+
}
1517+
if (nextResult !== null &&
1518+
(typeof nextResult === 'object' || typeof nextResult === 'function')) {
1519+
// Mirrors `await iterator.next()`: processIterResult runs at the
1520+
// microtask position the await resumed.
1521+
PromisePrototypeThen(
1522+
PromiseResolve(nextResult), processIterResult, rejectPull);
1523+
return kParkedAlgorithmResult;
14611524
}
1525+
// A non-thenable next() result fails validation a microtask later.
1526+
pendingIterResult = nextResult;
1527+
PromisePrototypeThen(kResolvedPromise, processPendingIterResult);
1528+
return kParkedAlgorithmResult;
14621529
}
14631530

14641531
async function cancelAlgorithm(reason) {

‎lib/internal/webstreams/util.js‎

Lines changed: 1 addition & 1 deletion

0 commit comments

Comments
 (0)