code wiki / _hdl_build / nx_gen_orchestrator_gate.nx
nx_gen_orchestrator_gate.nx source
↩ module page · 218 lines · 12285 B
1// nx_gen_orchestrator_gate.nx -- end-to-end gate for the gen daemon over REAL loopback sockets.
2// Parent creates two listening sockets (worker + orchestrator) BEFORE forking (backlog removes any
3// race), forks a MOCK GPU worker (returns {"images":[b64,b64]} for /sdapi/v1/txt2img) and an
4// orchestrator server (go_serve_conn), then acts as the HTTP client:
5// (1) POST /api/generate -> 200, "count":2, real CIDs returned (socket dispatch + pipeline)
6// (2) GET /img/<cid> -> 200 image/png with a real PNG body (sidecar lookup + blob serve)
7// (3) GET / -> 200 with the gen UI
8// license_tier: ORIGINAL
9import "nx_gen_orchestrator.nx"
10import "nx_connect.nx" // bounded connect: a raw sys_connect hangs ~127s on a black-holed host
11import "nx_syscalls.nx"
12
13func og(s: *u8) -> i64 { var n: i64=0; while s[n]!=(0 as u8){n=n+1} sys_write(1,s,n); return 0 }
14func og_slen(s: *u8) -> i64 { var n: i64=0; while s[n]!=(0 as u8){n=n+1} return n }
15func og_cat(dst: *u8, off: i64, s: *u8) -> i64 { var o: i64=off; var i: i64=0; while s[i]!=(0 as u8){dst[o]=s[i]; o=o+1; i=i+1} return o }
16func og_catb(dst: *u8, off: i64, src: *u8, n: i64) -> i64 { var o: i64=off; var i: i64=0; while i<n {dst[o]=src[i]; o=o+1; i=i+1} return o }
17func og_itoa(dst: *u8, off: i64, v: i64) -> i64 { let t: *u8=sys_mmap(28); var m: i64=v; var k: i64=0; if m==0{t[0]=48 as u8;k=1} while m>0{t[k]=(48+(m%10)) as u8;m=m/10;k=k+1} var o: i64=off; var q: i64=k-1; while q>=0{dst[o]=t[q];o=o+1;q=q-1} return o }
18func og_find(buf: *u8, n: i64, needle: *u8) -> i64 {
19 let nl: i64 = og_slen(needle)
20 if nl==0 {return 0}
21 var i: i64=0
22 while i+nl<=n { var j: i64=0; var ok: i64=1; while j<nl { if (buf[i+j]&0xff)!=(needle[j]&0xff){ok=0;j=nl} else {j=j+1} } if ok==1{return i} i=i+1 }
23 return 0-1
24}
25
26func og_basepng(b: *u8) -> i64 {
27 b[0]=137 as u8; b[1]=80 as u8; b[2]=78 as u8; b[3]=71 as u8
28 b[4]=13 as u8; b[5]=10 as u8; b[6]=26 as u8; b[7]=10 as u8
29 b[8]=0 as u8; b[9]=0 as u8; b[10]=0 as u8; b[11]=14 as u8
30 b[12]=73 as u8; b[13]=72 as u8; b[14]=68 as u8; b[15]=82 as u8
31 b[16]=0 as u8; b[17]=0 as u8; b[18]=0 as u8; b[19]=0 as u8; b[20]=1 as u8
32 b[21]=0 as u8; b[22]=0 as u8; b[23]=0 as u8; b[24]=1 as u8; b[25]=8 as u8
33 b[26]=6 as u8; b[27]=0 as u8; b[28]=0 as u8; b[29]=0 as u8
34 b[30]=115 as u8; b[31]=22 as u8; b[32]=206 as u8; b[33]=242 as u8
35 b[34]=0 as u8; b[35]=0 as u8; b[36]=0 as u8; b[37]=0 as u8
36 b[38]=73 as u8; b[39]=69 as u8; b[40]=78 as u8; b[41]=68 as u8
37 b[42]=174 as u8; b[43]=66 as u8; b[44]=96 as u8; b[45]=130 as u8
38 return 46
39}
40func og_enc(v: i64) -> i64 { let x: i64=v&0x3f; if x<26 {return 65+x} if x<52 {return 97+(x-26)} if x<62 {return 48+(x-52)} if x==62 {return 43} return 47 }
41func og_b64encode(inp: *u8, n: i64, out: *u8) -> i64 {
42 var i: i64 = 0
43 var o: i64 = 0
44 while i + 3 <= n {
45 let b0: i64=inp[i]&0xff; let b1: i64=inp[i+1]&0xff; let b2: i64=inp[i+2]&0xff
46 out[o]=og_enc(b0>>2) as u8; out[o+1]=og_enc(((b0<<4)|(b1>>4))&0x3f) as u8; out[o+2]=og_enc(((b1<<2)|(b2>>6))&0x3f) as u8; out[o+3]=og_enc(b2&0x3f) as u8
47 o=o+4; i=i+3
48 }
49 let rem: i64 = n - i
50 if rem == 1 { let b0: i64=inp[i]&0xff; out[o]=og_enc(b0>>2) as u8; out[o+1]=og_enc((b0<<4)&0x3f) as u8; out[o+2]=61 as u8; out[o+3]=61 as u8; o=o+4 }
51 if rem == 2 { let b0: i64=inp[i]&0xff; let b1: i64=inp[i+1]&0xff; out[o]=og_enc(b0>>2) as u8; out[o+1]=og_enc(((b0<<4)|(b1>>4))&0x3f) as u8; out[o+2]=og_enc((b1<<2)&0x3f) as u8; out[o+3]=61 as u8; o=o+4 }
52 return o
53}
54
55func og_listen(port: i64) -> i64 {
56 let lfd: i64 = sys_socket(2,1,0); if lfd<0 {return 0-1}
57 let optv: *u8 = sys_mmap(4); optv[0]=1 as u8; sys_setsockopt(lfd,1,2,optv,4)
58 let addr: *u8 = sys_mmap(16)
59 addr[0]=2 as u8; addr[1]=0 as u8; addr[2]=((port>>8)&0xff) as u8; addr[3]=(port&0xff) as u8
60 addr[4]=127 as u8; addr[5]=0 as u8; addr[6]=0 as u8; addr[7]=1 as u8
61 var zi: i64=8; while zi<16 {addr[zi]=0 as u8; zi=zi+1}
62 if sys_bind(lfd, addr, 16) < 0 { return 0-1 }
63 if sys_listen(lfd, 16) < 0 { return 0-1 }
64 return lfd
65}
66func og_connect(port: i64) -> i64 {
67 let fd: i64 = sys_socket(2,1,0); if fd<0 {return 0-1}
68 sys_set_socket_timeout(fd, 60)
69 let a: *u8 = sys_mmap(16)
70 a[0]=2 as u8; a[1]=0 as u8; a[2]=((port>>8)&0xff) as u8; a[3]=(port&0xff) as u8
71 a[4]=127 as u8; a[5]=0 as u8; a[6]=0 as u8; a[7]=1 as u8
72 var zi: i64=8; while zi<16 {a[zi]=0 as u8; zi=zi+1}
73 if nx_connect_bounded(fd, a, 16, NX_CONN_DEFAULT_MS) != 0 { sys_close(fd); return 0-1 }
74 return fd
75}
76func og_freshns(prefix: *u8) -> i64 {
77 let mfn: *u8 = sys_mmap(128); var mo: i64=0
78 while (prefix[mo]&0xff)!=0 { mfn[mo]=prefix[mo]; mo=mo+1 }
79 let suf: *u8 = "manifest.txt" as *u8; var so: i64=0
80 while (suf[so]&0xff)!=0 { mfn[mo+so]=suf[so]; so=so+1 }
81 mfn[mo+so]=0 as u8
82 let fd: i64 = sys_openat_wr(mfn, 0x1a4); if fd>=0 {sys_close(fd)}
83 return 0
84}
85func og_truncate(path: *u8) -> i64 { let fd: i64 = sys_openat_wr(path, 0x1a4); if fd>=0 {sys_close(fd)} return 0 }
86
87// read a full HTTP response until the peer closes
88func og_readresp(fd: i64, resp: *u8, cap: i64) -> i64 {
89 var rn: i64 = 0; var g: i64 = 1
90 while g==1 { let r: i64 = sys_read(fd, (resp as i64+rn) as *u8, cap-rn); if r<=0 {g=0} else { rn=rn+r; if rn>=cap {g=0} } }
91 return rn
92}
93
94func main() -> i64 {
95 let png: *u8 = sys_mmap(256); let plen: i64 = og_basepng(png)
96 let b64: *u8 = sys_mmap(512); let b64len: i64 = og_b64encode(png, plen, b64)
97
98 let WPORT: i64 = 39271
99 let OPORT: i64 = 39272
100 og_freshns("/tmp/genorch-" as *u8)
101 og_truncate("/tmp/genorch_cids.tsv" as *u8)
102
103 let cfg: *i64 = sys_mmap(8*16) as *i64
104 cfg[0]=127; cfg[1]=0; cfg[2]=0; cfg[3]=1; cfg[4]=WPORT
105 let sp_store: *u8 = "/tmp/genorch-" as *u8
106 let sp_blob: *u8 = "/tmp/genorchblob-" as *u8
107 let sp_side: *u8 = "/tmp/genorch_cids.tsv" as *u8
108 let sp_host: *u8 = "laptop" as *u8
109 cfg[5]=sp_store as i64; cfg[6]=sp_blob as i64; cfg[7]=sp_side as i64; cfg[8]=sp_host as i64
110
111 let wfd: i64 = og_listen(WPORT); if wfd<0 { og("GATE FAIL: worker listen\n" as *u8); sys_exit(1); return 1 }
112 let ofd: i64 = og_listen(OPORT); if ofd<0 { og("GATE FAIL: orch listen\n" as *u8); sys_exit(1); return 1 }
113
114 // mock GPU worker child: serve 1 REAL request. The orchestrator's MULTI-WORKER availability probe
115 // (go_worker_up) connects-and-closes BEFORE dispatching, so skip 0-byte probe conns and answer the
116 // first conn that actually sends a request (fixture updated to the multi-worker contract).
117 let pidW: i64 = sys_fork()
118 if pidW == 0 {
119 sys_close(ofd)
120 var servedw: i64 = 0
121 var tries: i64 = 0
122 while servedw < 1 {
123 if tries >= 8 { servedw = 1 } else {
124 let c: i64 = sys_accept(wfd)
125 if c >= 0 {
126 let rb: *u8 = sys_mmap(16384)
127 let rr: i64 = sys_read(c, rb, 16384)
128 if rr > 0 {
129 let bod: *u8 = sys_mmap(4096); var bo: i64 = 0
130 bo = og_cat(bod, bo, "{\"images\":[\"" as *u8); bo = og_catb(bod, bo, b64, b64len)
131 bo = og_cat(bod, bo, "\",\"" as *u8); bo = og_catb(bod, bo, b64, b64len); bo = og_cat(bod, bo, "\"]}" as *u8)
132 let resp: *u8 = sys_mmap(8192); var o: i64 = 0
133 o = og_cat(resp, o, "HTTP/1.1 200 OK\r\nContent-Type: application/json\r\nConnection: close\r\nContent-Length: " as *u8)
134 o = og_itoa(resp, o, bo); o = og_cat(resp, o, "\r\n\r\n" as *u8); o = og_catb(resp, o, bod, bo)
135 sys_write(c, resp, o)
136 servedw = 1
137 }
138 sys_close(c)
139 }
140 tries = tries + 1 }
141 }
142 sys_exit(0)
143 }
144 // orchestrator child: serve 4 conns serially
145 let pidO: i64 = sys_fork()
146 if pidO == 0 {
147 sys_close(wfd)
148 var i: i64 = 0
149 while i < 4 { let c: i64 = sys_accept(ofd); if c>=0 { go_serve_conn(c, cfg); sys_close(c) } i=i+1 }
150 sys_exit(0)
151 }
152 sys_close(wfd); sys_close(ofd)
153
154 // (1) POST /api/generate
155 let fd1: i64 = og_connect(OPORT)
156 if fd1 < 0 { og("GATE FAIL: connect orch (generate)\n" as *u8); sys_exit(1); return 1 }
157 let gbody: *u8 = "{\"prompt\":\"a red fox\",\"width\":512,\"height\":512,\"steps\":8,\"cfg\":\"1.0\",\"seed\":1000,\"count\":2,\"sampler\":\"euler\"}" as *u8
158 var gbl: i64 = 0; while gbody[gbl]!=(0 as u8){gbl=gbl+1}
159 let greq: *u8 = sys_mmap(4096); var go1: i64 = 0
160 go1 = og_cat(greq, go1, "POST /api/generate HTTP/1.1\r\nHost: x\r\nConnection: close\r\nContent-Type: application/json\r\nContent-Length: " as *u8)
161 go1 = og_itoa(greq, go1, gbl); go1 = og_cat(greq, go1, "\r\n\r\n" as *u8); go1 = og_catb(greq, go1, gbody, gbl)
162 sys_write(fd1, greq, go1)
163 let gresp: *u8 = sys_mmap(131072)
164 let grn: i64 = og_readresp(fd1, gresp, 131072)
165 sys_close(fd1)
166 if og_find(gresp, grn, "200 OK" as *u8) < 0 { og("GATE FAIL: /api/generate not 200\n" as *u8); sys_write(1, gresp, grn); sys_exit(1); return 1 }
167 if og_find(gresp, grn, "\"count\":2" as *u8) < 0 { og("GATE FAIL: count != 2 in generate response\n" as *u8); sys_write(1, gresp, grn); sys_exit(1); return 1 }
168 og("GATE ROW generate: GREEN -- POST /api/generate dispatched to worker, processed 2 images, returned CIDs\n" as *u8)
169
170 // extract first cid
171 let cidp: i64 = og_find(gresp, grn, "\"cids\":[\"" as *u8)
172 if cidp < 0 { og("GATE FAIL: no cids[] in response\n" as *u8); sys_exit(1); return 1 }
173 let cid: *u8 = sys_mmap(128)
174 var ci: i64 = cidp + 9
175 var co: i64 = 0
176 var cstop: i64 = 0
177 while cstop==0 { if ci>=grn { cstop=1 } else { if (gresp[ci]&0xff)==34 { cstop=1 } else { cid[co]=gresp[ci]; co=co+1; ci=ci+1 } } }
178 cid[co]=0 as u8
179
180 // (2) GET /img/<cid>
181 let fd2: i64 = og_connect(OPORT)
182 if fd2 < 0 { og("GATE FAIL: connect orch (img)\n" as *u8); sys_exit(1); return 1 }
183 let ireq: *u8 = sys_mmap(256); var io: i64 = 0
184 io = og_cat(ireq, io, "GET /img/" as *u8); io = og_catb(ireq, io, cid, co)
185 io = og_cat(ireq, io, " HTTP/1.1\r\nHost: x\r\nConnection: close\r\n\r\n" as *u8)
186 sys_write(fd2, ireq, io)
187 let iresp: *u8 = sys_mmap(131072)
188 let irn: i64 = og_readresp(fd2, iresp, 131072)
189 sys_close(fd2)
190 if og_find(iresp, irn, "200 OK" as *u8) < 0 { og("GATE FAIL: /img not 200\n" as *u8); sys_write(1, iresp, irn); sys_exit(1); return 1 }
191 if og_find(iresp, irn, "image/png" as *u8) < 0 { og("GATE FAIL: /img not image/png\n" as *u8); sys_exit(1); return 1 }
192 // PNG signature byte 137 must appear in the body
193 if og_find(iresp, irn, "IEND" as *u8) < 0 { og("GATE FAIL: /img body is not a PNG (no IEND)\n" as *u8); sys_exit(1); return 1 }
194 og("GATE ROW serve-img: GREEN -- GET /img/<cid> -> 200 image/png with a real PNG blob (sidecar lookup ok)\n" as *u8)
195
196 // (3) GET / (UI)
197 let fd3: i64 = og_connect(OPORT)
198 if fd3 < 0 { og("GATE FAIL: connect orch (ui)\n" as *u8); sys_exit(1); return 1 }
199 let ureq: *u8 = sys_mmap(128); let ul: i64 = og_cat(ureq, 0, "GET / HTTP/1.1\r\nHost: x\r\nConnection: close\r\n\r\n" as *u8)
200 sys_write(fd3, ureq, ul)
201 let uresp: *u8 = sys_mmap(65536)
202 let urn: i64 = og_readresp(fd3, uresp, 65536)
203 sys_close(fd3)
204 if og_find(uresp, urn, "200 OK" as *u8) < 0 { og("GATE FAIL: GET / not 200\n" as *u8); sys_exit(1); return 1 }
205 if og_find(uresp, urn, "Elder AI" as *u8) < 0 { og("GATE FAIL: GET / missing the gen UI\n" as *u8); sys_exit(1); return 1 }
206 og("GATE ROW ui: GREEN -- GET / -> 200 with the Elder AI gen UI (batch form)\n" as *u8)
207
208 // (4) one more orch conn to let the server loop reach 4 + exit cleanly, then reap
209 let fd4: i64 = og_connect(OPORT)
210 if fd4 >= 0 { let hreq: *u8 = sys_mmap(128); let hl: i64 = og_cat(hreq, 0, "GET /health HTTP/1.1\r\nHost: x\r\nConnection: close\r\n\r\n" as *u8); sys_write(fd4, hreq, hl); let hr: *u8 = sys_mmap(1024); og_readresp(fd4, hr, 1024); sys_close(fd4) }
211 let stx: *i64 = sys_mmap(16) as *i64
212 sys_wait4(pidW, stx, 0)
213 sys_wait4(pidO, stx, 0)
214
215 og("GEN-ORCH-GATE 3/3 GREEN -- generate(socket dispatch+pipeline) + serve-img(blob) + UI, all over real loopback HTTP\n" as *u8)
216 sys_exit(0)
217 return 0
218}