Commit | Line | Data |
---|---|---|
bcfb06ce JB |
1 | import { cpus } from 'node:os' |
2 | ||
3 | import type { IPool } from '../pool.js' | |
4 | import type { IWorker } from '../worker.js' | |
5 | import { FairShareWorkerChoiceStrategy } from './fair-share-worker-choice-strategy.js' | |
6 | import { InterleavedWeightedRoundRobinWorkerChoiceStrategy } from './interleaved-weighted-round-robin-worker-choice-strategy.js' | |
7 | import { LeastBusyWorkerChoiceStrategy } from './least-busy-worker-choice-strategy.js' | |
8 | import { LeastEluWorkerChoiceStrategy } from './least-elu-worker-choice-strategy.js' | |
9 | import { LeastUsedWorkerChoiceStrategy } from './least-used-worker-choice-strategy.js' | |
10 | import { RoundRobinWorkerChoiceStrategy } from './round-robin-worker-choice-strategy.js' | |
11 | import { | |
12 | type IWorkerChoiceStrategy, | |
763abb1c | 13 | type MeasurementStatisticsRequirements, |
19b8be8b JB |
14 | type StrategyPolicy, |
15 | type TaskStatisticsRequirements, | |
bcfb06ce JB |
16 | WorkerChoiceStrategies, |
17 | type WorkerChoiceStrategy, | |
3a502712 | 18 | type WorkerChoiceStrategyOptions, |
bcfb06ce JB |
19 | } from './selection-strategies-types.js' |
20 | import { WeightedRoundRobinWorkerChoiceStrategy } from './weighted-round-robin-worker-choice-strategy.js' | |
21 | import type { WorkerChoiceStrategiesContext } from './worker-choice-strategies-context.js' | |
22 | ||
bcfb06ce JB |
23 | const estimatedCpuSpeed = (): number => { |
24 | const runs = 150000000 | |
25 | const begin = performance.now() | |
26 | // eslint-disable-next-line no-empty | |
27 | for (let i = runs; i > 0; i--) {} | |
28 | const end = performance.now() | |
29 | const duration = end - begin | |
30 | return Math.trunc(runs / duration / 1000) // in MHz | |
31 | } | |
32 | ||
33 | const getDefaultWorkerWeight = (): number => { | |
34 | const currentCpus = cpus() | |
35 | let estCpuSpeed: number | undefined | |
bcfb06ce JB |
36 | if (currentCpus.every(cpu => cpu.speed == null || cpu.speed === 0)) { |
37 | estCpuSpeed = estimatedCpuSpeed() | |
38 | } | |
39 | let cpusCycleTimeWeight = 0 | |
40 | for (const cpu of currentCpus) { | |
bcfb06ce JB |
41 | if (cpu.speed == null || cpu.speed === 0) { |
42 | cpu.speed = | |
bcfb06ce JB |
43 | currentCpus.find(cpu => cpu.speed != null && cpu.speed !== 0)?.speed ?? |
44 | estCpuSpeed ?? | |
45 | 2000 | |
46 | } | |
47 | // CPU estimated cycle time | |
48 | const numberOfDigits = cpu.speed.toString().length - 1 | |
49 | const cpuCycleTime = 1 / (cpu.speed / Math.pow(10, numberOfDigits)) | |
50 | cpusCycleTimeWeight += cpuCycleTime * Math.pow(10, numberOfDigits) | |
51 | } | |
52 | return Math.round(cpusCycleTimeWeight / currentCpus.length) | |
53 | } | |
54 | ||
55 | const getDefaultWeights = ( | |
56 | poolMaxSize: number, | |
57 | defaultWorkerWeight?: number | |
58 | ): Record<number, number> => { | |
59 | defaultWorkerWeight = defaultWorkerWeight ?? getDefaultWorkerWeight() | |
60 | const weights: Record<number, number> = {} | |
61 | for (let workerNodeKey = 0; workerNodeKey < poolMaxSize; workerNodeKey++) { | |
62 | weights[workerNodeKey] = defaultWorkerWeight | |
63 | } | |
64 | return weights | |
65 | } | |
66 | ||
67 | export const getWorkerChoiceStrategiesRetries = < | |
68 | Worker extends IWorker, | |
69 | Data, | |
70 | Response | |
71 | >( | |
72 | pool: IPool<Worker, Data, Response>, | |
73 | opts?: WorkerChoiceStrategyOptions | |
74 | ): number => { | |
75 | return ( | |
76 | pool.info.maxSize + | |
77 | Object.keys(opts?.weights ?? getDefaultWeights(pool.info.maxSize)).length | |
78 | ) | |
79 | } | |
80 | ||
81 | export const buildWorkerChoiceStrategyOptions = < | |
82 | Worker extends IWorker, | |
83 | Data, | |
84 | Response | |
85 | >( | |
86 | pool: IPool<Worker, Data, Response>, | |
87 | opts?: WorkerChoiceStrategyOptions | |
88 | ): WorkerChoiceStrategyOptions => { | |
85bbc7ab | 89 | opts = structuredClone(opts ?? {}) |
bcfb06ce JB |
90 | opts.weights = opts.weights ?? getDefaultWeights(pool.info.maxSize) |
91 | return { | |
92 | ...{ | |
93 | runTime: { median: false }, | |
94 | waitTime: { median: false }, | |
3a502712 | 95 | elu: { median: false }, |
bcfb06ce | 96 | }, |
3a502712 | 97 | ...opts, |
bcfb06ce JB |
98 | } |
99 | } | |
100 | ||
763abb1c JB |
101 | export const toggleMedianMeasurementStatisticsRequirements = ( |
102 | measurementStatisticsRequirements: MeasurementStatisticsRequirements, | |
103 | toggleMedian: boolean | |
104 | ): void => { | |
105 | if (measurementStatisticsRequirements.average && toggleMedian) { | |
106 | measurementStatisticsRequirements.average = false | |
107 | measurementStatisticsRequirements.median = toggleMedian | |
108 | } | |
109 | if (measurementStatisticsRequirements.median && !toggleMedian) { | |
110 | measurementStatisticsRequirements.average = true | |
111 | measurementStatisticsRequirements.median = toggleMedian | |
112 | } | |
113 | } | |
114 | ||
19b8be8b JB |
115 | export const buildWorkerChoiceStrategiesPolicy = ( |
116 | workerChoiceStrategies: Map<WorkerChoiceStrategy, IWorkerChoiceStrategy> | |
117 | ): StrategyPolicy => { | |
63af5400 JB |
118 | const policies: StrategyPolicy[] = Array.from( |
119 | workerChoiceStrategies, | |
120 | ([_, workerChoiceStrategy]) => workerChoiceStrategy.strategyPolicy | |
121 | ) | |
19b8be8b JB |
122 | return { |
123 | dynamicWorkerUsage: policies.some(p => p.dynamicWorkerUsage), | |
3a502712 | 124 | dynamicWorkerReady: policies.some(p => p.dynamicWorkerReady), |
19b8be8b JB |
125 | } |
126 | } | |
127 | ||
128 | export const buildWorkerChoiceStrategiesTaskStatisticsRequirements = ( | |
129 | workerChoiceStrategies: Map<WorkerChoiceStrategy, IWorkerChoiceStrategy> | |
130 | ): TaskStatisticsRequirements => { | |
63af5400 JB |
131 | const taskStatisticsRequirements: TaskStatisticsRequirements[] = Array.from( |
132 | workerChoiceStrategies, | |
133 | ([_, workerChoiceStrategy]) => | |
19b8be8b | 134 | workerChoiceStrategy.taskStatisticsRequirements |
63af5400 | 135 | ) |
19b8be8b JB |
136 | return { |
137 | runTime: { | |
138 | aggregate: taskStatisticsRequirements.some(r => r.runTime.aggregate), | |
139 | average: taskStatisticsRequirements.some(r => r.runTime.average), | |
3a502712 | 140 | median: taskStatisticsRequirements.some(r => r.runTime.median), |
19b8be8b JB |
141 | }, |
142 | waitTime: { | |
143 | aggregate: taskStatisticsRequirements.some(r => r.waitTime.aggregate), | |
144 | average: taskStatisticsRequirements.some(r => r.waitTime.average), | |
3a502712 | 145 | median: taskStatisticsRequirements.some(r => r.waitTime.median), |
19b8be8b JB |
146 | }, |
147 | elu: { | |
148 | aggregate: taskStatisticsRequirements.some(r => r.elu.aggregate), | |
149 | average: taskStatisticsRequirements.some(r => r.elu.average), | |
3a502712 JB |
150 | median: taskStatisticsRequirements.some(r => r.elu.median), |
151 | }, | |
19b8be8b JB |
152 | } |
153 | } | |
154 | ||
bcfb06ce JB |
155 | export const getWorkerChoiceStrategy = <Worker extends IWorker, Data, Response>( |
156 | workerChoiceStrategy: WorkerChoiceStrategy, | |
157 | pool: IPool<Worker, Data, Response>, | |
158 | context: ThisType<WorkerChoiceStrategiesContext<Worker, Data, Response>>, | |
159 | opts?: WorkerChoiceStrategyOptions | |
160 | ): IWorkerChoiceStrategy => { | |
161 | switch (workerChoiceStrategy) { | |
162 | case WorkerChoiceStrategies.ROUND_ROBIN: | |
163 | return new (RoundRobinWorkerChoiceStrategy.bind(context))(pool, opts) | |
164 | case WorkerChoiceStrategies.LEAST_USED: | |
165 | return new (LeastUsedWorkerChoiceStrategy.bind(context))(pool, opts) | |
166 | case WorkerChoiceStrategies.LEAST_BUSY: | |
167 | return new (LeastBusyWorkerChoiceStrategy.bind(context))(pool, opts) | |
168 | case WorkerChoiceStrategies.LEAST_ELU: | |
169 | return new (LeastEluWorkerChoiceStrategy.bind(context))(pool, opts) | |
170 | case WorkerChoiceStrategies.FAIR_SHARE: | |
171 | return new (FairShareWorkerChoiceStrategy.bind(context))(pool, opts) | |
172 | case WorkerChoiceStrategies.WEIGHTED_ROUND_ROBIN: | |
173 | return new (WeightedRoundRobinWorkerChoiceStrategy.bind(context))( | |
174 | pool, | |
175 | opts | |
176 | ) | |
177 | case WorkerChoiceStrategies.INTERLEAVED_WEIGHTED_ROUND_ROBIN: | |
178 | return new (InterleavedWeightedRoundRobinWorkerChoiceStrategy.bind( | |
179 | context | |
180 | ))(pool, opts) | |
181 | default: | |
182 | throw new Error( | |
bcfb06ce JB |
183 | `Worker choice strategy '${workerChoiceStrategy}' is not valid` |
184 | ) | |
185 | } | |
186 | } |