Worker Management

/**
 * Creates a new worker farm for parallel task execution
 * @param workerPath - Absolute path or URL to the worker module
 * @param options - Configuration options for the worker farm
 */
class Worker {
  constructor(workerPath: string | URL, options?: WorkerFarmOptions);
  
  /** Returns readable stream of all workers' stdout */
  getStdout(): NodeJS.ReadableStream;
  
  /** Returns readable stream of all workers' stderr */
  getStderr(): NodeJS.ReadableStream;
  
  /** Starts all workers and calls their setup functions */
  start(): Promise<void>;
  
  /** Terminates all workers gracefully */
  end(): Promise<PoolExitResult>;
}

import { Worker } from "jest-worker";

// Basic worker creation
const worker = new Worker(require.resolve("./my-worker"));

// Worker with custom configuration
const configuredWorker = new Worker("./worker.js", {
  numWorkers: 8,
  enableWorkerThreads: true,
  maxRetries: 5,
  idleMemoryLimit: 0.1, // 10% of system memory
  computeWorkerKey: (method, filename) => filename // Bind by filename
});

// Using the worker
const result = await worker.processFile("data.txt");
await worker.end();

import type { ForkOptions } from 'child_process';
import type { ResourceLimits } from 'worker_threads';

interface WorkerFarmOptions {
  /** Function to compute worker binding key for task caching */
  computeWorkerKey?: (method: string, ...args: Array<unknown>) => string | null;
  
  /** Use worker_threads instead of child_process.fork */
  enableWorkerThreads?: boolean;
  
  /** Explicitly specify which methods to expose from worker module */
  exposedMethods?: ReadonlyArray<string>;
  
  /** Options passed to child_process.fork */
  forkOptions?: ForkOptions;
  
  /** Maximum number of retries for failed tasks */
  maxRetries?: number;
  
  /** Number of worker processes/threads to spawn */
  numWorkers?: number;
  
  /** Resource limits for worker_threads */
  resourceLimits?: ResourceLimits;
  
  /** Arguments passed to worker setup function */
  setupArgs?: Array<unknown>;
  
  /** Custom task queue implementation */
  taskQueue?: TaskQueue;
  
  /** Custom worker pool implementation */
  WorkerPool?: new (workerPath: string, options?: WorkerPoolOptions) => WorkerPoolInterface;
  
  /** Strategy for assigning tasks to idle workers */
  workerSchedulingPolicy?: WorkerSchedulingPolicy;
  
  /** Memory limit for worker recycling */
  idleMemoryLimit?: number;
}

// Memory-conscious configuration
const memoryWorker = new Worker("./processor.js", {
  idleMemoryLimit: 0.05, // Restart workers using >5% system memory
  numWorkers: 2,
  maxRetries: 1
});

// High-performance configuration
const fastWorker = new Worker("./fast-processor.js", {
  enableWorkerThreads: true,
  numWorkers: 16,
  workerSchedulingPolicy: "round-robin",
  resourceLimits: {
    maxOldGenerationSizeMb: 100,
    maxYoungGenerationSizeMb: 50
  }
});

// Bound worker for caching
const cachingWorker = new Worker("./file-processor.js", {
  computeWorkerKey: (method, filepath) => {
    // Same files always go to same worker for caching
    return method === "processFile" ? filepath : null;
  },
  numWorkers: 4
});

interface WorkerPoolInterface {
  /** Get combined stderr stream from all workers */
  getStderr(): NodeJS.ReadableStream;
  
  /** Get combined stdout stream from all workers */
  getStdout(): NodeJS.ReadableStream;
  
  /** Get array of all worker instances */
  getWorkers(): Array<WorkerInterface>;
  
  /** Create a new worker with specified options */
  createWorker(options: WorkerOptions): WorkerInterface;
  
  /** Send message to specific worker */
  send: WorkerCallback;
  
  /** Start all workers in the pool */
  start(): Promise<void>;
  
  /** Terminate all workers in the pool */
  end(): Promise<PoolExitResult>;
}

interface WorkerPoolOptions {
  setupArgs: Array<unknown>;
  forkOptions: ForkOptions;
  resourceLimits: ResourceLimits;
  maxRetries: number;
  numWorkers: number;
  enableWorkerThreads: boolean;
  idleMemoryLimit?: number;
}

interface WorkerInterface {
  /** Current state of the worker */
  readonly state: WorkerStates;
  
  /** Send a message to this specific worker */
  send(
    request: ChildMessage,
    onProcessStart: OnStart,
    onProcessEnd: OnEnd,
    onCustomMessage: OnCustomMessage
  ): void;
  
  /** Wait for worker to exit gracefully */
  waitForExit(): Promise<void>;
  
  /** Force worker termination */
  forceExit(): void;
  
  /** Get unique worker identifier */
  getWorkerId(): number;
  
  /** Get worker's stderr stream */
  getStderr(): NodeJS.ReadableStream | null;
  
  /** Get worker's stdout stream */
  getStdout(): NodeJS.ReadableStream | null;
  
  /** Get system-level worker identifier (PID, thread ID, etc.) */
  getWorkerSystemId(): number;
  
  /** Get current memory usage of worker */
  getMemoryUsage(): Promise<number | null>;
  
  /** Check if worker process/thread is running */
  isWorkerRunning(): boolean;
  
  /** Wait for worker to be ready to handle requests */
  waitForWorkerReady(): Promise<void>;
}

enum WorkerStates {
  STARTING = 'starting',
  OK = 'ok',
  OUT_OF_MEMORY = 'oom',
  RESTARTING = 'restarting',
  SHUTTING_DOWN = 'shutting-down',
  SHUT_DOWN = 'shut-down'
}

interface PoolExitResult {
  /** True if at least one worker required force termination */
  forceExited: boolean;
}

const worker = new Worker("./long-running-worker.js");

// Do work...
await worker.processLongTask();

// Clean shutdown
const result = await worker.end();

if (result.forceExited) {
  console.warn("Some workers had to be force-terminated");
  console.warn("Check for memory leaks or hanging operations");
}

// my-worker.js

// Export functions that will be exposed as worker methods
exports.processData = function(data) {
  // Synchronous or asynchronous work
  return { result: data.processed };
};

exports.heavyComputation = async function(input) {
  // Async work is fully supported
  const result = await someAsyncOperation(input);
  return result;
};

// Optional lifecycle hooks
exports.setup = function(...setupArgs) {
  // Called once when worker starts
  // setupArgs come from WorkerFarmOptions.setupArgs
  console.log("Worker started with args:", setupArgs);
};

exports.teardown = function() {
  // Called when worker is shutting down
  console.log("Worker shutting down");
};

// In worker module
exports.getWorkerId = function() {
  return process.env.JEST_WORKER_ID;
};

const worker = new Worker("./worker.js", {
  exposedMethods: ["processData", "compute"] // Only expose these methods
});