==> src/meshd.c <==1/* meshd (main.rs): the host daemon of the GPU mesh. It owns the GPU backend (CUDA: the context, the
2 * mailbox and every program's resident kernel), the CPU executors beside it (wasm lanes, native
3 * worker processes), and the driver, which publishes tasks and runs the program lifecycle on the main
4 * thread; the control, HTTP and sampler threads only talk to the driver (commands) or read the
5 * mailbox. */
6#include "meshd.h"
7
8#include "mesh/abi.h"
9#include "mesh/args.h"
10#include "mesh/backend.h"
11#include "mesh/buf.h"
12#include "mesh/control.h"
13#include "mesh/cuda.h"
14#include "mesh/driver.h"
15#include "mesh/exec_wasm.h"
16#include "mesh/http_relay.h"
17#include "mesh/infer_svc.h"
18#include "mesh/log.h"
19#include "mesh/mps.h"
20#include "mesh/mps_worker.h"
21#include "mesh/native.h"
22#include "mesh/nvjitlink.h"
23#include "mesh/rsfmt.h"
24#include "mesh/samplers.h"
25#include "mesh/shared.h"
26#include "mesh/shm.h"
27
28#include <arpa/inet.h>
29#include <errno.h>
30#include <limits.h>
31#include <netdb.h>
32#include <netinet/in.h>
33#include <pthread.h>
34#include <signal.h>
35#include <stdatomic.h>
36#include <stdio.h>
37#include <stdlib.h>
38#include <string.h>
39#include <sys/socket.h>
40#include <sys/stat.h>
41#include <unistd.h>
42
43static _Atomic bool QUIT;
44
45#if defined(__x86_64__)
46#define ARCH "x86_64"
47#elif defined(__aarch64__)
48#define ARCH "aarch64"
49#else
50#define ARCH "unknown"
51#endif
52
53static void on_signal(int s) {
54 (void)s;
55 atomic_store(&QUIT, true);
56}
57
58static void *cleanup_thread(void *done) {
59 mesh_backend_cleanup();
60 atomic_store((_Atomic bool *)done, true);
61 return NULL;
62}
63
64/* Waits for QUIT, then the limit; if the process is still there, logs it and ends it (_exit(3)). */
65static void *watchdog(void *arg) {
66 (void)arg;
67 while (!atomic_load(&QUIT)) mesh_sleep_us(20000);
68 mesh_sleep_us(10000000);
69 mesh_error("shutdown: still running 10s after the signal; exiting at once (watchdog)");
70 /* Cleanup is bounded, but the watchdog does not depend on it: at most a second more. */
71 static _Atomic bool done;
72 pthread_t t;
73 if (pthread_create(&t, NULL, cleanup_thread, &done) == 0) pthread_detach(t);
74 for (int i = 0; i < 100 && !atomic_load(&done); i++) mesh_sleep_us(10000);
75 _exit(3);
76}
77
78static void install_signals(void) {
79 struct sigaction sa;
80 memset(&sa, 0, sizeof sa);
81 sa.sa_handler = on_signal;
82 sigemptyset(&sa.sa_mask);
83 sigaction(SIGINT, &sa, NULL);
84 sigaction(SIGTERM, &sa, NULL);
85 signal(SIGPIPE, SIG_IGN);
86 pthread_t t;
87 if (pthread_create(&t, NULL, watchdog, NULL) != 0) { mesh_error("the watchdog thread"); abort(); }
88 pthread_detach(t);
89}
90
91/* Every exit after the MPS daemon may have started quits it (a no-op in blink mode). */
92static _Noreturn void fail(char *e) {
93 mesh_error("%s", e);
94 free(e);
95 mesh_backend_cleanup();
96 mesh_log_exit(1);
97}
98
99/* host:port to a listening TCP socket: the fd, or -1 with *err (std's words). */
100static int listen_on(const char *addr, char **err) {
101 const char *c = strrchr(addr, ':');
102 if (!c) { *err = mesh_xstrdup("invalid socket address"); return -1; }
103 char host[256];
104 snprintf(host, sizeof host, "%.*s", (int)(c - addr), addr);
105 struct addrinfo hints, *res = NULL;
106 memset(&hints, 0, sizeof hints);
107 hints.ai_family = AF_INET;
108 hints.ai_socktype = SOCK_STREAM;
109 hints.ai_flags = AI_NUMERICSERV;
110 int g = getaddrinfo(host, c + 1, &hints, &res);
111 if (g != 0) { *err = mesh_buf_fmt("%s", gai_strerror(g)); return -1; }
112 int fd = socket(AF_INET, SOCK_STREAM | SOCK_CLOEXEC, 0);
113 int one = 1;
114 setsockopt(fd, SOL_SOCKET, SO_REUSEADDR, &one, sizeof one);
115 if (bind(fd, res->ai_addr, res->ai_addrlen) != 0 || listen(fd, 128) != 0) {
116 int e = errno;
117 *err = mesh_buf_fmt("%s (os error %d)", strerror(e), e);
118 close(fd);
119 freeaddrinfo(res);
120 return -1;
121 }
122 freeaddrinfo(res);
123 return fd;
124}
125
126static int bind_or_exit(const char *what, const char *addr) {
127 char *err = NULL;
128 int fd = listen_on(addr, &err);
129 if (fd < 0) {
130 mesh_error("%s: cannot bind %s: %s", what, addr, err);
131 free(err);
132 mesh_log_exit(2);
133 }
134 mesh_info("%s on %s", what, addr);
135 return fd;
136}
137
138typedef struct {
139 mesh_shared *sh;
140 int fd;
141} serve_arg;
142
143static void *control_thread(void *p) { serve_arg *a = p; mesh_control_serve(a->sh, a->fd); return NULL; }
144static void *http_thread(void *p) { serve_arg *a = p; mesh_http_serve(a->sh, a->fd); return NULL; }
145static void *net_thread(void *p) { mesh_net_sampler(p); return NULL; }
146static void *load_thread(void *p) { mesh_load_sampler(p); return NULL; }
147
148static void spawn(void *(*fn)(void *), void *arg) {
149 pthread_t t;
150 if (pthread_create(&t, NULL, fn, arg) != 0) { mesh_error("cannot start a thread: %s", strerror(errno)); abort(); }
151 pthread_detach(t);
152}
153
154static const char *keys_get(void *ctx, const char *k) { return ms_get(ctx, k); }
155
156/* The CUDA backend: nvJitLink, then the context (blink) or the MPS daemon and workers. */
157static mesh_backend *open_cuda(const mesh_args *a, const mesh_shm *shm, char **err) {
158 /* The program loader's JIT (no fallback to the driver's): refuse to start without it. */
159 mesh_nvjitlink *jit = mesh_xcalloc(1, sizeof *jit);
160 if (!mesh_nvjitlink_open(a->nvjitlink, jit, err)) { free(jit); return NULL; }
161 mesh_info("nvJitLink %u.%u from %s; module loading EAGER", jit->major, jit->minor, jit->path);
162 mesh_backend *b = NULL;
163 if (!mesh_cuda_backend_open(a->isolation, a->has_sms ? a->sms : 0, a->mps_dir, a->has_mps_pool ? a->mps_pool : 8, shm, jit, &b, err)) return NULL;
164 const mesh_caps *c = b->ops->caps(b);
165 bool mps = mesh_cuda_backend_mps(b);
166 mesh_info("GPU %s sm_%d%d, %u SMs, %u blocks/SM max, host-native atomics %s; programs run on %u SMs%s%s", c->name, c->cc_major, c->cc_minor,
167 c->sms_total, c->blocks_per_sm,
168 mesh_cuda_backend_host_atomics(b) ? "yes" : "no (each pipe inbox has a GPU-side ring head of its own)", c->sms,
169 c->green ? " (green context)" : " (primary context)",
170 mps ? "; isolation mps: one worker process per image" : "; isolation blink: one context");
171 char *lib = mesh_cuda_check_lib(err);
172 if (!lib) return NULL;
173 mesh_info("CUDA driver library %s", lib);
174 free(lib);
175 return b;
176}
177
178_Noreturn int meshd_main(int argc, char **argv) {
179 /* argv without the program name. */
180 int n = argc - 1;
181 char **av = argv + 1;
182 uint32_t kind;
183 char *err = NULL;
184 if (mesh_args_backend_of(n, av, &kind, &err)) {
185 fprintf(stderr, "%s\n", err);
186 mesh_log_exit(2);
187 }
188 if (kind == MESH_BACKEND_CUDA) {
189 /* First (it may re-exec): the per-machine libcuda dir, --cuda-lib or $MESH_CUDA_LIB; workers inherit it. */
190 char *dir = mesh_cuda_lib_dir(n, av);
191 if (!mesh_cuda_use_lib(dir, argc, argv, &err)) { mesh_error("%s", err); mesh_log_exit(2); }
192 free(dir);
193 /* --isolation mps: meshd starts itself as each program's worker process. */
194 if (n > 0 && strcmp(av[0], "worker") == 0) mesh_mps_worker_main(n - 1, av + 1);
195 /* Before anything else (no other thread exists yet): EAGER, or refuse. */
196 if (!mesh_cuda_force_eager(&err)) { mesh_error("%s", err); mesh_log_exit(2); }
197 }
198 mesh_args a;
199 if (mesh_args_parse(n, av, &a, &err)) {
200 fprintf(stderr, "%s\n", err);
201 mesh_log_exit(2);
202 }
203 install_signals();
204 struct stat st;
205 if (stat(a.programs, &st) != 0 || !S_ISDIR(st.st_mode)) fail(mesh_buf_fmt("--programs %s: not a directory", a.programs));
206 if (a.cluster)
207 fail(mesh_xstrdup("--cluster/--peer: this port's fleet thread (peer.rs run) is not wired into meshd yet; run a single meshd"));
208
209 /* Ports first: a busy port should fail before the GPU is touched. */
210 int control = bind_or_exit("control", a.control);
211 int web = bind_or_exit("http", a.http);
212 /* The mailbox: one shared-memory file the GPU backend wraps and native workers map. */
213 mesh_shm shm;
214 if (!mesh_shm_new(&shm, &err)) fail(err);
215 /* No fallback: the backend asked for (or the platform's) is the one that runs. */
216 mesh_backend *gpu = NULL;
217 if (kind == MESH_BACKEND_CUDA) gpu = open_cuda(&a, &shm, &err);
218 else if (kind == MESH_BACKEND_WASM) {
219 /* A CPU-only meshd: wasm lanes on the shared mailbox (native workers beside them). */
220 if (!mesh_wasm_backend_open(shm.base, a.wasm_cache, &gpu, &err)) gpu = NULL;
221 } else err = mesh_buf_fmt("--backend %s: this meshd (linux) was built without it", MESH_BACKENDS[kind]);
222 if (!gpu) fail(err);
223 const mesh_caps *caps = gpu->ops->caps(gpu);
224 uint8_t *base;
225 uint64_t dev;
226 if (!gpu->ops->mailbox(gpu, &base, &dev, &err)) fail(err);
227 if (base != shm.base) fail(mesh_xstrdup("internal: the backend's mailbox is not the shared file"));
228 mesh_mailbox mb = mesh_mailbox_new(base, dev);
229 mesh_info("mailbox %d bytes (shared memory, fd %d), device address 0x%llx; catalog %s", MESH_MAILBOX_BYTES, shm.fd, (unsigned long long)dev,
230 a.programs);
231 /* The CPU executors beside the GPU, on the same mailbox: wasm lanes (none beside itself) and native worker processes. */
232 mesh_backend *cpus[2];
233 size_t ncpus = 0;
234 if (kind != MESH_BACKEND_WASM) {
235 if (!mesh_wasm_backend_open(shm.base, a.wasm_cache, &cpus[ncpus], &err)) fail(err);
236 ncpus++;
237 }
238 /* Where the inference services listen, and what native workers are started with. */
239 uint32_t me = a.cluster ? a.topo.me : 0, nsys = a.cluster ? a.topo.n : 1;
240 mesh_infer_svc svc;
241 bool ok = a.has_keys ? mesh_infer_svc_cluster(&svc, keys_get, &a.keys, me, nsys, &err) == 0 : mesh_infer_svc_single(&svc, a.control, &err) == 0;
242 if (!ok) fail(err);
243 mesh_worker_cfg wcfg;
244 memset(&wcfg, 0, sizeof wcfg);
245 const char *access;
246 if (kind == MESH_BACKEND_CUDA) {
247 char *pipe = mesh_mps_native_pipe_dir();
248 if (pipe) {
249 wcfg.gpu = MESH_GPU_ACCESS_MPS;
250 wcfg.mps_pipe_dir = pipe;
251 wcfg.check_gpu_client = mesh_mps_check_native;
252 mesh_buf q = MESH_BUF_INIT;
253 mesh_buf_puts(&q, "Mps(");
254 mesh_rs_debug_cstr(&q, pipe);
255 mesh_buf_putc(&q, ')');
256 access = mesh_buf_take(&q);
257 } else { wcfg.gpu = MESH_GPU_ACCESS_BLINK; access = "Blink"; }
258 } else { wcfg.gpu = MESH_GPU_ACCESS_NONE; access = "None"; }
259 char *env[4] = {mesh_buf_fmt("MESH_SERVICE_BASE=%u", svc.bases[me]), mesh_buf_fmt("MESH_SYSTEM=%u", me), NULL, NULL};
260 if (a.models) {
261 char real[PATH_MAX];
262 if (!realpath(a.models, real)) fail(mesh_buf_fmt("--models %s: %s (os error %d)", a.models, strerror(errno), errno));
263 env[2] = mesh_buf_fmt("MESH_MODELS=%s", real);
264 }
265 wcfg.env = env;
266 mesh_buf envs = MESH_BUF_INIT;
267 for (int i = 0; env[i]; i++) mesh_buf_printf(&envs, "%s%s", i ? " " : "", env[i]);
268 mesh_info("native workers: GPU access %s; services based at port %u (%s)", access, svc.bases[me], mesh_buf_str(&envs));
269 mesh_buf_free(&envs);
270 cpus[ncpus++] = mesh_native_backend_new(&shm, a.logs, &wcfg);
271 char *label = mesh_backend_label(mesh_caps_word(caps));
272 mesh_info("executors: %s, %snative workers (%s binaries, output to %s/<name>.log); wasm: the C interpreter, NaN canonicalisation, task "
273 "deadlines%s%s",
274 label, kind != MESH_BACKEND_WASM ? "wasm (CPU lanes), " : "", mesh_native_target(), a.logs,
275 a.wasm_cache ? ", compiled modules cached in " : "", a.wasm_cache ? a.wasm_cache : "");
276 free(label);
277
278 uint32_t system = a.cluster ? a.topo.me : MESH_PEER_NONE;
279 /* The fleet's id and size (viewers; the http program). */
280 mesh_mailbox_set(mb, MESH_H_PEER, system);
281 mesh_mailbox_set(mb, MESH_H_SYSTEMS, nsys);
282 /* This machine's GPU facts: the header is their one source of truth. */
283 mesh_mailbox_set(mb, MESH_H_SMS, caps->sms);
284 mesh_mailbox_set(mb, MESH_H_SMS_TOTAL, caps->sms_total);
285 mesh_mailbox_set(mb, MESH_H_CC, (uint32_t)(caps->cc_major * 10 + caps->cc_minor));
286 mesh_mailbox_set(mb, MESH_H_BLOCKS_PER_SM, caps->blocks_per_sm);
287 mesh_mailbox_set(mb, MESH_H_BACKEND, mesh_caps_word(caps));
288 mesh_mailbox_set(mb, MESH_H_MAX_TG_THREADS, caps->max_tg_threads);
289 mesh_mailbox_set(mb, MESH_H_MAX_SHARED, caps->max_shared);
290 mesh_shared *sh = mesh_shared_new(mb, a.programs, a.cluster, a.cluster ? system : 0, nsys, &QUIT);
291 mesh_host_block *host = &sh->host;
292 char hostname[256] = "";
293 gethostname(hostname, sizeof hostname - 1);
294 mesh_host_block_set_u32(host, MESH_S_SYSTEM, system);
295 mesh_host_block_set_u32(host, MESH_S_SYSTEMS, nsys);
296 mesh_host_block_set_str(host, MESH_S_ARCH, 16, ARCH);
297 mesh_host_block_set_u32(host, MESH_S_SMS, mesh_mailbox_get(mb, MESH_H_SMS));
298 mesh_host_block_set_u32(host, MESH_S_SMS_TOTAL, mesh_mailbox_get(mb, MESH_H_SMS_TOTAL));
299 mesh_host_block_set_u32(host, MESH_S_GREEN, caps->green);
300 mesh_host_block_set_u32(host, MESH_S_ISOLATION, gpu->ops->isolation(gpu));
301 mesh_host_block_set_u32(host, MESH_S_HEAP_FREE, (uint32_t)MESH_VIEW_PAGES);
302 mesh_host_block_set_u32(host, MESH_S_HEAP_RUN, (uint32_t)MESH_VIEW_PAGES);
303 mesh_host_block_set_str(host, MESH_S_HOSTNAME, 32, hostname);
304 mesh_host_block_set_str(host, MESH_S_GPU, 32, caps->name ? caps->name : "");
305 mesh_host_block_set_str(host, MESH_S_PROGRAMS_DIR, 64, a.programs);
306 struct sockaddr_in sa;
307 socklen_t sl = sizeof sa;
308 char http_addr[64];
309 if (getsockname(web, (struct sockaddr *)&sa, &sl) == 0) {
310 char ip[32];
311 inet_ntop(AF_INET, &sa.sin_addr, ip, sizeof ip);
312 snprintf(http_addr, sizeof http_addr, "%s:%u", ip, ntohs(sa.sin_port));
313 } else snprintf(http_addr, sizeof http_addr, "%s", a.http);
314 mesh_host_block_set_str(host, MESH_S_HTTP_ADDR, 64, http_addr);
315 sh->svc = svc;
316 sh->has_svc = true;
317 mesh_shared_set_peer_state(sh, a.cluster ? MESH_PEER_WAITING : MESH_PEER_SINGLE);
318 /* The jobs and blob threads (jobd.rs, blobd.rs) are not wired yet: `jobs` requests answer at once. */
319 mesh_chan_close_rx(sh->jobs);
320
321 mesh_driver_config cfg = {a.cluster, a.state, a.autoload, false};
322 mesh_driver *d = mesh_driver_new(sh, gpu, cpus, ncpus, cfg);
323 if (!mesh_driver_start(d, &err)) fail(err);
324 static serve_arg cs, hs;
325 cs = (serve_arg){sh, control};
326 hs = (serve_arg){sh, web};
327 spawn(control_thread, &cs);
328 spawn(http_thread, &hs);
329 spawn(net_thread, sh);
330 spawn(load_thread, sh);
331 /* The driver runs on the main thread (it made the context current in start). */
332 int code = mesh_driver_run(d);
333 /* A fail-stop: the fleet state or the job log could not be made durable. */
334 if (atomic_load(&sh->fatal)) code = 3;
335 mesh_backend_cleanup();
336 mesh_driver_free(d); /* the executors with it; the port threads only use sh, which lives until exit */
337 uint64_t dropped = mesh_log_dropped();
338 if (dropped) mesh_info("exit %d (%llu log lines were not written to stderr)", code, (unsigned long long)dropped);
339 else mesh_info("exit %d", code);
340 mesh_log_exit(code);
341}