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