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 workerScript
  18    * @param maxElementsPerWorker
  19    * @param workerStartDelay
  20    * @param messageListenerCallback
  21    */
  22   constructor(workerScript: string, maxElementsPerWorker = 1, workerStartDelay?: number, messageListenerCallback: (message: any) => void = () => { /* This is intentional */ }) {
  23     super(workerScript, workerStartDelay, messageListenerCallback);
  24     this.workerSet = new Set<WorkerSetElement>();
  25     this.maxElementsPerWorker = maxElementsPerWorker;
  26   }
  27
  28   get size(): number {
  29     return this.workerSet.size;
  30   }
  31
  32   /**
  33    *
  34    * @param elementData
  35    * @returns
  36    * @public
  37    */
  38   public async addElement(elementData: T): Promise<void> {
  39     if (!this.workerSet) {
  40       throw new Error('Cannot add a WorkerSet element: workers\' set does not exist');
  41     }
  42     if (this.getLastWorkerSetElement().numberOfWorkerElements >= this.maxElementsPerWorker) {
  43       this.startWorker();
  44       // Start worker sequentially to optimize memory at startup
  45       await Utils.sleep(this.workerStartDelay);
  46     }
  47     this.getLastWorker().postMessage({ id: WorkerEvents.START_WORKER_ELEMENT, data: elementData });
  48     this.getLastWorkerSetElement().numberOfWorkerElements++;
  49   }
  50
  51   /**
  52    *
  53    * @returns
  54    * @public
  55    */
  56   public async start(): Promise<void> {
  57     this.startWorker();
  58     // Start worker sequentially to optimize memory at startup
  59     await Utils.sleep(this.workerStartDelay);
  60   }
  61
  62   /**
  63    *
  64    * @returns
  65    * @public
  66    */
  67   public async stop(): Promise<void> {
  68     for (const workerSetElement of this.workerSet) {
  69       await workerSetElement.worker.terminate();
  70     }
  71     this.workerSet.clear();
  72   }
  73
  74   /**
  75    *
  76    * @private
  77    */
  78   private startWorker(): void {
  79     const worker = new Worker(this.workerScript);
  80     worker.on('message', this.messageListener);
  81     worker.on('error', () => { /* This is intentional */ });
  82     worker.on('exit', (code) => {
  83       WorkerUtils.defaultExitHandler(code);
  84       this.workerSet.delete(this.getWorkerSetElementByWorker(worker));
  85     });
  86     this.workerSet.add({ worker, numberOfWorkerElements: 0 });
  87   }
  88
  89   private getLastWorkerSetElement(): WorkerSetElement {
  90     let workerSetElement: WorkerSetElement;
  91     // eslint-disable-next-line no-empty
  92     for (workerSetElement of this.workerSet) { /* This is intentional */ }
  93     return workerSetElement;
  94   }
  95
  96   private getLastWorker(): Worker {
  97     return this.getLastWorkerSetElement().worker;
  98   }
  99
 100   private getWorkerSetElementByWorker(worker: Worker): WorkerSetElement {
 101     let workerSetElt: WorkerSetElement;
 102     for (const workerSetElement of this.workerSet) {
 103       if (workerSetElement.worker.threadId === worker.threadId) {
 104         workerSetElt = workerSetElement;
 105         break;
 106       }
 107     }
 108     return workerSetElt;
 109   }
 110 }