6 } from
'worker_threads'
11 export type Draft
<T
> = { -readonly [P
in keyof T
]?: T
[P
] }
13 export type WorkerWithMessageChannel
= Worker
& Draft
<MessageChannel
>
15 export interface FixedThreadPoolOptions
{
17 * A function that will listen for error event on each worker thread.
19 errorHandler
?: (this: Worker
, e
: Error) => void
21 * A function that will listen for online event on each worker thread.
23 onlineHandler
?: (this: Worker
) => void
25 * A function that will listen for exit event on each worker thread.
27 exitHandler
?: (this: Worker
, code
: number) => void
29 * This is just to avoid not useful warnings message, is used to set `maxListeners` on event emitters (workers are event emitters).
37 * A thread pool with a static number of threads, is possible to execute tasks in sync or async mode as you prefer.
39 * This pool will select the worker thread in a round robin fashion.
41 * @author [Alessandro Pio Ardizio](https://github.com/pioardi)
44 export default class FixedThreadPool
<Data
= any, Response
= any> {
45 public readonly workers
: WorkerWithMessageChannel
[] = []
46 public nextWorker
: number = 0
48 // threadId as key and an integer value
49 public readonly tasks
: Map
<WorkerWithMessageChannel
, number> = new Map
<
50 WorkerWithMessageChannel
,
54 protected _id
: number = 0
57 * @param numThreads Num of threads for this worker pool.
58 * @param filePath A file path with implementation of `ThreadWorker` class, relative path is fine.
59 * @param opts An object with possible options for example `errorHandler`, `onlineHandler`. Default: `{ maxTasks: 1000 }`
62 public readonly numThreads
: number,
63 public readonly filePath
: string,
64 public readonly opts
: FixedThreadPoolOptions
= { maxTasks
: 1000 }
67 throw new Error('Cannot start a thread pool from a worker thread !!!')
69 throw new Error('Please specify a file with a worker implementation')
71 for (let i
= 1; i
<= this.numThreads
; i
++) {
76 public async destroy (): Promise
<void> {
77 for (const worker
of this.workers
) {
78 await worker
.terminate()
83 * Execute the task specified into the constructor with the data parameter.
85 * @param data The input for the task specified.
86 * @returns Promise that is resolved when the task is done.
88 public async execute (data
: Data
): Promise
<Response
> {
89 // configure worker to handle message with the specified task
90 const worker
= this._chooseWorker()
91 this.tasks
.set(worker
, this.tasks
.get(worker
) + 1)
93 const res
= this._execute(worker
, id
)
94 worker
.postMessage({ data
: data
|| _void
, _id
: id
})
99 worker
: WorkerWithMessageChannel
,
101 ): Promise
<Response
> {
102 return new Promise((resolve
, reject
) => {
103 const listener
= (message
) => {
104 if (message
._id
=== id
) {
105 worker
.port2
.removeListener('message', listener
)
106 this.tasks
.set(worker
, this.tasks
.get(worker
) - 1)
107 if (message
.error
) reject(message
.error
)
108 else resolve(message
.data
)
111 worker
.port2
.on('message', listener
)
115 protected _chooseWorker (): WorkerWithMessageChannel
{
116 if (this.workers
.length
- 1 === this.nextWorker
) {
118 return this.workers
[this.nextWorker
]
121 return this.workers
[this.nextWorker
]
125 protected _newWorker (): WorkerWithMessageChannel
{
126 const worker
: WorkerWithMessageChannel
= new Worker(this.filePath
, {
129 worker
.on('error', this.opts
.errorHandler
|| empty
)
130 worker
.on('online', this.opts
.onlineHandler
|| empty
)
131 // TODO handle properly when a thread exit
132 worker
.on('exit', this.opts
.exitHandler
|| empty
)
133 this.workers
.push(worker
)
134 const { port1
, port2
} = new MessageChannel()
135 worker
.postMessage({ parent: port1
}, [port1
])
138 // we will attach a listener for every task,
139 // when task is completed the listener will be removed but to avoid warnings we are increasing the max listeners size
140 worker
.port2
.setMaxListeners(this.opts
.maxTasks
|| 1000)
142 this.tasks
.set(worker
, 0)
147 module.exports
= FixedThreadPool