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