==> src/fleet_edit.c <==1/* fleet.rs: Req (parse), Image, EditCtx, edit, revert. */
2#include "mesh/fleet_edit.h"
3#include "mesh/abi/base.h"
4#include "mesh/abi/placement.h"
5#include "mesh/fleet_groups.h"
6#include "mesh/parse.h"
7#include "mesh/place_mode.h"
8#include "mesh/rust_fmt.h"
9
10#include <stdlib.h>
11#include <string.h>
12
13void mfe_req_free(mfe_req *r) {
14 free(r->name);
15 free(r->dir);
16 mt_strs_free(&r->words);
17 ml_opt_free(&r->pin);
18 *r = (mfe_req){0};
19}
20
21void mfe_image_free(mfe_image *im) {
22 free(im->dir);
23 free(im->name);
24 mt_strs_free(&im->backends);
25 mesh_needs_free(&im->needs);
26 *im = (mfe_image){0};
27}
28
29/* "0x.." in hex, else decimal (fleet.rs int). */
30static bool int_word(const char *s, uint32_t *out) { return mt_starts(s, "0x") ? mp_hex_u32(s + 2, out) : mp_u32(s, out); }
31
32static bool w_is(const mt_strs *w, size_t n, const char *const *lit) {
33 if (w->n != n) return false;
34 for (size_t i = 0; i < n; i++)
35 if (lit[i] && strcmp(w->v[i], lit[i])) return false;
36 return true;
37}
38
39#define IS(...) w_is(&w, sizeof((const char *const[]){__VA_ARGS__}) / sizeof(const char *), (const char *const[]){__VA_ARGS__})
40
41bool mfe_parse(const char *line, mfe_req *out, char **err) {
42 mt_strs w = mt_words(line);
43 mfe_req r = {0};
44 bool ok = true;
45 char **v = w.v;
46 if (IS("load", NULL)) r.kind = MFE_LOAD, r.name = mt_strdup(v[1]);
47 else if (IS("upgrade", NULL)) r.kind = MFE_UPGRADE, r.name = mt_strdup(v[1]);
48 else if (IS("upgrade", NULL, NULL)) r.kind = MFE_UPGRADE, r.name = mt_strdup(v[1]), r.dir = mt_strdup(v[2]);
49 else if (IS("rollback", NULL)) r.kind = MFE_ROLLBACK, r.name = mt_strdup(v[1]);
50 else if (IS("unload", NULL) || IS("unload", NULL, "--now")) r.kind = MFE_UNLOAD, r.name = mt_strdup(v[1]);
51 else if (IS("start", NULL)) r.kind = MFE_RUN, r.name = mt_strdup(v[1]), r.on = true;
52 else if (IS("stop", NULL)) r.kind = MFE_RUN, r.name = mt_strdup(v[1]), r.on = false;
53 else if (IS("reset", NULL)) r.kind = MFE_RESET, r.name = mt_strdup(v[1]);
54 else if (IS("task", NULL, NULL, NULL)) {
55 r.kind = MFE_TASK;
56 if (!int_word(v[2], &r.op) || !int_word(v[3], &r.arg)) {
57 *err = mt_strdup("usage: task <name> <op> <arg>");
58 ok = false;
59 } else r.name = mt_strdup(v[1]);
60 } else if (w.n >= 3 && strcmp(v[0], "place") == 0) {
61 r.kind = MFE_PLACE;
62 r.name = mt_strdup(v[1]);
63 for (size_t i = 2; i < w.n; i++) mt_push(&r.words, v[i]);
64 } else if (IS("group", NULL, "reform")) r.kind = MFE_REFORM, r.name = mt_strdup(v[1]);
65 else if (IS("group", NULL, "on_loss", "pause")) r.kind = MFE_ON_LOSS, r.name = mt_strdup(v[1]), r.on_loss = MFD_LOSS_PAUSE;
66 else if (IS("group", NULL, "on_loss", "reform")) r.kind = MFE_ON_LOSS, r.name = mt_strdup(v[1]), r.on_loss = MFD_LOSS_REFORM;
67 else if (IS("group", NULL, "on_loss", "reform", NULL)) {
68 r.kind = MFE_ON_LOSS, r.on_loss = MFD_LOSS_REFORM;
69 if (!mp_u32(v[4], &r.ms)) {
70 char *q = rf_debug_str(v[4]);
71 *err = mt_fmt("%s: milliseconds", q);
72 free(q);
73 ok = false;
74 } else r.name = mt_strdup(v[1]);
75 } else if (IS("sched", "pin", NULL, NULL)) {
76 r.kind = MFE_PIN;
77 char *e = NULL;
78 r.pin.some = true;
79 if (!ml_parse(v[3], &r.pin.l, &e)) {
80 *err = mt_fmt("usage: sched pin <name> <member[,member..]>: %s", e);
81 free(e);
82 ok = false;
83 } else r.name = mt_strdup(v[2]);
84 } else if (IS("sched", "unpin", NULL)) r.kind = MFE_PIN, r.name = mt_strdup(v[2]);
85 else if (IS("sched", "pause")) r.kind = MFE_SCHED_PAUSE, r.on = true;
86 else if (IS("sched", "resume")) r.kind = MFE_SCHED_PAUSE, r.on = false;
87 else if (IS("sched", "autoreform", "on")) r.kind = MFE_AUTOREFORM, r.on = true;
88 else if (IS("sched", "autoreform", "off")) r.kind = MFE_AUTOREFORM, r.on = false;
89 else if (IS("sched", "cordon", NULL) || IS("sched", "uncordon", NULL)) {
90 r.kind = MFE_CORDON;
91 r.on = strcmp(v[1], "cordon") == 0;
92 if (!mp_u32(v[2], &r.k) || r.k >= MESH_MAX_SYSTEMS) {
93 char *q = rf_debug_str(v[2]);
94 *err = mt_fmt("%s: a member", q);
95 free(q);
96 ok = false;
97 }
98 } else {
99 char *q = rf_debug_str(line);
100 *err = mt_fmt("not a change: %s", q);
101 free(q);
102 ok = false;
103 }
104 mt_strs_free(&w);
105 if (!ok) {
106 mfe_req_free(&r);
107 return false;
108 }
109 *out = r;
110 return true;
111}
112
113static char *missing(const char *n) {
114 char *q = rf_debug_str(n);
115 char *e = mt_fmt("no program %s in the desired state (`fleet` lists it)", q);
116 free(q);
117 return e;
118}
119
120static char *no_group(const char *n) {
121 char *q = rf_debug_str(n);
122 char *e = mt_fmt("no group %s (`groups` lists them)", q);
123 free(q);
124 return e;
125}
126
127static mfd_group *running_group(mfd_doc *d, const char *program) {
128 for (size_t i = 0; i < d->ngroups; i++)
129 if (strcmp(d->groups[i].program, program) == 0 && d->groups[i].state == MFD_RUNNING) return &d->groups[i];
130 return NULL;
131}
132
133static bool load(mfd_doc *d, const mfe_req *req, const mfe_ctx *cx, char **what, char **err) {
134 mfe_image im;
135 if (!cx->catalog(cx->ctx, req->name, &im, err)) return false;
136 bool ok = false;
137 if (mfd_prog_get(d, im.name)) {
138 *err = mt_fmt("name exists: %s is in the desired state", im.name);
139 goto done;
140 }
141 if (im.lockstep) {
142 for (size_t i = 0; i < d->nprogs; i++)
143 if (d->prog[i].lockstep) {
144 *err = mt_fmt("manifest: %s is the lockstep program already; only one may be loaded", d->pname[i]);
145 goto done;
146 }
147 ml_list members = cx->eligible_up(cx->ctx, &im.backends);
148 if (!members.n) {
149 char *b = mt_join(&im.backends, ", ");
150 *err = mt_fmt("unsupported: no up member runs %s (backends %s)", im.name, b);
151 free(b);
152 ml_free(&members);
153 goto done;
154 }
155 const char *gname = im.needs.group ? im.needs.group : im.name;
156 uint32_t epoch = 0;
157 for (size_t i = 0; i < d->ngroups; i++)
158 if (strcmp(d->groups[i].name, gname) == 0 && d->groups[i].epoch > epoch) epoch = d->groups[i].epoch;
159 for (size_t i = d->ngroups; i-- > 0;)
160 if (strcmp(d->groups[i].name, gname) == 0) mfd_group_remove_at(d, i);
161 mfd_group_push(d, (mfd_group){.name = mt_strdup(gname), .program = mt_strdup(im.name), .epoch = epoch + 1, .members = members, .state = MFD_RUNNING,
162 .on_loss = MFD_LOSS_PAUSE});
163 }
164 mfd_prog p = mfd_prog_new(im.dir, im.hash, im.autostart, &im.needs);
165 p.lockstep = im.lockstep;
166 p.backends = mt_strs_clone(&im.backends);
167 mt_set(&p.why, "new: the scheduler places it");
168 mfd_prog_put(d, im.name, p);
169 *what = mt_fmt("load %s from %s (%016llx)", im.name, im.dir, (unsigned long long)im.hash);
170 ok = true;
171done:
172 mfe_image_free(&im);
173 return ok;
174}
175
176static bool upgrade(mfd_doc *d, const mfe_req *req, const mfe_ctx *cx, char **what, char **err) {
177 const char *n = req->name;
178 mfd_prog *cur = mfd_prog_get(d, n);
179 if (!cur) return *err = missing(n), false;
180 mfd_prog p = mfd_prog_clone(cur);
181 mfe_image im;
182 bool ok = false;
183 if (!cx->catalog(cx->ctx, req->dir ? req->dir : p.dir, &im, err)) {
184 mfd_prog_free(&p);
185 return false;
186 }
187 if (strcmp(im.name, n) != 0) {
188 char *a = rf_debug_str(im.name), *b = rf_debug_str(n);
189 *err = mt_fmt("manifest: %s names %s, not %s", im.dir, a, b);
190 free(a), free(b);
191 goto done;
192 }
193 if (im.lockstep != p.lockstep) {
194 *err = mt_fmt("manifest: %s: an upgrade cannot change `peer` (unload and load it)", im.dir);
195 goto done;
196 }
197 if (im.hash == p.hash && strcmp(im.dir, p.dir) == 0) {
198 *what = mt_fmt("upgrade %s: already at %016llx (%s): nothing changes", n, (unsigned long long)p.hash, p.dir);
199 ok = true;
200 goto done;
201 }
202 /* A running lockstep group upgrades in one cycle on every member: each must be able to run the new image. */
203 mfd_group *g = running_group(d, n);
204 if (g) {
205 ml_list can = cx->eligible_up(cx->ctx, &im.backends), cannot = {0};
206 for (size_t i = 0; i < g->members.n; i++)
207 if (!ml_has(&can, g->members.v[i])) ml_push(&cannot, g->members.v[i]);
208 ml_free(&can);
209 if (cannot.n) {
210 char *l = ml_text(&cannot), *b = mt_join(&im.backends, ",");
211 *err = mt_fmt("unsupported: group %s member(s) %s cannot run %s (backends %s): every member of a lockstep group runs the same image (`group %s reform` "
212 "without them first)",
213 g->name, l, im.dir, b, g->name);
214 free(l), free(b);
215 ml_free(&cannot);
216 goto done;
217 }
218 }
219 mfd_prog *e = mfd_prog_get(d, n);
220 e->has_prev = true;
221 mt_set(&e->prev_dir, p.dir);
222 e->prev_hash = p.hash;
223 mt_set(&e->dir, im.dir);
224 e->hash = im.hash;
225 mt_strs_free(&e->backends);
226 e->backends = mt_strs_clone(&im.backends);
227 mesh_needs_free(&e->needs);
228 e->needs = msn_clone(&im.needs);
229 if (g) {
230 g->cycle += 1;
231 g->op = (mfd_op){MFD_OP_UPGRADE, 0, 0};
232 }
233 *what = mt_fmt("upgrade %s %016llx -> %016llx (%s)", n, (unsigned long long)p.hash, (unsigned long long)im.hash, im.dir);
234 ok = true;
235done:
236 mfe_image_free(&im);
237 mfd_prog_free(&p);
238 return ok;
239}
240
241static bool place(mfd_doc *d, const mfe_req *req, const mfe_ctx *cx, char **what, char **err) {
242 const char *n = req->name;
243 mfd_prog *p = mfd_prog_get(d, n);
244 if (!p) return *err = missing(n), false;
245 mfe_image im;
246 if (!cx->catalog(cx->ctx, p->dir, &im, err)) return false;
247 bool ok = false;
248 char *e = NULL;
249 if (!im.placeable) {
250 *err = mt_fmt("%s declares no `place` in its manifest", n);
251 goto done;
252 }
253 mpm_target t;
254 if (!mpm_parse(req->words.v, req->words.n, cx->n, im.place_fanout, &t, err)) goto done;
255 mpm_mode m = t.is_auto ? mpm_split(mpm_even((uint32_t[]){0}, 1, im.place_fanout)) : t.mode;
256 if (!mpm_allowed(&m, im.place_cap, &e)) {
257 *err = mt_fmt("%s: %s", n, e);
258 free(e);
259 goto done;
260 }
261 if (req->words.n == 1 && strcmp(req->words.v[0], "local") == 0) mfd_place_remove(d, n);
262 else mfd_place_put(d, n, mt_strs_clone(&req->words));
263 char *j = mt_join(&req->words, " ");
264 *what = mt_fmt("place %s %s", n, j);
265 free(j);
266 ok = true;
267done:
268 mfe_image_free(&im);
269 return ok;
270}
271
272bool mfe_edit(const mfd_doc *doc, const mfe_req *req, const mfe_ctx *cx, mfd_doc *out, char **what, char **err) {
273 mfd_doc d = mfd_doc_clone(doc);
274 bool ok = true;
275 const char *n = req->name;
276 switch (req->kind) {
277 case MFE_LOAD: ok = load(&d, req, cx, what, err); break;
278 case MFE_UPGRADE: ok = upgrade(&d, req, cx, what, err); break;
279 case MFE_ROLLBACK: {
280 mfd_prog *e = mfd_prog_get(&d, n);
281 if (!e) {
282 *err = missing(n);
283 ok = false;
284 break;
285 }
286 if (!e->has_prev) {
287 *err = mt_fmt("%s: no previous image to roll back to", n);
288 ok = false;
289 break;
290 }
291 char *pd = mt_strdup(e->prev_dir), *od = mt_strdup(e->dir);
292 uint64_t ph = e->prev_hash, oh = e->hash;
293 mt_set(&e->prev_dir, od);
294 e->prev_hash = oh;
295 mt_set(&e->dir, pd);
296 e->hash = ph;
297 free(pd), free(od);
298 mfd_group *g = running_group(&d, n);
299 if (g) {
300 g->cycle += 1;
301 g->op = (mfd_op){MFD_OP_ROLLBACK, 0, 0};
302 }
303 *what = mt_fmt("rollback %s %016llx -> %016llx", n, (unsigned long long)oh, (unsigned long long)ph);
304 break;
305 }
306 case MFE_UNLOAD:
307 if (!mfd_prog_remove(&d, n)) {
308 *err = missing(n);
309 ok = false;
310 break;
311 }
312 for (size_t i = d.ngroups; i-- > 0;)
313 if (strcmp(d.groups[i].program, n) == 0) mfd_group_remove_at(&d, i);
314 mfd_place_remove(&d, n);
315 *what = mt_fmt("unload %s", n);
316 break;
317 case MFE_RUN: {
318 mfd_prog *p = mfd_prog_get(&d, n);
319 if (!p) {
320 *err = missing(n);
321 ok = false;
322 break;
323 }
324 p->run = req->on;
325 *what = mt_fmt("%s %s", req->on ? "start" : "stop", n);
326 break;
327 }
328 case MFE_RESET: {
329 mfd_prog *p = mfd_prog_get(&d, n);
330 if (!p) {
331 *err = missing(n);
332 ok = false;
333 break;
334 }
335 if (p->lockstep) {
336 /* A lockstep reset is the group's next epoch over the same members (rounds restart at 1 on all). */
337 mfd_group *g = mfd_group_of(&d, n);
338 if (!g) {
339 *err = mt_fmt("%s: no group", n);
340 ok = false;
341 break;
342 }
343 mfd_gstate st = g->state;
344 mfg_new_epoch(g, ml_clone(&g->members));
345 g->state = st;
346 *what = mt_fmt("reset %s: group epoch %u", n, g->epoch);
347 } else {
348 p->reset += 1;
349 *what = mt_fmt("reset %s", n);
350 }
351 break;
352 }
353 case MFE_TASK: {
354 mfd_group *g = mfd_group_of(&d, n);
355 if (!g) {
356 *err = mt_fmt("%s is not a lockstep program: its tasks are not changes (they run where asked)", n);
357 ok = false;
358 break;
359 }
360 if (g->state != MFD_RUNNING) {
361 *err = mt_fmt("peer: group %s is paused (a member is missing): its tasks run on every member or none", g->name);
362 ok = false;
363 break;
364 }
365 g->cycle += 1;
366 g->op = (mfd_op){MFD_OP_TASK, req->op, req->arg};
367 *what = mt_fmt("task %s %u %u (group %s cycle %u)", n, req->op, req->arg, g->name, g->cycle);
368 break;
369 }
370 case MFE_PLACE: ok = place(&d, req, cx, what, err); break;
371 case MFE_REFORM: {
372 mfd_group *g = mfd_group_named(&d, n);
373 if (!g) {
374 *err = no_group(n);
375 ok = false;
376 break;
377 }
378 const mfd_prog *p = mfd_prog_get(doc, g->program);
379 if (!p) {
380 *err = missing(g->program);
381 ok = false;
382 break;
383 }
384 ml_list up = cx->eligible_up(cx->ctx, &p->backends);
385 mfd_group ng;
386 ok = mfg_reform(g, &up, &ng, err);
387 ml_free(&up);
388 if (!ok) break;
389 mfd_group_free(g);
390 *g = ng;
391 char *l = ml_text(&g->members);
392 *what = mt_fmt("group %s reformed: epoch %u members %s", n, g->epoch, l);
393 free(l);
394 break;
395 }
396 case MFE_PIN: {
397 mfd_prog *p = mfd_prog_get(&d, n);
398 if (!p) {
399 *err = missing(n);
400 ok = false;
401 break;
402 }
403 if (req->pin.some) {
404 if (!req->pin.l.n) {
405 *err = mt_strdup("sched pin: at least one member");
406 ok = false;
407 break;
408 }
409 for (size_t i = 0; i < req->pin.l.n && ok; i++)
410 if (req->pin.l.v[i] >= cx->n) {
411 *err = mt_fmt("sched pin: system %u: the fleet has 0..%u", req->pin.l.v[i], cx->n - 1);
412 ok = false;
413 }
414 if (!ok) break;
415 if (p->lockstep) {
416 *err = mt_fmt("%s is the lockstep program: its group's members are `group %s reform`'s, not a pin", n, n);
417 ok = false;
418 break;
419 }
420 const char *a = p->needs.colocate_with;
421 if (a && mfd_prog_get(doc, a)) {
422 *err = mt_fmt("%s is colocated with %s: pin %s (what is colocated with it moves with it)", n, a, a);
423 ok = false;
424 break;
425 }
426 }
427 ml_opt_free(&p->pin);
428 p->pin = ml_opt_clone(&req->pin);
429 if (req->pin.some) {
430 char *l = ml_text(&req->pin.l);
431 *what = mt_fmt("sched pin %s %s", n, l);
432 free(l);
433 } else *what = mt_fmt("sched unpin %s", n);
434 break;
435 }
436 case MFE_SCHED_PAUSE:
437 d.sched_paused = req->on;
438 *what = mt_fmt("sched %s", req->on ? "pause" : "resume");
439 break;
440 case MFE_AUTOREFORM:
441 d.autoreform = req->on;
442 *what = mt_fmt("sched autoreform %s", req->on ? "on" : "off");
443 break;
444 case MFE_CORDON:
445 if (req->k >= cx->n) {
446 *err = mt_fmt("sched cordon: system %u: the fleet has 0..%u", req->k, cx->n - 1);
447 ok = false;
448 break;
449 }
450 if (req->on) d.cordon |= 1u << req->k;
451 else d.cordon &= ~(1u << req->k);
452 *what = mt_fmt("sched %scordon %u", req->on ? "" : "un", req->k);
453 break;
454 case MFE_ON_LOSS: {
455 mfd_group *g = mfd_group_named(&d, n);
456 if (!g) {
457 *err = no_group(n);
458 ok = false;
459 break;
460 }
461 g->on_loss = req->on_loss;
462 g->auto_ms = req->ms;
463 char *after = req->ms > 0 ? mt_fmt(" after %u ms", req->ms) : mt_strdup("");
464 *what = mt_fmt("group %s on_loss %s%s", n, req->on_loss == MFD_LOSS_PAUSE ? "pause" : "reform", after);
465 free(after);
466 break;
467 }
468 }
469 if (!ok) {
470 mfd_doc_free(&d);
471 return false;
472 }
473 *out = d;
474 return true;
475}
476
477/* The union of two sorted name arrays, sorted (a BTreeSet's iteration). */
478static mt_strs union_sorted(char *const *a, size_t na, char *const *b, size_t nb) {
479 mt_strs u = {0};
480 size_t i = 0, j = 0;
481 while (i < na || j < nb) {
482 int c = i == na ? 1 : j == nb ? -1 : strcmp(a[i], b[j]);
483 if (c <= 0) mt_push(&u, a[i]);
484 else mt_push(&u, b[j]);
485 if (c <= 0) i++;
486 if (c >= 0) j++;
487 }
488 return u;
489}
490
491static int cmp_str(const void *a, const void *b) { return strcmp(*(char *const *)a, *(char *const *)b); }
492
493mfd_doc mfe_revert(const mfd_doc *cur, const mfd_doc *before, const mfd_doc *after) {
494 mfd_doc d = mfd_doc_clone(cur);
495 mt_strs keys = union_sorted(before->pname, before->nprogs, after->pname, after->nprogs);
496 for (size_t i = 0; i < keys.n; i++) {
497 const mfd_prog *b = mfd_prog_get(before, keys.v[i]), *a = mfd_prog_get(after, keys.v[i]);
498 if ((!b && !a) || (b && a && mfd_prog_eq(b, a))) continue;
499 if (b) mfd_prog_put(&d, keys.v[i], mfd_prog_clone(b));
500 else mfd_prog_remove(&d, keys.v[i]);
501 }
502 mt_strs_free(&keys);
503 mt_strs names = {0};
504 for (size_t i = 0; i < before->ngroups; i++)
505 if (!mt_strs_has(&names, before->groups[i].name)) mt_push(&names, before->groups[i].name);
506 for (size_t i = 0; i < after->ngroups; i++)
507 if (!mt_strs_has(&names, after->groups[i].name)) mt_push(&names, after->groups[i].name);
508 if (names.n > 1) qsort(names.v, names.n, sizeof *names.v, cmp_str);
509 for (size_t i = 0; i < names.n; i++) {
510 const mfd_group *b = mfd_group_named(before, names.v[i]), *a = mfd_group_named(after, names.v[i]);
511 if ((!b && !a) || (b && a && mfd_group_eq(b, a))) continue;
512 for (size_t j = d.ngroups; j-- > 0;)
513 if (strcmp(d.groups[j].name, names.v[i]) == 0) mfd_group_remove_at(&d, j);
514 if (b) mfd_group_push(&d, mfd_group_clone(b));
515 }
516 mt_strs_free(&names);
517 keys = union_sorted(before->place_name, before->nplaces, after->place_name, after->nplaces);
518 for (size_t i = 0; i < keys.n; i++) {
519 const mt_strs *b = mfd_place_get(before, keys.v[i]), *a = mfd_place_get(after, keys.v[i]);
520 if ((!b && !a) || (b && a && mt_strs_eq(b, a))) continue;
521 if (b) mfd_place_put(&d, keys.v[i], mt_strs_clone(b));
522 else mfd_place_remove(&d, keys.v[i]);
523 }
524 mt_strs_free(&keys);
525 if (before->sched_paused != after->sched_paused) d.sched_paused = before->sched_paused;
526 if (before->autoreform != after->autoreform) d.autoreform = before->autoreform;
527 if (before->cordon != after->cordon) d.cordon = before->cordon;
528 return d;
529}