1 import { existsSync
} from
'node:fs'
2 import cluster from
'node:cluster'
3 import { SHARE_ENV
, Worker
, type WorkerOptions
} from
'node:worker_threads'
4 import { env
} from
'node:process'
5 import { average
, isPlainObject
, max
, median
, min
} from
'../utils.js'
6 import type { MessageValue
, Task
} from
'../utility-types.js'
8 type MeasurementStatisticsRequirements
,
9 WorkerChoiceStrategies
,
10 type WorkerChoiceStrategy
11 } from
'./selection-strategies/selection-strategies-types.js'
12 import type { TasksQueueOptions
} from
'./pool.js'
16 type MeasurementStatistics
,
17 type WorkerNodeOptions
,
22 import type { WorkerChoiceStrategyContext
} from
'./selection-strategies/worker-choice-strategy-context.js'
24 export const getDefaultTasksQueueOptions
= (
26 ): Required
<TasksQueueOptions
> => {
28 size
: Math.pow(poolMaxSize
, 2),
31 tasksStealingOnBackPressure
: true,
32 tasksFinishedTimeout
: 2000
36 export const checkFilePath
= (filePath
: string): void => {
37 if (filePath
== null) {
38 throw new TypeError('The worker file path must be specified')
40 if (typeof filePath
!== 'string') {
41 throw new TypeError('The worker file path must be a string')
43 if (!existsSync(filePath
)) {
44 throw new Error(`Cannot find the worker file '${filePath}'`)
48 export const checkDynamicPoolSize
= (min
: number, max
: number): void => {
51 'Cannot instantiate a dynamic pool without specifying the maximum pool size'
53 } else if (!Number.isSafeInteger(max
)) {
55 'Cannot instantiate a dynamic pool with a non safe integer maximum pool size'
57 } else if (min
> max
) {
59 'Cannot instantiate a dynamic pool with a maximum pool size inferior to the minimum pool size'
61 } else if (max
=== 0) {
63 'Cannot instantiate a dynamic pool with a maximum pool size equal to zero'
65 } else if (min
=== max
) {
67 'Cannot instantiate a dynamic pool with a minimum pool size equal to the maximum pool size. Use a fixed pool instead'
72 export const checkValidWorkerChoiceStrategy
= (
73 workerChoiceStrategy
: WorkerChoiceStrategy
76 workerChoiceStrategy
!= null &&
77 !Object.values(WorkerChoiceStrategies
).includes(workerChoiceStrategy
)
79 throw new Error(`Invalid worker choice strategy '${workerChoiceStrategy}'`)
83 export const checkValidTasksQueueOptions
= (
84 tasksQueueOptions
: TasksQueueOptions
86 if (tasksQueueOptions
!= null && !isPlainObject(tasksQueueOptions
)) {
87 throw new TypeError('Invalid tasks queue options: must be a plain object')
90 tasksQueueOptions
?.concurrency
!= null &&
91 !Number.isSafeInteger(tasksQueueOptions
.concurrency
)
94 'Invalid worker node tasks concurrency: must be an integer'
98 tasksQueueOptions
?.concurrency
!= null &&
99 tasksQueueOptions
.concurrency
<= 0
101 throw new RangeError(
102 `Invalid worker node tasks concurrency: ${tasksQueueOptions.concurrency} is a negative integer or zero`
106 tasksQueueOptions
?.size
!= null &&
107 !Number.isSafeInteger(tasksQueueOptions
.size
)
110 'Invalid worker node tasks queue size: must be an integer'
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`
120 export const checkWorkerNodeArguments
= (
123 opts
: WorkerNodeOptions
126 throw new TypeError('Cannot construct a worker node without a worker type')
128 if (!Object.values(WorkerTypes
).includes(type)) {
130 `Cannot construct a worker node with an invalid worker type '${type}'`
133 checkFilePath(filePath
)
136 'Cannot construct a worker node without worker node options'
139 if (!isPlainObject(opts
)) {
141 'Cannot construct a worker node with invalid options: must be a plain object'
144 if (opts
.tasksQueueBackPressureSize
== null) {
146 'Cannot construct a worker node without a tasks queue back pressure size option'
149 if (!Number.isSafeInteger(opts
.tasksQueueBackPressureSize
)) {
151 'Cannot construct a worker node with a tasks queue back pressure size option that is not an integer'
154 if (opts
.tasksQueueBackPressureSize
<= 0) {
155 throw new RangeError(
156 'Cannot construct a worker node with a tasks queue back pressure size option that is not a positive integer'
162 * Updates the given measurement statistics.
164 * @param measurementStatistics - The measurement statistics to update.
165 * @param measurementRequirements - The measurement statistics requirements.
166 * @param measurementValue - The measurement value.
169 const updateMeasurementStatistics
= (
170 measurementStatistics
: MeasurementStatistics
,
171 measurementRequirements
: MeasurementStatisticsRequirements
,
172 measurementValue
: number
174 if (measurementRequirements
.aggregate
) {
175 measurementStatistics
.aggregate
=
176 (measurementStatistics
.aggregate
?? 0) + measurementValue
177 measurementStatistics
.minimum
= min(
179 measurementStatistics
.minimum
?? Infinity
181 measurementStatistics
.maximum
= max(
183 measurementStatistics
.maximum
?? -Infinity
186 (measurementRequirements
.average
|| measurementRequirements
.median
) &&
187 measurementValue
!= null
189 measurementStatistics
.history
.push(measurementValue
)
190 if (measurementRequirements
.average
) {
191 measurementStatistics
.average
= average(measurementStatistics
.history
)
192 } else if (measurementStatistics
.average
!= null) {
193 delete measurementStatistics
.average
195 if (measurementRequirements
.median
) {
196 measurementStatistics
.median
= median(measurementStatistics
.history
)
197 } else if (measurementStatistics
.median
!= null) {
198 delete measurementStatistics
.median
203 if (env
.NODE_ENV
=== 'test') {
204 // eslint-disable-next-line @typescript-eslint/no-unsafe-member-access
205 exports
.updateMeasurementStatistics
= updateMeasurementStatistics
208 export const updateWaitTimeWorkerUsage
= <
209 Worker
extends IWorker
,
213 workerChoiceStrategyContext
: WorkerChoiceStrategyContext
<
218 workerUsage
: WorkerUsage
,
221 const timestamp
= performance
.now()
222 const taskWaitTime
= timestamp
- (task
.timestamp
?? timestamp
)
223 updateMeasurementStatistics(
224 workerUsage
.waitTime
,
225 workerChoiceStrategyContext
.getTaskStatisticsRequirements().waitTime
,
230 export const updateTaskStatisticsWorkerUsage
= <Response
= unknown
>(
231 workerUsage
: WorkerUsage
,
232 message
: MessageValue
<Response
>
234 const workerTaskStatistics
= workerUsage
.tasks
236 workerTaskStatistics
.executing
!= null &&
237 workerTaskStatistics
.executing
> 0
239 --workerTaskStatistics
.executing
241 if (message
.workerError
== null) {
242 ++workerTaskStatistics
.executed
244 ++workerTaskStatistics
.failed
248 export const updateRunTimeWorkerUsage
= <
249 Worker
extends IWorker
,
253 workerChoiceStrategyContext
: WorkerChoiceStrategyContext
<
258 workerUsage
: WorkerUsage
,
259 message
: MessageValue
<Response
>
261 if (message
.workerError
!= null) {
264 updateMeasurementStatistics(
266 workerChoiceStrategyContext
.getTaskStatisticsRequirements().runTime
,
267 message
.taskPerformance
?.runTime
?? 0
271 export const updateEluWorkerUsage
= <
272 Worker
extends IWorker
,
276 workerChoiceStrategyContext
: WorkerChoiceStrategyContext
<
281 workerUsage
: WorkerUsage
,
282 message
: MessageValue
<Response
>
284 if (message
.workerError
!= null) {
287 const eluTaskStatisticsRequirements
: MeasurementStatisticsRequirements
=
288 workerChoiceStrategyContext
.getTaskStatisticsRequirements().elu
289 updateMeasurementStatistics(
290 workerUsage
.elu
.active
,
291 eluTaskStatisticsRequirements
,
292 message
.taskPerformance
?.elu
?.active
?? 0
294 updateMeasurementStatistics(
295 workerUsage
.elu
.idle
,
296 eluTaskStatisticsRequirements
,
297 message
.taskPerformance
?.elu
?.idle
?? 0
299 if (eluTaskStatisticsRequirements
.aggregate
) {
300 if (message
.taskPerformance
?.elu
!= null) {
301 if (workerUsage
.elu
.utilization
!= null) {
302 workerUsage
.elu
.utilization
=
303 (workerUsage
.elu
.utilization
+
304 message
.taskPerformance
.elu
.utilization
) /
307 workerUsage
.elu
.utilization
= message
.taskPerformance
.elu
.utilization
313 export const createWorker
= <Worker
extends IWorker
>(
316 opts
: { env
?: Record
<string, unknown
>, workerOptions
?: WorkerOptions
}
319 case WorkerTypes
.thread
:
320 return new Worker(filePath
, {
322 ...opts
?.workerOptions
323 }) as unknown
as Worker
324 case WorkerTypes
.cluster
:
325 return cluster
.fork(opts
?.env
) as unknown
as Worker
327 // eslint-disable-next-line @typescript-eslint/restrict-template-expressions
328 throw new Error(`Unknown worker type '${type}'`)
332 export const waitWorkerNodeEvents
= async <
333 Worker
extends IWorker
,
336 workerNode
: IWorkerNode
<Worker
, Data
>,
337 workerNodeEvent
: string,
338 numberOfEventsToWait
: number,
340 ): Promise
<number> => {
341 return await new Promise
<number>(resolve
=> {
343 if (numberOfEventsToWait
=== 0) {
347 workerNode
.on(workerNodeEvent
, () => {
349 if (events
=== numberOfEventsToWait
) {