==> src/recon.c <==1/* The reconciler's ops (reconcile.rs's Rec methods that touch the driver, the catalog and the fleet
2 * thread). */
3#include "fleetd_int.h"
4
5#include "mesh/abi.h"
6#include "mesh/buf.h"
7#include "mesh/loader.h"
8#include "mesh/log.h"
9#include "mesh/rsfmt.h"
10
11#include <stdlib.h>
12#include <string.h>
13
14#define F(c) ((mesh_fleetd *)(c))
15#define SH(c) (F(c)->sh)
16
17static mfl_actual *inventory(void *c, size_t *n) { return fd_inventory(SH(c), 10000, n); }
18
19/* An image of name at hash: the local catalog's when it is that image, else a member's. */
20static bool source(void *c, const char *name, const char *dir, uint64_t hash, void **src, char **err) {
21 mesh_fleetd *f = c;
22 mesh_shared *sh = f->sh;
23 mesh_payload p;
24 mesh_manifest m;
25 uint64_t h;
26 char *e = NULL;
27 if (mesh_loader_identity(sh->programs, dir, &p, &m, &h, &e)) {
28 bool same = h == hash && strcmp(m.name, name) == 0;
29 mesh_manifest_free(&m);
30 if (same) {
31 mesh_src *s = mesh_xmalloc(sizeof *s);
32 *s = mesh_src_bytes(&p, hash);
33 mesh_payload_free(&p);
34 *src = s;
35 return true;
36 }
37 mesh_payload_free(&p);
38 } else free(e);
39 mpr_relay *r = fd_relay(f);
40 if (!r) {
41 *err = mesh_xstrdup("fleet thread gone");
42 return false;
43 }
44 mpr_payload got;
45 e = NULL;
46 if (!mpr_fetch_image(r, name, hash, dir, FD_FETCH_MS, &got, &e)) {
47 *err = mesh_buf_fmt("image %s %016llx: %s", name, (unsigned long long)hash, e ? e : "");
48 free(e);
49 return false;
50 }
51 mesh_info("reconcile: %s %016llx: image from a member (%zu bytes of PTX)", name, (unsigned long long)hash, got.ptx.n);
52 mesh_payload q = {got.dir, got.manifest, got.ptx.b, got.ptx.n};
53 mesh_src *s = mesh_xmalloc(sizeof *s);
54 *s = mesh_src_bytes(&q, hash);
55 mpr_payload_free(&got);
56 *src = s;
57 return true;
58}
59
60static void src_free(void *c, void *src) {
61 (void)c;
62 mesh_src_free(src);
63 free(src);
64}
65
66/* sh.ask(.., APPLY): the driver's Ok text, or its error. */
67static bool ask(mesh_shared *sh, mesh_cmd cmd, int64_t ms, char **text, char **err) {
68 char *t = NULL;
69 if (mesh_shared_ask(sh, cmd, ms, &t)) {
70 *text = t;
71 return true;
72 }
73 *err = t;
74 return false;
75}
76
77static mesh_cmd named(mesh_cmd_kind k, const char *name) {
78 mesh_cmd c = mesh_cmd_new(k);
79 c.name = mesh_xstrdup(name);
80 return c;
81}
82
83static bool load(void *c, void *src, bool run, char **text, char **err) {
84 mesh_cmd x = mesh_cmd_new(MESH_CMD_LOAD);
85 x.has_src = true;
86 x.src = mesh_src_copy(src);
87 x.run = run;
88 return ask(SH(c), x, FD_APPLY_MS, text, err);
89}
90
91static bool upgrade(void *c, const char *name, void *src, char **text, char **err) {
92 mesh_cmd x = named(MESH_CMD_UPGRADE, name);
93 x.has_src = true;
94 x.src = mesh_src_copy(src);
95 return ask(SH(c), x, FD_APPLY_MS, text, err);
96}
97
98static bool unload(void *c, const char *name, char **text, char **err) {
99 mesh_cmd x = named(MESH_CMD_UNLOAD, name);
100 x.now = false;
101 return ask(SH(c), x, FD_APPLY_MS, text, err);
102}
103
104static bool rollback(void *c, const char *name, char **text, char **err) { return ask(SH(c), named(MESH_CMD_ROLLBACK, name), FD_APPLY_MS, text, err); }
105
106static bool prog_run(void *c, const char *name, bool on, char **text, char **err) {
107 mesh_cmd x = named(MESH_CMD_PROG_RUN, name);
108 x.on = on;
109 return ask(SH(c), x, FD_APPLY_MS, text, err);
110}
111
112static bool reset(void *c, const char *name, char **text, char **err) { return ask(SH(c), named(MESH_CMD_RESET, name), FD_APPLY_MS, text, err); }
113
114/* A lockstep cycle's task: its result, or "no result in 60s". */
115static bool task(void *c, const char *name, uint32_t op, uint32_t arg, uint32_t *result, char **err) {
116 mesh_cmd x = named(MESH_CMD_TASK, name);
117 x.op = op;
118 x.arg = arg;
119 mesh_result res;
120 char *e = NULL;
121 if (!fd_ask(SH(c), x, FD_APPLY_MS, &res, &e)) {
122 if (strncmp(e, "no answer from the driver in ", 29) == 0) {
123 free(e);
124 mesh_buf b = MESH_BUF_INIT;
125 mesh_buf_puts(&b, "no result in ");
126 mesh_rs_duration_ms(&b, FD_APPLY_MS);
127 e = mesh_buf_take(&b);
128 }
129 *err = e;
130 return false;
131 }
132 *result = res.value;
133 mesh_result_free(&res);
134 return true;
135}
136
137static bool place(void *c, const char *name, const mt_strs *words, char **text, char **err) {
138 mesh_cmd x = named(MESH_CMD_PLACE, name);
139 x.has_words = true;
140 x.nwords = words->n;
141 x.words = mesh_xcalloc(words->n + 1, sizeof *x.words);
142 for (size_t i = 0; i < words->n; i++) x.words[i] = mesh_xstrdup(words->v[i]);
143 return ask(SH(c), x, FD_PLACE_MS, text, err);
144}
145
146static bool fetch_view(void *c, const char *name, uint64_t hash, uint32_t from, uint32_t *view_abi, mpw_bytes *bytes, char **err) {
147 mpr_relay *r = fd_relay(F(c));
148 if (!r) {
149 *err = mesh_xstrdup("fleet thread gone");
150 return false;
151 }
152 return mpr_fetch_view(r, name, hash, from, FD_FETCH_MS, view_abi, bytes, err);
153}
154
155static bool view_put(void *c, const char *name, uint32_t view_abi, const mpw_bytes *bytes, char **text, char **err) {
156 mesh_cmd x = named(MESH_CMD_VIEW_PUT, name);
157 x.view_abi = view_abi;
158 x.len = bytes->n;
159 x.bytes = mesh_xmalloc(bytes->n ? bytes->n : 1);
160 if (bytes->n) memcpy(x.bytes, bytes->b, bytes->n);
161 return ask(SH(c), x, FD_APPLY_MS, text, err);
162}
163
164/* Stop feeding the lockstep program and wait (2 s at most) until the fleet thread saw it. */
165static void quiesce(void *c, bool mirror, bool member) {
166 mesh_fleetd *f = c;
167 atomic_store(&f->sh->lock_feed, false);
168 pthread_mutex_lock(&f->mu);
169 f->want_set = false;
170 f->mirror = mirror;
171 f->member = member;
172 pthread_mutex_unlock(&f->mu);
173 mpr_relay *r = fd_relay(f);
174 if (!r) return;
175 mpr_group_want(r, NULL, mirror, member);
176 int64_t t0 = mesh_mono_ms();
177 for (;;) {
178 uint32_t s = mpr_group_state(r);
179 if ((s != MESH_GL_RUNNING && s != MESH_GL_HELLO) || mesh_mono_ms() - t0 >= 2000) break;
180 mesh_sleep_us(2000);
181 }
182}
183
184static bool want_set(void *c) {
185 mesh_fleetd *f = c;
186 pthread_mutex_lock(&f->mu);
187 bool w = f->want_set;
188 pthread_mutex_unlock(&f->mu);
189 return w;
190}
191
192static void group_words(void *c, uint32_t rank, uint32_t n, uint32_t epoch) {
193 mesh_mailbox mb = SH(c)->mb;
194 mesh_mailbox_set(mb, MESH_H_GROUP_RANK, rank);
195 mesh_mailbox_set(mb, MESH_H_GROUP_EPOCH, epoch);
196 mesh_mailbox_set(mb, MESH_H_GROUP_N, n);
197}
198
199static void ready(void *c, const mpk_hello *spec) {
200 mesh_fleetd *f = c;
201 pthread_mutex_lock(&f->mu);
202 f->want_set = true;
203 bool mirror = f->mirror, member = f->member;
204 pthread_mutex_unlock(&f->mu);
205 mpr_relay *r = fd_relay(f);
206 if (r) mpr_group_want(r, spec, mirror, member);
207}
208
209static bool is_leader(void *c) {
210 mesh_shared *sh = SH(c);
211 return mesh_host_block_get_u32(&sh->host, MESH_S_LEADER) == sh->me;
212}
213
214static void status(void *c, uint64_t applied, uint32_t st, const char *reason, const mst_outcome *moves, size_t nmoves) {
215 mesh_fleetd *f = c;
216 pthread_mutex_lock(&f->mu);
217 f->applied = applied;
218 f->status = st;
219 free(f->reason);
220 f->reason = mesh_xstrdup(reason);
221 for (size_t i = 0; i < f->nmoves; i++) free(f->moves[i].name);
222 free(f->moves);
223 f->moves = nmoves ? mesh_xcalloc(nmoves, sizeof *f->moves) : NULL;
224 for (size_t i = 0; i < nmoves; i++) {
225 f->moves[i] = moves[i];
226 f->moves[i].name = mesh_xstrdup(moves[i].name);
227 }
228 f->nmoves = nmoves;
229 pthread_mutex_unlock(&f->mu);
230 mpr_relay *r = atomic_load(&f->relay);
231 if (r) mpr_wake(r);
232}
233
234static void fails(void *c, char *const *names, const uint64_t *hashes, const bool *unsupported, char *const *whys, size_t n) {
235 mesh_fleetd *f = c;
236 msr_fail *v = n ? mesh_xcalloc(n, sizeof *v) : NULL;
237 for (size_t i = 0; i < n; i++) v[i] = (msr_fail){mesh_xstrdup(names[i]), hashes[i], unsupported[i], mesh_xstrdup(whys[i])};
238 pthread_mutex_lock(&f->mu);
239 for (size_t i = 0; i < f->nfails; i++) {
240 free(f->fails[i].name);
241 free(f->fails[i].why);
242 }
243 free(f->fails);
244 f->fails = v;
245 f->nfails = n;
246 pthread_mutex_unlock(&f->mu);
247}
248
249static uint64_t applied(void *c) {
250 mesh_fleetd *f = c;
251 pthread_mutex_lock(&f->mu);
252 uint64_t a = f->applied;
253 pthread_mutex_unlock(&f->mu);
254 return a;
255}
256
257static void logline(void *c, bool error, const char *line) {
258 (void)c;
259 if (error) mesh_error("%s", line);
260 else mesh_info("%s", line);
261}
262
263static void sleep_ms(void *c, uint32_t ms) {
264 (void)c;
265 mesh_sleep_us((int64_t)ms * 1000);
266}
267
268mrc_ops fd_recon_ops(mesh_fleetd *f) {
269 return (mrc_ops){
270 .ctx = f,
271 .me = f->sh->me,
272 .inventory = inventory,
273 .source = source,
274 .src_free = src_free,
275 .load = load,
276 .upgrade = upgrade,
277 .unload = unload,
278 .rollback = rollback,
279 .prog_run = prog_run,
280 .reset = reset,
281 .task = task,
282 .place = place,
283 .fetch_view = fetch_view,
284 .view_put = view_put,
285 .quiesce = quiesce,
286 .want_set = want_set,
287 .group_words = group_words,
288 .ready = ready,
289 .is_leader = is_leader,
290 .status = status,
291 .fails = fails,
292 .applied = applied,
293 .log = logline,
294 .sleep_ms = sleep_ms,
295 };
296}