stream: speed up async iteration of Readable by mcollina · Pull Request #64447 · nodejs/node

mcollina · 2026-07-12T08:27:10Z

Replace the async generator backing Readable.prototype[Symbol.asyncIterator] (and .iterator()) with a hand-rolled iterator.

The generator machinery costs several extra promise allocations and microtask hops per chunk: yield awaits the yielded value and resolves the pending request through separate promises, and every next() goes through the async generator request queue. The hand-rolled iterator delivers buffered chunks as an already-resolved promise.

The observable semantics are preserved:

thenable chunks (object mode) are still awaited before delivery, and a rejected thenable still tears down the iterator and the stream;
next()/return()/throw() calls received while a request is outstanding are queued and processed in order — including return() while waiting for data, which still completes only once the pending read settles;
return()/throw() before the first next() complete the iterator without attaching listeners or destroying the stream;
the finally teardown logic (destroyOnReturn, autoDestroy, half-open duplex preservation) is unchanged;
error aggregation via aggregateTwoErrors is unchanged.

The one observable difference is that buffered chunks are delivered one microtask sooner than before, since the generator's yield performed an implicit Await on the yielded value. This is visible to code racing an abort against the first chunk. The flatMap AbortSignal test relied on such a race (a queueMicrotask'd abort beating the first chunk); it is reworked to be deterministic and timer-free: two mappers block until their signal aborts, the abort fires while both are in flight, and the test asserts the concurrency limit is respected (exactly two mappers start), in-flight mappers are cancelled through their signal, and iteration rejects with AbortError. The reworked test passes against both the old and the new implementation.

New regression tests cover the subtler iterator behaviors (thenable unwrapping, rejected thenables, throw(), pre-start throw(), concurrent next() ordering, and return() queued behind a pending next()); they also pass against both implementations.

Benchmark (benchmark/compare.js, 30 runs):

                                                       confidence improvement accuracy (*)   (**)  (***)
streams/readable-async-iterator.js sync='no' n=100000         ***      9.84 %       ±3.04% ±4.05% ±5.27%
streams/readable-async-iterator.js sync='yes' n=100000        ***     32.59 %       ±5.49% ±7.34% ±9.62%

No changes on pipe.js / readable-readall.js.

🤖 Generated with Claude Code

nodejs-github-bot · 2026-07-12T08:27:15Z

Review requested:

@nodejs/streams

Replace the async generator backing Symbol.asyncIterator with a hand-rolled iterator. The generator machinery costs several extra promise allocations and microtask hops per chunk: yield awaits the yielded value and resolves the pending request through separate promises. Buffered chunks are now delivered as an already-resolved promise, one microtask sooner than before. Thenable chunks are still awaited before delivery, requests received while a next() is outstanding are queued, and return()/throw() before the first next() complete the iterator without touching the stream. The earlier delivery is observable by code racing an abort against the first chunk. The flatMap AbortSignal test relied on such a race; it is reworked to abort deterministically while two mappers are in flight, asserting the concurrency limit, in-flight cancellation and rejection, without depending on delivery timing or timers. streams/readable-async-iterator.js sync='yes': +32.59% (***) streams/readable-async-iterator.js sync='no': +9.84% (***) Assisted-by: Claude Fable 5 Signed-off-by: Matteo Collina <matteo.collina@gmail.com>

aduh95 · 2026-07-12T09:18:49Z

+      if (typeof chunk.then === 'function') {
+        PromisePrototypeThen(PromiseResolve(chunk), (value) => {
+          inFlight = false;
+          resolve({ done: false, value });
+          if (queue !== null) drain();
+        }, (err) => {
+          inFlight = false;
+          settleError(err, reject);
+          if (queue !== null) drain();
+        });
+        return;
+      }


If then is a getter that e.g. throws on the second access, this code might throw when previously it wouldn't have. We can protect against that by storing the initial value we're getting (that might also avoid an additional promise allocation).

Suggested change

if (typeof chunk.then === 'function') {

PromisePrototypeThen(PromiseResolve(chunk), (value) => {

inFlight = false;

resolve({ done: false, value });

if (queue !== null) drain();

}, (err) => {

inFlight = false;

settleError(err, reject);

if (queue !== null) drain();

});

return;

}

const { then } = chunk;

if (typeof then === 'function') {

FunctionPrototypeCall(then, chunk, (value) => {

inFlight = false;

resolve({ done: false, value });

if (queue !== null) drain();

}, (err) => {

inFlight = false;

settleError(err, reject);

if (queue !== null) drain();

});

return;

}

aduh95 · 2026-07-12T09:21:28Z

+          if (typeof chunk.then === 'function') {
+            inFlight = true;
+            return PromisePrototypeThen(
+              PromiseResolve(chunk), onChunkFulfilled, onChunkRejected);


Same here

Suggested change

if (typeof chunk.then === 'function') {

inFlight = true;

return PromisePrototypeThen(

PromiseResolve(chunk), onChunkFulfilled, onChunkRejected);

const { then } = chunk;

if (typeof then === 'function') {

inFlight = true;

return FunctionPrototypeCall(then, chunk, onChunkFulfilled, onChunkRejected);

Renegade334 · 2026-07-12T10:00:58Z

+    settleError(err, reject);
+  }
+
+  return {


This object should ideally have a prototype of AsyncIteratorPrototype.

codecov · 2026-07-12T10:11:17Z

Codecov Report

❌ Patch coverage is 89.34010% with 21 lines in your changes missing coverage. Please review.
✅ Project coverage is 90.23%. Comparing base (8a3b11c) to head (6f9a2f3).
⚠️ Report is 4 commits behind head on main.

Files with missing lines	Patch %	Lines
lib/internal/streams/readable.js	89.34%	20 Missing and 1 partial ⚠️

Additional details and impacted files

@@            Coverage Diff             @@
##             main   #64447      +/-   ##
==========================================
- Coverage   90.24%   90.23%   -0.02%     
==========================================
  Files         741      741              
  Lines      241384   241544     +160     
  Branches    45480    45527      +47     
==========================================
+ Hits       217844   217954     +110     
- Misses      15097    15149      +52     
+ Partials     8443     8441       -2

Files with missing lines	Coverage Δ
lib/internal/streams/readable.js	`96.51% <89.34%> (-0.76%)`	⬇️

... and 41 files with indirect coverage changes

🚀 New features to boost your workflow:

❄️ Test Analytics: Detect flaky tests, report on failures, and find test suite problems.
📦 JS Bundle Analysis: Save yourself from yourself by tracking and limiting bundle sizes in JS merges.

mertcanaltin · 2026-07-12T10:49:59Z

+      return new Promise((resolve, reject) => {
+        if (inFlight) {
+          queue ??= [];
+          queue.push({ type: 'return', value, resolve, reject });


Suggested change

queue.push({ type: 'return', value, resolve, reject });

queue.push({ __proto__: null, type: 'return', value, resolve, reject });

mertcanaltin · 2026-07-12T10:50:11Z

+      return new Promise((resolve, reject) => {
+        if (inFlight) {
+          queue ??= [];
+          queue.push({ type: 'next', value: undefined, resolve, reject });


Suggested change

queue.push({ type: 'next', value: undefined, resolve, reject });

queue.push({ __proto__: null, type: 'next', value: undefined, resolve, reject });

mertcanaltin · 2026-07-12T10:50:27Z

+      return new Promise((resolve, reject) => {
+        if (inFlight) {
+          queue ??= [];
+          queue.push({ type: 'throw', value: err, resolve, reject });


Suggested change

queue.push({ type: 'throw', value: err, resolve, reject });

queue.push({ __proto__: null, type: 'throw', value: err, resolve, reject });

mertcanaltin · 2026-07-12T11:10:27Z

+
+  function drain() {
+    while (!inFlight && queue.length > 0) {
+      const req = queue.shift();


Can we replace queue.shift(); with an index-based queue for high throughput?

gurgunday · 2026-07-12T11:42:16Z

+      }
+      return new Promise((resolve, reject) => {
+        if (inFlight) {
+          queue ??= [];


We should use FixedQueue here since a normal Array shift inside a while loop to empty it is O(n2):

node/lib/internal/fixed_queue.js

Line 91 in 17163ea

module.exports = class FixedQueue {

gurgunday · 2026-07-12T11:50:53Z

+    while (!inFlight && queue.length > 0) {
+      const req = queue.shift();
+      if (req.type === 'next') {
+        processNext(req.resolve, req.reject);
+      } else if (req.type === 'return') {
+        processReturn(req.value, req.resolve);
+      } else {
+        processThrow(req.value, req.reject);
+      }
+    }


We need a guard here I think because drain() is recursive when chunks are already buffered

Each queued request adds another nested drain -> processNext -> pump cycle

Suggested change

while (!inFlight && queue.length > 0) {

const req = queue.shift();

if (req.type === 'next') {

processNext(req.resolve, req.reject);

} else if (req.type === 'return') {

processReturn(req.value, req.resolve);

} else {

processThrow(req.value, req.reject);

}

}

if (draining) {

return;

}

draining = true;

try {

while (!inFlight && queue.length > 0) {

const req = queue.shift();

if (req.type === 'next') {

processNext(req.resolve, req.reject);

} else if (req.type === 'return') {

processReturn(req.value, req.resolve);

} else {

processThrow(req.value, req.reject);

}

}

} finally {

draining = false;

}

Currently, this script throws a stack error with this patch:

'use strict'; const assert = require('node:assert'); const { Readable } = require('node:stream'); const count = 20_000; const stream = new Readable({ objectMode: true, read() {}, }); const iterator = stream[Symbol.asyncIterator](); const requests = []; for (let i = 0; i < count; i++) { requests.push(iterator.next()); } for (let i = 0; i < count; i++) { stream.push(i); } stream.push(null); Promise.all(requests).then((results) => { assert.strictEqual(results.length, count); for (let i = 0; i < count; i++) { assert.deepStrictEqual(results[i], { done: false, value: i, }); } console.log('All requests completed without recursion'); }); // RangeError: Maximum call stack size exceeded

nodejs-github-bot added needs-ci PRs that need a full CI run. stream Issues and PRs related to the stream subsystem. labels Jul 12, 2026

mcollina force-pushed the stream-async-iterator-perf branch from 2307152 to 6f9a2f3 Compare July 12, 2026 08:36

aduh95 reviewed Jul 12, 2026

View reviewed changes

Renegade334 reviewed Jul 12, 2026

View reviewed changes

mertcanaltin reviewed Jul 12, 2026

View reviewed changes

gurgunday reviewed Jul 12, 2026

View reviewed changes

	queue.push({ type: 'return', value, resolve, reject });
	queue.push({ __proto__: null, type: 'return', value, resolve, reject });

	queue.push({ type: 'next', value: undefined, resolve, reject });
	queue.push({ __proto__: null, type: 'next', value: undefined, resolve, reject });

	queue.push({ type: 'throw', value: err, resolve, reject });
	queue.push({ __proto__: null, type: 'throw', value: err, resolve, reject });

Uh oh!

Uh oh!

Conversation

mcollina commented Jul 12, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

nodejs-github-bot commented Jul 12, 2026

Uh oh!

aduh95 Jul 12, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

aduh95 Jul 12, 2026

Choose a reason for hiding this comment

Uh oh!

Renegade334 Jul 12, 2026

Choose a reason for hiding this comment

Uh oh!

codecov Bot commented Jul 12, 2026

Codecov Report

Uh oh!

mertcanaltin Jul 12, 2026

Choose a reason for hiding this comment

Uh oh!

mertcanaltin Jul 12, 2026

Choose a reason for hiding this comment

Uh oh!

mertcanaltin Jul 12, 2026

Choose a reason for hiding this comment

Uh oh!

mertcanaltin Jul 12, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

gurgunday Jul 12, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

gurgunday Jul 12, 2026

Choose a reason for hiding this comment

Uh oh!

gurgunday Jul 12, 2026

Choose a reason for hiding this comment

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

6 participants

mcollina commented Jul 12, 2026 •

edited

Loading

aduh95 Jul 12, 2026 •

edited

Loading

mertcanaltin Jul 12, 2026 •

edited

Loading

gurgunday Jul 12, 2026 •

edited

Loading