X-Git-Url: https://git.piment-noir.org/?a=blobdiff_plain;ds=sidebyside;f=src%2Fpools%2Futils.ts;h=24f2cfb2a6d075b2133473bef04aa9a650607c5b;hb=6e5d7052fe741b50e68f8614b33d3754be41415f;hp=a97f3e6e121d1cb3e366aa277a4bcbd43b6d6429;hpb=f4d0a470d13364acc493b877f7e51a3cbb8c29ed;p=poolifier.git diff --git a/src/pools/utils.ts b/src/pools/utils.ts index a97f3e6e..24f2cfb2 100644 --- a/src/pools/utils.ts +++ b/src/pools/utils.ts @@ -1,15 +1,21 @@ +import cluster, { Worker as ClusterWorker } from 'node:cluster' import { existsSync } from 'node:fs' -import cluster from 'node:cluster' -import { SHARE_ENV, Worker, type WorkerOptions } from 'node:worker_threads' import { env } from 'node:process' -import { average, isPlainObject, max, median, min } from '../utils.js' +import { + SHARE_ENV, + Worker as ThreadWorker, + type WorkerOptions, +} from 'node:worker_threads' + import type { MessageValue, Task } from '../utility-types.js' +import { average, isPlainObject, max, median, min } from '../utils.js' +import type { TasksQueueOptions } from './pool.js' import { type MeasurementStatisticsRequirements, WorkerChoiceStrategies, - type WorkerChoiceStrategy + type WorkerChoiceStrategy, } from './selection-strategies/selection-strategies-types.js' -import type { TasksQueueOptions } from './pool.js' +import type { WorkerChoiceStrategiesContext } from './selection-strategies/worker-choice-strategies-context.js' import { type IWorker, type IWorkerNode, @@ -17,9 +23,18 @@ import { type WorkerNodeOptions, type WorkerType, WorkerTypes, - type WorkerUsage + type WorkerUsage, } from './worker.js' -import type { WorkerChoiceStrategyContext } from './selection-strategies/worker-choice-strategy-context.js' + +/** + * Default measurement statistics requirements. + */ +export const DEFAULT_MEASUREMENT_STATISTICS_REQUIREMENTS: MeasurementStatisticsRequirements = + { + aggregate: false, + average: false, + median: false, + } export const getDefaultTasksQueueOptions = ( poolMaxSize: number @@ -28,8 +43,8 @@ export const getDefaultTasksQueueOptions = ( size: Math.pow(poolMaxSize, 2), concurrency: 1, taskStealing: true, - tasksStealingOnBackPressure: true, - tasksFinishedTimeout: 2000 + tasksStealingOnBackPressure: false, + tasksFinishedTimeout: 2000, } } @@ -72,6 +87,19 @@ export const checkDynamicPoolSize = ( } } +export const checkValidPriority = (priority: number | undefined): void => { + if (priority != null && !Number.isSafeInteger(priority)) { + throw new TypeError(`Invalid property 'priority': '${priority.toString()}'`) + } + if ( + priority != null && + Number.isSafeInteger(priority) && + (priority < -20 || priority > 19) + ) { + throw new RangeError("Property 'priority' must be between -20 and 19") + } +} + export const checkValidWorkerChoiceStrategy = ( workerChoiceStrategy: WorkerChoiceStrategy | undefined ): void => { @@ -102,7 +130,7 @@ export const checkValidTasksQueueOptions = ( tasksQueueOptions.concurrency <= 0 ) { throw new RangeError( - `Invalid worker node tasks concurrency: ${tasksQueueOptions.concurrency} is a negative integer or zero` + `Invalid worker node tasks concurrency: ${tasksQueueOptions.concurrency.toString()} is a negative integer or zero` ) } if ( @@ -115,7 +143,7 @@ export const checkValidTasksQueueOptions = ( } if (tasksQueueOptions?.size != null && tasksQueueOptions.size <= 0) { throw new RangeError( - `Invalid worker node tasks queue size: ${tasksQueueOptions.size} is a negative integer or zero` + `Invalid worker node tasks queue size: ${tasksQueueOptions.size.toString()} is a negative integer or zero` ) } } @@ -141,7 +169,7 @@ export const checkWorkerNodeArguments = ( } if (!isPlainObject(opts)) { throw new TypeError( - 'Cannot construct a worker node with invalid options: must be a plain object' + 'Cannot construct a worker node with invalid worker node options: must be a plain object' ) } if (opts.tasksQueueBackPressureSize == null) { @@ -159,11 +187,35 @@ export const checkWorkerNodeArguments = ( 'Cannot construct a worker node with a tasks queue back pressure size option that is not a positive integer' ) } + if (opts.tasksQueueBucketSize == null) { + throw new TypeError( + 'Cannot construct a worker node without a tasks queue bucket size option' + ) + } + if (!Number.isSafeInteger(opts.tasksQueueBucketSize)) { + throw new TypeError( + 'Cannot construct a worker node with a tasks queue bucket size option that is not an integer' + ) + } + if (opts.tasksQueueBucketSize <= 0) { + throw new RangeError( + 'Cannot construct a worker node with a tasks queue bucket size option that is not a positive integer' + ) + } + if (opts.tasksQueuePriority == null) { + throw new TypeError( + 'Cannot construct a worker node without a tasks queue priority option' + ) + } + if (typeof opts.tasksQueuePriority !== 'boolean') { + throw new TypeError( + 'Cannot construct a worker node with a tasks queue priority option that is not a boolean' + ) + } } /** * Updates the given measurement statistics. - * * @param measurementStatistics - The measurement statistics to update. * @param measurementRequirements - The measurement statistics requirements. * @param measurementValue - The measurement value. @@ -183,21 +235,25 @@ const updateMeasurementStatistics = ( (measurementStatistics.aggregate ?? 0) + measurementValue measurementStatistics.minimum = min( measurementValue, - measurementStatistics.minimum ?? Infinity + measurementStatistics.minimum ?? Number.POSITIVE_INFINITY ) measurementStatistics.maximum = max( measurementValue, - measurementStatistics.maximum ?? -Infinity + measurementStatistics.maximum ?? Number.NEGATIVE_INFINITY ) if (measurementRequirements.average || measurementRequirements.median) { - measurementStatistics.history.push(measurementValue) + measurementStatistics.history.put(measurementValue) if (measurementRequirements.average) { - measurementStatistics.average = average(measurementStatistics.history) + measurementStatistics.average = average( + measurementStatistics.history.toArray() + ) } else if (measurementStatistics.average != null) { delete measurementStatistics.average } if (measurementRequirements.median) { - measurementStatistics.median = median(measurementStatistics.history) + measurementStatistics.median = median( + measurementStatistics.history.toArray() + ) } else if (measurementStatistics.median != null) { delete measurementStatistics.median } @@ -214,8 +270,8 @@ export const updateWaitTimeWorkerUsage = < Data = unknown, Response = unknown >( - workerChoiceStrategyContext: - | WorkerChoiceStrategyContext + workerChoiceStrategiesContext: + | WorkerChoiceStrategiesContext | undefined, workerUsage: WorkerUsage, task: Task @@ -224,7 +280,7 @@ export const updateWaitTimeWorkerUsage = < const taskWaitTime = timestamp - (task.timestamp ?? timestamp) updateMeasurementStatistics( workerUsage.waitTime, - workerChoiceStrategyContext?.getTaskStatisticsRequirements().waitTime, + workerChoiceStrategiesContext?.getTaskStatisticsRequirements().waitTime, taskWaitTime ) } @@ -253,8 +309,8 @@ export const updateRunTimeWorkerUsage = < Data = unknown, Response = unknown >( - workerChoiceStrategyContext: - | WorkerChoiceStrategyContext + workerChoiceStrategiesContext: + | WorkerChoiceStrategiesContext | undefined, workerUsage: WorkerUsage, message: MessageValue @@ -264,7 +320,7 @@ export const updateRunTimeWorkerUsage = < } updateMeasurementStatistics( workerUsage.runTime, - workerChoiceStrategyContext?.getTaskStatisticsRequirements().runTime, + workerChoiceStrategiesContext?.getTaskStatisticsRequirements().runTime, message.taskPerformance?.runTime ?? 0 ) } @@ -274,8 +330,8 @@ export const updateEluWorkerUsage = < Data = unknown, Response = unknown >( - workerChoiceStrategyContext: - | WorkerChoiceStrategyContext + workerChoiceStrategiesContext: + | WorkerChoiceStrategiesContext | undefined, workerUsage: WorkerUsage, message: MessageValue @@ -284,7 +340,7 @@ export const updateEluWorkerUsage = < return } const eluTaskStatisticsRequirements = - workerChoiceStrategyContext?.getTaskStatisticsRequirements().elu + workerChoiceStrategiesContext?.getTaskStatisticsRequirements().elu updateMeasurementStatistics( workerUsage.elu.active, eluTaskStatisticsRequirements, @@ -312,13 +368,13 @@ export const updateEluWorkerUsage = < export const createWorker = ( type: WorkerType, filePath: string, - opts: { env?: Record, workerOptions?: WorkerOptions } + opts: { env?: Record; workerOptions?: WorkerOptions } ): Worker => { switch (type) { case WorkerTypes.thread: - return new Worker(filePath, { + return new ThreadWorker(filePath, { env: SHARE_ENV, - ...opts.workerOptions + ...opts.workerOptions, }) as unknown as Worker case WorkerTypes.cluster: return cluster.fork(opts.env) as unknown as Worker @@ -328,6 +384,34 @@ export const createWorker = ( } } +/** + * Returns the worker type of the given worker. + * @param worker - The worker to get the type of. + * @returns The worker type of the given worker. + * @internal + */ +export const getWorkerType = (worker: IWorker): WorkerType | undefined => { + if (worker instanceof ThreadWorker) { + return WorkerTypes.thread + } else if (worker instanceof ClusterWorker) { + return WorkerTypes.cluster + } +} + +/** + * Returns the worker id of the given worker. + * @param worker - The worker to get the id of. + * @returns The worker id of the given worker. + * @internal + */ +export const getWorkerId = (worker: IWorker): number | undefined => { + if (worker instanceof ThreadWorker) { + return worker.threadId + } else if (worker instanceof ClusterWorker) { + return worker.id + } +} + export const waitWorkerNodeEvents = async < Worker extends IWorker, Data = unknown @@ -343,12 +427,20 @@ export const waitWorkerNodeEvents = async < resolve(events) return } - workerNode.on(workerNodeEvent, () => { - ++events - if (events === numberOfEventsToWait) { - resolve(events) - } - }) + switch (workerNodeEvent) { + case 'idle': + case 'backPressure': + case 'taskFinished': + workerNode.on(workerNodeEvent, () => { + ++events + if (events === numberOfEventsToWait) { + resolve(events) + } + }) + break + default: + throw new Error('Invalid worker node event') + } if (timeout >= 0) { setTimeout(() => { resolve(events)