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