From 25ea1c104c69657e403327c8560aab5686f82dc1 Mon Sep 17 00:00:00 2001 From: Devin Ivy Date: Sat, 18 Apr 2026 00:51:19 -0400 Subject: [PATCH] feat: add inert() and map() helpers for bounded fan-out - inert(task) returns a plain task descriptor without PromiseLike or AsyncIterable protocols, safe to yield from an (async) generator without triggering auto-await - map(run, items, { concurrency, signal }) dispatches an iterable or async iterable of tasks to a Runner with bounded concurrency, yielding results in completion order - Accepts mixed task types: Task | Task yields string | number - Supports AbortSignal for stream cancellation; moroutine auto-transfers signals passed as task args so in-flight work can observe the same abort Includes test coverage, a bounded-map example that hashes a directory tree via recursive async generator, and a README section. Co-Authored-By: Claude Opus 4.7 (1M context) --- .changeset/inert-and-map.md | 10 +++ .nvmrc | 1 + README.md | 48 ++++++++++++ examples/bounded-map/hash-file.ts | 10 +++ examples/bounded-map/main.ts | 35 +++++++++ src/index.ts | 3 + src/inert.ts | 14 ++++ src/map.ts | 32 ++++++++ test/fixtures/map.ts | 17 +++++ test/map.test.ts | 122 ++++++++++++++++++++++++++++++ 10 files changed, 292 insertions(+) create mode 100644 .changeset/inert-and-map.md create mode 100644 .nvmrc create mode 100644 examples/bounded-map/hash-file.ts create mode 100644 examples/bounded-map/main.ts create mode 100644 src/inert.ts create mode 100644 src/map.ts create mode 100644 test/fixtures/map.ts create mode 100644 test/map.test.ts diff --git a/.changeset/inert-and-map.md b/.changeset/inert-and-map.md new file mode 100644 index 0000000..2e817ab --- /dev/null +++ b/.changeset/inert-and-map.md @@ -0,0 +1,10 @@ +--- +'moroutine': minor +--- + +Add `inert()` and `map()` helpers for fan-out over a worker pool + +- `inert(task)` returns a plain task descriptor without `PromiseLike` or `AsyncIterable` protocols — safe to yield from an (async) generator without triggering auto-await +- `map(run, items, { concurrency, signal })` dispatches an iterable or async iterable of tasks to a `Runner` with bounded concurrency, yielding results in completion order; accepts mixed task types (`Task | Task` → `string | number`) +- New example: `examples/bounded-map` — recursive directory walk hashing every file with bounded concurrency +- `Task` now carries a type-only arg brand to enable accurate result inference through `map()`; live tasks returned by `mo()` continue to be `PromiseLike` / `AsyncIterable` as before diff --git a/.nvmrc b/.nvmrc new file mode 100644 index 0000000..a45fd52 --- /dev/null +++ b/.nvmrc @@ -0,0 +1 @@ +24 diff --git a/README.md b/README.md index 121d1b4..bda93ac 100644 --- a/README.md +++ b/README.md @@ -389,6 +389,53 @@ Use `assign(worker, task)` to pin a task to a specific worker. `run.workers` is Without `channel()`, `AsyncIterable` and streaming task arguments are auto-detected and streamed to a single consumer. `channel()` is only needed for fan-out. +### `map()` — Bounded Fan-out + +Dispatch a stream of tasks to a pool with bounded concurrency, yielding results in completion order. Wrap each task with `inert()` so it passes through the stream as-is instead of being auto-awaited. + +```ts +// main.ts +import { readdir } from 'node:fs/promises'; +import { join } from 'node:path'; +import { workers, map, inert } from 'moroutine'; +import type { Task } from 'moroutine'; +import { hashFile, type FileHash } from './hash-file.ts'; + +{ + using run = workers(); + for await (const { path, hash } of map(run, walk('./src'), { concurrency: 4 })) { + console.log(`${hash.slice(0, 12)} ${path}`); + } +} + +async function* walk(dir: string): AsyncGenerator> { + for (const entry of await readdir(dir, { withFileTypes: true })) { + const p = join(dir, entry.name); + if (entry.isDirectory()) { + yield* walk(p); + } else { + yield inert(hashFile(p)); + } + } +} +``` + +```ts +// hash-file.ts +import { readFile } from 'node:fs/promises'; +import { createHash } from 'node:crypto'; +import { mo } from 'moroutine'; + +export type FileHash = { path: string; hash: string }; + +export const hashFile = mo(import.meta, async (path: string): Promise => { + const buf = await readFile(path); + return { path, hash: createHash('sha256').update(buf).digest('hex') }; +}); +``` + +`map()` accepts a sync iterable, async iterable, or generator of tasks. `concurrency` caps in-flight dispatches (default `1`). Mixed task types unify: `map` over `Task | Task` yields `string | number`. An optional `signal` aborts iteration — and, since moroutine auto-transfers `AbortSignal` args, the same signal passed to tasks will also cancel in-flight work. + ### Pipelines Chain streaming moroutines by passing one as an argument to the next. Each stage runs on its own dedicated worker. @@ -429,5 +476,6 @@ All examples require Node v24+ and can be run directly, e.g. `node examples/prim - [`examples/sqlite`](examples/sqlite) -- shared SQLite database on a worker via task-arg caching - [`examples/pipeline`](examples/pipeline) -- streaming pipeline across dedicated workers - [`examples/channel-fanout`](examples/channel-fanout) -- fan-out a channel to multiple workers via work stealing +- [`examples/bounded-map`](examples/bounded-map) -- bounded-concurrency fan-out with `map()` and a shared abort signal - [`examples/load-balancing`](examples/load-balancing) -- round-robin vs least-busy with variable-cost tasks - [`examples/benchmark`](examples/benchmark) -- roundtrip channel throughput with 1–N workers diff --git a/examples/bounded-map/hash-file.ts b/examples/bounded-map/hash-file.ts new file mode 100644 index 0000000..5b770bf --- /dev/null +++ b/examples/bounded-map/hash-file.ts @@ -0,0 +1,10 @@ +import { readFile } from 'node:fs/promises'; +import { createHash } from 'node:crypto'; +import { mo } from '../../src/index.ts'; + +export type FileHash = { path: string; hash: string }; + +export const hashFile = mo(import.meta, async (path: string): Promise => { + const buf = await readFile(path); + return { path, hash: createHash('sha256').update(buf).digest('hex') }; +}); diff --git a/examples/bounded-map/main.ts b/examples/bounded-map/main.ts new file mode 100644 index 0000000..6aae535 --- /dev/null +++ b/examples/bounded-map/main.ts @@ -0,0 +1,35 @@ +// Walk a directory tree with a recursive async generator and hash every file +// using map() with bounded concurrency. Results stream back in completion order. +// Requires Node v24+. +// +// Run: node examples/bounded-map/main.ts [dir] + +import { readdir } from 'node:fs/promises'; +import { join } from 'node:path'; +import { workers, map, inert } from '../../src/index.ts'; +import type { Task } from '../../src/index.ts'; +import { hashFile, type FileHash } from './hash-file.ts'; + +const root = process.argv[2] ?? './src'; + +{ + using run = workers(); + const start = performance.now(); + let count = 0; + for await (const { path, hash } of map(run, walk(root), { concurrency: 4 })) { + console.log(`${hash.slice(0, 12)} ${path}`); + count++; + } + console.log(`\nhashed ${count} files in ${(performance.now() - start).toFixed(0)}ms`); +} + +async function* walk(dir: string): AsyncGenerator> { + for (const entry of await readdir(dir, { withFileTypes: true })) { + const p = join(dir, entry.name); + if (entry.isDirectory()) { + yield* walk(p); + } else { + yield inert(hashFile(p)); + } + } +} diff --git a/src/index.ts b/src/index.ts index 1ccdb59..9235293 100644 --- a/src/index.ts +++ b/src/index.ts @@ -7,6 +7,9 @@ export type { ChannelOptions } from './channel.ts'; export { workers } from './worker-pool.ts'; export { transfer } from './transfer.ts'; export { assign } from './assign.ts'; +export { inert } from './inert.ts'; +export { map } from './map.ts'; +export type { MapOptions } from './map.ts'; export { roundRobin, leastBusy } from './balancers.ts'; export type { Task, RunResult, Balancer, Runner, WorkerHandle, WorkerOptions } from './runner.ts'; export { diff --git a/src/inert.ts b/src/inert.ts new file mode 100644 index 0000000..8607eb9 --- /dev/null +++ b/src/inert.ts @@ -0,0 +1,14 @@ +import type { Task } from './runner.ts'; + +/** + * Returns an inert copy of a task — same `uid`, `id`, `args`, and `worker` + * but without `PromiseLike` or `AsyncIterable` protocols. + * Safe to `yield` from an async generator without triggering auto-await. + * + * @param task - A task created by a {@link mo}-wrapped function. + * @returns A plain object with the same task identity, stripped of any thenable or iterable protocols. + */ +export function inert(task: Task): Task { + const { uid, id, args, worker } = task; + return { uid, id, args, worker }; +} diff --git a/src/map.ts b/src/map.ts new file mode 100644 index 0000000..01c2c34 --- /dev/null +++ b/src/map.ts @@ -0,0 +1,32 @@ +import { Readable } from 'node:stream'; +import type { Task, Runner } from './runner.ts'; + +type TaskResult = I extends Task ? T : never; + +/** Options controlling how {@link map} dispatches tasks to a {@link Runner}. */ +export interface MapOptions { + /** Maximum number of in-flight tasks at once. Defaults to `1`. */ + concurrency?: number; + /** Signal that, when aborted, stops pulling from `items` and ends iteration. */ + signal?: AbortSignal; +} + +/** + * Dispatches tasks yielded by `items` to a worker pool with bounded concurrency, + * emitting results in completion order. + * + * Accepts a union of task types (e.g. `Task | Task`) and emits + * the corresponding union of results. + * + * @param run - A {@link Runner} returned by {@link workers}. + * @param items - Iterable or async iterable of inert tasks to dispatch. Use {@link inert} to yield tasks from an async generator without triggering auto-await. + * @param opts - Concurrency and cancellation options. + * @returns Async iterable of task results in completion order. + */ +export function map>( + run: Runner, + items: Iterable | AsyncIterable, + opts?: MapOptions, +): AsyncIterable> { + return Readable.from(items).map(run, opts) as AsyncIterable>; +} diff --git a/test/fixtures/map.ts b/test/fixtures/map.ts new file mode 100644 index 0000000..78edba9 --- /dev/null +++ b/test/fixtures/map.ts @@ -0,0 +1,17 @@ +import { setTimeout } from 'node:timers/promises'; +import { mo } from 'moroutine'; + +export const delayedSquare = mo(import.meta, async (n: number, ms: number): Promise => { + await setTimeout(ms); + return n * n; +}); + +export const toUpper = mo(import.meta, (s: string): string => s.toUpperCase()); + +export const waitAborted = mo(import.meta, async (n: number, signal: AbortSignal): Promise => { + await new Promise((resolve) => { + if (signal.aborted) return resolve(); + signal.addEventListener('abort', () => resolve()); + }); + return n; +}); diff --git a/test/map.test.ts b/test/map.test.ts new file mode 100644 index 0000000..6deb73d --- /dev/null +++ b/test/map.test.ts @@ -0,0 +1,122 @@ +import { describe, it } from 'node:test'; +import assert from 'node:assert/strict'; +import { inert, map, workers } from 'moroutine'; +import { delayedSquare, toUpper, waitAborted } from './fixtures/map.ts'; + +describe('map', () => { + it('dispatches tasks and yields results', async () => { + using run = workers(2); + async function* tasks() { + for (const n of [1, 2, 3, 4]) yield inert(delayedSquare(n, 10)); + } + const results: number[] = []; + for await (const r of map(run, tasks(), { concurrency: 2 })) { + results.push(r); + } + assert.deepEqual( + results.sort((a, b) => a - b), + [1, 4, 9, 16], + ); + }); + + it('respects the concurrency limit', async () => { + using run = workers(4); + let active = 0; + let peak = 0; + async function* tasks() { + for (let i = 0; i < 8; i++) { + active++; + peak = Math.max(peak, active); + yield inert(delayedSquare(i, 30)); + active--; + } + } + const results: number[] = []; + for await (const r of map(run, tasks(), { concurrency: 2 })) { + results.push(r); + } + assert.equal(results.length, 8); + assert.ok(peak <= 2, `expected peak in-flight <= 2, got ${peak}`); + }); + + it('yields the union of result types for mixed tasks', async () => { + using run = workers(2); + async function* tasks() { + yield inert(delayedSquare(3, 5)); + yield inert(toUpper('hi')); + yield inert(delayedSquare(4, 5)); + yield inert(toUpper('there')); + } + const results: Array = []; + for await (const r of map(run, tasks(), { concurrency: 2 })) { + results.push(r); + } + assert.equal(results.length, 4); + assert.equal(results.filter((r) => typeof r === 'number').length, 2); + assert.equal(results.filter((r) => typeof r === 'string').length, 2); + }); + + it('stops iteration when signal aborts', async () => { + using run = workers(2); + const ac = new AbortController(); + async function* tasks() { + for (let i = 0; i < 100; i++) yield inert(waitAborted(i, ac.signal)); + } + setTimeout(() => ac.abort(), 30); + const results: number[] = []; + await assert.rejects( + async () => { + for await (const r of map(run, tasks(), { concurrency: 4, signal: ac.signal })) { + results.push(r); + } + }, + (err: Error) => err.name === 'AbortError', + ); + }); + + it('accepts a sync iterable of tasks', async () => { + using run = workers(2); + const tasks = [1, 2, 3].map((n) => inert(delayedSquare(n, 5))); + const results: number[] = []; + for await (const r of map(run, tasks, { concurrency: 2 })) { + results.push(r); + } + assert.deepEqual( + results.sort((a, b) => a - b), + [1, 4, 9], + ); + }); + + it('accepts a sync generator of tasks', async () => { + using run = workers(2); + function* gen() { + for (const n of [2, 4, 6]) yield inert(delayedSquare(n, 5)); + } + const results: number[] = []; + for await (const r of map(run, gen(), { concurrency: 2 })) { + results.push(r); + } + assert.deepEqual( + results.sort((a, b) => a - b), + [4, 16, 36], + ); + }); + + it('defaults concurrency to 1', async () => { + using run = workers(2); + const order: number[] = []; + async function* tasks() { + for (const n of [0, 1, 2]) { + order.push(n); + yield inert(delayedSquare(n, 20)); + } + } + const results: number[] = []; + for await (const r of map(run, tasks())) { + results.push(r); + } + // With concurrency=1, generator shouldn't get ahead of consumption — all 3 produced sequentially + assert.deepEqual(results, [0, 1, 4]); + assert.deepEqual(order, [0, 1, 2]); + }); +}); -- 2.51.2