==> 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 if (on != want) {
292 char *why = drv_placed_here(d, s);
293 if (why) {
294 mesh_info("run request ignored: %s", why);
295 free(why);
296 return;
297 }
298 }
299 /* In a fleet the run state is the desired state's: the request goes through the leader. */
300 if (d->sh->peer) {
301 if (on != want) {
302 mesh_info("run request: %s %s (a change of the desired state)", on ? "start" : "stop", drv_name(x));
303 mesh_peer_event e = mesh_peer_event_new(MESH_PEV_CHANGE);
304 e.line = mesh_buf_fmt("%s %s", on ? "start" : "stop", drv_name(x));
305 if (mesh_chan_send(d->sh->peer, &e) != MESH_CHAN_OK) mesh_peer_event_free(&e);
306 }
307 return;
308 }
309 mesh_mailbox_set(mb, ctl(s, MESH_P_RUN), on);
310 if (on != x->want_run) mesh_info("run request: %s %s", on ? "start" : "stop", drv_name(x));
311 if (on) { free(x->failed); x->failed = NULL; }
312 x->want_run = on;
313 d->state_dirty = true;
314}
315
316/* ---------------------------------------------------------------- completions */
317
318/* The lockstep program's progress for viewers: its next arg (u64 with carry) and view prefix, under one lock. */
319static void lock_progress(mesh_driver *d, uint32_t s) {
320 mesh_mailbox mb = d->sh->mb;
321 dslot *x = d->slots[s];
322 const mesh_manifest *m = &x->image.manifest;
323 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));
324 if (lo < d->lock_lo) d->lock_hi++;
325 d->lock_lo = lo;
326 uint64_t step = (uint64_t)d->lock_hi << 32 | lo;
327 uint8_t sb[8], view[32], name[MESH_NAME_BYTES];
328 for (int i = 0; i < 8; i++) sb[i] = (uint8_t)(step >> (8 * i));
329 memset(view, 0, sizeof view);
330 mesh_mailbox_read(mb, x->view_off, view, m->view_bytes < 32 ? m->view_bytes : 32);
331 memset(name, 0, sizeof name);
332 memcpy(name, m->name, strlen(m->name) < MESH_NAME_BYTES ? strlen(m->name) : MESH_NAME_BYTES);
333 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}};
334 mesh_host_block_set_many(&d->sh->host, f, 3);
335}
336
337void drv_collect(mesh_driver *d, uint32_t s) {
338 mesh_mailbox mb = d->sh->mb;
339 dslot *x = d->slots[s];
340 if (!drv_live(x)) return;
341 mesh_done *done;
342 size_t n;
343 mesh_pring_collect(&x->ring, mb, &done, &n);
344 bool lockstep = x->image.manifest.peer == MESH_PEER_MODE_LOCKSTEP;
345 bool own = false, follow = false;
346 for (size_t i = 0; i < n; i++) {
347 mesh_out *o = &done[i].out;
348 uint32_t r = done[i].result;
349 x = d->slots[s];
350 switch (o->kind.tk) {
351 case MESH_TK_AUTO:
352 own = true;
353 if (lockstep) lock_progress(d, s);
354 break;
355 case MESH_TK_FOLLOW: follow = true; break;
356 case MESH_TK_INIT: mesh_info("%s: OP_INIT done", drv_name(x)); break;
357 case MESH_TK_JOB: {
358 djobslot *j = &d->jobw[s];
359 if (j->inflight) j->inflight--;
360 if (r == MESH_R_JOB_FAIL) {
361 char *why = mesh_buf_fmt("%s: item %llu: result 0x%x (R_JOB_FAIL)", drv_name(x), (unsigned long long)o->kind.item, r);
362 drv_job_done(d, o->kind.lease, o->kind.job, o->kind.item, MESH_JOBS_RS_FAIL, r, why);
363 free(why);
364 } else {
365 const mesh_manifest *m = &x->image.manifest;
366 mesh_job_event e = mesh_job_event_new(MESH_JEV_DONE);
367 e.res.lease = o->kind.lease;
368 e.res.job = o->kind.job;
369 e.res.item = o->kind.item;
370 e.res.status = MESH_JOBS_RS_OK;
371 e.res.result = r;
372 /* The item's job_out record, at its completion slot (read before the slot's next ticket is published). */
373 size_t slot = o->ticket % MESH_PROG_RESULTS;
374 if (m->has_job_out) {
375 e.res.out_len = m->job_out_rec;
376 e.res.out = mesh_xmalloc(m->job_out_rec ? m->job_out_rec : 1);
377 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);
378 }
379 /* Its payload (`job_payload`), at the same slot. */
380 if (m->has_job_payload) {
381 e.has_payload = true;
382 e.payload_len = m->job_payload_len;
383 e.payload = mesh_xmalloc(m->job_payload_len ? m->job_payload_len : 1);
384 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);
385 }
386 if (mesh_chan_send(d->sh->jobs, &e) != MESH_CHAN_OK) mesh_job_event_free(&e);
387 }
388 break;
389 }
390 case MESH_TK_JOB_SETUP: {
391 char *err = NULL;
392 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);
393 mesh_setup_state_completed(&d->jobw[s].setup, o->kind.job, err);
394 free(err);
395 break;
396 }
397 case MESH_TK_TASK: mesh_reply_value(o->kind.reply, r); o->kind.reply = NULL; break;
398 case MESH_TK_REQ: {
399 const mesh_manifest *m = &x->image.manifest;
400 if (mesh_mailbox_get64(mb, ctl(s, MESH_P_RESP_SEQ)) == o->kind.seq) {
401 uint32_t len = mesh_mailbox_get(mb, ctl(s, MESH_P_RESP_LEN));
402 if (len > m->request.resp_max) len = m->request.resp_max;
403 uint8_t *buf = mesh_xmalloc(len ? len : 1);
404 mesh_mailbox_read(mb, (size_t)x->view_off + m->request.resp_off, buf, len);
405 mesh_reply_bytes(o->kind.reply, buf, len);
406 free(buf);
407 } else {
408 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);
409 }
410 o->kind.reply = NULL;
411 break;
412 }
413 }
414 }
415 free(done);
416 if ((own || follow) && d->place[s]) drv_place_frame(d, s, own, follow);
417}
418
419/* ---------------------------------------------------------------- phases */
420
421bool drv_halted(mesh_driver *d, uint32_t s, bool *halted, char **err) {
422 dslot *x = d->slots[s];
423 mesh_backend *ex = drv_be(d, x->image.exec);
424 /* A native worker: halted (or exited, or dead) as its process says. */
425 if (mesh_backend_owns_halt(ex)) return ex->ops->query(ex, x->stream, x->image.kernel, halted, err);
426 if (mesh_mailbox_get(d->sh->mb, ctl(s, MESH_P_HALTED)) < x->image.manifest.nodes) { *halted = false; return true; }
427 return ex->ops->query(ex, x->stream, x->image.kernel, halted, err);
428}
429
430void drv_resolve(mesh_driver *d, uint32_t s, bool on, bool ok, const char *text) {
431 size_t k = 0;
432 for (size_t i = 0; i < d->nwaiters; i++) {
433 dwaiter *w = &d->waiters[i];
434 if (w->slot != s || w->on != on || w->unload) { d->waiters[k++] = *w; continue; }
435 drv_answer(w->reply, ok, text);
436 }
437 d->nwaiters = k;
438}
439
440void drv_running(mesh_driver *d, uint32_t s, int64_t t0) {
441 dslot *x = d->slots[s];
442 x->phase = PH_RUNNING;
443 x->last_auto = mesh_mono_ns();
444 long long us = (mesh_mono_ns() - t0) / 1000;
445 mesh_info("%s running: %u nodes online in %lld us", drv_name(x), x->image.manifest.nodes, us);
446 char t[32];
447 snprintf(t, sizeof t, "%lld us", us);
448 drv_resolve(d, s, true, true, t);
449}
450
451/* Tickets that will never run: their waiters get an error, an OP_INIT is due again, a job item goes back (or fails). */
452void drv_not_run(mesh_driver *d, uint32_t s, mesh_out *outs, size_t n) {
453 if (!n) { free(outs); return; }
454 dslot *x = d->slots[s];
455 char *why = mesh_buf_fmt("%s stopped before running it", drv_name(x));
456 for (size_t i = 0; i < n; i++) {
457 mesh_kind *k = &outs[i].kind;
458 switch (k->tk) {
459 case MESH_TK_INIT: x->needs_init = true; break;
460 case MESH_TK_FOLLOW: drv_place_follow_lost(d, s); break;
461 case MESH_TK_TASK:
462 case MESH_TK_REQ: mesh_reply_err(k->reply, why); k->reply = NULL; break;
463 case MESH_TK_JOB:
464 if (x->failed) {
465 char *e = mesh_buf_fmt("the program failed while it ran: %s", x->failed);
466 drv_job_done(d, k->lease, k->job, k->item, MESH_JOBS_RS_FAIL, 0, e);
467 free(e);
468 } else drv_job_done(d, k->lease, k->job, k->item, MESH_JOBS_RS_RETURNED, 0, "the program stopped");
469 break;
470 default: break;
471 }
472 }
473 free(why);
474 free(outs);
475}
476
477void drv_stopped(mesh_driver *d, uint32_t s) {
478 mesh_mailbox mb = d->sh->mb;
479 drv_collect(d, s);
480 drv_producer_stopped(d, s);
481 dslot *x = d->slots[s];
482 x->phase = PH_STOPPED;
483 mesh_out *left;
484 size_t nleft;
485 mesh_pring_rewind(&x->ring, mb, &left, &nleft);
486 drv_not_run(d, s, left, nleft);
487 char *why = mesh_buf_fmt("%s stopped before running it", drv_name(x));
488 bool stop = !x->want_run || x->failed || x->unloading;
489 /* The program halted: its job setup is run again before the next item. */
490 djobslot *j = &d->jobw[s];
491 j->has_ctx = false;
492 j->inflight = 0;
493 mesh_setup_state_reset(&j->setup);
494 if (stop) {
495 size_t nq = j->n;
496 dlease *q = j->q;
497 j->q = NULL;
498 j->n = j->cap = 0;
499 for (size_t i = 0; i < nq; i++) {
500 drv_jobs_returned(d, &q[i], MESH_JOBS_RS_RETURNED, "the program stopped");
501 drv_lease_free(&q[i]);
502 }
503 free(q);
504 }
505 if (!x->want_run || x->failed) {
506 for (size_t i = 0; i < x->ntasks; i++) mesh_reply_err(x->tasks[i].reply, why);
507 x->ntasks = 0;
508 for (size_t i = 0; i < x->nreqs; i++) { mesh_reply_err(x->reqs[i].reply, why); free(x->reqs[i].bytes); }
509 x->nreqs = 0;
510 }
511 free(why);
512 mesh_mailbox_publish(mb, ctl(s, MESH_P_SHUTDOWN), 0);
513 mesh_pt_set_state(mb, s, drv_state(x));
514 if (x->failed) {
515 char *f = mesh_xstrdup(x->failed);
516 drv_resolve(d, s, true, false, f);
517 free(f);
518 }
519 if (x->unloading) drv_remove(d, s);
520 else {
521 for (size_t i = 0; i < d->nwaiters; i++) {
522 dwaiter *w = &d->waiters[i];
523 if (w->slot == s && !w->on && !w->unload) {
524 char t[32];
525 snprintf(t, sizeof t, "%lld us", (long long)((mesh_mono_ns() - w->t0) / 1000));
526 drv_resolve(d, s, false, true, t);
527 break;
528 }
529 }
530 }
531}
532
533void drv_fail(mesh_driver *d, uint32_t s, char *why) {
534 mesh_mailbox mb = d->sh->mb;
535 dslot *x = d->slots[s];
536 mesh_error("%s FAILED: %s", drv_name(x), why);
537 free(x->failed);
538 x->failed = why;
539 switch (x->phase) {
540 case PH_STOPPED: drv_stopped(d, s); break;
541 case PH_HALTING: mesh_pt_set_state(mb, s, MESH_ST_FAILED); break;
542 default:
543 mesh_mailbox_publish(mb, ctl(s, MESH_P_SHUTDOWN), MESH_SHUTDOWN_EXIT);
544 x->phase = PH_HALTING;
545 x->phase_t0 = mesh_mono_ns();
546 mesh_pt_set_state(mb, s, MESH_ST_FAILED);
547 }
548}
549
550bool drv_launch_slot(mesh_driver *d, uint32_t s, char **err) {
551 mesh_mailbox mb = d->sh->mb;
552 dslot *x = d->slots[s];
553 mesh_backend *be = drv_be(d, x->image.exec);
554 bool idle = false;
555 if (!be->ops->query(be, x->stream, x->image.kernel, &idle, err)) return false;
556 if (!idle) { *err = mesh_xstrdup("its previous kernel is still on the stream"); return false; }
557 /* stopped() rewound the ring already: anything outstanding now was never run. */
558 mesh_out *left;
559 size_t nleft;
560 mesh_pring_rewind(&x->ring, mb, &left, &nleft);
561 if (nleft) mesh_error("%s: %zu ticket(s) outstanding at launch (not run)", drv_name(x), nleft);
562 drv_not_run(d, s, left, nleft);
563 /* A relaunch after FAILED starts clean (P_FAULT keeps only the first fault). */
564 mesh_mailbox_set(mb, ctl(s, MESH_P_FAULT), 0);
565 mesh_mailbox_publish(mb, ctl(s, MESH_P_SHUTDOWN), 0);
566 mesh_mailbox_fetch_add(mb, ctl(s, MESH_P_LAUNCHES), 1);
567 if (!be->ops->launch(be, x->stream, x->image.kernel, x->image.manifest.nodes, mb.dev, s, err)) return false;
568 x->phase = PH_STARTING;
569 x->phase_t0 = mesh_mono_ns();
570 bool init = x->needs_init;
571 if (init) {
572 /* First on the ring: every later task sees an initialised view. */
573 x->needs_init = false;
574 mesh_pring_publish(&x->ring, mb, MESH_OP_INIT, 0, mesh_kind_of(MESH_TK_INIT), NULL);
575 }
576 mesh_pt_set_state(mb, s, MESH_ST_RUNNING);
577 /* A fresh view restarts the lockstep program's step. */
578 if (init && x->image.manifest.peer == MESH_PEER_MODE_LOCKSTEP) d->lock_hi = d->lock_lo = 0;
579 return true;
580}
581
582void drv_phase_step(mesh_driver *d, uint32_t s, int64_t now, bool in_blink) {
583 uint32_t exec = drv_exec_of(d, s);
584 bool gpu = exec == drv_gpu_kind(d);
585 mesh_backend *be = drv_be(d, exec);
586 dslot *x = d->slots[s];
587 /* A native worker that died: FAILED with its exit status and output (nothing else stops). */
588 if (drv_live(x)) {
589 char *why = mesh_backend_crashed(be, x->stream, x->image.kernel);
590 if (why) {
591 if (x->failed) { mesh_error("%s: %s", drv_name(x), why); free(why); }
592 else drv_fail(d, s, why);
593 return;
594 }
595 }
596 int64_t online = (int64_t)mesh_backend_online_timeout_ms(be) * 1000000;
597 mesh_mailbox mb = d->sh->mb;
598 uint32_t nodes = x->image.manifest.nodes;
599 char *err = NULL;
600 switch (x->phase) {
601 case PH_STOPPED:
602 /* A lost CUDA context stops only GPU slots. */
603 if (x->want_run && !x->failed && !x->unloading && !in_blink && (!d->fault || !gpu)) {
604 if (!drv_launch_slot(d, s, &err)) {
605 char *why = mesh_buf_fmt("launch: %s", err);
606 free(err);
607 drv_fail(d, s, why);
608 }
609 }
610 break;
611 case PH_STARTING: {
612 uint32_t on = mesh_mailbox_get(mb, ctl(s, MESH_P_ONLINE));
613 if (on >= nodes) drv_running(d, s, x->phase_t0);
614 else if (now - x->phase_t0 > online) {
615 mesh_buf b = MESH_BUF_INIT;
616 mesh_buf_printf(&b, "%u of %u nodes online after ", on, nodes);
617 mesh_rs_duration_ms(&b, (uint64_t)(online / 1000000));
618 drv_fail(d, s, mesh_buf_take(&b));
619 }
620 break;
621 }
622 case PH_RUNNING: {
623 uint32_t f = mesh_mailbox_get(mb, ctl(s, MESH_P_FAULT));
624 if (f != 0 && !x->failed) drv_fail(d, s, mesh_buf_fmt("fault %u (detail 0x%x)", f & 0xffu, f >> 8));
625 else if (!x->want_run || x->unloading) {
626 x->phase = PH_DRAINING;
627 x->phase_t0 = now;
628 mesh_pt_set_state(mb, s, MESH_ST_STOPPING);
629 }
630 break;
631 }
632 case PH_DRAINING:
633 if (mesh_pring_idle(&x->ring) || now - x->phase_t0 > DRAIN_NS) {
634 if (!mesh_pring_idle(&x->ring))
635 mesh_error("%s: %zu task(s) did not drain in 3s; stopping anyway", drv_name(x), mesh_pring_in_flight(&x->ring));
636 mesh_mailbox_publish(mb, ctl(s, MESH_P_SHUTDOWN), MESH_SHUTDOWN_EXIT);
637 x->phase = PH_HALTING;
638 x->phase_t0 = now;
639 }
640 break;
641 case PH_HALTING: {
642 bool failed = x->failed != NULL, h = false;
643 if (!drv_halted(d, s, &h, &err)) { drv_context_fault(d, err); free(err); break; }
644 if (h) drv_stopped(d, s);
645 else if (now - x->phase_t0 > HALT_NS && !failed) {
646 uint32_t hn = mesh_mailbox_get(mb, ctl(s, MESH_P_HALTED));
647 drv_fail(d, s, mesh_buf_fmt("%u of %u nodes halted in 3s", hn, nodes));
648 }
649 break;
650 }
651 }
652}
653
654/* ---------------------------------------------------------------- unload */
655
656void drv_remove(mesh_driver *d, uint32_t s) {
657 dslot *x = d->slots[s];
658 d->slots[s] = NULL;
659 bool lockstep = x->image.manifest.peer == MESH_PEER_MODE_LOCKSTEP;
660 uint32_t exec = x->image.exec;
661 char *name = mesh_xstrdup(drv_name(x));
662 uint32_t view_off = x->view_off, view_bytes = x->image.manifest.view_bytes;
663 bool inbox = x->image.manifest.inbox != NULL;
664 dzombie z = {NULL, view_off, view_bytes, exec, inbox, x->fed_mask, mesh_mono_ns(), false};
665 if (exec == MESH_BACKEND_NATIVE) {
666 /* A native image is no module: its processes exit now; only the view waits for a blink. */
667 mesh_backend *ex = drv_be(d, exec);
668 char *e = NULL;
669 if (!ex->ops->unload(ex, x->image.kernel, &e)) free(e);
670 x->image.kernel = NULL;
671 if (x->has_prev) { if (!ex->ops->unload(ex, x->prev.kernel, &e)) free(e); x->prev.kernel = NULL; }
672 mesh_slot_use *u = &mesh_cpuuse_slots[s];
673 atomic_store_explicit(&u->rss, 0, memory_order_relaxed);
674 atomic_store_explicit(&u->units, 0, memory_order_relaxed);
675 d->native_cpu[s].key = 0;
676 d->native_cpu[s].cpu = 0;
677 PUSH(d->zombies, d->nzombies, d->czombies, z);
678 } else {
679 z.kernel = x->image.kernel;
680 x->image.kernel = NULL;
681 PUSH(d->zombies, d->nzombies, d->czombies, z);
682 if (x->has_prev) {
683 dzombie p = {x->prev.kernel, 0, 0, x->prev.exec, false, 0, mesh_mono_ns(), false};
684 x->prev.kernel = NULL;
685 PUSH(d->zombies, d->nzombies, d->czombies, p);
686 }
687 }
688 char *gone = mesh_buf_fmt("%s was unloaded", name);
689 for (size_t i = 0; i < x->ntasks; i++) mesh_reply_err(x->tasks[i].reply, gone);
690 djobslot j = d->jobw[s];
691 memset(&d->jobw[s], 0, sizeof d->jobw[s]);
692 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]); }
693 free(j.q);
694 mesh_setup_state_reset(&j.setup);
695 for (size_t i = 0; i < x->nreqs; i++) { mesh_reply_err(x->reqs[i].reply, gone); free(x->reqs[i].bytes); }
696 free(gone);
697 mesh_pt_clear(d->sh->mb, s);
698 size_t k = 0;
699 for (size_t i = 0; i < d->ndead; i++) if (d->dead[i].consumer != s) d->dead[k++] = d->dead[i];
700 d->ndead = k;
701 drv_connect_pipes(d);
702 if (lockstep) drv_zero_exchange(d);
703 drv_place_drop(d, s, name);
704 d->state_dirty = true;
705 mesh_info("%s unloaded from slot %u (module and %zu view pages freed at the next blink)", name, s, mesh_view_pages(view_bytes));
706 /* The slot's own resources. */
707 drv_image_free(&x->image);
708 if (x->has_prev) drv_image_free(&x->prev);
709 mesh_stream_drop(x->stream);
710 mesh_pring_free(&x->ring);
711 free(x->tasks);
712 free(x->reqs);
713 free(x->failed);
714 free(x);
715 mesh_reply *now = NULL;
716 mesh_reply *done[64];
717 size_t ndone = 0;
718 k = 0;
719 for (size_t i = 0; i < d->nwaiters; i++) {
720 dwaiter *w = &d->waiters[i];
721 if (w->slot != s) { d->waiters[k++] = *w; continue; }
722 if (w->unload && w->now) now = w->reply;
723 else if (w->unload && ndone < 64) done[ndone++] = w->reply;
724 else mesh_reply_errf(w->reply, "%s was unloaded", name);
725 }
726 d->nwaiters = k;
727 for (size_t i = 0; i < ndone; i++) mesh_reply_ok(done[i], "");
728 if (now) {
729 uint32_t scope = 0;
730 mesh_backend *be = drv_be(d, exec);
731 bool has = be->ops->module_scope(be, false, 0, &scope);
732 dblink b = {OP_FREE, NULL, -1, 0, now, has, scope, exec, mesh_mono_ns()};
733 drv_push_blink(d, b);
734 }
735 free(name);
736}
737
738/* ---------------------------------------------------------------- publishing */
739
740static void publish_slot(mesh_driver *d, uint32_t s, int64_t now) {
741 mesh_mailbox mb = d->sh->mb;
742 /* In a fleet the lockstep program is fed only while its group's epoch runs here. */
743 bool lock_ok = !d->cfg.peered || atomic_load(&d->sh->lock_feed);
744 dslot *x = d->slots[s];
745 if (mesh_pring_has_init(&x->ring)) return;
746 size_t t = 0;
747 while (t < x->ntasks && mesh_pring_can_publish(&x->ring)) {
748 mesh_kind k = mesh_kind_of(MESH_TK_TASK);
749 k.reply = x->tasks[t].reply;
750 if (!mesh_pring_publish(&x->ring, mb, x->tasks[t].op, x->tasks[t].arg, k, NULL)) break;
751 t++;
752 }
753 if (t) { memmove(x->tasks, x->tasks + t, (x->ntasks - t) * sizeof *x->tasks); x->ntasks -= t; }
754 if (!mesh_pring_has_req(&x->ring) && mesh_pring_can_publish(&x->ring) && x->nreqs) {
755 dreq r = x->reqs[0];
756 memmove(x->reqs, x->reqs + 1, (x->nreqs - 1) * sizeof *x->reqs);
757 x->nreqs--;
758 const mesh_manifest_window *w = &x->image.manifest.request;
759 /* Word writes: never past req_max (checked at the request). */
760 size_t span = (r.len + 3) / 4 * 4;
761 if (span > w->req_max) abort();
762 mesh_mailbox_write(mb, (size_t)x->view_off + w->req_off, r.bytes, r.len, span);
763 mesh_mailbox_set(mb, ctl(s, MESH_P_REQ_LEN), (uint32_t)r.len);
764 uint64_t seq = mesh_mailbox_get64(mb, ctl(s, MESH_P_REQ_SEQ)) + 1;
765 mesh_mailbox_publish64(mb, ctl(s, MESH_P_REQ_SEQ), seq);
766 mesh_kind k = mesh_kind_of(MESH_TK_REQ);
767 k.seq = seq;
768 k.reply = r.reply;
769 mesh_pring_publish(&x->ring, mb, MESH_OP_REQ, 0, k, NULL);
770 free(r.bytes);
771 }
772 if (x->image.manifest.policy == MESH_POLICY_JOB && x->want_run && x->phase == PH_RUNNING) { drv_publish_jobs(d, s); return; }
773 const mesh_manifest *m = &x->image.manifest;
774 /* Placement: the follow batch (split, a follower's parts) on the owner's state, written just before. */
775 bool paused;
776 mpm_role role;
777 uint32_t parts;
778 drv_place_publish(d, s, &paused, &role, &parts);
779 if (x->want_run && !paused && (m->peer != MESH_PEER_MODE_LOCKSTEP || lock_ok)) {
780 switch (role) {
781 case MPM_ROLE_LOCAL:
782 case MPM_ROLE_RUNNER: mesh_publish_auto(mb, m, x->view_off, &x->ring, &x->last_auto, now); break;
783 case MPM_ROLE_OWNER: mesh_publish_parts(mb, m, x->view_off, &x->ring, &x->last_auto, now, parts); break;
784 case MPM_ROLE_FOLLOWER:
785 case MPM_ROLE_MIRROR: break;
786 }
787 }
788}
789
790void drv_programs_tick(mesh_driver *d, int64_t now) {
791 bool blink = d->nblinks > 0;
792 bool has_scope = blink && d->blinks[0].has_scope;
793 uint32_t scope = blink ? d->blinks[0].scope : 0;
794 for (uint32_t s = 0; s < MESH_PROG_SLOTS; s++) {
795 if (!d->slots[s]) continue;
796 run_request(d, s);
797 drv_collect(d, s);
798 bool in_blink = blink && drv_in_scope(d, has_scope, scope, s);
799 drv_phase_step(d, s, now, in_blink);
800 drv_native_words(d, s);
801 dslot *x = d->slots[s];
802 if (!x) continue;
803 if (!in_blink && (x->phase == PH_RUNNING || x->phase == PH_STARTING) && !x->failed) publish_slot(d, s, now);
804 }
805 drv_place_tick(d);
806 http_load(d);
807 dead_cells_tick(d);
808}