Merge pull request #38 from poolifier/fix-worker-node-readiness

fix: wait for worker node readiness

Author: jerome-benoit

CHANGELOG.md

@@ -8,6 +8,10 @@ and this project adheres to
 
 ## [Unreleased]
 
+### Fixed
+
+- Ensure worker choice strategies implementation wait for worker node readiness.
+
 ## [0.1.6] - 2023-12-18
 
 ### Fixed

docs/api.md

@@ -151,8 +151,6 @@ An object with these properties:
   object to use in this pool.\
   Properties:
 
-  - `retries` (optional) - The number of retries to perform if no worker is
-    eligible.
   - `measurement` (optional) - The measurement to use in worker choice
     strategies: `runTime`, `waitTime` <!-- or `elu`. -->
   - `runTime` (optional) - Use the tasks
@@ -170,9 +168,8 @@ An object with these properties:
   - `weights` (optional) - The worker weights to use in weighted round robin
     worker choice strategies: `{ 0: 200, 1: 300, ..., n: 100 }`.
 
-  Default:
-  `{ retries: 6, runTime: { median: false }, waitTime: { median: false } }`
-  <!-- `{ retries: 6, runTime: { median: false }, waitTime: { median: false }, elu: { median: false } }` -->
+  Default: `{ runTime: { median: false }, waitTime: { median: false } }`
+  <!-- `{ runTime: { median: false }, waitTime: { median: false }, elu: { median: false } }` -->
 
 - `startWorkers` (optional) - Start the minimum number of workers at pool
   initialization.\

src/pools/abstract-pool.ts

@@ -8,7 +8,6 @@ import type {
 import {
   average,
   DEFAULT_TASK_NAME,
-  DEFAULT_WORKER_CHOICE_STRATEGY_OPTIONS,
   EMPTY_FUNCTION,
   exponentialDelay,
   isKillBehavior,
@@ -78,11 +77,6 @@ export abstract class AbstractPool<
   /** @inheritDoc */
   public emitter?: EventEmitter
 
-  /**
-   * Dynamic pool maximum size property placeholder.
-   */
-  protected readonly max?: number
-
   /**
    * The task execution response promise map:
    * - `key`: The message id of each submitted task.
@@ -133,22 +127,25 @@ export abstract class AbstractPool<
   /**
    * Constructs a new poolifier pool.
    *
-   * @param numberOfWorkers - Number of workers that this pool should manage.
+   * @param minimumNumberOfWorkers - Minimum number of workers that this pool manages.
    * @param fileURL - URL to the worker file.
    * @param opts - Options for the pool.
+   * @param maximumNumberOfWorkers - Maximum number of workers that this pool manages.
    */
   public constructor(
-    protected readonly numberOfWorkers: number,
+    protected readonly minimumNumberOfWorkers: number,
     protected readonly fileURL: URL,
     protected readonly opts: PoolOptions<Data>,
+    protected readonly maximumNumberOfWorkers?: number,
   ) {
     if (!this.isMain()) {
       throw new Error(
         'Cannot start a pool from a worker with the same type as the pool',
       )
     }
+    this.checkPoolType()
     checkFileURL(this.fileURL)
-    this.checkNumberOfWorkers(this.numberOfWorkers)
+    this.checkMinimumNumberOfWorkers(this.minimumNumberOfWorkers)
     this.checkPoolOptions(this.opts)
 
     this.chooseWorkerNode = this.chooseWorkerNode.bind(this)
@@ -183,20 +180,28 @@ export abstract class AbstractPool<
     this.startTimestamp = performance.now()
   }
 
-  private checkNumberOfWorkers(numberOfWorkers: number): void {
-    if (numberOfWorkers == null) {
+  private checkPoolType(): void {
+    if (this.type === PoolTypes.fixed && this.maximumNumberOfWorkers != null) {
+      throw new Error(
+        'Cannot instantiate a fixed pool with a maximum number of workers specified at initialization',
+      )
+    }
+  }
+
+  private checkMinimumNumberOfWorkers(minimumNumberOfWorkers: number): void {
+    if (minimumNumberOfWorkers == null) {
       throw new Error(
         'Cannot instantiate a pool without specifying the number of workers',
       )
-    } else if (!Number.isSafeInteger(numberOfWorkers)) {
+    } else if (!Number.isSafeInteger(minimumNumberOfWorkers)) {
       throw new TypeError(
         'Cannot instantiate a pool with a non safe integer number of workers',
       )
-    } else if (numberOfWorkers < 0) {
+    } else if (minimumNumberOfWorkers < 0) {
       throw new RangeError(
         'Cannot instantiate a pool with a negative number of workers',
       )
-    } else if (this.type === PoolTypes.fixed && numberOfWorkers === 0) {
+    } else if (this.type === PoolTypes.fixed && minimumNumberOfWorkers === 0) {
       throw new RangeError('Cannot instantiate a fixed pool with zero worker')
     }
   }
@@ -212,9 +217,8 @@ export abstract class AbstractPool<
       this.checkValidWorkerChoiceStrategyOptions(
         opts.workerChoiceStrategyOptions as WorkerChoiceStrategyOptions,
       )
-      this.opts.workerChoiceStrategyOptions = {
-        ...DEFAULT_WORKER_CHOICE_STRATEGY_OPTIONS,
-        ...opts.workerChoiceStrategyOptions,
+      if (opts.workerChoiceStrategyOptions != null) {
+        this.opts.workerChoiceStrategyOptions = opts.workerChoiceStrategyOptions
       }
       this.opts.restartWorkerOnError = opts.restartWorkerOnError ?? true
       this.opts.enableEvents = opts.enableEvents ?? true
@@ -243,25 +247,10 @@ export abstract class AbstractPool<
         'Invalid worker choice strategy options: must be a plain object',
       )
     }
-    if (
-      workerChoiceStrategyOptions?.retries != null &&
-      !Number.isSafeInteger(workerChoiceStrategyOptions.retries)
-    ) {
-      throw new TypeError(
-        'Invalid worker choice strategy options: retries must be an integer',
-      )
-    }
-    if (
-      workerChoiceStrategyOptions?.retries != null &&
-      workerChoiceStrategyOptions.retries < 0
-    ) {
-      throw new RangeError(
-        `Invalid worker choice strategy options: retries '${workerChoiceStrategyOptions.retries}' must be greater or equal than zero`,
-      )
-    }
     if (
       workerChoiceStrategyOptions?.weights != null &&
-      Object.keys(workerChoiceStrategyOptions.weights).length !== this.maxSize
+      Object.keys(workerChoiceStrategyOptions.weights).length !==
+        (this.maximumNumberOfWorkers ?? this.minimumNumberOfWorkers)
     ) {
       throw new Error(
         'Invalid worker choice strategy options: must have a weight for each worker node',
@@ -292,11 +281,11 @@ export abstract class AbstractPool<
       started: this.started,
       ready: this.ready,
       strategy: this.opts.workerChoiceStrategy as WorkerChoiceStrategy,
-      minSize: this.minSize,
-      maxSize: this.maxSize,
-      ...(this.workerChoiceStrategyContext.getTaskStatisticsRequirements()
+      minSize: this.minimumNumberOfWorkers,
+      maxSize: (this.maximumNumberOfWorkers ?? this.minimumNumberOfWorkers),
+      ...(this.workerChoiceStrategyContext?.getTaskStatisticsRequirements()
         .runTime.aggregate &&
-        this.workerChoiceStrategyContext.getTaskStatisticsRequirements()
+        this.workerChoiceStrategyContext?.getTaskStatisticsRequirements()
           .waitTime.aggregate &&
         { utilization: round(this.utilization) }),
       workerNodes: this.workerNodes.length,
@@ -351,7 +340,7 @@ export abstract class AbstractPool<
           accumulator + workerNode.usage.tasks.failed,
         0,
       ),
-      ...(this.workerChoiceStrategyContext.getTaskStatisticsRequirements()
+      ...(this.workerChoiceStrategyContext?.getTaskStatisticsRequirements()
         .runTime.aggregate && {
         runTime: {
           minimum: round(
@@ -368,7 +357,7 @@ export abstract class AbstractPool<
               ),
             ),
           ),
-          ...(this.workerChoiceStrategyContext.getTaskStatisticsRequirements()
+          ...(this.workerChoiceStrategyContext?.getTaskStatisticsRequirements()
             .runTime.average && {
             average: round(
               average(
@@ -380,7 +369,7 @@ export abstract class AbstractPool<
               ),
             ),
           }),
-          ...(this.workerChoiceStrategyContext.getTaskStatisticsRequirements()
+          ...(this.workerChoiceStrategyContext?.getTaskStatisticsRequirements()
             .runTime.median && {
             median: round(
               median(
@@ -394,7 +383,7 @@ export abstract class AbstractPool<
           }),
         },
       }),
-      ...(this.workerChoiceStrategyContext.getTaskStatisticsRequirements()
+      ...(this.workerChoiceStrategyContext?.getTaskStatisticsRequirements()
         .waitTime.aggregate && {
         waitTime: {
           minimum: round(
@@ -411,7 +400,7 @@ export abstract class AbstractPool<
               ),
             ),
           ),
-          ...(this.workerChoiceStrategyContext.getTaskStatisticsRequirements()
+          ...(this.workerChoiceStrategyContext?.getTaskStatisticsRequirements()
             .waitTime.average && {
             average: round(
               average(
@@ -423,7 +412,7 @@ export abstract class AbstractPool<
               ),
             ),
           }),
-          ...(this.workerChoiceStrategyContext.getTaskStatisticsRequirements()
+          ...(this.workerChoiceStrategyContext?.getTaskStatisticsRequirements()
             .waitTime.median && {
             median: round(
               median(
@@ -451,7 +440,7 @@ export abstract class AbstractPool<
             ? accumulator + 1
             : accumulator,
         0,
-      ) >= this.minSize
+      ) >= this.minimumNumberOfWorkers
     )
   }
 
@@ -462,7 +451,7 @@ export abstract class AbstractPool<
    */
   private get utilization(): number {
     const poolTimeCapacity = (performance.now() - this.startTimestamp) *
-      this.maxSize
+      (this.maximumNumberOfWorkers ?? this.minimumNumberOfWorkers)
     const totalTasksRunTime = this.workerNodes.reduce(
       (accumulator, workerNode) =>
         accumulator + (workerNode.usage.runTime?.aggregate ?? 0),
@@ -488,20 +477,6 @@ export abstract class AbstractPool<
    */
   protected abstract get worker(): WorkerType
 
-  /**
-   * The pool minimum size.
-   */
-  protected get minSize(): number {
-    return this.numberOfWorkers
-  }
-
-  /**
-   * The pool maximum size.
-   */
-  protected get maxSize(): number {
-    return this.max ?? this.numberOfWorkers
-  }
-
   /**
    * Checks if the worker id sent in the received message from a worker is valid.
    *
@@ -559,11 +534,11 @@ export abstract class AbstractPool<
     workerChoiceStrategyOptions: WorkerChoiceStrategyOptions,
   ): void {
     this.checkValidWorkerChoiceStrategyOptions(workerChoiceStrategyOptions)
-    this.opts.workerChoiceStrategyOptions = {
-      ...DEFAULT_WORKER_CHOICE_STRATEGY_OPTIONS,
-      ...workerChoiceStrategyOptions,
+    if (workerChoiceStrategyOptions != null) {
+      this.opts.workerChoiceStrategyOptions = workerChoiceStrategyOptions
     }
     this.workerChoiceStrategyContext.setOptions(
+      this,
       this.opts.workerChoiceStrategyOptions as WorkerChoiceStrategyOptions,
     )
   }
@@ -611,7 +586,9 @@ export abstract class AbstractPool<
     tasksQueueOptions: TasksQueueOptions,
   ): TasksQueueOptions {
     return {
-      ...getDefaultTasksQueueOptions(this.maxSize),
+      ...getDefaultTasksQueueOptions(
+        this.maximumNumberOfWorkers ?? this.minimumNumberOfWorkers,
+      ),
       ...tasksQueueOptions,
     }
   }
@@ -664,7 +641,8 @@ export abstract class AbstractPool<
    * The pool filling boolean status.
    */
   protected get full(): boolean {
-    return this.workerNodes.length >= this.maxSize
+    return this.workerNodes.length >=
+      (this.maximumNumberOfWorkers ?? this.minimumNumberOfWorkers)
   }
 
   /**
@@ -961,7 +939,7 @@ export abstract class AbstractPool<
         (accumulator, workerNode) =>
           !workerNode.info.dynamic ? accumulator + 1 : accumulator,
         0,
-      ) < this.numberOfWorkers
+      ) < this.minimumNumberOfWorkers
     ) {
       this.createAndSetupWorkerNode()
     }
@@ -993,9 +971,7 @@ export abstract class AbstractPool<
     this.started = false
   }
 
-  protected async sendKillMessageToWorker(
-    workerNodeKey: number,
-  ): Promise<void> {
+  private async sendKillMessageToWorker(workerNodeKey: number): Promise<void> {
     await new Promise<void>((resolve, reject) => {
       if (this.workerNodes?.[workerNodeKey] == null) {
         // FIXME: should reject with an error
@@ -1034,7 +1010,9 @@ export abstract class AbstractPool<
       'taskFinished',
       flushedTasks,
       this.opts.tasksQueueOptions?.tasksFinishedTimeout ??
-        getDefaultTasksQueueOptions(this.maxSize).tasksFinishedTimeout,
+        getDefaultTasksQueueOptions(
+          this.maximumNumberOfWorkers ?? this.minimumNumberOfWorkers,
+        ).tasksFinishedTimeout,
     )
     await this.sendKillMessageToWorker(workerNodeKey)
     workerNode.terminate()
@@ -1764,7 +1742,9 @@ export abstract class AbstractPool<
       {
         workerOptions: this.opts.workerOptions,
         tasksQueueBackPressureSize: this.opts.tasksQueueOptions?.size ??
-          getDefaultTasksQueueOptions(this.maxSize).size,
+          getDefaultTasksQueueOptions(
+            this.maximumNumberOfWorkers ?? this.minimumNumberOfWorkers,
+          ).size,
       },
     )
     // Flag the worker node as ready at pool startup.