==> src/blink.c <==1/* Loads and upgrades: images compiled on a JIT thread, then blinks (halt what the change touches, the
2 * module operations, relaunch); rollback, reset; the table rewritten; the occupancy budget. */
3#include "driver_int.h"
4
5#include "mesh/native.h"
6#include "mesh/occupancy.h"
7#include "mesh/rsfmt.h"
8
9#include <pthread.h>
10#include <stdio.h>
11#include <stdlib.h>
12#include <string.h>
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 double ms(int64_t ns) { return (double)ns / 1e6; }
24
25/* What an executor compiles a program to (for logs). */
26static const char *code_word(uint32_t exec) {
27 switch (exec) {
28 case MESH_BACKEND_METAL: return "MSL";
29 case MESH_BACKEND_WASM: return "wasm";
30 case MESH_BACKEND_NATIVE: return "native binary";
31 default: return "cubin";
32 }
33}
34
35static bool module_scope(mesh_driver *d, uint32_t exec, bool has_slot, uint32_t slot, uint32_t *scope) {
36 mesh_backend *be = drv_be(d, exec);
37 return be->ops->module_scope(be, has_slot, slot, scope);
38}
39
40/* ---------------------------------------------------------------- the JIT thread */
41
42typedef struct {
43 mesh_compiler_entry comps[8];
44 size_t ncomps;
45 char *root;
46 mesh_src src;
47 int warm_run; /* -1: the manifest's autostart */
48 mesh_reply *tx;
49} jit_job;
50
51static void prepared_free_heap(void *p) {
52 if (!p) return;
53 mesh_prepared_free(p);
54 free(p);
55}
56
57static void *jit(void *arg) {
58 jit_job *j = arg;
59 mesh_payload p;
60 char *err = NULL;
61 bool ok;
62 mesh_prepared *out = mesh_xcalloc(1, sizeof *out);
63 if (!j->src.is_bytes) {
64 ok = mesh_loader_read(j->root, j->src.dir, false, &p, &err) && mesh_prepare_any(j->comps, j->ncomps, &p, NULL, j->root, out, &err);
65 } else {
66 mesh_payload_copy(&p, &j->src.payload);
67 ok = mesh_prepare_any(j->comps, j->ncomps, &p, &j->src.hash, j->root, out, &err);
68 }
69 /* A native image that will run at once gets its process started here, off the driver thread. */
70 if (ok && out->kernel && mesh_kernel_native(out->kernel)) {
71 bool run = j->warm_run < 0 ? out->manifest.autostart : j->warm_run == 1;
72 if (run && !mesh_native_worker_warm(mesh_kernel_native(out->kernel), &err)) { mesh_prepared_free(out); ok = false; }
73 }
74 if (ok) mesh_reply_ptr(j->tx, out, prepared_free_heap);
75 else {
76 free(out);
77 mesh_reply_err(j->tx, err ? err : "jit: failed");
78 }
79 free(err);
80 free(j->root);
81 mesh_src_free(&j->src);
82 free(j);
83 return NULL;
84}
85
86/* The job a load or upgrade is (for log lines). */
87static char *what(const mesh_src *s) { return mesh_src_what(s); }
88
89bool drv_precheck(mesh_driver *d, const mesh_manifest *m, char **err) {
90 if (drv_find(d, m->name) >= 0) { *err = mesh_buf_fmt("name exists: %s is loaded", m->name); return false; }
91 if (!drv_pipe_check(d, m, false, 0, err)) return false;
92 /* The exchange page has one owner. */
93 if (m->peer == MESH_PEER_MODE_LOCKSTEP) {
94 for (uint32_t s = 0; s < MESH_PROG_SLOTS; s++) {
95 dslot *x = d->slots[s];
96 if (x && x->image.manifest.peer == MESH_PEER_MODE_LOCKSTEP) {
97 *err = mesh_buf_fmt("manifest: %s is the lockstep program already; only one may be loaded", drv_name(x));
98 return false;
99 }
100 }
101 }
102 bool full = true;
103 for (uint32_t s = 0; s < MESH_PROG_SLOTS; s++) full &= d->slots[s] != NULL;
104 if (full) { *err = mesh_buf_fmt("no slot: all %d slots are loaded", MESH_PROG_SLOTS); return false; }
105 mesh_view_heap h = d->heap;
106 for (size_t i = 0; i < d->nzombies; i++) mesh_view_heap_free(&h, d->zombies[i].view_off, d->zombies[i].view_bytes);
107 uint32_t off;
108 if (!mesh_view_heap_alloc(&h, m->view_bytes, &off)) {
109 *err = mesh_buf_fmt("view heap: %zu pages wanted, %u free (fragmented)", mesh_view_pages(m->view_bytes), mesh_view_heap_free_pages(&h));
110 return false;
111 }
112 return true;
113}
114
115void drv_jobs_tick(mesh_driver *d) {
116 if (!d->working && !d->nblinks && d->njobs) {
117 djob job = d->jobs[0];
118 memmove(d->jobs, d->jobs + 1, (d->njobs - 1) * sizeof *d->jobs);
119 d->njobs--;
120 if (d->fault) {
121 mesh_reply_errf(job.reply, "the CUDA context is lost (%s); restart meshd", d->fault);
122 mesh_src_free(&job.src);
123 return;
124 }
125 jit_job *j = mesh_xcalloc(1, sizeof *j);
126 mesh_backend *all[9];
127 size_t nall = 0;
128 all[nall++] = d->gpu;
129 for (size_t i = 0; i < d->ncpus && nall < 8; i++) all[nall++] = d->cpus[i];
130 for (size_t i = 0; i < nall; i++) {
131 j->comps[i].kind = all[i]->ops->caps(all[i])->kind;
132 j->comps[i].comp = all[i]->ops->compiler(all[i]);
133 }
134 j->ncomps = nall;
135 j->root = mesh_xstrdup(d->sh->programs);
136 j->src = mesh_src_copy(&job.src);
137 j->warm_run = job.upgrade ? 1 : job.run;
138 mesh_reply *rx = mesh_reply_new();
139 j->tx = mesh_reply_sender(rx);
140 pthread_t th;
141 pthread_attr_t a;
142 pthread_attr_init(&a);
143 pthread_attr_setdetachstate(&a, PTHREAD_CREATE_DETACHED);
144 if (pthread_create(&th, &a, jit, j) != 0) {
145 mesh_reply_err(j->tx, "jit: the worker thread died");
146 free(j->root);
147 mesh_src_free(&j->src);
148 free(j);
149 }
150 pthread_attr_destroy(&a);
151 d->working = true;
152 d->work_rx = rx;
153 d->work_job = job;
154 }
155 if (!d->working) return;
156 mesh_result r;
157 int got = mesh_reply_poll(d->work_rx, &r);
158 if (got == MESH_CHAN_EMPTY) return;
159 if (got != MESH_CHAN_OK) r = mesh_result_err("jit: the worker thread died");
160 mesh_reply_release(d->work_rx);
161 d->working = false;
162 d->work_rx = NULL;
163 djob job = d->work_job;
164 int64_t since = mesh_mono_ns();
165 char *w = what(&job.src);
166 if (!r.ok) {
167 if (job.upgrade) mesh_error("upgrade from %s: %s", w, r.text);
168 else {
169 mesh_error("load %s: %s", w, r.text);
170 char *e = mesh_buf_fmt("%s: %s", w, r.text);
171 drv_startup_failed(d, job.reply, e);
172 free(e);
173 }
174 mesh_reply_err(job.reply, r.text);
175 } else {
176 mesh_prepared *p = r.ptr;
177 r.ptr = NULL;
178 if (!job.upgrade) {
179 char *e = NULL;
180 /* Refusals that need no module are answered without a blink. */
181 if (!drv_precheck(d, &p->manifest, &e)) {
182 mesh_error("load %s: %s", mesh_prepared_dir(p), e);
183 char *f = mesh_buf_fmt("%s: %s", mesh_prepared_dir(p), e);
184 drv_startup_failed(d, job.reply, f);
185 free(f);
186 mesh_reply_err(job.reply, e);
187 free(e);
188 mesh_prepared_free(p);
189 free(p);
190 } else {
191 mesh_buf l = MESH_BUF_INIT;
192 mesh_buf_printf(&l, "load %s: JIT ", mesh_prepared_dir(p));
193 mesh_rs_fixed(&l, p->compile_ms, 2);
194 mesh_buf_printf(&l, " ms, %zu bytes of %s (%s); blinking", p->code_len, code_word(p->exec), MESH_BACKENDS[p->exec]);
195 mesh_info("%s", mesh_buf_str(&l));
196 mesh_buf_free(&l);
197 uint32_t scope = 0;
198 bool has = module_scope(d, p->exec, false, 0, &scope);
199 dblink b = {OP_LOAD, p, job.run, 0, job.reply, has, scope, p->exec, since};
200 drv_push_blink(d, b);
201 }
202 } else {
203 mesh_buf l = MESH_BUF_INIT;
204 mesh_buf_printf(&l, "upgrade %s from %s: JIT ", p->manifest.name, mesh_prepared_dir(p));
205 mesh_rs_fixed(&l, p->compile_ms, 2);
206 mesh_buf_puts(&l, " ms; blinking");
207 mesh_info("%s", mesh_buf_str(&l));
208 mesh_buf_free(&l);
209 /* The slot's executor decides the scope (an image for another one is refused in the blink). */
210 uint32_t exec = drv_exec_of(d, job.slot), scope = 0;
211 bool has = module_scope(d, exec, true, job.slot, &scope);
212 dblink b = {OP_UPGRADE, p, -1, job.slot, job.reply, has, scope, exec, since};
213 drv_push_blink(d, b);
214 }
215 }
216 free(w);
217 mesh_src_free(&job.src);
218 mesh_result_free(&r);
219}
220
221/* ---------------------------------------------------------------- the blink */
222
223/* mps: the scope's kernel halts across ticks; true once halted, or overdue (do_blink then fails it). */
224static bool halt_scope(mesh_driver *d, bool has, uint32_t scope) {
225 if (!has || scope == MESH_NO_SLOT) return true;
226 if (!d->slots[scope] || !drv_live(d->slots[scope])) return true;
227 if (!d->halting) {
228 mesh_mailbox_publish(d->sh->mb, mesh_prog_ctl(scope) + MESH_P_SHUTDOWN, MESH_SHUTDOWN_HALT);
229 d->halting = true;
230 d->halting_t = mesh_mono_ns();
231 }
232 bool h = true;
233 char *e = NULL;
234 if (!drv_halted(d, scope, &h, &e)) { free(e); h = true; }
235 return h || mesh_mono_ns() - d->halting_t > HALT_NS;
236}
237
238static bool size_stack(mesh_driver *d, char **err) {
239 uint32_t need = 0;
240 for (uint32_t s = 0; s < MESH_PROG_SLOTS; s++) {
241 dslot *x = d->slots[s];
242 if (!x) continue;
243 if (x->image.local > need) need = x->image.local;
244 if (x->has_prev && x->prev.local > need) need = x->prev.local;
245 }
246 size_t have;
247 if (!d->gpu->ops->stack_limit(d->gpu, &have, err)) return false;
248 if (need > have) {
249 int64_t t0 = mesh_mono_ns();
250 if (!d->gpu->ops->set_stack_limit(d->gpu, need, err)) return false;
251 size_t now;
252 if (!d->gpu->ops->stack_limit(d->gpu, &now, err)) return false;
253 mesh_buf l = MESH_BUF_INIT;
254 mesh_buf_printf(&l, "context stack %zu -> %zu bytes per thread (", have, now);
255 mesh_rs_fixed(&l, ms(mesh_mono_ns() - t0), 2);
256 mesh_buf_puts(&l, " ms)");
257 mesh_info("%s", mesh_buf_str(&l));
258 mesh_buf_free(&l);
259 }
260 return true;
261}
262
263/* Every halted slot's entry rewritten, reads masks re-resolved by name; the running ones whose mask changed are returned. */
264static size_t rewrite_entries(mesh_driver *d, uint32_t *rebind) {
265 const char *names[MESH_PROG_SLOTS];
266 for (uint32_t s = 0; s < MESH_PROG_SLOTS; s++) names[s] = d->slots[s] ? drv_name(d->slots[s]) : NULL;
267 mesh_mailbox mb = d->sh->mb;
268 size_t n = 0;
269 for (uint32_t s = 0; s < MESH_PROG_SLOTS; s++) {
270 dslot *x = d->slots[s];
271 if (!x) continue;
272 uint32_t mask = mesh_reads_mask(&x->image.manifest, names, MESH_PROG_SLOTS);
273 uint8_t e[MESH_PT_BYTES], ext[MESH_PX_BYTES];
274 if (drv_live(x) && mesh_backend_reads_live(drv_be(d, x->image.exec))) {
275 /* wasm lanes and native workers read the mask under the seqlock at every peer view: rewritten in place. */
276 if (mask != x->reads_mask) {
277 x->reads_mask = mask;
278 drv_entry(x, e);
279 drv_ext(x, ext);
280 mesh_pt_write_entry(mb, s, e, ext, drv_state(x));
281 }
282 continue;
283 }
284 if (drv_live(x)) {
285 if (mask != x->reads_mask) rebind[n++] = s;
286 continue;
287 }
288 x->reads_mask = mask;
289 drv_entry(x, e);
290 drv_ext(x, ext);
291 mesh_pt_write_entry(mb, s, e, ext, drv_state(x));
292 }
293 drv_connect_pipes(d);
294 return n;
295}
296
297/* The image's module (blink mode: loaded here) plus the checks that need it. Consumes p. */
298static bool load_image(mesh_driver *d, mesh_prepared *p, dimage *out, char **err) {
299 uint32_t exec = p->exec, reg_cap = p->reg_cap;
300 mesh_backend *be = drv_be(d, exec);
301 mesh_kernel *k = NULL;
302 char *e = NULL;
303 if (!be->ops->load(be, p, &k, &e)) {
304 *err = mesh_buf_fmt("load: %s", e);
305 free(e);
306 mesh_prepared_free(p);
307 return false;
308 }
309 uint32_t rt = 0, regs = 0, per_sm = 0, local = 0;
310 bool ok = true;
311 if (!mesh_kernel_global_u32(k, MESH_RT_ABI_SYMBOL, &rt, &e)) { *err = mesh_buf_fmt("abi: %s", e); ok = false; }
312 else if (rt != MESH_RT_ABI) { *err = mesh_buf_fmt("abi: the image's %s is %u, meshd runs %u", MESH_RT_ABI_SYMBOL, rt, MESH_RT_ABI); ok = false; }
313 else if (!mesh_kernel_regs(k, ®s, &e)) { *err = mesh_buf_fmt("load: %s", e); ok = false; }
314 else if (!mesh_kernel_max_blocks_per_sm(k, &per_sm, &e)) { *err = mesh_buf_fmt("load: %s", e); ok = false; }
315 else if (per_sm == 0) {
316 *err = mesh_buf_fmt("regs: %u registers per thread: not one block of mesh_main (%u threads, %u shared bytes) fits an SM", regs,
317 mesh_manifest_threads(&p->manifest), mesh_manifest_smem(&p->manifest));
318 ok = false;
319 } else if (!mesh_kernel_local_bytes(k, &local, &e)) { *err = mesh_buf_fmt("load: %s", e); ok = false; }
320 free(e);
321 if (!ok) {
322 char *ue = NULL;
323 if (!be->ops->unload(be, k, &ue)) free(ue);
324 mesh_prepared_free(p);
325 return false;
326 }
327 memset(out, 0, sizeof *out);
328 out->manifest = p->manifest;
329 memset(&p->manifest, 0, sizeof p->manifest);
330 out->dir = mesh_xstrdup(p->src.dir);
331 out->kernel = k;
332 out->hash = p->hash;
333 out->code_bytes = (uint32_t)p->code_len;
334 out->regs = regs;
335 out->per_sm = per_sm;
336 out->reg_cap = reg_cap;
337 out->local = local;
338 out->rt_abi = rt;
339 out->exec = exec;
340 mesh_prepared_free(p);
341 return true;
342}
343
344static void unload_image_kernel(mesh_driver *d, dimage *im) {
345 if (!im->kernel) return;
346 char *e = NULL;
347 if (!drv_be(d, im->exec)->ops->unload(drv_be(d, im->exec), im->kernel, &e)) free(e);
348 im->kernel = NULL;
349}
350
351/* Co-residency: every loaded program's nodes must fit at once (backend.rs co_resident); nodes = "fit" decided here. */
352static bool budget(mesh_driver *d, dimage *im, bool has_replace, uint32_t replace, char **err) {
353 if (!drv_be(d, im->exec)->ops->resident(drv_be(d, im->exec))) return true;
354 const mesh_caps *caps = d->gpu->ops->caps(d->gpu);
355 uint32_t sms = caps->sms, bps = caps->blocks_per_sm;
356 mesh_occ_kernel all[MESH_PROG_SLOTS + 1];
357 size_t n = 0;
358 for (uint32_t s = 0; s < MESH_PROG_SLOTS; s++) {
359 dslot *x = d->slots[s];
360 if (!x || (has_replace && replace == s) || !drv_be(d, x->image.exec)->ops->resident(drv_be(d, x->image.exec))) continue;
361 all[n].nodes = x->image.manifest.nodes;
362 all[n].per_sm = x->image.per_sm;
363 n++;
364 }
365 mesh_manifest *m = &im->manifest;
366 if (m->nodes_fit) {
367 /* At most three quarters of the device's capacity for its shape: the rest stays for other programs. */
368 uint32_t q = im->per_sm * sms * 3 / 4;
369 uint32_t max = MESH_PROG_MAX_NODES < (q > 1 ? q : 1) ? MESH_PROG_MAX_NODES : (q > 1 ? q : 1);
370 uint32_t fit = mesh_occ_room(all, n, im->per_sm, max, sms, bps);
371 if (fit == 0) {
372 *err = mesh_buf_fmt("occupancy: %s (nodes = \"fit\"): no room for one block of %u threads beside the loaded kernels", m->name,
373 mesh_manifest_threads(m));
374 return false;
375 }
376 m->nodes = fit;
377 }
378 if (m->nodes > im->per_sm * sms) {
379 *err = mesh_buf_fmt("occupancy: %s needs %u resident blocks of %u threads; its %u regs and %u shared bytes allow %u per SM x %u SMs", m->name,
380 m->nodes, mesh_manifest_threads(m), im->regs, mesh_manifest_smem(m), im->per_sm, sms);
381 return false;
382 }
383 all[n].nodes = m->nodes;
384 all[n].per_sm = im->per_sm;
385 char *e = NULL;
386 if (!mesh_co_resident(all, n + 1, sms, bps, &e)) {
387 *err = mesh_buf_fmt("occupancy: %s: %s", m->name, e);
388 free(e);
389 return false;
390 }
391 return true;
392}
393
394static void slot_free(dslot *x) {
395 drv_image_free(&x->image);
396 if (x->has_prev) drv_image_free(&x->prev);
397 mesh_stream_drop(x->stream);
398 mesh_pring_free(&x->ring);
399 free(x->tasks);
400 free(x->reqs);
401 free(x->failed);
402 free(x);
403}
404
405static bool op_load(mesh_driver *d, mesh_prepared *p, int run, char **out) {
406 char *err = NULL;
407 if (!drv_precheck(d, &p->manifest, &err)) { mesh_prepared_free(p); free(p); *out = err; return false; }
408 uint32_t s = 0;
409 while (d->slots[s]) s++;
410 dimage im;
411 bool ok = load_image(d, p, &im, &err);
412 free(p);
413 if (!ok) { *out = err; return false; }
414 mesh_backend *ex = drv_be(d, im.exec);
415 if (!budget(d, &im, false, 0, &err)) { unload_image_kernel(d, &im); drv_image_free(&im); *out = err; return false; }
416 mesh_stream *stream = NULL;
417 if (!ex->ops->new_stream(ex, s, im.kernel, &stream, &err)) {
418 *out = mesh_buf_fmt("load: %s", err);
419 free(err);
420 unload_image_kernel(d, &im);
421 drv_image_free(&im);
422 return false;
423 }
424 const mesh_manifest *m = &im.manifest;
425 uint32_t view_off = 0;
426 mesh_view_heap_alloc(&d->heap, m->view_bytes, &view_off); /* prechecked */
427 mesh_mailbox mb = d->sh->mb;
428 mesh_pt_reset_slot(mb, s);
429 /* P_RUN_REQ is the GPU's word, never written here: requests count from its value now. */
430 uint32_t run_seq = mesh_mailbox_get(mb, mesh_prog_ctl(s) + MESH_P_RUN_REQ);
431 mesh_mailbox_zero(mb, view_off, m->view_bytes);
432 if (m->peer == MESH_PEER_MODE_LOCKSTEP) drv_zero_exchange(d);
433 bool want_run = run < 0 ? m->autostart : run == 1;
434 mesh_mailbox_set(mb, mesh_prog_ctl(s) + MESH_P_RUN, want_run);
435 if (mesh_kernel_native(im.kernel)) {
436 mesh_info("load %s into slot %u (native): worker %s, %u bytes, hash %016llx, view 0x%x+0x%x, %s", m->name, s,
437 mesh_native_worker_bin(mesh_kernel_native(im.kernel)), im.code_bytes, (unsigned long long)im.hash, view_off, m->view_bytes,
438 want_run ? "starting" : "stopped");
439 } else {
440 char cap[32] = "";
441 if (im.reg_cap > 0) snprintf(cap, sizeof cap, " (cap %u)", im.reg_cap);
442 mesh_info("load %s into slot %u (%s): %u nodes of %u warp(s), %u shared bytes, %u regs%s, %u B local, %u blocks/SM, %u bytes of %s, hash "
443 "%016llx, view 0x%x+0x%x, %s",
444 m->name, s, MESH_BACKENDS[im.exec], m->nodes, m->warps, m->shared_bytes, im.regs, cap, im.local, im.per_sm, im.code_bytes,
445 code_word(im.exec), (unsigned long long)im.hash, view_off, m->view_bytes, want_run ? "starting" : "stopped");
446 }
447 dslot *x = mesh_xcalloc(1, sizeof *x);
448 x->image = im;
449 x->stream = stream;
450 x->view_off = view_off;
451 x->loads = 1;
452 x->phase = PH_STOPPED;
453 x->want_run = want_run;
454 x->run_seq = run_seq;
455 x->needs_init = true;
456 mesh_pring_init(&x->ring, s);
457 x->last_auto = mesh_mono_ns();
458 d->slots[s] = x;
459 *out = mesh_buf_fmt("slot %u", s);
460 return true;
461}
462
463/* Gives slot s a view for `to` (from `from`): kept iff compatible, else a fresh zeroed one and OP_INIT. */
464static bool rebind_view(mesh_driver *d, uint32_t s, const mesh_manifest *from, const mesh_manifest *to, const char **res, char **err) {
465 if (drv_view_compatible(from, to)) { *res = "kept"; return true; }
466 dslot *x = d->slots[s];
467 uint32_t old = x->view_off, off;
468 if (!mesh_view_heap_alloc(&d->heap, to->view_bytes, &off)) {
469 *err = mesh_buf_fmt("view heap: %zu pages wanted, %u free", mesh_view_pages(to->view_bytes), mesh_view_heap_free_pages(&d->heap));
470 return false;
471 }
472 /* The old view is freed at the next module change, not now: a reader (`reads`) may hold a View on it until then. */
473 dzombie z = {NULL, old, from->view_bytes, drv_exec_of(d, s), from->inbox != NULL, x->fed_mask, mesh_mono_ns(), false};
474 PUSH(d->zombies, d->nzombies, d->czombies, z);
475 size_t k = 0;
476 for (size_t i = 0; i < d->ndead; i++) if (d->dead[i].consumer != s) d->dead[k++] = d->dead[i];
477 d->ndead = k;
478 mesh_mailbox_zero(d->sh->mb, off, to->view_bytes);
479 if (to->peer == MESH_PEER_MODE_LOCKSTEP) drv_zero_exchange(d);
480 x->view_off = off;
481 x->needs_init = true;
482 *res = "reset";
483 return true;
484}
485
486static bool op_upgrade(mesh_driver *d, uint32_t s, mesh_prepared *p, char **out) {
487 if (!d->slots[s]) { mesh_prepared_free(p); free(p); *out = mesh_xstrdup("the program was unloaded meanwhile"); return false; }
488 const char *name = drv_name(d->slots[s]);
489 if (strcmp(p->manifest.name, name) != 0) {
490 mesh_buf b = MESH_BUF_INIT;
491 mesh_buf_printf(&b, "manifest: %s names ", mesh_prepared_dir(p));
492 mesh_rs_debug_cstr(&b, p->manifest.name);
493 mesh_buf_puts(&b, ", not ");
494 mesh_rs_debug_cstr(&b, name);
495 *out = mesh_buf_take(&b);
496 mesh_prepared_free(p);
497 free(p);
498 return false;
499 }
500 uint32_t exec = drv_exec_of(d, s);
501 if (p->exec != exec) {
502 *out = mesh_buf_fmt("exec: %s runs on %s here and %s would run on %s (its manifest's `backends`): unload and load it to change executors", name,
503 MESH_BACKENDS[exec], mesh_prepared_dir(p), MESH_BACKENDS[p->exec]);
504 if (p->kernel) {
505 char *e = NULL;
506 if (!drv_be(d, p->exec)->ops->unload(drv_be(d, p->exec), p->kernel, &e)) free(e);
507 p->kernel = NULL;
508 }
509 mesh_prepared_free(p);
510 free(p);
511 return false;
512 }
513 dimage im;
514 char *err = NULL;
515 bool ok = load_image(d, p, &im, &err);
516 free(p);
517 if (!ok) { *out = err; return false; }
518 if (!budget(d, &im, true, s, &err) || !drv_pipe_check(d, &im.manifest, true, s, &err)) {
519 unload_image_kernel(d, &im);
520 drv_image_free(&im);
521 *out = err;
522 return false;
523 }
524 dslot *x = d->slots[s];
525 const char *view;
526 if (!rebind_view(d, s, &x->image.manifest, &im.manifest, &view, &err)) {
527 unload_image_kernel(d, &im);
528 drv_image_free(&im);
529 *out = err;
530 return false;
531 }
532 uint64_t old_hash = x->image.hash, new_hash = im.hash;
533 char *dir = mesh_xstrdup(im.dir);
534 /* One previous image is retained; an older one is unloaded now. */
535 dimage older = x->prev;
536 bool had = x->has_prev;
537 x->prev = x->image;
538 x->has_prev = true;
539 x->image = im;
540 x->loads++;
541 if (had) {
542 unload_image_kernel(d, &older);
543 drv_image_free(&older);
544 }
545 mesh_info("upgrade %s (slot %u) %016llx -> %016llx from %s: view %s", drv_name(x), s, (unsigned long long)old_hash, (unsigned long long)new_hash,
546 dir, view);
547 free(dir);
548 *out = mesh_buf_fmt("view %s", view);
549 return true;
550}
551
552static bool op_rollback(mesh_driver *d, uint32_t s, char **out) {
553 dslot *x = d->slots[s];
554 if (!x) { *out = mesh_xstrdup("the program was unloaded meanwhile"); return false; }
555 if (!x->has_prev) { *out = mesh_buf_fmt("%s: no previous image", drv_name(x)); return false; }
556 dimage cur = x->image;
557 x->image = x->prev;
558 x->prev = cur;
559 const char *view;
560 char *err = NULL;
561 if (!rebind_view(d, s, &x->prev.manifest, &x->image.manifest, &view, &err)) {
562 dimage back = x->image;
563 x->image = x->prev;
564 x->prev = back;
565 *out = err;
566 return false;
567 }
568 const char *names[MESH_PROG_SLOTS];
569 for (uint32_t i = 0; i < MESH_PROG_SLOTS; i++) names[i] = d->slots[i] ? drv_name(d->slots[i]) : NULL;
570 x->loads++;
571 x->reads_mask = mesh_reads_mask(&x->image.manifest, names, MESH_PROG_SLOTS);
572 uint8_t e[MESH_PT_BYTES], ext[MESH_PX_BYTES];
573 drv_entry(x, e);
574 drv_ext(x, ext);
575 mesh_pt_write_entry(d->sh->mb, s, e, ext, drv_state(x));
576 mesh_info("rollback %s (slot %u) to %016llx (%s): view %s", drv_name(x), s, (unsigned long long)x->image.hash, x->image.dir, view);
577 *out = mesh_buf_fmt("view %s", view);
578 return true;
579}
580
581/* Zero the view; OP_INIT at the (next) launch. An inbox in it is held first, and zeroed only once no push into it is in flight. */
582bool drv_op_reset(mesh_driver *d, uint32_t s, char **text) {
583 mesh_mailbox mb = d->sh->mb;
584 dslot *x = d->slots[s];
585 bool inbox = x->image.manifest.inbox != NULL;
586 uint32_t fed = x->fed_mask & ~(1u << s);
587 if (inbox) {
588 mesh_pt_set_hold(mb, s, true);
589 uint32_t busy = drv_pipe_quiet(d, fed, PIPE_DEADLINE_NS);
590 if (busy) {
591 mesh_pt_set_hold(mb, s, false);
592 *text = mesh_buf_fmt("reset %s: producers 0x%x still had a push into its inbox in flight after 500ms; nothing changed", drv_name(x), busy);
593 return false;
594 }
595 }
596 mesh_mailbox_zero(mb, x->view_off, x->image.manifest.view_bytes);
597 size_t k = 0;
598 for (size_t i = 0; i < d->ndead; i++) if (d->dead[i].consumer != s) d->dead[k++] = d->dead[i];
599 d->ndead = k;
600 if (inbox) mesh_pt_set_hold(mb, s, false);
601 x->needs_init = true;
602 if (x->image.manifest.peer == MESH_PEER_MODE_LOCKSTEP) drv_zero_exchange(d);
603 mesh_info("reset %s (slot %u): view zeroed, OP_INIT %s", drv_name(x), s, x->want_run ? "now" : "at the next start");
604 *text = mesh_xstrdup("");
605 return true;
606}
607
608static bool apply(mesh_driver *d, dblink *b, char **out) {
609 switch (b->k) {
610 case OP_LOAD: { mesh_prepared *p = b->p; b->p = NULL; return op_load(d, p, b->run, out); }
611 case OP_UPGRADE: { mesh_prepared *p = b->p; b->p = NULL; return op_upgrade(d, b->slot, p, out); }
612 case OP_ROLLBACK: return op_rollback(d, b->slot, out);
613 case OP_RESET: return drv_op_reset(d, b->slot, out);
614 default: *out = mesh_xstrdup(""); return true;
615 }
616}
617
618static void do_blink(mesh_driver *d, dblink b) {
619 mesh_mailbox mb = d->sh->mb;
620 int64_t drain = mesh_mono_ns() - b.since;
621 bool global = !b.has_scope;
622 /* mps, the CPU executors: a module change halts only its scope but frees and counts like a whole-context blink. */
623 bool iso = drv_be(d, b.exec)->ops->isolated(drv_be(d, b.exec));
624 bool modules = global || (iso && (b.k == OP_LOAD || b.k == OP_UPGRADE || b.k == OP_FREE));
625 bool scope_mask[MESH_PROG_SLOTS], filled[MESH_PROG_SLOTS];
626 for (uint32_t s = 0; s < MESH_PROG_SLOTS; s++) { scope_mask[s] = drv_in_scope(d, b.has_scope, b.scope, s); filled[s] = d->slots[s] != NULL; }
627 int64_t t0 = d->halting ? d->halting_t : mesh_mono_ns();
628 d->halting = false;
629 char *load_dir = b.k == OP_LOAD && b.p ? mesh_xstrdup(mesh_prepared_dir(b.p)) : NULL;
630 /* H_BLINK only now, after the drain: nodes holding no ticket take none while it is set. */
631 if (global) mesh_mailbox_publish(mb, MESH_H_BLINK, 1);
632 /* 1. Halt. */
633 uint32_t live[MESH_PROG_SLOTS], left[MESH_PROG_SLOTS];
634 size_t nlive = 0, nleft = 0;
635 for (uint32_t s = 0; s < MESH_PROG_SLOTS; s++)
636 if (scope_mask[s] && d->slots[s] && drv_live(d->slots[s])) live[nlive++] = s;
637 for (size_t i = 0; i < nlive; i++) { mesh_mailbox_publish(mb, mesh_prog_ctl(live[i]) + MESH_P_SHUTDOWN, MESH_SHUTDOWN_HALT); left[nleft++] = live[i]; }
638 int64_t deadline = t0 + HALT_NS;
639 char *stuck = NULL;
640 while (nleft) {
641 size_t k = 0;
642 char *err = NULL;
643 for (size_t i = 0; i < nleft; i++) {
644 bool h = false;
645 char *e = NULL;
646 if (!drv_halted(d, left[i], &h, &e)) { if (!err) err = e; else free(e); continue; }
647 if (!h) left[k++] = left[i];
648 }
649 nleft = k;
650 if (err) { drv_context_fault(d, err); free(err); }
651 if (mesh_mono_ns() > deadline && nleft) { stuck = mesh_buf_fmt("%zu program kernel(s) did not halt in 3s", nleft); break; }
652 }
653 for (size_t i = 0; i < nlive; i++) {
654 bool still = false;
655 for (size_t j = 0; j < nleft; j++) still |= left[j] == live[i];
656 if (!still && d->slots[live[i]]) drv_stopped(d, live[i]);
657 }
658 for (size_t i = 0; i < nleft; i++) if (d->slots[left[i]]) drv_fail(d, left[i], mesh_xstrdup("did not halt in 3s (blink)"));
659 int64_t t_halt = mesh_mono_ns() - t0;
660 /* 2. Module operations: only when nothing of the context runs. */
661 bool ok;
662 char *result = NULL;
663 if (stuck) {
664 ok = false;
665 result = mesh_buf_fmt("blink: %s; no module changed", stuck);
666 } else if (d->fault) {
667 ok = false;
668 result = mesh_xstrdup("the CUDA context is lost; restart meshd");
669 } else {
670 if (modules) {
671 /* A GPU module unloads only when nothing of its context runs; a CPU one, and views, at any module blink of theirs. */
672 bool gpu_ok = global || d->gpu->ops->isolated(d->gpu);
673 uint32_t gpu_kind = drv_gpu_kind(d);
674 size_t k = 0;
675 for (size_t i = 0; i < d->nzombies; i++) {
676 dzombie z = d->zombies[i];
677 if (z.exec == gpu_kind && !gpu_ok) { d->zombies[k++] = z; continue; }
678 if (z.kernel) {
679 char *e = NULL;
680 if (!drv_be(d, z.exec)->ops->unload(drv_be(d, z.exec), z.kernel, &e)) { mesh_error("module unload: %s", e); free(e); }
681 z.kernel = NULL;
682 }
683 /* A dropped inbox: its pages are reused only once no producer can still be writing into it. */
684 if (z.has_inbox) {
685 uint32_t busy = drv_pipe_quiet(d, z.fed, PIPE_SAMPLE_NS);
686 if (busy) {
687 if (mesh_mono_ns() - z.since > PIPE_DEADLINE_NS && !z.warned) {
688 z.warned = true;
689 mesh_buf l = MESH_BUF_INIT;
690 mesh_buf_printf(&l, "pipes: a dropped inbox's view (%zu pages at 0x%x) is kept: producers 0x%x have had pushes in flight since ",
691 mesh_view_pages(z.view_bytes), z.view_off, busy);
692 mesh_rs_fixed(&l, ms(mesh_mono_ns() - z.since), 0);
693 mesh_buf_puts(&l, " ms");
694 mesh_error("%s", mesh_buf_str(&l));
695 mesh_buf_free(&l);
696 }
697 d->zombies[k++] = z;
698 continue;
699 }
700 }
701 mesh_view_heap_free(&d->heap, z.view_off, z.view_bytes);
702 }
703 d->nzombies = k;
704 }
705 ok = apply(d, &b, &result);
706 if (global || iso) {
707 uint32_t rebind[MESH_PROG_SLOTS];
708 size_t nr = rewrite_entries(d, rebind);
709 /* Only isolated executors leave kernels running across a module change. */
710 for (size_t i = 0; i < nr; i++) {
711 uint32_t s = rebind[i], exec = drv_exec_of(d, s);
712 if (!drv_be(d, exec)->ops->isolated(drv_be(d, exec))) continue;
713 dblink rb = {OP_REBIND, NULL, -1, 0, NULL, true, s, exec, mesh_mono_ns()};
714 drv_push_blink(d, rb);
715 }
716 }
717 char *e = NULL;
718 if (!size_stack(d, &e)) {
719 drv_context_fault(d, e);
720 free(result);
721 result = e;
722 ok = false;
723 }
724 }
725 free(stuck);
726 int64_t t_mod = mesh_mono_ns() - t0;
727 /* 3. Relaunch everything that runs (and a slot the op filled), then wait for all of it. */
728 uint32_t starting[MESH_PROG_SLOTS];
729 size_t nstart = 0;
730 for (uint32_t s = 0; s < MESH_PROG_SLOTS; s++) {
731 dslot *x = d->slots[s];
732 if (!(scope_mask[s] || !filled[s]) || !x || x->phase != PH_STOPPED || !x->want_run || x->failed || x->unloading || d->fault) continue;
733 char *e = NULL;
734 if (drv_launch_slot(d, s, &e)) starting[nstart++] = s;
735 else {
736 char *why = mesh_buf_fmt("launch: %s", e);
737 free(e);
738 drv_fail(d, s, why);
739 }
740 }
741 int64_t t_launch = mesh_mono_ns();
742 /* mps and the CPU executors: no waiting here either; phase_step sees them online (or fails them). */
743 int64_t wait_until = t_launch + (iso ? 0 : ONLINE_NS);
744 for (;;) {
745 bool up = true;
746 for (size_t i = 0; i < nstart; i++)
747 up &= mesh_mailbox_get(mb, mesh_prog_ctl(starting[i]) + MESH_P_ONLINE) >= d->slots[starting[i]]->image.manifest.nodes;
748 if (up || mesh_mono_ns() > wait_until) break;
749 }
750 for (size_t i = 0; i < nstart; i++) {
751 uint32_t s = starting[i];
752 if (!d->slots[s]) continue;
753 uint32_t n = d->slots[s]->image.manifest.nodes, on = mesh_mailbox_get(mb, mesh_prog_ctl(s) + MESH_P_ONLINE);
754 if (on >= n) drv_running(d, s, t_launch);
755 else if (!iso) drv_fail(d, s, mesh_buf_fmt("%u of %u nodes online after 2s (blink)", on, n));
756 }
757 int64_t total = mesh_mono_ns() - t0;
758 if (global) mesh_mailbox_publish(mb, MESH_H_BLINK, 0);
759 if (modules) {
760 d->epoch++;
761 mesh_mailbox_publish(mb, MESH_H_EPOCH, d->epoch);
762 mesh_host_block *h = &d->sh->host;
763 mesh_host_block_set_u32(h, MESH_S_EPOCH, d->epoch);
764 mesh_host_block_set_u32(h, MESH_S_BLINK_US, (uint32_t)(total / 1000));
765 mesh_host_block_set_u64(h, MESH_S_LAST_BLINK_MS, mesh_host_block_uptime_ms(h));
766 mesh_host_block_set_u32(h, MESH_S_HEAP_FREE, mesh_view_heap_free_pages(&d->heap));
767 mesh_host_block_set_u32(h, MESH_S_HEAP_RUN, mesh_view_heap_largest_run(&d->heap));
768 }
769 long long us = total / 1000;
770 mesh_buf l = MESH_BUF_INIT;
771 mesh_buf_puts(&l, "blink");
772 if (modules) mesh_buf_printf(&l, " epoch %u", d->epoch);
773 if (!b.has_scope) mesh_buf_puts(&l, " (whole context)");
774 else if (b.scope == MESH_NO_SLOT) mesh_buf_puts(&l, " (no program halted)");
775 else mesh_buf_printf(&l, " (slot %u)", b.scope);
776 mesh_buf_puts(&l, ": drain ");
777 mesh_rs_fixed(&l, ms(drain), 2);
778 mesh_buf_printf(&l, " ms, halt %lld us, modules %lld us, relaunch %lld us (%zu programs), total %lld us -> %s %s", (long long)(t_halt / 1000),
779 (long long)((t_mod - t_halt) / 1000), (long long)((total - t_mod) / 1000), nstart, us, ok ? "ok" : "err", result);
780 mesh_info("%s", mesh_buf_str(&l));
781 mesh_buf_free(&l);
782 /* Replies: load `slot S blink U`, the others `blink U [view kept|reset]`. */
783 char *reply;
784 if (!ok) reply = mesh_xstrdup(result);
785 else if (!*result) reply = mesh_buf_fmt("blink %lld", us);
786 else if (strncmp(result, "slot ", 5) == 0) reply = mesh_buf_fmt("%s blink %lld", result, us);
787 else reply = mesh_buf_fmt("blink %lld %s", us, result);
788 d->state_dirty = true;
789 if (load_dir && !ok) {
790 char *e = mesh_buf_fmt("%s: %s", load_dir, reply);
791 drv_startup_failed(d, b.reply, e);
792 free(e);
793 }
794 drv_answer(b.reply, ok, reply);
795 free(reply);
796 free(result);
797 free(load_dir);
798 if (b.p) { mesh_prepared_free(b.p); free(b.p); }
799}
800
801void drv_blink_tick(mesh_driver *d) {
802 if (!d->nblinks) return;
803 dblink *b = &d->blinks[0];
804 if (d->fault) {
805 dblink x = *b;
806 memmove(d->blinks, d->blinks + 1, (d->nblinks - 1) * sizeof *d->blinks);
807 d->nblinks--;
808 mesh_reply_err(x.reply, "the CUDA context is lost; restart meshd");
809 if (x.p) { mesh_prepared_free(x.p); free(x.p); }
810 return;
811 }
812 bool iso = drv_be(d, b->exec)->ops->isolated(drv_be(d, b->exec));
813 bool drained = true;
814 for (uint32_t s = 0; s < MESH_PROG_SLOTS && drained; s++)
815 if (drv_in_scope(d, b->has_scope, b->scope, s) && d->slots[s] && !mesh_pring_idle(&d->slots[s]->ring)) drained = false;
816 if (drained) {
817 if (iso && !halt_scope(d, b->has_scope, b->scope)) return;
818 dblink x = *b;
819 memmove(d->blinks, d->blinks + 1, (d->nblinks - 1) * sizeof *d->blinks);
820 d->nblinks--;
821 do_blink(d, x);
822 } else if (mesh_mono_ns() - b->since > DRAIN_NS) {
823 dblink x = *b;
824 memmove(d->blinks, d->blinks + 1, (d->nblinks - 1) * sizeof *d->blinks);
825 d->nblinks--;
826 const char *e = "blink: in-flight tasks did not drain in 3s; nothing changed";
827 mesh_error("%s", e);
828 mesh_reply_err(x.reply, e);
829 if (x.p) { mesh_prepared_free(x.p); free(x.p); }
830 }
831}
832
833/* Keeps slot_free referenced for the unload paths that free a whole slot here. */
834void drv_slot_free(dslot *x);
835void drv_slot_free(dslot *x) { slot_free(x); }