1 import { isMainThread
, MessageChannel
, SHARE_ENV
, Worker
} from
'worker_threads'
2 import type { Draft
, MessageValue
} from
'../../utility-types'
3 import { AbstractPool
} from
'../abstract-pool'
4 import type { PoolOptions
} from
'../pool'
5 import { PoolType
} from
'../pool-internal'
8 * A thread worker with message channels for communication between main thread and thread worker.
10 export type ThreadWorkerWithMessageChannel
= Worker
& Draft
<MessageChannel
>
13 * A thread pool with a fixed number of threads.
15 * It is possible to perform tasks in sync or asynchronous mode as you prefer.
17 * This pool selects the threads in a round robin fashion.
19 * @template DataType of data sent to the worker. This can only be serializable data.
20 * @template ResponseType 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.
36 numberOfThreads
: number,
38 opts
: PoolOptions
<ThreadWorkerWithMessageChannel
> = {}
40 super(numberOfThreads
, filePath
, opts
)
44 protected isMain (): boolean {
49 public async destroyWorker (
50 worker
: ThreadWorkerWithMessageChannel
52 this.sendToWorker(worker
, { kill
: 1 })
53 await worker
.terminate()
57 protected sendToWorker (
58 worker
: ThreadWorkerWithMessageChannel
,
59 message
: MessageValue
<Data
>
61 worker
.postMessage(message
)
65 public registerWorkerMessageListener
<Message
extends Data
| Response
> (
66 messageChannel
: ThreadWorkerWithMessageChannel
,
67 listener
: (message
: MessageValue
<Message
>) => void
69 messageChannel
.port2
?.on('message', listener
)
73 protected createWorker (): ThreadWorkerWithMessageChannel
{
74 return new Worker(this.filePath
, {
80 protected afterWorkerSetup (worker
: ThreadWorkerWithMessageChannel
): void {
81 const { port1
, port2
} = new MessageChannel()
82 worker
.postMessage({ parent: port1
}, [port1
])
85 // Listen worker messages.
86 this.registerWorkerMessageListener(worker
, super.workerListener())
90 public get
type (): PoolType
{
95 public get
busy (): boolean {
96 return this.internalGetBusyStatus()