stream: add stream/iter benchmarks

Author: jasnell

benchmark/streams/iter/creation.js

@@ -0,0 +1,92 @@
+// Object creation overhead benchmark.
+// Measures the cost of constructing stream infrastructure (no data flow).
+'use strict';
+
+const common = require('../../common.js');
+const { Readable, Writable, Transform, PassThrough } = require('stream');
+
+const bench = common.createBenchmark(main, {
+  api: ['classic', 'webstream', 'iter'],
+  type: ['readable', 'writable', 'transform', 'pair'],
+  n: [1e5],
+}, {
+  flags: ['--experimental-stream-iter'],
+  // Iter has no standalone Transform class; transforms are plain functions.
+  combinationFilter: ({ api, type }) =>
+    !(api === 'iter' && type === 'transform'),
+});
+
+function main({ api, type, n }) {
+  switch (api) {
+    case 'classic':
+      return benchClassic(type, n);
+    case 'webstream':
+      return benchWebStream(type, n);
+    case 'iter':
+      return benchIter(type, n);
+  }
+}
+
+function benchClassic(type, n) {
+  bench.start();
+  switch (type) {
+    case 'readable':
+      for (let i = 0; i < n; i++) new Readable({ read() {} });
+      break;
+    case 'writable':
+      for (let i = 0; i < n; i++) new Writable({ write(c, e, cb) { cb(); } });
+      break;
+    case 'transform':
+      for (let i = 0; i < n; i++) new Transform({
+        transform(c, e, cb) { cb(null, c); },
+      });
+      break;
+    case 'pair':
+      for (let i = 0; i < n; i++) new PassThrough();
+      break;
+  }
+  bench.end(n);
+}
+
+function benchWebStream(type, n) {
+  bench.start();
+  switch (type) {
+    case 'readable':
+      for (let i = 0; i < n; i++) new ReadableStream({ pull() {} });
+      break;
+    case 'writable':
+      for (let i = 0; i < n; i++) new WritableStream({ write() {} });
+      break;
+    case 'transform':
+      for (let i = 0; i < n; i++) new TransformStream({
+        transform(c, controller) { controller.enqueue(c); },
+      });
+      break;
+    case 'pair': {
+      // TransformStream gives a readable+writable pair
+      for (let i = 0; i < n; i++) new TransformStream();
+      break;
+    }
+  }
+  bench.end(n);
+}
+
+function benchIter(type, n) {
+  const { push, from, duplex } = require('stream/iter');
+
+  bench.start();
+  switch (type) {
+    case 'readable':
+      for (let i = 0; i < n; i++) from('x');
+      break;
+    case 'writable':
+      // push() creates a writer+readable pair
+      for (let i = 0; i < n; i++) push();
+      break;
+    case 'pair':
+      // duplex() creates a bidirectional channel pair
+      for (let i = 0; i < n; i++) duplex();
+      break;
+  }
+  bench.end(n);
+}

benchmark/streams/iter/file-read.js

@@ -0,0 +1,107 @@
+// File reading throughput benchmark.
+// Reads a real file through the three stream APIs.
+'use strict';
+
+const common = require('../../common.js');
+const fs = require('fs');
+const { Writable, pipeline } = require('stream');
+const tmpdir = require('../../../test/common/tmpdir');
+
+tmpdir.refresh();
+const filename = tmpdir.resolve(`.removeme-bench-file-read-${process.pid}`);
+
+const bench = common.createBenchmark(main, {
+  api: ['classic', 'webstream', 'iter'],
+  filesize: [1024 * 1024, 16 * 1024 * 1024, 64 * 1024 * 1024],
+  n: [5],
+}, {
+  flags: ['--experimental-stream-iter'],
+});
+
+function main({ api, filesize, n }) {
+  // Create fixture file
+  const chunk = Buffer.alloc(Math.min(filesize, 64 * 1024), 'abcdefghij');
+  const fd = fs.openSync(filename, 'w');
+  let remaining = filesize;
+  while (remaining > 0) {
+    const size = Math.min(remaining, chunk.length);
+    fs.writeSync(fd, chunk, 0, size);
+    remaining -= size;
+  }
+  fs.closeSync(fd);
+
+  const totalOps = (filesize * n) / (1024 * 1024);
+
+  switch (api) {
+    case 'classic':
+      return benchClassic(filesize, n, totalOps);
+    case 'webstream':
+      return benchWebStream(filesize, n, totalOps);
+    case 'iter':
+      return benchIter(filesize, n, totalOps);
+  }
+}
+
+function benchClassic(filesize, n, totalOps) {
+  function run(cb) {
+    const r = fs.createReadStream(filename);
+    const w = new Writable({ write(data, enc, cb) { cb(); } });
+    pipeline(r, w, cb);
+  }
+
+  // Warmup
+  run(() => {
+    let i = 0;
+    bench.start();
+    (function next() {
+      if (i++ >= n) {
+        fs.unlinkSync(filename);
+        return bench.end(totalOps);
+      }
+      run(next);
+    })();
+  });
+}
+
+function benchWebStream(filesize, n, totalOps) {
+  const fsp = require('fs/promises');
+
+  async function run() {
+    const fh = await fsp.open(filename, 'r');
+    const rs = fh.readableWebStream();
+    const ws = new WritableStream({ write() {} });
+    await rs.pipeTo(ws);
+    await fh.close();
+  }
+
+  (async () => {
+    // Warmup
+    await run();
+
+    bench.start();
+    for (let i = 0; i < n; i++) await run();
+    fs.unlinkSync(filename);
+    bench.end(totalOps);
+  })();
+}
+
+function benchIter(filesize, n, totalOps) {
+  const fsp = require('fs/promises');
+  const { pipeTo } = require('stream/iter');
+
+  async function run() {
+    const fh = await fsp.open(filename, 'r');
+    await pipeTo(fh.pull(), { write() {} });
+    await fh.close();
+  }
+
+  (async () => {
+    // Warmup
+    await run();
+
+    bench.start();
+    for (let i = 0; i < n; i++) await run();
+    fs.unlinkSync(filename);
+    bench.end(totalOps);
+  })();
+}

benchmark/streams/iter/throughput-broadcast.js

@@ -0,0 +1,145 @@
+// Throughput benchmark: fan-out data to N consumers simultaneously.
+// Classic streams use PassThrough + pipe, Web Streams use tee() chains,
+// stream/iter uses broadcast() with push() consumers.
+'use strict';
+
+const common = require('../../common.js');
+const { PassThrough, Writable } = require('stream');
+
+const bench = common.createBenchmark(main, {
+  api: ['classic', 'webstream', 'iter'],
+  consumers: [1, 2, 4],
+  datasize: [1024 * 1024, 16 * 1024 * 1024],
+  n: [5],
+}, {
+  flags: ['--experimental-stream-iter'],
+});
+
+const CHUNK_SIZE = 64 * 1024;
+
+function main({ api, consumers, datasize, n }) {
+  const chunk = Buffer.alloc(CHUNK_SIZE, 'abcdefghij');
+  const totalOps = (datasize * n) / (1024 * 1024);
+
+  switch (api) {
+    case 'classic':
+      return benchClassic(chunk, consumers, datasize, n, totalOps);
+    case 'webstream':
+      return benchWebStream(chunk, consumers, datasize, n, totalOps);
+    case 'iter':
+      return benchIter(chunk, consumers, datasize, n, totalOps);
+  }
+}
+
+function benchClassic(chunk, numConsumers, datasize, n, totalOps) {
+  function run(cb) {
+    const source = new PassThrough();
+    const sinks = [];
+    let finished = 0;
+
+    for (let c = 0; c < numConsumers; c++) {
+      const w = new Writable({ write(data, enc, cb) { cb(); } });
+      source.pipe(w);
+      w.on('finish', () => { if (++finished === numConsumers) cb(); });
+      sinks.push(w);
+    }
+
+    let remaining = datasize;
+    function write() {
+      let ok = true;
+      while (ok && remaining > 0) {
+        const size = Math.min(remaining, chunk.length);
+        remaining -= size;
+        const buf = size === chunk.length ? chunk : chunk.subarray(0, size);
+        ok = source.write(buf);
+      }
+      if (remaining > 0) {
+        source.once('drain', write);
+      } else {
+        source.end();
+      }
+    }
+    write();
+  }
+
+  let i = 0;
+  bench.start();
+  (function next() {
+    if (i++ >= n) return bench.end(totalOps);
+    run(next);
+  })();
+}
+
+function benchWebStream(chunk, numConsumers, datasize, n, totalOps) {
+  async function run() {
+    let remaining = datasize;
+    const rs = new ReadableStream({
+      pull(controller) {
+        if (remaining <= 0) { controller.close(); return; }
+        const size = Math.min(remaining, chunk.length);
+        remaining -= size;
+        controller.enqueue(
+          size === chunk.length ? chunk : chunk.subarray(0, size));
+      },
+    });
+
+    // Chain tee() calls to get numConsumers branches.
+    // tee() gives 2; for 4 we tee twice at each level.
+    const branches = [];
+    if (numConsumers === 1) {
+      branches.push(rs);
+    } else {
+      const pending = [rs];
+      while (branches.length + pending.length < numConsumers) {
+        const stream = pending.shift();
+        const [a, b] = stream.tee();
+        pending.push(a, b);
+      }
+      branches.push(...pending);
+    }
+
+    const ws = () => new WritableStream({ write() {} });
+    await Promise.all(branches.map((b) => b.pipeTo(ws())));
+  }
+
+  (async () => {
+    bench.start();
+    for (let i = 0; i < n; i++) await run();
+    bench.end(totalOps);
+  })();
+}
+
+function benchIter(chunk, numConsumers, datasize, n, totalOps) {
+  const { broadcast } = require('stream/iter');
+
+  // No-op consumer: drain all batches without collecting
+  async function drain(source) {
+    // eslint-disable-next-line no-unused-vars
+    for await (const _ of source) { /* drain */ }
+  }
+
+  async function run() {
+    const { writer, broadcast: bc } = broadcast();
+    const consumers = [];
+    for (let c = 0; c < numConsumers; c++) {
+      consumers.push(drain(bc.push()));
+    }
+
+    let remaining = datasize;
+    while (remaining > 0) {
+      const size = Math.min(remaining, chunk.length);
+      remaining -= size;
+      const buf = size === chunk.length ? chunk : chunk.subarray(0, size);
+      writer.writeSync(buf);
+    }
+    writer.endSync();
+
+    await Promise.all(consumers);
+  }
+
+  (async () => {
+    bench.start();
+    for (let i = 0; i < n; i++) await run();
+    bench.end(totalOps);
+  })();
+}