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

stream: cache minimum cursor count in broadcast · nodejs/node@3bdb64d · GitHub

/ node Public

Commit 3bdb64d

Browse files
trivikr
authored and
aduh95
committed
stream: cache minimum cursor count in broadcast
Track how many broadcast consumers share the cached minimum cursor, matching the share implementation. This lets buffer trimming avoid calling getMinCursor() until the last minimum-cursor consumer advances or detaches. Add coverage for fan-out trimming behavior. Signed-off-by: Kamat, Trivikram <16024985+trivikr@users.noreply.github.com> Assisted-by: openai:gpt-5.5 PR-URL: #63322 Reviewed-By: James M Snell <jasnell@gmail.com> Reviewed-By: Ethan Arrowood <ethan@arrowood.dev> Reviewed-By: Gürgün Dayıoğlu <hey@gurgun.day>
1 parent 81819ad commit 3bdb64d

2 files changed

Lines changed: 74 additions & 20 deletions

File tree

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

Lines changed: 46 additions & 20 deletions
Original file line numberDiff line numberDiff line change
@@ -92,7 +92,7 @@ class BroadcastImpl {
9292
#options;
9393
#writer = null;
9494
#cachedMinCursor = 0;
95-
#minCursorDirty = false;
95+
#cachedMinCursorConsumers = 0;
9696

9797
constructor(options) {
9898
this.#options = options;
@@ -150,13 +150,13 @@ class BroadcastImpl {
150150
};
151151

152152
this.#consumers.add(state);
153-
// New consumer starts at buffer start; recalculate min cursor
154-
// since this consumer may now be the slowest.
155153
if (this.#consumers.size === 1) {
156154
this.#cachedMinCursor = state.cursor;
157-
this.#minCursorDirty = false;
155+
this.#cachedMinCursorConsumers = 1;
156+
} else if (state.cursor === this.#cachedMinCursor) {
157+
this.#cachedMinCursorConsumers++;
158158
} else {
159-
this.#minCursorDirty = true;
159+
this.#recomputeMinCursor();
160160
}
161161
const self = this;
162162

@@ -167,9 +167,9 @@ class BroadcastImpl {
167167
state.detached = true;
168168
state.resolve = null;
169169
state.reject = null;
170-
self.#consumers.delete(state);
171-
self.#minCursorDirty = true;
172-
self.#tryTrimBuffer();
170+
if (self.#deleteConsumer(state)) {
171+
self.#tryTrimBuffer();
172+
}
173173
}
174174

175175
return {
@@ -186,19 +186,19 @@ class BroadcastImpl {
186186
const bufferIndex = state.cursor - self.#bufferStart;
187187
if (bufferIndex < self.#buffer.length) {
188188
const chunk = self.#buffer.get(bufferIndex);
189-
// If this consumer was at the min cursor, mark dirty
190-
if (state.cursor <= self.#cachedMinCursor) {
191-
self.#minCursorDirty = true;
192-
}
189+
const cursor = state.cursor;
193190
state.cursor++;
194-
self.#tryTrimBuffer();
191+
if (cursor === self.#cachedMinCursor &&
192+
--self.#cachedMinCursorConsumers === 0) {
193+
self.#tryTrimBuffer();
194+
}
195195
return PromiseResolve(
196196
{ __proto__: null, done: false, value: chunk });
197197
}
198198

199199
if (self.#error) {
200200
state.detached = true;
201-
self.#consumers.delete(state);
201+
self.#deleteConsumer(state);
202202
return PromiseReject(self.#error);
203203
}
204204

@@ -253,6 +253,7 @@ class BroadcastImpl {
253253
consumer.detached = true;
254254
}
255255
this.#consumers.clear();
256+
this.#cachedMinCursorConsumers = 0;
256257
}
257258

258259
[SymbolDispose]() {
@@ -274,9 +275,11 @@ class BroadcastImpl {
274275
this.#bufferStart++;
275276
for (const consumer of this.#consumers) {
276277
if (consumer.cursor < this.#bufferStart) {
278+
this.#deleteConsumerFromMin(consumer);
277279
consumer.cursor = this.#bufferStart;
278280
}
279281
}
282+
this.#recomputeMinCursor();
280283
break;
281284
case 'drop-newest':
282285
return true;
@@ -297,7 +300,12 @@ class BroadcastImpl {
297300
const bufferIndex = consumer.cursor - this.#bufferStart;
298301
if (bufferIndex < this.#buffer.length) {
299302
const chunk = this.#buffer.get(bufferIndex);
303+
const cursor = consumer.cursor;
300304
consumer.cursor++;
305+
if (cursor === this.#cachedMinCursor &&
306+
--this.#cachedMinCursorConsumers === 0) {
307+
this.#tryTrimBuffer();
308+
}
301309
consumer.resolve({ __proto__: null, done: false, value: chunk });
302310
} else {
303311
consumer.resolve({ __proto__: null, done: true, value: undefined });
@@ -323,6 +331,7 @@ class BroadcastImpl {
323331
consumer.detached = true;
324332
}
325333
this.#consumers.clear();
334+
this.#cachedMinCursorConsumers = 0;
326335
}
327336

328337
[kGetDesiredSize]() {
@@ -343,14 +352,14 @@ class BroadcastImpl {
343352
// Private methods
344353

345354
#recomputeMinCursor() {
346-
const { minCursor } = getMinCursor(
355+
const { minCursor, minCursorConsumers } = getMinCursor(
347356
this.#consumers, this.#bufferStart + this.#buffer.length);
348357
this.#cachedMinCursor = minCursor;
349-
this.#minCursorDirty = false;
358+
this.#cachedMinCursorConsumers = minCursorConsumers;
350359
}
351360

352361
#tryTrimBuffer() {
353-
if (this.#minCursorDirty) {
362+
if (this.#cachedMinCursorConsumers === 0) {
354363
this.#recomputeMinCursor();
355364
}
356365
const trimCount = this.#cachedMinCursor - this.#bufferStart;
@@ -377,10 +386,12 @@ class BroadcastImpl {
377386
const bufferIndex = consumer.cursor - this.#bufferStart;
378387
if (bufferIndex < this.#buffer.length) {
379388
const chunk = this.#buffer.get(bufferIndex);
380-
if (consumer.cursor <= this.#cachedMinCursor) {
381-
this.#minCursorDirty = true;
382-
}
389+
const cursor = consumer.cursor;
383390
consumer.cursor++;
391+
if (cursor === this.#cachedMinCursor &&
392+
--this.#cachedMinCursorConsumers === 0) {
393+
this.#tryTrimBuffer();
394+
}
384395
const resolve = consumer.resolve;
385396
consumer.resolve = null;
386397
consumer.reject = null;
@@ -392,6 +403,21 @@ class BroadcastImpl {
392403
}
393404
}
394405
}
406+
407+
#deleteConsumerFromMin(consumer) {
408+
if (consumer.cursor === this.#cachedMinCursor) {
409+
this.#cachedMinCursorConsumers--;
410+
return this.#cachedMinCursorConsumers === 0;
411+
}
412+
return false;
413+
}
414+
415+
#deleteConsumer(consumer) {
416+
if (this.#consumers.delete(consumer)) {
417+
return this.#deleteConsumerFromMin(consumer);
418+
}
419+
return false;
420+
}
395421
}
396422

397423
// =============================================================================

‎test/parallel/test-stream-iter-broadcast-coverage.js‎

Lines changed: 28 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -96,6 +96,33 @@ async function testRingbufferGrow() {
9696
}
9797
}
9898

99+
// Multiple consumers at the minimum cursor should trim only after the last
100+
// one advances or detaches.
101+
async function testFanOutMinCursorTrimming() {
102+
const { writer, broadcast: bc } = broadcast({ highWaterMark: 4 });
103+
const iter1 = bc.push()[Symbol.asyncIterator]();
104+
const iter2 = bc.push()[Symbol.asyncIterator]();
105+
106+
writer.writeSync(new Uint8Array([1]));
107+
writer.writeSync(new Uint8Array([2]));
108+
assert.strictEqual(bc.bufferSize, 2);
109+
110+
assert.strictEqual((await iter1.next()).done, false);
111+
assert.strictEqual(bc.bufferSize, 2);
112+
113+
assert.strictEqual((await iter2.next()).done, false);
114+
assert.strictEqual(bc.bufferSize, 1);
115+
116+
await iter1.return();
117+
assert.strictEqual(bc.bufferSize, 1);
118+
119+
assert.strictEqual((await iter2.next()).done, false);
120+
assert.strictEqual(bc.bufferSize, 0);
121+
122+
writer.endSync();
123+
assert.strictEqual((await iter2.next()).done, true);
124+
}
125+
99126
// Broadcast drainableProtocol after close returns null
100127
async function testDrainableAfterClose() {
101128
const { drainableProtocol } = require('stream/iter');
@@ -111,5 +138,6 @@ Promise.all([
111138
testBroadcastFromSyncIterable(),
112139
testBroadcastFromSyncIterableStrings(),
113140
testRingbufferGrow(),
141+
testFanOutMinCursorTrimming(),
114142
testDrainableAfterClose(),
115143
]).then(common.mustCall());

0 commit comments

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