X-Git-Url: https://git.piment-noir.org/?a=blobdiff_plain;f=src%2Fpools%2Fabstract-pool.ts;h=b5e528ec643bb7d58325ab46d49f82c64f0f442b;hb=98f60ddd83d32e108adcfddebeb902f0d5a197eb;hp=536f4a03d8a0b2e052bc515b972d1de8098a94ae;hpb=e44639e9af74427b71f1556ff7ec9f7606373e0d;p=poolifier.git

diff --git a/src/pools/abstract-pool.ts b/src/pools/abstract-pool.ts
index 536f4a03..b5e528ec 100644
--- a/src/pools/abstract-pool.ts
+++ b/src/pools/abstract-pool.ts
@@ -1,17 +1,18 @@
+import { AsyncResource } from 'node:async_hooks'
 import { randomUUID } from 'node:crypto'
+import { EventEmitterAsyncResource } from 'node:events'
 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.js'
 import {
+  average,
   DEFAULT_TASK_NAME,
   EMPTY_FUNCTION,
-  average,
   exponentialDelay,
   isKillBehavior,
   isPlainObject,
@@ -21,8 +22,8 @@ import {
   round,
   sleep
 } from '../utils.js'
-import { KillBehaviors } from '../worker/worker-options.js'
 import type { TaskFunction } from '../worker/task-functions.js'
+import { KillBehaviors } from '../worker/worker-options.js'
 import {
   type IPool,
   PoolEvents,
@@ -32,13 +33,6 @@ import {
   PoolTypes,
   type TasksQueueOptions
 } from './pool.js'
-import type {
-  IWorker,
-  IWorkerNode,
-  WorkerInfo,
-  WorkerNodeEventDetail,
-  WorkerType
-} from './worker.js'
 import {
   Measurements,
   WorkerChoiceStrategies,
@@ -46,8 +40,6 @@ import {
   type WorkerChoiceStrategyOptions
 } 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,
@@ -59,6 +51,15 @@ import {
   updateWaitTimeWorkerUsage,
   waitWorkerNodeEvents
 } from './utils.js'
+import { version } from './version.js'
+import type {
+  IWorker,
+  IWorkerNode,
+  WorkerInfo,
+  WorkerNodeEventDetail,
+  WorkerType
+} from './worker.js'
+import { WorkerNode } from './worker-node.js'
 
 /**
  * Base class that implements some shared logic for all poolifier pools.
@@ -91,7 +92,7 @@ export abstract class AbstractPool<
   /**
    * Worker choice strategy context referencing a worker choice algorithm implementation.
    */
-  protected workerChoiceStrategyContext: WorkerChoiceStrategyContext<
+  protected workerChoiceStrategyContext?: WorkerChoiceStrategyContext<
   Worker,
   Data,
   Response
@@ -116,6 +117,10 @@ export abstract class AbstractPool<
    * Whether the pool is destroying or not.
    */
   private destroying: boolean
+  /**
+   * Whether the minimum number of workers is starting or not.
+   */
+  private startingMinimumNumberOfWorkers: boolean
   /**
    * Whether the pool ready event has been emitted or not.
    */
@@ -174,6 +179,7 @@ export abstract class AbstractPool<
     this.starting = false
     this.destroying = false
     this.readyEventEmitted = false
+    this.startingMinimumNumberOfWorkers = false
     if (this.opts.startWorkers === true) {
       this.start()
     }
@@ -189,7 +195,9 @@ export abstract class AbstractPool<
     }
   }
 
-  private checkMinimumNumberOfWorkers (minimumNumberOfWorkers: number): void {
+  private checkMinimumNumberOfWorkers (
+    minimumNumberOfWorkers: number | undefined
+  ): void {
     if (minimumNumberOfWorkers == null) {
       throw new Error(
         'Cannot instantiate a pool without specifying the number of workers'
@@ -210,13 +218,11 @@ export abstract class AbstractPool<
   private checkPoolOptions (opts: PoolOptions<Worker>): void {
     if (isPlainObject(opts)) {
       this.opts.startWorkers = opts.startWorkers ?? true
-      // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
-      checkValidWorkerChoiceStrategy(opts.workerChoiceStrategy!)
+      checkValidWorkerChoiceStrategy(opts.workerChoiceStrategy)
       this.opts.workerChoiceStrategy =
         opts.workerChoiceStrategy ?? WorkerChoiceStrategies.ROUND_ROBIN
       this.checkValidWorkerChoiceStrategyOptions(
-        // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
-        opts.workerChoiceStrategyOptions!
+        opts.workerChoiceStrategyOptions
       )
       if (opts.workerChoiceStrategyOptions != null) {
         this.opts.workerChoiceStrategyOptions = opts.workerChoiceStrategyOptions
@@ -225,11 +231,9 @@ export abstract class AbstractPool<
       this.opts.enableEvents = opts.enableEvents ?? true
       this.opts.enableTasksQueue = opts.enableTasksQueue ?? false
       if (this.opts.enableTasksQueue) {
-        // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
-        checkValidTasksQueueOptions(opts.tasksQueueOptions!)
+        checkValidTasksQueueOptions(opts.tasksQueueOptions)
         this.opts.tasksQueueOptions = this.buildTasksQueueOptions(
-          // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
-          opts.tasksQueueOptions!
+          opts.tasksQueueOptions
         )
       }
     } else {
@@ -238,7 +242,7 @@ export abstract class AbstractPool<
   }
 
   private checkValidWorkerChoiceStrategyOptions (
-    workerChoiceStrategyOptions: WorkerChoiceStrategyOptions
+    workerChoiceStrategyOptions: WorkerChoiceStrategyOptions | undefined
   ): void {
     if (
       workerChoiceStrategyOptions != null &&
@@ -285,12 +289,15 @@ export abstract class AbstractPool<
       ready: this.ready,
       // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
       strategy: this.opts.workerChoiceStrategy!,
+      strategyRetries: this.workerChoiceStrategyContext?.retriesCount ?? 0,
       minSize: this.minimumNumberOfWorkers,
       maxSize: this.maximumNumberOfWorkers ?? this.minimumNumberOfWorkers,
       ...(this.workerChoiceStrategyContext?.getTaskStatisticsRequirements()
-        .runTime.aggregate &&
-        this.workerChoiceStrategyContext?.getTaskStatisticsRequirements()
-          .waitTime.aggregate && { utilization: round(this.utilization) }),
+        .runTime.aggregate === true &&
+        this.workerChoiceStrategyContext.getTaskStatisticsRequirements()
+          .waitTime.aggregate && {
+        utilization: round(this.utilization)
+      }),
       workerNodes: this.workerNodes.length,
       idleWorkerNodes: this.workerNodes.reduce(
         (accumulator, workerNode) =>
@@ -331,7 +338,7 @@ export abstract class AbstractPool<
       ...(this.opts.enableTasksQueue === true && {
         maxQueuedTasks: this.workerNodes.reduce(
           (accumulator, workerNode) =>
-            accumulator + (workerNode.usage.tasks?.maxQueued ?? 0),
+            accumulator + (workerNode.usage.tasks.maxQueued ?? 0),
           0
         )
       }),
@@ -351,23 +358,23 @@ export abstract class AbstractPool<
         0
       ),
       ...(this.workerChoiceStrategyContext?.getTaskStatisticsRequirements()
-        .runTime.aggregate && {
+        .runTime.aggregate === true && {
         runTime: {
           minimum: round(
             min(
               ...this.workerNodes.map(
-                workerNode => workerNode.usage.runTime?.minimum ?? Infinity
+                workerNode => workerNode.usage.runTime.minimum ?? Infinity
               )
             )
           ),
           maximum: round(
             max(
               ...this.workerNodes.map(
-                workerNode => workerNode.usage.runTime?.maximum ?? -Infinity
+                workerNode => workerNode.usage.runTime.maximum ?? -Infinity
               )
             )
           ),
-          ...(this.workerChoiceStrategyContext?.getTaskStatisticsRequirements()
+          ...(this.workerChoiceStrategyContext.getTaskStatisticsRequirements()
             .runTime.average && {
             average: round(
               average(
@@ -379,7 +386,7 @@ export abstract class AbstractPool<
               )
             )
           }),
-          ...(this.workerChoiceStrategyContext?.getTaskStatisticsRequirements()
+          ...(this.workerChoiceStrategyContext.getTaskStatisticsRequirements()
             .runTime.median && {
             median: round(
               median(
@@ -394,23 +401,23 @@ export abstract class AbstractPool<
         }
       }),
       ...(this.workerChoiceStrategyContext?.getTaskStatisticsRequirements()
-        .waitTime.aggregate && {
+        .waitTime.aggregate === true && {
         waitTime: {
           minimum: round(
             min(
               ...this.workerNodes.map(
-                workerNode => workerNode.usage.waitTime?.minimum ?? Infinity
+                workerNode => workerNode.usage.waitTime.minimum ?? Infinity
               )
             )
           ),
           maximum: round(
             max(
               ...this.workerNodes.map(
-                workerNode => workerNode.usage.waitTime?.maximum ?? -Infinity
+                workerNode => workerNode.usage.waitTime.maximum ?? -Infinity
               )
             )
           ),
-          ...(this.workerChoiceStrategyContext?.getTaskStatisticsRequirements()
+          ...(this.workerChoiceStrategyContext.getTaskStatisticsRequirements()
             .waitTime.average && {
             average: round(
               average(
@@ -422,7 +429,7 @@ export abstract class AbstractPool<
               )
             )
           }),
-          ...(this.workerChoiceStrategyContext?.getTaskStatisticsRequirements()
+          ...(this.workerChoiceStrategyContext.getTaskStatisticsRequirements()
             .waitTime.median && {
             median: round(
               median(
@@ -443,6 +450,9 @@ export abstract class AbstractPool<
    * The pool readiness boolean status.
    */
   private get ready (): boolean {
+    if (this.empty) {
+      return false
+    }
     return (
       this.workerNodes.reduce(
         (accumulator, workerNode) =>
@@ -454,6 +464,13 @@ export abstract class AbstractPool<
     )
   }
 
+  /**
+   * The pool emptiness boolean status.
+   */
+  protected get empty (): boolean {
+    return this.minimumNumberOfWorkers === 0 && this.workerNodes.length === 0
+  }
+
   /**
    * The approximate pool utilization.
    *
@@ -465,12 +482,12 @@ export abstract class AbstractPool<
       (this.maximumNumberOfWorkers ?? this.minimumNumberOfWorkers)
     const totalTasksRunTime = this.workerNodes.reduce(
       (accumulator, workerNode) =>
-        accumulator + (workerNode.usage.runTime?.aggregate ?? 0),
+        accumulator + (workerNode.usage.runTime.aggregate ?? 0),
       0
     )
     const totalTasksWaitTime = this.workerNodes.reduce(
       (accumulator, workerNode) =>
-        accumulator + (workerNode.usage.waitTime?.aggregate ?? 0),
+        accumulator + (workerNode.usage.waitTime.aggregate ?? 0),
       0
     )
     return (totalTasksRunTime + totalTasksWaitTime) / poolTimeCapacity
@@ -523,7 +540,7 @@ export abstract class AbstractPool<
   ): void {
     checkValidWorkerChoiceStrategy(workerChoiceStrategy)
     this.opts.workerChoiceStrategy = workerChoiceStrategy
-    this.workerChoiceStrategyContext.setWorkerChoiceStrategy(
+    this.workerChoiceStrategyContext?.setWorkerChoiceStrategy(
       this.opts.workerChoiceStrategy
     )
     if (workerChoiceStrategyOptions != null) {
@@ -537,13 +554,13 @@ export abstract class AbstractPool<
 
   /** @inheritDoc */
   public setWorkerChoiceStrategyOptions (
-    workerChoiceStrategyOptions: WorkerChoiceStrategyOptions
+    workerChoiceStrategyOptions: WorkerChoiceStrategyOptions | undefined
   ): void {
     this.checkValidWorkerChoiceStrategyOptions(workerChoiceStrategyOptions)
     if (workerChoiceStrategyOptions != null) {
       this.opts.workerChoiceStrategyOptions = workerChoiceStrategyOptions
     }
-    this.workerChoiceStrategyContext.setOptions(
+    this.workerChoiceStrategyContext?.setOptions(
       this.opts.workerChoiceStrategyOptions
     )
   }
@@ -559,12 +576,13 @@ export abstract class AbstractPool<
       this.flushTasksQueues()
     }
     this.opts.enableTasksQueue = enable
-    // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
-    this.setTasksQueueOptions(tasksQueueOptions!)
+    this.setTasksQueueOptions(tasksQueueOptions)
   }
 
   /** @inheritDoc */
-  public setTasksQueueOptions (tasksQueueOptions: TasksQueueOptions): void {
+  public setTasksQueueOptions (
+    tasksQueueOptions: TasksQueueOptions | undefined
+  ): void {
     if (this.opts.enableTasksQueue === true) {
       checkValidTasksQueueOptions(tasksQueueOptions)
       this.opts.tasksQueueOptions =
@@ -589,7 +607,7 @@ export abstract class AbstractPool<
   }
 
   private buildTasksQueueOptions (
-    tasksQueueOptions: TasksQueueOptions
+    tasksQueueOptions: TasksQueueOptions | undefined
   ): TasksQueueOptions {
     return {
       ...getDefaultTasksQueueOptions(
@@ -702,24 +720,17 @@ export abstract class AbstractPool<
         message: MessageValue<Response>
       ): void => {
         this.checkMessageWorkerId(message)
-        // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
-        const workerId = this.getWorkerInfo(workerNodeKey).id!
+        const workerId = this.getWorkerInfo(workerNodeKey)?.id
         if (
           message.taskFunctionOperationStatus != null &&
           message.workerId === workerId
         ) {
           if (message.taskFunctionOperationStatus) {
             resolve(true)
-          } else if (!message.taskFunctionOperationStatus) {
+          } else {
             reject(
               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: '${
-                  // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
-                  message.workerError!.message
-                }'`
+                `Task function operation '${message.taskFunctionOperation}' failed on worker ${message.workerId} with error: '${message.workerError?.message}'`
               )
             )
           }
@@ -767,11 +778,8 @@ 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 ${errorResponse!
-                    .workerId!} with error: '${
-                    // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
-                    errorResponse!.workerError!.message
+                  }' failed on worker ${errorResponse?.workerId} with error: '${
+                    errorResponse?.workerError?.message
                   }'`
                 )
               )
@@ -946,6 +954,23 @@ export abstract class AbstractPool<
     })
   }
 
+  /**
+   * Starts the minimum number of workers.
+   */
+  private startMinimumNumberOfWorkers (): void {
+    this.startingMinimumNumberOfWorkers = true
+    while (
+      this.workerNodes.reduce(
+        (accumulator, workerNode) =>
+          !workerNode.info.dynamic ? accumulator + 1 : accumulator,
+        0
+      ) < this.minimumNumberOfWorkers
+    ) {
+      this.createAndSetupWorkerNode()
+    }
+    this.startingMinimumNumberOfWorkers = false
+  }
+
   /** @inheritdoc */
   public start (): void {
     if (this.started) {
@@ -958,15 +983,7 @@ export abstract class AbstractPool<
       throw new Error('Cannot start a destroying pool')
     }
     this.starting = true
-    while (
-      this.workerNodes.reduce(
-        (accumulator, workerNode) =>
-          !workerNode.info.dynamic ? accumulator + 1 : accumulator,
-        0
-      ) < this.minimumNumberOfWorkers
-    ) {
-      this.createAndSetupWorkerNode()
-    }
+    this.startMinimumNumberOfWorkers()
     this.starting = false
     this.started = true
   }
@@ -990,7 +1007,6 @@ export abstract class AbstractPool<
     )
     this.emitter?.emit(PoolEvents.destroy, this.info)
     this.emitter?.emitDestroy()
-    this.emitter?.removeAllListeners()
     this.readyEventEmitted = false
     this.destroying = false
     this.started = false
@@ -998,7 +1014,8 @@ export abstract class AbstractPool<
 
   private async sendKillMessageToWorker (workerNodeKey: number): Promise<void> {
     await new Promise<void>((resolve, reject) => {
-      if (this.workerNodes?.[workerNodeKey] == null) {
+      // eslint-disable-next-line @typescript-eslint/no-unnecessary-condition
+      if (this.workerNodes[workerNodeKey] == null) {
         resolve()
         return
       }
@@ -1009,8 +1026,7 @@ export abstract class AbstractPool<
         } else if (message.kill === 'failure') {
           reject(
             new Error(
-              // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
-              `Kill message handling failed on worker ${message.workerId!}`
+              `Kill message handling failed on worker ${message.workerId}`
             )
           )
         }
@@ -1054,7 +1070,9 @@ export abstract class AbstractPool<
   }
 
   /**
-   * Should return whether the worker is the main worker or not.
+   * Returns whether the worker is the main worker or not.
+   *
+   * @returns `true` if the worker is the main worker, `false` otherwise.
    */
   protected abstract isMain (): boolean
 
@@ -1069,6 +1087,7 @@ export abstract class AbstractPool<
     workerNodeKey: number,
     task: Task<Data>
   ): void {
+    // eslint-disable-next-line @typescript-eslint/no-unnecessary-condition
     if (this.workerNodes[workerNodeKey]?.usage != null) {
       const workerUsage = this.workerNodes[workerNodeKey].usage
       ++workerUsage.tasks.executing
@@ -1110,6 +1129,7 @@ export abstract class AbstractPool<
     message: MessageValue<Response>
   ): void {
     let needWorkerChoiceStrategyUpdate = false
+    // eslint-disable-next-line @typescript-eslint/no-unnecessary-condition
     if (this.workerNodes[workerNodeKey]?.usage != null) {
       const workerUsage = this.workerNodes[workerNodeKey].usage
       updateTaskStatisticsWorkerUsage(workerUsage, message)
@@ -1151,7 +1171,7 @@ export abstract class AbstractPool<
       needWorkerChoiceStrategyUpdate = true
     }
     if (needWorkerChoiceStrategyUpdate) {
-      this.workerChoiceStrategyContext.update(workerNodeKey)
+      this.workerChoiceStrategyContext?.update(workerNodeKey)
     }
   }
 
@@ -1181,12 +1201,14 @@ export abstract class AbstractPool<
     if (this.shallCreateDynamicWorker()) {
       const workerNodeKey = this.createAndSetupDynamicWorkerNode()
       if (
-        this.workerChoiceStrategyContext.getStrategyPolicy().dynamicWorkerUsage
+        this.workerChoiceStrategyContext?.getStrategyPolicy()
+          .dynamicWorkerUsage === true
       ) {
         return workerNodeKey
       }
     }
-    return this.workerChoiceStrategyContext.execute()
+    // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
+    return this.workerChoiceStrategyContext!.execute()
   }
 
   /**
@@ -1228,7 +1250,7 @@ export abstract class AbstractPool<
       'error',
       this.opts.errorHandler ?? EMPTY_FUNCTION
     )
-    workerNode.registerWorkerEventHandler('error', (error: Error) => {
+    workerNode.registerOnceWorkerEventHandler('error', (error: Error) => {
       workerNode.info.ready = false
       this.emitter?.emit(PoolEvents.error, error)
       if (
@@ -1238,8 +1260,8 @@ export abstract class AbstractPool<
       ) {
         if (workerNode.info.dynamic) {
           this.createAndSetupDynamicWorkerNode()
-        } else {
-          this.createAndSetupWorkerNode()
+        } else if (!this.startingMinimumNumberOfWorkers) {
+          this.startMinimumNumberOfWorkers()
         }
       }
       if (
@@ -1249,7 +1271,8 @@ export abstract class AbstractPool<
       ) {
         this.redistributeQueuedTasks(this.workerNodes.indexOf(workerNode))
       }
-      workerNode?.terminate().catch(error => {
+      // eslint-disable-next-line @typescript-eslint/no-unnecessary-condition
+      workerNode?.terminate().catch((error: unknown) => {
         this.emitter?.emit(PoolEvents.error, error)
       })
     })
@@ -1259,6 +1282,13 @@ export abstract class AbstractPool<
     )
     workerNode.registerOnceWorkerEventHandler('exit', () => {
       this.removeWorkerNode(workerNode)
+      if (
+        this.started &&
+        !this.startingMinimumNumberOfWorkers &&
+        !this.destroying
+      ) {
+        this.startMinimumNumberOfWorkers()
+      }
     })
     const workerNodeKey = this.addWorkerNode(workerNode)
     this.afterWorkerNodeSetup(workerNodeKey)
@@ -1277,7 +1307,7 @@ export abstract class AbstractPool<
       const localWorkerNodeKey = this.getWorkerNodeKeyByWorkerId(
         message.workerId
       )
-      const workerUsage = this.workerNodes[localWorkerNodeKey].usage
+      const workerUsage = this.workerNodes[localWorkerNodeKey]?.usage
       // Kill message received from worker
       if (
         isKillBehavior(KillBehaviors.HARD, message.kill) ||
@@ -1290,7 +1320,7 @@ export abstract class AbstractPool<
       ) {
         // Flag the worker node as not ready immediately
         this.flagWorkerNodeAsNotReady(localWorkerNodeKey)
-        this.destroyWorkerNode(localWorkerNodeKey).catch(error => {
+        this.destroyWorkerNode(localWorkerNodeKey).catch((error: unknown) => {
           this.emitter?.emit(PoolEvents.error, error)
         })
       }
@@ -1304,7 +1334,7 @@ export abstract class AbstractPool<
           taskFunctionOperation: 'add',
           taskFunctionName,
           taskFunction: taskFunction.toString()
-        }).catch(error => {
+        }).catch((error: unknown) => {
           this.emitter?.emit(PoolEvents.error, error)
         })
       }
@@ -1312,8 +1342,10 @@ export abstract class AbstractPool<
     const workerNode = this.workerNodes[workerNodeKey]
     workerNode.info.dynamic = true
     if (
-      this.workerChoiceStrategyContext.getStrategyPolicy().dynamicWorkerReady ||
-      this.workerChoiceStrategyContext.getStrategyPolicy().dynamicWorkerUsage
+      this.workerChoiceStrategyContext?.getStrategyPolicy()
+        .dynamicWorkerReady === true ||
+      this.workerChoiceStrategyContext?.getStrategyPolicy()
+        .dynamicWorkerUsage === true
     ) {
       workerNode.info.ready = true
     }
@@ -1408,10 +1440,11 @@ export abstract class AbstractPool<
     this.sendToWorker(workerNodeKey, {
       statistics: {
         runTime:
-          this.workerChoiceStrategyContext.getTaskStatisticsRequirements()
-            .runTime.aggregate,
-        elu: this.workerChoiceStrategyContext.getTaskStatisticsRequirements()
-          .elu.aggregate
+          this.workerChoiceStrategyContext?.getTaskStatisticsRequirements()
+            .runTime.aggregate ?? false,
+        elu:
+          this.workerChoiceStrategyContext?.getTaskStatisticsRequirements().elu
+            .aggregate ?? false
       }
     })
   }
@@ -1456,6 +1489,7 @@ export abstract class AbstractPool<
     taskName: string
   ): void {
     const workerNode = this.workerNodes[workerNodeKey]
+    // eslint-disable-next-line @typescript-eslint/no-unnecessary-condition
     if (workerNode?.usage != null) {
       ++workerNode.usage.tasks.stolen
     }
@@ -1474,6 +1508,7 @@ export abstract class AbstractPool<
     workerNodeKey: number
   ): void {
     const workerNode = this.workerNodes[workerNodeKey]
+    // eslint-disable-next-line @typescript-eslint/no-unnecessary-condition
     if (workerNode?.usage != null) {
       ++workerNode.usage.tasks.sequentiallyStolen
     }
@@ -1499,6 +1534,7 @@ export abstract class AbstractPool<
     workerNodeKey: number
   ): void {
     const workerNode = this.workerNodes[workerNodeKey]
+    // eslint-disable-next-line @typescript-eslint/no-unnecessary-condition
     if (workerNode?.usage != null) {
       workerNode.usage.tasks.sequentiallyStolen = 0
     }
@@ -1527,30 +1563,31 @@ export abstract class AbstractPool<
     const { workerNodeKey } = eventDetail
     if (workerNodeKey == null) {
       throw new Error(
-        'WorkerNode event detail workerNodeKey property must be defined'
+        "WorkerNode event detail 'workerNodeKey' property must be defined"
       )
     }
+    const workerInfo = this.getWorkerInfo(workerNodeKey)
     if (
       this.cannotStealTask() ||
-      // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
-      this.info.stealingWorkerNodes! > Math.floor(this.workerNodes.length / 2)
+      (this.info.stealingWorkerNodes ?? 0) >
+        Math.floor(this.workerNodes.length / 2)
     ) {
-      if (previousStolenTask != null) {
-        this.getWorkerInfo(workerNodeKey).stealing = false
+      if (workerInfo != null && previousStolenTask != null) {
+        workerInfo.stealing = false
       }
       return
     }
     const workerNodeTasksUsage = this.workerNodes[workerNodeKey].usage.tasks
     if (
+      workerInfo != null &&
       previousStolenTask != null &&
       workerNodeTasksUsage.sequentiallyStolen > 0 &&
       (workerNodeTasksUsage.executing > 0 ||
         this.tasksQueueSize(workerNodeKey) > 0)
     ) {
-      this.getWorkerInfo(workerNodeKey).stealing = false
+      workerInfo.stealing = false
       // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
-      for (const taskName of this.workerNodes[workerNodeKey].info
-        .taskFunctionNames!) {
+      for (const taskName of workerInfo.taskFunctionNames!) {
         this.resetTaskSequentiallyStolenStatisticsTaskFunctionWorkerUsage(
           workerNodeKey,
           taskName
@@ -1559,7 +1596,12 @@ export abstract class AbstractPool<
       this.resetTaskSequentiallyStolenStatisticsWorkerUsage(workerNodeKey)
       return
     }
-    this.getWorkerInfo(workerNodeKey).stealing = true
+    if (workerInfo == null) {
+      throw new Error(
+        `Worker node with key '${workerNodeKey}' not found in pool`
+      )
+    }
+    workerInfo.stealing = true
     const stolenTask = this.workerNodeStealTask(workerNodeKey)
     if (
       this.shallUpdateTaskFunctionWorkerUsage(workerNodeKey) &&
@@ -1594,7 +1636,9 @@ export abstract class AbstractPool<
         this.handleWorkerNodeIdleEvent(eventDetail, stolenTask)
         return undefined
       })
-      .catch(EMPTY_FUNCTION)
+      .catch((error: unknown) => {
+        this.emitter?.emit(PoolEvents.error, error)
+      })
   }
 
   private readonly workerNodeStealTask = (
@@ -1629,8 +1673,8 @@ export abstract class AbstractPool<
   ): void => {
     if (
       this.cannotStealTask() ||
-      // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
-      this.info.stealingWorkerNodes! > Math.floor(this.workerNodes.length / 2)
+      (this.info.stealingWorkerNodes ?? 0) >
+        Math.floor(this.workerNodes.length / 2)
     ) {
       return
     }
@@ -1658,13 +1702,19 @@ export abstract class AbstractPool<
           // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
           this.opts.tasksQueueOptions!.size! - sizeOffset
       ) {
-        this.getWorkerInfo(workerNodeKey).stealing = true
+        const workerInfo = this.getWorkerInfo(workerNodeKey)
+        if (workerInfo == null) {
+          throw new Error(
+            `Worker node with key '${workerNodeKey}' not found in pool`
+          )
+        }
+        workerInfo.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
+        workerInfo.stealing = false
       }
     }
   }
@@ -1685,26 +1735,32 @@ export abstract class AbstractPool<
       this.handleTaskExecutionResponse(message)
     } else if (taskFunctionNames != null) {
       // Task function names message received from worker
-      this.getWorkerInfo(
+      const workerInfo = this.getWorkerInfo(
         this.getWorkerNodeKeyByWorkerId(workerId)
-      ).taskFunctionNames = taskFunctionNames
+      )
+      if (workerInfo != null) {
+        workerInfo.taskFunctionNames = taskFunctionNames
+      }
+    }
+  }
+
+  private checkAndEmitReadyEvent (): void {
+    if (!this.readyEventEmitted && this.ready) {
+      this.emitter?.emit(PoolEvents.ready, this.info)
+      this.readyEventEmitted = true
     }
   }
 
   private handleWorkerReadyResponse (message: MessageValue<Response>): void {
     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`)
+      throw new Error(`Worker ${workerId} failed to initialize`)
     }
     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.checkAndEmitReadyEvent()
   }
 
   private handleTaskExecutionResponse (message: MessageValue<Response>): void {
@@ -1732,8 +1788,14 @@ export abstract class AbstractPool<
       this.afterTaskExecutionHook(workerNodeKey, message)
       // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
       this.promiseResponseMap.delete(taskId!)
+      // eslint-disable-next-line @typescript-eslint/no-unnecessary-condition
       workerNode?.emit('taskFinished', taskId)
-      if (this.opts.enableTasksQueue === true && !this.destroying) {
+      if (
+        this.opts.enableTasksQueue === true &&
+        !this.destroying &&
+        // eslint-disable-next-line @typescript-eslint/no-unnecessary-condition
+        workerNode != null
+      ) {
         const workerNodeTasksUsage = workerNode.usage.tasks
         if (
           this.tasksQueueSize(workerNodeKey) > 0 &&
@@ -1750,8 +1812,7 @@ export abstract class AbstractPool<
           workerNodeTasksUsage.sequentiallyStolen === 0
         ) {
           workerNode.emit('idle', {
-            // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
-            workerId: workerId!,
+            workerId,
             workerNodeKey
           })
         }
@@ -1782,7 +1843,7 @@ export abstract class AbstractPool<
    * @param workerNodeKey - The worker node key.
    * @returns The worker information.
    */
-  protected getWorkerInfo (workerNodeKey: number): WorkerInfo {
+  protected getWorkerInfo (workerNodeKey: number): WorkerInfo | undefined {
     return this.workerNodes[workerNodeKey]?.info
   }
 
@@ -1828,6 +1889,13 @@ export abstract class AbstractPool<
     return workerNodeKey
   }
 
+  private checkAndEmitEmptyEvent (): void {
+    if (this.empty) {
+      this.emitter?.emit(PoolEvents.empty, this.info)
+      this.readyEventEmitted = false
+    }
+  }
+
   /**
    * Removes the worker node from the pool worker nodes.
    *
@@ -1837,20 +1905,16 @@ export abstract class AbstractPool<
     const workerNodeKey = this.workerNodes.indexOf(workerNode)
     if (workerNodeKey !== -1) {
       this.workerNodes.splice(workerNodeKey, 1)
-      this.workerChoiceStrategyContext.remove(workerNodeKey)
+      this.workerChoiceStrategyContext?.remove(workerNodeKey)
     }
+    this.checkAndEmitEmptyEvent()
   }
 
   protected flagWorkerNodeAsNotReady (workerNodeKey: number): void {
-    this.getWorkerInfo(workerNodeKey).ready = false
-  }
-
-  /** @inheritDoc */
-  public hasWorkerNodeBackPressure (workerNodeKey: number): boolean {
-    return (
-      this.opts.enableTasksQueue === true &&
-      this.workerNodes[workerNodeKey].hasBackPressure()
-    )
+    const workerInfo = this.getWorkerInfo(workerNodeKey)
+    if (workerInfo != null) {
+      workerInfo.ready = false
+    }
   }
 
   private hasBackPressure (): boolean {