stream: cache minimum cursor count in share

Track the number of consumers at the cached minimum cursor in share()
so the minimum is only recomputed when the last consumer at that cursor
advances or detaches.
This avoids scanning every consumer on each trim attempt when multiple
consumers advance through a shared buffer.
Signed-off-by: Kamat, Trivikram <16024985+trivikr@users.noreply.github.com>
Assisted-by: openai:gpt-5.5
PR-URL: #63262
Reviewed-By: James M Snell <jasnell@gmail.com>
Reviewed-By: René <contact.9a5d6388@renegade334.me.uk>

Author: trivikr

benchmark/streams/iter-throughput-share.js

@@ -0,0 +1,32 @@
+'use strict';
+
+constcommon=require('../common.js');
+
+constbench=common.createBenchmark(main,{
+consumers: [2,8,32],
+batches: [1e4],
+backpressure: ['block'],
+n: [5],
+},{
+flags: ['--experimental-stream-iter'],
+});
+
+asyncfunctionmain({ consumers, batches, backpressure, n }){
+const{ share, array }=require('stream/iter');
+constchunk=Buffer.alloc(1024);
+consttotalOps=batches*consumers*n;
+
+asyncfunction*source(){
+for(leti=0;i<batches;i++){
+yield[chunk];
+}
+}
+
+bench.start();
+for(leti=0;i<n;i++){
+constshared=share(source(),{highWaterMark: 64, backpressure });
+constreaders=Array.from({length: consumers},()=>array(shared.pull()));
+awaitPromise.all(readers);
+}
+bench.end(totalOps);
+}

lib/internal/streams/iter/broadcast.js

@@ -343,8 +343,9 @@ class BroadcastImpl {
 // Private methods
 
  #recomputeMinCursor(){
-this.#cachedMinCursor=getMinCursor(
+const{ minCursor }=getMinCursor(
 this.#consumers,this.#bufferStart +this.#buffer.length);
+this.#cachedMinCursor =minCursor;
 this.#minCursorDirty =false;
 }
 

lib/internal/streams/iter/share.js

@@ -77,6 +77,8 @@ class ShareImpl {
  #cancelled =false;
  #pulling =false;
  #pullWaiters =[];
+ #cachedMinCursor =0;
+ #cachedMinCursorConsumers =0;
 
 constructor(source,options){
 this.#source =source;
@@ -114,6 +116,14 @@ class ShareImpl {
 };
 
 this.#consumers.add(state);
+if(this.#consumers.size===1){
+this.#cachedMinCursor =state.cursor;
+this.#cachedMinCursorConsumers =1;
+}elseif(state.cursor===this.#cachedMinCursor){
+this.#cachedMinCursorConsumers++;
+}else{
+this.#recomputeMinCursor();
+}
 constself=this;
 
 return{
@@ -139,22 +149,26 @@ class ShareImpl {
 
 if(self.#cancelled){
 state.detached=true;
-self.#consumers.delete(state);
+self.#deleteConsumer(state);
 return{__proto__: null,done: true,value: undefined};
 }
 
 // Check if data is available in buffer
 constbufferIndex=state.cursor-self.#bufferStart;
 if(bufferIndex<self.#buffer.length){
 constchunk=self.#buffer.get(bufferIndex);
+constcursor=state.cursor;
 state.cursor++;
-self.#tryTrimBuffer();
+if(cursor===self.#cachedMinCursor &&
+--self.#cachedMinCursorConsumers ===0){
+self.#tryTrimBuffer();
+}
 return{__proto__: null,done: false,value: chunk};
 }
 
 if(self.#sourceExhausted){
 state.detached=true;
-self.#consumers.delete(state);
+self.#deleteConsumer(state);
 if(self.#sourceError)throwself.#sourceError;
 return{__proto__: null,done: true,value: undefined};
 }
@@ -163,7 +177,7 @@ class ShareImpl {
 constcanPull=awaitself.#waitForBufferSpace();
 if(!canPull){
 state.detached=true;
-self.#consumers.delete(state);
+self.#deleteConsumer(state);
 if(self.#sourceError)throwself.#sourceError;
 return{__proto__: null,done: true,value: undefined};
 }
@@ -176,17 +190,19 @@ class ShareImpl {
 state.detached=true;
 state.resolve=null;
 state.reject=null;
-self.#consumers.delete(state);
-self.#tryTrimBuffer();
+if(self.#deleteConsumer(state)){
+self.#tryTrimBuffer();
+}
 return{__proto__: null,done: true,value: undefined};
 },
 
 asyncthrow(){
 state.detached=true;
 state.resolve=null;
 state.reject=null;
-self.#consumers.delete(state);
-self.#tryTrimBuffer();
+if(self.#deleteConsumer(state)){
+self.#tryTrimBuffer();
+}
 return{__proto__: null,done: true,value: undefined};
 },
 };
@@ -254,9 +270,11 @@ class ShareImpl {
 this.#bufferStart++;
 for(constconsumerofthis.#consumers){
 if(consumer.cursor<this.#bufferStart){
+this.#deleteConsumerFromMin(consumer);
 consumer.cursor=this.#bufferStart;
 }
 }
+this.#recomputeMinCursor();
 returntrue;
 case'drop-newest':
 returntrue;
@@ -324,18 +342,41 @@ class ShareImpl {
 }
 
  #tryTrimBuffer(){
-constminCursor=getMinCursor(
-this.#consumers,this.#bufferStart +this.#buffer.length);
-consttrimCount=minCursor-this.#bufferStart;
+if(this.#cachedMinCursorConsumers ===0){
+this.#recomputeMinCursor();
+}
+consttrimCount=this.#cachedMinCursor -this.#bufferStart;
 if(trimCount>0){
 this.#buffer.trimFront(trimCount);
-this.#bufferStart =minCursor;
+this.#bufferStart =this.#cachedMinCursor;
 for(leti=0;i<this.#pullWaiters.length;i++){
 this.#pullWaiters[i]();
 }
 this.#pullWaiters =[];
 }
 }
+
+ #recomputeMinCursor(){
+const{ minCursor, minCursorConsumers }=getMinCursor(
+this.#consumers,this.#bufferStart +this.#buffer.length);
+this.#cachedMinCursor =minCursor;
+this.#cachedMinCursorConsumers =minCursorConsumers;
+}
+
+ #deleteConsumerFromMin(consumer){
+if(consumer.cursor===this.#cachedMinCursor){
+this.#cachedMinCursorConsumers--;
+returnthis.#cachedMinCursorConsumers ===0;
+}
+returnfalse;
+}
+
+ #deleteConsumer(consumer){
+if(this.#consumers.delete(consumer)){
+returnthis.#deleteConsumerFromMin(consumer);
+}
+returnfalse;
+}
 }
 
 // =============================================================================
@@ -352,6 +393,8 @@ class SyncShareImpl {
  #sourceExhausted =false;
  #sourceError =null;
  #cancelled =false;
+ #cachedMinCursor =0;
+ #cachedMinCursorConsumers =0;
 
 constructor(source,options){
 this.#source =source;
@@ -383,6 +426,14 @@ class SyncShareImpl {
 };
 
 this.#consumers.add(state);
+if(this.#consumers.size===1){
+this.#cachedMinCursor =state.cursor;
+this.#cachedMinCursorConsumers =1;
+}elseif(state.cursor===this.#cachedMinCursor){
+this.#cachedMinCursorConsumers++;
+}else{
+this.#recomputeMinCursor();
+}
 constself=this;
 
 return{
@@ -396,26 +447,30 @@ class SyncShareImpl {
 }
 if(self.#sourceError){
 state.detached=true;
-self.#consumers.delete(state);
+self.#deleteConsumer(state);
 throwself.#sourceError;
 }
 if(self.#cancelled){
 state.detached=true;
-self.#consumers.delete(state);
+self.#deleteConsumer(state);
 return{__proto__: null,done: true,value: undefined};
 }
 
 constbufferIndex=state.cursor-self.#bufferStart;
 if(bufferIndex<self.#buffer.length){
 constchunk=self.#buffer.get(bufferIndex);
+constcursor=state.cursor;
 state.cursor++;
-self.#tryTrimBuffer();
+if(cursor===self.#cachedMinCursor &&
+--self.#cachedMinCursorConsumers ===0){
+self.#tryTrimBuffer();
+}
 return{__proto__: null,done: false,value: chunk};
 }
 
 if(self.#sourceExhausted){
 state.detached=true;
-self.#consumers.delete(state);
+self.#deleteConsumer(state);
 return{__proto__: null,done: true,value: undefined};
 }
 
@@ -436,13 +491,15 @@ class SyncShareImpl {
 self.#bufferStart++;
 for(constconsumerofself.#consumers){
 if(consumer.cursor<self.#bufferStart){
+self.#deleteConsumerFromMin(consumer);
 consumer.cursor=self.#bufferStart;
 }
 }
+self.#recomputeMinCursor();
 break;
 case'drop-newest':
 state.detached=true;
-self.#consumers.delete(state);
+self.#deleteConsumer(state);
 return{__proto__: null,done: true,value: undefined};
 }
 }
@@ -451,21 +508,25 @@ class SyncShareImpl {
 
 if(self.#sourceError){
 state.detached=true;
-self.#consumers.delete(state);
+self.#deleteConsumer(state);
 throwself.#sourceError;
 }
 
 constnewBufferIndex=state.cursor-self.#bufferStart;
 if(newBufferIndex<self.#buffer.length){
 constchunk=self.#buffer.get(newBufferIndex);
+constcursor=state.cursor;
 state.cursor++;
-self.#tryTrimBuffer();
+if(cursor===self.#cachedMinCursor &&
+--self.#cachedMinCursorConsumers ===0){
+self.#tryTrimBuffer();
+}
 return{__proto__: null,done: false,value: chunk};
 }
 
 if(self.#sourceExhausted){
 state.detached=true;
-self.#consumers.delete(state);
+self.#deleteConsumer(state);
 return{__proto__: null,done: true,value: undefined};
 }
 
@@ -474,15 +535,17 @@ class SyncShareImpl {
 
 return(){
 state.detached=true;
-self.#consumers.delete(state);
-self.#tryTrimBuffer();
+if(self.#deleteConsumer(state)){
+self.#tryTrimBuffer();
+}
 return{__proto__: null,done: true,value: undefined};
 },
 
 throw(){
 state.detached=true;
-self.#consumers.delete(state);
-self.#tryTrimBuffer();
+if(self.#deleteConsumer(state)){
+self.#tryTrimBuffer();
+}
 return{__proto__: null,done: true,value: undefined};
 },
 };
@@ -532,13 +595,36 @@ class SyncShareImpl {
 }
 
  #tryTrimBuffer(){
-constminCursor=getMinCursor(
-this.#consumers,this.#bufferStart +this.#buffer.length);
-consttrimCount=minCursor-this.#bufferStart;
+if(this.#cachedMinCursorConsumers ===0){
+this.#recomputeMinCursor();
+}
+consttrimCount=this.#cachedMinCursor -this.#bufferStart;
 if(trimCount>0){
 this.#buffer.trimFront(trimCount);
-this.#bufferStart =minCursor;
+this.#bufferStart =this.#cachedMinCursor;
+}
+}
+
+ #recomputeMinCursor(){
+const{ minCursor, minCursorConsumers }=getMinCursor(
+this.#consumers,this.#bufferStart +this.#buffer.length);
+this.#cachedMinCursor =minCursor;
+this.#cachedMinCursorConsumers =minCursorConsumers;
+}
+
+ #deleteConsumerFromMin(consumer){
+if(consumer.cursor===this.#cachedMinCursor){
+this.#cachedMinCursorConsumers--;
+returnthis.#cachedMinCursorConsumers ===0;
+}
+returnfalse;
+}
+
+ #deleteConsumer(consumer){
+if(this.#consumers.delete(consumer)){
+returnthis.#deleteConsumerFromMin(consumer);
 }
+returnfalse;
 }
 }
 

lib/internal/streams/iter/utils.js

@@ -70,20 +70,24 @@ function onSignalAbort(signal, handler) {
 }
 
 /**
- * Compute the minimum cursor across a set of consumers.
- * Returns fallback if the set is empty.
+ * Compute the minimum cursor across a set of consumers and count how many
+ * consumers are at that cursor.
  * @param {Set} consumers - Set of objects with a `cursor` property
- * @param {number} fallback - Value to return when set is empty
- * @returns {number}
+ * @param {number} fallback - Cursor to return when set is empty
+ * @returns {{ minCursor: number, minCursorConsumers: number }}
  */
 functiongetMinCursor(consumers,fallback){
-letmin=Infinity;
+letminCursor=fallback;
+letminCursorConsumers=0;
 for(constconsumerofconsumers){
-if(consumer.cursor<min){
-min=consumer.cursor;
+if(consumer.cursor<minCursor){
+minCursor=consumer.cursor;
+minCursorConsumers=1;
+}elseif(consumer.cursor===minCursor){
+minCursorConsumers++;
 }
 }
-returnmin===Infinity ? fallback : min;
+return{__proto__: null, minCursor, minCursorConsumers };
 }
 
 /**