@fideus-labs/worker-pool

A worker pool with bounded concurrency, plus a companion @fideus-labs/fizarrita package that accelerates zarrita codec operations on workers. Runs on Web Workers in the browser and on node:worker_threads in Node, behind one interface.

Features

• Bounded concurrency — at most poolSize workers run simultaneously.
• Worker recycling — workers are reused (LIFO) across tasks instead of being re-created.
• ChunkQueue interface — add() + onIdle(), compatible with zarrita.js and p-queue patterns.
• Batch interface — runTasks() with progress reporting and cancellation.
• Browser and Node — the same task code runs on Web Workers and on node:worker_threads; see Node.js.
• Zero runtime dependencies.

Installation

npm add @fideus-labs/worker-pool

Usage

Task function contract

Every task function receives an available worker (or null when the pool needs a new worker created) and must return an object with the worker to recycle and the result:

type WorkerPoolTask<T> = (
  worker: WorkerLike | null
) => Promise<{ worker: WorkerLike; result: T }>

WorkerLike is the slice of the Worker API the pool and its tasks use — postMessage, terminate, addEventListener, removeEventListener. A browser Worker satisfies it, and so does the node:worker_threads adapter, which is what lets one task function serve both runtimes.

ChunkQueue interface (add / onIdle)

import { WorkerPool } from '@fideus-labs/worker-pool'
import type { WorkerLike } from '@fideus-labs/worker-pool'

function createTask(input: number) {
  return (worker: WorkerLike | null) => {
    // Two things worth copying here:
    //  - the literal `new Worker(new URL(...), ...)` form, which is the only
    //    shape bundlers recognise as a worker entry point;
    //  - annotating the local, because `WorkerLike | Worker` is a union
    //    TypeScript resolves to the wrong `postMessage` overload.
    const w: WorkerLike =
      worker ??
      new Worker(new URL('./my-worker.js', import.meta.url), { type: 'module' })
    return new Promise<{ worker: WorkerLike; result: number }>((resolve, reject) => {
      // Handle 'error' too, or a failing worker leaves onIdle() pending forever.
      const onMessage = (e: { data: number }) => {
        detach()
        resolve({ worker: w, result: e.data })
      }
      const onError = (e: { message: string }) => {
        detach()
        reject(new Error(e.message))
      }
      const detach = () => {
        w.removeEventListener('message', onMessage)
        w.removeEventListener('error', onError)
      }
      w.addEventListener('message', onMessage)
      w.addEventListener('error', onError)
      w.postMessage(input)
    })
  }
}

const pool = new WorkerPool(4) // 4 concurrent workers

pool.add(createTask(1))
pool.add(createTask(2))
pool.add(createTask(3))

try {
  const results = await pool.onIdle<number>()
  // results: [result1, result2, result3] — in add() order
} finally {
  pool.terminateWorkers()
}

Batch interface (runTasks)

Submit an array of tasks at once with optional progress reporting and cancellation:

const pool = new WorkerPool(2)

const tasks = inputs.map((input) => createTask(input))

const { promise, runId } = pool.runTasks(tasks, (completed, total) => {
  console.log(`${completed}/${total}`)
})

// Cancel if needed:
// pool.cancel(runId)

try {
  const results = await promise
} finally {
  pool.terminateWorkers()
}

A batch can also be tied to an AbortSignal. When the signal fires, tasks that have not started are dropped and the promise rejects with the signal's reason; tasks already running finish, but their results are discarded and their workers return to the pool:

const controller = new AbortController()

const { promise } = pool.runTasks(tasks, null, { signal: controller.signal })

// e.g. the viewport moved and these results are stale:
controller.abort(new Error('viewport moved'))

API

new WorkerPool(poolSize: number)

Create a pool with at most poolSize concurrent workers.

pool.add<T>(fn: WorkerPoolTask<T>): void

Enqueue a task. Tasks are started when onIdle() is called.

pool.onIdle<T>(): Promise<T[]>

Execute all enqueued tasks and wait for completion. Returns results in the order tasks were added.

pool.runTasks<T>(taskFns, progressCallback?, options?): { promise, runId }

Submit a batch of tasks. The promise resolves with ordered results. The optional progressCallback is invoked as (completedTasks: number, totalTasks: number) => void after each task completes. options.signal accepts an AbortSignal: when it fires, tasks that have not started are dropped and the promise rejects with the signal's reason. A signal that is already aborted rejects the batch before any task starts.

pool.cancel(runId: number): void

Cancel a pending runTasks batch. The promise rejects with 'Remaining tasks canceled'.

pool.terminateWorkers(): void

Terminate all idle workers. The pool can still be used after this — new workers will be created as needed.

createWorker(url, options?): WorkerLike

Create a worker for the current runtime: a module Worker where one exists (browsers, Deno, Bun), a NodeWorker on Node. Throws if neither is available.

Bundled browser code: bundlers only detect a worker entry point from the literal new Worker(new URL('./w.js', import.meta.url), { type: 'module' }) form. A call through createWorker is opaque to them, so in a bundled browser app the worker script has to reach the output another way (a ?worker import, a copied asset, or a literal new Worker(...) on the browser branch).

new NodeWorker(url, options?)

A node:worker_threads worker behind the WorkerLike interface. The underlying thread is created asynchronously — node:worker_threads is imported on demand so browser bundles never try to resolve it — but the constructor is synchronous and messages posted before the thread exists are queued in order.

options accepts name and workerData.

isNodeRuntime(): boolean

Whether the current runtime reports itself as Node.

Node.js

Plain Node has no global Worker, so new Worker(...) in a task throws ReferenceError: Worker is not defined. Use createWorker, which picks the right implementation for the runtime:

import { createWorker, WorkerPool } from '@fideus-labs/worker-pool'
import type { WorkerLike } from '@fideus-labs/worker-pool'

const workerUrl = new URL('./my-worker.mjs', import.meta.url)

function createTask(input: number) {
  return async (worker: WorkerLike | null) => {
    const w = worker ?? createWorker(workerUrl)
    const result = await new Promise<number>((resolve, reject) => {
      const onMessage = (e: { data: number }) => { detach(); resolve(e.data) }
      const onError = (e: { message: string }) => { detach(); reject(new Error(e.message)) }
      const detach = () => {
        w.removeEventListener('message', onMessage)
        w.removeEventListener('error', onError)
      }
      w.addEventListener('message', onMessage)
      w.addEventListener('error', onError)
      w.postMessage(input)
    })
    return { worker: w, result }
  }
}

const pool = new WorkerPool(4)
const { promise } = pool.runTasks(inputs.map(createTask))

try {
  const results = await promise
} finally {
  // Required, and in a finally: worker threads hold the event loop open, so a
  // failed run would otherwise leave the process unable to exit.
  pool.terminateWorkers()
}

The worker script itself is an ordinary node:worker_threads module:

import { parentPort } from 'node:worker_threads'

parentPort.on('message', (input) => parentPort.postMessage(input * input))

Notes:

• Always call pool.terminateWorkers(). A live worker thread keeps the Node event loop alive, so the process will not exit without it.
• Event shape. Node delivers a raw value to on('message') and an Error to on('error'); the adapter wraps them as { data } and { message, error } so listeners read the same in both runtimes.
• SharedArrayBuffer and transfer lists work as they do in the browser.

zarrita.js Integration

The @fideus-labs/fizarrita package provides getWorker and setWorker as drop-in replacements for zarrita's get and set, offloading codec encode/decode to workers via the worker pool. It ships a codec worker for each runtime and picks between them automatically, so the examples below work unchanged in Node.

Installation

pnpm add @fideus-labs/fizarrita @fideus-labs/worker-pool zarrita

Basic usage

import { WorkerPool } from '@fideus-labs/worker-pool'
import { getWorker, setWorker } from '@fideus-labs/fizarrita'
import * as zarr from 'zarrita'

const pool = new WorkerPool(4)

// Open an array
const store = new zarr.FetchStore('https://example.com/data.zarr')
const arr = await zarr.open(store, { kind: 'array' })

try {
  // Read with codec decode offloaded to workers
  const chunk = await getWorker(arr, null, { pool })

  // Write with codec encode offloaded to workers
  await setWorker(arr, null, chunk, { pool })
} finally {
  pool.terminateWorkers()
}

SharedArrayBuffer support

Both getWorker and setWorker support a useSharedArrayBuffer option for additional performance:

// Read — output allocated on SharedArrayBuffer, workers decode directly
// into shared memory (eliminates one transfer + one copy per chunk)
const chunk = await getWorker(arr, null, {
  pool,
  useSharedArrayBuffer: true,
})

// chunk.data.buffer instanceof SharedArrayBuffer === true
// The chunk can be shared with other workers without copying.

// Write — intermediate buffers use SharedArrayBuffer for zero-transfer
// sharing between main thread and codec workers
await setWorker(arr, null, chunk, {
  pool,
  useSharedArrayBuffer: true,
})

getWorker with SAB:

• Output TypedArray is backed by SharedArrayBuffer
• Codec workers decode chunks AND write directly into the shared output buffer via the decode_into message protocol
• Eliminates 1 ArrayBuffer transfer (worker to main) and 1 main-thread set_from_chunk copy per chunk
• Fill-value chunks are still handled on the main thread

setWorker with SAB:

• Intermediate chunk buffers for partial updates use SharedArrayBuffer
• Reduces ArrayBuffer transfers between main thread and codec workers during the decode-modify-encode cycle

COOP/COEP headers

SharedArrayBuffer requires the page to be served with these HTTP headers:

Cross-Origin-Opener-Policy: same-origin
Cross-Origin-Embedder-Policy: require-corp

If these headers are missing, useSharedArrayBuffer: true will throw with a descriptive error message.

Vite example:

// vite.config.ts
export default defineConfig({
  server: {
    headers: {
      'Cross-Origin-Opener-Policy': 'same-origin',
      'Cross-Origin-Embedder-Policy': 'require-corp',
    },
  },
})

getWorker options

| Option | Type | Description | |---|---|---| | pool | WorkerPool | Required. The worker pool to use. | | workerUrl | string \| URL | URL of the codec worker script. Uses built-in default if omitted. | | opts | StoreOpts | Pass-through options for the store's get method (e.g., RequestInit). | | useSharedArrayBuffer | boolean | Allocate output on SharedArrayBuffer with decode-into-shared optimization. |

setWorker options

| Option | Type | Description | |---|---|---| | pool | WorkerPool | Required. The worker pool to use. | | workerUrl | string \| URL | URL of the codec worker script. Uses built-in default if omitted. | | useSharedArrayBuffer | boolean | Use SharedArrayBuffer for intermediate chunk buffers during partial updates. |

Worker message protocol

The codec worker handles four message types:

| Request | Response | Description | |---|---|---| | init | init_ok | Register codec metadata (sent once per worker per unique array config) | | decode | decoded | Decode raw bytes, transfer decoded ArrayBuffer back | | decode_into | decode_into_ok | Decode raw bytes and write directly into SharedArrayBuffer output | | encode | encoded | Encode chunk data, transfer encoded bytes back |

Benchmark

Live Benchmark — try it in your browser.

The repository includes a benchmark app that compares vanilla zarrita get/set with getWorker/setWorker (with and without SharedArrayBuffer):

pnpm bench
# Opens at http://localhost:5174

The benchmark supports both synthetic in-memory arrays and remote OME-Zarr datasets from AWS S3. The live version is deployed to GitHub Pages on every push to main.

Development

pnpm install
pnpm dev           # Start test app dev server (port 5173)
pnpm bench         # Start benchmark app (port 5174)
pnpm test          # Run Playwright browser tests
pnpm test:ui       # Interactive Playwright UI
pnpm test:node     # Build both packages, then run the Node tests

License

MIT