==> src/lifecycle.c <==1/* Per tick and per slot: run requests, completions, phases, publishing; launching, stopping, failing,
2 * unloading; pipes between programs; the http program's load requests. */
3#include "driver_int.h"
4
5#include "mesh/cpuuse.h"
6#include "mesh/rsfmt.h"
7
8#include <stdio.h>
9#include <stdlib.h>
10#include <string.h>
11
12void drv_native_words(mesh_driver *d, uint32_t s);
13
14#define PUSH(arr, n, cap, v) \
15 do { \
16 if ((n) == (cap)) { \
17 (cap) = (cap) ? 2 * (cap) : 8; \
18 (arr) = mesh_xrealloc((arr), (cap) * sizeof *(arr)); \
19 } \
20 (arr)[(n)++] = (v); \
21 } while (0)
22
23static size_t ctl(uint32_t s, size_t off) { return mesh_prog_ctl(s) + off; }
24
25/* A stream reported an error: the CUDA context is sticky-broken. */
26void drv_context_fault(mesh_driver *d, const char *e) {
27 if (d->fault) return;
28 mesh_error("kernel fault: %s; the CUDA context is lost, restart meshd", e);
29 d->fault = mesh_xstrdup(e);
30 /* The CPU executors (wasm lanes, native workers) do not run in the context: they go on. */
31 uint32_t gpu = drv_gpu_kind(d);
32 for (uint32_t s = 0; s < MESH_PROG_SLOTS; s++) {
33 dslot *x = d->slots[s];
34 if (x && x->image.exec == gpu && !x->failed) x->failed = mesh_buf_fmt("CUDA context lost: %s", e);
35 }
36}
37
38/* Appends to the slot's input queue (IQ_*): bytes at IQ_DATA + head % len, then head (release). */
39int drv_input(mesh_driver *d, uint32_t s, const uint8_t *bytes, size_t n, char **text) {
40 mesh_mailbox mb = d->sh->mb;
41 dslot *x = d->slots[s];
42 const mesh_manifest *m = &x->image.manifest;
43 if (m->input_len == 0) { *text = mesh_buf_fmt("%s declares no input queue", m->name); return -1; }
44 size_t q = (size_t)x->view_off + m->input_off;
45 uint32_t head = mesh_mailbox_get(mb, q + MESH_IQ_HEAD), tail = mesh_mailbox_get(mb, q + MESH_IQ_TAIL);
46 uint32_t used = head - tail < m->input_len ? head - tail : m->input_len;
47 uint32_t free_ = m->input_len - used;
48 if (n > free_) { *text = mesh_buf_fmt("%s: input queue full (%u bytes free)", m->name, free_); return -1; }
49 uint32_t h = head;
50 for (size_t i = 0; i < n; i++) {
51 /* A word RMW: the GPU only reads the data bytes, and only those before head. */
52 size_t at = q + MESH_IQ_DATA + (h % m->input_len);
53 size_t w = at & ~(size_t)3, sh = 8 * (at & 3);
54 mesh_mailbox_set(mb, w, (mesh_mailbox_get(mb, w) & ~(0xffu << sh)) | (uint32_t)bytes[i] << sh);
55 h++;
56 }
57 mesh_mailbox_publish(mb, q + MESH_IQ_HEAD, h);
58 *text = mesh_buf_fmt("%zu bytes", n);
59 return 0;
60}
61
62/* ---------------------------------------------------------------- pipes */
63
64/* Releases abandoned pipe cells whose consumers reached them. */
65static void dead_cells_tick(mesh_driver *d) {
66 if (!d->ndead) return;
67 mesh_mailbox mb = d->sh->mb;
68 mesh_words w = mesh_mailbox_words(mb);
69 size_t k = 0;
70 for (size_t i = 0; i < d->ndead; i++) {
71 ddead *c = &d->dead[i];
72 mesh_pipe_inbox now;
73 /* The consumer's inbox as it is now: another view (upgraded, unloaded) is not that ring any more. */
74 if (mesh_pipe_target(&w, c->consumer, c->consumer, &now) != 0 || now.base != c->inbox.base || now.cap != c->inbox.cap ||
75 now.rec != c->inbox.rec)
76 continue;
77 mesh_pipe_ring r = mesh_pipe_inbox_side(&c->inbox, c->side);
78 int got = mesh_pipe_ring_skip_dead(&r, &w, c->producer, c->pos);
79 if (got == 1) {
80 char cn[32], pn[32];
81 if (d->slots[c->consumer]) snprintf(cn, sizeof cn, "%s", drv_name(d->slots[c->consumer]));
82 else snprintf(cn, sizeof cn, "slot %u", c->consumer);
83 if (d->slots[c->producer]) snprintf(pn, sizeof pn, "%s", drv_name(d->slots[c->producer]));
84 else snprintf(pn, sizeof pn, "slot %u", c->producer);
85 mesh_info("pipes: %s's inbox: released position %u that %s reserved and never committed", cn, c->pos, pn);
86 continue;
87 }
88 if (got == 0) d->dead[k++] = *c;
89 }
90 d->ndead = k;
91}
92
93typedef struct {
94 mesh_driver *d;
95 uint32_t consumer, side, producer;
96 mesh_pipe_inbox inbox;
97 size_t found;
98} abandon_ctx;
99
100static void on_abandoned(void *arg, uint32_t pos) {
101 abandon_ctx *a = arg;
102 mesh_driver *d = a->d;
103 for (size_t i = 0; i < d->ndead; i++)
104 if (d->dead[i].consumer == a->consumer && d->dead[i].side == a->side && d->dead[i].pos == pos) return;
105 ddead c = {a->consumer, a->inbox, a->side, a->producer, pos};
106 PUSH(d->dead, d->ndead, d->cdead, c);
107 a->found++;
108}
109
110/* Producer slot s runs no code any more: its pushes in flight are over, and the cells
111 * it reserved and never committed are found for release. */
112void drv_producer_stopped(mesh_driver *d, uint32_t s) {
113 mesh_mailbox mb = d->sh->mb;
114 mesh_words w = mesh_mailbox_words(mb);
115 mesh_pt_flight_clear(mb, s);
116 uint32_t feeds = d->slots[s] ? d->slots[s]->feeds_mask : 0;
117 for (uint32_t c = 0; c < MESH_PROG_SLOTS; c++) {
118 if (!((feeds >> c) & 1u)) continue;
119 mesh_pipe_inbox inbox;
120 if (mesh_pipe_target(&w, c, c, &inbox) != 0) continue;
121 abandon_ctx a = {d, c, 0, s, inbox, 0};
122 const uint32_t sides[2] = {MESH_PIPE_SIDE_CPU, MESH_PIPE_SIDE_GPU};
123 for (int i = 0; i < 2; i++) {
124 a.side = sides[i];
125 mesh_pipe_ring r = mesh_pipe_inbox_side(&inbox, sides[i]);
126 mesh_pipe_ring_abandoned(&r, &w, s, on_abandoned, &a);
127 }
128 if (a.found > 0)
129 mesh_error("pipes: slot %u stopped with %zu cell(s) of slot %u's inbox reserved and not committed: released when the consumer reaches them",
130 s, a.found, c);
131 }
132}
133
134/* Spins until no producer in fed has a push in flight that began before this
135 * call, or limit passed; returns the producers not seen quiet (0: quiet). Two
136 * rounds, as RCU's grace period. */
137uint32_t drv_pipe_quiet(mesh_driver *d, uint32_t fed, int64_t limit_ns) {
138 mesh_mailbox mb = d->sh->mb;
139 if (!fed) return 0;
140 int64_t t0 = mesh_mono_ns();
141 for (int round = 0; round < 2; round++) {
142 uint32_t parity = mesh_pt_flight_epoch(mb), left = fed;
143 for (;;) {
144 for (uint32_t x = 0; x < MESH_PROG_SLOTS; x++)
145 if (((left >> x) & 1u) && mesh_pt_flight(mb, x, parity) == 0) left &= ~(1u << x);
146 if (!left) break;
147 if (mesh_mono_ns() - t0 > limit_ns) return left;
148 }
149 }
150 return 0;
151}
152
153/* Connects the pipes: a producer's `feeds` entry reaches a loaded consumer whose
154 * inbox has that type. Every slot's pipe block is rewritten in place. */
155void drv_connect_pipes(mesh_driver *d) {
156 mesh_mailbox mb = d->sh->mb;
157 uint32_t feeds[MESH_PROG_SLOTS] = {0}, fed[MESH_PROG_SLOTS] = {0};
158 for (uint32_t p = 0; p < MESH_PROG_SLOTS; p++) {
159 dslot *x = d->slots[p];
160 if (!x || x->unloading) continue;
161 for (size_t f = 0; f < x->image.manifest.nfeeds; f++) {
162 const mesh_manifest_feed *fd = &x->image.manifest.feeds[f];
163 for (uint32_t c = 0; c < MESH_PROG_SLOTS; c++) {
164 dslot *y = d->slots[c];
165 if (y && !y->unloading && strcmp(drv_name(y), fd->consumer) == 0 && y->image.manifest.inbox &&
166 strcmp(y->image.manifest.inbox->ty, fd->ty) == 0) {
167 feeds[p] |= 1u << c;
168 fed[c] |= 1u << p;
169 break;
170 }
171 }
172 }
173 }
174 for (uint32_t s = 0; s < MESH_PROG_SLOTS; s++) {
175 dslot *x = d->slots[s];
176 if (!x) continue;
177 if (x->feeds_mask != feeds[s] || x->fed_mask != fed[s]) mesh_info("pipes: %s feeds 0x%x, fed by 0x%x", drv_name(x), feeds[s], fed[s]);
178 x->feeds_mask = feeds[s];
179 x->fed_mask = fed[s];
180 uint8_t pp[MESH_PP_BYTES];
181 drv_pipe(x, pp);
182 mesh_pt_write_pipe(mb, s, pp);
183 }
184}
185
186/* A pipe's two ends must agree on its record type: m against the loaded programs (except: the slot m replaces). */
187bool drv_pipe_check(mesh_driver *d, const mesh_manifest *m, bool has_except, uint32_t except, char **err) {
188 for (uint32_t s = 0; s < MESH_PROG_SLOTS; s++) {
189 dslot *x = d->slots[s];
190 if (!x || (has_except && except == s)) continue;
191 const mesh_manifest *o = &x->image.manifest;
192 for (size_t f = 0; f < m->nfeeds; f++) {
193 const char *c = m->feeds[f].consumer, *ty = m->feeds[f].ty;
194 if (strcmp(c, o->name) != 0) continue;
195 if (o->inbox && strcmp(o->inbox->ty, ty) == 0) continue;
196 if (o->inbox) *err = mesh_buf_fmt("manifest: %s feeds %s records of type %s, but %s's inbox carries %s", m->name, c, ty, c, o->inbox->ty);
197 else *err = mesh_buf_fmt("manifest: %s feeds %s, which has no inbox", m->name, c);
198 return false;
199 }
200 if (m->inbox) {
201 for (size_t f = 0; f < o->nfeeds; f++) {
202 if (strcmp(o->feeds[f].consumer, m->name) == 0 && strcmp(o->feeds[f].ty, m->inbox->ty) != 0) {
203 *err = mesh_buf_fmt("manifest: %s feeds %s records of type %s, but this inbox carries %s", o->name, m->name, o->feeds[f].ty, m->inbox->ty);
204 return false;
205 }
206 }
207 }
208 }
209 return true;
210}
211
212/* The exchange page (every system's counter and slots) zeroed: only while the lockstep program is halted. */
213void drv_zero_exchange(mesh_driver *d) {
214 mesh_mailbox mb = d->sh->mb;
215 for (uint32_t k = 0; k < MESH_MAX_SYSTEMS; k++) {
216 mesh_mailbox_zero(mb, mesh_ex_slots(k), MESH_EX_REGION_BYTES);
217 mesh_mailbox_publish64(mb, mesh_ex_counter(k), 0);
218 }
219}
220
221/* ---------------------------------------------------------------- the http program's loads */
222
223static void http_load(mesh_driver *d) {
224 mesh_mailbox mb = d->sh->mb;
225 if (d->gpu_load) {
226 mesh_result r;
227 int got = mesh_reply_poll(d->gpu_load_rx, &r);
228 if (got == MESH_CHAN_EMPTY) return;
229 if (got != MESH_CHAN_OK) { r = mesh_result_err("load: dropped"); }
230 uint32_t s = d->gpu_load_slot, seq = d->gpu_load_seq;
231 mesh_reply_release(d->gpu_load_rx);
232 d->gpu_load = false;
233 d->gpu_load_rx = NULL;
234 /* Still the same http program? Then answer it: 0 ok, else 1 + the reason's index. */
235 dslot *x = d->slots[s];
236 if (x && strcmp(drv_name(x), "http") == 0) {
237 uint32_t code = MESH_HTTP_LOAD_OK;
238 if (!r.ok) {
239 code = 1 + 8;
240 for (uint32_t i = 0; i < MESH_LOAD_REASONS_LEN; i++) {
241 size_t k = strlen(MESH_LOAD_REASONS[i]);
242 if (r.text && strncmp(r.text, MESH_LOAD_REASONS[i], k) == 0) { code = 1 + i; break; }
243 }
244 }
245 mesh_mailbox_set(mb, (size_t)x->view_off + MESH_HTTP_HV_LOAD_RESULT, code);
246 mesh_mailbox_publish(mb, (size_t)x->view_off + MESH_HTTP_HV_LOAD_ACK, seq);
247 }
248 mesh_result_free(&r);
249 return;
250 }
251 int s = drv_find(d, "http");
252 if (s < 0) return;
253 dslot *x = d->slots[s];
254 if (x->image.manifest.view_bytes < MESH_HTTP_HV_LOAD_NAME + 16) return;
255 size_t v = x->view_off;
256 uint32_t seq = mesh_mailbox_get(mb, v + MESH_HTTP_HV_LOAD_SEQ);
257 if (seq == mesh_mailbox_get(mb, v + MESH_HTTP_HV_LOAD_ACK)) return;
258 uint8_t b[17];
259 mesh_mailbox_read(mb, v + MESH_HTTP_HV_LOAD_NAME, b, 16);
260 b[16] = 0;
261 mesh_buf name = MESH_BUF_INIT;
262 mesh_rs_lossy(&name, b, strlen((char *)b));
263 mesh_info("http: GET /load/%s", mesh_buf_str(&name));
264 mesh_reply *rx = mesh_reply_new();
265 /* In a fleet a load is a change of the desired state (the leader commits it, every member loads it). */
266 if (d->sh->peer) {
267 mesh_peer_event e = mesh_peer_event_new(MESH_PEV_CHANGE);
268 e.line = mesh_buf_fmt("load %s", mesh_buf_str(&name));
269 e.reply = mesh_reply_sender(rx);
270 if (mesh_chan_send(d->sh->peer, &e) != MESH_CHAN_OK) mesh_peer_event_free(&e);
271 } else {
272 djob j = {false, mesh_src_catalog(mesh_buf_str(&name)), -1, 0, mesh_reply_sender(rx)};
273 drv_push_job(d, j);
274 }
275 mesh_buf_free(&name);
276 d->gpu_load = true;
277 d->gpu_load_slot = (uint32_t)s;
278 d->gpu_load_seq = seq;
279 d->gpu_load_rx = rx;
280}
281
282/* P_RUN_REQ changed: a GPU program (http) asked to start/stop slot s. */
283static void run_request(mesh_driver *d, uint32_t s) {
284 mesh_mailbox mb = d->sh->mb;
285 uint32_t req = mesh_mailbox_get(mb, ctl(s, MESH_P_RUN_REQ));
286 dslot *x = d->slots[s];
287 if (req >> 1 == x->run_seq >> 1) return;
288 x->run_seq = req;
289 bool on = (req & 1u) && !x->unloading;
290 bool want = x->want_run;
291 /* In a fleet the run state is the desired state's: the request goes through the leader. */
292 if (d->sh->peer) {
293 if (on != want) {
294 mesh_info("run request: %s %s (a change of the desired state)", on ? "start" : "stop", drv_name(x));
295 mesh_peer_event e = mesh_peer_event_new(MESH_PEV_CHANGE);
296 e.line = mesh_buf_fmt("%s %s", on ? "start" : "stop", drv_name(x));
297 if (mesh_chan_send(d->sh->peer, &e) != MESH_CHAN_OK) mesh_peer_event_free(&e);
298 }
299 return;
300 }
301 mesh_mailbox_set(mb, ctl(s, MESH_P_RUN), on);
302 if (on != x->want_run) mesh_info("run request: %s %s", on ? "start" : "stop", drv_name(x));
303 if (on) { free(x->failed); x->failed = NULL; }
304 x->want_run = on;
305 d->state_dirty = true;
306}
307
308/* ---------------------------------------------------------------- completions */
309
310/* The lockstep program's progress for viewers: its next arg (u64 with carry) and view prefix, under one lock. */
311static void lock_progress(mesh_driver *d, uint32_t s) {
312 mesh_mailbox mb = d->sh->mb;
313 dslot *x = d->slots[s];
314 const mesh_manifest *m = &x->image.manifest;
315 uint32_t lo = m->arg_mode == MESH_ARG_VIEW_U32 ? mesh_mailbox_get(mb, (size_t)x->view_off + m->arg_off) : mesh_mailbox_get(mb, ctl(s, MESH_P_DONE));
316 if (lo < d->lock_lo) d->lock_hi++;
317 d->lock_lo = lo;
318 uint64_t step = (uint64_t)d->lock_hi << 32 | lo;
319 uint8_t sb[8], view[32], name[MESH_NAME_BYTES];
320 for (int i = 0; i < 8; i++) sb[i] = (uint8_t)(step >> (8 * i));
321 memset(view, 0, sizeof view);
322 mesh_mailbox_read(mb, x->view_off, view, m->view_bytes < 32 ? m->view_bytes : 32);
323 memset(name, 0, sizeof name);
324 memcpy(name, m->name, strlen(m->name) < MESH_NAME_BYTES ? strlen(m->name) : MESH_NAME_BYTES);
325 mesh_host_field f[3] = {{MESH_S_LOCK_STEP, sb, 8}, {MESH_S_LOCK_VIEW, view, 32}, {MESH_S_LOCK_NAME, name, MESH_NAME_BYTES}};
326 mesh_host_block_set_many(&d->sh->host, f, 3);
327}
328
329void drv_collect(mesh_driver *d, uint32_t s) {
330 mesh_mailbox mb = d->sh->mb;
331 dslot *x = d->slots[s];
332 if (!drv_live(x)) return;
333 mesh_done *done;
334 size_t n;
335 mesh_pring_collect(&x->ring, mb, &done, &n);
336 bool lockstep = x->image.manifest.peer == MESH_PEER_MODE_LOCKSTEP;
337 for (size_t i = 0; i < n; i++) {
338 mesh_out *o = &done[i].out;
339 uint32_t r = done[i].result;
340 x = d->slots[s];
341 switch (o->kind.tk) {
342 case MESH_TK_AUTO:
343 if (lockstep) lock_progress(d, s);
344 break;
345 case MESH_TK_FOLLOW: break;
346 case MESH_TK_INIT: mesh_info("%s: OP_INIT done", drv_name(x)); break;
347 case MESH_TK_JOB: {
348 djobslot *j = &d->jobw[s];
349 if (j->inflight) j->inflight--;
350 if (r == MESH_R_JOB_FAIL) {
351 char *why = mesh_buf_fmt("%s: item %llu: result 0x%x (R_JOB_FAIL)", drv_name(x), (unsigned long long)o->kind.item, r);
352 drv_job_done(d, o->kind.lease, o->kind.job, o->kind.item, MESH_JOBS_RS_FAIL, r, why);
353 free(why);
354 } else {
355 const mesh_manifest *m = &x->image.manifest;
356 mesh_job_event e = mesh_job_event_new(MESH_JEV_DONE);
357 e.res.lease = o->kind.lease;
358 e.res.job = o->kind.job;
359 e.res.item = o->kind.item;
360 e.res.status = MESH_JOBS_RS_OK;
361 e.res.result = r;
362 /* The item's job_out record, at its completion slot (read before the slot's next ticket is published). */
363 size_t slot = o->ticket % MESH_PROG_RESULTS;
364 if (m->has_job_out) {
365 e.res.out_len = m->job_out_rec;
366 e.res.out = mesh_xmalloc(m->job_out_rec ? m->job_out_rec : 1);
367 mesh_mailbox_read(mb, (size_t)x->view_off + m->job_out_off + (size_t)m->job_out_rec * slot, e.res.out, m->job_out_rec);
368 }
369 /* Its payload (`job_payload`), at the same slot. */
370 if (m->has_job_payload) {
371 e.has_payload = true;
372 e.payload_len = m->job_payload_len;
373 e.payload = mesh_xmalloc(m->job_payload_len ? m->job_payload_len : 1);
374 mesh_mailbox_read(mb, (size_t)x->view_off + m->job_payload_off + (size_t)m->job_payload_len * slot, e.payload, m->job_payload_len);
375 }
376 if (mesh_chan_send(d->sh->jobs, &e) != MESH_CHAN_OK) mesh_job_event_free(&e);
377 }
378 break;
379 }
380 case MESH_TK_JOB_SETUP: {
381 char *err = NULL;
382 if (r == MESH_R_BAD_OP) err = mesh_buf_fmt("%s: setup task %u %u -> R_BAD_OP", drv_name(x), o->kind.op, o->kind.arg);
383 mesh_setup_state_completed(&d->jobw[s].setup, o->kind.job, err);
384 free(err);
385 break;
386 }
387 case MESH_TK_TASK: mesh_reply_value(o->kind.reply, r); o->kind.reply = NULL; break;
388 case MESH_TK_REQ: {
389 const mesh_manifest *m = &x->image.manifest;
390 if (mesh_mailbox_get64(mb, ctl(s, MESH_P_RESP_SEQ)) == o->kind.seq) {
391 uint32_t len = mesh_mailbox_get(mb, ctl(s, MESH_P_RESP_LEN));
392 if (len > m->request.resp_max) len = m->request.resp_max;
393 uint8_t *buf = mesh_xmalloc(len ? len : 1);
394 mesh_mailbox_read(mb, (size_t)x->view_off + m->request.resp_off, buf, len);
395 mesh_reply_bytes(o->kind.reply, buf, len);
396 free(buf);
397 } else {
398 mesh_reply_errf(o->kind.reply, "%s: OP_REQ completed without a response to request %llu", drv_name(x), (unsigned long long)o->kind.seq);
399 }
400 o->kind.reply = NULL;
401 break;
402 }
403 }
404 }
405 free(done);
406}
407
408/* ---------------------------------------------------------------- phases */
409
410bool drv_halted(mesh_driver *d, uint32_t s, bool *halted, char **err) {
411 dslot *x = d->slots[s];
412 mesh_backend *ex = drv_be(d, x->image.exec);
413 /* A native worker: halted (or exited, or dead) as its process says. */
414 if (mesh_backend_owns_halt(ex)) return ex->ops->query(ex, x->stream, x->image.kernel, halted, err);
415 if (mesh_mailbox_get(d->sh->mb, ctl(s, MESH_P_HALTED)) < x->image.manifest.nodes) { *halted = false; return true; }
416 return ex->ops->query(ex, x->stream, x->image.kernel, halted, err);
417}
418
419void drv_resolve(mesh_driver *d, uint32_t s, bool on, bool ok, const char *text) {
420 size_t k = 0;
421 for (size_t i = 0; i < d->nwaiters; i++) {
422 dwaiter *w = &d->waiters[i];
423 if (w->slot != s || w->on != on || w->unload) { d->waiters[k++] = *w; continue; }
424 drv_answer(w->reply, ok, text);
425 }
426 d->nwaiters = k;
427}
428
429void drv_running(mesh_driver *d, uint32_t s, int64_t t0) {
430 dslot *x = d->slots[s];
431 x->phase = PH_RUNNING;
432 x->last_auto = mesh_mono_ns();
433 long long us = (mesh_mono_ns() - t0) / 1000;
434 mesh_info("%s running: %u nodes online in %lld us", drv_name(x), x->image.manifest.nodes, us);
435 char t[32];
436 snprintf(t, sizeof t, "%lld us", us);
437 drv_resolve(d, s, true, true, t);
438}
439
440/* Tickets that will never run: their waiters get an error, an OP_INIT is due again, a job item goes back (or fails). */
441void drv_not_run(mesh_driver *d, uint32_t s, mesh_out *outs, size_t n) {
442 if (!n) { free(outs); return; }
443 dslot *x = d->slots[s];
444 char *why = mesh_buf_fmt("%s stopped before running it", drv_name(x));
445 for (size_t i = 0; i < n; i++) {
446 mesh_kind *k = &outs[i].kind;
447 switch (k->tk) {
448 case MESH_TK_INIT: x->needs_init = true; break;
449 case MESH_TK_TASK:
450 case MESH_TK_REQ: mesh_reply_err(k->reply, why); k->reply = NULL; break;
451 case MESH_TK_JOB:
452 if (x->failed) {
453 char *e = mesh_buf_fmt("the program failed while it ran: %s", x->failed);
454 drv_job_done(d, k->lease, k->job, k->item, MESH_JOBS_RS_FAIL, 0, e);
455 free(e);
456 } else drv_job_done(d, k->lease, k->job, k->item, MESH_JOBS_RS_RETURNED, 0, "the program stopped");
457 break;
458 default: break;
459 }
460 }
461 free(why);
462 free(outs);
463}
464
465void drv_stopped(mesh_driver *d, uint32_t s) {
466 mesh_mailbox mb = d->sh->mb;
467 drv_collect(d, s);
468 drv_producer_stopped(d, s);
469 dslot *x = d->slots[s];
470 x->phase = PH_STOPPED;
471 mesh_out *left;
472 size_t nleft;
473 mesh_pring_rewind(&x->ring, mb, &left, &nleft);
474 drv_not_run(d, s, left, nleft);
475 char *why = mesh_buf_fmt("%s stopped before running it", drv_name(x));
476 bool stop = !x->want_run || x->failed || x->unloading;
477 /* The program halted: its job setup is run again before the next item. */
478 djobslot *j = &d->jobw[s];
479 j->has_ctx = false;
480 j->inflight = 0;
481 mesh_setup_state_reset(&j->setup);
482 if (stop) {
483 size_t nq = j->n;
484 dlease *q = j->q;
485 j->q = NULL;
486 j->n = j->cap = 0;
487 for (size_t i = 0; i < nq; i++) {
488 drv_jobs_returned(d, &q[i], MESH_JOBS_RS_RETURNED, "the program stopped");
489 drv_lease_free(&q[i]);
490 }
491 free(q);
492 }
493 if (!x->want_run || x->failed) {
494 for (size_t i = 0; i < x->ntasks; i++) mesh_reply_err(x->tasks[i].reply, why);
495 x->ntasks = 0;
496 for (size_t i = 0; i < x->nreqs; i++) { mesh_reply_err(x->reqs[i].reply, why); free(x->reqs[i].bytes); }
497 x->nreqs = 0;
498 }
499 free(why);
500 mesh_mailbox_publish(mb, ctl(s, MESH_P_SHUTDOWN), 0);
501 mesh_pt_set_state(mb, s, drv_state(x));
502 if (x->failed) {
503 char *f = mesh_xstrdup(x->failed);
504 drv_resolve(d, s, true, false, f);
505 free(f);
506 }
507 if (x->unloading) drv_remove(d, s);
508 else {
509 for (size_t i = 0; i < d->nwaiters; i++) {
510 dwaiter *w = &d->waiters[i];
511 if (w->slot == s && !w->on && !w->unload) {
512 char t[32];
513 snprintf(t, sizeof t, "%lld us", (long long)((mesh_mono_ns() - w->t0) / 1000));
514 drv_resolve(d, s, false, true, t);
515 break;
516 }
517 }
518 }
519}
520
521void drv_fail(mesh_driver *d, uint32_t s, char *why) {
522 mesh_mailbox mb = d->sh->mb;
523 dslot *x = d->slots[s];
524 mesh_error("%s FAILED: %s", drv_name(x), why);
525 free(x->failed);
526 x->failed = why;
527 switch (x->phase) {
528 case PH_STOPPED: drv_stopped(d, s); break;
529 case PH_HALTING: mesh_pt_set_state(mb, s, MESH_ST_FAILED); break;
530 default:
531 mesh_mailbox_publish(mb, ctl(s, MESH_P_SHUTDOWN), MESH_SHUTDOWN_EXIT);
532 x->phase = PH_HALTING;
533 x->phase_t0 = mesh_mono_ns();
534 mesh_pt_set_state(mb, s, MESH_ST_FAILED);
535 }
536}
537
538bool drv_launch_slot(mesh_driver *d, uint32_t s, char **err) {
539 mesh_mailbox mb = d->sh->mb;
540 dslot *x = d->slots[s];
541 mesh_backend *be = drv_be(d, x->image.exec);
542 bool idle = false;
543 if (!be->ops->query(be, x->stream, x->image.kernel, &idle, err)) return false;
544 if (!idle) { *err = mesh_xstrdup("its previous kernel is still on the stream"); return false; }
545 /* stopped() rewound the ring already: anything outstanding now was never run. */
546 mesh_out *left;
547 size_t nleft;
548 mesh_pring_rewind(&x->ring, mb, &left, &nleft);
549 if (nleft) mesh_error("%s: %zu ticket(s) outstanding at launch (not run)", drv_name(x), nleft);
550 drv_not_run(d, s, left, nleft);
551 /* A relaunch after FAILED starts clean (P_FAULT keeps only the first fault). */
552 mesh_mailbox_set(mb, ctl(s, MESH_P_FAULT), 0);
553 mesh_mailbox_publish(mb, ctl(s, MESH_P_SHUTDOWN), 0);
554 mesh_mailbox_fetch_add(mb, ctl(s, MESH_P_LAUNCHES), 1);
555 if (!be->ops->launch(be, x->stream, x->image.kernel, x->image.manifest.nodes, mb.dev, s, err)) return false;
556 x->phase = PH_STARTING;
557 x->phase_t0 = mesh_mono_ns();
558 bool init = x->needs_init;
559 if (init) {
560 /* First on the ring: every later task sees an initialised view. */
561 x->needs_init = false;
562 mesh_pring_publish(&x->ring, mb, MESH_OP_INIT, 0, mesh_kind_of(MESH_TK_INIT), NULL);
563 }
564 mesh_pt_set_state(mb, s, MESH_ST_RUNNING);
565 /* A fresh view restarts the lockstep program's step. */
566 if (init && x->image.manifest.peer == MESH_PEER_MODE_LOCKSTEP) d->lock_hi = d->lock_lo = 0;
567 return true;
568}
569
570void drv_phase_step(mesh_driver *d, uint32_t s, int64_t now, bool in_blink) {
571 uint32_t exec = drv_exec_of(d, s);
572 bool gpu = exec == drv_gpu_kind(d);
573 mesh_backend *be = drv_be(d, exec);
574 dslot *x = d->slots[s];
575 /* A native worker that died: FAILED with its exit status and output (nothing else stops). */
576 if (drv_live(x)) {
577 char *why = mesh_backend_crashed(be, x->stream, x->image.kernel);
578 if (why) {
579 if (x->failed) { mesh_error("%s: %s", drv_name(x), why); free(why); }
580 else drv_fail(d, s, why);
581 return;
582 }
583 }
584 int64_t online = (int64_t)mesh_backend_online_timeout_ms(be) * 1000000;
585 mesh_mailbox mb = d->sh->mb;
586 uint32_t nodes = x->image.manifest.nodes;
587 char *err = NULL;
588 switch (x->phase) {
589 case PH_STOPPED:
590 /* A lost CUDA context stops only GPU slots. */
591 if (x->want_run && !x->failed && !x->unloading && !in_blink && (!d->fault || !gpu)) {
592 if (!drv_launch_slot(d, s, &err)) {
593 char *why = mesh_buf_fmt("launch: %s", err);
594 free(err);
595 drv_fail(d, s, why);
596 }
597 }
598 break;
599 case PH_STARTING: {
600 uint32_t on = mesh_mailbox_get(mb, ctl(s, MESH_P_ONLINE));
601 if (on >= nodes) drv_running(d, s, x->phase_t0);
602 else if (now - x->phase_t0 > online) {
603 mesh_buf b = MESH_BUF_INIT;
604 mesh_buf_printf(&b, "%u of %u nodes online after ", on, nodes);
605 mesh_rs_duration_ms(&b, (uint64_t)(online / 1000000));
606 drv_fail(d, s, mesh_buf_take(&b));
607 }
608 break;
609 }
610 case PH_RUNNING: {
611 uint32_t f = mesh_mailbox_get(mb, ctl(s, MESH_P_FAULT));
612 if (f != 0 && !x->failed) drv_fail(d, s, mesh_buf_fmt("fault %u (detail 0x%x)", f & 0xffu, f >> 8));
613 else if (!x->want_run || x->unloading) {
614 x->phase = PH_DRAINING;
615 x->phase_t0 = now;
616 mesh_pt_set_state(mb, s, MESH_ST_STOPPING);
617 }
618 break;
619 }
620 case PH_DRAINING:
621 if (mesh_pring_idle(&x->ring) || now - x->phase_t0 > DRAIN_NS) {
622 if (!mesh_pring_idle(&x->ring))
623 mesh_error("%s: %zu task(s) did not drain in 3s; stopping anyway", drv_name(x), mesh_pring_in_flight(&x->ring));
624 mesh_mailbox_publish(mb, ctl(s, MESH_P_SHUTDOWN), MESH_SHUTDOWN_EXIT);
625 x->phase = PH_HALTING;
626 x->phase_t0 = now;
627 }
628 break;
629 case PH_HALTING: {
630 bool failed = x->failed != NULL, h = false;
631 if (!drv_halted(d, s, &h, &err)) { drv_context_fault(d, err); free(err); break; }
632 if (h) drv_stopped(d, s);
633 else if (now - x->phase_t0 > HALT_NS && !failed) {
634 uint32_t hn = mesh_mailbox_get(mb, ctl(s, MESH_P_HALTED));
635 drv_fail(d, s, mesh_buf_fmt("%u of %u nodes halted in 3s", hn, nodes));
636 }
637 break;
638 }
639 }
640}
641
642/* ---------------------------------------------------------------- unload */
643
644void drv_remove(mesh_driver *d, uint32_t s) {
645 dslot *x = d->slots[s];
646 d->slots[s] = NULL;
647 bool lockstep = x->image.manifest.peer == MESH_PEER_MODE_LOCKSTEP;
648 uint32_t exec = x->image.exec;
649 char *name = mesh_xstrdup(drv_name(x));
650 uint32_t view_off = x->view_off, view_bytes = x->image.manifest.view_bytes;
651 bool inbox = x->image.manifest.inbox != NULL;
652 dzombie z = {NULL, view_off, view_bytes, exec, inbox, x->fed_mask, mesh_mono_ns(), false};
653 if (exec == MESH_BACKEND_NATIVE) {
654 /* A native image is no module: its processes exit now; only the view waits for a blink. */
655 mesh_backend *ex = drv_be(d, exec);
656 char *e = NULL;
657 if (!ex->ops->unload(ex, x->image.kernel, &e)) free(e);
658 x->image.kernel = NULL;
659 if (x->has_prev) { if (!ex->ops->unload(ex, x->prev.kernel, &e)) free(e); x->prev.kernel = NULL; }
660 mesh_slot_use *u = &mesh_cpuuse_slots[s];
661 atomic_store_explicit(&u->rss, 0, memory_order_relaxed);
662 atomic_store_explicit(&u->units, 0, memory_order_relaxed);
663 d->native_cpu[s].key = 0;
664 d->native_cpu[s].cpu = 0;
665 PUSH(d->zombies, d->nzombies, d->czombies, z);
666 } else {
667 z.kernel = x->image.kernel;
668 x->image.kernel = NULL;
669 PUSH(d->zombies, d->nzombies, d->czombies, z);
670 if (x->has_prev) {
671 dzombie p = {x->prev.kernel, 0, 0, x->prev.exec, false, 0, mesh_mono_ns(), false};
672 x->prev.kernel = NULL;
673 PUSH(d->zombies, d->nzombies, d->czombies, p);
674 }
675 }
676 char *gone = mesh_buf_fmt("%s was unloaded", name);
677 for (size_t i = 0; i < x->ntasks; i++) mesh_reply_err(x->tasks[i].reply, gone);
678 djobslot j = d->jobw[s];
679 memset(&d->jobw[s], 0, sizeof d->jobw[s]);
680 for (size_t i = 0; i < j.n; i++) { drv_jobs_returned(d, &j.q[i], MESH_JOBS_RS_RETURNED, "unloaded"); drv_lease_free(&j.q[i]); }
681 free(j.q);
682 mesh_setup_state_reset(&j.setup);
683 for (size_t i = 0; i < x->nreqs; i++) { mesh_reply_err(x->reqs[i].reply, gone); free(x->reqs[i].bytes); }
684 free(gone);
685 mesh_pt_clear(d->sh->mb, s);
686 size_t k = 0;
687 for (size_t i = 0; i < d->ndead; i++) if (d->dead[i].consumer != s) d->dead[k++] = d->dead[i];
688 d->ndead = k;
689 drv_connect_pipes(d);
690 if (lockstep) drv_zero_exchange(d);
691 d->state_dirty = true;
692 mesh_info("%s unloaded from slot %u (module and %zu view pages freed at the next blink)", name, s, mesh_view_pages(view_bytes));
693 /* The slot's own resources. */
694 drv_image_free(&x->image);
695 if (x->has_prev) drv_image_free(&x->prev);
696 mesh_stream_drop(x->stream);
697 mesh_pring_free(&x->ring);
698 free(x->tasks);
699 free(x->reqs);
700 free(x->failed);
701 free(x);
702 mesh_reply *now = NULL;
703 mesh_reply *done[64];
704 size_t ndone = 0;
705 k = 0;
706 for (size_t i = 0; i < d->nwaiters; i++) {
707 dwaiter *w = &d->waiters[i];
708 if (w->slot != s) { d->waiters[k++] = *w; continue; }
709 if (w->unload && w->now) now = w->reply;
710 else if (w->unload && ndone < 64) done[ndone++] = w->reply;
711 else mesh_reply_errf(w->reply, "%s was unloaded", name);
712 }
713 d->nwaiters = k;
714 for (size_t i = 0; i < ndone; i++) mesh_reply_ok(done[i], "");
715 if (now) {
716 uint32_t scope = 0;
717 mesh_backend *be = drv_be(d, exec);
718 bool has = be->ops->module_scope(be, false, 0, &scope);
719 dblink b = {OP_FREE, NULL, -1, 0, now, has, scope, exec, mesh_mono_ns()};
720 drv_push_blink(d, b);
721 }
722 free(name);
723}
724
725/* ---------------------------------------------------------------- publishing */
726
727static void publish_slot(mesh_driver *d, uint32_t s, int64_t now) {
728 mesh_mailbox mb = d->sh->mb;
729 /* In a fleet the lockstep program is fed only while its group's epoch runs here. */
730 bool lock_ok = !d->cfg.peered || atomic_load(&d->sh->lock_feed);
731 dslot *x = d->slots[s];
732 if (mesh_pring_has_init(&x->ring)) return;
733 size_t t = 0;
734 while (t < x->ntasks && mesh_pring_can_publish(&x->ring)) {
735 mesh_kind k = mesh_kind_of(MESH_TK_TASK);
736 k.reply = x->tasks[t].reply;
737 if (!mesh_pring_publish(&x->ring, mb, x->tasks[t].op, x->tasks[t].arg, k, NULL)) break;
738 t++;
739 }
740 if (t) { memmove(x->tasks, x->tasks + t, (x->ntasks - t) * sizeof *x->tasks); x->ntasks -= t; }
741 if (!mesh_pring_has_req(&x->ring) && mesh_pring_can_publish(&x->ring) && x->nreqs) {
742 dreq r = x->reqs[0];
743 memmove(x->reqs, x->reqs + 1, (x->nreqs - 1) * sizeof *x->reqs);
744 x->nreqs--;
745 const mesh_manifest_window *w = &x->image.manifest.request;
746 /* Word writes: never past req_max (checked at the request). */
747 size_t span = (r.len + 3) / 4 * 4;
748 if (span > w->req_max) abort();
749 mesh_mailbox_write(mb, (size_t)x->view_off + w->req_off, r.bytes, r.len, span);
750 mesh_mailbox_set(mb, ctl(s, MESH_P_REQ_LEN), (uint32_t)r.len);
751 uint64_t seq = mesh_mailbox_get64(mb, ctl(s, MESH_P_REQ_SEQ)) + 1;
752 mesh_mailbox_publish64(mb, ctl(s, MESH_P_REQ_SEQ), seq);
753 mesh_kind k = mesh_kind_of(MESH_TK_REQ);
754 k.seq = seq;
755 k.reply = r.reply;
756 mesh_pring_publish(&x->ring, mb, MESH_OP_REQ, 0, k, NULL);
757 free(r.bytes);
758 }
759 if (x->image.manifest.policy == MESH_POLICY_JOB && x->want_run && x->phase == PH_RUNNING) { drv_publish_jobs(d, s); return; }
760 const mesh_manifest *m = &x->image.manifest;
761 if (x->want_run && (m->peer != MESH_PEER_MODE_LOCKSTEP || lock_ok))
762 mesh_publish_auto(mb, m, x->view_off, &x->ring, &x->last_auto, now);
763}
764
765void drv_programs_tick(mesh_driver *d, int64_t now) {
766 bool blink = d->nblinks > 0;
767 bool has_scope = blink && d->blinks[0].has_scope;
768 uint32_t scope = blink ? d->blinks[0].scope : 0;
769 for (uint32_t s = 0; s < MESH_PROG_SLOTS; s++) {
770 if (!d->slots[s]) continue;
771 run_request(d, s);
772 drv_collect(d, s);
773 bool in_blink = blink && drv_in_scope(d, has_scope, scope, s);
774 drv_phase_step(d, s, now, in_blink);
775 drv_native_words(d, s);
776 dslot *x = d->slots[s];
777 if (!x) continue;
778 if (!in_blink && (x->phase == PH_RUNNING || x->phase == PH_STARTING) && !x->failed) publish_slot(d, s, now);
779 }
780 http_load(d);
781 dead_cells_tick(d);
782}