-import { MessageChannel, SHARE_ENV, Worker, isMainThread } from 'worker_threads'
-
-function empty () {}
-const _void = {}
+import { isMainThread, MessageChannel, SHARE_ENV, Worker } from 'worker_threads'
export type Draft<T> = { -readonly [P in keyof T]?: T[P] }
* @author [Alessandro Pio Ardizio](https://github.com/pioardi)
* @since 0.0.1
*/
-export default class FixedThreadPool<Data = any, Response = any> {
+// eslint-disable-next-line @typescript-eslint/no-explicit-any
+export class FixedThreadPool<Data = any, Response = any> {
public readonly workers: WorkerWithMessageChannel[] = []
public nextWorker: number = 0
number
>()
- protected _id: number = 0
+ protected id: number = 0
/**
* @param numThreads Num of threads for this worker pool.
if (!isMainThread) {
throw new Error('Cannot start a thread pool from a worker thread !!!')
}
+ // TODO christopher 2021-02-07: Improve this check e.g. with a pattern or blank check
if (!this.filePath) {
throw new Error('Please specify a file with a worker implementation')
}
for (let i = 1; i <= this.numThreads; i++) {
- this._newWorker()
+ this.newWorker()
}
}
* @param data The input for the task specified.
* @returns Promise that is resolved when the task is done.
*/
- public async execute (data: Data): Promise<Response> {
+ public execute (data: Data): Promise<Response> {
// configure worker to handle message with the specified task
- const worker = this._chooseWorker()
- this.tasks.set(worker, this.tasks.get(worker) + 1)
- const id = ++this._id
- const res = this._execute(worker, id)
- worker.postMessage({ data: data || _void, _id: id })
+ const worker = this.chooseWorker()
+ const previousWorkerIndex = this.tasks.get(worker)
+ if (previousWorkerIndex !== undefined) {
+ this.tasks.set(worker, previousWorkerIndex + 1)
+ } else {
+ throw Error('Worker could not be found in tasks map')
+ }
+ const id = ++this.id
+ const res = this.internalExecute(worker, id)
+ worker.postMessage({ data: data || {}, id: id })
return res
}
- protected _execute (
+ protected internalExecute (
worker: WorkerWithMessageChannel,
id: number
): Promise<Response> {
return new Promise((resolve, reject) => {
- const listener = (message) => {
- if (message._id === id) {
- worker.port2.removeListener('message', listener)
- this.tasks.set(worker, this.tasks.get(worker) - 1)
+ const listener = (message: {
+ id: number
+ error?: string
+ data: Response
+ }): void => {
+ if (message.id === id) {
+ worker.port2?.removeListener('message', listener)
+ const previousWorkerIndex = this.tasks.get(worker)
+ if (previousWorkerIndex !== undefined) {
+ this.tasks.set(worker, previousWorkerIndex + 1)
+ } else {
+ throw Error('Worker could not be found in tasks map')
+ }
if (message.error) reject(message.error)
else resolve(message.data)
}
}
- worker.port2.on('message', listener)
+ worker.port2?.on('message', listener)
})
}
- protected _chooseWorker (): WorkerWithMessageChannel {
+ protected chooseWorker (): WorkerWithMessageChannel {
if (this.workers.length - 1 === this.nextWorker) {
this.nextWorker = 0
return this.workers[this.nextWorker]
}
}
- protected _newWorker (): WorkerWithMessageChannel {
+ protected newWorker (): WorkerWithMessageChannel {
const worker: WorkerWithMessageChannel = new Worker(this.filePath, {
env: SHARE_ENV
})
- worker.on('error', this.opts.errorHandler || empty)
- worker.on('online', this.opts.onlineHandler || empty)
+ worker.on('error', this.opts.errorHandler ?? (() => {}))
+ worker.on('online', this.opts.onlineHandler ?? (() => {}))
// TODO handle properly when a thread exit
- worker.on('exit', this.opts.exitHandler || empty)
+ worker.on('exit', this.opts.exitHandler ?? (() => {}))
this.workers.push(worker)
const { port1, port2 } = new MessageChannel()
worker.postMessage({ parent: port1 }, [port1])
worker.port2 = port2
// we will attach a listener for every task,
// when task is completed the listener will be removed but to avoid warnings we are increasing the max listeners size
- worker.port2.setMaxListeners(this.opts.maxTasks || 1000)
+ worker.port2.setMaxListeners(this.opts.maxTasks ?? 1000)
// init tasks map
this.tasks.set(worker, 0)
return worker
}
}
-
-module.exports = FixedThreadPool