==> src/daemon.c <==1/* mps.rs's daemon half: our private nvidia-cuda-mps-control, its requests,
2 * stale daemons of dead meshds, and the exit cleanup of workers and daemon. */
3#include "mps_internal.h"
4
5#include <dirent.h>
6#include <errno.h>
7#include <fcntl.h>
8#include <signal.h>
9#include <stdio.h>
10#include <stdlib.h>
11#include <string.h>
12#include <sys/stat.h>
13#include <sys/wait.h>
14#include <time.h>
15#include <unistd.h>
16
17char *mps_os_error(int e) { return mesh_fmt("%s (os error %d)", strerror(e), e); }
18
19uint64_t mps_now_ns(void) {
20 struct timespec t;
21 clock_gettime(CLOCK_MONOTONIC, &t);
22 return (uint64_t)t.tv_sec * 1000000000u + (uint64_t)t.tv_nsec;
23}
24
25void mps_sleep_us(unsigned us) {
26 struct timespec t = {(time_t)(us / 1000000), (long)(us % 1000000) * 1000L};
27 nanosleep(&t, NULL);
28}
29
30/* Rust's ExitStatus Display: "exit status: 1", "signal: 9 (SIGKILL)". */
31char *mps_status_display(int st) {
32 if (WIFEXITED(st)) return mesh_fmt("exit status: %d", WEXITSTATUS(st));
33 if (WIFSIGNALED(st)) return mesh_fmt("signal: %d (SIG%s)", WTERMSIG(st), sigabbrev_np(WTERMSIG(st)) ? sigabbrev_np(WTERMSIG(st)) : "?");
34 return mesh_fmt("wait status: %d", st);
35}
36
37static bool mkdir_private(const char *p, char **err) {
38 /* create_dir_all */
39 char *tmp = mesh_strdup(p);
40 for (char *s = tmp + 1; *s; s++)
41 if (*s == '/') {
42 *s = 0;
43 mkdir(tmp, 0755);
44 *s = '/';
45 }
46 if (mkdir(tmp, 0700) != 0 && errno != EEXIST) {
47 char *e = mps_os_error(errno);
48 mesh_errf(err, "mps: %s: %s", p, e);
49 free(e);
50 free(tmp);
51 return false;
52 }
53 free(tmp);
54 if (chmod(p, 0700) != 0) {
55 char *e = mps_os_error(errno);
56 mesh_errf(err, "mps: chmod %s: %s", p, e);
57 free(e);
58 return false;
59 }
60 return true;
61}
62
63static void read_all(int fd, mesh_text *t) {
64 char buf[4096];
65 ssize_t n;
66 while ((n = read(fd, buf, sizeof buf)) > 0) mesh_text_putn(t, buf, (size_t)n);
67}
68
69/* The env of a request to our daemon. */
70static char **daemon_env(const char *dir) {
71 char **env = mesh_alloc(3 * sizeof *env);
72 env[0] = mesh_fmt("%s=%s/pipe", MESH_MPS_PIPE_ENV, dir);
73 env[1] = mesh_fmt("%s=%s/log", MESH_MPS_LOG_ENV, dir);
74 return env;
75}
76
77static void free_env(char **env) {
78 for (size_t i = 0; env[i]; i++) free(env[i]);
79 free(env);
80}
81
82char *mps_control(const char *dir, const char *req, char **err) {
83 int in[2], out[2], er[2];
84 if (!mesh_pipe_cloexec(in, err) || !mesh_pipe_cloexec(out, err) || !mesh_pipe_cloexec(er, err)) return NULL;
85 char **env = daemon_env(dir);
86 char *argv[] = {MESH_MPS_CONTROL, NULL};
87 mesh_spawn_fd fds[] = {{in[0], 0}, {out[1], 1}, {er[1], 2}};
88 pid_t pid;
89 char *why = NULL;
90 bool ok = mesh_spawn(argv, env, fds, 3, &pid, &why);
91 free_env(env);
92 close(in[0]);
93 close(out[1]);
94 close(er[1]);
95 if (!ok) {
96 /* "posix_spawn nvidia-cuda-mps-control: No such file..." -> the io error part */
97 const char *e = strstr(why, ": ");
98 mesh_errf(err, "mps: %s: %s", MESH_MPS_CONTROL, e ? e + 2 : why);
99 free(why);
100 close(in[1]);
101 close(out[0]);
102 close(er[0]);
103 return NULL;
104 }
105 char *line = mesh_fmt("%s\n", req);
106 ssize_t w = write(in[1], line, strlen(line));
107 (void)w;
108 free(line);
109 close(in[1]);
110 uint64_t t0 = mps_now_ns();
111 int st = 0;
112 for (;;) {
113 pid_t r = waitpid(pid, &st, WNOHANG);
114 if (r == pid) break;
115 if (r < 0 || mps_now_ns() - t0 > 5000000000u) {
116 kill(pid, SIGKILL);
117 waitpid(pid, &st, 0);
118 close(out[0]);
119 close(er[0]);
120 mesh_errf(err, "mps: `%s` got no answer in 5 s", req);
121 return NULL;
122 }
123 mps_sleep_us(1000);
124 }
125 mesh_text o = {0}, e = {0};
126 read_all(out[0], &o);
127 read_all(er[0], &e);
128 close(out[0]);
129 close(er[0]);
130 if (!(WIFEXITED(st) && WEXITSTATUS(st) == 0)) {
131 char *d = mps_status_display(st);
132 const char *s = mesh_text_str(&e);
133 size_t n = e.len;
134 mesh_trim(&s, &n);
135 mesh_errf(err, "mps: `%s`: %s %.*s", req, d, (int)n, s);
136 free(d);
137 mesh_text_free(&o);
138 mesh_text_free(&e);
139 return NULL;
140 }
141 const char *s = mesh_text_str(&o);
142 size_t n = o.len;
143 mesh_trim(&s, &n);
144 char *r = mesh_strndup(s, n);
145 mesh_text_free(&o);
146 mesh_text_free(&e);
147 return r;
148}
149
150pid_t mps_daemon_pid(const char *dir) {
151 char *p = mesh_fmt("%s/pipe/nvidia-cuda-mps-control.pid", dir);
152 FILE *f = fopen(p, "r");
153 free(p);
154 if (!f) return 0;
155 long v = 0;
156 int n = fscanf(f, "%ld", &v);
157 fclose(f);
158 return n == 1 && v > 0 ? (pid_t)v : 0;
159}
160
161static bool exists(const char *p) {
162 struct stat st;
163 return stat(p, &st) == 0;
164}
165
166static void rm_rf(const char *p) {
167 DIR *d = opendir(p);
168 if (d) {
169 struct dirent *e;
170 while ((e = readdir(d))) {
171 if (strcmp(e->d_name, ".") == 0 || strcmp(e->d_name, "..") == 0) continue;
172 char *c = mesh_fmt("%s/%s", p, e->d_name);
173 struct stat st;
174 if (lstat(c, &st) == 0 && S_ISDIR(st.st_mode)) rm_rf(c);
175 else unlink(c);
176 free(c);
177 }
178 closedir(d);
179 }
180 rmdir(p);
181}
182
183/* A meshd killed with SIGKILL cannot quit its daemon: such daemons under the
184 * default dirs, whose meshd is gone, are quit here. */
185static void reap_stale(void) {
186 DIR *d = opendir("/tmp");
187 if (!d) return;
188 struct dirent *e;
189 while ((e = readdir(d))) {
190 if (strncmp(e->d_name, "meshd-mps.", 10) != 0) continue;
191 char *end;
192 long pid = strtol(e->d_name + 10, &end, 10);
193 if (*end || pid <= 0) continue;
194 char *comm = mesh_fmt("/proc/%ld/comm", pid), buf[64] = {0};
195 FILE *f = fopen(comm, "r");
196 free(comm);
197 bool alive = false;
198 if (f) {
199 if (fgets(buf, sizeof buf, f)) alive = strncmp(buf, "meshd", 5) == 0 && (buf[5] == '\n' || buf[5] == 0);
200 fclose(f);
201 }
202 char *dir = mesh_fmt("/tmp/%s", e->d_name), *ctl = mesh_fmt("%s/pipe/control", dir);
203 bool has_ctl = exists(ctl);
204 free(ctl);
205 if (alive || !has_ctl) {
206 free(dir);
207 continue;
208 }
209 pid_t dp = mps_daemon_pid(dir);
210 if (dp && kill(dp, 0) == 0) {
211 char *err = NULL, *r = mps_control(dir, "quit", &err);
212 if (r) mesh_info("mps: quit the daemon %d of meshd %ld, which is gone (%s)", (int)dp, pid, dir);
213 else {
214 mesh_error("mps: stale daemon %d in %s: %s", (int)dp, dir, err);
215 free(err);
216 free(dir);
217 continue;
218 }
219 free(r);
220 } else {
221 mesh_info("mps: removing %s: its meshd and daemon are gone", dir);
222 }
223 rm_rf(dir);
224 free(dir);
225 }
226 closedir(d);
227}
228
229bool mps_start_daemon(const char *dir, bool temporary, char **err) {
230 reap_stale();
231 char *pipe_dir = mesh_fmt("%s/pipe", dir), *log_dir = mesh_fmt("%s/log", dir), *ctl = mesh_fmt("%s/pipe/control", dir);
232 bool ok = mkdir_private(dir, err) && mkdir_private(pipe_dir, err) && mkdir_private(log_dir, err);
233 if (ok && exists(ctl)) ok = mesh_errf(err, "mps: %s already has a control pipe (another daemon?); refusing", dir);
234 if (ok) {
235 char **env = daemon_env(dir);
236 char *argv[] = {MESH_MPS_CONTROL, "-d", NULL};
237 int devnull = open("/dev/null", O_RDWR | O_CLOEXEC);
238 mesh_spawn_fd fds[] = {{devnull, 0}, {devnull, 1}, {2, 2}};
239 pid_t pid;
240 char *why = NULL;
241 ok = mesh_spawn(argv, env, fds, 3, &pid, &why);
242 free_env(env);
243 close(devnull);
244 if (!ok) {
245 const char *e = strstr(why, ": ");
246 mesh_errf(err, "mps: cannot run %s: %s; refusing (--isolation mps needs it)", MESH_MPS_CONTROL, e ? e + 2 : why);
247 free(why);
248 } else {
249 int st;
250 waitpid(pid, &st, 0);
251 if (!(WIFEXITED(st) && WEXITSTATUS(st) == 0)) {
252 char *d = mps_status_display(st);
253 ok = mesh_errf(err, "mps: %s -d: %s", MESH_MPS_CONTROL, d);
254 free(d);
255 }
256 }
257 }
258 if (ok) {
259 mps_set_daemon(dir, temporary);
260 uint64_t t0 = mps_now_ns();
261 while (!exists(ctl)) {
262 if (mps_now_ns() - t0 > 5000000000u) {
263 ok = mesh_errf(err, "mps: no control pipe in %s after 5 s", dir);
264 break;
265 }
266 mps_sleep_us(5000);
267 }
268 }
269 if (ok) {
270 char *servers = mps_control(dir, "get_server_list", err);
271 if (!servers) ok = false;
272 else {
273 char *q = mesh_debug_str(servers[0] ? servers : "none yet");
274 mesh_info("mps: private control daemon (pid %d) on %s; servers %s", (int)mps_daemon_pid(dir), dir, q);
275 free(q);
276 free(servers);
277 }
278 }
279 free(pipe_dir);
280 free(log_dir);
281 free(ctl);
282 return ok;
283}
284
285void mps_quit_daemon(const char *dir, bool temporary) {
286 pid_t pid = mps_daemon_pid(dir);
287 char *err = NULL, *r = mps_control(dir, "quit", &err);
288 if (!r) {
289 mesh_error("mps: quit: %s", err);
290 free(err);
291 }
292 free(r);
293 uint64_t t0 = mps_now_ns();
294 while (pid && kill(pid, 0) == 0) {
295 if (mps_now_ns() - t0 > 5000000000u) {
296 mesh_error("mps: control daemon %d still running 5 s after quit", (int)pid);
297 return;
298 }
299 mps_sleep_us(5000);
300 }
301 if (temporary) rm_rf(dir);
302 else {
303 char *p = mesh_fmt("%s/pipe", dir);
304 rm_rf(p);
305 free(p);
306 }
307 if (temporary) mesh_info("mps: control daemon quit");
308 else mesh_info("mps: control daemon quit (logs in %s/log)", dir);
309}