Vihren Atlas

Atlas › Test

TestFairness_MapCounter

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

Package
go.temporal.io/server/tools/fairsim
Suite / test hierarchy
TestFairness_MapCounter
Test
TestFairness_MapCounter
Introduced at
TestFairness_MapCounter, TestFairness_UniformDistribution_50 Test frontier
Covered ranges
38
Covered lines
166
Covered files
3

Co-introduced tests

1 other test enter at the same concept.

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/map.go 36 covered LOC · 10 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 map.go
35 > count := max(base, prev+inc)
36 > // inline simple case of updateHeap
37 > m.heap[idx].Count = count
38 > heap.Fix(m, idx)
39 > return count, false
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
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
58 m.heap[idx].Count = count
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
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 {
go.temporal.io/server/service/matching/counter/hybrid.go 12 covered LOC · 4 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)
55 // after migration, continue updating top-K tracker
58 }
59
60 > p, overflow := h.mapCounter.getPassWithOverflow(key, base, inc) hybrid.go
61 > if overflow {
62 h.migrateToCMS()
63 }
64 > return p hybrid.go
65 }
66