stream: keep consumer state in fast mode

benchmark/streams/iter-throughput-share-sync.js

@@ -0,0 +1,37 @@
+'use strict';
+
+const common = require('../common.js');
+
+const bench = common.createBenchmark(main, {
+  consumers: [2, 8, 32],
+  batches: [1e4],
+  n: [5],
+}, {
+  flags: ['--experimental-stream-iter'],
+});
+
+function main({ consumers, batches, n }) {
+  const { shareSync } = require('stream/iter');
+  const chunk = Buffer.alloc(1024);
+  let bytes = 0;
+
+  function* source() {
+    for (let i = 0; i < batches; i++) yield [chunk];
+  }
+
+  bench.start();
+  for (let run = 0; run < n; run++) {
+    const shared = shareSync(source(), { budget: 65536 });
+    const readers = Array.from({ length: consumers }, () =>
+      shared.pull()[Symbol.iterator]());
+    for (let i = 0; i < batches; i++) {
+      for (let j = 0; j < consumers; j++) {
+        bytes += readers[j].next().value[0].byteLength;
+      }
+    }
+  }
+  if (bytes !== batches * consumers * n * chunk.byteLength) {
+    throw new Error('Incorrect byte count');
+  }
+  bench.end(batches * consumers * n);
+}

lib/internal/streams/iter/broadcast.js

@@ -11,6 +11,7 @@ const {
   ArrayPrototypePush,
   ArrayPrototypeShift,
   FunctionPrototypeCall,
+  ObjectSetPrototypeOf,
   PromisePrototypeThen,
   PromiseReject,
   PromiseResolve,
@@ -183,8 +184,7 @@ class BroadcastImpl {
   }
 
   #createRawConsumer() {
-    const state = {
-      __proto__: null,
+    const state = ObjectSetPrototypeOf({
       // Start at the oldest buffered entry so late-joining consumers
       // can read data already in the buffer.
       cursor: this.#bufferStart,
@@ -193,7 +193,7 @@ class BroadcastImpl {
       pending: [],
       detached: false,
       error: kNoBroadcastError,
-    };
+    }, null);
 
     this.#consumers.add(state);
     if (this.#consumers.size === 1) {

lib/internal/streams/iter/share.js

@@ -8,6 +8,7 @@
 const {
   ArrayPrototypePush,
   FunctionPrototypeCall,
+  ObjectSetPrototypeOf,
   PromisePrototypeThen,
   PromiseResolve,
   PromiseWithResolvers,
@@ -141,15 +142,14 @@ class ShareImpl {
   }
 
   #createRawConsumer() {
-    const state = {
-      __proto__: null,
+    const state = ObjectSetPrototypeOf({
       cursor: this.#bufferStart,
       resolve: null,
       reject: null,
       detached: false,
       error: kNoShareError,
       pendingNext: PromiseResolve(),
-    };
+    }, null);
 
     this.#consumers.add(state);
     if (this.#consumers.size === 1) {
@@ -546,12 +546,11 @@ class SyncShareImpl {
   }
 
   #createRawConsumer() {
-    const state = {
-      __proto__: null,
+    const state = ObjectSetPrototypeOf({
       cursor: this.#bufferStart,
       detached: false,
       error: kNoShareError,
-    };
+    }, null);
 
     this.#consumers.add(state);
     if (this.#consumers.size === 1) {