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, | |
13 | WorkerChoiceStrategies, | |
14 | type WorkerChoiceStrategy, | |
15 | type WorkerChoiceStrategyOptions | |
16 | } from './selection-strategies-types.js' | |
17 | import { WeightedRoundRobinWorkerChoiceStrategy } from './weighted-round-robin-worker-choice-strategy.js' | |
18 | import type { WorkerChoiceStrategiesContext } from './worker-choice-strategies-context.js' | |
19 | ||
20 | const clone = <T>(object: T): T => { | |
21 | return structuredClone<T>(object) | |
22 | } | |
23 | ||
24 | const estimatedCpuSpeed = (): number => { | |
25 | const runs = 150000000 | |
26 | const begin = performance.now() | |
27 | // eslint-disable-next-line no-empty | |
28 | for (let i = runs; i > 0; i--) {} | |
29 | const end = performance.now() | |
30 | const duration = end - begin | |
31 | return Math.trunc(runs / duration / 1000) // in MHz | |
32 | } | |
33 | ||
34 | const getDefaultWorkerWeight = (): number => { | |
35 | const currentCpus = cpus() | |
36 | let estCpuSpeed: number | undefined | |
37 | // eslint-disable-next-line @typescript-eslint/no-unnecessary-condition | |
38 | if (currentCpus.every(cpu => cpu.speed == null || cpu.speed === 0)) { | |
39 | estCpuSpeed = estimatedCpuSpeed() | |
40 | } | |
41 | let cpusCycleTimeWeight = 0 | |
42 | for (const cpu of currentCpus) { | |
43 | // eslint-disable-next-line @typescript-eslint/no-unnecessary-condition | |
44 | if (cpu.speed == null || cpu.speed === 0) { | |
45 | cpu.speed = | |
46 | // eslint-disable-next-line @typescript-eslint/no-unnecessary-condition | |
47 | currentCpus.find(cpu => cpu.speed != null && cpu.speed !== 0)?.speed ?? | |
48 | estCpuSpeed ?? | |
49 | 2000 | |
50 | } | |
51 | // CPU estimated cycle time | |
52 | const numberOfDigits = cpu.speed.toString().length - 1 | |
53 | const cpuCycleTime = 1 / (cpu.speed / Math.pow(10, numberOfDigits)) | |
54 | cpusCycleTimeWeight += cpuCycleTime * Math.pow(10, numberOfDigits) | |
55 | } | |
56 | return Math.round(cpusCycleTimeWeight / currentCpus.length) | |
57 | } | |
58 | ||
59 | const getDefaultWeights = ( | |
60 | poolMaxSize: number, | |
61 | defaultWorkerWeight?: number | |
62 | ): Record<number, number> => { | |
63 | defaultWorkerWeight = defaultWorkerWeight ?? getDefaultWorkerWeight() | |
64 | const weights: Record<number, number> = {} | |
65 | for (let workerNodeKey = 0; workerNodeKey < poolMaxSize; workerNodeKey++) { | |
66 | weights[workerNodeKey] = defaultWorkerWeight | |
67 | } | |
68 | return weights | |
69 | } | |
70 | ||
71 | export const getWorkerChoiceStrategiesRetries = < | |
72 | Worker extends IWorker, | |
73 | Data, | |
74 | Response | |
75 | >( | |
76 | pool: IPool<Worker, Data, Response>, | |
77 | opts?: WorkerChoiceStrategyOptions | |
78 | ): number => { | |
79 | return ( | |
80 | pool.info.maxSize + | |
81 | Object.keys(opts?.weights ?? getDefaultWeights(pool.info.maxSize)).length | |
82 | ) | |
83 | } | |
84 | ||
85 | export const buildWorkerChoiceStrategyOptions = < | |
86 | Worker extends IWorker, | |
87 | Data, | |
88 | Response | |
89 | >( | |
90 | pool: IPool<Worker, Data, Response>, | |
91 | opts?: WorkerChoiceStrategyOptions | |
92 | ): WorkerChoiceStrategyOptions => { | |
93 | opts = clone(opts ?? {}) | |
94 | opts.weights = opts.weights ?? getDefaultWeights(pool.info.maxSize) | |
95 | return { | |
96 | ...{ | |
97 | runTime: { median: false }, | |
98 | waitTime: { median: false }, | |
99 | elu: { median: false } | |
100 | }, | |
101 | ...opts | |
102 | } | |
103 | } | |
104 | ||
105 | export const getWorkerChoiceStrategy = <Worker extends IWorker, Data, Response>( | |
106 | workerChoiceStrategy: WorkerChoiceStrategy, | |
107 | pool: IPool<Worker, Data, Response>, | |
108 | context: ThisType<WorkerChoiceStrategiesContext<Worker, Data, Response>>, | |
109 | opts?: WorkerChoiceStrategyOptions | |
110 | ): IWorkerChoiceStrategy => { | |
111 | switch (workerChoiceStrategy) { | |
112 | case WorkerChoiceStrategies.ROUND_ROBIN: | |
113 | return new (RoundRobinWorkerChoiceStrategy.bind(context))(pool, opts) | |
114 | case WorkerChoiceStrategies.LEAST_USED: | |
115 | return new (LeastUsedWorkerChoiceStrategy.bind(context))(pool, opts) | |
116 | case WorkerChoiceStrategies.LEAST_BUSY: | |
117 | return new (LeastBusyWorkerChoiceStrategy.bind(context))(pool, opts) | |
118 | case WorkerChoiceStrategies.LEAST_ELU: | |
119 | return new (LeastEluWorkerChoiceStrategy.bind(context))(pool, opts) | |
120 | case WorkerChoiceStrategies.FAIR_SHARE: | |
121 | return new (FairShareWorkerChoiceStrategy.bind(context))(pool, opts) | |
122 | case WorkerChoiceStrategies.WEIGHTED_ROUND_ROBIN: | |
123 | return new (WeightedRoundRobinWorkerChoiceStrategy.bind(context))( | |
124 | pool, | |
125 | opts | |
126 | ) | |
127 | case WorkerChoiceStrategies.INTERLEAVED_WEIGHTED_ROUND_ROBIN: | |
128 | return new (InterleavedWeightedRoundRobinWorkerChoiceStrategy.bind( | |
129 | context | |
130 | ))(pool, opts) | |
131 | default: | |
132 | throw new Error( | |
133 | // eslint-disable-next-line @typescript-eslint/restrict-template-expressions | |
134 | `Worker choice strategy '${workerChoiceStrategy}' is not valid` | |
135 | ) | |
136 | } | |
137 | } |