Worker Mutex

Re-entrant mutex for Node.js worker_threads based on SharedArrayBuffer + Atomics.

- Works across workers and the main thread.
- Supports recursive lock by the same thread.
- Supports blocking (lock) and non-blocking (lockAsync) modes.
- Uses one SharedArrayBuffer per mutex.

---

Installation

``bash
npm install worker-mutex
`

---

Quick start

`ts
import { WorkerMutex } from 'worker-mutex';

const shared = WorkerMutex.createSharedBuffer();
const mutex = new WorkerMutex(shared);

mutex.lock();
try {
mutex.lock(); // re-entrant lock (same thread)
try {
// critical section
} finally {
mutex.unlock();
}
} finally {
mutex.unlock();
}
`

---


Quick start with worker_threads


`ts
// main.ts (transpile with "module": "CommonJS")
import * as path from 'path';
import { Worker } from 'worker_threads';
import { WorkerMutex } from 'worker-mutex';

const mutexBuffer = WorkerMutex.createSharedBuffer();
const counterBuffer = new SharedArrayBuffer(Int32Array.BYTES_PER_ELEMENT);
const counter = new Int32Array(counterBuffer);

function runWorker() {
return new Promise((resolve, reject) => {
const worker = new Worker(path.join(__dirname, 'worker.js'), {
workerData: { mutexBuffer, counterBuffer },
});
WorkerMutex.bindWorkerExit(worker, mutexBuffer);

worker.once('error', reject);
worker.once('exit', (code) => {
if (code !== 0) {
reject(new Error(Worker exited with code ${code}));
return;
}

resolve();
});
});
}

Promise.all(Array.from({ length: 4 }, () => runWorker()))
.then(() => {
console.log(counter[0]); // expected: 40000
})
.catch((error) => {
console.error(error);
process.exitCode = 1;
});
`

`ts
// worker.ts (runtime file is worker.js after transpile)
import { workerData } from 'worker_threads';
import { WorkerMutex } from 'worker-mutex';

const mutex = new WorkerMutex(workerData.mutexBuffer);
const counter = new Int32Array(workerData.counterBuffer);

for (let i = 0; i < 10_000; i += 1) {
mutex.lock();
try {
counter[0] += 1;
} finally {
mutex.unlock();
}
}
`

---


Memory layout


Each mutex occupies 3 Int32 slots in the shared buffer:

1. flag (0 = unlocked, 1 = locked)
2. owner (threadId of the owning thread; meaningful only when flag = 1)
3. recursionCount (current re-entrant depth)

createSharedBuffer() allocates 3 * Int32Array.BYTES_PER_ELEMENT bytes.

---


API reference

$3

Creates a shared buffer for a single mutex.

$3

Binds automatic stale-lock cleanup to worker termination.

- On worker exit, if mutex is still locked and owner === worker.threadId,
mutex state is force-reset (flag/owner/recursionCount -> 0) and waiters are notified.
- Returns unsubscribe function for optional early detach; after exit, manual detach is not required (once listener).
- worker must support once('exit', ...) and have positive integer threadId,
otherwise WORKER_INSTANCE_MUST_SUPPORT_EXIT_EVENT or
WORKER_THREAD_ID_MUST_BE_A_POSITIVE_INTEGER is thrown.
- Binding to an already terminated worker throws WORKER_IS_ALREADY_EXITED.
- sharedBuffer validation is the same as constructor validation.

$3

Creates a mutex over an existing shared buffer.

- sharedBuffer must be a SharedArrayBuffer;
otherwise HANDLE_MUST_BE_A_SHARED_ARRAY_BUFFER is thrown.
- sharedBuffer.byteLength must match one mutex layout (3 * Int32);
otherwise MUTEX_BUFFER_SIZE_MUST_MATCH_SINGLE_MUTEX is thrown.

$3

Blocking lock.

- If mutex is free, acquires it.
- If current thread already owns it, increases recursion depth.
- Otherwise waits using Atomics.wait.

$3

Non-blocking lock for async flows.

- Uses Atomics.waitAsync when available.
- Falls back to soft backoff with setTimeout when waitAsync is not available.

$3

Unlocks one recursion level.

- Throws if current thread is not the owner.
- Fully releases mutex only when recursion depth reaches 0.

$3


Returns original SharedArrayBuffer.

---

Errors

All custom errors are instances of WorkerMutexError.

Possible error codes:

- HANDLE_MUST_BE_A_SHARED_ARRAY_BUFFER
- MUTEX_BUFFER_SIZE_MUST_MATCH_SINGLE_MUTEX
- MUTEX_IS_NOT_OWNED_BY_CURRENT_THREAD
- MUTEX_RECURSION_COUNT_UNDERFLOW
- MUTEX_RECURSION_COUNT_OVERFLOW
(can be thrown by re-entrant lock()/lockAsync() when recursion depth reaches Int32 max)
- WORKER_INSTANCE_MUST_SUPPORT_EXIT_EVENT
- WORKER_THREAD_ID_MUST_BE_A_POSITIVE_INTEGER
- WORKER_IS_ALREADY_EXITED

---


Notes and limitations


- lock() is blocking and can pause the main thread event loop while waiting.
- On the first lock() call from the main thread, the library emits a process warning:
WORKER_MUTEX_LOCK_ON_MAIN_THREAD_BLOCKS_EVENT_LOOP.
- For crash-safe cleanup, call WorkerMutex.bindWorkerExit(worker, mutexBuffer) in code that creates workers.
- Fairness is not guaranteed under heavy contention.
- Always pair lock/lockAsync with unlock in try/finally`.

License

MIT