Commit e48e287
stream: fast-path stateless transform flush results
Avoid generator-based normalization for common flush result types in
fused stateless stream/iter transforms. This keeps the corrected flush
ordering while reducing overhead for already-normalized batches and
primitive byte outputs.
Signed-off-by: Kamat, Trivikram <16024985+trivikr@users.noreply.github.com>
Assisted-by: openai:gpt-5.5
PR-URL: #63605
Backport-PR-URL: #64675
Reviewed-By: Matteo Collina <matteo.collina@gmail.com>1 parent f0ee316 commit e48e287
1 file changed
Lines changed: 64 additions & 4 deletions
| Original file line number | Diff line number | Diff line change | |
|---|---|---|---|
@@ -280,6 +280,31 @@ function* processTransformResultSync(result) { | |||
| 280 | 280 | * @param {*} result | |
| 281 | 281 | */ | |
| 282 | 282 | function appendTransformResultSync(target, result) { | |
| 283 | + if (result === null) { | ||
| 284 | + return; | ||
| 285 | + } | ||
| 286 | + if (isUint8ArrayBatch(result)) { | ||
| 287 | + if (result.length > 0) { | ||
| 288 | + ArrayPrototypePush(target, result); | ||
| 289 | + } | ||
| 290 | + return; | ||
| 291 | + } | ||
| 292 | + if (isUint8Array(result)) { | ||
| 293 | + ArrayPrototypePush(target, [result]); | ||
| 294 | + return; | ||
| 295 | + } | ||
| 296 | + if (typeof result === 'string') { | ||
| 297 | + ArrayPrototypePush(target, [toUint8Array(result)]); | ||
| 298 | + return; | ||
| 299 | + } | ||
| 300 | + if (isAnyArrayBuffer(result)) { | ||
| 301 | + ArrayPrototypePush(target, [new Uint8Array(result)]); | ||
| 302 | + return; | ||
| 303 | + } | ||
| 304 | + if (ArrayBufferIsView(result)) { | ||
| 305 | + ArrayPrototypePush(target, [arrayBufferViewToUint8Array(result)]); | ||
| 306 | + return; | ||
| 307 | + } | ||
| 283 | 308 | for (const batch of processTransformResultSync(result)) { | |
| 284 | 309 | ArrayPrototypePush(target, batch); | |
| 285 | 310 | } | |
@@ -372,9 +397,38 @@ async function* processTransformResultAsync(result) { | |||
| 372 | 397 | * Append normalized transform result batches to an array (async). | |
| 373 | 398 | * @param {Array<Uint8Array[]>} target | |
| 374 | 399 | * @param {*} result | |
| 375 | - * @returns {Promise<void>} | ||
| 400 | + * @returns {Promise<void>|undefined} | ||
| 376 | 401 | */ | |
| 377 | - async function appendTransformResultAsync(target, result) { | ||
| 402 | + function appendTransformResultAsync(target, result) { | ||
| 403 | + if (result === null) { | ||
| 404 | + return; | ||
| 405 | + } | ||
| 406 | + if (isUint8ArrayBatch(result)) { | ||
| 407 | + if (result.length > 0) { | ||
| 408 | + ArrayPrototypePush(target, result); | ||
| 409 | + } | ||
| 410 | + return; | ||
| 411 | + } | ||
| 412 | + if (isUint8Array(result)) { | ||
| 413 | + ArrayPrototypePush(target, [result]); | ||
| 414 | + return; | ||
| 415 | + } | ||
| 416 | + if (typeof result === 'string') { | ||
| 417 | + ArrayPrototypePush(target, [toUint8Array(result)]); | ||
| 418 | + return; | ||
| 419 | + } | ||
| 420 | + if (isAnyArrayBuffer(result)) { | ||
| 421 | + ArrayPrototypePush(target, [new Uint8Array(result)]); | ||
| 422 | + return; | ||
| 423 | + } | ||
| 424 | + if (ArrayBufferIsView(result)) { | ||
| 425 | + ArrayPrototypePush(target, [arrayBufferViewToUint8Array(result)]); | ||
| 426 | + return; | ||
| 427 | + } | ||
| 428 | + return appendTransformResultAsyncSlow(target, result); | ||
| 429 | + } | ||
| 430 | + | ||
| 431 | + async function appendTransformResultAsyncSlow(target, result) { | ||
| 378 | 432 | for await (const batch of processTransformResultAsync(result)) { | |
| 379 | 433 | ArrayPrototypePush(target, batch); | |
| 380 | 434 | } | |
@@ -553,13 +607,19 @@ async function* applyFusedStatelessAsyncTransforms(source, run, signal) { | |||
| 553 | 607 | for (let i = 0; i < run.length; i++) { | |
| 554 | 608 | const next = []; | |
| 555 | 609 | for (let j = 0; j < pending.length; j++) { | |
| 556 | - await appendTransformResultAsync( | ||
| 610 | + const pendingResult = appendTransformResultAsync( | ||
| 557 | 611 | next, | |
| 558 | 612 | run[i](pending[j], { __proto__: null, signal })); | |
| 613 | + if (pendingResult !== undefined) { | ||
| 614 | + await pendingResult; | ||
| 615 | + } | ||
| 559 | 616 | } | |
| 560 | - await appendTransformResultAsync( | ||
| 617 | + const flushResult = appendTransformResultAsync( | ||
| 561 | 618 | next, | |
| 562 | 619 | run[i](null, { __proto__: null, signal })); | |
| 620 | + if (flushResult !== undefined) { | ||
| 621 | + await flushResult; | ||
| 622 | + } | ||
| 563 | 623 | pending = next; | |
| 564 | 624 | } | |
| 565 | 625 | for (let i = 0; i < pending.length; i++) { | |
0 commit comments