1 import cluster
, { type ClusterSettings
, type Worker
} from
'node:cluster'
2 import type { MessageValue
} from
'../../utility-types'
3 import { AbstractPool
} from
'../abstract-pool'
4 import { type PoolOptions
, type PoolType
, PoolTypes
} from
'../pool'
5 import { type WorkerType
, WorkerTypes
} from
'../worker'
8 * Options for a poolifier cluster pool.
10 export interface ClusterPoolOptions
extends PoolOptions
<Worker
> {
12 * Key/value pairs to add to worker process environment.
14 * @see https://nodejs.org/api/cluster.html#cluster_cluster_fork_env
16 env
?: Record
<string, unknown
>
20 * @see https://nodejs.org/api/cluster.html#cluster_cluster_settings
22 settings
?: ClusterSettings
26 * A cluster pool with a fixed number of workers.
28 * @typeParam Data - Type of data sent to the worker. This can only be structured-cloneable data.
29 * @typeParam Response - Type of execution response. This can only be structured-cloneable data.
30 * @author [Christopher Quadflieg](https://github.com/Shinigami92)
33 export class FixedClusterPool
<
36 > extends AbstractPool
<Worker
, Data
, Response
> {
38 * Constructs a new poolifier fixed cluster pool.
40 * @param numberOfWorkers - Number of workers for this pool.
41 * @param filePath - Path to an implementation of a `ClusterWorker` file, which can be relative or absolute.
42 * @param opts - Options for this fixed cluster pool.
45 numberOfWorkers
: number,
47 protected readonly opts
: ClusterPoolOptions
= {}
49 super(numberOfWorkers
, filePath
, opts
)
53 protected setupHook (): void {
54 cluster
.setupPrimary({ ...this.opts
.settings
, exec
: this.filePath
})
58 protected isMain (): boolean {
59 return cluster
.isPrimary
63 protected async destroyWorkerNode (workerNodeKey
: number): Promise
<void> {
64 this.flushTasksQueue(workerNodeKey
)
65 // FIXME: wait for tasks to be finished
66 const workerNode
= this.workerNodes
[workerNodeKey
]
67 const worker
= workerNode
.worker
68 const waitWorkerExit
= new Promise
<void>((resolve
) => {
69 worker
.on('exit', () => {
73 worker
.on('disconnect', () => {
76 await this.sendKillMessageToWorker(
78 workerNode
.info
.id
as number
85 protected sendToWorker (
86 workerNodeKey
: number,
87 message
: MessageValue
<Data
>
89 this.workerNodes
[workerNodeKey
].worker
.send(message
)
93 protected sendStartupMessageToWorker (workerNodeKey
: number): void {
94 this.sendToWorker(workerNodeKey
, {
96 workerId
: this.workerNodes
[workerNodeKey
].info
.id
as number
101 protected registerWorkerMessageListener
<Message
extends Data
| Response
>(
102 workerNodeKey
: number,
103 listener
: (message
: MessageValue
<Message
>) => void
105 this.workerNodes
[workerNodeKey
].worker
.on('message', listener
)
109 protected createWorker (): Worker
{
110 return cluster
.fork(this.opts
.env
)
114 protected get
type (): PoolType
{
115 return PoolTypes
.fixed
119 protected get
worker (): WorkerType
{
120 return WorkerTypes
.cluster
124 protected get
busy (): boolean {
125 return this.internalBusy()