| 71 | } |
| 72 | |
| 73 | function benchWebStream(chunk, numConsumers, datasize, n, totalOps) { |
| 74 | async function run() { |
| 75 | let remaining = datasize; |
| 76 | const rs = new ReadableStream({ |
| 77 | pull(controller) { |
| 78 | if (remaining <= 0) { controller.close(); return; } |
| 79 | const size = Math.min(remaining, chunk.length); |
| 80 | remaining -= size; |
| 81 | controller.enqueue( |
| 82 | size === chunk.length ? chunk : chunk.subarray(0, size)); |
| 83 | }, |
| 84 | }); |
| 85 | |
| 86 | // Chain tee() calls to get numConsumers branches. |
| 87 | // tee() gives 2; for 4 we tee twice at each level. |
| 88 | const branches = []; |
| 89 | if (numConsumers === 1) { |
| 90 | branches.push(rs); |
| 91 | } else { |
| 92 | const pending = [rs]; |
| 93 | while (branches.length + pending.length < numConsumers) { |
| 94 | const stream = pending.shift(); |
| 95 | const [a, b] = stream.tee(); |
| 96 | pending.push(a, b); |
| 97 | } |
| 98 | branches.push(...pending); |
| 99 | } |
| 100 | |
| 101 | const ws = () => new WritableStream({ write() {} }); |
| 102 | await Promise.all(branches.map((b) => b.pipeTo(ws()))); |
| 103 | } |
| 104 | |
| 105 | (async () => { |
| 106 | bench.start(); |
| 107 | for (let i = 0; i < n; i++) await run(); |
| 108 | bench.end(totalOps); |
| 109 | })(); |
| 110 | } |
| 111 | |
| 112 | function benchIter(chunk, numConsumers, datasize, n, totalOps) { |
| 113 | const { broadcast } = require('stream/iter'); |