==> include/mesh/rt/ctx.h <==1/* @ajax/mesh-rt-ctx: what a program's init and task see (Rust's `Ctx`): its view, other
2 * programs' views (gated by the manifest's `reads`), the input queue, pipes,
3 * the request window, the exchange page (lockstep only), counters, block
4 * collectives, shared memory, scratch, and node/warp/lane ids.
5 *
6 * The own entry is read once at kernel start under its PT_GEN seqlock: meshd
7 * rewrites it only while the slot's kernel is not running. Other slots'
8 * entries can change under a running reader, so peer_view reads them under
9 * their seqlock.
10 *
11 * Warp collectives (input_pop, respond, push_with, pop_with, pipe_len,
12 * pipe_free) must be called by all 32 lanes of a warp, never after a branch
13 * the lanes may take differently; the calling warp's lane 0 acts. In a
14 * multi-warp node call them from one warp (`if (mesh_ctx_warp(c) == 0)`).
15 * Block collectives must be called by every thread of the node. Everything
16 * else may be called by any thread. View and shared accessors are bounds
17 * checked (an out-of-range access reads 0 / drops the write, and P_FAULT).
18 *
19 * Ids: node (the block, 0..nodes), warp (0..warps), lane (0..32 in the warp),
20 * thread (0..threads = 32 x warps); leader is thread 0 of the node, which
21 * makes side effects.
22 * Ported from alpha-rust-mesh crates/mesh-rt/src/ctx.rs. */
23#ifndef MESH_RT_CTX_H
24#define MESH_RT_CTX_H
25
26#include <stdbool.h>
27#include <stddef.h>
28#include <stdint.h>
29
30#include "mesh/abi/base.h"
31#include "mesh/abi/control_block.h"
32#include "mesh/abi/exchange.h"
33#include "mesh/abi/ext.h"
34#include "mesh/abi/fault.h"
35#include "mesh/abi/header.h"
36#include "mesh/abi/input_queue.h"
37#include "mesh/abi/pipe.h"
38#include "mesh/abi/pipe_ring.h"
39#include "mesh/abi/program_table.h"
40#include "mesh/abi/words.h"
41#include "mesh/rt/block.h"
42#include "mesh/rt/prim.h"
43#include "mesh/rt/view.h"
44
45/* ---- the mailbox's words for the pipe algorithm (system scope: the other end may be a CPU) */
46static inline uint32_t mesh_rt_gw_load(void *c, size_t a) { return mesh_rt_ld_acquire((uint8_t *)c + a); }
47static inline void mesh_rt_gw_store(void *c, size_t a, uint32_t v) { mesh_rt_st_release((uint8_t *)c + a, v); }
48static inline uint32_t mesh_rt_gw_cas(void *c, size_t a, uint32_t cmp, uint32_t desired) {
49 return mesh_rt_atom_cas((uint8_t *)c + a, cmp, desired);
50}
51static inline uint32_t mesh_rt_gw_add(void *c, size_t a, uint32_t v) { return mesh_rt_atom_add((uint8_t *)c + a, v); }
52static inline void mesh_rt_gw_fence(void *c) { (void)c; mesh_rt_fence_sc(); }
53MESH_RT_INLINE mesh_words mesh_rt_gw(uint8_t *mb) {
54 mesh_words w = {mb, mesh_rt_gw_load, mesh_rt_gw_store, mesh_rt_gw_cas, mesh_rt_gw_add, mesh_rt_gw_fence, NULL, NULL};
55 return w;
56}
57
58/* The exchange page (EX_BASE) for the lockstep program: system k's counter and
59 * slots (mesh_ex_counter / mesh_ex_slots relative to EX_BASE), k < MAX_SYSTEMS. */
60typedef struct {
61 uint8_t *base;
62} mesh_exchange;
63MESH_RT_INLINE uint8_t *mesh_exchange_base(mesh_exchange x) { return x.base; }
64/* System k's u64 round counter. */
65MESH_RT_INLINE uint8_t *mesh_exchange_counter(mesh_exchange x, uint32_t k) { return x.base + (mesh_ex_counter(k) - MESH_EX_BASE); }
66/* System k's two EX_SLOT_BYTES slots (parity 0, 1). */
67MESH_RT_INLINE uint8_t *mesh_exchange_slots(mesh_exchange x, uint32_t k) { return x.base + (mesh_ex_slots(k) - MESH_EX_BASE); }
68
69typedef struct {
70 uint8_t *mb;
71 uint8_t *ctl;
72 mesh_view view;
73 uint32_t slot;
74 uint32_t node;
75 uint32_t lane;
76 /* The thread in the node (0..threads), its warp, the node's warps. */
77 uint32_t thread;
78 uint32_t warp;
79 uint32_t warps;
80 /* The program's shared memory (after the runtime's RT_SHARED_BYTES) and its bytes. */
81 uint8_t *smem;
82 uint32_t smem_len;
83 /* The program's scratch (manifest `scratch_bytes`): device memory of its own; NULL / 0 none. */
84 uint8_t *scratch;
85 uint32_t scratch_len;
86 uint32_t nodes;
87 uint32_t reads_mask;
88 uint32_t peer_mode;
89 uint32_t input_off;
90 uint32_t input_len;
91 uint32_t req_off;
92 uint32_t req_max;
93 uint32_t resp_off;
94 uint32_t resp_max;
95 /* The running task's ticket (set by the node loop before each op). */
96 uint32_t ticket;
97} mesh_ctx;
98
99/* Reads the slot's table entry under its seqlock (PT_GEN even and unchanged). */
100MESH_RT_INLINE mesh_ctx mesh_ctx_new(uint8_t *mb, uint32_t slot) {
101 uint8_t *e = mb + mesh_prog_entry(slot);
102 uint8_t *ctl = mb + mesh_prog_ctl(slot);
103 for (;;) {
104 uint32_t g = mesh_rt_ld_acquire(e + MESH_PT_GEN);
105 if (g & 1u) { mesh_rt_nanosleep(1000); continue; }
106 uint8_t *x = mb + mesh_prog_ext(slot);
107 uint32_t dyn = mesh_rt_dyn_smem();
108 mesh_ctx c;
109 c.mb = mb;
110 c.ctl = ctl;
111 c.view = (mesh_view){mb + mesh_rt_rd(e + MESH_PT_VIEW_OFF), mesh_rt_rd(e + MESH_PT_VIEW_BYTES), ctl};
112 c.slot = slot;
113 c.node = mesh_rt_ctaid();
114 c.lane = mesh_rt_lane();
115 c.thread = mesh_rt_tid();
116 c.warp = mesh_rt_tid() >> 5;
117 c.warps = mesh_rt_ntid() >> 5;
118 c.smem = mesh_rt_smem_base() + MESH_RT_SHARED_BYTES;
119 c.smem_len = dyn > MESH_RT_SHARED_BYTES ? dyn - MESH_RT_SHARED_BYTES : 0;
120 c.scratch = (uint8_t *)(uintptr_t)mesh_rt_rd64(x + MESH_PX_SCRATCH);
121 c.scratch_len = mesh_rt_rd(x + MESH_PX_SCRATCH_BYTES);
122 c.nodes = mesh_rt_rd(e + MESH_PT_NODES);
123 c.reads_mask = mesh_rt_rd(e + MESH_PT_READS_MASK);
124 c.peer_mode = mesh_rt_rd(e + MESH_PT_PEER);
125 c.input_off = mesh_rt_rd(e + MESH_PT_INPUT_OFF);
126 c.input_len = mesh_rt_rd(e + MESH_PT_INPUT_LEN);
127 c.req_off = mesh_rt_rd(e + MESH_PT_REQ_OFF);
128 c.req_max = mesh_rt_rd(e + MESH_PT_REQ_MAX);
129 c.resp_off = mesh_rt_rd(e + MESH_PT_RESP_OFF);
130 c.resp_max = mesh_rt_rd(e + MESH_PT_RESP_MAX);
131 c.ticket = 0;
132 mesh_rt_fence_acq_rel();
133 if (mesh_rt_ld_acquire(e + MESH_PT_GEN) == g) return c;
134 }
135}
136
137/* ---- ids */
138MESH_RT_INLINE uint32_t mesh_ctx_slot(const mesh_ctx *c) { return c->slot; }
139/* This node's number (its block), 0..nodes. */
140MESH_RT_INLINE uint32_t mesh_ctx_node(const mesh_ctx *c) { return c->node; }
141/* The thread's lane in its warp, 0..32. */
142MESH_RT_INLINE uint32_t mesh_ctx_lane(const mesh_ctx *c) { return c->lane; }
143MESH_RT_INLINE uint32_t mesh_ctx_nodes(const mesh_ctx *c) { return c->nodes; }
144/* The thread's warp in the node, 0..warps. */
145MESH_RT_INLINE uint32_t mesh_ctx_warp(const mesh_ctx *c) { return c->warp; }
146/* The node's warps (the manifest's `warps`). */
147MESH_RT_INLINE uint32_t mesh_ctx_warps(const mesh_ctx *c) { return c->warps; }
148/* The thread's index in the node, 0..threads (= 32 x warp + lane). */
149MESH_RT_INLINE uint32_t mesh_ctx_thread(const mesh_ctx *c) { return c->thread; }
150MESH_RT_INLINE uint32_t mesh_ctx_thread_id(const mesh_ctx *c) { return c->thread; }
151/* Threads per node (32 x warps). */
152MESH_RT_INLINE uint32_t mesh_ctx_threads(const mesh_ctx *c) { return c->warps * 32u; }
153/* Thread 0 of the node: the one that makes side effects. */
154MESH_RT_INLINE bool mesh_ctx_leader(const mesh_ctx *c) { return c->thread == 0; }
155
156/* ---- the block (every thread of the node calls these; @ajax/mesh-rt-block) */
157MESH_RT_INLINE void mesh_ctx_block_barrier(const mesh_ctx *c) { (void)c; mesh_rt_bar_sync(); }
158MESH_RT_INLINE uint32_t mesh_ctx_block_broadcast_u32(const mesh_ctx *c, uint32_t v, uint32_t src) { (void)c; return mesh_rt_block_broadcast_u32(v, src); }
159MESH_RT_INLINE float mesh_ctx_block_broadcast_f32(const mesh_ctx *c, float v, uint32_t src) { (void)c; return mesh_rt_block_broadcast_f32(v, src); }
160MESH_RT_INLINE uint64_t mesh_ctx_block_broadcast_u64(const mesh_ctx *c, uint64_t v, uint32_t src) { (void)c; return mesh_rt_block_broadcast_u64(v, src); }
161MESH_RT_INLINE uint32_t mesh_ctx_block_reduce_u32(const mesh_ctx *c, uint32_t v, uint32_t id, mesh_rt_op_u32 op) { (void)c; return mesh_rt_block_reduce_u32(v, id, op); }
162MESH_RT_INLINE float mesh_ctx_block_reduce_f32(const mesh_ctx *c, float v, float id, mesh_rt_op_f32 op) { (void)c; return mesh_rt_block_reduce_f32(v, id, op); }
163MESH_RT_INLINE uint64_t mesh_ctx_block_reduce_u64(const mesh_ctx *c, uint64_t v, uint64_t id, mesh_rt_op_u64 op) { (void)c; return mesh_rt_block_reduce_u64(v, id, op); }
164MESH_RT_INLINE uint32_t mesh_ctx_block_scan_u32_op(const mesh_ctx *c, uint32_t v, uint32_t id, mesh_rt_op_u32 op, bool inclusive) { (void)c; return mesh_rt_block_scan_u32(v, id, op, inclusive); }
165MESH_RT_INLINE float mesh_ctx_block_scan_f32_op(const mesh_ctx *c, float v, float id, mesh_rt_op_f32 op, bool inclusive) { (void)c; return mesh_rt_block_scan_f32(v, id, op, inclusive); }
166MESH_RT_INLINE uint64_t mesh_ctx_block_scan_u64_op(const mesh_ctx *c, uint64_t v, uint64_t id, mesh_rt_op_u64 op, bool inclusive) { (void)c; return mesh_rt_block_scan_u64(v, id, op, inclusive); }
167MESH_RT_INLINE float mesh_ctx_block_sum_f32(const mesh_ctx *c, float v) { (void)c; return mesh_rt_block_reduce_f32(v, MESH_RT_SUM_F32_ID, mesh_rt_add_f32); }
168MESH_RT_INLINE float mesh_ctx_block_max_f32(const mesh_ctx *c, float v) {
169 (void)c;
170 return mesh_rt_block_reduce_f32(v, mesh_f32_from_bits(MESH_RT_MAX_F32_ID_BITS), mesh_rt_max_f32);
171}
172MESH_RT_INLINE uint32_t mesh_ctx_block_sum_u32(const mesh_ctx *c, uint32_t v) { (void)c; return mesh_rt_block_reduce_u32(v, 0, mesh_rt_add_u32); }
173MESH_RT_INLINE uint32_t mesh_ctx_block_max_u32(const mesh_ctx *c, uint32_t v) { (void)c; return mesh_rt_block_reduce_u32(v, 0, mesh_rt_max_u32); }
174MESH_RT_INLINE uint64_t mesh_ctx_block_sum_u64(const mesh_ctx *c, uint64_t v) { (void)c; return mesh_rt_block_reduce_u64(v, 0, mesh_rt_add_u64); }
175MESH_RT_INLINE uint64_t mesh_ctx_block_max_u64(const mesh_ctx *c, uint64_t v) { (void)c; return mesh_rt_block_reduce_u64(v, 0, mesh_rt_max_u64); }
176/* Inclusive / exclusive prefix sums over the node's threads in thread order. */
177MESH_RT_INLINE float mesh_ctx_block_scan_f32(const mesh_ctx *c, float v, bool inclusive) { (void)c; return mesh_rt_block_scan_f32(v, MESH_RT_SUM_F32_ID, mesh_rt_add_f32, inclusive); }
178MESH_RT_INLINE uint32_t mesh_ctx_block_scan_u32(const mesh_ctx *c, uint32_t v, bool inclusive) { (void)c; return mesh_rt_block_scan_u32(v, 0, mesh_rt_add_u32, inclusive); }
179MESH_RT_INLINE uint64_t mesh_ctx_block_scan_u64(const mesh_ctx *c, uint64_t v, bool inclusive) { (void)c; return mesh_rt_block_scan_u64(v, 0, mesh_rt_add_u64, inclusive); }
180/* A block-strided loop: i = thread(), thread() + threads(), .. < n (Rust's for_strided). */
181#define MESH_CTX_FOR_STRIDED(c, n, i) for (uint32_t i = (c)->thread, i##_end_ = (n); i < i##_end_; i += mesh_ctx_threads(c))
182
183/* The program's scratch: device memory of its own (manifest `scratch_bytes`,
184 * zeroed at load, kept while the image is loaded, never seen by the host),
185 * bounds checked like the view. Empty without one. */
186MESH_RT_INLINE mesh_view mesh_ctx_scratch_view(const mesh_ctx *c) {
187 return (mesh_view){c->scratch, c->scratch ? c->scratch_len : 0, c->ctl};
188}
189/* Its bytes (the manifest's `scratch_bytes`; 0 without one). */
190MESH_RT_INLINE uint32_t mesh_ctx_scratch_bytes(const mesh_ctx *c) { return c->scratch ? c->scratch_len : 0; }
191/* Its raw base (unchecked access: a kernel's inner loops; plain loads, cached). */
192MESH_RT_INLINE uint8_t *mesh_ctx_scratch_ptr(const mesh_ctx *c, uint32_t *len) {
193 *len = mesh_ctx_scratch_bytes(c);
194 return c->scratch;
195}
196/* The program's shared memory bytes per node (the manifest's `shared_bytes`). */
197MESH_RT_INLINE uint32_t mesh_ctx_shared_len(const mesh_ctx *c) { return c->smem_len; }
198/* len elements of esize bytes (alignment esize) at byte off of the program's
199 * shared memory: the same memory for every thread of the node. False (and
200 * FAULT_SHARED_BOUNDS | off << 8) when outside `shared_bytes` or misaligned. */
201MESH_RT_INLINE bool mesh_ctx_shared(const mesh_ctx *c, uint32_t off, uint32_t len, uint32_t esize, mesh_shared *out) {
202 bool ok = off % esize == 0 && (uint64_t)off + (uint64_t)esize * len <= c->smem_len;
203 if (!ok) { mesh_rt_fault(c->ctl, MESH_FAULT_SHARED_BOUNDS | off << 8); return false; }
204 *out = (mesh_shared){c->smem + off, len, c->ctl};
205 return true;
206}
207/* The running task's ring ticket (the same on every lane). Tickets of one
208 * publish are consecutive, so a batch's task with arg index k has ticket t0 + k. */
209MESH_RT_INLINE uint32_t mesh_ctx_ticket(const mesh_ctx *c) { return c->ticket; }
210/* This system's number in the cluster (0..systems()), or PEER_NONE. */
211MESH_RT_INLINE uint32_t mesh_ctx_system(const mesh_ctx *c) { return mesh_rt_ld_relaxed(c->mb + MESH_H_PEER); }
212/* The cluster's size N (H_SYSTEMS; 1 single). */
213MESH_RT_INLINE uint32_t mesh_ctx_systems(const mesh_ctx *c) { return mesh_rt_ld_relaxed(c->mb + MESH_H_SYSTEMS); }
214/* The lockstep group this member is in: its rank r and the group's size N,
215 * with 2 <= N <= MAX_SYSTEMS and r < N; false outside a group. */
216MESH_RT_INLINE bool mesh_ctx_group(const mesh_ctx *c, uint32_t *rank, uint32_t *n) {
217 uint32_t r = mesh_rt_ld_relaxed(c->mb + MESH_H_GROUP_RANK), k = mesh_rt_ld_relaxed(c->mb + MESH_H_GROUP_N);
218 if (!(k >= 2 && k <= MESH_MAX_SYSTEMS && r < k)) return false;
219 *rank = r;
220 *n = k;
221 return true;
222}
223/* The manifest's `param` (PX_PARAM; 0 when the manifest has none). */
224MESH_RT_INLINE uint32_t mesh_ctx_param(const mesh_ctx *c) { return mesh_rt_ld_relaxed(c->mb + mesh_prog_ext(c->slot) + MESH_PX_PARAM); }
225/* The device's SM count (H_SMS). */
226MESH_RT_INLINE uint32_t mesh_ctx_sms(const mesh_ctx *c) { return mesh_rt_ld_relaxed(c->mb + MESH_H_SMS); }
227
228/* ---- own view */
229MESH_RT_INLINE mesh_view mesh_ctx_view(const mesh_ctx *c) { return c->view; }
230MESH_RT_INLINE uint32_t mesh_ctx_view_len(const mesh_ctx *c) { return c->view.len; }
231MESH_RT_INLINE uint32_t mesh_ctx_view_u32(const mesh_ctx *c, uint32_t off) { return mesh_view_u32(c->view, off); }
232MESH_RT_INLINE float mesh_ctx_view_f32(const mesh_ctx *c, uint32_t off) { return mesh_view_f32(c->view, off); }
233MESH_RT_INLINE uint64_t mesh_ctx_view_u64(const mesh_ctx *c, uint32_t off) { return mesh_view_u64(c->view, off); }
234MESH_RT_INLINE uint8_t mesh_ctx_view_u8(const mesh_ctx *c, uint32_t off) { return mesh_view_u8(c->view, off); }
235MESH_RT_INLINE void mesh_ctx_set_u32(const mesh_ctx *c, uint32_t off, uint32_t v) { mesh_view_set_u32(c->view, off, v); }
236MESH_RT_INLINE void mesh_ctx_set_f32(const mesh_ctx *c, uint32_t off, float v) { mesh_view_set_f32(c->view, off, v); }
237MESH_RT_INLINE void mesh_ctx_set_u64(const mesh_ctx *c, uint32_t off, uint64_t v) { mesh_view_set_u64(c->view, off, v); }
238MESH_RT_INLINE void mesh_ctx_set_u8(const mesh_ctx *c, uint32_t off, uint8_t v) { mesh_view_set_u8(c->view, off, v); }
239
240/* ---- other programs */
241/* Slot s's view, if the manifest's `reads` lists it and the slot holds a
242 * program; false otherwise (an undeclared read also records FAULT_PEER_VIEW).
243 * Writes through it are the caller's responsibility (`reads` is advisory). */
244MESH_RT_INLINE bool mesh_ctx_peer_view(const mesh_ctx *c, uint32_t s, mesh_view *out) {
245 if (s >= MESH_PROG_SLOTS) return false;
246 if ((c->reads_mask & (1u << s)) == 0) { mesh_rt_fault(c->ctl, MESH_FAULT_PEER_VIEW | s << 8); return false; }
247 /* Slot s's entry may be rewritten while we run (its rollback, its unload):
248 * offset, length and state are read as one, under its seqlock. */
249 uint8_t *e = c->mb + mesh_prog_entry(s);
250 for (;;) {
251 uint32_t g = mesh_rt_ld_acquire(e + MESH_PT_GEN);
252 if (g & 1u) { mesh_rt_nanosleep(1000); continue; }
253 uint32_t st = mesh_rt_ld_relaxed(e + MESH_PT_STATE), off = mesh_rt_rd(e + MESH_PT_VIEW_OFF), len = mesh_rt_rd(e + MESH_PT_VIEW_BYTES);
254 mesh_rt_fence_acq_rel();
255 if (mesh_rt_ld_acquire(e + MESH_PT_GEN) != g) continue;
256 if (st == MESH_ST_EMPTY || st == MESH_ST_FAILED) return false;
257 *out = (mesh_view){c->mb + off, len, c->ctl};
258 return true;
259 }
260}
261MESH_RT_INLINE uint32_t mesh_ctx_prog_state(const mesh_ctx *c, uint32_t s) { return mesh_rt_ld_relaxed(c->mb + mesh_prog_entry(s) + MESH_PT_STATE); }
262/* Slot s's name, NUL padded. */
263MESH_RT_INLINE void mesh_ctx_prog_name(const mesh_ctx *c, uint32_t s, uint8_t out[MESH_NAME_BYTES]) {
264 for (uint32_t i = 0; i < MESH_NAME_BYTES; i++) out[i] = *(const volatile uint8_t *)(c->mb + mesh_prog_entry(s) + MESH_PT_NAME + i);
265}
266MESH_RT_INLINE bool mesh_ctx_prog_name_is(const mesh_ctx *c, uint32_t s, const char *name, uint32_t len) {
267 uint8_t n[MESH_NAME_BYTES];
268 mesh_ctx_prog_name(c, s, n);
269 if (len > MESH_NAME_MAX) return false;
270 for (uint32_t i = 0; i < len; i++)
271 if (n[i] != (uint8_t)name[i]) return false;
272 return n[len] == 0;
273}
274/* The slot holding the program `name` (len bytes), if any (the table is public). */
275MESH_RT_INLINE bool mesh_ctx_find(const mesh_ctx *c, const char *name, uint32_t len, uint32_t *slot) {
276 for (uint32_t s = 0; s < MESH_PROG_SLOTS; s++)
277 if (mesh_ctx_prog_state(c, s) != MESH_ST_EMPTY && mesh_ctx_prog_name_is(c, s, name, len)) { *slot = s; return true; }
278 return false;
279}
280/* Asks meshd to start/stop slot s: advances the GPU-owned P_RUN_REQ
281 * (n << 1 | on) with atom.cas; meshd only reads it. Only for a program with
282 * reads = ["*"] (http). Call from one lane. */
283MESH_RT_INLINE bool mesh_ctx_request_run(const mesh_ctx *c, uint32_t s, bool on) {
284 if (c->reads_mask != MESH_READS_ALL || s >= MESH_PROG_SLOTS) return false;
285 uint8_t *w = c->mb + mesh_prog_ctl(s) + MESH_P_RUN_REQ;
286 mesh_rt_fence_acq_rel();
287 uint32_t cur = mesh_rt_ld_relaxed(w);
288 for (;;) {
289 uint32_t seen = mesh_rt_atom_cas(w, cur, mesh_run_req_next(cur, on));
290 if (seen == cur) break;
291 cur = seen;
292 }
293 return true;
294}
295/* The whole mailbox, for a READS_ALL program (http renders the table); NULL otherwise. */
296MESH_RT_INLINE uint8_t *mesh_ctx_mailbox(const mesh_ctx *c) { return c->reads_mask == MESH_READS_ALL ? c->mb : NULL; }
297
298/* ---- input queue (manifest `input`) */
299/* Bytes waiting in the input queue. */
300MESH_RT_INLINE uint32_t mesh_ctx_input_pending(const mesh_ctx *c) {
301 if (c->input_len == 0) return 0;
302 uint8_t *q = c->view.base + c->input_off;
303 return mesh_rt_ld_acquire(q + MESH_IQ_HEAD) - mesh_rt_rd(q + MESH_IQ_TAIL);
304}
305/* Collective: the next input byte, the same on every lane; false when empty. */
306MESH_RT_INLINE bool mesh_ctx_input_pop(const mesh_ctx *c, uint8_t *out) {
307 uint32_t v = 0x100;
308 if (c->lane == 0 && c->input_len != 0) {
309 uint8_t *q = c->view.base + c->input_off;
310 uint32_t tail = mesh_rt_rd(q + MESH_IQ_TAIL);
311 if (mesh_rt_ld_acquire(q + MESH_IQ_HEAD) != tail) {
312 v = *(const volatile uint8_t *)(q + MESH_IQ_DATA + (tail & (c->input_len - 1)));
313 mesh_rt_st_release(q + MESH_IQ_TAIL, tail + 1);
314 }
315 }
316 v = mesh_rt_shfl_idx(v, 0);
317 if (v >= 0x100) return false;
318 *out = (uint8_t)v;
319 return true;
320}
321
322/* ---- pipes (manifest `inbox`, `feeds`; @ajax/mesh-abi-pipe) */
323typedef void (*mesh_ctx_record_fn)(mesh_view rec, void *user);
324
325/* Collective: pushes one record into slot s's inbox. Lane 0 claims a cell;
326 * then every lane calls fill with a view of the cell (the consumer's record
327 * size: lane l typically writes words l, l + 32, ...), and lane 0 publishes
328 * it. PIPE_OK, PIPE_NONE (full: keep the record, try later), or a negative
329 * code (not fed / gone; not fed also records FAULT_PIPE). The same on every lane. */
330MESH_RT_INLINE int32_t mesh_ctx_push_with(const mesh_ctx *c, uint32_t s, mesh_ctx_record_fn fill, void *user) {
331 mesh_words w = mesh_rt_gw(c->mb);
332 uint32_t code = (uint32_t)MESH_PIPE_NONE, pos = 0, base = 0, rec = 0, parity = 0;
333 bool have = false;
334 mesh_pipe_ring ring = {0, 0, 0};
335 if (c->lane == 0) {
336 /* Counted in flight (flight_word) from before the block is read until after the commit. */
337 parity = mesh_pipe_flight_begin(&w, c->slot);
338 mesh_pipe_inbox in;
339 int r = mesh_pipe_target(&w, c->slot, s, &in);
340 if (r < 0) {
341 if (r == -MESH_PIPE_ERR_NOT_CONNECTED) mesh_rt_fault(c->ctl, MESH_FAULT_PIPE | s << 8);
342 code = (uint32_t)mesh_pipe_code(r);
343 } else {
344 /* A GPU producer pushes on the inbox's GPU side (its head is CASed by GPU warps only). */
345 mesh_pipe_ring rr = mesh_pipe_inbox_side(&in, MESH_PIPE_SIDE_GPU);
346 uint32_t p;
347 if (mesh_pipe_ring_reserve_push(&rr, &w, c->slot, &p)) {
348 code = (uint32_t)MESH_PIPE_OK;
349 pos = p;
350 base = (uint32_t)mesh_pipe_ring_cell(&rr, p);
351 rec = rr.rec;
352 ring = rr;
353 have = true;
354 }
355 }
356 if (!have) mesh_pipe_flight_end(&w, c->slot, parity);
357 }
358 int32_t cd = (int32_t)mesh_rt_shfl_idx(code, 0);
359 if (cd != MESH_PIPE_OK) return cd;
360 base = mesh_rt_shfl_idx(base, 0);
361 rec = mesh_rt_shfl_idx(rec, 0);
362 fill((mesh_view){c->mb + base, rec, c->ctl}, user);
363 /* Every lane's record words before lane 0's release of the cell. */
364 mesh_rt_fence_acq_rel();
365 mesh_rt_warp_sync();
366 /* The record's sum (pipe rec_sum), lanes summing words l, l + 32, ... */
367 uint32_t part = 0;
368 for (uint32_t k = c->lane; k < rec / 4; k += 32) part += mesh_pipe_sum_part(k, mesh_rt_rd(c->mb + base + 4 * k));
369 for (uint32_t m = 16; m > 0; m >>= 1) part += mesh_rt_shfl_bfly(part, m);
370 if (have) {
371 mesh_rt_st_relaxed(c->mb + mesh_pipe_ring_sum(&ring, pos), part ^ mesh_pipe_pos_mix(pos));
372 mesh_pipe_ring_commit_push(&ring, &w, pos);
373 mesh_pipe_flight_end(&w, c->slot, parity);
374 }
375 mesh_rt_warp_sync();
376 return cd;
377}
378/* Collective: pops the own inbox's next record. Every lane calls read with a
379 * view of it; then lane 0 frees the cell. PIPE_OK, PIPE_NONE (empty), or a
380 * negative code (no inbox). The same on every lane. */
381MESH_RT_INLINE int32_t mesh_ctx_pop_with(const mesh_ctx *c, mesh_ctx_record_fn read, void *user) {
382 mesh_words w = mesh_rt_gw(c->mb);
383 uint32_t code = (uint32_t)MESH_PIPE_NONE, pos = 0, base = 0, rec = 0;
384 bool have = false;
385 mesh_pipe_ring ring = {0, 0, 0};
386 if (c->lane == 0) {
387 mesh_pipe_inbox in;
388 int r = mesh_pipe_target(&w, c->slot, c->slot, &in);
389 if (r < 0) {
390 code = (uint32_t)mesh_pipe_code(r);
391 } else {
392 /* Both sides, the fuller first. */
393 uint32_t k = mesh_pipe_inbox_first(&in, &w);
394 uint32_t sides[2] = {k, k ^ 1u};
395 for (int i = 0; i < 2; i++) {
396 mesh_pipe_ring rr = mesh_pipe_inbox_side(&in, sides[i]);
397 uint32_t t;
398 if (mesh_pipe_ring_reserve_pop(&rr, &w, &t)) {
399 code = (uint32_t)MESH_PIPE_OK;
400 pos = t;
401 base = (uint32_t)mesh_pipe_ring_cell(&rr, t);
402 rec = rr.rec;
403 ring = rr;
404 have = true;
405 break;
406 }
407 }
408 if (!have) mesh_rt_red_add(c->mb + mesh_pipe_inbox_side(&in, MESH_PIPE_SIDE_CPU).base + MESH_PIPE_PI_EMPTY, 1);
409 }
410 }
411 int32_t cd = (int32_t)mesh_rt_shfl_idx(code, 0);
412 if (cd != MESH_PIPE_OK) return cd;
413 base = mesh_rt_shfl_idx(base, 0);
414 rec = mesh_rt_shfl_idx(rec, 0);
415 /* Lane 0 acquired the cell's lap word: order every lane's reads after it. */
416 mesh_rt_fence_acq_rel();
417 read((mesh_view){c->mb + base, rec, c->ctl}, user);
418 mesh_rt_warp_sync();
419 if (have) mesh_pipe_ring_commit_pop(&ring, &w, pos);
420 mesh_rt_warp_sync();
421 return cd;
422}
423/* Collective: free cells in slot s's inbox (one this program feeds), as lane
424 * 0 read them (the same on every lane); false when it may not push there. */
425MESH_RT_INLINE bool mesh_ctx_pipe_free(const mesh_ctx *c, uint32_t s, uint32_t *out) {
426 uint32_t v = 0;
427 if (c->lane == 0) {
428 mesh_words w = mesh_rt_gw(c->mb);
429 mesh_pipe_inbox in;
430 if (mesh_pipe_target(&w, c->slot, s, &in) < 0) v = UINT32_MAX;
431 else {
432 mesh_pipe_ring rr = mesh_pipe_inbox_side(&in, MESH_PIPE_SIDE_GPU);
433 v = mesh_pipe_ring_free(&rr, &w);
434 }
435 }
436 v = mesh_rt_shfl_idx(v, 0);
437 if (v == UINT32_MAX) return false;
438 *out = v;
439 return true;
440}
441/* Collective: records waiting in slot s's inbox (the own slot, or one this
442 * program feeds), as lane 0 read them; false otherwise. */
443MESH_RT_INLINE bool mesh_ctx_pipe_len(const mesh_ctx *c, uint32_t s, uint32_t *out) {
444 uint32_t v = 0;
445 if (c->lane == 0) {
446 mesh_words w = mesh_rt_gw(c->mb);
447 mesh_pipe_inbox in;
448 v = mesh_pipe_target(&w, c->slot, s, &in) < 0 ? UINT32_MAX : mesh_pipe_inbox_len(&in, &w);
449 }
450 v = mesh_rt_shfl_idx(v, 0);
451 if (v == UINT32_MAX) return false;
452 *out = v;
453 return true;
454}
455
456/* ---- request window (manifest `request`, op OP_REQ) */
457/* The request's bytes (P_REQ_LEN, clamped to req_max). */
458MESH_RT_INLINE mesh_view mesh_ctx_request(const mesh_ctx *c) {
459 uint32_t len = mesh_rt_ld_relaxed(c->ctl + MESH_P_REQ_LEN);
460 if (len > c->req_max) len = c->req_max;
461 return (mesh_view){c->view.base + c->req_off, len, c->ctl};
462}
463/* The response window (resp_max bytes). */
464MESH_RT_INLINE mesh_view mesh_ctx_response(const mesh_ctx *c) { return (mesh_view){c->view.base + c->resp_off, c->resp_max, c->ctl}; }
465/* Collective: publishes len response bytes (P_RESP_LEN, then P_RESP_SEQ = P_REQ_SEQ, release). */
466MESH_RT_INLINE void mesh_ctx_respond(const mesh_ctx *c, uint32_t len) {
467 mesh_rt_warp_sync();
468 if (c->lane == 0) {
469 if (len > c->resp_max) {
470 mesh_rt_fault(c->ctl, MESH_FAULT_RESP_BOUNDS | len << 8);
471 len = c->resp_max;
472 }
473 mesh_rt_wr(c->ctl + MESH_P_RESP_LEN, len);
474 mesh_rt_st_release64(c->ctl + MESH_P_RESP_SEQ, mesh_rt_ld_acquire64(c->ctl + MESH_P_REQ_SEQ));
475 }
476 mesh_rt_warp_sync();
477}
478
479/* ---- lockstep */
480/* The exchange page, only for the program with peer = "lockstep". */
481MESH_RT_INLINE bool mesh_ctx_exchange(const mesh_ctx *c, mesh_exchange *out) {
482 if (c->peer_mode != MESH_PEER_MODE_LOCKSTEP) return false;
483 out->base = c->mb + MESH_EX_BASE;
484 return true;
485}
486
487/* ---- counters (manifest `counters`) */
488/* Adds n to generic counter i (0..4). Call from one lane. */
489MESH_RT_INLINE void mesh_ctx_count(const mesh_ctx *c, uint32_t i, uint32_t n) {
490 if (i < MESH_COUNTERS) mesh_rt_red_add(c->ctl + MESH_P_COUNTERS + 4 * i, n);
491}
492MESH_RT_INLINE uint32_t mesh_ctx_counter(const mesh_ctx *c, uint32_t i) {
493 return i < MESH_COUNTERS ? mesh_rt_ld_relaxed(c->ctl + MESH_P_COUNTERS + 4 * i) : 0;
494}
495
496#endif