src/worker/WorkerStaticPool.ts

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