==> src/host.c <==1/* The relay's host (mpr_host) over mesh_shared: what peer.rs reads and writes of Shared, and what it
2 * asks the driver and the catalog. */
3#include "fleetd_int.h"
4
5#include "mesh/abi.h"
6#include "mesh/buf.h"
7#include "mesh/jobd/jobd.h"
8#include "mesh/loader.h"
9#include "mesh/peer_link.h"
10#include "mesh/log.h"
11#include "mesh/rsfmt.h"
12#include "mesh/views.h"
13
14#include <stdlib.h>
15#include <string.h>
16
17/* ---- driver commands */
18
19bool fd_ask(mesh_shared *sh, mesh_cmd c, int64_t within_ms, mesh_result *out, char **err) {
20 mesh_reply *r = mesh_reply_new();
21 mesh_reply_drop(c.reply);
22 c.reply = mesh_reply_sender(r);
23 if (mesh_chan_send(sh->cmd, &c) != MESH_CHAN_OK) {
24 mesh_cmd_free(&c);
25 mesh_reply_release(r);
26 *err = mesh_xstrdup("driver gone");
27 return false;
28 }
29 int w = mesh_reply_wait(r, within_ms, out);
30 mesh_reply_release(r);
31 if (w != MESH_CHAN_OK) {
32 mesh_buf b = MESH_BUF_INIT;
33 mesh_buf_puts(&b, "no answer from the driver in ");
34 mesh_rs_duration_ms(&b, (uint64_t)within_ms);
35 *err = mesh_buf_take(&b);
36 return false;
37 }
38 if (!out->ok) {
39 *err = out->text ? out->text : mesh_xstrdup("");
40 out->text = NULL;
41 mesh_result_free(out);
42 return false;
43 }
44 return true;
45}
46
47/* The driver's inventory as fleet.rs Actuals (none when the driver does not answer in time). */
48mfl_actual *fd_inventory(mesh_shared *sh, int64_t within_ms, size_t *n) {
49 *n = 0;
50 mesh_result x;
51 char *err = NULL;
52 if (!fd_ask(sh, mesh_cmd_new(MESH_CMD_INVENTORY), within_ms, &x, &err)) {
53 free(err);
54 return NULL;
55 }
56 const mesh_actuals *a = x.ptr;
57 mfl_actual *v = NULL;
58 if (a && a->n) {
59 v = mesh_xcalloc(a->n, sizeof *v);
60 for (size_t i = 0; i < a->n; i++) {
61 const mesh_actual *y = &a->v[i];
62 v[i] = mfl_actual_new(y->name, y->dir, y->hash, y->has_prev, y->prev, y->run);
63 v[i].failed = y->failed;
64 v[i].lockstep = y->lockstep;
65 }
66 *n = a->n;
67 }
68 mesh_result_free(&x);
69 return v;
70}
71
72/* ---- the status block and the mailbox */
73
74#define F(c) ((mesh_fleetd *)(c))
75#define SH(c) (F(c)->sh)
76
77static void host_u32(void *c, size_t off, uint32_t v) {
78 mesh_shared *sh = SH(c);
79 mesh_host_block_set_u32(&sh->host, off, v);
80 if (off != MESH_S_LINKED) return;
81 /* The links' frame queues follow the linked members (peer.rs sets link_out / bulk_out on a link,
82 * and a link's end drops them): mesh_shared_send_peer refuses a member that is not linked. */
83 pthread_mutex_lock(&sh->mu);
84 for (uint32_t k = 0; k < sh->n && k < MESH_MAX_SYSTEMS; k++) {
85 if (k == sh->me) continue;
86 bool on = v >> k & 1;
87 if (on && !sh->link_out[k]) {
88 sh->link_out[k] = mesh_link_queue_new(MPN_LINK_QUEUE);
89 sh->bulk_out[k] = mesh_link_queue_new(MPN_BULK_QUEUE);
90 } else if (!on && sh->link_out[k]) {
91 mesh_chan_release(sh->link_out[k]);
92 mesh_chan_release(sh->bulk_out[k]);
93 sh->link_out[k] = sh->bulk_out[k] = NULL;
94 }
95 }
96 pthread_mutex_unlock(&sh->mu);
97}
98static void host_u64(void *c, size_t off, uint64_t v) { mesh_host_block_set_u64(&SH(c)->host, off, v); }
99static void host_str(void *c, size_t off, size_t cap, const char *s) { mesh_host_block_set_str(&SH(c)->host, off, cap, s); }
100static void host_bytes(void *c, size_t off, const uint8_t *b, size_t n) {
101 mesh_host_field f = {off, b, n};
102 mesh_host_block_set_many(&SH(c)->host, &f, 1);
103}
104static void host_read(void *c, size_t off, uint8_t *out, size_t n) {
105 memset(out, 0, n);
106 if (off >= MESH_SNAP_HOST) return;
107 if (n > MESH_SNAP_HOST - off) n = MESH_SNAP_HOST - off;
108 uint8_t *b = mesh_xmalloc(MESH_SNAP_HOST);
109 mesh_host_block_bytes(&SH(c)->host, b);
110 memcpy(out, b + off, n);
111 free(b);
112}
113static uint64_t uptime_ms(void *c) { return mesh_host_block_uptime_ms(&SH(c)->host); }
114static uint64_t mb_get64(void *c, size_t off) { return mesh_mailbox_get64(SH(c)->mb, off); }
115static uint32_t mb_get32(void *c, size_t off) { return mesh_mailbox_get(SH(c)->mb, off); }
116static void mb_read(void *c, size_t off, uint8_t *out, size_t n) { mesh_mailbox_read(SH(c)->mb, off, out, n); }
117static void mb_write(void *c, size_t off, const uint8_t *b, size_t n) { mesh_mailbox_write(SH(c)->mb, off, b, n, n); }
118static void mb_publish64(void *c, size_t off, uint64_t v) { mesh_mailbox_publish64(SH(c)->mb, off, v); }
119static void lock_feed(void *c, bool on) { atomic_store(&SH(c)->lock_feed, on); }
120
121/* peer.rs round_stats: the member's round counter, our rounds' trips and the wait for its round. */
122static void round_stats(void *c, uint32_t k, uint64_t n, double rtt_us, double wait_us) {
123 mesh_shared *sh = SH(c);
124 if (k >= MESH_MAX_SYSTEMS) return;
125 atomic_store(&sh->round_of[k], n);
126 pthread_mutex_lock(&sh->mu);
127 if (rtt_us >= 0) {
128 sh->rtt[k].sum += rtt_us;
129 sh->rtt[k].n++;
130 }
131 sh->wait[k].sum += wait_us;
132 sh->wait[k].n++;
133 pthread_mutex_unlock(&sh->mu);
134}
135
136/* ---- the reconciler */
137
138static void committed(void *c, const mfd_doc *doc) {
139 mesh_fleetd *f = c;
140 if (f->rec) mrc_post(f->rec, doc);
141}
142
143static void recon(void *c, mpr_recon *out) {
144 mesh_fleetd *f = c;
145 pthread_mutex_lock(&f->mu);
146 *out = (mpr_recon){.applied = f->applied, .status = f->status, .reason = mesh_xstrdup(f->reason ? f->reason : "")};
147 if (f->nmoves) {
148 out->moves = mesh_xcalloc(f->nmoves, sizeof *out->moves);
149 for (size_t i = 0; i < f->nmoves; i++) {
150 out->moves[i] = f->moves[i];
151 out->moves[i].name = mesh_xstrdup(f->moves[i].name);
152 }
153 out->nmoves = f->nmoves;
154 }
155 pthread_mutex_unlock(&f->mu);
156}
157
158/* ---- the catalog */
159
160/* peer.rs Relay::catalog: the leader's view of an image (its own catalog). */
161static bool catalog(void *c, const char *dir, mfe_image *out, char **err) {
162 mesh_shared *sh = SH(c);
163 mesh_payload p;
164 mesh_manifest m;
165 uint64_t hash;
166 if (!mesh_loader_identity(sh->programs, dir, &p, &m, &hash, err)) return false;
167 mesh_payload_free(&p);
168 bool ok = false;
169 char *name = NULL;
170 if (m.abi != MESH_RT_ABI) *err = mesh_buf_fmt("abi: %s declares abi %u, meshd runs %u", m.name, m.abi, MESH_RT_ABI);
171 else if ((name = mesh_catalog_name(dir, err))) {
172 memset(out, 0, sizeof *out);
173 out->dir = name;
174 out->name = mesh_xstrdup(m.name);
175 out->hash = hash;
176 out->lockstep = m.peer == MESH_PEER_MODE_LOCKSTEP;
177 for (size_t i = 0; i < m.nbackends; i++) mt_push(&out->backends, m.backends[i]);
178 out->autostart = m.autostart;
179 if (m.place) {
180 out->placeable = true;
181 out->place_cap = m.place->cap;
182 out->place_fanout = m.fanout;
183 }
184 mesh_needs_copy(&out->needs, &m.needs);
185 ok = true;
186 }
187 mesh_manifest_free(&m);
188 return ok;
189}
190
191static mfl_actual *inventory(void *c, size_t *n) { return fd_inventory(SH(c), 5000, n); }
192
193/* peer.rs serve_image: this member's catalog image of dir, if it is name at hash. */
194static bool image(void *c, const char *dir, const char *name, uint64_t hash, mpr_payload *out, char **err) {
195 mesh_shared *sh = SH(c);
196 mesh_payload p;
197 mesh_manifest m;
198 uint64_t h;
199 if (!mesh_loader_identity(sh->programs, dir, &p, &m, &h, err)) return false;
200 bool ok = false;
201 if (strcmp(m.name, name) != 0) *err = mesh_buf_fmt("system %u: its catalog's %s is %s, not %s", sh->me, dir, m.name, name);
202 else if (h != hash) *err = mesh_buf_fmt("system %u: its catalog's %s is %016llx, not %016llx", sh->me, dir, (unsigned long long)h, (unsigned long long)hash);
203 else {
204 memset(out, 0, sizeof *out);
205 out->dir = p.dir, p.dir = NULL;
206 out->manifest = p.manifest, p.manifest = NULL;
207 mpw_put(&out->ptx, p.ptx, p.ptx_len);
208 ok = true;
209 }
210 mesh_payload_free(&p);
211 mesh_manifest_free(&m);
212 return ok;
213}
214
215/* peer_sched.rs's view-out thread: stop the program here, then its view from the driver. */
216static bool stop_and_view(void *c, const char *name, uint32_t *view_abi, mpw_bytes *view, char **err) {
217 mesh_shared *sh = SH(c);
218 mesh_cmd run = mesh_cmd_new(MESH_CMD_PROG_RUN);
219 run.name = mesh_xstrdup(name);
220 run.on = false;
221 char *t = NULL;
222 if (!mesh_shared_ask(sh, run, 30000, &t)) {
223 *err = t;
224 return false;
225 }
226 free(t);
227 mesh_cmd get = mesh_cmd_new(MESH_CMD_VIEW_GET);
228 get.name = mesh_xstrdup(name);
229 mesh_result x;
230 char *e = NULL;
231 if (!fd_ask(sh, get, 10000, &x, &e)) {
232 /* The Rust's words for a view that did not come. */
233 if (strncmp(e, "no answer from the driver", 25) == 0) {
234 free(e);
235 e = mesh_xstrdup("no view from the driver");
236 }
237 *err = e;
238 return false;
239 }
240 const mesh_view_bytes *v = x.ptr;
241 *view_abi = v ? v->view_abi : 0;
242 memset(view, 0, sizeof *view);
243 if (v && v->len) mpw_put(view, v->bytes, v->len);
244 mesh_result_free(&x);
245 return true;
246}
247
248/* peer.rs send_view: the fleet's view for the driver's placement. */
249static void driver_view(void *c, uint32_t up, bool has_leader, uint32_t leader, const mfd_doc *places) {
250 mesh_shared *sh = SH(c);
251 mesh_cmd v = mesh_cmd_new(MESH_CMD_FLEET_VIEW);
252 v.up = up;
253 v.has_leader = has_leader;
254 v.leader = leader;
255 if (places && places->nplaces) {
256 v.places = mesh_xcalloc(places->nplaces, sizeof *v.places);
257 for (size_t i = 0; i < places->nplaces; i++) {
258 mesh_place_req *q = &v.places[i];
259 q->name = mesh_xstrdup(places->place_name[i]);
260 q->nwords = places->place_words[i].n;
261 q->words = mesh_xcalloc(q->nwords + 1, sizeof *q->words);
262 for (size_t j = 0; j < q->nwords; j++) q->words[j] = mesh_xstrdup(places->place_words[i].v[j]);
263 }
264 v.nplaces = places->nplaces;
265 }
266 if (mesh_chan_send(sh->cmd, &v) != MESH_CHAN_OK) mesh_cmd_free(&v);
267}
268
269/* ---- frames for others */
270
271static void place_free(void *m) {
272 mpw_place_free(m);
273 free(m);
274}
275static void frame_free(void *m) {
276 mpw_frame_free(m);
277 free(m);
278}
279
280/* F_PLACE / F_FRAME / F_INPUT straight to the driver (peer.rs's reader), decoded (the relay checked them). */
281static void place_frame(void *c, uint32_t from, mpw_bytes f) {
282 mesh_shared *sh = SH(c);
283 const uint8_t *b = f.b + 1;
284 char *err = NULL;
285 mesh_cmd cmd;
286 bool ok = false;
287 size_t rest;
288 switch (f.b[0]) {
289 case MESH_F_PLACE: {
290 mpw_place *m = mesh_xcalloc(1, sizeof *m);
291 if (mpw_place_rest_len(b, &rest, &err) && mpw_place_decode(b, b + MPW_PLACE_HEAD, rest, m, &err)) {
292 cmd = mesh_cmd_new(MESH_CMD_PEER_PLACE);
293 cmd.msg = m, cmd.msg_free = place_free, cmd.from = from;
294 ok = true;
295 } else free(m);
296 break;
297 }
298 case MESH_F_FRAME: {
299 mpw_frame *m = mesh_xcalloc(1, sizeof *m);
300 if (mpw_frame_rest_len(b, &rest, &err) && mpw_frame_decode(b, b + MPW_FRAME_HEAD, rest, m, &err)) {
301 cmd = mesh_cmd_new(MESH_CMD_PEER_FRAME);
302 cmd.msg = m, cmd.msg_free = frame_free, cmd.from = from;
303 ok = true;
304 } else free(m);
305 break;
306 }
307 case MESH_F_INPUT: {
308 char name[MESH_NAME_BYTES + 1];
309 if (mpw_input_len(b, &rest, &err) && mpw_name_of(b, MESH_NAME_BYTES, name, &err)) {
310 cmd = mesh_cmd_new(MESH_CMD_PEER_INPUT);
311 cmd.name = mesh_xstrdup(name);
312 cmd.bytes = mesh_xmalloc(rest ? rest : 1);
313 memcpy(cmd.bytes, b + MPW_INPUT_HEAD, rest);
314 cmd.len = rest;
315 ok = true;
316 }
317 break;
318 }
319 default: err = mesh_buf_fmt("bad frame type %u", f.b[0]);
320 }
321 if (ok) {
322 if (mesh_chan_send(sh->cmd, &cmd) != MESH_CHAN_OK) mesh_cmd_free(&cmd);
323 } else {
324 mesh_error("peer: system %u: %s", from, err ? err : "");
325 free(err);
326 }
327 mpw_bytes_free(&f);
328}
329
330static void jobs_frame(void *c, uint32_t from, uint8_t type, const uint8_t *body, size_t n) { mjd_jobs_frame(SH(c), from, type, body, n); }
331static void blobs_frame(void *c, uint32_t from, uint8_t type, const uint8_t *body, size_t n) { mjd_blobs_frame(SH(c), from, type, body, n); }
332
333/* The bridge's answer to a change: the control port's (or the jobs thread's) one-shot reply. */
334void fd_reply_fn(void *ctx, bool ok, const char *text) {
335 if (!ctx) return;
336 if (ok) mesh_reply_ok(ctx, text);
337 else mesh_reply_err(ctx, text);
338}
339
340typedef struct {
341 mesh_reply *r;
342 mpr_reply to;
343} pump_t;
344
345/* A local `job ...` answer for someone else's mpr_reply: waits for the jobs thread's. */
346static void *pump(void *p) {
347 pump_t *x = p;
348 mesh_result res;
349 if (mesh_reply_wait(x->r, FD_APPLY_MS, &res) == MESH_CHAN_OK) {
350 x->to.fn(x->to.ctx, res.ok, res.text ? res.text : "");
351 mesh_result_free(&res);
352 } else x->to.fn(x->to.ctx, false, "jobs thread gone");
353 mesh_reply_release(x->r);
354 free(x);
355 return NULL;
356}
357
358/* A `job ...` request for the jobs thread (JobEvent::Ctl): local with its reply, or member from's. */
359static void jobs_ctl(void *c, const char *line, bool remote, uint32_t from, uint32_t id, mpr_reply reply) {
360 mesh_shared *sh = SH(c);
361 mesh_job_event e = mesh_job_event_new(MESH_JEV_CTL);
362 e.line = mesh_xstrdup(line);
363 if (remote) {
364 e.reply_to = MESH_JOB_REPLY_REMOTE;
365 e.remote_from = from;
366 e.remote_id = id;
367 } else if (reply.fn == fd_reply_fn) {
368 /* The bridge's reply: the jobs thread answers the waiter directly. */
369 e.reply_to = MESH_JOB_REPLY_LOCAL;
370 e.reply = reply.ctx;
371 } else if (reply.fn) {
372 mesh_reply *r = mesh_reply_new();
373 e.reply_to = MESH_JOB_REPLY_LOCAL;
374 e.reply = mesh_reply_sender(r);
375 pump_t *x = mesh_xmalloc(sizeof *x);
376 *x = (pump_t){r, reply};
377 pthread_t t;
378 if (pthread_create(&t, NULL, pump, x) == 0) pthread_detach(t);
379 else {
380 mesh_reply_release(r);
381 free(x);
382 }
383 }
384 if (mesh_chan_send(sh->jobs, &e) != MESH_CHAN_OK) {
385 if (e.reply_to == MESH_JOB_REPLY_LOCAL && e.reply) {
386 mesh_reply_err(e.reply, "jobs thread gone");
387 e.reply = NULL;
388 }
389 mesh_job_event_free(&e);
390 }
391}
392
393static void jobs_view(void *c, uint64_t term, bool has_leader, uint32_t leader, uint32_t alive, uint32_t linked) {
394 mjd_set_view(SH(c), (mj_view){term, has_leader, leader, alive, linked});
395}
396static void jobs_sched(void *c, const msm_member *members, size_t n, const mfd_doc *doc, const msm_perfs *perf, const msw_weights *w) {
397 mjd_jobs_sched(SH(c), members, n, doc, perf, w);
398}
399
400/* ---- texts */
401
402static void texts(void *c, const char *fleet, const char *groups) {
403 mesh_shared *sh = SH(c);
404 mesh_shared_set_text(sh, &sh->fleet_text, fleet);
405 mesh_shared_set_text(sh, &sh->groups_text, groups);
406}
407
408static void sched_texts(void *c, const char *all, const msp_explains *explains, const char *dry, const char *records) {
409 mesh_shared *sh = SH(c);
410 mesh_explain *ex = explains->n ? mesh_xcalloc(explains->n, sizeof *ex) : NULL;
411 for (size_t i = 0; i < explains->n; i++) {
412 ex[i].name = mesh_xstrdup(explains->v[i].program);
413 ex[i].text = msp_explain_text(&explains->v[i]);
414 }
415 mesh_recs recs = {0};
416 mesh_views_parse(records, &recs);
417 pthread_mutex_lock(&sh->mu);
418 free(sh->sched_all);
419 sh->sched_all = mesh_xstrdup(all);
420 free(sh->sched_dry);
421 sh->sched_dry = mesh_xstrdup(dry);
422 for (size_t i = 0; i < sh->nexplain; i++) {
423 free(sh->sched_explain[i].name);
424 free(sh->sched_explain[i].text);
425 }
426 free(sh->sched_explain);
427 sh->sched_explain = ex;
428 sh->nexplain = explains->n;
429 mesh_recs old = sh->fleet_view;
430 sh->fleet_view = recs;
431 pthread_mutex_unlock(&sh->mu);
432 mesh_recs_free(&old);
433}
434
435static void peer_state(void *c, uint32_t s) { mesh_shared_set_peer_state(SH(c), s); }
436
437static void net_count(void *c, uint32_t k, bool tx, uint8_t type, size_t bytes) {
438 mesh_shared *sh = SH(c);
439 if (k >= MESH_MAX_SYSTEMS) return;
440 if (tx) mesh_net_counters_tx(&sh->host.net, k, &type, bytes);
441 else mesh_net_counters_rx(&sh->host.net, k, type, bytes);
442}
443
444static void logline(void *c, bool error, const char *line) {
445 (void)c;
446 if (error) mesh_error("%s", line);
447 else mesh_info("%s", line);
448}
449
450static void fail_stop(void *c, const char *why) { mesh_shared_fail_stop(SH(c), why); }
451
452mpr_host fd_host(mesh_fleetd *f) {
453 return (mpr_host){
454 .ctx = f,
455 .host_u32 = host_u32,
456 .host_u64 = host_u64,
457 .host_str = host_str,
458 .host_bytes = host_bytes,
459 .host_read = host_read,
460 .uptime_ms = uptime_ms,
461 .mb_get64 = mb_get64,
462 .mb_get32 = mb_get32,
463 .mb_read = mb_read,
464 .mb_write = mb_write,
465 .mb_publish64 = mb_publish64,
466 .lock_feed = lock_feed,
467 .round_stats = round_stats,
468 .committed = committed,
469 .recon = recon,
470 .catalog = catalog,
471 .inventory = inventory,
472 .image = image,
473 .stop_and_view = stop_and_view,
474 .driver_view = driver_view,
475 .place_frame = place_frame,
476 .jobs_frame = jobs_frame,
477 .blobs_frame = blobs_frame,
478 .jobs_ctl = jobs_ctl,
479 .jobs_view = jobs_view,
480 .jobs_sched = jobs_sched,
481 .texts = texts,
482 .sched_texts = sched_texts,
483 .peer_state = peer_state,
484 .net_count = net_count,
485 .log = logline,
486 .fail_stop = fail_stop,
487 };
488}