Something went wrong. Try again.
Offload functions to worker threads with shared memory primitives for Node.js.
Something went wrong. Try again.
6.3 kB · 164 lines
TypeScript
at main
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165// Measures pure round-trip dispatch overhead for value tasks — a noop// function is dispatched N times and timed. Surfaces the cost of// postMessage + event loop re-entry per task.//// Three benchmarks:// - Ping-pong latency: strict await-each-dispatch, 1 worker// - Pipelined throughput: N in-flight at once, 1 worker// - Parallel throughput: N in-flight spread across pool//// Requires Node v24+.// Run: node examples/benchmark-dispatch/main.ts [--atomics]
import { availableParallelism } from 'node:os';import { workers, channel, assign } from '../../src/index.ts';import { noop, noopBatch, noopStream, noopStreamGen, noopConsume } from './noop.ts';
const atomicsFlag = process.argv.includes('--atomics');const maxWorkers = availableParallelism();
// Passing `atomics: true` is forward-compatible — the option is ignored// until the feature lands, so the same bench file times both modes.const poolOpts = atomicsFlag ? ({ atomics: true } as any) : {};
async function warmup(run: ReturnType<typeof workers>, n: number) { for (let i = 0; i < n; i++) await run(noop(i));}
async function pingPong(iters: number): Promise<number> { using run = workers(1, poolOpts); await warmup(run, 100); const t0 = performance.now(); for (let i = 0; i < iters; i++) await run(noop(i)); return iters / ((performance.now() - t0) / 1000);}
async function pipelined(iters: number, inflight: number): Promise<number> { using run = workers(1, poolOpts); await warmup(run, 100); const t0 = performance.now(); let next = 0; async function slot() { while (next < iters) { const i = next++; await run(noop(i)); } } await Promise.all(Array.from({ length: inflight }, () => slot())); return iters / ((performance.now() - t0) / 1000);}
async function parallel(iters: number, numWorkers: number): Promise<number> { using run = workers(numWorkers, poolOpts); await warmup(run, 100); const t0 = performance.now(); const promises = Array.from({ length: iters }, (_, i) => run(noop(i))); await Promise.all(promises); return iters / ((performance.now() - t0) / 1000);}
// One task carries the whole batch — items are not dispatched individually.async function batched(iters: number): Promise<number> { using run = workers(1, poolOpts); await run(noopBatch([1, 2, 3])); // warm the handler const items = Array.from({ length: iters }, (_, i) => i); const t0 = performance.now(); await run(noopBatch(items)); return iters / ((performance.now() - t0) / 1000);}
// 1-direction stream: worker generates items, parent consumes. Fair// comparison to channel (one cross-thread hop per item in both cases).async function streamingOneWay(iters: number): Promise<number> { using run = workers(1, poolOpts); for await (const _ of run(noopStreamGen(3))) void _; // warm const t0 = performance.now(); let count = 0; for await (const _ of run(noopStreamGen(iters))) { count++; void _; } if (count !== iters) throw new Error(`expected ${iters} items, got ${count}`); return iters / ((performance.now() - t0) / 1000);}
async function channelFanout(iters: number, numWorkers: number): Promise<number> { using run = workers(numWorkers, poolOpts); async function* source() { for (let i = 0; i < iters; i++) yield i; } const ch = channel(source()); const tasks = run.workers.map((w) => assign(w, noopConsume(ch))); const t0 = performance.now(); const counts = (await run(tasks)) as number[]; const total = counts.reduce((a, b) => a + b, 0); if (total !== iters) throw new Error(`expected ${iters} items, got ${total}`); return iters / ((performance.now() - t0) / 1000);}
// Round-trip stream: parent generates → worker re-yields → parent consumes.// Two hops per item, so ~half the throughput of a one-way test.async function streaming(iters: number): Promise<number> { using run = workers(1, poolOpts); async function* probe() { for (let i = 0; i < 3; i++) yield i; } for await (const _ of run(noopStream(probe()))) void _; // warm
async function* gen() { for (let i = 0; i < iters; i++) yield i; } const t0 = performance.now(); let count = 0; for await (const _ of run(noopStream(gen()))) { count++; void _; } if (count !== iters) throw new Error(`expected ${iters} items, got ${count}`); return iters / ((performance.now() - t0) / 1000);}
const SEQ = 20_000;const PIPE = 50_000;const PAR = 100_000;
console.log(`mode: ${atomicsFlag ? 'atomics' : 'default (postMessage events)'}`);console.log(`cpus: ${maxWorkers}\n`);
console.log('ping-pong latency (strict await-each, 1 worker):');const pp = await pingPong(SEQ);console.log( ` ${SEQ.toLocaleString()} tasks ${Math.round(pp).toLocaleString().padStart(9)} ops/s (${(1e6 / pp).toFixed(1)} µs/roundtrip)`,);
console.log('\npipelined throughput (1 worker, in-flight window):');for (const window of [4, 16, 64, 256]) { const ips = await pipelined(PIPE, window); console.log(` window=${String(window).padStart(3)} ${Math.round(ips).toLocaleString().padStart(9)} ops/s`);}
console.log('\nparallel throughput (N workers, 100K tasks):');for (const n of [1, 2, 4, maxWorkers]) { const ips = await parallel(PAR, n); console.log(` ${String(n).padStart(2)} workers ${Math.round(ips).toLocaleString().padStart(9)} ops/s`);}
console.log('\nfair 1-direction comparison (1 worker, 100K items):');const str1 = await streamingOneWay(PAR);const ch1 = await channelFanout(PAR, 1);const strRT = await streaming(PAR);console.log( ` stream (worker generates → parent consumes) ${Math.round(str1).toLocaleString().padStart(9)} items/s`,);console.log( ` channel (parent generates → worker consumes) ${Math.round(ch1).toLocaleString().padStart(9)} items/s`,);console.log( ` round-trip stream (parent → worker → parent, 2 hops) ${Math.round(strRT).toLocaleString().padStart(9)} items/s`,);
console.log('\nbatch vs stream vs per-task (1 worker, 100K items):');const perTask = await parallel(PAR, 1);const bat = await batched(PAR);console.log(` per-task (run(noop(i)) ×N) ${Math.round(perTask).toLocaleString().padStart(9)} items/s`);console.log(` batch (run(noopBatch([...N]))) ${Math.round(bat).toLocaleString().padStart(9)} items/s`);