stream: cache minimum cursor counts

Cache the minimum consumer cursor and track how many consumers are
currently positioned at that cursor. This avoids scanning every consumer
on each trim attempt when multiple consumers advance through shared
buffers.

Update share() and broadcast() to recompute the minimum cursor only when
the last consumer at the cached minimum advances or detaches.

Signed-off-by: Kamat, Trivikram <16024985+trivikr@users.noreply.github.com>
Assisted-by: openai:gpt-5.5

Author: trivikr

benchmark/streams/iter-throughput-share.js

@@ -0,0 +1,32 @@
+'use strict';
+
+const common = require('../common.js');
+
+const bench = common.createBenchmark(main, {
+  consumers: [8, 32, 128],
+  batches: [1e4],
+  backpressure: ['block'],
+  n: [5],
+}, {
+  flags: ['--experimental-stream-iter'],
+});
+
+async function main({ consumers, batches, backpressure, n }) {
+  const { share, array } = require('stream/iter');
+  const chunk = Buffer.alloc(1024);
+  const totalOps = batches * consumers * n;
+
+  async function* source() {
+    for (let i = 0; i < batches; i++) {
+      yield [chunk];
+    }
+  }
+
+  bench.start();
+  for (let i = 0; i < n; i++) {
+    const shared = share(source(), { highWaterMark: 64, backpressure });
+    const readers = Array.from({ length: consumers }, () => array(shared.pull()));
+    await Promise.all(readers);
+  }
+  bench.end(totalOps);
+}

lib/internal/streams/iter/broadcast.js

@@ -92,7 +92,7 @@ class BroadcastImpl {
   #options;
   #writer = null;
   #cachedMinCursor = 0;
-  #minCursorDirty = false;
+  #cachedMinCursorConsumers = 0;
 
   constructor(options) {
     this.#options = options;
@@ -154,9 +154,11 @@ class BroadcastImpl {
     // since this consumer may now be the slowest.
     if (this.#consumers.size === 1) {
       this.#cachedMinCursor = state.cursor;
-      this.#minCursorDirty = false;
+      this.#cachedMinCursorConsumers = 1;
+    } else if (state.cursor === this.#cachedMinCursor) {
+      this.#cachedMinCursorConsumers++;
     } else {
-      this.#minCursorDirty = true;
+      this.#recomputeMinCursor();
     }
     const self = this;
 
@@ -167,9 +169,9 @@ class BroadcastImpl {
       state.detached = true;
       state.resolve = null;
       state.reject = null;
-      self.#consumers.delete(state);
-      self.#minCursorDirty = true;
-      self.#tryTrimBuffer();
+      if (self.#deleteConsumer(state)) {
+        self.#tryTrimBuffer();
+      }
     }
 
     return {
@@ -186,19 +188,21 @@ class BroadcastImpl {
             const bufferIndex = state.cursor - self.#bufferStart;
             if (bufferIndex < self.#buffer.length) {
               const chunk = self.#buffer.get(bufferIndex);
-              // If this consumer was at the min cursor, mark dirty
-              if (state.cursor <= self.#cachedMinCursor) {
-                self.#minCursorDirty = true;
-              }
+              const cursor = state.cursor;
               state.cursor++;
-              self.#tryTrimBuffer();
+              if (cursor === self.#cachedMinCursor &&
+                  --self.#cachedMinCursorConsumers === 0) {
+                self.#tryTrimBuffer();
+              }
               return PromiseResolve(
                 { __proto__: null, done: false, value: chunk });
             }
 
             if (self.#error) {
               state.detached = true;
-              self.#consumers.delete(state);
+              if (self.#deleteConsumer(state)) {
+                self.#tryTrimBuffer();
+              }
               return PromiseReject(self.#error);
             }
 
@@ -274,9 +278,11 @@ class BroadcastImpl {
           this.#bufferStart++;
           for (const consumer of this.#consumers) {
             if (consumer.cursor < this.#bufferStart) {
+              this.#deleteConsumerFromMin(consumer);
               consumer.cursor = this.#bufferStart;
             }
           }
+          this.#recomputeMinCursor();
           break;
         case 'drop-newest':
           return true;
@@ -297,6 +303,7 @@ class BroadcastImpl {
         const bufferIndex = consumer.cursor - this.#bufferStart;
         if (bufferIndex < this.#buffer.length) {
           const chunk = this.#buffer.get(bufferIndex);
+          this.#advanceConsumer(consumer);
           consumer.cursor++;
           consumer.resolve({ __proto__: null, done: false, value: chunk });
         } else {
@@ -306,6 +313,9 @@ class BroadcastImpl {
         consumer.reject = null;
       }
     }
+    if (this.#consumers.size > 0 && this.#cachedMinCursorConsumers === 0) {
+      this.#tryTrimBuffer();
+    }
   }
 
   [kAbort](reason) {
@@ -343,13 +353,14 @@ class BroadcastImpl {
   // Private methods
 
   #recomputeMinCursor() {
-    this.#cachedMinCursor = getMinCursor(
+    const [minCursor, minCursorConsumers] = getMinCursor(
       this.#consumers, this.#bufferStart + this.#buffer.length);
-    this.#minCursorDirty = false;
+    this.#cachedMinCursor = minCursor;
+    this.#cachedMinCursorConsumers = minCursorConsumers;
   }
 
   #tryTrimBuffer() {
-    if (this.#minCursorDirty) {
+    if (this.#cachedMinCursorConsumers === 0) {
       this.#recomputeMinCursor();
     }
     const trimCount = this.#cachedMinCursor - this.#bufferStart;
@@ -364,6 +375,26 @@ class BroadcastImpl {
     }
   }
 
+  #advanceConsumer(consumer) {
+    if (consumer.cursor === this.#cachedMinCursor) {
+      this.#cachedMinCursorConsumers--;
+    }
+  }
+
+  #deleteConsumerFromMin(consumer) {
+    if (consumer.cursor === this.#cachedMinCursor) {
+      this.#cachedMinCursorConsumers--;
+    }
+  }
+
+  #deleteConsumer(consumer) {
+    if (this.#consumers.delete(consumer)) {
+      this.#deleteConsumerFromMin(consumer);
+      return this.#cachedMinCursorConsumers === 0;
+    }
+    return false;
+  }
+
   #notifyConsumers() {
     const waiters = this.#waiters;
     if (waiters.length === 0) return;
@@ -376,9 +407,7 @@ class BroadcastImpl {
         const bufferIndex = consumer.cursor - this.#bufferStart;
         if (bufferIndex < this.#buffer.length) {
           const chunk = this.#buffer.get(bufferIndex);
-          if (consumer.cursor <= this.#cachedMinCursor) {
-            this.#minCursorDirty = true;
-          }
+          this.#advanceConsumer(consumer);
           consumer.cursor++;
           const resolve = consumer.resolve;
           consumer.resolve = null;
@@ -390,6 +419,9 @@ class BroadcastImpl {
         }
       }
     }
+    if (this.#cachedMinCursorConsumers === 0) {
+      this.#tryTrimBuffer();
+    }
   }
 }