Something went wrong. Try again.
Offload functions to worker threads with shared memory primitives for Node.js.
Something went wrong. Try again.
TypeScript
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146import { isMainThread } from 'node:worker_threads';import { workers } from './worker-pool.ts';import { isSharedValue } from './shared/reconstruct.ts';import { getDefineId } from './define.ts';import { defaultRuntime } from './runtime-definition.ts';import type { RuntimeDefinition } from './runtime-definition.ts';import type { Task, TaskResults, RunResult, Runner, WorkerHandle } from './runner.ts';import type { ChannelOptions } from './channel.ts';
// Process-wide singleton state, one runtime per process lifetime — no reset// hook by design; fresh-state tests use child processes.let registered: RuntimeDefinition | null = null;let pool: Runner | null = null;let shutdownStarted = false;let shutdownPromise: Promise<void> | null = null;
/** * Registers a custom runtime definition as the process's global runtime. * Must be called from the main thread before the first bare dispatch * (`await task`) or `runtime.run()` call — the runtime boots lazily on * first dispatch, and re-configuring a booted runtime would be a silent * behavior change, so this throws instead. */export function registerRuntime(definition: RuntimeDefinition): void { if (shutdownStarted) { throw new Error('Cannot registerRuntime(): the global runtime has been shut down.'); } if (pool !== null) { throw new Error( 'Cannot registerRuntime(): the global runtime has already booted. ' + 'Register before the first dispatch (bare `await task` or runtime.run()).', ); } if (registered !== null) { throw new Error('Cannot registerRuntime(): a runtime definition is already registered.'); } if (getDefineId(definition) === undefined) { throw new Error( 'registerRuntime() requires a define()-branded runtime definition ' + '(create it with define(import.meta, {...}) in a side-effect-free module).', ); } registered = definition;}
/** Boots (if needed) and returns the global runtime pool. @internal */export function getRuntime(): Runner { if (shutdownStarted) { throw new Error('The global runtime has been shut down.'); } if (pool === null) { if (!isMainThread) { throw new Error( 'The global runtime cannot boot on a worker thread yet — dispatch from the main thread. ' + '(Worker-side runtime participation is planned.)', ); } const def = registered ?? defaultRuntime; // Balancer state crosses the future worker handshake — validate at boot // that it can survive that trip intact (crash early, not corrupt later). const state = def.balance?.initialState?.(); if (state !== undefined && !isSharedValue(state) && !isPlainSerializable(state)) { throw new Error( 'Global runtime balancer state must be a single shared value (e.g. uint32atomic() ' + 'or shared({...})) or a plain serializable value — got a non-shared object, which ' + 'cannot stay consistent across scheduling threads.', ); } pool = def.size !== undefined ? workers(def.size, { balance: def.balance, shutdownTimeout: def.shutdownTimeout, refMode: 'lazy', balancerState: state, }) : workers({ balance: def.balance, shutdownTimeout: def.shutdownTimeout, refMode: 'lazy', balancerState: state, }); } return pool;}
function isPlainSerializable(value: unknown): boolean { // Primitives serialize fine; objects/functions (other than shared values, // checked by the caller) do not survive the worker handshake intact. return value === null || (typeof value !== 'object' && typeof value !== 'function');}
/** Returns the booted pool or throws — for surface that must not boot. */function bootedRuntime(): Runner { if (shutdownStarted) { throw new Error('The global runtime has been shut down.'); } if (pool === null) { throw new Error('The global runtime is not booted (it boots on first dispatch).'); } return pool;}
/** * The ambient global runtime handle. Task code imports this — never a specific * runtime definition module — so it stays portable across runtimes. * * Dispatch (`runtime.run(...)` or bare `await task`) boots the runtime on * first use; property access (`runtime.workers`, `runtime.signal`) never * boots and throws before boot. *//** Dispatches a batch of tasks in parallel to the global runtime and returns all results. */function run<T extends Task<any>[]>(tasks: [...T]): Promise<TaskResults<T>>;/** Dispatches a task to the global runtime. Returns `AsyncIterable<T>` for streaming tasks, `Promise<T>` otherwise. */function run<T>(task: Task<T>, opts?: ChannelOptions): RunResult<T>;function run(taskOrTasks: Task<any> | Task<any>[], opts?: ChannelOptions): any { return getRuntime()(taskOrTasks as any, opts);}
export const runtime = { /** Dispatches a task, batch, or streaming task to the global runtime. Boots it on first use. */ run, /** Uniform worker handles, one per runtime worker. Throws before boot. */ get workers(): readonly WorkerHandle[] { return bootedRuntime().workers; }, /** Fires when shutdown begins. Throws before boot. */ get signal(): AbortSignal { return bootedRuntime().signal; }, /** * Graceful opt-in teardown: fires `signal`, awaits in-flight tasks (up to * the definition's `shutdownTimeout`), then terminates workers. After * shutdown, dispatch throws; the runtime cannot be re-registered or * re-booted (one runtime per process lifetime). */ async shutdown(): Promise<void> { shutdownStarted = true; if (pool === null) return; // never booted: nothing to drain, lock the door shutdownPromise ??= pool[Symbol.asyncDispose](); await shutdownPromise; },};
export type { RuntimeDefinition };