diff --git a/examples/benchmark-dispatch/main.ts b/examples/benchmark-dispatch/main.ts index 68110da..740ad83 100644 --- a/examples/benchmark-dispatch/main.ts +++ b/examples/benchmark-dispatch/main.ts @@ -11,8 +11,8 @@ // Run: node examples/benchmark-dispatch/main.ts [--atomics] import { availableParallelism } from 'node:os'; -import { workers } from '../../src/index.ts'; -import { noop, noopBatch, noopStream } from './noop.ts'; +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(); @@ -67,8 +67,37 @@ async function batched(iters: number): Promise { return iters / ((performance.now() - t0) / 1000); } -// Streaming: items flow over a MessageChannel to the worker-side async -// generator. Worker drains at its own pace — no per-item task envelope. +// 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 { + 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 { + 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 { using run = workers(1, poolOpts); async function* probe() { @@ -112,10 +141,16 @@ for (const n of [1, 2, 4, maxWorkers]) { 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); -const str = await streaming(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`); -console.log(` stream (run(noopStream(gen()))) ${Math.round(str).toLocaleString().padStart(9)} items/s`); diff --git a/examples/benchmark-dispatch/noop.ts b/examples/benchmark-dispatch/noop.ts index 368acdd..fe41034 100644 --- a/examples/benchmark-dispatch/noop.ts +++ b/examples/benchmark-dispatch/noop.ts @@ -13,3 +13,14 @@ export const noopBatch = mo(import.meta, (items: number[]): number[] => items); export const noopStream = mo(import.meta, async function* (items: AsyncIterable) { for await (const item of items) yield item; }); + +// Worker-side generator — produces N items. One-direction: worker → parent. +export const noopStreamGen = mo(import.meta, async function* (n: number) { + for (let i = 0; i < n; i++) yield i; +}); + +export const noopConsume = mo(import.meta, async (items: AsyncIterable): Promise => { + let count = 0; + for await (const _ of items) count++; + return count; +});