code wiki / _hdl_build / nx_gen_worker.nx
nx_gen_worker.nx source
↩ module page · 116 lines · 7233 B
1// nx_gen_worker.nx -- the GenWorker object: the GPU-worker CLIENT. Owns dispatch to a worker (a.b.c.d:port) for
2// image gen (/v1/images/generations) and chat (/v1/chat/completions), plus the non-blocking TCP health-check that
3// makes multi-worker availability-based selection + failover safe (a powered-off worker must never hang a dispatch).
4// Extracted from nx_gen_orchestrator (monolith -> objects); behaviour-preserving. THIS is where ER-8 model-select-by-
5// intent will land (route SFW->controlled model, explicit->turbo). Self-contained (private gw_* helpers). Object
6// model: companion_arch.md. license_tier: ORIGINAL
7import "nx_syscalls.nx"
8
9func gw_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 }
10func gw_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 }
11func gw_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 }
12
13// dispatch an image-gen request: POST /v1/images/generations to worker (a.b.c.d:port), read full response.
14func go_worker_dispatch(ha: i64, hb: i64, hc: i64, hd: i64, port: i64, body: *u8, blen: i64, respbuf: *u8, respcap: i64) -> i64 {
15 let fd: i64 = sys_socket(2, 1, 0)
16 if fd < 0 { return 0 - 1 }
17 sys_set_socket_timeout(fd, 180)
18 let a: *u8 = sys_mmap(16)
19 a[0]=2 as u8; a[1]=0 as u8; a[2]=((port>>8)&0xff) as u8; a[3]=(port&0xff) as u8
20 a[4]=ha as u8; a[5]=hb as u8; a[6]=hc as u8; a[7]=hd as u8
21 var zi: i64=8; while zi<16 { a[zi]=0 as u8; zi=zi+1 }
22 if sys_connect(fd, a, 16) != 0 { sys_close(fd); return 0 - 2 }
23 let rq: *u8 = sys_mmap(16384); var o: i64 = 0
24 o = gw_cat(rq, o, "POST /v1/images/generations HTTP/1.1\r\nHost: 127.0.0.1\r\nAccept: */*\r\nConnection: close\r\nContent-Type: application/json\r\nContent-Length: " as *u8)
25 o = gw_itoa(rq, o, blen); o = gw_cat(rq, o, "\r\n\r\n" as *u8); o = gw_catb(rq, o, body, blen)
26 sys_write(fd, rq, o)
27 var total: i64 = 0
28 var gof: i64 = 1
29 while gof == 1 {
30 let r: i64 = sys_read(fd, (respbuf as i64 + total) as *u8, respcap - total)
31 if r <= 0 { gof = 0 } else { total = total + r; if total >= respcap { gof = 0 } }
32 }
33 sys_close(fd)
34 return total
35}
36// dispatch an IMG2IMG request: POST /sdapi/v1/img2img (A1111 shape, which stable-diffusion.cpp serves) to the worker.
37// The body carries a base64 init image (~1.4MB+ for 512px) so we write a SMALL header then STREAM the caller's body
38// from its pointer in a write-loop (no multi-MB request copy, and one sys_write cannot flush a MB in a single call).
39// Response is {"images":["<b64>"]} -> the SAME gp_response A1111 branch the txt2img path already ingests. This is
40// where structure-preserving edits (digital-twin refine, attribute change, and mask-based inpaint) dispatch.
41func go_worker_img2img(ha: i64, hb: i64, hc: i64, hd: i64, port: i64, body: *u8, blen: i64, respbuf: *u8, respcap: i64) -> i64 {
42 let fd: i64 = sys_socket(2, 1, 0)
43 if fd < 0 { return 0 - 1 }
44 sys_set_socket_timeout(fd, 200)
45 let a: *u8 = sys_mmap(16)
46 a[0]=2 as u8; a[1]=0 as u8; a[2]=((port>>8)&0xff) as u8; a[3]=(port&0xff) as u8
47 a[4]=ha as u8; a[5]=hb as u8; a[6]=hc as u8; a[7]=hd as u8
48 var zi: i64=8; while zi<16 { a[zi]=0 as u8; zi=zi+1 }
49 if sys_connect(fd, a, 16) != 0 { sys_close(fd); return 0 - 2 }
50 let hdr: *u8 = sys_mmap(1024); var ho: i64 = 0
51 ho = gw_cat(hdr, 0, "POST /sdapi/v1/img2img HTTP/1.1\r\nHost: 127.0.0.1\r\nAccept: */*\r\nConnection: close\r\nContent-Type: application/json\r\nContent-Length: " as *u8)
52 ho = gw_itoa(hdr, ho, blen); ho = gw_cat(hdr, ho, "\r\n\r\n" as *u8)
53 sys_write(fd, hdr, ho)
54 var sent: i64 = 0
55 while sent < blen { let w: i64 = sys_write(fd, (body as i64 + sent) as *u8, blen - sent); if w <= 0 { sent = blen } else { sent = sent + w } }
56 var total: i64 = 0
57 var gof: i64 = 1
58 while gof == 1 {
59 let r: i64 = sys_read(fd, (respbuf as i64 + total) as *u8, respcap - total)
60 if r <= 0 { gof = 0 } else { total = total + r; if total >= respcap { gof = 0 } }
61 }
62 sys_close(fd)
63 return total
64}
65// non-blocking TCP health-check ("electronic availability"): connect non-blocking + poll(POLLOUT, 2s) + SO_ERROR.
66// 1=reachable, 0=down/refused/timeout. = the multi-worker availability selector's liveness probe.
67func go_worker_up(ha: i64, hb: i64, hc: i64, hd: i64, port: i64) -> i64 {
68 let fd: i64 = sys_socket(2, 1, 0)
69 if fd < 0 { return 0 }
70 let fl: i64 = __syscall(72, fd, 3, 0, 0, 0, 0) // F_GETFL
71 __syscall(72, fd, 4, fl | 2048, 0, 0, 0) // F_SETFL | O_NONBLOCK(04000)
72 let a: *u8 = sys_mmap(16)
73 a[0]=2 as u8; a[1]=0 as u8; a[2]=((port>>8)&0xff) as u8; a[3]=(port&0xff) as u8
74 a[4]=ha as u8; a[5]=hb as u8; a[6]=hc as u8; a[7]=hd as u8
75 var zi: i64=8; while zi<16 { a[zi]=0 as u8; zi=zi+1 }
76 let rc: i64 = sys_connect(fd, a, 16)
77 var up: i64 = 0
78 if rc == 0 { up = 1 }
79 else { if rc == (0 - 115) { // EINPROGRESS -> wait for connect to settle
80 let pfd: *u8 = sys_mmap(8)
81 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
82 pfd[4]=4 as u8; pfd[5]=0 as u8 // POLLOUT
83 if sys_poll(pfd, 1, 2000) > 0 {
84 let so: *i64 = sys_mmap(16) as *i64; so[0]=0
85 let sl: *i64 = sys_mmap(16) as *i64; sl[0]=4
86 __syscall(55, fd, 1, 4, so as i64, sl as i64, 0) // getsockopt SOL_SOCKET SO_ERROR
87 if (so[0] & 0xffffffff) == 0 { up = 1 }
88 }
89 } }
90 sys_close(fd)
91 return up
92}
93// dispatch a chat request: POST /v1/chat/completions to the worker (same engine; Qwen text-encoder doubles as the
94// companion LLM). Thin authed proxy; the caller owns the messages array.
95func go_chat_dispatch(ha: i64, hb: i64, hc: i64, hd: i64, port: i64, body: *u8, blen: i64, respbuf: *u8, respcap: i64) -> i64 {
96 let fd: i64 = sys_socket(2, 1, 0)
97 if fd < 0 { return 0 - 1 }
98 sys_set_socket_timeout(fd, 25)
99 let a: *u8 = sys_mmap(16)
100 a[0]=2 as u8; a[1]=0 as u8; a[2]=((port>>8)&0xff) as u8; a[3]=(port&0xff) as u8
101 a[4]=ha as u8; a[5]=hb as u8; a[6]=hc as u8; a[7]=hd as u8
102 var zi: i64=8; while zi<16 { a[zi]=0 as u8; zi=zi+1 }
103 if sys_connect(fd, a, 16) != 0 { sys_close(fd); return 0 - 2 }
104 let rq: *u8 = sys_mmap(131072); var o: i64 = 0
105 o = gw_cat(rq, o, "POST /v1/chat/completions HTTP/1.1\r\nHost: 127.0.0.1\r\nAccept: */*\r\nConnection: close\r\nContent-Type: application/json\r\nContent-Length: " as *u8)
106 o = gw_itoa(rq, o, blen); o = gw_cat(rq, o, "\r\n\r\n" as *u8); o = gw_catb(rq, o, body, blen)
107 sys_write(fd, rq, o)
108 var total: i64 = 0
109 var gof: i64 = 1
110 while gof == 1 {
111 let r: i64 = sys_read(fd, (respbuf as i64 + total) as *u8, respcap - total)
112 if r <= 0 { gof = 0 } else { total = total + r; if total >= respcap { gof = 0 } }
113 }
114 sys_close(fd)
115 return total
116}