Skip to content

Commit 153ec89

Browse files
jasnellnodejs-github-bot
authored andcommitted
stream: make stream/iter from() cancellation waits cheaper
To stay cancellable while a source is pending, from() waits for every value of an async source through waitForNormalization(). For each value it created a PromiseWithResolvers(), raced it against the value with SafePromiseRace(), which wraps both in new promises, and ran an async function with try/finally. This was the largest per-batch cost of normalizing an async source. Wait with a single promise and a single reaction on the value instead, and reject that promise directly on cancellation. The outcome is unchanged: the value's result, its rejection, or the cancellation reason, whichever comes first, and the cancellation reason if the normalization was cancelled by the time the value fulfills. With an async generator yielding 16-byte chunks, pipeTo() is about 1.7x faster and iterating from() about 1.75x faster. Assisted-by: OpenCode Signed-off-by: James M Snell <jasnell@gmail.com> PR-URL: #66648 Reviewed-By: Trivikram Kamat <trivikr.dev@gmail.com> Reviewed-By: Gürgün Dayıoğlu <hey@gurgun.day>
1 parent 818f348 commit 153ec89

1 file changed

Lines changed: 41 additions & 17 deletions

File tree

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

Lines changed: 41 additions & 17 deletions
Original file line numberDiff line numberDiff line change
@@ -19,7 +19,6 @@ const {
1919
PromisePrototypeThen,
2020
PromiseResolve,
2121
PromiseWithResolvers,
22-
SafePromiseRace,
2322
Symbol,
2423
SymbolAsyncIterator,
2524
SymbolIterator,
@@ -54,6 +53,7 @@ const {
5453

5554
const {
5655
IterResult,
56+
kResolvedPromise,
5757
getProtocolMethod,
5858
toUint8Array,
5959
} = require('internal/streams/iter/utils');
@@ -62,7 +62,6 @@ const {
6262
// Bounds peak memory when arrays flow through transforms, which must
6363
// allocate output for the entire batch at once.
6464
const FROM_BATCH_SIZE = 128;
65-
const kNormalizationCancelled = Symbol('kNormalizationCancelled');
6665
// Yielded by normalizeAsyncValue() (only when `emitFlush` is true) right
6766
// before it waits on a promise or on a nested async iterable. Callers that
6867
// batch chunks yield whatever they have collected so far, so that chunks that
@@ -83,31 +82,56 @@ function cancelNormalization(context, reason, suppressCleanup = false) {
8382
context.cancelled = true;
8483
context.reason = reason;
8584
context.suppressCleanup = suppressCleanup;
86-
context.resolve?.(kNormalizationCancelled);
85+
context.resolve?.();
8786
}
8887

8988
function throwIfNormalizationCancelled(context) {
9089
if (context?.cancelled) throw context.reason;
9190
}
9291

93-
async function waitForNormalization(value, context) {
92+
/**
93+
* Wait for `value`, but stop waiting if the normalization is cancelled.
94+
* Settles with the first of:
95+
* - `value` fulfilling: its value, or the cancellation reason if the
96+
* normalization has been cancelled by then;
97+
* - `value` rejecting: its rejection reason;
98+
* - cancellation: the cancellation reason.
99+
* This runs for every value of a normalized async source, so it uses a
100+
* single promise and a single reaction on `value` rather than racing
101+
* promises.
102+
* @param {any} value
103+
* @param {object} [context]
104+
* @returns {Promise<any>|any}
105+
*/
106+
function waitForNormalization(value, context) {
94107
if (context === undefined) return value;
95-
const { promise, resolve } = PromiseWithResolvers();
108+
const { promise, resolve, reject } = PromiseWithResolvers();
109+
const onCancel = () => {
110+
if (context.resolve === onCancel) context.resolve = null;
111+
reject(context.reason);
112+
};
113+
PromisePrototypeThen(
114+
PromiseResolve(value),
115+
(result) => {
116+
if (context.resolve === onCancel) context.resolve = null;
117+
if (context.cancelled) {
118+
reject(context.reason);
119+
} else {
120+
resolve(result);
121+
}
122+
},
123+
(error) => {
124+
if (context.resolve === onCancel) context.resolve = null;
125+
reject(error);
126+
});
96127
if (context.cancelled) {
97-
resolve(kNormalizationCancelled);
128+
// Already cancelled: a `value` that has already settled still takes
129+
// precedence, as its reaction above runs first.
130+
PromisePrototypeThen(kResolvedPromise, onCancel);
98131
} else {
99-
context.resolve = resolve;
100-
}
101-
try {
102-
const result = await SafePromiseRace([
103-
PromiseResolve(value),
104-
promise,
105-
]);
106-
throwIfNormalizationCancelled(context);
107-
return result;
108-
} finally {
109-
if (context.resolve === resolve) context.resolve = null;
132+
context.resolve = onCancel;
110133
}
134+
return promise;
111135
}
112136

113137
function createNormalizationIterator(createIterator) {

0 commit comments

Comments
 (0)