1/* jobs.rs's Node: frames (on_frame, answer, follow, gap). */
2#include "internal.h"
3#include "mesh/abi/peer_frames.h"
4
5#include <stdlib.h>
6#include <string.h>
7
8mj_lead *mj_lead_new(uint32_t n);
9void mj_lead_activate(mj_node *n, uint64_t now, uint64_t unix_ms, mj_outs *out);
10
11mj_op_parts mj_op_parts_of(const mj_ops *ops) {
12    mj_parts p = mj_parts_of(ops->v, ops->n);
13    mj_op_parts r = {mxalloc(p.n * sizeof *r.v), p.n};
14    for (size_t i = 0; i < p.n; i++)
15        if (!mj_decode_ops(p.v[i].p, p.v[i].n, &r.v[i], NULL)) r.v[i] = (mj_ops){0};
16    mj_parts_free(&p);
17    return r;
18}
19
20void mj_op_parts_free(mj_op_parts *p) {
21    for (size_t i = 0; i < p->n; i++) mj_ops_free(&p->v[i]);
22    free(p->v);
23    *p = (mj_op_parts){0};
24}
25
26/* Whether frames of leadership `lid` from `from` are to be followed (a newer one is adopted, and persisted). */
27static bool follow(mj_node *n, uint32_t from, uint64_t lid, mj_outs *out) {
28    if (lid < n->rep.lid) return false;
29    if (lid > n->rep.lid) {
30        if (from != n->cfg.me && n->lead) {
31            mj_outs o = mj_node_step_down(n);
32            mj_outs_append(out, &o);
33        }
34        mj_node_log(n, mj_log_lid(lid), out);
35        n->has_lid_from = true;
36        n->lid_from = from;
37        n->next = 0;
38        n->has_staging = false;
39        return true;
40    }
41    /* The leadership this member followed before a restart: its leader is whoever speaks for it. */
42    if (!n->has_lid_from) {
43        n->has_lid_from = true;
44        n->lid_from = from;
45        n->next = 0;
46        n->has_staging = false;
47    }
48    return n->lid_from == from;
49}
50
51static void gap(mj_node *n, uint32_t from, uint64_t lid, uint64_t now, mj_outs *out) {
52    if ((now > n->gap_at ? now - n->gap_at : 0) < 500) return;
53    n->gap_at = now;
54    mj_node_send(n, from, mj_frame_ack(lid, n->next ? n->next - 1 : 0, true), out);
55}
56
57static mj_held *held_now(const mj_node *n, size_t *count) {
58    size_t nh = momap_len(n->held), i = 0;
59    mj_held *v = nh ? mxalloc(nh * sizeof *v) : NULL;
60    uint64_t k;
61    momap_iter cur1_;
62    momap_iter_init(&cur1_, n->held);
63    for (const mj_held *h = momap_iter_next1(&cur1_, &k); h; h = momap_iter_next1(&cur1_, &k)) v[i++] = mj_held_clone(h);
64    *count = nh;
65    return v;
66}
67
68typedef struct {
69    mj_frame *v;
70    size_t n;
71} frames;
72
73static void put_parts(frames *f, uint64_t lid, uint32_t kind, uint64_t s0, uint64_t s1, const mj_ops *ops, uint32_t *seq) {
74    mj_op_parts p = mj_op_parts_of(ops);
75    for (size_t i = 0; i < p.n; i++) {
76        f->v = mxrealloc(f->v, (f->n + 1) * sizeof *f->v);
77        f->v[f->n++] = mj_frame_have(lid, kind, (*seq)++, s0, s1, p.v[i], NULL, 0);
78        p.v[i] = (mj_ops){0};
79    }
80    mj_op_parts_free(&p);
81}
82
83void mj_node_answer(const mj_node *n, uint64_t lid, mj_frame **out, size_t *count) {
84    frames f = {0};
85    uint32_t seq = 0;
86    mj_ops committed = {0};
87    mj_table_ops(&n->rep.committed, &committed);
88    put_parts(&f, lid, MESH_JOBS_HAVE_COMMITTED, n->rep.commit_lid, n->rep.commit_n, &committed, &seq);
89    mj_ops_free(&committed);
90    /* The entries of one leadership as one stream (the merge takes them in any order within it). */
91    size_t i = 0;
92    while (i < n->rep.npending) {
93        uint64_t l = n->rep.pending[i].lid, last = 0;
94        mj_ops ops = {0};
95        while (i < n->rep.npending && n->rep.pending[i].lid == l) {
96            const mj_entry *e = &n->rep.pending[i];
97            last = e->n;
98            if (e->replace) mj_table_ops(&e->table, &ops);
99            else
100                for (size_t k = 0; k < e->ops.n; k++) mj_ops_push(&ops, mj_op_clone(&e->ops.v[k]));
101            i++;
102        }
103        put_parts(&f, lid, MESH_JOBS_HAVE_DELTA, l, last, &ops, &seq);
104        mj_ops_free(&ops);
105    }
106    size_t nh;
107    mj_held *held = held_now(n, &nh);
108    uint32_t total = (uint32_t)f.n;
109    f.v = mxrealloc(f.v, (f.n + 1) * sizeof *f.v);
110    f.v[f.n++] = mj_frame_have(lid, MESH_JOBS_HAVE_END, total, 0, 0, (mj_ops){0}, held, nh);
111    *out = f.v;
112    *count = f.n;
113}
114
115static void on_have(mj_node *n, uint32_t from, mj_frame *f, uint64_t now, uint64_t unix_ms, mj_outs *out) {
116    uint32_t quorum = n->cfg.quorum;
117    mj_lead *l = n->lead;
118    if (!l || l->lid != f->lid) return;
119    if (l->active) {
120        /* A late answer: its results are welcome (the table is a union, and the working table holds every
121         * committed entry already), and so are its leases for the items still queued (not leased again). */
122        for (size_t i = 0; i < f->ops.n; i++) {
123            mj_table_apply(&l->working, &f->ops.v[i]);
124            mj_ops_push(&l->ops, f->ops.v[i]);
125            f->ops.v[i] = (mj_op){0};
126        }
127        uint64_t dead = n->cfg.dead_ms;
128        for (size_t i = 0; i < f->nheld; i++) {
129            mj_held *h = &f->held[i];
130            if (momap_get1(l->disp.leases, h->lease)) continue;
131            mj_queue *q = momap_get1(l->disp.queues, h->job);
132            if (!q) continue;
133            mj_ranges items = mj_ranges_intersect(&h->items, &q->fresh), r2 = mj_ranges_intersect(&h->items, &q->retry);
134            mj_ranges_union(&items, &r2);
135            mj_ranges_free(&r2);
136            if (mj_ranges_is_empty(&items)) {
137                mj_ranges_free(&items);
138                continue;
139            }
140            mj_ranges_subtract(&q->fresh, &items);
141            mj_ranges_subtract(&q->retry, &items);
142            uint64_t cnt = mj_ranges_len(&items);
143            mj_outs_push(out, mj_out_log(mfmt("jobs: system %u's late answer: lease %llx of %llu item(s) adopted", from,
144                                              (unsigned long long)h->lease, (unsigned long long)cnt)));
145            if (h->lease > l->disp.next_lease) l->disp.next_lease = h->lease;
146            bool fresh;
147            mj_lease *x = momap_put1(l->disp.leases, h->lease, &fresh);
148            *x = (mj_lease){h->lease, h->job, from, mstrdup(h->program), items, cnt, now, now + dead + 10000};
149        }
150        return;
151    }
152    if (l->responded >> from & 1) return;
153    l->heard_mask |= (uint32_t)1 << from;
154    l->heard[from] = now;
155    /* A part numbered 0 starts an answer (a repeated pull's answer replaces an earlier one). */
156    if (f->seq == 0 && f->kind != MESH_JOBS_HAVE_END && l->answers[from]) {
157        mj_answer_free(l->answers[from]);
158        free(l->answers[from]);
159        l->answers[from] = NULL;
160    }
161    if (!l->answers[from]) {
162        l->answers[from] = mxalloc(sizeof(mj_answer));
163        mj_answer_init(l->answers[from]);
164    }
165    mj_answer *a = l->answers[from];
166    if (f->kind == MESH_JOBS_HAVE_END) {
167        if (a->spoilt || a->seq != f->seq) {
168            mj_outs_push(out, mj_out_log(mfmt("jobs: system %u's table arrived incomplete (%u of %u parts); asking again", from, a->seq, f->seq)));
169            mj_answer_free(a);
170            free(a);
171            l->answers[from] = NULL;
172            mj_outs_push(out, mj_out_send(from, mj_frame_pull(f->lid)));
173            return;
174        }
175        l->responded |= (uint32_t)1 << from;
176        for (size_t i = 0; i < f->nheld; i++) {
177            if (l->nadopt == l->capadopt) {
178                l->capadopt = l->capadopt ? 2 * l->capadopt : 8;
179                l->adopt = mxrealloc(l->adopt, l->capadopt * sizeof *l->adopt);
180            }
181            l->adopt[l->nadopt++] = (mj_adopt){from, f->held[i]};
182            f->held[i] = (mj_held){0};
183        }
184        uint32_t cnt = 0;
185        for (uint32_t k = 0; k < 32; k++) cnt += l->responded >> k & 1;
186        if (cnt >= quorum) mj_lead_activate(n, now, unix_ms, out);
187        return;
188    }
189    if (a->spoilt || a->seq != f->seq) {
190        a->spoilt = true;
191        return;
192    }
193    a->seq++;
194    if (f->kind == MESH_JOBS_HAVE_COMMITTED) {
195        a->commit_lid = f->src0;
196        a->commit_n = f->src1;
197        for (size_t i = 0; i < f->ops.n; i++) mj_table_apply(&a->committed, &f->ops.v[i]);
198    } else {
199        mj_ops *e = mj_answer_entry(a, f->src0, f->src1);
200        mj_ops_append(e, &f->ops);
201    }
202}
203
204static void on_state(mj_node *n, uint32_t from, mj_frame *f, uint64_t now, mj_outs *out) {
205    uint64_t lid = f->lid;
206    switch (f->kind) {
207    case MESH_JOBS_ST_PART:
208        /* Numbered: part 0 starts the table; any other must follow the last (review finding 9). */
209        if (f->seq == 0) {
210            mj_table_free(&n->staging);
211            mj_table_init(&n->staging);
212            for (size_t i = 0; i < f->ops.n; i++) mj_table_apply(&n->staging, &f->ops.v[i]);
213            n->has_staging = true;
214            n->staging_n = f->n;
215            n->staging_seq = 1;
216        } else if (n->has_staging && n->staging_n == f->n && n->staging_seq == f->seq) {
217            for (size_t i = 0; i < f->ops.n; i++) mj_table_apply(&n->staging, &f->ops.v[i]);
218            n->staging_seq++;
219        } else n->has_staging = false;
220        break;
221    case MESH_JOBS_ST_SNAP_END:
222    case MESH_JOBS_ST_REPLACE_END: {
223        bool whole = n->has_staging && n->staging_n == f->n && n->staging_seq == f->seq;
224        n->has_staging = false;
225        if (whole && f->kind == MESH_JOBS_ST_SNAP_END && f->n == 0) {
226            /* A snapshot at n = 0 would be a leader's table from before it led: never installed. */
227        } else if (whole && f->kind == MESH_JOBS_ST_SNAP_END) {
228            /* The committed table at the leader's commit point n; the entries of this leadership past it that
229             * this member has stay (Replica::apply). */
230            mj_ops ops = {0};
231            mj_table_ops(&n->staging, &ops);
232            mj_node_log(n, mj_log_snapshot(lid, f->n, ops), out);
233            uint64_t last = mj_replica_last(&n->rep, lid);
234            n->next = (last > f->n ? last : f->n) + 1;
235            mj_node_send(n, from, mj_frame_ack(lid, n->next - 1, false), out);
236        } else if (whole && f->kind == MESH_JOBS_ST_REPLACE_END) {
237            if (f->n == n->next || (n->next == 0 && f->n == 1)) {
238                mj_ops ops = {0};
239                mj_table_ops(&n->staging, &ops);
240                mj_node_log(n, mj_log_entry(lid, f->n, true, ops), out);
241                n->next = f->n + 1;
242                mj_node_send(n, from, mj_frame_ack(lid, f->n, false), out);
243            } else if (f->n > n->next) gap(n, from, lid, now, out);
244        } else {
245            /* A part was lost: the table is not installed (a gap: the leader sends it again). */
246            mj_outs_push(out, mj_out_log(mfmt("jobs: a table from system %u arrived incomplete; asking for it again", from)));
247            gap(n, from, lid, now, out);
248        }
249        break;
250    }
251    case MESH_JOBS_ST_DELTA:
252        if (f->n == n->next && f->n > 0) {
253            mj_node_log(n, mj_log_entry(lid, f->n, false, f->ops), out);
254            f->ops = (mj_ops){0};
255            n->next = f->n + 1;
256            mj_node_send(n, from, mj_frame_ack(lid, f->n, false), out);
257        } else if (f->n > n->next) gap(n, from, lid, now, out);
258        break;
259    case MESH_JOBS_ST_COMMIT:
260        /* A commit point past the entries here: some were lost on the way: a gap. */
261        if (f->commit >= (n->next > 1 ? n->next : 1)) gap(n, from, lid, now, out);
262        break;
263    default: break;
264    }
265    /* The leader's commit point: what it covers of this leadership's entries here is committed. */
266    uint64_t top = n->next ? n->next - 1 : 0;
267    uint64_t c = f->commit < top ? f->commit : top;
268    if (n->rep.npending && n->rep.pending[0].lid == lid && n->rep.pending[0].n <= c) mj_node_log(n, mj_log_commit(lid, c), out);
269}
270
271void mj_node_on_frame(mj_node *n, uint32_t from, mj_frame f, uint64_t now, uint64_t unix_ms, mj_outs *out) {
272    uint32_t me = n->cfg.me;
273    switch (f.ty) {
274    case MESH_F_JOB_AVAIL:
275        if (n->lead) {
276            mj_lead *l = n->lead;
277            if (!(from < MJ_MEMBERS && l->disp.progs.has[from] && mj_progs_eq(l->disp.progs.v[from], l->disp.progs.n[from], f.progs, f.nprogs))) {
278                msb b = {0};
279                if (!f.nprogs) msb_puts(&b, "no job program");
280                for (size_t i = 0; i < f.nprogs; i++) msb_printf(&b, "%s%s", i ? ", " : "", f.progs[i].name);
281                char *names = msb_take(&b);
282                mj_outs_push(out, mj_out_log(mfmt("jobs: system %u runs %s", from, names)));
283                free(names);
284                if (from < MJ_MEMBERS) {
285                    mj_member_progs_set(&l->disp.progs, from, f.progs, f.nprogs);
286                    f.progs = NULL;
287                    f.nprogs = 0;
288                }
289            }
290        }
291        break;
292    case MESH_F_RESULT:
293        if (mj_node_active(n)) {
294            mj_node_on_results(n, from, f.items, f.nitems, now, unix_ms, out);
295            f.items = NULL;
296            f.nitems = 0;
297        }
298        break;
299    case MESH_F_JOB_ACK:
300        if (n->lead && n->lead->lid == f.lid && n->lead->active && from < n->cfg.n) {
301            mj_lead *l = n->lead;
302            /* A gap: it gets a snapshot; what it says it has is all it has (review finding 6). */
303            if (f.gap) {
304                l->snap_due |= (uint32_t)1 << from;
305                if (f.n < l->acked[from]) l->acked[from] = f.n;
306            } else if (f.n > l->acked[from]) l->acked[from] = f.n;
307            mj_node_commit_check(n, out);
308        }
309        break;
310    case MESH_F_JOB_HAVE: on_have(n, from, &f, now, unix_ms, out); break;
311    case MESH_F_JOB_PULL:
312        if (follow(n, from, f.lid, out)) {
313            mj_frame *v;
314            size_t cnt;
315            mj_node_answer(n, f.lid, &v, &cnt);
316            for (size_t i = 0; i < cnt; i++) mj_node_send(n, from, v[i], out);
317            free(v);
318        }
319        break;
320    case MESH_F_JOB_STATE:
321        if (from != me && follow(n, from, f.lid, out)) on_state(n, from, &f, now, out);
322        break;
323    case MESH_F_LEASE:
324        if (f.lid < n->rep.lid && from != me) break;
325        {
326            bool fresh;
327            mj_held *h = momap_put1(n->held, f.l.lease, &fresh);
328            if (!fresh) mj_held_free(h);
329            *h = (mj_held){f.l.lease, f.l.job, mstrdup(f.l.program), mj_ranges_clone(&f.l.items)};
330            n->avail_dirty = true;
331            mj_outs_push(out, mj_out_run(f.l));
332            f.l = (mj_lease_msg){0};
333            f.ty = 0;
334        }
335        break;
336    case MESH_F_LEASE_CUT:
337        if (f.lid < n->rep.lid && from != me) break;
338        {
339            mj_held *h = momap_get1(n->held, f.lease);
340            if (h) {
341                mj_ranges_remove(&h->items, f.cut, UINT64_MAX);
342                if (mj_ranges_is_empty(&h->items)) momap_del1(n->held, f.lease);
343                n->avail_dirty = true;
344            }
345            mj_outs_push(out, mj_out_cut(f.lease, f.cut));
346        }
347        break;
348    default: break;
349    }
350    mj_frame_free(&f);
351}