From 8fb9ac57b4fdfebe7356ecc6131d7ad738ddc41a Mon Sep 17 00:00:00 2001 From: Devin Ivy Date: Mon, 20 Jul 2026 07:32:06 +0000 Subject: [PATCH] docs: define(), balancer state, leastBusy default --- README.md | 87 ++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++--------------- .changeset/runtime-foundations.md | 14 ++++++++++++++ 2 file(s) changed, 86 insertion(s)(+), 15 deletion(s)(-) diff --git a/README.md b/README.md --- a/README.md +++ b/README.md @@ -52,7 +52,7 @@ ### `workers(size?, opts?)` -Creates a pool of worker threads. Returns a `Runner` that dispatches tasks. Disposable via `using` or `[Symbol.dispose]()`. Defaults to `os.availableParallelism()` workers and round-robin scheduling when arguments are omitted. +Creates a pool of worker threads. Returns a `Runner` that dispatches tasks. Disposable via `using` or `[Symbol.dispose]()`. Defaults to `os.availableParallelism()` workers and least-busy scheduling when arguments are omitted. ```ts import { workers } from 'moroutine'; @@ -65,6 +65,36 @@ const [a, b] = await run([add(1, 2), add(3, 4)]); // batch } ``` + +### `define(import.meta, value)` + +Tags an object with module identity so it crosses threads **by reference**: workers resolve +it by evaluating its module, like moroutine functions themselves. Useful for config objects, +lookup tables, and values containing non-serializable members (functions). + +```ts +// config.ts +import { define } from 'moroutine'; + +export const limits = define(import.meta, { maxBatch: 64, retry: 3 }); +``` + +Pass it to any moroutine — the reference crosses, and the worker resolves its own copy from +the module: + +```ts +// main.ts +import { workers } from 'moroutine'; +import { readLimits } from './read-limits.ts'; +import { limits } from './config.ts'; + +using run = workers(); +await run(readLimits(limits)); +``` + +Since the value is produced by evaluating the module on each thread, non-shared mutable +state inside it is per-thread. Cross-thread state must use shared memory types. Like `mo()`, +`define()` must be called at module scope in a side-effect-free module. ### Dedicated Workers @@ -529,51 +559,78 @@ ## Load Balancing -The pool uses round-robin scheduling by default. Pass a `balance` option to change the strategy: +The pool uses least-busy scheduling by default. Pass a `balance` option to change the strategy: ```ts -import { workers, leastBusy } from 'moroutine'; +import { workers, roundRobin } from 'moroutine'; { - using run = workers(4, { balance: leastBusy() }); - // tasks dispatched to whichever worker has the fewest in-flight tasks + using run = workers(4, { balance: roundRobin() }); + // tasks dispatched to workers in cyclic order } ``` Built-in balancers: -- `roundRobin()` - cycles through workers in order (default) -- `leastBusy()` - picks the worker with the lowest active task count +- `leastBusy()` - picks the worker with the lowest active task count (default) +- `roundRobin()` - cycles through workers in order; its cursor lives in shared memory -Custom balancers implement the `Balancer` interface: +Custom balancers implement the `Balancer` interface. Stateless strategies just implement +`select`: ```ts -import type { Balancer, WorkerHandle, Task } from 'moroutine'; +import type { Balancer } from 'moroutine'; const random: Balancer = { - select(workers: readonly WorkerHandle[], task: Task) { + select(workers, task) { return workers[Math.floor(Math.random() * workers.length)]; }, }; ``` -Each `WorkerHandle` exposes `activeCount` (in-flight tasks) and `thread` (the underlying `worker_threads.Worker`) for building custom strategies. +Stateful strategies produce their state once via `initialState()`, and `select` receives it +as a third argument. Put cross-thread state (counters, cursors) in shared memory types: + +```ts +import { uint32atomic } from 'moroutine'; +import type { Balancer, Uint32Atomic } from 'moroutine'; + +const everyOther: Balancer<{ cursor: Uint32Atomic }> = { + initialState: () => ({ cursor: uint32atomic() }), + select(workers, task, { cursor }) { + return workers[(cursor.add(1) * 2) % workers.length]; + }, +}; +``` + +Each `WorkerHandle` exposes `index` (its position in the pool) and `activeCount` (in-flight tasks) for building custom strategies. `isTask(moroutine, task)` narrows a task to the descriptor type produced by a specific moroutine — useful inside a balancer to route by task kind or by a key in the args. For example, a worker-affinity balancer can hash a shard key out of the args so that every call for the same key hits the worker that already has its state loaded: ```ts import { isTask, roundRobin } from 'moroutine'; -import type { Balancer } from 'moroutine'; +import type { Balancer, RoundRobinState, Task, WorkerHandle } from 'moroutine'; import { increment, read } from './counter.ts'; -export function keyAffinity(): Balancer { +// Routes tasks to workers based on a key in the task's args, so that +// per-worker state (caches, connections, compiled resources) can be reused +// across successive calls for the same key. Falls back to round-robin for +// any task that doesn't carry a routable key. +// +// The fallback's state is composed into keyAffinity's own initialState() +// rather than captured in a closure: closure state only stays consistent +// under single-threaded scheduling (a workers() pool), while state produced +// by initialState() and shared via shared-memory types stays consistent +// even when scheduling moves to multiple threads. +export function keyAffinity(): Balancer<{ fallbackState: RoundRobinState }> { const fallback = roundRobin(); return { - select(workers, task) { + initialState: () => ({ fallbackState: fallback.initialState() }), + select(workers: readonly WorkerHandle[], task: Task, { fallbackState }): WorkerHandle { let key: string | undefined; if (isTask(increment, task)) key = task.args[0]; else if (isTask(read, task)) key = task.args[0]; - if (key === undefined) return fallback.select(workers, task); + if (key === undefined) return fallback.select(workers, task, fallbackState); return workers[hash(key) % workers.length]; }, }; diff --git a/.changeset/runtime-foundations.md b/.changeset/runtime-foundations.md new file mode 100644 --- /dev/null +++ b/.changeset/runtime-foundations.md @@ -0,0 +1,14 @@ +--- +'moroutine': major +--- + +Runtime foundations: + +- New `define()`: module-identified values that cross threads by reference and resolve per + thread via module evaluation; supported as task args. +- **Breaking:** default balancer is now `leastBusy()` (was round-robin). +- **Breaking:** `Balancer` gains `initialState()` (required for stateful balancers) and + `select` receives the state as a third argument. Plain stateless `{ select }` balancers + remain valid. +- **Breaking:** `WorkerHandle.thread` removed; `WorkerHandle.index` added. Active counts + are SharedArrayBuffer-backed. -- tangled.sh