==> tests/lifecycle.c <==1/* mesh-worker tests/lifecycle.rs: the native executor against the fixture
2 * worker (apps/mesh-test-worker) in a real shared mailbox: spawn and `ready`,
3 * launch, tasks, a halt that keeps the process (relaunch, same pid), a stop
4 * that ends it (reaped, no crash), a crash (FAILED reason: status, output
5 * tail), deaths mid-task, refusals (no binary, not executable, no
6 * source.hash, a script, a peer's hash, abi), identities, and on Linux the
7 * memory cap (OOM kill) and the CPU cap where systemd's user scopes work. */
8#include "mesh/check.h"
9#include "mesh/mailbox.h"
10#include "mesh/manifest_encode.h"
11#include "mesh/native.h"
12#include "mesh/proc_usage.h"
13#include "mesh/text.h"
14
15#include "mesh/abi/control_block.h"
16#include "mesh/abi/ext.h"
17#include "mesh/abi/header.h"
18#include "mesh/abi/ops.h"
19#include "mesh/abi/program_table.h"
20#include "mesh/abi/ring.h"
21
22#include <limits.h>
23#include <stdio.h>
24#include <stdlib.h>
25#include <string.h>
26#include <sys/stat.h>
27#include <sys/wait.h>
28#include <time.h>
29#include <unistd.h>
30
31static char *fixture;
32
33typedef struct {
34 char root[PATH_MAX];
35 mesh_shm shm;
36 mesh_mailbox mb;
37 mesh_backend *b;
38 mesh_native_host *host;
39 const mesh_compiler *comp;
40} env;
41
42static void run(char *const argv[]) {
43 pid_t p = fork();
44 if (p == 0) {
45 execvp(argv[0], argv);
46 _exit(127);
47 }
48 int st;
49 waitpid(p, &st, 0);
50}
51
52static void env_new(env *e, const char *tag) {
53 snprintf(e->root, sizeof e->root, "/tmp/mesh-native-%s-%d", tag, (int)getpid());
54 run((char *[]){"rm", "-rf", e->root, NULL});
55 mkdir(e->root, 0755);
56 char *err = NULL;
57 if (!mesh_shm_new(&e->shm, &err)) {
58 fprintf(stderr, "%s\n", err);
59 exit(1);
60 }
61 e->mb = mesh_mailbox_new(e->shm.base, 0);
62 mesh_mailbox_set(e->mb, MESH_H_MAGIC, MESH_MAGIC);
63 mesh_mailbox_set(e->mb, MESH_H_VERSION, MESH_VERSION);
64 char *logs = mesh_fmt("%s/logs", e->root);
65 mesh_worker_cfg cfg = {0};
66 e->b = mesh_native_backend_new(&e->shm, logs, &cfg);
67 free(logs);
68 e->host = mesh_native_backend_host(e->b);
69 e->comp = e->b->ops->compiler(e->b);
70}
71
72static void env_end(env *e) {
73 run((char *[]){"rm", "-rf", e->root, NULL});
74 mesh_backend_destroy(e->b);
75}
76
77static void write_file(const char *path, const char *s) {
78 FILE *f = fopen(path, "w");
79 fputs(s, f);
80 fclose(f);
81}
82
83static mesh_payload catalog(env *e, const char *dir, const char *manifest) {
84 char *d = mesh_fmt("%s/%s", e->root, dir), *t = mesh_fmt("%s/%s", d, mesh_native_target());
85 mkdir(d, 0755);
86 mkdir(t, 0755);
87 char *p = mesh_fmt("%s/program.toml", d), *h = mesh_fmt("%s/source.hash", d), *bin = mesh_native_binary(d, "nt");
88 write_file(p, manifest);
89 write_file(h, "0123456789abcdef\n");
90 run((char *[]){"cp", fixture, bin, NULL});
91 free(d);
92 free(t);
93 free(p);
94 free(h);
95 free(bin);
96 return (mesh_payload){mesh_strdup(dir), mesh_strdup(manifest), NULL, 0};
97}
98
99static char *manifest(const char *extra) {
100 return mesh_fmt("name = \"nt\"\nabi = 6\nbackends = [\"native\"]\nview_bytes = 0x1000\n%s%s", strstr(extra, "nodes") ? "" : "nodes = 2\n", extra);
101}
102
103/* Prepared and installed in slot (the table entry meshd writes): the worker kernel. */
104static mesh_kernel *prepare(env *e, mesh_payload p, uint32_t slot, char **err) {
105 mesh_prepared pr;
106 if (!e->comp->ops->prepare(e->comp, &p, NULL, e->root, &pr, err)) return NULL;
107 mesh_image im = {0x100000, 0, pr.hash, 0, 0, 0, 1};
108 uint8_t entry[MESH_PT_BYTES], ext[MESH_PX_BYTES];
109 mesh_manifest_encode_entry(&pr.manifest, &im, entry);
110 mesh_manifest_encode_ext(&pr.manifest, ext);
111 for (size_t i = 2; i < MESH_PT_BYTES / 4; i++) {
112 uint32_t w;
113 memcpy(&w, entry + 4 * i, 4);
114 mesh_mailbox_set(e->mb, mesh_prog_entry(slot) + 4 * i, w);
115 }
116 for (size_t i = 0; i < MESH_PX_BYTES / 4; i++) {
117 uint32_t w;
118 memcpy(&w, ext + 4 * i, 4);
119 mesh_mailbox_set(e->mb, mesh_prog_ext(slot) + 4 * i, w);
120 }
121 mesh_mailbox_set(e->mb, mesh_prog_entry(slot) + MESH_PT_STATE, MESH_ST_LOADED);
122 mesh_kernel *k = pr.kernel;
123 pr.kernel = NULL;
124 mesh_prepared_free(&pr);
125 return k;
126}
127
128static mesh_native_worker *W(mesh_kernel *k) { return mesh_kernel_native(k); }
129
130static uint64_t now_ms(void) {
131 struct timespec t;
132 clock_gettime(CLOCK_MONOTONIC, &t);
133 return (uint64_t)t.tv_sec * 1000 + (uint64_t)t.tv_nsec / 1000000;
134}
135
136static void launch(env *e, mesh_kernel *k, uint32_t slot, uint32_t tail) {
137 size_t c = mesh_prog_ctl(slot);
138 mesh_mailbox_set(e->mb, c + MESH_P_RING_HEAD, tail);
139 mesh_mailbox_set(e->mb, c + MESH_P_ONLINE, 0);
140 mesh_mailbox_set(e->mb, c + MESH_P_HALTED, 0);
141 mesh_mailbox_publish(e->mb, c + MESH_P_SHUTDOWN, 0);
142 char *err = NULL;
143 CHECK(mesh_native_worker_launch(W(k), slot, &err));
144 if (err) fprintf(stderr, "launch: %s\n", err);
145 free(err);
146}
147
148#define WAIT(what, ms, cond) \
149 do { \
150 uint64_t t0_ = now_ms(); \
151 while (!(cond)) { \
152 if (now_ms() - t0_ > (ms)) { \
153 CHECK(!what); \
154 break; \
155 } \
156 usleep(200); \
157 } \
158 } while (0)
159
160/* Publishes (op, arg) as ticket t and waits for its result; false with *why on a crash. */
161static bool task(env *e, mesh_kernel *k, uint32_t slot, uint32_t t, uint32_t op, uint32_t arg, uint32_t *res, char **why) {
162 size_t c = mesh_prog_ctl(slot), en = c + MESH_P_RING + MESH_ENTRY_BYTES * (t % MESH_PROG_RING), kk = t % MESH_PROG_RESULTS;
163 mesh_mailbox_set(e->mb, en + MESH_E_OP, op);
164 mesh_mailbox_set(e->mb, en + MESH_E_ARG, arg);
165 mesh_mailbox_set(e->mb, en + MESH_E_SLOT, t % MESH_PROG_RESULTS);
166 mesh_mailbox_publish(e->mb, en + MESH_E_SEQ, t + 1);
167 uint64_t t0 = now_ms();
168 for (;;) {
169 if (mesh_mailbox_get(e->mb, c + MESH_P_MARKS + 4 * kk) == t + 1) {
170 *res = mesh_mailbox_get(e->mb, c + MESH_P_RESULTS + 4 * kk);
171 return true;
172 }
173 char *cr = mesh_native_worker_crashed(W(k));
174 if (cr) {
175 *why = cr;
176 return false;
177 }
178 if (now_ms() - t0 > 20000) {
179 *why = mesh_fmt("ticket %u: no completion", (unsigned)t);
180 return false;
181 }
182 usleep(100);
183 }
184}
185
186static uint32_t ok_task(env *e, mesh_kernel *k, uint32_t slot, uint32_t t, uint32_t op, uint32_t arg) {
187 uint32_t r = 0;
188 char *why = NULL;
189 if (!CHECK(task(e, k, slot, t, op, arg, &r, &why))) {
190 fprintf(stderr, "task %u: %s\n", (unsigned)t, why);
191 free(why);
192 }
193 return r;
194}
195
196static char *err_task(env *e, mesh_kernel *k, uint32_t slot, uint32_t t, uint32_t op, uint32_t arg) {
197 uint32_t r;
198 char *why = NULL;
199 CHECK(!task(e, k, slot, t, op, arg, &r, &why));
200 return why ? why : mesh_strdup("");
201}
202
203static bool alive(pid_t pid) {
204 uint64_t c, r;
205 return mesh_proc_usage(pid, &c, &r);
206}
207
208static void spawn_tasks_halt_relaunch_stop(void) {
209 env e;
210 env_new(&e, "life");
211 char *m = manifest("param = 2"), *err = NULL;
212 mesh_kernel *k = prepare(&e, catalog(&e, "nt", m), 5, &err);
213 free(m);
214 if (!CHECK(k)) {
215 fprintf(stderr, "%s\n", err);
216 return;
217 }
218 /* A start after load with no process: spawned off the calling thread. */
219 uint64_t t0 = now_ms();
220 launch(&e, k, 5, 0);
221 CHECK(now_ms() - t0 < 50);
222 WAIT("online", 20000, mesh_mailbox_get(e.mb, mesh_prog_ctl(5) + MESH_P_ONLINE) == 2);
223 CHECK_U(ok_task(&e, k, 5, 0, MESH_OP_INIT, 0), MESH_R_NONE);
224 CHECK_U(mesh_mailbox_get(e.mb, 0x100000), 0x57544e4e);
225 for (uint32_t t = 1; t < 50; t++) CHECK_U(ok_task(&e, k, 5, t, MESH_OP_TASK, t), t * 3 + 7 + 2000);
226 pid_t pid = (pid_t)ok_task(&e, k, 5, 50, 5, 0);
227 CHECK(!mesh_native_worker_query(W(k)));
228 /* A halt (a blink): the nodes stop, the process stays for the next run. */
229 mesh_mailbox_publish(e.mb, mesh_prog_ctl(5) + MESH_P_SHUTDOWN, MESH_SHUTDOWN_HALT);
230 WAIT("halted", 5000, mesh_native_worker_query(W(k)));
231 CHECK_U(mesh_mailbox_get(e.mb, mesh_prog_ctl(5) + MESH_P_HALTED), 2);
232 CHECK(alive(pid));
233 launch(&e, k, 5, 51);
234 WAIT("online again", 5000, mesh_mailbox_get(e.mb, mesh_prog_ctl(5) + MESH_P_ONLINE) == 2);
235 CHECK_I(ok_task(&e, k, 5, 51, 5, 0), pid);
236 mesh_native_stats st = mesh_native_worker_stats(W(k));
237 CHECK(st.pid == pid && st.spawns == 1 && st.crashes == 0);
238 /* A stop: halt and exit; reaped, no crash. */
239 mesh_mailbox_publish(e.mb, mesh_prog_ctl(5) + MESH_P_SHUTDOWN, MESH_SHUTDOWN_EXIT);
240 WAIT("stopped", 5000, mesh_native_worker_query(W(k)));
241 char *cr = mesh_native_worker_crashed(W(k));
242 CHECK(cr == NULL);
243 if (cr) fprintf(stderr, "crashed: %s\n", cr);
244 free(cr);
245 WAIT("reaped", 2000, !alive(pid));
246 /* Its output went to the log file. */
247 char *lp = mesh_fmt("%s/logs/nt.log", e.root), *want = mesh_fmt("test worker %d set up", (int)pid);
248 FILE *f = fopen(lp, "r");
249 char log[8192] = {0};
250 if (f) {
251 size_t n = fread(log, 1, sizeof log - 1, f);
252 log[n] = 0;
253 fclose(f);
254 }
255 CHECK(strstr(log, want) != NULL);
256 free(lp);
257 free(want);
258 /* The next start spawns a new process. */
259 launch(&e, k, 5, 52);
260 WAIT("online after a stop", 20000, mesh_mailbox_get(e.mb, mesh_prog_ctl(5) + MESH_P_ONLINE) == 2);
261 pid_t pid2 = (pid_t)ok_task(&e, k, 5, 52, 5, 0);
262 CHECK(pid2 != pid);
263 CHECK_U(mesh_native_worker_stats(W(k)).spawns, 2);
264 mesh_native_worker_release(W(k));
265 WAIT("released", 5000, !alive(pid2));
266 mesh_kernel_drop(k);
267 env_end(&e);
268}
269
270static void warm_then_launch_is_immediate(void) {
271 env e;
272 env_new(&e, "warm");
273 char *m = manifest(""), *err = NULL;
274 mesh_kernel *k = prepare(&e, catalog(&e, "nt", m), 1, &err);
275 free(m);
276 CHECK(mesh_native_worker_warm(W(k), &err));
277 pid_t pid = mesh_native_worker_stats(W(k)).pid;
278 CHECK(pid != 0);
279 uint64_t t0 = now_ms();
280 launch(&e, k, 1, 0);
281 WAIT("online", 5000, mesh_mailbox_get(e.mb, mesh_prog_ctl(1) + MESH_P_ONLINE) == 2);
282 uint64_t up = now_ms() - t0;
283 CHECK_I(ok_task(&e, k, 1, 0, 5, 0), pid);
284 CHECK(up < 100);
285 mesh_kernel_drop(k);
286 env_end(&e);
287}
288
289static void crashes_fail_with_status_and_output(void) {
290 env e;
291 env_new(&e, "crash");
292 char *m = manifest(""), *err = NULL;
293 mesh_kernel *k = prepare(&e, catalog(&e, "nt", m), 2, &err);
294 free(m);
295 launch(&e, k, 2, 0);
296 WAIT("online", 20000, mesh_mailbox_get(e.mb, mesh_prog_ctl(2) + MESH_P_ONLINE) == 2);
297 char *r = err_task(&e, k, 2, 0, 9, 42);
298 CHECK(strstr(r, "exit 101") && strstr(r, "deliberate crash (op 9, arg 42)") && strstr(r, "while running"));
299 fprintf(stderr, "crash report: %s\n", r);
300 free(r);
301 CHECK(mesh_native_worker_query(W(k)));
302 mesh_native_stats st = mesh_native_worker_stats(W(k));
303 CHECK(st.crashes == 1 && st.pid == 0);
304 char *again = mesh_native_worker_crashed(W(k));
305 CHECK(again == NULL);
306 free(again);
307 /* exit(3) mid-task is a crash too, and so is a signal; the next launch is a new process. */
308 launch(&e, k, 2, 1);
309 WAIT("online", 20000, mesh_mailbox_get(e.mb, mesh_prog_ctl(2) + MESH_P_ONLINE) == 2);
310 r = err_task(&e, k, 2, 1, 13, 3);
311 CHECK(strstr(r, "exit 3") != NULL);
312 free(r);
313 launch(&e, k, 2, 2);
314 WAIT("online", 20000, mesh_mailbox_get(e.mb, mesh_prog_ctl(2) + MESH_P_ONLINE) == 2);
315 r = err_task(&e, k, 2, 2, 13, 0);
316 CHECK(strstr(r, "signal 15") != NULL);
317 free(r);
318 CHECK_U(mesh_native_worker_stats(W(k)).crashes, 3);
319 mesh_kernel_drop(k);
320 env_end(&e);
321}
322
323static char *prepare_err(env *e, mesh_payload p, const uint64_t *expect) {
324 mesh_prepared pr;
325 char *err = NULL;
326 if (e->comp->ops->prepare(e->comp, &p, expect, e->root, &pr, &err)) {
327 mesh_prepared_free(&pr);
328 return mesh_strdup("(prepared)");
329 }
330 return err;
331}
332
333static void refusals(void) {
334 env e;
335 env_new(&e, "refuse");
336 char *m = manifest(""), *r, *d, *b;
337 /* No binary for this target: `unsupported`. */
338 mesh_payload p = catalog(&e, "nobin", m);
339 d = mesh_fmt("%s/nobin", e.root);
340 b = mesh_native_binary(d, "nt");
341 unlink(b);
342 r = prepare_err(&e, p, NULL);
343 CHECK(strncmp(r, "unsupported: nt has no binary for", 33) == 0);
344 free(r), free(d), free(b);
345 /* Not executable. */
346 p = catalog(&e, "noexec", m);
347 d = mesh_fmt("%s/noexec", e.root);
348 b = mesh_native_binary(d, "nt");
349 chmod(b, 0644);
350 r = prepare_err(&e, p, NULL);
351 CHECK(strstr(r, "not executable") != NULL);
352 free(r), free(d), free(b);
353 /* A script is not an executable of this target. */
354 p = catalog(&e, "script", m);
355 d = mesh_fmt("%s/script", e.root);
356 b = mesh_native_binary(d, "nt");
357 write_file(b, "#!/bin/sh\necho hi\n");
358 r = prepare_err(&e, p, NULL);
359 CHECK(strstr(r, "not an executable for") != NULL);
360 free(r), free(d), free(b);
361 /* No source.hash: no identity. */
362 p = catalog(&e, "nohash", m);
363 d = mesh_fmt("%s/nohash/source.hash", e.root);
364 unlink(d);
365 r = prepare_err(&e, p, NULL);
366 CHECK(strncmp(r, "manifest: nt:", 13) == 0);
367 free(r), free(d);
368 /* A peer's hash must match. */
369 uint64_t one = 1;
370 r = prepare_err(&e, catalog(&e, "hash", m), &one);
371 CHECK(strncmp(r, "peer:", 5) == 0);
372 free(r);
373 /* abi. */
374 char *m5 = mesh_fmt("name = \"nt\"\nabi = 5\nbackends = [\"native\"]\nview_bytes = 0x1000\nnodes = 2\n");
375 r = prepare_err(&e, catalog(&e, "abi", m5), NULL);
376 CHECK(strncmp(r, "abi:", 4) == 0);
377 free(r), free(m5), free(m);
378 env_end(&e);
379}
380
381static uint64_t prepared_hash(env *e, mesh_payload p, const uint64_t *expect, bool *ok) {
382 mesh_prepared pr;
383 char *err = NULL;
384 *ok = e->comp->ops->prepare(e->comp, &p, expect, e->root, &pr, &err);
385 free(err);
386 if (!*ok) return 0;
387 uint64_t h = pr.hash;
388 mesh_prepared_free(&pr);
389 return h;
390}
391
392static void identity_is_the_manifest_and_source(void) {
393 env e;
394 env_new(&e, "ident");
395 char *m = manifest(""), *mp = manifest("param = 1");
396 bool ok;
397 uint64_t a = prepared_hash(&e, catalog(&e, "a", m), NULL, &ok);
398 uint64_t b = prepared_hash(&e, catalog(&e, "b", m), NULL, &ok);
399 uint64_t c = prepared_hash(&e, catalog(&e, "c", mp), NULL, &ok);
400 CHECK(a == b);
401 CHECK(a != c);
402 prepared_hash(&e, catalog(&e, "d", m), &a, &ok);
403 CHECK(ok);
404 free(m);
405 free(mp);
406 env_end(&e);
407}
408
409static void memory_cap_kills(void) {
410 mesh_native_limits l = {0, 64u << 20, 0, false};
411 char *err = NULL;
412 if (!mesh_native_limits_available(&l, &err)) {
413 printf("memory_cap_kills: skipped (%s)\n", err);
414 free(err);
415 return;
416 }
417 env e;
418 env_new(&e, "mem");
419 char *m = manifest("mem_mb = 64");
420 mesh_kernel *k = prepare(&e, catalog(&e, "nt", m), 3, &err);
421 free(m);
422 if (!CHECK(k)) {
423 fprintf(stderr, "%s\n", err);
424 return;
425 }
426 launch(&e, k, 3, 0);
427 WAIT("online", 20000, mesh_mailbox_get(e.mb, mesh_prog_ctl(3) + MESH_P_ONLINE) == 2);
428 CHECK_U(mesh_native_worker_stats(W(k)).limits, MESH_LIMIT_MEM);
429 CHECK_U(ok_task(&e, k, 3, 0, 10, 16), 16);
430 char *r = err_task(&e, k, 3, 1, 10, 256);
431 CHECK(strstr(r, "signal 9") != NULL);
432 free(r);
433 mesh_kernel_drop(k);
434 env_end(&e);
435}
436
437static void cpu_cap_holds(void) {
438 mesh_native_limits l = {50, 0, 0, false};
439 char *err = NULL;
440 if (!mesh_native_limits_available(&l, &err)) {
441 printf("cpu_cap_holds: skipped (%s)\n", err);
442 free(err);
443 return;
444 }
445 env e;
446 env_new(&e, "cpu");
447 char *m = manifest("cpu = 0.5\nnodes = 1");
448 mesh_kernel *k = prepare(&e, catalog(&e, "nt", m), 4, &err);
449 free(m);
450 launch(&e, k, 4, 0);
451 WAIT("online", 20000, mesh_mailbox_get(e.mb, mesh_prog_ctl(4) + MESH_P_ONLINE) == 1);
452 pid_t pid = (pid_t)ok_task(&e, k, 4, 0, 5, 0);
453 uint64_t c0, c1, r;
454 mesh_proc_usage(pid, &c0, &r);
455 uint64_t t0 = now_ms();
456 ok_task(&e, k, 4, 1, 12, 1000);
457 mesh_proc_usage(pid, &c1, &r);
458 double share = (double)(c1 - c0) / ((double)(now_ms() - t0) * 1e6);
459 CHECK(share < 0.6);
460 CHECK_U(mesh_native_worker_stats(W(k)).limits, MESH_LIMIT_CPU);
461 mesh_kernel_drop(k);
462 env_end(&e);
463}
464
465int main(void) {
466 char self[PATH_MAX];
467 ssize_t n = readlink("/proc/self/exe", self, sizeof self - 1);
468 if (n <= 0) return 1;
469 self[n] = 0;
470 /* .sexc/out/<profile>/test/<pkg>/lifecycle -> .sexc/out/<profile>/bin/mesh-test-worker */
471 char *s = strstr(self, "/test/");
472 if (s) *s = 0;
473 fixture = mesh_fmt("%s/bin/mesh-test-worker", self);
474 if (access(fixture, X_OK) != 0) {
475 printf("skipped: %s is not built (run `sexc build @ajax/mesh-test-worker` first)\n", fixture);
476 return 0;
477 }
478 spawn_tasks_halt_relaunch_stop();
479 warm_then_launch_is_immediate();
480 crashes_fail_with_status_and_output();
481 refusals();
482 identity_is_the_manifest_and_source();
483 memory_cap_kills();
484 cpu_cap_holds();
485 mesh_backend_cleanup();
486 return mesh_check_done();
487}