build(deps-dev): apply updates

[poolifier.git] / src / pools / abstract-pool.ts

diff --git a/src/pools/abstract-pool.ts b/src/pools/abstract-pool.ts
index 0c366911a1ddb059f3c993f69bfdabfa8404173a..b7b30161ac7676511f0805d56a5a9284299c7f66 100644 (file)
--- 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.
@@ -283,10 +284,11 @@ 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 === true &&
+        .runTime.aggregate === true &&
         this.workerChoiceStrategyContext.getTaskStatisticsRequirements()
           .waitTime.aggregate && {
         utilization: round(this.utilization)
@@ -351,7 +353,7 @@ export abstract class AbstractPool<
         0
       ),
       ...(this.workerChoiceStrategyContext?.getTaskStatisticsRequirements()
-        ?.runTime.aggregate === true && {
+        .runTime.aggregate === true && {
         runTime: {
           minimum: round(
             min(
@@ -394,7 +396,7 @@ export abstract class AbstractPool<
         }
       }),
       ...(this.workerChoiceStrategyContext?.getTaskStatisticsRequirements()
-        ?.waitTime.aggregate === true && {
+        .waitTime.aggregate === true && {
         waitTime: {
           minimum: round(
             min(
@@ -443,6 +445,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 +459,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.
    *
@@ -981,7 +993,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
@@ -1245,7 +1256,7 @@ export abstract class AbstractPool<
         this.redistributeQueuedTasks(this.workerNodes.indexOf(workerNode))
       }
       // eslint-disable-next-line @typescript-eslint/no-unnecessary-condition
-      workerNode?.terminate().catch(error => {
+      workerNode?.terminate().catch((error: unknown) => {
         this.emitter?.emit(PoolEvents.error, error)
       })
     })
@@ -1273,7 +1284,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) ||
@@ -1286,7 +1297,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)
         })
       }
@@ -1300,7 +1311,7 @@ export abstract class AbstractPool<
           taskFunctionOperation: 'add',
           taskFunctionName,
           taskFunction: taskFunction.toString()
-        }).catch(error => {
+        }).catch((error: unknown) => {
           this.emitter?.emit(PoolEvents.error, error)
         })
       }
@@ -1407,9 +1418,9 @@ export abstract class AbstractPool<
       statistics: {
         runTime:
           this.workerChoiceStrategyContext?.getTaskStatisticsRequirements()
-            ?.runTime.aggregate ?? false,
+            .runTime.aggregate ?? false,
         elu:
-          this.workerChoiceStrategyContext?.getTaskStatisticsRequirements()?.elu
+          this.workerChoiceStrategyContext?.getTaskStatisticsRequirements().elu
             .aggregate ?? false
       }
     })
@@ -1553,8 +1564,7 @@ export abstract class AbstractPool<
     ) {
       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
@@ -1603,7 +1613,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 = (
@@ -1709,6 +1721,13 @@ export abstract class AbstractPool<
     }
   }
 
+  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) {
@@ -1718,10 +1737,7 @@ export abstract class AbstractPool<
       this.workerNodes[this.getWorkerNodeKeyByWorkerId(workerId)]
     workerNode.info.ready = ready
     workerNode.info.taskFunctionNames = taskFunctionNames
-    if (!this.readyEventEmitted && this.ready) {
-      this.emitter?.emit(PoolEvents.ready, this.info)
-      this.readyEventEmitted = true
-    }
+    this.checkAndEmitReadyEvent()
   }
 
   private handleTaskExecutionResponse (message: MessageValue<Response>): void {
@@ -1751,7 +1767,12 @@ export abstract class AbstractPool<
       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 &&
@@ -1768,8 +1789,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
           })
         }
@@ -1846,6 +1866,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.
    *
@@ -1857,6 +1884,7 @@ export abstract class AbstractPool<
       this.workerNodes.splice(workerNodeKey, 1)
       this.workerChoiceStrategyContext?.remove(workerNodeKey)
     }
+    this.checkAndEmitEmptyEvent()
   }
 
   protected flagWorkerNodeAsNotReady (workerNodeKey: number): void {