X-Git-Url: https://git.piment-noir.org/?a=blobdiff_plain;f=src%2Fpools%2Fcluster%2Ffixed.ts;h=6e9b4b07113bac62000a84ae0a2d8aa05b333532;hb=3d6dd312a7825521cce506ebb7443bae36a111e6;hp=332daa4d2b4cfa33225f3ff10a557d5f1a2fbde7;hpb=325f50bc1777ea44abc9736ce9d780ec0c8f90e2;p=poolifier.git diff --git a/src/pools/cluster/fixed.ts b/src/pools/cluster/fixed.ts index 332daa4d..6e9b4b07 100644 --- a/src/pools/cluster/fixed.ts +++ b/src/pools/cluster/fixed.ts @@ -1,162 +1,105 @@ -import type { SendHandle } from 'child_process' -import { fork, isMaster, setupMaster, Worker } from 'cluster' +import cluster, { type ClusterSettings, type Worker } from 'node:cluster' import type { MessageValue } from '../../utility-types' +import { AbstractPool } from '../abstract-pool' +import { type PoolOptions, type PoolType, PoolTypes } from '../pool' +import { type WorkerType, WorkerTypes } from '../worker' -export type WorkerWithMessageChannel = Worker // & Draft - -export interface FixedClusterPoolOptions { - /** - * A function that will listen for error event on each worker. - */ - errorHandler?: (this: Worker, e: Error) => void - /** - * A function that will listen for online event on each worker. - */ - onlineHandler?: (this: Worker) => void - /** - * A function that will listen for exit event on each worker. - */ - exitHandler?: (this: Worker, code: number) => void +/** + * Options for a poolifier cluster pool. + */ +export interface ClusterPoolOptions extends PoolOptions { /** - * This is just to avoid not useful warnings message, is used to set `maxListeners` on event emitters (workers are event emitters). + * Key/value pairs to add to worker process environment. * - * @default 1000 + * @see https://nodejs.org/api/cluster.html#cluster_cluster_fork_env */ - maxTasks?: number + env?: Record /** - * Key/value pairs to add to worker process environment. + * Cluster settings. * - * @see https://nodejs.org/api/cluster.html#cluster_cluster_fork_env + * @see https://nodejs.org/api/cluster.html#cluster_cluster_settings */ - // eslint-disable-next-line @typescript-eslint/no-explicit-any - env?: any + settings?: ClusterSettings } /** - * A cluster pool with a static number of workers, is possible to execute tasks in sync or async mode as you prefer. - * - * This pool will select the worker in a round robin fashion. + * A cluster pool with a fixed number of workers. * + * @typeParam Data - Type of data sent to the worker. This can only be structured-cloneable data. + * @typeParam Response - Type of execution response. This can only be structured-cloneable data. * @author [Christopher Quadflieg](https://github.com/Shinigami92) * @since 2.0.0 */ -// eslint-disable-next-line @typescript-eslint/no-explicit-any -export class FixedClusterPool { - public readonly workers: WorkerWithMessageChannel[] = [] - public nextWorker: number = 0 - - // workerId as key and an integer value - public readonly tasks: Map = new Map< - WorkerWithMessageChannel, - number - >() - - protected id: number = 0 - +export class FixedClusterPool< + Data = unknown, + Response = unknown +> extends AbstractPool { /** - * @param numWorkers Number of workers for this pool. - * @param filePath A file path with implementation of `ClusterWorker` class, relative path is fine. - * @param opts An object with possible options for example `errorHandler`, `onlineHandler`. Default: `{ maxTasks: 1000 }` + * Constructs a new poolifier fixed cluster pool. + * + * @param numberOfWorkers - Number of workers for this pool. + * @param filePath - Path to an implementation of a `ClusterWorker` file, which can be relative or absolute. + * @param opts - Options for this fixed cluster pool. */ public constructor ( - public readonly numWorkers: number, - public readonly filePath: string, - public readonly opts: FixedClusterPoolOptions = { maxTasks: 1000 } + numberOfWorkers: number, + filePath: string, + protected readonly opts: ClusterPoolOptions = {} ) { - if (!isMaster) { - throw new Error('Cannot start a cluster pool from a worker!') - } - // TODO christopher 2021-02-09: Improve this check e.g. with a pattern or blank check - if (!this.filePath) { - throw new Error('Please specify a file with a worker implementation') - } + super(numberOfWorkers, filePath, opts) + } - setupMaster({ - exec: this.filePath - }) + /** @inheritDoc */ + protected setupHook (): void { + cluster.setupPrimary({ ...this.opts.settings, exec: this.filePath }) + } - for (let i = 1; i <= this.numWorkers; i++) { - this.newWorker() - } + /** @inheritDoc */ + protected isMain (): boolean { + return cluster.isPrimary } - public destroy (): void { - for (const worker of this.workers) { + /** @inheritDoc */ + protected destroyWorker (worker: Worker): void { + this.sendToWorker(worker, { kill: true, workerId: worker.id }) + worker.on('disconnect', () => { worker.kill() - } + }) + worker.disconnect() } - /** - * Execute the task specified into the constructor with the data parameter. - * - * @param data The input for the task specified. - * @returns Promise that is resolved when the task is done. - */ - public execute (data: Data): Promise { - // configure worker to handle message with the specified task - const worker: WorkerWithMessageChannel = this.chooseWorker() - // console.log('FixedClusterPool#execute choosen worker:', worker) - const previousWorkerIndex = this.tasks.get(worker) - if (previousWorkerIndex !== undefined) { - this.tasks.set(worker, previousWorkerIndex + 1) - } else { - throw Error('Worker could not be found in tasks map') - } - const id: number = ++this.id - const res: Promise = this.internalExecute(worker, id) - // console.log('FixedClusterPool#execute send data to worker:', worker) - worker.send({ data: data || {}, id: id }) - return res + /** @inheritDoc */ + protected sendToWorker (worker: Worker, message: MessageValue): void { + worker.send(message) } - protected internalExecute ( - worker: WorkerWithMessageChannel, - id: number - ): Promise { - return new Promise((resolve, reject) => { - const listener: ( - message: MessageValue, - handle: SendHandle - ) => void = message => { - // console.log('FixedClusterPool#internalExecute listener:', message) - if (message.id === id) { - worker.removeListener('message', listener) - const previousWorkerIndex = this.tasks.get(worker) - if (previousWorkerIndex !== undefined) { - this.tasks.set(worker, previousWorkerIndex + 1) - } else { - throw Error('Worker could not be found in tasks map') - } - if (message.error) reject(message.error) - else resolve(message.data as Response) - } - } - worker.on('message', listener) - }) + /** @inheritDoc */ + protected createWorker (): Worker { + return cluster.fork(this.opts.env) + } + + /** @inheritDoc */ + protected get type (): PoolType { + return PoolTypes.fixed + } + + /** @inheritDoc */ + protected get worker (): WorkerType { + return WorkerTypes.cluster + } + + /** @inheritDoc */ + protected get minSize (): number { + return this.numberOfWorkers } - protected chooseWorker (): WorkerWithMessageChannel { - if (this.workers.length - 1 === this.nextWorker) { - this.nextWorker = 0 - return this.workers[this.nextWorker] - } else { - this.nextWorker++ - return this.workers[this.nextWorker] - } + /** @inheritDoc */ + protected get maxSize (): number { + return this.numberOfWorkers } - protected newWorker (): WorkerWithMessageChannel { - const worker: WorkerWithMessageChannel = fork(this.opts.env) - worker.on('error', this.opts.errorHandler ?? (() => {})) - worker.on('online', this.opts.onlineHandler ?? (() => {})) - // TODO handle properly when a worker exit - worker.on('exit', this.opts.exitHandler ?? (() => {})) - this.workers.push(worker) - // we will attach a listener for every task, - // when task is completed the listener will be removed but to avoid warnings we are increasing the max listeners size - worker.setMaxListeners(this.opts.maxTasks ?? 1000) - // init tasks map - this.tasks.set(worker, 0) - return worker + /** @inheritDoc */ + protected get busy (): boolean { + return this.internalBusy() } }