Vihren Atlas

Atlas › Test

TestCMSketch_Grow_PreservedOnResize

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

Package
go.temporal.io/server/service/matching/counter
Suite / test hierarchy
TestCMSketch_Grow_PreservedOnResize
Test
TestCMSketch_Grow_PreservedOnResize
Introduced at
cmsketch.go ×2 Joint frontier
Covered ranges
23
Covered lines
105
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 105 covered LOC · 23 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
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 > }