==> 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 d->place[s] = NULL;
460 drv_sync_place(d, s);
461 *out = mesh_buf_fmt("slot %u", s);
462 return true;
463}
464
465/* Gives slot s a view for `to` (from `from`): kept iff compatible, else a fresh zeroed one and OP_INIT. */
466static bool rebind_view(mesh_driver *d, uint32_t s, const mesh_manifest *from, const mesh_manifest *to, const char **res, char **err) {
467 if (drv_view_compatible(from, to)) { *res = "kept"; return true; }
468 dslot *x = d->slots[s];
469 uint32_t old = x->view_off, off;
470 if (!mesh_view_heap_alloc(&d->heap, to->view_bytes, &off)) {
471 *err = mesh_buf_fmt("view heap: %zu pages wanted, %u free", mesh_view_pages(to->view_bytes), mesh_view_heap_free_pages(&d->heap));
472 return false;
473 }
474 /* The old view is freed at the next module change, not now: a reader (`reads`) may hold a View on it until then. */
475 dzombie z = {NULL, old, from->view_bytes, drv_exec_of(d, s), from->inbox != NULL, x->fed_mask, mesh_mono_ns(), false};
476 PUSH(d->zombies, d->nzombies, d->czombies, z);
477 size_t k = 0;
478 for (size_t i = 0; i < d->ndead; i++) if (d->dead[i].consumer != s) d->dead[k++] = d->dead[i];
479 d->ndead = k;
480 mesh_mailbox_zero(d->sh->mb, off, to->view_bytes);
481 if (to->peer == MESH_PEER_MODE_LOCKSTEP) drv_zero_exchange(d);
482 x->view_off = off;
483 x->needs_init = true;
484 *res = "reset";
485 return true;
486}
487
488static bool op_upgrade(mesh_driver *d, uint32_t s, mesh_prepared *p, char **out) {
489 if (!d->slots[s]) { mesh_prepared_free(p); free(p); *out = mesh_xstrdup("the program was unloaded meanwhile"); return false; }
490 const char *name = drv_name(d->slots[s]);
491 if (strcmp(p->manifest.name, name) != 0) {
492 mesh_buf b = MESH_BUF_INIT;
493 mesh_buf_printf(&b, "manifest: %s names ", mesh_prepared_dir(p));
494 mesh_rs_debug_cstr(&b, p->manifest.name);
495 mesh_buf_puts(&b, ", not ");
496 mesh_rs_debug_cstr(&b, name);
497 *out = mesh_buf_take(&b);
498 mesh_prepared_free(p);
499 free(p);
500 return false;
501 }
502 uint32_t exec = drv_exec_of(d, s);
503 if (p->exec != exec) {
504 *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,
505 MESH_BACKENDS[exec], mesh_prepared_dir(p), MESH_BACKENDS[p->exec]);
506 if (p->kernel) {
507 char *e = NULL;
508 if (!drv_be(d, p->exec)->ops->unload(drv_be(d, p->exec), p->kernel, &e)) free(e);
509 p->kernel = NULL;
510 }
511 mesh_prepared_free(p);
512 free(p);
513 return false;
514 }
515 dimage im;
516 char *err = NULL;
517 bool ok = load_image(d, p, &im, &err);
518 free(p);
519 if (!ok) { *out = err; return false; }
520 if (!budget(d, &im, true, s, &err) || !drv_pipe_check(d, &im.manifest, true, s, &err)) {
521 unload_image_kernel(d, &im);
522 drv_image_free(&im);
523 *out = err;
524 return false;
525 }
526 dslot *x = d->slots[s];
527 const char *view;
528 if (!rebind_view(d, s, &x->image.manifest, &im.manifest, &view, &err)) {
529 unload_image_kernel(d, &im);
530 drv_image_free(&im);
531 *out = err;
532 return false;
533 }
534 uint64_t old_hash = x->image.hash, new_hash = im.hash;
535 char *dir = mesh_xstrdup(im.dir);
536 /* One previous image is retained; an older one is unloaded now. */
537 dimage older = x->prev;
538 bool had = x->has_prev;
539 x->prev = x->image;
540 x->has_prev = true;
541 x->image = im;
542 x->loads++;
543 if (had) {
544 unload_image_kernel(d, &older);
545 drv_image_free(&older);
546 }
547 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,
548 dir, view);
549 free(dir);
550 drv_sync_place(d, s);
551 *out = mesh_buf_fmt("view %s", view);
552 return true;
553}
554
555static bool op_rollback(mesh_driver *d, uint32_t s, char **out) {
556 dslot *x = d->slots[s];
557 if (!x) { *out = mesh_xstrdup("the program was unloaded meanwhile"); return false; }
558 if (!x->has_prev) { *out = mesh_buf_fmt("%s: no previous image", drv_name(x)); return false; }
559 dimage cur = x->image;
560 x->image = x->prev;
561 x->prev = cur;
562 const char *view;
563 char *err = NULL;
564 if (!rebind_view(d, s, &x->prev.manifest, &x->image.manifest, &view, &err)) {
565 dimage back = x->image;
566 x->image = x->prev;
567 x->prev = back;
568 *out = err;
569 return false;
570 }
571 const char *names[MESH_PROG_SLOTS];
572 for (uint32_t i = 0; i < MESH_PROG_SLOTS; i++) names[i] = d->slots[i] ? drv_name(d->slots[i]) : NULL;
573 x->loads++;
574 x->reads_mask = mesh_reads_mask(&x->image.manifest, names, MESH_PROG_SLOTS);
575 uint8_t e[MESH_PT_BYTES], ext[MESH_PX_BYTES];
576 drv_entry(x, e);
577 drv_ext(x, ext);
578 mesh_pt_write_entry(d->sh->mb, s, e, ext, drv_state(x));
579 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);
580 drv_sync_place(d, s);
581 *out = mesh_buf_fmt("view %s", view);
582 return true;
583}
584
585/* 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. */
586bool drv_op_reset(mesh_driver *d, uint32_t s, char **text) {
587 mesh_mailbox mb = d->sh->mb;
588 dslot *x = d->slots[s];
589 bool inbox = x->image.manifest.inbox != NULL;
590 uint32_t fed = x->fed_mask & ~(1u << s);
591 if (inbox) {
592 mesh_pt_set_hold(mb, s, true);
593 uint32_t busy = drv_pipe_quiet(d, fed, PIPE_DEADLINE_NS);
594 if (busy) {
595 mesh_pt_set_hold(mb, s, false);
596 *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);
597 return false;
598 }
599 }
600 mesh_mailbox_zero(mb, x->view_off, x->image.manifest.view_bytes);
601 size_t k = 0;
602 for (size_t i = 0; i < d->ndead; i++) if (d->dead[i].consumer != s) d->dead[k++] = d->dead[i];
603 d->ndead = k;
604 if (inbox) mesh_pt_set_hold(mb, s, false);
605 x->needs_init = true;
606 if (x->image.manifest.peer == MESH_PEER_MODE_LOCKSTEP) drv_zero_exchange(d);
607 mesh_info("reset %s (slot %u): view zeroed, OP_INIT %s", drv_name(x), s, x->want_run ? "now" : "at the next start");
608 *text = mesh_xstrdup("");
609 return true;
610}
611
612static bool apply(mesh_driver *d, dblink *b, char **out) {
613 switch (b->k) {
614 case OP_LOAD: { mesh_prepared *p = b->p; b->p = NULL; return op_load(d, p, b->run, out); }
615 case OP_UPGRADE: { mesh_prepared *p = b->p; b->p = NULL; return op_upgrade(d, b->slot, p, out); }
616 case OP_ROLLBACK: return op_rollback(d, b->slot, out);
617 case OP_RESET: return drv_op_reset(d, b->slot, out);
618 default: *out = mesh_xstrdup(""); return true;
619 }
620}
621
622static void do_blink(mesh_driver *d, dblink b) {
623 mesh_mailbox mb = d->sh->mb;
624 int64_t drain = mesh_mono_ns() - b.since;
625 bool global = !b.has_scope;
626 /* mps, the CPU executors: a module change halts only its scope but frees and counts like a whole-context blink. */
627 bool iso = drv_be(d, b.exec)->ops->isolated(drv_be(d, b.exec));
628 bool modules = global || (iso && (b.k == OP_LOAD || b.k == OP_UPGRADE || b.k == OP_FREE));
629 bool scope_mask[MESH_PROG_SLOTS], filled[MESH_PROG_SLOTS];
630 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; }
631 int64_t t0 = d->halting ? d->halting_t : mesh_mono_ns();
632 d->halting = false;
633 char *load_dir = b.k == OP_LOAD && b.p ? mesh_xstrdup(mesh_prepared_dir(b.p)) : NULL;
634 /* H_BLINK only now, after the drain: nodes holding no ticket take none while it is set. */
635 if (global) mesh_mailbox_publish(mb, MESH_H_BLINK, 1);
636 /* 1. Halt. */
637 uint32_t live[MESH_PROG_SLOTS], left[MESH_PROG_SLOTS];
638 size_t nlive = 0, nleft = 0;
639 for (uint32_t s = 0; s < MESH_PROG_SLOTS; s++)
640 if (scope_mask[s] && d->slots[s] && drv_live(d->slots[s])) live[nlive++] = s;
641 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]; }
642 int64_t deadline = t0 + HALT_NS;
643 char *stuck = NULL;
644 while (nleft) {
645 size_t k = 0;
646 char *err = NULL;
647 for (size_t i = 0; i < nleft; i++) {
648 bool h = false;
649 char *e = NULL;
650 if (!drv_halted(d, left[i], &h, &e)) { if (!err) err = e; else free(e); continue; }
651 if (!h) left[k++] = left[i];
652 }
653 nleft = k;
654 if (err) { drv_context_fault(d, err); free(err); }
655 if (mesh_mono_ns() > deadline && nleft) { stuck = mesh_buf_fmt("%zu program kernel(s) did not halt in 3s", nleft); break; }
656 }
657 for (size_t i = 0; i < nlive; i++) {
658 bool still = false;
659 for (size_t j = 0; j < nleft; j++) still |= left[j] == live[i];
660 if (!still && d->slots[live[i]]) drv_stopped(d, live[i]);
661 }
662 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)"));
663 int64_t t_halt = mesh_mono_ns() - t0;
664 /* 2. Module operations: only when nothing of the context runs. */
665 bool ok;
666 char *result = NULL;
667 if (stuck) {
668 ok = false;
669 result = mesh_buf_fmt("blink: %s; no module changed", stuck);
670 } else if (d->fault) {
671 ok = false;
672 result = mesh_xstrdup("the CUDA context is lost; restart meshd");
673 } else {
674 if (modules) {
675 /* A GPU module unloads only when nothing of its context runs; a CPU one, and views, at any module blink of theirs. */
676 bool gpu_ok = global || d->gpu->ops->isolated(d->gpu);
677 uint32_t gpu_kind = drv_gpu_kind(d);
678 size_t k = 0;
679 for (size_t i = 0; i < d->nzombies; i++) {
680 dzombie z = d->zombies[i];
681 if (z.exec == gpu_kind && !gpu_ok) { d->zombies[k++] = z; continue; }
682 if (z.kernel) {
683 char *e = NULL;
684 if (!drv_be(d, z.exec)->ops->unload(drv_be(d, z.exec), z.kernel, &e)) { mesh_error("module unload: %s", e); free(e); }
685 z.kernel = NULL;
686 }
687 /* A dropped inbox: its pages are reused only once no producer can still be writing into it. */
688 if (z.has_inbox) {
689 uint32_t busy = drv_pipe_quiet(d, z.fed, PIPE_SAMPLE_NS);
690 if (busy) {
691 if (mesh_mono_ns() - z.since > PIPE_DEADLINE_NS && !z.warned) {
692 z.warned = true;
693 mesh_buf l = MESH_BUF_INIT;
694 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 ",
695 mesh_view_pages(z.view_bytes), z.view_off, busy);
696 mesh_rs_fixed(&l, ms(mesh_mono_ns() - z.since), 0);
697 mesh_buf_puts(&l, " ms");
698 mesh_error("%s", mesh_buf_str(&l));
699 mesh_buf_free(&l);
700 }
701 d->zombies[k++] = z;
702 continue;
703 }
704 }
705 mesh_view_heap_free(&d->heap, z.view_off, z.view_bytes);
706 }
707 d->nzombies = k;
708 }
709 ok = apply(d, &b, &result);
710 if (global || iso) {
711 uint32_t rebind[MESH_PROG_SLOTS];
712 size_t nr = rewrite_entries(d, rebind);
713 /* Only isolated executors leave kernels running across a module change. */
714 for (size_t i = 0; i < nr; i++) {
715 uint32_t s = rebind[i], exec = drv_exec_of(d, s);
716 if (!drv_be(d, exec)->ops->isolated(drv_be(d, exec))) continue;
717 dblink rb = {OP_REBIND, NULL, -1, 0, NULL, true, s, exec, mesh_mono_ns()};
718 drv_push_blink(d, rb);
719 }
720 }
721 char *e = NULL;
722 if (!size_stack(d, &e)) {
723 drv_context_fault(d, e);
724 free(result);
725 result = e;
726 ok = false;
727 }
728 }
729 free(stuck);
730 int64_t t_mod = mesh_mono_ns() - t0;
731 /* 3. Relaunch everything that runs (and a slot the op filled), then wait for all of it. */
732 uint32_t starting[MESH_PROG_SLOTS];
733 size_t nstart = 0;
734 for (uint32_t s = 0; s < MESH_PROG_SLOTS; s++) {
735 dslot *x = d->slots[s];
736 if (!(scope_mask[s] || !filled[s]) || !x || x->phase != PH_STOPPED || !x->want_run || x->failed || x->unloading || d->fault) continue;
737 char *e = NULL;
738 if (drv_launch_slot(d, s, &e)) starting[nstart++] = s;
739 else {
740 char *why = mesh_buf_fmt("launch: %s", e);
741 free(e);
742 drv_fail(d, s, why);
743 }
744 }
745 int64_t t_launch = mesh_mono_ns();
746 /* mps and the CPU executors: no waiting here either; phase_step sees them online (or fails them). */
747 int64_t wait_until = t_launch + (iso ? 0 : ONLINE_NS);
748 for (;;) {
749 bool up = true;
750 for (size_t i = 0; i < nstart; i++)
751 up &= mesh_mailbox_get(mb, mesh_prog_ctl(starting[i]) + MESH_P_ONLINE) >= d->slots[starting[i]]->image.manifest.nodes;
752 if (up || mesh_mono_ns() > wait_until) break;
753 }
754 for (size_t i = 0; i < nstart; i++) {
755 uint32_t s = starting[i];
756 if (!d->slots[s]) continue;
757 uint32_t n = d->slots[s]->image.manifest.nodes, on = mesh_mailbox_get(mb, mesh_prog_ctl(s) + MESH_P_ONLINE);
758 if (on >= n) drv_running(d, s, t_launch);
759 else if (!iso) drv_fail(d, s, mesh_buf_fmt("%u of %u nodes online after 2s (blink)", on, n));
760 }
761 int64_t total = mesh_mono_ns() - t0;
762 if (global) mesh_mailbox_publish(mb, MESH_H_BLINK, 0);
763 if (modules) {
764 d->epoch++;
765 mesh_mailbox_publish(mb, MESH_H_EPOCH, d->epoch);
766 mesh_host_block *h = &d->sh->host;
767 mesh_host_block_set_u32(h, MESH_S_EPOCH, d->epoch);
768 mesh_host_block_set_u32(h, MESH_S_BLINK_US, (uint32_t)(total / 1000));
769 mesh_host_block_set_u64(h, MESH_S_LAST_BLINK_MS, mesh_host_block_uptime_ms(h));
770 mesh_host_block_set_u32(h, MESH_S_HEAP_FREE, mesh_view_heap_free_pages(&d->heap));
771 mesh_host_block_set_u32(h, MESH_S_HEAP_RUN, mesh_view_heap_largest_run(&d->heap));
772 }
773 long long us = total / 1000;
774 mesh_buf l = MESH_BUF_INIT;
775 mesh_buf_puts(&l, "blink");
776 if (modules) mesh_buf_printf(&l, " epoch %u", d->epoch);
777 if (!b.has_scope) mesh_buf_puts(&l, " (whole context)");
778 else if (b.scope == MESH_NO_SLOT) mesh_buf_puts(&l, " (no program halted)");
779 else mesh_buf_printf(&l, " (slot %u)", b.scope);
780 mesh_buf_puts(&l, ": drain ");
781 mesh_rs_fixed(&l, ms(drain), 2);
782 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),
783 (long long)((t_mod - t_halt) / 1000), (long long)((total - t_mod) / 1000), nstart, us, ok ? "ok" : "err", result);
784 mesh_info("%s", mesh_buf_str(&l));
785 mesh_buf_free(&l);
786 /* Replies: load `slot S blink U`, the others `blink U [view kept|reset]`. */
787 char *reply;
788 if (!ok) reply = mesh_xstrdup(result);
789 else if (!*result) reply = mesh_buf_fmt("blink %lld", us);
790 else if (strncmp(result, "slot ", 5) == 0) reply = mesh_buf_fmt("%s blink %lld", result, us);
791 else reply = mesh_buf_fmt("blink %lld %s", us, result);
792 d->state_dirty = true;
793 if (load_dir && !ok) {
794 char *e = mesh_buf_fmt("%s: %s", load_dir, reply);
795 drv_startup_failed(d, b.reply, e);
796 free(e);
797 }
798 drv_answer(b.reply, ok, reply);
799 free(reply);
800 free(result);
801 free(load_dir);
802 if (b.p) { mesh_prepared_free(b.p); free(b.p); }
803}
804
805void drv_blink_tick(mesh_driver *d) {
806 if (!d->nblinks) return;
807 dblink *b = &d->blinks[0];
808 if (d->fault) {
809 dblink x = *b;
810 memmove(d->blinks, d->blinks + 1, (d->nblinks - 1) * sizeof *d->blinks);
811 d->nblinks--;
812 mesh_reply_err(x.reply, "the CUDA context is lost; restart meshd");
813 if (x.p) { mesh_prepared_free(x.p); free(x.p); }
814 return;
815 }
816 bool iso = drv_be(d, b->exec)->ops->isolated(drv_be(d, b->exec));
817 bool drained = true;
818 for (uint32_t s = 0; s < MESH_PROG_SLOTS && drained; s++)
819 if (drv_in_scope(d, b->has_scope, b->scope, s) && d->slots[s] && !mesh_pring_idle(&d->slots[s]->ring)) drained = false;
820 if (drained) {
821 if (iso && !halt_scope(d, b->has_scope, b->scope)) return;
822 dblink x = *b;
823 memmove(d->blinks, d->blinks + 1, (d->nblinks - 1) * sizeof *d->blinks);
824 d->nblinks--;
825 do_blink(d, x);
826 } else if (mesh_mono_ns() - b->since > DRAIN_NS) {
827 dblink x = *b;
828 memmove(d->blinks, d->blinks + 1, (d->nblinks - 1) * sizeof *d->blinks);
829 d->nblinks--;
830 const char *e = "blink: in-flight tasks did not drain in 3s; nothing changed";
831 mesh_error("%s", e);
832 mesh_reply_err(x.reply, e);
833 if (x.p) { mesh_prepared_free(x.p); free(x.p); }
834 }
835}
836
837/* Keeps slot_free referenced for the unload paths that free a whole slot here. */
838void drv_slot_free(dslot *x);
839void drv_slot_free(dslot *x) { slot_free(x); }