1/* driver.rs's tests (job_setup_in_order, reads_by_name), and the driver end to end over a fake CPU
2 * executor whose "nodes" are a thread playing mesh-rt's node loop on the mailbox: autoload, the blink,
3 * OP_INIT, on-demand tasks, stop and start, refusals, unload, --state, shutdown. */
4#define _XOPEN_SOURCE 700 /* sexc compiles tests without the package's cflags (reported) */
5#include "mesh/check.h"
6#include "mesh/driver.h"
7#include "mesh/log.h"
8#include "mesh/manifest.h"
9
10#include <pthread.h>
11#include <stdatomic.h>
12#include <stdio.h>
13#include <stdlib.h>
14#include <string.h>
15#include <sys/stat.h>
16#include <unistd.h>
17
18/* ---- a fake executor ---- */
19
20static mesh_mailbox MB;
21
22typedef struct {
23    pthread_t th;
24    bool started;
25    _Atomic bool running;
26    uint32_t slot, nodes;
27} fake_kernel;
28
29static bool k_zero(const mesh_kernel *k, uint32_t *o, char **e) { (void)k, (void)e; *o = 0; return true; }
30static bool k_one(const mesh_kernel *k, uint32_t *o, char **e) { (void)k, (void)e; *o = 1; return true; }
31static uint64_t k_scratch(const mesh_kernel *k) { (void)k; return 0; }
32static bool k_global(const mesh_kernel *k, const char *n, uint32_t *o, char **e) {
33    (void)k;
34    if (strcmp(n, MESH_RT_ABI_SYMBOL)) { *e = strdup("no such global"); return false; }
35    *o = MESH_RT_ABI;
36    return true;
37}
38static void k_drop(mesh_kernel *k) {
39    fake_kernel *f = k->self;
40    if (f->started) pthread_join(f->th, NULL);
41    free(f);
42}
43static const mesh_kernel_ops KOPS = {MESH_BACKEND_WASM, k_zero, k_zero, k_one, k_scratch, k_global, k_drop};
44
45/* The node loop: online, take tickets in order, run them (OP_INIT writes a magic; a task is arg * 3 + 7), halt on P_SHUTDOWN. */
46static void *nodes(void *arg) {
47    fake_kernel *f = arg;
48    size_t c = mesh_prog_ctl(f->slot);
49    mesh_mailbox_fetch_add(MB, c + MESH_P_ONLINE, f->nodes);
50    uint32_t view = mesh_mailbox_get(MB, mesh_prog_entry(f->slot) + MESH_PT_VIEW_OFF);
51    for (;;) {
52        if (mesh_mailbox_get(MB, c + MESH_P_SHUTDOWN)) break;
53        uint32_t t = mesh_mailbox_get(MB, c + MESH_P_RING_HEAD);
54        size_t e = c + MESH_P_RING + MESH_ENTRY_BYTES * (t % MESH_PROG_RING);
55        if (mesh_mailbox_get(MB, e + MESH_E_SEQ) != t + 1) { mesh_sleep_us(50); continue; }
56        uint32_t op = mesh_mailbox_get(MB, e + MESH_E_OP), a = mesh_mailbox_get(MB, e + MESH_E_ARG), k = mesh_mailbox_get(MB, e + MESH_E_SLOT);
57        uint32_t r = 0;
58        if (op == MESH_OP_INIT) mesh_mailbox_set(MB, view, 0x4f4c4548u);
59        else if (op == MESH_OP_TASK) r = a * 3 + 7;
60        else r = MESH_R_BAD_OP;
61        mesh_mailbox_set(MB, c + MESH_P_RING_HEAD, t + 1);
62        mesh_mailbox_set(MB, c + MESH_P_RESULTS + 4 * k, r);
63        mesh_mailbox_publish(MB, c + MESH_P_MARKS + 4 * k, t + 1);
64        mesh_mailbox_fetch_add(MB, c + MESH_P_DONE, 1);
65    }
66    mesh_mailbox_fetch_add(MB, c + MESH_P_HALTED, f->nodes);
67    atomic_store(&f->running, false);
68    return NULL;
69}
70
71static mesh_caps CAPS = {MESH_BACKEND_WASM, "fake", 0, 0, 1, 1, false, 0, 0, 0, 0, "fake"};
72static const mesh_caps *b_caps(const mesh_backend *b) { (void)b; return &CAPS; }
73static bool b_true(const mesh_backend *b) { (void)b; return true; }
74static bool b_false(const mesh_backend *b) { (void)b; return false; }
75static uint32_t b_iso(const mesh_backend *b) { (void)b; return MESH_ISOLATION_LEASE; }
76static bool b_mailbox(const mesh_backend *b, uint8_t **h, uint64_t *dev, char **e) { (void)b, (void)e; *h = MB.base; *dev = 0; return true; }
77static bool b_current(const mesh_backend *b, char **e) { (void)b, (void)e; return true; }
78
79static bool c_needs_ptx(const mesh_compiler *c) { (void)c; return false; }
80static bool c_prepare(const mesh_compiler *c, mesh_payload *src, const uint64_t *expect, const char *root, mesh_prepared *out, char **err) {
81    (void)c, (void)expect, (void)root;
82    memset(out, 0, sizeof *out);
83    if (!mesh_check_manifest(src, MESH_BACKEND_WASM, &out->manifest, err)) { mesh_payload_free(src); return false; }
84    out->hash = 0x1234;
85    out->code_len = strlen(src->manifest);
86    out->code = (uint8_t *)strdup(src->manifest);
87    out->src = *src;
88    memset(src, 0, sizeof *src);
89    out->exec = MESH_BACKEND_WASM;
90    out->kernel = calloc(1, sizeof *out->kernel);
91    out->kernel->ops = &KOPS;
92    out->kernel->self = calloc(1, sizeof(fake_kernel));
93    return true;
94}
95static const mesh_compiler_ops COPS = {c_needs_ptx, c_prepare};
96static const mesh_compiler COMP = {&COPS, NULL};
97static const mesh_compiler *b_compiler(const mesh_backend *b) { (void)b; return &COMP; }
98
99static bool b_load(const mesh_backend *b, mesh_prepared *p, mesh_kernel **out, char **e) {
100    (void)b, (void)e;
101    *out = p->kernel;
102    p->kernel = NULL;
103    return true;
104}
105static bool b_stream(const mesh_backend *b, uint32_t slot, const mesh_kernel *k, mesh_stream **out, char **e) {
106    (void)b, (void)k, (void)e;
107    *out = calloc(1, sizeof **out);
108    (*out)->exec = MESH_BACKEND_WASM;
109    (*out)->slot = slot;
110    return true;
111}
112static bool b_launch(const mesh_backend *b, const mesh_stream *st, const mesh_kernel *k, uint32_t n, uint64_t dev, uint32_t slot, char **e) {
113    (void)b, (void)st, (void)dev, (void)e;
114    fake_kernel *f = k->self;
115    if (f->started) pthread_join(f->th, NULL);
116    f->slot = slot;
117    f->nodes = n;
118    atomic_store(&f->running, true);
119    f->started = true;
120    pthread_create(&f->th, NULL, nodes, f);
121    return true;
122}
123static bool b_query(const mesh_backend *b, const mesh_stream *st, const mesh_kernel *k, bool *done, char **e) {
124    (void)b, (void)st, (void)e;
125    *done = !atomic_load(&((fake_kernel *)k->self)->running);
126    return true;
127}
128static bool b_unload(const mesh_backend *b, mesh_kernel *k, char **e) {
129    (void)b, (void)e;
130    mesh_kernel_drop(k);
131    return true;
132}
133static bool b_stack(const mesh_backend *b, size_t *o, char **e) { (void)b, (void)e; *o = 0; return true; }
134static bool b_set_stack(const mesh_backend *b, size_t n, char **e) { (void)b, (void)n, (void)e; return true; }
135static bool b_scope(const mesh_backend *b, bool has, uint32_t slot, uint32_t *scope) { (void)b; *scope = has ? slot : MESH_NO_SLOT; return true; }
136
137static void b_destroy(mesh_backend *b) { free(b); }
138static const mesh_backend_ops BOPS = {b_caps, b_true, b_false, b_iso, b_mailbox, b_current, b_compiler, b_load, b_stream, b_launch, b_query,
139                                       b_unload, b_stack, b_set_stack, b_scope, NULL, NULL, NULL, NULL, NULL, b_destroy};
140
141/* ---- helpers ---- */
142
143static mesh_shared *SH;
144
145static bool ask(mesh_cmd_kind k, const char *name, bool on, char **text) {
146    mesh_cmd c = mesh_cmd_new(k);
147    if (name) c.name = strdup(name);
148    c.on = on;
149    return mesh_shared_ask(SH, c, 10000, text);
150}
151
152static bool load(const char *dir, char **text) {
153    mesh_cmd c = mesh_cmd_new(MESH_CMD_LOAD);
154    c.has_src = true;
155    c.src = mesh_src_catalog(dir);
156    return mesh_shared_ask(SH, c, 10000, text);
157}
158
159static int task(const char *name, uint32_t op, uint32_t arg, uint32_t *v, char **text) {
160    mesh_reply *r = mesh_reply_new();
161    mesh_cmd c = mesh_cmd_new(MESH_CMD_TASK);
162    c.name = strdup(name);
163    c.op = op;
164    c.arg = arg;
165    c.reply = mesh_reply_sender(r);
166    mesh_chan_send(SH->cmd, &c);
167    mesh_result x;
168    int w = mesh_reply_wait(r, 10000, &x);
169    mesh_reply_release(r);
170    if (w != MESH_CHAN_OK) return -1;
171    *v = x.value;
172    *text = x.text ? strdup(x.text) : NULL;
173    int ok = x.ok;
174    mesh_result_free(&x);
175    return ok;
176}
177
178static void *run(void *arg) {
179    *(int *)((void **)arg)[0] = mesh_driver_run((mesh_driver *)((void **)arg)[1]);
180    return NULL;
181}
182
183static void write_file(const char *path, const char *text) {
184    FILE *f = fopen(path, "w");
185    fputs(text, f);
186    fclose(f);
187}
188
189int main(void) {
190    /* ---- the Rust tests ---- */
191    mesh_setup_state j;
192    memset(&j, 0, sizeof j);
193    const mj_setup setup[2] = {{5, 0}, {4, 150}};
194    mesh_setup_step st = mesh_setup_state_step(&j, 7, setup, 2);
195    CHECK(st.k == MESH_SETUP_PUBLISH && st.op == 5 && st.arg == 0);
196    mesh_setup_state_published(&j);
197    CHECK(mesh_setup_state_step(&j, 7, setup, 2).k == MESH_SETUP_WAIT);
198    CHECK(mesh_setup_state_step(&j, 7, setup, 2).k == MESH_SETUP_WAIT);
199    mesh_setup_state_completed(&j, 7, NULL);
200    st = mesh_setup_state_step(&j, 7, setup, 2);
201    CHECK(st.k == MESH_SETUP_PUBLISH && st.op == 4 && st.arg == 150);
202    mesh_setup_state_published(&j);
203    CHECK(mesh_setup_state_step(&j, 7, setup, 2).k == MESH_SETUP_WAIT);
204    mesh_setup_state_completed(&j, 7, NULL);
205    CHECK(mesh_setup_state_step(&j, 7, setup, 2).k == MESH_SETUP_READY);
206    CHECK(mesh_setup_state_step(&j, 7, setup, 2).k == MESH_SETUP_READY);
207    mesh_setup_state k;
208    memset(&k, 0, sizeof k);
209    CHECK(mesh_setup_state_step(&k, 1, setup, 2).k == MESH_SETUP_PUBLISH);
210    CHECK(mesh_setup_state_step(&k, 1, setup, 2).k == MESH_SETUP_PUBLISH);
211    const mj_setup s8[1] = {{6, 32}};
212    st = mesh_setup_state_step(&j, 8, s8, 1);
213    CHECK(st.k == MESH_SETUP_PUBLISH && st.op == 6);
214    mesh_setup_state_published(&j);
215    mesh_setup_state_completed(&j, 7, "old");
216    CHECK(mesh_setup_state_step(&j, 8, s8, 1).k == MESH_SETUP_WAIT);
217    mesh_setup_state_completed(&j, 8, NULL);
218    CHECK(mesh_setup_state_step(&j, 8, s8, 1).k == MESH_SETUP_READY);
219    mesh_setup_state f;
220    memset(&f, 0, sizeof f);
221    CHECK(mesh_setup_state_step(&f, 2, setup, 2).k == MESH_SETUP_PUBLISH);
222    mesh_setup_state_published(&f);
223    mesh_setup_state_completed(&f, 2, "bad op");
224    st = mesh_setup_state_step(&f, 2, setup, 2);
225    CHECK(st.k == MESH_SETUP_FAILED && strcmp(st.err, "bad op") == 0);
226    mesh_setup_state_reset(&j);
227    CHECK(mesh_setup_state_step(&j, 8, s8, 1).k == MESH_SETUP_PUBLISH);
228    mesh_setup_state z;
229    memset(&z, 0, sizeof z);
230    CHECK(mesh_setup_state_step(&z, 3, NULL, 0).k == MESH_SETUP_READY);
231    mesh_setup_state_reset(&j);
232    mesh_setup_state_reset(&k);
233    mesh_setup_state_reset(&f);
234    mesh_setup_state_reset(&z);
235
236    mesh_manifest m;
237    char *err = NULL;
238    const char *t1 = "name = \"infer\"\nabi = 6\nreads = [\"train\", \"doom\"]";
239    CHECK(mesh_manifest_parse(t1, strlen(t1), &m, &err));
240    const char *names[4] = {"train", NULL, "hello", "doom"};
241    CHECK_U(mesh_reads_mask(&m, names, 4), 9);
242    mesh_manifest_free(&m);
243    const char *t2 = "name = \"http\"\nabi = 6\nreads = [\"*\"]";
244    CHECK(mesh_manifest_parse(t2, strlen(t2), &m, &err));
245    CHECK_U(mesh_reads_mask(&m, names, 4), MESH_READS_ALL);
246    mesh_manifest_free(&m);
247
248    /* ---- end to end ---- */
249    char root[64];
250    snprintf(root, sizeof root, "/tmp/mesh-driver-%d", (int)getpid());
251    char path[256];
252    mkdir(root, 0700);
253    snprintf(path, sizeof path, "%s/hello", root);
254    mkdir(path, 0700);
255    snprintf(path, sizeof path, "%s/hello/program.toml", root);
256    write_file(path, "name = \"hello\"\nabi = 6\nview_bytes = 0x1000\nsnap_bytes = 0x40\nnodes = 2\npolicy = \"on_demand\"\nautostart = true\nbackends = [\"wasm\"]\n");
257    snprintf(path, sizeof path, "%s/gpu", root);
258    mkdir(path, 0700);
259    snprintf(path, sizeof path, "%s/gpu/program.toml", root);
260    write_file(path, "name = \"gpu\"\nabi = 6\nbackends = [\"cuda\"]\nautostart = true\n");
261    char state[256];
262    snprintf(state, sizeof state, "%s/state", root);
263    mesh_log_set_fd(-1);
264
265    MB = mesh_mailbox_test();
266    static _Atomic bool quit;
267    SH = mesh_shared_new(MB, root, false, 0, 1, &quit);
268    mesh_backend *gpu = calloc(1, sizeof *gpu);
269    gpu->ops = &BOPS;
270    mesh_driver_config cfg = {false, state, true, false};
271    mesh_driver *d = mesh_driver_new(SH, gpu, NULL, 0, cfg);
272    CHECK(mesh_driver_start(d, &err));
273    void *arg[2] = {0, d};
274    int code = -1;
275    arg[0] = &code;
276    pthread_t th;
277    pthread_create(&th, NULL, run, arg);
278    /* The autoload: hello loads and starts (OP_INIT first); gpu is not loaded here (unsupported). */
279    for (int i = 0; i < 2000 && !atomic_load(&SH->ready); i++) mesh_sleep_us(1000);
280    CHECK(atomic_load(&SH->ready));
281    for (int i = 0; i < 2000 && mesh_mailbox_get(MB, mesh_prog_entry(0) + MESH_PT_STATE) != MESH_ST_RUNNING; i++) mesh_sleep_us(1000);
282    CHECK_U(mesh_mailbox_get(MB, mesh_prog_entry(0) + MESH_PT_STATE), MESH_ST_RUNNING);
283    CHECK_U(mesh_mailbox_get(MB, MESH_H_MAGIC), MESH_MAGIC);
284    uint32_t v = 0;
285    char *text = NULL;
286    CHECK_I(task("hello", MESH_OP_TASK, 5, &v, &text), 1);
287    CHECK_U(v, 22);
288    free(text);
289    mesh_sleep_us(5000);
290    CHECK_U(mesh_mailbox_get(MB, mesh_mailbox_get(MB, mesh_prog_entry(0) + MESH_PT_VIEW_OFF)), 0x4f4c4548u);
291    /* Refusals. */
292    CHECK(!load("hello", &text));
293    CHECK_S(text, "name exists: hello is loaded");
294    free(text);
295    CHECK(!load("../etc", &text));
296    CHECK(strncmp(text, "manifest: ", 10) == 0);
297    free(text);
298    CHECK(!load("gpu", &text));
299    CHECK_S(text, "unsupported: gpu runs on cuda (manifest `backends`); this meshd runs wasm");
300    free(text);
301    CHECK(!ask(MESH_CMD_ROLLBACK, "hello", false, &text));
302    CHECK_S(text, "hello: no previous image to roll back to");
303    free(text);
304    CHECK(!ask(MESH_CMD_PROG_RUN, "nope", true, &text));
305    CHECK_S(text, "no program \"nope\" (`programs` lists them)");
306    free(text);
307    CHECK_I(task("nope", 0, 0, &v, &text), 0);
308    CHECK_S(text, "no program \"nope\" (`programs` lists them)");
309    free(text);
310    /* Stop and start: zero gap for the others, answered once halted / online. */
311    CHECK(ask(MESH_CMD_PROG_RUN, "hello", false, &text));
312    CHECK(strstr(text, " us") != NULL);
313    free(text);
314    CHECK_U(mesh_mailbox_get(MB, mesh_prog_entry(0) + MESH_PT_STATE), MESH_ST_LOADED);
315    CHECK_I(task("hello", MESH_OP_TASK, 1, &v, &text), 0);
316    CHECK_S(text, "hello is not running");
317    free(text);
318    CHECK(ask(MESH_CMD_PROG_RUN, "hello", true, &text));
319    free(text);
320    CHECK_I(task("hello", MESH_OP_TASK, 10, &v, &text), 1);
321    CHECK_U(v, 37);
322    free(text);
323    /* A bad op: the program's answer. */
324    CHECK_I(task("hello", 9, 0, &v, &text), 1);
325    CHECK_U(v, MESH_R_BAD_OP);
326    free(text);
327    /* Reset while running: a blink of its slot; the view zeroed, OP_INIT again. */
328    CHECK(ask(MESH_CMD_RESET, "hello", false, &text));
329    CHECK(strncmp(text, "blink ", 6) == 0);
330    free(text);
331    /* --state records it. */
332    mesh_sleep_us(20000);
333    FILE *sf = fopen(state, "r");
334    char line[256] = "";
335    if (sf) {
336        char *got = fgets(line, sizeof line, sf);
337        got = fgets(line, sizeof line, sf);
338        (void)got;
339        fclose(sf);
340    }
341    CHECK_S(line, "hello hello 1\n");
342    /* Unload. */
343    CHECK(ask(MESH_CMD_UNLOAD, "hello", false, &text));
344    CHECK_S(text, "");
345    free(text);
346    CHECK_U(mesh_mailbox_get(MB, mesh_prog_entry(0) + MESH_PT_STATE), MESH_ST_EMPTY);
347    /* Loaded again into the freed slot (the blink frees the old view). */
348    CHECK(load("hello", &text));
349    CHECK(strncmp(text, "slot 0 blink ", 13) == 0);
350    free(text);
351    atomic_store(&quit, true);
352    pthread_join(th, NULL);
353    CHECK_I(code, 0);
354    mesh_driver_free(d);
355    mesh_shared_free(SH);
356    free(MB.base);
357    char cmd[300];
358    snprintf(cmd, sizeof cmd, "%s/hello/program.toml", root);
359    unlink(cmd);
360    snprintf(cmd, sizeof cmd, "%s/gpu/program.toml", root);
361    unlink(cmd);
362    unlink(state);
363    snprintf(cmd, sizeof cmd, "%s/hello", root);
364    rmdir(cmd);
365    snprintf(cmd, sizeof cmd, "%s/gpu", root);
366    rmdir(cmd);
367    rmdir(root);
368    return mesh_check_done();
369}