X-Git-Url: https://git.piment-noir.org/?a=blobdiff_plain;f=src%2Fpools%2Fcluster%2Ffixed.ts;h=81e7b3b7d08b4bb5624d993a15f7850d9d3bf2ba;hb=c319c66bad0611acf6087950a1f8a20f8124167b;hp=332daa4d2b4cfa33225f3ff10a557d5f1a2fbde7;hpb=325f50bc1777ea44abc9736ce9d780ec0c8f90e2;p=poolifier.git diff --git a/src/pools/cluster/fixed.ts b/src/pools/cluster/fixed.ts index 332daa4d..81e7b3b7 100644 --- a/src/pools/cluster/fixed.ts +++ b/src/pools/cluster/fixed.ts @@ -1,28 +1,14 @@ -import type { SendHandle } from 'child_process' -import { fork, isMaster, setupMaster, Worker } from 'cluster' +import type { ClusterSettings, Worker } from 'node:cluster' +import cluster from 'node:cluster' import type { MessageValue } from '../../utility-types' +import { AbstractPool } from '../abstract-pool' +import type { PoolOptions } from '../pool' +import { PoolType } from '../pool' -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 - /** - * This is just to avoid not useful warnings message, is used to set `maxListeners` on event emitters (workers are event emitters). - * - * @default 1000 - */ - maxTasks?: number +/** + * Options for a poolifier cluster pool. + */ +export interface ClusterPoolOptions extends PoolOptions { /** * Key/value pairs to add to worker process environment. * @@ -30,133 +16,97 @@ export interface FixedClusterPoolOptions { */ // eslint-disable-next-line @typescript-eslint/no-explicit-any env?: any + /** + * Cluster settings. + * + * @see https://nodejs.org/api/cluster.html#cluster_cluster_settings + */ + settings?: ClusterSettings } /** - * A cluster pool with a static number of workers, is possible to execute tasks in sync or async mode as you prefer. + * A cluster pool with a fixed number of workers. + * + * It is possible to perform tasks in sync or asynchronous mode as you prefer. * - * This pool will select the worker in a round robin fashion. + * This pool selects the workers in a round robin fashion. * + * @typeParam Data - Type of data sent to the worker. This can only be serializable data. + * @typeParam Response - Type of response of execution. This can only be serializable 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, + public 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) { - worker.kill() - } + /** @inheritDoc */ + public destroyWorker (worker: Worker): void { + this.sendToWorker(worker, { kill: 1 }) + worker.kill() } - /** - * 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) + } + + /** @inheritDoc */ + protected registerWorkerMessageListener( + worker: Worker, + listener: (message: MessageValue) => void + ): void { + worker.on('message', listener) + } + + /** @inheritDoc */ + protected createWorker (): Worker { + return cluster.fork(this.opts.env) + } + + /** @inheritDoc */ + protected afterWorkerSetup (worker: Worker): void { + // Listen to worker messages. + this.registerWorkerMessageListener(worker, super.workerListener()) } - 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 */ + public get type (): PoolType { + return PoolType.FIXED } - 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 full (): boolean { + return this.workerNodes.length === 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() } }