X-Git-Url: https://git.piment-noir.org/?a=blobdiff_plain;f=src%2Fpools%2Fabstract-pool.ts;h=ad657b17cffd21b373473444a802fba34c6cefb8;hb=f751010550f07200428ca708afb2ec1d242cf7a7;hp=0069fb7fadc61c95196982365adfaf60561069fa;hpb=c05f0d5094264db0ff50636e794201e60ba7f448;p=poolifier.git

diff --git a/src/pools/abstract-pool.ts b/src/pools/abstract-pool.ts
index 0069fb7f..ad657b17 100644
--- a/src/pools/abstract-pool.ts
+++ b/src/pools/abstract-pool.ts
@@ -38,6 +38,7 @@ import {
   type WorkerChoiceStrategyOptions
 } from './selection-strategies/selection-strategies-types'
 import { WorkerChoiceStrategyContext } from './selection-strategies/worker-choice-strategy-context'
+import { version } from './version'
 
 /**
  * Base class that implements some shared logic for all poolifier pools.
@@ -249,14 +250,15 @@ export abstract class AbstractPool<
   /** @inheritDoc */
   public get info (): PoolInfo {
     return {
+      version,
       type: this.type,
       worker: this.worker,
       minSize: this.minSize,
       maxSize: this.maxSize,
       ...(this.workerChoiceStrategyContext.getTaskStatisticsRequirements()
         .runTime.aggregate &&
-        this.workerChoiceStrategyContext.getTaskStatisticsRequirements().runTime
-          .aggregate && { utilization: round(this.utilization) }),
+        this.workerChoiceStrategyContext.getTaskStatisticsRequirements()
+          .waitTime.aggregate && { utilization: round(this.utilization) }),
       workerNodes: this.workerNodes.length,
       idleWorkerNodes: this.workerNodes.reduce(
         (accumulator, workerNode) =>
@@ -298,22 +300,14 @@ 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 poolRunTimeCapacity =
+      (performance.now() - this.startTimestamp) * this.maxSize
     const totalTasksRunTime = this.workerNodes.reduce(
       (accumulator, workerNode) =>
         accumulator + workerNode.usage.runTime.aggregate,
@@ -505,7 +499,13 @@ export abstract class AbstractPool<
       this.workerNodes.map(async (workerNode, workerNodeKey) => {
         this.flushTasksQueue(workerNodeKey)
         // FIXME: wait for tasks to be finished
+        const workerExitPromise = new Promise<void>(resolve => {
+          workerNode.worker.on('exit', () => {
+            resolve()
+          })
+        })
         await this.destroyWorker(workerNode.worker)
+        await workerExitPromise
       })
     )
   }
@@ -585,7 +585,16 @@ export abstract class AbstractPool<
       this.workerChoiceStrategyContext.getTaskStatisticsRequirements().runTime
         .aggregate
     ) {
-      workerUsage.runTime.aggregate += message.taskPerformance?.runTime ?? 0
+      const taskRunTime = message.taskPerformance?.runTime ?? 0
+      workerUsage.runTime.aggregate += taskRunTime
+      workerUsage.runTime.minimum = Math.min(
+        taskRunTime,
+        workerUsage.runTime.minimum
+      )
+      workerUsage.runTime.maximum = Math.max(
+        taskRunTime,
+        workerUsage.runTime.maximum
+      )
       if (
         this.workerChoiceStrategyContext.getTaskStatisticsRequirements().runTime
           .average &&
@@ -616,7 +625,15 @@ export abstract class AbstractPool<
       this.workerChoiceStrategyContext.getTaskStatisticsRequirements().waitTime
         .aggregate
     ) {
-      workerUsage.waitTime.aggregate += taskWaitTime ?? 0
+      workerUsage.waitTime.aggregate += taskWaitTime
+      workerUsage.waitTime.minimum = Math.min(
+        taskWaitTime,
+        workerUsage.waitTime.minimum
+      )
+      workerUsage.waitTime.maximum = Math.max(
+        taskWaitTime,
+        workerUsage.waitTime.maximum
+      )
       if (
         this.workerChoiceStrategyContext.getTaskStatisticsRequirements()
           .waitTime.average &&
@@ -645,17 +662,33 @@ export abstract class AbstractPool<
       this.workerChoiceStrategyContext.getTaskStatisticsRequirements().elu
         .aggregate
     ) {
-      if (workerUsage.elu != null && message.taskPerformance?.elu != null) {
+      if (message.taskPerformance?.elu != null) {
         workerUsage.elu.idle.aggregate += message.taskPerformance.elu.idle
         workerUsage.elu.active.aggregate += message.taskPerformance.elu.active
-        workerUsage.elu.utilization =
-          (workerUsage.elu.utilization +
-            message.taskPerformance.elu.utilization) /
-          2
-      } else if (message.taskPerformance?.elu != null) {
-        workerUsage.elu.idle.aggregate = message.taskPerformance.elu.idle
-        workerUsage.elu.active.aggregate = message.taskPerformance.elu.active
-        workerUsage.elu.utilization = message.taskPerformance.elu.utilization
+        if (workerUsage.elu.utilization != null) {
+          workerUsage.elu.utilization =
+            (workerUsage.elu.utilization +
+              message.taskPerformance.elu.utilization) /
+            2
+        } else {
+          workerUsage.elu.utilization = message.taskPerformance.elu.utilization
+        }
+        workerUsage.elu.idle.minimum = Math.min(
+          message.taskPerformance.elu.idle,
+          workerUsage.elu.idle.minimum
+        )
+        workerUsage.elu.idle.maximum = Math.max(
+          message.taskPerformance.elu.idle,
+          workerUsage.elu.idle.maximum
+        )
+        workerUsage.elu.active.minimum = Math.min(
+          message.taskPerformance.elu.active,
+          workerUsage.elu.active.minimum
+        )
+        workerUsage.elu.active.maximum = Math.max(
+          message.taskPerformance.elu.active,
+          workerUsage.elu.active.maximum
+        )
       }
       if (
         this.workerChoiceStrategyContext.getTaskStatisticsRequirements().elu
@@ -766,6 +799,33 @@ export abstract class AbstractPool<
       if (this.emitter != null) {
         this.emitter.emit(PoolEvents.error, error)
       }
+      if (this.opts.enableTasksQueue === true) {
+        const workerNodeKey = this.getWorkerNodeKey(worker)
+        while (this.tasksQueueSize(workerNodeKey) > 0) {
+          let targetWorkerNodeKey: number = workerNodeKey
+          let minQueuedTasks = Infinity
+          for (const [workerNodeId, workerNode] of this.workerNodes.entries()) {
+            if (
+              workerNodeId !== workerNodeKey &&
+              workerNode.usage.tasks.queued === 0
+            ) {
+              targetWorkerNodeKey = workerNodeId
+              break
+            }
+            if (
+              workerNodeId !== workerNodeKey &&
+              workerNode.usage.tasks.queued < minQueuedTasks
+            ) {
+              minQueuedTasks = workerNode.usage.tasks.queued
+              targetWorkerNodeKey = workerNodeId
+            }
+          }
+          this.enqueueTask(
+            targetWorkerNodeKey,
+            this.dequeueTask(workerNodeKey) as Task<Data>
+          )
+        }
+      }
       if (this.opts.restartWorkerOnError === true) {
         this.createAndSetupWorker()
       }
@@ -987,13 +1047,11 @@ export abstract class AbstractPool<
   }
 
   private flushTasksQueue (workerNodeKey: number): void {
-    if (this.tasksQueueSize(workerNodeKey) > 0) {
-      for (let i = 0; i < this.tasksQueueSize(workerNodeKey); i++) {
-        this.executeTask(
-          workerNodeKey,
-          this.dequeueTask(workerNodeKey) as Task<Data>
-        )
-      }
+    while (this.tasksQueueSize(workerNodeKey) > 0) {
+      this.executeTask(
+        workerNodeKey,
+        this.dequeueTask(workerNodeKey) as Task<Data>
+      )
     }
     this.workerNodes[workerNodeKey].tasksQueue.clear()
   }
@@ -1037,12 +1095,16 @@ export abstract class AbstractPool<
       },
       runTime: {
         aggregate: 0,
+        maximum: 0,
+        minimum: 0,
         average: 0,
         median: 0,
         history: new CircularArray()
       },
       waitTime: {
         aggregate: 0,
+        maximum: 0,
+        minimum: 0,
         average: 0,
         median: 0,
         history: new CircularArray()
@@ -1050,17 +1112,20 @@ export abstract class AbstractPool<
       elu: {
         idle: {
           aggregate: 0,
+          maximum: 0,
+          minimum: 0,
           average: 0,
           median: 0,
           history: new CircularArray()
         },
         active: {
           aggregate: 0,
+          maximum: 0,
+          minimum: 0,
           average: 0,
           median: 0,
           history: new CircularArray()
-        },
-        utilization: 0
+        }
       }
     }
   }