1 import { isMainThread
, MessageChannel
, SHARE_ENV
, Worker
} from
'worker_threads'
2 import type { Draft
, MessageValue
} from
'../../utility-types'
3 import type { PoolOptions
} from
'../abstract-pool'
4 import { AbstractPool
} from
'../abstract-pool'
7 * A thread worker with message channels for communication between main thread and thread worker.
9 export type ThreadWorkerWithMessageChannel
= Worker
& Draft
<MessageChannel
>
12 * A thread pool with a fixed number of threads.
14 * It is possible to perform tasks in sync or asynchronous mode as you prefer.
16 * This pool selects the threads in a round robin fashion.
18 * @template Data Type of data sent to the worker. This can only be serializable data.
19 * @template Response Type of response of execution. This can only be serializable data.
21 * @author [Alessandro Pio Ardizio](https://github.com/pioardi)
24 export class FixedThreadPool
<
27 > extends AbstractPool
<ThreadWorkerWithMessageChannel
, Data
, Response
> {
29 * Constructs a new poolifier fixed thread pool.
31 * @param numberOfThreads Number of threads for this pool.
32 * @param filePath Path to an implementation of a `ThreadWorker` file, which can be relative or absolute.
33 * @param opts Options for this fixed thread pool. Default: `{ maxTasks: 1000 }`
36 numberOfThreads
: number,
38 opts
: PoolOptions
<ThreadWorkerWithMessageChannel
> = { maxTasks
: 1000 }
40 super(numberOfThreads
, filePath
, opts
)
43 protected isMain (): boolean {
48 public async destroyWorker (
49 worker
: ThreadWorkerWithMessageChannel
51 this.sendToWorker(worker
, { kill
: 1 })
52 await worker
.terminate()
55 protected sendToWorker (
56 worker
: ThreadWorkerWithMessageChannel
,
57 message
: MessageValue
<Data
>
59 worker
.postMessage(message
)
63 public registerWorkerMessageListener
<Message
extends Data
| Response
> (
64 messageChannel
: ThreadWorkerWithMessageChannel
,
65 listener
: (message
: MessageValue
<Message
>) => void
67 messageChannel
.port2
?.on('message', listener
)
70 protected createWorker (): ThreadWorkerWithMessageChannel
{
71 return new Worker(this.filePath
, {
76 protected afterWorkerSetup (worker
: ThreadWorkerWithMessageChannel
): void {
77 const { port1
, port2
} = new MessageChannel()
78 worker
.postMessage({ parent: port1
}, [port1
])
81 // We will attach a listener for every task,
82 // when the task is completed the listener will be removed but to avoid warnings we are increasing the max listeners size.
83 worker
.port2
.setMaxListeners(this.opts
.maxTasks
?? 1000)
84 this.registerWorkerMessageListener(worker
, super.workerListener())