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