FazBrowse GitHub Viewer | Trending |
URL:
| Home
Tools: [Download Repo ZIP]   [Original HTTPS Page]

stream: refine the stream/iter backpressure · nodejs/node@a46ddad · GitHub

/ node Public

Commit a46ddad

Browse files
jasnell
authored and
aduh95
committed
stream: refine the stream/iter backpressure
In `Writer`, there are two queues that matter: the slot queue and the pending queue. The sync `writeSync`/`writevSync` only use the slot queue. If writes cannot be accepted directly into slots, they return `false`. The sync methods *never* enqueue into the pending queue. The async `write`/`writev` will first attempt to add to the slot queue; if it is full, then it will attempt to add to the pending queue; if that is also full, the backpressure policy kicks in. Signed-off-by: James M Snell <jasnell@gmail.com> PR-URL: #63697 Backport-PR-URL: #64675 Reviewed-By: Trivikram Kamat <trivikr.dev@gmail.com> Reviewed-By: Matteo Collina <matteo.collina@gmail.com> Reviewed-By: Gürgün Dayıoğlu <hey@gurgun.day>
1 parent 557bf99 commit a46ddad

9 files changed

Lines changed: 185 additions & 201 deletions

File tree

‎doc/api/stream_iter.md‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -1649,7 +1649,7 @@ Creates a classic [`stream.Writable`][] backed by a stream/iter Writer.
16491649

16501650
Each `_write()` / `_writev()` call attempts the Writer's synchronous method
16511651
first (`writeSync` / `writevSync`), falling back to the async method if the
1652-
sync path returns `false` or throws. Similarly, `_final()` tries `endSync()`
1652+
sync path returns `false`. Similarly, `_final()` tries `endSync()`
16531653
before `end()`. When the sync path succeeds, the callback is deferred via
16541654
`queueMicrotask` to preserve the async resolution contract.
16551655

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

Lines changed: 16 additions & 51 deletions
Original file line numberDiff line numberDiff line change
@@ -57,7 +57,6 @@ const {
5757
const {
5858
toAsyncStreamable: kToAsyncStreamable,
5959
kValidatedSource,
60-
kSyncWriteAccepted,
6160
drainableProtocol,
6261
} = require('internal/streams/iter/types');
6362

@@ -765,41 +764,11 @@ function toWritable(writer) {
765764
const hasEndSync = hasEnd &&
766765
typeof writer.endSync === 'function';
767766
const hasFail = typeof writer.fail === 'function';
768-
const hasSyncWriteAccepted =
769-
typeof writer[kSyncWriteAccepted] === 'function';
770-
771-
function syncWriteAccepted() {
772-
return hasSyncWriteAccepted && writer[kSyncWriteAccepted]();
773-
}
774-
775-
function finishAfterSyncBackpressure(cb) {
776-
let ondrain;
777-
try {
778-
if (typeof writer[drainableProtocol] === 'function') {
779-
ondrain = writer[drainableProtocol]();
780-
}
781-
} catch (err) {
782-
cb(err);
783-
return;
784-
}
785-
if (ondrain !== null && ondrain !== undefined) {
786-
PromisePrototypeThen(ondrain, (drained) => {
787-
if (drained === false) {
788-
cb(new ERR_INVALID_STATE.TypeError('Stream closed by consumer'));
789-
return;
790-
}
791-
cb();
792-
}, cb);
793-
return;
794-
}
795-
queueMicrotask(cb);
796-
}
797-
798767
// Try-sync-first pattern: attempt the synchronous method and fall back to the
799-
// async method if it returns false without accepting the data, or if it
800-
// throws. When the sync path succeeds, the callback is deferred via
801-
// queueMicrotask to preserve the async resolution contract that Writable
802-
// internals expect from _write/_writev/_final callbacks.
768+
// async method if it returns false (data not accepted synchronously).
769+
// When the sync path succeeds, the callback is deferred via queueMicrotask
770+
// to preserve the async resolution contract that Writable internals expect
771+
// from _write/_writev/_final callbacks.
803772

804773
function _write(chunk, encoding, cb) {
805774
const bytes = typeof chunk === 'string' ?
@@ -810,13 +779,10 @@ function toWritable(writer) {
810779
queueMicrotask(cb);
811780
return;
812781
}
813-
if (syncWriteAccepted()) {
814-
// The chunk was accepted; false only signaled backpressure.
815-
finishAfterSyncBackpressure(cb);
816-
return;
817-
}
818-
} catch {
819-
// Sync path threw -- fall through to async.
782+
// WriteSync returned false: not accepted, fall through to async.
783+
} catch (err) {
784+
cb(err);
785+
return;
820786
}
821787
}
822788
try {
@@ -839,13 +805,10 @@ function toWritable(writer) {
839805
queueMicrotask(cb);
840806
return;
841807
}
842-
if (syncWriteAccepted()) {
843-
// The chunks were accepted; false only signaled backpressure.
844-
finishAfterSyncBackpressure(cb);
845-
return;
846-
}
847-
} catch {
848-
// Sync path threw -- fall through to async.
808+
// WritevSync returned false: not accepted, fall through to async.
809+
} catch (err) {
810+
cb(err);
811+
return;
849812
}
850813
}
851814
try {
@@ -867,8 +830,10 @@ function toWritable(writer) {
867830
queueMicrotask(cb);
868831
return;
869832
}
870-
} catch {
871-
// Sync path threw -- fall through to async.
833+
// Result < 0: can't end synchronously, fall through to async.
834+
} catch (err) {
835+
cb(err);
836+
return;
872837
}
873838
}
874839
try {

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

Lines changed: 35 additions & 10 deletions
Original file line numberDiff line numberDiff line change
@@ -480,6 +480,7 @@ function merge(...args) {
480480
);
481481
}
482482

483+
let primaryError;
483484
try {
484485
while (activeCount > 0 || ready.length > 0) {
485486
signal?.throwIfAborted();
@@ -500,22 +501,46 @@ function merge(...args) {
500501
});
501502
}
502503
}
504+
} catch (err) {
505+
primaryError = err;
503506
} finally {
504-
// Clean up: return all iterators
505-
await SafePromiseAllReturnVoid(iterators, async (iterator) => {
506-
if (iterator.return) {
507-
try {
508-
await iterator.return();
509-
} catch {
510-
// Ignore return errors
511-
}
512-
}
513-
});
507+
// Clean up: return all iterators. Cleanup errors are not
508+
// swallowed - a broken iterator.return() (e.g., failing to
509+
// release a resource) should be visible to the caller.
510+
await cleanupIterators(iterators, primaryError);
514511
}
515512
},
516513
};
517514
}
518515

516+
async function cleanupIterators(iterators, primaryError) {
517+
let cleanupError;
518+
await SafePromiseAllReturnVoid(iterators, async (iterator) => {
519+
if (iterator.return) {
520+
try {
521+
await iterator.return();
522+
} catch (err) {
523+
// Keep the first cleanup error encountered.
524+
cleanupError ??= err;
525+
}
526+
}
527+
});
528+
if (cleanupError !== undefined) {
529+
if (primaryError !== undefined) {
530+
// Both a primary error and a cleanup error occurred.
531+
// Wrap in SuppressedError so neither is lost:
532+
// .error = primaryError, .suppressed = cleanupError.
533+
// eslint-disable-next-line no-restricted-syntax
534+
throw new SuppressedError(primaryError, cleanupError);
535+
}
536+
// No primary error - the cleanup error is the only error.
537+
throw cleanupError;
538+
}
539+
if (primaryError !== undefined) {
540+
throw primaryError;
541+
}
542+
}
543+
519544
module.exports = {
520545
array,
521546
arrayBuffer,

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

Lines changed: 5 additions & 51 deletions
Original file line numberDiff line numberDiff line change
@@ -63,8 +63,6 @@ const {
6363
} = require('internal/streams/iter/utils');
6464

6565
const {
66-
drainableProtocol,
67-
kSyncWriteAcceptedOnFalse,
6866
kValidatedSource,
6967
kValidatedTransform,
7068
toAsyncStreamable,
@@ -863,18 +861,6 @@ async function* createAsyncPipeline(source, transforms, signal) {
863861
}
864862
}
865863

866-
/**
867-
* Check if a false sync write result means accepted backpressure.
868-
* @param {object} writer - The writer whose sync method returned.
869-
* @param {*} result - The return value from writeSync() or writevSync().
870-
* @returns {boolean}
871-
*/
872-
function isAcceptedSyncWriteBackpressure(writer, result) {
873-
return result === false &&
874-
writer[kSyncWriteAcceptedOnFalse] === true &&
875-
writer.desiredSize === 0;
876-
}
877-
878864
// =============================================================================
879865
// Public API: pull() and pullSync()
880866
// =============================================================================
@@ -963,9 +949,7 @@ function pipeToSync(source, ...args) {
963949
break;
964950
}
965951
if (hasWritevSync && batch.length > 1) {
966-
const result = writer.writevSync(batch);
967-
if (result === false &&
968-
!isAcceptedSyncWriteBackpressure(writer, result)) {
952+
if (writer.writevSync(batch) === false) {
969953
break;
970954
}
971955
for (let i = 0; i < batch.length; i++) {
@@ -974,9 +958,7 @@ function pipeToSync(source, ...args) {
974958
} else {
975959
for (let i = 0; i < batch.length; i++) {
976960
const chunk = batch[i];
977-
const result = writer.writeSync(chunk);
978-
if (result === false &&
979-
!isAcceptedSyncWriteBackpressure(writer, result)) {
961+
if (writer.writeSync(chunk) === false) {
980962
canContinue = false;
981963
break;
982964
}
@@ -1027,28 +1009,13 @@ async function pipeTo(source, ...args) {
10271009
const hasWritevSync = typeof writer.writevSync === 'function';
10281010
const hasEndSync = typeof writer.endSync === 'function';
10291011

1030-
function waitForSyncBackpressure() {
1031-
const ondrain = writer[drainableProtocol];
1032-
return ondrain?.call(writer);
1033-
}
1034-
1035-
async function writeBatchAfterAcceptedBackpressure(batch, startIndex) {
1036-
await waitForSyncBackpressure();
1037-
await writeBatchAsyncFallback(batch, startIndex);
1038-
}
1039-
10401012
// Async fallback for writeBatch when sync write fails partway through.
10411013
// Continues writing from batch[startIndex] using async write().
10421014
async function writeBatchAsyncFallback(batch, startIndex) {
10431015
for (let i = startIndex; i < batch.length; i++) {
10441016
const chunk = batch[i];
1045-
const result = hasWriteSync && writer.writeSync(chunk);
1046-
if (result) {
1017+
if (hasWriteSync && writer.writeSync(chunk)) {
10471018
// Sync retry succeeded
1048-
} else if (isAcceptedSyncWriteBackpressure(writer, result)) {
1049-
totalBytes += TypedArrayPrototypeGetByteLength(chunk);
1050-
await waitForSyncBackpressure();
1051-
continue;
10521019
} else {
10531020
const result = writer.write(
10541021
chunk, signal ? { __proto__: null, signal } : undefined);
@@ -1065,14 +1032,7 @@ async function pipeTo(source, ...args) {
10651032
// is required. Callers must check: const p = writeBatch(b); if (p) await p;
10661033
function writeBatch(batch) {
10671034
if (hasWritev && batch.length > 1) {
1068-
const result = hasWritevSync && writer.writevSync(batch);
1069-
if (!result) {
1070-
if (isAcceptedSyncWriteBackpressure(writer, result)) {
1071-
for (let i = 0; i < batch.length; i++) {
1072-
totalBytes += TypedArrayPrototypeGetByteLength(batch[i]);
1073-
}
1074-
return waitForSyncBackpressure();
1075-
}
1035+
if (!hasWritevSync || !writer.writevSync(batch)) {
10761036
const opts = signal ? { __proto__: null, signal } : undefined;
10771037
const writevResult = writer.writev(batch, opts);
10781038
if (writevResult === undefined) {
@@ -1094,14 +1054,8 @@ async function pipeTo(source, ...args) {
10941054
}
10951055
for (let i = 0; i < batch.length; i++) {
10961056
const chunk = batch[i];
1097-
const result = hasWriteSync && writer.writeSync(chunk);
1098-
if (!result) {
1099-
if (isAcceptedSyncWriteBackpressure(writer, result)) {
1100-
totalBytes += TypedArrayPrototypeGetByteLength(chunk);
1101-
return writeBatchAfterAcceptedBackpressure(batch, i + 1);
1102-
}
1057+
if (!hasWriteSync || !writer.writeSync(chunk)) {
11031058
// Sync path failed at index i - fall back to async for the rest.
1104-
// Count bytes for chunks already written synchronously (0..i-1).
11051059
return writeBatchAsyncFallback(batch, i);
11061060
}
11071061
totalBytes += TypedArrayPrototypeGetByteLength(chunk);

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

Lines changed: 2 additions & 46 deletions
Original file line numberDiff line numberDiff line change
@@ -32,8 +32,6 @@ const {
3232

3333
const {
3434
drainableProtocol,
35-
kSyncWriteAccepted,
36-
kSyncWriteAcceptedOnFalse,
3735
} = require('internal/streams/iter/types');
3836

3937
const {
@@ -364,19 +362,6 @@ class PushQueue {
364362
this.#pendingEnd = pending;
365363
}
366364

367-
/**
368-
* Force-enqueue chunks into the slots buffer, bypassing capacity checks.
369-
* Used by PushWriter.writeSync() for 'block' policy where the data is
370-
* accepted but false is returned as a backpressure signal.
371-
*/
372-
forceEnqueue(chunks) {
373-
this.#slots.push(chunks);
374-
for (let i = 0; i < chunks.length; i++) {
375-
this.#bytesWritten += TypedArrayPrototypeGetByteLength(chunks[i]);
376-
}
377-
this.#resolvePendingReads();
378-
}
379-
380365
/**
381366
* Wait for backpressure to clear (desiredSize > 0).
382367
* @returns {Promise<void>}
@@ -558,16 +543,11 @@ class PushQueue {
558543

559544
class PushWriter {
560545
#queue;
561-
#syncWriteAccepted = false;
562546

563547
constructor(queue) {
564548
this.#queue = queue;
565549
}
566550

567-
[kSyncWriteAccepted]() {
568-
return this.#syncWriteAccepted;
569-
}
570-
571551
[drainableProtocol]() {
572552
const desired = this.desiredSize;
573553
if (desired === null) return null;
@@ -579,10 +559,6 @@ class PushWriter {
579559
return this.#queue.desiredSize;
580560
}
581561

582-
get [kSyncWriteAcceptedOnFalse]() {
583-
return this.#queue.backpressurePolicy === 'block';
584-
}
585-
586562
write(chunk, options) {
587563
if (!options?.signal && this.#queue.canWriteSync()) {
588564
const bytes = toUint8Array(chunk);
@@ -607,36 +583,16 @@ class PushWriter {
607583
}
608584

609585
writeSync(chunk) {
610-
this.#syncWriteAccepted = false;
611586
const bytes = toUint8Array(chunk);
612-
const result = this.#queue.writeSync([bytes]);
613-
if (!result && this.#queue.backpressurePolicy === 'block' &&
614-
this.#queue.desiredSize === 0) {
615-
// Block policy: force-enqueue and return false as backpressure signal.
616-
// Data IS accepted; false tells caller to slow down.
617-
this.#queue.forceEnqueue([bytes]);
618-
this.#syncWriteAccepted = true;
619-
return false;
620-
}
621-
this.#syncWriteAccepted = result;
622-
return result;
587+
return this.#queue.writeSync([bytes]);
623588
}
624589

625590
writevSync(chunks) {
626-
this.#syncWriteAccepted = false;
627591
if (!ArrayIsArray(chunks)) {
628592
throw new ERR_INVALID_ARG_TYPE('chunks', 'Array', chunks);
629593
}
630594
const bytes = convertChunks(chunks);
631-
const result = this.#queue.writeSync(bytes);
632-
if (!result && this.#queue.backpressurePolicy === 'block' &&
633-
this.#queue.desiredSize === 0) {
634-
this.#queue.forceEnqueue(bytes);
635-
this.#syncWriteAccepted = true;
636-
return false;
637-
}
638-
this.#syncWriteAccepted = result;
639-
return result;
595+
return this.#queue.writeSync(bytes);
640596
}
641597

642598
end(options) {

0 commit comments

Comments
 (0)
Back | FazBrowse Home | New Git URL