X-Git-Url: https://git.piment-noir.org/?a=blobdiff_plain;f=src%2Fpools%2Fabstract-pool.ts;h=55c380b80c5b92de72bb1c4c87e57f4614c56f37;hb=184855e69fea29f1018024a34be10de2c8e3141a;hp=724785f534b6a4c3b971d97aad8b7d7d598de666;hpb=a32959dad7ec3668f607e26153ad7610ff5438f9;p=poolifier.git

diff --git a/src/pools/abstract-pool.ts b/src/pools/abstract-pool.ts
index 724785f5..55c380b8 100644
--- a/src/pools/abstract-pool.ts
+++ b/src/pools/abstract-pool.ts
@@ -13,9 +13,12 @@ import {
   type IPool,
   PoolEmitter,
   PoolEvents,
+  type PoolInfo,
   type PoolOptions,
-  PoolType,
-  type TasksQueueOptions
+  type PoolType,
+  PoolTypes,
+  type TasksQueueOptions,
+  type WorkerType
 } from './pool'
 import type { IWorker, Task, TasksUsage, WorkerNode } from './worker'
 import {
@@ -133,7 +136,7 @@ export abstract class AbstractPool<
       throw new RangeError(
         'Cannot instantiate a pool with a negative number of workers'
       )
-    } else if (this.type === PoolType.FIXED && numberOfWorkers === 0) {
+    } else if (this.type === PoolTypes.fixed && numberOfWorkers === 0) {
       throw new Error('Cannot instantiate a fixed pool with no worker')
     }
   }
@@ -149,6 +152,7 @@ export abstract class AbstractPool<
       this.checkValidWorkerChoiceStrategyOptions(
         this.opts.workerChoiceStrategyOptions
       )
+      this.opts.restartWorkerOnError = opts.restartWorkerOnError ?? true
       this.opts.enableEvents = opts.enableEvents ?? true
       this.opts.enableTasksQueue = opts.enableTasksQueue ?? false
       if (this.opts.enableTasksQueue) {
@@ -184,7 +188,7 @@ export abstract class AbstractPool<
     }
     if (
       workerChoiceStrategyOptions.weights != null &&
-      Object.keys(workerChoiceStrategyOptions.weights).length !== this.size
+      Object.keys(workerChoiceStrategyOptions.weights).length !== this.maxSize
     ) {
       throw new Error(
         'Invalid worker choice strategy options: must have a weight for each worker node'
@@ -211,30 +215,54 @@ export abstract class AbstractPool<
   public abstract get type (): PoolType
 
   /** @inheritDoc */
-  public abstract get size (): number
+  public get info (): PoolInfo {
+    return {
+      type: this.type,
+      worker: this.worker,
+      minSize: this.minSize,
+      maxSize: this.maxSize,
+      workerNodes: this.workerNodes.length,
+      idleWorkerNodes: this.workerNodes.reduce(
+        (accumulator, workerNode) =>
+          workerNode.tasksUsage.running === 0 ? accumulator + 1 : accumulator,
+        0
+      ),
+      busyWorkerNodes: this.workerNodes.reduce(
+        (accumulator, workerNode) =>
+          workerNode.tasksUsage.running > 0 ? accumulator + 1 : accumulator,
+        0
+      ),
+      runningTasks: this.workerNodes.reduce(
+        (accumulator, workerNode) =>
+          accumulator + workerNode.tasksUsage.running,
+        0
+      ),
+      queuedTasks: this.workerNodes.reduce(
+        (accumulator, workerNode) => accumulator + workerNode.tasksQueue.size,
+        0
+      ),
+      maxQueuedTasks: this.workerNodes.reduce(
+        (accumulator, workerNode) =>
+          accumulator + workerNode.tasksQueue.maxSize,
+        0
+      )
+    }
+  }
 
   /**
-   * Number of tasks running in the pool.
+   * Gets the worker type.
    */
-  private get numberOfRunningTasks (): number {
-    return this.workerNodes.reduce(
-      (accumulator, workerNode) => accumulator + workerNode.tasksUsage.running,
-      0
-    )
-  }
+  protected abstract get worker (): WorkerType
 
   /**
-   * Number of tasks queued in the pool.
+   * Pool minimum size.
    */
-  private get numberOfQueuedTasks (): number {
-    if (this.opts.enableTasksQueue === false) {
-      return 0
-    }
-    return this.workerNodes.reduce(
-      (accumulator, workerNode) => accumulator + workerNode.tasksQueue.size,
-      0
-    )
-  }
+  protected abstract get minSize (): number
+
+  /**
+   * Pool maximum size.
+   */
+  protected abstract get maxSize (): number
 
   /**
    * Gets the given worker its worker node key.
@@ -438,7 +466,7 @@ export abstract class AbstractPool<
       ++workerTasksUsage.error
     }
     this.updateRunTimeTasksUsage(workerTasksUsage, message)
-    this.updateWaitTasksUsage(workerTasksUsage, message)
+    this.updateWaitTimeTasksUsage(workerTasksUsage, message)
   }
 
   private updateRunTimeTasksUsage (
@@ -464,7 +492,7 @@ export abstract class AbstractPool<
     }
   }
 
-  private updateWaitTasksUsage (
+  private updateWaitTimeTasksUsage (
     workerTasksUsage: TasksUsage,
     message: MessageValue<Response>
   ): void {
@@ -496,7 +524,7 @@ export abstract class AbstractPool<
    */
   protected chooseWorkerNode (): number {
     let workerNodeKey: number
-    if (this.type === PoolType.DYNAMIC && !this.full && this.internalBusy()) {
+    if (this.type === PoolTypes.dynamic && !this.full && this.internalBusy()) {
       const workerCreated = this.createAndSetupWorker()
       this.registerWorkerMessageListener(workerCreated, message => {
         const currentWorkerNodeKey = this.getWorkerNodeKey(workerCreated)
@@ -562,6 +590,16 @@ export abstract class AbstractPool<
 
     worker.on('message', this.opts.messageHandler ?? EMPTY_FUNCTION)
     worker.on('error', this.opts.errorHandler ?? EMPTY_FUNCTION)
+    worker.on('error', error => {
+      if (this.emitter != null) {
+        this.emitter.emit(PoolEvents.error, error)
+      }
+    })
+    if (this.opts.restartWorkerOnError === true) {
+      worker.on('error', () => {
+        this.createAndSetupWorker()
+      })
+    }
     worker.on('online', this.opts.onlineHandler ?? EMPTY_FUNCTION)
     worker.on('exit', this.opts.exitHandler ?? EMPTY_FUNCTION)
     worker.once('exit', () => {
@@ -588,6 +626,12 @@ export abstract class AbstractPool<
         if (promiseResponse != null) {
           if (message.error != null) {
             promiseResponse.reject(message.error)
+            if (this.emitter != null) {
+              this.emitter.emit(PoolEvents.taskError, {
+                error: message.error,
+                errorData: message.errorData
+              })
+            }
           } else {
             promiseResponse.resolve(message.data as Response)
           }
@@ -609,12 +653,12 @@ export abstract class AbstractPool<
   }
 
   private checkAndEmitEvents (): void {
-    if (this.opts.enableEvents === true) {
+    if (this.emitter != null) {
       if (this.busy) {
-        this.emitter?.emit(PoolEvents.busy)
+        this.emitter?.emit(PoolEvents.busy, this.info)
       }
-      if (this.type === PoolType.DYNAMIC && this.full) {
-        this.emitter?.emit(PoolEvents.full)
+      if (this.type === PoolTypes.dynamic && this.full) {
+        this.emitter?.emit(PoolEvents.full, this.info)
       }
     }
   }
@@ -686,8 +730,10 @@ export abstract class AbstractPool<
    */
   private removeWorkerNode (worker: Worker): void {
     const workerNodeKey = this.getWorkerNodeKey(worker)
-    this.workerNodes.splice(workerNodeKey, 1)
-    this.workerChoiceStrategyContext.remove(workerNodeKey)
+    if (workerNodeKey !== -1) {
+      this.workerNodes.splice(workerNodeKey, 1)
+      this.workerChoiceStrategyContext.remove(workerNodeKey)
+    }
   }
 
   private executeTask (workerNodeKey: number, task: Task<Data>): void {