X-Git-Url: https://git.piment-noir.org/?a=blobdiff_plain;f=src%2Futils.ts;h=b38cf948e833ef41ccc7d183db0498a8e4257ed0;hb=efc4d37dd21d0c80e5fa2d58009e6298eefe5d48;hp=82c38c09e7161fd06e2c6f1628fdaad1e8f4c63f;hpb=bdb9d7125f62a36dd65cba6aa9110ce269359f0a;p=poolifier.git diff --git a/src/utils.ts b/src/utils.ts index 82c38c09..b38cf948 100644 --- a/src/utils.ts +++ b/src/utils.ts @@ -1,18 +1,15 @@ import * as os from 'node:os' -import { webcrypto } from 'node:crypto' +import { getRandomValues, randomInt } from 'node:crypto' import { Worker as ClusterWorker } from 'node:cluster' import { Worker as ThreadWorker } from 'node:worker_threads' +import { cpus } from 'node:os' import type { MeasurementStatisticsRequirements, WorkerChoiceStrategyOptions -} from './pools/selection-strategies/selection-strategies-types' -import type { KillBehavior } from './worker/worker-options' -import { - type IWorker, - type MeasurementStatistics, - type WorkerType, - WorkerTypes -} from './pools/worker' +} from './pools/selection-strategies/selection-strategies-types.js' +import type { KillBehavior } from './worker/worker-options.js' +import { type IWorker, type WorkerType, WorkerTypes } from './pools/worker.js' +import type { IPool } from './pools/pool.js' /** * Default task name. @@ -26,17 +23,6 @@ export const EMPTY_FUNCTION: () => void = Object.freeze(() => { /* Intentionally empty */ }) -/** - * Default worker choice strategy options. - */ -export const DEFAULT_WORKER_CHOICE_STRATEGY_OPTIONS: WorkerChoiceStrategyOptions = - { - retries: 6, - runTime: { median: false }, - waitTime: { median: false }, - elu: { median: false } - } - /** * Default measurement statistics requirements. */ @@ -52,16 +38,15 @@ export const DEFAULT_MEASUREMENT_STATISTICS_REQUIREMENTS: MeasurementStatisticsR * Always returns a value greater than zero. * * @returns The host OS optimized maximum pool size. - * @internal */ export const availableParallelism = (): number => { let availableParallelism = 1 try { availableParallelism = os.availableParallelism() } catch { - const numberOfCpus = os.cpus() - if (Array.isArray(numberOfCpus) && numberOfCpus.length > 0) { - availableParallelism = numberOfCpus.length + const cpus = os.cpus() + if (Array.isArray(cpus) && cpus.length > 0) { + availableParallelism = cpus.length } } return availableParallelism @@ -74,9 +59,7 @@ export const availableParallelism = (): number => { * @returns The worker type of the given worker. * @internal */ -export const getWorkerType = ( - worker: Worker -): WorkerType | undefined => { +export const getWorkerType = (worker: IWorker): WorkerType | undefined => { if (worker instanceof ThreadWorker) { return WorkerTypes.thread } else if (worker instanceof ClusterWorker) { @@ -91,9 +74,7 @@ export const getWorkerType = ( * @returns The worker id of the given worker. * @internal */ -export const getWorkerId = ( - worker: Worker -): number | undefined => { +export const getWorkerId = (worker: IWorker): number | undefined => { if (worker instanceof ThreadWorker) { return worker.threadId } else if (worker instanceof ClusterWorker) { @@ -106,9 +87,10 @@ export const getWorkerId = ( * * @param ms - The amount of milliseconds to sleep. * @returns A promise that resolves after the given amount of milliseconds. + * @internal */ export const sleep = async (ms: number): Promise => { - await new Promise((resolve) => { + await new Promise(resolve => { setTimeout(resolve, ms) }) } @@ -179,6 +161,7 @@ export const median = (dataSet: number[]): number => { * @param num - The number to round. * @param scale - The scale to round to. * @returns The rounded number. + * @internal */ export const round = (num: number, scale = 2): number => { const rounder = Math.pow(10, scale) @@ -190,11 +173,12 @@ export const round = (num: number, scale = 2): number => { * * @param obj - The object to check. * @returns `true` if the given object is a plain object, `false` otherwise. + * @internal */ export const isPlainObject = (obj: unknown): boolean => typeof obj === 'object' && obj !== null && - obj?.constructor === Object && + obj.constructor === Object && Object.prototype.toString.call(obj) === '[object Object]' /** @@ -218,6 +202,7 @@ export const isKillBehavior = ( * * @param fn - Any value. * @returns `true` if `fn` was an asynchronous function, otherwise `false`. + * @internal */ export const isAsyncFunction = ( fn: unknown @@ -226,54 +211,127 @@ export const isAsyncFunction = ( } /** - * Updates the given measurement statistics. + * Generates a cryptographically secure random number in the [0,1[ range * - * @param measurementStatistics - The measurement statistics to update. - * @param measurementRequirements - The measurement statistics requirements. - * @param measurementValue - The measurement value. - * @param numberOfMeasurements - The number of measurements. + * @returns A number in the [0,1[ range * @internal */ -export const updateMeasurementStatistics = ( - measurementStatistics: MeasurementStatistics, - measurementRequirements: MeasurementStatisticsRequirements, - measurementValue: number -): void => { - if (measurementRequirements.aggregate) { - measurementStatistics.aggregate = - (measurementStatistics.aggregate ?? 0) + measurementValue - measurementStatistics.minimum = Math.min( - measurementValue, - measurementStatistics.minimum ?? Infinity - ) - measurementStatistics.maximum = Math.max( - measurementValue, - measurementStatistics.maximum ?? -Infinity - ) - if ( - (measurementRequirements.average || measurementRequirements.median) && - measurementValue != null - ) { - measurementStatistics.history.push(measurementValue) - if (measurementRequirements.average) { - measurementStatistics.average = average(measurementStatistics.history) - } else if (measurementStatistics.average != null) { - delete measurementStatistics.average - } - if (measurementRequirements.median) { - measurementStatistics.median = median(measurementStatistics.history) - } else if (measurementStatistics.median != null) { - delete measurementStatistics.median - } - } - } +export const secureRandom = (): number => { + return getRandomValues(new Uint32Array(1))[0] / 0x100000000 } /** - * Generate a cryptographically secure random number in the [0,1[ range + * Returns the minimum of the given numbers. + * If no numbers are given, `Infinity` is returned. * - * @returns A number in the [0,1[ range + * @param args - The numbers to get the minimum of. + * @returns The minimum of the given numbers. + * @internal */ -export const secureRandom = (): number => { - return webcrypto.getRandomValues(new Uint32Array(1))[0] / 0x100000000 +export const min = (...args: number[]): number => + args.reduce((minimum, num) => (minimum < num ? minimum : num), Infinity) + +/** + * Returns the maximum of the given numbers. + * If no numbers are given, `-Infinity` is returned. + * + * @param args - The numbers to get the maximum of. + * @returns The maximum of the given numbers. + * @internal + */ +export const max = (...args: number[]): number => + args.reduce((maximum, num) => (maximum > num ? maximum : num), -Infinity) + +/** + * Wraps a function so that it can only be called once. + * + * @param fn - The function to wrap. + * @param context - The context to bind the function to. + * @returns The wrapped function. + * @internal + */ +// eslint-disable-next-line @typescript-eslint/no-explicit-any +export const once = ( + fn: (...args: A) => R, + context: T +): ((...args: A) => R) => { + let result: R + return (...args: A) => { + // eslint-disable-next-line @typescript-eslint/no-unnecessary-condition + if (fn != null) { + result = fn.apply(context, args) + ;(fn as unknown as undefined) = (context as unknown as undefined) = + undefined + } + return result + } +} + +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 + ) +} + +const clone = (object: T): T => { + return structuredClone(object) +} + +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 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 getDefaultWorkerWeight = (): number => { + const cpuSpeed = randomInt(500, 2500) + let cpusCycleTimeWeight = 0 + for (const cpu of cpus()) { + // 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 + cpus().find(cpu => cpu.speed != null && cpu.speed !== 0)?.speed ?? + cpuSpeed + } + // 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 / cpus().length) }