==> tests/sim.h <==1/* fleet.rs's test network, ported: a deterministic network of n cores -- messages delayed, dropped,
2 * partitions, crashes and restarts, disks that fill up -- that asserts at most one leader per term,
3 * one document per committed version and per accepted (term, version), no committed change lost,
4 * and no member's committed document going back. */
5#ifndef SIM_H
6#define SIM_H
7
8#include "mesh/check.h"
9#include "mesh/fleet_core.h"
10#include "mesh/mt.h"
11
12#include <stdio.h>
13#include <stdlib.h>
14#include <string.h>
15
16typedef struct {
17 uint64_t at;
18 uint32_t from, to;
19 mfc_msg msg;
20} wire_t;
21
22typedef struct {
23 uint32_t n;
24 mfc_core *cores[8];
25 mfp_persist disk[8];
26 uint64_t now;
27 wire_t *wire;
28 size_t nwire, capwire;
29 uint32_t side[8];
30 uint64_t rng, loss;
31 uint64_t leader_term[4096];
32 uint32_t leader_of[4096];
33 size_t nleaders;
34 mfd_doc *committed; /* by version, in insertion order */
35 size_t ncommitted;
36 mfd_doc *accepted;
37 size_t naccepted;
38 mfc_id high[8];
39 bool bad_disk[8];
40 uint32_t fail_stops;
41 int failed; /* an invariant broke */
42} sim_t;
43
44static uint64_t sim_rand(sim_t *s, uint64_t n) {
45 s->rng ^= s->rng << 13;
46 s->rng ^= s->rng >> 7;
47 s->rng ^= s->rng << 17;
48 return s->rng % (n ? n : 1);
49}
50
51static mfc_core *sim_boot(sim_t *s, uint32_t k) {
52 uint64_t seed = sim_rand(s, 1u << 30) + 1;
53 return mfc_new((mfc_config){k, s->n, 3000, seed}, mfp_clone(&s->disk[k]), s->now);
54}
55
56static bool sim_connected(sim_t *s, uint32_t a, uint32_t b) { return s->cores[a] && s->cores[b] && s->side[a] == s->side[b]; }
57
58static void sim_crash(sim_t *s, uint32_t k);
59
60typedef struct {
61 sim_t *s;
62 uint32_t k;
63} persist_ctx;
64
65static bool sim_persist(void *ctx, char **err) {
66 persist_ctx *p = ctx;
67 if (p->s->bad_disk[p->k]) {
68 *err = mt_strdup("injected: no space left on device");
69 return false;
70 }
71 mfp_free(&p->s->disk[p->k]);
72 p->s->disk[p->k] = mfp_clone(mfc_persist(p->s->cores[p->k]));
73 return true;
74}
75
76static mfd_doc *find_version(sim_t *s, uint64_t v) {
77 for (size_t i = 0; i < s->ncommitted; i++)
78 if (s->committed[i].version == v) return &s->committed[i];
79 return NULL;
80}
81
82static void sim_committed(sim_t *s, const mfd_doc *d) {
83 mfd_doc *o = find_version(s, d->version);
84 if (o) {
85 if (!mfd_doc_eq(o, d)) {
86 fprintf(stderr, "two different documents committed as version %llu\n", (unsigned long long)d->version);
87 s->failed++;
88 }
89 return;
90 }
91 /* No committed version is lost: every change (a tag) of a lower committed version is in this one,
92 * and this one's are in every higher one. */
93 for (size_t i = 0; i < s->ncommitted; i++) {
94 const mfd_doc *x = &s->committed[i];
95 const mfd_doc *lo = x->version < d->version ? x : d, *hi = x->version < d->version ? d : x;
96 for (size_t t = 0; t < lo->nplaces; t++)
97 if (!mfd_place_get(hi, lo->place_name[t])) {
98 fprintf(stderr, "committed version %llu lost %s of committed version %llu\n", (unsigned long long)hi->version, lo->place_name[t],
99 (unsigned long long)lo->version);
100 s->failed++;
101 }
102 }
103 s->committed = mt_realloc(s->committed, (s->ncommitted + 1) * sizeof *s->committed);
104 s->committed[s->ncommitted++] = mfd_doc_clone(d);
105}
106
107static void sim_apply(sim_t *s, uint32_t k, mfc_outs out) {
108 persist_ctx pc = {s, k};
109 char *err = NULL;
110 if (!mfc_durable(&out, sim_persist, &pc, &err)) {
111 free(err);
112 mfc_outs_free(&out);
113 s->fail_stops++;
114 sim_crash(s, k);
115 return;
116 }
117 for (size_t i = 0; i < out.n; i++) {
118 mfc_out *o = &out.v[i];
119 if (o->kind == MFC_OUT_SEND) {
120 if (sim_rand(s, 100) < s->loss || !sim_connected(s, k, o->to)) continue;
121 uint64_t at = s->now + 1 + sim_rand(s, 40);
122 if (s->nwire == s->capwire) {
123 s->capwire = s->capwire ? s->capwire * 2 : 64;
124 s->wire = mt_realloc(s->wire, s->capwire * sizeof *s->wire);
125 }
126 s->wire[s->nwire++] = (wire_t){at, k, o->to, o->msg};
127 o->msg = (mfc_msg){0}; /* moved */
128 } else if (o->kind == MFC_OUT_COMMITTED) {
129 sim_committed(s, mfc_committed(s->cores[k]));
130 } else if (o->kind == MFC_OUT_LEADER && o->has_leader && o->leader == k) {
131 uint64_t t = mfc_term(s->cores[k]);
132 for (size_t j = 0; j < s->nleaders; j++)
133 if (s->leader_term[j] == t && s->leader_of[j] != k) {
134 fprintf(stderr, "two leaders in term %llu: %u and %u\n", (unsigned long long)t, s->leader_of[j], k);
135 s->failed++;
136 }
137 s->leader_term[s->nleaders] = t;
138 s->leader_of[s->nleaders++] = k;
139 }
140 }
141 mfc_outs_free(&out);
142 if (s->cores[k]) {
143 /* A member's committed document never goes back (nor across its restarts). */
144 mfc_id id = mfc_doc_id(mfc_committed(s->cores[k]));
145 if (mfc_id_cmp(id, s->high[k]) < 0) {
146 fprintf(stderr, "member %u: committed went back\n", k);
147 s->failed++;
148 }
149 s->high[k] = id;
150 const mfd_doc *a = mfc_accepted(s->cores[k]);
151 if (a->version > 0) {
152 bool seen = false;
153 for (size_t i = 0; i < s->naccepted; i++)
154 if (s->accepted[i].term == a->term && s->accepted[i].version == a->version) {
155 seen = true;
156 if (!mfd_doc_eq(&s->accepted[i], a)) {
157 fprintf(stderr, "two documents accepted as (%llu, %llu)\n", (unsigned long long)a->term, (unsigned long long)a->version);
158 s->failed++;
159 }
160 }
161 if (!seen) {
162 s->accepted = mt_realloc(s->accepted, (s->naccepted + 1) * sizeof *s->accepted);
163 s->accepted[s->naccepted++] = mfd_doc_clone(a);
164 }
165 }
166 }
167}
168
169static void sim_relink(sim_t *s) {
170 for (uint32_t a = 0; a < s->n; a++)
171 for (uint32_t b = 0; b < s->n; b++) {
172 if (a == b || !s->cores[a]) continue;
173 bool up = sim_connected(s, a, b);
174 bool was = mfc_linked(s->cores[a]) >> b & 1;
175 if (up != was) sim_apply(s, a, mfc_link(s->cores[a], b, up, s->now));
176 }
177}
178
179static void sim_crash(sim_t *s, uint32_t k) {
180 if (!s->cores[k]) return;
181 mfc_free(s->cores[k]);
182 s->cores[k] = NULL;
183 sim_relink(s);
184}
185
186static void sim_restart(sim_t *s, uint32_t k) {
187 s->bad_disk[k] = false;
188 s->cores[k] = sim_boot(s, k);
189 sim_relink(s);
190}
191
192static void sim_init(sim_t *s, uint32_t n, uint64_t seed, uint64_t loss) {
193 memset(s, 0, sizeof *s);
194 s->n = n;
195 s->now = 1;
196 s->rng = seed | 1;
197 s->loss = loss;
198 for (uint32_t k = 0; k < n; k++) s->cores[k] = sim_boot(s, k);
199 sim_relink(s);
200}
201
202static void sim_free(sim_t *s) {
203 for (uint32_t k = 0; k < s->n; k++) {
204 mfc_free(s->cores[k]);
205 mfp_free(&s->disk[k]);
206 }
207 for (size_t i = 0; i < s->nwire; i++) mfc_msg_free(&s->wire[i].msg);
208 free(s->wire);
209 for (size_t i = 0; i < s->ncommitted; i++) mfd_doc_free(&s->committed[i]);
210 free(s->committed);
211 for (size_t i = 0; i < s->naccepted; i++) mfd_doc_free(&s->accepted[i]);
212 free(s->accepted);
213}
214
215/* peer.rs `lead`: a leader whose term has no committed document proposes its first one. */
216static void sim_lead(sim_t *s, uint32_t k) {
217 mfc_core *c = s->cores[k];
218 if (!c || mfc_role_of(c) != MFC_LEADER || mfc_busy(c) || mfc_term_committed(c)) return;
219 mfd_doc d = {0};
220 mfc_first_doc(c, &d);
221 uint64_t v;
222 mfc_outs out;
223 char *err = NULL;
224 if (mfc_propose(c, d, s->now, &v, &out, &err)) sim_apply(s, k, out);
225 else free(err);
226}
227
228static void sim_step(sim_t *s, uint64_t ms) {
229 uint64_t end = s->now + ms;
230 while (s->now < end) {
231 s->now++;
232 /* Deliver what is due, in order (links that broke meanwhile lose it). */
233 size_t nd = 0;
234 wire_t *due = NULL;
235 size_t j = 0;
236 for (size_t i = 0; i < s->nwire; i++) {
237 if (s->wire[i].at <= s->now) {
238 due = mt_realloc(due, (nd + 1) * sizeof *due);
239 due[nd++] = s->wire[i];
240 } else s->wire[j++] = s->wire[i];
241 }
242 s->nwire = j;
243 for (size_t i = 0; i < nd; i++) {
244 if (!sim_connected(s, due[i].from, due[i].to)) {
245 mfc_msg_free(&due[i].msg);
246 continue;
247 }
248 sim_apply(s, due[i].to, mfc_recv(s->cores[due[i].to], due[i].from, due[i].msg, s->now));
249 }
250 free(due);
251 if (s->now % 25 == 0) {
252 for (uint32_t k = 0; k < s->n; k++) {
253 mfc_core *c = s->cores[k];
254 if (!c) continue;
255 mfc_outs out = mfc_tick(c, s->now);
256 if (s->now % 250 == 0) {
257 mfc_msg b = mfc_beat(c);
258 for (uint32_t t = 0; t < s->n; t++)
259 if (t != k) {
260 if (out.n == out.cap) {
261 out.cap = out.cap ? out.cap * 2 : 8;
262 out.v = mt_realloc(out.v, out.cap * sizeof *out.v);
263 }
264 out.v[out.n++] = (mfc_out){.kind = MFC_OUT_SEND, .to = t, .msg = b};
265 }
266 }
267 sim_apply(s, k, out);
268 sim_lead(s, k);
269 }
270 }
271 }
272}
273
274/* The single leader, or -1. */
275static int sim_leader(sim_t *s) {
276 int l = -1, count = 0;
277 for (uint32_t k = 0; k < s->n; k++)
278 if (s->cores[k] && mfc_role_of(s->cores[k]) == MFC_LEADER) l = (int)k, count++;
279 return count == 1 ? l : -1;
280}
281
282/* A change built on the committed document, once this term committed one: version, or 0 with *err. */
283static uint64_t sim_propose(sim_t *s, uint32_t k, const char *tag, char **err) {
284 mfc_core *c = s->cores[k];
285 if (mfc_role_of(c) == MFC_LEADER && !mfc_term_committed(c)) {
286 *err = mt_strdup("the term's first document is not committed yet");
287 return 0;
288 }
289 mfd_doc d = mfd_doc_clone(mfc_committed(c));
290 mt_strs w = {0};
291 mt_push(&w, "local");
292 mfd_place_put(&d, tag, w);
293 uint64_t v;
294 mfc_outs out;
295 if (!mfc_propose(c, d, s->now, &v, &out, err)) return 0;
296 sim_apply(s, k, out);
297 return v;
298}
299
300#endif