1/* meshd's shared state, its messages, and the requests other threads make of the driver and the fleet. */
2#include "mesh/shared.h"
3#include "mesh/buf.h"
4#include "mesh/log.h"
5#include "mesh/rsfmt.h"
6
7#include <stdlib.h>
8#include <string.h>
9
10/* ---- Src ---- */
11
12mesh_src mesh_src_catalog(const char *dir) {
13    mesh_src s;
14    memset(&s, 0, sizeof s);
15    s.dir = mesh_xstrdup(dir);
16    return s;
17}
18
19mesh_src mesh_src_bytes(const mesh_payload *p, uint64_t hash) {
20    mesh_src s;
21    memset(&s, 0, sizeof s);
22    s.is_bytes = true;
23    mesh_payload_copy(&s.payload, p);
24    s.hash = hash;
25    return s;
26}
27
28mesh_src mesh_src_copy(const mesh_src *s) {
29    if (s->is_bytes) return mesh_src_bytes(&s->payload, s->hash);
30    return mesh_src_catalog(s->dir ? s->dir : "");
31}
32
33void mesh_src_free(mesh_src *s) {
34    free(s->dir);
35    if (s->is_bytes) mesh_payload_free(&s->payload);
36    memset(s, 0, sizeof *s);
37}
38
39char *mesh_src_what(const mesh_src *s) {
40    if (s->is_bytes) return mesh_buf_fmt("%s (from the peer)", s->payload.dir ? s->payload.dir : "");
41    return mesh_xstrdup(s->dir ? s->dir : "");
42}
43
44/* ---- Cmd ---- */
45
46mesh_cmd mesh_cmd_new(mesh_cmd_kind k) {
47    mesh_cmd c;
48    memset(&c, 0, sizeof c);
49    c.k = k;
50    c.run = -1;
51    return c;
52}
53
54static void strv_free(char **v, size_t n) {
55    for (size_t i = 0; i < n; i++) free(v[i]);
56    free(v);
57}
58
59void mesh_cmd_free(mesh_cmd *c) {
60    free(c->name);
61    if (c->has_src) mesh_src_free(&c->src);
62    free(c->bytes);
63    strv_free(c->words, c->nwords);
64    if (c->msg && c->msg_free) c->msg_free(c->msg);
65    for (size_t i = 0; i < c->nplaces; i++) { free(c->places[i].name); strv_free(c->places[i].words, c->places[i].nwords); }
66    free(c->places);
67    if (c->k == MESH_CMD_JOB_LEASE) mj_lease_msg_free(&c->lease);
68    mesh_reply_drop(c->reply);
69    memset(c, 0, sizeof *c);
70}
71
72void mesh_actuals_free(void *p) {
73    mesh_actuals *a = p;
74    if (!a) return;
75    for (size_t i = 0; i < a->n; i++) { free(a->v[i].name); free(a->v[i].dir); }
76    free(a->v);
77    free(a);
78}
79
80void mesh_view_bytes_free(void *p) {
81    mesh_view_bytes *v = p;
82    if (!v) return;
83    free(v->bytes);
84    free(v);
85}
86
87void mesh_job_progs_free(void *p) {
88    mesh_job_progs *j = p;
89    if (!j) return;
90    mj_progs_free(j->v, j->n);
91    free(j);
92}
93
94/* ---- PeerEvent, JobEvent ---- */
95
96mesh_peer_event mesh_peer_event_new(mesh_peer_event_kind k) {
97    mesh_peer_event e;
98    memset(&e, 0, sizeof e);
99    e.k = k;
100    return e;
101}
102
103void mesh_peer_event_free(mesh_peer_event *e) {
104    free(e->line);
105    free(e->text);
106    free(e->name);
107    free(e->dir);
108    if (e->msg && e->msg_free) e->msg_free(e->msg);
109    mesh_reply_drop(e->reply);
110    memset(e, 0, sizeof *e);
111}
112
113mesh_job_event mesh_job_event_new(mesh_job_event_kind k) {
114    mesh_job_event e;
115    memset(&e, 0, sizeof e);
116    e.k = k;
117    return e;
118}
119
120void mesh_job_event_free(mesh_job_event *e) {
121    free(e->body);
122    free(e->line);
123    mesh_reply_drop(e->reply);
124    if (e->k == MESH_JEV_DONE) mj_item_res_free(&e->res);
125    free(e->payload);
126    if (e->msg && e->msg_free) e->msg_free(e->msg);
127    memset(e, 0, sizeof *e);
128}
129
130/* ---- link frames ---- */
131
132static void drop_cmd(void *p) { mesh_cmd_free(p); }
133static void drop_pev(void *p) { mesh_peer_event_free(p); }
134static void drop_jev(void *p) { mesh_job_event_free(p); }
135static void drop_frame(void *p) { free(((mesh_link_frame *)p)->b); }
136
137mesh_chan *mesh_link_queue_new(size_t bound) { return mesh_chan_new(sizeof(mesh_link_frame), bound, drop_frame); }
138
139/* ---- Shared ---- */
140
141uint64_t mesh_shared_mono_us(void) {
142    static _Atomic int64_t t0;
143    int64_t z = 0, now = mesh_mono_us();
144    atomic_compare_exchange_strong(&t0, &z, now);
145    return (uint64_t)(now - atomic_load(&t0));
146}
147
148mesh_shared *mesh_shared_new(mesh_mailbox mb, const char *programs, bool peered, uint32_t me, uint32_t n, _Atomic bool *quit) {
149    mesh_shared_mono_us();
150    mesh_shared *sh = mesh_xcalloc(1, sizeof *sh);
151    sh->mb = mb;
152    mesh_host_block_init(&sh->host);
153    sh->programs = mesh_xstrdup(programs);
154    sh->cmd = mesh_chan_new(sizeof(mesh_cmd), 0, drop_cmd);
155    sh->peer = peered ? mesh_chan_new(sizeof(mesh_peer_event), 0, drop_pev) : NULL;
156    sh->jobs = mesh_chan_new(sizeof(mesh_job_event), 0, drop_jev);
157    sh->blobs = NULL;
158    sh->me = me;
159    sh->n = n;
160    atomic_init(&sh->peer_state, MESH_PEER_SINGLE);
161    atomic_init(&sh->lock_feed, false);
162    for (size_t k = 0; k < MESH_MAX_SYSTEMS; k++) atomic_init(&sh->round_of[k], 0);
163    atomic_init(&sh->ready, false);
164    atomic_init(&sh->fatal, false);
165    atomic_init(&sh->blob_mem, 0);
166    sh->quit = quit;
167    sh->fleet_text = mesh_xstrdup("");
168    sh->groups_text = mesh_xstrdup("");
169    sh->sched_all = mesh_xstrdup("");
170    sh->sched_dry = mesh_xstrdup("");
171    pthread_mutex_init(&sh->mu, NULL);
172    return sh;
173}
174
175void mesh_shared_free(mesh_shared *sh) {
176    if (!sh) return;
177    mesh_chan_release(sh->cmd);
178    if (sh->peer) mesh_chan_release(sh->peer);
179    mesh_chan_release(sh->jobs);
180    for (size_t k = 0; k < MESH_MAX_SYSTEMS; k++) {
181        if (sh->link_out[k]) mesh_chan_release(sh->link_out[k]);
182        if (sh->bulk_out[k]) mesh_chan_release(sh->bulk_out[k]);
183    }
184    free(sh->programs);
185    free(sh->fleet_text);
186    free(sh->groups_text);
187    free(sh->sched_all);
188    free(sh->sched_dry);
189    for (size_t i = 0; i < sh->nexplain; i++) { free(sh->sched_explain[i].name); free(sh->sched_explain[i].text); }
190    free(sh->sched_explain);
191    mesh_recs_free(&sh->fleet_view);
192    mesh_recs_free(&sh->jobs_view_recs);
193    free(sh->pipe_hist);
194    pthread_mutex_destroy(&sh->mu);
195    free(sh);
196}
197
198static mesh_link_frame frame_copy(const uint8_t *f, size_t n) {
199    mesh_link_frame x = {mesh_xmalloc(n ? n : 1), n};
200    if (n) memcpy(x.b, f, n);
201    return x;
202}
203
204/* try_send of a copy to a queue held under mu; the copy is freed when refused. */
205static int try_send(mesh_chan *q, const uint8_t *f, size_t n) {
206    mesh_link_frame x = frame_copy(f, n);
207    int r = mesh_chan_try_send(q, &x);
208    if (r != MESH_CHAN_OK) free(x.b);
209    return r;
210}
211
212bool mesh_shared_send_peer(mesh_shared *sh, uint32_t to, const uint8_t *frame, size_t n) {
213    if (atomic_load(&sh->fatal) || to >= MESH_MAX_SYSTEMS) return false;
214    pthread_mutex_lock(&sh->mu);
215    mesh_chan *q = sh->link_out[to];
216    int r = q ? try_send(q, frame, n) : MESH_CHAN_CLOSED;
217    pthread_mutex_unlock(&sh->mu);
218    return r == MESH_CHAN_OK;
219}
220
221bool mesh_shared_send_peer_wait(mesh_shared *sh, uint32_t to, const uint8_t *frame, size_t n, int64_t within_ms) {
222    if (atomic_load(&sh->fatal) || to >= MESH_MAX_SYSTEMS) return false;
223    int64_t t0 = mesh_mono_ms();
224    for (;;) {
225        pthread_mutex_lock(&sh->mu);
226        mesh_chan *q = sh->link_out[to];
227        int r = q ? try_send(q, frame, n) : MESH_CHAN_CLOSED;
228        pthread_mutex_unlock(&sh->mu);
229        if (r == MESH_CHAN_OK) return true;
230        if (r != MESH_CHAN_FULL || mesh_mono_ms() - t0 >= within_ms) return false;
231        mesh_sleep_us(500);
232    }
233}
234
235bool mesh_shared_send_bulk(mesh_shared *sh, uint32_t to, const uint8_t *frame, size_t n) {
236    if (atomic_load(&sh->fatal) || to >= MESH_MAX_SYSTEMS) return false;
237    pthread_mutex_lock(&sh->mu);
238    mesh_chan *q = sh->bulk_out[to];
239    bool ok = q && try_send(q, frame, n) == MESH_CHAN_OK;
240    /* A doorbell (an empty frame) wakes the writer if it waits on its control queue; a full one is awake. */
241    if (ok && sh->link_out[to]) try_send(sh->link_out[to], NULL, 0);
242    pthread_mutex_unlock(&sh->mu);
243    return ok;
244}
245
246bool mesh_shared_send_mask(mesh_shared *sh, uint32_t mask, const uint8_t *frame, size_t n) {
247    if (atomic_load(&sh->fatal)) return false;
248    pthread_mutex_lock(&sh->mu);
249    bool any = false, ok = true;
250    for (uint32_t k = 0; k < MESH_MAX_SYSTEMS; k++) {
251        if (!((mask >> k) & 1u) || !sh->link_out[k]) continue;
252        any = true;
253        ok &= try_send(sh->link_out[k], frame, n) == MESH_CHAN_OK;
254    }
255    pthread_mutex_unlock(&sh->mu);
256    return any && ok;
257}
258
259bool mesh_shared_broadcast(mesh_shared *sh, const uint8_t *frame, size_t n) { return mesh_shared_send_mask(sh, UINT32_MAX, frame, n); }
260
261void mesh_shared_fail_stop(mesh_shared *sh, const char *why) {
262    if (!atomic_exchange(&sh->fatal, true)) mesh_error("FATAL: %s: fail-stop (nothing more is sent; meshd exits)", why);
263    atomic_store(sh->quit, true);
264}
265
266static char *within_text(const char *prefix, int64_t within_ms) {
267    mesh_buf b = MESH_BUF_INIT;
268    mesh_buf_puts(&b, prefix);
269    mesh_rs_duration_ms(&b, (uint64_t)within_ms);
270    return mesh_buf_take(&b);
271}
272
273bool mesh_shared_ask(mesh_shared *sh, mesh_cmd cmd, int64_t within_ms, char **text) {
274    mesh_reply *r = mesh_reply_new();
275    mesh_reply_drop(cmd.reply);
276    cmd.reply = mesh_reply_sender(r);
277    if (mesh_chan_send(sh->cmd, &cmd) != MESH_CHAN_OK) {
278        mesh_cmd_free(&cmd);
279        mesh_reply_release(r);
280        *text = mesh_xstrdup("driver gone");
281        return false;
282    }
283    mesh_result x;
284    int w = mesh_reply_wait(r, within_ms, &x);
285    mesh_reply_release(r);
286    if (w != MESH_CHAN_OK) { *text = within_text("no answer from the driver in ", within_ms); return false; }
287    *text = x.text ? x.text : mesh_xstrdup("");
288    x.text = NULL;
289    bool ok = x.ok;
290    mesh_result_free(&x);
291    return ok;
292}
293
294void mesh_shared_set_peer_state(mesh_shared *sh, uint32_t s) {
295    atomic_store(&sh->peer_state, s);
296    mesh_host_block_set_u32(&sh->host, MESH_S_PEER_STATE, s);
297}
298
299bool mesh_shared_change(mesh_shared *sh, const char *line, int64_t within_ms, char **text) {
300    if (!sh->peer) { *text = mesh_xstrdup("not in a fleet"); return false; }
301    mesh_reply *r = mesh_reply_new();
302    mesh_peer_event e = mesh_peer_event_new(MESH_PEV_CHANGE);
303    e.line = mesh_xstrdup(line);
304    e.reply = mesh_reply_sender(r);
305    if (mesh_chan_send(sh->peer, &e) != MESH_CHAN_OK) {
306        mesh_peer_event_free(&e);
307        mesh_reply_release(r);
308        *text = mesh_xstrdup("fleet thread gone");
309        return false;
310    }
311    mesh_result x;
312    int w = mesh_reply_wait(r, within_ms, &x);
313    mesh_reply_release(r);
314    if (w != MESH_CHAN_OK) { *text = within_text("peer: no answer from the fleet in ", within_ms); return false; }
315    *text = x.text ? x.text : mesh_xstrdup("");
316    x.text = NULL;
317    bool ok = x.ok;
318    mesh_result_free(&x);
319    return ok;
320}
321
322bool mesh_shared_http(mesh_shared *sh, const uint8_t *head, size_t n, uint8_t **resp, size_t *rn, char **err) {
323    size_t l = 0;
324    while (l < n && head[l] != '\r' && head[l] != '\n') l++;
325    mesh_buf line = MESH_BUF_INIT;
326    mesh_rs_lossy(&line, head, l);
327    mesh_reply *r = mesh_reply_new();
328    mesh_cmd c = mesh_cmd_new(MESH_CMD_REQ);
329    c.name = mesh_xstrdup("http");
330    c.bytes = mesh_xmalloc(n ? n : 1);
331    memcpy(c.bytes, head, n);
332    c.len = n;
333    c.reply = mesh_reply_sender(r);
334    if (mesh_chan_send(sh->cmd, &c) != MESH_CHAN_OK) {
335        mesh_cmd_free(&c);
336        mesh_reply_release(r);
337        mesh_buf_free(&line);
338        *err = mesh_xstrdup("driver gone");
339        return false;
340    }
341    mesh_result x;
342    int w = mesh_reply_wait(r, 5000, &x);
343    mesh_reply_release(r);
344    if (w != MESH_CHAN_OK) { mesh_buf_free(&line); *err = mesh_xstrdup("GPU http timed out"); return false; }
345    if (!x.ok) {
346        *err = x.text ? x.text : mesh_xstrdup("");
347        x.text = NULL;
348        mesh_result_free(&x);
349        mesh_buf_free(&line);
350        return false;
351    }
352    mesh_host_block_add_u32(&sh->host, MESH_S_HTTP_REQUESTS, 1);
353    mesh_host_block_set_str(&sh->host, MESH_S_LAST_HTTP, 64, mesh_buf_str(&line));
354    mesh_buf_free(&line);
355    *resp = x.bytes ? x.bytes : mesh_xmalloc(1);
356    *rn = x.len;
357    x.bytes = NULL;
358    mesh_result_free(&x);
359    return true;
360}
361
362char *mesh_shared_text(mesh_shared *sh, char *const *field) {
363    pthread_mutex_lock(&sh->mu);
364    char *t = mesh_xstrdup(*field ? *field : "");
365    pthread_mutex_unlock(&sh->mu);
366    return t;
367}
368
369void mesh_shared_set_text(mesh_shared *sh, char **field, const char *text) {
370    char *t = mesh_xstrdup(text);
371    pthread_mutex_lock(&sh->mu);
372    free(*field);
373    *field = t;
374    pthread_mutex_unlock(&sh->mu);
375}