==> src/peer_codec.c <==1/* peer.rs: Caps, Join, BeatStatus, frame, msg_frame, beat_frame, moves_bytes, moves_decode,
2 * beat_decode, hello_frame, hello_decode; shared.rs HelloSpec. */
3#include "mesh/peer_codec.h"
4#include "mesh/abi/peer_frames.h"
5#include "mesh/abi/peer_sizes.h"
6#include "mesh/mt.h"
7
8#include <stdlib.h>
9#include <string.h>
10
11static void put64(mpw_bytes *o, uint64_t v) {
12 mpw_put32(o, (uint32_t)v);
13 mpw_put32(o, (uint32_t)(v >> 32));
14}
15
16static void str_field(mpw_bytes *o, const char *s, size_t n) {
17 uint8_t b[64] = {0};
18 size_t k = strlen(s);
19 if (k > n - 1) k = n - 1;
20 memcpy(b, s, k);
21 mpw_put(o, b, n);
22}
23
24static void cstr(char *out, const uint8_t *b, size_t n) {
25 size_t k = 0;
26 while (k < n && b[k]) k++;
27 memcpy(out, b, k);
28 out[k] = 0;
29}
30
31mpw_bytes mpk_join_encode(const mpk_join *j) {
32 mpw_bytes o = {0};
33 uint8_t t = MESH_F_JOIN;
34 const mpk_caps *c = &j->caps;
35 mpw_put(&o, &t, 1);
36 uint32_t v[] = {j->system, j->version, j->systems, c->cc, c->sms, c->sms_total, c->blocks_per_sm, c->backend};
37 for (int i = 0; i < 8; i++) mpw_put32(&o, v[i]);
38 str_field(&o, c->gpu, 32);
39 str_field(&o, c->arch, 16);
40 str_field(&o, c->host, 32);
41 mpw_put32(&o, c->max_threads);
42 mpw_put32(&o, c->max_shared);
43 return o;
44}
45
46mpk_join mpk_join_decode(const uint8_t *b) {
47 mpk_join j = {mpw_u32_at(b, 0), mpw_u32_at(b, 4), mpw_u32_at(b, 8), {0}};
48 mpk_caps *c = &j.caps;
49 c->cc = mpw_u32_at(b, 12), c->sms = mpw_u32_at(b, 16), c->sms_total = mpw_u32_at(b, 20), c->blocks_per_sm = mpw_u32_at(b, 24);
50 c->backend = mpw_u32_at(b, 28);
51 cstr(c->gpu, b + 32, 32);
52 cstr(c->arch, b + 64, 16);
53 cstr(c->host, b + 80, 32);
54 c->max_threads = mpw_u32_at(b, 112), c->max_shared = mpw_u32_at(b, 116);
55 return j;
56}
57
58bool mpk_join_check(const mpk_join *j, uint32_t me, uint32_t n, bool has_expect, uint32_t expect, char **err) {
59 if (j->version != MESH_PEER_VERSION) {
60 *err = mt_fmt("system %u runs peer-link version %u, we run %u", j->system, j->version, (unsigned)MESH_PEER_VERSION);
61 return false;
62 }
63 if (j->systems != n) {
64 *err = mt_fmt("system %u is in a cluster of %u, we are in one of %u", j->system, j->systems, n);
65 return false;
66 }
67 if (j->system >= n || j->system == me) {
68 *err = mt_fmt("a JOIN from system %u (we are system %u of %u)", j->system, me, n);
69 return false;
70 }
71 if (has_expect && j->system != expect) {
72 *err = mt_fmt("dialed system %u, it says it is system %u", expect, j->system);
73 return false;
74 }
75 if (!has_expect && j->system >= me) {
76 *err = mt_fmt("system %u dialed us (system %u); only lower systems dial", j->system, me);
77 return false;
78 }
79 return true;
80}
81
82bool mpk_caps_eq(const mpk_caps *a, const mpk_caps *b) {
83 return a->cc == b->cc && a->sms == b->sms && a->sms_total == b->sms_total && a->blocks_per_sm == b->blocks_per_sm && !strcmp(a->gpu, b->gpu) &&
84 !strcmp(a->arch, b->arch) && !strcmp(a->host, b->host) && a->backend == b->backend && a->max_threads == b->max_threads && a->max_shared == b->max_shared;
85}
86
87void mpk_beat_status_free(mpk_beat_status *s) {
88 free(s->reason);
89 for (size_t i = 0; i < s->nmoves; i++) free(s->moves[i].name);
90 free(s->moves);
91 *s = (mpk_beat_status){0};
92}
93
94mpk_beat_status mpk_beat_status_clone(const mpk_beat_status *s) {
95 mpk_beat_status c = *s;
96 c.reason = mt_strdup(s->reason ? s->reason : "");
97 c.moves = mt_alloc((s->nmoves ? s->nmoves : 1) * sizeof *c.moves);
98 for (size_t i = 0; i < s->nmoves; i++) c.moves[i] = (mst_outcome){mt_strdup(s->moves[i].name), s->moves[i].version, s->moves[i].ok, s->moves[i].gap_us};
99 return c;
100}
101
102bool mpk_beat_status_eq(const mpk_beat_status *a, const mpk_beat_status *b) {
103 if (a->has_leader != b->has_leader || (a->has_leader && a->leader != b->leader) || a->applied != b->applied || a->status != b->status || a->up != b->up ||
104 a->group_epoch != b->group_epoch || a->group_state != b->group_state || strcmp(a->reason ? a->reason : "", b->reason ? b->reason : "") || a->nmoves != b->nmoves)
105 return false;
106 for (size_t i = 0; i < a->nmoves; i++)
107 if (strcmp(a->moves[i].name, b->moves[i].name) || a->moves[i].version != b->moves[i].version || a->moves[i].ok != b->moves[i].ok || a->moves[i].gap_us != b->moves[i].gap_us)
108 return false;
109 return true;
110}
111
112mpw_bytes mpk_msg_frame(const mfc_msg *m) {
113 mpw_bytes o = {0};
114 uint8_t t;
115 switch (m->kind) {
116 case MFC_VOTE_REQ:
117 t = MESH_F_VOTE_REQ;
118 mpw_put(&o, &t, 1);
119 mpw_put32(&o, m->pre);
120 put64(&o, m->term);
121 put64(&o, m->acc.term);
122 put64(&o, m->acc.version);
123 break;
124 case MFC_VOTE:
125 t = MESH_F_VOTE;
126 mpw_put(&o, &t, 1);
127 mpw_put32(&o, m->pre);
128 put64(&o, m->term);
129 mpw_put32(&o, m->granted);
130 break;
131 case MFC_STATE: {
132 t = MESH_F_STATE;
133 mpw_put(&o, &t, 1);
134 char *doc = mfd_encode(&m->doc);
135 put64(&o, m->term);
136 mpw_put32(&o, (uint32_t)strlen(doc));
137 mpw_put(&o, doc, strlen(doc));
138 free(doc);
139 break;
140 }
141 case MFC_STATE_ACK:
142 case MFC_COMMIT:
143 t = m->kind == MFC_STATE_ACK ? MESH_F_STATE_ACK : MESH_F_COMMIT;
144 mpw_put(&o, &t, 1);
145 put64(&o, m->term);
146 put64(&o, m->ver.term);
147 put64(&o, m->ver.version);
148 break;
149 case MFC_BEAT: /* beats carry the status: mpk_beat_frame */
150 break;
151 }
152 return o;
153}
154
155mpw_bytes mpk_beat_frame(const mfc_msg *m, const mpk_beat_status *st) {
156 mpw_bytes o = {0};
157 uint8_t t = MESH_F_BEAT;
158 const char *reason = st->reason ? st->reason : "";
159 size_t rn = strlen(reason);
160 if (rn > MESH_BEAT_REASON_MAX) rn = MESH_BEAT_REASON_MAX;
161 mpw_put(&o, &t, 1);
162 put64(&o, m->term);
163 mpw_put32(&o, m->leader ? MESH_BEAT_LEADER : 0);
164 mpw_put32(&o, st->has_leader ? st->leader : MESH_PEER_NONE);
165 put64(&o, m->commit.term);
166 put64(&o, m->commit.version);
167 put64(&o, m->acc.term);
168 put64(&o, m->acc.version);
169 put64(&o, st->applied);
170 mpw_put32(&o, st->status);
171 mpw_put32(&o, st->up);
172 mpw_put32(&o, st->group_epoch);
173 mpw_put32(&o, st->group_state);
174 mpw_put32(&o, (uint32_t)rn);
175 mpw_put(&o, reason, rn);
176 /* F_BEAT's tail: u32 count, then per move outcome its name, version, ok and gap. */
177 size_t k = st->nmoves < MESH_BEAT_MOVES_MAX ? st->nmoves : MESH_BEAT_MOVES_MAX;
178 mpw_put32(&o, (uint32_t)k);
179 for (size_t i = 0; i < k; i++) {
180 uint8_t name[MESH_NAME_BYTES];
181 mpw_name16(st->moves[i].name, name);
182 mpw_put(&o, name, sizeof name);
183 put64(&o, st->moves[i].version);
184 mpw_put32(&o, st->moves[i].ok);
185 mpw_put32(&o, st->moves[i].gap_us);
186 }
187 return o;
188}
189
190bool mpk_beat_decode(const uint8_t *h, const uint8_t *reason, size_t nreason, const uint8_t *moves, size_t nmoves, mfc_msg *msg, mpk_beat_status *st, char **err) {
191 uint32_t l = mpw_u32_at(h, 12);
192 *msg = (mfc_msg){.kind = MFC_BEAT, .term = mpw_u64_at(h, 0), .leader = (mpw_u32_at(h, 8) & MESH_BEAT_LEADER) != 0, .commit = {mpw_u64_at(h, 16), mpw_u64_at(h, 24)},
193 .acc = {mpw_u64_at(h, 32), mpw_u64_at(h, 40)}};
194 mpk_beat_status s = {.has_leader = l != MESH_PEER_NONE, .leader = l != MESH_PEER_NONE ? l : 0, .applied = mpw_u64_at(h, 48), .status = mpw_u32_at(h, 56),
195 .up = mpw_u32_at(h, 60), .group_epoch = mpw_u32_at(h, 64), .group_state = mpw_u32_at(h, 68)};
196 /* String::from_utf8_lossy: the bytes as they are (the fleet's reasons are ASCII). */
197 s.reason = mt_strndup((const char *)reason, nreason);
198 size_t k = nmoves / MESH_BEAT_MOVE_BYTES;
199 s.moves = mt_alloc((k ? k : 1) * sizeof *s.moves);
200 for (size_t i = 0; i < k; i++) {
201 const uint8_t *c = moves + i * MESH_BEAT_MOVE_BYTES;
202 char name[MESH_NAME_BYTES + 1];
203 if (!mpw_name_of(c, MESH_NAME_BYTES, name, err)) {
204 mpk_beat_status_free(&s);
205 return false;
206 }
207 s.moves[s.nmoves++] = (mst_outcome){mt_strdup(name), mpw_u64_at(c, MESH_NAME_BYTES), mpw_u32_at(c, MESH_NAME_BYTES + 8) != 0, mpw_u32_at(c, MESH_NAME_BYTES + 12)};
208 }
209 *st = s;
210 return true;
211}
212
213mfc_msg mpk_vote_req_decode(const uint8_t *b) {
214 return (mfc_msg){.kind = MFC_VOTE_REQ, .pre = mpw_u32_at(b, 0) != 0, .term = mpw_u64_at(b, 4), .acc = {mpw_u64_at(b, 12), mpw_u64_at(b, 20)}};
215}
216
217mfc_msg mpk_vote_decode(const uint8_t *b) { return (mfc_msg){.kind = MFC_VOTE, .pre = mpw_u32_at(b, 0) != 0, .term = mpw_u64_at(b, 4), .granted = mpw_u32_at(b, 12) != 0}; }
218
219mfc_msg mpk_ack_decode(uint8_t type, const uint8_t *b) {
220 return (mfc_msg){.kind = type == MESH_F_STATE_ACK ? MFC_STATE_ACK : MFC_COMMIT, .term = mpw_u64_at(b, 0), .ver = {mpw_u64_at(b, 8), mpw_u64_at(b, 16)}};
221}
222
223bool mpk_hello_rank(const mpk_hello *h, uint32_t k, uint32_t *rank) {
224 for (uint32_t r = 0; r < MESH_MAX_SYSTEMS; r++)
225 if (h->order[r] == k) {
226 if (rank) *rank = r;
227 return true;
228 }
229 return false;
230}
231
232uint32_t mpk_hello_n(const mpk_hello *h) {
233 uint32_t n = 0;
234 for (uint32_t m = h->mask; m; m &= m - 1) n++;
235 return n;
236}
237
238bool mpk_hello_eq(const mpk_hello *a, const mpk_hello *b) {
239 return !strcmp(a->group, b->group) && a->epoch == b->epoch && a->cycle == b->cycle && a->mask == b->mask && !memcmp(a->order, b->order, sizeof a->order) &&
240 a->hash == b->hash;
241}
242
243mpw_bytes mpk_hello_frame(uint32_t me, const mpk_hello *h) {
244 mpw_bytes o = {0};
245 uint8_t t = MESH_F_HELLO, name[MESH_NAME_BYTES];
246 mpw_put(&o, &t, 1);
247 mpw_put32(&o, me);
248 mpw_put32(&o, MESH_PEER_VERSION);
249 mpw_name16(h->group, name);
250 mpw_put(&o, name, sizeof name);
251 mpw_put32(&o, h->epoch);
252 mpw_put32(&o, h->cycle);
253 mpw_put32(&o, h->mask);
254 mpw_put(&o, h->order, MESH_MAX_SYSTEMS);
255 put64(&o, h->hash);
256 return o;
257}
258
259bool mpk_hello_decode(const uint8_t *b, uint32_t *system, uint32_t *version, mpk_hello *out, char **err) {
260 size_t at = 8 + MESH_NAME_BYTES;
261 mpk_hello h = {.epoch = mpw_u32_at(b, at), .cycle = mpw_u32_at(b, at + 4), .mask = mpw_u32_at(b, at + 8), .hash = mpw_u64_at(b, at + 12 + MESH_MAX_SYSTEMS)};
262 if (!mpw_name_of(b + 8, MESH_NAME_BYTES, h.group, err)) return false;
263 memcpy(h.order, b + at + 12, MESH_MAX_SYSTEMS);
264 *system = mpw_u32_at(b, 0);
265 *version = mpw_u32_at(b, 4);
266 *out = h;
267 return true;
268}