Commit | Line | Data |
---|---|---|
059438ff | 1 | import cluster, { type ClusterSettings, type Worker } from 'node:cluster' |
deb85c12 | 2 | import type { MessageValue } from '../../utility-types' |
c97c7edb | 3 | import { AbstractPool } from '../abstract-pool' |
4b628b48 JB |
4 | import { type PoolOptions, type PoolType, PoolTypes } from '../pool' |
5 | import { type WorkerType, WorkerTypes } from '../worker' | |
4ade5f1f | 6 | |
729c563d S |
7 | /** |
8 | * Options for a poolifier cluster pool. | |
9 | */ | |
c97c7edb | 10 | export 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 | 33 | export 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) |
78f60f82 | 78 | workerNode.removeAllListeners() |
5b59d53e | 79 | worker.disconnect() |
c2301b8e | 80 | await waitWorkerExit |
4ade5f1f S |
81 | } |
82 | ||
afc003b2 | 83 | /** @inheritDoc */ |
aa9eede8 JB |
84 | protected sendToWorker ( |
85 | workerNodeKey: number, | |
86 | message: MessageValue<Data> | |
87 | ): void { | |
72ae84a2 JB |
88 | this.workerNodes[workerNodeKey].worker.send({ |
89 | ...message, | |
dbfa7948 | 90 | workerId: this.getWorkerInfo(workerNodeKey).id as number |
72ae84a2 | 91 | }) |
4ade5f1f S |
92 | } |
93 | ||
85aeb3f3 | 94 | /** @inheritDoc */ |
aa9eede8 JB |
95 | protected sendStartupMessageToWorker (workerNodeKey: number): void { |
96 | this.sendToWorker(workerNodeKey, { | |
e9dd5b66 | 97 | ready: false |
85aeb3f3 JB |
98 | }) |
99 | } | |
100 | ||
101 | /** @inheritDoc */ | |
102 | protected registerWorkerMessageListener<Message extends Data | Response>( | |
aa9eede8 | 103 | workerNodeKey: number, |
85aeb3f3 JB |
104 | listener: (message: MessageValue<Message>) => void |
105 | ): void { | |
aa9eede8 | 106 | this.workerNodes[workerNodeKey].worker.on('message', listener) |
85aeb3f3 JB |
107 | } |
108 | ||
ae036c3e JB |
109 | /** @inheritDoc */ |
110 | protected registerOnceWorkerMessageListener<Message extends Data | Response>( | |
111 | workerNodeKey: number, | |
112 | listener: (message: MessageValue<Message>) => void | |
113 | ): void { | |
114 | this.workerNodes[workerNodeKey].worker.once('message', listener) | |
115 | } | |
116 | ||
117 | /** @inheritDoc */ | |
118 | protected deregisterWorkerMessageListener<Message extends Data | Response>( | |
119 | workerNodeKey: number, | |
120 | listener: (message: MessageValue<Message>) => void | |
121 | ): void { | |
122 | this.workerNodes[workerNodeKey].worker.off('message', listener) | |
123 | } | |
124 | ||
afc003b2 | 125 | /** @inheritDoc */ |
280c2a77 | 126 | protected createWorker (): Worker { |
7e0d447f | 127 | return cluster.fork(this.opts.env) |
4ade5f1f S |
128 | } |
129 | ||
afc003b2 | 130 | /** @inheritDoc */ |
8881ae32 | 131 | protected get type (): PoolType { |
6b27d407 | 132 | return PoolTypes.fixed |
7c0ba920 JB |
133 | } |
134 | ||
184855e6 JB |
135 | /** @inheritDoc */ |
136 | protected get worker (): WorkerType { | |
137 | return WorkerTypes.cluster | |
138 | } | |
139 | ||
afc003b2 | 140 | /** @inheritDoc */ |
c319c66b | 141 | protected get busy (): boolean { |
c2ade475 | 142 | return this.internalBusy() |
7c0ba920 | 143 | } |
4ade5f1f | 144 | } |