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