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