==> src/driver.c <==1/* The driver: its state, the loop, startup and shutdown, statistics, and the commands other threads send. */
2#include "driver_int.h"
3
4#include "mesh/cpuuse.h"
5#include "mesh/native.h"
6#include "mesh/rsfmt.h"
7#include "mesh/state_file.h"
8
9#include <stdio.h>
10#include <stdlib.h>
11#include <string.h>
12
13
14/* ---------------------------------------------------------------- helpers */
15
16uint32_t drv_gpu_kind(const mesh_driver *d) { return d->gpu->ops->caps(d->gpu)->kind; }
17
18/* The executor of kind exec among the GPU backend and the CPU executors. */
19mesh_backend *drv_be(const mesh_driver *d, uint32_t exec) {
20 if (exec == drv_gpu_kind(d)) return d->gpu;
21 for (size_t i = 0; i < d->ncpus; i++)
22 if (d->cpus[i]->ops->caps(d->cpus[i])->kind == exec) return d->cpus[i];
23 mesh_error("internal: no executor %s", exec < MESH_BACKENDS_LEN ? MESH_BACKENDS[exec] : "?");
24 abort();
25}
26
27uint32_t drv_exec_of(const mesh_driver *d, uint32_t s) { return d->slots[s] ? d->slots[s]->image.exec : drv_gpu_kind(d); }
28
29bool drv_in_scope(const mesh_driver *d, bool has_scope, uint32_t scope, uint32_t s) {
30 if (has_scope) return scope == s;
31 return drv_exec_of(d, s) == drv_gpu_kind(d);
32}
33
34const char *drv_name(const dslot *x) { return x->image.manifest.name; }
35bool drv_live(const dslot *x) { return x->phase != PH_STOPPED; }
36
37uint32_t drv_state(const dslot *x) {
38 if (x->failed && !drv_live(x)) return MESH_ST_FAILED;
39 switch (x->phase) {
40 case PH_STOPPED: return MESH_ST_LOADED;
41 case PH_STARTING:
42 case PH_RUNNING: return MESH_ST_RUNNING;
43 default: return MESH_ST_STOPPING;
44 }
45}
46
47int drv_find(const mesh_driver *d, const char *name) {
48 for (uint32_t s = 0; s < MESH_PROG_SLOTS; s++)
49 if (d->slots[s] && !d->slots[s]->unloading && strcmp(drv_name(d->slots[s]), name) == 0) return (int)s;
50 return -1;
51}
52
53void drv_entry(const dslot *x, uint8_t out[MESH_PT_BYTES]) {
54 const dimage *i = &x->image;
55 mesh_image im = {x->view_off, x->reads_mask, i->hash, i->regs, i->code_bytes, i->rt_abi, x->loads};
56 mesh_manifest_encode_entry(&i->manifest, &im, out);
57}
58
59static void put32(uint8_t *b, size_t o, uint32_t v) { for (int i = 0; i < 4; i++) b[o + (size_t)i] = (uint8_t)(v >> (8 * i)); }
60
61void drv_pipe(const dslot *x, uint8_t b[MESH_PP_BYTES]) {
62 memset(b, 0, MESH_PP_BYTES);
63 const mesh_manifest_inbox *in = x->image.manifest.inbox;
64 if (in) {
65 put32(b, MESH_PP_IN_OFF, in->off);
66 put32(b, MESH_PP_IN_CAP, in->cap);
67 put32(b, MESH_PP_IN_REC, in->rec);
68 put32(b, MESH_PP_IN_TYPE, mesh_pipe_type_hash((const uint8_t *)in->ty, strlen(in->ty)));
69 }
70 put32(b, MESH_PP_FEEDS_MASK, x->feeds_mask);
71 put32(b, MESH_PP_FED_MASK, x->fed_mask);
72 if (in) memcpy(b + MESH_PP_TYPE_NAME, in->ty, strlen(in->ty));
73}
74
75void drv_ext(const dslot *x, uint8_t b[MESH_PX_BYTES]) {
76 mesh_manifest_encode_ext(&x->image.manifest, b);
77 put32(b, MESH_PX_PER_SM, x->image.per_sm);
78 put32(b, MESH_PX_LOCAL, x->image.local);
79 put32(b, MESH_PX_EXEC, x->image.exec);
80 put32(b, MESH_PX_REG_CAP, x->image.reg_cap);
81 uint64_t sc = mesh_kernel_scratch(x->image.kernel);
82 put32(b, MESH_PX_SCRATCH, (uint32_t)sc);
83 put32(b, MESH_PX_SCRATCH + 4, (uint32_t)(sc >> 32));
84}
85
86/* Upgrade/rollback keep the view iff view_abi, view_bytes and the inbox are unchanged. */
87bool drv_view_compatible(const mesh_manifest *a, const mesh_manifest *b) {
88 if (a->view_abi != b->view_abi || a->view_bytes != b->view_bytes) return false;
89 if (!a->inbox || !b->inbox) return !a->inbox && !b->inbox;
90 return a->inbox->off == b->inbox->off && a->inbox->cap == b->inbox->cap && a->inbox->rec == b->inbox->rec &&
91 strcmp(a->inbox->ty, b->inbox->ty) == 0;
92}
93
94void drv_answer(mesh_reply *reply, bool ok, const char *text) {
95 if (!reply) return;
96 if (ok) mesh_reply_ok(reply, text);
97 else mesh_reply_err(reply, text);
98}
99
100/* A load with no one to answer (--state, --autoload) failed during startup: fatal. */
101void drv_startup_failed(mesh_driver *d, const mesh_reply *reply, const char *e) {
102 if (!reply && d->restoring && !d->startup_error) d->startup_error = mesh_xstrdup(e);
103}
104
105void drv_image_free(dimage *im) {
106 mesh_manifest_free(&im->manifest);
107 free(im->dir);
108 if (im->kernel) mesh_kernel_drop(im->kernel);
109 memset(im, 0, sizeof *im);
110}
111
112#define PUSH(arr, n, cap, v) \
113 do { \
114 if ((n) == (cap)) { \
115 (cap) = (cap) ? 2 * (cap) : 8; \
116 (arr) = mesh_xrealloc((arr), (cap) * sizeof *(arr)); \
117 } \
118 (arr)[(n)++] = (v); \
119 } while (0)
120
121void drv_push_blink(mesh_driver *d, dblink b) { PUSH(d->blinks, d->nblinks, d->cblinks, b); }
122void drv_push_job(mesh_driver *d, djob j) { PUSH(d->jobs, d->njobs, d->cjobs, j); }
123static void push_waiter(mesh_driver *d, dwaiter w) { PUSH(d->waiters, d->nwaiters, d->cwaiters, w); }
124
125uint32_t mesh_reads_mask(const mesh_manifest *m, const char *const *names, size_t n) {
126 for (size_t i = 0; i < m->nreads; i++) if (strcmp(m->reads[i], "*") == 0) return MESH_READS_ALL;
127 uint32_t mask = 0;
128 for (size_t s = 0; s < n; s++) {
129 if (!names[s]) continue;
130 for (size_t i = 0; i < m->nreads; i++) if (strcmp(m->reads[i], names[s]) == 0) { mask |= 1u << s; break; }
131 }
132 return mask;
133}
134
135/* ---------------------------------------------------------------- job setup (SetupState) */
136
137mesh_setup_step mesh_setup_state_step(mesh_setup_state *s, uint64_t job, const mj_setup *setup, size_t n) {
138 mesh_setup_step r = {MESH_SETUP_READY, 0, 0, NULL};
139 if (!s->has_job || s->job != job) {
140 mesh_setup_state_reset(s);
141 s->has_job = true;
142 s->job = job;
143 }
144 if (s->busy) { r.k = MESH_SETUP_WAIT; return r; }
145 if (s->err) { r.k = MESH_SETUP_FAILED; r.err = s->err; return r; }
146 if (s->next < n) { r.k = MESH_SETUP_PUBLISH; r.op = setup[s->next].op; r.arg = setup[s->next].arg; }
147 return r;
148}
149
150void mesh_setup_state_published(mesh_setup_state *s) {
151 s->next++;
152 s->busy = true;
153}
154
155void mesh_setup_state_completed(mesh_setup_state *s, uint64_t job, const char *err) {
156 if (!s->has_job || s->job != job) return;
157 s->busy = false;
158 if (!s->err && err) s->err = mesh_xstrdup(err);
159}
160
161void mesh_setup_state_reset(mesh_setup_state *s) {
162 free(s->err);
163 memset(s, 0, sizeof *s);
164}
165
166/* ---------------------------------------------------------------- construction, startup */
167
168mesh_driver *mesh_driver_new(mesh_shared *sh, mesh_backend *gpu, mesh_backend **cpus, size_t ncpus, mesh_driver_config cfg) {
169 mesh_driver *d = mesh_xcalloc(1, sizeof *d);
170 mesh_mailbox_set(sh->mb, MESH_H_MAGIC, MESH_MAGIC);
171 mesh_mailbox_set(sh->mb, MESH_H_VERSION, MESH_VERSION);
172 d->sh = sh;
173 d->gpu = gpu;
174 d->cpus = mesh_xcalloc(ncpus ? ncpus : 1, sizeof *d->cpus);
175 for (size_t i = 0; i < ncpus; i++) d->cpus[i] = cpus[i];
176 d->ncpus = ncpus;
177 d->cfg = cfg;
178 int64_t now = mesh_mono_ns();
179 d->last_stats = d->last_log = now;
180 mesh_view_heap_init(&d->heap);
181 d->restoring = true;
182 d->me = sh->me;
183 d->n = sh->n;
184 return d;
185}
186
187/* The executors' kinds, the GPU's first. */
188static size_t execs(const mesh_driver *d, uint32_t *out) {
189 size_t n = 0;
190 out[n++] = drv_gpu_kind(d);
191 for (size_t i = 0; i < d->ncpus; i++) out[n++] = d->cpus[i]->ops->caps(d->cpus[i])->kind;
192 return n;
193}
194
195static bool startup_loads(mesh_driver *d, char **err) {
196 if (d->cfg.skip_startup) {
197 mesh_info("startup: the fleet's committed desired state decides what loads (--state / --autoload skipped)");
198 return true;
199 }
200 char **names = NULL;
201 size_t nnames = 0, cnames = 0;
202 if (d->cfg.state) {
203 mesh_saved *set;
204 size_t n;
205 if (!mesh_state_read(d->cfg.state, &set, &n, err)) return false;
206 for (size_t i = 0; i < n; i++) {
207 mesh_info("--state: %s from %s (%s)", set[i].name, set[i].dir, set[i].running ? "running" : "stopped");
208 PUSH(names, nnames, cnames, mesh_xstrdup(set[i].name));
209 djob j = {false, mesh_src_catalog(set[i].dir), set[i].running ? 1 : 0, 0, NULL};
210 drv_push_job(d, j);
211 }
212 mesh_state_free(set, n);
213 }
214 if (d->cfg.autoload) {
215 uint32_t ex[8];
216 size_t nex = execs(d, ex);
217 mesh_buf me = MESH_BUF_INIT;
218 for (size_t i = 0; i < nex; i++) mesh_buf_printf(&me, "%s%s", i ? ", " : "", MESH_BACKENDS[ex[i]]);
219 char **cat;
220 size_t ncat;
221 if (!mesh_catalog(d->sh->programs, &cat, &ncat, err)) {
222 mesh_buf_free(&me);
223 for (size_t i = 0; i < nnames; i++) free(names[i]);
224 free(names);
225 return false;
226 }
227 for (size_t c = 0; c < ncat; c++) {
228 char *dir = mesh_buf_fmt("%s/%s", d->sh->programs, cat[c]);
229 mesh_manifest m;
230 char *ptx = NULL, *e = NULL;
231 if (!mesh_manifest_load(dir, &m, &ptx, &e)) {
232 mesh_error("--autoload: %s", e);
233 free(e);
234 free(dir);
235 continue;
236 }
237 free(ptx);
238 bool known = false;
239 for (size_t i = 0; i < nnames; i++) known |= strcmp(names[i], m.name) == 0;
240 if (m.autostart && !mesh_runs_here(dir, &m, ex, nex)) {
241 mesh_buf b = MESH_BUF_INIT;
242 for (size_t i = 0; i < m.nbackends; i++) mesh_buf_printf(&b, "%s%s", i ? ", " : "", m.backends[i]);
243 char *suffix = mesh_manifest_native(&m) ? mesh_buf_fmt(", and has no native binary of it for %s", mesh_native_target())
244 : mesh_xstrdup("");
245 mesh_info("--autoload: %s: not loaded: unsupported here (it runs on %s; this meshd runs %s%s)", m.name, mesh_buf_str(&b),
246 mesh_buf_str(&me), suffix);
247 free(suffix);
248 mesh_buf_free(&b);
249 } else if (m.autostart && !known) {
250 mesh_info("--autoload: %s", m.name);
251 PUSH(names, nnames, cnames, mesh_xstrdup(m.name));
252 djob j = {false, mesh_src_catalog(cat[c]), -1, 0, NULL};
253 drv_push_job(d, j);
254 }
255 mesh_manifest_free(&m);
256 free(dir);
257 }
258 for (size_t c = 0; c < ncat; c++) free(cat[c]);
259 free(cat);
260 mesh_buf_free(&me);
261 }
262 for (size_t i = 0; i < nnames; i++) free(names[i]);
263 free(names);
264 return true;
265}
266
267bool mesh_driver_start(mesh_driver *d, char **err) {
268 if (!d->gpu->ops->make_current(d->gpu, err)) return false;
269 return startup_loads(d, err);
270}
271
272/* ---------------------------------------------------------------- the loop */
273
274static void save_state(mesh_driver *d) {
275 d->state_dirty = false;
276 if (!d->cfg.state) return;
277 mesh_saved set[MESH_PROG_SLOTS];
278 size_t n = 0;
279 for (uint32_t s = 0; s < MESH_PROG_SLOTS; s++) {
280 dslot *x = d->slots[s];
281 if (!x || x->unloading) continue;
282 set[n].name = (char *)drv_name(x);
283 set[n].dir = x->image.dir;
284 set[n].running = x->want_run;
285 n++;
286 }
287 char *err = NULL;
288 if (!mesh_state_write(d->cfg.state, set, n, &err)) { mesh_error("%s", err); free(err); }
289}
290
291/* Something wants prompt attention: a continuous program, a transition, a blink, job work. */
292static bool busy(const mesh_driver *d) {
293 for (uint32_t s = 0; s < MESH_PROG_SLOTS; s++) if (d->jobw[s].n || d->jobw[s].inflight) return true;
294 if (d->nblinks || d->working) return true;
295 for (uint32_t s = 0; s < MESH_PROG_SLOTS; s++) {
296 const dslot *x = d->slots[s];
297 if (!x) continue;
298 switch (x->phase) {
299 case PH_STOPPED: break;
300 case PH_RUNNING:
301 if (x->image.manifest.policy == MESH_POLICY_CONTINUOUS || !mesh_pring_idle(&x->ring)) return true;
302 break;
303 default: return true;
304 }
305 }
306 return false;
307}
308
309static void commands(mesh_driver *d) {
310 mesh_cmd c;
311 while (mesh_chan_try_recv(d->sh->cmd, &c) == MESH_CHAN_OK) {
312 drv_program_cmd(d, &c);
313 mesh_cmd_free(&c);
314 }
315}
316
317/* Per member over the last second: exchange RTT, wait per round, rounds received (S_MEMBERS). */
318static void member_stats(mesh_driver *d, float dt, float *rtt_out, uint32_t *cnt_out, float *worst_out) {
319 mesh_rtt_acc rtt[MESH_MAX_SYSTEMS], wait[MESH_MAX_SYSTEMS];
320 pthread_mutex_lock(&d->sh->mu);
321 memcpy(rtt, d->sh->rtt, sizeof rtt);
322 memcpy(wait, d->sh->wait, sizeof wait);
323 memset(d->sh->rtt, 0, sizeof d->sh->rtt);
324 memset(d->sh->wait, 0, sizeof d->sh->wait);
325 pthread_mutex_unlock(&d->sh->mu);
326 mesh_host_block *h = &d->sh->host;
327 float sum = 0, worst = 0;
328 uint32_t cnt = 0;
329 for (uint32_t k = 0; k < d->n && k < MESH_MAX_SYSTEMS; k++) {
330 size_t m = mesh_member(k);
331 float r = rtt[k].n ? (float)(rtt[k].sum / rtt[k].n) : 0.0f, w = wait[k].n ? (float)(wait[k].sum / wait[k].n) : 0.0f;
332 mesh_host_block_set_f32(h, m + MESH_M_RTT_US, r);
333 mesh_host_block_set_f32(h, m + MESH_M_WAIT_US, w);
334 float per = (float)wait[k].n / (dt > 1e-3f ? dt : 1e-3f);
335 mesh_host_block_set_u32(h, m + MESH_M_ROUNDS, (uint32_t)(per + 0.5f));
336 uint64_t round;
337 if (k == d->me) {
338 uint32_t rank = mesh_mailbox_get(d->sh->mb, MESH_H_GROUP_RANK);
339 round = mesh_mailbox_get64(d->sh->mb, mesh_ex_counter(rank < MESH_MAX_SYSTEMS - 1 ? rank : MESH_MAX_SYSTEMS - 1));
340 } else round = atomic_load(&d->sh->round_of[k]);
341 mesh_host_block_set_u64(h, m + MESH_M_ROUND, round);
342 if (rtt[k].n) { sum += r; cnt++; }
343 if (w > worst) worst = w;
344 }
345 *rtt_out = cnt ? sum / (float)cnt : 0.0f;
346 *cnt_out = cnt;
347 *worst_out = worst;
348}
349
350static void stats(mesh_driver *d) {
351 int64_t now = mesh_mono_ns();
352 float dt = (float)((double)(now - d->last_stats) / 1e9);
353 d->last_stats = now;
354 float rtt, wait;
355 uint32_t n;
356 member_stats(d, dt, &rtt, &n, &wait);
357 float total = 0, rates[MESH_PROG_SLOTS] = {0};
358 for (uint32_t s = 0; s < MESH_PROG_SLOTS; s++) {
359 dslot *x = d->slots[s];
360 if (!x) continue;
361 uint32_t done = mesh_mailbox_get(d->sh->mb, mesh_prog_ctl(s) + MESH_P_DONE);
362 x->rate = (float)(done - x->last_done) / dt;
363 x->last_done = done;
364 rates[s] = x->rate;
365 total += x->rate;
366 }
367 mesh_backend_sample(d->gpu);
368 for (size_t i = 0; i < d->ncpus; i++) mesh_backend_sample(d->cpus[i]);
369 mesh_host_block *h = &d->sh->host;
370 uint8_t rb[4 * MESH_PROG_SLOTS];
371 for (uint32_t s = 0; s < MESH_PROG_SLOTS; s++) memcpy(rb + 4 * s, &rates[s], 4);
372 mesh_host_field f = {MESH_S_RATES, rb, sizeof rb};
373 mesh_host_block_set_many(h, &f, 1);
374 mesh_host_block_set_f32(h, MESH_S_TASKS_PER_S, total);
375 mesh_host_block_set_f32(h, MESH_S_PEER_RTT_US, rtt);
376 mesh_host_block_set_f32(h, MESH_S_LOCK_WAIT_US, wait);
377 uint32_t progs = 0;
378 for (uint32_t s = 0; s < MESH_PROG_SLOTS; s++) progs += d->slots[s] != NULL;
379 mesh_host_block_set_u32(h, MESH_S_PROGRAMS, progs);
380 mesh_host_block_set_u32(h, MESH_S_HEAP_FREE, mesh_view_heap_free_pages(&d->heap));
381 mesh_host_block_set_u32(h, MESH_S_HEAP_RUN, mesh_view_heap_largest_run(&d->heap));
382 if (now - d->last_log >= 10000000000ll) {
383 d->last_log = now;
384 mesh_buf p = MESH_BUF_INIT;
385 int first = 1;
386 for (uint32_t s = 0; s < MESH_PROG_SLOTS; s++) {
387 dslot *x = d->slots[s];
388 if (!x) continue;
389 mesh_buf_printf(&p, "%s%s %s ", first ? "" : ", ", drv_name(x), MESH_STATES[drv_state(x)]);
390 mesh_rs_fixed(&p, x->rate, 0);
391 mesh_buf_puts(&p, "/s");
392 first = 0;
393 }
394 mesh_buf line = MESH_BUF_INIT;
395 mesh_rs_fixed(&line, total, 0);
396 mesh_buf_printf(&line, " tasks/s: %s (epoch %u, peer state %u", first ? "no programs" : mesh_buf_str(&p), d->epoch,
397 atomic_load(&d->sh->peer_state));
398 if (n > 0) {
399 mesh_buf_puts(&line, ", exchange rtt ");
400 mesh_rs_fixed(&line, rtt, 0);
401 mesh_buf_puts(&line, " us, slowest member's wait ");
402 mesh_rs_fixed(&line, wait, 0);
403 mesh_buf_puts(&line, " us/round");
404 }
405 mesh_buf_putc(&line, ')');
406 mesh_info("%s", mesh_buf_str(&line));
407 mesh_buf_free(&line);
408 mesh_buf_free(&p);
409 }
410}
411
412/* A native slot's worker words (P_PID .. P_LIMITS) every tick, and its CPU time and RSS for the load sampler. */
413void drv_native_words(mesh_driver *d, uint32_t s);
414void drv_native_words(mesh_driver *d, uint32_t s) {
415 dslot *x = d->slots[s];
416 if (!x || !mesh_kernel_native(x->image.kernel)) return;
417 mesh_native_stats st = mesh_native_worker_stats(mesh_kernel_native(x->image.kernel)), pst = {0};
418 bool prev = x->has_prev && mesh_kernel_native(x->prev.kernel);
419 if (prev) pst = mesh_native_worker_stats(mesh_kernel_native(x->prev.kernel));
420 mesh_mailbox mb = d->sh->mb;
421 size_t c = mesh_prog_ctl(s);
422 mesh_mailbox_set(mb, c + MESH_P_PID, (uint32_t)st.pid);
423 mesh_mailbox_set(mb, c + MESH_P_SPAWNS, st.spawns);
424 mesh_mailbox_set(mb, c + MESH_P_CRASHES, st.crashes);
425 mesh_mailbox_set(mb, c + MESH_P_EXIT, st.exit);
426 mesh_mailbox_set(mb, c + MESH_P_LIMITS, st.limits);
427 /* The slot's CPU counter is monotonic across images: add this image's growth. */
428 uintptr_t key = (uintptr_t)mesh_kernel_native(x->image.kernel);
429 if (d->native_cpu[s].key != key) { d->native_cpu[s].key = key; d->native_cpu[s].cpu = 0; }
430 mesh_slot_use *u = &mesh_cpuuse_slots[s];
431 mesh_slot_use_add_cpu(u, st.cpu_ns > d->native_cpu[s].cpu ? st.cpu_ns - d->native_cpu[s].cpu : 0);
432 if (st.cpu_ns > d->native_cpu[s].cpu) d->native_cpu[s].cpu = st.cpu_ns;
433 atomic_store_explicit(&u->rss, st.rss + (prev ? pst.rss : 0), memory_order_relaxed);
434 atomic_store_explicit(&u->units, (uint32_t)(st.pid != 0) + (uint32_t)(prev && pst.pid != 0), memory_order_relaxed);
435}
436
437static bool all_idle(const mesh_driver *d) {
438 for (uint32_t s = 0; s < MESH_PROG_SLOTS; s++) if (d->slots[s] && !mesh_pring_idle(&d->slots[s]->ring)) return false;
439 return true;
440}
441
442static int shutdown_(mesh_driver *d) {
443 mesh_info("signal: quiescing");
444 mesh_mailbox mb = d->sh->mb;
445 int64_t deadline = mesh_mono_ns() + 400000000ll;
446 while (!all_idle(d) && mesh_mono_ns() < deadline) {
447 for (uint32_t s = 0; s < MESH_PROG_SLOTS; s++) {
448 if (!d->slots[s]) continue;
449 mesh_done *done;
450 size_t n;
451 mesh_pring_collect(&d->slots[s]->ring, mb, &done, &n);
452 for (size_t i = 0; i < n; i++) mesh_reply_drop(done[i].out.kind.reply);
453 free(done);
454 }
455 mesh_sleep_us(TICK_NS / 1000);
456 }
457 uint32_t left[MESH_PROG_SLOTS];
458 size_t nleft = 0;
459 for (uint32_t s = 0; s < MESH_PROG_SLOTS; s++)
460 if (d->slots[s] && drv_live(d->slots[s])) {
461 mesh_mailbox_publish(mb, mesh_prog_ctl(s) + MESH_P_SHUTDOWN, MESH_SHUTDOWN_EXIT);
462 left[nleft++] = s;
463 }
464 deadline = mesh_mono_ns() + 500000000ll;
465 while (nleft && mesh_mono_ns() < deadline) {
466 size_t k = 0;
467 for (size_t i = 0; i < nleft; i++) {
468 bool h = true;
469 char *e = NULL;
470 if (!drv_halted(d, left[i], &h, &e)) { free(e); h = true; }
471 if (!h) left[k++] = left[i];
472 }
473 nleft = k;
474 mesh_sleep_us(100);
475 }
476 if (!d->restoring) save_state(d);
477 if (!nleft) { mesh_info("programs halted, exiting"); return 0; }
478 mesh_buf b = MESH_BUF_INIT;
479 mesh_buf_putc(&b, '[');
480 for (size_t i = 0; i < nleft; i++) mesh_buf_printf(&b, "%s%u", i ? ", " : "", left[i]);
481 mesh_buf_putc(&b, ']');
482 mesh_error("shutdown: programs in slots %s did not halt", mesh_buf_str(&b));
483 mesh_buf_free(&b);
484 return 1;
485}
486
487int mesh_driver_run(mesh_driver *d) {
488 for (;;) {
489 if (atomic_load(d->sh->quit)) return shutdown_(d);
490 int64_t now = mesh_mono_ns();
491 commands(d);
492 drv_programs_tick(d, now);
493 drv_jobs_tick(d);
494 drv_blink_tick(d);
495 if (d->startup_error) {
496 mesh_error("FAILED startup load (%s): --state is left as it was; exiting", d->startup_error);
497 free(d->startup_error);
498 d->startup_error = NULL;
499 shutdown_(d);
500 return 1;
501 }
502 if (mesh_mono_ns() - d->last_stats >= 1000000000ll) stats(d);
503 if (d->restoring && !d->njobs && !d->working && !d->nblinks) {
504 d->restoring = false;
505 d->state_dirty = true;
506 atomic_store(&d->sh->ready, true);
507 size_t n = 0;
508 for (uint32_t s = 0; s < MESH_PROG_SLOTS; s++) n += d->slots[s] != NULL;
509 mesh_info("startup loads done: %zu program(s) loaded", n);
510 }
511 if (!d->restoring && d->state_dirty) save_state(d);
512 mesh_sleep_us((busy(d) ? FAST_TICK_NS : TICK_NS) / 1000);
513 }
514}
515
516void mesh_driver_free(mesh_driver *d) {
517 if (!d) return;
518 for (uint32_t s = 0; s < MESH_PROG_SLOTS; s++) {
519 dslot *x = d->slots[s];
520 djobslot *j = &d->jobw[s];
521 for (size_t i = 0; i < j->n; i++) drv_lease_free(&j->q[i]);
522 free(j->q);
523 mesh_setup_state_reset(&j->setup);
524 if (!x) continue;
525 mesh_backend *be = drv_be(d, x->image.exec);
526 char *e = NULL;
527 if (x->image.kernel && !be->ops->unload(be, x->image.kernel, &e)) free(e);
528 x->image.kernel = NULL;
529 if (x->has_prev && x->prev.kernel && !drv_be(d, x->prev.exec)->ops->unload(drv_be(d, x->prev.exec), x->prev.kernel, &e)) free(e);
530 x->prev.kernel = NULL;
531 drv_image_free(&x->image);
532 if (x->has_prev) drv_image_free(&x->prev);
533 mesh_stream_drop(x->stream);
534 mesh_pring_free(&x->ring);
535 for (size_t i = 0; i < x->ntasks; i++) mesh_reply_drop(x->tasks[i].reply);
536 for (size_t i = 0; i < x->nreqs; i++) { mesh_reply_drop(x->reqs[i].reply); free(x->reqs[i].bytes); }
537 free(x->tasks);
538 free(x->reqs);
539 free(x->failed);
540 free(x);
541 }
542 for (size_t i = 0; i < d->nzombies; i++) {
543 char *e = NULL;
544 if (d->zombies[i].kernel && !drv_be(d, d->zombies[i].exec)->ops->unload(drv_be(d, d->zombies[i].exec), d->zombies[i].kernel, &e)) free(e);
545 }
546 free(d->zombies);
547 for (size_t i = 0; i < d->njobs; i++) { mesh_src_free(&d->jobs[i].src); mesh_reply_drop(d->jobs[i].reply); }
548 free(d->jobs);
549 if (d->working) {
550 /* The JIT thread still holds its sender: let go of our end. */
551 mesh_reply_release(d->work_rx);
552 mesh_src_free(&d->work_job.src);
553 mesh_reply_drop(d->work_job.reply);
554 }
555 for (size_t i = 0; i < d->nblinks; i++) {
556 if (d->blinks[i].p) { mesh_prepared_free(d->blinks[i].p); free(d->blinks[i].p); }
557 mesh_reply_drop(d->blinks[i].reply);
558 }
559 free(d->blinks);
560 for (size_t i = 0; i < d->nwaiters; i++) mesh_reply_drop(d->waiters[i].reply);
561 free(d->waiters);
562 if (d->gpu_load) mesh_reply_release(d->gpu_load_rx);
563 free(d->dead);
564 free(d->fault);
565 free(d->startup_error);
566 mesh_backend_destroy(d->gpu);
567 for (size_t i = 0; i < d->ncpus; i++) mesh_backend_destroy(d->cpus[i]);
568 free(d->cpus);
569 free(d);
570}
571
572/* ---------------------------------------------------------------- commands */
573
574static char *unknown(const char *n) {
575 mesh_buf b = MESH_BUF_INIT;
576 mesh_buf_puts(&b, "no program ");
577 mesh_rs_debug_cstr(&b, n);
578 mesh_buf_puts(&b, " (`programs` lists them)");
579 return mesh_buf_take(&b);
580}
581
582static void err_unknown(mesh_reply **reply, const char *n) {
583 char *e = unknown(n);
584 mesh_reply_err(*reply, e);
585 *reply = NULL;
586 free(e);
587}
588
589static mesh_reply *take(mesh_reply **r) {
590 mesh_reply *x = *r;
591 *r = NULL;
592 return x;
593}
594
595static void cmd_unload(mesh_driver *d, mesh_cmd *c) {
596 int s = drv_find(d, c->name);
597 if (s < 0) { err_unknown(&c->reply, c->name); return; }
598 mesh_info("unload %s (slot %d)%s", c->name, s, c->now ? " --now" : "");
599 dslot *x = d->slots[s];
600 x->unloading = true;
601 x->want_run = false;
602 mesh_mailbox_set(d->sh->mb, mesh_prog_ctl((uint32_t)s) + MESH_P_RUN, 0);
603 dwaiter w = {(uint32_t)s, false, true, c->now, take(&c->reply), mesh_mono_ns()};
604 push_waiter(d, w);
605 if (!drv_live(x)) drv_remove(d, (uint32_t)s);
606}
607
608static void cmd_prog_run(mesh_driver *d, mesh_cmd *c) {
609 int s = drv_find(d, c->name);
610 if (s < 0) { err_unknown(&c->reply, c->name); return; }
611 /* P_RUN is the host's word; GPU requests come through the GPU-owned P_RUN_REQ. */
612 mesh_mailbox_set(d->sh->mb, mesh_prog_ctl((uint32_t)s) + MESH_P_RUN, c->on);
613 dslot *x = d->slots[s];
614 if (d->fault && drv_exec_of(d, (uint32_t)s) == drv_gpu_kind(d)) {
615 mesh_reply_errf(take(&c->reply), "the CUDA context is lost (%s); restart meshd", d->fault);
616 return;
617 }
618 if (c->on && x->failed) {
619 mesh_info("start %s: clearing FAILED (%s)", c->name, x->failed);
620 free(x->failed);
621 x->failed = NULL;
622 }
623 if (x->want_run != c->on) mesh_info("%s %s", c->on ? "start" : "stop", c->name);
624 x->want_run = c->on;
625 bool done = c->on ? x->phase == PH_RUNNING : x->phase == PH_STOPPED;
626 if (done) mesh_reply_ok(take(&c->reply), "0 us");
627 else {
628 dwaiter w = {(uint32_t)s, c->on, false, false, take(&c->reply), mesh_mono_ns()};
629 push_waiter(d, w);
630 }
631 d->state_dirty = true;
632}
633
634static void cmd_task(mesh_driver *d, mesh_cmd *c) {
635 int s = drv_find(d, c->name);
636 if (s < 0) { err_unknown(&c->reply, c->name); return; }
637 dslot *x = d->slots[s];
638 if (!x->want_run || x->failed) mesh_reply_errf(take(&c->reply), "%s is not running", c->name);
639 else if (x->ntasks >= MAX_QUEUED) mesh_reply_errf(take(&c->reply), "%s: %d tasks queued already", c->name, MAX_QUEUED);
640 else {
641 dtask t = {c->op, c->arg, take(&c->reply)};
642 PUSH(x->tasks, x->ntasks, x->ctasks, t);
643 }
644}
645
646static void cmd_req(mesh_driver *d, mesh_cmd *c) {
647 int s = drv_find(d, c->name);
648 if (s < 0) { err_unknown(&c->reply, c->name); return; }
649 dslot *x = d->slots[s];
650 const mesh_manifest *m = &x->image.manifest;
651 /* Word writes never pass req_max: a body that does not fit is refused, never cut. */
652 size_t max = m->has_request ? (m->request.req_max & ~3u) : 0;
653 if (!m->has_request) mesh_reply_errf(take(&c->reply), "%s declares no request window", c->name);
654 else if (c->len > max) mesh_reply_errf(take(&c->reply), "%s: request of %zu bytes > its req_max %zu", c->name, c->len, max);
655 else if (!x->want_run || x->failed) mesh_reply_errf(take(&c->reply), "%s is not running", c->name);
656 else if (x->nreqs >= MAX_QUEUED) mesh_reply_errf(take(&c->reply), "%s: %d requests queued already", c->name, MAX_QUEUED);
657 else {
658 dreq r = {c->bytes, c->len, take(&c->reply)};
659 c->bytes = NULL;
660 PUSH(x->reqs, x->nreqs, x->creqs, r);
661 }
662}
663
664static void cmd_input(mesh_driver *d, mesh_cmd *c) {
665 int s = drv_find(d, c->name);
666 if (s < 0) { err_unknown(&c->reply, c->name); return; }
667 /* Placed programs forward their input to the machine that applies it: not wired (no placer is ever created here). */
668 char *text = NULL;
669 bool ok = drv_input(d, (uint32_t)s, c->bytes, c->len, &text) == 0;
670 drv_answer(take(&c->reply), ok, text);
671 free(text);
672}
673
674static void cmd_view_get(mesh_driver *d, mesh_cmd *c) {
675 int s = drv_find(d, c->name);
676 if (s < 0) { err_unknown(&c->reply, c->name); return; }
677 dslot *x = d->slots[s];
678 if (drv_live(x)) { mesh_reply_errf(take(&c->reply), "%s is running here: a view is taken from a stopped program", c->name); return; }
679 mesh_view_bytes *v = mesh_xcalloc(1, sizeof *v);
680 v->view_abi = x->image.manifest.view_abi;
681 v->len = x->image.manifest.view_bytes;
682 v->bytes = mesh_xmalloc(v->len ? v->len : 1);
683 mesh_mailbox_read(d->sh->mb, x->view_off, v->bytes, v->len);
684 mesh_reply_ptr(take(&c->reply), v, mesh_view_bytes_free);
685}
686
687static void cmd_view_put(mesh_driver *d, mesh_cmd *c) {
688 int s = drv_find(d, c->name);
689 if (s < 0) { err_unknown(&c->reply, c->name); return; }
690 dslot *x = d->slots[s];
691 const mesh_manifest *m = &x->image.manifest;
692 if (drv_live(x)) { mesh_reply_errf(take(&c->reply), "%s is running here: a view is written into a stopped program", c->name); return; }
693 if (m->view_abi != c->view_abi || m->view_bytes != c->len) {
694 mesh_reply_errf(take(&c->reply), "view: the source's is view_abi %u of %zu bytes, this image's view_abi %u of %u bytes", c->view_abi, c->len,
695 m->view_abi, m->view_bytes);
696 return;
697 }
698 mesh_mailbox_write(d->sh->mb, x->view_off, c->bytes, c->len, c->len);
699 /* It holds a live program's state now: no OP_INIT over it. */
700 x->needs_init = false;
701 mesh_reply_okf(take(&c->reply), "%zu bytes", c->len);
702}
703
704static void cmd_job_lease(mesh_driver *d, mesh_cmd *c) {
705 mj_lease_msg *l = &c->lease;
706 int s = drv_find(d, l->program);
707 bool ok = false;
708 if (s >= 0) {
709 dslot *x = d->slots[s];
710 ok = x->image.manifest.policy == MESH_POLICY_JOB && x->want_run && !x->failed && !x->unloading;
711 }
712 if (ok) {
713 dlease q;
714 q.lease = l->lease;
715 q.job = l->job;
716 q.op = l->op;
717 q.setup = l->setup;
718 q.nsetup = l->nsetup;
719 q.items = l->items;
720 memset(l, 0, sizeof *l);
721 djobslot *j = &d->jobw[s];
722 PUSH(j->q, j->n, j->cap, q);
723 } else {
724 char *why = mesh_buf_fmt("%s does not run here", l->program);
725 dlease tmp = {l->lease, l->job, l->op, NULL, 0, l->items};
726 drv_jobs_returned(d, &tmp, MESH_JOBS_RS_RETURNED, why);
727 free(why);
728 }
729}
730
731static void cmd_job_cut(mesh_driver *d, mesh_cmd *c) {
732 for (uint32_t s = 0; s < MESH_PROG_SLOTS; s++) {
733 djobslot *j = &d->jobw[s];
734 size_t k = 0;
735 for (size_t i = 0; i < j->n; i++) {
736 if (j->q[i].lease == c->lease_id) mj_ranges_remove(&j->q[i].items, c->cut, UINT64_MAX);
737 if (mj_ranges_is_empty(&j->q[i].items)) drv_lease_free(&j->q[i]);
738 else j->q[k++] = j->q[i];
739 }
740 j->n = k;
741 }
742}
743
744static void cmd_job_progs(mesh_driver *d, mesh_cmd *c) {
745 mesh_job_progs *p = mesh_xcalloc(1, sizeof *p);
746 p->v = mesh_xcalloc(MESH_PROG_SLOTS, sizeof *p->v);
747 for (uint32_t s = 0; s < MESH_PROG_SLOTS; s++) {
748 dslot *x = d->slots[s];
749 if (!x || x->image.manifest.policy != MESH_POLICY_JOB || !x->want_run || x->failed || x->unloading || x->phase != PH_RUNNING) continue;
750 mj_prog *q = &p->v[p->n++];
751 q->name = mesh_xstrdup(drv_name(x));
752 q->exec = x->image.exec;
753 q->cap = drv_job_cap(&x->image.manifest, x->image.exec);
754 }
755 mesh_reply_ptr(take(&c->reply), p, mesh_job_progs_free);
756}
757
758static void cmd_inventory(mesh_driver *d, mesh_cmd *c) {
759 mesh_actuals *a = mesh_xcalloc(1, sizeof *a);
760 a->v = mesh_xcalloc(MESH_PROG_SLOTS, sizeof *a->v);
761 for (uint32_t s = 0; s < MESH_PROG_SLOTS; s++) {
762 dslot *x = d->slots[s];
763 if (!x || x->unloading) continue;
764 mesh_actual *y = &a->v[a->n++];
765 y->name = mesh_xstrdup(drv_name(x));
766 y->dir = mesh_xstrdup(x->image.dir);
767 y->hash = x->image.hash;
768 y->has_prev = x->has_prev;
769 y->prev = x->has_prev ? x->prev.hash : 0;
770 y->run = x->want_run;
771 y->failed = x->failed != NULL;
772 y->lockstep = x->image.manifest.peer == MESH_PEER_MODE_LOCKSTEP;
773 }
774 mesh_reply_ptr(take(&c->reply), a, mesh_actuals_free);
775}
776
777void drv_program_cmd(mesh_driver *d, mesh_cmd *c) {
778 switch (c->k) {
779 case MESH_CMD_LOAD: {
780 djob j = {false, c->src, c->run, 0, take(&c->reply)};
781 c->has_src = false;
782 memset(&c->src, 0, sizeof c->src);
783 drv_push_job(d, j);
784 break;
785 }
786 case MESH_CMD_UPGRADE: {
787 int s = drv_find(d, c->name);
788 if (s < 0) { err_unknown(&c->reply, c->name); break; }
789 djob j;
790 memset(&j, 0, sizeof j);
791 j.upgrade = true;
792 j.slot = (uint32_t)s;
793 j.reply = take(&c->reply);
794 if (c->has_src) { j.src = c->src; c->has_src = false; memset(&c->src, 0, sizeof c->src); }
795 else j.src = mesh_src_catalog(d->slots[s]->image.dir);
796 drv_push_job(d, j);
797 break;
798 }
799 case MESH_CMD_ROLLBACK: {
800 int s = drv_find(d, c->name);
801 if (s < 0) { err_unknown(&c->reply, c->name); break; }
802 if (!d->slots[s]->has_prev) { mesh_reply_errf(take(&c->reply), "%s: no previous image to roll back to", c->name); break; }
803 dblink b = {OP_ROLLBACK, NULL, -1, (uint32_t)s, take(&c->reply), true, (uint32_t)s, drv_exec_of(d, (uint32_t)s), mesh_mono_ns()};
804 drv_push_blink(d, b);
805 break;
806 }
807 case MESH_CMD_RESET: {
808 int s = drv_find(d, c->name);
809 if (s < 0) { err_unknown(&c->reply, c->name); break; }
810 if (!drv_live(d->slots[s])) {
811 char *t = NULL;
812 bool ok = drv_op_reset(d, (uint32_t)s, &t);
813 if (ok) { free(t); t = mesh_xstrdup("blink 0"); }
814 drv_answer(take(&c->reply), ok, t);
815 free(t);
816 } else {
817 dblink b = {OP_RESET, NULL, -1, (uint32_t)s, take(&c->reply), true, (uint32_t)s, drv_exec_of(d, (uint32_t)s), mesh_mono_ns()};
818 drv_push_blink(d, b);
819 }
820 break;
821 }
822 case MESH_CMD_UNLOAD: cmd_unload(d, c); break;
823 case MESH_CMD_PROG_RUN: cmd_prog_run(d, c); break;
824 case MESH_CMD_TASK: cmd_task(d, c); break;
825 case MESH_CMD_REQ: cmd_req(d, c); break;
826 case MESH_CMD_INPUT: cmd_input(d, c); break;
827 case MESH_CMD_PLACE: {
828 int s = drv_find(d, c->name);
829 if (s < 0) { err_unknown(&c->reply, c->name); break; }
830 /* A placer exists only when peered and the manifest declares `place` (wired with the fleet's placement). */
831 if (!d->cfg.peered) mesh_reply_err(take(&c->reply), "not peered: placement needs a cluster (--cluster, --system)");
832 else mesh_reply_errf(take(&c->reply), "%s declares no `place` in its manifest", c->name);
833 break;
834 }
835 case MESH_CMD_PEER_PLACE:
836 mesh_error("place: F_PLACE for a program from system %u, which is not placeable here; ignored", c->from);
837 break;
838 case MESH_CMD_PEER_FRAME: break;
839 case MESH_CMD_PEER_INPUT:
840 mesh_error("place: %zu byte(s) of input for %s from a member, which does not apply its input here; dropped", c->len, c->name);
841 break;
842 case MESH_CMD_FLEET_VIEW:
843 d->fleet_up = c->up;
844 d->has_fleet_leader = c->has_leader;
845 d->fleet_leader = c->leader;
846 break;
847 case MESH_CMD_INVENTORY: cmd_inventory(d, c); break;
848 case MESH_CMD_VIEW_GET: cmd_view_get(d, c); break;
849 case MESH_CMD_VIEW_PUT: cmd_view_put(d, c); break;
850 case MESH_CMD_JOB_LEASE: cmd_job_lease(d, c); break;
851 case MESH_CMD_JOB_CUT: cmd_job_cut(d, c); break;
852 case MESH_CMD_JOB_PROGS: cmd_job_progs(d, c); break;
853 }
854}