@nxtedition/shared

Cross-thread primitives for Node.js worker threads.

Install

npm install @nxtedition/shared

Requirements

• Node.js 26.1 or newer. The implementation uses Atomics.pause and the native binding relies on the experimental SharedArrayBuffer N-API entry points, including external backing-store finalization, available in the supported Node 26.1+ toolchain.
• CPU architecture with always-lock-free 32-bit atomics (x86-64, ARM64). The module verifies this at import via std::atomic<uint32_t>::is_always_lock_free and throws on unsupported platforms, because the ring buffer's lock-free protocol relies on non-tearing 32-bit loads/stores.
• Windows 10 1803 / Server 1607 or newer. The ring buffer's double-mapping uses VirtualAlloc2 / MapViewOfFile3, introduced in that release.

Prebuilt binaries are currently shipped for linux-x64. On other platforms the package builds the native addon from source on install, which requires a C++20 toolchain and node-gyp. node-gyp is not a dependency of this package: the install script (node-gyp-build) runs a node-gyp installed in the dependent project if there is one, and otherwise the node-gyp on PATH — npm, yarn and pnpm put their bundled node-gyp on the PATH while running install scripts. In environments that do neither, install node-gyp yourself.

API

SharedArrayBuffer registry

A process-wide, thread-safe registry for SharedArrayBuffers. Keys are strong references; backing stores are stored as weak references. If all JS references to a SharedArrayBuffer are garbage collected, the factory is called again on the next getOrCreate.

import { getOrCreate } from '@nxtedition/shared'

// Pass a byte size — a SharedArrayBuffer is created automatically
const sab = getOrCreate('my-buffer', 1024)

// Or pass a factory function for custom creation
const sab2 = getOrCreate('my-other-buffer', (key) => new SharedArrayBuffer(2048))

A practical use case is application-wide stats counters. Instead of plumbing parentPort.postMessage calls to propagate metrics from workers to the main thread, every thread can atomically update a shared counter directly:

// stats.js — import from any thread
import { getOrCreate } from '@nxtedition/shared'

const counters = new Int32Array(getOrCreate('app:stats', 4 * 4))

export const REQUESTS = 0
export const ERRORS = 1
export const BYTES_IN = 2
export const BYTES_OUT = 3

export function inc(index, delta = 1) {
  Atomics.add(counters, index, delta)
}

export function snapshot() {
  return {
    requests: Atomics.load(counters, REQUESTS),
    errors: Atomics.load(counters, ERRORS),
    bytesIn: Atomics.load(counters, BYTES_IN),
    bytesOut: Atomics.load(counters, BYTES_OUT),
  }
}

Any worker calls inc(REQUESTS) on the hot path; the main thread calls snapshot() to read all counters without any message passing overhead.

getOrCreate(key, sizeOrCallbackFn)

Returns an existing SharedArrayBuffer for key, or creates and registers one. Thread-safe; live lookups share the registry lock and first creation is coordinated per key.

• key string -- Registry key. Keys starting with __@nxtedition/shared/lock: are reserved for withLock.
• sizeOrCallbackFn number | (key: string) => SharedArrayBuffer -- Either a positive integer byte size (creates a new SharedArrayBuffer(size) automatically), or a factory function called when the key has no live entry.

With the size form, a RangeError is thrown if an entry already exists with a different byteLength. The factory form does no such validation — an existing entry is returned as-is, whatever its size.

Factories are serialized per key and run without holding the process-wide registry lock, so a slow factory does not block lookups or creation for other keys. Callers racing to create the same key wait for the first factory and, while its result remains live, receive the same backing store. If that factory throws or returns an invalid value, a waiting caller may retry with its own factory. Calling getOrCreate recursively from within a factory throws.

The registry and native ring handles are process-wide only within one loaded copy of the native addon. All participating threads must resolve the same physical @nxtedition/shared installation; deduplicate the package if your dependency graph would otherwise load multiple copies.

Cross-thread lock

A simple mutex built on SharedArrayBuffer and Atomics. Inspired by the Web Locks API but works across worker threads without message passing.

import { withLock } from '@nxtedition/shared'

const result = await withLock('my-resource', async () => {
  // exclusive access across all threads
  return doWork()
})

withLock(key, fn, opaque?, opts?)

Acquires a cross-thread lock identified by key, executes fn, and releases the lock. If the lock is held by another thread, waits asynchronously (via Atomics.waitAsync) until it becomes available. The lock is released when fn returns or throws.

• key string -- Lock name. The lock state lives in the registry under the reserved key prefix __@nxtedition/shared/lock: — do not create registry entries with that prefix.
• fn (opaque?) => T | PromiseLike<T> -- Function to execute under the lock.
• opaque (optional) -- Passed through to fn to avoid closures on hot paths.
• opts.signal AbortSignal (optional) -- Aborts the lock acquisition. If the signal fires while waiting, the promise rejects and the lock is not acquired.

Returns Promise<T> with the return value of fn.

Caveats:

• Options vs opaque. A lone 3rd argument is treated as opts only when fn declares no parameters. If fn has a parameter, a lone 3rd argument is the opaque value (any options in it are ignored) — pass opts as the 4th argument instead: withLock(key, fn, opaque, { signal }).
• Held locks die with their thread. The lock is only released by a live thread. If the holder is terminated or crashes inside fn, the lock is never released and all waiters wait forever. Do not terminate workers that may hold locks; pass signal (e.g. AbortSignal.timeout(ms)) to bound the wait.
• No fairness. Waiters re-contend on every release; under sustained contention an individual waiter can be starved indefinitely. Again, use signal to bound waiting time.
• Non-reentrant. Acquiring the same key from within its own callback (on the same thread) would deadlock, so it throws instead. Independent concurrent withLock calls on the same thread queue normally.
• Throws on corrupted state. If a registry entry already exists under the reserved key with the wrong size, a RangeError is thrown; if the lock word holds a value outside the internal unlocked, locked, and contended states, an error is thrown rather than waiting forever.

Ring buffer

A high-performance, lock-free ring buffer for inter-thread communication using SharedArrayBuffer.

A single SharedArrayBuffer is mapped into both threads. The writer appends messages by advancing a write pointer; the reader consumes them by advancing a read pointer. No copies, no ownership transfers, no cloning overhead. Reads are zero-copy: the reader callback receives a view directly into the shared buffer, and the delivered bytes stay valid until the current synchronous execution completes (see reader.readSome). Writes are batched -- the write pointer is only published after a high-water mark is reached or, via a deferred microtask, when the current synchronous execution completes.

import { Reader, Writer } from '@nxtedition/shared'

const w = new Writer(1024 * 1024) // 1 MB ring buffer

const payload = Buffer.from('hello world')
w.writeSync(payload.length, (data) => {
  payload.copy(data.buffer, data.byteOffset)
  return data.byteOffset + payload.length
})
w.flushSync() // writes publish in a deferred microtask — flush so a synchronous read sees them

// Pass w.handle to the other thread via workerData
const r = new Reader(w.handle)

r.readSome((data) => {
  const msg = data.buffer.subarray(data.byteOffset, data.byteOffset + data.byteLength).toString()
  console.log(msg) // 'hello world'
})

new Reader(handleOrSize)

Creates a reader for the ring buffer.

• handleOrSize -- SharedHandle from writer.handle, or a positive integer to allocate a new ring buffer.

A size is the guaranteed maximum payload for a single write, at most 2**30 - 8 bytes. The allocator adds framing overhead and rounds the data region up to a page-aligned power of two; the resulting limit is writer.maxMessageSize.

reader.handle

The underlying SharedHandle. Pass to another thread via workerData.

reader.size

Physical size in bytes of the ring buffer's data region (page-aligned, rounded up to a power of two by the native allocator).

reader.hugePages

Whether the data region is backed by explicit huge pages (Linux 2 MiB hugetlb). false on other platforms and when the huge-page allocation fell back to regular pages.

reader.stats

Returns { readCount, readBytes }.

reader.readSome(next, opaque?)

Reads a batch of available messages, calling next(data, opaque) for each. Returns the number of messages consumed.

• A single call does not necessarily drain the ring: the batch ends after ~256 KiB. Call repeatedly until it returns 0 to drain.
• data (buffer, view, byteOffset, byteLength) is a zero-copy view into the live ring. The object is reused for every message — capture byteOffset/byteLength inside the callback. The bytes they point at stay valid until the current synchronous execution completes: the read position is published to the writer in a deferred microtask, so the writer cannot reclaim delivered bytes — across all readSome calls in the same tick — until the microtask queue runs. Yielding to the microtask queue (e.g. await) ends the validity window; copy anything you need to retain beyond it, e.g. Buffer.from(data.buffer.subarray(data.byteOffset, data.byteOffset + data.byteLength)).
• Return false from next to stop early.
• Every message handed to next counts as consumed and is never re-delivered — including the one on which next returned false and, if next throws, the one whose callback threw.
• Not re-entrant: calling readSome from inside next throws.

reader.flushSync()

Publishes the read position to the writer immediately instead of waiting for the deferred microtask, ending the readSome validity window early: bytes delivered by earlier readSome calls may be overwritten by the writer as soon as this returns. Required when the writer blocks for space on the same thread within the current synchronous execution (e.g. draining from a yield callback), where microtasks cannot run.

new Writer(handleOrSize, options?)

Creates a writer for the ring buffer.

Options:

• yield?: () => void -- Called when the writer must wait for the reader.
• logger?: { warn(obj, msg): void } -- Logger for yield warnings.

writer.handle

The underlying SharedHandle.

writer.size / writer.hugePages

Same as reader.size / reader.hugePages.

writer.maxMessageSize

Maximum payload size for a single write (writer.size - 8).

writer.stats

Returns { yieldCount, yieldTime, waitNotifyCount, waitTimeoutCount, writeCount, writeBytes }. The wait counters distinguish reader notifications from the bounded timeout fallback; waits that find an already-advanced read index increment neither.

writer.writeSync(len, fn, opaque?)

Synchronously writes a message. fn(data, opaque) must return the end position (data.byteOffset + bytesWritten). If the buffer is full, blocks (via Atomics.wait) until the reader frees space — and throws if no space appears within 60 seconds. Not re-entrant: calling writeSync/tryWrite from the write callback, yield hook, or logger while a write is in progress throws.

writer.tryWrite(len, fn, opaque?)

Non-blocking write attempt. Returns false if the buffer is full. On a full ring it first publishes pending corked writes (as flushSync does) so the reader can drain — a false return means genuinely full against the reader. Not re-entrant (see writeSync).

writer.cork(callback?)

Batches writes to reduce publish frequency. Pending writes are published when the outermost cork is released. Corking is a batching hint, not a transaction: the writer still publishes early when pending bytes reach the high-water mark (256 KiB, or a quarter of the ring for small rings) and when it must wait for the reader to free space, so the reader can observe a partial batch.

writer.uncork()

Releases one cork level and publishes pending writes when the count reaches zero.

writer.flushSync()

Immediately publishes pending writes regardless of cork state.

Native binding

The ring buffer relies on a native C++ addon for double-mapped virtual memory (contiguous reads/writes across the ring boundary) and huge page support on Linux. The SharedArrayBuffer registry uses V8's BackingStore API to hold weak references to backing stores across threads.

License

MIT