lib/dynamic.js

   1 'use strict'
   2 const FixedThreadPool = require('./fixed')
   3 const { randomWorker } = require('./util')
   4 const EventEmitter = require('events')
   5 class MyEmitter extends EventEmitter {}
   6
   7 /**
   8  * A thread pool with a min/max number of threads , is possible to execute tasks in sync or async mode as you prefer. <br>
   9  * This thread pool will create new workers when the other ones are busy, until the max number of threads,
  10  * when the max number of threads is reached, an event will be emitted , if you want to listen this event use the emitter method.
  11  * @author Alessandro Pio Ardizio
  12  * @since 0.0.1
  13  */
  14 class DynamicThreadPool extends FixedThreadPool {
  15   /**
  16     *
  17     * @param {Number} min  Min number of threads that will be always active
  18     * @param {Number} max  Max number of threads that will be active
  19     * @param {Object} an object with possible options for example maxConcurrency
  20   */
  21   constructor (min, max, filename, opts) {
  22     super(min, filename, opts)
  23     this.max = max
  24     this.emitter = new MyEmitter()
  25   }
  26
  27   /**
  28    * Return an event emitter that will send some messages, for example
  29    * a message will be sent when max number of threads is reached and all threads are busy
  30    * in this case it will emit a message
  31    */
  32   emitter () {
  33     return this.emitter
  34   }
  35
  36   _chooseWorker () {
  37     let worker
  38     for (const entry of this.tasks) {
  39       if (entry[1] === 0) {
  40         worker = entry[0]
  41         break
  42       }
  43     }
  44
  45     if (worker) {
  46       // a worker is free, use it
  47       return worker
  48     } else {
  49       if (this.workers.length === this.max) {
  50         this.emitter.emit('FullPool')
  51         return randomWorker(this.tasks)
  52       }
  53       // all workers are busy create a new worker
  54       const worker = this._newWorker()
  55       worker.port2.on('message', (message) => {
  56         if (message.kill) {
  57           worker.postMessage({ kill: 1 })
  58           worker.terminate()
  59         }
  60       })
  61       return worker
  62     }
  63   }
  64 }
  65
  66 module.exports = DynamicThreadPool