1/* fleet.rs document_round_trips (the document part), group rank/order, can_run, decode errors. */
2#include "mesh/check.h"
3#include "mesh/fleet_doc.h"
4
5#include <stdlib.h>
6
7static mfd_doc doc(void) {
8    mfd_doc d = {.version = 7, .term = 3};
9    mesh_needs n = msn_default();
10    mfd_prog h = mfd_prog_new("hello", 0xabc, true, &n);
11    h.reset = 2;
12    mt_push(&h.backends, "cuda");
13    mt_push(&h.backends, "wasm");
14    mfd_prog_put(&d, "hello", h);
15    mfd_prog t = mfd_prog_new("train-b", 0x123, true, &n);
16    t.lockstep = true;
17    mt_push(&t.backends, "cuda");
18    t.has_prev = true;
19    t.prev_dir = mt_strdup("train");
20    t.prev_hash = 0x99;
21    mfd_prog_put(&d, "train", t);
22    mesh_needs_free(&n);
23    mfd_group g = {.name = mt_strdup("train"), .program = mt_strdup("train"), .epoch = 4, .cycle = 2, .op = {MFD_OP_TASK, 3, 1000593162},
24                   .members = ml_of((uint32_t[]){0, 2, 3}, 3), .state = MFD_RUNNING, .on_loss = MFD_LOSS_REFORM, .auto_ms = 5000};
25    mfd_group_push(&d, g);
26    mt_strs w = {0};
27    mt_push(&w, "split");
28    mt_push(&w, "auto");
29    mfd_place_put(&d, "doom", w);
30    return d;
31}
32
33static void round_trip(const mfd_doc *d) {
34    char *e = mfd_encode(d);
35    mfd_doc back;
36    char *err = NULL;
37    CHECK(mfd_decode(e, &back, &err));
38    CHECK(mfd_doc_eq(&back, d));
39    mfd_doc_free(&back);
40    free(e);
41}
42
43static void bad(const char *text, const char *want) {
44    mfd_doc d;
45    char *err = NULL;
46    CHECK(!mfd_decode(text, &d, &err));
47    if (want) CHECK_S(err, want);
48    free(err);
49}
50
51int main(void) {
52    mfd_doc d = doc();
53    round_trip(&d);
54    char *e = mfd_encode(&d);
55    CHECK_S(e, "doc version 7 term 3\n"
56               "program hello dir hello hash 0000000000000abc run 1 reset 2 lockstep 0 backends cuda,wasm prev - 0 at * from - pin * needs "
57               "b=cuda;ar=-;mm=0;pf=-;av=-;r=all;sp=spread;co=-;fd=-;g=-;gs=0;gm=-;pr=0;cpu=0;mem=0;mg=state;ls=0;vb=0;w=1;sh=0 why \n"
58               "program train dir train-b hash 0000000000000123 run 1 reset 0 lockstep 1 backends cuda prev train 0000000000000099 at * from - pin * needs "
59               "b=cuda;ar=-;mm=0;pf=-;av=-;r=all;sp=spread;co=-;fd=-;g=-;gs=0;gm=-;pr=0;cpu=0;mem=0;mg=state;ls=0;vb=0;w=1;sh=0 why \n"
60               "group train program train epoch 4 cycle 2 op task:3:1000593162 members 0,2,3 state running on_loss reform auto_ms 5000\n"
61               "place doom split,auto\n"
62               "sched paused 0 autoreform 0 cordon -\n");
63    free(e);
64    mfd_doc empty = {0};
65    round_trip(&empty);
66    /* The scheduler's fields: a placement, a migration, a pin, the needs, the why; the sched flags. */
67    mfd_doc x = mfd_doc_clone(&d);
68    mfd_prog *h = mfd_prog_get(&x, "hello");
69    ml_opt_free(&h->at);
70    h->at = ml_some((uint32_t[]){1, 3}, 2);
71    h->has_from = h->has_to = true;
72    h->from = 2;
73    h->to = 3;
74    h->since = 41;
75    h->pin = ml_some((uint32_t[]){3}, 1);
76    h->needs.replicas = (mesh_replicas){false, 2};
77    msn_terms_push(&h->needs.prefer, &h->needs.nprefer, msn_term_new(MESH_TERM_HOST, "dgx-00", 0));
78    mt_set(&h->why, "replicas 2: system 1 0.812, system 3 0.700; next system 0 0.6");
79    mfd_prog *t = mfd_prog_get(&x, "train");
80    t->at = ml_some(NULL, 0);
81    x.sched_paused = true;
82    x.autoreform = true;
83    x.cordon = 9;
84    round_trip(&x);
85    /* A migration's source alone (no target). */
86    h->has_to = false;
87    h->since = 0;
88    round_trip(&x);
89    mfd_doc_free(&x);
90    bad("program x", "fleet document line 1: \"program x\"");
91    bad("", "fleet document: no `doc version .. term ..` line");
92    bad("doc version 1 term 1\ngroup g program p epoch 1 cycle 0 op bogus members - state running on_loss pause auto_ms 0", "bad op \"bogus\"");
93    bad("doc version 1 term 1\ngroup g program p epoch 1 cycle 0 op none members 9 state running on_loss pause auto_ms 0", "member 9 >= 8");
94    bad("doc version 1 term 1\ngroup g program p epoch 1 cycle 0 op none members - state sleeping on_loss pause auto_ms 0", NULL);
95    bad("doc version x term 1", "fleet document line 1: \"doc version x term 1\"");
96    /* Group rank and order. */
97    uint32_t r;
98    CHECK(mfd_group_rank(&d.groups[0], 2, &r) && r == 1 && !mfd_group_rank(&d.groups[0], 1, &r));
99    uint8_t o[MESH_MAX_SYSTEMS];
100    mfd_group_order(&d.groups[0], o);
101    CHECK(o[0] == 0 && o[1] == 2 && o[2] == 3 && o[3] == 0xff && o[7] == 0xff);
102    CHECK_U(mfd_group_mask(&d.groups[0]), 0xd);
103    /* here: placed there, or the source of a migration. */
104    mfd_prog *hp = mfd_prog_get(&d, "hello");
105    CHECK(mfd_prog_here(hp, 5));
106    hp->at = ml_some((uint32_t[]){1}, 1);
107    CHECK(mfd_prog_here(hp, 1) && !mfd_prog_here(hp, 2));
108    hp->has_from = true;
109    hp->from = 2;
110    CHECK(mfd_prog_here(hp, 2));
111    /* can_run: the member's GPU, or wasm / native anywhere. */
112    mt_strs b = {0};
113    mt_push(&b, "metal");
114    CHECK(mfd_can_run(&b, 1) && !mfd_can_run(&b, 0) && !mfd_can_run(&b, 99));
115    mt_push(&b, "native");
116    CHECK(mfd_can_run(&b, 0));
117    mt_strs_free(&b);
118    /* Maps: replace, remove. */
119    CHECK(mfd_prog_remove(&d, "hello") && !mfd_prog_remove(&d, "hello") && d.nprogs == 1);
120    CHECK(mfd_place_remove(&d, "doom") && mfd_place_get(&d, "doom") == NULL);
121    CHECK(mfd_group_of(&d, "train") && mfd_group_named(&d, "train") && !mfd_group_of(&d, "x"));
122    mfd_group_remove_at(&d, 0);
123    CHECK_U(d.ngroups, 0);
124    CHECK(mfd_id_cmp(3, 9, 4, 1) < 0 && mfd_id_cmp(4, 2, 4, 1) > 0 && mfd_id_cmp(1, 1, 1, 1) == 0);
125    mfd_doc_free(&d);
126    return mesh_check_done();
127}