==> src/proc.c <==1/* native.rs's Proc and Host: one worker process (its pipes, status, output
2 * tail, exit status, samples), and the processes of one meshd. */
3#include "native_internal.h"
4
5#include "mesh/abi/control_block.h"
6#include "mesh/abi/native.h"
7
8#include <errno.h>
9#include <signal.h>
10#include <stdio.h>
11#include <stdlib.h>
12#include <string.h>
13#include <sys/stat.h>
14#include <sys/wait.h>
15#include <time.h>
16#include <unistd.h>
17
18uint64_t mesh_native_now_ns(void) {
19 struct timespec t;
20 clock_gettime(CLOCK_MONOTONIC, &t);
21 return (uint64_t)t.tv_sec * 1000000000u + (uint64_t)t.tv_nsec;
22}
23
24void mesh_native_sleep_us(unsigned us) {
25 struct timespec t = {(time_t)(us / 1000000), (long)(us % 1000000) * 1000L};
26 nanosleep(&t, NULL);
27}
28
29void proc_ref(proc *p) { atomic_fetch_add(&p->refs, 1); }
30
31void proc_unref(proc *p) {
32 if (!p || atomic_fetch_sub(&p->refs, 1) != 1) return;
33 if (p->stdin_fd >= 0) close(p->stdin_fd);
34 for (size_t i = 0; i < p->tail_len; i++) free(p->tail[(p->tail_head + i) % MESH_NATIVE_TAIL]);
35 free(p->name);
36 pthread_mutex_destroy(&p->stdin_mu);
37 pthread_mutex_destroy(&p->tail_mu);
38 pthread_mutex_destroy(&p->status_mu);
39 pthread_mutex_destroy(&p->sample_mu);
40 free(p);
41}
42
43/* True when SIGPIPE is pending for this thread or the process. */
44static bool sigpipe_pending(void) {
45 sigset_t pending;
46 return sigpending(&pending) == 0 && sigismember(&pending, SIGPIPE) == 1;
47}
48
49/* A worker that has exited has closed its stdin, and writing to it raises SIGPIPE, which by
50 * default kills the whole process. Rust ignores SIGPIPE from startup, so the Rust executor only
51 * sees EPIPE; a library can't assume its host program does the same, so SIGPIPE is blocked
52 * around the write and one raised by this write is consumed before it is unblocked. */
53bool proc_send(proc *p, const char *line) {
54 pthread_mutex_lock(&p->stdin_mu);
55 bool ok = false;
56 if (p->stdin_fd >= 0) {
57 sigset_t pipe_set, old;
58 sigemptyset(&pipe_set);
59 sigaddset(&pipe_set, SIGPIPE);
60 pthread_sigmask(SIG_BLOCK, &pipe_set, &old);
61 bool was_pending = sigpipe_pending();
62 char *l = mesh_fmt("%s\n", line);
63 size_t n = strlen(l), off = 0;
64 ok = true;
65 while (off < n) {
66 ssize_t k = write(p->stdin_fd, l + off, n - off);
67 if (k > 0) off += (size_t)k;
68 else if (k < 0 && errno == EINTR) continue;
69 else {
70 if (k < 0 && errno == EPIPE && !was_pending) {
71 struct timespec zero = {0, 0};
72 while (sigtimedwait(&pipe_set, NULL, &zero) < 0 && errno == EINTR) {}
73 }
74 ok = false;
75 break;
76 }
77 }
78 free(l);
79 pthread_sigmask(SIG_SETMASK, &old, NULL);
80 }
81 pthread_mutex_unlock(&p->stdin_mu);
82 return ok;
83}
84
85void proc_tell_exit(proc *p) {
86 if (atomic_exchange(&p->exit_sent, true)) return;
87 proc_send(p, "exit");
88 pthread_mutex_lock(&p->stdin_mu);
89 if (p->stdin_fd >= 0) close(p->stdin_fd);
90 p->stdin_fd = -1;
91 pthread_mutex_unlock(&p->stdin_mu);
92}
93
94bool proc_exited(proc *p, int *status) {
95 pthread_mutex_lock(&p->status_mu);
96 if (!p->exited) {
97 int st = 0;
98 if (waitpid(p->pid, &st, WNOHANG) == p->pid) {
99 p->exited = true;
100 p->status = st;
101 }
102 }
103 bool e = p->exited;
104 if (e && status) *status = p->status;
105 pthread_mutex_unlock(&p->status_mu);
106 return e;
107}
108
109void proc_kill(proc *p) {
110 pthread_mutex_lock(&p->status_mu);
111 if (p->exited) {
112 pthread_mutex_unlock(&p->status_mu);
113 return;
114 }
115 kill(p->pid, SIGKILL);
116 uint64_t t0 = mesh_native_now_ns();
117 for (;;) {
118 int st = 0;
119 pid_t r = waitpid(p->pid, &st, WNOHANG);
120 if (r == p->pid) {
121 p->exited = true;
122 p->status = st;
123 pthread_mutex_unlock(&p->status_mu);
124 return;
125 }
126 if (r < 0 || mesh_native_now_ns() - t0 > KILL_REAP_MS * 1000000ull) break;
127 mesh_native_sleep_us(1000);
128 }
129 pthread_mutex_unlock(&p->status_mu);
130 mesh_error("native: %s worker %d not reaped %ums after SIGKILL; left behind", p->name, (int)p->pid, KILL_REAP_MS);
131}
132
133char *proc_tail_text(proc *p) {
134 pthread_mutex_lock(&p->tail_mu);
135 char *s;
136 if (p->tail_len == 0) s = mesh_strdup("(no output)");
137 else {
138 mesh_text t = {0};
139 for (size_t i = 0; i < p->tail_len; i++) {
140 if (i) mesh_text_puts(&t, " | ");
141 mesh_text_puts(&t, p->tail[(p->tail_head + i) % MESH_NATIVE_TAIL]);
142 }
143 s = mesh_text_take(&t);
144 }
145 pthread_mutex_unlock(&p->tail_mu);
146 return s;
147}
148
149static void tail_push(proc *p, const char *l) {
150 pthread_mutex_lock(&p->tail_mu);
151 if (p->tail_len == MESH_NATIVE_TAIL) {
152 free(p->tail[p->tail_head]);
153 p->tail[p->tail_head] = mesh_strdup(l);
154 p->tail_head = (p->tail_head + 1) % MESH_NATIVE_TAIL;
155 } else {
156 p->tail[(p->tail_head + p->tail_len++) % MESH_NATIVE_TAIL] = mesh_strdup(l);
157 }
158 pthread_mutex_unlock(&p->tail_mu);
159}
160
161void proc_sample(proc *p, uint64_t *cpu, uint64_t *rss, float *pct) {
162 pthread_mutex_lock(&p->sample_mu);
163 uint64_t c, r;
164 if (!proc_exited(p, NULL) && mesh_proc_usage(p->pid, &c, &r)) {
165 uint64_t now = mesh_native_now_ns();
166 double dt = (double)(now - p->sample_t) / 1e9;
167 float pc = dt > 0.05 ? (float)((double)(c > p->sample_cpu ? c - p->sample_cpu : 0) / 1e9 / dt * 100.0) : p->sample_pct;
168 p->sample_t = now;
169 p->sample_cpu = c;
170 p->sample_rss = r;
171 p->sample_pct = pc;
172 }
173 *cpu = p->sample_cpu;
174 *rss = p->sample_rss;
175 *pct = p->sample_pct;
176 pthread_mutex_unlock(&p->sample_mu);
177}
178
179void proc_last(proc *p, uint64_t *cpu, uint64_t *rss, float *pct) {
180 pthread_mutex_lock(&p->sample_mu);
181 *cpu = p->sample_cpu;
182 *rss = p->sample_rss;
183 *pct = p->sample_pct;
184 pthread_mutex_unlock(&p->sample_mu);
185}
186
187/* A status line from the worker: `ready <abi>` or `halted <gen>`. */
188void mesh_native_status_line(proc *p, const char *l) {
189 char w0[16];
190 unsigned long v;
191 char rest[8];
192 int n = sscanf(l, "%15s %lu %7s", w0, &v, rest);
193 if (n < 2 || v > UINT32_MAX) return;
194 if (strcmp(w0, "ready") == 0) atomic_store(&p->ready, v > 1 ? (uint32_t)v : 1u);
195 else if (strcmp(w0, "halted") == 0) atomic_store(&p->halted_gen, (uint32_t)v);
196}
197
198typedef struct {
199 proc *p;
200 int fd;
201 FILE *log;
202 pthread_mutex_t *log_mu;
203} reader;
204
205static void *status_reader(void *a) {
206 reader *r = a;
207 FILE *f = fdopen(r->fd, "r");
208 char *line = NULL;
209 size_t cap = 0;
210 ssize_t n;
211 while (f && (n = getline(&line, &cap, f)) > 0) {
212 if (line[n - 1] == '\n') line[n - 1] = 0;
213 mesh_native_status_line(r->p, line);
214 }
215 free(line);
216 if (f) fclose(f);
217 else close(r->fd);
218 atomic_store(&r->p->eof, true);
219 proc_unref(r->p);
220 free(r);
221 return NULL;
222}
223
224/* A shared, refcounted log file for a process's two output readers. */
225typedef struct {
226 atomic_int refs;
227 FILE *f;
228 pthread_mutex_t mu;
229} logfile;
230
231static void logfile_unref(logfile *l) {
232 if (!l || atomic_fetch_sub(&l->refs, 1) != 1) return;
233 if (l->f) fclose(l->f);
234 pthread_mutex_destroy(&l->mu);
235 free(l);
236}
237
238typedef struct {
239 proc *p;
240 int fd;
241 logfile *log;
242} out_reader;
243
244static void *output_reader(void *a) {
245 out_reader *r = a;
246 FILE *f = fdopen(r->fd, "r");
247 char *line = NULL;
248 size_t cap = 0;
249 ssize_t n;
250 while (f && (n = getline(&line, &cap, f)) > 0) {
251 if (line[n - 1] == '\n') line[--n] = 0;
252 if (r->log && r->log->f) {
253 char st[16];
254 mesh_log_stamp(st);
255 pthread_mutex_lock(&r->log->mu);
256 fprintf(r->log->f, "%s [%d] %s\n", st, (int)r->p->pid, line);
257 fflush(r->log->f);
258 pthread_mutex_unlock(&r->log->mu);
259 }
260 mesh_info("%s[%d]: %s", r->p->name, (int)r->p->pid, line);
261 tail_push(r->p, line);
262 }
263 free(line);
264 if (f) fclose(f);
265 else close(r->fd);
266 atomic_fetch_add(&r->p->outs_done, 1);
267 logfile_unref(r->log);
268 proc_unref(r->p);
269 free(r);
270 return NULL;
271}
272
273static bool start_thread(void *(*fn)(void *), void *arg) {
274 pthread_t t;
275 pthread_attr_t at;
276 pthread_attr_init(&at);
277 pthread_attr_setdetachstate(&at, PTHREAD_CREATE_DETACHED);
278 bool ok = pthread_create(&t, &at, fn, arg) == 0;
279 pthread_attr_destroy(&at);
280 return ok;
281}
282
283/* ---- the host -------------------------------------------------------------------------- */
284
285static pthread_mutex_t hosts_mu = PTHREAD_MUTEX_INITIALIZER;
286static mesh_native_host **hosts;
287static size_t nhosts;
288
289mesh_native_host *mesh_native_host_new(const mesh_shm *shm, const char *logs, const mesh_worker_cfg *cfg) {
290 mesh_native_host *h = mesh_alloc(sizeof *h);
291 h->mb = mesh_mailbox_new(shm->base, 0);
292 h->shm = *shm;
293 h->logs = mesh_strdup(logs);
294 h->cfg = *cfg;
295 size_t n = 0;
296 while (cfg->env && cfg->env[n]) n++;
297 h->cfg.env = mesh_alloc((n + 1) * sizeof *h->cfg.env);
298 for (size_t i = 0; i < n; i++) h->cfg.env[i] = mesh_strdup(cfg->env[i]);
299 h->cfg.mps_pipe_dir = cfg->mps_pipe_dir ? mesh_strdup(cfg->mps_pipe_dir) : NULL;
300 pthread_mutex_init(&h->mu, NULL);
301 pthread_mutex_lock(&hosts_mu);
302 hosts = mesh_realloc(hosts, (nhosts + 1) * sizeof *hosts);
303 hosts[nhosts++] = h;
304 pthread_mutex_unlock(&hosts_mu);
305 return h;
306}
307
308void host_forget(mesh_native_host *h, proc *p) {
309 pthread_mutex_lock(&h->mu);
310 for (size_t i = 0; i < h->nprocs; i++)
311 if (h->procs[i] == p) {
312 h->procs[i] = h->procs[--h->nprocs];
313 pthread_mutex_unlock(&h->mu);
314 proc_unref(p);
315 return;
316 }
317 pthread_mutex_unlock(&h->mu);
318}
319
320/* A snapshot of the host's processes, each with a reference. */
321static proc **host_procs(mesh_native_host *h, size_t *n) {
322 pthread_mutex_lock(&h->mu);
323 proc **v = mesh_alloc((h->nprocs ? h->nprocs : 1) * sizeof *v);
324 for (size_t i = 0; i < h->nprocs; i++) {
325 v[i] = h->procs[i];
326 proc_ref(v[i]);
327 }
328 *n = h->nprocs;
329 pthread_mutex_unlock(&h->mu);
330 return v;
331}
332
333void mesh_native_host_counts(mesh_native_host *h, uint32_t *live, uint32_t *spawns, uint32_t *crashes) {
334 size_t n;
335 proc **v = host_procs(h, &n);
336 uint32_t l = 0;
337 for (size_t i = 0; i < n; i++) {
338 if (!proc_exited(v[i], NULL)) l++;
339 proc_unref(v[i]);
340 }
341 free(v);
342 *live = l;
343 *spawns = atomic_load(&h->spawns);
344 *crashes = atomic_load(&h->crashes);
345}
346
347mesh_native_totals mesh_native_host_totals(mesh_native_host *h) {
348 size_t n;
349 proc **v = host_procs(h, &n);
350 mesh_native_totals t = {0, 0, 0, atomic_load(&h->spawns), atomic_load(&h->crashes)};
351 for (size_t i = 0; i < n; i++) {
352 if (!proc_exited(v[i], NULL)) {
353 uint64_t c, r;
354 float pct;
355 proc_sample(v[i], &c, &r, &pct);
356 t.procs++;
357 t.cpu_pct += pct;
358 t.rss += r;
359 }
360 proc_unref(v[i]);
361 }
362 free(v);
363 return t;
364}
365
366typedef struct {
367 proc *p;
368 mesh_native_host *h;
369} retiree;
370
371static void *retire_thread(void *a) {
372 retiree *r = a;
373 uint64_t t0 = mesh_native_now_ns();
374 while (!proc_exited(r->p, NULL)) {
375 if (mesh_native_now_ns() - t0 > (uint64_t)MESH_NATIVE_EXIT_GRACE_MS * 1000000u) {
376 mesh_error("native: %s worker %d did not exit in %us; killing it", r->p->name, (int)r->p->pid, MESH_NATIVE_EXIT_GRACE_MS / 1000);
377 proc_kill(r->p);
378 break;
379 }
380 mesh_native_sleep_us(2000);
381 }
382 host_forget(r->h, r->p);
383 proc_unref(r->p);
384 free(r);
385 return NULL;
386}
387
388/* Sends `exit`, then reaps it off the calling thread (killed after EXIT_GRACE). Takes p's reference. */
389void proc_retire(proc *p, mesh_native_host *h) {
390 proc_tell_exit(p);
391 retiree *r = mesh_alloc(sizeof *r);
392 r->p = p;
393 r->h = h;
394 if (!start_thread(retire_thread, r)) retire_thread(r);
395}
396
397static void close_all(const int *fds, size_t n) {
398 for (size_t i = 0; i < n; i++)
399 if (fds[i] >= 0) close(fds[i]);
400}
401
402bool host_spawn(mesh_native_host *h, spec *s, proc **out, char **err) {
403 uint64_t t0 = mesh_native_now_ns();
404 static atomic_uint seq;
405 unsigned n = atomic_fetch_add(&seq, 1);
406 const mesh_native_limits *l = &s->limits;
407 char *argv[24];
408 char *owned[8];
409 size_t na = 0, no = 0;
410 if (mesh_native_limits_scoped(l)) {
411 argv[na++] = "systemd-run";
412 argv[na++] = "--user";
413 argv[na++] = "--scope";
414 argv[na++] = "-q";
415 argv[na++] = owned[no++] = mesh_fmt("--unit=meshd-%d-%s-%u", (int)getpid(), s->name, n);
416 if (l->cpu_pct > 0) {
417 argv[na++] = "-p";
418 argv[na++] = owned[no++] = mesh_fmt("CPUQuota=%u%%", (unsigned)l->cpu_pct);
419 }
420 if (l->mem > 0) {
421 argv[na++] = "-p";
422 argv[na++] = owned[no++] = mesh_fmt("MemoryMax=%llu", (unsigned long long)l->mem);
423 argv[na++] = "-p";
424 argv[na++] = "MemorySwapMax=0";
425 }
426 argv[na++] = "--";
427 }
428 if (l->nice > 0) {
429 argv[na++] = "/usr/bin/nice";
430 argv[na++] = "-n";
431 argv[na++] = owned[no++] = mesh_fmt("%u", (unsigned)l->nice);
432 }
433 if (l->background) {
434#if defined(__APPLE__)
435 argv[na++] = "/usr/sbin/taskpolicy";
436 argv[na++] = "-b";
437#else
438 argv[na++] = "/usr/bin/chrt";
439 argv[na++] = "--idle";
440 argv[na++] = "0";
441#endif
442 }
443 argv[na++] = s->bin;
444 argv[na] = NULL;
445 size_t ne = 0;
446 while (h->cfg.env && h->cfg.env[ne]) ne++;
447 char **env = mesh_alloc((ne + 4) * sizeof *env);
448 size_t k = 0;
449 env[k++] = mesh_fmt("%s=%u", MESH_NATIVE_ENV_ABI, (unsigned)MESH_NATIVE_ABI);
450 env[k++] = mesh_fmt("%s=%s", MESH_NATIVE_ENV_NAME, s->name);
451 for (size_t i = 0; i < ne; i++) env[k++] = mesh_strdup(h->cfg.env[i]);
452 if (s->gpu && h->cfg.gpu == MESH_GPU_ACCESS_MPS && h->cfg.mps_pipe_dir)
453 env[k++] = mesh_fmt("CUDA_MPS_PIPE_DIRECTORY=%s", h->cfg.mps_pipe_dir);
454 env[k] = NULL;
455 int inp[2] = {-1, -1}, outp[2] = {-1, -1}, errp[2] = {-1, -1}, stp[2] = {-1, -1};
456 bool ok = mesh_pipe_cloexec(inp, err) && mesh_pipe_cloexec(outp, err) && mesh_pipe_cloexec(errp, err) && mesh_pipe_cloexec(stp, err);
457 pid_t pid = 0;
458 uint64_t ts = mesh_native_now_ns();
459 if (ok) {
460 mesh_spawn_fd fds[] = {{inp[0], 0}, {outp[1], 1}, {errp[1], 2}, {stp[1], MESH_NATIVE_STATUS_FD}, {h->shm.fd, MESH_NATIVE_MAILBOX_FD}};
461 char *why = NULL;
462 ok = mesh_spawn(argv, env, fds, 5, &pid, &why);
463 if (!ok) {
464 mesh_errf(err, "load: %s: cannot start %s: %s", s->name, s->bin, why);
465 free(why);
466 }
467 }
468 double spawn_ms = (double)(mesh_native_now_ns() - ts) / 1e6;
469 int child_ends[] = {inp[0], outp[1], errp[1], stp[1]};
470 close_all(child_ends, 4);
471 for (size_t i = 0; i < no; i++) free(owned[i]);
472 for (size_t i = 0; i < k; i++) free(env[i]);
473 free(env);
474 if (!ok) {
475 int parent_ends[] = {inp[1], outp[0], errp[0], stp[0]};
476 close_all(parent_ends, 4);
477 return false;
478 }
479 proc *p = mesh_alloc(sizeof *p);
480 atomic_init(&p->refs, 1);
481 p->pid = pid;
482 p->name = mesh_strdup(s->name);
483 p->stdin_fd = inp[1];
484 pthread_mutex_init(&p->stdin_mu, NULL);
485 pthread_mutex_init(&p->tail_mu, NULL);
486 pthread_mutex_init(&p->status_mu, NULL);
487 pthread_mutex_init(&p->sample_mu, NULL);
488 p->sample_t = mesh_native_now_ns();
489 p->limits = mesh_native_limits_bits(l);
490 reader *sr = mesh_alloc(sizeof *sr);
491 proc_ref(p);
492 *sr = (reader){p, stp[0], NULL, NULL};
493 if (!start_thread(status_reader, sr)) mesh_worker_fatal_thread();
494 mkdir(h->logs, 0755);
495 char *lp = mesh_fmt("%s/%s.log", h->logs, s->name);
496 logfile *lf = mesh_alloc(sizeof *lf);
497 atomic_init(&lf->refs, 2);
498 lf->f = fopen(lp, "a");
499 pthread_mutex_init(&lf->mu, NULL);
500 free(lp);
501 int outs[] = {outp[0], errp[0]};
502 for (int i = 0; i < 2; i++) {
503 out_reader *r = mesh_alloc(sizeof *r);
504 proc_ref(p);
505 *r = (out_reader){p, outs[i], lf};
506 if (!start_thread(output_reader, r)) mesh_worker_fatal_thread();
507 }
508 pthread_mutex_lock(&h->mu);
509 h->procs = mesh_realloc(h->procs, (h->nprocs + 1) * sizeof *h->procs);
510 proc_ref(p);
511 h->procs[h->nprocs++] = p;
512 pthread_mutex_unlock(&h->mu);
513 atomic_fetch_add(&h->spawns, 1);
514 atomic_fetch_add(&mesh_cpuuse_worker_spawns, 1);
515 atomic_fetch_add(&s->spawns, 1);
516 /* Up to `ready` (its setup done), or its death, or READY. */
517 bool good = false;
518 for (;;) {
519 uint32_t abi = atomic_load(&p->ready);
520 int st;
521 if (abi != 0) {
522 if (abi != MESH_NATIVE_ABI)
523 mesh_errf(err, "abi: %s speaks native worker ABI %u, meshd %u (rebuild it)", s->name, (unsigned)abi, (unsigned)MESH_NATIVE_ABI);
524 else good = true;
525 break;
526 }
527 uint64_t el = mesh_native_now_ns() - t0;
528 if (proc_exited(p, &st)) {
529 /* Its last output may still be in the pipes: give the readers a moment. */
530 for (int i = 0; i < 100 && atomic_load(&p->outs_done) < 2; i++) mesh_native_sleep_us(1000);
531 char *d = mesh_describe_status(st), *tail = proc_tail_text(p);
532 mesh_errf(err, "load: %s worker %d exited before it was ready (%s); output: %s", s->name, (int)pid, d, tail);
533 free(d);
534 free(tail);
535 break;
536 }
537 if (atomic_load(&p->eof) && el > 200000000u && !proc_exited(p, NULL)) {
538 mesh_errf(err, "load: %s worker %d closed its status pipe without `ready` (not a mesh-worker program?)", s->name, (int)pid);
539 break;
540 }
541 if (el > (uint64_t)MESH_NATIVE_READY_MS * 1000000u) {
542 char *tail = proc_tail_text(p);
543 mesh_errf(err, "load: %s worker %d not ready in %us; output: %s", s->name, (int)pid, MESH_NATIVE_READY_MS / 1000, tail);
544 free(tail);
545 break;
546 }
547 mesh_native_sleep_us(200);
548 }
549#if defined(__linux__)
550 if (good && mesh_native_limits_scoped(l)) good = mesh_native_check_cgroup(pid, l, err);
551#endif
552 if (good && s->gpu && h->cfg.gpu == MESH_GPU_ACCESS_MPS && h->cfg.check_gpu_client) {
553 char *why = NULL;
554 good = h->cfg.check_gpu_client(pid, &why);
555 if (!good) mesh_errf(err, "load: %s: device gpu: %s", s->name, why);
556 free(why);
557 }
558 if (!good) {
559 proc_kill(p);
560 host_forget(h, p);
561 proc_unref(p);
562 return false;
563 }
564 mesh_native_limits none = {0, 0, 0, false};
565 char *lim = mesh_native_limits_eq(l, &none) ? mesh_strdup("") : NULL;
566 if (!lim) {
567 char *d = mesh_native_limits_debug(l);
568 lim = mesh_fmt("; limits %s", d);
569 free(d);
570 }
571 mesh_info("native: %s worker %d ready in %.1f ms (spawn %.2f ms; %s%s)", s->name, (int)pid, (double)(mesh_native_now_ns() - t0) / 1e6,
572 spawn_ms, s->bin, lim);
573 free(lim);
574 *out = p;
575 return true;
576}
577
578void mesh_native_cleanup(void) {
579 pthread_mutex_lock(&hosts_mu);
580 proc **all = NULL;
581 size_t n = 0;
582 for (size_t i = 0; i < nhosts; i++) {
583 mesh_native_host *h = hosts[i];
584 pthread_mutex_lock(&h->mu);
585 all = mesh_realloc(all, (n + h->nprocs + 1) * sizeof *all);
586 for (size_t j = 0; j < h->nprocs; j++) all[n++] = h->procs[j];
587 h->nprocs = 0;
588 pthread_mutex_unlock(&h->mu);
589 }
590 pthread_mutex_unlock(&hosts_mu);
591 for (size_t i = 0; i < n; i++) proc_tell_exit(all[i]);
592 uint64_t t0 = mesh_native_now_ns();
593 for (;;) {
594 bool any = false;
595 for (size_t i = 0; i < n; i++) any = any || !proc_exited(all[i], NULL);
596 if (!any || mesh_native_now_ns() - t0 > 2000000000u) break;
597 mesh_native_sleep_us(2000);
598 }
599 for (size_t i = 0; i < n; i++)
600 if (!proc_exited(all[i], NULL)) {
601 mesh_error("native: %s worker %d still there at exit; killing it", all[i]->name, (int)all[i]->pid);
602 proc_kill(all[i]);
603 }
604 if (n) mesh_info("native: %zu worker process(es) gone", n);
605 for (size_t i = 0; i < n; i++) proc_unref(all[i]);
606 free(all);
607}