==> tests/ring.c <==1/* The ring tests of alpha-rust-mesh's mesh-abi pipe.rs, ported one for one:
2 * shape_and_laps, fifo_full_and_wrap, mpmc_threads_lose_and_duplicate_nothing,
3 * position_is_in_the_sum, reserve_push_is_bounded, skip_dead_waits_for_the_front. */
4#include "mesh/abi/pipe_ring.h"
5#include "mesh/check.h"
6#include "mem.h"
7
8#include <threads.h>
9
10static void shape_and_laps(void) {
11 CHECK_U(mesh_pipe_inbox_sums(4), 0x90);
12 CHECK_U(mesh_pipe_inbox_data(4), 0xa0);
13 CHECK_U(mesh_pipe_ring_bytes(256, 32), 0x880 + 256 * 32);
14 CHECK_U(mesh_pipe_inbox_bytes(512, 32), 2 * (0x880 + 256 * 32));
15 CHECK_U(mesh_pipe_ring_bytes(4, 8), 0xc0);
16 mesh_pipe_inbox i = {0x100, 8, 8};
17 mesh_pipe_ring s1 = mesh_pipe_inbox_side(&i, 1);
18 CHECK_U(s1.base, 0x100 + 0xc0);
19 CHECK_U(s1.cap, 4);
20 mesh_pipe_inbox bad = {0, 2, 8};
21 CHECK(mesh_pipe_inbox_valid(&i) && !mesh_pipe_inbox_valid(&bad));
22 uint8_t a[8] = {1, 0, 0, 0, 2, 0, 0, 0}, b[8] = {2, 0, 0, 0, 1, 0, 0, 0};
23 CHECK_U(mesh_pipe_rec_sum(a, 8), mesh_pipe_sum_part(0, 1) + mesh_pipe_sum_part(1, 2));
24 CHECK(mesh_pipe_rec_sum(a, 8) != mesh_pipe_rec_sum(b, 8));
25 mesh_pipe_ring r = {0, 4, 8}, r3 = {0, 3, 8}, r6 = {0, 4, 6};
26 CHECK(mesh_pipe_ring_valid(&r) && !mesh_pipe_ring_valid(&r3) && !mesh_pipe_ring_valid(&r6));
27 CHECK_U(mesh_pipe_ring_empty(&r, 0), 0);
28 CHECK_U(mesh_pipe_ring_full(&r, 0), 1);
29 CHECK_U(mesh_pipe_ring_empty(&r, 4), 2);
30 CHECK_U(mesh_pipe_ring_full(&r, 7), 3);
31 /* Positions wrap u32: the lap of p = 2^32 - 1 is followed by lap 0's words. */
32 uint32_t last = UINT32_MAX;
33 CHECK_U(mesh_pipe_ring_empty(&r, last + 1), (mesh_pipe_ring_full(&r, last) + 1) & mesh_pipe_ring_lap_mask(&r));
34 CHECK_U(mesh_pipe_type_hash((const uint8_t *)"", 0), 0x811c9dc5u);
35}
36
37static void fifo_full_and_wrap(void) {
38 mesh_pipe_ring r = {0x100, 4, 8};
39 mem *m = mem_new(0x100 + mesh_pipe_ring_bytes(4, 8));
40 const mesh_words *w = &m->words;
41 /* Start near the u32 wrap to cross it. */
42 uint32_t start = UINT32_MAX - 5;
43 mem_set(m, 0x100 + MESH_PIPE_PI_HEAD, start);
44 mem_set(m, 0x100 + MESH_PIPE_PI_TAIL, start);
45 for (uint32_t i = 0; i < 4; i++) mem_set(m, mesh_pipe_ring_lap(&r, start + i), mesh_pipe_ring_empty(&r, start + i));
46 uint32_t next = 0, got = 0, p, t;
47 for (uint32_t round = 0; round < 10; round++) {
48 while (mesh_pipe_ring_reserve_push(&r, w, 1, &p)) {
49 uint32_t v[2] = {next, ~next};
50 mem_put(m, &r, p, v, 2);
51 mesh_pipe_ring_commit_push(&r, w, p);
52 next++;
53 }
54 CHECK_U(mesh_pipe_ring_len(&r, w), 4);
55 for (uint32_t k = 0; k < 1 + round % 4; k++) {
56 CHECK(mesh_pipe_ring_reserve_pop(&r, w, &t));
57 CHECK_U(mem_cell_word(m, &r, t, 0), got);
58 CHECK_U(mem_cell_word(m, &r, t, 1), ~got);
59 mesh_pipe_ring_commit_pop(&r, w, t);
60 got++;
61 }
62 }
63 while (mesh_pipe_ring_reserve_pop(&r, w, &t)) {
64 CHECK_U(mem_cell_word(m, &r, t, 0), got);
65 mesh_pipe_ring_commit_pop(&r, w, t);
66 got++;
67 }
68 CHECK_U(got, next);
69 CHECK(mem_get(m, 0x100 + MESH_PIPE_PI_FULL) >= 10);
70 mem_free(m);
71}
72
73enum { PRODUCERS = 4, PER = 50000, CONSUMERS = 3 };
74static const mesh_pipe_ring MPMC = {0, 64, 8};
75static mem *mpmc_mem;
76static _Atomic uint32_t seen[PRODUCERS * PER];
77static _Atomic uint32_t done, torn;
78
79static int producer(void *arg) {
80 uint32_t k = (uint32_t)(uintptr_t)arg, p;
81 for (uint32_t i = 0; i < PER; i++) {
82 uint32_t v = k * PER + i;
83 for (;;) {
84 if (mesh_pipe_ring_reserve_push(&MPMC, &mpmc_mem->words, k, &p)) {
85 uint32_t rec[2] = {v, v ^ 0xdeadbeefu};
86 mem_put(mpmc_mem, &MPMC, p, rec, 2);
87 mesh_pipe_ring_commit_push(&MPMC, &mpmc_mem->words, p);
88 break;
89 }
90 thrd_yield();
91 }
92 }
93 return 0;
94}
95static int consumer(void *arg) {
96 (void)arg;
97 uint32_t t;
98 while (atomic_load_explicit(&done, memory_order_relaxed) < PRODUCERS * PER) {
99 if (mesh_pipe_ring_reserve_pop(&MPMC, &mpmc_mem->words, &t)) {
100 uint32_t x0 = mem_cell_word(mpmc_mem, &MPMC, t, 0), x1 = mem_cell_word(mpmc_mem, &MPMC, t, 1);
101 mesh_pipe_ring_commit_pop(&MPMC, &mpmc_mem->words, t);
102 if (x1 != (x0 ^ 0xdeadbeefu) || x0 >= PRODUCERS * PER) atomic_fetch_add(&torn, 1);
103 else atomic_fetch_add_explicit(&seen[x0], 1, memory_order_relaxed);
104 atomic_fetch_add_explicit(&done, 1, memory_order_relaxed);
105 } else {
106 thrd_yield();
107 }
108 }
109 return 0;
110}
111static void mpmc_threads_lose_and_duplicate_nothing(void) {
112 mpmc_mem = mem_new(mesh_pipe_ring_bytes(64, 8));
113 thrd_t th[PRODUCERS + CONSUMERS];
114 for (uint32_t k = 0; k < PRODUCERS; k++) thrd_create(&th[k], producer, (void *)(uintptr_t)k);
115 for (int k = 0; k < CONSUMERS; k++) thrd_create(&th[PRODUCERS + k], consumer, NULL);
116 for (int k = 0; k < PRODUCERS + CONSUMERS; k++) thrd_join(th[k], NULL);
117 CHECK_U(atomic_load(&torn), 0); /* "torn record" */
118 uint32_t once = 0;
119 for (uint32_t i = 0; i < PRODUCERS * PER; i++) once += atomic_load(&seen[i]) == 1;
120 CHECK_U(once, PRODUCERS * PER);
121 CHECK_U(mesh_pipe_ring_len(&MPMC, &mpmc_mem->words), 0);
122 mem_free(mpmc_mem);
123}
124
125static void position_is_in_the_sum(void) {
126 /* A cell and sum both left from the previous lap behind a fresh lap word fail the check (finding 5). */
127 mesh_pipe_ring r = {0, 4, 8};
128 mem *m = mem_new(mesh_pipe_ring_bytes(4, 8));
129 const mesh_words *w = &m->words;
130 uint8_t out[8];
131 uint32_t p0, t, p;
132 CHECK(mesh_pipe_pos_mix(0) != mesh_pipe_pos_mix(4));
133 CHECK(mesh_pipe_ring_reserve_push(&r, w, 3, &p0));
134 CHECK_U(mem_get(m, mesh_pipe_ring_sum(&r, p0)), mesh_pipe_owner_tag(3));
135 uint32_t v78[2] = {7, 8};
136 mem_put(m, &r, p0, v78, 2);
137 mesh_pipe_ring_commit_push(&r, w, p0);
138 CHECK(mesh_pipe_ring_take(&r, w, out, 8, &t));
139 mesh_pipe_ring_commit_pop(&r, w, t);
140 /* Producer of position 4 (same cell) advanced head and published its lap word, but the
141 * consumer still sees the old cell and sum (a stale cache line): not taken. */
142 for (int i = 1; i < 4; i++) {
143 CHECK(mesh_pipe_ring_reserve_push(&r, w, 3, &p));
144 uint32_t v[2] = {p, p};
145 mem_put(m, &r, p, v, 2);
146 mesh_pipe_ring_commit_push(&r, w, p);
147 CHECK(mesh_pipe_ring_take(&r, w, out, 8, &t));
148 mesh_pipe_ring_commit_pop(&r, w, t);
149 }
150 uint32_t p4;
151 CHECK(mesh_pipe_ring_reserve_push(&r, w, 3, &p4));
152 CHECK_U(p4, 4);
153 const uint8_t old[8] = {7, 0, 0, 0, 8, 0, 0, 0};
154 mem_set(m, mesh_pipe_ring_sum(&r, p4), mesh_pipe_rec_sum_at(old, 8, 0));
155 mem_set(m, mesh_pipe_ring_cell(&r, p4), 7);
156 mem_set(m, mesh_pipe_ring_cell(&r, p4) + 4, 8);
157 mesh_pipe_ring_commit_push(&r, w, p4);
158 CHECK(!mesh_pipe_ring_take(&r, w, out, 8, &t)); /* a stale record from position 0 passed as position 4 */
159 mem_put(m, &r, p4, v78, 2);
160 CHECK(mesh_pipe_ring_take(&r, w, out, 8, &t) && t == 4);
161 mem_free(m);
162}
163
164static void collect(void *ctx, uint32_t p) {
165 uint32_t *cells = ctx;
166 cells[1 + cells[0]++] = p;
167}
168static void fail_cell(void *ctx, uint32_t p) {
169 (void)ctx;
170 CHECK(!"slot 1 committed everything");
171 (void)p;
172}
173static void reserve_push_is_bounded(void) {
174 /* A lap word that is neither empty(p) nor the previous lap's full: busy, at once (finding 2). */
175 mesh_pipe_ring r = {0, 4, 8};
176 mem *m = mem_new(mesh_pipe_ring_bytes(4, 8));
177 uint32_t p, t;
178 mem_set(m, mesh_pipe_ring_lap(&r, 0), 0xbb);
179 CHECK(!mesh_pipe_ring_reserve_push(&r, &m->words, 1, &p));
180 CHECK_U(mem_get(m, MESH_PIPE_PI_FULL), 0); /* not counted as backpressure */
181 mem_free(m);
182 /* A producer that reserved position 0 and died; the ring wrapped to its cell: busy, not a spin. */
183 m = mem_new(mesh_pipe_ring_bytes(4, 8));
184 const mesh_words *w = &m->words;
185 uint32_t dead;
186 CHECK(mesh_pipe_ring_reserve_push(&r, w, 9, &dead));
187 for (int i = 1; i < 4; i++) {
188 CHECK(mesh_pipe_ring_reserve_push(&r, w, 1, &p));
189 uint32_t v[2] = {p, 0};
190 mem_put(m, &r, p, v, 2);
191 mesh_pipe_ring_commit_push(&r, w, p);
192 }
193 CHECK(!mesh_pipe_ring_reserve_push(&r, w, 1, &p));
194 uint8_t out[8];
195 CHECK(!mesh_pipe_ring_take(&r, w, out, 8, &t)); /* the dead cell blocks the consumer */
196 /* meshd: the dead slot's cells, released at the front. */
197 uint32_t cells[8] = {0};
198 mesh_pipe_ring_abandoned(&r, w, 9, collect, cells);
199 CHECK_U(cells[0], 1);
200 CHECK_U(cells[1], dead);
201 mesh_pipe_ring_abandoned(&r, w, 1, fail_cell, NULL);
202 CHECK_I(mesh_pipe_ring_skip_dead(&r, w, 9, dead), 1);
203 CHECK_I(mesh_pipe_ring_skip_dead(&r, w, 9, dead), -1);
204 for (uint32_t want = 1; want < 4; want++) {
205 CHECK(mesh_pipe_ring_take(&r, w, out, 8, &t));
206 CHECK_U(t, want);
207 mesh_pipe_ring_commit_pop(&r, w, t);
208 }
209 /* The dead cell is usable again: the ring is whole. */
210 for (int i = 0; i < 4; i++) {
211 CHECK(mesh_pipe_ring_reserve_push(&r, w, 1, &p));
212 uint32_t v[2] = {p, 0};
213 mem_put(m, &r, p, v, 2);
214 mesh_pipe_ring_commit_push(&r, w, p);
215 }
216 CHECK_U(mesh_pipe_ring_len(&r, w), 4);
217 mem_free(m);
218}
219
220static void skip_dead_waits_for_the_front(void) {
221 mesh_pipe_ring r = {0, 8, 8};
222 mem *m = mem_new(mesh_pipe_ring_bytes(8, 8));
223 const mesh_words *w = &m->words;
224 uint32_t a, dead, t;
225 CHECK(mesh_pipe_ring_reserve_push(&r, w, 1, &a));
226 uint32_t v[2] = {1, 1};
227 mem_put(m, &r, a, v, 2);
228 mesh_pipe_ring_commit_push(&r, w, a);
229 CHECK(mesh_pipe_ring_reserve_push(&r, w, 2, &dead));
230 CHECK_I(mesh_pipe_ring_skip_dead(&r, w, 2, dead), 0); /* position 0 is still queued */
231 uint8_t out[8];
232 CHECK(mesh_pipe_ring_take(&r, w, out, 8, &t));
233 mesh_pipe_ring_commit_pop(&r, w, t);
234 CHECK_I(mesh_pipe_ring_skip_dead(&r, w, 2, dead), 1);
235 CHECK_U(mesh_pipe_ring_len(&r, w), 0);
236 mem_free(m);
237}
238
239int main(void) {
240 shape_and_laps();
241 fifo_full_and_wrap();
242 mpmc_threads_lose_and_duplicate_nothing();
243 position_is_in_the_sum();
244 reserve_push_is_bounded();
245 skip_dead_waits_for_the_front();
246 return mesh_check_done();
247}