diff --git a/README.md b/README.md index 6e1200f..518e546 100644 --- a/README.md +++ b/README.md @@ -589,15 +589,17 @@ const random: Balancer = { ``` 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: +as a third argument. State that must stay consistent across scheduling threads should be a +single shared-memory value — a shared primitive like `uint32atomic()`, or a `shared({...})` +struct for compound state: ```ts import { uint32atomic } from 'moroutine'; import type { Balancer, Uint32Atomic } from 'moroutine'; -const everyOther: Balancer<{ cursor: Uint32Atomic }> = { - initialState: () => ({ cursor: uint32atomic() }), - select(workers, task, { cursor }) { +const everyOther: Balancer = { + initialState: () => uint32atomic(), + select(workers, task, cursor) { return workers[(cursor.add(1) * 2) % workers.length]; }, }; @@ -620,13 +622,13 @@ import { increment, read } from './counter.ts'; // 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 }> { +// by initialState() as a shared-memory value stays consistent even when +// scheduling moves to multiple threads. +export function keyAffinity(): Balancer { const fallback = roundRobin(); return { - initialState: () => ({ fallbackState: fallback.initialState() }), - select(workers: readonly WorkerHandle[], task: Task, { fallbackState }): WorkerHandle { + initialState: () => 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]; diff --git a/examples/worker-affinity/key-affinity.ts b/examples/worker-affinity/key-affinity.ts index cb875d4..fae5b4e 100644 --- a/examples/worker-affinity/key-affinity.ts +++ b/examples/worker-affinity/key-affinity.ts @@ -10,13 +10,13 @@ import { increment, read } from './counter.ts'; // 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 }> { +// by initialState() as a shared-memory value stays consistent even when +// scheduling moves to multiple threads. +export function keyAffinity(): Balancer { const fallback = roundRobin(); return { - initialState: () => ({ fallbackState: fallback.initialState() }), - select(workers: readonly WorkerHandle[], task: Task, { fallbackState }): WorkerHandle { + initialState: () => 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]; diff --git a/src/balancers.ts b/src/balancers.ts index f2eaa70..caa8bfc 100644 --- a/src/balancers.ts +++ b/src/balancers.ts @@ -3,7 +3,7 @@ import type { Uint32Atomic } from './shared/uint32-atomic.ts'; import type { Balancer, WorkerHandle } from './runner.ts'; /** Shared state for {@link roundRobin}: a cursor advanced atomically on every select. */ -export type RoundRobinState = { cursor: Uint32Atomic }; +export type RoundRobinState = Uint32Atomic; /** * Creates a round-robin balancer that cycles through workers in order. @@ -15,8 +15,8 @@ export type RoundRobinState = { cursor: Uint32Atomic }; */ export function roundRobin(): Balancer { return { - initialState: () => ({ cursor: uint32atomic() }), - select(workers: readonly WorkerHandle[], _task, { cursor }): WorkerHandle { + initialState: () => uint32atomic(), + select(workers: readonly WorkerHandle[], _task, cursor): WorkerHandle { return workers[cursor.add(1) % workers.length]; }, }; diff --git a/src/runner.ts b/src/runner.ts index 1a52138..08813b2 100644 --- a/src/runner.ts +++ b/src/runner.ts @@ -24,12 +24,14 @@ type TaskResults[]> = { /** * A load balancing strategy for choosing which worker runs a task. * - * `initialState()` produces the balancer's state once at pool/runtime boot. May - * contain shared memory (which crosses threads by reference); plain values ride - * along by serialization. Closure state is safe only where scheduling is - * single-threaded (a `workers()` pool); shared, declared state works everywhere. - * Required whenever `S` is non-`void`, so a stateful balancer can't omit it and - * crash at first dispatch. + * `initialState()` produces the balancer's state once at pool/runtime boot. + * State that must stay consistent across scheduling threads should be a single + * shared-memory value — a shared primitive like `uint32atomic()`, or a + * `shared({...})` struct for compound state — which crosses threads by + * reference. Closure state is safe only where scheduling is single-threaded + * (a `workers()` pool); shared state works everywhere. Required whenever `S` + * is non-`void`, so a stateful balancer can't omit it and crash at first + * dispatch. */ export type Balancer = { /** Choose a worker for the given task. Called synchronously on every dispatch. */