---
[poolifier.git] / src / pools / utils.ts
1 import cluster, { Worker as ClusterWorker } from 'node:cluster'
2 import { existsSync } from 'node:fs'
3 import { env } from 'node:process'
4 import {
5 SHARE_ENV,
6 Worker as ThreadWorker,
7 type WorkerOptions
8 } from 'node:worker_threads'
9
10 import type { MessageValue, Task } from '../utility-types.js'
11 import { average, isPlainObject, max, median, min } from '../utils.js'
12 import type { TasksQueueOptions } from './pool.js'
13 import {
14 type MeasurementStatisticsRequirements,
15 WorkerChoiceStrategies,
16 type WorkerChoiceStrategy
17 } from './selection-strategies/selection-strategies-types.js'
18 import type { WorkerChoiceStrategiesContext } from './selection-strategies/worker-choice-strategies-context.js'
19 import {
20 type IWorker,
21 type IWorkerNode,
22 type MeasurementStatistics,
23 type WorkerNodeOptions,
24 type WorkerType,
25 WorkerTypes,
26 type WorkerUsage
27 } from './worker.js'
28
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
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: false,
47 tasksFinishedTimeout: 2000
48 }
49 }
50
51 export const checkFilePath = (filePath: string | undefined): void => {
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 }
58 if (!existsSync(filePath)) {
59 throw new Error(`Cannot find the worker file '${filePath}'`)
60 }
61 }
62
63 export const checkDynamicPoolSize = (
64 min: number,
65 max: number | undefined
66 ): void => {
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
90 export const checkValidPriority = (priority: number | undefined): void => {
91 if (priority != null && !Number.isSafeInteger(priority)) {
92 throw new TypeError(`Invalid property 'priority': '${priority}'`)
93 }
94 if (
95 priority != null &&
96 Number.isSafeInteger(priority) &&
97 (priority < -20 || priority > 19)
98 ) {
99 throw new RangeError("Property 'priority' must be between -20 and 19")
100 }
101 }
102
103 export const checkValidWorkerChoiceStrategy = (
104 workerChoiceStrategy: WorkerChoiceStrategy | undefined
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 = (
115 tasksQueueOptions: TasksQueueOptions | undefined
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 }
150
151 export const checkWorkerNodeArguments = (
152 type: WorkerType | undefined,
153 filePath: string | undefined,
154 opts: WorkerNodeOptions | undefined
155 ): void => {
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 )
163 }
164 checkFilePath(filePath)
165 if (opts == null) {
166 throw new TypeError(
167 'Cannot construct a worker node without worker node options'
168 )
169 }
170 if (!isPlainObject(opts)) {
171 throw new TypeError(
172 'Cannot construct a worker node with invalid options: must be a plain object'
173 )
174 }
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) {
186 throw new RangeError(
187 'Cannot construct a worker node with a tasks queue back pressure size option that is not a positive integer'
188 )
189 }
190 if (opts.tasksQueueBucketSize == null) {
191 throw new TypeError(
192 'Cannot construct a worker node without a tasks queue bucket size option'
193 )
194 }
195 if (!Number.isSafeInteger(opts.tasksQueueBucketSize)) {
196 throw new TypeError(
197 'Cannot construct a worker node with a tasks queue bucket size option that is not an integer'
198 )
199 }
200 if (opts.tasksQueueBucketSize <= 0) {
201 throw new RangeError(
202 'Cannot construct a worker node with a tasks queue bucket size option that is not a positive integer'
203 )
204 }
205 }
206
207 /**
208 * Updates the given measurement statistics.
209 *
210 * @param measurementStatistics - The measurement statistics to update.
211 * @param measurementRequirements - The measurement statistics requirements.
212 * @param measurementValue - The measurement value.
213 * @internal
214 */
215 const updateMeasurementStatistics = (
216 measurementStatistics: MeasurementStatistics,
217 measurementRequirements: MeasurementStatisticsRequirements | undefined,
218 measurementValue: number | undefined
219 ): void => {
220 if (
221 measurementRequirements != null &&
222 measurementValue != null &&
223 measurementRequirements.aggregate
224 ) {
225 measurementStatistics.aggregate =
226 (measurementStatistics.aggregate ?? 0) + measurementValue
227 measurementStatistics.minimum = min(
228 measurementValue,
229 measurementStatistics.minimum ?? Infinity
230 )
231 measurementStatistics.maximum = max(
232 measurementValue,
233 measurementStatistics.maximum ?? -Infinity
234 )
235 if (measurementRequirements.average || measurementRequirements.median) {
236 measurementStatistics.history.push(measurementValue)
237 if (measurementRequirements.average) {
238 measurementStatistics.average = average(measurementStatistics.history)
239 } else if (measurementStatistics.average != null) {
240 delete measurementStatistics.average
241 }
242 if (measurementRequirements.median) {
243 measurementStatistics.median = median(measurementStatistics.history)
244 } else if (measurementStatistics.median != null) {
245 delete measurementStatistics.median
246 }
247 }
248 }
249 }
250 if (env.NODE_ENV === 'test') {
251 // eslint-disable-next-line @typescript-eslint/no-unsafe-member-access
252 exports.updateMeasurementStatistics = updateMeasurementStatistics
253 }
254
255 export const updateWaitTimeWorkerUsage = <
256 Worker extends IWorker,
257 Data = unknown,
258 Response = unknown
259 >(
260 workerChoiceStrategiesContext:
261 | WorkerChoiceStrategiesContext<Worker, Data, Response>
262 | undefined,
263 workerUsage: WorkerUsage,
264 task: Task<Data>
265 ): void => {
266 const timestamp = performance.now()
267 const taskWaitTime = timestamp - (task.timestamp ?? timestamp)
268 updateMeasurementStatistics(
269 workerUsage.waitTime,
270 workerChoiceStrategiesContext?.getTaskStatisticsRequirements().waitTime,
271 taskWaitTime
272 )
273 }
274
275 export const updateTaskStatisticsWorkerUsage = <Response = unknown>(
276 workerUsage: WorkerUsage,
277 message: MessageValue<Response>
278 ): void => {
279 const workerTaskStatistics = workerUsage.tasks
280 if (
281 // eslint-disable-next-line @typescript-eslint/no-unnecessary-condition
282 workerTaskStatistics.executing != null &&
283 workerTaskStatistics.executing > 0
284 ) {
285 --workerTaskStatistics.executing
286 }
287 if (message.workerError == null) {
288 ++workerTaskStatistics.executed
289 } else {
290 ++workerTaskStatistics.failed
291 }
292 }
293
294 export const updateRunTimeWorkerUsage = <
295 Worker extends IWorker,
296 Data = unknown,
297 Response = unknown
298 >(
299 workerChoiceStrategiesContext:
300 | WorkerChoiceStrategiesContext<Worker, Data, Response>
301 | undefined,
302 workerUsage: WorkerUsage,
303 message: MessageValue<Response>
304 ): void => {
305 if (message.workerError != null) {
306 return
307 }
308 updateMeasurementStatistics(
309 workerUsage.runTime,
310 workerChoiceStrategiesContext?.getTaskStatisticsRequirements().runTime,
311 message.taskPerformance?.runTime ?? 0
312 )
313 }
314
315 export const updateEluWorkerUsage = <
316 Worker extends IWorker,
317 Data = unknown,
318 Response = unknown
319 >(
320 workerChoiceStrategiesContext:
321 | WorkerChoiceStrategiesContext<Worker, Data, Response>
322 | undefined,
323 workerUsage: WorkerUsage,
324 message: MessageValue<Response>
325 ): void => {
326 if (message.workerError != null) {
327 return
328 }
329 const eluTaskStatisticsRequirements =
330 workerChoiceStrategiesContext?.getTaskStatisticsRequirements().elu
331 updateMeasurementStatistics(
332 workerUsage.elu.active,
333 eluTaskStatisticsRequirements,
334 message.taskPerformance?.elu?.active ?? 0
335 )
336 updateMeasurementStatistics(
337 workerUsage.elu.idle,
338 eluTaskStatisticsRequirements,
339 message.taskPerformance?.elu?.idle ?? 0
340 )
341 if (eluTaskStatisticsRequirements?.aggregate === true) {
342 if (message.taskPerformance?.elu != null) {
343 if (workerUsage.elu.utilization != null) {
344 workerUsage.elu.utilization =
345 (workerUsage.elu.utilization +
346 message.taskPerformance.elu.utilization) /
347 2
348 } else {
349 workerUsage.elu.utilization = message.taskPerformance.elu.utilization
350 }
351 }
352 }
353 }
354
355 export const createWorker = <Worker extends IWorker>(
356 type: WorkerType,
357 filePath: string,
358 opts: { env?: Record<string, unknown>, workerOptions?: WorkerOptions }
359 ): Worker => {
360 switch (type) {
361 case WorkerTypes.thread:
362 return new ThreadWorker(filePath, {
363 env: SHARE_ENV,
364 ...opts.workerOptions
365 }) as unknown as Worker
366 case WorkerTypes.cluster:
367 return cluster.fork(opts.env) as unknown as Worker
368 default:
369 // eslint-disable-next-line @typescript-eslint/restrict-template-expressions
370 throw new Error(`Unknown worker type '${type}'`)
371 }
372 }
373
374 /**
375 * Returns the worker type of the given worker.
376 *
377 * @param worker - The worker to get the type of.
378 * @returns The worker type of the given worker.
379 * @internal
380 */
381 export const getWorkerType = (worker: IWorker): WorkerType | undefined => {
382 if (worker instanceof ThreadWorker) {
383 return WorkerTypes.thread
384 } else if (worker instanceof ClusterWorker) {
385 return WorkerTypes.cluster
386 }
387 }
388
389 /**
390 * Returns the worker id of the given worker.
391 *
392 * @param worker - The worker to get the id of.
393 * @returns The worker id of the given worker.
394 * @internal
395 */
396 export const getWorkerId = (worker: IWorker): number | undefined => {
397 if (worker instanceof ThreadWorker) {
398 return worker.threadId
399 } else if (worker instanceof ClusterWorker) {
400 return worker.id
401 }
402 }
403
404 export const waitWorkerNodeEvents = async <
405 Worker extends IWorker,
406 Data = unknown
407 >(
408 workerNode: IWorkerNode<Worker, Data>,
409 workerNodeEvent: string,
410 numberOfEventsToWait: number,
411 timeout: number
412 ): Promise<number> => {
413 return await new Promise<number>(resolve => {
414 let events = 0
415 if (numberOfEventsToWait === 0) {
416 resolve(events)
417 return
418 }
419 switch (workerNodeEvent) {
420 case 'idle':
421 case 'backPressure':
422 case 'taskFinished':
423 workerNode.on(workerNodeEvent, () => {
424 ++events
425 if (events === numberOfEventsToWait) {
426 resolve(events)
427 }
428 })
429 break
430 default:
431 throw new Error('Invalid worker node event')
432 }
433 if (timeout >= 0) {
434 setTimeout(() => {
435 resolve(events)
436 }, timeout)
437 }
438 })
439 }