src/pools/cluster/fixed.ts

   1 import cluster, { type Worker } from 'node:cluster'
   2 import type { MessageValue } from '../../utility-types'
   3 import { AbstractPool } from '../abstract-pool'
   4 import { type PoolOptions, type PoolType, PoolTypes } from '../pool'
   5 import { type WorkerType, WorkerTypes } from '../worker'
   6
   7 /**
   8  * A cluster pool with a fixed number of workers.
   9  *
  10  * @typeParam Data - Type of data sent to the worker. This can only be structured-cloneable data.
  11  * @typeParam Response - Type of execution response. This can only be structured-cloneable data.
  12  * @author [Christopher Quadflieg](https://github.com/Shinigami92)
  13  * @since 2.0.0
  14  */
  15 export class FixedClusterPool<
  16   Data = unknown,
  17   Response = unknown
  18 > extends AbstractPool<Worker, Data, Response> {
  19   /**
  20    * Constructs a new poolifier fixed cluster pool.
  21    *
  22    * @param numberOfWorkers - Number of workers for this pool.
  23    * @param filePath - Path to an implementation of a `ClusterWorker` file, which can be relative or absolute.
  24    * @param opts - Options for this fixed cluster pool.
  25    */
  26   public constructor (
  27     numberOfWorkers: number,
  28     filePath: string,
  29     protected readonly opts: PoolOptions<Worker> = {}
  30   ) {
  31     super(numberOfWorkers, filePath, opts)
  32   }
  33
  34   /** @inheritDoc */
  35   protected setupHook (): void {
  36     cluster.setupPrimary({ ...this.opts.settings, exec: this.filePath })
  37   }
  38
  39   /** @inheritDoc */
  40   protected isMain (): boolean {
  41     return cluster.isPrimary
  42   }
  43
  44   /** @inheritDoc */
  45   protected async destroyWorkerNode (workerNodeKey: number): Promise<void> {
  46     this.flagWorkerNodeAsNotReady(workerNodeKey)
  47     this.flushTasksQueue(workerNodeKey)
  48     // FIXME: wait for tasks to be finished
  49     const workerNode = this.workerNodes[workerNodeKey]
  50     const waitWorkerExit = new Promise<void>(resolve => {
  51       workerNode.registerOnceWorkerEventHandler('exit', () => {
  52         resolve()
  53       })
  54     })
  55     workerNode.registerOnceWorkerEventHandler('disconnect', () => {
  56       workerNode.worker.kill()
  57     })
  58     await this.sendKillMessageToWorker(workerNodeKey)
  59     workerNode.removeAllListeners()
  60     workerNode.worker.disconnect()
  61     await waitWorkerExit
  62   }
  63
  64   /** @inheritDoc */
  65   protected sendToWorker (
  66     workerNodeKey: number,
  67     message: MessageValue<Data>
  68   ): void {
  69     this.workerNodes[workerNodeKey].worker.send({
  70       ...message,
  71       workerId: this.getWorkerInfo(workerNodeKey).id as number
  72     })
  73   }
  74
  75   /** @inheritDoc */
  76   protected sendStartupMessageToWorker (workerNodeKey: number): void {
  77     this.sendToWorker(workerNodeKey, {
  78       ready: false
  79     })
  80   }
  81
  82   /** @inheritDoc */
  83   protected registerWorkerMessageListener<Message extends Data | Response>(
  84     workerNodeKey: number,
  85     listener: (message: MessageValue<Message>) => void
  86   ): void {
  87     this.workerNodes[workerNodeKey].worker.on('message', listener)
  88   }
  89
  90   /** @inheritDoc */
  91   protected registerOnceWorkerMessageListener<Message extends Data | Response>(
  92     workerNodeKey: number,
  93     listener: (message: MessageValue<Message>) => void
  94   ): void {
  95     this.workerNodes[workerNodeKey].worker.once('message', listener)
  96   }
  97
  98   /** @inheritDoc */
  99   protected deregisterWorkerMessageListener<Message extends Data | Response>(
 100     workerNodeKey: number,
 101     listener: (message: MessageValue<Message>) => void
 102   ): void {
 103     this.workerNodes[workerNodeKey].worker.off('message', listener)
 104   }
 105
 106   /** @inheritDoc */
 107   protected get type (): PoolType {
 108     return PoolTypes.fixed
 109   }
 110
 111   /** @inheritDoc */
 112   protected get worker (): WorkerType {
 113     return WorkerTypes.cluster
 114   }
 115
 116   /** @inheritDoc */
 117   protected get busy (): boolean {
 118     return this.internalBusy()
 119   }
 120 }