nx_linkqual.nx source
↩ module page · 305 lines · 12417 B
1// nx_linkqual.nx -- ONE sovereign real-time LINK-QUALITY substrate that
2// improves GAME MULTIPLAYER, VIDEO ROOMS, and STREAMING from a single
3// per-flow metrics core. Integer-only (no float), pure/deterministic
4// (fed timestamped sample events -> KAT-able offline), raw-syscall-only
5// (imports only nx_syscalls), zero external/licensed deps.
6//
7// Research-grounded (workflow wf_5c50a76d) in the canonical RFCs, all
8// re-implemented from the SPEC (concepts, not code):
9// - RTT smoothing: RFC 6298 / Jacobson-Karels (SRTT*8, RTTVAR*4, RTO).
10// - Interarrival jitter: RFC 3550 A.8 (J += (|D|-J)/16, pre-scaled *16).
11// - Loss fraction: RFC 3550 A.3 (expected-vs-received, Q16.16 fraction).
12// - Throughput EWMA (BBR-style delivery-rate, gain 1/8).
13// - Latency quantiles: HDR-style power-of-two magnitude buckets + a
14// first-class max (Tene: never discard the tail -- that IS the signal).
15// SUPERSEDES nx_bandwidth_meter for real-time use (adds jitter, quantiles,
16// proper RTTVAR/RTO it lacks); pairs with nx_adaptive_bitrate for tiers.
17//
18// Consumers (the products): lq_game_advise (rollback input-delay frames),
19// lq_jitter_buffer_ms (video-room NetEQ-style target), lq_bitrate_decision
20// (streaming ABR up/hold/down), lq_score + lq_verdict_band (easy-for-
21// anyone Excellent/Good/Fair/Poor + one number).
22//
23// Constants are bootstrap defaults destined for svc-config (CLAUDE.md 11).
24// license_tier: ORIGINAL
25
26import "nx_syscalls.nx"
27const LQ_MAGIC_32768: i64 = 32768
28const LQ_MAGIC_65536: i64 = 65536
29
30const LQ_NBUCKETS: i64 = 32
31const LQ_SRTT_SH: i64 = 3 // SRTT stored *8
32const LQ_RTTVAR_SH: i64 = 2 // RTTVAR stored *4
33const LQ_JIT_SH: i64 = 4 // jitter stored *16 (RFC 3550 A.8)
34const LQ_RATE_SH: i64 = 3 // throughput EWMA gain 1/8
35const LQ_Q16: i64 = 16 // Q16.16 fixed-point shift (loss fraction)
36const LQ_RTO_MIN_MS: i64 = 200 // (RFC6298 says 1000; 200 fits real-time UDP)
37const LQ_RTO_MAX_MS: i64 = 60000
38
39// LinkFlow: 16 i64 header (128B) + buckets[32] i64 (256B) = 384B, flat mmap.
40struct LinkFlow {
41 srtt_x8: i64,
42 rttvar_x4: i64,
43 rto_ms: i64,
44 transit_prev: i64,
45 have_transit: i64,
46 jitter_x16: i64,
47 base_seq: i64,
48 max_seq: i64,
49 cycles: i64,
50 received: i64,
51 expected_prior: i64,
52 received_prior: i64,
53 rate_ewma_bps: i64,
54 sample_count: i64,
55 max_latency_ms: i64,
56 loss_frac_q16: i64,
57}
58const LQ_FLOW_HDR_BYTES: i64 = 128
59const LQ_FLOW_BYTES: i64 = 384
60
61// quality bands
62const LQ_EXCELLENT: i64 = 0
63const LQ_GOOD: i64 = 1
64const LQ_FAIR: i64 = 2
65const LQ_POOR: i64 = 3
66
67// products (for lq_render)
68const LQ_PROD_GAME: i64 = 0
69const LQ_PROD_VIDEO: i64 = 1
70const LQ_PROD_STREAM: i64 = 2
71
72struct GameAdvice { delay_frames: i64, score: i64, verdict: i64 }
73const LQ_GAME_ADVICE_BYTES: i64 = 24
74
75func lq_flow_new() -> *LinkFlow {
76 let raw: *u8 = sys_mmap(LQ_FLOW_BYTES) // mmap is zeroed
77 return raw as *LinkFlow
78}
79
80// ===== metrics core =====
81
82// RTT -- RFC 6298, RTTVAR updated FIRST (uses old SRTT). r_ms = clean RTT.
83func lq_on_rtt(f: *LinkFlow, r_ms: i64) -> i64 {
84 if r_ms < 0 { return -1 }
85 if f.srtt_x8 == 0 {
86 f.srtt_x8 = r_ms << LQ_SRTT_SH
87 f.rttvar_x4 = (r_ms >> 1) << LQ_RTTVAR_SH
88 } else {
89 var err: i64 = r_ms - (f.srtt_x8 >> LQ_SRTT_SH)
90 f.srtt_x8 = f.srtt_x8 + err
91 if err < 0 { err = 0 - err }
92 err = err - (f.rttvar_x4 >> LQ_RTTVAR_SH)
93 f.rttvar_x4 = f.rttvar_x4 + err
94 }
95 let srtt: i64 = f.srtt_x8 >> LQ_SRTT_SH
96 let rttvar: i64 = f.rttvar_x4 >> LQ_RTTVAR_SH
97 var rto: i64 = srtt + (rttvar << 2)
98 if rto < LQ_RTO_MIN_MS { rto = LQ_RTO_MIN_MS }
99 if rto > LQ_RTO_MAX_MS { rto = LQ_RTO_MAX_MS }
100 f.rto_ms = rto
101 return 0
102}
103func lq_srtt_ms(f: *LinkFlow) -> i64 { return f.srtt_x8 >> LQ_SRTT_SH }
104func lq_rttvar_ms(f: *LinkFlow) -> i64 { return f.rttvar_x4 >> LQ_RTTVAR_SH }
105
106// Jitter -- RFC 3550 A.8. send_ts/recv_ts same units (ms); skew cancels in D.
107func lq_on_arrival(f: *LinkFlow, send_ts: i64, recv_ts: i64) -> i64 {
108 let transit: i64 = recv_ts - send_ts
109 if f.have_transit == 0 { f.transit_prev = transit; f.have_transit = 1; return 0 }
110 var d: i64 = transit - f.transit_prev
111 f.transit_prev = transit
112 // Branchless abs: arithmetic >>63 -> 0 (d>=0) or all-ones (d<0), then
113 // (d^s)-s. Over `if d<0 {d=0-d}` on realistic random-sign jitter, a
114 // rigorous same-process interleaved min-of-25 A/B (nx_abs_ab.nx, __rdtsc)
115 // shows it CONSISTENTLY but only SLIGHTLY faster: ~0.1-0.7 cyc/op (~2-3%)
116 // -- removes the mispredict + the bb-boundary spill. Honest magnitude:
117 // small. The big lever for this hot path is G1 register allocation (the
118 // ~40-insn body is stack-spill-bound, not branch-bound). KAT-identical.
119 let s: i64 = d >> 63
120 d = (d ^ s) - s
121 f.jitter_x16 = f.jitter_x16 + (d - (f.jitter_x16 >> LQ_JIT_SH))
122 return 0
123}
124func lq_jitter_ms(f: *LinkFlow) -> i64 { return f.jitter_x16 >> LQ_JIT_SH }
125
126// Loss -- RFC 3550 A.3. on_seq per packet; loss_tick per report interval.
127func lq_on_seq(f: *LinkFlow, seq: i64) -> i64 {
128 if f.received == 0 { f.base_seq = seq; f.max_seq = seq }
129 if seq < f.max_seq { if (f.max_seq - seq) > LQ_MAGIC_32768 { f.cycles = f.cycles + LQ_MAGIC_65536 } }
130 if seq > f.max_seq { f.max_seq = seq }
131 f.received = f.received + 1
132 return 0
133}
134func lq_loss_tick(f: *LinkFlow) -> i64 {
135 let extended_max: i64 = f.cycles + f.max_seq
136 let expected: i64 = extended_max - f.base_seq + 1
137 let exp_int: i64 = expected - f.expected_prior
138 let rcv_int: i64 = f.received - f.received_prior
139 let lost_int: i64 = exp_int - rcv_int
140 f.expected_prior = expected
141 f.received_prior = f.received
142 if exp_int <= 0 { f.loss_frac_q16 = 0; return 0 }
143 if lost_int <= 0 { f.loss_frac_q16 = 0; return 0 }
144 f.loss_frac_q16 = (lost_int << LQ_Q16) / exp_int
145 return f.loss_frac_q16
146}
147// rounded (not truncated) -- avoids the double-truncation that makes 40% read as 39%.
148func lq_loss_pct(f: *LinkFlow) -> i64 { return (f.loss_frac_q16 * 100 + LQ_MAGIC_32768) >> LQ_Q16 }
149
150// Throughput -- BBR-style delivery-rate EWMA, bits/sec.
151func lq_on_delivery(f: *LinkFlow, bytes: i64, interval_ms: i64) -> i64 {
152 if interval_ms <= 0 { return f.rate_ewma_bps }
153 let sample_bps: i64 = (bytes * 8 * 1000) / interval_ms
154 if f.rate_ewma_bps == 0 { f.rate_ewma_bps = sample_bps }
155 else { f.rate_ewma_bps = f.rate_ewma_bps + ((sample_bps - f.rate_ewma_bps) >> LQ_RATE_SH) }
156 return f.rate_ewma_bps
157}
158
159// Quantiles -- HDR power-of-two magnitude buckets; max kept first-class.
160func lq_buckets(f: *LinkFlow) -> *i64 { return ((f as i64) + LQ_FLOW_HDR_BYTES) as *i64 }
161// bucket = floor(log2(lat_ms)). O(1) via the hardware CLZ (lzcnt) instead
162// of the bit-length loop -- floor(log2(x)) == 31 - clz32(x) for x in 32-bit.
163// Cut lq_record_latency from ~23ns to single-instruction-class on this HW.
164func lq_bucket_of(lat_ms: i64) -> i64 {
165 if lat_ms <= 1 { return 0 }
166 var x: i64 = lat_ms
167 if x > 0x7fffffff { x = 0x7fffffff } // clamp to 32-bit domain for clz32
168 var b: i64 = 31 - __clz32(x)
169 if b < 0 { b = 0 }
170 if b >= LQ_NBUCKETS { b = LQ_NBUCKETS - 1 }
171 return b
172}
173func lq_record_latency(f: *LinkFlow, lat_ms: i64) -> i64 {
174 let bk: *i64 = lq_buckets(f)
175 let i: i64 = lq_bucket_of(lat_ms)
176 bk[i] = bk[i] + 1
177 f.sample_count = f.sample_count + 1
178 if lat_ms > f.max_latency_ms { f.max_latency_ms = lat_ms }
179 return 0
180}
181// p in [0,100]; returns bucket UPPER edge (2^(i+1) ms).
182func lq_quantile_ms(f: *LinkFlow, p: i64) -> i64 {
183 if f.sample_count == 0 { return 0 }
184 let target: i64 = (p * f.sample_count + 99) / 100
185 let bk: *i64 = lq_buckets(f)
186 var acc: i64 = 0
187 var i: i64 = 0
188 while i < LQ_NBUCKETS {
189 acc = acc + bk[i]
190 if acc >= target { let edge: i64 = 1 << (i + 1); return edge }
191 i = i + 1
192 }
193 return f.max_latency_ms
194}
195
196// ===== easy-for-anyone score + verdict =====
197// 0..100; penalize RTT, jitter, loss against real-time targets.
198func lq_score(f: *LinkFlow) -> i64 {
199 let srtt: i64 = lq_srtt_ms(f)
200 let jit: i64 = lq_jitter_ms(f)
201 let lossp: i64 = lq_loss_pct(f)
202 var s: i64 = 100
203 var pr: i64 = srtt / 5
204 if pr > 40 { pr = 40 }
205 s = s - pr // 200ms RTT -> -40
206 var pj: i64 = jit * 2
207 if pj > 30 { pj = 30 }
208 s = s - pj // 15ms jitter -> -30
209 var pl: i64 = lossp * 8
210 if pl > 40 { pl = 40 }
211 s = s - pl // 5% loss -> -40
212 if s < 0 { s = 0 }
213 return s
214}
215func lq_verdict_band(score: i64) -> i64 {
216 if score >= 80 { return LQ_EXCELLENT }
217 if score >= 60 { return LQ_GOOD }
218 if score >= 40 { return LQ_FAIR }
219 return LQ_POOR
220}
221
222// ===== consumer 1: GAME -- rollback input-delay frames =====
223// budget_ms ~ one-way (SRTT/2) + jitter safety margin (2*jitter);
224// frames = ceil(budget_ms * fps / 1000), clamped to a sane rollback window.
225func lq_game_advise(f: *LinkFlow, fps: i64, out: *GameAdvice) -> i64 {
226 let srtt: i64 = lq_srtt_ms(f)
227 let jit: i64 = lq_jitter_ms(f)
228 let budget_ms: i64 = (srtt / 2) + (jit * 2)
229 var frames: i64 = (budget_ms * fps + 999) / 1000
230 if frames < 0 { frames = 0 }
231 if frames > 9 { frames = 9 }
232 out.delay_frames = frames
233 out.score = lq_score(f)
234 out.verdict = lq_verdict_band(out.score)
235 return 0
236}
237
238// ===== consumer 2: VIDEO ROOM -- NetEQ-style target jitter-buffer ms =====
239// target ~ base + 4*jitter (covers the bulk of arrival spread), +room for
240// loss/NACK; clamped [20,500].
241func lq_jitter_buffer_ms(f: *LinkFlow) -> i64 {
242 let jit: i64 = lq_jitter_ms(f)
243 let lossp: i64 = lq_loss_pct(f)
244 var t: i64 = 20 + jit * 4
245 t = t + lossp * 5
246 if t < 20 { t = 20 }
247 if t > 500 { t = 500 }
248 return t
249}
250
251// ===== consumer 3: STREAMING -- ABR up/hold/down (-1/0/+1) =====
252// down on congestion (loss high OR p95 latency spike vs SRTT); up only when
253// clean (no loss, p95 near SRTT); else hold (hysteresis -> no oscillation).
254func lq_bitrate_decision(f: *LinkFlow) -> i64 {
255 let lossp: i64 = lq_loss_pct(f)
256 let p95: i64 = lq_quantile_ms(f, 95)
257 let srtt: i64 = lq_srtt_ms(f)
258 if lossp >= 5 { return 0 - 1 }
259 if p95 > (srtt * 2 + 100) { return 0 - 1 }
260 if lossp == 0 { if p95 <= (srtt + 50) { return 1 } }
261 return 0
262}
263
264// ===== easy-for-anyone render =====
265func lq_band_word(b: i64) -> i64 {
266 if b == LQ_EXCELLENT { sys_write(1, "EXCELLENT" as *u8, 9); return 0 }
267 if b == LQ_GOOD { sys_write(1, "GOOD" as *u8, 4); return 0 }
268 if b == LQ_FAIR { sys_write(1, "FAIR" as *u8, 4); return 0 }
269 sys_write(1, "POOR" as *u8, 4)
270 return 0
271}
272func lq_dec(n: i64) -> i64 {
273 if n == 0 { sys_write(1, "0" as *u8, 1); return 0 }
274 var m: i64 = n
275 if m < 0 { sys_write(1, "-" as *u8, 1); m = 0 - m }
276 let d: *u8 = sys_mmap(24)
277 var k: i64 = 0
278 while m > 0 { d[k] = (0x30 + (m % 10)) as u8; m = m / 10; k = k + 1 }
279 var i: i64 = k - 1
280 while i >= 0 { let one: *u8 = sys_mmap(1); one[0] = d[i]; sys_write(1, one, 1); i = i - 1 }
281 return 0
282}
283func lq_render(f: *LinkFlow, product: i64) -> i64 {
284 let sc: i64 = lq_score(f)
285 sys_write(1, " connection quality: ", 22); lq_band_word(lq_verdict_band(sc))
286 sys_write(1, " (", 3); lq_dec(sc); sys_write(1, "/100) -- ", 9)
287 sys_write(1, "RTT ", 4); lq_dec(lq_srtt_ms(f)); sys_write(1, "ms, jitter ", 11)
288 lq_dec(lq_jitter_ms(f)); sys_write(1, "ms, loss ", 9); lq_dec(lq_loss_pct(f)); sys_write(1, "%\n", 2)
289 if product == LQ_PROD_GAME {
290 let ga: *GameAdvice = sys_mmap(LQ_GAME_ADVICE_BYTES) as *GameAdvice
291 lq_game_advise(f, 60, ga)
292 sys_write(1, " game: use ", 12); lq_dec(ga.delay_frames); sys_write(1, " input-delay frame(s) @60fps\n", 29)
293 }
294 if product == LQ_PROD_VIDEO {
295 sys_write(1, " video room: set jitter buffer to ", 35); lq_dec(lq_jitter_buffer_ms(f)); sys_write(1, "ms\n", 3)
296 }
297 if product == LQ_PROD_STREAM {
298 let dcn: i64 = lq_bitrate_decision(f)
299 sys_write(1, " streaming: ", 13)
300 if dcn > 0 { sys_write(1, "raise bitrate (headroom)\n", 25) }
301 if dcn == 0 { sys_write(1, "hold bitrate (stable)\n", 22) }
302 if dcn < 0 { sys_write(1, "lower bitrate (congestion)\n", 27) }
303 }
304 return 0
305}