chore: v2.6.14
[poolifier.git] / src / pools / abstract-pool.ts
index bfd0268aebff7a880c59a1bb77ec67603a16e3f7..c200d5317da6ba47c1154cdeb15a9b5ca5c9ee30 100644 (file)
@@ -1,7 +1,12 @@
-import crypto from 'node:crypto'
+import { randomUUID } from 'node:crypto'
 import { performance } from 'node:perf_hooks'
-import type { MessageValue, PromiseResponseWrapper } from '../utility-types'
+import type {
+  MessageValue,
+  PromiseResponseWrapper,
+  Task
+} from '../utility-types'
 import {
+  DEFAULT_TASK_NAME,
   DEFAULT_WORKER_CHOICE_STRATEGY_OPTIONS,
   EMPTY_FUNCTION,
   isKillBehavior,
@@ -10,8 +15,6 @@ import {
   round
 } from '../utils'
 import { KillBehaviors } from '../worker/worker-options'
-import { CircularArray } from '../circular-array'
-import { Queue } from '../queue'
 import {
   type IPool,
   PoolEmitter,
@@ -20,16 +23,14 @@ import {
   type PoolOptions,
   type PoolType,
   PoolTypes,
-  type TasksQueueOptions,
-  type WorkerType,
-  WorkerTypes
+  type TasksQueueOptions
 } from './pool'
 import type {
   IWorker,
+  IWorkerNode,
   MessageHandler,
-  Task,
   WorkerInfo,
-  WorkerNode,
+  WorkerType,
   WorkerUsage
 } from './worker'
 import {
@@ -40,6 +41,7 @@ import {
 } from './selection-strategies/selection-strategies-types'
 import { WorkerChoiceStrategyContext } from './selection-strategies/worker-choice-strategy-context'
 import { version } from './version'
+import { WorkerNode } from './worker-node'
 
 /**
  * Base class that implements some shared logic for all poolifier pools.
@@ -54,7 +56,7 @@ export abstract class AbstractPool<
   Response = unknown
 > implements IPool<Worker, Data, Response> {
   /** @inheritDoc */
-  public readonly workerNodes: Array<WorkerNode<Worker, Data>> = []
+  public readonly workerNodes: Array<IWorkerNode<Worker, Data>> = []
 
   /** @inheritDoc */
   public readonly emitter?: PoolEmitter
@@ -155,7 +157,25 @@ export abstract class AbstractPool<
         'Cannot instantiate a pool with a negative number of workers'
       )
     } else if (this.type === PoolTypes.fixed && numberOfWorkers === 0) {
-      throw new Error('Cannot instantiate a fixed pool with no worker')
+      throw new RangeError('Cannot instantiate a fixed pool with zero worker')
+    }
+  }
+
+  protected checkDynamicPoolSize (min: number, max: number): void {
+    if (this.type === PoolTypes.dynamic) {
+      if (min > max) {
+        throw new RangeError(
+          'Cannot instantiate a dynamic pool with a maximum pool size inferior to the minimum pool size'
+        )
+      } else if (min === 0 && max === 0) {
+        throw new RangeError(
+          'Cannot instantiate a dynamic pool with a minimum pool size and a maximum pool size equal to zero'
+        )
+      } else if (min === max) {
+        throw new RangeError(
+          'Cannot instantiate a dynamic pool with a minimum pool size equal to the maximum pool size. Use a fixed pool instead'
+        )
+      }
     }
   }
 
@@ -254,6 +274,8 @@ export abstract class AbstractPool<
       version,
       type: this.type,
       worker: this.worker,
+      ready: this.ready,
+      strategy: this.opts.workerChoiceStrategy as WorkerChoiceStrategy,
       minSize: this.minSize,
       maxSize: this.maxSize,
       ...(this.workerChoiceStrategyContext.getTaskStatisticsRequirements()
@@ -290,17 +312,114 @@ export abstract class AbstractPool<
       ),
       maxQueuedTasks: this.workerNodes.reduce(
         (accumulator, workerNode) =>
-          accumulator + workerNode.usage.tasks.maxQueued,
+          accumulator + (workerNode.usage.tasks?.maxQueued ?? 0),
         0
       ),
       failedTasks: this.workerNodes.reduce(
         (accumulator, workerNode) =>
           accumulator + workerNode.usage.tasks.failed,
         0
-      )
+      ),
+      ...(this.workerChoiceStrategyContext.getTaskStatisticsRequirements()
+        .runTime.aggregate && {
+        runTime: {
+          minimum: round(
+            Math.min(
+              ...this.workerNodes.map(
+                workerNode => workerNode.usage.runTime?.minimum ?? Infinity
+              )
+            )
+          ),
+          maximum: round(
+            Math.max(
+              ...this.workerNodes.map(
+                workerNode => workerNode.usage.runTime?.maximum ?? -Infinity
+              )
+            )
+          ),
+          average: round(
+            this.workerNodes.reduce(
+              (accumulator, workerNode) =>
+                accumulator + (workerNode.usage.runTime?.aggregate ?? 0),
+              0
+            ) /
+              this.workerNodes.reduce(
+                (accumulator, workerNode) =>
+                  accumulator + (workerNode.usage.tasks?.executed ?? 0),
+                0
+              )
+          ),
+          ...(this.workerChoiceStrategyContext.getTaskStatisticsRequirements()
+            .runTime.median && {
+            median: round(
+              median(
+                this.workerNodes.map(
+                  workerNode => workerNode.usage.runTime?.median ?? 0
+                )
+              )
+            )
+          })
+        }
+      }),
+      ...(this.workerChoiceStrategyContext.getTaskStatisticsRequirements()
+        .waitTime.aggregate && {
+        waitTime: {
+          minimum: round(
+            Math.min(
+              ...this.workerNodes.map(
+                workerNode => workerNode.usage.waitTime?.minimum ?? Infinity
+              )
+            )
+          ),
+          maximum: round(
+            Math.max(
+              ...this.workerNodes.map(
+                workerNode => workerNode.usage.waitTime?.maximum ?? -Infinity
+              )
+            )
+          ),
+          average: round(
+            this.workerNodes.reduce(
+              (accumulator, workerNode) =>
+                accumulator + (workerNode.usage.waitTime?.aggregate ?? 0),
+              0
+            ) /
+              this.workerNodes.reduce(
+                (accumulator, workerNode) =>
+                  accumulator + (workerNode.usage.tasks?.executed ?? 0),
+                0
+              )
+          ),
+          ...(this.workerChoiceStrategyContext.getTaskStatisticsRequirements()
+            .waitTime.median && {
+            median: round(
+              median(
+                this.workerNodes.map(
+                  workerNode => workerNode.usage.waitTime?.median ?? 0
+                )
+              )
+            )
+          })
+        }
+      })
     }
   }
 
+  private get starting (): boolean {
+    return (
+      this.workerNodes.length < this.minSize ||
+      (this.workerNodes.length >= this.minSize &&
+        this.workerNodes.some(workerNode => !workerNode.info.ready))
+    )
+  }
+
+  private get ready (): boolean {
+    return (
+      this.workerNodes.length >= this.minSize &&
+      this.workerNodes.every(workerNode => workerNode.info.ready)
+    )
+  }
+
   /**
    * Gets the approximate pool utilization.
    *
@@ -311,12 +430,12 @@ export abstract class AbstractPool<
       (performance.now() - this.startTimestamp) * this.maxSize
     const totalTasksRunTime = this.workerNodes.reduce(
       (accumulator, workerNode) =>
-        accumulator + workerNode.usage.runTime.aggregate,
+        accumulator + (workerNode.usage.runTime?.aggregate ?? 0),
       0
     )
     const totalTasksWaitTime = this.workerNodes.reduce(
       (accumulator, workerNode) =>
-        accumulator + workerNode.usage.waitTime.aggregate,
+        accumulator + (workerNode.usage.waitTime?.aggregate ?? 0),
       0
     )
     return (totalTasksRunTime + totalTasksWaitTime) / poolRunTimeCapacity
@@ -355,6 +474,17 @@ export abstract class AbstractPool<
       ?.worker
   }
 
+  private checkMessageWorkerId (message: MessageValue<Response>): void {
+    if (
+      message.workerId != null &&
+      this.getWorkerById(message.workerId) == null
+    ) {
+      throw new Error(
+        `Worker message received from unknown worker '${message.workerId}'`
+      )
+    }
+  }
+
   /**
    * Gets the given worker its worker node key.
    *
@@ -381,10 +511,7 @@ export abstract class AbstractPool<
       this.setWorkerChoiceStrategyOptions(workerChoiceStrategyOptions)
     }
     for (const workerNode of this.workerNodes) {
-      this.setWorkerNodeTasksUsage(
-        workerNode,
-        this.getInitialWorkerUsage(workerNode.worker)
-      )
+      workerNode.resetUsage()
       this.setWorkerStatistics(workerNode.worker)
     }
   }
@@ -465,11 +592,12 @@ export abstract class AbstractPool<
     const timestamp = performance.now()
     const workerNodeKey = this.chooseWorkerNode()
     const submittedTask: Task<Data> = {
-      name,
+      name: name ?? DEFAULT_TASK_NAME,
       // eslint-disable-next-line @typescript-eslint/consistent-type-assertions
       data: data ?? ({} as Data),
       timestamp,
-      id: crypto.randomUUID()
+      workerId: this.getWorkerInfo(workerNodeKey).id as number,
+      id: randomUUID()
     }
     const res = new Promise<Response>((resolve, reject) => {
       this.promiseResponseMap.set(submittedTask.id as string, {
@@ -547,6 +675,11 @@ export abstract class AbstractPool<
     const workerUsage = this.workerNodes[workerNodeKey].usage
     ++workerUsage.tasks.executing
     this.updateWaitTimeWorkerUsage(workerUsage, task)
+    const taskWorkerUsage = this.workerNodes[workerNodeKey].getTaskWorkerUsage(
+      task.name as string
+    ) as WorkerUsage
+    ++taskWorkerUsage.tasks.executing
+    this.updateWaitTimeWorkerUsage(taskWorkerUsage, task)
   }
 
   /**
@@ -560,10 +693,17 @@ export abstract class AbstractPool<
     worker: Worker,
     message: MessageValue<Response>
   ): void {
-    const workerUsage = this.workerNodes[this.getWorkerNodeKey(worker)].usage
+    const workerNodeKey = this.getWorkerNodeKey(worker)
+    const workerUsage = this.workerNodes[workerNodeKey].usage
     this.updateTaskStatisticsWorkerUsage(workerUsage, message)
     this.updateRunTimeWorkerUsage(workerUsage, message)
     this.updateEluWorkerUsage(workerUsage, message)
+    const taskWorkerUsage = this.workerNodes[workerNodeKey].getTaskWorkerUsage(
+      message.taskPerformance?.name ?? DEFAULT_TASK_NAME
+    ) as WorkerUsage
+    this.updateTaskStatisticsWorkerUsage(taskWorkerUsage, message)
+    this.updateRunTimeWorkerUsage(taskWorkerUsage, message)
+    this.updateEluWorkerUsage(taskWorkerUsage, message)
   }
 
   private updateTaskStatisticsWorkerUsage (
@@ -572,8 +712,9 @@ export abstract class AbstractPool<
   ): void {
     const workerTaskStatistics = workerUsage.tasks
     --workerTaskStatistics.executing
-    ++workerTaskStatistics.executed
-    if (message.taskError != null) {
+    if (message.taskError == null) {
+      ++workerTaskStatistics.executed
+    } else {
       ++workerTaskStatistics.failed
     }
   }
@@ -587,14 +728,15 @@ export abstract class AbstractPool<
         .aggregate
     ) {
       const taskRunTime = message.taskPerformance?.runTime ?? 0
-      workerUsage.runTime.aggregate += taskRunTime
+      workerUsage.runTime.aggregate =
+        (workerUsage.runTime.aggregate ?? 0) + taskRunTime
       workerUsage.runTime.minimum = Math.min(
         taskRunTime,
-        workerUsage.runTime.minimum ?? Infinity
+        workerUsage.runTime?.minimum ?? Infinity
       )
       workerUsage.runTime.maximum = Math.max(
         taskRunTime,
-        workerUsage.runTime.maximum ?? -Infinity
+        workerUsage.runTime?.maximum ?? -Infinity
       )
       if (
         this.workerChoiceStrategyContext.getTaskStatisticsRequirements().runTime
@@ -602,8 +744,7 @@ export abstract class AbstractPool<
         workerUsage.tasks.executed !== 0
       ) {
         workerUsage.runTime.average =
-          workerUsage.runTime.aggregate /
-          (workerUsage.tasks.executed - workerUsage.tasks.failed)
+          workerUsage.runTime.aggregate / workerUsage.tasks.executed
       }
       if (
         this.workerChoiceStrategyContext.getTaskStatisticsRequirements().runTime
@@ -626,14 +767,15 @@ export abstract class AbstractPool<
       this.workerChoiceStrategyContext.getTaskStatisticsRequirements().waitTime
         .aggregate
     ) {
-      workerUsage.waitTime.aggregate += taskWaitTime
+      workerUsage.waitTime.aggregate =
+        (workerUsage.waitTime?.aggregate ?? 0) + taskWaitTime
       workerUsage.waitTime.minimum = Math.min(
         taskWaitTime,
-        workerUsage.waitTime.minimum ?? Infinity
+        workerUsage.waitTime?.minimum ?? Infinity
       )
       workerUsage.waitTime.maximum = Math.max(
         taskWaitTime,
-        workerUsage.waitTime.maximum ?? -Infinity
+        workerUsage.waitTime?.maximum ?? -Infinity
       )
       if (
         this.workerChoiceStrategyContext.getTaskStatisticsRequirements()
@@ -641,13 +783,11 @@ export abstract class AbstractPool<
         workerUsage.tasks.executed !== 0
       ) {
         workerUsage.waitTime.average =
-          workerUsage.waitTime.aggregate /
-          (workerUsage.tasks.executed - workerUsage.tasks.failed)
+          workerUsage.waitTime.aggregate / workerUsage.tasks.executed
       }
       if (
         this.workerChoiceStrategyContext.getTaskStatisticsRequirements()
-          .waitTime.median &&
-        taskWaitTime != null
+          .waitTime.median
       ) {
         workerUsage.waitTime.history.push(taskWaitTime)
         workerUsage.waitTime.median = median(workerUsage.waitTime.history)
@@ -664,8 +804,12 @@ export abstract class AbstractPool<
         .aggregate
     ) {
       if (message.taskPerformance?.elu != null) {
-        workerUsage.elu.idle.aggregate += message.taskPerformance.elu.idle
-        workerUsage.elu.active.aggregate += message.taskPerformance.elu.active
+        workerUsage.elu.idle.aggregate =
+          (workerUsage.elu.idle?.aggregate ?? 0) +
+          message.taskPerformance.elu.idle
+        workerUsage.elu.active.aggregate =
+          (workerUsage.elu.active?.aggregate ?? 0) +
+          message.taskPerformance.elu.active
         if (workerUsage.elu.utilization != null) {
           workerUsage.elu.utilization =
             (workerUsage.elu.utilization +
@@ -676,42 +820,41 @@ export abstract class AbstractPool<
         }
         workerUsage.elu.idle.minimum = Math.min(
           message.taskPerformance.elu.idle,
-          workerUsage.elu.idle.minimum ?? Infinity
+          workerUsage.elu.idle?.minimum ?? Infinity
         )
         workerUsage.elu.idle.maximum = Math.max(
           message.taskPerformance.elu.idle,
-          workerUsage.elu.idle.maximum ?? -Infinity
+          workerUsage.elu.idle?.maximum ?? -Infinity
         )
         workerUsage.elu.active.minimum = Math.min(
           message.taskPerformance.elu.active,
-          workerUsage.elu.active.minimum ?? Infinity
+          workerUsage.elu.active?.minimum ?? Infinity
         )
         workerUsage.elu.active.maximum = Math.max(
           message.taskPerformance.elu.active,
-          workerUsage.elu.active.maximum ?? -Infinity
+          workerUsage.elu.active?.maximum ?? -Infinity
         )
-      }
-      if (
-        this.workerChoiceStrategyContext.getTaskStatisticsRequirements().elu
-          .average &&
-        workerUsage.tasks.executed !== 0
-      ) {
-        const executedTasks =
-          workerUsage.tasks.executed - workerUsage.tasks.failed
-        workerUsage.elu.idle.average =
-          workerUsage.elu.idle.aggregate / executedTasks
-        workerUsage.elu.active.average =
-          workerUsage.elu.active.aggregate / executedTasks
-      }
-      if (
-        this.workerChoiceStrategyContext.getTaskStatisticsRequirements().elu
-          .median &&
-        message.taskPerformance?.elu != null
-      ) {
-        workerUsage.elu.idle.history.push(message.taskPerformance.elu.idle)
-        workerUsage.elu.active.history.push(message.taskPerformance.elu.active)
-        workerUsage.elu.idle.median = median(workerUsage.elu.idle.history)
-        workerUsage.elu.active.median = median(workerUsage.elu.active.history)
+        if (
+          this.workerChoiceStrategyContext.getTaskStatisticsRequirements().elu
+            .average &&
+          workerUsage.tasks.executed !== 0
+        ) {
+          workerUsage.elu.idle.average =
+            workerUsage.elu.idle.aggregate / workerUsage.tasks.executed
+          workerUsage.elu.active.average =
+            workerUsage.elu.active.aggregate / workerUsage.tasks.executed
+        }
+        if (
+          this.workerChoiceStrategyContext.getTaskStatisticsRequirements().elu
+            .median
+        ) {
+          workerUsage.elu.idle.history.push(message.taskPerformance.elu.idle)
+          workerUsage.elu.active.history.push(
+            message.taskPerformance.elu.active
+          )
+          workerUsage.elu.idle.median = median(workerUsage.elu.idle.history)
+          workerUsage.elu.active.median = median(workerUsage.elu.active.history)
+        }
       }
     }
   }
@@ -784,6 +927,13 @@ export abstract class AbstractPool<
   protected afterWorkerSetup (worker: Worker): void {
     // Listen to worker messages.
     this.registerWorkerMessageListener(worker, this.workerListener())
+    // Send startup message to worker.
+    this.sendToWorker(worker, {
+      ready: false,
+      workerId: this.getWorkerInfo(this.getWorkerNodeKey(worker)).id as number
+    })
+    // Setup worker task statistics computation.
+    this.setWorkerStatistics(worker)
   }
 
   /**
@@ -801,33 +951,9 @@ export abstract class AbstractPool<
         this.emitter.emit(PoolEvents.error, error)
       }
       if (this.opts.enableTasksQueue === true) {
-        const workerNodeKey = this.getWorkerNodeKey(worker)
-        while (this.tasksQueueSize(workerNodeKey) > 0) {
-          let targetWorkerNodeKey: number = workerNodeKey
-          let minQueuedTasks = Infinity
-          for (const [workerNodeId, workerNode] of this.workerNodes.entries()) {
-            if (
-              workerNodeId !== workerNodeKey &&
-              workerNode.usage.tasks.queued === 0
-            ) {
-              targetWorkerNodeKey = workerNodeId
-              break
-            }
-            if (
-              workerNodeId !== workerNodeKey &&
-              workerNode.usage.tasks.queued < minQueuedTasks
-            ) {
-              minQueuedTasks = workerNode.usage.tasks.queued
-              targetWorkerNodeKey = workerNodeId
-            }
-          }
-          this.enqueueTask(
-            targetWorkerNodeKey,
-            this.dequeueTask(workerNodeKey) as Task<Data>
-          )
-        }
+        this.redistributeQueuedTasks(worker)
       }
-      if (this.opts.restartWorkerOnError === true) {
+      if (this.opts.restartWorkerOnError === true && !this.starting) {
         if (this.getWorkerInfo(this.getWorkerNodeKey(worker)).dynamic) {
           this.createAndSetupDynamicWorker()
         } else {
@@ -843,13 +969,39 @@ export abstract class AbstractPool<
 
     this.pushWorkerNode(worker)
 
-    this.setWorkerStatistics(worker)
-
     this.afterWorkerSetup(worker)
 
     return worker
   }
 
+  private redistributeQueuedTasks (worker: Worker): void {
+    const workerNodeKey = this.getWorkerNodeKey(worker)
+    while (this.tasksQueueSize(workerNodeKey) > 0) {
+      let targetWorkerNodeKey: number = workerNodeKey
+      let minQueuedTasks = Infinity
+      for (const [workerNodeId, workerNode] of this.workerNodes.entries()) {
+        if (
+          workerNodeId !== workerNodeKey &&
+          workerNode.usage.tasks.queued === 0
+        ) {
+          targetWorkerNodeKey = workerNodeId
+          break
+        }
+        if (
+          workerNodeId !== workerNodeKey &&
+          workerNode.usage.tasks.queued < minQueuedTasks
+        ) {
+          minQueuedTasks = workerNode.usage.tasks.queued
+          targetWorkerNodeKey = workerNodeId
+        }
+      }
+      this.enqueueTask(
+        targetWorkerNodeKey,
+        this.dequeueTask(workerNodeKey) as Task<Data>
+      )
+    }
+  }
+
   /**
    * Creates a new dynamic worker and sets it up completely in the pool worker nodes.
    *
@@ -857,7 +1009,6 @@ export abstract class AbstractPool<
    */
   protected createAndSetupDynamicWorker (): Worker {
     const worker = this.createAndSetupWorker()
-    this.getWorkerInfo(this.getWorkerNodeKey(worker)).dynamic = true
     this.registerWorkerMessageListener(worker, message => {
       const workerNodeKey = this.getWorkerNodeKey(worker)
       if (
@@ -873,6 +1024,12 @@ export abstract class AbstractPool<
         void (this.destroyWorker(worker) as Promise<void>)
       }
     })
+    const workerInfo = this.getWorkerInfo(this.getWorkerNodeKey(worker))
+    workerInfo.dynamic = true
+    this.sendToWorker(worker, {
+      checkAlive: true,
+      workerId: workerInfo.id as number
+    })
     return worker
   }
 
@@ -883,9 +1040,10 @@ export abstract class AbstractPool<
    */
   protected workerListener (): (message: MessageValue<Response>) => void {
     return message => {
-      if (message.workerId != null && message.started != null) {
-        // Worker started message received
-        this.handleWorkerStartedMessage(message)
+      this.checkMessageWorkerId(message)
+      if (message.ready != null && message.workerId != null) {
+        // Worker ready message received
+        this.handleWorkerReadyMessage(message)
       } else if (message.id != null) {
         // Task execution response received
         this.handleTaskExecutionResponse(message)
@@ -893,18 +1051,12 @@ export abstract class AbstractPool<
     }
   }
 
-  private handleWorkerStartedMessage (message: MessageValue<Response>): void {
-    // Worker started message received
-    const worker = this.getWorkerById(message.workerId as number)
-    if (worker != null) {
-      this.workerNodes[this.getWorkerNodeKey(worker)].info.started =
-        message.started as boolean
-    } else {
-      throw new Error(
-        `Worker started message received from unknown worker '${
-          message.workerId as number
-        }'`
-      )
+  private handleWorkerReadyMessage (message: MessageValue<Response>): void {
+    const worker = this.getWorkerById(message.workerId)
+    this.getWorkerInfo(this.getWorkerNodeKey(worker as Worker)).ready =
+      message.ready as boolean
+    if (this.emitter != null && this.ready) {
+      this.emitter.emit(PoolEvents.ready, this.info)
     }
   }
 
@@ -938,27 +1090,14 @@ export abstract class AbstractPool<
   private checkAndEmitEvents (): void {
     if (this.emitter != null) {
       if (this.busy) {
-        this.emitter?.emit(PoolEvents.busy, this.info)
+        this.emitter.emit(PoolEvents.busy, this.info)
       }
       if (this.type === PoolTypes.dynamic && this.full) {
-        this.emitter?.emit(PoolEvents.full, this.info)
+        this.emitter.emit(PoolEvents.full, this.info)
       }
     }
   }
 
-  /**
-   * Sets the given worker node its tasks usage in the pool.
-   *
-   * @param workerNode - The worker node.
-   * @param workerUsage - The worker usage.
-   */
-  private setWorkerNodeTasksUsage (
-    workerNode: WorkerNode<Worker, Data>,
-    workerUsage: WorkerUsage
-  ): void {
-    workerNode.usage = workerUsage
-  }
-
   /**
    * Gets the worker information.
    *
@@ -975,57 +1114,9 @@ export abstract class AbstractPool<
    * @returns The worker nodes length.
    */
   private pushWorkerNode (worker: Worker): number {
-    this.workerNodes.push({
-      worker,
-      info: this.getInitialWorkerInfo(worker),
-      usage: this.getInitialWorkerUsage(),
-      tasksQueue: new Queue<Task<Data>>()
-    })
-    this.setWorkerNodeTasksUsage(
-      this.workerNodes[this.getWorkerNodeKey(worker)],
-      this.getInitialWorkerUsage(worker)
-    )
-    return this.workerNodes.length
+    return this.workerNodes.push(new WorkerNode(worker, this.worker))
   }
 
-  /**
-   * Gets the worker id.
-   *
-   * @param worker - The worker.
-   * @returns The worker id.
-   */
-  private getWorkerId (worker: Worker): number | undefined {
-    if (this.worker === WorkerTypes.thread) {
-      return worker.threadId
-    } else if (this.worker === WorkerTypes.cluster) {
-      return worker.id
-    }
-  }
-
-  // /**
-  //  * Sets the given worker in the pool worker nodes.
-  //  *
-  //  * @param workerNodeKey - The worker node key.
-  //  * @param worker - The worker.
-  //  * @param workerInfo - The worker info.
-  //  * @param workerUsage - The worker usage.
-  //  * @param tasksQueue - The worker task queue.
-  //  */
-  // private setWorkerNode (
-  //   workerNodeKey: number,
-  //   worker: Worker,
-  //   workerInfo: WorkerInfo,
-  //   workerUsage: WorkerUsage,
-  //   tasksQueue: Queue<Task<Data>>
-  // ): void {
-  //   this.workerNodes[workerNodeKey] = {
-  //     worker,
-  //     info: workerInfo,
-  //     usage: workerUsage,
-  //     tasksQueue
-  //   }
-  // }
-
   /**
    * Removes the given worker from the pool worker nodes.
    *
@@ -1045,19 +1136,15 @@ export abstract class AbstractPool<
   }
 
   private enqueueTask (workerNodeKey: number, task: Task<Data>): number {
-    return this.workerNodes[workerNodeKey].tasksQueue.enqueue(task)
+    return this.workerNodes[workerNodeKey].enqueueTask(task)
   }
 
   private dequeueTask (workerNodeKey: number): Task<Data> | undefined {
-    return this.workerNodes[workerNodeKey].tasksQueue.dequeue()
+    return this.workerNodes[workerNodeKey].dequeueTask()
   }
 
   private tasksQueueSize (workerNodeKey: number): number {
-    return this.workerNodes[workerNodeKey].tasksQueue.size
-  }
-
-  private tasksMaxQueueSize (workerNodeKey: number): number {
-    return this.workerNodes[workerNodeKey].tasksQueue.maxSize
+    return this.workerNodes[workerNodeKey].tasksQueueSize()
   }
 
   private flushTasksQueue (workerNodeKey: number): void {
@@ -1067,7 +1154,7 @@ export abstract class AbstractPool<
         this.dequeueTask(workerNodeKey) as Task<Data>
       )
     }
-    this.workerNodes[workerNodeKey].tasksQueue.clear()
+    this.workerNodes[workerNodeKey].clearTasksQueue()
   }
 
   private flushTasksQueues (): void {
@@ -1084,71 +1171,8 @@ export abstract class AbstractPool<
             .runTime.aggregate,
         elu: this.workerChoiceStrategyContext.getTaskStatisticsRequirements()
           .elu.aggregate
-      }
-    })
-  }
-
-  private getInitialWorkerUsage (worker?: Worker): WorkerUsage {
-    const getTasksQueueSize = (worker?: Worker): number => {
-      return worker != null
-        ? this.tasksQueueSize(this.getWorkerNodeKey(worker))
-        : 0
-    }
-    const getTasksMaxQueueSize = (worker?: Worker): number => {
-      return worker != null
-        ? this.tasksMaxQueueSize(this.getWorkerNodeKey(worker))
-        : 0
-    }
-    return {
-      tasks: {
-        executed: 0,
-        executing: 0,
-        get queued (): number {
-          return getTasksQueueSize(worker)
-        },
-        get maxQueued (): number {
-          return getTasksMaxQueueSize(worker)
-        },
-        failed: 0
-      },
-      runTime: {
-        aggregate: 0,
-        maximum: 0,
-        minimum: 0,
-        average: 0,
-        median: 0,
-        history: new CircularArray()
-      },
-      waitTime: {
-        aggregate: 0,
-        maximum: 0,
-        minimum: 0,
-        average: 0,
-        median: 0,
-        history: new CircularArray()
       },
-      elu: {
-        idle: {
-          aggregate: 0,
-          maximum: 0,
-          minimum: 0,
-          average: 0,
-          median: 0,
-          history: new CircularArray()
-        },
-        active: {
-          aggregate: 0,
-          maximum: 0,
-          minimum: 0,
-          average: 0,
-          median: 0,
-          history: new CircularArray()
-        }
-      }
-    }
-  }
-
-  private getInitialWorkerInfo (worker: Worker): WorkerInfo {
-    return { id: this.getWorkerId(worker), dynamic: false, started: true }
+      workerId: this.getWorkerInfo(this.getWorkerNodeKey(worker)).id as number
+    })
   }
 }