X-Git-Url: https://git.piment-noir.org/?a=blobdiff_plain;f=src%2Fpools%2Fcluster%2Ffixed.ts;h=616523cb2885964ef2fee6e4165aaf5dad979000;hb=deb85c12b77faf6974551cefcd9676e62a392086;hp=b4071407f6e23864f2f34e4ea4da5d5d042333fc;hpb=d3c8a1a8022726094160d1413f1f289c5d84d715;p=poolifier.git diff --git a/src/pools/cluster/fixed.ts b/src/pools/cluster/fixed.ts index b4071407..616523cb 100644 --- a/src/pools/cluster/fixed.ts +++ b/src/pools/cluster/fixed.ts @@ -1,8 +1,11 @@ import { fork, isMaster, setupMaster, Worker } from 'cluster' -import type { JSONValue, MessageValue } from '../../utility-types' +import type { MessageValue } from '../../utility-types' import type { PoolOptions } from '../abstract-pool' import { AbstractPool } from '../abstract-pool' +/** + * Options for a poolifier cluster pool. + */ export interface ClusterPoolOptions extends PoolOptions { /** * Key/value pairs to add to worker process environment. @@ -14,28 +17,35 @@ export interface ClusterPoolOptions extends PoolOptions { } /** - * 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. + * + * @template Data Type of data sent to the worker. This can only be serializable data. + * @template Response Type of response of execution. This can only be serializable data. * * @author [Christopher Quadflieg](https://github.com/Shinigami92) * @since 2.0.0 */ export class FixedClusterPool< - Data extends JSONValue = JSONValue, - Response extends JSONValue = JSONValue + 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. Default: `{ maxTasks: 1000 }` */ public constructor ( - numWorkers: number, + numberOfWorkers: number, filePath: string, public readonly opts: ClusterPoolOptions = { maxTasks: 1000 } ) { - super(numWorkers, filePath, opts) + super(numberOfWorkers, filePath, opts) } protected setupHook (): void { @@ -49,6 +59,7 @@ export class FixedClusterPool< } protected destroyWorker (worker: Worker): void { + this.sendToWorker(worker, { kill: 1 }) worker.kill() } @@ -56,25 +67,25 @@ export class FixedClusterPool< worker.send(message) } - protected registerWorkerMessageListener ( - port: Worker, - listener: (message: MessageValue) => void + protected registerWorkerMessageListener ( + worker: Worker, + listener: (message: MessageValue) => void ): void { - port.on('message', listener) + worker.on('message', listener) } - protected unregisterWorkerMessageListener ( - port: Worker, - listener: (message: MessageValue) => void + protected unregisterWorkerMessageListener ( + worker: Worker, + listener: (message: MessageValue) => void ): void { - port.removeListener('message', listener) + worker.removeListener('message', listener) } - protected newWorker (): Worker { + protected createWorker (): Worker { return fork(this.opts.env) } - protected afterNewWorkerPushed (worker: Worker): void { + protected afterWorkerSetup (worker: Worker): void { // 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)