Merge pull request #1497 from poolifier/fix-dynamic-worker-removal-race
[poolifier.git] / src / pools / cluster / fixed.ts
CommitLineData
059438ff 1import cluster, { type ClusterSettings, type Worker } from 'node:cluster'
deb85c12 2import type { MessageValue } from '../../utility-types'
c97c7edb 3import { AbstractPool } from '../abstract-pool'
4b628b48
JB
4import { type PoolOptions, type PoolType, PoolTypes } from '../pool'
5import { type WorkerType, WorkerTypes } from '../worker'
4ade5f1f 6
729c563d
S
7/**
8 * Options for a poolifier cluster pool.
9 */
c97c7edb 10export interface ClusterPoolOptions extends PoolOptions<Worker> {
325f50bc
S
11 /**
12 * Key/value pairs to add to worker process environment.
13 *
14 * @see https://nodejs.org/api/cluster.html#cluster_cluster_fork_env
15 */
82f36766 16 env?: Record<string, unknown>
1a76932b
JB
17 /**
18 * Cluster settings.
19 *
20 * @see https://nodejs.org/api/cluster.html#cluster_cluster_settings
21 */
22 settings?: ClusterSettings
4ade5f1f
S
23}
24
25/**
729c563d
S
26 * A cluster pool with a fixed number of workers.
27 *
e102732c
JB
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.
325f50bc
S
30 * @author [Christopher Quadflieg](https://github.com/Shinigami92)
31 * @since 2.0.0
4ade5f1f 32 */
d3c8a1a8 33export class FixedClusterPool<
deb85c12
JB
34 Data = unknown,
35 Response = unknown
d3c8a1a8 36> extends AbstractPool<Worker, Data, Response> {
4ade5f1f 37 /**
729c563d
S
38 * Constructs a new poolifier fixed cluster pool.
39 *
38e795c1
JB
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.
4ade5f1f
S
43 */
44 public constructor (
5c5a1fb7 45 numberOfWorkers: number,
c97c7edb 46 filePath: string,
dea903a8 47 protected readonly opts: ClusterPoolOptions = {}
4ade5f1f 48 ) {
5c5a1fb7 49 super(numberOfWorkers, filePath, opts)
c97c7edb 50 }
4ade5f1f 51
afc003b2 52 /** @inheritDoc */
c97c7edb 53 protected setupHook (): void {
1a76932b 54 cluster.setupPrimary({ ...this.opts.settings, exec: this.filePath })
c97c7edb 55 }
325f50bc 56
afc003b2 57 /** @inheritDoc */
c97c7edb 58 protected isMain (): boolean {
7e0d447f 59 return cluster.isPrimary
4ade5f1f
S
60 }
61
afc003b2 62 /** @inheritDoc */
81c02522 63 protected async destroyWorkerNode (workerNodeKey: number): Promise<void> {
ae3ab61d 64 this.flagWorkerNodeAsNotReady(workerNodeKey)
81c02522
JB
65 this.flushTasksQueue(workerNodeKey)
66 // FIXME: wait for tasks to be finished
75de9f41
JB
67 const workerNode = this.workerNodes[workerNodeKey]
68 const worker = workerNode.worker
041dc05b 69 const waitWorkerExit = new Promise<void>(resolve => {
ae036c3e 70 worker.once('exit', () => {
81c02522
JB
71 resolve()
72 })
73 })
ae036c3e 74 worker.once('disconnect', () => {
5b59d53e
JB
75 worker.kill()
76 })
72ae84a2 77 await this.sendKillMessageToWorker(workerNodeKey)
5b59d53e 78 worker.disconnect()
c2301b8e 79 await waitWorkerExit
4ade5f1f
S
80 }
81
afc003b2 82 /** @inheritDoc */
aa9eede8
JB
83 protected sendToWorker (
84 workerNodeKey: number,
85 message: MessageValue<Data>
86 ): void {
72ae84a2
JB
87 this.workerNodes[workerNodeKey].worker.send({
88 ...message,
dbfa7948 89 workerId: this.getWorkerInfo(workerNodeKey).id as number
72ae84a2 90 })
4ade5f1f
S
91 }
92
85aeb3f3 93 /** @inheritDoc */
aa9eede8
JB
94 protected sendStartupMessageToWorker (workerNodeKey: number): void {
95 this.sendToWorker(workerNodeKey, {
e9dd5b66 96 ready: false
85aeb3f3
JB
97 })
98 }
99
100 /** @inheritDoc */
101 protected registerWorkerMessageListener<Message extends Data | Response>(
aa9eede8 102 workerNodeKey: number,
85aeb3f3
JB
103 listener: (message: MessageValue<Message>) => void
104 ): void {
aa9eede8 105 this.workerNodes[workerNodeKey].worker.on('message', listener)
85aeb3f3
JB
106 }
107
ae036c3e
JB
108 /** @inheritDoc */
109 protected registerOnceWorkerMessageListener<Message extends Data | Response>(
110 workerNodeKey: number,
111 listener: (message: MessageValue<Message>) => void
112 ): void {
113 this.workerNodes[workerNodeKey].worker.once('message', listener)
114 }
115
116 /** @inheritDoc */
117 protected deregisterWorkerMessageListener<Message extends Data | Response>(
118 workerNodeKey: number,
119 listener: (message: MessageValue<Message>) => void
120 ): void {
121 this.workerNodes[workerNodeKey].worker.off('message', listener)
122 }
123
afc003b2 124 /** @inheritDoc */
280c2a77 125 protected createWorker (): Worker {
7e0d447f 126 return cluster.fork(this.opts.env)
4ade5f1f
S
127 }
128
afc003b2 129 /** @inheritDoc */
8881ae32 130 protected get type (): PoolType {
6b27d407 131 return PoolTypes.fixed
7c0ba920
JB
132 }
133
184855e6
JB
134 /** @inheritDoc */
135 protected get worker (): WorkerType {
136 return WorkerTypes.cluster
137 }
138
afc003b2 139 /** @inheritDoc */
c319c66b 140 protected get busy (): boolean {
c2ade475 141 return this.internalBusy()
7c0ba920 142 }
4ade5f1f 143}