1/* sched.rs: Replicas, Migrate, Term, Needs. */
2#include "mesh/sched_needs.h"
3#include "mesh/abi/backend.h"
4#include "mesh/abi/base.h"
5#include "mesh/parse.h"
6#include "mesh/rust_fmt.h"
7
8#include <math.h>
9#include <stdlib.h>
10#include <string.h>
11
12mesh_term msn_term_new(mesh_term_kind kind, const char *s, uint32_t system) { return (mesh_term){kind, s ? mt_strdup(s) : NULL, system}; }
13
14bool msn_term_matches(const mesh_term *t, const msm_member *m) {
15    switch (t->kind) {
16    case MESH_TERM_BACKEND: return strcmp(m->gpu, t->s) == 0;
17    case MESH_TERM_HOST: {
18        if (strcmp(m->host, t->s) == 0) return true;
19        const char *dot = strchr(m->host, '.');
20        size_t n = dot ? (size_t)(dot - m->host) : strlen(m->host);
21        return strlen(t->s) == n && memcmp(m->host, t->s, n) == 0;
22    }
23    case MESH_TERM_SYSTEM: return m->id == t->system;
24    default: return strcmp(m->arch, t->s) == 0;
25    }
26}
27
28void msn_terms_push(mesh_term **v, size_t *n, mesh_term t) {
29    *v = mt_realloc(*v, (*n + 1) * sizeof **v);
30    (*v)[(*n)++] = t;
31}
32
33void msn_strs_push(char ***v, size_t *n, const char *s) {
34    *v = mt_realloc(*v, (*n + 1) * sizeof **v);
35    (*v)[(*n)++] = mt_strdup(s);
36}
37
38char *msn_terms_text(const mesh_term *v, size_t n) {
39    if (!n) return mt_strdup("-");
40    mt_buf b = {0};
41    for (size_t i = 0; i < n; i++) {
42        char *w = mesh_term_word(&v[i]);
43        if (i) mt_putc(&b, ',');
44        mt_puts(&b, w);
45        free(w);
46    }
47    return mt_take(&b);
48}
49
50mesh_needs msn_default(void) {
51    mesh_needs n;
52    mesh_needs_default(&n);
53    return n;
54}
55
56mesh_needs msn_clone(const mesh_needs *n) {
57    mesh_needs c;
58    mesh_needs_copy(&c, n);
59    return c;
60}
61
62char *msn_strs_join(char *const *v, size_t n, const char *sep) {
63    mt_buf b = {0};
64    for (size_t i = 0; i < n; i++) {
65        if (i) mt_puts(&b, sep);
66        mt_puts(&b, v[i]);
67    }
68    return n ? mt_take(&b) : mt_strdup("");
69}
70
71bool msn_strs_has(char *const *v, size_t n, const char *s) {
72    for (size_t i = 0; i < n; i++)
73        if (strcmp(v[i], s) == 0) return true;
74    return false;
75}
76
77static char *join(char *const *v, size_t n, const char *empty) { return n ? msn_strs_join(v, n, ",") : mt_strdup(empty); }
78
79char *msn_replicas_word(mesh_replicas r) { return r.all ? mt_strdup("all") : mt_fmt("%u", r.n); }
80
81char *msn_encode(const mesh_needs *n) {
82    char *b = join(n->backends, n->nbackends, "-"), *pf = msn_terms_text(n->prefer, n->nprefer), *av = msn_terms_text(n->avoid, n->navoid), *r = msn_replicas_word(n->replicas),
83         *fd = join(n->feeds, n->nfeeds, "-"), *gm = msn_terms_text(n->group_members, n->ngroup_members);
84    char cpu[RF_FLOAT_MAX];
85    char *s = mt_fmt("b=%s;ar=%s;mm=%u;pf=%s;av=%s;r=%s;sp=%s;co=%s;fd=%s;g=%s;gs=%u;gm=%s;pr=%d;cpu=%s;mem=%u;mg=%s;ls=%u;vb=%u;w=%u;sh=%u", b,
86                     n->arch ? n->arch : "-", n->min_mem_mb, pf, av, r, n->pack ? "pack" : "spread", n->colocate_with ? n->colocate_with : "-", fd,
87                     n->group ? n->group : "-", n->group_size, gm, n->priority, rf_f32(cpu, n->cpu), n->mem_mb, mesh_migrate_word(n->migrate),
88                     (unsigned)n->lockstep, n->view_bytes, n->warps, n->shared);
89    free(b), free(pf), free(av), free(r), free(fd), free(gm);
90    return s;
91}
92
93static void parse_words(const char *s, char ***v, size_t *n) {
94    mesh_strs_free(*v, *n);
95    *v = NULL, *n = 0;
96    if (strcmp(s, "-") == 0) return;
97    mt_strs w = mt_split(s, ',');
98    for (size_t i = 0; i < w.n; i++) msn_strs_push(v, n, w.v[i]);
99    mt_strs_free(&w);
100}
101
102static bool parse_terms(const char *s, mesh_term **v, size_t *n, char **err) {
103    for (size_t i = 0; i < *n; i++) mesh_term_free(&(*v)[i]);
104    free(*v);
105    *v = NULL, *n = 0;
106    if (strcmp(s, "-") == 0) return true;
107    mt_strs w = mt_split(s, ',');
108    for (size_t i = 0; i < w.n; i++) {
109        mesh_term t;
110        char *e = NULL;
111        if (!mesh_term_parse(w.v[i], &t, &e)) {
112            *err = e;
113            mt_strs_free(&w);
114            return false;
115        }
116        msn_terms_push(v, n, t);
117    }
118    mt_strs_free(&w);
119    return true;
120}
121
122static char *bad(const char *k, const char *v) {
123    char *q = rf_debug_str(v);
124    char *e = mt_fmt("needs: %s=%s", k, q);
125    free(q);
126    return e;
127}
128
129static void set_opt(char **dst, const char *v) { mt_set(dst, strcmp(v, "-") == 0 ? NULL : v); }
130
131static bool field(mesh_needs *n, const char *k, const char *v, char **err) {
132    if (strcmp(k, "b") == 0) {
133        parse_words(v, &n->backends, &n->nbackends);
134    } else if (strcmp(k, "ar") == 0) set_opt(&n->arch, v);
135    else if (strcmp(k, "mm") == 0) {
136        if (!mp_u32(v, &n->min_mem_mb)) return *err = bad(k, v), false;
137    } else if (strcmp(k, "pf") == 0) return parse_terms(v, &n->prefer, &n->nprefer, err);
138    else if (strcmp(k, "av") == 0) return parse_terms(v, &n->avoid, &n->navoid, err);
139    else if (strcmp(k, "r") == 0) {
140        if (strcmp(v, "all") == 0) n->replicas = (mesh_replicas){true, 0};
141        else {
142            uint32_t r;
143            if (!mp_u32(v, &r)) return *err = bad(k, v), false;
144            n->replicas = (mesh_replicas){false, r};
145        }
146    } else if (strcmp(k, "sp") == 0) n->pack = strcmp(v, "pack") == 0;
147    else if (strcmp(k, "co") == 0) set_opt(&n->colocate_with, v);
148    else if (strcmp(k, "fd") == 0) {
149        parse_words(v, &n->feeds, &n->nfeeds);
150    } else if (strcmp(k, "g") == 0) set_opt(&n->group, v);
151    else if (strcmp(k, "gs") == 0) {
152        if (!mp_u32(v, &n->group_size)) return *err = bad(k, v), false;
153    } else if (strcmp(k, "gm") == 0) return parse_terms(v, &n->group_members, &n->ngroup_members, err);
154    else if (strcmp(k, "pr") == 0) {
155        if (!mp_i32(v, &n->priority)) return *err = bad(k, v), false;
156    } else if (strcmp(k, "cpu") == 0) {
157        float c;
158        if (!mp_f32(v, &c) || !isfinite(c)) return *err = bad(k, v), false;
159        n->cpu = c;
160    } else if (strcmp(k, "mem") == 0) {
161        if (!mp_u32(v, &n->mem_mb)) return *err = bad(k, v), false;
162    } else if (strcmp(k, "mg") == 0) {
163        if (strcmp(v, "state") == 0) n->migrate = MESH_MIGRATE_STATE;
164        else if (strcmp(v, "restart") == 0) n->migrate = MESH_MIGRATE_RESTART;
165        else if (strcmp(v, "never") == 0) n->migrate = MESH_MIGRATE_NEVER;
166        else return *err = bad(k, v), false;
167    } else if (strcmp(k, "ls") == 0) n->lockstep = strcmp(v, "1") == 0;
168    else if (strcmp(k, "vb") == 0) {
169        if (!mp_u32(v, &n->view_bytes)) return *err = bad(k, v), false;
170    } else if (strcmp(k, "w") == 0) {
171        uint32_t w;
172        if (!mp_u32(v, &w) || w < 1 || w > MESH_MAX_WARPS) return *err = bad(k, v), false;
173        n->warps = w;
174    } else if (strcmp(k, "sh") == 0) {
175        if (!mp_u32(v, &n->shared)) return *err = bad(k, v), false;
176    } else {
177        char *q = rf_debug_str(k);
178        *err = mt_fmt("needs: unknown key %s", q);
179        free(q);
180        return false;
181    }
182    return true;
183}
184
185bool msn_decode(const char *s, mesh_needs *out, char **err) {
186    mesh_needs n = msn_default();
187    mt_strs kvs = mt_split(s, ';');
188    bool ok = true;
189    for (size_t i = 0; i < kvs.n && ok; i++) {
190        char *k, *v;
191        if (!mt_split_once(kvs.v[i], '=', &k, &v)) {
192            char *q = rf_debug_str(kvs.v[i]);
193            *err = mt_fmt("needs: %s", q);
194            free(q);
195            ok = false;
196            break;
197        }
198        ok = field(&n, k, v, err);
199        free(k);
200        free(v);
201    }
202    mt_strs_free(&kvs);
203    if (!ok) {
204        mesh_needs_free(&n);
205        return false;
206    }
207    *out = n;
208    return true;
209}
210
211char *msn_summary(const mesh_needs *n) {
212    mt_strs s = {0};
213    char *r = msn_replicas_word(n->replicas);
214    mt_push_own(&s, mt_fmt("replicas %s", r));
215    free(r);
216    if (n->pack) mt_push(&s, "pack");
217    char *b = join(n->backends, n->nbackends, "");
218    mt_push_own(&s, mt_fmt("backends %s", b));
219    free(b);
220    if (n->arch) mt_push_own(&s, mt_fmt("arch %s", n->arch));
221    if (n->min_mem_mb > 0) mt_push_own(&s, mt_fmt("min_mem_mb %u", n->min_mem_mb));
222    if (n->nprefer) {
223        char *t = msn_terms_text(n->prefer, n->nprefer);
224        mt_push_own(&s, mt_fmt("prefer %s", t));
225        free(t);
226    }
227    if (n->navoid) {
228        char *t = msn_terms_text(n->avoid, n->navoid);
229        mt_push_own(&s, mt_fmt("avoid %s", t));
230        free(t);
231    }
232    if (n->colocate_with) mt_push_own(&s, mt_fmt("colocate_with %s", n->colocate_with));
233    if (n->group) {
234        char *t = msn_terms_text(n->group_members, n->ngroup_members);
235        mt_push_own(&s, mt_fmt("group %s size %u members %s", n->group, n->group_size, t));
236        free(t);
237    }
238    if (n->priority != 0) mt_push_own(&s, mt_fmt("priority %d", n->priority));
239    if (n->cpu > 0.0f) mt_push_own(&s, mt_fmt("cpu %.2f", (double)n->cpu));
240    if (n->mem_mb > 0) mt_push_own(&s, mt_fmt("mem_mb %u", n->mem_mb));
241    mt_push_own(&s, mt_fmt("view %u KiB", n->view_bytes / 1024 + (n->view_bytes % 1024 != 0)));
242    if (n->warps > 1 || n->shared > 0) mt_push_own(&s, mt_fmt("node %u warps, %u shared bytes", n->warps, n->shared));
243    mt_push_own(&s, mt_fmt("migrate %s", mesh_migrate_word(n->migrate)));
244    char *out = mt_join(&s, " ");
245    mt_strs_free(&s);
246    return out;
247}