nx_remote_worker.nx source
↩ module page · 144 lines · 5405 B
1// nx_remote_worker.nx -- distributed-MIMD pattern: orchestrator
2// dispatches typed tasks to a worker over NetChan; worker computes
3// and returns results. Single-process demo (worker is a spawned
4// thread that listens on TCP); same shape works across hosts when
5// the worker side runs as a remote process.
6//
7// Composes against:
8// [[nx_net_chan_tcp_i64]] -- transport
9// [[thread_clone_native_trampoline]] -- worker thread
10// [[fn_ptr_indirect_call]] -- worker dispatches via typed fn-ptr
11//
12// Worker protocol (per task):
13// 1. orchestrator sends task_id (i64)
14// 2. orchestrator sends arg (i64)
15// 3. worker runs registered_fn(task_id, arg) -> result (i64)
16// 4. worker sends result (i64)
17// 5. orchestrator recvs result
18//
19// Sentinel task_id (-1) signals worker to exit cleanly.
20
21// nx_safety_envelope:
22// intended_use: AUTO_APPLIED -- primitive-specific tuning queued
23// sil_target: SIL1
24// evidence: [bulk_applied_2026-05-16, see-file-comment-for-detail]
25// verdict: NOT_YET_EVALUATED
26
27import "nx_syscalls.nx"
28import "nx_atom.nx"
29import "nx_thread.nx"
30import "nx_socket.nx"
31import "nx_net_chan.nx"
32const NX_MAGIC_65536: i64 = 65536
33const NX_MAGIC_100000000: i64 = 100000000
34
35const NX_REMOTE_SENTINEL: i64 = -1
36
37struct NxRemoteWorker {
38 listen_port: i64,
39 handler_fn: func(i64, i64) -> i64, // (task_id, arg) -> result
40 ready_flag: i64, // atomic; 1 when listening
41 exited_flag: i64, // atomic; 1 when worker thread done
42}
43
44const NX_REMOTE_WORKER_BYTES: i64 = 32
45
46// Worker thread main. arg is *NxRemoteWorker. Listens on TCP,
47// accepts one connection, then runs task-dispatch loop until
48// receiving sentinel task_id.
49func _nx_remote_worker_main(arg: *u8) -> i64 {
50 let w: *NxRemoteWorker = arg as *NxRemoteWorker
51 let ready_addr: *i64 = ((arg as i64) + 16) as *i64
52 let exited_addr: *i64 = ((arg as i64) + 24) as *i64
53
54 let lfd: i64 = nx_sock_socket(NX_AF_INET, NX_SOCK_STREAM, NX_IPPROTO_TCP)
55 if lfd < 0 { nx_atom_store_i64(exited_addr, 1, NX_MO_SEQ_CST); return 0 }
56 nx_sock_reuseaddr(lfd)
57 let addr: *u8 = sys_mmap(16)
58 nx_sock_sin_init(addr, 0, w.listen_port)
59 if nx_sock_bind(lfd, addr, 16) < 0 { nx_atom_store_i64(exited_addr, 1, NX_MO_SEQ_CST); return 0 }
60 if nx_sock_listen(lfd, 1) < 0 { nx_atom_store_i64(exited_addr, 1, NX_MO_SEQ_CST); return 0 }
61
62 // Signal orchestrator that listener is up.
63 nx_atom_store_i64(ready_addr, 1, NX_MO_SEQ_CST)
64
65 let cfd: i64 = nx_sock_accept(lfd, 0 as *u8, 0 as *i64)
66 if cfd < 0 { nx_atom_store_i64(exited_addr, 1, NX_MO_SEQ_CST); return 0 }
67 let nc: *NxNetChan = nx_net_chan_from_fd(cfd)
68
69 let id_raw: *u8 = sys_mmap(16)
70 let arg_raw: *u8 = sys_mmap(16)
71 let id_v: *i64 = id_raw as *i64
72 let arg_v: *i64 = arg_raw as *i64
73
74 var running: i64 = 1
75 while running == 1 {
76 if nx_net_chan_recv(nc, id_v) != 0 { running = 0 }
77 else {
78 if *id_v == NX_REMOTE_SENTINEL { running = 0 }
79 else {
80 if nx_net_chan_recv(nc, arg_v) != 0 { running = 0 }
81 else {
82 let fp: func(i64, i64) -> i64 = w.handler_fn
83 let r: i64 = fp(*id_v, *arg_v)
84 if nx_net_chan_send(nc, r) != 0 { running = 0 }
85 }
86 }
87 }
88 }
89 nx_net_chan_close(nc)
90 sys_close(lfd)
91 nx_atom_store_i64(exited_addr, 1, NX_MO_SEQ_CST)
92 return 0
93}
94
95// Spawn the worker thread + wait until it's listening.
96func nx_remote_worker_start(handler: func(i64, i64) -> i64, port: i64) -> *NxRemoteWorker {
97 let raw: *u8 = sys_mmap(NX_REMOTE_WORKER_BYTES)
98 let w: *NxRemoteWorker = raw as *NxRemoteWorker
99 w.listen_port = port
100 w.handler_fn = handler
101 w.ready_flag = 0
102 w.exited_flag = 0
103
104 nx_thread_spawn_fn(_nx_remote_worker_main, raw, NX_MAGIC_65536)
105
106 // Spin until worker signals ready.
107 let ready_addr: *i64 = ((raw as i64) + 16) as *i64
108 var spins: i64 = 0
109 while nx_atom_load_i64(ready_addr, NX_MO_SEQ_CST) == 0 {
110 nx_thread_yield()
111 spins = spins + 1
112 if spins > NX_MAGIC_100000000 { return w } // caller can detect via still-zero ready
113 }
114 return w
115}
116
117// Orchestrator-side: open NetChan to worker. Returns NULL-ish on failure.
118func nx_remote_orchestrator_connect(port: i64) -> *NxNetChan {
119 let fd: i64 = nx_sock_socket(NX_AF_INET, NX_SOCK_STREAM, NX_IPPROTO_TCP)
120 if fd < 0 { return 0 as *NxNetChan }
121 let addr: *u8 = sys_mmap(16)
122 nx_sock_sin_init(addr, 0x0100007F, port)
123 if nx_sock_connect(fd, addr, 16) < 0 { sys_close(fd); return 0 as *NxNetChan }
124 return nx_net_chan_from_fd(fd)
125}
126
127// Send one task + recv its result. Returns 0 + writes *out on success.
128func nx_remote_dispatch(nc: *NxNetChan, task_id: i64, arg: i64, out: *i64) -> i64 {
129 if nx_net_chan_send(nc, task_id) != 0 { return -1 }
130 if nx_net_chan_send(nc, arg) != 0 { return -2 }
131 return nx_net_chan_recv(nc, out)
132}
133
134// Send sentinel to ask worker to exit cleanly.
135func nx_remote_shutdown(nc: *NxNetChan) -> i64 {
136 nx_net_chan_send(nc, NX_REMOTE_SENTINEL)
137 return nx_net_chan_close(nc)
138}
139
140func nx_remote_worker_alive(w: *NxRemoteWorker) -> i64 {
141 let exited_addr: *i64 = ((w as i64) + 24) as *i64
142 if nx_atom_load_i64(exited_addr, NX_MO_SEQ_CST) == 1 { return 0 }
143 return 1
144}