1 import { isMainThread
, MessageChannel
, SHARE_ENV
, Worker
} from
'worker_threads'
2 import type { Draft
, JSONValue
, 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.
19 * @template Response Type of response of execution.
21 * @author [Alessandro Pio Ardizio](https://github.com/pioardi)
24 export class FixedThreadPool
<
25 Data
extends JSONValue
= JSONValue
,
26 Response
extends JSONValue
= JSONValue
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 {
47 protected async destroyWorker (
48 worker
: ThreadWorkerWithMessageChannel
50 this.sendToWorker(worker
, { kill
: 1 })
51 await worker
.terminate()
54 protected sendToWorker (
55 worker
: ThreadWorkerWithMessageChannel
,
56 message
: MessageValue
<Data
>
58 worker
.postMessage(message
)
61 protected registerWorkerMessageListener
<Message
extends Data
| Response
> (
62 messageChannel
: ThreadWorkerWithMessageChannel
,
63 listener
: (message
: MessageValue
<Message
>) => void
65 messageChannel
.port2
?.on('message', listener
)
68 protected unregisterWorkerMessageListener
<Message
extends Data
| Response
> (
69 messageChannel
: ThreadWorkerWithMessageChannel
,
70 listener
: (message
: MessageValue
<Message
>) => void
72 messageChannel
.port2
?.removeListener('message', listener
)
75 protected createWorker (): ThreadWorkerWithMessageChannel
{
76 return new Worker(this.filePath
, {
81 protected afterWorkerSetup (worker
: ThreadWorkerWithMessageChannel
): void {
82 const { port1
, port2
} = new MessageChannel()
83 worker
.postMessage({ parent: port1
}, [port1
])
86 // we will attach a listener for every task,
87 // when task is completed the listener will be removed but to avoid warnings we are increasing the max listeners size
88 worker
.port2
.setMaxListeners(this.opts
.maxTasks
?? 1000)