refactor: use task performance data structure in messages

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

diff --git a/src/pools/abstract-pool.ts b/src/pools/abstract-pool.ts
index 58fc28ce12d6a56416b6db05cec3e5186c549a14..85a47e38f7c248f7a92ff9e53166e2121048015c 100644 (file)
--- a/src/pools/abstract-pool.ts
+++ b/src/pools/abstract-pool.ts
@@ -1,4 +1,5 @@
 import crypto from 'node:crypto'
+import { performance } from 'node:perf_hooks'
 import type { MessageValue, PromiseResponseWrapper } from '../utility-types'
 import {
   DEFAULT_WORKER_CHOICE_STRATEGY_OPTIONS,
@@ -17,7 +18,8 @@ import {
   type PoolOptions,
   type PoolType,
   PoolTypes,
-  type TasksQueueOptions
+  type TasksQueueOptions,
+  type WorkerType
 } from './pool'
 import type { IWorker, Task, TasksUsage, WorkerNode } from './worker'
 import {
@@ -77,9 +79,9 @@ export abstract class AbstractPool<
    * @param opts - Options for the pool.
    */
   public constructor (
-    public readonly numberOfWorkers: number,
-    public readonly filePath: string,
-    public readonly opts: PoolOptions<Worker>
+    protected readonly numberOfWorkers: number,
+    protected readonly filePath: string,
+    protected readonly opts: PoolOptions<Worker>
   ) {
     if (!this.isMain()) {
       throw new Error('Cannot start a pool from a worker!')
@@ -93,12 +95,6 @@ export abstract class AbstractPool<
     this.enqueueTask = this.enqueueTask.bind(this)
     this.checkAndEmitEvents = this.checkAndEmitEvents.bind(this)
 
-    this.setupHook()
-
-    for (let i = 1; i <= this.numberOfWorkers; i++) {
-      this.createAndSetupWorker()
-    }
-
     if (this.opts.enableEvents === true) {
       this.emitter = new PoolEmitter()
     }
@@ -111,6 +107,12 @@ export abstract class AbstractPool<
       this.opts.workerChoiceStrategy,
       this.opts.workerChoiceStrategyOptions
     )
+
+    this.setupHook()
+
+    for (let i = 1; i <= this.numberOfWorkers; i++) {
+      this.createAndSetupWorker()
+    }
   }
 
   private checkFilePath (filePath: string): void {
@@ -210,13 +212,11 @@ export abstract class AbstractPool<
     }
   }
 
-  /** @inheritDoc */
-  public abstract get type (): PoolType
-
   /** @inheritDoc */
   public get info (): PoolInfo {
     return {
       type: this.type,
+      worker: this.worker,
       minSize: this.minSize,
       maxSize: this.maxSize,
       workerNodes: this.workerNodes.length,
@@ -247,6 +247,18 @@ export abstract class AbstractPool<
     }
   }
 
+  /**
+   * Pool type.
+   *
+   * If it is `'dynamic'`, it provides the `max` property.
+   */
+  protected abstract get type (): PoolType
+
+  /**
+   * Gets the worker type.
+   */
+  protected abstract get worker (): WorkerType
+
   /**
    * Pool minimum size.
    */
@@ -276,9 +288,15 @@ export abstract class AbstractPool<
   ): void {
     this.checkValidWorkerChoiceStrategy(workerChoiceStrategy)
     this.opts.workerChoiceStrategy = workerChoiceStrategy
+    this.workerChoiceStrategyContext.setWorkerChoiceStrategy(
+      this.opts.workerChoiceStrategy
+    )
+    if (workerChoiceStrategyOptions != null) {
+      this.setWorkerChoiceStrategyOptions(workerChoiceStrategyOptions)
+    }
     for (const workerNode of this.workerNodes) {
       this.setWorkerNodeTasksUsage(workerNode, {
-        run: 0,
+        ran: 0,
         running: 0,
         runTime: 0,
         runTimeHistory: new CircularArray(),
@@ -288,14 +306,10 @@ export abstract class AbstractPool<
         waitTimeHistory: new CircularArray(),
         avgWaitTime: 0,
         medWaitTime: 0,
-        error: 0
+        error: 0,
+        elu: undefined
       })
-    }
-    this.workerChoiceStrategyContext.setWorkerChoiceStrategy(
-      this.opts.workerChoiceStrategy
-    )
-    if (workerChoiceStrategyOptions != null) {
-      this.setWorkerChoiceStrategyOptions(workerChoiceStrategyOptions)
+      this.setWorkerStatistics(workerNode.worker)
     }
   }
 
@@ -328,7 +342,7 @@ export abstract class AbstractPool<
       this.checkValidTasksQueueOptions(tasksQueueOptions)
       this.opts.tasksQueueOptions =
         this.buildTasksQueueOptions(tasksQueueOptions)
-    } else {
+    } else if (this.opts.tasksQueueOptions != null) {
       delete this.opts.tasksQueueOptions
     }
   }
@@ -346,7 +360,9 @@ export abstract class AbstractPool<
    *
    * The pool filling boolean status.
    */
-  protected abstract get full (): boolean
+  protected get full (): boolean {
+    return this.workerNodes.length >= this.maxSize
+  }
 
   /**
    * Whether the pool is busy or not.
@@ -365,13 +381,13 @@ export abstract class AbstractPool<
 
   /** @inheritDoc */
   public async execute (data?: Data, name?: string): Promise<Response> {
-    const submissionTimestamp = performance.now()
+    const timestamp = performance.now()
     const workerNodeKey = this.chooseWorkerNode()
     const submittedTask: Task<Data> = {
       name,
       // eslint-disable-next-line @typescript-eslint/consistent-type-assertions
       data: data ?? ({} as Data),
-      submissionTimestamp,
+      timestamp,
       id: crypto.randomUUID()
     }
     const res = new Promise<Response>((resolve, reject) => {
@@ -403,6 +419,7 @@ export abstract class AbstractPool<
     await Promise.all(
       this.workerNodes.map(async (workerNode, workerNodeKey) => {
         this.flushTasksQueue(workerNodeKey)
+        // FIXME: wait for tasks to be finished
         await this.destroyWorker(workerNode.worker)
       })
     )
@@ -454,32 +471,33 @@ export abstract class AbstractPool<
     const workerTasksUsage =
       this.workerNodes[this.getWorkerNodeKey(worker)].tasksUsage
     --workerTasksUsage.running
-    ++workerTasksUsage.run
+    ++workerTasksUsage.ran
     if (message.error != null) {
       ++workerTasksUsage.error
     }
     this.updateRunTimeTasksUsage(workerTasksUsage, message)
     this.updateWaitTimeTasksUsage(workerTasksUsage, message)
+    this.updateEluTasksUsage(workerTasksUsage, message)
   }
 
   private updateRunTimeTasksUsage (
     workerTasksUsage: TasksUsage,
     message: MessageValue<Response>
   ): void {
-    if (this.workerChoiceStrategyContext.getRequiredStatistics().runTime) {
-      workerTasksUsage.runTime += message.runTime ?? 0
+    if (this.workerChoiceStrategyContext.getTaskStatistics().runTime) {
+      workerTasksUsage.runTime += message.taskPerformance?.runTime ?? 0
       if (
-        this.workerChoiceStrategyContext.getRequiredStatistics().avgRunTime &&
-        workerTasksUsage.run !== 0
+        this.workerChoiceStrategyContext.getTaskStatistics().avgRunTime &&
+        workerTasksUsage.ran !== 0
       ) {
         workerTasksUsage.avgRunTime =
-          workerTasksUsage.runTime / workerTasksUsage.run
+          workerTasksUsage.runTime / workerTasksUsage.ran
       }
       if (
-        this.workerChoiceStrategyContext.getRequiredStatistics().medRunTime &&
-        message.runTime != null
+        this.workerChoiceStrategyContext.getTaskStatistics().medRunTime &&
+        message.taskPerformance?.runTime != null
       ) {
-        workerTasksUsage.runTimeHistory.push(message.runTime)
+        workerTasksUsage.runTimeHistory.push(message.taskPerformance.runTime)
         workerTasksUsage.medRunTime = median(workerTasksUsage.runTimeHistory)
       }
     }
@@ -489,25 +507,49 @@ export abstract class AbstractPool<
     workerTasksUsage: TasksUsage,
     message: MessageValue<Response>
   ): void {
-    if (this.workerChoiceStrategyContext.getRequiredStatistics().waitTime) {
-      workerTasksUsage.waitTime += message.waitTime ?? 0
+    if (this.workerChoiceStrategyContext.getTaskStatistics().waitTime) {
+      workerTasksUsage.waitTime += message.taskPerformance?.waitTime ?? 0
       if (
-        this.workerChoiceStrategyContext.getRequiredStatistics().avgWaitTime &&
-        workerTasksUsage.run !== 0
+        this.workerChoiceStrategyContext.getTaskStatistics().avgWaitTime &&
+        workerTasksUsage.ran !== 0
       ) {
         workerTasksUsage.avgWaitTime =
-          workerTasksUsage.waitTime / workerTasksUsage.run
+          workerTasksUsage.waitTime / workerTasksUsage.ran
       }
       if (
-        this.workerChoiceStrategyContext.getRequiredStatistics().medWaitTime &&
-        message.waitTime != null
+        this.workerChoiceStrategyContext.getTaskStatistics().medWaitTime &&
+        message.taskPerformance?.waitTime != null
       ) {
-        workerTasksUsage.waitTimeHistory.push(message.waitTime)
+        workerTasksUsage.waitTimeHistory.push(message.taskPerformance.waitTime)
         workerTasksUsage.medWaitTime = median(workerTasksUsage.waitTimeHistory)
       }
     }
   }
 
+  private updateEluTasksUsage (
+    workerTasksUsage: TasksUsage,
+    message: MessageValue<Response>
+  ): void {
+    if (this.workerChoiceStrategyContext.getTaskStatistics().elu) {
+      if (
+        workerTasksUsage.elu != null &&
+        message.taskPerformance?.elu != null
+      ) {
+        workerTasksUsage.elu = {
+          idle: workerTasksUsage.elu.idle + message.taskPerformance.elu.idle,
+          active:
+            workerTasksUsage.elu.active + message.taskPerformance.elu.active,
+          utilization:
+            (workerTasksUsage.elu.utilization +
+              message.taskPerformance.elu.utilization) /
+            2
+        }
+      } else if (message.taskPerformance?.elu != null) {
+        workerTasksUsage.elu = message.taskPerformance.elu
+      }
+    }
+  }
+
   /**
    * Chooses a worker node for the next task.
    *
@@ -528,6 +570,7 @@ export abstract class AbstractPool<
         ) {
           // Kill message received from the worker: no new tasks are submitted to that worker for a while ( > maxInactiveTime)
           this.flushTasksQueue(currentWorkerNodeKey)
+          // FIXME: wait for tasks to be finished
           void (this.destroyWorker(workerCreated) as Promise<void>)
         }
       })
@@ -588,11 +631,11 @@ export abstract class AbstractPool<
         this.emitter.emit(PoolEvents.error, error)
       }
     })
-    if (this.opts.restartWorkerOnError === true) {
-      worker.on('error', () => {
+    worker.on('error', () => {
+      if (this.opts.restartWorkerOnError === true) {
         this.createAndSetupWorker()
-      })
-    }
+      }
+    })
     worker.on('online', this.opts.onlineHandler ?? EMPTY_FUNCTION)
     worker.on('exit', this.opts.exitHandler ?? EMPTY_FUNCTION)
     worker.once('exit', () => {
@@ -601,6 +644,8 @@ export abstract class AbstractPool<
 
     this.pushWorkerNode(worker)
 
+    this.setWorkerStatistics(worker)
+
     this.afterWorkerSetup(worker)
 
     return worker
@@ -679,7 +724,7 @@ export abstract class AbstractPool<
     return this.workerNodes.push({
       worker,
       tasksUsage: {
-        run: 0,
+        ran: 0,
         running: 0,
         runTime: 0,
         runTimeHistory: new CircularArray(),
@@ -689,7 +734,8 @@ export abstract class AbstractPool<
         waitTimeHistory: new CircularArray(),
         avgWaitTime: 0,
         medWaitTime: 0,
-        error: 0
+        error: 0,
+        elu: undefined
       },
       tasksQueue: new Queue<Task<Data>>()
     })
@@ -762,4 +808,14 @@ export abstract class AbstractPool<
       this.flushTasksQueue(workerNodeKey)
     }
   }
+
+  private setWorkerStatistics (worker: Worker): void {
+    this.sendToWorker(worker, {
+      statistics: {
+        runTime: this.workerChoiceStrategyContext.getTaskStatistics().runTime,
+        waitTime: this.workerChoiceStrategyContext.getTaskStatistics().waitTime,
+        elu: this.workerChoiceStrategyContext.getTaskStatistics().elu
+      }
+    })
+  }
 }