From ee49ea1026948ce62f5272a2b1300b68d9260a50 Mon Sep 17 00:00:00 2001 From: Devin Ivy Date: Sun, 19 Apr 2026 16:32:00 -0400 Subject: [PATCH] bench: fair 1-direction stream+channel, expand channel fan-out MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Add noopStreamGen and noopConsume moroutines + matching benches that pit 1-direction stream (worker→parent) against 1-direction channel (parent→worker) on equal footing — both are one cross-thread hop per item. Findings (on this branch, HEAD): stream (worker generates → parent consumes) ~746K items/s channel (parent generates → worker consumes) ~652K items/s stream round-trip (parent → worker → parent) ~464K items/s vs unoptimized (bb41d5d): stream 1-dir: ~66K (+11.3x) channel 1-dir: ~67K (+9.7x) Two takeaways: - The earlier "channel beats stream" observation was an artifact of comparing channel's 1-hop case to stream's 2-hop (pass-through) noopStream-with-input case. When measured 1-direction-vs-1-direction, stream and channel are tied pre-optimization. - Post-optimization, stream is ~15% faster than channel — that's the win from atomics-based backpressure over message+adaptive-yield. Applying atomics to the Distributor would likely close that gap. Also keeps channelFanout at varying worker counts as a separate section — shows the Distributor's single-threaded producer bottleneck (peak at ~2 consumers, regresses past 4). Co-Authored-By: Claude Opus 4.7 (1M context) --- examples/benchmark-dispatch/main.ts | 47 +++++++++++++++++++++++++---- examples/benchmark-dispatch/noop.ts | 11 +++++++ 2 files changed, 52 insertions(+), 6 deletions(-) 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; +}); -- 2.51.2