==> src/reconcile.c <==1/* reconcile.rs: Rec (pass, act, migrate, group, place), cycle_ready, image_ready, run. */
2#include "mesh/reconcile.h"
3#include "mesh/abi/peer_sizes.h"
4#include "mesh/abi/reconcile.h"
5#include "mesh/strmap.h"
6#include "mesh/mt.h"
7
8#include <pthread.h>
9#include <stdio.h>
10#include <stdlib.h>
11#include <string.h>
12#include <time.h>
13
14#define RETRY_MS 30000
15
16static uint64_t mono_us(void) {
17 struct timespec ts;
18 clock_gettime(CLOCK_MONOTONIC, &ts);
19 return (uint64_t)ts.tv_sec * 1000000u + (uint64_t)ts.tv_nsec / 1000u;
20}
21
22static const char *op_word(mfd_op op) {
23 return op.kind == MFD_OP_NONE ? "none" : op.kind == MFD_OP_UPGRADE ? "upgrade" : op.kind == MFD_OP_ROLLBACK ? "rollback" : "task";
24}
25
26/* The program runs here at the group's image (the inventory after the cycle's actions). */
27static bool image_ready(const mfd_group *g, uint64_t hash, const mfl_actual *inv, size_t n, char **why) {
28 for (size_t i = 0; i < n; i++)
29 if (strcmp(inv[i].name, g->program) == 0) {
30 if (inv[i].hash == hash) return true;
31 *why = mt_fmt("group %s cycle %u (%s): %s runs %016llx here, the group %016llx; this member stays out of the cycle (paused)", g->name, g->cycle,
32 op_word(g->op), g->program, (unsigned long long)inv[i].hash, (unsigned long long)hash);
33 return false;
34 }
35 *why = mt_fmt("group %s cycle %u: %s is not loaded here; this member stays out of the cycle (paused)", g->name, g->cycle, g->program);
36 return false;
37}
38
39bool mrc_cycle_ready(const mfd_group *g, uint64_t hash, const mfl_actual *inv, size_t ninv, const bool *task_ok, const char *task_err, char **why) {
40 if (!image_ready(g, hash, inv, ninv, why)) return false;
41 if (g->op.kind == MFD_OP_TASK) {
42 if (task_ok && !*task_ok) {
43 *why = mt_fmt("group %s cycle %u: task %u %u failed here (%s); this member stays paused until the group's next epoch (`group %s reform` or `reset %s`)", g->name,
44 g->cycle, g->op.op, g->op.arg, task_err, g->name, g->program);
45 return false;
46 }
47 if (!task_ok) {
48 *why = mt_fmt("group %s cycle %u: task %u %u not run here", g->name, g->cycle, g->op.op, g->op.arg);
49 return false;
50 }
51 }
52 return true;
53}
54
55mt_strs mrc_stop_order(const mfd_doc *doc, const mt_strs *names) {
56 mt_strs left = mt_strs_clone(names), out = {0};
57 while (left.n) {
58 size_t pick = 0;
59 for (size_t i = 0; i < left.n; i++) {
60 bool fed = false;
61 for (size_t j = 0; j < left.n && !fed; j++) {
62 if (!strcmp(left.v[j], left.v[i])) continue;
63 const mfd_prog *o = mfd_prog_get(doc, left.v[j]);
64 if (o && msn_strs_has(o->needs.feeds, o->needs.nfeeds, left.v[i])) fed = true;
65 }
66 if (!fed) {
67 pick = i;
68 break;
69 }
70 }
71 mt_push(&out, left.v[pick]);
72 mt_strs_remove(&left, pick);
73 }
74 mt_strs_free(&left);
75 return out;
76}
77
78/* ---- the reconciler's memory */
79
80typedef struct {
81 char *name;
82 uint64_t hash;
83 char *why;
84 bool unsupported;
85} skip_t;
86
87struct mrc_rec {
88 mrc_ops ops;
89 ms_map resets; /* name -> the reset generation applied (decimal) */
90 skip_t *skip; /* programs not to try again at this hash, sorted by name */
91 size_t nskip;
92 bool has_group; /* the group epoch applied here: (group, epoch, cycle, members) */
93 char *g_name;
94 uint32_t g_epoch, g_cycle, g_mask;
95 bool has_cycle_failed; /* a cycle this member could not apply: (group, epoch, cycle, why) */
96 char *cf_group, *cf_why;
97 uint32_t cf_epoch, cf_cycle;
98 char **place_name; /* the placement requests applied (on the leader) */
99 mt_strs *place_words;
100 size_t nplaces;
101 mt_strs done, failed;
102 mst_outcome *moves; /* per program loaded or migrated here: the outcome, sorted by name */
103 size_t nmoves;
104 mfd_doc doc; /* the last document */
105 bool has_doc;
106};
107
108mrc_rec *mrc_new(const mrc_ops *ops) {
109 mrc_rec *r = mt_alloc(sizeof *r);
110 r->ops = *ops;
111 return r;
112}
113
114static void skip_clear_at(mrc_rec *r, size_t i) {
115 free(r->skip[i].name);
116 free(r->skip[i].why);
117 memmove(r->skip + i, r->skip + i + 1, (r->nskip - i - 1) * sizeof *r->skip);
118 r->nskip--;
119}
120
121static void places_clear(mrc_rec *r) {
122 for (size_t i = 0; i < r->nplaces; i++) {
123 free(r->place_name[i]);
124 mt_strs_free(&r->place_words[i]);
125 }
126 free(r->place_name);
127 free(r->place_words);
128 r->place_name = NULL;
129 r->place_words = NULL;
130 r->nplaces = 0;
131}
132
133static void moves_clear_at(mrc_rec *r, size_t i) {
134 free(r->moves[i].name);
135 memmove(r->moves + i, r->moves + i + 1, (r->nmoves - i - 1) * sizeof *r->moves);
136 r->nmoves--;
137}
138
139void mrc_free(mrc_rec *r) {
140 if (!r) return;
141 ms_free(&r->resets);
142 while (r->nskip) skip_clear_at(r, r->nskip - 1);
143 free(r->skip);
144 free(r->g_name);
145 free(r->cf_group);
146 free(r->cf_why);
147 places_clear(r);
148 mt_strs_free(&r->done);
149 mt_strs_free(&r->failed);
150 while (r->nmoves) moves_clear_at(r, r->nmoves - 1);
151 free(r->moves);
152 if (r->has_doc) mfd_doc_free(&r->doc);
153 free(r);
154}
155
156static skip_t *skip_get(mrc_rec *r, const char *name) {
157 for (size_t i = 0; i < r->nskip; i++)
158 if (!strcmp(r->skip[i].name, name)) return &r->skip[i];
159 return NULL;
160}
161
162static void skip_set(mrc_rec *r, const char *name, uint64_t hash, const char *why, bool unsupported) {
163 skip_t *s = skip_get(r, name);
164 if (s) {
165 s->hash = hash;
166 mt_set(&s->why, why);
167 s->unsupported = unsupported;
168 return;
169 }
170 size_t i = 0;
171 while (i < r->nskip && strcmp(r->skip[i].name, name) < 0) i++;
172 r->skip = mt_realloc(r->skip, (r->nskip + 1) * sizeof *r->skip);
173 memmove(r->skip + i + 1, r->skip + i, (r->nskip - i) * sizeof *r->skip);
174 r->skip[i] = (skip_t){mt_strdup(name), hash, mt_strdup(why), unsupported};
175 r->nskip++;
176}
177
178/* skip.retain(|_, s| s.2): keep only the unsupported. */
179static void skip_keep_unsupported(mrc_rec *r) {
180 for (size_t i = r->nskip; i-- > 0;)
181 if (!r->skip[i].unsupported) skip_clear_at(r, i);
182}
183
184static void reset_set(mrc_rec *r, const char *name, const mfd_doc *doc) {
185 const mfd_prog *p = mfd_prog_get(doc, name);
186 char v[16];
187 snprintf(v, sizeof v, "%u", p ? p->reset : 0);
188 ms_set(&r->resets, name, v, true);
189}
190
191/* A program's outcome here at doc (a move's target reports it). */
192static void moved(mrc_rec *r, const mfd_doc *doc, const char *name, bool ok, uint32_t gap_us) {
193 size_t i = 0;
194 while (i < r->nmoves && strcmp(r->moves[i].name, name) < 0) i++;
195 if (i < r->nmoves && !strcmp(r->moves[i].name, name)) {
196 r->moves[i] = (mst_outcome){r->moves[i].name, doc->version, ok, gap_us};
197 return;
198 }
199 r->moves = mt_realloc(r->moves, (r->nmoves + 1) * sizeof *r->moves);
200 memmove(r->moves + i + 1, r->moves + i, (r->nmoves - i) * sizeof *r->moves);
201 r->moves[i] = (mst_outcome){mt_strdup(name), doc->version, ok, gap_us};
202 r->nmoves++;
203}
204
205static void logf_(mrc_rec *r, bool error, char *line) {
206 if (r->ops.log) r->ops.log(r->ops.ctx, error, line);
207 free(line);
208}
209
210static void status(mrc_rec *r, uint64_t applied, uint32_t st, const char *reason) {
211 /* reason.chars().take(BEAT_REASON_MAX) */
212 char *cut = mt_strndup(reason, strlen(reason) > MESH_BEAT_REASON_MAX ? MESH_BEAT_REASON_MAX : strlen(reason));
213 r->ops.status(r->ops.ctx, applied, st, cut, r->moves, r->nmoves);
214 free(cut);
215}
216
217static mfl_actual *inventory(mrc_rec *r, size_t *n) {
218 mfl_actual *inv = r->ops.inventory(r->ops.ctx, n);
219 return inv;
220}
221
222static void inv_free(mfl_actual *inv, size_t n) {
223 for (size_t i = 0; i < n; i++) mfl_actual_free(&inv[i]);
224 free(inv);
225}
226
227static const char *verb(mfl_kind k) {
228 static const char *const w[] = {"unload", "load", "upgrade", "rollback", "run", "reset", "migrate"};
229 return w[k];
230}
231
232static void migrate(mrc_rec *r, const mfd_doc *doc, const mfl_action *migs, size_t n);
233
234/* One action; its result goes into done / failed / skip. */
235static bool act(mrc_rec *r, const mfd_doc *doc, const mfl_action *a) {
236 const char *n = a->name;
237 char *text = NULL, *err = NULL, *t = NULL;
238 bool ok = false;
239 void *src = NULL;
240 switch (a->kind) {
241 case MFL_UNLOAD:
242 ok = r->ops.unload(r->ops.ctx, n, &text, &err);
243 if (ok) t = mt_fmt("unload %s", n);
244 break;
245 case MFL_LOAD:
246 ok = r->ops.source(r->ops.ctx, n, a->dir, a->hash, &src, &err) && r->ops.load(r->ops.ctx, src, a->run, &text, &err);
247 if (ok) {
248 reset_set(r, n, doc);
249 t = mt_fmt("load %s (%s)", n, text);
250 }
251 break;
252 case MFL_UPGRADE:
253 ok = r->ops.source(r->ops.ctx, n, a->dir, a->hash, &src, &err) && r->ops.upgrade(r->ops.ctx, n, src, &text, &err);
254 if (ok) t = mt_fmt("upgrade %s to %016llx (%s)", n, (unsigned long long)a->hash, text);
255 break;
256 case MFL_ROLLBACK:
257 ok = r->ops.rollback(r->ops.ctx, n, &text, &err);
258 if (ok) t = mt_fmt("rollback %s (%s)", n, text);
259 break;
260 case MFL_RUN:
261 ok = r->ops.prog_run(r->ops.ctx, n, a->run, &text, &err);
262 if (ok) t = mt_fmt("%s %s", a->run ? "start" : "stop", n);
263 break;
264 case MFL_RESET:
265 ok = r->ops.reset(r->ops.ctx, n, &text, &err);
266 if (ok) {
267 reset_set(r, n, doc);
268 t = mt_fmt("reset %s (%s)", n, text);
269 }
270 break;
271 case MFL_MIGRATE:
272 migrate(r, doc, a, 1);
273 return true;
274 }
275 if (src) r->ops.src_free(r->ops.ctx, src);
276 free(text);
277 const mfd_prog *p = mfd_prog_get(doc, n);
278 uint64_t hash = p ? p->hash : 0;
279 if (a->kind == MFL_LOAD) moved(r, doc, n, ok, 0);
280 if (ok) {
281 mt_push_own(&r->done, t);
282 return true;
283 }
284 if (mt_starts(err, "unsupported")) {
285 if (!skip_get(r, n)) logf_(r, false, mt_fmt("reconcile: %s: not on this member (%s)", n, err));
286 skip_set(r, n, hash, err, true);
287 } else {
288 char *f = mt_fmt("%s %s: %s", verb(a->kind), n, err);
289 mt_push(&r->failed, f);
290 skip_set(r, n, hash, f, false);
291 free(f);
292 }
293 free(err);
294 return false;
295}
296
297/* A view from member from (the source may not have the committed document yet: retried for 3 s). */
298static bool fetch_view(mrc_rec *r, const char *name, uint64_t hash, uint32_t from, uint32_t *abi, mpw_bytes *b, char **err) {
299 uint64_t t0 = mono_us();
300 for (;;) {
301 char *e = NULL;
302 if (r->ops.fetch_view(r->ops.ctx, name, hash, from, abi, b, &e)) return true;
303 if (mt_contains(e, "not yet committed") && mono_us() - t0 < 3000000) {
304 free(e);
305 r->ops.sleep_ms(r->ops.ctx, 50);
306 continue;
307 }
308 *err = e;
309 return false;
310 }
311}
312
313typedef struct {
314 char *name;
315 uint64_t hash;
316 bool run;
317 uint32_t from;
318} loaded_t;
319
320/* Programs moving here from another member: load each stopped; take each view from the source,
321 * producers first; write the views (no OP_INIT); start them, consumers first. The gap is from the
322 * first view request to the last start. A failure is reported (the leader puts them back). */
323static void migrate(mrc_rec *r, const mfd_doc *doc, const mfl_action *migs, size_t n) {
324 mt_strs names = {0};
325 for (size_t i = 0; i < n; i++) mt_push(&names, migs[i].name);
326 mt_strs order = mrc_stop_order(doc, &names);
327 loaded_t *loaded = mt_alloc((n ? n : 1) * sizeof *loaded);
328 size_t nl = 0;
329 for (size_t oi = 0; oi < order.n; oi++) {
330 const mfl_action *a = NULL;
331 for (size_t i = 0; i < n; i++)
332 if (!strcmp(migs[i].name, order.v[oi])) a = &migs[i];
333 if (!a || a->kind != MFL_MIGRATE) continue;
334 void *src = NULL;
335 char *text = NULL, *err = NULL;
336 bool ok = r->ops.source(r->ops.ctx, a->name, a->dir, a->hash, &src, &err) && r->ops.load(r->ops.ctx, src, false, &text, &err);
337 if (src) r->ops.src_free(r->ops.ctx, src);
338 free(text);
339 if (ok) {
340 reset_set(r, a->name, doc);
341 loaded[nl++] = (loaded_t){a->name, a->hash, a->run, a->from};
342 } else if (mt_starts(err, "unsupported")) {
343 moved(r, doc, a->name, false, 0);
344 skip_set(r, a->name, a->hash, err, true);
345 free(err);
346 } else {
347 moved(r, doc, a->name, false, 0);
348 mt_push_own(&r->failed, mt_fmt("migrate %s: load: %s", a->name, err));
349 skip_set(r, a->name, a->hash, err, false);
350 free(err);
351 }
352 }
353 if (!nl) goto done;
354 /* Any failure below fails every program loaded for this move (they stay stopped; the leader puts them back). */
355 for (size_t i = 0; i < nl; i++) moved(r, doc, loaded[i].name, false, 0);
356 uint64_t t0 = mono_us();
357 size_t bytes = 0;
358 bool ok = true;
359 for (size_t i = 0; i < nl; i++) {
360 uint32_t abi = 0;
361 mpw_bytes v = {0};
362 char *err = NULL, *text = NULL;
363 if (fetch_view(r, loaded[i].name, loaded[i].hash, loaded[i].from, &abi, &v, &err)) {
364 bytes += v.n;
365 if (!r->ops.view_put(r->ops.ctx, loaded[i].name, abi, &v, &text, &err)) goto view_fail;
366 free(text);
367 mpw_bytes_free(&v);
368 continue;
369 }
370 view_fail:
371 ok = false;
372 mt_push_own(&r->failed, mt_fmt("migrate %s from system %u: %s", loaded[i].name, loaded[i].from, err));
373 skip_set(r, loaded[i].name, loaded[i].hash, err, false);
374 free(err);
375 mpw_bytes_free(&v);
376 }
377 if (!ok) goto done;
378 double view_ms = (double)(mono_us() - t0) / 1e3;
379 for (size_t i = nl; i-- > 0;) {
380 if (!loaded[i].run) continue;
381 char *err = NULL, *text = NULL;
382 if (!r->ops.prog_run(r->ops.ctx, loaded[i].name, true, &text, &err)) {
383 ok = false;
384 mt_push_own(&r->failed, mt_fmt("migrate %s: start: %s", loaded[i].name, err));
385 skip_set(r, loaded[i].name, loaded[i].hash, err, false);
386 free(err);
387 }
388 free(text);
389 }
390 if (!ok) goto done;
391 double gap = (double)(mono_us() - t0) / 1e3;
392 for (size_t i = 0; i < nl; i++) moved(r, doc, loaded[i].name, true, (uint32_t)(gap * 1e3));
393 char *joined = mt_join(&names, ",");
394 logf_(r, false, mt_fmt("reconcile: migrated %s from system %u: views %zu bytes in %.1f ms, gap %.1f ms", joined, loaded[0].from, bytes, view_ms, gap));
395 mt_push_own(&r->done, mt_fmt("migrate %s from %u: view %zu B, gap %.1f ms", joined, loaded[0].from, bytes, gap));
396 free(joined);
397done:
398 free(loaded);
399 mt_strs_free(&order);
400 mt_strs_free(&names);
401}
402
403static void set_group(mrc_rec *r, const mfd_group *g) {
404 mt_set(&r->g_name, g->name);
405 r->g_epoch = g->epoch, r->g_cycle = g->cycle, r->g_mask = mfd_group_mask(g);
406 r->has_group = true;
407}
408
409static void ready(mrc_rec *r, const mfd_group *g, uint64_t hash) {
410 mpk_hello spec = {.epoch = g->epoch, .cycle = g->cycle, .mask = mfd_group_mask(g), .hash = hash};
411 snprintf(spec.group, sizeof spec.group, "%s", g->name);
412 mfd_group_order(g, spec.order);
413 r->ops.ready(r->ops.ctx, &spec);
414}
415
416static bool loaded_at(mrc_rec *r, const char *program, uint64_t hash) {
417 size_t n;
418 mfl_actual *inv = inventory(r, &n);
419 bool yes = false;
420 for (size_t i = 0; i < n; i++) yes |= !strcmp(inv[i].name, program) && inv[i].hash == hash;
421 inv_free(inv, n);
422 return yes;
423}
424
425/* The lockstep program's group: a new epoch (reset), a cycle (upgrade, rollback, task), or its steps as they are. */
426static void group(mrc_rec *r, const mfd_doc *doc, const mfd_group *g, const mfl_actions *acts) {
427 uint32_t rank;
428 bool member = mfd_group_rank(g, r->ops.me, &rank);
429 const mfd_prog *p = mfd_prog_get(doc, g->program);
430 uint64_t hash = p ? p->hash : 0;
431 if (g->state == MFD_PAUSED || !member) {
432 /* Paused, or left out of this epoch: unfed; the next epoch resets it anyway. */
433 if (r->has_group || r->ops.want_set(r->ops.ctx)) r->ops.quiesce(r->ops.ctx, false, member);
434 for (size_t i = 0; i < acts->n; i++) act(r, doc, &acts->v[i]);
435 if (!member) r->ops.group_words(r->ops.ctx, 0, 0, g->epoch);
436 bool had = r->has_group;
437 r->has_group = false;
438 if (had || acts->n) mt_push_own(&r->done, mt_fmt("group %s epoch %u: %s", g->name, g->epoch, !member ? "not a member here (unfed)" : "paused"));
439 return;
440 }
441 bool same_epoch = r->has_group && !strcmp(r->g_name, g->name) && r->g_epoch == g->epoch && r->g_mask == mfd_group_mask(g);
442 if (!same_epoch) {
443 r->ops.quiesce(r->ops.ctx, false, true);
444 for (size_t i = 0; i < acts->n; i++) act(r, doc, &acts->v[i]);
445 if (!loaded_at(r, g->program, hash)) {
446 char *why = mt_fmt("group %s epoch %u: %s is not loaded here at %016llx; this member cannot join", g->name, g->epoch, g->program, (unsigned long long)hash);
447 skip_t *s = skip_get(r, g->program);
448 if (!(s && s->unsupported)) mt_push_own(&r->failed, why);
449 else mt_push_own(&r->done, why);
450 return;
451 }
452 r->has_cycle_failed = false;
453 r->ops.group_words(r->ops.ctx, rank, (uint32_t)g->members.n, g->epoch);
454 char *text = NULL, *err = NULL;
455 if (!r->ops.reset(r->ops.ctx, g->program, &text, &err)) {
456 mt_push_own(&r->failed, mt_fmt("group %s epoch %u: reset %s: %s", g->name, g->epoch, g->program, err));
457 free(err);
458 return;
459 }
460 free(text);
461 set_group(r, g);
462 mt_push_own(&r->done, mt_fmt("group %s epoch %u: rank %u of %zu (reset)", g->name, g->epoch, rank, g->members.n));
463 ready(r, g, hash);
464 } else if (r->has_group && r->g_cycle != g->cycle) {
465 /* A cycle inside the epoch: drain in step, apply, HELLO -- only once this member runs the image the
466 * document names and the cycle's task ran (cycle_ready). A task that failed here earlier in this
467 * epoch: a step behind for every later cycle too. */
468 if (r->has_cycle_failed && !strcmp(r->cf_group, g->name) && r->cf_epoch == g->epoch && r->cf_cycle <= g->cycle) {
469 mt_push(&r->failed, r->cf_why);
470 return;
471 }
472 r->ops.quiesce(r->ops.ctx, true, true);
473 for (size_t i = 0; i < acts->n; i++) act(r, doc, &acts->v[i]);
474 size_t ninv;
475 mfl_actual *inv = inventory(r, &ninv);
476 bool ran = false, task_ok = false;
477 char *task_err = NULL, *probe = NULL;
478 uint32_t result = 0;
479 if (g->op.kind == MFD_OP_TASK && image_ready(g, hash, inv, ninv, &probe)) {
480 ran = true;
481 task_ok = r->ops.task(r->ops.ctx, g->program, g->op.op, g->op.arg, &result, &task_err);
482 }
483 free(probe);
484 char *why = NULL;
485 if (!mrc_cycle_ready(g, hash, inv, ninv, ran ? &task_ok : NULL, task_err, &why)) {
486 logf_(r, true, mt_fmt("reconcile: %s", why));
487 /* A task that failed is not run again; a missing image is tried again (the retry, the next document). */
488 if (ran) {
489 r->has_cycle_failed = true;
490 mt_set(&r->cf_group, g->name);
491 r->cf_epoch = g->epoch, r->cf_cycle = g->cycle;
492 mt_set(&r->cf_why, why);
493 }
494 mt_push_own(&r->failed, why);
495 free(task_err);
496 inv_free(inv, ninv);
497 return;
498 }
499 free(task_err);
500 inv_free(inv, ninv);
501 if (ran && task_ok) mt_push_own(&r->done, mt_fmt("task %s -> %u (cycle %u)", g->program, result, g->cycle));
502 r->g_cycle = g->cycle;
503 mt_push_own(&r->done, mt_fmt("group %s cycle %u", g->name, g->cycle));
504 ready(r, g, hash);
505 } else {
506 for (size_t i = 0; i < acts->n; i++) act(r, doc, &acts->v[i]);
507 }
508}
509
510static const mt_strs *place_get(const mrc_rec *r, const char *name) {
511 for (size_t i = 0; i < r->nplaces; i++)
512 if (!strcmp(r->place_name[i], name)) return &r->place_words[i];
513 return NULL;
514}
515
516static void place_set(mrc_rec *r, const char *name, const mt_strs *w) {
517 for (size_t i = 0; i < r->nplaces; i++)
518 if (!strcmp(r->place_name[i], name)) {
519 mt_strs_free(&r->place_words[i]);
520 r->place_words[i] = mt_strs_clone(w);
521 return;
522 }
523 r->place_name = mt_realloc(r->place_name, (r->nplaces + 1) * sizeof *r->place_name);
524 r->place_words = mt_realloc(r->place_words, (r->nplaces + 1) * sizeof *r->place_words);
525 r->place_name[r->nplaces] = mt_strdup(name);
526 r->place_words[r->nplaces++] = mt_strs_clone(w);
527}
528
529static void place_remove(mrc_rec *r, const char *name) {
530 for (size_t i = 0; i < r->nplaces; i++)
531 if (!strcmp(r->place_name[i], name)) {
532 free(r->place_name[i]);
533 mt_strs_free(&r->place_words[i]);
534 r->place_name[i] = r->place_name[r->nplaces - 1];
535 r->place_words[i] = r->place_words[--r->nplaces];
536 return;
537 }
538}
539
540static int cmp_str(const void *a, const void *b) { return strcmp(*(char *const *)a, *(char *const *)b); }
541
542/* The leader runs each changed placement request (the driver moves the state). */
543static void place(mrc_rec *r, const mfd_doc *doc) {
544 /* want: the document's requests, and `local` for one dropped from it (sorted by name). */
545 mt_strs names = {0};
546 for (size_t i = 0; i < doc->nplaces; i++) mt_push(&names, doc->place_name[i]);
547 for (size_t i = 0; i < r->nplaces; i++)
548 if (!mfd_place_get(doc, r->place_name[i]) && mfd_prog_get(doc, r->place_name[i])) mt_push(&names, r->place_name[i]);
549 if (names.n > 1) qsort(names.v, names.n, sizeof *names.v, cmp_str);
550 for (size_t i = 0; i < names.n; i++) {
551 const mt_strs *dw = mfd_place_get(doc, names.v[i]);
552 mt_strs w = dw ? mt_strs_clone(dw) : (mt_strs){0};
553 if (!dw) mt_push(&w, "local");
554 const mt_strs *have = place_get(r, names.v[i]);
555 if (have && mt_strs_eq(have, &w)) {
556 mt_strs_free(&w);
557 continue;
558 }
559 char *text = NULL, *err = NULL, *j = mt_join(&w, " ");
560 if (r->ops.place(r->ops.ctx, names.v[i], &w, &text, &err)) {
561 mt_push_own(&r->done, mt_fmt("place %s %s: %s", names.v[i], j, text));
562 if (w.n == 1 && !strcmp(w.v[0], "local")) place_remove(r, names.v[i]);
563 else place_set(r, names.v[i], &w);
564 free(text);
565 } else {
566 mt_push_own(&r->failed, mt_fmt("place %s %s: %s", names.v[i], j, err));
567 free(err);
568 }
569 free(j);
570 mt_strs_free(&w);
571 }
572 mt_strs_free(&names);
573}
574
575typedef struct {
576 mrc_rec *r;
577} skipctx;
578
579static bool skip_fn(void *ctx, const char *name, uint64_t hash) {
580 mrc_rec *r = ctx;
581 skip_t *s = skip_get(r, name);
582 return s && s->hash == hash;
583}
584
585static void pass(mrc_rec *r, const mfd_doc *doc) {
586 uint32_t me = r->ops.me;
587 for (size_t i = r->nmoves; i-- > 0;) {
588 const mfd_prog *p = mfd_prog_get(doc, r->moves[i].name);
589 if (!p || !mfd_prog_here(p, me)) moves_clear_at(r, i);
590 }
591 /* 1. The programs (the lockstep program's steps wait for the group, below). */
592 size_t ninv;
593 mfl_actual *inv = inventory(r, &ninv);
594 char **rn = mt_alloc((r->resets.n ? r->resets.n : 1) * sizeof *rn);
595 uint32_t *rg = mt_alloc((r->resets.n ? r->resets.n : 1) * sizeof *rg);
596 for (size_t i = 0; i < r->resets.n; i++) rn[i] = r->resets.k[i], rg[i] = (uint32_t)strtoul(r->resets.v[i], NULL, 10);
597 mfl_actions actions = mfl_plan(doc, inv, ninv, skip_fn, r, (const char *const *)rn, rg, r->resets.n, me);
598 free(rn), free(rg);
599 inv_free(inv, ninv);
600 const char *lock = doc->ngroups ? doc->groups[0].program : NULL;
601 mfl_actions lock_acts = {0}, migs = {0}, rest = {0};
602 for (size_t i = 0; i < actions.n; i++) {
603 mfl_actions *to = lock && !strcmp(actions.v[i].name, lock) ? &lock_acts : actions.v[i].kind == MFL_MIGRATE ? &migs : &rest;
604 to->v = mt_realloc(to->v, (to->n + 1) * sizeof *to->v);
605 to->v[to->n++] = actions.v[i];
606 }
607 free(actions.v);
608 /* A lockstep program leaving the document: its group goes too. */
609 if (!lock && r->has_group) {
610 r->ops.quiesce(r->ops.ctx, false, false);
611 r->ops.group_words(r->ops.ctx, 0, 0, 0);
612 r->has_group = false;
613 }
614 for (size_t i = 0; i < rest.n; i++) act(r, doc, &rest.v[i]);
615 /* Programs migrating here, with their views (the scheduler's moves). */
616 if (migs.n) migrate(r, doc, migs.v, migs.n);
617 /* 2. The group. */
618 if (doc->ngroups) group(r, doc, &doc->groups[0], &lock_acts);
619 /* 3. Placement (the leader coordinates it). */
620 if (r->ops.is_leader(r->ops.ctx)) place(r, doc);
621 else places_clear(r);
622 /* Unsupported programs are reported, not failures. */
623 char **names = mt_alloc((r->nskip ? r->nskip : 1) * sizeof *names), **whys = mt_alloc((r->nskip ? r->nskip : 1) * sizeof *whys);
624 uint64_t *hashes = mt_alloc((r->nskip ? r->nskip : 1) * sizeof *hashes);
625 bool *uns = mt_alloc((r->nskip ? r->nskip : 1) * sizeof *uns);
626 for (size_t i = 0; i < r->nskip; i++) {
627 names[i] = r->skip[i].name, hashes[i] = r->skip[i].hash, uns[i] = r->skip[i].unsupported;
628 whys[i] = mt_strndup(r->skip[i].why, strlen(r->skip[i].why) > 200 ? 200 : strlen(r->skip[i].why));
629 }
630 r->ops.fails(r->ops.ctx, names, hashes, uns, whys, r->nskip);
631 for (size_t i = 0; i < r->nskip; i++) free(whys[i]);
632 free(names), free(whys), free(hashes), free(uns);
633 mt_strs us = {0};
634 for (size_t i = 0; i < r->nskip; i++) {
635 const mfd_prog *p = mfd_prog_get(doc, r->skip[i].name);
636 if (r->skip[i].unsupported && p && p->hash == r->skip[i].hash) mt_push(&us, r->skip[i].name);
637 }
638 if (us.n) {
639 char *j = mt_join(&us, ",");
640 mt_push_own(&r->done, mt_fmt("unsupported here: %s", j));
641 free(j);
642 }
643 mt_strs_free(&us);
644 mfl_actions_free(&lock_acts), mfl_actions_free(&migs), mfl_actions_free(&rest);
645}
646
647/* One pass over the current document, with its status before and after; true when it failed. */
648static bool run_pass(mrc_rec *r) {
649 const mfd_doc *d = &r->doc;
650 char *pending = mt_fmt("reconciling version %llu", (unsigned long long)d->version);
651 status(r, r->ops.applied(r->ops.ctx), MESH_RS_PENDING, pending);
652 free(pending);
653 mt_strs_free(&r->done);
654 mt_strs_free(&r->failed);
655 uint64_t t0 = mono_us();
656 pass(r, d);
657 double ms = (double)(mono_us() - t0) / 1e3;
658 if (!r->failed.n) {
659 char *done = mt_join(&r->done, ", ");
660 char *s = r->done.n ? mt_fmt("ok: %s", done) : mt_strdup("ok");
661 if (r->done.n) logf_(r, false, mt_fmt("reconcile: version %llu applied in %.1f ms: %s", (unsigned long long)d->version, ms, done));
662 status(r, d->version, MESH_RS_SYNC, s);
663 free(s), free(done);
664 return false;
665 }
666 char *s = mt_join(&r->failed, ", "), *done = mt_join(&r->done, ", ");
667 char *extra = r->done.n ? mt_fmt(" (done: %s)", done) : mt_strdup("");
668 logf_(r, true, mt_fmt("reconcile: version %llu (%.1f ms): %s%s", (unsigned long long)d->version, ms, s, extra));
669 status(r, d->version, MESH_RS_FAILED, s);
670 free(s), free(done), free(extra);
671 return true;
672}
673
674bool mrc_document(mrc_rec *r, const mfd_doc *doc) {
675 if (!r->has_doc || r->doc.version != doc->version) skip_keep_unsupported(r);
676 if (r->has_doc) mfd_doc_free(&r->doc);
677 r->doc = mfd_doc_clone(doc);
678 r->has_doc = true;
679 return run_pass(r);
680}
681
682bool mrc_retry(mrc_rec *r) {
683 skip_keep_unsupported(r);
684 return r->has_doc ? run_pass(r) : false;
685}
686
687/* ---- the thread */
688
689struct mrc_thread {
690 pthread_t th;
691 pthread_mutex_t mu;
692 pthread_cond_t cv;
693 mfd_doc newest;
694 bool has_newest, stop;
695 mrc_rec *rec;
696};
697
698static void *thread_main(void *arg) {
699 mrc_thread *t = arg;
700 bool retry = false;
701 struct timespec until = {0};
702 for (;;) {
703 pthread_mutex_lock(&t->mu);
704 while (!t->has_newest && !t->stop) {
705 if (!retry) pthread_cond_wait(&t->cv, &t->mu);
706 else if (pthread_cond_timedwait(&t->cv, &t->mu, &until) != 0) break;
707 }
708 if (t->stop) {
709 pthread_mutex_unlock(&t->mu);
710 return NULL;
711 }
712 bool got = t->has_newest;
713 mfd_doc d = t->newest;
714 t->has_newest = false;
715 t->newest = (mfd_doc){0};
716 pthread_mutex_unlock(&t->mu);
717 retry = got ? mrc_document(t->rec, &d) : mrc_retry(t->rec);
718 if (got) mfd_doc_free(&d);
719 if (retry) {
720 clock_gettime(CLOCK_REALTIME, &until);
721 until.tv_sec += RETRY_MS / 1000;
722 }
723 }
724}
725
726mrc_thread *mrc_thread_start(const mrc_ops *ops) {
727 mrc_thread *t = mt_alloc(sizeof *t);
728 pthread_mutex_init(&t->mu, NULL);
729 pthread_cond_init(&t->cv, NULL);
730 t->rec = mrc_new(ops);
731 pthread_create(&t->th, NULL, thread_main, t);
732 return t;
733}
734
735void mrc_post(mrc_thread *t, const mfd_doc *doc) {
736 pthread_mutex_lock(&t->mu);
737 if (t->has_newest) mfd_doc_free(&t->newest);
738 t->newest = mfd_doc_clone(doc);
739 t->has_newest = true;
740 pthread_cond_signal(&t->cv);
741 pthread_mutex_unlock(&t->mu);
742}
743
744void mrc_thread_stop(mrc_thread *t) {
745 pthread_mutex_lock(&t->mu);
746 t->stop = true;
747 pthread_cond_signal(&t->cv);
748 pthread_mutex_unlock(&t->mu);
749 pthread_join(t->th, NULL);
750 if (t->has_newest) mfd_doc_free(&t->newest);
751 mrc_free(t->rec);
752 pthread_mutex_destroy(&t->mu);
753 pthread_cond_destroy(&t->cv);
754 free(t);
755}