nx_mesh_lb.nx source
↩ module page · 227 lines · 10322 B
1// nx_mesh_lb.nx -- WORKER MESH: multi-worker LOAD BALANCER + HEALTH-AWARE routing (the Triton/Ray replica-routing
2// analog). A worker CLASS (e.g. gpu-image) has N registered endpoints (laptop 5080, west 3090, ...). On a request the
3// balancer HEALTH-PROBES each endpoint (bounded, non-blocking connect so a down replica is skipped in ~1s), then
4// picks a HEALTHY one by round-robin (persisted counter -> spreads load), skipping every unhealthy replica. If none
5// is healthy it refuses. This is what turns "one worker per class" into a real load-balanced mesh, and it is exactly
6// how a second worker (west 3090) or a cloud replica joins: add an endpoint row, zero code.
7//
8// CLI: (no args) -> self-test GATE (round-robin + health-skip + no-healthy + liar-kills)
9// pick <class> -> live: probe the class's endpoints, print health vector + the chosen replica
10// Classes: gpu-image (laptop 5080 + west 3090). NO fake greens: the gate proves it NEVER picks an unhealthy replica.
11// license_tier: ORIGINAL
12import "nx_syscalls.nx"
13import "nx_estate_path.nx" // ep_anchor: the CWD must not decide this organ's verdict
14import "nx_connect.nx" // bounded connect: a raw sys_connect hangs ~127s on a black-holed host
15import "nx_runtime.nx"
16import "nx_http_client.nx"
17import "nx_swarm_endpoint_lib.nx" // role->addr SSOT (seq1485)
18
19// "<label><host>" -- this renderer wants the host WITHOUT the port, so it takes a narrower slice of
20// the same truth rather than re-spelling the address. Fail-closed: no row -> no fabricated host.
21func ml_where_role(role: *u8, label: *u8) -> *u8 {
22 let h: *u8 = se_host(role)
23 if (h as i64) == 0 { return "unknown (role not in endpoint table)" as *u8 }
24 let o: *u8 = sys_mmap(256)
25 var k: i64 = se_put(o, 0, label)
26 k = se_put(o, k, h)
27 o[k] = 0 as u8
28 return o
29}
30const LB_MAGIC_5080: i64 = 5080
31const LB_MAGIC_3090: i64 = 3090
32const LB_MAGIC_7861: i64 = 7861
33const LB_MAGIC_2048: i64 = 2048
34const LB_MAGIC_2000: i64 = 2000
35const LB_MAGIC_16384: i64 = 16384
36const LB_MAGIC_16383: i64 = 16383
37
38func lb_w(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 }
39func lb_wn(fd: i64, v: i64) -> i64 {
40 let bb: *u8 = sys_mmap(28)
41 var m: i64 = v
42 if m < 0 { sys_write(fd, "-" as *u8, 1); m = 0 - m }
43 let tt: *u8 = sys_mmap(28)
44 var k: i64 = 0
45 if m == 0 { tt[0] = 48 as u8; k = 1 }
46 while m > 0 { tt[k] = (48 + (m - (m/10)*10)) as u8; m = m / 10; k = k + 1 }
47 var i: i64 = 0
48 while i < k { bb[i] = tt[k-1-i]; i = i + 1 }
49 sys_write(fd, bb, k)
50 return 0
51}
52func lb_p(s: *u8) -> i64 { return lb_w(1, s) }
53func lb_pn(v: i64) -> i64 { return lb_wn(1, v) }
54func lb_streq(a: *u8, b: *u8) -> i64 { var i: i64 = 0; while b[i] != (0 as u8) { if a[i] != b[i] { return 0 } i = i + 1 } if a[i] != (0 as u8) { return 0 } return 1 }
55
56// ---- endpoint registry for gpu-image: octets a,b,c,d + a label. (west joins by adding a row.) ----
57const LB_N_IMAGE: i64 = 2
58func lb_ep(idx: i64, oct: *i64) -> i64 {
59 if idx == 0 { oct[0]=192; oct[1]=168; oct[2]=8; oct[3]=193; return 1 } // laptop RTX LB_MAGIC_5080
60 if idx == 1 { oct[0]=10; oct[1]=0; oct[2]=4; oct[3]=13; return 1 } // west dual LB_MAGIC_3090
61 return 0
62}
63func lb_ep_label(idx: i64) -> *u8 {
64 if idx == 0 { return ml_where_role("gpu-image" as *u8, "laptop-5080 @ " as *u8) }
65 if idx == 1 { return ml_where_role("gpu-video" as *u8, "west-3090 @ " as *u8) }
66 return "?" as *u8
67}
68
69// ---- THE PICKER (pure): round-robin among HEALTHY replicas; -1 if none. ----
70func lb_pick(health: *i64, n: i64, counter: i64) -> i64 {
71 var h: i64 = 0
72 var i: i64 = 0
73 while i < n { if health[i] == 1 { h = h + 1 } i = i + 1 }
74 if h == 0 { return 0 - 1 }
75 let target: i64 = counter - (counter / h) * h // counter mod healthy-count
76 var seen: i64 = 0
77 var j: i64 = 0
78 while j < n {
79 if health[j] == 1 {
80 if seen == target { return j }
81 seen = seen + 1
82 }
83 j = j + 1
84 }
85 return 0 - 1
86}
87
88// ---- bounded health probe (non-blocking connect + poll; ~1s) ----
89func lb_pollfd(pfd: *u8, fd: i64, events: i64) -> i64 {
90 pfd[0] = (fd & 0xff) as u8; pfd[1] = ((fd >> 8) & 0xff) as u8; pfd[2] = ((fd >> 16) & 0xff) as u8; pfd[3] = ((fd >> 24) & 0xff) as u8
91 pfd[4] = (events & 0xff) as u8; pfd[5] = ((events >> 8) & 0xff) as u8
92 pfd[6] = 0 as u8; pfd[7] = 0 as u8
93 return 0
94}
95func lb_resp_2xx(buf: *u8, n: i64) -> i64 {
96 var i: i64 = 0
97 while i + 3 < n {
98 if buf[i] == (50 as u8) { if buf[i+1] == (48 as u8) { if buf[i+2] == (48 as u8) { return 1 } } }
99 if buf[i] == (10 as u8) { i = n }
100 i = i + 1
101 }
102 return 0
103}
104func lb_probe(a: i64, b: i64, c: i64, d: i64) -> i64 {
105 let sa: *u8 = sys_mmap(16)
106 nx_http_client_sockaddr_ipv4(sa, a, b, c, d, LB_MAGIC_7861)
107 let fd: i64 = sys_socket(AF_INET, SOCK_STREAM | LB_MAGIC_2048, 0)
108 if fd < 0 { return 0 }
109 nx_connect_bounded(fd, sa, 16, NX_CONN_DEFAULT_MS)
110 let pfd: *u8 = sys_mmap(8)
111 lb_pollfd(pfd, fd, 4)
112 let pn: i64 = sys_poll(pfd, 1, 1000)
113 if pn <= 0 { sys_close(fd); return 0 }
114 let rev: i64 = (pfd[6] as i64) | ((pfd[7] as i64) << 8)
115 if (rev & 8) != 0 { sys_close(fd); return 0 }
116 if (rev & 16) != 0 { sys_close(fd); return 0 }
117 let req: *u8 = sys_mmap(512)
118 let rl: i64 = nx_http_client_build_request("/v1/models" as *u8, 10, "worker" as *u8, 6, req)
119 sys_write(fd, req, rl)
120 lb_pollfd(pfd, fd, 1)
121 let pn2: i64 = sys_poll(pfd, 1, LB_MAGIC_2000)
122 if pn2 <= 0 { sys_close(fd); return 0 }
123 let out: *u8 = sys_mmap(LB_MAGIC_16384)
124 let got: i64 = sys_read(fd, out, LB_MAGIC_16383)
125 sys_close(fd)
126 if got <= 0 { return 0 }
127 return lb_resp_2xx(out, got)
128}
129
130// round-robin counter persistence
131func lb_counter_read() -> i64 {
132 let fd: i64 = sys_openat_rd("knowledge/status/mesh_lb_counter" as *u8)
133 if fd < 0 { return 0 }
134 let b: *u8 = sys_mmap(28)
135 let n: i64 = sys_read(fd, b, 27)
136 sys_close(fd)
137 var v: i64 = 0
138 var i: i64 = 0
139 while i < n { let c: i64 = b[i] as i64; if c >= 48 { if c <= 57 { v = v * 10 + (c - 48) } } i = i + 1 }
140 return v
141}
142func lb_counter_write(v: i64) -> i64 {
143 let fd: i64 = sys_openat_wr("knowledge/status/mesh_lb_counter" as *u8, 0x1a4)
144 if fd < 0 { return 0 }
145 lb_wn(fd, v)
146 sys_close(fd)
147 return 0
148}
149
150func lb_gate() -> i64 {
151 lb_p("=== nx_mesh_lb: multi-worker load balancer + health-aware routing (Triton/Ray replica analog) ===\n" as *u8)
152 let hv: *i64 = sys_mmap(64) as *i64
153 // T1 round-robin among 2 healthy: counter 0->0, 1->1, 2->0
154 hv[0]=1; hv[1]=1
155 var t1: i64 = 0
156 if lb_pick(hv, 2, 0) == 0 { if lb_pick(hv, 2, 1) == 1 { if lb_pick(hv, 2, 2) == 0 { t1 = 1 } } }
157 // T2 health-skip: [healthy, DOWN] -> always 0
158 hv[0]=1; hv[1]=0
159 var t2: i64 = 0
160 if lb_pick(hv, 2, 0) == 0 { if lb_pick(hv, 2, 1) == 0 { if lb_pick(hv, 2, 7) == 0 { t2 = 1 } } }
161 // T3 failover: [DOWN, healthy] -> always 1
162 hv[0]=0; hv[1]=1
163 var t3: i64 = 0
164 if lb_pick(hv, 2, 0) == 1 { if lb_pick(hv, 2, 5) == 1 { t3 = 1 } }
165 // T4 none-healthy -> -1
166 hv[0]=0; hv[1]=0
167 var t4: i64 = 0
168 if lb_pick(hv, 2, 3) == (0 - 1) { t4 = 1 }
169 // T5 3 healthy round-robin: counter 5 -> 5 mod 3 = 2
170 hv[0]=1; hv[1]=1; hv[2]=1
171 var t5: i64 = 0
172 if lb_pick(hv, 3, 5) == 2 { if lb_pick(hv, 3, 3) == 0 { t5 = 1 } }
173 // neg1 liar-kill: the picked index is NEVER unhealthy, across a mixed vector + many counters
174 hv[0]=0; hv[1]=1; hv[2]=0; hv[3]=1
175 var neg1: i64 = 1
176 var k: i64 = 0
177 while k < 20 { let pk: i64 = lb_pick(hv, 4, k); if pk < 0 { neg1 = 0 } else { if hv[pk] != 1 { neg1 = 0 } } k = k + 1 }
178
179 lb_p(" T1 round-robin 2-healthy: " as *u8); lb_pn(t1)
180 lb_p(" | T2 health-skip down: " as *u8); lb_pn(t2)
181 lb_p(" | T3 failover to healthy: " as *u8); lb_pn(t3)
182 lb_p(" | T4 none-healthy->refuse: " as *u8); lb_pn(t4)
183 lb_p(" | T5 3-replica round-robin: " as *u8); lb_pn(t5)
184 lb_p(" | neg1 never-picks-unhealthy: " as *u8); lb_pn(neg1); lb_p("\n" as *u8)
185
186 var pass: i64 = 0
187 if t1==1 { if t2==1 { if t3==1 { if t4==1 { if t5==1 { if neg1==1 { pass = 1 } } } } } }
188 if pass == 1 { lb_p("MESHLBGATE verdict=GREEN (round-robin among healthy; unhealthy skipped; none-healthy refused; liar-killed)\n" as *u8); return 0 }
189 lb_p("MESHLBGATE verdict=RED\n" as *u8)
190 return 1
191}
192
193func main(argc: i64, argv: *i64) -> i64 {
194 // ANCHOR FIRST (2026-08-04, nx_cwdguard finding): this organ reads a RELATIVE
195 // knowledge/ path, so its answer depended on where it was launched. No-op when
196 // already at the estate root, so the cron/MCP context is unchanged.
197 ep_anchor()
198 if argc >= 2 {
199 let cmd: *u8 = argv[1] as *u8
200 if lb_streq(cmd, "pick" as *u8) == 1 {
201 if argc < 3 { lb_w(2, "usage: nx_mesh_lb pick <class>\n" as *u8); return 2 }
202 let cls: *u8 = argv[2] as *u8
203 if lb_streq(cls, "gpu-image" as *u8) == 0 { lb_w(2, "only gpu-image registered\n" as *u8); return 3 }
204 lb_p("pick class=gpu-image -- health-probing " as *u8); lb_pn(LB_N_IMAGE); lb_p(" replicas...\n" as *u8)
205 let hv: *i64 = sys_mmap(64) as *i64
206 let oct: *i64 = sys_mmap(64) as *i64
207 var i: i64 = 0
208 while i < LB_N_IMAGE {
209 lb_ep(i, oct)
210 let h: i64 = lb_probe(oct[0], oct[1], oct[2], oct[3])
211 hv[i] = h
212 lb_p(" [" as *u8); lb_pn(i); lb_p("] " as *u8); lb_p(lb_ep_label(i)); lb_p(" -> " as *u8)
213 if h == 1 { lb_p("HEALTHY\n" as *u8) } else { lb_p("down\n" as *u8) }
214 i = i + 1
215 }
216 let ctr: i64 = lb_counter_read()
217 let pick: i64 = lb_pick(hv, LB_N_IMAGE, ctr)
218 lb_counter_write(ctr + 1)
219 if pick < 0 { lb_p(" NO HEALTHY REPLICA -- refuse (would 503)\n" as *u8); return 5 }
220 lb_p(" ROUTE -> [" as *u8); lb_pn(pick); lb_p("] " as *u8); lb_p(lb_ep_label(pick)); lb_p(" (round-robin ctr=" as *u8); lb_pn(ctr); lb_p(", health-aware)\n" as *u8)
221 return 0
222 }
223 lb_w(2, "usage: nx_mesh_lb pick <class> (no args = gate)\n" as *u8)
224 return 2
225 }
226 return lb_gate()
227}