Fixes to worker selection strategies

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

diff --git a/src/pools/abstract-pool.ts b/src/pools/abstract-pool.ts
index 0136f048cdb761ad6bc8186197634b635d55c3ba..40e06b158fe3af59ec49527a6cc170822520657f 100644 (file)
--- a/src/pools/abstract-pool.ts
+++ b/src/pools/abstract-pool.ts
@@ -4,10 +4,10 @@ import type {
 } from '../utility-types'
 import { EMPTY_FUNCTION } from '../utils'
 import { isKillBehavior, KillBehaviors } from '../worker/worker-options'
-import type { AbstractPoolWorker } from './abstract-pool-worker'
 import type { PoolOptions } from './pool'
 import type { IPoolInternal, TasksUsage } from './pool-internal'
 import { PoolEmitter, PoolType } from './pool-internal'
+import type { IPoolWorker } from './pool-worker'
 import {
   WorkerChoiceStrategies,
   WorkerChoiceStrategy
@@ -18,27 +18,22 @@ const WORKER_NOT_FOUND_TASKS_USAGE_MAP =
   'Worker could not be found in worker tasks usage map'
 
 /**
- * Base class containing some shared logic for all poolifier pools.
+ * Base class that implements some shared logic for all poolifier pools.
  *
  * @template Worker Type of worker which manages this pool.
  * @template Data Type of data sent to the worker. This can only be serializable data.
  * @template Response Type of response of execution. This can only be serializable data.
  */
 export abstract class AbstractPool<
-  Worker extends AbstractPoolWorker,
+  Worker extends IPoolWorker,
   Data = unknown,
   Response = unknown
 > implements IPoolInternal<Worker, Data, Response> {
   /** @inheritDoc */
   public readonly workers: Worker[] = []
 
-  /**
-   * The workers tasks usage map.
-   *
-   *  `key`: The `Worker`
-   *  `value`: Worker tasks usage statistics.
-   */
-  protected workersTasksUsage: Map<Worker, TasksUsage> = new Map<
+  /** @inheritDoc */
+  public readonly workersTasksUsage: Map<Worker, TasksUsage> = new Map<
     Worker,
     TasksUsage
   >()
@@ -135,7 +130,7 @@ export abstract class AbstractPool<
       throw new Error(
         'Cannot instantiate a pool without specifying the number of workers'
       )
-    } else if (!Number.isSafeInteger(numberOfWorkers)) {
+    } else if (Number.isSafeInteger(numberOfWorkers) === false) {
       throw new Error(
         'Cannot instantiate a pool with a non integer number of workers'
       )
@@ -182,6 +177,9 @@ export abstract class AbstractPool<
     workerChoiceStrategy: WorkerChoiceStrategy
   ): void {
     this.opts.workerChoiceStrategy = workerChoiceStrategy
+    for (const worker of this.workers) {
+      this.resetWorkerTasksUsage(worker)
+    }
     this.workerChoiceStrategyContext.setWorkerChoiceStrategy(
       workerChoiceStrategy
     )
@@ -274,7 +272,7 @@ export abstract class AbstractPool<
   /**
    * Removes the given worker from the pool.
    *
-   * @param worker Worker that will be removed.
+   * @param worker The worker that will be removed.
    */
   protected removeWorker (worker: Worker): void {
     // Clean worker from data structure
@@ -305,10 +303,10 @@ export abstract class AbstractPool<
   ): void
 
   /**
-   * Register a listener callback on a given worker.
+   * Registers a listener callback on a given worker.
    *
-   * @param worker A worker.
-   * @param listener A message listener callback.
+   * @param worker The worker which should register a listener.
+   * @param listener The message listener callback.
    */
   protected abstract registerWorkerMessageListener<
     Message extends Data | Response
@@ -355,12 +353,7 @@ export abstract class AbstractPool<
     this.workers.push(worker)
 
     // Init worker tasks usage map
-    this.workersTasksUsage.set(worker, {
-      run: 0,
-      running: 0,
-      runTime: 0,
-      avgRunTime: 0
-    })
+    this.initWorkerTasksUsage(worker)
 
     this.afterWorkerSetup(worker)
 
@@ -443,10 +436,10 @@ export abstract class AbstractPool<
   }
 
   /**
-   * Updates tasks run time for the given worker.
+   * Updates tasks runtime for the given worker.
    *
    * @param worker Worker which run the task.
-   * @param taskRunTime Worker task run time.
+   * @param taskRunTime Worker task runtime.
    */
   private updateWorkerTasksRunTime (
     worker: Worker,
@@ -469,6 +462,20 @@ export abstract class AbstractPool<
     }
   }
 
+  /**
+   * Initializes tasks usage statistics.
+   *
+   * @param worker The worker.
+   */
+  initWorkerTasksUsage (worker: Worker): void {
+    this.workersTasksUsage.set(worker, {
+      run: 0,
+      running: 0,
+      runTime: 0,
+      avgRunTime: 0
+    })
+  }
+
   /**
    * Removes worker tasks usage statistics.
    *
@@ -477,4 +484,14 @@ export abstract class AbstractPool<
   private removeWorkerTasksUsage (worker: Worker): void {
     this.workersTasksUsage.delete(worker)
   }
+
+  /**
+   * Resets worker tasks usage statistics.
+   *
+   * @param worker The worker.
+   */
+  private resetWorkerTasksUsage (worker: Worker): void {
+    this.removeWorkerTasksUsage(worker)
+    this.initWorkerTasksUsage(worker)
+  }
 }