dee28c53fc00d0f62f5d9ea753a8ecad3151bba3
1 import { MessageChannel
, SHARE_ENV
, Worker
, isMainThread
} from
'worker_threads'
6 export type Draft
<T
> = { -readonly [P
in keyof T
]?: T
[P
] }
8 export type WorkerWithMessageChannel
= Worker
& Draft
<MessageChannel
>
10 export interface FixedThreadPoolOptions
{
12 * A function that will listen for error event on each worker thread.
14 errorHandler
?: (this: Worker
, e
: Error) => void
16 * A function that will listen for online event on each worker thread.
18 onlineHandler
?: (this: Worker
) => void
20 * A function that will listen for exit event on each worker thread.
22 exitHandler
?: (this: Worker
, code
: number) => void
24 * This is just to avoid not useful warnings message, is used to set `maxListeners` on event emitters (workers are event emitters).
32 * A thread pool with a static number of threads, is possible to execute tasks in sync or async mode as you prefer.
34 * This pool will select the worker thread in a round robin fashion.
36 * @author [Alessandro Pio Ardizio](https://github.com/pioardi)
39 export default class FixedThreadPool
<Data
= any, Response
= any> {
40 public readonly workers
: WorkerWithMessageChannel
[] = []
41 public nextWorker
: number = 0
43 // threadId as key and an integer value
44 public readonly tasks
: Map
<WorkerWithMessageChannel
, number> = new Map
<
45 WorkerWithMessageChannel
,
49 protected _id
: number = 0
52 * @param numThreads Num of threads for this worker pool.
53 * @param filePath A file path with implementation of `ThreadWorker` class, relative path is fine.
54 * @param opts An object with possible options for example `errorHandler`, `onlineHandler`. Default: `{ maxTasks: 1000 }`
57 public readonly numThreads
: number,
58 public readonly filePath
: string,
59 public readonly opts
: FixedThreadPoolOptions
= { maxTasks
: 1000 }
62 throw new Error('Cannot start a thread pool from a worker thread !!!')
65 throw new Error('Please specify a file with a worker implementation')
68 for (let i
= 1; i
<= this.numThreads
; i
++) {
73 public async destroy (): Promise
<void> {
74 for (const worker
of this.workers
) {
75 await worker
.terminate()
80 * Execute the task specified into the constructor with the data parameter.
82 * @param data The input for the task specified.
83 * @returns Promise that is resolved when the task is done.
85 public async execute (data
: Data
): Promise
<Response
> {
86 // configure worker to handle message with the specified task
87 const worker
= this._chooseWorker()
88 this.tasks
.set(worker
, this.tasks
.get(worker
) + 1)
90 const res
= this._execute(worker
, id
)
91 worker
.postMessage({ data
: data
|| _void
, _id
: id
})
96 worker
: WorkerWithMessageChannel
,
98 ): Promise
<Response
> {
99 return new Promise((resolve
, reject
) => {
100 const listener
= (message
) => {
101 if (message
._id
=== id
) {
102 worker
.port2
.removeListener('message', listener
)
103 this.tasks
.set(worker
, this.tasks
.get(worker
) - 1)
104 if (message
.error
) reject(message
.error
)
105 else resolve(message
.data
)
108 worker
.port2
.on('message', listener
)
112 protected _chooseWorker (): WorkerWithMessageChannel
{
113 if (this.workers
.length
- 1 === this.nextWorker
) {
115 return this.workers
[this.nextWorker
]
118 return this.workers
[this.nextWorker
]
122 protected _newWorker (): WorkerWithMessageChannel
{
123 const worker
: WorkerWithMessageChannel
= new Worker(this.filePath
, {
126 worker
.on('error', this.opts
.errorHandler
|| empty
)
127 worker
.on('online', this.opts
.onlineHandler
|| empty
)
128 // TODO handle properly when a thread exit
129 worker
.on('exit', this.opts
.exitHandler
|| empty
)
130 this.workers
.push(worker
)
131 const { port1
, port2
} = new MessageChannel()
132 worker
.postMessage({ parent: port1
}, [port1
])
135 // we will attach a listener for every task,
136 // when task is completed the listener will be removed but to avoid warnings we are increasing the max listeners size
137 worker
.port2
.setMaxListeners(this.opts
.maxTasks
|| 1000)
139 this.tasks
.set(worker
, 0)
144 module.exports
= FixedThreadPool