X-Git-Url: https://git.piment-noir.org/?a=blobdiff_plain;f=src%2Futils.ts;h=400a92b14a7f42e8518878af1744623c8f6c2290;hb=563b18cd8fe9c211f15e9da91a299f2819abea01;hp=0becc5ed9aed9147cfaa5351d60a5e757ba0554b;hpb=ff128cc9dc9cee365f55b15721d5b90707577a0a;p=poolifier.git diff --git a/src/utils.ts b/src/utils.ts index 0becc5ed..400a92b1 100644 --- a/src/utils.ts +++ b/src/utils.ts @@ -1,9 +1,18 @@ import * as os from 'node:os' +import { webcrypto } from 'node:crypto' +import { Worker as ClusterWorker } from 'node:cluster' +import { Worker as ThreadWorker } from 'node:worker_threads' 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' /** * Default task name. @@ -22,6 +31,7 @@ export const EMPTY_FUNCTION: () => void = Object.freeze(() => { */ export const DEFAULT_WORKER_CHOICE_STRATEGY_OPTIONS: WorkerChoiceStrategyOptions = { + retries: 6, runTime: { median: false }, waitTime: { median: false }, elu: { median: false } @@ -48,19 +58,100 @@ export const availableParallelism = (): number => { 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 } +/** + * 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 + } +} + +/** + * Sleeps for the given amount of milliseconds. + * + * @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 => { + setTimeout(resolve, ms) + }) +} + +/** + * Computes the retry delay in milliseconds using an exponential back off algorithm. + * + * @param retryNumber - The number of retries that have already been attempted + * @param delayFactor - The base delay factor in milliseconds + * @returns Delay in milliseconds + * @internal + */ +export const exponentialDelay = ( + retryNumber = 0, + delayFactor = 100 +): number => { + const delay = Math.pow(2, retryNumber) * delayFactor + const randomSum = delay * 0.2 * secureRandom() // 0-20% of the delay + return delay + randomSum +} + +/** + * Computes the average of the given data set. + * + * @param dataSet - Data set. + * @returns The average of the given data set. + * @internal + */ +export const average = (dataSet: number[]): number => { + if (Array.isArray(dataSet) && dataSet.length === 0) { + return 0 + } + if (Array.isArray(dataSet) && dataSet.length === 1) { + return dataSet[0] + } + return ( + dataSet.reduce((accumulator, number) => accumulator + number, 0) / + dataSet.length + ) +} + /** * Computes the median of the given data set. * * @param dataSet - Data set. * @returns The median of the given data set. + * @internal */ export const median = (dataSet: number[]): number => { if (Array.isArray(dataSet) && dataSet.length === 0) { @@ -84,6 +175,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) @@ -95,6 +187,7 @@ 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' && @@ -109,6 +202,7 @@ export const isPlainObject = (obj: unknown): boolean => * @param killBehavior - Which kind of kill behavior to detect. * @param value - Any value. * @returns `true` if `value` was strictly equals to `killBehavior`, otherwise `false`. + * @internal */ export const isKillBehavior = ( killBehavior: KB, @@ -122,9 +216,86 @@ export const isKillBehavior = ( * * @param fn - Any value. * @returns `true` if `fn` was an asynchronous function, otherwise `false`. + * @internal */ export const isAsyncFunction = ( fn: unknown ): fn is (...args: unknown[]) => Promise => { return typeof fn === 'function' && fn.constructor.name === 'AsyncFunction' } + +/** + * Updates the given measurement statistics. + * + * @param measurementStatistics - The measurement statistics to update. + * @param measurementRequirements - The measurement statistics requirements. + * @param measurementValue - The measurement value. + * @param numberOfMeasurements - The number of measurements. + * @internal + */ +export const updateMeasurementStatistics = ( + measurementStatistics: MeasurementStatistics, + measurementRequirements: MeasurementStatisticsRequirements, + measurementValue: number +): void => { + if (measurementRequirements.aggregate) { + measurementStatistics.aggregate = + (measurementStatistics.aggregate ?? 0) + measurementValue + measurementStatistics.minimum = min( + measurementValue, + measurementStatistics.minimum ?? Infinity + ) + measurementStatistics.maximum = 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 + } + } + } +} + +/** + * Generates a cryptographically secure random number in the [0,1[ range + * + * @returns A number in the [0,1[ range + * @internal + */ +export const secureRandom = (): number => { + return webcrypto.getRandomValues(new Uint32Array(1))[0] / 0x100000000 +} + +/** + * Returns the minimum of the given numbers. + * If no numbers are given, `Infinity` is returned. + * + * @param args - The numbers to get the minimum of. + * @returns The minimum of the given numbers. + * @internal + */ +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)