Fix tests run on node 10.x and add test assertion
[poolifier.git] / lib / fixed.js
1 'use strict'
2 const {
3 Worker, isMainThread, MessageChannel, SHARE_ENV
4 } = require('worker_threads')
5 const { generateID } = require('./util')
6
7 function empty () {}
8 /**
9 * A thread pool with a static number of threads , is possible to execute tasks in sync or async mode as you prefer. <br>
10 * This pool will select the worker thread in a round robin fashion. <br>
11 * @author Alessandro Pio Ardizio
12 * @since 0.0.1
13 */
14 class FixedThreadPool {
15 /**
16 *
17 * @param {Number} numThreads Num of threads for this worker pool
18 * @param {string} a file path with implementation of @see ThreadWorker class, relative path is fine
19 * @param {Object} an object with possible options for example errorHandler, onlineHandler.
20 */
21 constructor (numThreads, filePath, opts) {
22 if (!isMainThread) throw new Error('Cannot start a thread pool from a worker thread !!!')
23 if (!filePath) throw new Error('Please specify a file with a worker implementation')
24 this.numThreads = numThreads
25 this.workers = []
26 this.nextWorker = 0
27 this.opts = opts || { maxTasks: 1000 }
28 this.filePath = filePath
29 // threadId as key and an integer value
30 this.tasks = new Map()
31 for (let i = 1; i <= numThreads; i++) {
32 this._newWorker()
33 }
34 }
35
36 destroy () {
37 for (const worker of this.workers) {
38 worker.terminate()
39 }
40 }
41
42 /**
43 * Execute the task specified into the constructor with the data parameter.
44 * @param {Any} the input for the task specified
45 * @returns {Promise} that is resolved when the task is done
46 */
47 async execute (data) {
48 // configure worker to handle message with the specified task
49 const worker = this._chooseWorker()
50 this.tasks.set(worker, this.tasks.get(worker) + 1)
51 const id = generateID()
52 const res = this._execute(worker, id)
53 worker.postMessage({ data: data, _id: id })
54 return res
55 }
56
57 _execute (worker, id) {
58 return new Promise((resolve, reject) => {
59 const listener = (message) => {
60 if (message._id === id) {
61 worker.port2.removeListener('message', listener)
62 this.tasks.set(worker, this.tasks.get(worker) - 1)
63 resolve(message.data)
64 }
65 }
66 worker.port2.on('message', listener)
67 })
68 }
69
70 _chooseWorker () {
71 if ((this.workers.length - 1) === this.nextWorker) {
72 this.nextWorker = 0
73 return this.workers[this.nextWorker]
74 } else {
75 this.nextWorker++
76 return this.workers[this.nextWorker]
77 }
78 }
79
80 _newWorker () {
81 const worker = new Worker(this.filePath, { env: SHARE_ENV })
82 worker.on('error', this.opts.errorHandler || empty)
83 worker.on('online', this.opts.onlineHandler || empty)
84 // TODO handle properly when a thread exit
85 worker.on('exit', this.opts.exitHandler || empty)
86 this.workers.push(worker)
87 const { port1, port2 } = new MessageChannel()
88 worker.postMessage({ parent: port1 }, [port1])
89 worker.port1 = port1
90 worker.port2 = port2
91 // we will attach a listener for every task,
92 // when task is completed the listener will be removed but to avoid warnings we are increasing the max listeners size
93 worker.port2.setMaxListeners(this.opts.maxTasks || 1000)
94 // init tasks map
95 this.tasks.set(worker, 0)
96 return worker
97 }
98 }
99
100 module.exports = FixedThreadPool