==> 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 /* A placed program's frames hop between the machines: see them promptly. */
294 if (drv_place_busy(d)) return true;
295 for (uint32_t s = 0; s < MESH_PROG_SLOTS; s++) if (d->jobw[s].n || d->jobw[s].inflight) return true;
296 if (d->nblinks || d->working) return true;
297 for (uint32_t s = 0; s < MESH_PROG_SLOTS; s++) {
298 const dslot *x = d->slots[s];
299 if (!x) continue;
300 switch (x->phase) {
301 case PH_STOPPED: break;
302 case PH_RUNNING:
303 if (x->image.manifest.policy == MESH_POLICY_CONTINUOUS || !mesh_pring_idle(&x->ring)) return true;
304 break;
305 default: return true;
306 }
307 }
308 return false;
309}
310
311static void commands(mesh_driver *d) {
312 mesh_cmd c;
313 while (mesh_chan_try_recv(d->sh->cmd, &c) == MESH_CHAN_OK) {
314 drv_program_cmd(d, &c);
315 mesh_cmd_free(&c);
316 }
317}
318
319/* Per member over the last second: exchange RTT, wait per round, rounds received (S_MEMBERS). */
320static void member_stats(mesh_driver *d, float dt, float *rtt_out, uint32_t *cnt_out, float *worst_out) {
321 mesh_rtt_acc rtt[MESH_MAX_SYSTEMS], wait[MESH_MAX_SYSTEMS];
322 pthread_mutex_lock(&d->sh->mu);
323 memcpy(rtt, d->sh->rtt, sizeof rtt);
324 memcpy(wait, d->sh->wait, sizeof wait);
325 memset(d->sh->rtt, 0, sizeof d->sh->rtt);
326 memset(d->sh->wait, 0, sizeof d->sh->wait);
327 pthread_mutex_unlock(&d->sh->mu);
328 mesh_host_block *h = &d->sh->host;
329 float sum = 0, worst = 0;
330 uint32_t cnt = 0;
331 for (uint32_t k = 0; k < d->n && k < MESH_MAX_SYSTEMS; k++) {
332 size_t m = mesh_member(k);
333 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;
334 mesh_host_block_set_f32(h, m + MESH_M_RTT_US, r);
335 mesh_host_block_set_f32(h, m + MESH_M_WAIT_US, w);
336 float per = (float)wait[k].n / (dt > 1e-3f ? dt : 1e-3f);
337 mesh_host_block_set_u32(h, m + MESH_M_ROUNDS, (uint32_t)(per + 0.5f));
338 uint64_t round;
339 if (k == d->me) {
340 uint32_t rank = mesh_mailbox_get(d->sh->mb, MESH_H_GROUP_RANK);
341 round = mesh_mailbox_get64(d->sh->mb, mesh_ex_counter(rank < MESH_MAX_SYSTEMS - 1 ? rank : MESH_MAX_SYSTEMS - 1));
342 } else round = atomic_load(&d->sh->round_of[k]);
343 mesh_host_block_set_u64(h, m + MESH_M_ROUND, round);
344 if (rtt[k].n) { sum += r; cnt++; }
345 if (w > worst) worst = w;
346 }
347 /* Placed programs: the measured time per part (for `split auto`). */
348 drv_place_part_ns(d);
349 *rtt_out = cnt ? sum / (float)cnt : 0.0f;
350 *cnt_out = cnt;
351 *worst_out = worst;
352}
353
354static void stats(mesh_driver *d) {
355 int64_t now = mesh_mono_ns();
356 float dt = (float)((double)(now - d->last_stats) / 1e9);
357 d->last_stats = now;
358 float rtt, wait;
359 uint32_t n;
360 member_stats(d, dt, &rtt, &n, &wait);
361 float total = 0, rates[MESH_PROG_SLOTS] = {0};
362 for (uint32_t s = 0; s < MESH_PROG_SLOTS; s++) {
363 dslot *x = d->slots[s];
364 if (!x) continue;
365 uint32_t done = mesh_mailbox_get(d->sh->mb, mesh_prog_ctl(s) + MESH_P_DONE);
366 x->rate = (float)(done - x->last_done) / dt;
367 x->last_done = done;
368 rates[s] = x->rate;
369 total += x->rate;
370 }
371 mesh_backend_sample(d->gpu);
372 for (size_t i = 0; i < d->ncpus; i++) mesh_backend_sample(d->cpus[i]);
373 mesh_host_block *h = &d->sh->host;
374 uint8_t rb[4 * MESH_PROG_SLOTS];
375 for (uint32_t s = 0; s < MESH_PROG_SLOTS; s++) memcpy(rb + 4 * s, &rates[s], 4);
376 mesh_host_field f = {MESH_S_RATES, rb, sizeof rb};
377 mesh_host_block_set_many(h, &f, 1);
378 mesh_host_block_set_f32(h, MESH_S_TASKS_PER_S, total);
379 mesh_host_block_set_f32(h, MESH_S_PEER_RTT_US, rtt);
380 mesh_host_block_set_f32(h, MESH_S_LOCK_WAIT_US, wait);
381 uint32_t progs = 0;
382 for (uint32_t s = 0; s < MESH_PROG_SLOTS; s++) progs += d->slots[s] != NULL;
383 mesh_host_block_set_u32(h, MESH_S_PROGRAMS, progs);
384 mesh_host_block_set_u32(h, MESH_S_HEAP_FREE, mesh_view_heap_free_pages(&d->heap));
385 mesh_host_block_set_u32(h, MESH_S_HEAP_RUN, mesh_view_heap_largest_run(&d->heap));
386 if (now - d->last_log >= 10000000000ll) {
387 d->last_log = now;
388 mesh_buf p = MESH_BUF_INIT;
389 int first = 1;
390 for (uint32_t s = 0; s < MESH_PROG_SLOTS; s++) {
391 dslot *x = d->slots[s];
392 if (!x) continue;
393 mesh_buf_printf(&p, "%s%s %s ", first ? "" : ", ", drv_name(x), MESH_STATES[drv_state(x)]);
394 mesh_rs_fixed(&p, x->rate, 0);
395 mesh_buf_puts(&p, "/s");
396 first = 0;
397 }
398 mesh_buf line = MESH_BUF_INIT;
399 mesh_rs_fixed(&line, total, 0);
400 mesh_buf_printf(&line, " tasks/s: %s (epoch %u, peer state %u", first ? "no programs" : mesh_buf_str(&p), d->epoch,
401 atomic_load(&d->sh->peer_state));
402 if (n > 0) {
403 mesh_buf_puts(&line, ", exchange rtt ");
404 mesh_rs_fixed(&line, rtt, 0);
405 mesh_buf_puts(&line, " us, slowest member's wait ");
406 mesh_rs_fixed(&line, wait, 0);
407 mesh_buf_puts(&line, " us/round");
408 }
409 mesh_buf_putc(&line, ')');
410 mesh_info("%s", mesh_buf_str(&line));
411 mesh_buf_free(&line);
412 mesh_buf_free(&p);
413 drv_place_log(d);
414 }
415}
416
417/* A native slot's worker words (P_PID .. P_LIMITS) every tick, and its CPU time and RSS for the load sampler. */
418void drv_native_words(mesh_driver *d, uint32_t s);
419void drv_native_words(mesh_driver *d, uint32_t s) {
420 dslot *x = d->slots[s];
421 if (!x || !mesh_kernel_native(x->image.kernel)) return;
422 mesh_native_stats st = mesh_native_worker_stats(mesh_kernel_native(x->image.kernel)), pst = {0};
423 bool prev = x->has_prev && mesh_kernel_native(x->prev.kernel);
424 if (prev) pst = mesh_native_worker_stats(mesh_kernel_native(x->prev.kernel));
425 mesh_mailbox mb = d->sh->mb;
426 size_t c = mesh_prog_ctl(s);
427 mesh_mailbox_set(mb, c + MESH_P_PID, (uint32_t)st.pid);
428 mesh_mailbox_set(mb, c + MESH_P_SPAWNS, st.spawns);
429 mesh_mailbox_set(mb, c + MESH_P_CRASHES, st.crashes);
430 mesh_mailbox_set(mb, c + MESH_P_EXIT, st.exit);
431 mesh_mailbox_set(mb, c + MESH_P_LIMITS, st.limits);
432 /* The slot's CPU counter is monotonic across images: add this image's growth. */
433 uintptr_t key = (uintptr_t)mesh_kernel_native(x->image.kernel);
434 if (d->native_cpu[s].key != key) { d->native_cpu[s].key = key; d->native_cpu[s].cpu = 0; }
435 mesh_slot_use *u = &mesh_cpuuse_slots[s];
436 mesh_slot_use_add_cpu(u, st.cpu_ns > d->native_cpu[s].cpu ? st.cpu_ns - d->native_cpu[s].cpu : 0);
437 if (st.cpu_ns > d->native_cpu[s].cpu) d->native_cpu[s].cpu = st.cpu_ns;
438 atomic_store_explicit(&u->rss, st.rss + (prev ? pst.rss : 0), memory_order_relaxed);
439 atomic_store_explicit(&u->units, (uint32_t)(st.pid != 0) + (uint32_t)(prev && pst.pid != 0), memory_order_relaxed);
440}
441
442static bool all_idle(const mesh_driver *d) {
443 for (uint32_t s = 0; s < MESH_PROG_SLOTS; s++) if (d->slots[s] && !mesh_pring_idle(&d->slots[s]->ring)) return false;
444 return true;
445}
446
447static int shutdown_(mesh_driver *d) {
448 mesh_info("signal: quiescing");
449 mesh_mailbox mb = d->sh->mb;
450 int64_t deadline = mesh_mono_ns() + 400000000ll;
451 while (!all_idle(d) && mesh_mono_ns() < deadline) {
452 for (uint32_t s = 0; s < MESH_PROG_SLOTS; s++) {
453 if (!d->slots[s]) continue;
454 mesh_done *done;
455 size_t n;
456 mesh_pring_collect(&d->slots[s]->ring, mb, &done, &n);
457 for (size_t i = 0; i < n; i++) mesh_reply_drop(done[i].out.kind.reply);
458 free(done);
459 }
460 mesh_sleep_us(TICK_NS / 1000);
461 }
462 uint32_t left[MESH_PROG_SLOTS];
463 size_t nleft = 0;
464 for (uint32_t s = 0; s < MESH_PROG_SLOTS; s++)
465 if (d->slots[s] && drv_live(d->slots[s])) {
466 mesh_mailbox_publish(mb, mesh_prog_ctl(s) + MESH_P_SHUTDOWN, MESH_SHUTDOWN_EXIT);
467 left[nleft++] = s;
468 }
469 deadline = mesh_mono_ns() + 500000000ll;
470 while (nleft && mesh_mono_ns() < deadline) {
471 size_t k = 0;
472 for (size_t i = 0; i < nleft; i++) {
473 bool h = true;
474 char *e = NULL;
475 if (!drv_halted(d, left[i], &h, &e)) { free(e); h = true; }
476 if (!h) left[k++] = left[i];
477 }
478 nleft = k;
479 mesh_sleep_us(100);
480 }
481 if (!d->restoring) save_state(d);
482 if (!nleft) { mesh_info("programs halted, exiting"); return 0; }
483 mesh_buf b = MESH_BUF_INIT;
484 mesh_buf_putc(&b, '[');
485 for (size_t i = 0; i < nleft; i++) mesh_buf_printf(&b, "%s%u", i ? ", " : "", left[i]);
486 mesh_buf_putc(&b, ']');
487 mesh_error("shutdown: programs in slots %s did not halt", mesh_buf_str(&b));
488 mesh_buf_free(&b);
489 return 1;
490}
491
492int mesh_driver_run(mesh_driver *d) {
493 for (;;) {
494 if (atomic_load(d->sh->quit)) return shutdown_(d);
495 int64_t now = mesh_mono_ns();
496 commands(d);
497 drv_programs_tick(d, now);
498 drv_jobs_tick(d);
499 drv_blink_tick(d);
500 if (d->startup_error) {
501 mesh_error("FAILED startup load (%s): --state is left as it was; exiting", d->startup_error);
502 free(d->startup_error);
503 d->startup_error = NULL;
504 shutdown_(d);
505 return 1;
506 }
507 if (mesh_mono_ns() - d->last_stats >= 1000000000ll) stats(d);
508 if (d->restoring && !d->njobs && !d->working && !d->nblinks) {
509 d->restoring = false;
510 d->state_dirty = true;
511 atomic_store(&d->sh->ready, true);
512 size_t n = 0;
513 for (uint32_t s = 0; s < MESH_PROG_SLOTS; s++) n += d->slots[s] != NULL;
514 mesh_info("startup loads done: %zu program(s) loaded", n);
515 }
516 if (!d->restoring && d->state_dirty) save_state(d);
517 mesh_sleep_us((busy(d) ? FAST_TICK_NS : TICK_NS) / 1000);
518 }
519}
520
521void mesh_driver_free(mesh_driver *d) {
522 if (!d) return;
523 drv_place_free_all(d);
524 for (uint32_t s = 0; s < MESH_PROG_SLOTS; s++) {
525 dslot *x = d->slots[s];
526 djobslot *j = &d->jobw[s];
527 for (size_t i = 0; i < j->n; i++) drv_lease_free(&j->q[i]);
528 free(j->q);
529 mesh_setup_state_reset(&j->setup);
530 if (!x) continue;
531 mesh_backend *be = drv_be(d, x->image.exec);
532 char *e = NULL;
533 if (x->image.kernel && !be->ops->unload(be, x->image.kernel, &e)) free(e);
534 x->image.kernel = NULL;
535 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);
536 x->prev.kernel = NULL;
537 drv_image_free(&x->image);
538 if (x->has_prev) drv_image_free(&x->prev);
539 mesh_stream_drop(x->stream);
540 mesh_pring_free(&x->ring);
541 for (size_t i = 0; i < x->ntasks; i++) mesh_reply_drop(x->tasks[i].reply);
542 for (size_t i = 0; i < x->nreqs; i++) { mesh_reply_drop(x->reqs[i].reply); free(x->reqs[i].bytes); }
543 free(x->tasks);
544 free(x->reqs);
545 free(x->failed);
546 free(x);
547 }
548 for (size_t i = 0; i < d->nzombies; i++) {
549 char *e = NULL;
550 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);
551 }
552 free(d->zombies);
553 for (size_t i = 0; i < d->njobs; i++) { mesh_src_free(&d->jobs[i].src); mesh_reply_drop(d->jobs[i].reply); }
554 free(d->jobs);
555 if (d->working) {
556 /* The JIT thread still holds its sender: let go of our end. */
557 mesh_reply_release(d->work_rx);
558 mesh_src_free(&d->work_job.src);
559 mesh_reply_drop(d->work_job.reply);
560 }
561 for (size_t i = 0; i < d->nblinks; i++) {
562 if (d->blinks[i].p) { mesh_prepared_free(d->blinks[i].p); free(d->blinks[i].p); }
563 mesh_reply_drop(d->blinks[i].reply);
564 }
565 free(d->blinks);
566 for (size_t i = 0; i < d->nwaiters; i++) mesh_reply_drop(d->waiters[i].reply);
567 free(d->waiters);
568 if (d->gpu_load) mesh_reply_release(d->gpu_load_rx);
569 free(d->dead);
570 free(d->fault);
571 free(d->startup_error);
572 mesh_backend_destroy(d->gpu);
573 for (size_t i = 0; i < d->ncpus; i++) mesh_backend_destroy(d->cpus[i]);
574 free(d->cpus);
575 free(d);
576}
577
578/* ---------------------------------------------------------------- commands */
579
580static char *unknown(const char *n) {
581 mesh_buf b = MESH_BUF_INIT;
582 mesh_buf_puts(&b, "no program ");
583 mesh_rs_debug_cstr(&b, n);
584 mesh_buf_puts(&b, " (`programs` lists them)");
585 return mesh_buf_take(&b);
586}
587
588static void err_unknown(mesh_reply **reply, const char *n) {
589 char *e = unknown(n);
590 mesh_reply_err(*reply, e);
591 *reply = NULL;
592 free(e);
593}
594
595static mesh_reply *take(mesh_reply **r) {
596 mesh_reply *x = *r;
597 *r = NULL;
598 return x;
599}
600
601static void cmd_unload(mesh_driver *d, mesh_cmd *c) {
602 int s = drv_find(d, c->name);
603 if (s < 0) { err_unknown(&c->reply, c->name); return; }
604 mesh_info("unload %s (slot %d)%s", c->name, s, c->now ? " --now" : "");
605 dslot *x = d->slots[s];
606 x->unloading = true;
607 x->want_run = false;
608 mesh_mailbox_set(d->sh->mb, mesh_prog_ctl((uint32_t)s) + MESH_P_RUN, 0);
609 dwaiter w = {(uint32_t)s, false, true, c->now, take(&c->reply), mesh_mono_ns()};
610 push_waiter(d, w);
611 if (!drv_live(x)) drv_remove(d, (uint32_t)s);
612}
613
614static void cmd_prog_run(mesh_driver *d, mesh_cmd *c) {
615 int s = drv_find(d, c->name);
616 if (s < 0) { err_unknown(&c->reply, c->name); return; }
617 char *why = drv_placed_here(d, (uint32_t)s);
618 if (why) { mesh_reply_err(take(&c->reply), why); free(why); return; }
619 /* P_RUN is the host's word; GPU requests come through the GPU-owned P_RUN_REQ. */
620 mesh_mailbox_set(d->sh->mb, mesh_prog_ctl((uint32_t)s) + MESH_P_RUN, c->on);
621 dslot *x = d->slots[s];
622 if (d->fault && drv_exec_of(d, (uint32_t)s) == drv_gpu_kind(d)) {
623 mesh_reply_errf(take(&c->reply), "the CUDA context is lost (%s); restart meshd", d->fault);
624 return;
625 }
626 if (c->on && x->failed) {
627 mesh_info("start %s: clearing FAILED (%s)", c->name, x->failed);
628 free(x->failed);
629 x->failed = NULL;
630 }
631 if (x->want_run != c->on) mesh_info("%s %s", c->on ? "start" : "stop", c->name);
632 x->want_run = c->on;
633 bool done = c->on ? x->phase == PH_RUNNING : x->phase == PH_STOPPED;
634 if (done) mesh_reply_ok(take(&c->reply), "0 us");
635 else {
636 dwaiter w = {(uint32_t)s, c->on, false, false, take(&c->reply), mesh_mono_ns()};
637 push_waiter(d, w);
638 }
639 d->state_dirty = true;
640}
641
642static void cmd_task(mesh_driver *d, mesh_cmd *c) {
643 int s = drv_find(d, c->name);
644 if (s < 0) { err_unknown(&c->reply, c->name); return; }
645 dslot *x = d->slots[s];
646 if (!x->want_run || x->failed) mesh_reply_errf(take(&c->reply), "%s is not running", c->name);
647 else if (x->ntasks >= MAX_QUEUED) mesh_reply_errf(take(&c->reply), "%s: %d tasks queued already", c->name, MAX_QUEUED);
648 else {
649 dtask t = {c->op, c->arg, take(&c->reply)};
650 PUSH(x->tasks, x->ntasks, x->ctasks, t);
651 }
652}
653
654static void cmd_req(mesh_driver *d, mesh_cmd *c) {
655 int s = drv_find(d, c->name);
656 if (s < 0) { err_unknown(&c->reply, c->name); return; }
657 dslot *x = d->slots[s];
658 const mesh_manifest *m = &x->image.manifest;
659 /* Word writes never pass req_max: a body that does not fit is refused, never cut. */
660 size_t max = m->has_request ? (m->request.req_max & ~3u) : 0;
661 if (!m->has_request) mesh_reply_errf(take(&c->reply), "%s declares no request window", c->name);
662 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);
663 else if (!x->want_run || x->failed) mesh_reply_errf(take(&c->reply), "%s is not running", c->name);
664 else if (x->nreqs >= MAX_QUEUED) mesh_reply_errf(take(&c->reply), "%s: %d requests queued already", c->name, MAX_QUEUED);
665 else {
666 dreq r = {c->bytes, c->len, take(&c->reply)};
667 c->bytes = NULL;
668 PUSH(x->reqs, x->nreqs, x->creqs, r);
669 }
670}
671
672static void cmd_input(mesh_driver *d, mesh_cmd *c) {
673 int s = drv_find(d, c->name);
674 if (s < 0) { err_unknown(&c->reply, c->name); return; }
675 /* Placed: the input goes to the machine that applies it (pin k: k; split: the owner). */
676 uint32_t to = drv_input_to(d, (uint32_t)s);
677 char *text = NULL;
678 bool ok;
679 if (to == d->me) ok = drv_input(d, (uint32_t)s, c->bytes, c->len, &text) == 0;
680 else if (c->len > MESH_F_INPUT_MAX) ok = false, text = mesh_buf_fmt("%s: at most %d bytes forwarded at once", c->name, MESH_F_INPUT_MAX);
681 else {
682 mpw_bytes f = mpw_input_encode(c->name, c->bytes, c->len);
683 ok = mesh_shared_send_peer(d->sh, to, f.b, f.n);
684 mpw_bytes_free(&f);
685 text = ok ? mesh_buf_fmt("%zu bytes to system %u", c->len, to) : mesh_buf_fmt("peer: the cluster is not up; %s's input goes to system %u", c->name, to);
686 }
687 drv_answer(take(&c->reply), ok, text);
688 free(text);
689}
690
691static void cmd_view_get(mesh_driver *d, mesh_cmd *c) {
692 int s = drv_find(d, c->name);
693 if (s < 0) { err_unknown(&c->reply, c->name); return; }
694 dslot *x = d->slots[s];
695 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; }
696 mesh_view_bytes *v = mesh_xcalloc(1, sizeof *v);
697 v->view_abi = x->image.manifest.view_abi;
698 v->len = x->image.manifest.view_bytes;
699 v->bytes = mesh_xmalloc(v->len ? v->len : 1);
700 mesh_mailbox_read(d->sh->mb, x->view_off, v->bytes, v->len);
701 mesh_reply_ptr(take(&c->reply), v, mesh_view_bytes_free);
702}
703
704static void cmd_view_put(mesh_driver *d, mesh_cmd *c) {
705 int s = drv_find(d, c->name);
706 if (s < 0) { err_unknown(&c->reply, c->name); return; }
707 dslot *x = d->slots[s];
708 const mesh_manifest *m = &x->image.manifest;
709 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; }
710 if (m->view_abi != c->view_abi || m->view_bytes != c->len) {
711 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,
712 m->view_abi, m->view_bytes);
713 return;
714 }
715 mesh_mailbox_write(d->sh->mb, x->view_off, c->bytes, c->len, c->len);
716 /* It holds a live program's state now: no OP_INIT over it. */
717 x->needs_init = false;
718 mesh_reply_okf(take(&c->reply), "%zu bytes", c->len);
719}
720
721static void cmd_job_lease(mesh_driver *d, mesh_cmd *c) {
722 mj_lease_msg *l = &c->lease;
723 int s = drv_find(d, l->program);
724 bool ok = false;
725 if (s >= 0) {
726 dslot *x = d->slots[s];
727 ok = x->image.manifest.policy == MESH_POLICY_JOB && x->want_run && !x->failed && !x->unloading;
728 }
729 if (ok) {
730 dlease q;
731 q.lease = l->lease;
732 q.job = l->job;
733 q.op = l->op;
734 q.setup = l->setup;
735 q.nsetup = l->nsetup;
736 q.items = l->items;
737 free(l->program); /* the queue keeps no name: the slot is the program */
738 memset(l, 0, sizeof *l);
739 djobslot *j = &d->jobw[s];
740 PUSH(j->q, j->n, j->cap, q);
741 } else {
742 char *why = mesh_buf_fmt("%s does not run here", l->program);
743 dlease tmp = {l->lease, l->job, l->op, NULL, 0, l->items};
744 drv_jobs_returned(d, &tmp, MESH_JOBS_RS_RETURNED, why);
745 free(why);
746 }
747}
748
749static void cmd_job_cut(mesh_driver *d, mesh_cmd *c) {
750 for (uint32_t s = 0; s < MESH_PROG_SLOTS; s++) {
751 djobslot *j = &d->jobw[s];
752 size_t k = 0;
753 for (size_t i = 0; i < j->n; i++) {
754 if (j->q[i].lease == c->lease_id) mj_ranges_remove(&j->q[i].items, c->cut, UINT64_MAX);
755 if (mj_ranges_is_empty(&j->q[i].items)) drv_lease_free(&j->q[i]);
756 else j->q[k++] = j->q[i];
757 }
758 j->n = k;
759 }
760}
761
762static void cmd_job_progs(mesh_driver *d, mesh_cmd *c) {
763 mesh_job_progs *p = mesh_xcalloc(1, sizeof *p);
764 p->v = mesh_xcalloc(MESH_PROG_SLOTS, sizeof *p->v);
765 for (uint32_t s = 0; s < MESH_PROG_SLOTS; s++) {
766 dslot *x = d->slots[s];
767 if (!x || x->image.manifest.policy != MESH_POLICY_JOB || !x->want_run || x->failed || x->unloading || x->phase != PH_RUNNING) continue;
768 mj_prog *q = &p->v[p->n++];
769 q->name = mesh_xstrdup(drv_name(x));
770 q->exec = x->image.exec;
771 q->cap = drv_job_cap(&x->image.manifest, x->image.exec);
772 }
773 mesh_reply_ptr(take(&c->reply), p, mesh_job_progs_free);
774}
775
776static void cmd_inventory(mesh_driver *d, mesh_cmd *c) {
777 mesh_actuals *a = mesh_xcalloc(1, sizeof *a);
778 a->v = mesh_xcalloc(MESH_PROG_SLOTS, sizeof *a->v);
779 for (uint32_t s = 0; s < MESH_PROG_SLOTS; s++) {
780 dslot *x = d->slots[s];
781 if (!x || x->unloading) continue;
782 mesh_actual *y = &a->v[a->n++];
783 y->name = mesh_xstrdup(drv_name(x));
784 y->dir = mesh_xstrdup(x->image.dir);
785 y->hash = x->image.hash;
786 y->has_prev = x->has_prev;
787 y->prev = x->has_prev ? x->prev.hash : 0;
788 y->run = x->want_run || drv_place_stopped(d, drv_name(x));
789 y->failed = x->failed != NULL;
790 y->lockstep = x->image.manifest.peer == MESH_PEER_MODE_LOCKSTEP;
791 }
792 mesh_reply_ptr(take(&c->reply), a, mesh_actuals_free);
793}
794
795void drv_program_cmd(mesh_driver *d, mesh_cmd *c) {
796 switch (c->k) {
797 case MESH_CMD_LOAD: {
798 djob j = {false, c->src, c->run, 0, take(&c->reply)};
799 c->has_src = false;
800 memset(&c->src, 0, sizeof c->src);
801 drv_push_job(d, j);
802 break;
803 }
804 case MESH_CMD_UPGRADE: {
805 int s = drv_find(d, c->name);
806 if (s < 0) { err_unknown(&c->reply, c->name); break; }
807 djob j;
808 memset(&j, 0, sizeof j);
809 j.upgrade = true;
810 j.slot = (uint32_t)s;
811 j.reply = take(&c->reply);
812 if (c->has_src) { j.src = c->src; c->has_src = false; memset(&c->src, 0, sizeof c->src); }
813 else j.src = mesh_src_catalog(d->slots[s]->image.dir);
814 drv_push_job(d, j);
815 break;
816 }
817 case MESH_CMD_ROLLBACK: {
818 int s = drv_find(d, c->name);
819 if (s < 0) { err_unknown(&c->reply, c->name); break; }
820 if (!d->slots[s]->has_prev) { mesh_reply_errf(take(&c->reply), "%s: no previous image to roll back to", c->name); break; }
821 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()};
822 drv_push_blink(d, b);
823 break;
824 }
825 case MESH_CMD_RESET: {
826 int s = drv_find(d, c->name);
827 if (s < 0) { err_unknown(&c->reply, c->name); break; }
828 char *why = drv_placed_here(d, (uint32_t)s);
829 if (why) { mesh_reply_err(take(&c->reply), why); free(why); break; }
830 if (!drv_live(d->slots[s])) {
831 char *t = NULL;
832 bool ok = drv_op_reset(d, (uint32_t)s, &t);
833 if (ok) { free(t); t = mesh_xstrdup("blink 0"); }
834 drv_answer(take(&c->reply), ok, t);
835 free(t);
836 } else {
837 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()};
838 drv_push_blink(d, b);
839 }
840 break;
841 }
842 case MESH_CMD_UNLOAD: cmd_unload(d, c); break;
843 case MESH_CMD_PROG_RUN: cmd_prog_run(d, c); break;
844 case MESH_CMD_TASK: cmd_task(d, c); break;
845 case MESH_CMD_REQ: cmd_req(d, c); break;
846 case MESH_CMD_INPUT: cmd_input(d, c); break;
847 case MESH_CMD_PLACE:
848 if (drv_find(d, c->name) < 0) err_unknown(&c->reply, c->name);
849 else drv_cmd_place(d, c, take(&c->reply));
850 break;
851 case MESH_CMD_PEER_PLACE: drv_cmd_peer_place(d, c); break;
852 case MESH_CMD_PEER_FRAME: drv_cmd_peer_frame(d, c); break;
853 case MESH_CMD_PEER_INPUT: drv_cmd_peer_input(d, c); break;
854 case MESH_CMD_FLEET_VIEW: drv_cmd_fleet_view(d, c); break;
855 case MESH_CMD_INVENTORY: cmd_inventory(d, c); break;
856 case MESH_CMD_VIEW_GET: cmd_view_get(d, c); break;
857 case MESH_CMD_VIEW_PUT: cmd_view_put(d, c); break;
858 case MESH_CMD_JOB_LEASE: cmd_job_lease(d, c); break;
859 case MESH_CMD_JOB_CUT: cmd_job_cut(d, c); break;
860 case MESH_CMD_JOB_PROGS: cmd_job_progs(d, c); break;
861 }
862}