==> tests/meshd.c <==1/* A whole CPU-only meshd (--backend wasm) in a child process, on a catalog holding the real hello
2 * program (programs/hello-wasm, built by sexc), driven over its control and HTTP ports: autoload,
3 * tasks, stop/start, reset, unload/load, peek, the views, --state, SIGINT and exit 0. Also: the
4 * CUDA backend without a GPU fails cleanly (exit 1), and a bad flag exits 2. */
5#define _XOPEN_SOURCE 700 /* sexc compiles tests without the package's cflags (reported) */
6#include "meshd.h"
7#include "mesh/check.h"
8#include "mesh/exec_wasm_testkit.h"
9#include "mesh/backend.h"
10
11#include <limits.h>
12
13#include <arpa/inet.h>
14#include <fcntl.h>
15#include <netinet/in.h>
16#include <signal.h>
17#include <stdio.h>
18#include <stdlib.h>
19#include <string.h>
20#include <sys/socket.h>
21#include <sys/stat.h>
22#include <sys/wait.h>
23#include <time.h>
24#include <unistd.h>
25
26static int free_port(void) {
27 int fd = socket(AF_INET, SOCK_STREAM, 0);
28 struct sockaddr_in sa = {0};
29 sa.sin_family = AF_INET;
30 sa.sin_addr.s_addr = htonl(INADDR_LOOPBACK);
31 bind(fd, (struct sockaddr *)&sa, sizeof sa);
32 socklen_t sl = sizeof sa;
33 getsockname(fd, (struct sockaddr *)&sa, &sl);
34 close(fd);
35 return ntohs(sa.sin_port);
36}
37
38static int conn(int port) {
39 for (int i = 0; i < 300; i++) {
40 int fd = socket(AF_INET, SOCK_STREAM, 0);
41 struct sockaddr_in sa = {0};
42 sa.sin_family = AF_INET;
43 sa.sin_port = htons((uint16_t)port);
44 sa.sin_addr.s_addr = htonl(INADDR_LOOPBACK);
45 if (connect(fd, (struct sockaddr *)&sa, sizeof sa) == 0) return fd;
46 close(fd);
47 struct timespec t = {0, 20000000};
48 nanosleep(&t, NULL);
49 }
50 return -1;
51}
52
53static int FD;
54static char *rline(void) {
55 static char b[4096];
56 size_t i = 0;
57 while (i < sizeof b - 1 && read(FD, b + i, 1) == 1) if (b[i++] == '\n') break;
58 b[i] = 0;
59 return b;
60}
61static char *ask(const char *req) {
62 if (write(FD, req, strlen(req)) < 0 || write(FD, "\n", 1) < 0) abort();
63 return rline();
64}
65static char *frame(const char *req, size_t *len) {
66 if (write(FD, req, strlen(req)) < 0 || write(FD, "\n", 1) < 0) abort();
67 uint8_t h[4];
68 for (size_t got = 0; got < 4;) { ssize_t k = read(FD, h + got, 4 - got); if (k <= 0) return NULL; got += (size_t)k; }
69 size_t n = (size_t)h[0] | (size_t)h[1] << 8 | (size_t)h[2] << 16 | (size_t)h[3] << 24;
70 char *b = malloc(n + 1);
71 for (size_t got = 0; got < n;) { ssize_t k = read(FD, b + got, n - got); if (k <= 0) break; got += (size_t)k; }
72 b[n] = 0;
73 if (len) *len = n;
74 return b;
75}
76
77static pid_t run(char **argv, int argc, const char *logpath) {
78 pid_t pid = fork();
79 if (pid == 0) {
80 int fd = open(logpath, O_WRONLY | O_CREAT | O_TRUNC, 0600);
81 dup2(fd, 2);
82 dup2(fd, 1);
83 meshd_main(argc, argv);
84 }
85 return pid;
86}
87
88static int wait_exit(pid_t pid) {
89 int st = 0;
90 waitpid(pid, &st, 0);
91 return WIFEXITED(st) ? WEXITSTATUS(st) : 128 + WTERMSIG(st);
92}
93
94int main(void) {
95 char *wasm = mesh_kit_artifact("ajax-mesh-prog-hello-wasm");
96 if (!wasm) {
97 fprintf(stderr, "meshd test: programs/hello-wasm is not built (run `sexc test` for the workspace); skipped\n");
98 return 0;
99 }
100 char root[128], cat[192], path[256], state[192], logs[192], logf[192];
101 snprintf(root, sizeof root, "/tmp/meshd-e2e-%d", (int)getpid());
102 snprintf(cat, sizeof cat, "%s/programs", root);
103 snprintf(state, sizeof state, "%s/state", root);
104 snprintf(logs, sizeof logs, "%s/logs", root);
105 snprintf(logf, sizeof logf, "%s/meshd.log", root);
106 mkdir(root, 0700);
107 mkdir(cat, 0700);
108 snprintf(path, sizeof path, "%s/hello", cat);
109 mkdir(path, 0700);
110 snprintf(path, sizeof path, "%s/hello/program.toml", cat);
111 FILE *f = fopen(path, "w");
112 fputs("# hello: task = arg * 3 + 7\nname = \"hello\"\nabi = 6\nview_bytes = 0x1000\nview_abi = 1\nsnap_bytes = 0x40\nnodes = 2\n"
113 "policy = \"on_demand\"\ncounters = [\"tasks\", \"inits\", \"\", \"\"]\nfields = [\"magic:u32:0x0\", \"last_arg:u32:0x4\", \"last_result:u32:0x8\"]\n"
114 "autostart = true\nbackends = [\"cuda\", \"metal\", \"wasm\"]\n",
115 f);
116 fclose(f);
117 snprintf(path, sizeof path, "%s/hello/hello.wasm", cat);
118 FILE *in = fopen(wasm, "rb"), *out = fopen(path, "wb");
119 char buf[65536];
120 size_t k;
121 while ((k = fread(buf, 1, sizeof buf, in)) > 0) fwrite(buf, 1, k, out);
122 fclose(in);
123 fclose(out);
124 free(wasm);
125
126 /* A native worker program (apps/mesh-test-worker): its binary for this target, a source hash, param 2. */
127 char self[PATH_MAX], fixture[PATH_MAX + 64];
128 ssize_t sn = readlink("/proc/self/exe", self, sizeof self - 1);
129 self[sn > 0 ? sn : 0] = 0;
130 char *tp = strstr(self, "/test/");
131 if (tp) *tp = 0;
132 snprintf(fixture, sizeof fixture, "%s/bin/mesh-test-worker", self);
133 bool native = access(fixture, X_OK) == 0;
134 if (native) {
135 snprintf(path, sizeof path, "%s/nt", cat);
136 mkdir(path, 0700);
137 snprintf(path, sizeof path, "%s/nt/%s", cat, mesh_native_target());
138 mkdir(path, 0700);
139 snprintf(path, sizeof path, "%s/nt/program.toml", cat);
140 f = fopen(path, "w");
141 fputs("name = \"nt\"\nabi = 6\nbackends = [\"native\"]\nview_bytes = 0x1000\nnodes = 2\nparam = 2\n", f);
142 fclose(f);
143 snprintf(path, sizeof path, "%s/nt/source.hash", cat);
144 f = fopen(path, "w");
145 fputs("0123456789abcdef\n", f);
146 fclose(f);
147 snprintf(path, sizeof path, "%s/nt/%s/nt", cat, mesh_native_target());
148 in = fopen(fixture, "rb");
149 out = fopen(path, "wb");
150 while ((k = fread(buf, 1, sizeof buf, in)) > 0) fwrite(buf, 1, k, out);
151 fclose(in);
152 fclose(out);
153 chmod(path, 0755);
154 } else fprintf(stderr, "meshd test: %s is not built; the native part is skipped\n", fixture);
155
156 /* No GPU here: the CUDA backend refuses to start (exit 1); a bad flag is exit 2. */
157 int cp = free_port(), hp = free_port();
158 char ctl[32], http[32];
159 snprintf(ctl, sizeof ctl, "127.0.0.1:%d", cp);
160 snprintf(http, sizeof http, "127.0.0.1:%d", hp);
161 char *cuda[] = {"meshd", "--backend", "cuda", "--nvjitlink", "/nonexistent", "--programs", cat, "--control", ctl, "--http", http, NULL};
162 CHECK_I(wait_exit(run(cuda, 11, logf)), 1);
163 char *badf[] = {"meshd", "--bogus", NULL};
164 CHECK_I(wait_exit(run(badf, 2, logf)), 2);
165
166 char *argv[] = {"meshd", "--backend", "wasm", "--programs", cat, "--autoload", "--state", state, "--control", ctl, "--http", http, "--logs", logs, NULL};
167 pid_t pid = run(argv, 14, logf);
168 FD = conn(cp);
169 CHECK(FD >= 0);
170 /* The autoloaded hello: running once its OP_INIT is done. */
171 char *r = NULL;
172 for (int i = 0; i < 100; i++) {
173 r = ask("task hello 0 5");
174 if (strcmp(r, "ok 22\n") == 0) break;
175 struct timespec t = {0, 50000000};
176 nanosleep(&t, NULL);
177 }
178 CHECK_S(r, "ok 22\n");
179 CHECK_S(ask("task hello 0 0x10"), "ok 55\n");
180 size_t n;
181 char *b = frame("peek hello 0 12", &n);
182 CHECK(n == 12 && memcmp(b, "HELO", 4) == 0 && (uint8_t)b[4] == 16 && (uint8_t)b[8] == 55);
183 free(b);
184 b = frame("programs", NULL);
185 CHECK(strncmp(b, "0 hello running 2 1 on_demand ", 30) == 0);
186 CHECK(strstr(b, " 0x100000+0x1000 kind wasm lanes ") != NULL);
187 CHECK(strstr(b, "\nfields hello magic:u32:0x0 last_arg:u32:0x4 last_result:u32:0x8\ncounters hello tasks inits - -\n") != NULL);
188 free(b);
189 r = ask("stop hello");
190 CHECK(strncmp(r, "ok ", 3) == 0 && strstr(r, " us\n"));
191 CHECK_S(ask("task hello 0 1"), "err hello is not running\n");
192 r = ask("start hello");
193 CHECK(strncmp(r, "ok ", 3) == 0 && strstr(r, " us\n"));
194 CHECK_S(ask("task hello 0 2"), "ok 13\n");
195 r = ask("reset hello");
196 CHECK(strncmp(r, "ok blink ", 9) == 0);
197 CHECK_S(ask("load hello"), "err name exists: hello is loaded\n");
198 CHECK_S(ask("unload hello"), "ok\n");
199 r = ask("load hello");
200 CHECK(strncmp(r, "ok slot 0 blink ", 16) == 0);
201 for (int i = 0; i < 100; i++) {
202 r = ask("task hello 0 9");
203 if (strcmp(r, "ok 34\n") == 0) break;
204 struct timespec t = {0, 20000000};
205 nanosleep(&t, NULL);
206 }
207 CHECK_S(r, "ok 34\n");
208 b = frame("catalog", NULL);
209 CHECK_S(b, native ? "hello\nnt\n" : "hello\n");
210 free(b);
211 if (native) {
212 /* A native worker beside the wasm lanes: its own process, its counts in `programs`, a crash FAILs only it. */
213 r = ask("load nt");
214 CHECK(strncmp(r, "ok slot 1 blink ", 16) == 0);
215 r = ask("start nt"); /* no autostart: loaded stopped */
216 CHECK(strncmp(r, "ok ", 3) == 0 && strstr(r, " us\n"));
217 for (int i = 0; i < 200; i++) {
218 r = ask("task nt 0 5");
219 if (strcmp(r, "ok 2022\n") == 0) break;
220 struct timespec t = {0, 20000000};
221 nanosleep(&t, NULL);
222 }
223 CHECK_S(r, "ok 2022\n");
224 b = frame("programs", NULL);
225 CHECK(strstr(b, "\n1 nt running 2 1 on_demand ") && strstr(b, " kind native pid ") && strstr(b, " spawns 1 crashes 0 exit 0/0 "));
226 free(b);
227 r = ask("task nt 9 0");
228 CHECK(strncmp(r, "err ", 4) == 0);
229 for (int i = 0; i < 200; i++) {
230 b = frame("programs", NULL);
231 bool failed = strstr(b, "\n1 nt failed ") != NULL;
232 free(b);
233 if (failed) break;
234 struct timespec t = {0, 20000000};
235 nanosleep(&t, NULL);
236 }
237 b = frame("programs", NULL);
238 CHECK(strstr(b, "\n1 nt failed ") && strstr(b, " crashes 1 "));
239 free(b);
240 CHECK_S(ask("task hello 0 1"), "ok 10\n");
241 CHECK_S(ask("unload nt"), "ok\n");
242 }
243 b = frame("view pipes", NULL);
244 CHECK_S(b, "pipes system=4294967295 inboxes=0\n");
245 free(b);
246 CHECK_S(ask("fleet"), "err not in a fleet (single meshd: --cluster / --peer)\n");
247 /* The jobs thread runs (a single meshd leads its own job table). */
248 b = frame("jobs", NULL);
249 CHECK(b && strncmp(b, "jobs system 0 leader 0 role leader ", 35) == 0);
250 free(b);
251 b = frame("http /status", NULL);
252 CHECK_S(b, "err no program \"http\" (`programs` lists them)\n");
253 free(b);
254 b = frame("snap", &n);
255 CHECK(n > 0x3000);
256 free(b);
257 /* The HTTP port answers the views itself. */
258 int h = conn(hp);
259 const char *get = "GET /fleet HTTP/1.0\r\n\r\n";
260 if (write(h, get, strlen(get)) < 0) abort();
261 char resp[1024] = "";
262 size_t got = 0;
263 ssize_t kk;
264 while (got < sizeof resp - 1 && (kk = read(h, resp + got, sizeof resp - 1 - got)) > 0) got += (size_t)kk;
265 resp[got] = 0;
266 close(h);
267 CHECK(strncmp(resp, "HTTP/1.0 200 OK\r\n", 17) == 0 && strstr(resp, "fleet system=4294967295 n=1 role=single"));
268 close(FD);
269 /* SIGINT: quiesce, halt, --state written, exit 0. */
270 kill(pid, SIGINT);
271 CHECK_I(wait_exit(pid), 0);
272 f = fopen(state, "r");
273 char line[256] = "", l2[256] = "";
274 if (f) {
275 if (!fgets(line, sizeof line, f) || !fgets(l2, sizeof l2, f)) l2[0] = 0;
276 fclose(f);
277 }
278 CHECK_S(l2, "hello hello 1\n");
279 f = fopen(logf, "r");
280 char all[65536];
281 size_t la = f ? fread(all, 1, sizeof all - 1, f) : 0;
282 all[la] = 0;
283 if (f) fclose(f);
284 CHECK(strstr(all, "INFO programs halted, exiting") && strstr(all, "INFO exit 0"));
285 if (mesh_check_failures) fprintf(stderr, "meshd's log:\n%s\n", all);
286 /* Leave nothing behind. */
287 char rm[300];
288 snprintf(rm, sizeof rm, "%s/hello/program.toml", cat);
289 unlink(rm);
290 snprintf(rm, sizeof rm, "%s/hello/hello.wasm", cat);
291 unlink(rm);
292 snprintf(rm, sizeof rm, "%s/hello", cat);
293 rmdir(rm);
294 if (native) {
295 snprintf(rm, sizeof rm, "%s/nt/%s/nt", cat, mesh_native_target());
296 unlink(rm);
297 snprintf(rm, sizeof rm, "%s/nt/%s", cat, mesh_native_target());
298 rmdir(rm);
299 snprintf(rm, sizeof rm, "%s/nt/program.toml", cat);
300 unlink(rm);
301 snprintf(rm, sizeof rm, "%s/nt/source.hash", cat);
302 unlink(rm);
303 snprintf(rm, sizeof rm, "%s/nt", cat);
304 rmdir(rm);
305 snprintf(rm, sizeof rm, "%s/nt.log", logs);
306 unlink(rm);
307 }
308 rmdir(cat);
309 unlink(state);
310 unlink(logf);
311 rmdir(logs);
312 rmdir(root);
313 return mesh_check_done();
314}