| FazBrowse GitHub Viewer | Trending | | Home |
| Tools: [Download Repo ZIP] [Original HTTPS Page] |
Sorry, something went wrong.
|
Review requested:
|
Sorry, something went wrong.
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>
| 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; | ||
| } |
There was a problem hiding this comment.
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).
| 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; | |
| } |
Sorry, something went wrong.
| if (typeof chunk.then === 'function') { | ||
| inFlight = true; | ||
| return PromisePrototypeThen( | ||
| PromiseResolve(chunk), onChunkFulfilled, onChunkRejected); |
There was a problem hiding this comment.
Same here
| 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); |
Sorry, something went wrong.
| settleError(err, reject); | ||
| } | ||
|
|
||
| return { |
There was a problem hiding this comment.
This object should ideally have a prototype of AsyncIteratorPrototype.
Sorry, something went wrong.
There was a problem hiding this comment.
I'm not sure it should if the sprecs doesn't require it – it does add a @@asyncDispose method which may or may not be desirable – it also adds a @@asyncIterator, which begs the question whether we should re-implement the method (re-implementing it ourselves means we're not subject to prototype tampering; otherwise, letting the built-in method be inherited would saves a bit of memory maybe?)
Sorry, something went wrong.
There was a problem hiding this comment.
The current implementation derives from AsyncIteratorPrototype (as a generator) so if we're trying to minimise observability then this is a fairly free move. (While the language currently only provides for the two well-known symbol methods, once we get into async iterator helpers territory, there will be significant advantages to keeping this inheritance.)
Sorry, something went wrong.
Codecov Report❌ Patch coverage is 90.23256% with 21 lines in your changes missing coverage. Please review.
@@ Coverage Diff @@
## main #64447 +/- ##
==========================================
+ Coverage 90.24% 90.29% +0.04%
==========================================
Files 741 760 +19
Lines 241384 247324 +5940
Branches 45480 46652 +1172
==========================================
+ Hits 217844 223310 +5466
- Misses 15097 15477 +380
- Partials 8443 8537 +94
... and 209 files with indirect coverage changes 🚀 New features to boost your workflow:
|
Sorry, something went wrong.
| return new Promise((resolve, reject) => { | ||
| if (inFlight) { | ||
| queue ??= []; | ||
| queue.push({ type: 'return', value, resolve, reject }); |
There was a problem hiding this comment.
| queue.push({ type: 'return', value, resolve, reject }); | |
| queue.push({ __proto__: null, type: 'return', value, resolve, reject }); |
Sorry, something went wrong.
| return new Promise((resolve, reject) => { | ||
| if (inFlight) { | ||
| queue ??= []; | ||
| queue.push({ type: 'next', value: undefined, resolve, reject }); |
There was a problem hiding this comment.
| queue.push({ type: 'next', value: undefined, resolve, reject }); | |
| queue.push({ __proto__: null, type: 'next', value: undefined, resolve, reject }); |
Sorry, something went wrong.
| return new Promise((resolve, reject) => { | ||
| if (inFlight) { | ||
| queue ??= []; | ||
| queue.push({ type: 'throw', value: err, resolve, reject }); |
There was a problem hiding this comment.
| queue.push({ type: 'throw', value: err, resolve, reject }); | |
| queue.push({ __proto__: null, type: 'throw', value: err, resolve, reject }); |
Sorry, something went wrong.
|
|
||
| function drain() { | ||
| while (!inFlight && queue.length > 0) { | ||
| const req = queue.shift(); |
There was a problem hiding this comment.
Can we replace queue.shift(); with an index-based queue for high throughput?
Sorry, something went wrong.
| await new Promise((resolve, reject) => { | ||
| if (signal.aborted) { | ||
| reject(signal.reason); | ||
| return; | ||
| } | ||
| signal.addEventListener('abort', () => reject(signal.reason), { once: true }); |
There was a problem hiding this comment.
| 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; |
Sorry, something went wrong.
Signed-off-by: Matteo Collina <hello@matteocollina.com>
There was a problem hiding this comment.
lgtm
Sorry, something went wrong.
Sorry, something went wrong.
Signed-off-by: Matteo Collina <hello@matteocollina.com>
There was a problem hiding this comment.
lgtm
Sorry, something went wrong.
Signed-off-by: Matteo Collina <hello@matteocollina.com>
Sorry, something went wrong.
Commit Queue failed- Loading data for nodejs/node/pull/64447 ✔ Done loading data for nodejs/node/pull/64447 ----------------------------------- PR info ------------------------------------ Title stream: speed up async iteration of Readable (#64447) Author Matteo Collina <matteo.collina@gmail.com> (@mcollina) Branch mcollina:stream-async-iterator-perf -> nodejs:main Labels stream, author ready, needs-ci, commit-queue Commits 4 - stream: speed up async iteration of Readable - fixup: address review comments - fixup: remove [SymbolAsyncIterator] - stream: fix lint in readable async iterator Committers 1 - Matteo Collina <hello@matteocollina.com> PR-URL: https://github.com/nodejs/node/pull/64447 Reviewed-By: Gürgün Dayıoğlu <hey@gurgun.day> Reviewed-By: Mattias Buelens <mattias@buelens.com> Reviewed-By: Robert Nagy <ronagy@icloud.com> ------------------------------ Generated metadata ------------------------------ PR-URL: https://github.com/nodejs/node/pull/64447 Reviewed-By: Gürgün Dayıoğlu <hey@gurgun.day> Reviewed-By: Mattias Buelens <mattias@buelens.com> Reviewed-By: Robert Nagy <ronagy@icloud.com> -------------------------------------------------------------------------------- ℹ This PR was created on Sun, 12 Jul 2026 08:27:10 GMT ✔ Approvals: 3 ✔ - Gürgün Dayıoğlu (@gurgunday): https://github.com/nodejs/node/pull/64447#pullrequestreview-4707049635 ✔ - Mattias Buelens (@MattiasBuelens): https://github.com/nodejs/node/pull/64447#pullrequestreview-4706259826 ✔ - Robert Nagy (@ronag) (TSC): https://github.com/nodejs/node/pull/64447#pullrequestreview-4838491319 ✔ Last GitHub CI successful ℹ Last Full PR CI on 2026-08-06T13:07:45Z: https://ci.nodejs.org/job/node-test-pull-request/75563/ - Querying data for job/node-test-pull-request/75563/ ✔ Build data downloaded ✔ Last Jenkins CI successful -------------------------------------------------------------------------------- ✔ No git cherry-pick in progress ✔ No git am in progress ✔ No git rebase in progress -------------------------------------------------------------------------------- - Bringing origin/main up to date... From https://github.com/nodejs/node * branch main -> FETCH_HEAD ✔ origin/main is now up-to-date - Downloading patch for 64447 From https://github.com/nodejs/node * branch refs/pull/64447/merge -> FETCH_HEAD ✔ Fetched commits as 9e23066b8af4..873a47ef6ae8 -------------------------------------------------------------------------------- Auto-merging lib/internal/streams/readable.js [main 0a0be7f763] stream: speed up async iteration of Readable Author: Matteo Collina <hello@matteocollina.com> Date: Sat Jul 11 23:41:55 2026 +0200 3 files changed, 307 insertions(+), 35 deletions(-) Auto-merging lib/internal/streams/readable.js [main 22401b1cb5] fixup: address review comments Author: Matteo Collina <hello@matteocollina.com> Date: Mon Jul 13 10:00:44 2026 +0200 3 files changed, 73 insertions(+), 26 deletions(-) Auto-merging lib/internal/streams/readable.js [main 0b186829dd] fixup: remove [SymbolAsyncIterator] Author: Matteo Collina <hello@matteocollina.com> Date: Wed Jul 15 17:34:22 2026 +0200 1 file changed, 1 insertion(+), 4 deletions(-) Auto-merging lib/internal/streams/readable.js [main 656bc697b3] stream: fix lint in readable async iterator Author: Matteo Collina <hello@matteocollina.com> Date: Sun Aug 2 09:08:31 2026 +0000 1 file changed, 1 insertion(+), 1 deletion(-) ✔ Patches applied There are 4 commits in the PR. Attempting autorebase. (node:388) [DEP0190] DeprecationWarning: Passing args to a child process with shell option true can lead to security vulnerabilities, as the arguments are not escaped, only concatenated. (Use `node --trace-deprecation ...` to show where the warning was created) Rebasing (2/8) Executing: git node land --amend --yes --------------------------------- New Message ---------------------------------- stream: speed up async iteration of Readablehttps://github.com/nodejs/node/actions/runs/31846615170 |
Sorry, something went wrong.
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> PR-URL: #64447 Reviewed-By: Gürgün Dayıoğlu <hey@gurgun.day> Reviewed-By: Mattias Buelens <mattias@buelens.com> Reviewed-By: Robert Nagy <ronagy@icloud.com>
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> PR-URL: #64447 Reviewed-By: Gürgün Dayıoğlu <hey@gurgun.day> Reviewed-By: Mattias Buelens <mattias@buelens.com> Reviewed-By: Robert Nagy <ronagy@icloud.com>
| Back | FazBrowse Home | New Git URL |
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:
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):
No changes on pipe.js / readable-readall.js.
🤖 Generated with Claude Code