Offload functions to worker threads with shared memory primitives for Node.js.
test: cross-worker tests for AtomicInt32, Mutex, and RwLock
Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
+80
+28
test/fixtures/sync.ts
+28
test/fixtures/sync.ts
···
1
+
import { mo } from 'moroutine';
2
+
import type { AtomicInt32, Mutex, RwLock } from 'moroutine';
3
+
4
+
export const atomicAdd = mo(import.meta, (counter: AtomicInt32, value: number): number => {
5
+
return counter.add(value);
6
+
});
7
+
8
+
export const mutexIncrement = mo(
9
+
import.meta,
10
+
async (mutex: Mutex, counter: AtomicInt32, times: number): Promise<void> => {
11
+
for (let i = 0; i < times; i++) {
12
+
const guard = await mutex.lock();
13
+
const val = counter.load();
14
+
counter.store(val + 1);
15
+
mutex.unlock();
16
+
}
17
+
},
18
+
);
19
+
20
+
export const rwlockRead = mo(
21
+
import.meta,
22
+
async (rwlock: RwLock, counter: AtomicInt32): Promise<number> => {
23
+
const guard = await rwlock.readLock();
24
+
const val = counter.load();
25
+
rwlock.readUnlock();
26
+
return val;
27
+
},
28
+
);
+52
test/sync/cross-worker.test.ts
+52
test/sync/cross-worker.test.ts
···
1
+
import { describe, it } from 'node:test';
2
+
import assert from 'node:assert/strict';
3
+
import { workerPool, AtomicInt32, Mutex, RwLock } from 'moroutine';
4
+
import { atomicAdd, mutexIncrement, rwlockRead } from '../fixtures/sync.ts';
5
+
6
+
describe('sync primitives across workers', () => {
7
+
it('AtomicInt32 is shared across worker threads', async () => {
8
+
const counter = new AtomicInt32();
9
+
const pool = workerPool(2);
10
+
try {
11
+
await Promise.all([
12
+
pool(atomicAdd(counter, 10)),
13
+
pool(atomicAdd(counter, 20)),
14
+
]);
15
+
assert.equal(counter.load(), 30);
16
+
} finally {
17
+
pool[Symbol.dispose]();
18
+
}
19
+
});
20
+
21
+
it('Mutex serializes access across workers', async () => {
22
+
const mutex = new Mutex();
23
+
const counter = new AtomicInt32();
24
+
const pool = workerPool(2);
25
+
try {
26
+
await Promise.all([
27
+
pool(mutexIncrement(mutex, counter, 100)),
28
+
pool(mutexIncrement(mutex, counter, 100)),
29
+
]);
30
+
assert.equal(counter.load(), 200);
31
+
} finally {
32
+
pool[Symbol.dispose]();
33
+
}
34
+
});
35
+
36
+
it('RwLock allows concurrent reads from workers', async () => {
37
+
const rwlock = new RwLock();
38
+
const counter = new AtomicInt32();
39
+
counter.store(42);
40
+
const pool = workerPool(2);
41
+
try {
42
+
const [a, b] = await Promise.all([
43
+
pool(rwlockRead(rwlock, counter)),
44
+
pool(rwlockRead(rwlock, counter)),
45
+
]);
46
+
assert.equal(a, 42);
47
+
assert.equal(b, 42);
48
+
} finally {
49
+
pool[Symbol.dispose]();
50
+
}
51
+
});
52
+
});