==> src/node.c <==1/* jobs.rs's Node: construction, the inputs (set_view, recv, result, blob_*, set_progs, request,
2 * tick) and the member side (stage). The leader's side is in lead.c, frames in frames.c. */
3#include "internal.h"
4#include "mesh/abi/peer_frames.h"
5#include "mesh/jobs/copies.h"
6#include "mesh/jobs/lid.h"
7
8#include <stdio.h>
9#include <stdlib.h>
10#include <string.h>
11
12char *mj_word(mj_job_id id) {
13 char w[MJ_JOB_WORD_MAX];
14 mj_job_word(id, w);
15 return mstrdup(w);
16}
17
18/* ---- (member, program) maps -------------------------------------------------------------------- */
19
20static int mp_cmp(uint32_t m, const char *p, const mj_mp *x) {
21 if (m != x->member) return m < x->member ? -1 : 1;
22 return strcmp(p, x->program);
23}
24
25mj_mp *mj_mpmap_get(const mj_mpmap *m, uint32_t member, const char *program) {
26 for (size_t i = 0; i < m->n; i++)
27 if (mp_cmp(member, program, &m->v[i]) == 0) return &m->v[i];
28 return NULL;
29}
30
31mj_mp *mj_mpmap_put(mj_mpmap *m, uint32_t member, const char *program) {
32 size_t at = 0;
33 while (at < m->n) {
34 int c = mp_cmp(member, program, &m->v[at]);
35 if (c == 0) return &m->v[at];
36 if (c < 0) break;
37 at++;
38 }
39 if (m->n == m->cap) {
40 m->cap = m->cap ? 2 * m->cap : 8;
41 m->v = mxrealloc(m->v, m->cap * sizeof *m->v);
42 }
43 memmove(&m->v[at + 1], &m->v[at], (m->n - at) * sizeof *m->v);
44 m->v[at] = (mj_mp){.member = member, .program = mstrdup(program)};
45 m->n++;
46 return &m->v[at];
47}
48
49static void mpmap_free(mj_mpmap *m) {
50 for (size_t i = 0; i < m->n; i++) free(m->v[i].program);
51 free(m->v);
52 *m = (mj_mpmap){0};
53}
54
55/* ---- map value releases ------------------------------------------------------------------------ */
56
57static void lease_free(void *v) {
58 mj_lease *x = v;
59 free(x->program);
60 mj_ranges_free(&x->out);
61}
62static void queue_free(void *v) {
63 mj_queue *q = v;
64 mj_ranges_free(&q->fresh);
65 mj_ranges_free(&q->retry);
66}
67static void picks_free(void *v) { mj_picks_free(v); }
68static void parked_free(void *v) {
69 mj_parked_list *p = v;
70 for (size_t i = 0; i < p->n; i++) mj_item_res_free(&p->v[i].r);
71 free(p->v);
72}
73static void held_free(void *v) { mj_held_free(v); }
74static void outbox_free(void *v) { mj_item_res_free(&((mj_outbox_entry *)v)->r); }
75static void staged_free(void *v) { mj_item_res_free(&((mj_staged *)v)->r); }
76
77mj_lead *mj_lead_new(uint32_t n);
78void mj_lead_free(mj_lead *l);
79
80mj_lead *mj_lead_new(uint32_t n) {
81 mj_lead *l = mxalloc(sizeof *l);
82 mj_table_init(&l->working);
83 l->acked = mxalloc((n ? n : 1) * sizeof *l->acked);
84 l->disp.leases = momap_new(1, sizeof(mj_lease), lease_free, NULL);
85 l->disp.queues = momap_new(1, sizeof(mj_queue), queue_free, NULL);
86 l->disp.picks = momap_new(1, sizeof(mj_picks), picks_free, NULL);
87 l->dupes = momap_new(1, sizeof(uint64_t), NULL, NULL);
88 l->parked = momap_new(5, sizeof(mj_parked_list), parked_free, NULL);
89 l->fetch_at = momap_new(5, sizeof(uint64_t), NULL, NULL);
90 return l;
91}
92
93void mj_lead_free(mj_lead *l) {
94 if (!l) return;
95 for (int k = 0; k < MJ_MEMBERS; k++)
96 if (l->answers[k]) {
97 mj_answer_free(l->answers[k]);
98 free(l->answers[k]);
99 }
100 for (size_t i = 0; i < l->nadopt; i++) mj_held_free(&l->adopt[i].h);
101 free(l->adopt);
102 mj_table_free(&l->working);
103 free(l->acked);
104 mj_ops_free(&l->ops);
105 for (size_t i = 0; i < l->nwaits; i++) free(l->waits[i].text);
106 free(l->waits);
107 momap_free(l->disp.leases);
108 momap_free(l->disp.queues);
109 mj_member_progs_free(&l->disp.progs);
110 mpmap_free(&l->disp.rates);
111 mpmap_free(&l->disp.holdoff);
112 momap_free(l->disp.picks);
113 momap_free(l->dupes);
114 for (size_t i = 0; i < l->nheld; i++) mj_submit_free(&l->held[i].s);
115 free(l->held);
116 momap_free(l->parked);
117 momap_free(l->fetch_at);
118 free(l);
119}
120
121/* ---- the node ---------------------------------------------------------------------------------- */
122
123mj_node *mj_node_new(mj_cfg cfg, mj_replica rep) {
124 mj_node *n = mxalloc(sizeof *n);
125 n->cfg = cfg;
126 n->rep = rep;
127 mj_table_init(&n->staging);
128 n->held = momap_new(1, sizeof(mj_held), held_free, NULL);
129 n->outbox = momap_new(3, sizeof(mj_outbox_entry), outbox_free, NULL);
130 n->avail_dirty = true;
131 n->staged = momap_new(3, sizeof(mj_staged), staged_free, NULL);
132 n->have = momap_new(5, 1, NULL, NULL);
133 return n;
134}
135
136static void sched_free(mj_sched *s) {
137 if (!s) return;
138 if (s->release) s->release(s->ctx);
139 free(s);
140}
141
142void mj_node_free(mj_node *n) {
143 if (!n) return;
144 mj_replica_free(&n->rep);
145 mj_table_free(&n->staging);
146 mj_lead_free(n->lead);
147 momap_free(n->held);
148 momap_free(n->outbox);
149 mj_progs_free(n->progs, n->nprogs);
150 for (size_t i = n->lo_head; i < n->nlo; i++) mj_frame_free(&n->lo[i]);
151 free(n->lo);
152 sched_free(n->sched);
153 momap_free(n->staged);
154 momap_free(n->have);
155 free(n);
156}
157
158const mj_table *mj_node_committed(const mj_node *n) { return &n->rep.committed; }
159
160void mj_node_set_sched(mj_node *n, mj_sched *cx) {
161 sched_free(n->sched);
162 n->sched = cx;
163}
164
165const mj_picks *mj_node_picks(const mj_node *n, mj_job_id job) { return n->lead ? momap_get1(n->lead->disp.picks, job) : NULL; }
166bool mj_node_leading(const mj_node *n) { return n->lead != NULL; }
167bool mj_node_active(const mj_node *n) { return n->lead && n->lead->active; }
168uint64_t mj_node_lid(const mj_node *n) { return n->rep.lid; }
169uint64_t mj_node_commit_n(const mj_node *n) { return n->lead ? n->lead->commit : n->rep.commit_n; }
170const mj_table *mj_node_working(const mj_node *n) { return n->lead && n->lead->active ? &n->lead->working : NULL; }
171const momap *mj_node_leases(const mj_node *n) { return n->lead ? n->lead->disp.leases : NULL; }
172const mj_mpmap *mj_node_rates(const mj_node *n) { return n->lead ? &n->lead->disp.rates : NULL; }
173
174bool mj_node_queued(const mj_node *n, mj_job_id job, uint64_t *count) {
175 if (!n->lead) return false;
176 const mj_queue *q = momap_get1(n->lead->disp.queues, job);
177 if (!q) return false;
178 uint64_t a = mj_ranges_len(&q->fresh), b = mj_ranges_len(&q->retry);
179 *count = a > UINT64_MAX - b ? UINT64_MAX : a + b;
180 return true;
181}
182
183double mj_node_my_rate(const mj_node *n) { return n->rate.ips; }
184size_t mj_node_unconfirmed(const mj_node *n) { return momap_len(n->outbox); }
185size_t mj_node_staged(const mj_node *n) { return momap_len(n->staged); }
186
187size_t mj_node_parked(const mj_node *n) {
188 if (!n->lead) return 0;
189 size_t total = 0;
190 uint64_t k[5];
191 momap_iter cur4_;
192 momap_iter_init(&cur4_, n->lead->parked);
193 const uint64_t *cur4_k;
194 for (const mj_parked_list *p = momap_iter_next(&cur4_, &cur4_k); p && (memcpy(k, cur4_k, sizeof k), 1); p = momap_iter_next(&cur4_, &cur4_k)) total += p->n;
195 return total;
196}
197
198bool mj_node_has_blob(const mj_node *n, mj_job_id job, const mj_hash *h) {
199 uint64_t k[5];
200 mj_key5(k, job, h);
201 return momap_get(n->have, k) != NULL;
202}
203
204void mj_node_send(mj_node *n, uint32_t k, mj_frame f, mj_outs *out) {
205 if (k == n->cfg.me) {
206 if (n->nlo == n->caplo) {
207 if (n->lo_head > 0) {
208 memmove(n->lo, n->lo + n->lo_head, (n->nlo - n->lo_head) * sizeof *n->lo);
209 n->nlo -= n->lo_head;
210 n->lo_head = 0;
211 }
212 if (n->nlo == n->caplo) {
213 n->caplo = n->caplo ? 2 * n->caplo : 16;
214 n->lo = mxrealloc(n->lo, n->caplo * sizeof *n->lo);
215 }
216 }
217 n->lo[n->nlo++] = f;
218 } else mj_outs_push(out, mj_out_send(k, f));
219}
220
221void mj_node_log(mj_node *n, mj_log l, mj_outs *out) {
222 mj_replica_apply(&n->rep, &l);
223 mj_outs_push(out, mj_out_persist(l));
224}
225
226void mj_node_pump(mj_node *n, uint64_t now, uint64_t unix_ms, mj_outs *out) {
227 while (n->lo_head < n->nlo) {
228 mj_frame f = n->lo[n->lo_head++];
229 if (n->lo_head == n->nlo) n->lo_head = n->nlo = 0;
230 mj_node_on_frame(n, n->cfg.me, f, now, unix_ms, out);
231 }
232}
233
234/* ---- inputs ------------------------------------------------------------------------------------ */
235
236mj_outs mj_node_set_view(mj_node *n, mj_view v, uint64_t now, uint64_t unix_ms) {
237 mj_outs out = {0};
238 mj_view old = n->view;
239 n->view = v;
240 uint32_t me = n->cfg.me;
241 bool lead_me = v.has_leader && v.leader == me;
242 if (n->lead && (!lead_me || n->lead->term != v.term)) {
243 char who[16];
244 if (v.has_leader) snprintf(who, sizeof who, "%u", v.leader);
245 else strcpy(who, "-");
246 mj_outs_push(&out, mj_out_log(mfmt("jobs: system %u no longer leads the job table (fleet leader %s, term %llu)", me, who,
247 (unsigned long long)v.term)));
248 mj_outs o = mj_node_step_down(n);
249 mj_outs_append(&out, &o);
250 }
251 if (lead_me && !n->lead) mj_node_start_lead(n, v.term, now, unix_ms, &out);
252 if (n->lead) {
253 mj_lead *l = n->lead;
254 /* A member linked since: it gets a snapshot (once this leadership is active). */
255 uint32_t fresh = v.linked & ~old.linked & ~((uint32_t)1 << me);
256 l->snap_due |= fresh;
257 /* Pull again from members newly linked while pulling. */
258 if (!l->active)
259 for (uint32_t k = 0; k < 32; k++)
260 if (fresh >> k & 1) mj_outs_push(&out, mj_out_send(k, mj_frame_pull(l->lid)));
261 }
262 bool same_leader = v.has_leader == old.has_leader && (!v.has_leader || v.leader == old.leader);
263 if (!same_leader) {
264 n->avail_dirty = true;
265 uint64_t k[3];
266 momap_iter cur5_;
267 momap_iter_init(&cur5_, n->outbox);
268 const uint64_t *cur5_k;
269 for (mj_outbox_entry *e = momap_iter_next(&cur5_, &cur5_k); e && (memcpy(k, cur5_k, sizeof k), 1); e = momap_iter_next(&cur5_, &cur5_k)) e->at = 0;
270 }
271 mj_node_pump(n, now, unix_ms, &out);
272 return out;
273}
274
275mj_outs mj_node_recv(mj_node *n, uint32_t from, mj_frame f, uint64_t now, uint64_t unix_ms) {
276 mj_outs out = {0};
277 mj_node_on_frame(n, from, f, now, unix_ms, &out);
278 mj_node_pump(n, now, unix_ms, &out);
279 return out;
280}
281
282mj_outs mj_node_result(mj_node *n, mj_item_res r, uint64_t now) {
283 mj_held *h = momap_get1(n->held, r.lease);
284 if (h) {
285 mj_ranges_take(&h->items, r.item);
286 if (mj_ranges_is_empty(&h->items)) {
287 momap_del1(n->held, r.lease);
288 n->avail_dirty = true;
289 }
290 }
291 if (r.status != MESH_JOBS_RS_RETURNED) mj_rate_count(&n->rate, 1, now);
292 uint64_t key[3];
293 mj_key3(key, r.job, r.item, r.lease);
294 if (r.has_blob && r.status == MESH_JOBS_RS_OK) {
295 uint32_t acked = mj_node_has_blob(n, r.job, &r.blob.hash) ? (uint32_t)1 << n->cfg.me : 0;
296 momap_del(n->staged, key);
297 mj_staged *s = momap_put(n->staged, key, NULL);
298 s->r = r;
299 s->acked = acked;
300 } else {
301 momap_del(n->outbox, key);
302 mj_outbox_entry *e = momap_put(n->outbox, key, NULL);
303 e->r = r;
304 e->at = 0;
305 }
306 return (mj_outs){0};
307}
308
309static bool staged_for(const mj_staged *s, mj_job_id job, const mj_hash *h) {
310 return s->r.job == job && s->r.has_blob && mj_hash_eq(&s->r.blob.hash, h);
311}
312
313mj_outs mj_node_blob_stored(mj_node *n, mj_job_id job, mj_hash hash, uint64_t now, uint64_t unix_ms) {
314 mj_outs out = {0};
315 uint64_t hk[5];
316 mj_key5(hk, job, &hash);
317 momap_put(n->have, hk, NULL);
318 uint32_t me = n->cfg.me;
319 uint64_t k[3];
320 momap_iter cur6_;
321 momap_iter_init(&cur6_, n->staged);
322 const uint64_t *cur6_k;
323 for (mj_staged *s = momap_iter_next(&cur6_, &cur6_k); s && (memcpy(k, cur6_k, sizeof k), 1); s = momap_iter_next(&cur6_, &cur6_k))
324 if (staged_for(s, job, &hash)) s->acked |= (uint32_t)1 << me;
325 if (n->lead && n->lead->active) {
326 mj_lead *l = n->lead;
327 momap_del(l->fetch_at, hk);
328 mj_parked_list v = {0};
329 if (momap_take(l->parked, hk, &v)) {
330 /* Grouped by the member it came from, in member order, each in arrival order. */
331 for (uint32_t from = 0; from < 32; from++) {
332 size_t cnt = 0;
333 for (size_t i = 0; i < v.n; i++) cnt += v.v[i].from == from;
334 if (!cnt) continue;
335 mj_item_res *items = mxalloc(cnt * sizeof *items);
336 size_t j = 0;
337 for (size_t i = 0; i < v.n; i++)
338 if (v.v[i].from == from) {
339 items[j++] = v.v[i].r;
340 v.v[i].r = (mj_item_res){0};
341 }
342 mj_node_on_results(n, from, items, cnt, now, unix_ms, &out);
343 }
344 parked_free(&v);
345 }
346 }
347 mj_node_pump(n, now, unix_ms, &out);
348 return out;
349}
350
351void mj_node_blob_gone(mj_node *n, mj_job_id job, const mj_hash *hash) {
352 uint64_t k[5];
353 mj_key5(k, job, hash);
354 momap_del(n->have, k);
355}
356
357void mj_node_blob_acked(mj_node *n, uint32_t from, mj_job_id job, mj_hash hash, bool ok) {
358 uint64_t k[3];
359 momap_iter cur7_;
360 momap_iter_init(&cur7_, n->staged);
361 const uint64_t *cur7_k;
362 for (mj_staged *s = momap_iter_next(&cur7_, &cur7_k); s && (memcpy(k, cur7_k, sizeof k), 1); s = momap_iter_next(&cur7_, &cur7_k))
363 if (staged_for(s, job, &hash)) {
364 if (ok) s->acked |= (uint32_t)1 << from;
365 else s->refused |= (uint32_t)1 << from;
366 }
367}
368
369void mj_node_set_progs(mj_node *n, mj_prog *progs, size_t count) {
370 if (mj_progs_eq(progs, count, n->progs, n->nprogs)) {
371 mj_progs_free(progs, count);
372 return;
373 }
374 mj_progs_free(n->progs, n->nprogs);
375 n->progs = progs;
376 n->nprogs = count;
377 n->avail_dirty = true;
378}
379
380/* Members (alive & allowed) whose programs include one of the submit's. */
381static uint32_t runs_mask(const mj_lead *l, const mj_submit *s, uint32_t alive) {
382 uint32_t allowed = s->members == 0 ? UINT32_MAX : s->members, mask = 0;
383 mj_spec probe = {.program = s->program};
384 for (uint32_t k = 0; k < MJ_MEMBERS; k++) {
385 if (!l->disp.progs.has[k] || ((alive & allowed) >> k & 1) == 0) continue;
386 for (size_t i = 0; i < l->disp.progs.n[k]; i++)
387 if (mj_spec_runs(&probe, l->disp.progs.v[k][i].name)) {
388 mask |= (uint32_t)1 << k;
389 break;
390 }
391 }
392 return mask;
393}
394
395static void push_wait(mj_lead *l, uint64_t n, uint64_t token, bool ok, char *text) {
396 if (l->nwaits == l->capwaits) {
397 l->capwaits = l->capwaits ? 2 * l->capwaits : 8;
398 l->waits = mxrealloc(l->waits, l->capwaits * sizeof *l->waits);
399 }
400 l->waits[l->nwaits++] = (mj_wait){n, token, ok, text};
401}
402
403static mj_outs request_in(mj_node *n, mj_req req, uint64_t token, uint64_t now, uint64_t unix_ms, bool may_wait) {
404 mj_outs out = {0};
405 uint32_t me = n->cfg.me;
406 uint32_t alive = n->view.alive | (uint32_t)1 << me;
407 mj_lead *l = n->lead && n->lead->active ? n->lead : NULL;
408 if (!l) {
409 mj_outs_push(&out, mj_out_answer(token, false,
410 mstrdup(n->lead ? "jobs: the new leader is still merging the members' job tables; try again in a moment"
411 : "jobs: this member does not lead the job table")));
412 mj_req_free(&req);
413 return out;
414 }
415 bool ok = false;
416 char *text = NULL;
417 switch (req.kind) {
418 case MJ_REQ_SUBMIT: {
419 mj_submit *s = &req.submit;
420 uint32_t runs = runs_mask(l, s, alive);
421 if (!runs && may_wait) {
422 /* Held until a member reports the program (or SUBMIT_WAIT_MS): tick retries it. */
423 if (l->nheld == l->capheld) {
424 l->capheld = l->capheld ? 2 * l->capheld : 4;
425 l->held = mxrealloc(l->held, l->capheld * sizeof *l->held);
426 }
427 l->held[l->nheld++] = (mj_held_submit){*s, token, now + MESH_JOBS_SUBMIT_WAIT_MS};
428 return out;
429 }
430 if (!runs) {
431 text = mfmt("no member runs %s (a loaded, running program with policy \"job\") after %u s; `jobs` shows who runs what", s->program,
432 MESH_JOBS_SUBMIT_WAIT_MS / 1000);
433 break;
434 }
435 mj_job_id id = mj_table_next_id(&l->working, l->term);
436 mj_spec spec = {.id = id, .program = mstrdup(s->program), .op = s->op, .items = mj_ranges_clone(&s->items), .priority = s->priority,
437 .attempts = s->attempts, .deadline_ms = s->deadline_s > 0 ? unix_ms + s->deadline_s * 1000 : 0, .mode = s->mode,
438 .setup = mj_setup_clone(s->setup, s->nsetup), .nsetup = s->nsetup, .submitted_ms = unix_ms, .members = s->members};
439 mj_op op = mj_op_submit(&spec);
440 mj_table_apply(&l->working, &op);
441 mj_ops_push(&l->ops, op);
442 bool fresh;
443 mj_queue *q = momap_put1(l->disp.queues, id, &fresh);
444 if (!fresh) queue_free(q);
445 *q = (mj_queue){mj_ranges_clone(&s->items), {0}};
446 msb b = {0};
447 char *w = mj_word(id);
448 msb_printf(&b, "job %s program %s items ", w, s->program);
449 free(w);
450 if (mj_spec_endless(&spec)) msb_puts(&b, "endless");
451 else msb_printf(&b, "%llu", (unsigned long long)mj_ranges_len(&s->items));
452 uint32_t cnt = 0;
453 for (uint32_t k = 0; k < 32; k++) cnt += runs >> k & 1;
454 msb_printf(&b, " on %u member(s) (", cnt);
455 bool first = true;
456 for (uint32_t k = 0; k < 32; k++)
457 if (runs >> k & 1) {
458 msb_printf(&b, "%s%u", first ? "" : ",", k);
459 first = false;
460 }
461 msb_putc(&b, ')');
462 mj_spec_free(&spec);
463 ok = true;
464 text = msb_take(&b);
465 break;
466 }
467 case MJ_REQ_CANCEL: {
468 mj_job *j = mj_table_get(&l->working, req.job);
469 char *w = mj_word(req.job);
470 if (!j) text = mfmt("no job %s", w);
471 else if (!mj_job_active(j)) text = mfmt("job %s is %s already", w, mj_job_state(j));
472 else {
473 mj_lead_cancel(l, req.job, "cancelled", MESH_JOBS_FIN_CANCELLED, unix_ms, &out);
474 ok = true;
475 text = mfmt("job %s cancelled", w);
476 }
477 free(w);
478 break;
479 }
480 case MJ_REQ_TONE: {
481 mj_job *j = mj_table_get(&l->working, req.job);
482 char *w = mj_word(req.job);
483 if (!j) text = mfmt("no job %s", w);
484 else {
485 mj_op op = mj_op_tone(req.job, j->tone_seq + 1, req.tone);
486 mj_table_apply(&l->working, &op);
487 mj_ops_push(&l->ops, op);
488 ok = true;
489 text = mfmt("job %s tone %u", w, req.tone);
490 }
491 free(w);
492 break;
493 }
494 case MJ_REQ_FORGET: {
495 mj_job *j = mj_table_get(&l->working, req.job);
496 char *w = mj_word(req.job);
497 if (!j) text = mfmt("no job %s", w);
498 else if (!j->finished) text = mfmt("job %s is %s: cancel it first", w, mj_job_state(j));
499 else {
500 mj_op op = mj_op_forget(req.job);
501 mj_table_apply(&l->working, &op);
502 mj_ops_push(&l->ops, op);
503 momap_del1(l->disp.queues, req.job);
504 uint64_t k[5];
505 for (mj_parked_list *p = momap_first(l->parked, k); p; p = momap_next(l->parked, k, k)) {
506 size_t keep = 0;
507 for (size_t i = 0; i < p->n; i++) {
508 if (p->v[i].r.job != req.job) p->v[keep++] = p->v[i];
509 else mj_item_res_free(&p->v[i].r);
510 }
511 p->n = keep;
512 if (!keep) momap_del(l->parked, k);
513 }
514 ok = true;
515 text = mfmt("job %s forgotten", w);
516 }
517 free(w);
518 break;
519 }
520 }
521 mj_req_free(&req);
522 if (!ok) mj_outs_push(&out, mj_out_answer(token, false, text));
523 else {
524 mj_node_flush(n, &out);
525 push_wait(n->lead, n->lead->n, token, true, text);
526 mj_node_commit_check(n, &out);
527 }
528 mj_node_pump(n, now, unix_ms, &out);
529 return out;
530}
531
532mj_outs mj_node_request(mj_node *n, mj_req req, uint64_t token, uint64_t now, uint64_t unix_ms) {
533 return request_in(n, req, token, now, unix_ms, true);
534}
535
536/* The member side of payload results: each staged result's copies (`copies_for`): pushes to the
537 * members that have not acknowledged one (again after PUSH_RESEND_MS), and the result to the outbox
538 * once every copy is durable (its holders recorded). */
539static void stage(mj_node *n, uint64_t now, mj_outs *out) {
540 uint32_t me = n->cfg.me, cnt = n->cfg.n;
541 uint32_t alive = n->view.alive | (uint32_t)1 << me;
542 uint64_t k[3];
543 uint64_t (*ready)[3] = NULL;
544 size_t nready = 0;
545 momap_iter cur8_;
546 momap_iter_init(&cur8_, n->staged);
547 const uint64_t *cur8_k;
548 for (mj_staged *s = momap_iter_next(&cur8_, &cur8_k); s && (memcpy(k, cur8_k, sizeof k), 1); s = momap_iter_next(&cur8_, &cur8_k)) {
549 if (!s->r.has_blob) continue;
550 uint32_t want;
551 if (!mj_copies_for(me, cnt, n->view.has_leader, n->view.leader, alive, s->refused, &want)) continue;
552 if ((want & ~s->acked) == 0) {
553 ready = mxrealloc(ready, (nready + 1) * sizeof *ready);
554 memcpy(ready[nready++], k, sizeof k);
555 continue;
556 }
557 uint32_t to = want & ~s->acked & ~((uint32_t)1 << me);
558 for (uint32_t m = 0; m < 32; m++) {
559 if (!(to >> m & 1)) continue;
560 if ((s->sent_mask >> m & 1) && (now > s->sent[m] ? now - s->sent[m] : 0) < MESH_JOBS_PUSH_RESEND_MS) continue;
561 s->sent_mask |= (uint32_t)1 << m;
562 s->sent[m] = now > 1 ? now : 1;
563 mj_outs_push(out, mj_out_push_blob(m, s->r.job, s->r.blob.hash));
564 }
565 }
566 for (size_t i = 0; i < nready; i++) {
567 mj_staged s;
568 momap_take(n->staged, ready[i], &s);
569 mj_item_res r = s.r;
570 if (r.has_blob) {
571 r.blob.holders = s.acked;
572 mj_outs_push(out, mj_out_blob_done(r.job, r.blob.hash));
573 }
574 momap_del(n->outbox, ready[i]);
575 mj_outbox_entry *e = momap_put(n->outbox, ready[i], NULL);
576 e->r = r;
577 e->at = 0;
578 }
579 free(ready);
580}
581
582mj_outs mj_node_tick(mj_node *n, uint64_t now, uint64_t unix_ms) {
583 mj_outs out = {0};
584 uint32_t me = n->cfg.me;
585 if (n->lead && !n->lead->active && (now > n->lead->pull_at ? now - n->lead->pull_at : 0) >= 1000) {
586 mj_lead *l = n->lead;
587 l->pull_at = now;
588 uint32_t ask = n->view.linked & ~l->responded & ~((uint32_t)1 << me);
589 for (uint32_t k = 0; k < 32; k++) {
590 if (!(ask >> k & 1)) continue;
591 if ((l->heard_mask >> k & 1) && (now > l->heard[k] ? now - l->heard[k] : 0) < 3000) continue;
592 mj_outs_push(&out, mj_out_send(k, mj_frame_pull(l->lid)));
593 }
594 }
595 if (mj_node_active(n)) {
596 /* Held submits: a member reports the program now, or their wait is over. */
597 uint32_t alive = n->view.alive | (uint32_t)1 << me;
598 mj_lead *l = n->lead;
599 mj_held_submit *go = NULL;
600 size_t ngo = 0, keep = 0;
601 for (size_t i = 0; i < l->nheld; i++) {
602 mj_held_submit *h = &l->held[i];
603 if (now >= h->until || runs_mask(l, &h->s, alive)) {
604 go = mxrealloc(go, (ngo + 1) * sizeof *go);
605 go[ngo++] = *h;
606 } else l->held[keep++] = *h;
607 }
608 l->nheld = keep;
609 for (size_t i = 0; i < ngo; i++) {
610 mj_req r = {.kind = MJ_REQ_SUBMIT, .submit = go[i].s};
611 mj_outs o = request_in(n, r, go[i].token, now, unix_ms, false);
612 mj_outs_append(&out, &o);
613 }
614 free(go);
615 mj_node_dispatch(n, now, unix_ms, &out);
616 mj_node_flush(n, &out);
617 mj_node_snapshots(n, &out);
618 mj_node_commit_check(n, &out);
619 }
620 /* The member side: availability, results, pruning. */
621 if (n->view.has_leader) {
622 uint32_t leader = n->view.leader;
623 if (n->avail_dirty || (now > n->avail_at ? now - n->avail_at : 0) >= MESH_JOBS_AVAIL_MS) {
624 n->avail_dirty = false;
625 n->avail_at = now;
626 size_t nh = momap_len(n->held), i = 0;
627 mj_held *held = nh ? mxalloc(nh * sizeof *held) : NULL;
628 uint64_t k;
629 momap_iter cur9_;
630 momap_iter_init(&cur9_, n->held);
631 for (const mj_held *h = momap_iter_next1(&cur9_, &k); h; h = momap_iter_next1(&cur9_, &k)) held[i++] = mj_held_clone(h);
632 mj_node_send(n, leader, mj_frame_avail(mj_node_lid(n), mj_progs_clone(n->progs, n->nprogs), n->nprogs, held, nh), &out);
633 }
634 stage(n, now, &out);
635 /* Results due: never sent, or not confirmed for RESEND_MS; in frames of 4096. */
636 uint64_t k[3];
637 mj_item_res *batch = NULL;
638 size_t nb = 0;
639 momap_iter cur10_;
640 momap_iter_init(&cur10_, n->outbox);
641 const uint64_t *cur10_k;
642 for (mj_outbox_entry *e = momap_iter_next(&cur10_, &cur10_k); e && (memcpy(k, cur10_k, sizeof k), 1); e = momap_iter_next(&cur10_, &cur10_k)) {
643 if (!(e->at == 0 || (now > e->at ? now - e->at : 0) >= MESH_JOBS_RESEND_MS)) continue;
644 e->at = now > 1 ? now : 1;
645 if (!batch) batch = mxalloc(4096 * sizeof *batch);
646 batch[nb++] = mj_item_res_clone(&e->r);
647 if (nb == 4096) {
648 mj_node_send(n, leader, mj_frame_result(mj_node_lid(n), batch, nb), &out);
649 batch = NULL;
650 nb = 0;
651 }
652 }
653 if (nb) mj_node_send(n, leader, mj_frame_result(mj_node_lid(n), batch, nb), &out);
654 else free(batch);
655 }
656 if ((now > n->prune_at ? now - n->prune_at : 0) >= 250) {
657 n->prune_at = now;
658 const mj_table *t = &n->rep.committed;
659 uint64_t k[3];
660 for (mj_outbox_entry *e = momap_first(n->outbox, k); e; e = momap_next(n->outbox, k, k)) {
661 const mj_job *j = mj_table_get(t, k[0]);
662 bool keep = j ? !mj_job_settled(j, k[1]) && mj_job_active(j) : !mj_table_forgot(t, k[0]);
663 if (!keep) momap_del(n->outbox, k);
664 }
665 /* A staged payload whose item is settled meanwhile (another run's result committed) is not needed. */
666 for (mj_staged *s = momap_first(n->staged, k); s; s = momap_next(n->staged, k, k)) {
667 const mj_job *j = mj_table_get(t, k[0]);
668 bool gone = j ? mj_job_settled(j, k[1]) || !mj_job_active(j) : mj_table_forgot(t, k[0]);
669 if (!gone) continue;
670 mj_staged x;
671 momap_take(n->staged, k, &x);
672 if (x.r.has_blob) mj_outs_push(&out, mj_out_blob_done(x.r.job, x.r.blob.hash));
673 mj_item_res_free(&x.r);
674 }
675 }
676 mj_node_pump(n, now, unix_ms, &out);
677 return out;
678}