==> src/table.c <==1/* jobs.rs's Job and Table: apply (the semilattice), ops (a table as ops), next_id, fold. */
2#include "mesh/jobs/table.h"
3
4#include <stdlib.h>
5#include <string.h>
6
7/* ---- map value helpers ---------------------------------------------------------------------- */
8
9static void res_free(void *v) { mj_res_free(v); }
10static void res_copy(void *d, const void *s) { *(mj_res *)d = mj_res_clone(s); }
11static void failed_free(void *v) { free(((mj_failed *)v)->why); }
12static void failed_copy(void *d, const void *s) {
13 const mj_failed *f = s;
14 *(mj_failed *)d = (mj_failed){f->tries, mstrdup(f->why ? f->why : "")};
15}
16static void jobp_free(void *v) { mj_job_free(*(mj_job **)v); }
17static void jobp_copy(void *d, const void *s) { *(mj_job **)d = mj_job_clone(*(mj_job *const *)s); }
18
19static momap *counts(void) { return momap_new(1, sizeof(uint64_t), NULL, NULL); }
20
21uint64_t mj_count_get(const momap *m, uint32_t member) {
22 const uint64_t *v = momap_get1(m, member);
23 return v ? *v : 0;
24}
25
26static void count_add(momap *m, uint32_t member, uint64_t n) {
27 uint64_t *v = momap_put1(m, member, NULL);
28 *v += n;
29}
30
31/* ---- Job -------------------------------------------------------------------------------------- */
32
33mj_job *mj_job_new(const mj_spec *spec) {
34 mj_job *j = mxalloc(sizeof *j);
35 j->spec = mj_spec_clone(spec);
36 j->results = momap_new(1, sizeof(mj_res), res_free, res_copy);
37 j->runs = momap_new(1, sizeof(mj_run), NULL, NULL);
38 j->failed = momap_new(1, sizeof(mj_failed), failed_free, failed_copy);
39 j->fails = momap_new(1, sizeof(uint32_t), NULL, NULL);
40 j->by_member = counts();
41 j->folded = counts();
42 return j;
43}
44
45void mj_job_free(mj_job *j) {
46 if (!j) return;
47 mj_spec_free(&j->spec);
48 mj_ranges_free(&j->done);
49 momap_free(j->results);
50 momap_free(j->runs);
51 momap_free(j->failed);
52 momap_free(j->fails);
53 momap_free(j->by_member);
54 momap_free(j->folded);
55 free(j->cancel);
56 free(j);
57}
58
59mj_job *mj_job_clone(const mj_job *j) {
60 mj_job *c = mxalloc(sizeof *c);
61 *c = *j;
62 c->spec = mj_spec_clone(&j->spec);
63 c->done = mj_ranges_clone(&j->done);
64 c->results = momap_clone(j->results);
65 c->runs = momap_clone(j->runs);
66 c->failed = momap_clone(j->failed);
67 c->fails = momap_clone(j->fails);
68 c->by_member = momap_clone(j->by_member);
69 c->folded = momap_clone(j->folded);
70 c->cancel = j->cancel ? mstrdup(j->cancel) : NULL;
71 return c;
72}
73
74/* Two maps equal key by key, values compared by eq (or bytes). */
75static bool map_eq(const momap *a, const momap *b, bool (*eq)(const void *, const void *)) {
76 if (momap_len(a) != momap_len(b)) return false;
77 momap_iter ia, ib;
78 momap_iter_init(&ia, a);
79 momap_iter_init(&ib, b);
80 const uint64_t *ka, *kb;
81 unsigned kw = momap_key_words(a);
82 const void *va = momap_iter_next(&ia, &ka), *vb = momap_iter_next(&ib, &kb);
83 while (va && vb) {
84 if (memcmp(ka, kb, 8 * kw) != 0) return false;
85 if (eq ? !eq(va, vb) : memcmp(va, vb, momap_val_size(a)) != 0) return false;
86 va = momap_iter_next(&ia, &ka);
87 vb = momap_iter_next(&ib, &kb);
88 }
89 return !va && !vb;
90}
91
92static bool res_eq(const void *a, const void *b) { return mj_res_eq(a, b); }
93static bool run_eq(const void *a, const void *b) {
94 const mj_run *x = a, *y = b;
95 return x->end == y->end && x->lease == y->lease && x->member == y->member;
96}
97static bool failed_eq(const void *a, const void *b) {
98 const mj_failed *x = a, *y = b;
99 return x->tries == y->tries && strcmp(x->why, y->why) == 0;
100}
101static bool u32_eq(const void *a, const void *b) { return *(const uint32_t *)a == *(const uint32_t *)b; }
102static bool u64_eq(const void *a, const void *b) { return *(const uint64_t *)a == *(const uint64_t *)b; }
103
104bool mj_job_eq(const mj_job *a, const mj_job *b) {
105 return mj_spec_eq(&a->spec, &b->spec) && mj_ranges_eq(&a->done, &b->done) && map_eq(a->results, b->results, res_eq) &&
106 map_eq(a->runs, b->runs, run_eq) && map_eq(a->failed, b->failed, failed_eq) && map_eq(a->fails, b->fails, u32_eq) &&
107 a->dupes == b->dupes && (a->cancel == NULL) == (b->cancel == NULL) && (!a->cancel || strcmp(a->cancel, b->cancel) == 0) &&
108 a->finished == b->finished && (!a->finished || (a->finish_at == b->finish_at && a->finish_state == b->finish_state)) &&
109 map_eq(a->by_member, b->by_member, u64_eq) && a->fold_end == b->fold_end && map_eq(a->folded, b->folded, u64_eq) &&
110 a->tone_seq == b->tone_seq && a->tone == b->tone;
111}
112
113uint64_t mj_job_total(const mj_job *j) { return mj_ranges_len(&j->spec.items); }
114
115bool mj_job_settled(const mj_job *j, uint64_t x) { return mj_ranges_contains(&j->done, x) || momap_get1(j->failed, x) != NULL; }
116
117bool mj_job_lease_of(const mj_job *j, uint64_t x, uint64_t *lease) {
118 const mj_res *r = momap_get1(j->results, x);
119 if (r) {
120 *lease = r->lease;
121 return true;
122 }
123 const mj_run *run = momap_floor(j->runs, &x, NULL);
124 if (run && x < run->end) {
125 *lease = run->lease;
126 return true;
127 }
128 return false;
129}
130
131bool mj_job_active(const mj_job *j) { return !j->finished && j->cancel == NULL; }
132
133bool mj_job_complete(const mj_job *j) {
134 if (mj_spec_endless(&j->spec)) return false;
135 uint64_t d = mj_ranges_len(&j->done), f = momap_len(j->failed);
136 uint64_t sum = d > UINT64_MAX - f ? UINT64_MAX : d + f;
137 return sum >= mj_job_total(j);
138}
139
140const char *mj_job_state(const mj_job *j) {
141 if (j->finished) {
142 switch (j->finish_state) {
143 case MESH_JOBS_FIN_DONE: return momap_len(j->failed) == 0 ? "done" : "done-with-failures";
144 case MESH_JOBS_FIN_CANCELLED: return "cancelled";
145 case MESH_JOBS_FIN_DEADLINE: return "deadline";
146 default: break;
147 }
148 }
149 return j->cancel ? "cancelling" : "running";
150}
151
152mj_ranges mj_job_left(const mj_job *j) {
153 mj_ranges r = mj_ranges_clone(&j->spec.items);
154 mj_ranges_subtract(&r, &j->done);
155 uint64_t x;
156 momap_iter it1_;
157 momap_iter_init(&it1_, j->failed);
158 for (const void *v = momap_iter_next1(&it1_, &x); v; v = momap_iter_next1(&it1_, &x)) mj_ranges_take(&r, x);
159 return r;
160}
161
162/* Folds the runs below the done prefix (the items done from the job's first on, without a gap)
163 * into per-member counts. Deterministic: every replica that applies the same ops folds the same. */
164static void fold(mj_job *j) {
165 uint64_t first;
166 if (!mj_ranges_first(&j->spec.items, &first)) return;
167 if (!j->done.n) return;
168 uint64_t s = j->done.v[0][0], w = j->done.v[0][1];
169 if (s != first || w <= j->fold_end) return;
170 momap *keep = momap_split_off(j->runs, &w);
171 uint64_t a;
172 momap_iter it2_;
173 momap_iter_init(&it2_, j->runs);
174 for (const mj_run *r = momap_iter_next1(&it2_, &a); r; r = momap_iter_next1(&it2_, &a)) count_add(j->folded, r->member, r->end - a);
175 momap_free(j->runs);
176 j->runs = keep;
177 j->fold_end = w;
178}
179
180/* ---- Table ------------------------------------------------------------------------------------ */
181
182void mj_table_init(mj_table *t) {
183 memset(t, 0, sizeof *t);
184 t->jobs = momap_new(1, sizeof(mj_job *), jobp_free, jobp_copy);
185}
186
187void mj_table_free(mj_table *t) {
188 momap_free(t->jobs);
189 free(t->forgotten);
190 memset(t, 0, sizeof *t);
191}
192
193mj_table mj_table_clone(const mj_table *t) {
194 mj_table c = {0};
195 c.jobs = momap_clone(t->jobs);
196 if (t->nforgotten) {
197 c.forgotten = mxalloc(t->nforgotten * sizeof *c.forgotten);
198 memcpy(c.forgotten, t->forgotten, t->nforgotten * sizeof *c.forgotten);
199 }
200 c.nforgotten = c.capforgotten = t->nforgotten;
201 return c;
202}
203
204static bool jobp_eq(const void *a, const void *b) { return mj_job_eq(*(mj_job *const *)a, *(mj_job *const *)b); }
205
206bool mj_table_eq(const mj_table *a, const mj_table *b) {
207 return map_eq(a->jobs, b->jobs, jobp_eq) && a->nforgotten == b->nforgotten &&
208 (a->nforgotten == 0 || memcmp(a->forgotten, b->forgotten, a->nforgotten * sizeof *a->forgotten) == 0);
209}
210
211mj_job *mj_table_get(const mj_table *t, mj_job_id id) {
212 mj_job **p = momap_get1(t->jobs, id);
213 return p ? *p : NULL;
214}
215
216bool mj_table_forgot(const mj_table *t, mj_job_id id) {
217 for (size_t i = 0; i < t->nforgotten; i++)
218 if (t->forgotten[i] == id) return true;
219 return false;
220}
221
222mj_job *mj_table_first(const mj_table *t, uint64_t *id) {
223 mj_job **p = momap_first(t->jobs, id);
224 return p ? *p : NULL;
225}
226
227mj_job *mj_table_next(const mj_table *t, uint64_t *id) {
228 mj_job **p = momap_next(t->jobs, id, id);
229 return p ? *p : NULL;
230}
231
232static void done_run(mj_job *j, const mj_op *op) {
233 mj_ranges span = mj_ranges_of(op->s, op->e);
234 mj_ranges fresh = mj_ranges_intersect(&span, &j->spec.items);
235 mj_ranges_free(&span);
236 mj_ranges_subtract(&fresh, &j->done);
237 uint64_t x;
238 momap_iter it3_;
239 momap_iter_init(&it3_, j->failed);
240 for (const void *v = momap_iter_next1(&it3_, &x); v; v = momap_iter_next1(&it3_, &x)) mj_ranges_take(&fresh, x);
241 for (size_t i = 0; i < fresh.n; i++) {
242 uint64_t a = fresh.v[i][0], b = fresh.v[i][1];
243 mj_ranges_insert(&j->done, a, b);
244 count_add(j->by_member, op->member, b - a);
245 /* Adjacent runs of one lease are one run. */
246 uint64_t pa;
247 const mj_run *p = momap_prev(j->runs, &a, &pa);
248 if (p && p->end == a && p->lease == op->lease && p->member == op->member) {
249 momap_del1(j->runs, pa);
250 a = pa;
251 }
252 const mj_run *n = momap_get1(j->runs, b);
253 if (n && n->lease == op->lease && n->member == op->member) {
254 uint64_t nb = n->end;
255 momap_del1(j->runs, b);
256 b = nb;
257 }
258 *(mj_run *)momap_put1(j->runs, a, NULL) = (mj_run){b, op->lease, op->member};
259 }
260 mj_ranges_free(&fresh);
261 fold(j);
262}
263
264static void apply_fold(mj_job *j, const mj_op *op) {
265 if (op->upto <= j->fold_end) return;
266 uint64_t first;
267 if (!mj_ranges_first(&j->spec.items, &first)) return;
268 /* Its runs below `upto` are in the tally now (and the items they did not cover are done there). */
269 momap *keep = momap_split_off(j->runs, &op->upto);
270 uint64_t a;
271 momap_iter it4_;
272 momap_iter_init(&it4_, j->runs);
273 for (const mj_run *r = momap_iter_next1(&it4_, &a); r; r = momap_iter_next1(&it4_, &a)) {
274 uint64_t *c = momap_get1(j->by_member, r->member);
275 if (c) *c = *c > r->end - a ? *c - (r->end - a) : 0;
276 }
277 momap_free(j->runs);
278 j->runs = keep;
279 uint64_t m;
280 momap_iter it5_;
281 momap_iter_init(&it5_, j->folded);
282 for (const uint64_t *n = momap_iter_next1(&it5_, &m); n; n = momap_iter_next1(&it5_, &m)) {
283 uint64_t *c = momap_get1(j->by_member, m);
284 if (c) *c = *c > *n ? *c - *n : 0;
285 }
286 momap_clear(j->folded);
287 for (size_t i = 0; i < op->ntally; i++) *(uint64_t *)momap_put1(j->folded, op->tally[i].member, NULL) = op->tally[i].n;
288 momap_iter it6_;
289 momap_iter_init(&it6_, j->folded);
290 for (const uint64_t *n = momap_iter_next1(&it6_, &m); n; n = momap_iter_next1(&it6_, &m)) count_add(j->by_member, (uint32_t)m, *n);
291 mj_ranges span = mj_ranges_of(first, op->upto);
292 mj_ranges r = mj_ranges_intersect(&span, &j->spec.items);
293 mj_ranges_free(&span);
294 uint64_t x;
295 momap_iter it7_;
296 momap_iter_init(&it7_, j->failed);
297 for (const void *v = momap_iter_next1(&it7_, &x); v; v = momap_iter_next1(&it7_, &x)) mj_ranges_take(&r, x);
298 momap_iter it8_;
299 momap_iter_init(&it8_, j->results);
300 for (const void *v = momap_iter_next1(&it8_, &x); v; v = momap_iter_next1(&it8_, &x)) mj_ranges_take(&r, x);
301 mj_ranges_union(&j->done, &r);
302 mj_ranges_free(&r);
303 j->fold_end = op->upto;
304 fold(j);
305}
306
307void mj_table_apply(mj_table *t, const mj_op *op) {
308 if (op->kind == MJ_OP_FORGET) {
309 momap_del1(t->jobs, op->job);
310 if (!mj_table_forgot(t, op->job)) {
311 if (t->nforgotten == t->capforgotten) {
312 t->capforgotten = t->capforgotten ? 2 * t->capforgotten : 4;
313 t->forgotten = mxrealloc(t->forgotten, t->capforgotten * sizeof *t->forgotten);
314 }
315 t->forgotten[t->nforgotten++] = op->job;
316 }
317 return;
318 }
319 if (op->kind == MJ_OP_SUBMIT) {
320 if (!mj_table_forgot(t, op->spec.id)) {
321 bool fresh;
322 mj_job **p = momap_put1(t->jobs, op->spec.id, &fresh);
323 if (fresh) *p = mj_job_new(&op->spec);
324 }
325 return;
326 }
327 mj_job *j = mj_table_get(t, mj_op_job(op));
328 if (!j) return;
329 switch (op->kind) {
330 case MJ_OP_CANCEL:
331 if (!j->cancel) j->cancel = mstrdup(op->why ? op->why : "");
332 break;
333 case MJ_OP_DONE:
334 if (mj_job_settled(j, op->item) || !mj_ranges_contains(&j->spec.items, op->item)) return;
335 mj_ranges_add(&j->done, op->item);
336 count_add(j->by_member, op->res.member, 1);
337 {
338 bool fresh;
339 mj_res *r = momap_put1(j->results, op->item, &fresh);
340 if (!fresh) mj_res_free(r);
341 *r = mj_res_clone(&op->res);
342 }
343 fold(j);
344 break;
345 case MJ_OP_DONE_RUN: done_run(j, op); break;
346 case MJ_OP_FOLD: apply_fold(j, op); break;
347 case MJ_OP_FAIL:
348 if (mj_job_settled(j, op->item)) return;
349 *(mj_failed *)momap_put1(j->failed, op->item, NULL) = (mj_failed){op->tries, mstrdup(op->why ? op->why : "")};
350 break;
351 case MJ_OP_FAILS: {
352 uint32_t *t2 = momap_put1(j->fails, op->item, NULL);
353 if (op->tries > *t2) *t2 = op->tries;
354 break;
355 }
356 case MJ_OP_DUPES:
357 if (op->n > j->dupes) j->dupes = op->n;
358 break;
359 case MJ_OP_FINISH:
360 if (!j->finished) {
361 j->finished = true;
362 j->finish_at = op->at_ms;
363 j->finish_state = op->state;
364 }
365 break;
366 case MJ_OP_TONE:
367 if (op->seq > j->tone_seq) {
368 j->tone_seq = op->seq;
369 j->tone = op->tone;
370 }
371 break;
372 default: break;
373 }
374}
375
376void mj_table_ops(const mj_table *t, mj_ops *out) {
377 uint64_t id;
378 for (const mj_job *j = mj_table_first(t, &id); j; j = mj_table_next(t, &id)) {
379 mj_job_id job = j->spec.id;
380 mj_ops_push(out, mj_op_submit(&j->spec));
381 uint64_t k;
382 momap_iter it9_;
383 momap_iter_init(&it9_, j->results);
384 for (const mj_res *r = momap_iter_next1(&it9_, &k); r; r = momap_iter_next1(&it9_, &k)) mj_ops_push(out, mj_op_done(job, k, r));
385 momap_iter it10_;
386 momap_iter_init(&it10_, j->failed);
387 for (const mj_failed *f = momap_iter_next1(&it10_, &k); f; f = momap_iter_next1(&it10_, &k))
388 mj_ops_push(out, mj_op_fail(job, k, f->tries, f->why));
389 /* After the results and failures (the fold's range is done but for those), before the runs above it. */
390 if (j->fold_end > 0) {
391 size_t n = momap_len(j->folded), i = 0;
392 mj_tally *tl = n ? mxalloc(n * sizeof *tl) : NULL;
393 momap_iter it11_;
394 momap_iter_init(&it11_, j->folded);
395 for (const uint64_t *c = momap_iter_next1(&it11_, &k); c; c = momap_iter_next1(&it11_, &k)) tl[i++] = (mj_tally){(uint32_t)k, *c};
396 mj_op o = {.kind = MJ_OP_FOLD, .job = job, .upto = j->fold_end, .tally = tl, .ntally = n};
397 mj_ops_push(out, o);
398 }
399 momap_iter it12_;
400 momap_iter_init(&it12_, j->runs);
401 for (const mj_run *r = momap_iter_next1(&it12_, &k); r; r = momap_iter_next1(&it12_, &k))
402 mj_ops_push(out, mj_op_done_run(job, k, r->end, r->lease, r->member));
403 momap_iter it13_;
404 momap_iter_init(&it13_, j->fails);
405 for (const uint32_t *tr = momap_iter_next1(&it13_, &k); tr; tr = momap_iter_next1(&it13_, &k)) mj_ops_push(out, mj_op_fails(job, k, *tr));
406 if (j->dupes > 0) mj_ops_push(out, mj_op_dupes(job, j->dupes));
407 if (j->cancel) mj_ops_push(out, mj_op_cancel(job, j->cancel));
408 if (j->finished) mj_ops_push(out, mj_op_finish(job, j->finish_at, j->finish_state));
409 if (j->tone_seq > 0) mj_ops_push(out, mj_op_tone(job, j->tone_seq, j->tone));
410 }
411 for (size_t i = 0; i < t->nforgotten; i++) mj_ops_push(out, mj_op_forget(t->forgotten[i]));
412}
413
414mj_job_id mj_table_next_id(const mj_table *t, uint64_t term) {
415 uint64_t n = 0, id;
416 for (const mj_job *j = mj_table_first(t, &id); j; j = mj_table_next(t, &id))
417 if (id >> 32 == term && (id & 0xffffffffu) > n) n = id & 0xffffffffu;
418 for (size_t i = 0; i < t->nforgotten; i++)
419 if (t->forgotten[i] >> 32 == term && (t->forgotten[i] & 0xffffffffu) > n) n = t->forgotten[i] & 0xffffffffu;
420 return term << 32 | (n + 1);
421}