stream: performance optimizations for stream/iter pipeline and broadcast

Optimize the stream/iter implementation based on benchmark analysis
comparing classic streams, web streams, and stream/iter.

- Eliminate `withFlushSignalAsync`/`withFlushSignalSync` generator wrappers
  from the stateless transform pipeline. Stateless transforms now handle
  their own flush (`null`) signal internally after the for-await loop,
  removing an entire async generator layer per pipeline. Stateful
  transforms retain the wrapper since their cost is dominated by the
  transform operation itself (compression, encryption, etc).
- Hoist writer capability checks in `pipeTo`/`pipeToSync`. Property lookups
  for `writeSync`/`writevSync`/`endSync`/`failSync` are done once before the
  loop instead of per-chunk via optional chaining. Split signal/no-signal
  loops to avoid per-batch null checks. Added `writevSync` batch write
  support to `pipeToSync`.
- Optimize `isUint8ArrayBatch` with single-element fast path and plain
  for loop. Replaces `ArrayPrototypeEvery` (function call per element)
  with direct indexed loop. Short-circuits on length 1 (most common
  from transforms) and checks first/last before iterating middle.
- Make broadcast consumer `next()`/`return()`/`throw()` non-async. Returns
  `PromiseResolve()` directly on the fast path (data in buffer) instead
  of wrapping through async function machinery. Caches the done result.
- `RingBuffer`: replace modulo with bitwise AND. Capacity is always a
  power of 2, so index computation uses & mask instead of % capacity.
- `Broadcast`: incremental min-cursor tracking. Replaces O(N) full scan
  of all consumers on every `next()` call with a cached min cursor that
  is only recomputed when dirty (consumer at the minimum advances or
  detaches). Eliminates O(N^2) per-write-cycle scaling.
- `Broadcast`: separate waiters list for `notifyConsumers`. Only iterates
  consumers with pending resolve callbacks instead of scanning all
  consumers on every write.
- `concatBytes`: cache per-chunk byte lengths to avoid reading `byteLength`
  twice per chunk (once for total, once for offset advance). Remove
  dead `totalByteLength` function.

Benchmark results (MB/s, higher is better):

| Benchmark        | classic | webstream | iter   | iter-sync | iter vs classic |
| ---------------- | ------- | --------- | ------ | --------- | --------------- |
| Identity 1MB     | 1,245   | 582       | **3,110**  | 16,658    | 2.5x            |
| Identity 64MB    | 31,410  | 14,980    | **33,894** | 62,111    | 1.1x            |
| Transform 1MB    | 287     | 227       | **325**    | 327       | 1.1x            |
| Transform 64MB   | 595     | 605       | **605**    | 573       | 1.0x            |
| Compression 1MB  | **123**     | 98        | 110    | --        | 0.9x            |
| Compression 64MB | **329**     | 303       | 308    | --        | 0.9x            |
| pipeTo 1MB       | 1,137   | 494       | **2,740**  | 13,611    | 2.4x            |
| pipeTo 64MB      | 22,081  | 15,377    | **30,036** | 60,976    | 1.4x            |
| Broadcast 1c 1MB | 1,365   | 521       | **1,991**  | --        | 1.5x            |
| Broadcast 2c 1MB | 1,285   | 439       | **1,962**  | --        | 1.5x            |
| Broadcast 4c 1MB | **1,217**   | 322       | 750    | --        | 0.6x            |
| File read 16MB   | 1,469   | 537       | **1,639**  | --        | 1.1x            |

The creation benchmarks show the raw cost of constructing the various
objects without any other activity. The `classic` Node.js streams are
faster here simply because they do less work on actual creation.

| Creation (ops/sec) | classic   | webstream | iter      | iter vs classic |
| ------------------ | --------- | --------- | --------- | --------------- |
| readable           | 8,662,361 | 505,889   | 1,144,385 | 0.1x            |
| writable           | 3,856,139 | 269,950   | 1,285,210 | 0.3x            |
| pair               | 3,120,224 | 141,988   | 349,176   | 0.1x            |

Author: jasnell

lib/internal/streams/iter/broadcast.js

@@ -15,6 +15,7 @@ const {
   MathMin,
   NumberMAX_SAFE_INTEGER,
   PromisePrototypeThen,
+  PromiseReject,
   PromiseResolve,
   PromiseWithResolvers,
   SafeSet,
@@ -111,11 +112,14 @@ class BroadcastImpl {
   #buffer = new RingBuffer();
   #bufferStart = 0;
   #consumers = new SafeSet();
+  #waiters = [];          // Consumers with pending resolve (subset of #consumers)
   #ended = false;
   #error = null;
   #cancelled = false;
   #options;
   #writer = null;
+  #cachedMinCursor = 0;
+  #minCursorDirty = false;
 
   constructor(options) {
     this.#options = options;
@@ -166,62 +170,76 @@ class BroadcastImpl {
     };
 
     this.#consumers.add(state);
+    // New consumer starts at the latest position; min cursor unchanged
+    // unless this is the first consumer.
+    if (this.#consumers.size === 1) {
+      this.#cachedMinCursor = state.cursor;
+      this.#minCursorDirty = false;
+    }
     const self = this;
 
+    const kDone = PromiseResolve(
+      { __proto__: null, done: true, value: undefined });
+
+    function detach() {
+      state.detached = true;
+      state.resolve = null;
+      state.reject = null;
+      self.#consumers.delete(state);
+      self.#minCursorDirty = true;
+      self.#tryTrimBuffer();
+    }
+
     return {
       __proto__: null,
       [SymbolAsyncIterator]() {
         return {
           __proto__: null,
-          async next() {
+          next() {
             if (state.detached) {
-              // If detached due to an error, throw the error
-              if (self.#error) throw self.#error;
-              return { __proto__: null, done: true, value: undefined };
+              if (self.#error) return PromiseReject(self.#error);
+              return kDone;
             }
 
             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;
+              }
               state.cursor++;
               self.#tryTrimBuffer();
-              return { __proto__: null, done: false, value: chunk };
+              return PromiseResolve(
+                { __proto__: null, done: false, value: chunk });
             }
 
             if (self.#error) {
               state.detached = true;
               self.#consumers.delete(state);
-              throw self.#error;
+              return PromiseReject(self.#error);
             }
 
             if (self.#ended || self.#cancelled) {
-              state.detached = true;
-              self.#consumers.delete(state);
-              return { __proto__: null, done: true, value: undefined };
+              detach();
+              return kDone;
             }
 
             const { promise, resolve, reject } = PromiseWithResolvers();
             state.resolve = resolve;
             state.reject = reject;
+            ArrayPrototypePush(self.#waiters, state);
             return promise;
           },
 
-          async return() {
-            state.detached = true;
-            state.resolve = null;
-            state.reject = null;
-            self.#consumers.delete(state);
-            self.#tryTrimBuffer();
-            return { __proto__: null, done: true, value: undefined };
+          return() {
+            detach();
+            return kDone;
           },
 
-          async throw() {
-            state.detached = true;
-            state.resolve = null;
-            state.reject = null;
-            self.#consumers.delete(state);
-            self.#tryTrimBuffer();
-            return { __proto__: null, done: true, value: undefined };
+          throw() {
+            detach();
+            return kDone;
           },
         };
       },
@@ -342,23 +360,26 @@ class BroadcastImpl {
 
   // Private methods
 
-  #getMinCursor() {
+  #recomputeMinCursor() {
     let min = Infinity;
     for (const consumer of this.#consumers) {
       if (consumer.cursor < min) {
         min = consumer.cursor;
       }
     }
-    return min === Infinity ?
+    this.#cachedMinCursor = min === Infinity ?
       this.#bufferStart + this.#buffer.length : min;
+    this.#minCursorDirty = false;
   }
 
   #tryTrimBuffer() {
-    const minCursor = this.#getMinCursor();
-    const trimCount = minCursor - this.#bufferStart;
+    if (this.#minCursorDirty) {
+      this.#recomputeMinCursor();
+    }
+    const trimCount = this.#cachedMinCursor - this.#bufferStart;
     if (trimCount > 0) {
       this.#buffer.trimFront(trimCount);
-      this.#bufferStart = minCursor;
+      this.#bufferStart = this.#cachedMinCursor;
 
       if (this[kOnBufferDrained] &&
           this.#buffer.length < this.#options.highWaterMark) {
@@ -368,16 +389,28 @@ class BroadcastImpl {
   }
 
   #notifyConsumers() {
-    for (const consumer of this.#consumers) {
+    const waiters = this.#waiters;
+    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 = [];
+    for (let i = 0; i < waiters.length; i++) {
+      const consumer = waiters[i];
       if (consumer.resolve) {
         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;
+          }
           consumer.cursor++;
           const resolve = consumer.resolve;
           consumer.resolve = null;
           consumer.reject = null;
           resolve({ __proto__: null, done: false, value: chunk });
+        } else {
+          // Still waiting -- put back
+          ArrayPrototypePush(this.#waiters, consumer);
         }
       }
     }

lib/internal/streams/iter/from.js

@@ -255,8 +255,17 @@ function* normalizeSyncValue(value) {
  */
 function isUint8ArrayBatch(value) {
   if (!ArrayIsArray(value)) return false;
-  if (value.length === 0) return true;
-  return ArrayPrototypeEvery(value, isUint8Array);
+  const len = value.length;
+  if (len === 0) return true;
+  // Fast path: single-element batch (most common from transforms)
+  if (len === 1) return isUint8Array(value[0]);
+  // Check first and last before iterating all elements
+  if (!isUint8Array(value[0]) || !isUint8Array(value[len - 1])) return false;
+  if (len === 2) return true;
+  for (let i = 1; i < len - 1; i++) {
+    if (!isUint8Array(value[i])) return false;
+  }
+  return true;
 }
 
 /**