1/* A fleet on loopback: N relays over real TCP links with a fake daemon behind each (a status block,
2 * a mailbox, a reconciler that applies every committed document at once).
3 * - they elect one leader;
4 * - a change asked for on a follower goes to the leader, commits, reconciles everywhere and is
5 *   answered;
6 * - a member that stops is seen gone (fleet core, status block, `fleet` text);
7 * - the leader stopping elects another among the majority left, which commits changes;
8 * - a member that comes back catches up to the committed document;
9 * - without a majority a change is refused. */
10#include "mesh/check.h"
11#include "mesh/abi/host_fleet.h"
12#include "mesh/abi/host_members.h"
13#include "mesh/abi/host_sched.h"
14#include "mesh/abi/reconcile.h"
15#include "mesh/peer_relay.h"
16
17#include <arpa/inet.h>
18#include <netinet/in.h>
19#include <pthread.h>
20#include <stdio.h>
21#include <stdlib.h>
22#include <string.h>
23#include <sys/socket.h>
24#include <time.h>
25#include <unistd.h>
26
27#define N 3
28#define HOST_BYTES 0x4000
29#define MB_BYTES 0x30000
30
31typedef struct {
32    uint32_t k;
33    pthread_mutex_t mu;
34    uint8_t host[HOST_BYTES];
35    uint8_t *mb;
36    uint64_t t0;
37    uint64_t applied;
38    uint32_t cordon; /* the committed document's, as the reconciler saw it */
39    uint32_t ncommitted;
40    char *fleet, *groups, *sched, *recs;
41    bool failed;
42} fake;
43
44static uint64_t mono_ms(void) {
45    struct timespec ts;
46    clock_gettime(CLOCK_MONOTONIC, &ts);
47    return (uint64_t)ts.tv_sec * 1000u + (uint64_t)ts.tv_nsec / 1000000u;
48}
49
50static void sleep_ms(long ms) {
51    struct timespec ts = {ms / 1000, (ms % 1000) * 1000000L};
52    nanosleep(&ts, NULL);
53}
54
55static bool verbose;
56
57static void put(fake *f, size_t off, const void *b, size_t n) {
58    if (off + n > HOST_BYTES) return;
59    pthread_mutex_lock(&f->mu);
60    memcpy(f->host + off, b, n);
61    pthread_mutex_unlock(&f->mu);
62}
63static void host_u32(void *c, size_t off, uint32_t v) { put(c, off, &v, 4); }
64static void host_u64(void *c, size_t off, uint64_t v) { put(c, off, &v, 8); }
65static void host_str(void *c, size_t off, size_t cap, const char *s) {
66    char b[256] = {0};
67    if (cap > sizeof b) cap = sizeof b;
68    strncpy(b, s, cap - 1);
69    put(c, off, b, cap);
70}
71static void host_bytes(void *c, size_t off, const uint8_t *b, size_t n) { put(c, off, b, n); }
72static void host_read(void *c, size_t off, uint8_t *out, size_t n) {
73    fake *f = c;
74    memset(out, 0, n);
75    if (off + n > HOST_BYTES) return;
76    pthread_mutex_lock(&f->mu);
77    memcpy(out, f->host + off, n);
78    pthread_mutex_unlock(&f->mu);
79}
80static uint32_t get32(fake *f, size_t off) {
81    uint32_t v;
82    host_read(f, off, (uint8_t *)&v, 4);
83    return v;
84}
85static uint64_t mono_ms(void);
86static void sleep_ms(long ms);
87
88/* Waits up to ms for the status block's u32 at off to be v (masked). */
89static bool wait_u32(fake *f, size_t off, uint32_t mask, uint32_t v, uint32_t ms) {
90    uint64_t end = mono_ms() + ms;
91    while ((get32(f, off) & mask) != v && mono_ms() < end) sleep_ms(20);
92    return (get32(f, off) & mask) == v;
93}
94
95static uint64_t uptime_ms(void *c) { return mono_ms() - ((fake *)c)->t0 + 1; }
96static uint64_t mb_get64(void *c, size_t off) {
97    uint64_t v = 0;
98    if (off + 8 <= MB_BYTES) memcpy(&v, ((fake *)c)->mb + off, 8);
99    return v;
100}
101static uint32_t mb_get32(void *c, size_t off) {
102    uint32_t v = 0;
103    if (off + 4 <= MB_BYTES) memcpy(&v, ((fake *)c)->mb + off, 4);
104    return v;
105}
106static void mb_read(void *c, size_t off, uint8_t *out, size_t n) {
107    memset(out, 0, n);
108    if (off + n <= MB_BYTES) memcpy(out, ((fake *)c)->mb + off, n);
109}
110static void mb_write(void *c, size_t off, const uint8_t *b, size_t n) {
111    if (off + n <= MB_BYTES) memcpy(((fake *)c)->mb + off, b, n);
112}
113static void mb_publish64(void *c, size_t off, uint64_t v) { mb_write(c, off, (const uint8_t *)&v, 8); }
114static void lock_feed(void *c, bool on) { (void)c, (void)on; }
115static void round_stats(void *c, uint32_t k, uint64_t n, double rtt, double wait) { (void)c, (void)k, (void)n, (void)rtt, (void)wait; }
116
117/* The reconciler: applies a committed document at once. */
118static void committed(void *c, const mfd_doc *doc) {
119    fake *f = c;
120    pthread_mutex_lock(&f->mu);
121    f->applied = doc->version;
122    f->cordon = doc->cordon;
123    f->ncommitted++;
124    pthread_mutex_unlock(&f->mu);
125}
126static void recon(void *c, mpr_recon *out) {
127    fake *f = c;
128    pthread_mutex_lock(&f->mu);
129    *out = (mpr_recon){.applied = f->applied, .status = f->applied ? MESH_RS_SYNC : MESH_RS_NONE, .reason = mt_strdup(f->applied ? "ok" : "")};
130    pthread_mutex_unlock(&f->mu);
131}
132static bool catalog(void *c, const char *dir, mfe_image *out, char **err) {
133    (void)c, (void)out;
134    *err = mt_fmt("no %s in the catalog", dir);
135    return false;
136}
137static mfl_actual *inventory(void *c, size_t *n) {
138    (void)c;
139    *n = 0;
140    return NULL;
141}
142static bool image(void *c, const char *dir, const char *name, uint64_t hash, mpr_payload *out, char **err) {
143    (void)c, (void)dir, (void)name, (void)hash, (void)out;
144    *err = mt_strdup("no images here");
145    return false;
146}
147static bool stop_and_view(void *c, const char *name, uint32_t *abi, mpw_bytes *view, char **err) {
148    (void)c, (void)name, (void)abi, (void)view;
149    *err = mt_strdup("no views here");
150    return false;
151}
152static void driver_view(void *c, uint32_t up, bool hl, uint32_t l, const mfd_doc *d) { (void)c, (void)up, (void)hl, (void)l, (void)d; }
153static void place_frame(void *c, uint32_t from, mpw_bytes f) {
154    (void)c, (void)from;
155    mpw_bytes_free(&f);
156}
157static void other_frame(void *c, uint32_t from, uint8_t t, const uint8_t *b, size_t n) { (void)c, (void)from, (void)t, (void)b, (void)n; }
158static void jobs_ctl(void *c, const char *line, bool remote, uint32_t from, uint32_t id, mpr_reply reply) {
159    (void)c, (void)line, (void)remote, (void)from, (void)id;
160    if (reply.fn) reply.fn(reply.ctx, false, "no jobs here");
161}
162static void jobs_view(void *c, uint64_t t, bool hl, uint32_t l, uint32_t a, uint32_t k) { (void)c, (void)t, (void)hl, (void)l, (void)a, (void)k; }
163static void jobs_sched(void *c, const msm_member *ms, size_t n, const mfd_doc *d, const msm_perfs *p, const msw_weights *w) {
164    (void)c, (void)ms, (void)n, (void)d, (void)p, (void)w;
165}
166static void texts(void *c, const char *fleet, const char *groups) {
167    fake *f = c;
168    pthread_mutex_lock(&f->mu);
169    mt_set(&f->fleet, fleet);
170    mt_set(&f->groups, groups);
171    pthread_mutex_unlock(&f->mu);
172}
173static void sched_texts(void *c, const char *all, const msp_explains *ex, const char *dry, const char *recs) {
174    (void)ex, (void)dry;
175    fake *f = c;
176    pthread_mutex_lock(&f->mu);
177    mt_set(&f->sched, all);
178    mt_set(&f->recs, recs);
179    pthread_mutex_unlock(&f->mu);
180}
181static void peer_state(void *c, uint32_t s) { (void)c, (void)s; }
182static void net_count(void *c, uint32_t k, bool tx, uint8_t t, size_t n) { (void)c, (void)k, (void)tx, (void)t, (void)n; }
183static void logline(void *c, bool error, const char *line) {
184    if (verbose) fprintf(stderr, "[%u]%s %s\n", ((fake *)c)->k, error ? " E" : "", line);
185}
186static void fail_stop(void *c, const char *why) {
187    ((fake *)c)->failed = true;
188    fprintf(stderr, "[%u] fail-stop: %s\n", ((fake *)c)->k, why);
189}
190
191static mpr_host host_of(fake *f) {
192    return (mpr_host){f,           host_u32,   host_u64,     host_str,    host_bytes,    host_read,   uptime_ms, mb_get64,    mb_get32,   mb_read,
193                      mb_write,    mb_publish64, lock_feed,  round_stats, committed,     recon,       catalog,   inventory,   image,      stop_and_view,
194                      driver_view, place_frame, other_frame, other_frame, jobs_ctl,      jobs_view,   jobs_sched, texts,      sched_texts, peer_state,
195                      net_count,   logline,     fail_stop};
196}
197
198/* ---- the fleet */
199
200static fake fakes[N];
201static mpr_relay *relays[N];
202static char *conf;
203
204static uint16_t free_port(void) {
205    int fd = socket(AF_INET, SOCK_STREAM, 0);
206    struct sockaddr_in a = {.sin_family = AF_INET, .sin_addr.s_addr = htonl(INADDR_LOOPBACK)};
207    socklen_t l = sizeof a;
208    bind(fd, (struct sockaddr *)&a, sizeof a);
209    getsockname(fd, (struct sockaddr *)&a, &l);
210    close(fd);
211    return ntohs(a.sin_port);
212}
213
214static void start(uint32_t k) {
215    fake *f = &fakes[k];
216    if (!f->mb) {
217        pthread_mutex_init(&f->mu, NULL);
218        f->mb = calloc(1, MB_BYTES);
219    }
220    pthread_mutex_lock(&f->mu);
221    memset(f->host, 0, sizeof f->host);
222    f->applied = 0, f->cordon = 0, f->ncommitted = 0;
223    f->k = k;
224    f->t0 = mono_ms();
225    pthread_mutex_unlock(&f->mu);
226    mcc_topology t;
227    char *err = NULL;
228    if (!mcc_topology_parse(conf, k, &t, &err)) {
229        fprintf(stderr, "%s\n", err);
230        abort();
231    }
232    mpk_caps caps = {.cc = 0, .backend = 0};
233    snprintf(caps.gpu, sizeof caps.gpu, "none");
234    snprintf(caps.arch, sizeof caps.arch, "x86_64");
235    snprintf(caps.host, sizeof caps.host, "loop-%u", k);
236    mpr_host h = host_of(f);
237    relays[k] = mpr_start(&t, &caps, NULL, &h, &err);
238    if (!relays[k]) {
239        fprintf(stderr, "start %u: %s\n", k, err);
240        abort();
241    }
242    mcc_topology_free(&t);
243}
244
245static void stop(uint32_t k) {
246    mpr_stop(relays[k]);
247    relays[k] = NULL;
248}
249
250/* The leader every running relay agrees on (and that leads), or -1. */
251static int agreed_leader(void) {
252    int l = -1;
253    for (uint32_t k = 0; k < N; k++) {
254        if (!relays[k]) continue;
255        uint32_t x;
256        if (!mpr_leader(relays[k], &x)) return -1;
257        if (l >= 0 && (uint32_t)l != x) return -1;
258        l = (int)x;
259    }
260    if (l < 0 || !relays[l] || mpr_role(relays[l]) != MESH_ROLE_LEADER) return -1;
261    return l;
262}
263
264static int wait_leader(uint32_t ms) {
265    uint64_t end = mono_ms() + ms;
266    int l;
267    while ((l = agreed_leader()) < 0 && mono_ms() < end) sleep_ms(20);
268    return l;
269}
270
271static bool wait_alive(uint32_t k, uint32_t mask, uint32_t ms) {
272    uint64_t end = mono_ms() + ms;
273    while (mono_ms() < end) {
274        if (mpr_alive(relays[k]) == mask) return true;
275        sleep_ms(20);
276    }
277    return false;
278}
279
280/* ---- a change and its answer */
281
282typedef struct {
283    pthread_mutex_t mu;
284    pthread_cond_t cv;
285    bool done, ok;
286    char *text;
287} answer;
288
289static void on_answer(void *ctx, bool ok, const char *text) {
290    answer *a = ctx;
291    pthread_mutex_lock(&a->mu);
292    a->done = true, a->ok = ok;
293    a->text = mt_strdup(text);
294    pthread_cond_signal(&a->cv);
295    pthread_mutex_unlock(&a->mu);
296}
297
298/* Asks member k for the change; true when answered ok within ms (*text: the answer). */
299static bool change(uint32_t k, const char *line, uint32_t ms, char **text) {
300    answer a = {.mu = PTHREAD_MUTEX_INITIALIZER, .cv = PTHREAD_COND_INITIALIZER};
301    mpr_change(relays[k], line, (mpr_reply){on_answer, &a});
302    uint64_t end = mono_ms() + ms;
303    pthread_mutex_lock(&a.mu);
304    while (!a.done && mono_ms() < end) {
305        pthread_mutex_unlock(&a.mu);
306        sleep_ms(10);
307        pthread_mutex_lock(&a.mu);
308    }
309    bool done = a.done, ok = a.ok;
310    pthread_mutex_unlock(&a.mu);
311    if (!done) {
312        fprintf(stderr, "change %s on %u: no answer in %u ms\n", line, k, ms);
313        abort();
314    }
315    if (verbose) fprintf(stderr, "change %s on %u: %s %s\n", line, k, ok ? "ok" : "refused", a.text);
316    if (text) *text = a.text;
317    else free(a.text);
318    return ok;
319}
320
321static bool wait_applied(uint32_t k, uint64_t version, uint32_t ms) {
322    uint64_t end = mono_ms() + ms;
323    while (mono_ms() < end) {
324        pthread_mutex_lock(&fakes[k].mu);
325        uint64_t a = fakes[k].applied;
326        pthread_mutex_unlock(&fakes[k].mu);
327        if (a >= version) return true;
328        sleep_ms(20);
329    }
330    return false;
331}
332
333static char *fleet_text(uint32_t k) {
334    pthread_mutex_lock(&fakes[k].mu);
335    char *s = mt_strdup(fakes[k].fleet ? fakes[k].fleet : "");
336    pthread_mutex_unlock(&fakes[k].mu);
337    return s;
338}
339
340static bool wait_fleet_text(uint32_t k, const char *needle, uint32_t ms) {
341    uint64_t end = mono_ms() + ms;
342    while (mono_ms() < end) {
343        char *s = fleet_text(k);
344        bool yes = strstr(s, needle) != NULL;
345        free(s);
346        if (yes) return true;
347        sleep_ms(20);
348    }
349    char *s = fleet_text(k);
350    fprintf(stderr, "member %u's fleet text has no \"%s\":\n%s", k, needle, s);
351    free(s);
352    return false;
353}
354
355static bool wait_recs(uint32_t k, const char *needle, uint32_t ms) {
356    uint64_t end = mono_ms() + ms;
357    while (mono_ms() < end) {
358        pthread_mutex_lock(&fakes[k].mu);
359        bool yes = fakes[k].recs && strstr(fakes[k].recs, needle);
360        pthread_mutex_unlock(&fakes[k].mu);
361        if (yes) return true;
362        sleep_ms(20);
363    }
364    return false;
365}
366
367int main(int argc, char **argv) {
368    verbose = argc > 1 && strcmp(argv[1], "-v") == 0;
369    mt_buf b = {0};
370    mt_printf(&b, "systems = %d\nping_ms = 50\ndead_ms = 600\n", N);
371    for (int k = 0; k < N; k++) mt_printf(&b, "%d.peer = 127.0.0.1:%u\n", k, free_port());
372    conf = mt_take(&b);
373    for (uint32_t k = 0; k < N; k++) start(k);
374
375    /* One leader, every member up. */
376    int l = wait_leader(10000);
377    CHECK(l >= 0);
378    for (uint32_t k = 0; k < N; k++) CHECK(wait_alive(k, (1u << N) - 1, 5000));
379    /* The leader's first document commits and reconciles everywhere. */
380    for (uint32_t k = 0; k < N; k++) CHECK(wait_applied(k, 1, 5000));
381    uint32_t f1 = (uint32_t)(l + 1) % N, f2 = (uint32_t)(l + 2) % N;
382    char lead_word[32];
383    snprintf(lead_word, sizeof lead_word, "loop-%d leader", l);
384    CHECK(wait_fleet_text(f1, lead_word, 3000));
385    CHECK(wait_u32(&fakes[f1], MESH_S_LEADER, ~0u, (uint32_t)l, 3000));
386    CHECK(wait_u32(&fakes[l], MESH_S_ROLE, ~0u, MESH_ROLE_LEADER, 3000));
387    CHECK(wait_u32(&fakes[f1], MESH_S_UP_MASK, ~0u, (1u << N) - 1, 3000));
388    CHECK(wait_u32(&fakes[f1], MESH_S_FLEET_FLAGS, MESH_FF_QUORUM, MESH_FF_QUORUM, 3000));
389
390    /* A change asked for on a follower: forwarded, committed, reconciled, answered. */
391    char *text = NULL;
392    CHECK(change(f1, "sched cordon 2", 15000, &text));
393    free(text);
394    uint64_t v = mpr_version(relays[l]);
395    for (uint32_t k = 0; k < N; k++) {
396        CHECK(wait_applied(k, v, 5000));
397        pthread_mutex_lock(&fakes[k].mu);
398        CHECK(fakes[k].cordon == 1u << 2);
399        pthread_mutex_unlock(&fakes[k].mu);
400    }
401    /* sched_publish (every second) puts the cordon in the status block and the view. */
402    uint64_t end = mono_ms() + 3000;
403    while (get32(&fakes[f2], MESH_S_CORDON) != 1u << 2 && mono_ms() < end) sleep_ms(20);
404    CHECK(get32(&fakes[f2], MESH_S_CORDON) == 1u << 2);
405    CHECK(wait_recs(f2, "cordon=2", 3000));
406    /* A change that is not one: refused by the leader, the answer says why. */
407    CHECK(!change(f2, "sched frobnicate", 15000, &text));
408    CHECK(text && strstr(text, "not a change"));
409    free(text);
410
411    /* A follower stops: the others see it gone. */
412    stop(f2);
413    uint32_t two = (1u << l) | (1u << f1);
414    CHECK(wait_alive((uint32_t)l, two, 5000));
415    CHECK(wait_alive(f1, two, 5000));
416    char gone[32];
417    snprintf(gone, sizeof gone, "member %u down", f2);
418    CHECK(wait_fleet_text(f1, gone, 3000));
419    CHECK(wait_u32(&fakes[l], mesh_member(f2) + MESH_M_STATE, ~0u, MESH_MS_ABSENT, 3000));
420    CHECK(wait_u32(&fakes[l], MESH_S_UP_MASK, ~0u, two, 3000));
421    /* Two of three are a majority: changes still commit. */
422    CHECK(change((uint32_t)l, "sched uncordon 2", 15000, NULL));
423
424    /* It comes back and catches up. */
425    start(f2);
426    CHECK(wait_alive((uint32_t)l, (1u << N) - 1, 5000));
427    v = mpr_version(relays[l]);
428    CHECK(wait_applied(f2, v, 5000));
429    pthread_mutex_lock(&fakes[f2].mu);
430    CHECK(fakes[f2].cordon == 0);
431    pthread_mutex_unlock(&fakes[f2].mu);
432
433    /* The leader stops: the two left elect one of them and go on committing. */
434    stop((uint32_t)l);
435    int l2 = wait_leader(10000);
436    CHECK(l2 >= 0 && l2 != l);
437    CHECK(change(l2 == (int)f1 ? f2 : f1, "sched pause", 15000, NULL));
438    v = mpr_version(relays[l2]);
439    CHECK(wait_applied(f1, v, 5000) && wait_applied(f2, v, 5000));
440    CHECK(wait_u32(&fakes[f1], MESH_S_SCHED_FLAGS, MESH_SF_PAUSED, MESH_SF_PAUSED, 3000));
441
442    /* One of three left: no majority, no leader, a change is refused. */
443    uint32_t last = l2 == (int)f1 ? f2 : f1;
444    stop((uint32_t)l2);
445    end = mono_ms() + 5000;
446    uint32_t x;
447    while (mpr_leader(relays[last], &x) && mono_ms() < end) sleep_ms(20);
448    CHECK(!mpr_leader(relays[last], &x));
449    CHECK(!change(last, "sched resume", 15000, &text));
450    CHECK(text && strstr(text, "unchanged"));
451    free(text);
452    CHECK(wait_u32(&fakes[last], MESH_S_FLEET_FLAGS, MESH_FF_QUORUM, 0, 3000));
453    CHECK(wait_u32(&fakes[last], MESH_S_LEADER, ~0u, MESH_PEER_NONE, 3000));
454    stop(last);
455
456    for (uint32_t k = 0; k < N; k++) {
457        CHECK(!fakes[k].failed);
458        free(fakes[k].mb), free(fakes[k].fleet), free(fakes[k].groups), free(fakes[k].sched), free(fakes[k].recs);
459    }
460    free(conf);
461    return mesh_check_done();
462}