stream: keep stream/iter broadcast waiters in reused lists

Broadcast notified waiting consumers by swapping #waiters for a new
SafeSet on every notification, so that a consumer re-waiting while being
notified was not processed twice. That happens for every chunk that
consumers wait for, and each Set operation allocates: the new Set, the
for...of iterator, and clear() or deleting the last entries (120-150
bytes each).

The waiters are now kept in two RingBuffers that are swapped back and
forth; the notified list is drained with shift() and cleared, keeping its
storage. A nested notification while the spare list is in use gets a new
one. A consumer detaching removes itself with indexOf()/removeAt(), as
it did with delete(); it is in the list at most once, as before.

Broadcast.from() fan-out, async source, two consumers, 16-byte chunks:
2036 -> 1873 bytes allocated per chunk.

Assisted-by: OpenCode

Author: jasnell

lib/internal/streams/iter/broadcast.js

@@ -124,7 +124,12 @@ class BroadcastImpl {
   #buffer = new RingBuffer();
   #bufferStart = 0;
   #consumers = new SafeSet();
-  #waiters = new SafeSet();  // Consumers with pending resolve
+  // Consumers with a pending resolve, each at most once. A RingBuffer
+  // rather than a Set: adding, clearing and iterating a Set allocate, and
+  // this changes for every chunk that consumers wait for.
+  #waiters = new RingBuffer(4);
+  // The list #notifyConsumers() swaps in for #waiters (see there).
+  #spareWaiters = new RingBuffer(4);
   #ended = false;
   #error;
   #errored = false;
@@ -224,7 +229,7 @@ class BroadcastImpl {
 
     function detach() {
       state.detached = true;
-      self.#waiters.delete(state);
+      self.#deleteWaiter(state);
       if (state.resolve) {
         state.resolve(new IterResult(true, undefined));
       }
@@ -286,7 +291,7 @@ class BroadcastImpl {
             const reject = takeReject();
             state.resolve = resolve;
             state.reject = reject;
-            self.#waiters.add(state);
+            self.#waiters.push(state);
             return promise;
           },
 
@@ -526,11 +531,30 @@ class BroadcastImpl {
 
   #notifyConsumers() {
     const waiters = this.#waiters;
-    if (waiters.size === 0) return;
+    if (waiters.length === 0) return;
     // Swap out the waiters list so consumers that re-wait during
-    // resolve don't get processed twice in this cycle.
-    this.#waiters = new SafeSet();
-    for (const consumer of waiters) {
+    // resolve don't get processed twice in this cycle. This runs for every
+    // write that consumers are waiting for, so the two lists are swapped
+    // back and forth rather than a new one created each time; a nested
+    // call, while the spare list is in use, gets a new one.
+    this.#waiters = this.#spareWaiters ?? new RingBuffer(4);
+    this.#spareWaiters = null;
+    try {
+      this.#notifyWaiters(waiters);
+    } finally {
+      waiters.clear();
+      this.#spareWaiters = waiters;
+    }
+  }
+
+  #deleteWaiter(consumer) {
+    const index = this.#waiters.indexOf(consumer);
+    if (index !== -1) this.#waiters.removeAt(index);
+  }
+
+  #notifyWaiters(waiters) {
+    let consumer;
+    while ((consumer = waiters.shift()) !== undefined) {
       if (consumer.resolve) {
         const bufferIndex = consumer.cursor - this.#bufferStart;
         if (bufferIndex < this.#buffer.length) {
@@ -549,11 +573,11 @@ class BroadcastImpl {
           if (consumer.detached && this.#deleteConsumer(consumer)) {
             this.#tryTrimBuffer();
           } else if (this.#promotePending(consumer)) {
-            this.#waiters.add(consumer);
+            this.#waiters.push(consumer);
           }
         } else {
           // Still waiting -- put back
-          this.#waiters.add(consumer);
+          this.#waiters.push(consumer);
         }
       }
     }

test/parallel/test-stream-iter-broadcast-waiter-cleanup.js

@@ -31,4 +31,55 @@ async function testDetachedWaitersAreReleased() {
   shared.cancel();
 }
 
-testDetachedWaitersAreReleased().then(common.mustCall());
+// Consumers waiting for writes are notified once per write, across many
+// writes: some wait with one next() at a time, one queues several next()
+// calls, and one detaches while it is waiting. Every remaining consumer
+// gets every chunk, in order.
+async function testWaitersAcrossWrites() {
+  // A 4-byte budget, so that the writer also waits for the consumers.
+  const { writer, broadcast: shared } = broadcast({ budget: 4 });
+  const n = 200;
+  const readAll = async (iterator) => {
+    const seen = [];
+    for (;;) {
+      const { done, value } = await iterator.next();
+      if (done) return seen;
+      for (const chunk of value) seen.push(chunk[0]);
+    }
+  };
+  const readQueued = async (iterator) => {
+    const seen = [];
+    for (;;) {
+      const results = await Promise.all(
+        [iterator.next(), iterator.next(), iterator.next()]);
+      for (const { done, value } of results) {
+        if (done) return seen;
+        for (const chunk of value) seen.push(chunk[0]);
+      }
+    }
+  };
+  const a = readAll(shared.push()[Symbol.asyncIterator]());
+  const b = readAll(shared.push()[Symbol.asyncIterator]());
+  const c = readQueued(shared.push()[Symbol.asyncIterator]());
+  for (let i = 0; i < n; i++) {
+    if (i === n / 2) {
+      // Joins at the current position, waits, and detaches while waiting.
+      const detaching = shared.push()[Symbol.asyncIterator]();
+      const pending = detaching.next();
+      await detaching.return();
+      await pending;
+    }
+    await writer.write(new Uint8Array([i % 256]));
+    if (i % 7 === 0) await new Promise(setImmediate);
+  }
+  await writer.end();
+  const expected = Array.from({ length: n }, (_, i) => i % 256);
+  assert.deepStrictEqual(await a, expected);
+  assert.deepStrictEqual(await b, expected);
+  assert.deepStrictEqual(await c, expected);
+  assert.strictEqual(shared.consumerCount, 0);
+}
+
+testDetachedWaitersAreReleased()
+  .then(testWaitersAcrossWrites)
+  .then(common.mustCall());