src/worker/WorkerDynamicPool.ts

   1 import { DynamicThreadPool, PoolOptions } from 'poolifier';
   2
   3 import Utils from '../utils/Utils';
   4 import { Worker } from 'worker_threads';
   5 import WorkerAbstract from './WorkerAbstract';
   6 import { WorkerData } from '../types/Worker';
   7
   8 export default class WorkerDynamicPool<T> extends WorkerAbstract {
   9   private pool: DynamicPool;
  10
  11   /**
  12    * Create a new `WorkerDynamicPool`.
  13    *
  14    * @param {string} workerScript
  15    * @param {number} min
  16    * @param {number} max
  17    * @param {number} workerStartDelay
  18    * @param {PoolOptions} opts
  19    */
  20   constructor(workerScript: string, min: number, max: number, workerStartDelay?: number, opts?: PoolOptions<Worker>) {
  21     super(workerScript, workerStartDelay);
  22     this.pool = DynamicPool.getInstance(min, max, this.workerScript, opts);
  23   }
  24
  25   get size(): number {
  26     return this.pool.workers.length;
  27   }
  28
  29   get maxElementsPerWorker(): number | null {
  30     return null;
  31   }
  32
  33   /**
  34    *
  35    * @returns {Promise<void>}
  36    * @public
  37    */
  38   // eslint-disable-next-line @typescript-eslint/no-empty-function
  39   public async start(): Promise<void> { }
  40
  41   /**
  42    *
  43    * @returns {Promise<void>}
  44    * @public
  45    */
  46   // eslint-disable-next-line @typescript-eslint/require-await
  47   public async stop(): Promise<void> {
  48     return this.pool.destroy();
  49   }
  50
  51   /**
  52    *
  53    * @param {T} elementData
  54    * @returns {Promise<void>}
  55    * @public
  56    */
  57   public async addElement(elementData: T): Promise<void> {
  58     await this.pool.execute(elementData);
  59     // Start worker sequentially to optimize memory at startup
  60     await Utils.sleep(this.workerStartDelay);
  61   }
  62 }
  63
  64 class DynamicPool extends DynamicThreadPool<WorkerData> {
  65   private static instance: DynamicPool;
  66
  67   private constructor(min: number, max: number, workerScript: string, opts?: PoolOptions<Worker>) {
  68     super(min, max, workerScript, opts);
  69   }
  70
  71   public static getInstance(min: number, max: number, workerScript: string, opts?: PoolOptions<Worker>): DynamicPool {
  72     if (!DynamicPool.instance) {
  73       opts.exitHandler = opts?.exitHandler ?? ((code) => {
  74         if (code !== 0) {
  75           console.error(`Worker stopped with exit code ${code}`);
  76         }
  77       });
  78       DynamicPool.instance = new DynamicPool(min, max, workerScript, opts);
  79     }
  80     return DynamicPool.instance;
  81   }
  82 }