GeezerVison is showing github.com — stream: add `highWaterMark` for the map operator · nodejs/node@651e450
File treeExpand file treeCollapse file tree Open diff view settings Expand file treeCollapse file tree Open diff view settings Collapse file Copy file name to clipboardExpand all lines: doc/api/stream.md +12Lines changed: 12 additions & 0 deletions - Display the source diff
- Display the rich diff
| Original file line number | Diff line number | Diff line change |
|---|
@@ -2008,6 +2008,10 @@ showBoth(); | 2008 | 2008 | added: | 2009 | 2009 | - v17.4.0 | 2010 | 2010 | - v16.14.0 | | 2011 | +changes: | | 2012 | + - version: REPLACEME | | 2013 | + pr-url: https://github.com/nodejs/node/pull/49249 | | 2014 | + description: added `highWaterMark` in options. | 2011 | 2015 | --> | 2012 | 2016 |
| 2013 | 2017 | > Stability: 1 - Experimental | @@ -2021,6 +2025,8 @@ added: | 2021 | 2025 | * `options` {Object} | 2022 | 2026 | * `concurrency` {number} the maximum concurrent invocation of `fn` to call | 2023 | 2027 | on the stream at once. **Default:** `1`. | | 2028 | + * `highWaterMark` {number} how many items to buffer while waiting for user | | 2029 | + consumption of the mapped items. **Default:** `concurrency * 2 - 1`. | 2024 | 2030 | * `signal` {AbortSignal} allows destroying the stream if the signal is | 2025 | 2031 | aborted. | 2026 | 2032 | * Returns: {Readable} a stream mapped with the function `fn`. | @@ -2055,6 +2061,10 @@ for await (const result of dnsResults) { | 2055 | 2061 | added: | 2056 | 2062 | - v17.4.0 | 2057 | 2063 | - v16.14.0 | | 2064 | +changes: | | 2065 | + - version: REPLACEME | | 2066 | + pr-url: https://github.com/nodejs/node/pull/49249 | | 2067 | + description: added `highWaterMark` in options. | 2058 | 2068 | --> | 2059 | 2069 |
| 2060 | 2070 | > Stability: 1 - Experimental | @@ -2067,6 +2077,8 @@ added: | 2067 | 2077 | * `options` {Object} | 2068 | 2078 | * `concurrency` {number} the maximum concurrent invocation of `fn` to call | 2069 | 2079 | on the stream at once. **Default:** `1`. | | 2080 | + * `highWaterMark` {number} how many items to buffer while waiting for user | | 2081 | + consumption of the filtered items. **Default:** `concurrency * 2 - 1`. | 2070 | 2082 | * `signal` {AbortSignal} allows destroying the stream if the signal is | 2071 | 2083 | aborted. | 2072 | 2084 | * Returns: {Readable} a stream filtered with the predicate `fn`. | |
Collapse file Copy file name to clipboardExpand all lines: lib/internal/streams/operators.js +41-14Lines changed: 41 additions & 14 deletions | Original file line number | Diff line number | Diff line change |
|---|
@@ -33,6 +33,7 @@ const { | 33 | 33 | NumberIsNaN, | 34 | 34 | Promise, | 35 | 35 | PromiseReject, | | 36 | + PromiseResolve, | 36 | 37 | PromisePrototypeThen, | 37 | 38 | Symbol, | 38 | 39 | } = primordials; | @@ -82,7 +83,15 @@ function map(fn, options) { | 82 | 83 | concurrency = MathFloor(options.concurrency); | 83 | 84 | } | 84 | 85 |
| 85 | | - validateInteger(concurrency, 'concurrency', 1); | | 86 | + let highWaterMark = concurrency - 1; | | 87 | + if (options?.highWaterMark != null) { | | 88 | + highWaterMark = MathFloor(options.highWaterMark); | | 89 | + } | | 90 | + | | 91 | + validateInteger(concurrency, 'options.concurrency', 1); | | 92 | + validateInteger(highWaterMark, 'options.highWaterMark', 0); | | 93 | + | | 94 | + highWaterMark += concurrency; | 86 | 95 |
| 87 | 96 | return async function* map() { | 88 | 97 | const signal = AbortSignal.any([options?.signal].filter(Boolean)); | @@ -93,9 +102,28 @@ function map(fn, options) { | 93 | 102 | let next; | 94 | 103 | let resume; | 95 | 104 | let done = false; | | 105 | + let cnt = 0; | 96 | 106 |
| 97 | | - function onDone() { | | 107 | + function onCatch() { | 98 | 108 | done = true; | | 109 | + afterItemProcessed(); | | 110 | + } | | 111 | + | | 112 | + function afterItemProcessed() { | | 113 | + cnt -= 1; | | 114 | + maybeResume(); | | 115 | + } | | 116 | + | | 117 | + function maybeResume() { | | 118 | + if ( | | 119 | + resume && | | 120 | + !done && | | 121 | + cnt < concurrency && | | 122 | + queue.length < highWaterMark | | 123 | + ) { | | 124 | + resume(); | | 125 | + resume = null; | | 126 | + } | 99 | 127 | } | 100 | 128 |
| 101 | 129 | async function pump() { | @@ -111,25 +139,27 @@ function map(fn, options) { | 111 | 139 |
| 112 | 140 | try { | 113 | 141 | val = fn(val, signalOpt); | | 142 | + | | 143 | + if (val === kEmpty) { | | 144 | + continue; | | 145 | + } | | 146 | + | | 147 | + val = PromiseResolve(val); | 114 | 148 | } catch (err) { | 115 | 149 | val = PromiseReject(err); | 116 | 150 | } | 117 | 151 |
| 118 | | - if (val === kEmpty) { | 119 | | - continue; | 120 | | - } | | 152 | + cnt += 1; | 121 | 153 |
| 122 | | - if (typeof val?.catch === 'function') { | 123 | | - val.catch(onDone); | 124 | | - } | | 154 | + PromisePrototypeThen(val, afterItemProcessed, onCatch); | 125 | 155 |
| 126 | 156 | queue.push(val); | 127 | 157 | if (next) { | 128 | 158 | next(); | 129 | 159 | next = null; | 130 | 160 | } | 131 | 161 |
| 132 | | - if (!done && queue.length && queue.length >= concurrency) { | | 162 | + if (!done && (queue.length >= highWaterMark || cnt >= concurrency)) { | 133 | 163 | await new Promise((resolve) => { | 134 | 164 | resume = resolve; | 135 | 165 | }); | @@ -138,7 +168,7 @@ function map(fn, options) { | 138 | 168 | queue.push(kEof); | 139 | 169 | } catch (err) { | 140 | 170 | const val = PromiseReject(err); | 141 | | - PromisePrototypeThen(val, undefined, onDone); | | 171 | + PromisePrototypeThen(val, afterItemProcessed, onCatch); | 142 | 172 | queue.push(val); | 143 | 173 | } finally { | 144 | 174 | done = true; | @@ -169,10 +199,7 @@ function map(fn, options) { | 169 | 199 | } | 170 | 200 |
| 171 | 201 | queue.shift(); | 172 | | - if (resume) { | 173 | | - resume(); | 174 | | - resume = null; | 175 | | - } | | 202 | + maybeResume(); | 176 | 203 | } | 177 | 204 |
| 178 | 205 | await new Promise((resolve) => { | |
Collapse file Copy file name to clipboardExpand all lines: test/parallel/test-stream-forEach.js +1-1Lines changed: 1 addition & 1 deletion | Original file line number | Diff line number | Diff line change |
|---|
@@ -96,7 +96,7 @@ const { once } = require('events'); | 96 | 96 | Readable.from([1, 2, 3, 4]).forEach(async (_, { signal }) => { | 97 | 97 | calls++; | 98 | 98 | await once(signal, 'abort'); | 99 | | - }, { signal: ac.signal, concurrency: 2 }); | | 99 | + }, { signal: ac.signal, concurrency: 2, highWaterMark: 0 }); | 100 | 100 | // pump | 101 | 101 | assert.rejects(async () => { | 102 | 102 | await forEachPromise; | |
Collapse file Copy file name to clipboardExpand all lines: test/parallel/test-stream-map.js +172-1Lines changed: 172 additions & 1 deletion | Original file line number | Diff line number | Diff line change |
|---|
@@ -8,6 +8,25 @@ const assert = require('assert'); | 8 | 8 | const { once } = require('events'); | 9 | 9 | const { setTimeout } = require('timers/promises'); | 10 | 10 |
| | 11 | +function createDependentPromises(n) { | | 12 | + const promiseAndResolveArray = []; | | 13 | + | | 14 | + for (let i = 0; i < n; i++) { | | 15 | + let res; | | 16 | + const promise = new Promise((resolve) => { | | 17 | + if (i === 0) { | | 18 | + res = resolve; | | 19 | + return; | | 20 | + } | | 21 | + res = () => promiseAndResolveArray[i - 1][0].then(resolve); | | 22 | + }); | | 23 | + | | 24 | + promiseAndResolveArray.push([promise, res]); | | 25 | + } | | 26 | + | | 27 | + return promiseAndResolveArray; | | 28 | +} | | 29 | + | 11 | 30 | { | 12 | 31 | // Map works on synchronous streams with a synchronous mapper | 13 | 32 | const stream = Readable.from([1, 2, 3, 4, 5]).map((x) => x + x); | @@ -143,7 +162,7 @@ const { setTimeout } = require('timers/promises'); | 143 | 162 | const stream = range.map(common.mustCall(async (_, { signal }) => { | 144 | 163 | await once(signal, 'abort'); | 145 | 164 | throw signal.reason; | 146 | | - }, 2), { signal: ac.signal, concurrency: 2 }); | | 165 | + }, 2), { signal: ac.signal, concurrency: 2, highWaterMark: 0 }); | 147 | 166 | // pump | 148 | 167 | assert.rejects(async () => { | 149 | 168 | for await (const item of stream) { | @@ -173,12 +192,164 @@ const { setTimeout } = require('timers/promises'); | 173 | 192 | })().then(common.mustCall()); | 174 | 193 | } | 175 | 194 |
| | 195 | + | | 196 | +{ | | 197 | + // highWaterMark with small concurrency | | 198 | + const finishOrder = []; | | 199 | + | | 200 | + const promises = createDependentPromises(4); | | 201 | + | | 202 | + const raw = Readable.from([2, 0, 1, 3]); | | 203 | + const stream = raw.map(async (item) => { | | 204 | + const [promise, resolve] = promises[item]; | | 205 | + resolve(); | | 206 | + | | 207 | + await promise; | | 208 | + finishOrder.push(item); | | 209 | + return item; | | 210 | + }, { concurrency: 2 }); | | 211 | + | | 212 | + (async () => { | | 213 | + await stream.toArray(); | | 214 | + | | 215 | + assert.deepStrictEqual(finishOrder, [0, 1, 2, 3]); | | 216 | + })().then(common.mustCall(), common.mustNotCall()); | | 217 | +} | | 218 | + | | 219 | +{ | | 220 | + // highWaterMark with a lot of items and large concurrency | | 221 | + const finishOrder = []; | | 222 | + | | 223 | + const promises = createDependentPromises(20); | | 224 | + | | 225 | + const input = [10, 1, 0, 3, 4, 2, 5, 7, 8, 9, 6, 11, 12, 13, 18, 15, 16, 17, 14, 19]; | | 226 | + const raw = Readable.from(input); | | 227 | + // Should be | | 228 | + // 10, 1, 0, 3, 4, 2 | next: 0 | | 229 | + // 10, 1, 3, 4, 2, 5 | next: 1 | | 230 | + // 10, 3, 4, 2, 5, 7 | next: 2 | | 231 | + // 10, 3, 4, 5, 7, 8 | next: 3 | | 232 | + // 10, 4, 5, 7, 8, 9 | next: 4 | | 233 | + // 10, 5, 7, 8, 9, 6 | next: 5 | | 234 | + // 10, 7, 8, 9, 6, 11 | next: 6 | | 235 | + // 10, 7, 8, 9, 11, 12 | next: 7 | | 236 | + // 10, 8, 9, 11, 12, 13 | next: 8 | | 237 | + // 10, 9, 11, 12, 13, 18 | next: 9 | | 238 | + // 10, 11, 12, 13, 18, 15 | next: 10 | | 239 | + // 11, 12, 13, 18, 15, 16 | next: 11 | | 240 | + // 12, 13, 18, 15, 16, 17 | next: 12 | | 241 | + // 13, 18, 15, 16, 17, 14 | next: 13 | | 242 | + // 18, 15, 16, 17, 14, 19 | next: 14 | | 243 | + // 18, 15, 16, 17, 19 | next: 15 | | 244 | + // 18, 16, 17, 19 | next: 16 | | 245 | + // 18, 17, 19 | next: 17 | | 246 | + // 18, 19 | next: 18 | | 247 | + // 19 | next: 19 | | 248 | + // | | 249 | + | | 250 | + const stream = raw.map(async (item) => { | | 251 | + const [promise, resolve] = promises[item]; | | 252 | + resolve(); | | 253 | + | | 254 | + await promise; | | 255 | + finishOrder.push(item); | | 256 | + return item; | | 257 | + }, { concurrency: 6 }); | | 258 | + | | 259 | + (async () => { | | 260 | + const outputOrder = await stream.toArray(); | | 261 | + | | 262 | + assert.deepStrictEqual(outputOrder, input); | | 263 | + assert.deepStrictEqual(finishOrder, [0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13, 14, 15, 16, 17, 18, 19]); | | 264 | + })().then(common.mustCall(), common.mustNotCall()); | | 265 | +} | | 266 | + | | 267 | +{ | | 268 | + // Custom highWaterMark with a lot of items and large concurrency | | 269 | + const finishOrder = []; | | 270 | + | | 271 | + const promises = createDependentPromises(20); | | 272 | + | | 273 | + const input = [11, 1, 0, 3, 4, 2, 5, 7, 8, 9, 6, 10, 12, 13, 18, 15, 16, 17, 14, 19]; | | 274 | + const raw = Readable.from(input); | | 275 | + // Should be | | 276 | + // 11, 1, 0, 3, 4 | next: 0, buffer: [] | | 277 | + // 11, 1, 3, 4, 2 | next: 1, buffer: [0] | | 278 | + // 11, 3, 4, 2, 5 | next: 2, buffer: [0, 1] | | 279 | + // 11, 3, 4, 5, 7 | next: 3, buffer: [0, 1, 2] | | 280 | + // 11, 4, 5, 7, 8 | next: 4, buffer: [0, 1, 2, 3] | | 281 | + // 11, 5, 7, 8, 9 | next: 5, buffer: [0, 1, 2, 3, 4] | | 282 | + // 11, 7, 8, 9, 6 | next: 6, buffer: [0, 1, 2, 3, 4, 5] | | 283 | + // 11, 7, 8, 9, 10 | next: 7, buffer: [0, 1, 2, 3, 4, 5, 6] -- buffer full | | 284 | + // 11, 8, 9, 10, 12 | next: 8, buffer: [0, 1, 2, 3, 4, 5, 6] | | 285 | + // 11, 9, 10, 12, 13 | next: 9, buffer: [0, 1, 2, 3, 4, 5, 6] | | 286 | + // 11, 10, 12, 13, 18 | next: 10, buffer: [0, 1, 2, 3, 4, 5, 6] | | 287 | + // 11, 12, 13, 18, 15 | next: 11, buffer: [0, 1, 2, 3, 4, 5, 6] | | 288 | + // 12, 13, 18, 15, 16 | next: 12, buffer: [] -- all items flushed as 11 is consumed and all the items wait for it | | 289 | + // 13, 18, 15, 16, 17 | next: 13, buffer: [] | | 290 | + // 18, 15, 16, 17, 14 | next: 14, buffer: [] | | 291 | + // 18, 15, 16, 17, 19 | next: 15, buffer: [14] | | 292 | + // 18, 16, 17, 19 | next: 16, buffer: [14, 15] | | 293 | + // 18, 17, 19 | next: 17, buffer: [14, 15, 16] | | 294 | + // 18, 19 | next: 18, buffer: [14, 15, 16, 17] | | 295 | + // 19 | next: 19, buffer: [] -- all items flushed | | 296 | + // | | 297 | + | | 298 | + const stream = raw.map(async (item) => { | | 299 | + const [promise, resolve] = promises[item]; | | 300 | + resolve(); | | 301 | + | | 302 | + await promise; | | 303 | + finishOrder.push(item); | | 304 | + return item; | | 305 | + }, { concurrency: 5, highWaterMark: 7 }); | | 306 | + | | 307 | + (async () => { | | 308 | + const outputOrder = await stream.toArray(); | | 309 | + | | 310 | + assert.deepStrictEqual(outputOrder, input); | | 311 | + assert.deepStrictEqual(finishOrder, [0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13, 14, 15, 16, 17, 18, 19]); | | 312 | + })().then(common.mustCall(), common.mustNotCall()); | | 313 | +} | | 314 | + | | 315 | +{ | | 316 | + // Where there is a delay between the first and the next item it should not wait for filled queue | | 317 | + // before yielding to the user | | 318 | + const promises = createDependentPromises(3); | | 319 | + | | 320 | + const raw = Readable.from([0, 1, 2]); | | 321 | + | | 322 | + const stream = raw | | 323 | + .map(async (item) => { | | 324 | + if (item !== 0) { | | 325 | + await promises[item][0]; | | 326 | + } | | 327 | + | | 328 | + return item; | | 329 | + }, { concurrency: 2 }) | | 330 | + .map((item) => { | | 331 | + // eslint-disable-next-line no-unused-vars | | 332 | + for (const [_, resolve] of promises) { | | 333 | + resolve(); | | 334 | + } | | 335 | + | | 336 | + return item; | | 337 | + }); | | 338 | + | | 339 | + (async () => { | | 340 | + await stream.toArray(); | | 341 | + })().then(common.mustCall(), common.mustNotCall()); | | 342 | +} | | 343 | + | 176 | 344 | { | 177 | 345 | // Error cases | 178 | 346 | assert.throws(() => Readable.from([1]).map(1), /ERR_INVALID_ARG_TYPE/); | 179 | 347 | assert.throws(() => Readable.from([1]).map((x) => x, { | 180 | 348 | concurrency: 'Foo' | 181 | 349 | }), /ERR_OUT_OF_RANGE/); | | 350 | + assert.throws(() => Readable.from([1]).map((x) => x, { | | 351 | + concurrency: -1 | | 352 | + }), /ERR_OUT_OF_RANGE/); | 182 | 353 | assert.throws(() => Readable.from([1]).map((x) => x, 1), /ERR_INVALID_ARG_TYPE/); | 183 | 354 | assert.throws(() => Readable.from([1]).map((x) => x, { signal: true }), /ERR_INVALID_ARG_TYPE/); | 184 | 355 | } | |
|