From 78e6ee08435ba86a430cd61ea1af58c1c84de3ab Mon Sep 17 00:00:00 2001 From: Devin Ivy Date: Sun, 19 Jul 2026 16:29:02 -0700 Subject: [PATCH] feat!: WorkerHandle index + shared active counts, drop thread --- src/active-counts.ts | 33 ++++++++++++++++++++++++++++ src/runner.ts | 7 +++--- src/worker-pool.ts | 43 ++++++++++++++++++------------------- test/active-counts.test.ts | 31 ++++++++++++++++++++++++++ test/balancers.test.ts | 4 ++-- test/load-balancing.test.ts | 8 +++---- 6 files changed, 94 insertions(+), 32 deletions(-) create mode 100644 src/active-counts.ts create mode 100644 test/active-counts.test.ts diff --git a/src/active-counts.ts b/src/active-counts.ts new file mode 100644 index 0000000..3d6a9dd --- /dev/null +++ b/src/active-counts.ts @@ -0,0 +1,33 @@ +/** Ints per slot: 64-byte stride pads each counter to its own cache line, + * avoiding false sharing between adjacent workers' counters. */ +const STRIDE = 16; + +/** + * Per-worker in-flight task counters on a SharedArrayBuffer, so any thread + * can read (`leastBusy`) or update (dispatching thread) them. + */ +export class ActiveCounts { + readonly #view: Int32Array; + + constructor(sizeOrBuffer: number | SharedArrayBuffer) { + this.#view = new Int32Array( + typeof sizeOrBuffer === 'number' ? new SharedArrayBuffer(sizeOrBuffer * STRIDE * 4) : sizeOrBuffer, + ); + } + + get buffer(): SharedArrayBuffer { + return this.#view.buffer; + } + + inc(index: number): void { + Atomics.add(this.#view, index * STRIDE, 1); + } + + dec(index: number): void { + Atomics.sub(this.#view, index * STRIDE, 1); + } + + get(index: number): number { + return Atomics.load(this.#view, index * STRIDE); + } +} diff --git a/src/runner.ts b/src/runner.ts index 53e25e0..fb18748 100644 --- a/src/runner.ts +++ b/src/runner.ts @@ -1,4 +1,3 @@ -import type { Worker } from 'node:worker_threads'; import type { ChannelOptions } from './channel.ts'; declare const resultBrand: unique symbol; @@ -40,12 +39,12 @@ export interface WorkerOptions { balance?: Balancer; } -/** A handle to a specific worker in a pool. */ +/** A handle to a specific worker in a pool. Uniform across threads. */ export interface WorkerHandle { /** Dispatches a task pinned to this worker. Returns `AsyncIterable` for streaming tasks, `Promise` otherwise. */ exec(task: Task, opts?: ChannelOptions): RunResult; - /** The underlying worker thread. */ - readonly thread: Worker; + /** Position of this worker in the pool. Stable across threads; the portable identity used by `assign()`. */ + readonly index: number; /** Number of currently in-flight tasks on this worker. */ readonly activeCount: number; } diff --git a/src/worker-pool.ts b/src/worker-pool.ts index 7558c3c..9568848 100644 --- a/src/worker-pool.ts +++ b/src/worker-pool.ts @@ -4,6 +4,7 @@ import { availableParallelism } from 'node:os'; import { setupWorker, execute, dispatchStream } from './execute.ts'; import { AsyncIterableTask } from './stream-task.ts'; import { roundRobin } from './balancers.ts'; +import { ActiveCounts } from './active-counts.ts'; import type { ChannelOptions } from './channel.ts'; import type { Task, Balancer, Runner, WorkerHandle, WorkerOptions } from './runner.ts'; @@ -42,11 +43,11 @@ export function workers(sizeOrOpts?: number | WorkerOptions, opts?: WorkerOption let disposed = false; const ac = new AbortController(); const inflight = new Set>(); - const activeCounts = new Map(); + const counts = new ActiveCounts(size); - async function trackValue(handle: WorkerHandle, promise: Promise): Promise { + async function trackValue(index: number, promise: Promise): Promise { inflight.add(promise); - activeCounts.set(handle, (activeCounts.get(handle) ?? 0) + 1); + counts.inc(index); try { return await promise; } catch (err) { @@ -57,16 +58,16 @@ export function workers(sizeOrOpts?: number | WorkerOptions, opts?: WorkerOption throw new Error(String(err), { cause: err }); } finally { inflight.delete(promise); - activeCounts.set(handle, (activeCounts.get(handle) ?? 1) - 1); + counts.dec(index); } } - function trackStream(handle: WorkerHandle, done: Promise): void { + function trackStream(index: number, done: Promise): void { inflight.add(done); - activeCounts.set(handle, (activeCounts.get(handle) ?? 0) + 1); + counts.inc(index); done.then(() => { inflight.delete(done); - activeCounts.set(handle, (activeCounts.get(handle) ?? 1) - 1); + counts.dec(index); }); } @@ -93,43 +94,41 @@ export function workers(sizeOrOpts?: number | WorkerOptions, opts?: WorkerOption } } - function resolveWorkerAndHandle(task: Task): { worker: Worker; handle: WorkerHandle } { + function resolveWorkerAndHandle(task: Task): { worker: Worker; handle: WorkerHandle; idx: number } { if (task.worker != null) { const idx = workerHandles.indexOf(task.worker); - if (idx !== -1) return { worker: pool[idx], handle: workerHandles[idx] }; + if (idx !== -1) return { worker: pool[idx], handle: workerHandles[idx], idx }; } const handle = balancer.select(workerHandles, task); const idx = workerHandles.indexOf(handle); - return { worker: pool[idx], handle }; + return { worker: pool[idx], handle, idx }; } function dispatch(task: Task): Promise { if (disposed) return Promise.reject(new Error('Worker pool is disposed')); - const { worker, handle } = resolveWorkerAndHandle(task); - return trackValue(handle, execute(worker, task.id, task.args)); + const { worker, idx } = resolveWorkerAndHandle(task); + return trackValue(idx, execute(worker, task.id, task.args)); } function makeWorkerHandle(worker: Worker, idx: number): WorkerHandle { - let handle: WorkerHandle; - handle = { + return { exec(task: Task, channelOpts?: ChannelOptions): any { if (task instanceof AsyncIterableTask) { if (disposed) throw new Error('Worker pool is disposed'); const { iterable, done } = dispatchStream(worker, task.id, task.args, channelOpts); - trackStream(handle, done); + trackStream(idx, done); return iterable; } if (disposed) return Promise.reject(new Error('Worker pool is disposed')); - return trackValue(handle, execute(worker, task.id, task.args)); + return trackValue(idx, execute(worker, task.id, task.args)); }, - get thread() { - return worker; + get index() { + return idx; }, get activeCount() { - return activeCounts.get(handle) ?? 0; + return counts.get(idx); }, }; - return handle; } const workerHandles: readonly WorkerHandle[] = Object.freeze(pool.map(makeWorkerHandle)); @@ -138,9 +137,9 @@ export function workers(sizeOrOpts?: number | WorkerOptions, opts?: WorkerOption (taskOrTasks: Task | Task[] | (Task & AsyncIterable), channelOpts?: ChannelOptions): any => { if (taskOrTasks instanceof AsyncIterableTask) { if (disposed) throw new Error('Worker pool is disposed'); - const { worker, handle } = resolveWorkerAndHandle(taskOrTasks); + const { worker, idx } = resolveWorkerAndHandle(taskOrTasks); const { iterable, done } = dispatchStream(worker, taskOrTasks.id, taskOrTasks.args, channelOpts); - trackStream(handle, done); + trackStream(idx, done); return iterable; } if (Array.isArray(taskOrTasks)) { diff --git a/test/active-counts.test.ts b/test/active-counts.test.ts new file mode 100644 index 0000000..1ebd4dc --- /dev/null +++ b/test/active-counts.test.ts @@ -0,0 +1,31 @@ +import { describe, it } from 'node:test'; +import assert from 'node:assert/strict'; +import { ActiveCounts } from '../src/active-counts.ts'; + +describe('ActiveCounts', () => { + it('starts at zero and tracks inc/dec per index', () => { + const counts = new ActiveCounts(4); + assert.equal(counts.get(0), 0); + counts.inc(0); + counts.inc(0); + counts.inc(3); + assert.equal(counts.get(0), 2); + assert.equal(counts.get(3), 1); + counts.dec(0); + assert.equal(counts.get(0), 1); + }); + + it('is backed by a SharedArrayBuffer and reconstructs over it', () => { + const counts = new ActiveCounts(2); + counts.inc(1); + const other = new ActiveCounts(counts.buffer); + assert.equal(other.get(1), 1); + other.inc(1); + assert.equal(counts.get(1), 2); + }); + + it('pads slots to 64-byte stride', () => { + const counts = new ActiveCounts(2); + assert.equal(counts.buffer.byteLength, 2 * 64); + }); +}); diff --git a/test/balancers.test.ts b/test/balancers.test.ts index 13a3ec4..103ce8d 100644 --- a/test/balancers.test.ts +++ b/test/balancers.test.ts @@ -3,8 +3,8 @@ import assert from 'node:assert/strict'; import { roundRobin, leastBusy } from 'moroutine'; import type { WorkerHandle } from 'moroutine'; -function mockHandle(activeCount: number): WorkerHandle { - return { activeCount, thread: {} as any, exec: {} as any }; +function mockHandle(activeCount: number, index = 0): WorkerHandle { + return { activeCount, index, exec: {} as any }; } describe('roundRobin()', () => { diff --git a/test/load-balancing.test.ts b/test/load-balancing.test.ts index 4385e5b..be25496 100644 --- a/test/load-balancing.test.ts +++ b/test/load-balancing.test.ts @@ -36,11 +36,11 @@ describe('load balancing', () => { } }); - it('exposes thread on WorkerHandle', () => { - const run = workers(1); + it('exposes index on WorkerHandle', () => { + const run = workers(2); try { - assert.ok(run.workers[0].thread); - assert.equal(typeof run.workers[0].thread.threadId, 'number'); + assert.equal(run.workers[0].index, 0); + assert.equal(run.workers[1].index, 1); } finally { run[Symbol.dispose](); } -- 2.51.2