==> src/publish.c <==1/* peer.rs publish, fleet_text, groups_text, set_member, group_line, and peer_sched.rs sched_publish and
2 * fleet_recs: what the fleet thread publishes (the status block's fleet, member and scheduler regions,
3 * the `fleet`, `groups` and `sched` texts, and the fleet view's records in views.rs's text form). */
4#include "relay.h"
5#include "mesh/abi/group_state.h"
6#include "mesh/abi/host_fleet.h"
7#include "mesh/abi/host_members.h"
8#include "mesh/abi/host_sched.h"
9#include "mesh/abi/program_table.h"
10#include "mesh/abi/reconcile.h"
11#include "mesh/rust_fmt.h"
12#include "mesh/sched_needs.h"
13#include "mesh/text.h"
14
15#include <math.h>
16#include <stdio.h>
17#include <stdlib.h>
18#include <string.h>
19
20#define SUSPECT_NONE 0
21
22static const char *gl_name(uint32_t s) { return s < MESH_GL_NAMES_LEN ? MESH_GL_NAMES[s] : "?"; }
23
24/* fleet::members_of(mask) as sched::list writes it ("-" when empty). */
25static char *mask_list(uint32_t mask) {
26 ml_list l = ml_members(mask);
27 char *s = ml_text(&l);
28 ml_free(&l);
29 return s;
30}
31
32/* The reconcile word: "pending" while behind (and not failed), else RS_NAMES. */
33static const char *rs_word(uint64_t applied, uint64_t version, uint32_t st) {
34 if (applied < version && st != MESH_RS_FAILED) return "pending";
35 return st < MESH_RS_NAMES_LEN ? MESH_RS_NAMES[st] : "?";
36}
37
38/* ---- views.rs Rec, written straight into its text form */
39
40static void rec(mt_buf *o, const char *kind) { mt_puts(o, kind); }
41
42/* A field: its value one token (whitespace and '=' become '_', empty becomes '-'). */
43static void kv(mt_buf *o, const char *k, const char *v) {
44 mt_printf(o, " %s=", k);
45 if (!v || !*v) {
46 mt_putc(o, '-');
47 return;
48 }
49 size_t n = strlen(v);
50 for (size_t i = 0; i < n;) {
51 size_t ws = mesh_utf8_ws_len(v + i, n - i);
52 if (ws) {
53 mt_putc(o, '_');
54 i += ws;
55 } else {
56 mt_putc(o, v[i] == '=' ? '_' : v[i]);
57 i++;
58 }
59 }
60}
61
62static void kv_own(mt_buf *o, const char *k, char *v) {
63 kv(o, k, v);
64 free(v);
65}
66
67static void kvu(mt_buf *o, const char *k, uint64_t v) { mt_printf(o, " %s=%llu", k, (unsigned long long)v); }
68static void kvi(mt_buf *o, const char *k, int64_t v) { mt_printf(o, " %s=%lld", k, (long long)v); }
69static void opt_u(mt_buf *o, const char *k, bool some, uint64_t v) {
70 if (some) kvu(o, k, v);
71 else kv(o, k, "-");
72}
73
74/* A float with p decimals ('-' when absent or not finite). */
75static void kf(mt_buf *o, const char *k, bool some, double v, int p) {
76 if (some && isfinite(v)) mt_printf(o, " %s=%.*f", k, p, v);
77 else kv(o, k, "-");
78}
79
80/* The text (newlines become spaces) and the end of the record. */
81static void rec_end(mt_buf *o, const char *text) {
82 if (text && *text) {
83 mt_puts(o, " -- ");
84 for (const char *c = text; *c; c++) mt_putc(o, *c == '\n' || *c == '\r' ? ' ' : *c);
85 }
86 mt_putc(o, '\n');
87}
88
89/* ---- the status block */
90
91void publish_member(mpr_relay *r, uint32_t k, uint32_t state, const mpk_caps *c) {
92 const mpr_host *h = &r->h;
93 size_t m = mesh_member(k);
94 h->host_u32(h->ctx, m + MESH_M_STATE, state);
95 if (!c) return;
96 h->host_u32(h->ctx, m + MESH_M_CC, c->cc);
97 h->host_u32(h->ctx, m + MESH_M_SMS, c->sms);
98 h->host_u32(h->ctx, m + MESH_M_SMS_TOTAL, c->sms_total);
99 h->host_u32(h->ctx, m + MESH_M_BLOCKS_PER_SM, c->blocks_per_sm);
100 h->host_str(h->ctx, m + MESH_M_GPU, 32, c->gpu);
101 h->host_str(h->ctx, m + MESH_M_ARCH, 16, c->arch);
102 h->host_str(h->ctx, m + MESH_M_HOST, 32, c->host);
103 h->host_u32(h->ctx, m + MESH_M_BACKEND, c->backend);
104}
105
106char *publish_group_line(const mfd_group *g) {
107 char *l = ml_text(&g->members), *a = g->auto_ms > 0 ? mt_fmt(" auto_ms %u", g->auto_ms) : mt_strdup("");
108 char *s = mt_fmt("group %s program %s epoch %u cycle %u members %s state %s on_loss %s%s", g->name, g->program, g->epoch, g->cycle, l,
109 g->state == MFD_RUNNING ? "running" : "paused", g->on_loss == MFD_LOSS_PAUSE ? "pause" : "reform", a);
110 free(l);
111 free(a);
112 return s;
113}
114
115/* A member's link state as `fleet` and the view say it. */
116static const char *link_word(mpr_relay *r, uint32_t k, uint32_t alive, uint32_t suspect) {
117 if (!(alive >> k & 1)) return relay_linked(r, k) ? "linked" : "down";
118 return suspect >> k & 1 ? "suspect" : "up";
119}
120
121/* `fleet` on the control port. */
122static char *fleet_text(mpr_relay *r, uint64_t now, uint32_t alive, uint32_t suspect, const mpr_recon *rs) {
123 const mfd_doc *doc = mfc_committed(r->core);
124 const mfc_config *cfg = mfc_cfg(r->core);
125 uint32_t leader;
126 bool has_leader = mfc_leader(r->core, &leader);
127 mt_buf s = {0};
128 char *up = mask_list(alive), *sus = mask_list(suspect);
129 char lw[16] = "-";
130 if (has_leader) snprintf(lw, sizeof lw, "%u", leader);
131 mt_printf(&s, "fleet system %u of %u role %s leader %s term %llu version %llu accepted %llu applied %llu quorum %u up %s suspect %s %s\n", r->me, r->n,
132 mfc_role_word(mfc_role_of(r->core)), lw, (unsigned long long)mfc_term(r->core), (unsigned long long)doc->version,
133 (unsigned long long)mfc_accepted(r->core)->version, (unsigned long long)rs->applied, mfc_quorum(cfg), up, sus,
134 mfc_fixed(cfg) ? "fixed-coordinator" : mfc_has_quorum(r->core, now) ? "majority" : "no-majority");
135 free(up);
136 free(sus);
137 uint64_t hs = r->h.uptime_ms(r->h.ctx);
138 for (uint32_t k = 0; k < r->n; k++) {
139 const char *state, *why, *host;
140 uint64_t applied;
141 uint32_t st;
142 char heard[24] = "-";
143 if (k == r->me) {
144 state = "up", applied = rs->applied, st = rs->status, why = rs->reason ? rs->reason : "";
145 snprintf(heard, sizeof heard, "0");
146 host = r->my_caps.host;
147 } else {
148 const mpk_beat_status *b = &r->st[k];
149 state = link_word(r, k, alive, suspect), applied = b->applied, st = b->status, why = b->reason ? b->reason : "";
150 bool linked;
151 uint64_t rx = relay_last_rx(r, k, &linked);
152 if (linked) snprintf(heard, sizeof heard, "%llu", (unsigned long long)(hs > rx ? hs - rx : 0));
153 host = r->has_caps[k] ? r->caps[k].host : "?";
154 }
155 mt_printf(&s, "member %u %s reconcile %s applied %llu heard_ms %s host %s%s", k, state, rs_word(applied, doc->version, st), (unsigned long long)applied,
156 heard, host, has_leader && leader == k ? " leader" : "");
157 if (*why && strcmp(why, "ok") != 0) mt_printf(&s, st == MESH_RS_SYNC ? " last %s" : " reason %s", why);
158 mt_putc(&s, '\n');
159 }
160 for (size_t i = 0; i < doc->nprogs; i++) {
161 const mfd_prog *p = &doc->prog[i];
162 char *at = ml_at_word(&p->at);
163 mt_printf(&s, "program %s dir %s hash %016llx run %u%s", doc->pname[i], p->dir, (unsigned long long)p->hash, (unsigned)p->run, p->lockstep ? " lockstep" : "");
164 if (p->has_prev) mt_printf(&s, " prev %s %016llx", p->prev_dir, (unsigned long long)p->prev_hash);
165 mt_printf(&s, " at %s", at);
166 free(at);
167 if (p->pin.some) {
168 char *l = ml_text(&p->pin.l);
169 mt_printf(&s, " pin %s", l);
170 free(l);
171 }
172 if (p->has_from) mt_printf(&s, " from %u", p->from);
173 mt_putc(&s, '\n');
174 }
175 for (size_t i = 0; i < doc->ngroups; i++) {
176 char *g = publish_group_line(&doc->groups[i]);
177 mt_printf(&s, "%s\n", g);
178 free(g);
179 }
180 for (size_t i = 0; i < doc->nplaces; i++) {
181 char *w = mt_join(&doc->place_words[i], " ");
182 mt_printf(&s, "place %s %s\n", doc->place_name[i], w);
183 free(w);
184 }
185 return mt_take(&s);
186}
187
188/* `groups` on the control port. */
189static char *groups_text(mpr_relay *r) {
190 const mfd_doc *doc = mfc_committed(r->core);
191 mt_buf s = {0};
192 for (size_t i = 0; i < doc->ngroups; i++) {
193 const mfd_group *g = &doc->groups[i];
194 char *line = publish_group_line(g);
195 uint32_t rank;
196 char rw[16] = "-";
197 if (mfd_group_rank(g, r->me, &rank)) snprintf(rw, sizeof rw, "%u", rank);
198 mt_printf(&s, "%s local %s rank %s", line, gl_name(atomic_load(&r->hello_state)), rw);
199 free(line);
200 for (size_t j = 0; j < g->members.n; j++) {
201 uint32_t k = g->members.v[j];
202 if (k == r->me) continue;
203 mt_printf(&s, " member%u %s@%u", k, gl_name(r->st[k].group_state), r->st[k].group_epoch);
204 }
205 mt_printf(&s, " why %s\n", r->hello_why && *r->hello_why ? r->hello_why : "-");
206 }
207 return s.len ? mt_take(&s) : mt_strdup("");
208}
209
210void publish_all(mpr_relay *r, uint64_t now) {
211 r->last_publish_ms = mpr_mono_ms();
212 const mpr_host *h = &r->h;
213 uint32_t alive = mfc_alive(r->core, now);
214 uint64_t suspect_after = r->tun.dead_ms / 3;
215 uint32_t suspect = SUSPECT_NONE;
216 uint64_t hs = h->uptime_ms(h->ctx);
217 for (uint32_t k = 0; k < r->n; k++) {
218 if (k == r->me) continue;
219 bool linked;
220 uint64_t rx = relay_last_rx(r, k, &linked);
221 uint32_t st;
222 if (!linked) st = MESH_MS_ABSENT;
223 else if (!(alive >> k & 1)) st = MESH_MS_LINKED;
224 else if ((hs > rx ? hs - rx : 0) > suspect_after) suspect |= 1u << k, st = MESH_MS_SUSPECT;
225 else st = MESH_MS_UP;
226 publish_member(r, k, st, NULL);
227 }
228 const mfd_doc *doc = mfc_committed(r->core);
229 mpr_recon rs = {0};
230 h->recon(h->ctx, &rs);
231 const mfc_config *cfg = mfc_cfg(r->core);
232 uint32_t leader;
233 h->host_u32(h->ctx, MESH_S_LEADER, mfc_leader(r->core, &leader) ? leader : MESH_PEER_NONE);
234 h->host_u32(h->ctx, MESH_S_ROLE, mfc_role_code(mfc_role_of(r->core)));
235 h->host_u64(h->ctx, MESH_S_TERM, mfc_term(r->core));
236 h->host_u64(h->ctx, MESH_S_STATE_VER, doc->version);
237 h->host_u64(h->ctx, MESH_S_APPLIED_VER, rs.applied);
238 h->host_u32(h->ctx, MESH_S_UP_MASK, alive);
239 h->host_u32(h->ctx, MESH_S_SUSPECT_MASK, suspect);
240 h->host_u32(h->ctx, MESH_S_QUORUM, mfc_quorum(cfg));
241 h->host_u32(h->ctx, MESH_S_FLEET_FLAGS, (mfc_has_quorum(r->core, now) ? MESH_FF_QUORUM : 0) | (mfc_fixed(cfg) ? MESH_FF_FIXED : 0));
242 h->host_u32(h->ctx, MESH_S_DEAD_MS, (uint32_t)r->tun.dead_ms);
243 h->host_u32(h->ctx, MESH_S_COMMIT_TERM, (uint32_t)doc->term);
244 uint32_t hello_state = atomic_load(&r->hello_state);
245 for (uint32_t k = 0; k < r->n; k++) {
246 size_t f = mesh_fmember(k);
247 uint32_t st, gs;
248 uint64_t ap, heard;
249 const char *why;
250 if (k == r->me) st = rs.status, gs = hello_state, ap = rs.applied, why = rs.reason, heard = hs;
251 else {
252 const mpk_beat_status *s = &r->st[k];
253 bool linked;
254 uint64_t rx = relay_last_rx(r, k, &linked);
255 st = s->status, gs = s->group_state, ap = s->applied, why = s->reason, heard = linked ? rx : 0;
256 }
257 h->host_u32(h->ctx, f + MESH_FM_STATUS, st);
258 h->host_u32(h->ctx, f + MESH_FM_GROUP, gs);
259 h->host_u64(h->ctx, f + MESH_FM_APPLIED, ap);
260 h->host_u64(h->ctx, f + MESH_FM_HEARD_MS, heard);
261 h->host_str(h->ctx, f + MESH_FM_REASON, 40, why ? why : "");
262 }
263 if (doc->ngroups) {
264 const mfd_group *g = &doc->groups[0];
265 uint8_t order[MESH_MAX_SYSTEMS];
266 uint32_t rank;
267 mfd_group_order(g, order);
268 h->host_str(h->ctx, MESH_SG_NAME, MESH_NAME_BYTES, g->name);
269 h->host_u32(h->ctx, MESH_SG_EPOCH, g->epoch);
270 h->host_u32(h->ctx, MESH_SG_CYCLE, g->cycle);
271 h->host_u32(h->ctx, MESH_SG_MASK, mfd_group_mask(g));
272 h->host_bytes(h->ctx, MESH_SG_ORDER, order, sizeof order);
273 h->host_u32(h->ctx, MESH_SG_STATE, g->state == MFD_RUNNING ? 1 : 2);
274 h->host_u32(h->ctx, MESH_SG_RANK, mfd_group_rank(g, r->me, &rank) ? rank : MESH_PEER_NONE);
275 h->host_u32(h->ctx, MESH_SG_ON_LOSS, g->on_loss == MFD_LOSS_PAUSE ? 0 : 1);
276 h->host_u32(h->ctx, MESH_SG_AUTO_MS, g->auto_ms);
277 } else {
278 h->host_str(h->ctx, MESH_SG_NAME, MESH_NAME_BYTES, "");
279 h->host_u32(h->ctx, MESH_SG_STATE, 0);
280 h->host_u32(h->ctx, MESH_SG_MASK, 0);
281 h->host_u32(h->ctx, MESH_SG_RANK, MESH_PEER_NONE);
282 }
283 h->host_u32(h->ctx, MESH_SG_LOCAL, hello_state);
284 char *ft = fleet_text(r, now, alive, suspect, &rs), *gt = groups_text(r);
285 h->texts(h->ctx, ft, gt);
286 free(ft);
287 free(gt);
288 if (mpr_mono_ms() - r->last_sched_text_ms >= 1000) {
289 r->last_sched_text_ms = mpr_mono_ms();
290 sched_publish(r, now, alive, suspect, &rs);
291 }
292 mpr_recon_free(&rs);
293}
294
295/* ---- the scheduler's texts and the fleet view */
296
297/* The program table's slot s: its name (empty slot: false). */
298static bool slot_name(mpr_relay *r, uint32_t s, char name[MESH_NAME_BYTES + 1]) {
299 size_t e = MESH_PROG_TABLE + (size_t)s * MESH_PT_BYTES;
300 if (r->h.mb_get32(r->h.ctx, e + MESH_PT_STATE) == MESH_ST_EMPTY) return false;
301 r->h.mb_read(r->h.ctx, e + MESH_PT_NAME, (uint8_t *)name, MESH_NAME_BYTES);
302 name[MESH_NAME_BYTES] = 0;
303 return true;
304}
305
306static void opt_f(char *out, size_t cap, mof32 v, int p) {
307 if (v.some) snprintf(out, cap, "%.*f", p, (double)v.v);
308 else snprintf(out, cap, "-");
309}
310
311/* views.rs records: fleet, member, program, cell, explain, group, gmember, proposal, migration,
312 * decision, note. */
313static char *fleet_recs(mpr_relay *r, const mfd_doc *doc, const msm_member *ms, size_t nms, const msm_perfs *perf, bool leading, uint32_t alive, uint32_t suspect,
314 const mpr_recon *rs) {
315 mt_buf o = {0};
316 const mfc_config *cfg = mfc_cfg(r->core);
317 uint32_t leader;
318 bool has_leader = mfc_leader(r->core, &leader);
319 ml_list busy = {0};
320 for (size_t i = 0; i < nms; i++)
321 if (ms[i].busy) ml_push(&busy, ms[i].id);
322 rec(&o, "fleet");
323 kvu(&o, "system", r->me);
324 kvu(&o, "n", r->n);
325 kv(&o, "role", mfc_role_word(mfc_role_of(r->core)));
326 opt_u(&o, "leader", has_leader, leader);
327 kvu(&o, "term", mfc_term(r->core));
328 kvu(&o, "version", doc->version);
329 kvu(&o, "applied", rs->applied);
330 kvu(&o, "quorum", mfc_quorum(cfg));
331 kv_own(&o, "up", mask_list(alive));
332 kv_own(&o, "suspect", mask_list(suspect));
333 kv(&o, "majority", mfc_fixed(cfg) ? "fixed" : mfc_has_quorum(r->core, relay_now(r)) ? "yes" : "no");
334 kvu(&o, "paused", doc->sched_paused);
335 kvu(&o, "autoreform", doc->autoreform);
336 kv_own(&o, "cordon", mask_list(doc->cordon));
337 kv_own(&o, "busy_members", ml_text(&busy));
338 opt_u(&o, "migrations", leading, r->moves_done);
339 kf(&o, "last_gap_ms", leading && r->last_gap_us > 0, (double)r->last_gap_us / 1e3, 1);
340 rec_end(&o, NULL);
341 ml_free(&busy);
342 uint64_t hs = r->h.uptime_ms(r->h.ctx);
343 uint32_t hello_state = atomic_load(&r->hello_state);
344 for (size_t i = 0; i < nms; i++) {
345 const msm_member *m = &ms[i];
346 uint32_t k = m->id, st, gs;
347 const char *state, *why;
348 uint64_t applied;
349 bool has_heard = true;
350 uint64_t heard = 0;
351 if (k == r->me) state = "up", applied = rs->applied, st = rs->status, why = rs->reason, gs = hello_state;
352 else {
353 const mpk_beat_status *b = &r->st[k];
354 state = link_word(r, k, alive, suspect), applied = b->applied, st = b->status, why = b->reason, gs = b->group_state;
355 uint64_t rx = relay_last_rx(r, k, &has_heard);
356 heard = hs > rx ? hs - rx : 0;
357 }
358 if (!why) why = "";
359 char *text;
360 if (m->busy) text = mt_fmt("busy: %s", m->busy);
361 else if (!*why || !strcmp(why, "ok")) text = mt_strdup("");
362 else if (st == MESH_RS_SYNC) text = mt_fmt("last %s", why);
363 else text = mt_strdup(why);
364 rec(&o, "member");
365 kvu(&o, "id", k);
366 kv(&o, "state", state);
367 kvu(&o, "leader", has_leader && leader == k);
368 kv(&o, "host", m->host);
369 kv(&o, "gpu", m->gpu);
370 kv(&o, "arch", m->arch);
371 kvu(&o, "cpus", m->cpus);
372 kvu(&o, "big", m->cores_big);
373 kvu(&o, "little", m->cores_little);
374 kvu(&o, "ram_mb", m->ram_mb);
375 kvu(&o, "avail_mb", m->mem_avail_mb);
376 kvu(&o, "disk_mb", m->disk_free_mb);
377 kf(&o, "share", m->gpu_share.some, m->gpu_share.v, 2);
378 kf(&o, "cpu", m->load_known, m->cpu_util, 2);
379 kf(&o, "psi_cpu", m->psi_cpu.some, m->psi_cpu.v, 3);
380 kf(&o, "psi_mem", m->psi_mem.some, m->psi_mem.v, 3);
381 kf(&o, "meshd_cpu", m->load_known, m->meshd_cpu, 2);
382 kvu(&o, "slots_free", m->slots_free);
383 kvu(&o, "heap_kb", m->heap_kb);
384 kvu(&o, "cordoned", m->cordoned);
385 kvu(&o, "busy", m->busy != NULL);
386 kv(&o, "reconcile", rs_word(applied, doc->version, st));
387 kvu(&o, "applied", applied);
388 opt_u(&o, "heard_ms", has_heard, heard);
389 kv(&o, "group", gl_name(gs));
390 rec_end(&o, text);
391 free(text);
392 }
393 for (size_t i = 0; i < doc->nprogs; i++) {
394 const char *n = doc->pname[i];
395 const mfd_prog *p = &doc->prog[i];
396 rec(&o, "program");
397 kv(&o, "name", n);
398 kv_own(&o, "hash", mt_fmt("%016llx", (unsigned long long)p->hash));
399 kvu(&o, "run", p->run);
400 kvu(&o, "lockstep", p->lockstep);
401 kv_own(&o, "at", ml_at_word(&p->at));
402 kv_own(&o, "pin", p->pin.some ? ml_text(&p->pin.l) : mt_strdup("-"));
403 opt_u(&o, "from", p->has_from, p->from);
404 kv_own(&o, "replicas", msn_replicas_word(p->needs.replicas));
405 kv_own(&o, "backends", msn_strs_join(p->needs.backends, p->needs.nbackends, ","));
406 kvi(&o, "priority", p->needs.priority);
407 kv(&o, "migrate", mesh_migrate_word(p->needs.migrate));
408 kf(&o, "cpu", true, p->needs.cpu, 2);
409 kvu(&o, "mem_mb", p->needs.mem_mb);
410 char *text = p->why && *p->why ? mt_strdup(p->why) : msn_summary(&p->needs);
411 rec_end(&o, text);
412 free(text);
413 const msp_explain *ex = msp_explains_get(&r->explains, n);
414 for (size_t j = 0; j < nms; j++) {
415 uint32_t k = ms[j].id;
416 const msp_row *row = NULL;
417 for (size_t x = 0; ex && x < ex->nrows; x++)
418 if (ex->rows[x].k == k) row = &ex->rows[x];
419 const msm_perf *pf = msm_perfs_get(perf, n, k);
420 bool placed = p->at.some ? ml_has(&p->at.l, k) : (row && row->ok) || pf != NULL;
421 const char *moving = p->has_from && p->from == k ? "from" : p->has_from && placed ? "to" : "-";
422 rec(&o, "cell");
423 kv(&o, "program", n);
424 kvu(&o, "member", k);
425 kvu(&o, "placed", placed);
426 kvu(&o, "pinned", p->pin.some && ml_has(&p->pin.l, k));
427 kv(&o, "moving", moving);
428 if (placed && (p->needs.cpu > 0.0f || p->needs.mem_mb > 0)) {
429 mt_buf res = {0};
430 if (p->needs.cpu > 0.0f) mt_printf(&res, "cpu:%.2f", (double)p->needs.cpu);
431 if (p->needs.mem_mb > 0) mt_printf(&res, "%smem_mb:%u", p->needs.cpu > 0.0f ? "," : "", p->needs.mem_mb);
432 kv_own(&o, "reserved", mt_take(&res));
433 } else kv(&o, "reserved", "-");
434 kv(&o, "state", !pf ? "-" : pf->run ? "running" : "stopped");
435 kv(&o, "exec", pf ? pf->exec : "-");
436 kf(&o, "task_us", pf && pf->task_us.some, pf ? pf->task_us.v : 0, 1);
437 kf(&o, "tasks_s", pf != NULL, pf ? pf->tasks_s : 0, 1);
438 rec_end(&o, NULL);
439 if (!row) continue;
440 rec(&o, "explain");
441 kv(&o, "program", n);
442 kvu(&o, "member", k);
443 if (row->ok) {
444 bool on = (!ex->chosen.some || ml_has(&ex->chosen.l, k));
445 kv(&o, "verdict", on ? "chosen" : "feasible");
446 kf(&o, "score", true, row->score.total, 3);
447 char *t = msc_explain(&row->score);
448 rec_end(&o, t);
449 free(t);
450 } else {
451 kv(&o, "verdict", "rejected");
452 kv(&o, "score", "-");
453 rec_end(&o, row->why);
454 }
455 }
456 }
457 for (size_t i = 0; i < doc->ngroups; i++) {
458 const mfd_group *g = &doc->groups[i];
459 rec(&o, "group");
460 kv(&o, "name", g->name);
461 kv(&o, "program", g->program);
462 kvu(&o, "epoch", g->epoch);
463 kvu(&o, "cycle", g->cycle);
464 kv_own(&o, "members", ml_text(&g->members));
465 kv(&o, "state", g->state == MFD_RUNNING ? "running" : "paused");
466 kv(&o, "on_loss", g->on_loss == MFD_LOSS_PAUSE ? "pause" : "reform");
467 kvu(&o, "auto_ms", g->auto_ms);
468 rec_end(&o, NULL);
469 for (size_t j = 0; j < g->members.n; j++) {
470 uint32_t k = g->members.v[j], gs, ep, rank;
471 if (k == r->me) gs = hello_state, ep = r->has_hello ? r->hello.epoch : 0;
472 else gs = r->st[k].group_state, ep = r->st[k].group_epoch;
473 bool has_rank = mfd_group_rank(g, k, &rank);
474 rec(&o, "gmember");
475 kv(&o, "group", g->name);
476 kvu(&o, "member", k);
477 opt_u(&o, "rank", has_rank, rank);
478 kv(&o, "state", gl_name(gs));
479 kvu(&o, "epoch", ep);
480 rec_end(&o, NULL);
481 }
482 }
483 if (r->group_prop) {
484 rec(&o, "proposal");
485 rec_end(&o, r->group_prop);
486 }
487 if (r->has_mig) {
488 rec(&o, "migration");
489 kv_own(&o, "programs", mt_join(&r->mig.mv.programs, ","));
490 kvu(&o, "from", r->mig.mv.from);
491 kvu(&o, "to", r->mig.mv.to);
492 kvu(&o, "version", r->mig.version);
493 kf(&o, "secs", true, (double)(mpr_mono_ms() - r->mig.t0_ms) / 1e3, 1);
494 rec_end(&o, r->mig.mv.why);
495 }
496 for (size_t i = r->sched_log.n; i-- > 0;) {
497 const char *d = r->sched_log.v[i], *sep = strstr(d, " s: ");
498 rec(&o, "decision");
499 if (sep) {
500 char *at = mt_strndup(d, (size_t)(sep - d));
501 kv_own(&o, "at_s", at);
502 rec_end(&o, sep + 4);
503 } else {
504 kv(&o, "at_s", "-");
505 rec_end(&o, d);
506 }
507 }
508 if (leading) {
509 const mt_strs *notes = mst_notes(r->tracker);
510 for (size_t i = 0; i < notes->n; i++) {
511 rec(&o, "note");
512 rec_end(&o, notes->v[i]);
513 }
514 }
515 return mt_take(&o);
516}
517
518/* `sched` (every member computes it from what it knows; the leader adds its rebalancer's state), the
519 * placed word of every program slot, the fleet view, and the jobs dispatcher's model. */
520void sched_publish(mpr_relay *r, uint64_t now, uint32_t alive, uint32_t suspect, const mpr_recon *rs) {
521 const mpr_host *h = &r->h;
522 mfd_doc doc = mfd_doc_clone(mfc_committed(r->core));
523 bool leading = mfc_role_of(r->core) == MFC_LEADER;
524 /* The explain: the leader's from its last pass; another member's computed here. */
525 if (!leading || r->explains.n == 0) {
526 mt_strs ch = {0};
527 mfd_doc d = sched_pass(r, &doc, now, &ch);
528 mfd_doc_free(&d);
529 mt_strs_free(&ch);
530 }
531 size_t nms;
532 msm_member *ms = sched_members(r, now, &nms);
533 if (leading) mst_owner(r->tracker, ms, nms, now, &r->w);
534 bool migrating = false;
535 for (size_t i = 0; i < doc.nprogs; i++) migrating |= doc.prog[i].has_from;
536 h->host_u32(h->ctx, MESH_S_SCHED_FLAGS, (doc.sched_paused ? MESH_SF_PAUSED : 0) | (doc.autoreform ? MESH_SF_AUTOREFORM : 0) | (migrating ? MESH_SF_MIGRATING : 0));
537 h->host_u32(h->ctx, MESH_S_CORDON, doc.cordon);
538 uint32_t busy = 0;
539 ml_list busy_l = {0};
540 for (size_t i = 0; i < nms; i++)
541 if (ms[i].busy) busy |= 1u << ms[i].id, ml_push(&busy_l, ms[i].id);
542 h->host_u32(h->ctx, MESH_S_BUSY_MASK, busy);
543 if (leading) {
544 h->host_u32(h->ctx, MESH_S_SCHED_MOVES, r->moves_done);
545 h->host_u32(h->ctx, MESH_S_SCHED_GAP_US, r->last_gap_us);
546 }
547 for (uint32_t s = 0; s < MESH_PROG_SLOTS; s++) {
548 char name[MESH_NAME_BYTES + 1];
549 uint32_t w = 0;
550 const mfd_prog *p = slot_name(r, s, name) ? mfd_prog_get(&doc, name) : NULL;
551 if (p) {
552 uint32_t at = p->at.some ? ml_mask(&p->at.l) & MESH_PS_MASK : MESH_PS_ALL;
553 w = MESH_PS_KNOWN | at | (p->pin.some ? MESH_PS_PINNED : 0) | (p->has_from ? MESH_PS_MIGRATING | (p->from + 1) << MESH_PS_FROM_SHIFT : 0);
554 }
555 h->host_u32(h->ctx, mesh_placed(s), w);
556 }
557 /* The text. */
558 uint32_t leader;
559 bool has_leader = mfc_leader(r->core, &leader);
560 mt_buf t = {0};
561 char lw[16] = "-", gap[32] = "-", moves[16] = "-";
562 if (has_leader) snprintf(lw, sizeof lw, "%u", leader);
563 if (leading) snprintf(moves, sizeof moves, "%u", r->moves_done);
564 if (leading && r->last_gap_us > 0) snprintf(gap, sizeof gap, "%.1f", (double)r->last_gap_us / 1e3);
565 char *cordon = mask_list(doc.cordon), *bl = ml_text(&busy_l);
566 mt_printf(&t, "sched system %u leader %s version %llu paused %u autoreform %u cordon %s busy %s migrations %s last_gap_ms %s\n", r->me, lw,
567 (unsigned long long)doc.version, (unsigned)doc.sched_paused, (unsigned)doc.autoreform, cordon, bl, moves, gap);
568 free(cordon);
569 free(bl);
570 ml_free(&busy_l);
571 char os[RF_FLOAT_MAX], op[RF_FLOAT_MAX];
572 mt_printf(&t, "policy improve %.0f%% hold_ms %llu dwell_ms %llu per_min %u replace_ms %llu owner share<%s psi>%s for %llu ms; tick_ms %llu\n", r->w.improve * 100.0,
573 (unsigned long long)r->w.hold_ms, (unsigned long long)r->w.dwell_ms, r->w.per_min, (unsigned long long)r->w.replace_ms, rf_f32(os, r->w.owner_share),
574 rf_f32(op, r->w.owner_psi), (unsigned long long)r->w.owner_ms, (unsigned long long)r->w.tick_ms);
575 for (size_t i = 0; i < nms; i++) {
576 const msm_member *m = &ms[i];
577 char share[32], pc[32], pm[32];
578 opt_f(share, sizeof share, m->gpu_share, 2);
579 opt_f(pc, sizeof pc, m->psi_cpu, 3);
580 opt_f(pm, sizeof pm, m->psi_mem, 3);
581 mt_buf rtt = {0};
582 for (size_t j = 0; j < m->nrtt; j++)
583 if (m->rtt_us[j].some) mt_printf(&rtt, "%s%zu:%.0f", rtt.len ? "," : "", j, (double)m->rtt_us[j].v);
584 char *rt = rtt.len ? mt_take(&rtt) : mt_strdup("-");
585 mt_printf(&t, "member %u %s host %s gpu %s arch %s cpus %u ram_mb %u avail_mb %u share %s cpu %.2f psi_cpu %s psi_mem %s meshd %.2f rtt_us %s%s", m->id,
586 m->up ? "up" : "down", *m->host ? m->host : "?", *m->gpu ? m->gpu : "?", m->arch, m->cpus, m->ram_mb, m->mem_avail_mb, share, (double)m->cpu_util, pc, pm,
587 (double)m->meshd_cpu, rt, m->cordoned ? " cordoned" : "");
588 free(rt);
589 if (m->busy) mt_printf(&t, " busy: %s", m->busy);
590 mt_putc(&t, '\n');
591 }
592 for (size_t i = 0; i < doc.nprogs; i++) {
593 const mfd_prog *p = &doc.prog[i];
594 char *at = ml_at_word(&p->at), *pin = p->pin.some ? ml_text(&p->pin.l) : mt_strdup("-"), *sum = msn_summary(&p->needs);
595 char from[16] = "-";
596 if (p->has_from) snprintf(from, sizeof from, "%u", p->from);
597 mt_printf(&t, "program %s at %s pin %s from %s %s why %s\n", doc.pname[i], at, pin, from, sum, p->why && *p->why ? p->why : "-");
598 free(at), free(pin), free(sum);
599 }
600 if (r->has_mig) {
601 char *pl = mt_join(&r->mig.mv.programs, ",");
602 mt_printf(&t, "migration %s from %u to %u version %llu for %.1f s\n", pl, r->mig.mv.from, r->mig.mv.to, (unsigned long long)r->mig.version,
603 (double)(mpr_mono_ms() - r->mig.t0_ms) / 1e3);
604 free(pl);
605 }
606 if (leading) {
607 const mt_strs *notes = mst_notes(r->tracker);
608 for (size_t i = 0; i < notes->n; i++) mt_printf(&t, "note %s\n", notes->v[i]);
609 }
610 if (r->group_prop) mt_printf(&t, "proposal %s\n", r->group_prop);
611 for (size_t i = 0; i < r->sched_log.n; i++) mt_printf(&t, "log %s\n", r->sched_log.v[i]);
612 /* The dry run: what a pass and an unhurried rebalance would do now. */
613 msm_perfs perf = sched_perf_map(r, now);
614 msp_input in = {&doc, ms, nms, &perf, &r->w};
615 mt_strs ch = {0};
616 msp_explains ex0 = {0};
617 mfd_doc placed = msp_place(&in, &ch, &ex0);
618 mfd_doc_free(&placed);
619 msp_explains_free(&ex0);
620 mt_buf dry = {0};
621 for (size_t i = 0; i < ch.n; i++) mt_printf(&dry, "place %s\n", ch.v[i]);
622 mt_strs_free(&ch);
623 mst_tracker *eager = mst_new();
624 msw_weights w0 = r->w;
625 w0.hold_ms = 0, w0.dwell_ms = 0;
626 msp_input in0 = {&doc, ms, nms, &perf, &w0};
627 mst_move mv;
628 if (mst_step(eager, &in0, now, false, &mv)) {
629 char *pl = mt_join(&mv.programs, ",");
630 mt_printf(&dry, "move %s from %u to %u: %s\n", pl, mv.from, mv.to, mv.why);
631 free(pl);
632 mst_move_free(&mv);
633 }
634 mst_free(eager);
635 if (doc.ngroups) {
636 const mfd_group *g = &doc.groups[0];
637 ml_list members;
638 char *why;
639 if (msp_group_choice(&in, g, &members, &why)) {
640 if (!ml_eq(&members, &g->members)) {
641 char *l = ml_text(&members);
642 mt_printf(&dry, "reform group %s over %s: %s\n", g->name, l, why);
643 free(l);
644 }
645 ml_free(&members);
646 free(why);
647 }
648 }
649 char *dr = dry.len ? mt_take(&dry) : mt_strdup("nothing to change\n");
650 char *all = mt_take(&t);
651 char *recs = fleet_recs(r, &doc, ms, nms, &perf, leading, alive, suspect, rs);
652 h->sched_texts(h->ctx, all, &r->explains, dr, recs);
653 free(all), free(dr), free(recs);
654 /* The jobs dispatcher picks lease members with the same model (jobs::select). */
655 if (h->jobs_sched) h->jobs_sched(h->ctx, ms, nms, &doc, &perf, &r->w);
656 msm_perfs_free(&perf);
657 sched_members_free(ms, nms);
658 mfd_doc_free(&doc);
659}