src/worker/WorkerStaticPool.ts

   1 import { FixedThreadPool, 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 WorkerStaticPool<T> extends WorkerAbstract {
  10   private pool: StaticPool;
  11
  12   /**
  13    * Create a new `WorkerStaticPool`.
  14    *
  15    * @param {string} workerScript
  16    * @param {number} numberOfThreads
  17    * @param {number} startWorkerDelay
  18    * @param {PoolOptions} opts
  19    */
  20   constructor(workerScript: string, numberOfThreads: number, startWorkerDelay?: number, opts?: PoolOptions<Worker>) {
  21     super(workerScript, startWorkerDelay);
  22     this.pool = StaticPool.getInstance(numberOfThreads, 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   public async stop(): Promise<void> {
  47     return this.pool.destroy();
  48   }
  49
  50   /**
  51    *
  52    * @param {T} 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 StaticPool extends FixedThreadPool<WorkerData> {
  64   private static instance: StaticPool;
  65
  66   private constructor(numberOfThreads: number, workerScript: string, opts?: PoolOptions<Worker>) {
  67     super(numberOfThreads, workerScript, opts);
  68   }
  69
  70   public static getInstance(numberOfThreads: number, workerScript: string, opts?: PoolOptions<Worker>): StaticPool {
  71     if (!StaticPool.instance) {
  72       opts.exitHandler = opts?.exitHandler ?? WorkerUtils.defaultExitHandler;
  73       StaticPool.instance = new StaticPool(numberOfThreads, workerScript, opts);
  74     }
  75     return StaticPool.instance;
  76   }
  77 }