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