src/worker/WorkerSet.ts

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