==> src/app.c <==1/* crates/mesh-top/src/app.rs, ported. */
2#include "mesh/top_app.h"
3
4#include "pathtrace/view.h"
5#include "tui/utf8.h"
6
7#include <math.h>
8#include <stdarg.h>
9#include <stdio.h>
10#include <stdlib.h>
11#include <string.h>
12
13const char *const MT_TAB_NAMES[MT_TABS] = {"Mesh", "Programs", "Doom", "Train", "Infer", "HTTP",
14 "Net", "Load", "Fleet", "Jobs", "Pipes", "Render"};
15const char MT_TAB_KEYS[MT_TABS] = {'1', '2', '3', '4', '5', '6', '7', '8', '9', '0', '-', '='};
16const char *const MT_VIEWS[3] = {"fleet", "jobs", "pipes"};
17const char *const MT_RENDER_PROGRAMS[2] = {"pathtrace", "pathtrace-live"};
18
19static void *xmalloc(size_t n) {
20 void *p = malloc(n ? n : 1);
21 if (!p) abort();
22 return p;
23}
24static void *xrealloc(void *p, size_t n) {
25 p = realloc(p, n ? n : 1);
26 if (!p) abort();
27 return p;
28}
29static char *xstrdup(const char *s) {
30 char *d = xmalloc(strlen(s) + 1);
31 strcpy(d, s);
32 return d;
33}
34static char *xfmt(const char *f, ...)
35#if defined(__GNUC__) || defined(__clang__)
36 __attribute__((format(printf, 1, 2)))
37#endif
38 ;
39static char *xfmt(const char *f, ...) {
40 va_list ap, cp;
41 va_start(ap, f);
42 va_copy(cp, ap);
43 int n = vsnprintf(NULL, 0, f, cp);
44 va_end(cp);
45 char *s = xmalloc((size_t)n + 1);
46 vsnprintf(s, (size_t)n + 1, f, ap);
47 va_end(ap);
48 return s;
49}
50static void set_str(char **dst, char *s) {
51 free(*dst);
52 *dst = s;
53}
54
55/* ---- Rate */
56
57void mt_rate_push(mt_rate *r, mt_instant now, uint64_t v) {
58 if (r->n && v < r->v[r->n - 1]) r->n = 0; /* a reset or reload: a new count */
59 if (r->n == r->cap) {
60 r->cap = r->cap ? r->cap * 2 : 8;
61 r->t = xrealloc(r->t, r->cap * sizeof *r->t);
62 r->v = xrealloc(r->v, r->cap * sizeof *r->v);
63 }
64 r->t[r->n] = now;
65 r->v[r->n] = v;
66 r->n++;
67 while (r->n > 2 && mt_since(now, r->t[1]) >= MT_RATE_WINDOW) {
68 memmove(r->t, r->t + 1, (r->n - 1) * sizeof *r->t);
69 memmove(r->v, r->v + 1, (r->n - 1) * sizeof *r->v);
70 r->n--;
71 }
72}
73
74float mt_rate_per_s(const mt_rate *r) {
75 if (r->n == 0) return 0.0f;
76 mt_instant at = r->t[0], bt = r->t[r->n - 1];
77 if (!(bt > at)) return 0.0f;
78 return (float)(r->v[r->n - 1] - r->v[0]) / mt_secs_f32(mt_since(bt, at));
79}
80
81void mt_rate_free(mt_rate *r) {
82 free(r->t);
83 free(r->v);
84 *r = (mt_rate){NULL, NULL, 0, 0};
85}
86
87/* ---- FleetImage */
88
89static void kv_set(mt_fleet_image *f, const char *k, const char *v) {
90 for (size_t i = 0; i < f->nkv; i++)
91 if (strcmp(f->kv[i].key, k) == 0) {
92 set_str(&f->kv[i].value, xstrdup(v));
93 return;
94 }
95 f->kv = xrealloc(f->kv, (f->nkv + 1) * sizeof *f->kv);
96 f->kv[f->nkv++] = (mt_kv){xstrdup(k), xstrdup(v)};
97}
98
99void mt_fleet_image_free(mt_fleet_image *f) {
100 for (size_t i = 0; i < f->nkv; i++) {
101 free(f->kv[i].key);
102 free(f->kv[i].value);
103 }
104 free(f->kv);
105 free(f->members);
106 free(f->tiles);
107 free(f->px);
108 memset(f, 0, sizeof *f);
109}
110
111static size_t split_ws(char *s, char **w, size_t cap) {
112 size_t n = 0;
113 char *save = NULL;
114 for (char *t = strtok_r(s, " \t\n\r\f\v", &save); t && n < cap; t = strtok_r(NULL, " \t\n\r\f\v", &save)) w[n++] = t;
115 return n;
116}
117
118bool mt_fleet_image_parse(const uint8_t *b, size_t len, mt_fleet_image *out) {
119 memset(out, 0, sizeof *out);
120 const uint8_t *z = memchr(b, 0, len);
121 if (!z) return false;
122 char *text = tui_utf8_lossy(b, (size_t)(z - b), NULL);
123 bool ok = true;
124 for (char *line = text; line && *line && ok;) {
125 char *nl = strchr(line, '\n');
126 if (nl) *nl = 0;
127 char *w[64];
128 size_t n = split_ws(line, w, 64);
129 if (n && (strcmp(w[0], "render") == 0 || strcmp(w[0], "digests") == 0)) {
130 for (size_t i = 1; i + 1 < n; i += 2) kv_set(out, w[i], w[i + 1]); /* chunks(2); an odd last word is dropped */
131 } else if (n >= 6 && strcmp(w[0], "member") == 0) {
132 uint64_t k, folded, tiles;
133 if (!mt_parse_u64(w[1], &k) || k > UINT32_MAX || !mt_parse_u64(w[3], &folded) || !mt_parse_u64(w[5], &tiles) ||
134 tiles > UINT32_MAX) {
135 ok = false;
136 break;
137 }
138 out->members = xrealloc(out->members, (out->nmembers + 1) * sizeof *out->members);
139 out->members[out->nmembers].k = (uint32_t)k;
140 out->members[out->nmembers].folded = folded;
141 out->members[out->nmembers].tiles = (uint32_t)tiles;
142 out->nmembers++;
143 }
144 line = nl ? nl + 1 : NULL;
145 }
146 free(text);
147 const uint8_t *rest = z + 1;
148 size_t rl = len - (size_t)(z + 1 - b);
149 if (!ok || rl < 4) {
150 mt_fleet_image_free(out);
151 return false;
152 }
153 size_t n = mt_get_u32(rest, rl, 0);
154 if (rl < 4 + 8 * n) {
155 mt_fleet_image_free(out);
156 return false;
157 }
158 out->tiles = xmalloc(n * sizeof *out->tiles);
159 out->ntiles = n;
160 for (size_t t = 0; t < n; t++) {
161 out->tiles[t][0] = mt_get_u32(rest, rl, 4 + 4 * (2 * t));
162 out->tiles[t][1] = mt_get_u32(rest, rl, 4 + 4 * (2 * t + 1));
163 }
164 out->npx = rl - 4 - 8 * n;
165 out->px = xmalloc(out->npx);
166 memcpy(out->px, rest + 4 + 8 * n, out->npx);
167 if (out->npx != n * 4 * PT_TILE_PIX) {
168 mt_fleet_image_free(out);
169 return false;
170 }
171 return true;
172}
173
174const char *mt_fleet_image_get(const mt_fleet_image *f, const char *k) {
175 for (size_t i = 0; i < f->nkv; i++)
176 if (strcmp(f->kv[i].key, k) == 0) return f->kv[i].value;
177 return "";
178}
179
180uint64_t mt_fleet_image_num(const mt_fleet_image *f, const char *k) {
181 uint64_t v;
182 return mt_parse_u64(mt_fleet_image_get(f, k), &v) ? v : 0;
183}
184
185/* ---- train */
186
187uint64_t mt_train_next_step(const mt_section *s) {
188 return (uint64_t)mt_section_view_u32(s, MESH_TRAIN_T_STEP_HI) << 32 | mt_section_view_u32(s, MESH_TRAIN_T_NEXT_STEP);
189}
190
191size_t mt_train_losses(const mt_section *s, mt_loss out[256]) {
192 uint64_t next = mt_train_next_step(s), from = next > 256 ? next - 256 : 0;
193 size_t n = 0;
194 for (uint64_t t = from; t < next; t++)
195 out[n++] = (mt_loss){t, mt_section_view_f32(s, MESH_TRAIN_T_LOSS + 4 * (size_t)(t % 256))};
196 return n;
197}
198
199/* ---- Machine */
200
201const char *mt_machine_name(const mt_machine *m, char *out, size_t cap) {
202 if (m->has_snap) {
203 char h[97];
204 mt_snapshot_hostname(&m->snap, h);
205 if (*h) {
206 snprintf(out, cap, "%s", h);
207 return out;
208 }
209 }
210 snprintf(out, cap, "%s", m->endpoint);
211 return out;
212}
213
214bool mt_machine_system(const mt_machine *m, uint32_t *k) { return m->has_snap && mt_snapshot_system(&m->snap, k); }
215
216const mt_section *mt_machine_program(const mt_machine *m, const char *name) {
217 return m->has_snap ? mt_snapshot_by_name(&m->snap, name) : NULL;
218}
219
220static mt_prog_track *prog_track(const mt_machine *m, uint32_t slot) {
221 for (size_t i = 0; i < m->nprogs; i++)
222 if (m->progs[i].slot == slot) return &m->progs[i];
223 return NULL;
224}
225
226float mt_machine_counter_rate(const mt_machine *m, uint32_t slot, size_t i) {
227 mt_prog_track *p = prog_track(m, slot);
228 return p ? mt_rate_per_s(&p->counters[i]) : 0.0f;
229}
230
231const mt_node_track *mt_machine_node(const mt_machine *m, uint32_t slot, uint32_t node) {
232 for (size_t i = 0; i < m->nnodes; i++)
233 if (m->nodes[i].slot == slot && m->nodes[i].node == node) return &m->nodes[i];
234 return NULL;
235}
236
237bool mt_machine_stalled(const mt_machine *m, uint32_t slot, uint32_t node, mt_instant now) {
238 const mt_node_track *t = mt_machine_node(m, slot, node);
239 return t && mt_since(now, t->changed) > MT_STALL;
240}
241
242static int64_t lead(uint32_t e, uint32_t base) { return (int64_t)(int32_t)(e - base); }
243
244size_t mt_machine_flame(const mt_machine *m, mt_instant now, uint32_t window_ns, mt_flame_rec **out) {
245 *out = NULL;
246 size_t nf = 0;
247 const mt_trace_track **fresh = xmalloc((m->ntrace ? m->ntrace : 1) * sizeof *fresh);
248 for (size_t i = 0; i < m->ntrace; i++)
249 if (mt_since(now, m->trace[i].moved) <= MT_TRACE_STALE && m->trace[i].nrecs) fresh[nf++] = &m->trace[i];
250 if (nf == 0) {
251 free(fresh);
252 return 0;
253 }
254 uint32_t base = fresh[0]->recs[fresh[0]->nrecs - 1].end;
255 int64_t newest = INT64_MIN;
256 for (size_t i = 0; i < nf; i++) {
257 int64_t l = lead(fresh[i]->recs[fresh[i]->nrecs - 1].end, base);
258 if (l > newest) newest = l;
259 }
260 size_t n = 0, cap = 0;
261 mt_flame_rec *o = NULL;
262 for (size_t i = 0; i < nf; i++) {
263 const mt_trace_track *t = fresh[i];
264 int64_t age = newest - lead(t->recs[t->nrecs - 1].end, base);
265 uint32_t next_end = t->recs[t->nrecs - 1].end;
266 for (size_t k = t->nrecs; k-- > 0;) {
267 const mt_rec *r = &t->recs[k];
268 age += (int64_t)(int32_t)(next_end - r->end);
269 next_end = r->end;
270 if (age > (int64_t)window_ns + 50000000) break;
271 if (age >= 0 && age < (int64_t)window_ns) {
272 if (n == cap) {
273 cap = cap ? cap * 2 : 64;
274 o = xrealloc(o, cap * sizeof *o);
275 }
276 o[n++] = (mt_flame_rec){*r, (uint32_t)age};
277 }
278 }
279 }
280 free(fresh);
281 *out = o;
282 return n;
283}
284
285size_t mt_machine_fresh_trace(const mt_machine *m, mt_instant now, mt_rec **out) {
286 size_t n = 0;
287 for (size_t i = 0; i < m->ntrace; i++)
288 if (mt_since(now, m->trace[i].moved) <= MT_TRACE_STALE) n += m->trace[i].nrecs;
289 mt_rec *o = xmalloc(n * sizeof *o);
290 size_t k = 0;
291 for (size_t i = 0; i < m->ntrace; i++)
292 if (mt_since(now, m->trace[i].moved) <= MT_TRACE_STALE) {
293 if (m->trace[i].nrecs) memcpy(o + k, m->trace[i].recs, m->trace[i].nrecs * sizeof *o);
294 k += m->trace[i].nrecs;
295 }
296 *out = o;
297 return n;
298}
299
300static void machine_init(mt_machine *m, const char *endpoint) {
301 memset(m, 0, sizeof *m);
302 m->endpoint = xstrdup(endpoint);
303 m->status = xstrdup("connecting");
304}
305
306static void trace_clear(mt_machine *m) {
307 for (size_t i = 0; i < m->ntrace; i++) free(m->trace[i].recs);
308 free(m->trace);
309 m->trace = NULL;
310 m->ntrace = 0;
311}
312
313static void strv_free(char **v, size_t n) {
314 for (size_t i = 0; i < n; i++) free(v[i]);
315 free(v);
316}
317
318static void prog_free(mt_prog_track *p) {
319 for (size_t i = 0; i < MESH_COUNTERS; i++) mt_rate_free(&p->counters[i]);
320}
321
322static void machine_free(mt_machine *m) {
323 free(m->endpoint);
324 free(m->status);
325 if (m->has_snap) mt_snapshot_free(&m->snap);
326 free(m->nodes);
327 for (size_t i = 0; i < m->nprogs; i++) prog_free(&m->progs[i]);
328 free(m->progs);
329 free(m->tps);
330 free(m->params);
331 free(m->losses);
332 mt_rate_free(&m->train_steps);
333 mt_rate_free(&m->doom_shown);
334 free(m->http_resp);
335 trace_clear(m);
336 free(m->trace_err);
337 strv_free(m->log, m->nlog);
338 free(m->log_err);
339 strv_free(m->catalog, m->ncatalog);
340 mt_net_free(&m->net);
341 free(m->load_hist);
342 free(m->render_name);
343 free(m->render_px);
344 free(m->render_err);
345 if (m->has_fleet) mt_fleet_image_free(&m->fleet);
346}
347
348static void record_lock(mt_machine *m, const mt_snapshot *s) {
349 char name[3 * MESH_NAME_BYTES + 1];
350 uint64_t step;
351 uint8_t view[32];
352 mt_snapshot_lock_pair(s, name, &step, view);
353 if (!*name) return;
354 if (m->nparams && step < m->params[m->nparams - 1].step) m->nparams = 0; /* a reset: another run */
355 size_t i = 0;
356 while (i < m->nparams && m->params[i].step < step) i++;
357 if (i < m->nparams && m->params[i].step == step) {
358 memcpy(m->params[i].view, view, 32);
359 } else {
360 m->params = xrealloc(m->params, (m->nparams + 1) * sizeof *m->params);
361 memmove(&m->params[i + 1], &m->params[i], (m->nparams - i) * sizeof *m->params);
362 m->params[i].step = step;
363 memcpy(m->params[i].view, view, 32);
364 m->nparams++;
365 }
366 while (m->nparams > MT_PARAM_HISTORY) {
367 memmove(m->params, m->params + 1, (m->nparams - 1) * sizeof *m->params);
368 m->nparams--;
369 }
370}
371
372static void loss_push(mt_machine *m, mt_loss l) {
373 if (m->nlosses == m->caplosses) {
374 m->caplosses = m->caplosses ? m->caplosses * 2 : 512;
375 m->losses = xrealloc(m->losses, m->caplosses * sizeof *m->losses);
376 }
377 m->losses[m->nlosses++] = l;
378}
379
380static void record_losses(mt_machine *m, const mt_section *t) {
381 mt_loss ring[256];
382 size_t nr = mt_train_losses(t, ring);
383 uint64_t next = mt_train_next_step(t);
384 if (m->nlosses && m->losses[m->nlosses - 1].step >= next) m->nlosses = 0;
385 bool has_after = m->nlosses > 0;
386 uint64_t after = has_after ? m->losses[m->nlosses - 1].step : 0;
387 for (size_t i = 0; i < nr; i++)
388 if (!has_after || ring[i].step > after) loss_push(m, ring[i]);
389 if (m->nlosses > MT_LOSS_POINTS) {
390 mt_loss last = m->losses[m->nlosses - 1];
391 size_t w = 0;
392 for (size_t i = 0; i < m->nlosses; i += 2) m->losses[w++] = m->losses[i];
393 m->nlosses = w;
394 mt_loss l = m->losses[m->nlosses - 1];
395 if (!(l.step == last.step && l.loss == last.loss)) loss_push(m, last);
396 }
397}
398
399static void apply_snap(mt_machine *m, mt_snapshot s, mt_instant now) {
400 double dt_ns = m->has_snap_at ? (double)mt_since(now, m->snap_at) : 0.0;
401 size_t total_nodes = 0;
402 for (size_t i = 0; i < s.nsections; i++) total_nodes += mt_section_nodes(&s.sections[i]);
403 mt_node_track *nodes = xmalloc(total_nodes * sizeof *nodes);
404 mt_prog_track *progs = xmalloc(s.nsections * sizeof *progs);
405 size_t nn = 0;
406 for (size_t si = 0; si < s.nsections; si++) {
407 const mt_section *sec = &s.sections[si];
408 uint32_t slot = mt_section_slot(sec), loads = mt_section_pt(sec, MESH_PT_LOADS);
409 char name[3 * MESH_NAME_BYTES + 1];
410 mt_section_name(sec, name);
411 const mt_section *prev = NULL;
412 if (m->has_snap) {
413 const mt_section *p = mt_snapshot_by_slot(&m->snap, slot);
414 if (p && mt_section_name_is(p, name) && mt_section_pt(p, MESH_PT_LOADS) == loads &&
415 mt_section_nodes(p) == mt_section_nodes(sec))
416 prev = p;
417 }
418 for (size_t n = 0; n < mt_section_nodes(sec); n++) {
419 mt_node b = mt_section_node(sec, n);
420 const mt_node_track *t = mt_machine_node(m, slot, (uint32_t)n);
421 mt_node_track nt = {slot, (uint32_t)n, b.heartbeat, now, 0.0f};
422 if (prev && t) {
423 mt_node a = mt_section_node(prev, n);
424 float util = t->util;
425 if (dt_ns > 0.0) {
426 double u = (double)(b.busy_ns - a.busy_ns) / dt_ns;
427 u = u < 0.0 ? 0.0 : u > 1.0 ? 1.0 : u;
428 util = (float)u;
429 }
430 nt.changed = b.heartbeat != t->heartbeat ? now : t->changed;
431 nt.util = util;
432 }
433 nodes[nn++] = nt;
434 }
435 mt_prog_track *old = prog_track(m, slot);
436 mt_prog_track pt;
437 if (old && strcmp(old->name, name) == 0 && old->loads == loads) {
438 pt = *old;
439 memset(old, 0, sizeof *old); /* moved */
440 old->slot = UINT32_MAX;
441 } else {
442 memset(&pt, 0, sizeof pt);
443 pt.slot = slot;
444 snprintf(pt.name, sizeof pt.name, "%s", name);
445 pt.loads = loads;
446 }
447 pt.slot = slot;
448 for (size_t i = 0; i < MESH_COUNTERS; i++) mt_rate_push(&pt.counters[i], now, mt_section_counter(sec, i));
449 progs[si] = pt;
450 }
451 /* Nodes sorted by (slot, node), as the BTreeMap keeps them. */
452 for (size_t i = 1; i < nn; i++)
453 for (size_t j = i; j > 0 && (nodes[j - 1].slot > nodes[j].slot ||
454 (nodes[j - 1].slot == nodes[j].slot && nodes[j - 1].node > nodes[j].node));
455 j--) {
456 mt_node_track t = nodes[j];
457 nodes[j] = nodes[j - 1];
458 nodes[j - 1] = t;
459 }
460 free(m->nodes);
461 m->nodes = nodes;
462 m->nnodes = nn;
463 for (size_t i = 0; i < m->nprogs; i++) prog_free(&m->progs[i]);
464 free(m->progs);
465 m->progs = progs;
466 m->nprogs = s.nsections;
467 /* The trace keeps only slots still present. */
468 size_t w = 0;
469 for (size_t i = 0; i < m->ntrace; i++) {
470 if (mt_snapshot_by_slot(&s, m->trace[i].slot)) m->trace[w++] = m->trace[i];
471 else free(m->trace[i].recs);
472 }
473 m->ntrace = w;
474 float tps = mt_snapshot_host_f32(&s, MESH_S_TASKS_PER_S);
475 uint64_t tv = isfinite(tps) && tps > 0.0f ? (tps >= 1.8446744e19f ? UINT64_MAX : (uint64_t)tps) : 0;
476 if (m->ntps == MT_SPARK_LEN) {
477 memmove(m->tps, m->tps + 1, (MT_SPARK_LEN - 1) * sizeof *m->tps);
478 m->ntps--;
479 }
480 if (!m->tps) m->tps = xmalloc(MT_SPARK_LEN * sizeof *m->tps);
481 m->tps[m->ntps++] = tv;
482 record_lock(m, &s);
483 mt_net_update(&m->net, &s);
484 mt_load l;
485 if (mt_snapshot_load(&s, &l) && l.samples != m->load_seen) {
486 m->load_seen = l.samples;
487 if (!m->load_hist) m->load_hist = xmalloc(MT_SPARK_LEN * sizeof *m->load_hist);
488 if (m->nload_hist == MT_SPARK_LEN) {
489 memmove(m->load_hist, m->load_hist + 1, (MT_SPARK_LEN - 1) * sizeof *m->load_hist);
490 m->nload_hist--;
491 }
492#define PCT(x) ((uint64_t)fminf(fmaxf(roundf((x) * 100.0f), 0.0f), 100.0f))
493 m->load_hist[m->nload_hist][0] = l.share.some ? PCT(l.share.v) : 0;
494 m->load_hist[m->nload_hist][1] = PCT(l.cpu);
495#undef PCT
496 m->nload_hist++;
497 }
498 const mt_section *t = mt_snapshot_by_name(&s, "train");
499 if (t) {
500 mt_rate_push(&m->train_steps, now, mt_train_next_step(t));
501 record_losses(m, t);
502 }
503 const mt_section *d = mt_snapshot_by_name(&s, "doom");
504 if (d) mt_rate_push(&m->doom_shown, now, mt_section_view_u32(d, MESH_DOOM_D_SHOW + MESH_SH_SEQ));
505 if (m->has_snap) mt_snapshot_free(&m->snap);
506 m->snap = s;
507 m->has_snap = true;
508 m->snap_at = now;
509 m->has_snap_at = true;
510 m->online = true;
511 set_str(&m->status, xstrdup("online"));
512}
513
514static void apply_trace(mt_machine *m, const uint8_t *b, size_t len, mt_instant now) {
515 mt_trace_prog *v;
516 size_t n;
517 char *err = NULL;
518 if (!mt_parse_trace(b, len, &v, &n, &err)) {
519 set_str(&m->trace_err, err);
520 return;
521 }
522 set_str(&m->trace_err, NULL);
523 for (size_t i = 0; i < n; i++) {
524 mt_trace_track *t = NULL;
525 size_t at = 0;
526 while (at < m->ntrace && m->trace[at].slot < v[i].slot) at++;
527 if (at < m->ntrace && m->trace[at].slot == v[i].slot) t = &m->trace[at];
528 mt_instant moved = t && t->n == v[i].n ? t->moved : now;
529 if (!t) {
530 m->trace = xrealloc(m->trace, (m->ntrace + 1) * sizeof *m->trace);
531 memmove(&m->trace[at + 1], &m->trace[at], (m->ntrace - at) * sizeof *m->trace);
532 m->ntrace++;
533 t = &m->trace[at];
534 } else {
535 free(t->recs);
536 }
537 *t = (mt_trace_track){v[i].slot, v[i].n, moved, v[i].recs, v[i].nrecs};
538 v[i].recs = NULL;
539 }
540 mt_trace_free(v, n);
541}
542
543/* ---- Identity, Place */
544
545bool mt_params_identity(const mt_app *app, mt_identity *out) {
546 const mt_machine *with[64];
547 size_t nw = 0;
548 for (size_t i = 0; i < app->nmachines && nw < 64; i++)
549 if (app->machines[i].nparams) with[nw++] = &app->machines[i];
550 if (nw < 2) return false;
551 bool found = false;
552 uint64_t step = 0;
553 for (size_t k = with[0]->nparams; k-- > 0 && !found;) {
554 uint64_t s = with[0]->params[k].step;
555 bool all = true;
556 for (size_t i = 1; i < nw && all; i++) {
557 bool has = false;
558 for (size_t j = 0; j < with[i]->nparams; j++) has |= with[i]->params[j].step == s;
559 all = has;
560 }
561 if (all) {
562 found = true;
563 step = s;
564 }
565 }
566 if (!found) return false;
567 const uint8_t *views[64];
568 for (size_t i = 0; i < nw; i++)
569 for (size_t j = 0; j < with[i]->nparams; j++)
570 if (with[i]->params[j].step == step) views[i] = with[i]->params[j].view;
571 /* The majority's view; a tie goes to the one seen first. */
572 size_t best = 0, best_count = 0, best_first = SIZE_MAX;
573 for (size_t i = 0; i < nw; i++) {
574 size_t count = 0, first = SIZE_MAX;
575 for (size_t j = 0; j < nw; j++)
576 if (memcmp(views[j], views[i], 32) == 0) {
577 count++;
578 if (first == SIZE_MAX) first = j;
579 }
580 if (count > best_count || (count == best_count && first < best_first)) {
581 best = i;
582 best_count = count;
583 best_first = first;
584 }
585 }
586 memset(out, 0, sizeof *out);
587 out->step = step;
588 out->n = nw;
589 for (size_t i = 0; i < nw; i++) {
590 if (memcmp(views[i], views[best], 32) == 0) continue;
591 uint32_t k;
592 if (out->ndiffer < sizeof out->differ / sizeof out->differ[0]) {
593 if (mt_machine_system(with[i], &k)) snprintf(out->differ[out->ndiffer], 64, "system %u", k);
594 else snprintf(out->differ[out->ndiffer], 64, "%s", with[i]->endpoint);
595 out->ndiffer++;
596 }
597 }
598 out->identical = out->ndiffer == 0;
599 return true;
600}
601
602bool mt_place_of(const mt_section *s, mt_place *out) {
603 const uint8_t *x = mt_section_ext(s);
604 if (mt_get_u32(x, MESH_PX_BYTES, MESH_PX_PLACE_CAP) == MESH_PLACE_CAP_NONE) return false;
605 uint32_t mode = mt_get_u32(x, MESH_PX_BYTES, MESH_PX_PLACE), sys = mt_get_u32(x, MESH_PX_BYTES, MESH_PX_PLACE_SYS);
606 memset(out, 0, sizeof *out);
607 if (mode == MESH_PLACE_PIN) {
608 out->kind = MT_PLACE_PIN;
609 out->pin = sys;
610 } else if (mode == MESH_PLACE_SPLIT) {
611 out->kind = MT_PLACE_SPLIT;
612 memcpy(out->assign, x + MESH_PX_PLACE_ASSIGN, MESH_PLACE_PARTS_MAX);
613 } else {
614 out->kind = MT_PLACE_LOCAL;
615 }
616 return true;
617}
618
619void mt_place_label(mt_place p, char out[32]) {
620 if (p.kind == MT_PLACE_LOCAL) snprintf(out, 32, "local");
621 else if (p.kind == MT_PLACE_PIN) snprintf(out, 32, "pin %u", p.pin);
622 else snprintf(out, 32, "split");
623}
624
625uint32_t mt_place_input_to(mt_place p, uint32_t me) {
626 return p.kind == MT_PLACE_LOCAL ? me : p.kind == MT_PLACE_PIN ? p.pin : p.assign[0];
627}
628
629mt_place mt_place_next(mt_place p, uint32_t n) {
630 mt_place r;
631 memset(&r, 0, sizeof r);
632 if (p.kind == MT_PLACE_LOCAL) {
633 r.kind = MT_PLACE_PIN;
634 r.pin = 0;
635 } else if (p.kind == MT_PLACE_PIN && p.pin + 1 < (n > 1 ? n : 1)) {
636 r.kind = MT_PLACE_PIN;
637 r.pin = p.pin + 1;
638 } else if (p.kind == MT_PLACE_PIN) {
639 r.kind = MT_PLACE_SPLIT;
640 memset(r.assign, MESH_PART_NONE, sizeof r.assign);
641 } else {
642 r.kind = MT_PLACE_LOCAL;
643 }
644 return r;
645}
646
647bool mt_place_eq(mt_place a, mt_place b) {
648 if (a.kind != b.kind) return false;
649 if (a.kind == MT_PLACE_PIN) return a.pin == b.pin;
650 if (a.kind == MT_PLACE_SPLIT) return memcmp(a.assign, b.assign, sizeof a.assign) == 0;
651 return true;
652}
653
654size_t mt_part_runs(const uint8_t assign[MESH_PLACE_PARTS_MAX], uint32_t fanout, mt_part_run *out) {
655 size_t n = 0;
656 uint32_t lim = fanout < MESH_PLACE_PARTS_MAX ? fanout : MESH_PLACE_PARTS_MAX;
657 for (uint32_t b = 0; b < lim; b++) {
658 uint8_t k = assign[b];
659 if (k == MESH_PART_NONE) continue;
660 if (n && out[n - 1].system == k && out[n - 1].last + 1 == b) out[n - 1].last = b;
661 else out[n++] = (mt_part_run){b, b, k};
662 }
663 return n;
664}
665
666/* ---- Actions and prompts */
667
668void mt_action_free(mt_action *a) {
669 free(a->line);
670 a->line = NULL;
671}
672
673static mt_action none(void) { return (mt_action){MT_ACTION_NONE, 0, NULL}; }
674static mt_action quit(void) { return (mt_action){MT_ACTION_QUIT, 0, NULL}; }
675static mt_action send(size_t i, char *line) { return (mt_action){MT_ACTION_SEND, i, line}; }
676static mt_action poll_(size_t i, char *line) { return (mt_action){MT_ACTION_POLL, i, line}; }
677
678static void prompt_clear(mt_prompt *p) {
679 free(p->name);
680 free(p->line);
681 free(p->what);
682 memset(p, 0, sizeof *p);
683}
684
685/* ---- App */
686
687void mt_app_init(mt_app *app, const char *const *endpoints, size_t n) {
688 memset(app, 0, sizeof *app);
689 app->machines = xmalloc(n * sizeof *app->machines);
690 app->nmachines = n;
691 for (size_t i = 0; i < n; i++) machine_init(&app->machines[i], endpoints[i]);
692 app->tab = MT_TAB_MESH;
693 strcpy(app->http_input, "/");
694 app->last_reply = xstrdup("");
695}
696
697void mt_app_free(mt_app *app) {
698 for (size_t i = 0; i < app->nmachines; i++) machine_free(&app->machines[i]);
699 free(app->machines);
700 prompt_clear(&app->prompt);
701 free(app->last_reply);
702 for (int v = 0; v < 3; v++)
703 if (app->has_view[v]) mt_view_free(&app->views[v]);
704 memset(app, 0, sizeof *app);
705}
706
707const mt_machine *mt_app_focused(const mt_app *app) { return &app->machines[app->focus]; }
708
709size_t mt_app_view_machine(const mt_app *app, size_t view) {
710 if (view == 2) return app->focus;
711 for (size_t i = 0; i < app->nmachines; i++) {
712 mt_fleet f;
713 if (app->machines[i].has_snap && mt_snapshot_fleet(&app->machines[i].snap, &f) && f.role == MESH_ROLE_LEADER) return i;
714 }
715 return app->focus;
716}
717
718int mt_app_tab_view(const mt_app *app) {
719 return app->tab == MT_TAB_FLEET ? 0 : app->tab == MT_TAB_JOBS ? 1 : app->tab == MT_TAB_PIPES ? 2 : -1;
720}
721
722const mt_view *mt_app_view(const mt_app *app, size_t v) { return app->has_view[v] ? &app->views[v] : NULL; }
723
724void mt_app_fleet_sel(const mt_app *app, char **prog, bool *has_member, uint32_t *member) {
725 *prog = NULL;
726 *has_member = false;
727 const mt_view *v = mt_app_view(app, 0);
728 if (!v) return;
729 size_t np = mt_view_count(v, "program");
730 if (np) {
731 size_t i = app->fsel[0] < np - 1 ? app->fsel[0] : np - 1;
732 *prog = xstrdup(mt_vrec_s(mt_view_nth(v, "program", i), "name"));
733 }
734 uint32_t ids[256];
735 size_t nm = 0;
736 for (size_t i = 0, c = mt_view_count(v, "member"); i < c && nm < 256; i++) {
737 uint64_t id;
738 if (mt_vrec_u(mt_view_nth(v, "member", i), "id", &id)) ids[nm++] = (uint32_t)id;
739 }
740 if (nm) {
741 size_t i = app->fsel[1] < nm - 1 ? app->fsel[1] : nm - 1;
742 *has_member = true;
743 *member = ids[i];
744 }
745}
746
747static bool jsel_row(const mt_app *app, size_t *row) {
748 const mt_view *v = mt_app_view(app, 1);
749 if (!v) return false;
750 if (app->has_jsel) {
751 *row = app->jsel;
752 return true;
753 }
754 *row = 0;
755 for (size_t i = 0, c = mt_view_count(v, "job"); i < c; i++) {
756 const char *st = mt_vrec_get(mt_view_nth(v, "job", i), "state");
757 if (st && strcmp(st, "running") == 0) {
758 *row = i;
759 break;
760 }
761 }
762 return true;
763}
764
765char *mt_app_job_sel(const mt_app *app) {
766 const mt_view *v = mt_app_view(app, 1);
767 if (!v) return NULL;
768 size_t row, n = mt_view_count(v, "job");
769 if (!jsel_row(app, &row) || n == 0) return NULL;
770 if (row > n - 1) row = n - 1;
771 return xstrdup(mt_vrec_s(mt_view_nth(v, "job", row), "job"));
772}
773
774uint32_t mt_app_cluster_size(const mt_app *app) {
775 uint32_t reported = 0;
776 for (size_t i = 0; i < app->nmachines; i++) {
777 const mt_machine *m = &app->machines[i];
778 if (m->has_snap && mt_snapshot_system(&m->snap, NULL)) {
779 uint32_t n = mt_snapshot_systems(&m->snap);
780 if (n > reported) reported = n;
781 }
782 }
783 return reported > 0 ? reported : (uint32_t)(app->nmachines > 1 ? app->nmachines : 1);
784}
785
786bool mt_app_render_program(const mt_app *app, char name[64], uint32_t *tiles) {
787 const mt_machine *m = mt_app_focused(app);
788 if (!m->has_snap) return false;
789 const mt_section *p = NULL;
790 for (int i = 0; i < 2 && !p; i++) p = mt_snapshot_by_name(&m->snap, MT_RENDER_PROGRAMS[i]);
791 if (!p) return false;
792 uint32_t t = mt_section_view_u32(p, PT_V_TILES);
793 if (!(mt_section_view_u32(p, PT_V_MAGIC) == PT_VIEW_MAGIC && t >= 1 && t <= PT_MAX_TILES)) return false;
794 mt_section_name(p, name);
795 *tiles = t;
796 return true;
797}
798
799const mt_section *mt_app_selected(const mt_app *app) {
800 const mt_machine *m = mt_app_focused(app);
801 if (!m->has_snap || m->snap.nsections == 0) return NULL;
802 size_t i = app->sel < m->snap.nsections - 1 ? app->sel : m->snap.nsections - 1;
803 return &m->snap.sections[i];
804}
805
806size_t mt_app_tick(mt_app *app, mt_instant now, mt_action out[2]) {
807 if (app->tab == MT_TAB_RENDER) {
808 if (app->has_render_at && mt_since(now, app->render_at) < MT_RENDER_POLL) return 0;
809 app->render_at = now;
810 app->has_render_at = true;
811 size_t n = 0;
812 out[n++] = poll_(app->focus, xstrdup("render latest raw"));
813 char name[64];
814 uint32_t tiles;
815 if (!mt_app_focused(app)->has_fleet && mt_app_render_program(app, name, &tiles))
816 out[n++] = poll_(app->focus, xfmt("peek %s %u %u", name, (unsigned)PT_V_RGB, tiles * 4 * PT_TILE_PIX));
817 return n;
818 }
819 int v = mt_app_tab_view(app);
820 if (v >= 0) {
821 if (app->has_view_at && mt_since(now, app->view_at) < MT_VIEW_POLL) return 0;
822 app->view_at = now;
823 app->has_view_at = true;
824 out[0] = poll_(mt_app_view_machine(app, (size_t)v), xfmt("view %s", MT_VIEWS[v]));
825 return 1;
826 }
827 if (app->tab != MT_TAB_PROGRAMS || (app->has_aux_at && mt_since(now, app->aux_at) < MT_AUX_POLL)) return 0;
828 app->aux_at = now;
829 app->has_aux_at = true;
830 out[0] = poll_(app->focus, xstrdup("trace"));
831 out[1] = poll_(app->focus, xfmt("log %d", MT_LOG_LINES));
832 return 2;
833}
834
835/* text's lines (str::lines), owned; optionally trimmed and without empty ones. */
836static char **lines_of(const char *text, bool trim_skip, size_t *n) {
837 char **v = NULL;
838 size_t cnt = 0, len = strlen(text);
839 for (size_t start = 0; start < len;) {
840 const char *nl = memchr(text + start, '\n', len - start);
841 size_t end = nl ? (size_t)(nl - text) : len, e = end;
842 if (nl && e > start && text[e - 1] == '\r') e--;
843 const char *s = text + start;
844 size_t l = e - start;
845 start = nl ? end + 1 : len;
846 if (trim_skip) {
847 while (l && (*s == ' ' || (*s >= '\t' && *s <= '\r'))) s++, l--;
848 while (l && (s[l - 1] == ' ' || (s[l - 1] >= '\t' && s[l - 1] <= '\r'))) l--;
849 if (!l) continue;
850 }
851 v = xrealloc(v, (cnt + 1) * sizeof *v);
852 v[cnt] = xmalloc(l + 1);
853 memcpy(v[cnt], s, l);
854 v[cnt][l] = 0;
855 cnt++;
856 }
857 *n = cnt;
858 return v;
859}
860
861static void first_word(const char *line, int nth, char *out, size_t cap) {
862 out[0] = 0;
863 const char *p = line;
864 for (int i = 0; i <= nth; i++) {
865 while (*p == ' ' || (*p >= '\t' && *p <= '\r')) p++;
866 const char *w = p;
867 while (*p && !(*p == ' ' || (*p >= '\t' && *p <= '\r'))) p++;
868 if (i == nth) {
869 size_t l = (size_t)(p - w);
870 if (l >= cap) l = cap - 1;
871 memcpy(out, w, l);
872 out[l] = 0;
873 }
874 if (!*p) {
875 if (i < nth) out[0] = 0;
876 return;
877 }
878 }
879}
880
881static void apply_reply(mt_app *app, mt_update *u, mt_instant now) {
882 char name[256];
883 mt_machine_name(&app->machines[u->machine], name, sizeof name);
884 mt_machine *m = &app->machines[u->machine];
885 char verb[64], arg1[64];
886 first_word(u->text, 0, verb, sizeof verb);
887 first_word(u->text, 1, arg1, sizeof arg1);
888 char *lossy = u->bytes ? tui_utf8_lossy(u->data, u->len, NULL) : NULL;
889 const char *text = u->bytes ? lossy : (const char *)u->data;
890 if (strcmp(verb, "trace") == 0) {
891 if (u->bytes) apply_trace(m, u->data, u->len, now);
892 else set_str(&m->trace_err, xstrdup(text));
893 } else if (strcmp(verb, "log") == 0) {
894 if (u->bytes) {
895 strv_free(m->log, m->nlog);
896 m->log = lines_of(text, false, &m->nlog);
897 set_str(&m->log_err, NULL);
898 } else {
899 set_str(&m->log_err, xstrdup(text));
900 }
901 } else if (strcmp(verb, "view") == 0) {
902 for (int v = 0; v < 3; v++)
903 if (strcmp(arg1, MT_VIEWS[v]) == 0) {
904 if (app->has_view[v]) mt_view_free(&app->views[v]);
905 app->views[v] = mt_view_new(u->machine, now, text);
906 app->has_view[v] = true;
907 }
908 } else if (strcmp(verb, "render") == 0) {
909 mt_fleet_image f;
910 if (u->bytes && mt_fleet_image_parse(u->data, u->len, &f)) {
911 uint64_t folded = mt_fleet_image_num(&f, "folded");
912 if (m->has_fleet_prev && now > m->fleet_prev_t) {
913 double dt = mt_secs_f64(mt_since(now, m->fleet_prev_t));
914 double spp = (double)mt_fleet_image_num(&f, "spp");
915 f.msps = (double)(folded > m->fleet_prev_n ? folded - m->fleet_prev_n : 0) * (double)PT_TILE_PIX * spp / dt / 1e6;
916 }
917 m->has_fleet_prev = true;
918 m->fleet_prev_t = now;
919 m->fleet_prev_n = folded;
920 if (m->has_fleet) mt_fleet_image_free(&m->fleet);
921 m->fleet = f;
922 m->has_fleet = true;
923 } else {
924 if (m->has_fleet) mt_fleet_image_free(&m->fleet);
925 m->has_fleet = false;
926 m->has_fleet_prev = false;
927 }
928 } else if (strcmp(verb, "peek") == 0 && (strcmp(arg1, MT_RENDER_PROGRAMS[0]) == 0 || strcmp(arg1, MT_RENDER_PROGRAMS[1]) == 0)) {
929 if (u->bytes && u->len >= 3 && memcmp(u->data, "err", 3) == 0) {
930 const char *s = text;
931 size_t l = strlen(s);
932 while (l && (*s == ' ' || (*s >= '\t' && *s <= '\r'))) s++, l--;
933 while (l && (s[l - 1] == ' ' || (s[l - 1] >= '\t' && s[l - 1] <= '\r'))) l--;
934 char *t = xmalloc(l + 1);
935 memcpy(t, s, l);
936 t[l] = 0;
937 set_str(&m->render_err, t);
938 } else if (u->bytes) {
939 set_str(&m->render_name, xstrdup(arg1));
940 free(m->render_px);
941 m->render_px = u->data;
942 m->render_len = u->len;
943 u->data = NULL;
944 m->has_render = true;
945 set_str(&m->render_err, NULL);
946 } else {
947 set_str(&m->render_err, xstrdup(text));
948 }
949 } else if (strcmp(verb, "catalog") == 0 && u->bytes) {
950 strv_free(m->catalog, m->ncatalog);
951 m->catalog = lines_of(text, true, &m->ncatalog);
952 set_str(&app->last_reply, xfmt("%s: catalog -> %zu programs", name, m->ncatalog));
953 } else if (!u->bytes) {
954 bool key = strncmp(u->text, "input doom ", 11) == 0 || strncmp(u->text, "key ", 4) == 0;
955 if (!(key && strncmp(text, "ok", 2) == 0)) set_str(&app->last_reply, xfmt("%s: %s -> %s", name, u->text, text));
956 } else {
957 set_str(&app->last_reply, xfmt("%s: %s -> %zu bytes", name, u->text, u->len));
958 if (strcmp(verb, "http") == 0) {
959 free(m->http_resp);
960 m->http_resp = u->data;
961 m->http_resp_len = u->len;
962 u->data = NULL;
963 m->has_http_resp = true;
964 app->http_scroll = 0;
965 }
966 }
967 free(lossy);
968}
969
970void mt_app_apply(mt_app *app, mt_update *u, mt_instant now) {
971 mt_machine *m = &app->machines[u->machine];
972 switch (u->kind) {
973 case MT_UPDATE_ONLINE:
974 m->online = true;
975 set_str(&m->status, xstrdup("online"));
976 break;
977 case MT_UPDATE_OFFLINE:
978 m->online = false;
979 set_str(&m->status, xstrdup(u->text));
980 break;
981 case MT_UPDATE_SNAP:
982 apply_snap(m, u->snap, now);
983 u->snap = (mt_snapshot){NULL, 0, NULL, 0};
984 break;
985 case MT_UPDATE_SNAP_ERROR:
986 set_str(&m->status, xfmt("snap: %s", u->text));
987 break;
988 case MT_UPDATE_REPLY:
989 apply_reply(app, u, now);
990 break;
991 }
992 free(u->text);
993 free(u->data);
994 if (u->snap.bytes) mt_snapshot_free(&u->snap);
995 memset(u, 0, sizeof *u);
996}
997
998static mt_action note(mt_app *app, char *s) {
999 set_str(&app->last_reply, s);
1000 return none();
1001}
1002
1003static mt_action toggle(mt_app *app, const char *name) {
1004 const mt_section *p = mt_machine_program(mt_app_focused(app), name);
1005 if (p) return send(app->focus, xfmt("%s %s", mt_section_running(p) ? "stop" : "start", name));
1006 char mn[256];
1007 mt_machine_name(mt_app_focused(app), mn, sizeof mn);
1008 return note(app, xfmt("%s: %s is not loaded", mn, name));
1009}
1010
1011static bool is_char(tui_key k, uint32_t c) { return k.code == TUI_KEY_CHAR && k.ch == c; }
1012static bool is_ascii_graphic(uint32_t c) { return c >= 0x21 && c <= 0x7E; }
1013
1014static void common_key(mt_app *app, tui_key k) {
1015 size_t n = app->nmachines;
1016 if (k.code != TUI_KEY_CHAR) return;
1017 for (int i = 0; i < MT_TABS; i++)
1018 if (k.ch == (uint32_t)MT_TAB_KEYS[i]) {
1019 app->tab = i;
1020 app->has_view_at = false;
1021 return;
1022 }
1023 if (k.ch == '[') app->focus = (app->focus + n - 1) % n;
1024 else if (k.ch == ']') app->focus = (app->focus + 1) % n;
1025}
1026
1027static mt_action default_key(mt_app *app, tui_key k) {
1028 if (is_char(k, 'q') || k.code == TUI_KEY_ESC) return quit();
1029 if (is_char(k, 'd')) return toggle(app, "doom");
1030 if (is_char(k, 't')) return toggle(app, "train");
1031 if (is_char(k, 'i')) return toggle(app, "infer");
1032 if (is_char(k, 'X')) return send(app->focus, xstrdup("reset train"));
1033 common_key(app, k);
1034 return none();
1035}
1036
1037static mt_action confirm(mt_app *app, size_t machine, char *line, char *what) {
1038 prompt_clear(&app->prompt);
1039 app->prompt.kind = MT_PROMPT_CONFIRM;
1040 app->prompt.machine = machine;
1041 app->prompt.line = line;
1042 app->prompt.what = what;
1043 return none();
1044}
1045
1046static mt_action fleet_key(mt_app *app, tui_key k) {
1047 const mt_view *v = mt_app_view(app, 0);
1048 size_t np = v ? mt_view_count(v, "program") : 0, nm = v ? mt_view_count(v, "member") : 0;
1049 size_t at = v ? v->machine : mt_app_view_machine(app, 0);
1050 char *prog;
1051 bool has_m;
1052 uint32_t member = 0;
1053 mt_app_fleet_sel(app, &prog, &has_m, &member);
1054 mt_action r = none();
1055 size_t npm = np ? np - 1 : 0, nmm = nm ? nm - 1 : 0;
1056 if (k.code == TUI_KEY_UP) {
1057 size_t c = app->fsel[0] < npm ? app->fsel[0] : npm;
1058 app->fsel[0] = c ? c - 1 : 0;
1059 } else if (k.code == TUI_KEY_DOWN) {
1060 app->fsel[0] = app->fsel[0] + 1 < npm ? app->fsel[0] + 1 : npm;
1061 } else if (k.code == TUI_KEY_LEFT) {
1062 size_t c = app->fsel[1] < nmm ? app->fsel[1] : nmm;
1063 app->fsel[1] = c ? c - 1 : 0;
1064 } else if (k.code == TUI_KEY_RIGHT) {
1065 app->fsel[1] = app->fsel[1] + 1 < nmm ? app->fsel[1] + 1 : nmm;
1066 } else if (is_char(k, 'c')) {
1067 if (!has_m) r = note(app, xstrdup("no member selected"));
1068 else {
1069 bool cordoned = false;
1070 for (size_t i = 0; v && i < nm; i++) {
1071 const mt_vrec *mr = mt_view_nth(v, "member", i);
1072 uint64_t id;
1073 if (mt_vrec_u(mr, "id", &id) && id == member) {
1074 cordoned = mt_vrec_flag(mr, "cordoned");
1075 break;
1076 }
1077 }
1078 r = cordoned ? confirm(app, at, xfmt("sched uncordon %u", member),
1079 xfmt("uncordon system %u (new placements and job leases may go there again)", member))
1080 : confirm(app, at, xfmt("sched cordon %u", member),
1081 xfmt("cordon system %u (no new placements or job leases there; what runs stays)", member));
1082 }
1083 } else if (is_char(k, 'p')) {
1084 if (!prog || !has_m) r = note(app, xstrdup("select a program and a member"));
1085 else r = confirm(app, at, xfmt("sched pin %s %u", prog, member), xfmt("pin %s to system %u (it migrates there)", prog, member));
1086 } else if (is_char(k, 'u')) {
1087 if (!prog) r = note(app, xstrdup("no program selected"));
1088 else r = confirm(app, at, xfmt("sched unpin %s", prog), xfmt("unpin %s (the scheduler places it again)", prog));
1089 } else if (is_char(k, 'z')) {
1090 const mt_vrec *f = v ? mt_view_first(v, "fleet") : NULL;
1091 bool paused = f && mt_vrec_flag(f, "paused");
1092 r = paused ? confirm(app, at, xstrdup("sched resume"), xstrdup("resume the scheduler (placement and rebalancing)"))
1093 : confirm(app, at, xstrdup("sched pause"), xstrdup("pause the scheduler (no automatic placement changes or moves)"));
1094 } else if (is_char(k, 'g')) {
1095 const mt_vrec *g = v ? mt_view_first(v, "group") : NULL;
1096 if (!g) r = note(app, xstrdup("no lockstep group"));
1097 else {
1098 const char *gn = mt_vrec_s(g, "name");
1099 r = confirm(app, at, xfmt("group %s reform", gn),
1100 xfmt("re-form group %s over every eligible up member (a new epoch: the program resets)", gn));
1101 }
1102 } else if (k.code == TUI_KEY_ESC) {
1103 app->tab = MT_TAB_MESH;
1104 } else {
1105 r = default_key(app, k);
1106 }
1107 free(prog);
1108 return r;
1109}
1110
1111static mt_action jobs_key(mt_app *app, tui_key k) {
1112 const mt_view *v = mt_app_view(app, 1);
1113 size_t nj = v ? mt_view_count(v, "job") : 0, njm = nj ? nj - 1 : 0;
1114 size_t cur = 0;
1115 if (!jsel_row(app, &cur)) cur = 0;
1116 if (k.code == TUI_KEY_UP) {
1117 size_t c = cur < njm ? cur : njm;
1118 app->has_jsel = true;
1119 app->jsel = c ? c - 1 : 0;
1120 } else if (k.code == TUI_KEY_DOWN) {
1121 app->has_jsel = true;
1122 app->jsel = cur + 1 < njm ? cur + 1 : njm;
1123 } else if (is_char(k, 'x')) {
1124 char *id = mt_app_job_sel(app);
1125 if (!id) return note(app, xstrdup("no job selected"));
1126 size_t at = v ? v->machine : mt_app_view_machine(app, 1);
1127 mt_action r = confirm(app, at, xfmt("job cancel %s", id), xfmt("cancel job %s (what is left is given up; done items stay)", id));
1128 free(id);
1129 return r;
1130 } else if (k.code == TUI_KEY_ESC) {
1131 app->tab = MT_TAB_MESH;
1132 } else {
1133 return default_key(app, k);
1134 }
1135 return none();
1136}
1137
1138static mt_action render_key(mt_app *app, tui_key k) {
1139 char name[64];
1140 uint32_t tiles;
1141 bool prog = mt_app_render_program(app, name, &tiles);
1142 const mt_machine *m = mt_app_focused(app);
1143 if (m->has_fleet) {
1144 if (is_char(k, 't')) {
1145 const char *tone = mt_fleet_image_get(&m->fleet, "tone");
1146 return send(app->focus, xfmt("job tone %s %s", mt_fleet_image_get(&m->fleet, "job"), strcmp(tone, "reinhard") == 0 ? "aces" : "reinhard"));
1147 }
1148 if (is_char(k, 'r')) return note(app, xstrdup("a fleet job's image is its committed passes: nothing to restart"));
1149 }
1150 if (is_char(k, 'm')) {
1151 app->render_marks = !app->render_marks;
1152 return none();
1153 }
1154 if (is_char(k, 't') && prog) {
1155 const mt_section *p = mt_machine_program(m, name);
1156 uint32_t cur = p ? mt_section_view_u32(p, PT_V_TONEMAP) : 0;
1157 return send(app->focus, xfmt("task %s %u %u", name, (unsigned)PT_OP_TONEMAP, (cur + 1) % PT_TONEMAPS));
1158 }
1159 if (is_char(k, 'r') && prog) return send(app->focus, xfmt("task %s %u 0", name, (unsigned)PT_OP_RESTART));
1160 if ((is_char(k, 't') || is_char(k, 'r')) && !prog) return note(app, xstrdup("pathtrace is not loaded on this machine"));
1161 if (k.code == TUI_KEY_ESC) {
1162 app->tab = MT_TAB_MESH;
1163 return none();
1164 }
1165 if (is_char(k, 'q')) return quit();
1166 common_key(app, k);
1167 return none();
1168}
1169
1170static mt_action programs_key(mt_app *app, tui_key k) {
1171 const mt_machine *fm = mt_app_focused(app);
1172 size_t count = fm->has_snap ? fm->snap.nsections : 0, cm = count ? count - 1 : 0;
1173 const mt_section *s = mt_app_selected(app);
1174 char sname[64] = "";
1175 bool running = false;
1176 if (s) {
1177 mt_section_name(s, sname);
1178 running = mt_section_running(s);
1179 }
1180#define BY_SEL(verb) (s ? send(app->focus, xfmt("%s %s", verb, sname)) : note(app, xstrdup("no program selected")))
1181 if (k.code == TUI_KEY_UP) {
1182 size_t c = app->sel < cm ? app->sel : cm;
1183 app->sel = c ? c - 1 : 0;
1184 return none();
1185 }
1186 if (k.code == TUI_KEY_DOWN) {
1187 app->sel = app->sel + 1 < cm ? app->sel + 1 : cm;
1188 return none();
1189 }
1190 if (is_char(k, 's')) return running ? BY_SEL("stop") : BY_SEL("start");
1191 if (is_char(k, 'u')) return BY_SEL("upgrade");
1192 if (is_char(k, 'b')) return BY_SEL("rollback");
1193 if (is_char(k, 'r')) return BY_SEL("reset");
1194 if (is_char(k, 'x')) {
1195 if (!s) return note(app, xstrdup("no program selected"));
1196 prompt_clear(&app->prompt);
1197 app->prompt.kind = MT_PROMPT_UNLOAD;
1198 app->prompt.machine = app->focus;
1199 app->prompt.name = xstrdup(sname);
1200 return none();
1201 }
1202 if (is_char(k, 'L') || is_char(k, 'l')) {
1203 prompt_clear(&app->prompt);
1204 app->prompt.kind = MT_PROMPT_LOAD;
1205 return send(app->focus, xstrdup("catalog"));
1206 }
1207 if (k.code == TUI_KEY_ESC) {
1208 app->tab = MT_TAB_MESH;
1209 return none();
1210 }
1211 return default_key(app, k);
1212#undef BY_SEL
1213}
1214
1215static mt_action prompt_key(mt_app *app, tui_key k) {
1216 mt_prompt p = app->prompt;
1217 memset(&app->prompt, 0, sizeof app->prompt); /* taken */
1218 mt_action r = none();
1219 switch (p.kind) {
1220 case MT_PROMPT_UNLOAD:
1221 if (is_char(k, 'y') || is_char(k, 'Y')) r = send(p.machine, xfmt("unload %s", p.name));
1222 else r = note(app, xfmt("unload %s cancelled", p.name));
1223 break;
1224 case MT_PROMPT_CONFIRM:
1225 if (is_char(k, 'y') || is_char(k, 'Y')) {
1226 app->has_view_at = false;
1227 r = send(p.machine, xstrdup(p.line));
1228 } else {
1229 r = note(app, xfmt("%s: cancelled", p.line));
1230 }
1231 break;
1232 case MT_PROMPT_LOAD: {
1233 const mt_machine *m = mt_app_focused(app);
1234 size_t ncat = m->ncatalog;
1235 if (k.code == TUI_KEY_ESC) {
1236 r = note(app, xstrdup("load cancelled"));
1237 break;
1238 }
1239 if (k.code == TUI_KEY_ENTER) {
1240 if (*p.input) r = send(app->focus, xfmt("load %s", p.input));
1241 else if (p.has_pick && p.pick < ncat) r = send(app->focus, xfmt("load %s", m->catalog[p.pick]));
1242 else r = note(app, xstrdup("load: type a program dir or pick one"));
1243 break;
1244 }
1245 if (k.code == TUI_KEY_UP && ncat) {
1246 p.pick = p.has_pick ? (p.pick ? p.pick - 1 : 0) : ncat - 1;
1247 p.has_pick = true;
1248 p.input[0] = 0;
1249 } else if (k.code == TUI_KEY_DOWN && ncat) {
1250 p.pick = p.has_pick ? (p.pick + 1 < ncat - 1 ? p.pick + 1 : ncat - 1) : 0;
1251 p.has_pick = true;
1252 p.input[0] = 0;
1253 } else if (k.code == TUI_KEY_BACKSPACE) {
1254 size_t l = strlen(p.input);
1255 if (l) p.input[l - 1] = 0;
1256 } else if (k.code == TUI_KEY_CHAR && is_ascii_graphic(k.ch) && strlen(p.input) < 128) {
1257 size_t l = strlen(p.input);
1258 p.input[l] = (char)k.ch;
1259 p.input[l + 1] = 0;
1260 p.has_pick = false;
1261 }
1262 app->prompt = p;
1263 return none();
1264 }
1265 case MT_PROMPT_NONE:
1266 return none();
1267 }
1268 prompt_clear(&p);
1269 return r;
1270}
1271
1272static mt_action http_key(mt_app *app, tui_key k) {
1273 size_t l = strlen(app->http_input);
1274 if (k.code == TUI_KEY_ESC) app->tab = MT_TAB_MESH;
1275 else if (k.code == TUI_KEY_ENTER) return send(app->focus, xfmt("http %s", l ? app->http_input : "/"));
1276 else if (k.code == TUI_KEY_BACKSPACE) {
1277 if (l) app->http_input[l - 1] = 0;
1278 } else if (k.code == TUI_KEY_UP) app->http_scroll = app->http_scroll ? (uint16_t)(app->http_scroll - 1) : 0;
1279 else if (k.code == TUI_KEY_DOWN) app->http_scroll = app->http_scroll == UINT16_MAX ? UINT16_MAX : (uint16_t)(app->http_scroll + 1);
1280 else if (k.code == TUI_KEY_PAGE_UP) app->http_scroll = app->http_scroll > 10 ? (uint16_t)(app->http_scroll - 10) : 0;
1281 else if (k.code == TUI_KEY_PAGE_DOWN) app->http_scroll = app->http_scroll > UINT16_MAX - 10 ? UINT16_MAX : (uint16_t)(app->http_scroll + 10);
1282 else if (k.code == TUI_KEY_CHAR && is_ascii_graphic(k.ch) && l < MT_HTTP_INPUT_MAX) {
1283 app->http_input[l] = (char)k.ch;
1284 app->http_input[l + 1] = 0;
1285 }
1286 return none();
1287}
1288
1289static uint32_t ascii_lower(uint32_t c) { return c >= 'A' && c <= 'Z' ? c + 32 : c; }
1290
1291mt_action mt_app_handle_key(mt_app *app, tui_key k) {
1292 if (k.kind == TUI_KEY_RELEASE) return none();
1293 if ((k.mods & TUI_MOD_CONTROL) && is_char(k, 'c')) return quit();
1294 if (app->prompt.kind != MT_PROMPT_NONE) return prompt_key(app, k);
1295 size_t n = app->nmachines;
1296 if (k.code == TUI_KEY_TAB) {
1297 app->tab = (app->tab + 1) % MT_TABS;
1298 app->has_view_at = false;
1299 return none();
1300 }
1301 if (k.code == TUI_KEY_BACKTAB) {
1302 app->tab = (app->tab + MT_TABS - 1) % MT_TABS;
1303 app->has_view_at = false;
1304 return none();
1305 }
1306 if (k.code == TUI_KEY_F && k.f == 1) {
1307 app->focus = (app->focus + n - 1) % n;
1308 return none();
1309 }
1310 if (k.code == TUI_KEY_F && k.f == 2) {
1311 app->focus = (app->focus + 1) % n;
1312 return none();
1313 }
1314 switch (app->tab) {
1315 case MT_TAB_HTTP: return http_key(app, k);
1316 case MT_TAB_DOOM: {
1317 char key = 0;
1318 if (k.code == TUI_KEY_UP) key = 'w';
1319 else if (k.code == TUI_KEY_DOWN) key = 's';
1320 else if (k.code == TUI_KEY_LEFT) key = 'a';
1321 else if (k.code == TUI_KEY_RIGHT) key = 'd';
1322 else if (k.code == TUI_KEY_CHAR && k.ch < 0x80 && strchr("wsadqe", (int)ascii_lower(k.ch)) && ascii_lower(k.ch))
1323 key = (char)ascii_lower(k.ch);
1324 if (key) return send(app->focus, xfmt("input doom %c", key));
1325 if (is_char(k, 'p') || is_char(k, 'v')) {
1326 const mt_machine *m = mt_app_focused(app);
1327 char mn[256];
1328 mt_machine_name(m, mn, sizeof mn);
1329 const mt_section *d = mt_machine_program(m, "doom");
1330 mt_place cur;
1331 if (!d) return note(app, xfmt("%s: doom is not loaded", mn));
1332 if (!mt_place_of(d, &cur)) return note(app, xfmt("%s: doom declares no placement", mn));
1333 mt_place to;
1334 memset(&to, 0, sizeof to);
1335 if (is_char(k, 'v')) {
1336 if (cur.kind == MT_PLACE_SPLIT) to.kind = MT_PLACE_LOCAL;
1337 else {
1338 to.kind = MT_PLACE_SPLIT;
1339 memset(to.assign, MESH_PART_NONE, sizeof to.assign);
1340 }
1341 } else {
1342 to = mt_place_next(cur, mt_app_cluster_size(app));
1343 }
1344 char label[32];
1345 mt_place_label(to, label);
1346 return send(app->focus, xfmt("place doom %s", label));
1347 }
1348 if (k.code == TUI_KEY_ESC) app->tab = MT_TAB_MESH;
1349 else common_key(app, k);
1350 return none();
1351 }
1352 case MT_TAB_PROGRAMS: return programs_key(app, k);
1353 case MT_TAB_FLEET: return fleet_key(app, k);
1354 case MT_TAB_JOBS: return jobs_key(app, k);
1355 case MT_TAB_RENDER: return render_key(app, k);
1356 default: return default_key(app, k);
1357 }
1358}
1359
1360/* ---- update constructors */
1361
1362mt_update mt_update_snap(size_t i, mt_snapshot s) {
1363 mt_update u;
1364 memset(&u, 0, sizeof u);
1365 u.kind = MT_UPDATE_SNAP;
1366 u.machine = i;
1367 u.snap = s;
1368 return u;
1369}
1370
1371mt_update mt_update_reply_bytes(size_t i, const char *line, const void *data, size_t len) {
1372 mt_update u;
1373 memset(&u, 0, sizeof u);
1374 u.kind = MT_UPDATE_REPLY;
1375 u.machine = i;
1376 u.text = xstrdup(line);
1377 u.bytes = true;
1378 u.data = xmalloc(len);
1379 if (len) memcpy(u.data, data, len);
1380 u.len = len;
1381 return u;
1382}
1383
1384mt_update mt_update_reply_text(size_t i, const char *line, const char *text) {
1385 mt_update u;
1386 memset(&u, 0, sizeof u);
1387 u.kind = MT_UPDATE_REPLY;
1388 u.machine = i;
1389 u.text = xstrdup(line);
1390 u.data = (uint8_t *)xstrdup(text);
1391 u.len = strlen(text);
1392 return u;
1393}
1394
1395mt_update mt_update_offline(size_t i, const char *why) {
1396 mt_update u;
1397 memset(&u, 0, sizeof u);
1398 u.kind = MT_UPDATE_OFFLINE;
1399 u.machine = i;
1400 u.text = xstrdup(why);
1401 return u;
1402}
1403
1404mt_update mt_update_online(size_t i) {
1405 mt_update u;
1406 memset(&u, 0, sizeof u);
1407 u.kind = MT_UPDATE_ONLINE;
1408 u.machine = i;
1409 return u;
1410}