1/* sched.rs: Move, Tracker, unit, start_move, resume_moves, MigOutcome, mig_verdict, end_move. */
2#include "mesh/sched_tracker.h"
3#include "mesh/mt.h"
4
5#include <stdio.h>
6#include <stdlib.h>
7#include <string.h>
8
9void mst_move_free(mst_move *m) {
10    mt_strs_free(&m->programs);
11    free(m->why);
12    *m = (mst_move){0};
13}
14
15mst_move mst_move_clone(const mst_move *m) { return (mst_move){mt_strs_clone(&m->programs), m->from, m->to, m->state, mt_strdup(m->why), m->forced}; }
16
17bool mst_move_eq(const mst_move *a, const mst_move *b) {
18    return mt_strs_eq(&a->programs, &b->programs) && a->from == b->from && a->to == b->to && a->state == b->state && strcmp(a->why, b->why) == 0 &&
19           a->forced == b->forced;
20}
21
22typedef struct {
23    char *name;
24    uint32_t from;
25    uint64_t since;
26} cand_t;
27
28typedef struct {
29    uint32_t k;
30    uint64_t since;
31    bool timing, busy;
32} owner_t;
33
34struct mst_tracker {
35    cand_t *cand; /* sorted by name */
36    size_t ncand;
37    ms_map last_move; /* name -> decimal ms */
38    uint64_t *moves;  /* a queue of start times */
39    size_t nmoves;
40    owner_t owner[64]; /* by member */
41    size_t nowner;
42    bool has_group;
43    char *group_name;
44    ml_list group_members;
45    uint64_t group_since;
46    mt_strs notes;
47};
48
49mst_tracker *mst_new(void) { return mt_alloc(sizeof(mst_tracker)); }
50
51static void cand_remove_at(mst_tracker *t, size_t i) {
52    free(t->cand[i].name);
53    memmove(t->cand + i, t->cand + i + 1, (t->ncand - i - 1) * sizeof *t->cand);
54    t->ncand--;
55}
56
57static long cand_find(const mst_tracker *t, const char *name) {
58    for (size_t i = 0; i < t->ncand; i++)
59        if (strcmp(t->cand[i].name, name) == 0) return (long)i;
60    return -1;
61}
62
63static bool cand_remove(mst_tracker *t, const char *name) {
64    long i = cand_find(t, name);
65    if (i < 0) return false;
66    cand_remove_at(t, (size_t)i);
67    return true;
68}
69
70void mst_free(mst_tracker *t) {
71    if (!t) return;
72    while (t->ncand) cand_remove_at(t, t->ncand - 1);
73    free(t->cand);
74    ms_free(&t->last_move);
75    free(t->moves);
76    free(t->group_name);
77    ml_free(&t->group_members);
78    mt_strs_free(&t->notes);
79    free(t);
80}
81
82const mt_strs *mst_notes(const mst_tracker *t) { return &t->notes; }
83
84static owner_t *owner_entry(mst_tracker *t, uint32_t k, uint64_t now) {
85    for (size_t i = 0; i < t->nowner; i++)
86        if (t->owner[i].k == k) return &t->owner[i];
87    t->owner[t->nowner] = (owner_t){k, now, false, false};
88    return &t->owner[t->nowner++];
89}
90
91static void owner_remove(mst_tracker *t, uint32_t k) {
92    for (size_t i = 0; i < t->nowner; i++)
93        if (t->owner[i].k == k) {
94            t->owner[i] = t->owner[--t->nowner];
95            return;
96        }
97}
98
99void mst_owner(mst_tracker *t, msm_member *ms, size_t n, uint64_t now, const msw_weights *w) {
100    for (size_t i = 0; i < n; i++) {
101        msm_member *m = &ms[i];
102        if (!m->up || !m->load_known) {
103            free(m->busy);
104            m->busy = NULL;
105            owner_remove(t, m->id);
106            continue;
107        }
108        mof32 share = strcmp(m->gpu, "wasm") != 0 ? m->gpu_share : mof32_none();
109        bool hot = (share.some && share.v < w->owner_share) || (m->psi_cpu.some && m->psi_cpu.v > w->owner_psi);
110        bool cool = (!share.some || share.v > w->owner_share * 1.33f) && (!m->psi_cpu.some || m->psi_cpu.v < w->owner_psi * 0.6f);
111        owner_t *e = owner_entry(t, m->id, now);
112        bool want = e->busy ? !cool : hot;
113        if (want != e->busy) {
114            if (!e->timing) e->since = now, e->timing = true;
115            else if (now - (now > e->since ? e->since : now) >= w->owner_ms) e->since = now, e->timing = false, e->busy = want;
116        } else if (e->timing) e->since = now, e->timing = false;
117        free(m->busy);
118        m->busy = NULL;
119        if (e->busy) {
120            char s[32], p[32];
121            if (share.some) snprintf(s, sizeof s, "%.2f", (double)share.v);
122            else strcpy(s, "-");
123            if (m->psi_cpu.some) snprintf(p, sizeof p, "%.3f", (double)m->psi_cpu.v);
124            else strcpy(p, "-");
125            m->busy = mt_fmt("other tenants load it (share %s, PSI cpu %s)", s, p);
126        }
127    }
128}
129
130uint32_t mst_recent(mst_tracker *t, uint64_t now) {
131    size_t drop = 0;
132    while (drop < t->nmoves && now - (now > t->moves[drop] ? t->moves[drop] : now) >= 60000) drop++;
133    if (drop) memmove(t->moves, t->moves + drop, (t->nmoves - drop) * sizeof *t->moves);
134    t->nmoves -= drop;
135    return (uint32_t)t->nmoves;
136}
137
138static void last_move_set(mst_tracker *t, const char *name, uint64_t now) {
139    char v[24];
140    snprintf(v, sizeof v, "%llu", (unsigned long long)now);
141    ms_set(&t->last_move, name, v, true);
142}
143
144void mst_moved(mst_tracker *t, const mst_move *m, uint64_t now) {
145    t->moves = mt_realloc(t->moves, (t->nmoves + 1) * sizeof *t->moves);
146    t->moves[t->nmoves++] = now;
147    for (size_t i = 0; i < m->programs.n; i++) {
148        last_move_set(t, m->programs.v[i], now);
149        cand_remove(t, m->programs.v[i]);
150    }
151}
152
153mt_strs mst_unit(const mfd_doc *doc, const char *name, uint32_t from) {
154    mt_strs v = {0};
155    mt_push(&v, name);
156    for (size_t i = 0; i < v.n; i++) {
157        char *a = mt_strdup(v.v[i]);
158        for (size_t j = 0; j < doc->nprogs; j++) {
159            const mfd_prog *p = &doc->prog[j];
160            if (p->needs.colocate_with && strcmp(p->needs.colocate_with, a) == 0 && !mt_strs_has(&v, doc->pname[j]) && ml_opt_has_or_all(&p->at, from))
161                mt_push(&v, doc->pname[j]);
162        }
163        free(a);
164    }
165    /* Stop order: a program that feeds another in the unit stops before it. */
166    mt_strs out = {0};
167    while (v.n) {
168        size_t pick = 0;
169        bool found = false;
170        for (size_t i = 0; i < v.n && !found; i++) {
171            bool fed = false;
172            for (size_t j = 0; j < v.n; j++) {
173                if (j == i || strcmp(v.v[j], v.v[i]) == 0) continue;
174                const mfd_prog *o = mfd_prog_get(doc, v.v[j]);
175                if (o && msn_strs_has(o->needs.feeds, o->needs.nfeeds, v.v[i])) fed = true;
176            }
177            if (!fed) pick = i, found = true;
178        }
179        mt_push(&out, v.v[pick]);
180        mt_strs_remove(&v, pick);
181    }
182    mt_strs_free(&v);
183    return out;
184}
185
186static char *list_text(const ml_list *l) { return ml_text(l); }
187
188bool mst_step(mst_tracker *t, const msp_input *in, uint64_t now, bool busy, mst_move *out) {
189    mt_strs_free(&t->notes);
190    const mfd_doc *doc = in->doc;
191    mt_strs ord = msp_order(doc);
192    bool result = false;
193    /* Forced: a pin the placement left for a migration. */
194    for (size_t oi = 0; oi < ord.n; oi++) {
195        const char *name = ord.v[oi];
196        const mfd_prog *p = mfd_prog_get(doc, name);
197        if (!p->pin.some || !p->at.some) continue;
198        if (p->has_from || ml_eq(&p->pin.l, &p->at.l) || p->needs.migrate != MESH_MIGRATE_STATE) continue;
199        /* Colocated with a loaded anchor: it moves with the anchor's unit, never on its own pin. */
200        if (p->needs.colocate_with && mfd_prog_get(doc, p->needs.colocate_with)) continue;
201        long fi = -1, ti = -1;
202        for (size_t i = 0; i < p->at.l.n && fi < 0; i++)
203            if (!ml_has(&p->pin.l, p->at.l.v[i])) fi = (long)i;
204        if (fi < 0) continue;
205        for (size_t i = 0; i < p->pin.l.n && ti < 0; i++)
206            if (!ml_has(&p->at.l, p->pin.l.v[i])) ti = (long)i;
207        if (ti < 0) continue;
208        char *pl = list_text(&p->pin.l);
209        if (busy) {
210            mt_push_own(&t->notes, mt_fmt("%s: pinned to %s: waits for the migration in flight", name, pl));
211            free(pl);
212            goto done;
213        }
214        uint32_t from = p->at.l.v[fi];
215        *out = (mst_move){mst_unit(doc, name, from), from, p->pin.l.v[ti], true, mt_fmt("pinned by the operator to %s", pl), true};
216        free(pl);
217        result = true;
218        goto done;
219    }
220    if (doc->sched_paused) {
221        mt_push(&t->notes, "paused (`sched resume`)");
222        goto done;
223    }
224    msp_books books = msp_books_new(in->nmembers);
225    for (size_t oi = 0; oi < ord.n; oi++) {
226        const mfd_prog *p = mfd_prog_get(doc, ord.v[oi]);
227        ml_list on = msp_members_of_at(&p->at, in->members, in->nmembers, &p->needs);
228        for (size_t i = 0; i < on.n; i++)
229            if (on.v[i] < in->nmembers) msp_books_add(&books, on.v[i], &p->needs);
230        ml_free(&on);
231    }
232    /* ready: (gain, move) */
233    double *gain = NULL;
234    mst_move *ready = NULL;
235    size_t nready = 0;
236    mt_strs seen = {0};
237    for (size_t oi = 0; oi < ord.n; oi++) {
238        const char *name = ord.v[oi];
239        const mfd_prog *p = mfd_prog_get(doc, name);
240        const mesh_needs *n = &p->needs;
241        bool movable = !n->replicas.all && !p->pin.some && !n->lockstep && n->migrate != MESH_MIGRATE_NEVER && !p->has_from &&
242                       (!n->colocate_with || !mfd_prog_get(doc, n->colocate_with));
243        if (!p->at.some || !p->at.l.n || !movable) {
244            cand_remove(t, name);
245            continue;
246        }
247        mt_push(&seen, name);
248        msp_row *r;
249        size_t nr;
250        msp_rows(in, name, &p->at.l, &books, NULL, &r, &nr);
251        /* The worst-placed replica (ties: the higher id) against the best member it is not on. */
252        long from_i = -1;
253        double s_from = 0;
254        uint32_t from = 0;
255        for (size_t i = 0; i < p->at.l.n; i++) {
256            uint32_t k = p->at.l.v[i];
257            for (size_t j = 0; j < nr; j++)
258                if (r[j].k == k && r[j].ok) {
259                    double s = r[j].score.total;
260                    if (from_i < 0 || s < s_from || (s == s_from && k > from)) from_i = (long)i, s_from = s, from = k;
261                    break;
262                }
263        }
264        if (from_i < 0) {
265            msp_rows_free(r, nr);
266            continue;
267        }
268        uint32_t *ks = mt_alloc((nr ? nr : 1) * sizeof *ks);
269        double *ts = mt_alloc((nr ? nr : 1) * sizeof *ts);
270        size_t nok = msp_sorted_ok(r, nr, ks, ts);
271        long to_i = -1;
272        for (size_t i = 0; i < nok && to_i < 0; i++)
273            if (!ml_has(&p->at.l, ks[i])) to_i = (long)i;
274        if (to_i < 0) {
275            cand_remove(t, name);
276            free(ks), free(ts);
277            msp_rows_free(r, nr);
278            continue;
279        }
280        uint32_t to = ks[to_i];
281        double s_to = ts[to_i];
282        free(ks), free(ts);
283        msp_rows_free(r, nr);
284        double g = s_to / s_from - 1.0;
285        if (g <= in->w->improve) {
286            if (cand_remove(t, name)) mt_push_own(&t->notes, mt_fmt("%s: the gain fell to %.0f %%: stays on %u", name, g * 100.0, from));
287            continue;
288        }
289        long ci = cand_find(t, name);
290        if (ci < 0) {
291            t->cand = mt_realloc(t->cand, (t->ncand + 1) * sizeof *t->cand);
292            t->cand[t->ncand] = (cand_t){mt_strdup(name), from, now};
293            ci = (long)t->ncand++;
294        }
295        cand_t *c = &t->cand[ci];
296        if (c->from != from) c->from = from, c->since = now;
297        uint64_t held = now > c->since ? now - c->since : 0;
298        const char *lm = ms_get(&t->last_move, name);
299        bool has_dwell = lm != NULL;
300        uint64_t dwell = 0;
301        if (lm) {
302            uint64_t tm = strtoull(lm, NULL, 10);
303            dwell = now > tm ? now - tm : 0;
304        }
305        char *why = mt_fmt("system %u %.3f -> system %u %.3f (+%.0f %% for %.1f s)", from, s_from, to, s_to, g * 100.0, (double)held / 1e3);
306        if (held < in->w->hold_ms) {
307            mt_push_own(&t->notes, mt_fmt("%s: %s: holding (%llu ms needed)", name, why, (unsigned long long)in->w->hold_ms));
308            free(why);
309            continue;
310        }
311        if (has_dwell && dwell < in->w->dwell_ms) {
312            mt_push_own(&t->notes, mt_fmt("%s: %s: moved %.1f s ago (dwell %llu ms)", name, why, (double)dwell / 1e3, (unsigned long long)in->w->dwell_ms));
313            free(why);
314            continue;
315        }
316        gain = mt_realloc(gain, (nready + 1) * sizeof *gain);
317        ready = mt_realloc(ready, (nready + 1) * sizeof *ready);
318        gain[nready] = g;
319        ready[nready++] = (mst_move){mst_unit(doc, name, from), from, to, n->migrate == MESH_MIGRATE_STATE, why, false};
320    }
321    for (size_t i = t->ncand; i-- > 0;)
322        if (!mt_strs_has(&seen, t->cand[i].name)) cand_remove_at(t, i);
323    mt_strs_free(&seen);
324    uint32_t recent = mst_recent(t, now);
325    /* The biggest gain first (ties: the name). */
326    size_t best = 0;
327    for (size_t i = 1; i < nready; i++)
328        if (gain[i] > gain[best] || (gain[i] == gain[best] && strcmp(ready[i].programs.v[0], ready[best].programs.v[0]) < 0)) best = i;
329    if (nready) {
330        mst_move *m = &ready[best];
331        if (busy) mt_push_own(&t->notes, mt_fmt("%s: %s: waits for the migration in flight", m->programs.v[0], m->why));
332        else if (recent >= in->w->per_min)
333            mt_push_own(&t->notes, mt_fmt("%s: %s: budget spent (%u migrations in the last minute, at most %u)", m->programs.v[0], m->why, recent, in->w->per_min));
334        else {
335            *out = *m;
336            *m = (mst_move){0};
337            result = true;
338        }
339    }
340    for (size_t i = 0; i < nready; i++) mst_move_free(&ready[i]);
341    free(ready);
342    free(gain);
343    msp_books_free(&books);
344done:
345    mt_strs_free(&ord);
346    return result;
347}
348
349bool mst_group_step(mst_tracker *t, const msp_input *in, uint64_t now, char **group, ml_list *members, char **why) {
350    if (!in->doc->ngroups) return false;
351    const mfd_group *g = &in->doc->groups[0];
352    ml_list ideal;
353    char *w;
354    if (!msp_group_choice(in, g, &ideal, &w)) return false;
355    bool ok = false;
356    if (ml_eq(&ideal, &g->members)) {
357        t->has_group = false;
358        goto done;
359    }
360    double cur = msp_group_score(in, g, &g->members), nw = msp_group_score(in, g, &ideal);
361    bool lost = false;
362    for (size_t i = 0; i < g->members.n; i++)
363        if (g->members.v[i] >= in->nmembers || !in->members[g->members.v[i]].up) lost = true;
364    if (!lost && nw <= cur * (1.0 + in->w->improve) && ideal.n <= g->members.n) {
365        t->has_group = false;
366        goto done;
367    }
368    uint64_t since;
369    if (t->has_group && strcmp(t->group_name, g->name) == 0 && ml_eq(&t->group_members, &ideal)) since = t->group_since;
370    else {
371        mt_set(&t->group_name, g->name);
372        ml_free(&t->group_members);
373        t->group_members = ml_clone(&ideal);
374        t->group_since = now;
375        t->has_group = true;
376        since = now;
377    }
378    if ((now > since ? now - since : 0) < in->w->hold_ms) goto done;
379    char *l = ml_text(&g->members);
380    *why = mt_fmt("%s; now %s scores %.3f, proposed %.3f", w, l, cur, nw);
381    free(l);
382    *group = mt_strdup(g->name);
383    *members = ideal;
384    ideal = (ml_list){0};
385    ok = true;
386done:
387    ml_free(&ideal);
388    free(w);
389    return ok;
390}
391
392mfd_doc mst_start_move(const mfd_doc *doc, const mst_move *m, uint64_t since) {
393    mfd_doc d = mfd_doc_clone(doc);
394    for (size_t i = 0; i < m->programs.n; i++) {
395        mfd_prog *p = mfd_prog_get(&d, m->programs.v[i]);
396        if (!p) continue;
397        ml_list at = p->at.some ? ml_clone(&p->at.l) : (ml_list){0};
398        ml_remove_value(&at, m->from);
399        if (!ml_has(&at, m->to)) ml_push(&at, m->to);
400        ml_sort(&at);
401        ml_opt_free(&p->at);
402        p->at = (ml_opt){true, at};
403        p->has_from = p->has_to = m->state;
404        p->from = m->state ? m->from : 0;
405        p->to = m->state ? m->to : 0;
406        p->since = m->state ? since : 0;
407        free(p->why);
408        p->why = mt_fmt("moved from system %u to %u: %s", m->from, m->to, m->why);
409    }
410    return d;
411}
412
413mst_resumed mst_resume_moves(const mfd_doc *doc) {
414    mst_resumed r = {0};
415    mt_strs ord = msp_order(doc);
416    for (size_t oi = 0; oi < ord.n; oi++) {
417        const mfd_prog *p = mfd_prog_get(doc, ord.v[oi]);
418        if (!p->has_from || !p->has_to) continue;
419        size_t i = 0;
420        for (; i < r.n; i++)
421            if (r.moves[i].from == p->from && r.moves[i].to == p->to && r.since[i] == p->since) break;
422        if (i < r.n) mt_push(&r.moves[i].programs, ord.v[oi]);
423        else {
424            r.moves = mt_realloc(r.moves, (r.n + 1) * sizeof *r.moves);
425            r.since = mt_realloc(r.since, (r.n + 1) * sizeof *r.since);
426            mst_move m = {{0}, p->from, p->to, true, mt_strdup("taken over by a new leader"), false};
427            mt_push(&m.programs, ord.v[oi]);
428            r.moves[r.n] = m;
429            r.since[r.n++] = p->since;
430        }
431    }
432    mt_strs_free(&ord);
433    return r;
434}
435
436void mst_resumed_free(mst_resumed *r) {
437    for (size_t i = 0; i < r->n; i++) mst_move_free(&r->moves[i]);
438    free(r->moves);
439    free(r->since);
440    *r = (mst_resumed){0};
441}
442
443bool mst_mig_verdict(const mst_move *m, uint64_t since, bool target_up, const mst_outcome *outs, size_t nouts, const char *reason, uint64_t waited_ms,
444                     uint64_t limit_ms, bool *ok, char **why, double *gap_ms) {
445    *gap_ms = -1;
446    if (!target_up) {
447        *ok = false;
448        *why = mt_fmt("the target, system %u, was lost", m->to);
449        return true;
450    }
451    size_t got = 0;
452    uint32_t gap = 0;
453    mt_strs bad = {0};
454    for (size_t i = 0; i < m->programs.n; i++)
455        for (size_t j = 0; j < nouts; j++)
456            if (strcmp(outs[j].name, m->programs.v[i]) == 0 && outs[j].version >= since) {
457                got++;
458                if (!outs[j].ok) mt_push(&bad, outs[j].name);
459                if (outs[j].gap_us > gap) gap = outs[j].gap_us;
460                break;
461            }
462    if (got == m->programs.n) {
463        if (!bad.n) {
464            char *names = mt_join(&m->programs, ",");
465            *ok = true;
466            *why = mt_fmt("system %u runs %s", m->to, names);
467            free(names);
468            if (gap > 0) *gap_ms = (double)gap / 1e3;
469        } else {
470            char *names = mt_join(&bad, ",");
471            *ok = false;
472            *why = mt_fmt("system %u could not take %s (%s)", m->to, names, reason);
473            free(names);
474        }
475        mt_strs_free(&bad);
476        return true;
477    }
478    mt_strs_free(&bad);
479    if (waited_ms > limit_ms) {
480        *ok = false;
481        *why = mt_fmt("not done in %llu s", (unsigned long long)(limit_ms / 1000));
482        return true;
483    }
484    return false;
485}
486
487mfd_doc mst_end_move(const mfd_doc *doc, const mst_move *m, bool ok, const char *why) {
488    mfd_doc d = mfd_doc_clone(doc);
489    for (size_t i = 0; i < m->programs.n; i++) {
490        mfd_prog *p = mfd_prog_get(&d, m->programs.v[i]);
491        if (!p) continue;
492        p->has_from = p->has_to = false;
493        p->from = p->to = 0;
494        p->since = 0;
495        if (!ok) {
496            ml_list at = p->at.some ? ml_clone(&p->at.l) : (ml_list){0};
497            ml_remove_value(&at, m->to);
498            if (!ml_has(&at, m->from)) ml_push(&at, m->from);
499            ml_sort(&at);
500            ml_opt_free(&p->at);
501            p->at = (ml_opt){true, at};
502        }
503        mt_set(&p->why, why);
504    }
505    return d;
506}