==> src/client.c <==1/* client.rs, ported: the endpoint thread, the control-port framing, the shared updates queue.
2 * Error texts are Rust's io::Error Display, since the status line shows them. */
3#include "mesh/top_client.h"
4
5#include <errno.h>
6#include <fcntl.h>
7#include <netdb.h>
8#include <netinet/in.h>
9#include <netinet/tcp.h>
10#include <poll.h>
11#include <pthread.h>
12#include <stdatomic.h>
13#include <stdio.h>
14#include <stdlib.h>
15#include <string.h>
16#include <sys/socket.h>
17#include <sys/time.h>
18#include <time.h>
19#include <unistd.h>
20
21static void *xmalloc(size_t n) {
22 void *p = malloc(n ? n : 1);
23 if (!p) abort();
24 return p;
25}
26
27static char *xstrdup(const char *s) {
28 size_t n = strlen(s) + 1;
29 return memcpy(xmalloc(n), s, n);
30}
31
32static uint64_t mono_ms(void) {
33 struct timespec t;
34 clock_gettime(CLOCK_MONOTONIC, &t);
35 return (uint64_t)t.tv_sec * 1000u + (uint64_t)t.tv_nsec / 1000000u;
36}
37
38static struct timespec deadline_ts(uint64_t at_ms) {
39 struct timespec t = {.tv_sec = (time_t)(at_ms / 1000u), .tv_nsec = (long)(at_ms % 1000u) * 1000000L};
40 return t;
41}
42
43static void cond_init_mono(pthread_cond_t *c) {
44 pthread_condattr_t a;
45 pthread_condattr_init(&a);
46 pthread_condattr_setclock(&a, CLOCK_MONOTONIC);
47 pthread_cond_init(c, &a);
48 pthread_condattr_destroy(&a);
49}
50
51/* The first whitespace-separated word of line, into w (empty when none). */
52static void first_word(const char *line, char *w, size_t cap) {
53 while (*line == ' ' || (*line >= '\t' && *line <= '\r')) line++;
54 size_t n = 0;
55 while (line[n] && !(line[n] == ' ' || (line[n] >= '\t' && line[n] <= '\r')) && n + 1 < cap) {
56 w[n] = line[n];
57 n++;
58 }
59 w[n] = 0;
60}
61
62static bool word_in(const char *w, const char *const *set, size_t n) {
63 for (size_t i = 0; i < n; i++)
64 if (strcmp(w, set[i]) == 0) return true;
65 return false;
66}
67
68uint32_t mt_reply_timeout_ms(const char *line) {
69 static const char *const slow[] = {"load", "upgrade", "rollback", "unload"};
70 static const char *const reset[] = {"reset", "train-reset"};
71 static const char *const req[] = {"http", "req", "task"};
72 char w[32];
73 first_word(line, w, sizeof w);
74 if (word_in(w, slow, 4)) return 20000;
75 if (word_in(w, reset, 2)) return 15000;
76 if (word_in(w, req, 3)) return 8000;
77 return MT_IO_TIMEOUT_MS;
78}
79
80bool mt_binary_reply(const char *line) {
81 static const char *const framed[] = {"snap", "http", "req", "peek", "log", "programs", "catalog", "trace", "view", "render"};
82 /* `load` alone is the load report; `load <dir>` loads a program (one line). */
83 const char *s = line;
84 size_t l = strlen(s);
85 while (l && (*s == ' ' || (*s >= '\t' && *s <= '\r'))) s++, l--;
86 while (l && (s[l - 1] == ' ' || (s[l - 1] >= '\t' && s[l - 1] <= '\r'))) l--;
87 if (l == 4 && memcmp(s, "load", 4) == 0) return true;
88 char w[32];
89 first_word(line, w, sizeof w);
90 return word_in(w, framed, sizeof framed / sizeof *framed);
91}
92
93/* ---- the updates queue ---- */
94
95struct mt_updates {
96 pthread_mutex_t mu;
97 pthread_cond_t cv;
98 mt_update *q;
99 size_t head, n, cap;
100};
101
102mt_updates *mt_updates_new(void) {
103 mt_updates *q = xmalloc(sizeof *q);
104 memset(q, 0, sizeof *q);
105 pthread_mutex_init(&q->mu, NULL);
106 cond_init_mono(&q->cv);
107 return q;
108}
109
110void mt_update_free(mt_update *u) {
111 free(u->text);
112 free(u->data);
113 if (u->snap.bytes) mt_snapshot_free(&u->snap);
114 memset(u, 0, sizeof *u);
115}
116
117void mt_updates_free(mt_updates *q) {
118 if (!q) return;
119 for (size_t i = 0; i < q->n; i++) mt_update_free(&q->q[(q->head + i) % q->cap]);
120 free(q->q);
121 pthread_mutex_destroy(&q->mu);
122 pthread_cond_destroy(&q->cv);
123 free(q);
124}
125
126void mt_updates_push(mt_updates *q, mt_update u) {
127 pthread_mutex_lock(&q->mu);
128 if (q->n == q->cap) {
129 size_t cap = q->cap ? q->cap * 2 : 64;
130 mt_update *nq = xmalloc(cap * sizeof *nq);
131 for (size_t i = 0; i < q->n; i++) nq[i] = q->q[(q->head + i) % q->cap];
132 free(q->q);
133 q->q = nq;
134 q->cap = cap;
135 q->head = 0;
136 }
137 q->q[(q->head + q->n) % q->cap] = u;
138 q->n++;
139 pthread_cond_signal(&q->cv);
140 pthread_mutex_unlock(&q->mu);
141}
142
143bool mt_updates_pop(mt_updates *q, uint32_t timeout_ms, mt_update *out) {
144 pthread_mutex_lock(&q->mu);
145 struct timespec dl = deadline_ts(mono_ms() + timeout_ms);
146 while (q->n == 0 && timeout_ms) {
147 if (pthread_cond_timedwait(&q->cv, &q->mu, &dl) == ETIMEDOUT) break;
148 }
149 bool got = q->n > 0;
150 if (got) {
151 *out = q->q[q->head];
152 q->head = (q->head + 1) % q->cap;
153 q->n--;
154 }
155 pthread_mutex_unlock(&q->mu);
156 return got;
157}
158
159/* ---- the endpoint ---- */
160
161struct mt_endpoint {
162 size_t idx;
163 char *addr;
164 mt_updates *up;
165 atomic_size_t pending;
166 atomic_uint_fast64_t interval;
167 pthread_t th;
168 pthread_mutex_t mu; /* guards cmds, closed, fd */
169 pthread_cond_t cv;
170 char **cmds;
171 size_t head, n, cap;
172 bool closed;
173 int fd; /* the live connection, so free can interrupt a request */
174};
175
176/* Rust's io::Error Display for an OS error. */
177static char *os_err(int e) {
178 char b[160];
179 snprintf(b, sizeof b, "%s (os error %d)", strerror(e), e);
180 return xstrdup(b);
181}
182
183typedef enum { RECV_LINE, RECV_TIMEOUT, RECV_CLOSED } recv_res;
184
185/* rx.recv_timeout: a command line, a timeout, or the endpoint being dropped. */
186static recv_res recv_cmd(mt_endpoint *e, uint64_t until_ms, char **line) {
187 pthread_mutex_lock(&e->mu);
188 struct timespec dl = deadline_ts(until_ms);
189 for (;;) {
190 if (e->n) {
191 *line = e->cmds[e->head];
192 e->head = (e->head + 1) % e->cap;
193 e->n--;
194 pthread_mutex_unlock(&e->mu);
195 return RECV_LINE;
196 }
197 if (e->closed) break;
198 if (mono_ms() >= until_ms || pthread_cond_timedwait(&e->cv, &e->mu, &dl) == ETIMEDOUT) {
199 if (e->n) continue;
200 pthread_mutex_unlock(&e->mu);
201 return e->closed ? RECV_CLOSED : RECV_TIMEOUT;
202 }
203 }
204 pthread_mutex_unlock(&e->mu);
205 return RECV_CLOSED;
206}
207
208static bool is_closed(mt_endpoint *e) {
209 pthread_mutex_lock(&e->mu);
210 bool c = e->closed;
211 pthread_mutex_unlock(&e->mu);
212 return c;
213}
214
215/* `host:port` or `[v6]:port` resolved and connected with a 1 s timeout; NULL or the error. */
216static char *dial(const char *addr, int *out) {
217 const char *colon = strrchr(addr, ':');
218 if (!colon) return xstrdup("invalid socket address");
219 char host[256];
220 const char *h = addr;
221 size_t hl = (size_t)(colon - addr);
222 if (hl >= 2 && h[0] == '[' && h[hl - 1] == ']') h++, hl -= 2;
223 if (hl >= sizeof host) return xstrdup("invalid socket address");
224 memcpy(host, h, hl);
225 host[hl] = 0;
226 char *end;
227 unsigned long port = strtoul(colon + 1, &end, 10);
228 if (!colon[1] || *end || port > 65535) return xstrdup("invalid port value");
229 struct addrinfo hints, *res = NULL;
230 memset(&hints, 0, sizeof hints);
231 hints.ai_socktype = SOCK_STREAM;
232 int g = getaddrinfo(host, colon + 1, &hints, &res);
233 if (g != 0) {
234 char b[256];
235 snprintf(b, sizeof b, "failed to lookup address information: %s", gai_strerror(g));
236 return xstrdup(b);
237 }
238 if (!res) return xstrdup("address resolves to nothing");
239 int fd = socket(res->ai_family, SOCK_STREAM | SOCK_CLOEXEC, 0);
240 if (fd < 0) {
241 int er = errno;
242 freeaddrinfo(res);
243 return os_err(er);
244 }
245 int fl = fcntl(fd, F_GETFL);
246 fcntl(fd, F_SETFL, fl | O_NONBLOCK);
247 int r = connect(fd, res->ai_addr, res->ai_addrlen);
248 freeaddrinfo(res);
249 if (r != 0 && errno != EINPROGRESS) {
250 int er = errno;
251 close(fd);
252 return os_err(er);
253 }
254 if (r != 0) {
255 struct pollfd p = {.fd = fd, .events = POLLOUT};
256 int k;
257 do k = poll(&p, 1, 1000);
258 while (k < 0 && errno == EINTR);
259 if (k == 0) {
260 close(fd);
261 return xstrdup("connection timed out");
262 }
263 int so = 0;
264 socklen_t sl = sizeof so;
265 getsockopt(fd, SOL_SOCKET, SO_ERROR, &so, &sl);
266 if (k < 0 || so) {
267 int er = k < 0 ? errno : so;
268 close(fd);
269 return os_err(er);
270 }
271 }
272 fcntl(fd, F_SETFL, fl & ~O_NONBLOCK);
273 int one = 1;
274 setsockopt(fd, IPPROTO_TCP, TCP_NODELAY, &one, sizeof one);
275 struct timeval tv = {.tv_sec = MT_IO_TIMEOUT_MS / 1000};
276 setsockopt(fd, SOL_SOCKET, SO_SNDTIMEO, &tv, sizeof tv);
277 *out = fd;
278 return NULL;
279}
280
281/* A connection with BufReader's buffer. */
282typedef struct {
283 int fd;
284 uint8_t buf[8192];
285 size_t at, have;
286} conn;
287
288/* Fills the buffer; NULL, or the error. */
289static char *fill(conn *c) {
290 for (;;) {
291 ssize_t k = recv(c->fd, c->buf, sizeof c->buf, 0);
292 if (k < 0 && errno == EINTR) continue;
293 if (k < 0) return os_err(errno);
294 c->at = 0;
295 c->have = (size_t)k;
296 return NULL;
297 }
298}
299
300static char *read_exact(conn *c, uint8_t *b, size_t n) {
301 while (n) {
302 if (c->at == c->have) {
303 char *e = fill(c);
304 if (e) return e;
305 if (!c->have) return xstrdup("failed to fill whole buffer");
306 }
307 size_t k = c->have - c->at < n ? c->have - c->at : n;
308 memcpy(b, c->buf + c->at, k);
309 c->at += k;
310 b += k;
311 n -= k;
312 }
313 return NULL;
314}
315
316/* read_line onto prefix, then trim_end; *out, or the error. */
317static char *text_line(conn *c, const char *prefix, char **out) {
318 size_t cap = 256, n = strlen(prefix), got = 0;
319 char *s = xmalloc(cap + n);
320 memcpy(s, prefix, n);
321 cap += n;
322 for (;;) {
323 if (c->at == c->have) {
324 char *e = fill(c);
325 if (e) {
326 free(s);
327 return e;
328 }
329 if (!c->have) break;
330 }
331 uint8_t ch = c->buf[c->at++];
332 if (n + 1 >= cap) {
333 cap *= 2;
334 char *t = realloc(s, cap);
335 if (!t) abort();
336 s = t;
337 }
338 s[n++] = (char)ch;
339 got++;
340 if (ch == '\n') break;
341 }
342 if (!got) {
343 free(s);
344 return xstrdup("connection closed");
345 }
346 while (n && (s[n - 1] == ' ' || (s[n - 1] >= '\t' && s[n - 1] <= '\r'))) n--;
347 s[n] = 0;
348 *out = s;
349 return NULL;
350}
351
352static char *send_all(int fd, const char *p, size_t n) {
353 while (n) {
354 ssize_t k = send(fd, p, n, MSG_NOSIGNAL);
355 if (k < 0 && errno == EINTR) continue;
356 if (k < 0) return os_err(errno);
357 if (k == 0) return xstrdup("failed to write whole buffer");
358 p += k;
359 n -= (size_t)k;
360 }
361 return NULL;
362}
363
364typedef struct {
365 bool bytes;
366 uint8_t *data;
367 size_t len;
368} reply;
369
370static char *request(conn *c, const char *line, reply *r) {
371 uint32_t ms = mt_reply_timeout_ms(line);
372 struct timeval tv = {.tv_sec = ms / 1000, .tv_usec = (ms % 1000) * 1000};
373 setsockopt(c->fd, SOL_SOCKET, SO_RCVTIMEO, &tv, sizeof tv);
374 size_t ll = strlen(line);
375 char *msg = xmalloc(ll + 1);
376 memcpy(msg, line, ll);
377 msg[ll] = '\n';
378 char *e = send_all(c->fd, msg, ll + 1);
379 free(msg);
380 if (e) return e;
381 char *t;
382 if (!mt_binary_reply(line)) {
383 if ((e = text_line(c, "", &t))) return e;
384 r->bytes = false;
385 r->data = (uint8_t *)t;
386 r->len = strlen(t);
387 return NULL;
388 }
389 uint8_t len[4];
390 if ((e = read_exact(c, len, 4))) return e;
391 /* A refused framed request answers with a text `err ...` line; its first four bytes would
392 * decode to a ~544 MB length, so they are unambiguous. */
393 if (memcmp(len, "err ", 4) == 0) {
394 if ((e = text_line(c, "err ", &t))) return e;
395 r->bytes = false;
396 r->data = (uint8_t *)t;
397 r->len = strlen(t);
398 return NULL;
399 }
400 size_t n = (size_t)len[0] | (size_t)len[1] << 8 | (size_t)len[2] << 16 | (size_t)len[3] << 24;
401 if (n > MT_MAX_REPLY) {
402 char b[96];
403 snprintf(b, sizeof b, "reply length %zu exceeds %zu", n, MT_MAX_REPLY);
404 return xstrdup(b);
405 }
406 uint8_t *buf = xmalloc(n);
407 if ((e = read_exact(c, buf, n))) {
408 free(buf);
409 return e;
410 }
411 r->bytes = true;
412 r->data = buf;
413 r->len = n;
414 return NULL;
415}
416
417static mt_update reply_update(size_t idx, char *line, reply r) {
418 mt_update u;
419 memset(&u, 0, sizeof u);
420 u.kind = MT_UPDATE_REPLY;
421 u.machine = idx;
422 u.text = line;
423 u.bytes = r.bytes;
424 u.data = r.data;
425 u.len = r.len;
426 return u;
427}
428
429static mt_update text_update(mt_update_kind k, size_t idx, char *text) {
430 mt_update u;
431 memset(&u, 0, sizeof u);
432 u.kind = k;
433 u.machine = idx;
434 u.text = text;
435 return u;
436}
437
438static void set_fd(mt_endpoint *e, int fd) {
439 pthread_mutex_lock(&e->mu);
440 e->fd = fd;
441 pthread_mutex_unlock(&e->mu);
442}
443
444/* Sleeps ms unless the endpoint is dropped first; false when it was. */
445static bool nap(mt_endpoint *e, uint64_t ms) {
446 pthread_mutex_lock(&e->mu);
447 struct timespec dl = deadline_ts(mono_ms() + ms);
448 while (!e->closed)
449 if (pthread_cond_timedwait(&e->cv, &e->mu, &dl) == ETIMEDOUT) break;
450 bool open = !e->closed;
451 pthread_mutex_unlock(&e->mu);
452 return open;
453}
454
455static void *run(void *arg) {
456 mt_endpoint *e = arg;
457 uint64_t backoff = MT_BACKOFF_MIN_MS;
458 conn *c = xmalloc(sizeof *c);
459 for (;;) {
460 if (is_closed(e)) break;
461 int fd = -1;
462 char *err = dial(e->addr, &fd);
463 if (err) {
464 mt_updates_push(e->up, text_update(MT_UPDATE_OFFLINE, e->idx, err));
465 /* Commands issued while offline fail at once instead of piling up. */
466 uint64_t deadline = mono_ms() + backoff;
467 bool dropped = false;
468 for (;;) {
469 char *line;
470 recv_res r = recv_cmd(e, deadline, &line);
471 if (r == RECV_TIMEOUT) break;
472 if (r == RECV_CLOSED) {
473 dropped = true;
474 break;
475 }
476 atomic_fetch_sub(&e->pending, 1);
477 reply rp = {false, (uint8_t *)xstrdup("err offline"), 11};
478 mt_updates_push(e->up, reply_update(e->idx, line, rp));
479 }
480 if (dropped) break;
481 backoff = backoff * 2 < MT_BACKOFF_MAX_MS ? backoff * 2 : MT_BACKOFF_MAX_MS;
482 continue;
483 }
484 c->fd = fd;
485 c->at = c->have = 0;
486 set_fd(e, fd);
487 backoff = MT_BACKOFF_MIN_MS;
488 mt_updates_push(e->up, text_update(MT_UPDATE_ONLINE, e->idx, NULL));
489 uint64_t next_snap = mono_ms();
490 bool dropped = false;
491 for (;;) {
492 char *line;
493 recv_res r = recv_cmd(e, next_snap, &line);
494 if (r == RECV_CLOSED) {
495 dropped = true;
496 break;
497 }
498 if (r == RECV_LINE) {
499 reply rp;
500 err = request(c, line, &rp);
501 atomic_fetch_sub(&e->pending, 1);
502 if (!err) {
503 /* Doom input shows on the next frame, not 66 ms later. */
504 if (strncmp(line, "input ", 6) == 0 || strncmp(line, "key ", 4) == 0) next_snap = mono_ms();
505 mt_updates_push(e->up, reply_update(e->idx, line, rp));
506 continue;
507 }
508 size_t el = strlen(err);
509 char *t = xmalloc(el + 5);
510 memcpy(t, "err ", 4);
511 memcpy(t + 4, err, el + 1);
512 rp.bytes = false;
513 rp.data = (uint8_t *)t;
514 rp.len = el + 4;
515 mt_updates_push(e->up, reply_update(e->idx, line, rp));
516 break;
517 }
518 reply rp;
519 err = request(c, "snap", &rp);
520 next_snap = mono_ms() + atomic_load(&e->interval);
521 if (err) break;
522 if (rp.bytes) {
523 mt_snapshot s;
524 char *perr = NULL;
525 if (mt_snapshot_parse(rp.data, rp.len, &s, &perr)) mt_updates_push(e->up, mt_update_snap(e->idx, s));
526 else mt_updates_push(e->up, text_update(MT_UPDATE_SNAP_ERROR, e->idx, perr));
527 free(rp.data);
528 } else {
529 mt_updates_push(e->up, text_update(MT_UPDATE_SNAP_ERROR, e->idx, (char *)rp.data));
530 }
531 }
532 set_fd(e, -1);
533 close(fd);
534 if (dropped) break;
535 mt_updates_push(e->up, text_update(MT_UPDATE_OFFLINE, e->idx, err));
536 if (!nap(e, backoff)) break;
537 }
538 free(c);
539 return NULL;
540}
541
542mt_endpoint *mt_endpoint_spawn(size_t idx, const char *addr, mt_updates *updates, uint64_t interval_ms) {
543 mt_endpoint *e = xmalloc(sizeof *e);
544 memset(e, 0, sizeof *e);
545 e->idx = idx;
546 e->addr = xstrdup(addr);
547 e->up = updates;
548 e->fd = -1;
549 atomic_init(&e->pending, 0);
550 atomic_init(&e->interval, interval_ms);
551 pthread_mutex_init(&e->mu, NULL);
552 cond_init_mono(&e->cv);
553 if (pthread_create(&e->th, NULL, run, e) != 0) abort();
554 return e;
555}
556
557void mt_endpoint_send(mt_endpoint *e, const char *line) {
558 atomic_fetch_add(&e->pending, 1);
559 pthread_mutex_lock(&e->mu);
560 if (e->closed) {
561 atomic_fetch_sub(&e->pending, 1);
562 pthread_mutex_unlock(&e->mu);
563 return;
564 }
565 if (e->n == e->cap) {
566 size_t cap = e->cap ? e->cap * 2 : 16;
567 char **nq = xmalloc(cap * sizeof *nq);
568 for (size_t i = 0; i < e->n; i++) nq[i] = e->cmds[(e->head + i) % e->cap];
569 free(e->cmds);
570 e->cmds = nq;
571 e->cap = cap;
572 e->head = 0;
573 }
574 e->cmds[(e->head + e->n) % e->cap] = xstrdup(line);
575 e->n++;
576 pthread_cond_signal(&e->cv);
577 pthread_mutex_unlock(&e->mu);
578}
579
580size_t mt_endpoint_pending(const mt_endpoint *e) { return atomic_load(&((mt_endpoint *)e)->pending); }
581
582void mt_endpoint_set_interval(mt_endpoint *e, uint64_t ms) { atomic_store(&e->interval, ms); }
583
584uint64_t mt_endpoint_interval(const mt_endpoint *e) { return atomic_load(&((mt_endpoint *)e)->interval); }
585
586const char *mt_endpoint_addr(const mt_endpoint *e) { return e->addr; }
587
588void mt_endpoint_free(mt_endpoint *e) {
589 if (!e) return;
590 pthread_mutex_lock(&e->mu);
591 e->closed = true;
592 if (e->fd >= 0) shutdown(e->fd, SHUT_RDWR);
593 pthread_cond_broadcast(&e->cv);
594 pthread_mutex_unlock(&e->mu);
595 pthread_join(e->th, NULL);
596 for (size_t i = 0; i < e->n; i++) free(e->cmds[(e->head + i) % e->cap]);
597 free(e->cmds);
598 free(e->addr);
599 pthread_mutex_destroy(&e->mu);
600 pthread_cond_destroy(&e->cv);
601 free(e);
602}