src/worker/WorkerSet.ts

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