From df5549767d0d688ee5210cecd122c1fb4c6539c8 Mon Sep 17 00:00:00 2001 From: Luciano Leggieri <230980@gmail.com> Date: Sat, 11 Apr 2026 14:07:07 -0700 Subject: [PATCH] stream: optimize iteration functions on Readable Signed-off-by: Luciano Leggieri <230980@gmail.com> Assisted-by: Sol 5.6 --- benchmark/streams/operator-throughput.js | 138 +++ lib/internal/streams/from.js | 10 +- lib/internal/streams/operators.js | 808 ++++++++++++++---- test/parallel/test-readable-from.js | 24 +- test/parallel/test-stream-filter.js | 48 ++ test/parallel/test-stream-forEach.js | 24 + test/parallel/test-stream-map.js | 158 +++- test/parallel/test-stream-reduce.js | 84 +- test/parallel/test-stream-some-find-every.mjs | 88 ++ test/parallel/test-stream-toArray.js | 69 ++ 10 files changed, 1271 insertions(+), 180 deletions(-) create mode 100644 benchmark/streams/operator-throughput.js diff --git a/benchmark/streams/operator-throughput.js b/benchmark/streams/operator-throughput.js new file mode 100644 index 000000000000..9256c160e49d --- /dev/null +++ b/benchmark/streams/operator-throughput.js @@ -0,0 +1,138 @@ +'use strict'; + +const assert = require('assert'); +const common = require('../common'); +const { Readable } = require('stream'); + +const bench = common.createBenchmark(main, { + n: [2e5], + operation: [ + 'reduce', + 'reduce-async', + 'map-sync-1', + 'map-sync-8', + 'map-async-1', + 'map-async-8', + 'find-sync-1', + 'find-sync-8', + 'find-async-1', + 'find-async-8', + 'filter', + 'forEach', + 'toArray', + ], + source: ['readable', 'iterator', 'array'], +}); + +function createSource(type, n) { + if (type === 'array') { + return Readable.from(Array.from({ length: n }, (_, i) => i)); + } + let i = 0; + + if (type === 'readable') { + return new Readable({ + objectMode: true, + read() { + this.push(i === n ? null : i++); + }, + }); + } + + return Readable.from((function* generate() { + while (i < n) { + yield i++; + } + })()); +} + +async function main({ n, operation, source }) { + const readable = createSource(source, n); + let result; + + bench.start(); + + switch (operation) { + case 'reduce': + result = await readable.reduce((sum, value) => sum + value, 0); + break; + case 'reduce-async': + result = await readable.reduce(async (sum, value) => sum + value, 0); + break; + case 'map-sync-1': + result = await readable + .map((value) => value + 1, { concurrency: 1 }) + .reduce((sum, value) => sum + value, 0); + break; + case 'map-sync-8': + result = await readable + .map((value) => value + 1, { concurrency: 8 }) + .reduce((sum, value) => sum + value, 0); + break; + case 'map-async-1': + result = await readable + .map(async (value) => value + 1, { concurrency: 1 }) + .reduce((sum, value) => sum + value, 0); + break; + case 'map-async-8': + result = await readable + .map(async (value) => value + 1, { concurrency: 8 }) + .reduce((sum, value) => sum + value, 0); + break; + case 'find-sync-1': + result = await readable.find((value) => value === n - 1, { concurrency: 1 }); + break; + case 'find-sync-8': + result = await readable.find((value) => value === n - 1, { concurrency: 8 }); + break; + case 'find-async-1': + result = await readable.find(async (value) => value === n - 1, { concurrency: 1 }); + break; + case 'find-async-8': + result = await readable.find(async (value) => value === n - 1, { concurrency: 8 }); + break; + case 'filter': + result = await readable + .filter((value) => (value & 1) === 0) + .reduce((sum, value) => sum + value, 0); + break; + case 'forEach': + result = 0; + await readable.forEach((value) => { result += value; }); + break; + case 'toArray': + result = await readable.toArray(); + break; + default: + throw new Error(`Unknown operation: ${operation}`); + } + + bench.end(n); + + const sum = n * (n - 1) / 2; + switch (operation) { + case 'map-sync-1': + case 'map-sync-8': + case 'map-async-1': + case 'map-async-8': + assert.strictEqual(result, sum + n); + break; + case 'find-sync-1': + case 'find-sync-8': + case 'find-async-1': + case 'find-async-8': + assert.strictEqual(result, n - 1); + break; + case 'filter': { + const evenCount = Math.ceil(n / 2); + assert.strictEqual(result, evenCount * (evenCount - 1)); + break; + } + case 'toArray': + assert.strictEqual(result.length, n); + assert.strictEqual(result[n - 1], n - 1); + break; + default: + assert.strictEqual(result, sum); + } +} diff --git a/lib/internal/streams/from.js b/lib/internal/streams/from.js index 6a943f026579..52331d0c976d 100644 --- a/lib/internal/streams/from.js +++ b/lib/internal/streams/from.js @@ -1,6 +1,7 @@ 'use strict'; const { + ArrayIsArray, PromisePrototypeThen, SymbolAsyncIterator, SymbolIterator, @@ -15,6 +16,10 @@ const { }, } = require('internal/errors'); +const { + getDefaultHighWaterMark, +} = require('internal/streams/state'); + function from(Readable, iterable, opts) { let iterator; if (typeof iterable === 'string' || iterable instanceof Buffer) { @@ -26,6 +31,8 @@ function from(Readable, iterable, opts) { this.push(null); }, }); + } else if (iterable instanceof Readable) { + return iterable; } let isAsync; @@ -39,10 +46,9 @@ function from(Readable, iterable, opts) { throw new ERR_INVALID_ARG_TYPE('iterable', ['Iterable'], iterable); } - const readable = new Readable({ objectMode: true, - highWaterMark: 1, + highWaterMark: ArrayIsArray(iterable) ? getDefaultHighWaterMark(true) : 1, // TODO(ronag): What options should be allowed? ...opts, }); diff --git a/lib/internal/streams/operators.js b/lib/internal/streams/operators.js index 6db2df0e3646..3f047ae99555 100644 --- a/lib/internal/streams/operators.js +++ b/lib/internal/streams/operators.js @@ -3,6 +3,7 @@ const { ArrayPrototypePush, Boolean, + FunctionPrototypeCall, MathFloor, Number, NumberIsNaN, @@ -10,6 +11,7 @@ const { PromisePrototypeThen, PromiseReject, PromiseResolve, + PromiseWithResolvers, Symbol, } = primordials; @@ -17,9 +19,11 @@ const { AbortController, AbortSignal } = require('internal/abort_controller'); const { AbortError, + aggregateTwoErrors, codes: { ERR_MISSING_ARGS, ERR_OUT_OF_RANGE, + ERR_STREAM_NULL_VALUES, }, } = require('internal/errors'); const { @@ -27,13 +31,63 @@ const { validateInteger, validateObject, validateFunction, + validateBoolean, } = require('internal/validators'); const { kWeakHandler, kResistStopPropagation } = require('internal/event_target'); -const { finished } = require('internal/streams/end-of-stream'); +const { eos, finished } = require('internal/streams/end-of-stream'); +const destroyImpl = require('internal/streams/destroy'); const kEmpty = Symbol('kEmpty'); const kEof = Symbol('kEof'); +const { + isPromise, +} = require('internal/util/types'); + +const nop = () => {}; +function createWaiter() { + let resolve = nop; + + function reset(r) { + resolve = r; + } + + function wait() { + return new Promise(reset); + } + + function notify() { + resolve(); + resolve = nop; + } + + return { __proto__: null, wait, reset, notify }; +} + +// Return native promises unchanged and normalize custom thenables exactly once. +// Capturing `then` avoids repeated getter access, while calling it with `value` +// as the receiver preserves thenables that depend on their `this` value. +function getThenablePromise(value) { + if (isPromise(value)) { + return value; + } + + const valueType = typeof value; + if ((valueType === 'object' && value !== null) || valueType === 'function') { + const then = value.then; + if (typeof then === 'function') { + return PromiseResolve({ + __proto__: null, + then(resolve, reject) { + FunctionPrototypeCall(then, value, resolve, reject); + }, + }); + } + } + + return undefined; +} + function map(fn, options) { validateFunction(fn, 'fn'); if (options != null) { @@ -43,184 +97,440 @@ function map(fn, options) { validateAbortSignal(options.signal, 'options.signal'); } - let concurrency = 1; - if (options?.concurrency != null) { - concurrency = MathFloor(options.concurrency); - } - - let highWaterMark = concurrency - 1; - if (options?.highWaterMark != null) { - highWaterMark = MathFloor(options.highWaterMark); - } + const concurrency = MathFloor(options?.concurrency ?? 1); + const highWaterMark = MathFloor(options?.highWaterMark ?? concurrency - 1); validateInteger(concurrency, 'options.concurrency', 1); validateInteger(highWaterMark, 'options.highWaterMark', 0); - highWaterMark += concurrency; - - return async function* map() { - const signal = AbortSignal.any([options?.signal].filter(Boolean)); - const stream = this; - const queue = []; - const signalOpt = { signal }; - - let next; - let resume; - let done = false; - let cnt = 0; + return createMappedStream(this, fn, options, concurrency, highWaterMark + concurrency); +} - function onCatch() { - done = true; - afterItemProcessed(); - } +function createMappedStream(source, fn, options, concurrency, highWaterMark) { + const ac = new AbortController(); + const signal = AbortSignal.any([ac.signal, options?.signal].filter(Boolean)); + const signalOpt = { signal }; + const queue = []; + const sourceReadable = createWaiter(); + const queueReadable = createWaiter(); + const sourceCapacity = createWaiter(); + + let stopped = false; + let activeMappers = 0; + + function atCapacity() { + return activeMappers >= concurrency || queue.length >= highWaterMark; + } - function afterItemProcessed() { - cnt -= 1; - maybeResume(); + function maybeResumeSource() { + if (!stopped && !atCapacity()) { + sourceCapacity.notify(); } + } - function maybeResume() { - if ( - resume && - !done && - cnt < concurrency && - queue.length < highWaterMark - ) { - resume(); - resume = null; - } - } + function mapperFinished() { + activeMappers--; + maybeResumeSource(); + } - async function pump() { - try { - for await (let val of stream) { - if (done) { - return; - } + function mapperRejected() { + stopped = true; + mapperFinished(); + queueReadable.notify(); + } - if (signal.aborted) { - throw new AbortError(); - } + function enqueueMappedChunk(chunk) { + activeMappers++; + const mapped = fn(chunk, signalOpt); - try { - val = fn(val, signalOpt); + if (mapped === kEmpty) { + mapperFinished(); + return mapped; + } - if (val === kEmpty) { - continue; - } + const mappedPromise = getThenablePromise(mapped); + if (mappedPromise !== undefined) { + PromisePrototypeThen(mappedPromise, mapperFinished, mapperRejected); + queue.push(mappedPromise); + } else { + queue.push({ chunk: mapped, afterItemProcessed: mapperFinished }); + } + queueReadable.notify(); + return mapped; + } - val = PromiseResolve(val); - } catch (err) { - val = PromiseReject(err); - } + async function pumpSource() { + let error; + source.on('readable', sourceReadable.notify); + signal.addEventListener('abort', sourceReadable.notify, { once: true }); + const cleanup = eos(source, { writable: false }, (err) => { + error = err ? aggregateTwoErrors(error, err) : null; + sourceReadable.notify(); + }); - cnt += 1; + try { + while (!stopped) { + if (signal.aborted) { + throw new AbortError(undefined, { cause: signal.reason }); + } - PromisePrototypeThen(val, afterItemProcessed, onCatch); + if (atCapacity()) { + await sourceCapacity.wait(); + continue; + } - queue.push(val); - if (next) { - next(); - next = null; + const chunk = source.destroyed ? null : source.read(); + if (chunk !== null) { + const mapped = enqueueMappedChunk(chunk); + if (mapped === kEof) { + while (queue.length > 0) { + await sourceCapacity.wait(); + } + return; } + continue; + } - if (!done && (queue.length >= highWaterMark || cnt >= concurrency)) { - await new Promise((resolve) => { - resume = resolve; - }); - } + if (error) { + throw error; } - queue.push(kEof); - } catch (err) { - const val = PromiseReject(err); - PromisePrototypeThen(val, afterItemProcessed, onCatch); - queue.push(val); - } finally { - done = true; - if (next) { - next(); - next = null; + if (error === null) { + queue.push(kEof); + return; } + await sourceReadable.wait(); } + } catch (err) { + error = aggregateTwoErrors(error, err); + throw error; + } finally { + if ( + (error || options?.destroyOnReturn !== false) && + (error === undefined || source._readableState.autoDestroy) + ) { + destroyImpl.destroyer(source, null); + } else { + source.off('readable', sourceReadable.notify); + cleanup(); + } + signal.removeEventListener('abort', sourceReadable.notify); + } + } + + async function runSourcePump() { + try { + await pumpSource(); + } catch (err) { + const rejected = PromiseReject(err); + PromisePrototypeThen(rejected, mapperFinished, mapperRejected); + queue.push(rejected); + } finally { + stopped = true; + queueReadable.notify(); } + } - pump(); + // This is lazy to avoid a circular dependency during stream initialization. + const Readable = require('internal/streams/readable'); + const readable = new Readable({ + objectMode: true, + highWaterMark, + }); + let readingQueue = false; + + readable._read = async function _read() { + if (readingQueue) { + return; + } + readingQueue = true; try { - while (true) { - while (queue.length > 0) { - const val = await queue[0]; + if (!stopped && queue.length === 0) { + await queueReadable.wait(); + } - if (val === kEof) { - return; - } + while (queue.length > 0) { + if (signal.aborted) { + throw new AbortError(); + } - if (signal.aborted) { - throw new AbortError(); - } + let item = queue[0]; + let syncItemFinished; + if (isPromise(item)) { + item = await item; + queue.shift(); + } else if (typeof item.afterItemProcessed === 'function') { + syncItemFinished = item.afterItemProcessed; + item = item.chunk; + // Keep the item queued until its mapper is marked as finished so + // async and synchronous items apply the same backpressure. + syncItemFinished(); + queue.shift(); + } else { + queue.shift(); + } - if (val !== kEmpty) { - yield val; + if (item === kEof) { + stopped = true; + this.push(null); + sourceCapacity.notify(); + queueReadable.notify(); + return; + } + + if (item === kEmpty) { + maybeResumeSource(); + if (!stopped && queue.length === 0) { + await queueReadable.wait(); } + continue; + } - queue.shift(); - maybeResume(); + if (item === null) { + throw new ERR_STREAM_NULL_VALUES(); } - await new Promise((resolve) => { - next = resolve; - }); + if (syncItemFinished) { + // Prevent stream.read() from re-entering _read before push clears its + // internal reading flag. + await PromiseResolve(); + } + const needsMore = this.push(item); + // Let async-iterator cleanup run before pulling another source chunk. + process.nextTick(maybeResumeSource); + if (!needsMore) { + return; + } } + } catch (err) { + destroyImpl.destroyer(this, err); } finally { - done = true; - if (resume) { - resume(); - resume = null; - } + readingQueue = false; } - }.call(this); + }; + + readable._destroy = function(err, cb) { + stopped = true; + ac.abort(err); + sourceReadable.notify(); + sourceCapacity.notify(); + queueReadable.notify(); + cb(err); + }; + + process.nextTick(runSourcePump); + return readable; } -async function some(fn, options = undefined) { - for await (const unused of filter.call(this, fn, options)) { - return true; +function nowOrLater(fn, fn2, args) { + const value = fn(...args); + const promise = getThenablePromise(value); + if (promise !== undefined) { + return PromisePrototypeThen(promise, fn2); } - return false; + return fn2(value); +} + +async function some(fn, options = undefined) { + validateFunction(fn, 'fn'); + const someFn = (...args) => { + return nowOrLater(fn, Boolean, args); + }; + return (await find.call(this, someFn, options)) !== undefined; } async function every(fn, options = undefined) { validateFunction(fn, 'fn'); + const everyFn = (...args) => { + return nowOrLater(fn, (value) => !value, args); + }; // https://en.wikipedia.org/wiki/De_Morgan%27s_laws - return !(await some.call(this, async (...args) => { - return !(await fn(...args)); - }, options)); + return !(await find.call(this, everyFn, options)); } -async function find(fn, options) { - for await (const result of filter.call(this, fn, options)) { - return result; +function find(fn, options) { + validateFunction(fn, 'fn'); + + if (options != null) { + validateObject(options, 'options'); } - return undefined; + const signal = options?.signal; + if (signal != null) { + validateAbortSignal(signal, 'options.signal'); + } + + const concurrency = MathFloor(options?.concurrency ?? 1); + validateInteger(concurrency, 'options.concurrency', 1); + + const destroyOnReturn = options?.destroyOnReturn ?? true; + validateBoolean(destroyOnReturn, 'options.destroyOnReturn'); + + const ac = new AbortController(); + const predicateSignal = AbortSignal.any([ac.signal, signal].filter(Boolean)); + const predicateOptions = { signal: predicateSignal }; + + // Concurrent predicates can settle out of order. Stop reading after any + // match, but keep the lowest index after all active predicates settle. + const stream = this; + const { promise, resolve } = PromiseWithResolvers(); + let match; + let error; + let activeEvaluations = 0; + let nextIndex = 0; + let ended = false; + let settled = false; + let draining = false; + + function settle() { + if (!settled) { + settled = true; + resolve(); + } + } + + function fail(err) { + if (settled) { + return; + } + error = aggregateTwoErrors(error, err); + destroyImpl.destroyer(stream, error); + settle(); + } + + function maybeSettle() { + if (activeEvaluations === 0 && (match !== undefined || ended)) { + settle(); + } + } + + function evaluationFinished(matches, chunk, index) { + if (matches && (match === undefined || index < match.index)) { + match = { index, value: chunk }; + } + activeEvaluations--; + maybeSettle(); + + if (!settled && match === undefined && !draining) { + onReadable(); + } + } + + function evaluationRejected(err) { + activeEvaluations--; + fail(err); + } + + function evaluate(chunk, index) { + activeEvaluations++; + + let matches; + try { + matches = fn(chunk, predicateOptions); + const matchesPromise = getThenablePromise(matches); + if (matchesPromise !== undefined) { + PromisePrototypeThen( + matchesPromise, + (result) => evaluationFinished(result, chunk, index), + evaluationRejected, + ); + return; + } + } catch (err) { + evaluationRejected(err); + return; + } + evaluationFinished(matches, chunk, index); + } + + function onReadable() { + if (draining || settled || match !== undefined) { + return; + } + + draining = true; + try { + while ( + !settled && + match === undefined && + activeEvaluations < concurrency + ) { + if (signal?.aborted) { + fail(new AbortError(undefined, { cause: signal.reason })); + return; + } + + const chunk = stream.destroyed ? null : stream.read(); + if (chunk === null) { + return; + } + evaluate(chunk, nextIndex++); + } + } catch (err) { + fail(err); + } finally { + draining = false; + } + } + + function onAbort() { + fail(new AbortError(undefined, { cause: signal.reason })); + } + + stream.on('readable', onReadable); + + const cleanup = eos(stream, { writable: false }, (err) => { + if (settled) { + return; + } + if (err) { + fail(err); + return; + } + ended = true; + maybeSettle(); + }); + + if (signal != null) { + signal.addEventListener('abort', onAbort, { once: true }); + if (signal.aborted) { + onAbort(); + } + } + + return PromisePrototypeThen(promise, () => { + stream.off('readable', onReadable); + signal?.removeEventListener('abort', onAbort); + ac.abort(); + + if ( + (error || destroyOnReturn !== false) && + (error === undefined || stream._readableState.autoDestroy) + ) { + destroyImpl.destroyer(stream, error); + } else { + cleanup(); + } + + if (error) { + return PromiseReject(error); + } + return match?.value; + }); } async function forEach(fn, options) { validateFunction(fn, 'fn'); - async function forEachFn(value, options) { - await fn(value, options); - return kEmpty; - } - // eslint-disable-next-line no-unused-vars - for await (const unused of map.call(this, forEachFn, options)); + const forEachFn = (...args) => { + return nowOrLater(fn, () => false, args); + }; + await find.call(this, forEachFn, options); } function filter(fn, options) { validateFunction(fn, 'fn'); - async function filterFn(value, options) { - if (await fn(value, options)) { - return value; - } - return kEmpty; - } + const filterFn = (value, options) => { + return nowOrLater(fn, (predicate) => (predicate ? value : kEmpty), [value, options]); + }; return map.call(this, filterFn, options); } @@ -233,65 +543,221 @@ class ReduceAwareErrMissingArgs extends ERR_MISSING_ARGS { } } -async function reduce(reducer, initialValue, options) { +function reduce(reducer, initialValue, options) { + try { + return reduceImpl(this, arguments.length > 1, reducer, initialValue, options); + } catch (err) { + return PromiseReject(err); + } +} + +function reduceImpl(stream, hasInitialValue, reducer, initialValue, options) { validateFunction(reducer, 'reducer'); if (options != null) { validateObject(options, 'options'); } - if (options?.signal != null) { - validateAbortSignal(options.signal, 'options.signal'); + const signal = options?.signal; + if (signal != null) { + validateAbortSignal(signal, 'options.signal'); } - let hasInitialValue = arguments.length > 1; - if (options?.signal?.aborted) { - const err = new AbortError(undefined, { cause: options.signal.reason }); - this.once('error', () => {}); // The error is already propagated - await finished(this.destroy(err)); - throw err; - } + const { promise, resolve } = PromiseWithResolvers(); const ac = new AbortController(); - const signal = ac.signal; - if (options?.signal) { - const opts = { once: true, [kWeakHandler]: this, [kResistStopPropagation]: true }; - options.signal.addEventListener('abort', () => ac.abort(), opts); + const reducerSignal = ac.signal; + const reducerOptions = { signal: reducerSignal }; + let reducing = false; + let draining = false; + let ended = false; + let settled = false; + let error; + + function settle() { + if (!settled) { + settled = true; + resolve(); + } } - let gotAnyItemFromStream = false; - try { - for await (const value of this) { - gotAnyItemFromStream = true; - if (options?.signal?.aborted) { - throw new AbortError(); - } - if (!hasInitialValue) { - initialValue = value; - hasInitialValue = true; - } else { - initialValue = await reducer(initialValue, value, { signal }); - } + + function fail(err) { + if (settled) { + return; + } + error = aggregateTwoErrors(error, err); + destroyImpl.destroyer(stream, error); + settle(); + } + + function finish() { + if (reducing || settled) { + return; } - if (!gotAnyItemFromStream && !hasInitialValue) { - throw new ReduceAwareErrMissingArgs(); + if (!hasInitialValue) { + fail(new ReduceAwareErrMissingArgs()); + return; } - } finally { + settle(); + } + + function reducerFulfilled(value) { + reducing = false; + initialValue = value; + if (settled) { + return; + } + if (ended) { + finish(); + } else { + onReadable(); + } + } + + function reducerRejected(err) { + reducing = false; + fail(err); + } + + function onReadable() { + if (draining || reducing || settled) { + return; + } + + draining = true; + try { + while (!reducing && !settled) { + if (signal?.aborted) { + fail(new AbortError(undefined, { cause: signal.reason })); + break; + } + + const value = stream.destroyed ? null : stream.read(); + if (value === null) { + break; + } + + if (!hasInitialValue) { + initialValue = value; + hasInitialValue = true; + continue; + } + + let result; + try { + result = reducer(initialValue, value, reducerOptions); + const resultPromise = getThenablePromise(result); + if (resultPromise === undefined) { + initialValue = result; + continue; + } + + reducing = true; + PromisePrototypeThen(resultPromise, reducerFulfilled, reducerRejected); + } catch (err) { + fail(err); + break; + } + } + } catch (err) { + fail(err); + } finally { + draining = false; + } + } + + function onAbort() { ac.abort(); + fail(new AbortError(undefined, { cause: signal.reason })); + } + + stream.on('readable', onReadable); + + const cleanup = eos(stream, { writable: false }, (err) => { + if (settled) { + return; + } + if (err) { + fail(err); + return; + } + ended = true; + finish(); + }); + + if (signal != null) { + const opts = { + once: true, + [kWeakHandler]: stream, + [kResistStopPropagation]: true, + }; + signal.addEventListener('abort', onAbort, opts); + if (signal.aborted) { + onAbort(); + } } - return initialValue; + + return PromisePrototypeThen(promise, () => { + stream.off('readable', onReadable); + signal?.removeEventListener('abort', onAbort); + ac.abort(); + + if (error) { + destroyImpl.destroyer(stream, error); + return PromiseReject(error); + } + + cleanup(); + return initialValue; + }); } async function toArray(options) { if (options != null) { validateObject(options, 'options'); } - if (options?.signal != null) { - validateAbortSignal(options.signal, 'options.signal'); + const signal = options?.signal; + if (signal != null) { + validateAbortSignal(signal, 'options.signal'); } + const destroyOnReturn = options?.destroyOnReturn ?? true; + validateBoolean(destroyOnReturn, 'options.destroyOnReturn'); + const result = []; - for await (const val of this) { - if (options?.signal?.aborted) { - throw new AbortError(undefined, { cause: options.signal.reason }); + const stream = this; + + function onReadable() { + while (true) { + const chunk = this.destroyed ? null : this.read(); + if (chunk !== null) { + ArrayPrototypePush(result, chunk); + continue; + } else if (signal?.aborted) { + this.destroy(new AbortError(undefined, { cause: signal.reason })); + } + return; + } + } + + function onAbort() { + stream.destroy(new AbortError(undefined, { cause: signal.reason })); + } + + this.on('readable', onReadable); + const done = finished(this, { writable: false, cleanup: true }); + if (signal != null) { + signal.addEventListener('abort', onAbort, { once: true }); + } + + try { + if (signal?.aborted) { + onAbort(); } - ArrayPrototypePush(result, val); + await done; + } finally { + this.off('readable', onReadable); + signal?.removeEventListener('abort', onAbort); + } + if (destroyOnReturn !== false) { + destroyImpl.destroyer(this, null); } return result; } @@ -327,19 +793,13 @@ function drop(number, options = undefined) { } number = toIntegerOrInfinity(number); - return async function* drop() { - if (options?.signal?.aborted) { - throw new AbortError(); - } - for await (const val of this) { - if (options?.signal?.aborted) { - throw new AbortError(); - } - if (number-- <= 0) { - yield val; - } + return filter.call(this, () => { + if (number > 0) { + number--; + return false; } - }.call(this); + return true; + }, options); } function take(number, options = undefined) { diff --git a/test/parallel/test-readable-from.js b/test/parallel/test-readable-from.js index 1d812ade3f23..e5e7e1d6974a 100644 --- a/test/parallel/test-readable-from.js +++ b/test/parallel/test-readable-from.js @@ -2,7 +2,7 @@ const { mustCall } = require('../common'); const { once } = require('events'); -const { Readable } = require('stream'); +const { getDefaultHighWaterMark, Readable } = require('stream'); const assert = require('assert'); const common = require('../common'); @@ -12,6 +12,28 @@ const common = require('../common'); }, /ERR_INVALID_ARG_TYPE/); } +{ + // Arrays use the default object-mode highWaterMark, while other iterables + // retain the one-at-a-time behavior. + const arrayStream = Readable.from([1, 2]); + assert.strictEqual( + arrayStream.readableHighWaterMark, + getDefaultHighWaterMark(true), + ); + arrayStream.destroy(); + + const setStream = Readable.from(new Set([1, 2])); + assert.strictEqual(setStream.readableHighWaterMark, 1); + setStream.destroy(); +} + +{ + // Readable inputs are already in the requested form. + const source = Readable.from([1]); + assert.strictEqual(Readable.from(source), source); + source.destroy(); +} + async function toReadableBasicSupport() { async function* generate() { yield 'a'; diff --git a/test/parallel/test-stream-filter.js b/test/parallel/test-stream-filter.js index 0b70c391c88f..340872a08299 100644 --- a/test/parallel/test-stream-filter.js +++ b/test/parallel/test-stream-filter.js @@ -8,6 +8,37 @@ const assert = require('assert'); const { once } = require('events'); const { setTimeout } = require('timers/promises'); +{ + // Filter works on empty streams with a synchronous predicate + const stream = Readable.from([]).filter((x) => true); + (async () => { + for await (const item of stream) { + assert.fail(`${item} should not exist`); + } + })().then(common.mustCall()); +} + +{ + // Filter works on synchronous streams with a synchronous predicate + const stream = Readable.from([1, 2, 3, 4, 5]).filter((x) => x < 10); + const result = [1, 2, 3, 4, 5]; + (async () => { + for await (const item of stream) { + assert.strictEqual(item, result.shift()); + } + })().then(common.mustCall()); +} + +{ + // Filter works on synchronous streams with a synchronous predicate + const stream = Readable.from([1, 2, 3, 4, 5]).filter((x) => x > 10); + (async () => { + for await (const item of stream) { + assert.fail(`${item} should not exist`); + } + })().then(common.mustCall()); +} + { // Filter works on synchronous streams with a synchronous predicate const stream = Readable.from([1, 2, 3, 4, 5]).filter((x) => x < 3); @@ -33,6 +64,23 @@ const { setTimeout } = require('timers/promises'); })().then(common.mustCall()); } +{ + // Filter awaits thenable predicates and preserves their receiver. + (async () => { + const thenable = { + then(resolve) { + thenable.receiver = this; + resolve(false); + }, + }; + assert.deepStrictEqual( + await Readable.from([1]).filter(() => thenable).toArray(), + [], + ); + assert.strictEqual(thenable.receiver, thenable); + })().then(common.mustCall()); +} + { // Map works on asynchronous streams with a asynchronous mapper const stream = Readable.from([1, 2, 3, 4, 5]).map(async (x) => { diff --git a/test/parallel/test-stream-forEach.js b/test/parallel/test-stream-forEach.js index cccd263adf4c..9dd68bb7ee75 100644 --- a/test/parallel/test-stream-forEach.js +++ b/test/parallel/test-stream-forEach.js @@ -43,6 +43,30 @@ const { once } = require('events'); })().then(common.mustCall()); } +{ + // forEach awaits thenables returned by the callback. + const visited = []; + const receivers = []; + const thenables = []; + (async () => { + await Readable.from([1, 2]).forEach((value) => { + const thenable = { + then(resolve) { + receivers.push(this); + setImmediate(() => { + visited.push(value); + resolve(); + }); + }, + }; + thenables.push(thenable); + return thenable; + }); + assert.deepStrictEqual(visited, [1, 2]); + assert.deepStrictEqual(receivers, thenables); + })().then(common.mustCall()); +} + { // forEach works on an infinite stream const ac = new AbortController(); diff --git a/test/parallel/test-stream-map.js b/test/parallel/test-stream-map.js index 212658313885..2e389b06df25 100644 --- a/test/parallel/test-stream-map.js +++ b/test/parallel/test-stream-map.js @@ -27,11 +27,32 @@ function createDependentPromises(n) { return promiseAndResolveArray; } +{ + // Map works on empty streams with a synchronous mapper + const stream = Readable.from([]).map((x) => x); + (async () => { + assert.deepStrictEqual(await stream.toArray(), []); + })().then(common.mustCall()); +} + { // Map works on synchronous streams with a synchronous mapper - const stream = Readable.from([1, 2, 3, 4, 5]).map((x) => x + x); + let mapperSignal; + const stream = Readable.from([1, 2, 3, 4, 5]).map((x, { signal }) => { + mapperSignal = signal; + return x + x; + }); (async () => { assert.deepStrictEqual(await stream.toArray(), [2, 4, 6, 8, 10]); + assert.strictEqual(mapperSignal.aborted, true); + })().then(common.mustCall()); +} + +{ + // Double Map works on synchronous streams with a synchronous mapper + const stream = Readable.from([1, 2, 3, 4, 5]).map((x) => x + x).map((x) => x + x); + (async () => { + assert.deepStrictEqual(await stream.toArray(), [4, 8, 12, 16, 20]); })().then(common.mustCall()); } @@ -46,6 +67,65 @@ function createDependentPromises(n) { })().then(common.mustCall()); } +{ + // Map awaits thenables exactly once and preserves their receiver. + (async () => { + let getterCalls = 0; + let thenCalls = 0; + const thenable = { + get then() { + getterCalls++; + return function(resolve) { + thenCalls++; + thenable.receiver = this; + resolve(2); + }; + }, + }; + + assert.deepStrictEqual( + await Readable.from([1]).map(() => thenable).toArray(), + [2], + ); + assert.strictEqual(getterCalls, 1); + assert.strictEqual(thenCalls, 1); + assert.strictEqual(thenable.receiver, thenable); + + const nonThenable = { then: null }; + assert.deepStrictEqual( + await Readable.from([1]).map(() => nonThenable).toArray(), + [nonThenable], + ); + })().then(common.mustCall()); +} + +{ + // Undefined mapper results remain valid object-mode chunks, while null uses + // the standard stream null-value error. + (async () => { + assert.deepStrictEqual( + await Readable.from([1]).map(() => undefined).toArray(), + [undefined], + ); + })().then(common.mustCall()); + + assert.rejects( + Readable.from([1]).map(() => null).toArray(), + { code: 'ERR_STREAM_NULL_VALUES' }, + ).then(common.mustCall()); +} + +{ + // Map works on synchronous streams with an asynchronous mapper + const stream = Readable.from([1, 2, 3, 4, 5]).map((x) => x + x).map(async (x) => { + await Promise.resolve(); + return x + x; + }); + (async () => { + assert.deepStrictEqual(await stream.toArray(), [4, 8, 12, 16, 20]); + })().then(common.mustCall()); +} + { // Map works on asynchronous streams with a asynchronous mapper const stream = Readable.from([1, 2, 3, 4, 5]).map(async (x) => { @@ -57,7 +137,23 @@ function createDependentPromises(n) { } { - // Map works on an infinite stream + // Map works on an infinite stream - sync + const stream = Readable.from(async function* () { + while (true) yield 1; + }()).map(common.mustCall((x) => { + return x + x; + }, 5)); + (async () => { + let i = 1; + for await (const item of stream) { + assert.strictEqual(item, 2); + if (++i === 5) break; + } + })().then(common.mustCall()); +} + +{ + // Map works on an infinite stream - async const stream = Readable.from(async function* () { while (true) yield 1; }()).map(common.mustCall(async (x) => { @@ -91,6 +187,29 @@ function createDependentPromises(n) { })().then(common.mustCall()); } +{ + // Mapping a Readable subclass does not invoke its constructor internally. + const required = Symbol('required'); + class CustomReadable extends Readable { + constructor(token) { + if (token !== required) { + throw new Error('CustomReadable requires its constructor token'); + } + super({ objectMode: true }); + } + + _read() { + this.push(1); + this.push(null); + } + } + + (async () => { + const stream = new CustomReadable(required).map((x) => x + 1); + assert.deepStrictEqual(await stream.toArray(), [2]); + })().then(common.mustCall()); +} + { // Does not care about data events const source = new Readable({ @@ -126,6 +245,23 @@ function createDependentPromises(n) { ).then(common.mustCall()); } +{ + // Errors from the source stream are propagated through the mapped stream. + const error = new Error('source boom'); + const source = new Readable({ + objectMode: true, + read() { + this.push(1); + this.destroy(error); + }, + }); + + assert.rejects( + source.map((x) => x).toArray(), + error, + ).then(common.mustCall()); +} + { // Throwing an error during `map` (sync) const stream = Readable.from([1, 2, 3, 4, 5]).map((x) => { @@ -177,6 +313,18 @@ function createDependentPromises(n) { }); } +{ + // AbortSignal wakes a mapper waiting on an idle source. + const ac = new AbortController(); + const source = new Readable({ read() {} }); + const result = source.map(common.mustNotCall(), { signal: ac.signal }).toArray(); + + setImmediate(() => ac.abort()); + assert.rejects(result, { name: 'AbortError' }).then(common.mustCall(() => { + assert.strictEqual(source.destroyed, true); + })); +} + { // Concurrency result order const stream = Readable.from([1, 2]).map(async (item, { signal }) => { @@ -350,6 +498,12 @@ function createDependentPromises(n) { assert.throws(() => Readable.from([1]).map((x) => x, { concurrency: -1 }), /ERR_OUT_OF_RANGE/); + assert.throws(() => Readable.from([1]).map((x) => x, { + highWaterMark: 'Foo' + }), /ERR_OUT_OF_RANGE/); + assert.throws(() => Readable.from([1]).map((x) => x, { + highWaterMark: -1 + }), /ERR_OUT_OF_RANGE/); assert.throws(() => Readable.from([1]).map((x) => x, 1), /ERR_INVALID_ARG_TYPE/); assert.throws(() => Readable.from([1]).map((x) => x, { signal: true }), /ERR_INVALID_ARG_TYPE/); } diff --git a/test/parallel/test-stream-reduce.js b/test/parallel/test-stream-reduce.js index 99029f6f8310..0f9b6bf080f4 100644 --- a/test/parallel/test-stream-reduce.js +++ b/test/parallel/test-stream-reduce.js @@ -50,12 +50,90 @@ function sum(p, c) { assert.strictEqual(six, 6); })().then(common.mustCall()); } + +{ + // The reducer receives a signal that is aborted when reduction completes. + let reducerSignal; + (async () => { + const result = await Readable.from([1]).reduce((previous, current, { signal }) => { + reducerSignal = signal; + return previous + current; + }, 0); + assert.strictEqual(result, 1); + assert.strictEqual(reducerSignal.aborted, true); + })().then(common.mustCall()); +} + +{ + // Works with thenables and preserves their receiver. + (async () => { + const receivers = []; + const thenables = []; + const result = await Readable.from([2, 3]).reduce((previous, current) => { + const thenable = { + value: previous + current, + then(resolve) { + receivers.push(this); + resolve(this.value); + }, + }; + thenables.push(thenable); + return thenable; + }, 1); + assert.strictEqual(result, 6); + assert.deepStrictEqual(receivers, thenables); + + const nonThenable = { then: null }; + const objectResult = await Readable.from([1]).reduce( + () => nonThenable, + 0, + ); + assert.strictEqual(objectResult, nonThenable); + })().then(common.mustCall()); +} + +{ + // Synchronous reducer failures, including a throwing `then` getter, + // reject the operation and destroy the stream. + const syncError = new Error('sync boom'); + const syncStream = Readable.from([1]); + assert.rejects(syncStream.reduce(() => { + throw syncError; + }, 0), syncError).then(common.mustCall(() => { + assert.strictEqual(syncStream.destroyed, true); + })); + + const getterError = new Error('then boom'); + const getterStream = Readable.from([1]); + assert.rejects(getterStream.reduce(() => ({ + get then() { + throw getterError; + }, + }), 0), getterError).then(common.mustCall(() => { + assert.strictEqual(getterStream.destroyed, true); + })); +} + +{ + // Errors from the source stream reject the reduction. + const error = new Error('source boom'); + const stream = new Readable({ + objectMode: true, + read() { + this.push(1); + this.destroy(error); + }, + }); + + assert.rejects(stream.reduce(sum, 0), error).then(common.mustCall()); +} + { // Works lazily assert.rejects(Readable.from([1, 2, 3, 4, 5, 6]) .map(common.mustCall((x) => { return x; - }, 3)) // Two consumed and one buffered by `map` due to default concurrency + }, 2)) .reduce(async (p, c) => { if (p === 1) { throw new Error('boom'); @@ -119,6 +197,10 @@ function sum(p, c) { { // Error cases + assert.rejects(Readable.from([]).reduce(sum), { + code: 'ERR_MISSING_ARGS', + message: 'Reduce of an empty stream requires an initial value', + }).then(common.mustCall()); assert.rejects(() => Readable.from([]).reduce(1), /TypeError/).then(common.mustCall()); assert.rejects(() => Readable.from([]).reduce('5'), /TypeError/).then(common.mustCall()); assert.rejects(() => Readable.from([]).reduce((x, y) => x + y, 0, 1), /ERR_INVALID_ARG_TYPE/).then(common.mustCall()); diff --git a/test/parallel/test-stream-some-find-every.mjs b/test/parallel/test-stream-some-find-every.mjs index 0617102bc471..790fb8243916 100644 --- a/test/parallel/test-stream-some-find-every.mjs +++ b/test/parallel/test-stream-some-find-every.mjs @@ -47,6 +47,43 @@ function oneTo5Async() { assert.strictEqual(await oneTo5().find(async (x) => x > 6), undefined); } +{ + // Some, find, and every await thenable predicates. + const expected = new Map([ + ['some', false], + ['every', false], + ['find', undefined], + ]); + + for (const op of expected.keys()) { + const thenable = { + then(resolve) { + thenable.receiver = this; + resolve(false); + }, + }; + assert.strictEqual( + await Readable.from([1])[op](() => thenable), + expected.get(op), + ); + assert.strictEqual(thenable.receiver, thenable); + } +} + +{ + // Predicates receive an options object containing an AbortSignal. + for (const op of ['some', 'every', 'find']) { + let predicateSignal; + await Readable.from([1])[op](common.mustCall((value, { signal }) => { + assert.strictEqual(value, 1); + assert.ok(signal instanceof AbortSignal); + predicateSignal = signal; + return true; + })); + assert.strictEqual(predicateSignal.aborted, true); + } +} + { // Some, find, and every work on asynchronous streams with an asynchronous predicate assert.strictEqual(await oneTo5Async().some(async (x) => x > 3), true); @@ -115,6 +152,52 @@ function oneTo5Async() { assert.strictEqual(found, 1); } +{ + // Errors from any active predicate take precedence over a concurrent match. + const first = Promise.withResolvers(); + const second = Promise.withResolvers(); + const error = new Error('boom'); + const result = Readable.from([1, 2]).find(common.mustCall((val) => { + return val === 1 ? first.promise : second.promise; + }, 2), { concurrency: 2 }); + + second.resolve(true); + await setTimeout(); + first.reject(error); + await assert.rejects(result, error); +} + +{ + // Synchronous throws and rejected promises from predicates are propagated. + for (const op of ['some', 'every', 'find']) { + const syncError = new Error(`${op} sync`); + const syncStream = oneTo5(); + await assert.rejects(syncStream[op](common.mustCall(() => { + throw syncError; + }, 1)), syncError); + assert.strictEqual(syncStream.destroyed, true); + + const asyncError = new Error(`${op} async`); + const asyncStream = oneTo5(); + await assert.rejects(asyncStream[op](common.mustCall(async () => { + throw asyncError; + }, 1)), asyncError); + assert.strictEqual(asyncStream.destroyed, true); + } +} + +{ + // Find can leave the source open after a match. + const stream = oneTo5(); + const found = await stream.find((x) => x === 2, { + destroyOnReturn: false, + }); + assert.strictEqual(found, 2); + assert.strictEqual(stream.destroyed, false); + assert.strictEqual(stream.read(), 3); + stream.destroy(); +} + { // Support for AbortSignal for (const op of ['some', 'every', 'find']) { @@ -158,6 +241,11 @@ function oneTo5Async() { signal: true }); }, /ERR_INVALID_ARG_TYPE/, `${op} should throw for invalid signal`).then(common.mustCall()); + assert.rejects(async () => { + await Readable.from([1])[op]((x) => x, { + destroyOnReturn: 'false' + }); + }, /ERR_INVALID_ARG_TYPE/, `${op} should throw for invalid destroyOnReturn`).then(common.mustCall()); } } { diff --git a/test/parallel/test-stream-toArray.js b/test/parallel/test-stream-toArray.js index 438e21724e40..ff19bf3f3c1f 100644 --- a/test/parallel/test-stream-toArray.js +++ b/test/parallel/test-stream-toArray.js @@ -54,6 +54,31 @@ const assert = require('assert'); })().then(common.mustCall()); } +{ + // destroyOnReturn can preserve streams that do not auto-destroy. + const stream = new Readable({ + objectMode: true, + autoDestroy: false, + read() { + this.push(1); + this.push(2); + this.push(null); + }, + }); + + (async () => { + assert.deepStrictEqual( + await stream.toArray({ destroyOnReturn: false }), + [1, 2], + ); + assert.strictEqual(stream.destroyed, false); + for (const event of ['end', 'finish', 'error', 'close']) { + assert.strictEqual(stream.listenerCount(event), 0); + } + stream.destroy(); + })().then(common.mustCall()); +} + { // Support for AbortSignal const ac = new AbortController(); @@ -74,6 +99,44 @@ const assert = require('assert'); })); ac.abort(); } + +{ + // AbortSignal wakes toArray while its source is idle. + const ac = new AbortController(); + const stream = new Readable({ read() {} }); + const result = stream.toArray({ signal: ac.signal }); + + setImmediate(() => ac.abort()); + assert.rejects(result, { name: 'AbortError' }).then(common.mustCall(() => { + assert.strictEqual(stream.listenerCount('readable'), 0); + })); +} + +{ + // A pre-aborted signal prevents the source from being read. + const stream = new Readable({ + read: common.mustNotCall(), + }); + + assert.rejects( + stream.toArray({ signal: AbortSignal.abort() }), + { name: 'AbortError' }, + ).then(common.mustCall()); +} + +{ + // Source errors reject and remove the readable listener. + const error = new Error('boom'); + const stream = new Readable({ + read() { + this.destroy(error); + }, + }); + + assert.rejects(stream.toArray(), error).then(common.mustCall(() => { + assert.strictEqual(stream.listenerCount('readable'), 0); + })); +} { // Test result is a Promise const result = Readable.from([1, 2, 3, 4, 5]).toArray(); @@ -90,4 +153,10 @@ const assert = require('assert'); signal: true }); }, /ERR_INVALID_ARG_TYPE/).then(common.mustCall()); + + assert.rejects(async () => { + await Readable.from([1]).toArray({ + destroyOnReturn: 'false' + }); + }, /ERR_INVALID_ARG_TYPE/).then(common.mustCall()); }