==> src/ctx.c <==1/* mesh-worker lib.rs's View and Ctx. */
2#include "mesh/worker_ctx.h"
3
4#include "mesh/abi/base.h"
5#include "mesh/abi/control_block.h"
6#include "mesh/abi/ext.h"
7#include "mesh/abi/fault.h"
8#include "mesh/abi/header.h"
9#include "mesh/abi/input_queue.h"
10#include "mesh/abi/pipe.h"
11#include "mesh/abi/program_table.h"
12
13#include <sched.h>
14#include <stdatomic.h>
15#include <string.h>
16#include <time.h>
17
18uint64_t mesh_worker_now_ns(void) {
19 struct timespec t = {0, 0};
20#if defined(__APPLE__)
21 clock_gettime(CLOCK_UPTIME_RAW, &t);
22#else
23 clock_gettime(CLOCK_MONOTONIC, &t);
24#endif
25 return (uint64_t)t.tv_sec * 1000000000u + (uint64_t)t.tv_nsec;
26}
27
28static void fault(uint8_t *ctl, uint32_t code) {
29 _Atomic uint32_t *w = (_Atomic uint32_t *)(void *)(ctl + MESH_P_FAULT);
30 uint32_t zero = 0;
31 atomic_compare_exchange_strong_explicit(w, &zero, code, memory_order_acq_rel, memory_order_relaxed);
32}
33
34static bool ok(mesh_view v, uint32_t off, uint32_t n, uint32_t align) {
35 uint64_t end = (uint64_t)off + n;
36 bool fine = end <= v.len && off % align == 0 && ((uintptr_t)v.base + off) % align == 0;
37 if (!fine) fault(v.ctl, MESH_FAULT_VIEW_BOUNDS | off << 8);
38 return fine;
39}
40
41static _Atomic uint32_t *w32(mesh_view v, uint32_t off) { return (_Atomic uint32_t *)(void *)(v.base + off); }
42static _Atomic uint64_t *w64(mesh_view v, uint32_t off) { return (_Atomic uint64_t *)(void *)(v.base + off); }
43
44uint32_t mesh_view_len(mesh_view v) { return v.len; }
45uint32_t mesh_view_u32(mesh_view v, uint32_t off) { return ok(v, off, 4, 4) ? atomic_load_explicit(w32(v, off), memory_order_acquire) : 0; }
46float mesh_view_f32(mesh_view v, uint32_t off) {
47 uint32_t u = mesh_view_u32(v, off);
48 float f;
49 memcpy(&f, &u, 4);
50 return f;
51}
52uint64_t mesh_view_u64(mesh_view v, uint32_t off) { return ok(v, off, 8, 8) ? atomic_load_explicit(w64(v, off), memory_order_acquire) : 0; }
53uint8_t mesh_view_u8(mesh_view v, uint32_t off) {
54 return ok(v, off, 1, 1) ? atomic_load_explicit((_Atomic uint8_t *)(void *)(v.base + off), memory_order_relaxed) : 0;
55}
56void mesh_view_set_u32(mesh_view v, uint32_t off, uint32_t x) {
57 if (ok(v, off, 4, 4)) atomic_store_explicit(w32(v, off), x, memory_order_release);
58}
59void mesh_view_set_f32(mesh_view v, uint32_t off, float x) {
60 uint32_t u;
61 memcpy(&u, &x, 4);
62 mesh_view_set_u32(v, off, u);
63}
64void mesh_view_set_u64(mesh_view v, uint32_t off, uint64_t x) {
65 if (ok(v, off, 8, 8)) atomic_store_explicit(w64(v, off), x, memory_order_release);
66}
67void mesh_view_set_u8(mesh_view v, uint32_t off, uint8_t x) {
68 if (ok(v, off, 1, 1)) atomic_store_explicit((_Atomic uint8_t *)(void *)(v.base + off), x, memory_order_relaxed);
69}
70uint32_t mesh_view_add_u32(mesh_view v, uint32_t off, uint32_t x) {
71 return ok(v, off, 4, 4) ? atomic_fetch_add_explicit(w32(v, off), x, memory_order_acq_rel) : 0;
72}
73bool mesh_view_read(mesh_view v, uint32_t off, uint8_t *out, size_t n) {
74 if (n > UINT32_MAX || !ok(v, off, (uint32_t)n, 1)) {
75 memset(out, 0, n);
76 return false;
77 }
78 memcpy(out, v.base + off, n);
79 return true;
80}
81bool mesh_view_write(mesh_view v, uint32_t off, const uint8_t *b, size_t n) {
82 if (n > UINT32_MAX || !ok(v, off, (uint32_t)n, 1)) return false;
83 memcpy(v.base + off, b, n);
84 return true;
85}
86bool mesh_view_write_f32s(mesh_view v, uint32_t off, const float *x, size_t n) { return mesh_view_write(v, off, (const uint8_t *)x, n * 4); }
87bool mesh_view_read_f32s(mesh_view v, uint32_t off, float *out, size_t n) { return mesh_view_read(v, off, (uint8_t *)out, n * 4); }
88bool mesh_view_sub(mesh_view v, uint32_t off, uint32_t len, mesh_view *out) {
89 if (!ok(v, off, len, 1)) return false;
90 *out = (mesh_view){v.base + off, len, v.ctl};
91 return true;
92}
93
94void mesh_worker_entry(mesh_mailbox mb, uint32_t s, uint32_t out[32]) {
95 size_t e = mesh_prog_entry(s);
96 for (;;) {
97 uint32_t g = mesh_mailbox_get(mb, e + MESH_PT_GEN);
98 if ((g & 1) == 0) {
99 for (size_t i = 0; i < MESH_PT_BYTES / 4; i++)
100 out[i] = atomic_load_explicit((_Atomic uint32_t *)(void *)(mb.base + e + 4 * i), memory_order_relaxed);
101 atomic_thread_fence(memory_order_acquire);
102 if (mesh_mailbox_get(mb, e + MESH_PT_GEN) == g) return;
103 }
104 sched_yield();
105 }
106}
107
108void mesh_worker_ctx_new(mesh_mailbox mb, uint32_t slot, uint32_t node, mesh_worker_ctx *c) {
109 uint32_t e[32];
110 mesh_worker_entry(mb, slot, e);
111#define F(o) e[(o) / 4]
112 size_t ctl = mesh_prog_ctl(slot);
113 *c = (mesh_worker_ctx){
114 .mb = mb, .ctl = ctl, .slot = slot, .node = node,
115 .view = {mb.base + F(MESH_PT_VIEW_OFF), F(MESH_PT_VIEW_BYTES), mb.base + ctl},
116 .nodes = F(MESH_PT_NODES), .peer_mode = F(MESH_PT_PEER), .input_off = F(MESH_PT_INPUT_OFF), .input_len = F(MESH_PT_INPUT_LEN),
117 .req_off = F(MESH_PT_REQ_OFF), .req_max = F(MESH_PT_REQ_MAX), .resp_off = F(MESH_PT_RESP_OFF), .resp_max = F(MESH_PT_RESP_MAX),
118 };
119#undef F
120}
121
122static uint32_t relaxed(mesh_mailbox mb, size_t off) {
123 return atomic_load_explicit((_Atomic uint32_t *)(void *)(mb.base + off), memory_order_relaxed);
124}
125static uint8_t *ctlp(const mesh_worker_ctx *c) { return c->mb.base + c->ctl; }
126
127uint32_t mesh_ctx_system(const mesh_worker_ctx *c) { return relaxed(c->mb, MESH_H_PEER); }
128uint32_t mesh_ctx_systems(const mesh_worker_ctx *c) { return relaxed(c->mb, MESH_H_SYSTEMS); }
129bool mesh_ctx_group(const mesh_worker_ctx *c, uint32_t *rank, uint32_t *size) {
130 uint32_t r = relaxed(c->mb, MESH_H_GROUP_RANK), n = relaxed(c->mb, MESH_H_GROUP_N);
131 if (n < 2 || n > MESH_MAX_SYSTEMS || r >= n) return false;
132 *rank = r;
133 *size = n;
134 return true;
135}
136uint32_t mesh_ctx_param(const mesh_worker_ctx *c) { return relaxed(c->mb, mesh_prog_ext(c->slot) + MESH_PX_PARAM); }
137uint32_t mesh_ctx_sms(const mesh_worker_ctx *c) { return relaxed(c->mb, MESH_H_SMS); }
138
139bool mesh_ctx_peer_view(const mesh_worker_ctx *c, uint32_t s, mesh_view *out) {
140 if (s >= MESH_PROG_SLOTS) return false;
141 uint32_t mine[32], e[32];
142 mesh_worker_entry(c->mb, c->slot, mine);
143 if ((mine[MESH_PT_READS_MASK / 4] & (1u << s)) == 0) {
144 fault(ctlp(c), MESH_FAULT_PEER_VIEW | s << 8);
145 return false;
146 }
147 mesh_worker_entry(c->mb, s, e);
148 uint32_t st = e[MESH_PT_STATE / 4], off = e[MESH_PT_VIEW_OFF / 4], len = e[MESH_PT_VIEW_BYTES / 4];
149 if (st == MESH_ST_EMPTY || st == MESH_ST_FAILED || (size_t)off + len > MESH_MAILBOX_BYTES) return false;
150 *out = (mesh_view){c->mb.base + off, len, ctlp(c)};
151 return true;
152}
153
154uint32_t mesh_ctx_prog_state(const mesh_worker_ctx *c, uint32_t s) { return relaxed(c->mb, mesh_prog_entry(s) + MESH_PT_STATE); }
155
156void mesh_ctx_prog_name(const mesh_worker_ctx *c, uint32_t s, char out[17]) {
157 uint32_t e[32];
158 mesh_worker_entry(c->mb, s, e);
159 memcpy(out, (const uint8_t *)e + MESH_PT_NAME, MESH_NAME_BYTES);
160 out[MESH_NAME_BYTES] = 0;
161}
162
163bool mesh_ctx_find(const mesh_worker_ctx *c, const char *name, uint32_t *slot) {
164 for (uint32_t s = 0; s < MESH_PROG_SLOTS; s++) {
165 if (mesh_ctx_prog_state(c, s) == MESH_ST_EMPTY) continue;
166 char n[17];
167 mesh_ctx_prog_name(c, s, n);
168 if (strcmp(n, name) == 0) {
169 *slot = s;
170 return true;
171 }
172 }
173 return false;
174}
175
176bool mesh_ctx_mailbox(const mesh_worker_ctx *c, mesh_mailbox *out) {
177 uint32_t e[32];
178 mesh_worker_entry(c->mb, c->slot, e);
179 if (e[MESH_PT_READS_MASK / 4] != MESH_READS_ALL) return false;
180 *out = c->mb;
181 return true;
182}
183
184static size_t queue_at(const mesh_worker_ctx *c) { return (size_t)(c->view.base - c->mb.base) + c->input_off; }
185
186uint32_t mesh_ctx_input_pending(const mesh_worker_ctx *c) {
187 if (c->input_len == 0) return 0;
188 size_t q = queue_at(c);
189 return mesh_mailbox_get(c->mb, q + MESH_IQ_HEAD) - mesh_mailbox_get(c->mb, q + MESH_IQ_TAIL);
190}
191
192bool mesh_ctx_input_pop(const mesh_worker_ctx *c, uint8_t *out) {
193 if (c->input_len == 0) return false;
194 size_t q = queue_at(c);
195 for (;;) {
196 uint32_t tail = mesh_mailbox_get(c->mb, q + MESH_IQ_TAIL);
197 if (mesh_mailbox_get(c->mb, q + MESH_IQ_HEAD) == tail) return false;
198 size_t at = q + MESH_IQ_DATA + (tail & (c->input_len - 1));
199 uint8_t b = (uint8_t)(mesh_mailbox_get(c->mb, at & ~(size_t)3) >> (8 * (at & 3)));
200 if (mesh_mailbox_cas(c->mb, q + MESH_IQ_TAIL, tail, tail + 1) == tail) {
201 *out = b;
202 return true;
203 }
204 }
205}
206
207int32_t mesh_ctx_push_code(const mesh_worker_ctx *c, uint32_t s, const uint8_t *rec, size_t n) {
208 mesh_words w = mesh_mailbox_words(c->mb);
209 int r = mesh_pipe_push(&w, c->slot, s, rec, n);
210 if (r == -MESH_PIPE_ERR_NOT_CONNECTED || r == -MESH_PIPE_ERR_SIZE) fault(ctlp(c), MESH_FAULT_PIPE | s << 8);
211 return mesh_pipe_code(r);
212}
213
214bool mesh_ctx_push(const mesh_worker_ctx *c, uint32_t s, const uint8_t *rec, size_t n) { return mesh_ctx_push_code(c, s, rec, n) == MESH_PIPE_OK; }
215
216bool mesh_ctx_pop(const mesh_worker_ctx *c, uint8_t *out, size_t n) {
217 mesh_words w = mesh_mailbox_words(c->mb);
218 int r = mesh_pipe_pop(&w, c->slot, out, n);
219 if (r == -MESH_PIPE_ERR_NOT_CONNECTED || r == -MESH_PIPE_ERR_SIZE) fault(ctlp(c), MESH_FAULT_PIPE | c->slot << 8);
220 return r == 1;
221}
222
223bool mesh_ctx_pipe_len(const mesh_worker_ctx *c, uint32_t s, uint32_t *len) {
224 mesh_words w = mesh_mailbox_words(c->mb);
225 mesh_pipe_inbox in;
226 if (mesh_pipe_target(&w, c->slot, s, &in) < 0) return false;
227 *len = mesh_pipe_inbox_len(&in, &w);
228 return true;
229}
230
231mesh_view mesh_ctx_request(const mesh_worker_ctx *c) {
232 uint32_t len = relaxed(c->mb, c->ctl + MESH_P_REQ_LEN);
233 if (len > c->req_max) len = c->req_max;
234 return (mesh_view){c->view.base + c->req_off, len, ctlp(c)};
235}
236
237mesh_view mesh_ctx_response(const mesh_worker_ctx *c) { return (mesh_view){c->view.base + c->resp_off, c->resp_max, ctlp(c)}; }
238
239void mesh_ctx_respond(const mesh_worker_ctx *c, uint32_t len) {
240 if (len > c->resp_max) {
241 fault(ctlp(c), MESH_FAULT_RESP_BOUNDS | len << 8);
242 len = c->resp_max;
243 }
244 mesh_mailbox_set(c->mb, c->ctl + MESH_P_RESP_LEN, len);
245 mesh_mailbox_publish64(c->mb, c->ctl + MESH_P_RESP_SEQ, mesh_mailbox_get64(c->mb, c->ctl + MESH_P_REQ_SEQ));
246}
247
248void mesh_ctx_count(const mesh_worker_ctx *c, uint32_t i, uint32_t n) {
249 if (i < MESH_COUNTERS) mesh_mailbox_fetch_add(c->mb, c->ctl + MESH_P_COUNTERS + 4 * i, n);
250}
251
252uint32_t mesh_ctx_counter(const mesh_worker_ctx *c, uint32_t i) {
253 return i < MESH_COUNTERS ? relaxed(c->mb, c->ctl + MESH_P_COUNTERS + 4 * i) : 0;
254}
255
256bool mesh_ctx_request_run(const mesh_worker_ctx *c, uint32_t s, bool on) {
257 mesh_mailbox all;
258 if (!mesh_ctx_mailbox(c, &all) || s >= MESH_PROG_SLOTS) return false;
259 size_t w = mesh_prog_ctl(s) + MESH_P_RUN_REQ;
260 uint32_t cur = mesh_mailbox_get(c->mb, w);
261 for (;;) {
262 uint32_t seen = mesh_mailbox_cas(c->mb, w, cur, mesh_run_req_next(cur, on));
263 if (seen == cur) return true;
264 cur = seen;
265 }
266}