==> src/jobd.c <==1/* The jobs thread (jobd.rs `run`): the job node over the peer links, the driver, the job log and the
2 * blob thread. */
3#include "jobs_thread.h"
4
5#include "mesh/abi/peer_frames.h"
6#include "mesh/abi/peer_sizes.h"
7#include "mesh/debug_str.h"
8#include "mesh/jobd/text.h"
9#include "mesh/jobs/req.h"
10#include "mesh/log.h"
11#include "mesh/ref/sha256d.h"
12#include "mesh/rs_parse.h"
13#include "mesh/strbuf.h"
14
15#include <stdlib.h>
16#include <string.h>
17#include <time.h>
18
19/* Who waits for an answer (shared.rs JobReply). */
20typedef struct {
21 mesh_job_reply k;
22 mesh_reply *r; /* LOCAL */
23 uint32_t from, id; /* REMOTE: the member and its request id */
24} jd_reply;
25
26static void jd_reply_free(void *p) { mesh_reply_drop(((jd_reply *)p)->r); }
27
28uint64_t jd_unix_ms(void) {
29 struct timespec ts;
30 clock_gettime(CLOCK_REALTIME, &ts);
31 return (uint64_t)ts.tv_sec * 1000u + (uint64_t)ts.tv_nsec / 1000000u;
32}
33
34static uint64_t jd_now(const jd *x) { return (uint64_t)(mesh_mono_ms() - x->t0) + 1; }
35
36mj_view jd_view(const jd *x) {
37 if (x->peered) return mjd_ext_view(x->sh);
38 uint32_t me = x->d->cfg.me;
39 return (mj_view){1, true, me, 1u << me, 1u << me};
40}
41
42static void answer(jd *x, jd_reply *to, bool ok, const char *text) {
43 switch (to->k) {
44 case MESH_JOB_REPLY_LOCAL:
45 if (ok) mesh_reply_ok(to->r, text);
46 else mesh_reply_err(to->r, text);
47 to->r = NULL;
48 break;
49 case MESH_JOB_REPLY_REMOTE: {
50 size_t n = strlen(text);
51 if (n > MESH_F_CTL_REPLY_MAX) n = MESH_F_CTL_REPLY_MAX;
52 msb f = {0};
53 msb_putc(&f, MESH_F_CTL_REPLY);
54 uint32_t v[3] = {to->id, ok ? 1u : 0u, (uint32_t)n};
55 for (int i = 0; i < 3; i++)
56 for (int b = 0; b < 4; b++) msb_putc(&f, (char)(v[i] >> (8 * b)));
57 msb_put(&f, text, n);
58 mesh_shared_send_peer(x->sh, to->from, (const uint8_t *)f.p, f.n);
59 msb_free(&f);
60 break;
61 }
62 case MESH_JOB_REPLY_NONE: break;
63 }
64}
65
66static void blobs_send(jd *x, mjd_bev e) { mjd_blobs_send(x->sh, &e); }
67
68static void send_cmd(jd *x, mesh_cmd *c) {
69 if (mesh_chan_send(x->sh->cmd, c) != MESH_CHAN_OK) mesh_cmd_free(c);
70}
71
72/* The node's outputs, in order: a persist is synced before any send or answer after it (mj_durable);
73 * one that fails stops this member (fail-stop). */
74static void apply(jd *x, mj_outs *out) {
75 mesh_shared *sh = x->sh;
76 if (atomic_load(&sh->fatal)) {
77 mj_outs_free(out);
78 return;
79 }
80 bool ok;
81 char *err = NULL;
82 mj_outs outs = mj_durable(out, &x->disk, &ok, &err);
83 if (!ok) {
84 char *why = mfmt("jobs: cannot persist the job log (%s)", err ? err : "");
85 mesh_shared_fail_stop(sh, why);
86 free(why);
87 free(err);
88 mj_outs_free(&outs);
89 return;
90 }
91 /* Members whose queue stayed full in this batch: the rest of their table frames are dropped at once. */
92 uint32_t stalled = 0;
93 for (size_t i = 0; i < outs.n; i++) {
94 mj_out *o = &outs.v[i];
95 switch (o->kind) {
96 case MJ_OUT_PERSIST: break;
97 case MJ_OUT_SEND: {
98 msb f = {0};
99 mj_frame_encode(&o->frame, &f);
100 if (o->frame.ty == MESH_F_JOB_STATE || o->frame.ty == MESH_F_JOB_HAVE) {
101 /* The replicated table (deltas, snapshots and merges in parts) and pull answers wait for room
102 * in the link's queue (a while): a burst of them larger than the queue would otherwise never
103 * arrive whole. */
104 if (o->to < 32 && !(stalled >> o->to & 1) &&
105 !mesh_shared_send_peer_wait(sh, o->to, (const uint8_t *)f.p, f.n, MESH_JOBD_TABLE_SEND_WAIT_MS))
106 stalled |= 1u << o->to;
107 } else {
108 /* A full queue drops the frame (a lease expires, a result is sent again, the availability every second). */
109 mesh_shared_send_peer(sh, o->to, (const uint8_t *)f.p, f.n);
110 }
111 msb_free(&f);
112 break;
113 }
114 case MJ_OUT_RUN: {
115 mesh_cmd c = mesh_cmd_new(MESH_CMD_JOB_LEASE);
116 c.lease = o->run;
117 memset(&o->run, 0, sizeof o->run);
118 send_cmd(x, &c);
119 break;
120 }
121 case MJ_OUT_PUSH_BLOB: blobs_send(x, (mjd_bev){.k = MJD_BEV_PUSH, .to = o->to, .job = o->job, .hash = o->hash}); break;
122 case MJ_OUT_FETCH_BLOB: blobs_send(x, (mjd_bev){.k = MJD_BEV_FETCH, .job = o->job, .hash = o->hash, .holders = o->holders}); break;
123 case MJ_OUT_BLOB_DONE: blobs_send(x, (mjd_bev){.k = MJD_BEV_DONE, .job = o->job, .hash = o->hash}); break;
124 case MJ_OUT_CUT: {
125 mesh_cmd c = mesh_cmd_new(MESH_CMD_JOB_CUT);
126 c.lease_id = o->lease;
127 c.cut = o->cut;
128 send_cmd(x, &c);
129 break;
130 }
131 case MJ_OUT_ANSWER: {
132 jd_reply r;
133 if (momap_take(x->replies, &o->token, &r)) {
134 answer(x, &r, o->ok, o->text ? o->text : "");
135 jd_reply_free(&r);
136 }
137 break;
138 }
139 case MJ_OUT_LOG: mesh_info("%s", o->text ? o->text : ""); break;
140 }
141 }
142 mj_outs_free(&outs);
143}
144
145/* ---- control: `jobs`, `job status|results|submit|cancel|forget ..` */
146
147static char *bad_id(const char *id) {
148 char *q = mdebug(id);
149 char *e = mfmt("%s: a job id is <term>.<n>", q);
150 free(q);
151 return e;
152}
153
154static bool number(char **rest, size_t nrest, size_t i, uint64_t d, uint64_t *out, char **err) {
155 if (i >= nrest) {
156 *out = d;
157 return true;
158 }
159 if (rs_u64(rest[i], strlen(rest[i]), out)) return true;
160 char *q = mdebug(rest[i]);
161 *err = mfmt("%s: a number", q);
162 free(q);
163 return false;
164}
165
166static size_t split(char *s, char ***out) {
167 size_t n = 0, cap = 8;
168 char **w = mxalloc(cap * sizeof *w);
169 for (char *p = s; *p;) {
170 while (*p == ' ' || *p == '\t' || *p == '\n' || *p == '\r' || *p == '\v' || *p == '\f') p++;
171 if (!*p) break;
172 char *b = p;
173 while (*p && !(*p == ' ' || *p == '\t' || *p == '\n' || *p == '\r' || *p == '\v' || *p == '\f')) p++;
174 if (*p) *p++ = 0;
175 if (n == cap) w = mxrealloc(w, (cap *= 2) * sizeof *w);
176 w[n++] = b;
177 }
178 *out = w;
179 return n;
180}
181
182static void ctl(jd *x, const char *line, jd_reply *reply, uint64_t now, uint64_t unix_ms, mj_outs *outs) {
183 char *copy = mstrdup(line);
184 char **w;
185 size_t n = split(copy, &w);
186 mjd_text_cx cx = {x->node, jd_view(x), x->peered, unix_ms};
187 bool read = true, ok = false;
188 char *text = NULL;
189 mj_job_id id;
190 if (n == 1 && !strcmp(w[0], "jobs")) {
191 text = mjd_jobs_text(&cx);
192 ok = true;
193 } else if (n == 3 && !strcmp(w[0], "job") && !strcmp(w[1], "status")) {
194 if (!mj_parse_job(w[2], &id)) text = bad_id(w[2]);
195 else ok = mjd_status_text(&cx, id, &text);
196 } else if (n >= 3 && !strcmp(w[0], "job") && !strcmp(w[1], "results")) {
197 uint64_t from, cnt;
198 if (!mj_parse_job(w[2], &id)) text = bad_id(w[2]);
199 else if (number(w + 3, n - 3, 0, 0, &from, &text) && number(w + 3, n - 3, 1, MESH_JOBS_RESULTS_MAX, &cnt, &text))
200 ok = mjd_results_text(&cx, id, from, cnt < MESH_JOBS_RESULTS_MAX ? cnt : MESH_JOBS_RESULTS_MAX, &text);
201 } else read = false;
202 if (read) {
203 answer(x, reply, ok, text ? text : "");
204 free(text);
205 goto done;
206 }
207 mj_req req;
208 char *err = NULL;
209 if (n == 0 || strcmp(w[0], "job")) {
210 char *q = mdebug(line);
211 err = mfmt("not a job request: %s", q);
212 free(q);
213 } else if (mj_req_parse((const char *const *)w + 1, n - 1, &req, &err)) {
214 if (mj_node_leading(x->node) || !x->peered) {
215 x->token++;
216 jd_reply *slot = momap_put1(x->replies, x->token, NULL);
217 *slot = *reply;
218 reply->r = NULL;
219 mj_outs o = mj_node_request(x->node, req, x->token, now, unix_ms);
220 mj_outs_append(outs, &o);
221 mj_outs_free(&o);
222 goto done;
223 }
224 mj_req_free(&req);
225 /* Another member leads: the fleet thread forwards it (F_CTL) and routes the answer. */
226 if (x->sh->peer && reply->k == MESH_JOB_REPLY_LOCAL) {
227 mesh_peer_event e = mesh_peer_event_new(MESH_PEV_CHANGE);
228 e.line = mstrdup(line);
229 e.reply = reply->r;
230 reply->r = NULL;
231 if (mesh_chan_send(x->sh->peer, &e) != MESH_CHAN_OK) mesh_peer_event_free(&e);
232 goto done;
233 }
234 err = mfmt("jobs: system %u does not lead the job table (the fleet's leader does)", x->d->cfg.me);
235 }
236 answer(x, reply, false, err ? err : "");
237 free(err);
238done:
239 free(w);
240 free(copy);
241}
242
243/* ---- events */
244
245static void event(jd *x, mesh_job_event *e, uint64_t now, uint64_t unix_ms, mj_outs *out) {
246 mj_outs o = {0};
247 switch (e->k) {
248 case MESH_JEV_FRAME: {
249 mj_frame f;
250 char *err = NULL;
251 if (mj_frame_decode(e->ty, e->body, e->len, &f, &err)) o = mj_node_recv(x->node, e->from, f, now, unix_ms);
252 else {
253 mesh_error("jobs: a frame from system %u: %s", e->from, err ? err : "");
254 free(err);
255 }
256 break;
257 }
258 case MESH_JEV_DONE: {
259 mj_item_res r = e->res;
260 memset(&e->res, 0, sizeof e->res);
261 /* A payload: its blob named and handed to blobd before the result (staged until its copies are durable). */
262 if (e->has_payload && r.status == MESH_JOBS_RS_OK) {
263 mj_hash h;
264 mesh_sha256(e->payload, e->payload_len, h.b);
265 r.has_blob = true;
266 r.blob = (mj_blob_ref){h, (uint32_t)e->payload_len, 0};
267 blobs_send(x, (mjd_bev){.k = MJD_BEV_LOCAL, .job = r.job, .hash = h, .bytes = e->payload, .n = e->payload_len});
268 e->payload = NULL;
269 e->has_payload = false;
270 }
271 o = mj_node_result(x->node, r, now);
272 break;
273 }
274 case MESH_JEV_BLOBS_STORED: {
275 const mb_refs *hs = e->msg;
276 for (size_t i = 0; hs && i < hs->n; i++) {
277 mj_outs b = mj_node_blob_stored(x->node, hs->v[i].job, hs->v[i].h, now, unix_ms);
278 mj_outs_append(&o, &b);
279 mj_outs_free(&b);
280 }
281 break;
282 }
283 case MESH_JEV_BLOB_ACKED: {
284 const mb_ref *r = e->msg;
285 if (r) mj_node_blob_acked(x->node, e->from, r->job, r->h, e->ok);
286 break;
287 }
288 case MESH_JEV_BLOB_GONE: {
289 const mb_refs *hs = e->msg;
290 for (size_t i = 0; hs && i < hs->n; i++) mj_node_blob_gone(x->node, hs->v[i].job, &hs->v[i].h);
291 break;
292 }
293 case MESH_JEV_CTL: {
294 jd_reply r = {e->reply_to, e->reply, e->remote_from, e->remote_id};
295 e->reply = NULL;
296 ctl(x, e->line ? e->line : "", &r, now, unix_ms, &o);
297 jd_reply_free(&r);
298 break;
299 }
300 }
301 mj_outs_append(out, &o);
302 mj_outs_free(&o);
303}
304
305/* The job programs running here: asked every 250 ms, the answer picked up when the driver gives it
306 * (never waited for). */
307static void progs(jd *x) {
308 mesh_shared *sh = x->sh;
309 if (x->progs) {
310 mesh_result r;
311 int w = mesh_reply_poll(x->progs, &r);
312 if (w == MESH_CHAN_OK) {
313 /* Backpressure: while blobd holds too many of our payloads (their copies not durable yet), no new leases. */
314 uint64_t mem = atomic_load_explicit(&sh->blob_mem, memory_order_relaxed);
315 bool full = mem > MESH_BLOBD_BLOB_MEM_MAX;
316 if (full != x->throttled) {
317 x->throttled = full;
318 mesh_info("jobs: %s leases (%llu MiB of payloads wait for their copies)", full ? "no new" : "taking", (unsigned long long)(mem >> 20));
319 }
320 mesh_job_progs *p = r.ok ? r.ptr : NULL;
321 if (p && !full) {
322 mj_node_set_progs(x->node, p->v, p->n);
323 p->v = NULL;
324 p->n = 0;
325 } else if (p) mj_node_set_progs(x->node, NULL, 0);
326 mesh_result_free(&r);
327 }
328 if (w != MESH_CHAN_EMPTY) {
329 mesh_reply_release(x->progs);
330 x->progs = NULL;
331 }
332 }
333 if (!x->progs && mesh_mono_ms() - x->progs_at >= MESH_JOBD_PROGS_MS) {
334 x->progs_at = mesh_mono_ms();
335 mesh_reply *r = mesh_reply_new();
336 mesh_cmd c = mesh_cmd_new(MESH_CMD_JOB_PROGS);
337 c.reply = mesh_reply_sender(r);
338 if (mesh_chan_send(sh->cmd, &c) == MESH_CHAN_OK) x->progs = r;
339 else {
340 mesh_cmd_free(&c);
341 mesh_reply_release(r);
342 }
343 }
344}
345
346static void hist_free(void *p) {
347 jd_hist *h = p;
348 free(h->at);
349 free(h->done);
350}
351static void ranges_free(void *p) { mj_ranges_free(p); }
352
353void *mjd_jobs_main(void *arg) {
354 mjd *d = arg;
355 mesh_shared *sh = d->sh;
356 jd x = {.d = d, .sh = sh, .peered = sh->peer != NULL};
357 x.t0 = mesh_mono_ms();
358 x.disk = mjd_store_disk(d->store);
359 size_t njobs = momap_len(d->rep.committed.jobs);
360 if (x.peered)
361 mesh_info("jobs: system %u of %u: the job table %s (%zu job(s)); led by the fleet's leader, committed on %u of %u", d->cfg.me, d->cfg.n,
362 mjd_store_persisted(d->store) ? "is persisted" : "is kept in memory (no --jobs-state)", njobs, d->cfg.quorum, d->cfg.n);
363 else
364 mesh_info("jobs: system %u of %u: the job table %s (%zu job(s)); this meshd leads it", d->cfg.me, d->cfg.n,
365 mjd_store_persisted(d->store) ? "is persisted" : "is kept in memory (no --jobs-state)", njobs);
366 x.node = mj_node_new(d->cfg, d->rep);
367 memset(&d->rep, 0, sizeof d->rep);
368 x.replies = momap_new(1, sizeof(jd_reply), jd_reply_free, NULL);
369 x.hist = momap_new(1, sizeof(jd_hist), hist_free, NULL);
370 x.sent = momap_new(1, sizeof(mj_ranges), ranges_free, NULL);
371 x.told = momap_new(1, 1, NULL, NULL);
372 x.progs_at = mesh_mono_ms() - 1000;
373 x.publish_at = mesh_mono_ms();
374 for (;;) {
375 /* Fail-stop (mj_durable): this thread does nothing more. */
376 if (atomic_load(&sh->fatal) || atomic_load(&d->stop)) break;
377 mesh_job_event first;
378 int w = mesh_chan_recv(sh->jobs, &first, 5);
379 if (w == MESH_CHAN_CLOSED) break;
380 uint64_t now = jd_now(&x), unix_ms = jd_unix_ms();
381 mj_outs out = mj_node_set_view(x.node, jd_view(&x), now, unix_ms);
382 apply(&x, &out);
383 if (w == MESH_CHAN_OK) {
384 size_t k = 0;
385 mesh_job_event e = first;
386 do {
387 mj_outs o = {0};
388 event(&x, &e, now, unix_ms, &o);
389 mesh_job_event_free(&e);
390 apply(&x, &o);
391 } while (++k < 4096 && mesh_chan_try_recv(sh->jobs, &e) == MESH_CHAN_OK);
392 }
393 progs(&x);
394 /* The fleet scheduler's model for the lease choice (mj_select), refreshed every second. */
395 mjd_ext *ext = d->ext;
396 pthread_mutex_lock(&ext->mu);
397 mj_sched *cx = ext->sched;
398 ext->sched = NULL;
399 pthread_mutex_unlock(&ext->mu);
400 if (cx) mj_node_set_sched(x.node, cx);
401 out = mj_node_tick(x.node, now, unix_ms);
402 apply(&x, &out);
403 if (mesh_mono_ms() - x.publish_at >= MESH_JOBD_PUBLISH_MS) {
404 x.publish_at = mesh_mono_ms();
405 jd_publish(&x);
406 if (mjd_store_wants_compact(d->store)) {
407 /* A failed compaction leaves the log as it was (the rewrite goes through a renamed file): appends go on. */
408 char *err = NULL;
409 if (!mjd_store_rewrite(d->store, &x.node->rep, &err)) mesh_error("jobs: cannot compact the job log: %s", err ? err : "");
410 free(err);
411 }
412 }
413 }
414 if (x.progs) mesh_reply_release(x.progs);
415 momap_free(x.replies);
416 momap_free(x.hist);
417 momap_free(x.sent);
418 momap_free(x.told);
419 mj_node_free(x.node);
420 return NULL;
421}