src/pools/abstract-pool.ts

   1 import crypto from 'node:crypto'
   2 import type { MessageValue, PromiseResponseWrapper } from '../utility-types'
   3 import {
   4   DEFAULT_WORKER_CHOICE_STRATEGY_OPTIONS,
   5   EMPTY_FUNCTION,
   6   INITIAL_TASKS_USAGE,
   7   median
   8 } from '../utils'
   9 import { KillBehaviors, isKillBehavior } from '../worker/worker-options'
  10 import {
  11   PoolEvents,
  12   type IPool,
  13   type PoolOptions,
  14   type TasksQueueOptions,
  15   PoolType
  16 } from './pool'
  17 import { PoolEmitter } from './pool'
  18 import type { IWorker, Task, TasksUsage, WorkerNode } from './worker'
  19 import {
  20   WorkerChoiceStrategies,
  21   type WorkerChoiceStrategy
  22 } from './selection-strategies/selection-strategies-types'
  23 import { WorkerChoiceStrategyContext } from './selection-strategies/worker-choice-strategy-context'
  24
  25 /**
  26  * Base class that implements some shared logic for all poolifier pools.
  27  *
  28  * @typeParam Worker - Type of worker which manages this pool.
  29  * @typeParam Data - Type of data sent to the worker. This can only be serializable data.
  30  * @typeParam Response - Type of execution response. This can only be serializable data.
  31  */
  32 export abstract class AbstractPool<
  33   Worker extends IWorker,
  34   Data = unknown,
  35   Response = unknown
  36 > implements IPool<Worker, Data, Response> {
  37   /** @inheritDoc */
  38   public readonly workerNodes: Array<WorkerNode<Worker, Data>> = []
  39
  40   /** @inheritDoc */
  41   public readonly emitter?: PoolEmitter
  42
  43   /**
  44    * The execution response promise map.
  45    *
  46    * - `key`: The message id of each submitted task.
  47    * - `value`: An object that contains the worker, the execution response promise resolve and reject callbacks.
  48    *
  49    * When we receive a message from the worker, we get a map entry with the promise resolve/reject bound to the message id.
  50    */
  51   protected promiseResponseMap: Map<
  52   string,
  53   PromiseResponseWrapper<Worker, Response>
  54   > = new Map<string, PromiseResponseWrapper<Worker, Response>>()
  55
  56   /**
  57    * Worker choice strategy context referencing a worker choice algorithm implementation.
  58    *
  59    * Default to a round robin algorithm.
  60    */
  61   protected workerChoiceStrategyContext: WorkerChoiceStrategyContext<
  62   Worker,
  63   Data,
  64   Response
  65   >
  66
  67   /**
  68    * Constructs a new poolifier pool.
  69    *
  70    * @param numberOfWorkers - Number of workers that this pool should manage.
  71    * @param filePath - Path to the worker-file.
  72    * @param opts - Options for the pool.
  73    */
  74   public constructor (
  75     public readonly numberOfWorkers: number,
  76     public readonly filePath: string,
  77     public readonly opts: PoolOptions<Worker>
  78   ) {
  79     if (!this.isMain()) {
  80       throw new Error('Cannot start a pool from a worker!')
  81     }
  82     this.checkNumberOfWorkers(this.numberOfWorkers)
  83     this.checkFilePath(this.filePath)
  84     this.checkPoolOptions(this.opts)
  85
  86     this.chooseWorkerNode.bind(this)
  87     this.executeTask.bind(this)
  88     this.enqueueTask.bind(this)
  89     this.checkAndEmitEvents.bind(this)
  90
  91     this.setupHook()
  92
  93     for (let i = 1; i <= this.numberOfWorkers; i++) {
  94       this.createAndSetupWorker()
  95     }
  96
  97     if (this.opts.enableEvents === true) {
  98       this.emitter = new PoolEmitter()
  99     }
 100     this.workerChoiceStrategyContext = new WorkerChoiceStrategyContext<
 101     Worker,
 102     Data,
 103     Response
 104     >(
 105       this,
 106       this.opts.workerChoiceStrategy,
 107       this.opts.workerChoiceStrategyOptions
 108     )
 109   }
 110
 111   private checkFilePath (filePath: string): void {
 112     if (
 113       filePath == null ||
 114       (typeof filePath === 'string' && filePath.trim().length === 0)
 115     ) {
 116       throw new Error('Please specify a file with a worker implementation')
 117     }
 118   }
 119
 120   private checkNumberOfWorkers (numberOfWorkers: number): void {
 121     if (numberOfWorkers == null) {
 122       throw new Error(
 123         'Cannot instantiate a pool without specifying the number of workers'
 124       )
 125     } else if (!Number.isSafeInteger(numberOfWorkers)) {
 126       throw new TypeError(
 127         'Cannot instantiate a pool with a non integer number of workers'
 128       )
 129     } else if (numberOfWorkers < 0) {
 130       throw new RangeError(
 131         'Cannot instantiate a pool with a negative number of workers'
 132       )
 133     } else if (this.type === PoolType.FIXED && numberOfWorkers === 0) {
 134       throw new Error('Cannot instantiate a fixed pool with no worker')
 135     }
 136   }
 137
 138   private checkPoolOptions (opts: PoolOptions<Worker>): void {
 139     this.opts.workerChoiceStrategy =
 140       opts.workerChoiceStrategy ?? WorkerChoiceStrategies.ROUND_ROBIN
 141     this.checkValidWorkerChoiceStrategy(this.opts.workerChoiceStrategy)
 142     this.opts.workerChoiceStrategyOptions =
 143       opts.workerChoiceStrategyOptions ?? DEFAULT_WORKER_CHOICE_STRATEGY_OPTIONS
 144     this.opts.enableEvents = opts.enableEvents ?? true
 145     this.opts.enableTasksQueue = opts.enableTasksQueue ?? false
 146     if (this.opts.enableTasksQueue) {
 147       if ((opts.tasksQueueOptions?.concurrency as number) <= 0) {
 148         throw new Error(
 149           `Invalid worker tasks concurrency '${
 150             (opts.tasksQueueOptions as TasksQueueOptions).concurrency as number
 151           }'`
 152         )
 153       }
 154       this.opts.tasksQueueOptions = {
 155         concurrency: opts.tasksQueueOptions?.concurrency ?? 1
 156       }
 157     }
 158   }
 159
 160   private checkValidWorkerChoiceStrategy (
 161     workerChoiceStrategy: WorkerChoiceStrategy
 162   ): void {
 163     if (!Object.values(WorkerChoiceStrategies).includes(workerChoiceStrategy)) {
 164       throw new Error(
 165         `Invalid worker choice strategy '${workerChoiceStrategy}'`
 166       )
 167     }
 168   }
 169
 170   /** @inheritDoc */
 171   public abstract get type (): PoolType
 172
 173   /**
 174    * Number of tasks running in the pool.
 175    */
 176   private get numberOfRunningTasks (): number {
 177     return this.workerNodes.reduce(
 178       (accumulator, workerNode) => accumulator + workerNode.tasksUsage.running,
 179       0
 180     )
 181   }
 182
 183   /**
 184    * Number of tasks queued in the pool.
 185    */
 186   private get numberOfQueuedTasks (): number {
 187     if (this.opts.enableTasksQueue === false) {
 188       return 0
 189     }
 190     return this.workerNodes.reduce(
 191       (accumulator, workerNode) => accumulator + workerNode.tasksQueue.length,
 192       0
 193     )
 194   }
 195
 196   /**
 197    * Gets the given worker its worker node key.
 198    *
 199    * @param worker - The worker.
 200    * @returns The worker node key if the worker is found in the pool worker nodes, `-1` otherwise.
 201    */
 202   private getWorkerNodeKey (worker: Worker): number {
 203     return this.workerNodes.findIndex(
 204       workerNode => workerNode.worker === worker
 205     )
 206   }
 207
 208   /** @inheritDoc */
 209   public setWorkerChoiceStrategy (
 210     workerChoiceStrategy: WorkerChoiceStrategy
 211   ): void {
 212     this.checkValidWorkerChoiceStrategy(workerChoiceStrategy)
 213     this.opts.workerChoiceStrategy = workerChoiceStrategy
 214     for (const workerNode of this.workerNodes) {
 215       this.setWorkerNodeTasksUsage(workerNode, INITIAL_TASKS_USAGE)
 216     }
 217     this.workerChoiceStrategyContext.setWorkerChoiceStrategy(
 218       workerChoiceStrategy
 219     )
 220   }
 221
 222   /**
 223    * Whether the pool is full or not.
 224    *
 225    * The pool filling boolean status.
 226    */
 227   protected abstract get full (): boolean
 228
 229   /**
 230    * Whether the pool is busy or not.
 231    *
 232    * The pool busyness boolean status.
 233    */
 234   protected abstract get busy (): boolean
 235
 236   protected internalBusy (): boolean {
 237     return this.findFreeWorkerNodeKey() === -1
 238   }
 239
 240   /** @inheritDoc */
 241   public findFreeWorkerNodeKey (): number {
 242     return this.workerNodes.findIndex(workerNode => {
 243       return workerNode.tasksUsage?.running === 0
 244     })
 245   }
 246
 247   /** @inheritDoc */
 248   public async execute (data: Data): Promise<Response> {
 249     const [workerNodeKey, workerNode] = this.chooseWorkerNode()
 250     const submittedTask: Task<Data> = {
 251       // eslint-disable-next-line @typescript-eslint/consistent-type-assertions
 252       data: data ?? ({} as Data),
 253       id: crypto.randomUUID()
 254     }
 255     const res = new Promise<Response>((resolve, reject) => {
 256       this.promiseResponseMap.set(submittedTask.id as string, {
 257         resolve,
 258         reject,
 259         worker: workerNode.worker
 260       })
 261     })
 262     if (
 263       this.opts.enableTasksQueue === true &&
 264       (this.busy ||
 265         this.workerNodes[workerNodeKey].tasksUsage.running >=
 266           ((this.opts.tasksQueueOptions as TasksQueueOptions)
 267             .concurrency as number))
 268     ) {
 269       this.enqueueTask(workerNodeKey, submittedTask)
 270     } else {
 271       this.executeTask(workerNodeKey, submittedTask)
 272     }
 273     this.checkAndEmitEvents()
 274     // eslint-disable-next-line @typescript-eslint/return-await
 275     return res
 276   }
 277
 278   /** @inheritDoc */
 279   public async destroy (): Promise<void> {
 280     await Promise.all(
 281       this.workerNodes.map(async (workerNode, workerNodeKey) => {
 282         this.flushTasksQueue(workerNodeKey)
 283         await this.destroyWorker(workerNode.worker)
 284       })
 285     )
 286   }
 287
 288   /**
 289    * Shutdowns the given worker.
 290    *
 291    * @param worker - A worker within `workerNodes`.
 292    */
 293   protected abstract destroyWorker (worker: Worker): void | Promise<void>
 294
 295   /**
 296    * Setup hook to execute code before worker node are created in the abstract constructor.
 297    * Can be overridden
 298    *
 299    * @virtual
 300    */
 301   protected setupHook (): void {
 302     // Intentionally empty
 303   }
 304
 305   /**
 306    * Should return whether the worker is the main worker or not.
 307    */
 308   protected abstract isMain (): boolean
 309
 310   /**
 311    * Hook executed before the worker task execution.
 312    * Can be overridden.
 313    *
 314    * @param workerNodeKey - The worker node key.
 315    */
 316   protected beforeTaskExecutionHook (workerNodeKey: number): void {
 317     ++this.workerNodes[workerNodeKey].tasksUsage.running
 318   }
 319
 320   /**
 321    * Hook executed after the worker task execution.
 322    * Can be overridden.
 323    *
 324    * @param worker - The worker.
 325    * @param message - The received message.
 326    */
 327   protected afterTaskExecutionHook (
 328     worker: Worker,
 329     message: MessageValue<Response>
 330   ): void {
 331     const workerTasksUsage = this.getWorkerTasksUsage(worker) as TasksUsage
 332     --workerTasksUsage.running
 333     ++workerTasksUsage.run
 334     if (message.error != null) {
 335       ++workerTasksUsage.error
 336     }
 337     if (this.workerChoiceStrategyContext.getRequiredStatistics().runTime) {
 338       workerTasksUsage.runTime += message.runTime ?? 0
 339       if (
 340         this.workerChoiceStrategyContext.getRequiredStatistics().avgRunTime &&
 341         workerTasksUsage.run !== 0
 342       ) {
 343         workerTasksUsage.avgRunTime =
 344           workerTasksUsage.runTime / workerTasksUsage.run
 345       }
 346       if (this.workerChoiceStrategyContext.getRequiredStatistics().medRunTime) {
 347         workerTasksUsage.runTimeHistory.push(message.runTime ?? 0)
 348         workerTasksUsage.medRunTime = median(workerTasksUsage.runTimeHistory)
 349       }
 350     }
 351   }
 352
 353   /**
 354    * Chooses a worker node for the next task.
 355    *
 356    * The default uses a round robin algorithm to distribute the load.
 357    *
 358    * @returns [worker node key, worker node].
 359    */
 360   protected chooseWorkerNode (): [number, WorkerNode<Worker, Data>] {
 361     let workerNodeKey: number
 362     if (this.type === PoolType.DYNAMIC && !this.full && this.internalBusy()) {
 363       const workerCreated = this.createAndSetupWorker()
 364       this.registerWorkerMessageListener(workerCreated, message => {
 365         if (
 366           isKillBehavior(KillBehaviors.HARD, message.kill) ||
 367           (message.kill != null &&
 368             this.getWorkerTasksUsage(workerCreated)?.running === 0)
 369         ) {
 370           // Kill message received from the worker: no new tasks are submitted to that worker for a while ( > maxInactiveTime)
 371           this.flushTasksQueueByWorker(workerCreated)
 372           void this.destroyWorker(workerCreated)
 373         }
 374       })
 375       workerNodeKey = this.getWorkerNodeKey(workerCreated)
 376     } else {
 377       workerNodeKey = this.workerChoiceStrategyContext.execute()
 378     }
 379     return [workerNodeKey, this.workerNodes[workerNodeKey]]
 380   }
 381
 382   /**
 383    * Sends a message to the given worker.
 384    *
 385    * @param worker - The worker which should receive the message.
 386    * @param message - The message.
 387    */
 388   protected abstract sendToWorker (
 389     worker: Worker,
 390     message: MessageValue<Data>
 391   ): void
 392
 393   /**
 394    * Registers a listener callback on the given worker.
 395    *
 396    * @param worker - The worker which should register a listener.
 397    * @param listener - The message listener callback.
 398    */
 399   protected abstract registerWorkerMessageListener<
 400     Message extends Data | Response
 401   >(worker: Worker, listener: (message: MessageValue<Message>) => void): void
 402
 403   /**
 404    * Returns a newly created worker.
 405    */
 406   protected abstract createWorker (): Worker
 407
 408   /**
 409    * Function that can be hooked up when a worker has been newly created and moved to the pool worker nodes.
 410    *
 411    * Can be used to update the `maxListeners` or binding the `main-worker`\<-\>`worker` connection if not bind by default.
 412    *
 413    * @param worker - The newly created worker.
 414    */
 415   protected abstract afterWorkerSetup (worker: Worker): void
 416
 417   /**
 418    * Creates a new worker and sets it up completely in the pool worker nodes.
 419    *
 420    * @returns New, completely set up worker.
 421    */
 422   protected createAndSetupWorker (): Worker {
 423     const worker = this.createWorker()
 424
 425     worker.on('message', this.opts.messageHandler ?? EMPTY_FUNCTION)
 426     worker.on('error', this.opts.errorHandler ?? EMPTY_FUNCTION)
 427     worker.on('online', this.opts.onlineHandler ?? EMPTY_FUNCTION)
 428     worker.on('exit', this.opts.exitHandler ?? EMPTY_FUNCTION)
 429     worker.once('exit', () => {
 430       this.removeWorkerNode(worker)
 431     })
 432
 433     this.pushWorkerNode(worker)
 434
 435     this.afterWorkerSetup(worker)
 436
 437     return worker
 438   }
 439
 440   /**
 441    * This function is the listener registered for each worker message.
 442    *
 443    * @returns The listener function to execute when a message is received from a worker.
 444    */
 445   protected workerListener (): (message: MessageValue<Response>) => void {
 446     return message => {
 447       if (message.id != null) {
 448         // Task execution response received
 449         const promiseResponse = this.promiseResponseMap.get(message.id)
 450         if (promiseResponse != null) {
 451           if (message.error != null) {
 452             promiseResponse.reject(message.error)
 453           } else {
 454             promiseResponse.resolve(message.data as Response)
 455           }
 456           this.afterTaskExecutionHook(promiseResponse.worker, message)
 457           this.promiseResponseMap.delete(message.id)
 458           const workerNodeKey = this.getWorkerNodeKey(promiseResponse.worker)
 459           if (
 460             this.opts.enableTasksQueue === true &&
 461             this.tasksQueueSize(workerNodeKey) > 0
 462           ) {
 463             this.executeTask(
 464               workerNodeKey,
 465               this.dequeueTask(workerNodeKey) as Task<Data>
 466             )
 467           }
 468         }
 469       }
 470     }
 471   }
 472
 473   private checkAndEmitEvents (): void {
 474     if (this.opts.enableEvents === true) {
 475       if (this.busy) {
 476         this.emitter?.emit(PoolEvents.busy)
 477       }
 478       if (this.type === PoolType.DYNAMIC && this.full) {
 479         this.emitter?.emit(PoolEvents.full)
 480       }
 481     }
 482   }
 483
 484   /**
 485    * Sets the given worker node its tasks usage in the pool.
 486    *
 487    * @param workerNode - The worker node.
 488    * @param tasksUsage - The worker node tasks usage.
 489    */
 490   private setWorkerNodeTasksUsage (
 491     workerNode: WorkerNode<Worker, Data>,
 492     tasksUsage: TasksUsage
 493   ): void {
 494     workerNode.tasksUsage = tasksUsage
 495   }
 496
 497   /**
 498    * Gets the given worker its tasks usage in the pool.
 499    *
 500    * @param worker - The worker.
 501    * @throws {@link Error} if the worker is not found in the pool worker nodes.
 502    * @returns The worker tasks usage.
 503    */
 504   private getWorkerTasksUsage (worker: Worker): TasksUsage | undefined {
 505     const workerNodeKey = this.getWorkerNodeKey(worker)
 506     if (workerNodeKey !== -1) {
 507       return this.workerNodes[workerNodeKey].tasksUsage
 508     }
 509     throw new Error('Worker could not be found in the pool worker nodes')
 510   }
 511
 512   /**
 513    * Pushes the given worker in the pool worker nodes.
 514    *
 515    * @param worker - The worker.
 516    * @returns The worker nodes length.
 517    */
 518   private pushWorkerNode (worker: Worker): number {
 519     return this.workerNodes.push({
 520       worker,
 521       tasksUsage: INITIAL_TASKS_USAGE,
 522       tasksQueue: []
 523     })
 524   }
 525
 526   /**
 527    * Sets the given worker in the pool worker nodes.
 528    *
 529    * @param workerNodeKey - The worker node key.
 530    * @param worker - The worker.
 531    * @param tasksUsage - The worker tasks usage.
 532    * @param tasksQueue - The worker task queue.
 533    */
 534   private setWorkerNode (
 535     workerNodeKey: number,
 536     worker: Worker,
 537     tasksUsage: TasksUsage,
 538     tasksQueue: Array<Task<Data>>
 539   ): void {
 540     this.workerNodes[workerNodeKey] = {
 541       worker,
 542       tasksUsage,
 543       tasksQueue
 544     }
 545   }
 546
 547   /**
 548    * Removes the given worker from the pool worker nodes.
 549    *
 550    * @param worker - The worker.
 551    */
 552   private removeWorkerNode (worker: Worker): void {
 553     const workerNodeKey = this.getWorkerNodeKey(worker)
 554     this.workerNodes.splice(workerNodeKey, 1)
 555     this.workerChoiceStrategyContext.remove(workerNodeKey)
 556   }
 557
 558   private executeTask (workerNodeKey: number, task: Task<Data>): void {
 559     this.beforeTaskExecutionHook(workerNodeKey)
 560     this.sendToWorker(this.workerNodes[workerNodeKey].worker, task)
 561   }
 562
 563   private enqueueTask (workerNodeKey: number, task: Task<Data>): number {
 564     return this.workerNodes[workerNodeKey].tasksQueue.push(task)
 565   }
 566
 567   private dequeueTask (workerNodeKey: number): Task<Data> | undefined {
 568     return this.workerNodes[workerNodeKey].tasksQueue.shift()
 569   }
 570
 571   private tasksQueueSize (workerNodeKey: number): number {
 572     return this.workerNodes[workerNodeKey].tasksQueue.length
 573   }
 574
 575   private flushTasksQueue (workerNodeKey: number): void {
 576     if (this.tasksQueueSize(workerNodeKey) > 0) {
 577       for (const task of this.workerNodes[workerNodeKey].tasksQueue) {
 578         this.executeTask(workerNodeKey, task)
 579       }
 580     }
 581   }
 582
 583   private flushTasksQueueByWorker (worker: Worker): void {
 584     const workerNodeKey = this.getWorkerNodeKey(worker)
 585     this.flushTasksQueue(workerNodeKey)
 586   }
 587 }