src/charging-station/Worker.ts

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