==> src/ext.c <==1/* mesh-jobd: opening (the job log replayed, the blob store indexed), the threads' lifetime, and what
2 * the other threads call: the fleet's view and scheduler, frame routing, `render`. */
3#include "internal.h"
4
5#include "mesh/abi/peer_frames.h"
6#include "mesh/abi/peer_sizes.h"
7#include "mesh/jobs/sched.h"
8#include "mesh/sched_needs.h"
9#include "mesh/log.h"
10#include "mesh/strbuf.h"
11
12#include <stdlib.h>
13#include <string.h>
14
15/* ---- the blob thread's events */
16
17void mjd_bev_free(void *p) {
18 mjd_bev *e = p;
19 free(e->bytes);
20 if (e->news) {
21 mbd_news_free(e->news);
22 free(e->news);
23 }
24 for (size_t i = 0; i < e->nwords; i++) free(e->words[i]);
25 free(e->words);
26 mesh_reply_drop(e->reply);
27 free(e->err);
28 memset(e, 0, sizeof *e);
29}
30
31void mjd_blobs_send(mesh_shared *sh, mjd_bev *e) {
32 if (!sh->blobs || mesh_chan_send(sh->blobs, e) != MESH_CHAN_OK) mjd_bev_free(e);
33}
34
35/* ---- sh->jobs_ext */
36
37mj_view mjd_ext_view(mesh_shared *sh) {
38 mjd_ext *x = mjd_ext_of(sh);
39 mj_view v = {0};
40 if (!x) return v;
41 pthread_mutex_lock(&x->mu);
42 v = x->view;
43 pthread_mutex_unlock(&x->mu);
44 return v;
45}
46
47size_t mjd_ext_watch(mesh_shared *sh, mj_job_id **out) {
48 mjd_ext *x = mjd_ext_of(sh);
49 *out = NULL;
50 if (!x) return 0;
51 pthread_mutex_lock(&x->mu);
52 size_t n = x->nwatch;
53 if (n) {
54 *out = mxalloc(n * sizeof **out);
55 memcpy(*out, x->watch, n * sizeof **out);
56 }
57 pthread_mutex_unlock(&x->mu);
58 return n;
59}
60
61/* The set is kept in id order (Rust's BTreeSet). */
62void mjd_ext_set_watch(mesh_shared *sh, mj_job_id id, bool on) {
63 mjd_ext *x = mjd_ext_of(sh);
64 if (!x) return;
65 pthread_mutex_lock(&x->mu);
66 size_t i = 0;
67 while (i < x->nwatch && x->watch[i] < id) i++;
68 bool has = i < x->nwatch && x->watch[i] == id;
69 if (on && !has) {
70 if (x->nwatch == x->capwatch) {
71 x->capwatch = x->capwatch ? 2 * x->capwatch : 8;
72 x->watch = mxrealloc(x->watch, x->capwatch * sizeof *x->watch);
73 }
74 memmove(&x->watch[i + 1], &x->watch[i], (x->nwatch - i) * sizeof *x->watch);
75 x->watch[i] = id;
76 x->nwatch++;
77 } else if (!on && has) {
78 memmove(&x->watch[i], &x->watch[i + 1], (x->nwatch - i - 1) * sizeof *x->watch);
79 x->nwatch--;
80 }
81 pthread_mutex_unlock(&x->mu);
82}
83
84void mjd_set_view(mesh_shared *sh, mj_view v) {
85 mjd_ext *x = mjd_ext_of(sh);
86 if (!x) return;
87 pthread_mutex_lock(&x->mu);
88 x->view = v;
89 pthread_mutex_unlock(&x->mu);
90}
91
92void mjd_set_sched(mesh_shared *sh, mj_sched *cx) {
93 mjd_ext *x = mjd_ext_of(sh);
94 if (!x) {
95 if (cx && cx->release) cx->release(cx->ctx);
96 free(cx);
97 return;
98 }
99 pthread_mutex_lock(&x->mu);
100 mj_sched *old = x->sched;
101 x->sched = cx;
102 pthread_mutex_unlock(&x->mu);
103 if (old) {
104 if (old->release) old->release(old->ctx);
105 free(old);
106 }
107}
108
109/* ---- frames and render */
110
111bool mjd_jobs_frame(mesh_shared *sh, uint32_t from, uint8_t ty, const uint8_t *body, size_t n) {
112 mesh_job_event e = mesh_job_event_new(MESH_JEV_FRAME);
113 e.from = from;
114 e.ty = ty;
115 e.len = n;
116 e.body = mxalloc(n ? n : 1);
117 memcpy(e.body, body, n);
118 if (mesh_chan_send(sh->jobs, &e) != MESH_CHAN_OK) {
119 mesh_job_event_free(&e);
120 return false;
121 }
122 return true;
123}
124
125bool mjd_blobs_frame(mesh_shared *sh, uint32_t from, uint8_t ty, const uint8_t *body, size_t n) {
126 if (!sh->blobs) return false;
127 mjd_bev e = {.k = MJD_BEV_FRAME, .from = from, .ty = ty, .n = n};
128 e.bytes = mxalloc(n ? n : 1);
129 memcpy(e.bytes, body, n);
130 if (mesh_chan_send(sh->blobs, &e) != MESH_CHAN_OK) {
131 mjd_bev_free(&e);
132 return false;
133 }
134 return true;
135}
136
137bool mjd_route_frame(mesh_shared *sh, uint32_t from, const uint8_t *frame, size_t n) {
138 if (n < 1 + MESH_JOB_HEAD) return false;
139 uint8_t ty = frame[0];
140 uint32_t len = (uint32_t)frame[1] | (uint32_t)frame[2] << 8 | (uint32_t)frame[3] << 16 | (uint32_t)frame[4] << 24;
141 if ((size_t)len != n - 1 - MESH_JOB_HEAD) return false;
142 const uint8_t *body = frame + 1 + MESH_JOB_HEAD;
143 if (ty >= MESH_F_JOB_AVAIL && ty <= MESH_F_RESULT) return mjd_jobs_frame(sh, from, ty, body, len);
144 if (ty >= MESH_F_BLOB_PUT && ty <= MESH_F_BLOB) return mjd_blobs_frame(sh, from, ty, body, len);
145 return false;
146}
147
148/* The scheduler's model as the fleet thread has it (peer_sched.rs): every program's needs from the
149 * committed document. */
150void mjd_jobs_sched(mesh_shared *sh, const msm_member *members, size_t n, const mfd_doc *doc, const msm_perfs *perf, const msw_weights *w) {
151 mjs_cx cx = {0};
152 cx.members = mxalloc((n ? n : 1) * sizeof *cx.members);
153 for (size_t i = 0; i < n; i++) cx.members[i] = msm_member_clone(&members[i]);
154 cx.nmembers = n;
155 cx.need_names = mxalloc((doc->nprogs ? doc->nprogs : 1) * sizeof *cx.need_names);
156 cx.needs = mxalloc((doc->nprogs ? doc->nprogs : 1) * sizeof *cx.needs);
157 for (size_t i = 0; i < doc->nprogs; i++) {
158 cx.need_names[i] = mstrdup(doc->pname[i]);
159 cx.needs[i] = msn_clone(&doc->prog[i].needs);
160 }
161 cx.nneeds = doc->nprogs;
162 cx.perf = msm_perfs_clone(perf);
163 cx.w = *w;
164 mjd_set_sched(sh, mjs_sched(cx));
165}
166
167bool mjd_render(mesh_shared *sh, char **words, size_t n, uint8_t **out, size_t *len, char **err) {
168 *out = NULL;
169 *len = 0;
170 *err = NULL;
171 if (!sh->blobs) {
172 *err = mstrdup("the blob thread is gone");
173 return false;
174 }
175 mesh_reply *r = mesh_reply_new();
176 mjd_bev e = {.k = MJD_BEV_RENDER, .nwords = n, .reply = mesh_reply_sender(r)};
177 e.words = mxalloc((n ? n : 1) * sizeof *e.words);
178 for (size_t i = 0; i < n; i++) e.words[i] = mstrdup(words[i]);
179 if (mesh_chan_send(sh->blobs, &e) != MESH_CHAN_OK) {
180 mjd_bev_free(&e);
181 mesh_reply_release(r);
182 *err = mstrdup("the blob thread is gone");
183 return false;
184 }
185 mesh_result x;
186 int w = mesh_reply_wait(r, 10000, &x);
187 mesh_reply_release(r);
188 if (w != MESH_CHAN_OK) {
189 *err = mstrdup(w == MESH_CHAN_CLOSED ? "the blob thread is gone" : "render: no answer from the blob thread in 10 s");
190 return false;
191 }
192 bool ok = x.ok;
193 if (ok) {
194 *out = x.bytes;
195 *len = x.len;
196 x.bytes = NULL;
197 } else {
198 *err = x.text ? x.text : mstrdup("");
199 x.text = NULL;
200 }
201 mesh_result_free(&x);
202 return ok;
203}
204
205/* ---- lifetime */
206
207static void store_log(void *ctx, const char *line) {
208 (void)ctx;
209 mesh_info("%s", line);
210}
211static void blob_log(void *ctx, const char *line) {
212 (void)ctx;
213 mesh_error("%s", line);
214}
215static void drop_write(void *p) { mbd_write_free(p); }
216
217mjd *mjd_open(mesh_shared *sh, const mjd_config *cfg, char **err) {
218 mjd *d = calloc(1, sizeof *d);
219 if (!d) abort();
220 d->sh = sh;
221 d->store = mjd_store_open(cfg->jobs_state, &d->rep, store_log, NULL, err);
222 if (!d->store) {
223 mj_replica_free(&d->rep);
224 free(d);
225 return NULL;
226 }
227 d->bstore = mb_open(cfg->blobs, cfg->blobs_cap, err);
228 if (!d->bstore) {
229 mjd_store_free(d->store);
230 mj_replica_free(&d->rep);
231 free(d);
232 return NULL;
233 }
234 mb_set_log(d->bstore, blob_log, NULL);
235 /* Its own leader when single; in a fleet the job table follows the fleet's leader (a majority commits). */
236 bool peered = sh->peer != NULL;
237 uint32_t n = sh->n ? sh->n : 1;
238 d->cfg = (mj_cfg){peered ? sh->me : 0, n, (!peered || n == 2) ? 1 : n / 2 + 1, cfg->dead_ms};
239 d->ext = calloc(1, sizeof *d->ext);
240 if (!d->ext) abort();
241 pthread_mutex_init(&d->ext->mu, NULL);
242 sh->jobs_ext = d->ext;
243 sh->blobs = mesh_chan_new(sizeof(mjd_bev), 0, mjd_bev_free);
244 d->wq = mesh_chan_new(sizeof(mbd_write), 0, drop_write);
245 return d;
246}
247
248static void spawn(pthread_t *t, void *(*f)(void *), void *arg) {
249 if (pthread_create(t, NULL, f, arg) != 0) {
250 mesh_error("jobs: cannot start a thread");
251 abort();
252 }
253}
254
255void mjd_start(mjd *d) {
256 if (d->started) return;
257 d->started = true;
258 for (unsigned i = 0; i < MESH_BLOBD_WRITERS; i++) spawn(&d->writers[i], mjd_writer_main, d);
259 spawn(&d->blobs_t, mjd_blobs_main, d);
260 spawn(&d->jobs_t, mjd_jobs_main, d);
261}
262
263void mjd_stop(mjd *d) {
264 if (!d->started) return;
265 atomic_store(&d->stop, true);
266 pthread_join(d->jobs_t, NULL);
267 pthread_join(d->blobs_t, NULL); /* its writer queue's sender goes with it: the writers drain and end */
268 for (unsigned i = 0; i < MESH_BLOBD_WRITERS; i++) pthread_join(d->writers[i], NULL);
269 d->started = false;
270}
271
272void mjd_free(mjd *d) {
273 if (!d) return;
274 mjd_stop(d);
275 mjd_store_free(d->store);
276 mj_replica_free(&d->rep);
277 mb_store_free(d->bstore);
278 mesh_chan_release(d->wq);
279 if (d->sh->jobs_ext == d->ext) d->sh->jobs_ext = NULL;
280 mesh_chan *b = d->sh->blobs;
281 d->sh->blobs = NULL;
282 if (b) mesh_chan_release(b);
283 if (d->ext->sched) {
284 if (d->ext->sched->release) d->ext->sched->release(d->ext->sched->ctx);
285 free(d->ext->sched);
286 }
287 free(d->ext->watch);
288 pthread_mutex_destroy(&d->ext->mu);
289 free(d->ext);
290 free(d);
291}