src/pools/abstract-pool.ts

   1 import crypto from 'node:crypto'
   2 import { performance } from 'node:perf_hooks'
   3 import type { MessageValue, PromiseResponseWrapper } from '../utility-types'
   4 import {
   5   DEFAULT_WORKER_CHOICE_STRATEGY_OPTIONS,
   6   EMPTY_FUNCTION,
   7   isPlainObject,
   8   median
   9 } from '../utils'
  10 import { KillBehaviors, isKillBehavior } from '../worker/worker-options'
  11 import { CircularArray } from '../circular-array'
  12 import { Queue } from '../queue'
  13 import {
  14   type IPool,
  15   PoolEmitter,
  16   PoolEvents,
  17   type PoolInfo,
  18   type PoolOptions,
  19   type PoolType,
  20   PoolTypes,
  21   type TasksQueueOptions,
  22   type WorkerType
  23 } from './pool'
  24 import type {
  25   IWorker,
  26   Task,
  27   TaskStatistics,
  28   WorkerNode,
  29   WorkerUsage
  30 } from './worker'
  31 import {
  32   WorkerChoiceStrategies,
  33   type WorkerChoiceStrategy,
  34   type WorkerChoiceStrategyOptions
  35 } from './selection-strategies/selection-strategies-types'
  36 import { WorkerChoiceStrategyContext } from './selection-strategies/worker-choice-strategy-context'
  37
  38 /**
  39  * Base class that implements some shared logic for all poolifier pools.
  40  *
  41  * @typeParam Worker - Type of worker which manages this pool.
  42  * @typeParam Data - Type of data sent to the worker. This can only be serializable data.
  43  * @typeParam Response - Type of execution response. This can only be serializable data.
  44  */
  45 export abstract class AbstractPool<
  46   Worker extends IWorker,
  47   Data = unknown,
  48   Response = unknown
  49 > implements IPool<Worker, Data, Response> {
  50   /** @inheritDoc */
  51   public readonly workerNodes: Array<WorkerNode<Worker, Data>> = []
  52
  53   /** @inheritDoc */
  54   public readonly emitter?: PoolEmitter
  55
  56   /**
  57    * The execution response promise map.
  58    *
  59    * - `key`: The message id of each submitted task.
  60    * - `value`: An object that contains the worker, the execution response promise resolve and reject callbacks.
  61    *
  62    * When we receive a message from the worker, we get a map entry with the promise resolve/reject bound to the message id.
  63    */
  64   protected promiseResponseMap: Map<
  65   string,
  66   PromiseResponseWrapper<Worker, Response>
  67   > = new Map<string, PromiseResponseWrapper<Worker, Response>>()
  68
  69   /**
  70    * Worker choice strategy context referencing a worker choice algorithm implementation.
  71    */
  72   protected workerChoiceStrategyContext: WorkerChoiceStrategyContext<
  73   Worker,
  74   Data,
  75   Response
  76   >
  77
  78   /**
  79    * Constructs a new poolifier pool.
  80    *
  81    * @param numberOfWorkers - Number of workers that this pool should manage.
  82    * @param filePath - Path to the worker file.
  83    * @param opts - Options for the pool.
  84    */
  85   public constructor (
  86     protected readonly numberOfWorkers: number,
  87     protected readonly filePath: string,
  88     protected readonly opts: PoolOptions<Worker>
  89   ) {
  90     if (!this.isMain()) {
  91       throw new Error('Cannot start a pool from a worker!')
  92     }
  93     this.checkNumberOfWorkers(this.numberOfWorkers)
  94     this.checkFilePath(this.filePath)
  95     this.checkPoolOptions(this.opts)
  96
  97     this.chooseWorkerNode = this.chooseWorkerNode.bind(this)
  98     this.executeTask = this.executeTask.bind(this)
  99     this.enqueueTask = this.enqueueTask.bind(this)
 100     this.checkAndEmitEvents = this.checkAndEmitEvents.bind(this)
 101
 102     if (this.opts.enableEvents === true) {
 103       this.emitter = new PoolEmitter()
 104     }
 105     this.workerChoiceStrategyContext = new WorkerChoiceStrategyContext<
 106     Worker,
 107     Data,
 108     Response
 109     >(
 110       this,
 111       this.opts.workerChoiceStrategy,
 112       this.opts.workerChoiceStrategyOptions
 113     )
 114
 115     this.setupHook()
 116
 117     for (let i = 1; i <= this.numberOfWorkers; i++) {
 118       this.createAndSetupWorker()
 119     }
 120   }
 121
 122   private checkFilePath (filePath: string): void {
 123     if (
 124       filePath == null ||
 125       (typeof filePath === 'string' && filePath.trim().length === 0)
 126     ) {
 127       throw new Error('Please specify a file with a worker implementation')
 128     }
 129   }
 130
 131   private checkNumberOfWorkers (numberOfWorkers: number): void {
 132     if (numberOfWorkers == null) {
 133       throw new Error(
 134         'Cannot instantiate a pool without specifying the number of workers'
 135       )
 136     } else if (!Number.isSafeInteger(numberOfWorkers)) {
 137       throw new TypeError(
 138         'Cannot instantiate a pool with a non safe integer number of workers'
 139       )
 140     } else if (numberOfWorkers < 0) {
 141       throw new RangeError(
 142         'Cannot instantiate a pool with a negative number of workers'
 143       )
 144     } else if (this.type === PoolTypes.fixed && numberOfWorkers === 0) {
 145       throw new Error('Cannot instantiate a fixed pool with no worker')
 146     }
 147   }
 148
 149   private checkPoolOptions (opts: PoolOptions<Worker>): void {
 150     if (isPlainObject(opts)) {
 151       this.opts.workerChoiceStrategy =
 152         opts.workerChoiceStrategy ?? WorkerChoiceStrategies.ROUND_ROBIN
 153       this.checkValidWorkerChoiceStrategy(this.opts.workerChoiceStrategy)
 154       this.opts.workerChoiceStrategyOptions =
 155         opts.workerChoiceStrategyOptions ??
 156         DEFAULT_WORKER_CHOICE_STRATEGY_OPTIONS
 157       this.checkValidWorkerChoiceStrategyOptions(
 158         this.opts.workerChoiceStrategyOptions
 159       )
 160       this.opts.restartWorkerOnError = opts.restartWorkerOnError ?? true
 161       this.opts.enableEvents = opts.enableEvents ?? true
 162       this.opts.enableTasksQueue = opts.enableTasksQueue ?? false
 163       if (this.opts.enableTasksQueue) {
 164         this.checkValidTasksQueueOptions(
 165           opts.tasksQueueOptions as TasksQueueOptions
 166         )
 167         this.opts.tasksQueueOptions = this.buildTasksQueueOptions(
 168           opts.tasksQueueOptions as TasksQueueOptions
 169         )
 170       }
 171     } else {
 172       throw new TypeError('Invalid pool options: must be a plain object')
 173     }
 174   }
 175
 176   private checkValidWorkerChoiceStrategy (
 177     workerChoiceStrategy: WorkerChoiceStrategy
 178   ): void {
 179     if (!Object.values(WorkerChoiceStrategies).includes(workerChoiceStrategy)) {
 180       throw new Error(
 181         `Invalid worker choice strategy '${workerChoiceStrategy}'`
 182       )
 183     }
 184   }
 185
 186   private checkValidWorkerChoiceStrategyOptions (
 187     workerChoiceStrategyOptions: WorkerChoiceStrategyOptions
 188   ): void {
 189     if (!isPlainObject(workerChoiceStrategyOptions)) {
 190       throw new TypeError(
 191         'Invalid worker choice strategy options: must be a plain object'
 192       )
 193     }
 194     if (
 195       workerChoiceStrategyOptions.weights != null &&
 196       Object.keys(workerChoiceStrategyOptions.weights).length !== this.maxSize
 197     ) {
 198       throw new Error(
 199         'Invalid worker choice strategy options: must have a weight for each worker node'
 200       )
 201     }
 202   }
 203
 204   private checkValidTasksQueueOptions (
 205     tasksQueueOptions: TasksQueueOptions
 206   ): void {
 207     if (tasksQueueOptions != null && !isPlainObject(tasksQueueOptions)) {
 208       throw new TypeError('Invalid tasks queue options: must be a plain object')
 209     }
 210     if ((tasksQueueOptions?.concurrency as number) <= 0) {
 211       throw new Error(
 212         `Invalid worker tasks concurrency '${
 213           tasksQueueOptions.concurrency as number
 214         }'`
 215       )
 216     }
 217   }
 218
 219   /** @inheritDoc */
 220   public get info (): PoolInfo {
 221     return {
 222       type: this.type,
 223       worker: this.worker,
 224       minSize: this.minSize,
 225       maxSize: this.maxSize,
 226       workerNodes: this.workerNodes.length,
 227       idleWorkerNodes: this.workerNodes.reduce(
 228         (accumulator, workerNode) =>
 229           workerNode.workerUsage.tasks.executing === 0
 230             ? accumulator + 1
 231             : accumulator,
 232         0
 233       ),
 234       busyWorkerNodes: this.workerNodes.reduce(
 235         (accumulator, workerNode) =>
 236           workerNode.workerUsage.tasks.executing > 0
 237             ? accumulator + 1
 238             : accumulator,
 239         0
 240       ),
 241       executedTasks: this.workerNodes.reduce(
 242         (accumulator, workerNode) =>
 243           accumulator + workerNode.workerUsage.tasks.executed,
 244         0
 245       ),
 246       executingTasks: this.workerNodes.reduce(
 247         (accumulator, workerNode) =>
 248           accumulator + workerNode.workerUsage.tasks.executing,
 249         0
 250       ),
 251       queuedTasks: this.workerNodes.reduce(
 252         (accumulator, workerNode) => accumulator + workerNode.tasksQueue.size,
 253         0
 254       ),
 255       maxQueuedTasks: this.workerNodes.reduce(
 256         (accumulator, workerNode) =>
 257           accumulator + workerNode.tasksQueue.maxSize,
 258         0
 259       ),
 260       failedTasks: this.workerNodes.reduce(
 261         (accumulator, workerNode) =>
 262           accumulator + workerNode.workerUsage.tasks.failed,
 263         0
 264       )
 265     }
 266   }
 267
 268   /**
 269    * Pool type.
 270    *
 271    * If it is `'dynamic'`, it provides the `max` property.
 272    */
 273   protected abstract get type (): PoolType
 274
 275   /**
 276    * Gets the worker type.
 277    */
 278   protected abstract get worker (): WorkerType
 279
 280   /**
 281    * Pool minimum size.
 282    */
 283   protected abstract get minSize (): number
 284
 285   /**
 286    * Pool maximum size.
 287    */
 288   protected abstract get maxSize (): number
 289
 290   /**
 291    * Gets the given worker its worker node key.
 292    *
 293    * @param worker - The worker.
 294    * @returns The worker node key if the worker is found in the pool worker nodes, `-1` otherwise.
 295    */
 296   private getWorkerNodeKey (worker: Worker): number {
 297     return this.workerNodes.findIndex(
 298       workerNode => workerNode.worker === worker
 299     )
 300   }
 301
 302   /** @inheritDoc */
 303   public setWorkerChoiceStrategy (
 304     workerChoiceStrategy: WorkerChoiceStrategy,
 305     workerChoiceStrategyOptions?: WorkerChoiceStrategyOptions
 306   ): void {
 307     this.checkValidWorkerChoiceStrategy(workerChoiceStrategy)
 308     this.opts.workerChoiceStrategy = workerChoiceStrategy
 309     this.workerChoiceStrategyContext.setWorkerChoiceStrategy(
 310       this.opts.workerChoiceStrategy
 311     )
 312     if (workerChoiceStrategyOptions != null) {
 313       this.setWorkerChoiceStrategyOptions(workerChoiceStrategyOptions)
 314     }
 315     for (const workerNode of this.workerNodes) {
 316       this.setWorkerNodeTasksUsage(
 317         workerNode,
 318         this.getWorkerUsage(workerNode.worker)
 319       )
 320       this.setWorkerStatistics(workerNode.worker)
 321     }
 322   }
 323
 324   /** @inheritDoc */
 325   public setWorkerChoiceStrategyOptions (
 326     workerChoiceStrategyOptions: WorkerChoiceStrategyOptions
 327   ): void {
 328     this.checkValidWorkerChoiceStrategyOptions(workerChoiceStrategyOptions)
 329     this.opts.workerChoiceStrategyOptions = workerChoiceStrategyOptions
 330     this.workerChoiceStrategyContext.setOptions(
 331       this.opts.workerChoiceStrategyOptions
 332     )
 333   }
 334
 335   /** @inheritDoc */
 336   public enableTasksQueue (
 337     enable: boolean,
 338     tasksQueueOptions?: TasksQueueOptions
 339   ): void {
 340     if (this.opts.enableTasksQueue === true && !enable) {
 341       this.flushTasksQueues()
 342     }
 343     this.opts.enableTasksQueue = enable
 344     this.setTasksQueueOptions(tasksQueueOptions as TasksQueueOptions)
 345   }
 346
 347   /** @inheritDoc */
 348   public setTasksQueueOptions (tasksQueueOptions: TasksQueueOptions): void {
 349     if (this.opts.enableTasksQueue === true) {
 350       this.checkValidTasksQueueOptions(tasksQueueOptions)
 351       this.opts.tasksQueueOptions =
 352         this.buildTasksQueueOptions(tasksQueueOptions)
 353     } else if (this.opts.tasksQueueOptions != null) {
 354       delete this.opts.tasksQueueOptions
 355     }
 356   }
 357
 358   private buildTasksQueueOptions (
 359     tasksQueueOptions: TasksQueueOptions
 360   ): TasksQueueOptions {
 361     return {
 362       concurrency: tasksQueueOptions?.concurrency ?? 1
 363     }
 364   }
 365
 366   /**
 367    * Whether the pool is full or not.
 368    *
 369    * The pool filling boolean status.
 370    */
 371   protected get full (): boolean {
 372     return this.workerNodes.length >= this.maxSize
 373   }
 374
 375   /**
 376    * Whether the pool is busy or not.
 377    *
 378    * The pool busyness boolean status.
 379    */
 380   protected abstract get busy (): boolean
 381
 382   protected internalBusy (): boolean {
 383     return (
 384       this.workerNodes.findIndex(workerNode => {
 385         return workerNode.workerUsage.tasks.executing === 0
 386       }) === -1
 387     )
 388   }
 389
 390   /** @inheritDoc */
 391   public async execute (data?: Data, name?: string): Promise<Response> {
 392     const timestamp = performance.now()
 393     const workerNodeKey = this.chooseWorkerNode()
 394     const submittedTask: Task<Data> = {
 395       name,
 396       // eslint-disable-next-line @typescript-eslint/consistent-type-assertions
 397       data: data ?? ({} as Data),
 398       timestamp,
 399       id: crypto.randomUUID()
 400     }
 401     const res = new Promise<Response>((resolve, reject) => {
 402       this.promiseResponseMap.set(submittedTask.id as string, {
 403         resolve,
 404         reject,
 405         worker: this.workerNodes[workerNodeKey].worker
 406       })
 407     })
 408     if (
 409       this.opts.enableTasksQueue === true &&
 410       (this.busy ||
 411         this.workerNodes[workerNodeKey].workerUsage.tasks.executing >=
 412           ((this.opts.tasksQueueOptions as TasksQueueOptions)
 413             .concurrency as number))
 414     ) {
 415       this.enqueueTask(workerNodeKey, submittedTask)
 416     } else {
 417       this.executeTask(workerNodeKey, submittedTask)
 418     }
 419     this.workerChoiceStrategyContext.update(workerNodeKey)
 420     this.checkAndEmitEvents()
 421     // eslint-disable-next-line @typescript-eslint/return-await
 422     return res
 423   }
 424
 425   /** @inheritDoc */
 426   public async destroy (): Promise<void> {
 427     await Promise.all(
 428       this.workerNodes.map(async (workerNode, workerNodeKey) => {
 429         this.flushTasksQueue(workerNodeKey)
 430         // FIXME: wait for tasks to be finished
 431         await this.destroyWorker(workerNode.worker)
 432       })
 433     )
 434   }
 435
 436   /**
 437    * Shutdowns the given worker.
 438    *
 439    * @param worker - A worker within `workerNodes`.
 440    */
 441   protected abstract destroyWorker (worker: Worker): void | Promise<void>
 442
 443   /**
 444    * Setup hook to execute code before worker node are created in the abstract constructor.
 445    * Can be overridden
 446    *
 447    * @virtual
 448    */
 449   protected setupHook (): void {
 450     // Intentionally empty
 451   }
 452
 453   /**
 454    * Should return whether the worker is the main worker or not.
 455    */
 456   protected abstract isMain (): boolean
 457
 458   /**
 459    * Hook executed before the worker task execution.
 460    * Can be overridden.
 461    *
 462    * @param workerNodeKey - The worker node key.
 463    * @param task - The task to execute.
 464    */
 465   protected beforeTaskExecutionHook (
 466     workerNodeKey: number,
 467     task: Task<Data>
 468   ): void {
 469     const workerUsage = this.workerNodes[workerNodeKey].workerUsage
 470     ++workerUsage.tasks.executing
 471     this.updateWaitTimeWorkerUsage(workerUsage, task)
 472   }
 473
 474   /**
 475    * Hook executed after the worker task execution.
 476    * Can be overridden.
 477    *
 478    * @param worker - The worker.
 479    * @param message - The received message.
 480    */
 481   protected afterTaskExecutionHook (
 482     worker: Worker,
 483     message: MessageValue<Response>
 484   ): void {
 485     const workerUsage =
 486       this.workerNodes[this.getWorkerNodeKey(worker)].workerUsage
 487     const workerTaskStatistics = workerUsage.tasks
 488     --workerTaskStatistics.executing
 489     ++workerTaskStatistics.executed
 490     if (message.taskError != null) {
 491       ++workerTaskStatistics.failed
 492     }
 493     this.updateRunTimeWorkerUsage(workerUsage, message)
 494     this.updateEluWorkerUsage(workerUsage, message)
 495   }
 496
 497   private updateRunTimeWorkerUsage (
 498     workerUsage: WorkerUsage,
 499     message: MessageValue<Response>
 500   ): void {
 501     if (
 502       this.workerChoiceStrategyContext.getTaskStatisticsRequirements().runTime
 503         .aggregate
 504     ) {
 505       workerUsage.runTime.aggregate += message.taskPerformance?.runTime ?? 0
 506       if (
 507         this.workerChoiceStrategyContext.getTaskStatisticsRequirements().runTime
 508           .average &&
 509         workerUsage.tasks.executed !== 0
 510       ) {
 511         workerUsage.runTime.average =
 512           workerUsage.runTime.aggregate / workerUsage.tasks.executed
 513       }
 514       if (
 515         this.workerChoiceStrategyContext.getTaskStatisticsRequirements().runTime
 516           .median &&
 517         message.taskPerformance?.runTime != null
 518       ) {
 519         workerUsage.runTime.history.push(message.taskPerformance.runTime)
 520         workerUsage.runTime.median = median(workerUsage.runTime.history)
 521       }
 522     }
 523   }
 524
 525   private updateWaitTimeWorkerUsage (
 526     workerUsage: WorkerUsage,
 527     task: Task<Data>
 528   ): void {
 529     const timestamp = performance.now()
 530     const taskWaitTime = timestamp - (task.timestamp ?? timestamp)
 531     if (
 532       this.workerChoiceStrategyContext.getTaskStatisticsRequirements().waitTime
 533         .aggregate
 534     ) {
 535       workerUsage.waitTime.aggregate += taskWaitTime ?? 0
 536       if (
 537         this.workerChoiceStrategyContext.getTaskStatisticsRequirements()
 538           .waitTime.average &&
 539         workerUsage.tasks.executed !== 0
 540       ) {
 541         workerUsage.waitTime.average =
 542           workerUsage.waitTime.aggregate / workerUsage.tasks.executed
 543       }
 544       if (
 545         this.workerChoiceStrategyContext.getTaskStatisticsRequirements()
 546           .waitTime.median &&
 547         taskWaitTime != null
 548       ) {
 549         workerUsage.waitTime.history.push(taskWaitTime)
 550         workerUsage.waitTime.median = median(workerUsage.waitTime.history)
 551       }
 552     }
 553   }
 554
 555   private updateEluWorkerUsage (
 556     workerUsage: WorkerUsage,
 557     message: MessageValue<Response>
 558   ): void {
 559     if (
 560       this.workerChoiceStrategyContext.getTaskStatisticsRequirements().elu
 561         .aggregate
 562     ) {
 563       if (workerUsage.elu != null && message.taskPerformance?.elu != null) {
 564         workerUsage.elu.idle.aggregate += message.taskPerformance.elu.idle
 565         workerUsage.elu.active.aggregate += message.taskPerformance.elu.active
 566         workerUsage.elu.utilization =
 567           (workerUsage.elu.utilization +
 568             message.taskPerformance.elu.utilization) /
 569           2
 570       } else if (message.taskPerformance?.elu != null) {
 571         workerUsage.elu.idle.aggregate = message.taskPerformance.elu.idle
 572         workerUsage.elu.active.aggregate = message.taskPerformance.elu.active
 573         workerUsage.elu.utilization = message.taskPerformance.elu.utilization
 574       }
 575       if (
 576         this.workerChoiceStrategyContext.getTaskStatisticsRequirements().elu
 577           .average &&
 578         workerUsage.tasks.executed !== 0
 579       ) {
 580         workerUsage.elu.idle.average =
 581           workerUsage.elu.idle.aggregate / workerUsage.tasks.executed
 582         workerUsage.elu.active.average =
 583           workerUsage.elu.active.aggregate / workerUsage.tasks.executed
 584       }
 585       if (
 586         this.workerChoiceStrategyContext.getTaskStatisticsRequirements().elu
 587           .median &&
 588         message.taskPerformance?.elu != null
 589       ) {
 590         workerUsage.elu.idle.history.push(message.taskPerformance.elu.idle)
 591         workerUsage.elu.active.history.push(message.taskPerformance.elu.active)
 592         workerUsage.elu.idle.median = median(workerUsage.elu.idle.history)
 593         workerUsage.elu.active.median = median(workerUsage.elu.active.history)
 594       }
 595     }
 596   }
 597
 598   /**
 599    * Chooses a worker node for the next task.
 600    *
 601    * The default worker choice strategy uses a round robin algorithm to distribute the load.
 602    *
 603    * @returns The worker node key
 604    */
 605   protected chooseWorkerNode (): number {
 606     let workerNodeKey: number
 607     if (this.type === PoolTypes.dynamic && !this.full && this.internalBusy()) {
 608       const workerCreated = this.createAndSetupWorker()
 609       this.registerWorkerMessageListener(workerCreated, message => {
 610         const currentWorkerNodeKey = this.getWorkerNodeKey(workerCreated)
 611         if (
 612           isKillBehavior(KillBehaviors.HARD, message.kill) ||
 613           (message.kill != null &&
 614             this.workerNodes[currentWorkerNodeKey].workerUsage.tasks
 615               .executing === 0)
 616         ) {
 617           // Kill message received from the worker: no new tasks are submitted to that worker for a while ( > maxInactiveTime)
 618           this.flushTasksQueue(currentWorkerNodeKey)
 619           // FIXME: wait for tasks to be finished
 620           void (this.destroyWorker(workerCreated) as Promise<void>)
 621         }
 622       })
 623       workerNodeKey = this.getWorkerNodeKey(workerCreated)
 624     } else {
 625       workerNodeKey = this.workerChoiceStrategyContext.execute()
 626     }
 627     return workerNodeKey
 628   }
 629
 630   /**
 631    * Sends a message to the given worker.
 632    *
 633    * @param worker - The worker which should receive the message.
 634    * @param message - The message.
 635    */
 636   protected abstract sendToWorker (
 637     worker: Worker,
 638     message: MessageValue<Data>
 639   ): void
 640
 641   /**
 642    * Registers a listener callback on the given worker.
 643    *
 644    * @param worker - The worker which should register a listener.
 645    * @param listener - The message listener callback.
 646    */
 647   protected abstract registerWorkerMessageListener<
 648     Message extends Data | Response
 649   >(worker: Worker, listener: (message: MessageValue<Message>) => void): void
 650
 651   /**
 652    * Returns a newly created worker.
 653    */
 654   protected abstract createWorker (): Worker
 655
 656   /**
 657    * Function that can be hooked up when a worker has been newly created and moved to the pool worker nodes.
 658    *
 659    * Can be used to update the `maxListeners` or binding the `main-worker`\<-\>`worker` connection if not bind by default.
 660    *
 661    * @param worker - The newly created worker.
 662    */
 663   protected abstract afterWorkerSetup (worker: Worker): void
 664
 665   /**
 666    * Creates a new worker and sets it up completely in the pool worker nodes.
 667    *
 668    * @returns New, completely set up worker.
 669    */
 670   protected createAndSetupWorker (): Worker {
 671     const worker = this.createWorker()
 672
 673     worker.on('message', this.opts.messageHandler ?? EMPTY_FUNCTION)
 674     worker.on('error', this.opts.errorHandler ?? EMPTY_FUNCTION)
 675     worker.on('error', error => {
 676       if (this.emitter != null) {
 677         this.emitter.emit(PoolEvents.error, error)
 678       }
 679     })
 680     worker.on('error', () => {
 681       if (this.opts.restartWorkerOnError === true) {
 682         this.createAndSetupWorker()
 683       }
 684     })
 685     worker.on('online', this.opts.onlineHandler ?? EMPTY_FUNCTION)
 686     worker.on('exit', this.opts.exitHandler ?? EMPTY_FUNCTION)
 687     worker.once('exit', () => {
 688       this.removeWorkerNode(worker)
 689     })
 690
 691     this.pushWorkerNode(worker)
 692
 693     this.setWorkerStatistics(worker)
 694
 695     this.afterWorkerSetup(worker)
 696
 697     return worker
 698   }
 699
 700   /**
 701    * This function is the listener registered for each worker message.
 702    *
 703    * @returns The listener function to execute when a message is received from a worker.
 704    */
 705   protected workerListener (): (message: MessageValue<Response>) => void {
 706     return message => {
 707       if (message.id != null) {
 708         // Task execution response received
 709         const promiseResponse = this.promiseResponseMap.get(message.id)
 710         if (promiseResponse != null) {
 711           if (message.taskError != null) {
 712             promiseResponse.reject(message.taskError.message)
 713             if (this.emitter != null) {
 714               this.emitter.emit(PoolEvents.taskError, message.taskError)
 715             }
 716           } else {
 717             promiseResponse.resolve(message.data as Response)
 718           }
 719           this.afterTaskExecutionHook(promiseResponse.worker, message)
 720           this.promiseResponseMap.delete(message.id)
 721           const workerNodeKey = this.getWorkerNodeKey(promiseResponse.worker)
 722           if (
 723             this.opts.enableTasksQueue === true &&
 724             this.tasksQueueSize(workerNodeKey) > 0
 725           ) {
 726             this.executeTask(
 727               workerNodeKey,
 728               this.dequeueTask(workerNodeKey) as Task<Data>
 729             )
 730           }
 731         }
 732       }
 733     }
 734   }
 735
 736   private checkAndEmitEvents (): void {
 737     if (this.emitter != null) {
 738       if (this.busy) {
 739         this.emitter?.emit(PoolEvents.busy, this.info)
 740       }
 741       if (this.type === PoolTypes.dynamic && this.full) {
 742         this.emitter?.emit(PoolEvents.full, this.info)
 743       }
 744     }
 745   }
 746
 747   /**
 748    * Sets the given worker node its tasks usage in the pool.
 749    *
 750    * @param workerNode - The worker node.
 751    * @param workerUsage - The worker usage.
 752    */
 753   private setWorkerNodeTasksUsage (
 754     workerNode: WorkerNode<Worker, Data>,
 755     workerUsage: WorkerUsage
 756   ): void {
 757     workerNode.workerUsage = workerUsage
 758   }
 759
 760   /**
 761    * Pushes the given worker in the pool worker nodes.
 762    *
 763    * @param worker - The worker.
 764    * @returns The worker nodes length.
 765    */
 766   private pushWorkerNode (worker: Worker): number {
 767     return this.workerNodes.push({
 768       worker,
 769       workerUsage: this.getWorkerUsage(worker),
 770       tasksQueue: new Queue<Task<Data>>()
 771     })
 772   }
 773
 774   // /**
 775   //  * Sets the given worker in the pool worker nodes.
 776   //  *
 777   //  * @param workerNodeKey - The worker node key.
 778   //  * @param worker - The worker.
 779   //  * @param workerUsage - The worker usage.
 780   //  * @param tasksQueue - The worker task queue.
 781   //  */
 782   // private setWorkerNode (
 783   //   workerNodeKey: number,
 784   //   worker: Worker,
 785   //   workerUsage: WorkerUsage,
 786   //   tasksQueue: Queue<Task<Data>>
 787   // ): void {
 788   //   this.workerNodes[workerNodeKey] = {
 789   //     worker,
 790   //     workerUsage,
 791   //     tasksQueue
 792   //   }
 793   // }
 794
 795   /**
 796    * Removes the given worker from the pool worker nodes.
 797    *
 798    * @param worker - The worker.
 799    */
 800   private removeWorkerNode (worker: Worker): void {
 801     const workerNodeKey = this.getWorkerNodeKey(worker)
 802     if (workerNodeKey !== -1) {
 803       this.workerNodes.splice(workerNodeKey, 1)
 804       this.workerChoiceStrategyContext.remove(workerNodeKey)
 805     }
 806   }
 807
 808   private executeTask (workerNodeKey: number, task: Task<Data>): void {
 809     this.beforeTaskExecutionHook(workerNodeKey, task)
 810     this.sendToWorker(this.workerNodes[workerNodeKey].worker, task)
 811   }
 812
 813   private enqueueTask (workerNodeKey: number, task: Task<Data>): number {
 814     return this.workerNodes[workerNodeKey].tasksQueue.enqueue(task)
 815   }
 816
 817   private dequeueTask (workerNodeKey: number): Task<Data> | undefined {
 818     return this.workerNodes[workerNodeKey].tasksQueue.dequeue()
 819   }
 820
 821   private tasksQueueSize (workerNodeKey: number): number {
 822     return this.workerNodes[workerNodeKey].tasksQueue.size
 823   }
 824
 825   private flushTasksQueue (workerNodeKey: number): void {
 826     if (this.tasksQueueSize(workerNodeKey) > 0) {
 827       for (let i = 0; i < this.tasksQueueSize(workerNodeKey); i++) {
 828         this.executeTask(
 829           workerNodeKey,
 830           this.dequeueTask(workerNodeKey) as Task<Data>
 831         )
 832       }
 833     }
 834   }
 835
 836   private flushTasksQueues (): void {
 837     for (const [workerNodeKey] of this.workerNodes.entries()) {
 838       this.flushTasksQueue(workerNodeKey)
 839     }
 840   }
 841
 842   private setWorkerStatistics (worker: Worker): void {
 843     this.sendToWorker(worker, {
 844       statistics: {
 845         runTime:
 846           this.workerChoiceStrategyContext.getTaskStatisticsRequirements()
 847             .runTime.aggregate,
 848         elu: this.workerChoiceStrategyContext.getTaskStatisticsRequirements()
 849           .elu.aggregate
 850       }
 851     })
 852   }
 853
 854   private getWorkerUsage (worker: Worker): WorkerUsage {
 855     return {
 856       tasks: this.getTaskStatistics(worker),
 857       runTime: {
 858         aggregate: 0,
 859         average: 0,
 860         median: 0,
 861         history: new CircularArray()
 862       },
 863       waitTime: {
 864         aggregate: 0,
 865         average: 0,
 866         median: 0,
 867         history: new CircularArray()
 868       },
 869       elu: {
 870         idle: {
 871           aggregate: 0,
 872           average: 0,
 873           median: 0,
 874           history: new CircularArray()
 875         },
 876         active: {
 877           aggregate: 0,
 878           average: 0,
 879           median: 0,
 880           history: new CircularArray()
 881         },
 882         utilization: 0
 883       }
 884     }
 885   }
 886
 887   private getTaskStatistics (worker: Worker): TaskStatistics {
 888     const queueSize =
 889       this.workerNodes[this.getWorkerNodeKey(worker)]?.tasksQueue?.size
 890     return {
 891       executed: 0,
 892       executing: 0,
 893       get queued (): number {
 894         return queueSize ?? 0
 895       },
 896       failed: 0
 897     }
 898   }
 899 }