Vihren Atlas

Atlas › Test

TestFairness_UniformDistribution_500

go.temporal.io/server/tools/fairsim/TestFairness_UniformDistribution_500

Package
go.temporal.io/server/tools/fairsim
Suite / test hierarchy
TestFairness_UniformDistribution_500
Test
TestFairness_UniformDistribution_500
Introduced at
TestFairness_UniformDistribution_500 Test frontier
Covered ranges
68
Covered lines
303
Covered files
4

Covered source

Expand a file to inspect source; the > gutter marks covered lines.

go.temporal.io/server/tools/fairsim/sim.go 118 covered LOC · 24 ranges

Open complete file

150 }
151
152 > func newLatencyStats() *latencyStats { sim.go
153 > return &latencyStats{
154 > byKey: make(map[string][]int64),
155 > byKeyNormalized: make(map[string][]float64),
156 > }
157 > }
158
159 > func newState(rnd *rand.Rand, counterFactory func() counter.Counter, partitions, strideFactor int) *state { sim.go
160 > return &state{
161 > rnd: rnd,
162 > counterFactory: counterFactory,
163 > partitions: make([]partitionState, partitions),
164 > strideFactor: float32(strideFactor),
165 > }
166 > }
167
168 > func newSimulator(state *state, stats *latencyStats, defaultPriority int, w io.Writer, verbose bool) *simulator { sim.go
169 > return &simulator{
170 > state: state,
171 > stats: stats,
172 > w: w,
173 > verbose: verbose,
174 > defaultPriority: defaultPriority,
175 > }
176 > }
177
178 // addTask adds a task to the simulator, assigning defaults and an index.
179 > func (sim *simulator) addTask(t *task) { sim.go
180 > t.pri = cmp.Or(t.pri, sim.defaultPriority)
181 > t.fweight = cmp.Or(t.fweight, 1.0)
182 > t.index = sim.nextIndex
183 > sim.nextIndex++
184 > sim.state.addTask(t)
185 > }
186
187 // processTask records stats for a dispatched task and returns the latency.
188 > func (sim *simulator) processTask(t *task) int64 { sim.go
189 > latency := sim.dispatchIndex - t.index
190 > sim.stats.byKey[t.fkey] = append(sim.stats.byKey[t.fkey], latency)
191 > sim.stats.overall = append(sim.stats.overall, latency)
192 > sim.dispatchIndex++
193 > return latency
194 > }
195
196 // printTask writes a single task's dispatch info to the writer.
197 > func (sim *simulator) printTask(t *task, partition int, latency int64) { sim.go
198 > if !sim.verbose {
199 > return
200 > }
201 fmt.Fprintf(sim.w, "task idx:%6d dsp:%6d lat:%6d pri:%2d fkey:%10q fweight:%3g part:%2d payload:%q\n",
202 t.index, sim.dispatchIndex-1, latency, t.pri, t.fkey, t.fweight, partition, t.payload)
204
205 // drainTasks pops and processes all remaining tasks, printing each one.
206 > func (sim *simulator) drainTasks() { sim.go
207 > for t, partition := sim.state.popTask(); t != nil; t, partition = sim.state.popTask() {
208 > latency := sim.processTask(t)
209 > sim.printTask(t, partition, latency)
210 > }
211 }
212
357
358 // addTask adds a task to a random partition, computing its pass via the counter.
359 > func (s *state) addTask(t *task) { sim.go
360 > partition := &s.partitions[s.rnd.IntN(len(s.partitions))]
361 >
362 > if partition.perPri == nil {
363 > partition.perPri = make(map[int]perPriState)
364 > }
365
366 > priState, exists := partition.perPri[t.pri] sim.go
367 > if !exists {
368 > priState = perPriState{c: s.counterFactory()}
369 > partition.perPri[t.pri] = priState
370 > }
371
372 > t.pass = priState.c.GetPass(t.fkey, 0, max(1, int64(s.strideFactor/t.fweight))) sim.go
373 > heap.Push(&partition.heap, t)
374 }
375
376 // popTask returns the task with minimum (pri, pass, index) from a random partition.
377 > func (s *state) popTask() (*task, int) { sim.go
378 > for _, idx := range s.rnd.Perm(len(s.partitions)) {
379 > partition := &s.partitions[idx]
380 > if partition.heap.Len() > 0 {
381 > t := heap.Pop(&partition.heap).(*task) //nolint:revive
382 > return t, idx
383 > }
384 }
385 > return nil, -1 sim.go
386 }
387
394 // --- taskHeap (heap.Interface) ---
395
396 > func (h taskHeap) Len() int { return len(h) } sim.go
397
398 > func (h taskHeap) Less(i, j int) bool { sim.go
399 > if h[i].pri != h[j].pri {
400 return h[i].pri < h[j].pri
401 }
402 > if h[i].pass != h[j].pass { sim.go
403 > return h[i].pass < h[j].pass
404 > }
405 > return h[i].index < h[j].index
406 }
407
408 > func (h taskHeap) Swap(i, j int) { h[i], h[j] = h[j], h[i] } sim.go
409
410 > func (h *taskHeap) Push(x any) { *h = append(*h, x.(*task)) } sim.go
411
412 > func (h *taskHeap) Pop() any { sim.go
413 > old := *h
414 > n := len(old)
415 > item := old[n-1]
416 > *h = old[:n-1]
417 > return item
418 > }
419
420 // --- latencyStats ---
421
422 > func (stats *latencyStats) calculateNormalized() { sim.go
423 > totalTasks := len(stats.overall)
424 > totalKeys := len(stats.byKey)
425 > if totalTasks == 0 || totalKeys == 0 {
426 return
427 }
431 // of total traffic, making values comparable across different workload sizes.
432 // Values center on 0: negative = expedited, positive = delayed.
433 > fairShareFactor := float64(totalKeys) / float64(totalTasks) sim.go
434 > stats.overallNormalized = stats.overallNormalized[:0]
435 > for key, latencies := range stats.byKey {
436 > taskCount := float64(len(latencies))
437 > normalizedLatencies := make([]float64, len(latencies))
438 > for i, rawLatency := range latencies {
439 > normalizedLatencies[i] = float64(rawLatency) / taskCount * fairShareFactor
440 > stats.overallNormalized = append(stats.overallNormalized, normalizedLatencies[i])
441 > }
442 > stats.byKeyNormalized[key] = normalizedLatencies
443 }
444 }
581 }
582
583 > func percentileOfPercentiles[T int64 | float64](dataByKey map[string][]T, keyPercentile float64, crossPercentile []float64) []float64 { sim.go
584 > var keyPercentiles []float64
585 >
586 > for _, values := range dataByKey {
587 > if len(values) == 0 {
588 continue
589 }
590 > sorted := make([]float64, len(values)) sim.go
591 > for i, v := range values {
592 > sorted[i] = float64(v)
593 > }
594 > slices.Sort(sorted)
595 > idx := int(keyPercentile / 100.0 * float64(len(sorted)-1))
596 > keyPercentiles = append(keyPercentiles, sorted[idx])
597 }
598
599 > if len(keyPercentiles) == 0 { sim.go
600 return nil
601 }
602
603 > slices.Sort(keyPercentiles) sim.go
604 >
605 > out := make([]float64, len(crossPercentile))
606 > for i, p := range crossPercentile {
607 > idx := int(p / 100.0 * float64(len(keyPercentiles)-1))
608 > out[i] = keyPercentiles[idx]
609 > }
610 > return out
611 }
go.temporal.io/server/service/matching/counter/cmsketch.go 107 covered LOC · 24 ranges

Open complete file

59 var _ Counter = (*cmSketch)(nil)
60
61 > func NewCMSketchCounter(params CMSketchParams, src rand.Source, topKProvider topKFunc) *cmSketch { cmsketch.go
62 > params.D = max(1, params.D)
63 > params.W = max(1, params.W)
64 > params.Grow.SkipRateDecay = max(1_000, params.Grow.SkipRateDecay)
65 > numRows := params.D + 1 // + 1 for shadow row
66 > return &cmSketch{
67 > params: params,
68 > seed0: maphash.MakeSeed(),
69 > seeds: makeSeeds(numRows, src),
70 > cells: make([]uint32, params.W*numRows),
71 > shadowRow: 0,
72 > src: src,
73 > topKProvider: topKProvider,
74 > }
75 > }
76
77 > func (s *cmSketch) GetPass(key string, base, inc int64) int64 { cmsketch.go
78 > if inc < 0 {
79 return base // we don't handle negatives here
80 }
81
82 > numRows := s.params.D + 1 cmsketch.go
83 > indexes := make([]int, numRows)
84 > s.fillIndexes(key, indexes)
85 >
86 > current := s.getByIndexes(indexes)
87 > pass := max(base, current+inc)
88 > s.skips += s.ensureByIndexes(indexes, pass)
89 >
90 > if s.incs++; s.incs > s.params.Grow.SkipRateDecay {
91 > s.maybeGrow() cmsketch.go
92 > s.skips >>= 1
93 > s.incs >>= 1
94 > }
95
96 > if s.reseedOps++; s.params.Reseed.Interval > 0 && s.reseedOps >= s.params.Reseed.Interval { cmsketch.go
97 s.reseed()
98 s.reseedOps = 0
99 }
100
101 > return int64(pass) cmsketch.go
102 }
103
104 > func (s *cmSketch) SkipRate() float64 { cmsketch.go
105 > return float64(s.skips) / float64(s.incs*s.params.D)
106 > }
107
108 func (s *cmSketch) EstimateDistinctKeys() int {
124 // fillIndexes computes cell indexes for all D+1 rows (D active + 1 shadow).
125 // len(indexes) must == len(s.seeds) == D+1
126 > func (s *cmSketch) fillIndexes(k string, indexes []int) { cmsketch.go
127 > w := s.params.W
128 > // get 64 bits of hash
129 > h0 := maphash.String(s.seed0, k)
130 >
131 > for i, seed := range s.seeds {
132 > h1 := bits.RotateLeft64(h0, i*39)
133 > h2l := mix(uint32(h1), uint32(seed))
134 > h2h := mix(uint32(h1>>32), uint32(seed>>32))
135 > h3 := mix(h2l, h2h)
136 > // https://lemire.me/blog/2016/06/27/a-fast-alternative-to-the-modulo-reduction/
137 > indexes[i] = i*w + int((uint64(h3)*uint64(w))>>32)
138 > }
139 }
140
141 > func (s *cmSketch) maybeGrow() { cmsketch.go
142 > if s.params.Grow.Threshold == 0 ||
143 > s.params.Grow.Ratio == 0 ||
144 > s.params.W >= s.params.Grow.MaxW ||
145 > s.SkipRate() < s.params.Grow.Threshold {
146 > return cmsketch.go
147 > }
148 // get top entries before resetting (if provider available)
149 > var topK []TopKEntry cmsketch.go
150 > if s.topKProvider != nil {
151 > topK = s.topKProvider() cmsketch.go
152 > }
153
154 > numRows := s.params.D + 1 cmsketch.go
155 > s.params.W = min(int(float64(s.params.W)*s.params.Grow.Ratio), s.params.Grow.MaxW)
156 > s.seed0 = maphash.MakeSeed()
157 > // we're resetting everything so might as well reseed now too
158 > s.seeds = makeSeeds(numRows, s.src)
159 > s.base = 0
160 > s.cells = make([]uint32, s.params.W*numRows)
161 > s.shadowRow = s.params.D // reset shadow to last row
162 > s.skips, s.incs, s.reseedOps = 0, 0, 0
163 >
164 > if len(topK) > 0 {
165 > // restore top entries after resize. GetPass can in theory call back into maybeGrow, cmsketch.go
166 > // but we can just reset the counters each time to prevent that.
167 > for _, entry := range topK {
168 > _ = s.GetPass(entry.Key, entry.Count, 0)
169 > s.skips, s.incs, s.reseedOps = 0, 0, 0
170 > }
171 }
172 }
188 }
189
190 > func (s *cmSketch) getByIndexes(indexes []int) int64 { cmsketch.go
191 > // TODO: consider using better estimator: https://dl.acm.org/doi/pdf/10.1145/3219819.3219975
192 > minVal := uint32(math.MaxUint32)
193 > for i, idx := range indexes {
194 > if i == s.shadowRow {
195 > continue // skip shadow row for reads
196 }
197 > minVal = min(minVal, s.cells[idx]) cmsketch.go
198 }
199 > return s.base + int64(minVal) cmsketch.go
200 }
201
202 > func (s *cmSketch) ensureByIndexes(indexes []int, target int64) (skips int) { cmsketch.go
203 > offset := target - s.base
204 > if offset < 0 {
205 // target is below our window floor, all cells are already high enough
206 return s.params.D // only count active rows for skips
207 }
208 > if offset > math.MaxUint32 { cmsketch.go
209 // would overflow uint32, need to slide the base up first
210 s.slideBase(offset + slideHeadroom - math.MaxUint32)
212 }
213
214 > uoffset := uint32(offset) cmsketch.go
215 > for i, idx := range indexes {
216 > if s.cells[idx] < uoffset {
217 > s.cells[idx] = uoffset
218 > } else if i != s.shadowRow {
219 > skips++ // only count skips for active rows, not shadow cmsketch.go
220 > }
221 }
222 > return cmsketch.go
223 }
224
247 }
248
249 > func makeSeeds(rows int, src rand.Source) []uint64 { cmsketch.go
250 > out := make([]uint64, rows)
251 > for i := range out {
252 > out[i] = src.Uint64()
253 > }
254 > return out
255 }
256
257 // from https://www.pcg-random.org/posts/developing-a-seed_seq-alternative.html
258 > func mix(x, y uint32) uint32 { cmsketch.go
259 > result := 0xca01f9dd*x - 0x4973f715*y
260 > result ^= result >> 16
261 > return result
262 > }
go.temporal.io/server/service/matching/counter/map.go 53 covered LOC · 13 ranges

Open complete file

18
19 // NewMapCounter creates a mapCounter that also tracks the top K entries.
20 > func NewMapCounter(limit int) *mapCounter { map.go
21 > return &mapCounter{
22 > m: make(map[string]int),
23 > limit: limit,
24 > }
25 > }
26
27 func (m *mapCounter) GetPass(key string, base, inc int64) int64 {
30 }
31
32 > func (m *mapCounter) getPassWithOverflow(key string, base, inc int64) (int64, bool) { map.go
33 > if idx, ok := m.m[key]; ok {
34 prev := m.heap[idx].Count
35 count := max(base, prev+inc)
40 }
41 // not present, fall back to full updateHeap
42 > count := max(base, inc) map.go
43 > return count, m.updateHeap(key, count)
44 }
45
49
50 // TopK returns the top-K entries by count.
51 > func (m *mapCounter) TopK() []TopKEntry { map.go
52 > return slices.Clone(m.heap)
53 > }
54
55 > func (m *mapCounter) updateHeap(key string, count int64) bool { map.go
56 > if idx, ok := m.m[key]; ok {
57 > // already in heap - update count and fix map.go
58 > m.heap[idx].Count = count
59 > heap.Fix(m, idx)
60 > return false
61 > }
62
63 > if len(m.heap) < m.limit { map.go
64 > // heap not full - add
65 > m.m[key] = len(m.heap)
66 > heap.Push(m, TopKEntry{Key: key, Count: count})
67 > return false
68 > }
69
70 // heap is full - only add if count > min
71 > if count > m.heap[0].Count { map.go
72 > // evict min
73 > evicted := heap.Pop(m).(TopKEntry) // nolint:revive // unchecked-type-assertion
74 > delete(m.m, evicted.Key)
75 > // add new
76 > m.m[key] = len(m.heap)
77 > heap.Push(m, TopKEntry{Key: key, Count: count})
78 > }
79 > return true
80 }
81
82 // implements heap.Interface using m.heap
83 > func (m *mapCounter) Len() int { return len(m.heap) } map.go
84 > func (m *mapCounter) Less(i, j int) bool { return m.heap[i].Count < m.heap[j].Count } map.go
85 > func (m *mapCounter) Swap(i, j int) { map.go
86 > m.heap[i], m.heap[j] = m.heap[j], m.heap[i]
87 > // don't forget to fix the map:
88 > m.m[m.heap[i].Key] = i
89 > m.m[m.heap[j].Key] = j
90 > }
91
92 > func (m *mapCounter) Push(x any) { map.go
93 > m.heap = append(m.heap, x.(TopKEntry))
94 > }
95
96 > func (m *mapCounter) Pop() any { map.go
97 > n := len(m.heap)
98 > entry := m.heap[n-1]
99 > m.heap[n-1] = TopKEntry{}
100 > m.heap = m.heap[0 : n-1]
101 > return entry
102 > }
go.temporal.io/server/service/matching/counter/hybrid.go 25 covered LOC · 7 ranges

Open complete file

42 }
43
44 > func NewHybridCounter(params CounterParams, src rand.Source) *hybridCounter { hybrid.go
45 > return &hybridCounter{
46 > mapCounter: *NewMapCounter(params.MapLimit),
47 > params: params,
48 > src: src,
49 > }
50 > }
51
52 > func (h *hybridCounter) GetPass(key string, base int64, inc int64) int64 { hybrid.go
53 > if h.cmSketch != nil {
54 > p := h.cmSketch.GetPass(key, base, inc) hybrid.go
55 > // after migration, continue updating top-K tracker
56 > _ = h.mapCounter.updateHeap(key, p)
57 > return p
58 > }
59
60 > p, overflow := h.mapCounter.getPassWithOverflow(key, base, inc) hybrid.go
61 > if overflow {
62 > h.migrateToCMS() hybrid.go
63 > }
64 > return p hybrid.go
65 }
66
67 > func (h *hybridCounter) migrateToCMS() { hybrid.go
68 > h.cmSketch = NewCMSketchCounter(h.params.CMS, h.src, h.mapCounter.TopK)
69 > // move existing counts into CMS
70 > for _, entry := range h.mapCounter.heap {
71 > _ = h.cmSketch.GetPass(entry.Key, entry.Count, 0)
72 > }
73 }
74