1/* jobs.rs's Node, leading: step_down, start_lead, activate, flush, snapshots, commit_check. */
2#include "internal.h"
3#include "mesh/jobs/lid.h"
4
5#include <stdlib.h>
6#include <string.h>
7
8mj_lead *mj_lead_new(uint32_t n);
9void mj_lead_free(mj_lead *l);
10void mj_lead_activate(mj_node *n, uint64_t now, uint64_t unix_ms, mj_outs *out);
11
12static unsigned ones(uint32_t x) {
13    unsigned c = 0;
14    for (; x; x &= x - 1) c++;
15    return c;
16}
17
18mj_outs mj_node_step_down(mj_node *n) {
19    mj_outs out = {0};
20    mj_lead *l = n->lead;
21    if (!l) return out;
22    n->lead = NULL;
23    for (size_t i = 0; i < l->nwaits; i++)
24        mj_outs_push(&out, mj_out_answer(l->waits[i].token, false,
25                                         mstrdup("jobs: the leader changed before the change was committed; `jobs` shows whether it was")));
26    for (size_t i = 0; i < l->nheld; i++)
27        mj_outs_push(&out, mj_out_answer(l->held[i].token, false,
28                                         mstrdup("jobs: the leader changed before the submit found a member running its program; submit again")));
29    /* Its entries not committed yet stay pending in the replica (a majority may have them: the next leader's pull). */
30    mj_lead_free(l);
31    return out;
32}
33
34void mj_node_start_lead(mj_node *n, uint64_t term, uint64_t now, uint64_t unix_ms, mj_outs *out) {
35    uint32_t me = n->cfg.me;
36    uint64_t lid;
37    if (!mj_lid_next(term, n->rep.lid, &lid)) {
38        if (n->behind != term) {
39            n->behind = term;
40            mj_outs_push(out, mj_out_log(mfmt("jobs: system %u leads fleet term %llu but knows job leadership %llx of term %llu: waiting for the fleet's view", me,
41                                              (unsigned long long)term, (unsigned long long)n->rep.lid, (unsigned long long)mj_lid_term(n->rep.lid))));
42        }
43        return;
44    }
45    /* Persisted before any frame of it leaves (a restart leads above it). */
46    mj_node_log(n, mj_log_lid(lid), out);
47    n->has_lid_from = true;
48    n->lid_from = me;
49    n->next = 0;
50    n->has_staging = false;
51    mj_outs_push(out, mj_out_log(mfmt("jobs: system %u leads the job table (term %llu, leadership %llx); pulling the members' tables", me,
52                                      (unsigned long long)term, (unsigned long long)lid)));
53    mj_lead *l = mj_lead_new(n->cfg.n);
54    l->lid = lid;
55    l->term = term;
56    l->responded = (uint32_t)1 << me;
57    l->answers[me] = mxalloc(sizeof(mj_answer));
58    *l->answers[me] = mj_answer_of(&n->rep);
59    uint64_t k;
60    momap_iter cur2_;
61    momap_iter_init(&cur2_, n->held);
62    for (const mj_held *h = momap_iter_next1(&cur2_, &k); h; h = momap_iter_next1(&cur2_, &k)) {
63        if (l->nadopt == l->capadopt) {
64            l->capadopt = l->capadopt ? 2 * l->capadopt : 8;
65            l->adopt = mxrealloc(l->adopt, l->capadopt * sizeof *l->adopt);
66        }
67        l->adopt[l->nadopt++] = (mj_adopt){me, mj_held_clone(h)};
68    }
69    l->pull_at = now;
70    n->lead = l;
71    uint32_t linked = n->view.linked & ~((uint32_t)1 << me);
72    for (uint32_t m = 0; m < 32; m++)
73        if (linked >> m & 1) mj_outs_push(out, mj_out_send(m, mj_frame_pull(lid)));
74    if (n->cfg.quorum <= 1) mj_lead_activate(n, now, unix_ms, out);
75}
76
77/* A majority's tables are merged: commit the merge as this leadership's first entry, rebuild the
78 * queues, adopt the leases. */
79void mj_lead_activate(mj_node *n, uint64_t now, uint64_t unix_ms, mj_outs *out) {
80    (void)unix_ms;
81    uint32_t me = n->cfg.me;
82    uint64_t dead = n->cfg.dead_ms;
83    uint32_t linked = n->view.linked & ~((uint32_t)1 << me);
84    mj_lead *l = n->lead;
85    uint64_t lid = l->lid;
86    uint32_t responded = l->responded;
87    mj_answer *in[MJ_MEMBERS];
88    size_t nin = 0;
89    for (int k = 0; k < MJ_MEMBERS; k++)
90        if (l->answers[k]) in[nin++] = l->answers[k];
91    mj_table working = mj_merge(in, nin);
92    mj_ops ops = {0};
93    mj_table_ops(&working, &ops);
94    /* The first entry: the merge (a replace: the entries of the leaderships before it are in it). */
95    mj_node_log(n, mj_log_entry(lid, 1, true, mj_ops_clone(&ops)), out);
96    n->next = 2;
97    l->active = true;
98    for (int k = 0; k < MJ_MEMBERS; k++)
99        if (l->answers[k]) {
100            mj_answer_free(l->answers[k]);
101            free(l->answers[k]);
102            l->answers[k] = NULL;
103        }
104    mj_table_free(&l->working);
105    l->working = working;
106    l->n = 1;
107    l->acked[me] = 1;
108    mj_op_parts ps = mj_op_parts_of(&ops);
109    for (size_t i = 0; i < ps.n; i++)
110        for (uint32_t k = 0; k < 32; k++)
111            if (linked >> k & 1) mj_outs_push(out, mj_out_send(k, mj_frame_state(lid, MESH_JOBS_ST_PART, 1, 0, (uint32_t)i, mj_ops_clone(&ps.v[i]))));
112    for (uint32_t k = 0; k < 32; k++)
113        if (linked >> k & 1) mj_outs_push(out, mj_out_send(k, mj_frame_state(lid, MESH_JOBS_ST_REPLACE_END, 1, 0, (uint32_t)ps.n, (mj_ops){0})));
114    mj_op_parts_free(&ps);
115    mj_ops_free(&ops);
116    /* The queues: every active job's items left, but those in adopted leases. */
117    momap_clear(l->disp.queues);
118    uint64_t id;
119    for (const mj_job *j = mj_table_first(&l->working, &id); j; j = mj_table_next(&l->working, &id))
120        if (mj_job_active(j)) {
121            mj_queue *q = momap_put1(l->disp.queues, id, NULL);
122            *q = (mj_queue){mj_job_left(j), {0}};
123        }
124    uint64_t adopted = 0, top = 0;
125    for (size_t i = 0; i < l->nadopt; i++) {
126        uint32_t k = l->adopt[i].member;
127        mj_held *h = &l->adopt[i].h;
128        if (h->lease > top) top = h->lease;
129        mj_queue *qq = momap_get1(l->disp.queues, h->job);
130        if (!qq) continue;
131        mj_ranges items = mj_ranges_intersect(&h->items, &qq->fresh);
132        if (mj_ranges_is_empty(&items)) {
133            mj_ranges_free(&items);
134            continue;
135        }
136        mj_ranges_subtract(&qq->fresh, &items);
137        mj_ranges in2 = mj_ranges_intersect(&items, &mj_table_get(&l->working, h->job)->spec.items);
138        mj_ranges_free(&items);
139        uint64_t size = mj_ranges_len(&in2);
140        adopted += size;
141        bool fresh;
142        mj_lease *x = momap_put1(l->disp.leases, h->lease, &fresh);
143        if (!fresh) {
144            free(x->program);
145            mj_ranges_free(&x->out);
146        }
147        *x = (mj_lease){h->lease, h->job, k, mstrdup(h->program), in2, size, now, now + dead + 10000};
148    }
149    for (size_t i = 0; i < l->nadopt; i++) mj_held_free(&l->adopt[i].h);
150    l->nadopt = 0;
151    /* Lease ids of this leadership, past any a member holds (review finding 10: never the wall clock's). */
152    uint64_t base = mj_lease_base(lid);
153    l->disp.next_lease = base > top ? base : top;
154    size_t jobs = 0, active = 0;
155    for (const mj_job *j = mj_table_first(&l->working, &id); j; j = mj_table_next(&l->working, &id)) {
156        jobs++;
157        active += mj_job_active(j);
158    }
159    mj_outs_push(out, mj_out_log(mfmt("jobs: system %u leads the job table: %u answer(s) merged, %zu job(s) (%zu active), %zu lease(s) of %llu item(s) adopted",
160                                      me, ones(responded), jobs, active, momap_len(l->disp.leases), (unsigned long long)adopted)));
161    mj_node_commit_check(n, out);
162}
163
164/* This tick's ops as one delta (persisted, sent to every member that is in step). */
165void mj_node_flush(mj_node *n, mj_outs *out) {
166    uint32_t me = n->cfg.me;
167    uint32_t linked = n->view.linked & ~((uint32_t)1 << me);
168    mj_lead *l = n->lead;
169    if (!l || !l->active) return;
170    /* Dupes counted this tick: one op per job. */
171    uint64_t job;
172    momap_iter cur3_;
173    momap_iter_init(&cur3_, l->dupes);
174    for (const uint64_t *cnt = momap_iter_next1(&cur3_, &job); cnt; cnt = momap_iter_next1(&cur3_, &job)) {
175        mj_job *j = mj_table_get(&l->working, job);
176        if (!j) continue;
177        uint64_t total = j->dupes > *cnt ? j->dupes : *cnt;
178        if (total > j->dupes) {
179            mj_op op = mj_op_dupes(job, total);
180            mj_table_apply(&l->working, &op);
181            mj_ops_push(&l->ops, op);
182        }
183    }
184    if (!l->ops.n) return;
185    mj_ops ops = l->ops;
186    l->ops = (mj_ops){0};
187    uint64_t lid = l->lid;
188    mj_op_parts ps = mj_op_parts_of(&ops);
189    mj_ops_free(&ops);
190    for (size_t i = 0; i < ps.n; i++) {
191        l->n++;
192        uint64_t nn = l->n, commit = l->commit;
193        uint32_t due = l->snap_due;
194        l->acked[me] = nn;
195        mj_node_log(n, mj_log_entry(lid, nn, false, mj_ops_clone(&ps.v[i])), out);
196        n->next = nn + 1;
197        uint32_t to = linked & ~due;
198        for (uint32_t k = 0; k < 32; k++)
199            if (to >> k & 1) mj_outs_push(out, mj_out_send(k, mj_frame_state(lid, MESH_JOBS_ST_DELTA, nn, commit, 0, mj_ops_clone(&ps.v[i]))));
200    }
201    mj_op_parts_free(&ps);
202}
203
204typedef struct {
205    mj_frame *v;
206    size_t n;
207} frames;
208
209static void push_frame(frames *f, mj_frame x) {
210    f->v = mxrealloc(f->v, (f->n + 1) * sizeof *f->v);
211    f->v[f->n++] = x;
212}
213
214static void table_parts(frames *f, uint64_t lid, uint64_t nn, uint64_t commit, const mj_table *t, uint32_t end_kind) {
215    mj_ops ops = {0};
216    mj_table_ops(t, &ops);
217    mj_op_parts ps = mj_op_parts_of(&ops);
218    mj_ops_free(&ops);
219    for (size_t i = 0; i < ps.n; i++) {
220        push_frame(f, mj_frame_state(lid, MESH_JOBS_ST_PART, nn, commit, (uint32_t)i, ps.v[i]));
221        ps.v[i] = (mj_ops){0};
222    }
223    push_frame(f, mj_frame_state(lid, end_kind, nn, commit, (uint32_t)ps.n, (mj_ops){0}));
224    mj_op_parts_free(&ps);
225}
226
227/* Members that need the committed table, then the entries after it. */
228void mj_node_snapshots(mj_node *n, mj_outs *out) {
229    uint32_t me = n->cfg.me;
230    uint32_t linked = n->view.linked & ~((uint32_t)1 << me);
231    mj_lead *l = n->lead;
232    if (!l) return;
233    uint32_t due = l->snap_due & linked;
234    if (!due) return;
235    l->snap_due &= ~due;
236    uint64_t lid = l->lid, commit = l->commit;
237    frames f = {0};
238    /* The committed table only once this leadership committed its first entry (the merge): before, this
239     * leader's committed table is its own from before it led, maybe behind the member's, which a snapshot
240     * would replace. Until then the member gets the entries alone (the merge first). */
241    if (commit > 0) table_parts(&f, lid, commit, commit, &n->rep.committed, MESH_JOBS_ST_SNAP_END);
242    size_t entries = 0;
243    for (size_t i = 0; i < n->rep.npending; i++) {
244        const mj_entry *e = &n->rep.pending[i];
245        if (e->lid != lid) continue;
246        entries++;
247        if (e->replace) table_parts(&f, lid, e->n, commit, &e->table, MESH_JOBS_ST_REPLACE_END);
248        else push_frame(&f, mj_frame_state(lid, MESH_JOBS_ST_DELTA, e->n, commit, 0, mj_ops_clone(&e->ops)));
249    }
250    for (uint32_t k = 0; k < 32; k++) {
251        if (!(due >> k & 1)) continue;
252        mj_outs_push(out, mj_out_log(mfmt("jobs: system %u gets the job table (commit %llu, %zu entries after it)", k, (unsigned long long)commit, entries)));
253        for (size_t i = 0; i < f.n; i++) mj_outs_push(out, mj_out_send(k, mj_frame_clone(&f.v[i])));
254    }
255    for (size_t i = 0; i < f.n; i++) mj_frame_free(&f.v[i]);
256    free(f.v);
257}
258
259static int desc(const void *a, const void *b) {
260    uint64_t x = *(const uint64_t *)a, y = *(const uint64_t *)b;
261    return x < y ? 1 : x > y ? -1 : 0;
262}
263
264/* The commit point: the highest entry a majority persisted. */
265void mj_node_commit_check(mj_node *n, mj_outs *out) {
266    uint32_t me = n->cfg.me;
267    uint32_t linked = n->view.linked & ~((uint32_t)1 << me);
268    uint32_t q = n->cfg.quorum > 1 ? n->cfg.quorum : 1;
269    mj_lead *l = n->lead;
270    if (!l || !l->active) return;
271    uint64_t *a = mxalloc(n->cfg.n * sizeof *a);
272    memcpy(a, l->acked, n->cfg.n * sizeof *a);
273    qsort(a, n->cfg.n, sizeof *a, desc);
274    uint64_t c = a[q - 1];
275    free(a);
276    if (c <= l->commit) return;
277    l->commit = c;
278    uint64_t lid = l->lid;
279    mj_wait *done = NULL;
280    size_t ndone = 0, keep = 0;
281    for (size_t i = 0; i < l->nwaits; i++) {
282        if (l->waits[i].n <= c) {
283            done = mxrealloc(done, (ndone + 1) * sizeof *done);
284            done[ndone++] = l->waits[i];
285        } else l->waits[keep++] = l->waits[i];
286    }
287    l->nwaits = keep;
288    uint32_t due = l->snap_due;
289    mj_node_log(n, mj_log_commit(lid, c), out);
290    for (size_t i = 0; i < ndone; i++) mj_outs_push(out, mj_out_answer(done[i].token, done[i].ok, done[i].text));
291    free(done);
292    uint32_t to = linked & ~due;
293    for (uint32_t k = 0; k < 32; k++)
294        if (to >> k & 1) mj_outs_push(out, mj_out_send(k, mj_frame_state(lid, MESH_JOBS_ST_COMMIT, c, c, 0, (mj_ops){0})));
295}