sketch_sliding_window.nx source
↩ module page · 181 lines · 5234 B
1// sketch_sliding_window.nx -- last-W-samples rolling aggregator.
2//
3// Circular buffer of size W with running sum + lazy min/max tracking.
4// Useful for "last 5 minutes / last 100 events" rolling computations
5// when the EXACT window is required (vs EWMA's exponential decay).
6//
7// AGGREGATES MAINTAINED:
8// sum -- running sum (O(1) update)
9// count -- min(window_size, observations) — useful at warm-up
10// min -- minimum (lazy: recompute via full scan when displaced)
11// max -- maximum (lazy: same)
12//
13// USE CASES:
14// - SRE: last-5-minute p99 latency window
15// - finance: 20-period moving average
16// - sensor: rolling-window anomaly thresholds
17// - rate-limiting: requests per fixed window
18//
19// MIN/MAX LAZY STRATEGY: cheaper than monotonic-deque for typical W.
20// Cached min/max valid until the displaced sample equaled the cached
21// value; then we rescan the buffer. Worst-case O(W) on eviction, O(1)
22// amortized for "no extremum eviction." For deterministic O(1)-per-op
23// monotonic-deque variant, swap in v2.
24
25import "syscalls.nx"
26import "sketch_types.nx"
27
28const NX_SW_MIN_W: i64 = 2
29const NX_SW_MAX_W: i64 = 1000000
30
31struct SlidingWindow {
32 buffer: *i64, // circular ring of W slots
33 window: i64, // capacity W
34 count: i64, // # samples (capped at W)
35 head: i64, // next write position
36 sum: i64,
37 cached_min: i64,
38 cached_max: i64,
39 cache_valid: i64, // 1 if cached_min/cached_max valid
40}
41
42// === construction =================================================
43
44func nx_sw_alloc(window: i64) -> *SlidingWindow {
45 if window < NX_SW_MIN_W { return 0 as *SlidingWindow }
46 if window > NX_SW_MAX_W { return 0 as *SlidingWindow }
47 let raw: *u8 = sys_mmap(72)
48 let s: *SlidingWindow = raw as *SlidingWindow
49 s.buffer = sys_mmap(window * 8) as *i64
50 var i: i64 = 0
51 while i < window {
52 s.buffer[i] = 0
53 i = i + 1
54 }
55 s.window = window
56 s.count = 0
57 s.head = 0
58 s.sum = 0
59 s.cached_min = 0
60 s.cached_max = 0
61 s.cache_valid = 0
62 return s
63}
64
65// === recompute min/max via full scan ============================
66
67func nx_sw_recompute_extrema(s: *SlidingWindow) -> i64 {
68 if s.count == 0 {
69 s.cached_min = 0
70 s.cached_max = 0
71 s.cache_valid = 0
72 return 0
73 }
74 var mn: i64 = s.buffer[0]
75 var mx: i64 = s.buffer[0]
76 var i: i64 = 1
77 while i < s.count {
78 let v: i64 = s.buffer[i]
79 if v < mn { mn = v }
80 if v > mx { mx = v }
81 i = i + 1
82 }
83 s.cached_min = mn
84 s.cached_max = mx
85 s.cache_valid = 1
86 return 0
87}
88
89// === push ==========================================================
90
91func nx_sw_push(s: *SlidingWindow, value: i64) -> i64 {
92 var evicted: i64 = 0
93 var has_evicted: i64 = 0
94 if s.count >= s.window {
95 evicted = s.buffer[s.head]
96 has_evicted = 1
97 s.sum = s.sum - evicted
98 }
99 if s.count < s.window {
100 s.count = s.count + 1
101 }
102 s.buffer[s.head] = value
103 s.sum = s.sum + value
104 s.head = (s.head + 1) % s.window
105
106 // Invalidate cached extrema only if the evicted value was an extremum.
107 if has_evicted == 1 {
108 if s.cache_valid == 1 {
109 if evicted == s.cached_min { s.cache_valid = 0 }
110 if evicted == s.cached_max { s.cache_valid = 0 }
111 }
112 }
113 // Track new value against cache (cheap).
114 if s.cache_valid == 1 {
115 if value < s.cached_min { s.cached_min = value }
116 if value > s.cached_max { s.cached_max = value }
117 }
118 return 0
119}
120
121// === queries =====================================================
122
123func nx_sw_sum(s: *SlidingWindow) -> i64 {
124 return s.sum
125}
126
127func nx_sw_mean(s: *SlidingWindow) -> i64 {
128 if s.count == 0 { return 0 }
129 return s.sum / s.count
130}
131
132func nx_sw_count(s: *SlidingWindow) -> i64 {
133 return s.count
134}
135
136func nx_sw_min(s: *SlidingWindow) -> i64 {
137 if s.count == 0 { return 0 }
138 if s.cache_valid == 0 {
139 nx_sw_recompute_extrema(s)
140 }
141 return s.cached_min
142}
143
144func nx_sw_max(s: *SlidingWindow) -> i64 {
145 if s.count == 0 { return 0 }
146 if s.cache_valid == 0 {
147 nx_sw_recompute_extrema(s)
148 }
149 return s.cached_max
150}
151
152func nx_sw_range(s: *SlidingWindow) -> i64 {
153 return nx_sw_max(s) - nx_sw_min(s)
154}
155
156// === typed envelope ==============================================
157//
158// All aggregates are EXACT integer computations. Production tier.
159
160func nx_sw_query_mean(s: *SlidingWindow) -> *ApproxI64 {
161 let m: i64 = nx_sw_mean(s)
162 return nx_approx_new(m, NX_ENV_ABS, 0, 1000000000,
163 NX_MATURITY_PRODUCTION,
164 NX_ADV_HONEST)
165}
166
167// === introspection ================================================
168
169func nx_sw_memory_bytes(s: *SlidingWindow) -> i64 {
170 return 72 + s.window * 8
171}
172
173func nx_sw_clear(s: *SlidingWindow) -> i64 {
174 s.count = 0
175 s.head = 0
176 s.sum = 0
177 s.cached_min = 0
178 s.cached_max = 0
179 s.cache_valid = 0
180 return 0
181}