From 11eb72c6e9205f3b8833fcc9d88d81a321b8ee47 Mon Sep 17 00:00:00 2001 From: Devin Ivy Date: Mon, 13 Apr 2026 23:05:46 -0400 Subject: [PATCH] docs: per-worker dispatch implementation plan Co-Authored-By: Claude Opus 4.6 (1M context) --- .../plans/2026-04-13-per-worker-dispatch.md | 604 ++++++++++++++++++ 1 file changed, 604 insertions(+) create mode 100644 docs/superpowers/plans/2026-04-13-per-worker-dispatch.md diff --git a/docs/superpowers/plans/2026-04-13-per-worker-dispatch.md b/docs/superpowers/plans/2026-04-13-per-worker-dispatch.md new file mode 100644 index 0000000..68f81d0 --- /dev/null +++ b/docs/superpowers/plans/2026-04-13-per-worker-dispatch.md @@ -0,0 +1,604 @@ +# Per-Worker Dispatch Implementation Plan + +> **For agentic workers:** REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (`- [ ]`) syntax for tracking. + +**Goal:** Allow pinning tasks to specific workers via `assign()` and `run.workers`, removing the need to know worker count or assume round-robin scheduling. + +**Architecture:** `Task` and `StreamTask` gain an optional `worker` property. `assign(handle, task)` creates a copy with `worker` set. `workers()` builds a frozen `WorkerHandle[]` exposed on the runner. Dispatch checks `task.worker` — if set, dispatches to that worker; if not, round-robin as before. + +**Tech Stack:** TypeScript (erasable syntax only), worker_threads, node:test. + +--- + +### Task 1: Add `worker` Property to Task and StreamTask + +**Files:** +- Modify: `src/task.ts` +- Modify: `src/stream-task.ts` + +- [ ] **Step 1: Add optional worker property to Task** + +Read `src/task.ts`. Add a `worker` property. The type is `unknown` here to avoid circular imports (the `WorkerHandle` type lives in `src/runner.ts` which imports `Task`). The property is optional and unset by default. + +Add after the existing `readonly args: unknown[];` line: + +```ts + worker?: unknown; +``` + +- [ ] **Step 2: Add optional worker property to StreamTask** + +Read `src/stream-task.ts`. Add the same property after `readonly args: unknown[];`: + +```ts + worker?: unknown; +``` + +- [ ] **Step 3: Run type check** + +Run: `pnpm tsc --noEmit 2>&1` +Expected: No errors. + +- [ ] **Step 4: Commit** + +```bash +git add src/task.ts src/stream-task.ts +git commit -m "feat: add optional worker property to Task and StreamTask" +``` + +--- + +### Task 2: Define WorkerHandle and Update Runner Type + +**Files:** +- Modify: `src/runner.ts` +- Modify: `src/index.ts` + +- [ ] **Step 1: Add WorkerHandle and update Runner** + +Replace `src/runner.ts` with: + +```ts +import type { Task } from './task.ts'; +import type { StreamTask } from './stream-task.ts'; +import type { ChannelOptions } from './channel.ts'; + +type TaskResults[]> = { [K in keyof T]: T[K] extends Task ? R : never }; + +/** Options for configuring a worker pool. */ +export interface WorkerOptions { + /** Maximum time in ms to wait for in-flight tasks during async dispose. If exceeded, workers are force-terminated. */ + shutdownTimeout?: number; +} + +/** A handle to a specific worker in a pool. */ +export interface WorkerHandle { + /** Dispatches a task pinned to this worker. */ + exec(task: Task): Promise; + /** Dispatches a streaming task pinned to this worker. */ + exec(task: StreamTask, opts?: ChannelOptions): AsyncIterable; +} + +/** + * A callable that dispatches tasks to a worker pool. Disposable via `using` or `[Symbol.dispose]()`. + * + * @param task - A single {@link Task} to run on a worker. + * @returns `Promise` for a single task, `Promise<[...results]>` for a batch, or `AsyncIterable` for a streaming task. + */ +export type Runner = { + /** Dispatches a single task and returns its result. */ + (task: Task): Promise; + /** Dispatches a batch of tasks in parallel and returns all results. */ + []>(tasks: [...T]): Promise>; + /** Dispatches a streaming task and returns an async iterable of yielded values. */ + (task: StreamTask, opts?: ChannelOptions): AsyncIterable; + /** AbortSignal that fires when the pool starts disposing. */ + readonly signal: AbortSignal; + /** Read-only array of worker handles, one per pool worker. */ + readonly workers: readonly WorkerHandle[]; + /** Terminates all workers immediately. */ + [Symbol.dispose](): void; + /** Aborts signal, waits for in-flight tasks to settle, then terminates workers. */ + [Symbol.asyncDispose](): Promise; +}; +``` + +- [ ] **Step 2: Export WorkerHandle from index.ts** + +In `src/index.ts`, change: +```ts +export type { Runner, WorkerOptions } from './runner.ts'; +``` +to: +```ts +export type { Runner, WorkerHandle, WorkerOptions } from './runner.ts'; +``` + +- [ ] **Step 3: Run type check** + +Run: `pnpm tsc --noEmit 2>&1` +Expected: No errors. + +- [ ] **Step 4: Commit** + +```bash +git add src/runner.ts src/index.ts +git commit -m "feat: add WorkerHandle type and workers property to Runner" +``` + +--- + +### Task 3: Create `assign()` Function + +**Files:** +- Create: `src/assign.ts` +- Modify: `src/index.ts` +- Create: `test/assign.test.ts` + +- [ ] **Step 1: Write failing tests** + +Create `test/assign.test.ts`: + +```ts +import { describe, it } from 'node:test'; +import assert from 'node:assert/strict'; +import { mo, workers, assign } from 'moroutine'; + +const double = mo(import.meta, (n: number): number => n * 2); + +describe('assign()', () => { + it('returns a new task with worker set', () => { + const run = workers(1); + try { + const task = double(5); + const assigned = assign(run.workers[0], task); + assert.notStrictEqual(assigned, task); + assert.equal(assigned.id, task.id); + assert.deepEqual(assigned.args, task.args); + assert.notEqual(assigned.uid, task.uid); + assert.equal(assigned.worker, run.workers[0]); + } finally { + run[Symbol.dispose](); + } + }); + + it('does not modify the original task', () => { + const run = workers(1); + try { + const task = double(5); + assign(run.workers[0], task); + assert.equal(task.worker, undefined); + } finally { + run[Symbol.dispose](); + } + }); +}); +``` + +- [ ] **Step 2: Run test to verify it fails** + +Run: `node --no-warnings --test test/assign.test.ts 2>&1 | tail -10` +Expected: FAIL — `assign` not exported. + +- [ ] **Step 3: Implement assign()** + +Create `src/assign.ts`: + +```ts +import { Task } from './task.ts'; +import { StreamTask } from './stream-task.ts'; +import type { WorkerHandle } from './runner.ts'; + +/** + * Returns a copy of the task pinned to a specific worker. + * The original task is unchanged. + * @param worker - The worker handle to pin the task to. + * @param task - The task or streaming task to assign. + * @returns A new task with the same id and args, pinned to the given worker. + */ +export function assign(worker: WorkerHandle, task: Task): Task; +export function assign(worker: WorkerHandle, task: StreamTask): StreamTask; +export function assign(worker: WorkerHandle, task: Task | StreamTask): Task | StreamTask { + if (task instanceof StreamTask) { + const copy = new StreamTask(task.id, task.args); + copy.worker = worker; + return copy; + } + const copy = new Task(task.id, task.args); + copy.worker = worker; + return copy; +} +``` + +- [ ] **Step 4: Export assign from index.ts** + +Add to `src/index.ts`: +```ts +export { assign } from './assign.ts'; +``` + +- [ ] **Step 5: Run tests** + +Run: `node --no-warnings --test test/assign.test.ts 2>&1 | tail -10` +Expected: All tests pass. + +Run: `pnpm tsc --noEmit 2>&1` +Expected: No errors. + +- [ ] **Step 6: Commit** + +```bash +git add src/assign.ts src/index.ts test/assign.test.ts +git commit -m "feat: assign() for pinning tasks to specific workers" +``` + +--- + +### Task 4: Implement WorkerHandle and `run.workers` in worker-pool.ts + +**Files:** +- Modify: `src/worker-pool.ts` +- Create: `test/worker-handle.test.ts` +- Create: `test/fixtures/worker-handle.ts` + +- [ ] **Step 1: Write test fixtures** + +Create `test/fixtures/worker-handle.ts`: + +```ts +import { mo } from 'moroutine'; + +export const identity = mo(import.meta, (n: number): number => n); + +export const countUp = mo(import.meta, async function* (n: number) { + for (let i = 0; i < n; i++) yield i; +}); +``` + +- [ ] **Step 2: Write failing tests** + +Create `test/worker-handle.test.ts`: + +```ts +import { describe, it } from 'node:test'; +import assert from 'node:assert/strict'; +import { workers, assign, channel } from 'moroutine'; +import { identity, countUp } from './fixtures/worker-handle.ts'; + +describe('WorkerHandle', () => { + it('run.workers is a frozen array matching pool size', () => { + const run = workers(3); + try { + assert.equal(run.workers.length, 3); + assert.ok(Object.isFrozen(run.workers)); + } finally { + run[Symbol.dispose](); + } + }); + + it('w.exec() dispatches a task to the specific worker', async () => { + const run = workers(2); + try { + const result = await run.workers[0].exec(identity(42)); + assert.equal(result, 42); + } finally { + run[Symbol.dispose](); + } + }); + + it('w.exec() dispatches a streaming task', async () => { + const run = workers(1); + try { + const results: number[] = []; + for await (const n of run.workers[0].exec(countUp(3))) { + results.push(n); + } + assert.deepEqual(results, [0, 1, 2]); + } finally { + run[Symbol.dispose](); + } + }); + + it('assign() pins task to a specific worker via run()', async () => { + const run = workers(2); + try { + const task = assign(run.workers[1], identity(99)); + const result = await run(task); + assert.equal(result, 99); + } finally { + run[Symbol.dispose](); + } + }); + + it('assign() works in a batch', async () => { + const run = workers(2); + try { + const results = await run([ + assign(run.workers[0], identity(1)), + assign(run.workers[1], identity(2)), + ]); + assert.deepEqual(results, [1, 2]); + } finally { + run[Symbol.dispose](); + } + }); + + it('channel fan-out via assign + run.workers.map', async () => { + const run = workers(2); + try { + const ch = channel(countUp(20)); + const results: number[][] = await run( + run.workers.map((w) => { + return assign(w, identity(0)); + }), + ); + // Just verify it dispatches and returns — identity doesn't consume a channel + assert.equal(results.length, 2); + } finally { + run[Symbol.dispose](); + } + }); + + it('w.exec() rejects after dispose', async () => { + const run = workers(1); + run[Symbol.dispose](); + await assert.rejects(() => run.workers[0].exec(identity(1)), { message: /disposed/ }); + }); +}); +``` + +- [ ] **Step 3: Run test to verify it fails** + +Run: `node --no-warnings --test test/worker-handle.test.ts 2>&1 | tail -10` +Expected: FAIL — `run.workers` not defined yet. + +- [ ] **Step 4: Implement WorkerHandle and run.workers** + +Replace `src/worker-pool.ts` with: + +```ts +import { setTimeout } from 'node:timers/promises'; +import { Worker } from 'node:worker_threads'; +import { availableParallelism } from 'node:os'; +import { setupWorker, execute, dispatchStream } from './execute.ts'; +import { Task } from './task.ts'; +import { StreamTask } from './stream-task.ts'; +import type { ChannelOptions } from './channel.ts'; +import type { Runner, WorkerHandle, WorkerOptions } from './runner.ts'; + +const workerEntryUrl = new URL('./worker-entry.ts', import.meta.url); + +/** + * Creates a pool of worker threads that dispatch tasks with round-robin scheduling. + * @param size - Number of worker threads in the pool. Defaults to the number of available CPUs. + * @param opts - Optional configuration including shutdown timeout. + * @returns A disposable {@link Runner} for dispatching tasks. + */ +export function workers(size: number = availableParallelism(), opts?: WorkerOptions): Runner { + const pool: Worker[] = []; + for (let i = 0; i < size; i++) { + const worker = new Worker(workerEntryUrl); + setupWorker(worker); + worker.unref(); + pool.push(worker); + } + + let next = 0; + let disposed = false; + const ac = new AbortController(); + const inflight = new Set>(); + + function track(promise: Promise): Promise { + inflight.add(promise); + promise.finally(() => inflight.delete(promise)); + return promise; + } + + function terminateAll(): void { + for (const worker of pool) { + worker.terminate(); + } + pool.length = 0; + } + + function resolveWorker(task: Task | StreamTask): Worker { + if (task.worker != null) { + const idx = handles.indexOf(task.worker as WorkerHandle); + if (idx !== -1) return pool[idx]; + } + const worker = pool[next % pool.length]; + next++; + return worker; + } + + function dispatch(task: Task): Promise { + if (disposed) return Promise.reject(new Error('Worker pool is disposed')); + const worker = resolveWorker(task); + return track(execute(worker, task.id, task.args)); + } + + function dispatchStreamTask(task: StreamTask, channelOpts?: ChannelOptions): AsyncIterable { + if (disposed) throw new Error('Worker pool is disposed'); + const worker = resolveWorker(task); + const { iterable, done } = dispatchStream(worker, task.id, task.args, channelOpts); + track(done); + return iterable; + } + + const handles: WorkerHandle[] = pool.map((_, idx) => ({ + exec(task: Task | StreamTask, channelOpts?: ChannelOptions): any { + if (task instanceof StreamTask) { + if (disposed) throw new Error('Worker pool is disposed'); + const { iterable, done } = dispatchStream(pool[idx], task.id, task.args, channelOpts); + track(done); + return iterable; + } + if (disposed) return Promise.reject(new Error('Worker pool is disposed')); + return track(execute(pool[idx], task.id, task.args)); + }, + })); + Object.freeze(handles); + + const run: Runner = Object.assign( + (taskOrTasks: Task | Task[] | StreamTask, channelOpts?: ChannelOptions): any => { + if (taskOrTasks instanceof StreamTask) { + return dispatchStreamTask(taskOrTasks, channelOpts); + } + if (Array.isArray(taskOrTasks)) { + return Promise.all(taskOrTasks.map((t) => dispatch(t))); + } + return dispatch(taskOrTasks); + }, + { + get signal() { + return ac.signal; + }, + get workers(): readonly WorkerHandle[] { + return handles; + }, + [Symbol.dispose]() { + disposed = true; + ac.abort(); + terminateAll(); + }, + async [Symbol.asyncDispose]() { + disposed = true; + ac.abort(); + const settle = Promise.allSettled(inflight); + if (opts?.shutdownTimeout != null) { + const timeoutAc = new AbortController(); + await Promise.race([ + settle.finally(() => timeoutAc.abort()), + setTimeout(opts.shutdownTimeout, undefined, { signal: timeoutAc.signal }).catch(() => {}), + ]); + } else { + await settle; + } + terminateAll(); + }, + }, + ); + + return run; +} +``` + +- [ ] **Step 5: Run tests** + +Run: `node --no-warnings --test test/worker-handle.test.ts 2>&1 | tail -15` +Expected: All tests pass. + +Run: `pnpm test 2>&1 | tail -10` +Expected: All existing tests pass. + +Run: `pnpm tsc --noEmit 2>&1` +Expected: No errors. + +- [ ] **Step 6: Commit** + +```bash +git add src/worker-pool.ts test/worker-handle.test.ts test/fixtures/worker-handle.ts +git commit -m "feat: WorkerHandle, run.workers, and assign() dispatch support" +``` + +--- + +### Task 5: Update Channel-Fanout Example + +**Files:** +- Modify: `examples/channel-fanout/main.ts` + +- [ ] **Step 1: Update example to use assign + run.workers** + +Replace `examples/channel-fanout/main.ts` with: + +```ts +// Fan-out a single channel to multiple workers using work stealing. +// Each item goes to whichever worker is ready first. +// Requires Node v24+. +// +// Run: node examples/channel-fanout/main.ts + +import { workers, channel, assign } from '../../src/index.ts'; +import { generate, process } from './work.ts'; + +{ + using run = workers(); + const ch = channel(generate(200)); + const fanout = run.workers.map((w) => { + return assign(w, process(ch)); + }); + const results: number[][] = await run(fanout); + + for (let i = 0; i < results.length; i++) { + console.log(`Worker ${i}: processed ${results[i].length} items`); + } + + const all = results.flat().sort((a, b) => a - b); + console.log(`\nTotal: ${all.length} items, none lost, none duplicated`); +} +``` + +- [ ] **Step 2: Run example** + +Run: `node --no-warnings examples/channel-fanout/main.ts` +Expected: Prints worker counts and total, exits cleanly. + +- [ ] **Step 3: Commit** + +```bash +git add examples/channel-fanout/main.ts +git commit -m "refactor: channel-fanout example uses assign() and run.workers" +``` + +--- + +### Task 6: Update README + +**Files:** +- Modify: `README.md` + +- [ ] **Step 1: Add per-worker dispatch docs** + +Read `README.md`. Find the `channel()` and Fan-out section. Update the fan-out example to use `assign()` and `run.workers`. + +Replace the existing fan-out code example: + +```ts +const ch = channel(generate(100)); + +{ + using run = workers(4); + const [a, b, c, d] = await run([process(ch), process(ch), process(ch), process(ch)]); + // Items distributed across workers — no duplicates, no gaps +} +``` + +With: + +```ts +const ch = channel(generate(100)); + +{ + using run = workers(); + const fanout = run.workers.map((w) => { + return assign(w, process(ch)); + }); + const results = await run(fanout); + // Items distributed across workers — no duplicates, no gaps +} +``` + +Update the import line in the code block above it from `import { workers, channel, mo } from 'moroutine';` to `import { workers, channel, assign, mo } from 'moroutine';`. + +- [ ] **Step 2: Run type check** + +Run: `pnpm tsc --noEmit 2>&1` +Expected: No errors. + +- [ ] **Step 3: Commit** + +```bash +git add README.md +git commit -m "docs: update fan-out example to use assign() and run.workers" +``` -- 2.51.2