X-Git-Url: https://git.piment-noir.org/?a=blobdiff_plain;f=src%2Fpools%2Futils.ts;h=724c4ab960c691e8f4a5d8661f9c442ff14d5b3d;hb=e334515bf28581c234d6b9f5239425c9a0c6a7ec;hp=99584d8528d65f5538d3f89e309a6dd77d4fde79;hpb=d35e571704515a8b729d3455e4784054f07c368f;p=poolifier.git diff --git a/src/pools/utils.ts b/src/pools/utils.ts index 99584d85..724c4ab9 100644 --- a/src/pools/utils.ts +++ b/src/pools/utils.ts @@ -1,15 +1,23 @@ +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 { cpus } from 'node:os' 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 { IPool, TasksQueueOptions } from './pool.js' import { type MeasurementStatisticsRequirements, WorkerChoiceStrategies, - type WorkerChoiceStrategy + type WorkerChoiceStrategy, + type WorkerChoiceStrategyOptions } from './selection-strategies/selection-strategies-types.js' -import type { TasksQueueOptions } from './pool.js' +import type { WorkerChoiceStrategyContext } from './selection-strategies/worker-choice-strategy-context.js' import { type IWorker, type IWorkerNode, @@ -19,7 +27,16 @@ import { WorkerTypes, 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 @@ -33,7 +50,92 @@ export const getDefaultTasksQueueOptions = ( } } -export const checkFilePath = (filePath: string): void => { +export const getWorkerChoiceStrategyRetries = < + Worker extends IWorker, + Data, + Response +>( + pool: IPool, + opts?: WorkerChoiceStrategyOptions + ): number => { + return ( + pool.info.maxSize + + Object.keys(opts?.weights ?? getDefaultWeights(pool.info.maxSize)).length + ) +} + +export const buildWorkerChoiceStrategyOptions = < + Worker extends IWorker, + Data, + Response +>( + pool: IPool, + opts?: WorkerChoiceStrategyOptions + ): WorkerChoiceStrategyOptions => { + opts = clone(opts ?? {}) + opts.weights = opts.weights ?? getDefaultWeights(pool.info.maxSize) + return { + ...{ + runTime: { median: false }, + waitTime: { median: false }, + elu: { median: false } + }, + ...opts + } +} + +const clone = (object: T): T => { + return structuredClone(object) +} + +const getDefaultWeights = ( + poolMaxSize: number, + defaultWorkerWeight?: number +): Record => { + defaultWorkerWeight = defaultWorkerWeight ?? getDefaultWorkerWeight() + const weights: Record = {} + for (let workerNodeKey = 0; workerNodeKey < poolMaxSize; workerNodeKey++) { + weights[workerNodeKey] = defaultWorkerWeight + } + return weights +} + +const estimatedCpuSpeed = (): number => { + const runs = 150000000 + const begin = performance.now() + // eslint-disable-next-line no-empty + for (let i = runs; i > 0; i--) {} + const end = performance.now() + const duration = end - begin + return Math.trunc(runs / duration / 1000) // in MHz +} + +const getDefaultWorkerWeight = (): number => { + const currentCpus = cpus() + let estCpuSpeed: number | undefined + // eslint-disable-next-line @typescript-eslint/no-unnecessary-condition + if (currentCpus.every(cpu => cpu.speed == null || cpu.speed === 0)) { + estCpuSpeed = estimatedCpuSpeed() + } + let cpusCycleTimeWeight = 0 + for (const cpu of currentCpus) { + // eslint-disable-next-line @typescript-eslint/no-unnecessary-condition + if (cpu.speed == null || cpu.speed === 0) { + cpu.speed = + // eslint-disable-next-line @typescript-eslint/no-unnecessary-condition + currentCpus.find(cpu => cpu.speed != null && cpu.speed !== 0)?.speed ?? + estCpuSpeed ?? + 2000 + } + // CPU estimated cycle time + const numberOfDigits = cpu.speed.toString().length - 1 + const cpuCycleTime = 1 / (cpu.speed / Math.pow(10, numberOfDigits)) + cpusCycleTimeWeight += cpuCycleTime * Math.pow(10, numberOfDigits) + } + return Math.round(cpusCycleTimeWeight / currentCpus.length) +} + +export const checkFilePath = (filePath: string | undefined): void => { if (filePath == null) { throw new TypeError('The worker file path must be specified') } @@ -45,7 +147,10 @@ export const checkFilePath = (filePath: string): void => { } } -export const checkDynamicPoolSize = (min: number, max: number): void => { +export const checkDynamicPoolSize = ( + min: number, + max: number | undefined +): void => { if (max == null) { throw new TypeError( 'Cannot instantiate a dynamic pool without specifying the maximum pool size' @@ -70,7 +175,7 @@ export const checkDynamicPoolSize = (min: number, max: number): void => { } export const checkValidWorkerChoiceStrategy = ( - workerChoiceStrategy: WorkerChoiceStrategy + workerChoiceStrategy: WorkerChoiceStrategy | undefined ): void => { if ( workerChoiceStrategy != null && @@ -81,7 +186,7 @@ export const checkValidWorkerChoiceStrategy = ( } export const checkValidTasksQueueOptions = ( - tasksQueueOptions: TasksQueueOptions + tasksQueueOptions: TasksQueueOptions | undefined ): void => { if (tasksQueueOptions != null && !isPlainObject(tasksQueueOptions)) { throw new TypeError('Invalid tasks queue options: must be a plain object') @@ -118,9 +223,9 @@ export const checkValidTasksQueueOptions = ( } export const checkWorkerNodeArguments = ( - type: WorkerType, - filePath: string, - opts: WorkerNodeOptions + type: WorkerType | undefined, + filePath: string | undefined, + opts: WorkerNodeOptions | undefined ): void => { if (type == null) { throw new TypeError('Cannot construct a worker node without a worker type') @@ -168,10 +273,14 @@ export const checkWorkerNodeArguments = ( */ const updateMeasurementStatistics = ( measurementStatistics: MeasurementStatistics, - measurementRequirements: MeasurementStatisticsRequirements, - measurementValue: number + measurementRequirements: MeasurementStatisticsRequirements | undefined, + measurementValue: number | undefined ): void => { - if (measurementRequirements.aggregate) { + if ( + measurementRequirements != null && + measurementValue != null && + measurementRequirements.aggregate + ) { measurementStatistics.aggregate = (measurementStatistics.aggregate ?? 0) + measurementValue measurementStatistics.minimum = min( @@ -182,10 +291,7 @@ const updateMeasurementStatistics = ( measurementValue, measurementStatistics.maximum ?? -Infinity ) - if ( - (measurementRequirements.average || measurementRequirements.median) && - measurementValue != null - ) { + if (measurementRequirements.average || measurementRequirements.median) { measurementStatistics.history.push(measurementValue) if (measurementRequirements.average) { measurementStatistics.average = average(measurementStatistics.history) @@ -210,11 +316,9 @@ export const updateWaitTimeWorkerUsage = < Data = unknown, Response = unknown >( - workerChoiceStrategyContext: WorkerChoiceStrategyContext< - Worker, - Data, - Response - >, + workerChoiceStrategyContext: + | WorkerChoiceStrategyContext + | undefined, workerUsage: WorkerUsage, task: Task ): void => { @@ -222,7 +326,7 @@ export const updateWaitTimeWorkerUsage = < const taskWaitTime = timestamp - (task.timestamp ?? timestamp) updateMeasurementStatistics( workerUsage.waitTime, - workerChoiceStrategyContext.getTaskStatisticsRequirements().waitTime, + workerChoiceStrategyContext?.getTaskStatisticsRequirements().waitTime, taskWaitTime ) } @@ -233,6 +337,7 @@ export const updateTaskStatisticsWorkerUsage = ( ): void => { const workerTaskStatistics = workerUsage.tasks if ( + // eslint-disable-next-line @typescript-eslint/no-unnecessary-condition workerTaskStatistics.executing != null && workerTaskStatistics.executing > 0 ) { @@ -250,11 +355,9 @@ export const updateRunTimeWorkerUsage = < Data = unknown, Response = unknown >( - workerChoiceStrategyContext: WorkerChoiceStrategyContext< - Worker, - Data, - Response - >, + workerChoiceStrategyContext: + | WorkerChoiceStrategyContext + | undefined, workerUsage: WorkerUsage, message: MessageValue ): void => { @@ -263,7 +366,7 @@ export const updateRunTimeWorkerUsage = < } updateMeasurementStatistics( workerUsage.runTime, - workerChoiceStrategyContext.getTaskStatisticsRequirements().runTime, + workerChoiceStrategyContext?.getTaskStatisticsRequirements().runTime, message.taskPerformance?.runTime ?? 0 ) } @@ -273,19 +376,17 @@ export const updateEluWorkerUsage = < Data = unknown, Response = unknown >( - workerChoiceStrategyContext: WorkerChoiceStrategyContext< - Worker, - Data, - Response - >, + workerChoiceStrategyContext: + | WorkerChoiceStrategyContext + | undefined, workerUsage: WorkerUsage, message: MessageValue ): void => { if (message.workerError != null) { return } - const eluTaskStatisticsRequirements: MeasurementStatisticsRequirements = - workerChoiceStrategyContext.getTaskStatisticsRequirements().elu + const eluTaskStatisticsRequirements = + workerChoiceStrategyContext?.getTaskStatisticsRequirements().elu updateMeasurementStatistics( workerUsage.elu.active, eluTaskStatisticsRequirements, @@ -296,7 +397,7 @@ export const updateEluWorkerUsage = < eluTaskStatisticsRequirements, message.taskPerformance?.elu?.idle ?? 0 ) - if (eluTaskStatisticsRequirements.aggregate) { + if (eluTaskStatisticsRequirements?.aggregate === true) { if (message.taskPerformance?.elu != null) { if (workerUsage.elu.utilization != null) { workerUsage.elu.utilization = @@ -317,18 +418,48 @@ export const createWorker = ( ): 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 + return cluster.fork(opts.env) as unknown as Worker default: // eslint-disable-next-line @typescript-eslint/restrict-template-expressions throw new Error(`Unknown worker type '${type}'`) } } +/** + * 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