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     if (this.shallCreateDynamicWorker()) {
 618       return this.getWorkerNodeKey(this.createAndSetupDynamicWorker())
 619     }
 620     return this.workerChoiceStrategyContext.execute()
 621   }
 622
 623   protected shallCreateDynamicWorker (): boolean {
 624     return this.type === PoolTypes.dynamic && !this.full && this.internalBusy()
 625   }
 626
 627   /**
 628    * Sends a message to the given worker.
 629    *
 630    * @param worker - The worker which should receive the message.
 631    * @param message - The message.
 632    */
 633   protected abstract sendToWorker (
 634     worker: Worker,
 635     message: MessageValue<Data>
 636   ): void
 637
 638   /**
 639    * Registers a listener callback on the given worker.
 640    *
 641    * @param worker - The worker which should register a listener.
 642    * @param listener - The message listener callback.
 643    */
 644   protected abstract registerWorkerMessageListener<
 645     Message extends Data | Response
 646   >(worker: Worker, listener: (message: MessageValue<Message>) => void): void
 647
 648   /**
 649    * Returns a newly created worker.
 650    */
 651   protected abstract createWorker (): Worker
 652
 653   /**
 654    * Function that can be hooked up when a worker has been newly created and moved to the pool worker nodes.
 655    *
 656    * Can be used to update the `maxListeners` or binding the `main-worker`\<-\>`worker` connection if not bind by default.
 657    *
 658    * @param worker - The newly created worker.
 659    */
 660   protected abstract afterWorkerSetup (worker: Worker): void
 661
 662   /**
 663    * Creates a new worker and sets it up completely in the pool worker nodes.
 664    *
 665    * @returns New, completely set up worker.
 666    */
 667   protected createAndSetupWorker (): Worker {
 668     const worker = this.createWorker()
 669
 670     worker.on('message', this.opts.messageHandler ?? EMPTY_FUNCTION)
 671     worker.on('error', this.opts.errorHandler ?? EMPTY_FUNCTION)
 672     worker.on('error', error => {
 673       if (this.emitter != null) {
 674         this.emitter.emit(PoolEvents.error, error)
 675       }
 676       if (this.opts.restartWorkerOnError === true) {
 677         this.createAndSetupWorker()
 678       }
 679     })
 680     worker.on('online', this.opts.onlineHandler ?? EMPTY_FUNCTION)
 681     worker.on('exit', this.opts.exitHandler ?? EMPTY_FUNCTION)
 682     worker.once('exit', () => {
 683       this.removeWorkerNode(worker)
 684     })
 685
 686     this.pushWorkerNode(worker)
 687
 688     this.setWorkerStatistics(worker)
 689
 690     this.afterWorkerSetup(worker)
 691
 692     return worker
 693   }
 694
 695   /**
 696    * Creates a new dynamic worker and sets it up completely in the pool worker nodes.
 697    *
 698    * @returns New, completely set up dynamic worker.
 699    */
 700   protected createAndSetupDynamicWorker (): Worker {
 701     const worker = this.createAndSetupWorker()
 702     this.registerWorkerMessageListener(worker, message => {
 703       const currentWorkerNodeKey = this.getWorkerNodeKey(worker)
 704       if (
 705         isKillBehavior(KillBehaviors.HARD, message.kill) ||
 706         (message.kill != null &&
 707           this.workerNodes[currentWorkerNodeKey].workerUsage.tasks.executing ===
 708             0)
 709       ) {
 710         // Kill message received from the worker: no new tasks are submitted to that worker for a while ( > maxInactiveTime)
 711         this.flushTasksQueue(currentWorkerNodeKey)
 712         // FIXME: wait for tasks to be finished
 713         void (this.destroyWorker(worker) as Promise<void>)
 714       }
 715     })
 716     return worker
 717   }
 718
 719   /**
 720    * This function is the listener registered for each worker message.
 721    *
 722    * @returns The listener function to execute when a message is received from a worker.
 723    */
 724   protected workerListener (): (message: MessageValue<Response>) => void {
 725     return message => {
 726       if (message.id != null) {
 727         // Task execution response received
 728         const promiseResponse = this.promiseResponseMap.get(message.id)
 729         if (promiseResponse != null) {
 730           if (message.taskError != null) {
 731             promiseResponse.reject(message.taskError.message)
 732             if (this.emitter != null) {
 733               this.emitter.emit(PoolEvents.taskError, message.taskError)
 734             }
 735           } else {
 736             promiseResponse.resolve(message.data as Response)
 737           }
 738           this.afterTaskExecutionHook(promiseResponse.worker, message)
 739           this.promiseResponseMap.delete(message.id)
 740           const workerNodeKey = this.getWorkerNodeKey(promiseResponse.worker)
 741           if (
 742             this.opts.enableTasksQueue === true &&
 743             this.tasksQueueSize(workerNodeKey) > 0
 744           ) {
 745             this.executeTask(
 746               workerNodeKey,
 747               this.dequeueTask(workerNodeKey) as Task<Data>
 748             )
 749           }
 750         }
 751       }
 752     }
 753   }
 754
 755   private checkAndEmitEvents (): void {
 756     if (this.emitter != null) {
 757       if (this.busy) {
 758         this.emitter?.emit(PoolEvents.busy, this.info)
 759       }
 760       if (this.type === PoolTypes.dynamic && this.full) {
 761         this.emitter?.emit(PoolEvents.full, this.info)
 762       }
 763     }
 764   }
 765
 766   /**
 767    * Sets the given worker node its tasks usage in the pool.
 768    *
 769    * @param workerNode - The worker node.
 770    * @param workerUsage - The worker usage.
 771    */
 772   private setWorkerNodeTasksUsage (
 773     workerNode: WorkerNode<Worker, Data>,
 774     workerUsage: WorkerUsage
 775   ): void {
 776     workerNode.workerUsage = workerUsage
 777   }
 778
 779   /**
 780    * Pushes the given worker in the pool worker nodes.
 781    *
 782    * @param worker - The worker.
 783    * @returns The worker nodes length.
 784    */
 785   private pushWorkerNode (worker: Worker): number {
 786     return this.workerNodes.push({
 787       worker,
 788       workerUsage: this.getWorkerUsage(worker),
 789       tasksQueue: new Queue<Task<Data>>()
 790     })
 791   }
 792
 793   // /**
 794   //  * Sets the given worker in the pool worker nodes.
 795   //  *
 796   //  * @param workerNodeKey - The worker node key.
 797   //  * @param worker - The worker.
 798   //  * @param workerUsage - The worker usage.
 799   //  * @param tasksQueue - The worker task queue.
 800   //  */
 801   // private setWorkerNode (
 802   //   workerNodeKey: number,
 803   //   worker: Worker,
 804   //   workerUsage: WorkerUsage,
 805   //   tasksQueue: Queue<Task<Data>>
 806   // ): void {
 807   //   this.workerNodes[workerNodeKey] = {
 808   //     worker,
 809   //     workerUsage,
 810   //     tasksQueue
 811   //   }
 812   // }
 813
 814   /**
 815    * Removes the given worker from the pool worker nodes.
 816    *
 817    * @param worker - The worker.
 818    */
 819   private removeWorkerNode (worker: Worker): void {
 820     const workerNodeKey = this.getWorkerNodeKey(worker)
 821     if (workerNodeKey !== -1) {
 822       this.workerNodes.splice(workerNodeKey, 1)
 823       this.workerChoiceStrategyContext.remove(workerNodeKey)
 824     }
 825   }
 826
 827   private executeTask (workerNodeKey: number, task: Task<Data>): void {
 828     this.beforeTaskExecutionHook(workerNodeKey, task)
 829     this.sendToWorker(this.workerNodes[workerNodeKey].worker, task)
 830   }
 831
 832   private enqueueTask (workerNodeKey: number, task: Task<Data>): number {
 833     return this.workerNodes[workerNodeKey].tasksQueue.enqueue(task)
 834   }
 835
 836   private dequeueTask (workerNodeKey: number): Task<Data> | undefined {
 837     return this.workerNodes[workerNodeKey].tasksQueue.dequeue()
 838   }
 839
 840   private tasksQueueSize (workerNodeKey: number): number {
 841     return this.workerNodes[workerNodeKey].tasksQueue.size
 842   }
 843
 844   private flushTasksQueue (workerNodeKey: number): void {
 845     if (this.tasksQueueSize(workerNodeKey) > 0) {
 846       for (let i = 0; i < this.tasksQueueSize(workerNodeKey); i++) {
 847         this.executeTask(
 848           workerNodeKey,
 849           this.dequeueTask(workerNodeKey) as Task<Data>
 850         )
 851       }
 852     }
 853   }
 854
 855   private flushTasksQueues (): void {
 856     for (const [workerNodeKey] of this.workerNodes.entries()) {
 857       this.flushTasksQueue(workerNodeKey)
 858     }
 859   }
 860
 861   private setWorkerStatistics (worker: Worker): void {
 862     this.sendToWorker(worker, {
 863       statistics: {
 864         runTime:
 865           this.workerChoiceStrategyContext.getTaskStatisticsRequirements()
 866             .runTime.aggregate,
 867         elu: this.workerChoiceStrategyContext.getTaskStatisticsRequirements()
 868           .elu.aggregate
 869       }
 870     })
 871   }
 872
 873   private getWorkerUsage (worker: Worker): WorkerUsage {
 874     return {
 875       tasks: this.getTaskStatistics(worker),
 876       runTime: {
 877         aggregate: 0,
 878         average: 0,
 879         median: 0,
 880         history: new CircularArray()
 881       },
 882       waitTime: {
 883         aggregate: 0,
 884         average: 0,
 885         median: 0,
 886         history: new CircularArray()
 887       },
 888       elu: {
 889         idle: {
 890           aggregate: 0,
 891           average: 0,
 892           median: 0,
 893           history: new CircularArray()
 894         },
 895         active: {
 896           aggregate: 0,
 897           average: 0,
 898           median: 0,
 899           history: new CircularArray()
 900         },
 901         utilization: 0
 902       }
 903     }
 904   }
 905
 906   private getTaskStatistics (worker: Worker): TaskStatistics {
 907     const queueSize =
 908       this.workerNodes[this.getWorkerNodeKey(worker)]?.tasksQueue?.size
 909     return {
 910       executed: 0,
 911       executing: 0,
 912       get queued (): number {
 913         return queueSize ?? 0
 914       },
 915       failed: 0
 916     }
 917   }
 918 }