==> src/backend.c <==1/* The backend and its lanes (mesh-host/src/wasm.rs `WasmBackend`, `Lane`, `trap_fault`). */
2#include "mesh/exec_wasm.h"
3
4#include "mesh/abi/backend.h"
5#include "mesh/abi/fault.h"
6#include "mesh/abi/host_status.h"
7#include "mesh/abi/ops.h"
8#include "mesh/cpuuse.h"
9#include "mesh/lease.h"
10#include "mesh/log.h"
11
12#include <pthread.h>
13#include <stdlib.h>
14#include <string.h>
15#include <unistd.h>
16
17#include "exec_internal.h"
18#include "rt_internal.h"
19
20uint32_t mesh_wasm_trap_fault(int trap) {
21 if (trap == WASM_TRAP_INTERRUPT) return MESH_FAULT_TIME;
22 if (trap >= 0) return MESH_FAULT_TRAP | ((uint32_t)trap + 1u) << 8;
23 return MESH_FAULT_TRAP;
24}
25
26/* ---- a lane: one node of a running slot, the lease's runner ---- */
27
28typedef struct {
29 mesh_wasm_program *p;
30 wasm_instance *inst;
31 mesh_wasm_lane_data data;
32 uint32_t init, task;
33 uint32_t nodes;
34 mesh_meter meter;
35 bool metered;
36 uint64_t ticked;
37 mesh_lease lease;
38 mesh_lane_stats stats;
39} lane;
40
41typedef struct {
42 _Atomic bool done;
43 pthread_t th;
44 bool started;
45} handle;
46
47typedef struct {
48 handle *h;
49 uint32_t n;
50} slot_lanes;
51
52typedef struct {
53 mesh_wasm_rt *rt;
54 mesh_mailbox mb;
55 mesh_caps caps;
56 mesh_compiler comp;
57 pthread_mutex_t mu;
58 slot_lanes lanes[MESH_PROG_SLOTS];
59} backend;
60
61static size_t lane_max_take(void *self) {
62 (void)self;
63 return 1;
64}
65
66static void lane_heartbeat(void *self) {
67 lane *l = self;
68 uint64_t mem = 0;
69 if (l->inst) wasm_memory(l->inst, &mem);
70 if (l->metered) mesh_meter_tick(&l->meter, mem);
71}
72
73static bool lane_run(void *self, mesh_lease *lease, const mesh_taken *taken, size_t n) {
74 lane *l = self;
75 for (size_t i = 0; i < n; i++) {
76 const mesh_taken *k = &taken[i];
77 if (!l->inst) {
78 /* Its re-instantiation failed after a trap: nothing to run on. */
79 mesh_lease_fault(lease, MESH_FAULT_TRAP);
80 return false;
81 }
82 mesh_wasm_lane_data_task(&l->data, k->t);
83 wasm_set_deadline(l->inst, atomic_load(&l->p->rt->epoch) + l->p->task_ms);
84 uint64_t io[2] = {k->op, k->arg};
85 uint64_t t0 = mesh_host_ns();
86 int trap = k->op == MESH_OP_INIT ? wasm_call(l->inst, l->init, io) : wasm_call(l->inst, l->task, io);
87 uint64_t t1 = mesh_host_ns();
88 if (trap == WASM_NUM_OK) {
89 mesh_lease_complete(lease, lease->first, k, k->op == MESH_OP_INIT ? MESH_R_NONE : (uint32_t)io[0], t0, t1);
90 continue;
91 }
92 uint32_t code = mesh_wasm_trap_fault(trap);
93 mesh_lease_fault(lease, code);
94 if (code == MESH_FAULT_TIME)
95 mesh_error("wasm %s slot %u node %u: ticket %u op %u ran past task_ms = %u ms (interrupted): wasm trap: %s", l->p->name,
96 lease->slot, lease->first, k->t, k->op, l->p->task_ms, wasm_trap_message(trap));
97 else
98 mesh_error("wasm %s slot %u node %u: ticket %u op %u trapped: wasm trap: %s", l->p->name, lease->slot, lease->first,
99 k->t, k->op, wasm_trap_message(trap));
100 /* A fresh instance (its memory may be anywhere mid-task); the driver fails the slot. */
101 char *err = NULL;
102 wasm_instance_free(l->inst);
103 l->inst = mesh_wasm_instantiate(l->p, lease->mb, lease->slot, lease->first, l->nodes, &l->data, &err);
104 if (!l->inst || !mesh_wasm_entry_points(l->p, l->inst, &l->init, &l->task, &err))
105 mesh_error("wasm %s: re-instantiate after the trap: %s", l->p->name, err ? err : "?");
106 free(err);
107 return false;
108 }
109 /* A busy lane accounts its CPU time about every ms (a thread-CPU read costs a syscall). */
110 if (mesh_host_ns() - l->ticked > 1000000) {
111 l->ticked = mesh_host_ns();
112 lane_heartbeat(l);
113 }
114 return true;
115}
116
117static const mesh_runner RUNNER = {lane_max_take, lane_run, lane_heartbeat};
118
119static void *lane_main(void *arg) {
120 lane *l = arg;
121 mesh_wasm_rt *rt = l->p->rt;
122 /* This thread's CPU time and the instance's memory count for the slot (PL_HOST). */
123 mesh_meter_start(&l->meter, l->lease.slot);
124 l->metered = true;
125 lane_heartbeat(l);
126 if (l->inst) mesh_lease_run(&l->lease, &RUNNER, l);
127 else atomic_store(l->lease.done, true);
128 atomic_fetch_sub(&rt->lanes, 1);
129 mesh_meter_end(&l->meter);
130 wasm_instance_free(l->inst);
131 mesh_wasm_program_free(l->p);
132 free(l);
133 return NULL;
134}
135
136/* ---- every open executor, for meshd's exit path (MESH_CLEANUP_WASM) ---- */
137
138static pthread_mutex_t open_mu = PTHREAD_MUTEX_INITIALIZER;
139static pthread_once_t cleanup_once = PTHREAD_ONCE_INIT;
140static void *open_backends[8];
141
142/* Asks every running lane to halt (P_SHUTDOWN): it ends at its next turn. Doesn't wait. */
143static void cleanup(void);
144
145static void register_cleanup(void) { mesh_backend_on_cleanup(MESH_CLEANUP_WASM, cleanup); }
146
147/* ---- the backend ---- */
148
149static const mesh_caps *b_caps(const mesh_backend *b) { return &((backend *)b->self)->caps; }
150static bool b_true(const mesh_backend *b) {
151 (void)b;
152 return true;
153}
154static bool b_false(const mesh_backend *b) {
155 (void)b;
156 return false;
157}
158static uint32_t b_isolation(const mesh_backend *b) {
159 (void)b;
160 return MESH_ISOLATION_LEASE;
161}
162static bool b_mailbox(const mesh_backend *b, uint8_t **host, uint64_t *dev, char **err) {
163 (void)err;
164 backend *x = b->self;
165 *host = x->mb.base;
166 *dev = x->mb.dev;
167 return true;
168}
169static bool b_make_current(const mesh_backend *b, char **err) {
170 (void)b, (void)err;
171 return true;
172}
173static const mesh_compiler *b_compiler(const mesh_backend *b) { return &((backend *)b->self)->comp; }
174
175static bool b_load(const mesh_backend *b, mesh_prepared *p, mesh_kernel **out, char **err) {
176 (void)b;
177 if (!p->kernel) {
178 if (err) *err = strdup("internal: a wasm image not compiled on the JIT thread");
179 return false;
180 }
181 *out = p->kernel;
182 p->kernel = NULL;
183 return true;
184}
185
186static bool b_new_stream(const mesh_backend *b, uint32_t slot, const mesh_kernel *k, mesh_stream **out, char **err) {
187 (void)b, (void)k;
188 mesh_stream *s = calloc(1, sizeof *s);
189 if (!s) {
190 if (err) *err = strdup("out of memory");
191 return false;
192 }
193 s->exec = MESH_BACKEND_WASM;
194 s->slot = slot;
195 *out = s;
196 return true;
197}
198
199static bool running(const slot_lanes *s) {
200 for (uint32_t i = 0; s->h && i < s->n; i++)
201 if (!atomic_load(&s->h[i].done)) return true;
202 return false;
203}
204
205/* Joins and frees a slot's handles (all done). */
206static void reap(slot_lanes *s) {
207 for (uint32_t i = 0; s->h && i < s->n; i++)
208 if (s->h[i].started) pthread_join(s->h[i].th, NULL);
209 free(s->h);
210 s->h = NULL;
211 s->n = 0;
212}
213
214/* One lane per node, each instantiated here (a failure fails the launch). */
215static bool b_launch(const mesh_backend *b, const mesh_stream *st, const mesh_kernel *k, uint32_t nodes, uint64_t dev,
216 uint32_t slot, char **err) {
217 (void)dev;
218 backend *x = b->self;
219 if (st->exec != MESH_BACKEND_WASM) {
220 if (err) *err = strdup("internal: a GPU stream on the wasm executor");
221 return false;
222 }
223 if (st->slot != slot) {
224 if (err) *err = mesh_wasm_fmt("internal: slot %u launched on lanes %u", slot, st->slot);
225 return false;
226 }
227 mesh_wasm_program *p = (mesh_wasm_program *)mesh_wasm_kernel_program(k);
228 if (!p) {
229 if (err) *err = strdup("internal: a GPU image on the wasm executor");
230 return false;
231 }
232 pthread_mutex_lock(&x->mu);
233 slot_lanes *sl = &x->lanes[slot];
234 if (running(sl)) {
235 pthread_mutex_unlock(&x->mu);
236 if (err) *err = mesh_wasm_fmt("slot %u: its lanes still run", slot);
237 return false;
238 }
239 reap(sl);
240 lane **ls = calloc(nodes ? nodes : 1, sizeof *ls);
241 handle *hs = calloc(nodes ? nodes : 1, sizeof *hs);
242 bool ok = ls && hs;
243 for (uint32_t node = 0; ok && node < nodes; node++) {
244 lane *l = calloc(1, sizeof *l);
245 if (!l) {
246 ok = false;
247 break;
248 }
249 ls[node] = l;
250 l->p = p;
251 l->nodes = nodes;
252 if (!(l->inst = mesh_wasm_instantiate(p, x->mb, slot, node, nodes, &l->data, err)) ||
253 !mesh_wasm_entry_points(p, l->inst, &l->init, &l->task, err))
254 ok = false;
255 }
256 if (!ok) {
257 for (uint32_t i = 0; ls && i < nodes; i++)
258 if (ls[i]) {
259 wasm_instance_free(ls[i]->inst);
260 free(ls[i]);
261 }
262 free(ls);
263 free(hs);
264 pthread_mutex_unlock(&x->mu);
265 if (err && !*err) *err = strdup("out of memory");
266 return false;
267 }
268 for (uint32_t node = 0; node < nodes; node++) {
269 lane *l = ls[node];
270 atomic_init(&hs[node].done, false);
271 mesh_lease_init(&l->lease, x->mb, slot, node, 1, false, &hs[node].done, &l->stats);
272 atomic_fetch_add(&p->refs, 1);
273 atomic_fetch_add(&x->rt->lanes, 1);
274 if (pthread_create(&hs[node].th, NULL, lane_main, l) != 0) {
275 atomic_fetch_sub(&x->rt->lanes, 1);
276 atomic_store(&hs[node].done, true);
277 wasm_instance_free(l->inst);
278 mesh_wasm_program_free(p);
279 free(l);
280 /* The lanes already started run on; the launch fails. */
281 for (uint32_t k2 = node + 1; k2 < nodes; k2++) {
282 wasm_instance_free(ls[k2]->inst);
283 free(ls[k2]);
284 atomic_store(&hs[k2].done, true);
285 }
286 sl->h = hs;
287 sl->n = nodes;
288 free(ls);
289 pthread_mutex_unlock(&x->mu);
290 if (err) *err = strdup("wasm lane thread: cannot start it");
291 return false;
292 }
293 hs[node].started = true;
294 }
295 free(ls);
296 sl->h = hs;
297 sl->n = nodes;
298 pthread_mutex_unlock(&x->mu);
299 return true;
300}
301
302static bool b_query(const mesh_backend *b, const mesh_stream *st, const mesh_kernel *k, bool *done, char **err) {
303 (void)k;
304 backend *x = b->self;
305 if (st->exec != MESH_BACKEND_WASM) {
306 if (err) *err = strdup("internal: a GPU stream on the wasm executor");
307 return false;
308 }
309 pthread_mutex_lock(&x->mu);
310 slot_lanes *sl = &x->lanes[st->slot];
311 if (!sl->h) *done = true;
312 else if (running(sl)) *done = false;
313 else {
314 reap(sl);
315 *done = true;
316 }
317 pthread_mutex_unlock(&x->mu);
318 return true;
319}
320
321static bool b_unload(const mesh_backend *b, mesh_kernel *k, char **err) {
322 (void)b, (void)err;
323 mesh_kernel_drop(k);
324 return true;
325}
326
327static bool b_stack_limit(const mesh_backend *b, size_t *out, char **err) {
328 (void)b, (void)err;
329 *out = SIZE_MAX;
330 return true;
331}
332
333static bool b_set_stack_limit(const mesh_backend *b, size_t bytes, char **err) {
334 (void)b, (void)bytes, (void)err;
335 return true;
336}
337
338static bool b_module_scope(const mesh_backend *b, bool has_slot, uint32_t slot, uint32_t *scope) {
339 (void)b;
340 *scope = has_slot ? slot : MESH_NO_SLOT;
341 return true;
342}
343
344static void track(backend *x, bool add) {
345 pthread_mutex_lock(&open_mu);
346 for (size_t i = 0; i < sizeof open_backends / sizeof open_backends[0]; i++) {
347 if (add && !open_backends[i]) {
348 open_backends[i] = x;
349 break;
350 }
351 if (!add && open_backends[i] == x) open_backends[i] = NULL;
352 }
353 pthread_mutex_unlock(&open_mu);
354}
355
356static void cleanup(void) {
357 pthread_mutex_lock(&open_mu);
358 for (size_t i = 0; i < sizeof open_backends / sizeof open_backends[0]; i++) {
359 backend *x = open_backends[i];
360 if (!x) continue;
361 pthread_mutex_lock(&x->mu);
362 for (uint32_t s = 0; s < MESH_PROG_SLOTS; s++)
363 if (running(&x->lanes[s])) mesh_mailbox_publish(x->mb, mesh_prog_ctl(s) + MESH_P_SHUTDOWN, 1);
364 pthread_mutex_unlock(&x->mu);
365 }
366 pthread_mutex_unlock(&open_mu);
367}
368
369static void b_destroy(mesh_backend *b) {
370 backend *x = b->self;
371 track(x, false);
372 for (uint32_t s = 0; s < MESH_PROG_SLOTS; s++) {
373 if (x->lanes[s].h && running(&x->lanes[s])) {
374 /* Lanes still running at exit: ask them to halt and wait. */
375 mesh_mailbox_publish(x->mb, mesh_prog_ctl(s) + MESH_P_SHUTDOWN, 1);
376 }
377 reap(&x->lanes[s]);
378 }
379 mesh_wasm_rt_free(x->rt);
380 mesh_caps_free(&x->caps);
381 pthread_mutex_destroy(&x->mu);
382 free(x);
383 free(b);
384}
385
386static const mesh_backend_ops OPS = {
387 .caps = b_caps,
388 .isolated = b_true,
389 .resident = b_false,
390 .isolation = b_isolation,
391 .mailbox = b_mailbox,
392 .make_current = b_make_current,
393 .compiler = b_compiler,
394 .load = b_load,
395 .new_stream = b_new_stream,
396 .launch = b_launch,
397 .query = b_query,
398 .unload = b_unload,
399 .stack_limit = b_stack_limit,
400 .set_stack_limit = b_set_stack_limit,
401 .module_scope = b_module_scope,
402 .crashed = NULL,
403 .owns_halt = NULL,
404 .reads_live = b_true,
405 .sample = NULL,
406 .online_timeout_ms = NULL,
407 .destroy = b_destroy,
408};
409
410#if defined(__x86_64__)
411#define ARCH "x86_64"
412#elif defined(__aarch64__)
413#define ARCH "aarch64"
414#else
415#define ARCH "unknown"
416#endif
417
418bool mesh_wasm_backend_open(uint8_t *mailbox, const char *cache, mesh_backend **out, char **err) {
419 mesh_wasm_rt *rt = mesh_wasm_rt_new(cache, err);
420 if (!rt) return false;
421 backend *x = calloc(1, sizeof *x);
422 mesh_backend *b = calloc(1, sizeof *b);
423 if (!x || !b) {
424 free(x);
425 free(b);
426 mesh_wasm_rt_free(rt);
427 if (err) *err = strdup("wasm: out of memory");
428 return false;
429 }
430 long n = sysconf(_SC_NPROCESSORS_ONLN);
431 uint32_t threads = n > 0 ? (uint32_t)n : 1;
432 x->rt = rt;
433 x->mb = mesh_mailbox_new(mailbox, (uint64_t)(uintptr_t)mailbox);
434 x->caps.kind = MESH_BACKEND_WASM;
435 x->caps.name = mesh_wasm_fmt("%s CPU, %u threads (wasm-interp)", ARCH, threads);
436 x->caps.sms_total = threads;
437 x->caps.sms = threads;
438 x->caps.arch = mesh_wasm_fmt("wasm32 on %s", ARCH);
439 x->comp.ops = &mesh_wasm_compiler_ops;
440 x->comp.self = rt;
441 pthread_mutex_init(&x->mu, NULL);
442 b->ops = &OPS;
443 b->self = x;
444 pthread_once(&cleanup_once, register_cleanup);
445 track(x, true);
446 *out = b;
447 return true;
448}