src/charging-station/Worker.ts

   1 import { Worker, WorkerOptions } from 'worker_threads';
   2
   3 import Configuration from '../utils/Configuration';
   4 import Constants from '../utils/Constants';
   5 import Pool from 'worker-threads-pool';
   6 import WorkerData from '../types/WorkerData';
   7
   8 export default class Wrk {
   9   private _workerScript: string;
  10   private _workerData: WorkerData;
  11   private _index: number;
  12   private _concurrentWorkers: number;
  13   private _worker: Worker;
  14
  15   /**
  16    * Create a new `Wrk`.
  17    *
  18    * @param {string} workerScript
  19    * @param {WorkerData} workerData
  20    * @param {number} numConcurrentWorkers
  21    */
  22   constructor(workerScript: string, workerData: WorkerData, numConcurrentWorkers: number) {
  23     this._workerData = workerData;
  24     this._index = workerData.index;
  25     this._workerScript = workerScript;
  26     if (Configuration.useWorkerPool()) {
  27       this._concurrentWorkers = Configuration.getWorkerPoolSize();
  28       WorkerPool.concurrentWorkers = this._concurrentWorkers;
  29     } else {
  30       this._concurrentWorkers = numConcurrentWorkers;
  31     }
  32   }
  33
  34   /**
  35    * @return {number}
  36    * @public
  37    */
  38   public get concurrentWorkers(): number {
  39     return this._concurrentWorkers;
  40   }
  41
  42   /**
  43    *
  44    * @return {Promise}
  45    * @public
  46    */
  47   async start(): Promise<Worker> {
  48     if (Configuration.useWorkerPool()) {
  49       await this._startWorkerWithPool();
  50     } else {
  51       await this._startWorker();
  52     }
  53     return this._worker;
  54   }
  55
  56   /**
  57    *
  58    * @return {void}
  59    * @public
  60    */
  61   addChargingStation(workerData: WorkerData, numConcurrentWorkers: number): void {
  62     this._workerData = workerData;
  63     this._index = workerData.index;
  64     this._concurrentWorkers = numConcurrentWorkers;
  65     this._worker.postMessage({ id : Constants.START_CHARGING_STATION, workerData: workerData });
  66   }
  67
  68   /**
  69    *
  70    * @return {Promise}
  71    * @private
  72    */
  73   private async _startWorkerWithPool() {
  74     return new Promise((resolve, reject) => {
  75       WorkerPool.acquire(this._workerScript, { workerData: this._workerData }, (err, worker) => {
  76         if (err) {
  77           return reject(err);
  78         }
  79         worker.once('message', resolve);
  80         worker.once('error', reject);
  81         this._worker = worker;
  82       });
  83     });
  84   }
  85
  86   /**
  87    *
  88    * @return {Promise}
  89    * @private
  90    */
  91   private async _startWorker() {
  92     return new Promise((resolve, reject) => {
  93       const worker = new Worker(this._workerScript, { workerData: this._workerData });
  94       worker.on('message', resolve);
  95       worker.on('error', reject);
  96       worker.on('exit', (code) => {
  97         if (code !== 0) {
  98           reject(new Error(`Worker id ${this._index} stopped with exit code ${code}`));
  99         }
 100       });
 101       this._worker = worker;
 102     });
 103   }
 104 }
 105
 106 class WorkerPool {
 107   public static concurrentWorkers: number;
 108   private static _instance: Pool;
 109
 110   private constructor() { }
 111
 112   public static getInstance(): Pool {
 113     if (!WorkerPool._instance || (WorkerPool._instance?.size === WorkerPool.concurrentWorkers)) {
 114       WorkerPool._instance = new Pool({ max: WorkerPool.concurrentWorkers });
 115     }
 116     return WorkerPool._instance;
 117   }
 118
 119   public static acquire(filename: string, options: WorkerOptions, callback: (error: Error | null, worker: Worker) => void): void {
 120     WorkerPool.getInstance().acquire(filename, options, callback);
 121   }
 122 }