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