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