refactor: cleanup queue API

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

diff --git a/src/pools/abstract-pool.ts b/src/pools/abstract-pool.ts
index 3bc00a6fffd36c987dd015fdeecd6b62f1e6a662..06cdc3cd5d635c1e13b209228631bea7e6960fd6 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.
@@ -69,7 +70,7 @@ export abstract class AbstractPool<
    * Constructs a new poolifier pool.
    *
    * @param numberOfWorkers - Number of workers that this pool should manage.
-   * @param filePath - Path to the worker-file.
+   * @param filePath - Path to the worker file.
    * @param opts - Options for the pool.
    */
   public constructor (
@@ -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
     )
   }
@@ -256,9 +257,7 @@ export abstract class AbstractPool<
     tasksQueueOptions?: TasksQueueOptions
   ): void {
     if (this.opts.enableTasksQueue === true && !enable) {
-      for (const [workerNodeKey] of this.workerNodes.entries()) {
-        this.flushTasksQueue(workerNodeKey)
-      }
+      this.flushTasksQueues()
     }
     this.opts.enableTasksQueue = enable
     this.setTasksQueueOptions(tasksQueueOptions as TasksQueueOptions)
@@ -298,18 +297,15 @@ 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): Promise<Response> {
     const [workerNodeKey, workerNode] = this.chooseWorkerNode()
     const submittedTask: Task<Data> = {
       // eslint-disable-next-line @typescript-eslint/consistent-type-assertions
@@ -392,7 +388,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) {
@@ -433,7 +429,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)
@@ -565,7 +561,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
@@ -591,7 +587,7 @@ export abstract class AbstractPool<
         medRunTime: 0,
         error: 0
       },
-      tasksQueue: []
+      tasksQueue: new Queue<Task<Data>>()
     })
   }
 
@@ -607,7 +603,7 @@ export abstract class AbstractPool<
     workerNodeKey: number,
     worker: Worker,
     tasksUsage: TasksUsage,
-    tasksQueue: Array<Task<Data>>
+    tasksQueue: Queue<Task<Data>>
   ): void {
     this.workerNodes[workerNodeKey] = {
       worker,
@@ -633,21 +629,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>
+        )
       }
     }
   }
@@ -656,4 +655,10 @@ export abstract class AbstractPool<
     const workerNodeKey = this.getWorkerNodeKey(worker)
     this.flushTasksQueue(workerNodeKey)
   }
+
+  private flushTasksQueues (): void {
+    for (const [workerNodeKey] of this.workerNodes.entries()) {
+      this.flushTasksQueue(workerNodeKey)
+    }
+  }
 }