==> src/group.c <==1/* peer.rs: the lockstep group's data plane: the reconciler's wish against the HELLO run here, the
2 * HELLO exchange, pausing, and the rounds (the sender thread and the reader's F_ROUND). */
3#include "relay.h"
4#include "mesh/abi/peer_sizes.h"
5#include "mesh/abi/exchange.h"
6#include "mesh/abi/group_state.h"
7#include "mesh/abi/peer_frames.h"
8
9#include <stdio.h>
10#include <stdlib.h>
11#include <string.h>
12#include <time.h>
13
14void group_set_state(mpr_relay *r, uint32_t s, const char *why) {
15 atomic_store(&r->hello_state, s);
16 mt_set(&r->hello_why, why);
17}
18
19uint32_t mpr_group_state(mpr_relay *r) { return atomic_load(&r->hello_state); }
20
21void mpr_group_want(mpr_relay *r, const mpk_hello *spec, bool mirror, bool member) {
22 pthread_mutex_lock(&r->group_mu);
23 r->want_set = spec != NULL;
24 if (spec) r->want = *spec;
25 r->want_mirror = mirror;
26 r->want_member = member;
27 pthread_mutex_unlock(&r->group_mu);
28 mpr_wake(r);
29}
30
31void group_hello_send(mpr_relay *r) {
32 r->hello_sent_ms = mpr_mono_ms();
33 if (!r->has_hello) return;
34 mpw_bytes f = mpk_hello_frame(r->me, &r->hello);
35 for (uint32_t k = 0; k < r->n; k++)
36 if (k != r->me && (r->hello.mask >> k & 1)) relay_send(r, k, &f);
37 mpw_bytes_free(&f);
38}
39
40/* Every other member's HELLO matches ours: run. */
41static void hello_check(mpr_relay *r) {
42 if (atomic_load(&r->hello_state) != MESH_GL_HELLO || !r->has_hello) return;
43 mpk_hello *h = &r->hello;
44 ml_list missing = {0};
45 for (uint32_t k = 0; k < r->n; k++) {
46 if (k == r->me || !(h->mask >> k & 1)) continue;
47 if (r->has_theirs[k] && mpk_hello_eq(&r->theirs[k], h)) continue;
48 if (r->has_theirs[k] && !(r->has_mismatch[k] && mpk_hello_eq(&r->mismatch[k], &r->theirs[k]))) {
49 const mpk_hello *t = &r->theirs[k];
50 ml_list tm = ml_members(t->mask), hm = ml_members(h->mask);
51 char *a = ml_text(&tm), *b = ml_text(&hm);
52 relay_log(r, false, mt_fmt("group %s: system %u is at epoch %u cycle %u members %s image %016llx; here epoch %u cycle %u members %s image %016llx (waiting)",
53 h->group, k, t->epoch, t->cycle, a, (unsigned long long)t->hash, h->epoch, h->cycle, b, (unsigned long long)h->hash));
54 free(a), free(b);
55 ml_free(&tm), ml_free(&hm);
56 r->has_mismatch[k] = true;
57 r->mismatch[k] = *t;
58 }
59 ml_push(&missing, k);
60 }
61 if (missing.n) {
62 char *l = ml_text(&missing), *why = mt_fmt("waiting for the HELLO of system(s) %s", l);
63 mt_set(&r->hello_why, why);
64 free(l), free(why);
65 ml_free(&missing);
66 return;
67 }
68 atomic_store(&r->mirroring, true);
69 r->h.lock_feed(r->h.ctx, true);
70 group_set_state(r, MESH_GL_RUNNING, "running");
71 ml_list hm = ml_members(h->mask);
72 char *ml = ml_text(&hm);
73 uint32_t rank = 0;
74 mpk_hello_rank(h, r->me, &rank);
75 relay_log(r, false, mt_fmt("group %s: epoch %u cycle %u running on %u member(s) %s (rank %u here)", h->group, h->epoch, h->cycle, mpk_hello_n(h), ml, rank));
76 free(ml);
77 ml_free(&hm);
78}
79
80void group_hello(mpr_relay *r, uint32_t from, const mpk_hello *spec) {
81 r->theirs[from] = *spec;
82 r->has_theirs[from] = true;
83 hello_check(r);
84}
85
86/* A new epoch or cycle is ready here: retag the rounds, empty the members' copies, HELLO them. */
87static void hello_begin(mpr_relay *r, const mpk_hello *spec) {
88 pthread_mutex_lock(&r->send_mu);
89 atomic_store(&r->mirroring, false);
90 pthread_mutex_lock(&r->tag_mu);
91 uint8_t rank_of[MAXS];
92 memset(rank_of, 0xff, sizeof rank_of);
93 for (uint32_t rk = 0; rk < MAXS; rk++)
94 if (spec->order[rk] < MAXS) rank_of[spec->order[rk]] = (uint8_t)rk;
95 r->tag = (tag_t){.epoch = spec->epoch, .cycle = spec->cycle};
96 memcpy(r->tag.rank_of, rank_of, sizeof rank_of);
97 r->tag.me = rank_of[r->me];
98 /* Only the members' rounds of this (epoch, cycle) land here from now on. */
99 uint8_t zero[MESH_EX_REGION_BYTES] = {0};
100 for (uint32_t rk = 0; rk < mpk_hello_n(spec); rk++)
101 if (rk != r->tag.me) {
102 r->h.mb_write(r->h.ctx, mesh_ex_slots(rk), zero, MESH_EX_REGION_BYTES);
103 r->h.mb_publish64(r->h.ctx, mesh_ex_counter(rk), 0);
104 }
105 pthread_mutex_unlock(&r->tag_mu);
106 pthread_mutex_unlock(&r->send_mu);
107 atomic_store(&r->last_sent, UINT64_MAX);
108 pthread_mutex_lock(&r->links_mu);
109 for (uint32_t k = 0; k < MAXS; k++)
110 if (r->links[k]) {
111 pthread_mutex_lock(&r->links[k]->sent_mu);
112 r->links[k]->sent_n = 0;
113 pthread_mutex_unlock(&r->links[k]->sent_mu);
114 atomic_store(&r->links[k]->last_round, 0);
115 }
116 pthread_mutex_unlock(&r->links_mu);
117 pthread_mutex_lock(&r->own_mu);
118 r->own_n = 0;
119 pthread_mutex_unlock(&r->own_mu);
120 memset(r->has_mismatch, 0, sizeof r->has_mismatch);
121 ml_list hm = ml_members(spec->mask);
122 char *ml = ml_text(&hm), rank[16];
123 uint32_t rk;
124 if (mpk_hello_rank(spec, r->me, &rk)) snprintf(rank, sizeof rank, "%u", rk);
125 else strcpy(rank, "-");
126 relay_log(r, false, mt_fmt("group %s: ready for epoch %u cycle %u (members %s, rank %s, image %016llx); HELLO", spec->group, spec->epoch, spec->cycle, ml, rank,
127 (unsigned long long)spec->hash));
128 free(ml);
129 ml_free(&hm);
130 r->hello = *spec;
131 r->has_hello = true;
132 group_set_state(r, MESH_GL_HELLO, "waiting for every member's HELLO");
133 group_hello_send(r);
134 hello_check(r);
135}
136
137/* The reconciler's wish against the HELLO run here. */
138void group_want(mpr_relay *r) {
139 pthread_mutex_lock(&r->group_mu);
140 bool set = r->want_set, mirror = r->want_mirror, member = r->want_member;
141 mpk_hello want = r->want;
142 pthread_mutex_unlock(&r->group_mu);
143 if (set) {
144 if (!r->has_hello || !mpk_hello_eq(&want, &r->hello)) hello_begin(r, &want);
145 return;
146 }
147 /* Quiesced by the reconciler (a cycle keeps mirroring until its HELLO; an epoch change stops now). */
148 if (!mirror && atomic_load(&r->mirroring)) {
149 pthread_mutex_lock(&r->send_mu);
150 atomic_store(&r->mirroring, false);
151 pthread_mutex_unlock(&r->send_mu);
152 }
153 r->h.lock_feed(r->h.ctx, false);
154 r->has_hello = false;
155 uint32_t st = member ? MESH_GL_PAUSED : MESH_GL_NONE;
156 if (atomic_load(&r->hello_state) != st) group_set_state(r, st, member ? "quiesced (the group changes)" : "not in the group's epoch");
157}
158
159void group_pause(mpr_relay *r, const char *why) {
160 pthread_mutex_lock(&r->send_mu);
161 atomic_store(&r->mirroring, false);
162 pthread_mutex_unlock(&r->send_mu);
163 r->h.lock_feed(r->h.ctx, false);
164 if (r->has_hello)
165 relay_log(r, true, mt_fmt("group %s: paused at epoch %u (%s); it resumes as a new epoch (reset) or re-forms", r->hello.group, r->hello.epoch, why));
166 group_set_state(r, MESH_GL_PAUSED, why);
167}
168
169/* A member reached round n: our rounds <= n made the trip (rtt), and how long after our own round n its
170 * round n came (the wait for it). */
171static void round_stats(mpr_relay *r, linkrec *lr, uint64_t n) {
172 if (n <= atomic_exchange(&lr->last_round, n)) return;
173 uint64_t now = mpr_mono_us();
174 double rtt = -1;
175 pthread_mutex_lock(&lr->sent_mu);
176 size_t i = 0;
177 uint64_t last = 0;
178 bool any = false;
179 while (i < lr->sent_n && lr->sent_c[i] <= n) last = lr->sent_t[i++], any = true;
180 memmove(lr->sent_c, lr->sent_c + i, (lr->sent_n - i) * sizeof *lr->sent_c);
181 memmove(lr->sent_t, lr->sent_t + i, (lr->sent_n - i) * sizeof *lr->sent_t);
182 lr->sent_n -= i;
183 pthread_mutex_unlock(&lr->sent_mu);
184 if (any) rtt = (double)(now - last);
185 double wait = 0;
186 pthread_mutex_lock(&r->own_mu);
187 for (size_t j = 0; j < r->own_n; j++)
188 if (r->own_c[j] >= n) {
189 wait = now > r->own_t[j] ? (double)(now - r->own_t[j]) : 0;
190 break;
191 }
192 pthread_mutex_unlock(&r->own_mu);
193 if (r->h.round_stats) r->h.round_stats(r->h.ctx, lr->from, n, rtt, wait);
194}
195
196void group_round(mpr_relay *r, linkrec *lr, const uint8_t *b) {
197 uint32_t e = mpw_u32_at(b, 0), c = mpw_u32_at(b, 4);
198 uint64_t n = mpw_u64_at(b, 8);
199 pthread_mutex_lock(&r->tag_mu);
200 uint8_t rank = r->tag.rank_of[lr->from];
201 /* A round of another epoch or cycle is stale (or early): dropped. */
202 if (e == r->tag.epoch && c == r->tag.cycle && rank != 0xff && rank != r->tag.me) {
203 r->h.mb_write(r->h.ctx, mesh_ex_slots(rank), b + MESH_ROUND_HEAD, MESH_EX_REGION_BYTES);
204 r->h.mb_publish64(r->h.ctx, mesh_ex_counter(rank), n);
205 pthread_mutex_unlock(&r->tag_mu);
206 round_stats(r, lr, n);
207 return;
208 }
209 pthread_mutex_unlock(&r->tag_mu);
210}
211
212static void sleep_us(long us) {
213 struct timespec ts = {us / 1000000, (us % 1000000) * 1000};
214 nanosleep(&ts, NULL);
215}
216
217/* Mirrors our exchange region to every other group member whenever our counter moves (~20 µs polling
218 * bounds the step rate), tagged with the group's epoch and cycle. */
219void *group_sender(void *arg) {
220 mpr_relay *r = arg;
221 while (!atomic_load(&r->quit)) {
222 if (!atomic_load(&r->mirroring)) {
223 sleep_us(2000);
224 continue;
225 }
226 /* The round and its targets under the lock (a retag waits for it); the writes after it. */
227 mpw_bytes f = {0};
228 uint64_t n = 0, now = 0;
229 linkrec *targets[MAXS];
230 size_t nt = 0;
231 pthread_mutex_lock(&r->send_mu);
232 if (atomic_load(&r->mirroring)) {
233 pthread_mutex_lock(&r->tag_mu);
234 tag_t t = r->tag;
235 pthread_mutex_unlock(&r->tag_mu);
236 n = r->h.mb_get64(r->h.ctx, mesh_ex_counter(t.me));
237 if (n != atomic_load(&r->last_sent)) {
238 now = mpr_mono_us();
239 uint8_t ty = MESH_F_ROUND, region[MESH_EX_REGION_BYTES], nb[8];
240 mpw_put(&f, &ty, 1);
241 mpw_put32(&f, t.epoch);
242 mpw_put32(&f, t.cycle);
243 memcpy(nb, &n, 8);
244 mpw_put(&f, nb, 8);
245 r->h.mb_read(r->h.ctx, mesh_ex_slots(t.me), region, sizeof region);
246 mpw_put(&f, region, sizeof region);
247 pthread_mutex_lock(&r->own_mu);
248 if (!r->own_n || r->own_c[r->own_n - 1] < n) {
249 if (r->own_n >= 64) {
250 memmove(r->own_c, r->own_c + 1, 63 * sizeof *r->own_c);
251 memmove(r->own_t, r->own_t + 1, 63 * sizeof *r->own_t);
252 r->own_n--;
253 }
254 r->own_c[r->own_n] = n, r->own_t[r->own_n++] = now;
255 }
256 pthread_mutex_unlock(&r->own_mu);
257 pthread_mutex_lock(&r->links_mu);
258 for (uint32_t k = 0; k < MAXS; k++) {
259 linkrec *lr = r->links[k];
260 if (!lr || k == r->me || t.rank_of[k] == 0xff || !mpn_link_alive(lr->l)) continue;
261 pthread_mutex_lock(&lr->sent_mu);
262 if (lr->sent_n >= 64) {
263 memmove(lr->sent_c, lr->sent_c + 1, 63 * sizeof *lr->sent_c);
264 memmove(lr->sent_t, lr->sent_t + 1, 63 * sizeof *lr->sent_t);
265 lr->sent_n--;
266 }
267 lr->sent_c[lr->sent_n] = n, lr->sent_t[lr->sent_n++] = now;
268 pthread_mutex_unlock(&lr->sent_mu);
269 atomic_fetch_add(&lr->refs, 1);
270 targets[nt++] = lr;
271 }
272 pthread_mutex_unlock(&r->links_mu);
273 atomic_store(&r->last_sent, n);
274 pthread_mutex_unlock(&r->send_mu);
275 /* The writes after the locks: a member that does not read (frozen, gone) never holds up
276 * a retag, a pause or the fleet thread. The reference keeps the link alive meanwhile. */
277 for (size_t i = 0; i < nt; i++) {
278 linkrec *lr = targets[i];
279 if (!mpn_write_now(lr->l, f.b, f.n)) mpn_link_close(lr->l);
280 atomic_fetch_sub(&lr->refs, 1);
281 }
282 mpw_bytes_free(&f);
283 sleep_us(20);
284 continue;
285 }
286 }
287 pthread_mutex_unlock(&r->send_mu);
288 sleep_us(20);
289 }
290 return NULL;
291}