src/pools/abstract-pool.ts

   1 import type {
   2   MessageValue,
   3   PromiseWorkerResponseWrapper
   4 } from '../utility-types'
   5 import { EMPTY_FUNCTION } from '../utils'
   6 import { KillBehaviors, isKillBehavior } from '../worker/worker-options'
   7 import type { PoolOptions } from './pool'
   8 import { PoolEmitter } from './pool'
   9 import type { IPoolInternal, TasksUsage } from './pool-internal'
  10 import { PoolType } from './pool-internal'
  11 import type { IPoolWorker } from './pool-worker'
  12 import {
  13   WorkerChoiceStrategies,
  14   type WorkerChoiceStrategy
  15 } from './selection-strategies/selection-strategies-types'
  16 import { WorkerChoiceStrategyContext } from './selection-strategies/worker-choice-strategy-context'
  17
  18 /**
  19  * Base class that implements some shared logic for all poolifier pools.
  20  *
  21  * @typeParam Worker - Type of worker which manages this pool.
  22  * @typeParam Data - Type of data sent to the worker. This can only be serializable data.
  23  * @typeParam Response - Type of response of execution. This can only be serializable data.
  24  */
  25 export abstract class AbstractPool<
  26   Worker extends IPoolWorker,
  27   Data = unknown,
  28   Response = unknown
  29 > implements IPoolInternal<Worker, Data, Response> {
  30   /** {@inheritDoc} */
  31   public readonly workers: Worker[] = []
  32
  33   /** {@inheritDoc} */
  34   public readonly workersTasksUsage: Map<Worker, TasksUsage> = new Map<
  35   Worker,
  36   TasksUsage
  37   >()
  38
  39   /** {@inheritDoc} */
  40   public readonly emitter?: PoolEmitter
  41
  42   /**
  43    * The promise map.
  44    *
  45    * - `key`: This is the message Id of each submitted task.
  46    * - `value`: An object that contains the worker, the resolve function and the reject function.
  47    *
  48    * When we receive a message from the worker we get a map entry and resolve/reject the promise based on the message.
  49    */
  50   protected promiseMap: Map<
  51   number,
  52   PromiseWorkerResponseWrapper<Worker, Response>
  53   > = new Map<number, PromiseWorkerResponseWrapper<Worker, Response>>()
  54
  55   /**
  56    * Id of the next message.
  57    */
  58   protected nextMessageId: number = 0
  59
  60   /**
  61    * Worker choice strategy instance implementing the worker choice algorithm.
  62    *
  63    * Default to a strategy implementing a round robin algorithm.
  64    */
  65   protected workerChoiceStrategyContext: WorkerChoiceStrategyContext<
  66   Worker,
  67   Data,
  68   Response
  69   >
  70
  71   /**
  72    * Constructs a new poolifier pool.
  73    *
  74    * @param numberOfWorkers - Number of workers that this pool should manage.
  75    * @param filePath - Path to the worker-file.
  76    * @param opts - Options for the pool.
  77    */
  78   public constructor (
  79     public readonly numberOfWorkers: number,
  80     public readonly filePath: string,
  81     public readonly opts: PoolOptions<Worker>
  82   ) {
  83     if (!this.isMain()) {
  84       throw new Error('Cannot start a pool from a worker!')
  85     }
  86     this.checkNumberOfWorkers(this.numberOfWorkers)
  87     this.checkFilePath(this.filePath)
  88     this.checkPoolOptions(this.opts)
  89     this.setupHook()
  90
  91     for (let i = 1; i <= this.numberOfWorkers; i++) {
  92       this.createAndSetupWorker()
  93     }
  94
  95     if (this.opts.enableEvents === true) {
  96       this.emitter = new PoolEmitter()
  97     }
  98     this.workerChoiceStrategyContext = new WorkerChoiceStrategyContext(
  99       this,
 100       () => {
 101         const workerCreated = this.createAndSetupWorker()
 102         this.registerWorkerMessageListener(workerCreated, message => {
 103           if (
 104             isKillBehavior(KillBehaviors.HARD, message.kill) ||
 105             this.getWorkerRunningTasks(workerCreated) === 0
 106           ) {
 107             // Kill received from the worker, means that no new tasks are submitted to that worker for a while ( > maxInactiveTime)
 108             void (this.destroyWorker(workerCreated) as Promise<void>)
 109           }
 110         })
 111         return workerCreated
 112       },
 113       this.opts.workerChoiceStrategy
 114     )
 115   }
 116
 117   private checkFilePath (filePath: string): void {
 118     if (filePath == null || filePath.length === 0) {
 119       throw new Error('Please specify a file with a worker implementation')
 120     }
 121   }
 122
 123   private checkNumberOfWorkers (numberOfWorkers: number): void {
 124     if (numberOfWorkers == null) {
 125       throw new Error(
 126         'Cannot instantiate a pool without specifying the number of workers'
 127       )
 128     } else if (!Number.isSafeInteger(numberOfWorkers)) {
 129       throw new TypeError(
 130         'Cannot instantiate a pool with a non integer number of workers'
 131       )
 132     } else if (numberOfWorkers < 0) {
 133       throw new RangeError(
 134         'Cannot instantiate a pool with a negative number of workers'
 135       )
 136     } else if (this.type === PoolType.FIXED && numberOfWorkers === 0) {
 137       throw new Error('Cannot instantiate a fixed pool with no worker')
 138     }
 139   }
 140
 141   private checkPoolOptions (opts: PoolOptions<Worker>): void {
 142     this.opts.workerChoiceStrategy =
 143       opts.workerChoiceStrategy ?? WorkerChoiceStrategies.ROUND_ROBIN
 144     this.opts.enableEvents = opts.enableEvents ?? true
 145   }
 146
 147   /** {@inheritDoc} */
 148   public abstract get type (): PoolType
 149
 150   /** {@inheritDoc} */
 151   public get numberOfRunningTasks (): number {
 152     return this.promiseMap.size
 153   }
 154
 155   /** {@inheritDoc} */
 156   public getWorkerIndex (worker: Worker): number {
 157     return this.workers.indexOf(worker)
 158   }
 159
 160   /** {@inheritDoc} */
 161   public getWorkerRunningTasks (worker: Worker): number | undefined {
 162     return this.workersTasksUsage.get(worker)?.running
 163   }
 164
 165   /** {@inheritDoc} */
 166   public getWorkerAverageTasksRunTime (worker: Worker): number | undefined {
 167     return this.workersTasksUsage.get(worker)?.avgRunTime
 168   }
 169
 170   /** {@inheritDoc} */
 171   public setWorkerChoiceStrategy (
 172     workerChoiceStrategy: WorkerChoiceStrategy
 173   ): void {
 174     this.opts.workerChoiceStrategy = workerChoiceStrategy
 175     for (const worker of this.workers) {
 176       this.resetWorkerTasksUsage(worker)
 177     }
 178     this.workerChoiceStrategyContext.setWorkerChoiceStrategy(
 179       workerChoiceStrategy
 180     )
 181   }
 182
 183   /** {@inheritDoc} */
 184   public abstract get busy (): boolean
 185
 186   protected internalGetBusyStatus (): boolean {
 187     return (
 188       this.numberOfRunningTasks >= this.numberOfWorkers &&
 189       this.findFreeWorker() === false
 190     )
 191   }
 192
 193   /** {@inheritDoc} */
 194   public findFreeWorker (): Worker | false {
 195     for (const worker of this.workers) {
 196       if (this.getWorkerRunningTasks(worker) === 0) {
 197         // A worker is free, return the matching worker
 198         return worker
 199       }
 200     }
 201     return false
 202   }
 203
 204   /** {@inheritDoc} */
 205   public async execute (data: Data): Promise<Response> {
 206     // Configure worker to handle message with the specified task
 207     const worker = this.chooseWorker()
 208     const res = this.internalExecute(worker, this.nextMessageId)
 209     this.checkAndEmitBusy()
 210     this.sendToWorker(worker, {
 211       // eslint-disable-next-line @typescript-eslint/consistent-type-assertions
 212       data: data ?? ({} as Data),
 213       id: this.nextMessageId
 214     })
 215     ++this.nextMessageId
 216     // eslint-disable-next-line @typescript-eslint/return-await
 217     return res
 218   }
 219
 220   /** {@inheritDoc} */
 221   public async destroy (): Promise<void> {
 222     await Promise.all(
 223       this.workers.map(async worker => {
 224         await this.destroyWorker(worker)
 225       })
 226     )
 227   }
 228
 229   /**
 230    * Shutdowns given worker.
 231    *
 232    * @param worker - A worker within `workers`.
 233    */
 234   protected abstract destroyWorker (worker: Worker): void | Promise<void>
 235
 236   /**
 237    * Setup hook that can be overridden by a Poolifier pool implementation
 238    * to run code before workers are created in the abstract constructor.
 239    */
 240   protected setupHook (): void {
 241     // Can be overridden
 242   }
 243
 244   /**
 245    * Should return whether the worker is the main worker or not.
 246    */
 247   protected abstract isMain (): boolean
 248
 249   /**
 250    * Hook executed before the worker task promise resolution.
 251    * Can be overridden.
 252    *
 253    * @param worker - The worker.
 254    */
 255   protected beforePromiseWorkerResponseHook (worker: Worker): void {
 256     this.increaseWorkerRunningTasks(worker)
 257   }
 258
 259   /**
 260    * Hook executed after the worker task promise resolution.
 261    * Can be overridden.
 262    *
 263    * @param message - The received message.
 264    * @param promise - The Promise response.
 265    */
 266   protected afterPromiseWorkerResponseHook (
 267     message: MessageValue<Response>,
 268     promise: PromiseWorkerResponseWrapper<Worker, Response>
 269   ): void {
 270     this.decreaseWorkerRunningTasks(promise.worker)
 271     this.stepWorkerRunTasks(promise.worker, 1)
 272     this.updateWorkerTasksRunTime(promise.worker, message.taskRunTime)
 273   }
 274
 275   /**
 276    * Removes the given worker from the pool.
 277    *
 278    * @param worker - The worker that will be removed.
 279    */
 280   protected removeWorker (worker: Worker): void {
 281     // Clean worker from data structure
 282     this.workers.splice(this.getWorkerIndex(worker), 1)
 283     this.removeWorkerTasksUsage(worker)
 284   }
 285
 286   /**
 287    * Chooses a worker for the next task.
 288    *
 289    * The default implementation uses a round robin algorithm to distribute the load.
 290    *
 291    * @returns Worker.
 292    */
 293   protected chooseWorker (): Worker {
 294     return this.workerChoiceStrategyContext.execute()
 295   }
 296
 297   /**
 298    * Sends a message to the given worker.
 299    *
 300    * @param worker - The worker which should receive the message.
 301    * @param message - The message.
 302    */
 303   protected abstract sendToWorker (
 304     worker: Worker,
 305     message: MessageValue<Data>
 306   ): void
 307
 308   /**
 309    * Registers a listener callback on a given worker.
 310    *
 311    * @param worker - The worker which should register a listener.
 312    * @param listener - The message listener callback.
 313    */
 314   protected abstract registerWorkerMessageListener<
 315     Message extends Data | Response
 316   >(worker: Worker, listener: (message: MessageValue<Message>) => void): void
 317
 318   /**
 319    * Returns a newly created worker.
 320    */
 321   protected abstract createWorker (): Worker
 322
 323   /**
 324    * Function that can be hooked up when a worker has been newly created and moved to the workers registry.
 325    *
 326    * Can be used to update the `maxListeners` or binding the `main-worker`\<-\>`worker` connection if not bind by default.
 327    *
 328    * @param worker - The newly created worker.
 329    */
 330   protected abstract afterWorkerSetup (worker: Worker): void
 331
 332   /**
 333    * Creates a new worker for this pool and sets it up completely.
 334    *
 335    * @returns New, completely set up worker.
 336    */
 337   protected createAndSetupWorker (): Worker {
 338     const worker = this.createWorker()
 339
 340     worker.on('message', this.opts.messageHandler ?? EMPTY_FUNCTION)
 341     worker.on('error', this.opts.errorHandler ?? EMPTY_FUNCTION)
 342     worker.on('online', this.opts.onlineHandler ?? EMPTY_FUNCTION)
 343     worker.on('exit', this.opts.exitHandler ?? EMPTY_FUNCTION)
 344     worker.once('exit', () => {
 345       this.removeWorker(worker)
 346     })
 347
 348     this.workers.push(worker)
 349
 350     // Init worker tasks usage map
 351     this.initWorkerTasksUsage(worker)
 352
 353     this.afterWorkerSetup(worker)
 354
 355     return worker
 356   }
 357
 358   /**
 359    * This function is the listener registered for each worker.
 360    *
 361    * @returns The listener function to execute when a message is received from a worker.
 362    */
 363   protected workerListener (): (message: MessageValue<Response>) => void {
 364     return message => {
 365       if (message.id !== undefined) {
 366         const promise = this.promiseMap.get(message.id)
 367         if (promise !== undefined) {
 368           if (message.error != null) {
 369             promise.reject(message.error)
 370           } else {
 371             promise.resolve(message.data as Response)
 372           }
 373           this.afterPromiseWorkerResponseHook(message, promise)
 374           this.promiseMap.delete(message.id)
 375         }
 376       }
 377     }
 378   }
 379
 380   private async internalExecute (
 381     worker: Worker,
 382     messageId: number
 383   ): Promise<Response> {
 384     this.beforePromiseWorkerResponseHook(worker)
 385     return await new Promise<Response>((resolve, reject) => {
 386       this.promiseMap.set(messageId, { resolve, reject, worker })
 387     })
 388   }
 389
 390   private checkAndEmitBusy (): void {
 391     if (this.opts.enableEvents === true && this.busy) {
 392       this.emitter?.emit('busy')
 393     }
 394   }
 395
 396   /**
 397    * Increases the number of tasks that the given worker has applied.
 398    *
 399    * @param worker - Worker which running tasks is increased.
 400    */
 401   private increaseWorkerRunningTasks (worker: Worker): void {
 402     this.stepWorkerRunningTasks(worker, 1)
 403   }
 404
 405   /**
 406    * Decreases the number of tasks that the given worker has applied.
 407    *
 408    * @param worker - Worker which running tasks is decreased.
 409    */
 410   private decreaseWorkerRunningTasks (worker: Worker): void {
 411     this.stepWorkerRunningTasks(worker, -1)
 412   }
 413
 414   /**
 415    * Steps the number of tasks that the given worker has applied.
 416    *
 417    * @param worker - Worker which running tasks are stepped.
 418    * @param step - Number of running tasks step.
 419    */
 420   private stepWorkerRunningTasks (worker: Worker, step: number): void {
 421     if (this.checkWorkerTasksUsage(worker)) {
 422       const tasksUsage = this.workersTasksUsage.get(worker) as TasksUsage
 423       tasksUsage.running = tasksUsage.running + step
 424       this.workersTasksUsage.set(worker, tasksUsage)
 425     }
 426   }
 427
 428   /**
 429    * Steps the number of tasks that the given worker has run.
 430    *
 431    * @param worker - Worker which has run tasks.
 432    * @param step - Number of run tasks step.
 433    */
 434   private stepWorkerRunTasks (worker: Worker, step: number): void {
 435     if (this.checkWorkerTasksUsage(worker)) {
 436       const tasksUsage = this.workersTasksUsage.get(worker) as TasksUsage
 437       tasksUsage.run = tasksUsage.run + step
 438       this.workersTasksUsage.set(worker, tasksUsage)
 439     }
 440   }
 441
 442   /**
 443    * Updates tasks runtime for the given worker.
 444    *
 445    * @param worker - Worker which run the task.
 446    * @param taskRunTime - Worker task runtime.
 447    */
 448   private updateWorkerTasksRunTime (
 449     worker: Worker,
 450     taskRunTime: number | undefined
 451   ): void {
 452     if (
 453       this.workerChoiceStrategyContext.getWorkerChoiceStrategy()
 454         .requiredStatistics.runTime &&
 455       this.checkWorkerTasksUsage(worker)
 456     ) {
 457       const tasksUsage = this.workersTasksUsage.get(worker) as TasksUsage
 458       tasksUsage.runTime += taskRunTime ?? 0
 459       if (tasksUsage.run !== 0) {
 460         tasksUsage.avgRunTime = tasksUsage.runTime / tasksUsage.run
 461       }
 462       this.workersTasksUsage.set(worker, tasksUsage)
 463     }
 464   }
 465
 466   /**
 467    * Checks if the given worker is registered in the workers tasks usage map.
 468    *
 469    * @param worker - Worker to check.
 470    * @returns `true` if the worker is registered in the workers tasks usage map. `false` otherwise.
 471    */
 472   private checkWorkerTasksUsage (worker: Worker): boolean {
 473     const hasTasksUsage = this.workersTasksUsage.has(worker)
 474     if (!hasTasksUsage) {
 475       throw new Error('Worker could not be found in workers tasks usage map')
 476     }
 477     return hasTasksUsage
 478   }
 479
 480   /**
 481    * Initializes tasks usage statistics.
 482    *
 483    * @param worker - The worker.
 484    */
 485   private initWorkerTasksUsage (worker: Worker): void {
 486     this.workersTasksUsage.set(worker, {
 487       run: 0,
 488       running: 0,
 489       runTime: 0,
 490       avgRunTime: 0
 491     })
 492   }
 493
 494   /**
 495    * Removes worker tasks usage statistics.
 496    *
 497    * @param worker - The worker.
 498    */
 499   private removeWorkerTasksUsage (worker: Worker): void {
 500     this.workersTasksUsage.delete(worker)
 501   }
 502
 503   /**
 504    * Resets worker tasks usage statistics.
 505    *
 506    * @param worker - The worker.
 507    */
 508   private resetWorkerTasksUsage (worker: Worker): void {
 509     this.removeWorkerTasksUsage(worker)
 510     this.initWorkerTasksUsage(worker)
 511   }
 512 }