Commit | Line | Data |
---|---|---|
bde6b5d7 | 1 | import { existsSync } from 'node:fs' |
c3719753 JB |
2 | import cluster from 'node:cluster' |
3 | import { SHARE_ENV, Worker, type WorkerOptions } from 'node:worker_threads' | |
c329fd41 | 4 | import { env } from 'node:process' |
bfc75cca | 5 | import { average, isPlainObject, max, median, min } from '../utils' |
c329fd41 | 6 | import type { MessageValue, Task } from '../utility-types' |
bde6b5d7 | 7 | import { |
bfc75cca | 8 | type MeasurementStatisticsRequirements, |
bde6b5d7 JB |
9 | WorkerChoiceStrategies, |
10 | type WorkerChoiceStrategy | |
11 | } from './selection-strategies/selection-strategies-types' | |
12 | import type { TasksQueueOptions } from './pool' | |
c3719753 JB |
13 | import { |
14 | type IWorker, | |
d41a44de | 15 | type IWorkerNode, |
c3719753 JB |
16 | type MeasurementStatistics, |
17 | type WorkerNodeOptions, | |
18 | type WorkerType, | |
c329fd41 JB |
19 | WorkerTypes, |
20 | type WorkerUsage | |
c3719753 | 21 | } from './worker' |
c329fd41 | 22 | import type { WorkerChoiceStrategyContext } from './selection-strategies/worker-choice-strategy-context' |
bde6b5d7 | 23 | |
32b141fd JB |
24 | export const getDefaultTasksQueueOptions = ( |
25 | poolMaxSize: number | |
26 | ): Required<TasksQueueOptions> => { | |
27 | return { | |
28 | size: Math.pow(poolMaxSize, 2), | |
29 | concurrency: 1, | |
30 | taskStealing: true, | |
31 | tasksStealingOnBackPressure: true, | |
568d0075 | 32 | tasksFinishedTimeout: 2000 |
32b141fd JB |
33 | } |
34 | } | |
35 | ||
bde6b5d7 | 36 | export const checkFilePath = (filePath: string): void => { |
c3719753 JB |
37 | if (filePath == null) { |
38 | throw new TypeError('The worker file path must be specified') | |
39 | } | |
40 | if (typeof filePath !== 'string') { | |
41 | throw new TypeError('The worker file path must be a string') | |
42 | } | |
bde6b5d7 JB |
43 | if (!existsSync(filePath)) { |
44 | throw new Error(`Cannot find the worker file '${filePath}'`) | |
45 | } | |
46 | } | |
47 | ||
48 | export const checkDynamicPoolSize = (min: number, max: number): void => { | |
49 | if (max == null) { | |
50 | throw new TypeError( | |
51 | 'Cannot instantiate a dynamic pool without specifying the maximum pool size' | |
52 | ) | |
53 | } else if (!Number.isSafeInteger(max)) { | |
54 | throw new TypeError( | |
55 | 'Cannot instantiate a dynamic pool with a non safe integer maximum pool size' | |
56 | ) | |
57 | } else if (min > max) { | |
58 | throw new RangeError( | |
59 | 'Cannot instantiate a dynamic pool with a maximum pool size inferior to the minimum pool size' | |
60 | ) | |
61 | } else if (max === 0) { | |
62 | throw new RangeError( | |
63 | 'Cannot instantiate a dynamic pool with a maximum pool size equal to zero' | |
64 | ) | |
65 | } else if (min === max) { | |
66 | throw new RangeError( | |
67 | 'Cannot instantiate a dynamic pool with a minimum pool size equal to the maximum pool size. Use a fixed pool instead' | |
68 | ) | |
69 | } | |
70 | } | |
71 | ||
72 | export const checkValidWorkerChoiceStrategy = ( | |
73 | workerChoiceStrategy: WorkerChoiceStrategy | |
74 | ): void => { | |
75 | if ( | |
76 | workerChoiceStrategy != null && | |
77 | !Object.values(WorkerChoiceStrategies).includes(workerChoiceStrategy) | |
78 | ) { | |
79 | throw new Error(`Invalid worker choice strategy '${workerChoiceStrategy}'`) | |
80 | } | |
81 | } | |
82 | ||
83 | export const checkValidTasksQueueOptions = ( | |
84 | tasksQueueOptions: TasksQueueOptions | |
85 | ): void => { | |
86 | if (tasksQueueOptions != null && !isPlainObject(tasksQueueOptions)) { | |
87 | throw new TypeError('Invalid tasks queue options: must be a plain object') | |
88 | } | |
89 | if ( | |
90 | tasksQueueOptions?.concurrency != null && | |
91 | !Number.isSafeInteger(tasksQueueOptions.concurrency) | |
92 | ) { | |
93 | throw new TypeError( | |
94 | 'Invalid worker node tasks concurrency: must be an integer' | |
95 | ) | |
96 | } | |
97 | if ( | |
98 | tasksQueueOptions?.concurrency != null && | |
99 | tasksQueueOptions.concurrency <= 0 | |
100 | ) { | |
101 | throw new RangeError( | |
102 | `Invalid worker node tasks concurrency: ${tasksQueueOptions.concurrency} is a negative integer or zero` | |
103 | ) | |
104 | } | |
105 | if ( | |
106 | tasksQueueOptions?.size != null && | |
107 | !Number.isSafeInteger(tasksQueueOptions.size) | |
108 | ) { | |
109 | throw new TypeError( | |
110 | 'Invalid worker node tasks queue size: must be an integer' | |
111 | ) | |
112 | } | |
113 | if (tasksQueueOptions?.size != null && tasksQueueOptions.size <= 0) { | |
114 | throw new RangeError( | |
115 | `Invalid worker node tasks queue size: ${tasksQueueOptions.size} is a negative integer or zero` | |
116 | ) | |
117 | } | |
118 | } | |
bfc75cca | 119 | |
c3719753 JB |
120 | export const checkWorkerNodeArguments = ( |
121 | type: WorkerType, | |
122 | filePath: string, | |
123 | opts: WorkerNodeOptions | |
9a38f99e | 124 | ): void => { |
c3719753 JB |
125 | if (type == null) { |
126 | throw new TypeError('Cannot construct a worker node without a worker type') | |
127 | } | |
128 | if (!Object.values(WorkerTypes).includes(type)) { | |
129 | throw new TypeError( | |
130 | `Cannot construct a worker node with an invalid worker type '${type}'` | |
131 | ) | |
9a38f99e | 132 | } |
c3719753 JB |
133 | checkFilePath(filePath) |
134 | if (opts == null) { | |
9a38f99e | 135 | throw new TypeError( |
c3719753 | 136 | 'Cannot construct a worker node without worker node options' |
9a38f99e JB |
137 | ) |
138 | } | |
9974369e | 139 | if (!isPlainObject(opts)) { |
9a38f99e | 140 | throw new TypeError( |
c3719753 | 141 | 'Cannot construct a worker node with invalid options: must be a plain object' |
9a38f99e JB |
142 | ) |
143 | } | |
c3719753 JB |
144 | if (opts.tasksQueueBackPressureSize == null) { |
145 | throw new TypeError( | |
146 | 'Cannot construct a worker node without a tasks queue back pressure size option' | |
147 | ) | |
148 | } | |
149 | if (!Number.isSafeInteger(opts.tasksQueueBackPressureSize)) { | |
150 | throw new TypeError( | |
151 | 'Cannot construct a worker node with a tasks queue back pressure size option that is not an integer' | |
152 | ) | |
153 | } | |
154 | if (opts.tasksQueueBackPressureSize <= 0) { | |
9a38f99e | 155 | throw new RangeError( |
c3719753 | 156 | 'Cannot construct a worker node with a tasks queue back pressure size option that is not a positive integer' |
9a38f99e JB |
157 | ) |
158 | } | |
159 | } | |
bfc75cca JB |
160 | |
161 | /** | |
162 | * Updates the given measurement statistics. | |
163 | * | |
164 | * @param measurementStatistics - The measurement statistics to update. | |
165 | * @param measurementRequirements - The measurement statistics requirements. | |
166 | * @param measurementValue - The measurement value. | |
167 | * @param numberOfMeasurements - The number of measurements. | |
168 | * @internal | |
169 | */ | |
c329fd41 | 170 | const updateMeasurementStatistics = ( |
bfc75cca JB |
171 | measurementStatistics: MeasurementStatistics, |
172 | measurementRequirements: MeasurementStatisticsRequirements, | |
173 | measurementValue: number | |
174 | ): void => { | |
175 | if (measurementRequirements.aggregate) { | |
176 | measurementStatistics.aggregate = | |
177 | (measurementStatistics.aggregate ?? 0) + measurementValue | |
178 | measurementStatistics.minimum = min( | |
179 | measurementValue, | |
180 | measurementStatistics.minimum ?? Infinity | |
181 | ) | |
182 | measurementStatistics.maximum = max( | |
183 | measurementValue, | |
184 | measurementStatistics.maximum ?? -Infinity | |
185 | ) | |
186 | if ( | |
187 | (measurementRequirements.average || measurementRequirements.median) && | |
188 | measurementValue != null | |
189 | ) { | |
190 | measurementStatistics.history.push(measurementValue) | |
191 | if (measurementRequirements.average) { | |
192 | measurementStatistics.average = average(measurementStatistics.history) | |
193 | } else if (measurementStatistics.average != null) { | |
194 | delete measurementStatistics.average | |
195 | } | |
196 | if (measurementRequirements.median) { | |
197 | measurementStatistics.median = median(measurementStatistics.history) | |
198 | } else if (measurementStatistics.median != null) { | |
199 | delete measurementStatistics.median | |
200 | } | |
201 | } | |
202 | } | |
203 | } | |
c329fd41 JB |
204 | if (env.NODE_ENV === 'test') { |
205 | // eslint-disable-next-line @typescript-eslint/no-unsafe-member-access | |
206 | exports.updateMeasurementStatistics = updateMeasurementStatistics | |
207 | } | |
208 | ||
209 | export const updateWaitTimeWorkerUsage = < | |
210 | Worker extends IWorker, | |
211 | Data = unknown, | |
212 | Response = unknown | |
213 | >( | |
214 | workerChoiceStrategyContext: WorkerChoiceStrategyContext< | |
215 | Worker, | |
216 | Data, | |
217 | Response | |
218 | >, | |
219 | workerUsage: WorkerUsage, | |
220 | task: Task<Data> | |
221 | ): void => { | |
222 | const timestamp = performance.now() | |
223 | const taskWaitTime = timestamp - (task.timestamp ?? timestamp) | |
224 | updateMeasurementStatistics( | |
225 | workerUsage.waitTime, | |
226 | workerChoiceStrategyContext.getTaskStatisticsRequirements().waitTime, | |
227 | taskWaitTime | |
228 | ) | |
229 | } | |
230 | ||
231 | export const updateTaskStatisticsWorkerUsage = <Response = unknown>( | |
232 | workerUsage: WorkerUsage, | |
233 | message: MessageValue<Response> | |
234 | ): void => { | |
235 | const workerTaskStatistics = workerUsage.tasks | |
236 | if ( | |
237 | workerTaskStatistics.executing != null && | |
238 | workerTaskStatistics.executing > 0 | |
239 | ) { | |
240 | --workerTaskStatistics.executing | |
241 | } | |
242 | if (message.workerError == null) { | |
243 | ++workerTaskStatistics.executed | |
244 | } else { | |
245 | ++workerTaskStatistics.failed | |
246 | } | |
247 | } | |
248 | ||
249 | export const updateRunTimeWorkerUsage = < | |
250 | Worker extends IWorker, | |
251 | Data = unknown, | |
252 | Response = unknown | |
253 | >( | |
254 | workerChoiceStrategyContext: WorkerChoiceStrategyContext< | |
255 | Worker, | |
256 | Data, | |
257 | Response | |
258 | >, | |
259 | workerUsage: WorkerUsage, | |
260 | message: MessageValue<Response> | |
261 | ): void => { | |
262 | if (message.workerError != null) { | |
263 | return | |
264 | } | |
265 | updateMeasurementStatistics( | |
266 | workerUsage.runTime, | |
267 | workerChoiceStrategyContext.getTaskStatisticsRequirements().runTime, | |
268 | message.taskPerformance?.runTime ?? 0 | |
269 | ) | |
270 | } | |
271 | ||
272 | export const updateEluWorkerUsage = < | |
273 | Worker extends IWorker, | |
274 | Data = unknown, | |
275 | Response = unknown | |
276 | >( | |
277 | workerChoiceStrategyContext: WorkerChoiceStrategyContext< | |
278 | Worker, | |
279 | Data, | |
280 | Response | |
281 | >, | |
282 | workerUsage: WorkerUsage, | |
283 | message: MessageValue<Response> | |
284 | ): void => { | |
285 | if (message.workerError != null) { | |
286 | return | |
287 | } | |
288 | const eluTaskStatisticsRequirements: MeasurementStatisticsRequirements = | |
289 | workerChoiceStrategyContext.getTaskStatisticsRequirements().elu | |
290 | updateMeasurementStatistics( | |
291 | workerUsage.elu.active, | |
292 | eluTaskStatisticsRequirements, | |
293 | message.taskPerformance?.elu?.active ?? 0 | |
294 | ) | |
295 | updateMeasurementStatistics( | |
296 | workerUsage.elu.idle, | |
297 | eluTaskStatisticsRequirements, | |
298 | message.taskPerformance?.elu?.idle ?? 0 | |
299 | ) | |
300 | if (eluTaskStatisticsRequirements.aggregate) { | |
301 | if (message.taskPerformance?.elu != null) { | |
302 | if (workerUsage.elu.utilization != null) { | |
303 | workerUsage.elu.utilization = | |
304 | (workerUsage.elu.utilization + | |
305 | message.taskPerformance.elu.utilization) / | |
306 | 2 | |
307 | } else { | |
308 | workerUsage.elu.utilization = message.taskPerformance.elu.utilization | |
309 | } | |
310 | } | |
311 | } | |
312 | } | |
c3719753 JB |
313 | |
314 | export const createWorker = <Worker extends IWorker>( | |
315 | type: WorkerType, | |
316 | filePath: string, | |
317 | opts: { env?: Record<string, unknown>, workerOptions?: WorkerOptions } | |
318 | ): Worker => { | |
319 | switch (type) { | |
320 | case WorkerTypes.thread: | |
321 | return new Worker(filePath, { | |
322 | env: SHARE_ENV, | |
323 | ...opts?.workerOptions | |
324 | }) as unknown as Worker | |
325 | case WorkerTypes.cluster: | |
326 | return cluster.fork(opts?.env) as unknown as Worker | |
327 | default: | |
328 | // eslint-disable-next-line @typescript-eslint/restrict-template-expressions | |
329 | throw new Error(`Unknown worker type '${type}'`) | |
330 | } | |
331 | } | |
d41a44de JB |
332 | |
333 | export const waitWorkerNodeEvents = async < | |
334 | Worker extends IWorker, | |
335 | Data = unknown | |
336 | >( | |
337 | workerNode: IWorkerNode<Worker, Data>, | |
338 | workerNodeEvent: string, | |
32b141fd JB |
339 | numberOfEventsToWait: number, |
340 | timeout: number | |
d41a44de JB |
341 | ): Promise<number> => { |
342 | return await new Promise<number>(resolve => { | |
343 | let events = 0 | |
344 | if (numberOfEventsToWait === 0) { | |
345 | resolve(events) | |
346 | return | |
347 | } | |
348 | workerNode.on(workerNodeEvent, () => { | |
349 | ++events | |
350 | if (events === numberOfEventsToWait) { | |
351 | resolve(events) | |
352 | } | |
353 | }) | |
6f3a391b | 354 | if (timeout >= 0) { |
32b141fd JB |
355 | setTimeout(() => { |
356 | resolve(events) | |
357 | }, timeout) | |
358 | } | |
d41a44de JB |
359 | }) |
360 | } |