Commit | Line | Data |
---|---|---|
bcfb06ce JB |
1 | import type { IPool } from '../pool.js' |
2 | import type { IWorker } from '../worker.js' | |
3 | import type { | |
4 | IWorkerChoiceStrategy, | |
5 | StrategyPolicy, | |
6 | TaskStatisticsRequirements, | |
7 | WorkerChoiceStrategy, | |
3a502712 | 8 | WorkerChoiceStrategyOptions, |
bcfb06ce | 9 | } from './selection-strategies-types.js' |
97231086 | 10 | |
bcfb06ce JB |
11 | import { WorkerChoiceStrategies } from './selection-strategies-types.js' |
12 | import { | |
19b8be8b JB |
13 | buildWorkerChoiceStrategiesPolicy, |
14 | buildWorkerChoiceStrategiesTaskStatisticsRequirements, | |
bcfb06ce | 15 | getWorkerChoiceStrategiesRetries, |
3a502712 | 16 | getWorkerChoiceStrategy, |
bcfb06ce JB |
17 | } from './selection-strategies-utils.js' |
18 | ||
19 | /** | |
20 | * The worker choice strategies context. | |
bcfb06ce JB |
21 | * @typeParam Worker - Type of worker. |
22 | * @typeParam Data - Type of data sent to the worker. This can only be structured-cloneable data. | |
23 | * @typeParam Response - Type of execution response. This can only be structured-cloneable data. | |
9eacd2e0 | 24 | * @internal |
bcfb06ce JB |
25 | */ |
26 | export class WorkerChoiceStrategiesContext< | |
27 | Worker extends IWorker, | |
28 | Data = unknown, | |
29 | Response = unknown | |
30 | > { | |
31 | /** | |
97231086 | 32 | * The default worker choice strategy in the context. |
bcfb06ce | 33 | */ |
97231086 | 34 | private defaultWorkerChoiceStrategy: WorkerChoiceStrategy |
bcfb06ce JB |
35 | |
36 | /** | |
97231086 | 37 | * The maximum number of worker choice strategies execution retries. |
bcfb06ce | 38 | */ |
97231086 | 39 | private readonly retries: number |
bcfb06ce JB |
40 | |
41 | /** | |
42 | * The worker choice strategies registered in the context. | |
43 | */ | |
44 | private readonly workerChoiceStrategies: Map< | |
3a502712 JB |
45 | WorkerChoiceStrategy, |
46 | IWorkerChoiceStrategy | |
bcfb06ce JB |
47 | > |
48 | ||
19b8be8b JB |
49 | /** |
50 | * The active worker choice strategies in the context policy. | |
51 | */ | |
52 | private workerChoiceStrategiesPolicy: StrategyPolicy | |
53 | ||
54 | /** | |
55 | * The active worker choice strategies in the context task statistics requirements. | |
56 | */ | |
57 | private workerChoiceStrategiesTaskStatisticsRequirements: TaskStatisticsRequirements | |
58 | ||
bcfb06ce | 59 | /** |
97231086 | 60 | * The number of worker choice strategies execution retries. |
bcfb06ce | 61 | */ |
97231086 | 62 | public retriesCount: number |
bcfb06ce JB |
63 | |
64 | /** | |
65 | * Worker choice strategies context constructor. | |
bcfb06ce JB |
66 | * @param pool - The pool instance. |
67 | * @param workerChoiceStrategies - The worker choice strategies. @defaultValue [WorkerChoiceStrategies.ROUND_ROBIN] | |
68 | * @param opts - The worker choice strategy options. | |
69 | */ | |
70 | public constructor ( | |
71 | private readonly pool: IPool<Worker, Data, Response>, | |
72 | workerChoiceStrategies: WorkerChoiceStrategy[] = [ | |
3a502712 | 73 | WorkerChoiceStrategies.ROUND_ROBIN, |
bcfb06ce JB |
74 | ], |
75 | opts?: WorkerChoiceStrategyOptions | |
76 | ) { | |
77 | this.execute = this.execute.bind(this) | |
78 | this.defaultWorkerChoiceStrategy = workerChoiceStrategies[0] | |
79 | this.workerChoiceStrategies = new Map< | |
3a502712 JB |
80 | WorkerChoiceStrategy, |
81 | IWorkerChoiceStrategy | |
bcfb06ce JB |
82 | >() |
83 | for (const workerChoiceStrategy of workerChoiceStrategies) { | |
84 | this.addWorkerChoiceStrategy(workerChoiceStrategy, this.pool, opts) | |
85 | } | |
19b8be8b JB |
86 | this.workerChoiceStrategiesPolicy = buildWorkerChoiceStrategiesPolicy( |
87 | this.workerChoiceStrategies | |
88 | ) | |
89 | this.workerChoiceStrategiesTaskStatisticsRequirements = | |
90 | buildWorkerChoiceStrategiesTaskStatisticsRequirements( | |
91 | this.workerChoiceStrategies | |
92 | ) | |
bcfb06ce | 93 | this.retriesCount = 0 |
d0bd5062 JB |
94 | this.retries = getWorkerChoiceStrategiesRetries<Worker, Data, Response>( |
95 | this.pool, | |
96 | opts | |
97 | ) | |
bcfb06ce JB |
98 | } |
99 | ||
100 | /** | |
97231086 JB |
101 | * Adds a worker choice strategy to the context. |
102 | * @param workerChoiceStrategy - The worker choice strategy to add. | |
103 | * @param pool - The pool instance. | |
bcfb06ce | 104 | * @param opts - The worker choice strategy options. |
97231086 | 105 | * @returns The worker choice strategies. |
bcfb06ce | 106 | */ |
97231086 | 107 | private addWorkerChoiceStrategy ( |
bcfb06ce | 108 | workerChoiceStrategy: WorkerChoiceStrategy, |
97231086 | 109 | pool: IPool<Worker, Data, Response>, |
bcfb06ce | 110 | opts?: WorkerChoiceStrategyOptions |
97231086 JB |
111 | ): Map<WorkerChoiceStrategy, IWorkerChoiceStrategy> { |
112 | if (!this.workerChoiceStrategies.has(workerChoiceStrategy)) { | |
113 | return this.workerChoiceStrategies.set( | |
114 | workerChoiceStrategy, | |
115 | getWorkerChoiceStrategy<Worker, Data, Response>( | |
116 | workerChoiceStrategy, | |
117 | pool, | |
118 | this, | |
119 | opts | |
120 | ) | |
121 | ) | |
85bbc7ab | 122 | } |
97231086 | 123 | return this.workerChoiceStrategies |
bcfb06ce JB |
124 | } |
125 | ||
126 | /** | |
127 | * Executes the given worker choice strategy. | |
bcfb06ce JB |
128 | * @param workerChoiceStrategy - The worker choice strategy. |
129 | * @returns The key of the worker node. | |
130 | * @throws {@link https://nodejs.org/api/errors.html#class-error} If after computed retries the worker node key is null or undefined. | |
131 | */ | |
132 | private executeStrategy (workerChoiceStrategy: IWorkerChoiceStrategy): number { | |
133 | let workerNodeKey: number | undefined | |
134 | let chooseCount = 0 | |
135 | let retriesCount = 0 | |
136 | do { | |
137 | workerNodeKey = workerChoiceStrategy.choose() | |
138 | if (workerNodeKey == null && chooseCount > 0) { | |
139 | ++retriesCount | |
140 | ++this.retriesCount | |
141 | } | |
142 | ++chooseCount | |
143 | } while (workerNodeKey == null && retriesCount < this.retries) | |
144 | if (workerNodeKey == null) { | |
145 | throw new Error( | |
6e5d7052 | 146 | `Worker node key chosen is null or undefined after ${retriesCount.toString()} retries` |
bcfb06ce JB |
147 | ) |
148 | } | |
149 | return workerNodeKey | |
150 | } | |
151 | ||
97231086 JB |
152 | /** |
153 | * Removes a worker choice strategy from the context. | |
154 | * @param workerChoiceStrategy - The worker choice strategy to remove. | |
155 | * @returns `true` if the worker choice strategy is removed, `false` otherwise. | |
156 | */ | |
157 | private removeWorkerChoiceStrategy ( | |
158 | workerChoiceStrategy: WorkerChoiceStrategy | |
159 | ): boolean { | |
160 | return this.workerChoiceStrategies.delete(workerChoiceStrategy) | |
161 | } | |
162 | ||
163 | /** | |
164 | * Executes the given worker choice strategy in the context algorithm. | |
165 | * @param workerChoiceStrategy - The worker choice strategy algorithm to execute. @defaultValue this.defaultWorkerChoiceStrategy | |
166 | * @returns The key of the worker node. | |
167 | * @throws {@link https://nodejs.org/api/errors.html#class-error} If after computed retries the worker node key is null or undefined. | |
168 | */ | |
169 | public execute ( | |
170 | workerChoiceStrategy: WorkerChoiceStrategy = this | |
171 | .defaultWorkerChoiceStrategy | |
172 | ): number { | |
173 | return this.executeStrategy( | |
174 | // eslint-disable-next-line @typescript-eslint/no-non-null-assertion | |
175 | this.workerChoiceStrategies.get(workerChoiceStrategy)! | |
176 | ) | |
177 | } | |
178 | ||
179 | /** | |
180 | * Gets the active worker choice strategies in the context policy. | |
181 | * @returns The strategies policy. | |
182 | */ | |
183 | public getPolicy (): StrategyPolicy { | |
184 | return this.workerChoiceStrategiesPolicy | |
185 | } | |
186 | ||
187 | /** | |
188 | * Gets the active worker choice strategies in the context task statistics requirements. | |
189 | * @returns The strategies task statistics requirements. | |
190 | */ | |
191 | public getTaskStatisticsRequirements (): TaskStatisticsRequirements { | |
192 | return this.workerChoiceStrategiesTaskStatisticsRequirements | |
193 | } | |
194 | ||
bcfb06ce JB |
195 | /** |
196 | * Removes the worker node key from the active worker choice strategies in the context. | |
bcfb06ce JB |
197 | * @param workerNodeKey - The worker node key. |
198 | * @returns `true` if the removal is successful, `false` otherwise. | |
199 | */ | |
200 | public remove (workerNodeKey: number): boolean { | |
63af5400 JB |
201 | return Array.from( |
202 | this.workerChoiceStrategies, | |
203 | ([_, workerChoiceStrategy]) => workerChoiceStrategy.remove(workerNodeKey) | |
204 | ).every(r => r) | |
bcfb06ce JB |
205 | } |
206 | ||
97231086 JB |
207 | /** |
208 | * Sets the default worker choice strategy to use in the context. | |
209 | * @param workerChoiceStrategy - The default worker choice strategy to set. | |
210 | * @param opts - The worker choice strategy options. | |
211 | */ | |
212 | public setDefaultWorkerChoiceStrategy ( | |
213 | workerChoiceStrategy: WorkerChoiceStrategy, | |
214 | opts?: WorkerChoiceStrategyOptions | |
215 | ): void { | |
216 | if (workerChoiceStrategy !== this.defaultWorkerChoiceStrategy) { | |
217 | this.defaultWorkerChoiceStrategy = workerChoiceStrategy | |
218 | this.addWorkerChoiceStrategy(workerChoiceStrategy, this.pool, opts) | |
219 | } | |
220 | } | |
221 | ||
bcfb06ce JB |
222 | /** |
223 | * Sets the active worker choice strategies in the context options. | |
bcfb06ce JB |
224 | * @param opts - The worker choice strategy options. |
225 | */ | |
97231086 | 226 | public setOptions (opts: undefined | WorkerChoiceStrategyOptions): void { |
bcfb06ce JB |
227 | for (const workerChoiceStrategy of this.workerChoiceStrategies.values()) { |
228 | workerChoiceStrategy.setOptions(opts) | |
229 | } | |
230 | } | |
231 | ||
85bbc7ab JB |
232 | /** |
233 | * Synchronizes the active worker choice strategies in the context with the given worker choice strategies. | |
85bbc7ab JB |
234 | * @param workerChoiceStrategies - The worker choice strategies to synchronize. |
235 | * @param opts - The worker choice strategy options. | |
236 | */ | |
237 | public syncWorkerChoiceStrategies ( | |
238 | workerChoiceStrategies: Set<WorkerChoiceStrategy>, | |
239 | opts?: WorkerChoiceStrategyOptions | |
240 | ): void { | |
241 | for (const workerChoiceStrategy of this.workerChoiceStrategies.keys()) { | |
242 | if (!workerChoiceStrategies.has(workerChoiceStrategy)) { | |
243 | this.removeWorkerChoiceStrategy(workerChoiceStrategy) | |
244 | } | |
245 | } | |
246 | for (const workerChoiceStrategy of workerChoiceStrategies) { | |
247 | if (!this.workerChoiceStrategies.has(workerChoiceStrategy)) { | |
248 | this.addWorkerChoiceStrategy(workerChoiceStrategy, this.pool, opts) | |
249 | } | |
250 | } | |
19b8be8b JB |
251 | this.workerChoiceStrategiesPolicy = buildWorkerChoiceStrategiesPolicy( |
252 | this.workerChoiceStrategies | |
253 | ) | |
254 | this.workerChoiceStrategiesTaskStatisticsRequirements = | |
255 | buildWorkerChoiceStrategiesTaskStatisticsRequirements( | |
256 | this.workerChoiceStrategies | |
257 | ) | |
85bbc7ab JB |
258 | } |
259 | ||
260 | /** | |
97231086 JB |
261 | * Updates the worker node key in the active worker choice strategies in the context internals. |
262 | * @param workerNodeKey - The worker node key. | |
263 | * @returns `true` if the update is successful, `false` otherwise. | |
85bbc7ab | 264 | */ |
97231086 JB |
265 | public update (workerNodeKey: number): boolean { |
266 | return Array.from( | |
267 | this.workerChoiceStrategies, | |
268 | ([_, workerChoiceStrategy]) => workerChoiceStrategy.update(workerNodeKey) | |
269 | ).every(r => r) | |
85bbc7ab | 270 | } |
bcfb06ce | 271 | } |