==> tests/jobd.c <==1/* The jobs thread and the blob thread over a mesh_shared in-process, with a fake driver: a single meshd
2 * runs a job with payloads to the end (leases, results, payload blobs made durable, the commit, the
3 * host block, the jobs view), the control answers and their errors, `render`, frame routing, backpressure
4 * on payload memory, a restart replaying the job log and indexing the blob store, and a follower in a
5 * fleet forwarding changes to the fleet thread and answering a remote request with F_CTL_REPLY. */
6#define _XOPEN_SOURCE 700 /* sexc compiles tests without the package's cflags */
7#include "mesh/jobd/jobd.h"
8
9#include "mesh/abi/host_jobs.h"
10#include "mesh/abi/peer_frames.h"
11#include "mesh/blobd/blobd.h"
12#include "mesh/blobs/durable.h"
13#include "mesh/blobs/store.h"
14#include "mesh/check.h"
15#include "mesh/sched_needs.h"
16#include "mesh/strbuf.h"
17
18#include <pthread.h>
19#include <stdatomic.h>
20#include <stdio.h>
21#include <stdlib.h>
22#include <string.h>
23#include <unistd.h>
24
25static _Atomic bool drv_stop;
26static _Atomic uint32_t leases, items_run;
27
28/* The driver: runs job program "p"; each item's result is 2 * item, its out the item (8 bytes), its
29 * payload 64 bytes made from the item. */
30static void *driver(void *arg) {
31 mesh_shared *sh = arg;
32 while (!atomic_load(&drv_stop)) {
33 mesh_cmd c;
34 if (mesh_chan_recv(sh->cmd, &c, 20) != MESH_CHAN_OK) continue;
35 if (c.k == MESH_CMD_JOB_PROGS) {
36 mesh_job_progs *p = mxalloc(sizeof *p);
37 p->v = mxalloc(sizeof *p->v);
38 p->v[0] = (mj_prog){mstrdup("p"), 0, 16};
39 p->n = 1;
40 mesh_reply_ptr(c.reply, p, mesh_job_progs_free);
41 c.reply = NULL;
42 } else if (c.k == MESH_CMD_JOB_LEASE) {
43 atomic_fetch_add(&leases, 1);
44 for (size_t r = 0; r < c.lease.items.n; r++)
45 for (uint64_t it = c.lease.items.v[r][0]; it < c.lease.items.v[r][1]; it++) {
46 mesh_job_event e = mesh_job_event_new(MESH_JEV_DONE);
47 e.res.lease = c.lease.lease;
48 e.res.job = c.lease.job;
49 e.res.item = it;
50 e.res.status = MESH_JOBS_RS_OK;
51 e.res.result = (uint32_t)(2 * it);
52 e.res.out = mxalloc(8);
53 for (int b = 0; b < 8; b++) e.res.out[b] = (uint8_t)(it >> (8 * b));
54 e.res.out_len = 8;
55 e.has_payload = true;
56 e.payload_len = 64;
57 e.payload = mxalloc(64);
58 for (int b = 0; b < 64; b++) e.payload[b] = (uint8_t)(it * 31 + (uint64_t)b);
59 atomic_fetch_add(&items_run, 1);
60 if (mesh_chan_send(sh->jobs, &e) != MESH_CHAN_OK) mesh_job_event_free(&e);
61 }
62 }
63 mesh_cmd_free(&c);
64 }
65 return NULL;
66}
67
68/* A control line through the jobs thread, as control.c sends it. */
69static bool ctl(mesh_shared *sh, const char *line, char **text) {
70 mesh_reply *r = mesh_reply_new();
71 mesh_job_event e = mesh_job_event_new(MESH_JEV_CTL);
72 e.line = mstrdup(line);
73 e.reply_to = MESH_JOB_REPLY_LOCAL;
74 e.reply = mesh_reply_sender(r);
75 CHECK(mesh_chan_send(sh->jobs, &e) == MESH_CHAN_OK);
76 mesh_result x;
77 int w = mesh_reply_wait(r, 20000, &x);
78 mesh_reply_release(r);
79 CHECK_I(w, MESH_CHAN_OK);
80 if (w != MESH_CHAN_OK) {
81 *text = mstrdup("");
82 return false;
83 }
84 bool ok = x.ok;
85 *text = x.text ? x.text : mstrdup("");
86 x.text = NULL;
87 mesh_result_free(&x);
88 return ok;
89}
90
91static bool wait_for(mesh_shared *sh, size_t off, uint64_t want, int ms) {
92 for (int i = 0; i < ms / 10; i++) {
93 uint64_t v = off == MESH_S_JOB_DONE ? (uint64_t)mesh_host_block_get_u32(&sh->host, off) | (uint64_t)mesh_host_block_get_u32(&sh->host, off + 4) << 32
94 : mesh_host_block_get_u32(&sh->host, off);
95 if (v == want) return true;
96 mesh_sleep_us(10000);
97 }
98 return false;
99}
100
101/* A field of the first record of kind (and job, when given) in the jobs view (a copy, or NULL). */
102static char *view_field(mesh_shared *sh, const char *kind, const char *job, const char *key) {
103 char *v = NULL;
104 pthread_mutex_lock(&sh->mu);
105 for (size_t i = 0; i < sh->jobs_view_recs.len && !v; i++) {
106 const mesh_rec *r = &sh->jobs_view_recs.items[i];
107 if (strcmp(r->kind, kind)) continue;
108 const char *j = mesh_rec_get(r, "job");
109 if (job && (!j || strcmp(j, job))) continue;
110 const char *x = mesh_rec_get(r, key);
111 v = mstrdup(x ? x : "-");
112 }
113 pthread_mutex_unlock(&sh->mu);
114 return v;
115}
116
117static size_t view_count(mesh_shared *sh, const char *kind) {
118 size_t n = 0;
119 pthread_mutex_lock(&sh->mu);
120 for (size_t i = 0; i < sh->jobs_view_recs.len; i++) n += strcmp(sh->jobs_view_recs.items[i].kind, kind) == 0;
121 pthread_mutex_unlock(&sh->mu);
122 return n;
123}
124
125static void single(const char *dir) {
126 static _Atomic bool quit;
127 char *jobs = mfmt("%s/jobs.log", dir), *blobs = mfmt("%s/blobs", dir);
128 mjd_config cfg = {jobs, blobs, 64u << 20, MESH_PEER_DEAD_MS};
129 for (int run = 0; run < 2; run++) {
130 mesh_mailbox mb = mesh_mailbox_test();
131 mesh_shared *sh = mesh_shared_new(mb, "x", false, 0, 1, &quit);
132 char *err = NULL;
133 mjd *d = mjd_open(sh, &cfg, &err);
134 CHECK(d != NULL);
135 if (!d) {
136 fprintf(stderr, "%s\n", err);
137 exit(1);
138 }
139 CHECK(sh->blobs != NULL && sh->jobs_ext != NULL);
140 atomic_store(&drv_stop, false);
141 pthread_t th;
142 pthread_create(&th, NULL, driver, sh);
143 mjd_start(d);
144 char *t;
145 if (run == 0) {
146 /* The scheduler's model (the relay's jobs_sched): no needs for p, so this member is eligible unscored. */
147 msm_member m = msm_member_default();
148 m.id = 0;
149 m.up = true;
150 mesh_needs nd = msn_default();
151 mfd_doc doc = {0};
152 doc.pname = mxalloc(sizeof *doc.pname);
153 doc.pname[0] = mstrdup("other");
154 doc.prog = mxalloc(sizeof *doc.prog);
155 doc.prog[0] = mfd_prog_new("other", 1, true, &nd);
156 doc.nprogs = 1;
157 msm_perfs perf = {0};
158 msw_weights w = msw_default();
159 mjd_jobs_sched(sh, &m, 1, &doc, &perf, &w);
160 mjd_jobs_sched(sh, &m, 1, &doc, &perf, &w); /* the second replaces the first before it is picked up, or after */
161 mfd_doc_free(&doc);
162 mesh_needs_free(&nd);
163 msm_member_free(&m);
164 CHECK(ctl(sh, "job submit p items 0-20", &t));
165 CHECK(strncmp(t, "job 1.1 program p items 20 on 1 member(s) (0)", 45) == 0);
166 free(t);
167 CHECK(wait_for(sh, MESH_S_JOB_DONE, 20, 20000));
168 CHECK(wait_for(sh, MESH_S_JOBS_ACTIVE, 0, 20000));
169 CHECK_U(atomic_load(&items_run), 20);
170 CHECK(atomic_load(&leases) >= 1);
171 mesh_sleep_us(400000); /* a publish after the last commit */
172 CHECK_U(mesh_host_block_get_u32(&sh->host, MESH_S_JOBS_TOTAL), 1);
173 CHECK_U(mesh_host_block_get_u32(&sh->host, MESH_S_JOB_FLAGS), MESH_JF_LEADER | MESH_JF_ACTIVE | MESH_JF_PERSIST);
174 char *v = view_field(sh, "jobs", NULL, "role");
175 CHECK_S(v, "leader");
176 free(v);
177 v = view_field(sh, "job", "1.1", "state");
178 CHECK_S(v, "done");
179 free(v);
180 v = view_field(sh, "job", "1.1", "done");
181 CHECK_S(v, "20");
182 free(v);
183 v = view_field(sh, "job", "1.1", "progress");
184 CHECK_S(v, "1.0000");
185 free(v);
186 v = view_field(sh, "share", "1.1", "done");
187 CHECK_S(v, "20");
188 free(v);
189 CHECK_U(view_count(sh, "result"), 3);
190 v = view_field(sh, "result", "1.1", "item");
191 CHECK_S(v, "19");
192 free(v);
193 v = view_field(sh, "result", "1.1", "out");
194 CHECK_S(v, "1300000000000000");
195 free(v);
196
197 /* The control answers and their errors. */
198 CHECK(ctl(sh, "job status 1.1", &t));
199 CHECK(strstr(t, "1.1") != NULL && strstr(t, "done") != NULL);
200 free(t);
201 CHECK(ctl(sh, "job results 1.1 18", &t));
202 const char *head = "results 1.1 total 20 from 18 count 2 done 20 failed 0 dupes 0 state done\n18 36 lease ";
203 CHECK(strncmp(t, head, strlen(head)) == 0);
204 CHECK(strstr(t, "\n19 38 lease ") != NULL && strstr(t, "\n17 ") == NULL);
205 CHECK(strstr(t, " out 1200000000000000 blob ") != NULL);
206 free(t);
207 CHECK(!ctl(sh, "job status x", &t));
208 CHECK_S(t, "\"x\": a job id is <term>.<n>");
209 free(t);
210 CHECK(!ctl(sh, "job results 1.1 zz", &t));
211 CHECK_S(t, "\"zz\": a number");
212 free(t);
213 CHECK(!ctl(sh, "frob it", &t));
214 CHECK_S(t, "not a job request: \"frob it\"");
215 free(t);
216 CHECK(!ctl(sh, "job status 1.9", &t));
217 CHECK(strstr(t, "no job 1.9") != NULL);
218 free(t);
219 CHECK(ctl(sh, "jobs", &t));
220 CHECK(strstr(t, "1.1") != NULL);
221 free(t);
222
223 /* render: not a pathtrace job; none at all. */
224 uint8_t *b;
225 size_t n;
226 char *w1[] = {"1.1"}, *w0[] = {"latest"};
227 CHECK(!mjd_render(sh, w1, 1, &b, &n, &err));
228 CHECK(strstr(err, "is not a pathtrace image") != NULL);
229 free(err);
230 CHECK(!mjd_render(sh, w0, 1, &b, &n, &err));
231 CHECK_S(err, "no pathtrace job in the committed table");
232 free(err);
233
234 /* Frame routing: neither a job nor a blob frame; a length that does not match; a short blob frame (logged). */
235 uint8_t f[9] = {MESH_F_CTL, 4, 0, 0, 0, 1, 2, 3, 4};
236 CHECK(!mjd_route_frame(sh, 1, f, 9));
237 f[0] = MESH_F_BLOB_GET;
238 CHECK(!mjd_route_frame(sh, 1, f, 8));
239 CHECK(mjd_route_frame(sh, 1, f, 9));
240
241 /* Backpressure: while blobd holds too many payloads in memory, this member takes no leases (it
242 * reports no job program: a submit finds no member that runs it). */
243 uint32_t before = atomic_load(&leases);
244 atomic_store(&sh->blob_mem, MESH_BLOBD_BLOB_MEM_MAX + 1);
245 mesh_sleep_us(600000);
246 CHECK(!ctl(sh, "job submit p items 0-5", &t));
247 CHECK(strncmp(t, "no member runs p", 16) == 0);
248 free(t);
249 CHECK_U(atomic_load(&leases), before);
250 atomic_store(&sh->blob_mem, 0);
251 mesh_sleep_us(600000);
252 CHECK(ctl(sh, "job submit p items 0-5", &t));
253 free(t);
254 CHECK(wait_for(sh, MESH_S_JOB_DONE, 25, 20000));
255 CHECK(atomic_load(&leases) > before);
256 CHECK(ctl(sh, "job forget 1.2", &t));
257 free(t);
258 } else {
259 /* The restart: the log replayed (1.2 forgotten), the blobs found. */
260 CHECK(wait_for(sh, MESH_S_JOBS_TOTAL, 1, 20000));
261 CHECK(wait_for(sh, MESH_S_JOB_DONE, 20, 20000));
262 CHECK(ctl(sh, "job status 1.1", &t));
263 CHECK(strstr(t, "done") != NULL);
264 free(t);
265 CHECK(!ctl(sh, "job status 1.2", &t));
266 free(t);
267 }
268 mjd_stop(d);
269 atomic_store(&drv_stop, true);
270 pthread_join(th, NULL);
271 mjd_free(d);
272 CHECK(sh->blobs == NULL && sh->jobs_ext == NULL);
273 mesh_shared_free(sh);
274 free(mb.base);
275 }
276 /* Every payload of job 1.1 is durable in the store (1.2's went with its forget). */
277 char *err = NULL;
278 mb_store *s = mb_open(blobs, 64u << 20, &err);
279 CHECK(s != NULL);
280 if (s) {
281 CHECK_U(mb_count(s), 20);
282 mb_store_free(s);
283 }
284 free(jobs);
285 free(blobs);
286}
287
288/* A follower of a fleet of 3 (system 1, member 0 leads): reads are answered here, changes go to the
289 * fleet thread, a remote request is answered with F_CTL_REPLY on the link. */
290static void follower(void) {
291 static _Atomic bool quit;
292 mesh_mailbox mb = mesh_mailbox_test();
293 mesh_shared *sh = mesh_shared_new(mb, "x", true, 1, 3, &quit);
294 mjd_config cfg = {NULL, NULL, 16u << 20, MESH_PEER_DEAD_MS};
295 char *err = NULL;
296 mjd *d = mjd_open(sh, &cfg, &err);
297 CHECK(d != NULL);
298 pthread_mutex_lock(&sh->mu);
299 sh->link_out[0] = mesh_link_queue_new(4096);
300 pthread_mutex_unlock(&sh->mu);
301 mjd_set_view(sh, (mj_view){5, true, 0, 7, 7});
302 atomic_store(&drv_stop, false);
303 pthread_t th;
304 pthread_create(&th, NULL, driver, sh);
305 mjd_start(d);
306 char *t;
307 CHECK(ctl(sh, "jobs", &t));
308 free(t);
309 /* A change: forwarded to the fleet thread, whose answer reaches the caller. */
310 mesh_reply *r = mesh_reply_new();
311 mesh_job_event e = mesh_job_event_new(MESH_JEV_CTL);
312 e.line = mstrdup("job cancel 1.1");
313 e.reply_to = MESH_JOB_REPLY_LOCAL;
314 e.reply = mesh_reply_sender(r);
315 CHECK(mesh_chan_send(sh->jobs, &e) == MESH_CHAN_OK);
316 mesh_peer_event pe;
317 CHECK_I(mesh_chan_recv(sh->peer, &pe, 20000), MESH_CHAN_OK);
318 CHECK(pe.k == MESH_PEV_CHANGE);
319 CHECK_S(pe.line, "job cancel 1.1");
320 mesh_reply_ok(pe.reply, "forwarded");
321 pe.reply = NULL;
322 mesh_peer_event_free(&pe);
323 mesh_result x;
324 CHECK_I(mesh_reply_wait(r, 20000, &x), MESH_CHAN_OK);
325 CHECK(x.ok);
326 CHECK_S(x.text, "forwarded");
327 mesh_result_free(&x);
328 mesh_reply_release(r);
329 /* A remote request (F_CTL from member 0, id 7): its answer on the link. */
330 e = mesh_job_event_new(MESH_JEV_CTL);
331 e.line = mstrdup("job status zz");
332 e.reply_to = MESH_JOB_REPLY_REMOTE;
333 e.remote_from = 0;
334 e.remote_id = 7;
335 CHECK(mesh_chan_send(sh->jobs, &e) == MESH_CHAN_OK);
336 bool seen = false;
337 for (int i = 0; i < 2000 && !seen; i++) {
338 mesh_link_frame f;
339 if (mesh_chan_recv(sh->link_out[0], &f, 10) != MESH_CHAN_OK) continue;
340 if (f.n >= 13 && f.b[0] == MESH_F_CTL_REPLY) {
341 seen = true;
342 const char *want = "\"zz\": a job id is <term>.<n>";
343 CHECK_U(f.b[1], 7);
344 CHECK_U(f.b[5], 0);
345 CHECK_U(f.b[9], strlen(want));
346 CHECK(f.n == 13 + strlen(want) && memcmp(f.b + 13, want, strlen(want)) == 0);
347 }
348 free(f.b);
349 }
350 CHECK(seen);
351 mesh_sleep_us(400000);
352 char *v = view_field(sh, "jobs", NULL, "role");
353 CHECK_S(v, "follower");
354 free(v);
355 v = view_field(sh, "jobs", NULL, "leader");
356 CHECK_S(v, "0");
357 free(v);
358 mjd_stop(d);
359 atomic_store(&drv_stop, true);
360 pthread_join(th, NULL);
361 mjd_free(d);
362 mesh_shared_free(sh);
363 free(mb.base);
364}
365
366int main(void) {
367 char tmpl[] = "/tmp/mesh-jobd-XXXXXX";
368 char *dir = mkdtemp(tmpl);
369 CHECK(dir != NULL);
370 single(dir);
371 follower();
372 mb_remove_tree(dir);
373 return mesh_check_done();
374}