src/pools/abstract-pool.ts

   1 import EventEmitter from 'events'
   2 import type { MessageValue } from '../utility-types'
   3 import type { IPool } from './pool'
   4
   5 /**
   6  * Callback invoked if the worker raised an error.
   7  */
   8 export type ErrorHandler<Worker> = (this: Worker, e: Error) => void
   9
  10 /**
  11  * Callback invoked when the worker has started successfully.
  12  */
  13 export type OnlineHandler<Worker> = (this: Worker) => void
  14
  15 /**
  16  * Callback invoked when the worker exits successfully.
  17  */
  18 export type ExitHandler<Worker> = (this: Worker, code: number) => void
  19
  20 /**
  21  * Basic interface that describes the minimum required implementation of listener events for a pool-worker.
  22  */
  23 export interface IWorker {
  24   /**
  25    * Register a listener to the error event.
  26    *
  27    * @param event `'error'`.
  28    * @param handler The error handler.
  29    */
  30   on(event: 'error', handler: ErrorHandler<this>): void
  31   /**
  32    * Register a listener to the online event.
  33    *
  34    * @param event `'online'`.
  35    * @param handler The online handler.
  36    */
  37   on(event: 'online', handler: OnlineHandler<this>): void
  38   /**
  39    * Register a listener to the exit event.
  40    *
  41    * @param event `'exit'`.
  42    * @param handler The exit handler.
  43    */
  44   on(event: 'exit', handler: ExitHandler<this>): void
  45   /**
  46    * Register a listener to the exit event that will only performed once.
  47    *
  48    * @param event `'exit'`.
  49    * @param handler The exit handler.
  50    */
  51   once(event: 'exit', handler: ExitHandler<this>): void
  52 }
  53
  54 /**
  55  * Options for a poolifier pool.
  56  */
  57 export interface PoolOptions<Worker> {
  58   /**
  59    * A function that will listen for error event on each worker.
  60    */
  61   errorHandler?: ErrorHandler<Worker>
  62   /**
  63    * A function that will listen for online event on each worker.
  64    */
  65   onlineHandler?: OnlineHandler<Worker>
  66   /**
  67    * A function that will listen for exit event on each worker.
  68    */
  69   exitHandler?: ExitHandler<Worker>
  70   /**
  71    * This is just to avoid non-useful warning messages.
  72    *
  73    * Will be used to set `maxListeners` on event emitters (workers are event emitters).
  74    *
  75    * @default 1000
  76    * @see [Node events emitter.setMaxListeners(n)](https://nodejs.org/api/events.html#events_emitter_setmaxlisteners_n)
  77    */
  78   maxTasks?: number
  79 }
  80
  81 /**
  82  * Internal poolifier pool emitter.
  83  */
  84 class PoolEmitter extends EventEmitter {}
  85
  86 /**
  87  * Base class containing some shared logic for all poolifier pools.
  88  *
  89  * @template Worker Type of worker which manages this pool.
  90  * @template Data Type of data sent to the worker.
  91  * @template Response Type of response of execution.
  92  */
  93 export abstract class AbstractPool<
  94   Worker extends IWorker,
  95   Data = unknown,
  96   Response = unknown
  97 > implements IPool<Data, Response> {
  98   /**
  99    * List of currently available workers.
 100    */
 101   public readonly workers: Worker[] = []
 102
 103   /**
 104    * Index for the next worker.
 105    */
 106   public nextWorkerIndex: number = 0
 107
 108   /**
 109    * The tasks map.
 110    *
 111    * - `key`: The `Worker`
 112    * - `value`: Number of tasks currently in progress on the worker.
 113    */
 114   public readonly tasks: Map<Worker, number> = new Map<Worker, number>()
 115
 116   /**
 117    * Emitter on which events can be listened to.
 118    *
 119    * Events that can currently be listened to:
 120    *
 121    * - `'FullPool'`
 122    */
 123   public readonly emitter: PoolEmitter
 124
 125   /**
 126    * ID of the next message.
 127    */
 128   protected nextMessageId: number = 0
 129
 130   /**
 131    * Constructs a new poolifier pool.
 132    *
 133    * @param numberOfWorkers Number of workers that this pool should manage.
 134    * @param filePath Path to the worker-file.
 135    * @param opts Options for the pool. Default: `{ maxTasks: 1000 }`
 136    */
 137   public constructor (
 138     public readonly numberOfWorkers: number,
 139     public readonly filePath: string,
 140     public readonly opts: PoolOptions<Worker> = { maxTasks: 1000 }
 141   ) {
 142     if (!this.isMain()) {
 143       throw new Error('Cannot start a pool from a worker!')
 144     }
 145     // TODO christopher 2021-02-07: Improve this check e.g. with a pattern or blank check
 146     if (!this.filePath) {
 147       throw new Error('Please specify a file with a worker implementation')
 148     }
 149     this.setupHook()
 150
 151     for (let i = 1; i <= this.numberOfWorkers; i++) {
 152       this.createAndSetupWorker()
 153     }
 154
 155     this.emitter = new PoolEmitter()
 156   }
 157
 158   /**
 159    * Number of workers that this pool should manage.
 160    *
 161    * @returns Number of workers that this pool manages.
 162    * @deprecated Only here for backward compatibility.
 163    */
 164   // eslint-disable-next-line spellcheck/spell-checker
 165   public get numWorkers (): number {
 166     return this.numberOfWorkers
 167   }
 168
 169   /**
 170    * Index for the next worker.
 171    *
 172    * @returns Index for the next worker.
 173    * @deprecated Only here for backward compatibility.
 174    */
 175   public get nextWorker (): number {
 176     return this.nextWorkerIndex
 177   }
 178
 179   /**
 180    * Perform the task specified in the constructor with the data parameter.
 181    *
 182    * @param data The input for the specified task.
 183    * @returns Promise that will be resolved when the task is successfully completed.
 184    */
 185   public execute (data: Data): Promise<Response> {
 186     // Configure worker to handle message with the specified task
 187     const worker = this.chooseWorker()
 188     this.increaseWorkersTask(worker)
 189     const messageId = ++this.nextMessageId
 190     const res = this.internalExecute(worker, messageId)
 191     this.sendToWorker(worker, { data: data || ({} as Data), id: messageId })
 192     return res
 193   }
 194
 195   /**
 196    * Shut down every current worker in this pool.
 197    */
 198   public async destroy (): Promise<void> {
 199     await Promise.all(this.workers.map(worker => this.destroyWorker(worker)))
 200   }
 201
 202   /**
 203    * Shut down given worker.
 204    *
 205    * @param worker A worker within `workers`.
 206    */
 207   protected abstract destroyWorker (worker: Worker): void | Promise<void>
 208
 209   /**
 210    * Setup hook that can be overridden by a Poolifier pool implementation
 211    * to run code before workers are created in the abstract constructor.
 212    */
 213   protected setupHook (): void {
 214     // Can be overridden
 215   }
 216
 217   /**
 218    * Should return whether the worker is the main worker or not.
 219    */
 220   protected abstract isMain (): boolean
 221
 222   /**
 223    * Increase the number of tasks that the given workers has done.
 224    *
 225    * @param worker Workers whose tasks are increased.
 226    */
 227   protected increaseWorkersTask (worker: Worker): void {
 228     const numberOfTasksInProgress = this.tasks.get(worker)
 229     if (numberOfTasksInProgress !== undefined) {
 230       this.tasks.set(worker, numberOfTasksInProgress + 1)
 231     } else {
 232       throw Error('Worker could not be found in tasks map')
 233     }
 234   }
 235
 236   /**
 237    * Decrease the number of tasks that the given workers has done.
 238    *
 239    * @param worker Workers whose tasks are decreased.
 240    */
 241   protected decreaseWorkersTasks (worker: Worker): void {
 242     const numberOfTasksInProgress = this.tasks.get(worker)
 243     if (numberOfTasksInProgress !== undefined) {
 244       this.tasks.set(worker, numberOfTasksInProgress - 1)
 245     } else {
 246       throw Error('Worker could not be found in tasks map')
 247     }
 248   }
 249
 250   /**
 251    * Removes the given worker from the pool.
 252    *
 253    * @param worker Worker that will be removed.
 254    */
 255   protected removeWorker (worker: Worker): void {
 256     // Clean worker from data structure
 257     const workerIndex = this.workers.indexOf(worker)
 258     this.workers.splice(workerIndex, 1)
 259     this.tasks.delete(worker)
 260   }
 261
 262   /**
 263    * Choose a worker for the next task.
 264    *
 265    * The default implementation uses a round robin algorithm to distribute the load.
 266    *
 267    * @returns Worker.
 268    */
 269   protected chooseWorker (): Worker {
 270     const chosenWorker = this.workers[this.nextWorkerIndex]
 271     this.nextWorkerIndex =
 272       this.workers.length - 1 === this.nextWorkerIndex
 273         ? 0
 274         : this.nextWorkerIndex + 1
 275     return chosenWorker
 276   }
 277
 278   /**
 279    * Send a message to the given worker.
 280    *
 281    * @param worker The worker which should receive the message.
 282    * @param message The message.
 283    */
 284   protected abstract sendToWorker (
 285     worker: Worker,
 286     message: MessageValue<Data>
 287   ): void
 288
 289   protected abstract registerWorkerMessageListener<
 290     Message extends Data | Response
 291   > (worker: Worker, listener: (message: MessageValue<Message>) => void): void
 292
 293   protected abstract unregisterWorkerMessageListener<
 294     Message extends Data | Response
 295   > (worker: Worker, listener: (message: MessageValue<Message>) => void): void
 296
 297   protected internalExecute (
 298     worker: Worker,
 299     messageId: number
 300   ): Promise<Response> {
 301     return new Promise((resolve, reject) => {
 302       const listener: (message: MessageValue<Response>) => void = message => {
 303         if (message.id === messageId) {
 304           this.unregisterWorkerMessageListener(worker, listener)
 305           this.decreaseWorkersTasks(worker)
 306           if (message.error) reject(message.error)
 307           else resolve(message.data as Response)
 308         }
 309       }
 310       this.registerWorkerMessageListener(worker, listener)
 311     })
 312   }
 313
 314   /**
 315    * Returns a newly created worker.
 316    */
 317   protected abstract createWorker (): Worker
 318
 319   /**
 320    * Function that can be hooked up when a worker has been newly created and moved to the workers registry.
 321    *
 322    * Can be used to update the `maxListeners` or binding the `main-worker`<->`worker` connection if not bind by default.
 323    *
 324    * @param worker The newly created worker.
 325    */
 326   protected abstract afterWorkerSetup (worker: Worker): void
 327
 328   /**
 329    * Creates a new worker for this pool and sets it up completely.
 330    *
 331    * @returns New, completely set up worker.
 332    */
 333   protected createAndSetupWorker (): Worker {
 334     const worker: Worker = this.createWorker()
 335
 336     worker.on('error', this.opts.errorHandler ?? (() => {}))
 337     worker.on('online', this.opts.onlineHandler ?? (() => {}))
 338     worker.on('exit', this.opts.exitHandler ?? (() => {}))
 339     worker.once('exit', () => this.removeWorker(worker))
 340
 341     this.workers.push(worker)
 342
 343     // Init tasks map
 344     this.tasks.set(worker, 0)
 345
 346     this.afterWorkerSetup(worker)
 347
 348     return worker
 349   }
 350 }