diff --git a/benchmark/streams/iter-throughput-share-sync.js b/benchmark/streams/iter-throughput-share-sync.js new file mode 100644 index 00000000000..3a5de56b592 --- /dev/null +++ b/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); +} diff --git a/lib/internal/streams/iter/broadcast.js b/lib/internal/streams/iter/broadcast.js index 3b37a5b5f2a..44344d8b1bf 100644 --- a/lib/internal/streams/iter/broadcast.js +++ b/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) { diff --git a/lib/internal/streams/iter/share.js b/lib/internal/streams/iter/share.js index 89170bdabd2..25e47dce5bf 100644 --- a/lib/internal/streams/iter/share.js +++ b/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) {