githubmirror/node

benchmark/streams/compose.js
99 строк2 KB
Trivikram Kamat
stream: use data listener for compose forwarding
18 июн 2026, 23:12
18 июн 2026, 23:12f954ab3
'use strict';
const common = require('../common.js');

const {
  PassThrough,
  Readable,
  Writable,
  compose,
} = require('node:stream');

const bench = common.createBenchmark(main, {
  type: ['creation', 'throughput'],
  n: [1, 1e3],
  streams: [100],
  chunks: [1e4],
}, {
  combinationFilter({ type, n }) {
    return type === 'creation' ? n === 1e3 : n === 1;
  },
  test: {
    n: [1, 1e3],
    type: ['creation', 'throughput'],
  },
});

function main({ type, n, streams, chunks }) {
  switch (type) {
    case 'creation':
      return benchCreation(n, streams);
    case 'throughput':
      return benchThroughput(n, streams, chunks);
  }
}

function benchCreation(n, numberOfPassThroughs) {
  const cachedPassThroughs = [];
  const cachedReadables = [];
  const cachedWritables = [];

  for (let i = 0; i < n; i++) {
    const passThroughs = [];

    for (let i = 0; i < numberOfPassThroughs; i++) {
      passThroughs.push(new PassThrough());
    }

    const readable = Readable.from(['hello', 'world']);
    const writable = new Writable({ objectMode: true, write: (chunk, encoding, cb) => cb() });

    cachedPassThroughs.push(passThroughs);
    cachedReadables.push(readable);
    cachedWritables.push(writable);
  }

  bench.start();
  for (let i = 0; i < n; i++) {
    const composed = compose(cachedReadables[i], ...cachedPassThroughs[i], cachedWritables[i]);
    composed.end();
  }
  bench.end(n);
}

function benchThroughput(n, numberOfPassThroughs, chunks) {
  const chunk = Buffer.alloc(1024);

  let i = 0;
  bench.start();

  function run() {
    if (i++ === n) {
      bench.end(n * chunks);
      return;
    }

    const passThroughs = [];
    for (let i = 0; i < numberOfPassThroughs; i++) {
      passThroughs.push(new PassThrough());
    }

    let remaining = chunks;
    const composed = compose(...passThroughs);
    composed.on('data', () => {});
    composed.on('end', run);

    write();

    function write() {
      while (remaining-- > 0) {
        if (!composed.write(chunk)) {
          composed.once('drain', write);
          return;
        }
      }
      composed.end();
    }
  }

  run();
}