==> src/sched.c <==1/* peer_sched.rs: the fleet thread's scheduler glue: the member model from F_JOIN, F_LOADREP and F_PERF;
2 * the leader's placement passes, rebalancing and migrations (mesh-sched-* decides, this commits); the
3 * view transfer of a migration (F_VIEW_REQ / F_VIEW).
4 *
5 * A migration of programs from member A to B (state):
6 * 1. the leader commits them placed on B with `from` A (A keeps them);
7 * 2. B's reconciler loads them stopped and asks A for each view, producers first; A checks the document
8 * names that move, stops the program and sends the view (F_VIEW, 1 MiB chunks);
9 * 3. B writes each view, starts them, consumers first, and reports the gap;
10 * 4. the leader sees B's outcome and commits `from` cleared: A unloads them. B failed, or lost: the
11 * leader puts them back on A. A lost: B starts them fresh, loudly. */
12#include "relay.h"
13#include "mesh/abi/backend.h"
14#include "mesh/abi/host_load.h"
15#include "mesh/abi/host_mload.h"
16#include "mesh/abi/peer_frames.h"
17#include "mesh/abi/peer_sizes.h"
18#include "mesh/abi/view_heap.h"
19#include "mesh/fleet_groups.h"
20
21#include <stdio.h>
22#include <stdlib.h>
23#include <string.h>
24#include <time.h>
25
26#define VIEW_WAIT_MS 30000u
27#define MIG_WAIT_MS 90000u /* a migration not done by then is put back */
28#define PLACE_MS 250u /* the placement pass's period on the leader */
29#define PERF_KEEP_MS 600000u /* measured performance is remembered this long after its member stopped reporting it */
30#define FAIL_KEEP_MS 300000u /* a program that failed on a member is not placed there again for this long */
31
32static uint32_t host_u32_at(mpr_relay *r, size_t off) {
33 uint32_t v;
34 r->h.host_read(r->h.ctx, off, (uint8_t *)&v, 4);
35 return v;
36}
37
38static float host_f32_at(mpr_relay *r, size_t off) {
39 float v;
40 r->h.host_read(r->h.ctx, off, (uint8_t *)&v, 4);
41 return v;
42}
43
44static uint64_t host_u64_at(mpr_relay *r, size_t off) {
45 uint64_t v;
46 r->h.host_read(r->h.ctx, off, (uint8_t *)&v, 8);
47 return v;
48}
49
50void sched_members_free(msm_member *ms, size_t n) {
51 for (size_t i = 0; i < n; i++) msm_member_free(&ms[i]);
52 free(ms);
53}
54
55static void fail_mem_set(mpr_relay *r, const char *name, uint32_t k, uint64_t hash, const char *why, bool uns, uint64_t at) {
56 for (size_t i = 0; i < r->nfail_mem; i++)
57 if (r->fail_mem[i].k == k && !strcmp(r->fail_mem[i].name, name)) {
58 r->fail_mem[i].hash = hash;
59 mt_set(&r->fail_mem[i].why, why);
60 r->fail_mem[i].unsupported = uns;
61 r->fail_mem[i].at = at;
62 return;
63 }
64 r->fail_mem = mt_realloc(r->fail_mem, (r->nfail_mem + 1) * sizeof *r->fail_mem);
65 r->fail_mem[r->nfail_mem++] = (fail_mem_t){mt_strdup(name), k, hash, mt_strdup(why), uns, at};
66}
67
68msm_member *sched_members(mpr_relay *r, uint64_t now, size_t *n) {
69 uint32_t alive = mfc_alive(r->core, now);
70 for (uint32_t k = 0; k < r->n; k++) {
71 bool up = alive >> k & 1;
72 if (up) r->down[k] = false;
73 else if (!r->down[k]) r->down[k] = true, r->down_since[k] = now;
74 }
75 const mfd_doc *doc = mfc_committed(r->core);
76 uint32_t cordon = doc->cordon;
77 uint64_t hs = r->h.uptime_ms(r->h.ctx);
78 board_t *b = &r->board;
79 pthread_mutex_lock(&b->mu);
80 /* Failures: what the members report now, remembered a while (at the program's current image). */
81 for (uint32_t k = 0; k < r->n; k++)
82 if (b->has[k])
83 for (size_t i = 0; i < b->rep[k].nfails; i++) {
84 const msr_fail *f = &b->rep[k].fails[i];
85 fail_mem_set(r, f->name, k, f->hash, f->why, f->unsupported, now);
86 }
87 for (size_t i = r->nfail_mem; i-- > 0;) {
88 fail_mem_t *f = &r->fail_mem[i];
89 const mfd_prog *p = mfd_prog_get(doc, f->name);
90 if (p && p->hash == f->hash && (f->unsupported || now - f->at < FAIL_KEEP_MS)) continue;
91 free(f->name), free(f->why);
92 r->fail_mem[i] = r->fail_mem[--r->nfail_mem];
93 }
94 msm_member *ms = mt_alloc(r->n * sizeof *ms);
95 for (uint32_t k = 0; k < r->n; k++) {
96 msm_member m = msm_member_default();
97 m.id = k;
98 m.up = alive >> k & 1;
99 m.down_ms = r->down[k] ? now - r->down_since[k] : 0;
100 m.cordoned = cordon >> k & 1;
101 const mpk_caps *c = k == r->me ? &r->my_caps : r->has_caps[k] ? &r->caps[k] : NULL;
102 uint32_t caps_kind = c ? c->backend : 0;
103 if (c) {
104 MSM_SET(m.host, c->host);
105 uint32_t kind = mesh_backend_kind(c->backend);
106 MSM_SET(m.gpu, kind < MESH_BACKENDS_LEN ? MESH_BACKENDS[kind] : "?");
107 MSM_SET(m.arch, c->arch);
108 m.cc = c->cc, m.sms = c->sms, m.blocks_per_sm = c->blocks_per_sm, m.max_threads = c->max_threads, m.max_shared = c->max_shared;
109 }
110 if (b->has[k]) {
111 const msr_report *rp = &b->rep[k];
112 m.cpus = rp->cpus, m.cores_big = rp->big, m.cores_little = rp->little, m.ram_mb = rp->ram_mb, m.disk_free_mb = rp->disk_free_mb;
113 m.heap_kb = rp->heap_kb, m.slots_free = rp->slots_free;
114 if (k != r->me) {
115 m.nrtt = r->n;
116 for (uint32_t j = 0; j < r->n; j++) {
117 m.rtt_us[j] = mof32_none();
118 for (size_t i = 0; i < rp->nrtt; i++)
119 if (rp->rtt[i].k == j) {
120 m.rtt_us[j] = mof32_some(rp->rtt[i].us);
121 break;
122 }
123 }
124 }
125 }
126 if (k == r->me) {
127 m.nrtt = r->n;
128 for (uint32_t j = 0; j < r->n; j++) m.rtt_us[j] = b->rtt_us[j];
129 }
130 for (size_t i = 0; i < r->nfail_mem; i++)
131 if (r->fail_mem[i].k == k) {
132 char *w = r->fail_mem[i].unsupported ? mt_fmt("unsupported (%s)", r->fail_mem[i].why) : mt_strdup(r->fail_mem[i].why);
133 ms_set(&m.fails, r->fail_mem[i].name, w, true);
134 free(w);
135 }
136 size_t l = mesh_mload(k);
137 uint64_t at = host_u64_at(r, l + MESH_ML_AT_MS);
138 if (at != 0 && (hs > at ? hs - at : 0) < 10000) {
139 uint32_t lf = host_u32_at(r, l + MESH_ML_FLAGS);
140 m.load_known = true;
141 /* Metal's figure is submission overhead and queueing, not time slicing: unknown to the scheduler. */
142 m.gpu_share = (lf & MESH_LF_GPU) && mesh_backend_kind(caps_kind) != MESH_BACKEND_METAL ? mof32_some(host_f32_at(r, l + MESH_ML_SHARE10)) : mof32_none();
143 m.cpu_util = host_f32_at(r, l + MESH_ML_CPU);
144 m.cpu_big = host_f32_at(r, l + MESH_ML_CPU_BIG);
145 m.cpu_little = host_f32_at(r, l + MESH_ML_CPU_LITTLE);
146 m.psi_cpu = lf & MESH_LF_PSI ? mof32_some(host_f32_at(r, l + MESH_ML_PSI_CPU)) : mof32_none();
147 m.psi_mem = lf & MESH_LF_PSI ? mof32_some(host_f32_at(r, l + MESH_ML_PSI_MEM)) : mof32_none();
148 m.meshd_cpu = host_f32_at(r, l + MESH_ML_MESHD_CPU);
149 /* Smoothed: the scheduler reads the averages. */
150 if (!r->has_load[k]) {
151 r->has_load[k] = true;
152 r->load_avg[k] = (mpv_load){0, m.cpu_util, m.psi_cpu, m.psi_mem, m.gpu_share};
153 }
154 mpv_load_add(&r->load_avg[k], at, m.cpu_util, m.psi_cpu, m.psi_mem, m.gpu_share);
155 m.cpu_util = r->load_avg[k].cpu, m.psi_cpu = r->load_avg[k].psi_cpu, m.psi_mem = r->load_avg[k].psi_mem, m.gpu_share = r->load_avg[k].share;
156 m.mem_avail_mb = host_u32_at(r, l + MESH_ML_MEM_AVAIL_MB);
157 if (m.ram_mb == 0) m.ram_mb = host_u32_at(r, l + MESH_ML_MEM_TOTAL_MB);
158 if (m.disk_free_mb == 0) m.disk_free_mb = host_u32_at(r, l + MESH_ML_DISK_FREE_MB);
159 }
160 ms[k] = m;
161 }
162 pthread_mutex_unlock(&b->mu);
163 *n = r->n;
164 return ms;
165}
166
167static perf_mem_t *perf_mem_get(mpr_relay *r, const char *name, uint32_t k) {
168 for (size_t i = 0; i < r->nperf_mem; i++)
169 if (r->perf_mem[i].k == k && !strcmp(r->perf_mem[i].name, name)) return &r->perf_mem[i];
170 return NULL;
171}
172
173msm_perfs sched_perf_map(mpr_relay *r, uint64_t now) {
174 board_t *b = &r->board;
175 uint64_t t = mpr_mono_us();
176 pthread_mutex_lock(&b->mu);
177 for (uint32_t k = 0; k < r->n; k++) {
178 if (!b->has[k] || t - b->at_us[k] > 10000000) continue;
179 for (size_t i = 0; i < b->rep[k].nprogs; i++) {
180 perf_mem_t *pm = perf_mem_get(r, b->rep[k].progs[i].name, k);
181 if (!pm) {
182 r->perf_mem = mt_realloc(r->perf_mem, (r->nperf_mem + 1) * sizeof *r->perf_mem);
183 pm = &r->perf_mem[r->nperf_mem++];
184 pm->name = mt_strdup(b->rep[k].progs[i].name);
185 pm->k = k;
186 }
187 pm->perf = b->rep[k].progs[i].perf;
188 pm->at = now;
189 }
190 }
191 for (size_t i = r->nperf_mem; i-- > 0;)
192 if (now - r->perf_mem[i].at >= PERF_KEEP_MS) {
193 free(r->perf_mem[i].name);
194 r->perf_mem[i] = r->perf_mem[--r->nperf_mem];
195 }
196 msm_perfs m = {0};
197 for (size_t i = 0; i < r->nperf_mem; i++) {
198 msm_perf p = r->perf_mem[i].perf;
199 uint32_t k = r->perf_mem[i].k;
200 /* A program not reported by a member that reports (it left there): not running there now. */
201 if (b->has[k] && t - b->at_us[k] < 10000000) {
202 bool listed = false;
203 for (size_t j = 0; j < b->rep[k].nprogs; j++) listed |= !strcmp(b->rep[k].progs[j].name, r->perf_mem[i].name);
204 if (!listed) p.run = false;
205 }
206 msm_perfs_set(&m, r->perf_mem[i].name, k, &p);
207 }
208 pthread_mutex_unlock(&b->mu);
209 return m;
210}
211
212mfd_doc sched_pass(mpr_relay *r, const mfd_doc *doc, uint64_t now, mt_strs *changes) {
213 size_t n;
214 msm_member *ms = sched_members(r, now, &n);
215 mst_owner(r->tracker, ms, n, now, &r->w);
216 msm_perfs perf = sched_perf_map(r, now);
217 msp_input in = {doc, ms, n, &perf, &r->w};
218 msp_explains_free(&r->explains);
219 mfd_doc d = msp_place(&in, changes, &r->explains);
220 msm_perfs_free(&perf);
221 sched_members_free(ms, n);
222 return d;
223}
224
225mfd_doc sched_edit(mpr_relay *r, const mfd_doc *base, mfd_doc doc, uint64_t now, mt_strs *changes) {
226 mfd_doc d = sched_pass(r, &doc, now, changes);
227 mfd_doc_free(&doc);
228 size_t n;
229 msm_member *ms = sched_members(r, now, &n);
230 msm_perfs perf = sched_perf_map(r, now);
231 for (size_t i = 0; i < d.ngroups; i++) {
232 mfd_group *g = &d.groups[i];
233 bool fresh = true;
234 for (size_t j = 0; j < base->ngroups; j++)
235 if (!strcmp(base->groups[j].name, g->name) && base->groups[j].epoch == g->epoch) fresh = false;
236 if (!fresh) continue;
237 const mfd_prog *p = mfd_prog_get(&d, g->program);
238 if (!p || !(p->needs.group_size > 0 || p->needs.ngroup_members)) continue;
239 msp_input in = {&d, ms, n, &perf, &r->w};
240 ml_list members;
241 char *why;
242 if (msp_group_choice(&in, g, &members, &why)) {
243 if (!ml_eq(&members, &g->members)) {
244 char *l = ml_text(&members);
245 mt_push_own(changes, mt_fmt("group %s epoch %u members %s (%s)", g->name, g->epoch, l, why));
246 free(l);
247 ml_free(&g->members);
248 g->members = members;
249 } else ml_free(&members);
250 free(why);
251 }
252 }
253 msm_perfs_free(&perf);
254 sched_members_free(ms, n);
255 return d;
256}
257
258void sched_log_decision(mpr_relay *r, const char *what) {
259 uint64_t t = r->h.uptime_ms(r->h.ctx);
260 if (r->sched_log.n >= 16) mt_strs_remove(&r->sched_log, 0);
261 mt_push_own(&r->sched_log, mt_fmt("%.1f s: %s", (double)t / 1e3, what));
262}
263
264static void start_migration(mpr_relay *r, const mfd_doc *doc, mst_move mv, uint64_t now) {
265 uint64_t since = mfc_next_version(r->core);
266 mfd_doc d = mst_start_move(doc, &mv, since);
267 char *progs = mt_join(&mv.programs, ",");
268 char *what = mt_fmt("sched: %s %s from system %u to %u%s: %s", mv.state ? "migrate" : "restart", progs, mv.from, mv.to, mv.forced ? " (pinned)" : "", mv.why);
269 free(progs);
270 relay_log(r, false, mt_strdup(what));
271 sched_log_decision(r, what);
272 mst_moved(r->tracker, &mv, now);
273 uint64_t v;
274 if (relay_propose(r, d, what, (replyto){0}, &v)) {
275 if (r->has_mig) mst_move_free(&r->mig.mv);
276 r->has_mig = true;
277 r->mig = (mig_t){mv, v, mpr_mono_ms()};
278 } else mst_move_free(&mv);
279}
280
281/* The migration in flight: the target's outcome (done: release the source), failed or lost (put it
282 * back), the source lost (the target starts fresh). */
283static void migration_check(mpr_relay *r, const mfd_doc *doc, uint64_t now) {
284 if (!r->has_mig) return;
285 mst_move *mv = &r->mig.mv;
286 uint64_t v = r->mig.version;
287 /* Not committed (the proposal failed): nothing moved. */
288 if (doc->version < v) {
289 if (!r->has_pending) {
290 char *p = mt_join(&mv->programs, ",");
291 relay_log(r, true, mt_fmt("sched: the migration of %s was not committed; dropped", p));
292 free(p);
293 mst_move_free(mv);
294 r->has_mig = false;
295 }
296 return;
297 }
298 bool any_from = false;
299 for (size_t i = 0; i < mv->programs.n; i++) {
300 const mfd_prog *p = mfd_prog_get(doc, mv->programs.v[i]);
301 any_from |= p && p->has_from;
302 }
303 if (!any_from && mv->state) {
304 mst_move_free(mv);
305 r->has_mig = false;
306 return;
307 }
308 uint32_t alive = mfc_alive(r->core, now);
309 /* The target's outcome for each moved program (its reconcile status as a whole is not the move's). */
310 mst_outcome *outs;
311 size_t nouts;
312 char *reason;
313 mpr_recon rs = {0};
314 if (mv->to == r->me) {
315 r->h.recon(r->h.ctx, &rs);
316 outs = rs.moves, nouts = rs.nmoves, reason = rs.reason ? rs.reason : "";
317 } else outs = r->st[mv->to].moves, nouts = r->st[mv->to].nmoves, reason = r->st[mv->to].reason ? r->st[mv->to].reason : "";
318 bool ok;
319 char *text;
320 double gap;
321 bool decided = mst_mig_verdict(mv, v, alive >> mv->to & 1, outs, nouts, reason, mpr_mono_ms() - r->mig.t0_ms, MIG_WAIT_MS, &ok, &text, &gap);
322 mpr_recon_free(&rs);
323 if (!decided) return;
324 char *names = mt_join(&mv->programs, ",");
325 bool source_up = alive >> mv->from & 1;
326 double secs = (double)(mpr_mono_ms() - r->mig.t0_ms) / 1e3;
327 mfd_doc d;
328 char *what;
329 if (ok) {
330 r->moves_done++;
331 if (gap >= 0) r->last_gap_us = (uint32_t)(gap * 1e3);
332 char *g = gap >= 0 ? mt_fmt("%.1f ms", gap) : mt_strdup("-");
333 char *why = mt_fmt("moved from system %u (%s; gap %s)", mv->from, mv->why, g);
334 d = mst_end_move(doc, mv, true, why);
335 char *gs = gap >= 0 ? mt_fmt(" (gap %.1f ms)", gap) : mt_strdup("");
336 what = mt_fmt("sched: %s moved from system %u to %u in %.2f s%s: the source releases it", names, mv->from, mv->to, secs, gs);
337 free(g), free(why), free(gs);
338 } else if (source_up) {
339 relay_log(r, true, mt_fmt("sched: the migration of %s from system %u to %u failed (%s); putting it back on system %u", names, mv->from, mv->to, text, mv->from));
340 char *why = mt_fmt("a move to system %u failed (%s): back", mv->to, text);
341 d = mst_end_move(doc, mv, false, why);
342 what = mt_fmt("sched: %s back on system %u (%s)", names, mv->from, text);
343 free(why);
344 } else {
345 relay_log(r, true, mt_fmt("sched: the source of %s, system %u, was lost during its migration (%s): system %u starts it fresh (its view is lost)", names,
346 mv->from, text, mv->to));
347 char *why = mt_fmt("the source, system %u, was lost during the move: started fresh (OP_INIT)", mv->from);
348 d = mst_end_move(doc, mv, true, why);
349 what = mt_fmt("sched: %s on system %u without its view (the source was lost)", names, mv->to);
350 free(why);
351 }
352 free(text), free(names);
353 relay_log(r, false, mt_strdup(what));
354 sched_log_decision(r, what);
355 mst_move_free(mv);
356 r->has_mig = false;
357 relay_propose(r, d, what, (replyto){0}, NULL);
358}
359
360void sched_schedule(mpr_relay *r, uint64_t now) {
361 if (now - r->last_sched < PLACE_MS) return;
362 r->last_sched = now;
363 mfd_doc doc = mfd_doc_clone(mfc_committed(r->core));
364 if (doc.version == 0) goto done;
365 /* A migration the document has in flight that this leader does not track: taken over. */
366 if (!r->has_mig) {
367 mst_resumed res = mst_resume_moves(&doc);
368 if (res.n) {
369 char *p = mt_join(&res.moves[0].programs, ",");
370 char *what = mt_fmt("sched: takes over the migration of %s from system %u to %u (version %llu)", p, res.moves[0].from, res.moves[0].to,
371 (unsigned long long)res.since[0]);
372 relay_log(r, false, mt_strdup(what));
373 sched_log_decision(r, what);
374 free(p), free(what);
375 r->has_mig = true;
376 r->mig = (mig_t){mst_move_clone(&res.moves[0]), res.since[0], mpr_mono_ms()};
377 }
378 mst_resumed_free(&res);
379 }
380 /* A migration in flight: done, failed, or waiting. */
381 if (r->has_mig) {
382 migration_check(r, &doc, now);
383 if (r->has_pending) goto done;
384 }
385 /* Placement: new programs, members gone or no longer fitting (not while paused). */
386 mt_strs ch;
387 mfd_doc d = sched_pass(r, &doc, now, &ch);
388 if (ch.n && !doc.sched_paused) {
389 char *j = mt_join(&ch, "; "), *what = mt_fmt("sched: %s", j);
390 free(j);
391 sched_log_decision(r, what);
392 mt_strs_free(&ch);
393 relay_propose(r, d, what, (replyto){0}, NULL);
394 goto done;
395 }
396 mt_strs_free(&ch);
397 mfd_doc_free(&d);
398 if (now - r->last_rebalance < r->w.tick_ms) goto done;
399 r->last_rebalance = now;
400 size_t n;
401 msm_member *ms = sched_members(r, now, &n);
402 mst_owner(r->tracker, ms, n, now, &r->w);
403 msm_perfs perf = sched_perf_map(r, now);
404 msp_input in = {&doc, ms, n, &perf, &r->w};
405 mst_move mv;
406 if (mst_step(r->tracker, &in, now, r->has_mig, &mv)) {
407 start_migration(r, &doc, mv, now);
408 goto free_model;
409 }
410 /* The lockstep group: a re-form proposed (committed with autoreform). */
411 char *name, *why;
412 ml_list members;
413 if (mst_group_step(r->tracker, &in, now, &name, &members, &why)) {
414 char *l = ml_text(&members), *line = mt_fmt("sched: proposes re-forming group %s over %s: %s", name, l, why);
415 free(l);
416 if (doc.autoreform && !doc.sched_paused) {
417 mfd_doc nd = mfd_doc_clone(&doc);
418 mfd_group *g = mfd_group_named(&nd, name);
419 if (g) mfg_new_epoch(g, ml_clone(&members));
420 char *logged = mt_fmt("%s (autoreform: committing)", line);
421 sched_log_decision(r, logged);
422 free(logged);
423 free(r->group_prop);
424 r->group_prop = NULL;
425 relay_propose(r, nd, line, (replyto){0}, NULL);
426 } else if (!mt_eq(r->group_prop, line)) {
427 /* Logged when the proposed set changes, not when only its scores move. */
428 const char *a = r->group_prop ? strstr(r->group_prop, ": ") : NULL, *b = strstr(line, ": ");
429 const char *ae = a ? strstr(a + 2, ": ") : NULL, *be = b ? strstr(b + 2, ": ") : NULL;
430 size_t al = a ? (ae ? (size_t)(ae - a) : strlen(a)) : 0, bl = b ? (be ? (size_t)(be - b) : strlen(b)) : 0;
431 if (!a || !b || al != bl || strncmp(a, b, al) != 0) {
432 relay_log(r, false, mt_fmt("%s (`group %s reform` or `sched autoreform on`)", line, name));
433 sched_log_decision(r, line);
434 }
435 free(r->group_prop);
436 r->group_prop = line;
437 line = NULL;
438 }
439 free(line), free(name), free(why);
440 ml_free(&members);
441 } else {
442 free(r->group_prop);
443 r->group_prop = NULL;
444 }
445free_model:
446 msm_perfs_free(&perf);
447 sched_members_free(ms, n);
448done:
449 mfd_doc_free(&doc);
450}
451
452/* ---- the view transfer */
453
454typedef struct {
455 mpr_relay *r;
456 uint32_t k, id;
457 char *name;
458} view_job;
459
460static void send_view_frame(mpr_relay *r, uint32_t k, uint32_t id, uint32_t ok, uint32_t abi, uint32_t total, uint32_t off, const uint8_t *b, size_t n, bool wait) {
461 mpw_bytes f = {0};
462 uint8_t t = MESH_F_VIEW;
463 mpw_put(&f, &t, 1);
464 mpw_put32(&f, id);
465 mpw_put32(&f, ok);
466 mpw_put32(&f, abi);
467 mpw_put32(&f, total);
468 mpw_put32(&f, off);
469 mpw_put32(&f, (uint32_t)n);
470 mpw_put(&f, b, n);
471 /* A full link queue: wait for room (a view is a few frames). */
472 uint64_t t0 = mpr_mono_ms();
473 while (!mpr_send(r, k, f.b, f.n, false)) {
474 if (!wait || mpr_mono_ms() - t0 > 10000) {
475 if (wait) relay_log(r, true, mt_fmt("sched: a view for system %u: the link's queue stayed full", k));
476 break;
477 }
478 struct timespec ts = {0, 2000000};
479 nanosleep(&ts, NULL);
480 }
481 mpw_bytes_free(&f);
482}
483
484static void view_fail(mpr_relay *r, uint32_t k, uint32_t id, const char *e) {
485 size_t n = strlen(e) > MESH_F_REASON_MAX ? MESH_F_REASON_MAX : strlen(e);
486 send_view_frame(r, k, id, 0, 0, 0, 0, (const uint8_t *)e, n, false);
487}
488
489static void *view_out(void *arg) {
490 view_job *j = arg;
491 mpr_relay *r = j->r;
492 uint64_t t0 = mpr_mono_us();
493 uint32_t abi = 0;
494 mpw_bytes v = {0};
495 char *err = NULL;
496 if (r->h.stop_and_view(r->h.ctx, j->name, &abi, &v, &err)) {
497 double stop_ms = (double)(mpr_mono_us() - t0) / 1e3;
498 size_t total = v.n, off = 0;
499 do {
500 size_t n = total - off < MESH_F_VIEW_CHUNK ? total - off : MESH_F_VIEW_CHUNK;
501 send_view_frame(r, j->k, j->id, 1, abi, (uint32_t)total, (uint32_t)off, v.b + off, n, true);
502 off += n;
503 } while (off < total);
504 relay_log(r, false, mt_fmt("sched: %s: stopped in %.1f ms, its view (%zu bytes) sent to system %u", j->name, stop_ms, total, j->k));
505 } else {
506 relay_log(r, true, mt_fmt("sched: %s's view for system %u: %s", j->name, j->k, err));
507 char *e = mt_fmt("system %u: %s: %s", r->me, j->name, err);
508 view_fail(r, j->k, j->id, e);
509 free(e), free(err);
510 }
511 mpw_bytes_free(&v);
512 free(j->name);
513 free(j);
514 return NULL;
515}
516
517/* The target asks for a view: authorised only by a document naming the move (committed or accepted). */
518void sched_serve_view(mpr_relay *r, uint32_t k, uint32_t id, const char *name, uint64_t hash) {
519 bool ok = false;
520 const mfd_doc *docs[2] = {mfc_committed(r->core), mfc_accepted(r->core)};
521 for (int i = 0; i < 2; i++) {
522 const mfd_prog *p = mfd_prog_get(docs[i], name);
523 if (p && p->hash == hash && p->has_from && p->from == r->me && p->at.some && ml_has(&p->at.l, k)) ok = true;
524 }
525 if (!ok) {
526 char *e = mt_fmt("system %u: no desired state moves %s %016llx from here to system %u (not yet committed?)", r->me, name, (unsigned long long)hash, k);
527 view_fail(r, k, id, e);
528 free(e);
529 return;
530 }
531 relay_log(r, false, mt_fmt("sched: system %u takes %s's view: stopping it here", k, name));
532 view_job *j = mt_alloc(sizeof *j);
533 *j = (view_job){r, k, id, mt_strdup(name)};
534 pthread_t t;
535 pthread_create(&t, NULL, view_out, j);
536 pthread_detach(t);
537}
538
539void sched_fetch_view(mpr_relay *r, const char *name, uint64_t hash, uint32_t from, waiter *w) {
540 if (!relay_linked(r, from)) {
541 waiter_done(w, false, mt_fmt("system %u is not linked", from));
542 return;
543 }
544 size_t slot = 0;
545 while (slot < 32 && r->views[slot].used) slot++;
546 if (slot == 32) {
547 waiter_done(w, false, mt_strdup("too many views in flight"));
548 return;
549 }
550 uint32_t id = r->next_id++;
551 mpw_bytes f = {0};
552 uint8_t t = MESH_F_VIEW_REQ, nm[MESH_NAME_BYTES], hb[8];
553 mpw_put(&f, &t, 1);
554 mpw_put32(&f, id);
555 mpw_name16(name, nm);
556 mpw_put(&f, nm, sizeof nm);
557 memcpy(hb, &hash, 8);
558 mpw_put(&f, hb, 8);
559 relay_send(r, from, &f);
560 mpw_bytes_free(&f);
561 /* At most the program's view as the document gives it (else the whole view heap). */
562 const mfd_prog *p = mfd_prog_get(mfc_committed(r->core), name);
563 size_t vb = p ? p->needs.view_bytes : 0;
564 size_t limit = vb > 0 ? vb : (size_t)MESH_VIEW_PAGES * MESH_VIEW_PAGE;
565 r->views[slot] = (view_wait){true, id, w, mpv_wait_new(from, mpr_mono_ms() + VIEW_WAIT_MS, limit)};
566}
567
568void sched_view_chunk(mpr_relay *r, uint32_t id, const uint8_t *b, size_t n) {
569 for (size_t i = 0; i < 32; i++) {
570 view_wait *vw = &r->views[i];
571 if (!vw->used || vw->id != id) continue;
572 uint32_t ok = mpw_u32_at(b, 4), abi = mpw_u32_at(b, 8), total = mpw_u32_at(b, 12), off = mpw_u32_at(b, 16), len = mpw_u32_at(b, 20);
573 const uint8_t *body = b + MESH_VIEW_HEAD;
574 if (MESH_VIEW_HEAD + (size_t)len > n) len = (uint32_t)(n - MESH_VIEW_HEAD);
575 if (!ok) {
576 vw->used = false;
577 mpv_wait_free(&vw->v);
578 waiter_done(vw->w, false, mt_strndup((const char *)body, len));
579 return;
580 }
581 char *err = NULL;
582 int t = mpv_take(&vw->v, total, off, body, len, &err);
583 if (t == 0) return;
584 vw->used = false;
585 if (t == 1) {
586 vw->w->view_abi = abi;
587 mpw_put(&vw->w->view, vw->v.buf, vw->v.total);
588 mpv_wait_free(&vw->v);
589 waiter_done(vw->w, true, NULL);
590 } else {
591 mpv_wait_free(&vw->v);
592 waiter_done(vw->w, false, err);
593 }
594 return;
595 }
596}
597
598void sched_view_timeouts(mpr_relay *r, bool has_lost, uint32_t lost) {
599 uint64_t t = mpr_mono_ms();
600 for (size_t i = 0; i < 32; i++) {
601 view_wait *vw = &r->views[i];
602 if (!vw->used) continue;
603 bool gone = has_lost && vw->v.from == lost;
604 if (!gone && t < vw->v.deadline) continue;
605 vw->used = false;
606 uint32_t from = vw->v.from;
607 mpv_wait_free(&vw->v);
608 waiter_done(vw->w, false, gone ? mt_fmt("system %u was lost before it sent the view", from) : mt_fmt("system %u did not send the view in 30s", from));
609 }
610}