Worker pool example

npm create @setu-ts my-app -- --example worker-pool
cd my-app && npm install
npm start

Run pnpm --dir apps/worker-pool run start here, then request GET /health on port 3000 (pass another port as the first argument). Run pnpm --dir apps/worker-pool run smoke to exercise the unstarted factory in setu.config.ts.

The app registers RuntimePlugin, HealthPlugin with details: true (so /health shows the pool’s budget, which is hidden by default) and WorkerPoolPlugin with maxWorkers: 1. Its two task modules in src/tasks/ share that slot: spin.task.ts does CPU-bound work off the event loop; fill.task.ts writes a byte pattern into a caller-provided buffer.

The smoke checks main-thread timer progress during a 300 ms spin, completion of a second module before the spin queue drains, shared-buffer writes visible to the caller, an unchanged ArrayBuffer control, and the budget reported through /health.

Wait for the task promise to settle before reading shared memory. A timeout kills the worker and may leave a partial write; concurrent access requires an Atomics protocol.

Read the example deep dive for the concept and source walkthrough.

Read, in order

• src/tasks/fill.task.ts
• src/tasks/index.ts
• src/tasks/spin.task.ts
• smoke.ts

import { defineWorkerTask } from '@setu-ts/runtime/worker';

/** The input `fill` receives: a buffer to write into and the byte to write. */
export interface FillInput {
  readonly buf: SharedArrayBuffer | ArrayBuffer;
  readonly pattern: number;
}

/** Fills the buffer with one byte value and returns how many bytes it wrote. */
defineWorkerTask<FillInput, number>(({ buf, pattern }) => {
  const view = new Uint8Array(buf);
  view.fill(pattern);
  return view.length;
});

/**
 * The task modules the pool runs, addressed by specifier.
 *
 * A worker loads its task by module URL, never by closure, so these are URLs
 * rather than imports.
 */
export const SPIN_TASK = new URL('./spin.task.ts', import.meta.url).href;
export const FILL_TASK = new URL('./fill.task.ts', import.meta.url).href;

export type { FillInput } from './fill.task.ts';

import { defineWorkerTask } from '@setu-ts/runtime/worker';

/** Busy-waits for the given number of milliseconds and returns how long it ran. */
defineWorkerTask<number, number>((milliseconds) => {
  const started = performance.now();
  while (performance.now() - started < milliseconds) {
    // Deliberately CPU-bound: yielding here would not prove the work left the main thread.
  }
  return performance.now() - started;
});

import assert from 'node:assert/strict';
import { after, before, describe, it } from 'node:test';
import { CAPABILITIES } from '@setu-ts/common';
import type { IWorkerPool } from '@setu-ts/common';
import type { IKernelApplication } from '@setu-ts/kernel';
import { FILL_TASK, SPIN_TASK } from './src/tasks/index.ts';
import type { FillInput } from './src/tasks/index.ts';
import { createApp } from './setu.config.ts';

const PATTERN = 0x5a;

interface WorkerPoolHealth {
  readonly checks: Record<string, { readonly data: { readonly budget: Budget } }>;
}

interface Budget {
  readonly maxWorkers: number;
  readonly workers: number;
}

describe('worker-pool example', () => {
  let app: IKernelApplication;
  let pool: IWorkerPool;

  before(async () => {
    app = createApp();
    await app.start();
    pool = app.services.get<IWorkerPool>(CAPABILITIES.WORKER_POOL);
  });

  after(async () => {
    await app.stop();
  });

  it('keeps main-thread timers running during CPU-bound work', async () => {
    let ticks = 0;
    const timer = setInterval(() => {
      ticks += 1;
    }, 10);
    try {
      await pool.run(SPIN_TASK, 300);
    } finally {
      clearInterval(timer);
    }

    assert.ok(ticks >= 5, `only ${ticks} timer ticks ran during the spin`);
  });

  it('lets a second task module through before a queue of the first drains', async () => {
    const order: string[] = [];
    const spins = Array.from({ length: 5 }, (_, index) =>
      pool.run(SPIN_TASK, 30).then(() => order.push(`spin-${index}`)),
    );
    const fill = pool
      .run<FillInput, number>(FILL_TASK, { buf: new SharedArrayBuffer(16), pattern: PATTERN })
      .then(() => order.push('fill'));
    await Promise.all([...spins, fill]);

    assert.ok(order.indexOf('fill') < order.indexOf('spin-4'), order.join(', '));
  });

  it("shows a worker's writes to a SharedArrayBuffer to the caller", async () => {
    const shared = new SharedArrayBuffer(1024);

    await pool.run<FillInput, number>(FILL_TASK, { buf: shared, pattern: PATTERN });

    assert.ok(new Uint8Array(shared).every((byte) => byte === PATTERN));
  });

  it('copies an ordinary ArrayBuffer into the worker', async () => {
    const copied = new ArrayBuffer(1024);

    const written = await pool.run<FillInput, number>(FILL_TASK, { buf: copied, pattern: PATTERN });

    assert.equal(written, copied.byteLength);
    assert.ok(new Uint8Array(copied).every((byte) => byte === 0));
  });

  it('reports the configured worker budget through /health', async () => {
    const response = await app.inject({ method: 'GET', url: '/health' });
    const budget = response.json<WorkerPoolHealth>().checks['worker-pool']?.data.budget;

    assert.equal(response.statusCode, 200);
    assert.equal(budget?.maxWorkers, 1);
    assert.ok((budget?.workers ?? Infinity) <= 1);
  });
});