Vihren Atlas

Atlas › Test

TestCMSketch_SlideBase_MultipleSlides

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

Package
go.temporal.io/server/service/matching/counter
Suite / test hierarchy
TestCMSketch_SlideBase_MultipleSlides
Test
TestCMSketch_SlideBase_MultipleSlides
Introduced at
cmsketch.go ×3 Joint frontier
Covered ranges
20
Covered lines
84
Covered files
1

Covered source

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

go.temporal.io/server/service/matching/counter/cmsketch.go 84 covered LOC · 20 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 cmsketch.go
210 > s.slideBase(offset + slideHeadroom - math.MaxUint32)
211 > offset = math.MaxUint32 - slideHeadroom
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
225 // slideBase increases the base by delta, subtracting delta from all cells.
226 // Cells that would go negative are clamped to 0 (those keys are "dragged up").
227 > func (s *cmSketch) slideBase(delta int64) { cmsketch.go
228 > if delta <= 0 {
229 return
230 }
231 > s.base += delta cmsketch.go
232 > if delta >= math.MaxUint32 {
233 for i := range s.cells {
234 s.cells[i] = 0
237 }
238 // delta fits in uint32
239 > udelta := uint32(delta) cmsketch.go
240 > for i := range s.cells {
241 > if s.cells[i] > udelta {
242 s.cells[i] -= udelta
243 > } else { cmsketch.go
244 > s.cells[i] = 0
245 > }
246 }
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 > }