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