fix: ensure worker node destroy semantic is always the same
[poolifier.git] / src / pools / cluster / fixed.ts
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'
6
7 /**
8 * Options for a poolifier cluster pool.
9 */
10 export interface ClusterPoolOptions extends PoolOptions<Worker> {
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 */
16 env?: Record<string, unknown>
17 /**
18 * Cluster settings.
19 *
20 * @see https://nodejs.org/api/cluster.html#cluster_cluster_settings
21 */
22 settings?: ClusterSettings
23 }
24
25 /**
26 * A cluster pool with a fixed number of workers.
27 *
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)
31 * @since 2.0.0
32 */
33 export class FixedClusterPool<
34 Data = unknown,
35 Response = unknown
36 > extends AbstractPool<Worker, Data, Response> {
37 /**
38 * Constructs a new poolifier fixed cluster pool.
39 *
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.
43 */
44 public constructor (
45 numberOfWorkers: number,
46 filePath: string,
47 protected readonly opts: ClusterPoolOptions = {}
48 ) {
49 super(numberOfWorkers, filePath, opts)
50 }
51
52 /** @inheritDoc */
53 protected setupHook (): void {
54 cluster.setupPrimary({ ...this.opts.settings, exec: this.filePath })
55 }
56
57 /** @inheritDoc */
58 protected isMain (): boolean {
59 return cluster.isPrimary
60 }
61
62 /** @inheritDoc */
63 protected async destroyWorkerNode (workerNodeKey: number): Promise<void> {
64 this.flushTasksQueue(workerNodeKey)
65 // FIXME: wait for tasks to be finished
66 const worker = this.workerNodes[workerNodeKey].worker
67 const workerExitPromise = new Promise<void>(resolve => {
68 worker.on('exit', () => {
69 resolve()
70 })
71 })
72 worker.on('disconnect', () => {
73 worker.kill()
74 })
75 this.sendToWorker(workerNodeKey, { kill: true, workerId: worker.id })
76 worker.disconnect()
77 await workerExitPromise
78 }
79
80 /** @inheritDoc */
81 protected sendToWorker (
82 workerNodeKey: number,
83 message: MessageValue<Data>
84 ): void {
85 this.workerNodes[workerNodeKey].worker.send(message)
86 }
87
88 /** @inheritDoc */
89 protected sendStartupMessageToWorker (workerNodeKey: number): void {
90 this.sendToWorker(workerNodeKey, {
91 ready: false,
92 workerId: this.workerNodes[workerNodeKey].worker.id
93 })
94 }
95
96 /** @inheritDoc */
97 protected registerWorkerMessageListener<Message extends Data | Response>(
98 workerNodeKey: number,
99 listener: (message: MessageValue<Message>) => void
100 ): void {
101 this.workerNodes[workerNodeKey].worker.on('message', listener)
102 }
103
104 /** @inheritDoc */
105 protected createWorker (): Worker {
106 return cluster.fork(this.opts.env)
107 }
108
109 /** @inheritDoc */
110 protected get type (): PoolType {
111 return PoolTypes.fixed
112 }
113
114 /** @inheritDoc */
115 protected get worker (): WorkerType {
116 return WorkerTypes.cluster
117 }
118
119 /** @inheritDoc */
120 protected get minSize (): number {
121 return this.numberOfWorkers
122 }
123
124 /** @inheritDoc */
125 protected get maxSize (): number {
126 return this.numberOfWorkers
127 }
128
129 /** @inheritDoc */
130 protected get busy (): boolean {
131 return this.internalBusy()
132 }
133 }