Offload functions to worker threads with shared memory primitives for Node.js.
feat: add examples and freeze module registry on first task dispatch
Restructure examples into separate directories so moroutine definitions
are isolated modules (preventing fork bombs from top-level side effects).
Freeze per-module mo() registrations once a task is dispatched, erroring
on any late dynamic mo() calls that would cause ID mismatches in workers.
Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
+153
-1
+10
examples/non-blocking/fibonacci.ts
+10
examples/non-blocking/fibonacci.ts
···
1
+
import { mo } from '../../src/index.ts';
2
+
3
+
export const fibonacci = mo(import.meta, (n: number): number => {
4
+
// Deliberately naive recursive fibonacci — CPU-heavy but bounded
5
+
function fib(x: number): number {
6
+
if (x <= 1) return x;
7
+
return fib(x - 1) + fib(x - 2);
8
+
}
9
+
return fib(n);
10
+
});
+30
examples/non-blocking/main.ts
+30
examples/non-blocking/main.ts
···
1
+
// Demonstrates that the main thread event loop stays responsive
2
+
// while heavy work runs on worker threads.
3
+
//
4
+
// Run: node --experimental-strip-types examples/non-blocking/main.ts
5
+
6
+
import { workerPool } from '../../src/index.ts';
7
+
import { fibonacci } from './fibonacci.ts';
8
+
9
+
// Tick a counter on the main thread to prove it's not blocked
10
+
let ticks = 0;
11
+
const interval = setInterval(() => {
12
+
ticks++;
13
+
process.stdout.write(`\r main thread tick #${ticks}`);
14
+
}, 100);
15
+
16
+
console.log('Computing fibonacci(42) on a worker pool...');
17
+
console.log('Meanwhile, the main thread keeps ticking:\n');
18
+
19
+
const pool = workerPool(2);
20
+
const start = performance.now();
21
+
const [a, b] = await Promise.all([
22
+
pool(fibonacci(42)),
23
+
pool(fibonacci(41)),
24
+
]);
25
+
const elapsed = (performance.now() - start).toFixed(0);
26
+
pool[Symbol.dispose]();
27
+
28
+
clearInterval(interval);
29
+
console.log(`\n\nResults: fib(42)=${a}, fib(41)=${b}`);
30
+
console.log(`Computed in ${elapsed}ms with ${ticks} main-thread ticks (not blocked!)`);
+8
examples/parallel-batch/heavy-work.ts
+8
examples/parallel-batch/heavy-work.ts
+30
examples/parallel-batch/main.ts
+30
examples/parallel-batch/main.ts
···
1
+
// Process a batch of work items in parallel using a worker pool.
2
+
// Compares sequential (dedicated worker) vs parallel (pool) execution.
3
+
//
4
+
// Run: node --experimental-strip-types examples/parallel-batch/main.ts
5
+
6
+
import { workerPool } from '../../src/index.ts';
7
+
import { heavyWork } from './heavy-work.ts';
8
+
9
+
const items = Array.from({ length: 20 }, (_, i) => i + 1);
10
+
11
+
// Sequential: each call awaits on the dedicated worker
12
+
console.log('Sequential (dedicated worker)...');
13
+
const seqStart = performance.now();
14
+
const seqResults = [];
15
+
for (const item of items) {
16
+
seqResults.push(await heavyWork(item));
17
+
}
18
+
const seqTime = (performance.now() - seqStart).toFixed(0);
19
+
console.log(` ${items.length} items in ${seqTime}ms\n`);
20
+
21
+
// Parallel: distribute across a pool of 4 workers
22
+
console.log('Parallel (worker pool, 4 workers)...');
23
+
const pool = workerPool(4);
24
+
const parStart = performance.now();
25
+
const parResults = await Promise.all(items.map((item) => pool(heavyWork(item))));
26
+
const parTime = (performance.now() - parStart).toFixed(0);
27
+
pool[Symbol.dispose]();
28
+
console.log(` ${items.length} items in ${parTime}ms\n`);
29
+
30
+
console.log(`Speedup: ~${(Number(seqTime) / Number(parTime)).toFixed(1)}x`);
+11
examples/primes/is-prime.ts
+11
examples/primes/is-prime.ts
···
1
+
import { mo } from '../../src/index.ts';
2
+
3
+
export const isPrime = mo(import.meta, (n: number): boolean => {
4
+
if (n < 2) return false;
5
+
if (n < 4) return true;
6
+
if (n % 2 === 0 || n % 3 === 0) return false;
7
+
for (let i = 5; i * i <= n; i += 6) {
8
+
if (n % i === 0 || n % (i + 2) === 0) return false;
9
+
}
10
+
return true;
11
+
});
+19
examples/primes/main.ts
+19
examples/primes/main.ts
···
1
+
// CPU-bound prime checking offloaded to a dedicated worker thread.
2
+
//
3
+
// Run: node --experimental-strip-types examples/primes/main.ts
4
+
5
+
import { isPrime } from './is-prime.ts';
6
+
7
+
const candidates = [
8
+
999_999_937,
9
+
1_000_000_007,
10
+
1_000_000_009,
11
+
1_000_000_021,
12
+
1_000_000_033,
13
+
999_999_938,
14
+
];
15
+
16
+
for (const n of candidates) {
17
+
const prime = await isPrime(n);
18
+
console.log(`${n} is ${prime ? 'prime' : 'not prime'}`);
19
+
}
+3
src/execute.ts
+3
src/execute.ts
···
1
1
import type { Worker } from 'node:worker_threads';
2
+
import { freezeModule } from './registry.ts';
2
3
3
4
let nextCallId = 0;
4
5
const pending = new Map<number, { resolve: (value: any) => void; reject: (reason: any) => void }>();
···
17
18
}
18
19
19
20
export function execute<T>(worker: Worker, id: string, args: unknown[]): Promise<T> {
21
+
const url = id.slice(0, id.lastIndexOf('#'));
22
+
freezeModule(url);
20
23
const callId = nextCallId++;
21
24
return new Promise<T>((resolve, reject) => {
22
25
pending.set(callId, { resolve, reject });
+9
-1
src/mo.ts
+9
-1
src/mo.ts
···
1
-
import { registry } from './registry.ts';
1
+
import { registry, isModuleFrozen } from './registry.ts';
2
2
import { Task } from './task.ts';
3
3
4
4
const counters = new Map<string, number>();
···
8
8
fn: (...args: A) => R,
9
9
): (...args: A) => Task<R> {
10
10
const url = importMeta.url;
11
+
12
+
if (isModuleFrozen(url)) {
13
+
throw new Error(
14
+
`Cannot call mo() for ${url} after a task from this module has been dispatched. ` +
15
+
'All mo() calls must happen at module load time.',
16
+
);
17
+
}
18
+
11
19
const index = counters.get(url) ?? 0;
12
20
counters.set(url, index + 1);
13
21
const id = `${url}#${index}`;
+10
src/registry.ts
+10
src/registry.ts
···
1
1
export const registry = new Map<string, (...args: any[]) => any>();
2
+
3
+
const frozen = new Set<string>();
4
+
5
+
export function freezeModule(url: string): void {
6
+
frozen.add(url);
7
+
}
8
+
9
+
export function isModuleFrozen(url: string): boolean {
10
+
return frozen.has(url);
11
+
}
+7
test/fixtures/freeze.ts
+7
test/fixtures/freeze.ts
+12
test/freeze.test.ts
+12
test/freeze.test.ts
···
1
+
import { describe, it } from 'node:test';
2
+
import assert from 'node:assert/strict';
3
+
import { double, lateRegistration } from './fixtures/freeze.ts';
4
+
5
+
describe('module freezing', () => {
6
+
it('throws when mo() is called after a task from that module has been dispatched', async () => {
7
+
await double(2);
8
+
assert.throws(lateRegistration, {
9
+
message: /Cannot call mo\(\).*after a task from this module has been dispatched/,
10
+
});
11
+
});
12
+
});