nx_request_pool_candidate_t230.nx source
↩ module page · 243 lines · 8902 B
1// Single-owner IPC pool. Admission policy supplies capacity; this layer owns
2// response-buffer reservations, message framing and child reaping.
3import "nx_resource_arbiter.nx"
4
5const RP_CELL: i64 = 8 // current native ABI
6const RP_FD_MASK: i64 = 4294967295
7const RP_POLLIN: i64 = 1
8const RP_EINTR: i64 = -4
9const RP_SIGKILL: i64 = 9
10
11struct NxRequestWorker {
12 pid: i64,
13 fd: i64,
14 buffer: *u8,
15 used: i64,
16 cleanup_error: i64,
17}
18
19struct NxRequestPool {
20 workers: *u8,
21 capacity: i64,
22 response_bytes: i64,
23 active: i64,
24 pollfds: *i64,
25 arbiter: *NxResourceArbiter,
26 child_fd: i64,
27 last_error: i64,
28 // ADDITIVE (2026-09-11, lane conn): the slot of the last PER-REQUEST worker
29 // failure (a child that exited non-zero, or an over-budget response), or -1
30 // for a fatal-to-the-pool failure (poll error, broken output fd, reap fault)
31 // and for any successful pump. rp_pump's RETURN VALUE and last_error are
32 // unchanged, so the generic-pool contract and its gate are untouched; a
33 // caller that wants per-request recovery reads this after a -1 to learn WHICH
34 // request failed and emit a correlated error instead of tearing the pool down.
35 failed_slot: i64,
36}
37
38func rp_worker(p: *NxRequestPool, slot: i64) -> *NxRequestWorker {
39 return (p.workers + slot * __size_of(NxRequestWorker)) as *NxRequestWorker
40}
41
42func rp_new(capacity: i64, response_bytes: i64) -> *NxRequestPool {
43 if capacity <= 0 { return 0 as *NxRequestPool }
44 if response_bytes <= 0 { return 0 as *NxRequestPool }
45 if response_bytes >= NX_RA_SIZE_MAX { return 0 as *NxRequestPool }
46 if capacity > NX_RA_SIZE_MAX / __size_of(NxRequestWorker) { return 0 as *NxRequestPool }
47 if capacity > NX_RA_SIZE_MAX / (response_bytes + 1) { return 0 as *NxRequestPool }
48 let p: *NxRequestPool = sys_mmap(__size_of(NxRequestPool)) as *NxRequestPool
49 p.workers = sys_mmap(capacity * __size_of(NxRequestWorker))
50 p.pollfds = sys_mmap((capacity + 1) * RP_CELL) as *i64
51 p.capacity = capacity
52 p.response_bytes = response_bytes
53 p.active = 0
54 p.child_fd = -1
55 p.last_error = 0
56 p.failed_slot = -1
57 p.arbiter = nx_ra_new(capacity)
58 nx_ra_set_budget(p.arbiter,NX_RA_KIND_RAM_BYTES,capacity * (response_bytes + 1))
59 nx_ra_set_max_share(p.arbiter,NX_RA_KIND_RAM_BYTES,NX_RA_Q10_ONE)
60 return p
61}
62
63// In child return 0 with child_fd set; in parent return PID. No worker
64// receives another worker's pipe writer, so death produces observable EOF.
65func rp_fork(p: *NxRequestPool) -> i64 {
66 var slot: i64 = 0
67 while slot < p.capacity {
68 if rp_worker(p,slot).pid == 0 { break }
69 slot = slot + 1
70 }
71 if slot == p.capacity { return -1 }
72 var granted: i64 = 0
73 if nx_ra_request(p.arbiter,slot+1,NX_RA_KIND_RAM_BYTES,p.response_bytes+1,0,0,&granted,0) != NX_RA_V_GRANTED_FULL { return -1 }
74 var pair: i64 = 0
75 let pipe_rc: i64 = sys_pipe2(&pair,0)
76 if pipe_rc != 0 {
77 nx_ra_release(p.arbiter,slot+1,NX_RA_KIND_RAM_BYTES)
78 p.last_error = pipe_rc
79 return -1
80 }
81 let rd: i64 = pair & RP_FD_MASK
82 let wr: i64 = (pair >> 32) & RP_FD_MASK
83 let pid: i64 = sys_fork()
84 if pid == 0 {
85 sys_close(rd)
86 var i: i64 = 0
87 while i < p.capacity {
88 if rp_worker(p,i).pid != 0 { sys_close(rp_worker(p,i).fd) }
89 i = i + 1
90 }
91 p.child_fd = wr
92 return 0
93 }
94 sys_close(wr)
95 if pid < 0 {
96 sys_close(rd)
97 nx_ra_release(p.arbiter,slot+1,NX_RA_KIND_RAM_BYTES)
98 p.last_error = pid
99 return -1
100 }
101 let w: *NxRequestWorker = rp_worker(p,slot)
102 w.pid = pid
103 w.fd = rd
104 w.buffer = sys_mmap(p.response_bytes+1)
105 w.used = 0
106 w.cleanup_error = 0
107 p.active = p.active + 1
108 return pid
109}
110
111func rp_write(fd: i64, data: *u8, n: i64) -> i64 {
112 var at: i64 = 0
113 while at < n {
114 let rc: i64 = sys_write(fd,data+at,n-at)
115 if rc == RP_EINTR { continue }
116 if rc <= 0 { return -1 }
117 at = at + rc
118 }
119 return 0
120}
121
122func rp_release(p: *NxRequestPool, slot: i64) -> i64 {
123 let w: *NxRequestWorker = rp_worker(p,slot)
124 sys_close(w.fd)
125 sys_munmap(w.buffer as *u8,p.response_bytes+1)
126 w.pid = 0
127 w.fd = -1
128 w.buffer = 0 as *u8
129 w.used = 0
130 p.active = p.active - 1
131 nx_ra_release(p.arbiter,slot+1,NX_RA_KIND_RAM_BYTES)
132 return 0
133}
134
135// Returns stdin readiness (1), worker progress (0), or failure (-1).
136// Buffer whole responses and emit only after successful worker termination.
137func rp_pump_timeout(p: *NxRequestPool, accept_input: i64, output_fd: i64, wait_ms: i64) -> i64 {
138 p.failed_slot = 0 - 1
139 p.pollfds[0] = RP_FD_MASK
140 if accept_input != 0 { p.pollfds[0] = RP_POLLIN << 32 }
141 var i: i64 = 0
142 while i < p.capacity {
143 let w: *NxRequestWorker = rp_worker(p,i)
144 p.pollfds[i+1] = RP_FD_MASK
145 if w.pid != 0 { p.pollfds[i+1] = w.fd | (RP_POLLIN << 32) }
146 i = i + 1
147 }
148 // Linux poll takes a signed 32-bit millisecond interval. Longer queue
149 // deadlines use multiple finite waits rather than truncating into infinity.
150 var poll_ms:i64=wait_ms
151 if poll_ms>2147483647{poll_ms=2147483647}
152 let polled: i64 = sys_poll(p.pollfds as *u8,p.capacity+1,poll_ms)
153 if polled == RP_EINTR { return 0 }
154 if polled < 0 { p.last_error = polled; return -1 }
155 i = 0
156 while i < p.capacity {
157 let w: *NxRequestWorker = rp_worker(p,i)
158 if w.pid != 0 {
159 if (p.pollfds[i+1] >> 48) != 0 {
160 let used: i64 = w.used
161 let buf: *u8 = w.buffer as *u8
162 let got: i64 = sys_read(w.fd,buf+used,p.response_bytes+1-used)
163 if got < 0 { if got != RP_EINTR { p.last_error = got; return -1 } }
164 if got > 0 {
165 w.used = used+got
166 if used+got > p.response_bytes { p.failed_slot = i; p.last_error = -2; return -1 }
167 }
168 if got == 0 {
169 var status: i64 = 0
170 var waited: i64 = RP_EINTR
171 while waited == RP_EINTR { waited = sys_wait4(w.pid,&status,0) }
172 if waited != w.pid { p.last_error = waited; return -1 }
173 var write_rc: i64 = 0
174 if status == 0 { write_rc = rp_write(output_fd,buf,used) }
175 rp_release(p,i)
176 if status != 0 { p.failed_slot = i; p.last_error = status; return -1 }
177 if write_rc != 0 { p.last_error = -3; return -1 }
178 }
179 }
180 }
181 i = i + 1
182 }
183 if (p.pollfds[0] >> 48) != 0 { return 1 }
184 return 0
185}
186
187// Stop and reap are separate so whole-pool shutdown signals every worker
188// before waiting. Failed observations retain the worker and its reservation.
189func rp_stop_worker(p: *NxRequestPool, slot: i64) -> i64 {
190 let w: *NxRequestWorker = rp_worker(p,slot)
191 if w.pid == 0 { return 0 }
192 w.cleanup_error = nx_kill(w.pid,RP_SIGKILL)
193 return w.cleanup_error
194}
195
196func rp_reap_stopped(p: *NxRequestPool, slot: i64) -> i64 {
197 let w: *NxRequestWorker = rp_worker(p,slot)
198 if w.pid == 0 { return 0 }
199 if w.cleanup_error != 0 { return w.cleanup_error }
200 var status: i64 = 0
201 var waited: i64 = RP_EINTR
202 while waited == RP_EINTR { waited = sys_wait4(w.pid,&status,0) }
203 if waited != w.pid { w.cleanup_error = waited; return waited }
204 return rp_release(p,slot)
205}
206
207// The expected PID protects against a delayed cancellation targeting a reused
208// slot. Only this parent reaps pool children, keeping live PIDs owned.
209func rp_cancel_owned(p: *NxRequestPool, slot: i64, expected_pid: i64) -> i64 {
210 if p == (0 as *NxRequestPool) { return -1 }
211 if slot < 0 { return -1 }
212 if slot >= p.capacity { return -1 }
213 if expected_pid <= 0 { return -1 }
214 let w: *NxRequestWorker = rp_worker(p,slot)
215 if w.pid == 0 { return 0 }
216 if w.pid != expected_pid { return -1 }
217 let stopped: i64 = rp_stop_worker(p,slot)
218 if stopped != 0 { p.last_error = stopped; return -1 }
219 let reaped: i64 = rp_reap_stopped(p,slot)
220 if reaped != 0 { p.last_error = reaped; return -1 }
221 return 0
222}
223
224func rp_cancel_all(p: *NxRequestPool) -> i64 {
225 var i: i64 = 0
226 var failure: i64 = 0
227 while i < p.capacity {
228 let rc: i64 = rp_stop_worker(p,i)
229 if rc != 0 { if failure == 0 { failure = rc } }
230 i = i + 1
231 }
232 i = 0
233 while i < p.capacity {
234 let rc: i64 = rp_reap_stopped(p,i)
235 if rc != 0 { if failure == 0 { failure = rc } }
236 i = i + 1
237 }
238 if failure != 0 { p.last_error = failure; return -1 }
239 return 0
240}
241
242// Legacy indefinite wait retained; queue consumers supply their nearest deadline.
243func rp_pump(p:*NxRequestPool,accept_input:i64,output_fd:i64)->i64 { return rp_pump_timeout(p,accept_input,output_fd,-1) }