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

stream: drain pending writes before broadcast end · nodejs/node@dc51c79 · GitHub

/ node Public

Commit dc51c79

Browse files
trivikr
authored and
aduh95
committed
stream: drain pending writes before broadcast end
Keep processing writes that were queued before end(). Signal the broadcast end only after those writes enter the shared buffer. Resolve end() after all consumers reach end-of-stream. This prevents backpressured writes from remaining pending forever. Signed-off-by: Kamat, Trivikram <16024985+trivikr@users.noreply.github.com> Assisted-by: codex:gpt-5.6-sol PR-URL: #65334 Fixes: #65333 Reviewed-By: James M Snell <jasnell@gmail.com>
1 parent 5fbf6c5 commit dc51c79

4 files changed

Lines changed: 258 additions & 50 deletions

File tree

‎doc/api/stream_iter.md‎

Lines changed: 6 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -431,14 +431,16 @@ the write. Use [`ondrain()`][] to wait for capacity rather than polling.
431431
the pending `end()` call; it does not fail the writer itself.
432432
* Returns: {Promise} Fulfills with the total number of bytes written.
433433

434-
Signal that no more data will be written.
434+
Signals that no more data will be written and waits for buffered data to drain.
435435

436436
#### `writer.endSync()`
437437

438-
* Returns: {number} Total bytes written, or `-1` if the writer is not open.
438+
* Returns: {number} Total bytes written, or `-1` if ending cannot complete
439+
synchronously.
439440

440-
Synchronous variant of `writer.end()`. Returns `-1` if the writer is already
441-
closed or errored. Can be used as a try-fallback pattern:
441+
Synchronous variant of `writer.end()`. A return value of `-1` means closing has
442+
started but requires asynchronous draining. Use the try-fallback pattern to
443+
await completion:
442444

443445
```cjs
444446
const result = writer.endSync();

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

Lines changed: 92 additions & 34 deletions
Original file line numberDiff line numberDiff line change
@@ -14,6 +14,8 @@ const {
1414
PromiseReject,
1515
PromiseResolve,
1616
PromiseWithResolvers,
17+
SafePromisePrototypeFinally,
18+
SafePromiseRace,
1719
SafeSet,
1820
Symbol,
1921
SymbolAsyncDispose,
@@ -77,6 +79,22 @@ const kEnd = Symbol('kEnd');
7779
const kAbort = Symbol('kAbort');
7880
const kCanWrite = Symbol('kCanWrite');
7981
const kOnBufferDrained = Symbol('kOnBufferDrained');
82+
const kOnEndDrained = Symbol('kOnEndDrained');
83+
const kPendingWriteRemoved = Symbol('kPendingWriteRemoved');
84+
85+
function raceEndWithSignal(promise, signal) {
86+
if (!signal) return promise;
87+
88+
const { promise: aborted, reject } = PromiseWithResolvers();
89+
const onAbort = () => reject(signal.reason);
90+
signal.addEventListener('abort', onAbort, { __proto__: null, once: true });
91+
if (signal.aborted) onAbort();
92+
93+
return SafePromisePrototypeFinally(
94+
SafePromiseRace([promise, aborted]),
95+
() => signal.removeEventListener('abort', onAbort),
96+
);
97+
}
8098

8199
// =============================================================================
82100
// Broadcast Implementation
@@ -100,6 +118,7 @@ class BroadcastImpl {
100118
constructor(options) {
101119
this.#options = options;
102120
this[kOnBufferDrained] = null;
121+
this[kOnEndDrained] = null;
103122
}
104123

105124
setWriter(writer) {
@@ -183,6 +202,7 @@ class BroadcastImpl {
183202
if (self.#deleteConsumer(state)) {
184203
self.#tryTrimBuffer();
185204
}
205+
self.#notifyEndDrained();
186206
}
187207

188208
return {
@@ -358,10 +378,11 @@ class BroadcastImpl {
358378
}
359379
}
360380
}
381+
this.#notifyEndDrained();
361382
}
362383

363384
[kAbort](reason) {
364-
if (this.#ended || this.#error !== undefined) return;
385+
if (this.#error !== undefined) return;
365386
this.#error = reason;
366387
this.#ended = true;
367388

@@ -396,6 +417,12 @@ class BroadcastImpl {
396417

397418
// Private methods
398419

420+
#notifyEndDrained() {
421+
if (this.#ended && this.#consumers.size === 0) {
422+
this[kOnEndDrained]?.();
423+
}
424+
}
425+
399426
#recomputeMinCursor() {
400427
const { minCursor, minCursorConsumers } = getMinCursor(
401428
this.#consumers, this.#bufferStart + this.#buffer.length);
@@ -516,8 +543,9 @@ let getBroadcastPendingWrites;
516543
class BroadcastWriter {
517544
#broadcast;
518545
#totalBytes = 0;
519-
#closed;
520-
#aborted = false;
546+
#state = 'open';
547+
#error;
548+
#pendingEnd;
521549
#pendingWrites = new RingBuffer();
522550
#pendingDrains = [];
523551

@@ -532,8 +560,11 @@ class BroadcastWriter {
532560

533561
this.#broadcast[kOnBufferDrained] = () => {
534562
this.#resolvePendingWrites();
535-
this.#resolvePendingDrains(true);
563+
if (this.#state === 'open') {
564+
this.#resolvePendingDrains(true);
565+
}
536566
};
567+
this.#broadcast[kOnEndDrained] = () => this.#endDrained();
537568
}
538569

539570
// The drainable protocol works with Stream.ondrain to provide a notification
@@ -547,20 +578,12 @@ class BroadcastWriter {
547578
return promise;
548579
}
549580

550-
#isClosed() {
551-
return this.#closed !== undefined;
552-
}
553-
554-
#isClosedOrAborted() {
555-
return this.#isClosed() || this.#aborted;
556-
}
557-
558581
get canWrite() {
559-
return this.#isClosedOrAborted() ? null : this.#broadcast[kCanWrite]();
582+
return this.#state === 'open' ? this.#broadcast[kCanWrite]() : null;
560583
}
561584

562585
#canUseWriteFastPath(signal) {
563-
return !signal && !this.#isClosed() && !this.#aborted &&
586+
return !signal && this.#state === 'open' &&
564587
this.#broadcast[kCanWrite]();
565588
}
566589

@@ -592,13 +615,15 @@ class BroadcastWriter {
592615
}
593616

594617
async #writevSlow(chunks, signal) {
595-
// Check for pre-aborted
596-
signal?.throwIfAborted();
597-
598-
if (this.#isClosedOrAborted()) {
618+
if (this.#state === 'errored') {
619+
throw this.#error;
620+
}
621+
if (this.#state !== 'open') {
599622
throw new ERR_INVALID_STATE.TypeError('Writer is closed');
600623
}
601624

625+
signal?.throwIfAborted();
626+
602627
const converted = convertChunks(chunks);
603628

604629
if (this.#broadcast[kWrite](converted)) {
@@ -624,7 +649,7 @@ class BroadcastWriter {
624649
}
625650

626651
writeSync(chunk) {
627-
if (this.#isClosedOrAborted()) return false;
652+
if (this.#state !== 'open') return false;
628653
if (!this.#broadcast[kCanWrite]()) return false;
629654
const converted =
630655
toUint8Array(chunk);
@@ -637,7 +662,7 @@ class BroadcastWriter {
637662

638663
writevSync(chunks) {
639664
validateArray(chunks, 'chunks');
640-
if (this.#isClosedOrAborted()) return false;
665+
if (this.#state !== 'open') return false;
641666
if (!this.#broadcast[kCanWrite]()) return false;
642667
const converted = convertChunks(chunks);
643668
if (this.#broadcast[kWrite](converted)) {
@@ -651,34 +676,43 @@ class BroadcastWriter {
651676

652677
end(options) {
653678
const signal = getWriterSignal(options);
679+
if (this.#state === 'errored') return PromiseReject(this.#error);
680+
if (this.#state === 'closed') return PromiseResolve(this.#totalBytes);
654681
if (signal?.aborted) return PromiseReject(signal.reason);
655682

656-
if (this.#isClosed()) return this.#closed;
657-
this.#closed = PromiseResolve(this.#totalBytes);
658-
this.#broadcast[kEnd]();
659-
this.#resolvePendingDrains(false);
660-
return this.#closed;
683+
const endPromise = this.#getEndPromise();
684+
if (this.#state === 'open') {
685+
this.#state = 'closing';
686+
this.#resolvePendingDrains(false);
687+
this.#finishEndIfReady();
688+
}
689+
690+
return raceEndWithSignal(endPromise, signal);
661691
}
662692

663693
endSync() {
664-
if (this.#closed) return this.#totalBytes;
665-
this.#closed = PromiseResolve(this.#totalBytes);
666-
this.#broadcast[kEnd]();
694+
if (this.#state === 'closed') return this.#totalBytes;
695+
if (this.#state === 'errored' || this.#state === 'closing') return -1;
696+
697+
this.#state = 'closing';
667698
this.#resolvePendingDrains(false);
668-
return this.#totalBytes;
699+
this.#finishEndIfReady();
700+
return this.#state === 'closed' ? this.#totalBytes : -1;
669701
}
670702

671703
fail(reason) {
672-
if (this.#isClosedOrAborted()) return;
673-
this.#aborted = true;
674-
this.#closed = PromiseResolve(this.#totalBytes);
704+
if (this.#state === 'errored' || this.#state === 'closed') return;
705+
this.#state = 'errored';
675706
const error = reason ?? new ERR_INVALID_STATE.TypeError('Failed');
707+
this.#error = error;
676708
this.#rejectPendingWrites(error);
677709
this.#rejectPendingDrains(error);
710+
this.#pendingEnd?.reject(error);
678711
this.#broadcast[kAbort](error);
679712
}
680713

681714
[SymbolAsyncDispose]() {
715+
if (this.#state === 'closing') return this.#getEndPromise();
682716
this.fail();
683717
return PromiseResolve();
684718
}
@@ -688,11 +722,33 @@ class BroadcastWriter {
688722
}
689723

690724
[kCancelWriter]() {
691-
if (this.#isClosed()) return;
692-
this.#closed = PromiseResolve(this.#totalBytes);
725+
if (this.#state === 'closed' || this.#state === 'errored') return;
726+
this.#state = 'closed';
693727
this.#rejectPendingWrites(
694728
lazyDOMException('Broadcast cancelled', 'AbortError'));
695729
this.#resolvePendingDrains(false);
730+
this.#pendingEnd?.resolve(this.#totalBytes);
731+
}
732+
733+
#getEndPromise() {
734+
this.#pendingEnd ??= PromiseWithResolvers();
735+
return this.#pendingEnd.promise;
736+
}
737+
738+
#finishEndIfReady() {
739+
if (this.#state === 'closing' && this.#pendingWrites.length === 0) {
740+
this.#broadcast[kEnd]();
741+
}
742+
}
743+
744+
#endDrained() {
745+
if (this.#state !== 'closing') return;
746+
this.#state = 'closed';
747+
this.#pendingEnd?.resolve(this.#totalBytes);
748+
}
749+
750+
[kPendingWriteRemoved]() {
751+
this.#finishEndIfReady();
696752
}
697753

698754
/**
@@ -724,6 +780,7 @@ class BroadcastWriter {
724780
break;
725781
}
726782
}
783+
this.#finishEndIfReady();
727784
}
728785

729786
#rejectPendingWrites(error) {
@@ -756,6 +813,7 @@ function wireBroadcastWriteSignal(entry, signal, resolve, reject, self) {
756813
if (idx !== -1) pendingWrites.removeAt(idx);
757814
entry.chunk = null;
758815
reject(signal.reason ?? lazyDOMException('Aborted', 'AbortError'));
816+
if (idx !== -1) self[kPendingWriteRemoved]();
759817
};
760818
entry.resolve = function() {
761819
signal.removeEventListener('abort', onAbort);

0 commit comments

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