Bump typedoc from 0.20.28 to 0.20.29 (#252)

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

diff --git a/src/pools/abstract-pool.ts b/src/pools/abstract-pool.ts
index f1ac76ed587b8059c87430edcf1bddea18264577..0c4b59c92fef071bbddd487d1490ed5d2f8db38b 100644 (file)
--- a/src/pools/abstract-pool.ts
+++ b/src/pools/abstract-pool.ts
@@ -2,6 +2,7 @@ import type {
   MessageValue,
   PromiseWorkerResponseWrapper
 } from '../utility-types'
+import { isKillBehavior, KillBehaviors } from '../worker/worker-options'
 import type { IPoolInternal } from './pool-internal'
 import { PoolEmitter } from './pool-internal'
 import type { WorkerChoiceStrategy } from './selection-strategies'
@@ -82,15 +83,6 @@ export interface PoolOptions<Worker> {
    * A function that will listen for exit event on each worker.
    */
   exitHandler?: ExitHandler<Worker>
-  /**
-   * This is just to avoid non-useful warning messages.
-   *
-   * Will be used to set `maxListeners` on event emitters (workers are event emitters).
-   *
-   * @default 1000
-   * @see [Node events emitter.setMaxListeners(n)](https://nodejs.org/api/events.html#events_emitter_setmaxlisteners_n)
-   */
-  maxTasks?: number
   /**
    * The work choice strategy to use in this pool.
    */
@@ -109,6 +101,15 @@ export abstract class AbstractPool<
   Data = unknown,
   Response = unknown
 > implements IPoolInternal<Worker, Data, Response> {
+  /** @inheritdoc */
+  public readonly workers: Worker[] = []
+
+  /** @inheritdoc */
+  public readonly tasks: Map<Worker, number> = new Map<Worker, number>()
+
+  /** @inheritdoc */
+  public readonly emitter: PoolEmitter
+
   /**
    * The promise map.
    *
@@ -122,15 +123,6 @@ export abstract class AbstractPool<
     PromiseWorkerResponseWrapper<Worker, Response>
   > = new Map<number, PromiseWorkerResponseWrapper<Worker, Response>>()
 
-  /** @inheritdoc */
-  public readonly workers: Worker[] = []
-
-  /** @inheritdoc */
-  public readonly tasks: Map<Worker, number> = new Map<Worker, number>()
-
-  /** @inheritdoc */
-  public readonly emitter: PoolEmitter
-
   /**
    * ID of the next message.
    */
@@ -152,12 +144,12 @@ export abstract class AbstractPool<
    *
    * @param numberOfWorkers Number of workers that this pool should manage.
    * @param filePath Path to the worker-file.
-   * @param opts Options for the pool. Default: `{ maxTasks: 1000 }`
+   * @param opts Options for the pool.
    */
   public constructor (
     public readonly numberOfWorkers: number,
     public readonly filePath: string,
-    public readonly opts: PoolOptions<Worker> = { maxTasks: 1000 }
+    public readonly opts: PoolOptions<Worker>
   ) {
     if (!this.isMain()) {
       throw new Error('Cannot start a pool from a worker!')
@@ -173,6 +165,20 @@ export abstract class AbstractPool<
     this.emitter = new PoolEmitter()
     this.workerChoiceStrategyContext = new WorkerChoiceStrategyContext(
       this,
+      () => {
+        const workerCreated = this.createAndSetupWorker()
+        this.registerWorkerMessageListener(workerCreated, message => {
+          const tasksInProgress = this.tasks.get(workerCreated)
+          if (
+            isKillBehavior(KillBehaviors.HARD, message.kill) ||
+            tasksInProgress === 0
+          ) {
+            // Kill received from the worker, means that no new tasks are submitted to that worker for a while ( > maxInactiveTime)
+            void this.destroyWorker(workerCreated)
+          }
+        })
+        return workerCreated
+      },
       opts.workerChoiceStrategy ?? WorkerChoiceStrategies.ROUND_ROBIN
     )
   }
@@ -196,13 +202,13 @@ export abstract class AbstractPool<
       throw new Error(
         'Cannot instantiate a pool with a negative number of workers'
       )
-    } else if (!this.isDynamic() && numberOfWorkers === 0) {
+    } else if (!this.dynamic && numberOfWorkers === 0) {
       throw new Error('Cannot instantiate a fixed pool with no worker')
     }
   }
 
   /** @inheritdoc */
-  public isDynamic (): boolean {
+  public get dynamic (): boolean {
     return false
   }
 
@@ -210,6 +216,7 @@ export abstract class AbstractPool<
   public setWorkerChoiceStrategy (
     workerChoiceStrategy: WorkerChoiceStrategy
   ): void {
+    this.opts.workerChoiceStrategy = workerChoiceStrategy
     this.workerChoiceStrategyContext.setWorkerChoiceStrategy(
       workerChoiceStrategy
     )
@@ -231,8 +238,12 @@ export abstract class AbstractPool<
     await Promise.all(this.workers.map(worker => this.destroyWorker(worker)))
   }
 
-  /** @inheritdoc */
-  public abstract destroyWorker (worker: Worker): void | Promise<void>
+  /**
+   * Shut down given worker.
+   *
+   * @param worker A worker within `workers`.
+   */
+  protected abstract destroyWorker (worker: Worker): void | Promise<void>
 
   /**
    * Setup hook that can be overridden by a Poolifier pool implementation
@@ -314,8 +325,13 @@ export abstract class AbstractPool<
     message: MessageValue<Data>
   ): void
 
-  /** @inheritdoc */
-  public abstract registerWorkerMessageListener<
+  /**
+   * Register a listener callback on a given worker.
+   *
+   * @param worker A worker.
+   * @param listener A message listener callback.
+   */
+  protected abstract registerWorkerMessageListener<
     Message extends Data | Response
   > (worker: Worker, listener: (message: MessageValue<Message>) => void): void
 
@@ -342,8 +358,12 @@ export abstract class AbstractPool<
    */
   protected abstract afterWorkerSetup (worker: Worker): void
 
-  /** @inheritdoc */
-  public createAndSetupWorker (): Worker {
+  /**
+   * Creates a new worker for this pool and sets it up completely.
+   *
+   * @returns New, completely set up worker.
+   */
+  protected createAndSetupWorker (): Worker {
     const worker: Worker = this.createWorker()
 
     worker.on('error', this.opts.errorHandler ?? EMPTY_FUNCTION)
@@ -367,7 +387,7 @@ export abstract class AbstractPool<
    * @returns The listener function to execute when a message is sent from a worker.
    */
   protected workerListener (): (message: MessageValue<Response>) => void {
-    const listener: (message: MessageValue<Response>) => void = message => {
+    return message => {
       if (message.id) {
         const value = this.promiseMap.get(message.id)
         if (value) {
@@ -378,6 +398,5 @@ export abstract class AbstractPool<
         }
       }
     }
-    return listener
   }
 }