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 await worker
.terminate()
53 protected sendToWorker (
54 worker
: ThreadWorkerWithMessageChannel
,
55 message
: MessageValue
<Data
>
57 worker
.postMessage(message
)
60 protected registerWorkerMessageListener
<Message
extends Data
| Response
> (
61 messageChannel
: ThreadWorkerWithMessageChannel
,
62 listener
: (message
: MessageValue
<Message
>) => void
64 messageChannel
.port2
?.on('message', listener
)
67 protected unregisterWorkerMessageListener
<Message
extends Data
| Response
> (
68 messageChannel
: ThreadWorkerWithMessageChannel
,
69 listener
: (message
: MessageValue
<Message
>) => void
71 messageChannel
.port2
?.removeListener('message', listener
)
74 protected createWorker (): ThreadWorkerWithMessageChannel
{
75 return new Worker(this.filePath
, {
80 protected afterWorkerSetup (worker
: ThreadWorkerWithMessageChannel
): void {
81 const { port1
, port2
} = new MessageChannel()
82 worker
.postMessage({ parent: port1
}, [port1
])
85 // we will attach a listener for every task,
86 // when task is completed the listener will be removed but to avoid warnings we are increasing the max listeners size
87 worker
.port2
.setMaxListeners(this.opts
.maxTasks
?? 1000)