X-Git-Url: https://git.piment-noir.org/?a=blobdiff_plain;f=src%2Fpools%2Fabstract-pool.ts;h=283b04d682b4a656b6a430a78714a9ae52582e72;hb=f294681c4244effd6671e9d3534bf54bbf122d7a;hp=6ebb30595d822c525ef98e226bf3239a1d4c475f;hpb=df593701c4bd494b0e99372fdcc3708412799942;p=poolifier.git

diff --git a/src/pools/abstract-pool.ts b/src/pools/abstract-pool.ts
index 6ebb3059..283b04d6 100644
--- a/src/pools/abstract-pool.ts
+++ b/src/pools/abstract-pool.ts
@@ -4,10 +4,12 @@ import type { MessageValue, PromiseResponseWrapper } from '../utility-types'
 import {
   DEFAULT_WORKER_CHOICE_STRATEGY_OPTIONS,
   EMPTY_FUNCTION,
+  isKillBehavior,
   isPlainObject,
-  median
+  median,
+  round
 } from '../utils'
-import { KillBehaviors, isKillBehavior } from '../worker/worker-options'
+import { KillBehaviors } from '../worker/worker-options'
 import { CircularArray } from '../circular-array'
 import { Queue } from '../queue'
 import {
@@ -19,9 +21,16 @@ import {
   type PoolType,
   PoolTypes,
   type TasksQueueOptions,
-  type WorkerType
+  type WorkerType,
+  WorkerTypes
 } from './pool'
-import type { IWorker, Task, WorkerNode, WorkerUsage } from './worker'
+import type {
+  IWorker,
+  MessageHandler,
+  Task,
+  WorkerNode,
+  WorkerUsage
+} from './worker'
 import {
   Measurements,
   WorkerChoiceStrategies,
@@ -29,13 +38,14 @@ 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.
  *
  * @typeParam Worker - Type of worker which manages this pool.
- * @typeParam Data - Type of data sent to the worker. This can only be serializable data.
- * @typeParam Response - Type of execution response. This can only be serializable data.
+ * @typeParam Data - Type of data sent to the worker. This can only be structured-cloneable data.
+ * @typeParam Response - Type of execution response. This can only be structured-cloneable data.
  */
 export abstract class AbstractPool<
   Worker extends IWorker,
@@ -70,6 +80,11 @@ export abstract class AbstractPool<
   Response
   >
 
+  /**
+   * The start timestamp of the pool.
+   */
+  private readonly startTimestamp
+
   /**
    * Constructs a new poolifier pool.
    *
@@ -109,9 +124,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 {
@@ -233,53 +250,77 @@ 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()
+          .waitTime.aggregate && { utilization: round(this.utilization) }),
       workerNodes: this.workerNodes.length,
       idleWorkerNodes: this.workerNodes.reduce(
         (accumulator, workerNode) =>
-          workerNode.workerUsage.tasks.executing === 0
+          workerNode.usage.tasks.executing === 0
             ? accumulator + 1
             : accumulator,
         0
       ),
       busyWorkerNodes: this.workerNodes.reduce(
         (accumulator, workerNode) =>
-          workerNode.workerUsage.tasks.executing > 0
-            ? accumulator + 1
-            : accumulator,
+          workerNode.usage.tasks.executing > 0 ? accumulator + 1 : accumulator,
         0
       ),
       executedTasks: this.workerNodes.reduce(
         (accumulator, workerNode) =>
-          accumulator + workerNode.workerUsage.tasks.executed,
+          accumulator + workerNode.usage.tasks.executed,
         0
       ),
       executingTasks: this.workerNodes.reduce(
         (accumulator, workerNode) =>
-          accumulator + workerNode.workerUsage.tasks.executing,
+          accumulator + workerNode.usage.tasks.executing,
         0
       ),
       queuedTasks: this.workerNodes.reduce(
         (accumulator, workerNode) =>
-          accumulator + workerNode.workerUsage.tasks.queued,
+          accumulator + workerNode.usage.tasks.queued,
         0
       ),
       maxQueuedTasks: this.workerNodes.reduce(
         (accumulator, workerNode) =>
-          accumulator + workerNode.workerUsage.tasks.maxQueued,
+          accumulator + workerNode.usage.tasks.maxQueued,
         0
       ),
       failedTasks: this.workerNodes.reduce(
         (accumulator, workerNode) =>
-          accumulator + workerNode.workerUsage.tasks.failed,
+          accumulator + workerNode.usage.tasks.failed,
         0
       )
     }
   }
 
+  /**
+   * Gets the approximate pool utilization.
+   *
+   * @returns The pool utilization.
+   */
+  private get utilization (): number {
+    const poolRunTimeCapacity =
+      (performance.now() - this.startTimestamp) * 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.
    *
@@ -302,11 +343,22 @@ export abstract class AbstractPool<
    */
   protected abstract get maxSize (): number
 
+  /**
+   * Get the worker given its id.
+   *
+   * @param workerId - The worker id.
+   * @returns The worker if found in the pool worker nodes, `undefined` otherwise.
+   */
+  private getWorkerById (workerId: number): Worker | undefined {
+    return this.workerNodes.find(workerNode => workerNode.info.id === workerId)
+      ?.worker
+  }
+
   /**
    * Gets the given worker its worker node key.
    *
    * @param worker - The worker.
-   * @returns The worker node key if the worker is found in the pool worker nodes, `-1` otherwise.
+   * @returns The worker node key if found in the pool worker nodes, `-1` otherwise.
    */
   private getWorkerNodeKey (worker: Worker): number {
     return this.workerNodes.findIndex(
@@ -402,7 +454,7 @@ export abstract class AbstractPool<
   protected internalBusy (): boolean {
     return (
       this.workerNodes.findIndex(workerNode => {
-        return workerNode.workerUsage.tasks.executing === 0
+        return workerNode.usage.tasks.executing === 0
       }) === -1
     )
   }
@@ -428,7 +480,7 @@ export abstract class AbstractPool<
     if (
       this.opts.enableTasksQueue === true &&
       (this.busy ||
-        this.workerNodes[workerNodeKey].workerUsage.tasks.executing >=
+        this.workerNodes[workerNodeKey].usage.tasks.executing >=
           ((this.opts.tasksQueueOptions as TasksQueueOptions)
             .concurrency as number))
     ) {
@@ -447,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
       })
     )
   }
@@ -460,8 +518,8 @@ export abstract class AbstractPool<
   protected abstract destroyWorker (worker: Worker): void | Promise<void>
 
   /**
-   * Setup hook to execute code before worker node are created in the abstract constructor.
-   * Can be overridden
+   * Setup hook to execute code before worker nodes are created in the abstract constructor.
+   * Can be overridden.
    *
    * @virtual
    */
@@ -485,7 +543,7 @@ export abstract class AbstractPool<
     workerNodeKey: number,
     task: Task<Data>
   ): void {
-    const workerUsage = this.workerNodes[workerNodeKey].workerUsage
+    const workerUsage = this.workerNodes[workerNodeKey].usage
     ++workerUsage.tasks.executing
     this.updateWaitTimeWorkerUsage(workerUsage, task)
   }
@@ -501,8 +559,7 @@ export abstract class AbstractPool<
     worker: Worker,
     message: MessageValue<Response>
   ): void {
-    const workerUsage =
-      this.workerNodes[this.getWorkerNodeKey(worker)].workerUsage
+    const workerUsage = this.workerNodes[this.getWorkerNodeKey(worker)].usage
     this.updateTaskStatisticsWorkerUsage(workerUsage, message)
     this.updateRunTimeWorkerUsage(workerUsage, message)
     this.updateEluWorkerUsage(workerUsage, message)
@@ -670,9 +727,12 @@ export abstract class AbstractPool<
    * @param worker - The worker which should register a listener.
    * @param listener - The message listener callback.
    */
-  protected abstract registerWorkerMessageListener<
-    Message extends Data | Response
-  >(worker: Worker, listener: (message: MessageValue<Message>) => void): void
+  private registerWorkerMessageListener<Message extends Data | Response>(
+    worker: Worker,
+    listener: (message: MessageValue<Message>) => void
+  ): void {
+    worker.on('message', listener as MessageHandler<Worker>)
+  }
 
   /**
    * Creates a new worker.
@@ -683,12 +743,14 @@ export abstract class AbstractPool<
 
   /**
    * Function that can be hooked up when a worker has been newly created and moved to the pool worker nodes.
-   *
-   * Can be used to update the `maxListeners` or binding the `main-worker`\<-\>`worker` connection if not bind by default.
+   * Can be overridden.
    *
    * @param worker - The newly created worker.
    */
-  protected abstract afterWorkerSetup (worker: Worker): void
+  protected afterWorkerSetup (worker: Worker): void {
+    // Listen to worker messages.
+    this.registerWorkerMessageListener(worker, this.workerListener())
+  }
 
   /**
    * Creates a new worker and sets it up completely in the pool worker nodes.
@@ -704,6 +766,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()
       }
@@ -736,11 +825,9 @@ export abstract class AbstractPool<
         isKillBehavior(KillBehaviors.HARD, message.kill) ||
         (message.kill != null &&
           ((this.opts.enableTasksQueue === false &&
-            this.workerNodes[workerNodeKey].workerUsage.tasks.executing ===
-              0) ||
+            this.workerNodes[workerNodeKey].usage.tasks.executing === 0) ||
             (this.opts.enableTasksQueue === true &&
-              this.workerNodes[workerNodeKey].workerUsage.tasks.executing ===
-                0 &&
+              this.workerNodes[workerNodeKey].usage.tasks.executing === 0 &&
               this.tasksQueueSize(workerNodeKey) === 0)))
       ) {
         // Kill message received from the worker: no new tasks are submitted to that worker for a while ( > maxInactiveTime)
@@ -757,33 +844,55 @@ export abstract class AbstractPool<
    */
   protected workerListener (): (message: MessageValue<Response>) => void {
     return message => {
-      if (message.id != null) {
+      if (message.workerId != null && message.started != null) {
+        // Worker started message received
+        this.handleWorkerStartedMessage(message)
+      } else if (message.id != null) {
         // Task execution response received
-        const promiseResponse = this.promiseResponseMap.get(message.id)
-        if (promiseResponse != null) {
-          if (message.taskError != null) {
-            if (this.emitter != null) {
-              this.emitter.emit(PoolEvents.taskError, message.taskError)
-            }
-            promiseResponse.reject(message.taskError.message)
-          } else {
-            promiseResponse.resolve(message.data as Response)
-          }
-          this.afterTaskExecutionHook(promiseResponse.worker, message)
-          this.promiseResponseMap.delete(message.id)
-          const workerNodeKey = this.getWorkerNodeKey(promiseResponse.worker)
-          if (
-            this.opts.enableTasksQueue === true &&
-            this.tasksQueueSize(workerNodeKey) > 0
-          ) {
-            this.executeTask(
-              workerNodeKey,
-              this.dequeueTask(workerNodeKey) as Task<Data>
-            )
-          }
-          this.workerChoiceStrategyContext.update(workerNodeKey)
+        this.handleTaskExecutionResponse(message)
+      }
+    }
+  }
+
+  private handleWorkerStartedMessage (message: MessageValue<Response>): void {
+    // Worker started message received
+    const worker = this.getWorkerById(message.workerId as number)
+    if (worker != null) {
+      this.workerNodes[this.getWorkerNodeKey(worker)].info.started =
+        message.started as boolean
+    } else {
+      throw new Error(
+        `Worker started message received from unknown worker '${
+          message.workerId as number
+        }'`
+      )
+    }
+  }
+
+  private handleTaskExecutionResponse (message: MessageValue<Response>): void {
+    const promiseResponse = this.promiseResponseMap.get(message.id as string)
+    if (promiseResponse != null) {
+      if (message.taskError != null) {
+        if (this.emitter != null) {
+          this.emitter.emit(PoolEvents.taskError, message.taskError)
         }
+        promiseResponse.reject(message.taskError.message)
+      } else {
+        promiseResponse.resolve(message.data as Response)
+      }
+      this.afterTaskExecutionHook(promiseResponse.worker, message)
+      this.promiseResponseMap.delete(message.id as string)
+      const workerNodeKey = this.getWorkerNodeKey(promiseResponse.worker)
+      if (
+        this.opts.enableTasksQueue === true &&
+        this.tasksQueueSize(workerNodeKey) > 0
+      ) {
+        this.executeTask(
+          workerNodeKey,
+          this.dequeueTask(workerNodeKey) as Task<Data>
+        )
       }
+      this.workerChoiceStrategyContext.update(workerNodeKey)
     }
   }
 
@@ -808,7 +917,7 @@ export abstract class AbstractPool<
     workerNode: WorkerNode<Worker, Data>,
     workerUsage: WorkerUsage
   ): void {
-    workerNode.workerUsage = workerUsage
+    workerNode.usage = workerUsage
   }
 
   /**
@@ -820,7 +929,8 @@ export abstract class AbstractPool<
   private pushWorkerNode (worker: Worker): number {
     this.workerNodes.push({
       worker,
-      workerUsage: this.getWorkerUsage(),
+      info: { id: this.getWorkerId(worker), started: true },
+      usage: this.getWorkerUsage(),
       tasksQueue: new Queue<Task<Data>>()
     })
     const workerNodeKey = this.getWorkerNodeKey(worker)
@@ -831,23 +941,40 @@ 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.
   //  *
   //  * @param workerNodeKey - The worker node key.
   //  * @param worker - The worker.
+  //  * @param workerInfo - The worker info.
   //  * @param workerUsage - The worker usage.
   //  * @param tasksQueue - The worker task queue.
   //  */
   // private setWorkerNode (
   //   workerNodeKey: number,
   //   worker: Worker,
+  //   workerInfo: WorkerInfo,
   //   workerUsage: WorkerUsage,
   //   tasksQueue: Queue<Task<Data>>
   // ): void {
   //   this.workerNodes[workerNodeKey] = {
   //     worker,
-  //     workerUsage,
+  //     info: workerInfo,
+  //     usage: workerUsage,
   //     tasksQueue
   //   }
   // }
@@ -887,13 +1014,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()
   }