==> tests/control.c <==1/* The control port over TCP against a stand-in driver: every reply shape (lines, framed bodies, errors),
2 * the texts (programs, pipes, views, cluster, peek), req bodies, the line limit, and control.rs's
3 * an_idle_client_is_closed. */
4#define _XOPEN_SOURCE 700 /* sexc compiles tests without the package's cflags (reported) */
5#include "mesh/abi.h"
6#include "mesh/check.h"
7#include "mesh/control.h"
8#include "mesh/infer_request.h"
9#include "mesh/log.h"
10#include "mesh/manifest.h"
11#include "mesh/manifest_encode.h"
12#include "mesh/prog_table.h"
13
14#include <arpa/inet.h>
15#include <netinet/in.h>
16#include <pthread.h>
17#include <stdio.h>
18#include <stdlib.h>
19#include <string.h>
20#include <sys/socket.h>
21#include <unistd.h>
22
23static mesh_shared *SH;
24
25static void *driver(void *arg) {
26 (void)arg;
27 mesh_cmd c;
28 while (mesh_chan_recv(SH->cmd, &c, -1) == MESH_CHAN_OK) {
29 switch (c.k) {
30 case MESH_CMD_LOAD: mesh_reply_ok(c.reply, "slot 0 blink 12"); break;
31 case MESH_CMD_TASK: mesh_reply_value(c.reply, c.arg * 3 + 7); break;
32 case MESH_CMD_REQ: mesh_reply_bytes(c.reply, c.bytes, c.len); break;
33 case MESH_CMD_INPUT: mesh_reply_okf(c.reply, "%zu bytes", c.len); break;
34 case MESH_CMD_UNLOAD: mesh_reply_ok(c.reply, ""); break;
35 case MESH_CMD_PROG_RUN: mesh_reply_errf(c.reply, "no program \"%s\" (`programs` lists them)", c.name); break;
36 default: mesh_reply_err(c.reply, "unexpected"); break;
37 }
38 c.reply = NULL;
39 mesh_cmd_free(&c);
40 }
41 return NULL;
42}
43
44static int conn(int port) {
45 int fd = socket(AF_INET, SOCK_STREAM, 0);
46 struct sockaddr_in sa = {0};
47 sa.sin_family = AF_INET;
48 sa.sin_port = htons((uint16_t)port);
49 sa.sin_addr.s_addr = htonl(INADDR_LOOPBACK);
50 if (connect(fd, (struct sockaddr *)&sa, sizeof sa) != 0) return -1;
51 return fd;
52}
53
54static int FD;
55static void send_s(const char *s) {
56 size_t n = strlen(s);
57 if (write(FD, s, n) != (ssize_t)n) abort();
58}
59static bool read_n(int fd, void *b, size_t n) {
60 char *p = b;
61 while (n) {
62 ssize_t k = read(fd, p, n);
63 if (k <= 0) return false;
64 p += k;
65 n -= (size_t)k;
66 }
67 return true;
68}
69/* One reply line. */
70static char *rline(void) {
71 static char b[8192];
72 size_t i = 0;
73 while (i < sizeof b - 1 && read(FD, b + i, 1) == 1) if (b[i++] == '\n') break;
74 b[i] = 0;
75 return b;
76}
77/* One framed reply, as text. */
78static char *rframe(size_t *len) {
79 uint8_t h[4];
80 if (!read_n(FD, h, 4)) return NULL;
81 size_t n = (size_t)h[0] | (size_t)h[1] << 8 | (size_t)h[2] << 16 | (size_t)h[3] << 24;
82 char *b = malloc(n + 1);
83 read_n(FD, b, n);
84 b[n] = 0;
85 if (len) *len = n;
86 return b;
87}
88
89static int lfd;
90static void *serve(void *arg) {
91 (void)arg;
92 mesh_control_serve(SH, lfd);
93 return NULL;
94}
95
96int main(void) {
97 mesh_log_set_fd(-1);
98 static _Atomic bool quit;
99 mesh_mailbox mb = mesh_mailbox_test();
100 SH = mesh_shared_new(mb, "/nonexistent-catalog", false, 0, 1, &quit);
101 mesh_chan_close_rx(SH->jobs); /* no jobs thread here */
102 /* A loaded wasm program in slot 0. */
103 mesh_manifest m;
104 char *err = NULL;
105 const char *mt = "name = \"hello\"\nabi = 6\nview_bytes = 0x1000\nnodes = 2\ncounters = [\"tasks\"]\nbackends = [\"wasm\"]";
106 CHECK(mesh_manifest_parse(mt, strlen(mt), &m, &err));
107 mesh_image im = {0x100000, 0, 0x1234, 0, 0, MESH_RT_ABI, 1};
108 uint8_t e[MESH_PT_BYTES], x[MESH_PX_BYTES];
109 mesh_manifest_encode_entry(&m, &im, e);
110 mesh_manifest_encode_ext(&m, x);
111 x[MESH_PX_EXEC] = MESH_BACKEND_WASM;
112 mesh_pt_write_entry(mb, 0, e, x, MESH_ST_RUNNING);
113 mesh_mailbox_set(mb, 0x100000, 0x4f4c4548u);
114 mesh_manifest_free(&m);
115 pthread_t dt, st;
116 pthread_create(&dt, NULL, driver, NULL);
117 lfd = socket(AF_INET, SOCK_STREAM, 0);
118 struct sockaddr_in sa = {0};
119 sa.sin_family = AF_INET;
120 sa.sin_addr.s_addr = htonl(INADDR_LOOPBACK);
121 bind(lfd, (struct sockaddr *)&sa, sizeof sa);
122 socklen_t sl = sizeof sa;
123 getsockname(lfd, (struct sockaddr *)&sa, &sl);
124 listen(lfd, 8);
125 int port = ntohs(sa.sin_port);
126 pthread_create(&st, NULL, serve, NULL);
127 FD = conn(port);
128 CHECK(FD >= 0);
129
130 send_s("load hello\n");
131 CHECK_S(rline(), "ok slot 0 blink 12\n");
132 send_s("task hello 5 0x10\n");
133 CHECK_S(rline(), "ok 55\n");
134 send_s("task x\n");
135 CHECK_S(rline(), "err usage: task <name> <op> <arg>\n");
136 send_s(" unload hello \r\n");
137 CHECK_S(rline(), "ok\n");
138 send_s("start nope\n");
139 CHECK_S(rline(), "err no program \"nope\" (`programs` lists them)\n");
140 send_s("start a b\n");
141 CHECK_S(rline(), "err unknown or malformed request \"start a b\"\n");
142 send_s("frobnicate\n");
143 CHECK_S(rline(), "err unknown or malformed request \"frobnicate\"\n");
144 send_s("\n");
145 CHECK_S(rline(), "err empty request\n");
146 send_s("fleet\n");
147 CHECK_S(rline(), "err not in a fleet (single meshd: --cluster / --peer)\n");
148 send_s("jobs\n");
149 CHECK_S(rline(), "err the jobs thread is gone\n");
150 send_s("input hello abc\n");
151 CHECK_S(rline(), "ok 3 bytes\n");
152 send_s("input hello\n");
153 CHECK_S(rline(), "err usage: input <name> <bytes>\n");
154 send_s("key\n");
155 CHECK_S(rline(), "err usage: key <k>\n");
156 send_s("place\n");
157 CHECK_S(rline(), "err usage: place <name> [local | pin <k> | split [auto | a,b,.. | a:n,b:m,..]]\n");
158
159 char *f;
160 send_s("programs\n");
161 f = rframe(NULL);
162 CHECK_S(f, "0 hello running 2 1 on_demand 0.0 0000000000001234 0 0x100000+0x1000 kind wasm lanes 0 cpu 0.00 rss_mb 0.0\n"
163 "fields hello\ncounters hello tasks - - -\n");
164 free(f);
165 send_s("req hello 5\nabcde");
166 size_t n;
167 f = rframe(&n);
168 CHECK(n == 5 && memcmp(f, "abcde", 5) == 0);
169 free(f);
170 send_s("req hello x\n");
171 f = rframe(NULL);
172 CHECK_S(f, "err usage: req <name> <len>, then the bytes\n");
173 free(f);
174 send_s("req hello 2000000\n");
175 f = rframe(NULL);
176 CHECK_S(f, "err req: at most 1048576 bytes\n");
177 free(f);
178 send_s("peek hello 0 4\n");
179 f = rframe(&n);
180 CHECK(n == 4 && memcmp(f, "HELO", 4) == 0);
181 free(f);
182 send_s("peek hello 2 4\n");
183 f = rframe(NULL);
184 CHECK_S(f, "err hello: 0x2+0x4 is outside its 0x1000-byte view (offsets 4-aligned)\n");
185 free(f);
186 send_s("peek nope 0 4\n");
187 f = rframe(NULL);
188 CHECK_S(f, "err no program \"nope\"\n");
189 free(f);
190 send_s("view pipes\n");
191 f = rframe(NULL);
192 CHECK_S(f, "pipes system=0 inboxes=0\n");
193 free(f);
194 send_s("view fleet json\n");
195 f = rframe(NULL);
196 CHECK_S(f, "{\"fleet\":[{\"system\":0,\"n\":1,\"role\":\"single\",\"text\":\"not in a fleet (single meshd: --cluster / --peer)\"}]}\n");
197 free(f);
198 send_s("view bogus\n");
199 f = rframe(NULL);
200 CHECK_S(f, "err view \"bogus\": fleet, jobs or pipes [json]\n");
201 free(f);
202 send_s("pipes\n");
203 f = rframe(&n);
204 CHECK_U(n, 0);
205 free(f);
206 send_s("cluster\n");
207 f = rframe(NULL);
208 CHECK_S(f, "cluster 0 system 0 state single leader 0 up 0x0 linked 0x0 lock_wait_us 0.0\n");
209 free(f);
210 send_s("catalog\n");
211 f = rframe(NULL);
212 CHECK(strncmp(f, "err --programs /nonexistent-catalog: ", 37) == 0);
213 free(f);
214 send_s("snap\n");
215 f = rframe(&n);
216 CHECK(n > MESH_SNAP_HOST && (uint8_t)f[MESH_SN_COUNT] == 1);
217 free(f);
218 send_s("trace\n");
219 f = rframe(&n);
220 CHECK_U(n, 8 + MESH_TRACE_BYTES * MESH_TRACE_PER_PROG);
221 free(f);
222 send_s("render\n");
223 f = rframe(NULL);
224 CHECK_S(f, "err the blob thread is gone\n");
225 free(f);
226 send_s("infer\n");
227 f = rframe(NULL);
228 CHECK_S(f, "err no prompt (" MESH_INFER_USAGE ")\n");
229 free(f);
230 send_s("log 1\n");
231 f = rframe(NULL);
232 CHECK(strstr(f, "control: unknown or malformed request \"frobnicate\"") != NULL);
233 free(f);
234 /* The control counters moved. */
235 CHECK(atomic_load(&SH->host.net.ctl[2]) >= 20);
236 /* A line over 4096 bytes: refused, then closed. */
237 char *big = malloc(5000);
238 memset(big, 'x', 4999);
239 big[4999] = 0;
240 send_s(big);
241 free(big);
242 CHECK_S(rline(), "err request line over 4096 bytes\n");
243 char b1;
244 CHECK(read(FD, &b1, 1) == 0);
245 close(FD);
246
247 /* control.rs an_idle_client_is_closed: a request is answered, then silence closes the connection. */
248 int sv[2];
249 socketpair(AF_UNIX, SOCK_STREAM, 0, sv);
250 FD = sv[0];
251 send_s("log 1\n");
252 mesh_control_client(SH, sv[1], 300);
253 f = rframe(NULL);
254 free(f);
255 CHECK(read(sv[0], &b1, 1) == 0);
256 close(sv[0]);
257 close(lfd);
258 mesh_chan_drop_sender(SH->cmd); /* the driver stand-in's recv now ends */
259 pthread_join(dt, NULL);
260 free(mb.base);
261 return mesh_check_done();
262}