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 { average
, isPlainObject
, max
, median
, min
} from
'../utils'
6 type MeasurementStatisticsRequirements
,
7 WorkerChoiceStrategies
,
8 type WorkerChoiceStrategy
9 } from
'./selection-strategies/selection-strategies-types'
10 import type { TasksQueueOptions
} from
'./pool'
13 type MeasurementStatistics
,
14 type WorkerNodeOptions
,
19 export const checkFilePath
= (filePath
: string): void => {
20 if (filePath
== null) {
21 throw new TypeError('The worker file path must be specified')
23 if (typeof filePath
!== 'string') {
24 throw new TypeError('The worker file path must be a string')
26 if (!existsSync(filePath
)) {
27 throw new Error(`Cannot find the worker file '${filePath}'`)
31 export const checkDynamicPoolSize
= (min
: number, max
: number): void => {
34 'Cannot instantiate a dynamic pool without specifying the maximum pool size'
36 } else if (!Number.isSafeInteger(max
)) {
38 'Cannot instantiate a dynamic pool with a non safe integer maximum pool size'
40 } else if (min
> max
) {
42 'Cannot instantiate a dynamic pool with a maximum pool size inferior to the minimum pool size'
44 } else if (max
=== 0) {
46 'Cannot instantiate a dynamic pool with a maximum pool size equal to zero'
48 } else if (min
=== max
) {
50 'Cannot instantiate a dynamic pool with a minimum pool size equal to the maximum pool size. Use a fixed pool instead'
55 export const checkValidWorkerChoiceStrategy
= (
56 workerChoiceStrategy
: WorkerChoiceStrategy
59 workerChoiceStrategy
!= null &&
60 !Object.values(WorkerChoiceStrategies
).includes(workerChoiceStrategy
)
62 throw new Error(`Invalid worker choice strategy '${workerChoiceStrategy}'`)
66 export const checkValidTasksQueueOptions
= (
67 tasksQueueOptions
: TasksQueueOptions
69 if (tasksQueueOptions
!= null && !isPlainObject(tasksQueueOptions
)) {
70 throw new TypeError('Invalid tasks queue options: must be a plain object')
73 tasksQueueOptions
?.concurrency
!= null &&
74 !Number.isSafeInteger(tasksQueueOptions
.concurrency
)
77 'Invalid worker node tasks concurrency: must be an integer'
81 tasksQueueOptions
?.concurrency
!= null &&
82 tasksQueueOptions
.concurrency
<= 0
85 `Invalid worker node tasks concurrency: ${tasksQueueOptions.concurrency} is a negative integer or zero`
89 tasksQueueOptions
?.size
!= null &&
90 !Number.isSafeInteger(tasksQueueOptions
.size
)
93 'Invalid worker node tasks queue size: must be an integer'
96 if (tasksQueueOptions
?.size
!= null && tasksQueueOptions
.size
<= 0) {
98 `Invalid worker node tasks queue size: ${tasksQueueOptions.size} is a negative integer or zero`
103 export const checkWorkerNodeArguments
= (
106 opts
: WorkerNodeOptions
109 throw new TypeError('Cannot construct a worker node without a worker type')
111 if (!Object.values(WorkerTypes
).includes(type)) {
113 `Cannot construct a worker node with an invalid worker type '${type}'`
116 checkFilePath(filePath
)
119 'Cannot construct a worker node without worker node options'
122 if (!isPlainObject(opts
)) {
124 'Cannot construct a worker node with invalid options: must be a plain object'
127 if (opts
.tasksQueueBackPressureSize
== null) {
129 'Cannot construct a worker node without a tasks queue back pressure size option'
132 if (!Number.isSafeInteger(opts
.tasksQueueBackPressureSize
)) {
134 'Cannot construct a worker node with a tasks queue back pressure size option that is not an integer'
137 if (opts
.tasksQueueBackPressureSize
<= 0) {
138 throw new RangeError(
139 'Cannot construct a worker node with a tasks queue back pressure size option that is not a positive integer'
145 * Updates the given measurement statistics.
147 * @param measurementStatistics - The measurement statistics to update.
148 * @param measurementRequirements - The measurement statistics requirements.
149 * @param measurementValue - The measurement value.
150 * @param numberOfMeasurements - The number of measurements.
153 export const updateMeasurementStatistics
= (
154 measurementStatistics
: MeasurementStatistics
,
155 measurementRequirements
: MeasurementStatisticsRequirements
,
156 measurementValue
: number
158 if (measurementRequirements
.aggregate
) {
159 measurementStatistics
.aggregate
=
160 (measurementStatistics
.aggregate
?? 0) + measurementValue
161 measurementStatistics
.minimum
= min(
163 measurementStatistics
.minimum
?? Infinity
165 measurementStatistics
.maximum
= max(
167 measurementStatistics
.maximum
?? -Infinity
170 (measurementRequirements
.average
|| measurementRequirements
.median
) &&
171 measurementValue
!= null
173 measurementStatistics
.history
.push(measurementValue
)
174 if (measurementRequirements
.average
) {
175 measurementStatistics
.average
= average(measurementStatistics
.history
)
176 } else if (measurementStatistics
.average
!= null) {
177 delete measurementStatistics
.average
179 if (measurementRequirements
.median
) {
180 measurementStatistics
.median
= median(measurementStatistics
.history
)
181 } else if (measurementStatistics
.median
!= null) {
182 delete measurementStatistics
.median
188 export const createWorker
= <Worker
extends IWorker
>(
191 opts
: { env
?: Record
<string, unknown
>, workerOptions
?: WorkerOptions
}
194 case WorkerTypes
.thread
:
195 return new Worker(filePath
, {
197 ...opts
?.workerOptions
198 }) as unknown
as Worker
199 case WorkerTypes
.cluster
:
200 return cluster
.fork(opts
?.env
) as unknown
as Worker
202 // eslint-disable-next-line @typescript-eslint/restrict-template-expressions
203 throw new Error(`Unknown worker type '${type}'`)