src/worker/WorkerPool.ts

   1 import Configuration from '../utils/Configuration';
   2 import Constants from '../utils/Constants';
   3 import Pool from 'worker-threads-pool';
   4 import Utils from '../utils/Utils';
   5 import { WorkerData } from '../types/Worker';
   6 import Wrk from './Wrk';
   7
   8 export default class WorkerPool extends Wrk {
   9   private pool: Pool;
  10
  11   /**
  12    * Create a new `WorkerPool`.
  13    *
  14    * @param {string} workerScript
  15    */
  16   constructor(workerScript: string) {
  17     super(workerScript);
  18     this.pool = UniquePool.getInstance();
  19   }
  20
  21   get size(): number {
  22     return this.pool.size;
  23   }
  24
  25   get maxElementsPerWorker(): number {
  26     return 1;
  27   }
  28
  29   /**
  30    *
  31    * @return {Promise<void>}
  32    * @public
  33    */
  34   // eslint-disable-next-line @typescript-eslint/no-empty-function
  35   public async start(): Promise<void> { }
  36
  37   /**
  38    *
  39    * @return {Promise<void>}
  40    * @public
  41    */
  42   public async addElement(elementData: WorkerData): Promise<void> {
  43     return new Promise((resolve, reject) => {
  44       this.pool.acquire(this.workerScript, { workerData: elementData }, (err, worker) => {
  45         if (err) {
  46           return reject(err);
  47         }
  48         worker.once('message', resolve);
  49         worker.once('error', reject);
  50       });
  51       // Start worker sequentially to optimize memory at startup
  52       void Utils.sleep(Constants.START_WORKER_DELAY);
  53     });
  54   }
  55 }
  56
  57 class UniquePool {
  58   private static instance: Pool;
  59
  60   private constructor() { }
  61
  62   public static getInstance(): Pool {
  63     if (!UniquePool.instance) {
  64       UniquePool.instance = new Pool({ max: Configuration.getWorkerPoolMaxSize() });
  65     }
  66     return UniquePool.instance;
  67   }
  68 }