1/* lib.rs's node_loop and run_nodes. */
2#include "mesh/worker_node.h"
3
4#include "mesh/abi/control_block.h"
5#include "mesh/abi/node.h"
6#include "mesh/abi/ops.h"
7#include "mesh/abi/program_table.h"
8#include "mesh/abi/ring.h"
9#include "mesh/abi/trace.h"
10
11#include <pthread.h>
12#include <stdio.h>
13#include <sched.h>
14#include <stdlib.h>
15#include <time.h>
16
17atomic_bool mesh_worker_stop_all;
18
19#define HEARTBEAT_NS 2000000u
20
21static uint32_t relaxed(mesh_mailbox mb, size_t off) {
22    return atomic_load_explicit((_Atomic uint32_t *)(void *)(mb.base + off), memory_order_relaxed);
23}
24
25static void nap_ns(long ns) {
26    struct timespec t = {0, ns};
27    nanosleep(&t, NULL);
28}
29
30/* Idle polling: spin right after work, then sleep longer the longer it stays idle. */
31static void idle(uint64_t since) {
32    uint64_t t = mesh_worker_now_ns() - since;
33    if (t < 50000u) sched_yield();
34    else if (t < 10000000u) nap_ns(20000);
35    else if (t < 1000000000u) nap_ns(200000);
36    else nap_ns(1000000);
37}
38
39static uint32_t cpu_of_this_thread(uint32_t node) {
40#if defined(__linux__)
41    int c = sched_getcpu();
42    if (c >= 0) return (uint32_t)c;
43#endif
44    return node;
45}
46
47void mesh_worker_node_loop(mesh_mailbox mb, uint32_t slot, uint32_t node, const mesh_worker_program *p, const atomic_bool *stop) {
48    mesh_worker_ctx ctx;
49    mesh_worker_ctx_new(mb, slot, node, &ctx);
50    size_t ctl = mesh_prog_ctl(slot), rec = mesh_node_rec(slot, node);
51    mesh_mailbox_set(mb, rec + MESH_N_SM, cpu_of_this_thread(node));
52    mesh_mailbox_fetch_add(mb, ctl + MESH_P_ONLINE, 1);
53    uint32_t beat = relaxed(mb, rec + MESH_N_HEARTBEAT);
54    uint64_t beat_at = mesh_worker_now_ns(), idle_since = beat_at;
55    for (;;) {
56        uint32_t t;
57        for (;;) {
58            t = mesh_mailbox_get(mb, ctl + MESH_P_RING_HEAD);
59            size_t e = ctl + MESH_P_RING + MESH_ENTRY_BYTES * (t % MESH_PROG_RING);
60            if (mesh_mailbox_get(mb, e + MESH_E_SEQ) == t + 1) {
61                if (mesh_mailbox_cas(mb, ctl + MESH_P_RING_HEAD, t, t + 1) == t) break;
62                continue;
63            }
64            /* Nothing published: a ticket published before the shutdown was claimed above first. */
65            if (relaxed(mb, ctl + MESH_P_SHUTDOWN) != 0 || (stop && atomic_load_explicit(stop, memory_order_relaxed)) ||
66                atomic_load_explicit(&mesh_worker_stop_all, memory_order_relaxed))
67                goto halt;
68            uint64_t now = mesh_worker_now_ns();
69            if (now - beat_at >= HEARTBEAT_NS) {
70                beat_at = now;
71                mesh_mailbox_set(mb, rec + MESH_N_HEARTBEAT, ++beat);
72            }
73            idle(idle_since);
74        }
75        size_t e = ctl + MESH_P_RING + MESH_ENTRY_BYTES * (t % MESH_PROG_RING);
76        uint32_t op = relaxed(mb, e + MESH_E_OP), arg = relaxed(mb, e + MESH_E_ARG), cslot = relaxed(mb, e + MESH_E_SLOT) % MESH_PROG_RESULTS;
77        ctx.ticket = t;
78        uint64_t start = mesh_worker_now_ns();
79        uint32_t result;
80        if (op == MESH_OP_INIT) {
81            if (p->init) p->init(&ctx, p->user);
82            result = MESH_R_NONE;
83        } else {
84            result = p->task(&ctx, op, arg, p->user);
85        }
86        uint64_t end = mesh_worker_now_ns();
87        mesh_mailbox_set(mb, ctl + MESH_P_RESULTS + 4 * cslot, result);
88        mesh_mailbox_publish(mb, ctl + MESH_P_MARKS + 4 * cslot, t + 1);
89        mesh_mailbox_fetch_add(mb, ctl + MESH_P_DONE, 1);
90        mesh_mailbox_set(mb, rec + MESH_N_TASKS, relaxed(mb, rec + MESH_N_TASKS) + 1);
91        mesh_mailbox_set(mb, rec + MESH_N_LAST_OP, op);
92        mesh_mailbox_set64(mb, rec + MESH_N_BUSY_NS, mesh_mailbox_get64(mb, rec + MESH_N_BUSY_NS) + (end - start));
93        mesh_mailbox_set(mb, rec + MESH_N_HEARTBEAT, ++beat);
94        uint32_t n = mesh_mailbox_fetch_add(mb, ctl + MESH_P_TRACE_N, 1);
95        size_t tr = mesh_trace_rec(slot, n);
96        mesh_mailbox_set(mb, tr, slot | node << 8 | op << 16);
97        mesh_mailbox_set(mb, tr + 4, t);
98        mesh_mailbox_set(mb, tr + 8, (uint32_t)start);
99        mesh_mailbox_set(mb, tr + 12, (uint32_t)end);
100        idle_since = mesh_worker_now_ns();
101    }
102halt:
103    mesh_mailbox_fetch_add(mb, ctl + MESH_P_HALTED, 1);
104}
105
106typedef struct {
107    mesh_mailbox mb;
108    uint32_t slot, node;
109    const mesh_worker_program *p;
110    const atomic_bool *stop;
111} node_arg;
112
113static void *node_main(void *a) {
114    node_arg *n = a;
115    mesh_worker_node_loop(n->mb, n->slot, n->node, n->p, n->stop);
116    return NULL;
117}
118
119void mesh_worker_run_nodes(mesh_mailbox mb, uint32_t slot, const mesh_worker_program *p, const atomic_bool *stop) {
120    uint32_t e[32];
121    mesh_worker_entry(mb, slot, e);
122    uint32_t nodes = e[MESH_PT_NODES / 4];
123    if (nodes < 1) nodes = 1;
124    if (nodes > MESH_PROG_MAX_NODES) nodes = MESH_PROG_MAX_NODES;
125    pthread_t th[MESH_PROG_MAX_NODES];
126    node_arg args[MESH_PROG_MAX_NODES];
127    uint32_t started = 0;
128    for (uint32_t n = 0; n < nodes; n++) {
129        args[n] = (node_arg){mb, slot, n, p, stop};
130        if (pthread_create(&th[n], NULL, node_main, &args[n]) != 0) {
131            fputs("mesh-worker: a node thread could not start\n", stderr);
132            abort();
133        }
134        started++;
135    }
136    for (uint32_t n = 0; n < started; n++) pthread_join(th[n], NULL);
137}