Merge branch 'master' into multiple-functions

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

diff --git a/src/pools/abstract-pool.ts b/src/pools/abstract-pool.ts
index ffbda872af76d22fd7d037b39dd54025568e39c5..edf2b29a6e9de328133735cea8a6c58b18609db0 100644 (file)
--- a/src/pools/abstract-pool.ts
+++ b/src/pools/abstract-pool.ts
@@ -6,14 +6,16 @@ import {
   median
 } from '../utils'
 import { KillBehaviors, isKillBehavior } from '../worker/worker-options'
+import { CircularArray } from '../circular-array'
+import { Queue } from '../queue'
 import {
-  PoolEvents,
   type IPool,
+  PoolEmitter,
+  PoolEvents,
   type PoolOptions,
-  type TasksQueueOptions,
-  PoolType
+  PoolType,
+  type TasksQueueOptions
 } from './pool'
-import { PoolEmitter } from './pool'
 import type { IWorker, Task, TasksUsage, WorkerNode } from './worker'
 import {
   WorkerChoiceStrategies,
@@ -21,7 +23,6 @@ import {
   type WorkerChoiceStrategyOptions
 } from './selection-strategies/selection-strategies-types'
 import { WorkerChoiceStrategyContext } from './selection-strategies/worker-choice-strategy-context'
-import { CircularArray } from '../circular-array'
 
 /**
  * Base class that implements some shared logic for all poolifier pools.
@@ -84,10 +85,10 @@ export abstract class AbstractPool<
     this.checkFilePath(this.filePath)
     this.checkPoolOptions(this.opts)
 
-    this.chooseWorkerNode.bind(this)
-    this.executeTask.bind(this)
-    this.enqueueTask.bind(this)
-    this.checkAndEmitEvents.bind(this)
+    this.chooseWorkerNode = this.chooseWorkerNode.bind(this)
+    this.executeTask = this.executeTask.bind(this)
+    this.enqueueTask = this.enqueueTask.bind(this)
+    this.checkAndEmitEvents = this.checkAndEmitEvents.bind(this)
 
     this.setupHook()
 
@@ -197,7 +198,7 @@ export abstract class AbstractPool<
       return 0
     }
     return this.workerNodes.reduce(
-      (accumulator, workerNode) => accumulator + workerNode.tasksQueue.length,
+      (accumulator, workerNode) => accumulator + workerNode.tasksQueue.size,
       0
     )
   }
@@ -296,20 +297,18 @@ export abstract class AbstractPool<
   protected abstract get busy (): boolean
 
   protected internalBusy (): boolean {
-    return this.findFreeWorkerNodeKey() === -1
-  }
-
-  /** @inheritDoc */
-  public findFreeWorkerNodeKey (): number {
-    return this.workerNodes.findIndex(workerNode => {
-      return workerNode.tasksUsage?.running === 0
-    })
+    return (
+      this.workerNodes.findIndex(workerNode => {
+        return workerNode.tasksUsage?.running === 0
+      }) === -1
+    )
   }
 
   /** @inheritDoc */
-  public async execute (data?: Data): Promise<Response> {
+  public async execute (data?: Data, name?: string): Promise<Response> {
     const [workerNodeKey, workerNode] = this.chooseWorkerNode()
     const submittedTask: Task<Data> = {
+      name,
       // eslint-disable-next-line @typescript-eslint/consistent-type-assertions
       data: data ?? ({} as Data),
       id: crypto.randomUUID()
@@ -390,7 +389,7 @@ export abstract class AbstractPool<
     worker: Worker,
     message: MessageValue<Response>
   ): void {
-    const workerTasksUsage = this.getWorkerTasksUsage(worker) as TasksUsage
+    const workerTasksUsage = this.getWorkerTasksUsage(worker)
     --workerTasksUsage.running
     ++workerTasksUsage.run
     if (message.error != null) {
@@ -431,7 +430,7 @@ export abstract class AbstractPool<
         ) {
           // Kill message received from the worker: no new tasks are submitted to that worker for a while ( > maxInactiveTime)
           this.flushTasksQueueByWorker(workerCreated)
-          void this.destroyWorker(workerCreated)
+          void (this.destroyWorker(workerCreated) as Promise<void>)
         }
       })
       workerNodeKey = this.getWorkerNodeKey(workerCreated)
@@ -563,7 +562,7 @@ export abstract class AbstractPool<
    * @throws Error if the worker is not found in the pool worker nodes.
    * @returns The worker tasks usage.
    */
-  private getWorkerTasksUsage (worker: Worker): TasksUsage | undefined {
+  private getWorkerTasksUsage (worker: Worker): TasksUsage {
     const workerNodeKey = this.getWorkerNodeKey(worker)
     if (workerNodeKey !== -1) {
       return this.workerNodes[workerNodeKey].tasksUsage
@@ -589,7 +588,7 @@ export abstract class AbstractPool<
         medRunTime: 0,
         error: 0
       },
-      tasksQueue: []
+      tasksQueue: new Queue<Task<Data>>()
     })
   }
 
@@ -605,7 +604,7 @@ export abstract class AbstractPool<
     workerNodeKey: number,
     worker: Worker,
     tasksUsage: TasksUsage,
-    tasksQueue: Array<Task<Data>>
+    tasksQueue: Queue<Task<Data>>
   ): void {
     this.workerNodes[workerNodeKey] = {
       worker,
@@ -631,21 +630,24 @@ export abstract class AbstractPool<
   }
 
   private enqueueTask (workerNodeKey: number, task: Task<Data>): number {
-    return this.workerNodes[workerNodeKey].tasksQueue.push(task)
+    return this.workerNodes[workerNodeKey].tasksQueue.enqueue(task)
   }
 
   private dequeueTask (workerNodeKey: number): Task<Data> | undefined {
-    return this.workerNodes[workerNodeKey].tasksQueue.shift()
+    return this.workerNodes[workerNodeKey].tasksQueue.dequeue()
   }
 
   private tasksQueueSize (workerNodeKey: number): number {
-    return this.workerNodes[workerNodeKey].tasksQueue.length
+    return this.workerNodes[workerNodeKey].tasksQueue.size
   }
 
   private flushTasksQueue (workerNodeKey: number): void {
     if (this.tasksQueueSize(workerNodeKey) > 0) {
-      for (const task of this.workerNodes[workerNodeKey].tasksQueue) {
-        this.executeTask(workerNodeKey, task)
+      for (let i = 0; i < this.tasksQueueSize(workerNodeKey); i++) {
+        this.executeTask(
+          workerNodeKey,
+          this.dequeueTask(workerNodeKey) as Task<Data>
+        )
       }
     }
   }