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