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 _worker: Worker;
  12   private _maxWorkerElements: number;
  13   private _numWorkerElements: number;
  14
  15   /**
  16    * Create a new `Wrk`.
  17    *
  18    * @param {string} workerScript
  19    * @param {WorkerData} workerData
  20    */
  21   constructor(workerScript: string, workerData: WorkerData, maxWorkerElements = 1) {
  22     this._workerData = workerData;
  23     this._workerScript = workerScript;
  24     this._maxWorkerElements = maxWorkerElements;
  25     this._numWorkerElements = 0;
  26     if (Configuration.useWorkerPool()) {
  27       WorkerPool.maxConcurrentWorkers = Configuration.getWorkerPoolMaxSize();
  28     }
  29   }
  30
  31   /**
  32    *
  33    * @return {Promise}
  34    * @public
  35    */
  36   async start(): Promise<Worker> {
  37     if (Configuration.useWorkerPool()) {
  38       await this._startWorkerWithPool();
  39     } else {
  40       await this._startWorker();
  41     }
  42     return this._worker;
  43   }
  44
  45   /**
  46    *
  47    * @return {void}
  48    * @public
  49    */
  50   addWorkerElement(workerData: WorkerData): void {
  51     if (Configuration.useWorkerPool()) {
  52       throw Error('Cannot add Wrk element if the worker pool is enabled');
  53     }
  54     if (this._numWorkerElements >= this._maxWorkerElements) {
  55       throw Error('Cannot add Wrk element: max number of elements per worker reached');
  56     }
  57     this._workerData = workerData;
  58     this._worker.postMessage({ id: Constants.START_WORKER_ELEMENT, workerData: workerData });
  59     this._numWorkerElements++;
  60   }
  61
  62   /**
  63    *
  64    * @return {number}
  65    * @public
  66    */
  67   public getWorkerPoolSize(): number {
  68     if (Configuration.useWorkerPool()) {
  69       return WorkerPool.getPoolSize();
  70     }
  71   }
  72
  73   /**
  74    *
  75    * @return {Promise}
  76    * @private
  77    */
  78   private async _startWorkerWithPool() {
  79     return new Promise((resolve, reject) => {
  80       WorkerPool.acquire(this._workerScript, { workerData: this._workerData }, (err, worker) => {
  81         if (err) {
  82           return reject(err);
  83         }
  84         worker.once('message', resolve);
  85         worker.once('error', reject);
  86         this._worker = worker;
  87       });
  88     });
  89   }
  90
  91   /**
  92    *
  93    * @return {Promise}
  94    * @private
  95    */
  96   private async _startWorker() {
  97     return new Promise((resolve, reject) => {
  98       const worker = new Worker(this._workerScript, { workerData: this._workerData });
  99       worker.on('message', resolve);
 100       worker.on('error', reject);
 101       worker.on('exit', (code) => {
 102         if (code !== 0) {
 103           reject(new Error(`Worker stopped with exit code ${code}`));
 104         }
 105       });
 106       this._numWorkerElements++;
 107       this._worker = worker;
 108     });
 109   }
 110 }
 111
 112
 113 class WorkerPool {
 114   public static maxConcurrentWorkers: number;
 115   private static _instance: Pool;
 116
 117   private constructor() { }
 118
 119   public static getInstance(): Pool {
 120     if (!WorkerPool._instance) {
 121       WorkerPool._instance = new Pool({ max: WorkerPool.maxConcurrentWorkers });
 122     }
 123     return WorkerPool._instance;
 124   }
 125
 126   public static acquire(filename: string, options: WorkerOptions, callback: (error: Error | null, worker: Worker) => void): void {
 127     WorkerPool.getInstance().acquire(filename, options, callback);
 128   }
 129
 130   public static getPoolSize(): number {
 131     return WorkerPool.getInstance().size;
 132   }
 133 }