X-Git-Url: https://git.piment-noir.org/?a=blobdiff_plain;f=src%2Fpools%2Fabstract-pool.ts;h=8e32218dfd521567663a27b09db1b2747d74b582;hb=1087637e65991bbcf07248b277d0fa99594a1528;hp=b14801b69269b43cfe878dee88ef4987151dfb33;hpb=55082af96253bead6fb8d4d648c454ba71a38fb6;p=poolifier.git

diff --git a/src/pools/abstract-pool.ts b/src/pools/abstract-pool.ts
index b14801b6..8e32218d 100644
--- a/src/pools/abstract-pool.ts
+++ b/src/pools/abstract-pool.ts
@@ -2,25 +2,27 @@ import { randomUUID } from 'node:crypto'
 import { performance } from 'node:perf_hooks'
 import type { TransferListItem } from 'node:worker_threads'
 import { EventEmitterAsyncResource } from 'node:events'
+import { AsyncResource } from 'node:async_hooks'
 import type {
   MessageValue,
   PromiseResponseWrapper,
   Task
-} from '../utility-types'
+} from '../utility-types.js'
 import {
   DEFAULT_TASK_NAME,
-  DEFAULT_WORKER_CHOICE_STRATEGY_OPTIONS,
   EMPTY_FUNCTION,
   average,
+  exponentialDelay,
   isKillBehavior,
   isPlainObject,
   max,
   median,
   min,
-  round
-} from '../utils'
-import { KillBehaviors } from '../worker/worker-options'
-import type { TaskFunction } from '../worker/task-functions'
+  round,
+  sleep
+} from '../utils.js'
+import { KillBehaviors } from '../worker/worker-options.js'
+import type { TaskFunction } from '../worker/task-functions.js'
 import {
   type IPool,
   PoolEvents,
@@ -29,31 +31,34 @@ import {
   type PoolType,
   PoolTypes,
   type TasksQueueOptions
-} from './pool'
+} from './pool.js'
 import type {
   IWorker,
   IWorkerNode,
   WorkerInfo,
   WorkerNodeEventDetail,
-  WorkerType,
-  WorkerUsage
-} from './worker'
+  WorkerType
+} from './worker.js'
 import {
-  type MeasurementStatisticsRequirements,
   Measurements,
   WorkerChoiceStrategies,
   type WorkerChoiceStrategy,
   type WorkerChoiceStrategyOptions
-} from './selection-strategies/selection-strategies-types'
-import { WorkerChoiceStrategyContext } from './selection-strategies/worker-choice-strategy-context'
-import { version } from './version'
-import { WorkerNode } from './worker-node'
+} from './selection-strategies/selection-strategies-types.js'
+import { WorkerChoiceStrategyContext } from './selection-strategies/worker-choice-strategy-context.js'
+import { version } from './version.js'
+import { WorkerNode } from './worker-node.js'
 import {
   checkFilePath,
   checkValidTasksQueueOptions,
   checkValidWorkerChoiceStrategy,
-  updateMeasurementStatistics
-} from './utils'
+  getDefaultTasksQueueOptions,
+  updateEluWorkerUsage,
+  updateRunTimeWorkerUsage,
+  updateTaskStatisticsWorkerUsage,
+  updateWaitTimeWorkerUsage,
+  waitWorkerNodeEvents
+} from './utils.js'
 
 /**
  * Base class that implements some shared logic for all poolifier pools.
@@ -73,11 +78,6 @@ export abstract class AbstractPool<
   /** @inheritDoc */
   public emitter?: EventEmitterAsyncResource
 
-  /**
-   * Dynamic pool maximum size property placeholder.
-   */
-  protected readonly max?: number
-
   /**
    * The task execution response promise map:
    * - `key`: The message id of each submitted task.
@@ -128,22 +128,25 @@ export abstract class AbstractPool<
   /**
    * Constructs a new poolifier pool.
    *
-   * @param numberOfWorkers - Number of workers that this pool should manage.
+   * @param minimumNumberOfWorkers - Minimum number of workers that this pool manages.
    * @param filePath - Path to the worker file.
    * @param opts - Options for the pool.
+   * @param maximumNumberOfWorkers - Maximum number of workers that this pool manages.
    */
   public constructor (
-    protected readonly numberOfWorkers: number,
+    protected readonly minimumNumberOfWorkers: number,
     protected readonly filePath: string,
-    protected readonly opts: PoolOptions<Worker>
+    protected readonly opts: PoolOptions<Worker>,
+    protected readonly maximumNumberOfWorkers?: number
   ) {
     if (!this.isMain()) {
       throw new Error(
         'Cannot start a pool from a worker with the same type as the pool'
       )
     }
+    this.checkPoolType()
     checkFilePath(this.filePath)
-    this.checkNumberOfWorkers(this.numberOfWorkers)
+    this.checkMinimumNumberOfWorkers(this.minimumNumberOfWorkers)
     this.checkPoolOptions(this.opts)
 
     this.chooseWorkerNode = this.chooseWorkerNode.bind(this)
@@ -178,20 +181,28 @@ export abstract class AbstractPool<
     this.startTimestamp = performance.now()
   }
 
-  private checkNumberOfWorkers (numberOfWorkers: number): void {
-    if (numberOfWorkers == null) {
+  private checkPoolType (): void {
+    if (this.type === PoolTypes.fixed && this.maximumNumberOfWorkers != null) {
+      throw new Error(
+        'Cannot instantiate a fixed pool with a maximum number of workers specified at initialization'
+      )
+    }
+  }
+
+  private checkMinimumNumberOfWorkers (minimumNumberOfWorkers: number): void {
+    if (minimumNumberOfWorkers == null) {
       throw new Error(
         'Cannot instantiate a pool without specifying the number of workers'
       )
-    } else if (!Number.isSafeInteger(numberOfWorkers)) {
+    } else if (!Number.isSafeInteger(minimumNumberOfWorkers)) {
       throw new TypeError(
         'Cannot instantiate a pool with a non safe integer number of workers'
       )
-    } else if (numberOfWorkers < 0) {
+    } else if (minimumNumberOfWorkers < 0) {
       throw new RangeError(
         'Cannot instantiate a pool with a negative number of workers'
       )
-    } else if (this.type === PoolTypes.fixed && numberOfWorkers === 0) {
+    } else if (this.type === PoolTypes.fixed && minimumNumberOfWorkers === 0) {
       throw new RangeError('Cannot instantiate a fixed pool with zero worker')
     }
   }
@@ -199,25 +210,26 @@ export abstract class AbstractPool<
   private checkPoolOptions (opts: PoolOptions<Worker>): void {
     if (isPlainObject(opts)) {
       this.opts.startWorkers = opts.startWorkers ?? true
-      checkValidWorkerChoiceStrategy(
-        opts.workerChoiceStrategy as WorkerChoiceStrategy
-      )
+      // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
+      checkValidWorkerChoiceStrategy(opts.workerChoiceStrategy!)
       this.opts.workerChoiceStrategy =
         opts.workerChoiceStrategy ?? WorkerChoiceStrategies.ROUND_ROBIN
       this.checkValidWorkerChoiceStrategyOptions(
-        opts.workerChoiceStrategyOptions as WorkerChoiceStrategyOptions
+        // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
+        opts.workerChoiceStrategyOptions!
       )
-      this.opts.workerChoiceStrategyOptions = {
-        ...DEFAULT_WORKER_CHOICE_STRATEGY_OPTIONS,
-        ...opts.workerChoiceStrategyOptions
+      if (opts.workerChoiceStrategyOptions != null) {
+        this.opts.workerChoiceStrategyOptions = opts.workerChoiceStrategyOptions
       }
       this.opts.restartWorkerOnError = opts.restartWorkerOnError ?? true
       this.opts.enableEvents = opts.enableEvents ?? true
       this.opts.enableTasksQueue = opts.enableTasksQueue ?? false
       if (this.opts.enableTasksQueue) {
-        checkValidTasksQueueOptions(opts.tasksQueueOptions as TasksQueueOptions)
+        // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
+        checkValidTasksQueueOptions(opts.tasksQueueOptions!)
         this.opts.tasksQueueOptions = this.buildTasksQueueOptions(
-          opts.tasksQueueOptions as TasksQueueOptions
+          // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
+          opts.tasksQueueOptions!
         )
       }
     } else {
@@ -236,25 +248,10 @@ export abstract class AbstractPool<
         'Invalid worker choice strategy options: must be a plain object'
       )
     }
-    if (
-      workerChoiceStrategyOptions?.retries != null &&
-      !Number.isSafeInteger(workerChoiceStrategyOptions.retries)
-    ) {
-      throw new TypeError(
-        'Invalid worker choice strategy options: retries must be an integer'
-      )
-    }
-    if (
-      workerChoiceStrategyOptions?.retries != null &&
-      workerChoiceStrategyOptions.retries < 0
-    ) {
-      throw new RangeError(
-        `Invalid worker choice strategy options: retries '${workerChoiceStrategyOptions.retries}' must be greater or equal than zero`
-      )
-    }
     if (
       workerChoiceStrategyOptions?.weights != null &&
-      Object.keys(workerChoiceStrategyOptions.weights).length !== this.maxSize
+      Object.keys(workerChoiceStrategyOptions.weights).length !==
+        (this.maximumNumberOfWorkers ?? this.minimumNumberOfWorkers)
     ) {
       throw new Error(
         'Invalid worker choice strategy options: must have a weight for each worker node'
@@ -286,12 +283,13 @@ export abstract class AbstractPool<
       worker: this.worker,
       started: this.started,
       ready: this.ready,
-      strategy: this.opts.workerChoiceStrategy as WorkerChoiceStrategy,
-      minSize: this.minSize,
-      maxSize: this.maxSize,
-      ...(this.workerChoiceStrategyContext.getTaskStatisticsRequirements()
+      // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
+      strategy: this.opts.workerChoiceStrategy!,
+      minSize: this.minimumNumberOfWorkers,
+      maxSize: this.maximumNumberOfWorkers ?? this.minimumNumberOfWorkers,
+      ...(this.workerChoiceStrategyContext?.getTaskStatisticsRequirements()
         .runTime.aggregate &&
-        this.workerChoiceStrategyContext.getTaskStatisticsRequirements()
+        this.workerChoiceStrategyContext?.getTaskStatisticsRequirements()
           .waitTime.aggregate && { utilization: round(this.utilization) }),
       workerNodes: this.workerNodes.length,
       idleWorkerNodes: this.workerNodes.reduce(
@@ -301,9 +299,16 @@ export abstract class AbstractPool<
             : accumulator,
         0
       ),
+      ...(this.opts.enableTasksQueue === true && {
+        stealingWorkerNodes: this.workerNodes.reduce(
+          (accumulator, workerNode) =>
+            workerNode.info.stealing ? accumulator + 1 : accumulator,
+          0
+        )
+      }),
       busyWorkerNodes: this.workerNodes.reduce(
-        (accumulator, workerNode) =>
-          workerNode.usage.tasks.executing > 0 ? accumulator + 1 : accumulator,
+        (accumulator, _workerNode, workerNodeKey) =>
+          this.isWorkerNodeBusy(workerNodeKey) ? accumulator + 1 : accumulator,
         0
       ),
       executedTasks: this.workerNodes.reduce(
@@ -345,7 +350,7 @@ export abstract class AbstractPool<
           accumulator + workerNode.usage.tasks.failed,
         0
       ),
-      ...(this.workerChoiceStrategyContext.getTaskStatisticsRequirements()
+      ...(this.workerChoiceStrategyContext?.getTaskStatisticsRequirements()
         .runTime.aggregate && {
         runTime: {
           minimum: round(
@@ -362,7 +367,7 @@ export abstract class AbstractPool<
               )
             )
           ),
-          ...(this.workerChoiceStrategyContext.getTaskStatisticsRequirements()
+          ...(this.workerChoiceStrategyContext?.getTaskStatisticsRequirements()
             .runTime.average && {
             average: round(
               average(
@@ -374,7 +379,7 @@ export abstract class AbstractPool<
               )
             )
           }),
-          ...(this.workerChoiceStrategyContext.getTaskStatisticsRequirements()
+          ...(this.workerChoiceStrategyContext?.getTaskStatisticsRequirements()
             .runTime.median && {
             median: round(
               median(
@@ -388,7 +393,7 @@ export abstract class AbstractPool<
           })
         }
       }),
-      ...(this.workerChoiceStrategyContext.getTaskStatisticsRequirements()
+      ...(this.workerChoiceStrategyContext?.getTaskStatisticsRequirements()
         .waitTime.aggregate && {
         waitTime: {
           minimum: round(
@@ -405,7 +410,7 @@ export abstract class AbstractPool<
               )
             )
           ),
-          ...(this.workerChoiceStrategyContext.getTaskStatisticsRequirements()
+          ...(this.workerChoiceStrategyContext?.getTaskStatisticsRequirements()
             .waitTime.average && {
             average: round(
               average(
@@ -417,7 +422,7 @@ export abstract class AbstractPool<
               )
             )
           }),
-          ...(this.workerChoiceStrategyContext.getTaskStatisticsRequirements()
+          ...(this.workerChoiceStrategyContext?.getTaskStatisticsRequirements()
             .waitTime.median && {
             median: round(
               median(
@@ -445,7 +450,7 @@ export abstract class AbstractPool<
             ? accumulator + 1
             : accumulator,
         0
-      ) >= this.minSize
+      ) >= this.minimumNumberOfWorkers
     )
   }
 
@@ -456,7 +461,8 @@ export abstract class AbstractPool<
    */
   private get utilization (): number {
     const poolTimeCapacity =
-      (performance.now() - this.startTimestamp) * this.maxSize
+      (performance.now() - this.startTimestamp) *
+      (this.maximumNumberOfWorkers ?? this.minimumNumberOfWorkers)
     const totalTasksRunTime = this.workerNodes.reduce(
       (accumulator, workerNode) =>
         accumulator + (workerNode.usage.runTime?.aggregate ?? 0),
@@ -482,20 +488,6 @@ export abstract class AbstractPool<
    */
   protected abstract get worker (): WorkerType
 
-  /**
-   * The pool minimum size.
-   */
-  protected get minSize (): number {
-    return this.numberOfWorkers
-  }
-
-  /**
-   * The pool maximum size.
-   */
-  protected get maxSize (): number {
-    return this.max ?? this.numberOfWorkers
-  }
-
   /**
    * Checks if the worker id sent in the received message from a worker is valid.
    *
@@ -512,18 +504,6 @@ export abstract class AbstractPool<
     }
   }
 
-  /**
-   * Gets the given worker its worker node key.
-   *
-   * @param worker - The worker.
-   * @returns The worker node key if found in the pool worker nodes, `-1` otherwise.
-   */
-  private getWorkerNodeKeyByWorker (worker: Worker): number {
-    return this.workerNodes.findIndex(
-      workerNode => workerNode.worker === worker
-    )
-  }
-
   /**
    * Gets the worker node key given its worker id.
    *
@@ -560,9 +540,8 @@ export abstract class AbstractPool<
     workerChoiceStrategyOptions: WorkerChoiceStrategyOptions
   ): void {
     this.checkValidWorkerChoiceStrategyOptions(workerChoiceStrategyOptions)
-    this.opts.workerChoiceStrategyOptions = {
-      ...DEFAULT_WORKER_CHOICE_STRATEGY_OPTIONS,
-      ...workerChoiceStrategyOptions
+    if (workerChoiceStrategyOptions != null) {
+      this.opts.workerChoiceStrategyOptions = workerChoiceStrategyOptions
     }
     this.workerChoiceStrategyContext.setOptions(
       this.opts.workerChoiceStrategyOptions
@@ -580,7 +559,8 @@ export abstract class AbstractPool<
       this.flushTasksQueues()
     }
     this.opts.enableTasksQueue = enable
-    this.setTasksQueueOptions(tasksQueueOptions as TasksQueueOptions)
+    // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
+    this.setTasksQueueOptions(tasksQueueOptions!)
   }
 
   /** @inheritDoc */
@@ -589,13 +569,16 @@ export abstract class AbstractPool<
       checkValidTasksQueueOptions(tasksQueueOptions)
       this.opts.tasksQueueOptions =
         this.buildTasksQueueOptions(tasksQueueOptions)
-      this.setTasksQueueSize(this.opts.tasksQueueOptions.size as number)
+      // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
+      this.setTasksQueueSize(this.opts.tasksQueueOptions.size!)
       if (this.opts.tasksQueueOptions.taskStealing === true) {
+        this.unsetTaskStealing()
         this.setTaskStealing()
       } else {
         this.unsetTaskStealing()
       }
       if (this.opts.tasksQueueOptions.tasksStealingOnBackPressure === true) {
+        this.unsetTasksStealingOnBackPressure()
         this.setTasksStealingOnBackPressure()
       } else {
         this.unsetTasksStealingOnBackPressure()
@@ -609,12 +592,9 @@ export abstract class AbstractPool<
     tasksQueueOptions: TasksQueueOptions
   ): TasksQueueOptions {
     return {
-      ...{
-        size: Math.pow(this.maxSize, 2),
-        concurrency: 1,
-        taskStealing: true,
-        tasksStealingOnBackPressure: true
-      },
+      ...getDefaultTasksQueueOptions(
+        this.maximumNumberOfWorkers ?? this.minimumNumberOfWorkers
+      ),
       ...tasksQueueOptions
     }
   }
@@ -627,36 +607,33 @@ export abstract class AbstractPool<
 
   private setTaskStealing (): void {
     for (const [workerNodeKey] of this.workerNodes.entries()) {
-      this.workerNodes[workerNodeKey].addEventListener(
-        'emptyQueue',
-        this.handleEmptyQueueEvent as EventListener
-      )
+      this.workerNodes[workerNodeKey].on('idle', this.handleWorkerNodeIdleEvent)
     }
   }
 
   private unsetTaskStealing (): void {
     for (const [workerNodeKey] of this.workerNodes.entries()) {
-      this.workerNodes[workerNodeKey].removeEventListener(
-        'emptyQueue',
-        this.handleEmptyQueueEvent as EventListener
+      this.workerNodes[workerNodeKey].off(
+        'idle',
+        this.handleWorkerNodeIdleEvent
       )
     }
   }
 
   private setTasksStealingOnBackPressure (): void {
     for (const [workerNodeKey] of this.workerNodes.entries()) {
-      this.workerNodes[workerNodeKey].addEventListener(
+      this.workerNodes[workerNodeKey].on(
         'backPressure',
-        this.handleBackPressureEvent as EventListener
+        this.handleWorkerNodeBackPressureEvent
       )
     }
   }
 
   private unsetTasksStealingOnBackPressure (): void {
     for (const [workerNodeKey] of this.workerNodes.entries()) {
-      this.workerNodes[workerNodeKey].removeEventListener(
+      this.workerNodes[workerNodeKey].off(
         'backPressure',
-        this.handleBackPressureEvent as EventListener
+        this.handleWorkerNodeBackPressureEvent
       )
     }
   }
@@ -667,7 +644,10 @@ export abstract class AbstractPool<
    * The pool filling boolean status.
    */
   protected get full (): boolean {
-    return this.workerNodes.length >= this.maxSize
+    return (
+      this.workerNodes.length >=
+      (this.maximumNumberOfWorkers ?? this.minimumNumberOfWorkers)
+    )
   }
 
   /**
@@ -689,7 +669,8 @@ export abstract class AbstractPool<
           workerNode =>
             workerNode.info.ready &&
             workerNode.usage.tasks.executing <
-              (this.opts.tasksQueueOptions?.concurrency as number)
+              // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
+              this.opts.tasksQueueOptions!.concurrency!
         ) === -1
       )
     }
@@ -701,6 +682,17 @@ export abstract class AbstractPool<
     )
   }
 
+  private isWorkerNodeBusy (workerNodeKey: number): boolean {
+    if (this.opts.enableTasksQueue === true) {
+      return (
+        this.workerNodes[workerNodeKey].usage.tasks.executing >=
+        // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
+        this.opts.tasksQueueOptions!.concurrency!
+      )
+    }
+    return this.workerNodes[workerNodeKey].usage.tasks.executing > 0
+  }
+
   private async sendTaskFunctionOperationToWorker (
     workerNodeKey: number,
     message: MessageValue<Data>
@@ -710,7 +702,8 @@ export abstract class AbstractPool<
         message: MessageValue<Response>
       ): void => {
         this.checkMessageWorkerId(message)
-        const workerId = this.getWorkerInfo(workerNodeKey).id as number
+        // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
+        const workerId = this.getWorkerInfo(workerNodeKey).id!
         if (
           message.taskFunctionOperationStatus != null &&
           message.workerId === workerId
@@ -722,8 +715,10 @@ export abstract class AbstractPool<
               new Error(
                 `Task function operation '${
                   message.taskFunctionOperation as string
+                  // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
                 }' failed on worker ${message.workerId} with error: '${
-                  message.workerError?.message as string
+                  // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
+                  message.workerError!.message
                 }'`
               )
             )
@@ -772,10 +767,11 @@ export abstract class AbstractPool<
                 new Error(
                   `Task function operation '${
                     message.taskFunctionOperation as string
-                  }' failed on worker ${
-                    errorResponse?.workerId as number
-                  } with error: '${
-                    errorResponse?.workerError?.message as string
+                    // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
+                  }' failed on worker ${errorResponse!
+                    .workerId!} with error: '${
+                    // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
+                    errorResponse!.workerError!.message
                   }'`
                 )
               )
@@ -880,7 +876,8 @@ export abstract class AbstractPool<
     return (
       this.tasksQueueSize(workerNodeKey) === 0 &&
       this.workerNodes[workerNodeKey].usage.tasks.executing <
-        (this.opts.tasksQueueOptions?.concurrency as number)
+        // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
+        this.opts.tasksQueueOptions!.concurrency!
     )
   }
 
@@ -925,10 +922,17 @@ export abstract class AbstractPool<
         timestamp,
         taskId: randomUUID()
       }
-      this.promiseResponseMap.set(task.taskId as string, {
+      // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
+      this.promiseResponseMap.set(task.taskId!, {
         resolve,
         reject,
-        workerNodeKey
+        workerNodeKey,
+        ...(this.emitter != null && {
+          asyncResource: new AsyncResource('poolifier:task', {
+            triggerAsyncId: this.emitter.asyncId,
+            requireManualDestroy: true
+          })
+        })
       })
       if (
         this.opts.enableTasksQueue === false ||
@@ -959,7 +963,7 @@ export abstract class AbstractPool<
         (accumulator, workerNode) =>
           !workerNode.info.dynamic ? accumulator + 1 : accumulator,
         0
-      ) < this.numberOfWorkers
+      ) < this.minimumNumberOfWorkers
     ) {
       this.createAndSetupWorkerNode()
     }
@@ -980,21 +984,24 @@ export abstract class AbstractPool<
     }
     this.destroying = true
     await Promise.all(
-      this.workerNodes.map(async (_, workerNodeKey) => {
+      this.workerNodes.map(async (_workerNode, workerNodeKey) => {
         await this.destroyWorkerNode(workerNodeKey)
       })
     )
     this.emitter?.emit(PoolEvents.destroy, this.info)
     this.emitter?.emitDestroy()
+    this.emitter?.removeAllListeners()
     this.readyEventEmitted = false
     this.destroying = false
     this.started = false
   }
 
-  protected async sendKillMessageToWorker (
-    workerNodeKey: number
-  ): Promise<void> {
+  private async sendKillMessageToWorker (workerNodeKey: number): Promise<void> {
     await new Promise<void>((resolve, reject) => {
+      if (this.workerNodes?.[workerNodeKey] == null) {
+        resolve()
+        return
+      }
       const killMessageListener = (message: MessageValue<Response>): void => {
         this.checkMessageWorkerId(message)
         if (message.kill === 'success') {
@@ -1002,9 +1009,8 @@ export abstract class AbstractPool<
         } else if (message.kill === 'failure') {
           reject(
             new Error(
-              `Kill message handling failed on worker ${
-                message.workerId as number
-              }`
+              // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
+              `Kill message handling failed on worker ${message.workerId!}`
             )
           )
         }
@@ -1020,7 +1026,22 @@ export abstract class AbstractPool<
    *
    * @param workerNodeKey - The worker node key.
    */
-  protected abstract destroyWorkerNode (workerNodeKey: number): Promise<void>
+  protected async destroyWorkerNode (workerNodeKey: number): Promise<void> {
+    this.flagWorkerNodeAsNotReady(workerNodeKey)
+    const flushedTasks = this.flushTasksQueue(workerNodeKey)
+    const workerNode = this.workerNodes[workerNodeKey]
+    await waitWorkerNodeEvents(
+      workerNode,
+      'taskFinished',
+      flushedTasks,
+      this.opts.tasksQueueOptions?.tasksFinishedTimeout ??
+        getDefaultTasksQueueOptions(
+          this.maximumNumberOfWorkers ?? this.minimumNumberOfWorkers
+        ).tasksFinishedTimeout
+    )
+    await this.sendKillMessageToWorker(workerNodeKey)
+    await workerNode.terminate()
+  }
 
   /**
    * Setup hook to execute code before worker nodes are created in the abstract constructor.
@@ -1051,19 +1072,29 @@ export abstract class AbstractPool<
     if (this.workerNodes[workerNodeKey]?.usage != null) {
       const workerUsage = this.workerNodes[workerNodeKey].usage
       ++workerUsage.tasks.executing
-      this.updateWaitTimeWorkerUsage(workerUsage, task)
+      updateWaitTimeWorkerUsage(
+        this.workerChoiceStrategyContext,
+        workerUsage,
+        task
+      )
     }
     if (
       this.shallUpdateTaskFunctionWorkerUsage(workerNodeKey) &&
-      this.workerNodes[workerNodeKey].getTaskFunctionWorkerUsage(
-        task.name as string
-      ) != null
+      // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
+      this.workerNodes[workerNodeKey].getTaskFunctionWorkerUsage(task.name!) !=
+        null
     ) {
+      // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
       const taskFunctionWorkerUsage = this.workerNodes[
         workerNodeKey
-      ].getTaskFunctionWorkerUsage(task.name as string) as WorkerUsage
+        // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
+      ].getTaskFunctionWorkerUsage(task.name!)!
       ++taskFunctionWorkerUsage.tasks.executing
-      this.updateWaitTimeWorkerUsage(taskFunctionWorkerUsage, task)
+      updateWaitTimeWorkerUsage(
+        this.workerChoiceStrategyContext,
+        taskFunctionWorkerUsage,
+        task
+      )
     }
   }
 
@@ -1078,26 +1109,49 @@ export abstract class AbstractPool<
     workerNodeKey: number,
     message: MessageValue<Response>
   ): void {
+    let needWorkerChoiceStrategyUpdate = false
     if (this.workerNodes[workerNodeKey]?.usage != null) {
       const workerUsage = this.workerNodes[workerNodeKey].usage
-      this.updateTaskStatisticsWorkerUsage(workerUsage, message)
-      this.updateRunTimeWorkerUsage(workerUsage, message)
-      this.updateEluWorkerUsage(workerUsage, message)
+      updateTaskStatisticsWorkerUsage(workerUsage, message)
+      updateRunTimeWorkerUsage(
+        this.workerChoiceStrategyContext,
+        workerUsage,
+        message
+      )
+      updateEluWorkerUsage(
+        this.workerChoiceStrategyContext,
+        workerUsage,
+        message
+      )
+      needWorkerChoiceStrategyUpdate = true
     }
     if (
       this.shallUpdateTaskFunctionWorkerUsage(workerNodeKey) &&
       this.workerNodes[workerNodeKey].getTaskFunctionWorkerUsage(
-        message.taskPerformance?.name as string
+        // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
+        message.taskPerformance!.name
       ) != null
     ) {
+      // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
       const taskFunctionWorkerUsage = this.workerNodes[
         workerNodeKey
-      ].getTaskFunctionWorkerUsage(
-        message.taskPerformance?.name as string
-      ) as WorkerUsage
-      this.updateTaskStatisticsWorkerUsage(taskFunctionWorkerUsage, message)
-      this.updateRunTimeWorkerUsage(taskFunctionWorkerUsage, message)
-      this.updateEluWorkerUsage(taskFunctionWorkerUsage, message)
+        // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
+      ].getTaskFunctionWorkerUsage(message.taskPerformance!.name)!
+      updateTaskStatisticsWorkerUsage(taskFunctionWorkerUsage, message)
+      updateRunTimeWorkerUsage(
+        this.workerChoiceStrategyContext,
+        taskFunctionWorkerUsage,
+        message
+      )
+      updateEluWorkerUsage(
+        this.workerChoiceStrategyContext,
+        taskFunctionWorkerUsage,
+        message
+      )
+      needWorkerChoiceStrategyUpdate = true
+    }
+    if (needWorkerChoiceStrategyUpdate) {
+      this.workerChoiceStrategyContext.update(workerNodeKey)
     }
   }
 
@@ -1116,84 +1170,6 @@ export abstract class AbstractPool<
     )
   }
 
-  private updateTaskStatisticsWorkerUsage (
-    workerUsage: WorkerUsage,
-    message: MessageValue<Response>
-  ): void {
-    const workerTaskStatistics = workerUsage.tasks
-    if (
-      workerTaskStatistics.executing != null &&
-      workerTaskStatistics.executing > 0
-    ) {
-      --workerTaskStatistics.executing
-    }
-    if (message.workerError == null) {
-      ++workerTaskStatistics.executed
-    } else {
-      ++workerTaskStatistics.failed
-    }
-  }
-
-  private updateRunTimeWorkerUsage (
-    workerUsage: WorkerUsage,
-    message: MessageValue<Response>
-  ): void {
-    if (message.workerError != null) {
-      return
-    }
-    updateMeasurementStatistics(
-      workerUsage.runTime,
-      this.workerChoiceStrategyContext.getTaskStatisticsRequirements().runTime,
-      message.taskPerformance?.runTime ?? 0
-    )
-  }
-
-  private updateWaitTimeWorkerUsage (
-    workerUsage: WorkerUsage,
-    task: Task<Data>
-  ): void {
-    const timestamp = performance.now()
-    const taskWaitTime = timestamp - (task.timestamp ?? timestamp)
-    updateMeasurementStatistics(
-      workerUsage.waitTime,
-      this.workerChoiceStrategyContext.getTaskStatisticsRequirements().waitTime,
-      taskWaitTime
-    )
-  }
-
-  private updateEluWorkerUsage (
-    workerUsage: WorkerUsage,
-    message: MessageValue<Response>
-  ): void {
-    if (message.workerError != null) {
-      return
-    }
-    const eluTaskStatisticsRequirements: MeasurementStatisticsRequirements =
-      this.workerChoiceStrategyContext.getTaskStatisticsRequirements().elu
-    updateMeasurementStatistics(
-      workerUsage.elu.active,
-      eluTaskStatisticsRequirements,
-      message.taskPerformance?.elu?.active ?? 0
-    )
-    updateMeasurementStatistics(
-      workerUsage.elu.idle,
-      eluTaskStatisticsRequirements,
-      message.taskPerformance?.elu?.idle ?? 0
-    )
-    if (eluTaskStatisticsRequirements.aggregate) {
-      if (message.taskPerformance?.elu != null) {
-        if (workerUsage.elu.utilization != null) {
-          workerUsage.elu.utilization =
-            (workerUsage.elu.utilization +
-              message.taskPerformance.elu.utilization) /
-            2
-        } else {
-          workerUsage.elu.utilization = message.taskPerformance.elu.utilization
-        }
-      }
-    }
-  }
-
   /**
    * Chooses a worker node for the next task.
    *
@@ -1218,9 +1194,7 @@ export abstract class AbstractPool<
    *
    * @returns Whether to create a dynamic worker or not.
    */
-  private shallCreateDynamicWorker (): boolean {
-    return this.type === PoolTypes.dynamic && !this.full && this.internalBusy()
-  }
+  protected abstract shallCreateDynamicWorker (): boolean
 
   /**
    * Sends a message to worker given its worker node key.
@@ -1235,55 +1209,59 @@ export abstract class AbstractPool<
     transferList?: TransferListItem[]
   ): void
 
-  /**
-   * Creates a new worker.
-   *
-   * @returns Newly created worker.
-   */
-  protected abstract createWorker (): Worker
-
   /**
    * Creates a new, completely set up worker node.
    *
    * @returns New, completely set up worker node key.
    */
   protected createAndSetupWorkerNode (): number {
-    const worker = this.createWorker()
-
-    worker.on('online', this.opts.onlineHandler ?? EMPTY_FUNCTION)
-    worker.on('message', this.opts.messageHandler ?? EMPTY_FUNCTION)
-    worker.on('error', this.opts.errorHandler ?? EMPTY_FUNCTION)
-    worker.on('error', error => {
-      const workerNodeKey = this.getWorkerNodeKeyByWorker(worker)
-      this.flagWorkerNodeAsNotReady(workerNodeKey)
-      const workerInfo = this.getWorkerInfo(workerNodeKey)
+    const workerNode = this.createWorkerNode()
+    workerNode.registerWorkerEventHandler(
+      'online',
+      this.opts.onlineHandler ?? EMPTY_FUNCTION
+    )
+    workerNode.registerWorkerEventHandler(
+      'message',
+      this.opts.messageHandler ?? EMPTY_FUNCTION
+    )
+    workerNode.registerWorkerEventHandler(
+      'error',
+      this.opts.errorHandler ?? EMPTY_FUNCTION
+    )
+    workerNode.registerWorkerEventHandler('error', (error: Error) => {
+      workerNode.info.ready = false
       this.emitter?.emit(PoolEvents.error, error)
-      this.workerNodes[workerNodeKey].closeChannel()
       if (
         this.started &&
-        !this.starting &&
         !this.destroying &&
         this.opts.restartWorkerOnError === true
       ) {
-        if (workerInfo.dynamic) {
+        if (workerNode.info.dynamic) {
           this.createAndSetupDynamicWorkerNode()
         } else {
           this.createAndSetupWorkerNode()
         }
       }
-      if (this.started && this.opts.enableTasksQueue === true) {
-        this.redistributeQueuedTasks(workerNodeKey)
+      if (
+        this.started &&
+        !this.destroying &&
+        this.opts.enableTasksQueue === true
+      ) {
+        this.redistributeQueuedTasks(this.workerNodes.indexOf(workerNode))
       }
+      workerNode?.terminate().catch(error => {
+        this.emitter?.emit(PoolEvents.error, error)
+      })
     })
-    worker.on('exit', this.opts.exitHandler ?? EMPTY_FUNCTION)
-    worker.once('exit', () => {
-      this.removeWorkerNode(worker)
+    workerNode.registerWorkerEventHandler(
+      'exit',
+      this.opts.exitHandler ?? EMPTY_FUNCTION
+    )
+    workerNode.registerOnceWorkerEventHandler('exit', () => {
+      this.removeWorkerNode(workerNode)
     })
-
-    const workerNodeKey = this.addWorkerNode(worker)
-
+    const workerNodeKey = this.addWorkerNode(workerNode)
     this.afterWorkerNodeSetup(workerNodeKey)
-
     return workerNodeKey
   }
 
@@ -1317,7 +1295,6 @@ export abstract class AbstractPool<
         })
       }
     })
-    const workerInfo = this.getWorkerInfo(workerNodeKey)
     this.sendToWorker(workerNodeKey, {
       checkActive: true
     })
@@ -1332,12 +1309,13 @@ export abstract class AbstractPool<
         })
       }
     }
-    workerInfo.dynamic = true
+    const workerNode = this.workerNodes[workerNodeKey]
+    workerNode.info.dynamic = true
     if (
       this.workerChoiceStrategyContext.getStrategyPolicy().dynamicWorkerReady ||
       this.workerChoiceStrategyContext.getStrategyPolicy().dynamicWorkerUsage
     ) {
-      workerInfo.ready = true
+      workerNode.info.ready = true
     }
     this.checkAndEmitDynamicWorkerCreationEvents()
     return workerNodeKey
@@ -1392,7 +1370,7 @@ export abstract class AbstractPool<
     // Listen to worker messages.
     this.registerWorkerMessageListener(
       workerNodeKey,
-      this.workerMessageListener.bind(this)
+      this.workerMessageListener
     )
     // Send the startup message to worker.
     this.sendStartupMessageToWorker(workerNodeKey)
@@ -1400,15 +1378,15 @@ export abstract class AbstractPool<
     this.sendStatisticsMessageToWorker(workerNodeKey)
     if (this.opts.enableTasksQueue === true) {
       if (this.opts.tasksQueueOptions?.taskStealing === true) {
-        this.workerNodes[workerNodeKey].addEventListener(
-          'emptyQueue',
-          this.handleEmptyQueueEvent as EventListener
+        this.workerNodes[workerNodeKey].on(
+          'idle',
+          this.handleWorkerNodeIdleEvent
         )
       }
       if (this.opts.tasksQueueOptions?.tasksStealingOnBackPressure === true) {
-        this.workerNodes[workerNodeKey].addEventListener(
+        this.workerNodes[workerNodeKey].on(
           'backPressure',
-          this.handleBackPressureEvent as EventListener
+          this.handleWorkerNodeBackPressureEvent
         )
       }
     }
@@ -1438,7 +1416,22 @@ export abstract class AbstractPool<
     })
   }
 
+  private cannotStealTask (): boolean {
+    return this.workerNodes.length <= 1 || this.info.queuedTasks === 0
+  }
+
+  private handleTask (workerNodeKey: number, task: Task<Data>): void {
+    if (this.shallExecuteTask(workerNodeKey)) {
+      this.executeTask(workerNodeKey, task)
+    } else {
+      this.enqueueTask(workerNodeKey, task)
+    }
+  }
+
   private redistributeQueuedTasks (workerNodeKey: number): void {
+    if (workerNodeKey === -1 || this.cannotStealTask()) {
+      return
+    }
     while (this.tasksQueueSize(workerNodeKey) > 0) {
       const destinationWorkerNodeKey = this.workerNodes.reduce(
         (minWorkerNodeKey, workerNode, workerNodeKey, workerNodes) => {
@@ -1450,12 +1443,11 @@ export abstract class AbstractPool<
         },
         0
       )
-      const task = this.dequeueTask(workerNodeKey) as Task<Data>
-      if (this.shallExecuteTask(destinationWorkerNodeKey)) {
-        this.executeTask(destinationWorkerNodeKey, task)
-      } else {
-        this.enqueueTask(destinationWorkerNodeKey, task)
-      }
+      this.handleTask(
+        destinationWorkerNodeKey,
+        // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
+        this.dequeueTask(workerNodeKey)!
+      )
     }
   }
 
@@ -1471,18 +1463,143 @@ export abstract class AbstractPool<
       this.shallUpdateTaskFunctionWorkerUsage(workerNodeKey) &&
       workerNode.getTaskFunctionWorkerUsage(taskName) != null
     ) {
-      const taskFunctionWorkerUsage = workerNode.getTaskFunctionWorkerUsage(
-        taskName
-      ) as WorkerUsage
+      const taskFunctionWorkerUsage =
+        // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
+        workerNode.getTaskFunctionWorkerUsage(taskName)!
       ++taskFunctionWorkerUsage.tasks.stolen
     }
   }
 
-  private readonly handleEmptyQueueEvent = (
-    event: CustomEvent<WorkerNodeEventDetail>
+  private updateTaskSequentiallyStolenStatisticsWorkerUsage (
+    workerNodeKey: number
+  ): void {
+    const workerNode = this.workerNodes[workerNodeKey]
+    if (workerNode?.usage != null) {
+      ++workerNode.usage.tasks.sequentiallyStolen
+    }
+  }
+
+  private updateTaskSequentiallyStolenStatisticsTaskFunctionWorkerUsage (
+    workerNodeKey: number,
+    taskName: string
+  ): void {
+    const workerNode = this.workerNodes[workerNodeKey]
+    if (
+      this.shallUpdateTaskFunctionWorkerUsage(workerNodeKey) &&
+      workerNode.getTaskFunctionWorkerUsage(taskName) != null
+    ) {
+      const taskFunctionWorkerUsage =
+        // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
+        workerNode.getTaskFunctionWorkerUsage(taskName)!
+      ++taskFunctionWorkerUsage.tasks.sequentiallyStolen
+    }
+  }
+
+  private resetTaskSequentiallyStolenStatisticsWorkerUsage (
+    workerNodeKey: number
+  ): void {
+    const workerNode = this.workerNodes[workerNodeKey]
+    if (workerNode?.usage != null) {
+      workerNode.usage.tasks.sequentiallyStolen = 0
+    }
+  }
+
+  private resetTaskSequentiallyStolenStatisticsTaskFunctionWorkerUsage (
+    workerNodeKey: number,
+    taskName: string
+  ): void {
+    const workerNode = this.workerNodes[workerNodeKey]
+    if (
+      this.shallUpdateTaskFunctionWorkerUsage(workerNodeKey) &&
+      workerNode.getTaskFunctionWorkerUsage(taskName) != null
+    ) {
+      const taskFunctionWorkerUsage =
+        // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
+        workerNode.getTaskFunctionWorkerUsage(taskName)!
+      taskFunctionWorkerUsage.tasks.sequentiallyStolen = 0
+    }
+  }
+
+  private readonly handleWorkerNodeIdleEvent = (
+    eventDetail: WorkerNodeEventDetail,
+    previousStolenTask?: Task<Data>
   ): void => {
-    const { workerId } = event.detail
-    const destinationWorkerNodeKey = this.getWorkerNodeKeyByWorkerId(workerId)
+    const { workerNodeKey } = eventDetail
+    if (workerNodeKey == null) {
+      throw new Error(
+        'WorkerNode event detail workerNodeKey property must be defined'
+      )
+    }
+    if (
+      this.cannotStealTask() ||
+      // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
+      this.info.stealingWorkerNodes! > Math.floor(this.workerNodes.length / 2)
+    ) {
+      if (previousStolenTask != null) {
+        this.getWorkerInfo(workerNodeKey).stealing = false
+      }
+      return
+    }
+    const workerNodeTasksUsage = this.workerNodes[workerNodeKey].usage.tasks
+    if (
+      previousStolenTask != null &&
+      workerNodeTasksUsage.sequentiallyStolen > 0 &&
+      (workerNodeTasksUsage.executing > 0 ||
+        this.tasksQueueSize(workerNodeKey) > 0)
+    ) {
+      this.getWorkerInfo(workerNodeKey).stealing = false
+      // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
+      for (const taskName of this.workerNodes[workerNodeKey].info
+        .taskFunctionNames!) {
+        this.resetTaskSequentiallyStolenStatisticsTaskFunctionWorkerUsage(
+          workerNodeKey,
+          taskName
+        )
+      }
+      this.resetTaskSequentiallyStolenStatisticsWorkerUsage(workerNodeKey)
+      return
+    }
+    this.getWorkerInfo(workerNodeKey).stealing = true
+    const stolenTask = this.workerNodeStealTask(workerNodeKey)
+    if (
+      this.shallUpdateTaskFunctionWorkerUsage(workerNodeKey) &&
+      stolenTask != null
+    ) {
+      // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
+      const taskFunctionTasksWorkerUsage = this.workerNodes[
+        workerNodeKey
+        // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
+      ].getTaskFunctionWorkerUsage(stolenTask.name!)!.tasks
+      if (
+        taskFunctionTasksWorkerUsage.sequentiallyStolen === 0 ||
+        (previousStolenTask != null &&
+          previousStolenTask.name === stolenTask.name &&
+          taskFunctionTasksWorkerUsage.sequentiallyStolen > 0)
+      ) {
+        this.updateTaskSequentiallyStolenStatisticsTaskFunctionWorkerUsage(
+          workerNodeKey,
+          // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
+          stolenTask.name!
+        )
+      } else {
+        this.resetTaskSequentiallyStolenStatisticsTaskFunctionWorkerUsage(
+          workerNodeKey,
+          // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
+          stolenTask.name!
+        )
+      }
+    }
+    sleep(exponentialDelay(workerNodeTasksUsage.sequentiallyStolen))
+      .then(() => {
+        this.handleWorkerNodeIdleEvent(eventDetail, stolenTask)
+        return undefined
+      })
+      .catch(EMPTY_FUNCTION)
+  }
+
+  private readonly workerNodeStealTask = (
+    workerNodeKey: number
+  ): Task<Data> | undefined => {
     const workerNodes = this.workerNodes
       .slice()
       .sort(
@@ -1490,31 +1607,37 @@ export abstract class AbstractPool<
           workerNodeB.usage.tasks.queued - workerNodeA.usage.tasks.queued
       )
     const sourceWorkerNode = workerNodes.find(
-      workerNode =>
-        workerNode.info.ready &&
-        workerNode.info.id !== workerId &&
-        workerNode.usage.tasks.queued > 0
+      (sourceWorkerNode, sourceWorkerNodeKey) =>
+        sourceWorkerNode.info.ready &&
+        !sourceWorkerNode.info.stealing &&
+        sourceWorkerNodeKey !== workerNodeKey &&
+        sourceWorkerNode.usage.tasks.queued > 0
     )
     if (sourceWorkerNode != null) {
-      const task = sourceWorkerNode.popTask() as Task<Data>
-      if (this.shallExecuteTask(destinationWorkerNodeKey)) {
-        this.executeTask(destinationWorkerNodeKey, task)
-      } else {
-        this.enqueueTask(destinationWorkerNodeKey, task)
-      }
-      this.updateTaskStolenStatisticsWorkerUsage(
-        destinationWorkerNodeKey,
-        task.name as string
-      )
+      // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
+      const task = sourceWorkerNode.popTask()!
+      this.handleTask(workerNodeKey, task)
+      this.updateTaskSequentiallyStolenStatisticsWorkerUsage(workerNodeKey)
+      // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
+      this.updateTaskStolenStatisticsWorkerUsage(workerNodeKey, task.name!)
+      return task
     }
   }
 
-  private readonly handleBackPressureEvent = (
-    event: CustomEvent<WorkerNodeEventDetail>
+  private readonly handleWorkerNodeBackPressureEvent = (
+    eventDetail: WorkerNodeEventDetail
   ): void => {
-    const { workerId } = event.detail
+    if (
+      this.cannotStealTask() ||
+      // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
+      this.info.stealingWorkerNodes! > Math.floor(this.workerNodes.length / 2)
+    ) {
+      return
+    }
+    const { workerId } = eventDetail
     const sizeOffset = 1
-    if ((this.opts.tasksQueueOptions?.size as number) <= sizeOffset) {
+    // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
+    if (this.opts.tasksQueueOptions!.size! <= sizeOffset) {
       return
     }
     const sourceWorkerNode =
@@ -1529,20 +1652,19 @@ export abstract class AbstractPool<
       if (
         sourceWorkerNode.usage.tasks.queued > 0 &&
         workerNode.info.ready &&
+        !workerNode.info.stealing &&
         workerNode.info.id !== workerId &&
         workerNode.usage.tasks.queued <
-          (this.opts.tasksQueueOptions?.size as number) - sizeOffset
+          // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
+          this.opts.tasksQueueOptions!.size! - sizeOffset
       ) {
-        const task = sourceWorkerNode.popTask() as Task<Data>
-        if (this.shallExecuteTask(workerNodeKey)) {
-          this.executeTask(workerNodeKey, task)
-        } else {
-          this.enqueueTask(workerNodeKey, task)
-        }
-        this.updateTaskStolenStatisticsWorkerUsage(
-          workerNodeKey,
-          task.name as string
-        )
+        this.getWorkerInfo(workerNodeKey).stealing = true
+        // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
+        const task = sourceWorkerNode.popTask()!
+        this.handleTask(workerNodeKey, task)
+        // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
+        this.updateTaskStolenStatisticsWorkerUsage(workerNodeKey, task.name!)
+        this.getWorkerInfo(workerNodeKey).stealing = false
       }
     }
   }
@@ -1550,63 +1672,89 @@ export abstract class AbstractPool<
   /**
    * This method is the message listener registered on each worker.
    */
-  protected workerMessageListener (message: MessageValue<Response>): void {
+  protected readonly workerMessageListener = (
+    message: MessageValue<Response>
+  ): void => {
     this.checkMessageWorkerId(message)
-    if (message.ready != null && message.taskFunctionNames != null) {
+    const { workerId, ready, taskId, taskFunctionNames } = message
+    if (ready != null && taskFunctionNames != null) {
       // Worker ready response received from worker
       this.handleWorkerReadyResponse(message)
-    } else if (message.taskId != null) {
+    } else if (taskId != null) {
       // Task execution response received from worker
       this.handleTaskExecutionResponse(message)
-    } else if (message.taskFunctionNames != null) {
+    } else if (taskFunctionNames != null) {
       // Task function names message received from worker
       this.getWorkerInfo(
-        this.getWorkerNodeKeyByWorkerId(message.workerId)
-      ).taskFunctionNames = message.taskFunctionNames
+        this.getWorkerNodeKeyByWorkerId(workerId)
+      ).taskFunctionNames = taskFunctionNames
     }
   }
 
   private handleWorkerReadyResponse (message: MessageValue<Response>): void {
-    if (message.ready === false) {
-      throw new Error(
-        `Worker ${message.workerId as number} failed to initialize`
-      )
+    const { workerId, ready, taskFunctionNames } = message
+    if (ready == null || !ready) {
+      // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
+      throw new Error(`Worker ${workerId!} failed to initialize`)
     }
-    const workerInfo = this.getWorkerInfo(
-      this.getWorkerNodeKeyByWorkerId(message.workerId)
-    )
-    workerInfo.ready = message.ready as boolean
-    workerInfo.taskFunctionNames = message.taskFunctionNames
+    const workerNode =
+      this.workerNodes[this.getWorkerNodeKeyByWorkerId(workerId)]
+    workerNode.info.ready = ready
+    workerNode.info.taskFunctionNames = taskFunctionNames
     if (!this.readyEventEmitted && this.ready) {
-      this.readyEventEmitted = true
       this.emitter?.emit(PoolEvents.ready, this.info)
+      this.readyEventEmitted = true
     }
   }
 
   private handleTaskExecutionResponse (message: MessageValue<Response>): void {
-    const { taskId, workerError, data } = message
-    const promiseResponse = this.promiseResponseMap.get(taskId as string)
+    const { workerId, taskId, workerError, data } = message
+    // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
+    const promiseResponse = this.promiseResponseMap.get(taskId!)
     if (promiseResponse != null) {
-      const { resolve, reject, workerNodeKey } = promiseResponse
+      const { resolve, reject, workerNodeKey, asyncResource } = promiseResponse
+      const workerNode = this.workerNodes[workerNodeKey]
       if (workerError != null) {
         this.emitter?.emit(PoolEvents.taskError, workerError)
-        reject(workerError.message)
+        asyncResource != null
+          ? asyncResource.runInAsyncScope(
+            reject,
+            this.emitter,
+            workerError.message
+          )
+          : reject(workerError.message)
       } else {
-        resolve(data as Response)
+        asyncResource != null
+          ? asyncResource.runInAsyncScope(resolve, this.emitter, data)
+          : resolve(data as Response)
       }
+      asyncResource?.emitDestroy()
       this.afterTaskExecutionHook(workerNodeKey, message)
-      this.workerChoiceStrategyContext.update(workerNodeKey)
-      this.promiseResponseMap.delete(taskId as string)
-      if (
-        this.opts.enableTasksQueue === true &&
-        this.tasksQueueSize(workerNodeKey) > 0 &&
-        this.workerNodes[workerNodeKey].usage.tasks.executing <
-          (this.opts.tasksQueueOptions?.concurrency as number)
-      ) {
-        this.executeTask(
-          workerNodeKey,
-          this.dequeueTask(workerNodeKey) as Task<Data>
-        )
+      // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
+      this.promiseResponseMap.delete(taskId!)
+      workerNode?.emit('taskFinished', taskId)
+      if (this.opts.enableTasksQueue === true && !this.destroying) {
+        const workerNodeTasksUsage = workerNode.usage.tasks
+        if (
+          this.tasksQueueSize(workerNodeKey) > 0 &&
+          workerNodeTasksUsage.executing <
+            // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
+            this.opts.tasksQueueOptions!.concurrency!
+        ) {
+          // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
+          this.executeTask(workerNodeKey, this.dequeueTask(workerNodeKey)!)
+        }
+        if (
+          workerNodeTasksUsage.executing === 0 &&
+          this.tasksQueueSize(workerNodeKey) === 0 &&
+          workerNodeTasksUsage.sequentiallyStolen === 0
+        ) {
+          workerNode.emit('idle', {
+            // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
+            workerId: workerId!,
+            workerNodeKey
+          })
+        }
       }
     }
   }
@@ -1623,13 +1771,10 @@ export abstract class AbstractPool<
     }
   }
 
-  private checkAndEmitDynamicWorkerCreationEvents (): void {
-    if (this.type === PoolTypes.dynamic) {
-      if (this.full) {
-        this.emitter?.emit(PoolEvents.full, this.info)
-      }
-    }
-  }
+  /**
+   * Emits dynamic worker creation events.
+   */
+  protected abstract checkAndEmitDynamicWorkerCreationEvents (): void
 
   /**
    * Gets the worker information given its worker node key.
@@ -1642,23 +1787,41 @@ export abstract class AbstractPool<
   }
 
   /**
-   * Adds the given worker in the pool worker nodes.
+   * Creates a worker node.
    *
-   * @param worker - The worker.
-   * @returns The added worker node key.
-   * @throws {@link https://nodejs.org/api/errors.html#class-error} If the added worker node is not found.
+   * @returns The created worker node.
    */
-  private addWorkerNode (worker: Worker): number {
+  private createWorkerNode (): IWorkerNode<Worker, Data> {
     const workerNode = new WorkerNode<Worker, Data>(
-      worker,
-      this.opts.tasksQueueOptions?.size ?? Math.pow(this.maxSize, 2)
+      this.worker,
+      this.filePath,
+      {
+        env: this.opts.env,
+        workerOptions: this.opts.workerOptions,
+        tasksQueueBackPressureSize:
+          this.opts.tasksQueueOptions?.size ??
+          getDefaultTasksQueueOptions(
+            this.maximumNumberOfWorkers ?? this.minimumNumberOfWorkers
+          ).size
+      }
     )
     // Flag the worker node as ready at pool startup.
     if (this.starting) {
       workerNode.info.ready = true
     }
+    return workerNode
+  }
+
+  /**
+   * Adds the given worker node in the pool worker nodes.
+   *
+   * @param workerNode - The worker node.
+   * @returns The added worker node key.
+   * @throws {@link https://nodejs.org/api/errors.html#class-error} If the added worker node is not found.
+   */
+  private addWorkerNode (workerNode: IWorkerNode<Worker, Data>): number {
     this.workerNodes.push(workerNode)
-    const workerNodeKey = this.getWorkerNodeKeyByWorker(worker)
+    const workerNodeKey = this.workerNodes.indexOf(workerNode)
     if (workerNodeKey === -1) {
       throw new Error('Worker added not found in worker nodes')
     }
@@ -1666,12 +1829,12 @@ export abstract class AbstractPool<
   }
 
   /**
-   * Removes the given worker from the pool worker nodes.
+   * Removes the worker node from the pool worker nodes.
    *
-   * @param worker - The worker.
+   * @param workerNode - The worker node.
    */
-  private removeWorkerNode (worker: Worker): void {
-    const workerNodeKey = this.getWorkerNodeKeyByWorker(worker)
+  private removeWorkerNode (workerNode: IWorkerNode<Worker, Data>): void {
+    const workerNodeKey = this.workerNodes.indexOf(workerNode)
     if (workerNodeKey !== -1) {
       this.workerNodes.splice(workerNodeKey, 1)
       this.workerChoiceStrategyContext.remove(workerNodeKey)
@@ -1682,14 +1845,6 @@ export abstract class AbstractPool<
     this.getWorkerInfo(workerNodeKey).ready = false
   }
 
-  /** @inheritDoc */
-  public hasWorkerNodeBackPressure (workerNodeKey: number): boolean {
-    return (
-      this.opts.enableTasksQueue === true &&
-      this.workerNodes[workerNodeKey].hasBackPressure()
-    )
-  }
-
   private hasBackPressure (): boolean {
     return (
       this.opts.enableTasksQueue === true &&
@@ -1725,14 +1880,15 @@ export abstract class AbstractPool<
     return this.workerNodes[workerNodeKey].tasksQueueSize()
   }
 
-  protected flushTasksQueue (workerNodeKey: number): void {
+  protected flushTasksQueue (workerNodeKey: number): number {
+    let flushedTasks = 0
     while (this.tasksQueueSize(workerNodeKey) > 0) {
-      this.executeTask(
-        workerNodeKey,
-        this.dequeueTask(workerNodeKey) as Task<Data>
-      )
+      // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
+      this.executeTask(workerNodeKey, this.dequeueTask(workerNodeKey)!)
+      ++flushedTasks
     }
     this.workerNodes[workerNodeKey].clearTasksQueue()
+    return flushedTasks
   }
 
   private flushTasksQueues (): void {