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   /**
 383    * Whether worker nodes are executing at least one task.
 384    *
 385    * @returns Worker nodes busyness boolean status.
 386    */
 387   protected internalBusy (): boolean {
 388     return (
 389       this.workerNodes.findIndex(workerNode => {
 390         return workerNode.workerUsage.tasks.executing === 0
 391       }) === -1
 392     )
 393   }
 394
 395   /** @inheritDoc */
 396   public async execute (data?: Data, name?: string): Promise<Response> {
 397     const timestamp = performance.now()
 398     const workerNodeKey = this.chooseWorkerNode()
 399     const submittedTask: Task<Data> = {
 400       name,
 401       // eslint-disable-next-line @typescript-eslint/consistent-type-assertions
 402       data: data ?? ({} as Data),
 403       timestamp,
 404       id: crypto.randomUUID()
 405     }
 406     const res = new Promise<Response>((resolve, reject) => {
 407       this.promiseResponseMap.set(submittedTask.id as string, {
 408         resolve,
 409         reject,
 410         worker: this.workerNodes[workerNodeKey].worker
 411       })
 412     })
 413     if (
 414       this.opts.enableTasksQueue === true &&
 415       (this.busy ||
 416         this.workerNodes[workerNodeKey].workerUsage.tasks.executing >=
 417           ((this.opts.tasksQueueOptions as TasksQueueOptions)
 418             .concurrency as number))
 419     ) {
 420       this.enqueueTask(workerNodeKey, submittedTask)
 421     } else {
 422       this.executeTask(workerNodeKey, submittedTask)
 423     }
 424     this.checkAndEmitEvents()
 425     // eslint-disable-next-line @typescript-eslint/return-await
 426     return res
 427   }
 428
 429   /** @inheritDoc */
 430   public async destroy (): Promise<void> {
 431     await Promise.all(
 432       this.workerNodes.map(async (workerNode, workerNodeKey) => {
 433         this.flushTasksQueue(workerNodeKey)
 434         // FIXME: wait for tasks to be finished
 435         await this.destroyWorker(workerNode.worker)
 436       })
 437     )
 438   }
 439
 440   /**
 441    * Terminates the given worker.
 442    *
 443    * @param worker - A worker within `workerNodes`.
 444    */
 445   protected abstract destroyWorker (worker: Worker): void | Promise<void>
 446
 447   /**
 448    * Setup hook to execute code before worker node are created in the abstract constructor.
 449    * Can be overridden
 450    *
 451    * @virtual
 452    */
 453   protected setupHook (): void {
 454     // Intentionally empty
 455   }
 456
 457   /**
 458    * Should return whether the worker is the main worker or not.
 459    */
 460   protected abstract isMain (): boolean
 461
 462   /**
 463    * Hook executed before the worker task execution.
 464    * Can be overridden.
 465    *
 466    * @param workerNodeKey - The worker node key.
 467    * @param task - The task to execute.
 468    */
 469   protected beforeTaskExecutionHook (
 470     workerNodeKey: number,
 471     task: Task<Data>
 472   ): void {
 473     const workerUsage = this.workerNodes[workerNodeKey].workerUsage
 474     ++workerUsage.tasks.executing
 475     this.updateWaitTimeWorkerUsage(workerUsage, task)
 476   }
 477
 478   /**
 479    * Hook executed after the worker task execution.
 480    * Can be overridden.
 481    *
 482    * @param worker - The worker.
 483    * @param message - The received message.
 484    */
 485   protected afterTaskExecutionHook (
 486     worker: Worker,
 487     message: MessageValue<Response>
 488   ): void {
 489     const workerUsage =
 490       this.workerNodes[this.getWorkerNodeKey(worker)].workerUsage
 491     this.updateTaskStatisticsWorkerUsage(workerUsage, message)
 492     this.updateRunTimeWorkerUsage(workerUsage, message)
 493     this.updateEluWorkerUsage(workerUsage, message)
 494   }
 495
 496   private updateTaskStatisticsWorkerUsage (
 497     workerUsage: WorkerUsage,
 498     message: MessageValue<Response>
 499   ): void {
 500     const workerTaskStatistics = workerUsage.tasks
 501     --workerTaskStatistics.executing
 502     ++workerTaskStatistics.executed
 503     if (message.taskError != null) {
 504       ++workerTaskStatistics.failed
 505     }
 506   }
 507
 508   private updateRunTimeWorkerUsage (
 509     workerUsage: WorkerUsage,
 510     message: MessageValue<Response>
 511   ): void {
 512     if (
 513       this.workerChoiceStrategyContext.getTaskStatisticsRequirements().runTime
 514         .aggregate
 515     ) {
 516       workerUsage.runTime.aggregate += message.taskPerformance?.runTime ?? 0
 517       if (
 518         this.workerChoiceStrategyContext.getTaskStatisticsRequirements().runTime
 519           .average &&
 520         workerUsage.tasks.executed !== 0
 521       ) {
 522         workerUsage.runTime.average =
 523           workerUsage.runTime.aggregate /
 524           (workerUsage.tasks.executed - workerUsage.tasks.failed)
 525       }
 526       if (
 527         this.workerChoiceStrategyContext.getTaskStatisticsRequirements().runTime
 528           .median &&
 529         message.taskPerformance?.runTime != null
 530       ) {
 531         workerUsage.runTime.history.push(message.taskPerformance.runTime)
 532         workerUsage.runTime.median = median(workerUsage.runTime.history)
 533       }
 534     }
 535   }
 536
 537   private updateWaitTimeWorkerUsage (
 538     workerUsage: WorkerUsage,
 539     task: Task<Data>
 540   ): void {
 541     const timestamp = performance.now()
 542     const taskWaitTime = timestamp - (task.timestamp ?? timestamp)
 543     if (
 544       this.workerChoiceStrategyContext.getTaskStatisticsRequirements().waitTime
 545         .aggregate
 546     ) {
 547       workerUsage.waitTime.aggregate += taskWaitTime ?? 0
 548       if (
 549         this.workerChoiceStrategyContext.getTaskStatisticsRequirements()
 550           .waitTime.average &&
 551         workerUsage.tasks.executed !== 0
 552       ) {
 553         workerUsage.waitTime.average =
 554           workerUsage.waitTime.aggregate /
 555           (workerUsage.tasks.executed - workerUsage.tasks.failed)
 556       }
 557       if (
 558         this.workerChoiceStrategyContext.getTaskStatisticsRequirements()
 559           .waitTime.median &&
 560         taskWaitTime != null
 561       ) {
 562         workerUsage.waitTime.history.push(taskWaitTime)
 563         workerUsage.waitTime.median = median(workerUsage.waitTime.history)
 564       }
 565     }
 566   }
 567
 568   private updateEluWorkerUsage (
 569     workerUsage: WorkerUsage,
 570     message: MessageValue<Response>
 571   ): void {
 572     if (
 573       this.workerChoiceStrategyContext.getTaskStatisticsRequirements().elu
 574         .aggregate
 575     ) {
 576       if (workerUsage.elu != null && message.taskPerformance?.elu != null) {
 577         workerUsage.elu.idle.aggregate += message.taskPerformance.elu.idle
 578         workerUsage.elu.active.aggregate += message.taskPerformance.elu.active
 579         workerUsage.elu.utilization =
 580           (workerUsage.elu.utilization +
 581             message.taskPerformance.elu.utilization) /
 582           2
 583       } else if (message.taskPerformance?.elu != null) {
 584         workerUsage.elu.idle.aggregate = message.taskPerformance.elu.idle
 585         workerUsage.elu.active.aggregate = message.taskPerformance.elu.active
 586         workerUsage.elu.utilization = message.taskPerformance.elu.utilization
 587       }
 588       if (
 589         this.workerChoiceStrategyContext.getTaskStatisticsRequirements().elu
 590           .average &&
 591         workerUsage.tasks.executed !== 0
 592       ) {
 593         const executedTasks =
 594           workerUsage.tasks.executed - workerUsage.tasks.failed
 595         workerUsage.elu.idle.average =
 596           workerUsage.elu.idle.aggregate / executedTasks
 597         workerUsage.elu.active.average =
 598           workerUsage.elu.active.aggregate / executedTasks
 599       }
 600       if (
 601         this.workerChoiceStrategyContext.getTaskStatisticsRequirements().elu
 602           .median &&
 603         message.taskPerformance?.elu != null
 604       ) {
 605         workerUsage.elu.idle.history.push(message.taskPerformance.elu.idle)
 606         workerUsage.elu.active.history.push(message.taskPerformance.elu.active)
 607         workerUsage.elu.idle.median = median(workerUsage.elu.idle.history)
 608         workerUsage.elu.active.median = median(workerUsage.elu.active.history)
 609       }
 610     }
 611   }
 612
 613   /**
 614    * Chooses a worker node for the next task.
 615    *
 616    * The default worker choice strategy uses a round robin algorithm to distribute the tasks.
 617    *
 618    * @returns The worker node key
 619    */
 620   private chooseWorkerNode (): number {
 621     if (this.shallCreateDynamicWorker()) {
 622       const worker = this.createAndSetupDynamicWorker()
 623       if (
 624         this.workerChoiceStrategyContext.getStrategyPolicy().useDynamicWorker
 625       ) {
 626         return this.getWorkerNodeKey(worker)
 627       }
 628     }
 629     return this.workerChoiceStrategyContext.execute()
 630   }
 631
 632   /**
 633    * Conditions for dynamic worker creation.
 634    *
 635    * @returns Whether to create a dynamic worker or not.
 636    */
 637   private shallCreateDynamicWorker (): boolean {
 638     return this.type === PoolTypes.dynamic && !this.full && this.internalBusy()
 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    * Creates a new worker.
 664    *
 665    * @returns Newly created worker.
 666    */
 667   protected abstract createWorker (): Worker
 668
 669   /**
 670    * Function that can be hooked up when a worker has been newly created and moved to the pool worker nodes.
 671    *
 672    * Can be used to update the `maxListeners` or binding the `main-worker`\<-\>`worker` connection if not bind by default.
 673    *
 674    * @param worker - The newly created worker.
 675    */
 676   protected abstract afterWorkerSetup (worker: Worker): void
 677
 678   /**
 679    * Creates a new worker and sets it up completely in the pool worker nodes.
 680    *
 681    * @returns New, completely set up worker.
 682    */
 683   protected createAndSetupWorker (): Worker {
 684     const worker = this.createWorker()
 685
 686     worker.on('message', this.opts.messageHandler ?? EMPTY_FUNCTION)
 687     worker.on('error', this.opts.errorHandler ?? EMPTY_FUNCTION)
 688     worker.on('error', error => {
 689       if (this.emitter != null) {
 690         this.emitter.emit(PoolEvents.error, error)
 691       }
 692       if (this.opts.restartWorkerOnError === true) {
 693         this.createAndSetupWorker()
 694       }
 695     })
 696     worker.on('online', this.opts.onlineHandler ?? EMPTY_FUNCTION)
 697     worker.on('exit', this.opts.exitHandler ?? EMPTY_FUNCTION)
 698     worker.once('exit', () => {
 699       this.removeWorkerNode(worker)
 700     })
 701
 702     this.pushWorkerNode(worker)
 703
 704     this.setWorkerStatistics(worker)
 705
 706     this.afterWorkerSetup(worker)
 707
 708     return worker
 709   }
 710
 711   /**
 712    * Creates a new dynamic worker and sets it up completely in the pool worker nodes.
 713    *
 714    * @returns New, completely set up dynamic worker.
 715    */
 716   protected createAndSetupDynamicWorker (): Worker {
 717     const worker = this.createAndSetupWorker()
 718     this.registerWorkerMessageListener(worker, message => {
 719       const workerNodeKey = this.getWorkerNodeKey(worker)
 720       if (
 721         isKillBehavior(KillBehaviors.HARD, message.kill) ||
 722         (message.kill != null &&
 723           ((this.opts.enableTasksQueue === false &&
 724             this.workerNodes[workerNodeKey].workerUsage.tasks.executing ===
 725               0) ||
 726             (this.opts.enableTasksQueue === true &&
 727               this.workerNodes[workerNodeKey].workerUsage.tasks.executing ===
 728                 0 &&
 729               this.tasksQueueSize(workerNodeKey) === 0)))
 730       ) {
 731         // Kill message received from the worker: no new tasks are submitted to that worker for a while ( > maxInactiveTime)
 732         void (this.destroyWorker(worker) as Promise<void>)
 733       }
 734     })
 735     return worker
 736   }
 737
 738   /**
 739    * This function is the listener registered for each worker message.
 740    *
 741    * @returns The listener function to execute when a message is received from a worker.
 742    */
 743   protected workerListener (): (message: MessageValue<Response>) => void {
 744     return message => {
 745       if (message.id != null) {
 746         // Task execution response received
 747         const promiseResponse = this.promiseResponseMap.get(message.id)
 748         if (promiseResponse != null) {
 749           if (message.taskError != null) {
 750             promiseResponse.reject(message.taskError.message)
 751             if (this.emitter != null) {
 752               this.emitter.emit(PoolEvents.taskError, message.taskError)
 753             }
 754           } else {
 755             promiseResponse.resolve(message.data as Response)
 756           }
 757           this.afterTaskExecutionHook(promiseResponse.worker, message)
 758           this.promiseResponseMap.delete(message.id)
 759           const workerNodeKey = this.getWorkerNodeKey(promiseResponse.worker)
 760           if (
 761             this.opts.enableTasksQueue === true &&
 762             this.tasksQueueSize(workerNodeKey) > 0
 763           ) {
 764             this.executeTask(
 765               workerNodeKey,
 766               this.dequeueTask(workerNodeKey) as Task<Data>
 767             )
 768           }
 769           this.workerChoiceStrategyContext.update(workerNodeKey)
 770         }
 771       }
 772     }
 773   }
 774
 775   private checkAndEmitEvents (): void {
 776     if (this.emitter != null) {
 777       if (this.busy) {
 778         this.emitter?.emit(PoolEvents.busy, this.info)
 779       }
 780       if (this.type === PoolTypes.dynamic && this.full) {
 781         this.emitter?.emit(PoolEvents.full, this.info)
 782       }
 783     }
 784   }
 785
 786   /**
 787    * Sets the given worker node its tasks usage in the pool.
 788    *
 789    * @param workerNode - The worker node.
 790    * @param workerUsage - The worker usage.
 791    */
 792   private setWorkerNodeTasksUsage (
 793     workerNode: WorkerNode<Worker, Data>,
 794     workerUsage: WorkerUsage
 795   ): void {
 796     workerNode.workerUsage = workerUsage
 797   }
 798
 799   /**
 800    * Pushes the given worker in the pool worker nodes.
 801    *
 802    * @param worker - The worker.
 803    * @returns The worker nodes length.
 804    */
 805   private pushWorkerNode (worker: Worker): number {
 806     return this.workerNodes.push({
 807       worker,
 808       workerUsage: this.getWorkerUsage(worker),
 809       tasksQueue: new Queue<Task<Data>>()
 810     })
 811   }
 812
 813   // /**
 814   //  * Sets the given worker in the pool worker nodes.
 815   //  *
 816   //  * @param workerNodeKey - The worker node key.
 817   //  * @param worker - The worker.
 818   //  * @param workerUsage - The worker usage.
 819   //  * @param tasksQueue - The worker task queue.
 820   //  */
 821   // private setWorkerNode (
 822   //   workerNodeKey: number,
 823   //   worker: Worker,
 824   //   workerUsage: WorkerUsage,
 825   //   tasksQueue: Queue<Task<Data>>
 826   // ): void {
 827   //   this.workerNodes[workerNodeKey] = {
 828   //     worker,
 829   //     workerUsage,
 830   //     tasksQueue
 831   //   }
 832   // }
 833
 834   /**
 835    * Removes the given worker from the pool worker nodes.
 836    *
 837    * @param worker - The worker.
 838    */
 839   private removeWorkerNode (worker: Worker): void {
 840     const workerNodeKey = this.getWorkerNodeKey(worker)
 841     if (workerNodeKey !== -1) {
 842       this.workerNodes.splice(workerNodeKey, 1)
 843       this.workerChoiceStrategyContext.remove(workerNodeKey)
 844     }
 845   }
 846
 847   private executeTask (workerNodeKey: number, task: Task<Data>): void {
 848     this.beforeTaskExecutionHook(workerNodeKey, task)
 849     this.sendToWorker(this.workerNodes[workerNodeKey].worker, task)
 850   }
 851
 852   private enqueueTask (workerNodeKey: number, task: Task<Data>): number {
 853     return this.workerNodes[workerNodeKey].tasksQueue.enqueue(task)
 854   }
 855
 856   private dequeueTask (workerNodeKey: number): Task<Data> | undefined {
 857     return this.workerNodes[workerNodeKey].tasksQueue.dequeue()
 858   }
 859
 860   private tasksQueueSize (workerNodeKey: number): number {
 861     return this.workerNodes[workerNodeKey].tasksQueue.size
 862   }
 863
 864   private flushTasksQueue (workerNodeKey: number): void {
 865     if (this.tasksQueueSize(workerNodeKey) > 0) {
 866       for (let i = 0; i < this.tasksQueueSize(workerNodeKey); i++) {
 867         this.executeTask(
 868           workerNodeKey,
 869           this.dequeueTask(workerNodeKey) as Task<Data>
 870         )
 871       }
 872     }
 873   }
 874
 875   private flushTasksQueues (): void {
 876     for (const [workerNodeKey] of this.workerNodes.entries()) {
 877       this.flushTasksQueue(workerNodeKey)
 878     }
 879   }
 880
 881   private setWorkerStatistics (worker: Worker): void {
 882     this.sendToWorker(worker, {
 883       statistics: {
 884         runTime:
 885           this.workerChoiceStrategyContext.getTaskStatisticsRequirements()
 886             .runTime.aggregate,
 887         elu: this.workerChoiceStrategyContext.getTaskStatisticsRequirements()
 888           .elu.aggregate
 889       }
 890     })
 891   }
 892
 893   private getWorkerUsage (worker: Worker): WorkerUsage {
 894     return {
 895       tasks: this.getTaskStatistics(worker),
 896       runTime: {
 897         aggregate: 0,
 898         average: 0,
 899         median: 0,
 900         history: new CircularArray()
 901       },
 902       waitTime: {
 903         aggregate: 0,
 904         average: 0,
 905         median: 0,
 906         history: new CircularArray()
 907       },
 908       elu: {
 909         idle: {
 910           aggregate: 0,
 911           average: 0,
 912           median: 0,
 913           history: new CircularArray()
 914         },
 915         active: {
 916           aggregate: 0,
 917           average: 0,
 918           median: 0,
 919           history: new CircularArray()
 920         },
 921         utilization: 0
 922       }
 923     }
 924   }
 925
 926   private getTaskStatistics (worker: Worker): TaskStatistics {
 927     const queueSize =
 928       this.workerNodes[this.getWorkerNodeKey(worker)]?.tasksQueue?.size
 929     return {
 930       executed: 0,
 931       executing: 0,
 932       get queued (): number {
 933         return queueSize ?? 0
 934       },
 935       failed: 0
 936     }
 937   }
 938 }