==> src/control.c <==1/* The control port: the client loop, the request dispatch, and the texts it answers with. */
2#include "mesh/control.h"
3#include "mesh/abi.h"
4#include "mesh/backend.h"
5#include "mesh/infer_request.h"
6#include "mesh/inferd.h"
7#include "mesh/load_text.h"
8#include "mesh/loader.h"
9#include "mesh/log.h"
10#include "mesh/prog_table.h"
11#include "mesh/rsfmt.h"
12#include "mesh/snapshot.h"
13
14#include <errno.h>
15#include <netinet/in.h>
16#include <netinet/tcp.h>
17#include <pthread.h>
18#include <stdatomic.h>
19#include <stdio.h>
20#include <stdlib.h>
21#include <string.h>
22#include <sys/socket.h>
23#include <sys/time.h>
24#include <unistd.h>
25
26bool (*mesh_control_render)(mesh_shared *sh, char **words, size_t n, uint8_t **out, size_t *len, char **err);
27
28#define LIFECYCLE_MS 30000
29#define TASK_MS 10000
30#define PLACE_MS 15000
31#define FORWARD_WAIT_MS 70000 /* peer.rs FORWARD_WAIT */
32#define REQ_MAX ((size_t)1 << 20)
33#define LINE_MAX 4096
34#define MAX_CLIENTS 32
35#define CLIENT_IDLE_MS 120000
36#define CLIENT_WRITE_MS 10000
37
38/* ---------------------------------------------------------------- reply shapes */
39
40static void framed(mesh_buf *out, const void *b, size_t n) {
41 mesh_buf_u32le(out, (uint32_t)n);
42 mesh_buf_put(out, b, n);
43}
44
45static void framed_str(mesh_buf *out, const char *s) { framed(out, s, strlen(s)); }
46
47static void framed_err(mesh_buf *out, const char *e) {
48 mesh_buf t = MESH_BUF_INIT;
49 mesh_buf_printf(&t, "err %s\n", e);
50 framed(out, t.data, t.len);
51 mesh_buf_free(&t);
52}
53
54static void line(mesh_buf *out, const char *s) { mesh_buf_printf(out, "%s\n", s); }
55
56/* Ok("") is `ok`, Ok(s) `ok s`, Err(e) `err e`. */
57static void reply(mesh_buf *out, bool ok, const char *text) {
58 if (ok && !*text) line(out, "ok");
59 else mesh_buf_printf(out, "%s %s\n", ok ? "ok" : "err", text);
60}
61
62/* ---------------------------------------------------------------- small helpers */
63
64static uint32_t rd32(const uint8_t *b, size_t o) { return (uint32_t)b[o] | (uint32_t)b[o + 1] << 8 | (uint32_t)b[o + 2] << 16 | (uint32_t)b[o + 3] << 24; }
65static uint64_t rd64(const uint8_t *b, size_t o) { return (uint64_t)rd32(b, o) | (uint64_t)rd32(b, o + 4) << 32; }
66static float rdf(const uint8_t *b, size_t o) { uint32_t u = rd32(b, o); float f; memcpy(&f, &u, 4); return f; }
67
68/* A NUL-padded string field as text. */
69static void cstr(mesh_buf *out, const uint8_t *b, size_t n) {
70 size_t k = 0;
71 while (k < n && b[k]) k++;
72 mesh_rs_lossy(out, b, k);
73}
74
75/* `0x` hex as Rust's {:#x}. */
76static void hexalt(mesh_buf *b, uint64_t v) { mesh_buf_printf(b, "0x%llx", (unsigned long long)v); }
77
78/* "0x.." or decimal u32 (control.rs int). */
79static bool int_(const char *s, uint32_t *v) {
80 if (strncmp(s, "0x", 2) == 0) {
81 const char *h = s + 2;
82 size_t i = (*h == '+' && h[1]) ? 1 : 0;
83 if (!h[i]) return false;
84 uint64_t x = 0;
85 for (; h[i]; i++) {
86 char c = h[i];
87 int d = c >= '0' && c <= '9' ? c - '0' : c >= 'a' && c <= 'f' ? c - 'a' + 10 : c >= 'A' && c <= 'F' ? c - 'A' + 10 : -1;
88 if (d < 0) return false;
89 x = x * 16 + (uint64_t)d;
90 if (x > UINT32_MAX) return false;
91 }
92 *v = (uint32_t)x;
93 return true;
94 }
95 return mesh_rs_parse_u32(s, strlen(s), v) == 0;
96}
97
98static bool is_ws(char c) { return c == ' ' || (c >= 9 && c <= 13); }
99
100/* split_whitespace: up to max words (pointers into a copy the caller frees). */
101static size_t split_ws(char *s, char **w, size_t max) {
102 size_t n = 0;
103 while (*s) {
104 while (*s && is_ws(*s)) s++;
105 if (!*s) break;
106 if (n < max) w[n] = s;
107 n++;
108 while (*s && !is_ws(*s)) s++;
109 if (*s) *s++ = 0;
110 }
111 return n;
112}
113
114static char *debug_q(const char *s) {
115 mesh_buf b = MESH_BUF_INIT;
116 mesh_rs_debug_cstr(&b, s);
117 return mesh_buf_take(&b);
118}
119
120/* In a fleet, and name is the loaded lockstep program (its input and requests are refused). */
121static bool lockstep(mesh_shared *sh, const char *name) {
122 uint32_t s;
123 return sh->peer && mesh_pt_find(sh->mb, name, &s) && mesh_mailbox_get(sh->mb, mesh_prog_entry(s) + MESH_PT_PEER) == MESH_PEER_MODE_LOCKSTEP;
124}
125
126/* A change of the desired state (fleet), answered by the leader. */
127static void change(mesh_shared *sh, const char *req, mesh_buf *out) {
128 char *text = NULL;
129 bool ok = mesh_shared_change(sh, req, FORWARD_WAIT_MS + 5000, &text);
130 if (!ok) {
131 char *q = debug_q(req);
132 mesh_error("control: %s: %s", q, text);
133 free(q);
134 }
135 reply(out, ok, text);
136 free(text);
137}
138
139static void ask(mesh_shared *sh, mesh_cmd c, int64_t within, mesh_buf *out) {
140 char *text = NULL;
141 bool ok = mesh_shared_ask(sh, c, within, &text);
142 reply(out, ok, text);
143 free(text);
144}
145
146/* ---------------------------------------------------------------- texts */
147
148static void names_of(mesh_shared *sh, uint32_t mask, mesh_buf *out, const char *none) {
149 size_t at = out->len;
150 for (uint32_t s = 0; s < MESH_PROG_SLOTS; s++) {
151 if (!((mask >> s) & 1u)) continue;
152 char n[MESH_NAME_BYTES + 1];
153 mesh_pt_name_of(sh->mb, s, n);
154 mesh_buf_printf(out, "%s%s", out->len > at ? "," : "", n);
155 }
156 if (out->len == at) mesh_buf_puts(out, none);
157}
158
159/* An inbox's (pushed, popped, refused, empty pops, waiting): its two sides' ring counters summed. */
160static void inbox_counts(mesh_shared *sh, mesh_pipe_inbox in, uint32_t c[5]) {
161 mesh_words w = mesh_mailbox_words(sh->mb);
162 mesh_pipe_ring a = mesh_pipe_inbox_side(&in, MESH_PIPE_SIDE_CPU), b = mesh_pipe_inbox_side(&in, MESH_PIPE_SIDE_GPU);
163 const size_t offs[4] = {MESH_PIPE_PI_HEAD, MESH_PIPE_PI_TAIL, MESH_PIPE_PI_FULL, MESH_PIPE_PI_EMPTY};
164 for (int i = 0; i < 4; i++) c[i] = mesh_words_load(&w, a.base + offs[i]) + mesh_words_load(&w, b.base + offs[i]);
165 c[4] = mesh_pipe_inbox_len(&in, &w);
166}
167
168static bool inbox_of(mesh_shared *sh, uint32_t s, mesh_pipe_inbox *in, char *ty) {
169 mesh_mailbox mb = sh->mb;
170 if (mesh_mailbox_get(mb, mesh_prog_entry(s) + MESH_PT_STATE) == MESH_ST_EMPTY) return false;
171 size_t p = mesh_pipe_entry(s);
172 uint32_t cap = mesh_mailbox_get(mb, p + MESH_PP_IN_CAP), rec = mesh_mailbox_get(mb, p + MESH_PP_IN_REC);
173 if (!cap) return false;
174 in->base = (size_t)mesh_mailbox_get(mb, mesh_prog_entry(s) + MESH_PT_VIEW_OFF) + mesh_mailbox_get(mb, p + MESH_PP_IN_OFF);
175 in->cap = cap;
176 in->rec = rec;
177 uint8_t t[MESH_NAME_BYTES];
178 mesh_mailbox_read(mb, p + MESH_PP_TYPE_NAME, t, MESH_NAME_BYTES);
179 mesh_buf b = MESH_BUF_INIT;
180 mesh_rs_lossy(&b, t, MESH_NAME_BYTES);
181 /* trim_end_matches('\0') */
182 while (b.len && b.data[b.len - 1] == 0) b.len--;
183 snprintf(ty, 64, "%.*s", (int)b.len, b.data ? b.data : "");
184 mesh_buf_free(&b);
185 return true;
186}
187
188void mesh_control_pipes(mesh_shared *sh, mesh_buf *out) {
189 mesh_mailbox mb = sh->mb;
190 for (uint32_t s = 0; s < MESH_PROG_SLOTS; s++) {
191 mesh_pipe_inbox in;
192 char ty[64];
193 if (!inbox_of(sh, s, &in, ty)) continue;
194 uint32_t c[5];
195 inbox_counts(sh, in, c);
196 char n[MESH_NAME_BYTES + 1];
197 mesh_pt_name_of(mb, s, n);
198 mesh_buf_printf(out, "inbox %s type %s cap %u rec %u head %u tail %u len %u full %u empty %u fed_by ", n, ty, in.cap, in.rec, c[0], c[1], c[4],
199 c[2], c[3]);
200 names_of(sh, mesh_mailbox_get(mb, mesh_pipe_entry(s) + MESH_PP_FED_MASK), out, "-");
201 mesh_buf_putc(out, '\n');
202 }
203 for (uint32_t s = 0; s < MESH_PROG_SLOTS; s++) {
204 uint32_t f = mesh_mailbox_get(mb, mesh_pipe_entry(s) + MESH_PP_FEEDS_MASK);
205 if (!f || mesh_mailbox_get(mb, mesh_prog_entry(s) + MESH_PT_STATE) == MESH_ST_EMPTY) continue;
206 char n[MESH_NAME_BYTES + 1];
207 mesh_pt_name_of(mb, s, n);
208 mesh_buf_printf(out, "feeds %s -> ", n);
209 names_of(sh, f, out, "-");
210 mesh_buf_putc(out, '\n');
211 }
212}
213
214void mesh_control_pipes_recs(mesh_shared *sh, mesh_recs *v) {
215 mesh_mailbox mb = sh->mb;
216 int64_t now = mesh_mono_ns();
217 mesh_recs inboxes = MESH_RECS_INIT;
218 char seen[MESH_PROG_SLOTS][MESH_NAME_BYTES + 1];
219 size_t nseen = 0;
220 pthread_mutex_lock(&sh->mu);
221 for (uint32_t s = 0; s < MESH_PROG_SLOTS; s++) {
222 mesh_pipe_inbox in;
223 char ty[64];
224 if (!inbox_of(sh, s, &in, ty)) continue;
225 uint32_t c[5];
226 inbox_counts(sh, in, c);
227 uint32_t head = c[0], tail = c[1], full = c[2], empty = c[3], depth = c[4];
228 char name[MESH_NAME_BYTES + 1];
229 mesh_pt_name_of(mb, s, name);
230 /* Rates since the last sample (>= 0.5 s ago); a reload (counts going back) restarts them. */
231 mesh_pipe_sample *e = NULL;
232 for (size_t i = 0; i < sh->npipe_hist; i++) if (strcmp(sh->pipe_hist[i].name, name) == 0) e = &sh->pipe_hist[i];
233 if (!e) {
234 sh->pipe_hist = mesh_xrealloc(sh->pipe_hist, (sh->npipe_hist + 1) * sizeof *sh->pipe_hist);
235 e = &sh->pipe_hist[sh->npipe_hist++];
236 memset(e, 0, sizeof *e);
237 snprintf(e->name, sizeof e->name, "%s", name);
238 e->at_ns = now;
239 e->head = head;
240 e->tail = tail;
241 e->full = full;
242 }
243 float dt = (float)((double)(now - e->at_ns) / 1e9);
244 if (head < e->head || tail < e->tail || full < e->full) {
245 e->at_ns = now;
246 e->head = head;
247 e->tail = tail;
248 e->full = full;
249 e->rates[0] = e->rates[1] = e->rates[2] = 0;
250 } else if (dt >= 0.5f) {
251 e->rates[0] = (float)(head - e->head) / dt;
252 e->rates[1] = (float)(tail - e->tail) / dt;
253 e->rates[2] = (float)(full - e->full) / dt;
254 e->at_ns = now;
255 e->head = head;
256 e->tail = tail;
257 e->full = full;
258 }
259 float rates[3] = {e->rates[0], e->rates[1], e->rates[2]};
260 snprintf(seen[nseen++], MESH_NAME_BYTES + 1, "%s", name);
261 mesh_rec *r = mesh_recs_add(&inboxes, "inbox");
262 mesh_rec_kv(r, "consumer", name);
263 mesh_rec_kv(r, "type", ty);
264 mesh_rec_kvu(r, "cap", in.cap);
265 mesh_rec_kvu(r, "rec", in.rec);
266 mesh_rec_kvu(r, "depth", depth);
267 mesh_rec_f(r, "fill", true, (double)depth / (double)(in.cap > 1 ? in.cap : 1), 3);
268 mesh_rec_kvu(r, "pushed", head);
269 mesh_rec_kvu(r, "popped", tail);
270 mesh_rec_kvu(r, "refused", full);
271 mesh_rec_kvu(r, "empty", empty);
272 mesh_buf fb = MESH_BUF_INIT;
273 names_of(sh, mesh_mailbox_get(mb, mesh_pipe_entry(s) + MESH_PP_FED_MASK), &fb, "");
274 mesh_rec_kv(r, "fed_by", mesh_buf_str(&fb));
275 mesh_buf_free(&fb);
276 mesh_rec_f(r, "in_s", true, rates[0], 1);
277 mesh_rec_f(r, "out_s", true, rates[1], 1);
278 mesh_rec_f(r, "refused_s", true, rates[2], 1);
279 }
280 size_t k = 0;
281 for (size_t i = 0; i < sh->npipe_hist; i++) {
282 bool keep = false;
283 for (size_t j = 0; j < nseen; j++) keep |= strcmp(seen[j], sh->pipe_hist[i].name) == 0;
284 if (keep) sh->pipe_hist[k++] = sh->pipe_hist[i];
285 }
286 sh->npipe_hist = k;
287 pthread_mutex_unlock(&sh->mu);
288 mesh_rec *p = mesh_recs_add(v, "pipes");
289 mesh_rec_kvu(p, "system", mesh_host_block_get_u32(&sh->host, MESH_S_SYSTEM));
290 mesh_rec_kvu(p, "inboxes", inboxes.len);
291 mesh_recs_extend(v, &inboxes);
292 mesh_recs_free(&inboxes);
293 for (uint32_t s = 0; s < MESH_PROG_SLOTS; s++) {
294 uint32_t f = mesh_mailbox_get(mb, mesh_pipe_entry(s) + MESH_PP_FEEDS_MASK);
295 if (!f || mesh_mailbox_get(mb, mesh_prog_entry(s) + MESH_PT_STATE) == MESH_ST_EMPTY) continue;
296 char n[MESH_NAME_BYTES + 1];
297 mesh_pt_name_of(mb, s, n);
298 mesh_rec *r = mesh_recs_add(v, "feed");
299 mesh_rec_kv(r, "producer", n);
300 mesh_buf cb = MESH_BUF_INIT;
301 names_of(sh, f, &cb, "");
302 mesh_rec_kv(r, "consumers", mesh_buf_str(&cb));
303 mesh_buf_free(&cb);
304 }
305}
306
307int mesh_control_view(mesh_shared *sh, const char *which, bool json, mesh_buf *out, char **err) {
308 mesh_recs recs = MESH_RECS_INIT;
309 if (strcmp(which, "fleet") == 0 && sh->peer) {
310 pthread_mutex_lock(&sh->mu);
311 mesh_recs_extend(&recs, &sh->fleet_view);
312 pthread_mutex_unlock(&sh->mu);
313 } else if (strcmp(which, "fleet") == 0) {
314 mesh_rec *r = mesh_recs_add(&recs, "fleet");
315 mesh_rec_kvu(r, "system", mesh_host_block_get_u32(&sh->host, MESH_S_SYSTEM));
316 mesh_rec_kvu(r, "n", 1);
317 mesh_rec_kv(r, "role", "single");
318 mesh_rec_text(r, "not in a fleet (single meshd: --cluster / --peer)");
319 } else if (strcmp(which, "jobs") == 0) {
320 pthread_mutex_lock(&sh->mu);
321 mesh_recs_extend(&recs, &sh->jobs_view_recs);
322 pthread_mutex_unlock(&sh->mu);
323 } else if (strcmp(which, "pipes") == 0) {
324 mesh_control_pipes_recs(sh, &recs);
325 } else {
326 mesh_buf b = MESH_BUF_INIT;
327 mesh_buf_puts(&b, "view ");
328 mesh_rs_debug_cstr(&b, which);
329 mesh_buf_puts(&b, ": fleet, jobs or pipes [json]");
330 *err = mesh_buf_take(&b);
331 return -1;
332 }
333 mesh_views_render(out, &recs, json);
334 mesh_recs_free(&recs);
335 return 0;
336}
337
338void mesh_control_programs(mesh_shared *sh, mesh_buf *out) {
339 mesh_mailbox mb = sh->mb;
340 uint8_t *host = malloc(MESH_SNAP_HOST);
341 mesh_host_block_bytes(&sh->host, host);
342 for (uint32_t s = 0; s < MESH_PROG_SLOTS; s++) {
343 if (mesh_mailbox_get(mb, mesh_prog_entry(s) + MESH_PT_STATE) == MESH_ST_EMPTY) continue;
344 uint8_t e[MESH_PT_BYTES];
345 mesh_pt_read_entry(mb, s, e);
346 char name[MESH_NAME_BYTES + 1];
347 mesh_pt_name_of(mb, s, name);
348 float rate = rdf(host, MESH_S_RATES + 4 * (size_t)s);
349 uint64_t hash = rd64(e, MESH_PT_PTX_HASH);
350 uint32_t st = rd32(e, MESH_PT_STATE), pol = rd32(e, MESH_PT_POLICY);
351 size_t c = mesh_prog_ctl(s), p = mesh_pload(s);
352 float cpu = rdf(host, p + MESH_PL_CPU);
353 double rss = (double)rd64(host, p + MESH_PL_RSS) / 1048576.0;
354 uint32_t units = rd32(host, p + MESH_PL_UNITS);
355 mesh_buf kind = MESH_BUF_INIT;
356 uint32_t ex = mesh_mailbox_get(mb, mesh_prog_ext(s) + MESH_PX_EXEC);
357 if (ex == MESH_BACKEND_NATIVE) {
358 uint32_t exit_ = mesh_mailbox_get(mb, c + MESH_P_EXIT);
359 mesh_buf_printf(&kind, "kind native pid %u cpu ", mesh_mailbox_get(mb, c + MESH_P_PID));
360 mesh_rs_fixed(&kind, cpu, 2);
361 mesh_buf_puts(&kind, " rss_mb ");
362 mesh_rs_fixed(&kind, rss, 1);
363 mesh_buf_printf(&kind, " spawns %u crashes %u exit %u/%u limits ", mesh_mailbox_get(mb, c + MESH_P_SPAWNS),
364 mesh_mailbox_get(mb, c + MESH_P_CRASHES), exit_ >> 8, exit_ & 0xffu);
365 hexalt(&kind, mesh_mailbox_get(mb, c + MESH_P_LIMITS));
366 } else if (ex == MESH_BACKEND_WASM) {
367 mesh_buf_printf(&kind, "kind wasm lanes %u cpu ", units);
368 mesh_rs_fixed(&kind, cpu, 2);
369 mesh_buf_puts(&kind, " rss_mb ");
370 mesh_rs_fixed(&kind, rss, 1);
371 } else {
372 char *label = mesh_backend_label(mesh_mailbox_get(mb, MESH_H_BACKEND));
373 mesh_buf_printf(&kind, "kind %s warps %u shared %u reg_cap %u", label, rd32(e, MESH_PT_WARPS), mesh_mailbox_get(mb, mesh_prog_ext(s) + MESH_PX_SHARED),
374 mesh_mailbox_get(mb, mesh_prog_ext(s) + MESH_PX_REG_CAP));
375 free(label);
376 }
377 mesh_buf_printf(out, "%u %s %s %u %u %s ", s, name, st < MESH_STATES_LEN ? MESH_STATES[st] : "?", rd32(e, MESH_PT_NODES), rd32(e, MESH_PT_WARPS),
378 pol < MESH_POLICIES_LEN ? MESH_POLICIES[pol] : "?");
379 mesh_rs_fixed(out, rate, 1);
380 mesh_buf_printf(out, " %016llx %u ", (unsigned long long)hash, rd32(e, MESH_PT_REGS));
381 hexalt(out, rd32(e, MESH_PT_VIEW_OFF));
382 mesh_buf_putc(out, '+');
383 hexalt(out, rd32(e, MESH_PT_VIEW_BYTES));
384 mesh_buf_printf(out, " %s\n", mesh_buf_str(&kind));
385 mesh_buf_free(&kind);
386 uint8_t x[MESH_PX_BYTES];
387 mesh_mailbox_read(mb, mesh_prog_ext(s), x, MESH_PX_BYTES);
388 uint32_t nf = rd32(x, MESH_PX_FIELD_COUNT);
389 if (nf > MESH_MAX_FIELDS) nf = MESH_MAX_FIELDS;
390 mesh_buf_printf(out, "fields %s", name);
391 for (uint32_t i = 0; i < nf; i++) {
392 size_t f = MESH_PX_FIELDS + MESH_FIELD_BYTES * (size_t)i;
393 uint32_t ty = rd32(x, f + MESH_FD_TYPE);
394 mesh_buf_putc(out, ' ');
395 cstr(out, x + f + MESH_FD_NAME, MESH_FD_NAME_MAX + 1);
396 mesh_buf_printf(out, ":%s:", ty < MESH_FIELD_TYPES_LEN ? MESH_FIELD_TYPES[ty] : "?");
397 hexalt(out, rd32(x, f + MESH_FD_OFF));
398 }
399 mesh_buf_printf(out, "\ncounters %s", name);
400 for (size_t i = 0; i < MESH_COUNTERS; i++) {
401 const uint8_t *cn = x + MESH_PX_COUNTER_NAMES + MESH_NAME_BYTES * i;
402 mesh_buf_putc(out, ' ');
403 if (!cn[0]) mesh_buf_putc(out, '-');
404 else cstr(out, cn, MESH_NAME_BYTES);
405 }
406 mesh_buf_putc(out, '\n');
407 }
408 free(host);
409}
410
411void mesh_control_cluster(mesh_shared *sh, mesh_buf *out) {
412 uint8_t *h = malloc(MESH_SNAP_HOST);
413 mesh_host_block_bytes(&sh->host, h);
414 uint32_t ps = rd32(h, MESH_S_PEER_STATE), leader = rd32(h, MESH_S_LEADER);
415 static const char *const states[3] = {"single", "no-leader", "up"};
416 mesh_buf_printf(out, "cluster %u system %u state %s leader ", rd32(h, MESH_S_SYSTEMS), sh->me, ps < 3 ? states[ps] : "?");
417 if (leader == MESH_PEER_NONE) mesh_buf_putc(out, '-');
418 else mesh_buf_printf(out, "%u", leader);
419 mesh_buf_puts(out, " up ");
420 hexalt(out, rd32(h, MESH_S_UP_MASK));
421 mesh_buf_puts(out, " linked ");
422 hexalt(out, rd32(h, MESH_S_LINKED));
423 mesh_buf_puts(out, " lock_wait_us ");
424 mesh_rs_fixed(out, rdf(h, MESH_S_LOCK_WAIT_US), 1);
425 mesh_buf_putc(out, '\n');
426 if (!sh->peer) { free(h); return; }
427 for (uint32_t k = 0; k < sh->n; k++) {
428 size_t m = mesh_member(k), l = mesh_mload(k);
429 uint32_t cc = rd32(h, m + MESH_M_CC), lf = rd32(h, l + MESH_ML_FLAGS), st = rd32(h, m + MESH_M_STATE);
430 uint64_t at = rd64(h, l + MESH_ML_AT_MS);
431 mesh_buf_printf(out, "member %u %s rtt_us ", k, st < MESH_MEMBER_STATES_LEN ? MESH_MEMBER_STATES[st] : "?");
432 mesh_rs_fixed(out, rdf(h, m + MESH_M_RTT_US), 1);
433 mesh_buf_puts(out, " wait_us ");
434 mesh_rs_fixed(out, rdf(h, m + MESH_M_WAIT_US), 1);
435 char *label = mesh_backend_label(rd32(h, m + MESH_M_BACKEND));
436 mesh_buf_printf(out, " rounds %u round %llu cc %u.%u sms %u/%u blocks_per_sm %u backend %s arch ", rd32(h, m + MESH_M_ROUNDS),
437 (unsigned long long)rd64(h, m + MESH_M_ROUND), cc / 10, cc % 10, rd32(h, m + MESH_M_SMS), rd32(h, m + MESH_M_SMS_TOTAL),
438 rd32(h, m + MESH_M_BLOCKS_PER_SM), label);
439 free(label);
440 cstr(out, h + m + MESH_M_ARCH, 16);
441 mesh_buf_putc(out, ' ');
442 if (at == 0) mesh_buf_puts(out, "load_age_ms - gpu_share - cpu - psi_cpu - psi_mem - meshd_cpu - mem_avail_mb -");
443 else {
444 uint64_t up = mesh_host_block_uptime_ms(&sh->host);
445 mesh_buf_printf(out, "load_age_ms %llu gpu_share ", (unsigned long long)(up > at ? up - at : 0));
446 if (lf & MESH_LF_GPU) mesh_rs_fixed(out, rdf(h, l + MESH_ML_SHARE10), 3);
447 else mesh_buf_putc(out, '-');
448 mesh_buf_puts(out, " cpu ");
449 mesh_rs_fixed(out, rdf(h, l + MESH_ML_CPU), 3);
450 mesh_buf_puts(out, " psi_cpu ");
451 if (lf & MESH_LF_PSI) mesh_rs_fixed(out, rdf(h, l + MESH_ML_PSI_CPU), 4);
452 else mesh_buf_putc(out, '-');
453 mesh_buf_puts(out, " psi_mem ");
454 if (lf & MESH_LF_PSI) mesh_rs_fixed(out, rdf(h, l + MESH_ML_PSI_MEM), 4);
455 else mesh_buf_putc(out, '-');
456 mesh_buf_puts(out, " meshd_cpu ");
457 mesh_rs_fixed(out, rdf(h, l + MESH_ML_MESHD_CPU), 3);
458 mesh_buf_printf(out, " mem_avail_mb %u", rd32(h, l + MESH_ML_MEM_AVAIL_MB));
459 }
460 mesh_buf_puts(out, " host ");
461 cstr(out, h + m + MESH_M_HOST, 32);
462 mesh_buf_puts(out, " gpu ");
463 cstr(out, h + m + MESH_M_GPU, 32);
464 mesh_buf_putc(out, '\n');
465 }
466 free(h);
467}
468
469/* Read-only bytes of a program's view (from the table; any thread). */
470static int peek(mesh_shared *sh, const char *name, uint32_t off, uint32_t len, mesh_buf *out, char **err) {
471 uint32_t s;
472 if (!mesh_pt_find(sh->mb, name, &s)) {
473 char *q = debug_q(name);
474 *err = mesh_buf_fmt("no program %s", q);
475 free(q);
476 return -1;
477 }
478 uint8_t e[MESH_PT_BYTES];
479 mesh_pt_read_entry(sh->mb, s, e);
480 uint32_t view = rd32(e, MESH_PT_VIEW_OFF), bytes = rd32(e, MESH_PT_VIEW_BYTES);
481 if ((uint64_t)off + len > bytes || off % 4 != 0) {
482 *err = mesh_buf_fmt("%s: 0x%x+0x%x is outside its 0x%x-byte view (offsets 4-aligned)", name, off, len, bytes);
483 return -1;
484 }
485 size_t at = out->len;
486 mesh_buf_zeros(out, len);
487 mesh_mailbox_read(sh->mb, (size_t)view + off, (uint8_t *)out->data + at, len);
488 return 0;
489}
490
491/* A byte request through a program's request window (OP_REQ); framed reply. */
492static void request(mesh_shared *sh, const char *name, uint8_t *bytes, size_t n, mesh_buf *out) {
493 mesh_reply *r = mesh_reply_new();
494 mesh_cmd c = mesh_cmd_new(MESH_CMD_REQ);
495 c.name = mesh_xstrdup(name);
496 c.bytes = bytes;
497 c.len = n;
498 c.reply = mesh_reply_sender(r);
499 if (mesh_chan_send(sh->cmd, &c) != MESH_CHAN_OK) {
500 mesh_cmd_free(&c);
501 mesh_reply_release(r);
502 framed_str(out, "err driver gone\n");
503 return;
504 }
505 mesh_result x;
506 int w = mesh_reply_wait(r, TASK_MS, &x);
507 mesh_reply_release(r);
508 if (w != MESH_CHAN_OK) { framed_str(out, "err the request did not complete in 10s\n"); return; }
509 if (x.ok) framed(out, x.bytes ? x.bytes : (uint8_t *)"", x.len);
510 else framed_err(out, x.text ? x.text : "");
511 mesh_result_free(&x);
512}
513
514static void jobs_req(mesh_shared *sh, const char *cmd, const char *req, const char *first, mesh_buf *out) {
515 bool read = strcmp(cmd, "jobs") == 0 || (first && (strcmp(first, "status") == 0 || strcmp(first, "results") == 0));
516 mesh_reply *r = mesh_reply_new();
517 mesh_job_event e = mesh_job_event_new(MESH_JEV_CTL);
518 e.line = mesh_xstrdup(req);
519 e.reply_to = MESH_JOB_REPLY_LOCAL;
520 e.reply = mesh_reply_sender(r);
521 if (mesh_chan_send(sh->jobs, &e) != MESH_CHAN_OK) {
522 mesh_job_event_free(&e);
523 mesh_reply_release(r);
524 line(out, "err the jobs thread is gone");
525 return;
526 }
527 int64_t within = read ? TASK_MS : FORWARD_WAIT_MS + 5000;
528 mesh_result x;
529 int w = mesh_reply_wait(r, within, &x);
530 mesh_reply_release(r);
531 char *text;
532 bool ok;
533 if (w != MESH_CHAN_OK) {
534 mesh_buf b = MESH_BUF_INIT;
535 mesh_buf_puts(&b, "jobs: no answer in ");
536 mesh_rs_duration_ms(&b, (uint64_t)within);
537 text = mesh_buf_take(&b);
538 ok = false;
539 } else {
540 ok = x.ok;
541 text = x.text ? x.text : mesh_xstrdup("");
542 x.text = NULL;
543 mesh_result_free(&x);
544 }
545 if (read) {
546 if (ok) framed_str(out, text);
547 else framed_err(out, text);
548 } else reply(out, ok, text);
549 free(text);
550}
551
552static void trace(mesh_shared *sh, mesh_buf *out) {
553 mesh_buf b = MESH_BUF_INIT;
554 size_t recs = (size_t)MESH_TRACE_BYTES * MESH_TRACE_PER_PROG;
555 for (uint32_t s = 0; s < MESH_PROG_SLOTS; s++) {
556 if (mesh_mailbox_get(sh->mb, mesh_prog_entry(s) + MESH_PT_STATE) == MESH_ST_EMPTY) continue;
557 mesh_buf_u32le(&b, s);
558 mesh_buf_u32le(&b, mesh_mailbox_get(sh->mb, mesh_prog_ctl(s) + MESH_P_TRACE_N));
559 size_t at = b.len;
560 mesh_buf_zeros(&b, recs);
561 mesh_mailbox_read(sh->mb, mesh_trace_rec(s, 0), (uint8_t *)b.data + at, recs);
562 }
563 framed(out, b.data ? b.data : "", b.len);
564 mesh_buf_free(&b);
565}
566
567/* ---------------------------------------------------------------- dispatch */
568
569void mesh_control_handle(mesh_shared *sh, const char *req, mesh_control_body body, void *ctx, mesh_buf *out) {
570 const char *sp = strchr(req, ' ');
571 char *cmd = sp ? mesh_xstrndup(req, (size_t)(sp - req)) : mesh_xstrdup(req);
572 /* arg: what follows the first space, trimmed. */
573 const char *a = sp ? sp + 1 : "";
574 while (*a && is_ws(*a)) a++;
575 size_t an = strlen(a);
576 while (an && is_ws(a[an - 1])) an--;
577 char *arg = mesh_xstrndup(a, an);
578 char *wcopy = mesh_xstrdup(arg);
579 char *w[16];
580 size_t nw = split_ws(wcopy, w, 16);
581 bool peer = sh->peer != NULL;
582#define IS(x) (strcmp(cmd, (x)) == 0)
583#define W(i) ((i) < nw && (i) < 16 ? w[i] : NULL)
584 if (IS("snap")) {
585 mesh_buf s = MESH_BUF_INIT;
586 mesh_snapshot(&s, sh->mb, &sh->host);
587 framed(out, s.data, s.len);
588 mesh_buf_free(&s);
589 } else if (IS("programs")) {
590 mesh_buf t = MESH_BUF_INIT;
591 mesh_control_programs(sh, &t);
592 framed(out, t.data ? t.data : "", t.len);
593 mesh_buf_free(&t);
594 } else if (IS("pipes")) {
595 mesh_buf t = MESH_BUF_INIT;
596 mesh_control_pipes(sh, &t);
597 framed(out, t.data ? t.data : "", t.len);
598 mesh_buf_free(&t);
599 } else if (IS("view")) {
600 mesh_buf t = MESH_BUF_INIT;
601 char *err = NULL;
602 const char *w0 = nw > 0 ? w[0] : "", *w1 = nw > 1 ? w[1] : "";
603 if (mesh_control_view(sh, w0, strcmp(w1, "json") == 0, &t, &err) == 0) framed(out, t.data ? t.data : "", t.len);
604 else { framed_err(out, err); free(err); }
605 mesh_buf_free(&t);
606 } else if (IS("trace")) {
607 trace(sh, out);
608 } else if (IS("catalog")) {
609 char **names;
610 size_t n;
611 char *err = NULL;
612 if (mesh_catalog(sh->programs, &names, &n, &err)) {
613 mesh_buf t = MESH_BUF_INIT;
614 for (size_t i = 0; i < n; i++) { mesh_buf_printf(&t, "%s\n", names[i]); free(names[i]); }
615 free(names);
616 framed(out, t.data ? t.data : "", t.len);
617 mesh_buf_free(&t);
618 } else { framed_err(out, err); free(err); }
619 } else if ((IS("load") || IS("upgrade") || IS("rollback") || IS("unload") || IS("start") || IS("stop") || IS("reset") || IS("group")) && peer && nw) {
620 change(sh, req, out);
621 } else if (IS("train-reset") && peer) {
622 change(sh, "reset train", out);
623 } else if (IS("task") && peer && nw == 3 && lockstep(sh, w[0])) {
624 change(sh, req, out);
625 } else if (IS("place") && peer && nw >= 2) {
626 change(sh, req, out);
627 } else if (IS("fleet") && peer) {
628 char *t = mesh_shared_text(sh, &sh->fleet_text);
629 framed_str(out, t);
630 free(t);
631 } else if (IS("sched") && peer) {
632 if (nw == 0) {
633 char *t = mesh_shared_text(sh, &sh->sched_all);
634 framed_str(out, t);
635 free(t);
636 } else if (nw == 2 && strcmp(w[0], "explain") == 0) {
637 char *t = NULL;
638 pthread_mutex_lock(&sh->mu);
639 for (size_t i = 0; i < sh->nexplain; i++) if (strcmp(sh->sched_explain[i].name, w[1]) == 0) t = mesh_xstrdup(sh->sched_explain[i].text);
640 pthread_mutex_unlock(&sh->mu);
641 if (t) framed_str(out, t);
642 else {
643 char *q = debug_q(w[1]);
644 char *e = mesh_buf_fmt("no program %s in the desired state", q);
645 framed_err(out, e);
646 free(e);
647 free(q);
648 }
649 free(t);
650 } else if (nw == 1 && strcmp(w[0], "dry-run") == 0) {
651 char *t = mesh_shared_text(sh, &sh->sched_dry);
652 framed_str(out, t);
653 free(t);
654 } else if ((nw == 3 && strcmp(w[0], "pin") == 0) || (nw == 2 && strcmp(w[0], "unpin") == 0) || (nw == 1 && strcmp(w[0], "pause") == 0) ||
655 (nw == 1 && strcmp(w[0], "resume") == 0) ||
656 (nw == 2 && strcmp(w[0], "autoreform") == 0 && (strcmp(w[1], "on") == 0 || strcmp(w[1], "off") == 0)) ||
657 (nw == 2 && strcmp(w[0], "cordon") == 0) || (nw == 2 && strcmp(w[0], "uncordon") == 0)) {
658 change(sh, req, out);
659 } else {
660 line(out, "err usage: sched [explain <name> | dry-run | pin <name> <member[,member..]> | unpin <name> | pause | resume | autoreform on|off "
661 "| cordon <k> | uncordon <k>]");
662 }
663 } else if (IS("groups") && peer) {
664 char *t = mesh_shared_text(sh, &sh->groups_text);
665 framed_str(out, t);
666 free(t);
667 } else if (IS("leader") && peer) {
668 uint8_t *h = malloc(MESH_SNAP_HOST);
669 mesh_host_block_bytes(&sh->host, h);
670 uint32_t l = rd32(h, MESH_S_LEADER);
671 unsigned long long t = (unsigned long long)rd64(h, MESH_S_TERM);
672 free(h);
673 if (l == MESH_PEER_NONE) mesh_buf_printf(out, "err peer: no leader (term %llu): this member does not hear a majority\n", t);
674 else mesh_buf_printf(out, "ok system %u term %llu\n", l, t);
675 } else if (IS("fleet") || IS("groups") || IS("leader") || IS("group") || IS("sched")) {
676 line(out, "err not in a fleet (single meshd: --cluster / --peer)");
677 } else if (IS("jobs") || IS("job")) {
678 jobs_req(sh, cmd, req, W(0), out);
679 } else if (IS("infer")) {
680 mesh_infer_request r;
681 mesh_infer_msg res;
682 char *err = NULL;
683 if (mesh_infer_parse(arg, &r, &err) == 0 && mesh_inferd_run(sh, &r, &res, &err) == 0) {
684 mesh_buf t = MESH_BUF_INIT;
685 mesh_buf_puts(&t, "ok\n");
686 mesh_infer_text(&t, &res);
687 framed(out, t.data, t.len);
688 mesh_buf_free(&t);
689 mesh_infer_msg_free(&res);
690 } else {
691 framed_err(out, err);
692 free(err);
693 }
694 mesh_infer_request_free(&r);
695 } else if (IS("render")) {
696 uint8_t *b = NULL;
697 size_t n = 0;
698 char *err = NULL;
699 bool ok = mesh_control_render ? mesh_control_render(sh, w, nw < 16 ? nw : 16, &b, &n, &err) : (err = mesh_xstrdup("the blob thread is gone"), false);
700 if (ok) framed(out, b ? b : (uint8_t *)"", n);
701 else framed_err(out, err);
702 free(b);
703 free(err);
704 } else if (IS("load") && nw == 1) {
705 mesh_cmd c = mesh_cmd_new(MESH_CMD_LOAD);
706 c.has_src = true;
707 c.src = mesh_src_catalog(arg);
708 ask(sh, c, 2 * LIFECYCLE_MS, out);
709 } else if (IS("upgrade") && (nw == 1 || nw == 2)) {
710 mesh_cmd c = mesh_cmd_new(MESH_CMD_UPGRADE);
711 c.name = mesh_xstrdup(w[0]);
712 if (nw == 2) { c.has_src = true; c.src = mesh_src_catalog(w[1]); }
713 ask(sh, c, 2 * LIFECYCLE_MS, out);
714 } else if (IS("rollback") && nw == 1) {
715 mesh_cmd c = mesh_cmd_new(MESH_CMD_ROLLBACK);
716 c.name = mesh_xstrdup(arg);
717 ask(sh, c, 2 * LIFECYCLE_MS, out);
718 } else if (IS("unload") && (nw == 1 || (nw == 2 && strcmp(w[1], "--now") == 0))) {
719 mesh_cmd c = mesh_cmd_new(MESH_CMD_UNLOAD);
720 c.name = mesh_xstrdup(w[0]);
721 c.now = nw == 2;
722 ask(sh, c, 2 * LIFECYCLE_MS, out);
723 } else if (IS("reset") || IS("train-reset")) {
724 mesh_cmd c = mesh_cmd_new(MESH_CMD_RESET);
725 c.name = mesh_xstrdup(IS("train-reset") ? "train" : arg);
726 ask(sh, c, 2 * LIFECYCLE_MS, out);
727 } else if ((IS("start") || IS("stop")) && nw == 1) {
728 mesh_cmd c = mesh_cmd_new(MESH_CMD_PROG_RUN);
729 c.name = mesh_xstrdup(arg);
730 c.on = IS("start");
731 ask(sh, c, 2 * LIFECYCLE_MS, out);
732 } else if (IS("task")) {
733 uint32_t op, ta;
734 if (nw != 3 || !int_(w[1], &op) || !int_(w[2], &ta)) line(out, "err usage: task <name> <op> <arg>");
735 else {
736 mesh_reply *r = mesh_reply_new();
737 mesh_cmd c = mesh_cmd_new(MESH_CMD_TASK);
738 c.name = mesh_xstrdup(w[0]);
739 c.op = op;
740 c.arg = ta;
741 c.reply = mesh_reply_sender(r);
742 if (mesh_chan_send(sh->cmd, &c) != MESH_CHAN_OK) { mesh_cmd_free(&c); line(out, "err driver gone"); }
743 else {
744 mesh_result x;
745 int got = mesh_reply_wait(r, TASK_MS, &x);
746 if (got != MESH_CHAN_OK) line(out, "err the task did not complete in 10s");
747 else if (x.ok) mesh_buf_printf(out, "ok %u\n", x.value);
748 else mesh_buf_printf(out, "err %s\n", x.text ? x.text : "");
749 if (got == MESH_CHAN_OK) mesh_result_free(&x);
750 }
751 mesh_reply_release(r);
752 }
753 } else if (IS("req")) {
754 uint64_t len = 0;
755 if (nw != 2 || mesh_rs_parse_u64(w[1], strlen(w[1]), &len) != 0) framed_str(out, "err usage: req <name> <len>, then the bytes\n");
756 else if (len > REQ_MAX) framed_str(out, "err req: at most 1048576 bytes\n");
757 else {
758 uint8_t *bytes = mesh_xmalloc(len ? len : 1);
759 if (!body || !body(ctx, bytes, len)) { free(bytes); framed_str(out, "err req: short body\n"); }
760 else if (lockstep(sh, w[0])) {
761 free(bytes);
762 char *e = mesh_buf_fmt("%s: the lockstep program takes no requests in a fleet (they would not reach its group)", w[0]);
763 framed_err(out, e);
764 free(e);
765 } else request(sh, w[0], bytes, len, out);
766 }
767 } else if (IS("input") || IS("key")) {
768 const char *name, *bytes;
769 char *nm = NULL;
770 if (IS("key")) { name = "doom"; bytes = arg; }
771 else {
772 const char *s2 = strchr(arg, ' ');
773 if (s2) { nm = mesh_xstrndup(arg, (size_t)(s2 - arg)); name = nm; bytes = s2 + 1; }
774 else { name = arg; bytes = ""; }
775 }
776 if (!*bytes) mesh_buf_printf(out, "err usage: %s\n", IS("key") ? "key <k>" : "input <name> <bytes>");
777 else if (lockstep(sh, name))
778 mesh_buf_printf(out, "err %s: the lockstep program takes no input in a fleet (it would not reach its group)\n", name);
779 else {
780 mesh_cmd c = mesh_cmd_new(MESH_CMD_INPUT);
781 c.name = mesh_xstrdup(name);
782 c.len = strlen(bytes);
783 c.bytes = (uint8_t *)mesh_xstrdup(bytes);
784 ask(sh, c, TASK_MS, out);
785 }
786 free(nm);
787 } else if (IS("place")) {
788 if (!nw) line(out, "err usage: place <name> [local | pin <k> | split [auto | a,b,.. | a:n,b:m,..]]");
789 else {
790 mesh_cmd c = mesh_cmd_new(MESH_CMD_PLACE);
791 c.name = mesh_xstrdup(w[0]);
792 if (nw > 1) {
793 c.has_words = true;
794 c.nwords = nw - 1;
795 c.words = mesh_xcalloc(nw, sizeof *c.words);
796 for (size_t i = 1; i < nw && i < 16; i++) c.words[i - 1] = mesh_xstrdup(w[i]);
797 }
798 ask(sh, c, PLACE_MS, out);
799 }
800 } else if (IS("peek")) {
801 uint32_t off, len;
802 if (nw == 3 && int_(w[1], &off) && int_(w[2], &len)) {
803 mesh_buf t = MESH_BUF_INIT;
804 char *err = NULL;
805 if (peek(sh, w[0], off, len, &t, &err) == 0) framed(out, t.data ? t.data : "", t.len);
806 else { framed_err(out, err); free(err); }
807 mesh_buf_free(&t);
808 } else framed_str(out, "err usage: peek <name> <off> <len>\n");
809 } else if (IS("cluster")) {
810 mesh_buf t = MESH_BUF_INIT;
811 mesh_control_cluster(sh, &t);
812 framed(out, t.data, t.len);
813 mesh_buf_free(&t);
814 } else if (IS("load") && nw == 0) {
815 mesh_buf t = MESH_BUF_INIT;
816 mesh_load_text(sh, &t);
817 framed(out, t.data, t.len);
818 mesh_buf_free(&t);
819 } else if (IS("log")) {
820 uint64_t n = 50;
821 if (*arg && mesh_rs_parse_u64(arg, strlen(arg), &n) != 0) n = 50;
822 char **lines;
823 size_t k = mesh_log_recent((size_t)n, &lines);
824 mesh_buf t = MESH_BUF_INIT;
825 for (size_t i = 0; i < k; i++) { mesh_buf_printf(&t, "%s\n", lines[i]); free(lines[i]); }
826 free(lines);
827 framed(out, t.data ? t.data : "", t.len);
828 mesh_buf_free(&t);
829 } else if (IS("http")) {
830 const char *path = *arg ? arg : "/";
831 char *head = mesh_buf_fmt("GET %s HTTP/1.0\r\n\r\n", path);
832 uint8_t *resp;
833 size_t rn;
834 char *err = NULL;
835 if (mesh_shared_http(sh, (uint8_t *)head, strlen(head), &resp, &rn, &err)) { framed(out, resp, rn); free(resp); }
836 else {
837 mesh_error("control http %s: %s", path, err);
838 framed_err(out, err);
839 free(err);
840 }
841 free(head);
842 } else if (!*cmd) {
843 line(out, "err empty request");
844 } else {
845 char *q = debug_q(req);
846 mesh_info("control: unknown or malformed request %s", q);
847 mesh_buf_printf(out, "err unknown or malformed request %s\n", q);
848 free(q);
849 }
850#undef IS
851#undef W
852 free(wcopy);
853 free(arg);
854 free(cmd);
855}
856
857/* ---------------------------------------------------------------- the client loop */
858
859/* The client's socket, buffered, counting the bytes read from it (S_CTL_RX). */
860typedef struct {
861 int fd;
862 uint8_t buf[8192];
863 size_t a, b;
864 uint64_t n;
865} creader;
866
867static bool fill(creader *r) {
868 if (r->a < r->b) return true;
869 for (;;) {
870 ssize_t k = read(r->fd, r->buf, sizeof r->buf);
871 if (k < 0 && errno == EINTR) continue;
872 if (k <= 0) return false;
873 r->a = 0;
874 r->b = (size_t)k;
875 r->n += (uint64_t)k;
876 return true;
877 }
878}
879
880/* read_line under take(max): bytes up to and including '\n', at most max. -1: error or EOF with nothing. */
881static long read_line(creader *r, mesh_buf *l, size_t max) {
882 mesh_buf_clear(l);
883 while (l->len < max) {
884 if (!fill(r)) return l->len ? (long)l->len : -1;
885 uint8_t c = r->buf[r->a++];
886 mesh_buf_putc(l, (char)c);
887 if (c == '\n') break;
888 }
889 return (long)l->len;
890}
891
892static bool body_read(void *ctx, uint8_t *out, size_t n) {
893 creader *r = ctx;
894 size_t got = 0;
895 while (got < n) {
896 if (!fill(r)) return false;
897 size_t k = r->b - r->a < n - got ? r->b - r->a : n - got;
898 memcpy(out + got, r->buf + r->a, k);
899 r->a += k;
900 got += k;
901 }
902 return true;
903}
904
905static bool write_all(int fd, const char *p, size_t n) {
906 while (n) {
907 ssize_t w = write(fd, p, n);
908 if (w < 0 && errno == EINTR) continue;
909 if (w <= 0) return false;
910 p += w;
911 n -= (size_t)w;
912 }
913 return true;
914}
915
916static void timeouts(int fd, int64_t idle_ms) {
917 struct timeval rt = {(time_t)(idle_ms / 1000), (suseconds_t)(idle_ms % 1000) * 1000}, wt = {CLIENT_WRITE_MS / 1000, 0};
918 setsockopt(fd, SOL_SOCKET, SO_RCVTIMEO, &rt, sizeof rt);
919 setsockopt(fd, SOL_SOCKET, SO_SNDTIMEO, &wt, sizeof wt);
920 int one = 1;
921 setsockopt(fd, IPPROTO_TCP, TCP_NODELAY, &one, sizeof one);
922}
923
924void mesh_control_client(mesh_shared *sh, int fd, int64_t idle_ms) {
925 timeouts(fd, idle_ms);
926 creader *r = mesh_xcalloc(1, sizeof *r);
927 r->fd = fd;
928 mesh_buf l = MESH_BUF_INIT, out = MESH_BUF_INIT;
929 uint64_t counted = 0;
930 for (;;) {
931 long n = read_line(r, &l, LINE_MAX + 1);
932 if (n <= 0 || !mesh_rs_utf8_valid(l.data, l.len)) break;
933 if (l.data[l.len - 1] != '\n' && l.len > LINE_MAX) {
934 mesh_error("control: a request line over %d bytes; closing that client", LINE_MAX);
935 const char *e = "err request line over 4096 bytes\n";
936 if (!write_all(fd, e, strlen(e))) {}
937 break;
938 }
939 /* trim */
940 const char *s = l.data;
941 size_t k = l.len;
942 while (k && is_ws(*s)) { s++; k--; }
943 while (k && is_ws(s[k - 1])) k--;
944 char *req = mesh_xstrndup(s, k);
945 mesh_buf_clear(&out);
946 mesh_control_handle(sh, req, body_read, r, &out);
947 free(req);
948 bool ok = write_all(fd, out.data, out.len);
949 /* A request: the bytes its line and body took off the socket (read ahead included), its reply. */
950 mesh_net_counters_ctl(&sh->host.net, (size_t)(r->n - counted), ok ? out.len : 0);
951 counted = r->n;
952 if (!ok) break;
953 }
954 mesh_buf_free(&l);
955 mesh_buf_free(&out);
956 free(r);
957 close(fd);
958}
959
960typedef struct {
961 mesh_shared *sh;
962 int fd;
963} client_arg;
964
965static _Atomic int clients;
966
967static void *client_thread(void *p) {
968 client_arg *a = p;
969 mesh_control_client(a->sh, a->fd, CLIENT_IDLE_MS);
970 atomic_fetch_sub(&clients, 1);
971 free(a);
972 return NULL;
973}
974
975void mesh_control_serve(mesh_shared *sh, int listen_fd) {
976 for (;;) {
977 int c = accept(listen_fd, NULL, NULL);
978 if (c < 0) {
979 if (errno == EINTR) continue;
980 mesh_error("control accept: %s (os error %d)", strerror(errno), errno);
981 if (errno == EBADF || errno == EINVAL) return;
982 continue;
983 }
984 if (atomic_load(&clients) >= MAX_CLIENTS) {
985 mesh_error("control: %d clients connected already; refusing one", MAX_CLIENTS);
986 const char *e = "err too many control clients\n";
987 if (!write_all(c, e, strlen(e))) {}
988 close(c);
989 continue;
990 }
991 atomic_fetch_add(&clients, 1);
992 client_arg *a = mesh_xmalloc(sizeof *a);
993 a->sh = sh;
994 a->fd = c;
995 pthread_t th;
996 pthread_attr_t at;
997 pthread_attr_init(&at);
998 pthread_attr_setdetachstate(&at, PTHREAD_CREATE_DETACHED);
999 if (pthread_create(&th, &at, client_thread, a) != 0) {
1000 atomic_fetch_sub(&clients, 1);
1001 close(c);
1002 free(a);
1003 }
1004 pthread_attr_destroy(&at);
1005 }
1006}