sketch_misra_gries.nx source
↩ module page · 225 lines · 7388 B
1// sketch_misra_gries.nx -- Misra-Gries frequent items (1982).
2//
3// Bounded set of (key, count) pairs of capacity (k-1). On arrival
4// of (key, w):
5// - if key tracked: count[key] += w
6// - else if free slot: install (key, w)
7// - else: decrement EVERY counter by w (with floor 0);
8// counters that reach 0 are freed for future use
9//
10// GUARANTEE (different from SpaceSaving):
11// true_count - N/k <= estimate <= true_count
12// Misra-Gries UNDERESTIMATES; SpaceSaving OVERESTIMATES. Both bound
13// the error by N/k. Symmetric error semantics let callers pick:
14// - SpaceSaving for upper-bound queries ("at most X")
15// - Misra-Gries for lower-bound queries ("at least X")
16//
17// CAPABILITY STOMP vs DataSketches:
18// DS ships only the "frequent items" variant (SpaceSaving-like).
19// We ship BOTH algorithmic approaches (SpaceSaving + Misra-Gries)
20// so callers pick by error-bound direction.
21//
22// LOSSLESS-LANGUAGE DISCIPLINE:
23// nx_mg_query returns ApproxI64 with NX_ENV_ABS; param_a = N/k
24// (the maximum possible UNDERESTIMATE); conf_ppb = 1e9 (deterministic).
25
26import "syscalls.nx"
27import "sketch_types.nx"
28
29const NX_MG_K_MIN: i64 = 2
30const NX_MG_K_MAX: i64 = 100000
31
32struct MgCounter {
33 key: i64,
34 count: i64,
35}
36
37struct MisraGries {
38 counters: *MgCounter, // k-1 slots
39 k: i64, // user-facing K; capacity = K - 1
40 n_tracked: i64, // <= k-1
41 total_seen: i64, // total weighted insertions
42}
43
44// === construction =================================================
45
46func nx_mg_alloc(k: i64) -> *MisraGries {
47 if k < NX_MG_K_MIN { return 0 as *MisraGries }
48 if k > NX_MG_K_MAX { return 0 as *MisraGries }
49 let raw: *u8 = sys_mmap(40)
50 let mg: *MisraGries = raw as *MisraGries
51 mg.k = k
52 let cap: i64 = k - 1
53 let ent_raw: *u8 = sys_mmap(cap * 16)
54 mg.counters = ent_raw as *MgCounter
55 var i: i64 = 0
56 while i < cap {
57 let c: *MgCounter = (mg.counters as i64 + i * 16) as *MgCounter
58 c.key = 0
59 c.count = 0
60 i = i + 1
61 }
62 mg.n_tracked = 0
63 mg.total_seen = 0
64 return mg
65}
66
67func nx_mg_counter_at(mg: *MisraGries, i: i64) -> *MgCounter {
68 return (mg.counters as i64 + i * 16) as *MgCounter
69}
70
71// === find helpers =================================================
72
73func nx_mg_find_key(mg: *MisraGries, key: i64) -> i64 {
74 var i: i64 = 0
75 while i < mg.n_tracked {
76 let c: *MgCounter = nx_mg_counter_at(mg, i)
77 if c.key == key { return i }
78 i = i + 1
79 }
80 return -1
81}
82
83func nx_mg_find_empty(mg: *MisraGries) -> i64 {
84 // After decrement-and-floor, slots with count==0 are free.
85 // Returns first free slot or -1 if all occupied.
86 var i: i64 = 0
87 while i < mg.n_tracked {
88 let c: *MgCounter = nx_mg_counter_at(mg, i)
89 if c.count == 0 { return i }
90 i = i + 1
91 }
92 if mg.n_tracked < mg.k - 1 { return mg.n_tracked }
93 return -1
94}
95
96// === add ==========================================================
97//
98// Add `count` occurrences of `key`. Three cases:
99// 1. key tracked: increment by count.
100// 2. free slot: install (key, count).
101// 3. no slot: decrement all counters by `dec` where
102// dec = min(count, min_count_in_table). Any counter
103// reaching 0 is freed. If dec < count, recurse
104// with remaining count via the now-free slot.
105
106func if_le(a: i64, b: i64) -> i64 {
107 if a < b { return a }
108 return b
109}
110
111func nx_mg_min_count(mg: *MisraGries) -> i64 {
112 if mg.n_tracked == 0 { return 0 }
113 let c0: *MgCounter = nx_mg_counter_at(mg, 0)
114 var m: i64 = c0.count
115 var i: i64 = 1
116 while i < mg.n_tracked {
117 let c: *MgCounter = nx_mg_counter_at(mg, i)
118 if c.count < m { m = c.count }
119 i = i + 1
120 }
121 return m
122}
123
124func nx_mg_add(mg: *MisraGries, key: i64, count: i64) -> i64 {
125 if count <= 0 { return 0 }
126 mg.total_seen = mg.total_seen + count
127
128 let pos: i64 = nx_mg_find_key(mg, key)
129 if pos >= 0 {
130 let c: *MgCounter = nx_mg_counter_at(mg, pos)
131 c.count = c.count + count
132 return 0
133 }
134 // Not tracked. Try free slot.
135 let empty_slot: i64 = nx_mg_find_empty(mg)
136 if empty_slot >= 0 {
137 let c: *MgCounter = nx_mg_counter_at(mg, empty_slot)
138 c.key = key
139 c.count = count
140 if empty_slot >= mg.n_tracked {
141 mg.n_tracked = mg.n_tracked + 1
142 }
143 return 0
144 }
145 // All k-1 slots full and non-zero. Decrement-loop:
146 // absorb = min(count, min_count); decrement everyone by absorb;
147 // remaining count = count - absorb. If remaining > 0, recurse
148 // (the decrement freed at least one slot).
149 var remaining: i64 = count
150 while remaining > 0 {
151 let min_c: i64 = nx_mg_min_count(mg)
152 if min_c == 0 {
153 // A free slot exists -- install there.
154 let slot: i64 = nx_mg_find_empty(mg)
155 if slot < 0 {
156 // Shouldn't happen if min == 0 and n_tracked == k-1.
157 return -1
158 }
159 let c: *MgCounter = nx_mg_counter_at(mg, slot)
160 c.key = key
161 c.count = remaining
162 remaining = 0
163 }
164 if min_c > 0 {
165 let absorb: i64 = if_le(remaining, min_c)
166 // Decrement every counter.
167 var i: i64 = 0
168 while i < mg.n_tracked {
169 let c: *MgCounter = nx_mg_counter_at(mg, i)
170 c.count = c.count - absorb
171 i = i + 1
172 }
173 remaining = remaining - absorb
174 }
175 }
176 return 0
177}
178
179// === query ========================================================
180//
181// estimate(key) = stored count if tracked, else 0.
182// Both are LOWER bounds on the true count (Misra-Gries underestimates).
183
184func nx_mg_estimate(mg: *MisraGries, key: i64) -> i64 {
185 let pos: i64 = nx_mg_find_key(mg, key)
186 if pos < 0 { return 0 }
187 let c: *MgCounter = nx_mg_counter_at(mg, pos)
188 if c.count < 0 { return 0 }
189 return c.count
190}
191
192// Maximum possible UNDERCOUNT: bounded by N/k per Misra-Gries.
193func nx_mg_max_undercount(mg: *MisraGries) -> i64 {
194 return (mg.total_seen + mg.k - 1) / mg.k
195}
196
197// Upper bound on true count = estimate + max_undercount.
198func nx_mg_upper_bound(mg: *MisraGries, key: i64) -> i64 {
199 return nx_mg_estimate(mg, key) + nx_mg_max_undercount(mg)
200}
201
202func nx_mg_query(mg: *MisraGries, key: i64) -> *ApproxI64 {
203 let est: i64 = nx_mg_estimate(mg, key)
204 return nx_approx_new(est, NX_ENV_ABS, nx_mg_max_undercount(mg),
205 1000000000,
206 NX_MATURITY_REFERENCE_IMPL,
207 NX_ADV_HONEST)
208}
209
210// === introspection ================================================
211
212func nx_mg_n_tracked(mg: *MisraGries) -> i64 {
213 var n: i64 = 0
214 var i: i64 = 0
215 while i < mg.n_tracked {
216 let c: *MgCounter = nx_mg_counter_at(mg, i)
217 if c.count > 0 { n = n + 1 }
218 i = i + 1
219 }
220 return n
221}
222
223func nx_mg_memory_bytes(mg: *MisraGries) -> i64 {
224 return 40 + (mg.k - 1) * 16
225}