Skip to content

stream: speed up async iteration of Readable #64447

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
mcollina wants to merge 1 commit into
base: main
Choose a base branch
from
+307 −35
Open

stream: speed up async iteration of Readable #64447

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
225 changes: 197 additions & 28 deletions lib/internal/streams/readable.js
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -30,6 +30,9 @@ const {
ObjectKeys,
ObjectSetPrototypeOf,
Promise,
PromisePrototypeThen,
PromiseReject,
PromiseResolve,
ReflectApply,
SafeSet,
Symbol,
Expand Down Expand Up @@ -1382,10 +1385,21 @@ function streamToAsyncIterator(stream, options) {
return iter;
}

async function* createAsyncIterator(stream, options) {
// Async iterator over a Readable. Requests received while another is
// outstanding are queued and processed in order.
function createAsyncIterator(stream, options) {
let callback = nop;

function next(resolve) {
let error; // undefined: active, null: ended cleanly, else: Error
let started = false;
let completed = false;
let inFlight = false; // An asynchronous request is outstanding
let queue = null; // Requests received while inFlight
let cleanup;

// Used both as the 'readable' listener (where `this === stream`) and
// as a promise executor storing the resolver that wakes up a pending
// pump().
function wakeup(resolve) {
if (this === stream) {
callback();
callback = nop;
Expand All @@ -1394,32 +1408,23 @@ async function* createAsyncIterator(stream, options) {
}
}

stream.on('readable', next);
function start() {
started = true;

let error;
const cleanup = eos(stream, { writable: false }, (err) => {
error = err ? aggregateTwoErrors(error, err) : null;
callback();
callback = nop;
});
stream.on('readable', wakeup);

cleanup = eos(stream, { writable: false }, (err) => {
error = err ? aggregateTwoErrors(error, err) : null;
callback();
callback = nop;
});
}

// Complete the iterator and either destroy the stream or detach
// from it.
function finalize() {
completed = true;

try {
while (true) {
const chunk = stream.destroyed ? null : stream.read();
if (chunk !== null) {
yield chunk;
} else if (error) {
throw error;
} else if (error === null) {
return;
} else {
await new Promise(next);
}
}
} catch (err) {
error = aggregateTwoErrors(error, err);
throw error;
} finally {
const preserveHalfOpenDuplex =
error === null &&
stream.allowHalfOpen === true &&
Expand All @@ -1433,10 +1438,174 @@ async function* createAsyncIterator(stream, options) {
) {
destroyImpl.destroyer(stream, null);
} else {
stream.off('readable', next);
stream.off('readable', wakeup);
cleanup();
}
}

function settleError(err, reject) {
error = aggregateTwoErrors(error, err);
finalize();
reject(error);
}

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

@mertcanaltin mertcanaltin Jul 12, 2026 •
edited
Loading

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

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

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);
}
}
Comment on lines +1453 to +1462

@gurgunday gurgunday Jul 12, 2026

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

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

@gurgunday gurgunday Jul 12, 2026

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

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

}

// Thenable chunks are unwrapped before delivery; a rejection tears
// down the iterator and the stream.
function onChunkFulfilled(value) {
inFlight = false;
if (queue !== null) drain();
return { done: false, value };
}

function onChunkRejected(err) {
inFlight = false;
error = aggregateTwoErrors(error, err);
finalize();
if (queue !== null) drain();
throw error;
}

// Runs with inFlight === true; settles the request and hands over to
// any requests that queued up behind it.
function pump(resolve, reject) {
const chunk = stream.destroyed ? null : stream.read();
if (chunk !== null) {
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;
}
Comment on lines +1486 to +1497

@aduh95 aduh95 Jul 12, 2026 •
edited
Loading

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

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

inFlight = false;
resolve({ done: false, value: chunk });
if (queue !== null) drain();
} else if (error) {
inFlight = false;
settleError(error, reject);
if (queue !== null) drain();
} else if (error === null) {
inFlight = false;
finalize();
resolve({ done: true, value: undefined });
if (queue !== null) drain();
} else {
// No data buffered yet; wait for 'readable' or end-of-stream and
// retry.
PromisePrototypeThen(new Promise(wakeup), () => pump(resolve, reject));
}
}

function processNext(resolve, reject) {
if (completed) {
resolve({ done: true, value: undefined });
return;
}
if (!started) start();
inFlight = true;
pump(resolve, reject);
}

function processReturn(value, resolve) {
if (!completed) {
if (started) {
finalize();
} else {
// Never started: complete without touching the stream.
completed = true;
}
}
resolve({ done: true, value });
}

function processThrow(err, reject) {
if (completed || !started) {
completed = true;
reject(err);
return;
}
settleError(err, reject);
}

return {

@Renegade334 Renegade334 Jul 12, 2026

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

This object should ideally have a prototype of AsyncIteratorPrototype.

next() {
if (!inFlight && !completed) {
if (!started) start();
// Fast path: a chunk is already buffered.
const chunk = stream.destroyed ? null : stream.read();
if (chunk !== null) {
if (typeof chunk.then === 'function') {
inFlight = true;
return PromisePrototypeThen(
PromiseResolve(chunk), onChunkFulfilled, onChunkRejected);
Comment on lines +1555 to +1558

@aduh95 aduh95 Jul 12, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

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

}
return PromiseResolve({ done: false, value: chunk });
}
if (error) {
finalize();
return PromiseReject(error);
}
if (error === null) {
finalize();
return PromiseResolve({ done: true, value: undefined });
}
// No data buffered yet; wait for 'readable' or end-of-stream.
inFlight = true;
return new Promise((resolve, reject) => {
PromisePrototypeThen(new Promise(wakeup), () => pump(resolve, reject));
});
}
return new Promise((resolve, reject) => {
if (inFlight) {
queue ??= [];

@gurgunday gurgunday Jul 12, 2026 •
edited
Loading

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

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 {

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

@mertcanaltin mertcanaltin Jul 12, 2026

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

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

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

@mertcanaltin mertcanaltin Jul 12, 2026

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

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

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

@mertcanaltin mertcanaltin Jul 12, 2026

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

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

} else {
processThrow(err, reject);
}
});
},
[SymbolAsyncIterator]() {
return this;
},
};
}

let composeImpl;
Expand Down
23 changes: 16 additions & 7 deletions test/parallel/test-stream-flatMap.js
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -72,10 +72,23 @@ function oneTo5() {

{
// Concurrency + AbortSignal
// Two mappers are started concurrently and block until their signal
// is aborted. Aborting while both are in flight must cancel them and
// reject the iteration, without ever starting a third mapper.
const ac = new AbortController();
const stream = oneTo5().flatMap(common.mustNotCall(async (_, { signal }) => {
await setTimeout(100, { signal });
}), { signal: ac.signal, concurrency: 2 });
const stream = oneTo5().flatMap(common.mustCall(async (x, { signal }) => {
if (x === 2) {
// Both mappers allowed by `concurrency` are now in flight.
ac.abort();
}
await new Promise((resolve, reject) => {
if (signal.aborted) {
reject(signal.reason);
return;
}
signal.addEventListener('abort', () => reject(signal.reason), { once: true });
Comment on lines +84 to +89

@jasnell jasnell Jul 12, 2026

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Suggested change
await new Promise((resolve, reject) => {
if (signal.aborted) {
reject(signal.reason);
return;
}
signal.addEventListener('abort', () => reject(signal.reason), { once: true });
const { promise, reject } = Promise.withResolvers();
if (signal.aborted) {
reject(signal.reason);
}
signal.addEventListener('abort', () => reject(signal.reason), { once: true });
// Promise is expected to reject.
await promise;

});
}, 2), { signal: ac.signal, concurrency: 2 });
// pump
assert.rejects(async () => {
for await (const item of stream) {
Expand All @@ -85,10 +98,6 @@ function oneTo5() {
}, {
name: 'AbortError',
}).then(common.mustCall());

queueMicrotask(() => {
ac.abort();
});
}

{
Expand Down
94 changes: 94 additions & 0 deletions test/parallel/test-stream-readable-async-iterators.js
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -869,5 +869,99 @@ async function tests() {
}));
}

{
// Thenable chunks are awaited before delivery.
(async () => {
const r = new Readable({ objectMode: true, read() {} });
r.push(Promise.resolve('unwrapped'));
r.push(null);

const it = r[Symbol.asyncIterator]();
const { value, done } = await it.next();
assert.strictEqual(done, false);
assert.strictEqual(value, 'unwrapped');
})().then(common.mustCall());
}

{
// A rejected thenable chunk tears down the iterator and the stream.
(async () => {
const r = new Readable({ objectMode: true, read() {} });
const rejected = Promise.reject(new Error('kaboom'));
rejected.catch(() => {});
r.push(rejected);
r.push(null);

const it = r[Symbol.asyncIterator]();
await assert.rejects(it.next(), { message: 'kaboom' });
assert.strictEqual((await it.next()).done, true);
assert.strictEqual(r.destroyed, true);
})().then(common.mustCall());
}

{
// throw() rejects with the passed error, destroys the stream and
// completes the iterator.
(async () => {
const r = new Readable({ objectMode: true, read() {} });
r.push('a');

const it = r[Symbol.asyncIterator]();
assert.strictEqual((await it.next()).value, 'a');
await assert.rejects(it.throw(new Error('kaboom')), { message: 'kaboom' });
assert.strictEqual(r.destroyed, true);
assert.strictEqual((await it.next()).done, true);
})().then(common.mustCall());
}

{
// throw() before the first next() completes the iterator without
// touching the stream.
(async () => {
const r = new Readable({ objectMode: true, read() {} });
const it = r[Symbol.asyncIterator]();
await assert.rejects(it.throw(new Error('kaboom')), { message: 'kaboom' });
assert.strictEqual(r.destroyed, false);
assert.strictEqual(r.listenerCount('readable'), 0);
assert.strictEqual((await it.next()).done, true);
r.destroy();
})().then(common.mustCall());
}

{
// Concurrent next() calls while waiting for data are served in order.
(async () => {
const r = new Readable({ objectMode: true, read() {} });
const it = r[Symbol.asyncIterator]();

const p1 = it.next();
const p2 = it.next();
r.push('a');
r.push('b');
r.push(null);

assert.strictEqual((await p1).value, 'a');
assert.strictEqual((await p2).value, 'b');
assert.strictEqual((await it.next()).done, true);
})().then(common.mustCall());
}

{
// return() while a next() is pending is processed after the pending
// next() settles, and destroys the stream.
(async () => {
const r = new Readable({ objectMode: true, read() {} });
const it = r[Symbol.asyncIterator]();

const p1 = it.next();
const p2 = it.return();
r.push('a');

assert.strictEqual((await p1).value, 'a');
assert.deepStrictEqual(await p2, { done: true, value: undefined });
assert.strictEqual(r.destroyed, true);
})().then(common.mustCall());
}

// To avoid missing some tests if a promise does not resolve
tests().then(common.mustCall());
Loading