nx_fabric_fair_gate.nx source
↩ module page · 225 lines · 10407 B
1// nx_fabric_fair_gate.nx -- R2.5 of the SkyHammer fabric exceed ladder: MULTI-FLOW
2// FAIRNESS, closing the disclosed single-flow caveat on R2 (nx_fabric_cc) / the L4S
3// controller (nx_room_l4s). N flows with UNEQUAL starting windows share ONE bottleneck;
4// the DCTCP/ECN controller must drive them to FAIR shares (Jain's index ~ 1) AND keep the
5// queue bounded near the ECN setpoint. Load-bearing neg-control: with the reaction REMOVED
6// (frozen windows), the initial imbalance PERSISTS -> unfair -> proves the controller (not
7// the FIFO drain) is what creates fairness. Integer-only fixed-point (no float). Same
8// discrete-event link shape as nx_fabric_cc, extended to N independent flows.
9// expect_exit: 0 license_tier: ORIGINAL
10import "nx_syscalls.nx"
11
12const FAIR_DCTCP: i64 = 0 // ECN-adaptive (AI + proportional cut on marks)
13const FAIR_OPEN: i64 = 1 // reaction REMOVED: windows frozen at init -> imbalance persists
14const FSCALE: i64 = 1024
15
16// run NF flows over a shared bottleneck. flow 0 starts at init_big, flows 1.. at init_small.
17// fills dsw[f] = steady-state delivered per flow. returns the shared-queue q_p99.
18func fair_run(mode: i64, NF: i64, D: i64, RTT: i64, BUFCAP: i64, K: i64,
19 init_big: i64, init_small: i64, dsw: *i64) -> i64 {
20 let MAXT: i64 = 4000
21 let RS: i64 = BUFCAP + 1
22 let qflow: *i64 = sys_mmap(RS * 8) as *i64 // flow id of each queued frame
23 let qmark: *i64 = sys_mmap(RS * 8) as *i64 // ECN mark bit of each queued frame
24 var qhead: i64 = 0
25 var qtail: i64 = 0
26 var qocc: i64 = 0
27 let ackc: *i64 = sys_mmap(MAXT * NF * 8) as *i64 // acks for flow f arriving at tick t
28 let ackm: *i64 = sys_mmap(MAXT * NF * 8) as *i64 // ...and how many were marked
29 let qhist: *i64 = sys_mmap((BUFCAP + 2) * 8) as *i64
30 let cwnd: *i64 = sys_mmap(NF * 8) as *i64
31 let inflight: *i64 = sys_mmap(NF * 8) as *i64
32 let alpha: *i64 = sys_mmap(NF * 8) as *i64
33 let acct: *i64 = sys_mmap(NF * 8) as *i64
34 let accm: *i64 = sys_mmap(NF * 8) as *i64
35
36 let WARMUP: i64 = RTT * 16
37 var f: i64 = 0
38 while f < NF {
39 if f == 0 { cwnd[f] = init_big } else { cwnd[f] = init_small }
40 inflight[f] = 0; alpha[f] = 0; acct[f] = 0; accm[f] = 0; dsw[f] = 0
41 f = f + 1
42 }
43 var rtt_timer: i64 = 0
44 var tick: i64 = 0
45 var samples: i64 = 0
46 var rr: i64 = 0
47
48 while tick < MAXT {
49 // 1. ACKs for this tick clear each flow's in-flight + accumulate its mark stats
50 f = 0
51 while f < NF {
52 let a: i64 = ackc[tick * NF + f]
53 inflight[f] = inflight[f] - a
54 acct[f] = acct[f] + a
55 accm[f] = accm[f] + ackm[tick * NF + f]
56 f = f + 1
57 }
58 // 2. DRAIN up to D from the shared FIFO; schedule each frame's ack RTT later
59 var dd: i64 = 0
60 while dd < D {
61 if qocc == 0 { dd = D } else {
62 let ff: i64 = qflow[qhead]
63 let mk: i64 = qmark[qhead]
64 qhead = qhead + 1; if qhead >= RS { qhead = 0 }
65 qocc = qocc - 1
66 if tick >= WARMUP { dsw[ff] = dsw[ff] + 1 }
67 let at: i64 = tick + RTT
68 if at < MAXT { ackc[at * NF + ff] = ackc[at * NF + ff] + 1; ackm[at * NF + ff] = ackm[at * NF + ff] + mk }
69 dd = dd + 1
70 }
71 }
72 // 3. shared-queue depth histogram (steady state)
73 if tick >= WARMUP {
74 samples = samples + 1
75 if qocc <= BUFCAP { qhist[qocc] = qhist[qocc] + 1 } else { qhist[BUFCAP] = qhist[BUFCAP] + 1 }
76 }
77 // 4. PER-RTT controller update, per flow (DCTCP). FAIR_OPEN never adapts.
78 rtt_timer = rtt_timer + 1
79 if rtt_timer >= RTT {
80 f = 0
81 while f < NF {
82 if mode == FAIR_DCTCP {
83 if acct[f] > 0 {
84 let frac: i64 = (accm[f] * FSCALE) / acct[f]
85 alpha[f] = alpha[f] + (frac - alpha[f]) / 16
86 }
87 cwnd[f] = cwnd[f] + 1
88 if accm[f] > 0 { cwnd[f] = cwnd[f] - (cwnd[f] * alpha[f]) / (2 * FSCALE) }
89 if cwnd[f] < 1 { cwnd[f] = 1 }
90 }
91 acct[f] = 0; accm[f] = 0
92 f = f + 1
93 }
94 rtt_timer = 0
95 }
96 // 5. SEND: round-robin fair injection opportunity (rr persists across ticks so no
97 // flow is structurally favored); each flow injects only while in_flight < its cwnd.
98 var s: i64 = 0
99 while s < 2 * D {
100 var placed: i64 = 0
101 var tries: i64 = 0
102 while tries < NF {
103 let g: i64 = rr % NF
104 rr = rr + 1
105 if inflight[g] < cwnd[g] {
106 if qocc < BUFCAP {
107 var mk2: i64 = 0
108 if qocc >= K { mk2 = 1 }
109 qflow[qtail] = g; qmark[qtail] = mk2
110 qtail = qtail + 1; if qtail >= RS { qtail = 0 }
111 qocc = qocc + 1; inflight[g] = inflight[g] + 1
112 placed = 1; tries = NF
113 } else { tries = NF }
114 } else { tries = tries + 1 }
115 }
116 if placed == 0 { s = 2 * D }
117 s = s + 1
118 }
119 tick = tick + 1
120 }
121
122 // shared-queue p99
123 let t99: i64 = (samples * 99) / 100
124 var cum: i64 = 0
125 var p99: i64 = 0
126 var g99: i64 = 0
127 var hh: i64 = 0
128 while hh <= BUFCAP {
129 cum = cum + qhist[hh]
130 if g99 == 0 { if cum > t99 { p99 = hh; g99 = 1 } }
131 if g99 == 1 { hh = BUFCAP }
132 hh = hh + 1
133 }
134 return p99
135}
136
137// Jain's fairness index x1000: (sum^2 * 1000) / (NF * sumsq). 1000 = perfectly fair.
138func jain_x1000(d: *i64, NF: i64) -> i64 {
139 var sum: i64 = 0
140 var sumsq: i64 = 0
141 var f: i64 = 0
142 while f < NF { sum = sum + d[f]; sumsq = sumsq + d[f] * d[f]; f = f + 1 }
143 if sumsq == 0 { return 0 }
144 return (sum * sum * 1000) / (NF * sumsq)
145}
146
147func gp(s: *u8) -> i64 { var n: i64=0; while s[n]!=(0 as u8){n=n+1} sys_write(1,s,n); return 0 }
148func fw(fd: i64, s: *u8) -> i64 { var n: i64=0; while s[n]!=(0 as u8){n=n+1} sys_write(fd,s,n); return 0 }
149func gn(v: i64) -> i64 {
150 let b: *u8 = sys_mmap(28); var m: i64 = v
151 if m < 0 { m = 0 - m; sys_write(1, "-" as *u8, 1) }
152 let t: *u8 = sys_mmap(28); var k: i64 = 0
153 if m == 0 { t[0] = 48 as u8; k = 1 }
154 while m > 0 { t[k] = (48 + (m % 10)) as u8; m = m / 10; k = k + 1 }
155 var i: i64 = 0
156 while i < k { b[i] = t[k - 1 - i]; i = i + 1 }
157 sys_write(1, b, k); return 0
158}
159func chk(name: *u8, ok: i64) -> i64 {
160 if ok == 1 { gp(" PASS " as *u8) } else { gp(" FAIL " as *u8) }
161 gp(name); gp("\n" as *u8); return ok
162}
163func show_flows(tag: *u8, d: *i64, NF: i64) -> i64 {
164 gp(tag); var f: i64 = 0
165 while f < NF { gp(" f" as *u8); gn(f); gp("=" as *u8); gn(d[f]); f = f + 1 }
166 gp("\n" as *u8); return 0
167}
168
169func main() -> i64 {
170 let NF: i64 = 4
171 let D: i64 = 4
172 let RTT: i64 = 8 // BDP = 32, fair share = 8/flow
173 let BUF: i64 = 400
174 let K: i64 = 16
175 let BIG: i64 = 60 // flow 0 starts hogging
176 let SMALL: i64 = 2 // flows 1-3 start window-starved (< fair share 8)
177
178 gp("nx_fabric_fair R2.5 -- multi-flow fairness (Jain index), DCTCP vs reaction-removed\n" as *u8)
179 gp("flows=" as *u8); gn(NF); gp(" drain=" as *u8); gn(D); gp("/tick RTT=" as *u8); gn(RTT)
180 gp(" (BDP=" as *u8); gn(D*RTT); gp(" fair-share=" as *u8); gn((D*RTT)/NF); gp(") ECN-K=" as *u8); gn(K)
181 gp(" init: f0=" as *u8); gn(BIG); gp(" others=" as *u8); gn(SMALL); gp("\n" as *u8)
182
183 let dctcp_d: *i64 = sys_mmap(NF * 8) as *i64
184 let open_d: *i64 = sys_mmap(NF * 8) as *i64
185 let q99_dctcp: i64 = fair_run(FAIR_DCTCP, NF, D, RTT, BUF, K, BIG, SMALL, dctcp_d)
186 let q99_open: i64 = fair_run(FAIR_OPEN, NF, D, RTT, BUF, K, BIG, SMALL, open_d)
187 let j_dctcp: i64 = jain_x1000(dctcp_d, NF)
188 let j_open: i64 = jain_x1000(open_d, NF)
189
190 show_flows("DCTCP delivered:" as *u8, dctcp_d, NF)
191 show_flows("OPEN delivered:" as *u8, open_d, NF)
192 gp("Jain(x1000) DCTCP=" as *u8); gn(j_dctcp); gp(" OPEN=" as *u8); gn(j_open)
193 gp(" q_p99 DCTCP=" as *u8); gn(q99_dctcp); gp(" OPEN=" as *u8); gn(q99_open); gp("\n" as *u8)
194
195 // no-starvation: every DCTCP flow delivered something in steady state
196 var nostarve: i64 = 1
197 var f: i64 = 0
198 while f < NF { if dctcp_d[f] <= 0 { nostarve = 0 } f = f + 1 }
199
200 var pass: i64 = 0; var tot: i64 = 0
201 var c1: i64 = 0; if j_dctcp >= 950 { c1 = 1 }
202 pass = pass + chk("1 DCTCP Jain index >= 0.95 (flows converge to FAIR shares)" as *u8, c1); tot = tot + 1
203 var c2: i64 = 0; if nostarve == 1 { c2 = 1 }
204 pass = pass + chk("2 no flow starved under DCTCP (all delivered > 0)" as *u8, c2); tot = tot + 1
205 var c3: i64 = 0; if q99_dctcp <= 3 * K { c3 = 1 }
206 pass = pass + chk("3 DCTCP shared-queue q_p99 <= 3*K (bounded WITH N flows)" as *u8, c3); tot = tot + 1
207 var c4: i64 = 0; if j_dctcp >= j_open + 200 { c4 = 1 }
208 pass = pass + chk("NC reaction-removed is MUCH less fair (controller, not FIFO, makes fairness)" as *u8, c4); tot = tot + 1
209 var c5: i64 = 0; if j_open <= 750 { c5 = 1 }
210 pass = pass + chk("NC frozen-window imbalance PERSISTS (unfair, J<=0.75 = neg-control fires)" as *u8, c5); tot = tot + 1
211
212 gp("---- nx_fabric_fair R2.5 gate: passed " as *u8); gn(pass); gp(" / " as *u8); gn(tot); gp("\n" as *u8)
213 if pass == tot {
214 let lfd: i64 = sys_openat_append("knowledge/status/fabric_nx_fabric_fair.log" as *u8, 0x1a4)
215 if lfd >= 0 {
216 fw(lfd, "R2.5-FABRIC-FAIR organ=nx_fabric_fair checks=5/5 jain_dctcp=" as *u8);
217 let nb: *u8 = sys_mmap(28); var m: i64=j_dctcp; var k: i64=0; if m==0{nb[0]=48 as u8;k=1} let tt: *u8=sys_mmap(28); while m>0{tt[k]=(48+(m%10)) as u8;m=m/10;k=k+1} var i: i64=0; while i<k{nb[i]=tt[k-1-i];i=i+1} sys_write(lfd,nb,k)
218 fw(lfd, " multi-flow-fairness+bounded-queue+NC verdict=GREEN\n" as *u8)
219 sys_close(lfd)
220 }
221 gp("R2.5 GREEN -- N flows converge fair + queue bounded; reaction is load-bearing\n" as *u8)
222 sys_exit(0); return 0
223 }
224 sys_exit(1); return 1
225}