src/pools/abstract-pool.ts

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