src/worker/WorkerSet.ts

   1 import { WorkerEvents, WorkerSetElement } from '../types/Worker';
   2
   3 import Utils from '../utils/Utils';
   4 import { Worker } from 'worker_threads';
   5 import WorkerAbstract from './WorkerAbstract';
   6
   7 export default class WorkerSet<T> extends WorkerAbstract {
   8   public maxElementsPerWorker: number;
   9   private workerSet: Set<WorkerSetElement>;
  10
  11   /**
  12    * Create a new `WorkerSet`.
  13    *
  14    * @param {string} workerScript
  15    * @param {number} maxElementsPerWorker
  16    * @param {number} workerStartDelay
  17    */
  18   constructor(workerScript: string, maxElementsPerWorker = 1, workerStartDelay?: number) {
  19     super(workerScript, workerStartDelay);
  20     this.workerSet = new Set<WorkerSetElement>();
  21     this.maxElementsPerWorker = maxElementsPerWorker;
  22   }
  23
  24   get size(): number {
  25     return this.workerSet.size;
  26   }
  27
  28   /**
  29    *
  30    * @param {T} elementData
  31    * @returns {Promise<void>}
  32    * @public
  33    */
  34   public async addElement(elementData: T): Promise<void> {
  35     if (!this.workerSet) {
  36       throw Error('Cannot add a WorkerSet element: workers\' set does not exist');
  37     }
  38     if (this.getLastWorkerSetElement().numberOfWorkerElements >= this.maxElementsPerWorker) {
  39       this.startWorker();
  40       // Start worker sequentially to optimize memory at startup
  41       await Utils.sleep(this.workerStartDelay);
  42     }
  43     this.getLastWorker().postMessage({ id: WorkerEvents.START_WORKER_ELEMENT, workerData: elementData });
  44     this.getLastWorkerSetElement().numberOfWorkerElements++;
  45   }
  46
  47   /**
  48    *
  49    * @returns {Promise<void>}
  50    * @public
  51    */
  52   public async start(): Promise<void> {
  53     this.startWorker();
  54     // Start worker sequentially to optimize memory at startup
  55     await Utils.sleep(this.workerStartDelay);
  56   }
  57
  58   /**
  59    *
  60    * @returns {Promise<void>}
  61    * @public
  62    */
  63   public async stop(): Promise<void> {
  64     for (const workerSetElement of this.workerSet) {
  65       await workerSetElement.worker.terminate();
  66     }
  67     this.workerSet.clear();
  68   }
  69
  70   /**
  71    *
  72    * @private
  73    */
  74   private startWorker(): void {
  75     const worker = new Worker(this.workerScript);
  76     worker.on('message', () => { });
  77     worker.on('error', () => { });
  78     worker.on('exit', (code) => {
  79       if (code !== 0) {
  80         console.error(`Worker stopped with exit code ${code}`);
  81       }
  82       this.workerSet.delete(this.getWorkerSetElementByWorker(worker));
  83     });
  84     this.workerSet.add({ worker, numberOfWorkerElements: 0 });
  85   }
  86
  87   private getLastWorkerSetElement(): WorkerSetElement {
  88     let workerSetElement: WorkerSetElement;
  89     // eslint-disable-next-line no-empty
  90     for (workerSetElement of this.workerSet) { }
  91     return workerSetElement;
  92   }
  93
  94   private getLastWorker(): Worker {
  95     return this.getLastWorkerSetElement().worker;
  96   }
  97
  98   private getWorkerSetElementByWorker(worker: Worker): WorkerSetElement {
  99     let workerSetElt: WorkerSetElement;
 100     this.workerSet.forEach((workerSetElement) => {
 101       if (workerSetElement.worker.threadId === worker.threadId) {
 102         workerSetElt = workerSetElement;
 103       }
 104     });
 105     return workerSetElt;
 106   }
 107 }