X-Git-Url: https://git.piment-noir.org/?a=blobdiff_plain;f=src%2Fpools%2Fabstract-pool.ts;h=52a2aa383da0dd89acf4b3db7e518fd18fc33f21;hb=985d0e7986b2cad23bb08ae0561b2a6ff9afdf9e;hp=6fde2491b6e1a565ba48ce26d821407eff67c8a2;hpb=bb4e9c3d36509745767905e4577d000087a23288;p=poolifier.git

diff --git a/src/pools/abstract-pool.ts b/src/pools/abstract-pool.ts
index 6fde2491..52a2aa38 100644
--- a/src/pools/abstract-pool.ts
+++ b/src/pools/abstract-pool.ts
@@ -5,7 +5,8 @@ import {
   DEFAULT_WORKER_CHOICE_STRATEGY_OPTIONS,
   EMPTY_FUNCTION,
   isPlainObject,
-  median
+  median,
+  round
 } from '../utils'
 import { KillBehaviors, isKillBehavior } from '../worker/worker-options'
 import { CircularArray } from '../circular-array'
@@ -19,7 +20,8 @@ import {
   type PoolType,
   PoolTypes,
   type TasksQueueOptions,
-  type WorkerType
+  type WorkerType,
+  WorkerTypes
 } from './pool'
 import type {
   IWorker,
@@ -76,6 +78,11 @@ export abstract class AbstractPool<
   Response
   >
 
+  /**
+   * The start timestamp of the pool.
+   */
+  private readonly startTimestamp
+
   /**
    * Constructs a new poolifier pool.
    *
@@ -115,9 +122,11 @@ export abstract class AbstractPool<
 
     this.setupHook()
 
-    for (let i = 1; i <= this.numberOfWorkers; i++) {
+    while (this.workerNodes.length < this.numberOfWorkers) {
       this.createAndSetupWorker()
     }
+
+    this.startTimestamp = performance.now()
   }
 
   private checkFilePath (filePath: string): void {
@@ -236,14 +245,6 @@ export abstract class AbstractPool<
     }
   }
 
-  private get starting (): boolean {
-    return this.workerNodes.some(workerNode => !workerNode.info.started)
-  }
-
-  private get started (): boolean {
-    return this.workerNodes.some(workerNode => workerNode.info.started)
-  }
-
   /** @inheritDoc */
   public get info (): PoolInfo {
     return {
@@ -251,6 +252,7 @@ export abstract class AbstractPool<
       worker: this.worker,
       minSize: this.minSize,
       maxSize: this.maxSize,
+      utilization: round(this.utilization),
       workerNodes: this.workerNodes.length,
       idleWorkerNodes: this.workerNodes.reduce(
         (accumulator, workerNode) =>
@@ -292,6 +294,35 @@ export abstract class AbstractPool<
     }
   }
 
+  /**
+   * Gets the pool run time.
+   *
+   * @returns The pool run time in milliseconds.
+   */
+  private get runTime (): number {
+    return performance.now() - this.startTimestamp
+  }
+
+  /**
+   * Gets the approximate pool utilization.
+   *
+   * @returns The pool utilization.
+   */
+  private get utilization (): number {
+    const poolRunTimeCapacity = this.runTime * this.maxSize
+    const totalTasksRunTime = this.workerNodes.reduce(
+      (accumulator, workerNode) =>
+        accumulator + workerNode.usage.runTime.aggregate,
+      0
+    )
+    const totalTasksWaitTime = this.workerNodes.reduce(
+      (accumulator, workerNode) =>
+        accumulator + workerNode.usage.waitTime.aggregate,
+      0
+    )
+    return (totalTasksRunTime + totalTasksWaitTime) / poolRunTimeCapacity
+  }
+
   /**
    * Pool type.
    *
@@ -731,7 +762,7 @@ export abstract class AbstractPool<
       if (this.emitter != null) {
         this.emitter.emit(PoolEvents.error, error)
       }
-      if (this.opts.restartWorkerOnError === true && !this.starting) {
+      if (this.opts.restartWorkerOnError === true) {
         this.createAndSetupWorker()
       }
     })
@@ -784,9 +815,15 @@ export abstract class AbstractPool<
     return message => {
       if (message.workerId != null && message.started != null) {
         // Worker started message received
-        this.workerNodes[
-          this.getWorkerNodeKey(this.getWorkerById(message.workerId) as Worker)
-        ].info.started = message.started
+        const worker = this.getWorkerById(message.workerId)
+        if (worker != null) {
+          this.workerNodes[this.getWorkerNodeKey(worker)].info.started =
+            message.started
+        } else {
+          throw new Error(
+            `Worker started message received from unknown worker '${message.workerId}'`
+          )
+        }
       } else if (message.id != null) {
         // Task execution response received
         const promiseResponse = this.promiseResponseMap.get(message.id)
@@ -850,7 +887,7 @@ export abstract class AbstractPool<
   private pushWorkerNode (worker: Worker): number {
     this.workerNodes.push({
       worker,
-      info: { id: worker.threadId ?? worker.id, started: false },
+      info: { id: this.getWorkerId(worker), started: true },
       usage: this.getWorkerUsage(),
       tasksQueue: new Queue<Task<Data>>()
     })
@@ -862,6 +899,20 @@ export abstract class AbstractPool<
     return this.workerNodes.length
   }
 
+  /**
+   * Gets the worker id.
+   *
+   * @param worker - The worker.
+   * @returns The worker id.
+   */
+  private getWorkerId (worker: Worker): number | undefined {
+    if (this.worker === WorkerTypes.thread) {
+      return worker.threadId
+    } else if (this.worker === WorkerTypes.cluster) {
+      return worker.id
+    }
+  }
+
   // /**
   //  * Sets the given worker in the pool worker nodes.
   //  *