1/* lib.rs's run: the process side of the native worker protocol. */
2#include "mesh/worker.h"
3
4#include "mesh/abi/base.h"
5#include "mesh/abi/control_block.h"
6#include "mesh/abi/header.h"
7#include "mesh/abi/native.h"
8#include "mesh/abi/program_table.h"
9
10#include <errno.h>
11#include <pthread.h>
12#include <stdarg.h>
13#include <stdio.h>
14#include <stdlib.h>
15#include <string.h>
16#include <sys/mman.h>
17#include <sys/stat.h>
18#include <unistd.h>
19
20_Noreturn void mesh_worker_panic(const char *fmt, ...) {
21    va_list ap;
22    va_start(ap, fmt);
23    fputs("mesh-worker: panic: ", stderr);
24    vfprintf(stderr, fmt, ap);
25    fputc('\n', stderr);
26    va_end(ap);
27    fflush(stderr);
28    _exit(101);
29}
30
31static _Noreturn void fail(const char *fmt, ...) {
32    va_list ap;
33    va_start(ap, fmt);
34    fputs("mesh-worker: ", stderr);
35    vfprintf(stderr, fmt, ap);
36    fputc('\n', stderr);
37    va_end(ap);
38    exit(2);
39}
40
41static mesh_mailbox map_mailbox(void) {
42    int fd = MESH_NATIVE_MAILBOX_FD;
43    struct stat st;
44    if (fstat(fd, &st) != 0) fail("fd %d (the mailbox): %s (os error %d)", fd, strerror(errno), errno);
45    if ((size_t)st.st_size < MESH_MAILBOX_BYTES)
46        fail("fd %d: %lld bytes, not a %d-byte mailbox", fd, (long long)st.st_size, MESH_MAILBOX_BYTES);
47    void *p = mmap(NULL, MESH_MAILBOX_BYTES, PROT_READ | PROT_WRITE, MAP_SHARED, fd, 0);
48    if (p == MAP_FAILED) fail("mmap of the mailbox: %s (os error %d)", strerror(errno), errno);
49    close(fd);
50    mesh_mailbox mb = mesh_mailbox_new(p, 0);
51    if (mesh_mailbox_get(mb, MESH_H_MAGIC) != MESH_MAGIC || mesh_mailbox_get(mb, MESH_H_VERSION) != MESH_VERSION)
52        fail("the mailbox has magic 0x%x version %u", mesh_mailbox_get(mb, MESH_H_MAGIC), mesh_mailbox_get(mb, MESH_H_VERSION));
53    return mb;
54}
55
56/* Commands from meshd, from the command thread to the main loop. */
57typedef struct {
58    bool exit;
59    uint32_t gen, slot;
60} cmd;
61
62static pthread_mutex_t mu = PTHREAD_MUTEX_INITIALIZER;
63static pthread_cond_t cv = PTHREAD_COND_INITIALIZER;
64static cmd queue[64];
65static size_t qhead, qlen;
66
67static void send(cmd c) {
68    pthread_mutex_lock(&mu);
69    if (qlen < 64) queue[(qhead + qlen++) % 64] = c;
70    pthread_cond_signal(&cv);
71    pthread_mutex_unlock(&mu);
72}
73
74static cmd receive(void) {
75    pthread_mutex_lock(&mu);
76    while (qlen == 0) pthread_cond_wait(&cv, &mu);
77    cmd c = queue[qhead];
78    qhead = (qhead + 1) % 64;
79    qlen--;
80    pthread_mutex_unlock(&mu);
81    return c;
82}
83
84/* One command line: "run <gen> <slot>" or "exit" (anything else is ignored). */
85static void command(char *line) {
86    char w0[16], w1[32], w2[32], extra[8];
87    int n = sscanf(line, "%15s %31s %31s %7s", w0, w1, w2, extra);
88    if (n == 1 && strcmp(w0, "exit") == 0) {
89        atomic_store(&mesh_worker_stop_all, true);
90        send((cmd){true, 0, 0});
91    } else if (n == 3 && strcmp(w0, "run") == 0) {
92        char *e1, *e2;
93        errno = 0;
94        unsigned long g = strtoul(w1, &e1, 10), s = strtoul(w2, &e2, 10);
95        if (*e1 || *e2 || errno || g > UINT32_MAX || s > UINT32_MAX || w1[0] == '-' || w2[0] == '-') return;
96        send((cmd){false, (uint32_t)g, (uint32_t)s});
97    }
98}
99
100/* Reads stdin with read(2), not stdio: a node thread that calls exit() flushes
101 * every FILE, which would wait forever on a reader blocked inside fgets. */
102static void *commands(void *arg) {
103    (void)arg;
104    char buf[512];
105    size_t len = 0;
106    for (;;) {
107        ssize_t k = read(0, buf + len, sizeof buf - 1 - len);
108        if (k < 0 && errno == EINTR) continue;
109        if (k <= 0) break;
110        len += (size_t)k;
111        buf[len] = 0;
112        char *nl;
113        while ((nl = memchr(buf, '\n', len))) {
114            *nl = 0;
115            command(buf);
116            size_t used = (size_t)(nl + 1 - buf);
117            memmove(buf, nl + 1, len - used);
118            len -= used;
119            buf[len] = 0;
120        }
121        if (len == sizeof buf - 1) len = 0; /* an overlong line: dropped */
122    }
123    /* meshd is gone (or closed our stdin): halt and leave. */
124    atomic_store(&mesh_worker_stop_all, true);
125    send((cmd){true, 0, 0});
126    return NULL;
127}
128
129_Noreturn void mesh_worker_run(const mesh_worker_program *p) {
130    const char *abi = getenv(MESH_NATIVE_ENV_ABI);
131    char want[16];
132    snprintf(want, sizeof want, "%u", (unsigned)MESH_NATIVE_ABI);
133    if (!abi) fail("a mesh native worker: meshd starts it (load it from a catalog; %s is not set)", MESH_NATIVE_ENV_ABI);
134    if (strcmp(abi, want) != 0) fail("meshd speaks native worker ABI %s, this program %s: rebuild it", abi, want);
135    mesh_mailbox mb = map_mailbox();
136    FILE *status = fdopen(MESH_NATIVE_STATUS_FD, "w");
137    if (!status) exit(0);
138    pthread_t t;
139    if (pthread_create(&t, NULL, commands, NULL) != 0) mesh_worker_panic("a command thread");
140    pthread_detach(t);
141    if (fprintf(status, "ready %u\n", (unsigned)MESH_NATIVE_ABI) < 0 || fflush(status) != 0) exit(0);
142    static atomic_bool stop;
143    for (;;) {
144        cmd c = receive();
145        if (c.exit) exit(0);
146        if (c.slot >= MESH_PROG_SLOTS) continue;
147        mesh_worker_run_nodes(mb, c.slot, p, &stop);
148        fprintf(status, "halted %u\n", (unsigned)c.gen);
149        fflush(status);
150        if (mesh_mailbox_get(mb, mesh_prog_ctl(c.slot) + MESH_P_SHUTDOWN) == MESH_SHUTDOWN_EXIT || atomic_load(&mesh_worker_stop_all))
151            exit(0);
152    }
153}