40dc5a1dc4Forward Node.js tail stream data through a pipe-style data listener instead of manually draining readable events. This keeps compose closer to the stream pipe hot path while preserving backpressure with pause and resume. Signed-off-by: Kamat, Trivikram <16024985+trivikr@users.noreply.github.com> Assisted-by: openai:gpt-5.5 PR-URL: https://github.com/nodejs/node/pull/63593 Reviewed-By: Robert Nagy <ronagy@icloud.com> Reviewed-By: Ethan Arrowood <ethan@arrowood.dev> Reviewed-By: Jake Yuesong Li <jake.yuesong@gmail.com> Reviewed-By: Stephen Belanger <admin@stephenbelanger.com>
99 lines
2.1 KiB
JavaScript
99 lines
2.1 KiB
JavaScript
'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();
|
|
}
|