Vihren Atlas

Atlas › Test

TestHybridCounter_Migrates

go.temporal.io/server/service/matching/counter/TestHybridCounter_Migrates

Package
go.temporal.io/server/service/matching/counter
Suite / test hierarchy
TestHybridCounter_Migrates
Test
TestHybridCounter_Migrates
Introduced at
hybrid.go ×2 Joint frontier
Covered ranges
32
Covered lines
135
Covered files
3

Covered source

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

go.temporal.io/server/service/matching/counter/cmsketch.go 70 covered LOC · 15 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()
92 s.skips >>= 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
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
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
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 45 covered LOC · 11 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
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
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 20 covered LOC · 6 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() 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