perf: alternate worker selection between start and end of worker nodes

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

diff --git a/src/pools/abstract-pool.ts b/src/pools/abstract-pool.ts
index 9290c2185e05f62137e7df7784f881988d02b022..67a22ffe04898f590ab70669a9d49934d6f90ad6 100644 (file)
--- a/src/pools/abstract-pool.ts
+++ b/src/pools/abstract-pool.ts
@@ -69,7 +69,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 +84,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()
 
@@ -251,21 +251,23 @@ export abstract class AbstractPool<
   }
 
   /** @inheritDoc */
-  public enableTasksQueue (enable: boolean, opts?: TasksQueueOptions): void {
+  public enableTasksQueue (
+    enable: boolean,
+    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(opts as TasksQueueOptions)
+    this.setTasksQueueOptions(tasksQueueOptions as TasksQueueOptions)
   }
 
   /** @inheritDoc */
-  public setTasksQueueOptions (opts: TasksQueueOptions): void {
+  public setTasksQueueOptions (tasksQueueOptions: TasksQueueOptions): void {
     if (this.opts.enableTasksQueue === true) {
-      this.checkValidTasksQueueOptions(opts)
-      this.opts.tasksQueueOptions = this.buildTasksQueueOptions(opts)
+      this.checkValidTasksQueueOptions(tasksQueueOptions)
+      this.opts.tasksQueueOptions =
+        this.buildTasksQueueOptions(tasksQueueOptions)
     } else {
       delete this.opts.tasksQueueOptions
     }
@@ -305,7 +307,21 @@ export abstract class AbstractPool<
   }
 
   /** @inheritDoc */
-  public async execute (data: Data): Promise<Response> {
+  public findLastFreeWorkerNodeKey (): number {
+    // It requires node >= 18.0.0
+    // return this.workerNodes.findLastIndex(workerNode => {
+    //   return workerNode.tasksUsage?.running === 0
+    // })
+    for (let i = this.workerNodes.length - 1; i >= 0; i--) {
+      if (this.workerNodes[i].tasksUsage?.running === 0) {
+        return i
+      }
+    }
+    return -1
+  }
+
+  /** @inheritDoc */
+  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
@@ -558,7 +574,7 @@ export abstract class AbstractPool<
    * Gets the given worker its tasks usage in the pool.
    *
    * @param worker - The worker.
-   * @throws {@link Error} if the worker is not found in the pool worker nodes.
+   * @throws Error if the worker is not found in the pool worker nodes.
    * @returns The worker tasks usage.
    */
   private getWorkerTasksUsage (worker: Worker): TasksUsage | undefined {
@@ -652,4 +668,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)
+    }
+  }
 }