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