moroutine
Offload functions to worker threads with shared memory primitives for Node.js.
Star
Fork
Watch
253 4 9
Files
.changeset
.tangled
docs
examples
src
test
.gitignore
.node-version
.nvmrc
.prettierignore
.prettierrc
CHANGELOG.md
LICENSE
package.json
pnpm-lock.yaml
README.md
tsconfig.build.json
tsconfig.json
README.md

moroutine #

Offload functions to worker threads with shared memory primitives for Node.js.

Installation #

npm install moroutine

Requires Node.js v24+.

Quick Start #

// main.ts
import { isPrime } from './is-prime.ts';

const result = await isPrime(999_999_937);
console.log(result); // true

// is-prime.ts
import { mo } from 'moroutine';

export const isPrime = mo(import.meta, (n: number): boolean => {
  if (n < 2) return false;
  for (let i = 2; i * i <= n; i++) {
    if (n % i === 0) return false;
  }
  return true;
});

Define a function with mo() in its own module, then import and run it on a worker pool. Moroutine modules must be side-effect free — workers import them to find the registered functions.

Core API #

mo(import.meta, fn) #

Wraps a function so it runs on a worker thread. The function must be defined at module scope (not dynamically).

// math.ts
import { mo } from 'moroutine';

export const add = mo(import.meta, (a: number, b: number): number => {
  return a + b;
});

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 least-busy scheduling when arguments are omitted.

import { workers } from 'moroutine';
import { add } from './math.ts';

{
  using run = workers(2);

  const result = await run(add(3, 4)); // single task
  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).

// 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:

// 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.

The Global Runtime #

Awaiting a task directly (without a pool) runs it on the global runtime — a process-wide pool that boots lazily on first dispatch and winds down automatically: idle runtime workers don't keep the process alive.

const result = await add(3, 4); // boots the global runtime on first use

Configure it by registering a runtime definition — a define()'d config in its own side-effect-free module — before the first dispatch:

// app-runtime.ts
import { define } from 'moroutine';
import type { RuntimeDefinition } from 'moroutine';

export default define<RuntimeDefinition>(import.meta, {
  size: 8,
});

// main.ts
import { registerRuntime, runtime } from 'moroutine';
import appRuntime from './app-runtime.ts';
import { add } from './math.ts';

registerRuntime(appRuntime); // must precede the first dispatch

const result = await runtime.run(add(3, 4)); // explicit form
const same = await add(3, 4); // bare await — same runtime

runtime is the ambient handle to the global runtime: runtime.run(task | tasks) dispatches (booting if needed), runtime.workers exposes uniform worker handles for pinning with assign(), and runtime.signal fires when shutdown begins. registerRuntime() after the runtime has booted throws — no silent reconfiguration. Bare dispatch and runtime.run() work on the main thread and on runtime workers alike — a moroutine running on the runtime can fan out nested work to the whole pool (including itself), with placement decided by the same balancer everywhere. Worker threads that are not part of the runtime (plain workers() pool workers) cannot dispatch to it; membership is a hard boundary.

// tree-sum.ts
import { mo, runtime } from 'moroutine';

export const treeSum = mo(import.meta, async (values: number[]): Promise<number> => {
  if (values.length <= 2) return values.reduce((a, b) => a + b, 0);
  const mid = values.length / 2;
  // Nested dispatch from inside a runtime worker — fork-join across the pool.
  const [left, right] = await runtime.run([treeSum(values.slice(0, mid)), treeSum(values.slice(mid))]);
  return left + right;
});

// main.ts
import { treeSum } from './tree-sum.ts';

console.log(await treeSum([1, 2, 3, 4, 5, 6, 7, 8])); // 36

Long-running apps (servers) can drain gracefully:

await runtime.shutdown(); // fires signal, awaits in-flight work, terminates workers

Scripts never need to call it — idle runtime workers don't keep the process alive.

Task-Args #

Pass a task as an argument to another task. The result is resolved on the worker and cached, so it never crosses back to the main thread. This is useful for non-transferable context like a database connection.

// db.ts
import { DatabaseSync } from 'node:sqlite';
import { mo } from 'moroutine';

export const openDb = mo(import.meta, (filename: string): DatabaseSync => {
  return new DatabaseSync(filename);
});

export const exec = mo(import.meta, (db: DatabaseSync, sql: string): void => {
  db.exec(sql);
});

export const query = mo(import.meta, (db: DatabaseSync, sql: string): unknown[] => {
  return db.prepare(sql).all();
});

import { workers } from 'moroutine';
import { openDb, exec, query } from './db.ts';

const db = openDb(':memory:');

{
  using run = workers(1);
  await run(exec(db, `CREATE TABLE users (id INTEGER PRIMARY KEY, name TEXT)`));
  await run(exec(db, `INSERT INTO users (name) VALUES ('Alice')`));
  const rows = await run(query(db, 'SELECT * FROM users')); // [{ id: 1, name: 'Alice' }]
}

openDb() returns a Task<DatabaseSync>, and exec()/query() accept it in place of DatabaseSync. The database is opened once on the worker and reused for every subsequent call — the main thread never touches it.

Server Threads #

Scale a Node HTTP server across worker threads by fanning out socket fds. Main accepts TCP connections; workers each run their own http.Server and handle requests. Imported from the moroutine/serve subpath.

// app-server.ts
import { createServer } from 'node:http';
import { mo } from 'moroutine';
import { listen, type ListenArgs } from 'moroutine/serve';

export const runServer = mo(import.meta, async (...args: ListenArgs): Promise<void> => {
  const srv = createServer((req, res) => {
    res.writeHead(200);
    res.end('hello');
  });
  await listen(srv, ...args);
});

// main.ts
import { once } from 'node:events';
import { createServer } from 'node:net';
import { workers, assign } from 'moroutine';
import { serverThreads } from 'moroutine/serve';
import { runServer } from './app-server.ts';

const server = createServer();

{
  await using run = workers(4);
  using threads = serverThreads(run.workers, server);
  const fanout = threads.map(([w, args]) => assign(w, runServer(...args)));
  void run(fanout);

  await once(server.listen(3000), 'listening');
  console.log('Listening on :3000');
  await once(process, 'SIGINT');
  server.close();
}

Connection routing defaults to leastConns(); pass { balance: roundRobin() } for simple fan-out. TLS termination runs on workers (https.createServer / tls.createServer inside your moroutine); the main-thread listener stays raw TCP. POSIX-only (Linux, macOS). Frameworks that expose an underlying http.Server compose without glue:

// express
const srv = createServer(express());
await listen(srv, ...args);

// fastify
const app = fastify();
await app.ready();
await listen(app.server, ...args);

Options:

serverThreads(run.workers, server, {
  balance: leastConns(), // or roundRobin(); default: leastConns()
  highWaterMark: 64, // per-worker cross-thread buffer (default 64)
  listen: { drainTimeout: 30_000 }, // per-worker drain budget on shutdown
});

highWaterMark bounds how many fds are in-transit to each worker's MessagePort at once. When a worker can't keep up, the cross-thread pipe parks until it catches up.

Graceful shutdown: call server.close() on the main listener; the cascade closes worker channels, drains in-flight requests (up to drainTimeout, default 30s), and resolves the moroutines. Combine with await using run = workers(...) for full pool teardown.

See examples/server-threads for a runnable demo.

Streaming #

Streaming Moroutines #

Wrap an async function* with mo() to create a streaming moroutine. Values are streamed between threads via MessageChannel with atomics-based backpressure — the producer parks when highWaterMark items (default 16) sit in flight and resumes when the consumer drains below the cap.

// count.ts
import { mo } from 'moroutine';

export const countUp = mo(import.meta, async function* (n: number) {
  for (let i = 0; i < n; i++) {
    yield i;
  }
});

Iterate directly (global runtime) or dispatch via a pool:

import { workers } from 'moroutine';
import { countUp } from './count.ts';

// Global runtime
for await (const n of countUp(5)) {
  console.log(n); // 0, 1, 2, 3, 4
}

// Worker pool
{
  using run = workers(2);
  for await (const n of run(countUp(5))) {
    console.log(n); // 0, 1, 2, 3, 4
  }
}

channel() and Fan-out #

When you pass the same AsyncIterable or streaming task argument to multiple tasks, each task gets its own copy of the data. Use channel() to share a single source across multiple workers — each item goes to exactly one consumer (work stealing).

import { workers, channel, assign, mo } from 'moroutine';

const generate = mo(import.meta, async function* (n: number) {
  for (let i = 0; i < n; i++) yield i;
});

const process = mo(import.meta, async (input: AsyncIterable<number>): Promise<number[]> => {
  const items: number[] = [];
  for await (const n of input) items.push(n);
  return items;
});

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
}

Use assign(worker, task) to pin a task to a specific worker. run.workers is a read-only array of worker handles, one per pool worker.

Without channel(), AsyncIterable and streaming task arguments are auto-detected and streamed to a single consumer. An undispatched streaming task used as a task arg or channel() source is a bare dispatch — it runs on the global runtime. 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 — a bare task in the stream would be dispatched to the global runtime rather than passed through.

// 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<Task<FileHash>> {
  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));
    }
  }
}

// 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<FileHash> => {
  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<string> | Task<number> 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. Stages are placed by the runtime's balancer; pin stages with assign() when placement matters.

const doubled = double(generate(5));
const squared = square(doubled);
for await (const n of squared) {
  console.log(n);
}

Transfers #

Use transfer() for zero-copy movement of ArrayBuffer, TypedArray, MessagePort, or streams.

import { transfer } from 'moroutine';

const buf = new ArrayBuffer(1024);
await run(processData(transfer(buf)));
// buf is now detached (zero-length) — ownership moved to worker

Return values from workers are auto-transferred when possible.

Shared Memory #

Descriptors and shared() #

Shared-memory types are created with descriptor functions or the shared() allocator.

import { shared, int32, bool, mutex, string, bytes } from 'moroutine';

Primitives #

const counter = int32(); // standalone Int32
const flag = bool(); // standalone Bool
const big = int64(); // standalone Int64 (bigint)

Atomics #

Atomic variants use Atomics.* for thread-safe operations without a lock.

const counter = int32atomic();
counter.add(1); // atomic increment, returns previous value
counter.load(); // atomic read

Full atomic operations: load, store, add, sub, and, or, xor, exchange, compareExchange.

Int32Atomic additionally exposes futex-style wait / wake:

const slot = int32atomic();

// In one thread — wait until the slot's value differs from 0,
// with an optional timeout.
await slot.waitAsync(0); // 'ok' | 'not-equal' | 'timed-out'
await slot.waitAsync(0, 100 /*ms*/); // with timeout

// In another thread — change the value and wake waiters.
slot.store(1);
slot.notify(); // wakes all waiters on this slot
slot.notify(2); // wakes at most 2

waitAsync returns 'not-equal' synchronously (no microtask hop) if the slot already holds a value different from expected. The underlying Atomics.wait/notify futex is available only on Int32Atomic.

Structs #

Plain objects in shared() create structs backed by a single SharedArrayBuffer.

const point = shared({ x: int32, y: int32 });

point.load(); // { x: 0, y: 0 }
point.store({ x: 10, y: 20 });
point.fields.x.store(10); // direct field access

Structs nest:

const rect = shared({
  pos: { x: int32, y: int32 },
  size: { w: int32, h: int32 },
});

Tuples #

Arrays in shared() create fixed-length tuples.

const pair = shared([int32, bool]);
pair.load(); // [0, false]
pair.store([42, true]);
pair.elements[0].store(99);

Bytes and Strings #

const buf = bytes(32); // fixed 32-byte buffer
buf.store(new Uint8Array(32)); // exact length required
buf.load(); // Readonly<Uint8Array> view
buf.view[0] = 0xff; // direct mutable access

const name = string(64); // UTF-8, max 64 bytes
name.store('hello');
name.load(); // 'hello'

Value Shorthand #

Primitive values in schemas infer their type.

shared(0); // Int32 initialized to 0
shared(true); // Bool initialized to true
shared(0n); // Int64 initialized to 0n
shared({ x: 10, y: 20 }); // struct with Int32 fields

Locks #

Mutex #

const mu = mutex();

using guard = await mu.lock();
// exclusive access
// auto-unlocks when guard is disposed

// or manually:
await mu.lock();
mu.unlock();

// Non-blocking attempt — returns null if held elsewhere.
using guard = mu.tryLock();
if (!guard) return;

RwLock #

const rw = rwlock();

using guard = await rw.readLock(); // multiple readers OK
using guard = await rw.writeLock(); // exclusive access

// Non-blocking variants — return null if unavailable.
using r = rw.tryReadLock(); // null if write-locked
using w = rw.tryWriteLock(); // null if any lock is held

Using with Workers #

Shared-memory types pass through postMessage automatically. They're reconstructed on the worker side with the same shared backing memory.

// update-position.ts
import { mo } from 'moroutine';
import type { Mutex, SharedStruct, Int32 } from 'moroutine';

type Position = SharedStruct<{ x: Int32; y: Int32 }>;

export const updatePosition = mo(
  import.meta,
  async (mu: Mutex, pos: Position, dx: number, dy: number): Promise<void> => {
    using guard = await mu.lock();
    const current = pos.load();
    pos.store({ x: current.x + dx, y: current.y + dy });
  },
);

// main.ts
import { workers, shared, int32, mutex } from 'moroutine';
import { updatePosition } from './update-position.ts';

const mu = mutex();
const pos = shared({ x: int32, y: int32 });

{
  using run = workers(4);
  await run([updatePosition(mu, pos, 1, 0), updatePosition(mu, pos, 0, 1)]);
}

console.log(pos.load()); // { x: 1, y: 1 }

Graceful Shutdown #

Use await using for graceful async shutdown. The pool exposes a signal that fires when disposal begins — thread it into tasks for cooperative cancellation.

{
  await using run = workers(4);

  run(longTask(run.signal)); // task can react to abort
  run(otherTask()); // runs to completion
}
// signal fired, waited for both tasks, then workers terminated

Use using (without await) for immediate termination, same as before.

A shutdownTimeout option force-terminates workers if graceful shutdown takes too long:

{
  await using run = workers(4, { shutdownTimeout: 5000 });
  // ...
}

Load Balancing #

The pool uses least-busy scheduling by default. Pass a balance option to change the strategy:

import { workers, roundRobin } from 'moroutine';

{
  using run = workers(4, { balance: roundRobin() });
  // tasks dispatched to workers in cyclic order
}

Built-in balancers:

  • 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. Stateless strategies just implement select:

import type { Balancer } from 'moroutine';

const random: Balancer = {
  select(workers, task) {
    return workers[Math.floor(Math.random() * workers.length)];
  },
};

Stateful strategies produce their state once via initialState(), and select receives it 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:

import { uint32atomic } from 'moroutine';
import type { Balancer, Uint32Atomic } from 'moroutine';

const everyOther: Balancer<Uint32Atomic> = {
  initialState: () => 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:

import { isTask, roundRobin } from 'moroutine';
import type { Balancer, RoundRobinState, Task, WorkerHandle } from 'moroutine';
import { increment, read } from './counter.ts';

// 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() as a shared-memory value stays consistent even when
// scheduling moves to multiple threads.
export function keyAffinity(): Balancer<RoundRobinState> {
  const fallback = roundRobin();
  return {
    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];
      if (key === undefined) return fallback.select(workers, task, fallbackState);
      return workers[hash(key) % workers.length];
    },
  };
}

Inside the isTask branch, task.args is typed as the moroutine's argument tuple (e.g. [key: string, n: number] for increment). See examples/worker-affinity for the full demo, including per-worker state that only stays consistent under affinity routing.

Examples #

All examples require Node v24+ and can be run directly, e.g. node examples/primes/main.ts.