1/* Placement in the driver (driver.rs's placement section) over a fake CPU executor: a peered driver,
2 * member 0 of 2, with a placeable program. Its placer exists from the load; `place` reports and asks for
3 * a change (as the leader: pin here, done at once with the other member down; as a follower: refused,
4 * the leader coordinates); the leader's PL_SET pins it on the other member (its state written into the
5 * view, the kernel stopped here, start and reset refused with the Rust's reason, input forwarded with
6 * F_INPUT, PL_DONE answered, the inventory still counting it run) and back to local (it runs again);
7 * the fleet's view reaches the placer; unload drops it. */
8#define _XOPEN_SOURCE 700 /* sexc compiles tests without the package's cflags (reported) */
9#include "mesh/check.h"
10#include "mesh/driver.h"
11#include "mesh/log.h"
12#include "mesh/manifest.h"
13#include "mesh/place_wire.h"
14
15#include <pthread.h>
16#include <stdatomic.h>
17#include <stdio.h>
18#include <stdlib.h>
19#include <string.h>
20#include <sys/stat.h>
21#include <unistd.h>
22
23/* ---- a fake executor ---- */
24
25static mesh_mailbox MB;
26
27typedef struct {
28    pthread_t th;
29    bool started;
30    _Atomic bool running;
31    uint32_t slot, nodes;
32} fake_kernel;
33
34static bool k_zero(const mesh_kernel *k, uint32_t *o, char **e) { (void)k, (void)e; *o = 0; return true; }
35static bool k_one(const mesh_kernel *k, uint32_t *o, char **e) { (void)k, (void)e; *o = 1; return true; }
36static uint64_t k_scratch(const mesh_kernel *k) { (void)k; return 0; }
37static bool k_global(const mesh_kernel *k, const char *n, uint32_t *o, char **e) {
38    (void)k;
39    if (strcmp(n, MESH_RT_ABI_SYMBOL)) { *e = strdup("no such global"); return false; }
40    *o = MESH_RT_ABI;
41    return true;
42}
43static void k_drop(mesh_kernel *k) {
44    fake_kernel *f = k->self;
45    if (f->started) pthread_join(f->th, NULL);
46    free(f);
47}
48static const mesh_kernel_ops KOPS = {MESH_BACKEND_WASM, k_zero, k_zero, k_one, k_scratch, k_global, k_drop};
49
50/* 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. */
51static void *nodes(void *arg) {
52    fake_kernel *f = arg;
53    size_t c = mesh_prog_ctl(f->slot);
54    mesh_mailbox_fetch_add(MB, c + MESH_P_ONLINE, f->nodes);
55    uint32_t view = mesh_mailbox_get(MB, mesh_prog_entry(f->slot) + MESH_PT_VIEW_OFF);
56    for (;;) {
57        if (mesh_mailbox_get(MB, c + MESH_P_SHUTDOWN)) break;
58        uint32_t t = mesh_mailbox_get(MB, c + MESH_P_RING_HEAD);
59        size_t e = c + MESH_P_RING + MESH_ENTRY_BYTES * (t % MESH_PROG_RING);
60        if (mesh_mailbox_get(MB, e + MESH_E_SEQ) != t + 1) { mesh_sleep_us(50); continue; }
61        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);
62        uint32_t r = 0;
63        if (op == MESH_OP_INIT) mesh_mailbox_set(MB, view, 0x4f4c4548u);
64        else if (op == MESH_OP_TASK) r = a * 3 + 7;
65        else r = MESH_R_BAD_OP;
66        mesh_mailbox_set(MB, c + MESH_P_RING_HEAD, t + 1);
67        mesh_mailbox_set(MB, c + MESH_P_RESULTS + 4 * k, r);
68        mesh_mailbox_publish(MB, c + MESH_P_MARKS + 4 * k, t + 1);
69        mesh_mailbox_fetch_add(MB, c + MESH_P_DONE, 1);
70    }
71    mesh_mailbox_fetch_add(MB, c + MESH_P_HALTED, f->nodes);
72    atomic_store(&f->running, false);
73    return NULL;
74}
75
76static mesh_caps CAPS = {MESH_BACKEND_WASM, "fake", 0, 0, 1, 1, false, 0, 0, 0, 0, "fake"};
77static const mesh_caps *b_caps(const mesh_backend *b) { (void)b; return &CAPS; }
78static bool b_true(const mesh_backend *b) { (void)b; return true; }
79static bool b_false(const mesh_backend *b) { (void)b; return false; }
80static uint32_t b_iso(const mesh_backend *b) { (void)b; return MESH_ISOLATION_LEASE; }
81static 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; }
82static bool b_current(const mesh_backend *b, char **e) { (void)b, (void)e; return true; }
83
84static bool c_needs_ptx(const mesh_compiler *c) { (void)c; return false; }
85static bool c_prepare(const mesh_compiler *c, mesh_payload *src, const uint64_t *expect, const char *root, mesh_prepared *out, char **err) {
86    (void)c, (void)expect, (void)root;
87    memset(out, 0, sizeof *out);
88    if (!mesh_check_manifest(src, MESH_BACKEND_WASM, &out->manifest, err)) { mesh_payload_free(src); return false; }
89    out->hash = 0x1234;
90    out->code_len = strlen(src->manifest);
91    out->code = (uint8_t *)strdup(src->manifest);
92    out->src = *src;
93    memset(src, 0, sizeof *src);
94    out->exec = MESH_BACKEND_WASM;
95    out->kernel = calloc(1, sizeof *out->kernel);
96    out->kernel->ops = &KOPS;
97    out->kernel->self = calloc(1, sizeof(fake_kernel));
98    return true;
99}
100static const mesh_compiler_ops COPS = {c_needs_ptx, c_prepare};
101static const mesh_compiler COMP = {&COPS, NULL};
102static const mesh_compiler *b_compiler(const mesh_backend *b) { (void)b; return &COMP; }
103
104static bool b_load(const mesh_backend *b, mesh_prepared *p, mesh_kernel **out, char **e) {
105    (void)b, (void)e;
106    *out = p->kernel;
107    p->kernel = NULL;
108    return true;
109}
110static bool b_stream(const mesh_backend *b, uint32_t slot, const mesh_kernel *k, mesh_stream **out, char **e) {
111    (void)b, (void)k, (void)e;
112    *out = calloc(1, sizeof **out);
113    (*out)->exec = MESH_BACKEND_WASM;
114    (*out)->slot = slot;
115    return true;
116}
117static 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) {
118    (void)b, (void)st, (void)dev, (void)e;
119    fake_kernel *f = k->self;
120    if (f->started) pthread_join(f->th, NULL);
121    f->slot = slot;
122    f->nodes = n;
123    atomic_store(&f->running, true);
124    f->started = true;
125    pthread_create(&f->th, NULL, nodes, f);
126    return true;
127}
128static bool b_query(const mesh_backend *b, const mesh_stream *st, const mesh_kernel *k, bool *done, char **e) {
129    (void)b, (void)st, (void)e;
130    *done = !atomic_load(&((fake_kernel *)k->self)->running);
131    return true;
132}
133static bool b_unload(const mesh_backend *b, mesh_kernel *k, char **e) {
134    (void)b, (void)e;
135    mesh_kernel_drop(k);
136    return true;
137}
138static bool b_stack(const mesh_backend *b, size_t *o, char **e) { (void)b, (void)e; *o = 0; return true; }
139static bool b_set_stack(const mesh_backend *b, size_t n, char **e) { (void)b, (void)n, (void)e; return true; }
140static 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; }
141
142static void b_destroy(mesh_backend *b) { free(b); }
143static 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,
144                                       b_unload, b_stack, b_set_stack, b_scope, NULL, NULL, NULL, NULL, NULL, b_destroy};
145
146/* ---- helpers ---- */
147
148static mesh_shared *SH;
149
150static bool ask(mesh_cmd_kind k, const char *name, bool on, char **text) {
151    mesh_cmd c = mesh_cmd_new(k);
152    if (name) c.name = strdup(name);
153    c.on = on;
154    return mesh_shared_ask(SH, c, 10000, text);
155}
156
157static bool load(const char *dir, char **text) {
158    mesh_cmd c = mesh_cmd_new(MESH_CMD_LOAD);
159    c.has_src = true;
160    c.src = mesh_src_catalog(dir);
161    return mesh_shared_ask(SH, c, 10000, text);
162}
163
164
165static bool place(const char *name, const char *words, char **text) {
166    mesh_cmd c = mesh_cmd_new(MESH_CMD_PLACE);
167    c.name = strdup(name);
168    if (words) {
169        c.has_words = true;
170        c.words = calloc(8, sizeof *c.words);
171        char *w = strdup(words), *save = NULL;
172        for (char *t = strtok_r(w, " ", &save); t; t = strtok_r(NULL, " ", &save)) c.words[c.nwords++] = strdup(t);
173        free(w);
174    }
175    return mesh_shared_ask(SH, c, 10000, text);
176}
177
178static void fleet_view(uint32_t up, bool has_leader, uint32_t leader) {
179    mesh_cmd c = mesh_cmd_new(MESH_CMD_FLEET_VIEW);
180    c.up = up;
181    c.has_leader = has_leader;
182    c.leader = leader;
183    mesh_chan_send(SH->cmd, &c);
184}
185
186static void place_free(void *m) {
187    mpw_place_free(m);
188    free(m);
189}
190
191/* F_PLACE from the leader (system 1): PL_SET mode, the members, the state from system `src`. */
192static void pl_set(mpm_mode mode, uint32_t id, uint32_t members, uint32_t src, size_t state_len) {
193    mpw_place *m = calloc(1, sizeof *m);
194    m->op = MESH_PL_SET;
195    m->id = id;
196    strcpy(m->name, "dm");
197    m->mode = mode;
198    m->aux = members | src << 8;
199    m->nbytes = state_len;
200    m->bytes = calloc(1, state_len ? state_len : 1);
201    for (size_t i = 0; i < state_len; i++) m->bytes[i] = (uint8_t)(i * 7 + 1);
202    mesh_cmd c = mesh_cmd_new(MESH_CMD_PEER_PLACE);
203    c.msg = m;
204    c.msg_free = place_free;
205    c.from = 1;
206    mesh_chan_send(SH->cmd, &c);
207}
208
209/* The next frame queued for system 1 (type byte), or -1 within 2 s. */
210static int next_frame(mesh_link_frame *out) {
211    for (int i = 0; i < 2000; i++) {
212        if (mesh_chan_try_recv(SH->link_out[1], out) == MESH_CHAN_OK) {
213            if (out->n == 0) continue;
214            return out->b[0];
215        }
216        mesh_sleep_us(1000);
217    }
218    return -1;
219}
220
221static void *run(void *arg) {
222    *(int *)((void **)arg)[0] = mesh_driver_run((mesh_driver *)((void **)arg)[1]);
223    return NULL;
224}
225
226static uint32_t p_run(void) { return mesh_mailbox_get(MB, mesh_prog_ctl(0) + MESH_P_RUN); }
227
228static bool wait_state(uint32_t st) {
229    for (int i = 0; i < 3000; i++) {
230        if (mesh_mailbox_get(MB, mesh_prog_entry(0) + MESH_PT_STATE) == st) return true;
231        mesh_sleep_us(1000);
232    }
233    return false;
234}
235
236int main(void) {
237    char root[64], path[256];
238    snprintf(root, sizeof root, "/tmp/mesh-driver-place-%d", (int)getpid());
239    mkdir(root, 0700);
240    snprintf(path, sizeof path, "%s/dm", root);
241    mkdir(path, 0700);
242    snprintf(path, sizeof path, "%s/dm/program.toml", root);
243    FILE *f = fopen(path, "w");
244    fputs("name = \"dm\"\nabi = 6\nview_bytes = 0x4000\nview_abi = 2\nnodes = 1\npolicy = \"periodic\"\nhz = 10\nin_flight = \"batch\"\nfanout = 4\ninput = 64\nautostart = true\n"
245          "backends = [\"wasm\"]\nplace = \"split\"\nplace_state = [\"0x0:0x24\", \"0x28:0x4\", \"0x100:0x400\"]\nplace_frame = 0x20\n"
246          "place_out = [\"0x1000:0x200\", \"0x1400:0x200\", \"0x1800:0x200\", \"0x1c00:0x200\"]\nplace_show = 0x2000\n",
247          f);
248    fclose(f);
249    mesh_log_set_fd(-1);
250    MB = mesh_mailbox_test();
251    static _Atomic bool quit;
252    SH = mesh_shared_new(MB, root, true, 0, 2, &quit);
253    SH->link_out[1] = mesh_link_queue_new(64);
254    SH->bulk_out[1] = mesh_link_queue_new(8);
255    mesh_backend *gpu = calloc(1, sizeof *gpu);
256    gpu->ops = &BOPS;
257    mesh_driver_config cfg = {true, NULL, false, false};
258    mesh_driver *d = mesh_driver_new(SH, gpu, NULL, 0, cfg);
259    char *err = NULL, *text = NULL;
260    CHECK(mesh_driver_start(d, &err));
261    void *arg[2] = {0, d};
262    int code = -1;
263    arg[0] = &code;
264    pthread_t th;
265    pthread_create(&th, NULL, run, arg);
266    if (!CHECK(load("dm", &text))) fprintf(stderr, "load dm: %s\n", text);
267    free(text);
268    CHECK(wait_state(MESH_ST_RUNNING));
269
270    /* Local at first; the report. */
271    CHECK(place("dm", NULL, &text));
272    CHECK(strncmp(text, "local role Local system 0 paused false shown 0 lat_n 0 lat_sum_us 0 lat_max_us 0 ignored 0 part_ns 0 uptime_ms ", 106) == 0);
273    free(text);
274    /* The leader with the other member down: pin here, at once. */
275    fleet_view(1, true, 0);
276    CHECK(place("dm", "pin 0", &text));
277    free(text);
278    CHECK(place("dm", NULL, &text));
279    CHECK(strncmp(text, "pin 0 role Runner system 0 ", 27) == 0);
280    free(text);
281    CHECK_U(mesh_mailbox_get(MB, mesh_prog_ext(0) + MESH_PX_PLACE), MESH_PLACE_PIN);
282    CHECK_U(p_run(), 1);
283    CHECK(!place("dm", "pin 7", &text));
284    CHECK(strstr(text, "7") != NULL);
285    free(text);
286
287    /* A follower now: the leader (system 1) coordinates. */
288    fleet_view(3, true, 1);
289    CHECK(!place("dm", "pin 0", &text));
290    CHECK_S(text, "peer: placement changes are coordinated by the leader, system 1");
291    free(text);
292    /* PL_SET pin 1 from the leader with its state: written here, the kernel stops, PL_DONE back. */
293    size_t st_len = 0x24 + 0x4 + 0x400;
294    pl_set(mpm_pin(1), 9, 3, 1, st_len);
295    mesh_link_frame fr;
296    int ty;
297    while ((ty = next_frame(&fr)) >= 0 && ty != MESH_F_PLACE) free(fr.b);
298    CHECK_I(ty, MESH_F_PLACE);
299    if (ty == MESH_F_PLACE) {
300        size_t rest = 0;
301        mpw_place got;
302        CHECK(mpw_place_rest_len(fr.b + 1, &rest, &err) && mpw_place_decode(fr.b + 1, fr.b + 1 + MPW_PLACE_HEAD, rest, &got, &err));
303        CHECK_U(got.op, MESH_PL_DONE);
304        CHECK_U(got.id, 9);
305        mpw_place_free(&got);
306        free(fr.b);
307    }
308    CHECK(wait_state(MESH_ST_LOADED));
309    CHECK_U(p_run(), 0);
310    uint32_t view = mesh_mailbox_get(MB, mesh_prog_entry(0) + MESH_PT_VIEW_OFF);
311    uint8_t b[4];
312    mesh_mailbox_read(MB, view + 0x100, b, 4);
313    CHECK(b[0] == (uint8_t)(0x28 * 7 + 1) && b[1] == (uint8_t)(0x29 * 7 + 1)); /* the third region's first bytes */
314    CHECK(!ask(MESH_CMD_PROG_RUN, "dm", true, &text));
315    CHECK_S(text, "dm is placed (pin 1): system 1 runs it; `place dm local` first, or change it there");
316    free(text);
317    CHECK(!ask(MESH_CMD_RESET, "dm", false, &text));
318    CHECK_S(text, "dm is placed (pin 1): system 1 runs it; `place dm local` first, or change it there");
319    free(text);
320    /* The inventory counts it run (its placement stopped it). */
321    mesh_reply *r = mesh_reply_new();
322    mesh_cmd inv = mesh_cmd_new(MESH_CMD_INVENTORY);
323    inv.reply = mesh_reply_sender(r);
324    mesh_chan_send(SH->cmd, &inv);
325    mesh_result x;
326    CHECK_I(mesh_reply_wait(r, 10000, &x), MESH_CHAN_OK);
327    mesh_reply_release(r);
328    const mesh_actuals *a = x.ptr;
329    CHECK(a && a->n == 1 && a->v[0].run);
330    mesh_result_free(&x);
331    /* Input goes to system 1. */
332    mesh_cmd in = mesh_cmd_new(MESH_CMD_INPUT);
333    in.name = strdup("dm");
334    in.bytes = (uint8_t *)strdup("wasd");
335    in.len = 4;
336    CHECK(mesh_shared_ask(SH, in, 10000, &text));
337    CHECK_S(text, "4 bytes to system 1");
338    free(text);
339    while ((ty = next_frame(&fr)) >= 0 && ty != MESH_F_INPUT) free(fr.b);
340    CHECK_I(ty, MESH_F_INPUT);
341    if (ty >= 0) free(fr.b);
342
343    /* Back to local from the leader: it runs here again. */
344    pl_set(mpm_local(), 10, 3, 1, 0);
345    CHECK(wait_state(MESH_ST_RUNNING));
346    CHECK_U(p_run(), 1);
347    CHECK(place("dm", NULL, &text));
348    CHECK(strncmp(text, "local role Local system 0 ", 26) == 0);
349    free(text);
350    CHECK(ask(MESH_CMD_PROG_RUN, "dm", false, &text));
351    free(text);
352
353    /* Unloaded: its placer goes with it. */
354    CHECK(ask(MESH_CMD_UNLOAD, "dm", false, &text));
355    free(text);
356    CHECK(!place("dm", NULL, &text));
357    CHECK_S(text, "no program \"dm\" (`programs` lists them)");
358    free(text);
359
360    atomic_store(&quit, true);
361    pthread_join(th, NULL);
362    CHECK_I(code, 0);
363    mesh_driver_free(d);
364    mesh_shared_free(SH);
365    free(MB.base);
366    unlink(path);
367    snprintf(path, sizeof path, "%s/dm", root);
368    rmdir(path);
369    rmdir(root);
370    return mesh_check_done();
371}