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