1/* @ajax/mesh-abi-pipe-ring: A pipe inbox's two lock-free MPMC rings: laps, record sums, reserve and commit.
2 *
3 * Part of alpha-c-mesh, a C port of alpha-rust-mesh (README.md says from where). */
4#ifndef MESH_ABI_PIPE_RING_H
5#define MESH_ABI_PIPE_RING_H
67#include <stdbool.h>
8#include <stddef.h>
9#include <stdint.h>
10#include "mesh/abi/words.h"
11#include "mesh/abi/pipe_hash.h"
1213/* The ring's words, relative to the inbox start. */
14#define MESH_PIPE_PI_HEAD 0x00
15#define MESH_PIPE_PI_FULL 0x04
16#define MESH_PIPE_PI_TAIL 0x40
17/* Pops that found the ring empty (consumers count them; informational). */
18#define MESH_PIPE_PI_EMPTY 0x44
19#define MESH_PIPE_PI_LAPS 0x80
20/* The largest record (bytes) and inbox (records, both sides). */
21#define MESH_PIPE_REC_MAX 4096u
22#define MESH_PIPE_CAP_MAX ((uint32_t)(1 << 16))
23/* An inbox's producer sides (`Inbox::side`). */
24#define MESH_PIPE_SIDE_CPU 0u
25#define MESH_PIPE_SIDE_GPU 1u
26/* The bound of every ring loop (a CAS lost to another producer or consumer is one try). */
27#define MESH_PIPE_TRIES 256u
28/* What a reserved, not yet committed cell's sum word holds: its producer's slot. */
29#define MESH_PIPE_OWNER_TAG 0x7e57a000u
3031/* What a reserved, not yet committed cell's sum word holds: its producer's slot. */
32static inline uint32_t mesh_pipe_owner_tag(uint32_t slot) { return MESH_PIPE_OWNER_TAG | (slot & 0xfffu); }
3334/* Where the record sums start, and the records. */
35static inline size_t mesh_pipe_inbox_sums(uint32_t cap) { return MESH_PIPE_PI_LAPS + 4 * (size_t)cap; }
36static inline size_t mesh_pipe_inbox_data(uint32_t cap) { return (MESH_PIPE_PI_LAPS + 8 * (size_t)cap + 15) & ~(size_t)15; }
37/* One ring's bytes (CAP records), 64-aligned. */
38static inline size_t mesh_pipe_ring_bytes(uint32_t cap, uint32_t rec) {
39 return (mesh_pipe_inbox_data(cap) + (size_t)cap * rec + 63) & ~(size_t)63;
40}
41/* The inbox's bytes in the view: two rings of cap / 2. */
42static inline size_t mesh_pipe_inbox_bytes(uint32_t cap, uint32_t rec) { return 2 * mesh_pipe_ring_bytes(cap / 2, rec); }
4344/* One ring: where it is and its shape. */
45typedef struct {
46 size_t base; /* the ring's first byte (absolute, as mesh_words takes it) */
47 uint32_t cap;
48 uint32_t rec;
49} mesh_pipe_ring;
5051/* One inbox: two rings of cap / 2, one per producer side. */
52typedef struct {
53 size_t base; /* the inbox's first byte (absolute, as mesh_words takes it) */
54 uint32_t cap; /* records, both sides (the manifest's CAP) */
55 uint32_t rec;
56} mesh_pipe_inbox;
5758static inline bool mesh_pipe_is_pow2(uint32_t x) { return x != 0 && (x & (x - 1)) == 0; }
59/* trailing_zeros, 32 for 0. */
60static inline uint32_t mesh_pipe_ctz(uint32_t x) {
61 uint32_t n = 0;
62 if (x == 0) return 32;
63 while ((x & 1u) == 0) x >>= 1, n++;
64 return n;
65}
66static inline uint32_t mesh_pipe_ring_shift(const mesh_pipe_ring *r) { return mesh_pipe_ctz(r->cap); }
67/* 2^33 / cap - 1: lap words live modulo 2^33 / cap. */
68static inline uint32_t mesh_pipe_ring_lap_mask(const mesh_pipe_ring *r) {
69 return (uint32_t)((UINT64_C(1) << (33 - mesh_pipe_ring_shift(r))) - 1);
70}
71static inline uint32_t mesh_pipe_ring_empty(const mesh_pipe_ring *r, uint32_t p) {
72 return (2 * (p >> mesh_pipe_ring_shift(r))) & mesh_pipe_ring_lap_mask(r);
73}
74static inline uint32_t mesh_pipe_ring_full(const mesh_pipe_ring *r, uint32_t p) {
75 return (2 * (p >> mesh_pipe_ring_shift(r)) + 1) & mesh_pipe_ring_lap_mask(r);
76}
77/* The lap word of position p. */
78static inline size_t mesh_pipe_ring_lap(const mesh_pipe_ring *r, uint32_t p) {
79 return r->base + MESH_PIPE_PI_LAPS + 4 * (size_t)(p & (r->cap - 1));
80}
81/* The sum word of position p's record. */
82static inline size_t mesh_pipe_ring_sum(const mesh_pipe_ring *r, uint32_t p) {
83 return r->base + mesh_pipe_inbox_sums(r->cap) + 4 * (size_t)(p & (r->cap - 1));
84}
85/* The record of position p. */
86static inline size_t mesh_pipe_ring_cell(const mesh_pipe_ring *r, uint32_t p) {
87 return r->base + mesh_pipe_inbox_data(r->cap) + (size_t)(p & (r->cap - 1)) * r->rec;
88}
89/* A shape the runtimes accept (meshd checks manifests the same way). */
90static inline bool mesh_pipe_ring_valid(const mesh_pipe_ring *r) {
91 return mesh_pipe_is_pow2(r->cap) && r->cap >= 2 && r->cap <= MESH_PIPE_CAP_MAX && r->rec >= 4 &&
92 r->rec <= MESH_PIPE_REC_MAX && r->rec % 4 == 0;
93}
9495/* Claims the next position to fill for producer slot `me` (its owner_tag goes
96 * into the cell's sum word): true with *p. False: full (PI_FULL counted), or a
97 * cell no producer may take now (TRIES spent). */
98static inline bool mesh_pipe_ring_reserve_push(const mesh_pipe_ring *r, const mesh_words *w, uint32_t me, uint32_t *p) {
99 for (uint32_t i = 0; i < MESH_PIPE_TRIES; i++) {
100 uint32_t q = mesh_words_load(w, r->base + MESH_PIPE_PI_HEAD);
101 uint32_t l = mesh_words_load(w, mesh_pipe_ring_lap(r, q));
102 if (l == mesh_pipe_ring_empty(r, q)) {
103 if (mesh_words_cas(w, r->base + MESH_PIPE_PI_HEAD, q, q + 1) == q) {
104 mesh_words_store(w, mesh_pipe_ring_sum(r, q), mesh_pipe_owner_tag(me));
105 *p = q;
106 return true;
107 }
108 } else if (mesh_words_load(w, r->base + MESH_PIPE_PI_HEAD) == q) {
109 /* The cell still holds position p - cap (full: backpressure), or a
110 * word that is neither: a cell its producer never committed, or a
111 * ring being reset. Busy either way. */
112 if (l == ((mesh_pipe_ring_empty(r, q) - 1) & mesh_pipe_ring_lap_mask(r)))
113 mesh_words_add(w, r->base + MESH_PIPE_PI_FULL, 1);
114 return false;
115 }
116 }
117 return false;
118}
119/* Publishes position p's record (written at cell(p), its sum at sum(p)). */
120static inline void mesh_pipe_ring_commit_push(const mesh_pipe_ring *r, const mesh_words *w, uint32_t p) {
121 mesh_words_store(w, mesh_pipe_ring_lap(r, p), mesh_pipe_ring_full(r, p));
122}
123124/* Claims the next position to drain: true with *t; false: empty. The caller
125 * checks the record's sum itself (kcheck's sink); runtimes use take. */
126static inline bool mesh_pipe_ring_reserve_pop(const mesh_pipe_ring *r, const mesh_words *w, uint32_t *t) {
127 for (uint32_t i = 0; i < MESH_PIPE_TRIES; i++) {
128 uint32_t q = mesh_words_load(w, r->base + MESH_PIPE_PI_TAIL);
129 uint32_t l = mesh_words_load(w, mesh_pipe_ring_lap(r, q));
130 if (l == mesh_pipe_ring_full(r, q)) {
131 if (mesh_words_cas(w, r->base + MESH_PIPE_PI_TAIL, q, q + 1) == q) {
132 *t = q;
133 return true;
134 }
135 } else if (mesh_words_load(w, r->base + MESH_PIPE_PI_TAIL) == q) {
136 /* Not yet written (empty, or a producer between its reserve and commit). */
137 return false;
138 }
139 }
140 return false;
141}
142/* Reads the next record into out (len bytes, the ring's rec) and claims it:
143 * its sum is checked before the claim, so a record whose words are not visible
144 * here yet (a producer on a device that is coherent only at its own
145 * boundaries) stays in the ring, as if not yet pushed. False: empty (or not
146 * yet visible). The cell is the caller's until commit_pop. */
147static inline bool mesh_pipe_ring_take(const mesh_pipe_ring *r, const mesh_words *w, uint8_t *out, size_t len, uint32_t *t) {
148 for (uint32_t i = 0; i < MESH_PIPE_TRIES; i++) {
149 uint32_t q = mesh_words_load(w, r->base + MESH_PIPE_PI_TAIL);
150 uint32_t l = mesh_words_load(w, mesh_pipe_ring_lap(r, q));
151 if (l == mesh_pipe_ring_full(r, q)) {
152 mesh_words_read(w, mesh_pipe_ring_cell(r, q), out, len);
153 if (mesh_pipe_rec_sum_at(out, len, q) != mesh_words_load(w, mesh_pipe_ring_sum(r, q))) return false;
154 /* A position's record cannot change before its consumer's commit_pop: what was read is it. */
155 if (mesh_words_cas(w, r->base + MESH_PIPE_PI_TAIL, q, q + 1) == q) {
156 *t = q;
157 return true;
158 }
159 } else if (mesh_words_load(w, r->base + MESH_PIPE_PI_TAIL) == q) {
160 return false;
161 }
162 }
163 return false;
164}
165/* Frees position t's cell for the producers (its record was read). */
166static inline void mesh_pipe_ring_commit_pop(const mesh_pipe_ring *r, const mesh_words *w, uint32_t t) {
167 mesh_words_store(w, mesh_pipe_ring_lap(r, t), mesh_pipe_ring_empty(r, t + r->cap));
168}
169170/* Records waiting (reserved or committed): head - tail, at most cap. */
171static inline uint32_t mesh_pipe_ring_len(const mesh_pipe_ring *r, const mesh_words *w) {
172 uint32_t n = mesh_words_load(w, r->base + MESH_PIPE_PI_HEAD) - mesh_words_load(w, r->base + MESH_PIPE_PI_TAIL);
173 return n < r->cap ? n : r->cap;
174}
175/* Free cells now (a hint: other producers may take them). */
176static inline uint32_t mesh_pipe_ring_free(const mesh_pipe_ring *r, const mesh_words *w) {
177 return r->cap - mesh_pipe_ring_len(r, w);
178}
179180/* Calls f(ctx, p) for every position in [tail, head) that producer slot s
181 * reserved and has not committed (its owner tag in the sum word). Only
182 * meaningful once s runs no code: then none of them will ever commit. */
183static inline void mesh_pipe_ring_abandoned(const mesh_pipe_ring *r, const mesh_words *w, uint32_t s,
184 void (*f)(void *ctx, uint32_t p), void *ctx) {
185 uint32_t head = mesh_words_load(w, r->base + MESH_PIPE_PI_HEAD);
186 uint32_t tail = mesh_words_load(w, r->base + MESH_PIPE_PI_TAIL);
187 uint32_t n = head - tail < r->cap ? head - tail : r->cap;
188 for (uint32_t i = 0; i < n; i++) {
189 uint32_t p = tail + i;
190 if (mesh_words_load(w, mesh_pipe_ring_lap(r, p)) == mesh_pipe_ring_empty(r, p) &&
191 mesh_words_load(w, mesh_pipe_ring_sum(r, p)) == mesh_pipe_owner_tag(s))
192 f(ctx, p);
193 }
194}
195/* Releases dead producer s's position p (see abandoned) once the consumers
196 * reached it: 1 skipped, 0 not yet at the front (call again later), -1 not
197 * that dead cell any more (committed after all, or the ring was reset).
198 * Consumers never claim a cell whose lap word is not full, so the CAS on
199 * PI_TAIL races no one. */
200static inline int mesh_pipe_ring_skip_dead(const mesh_pipe_ring *r, const mesh_words *w, uint32_t s, uint32_t p) {
201 if (mesh_words_load(w, mesh_pipe_ring_lap(r, p)) != mesh_pipe_ring_empty(r, p) ||
202 mesh_words_load(w, mesh_pipe_ring_sum(r, p)) != mesh_pipe_owner_tag(s))
203 return -1;
204 uint32_t t = mesh_words_load(w, r->base + MESH_PIPE_PI_TAIL);
205 /* Behind the tail already (a reset restarted the ring's positions): not this cell. */
206 if (p - t >= r->cap) return -1;
207 if (t != p) return 0;
208 if (mesh_words_cas(w, r->base + MESH_PIPE_PI_TAIL, p, p + 1) != p) return 0;
209 /* The tag goes, so a later lap's reserve of this cell is never mistaken for it. */
210 mesh_words_store(w, mesh_pipe_ring_sum(r, p), 0);
211 mesh_pipe_ring_commit_pop(r, w, p);
212 return 1;
213}
214215/* Side k's ring (SIDE_CPU, SIDE_GPU). */
216static inline mesh_pipe_ring mesh_pipe_inbox_side(const mesh_pipe_inbox *in, uint32_t k) {
217 mesh_pipe_ring r = {in->base + (size_t)(k & 1u) * mesh_pipe_ring_bytes(in->cap / 2, in->rec), in->cap / 2, in->rec};
218 return r;
219}
220/* A shape the runtimes accept (meshd checks manifests with it). */
221static inline bool mesh_pipe_inbox_valid(const mesh_pipe_inbox *in) {
222 mesh_pipe_ring r = mesh_pipe_inbox_side(in, 0);
223 return in->cap >= 4 && mesh_pipe_ring_valid(&r) && mesh_pipe_is_pow2(in->cap);
224}
225/* The side a consumer drains first: the fuller one (CPU on a tie). */
226static inline uint32_t mesh_pipe_inbox_first(const mesh_pipe_inbox *in, const mesh_words *w) {
227 mesh_pipe_ring g = mesh_pipe_inbox_side(in, MESH_PIPE_SIDE_GPU), c = mesh_pipe_inbox_side(in, MESH_PIPE_SIDE_CPU);
228 return mesh_pipe_ring_len(&g, w) > mesh_pipe_ring_len(&c, w) ? 1u : 0u;
229}
230/* ring_take on the fuller side, then the other: true with the ring and position taken. */
231static inline bool mesh_pipe_inbox_take(const mesh_pipe_inbox *in, const mesh_words *w, uint8_t *out, size_t len,
232 mesh_pipe_ring *ring, uint32_t *t) {
233 uint32_t k = mesh_pipe_inbox_first(in, w);
234 uint32_t sides[2] = {k, k ^ 1u};
235 for (int i = 0; i < 2; i++) {
236 mesh_pipe_ring r = mesh_pipe_inbox_side(in, sides[i]);
237 if (mesh_pipe_ring_take(&r, w, out, len, t)) {
238 *ring = r;
239 return true;
240 }
241 }
242 return false;
243}
244/* Records waiting, both sides. */
245static inline uint32_t mesh_pipe_inbox_len(const mesh_pipe_inbox *in, const mesh_words *w) {
246 mesh_pipe_ring a = mesh_pipe_inbox_side(in, 0), b = mesh_pipe_inbox_side(in, 1);
247 return mesh_pipe_ring_len(&a, w) + mesh_pipe_ring_len(&b, w);
248}
249250#endif