stream: limit iter from sync iterable batches · nodejs/node@c50d8e4 · GitHub
Skip to content

Commit c50d8e4

Browse files
trivikraduh95
trivikr
authored and
aduh95
committed
stream: limit iter from sync iterable batches
Bound sync iterable normalization in from() and fromSync() to FROM_BATCH_SIZE. This avoids unbounded batches for from() sync iterable fallbacks and lets fromSync() coalesce plain Uint8Array values for writev paths. Signed-off-by: Kamat, Trivikram <16024985+trivikr@users.noreply.github.com> Assisted-by: openai:gpt-5.5 PR-URL: #63324 Backport-PR-URL: #64675 Reviewed-By: James M Snell <jasnell@gmail.com> Reviewed-By: Ethan Arrowood <ethan@arrowood.dev>
1 parent 1d7c86d commit c50d8e4

5 files changed

Lines changed: 216 additions & 20 deletions

File tree

‎benchmark/streams/iter-from-batching.js‎

Lines changed: 100 additions & 0 deletions

‎lib/internal/streams/iter/from.js‎

Lines changed: 57 additions & 18 deletions
Original file line numberDiff line numberDiff line change
@@ -47,7 +47,7 @@ const {
4747
toUint8Array,
4848
} = require('internal/streams/iter/utils');
4949

50-
// Maximum number of chunks to yield per batch from from(Uint8Array[]).
50+
// Maximum number of chunks to yield per batch from from()/fromSync().
5151
// Bounds peak memory when arrays flow through transforms, which must
5252
// allocate output for the entire batch at once.
5353
const FROM_BATCH_SIZE = 128;
@@ -190,33 +190,66 @@ function isUint8ArrayBatch(value) {
190190
return true;
191191
}
192192

193+
function* yieldBoundedBatch(batch) {
194+
if (batch.length === 0) {
195+
return;
196+
}
197+
if (batch.length <= FROM_BATCH_SIZE) {
198+
yield batch;
199+
return;
200+
}
201+
for (let i = 0; i < batch.length; i += FROM_BATCH_SIZE) {
202+
yield ArrayPrototypeSlice(batch, i, i + FROM_BATCH_SIZE);
203+
}
204+
}
205+
193206
/**
194207
* Normalize a sync streamable source, yielding batches of Uint8Array.
195208
* @param {Iterable} source
196209
* @yields {Uint8Array[]}
197210
*/
198211
function* normalizeSyncSource(source) {
212+
let batch = [];
213+
199214
for (const value of source) {
200215
// Fast path 1: value is already a Uint8Array[] batch
201216
if (isUint8ArrayBatch(value)) {
202-
if (value.length > 0) {
203-
yield value;
217+
if (batch.length > 0) {
218+
yield batch;
219+
batch = [];
204220
}
221+
yield* yieldBoundedBatch(value);
205222
continue;
206223
}
207224
// Fast path 2: value is a single Uint8Array (very common)
208225
if (isUint8Array(value)) {
209-
yield [value];
226+
ArrayPrototypePush(batch, value);
227+
if (batch.length === FROM_BATCH_SIZE) {
228+
yield batch;
229+
batch = [];
230+
}
210231
continue;
211232
}
212233
// Slow path: normalize the value
213-
const batch = [];
214-
for (const chunk of normalizeSyncValue(value)) {
215-
ArrayPrototypePush(batch, chunk);
216-
}
217234
if (batch.length > 0) {
218235
yield batch;
236+
batch = [];
237+
}
238+
let valueBatch = [];
239+
for (const chunk of normalizeSyncValue(value)) {
240+
ArrayPrototypePush(valueBatch, chunk);
241+
if (valueBatch.length === FROM_BATCH_SIZE) {
242+
yield valueBatch;
243+
valueBatch = [];
244+
}
219245
}
246+
if (valueBatch.length > 0) {
247+
yield valueBatch;
248+
}
249+
}
250+
251+
if (batch.length > 0) {
252+
yield batch;
220253
}
221254
}
222255

@@ -329,36 +362,42 @@ async function* normalizeAsyncSource(source) {
329362
return;
330363
}
331364

332-
// Fall back to sync iteration - batch all sync values together
365+
// Fall back to sync iteration - batch sync values together with a bound.
333366
if (isSyncIterable(source)) {
334-
const batch = [];
367+
let batch = [];
335368

336369
for (const value of source) {
337370
// Fast path 1: value is already a Uint8Array[] batch
338371
if (isUint8ArrayBatch(value)) {
339372
// Flush any accumulated batch first
340373
if (batch.length > 0) {
341-
yield ArrayPrototypeSlice(batch);
342-
batch.length = 0;
343-
}
344-
if (value.length > 0) {
345-
yield value;
374+
yield batch;
375+
batch = [];
346376
}
377+
yield* yieldBoundedBatch(value);
347378
continue;
348379
}
349380
// Fast path 2: value is a single Uint8Array (very common)
350381
if (isUint8Array(value)) {
351382
ArrayPrototypePush(batch, value);
383+
if (batch.length === FROM_BATCH_SIZE) {
384+
yield batch;
385+
batch = [];
386+
}
352387
continue;
353388
}
354389
// Slow path: normalize the value - must flush and yield individually
355390
if (batch.length > 0) {
356-
yield ArrayPrototypeSlice(batch);
357-
batch.length = 0;
391+
yield batch;
392+
batch = [];
358393
}
359-
const asyncBatch = [];
394+
let asyncBatch = [];
360395
for await (const chunk of normalizeAsyncValue(value)) {
361396
ArrayPrototypePush(asyncBatch, chunk);
397+
if (asyncBatch.length === FROM_BATCH_SIZE) {
398+
yield asyncBatch;
399+
asyncBatch = [];
400+
}
362401
}
363402
if (asyncBatch.length > 0) {
364403
yield asyncBatch;

‎test/parallel/test-stream-iter-from-coverage.js‎

Lines changed: 34 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -31,6 +31,22 @@ async function testFromSyncSubBatching() {
3131
assert.strictEqual(totalChunks, 200);
3232
}
3333

34+
// fromSync: generic sync iterables of Uint8Array use bounded batches
35+
async function testFromSyncIterableSubBatching() {
36+
function* gen() {
37+
for (let i = 0; i < 200; i++) {
38+
yield new Uint8Array([i & 0xFF]);
39+
}
40+
}
41+
const batches = [];
42+
for (const batch of fromSync(gen())) {
43+
batches.push(batch);
44+
}
45+
assert.strictEqual(batches.length, 2);
46+
assert.strictEqual(batches[0].length, 128);
47+
assert.strictEqual(batches[1].length, 72);
48+
}
49+
3450
// from: Uint8Array[] with > 128 elements triggers sub-batching (async)
3551
async function testFromAsyncSubBatching() {
3652
const bigBatch = Array.from({ length: 200 },
@@ -44,6 +60,22 @@ async function testFromAsyncSubBatching() {
4460
assert.strictEqual(batches[1].length, 72);
4561
}
4662

63+
// from: sync iterables use bounded batches instead of one unbounded batch
64+
async function testFromAsyncSyncIterableSubBatching() {
65+
function* gen() {
66+
for (let i = 0; i < 200; i++) {
67+
yield new Uint8Array([i & 0xFF]);
68+
}
69+
}
70+
const batches = [];
71+
for await (const batch of from(gen())) {
72+
batches.push(batch);
73+
}
74+
assert.strictEqual(batches.length, 2);
75+
assert.strictEqual(batches[0].length, 128);
76+
assert.strictEqual(batches[1].length, 72);
77+
}
78+
4779
// Exact boundary: 128 elements → single batch (no split)
4880
async function testFromSubBatchingBoundary() {
4981
const exactBatch = Array.from({ length: 128 },
@@ -133,7 +165,9 @@ async function testFromSyncInvalidYield() {
133165

134166
Promise.all([
135167
testFromSyncSubBatching(),
168+
testFromSyncIterableSubBatching(),
136169
testFromAsyncSubBatching(),
170+
testFromAsyncSyncIterableSubBatching(),
137171
testFromSubBatchingBoundary(),
138172
testFromSubBatchingBoundaryPlus1(),
139173
testFromSyncDataViewInGenerator(),

‎test/parallel/test-stream-iter-from-sync.js‎

Lines changed: 3 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -66,9 +66,10 @@ function testFromSyncGenerator() {
6666
for (const batch of readable) {
6767
batches.push(batch);
6868
}
69-
assert.strictEqual(batches.length, 2);
69+
assert.strictEqual(batches.length, 1);
70+
assert.strictEqual(batches[0].length, 2);
7071
assert.deepStrictEqual(batches[0][0], new Uint8Array([1, 2]));
71-
assert.deepStrictEqual(batches[1][0], new Uint8Array([3, 4]));
72+
assert.deepStrictEqual(batches[0][1], new Uint8Array([3, 4]));
7273
}
7374

7475
function testFromSyncNestedIterables() {

‎test/parallel/test-stream-iter-pipeto-writev.js‎

Lines changed: 22 additions & 0 deletions

0 commit comments

Comments
 (0)