fixed.js

   1 'use strict'
   2 const {
   3   Worker, isMainThread, MessageChannel
   4 } = require('worker_threads')
   5
   6 // FixedThreadPool , TrampolineThreadPool
   7 /**
   8  * A thread pool with a static number of threads , is possible to execute tasks in sync or async mode as you prefer
   9  * @author Alessandro Pio Ardizio
  10  * @since 0.0.1
  11  */
  12 class FixedThreadPool {
  13   /**
  14      *
  15      * @param {Number} numThreads  Num of threads for this worker pool
  16      * @param {Object} an object with possible options for example maxConcurrency
  17      */
  18   constructor (numThreads, task, opts) {
  19     if (!isMainThread) {
  20       throw new Error('Cannot start a thread pool from a worker thread !!!')
  21     }
  22     this.numThreads = numThreads
  23     this.workers = []
  24     this.task = task
  25     this.nextWorker = 0
  26     for (let i = 1; i <= numThreads; i++) {
  27       const worker = new Worker(__filename)
  28       this.workers.push(worker)
  29       const { port1 } = new MessageChannel()
  30       worker.emitter = port1
  31     }
  32   }
  33
  34   /**
  35    *
  36    * @param {Function} task , a function to execute
  37    * @param {Any} the input for the task specified
  38    */
  39   execute (task, data) {
  40     // TODO select a worker
  41     // configure worker to handle message with the specified task
  42     const worker = this._chooseWorker()
  43     const id = generateID()
  44     const res = this._execute(worker, task, id)
  45     worker.emitter.emit(id, data)
  46     return res
  47   }
  48
  49   _execute (worker, task, id) {
  50     return new Promise((resolve, reject) => {
  51       console.log('Executing a task on worker thread ' + worker.threadId)
  52       worker.emitter.once(id, (data) => {
  53         console.log('Receivd a message')
  54         try {
  55           resolve(task(data))
  56         } catch (e) {
  57           reject(e)
  58         }
  59       })
  60     })
  61   }
  62
  63   _chooseWorker () {
  64     console.log(this.workers.length -1)
  65     console.log(this.nextWorker)
  66     if ((this.workers.length - 1) === this.nextWorker) {
  67       console.log('we are here')
  68       this.nextWorker = 0
  69       return this.workers[this.nextWorker]
  70     } else {
  71       this.nextWorker++
  72       return this.workers[this.nextWorker]
  73     }
  74   }
  75 }
  76
  77 /**
  78  * Return an id to be associated to a node.
  79  */
  80 const generateID = () => {
  81   return Math.random()
  82     .toString(36)
  83     .substring(7)
  84 }
  85
  86 module.exports = FixedThreadPool