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     this.updateTaskStatisticsWorkerUsage(workerUsage, message)
 488     this.updateRunTimeWorkerUsage(workerUsage, message)
 489     this.updateEluWorkerUsage(workerUsage, message)
 490   }
 491
 492   private updateTaskStatisticsWorkerUsage (
 493     workerUsage: WorkerUsage,
 494     message: MessageValue<Response>
 495   ): void {
 496     const workerTaskStatistics = workerUsage.tasks
 497     --workerTaskStatistics.executing
 498     ++workerTaskStatistics.executed
 499     if (message.taskError != null) {
 500       ++workerTaskStatistics.failed
 501     }
 502   }
 503
 504   private updateRunTimeWorkerUsage (
 505     workerUsage: WorkerUsage,
 506     message: MessageValue<Response>
 507   ): void {
 508     if (
 509       this.workerChoiceStrategyContext.getTaskStatisticsRequirements().runTime
 510         .aggregate
 511     ) {
 512       workerUsage.runTime.aggregate += message.taskPerformance?.runTime ?? 0
 513       if (
 514         this.workerChoiceStrategyContext.getTaskStatisticsRequirements().runTime
 515           .average &&
 516         workerUsage.tasks.executed !== 0
 517       ) {
 518         workerUsage.runTime.average =
 519           workerUsage.runTime.aggregate /
 520           (workerUsage.tasks.executed - workerUsage.tasks.failed)
 521       }
 522       if (
 523         this.workerChoiceStrategyContext.getTaskStatisticsRequirements().runTime
 524           .median &&
 525         message.taskPerformance?.runTime != null
 526       ) {
 527         workerUsage.runTime.history.push(message.taskPerformance.runTime)
 528         workerUsage.runTime.median = median(workerUsage.runTime.history)
 529       }
 530     }
 531   }
 532
 533   private updateWaitTimeWorkerUsage (
 534     workerUsage: WorkerUsage,
 535     task: Task<Data>
 536   ): void {
 537     const timestamp = performance.now()
 538     const taskWaitTime = timestamp - (task.timestamp ?? timestamp)
 539     if (
 540       this.workerChoiceStrategyContext.getTaskStatisticsRequirements().waitTime
 541         .aggregate
 542     ) {
 543       workerUsage.waitTime.aggregate += taskWaitTime ?? 0
 544       if (
 545         this.workerChoiceStrategyContext.getTaskStatisticsRequirements()
 546           .waitTime.average &&
 547         workerUsage.tasks.executed !== 0
 548       ) {
 549         workerUsage.waitTime.average =
 550           workerUsage.waitTime.aggregate /
 551           (workerUsage.tasks.executed - workerUsage.tasks.failed)
 552       }
 553       if (
 554         this.workerChoiceStrategyContext.getTaskStatisticsRequirements()
 555           .waitTime.median &&
 556         taskWaitTime != null
 557       ) {
 558         workerUsage.waitTime.history.push(taskWaitTime)
 559         workerUsage.waitTime.median = median(workerUsage.waitTime.history)
 560       }
 561     }
 562   }
 563
 564   private updateEluWorkerUsage (
 565     workerUsage: WorkerUsage,
 566     message: MessageValue<Response>
 567   ): void {
 568     if (
 569       this.workerChoiceStrategyContext.getTaskStatisticsRequirements().elu
 570         .aggregate
 571     ) {
 572       if (workerUsage.elu != null && 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 =
 576           (workerUsage.elu.utilization +
 577             message.taskPerformance.elu.utilization) /
 578           2
 579       } else if (message.taskPerformance?.elu != null) {
 580         workerUsage.elu.idle.aggregate = message.taskPerformance.elu.idle
 581         workerUsage.elu.active.aggregate = message.taskPerformance.elu.active
 582         workerUsage.elu.utilization = message.taskPerformance.elu.utilization
 583       }
 584       if (
 585         this.workerChoiceStrategyContext.getTaskStatisticsRequirements().elu
 586           .average &&
 587         workerUsage.tasks.executed !== 0
 588       ) {
 589         const executedTasks =
 590           workerUsage.tasks.executed - workerUsage.tasks.failed
 591         workerUsage.elu.idle.average =
 592           workerUsage.elu.idle.aggregate / executedTasks
 593         workerUsage.elu.active.average =
 594           workerUsage.elu.active.aggregate / executedTasks
 595       }
 596       if (
 597         this.workerChoiceStrategyContext.getTaskStatisticsRequirements().elu
 598           .median &&
 599         message.taskPerformance?.elu != null
 600       ) {
 601         workerUsage.elu.idle.history.push(message.taskPerformance.elu.idle)
 602         workerUsage.elu.active.history.push(message.taskPerformance.elu.active)
 603         workerUsage.elu.idle.median = median(workerUsage.elu.idle.history)
 604         workerUsage.elu.active.median = median(workerUsage.elu.active.history)
 605       }
 606     }
 607   }
 608
 609   /**
 610    * Chooses a worker node for the next task.
 611    *
 612    * The default worker choice strategy uses a round robin algorithm to distribute the load.
 613    *
 614    * @returns The worker node key
 615    */
 616   protected chooseWorkerNode (): number {
 617     let workerNodeKey: number
 618     if (this.type === PoolTypes.dynamic && !this.full && this.internalBusy()) {
 619       const workerCreated = this.createAndSetupWorker()
 620       this.registerWorkerMessageListener(workerCreated, message => {
 621         const currentWorkerNodeKey = this.getWorkerNodeKey(workerCreated)
 622         if (
 623           isKillBehavior(KillBehaviors.HARD, message.kill) ||
 624           (message.kill != null &&
 625             this.workerNodes[currentWorkerNodeKey].workerUsage.tasks
 626               .executing === 0)
 627         ) {
 628           // Kill message received from the worker: no new tasks are submitted to that worker for a while ( > maxInactiveTime)
 629           this.flushTasksQueue(currentWorkerNodeKey)
 630           // FIXME: wait for tasks to be finished
 631           void (this.destroyWorker(workerCreated) as Promise<void>)
 632         }
 633       })
 634       workerNodeKey = this.getWorkerNodeKey(workerCreated)
 635     } else {
 636       workerNodeKey = this.workerChoiceStrategyContext.execute()
 637     }
 638     return workerNodeKey
 639   }
 640
 641   /**
 642    * Sends a message to the given worker.
 643    *
 644    * @param worker - The worker which should receive the message.
 645    * @param message - The message.
 646    */
 647   protected abstract sendToWorker (
 648     worker: Worker,
 649     message: MessageValue<Data>
 650   ): void
 651
 652   /**
 653    * Registers a listener callback on the given worker.
 654    *
 655    * @param worker - The worker which should register a listener.
 656    * @param listener - The message listener callback.
 657    */
 658   protected abstract registerWorkerMessageListener<
 659     Message extends Data | Response
 660   >(worker: Worker, listener: (message: MessageValue<Message>) => void): void
 661
 662   /**
 663    * Returns a newly created worker.
 664    */
 665   protected abstract createWorker (): Worker
 666
 667   /**
 668    * Function that can be hooked up when a worker has been newly created and moved to the pool worker nodes.
 669    *
 670    * Can be used to update the `maxListeners` or binding the `main-worker`\<-\>`worker` connection if not bind by default.
 671    *
 672    * @param worker - The newly created worker.
 673    */
 674   protected abstract afterWorkerSetup (worker: Worker): void
 675
 676   /**
 677    * Creates a new worker and sets it up completely in the pool worker nodes.
 678    *
 679    * @returns New, completely set up worker.
 680    */
 681   protected createAndSetupWorker (): Worker {
 682     const worker = this.createWorker()
 683
 684     worker.on('message', this.opts.messageHandler ?? EMPTY_FUNCTION)
 685     worker.on('error', this.opts.errorHandler ?? EMPTY_FUNCTION)
 686     worker.on('error', error => {
 687       if (this.emitter != null) {
 688         this.emitter.emit(PoolEvents.error, error)
 689       }
 690       if (this.opts.restartWorkerOnError === true) {
 691         this.createAndSetupWorker()
 692       }
 693     })
 694     worker.on('online', this.opts.onlineHandler ?? EMPTY_FUNCTION)
 695     worker.on('exit', this.opts.exitHandler ?? EMPTY_FUNCTION)
 696     worker.once('exit', () => {
 697       this.removeWorkerNode(worker)
 698     })
 699
 700     this.pushWorkerNode(worker)
 701
 702     this.setWorkerStatistics(worker)
 703
 704     this.afterWorkerSetup(worker)
 705
 706     return worker
 707   }
 708
 709   /**
 710    * This function is the listener registered for each worker message.
 711    *
 712    * @returns The listener function to execute when a message is received from a worker.
 713    */
 714   protected workerListener (): (message: MessageValue<Response>) => void {
 715     return message => {
 716       if (message.id != null) {
 717         // Task execution response received
 718         const promiseResponse = this.promiseResponseMap.get(message.id)
 719         if (promiseResponse != null) {
 720           if (message.taskError != null) {
 721             promiseResponse.reject(message.taskError.message)
 722             if (this.emitter != null) {
 723               this.emitter.emit(PoolEvents.taskError, message.taskError)
 724             }
 725           } else {
 726             promiseResponse.resolve(message.data as Response)
 727           }
 728           this.afterTaskExecutionHook(promiseResponse.worker, message)
 729           this.promiseResponseMap.delete(message.id)
 730           const workerNodeKey = this.getWorkerNodeKey(promiseResponse.worker)
 731           if (
 732             this.opts.enableTasksQueue === true &&
 733             this.tasksQueueSize(workerNodeKey) > 0
 734           ) {
 735             this.executeTask(
 736               workerNodeKey,
 737               this.dequeueTask(workerNodeKey) as Task<Data>
 738             )
 739           }
 740         }
 741       }
 742     }
 743   }
 744
 745   private checkAndEmitEvents (): void {
 746     if (this.emitter != null) {
 747       if (this.busy) {
 748         this.emitter?.emit(PoolEvents.busy, this.info)
 749       }
 750       if (this.type === PoolTypes.dynamic && this.full) {
 751         this.emitter?.emit(PoolEvents.full, this.info)
 752       }
 753     }
 754   }
 755
 756   /**
 757    * Sets the given worker node its tasks usage in the pool.
 758    *
 759    * @param workerNode - The worker node.
 760    * @param workerUsage - The worker usage.
 761    */
 762   private setWorkerNodeTasksUsage (
 763     workerNode: WorkerNode<Worker, Data>,
 764     workerUsage: WorkerUsage
 765   ): void {
 766     workerNode.workerUsage = workerUsage
 767   }
 768
 769   /**
 770    * Pushes the given worker in the pool worker nodes.
 771    *
 772    * @param worker - The worker.
 773    * @returns The worker nodes length.
 774    */
 775   private pushWorkerNode (worker: Worker): number {
 776     return this.workerNodes.push({
 777       worker,
 778       workerUsage: this.getWorkerUsage(worker),
 779       tasksQueue: new Queue<Task<Data>>()
 780     })
 781   }
 782
 783   // /**
 784   //  * Sets the given worker in the pool worker nodes.
 785   //  *
 786   //  * @param workerNodeKey - The worker node key.
 787   //  * @param worker - The worker.
 788   //  * @param workerUsage - The worker usage.
 789   //  * @param tasksQueue - The worker task queue.
 790   //  */
 791   // private setWorkerNode (
 792   //   workerNodeKey: number,
 793   //   worker: Worker,
 794   //   workerUsage: WorkerUsage,
 795   //   tasksQueue: Queue<Task<Data>>
 796   // ): void {
 797   //   this.workerNodes[workerNodeKey] = {
 798   //     worker,
 799   //     workerUsage,
 800   //     tasksQueue
 801   //   }
 802   // }
 803
 804   /**
 805    * Removes the given worker from the pool worker nodes.
 806    *
 807    * @param worker - The worker.
 808    */
 809   private removeWorkerNode (worker: Worker): void {
 810     const workerNodeKey = this.getWorkerNodeKey(worker)
 811     if (workerNodeKey !== -1) {
 812       this.workerNodes.splice(workerNodeKey, 1)
 813       this.workerChoiceStrategyContext.remove(workerNodeKey)
 814     }
 815   }
 816
 817   private executeTask (workerNodeKey: number, task: Task<Data>): void {
 818     this.beforeTaskExecutionHook(workerNodeKey, task)
 819     this.sendToWorker(this.workerNodes[workerNodeKey].worker, task)
 820   }
 821
 822   private enqueueTask (workerNodeKey: number, task: Task<Data>): number {
 823     return this.workerNodes[workerNodeKey].tasksQueue.enqueue(task)
 824   }
 825
 826   private dequeueTask (workerNodeKey: number): Task<Data> | undefined {
 827     return this.workerNodes[workerNodeKey].tasksQueue.dequeue()
 828   }
 829
 830   private tasksQueueSize (workerNodeKey: number): number {
 831     return this.workerNodes[workerNodeKey].tasksQueue.size
 832   }
 833
 834   private flushTasksQueue (workerNodeKey: number): void {
 835     if (this.tasksQueueSize(workerNodeKey) > 0) {
 836       for (let i = 0; i < this.tasksQueueSize(workerNodeKey); i++) {
 837         this.executeTask(
 838           workerNodeKey,
 839           this.dequeueTask(workerNodeKey) as Task<Data>
 840         )
 841       }
 842     }
 843   }
 844
 845   private flushTasksQueues (): void {
 846     for (const [workerNodeKey] of this.workerNodes.entries()) {
 847       this.flushTasksQueue(workerNodeKey)
 848     }
 849   }
 850
 851   private setWorkerStatistics (worker: Worker): void {
 852     this.sendToWorker(worker, {
 853       statistics: {
 854         runTime:
 855           this.workerChoiceStrategyContext.getTaskStatisticsRequirements()
 856             .runTime.aggregate,
 857         elu: this.workerChoiceStrategyContext.getTaskStatisticsRequirements()
 858           .elu.aggregate
 859       }
 860     })
 861   }
 862
 863   private getWorkerUsage (worker: Worker): WorkerUsage {
 864     return {
 865       tasks: this.getTaskStatistics(worker),
 866       runTime: {
 867         aggregate: 0,
 868         average: 0,
 869         median: 0,
 870         history: new CircularArray()
 871       },
 872       waitTime: {
 873         aggregate: 0,
 874         average: 0,
 875         median: 0,
 876         history: new CircularArray()
 877       },
 878       elu: {
 879         idle: {
 880           aggregate: 0,
 881           average: 0,
 882           median: 0,
 883           history: new CircularArray()
 884         },
 885         active: {
 886           aggregate: 0,
 887           average: 0,
 888           median: 0,
 889           history: new CircularArray()
 890         },
 891         utilization: 0
 892       }
 893     }
 894   }
 895
 896   private getTaskStatistics (worker: Worker): TaskStatistics {
 897     const queueSize =
 898       this.workerNodes[this.getWorkerNodeKey(worker)]?.tasksQueue?.size
 899     return {
 900       executed: 0,
 901       executing: 0,
 902       get queued (): number {
 903         return queueSize ?? 0
 904       },
 905       failed: 0
 906     }
 907   }
 908 }