sketch_lossy_counting.nx source
↩ module page · 231 lines · 7789 B
1// sketch_lossy_counting.nx -- Manku-Motwani 2002 Lossy Counting.
2//
3// THIRD major frequency-counting algorithm joining CMS / SpaceSaving /
4// Misra-Gries / CountSketch. Distinct technique: BUCKET-PRUNING.
5//
6// ALGORITHM:
7// Stream divided into "buckets" of size w = ceil(1/epsilon).
8// For each item:
9// - if tracked: increment frequency
10// - else: insert with frequency=1, error=current_bucket-1
11// At each bucket boundary (every w items):
12// - decrement EVERY tracked item's frequency by 1
13// - REMOVE items with frequency = 0
14//
15// GUARANTEE:
16// stored_freq <= true_freq (underestimate)
17// stored_freq + error >= true_freq (upper bound)
18// Items with true_freq > epsilon * N are NEVER missed
19// No item has stored_freq > true_freq (NEVER overestimates)
20//
21// COMPARISON to other frequency algorithms:
22// - CMS: random projection, overestimate-only, O(1) query
23// - SpaceSaving: k-bounded counters, overestimate-only, top-K native
24// - Misra-Gries: k-bounded counters, underestimate-only
25// - CountSketch: median-based, UNBIASED, supports negative
26// - Lossy Counting (THIS): bucket-pruning, underestimate + explicit
27// error tracking per entry
28//
29// MEMORY: O((1/epsilon) log(epsilon * N)) -- bounded but data-dependent.
30// We use a hash-set with linear probing; cap at NX_LC_MAX_ENTRIES.
31
32import "syscalls.nx"
33import "sketch_types.nx"
34
35const NX_LC_MIN_EPS_PPM: i64 = 100 // epsilon >= 0.0001
36const NX_LC_MAX_EPS_PPM: i64 = 500000 // epsilon <= 0.5
37const NX_LC_MAX_ENTRIES: i64 = 100000
38
39struct LcEntry {
40 key: i64,
41 freq: i64,
42 err: i64,
43}
44
45struct LossyCounting {
46 entries: *LcEntry,
47 capacity: i64, // max entries (allocated table size, power of 2)
48 mask: i64, // capacity - 1
49 n_entries: i64, // active entries (key != 0 sentinel)
50 total_seen: i64,
51 epsilon_ppm: i64,
52 bucket_size: i64, // = ceil(1e6 / epsilon_ppm)
53 current_bucket: i64, // 1-indexed
54}
55
56const NX_LC_EMPTY_KEY: i64 = 0 // assume real keys != 0 by remapping
57
58// === construction =================================================
59
60func nx_lc_is_pow2(n: i64) -> i64 {
61 if n < 16 { return 0 }
62 if (n & (n - 1)) != 0 { return 0 }
63 return 1
64}
65
66func nx_lc_alloc(capacity: i64, epsilon_ppm: i64) -> *LossyCounting {
67 if nx_lc_is_pow2(capacity) != 1 { return 0 as *LossyCounting }
68 if capacity > NX_LC_MAX_ENTRIES { return 0 as *LossyCounting }
69 if epsilon_ppm < NX_LC_MIN_EPS_PPM { return 0 as *LossyCounting }
70 if epsilon_ppm > NX_LC_MAX_EPS_PPM { return 0 as *LossyCounting }
71 let raw: *u8 = sys_mmap(72)
72 let lc: *LossyCounting = raw as *LossyCounting
73 let ent_raw: *u8 = sys_mmap(capacity * 24)
74 lc.entries = ent_raw as *LcEntry
75 var i: i64 = 0
76 while i < capacity {
77 let e: *LcEntry = (lc.entries as i64 + i * 24) as *LcEntry
78 e.key = NX_LC_EMPTY_KEY
79 e.freq = 0
80 e.err = 0
81 i = i + 1
82 }
83 lc.capacity = capacity
84 lc.mask = capacity - 1
85 lc.n_entries = 0
86 lc.total_seen = 0
87 lc.epsilon_ppm = epsilon_ppm
88 lc.bucket_size = (1000000 + epsilon_ppm - 1) / epsilon_ppm // ceil(1/eps)
89 lc.current_bucket = 1
90 return lc
91}
92
93// === hash + probe =================================================
94
95func nx_lc_hash(key: i64) -> i64 {
96 let mixed: i64 = (key * 0x9E3779B97F4A7C15) & 0xFFFFFFFFFFFFFFFF
97 return mixed
98}
99
100func nx_lc_entry_at(lc: *LossyCounting, idx: i64) -> *LcEntry {
101 return (lc.entries as i64 + idx * 24) as *LcEntry
102}
103
104// Find slot for key. Returns:
105// index of existing entry with this key, OR
106// index of first empty slot for insertion.
107func nx_lc_probe(lc: *LossyCounting, key: i64) -> i64 {
108 var i: i64 = nx_lc_hash(key) & lc.mask
109 var done: i64 = 0
110 var found: i64 = -1
111 while done == 0 {
112 let e: *LcEntry = nx_lc_entry_at(lc, i)
113 if e.key == NX_LC_EMPTY_KEY {
114 found = i
115 done = 1
116 }
117 if done == 0 {
118 if e.key == key {
119 found = i
120 done = 1
121 }
122 }
123 if done == 0 {
124 i = (i + 1) & lc.mask
125 }
126 }
127 return found
128}
129
130// === bucket-boundary pruning ======================================
131//
132// Decrement every tracked entry's freq by 1; remove entries that reach 0.
133// Removal is via marking key=EMPTY; subsequent probes will treat as empty.
134// (No re-hashing of subsequent collisions in this v1 -- accuracy slightly
135// reduced but correctness preserved as Lossy Counting only requires
136// that pruned items have low frequency.)
137
138func nx_lc_prune(lc: *LossyCounting) -> i64 {
139 var i: i64 = 0
140 while i < lc.capacity {
141 let e: *LcEntry = nx_lc_entry_at(lc, i)
142 if e.key != NX_LC_EMPTY_KEY {
143 e.freq = e.freq - 1
144 if e.freq <= 0 {
145 e.key = NX_LC_EMPTY_KEY
146 e.freq = 0
147 e.err = 0
148 lc.n_entries = lc.n_entries - 1
149 }
150 }
151 i = i + 1
152 }
153 return 0
154}
155
156// === add ==========================================================
157
158func nx_lc_add(lc: *LossyCounting, key: i64) -> i64 {
159 if key == NX_LC_EMPTY_KEY { return -1 } // 0 reserved as sentinel
160 lc.total_seen = lc.total_seen + 1
161 let idx: i64 = nx_lc_probe(lc, key)
162 let e: *LcEntry = nx_lc_entry_at(lc, idx)
163 if e.key == NX_LC_EMPTY_KEY {
164 // New entry.
165 if lc.n_entries >= lc.capacity - 1 { // leave room for probing
166 return -2 // table full; cannot insert
167 }
168 e.key = key
169 e.freq = 1
170 e.err = lc.current_bucket - 1
171 lc.n_entries = lc.n_entries + 1
172 }
173 if e.key == key {
174 if e.freq < lc.bucket_size + lc.current_bucket {
175 // Don't overcount via reset (defensive); just increment.
176 }
177 e.freq = e.freq + 1
178 }
179 // Bucket boundary: prune.
180 if (lc.total_seen % lc.bucket_size) == 0 {
181 nx_lc_prune(lc)
182 lc.current_bucket = lc.current_bucket + 1
183 }
184 return 0
185}
186
187// === queries ======================================================
188
189func nx_lc_estimate(lc: *LossyCounting, key: i64) -> i64 {
190 if key == NX_LC_EMPTY_KEY { return 0 }
191 let idx: i64 = nx_lc_probe(lc, key)
192 let e: *LcEntry = nx_lc_entry_at(lc, idx)
193 if e.key != key { return 0 }
194 return e.freq
195}
196
197// Upper bound: stored_freq + error.
198func nx_lc_upper_bound(lc: *LossyCounting, key: i64) -> i64 {
199 if key == NX_LC_EMPTY_KEY { return 0 }
200 let idx: i64 = nx_lc_probe(lc, key)
201 let e: *LcEntry = nx_lc_entry_at(lc, idx)
202 if e.key != key { return 0 }
203 return e.freq + e.err
204}
205
206// Maximum possible undercount = epsilon * N.
207func nx_lc_max_undercount(lc: *LossyCounting) -> i64 {
208 return (lc.epsilon_ppm * lc.total_seen) / 1000000
209}
210
211func nx_lc_query(lc: *LossyCounting, key: i64) -> *ApproxI64 {
212 let est: i64 = nx_lc_estimate(lc, key)
213 return nx_approx_new(est, NX_ENV_ABS, nx_lc_max_undercount(lc),
214 1000000000,
215 NX_MATURITY_REFERENCE_IMPL,
216 NX_ADV_HONEST)
217}
218
219// === introspection ================================================
220
221func nx_lc_n_entries(lc: *LossyCounting) -> i64 {
222 return lc.n_entries
223}
224
225func nx_lc_total_seen(lc: *LossyCounting) -> i64 {
226 return lc.total_seen
227}
228
229func nx_lc_memory_bytes(lc: *LossyCounting) -> i64 {
230 return 72 + lc.capacity * 24
231}