==> src/op.c <==1/* jobs.rs's Res, BlobRef, Op, and their encoding (put_op, get_op, encode_ops, decode_ops, parts). */
2#include "mesh/jobs/op.h"
3#include "mesh/abi/base.h"
4
5#include <stdlib.h>
6#include <string.h>
7
8void mj_res_free(mj_res *r) {
9 free(r->out);
10 r->out = NULL;
11 r->out_len = 0;
12}
13
14static uint8_t *dup_bytes(const uint8_t *b, size_t n) {
15 if (!n) return NULL;
16 uint8_t *c = mxalloc(n);
17 memcpy(c, b, n);
18 return c;
19}
20
21mj_res mj_res_clone(const mj_res *r) {
22 mj_res c = *r;
23 c.out = dup_bytes(r->out, r->out_len);
24 return c;
25}
26
27bool mj_blob_ref_eq(const mj_blob_ref *a, const mj_blob_ref *b) {
28 return mj_hash_eq(&a->hash, &b->hash) && a->len == b->len && a->holders == b->holders;
29}
30
31bool mj_res_eq(const mj_res *a, const mj_res *b) {
32 return a->lease == b->lease && a->member == b->member && a->result == b->result && a->out_len == b->out_len &&
33 (a->out_len == 0 || memcmp(a->out, b->out, a->out_len) == 0) && a->has_blob == b->has_blob &&
34 (!a->has_blob || mj_blob_ref_eq(&a->blob, &b->blob));
35}
36
37void mj_op_free(mj_op *o) {
38 if (o->kind == MJ_OP_SUBMIT) mj_spec_free(&o->spec);
39 free(o->why);
40 mj_res_free(&o->res);
41 free(o->tally);
42 memset(o, 0, sizeof *o);
43}
44
45mj_op mj_op_clone(const mj_op *o) {
46 mj_op c = *o;
47 if (o->kind == MJ_OP_SUBMIT) c.spec = mj_spec_clone(&o->spec);
48 c.why = o->why ? mstrdup(o->why) : NULL;
49 c.res = mj_res_clone(&o->res);
50 if (o->ntally) {
51 c.tally = mxalloc(o->ntally * sizeof *c.tally);
52 memcpy(c.tally, o->tally, o->ntally * sizeof *c.tally);
53 }
54 return c;
55}
56
57static bool str_eq(const char *a, const char *b) { return strcmp(a ? a : "", b ? b : "") == 0; }
58
59bool mj_op_eq(const mj_op *a, const mj_op *b) {
60 if (a->kind != b->kind) return false;
61 switch (a->kind) {
62 case MJ_OP_SUBMIT: return mj_spec_eq(&a->spec, &b->spec);
63 case MJ_OP_CANCEL: return a->job == b->job && str_eq(a->why, b->why);
64 case MJ_OP_DONE: return a->job == b->job && a->item == b->item && mj_res_eq(&a->res, &b->res);
65 case MJ_OP_DONE_RUN: return a->job == b->job && a->s == b->s && a->e == b->e && a->lease == b->lease && a->member == b->member;
66 case MJ_OP_FAIL: return a->job == b->job && a->item == b->item && a->tries == b->tries && str_eq(a->why, b->why);
67 case MJ_OP_FAILS: return a->job == b->job && a->item == b->item && a->tries == b->tries;
68 case MJ_OP_DUPES: return a->job == b->job && a->n == b->n;
69 case MJ_OP_FINISH: return a->job == b->job && a->at_ms == b->at_ms && a->state == b->state;
70 case MJ_OP_FORGET: return a->job == b->job;
71 case MJ_OP_FOLD:
72 if (a->job != b->job || a->upto != b->upto || a->ntally != b->ntally) return false;
73 for (size_t i = 0; i < a->ntally; i++)
74 if (a->tally[i].member != b->tally[i].member || a->tally[i].n != b->tally[i].n) return false;
75 return true;
76 case MJ_OP_TONE: return a->job == b->job && a->seq == b->seq && a->tone == b->tone;
77 }
78 return false;
79}
80
81mj_op mj_op_submit(const mj_spec *s) { return (mj_op){.kind = MJ_OP_SUBMIT, .job = s->id, .spec = mj_spec_clone(s)}; }
82mj_op mj_op_cancel(mj_job_id job, const char *why) { return (mj_op){.kind = MJ_OP_CANCEL, .job = job, .why = mstrdup(why)}; }
83mj_op mj_op_done(mj_job_id job, uint64_t item, const mj_res *res) {
84 return (mj_op){.kind = MJ_OP_DONE, .job = job, .item = item, .res = mj_res_clone(res)};
85}
86mj_op mj_op_done_run(mj_job_id job, uint64_t s, uint64_t e, uint64_t lease, uint32_t member) {
87 return (mj_op){.kind = MJ_OP_DONE_RUN, .job = job, .s = s, .e = e, .lease = lease, .member = member};
88}
89mj_op mj_op_fail(mj_job_id job, uint64_t item, uint32_t tries, const char *why) {
90 return (mj_op){.kind = MJ_OP_FAIL, .job = job, .item = item, .tries = tries, .why = mstrdup(why)};
91}
92mj_op mj_op_fails(mj_job_id job, uint64_t item, uint32_t tries) { return (mj_op){.kind = MJ_OP_FAILS, .job = job, .item = item, .tries = tries}; }
93mj_op mj_op_dupes(mj_job_id job, uint64_t n) { return (mj_op){.kind = MJ_OP_DUPES, .job = job, .n = n}; }
94mj_op mj_op_finish(mj_job_id job, uint64_t at_ms, uint32_t state) { return (mj_op){.kind = MJ_OP_FINISH, .job = job, .at_ms = at_ms, .state = state}; }
95mj_op mj_op_forget(mj_job_id job) { return (mj_op){.kind = MJ_OP_FORGET, .job = job}; }
96mj_op mj_op_fold(mj_job_id job, uint64_t upto, const mj_tally *tally, size_t n) {
97 mj_op o = {.kind = MJ_OP_FOLD, .job = job, .upto = upto, .ntally = n};
98 if (n) {
99 o.tally = mxalloc(n * sizeof *o.tally);
100 memcpy(o.tally, tally, n * sizeof *o.tally);
101 }
102 return o;
103}
104mj_op mj_op_tone(mj_job_id job, uint32_t seq, uint32_t tone) { return (mj_op){.kind = MJ_OP_TONE, .job = job, .seq = seq, .tone = tone}; }
105
106void mj_ops_push(mj_ops *v, mj_op o) {
107 if (v->n == v->cap) {
108 v->cap = v->cap ? 2 * v->cap : 8;
109 v->v = mxrealloc(v->v, v->cap * sizeof *v->v);
110 }
111 v->v[v->n++] = o;
112}
113
114void mj_ops_free(mj_ops *v) {
115 for (size_t i = 0; i < v->n; i++) mj_op_free(&v->v[i]);
116 free(v->v);
117 *v = (mj_ops){0};
118}
119
120mj_ops mj_ops_clone(const mj_ops *v) {
121 mj_ops c = {0};
122 for (size_t i = 0; i < v->n; i++) mj_ops_push(&c, mj_op_clone(&v->v[i]));
123 return c;
124}
125
126bool mj_ops_eq(const mj_ops *a, const mj_ops *b) {
127 if (a->n != b->n) return false;
128 for (size_t i = 0; i < a->n; i++)
129 if (!mj_op_eq(&a->v[i], &b->v[i])) return false;
130 return true;
131}
132
133void mj_ops_append(mj_ops *dst, mj_ops *src) {
134 for (size_t i = 0; i < src->n; i++) mj_ops_push(dst, src->v[i]);
135 free(src->v);
136 *src = (mj_ops){0};
137}
138
139void mj_w_blob(msb *w, bool has, const mj_blob_ref *b) {
140 if (!has) {
141 mj_w_u8(w, 0);
142 return;
143 }
144 mj_w_u8(w, 1);
145 msb_put(w, b->hash.b, 32);
146 mj_w_u32(w, b->len);
147 mj_w_u32(w, b->holders);
148}
149
150bool mj_rd_blob(mj_rd *r, bool *has, mj_blob_ref *b) {
151 uint8_t t;
152 if (!mj_rd_u8(r, &t)) return false;
153 if (t == 0) {
154 *has = false;
155 memset(b, 0, sizeof *b);
156 return true;
157 }
158 if (t != 1) return mj_rd_fail(r, mfmt("job frame: blob tag %u?", t));
159 const uint8_t *p;
160 if (!mj_rd_take(r, 32, &p)) return false;
161 memcpy(b->hash.b, p, 32);
162 *has = true;
163 return mj_rd_u32(r, &b->len) && mj_rd_u32(r, &b->holders);
164}
165
166void mj_put_op(msb *w, const mj_op *o) {
167 mj_w_u8(w, (uint8_t)o->kind);
168 switch (o->kind) {
169 case MJ_OP_SUBMIT: {
170 msb s = {0};
171 mj_spec_encode(&o->spec, &s);
172 mj_w_bytes(w, s.p, s.n);
173 msb_free(&s);
174 break;
175 }
176 case MJ_OP_CANCEL: mj_w_u64(w, o->job); mj_w_str(w, o->why ? o->why : ""); break;
177 case MJ_OP_DONE:
178 mj_w_u64(w, o->job);
179 mj_w_u64(w, o->item);
180 mj_w_u64(w, o->res.lease);
181 mj_w_u32(w, o->res.member);
182 mj_w_u32(w, o->res.result);
183 mj_w_bytes(w, o->res.out, o->res.out_len);
184 mj_w_blob(w, o->res.has_blob, &o->res.blob);
185 break;
186 case MJ_OP_DONE_RUN:
187 mj_w_u64(w, o->job);
188 mj_w_u64(w, o->s);
189 mj_w_u64(w, o->e);
190 mj_w_u64(w, o->lease);
191 mj_w_u32(w, o->member);
192 break;
193 case MJ_OP_FAIL:
194 mj_w_u64(w, o->job);
195 mj_w_u64(w, o->item);
196 mj_w_u32(w, o->tries);
197 mj_w_str(w, o->why ? o->why : "");
198 break;
199 case MJ_OP_FAILS: mj_w_u64(w, o->job); mj_w_u64(w, o->item); mj_w_u32(w, o->tries); break;
200 case MJ_OP_DUPES: mj_w_u64(w, o->job); mj_w_u64(w, o->n); break;
201 case MJ_OP_FINISH: mj_w_u64(w, o->job); mj_w_u64(w, o->at_ms); mj_w_u32(w, o->state); break;
202 case MJ_OP_FORGET: mj_w_u64(w, o->job); break;
203 case MJ_OP_FOLD:
204 mj_w_u64(w, o->job);
205 mj_w_u64(w, o->upto);
206 mj_w_u32(w, (uint32_t)o->ntally);
207 for (size_t i = 0; i < o->ntally; i++) {
208 mj_w_u32(w, o->tally[i].member);
209 mj_w_u64(w, o->tally[i].n);
210 }
211 break;
212 case MJ_OP_TONE: mj_w_u64(w, o->job); mj_w_u32(w, o->seq); mj_w_u32(w, o->tone); break;
213 }
214}
215
216bool mj_get_op(mj_rd *r, mj_op *out) {
217 uint8_t t;
218 if (!mj_rd_u8(r, &t)) return false;
219 mj_op o = {.kind = (mj_op_kind)t};
220 bool ok;
221 switch (t) {
222 case MJ_OP_SUBMIT: {
223 char *text = NULL, *err = NULL;
224 if (!mj_rd_str(r, &text)) return false;
225 ok = mj_spec_decode(text, &o.spec, &err);
226 free(text);
227 if (!ok) return mj_rd_fail(r, err);
228 o.job = o.spec.id;
229 break;
230 }
231 case MJ_OP_CANCEL: ok = mj_rd_u64(r, &o.job) && mj_rd_str(r, &o.why); break;
232 case MJ_OP_DONE:
233 ok = mj_rd_u64(r, &o.job) && mj_rd_u64(r, &o.item) && mj_rd_u64(r, &o.res.lease) && mj_rd_u32(r, &o.res.member) &&
234 mj_rd_u32(r, &o.res.result) && mj_rd_bytes(r, &o.res.out, &o.res.out_len) && mj_rd_blob(r, &o.res.has_blob, &o.res.blob);
235 if (ok && o.res.out_len == 0) {
236 free(o.res.out);
237 o.res.out = NULL;
238 }
239 break;
240 case MJ_OP_DONE_RUN:
241 ok = mj_rd_u64(r, &o.job) && mj_rd_u64(r, &o.s) && mj_rd_u64(r, &o.e) && mj_rd_u64(r, &o.lease) && mj_rd_u32(r, &o.member);
242 break;
243 case MJ_OP_FAIL: ok = mj_rd_u64(r, &o.job) && mj_rd_u64(r, &o.item) && mj_rd_u32(r, &o.tries) && mj_rd_str(r, &o.why); break;
244 case MJ_OP_FAILS: ok = mj_rd_u64(r, &o.job) && mj_rd_u64(r, &o.item) && mj_rd_u32(r, &o.tries); break;
245 case MJ_OP_DUPES: ok = mj_rd_u64(r, &o.job) && mj_rd_u64(r, &o.n); break;
246 case MJ_OP_FINISH: ok = mj_rd_u64(r, &o.job) && mj_rd_u64(r, &o.at_ms) && mj_rd_u32(r, &o.state); break;
247 case MJ_OP_FORGET: ok = mj_rd_u64(r, &o.job); break;
248 case MJ_OP_FOLD: {
249 uint32_t n;
250 ok = mj_rd_u64(r, &o.job) && mj_rd_u64(r, &o.upto) && mj_rd_u32(r, &n);
251 if (!ok) break;
252 if (n > MESH_MAX_SYSTEMS) return mj_rd_fail(r, mfmt("job op fold: %u members", n));
253 o.tally = n ? mxalloc(n * sizeof *o.tally) : NULL;
254 o.ntally = n;
255 for (uint32_t i = 0; ok && i < n; i++) ok = mj_rd_u32(r, &o.tally[i].member) && mj_rd_u64(r, &o.tally[i].n);
256 break;
257 }
258 case MJ_OP_TONE: ok = mj_rd_u64(r, &o.job) && mj_rd_u32(r, &o.seq) && mj_rd_u32(r, &o.tone); break;
259 default: return mj_rd_fail(r, mfmt("job op %u?", t));
260 }
261 if (!ok) {
262 mj_op_free(&o);
263 return false;
264 }
265 *out = o;
266 return true;
267}
268
269void mj_encode_ops(const mj_op *ops, size_t n, msb *out) {
270 for (size_t i = 0; i < n; i++) mj_put_op(out, &ops[i]);
271}
272
273bool mj_rd_ops(mj_rd *r, mj_ops *out) {
274 while (!mj_rd_done(r)) {
275 mj_op o;
276 if (!mj_get_op(r, &o)) return false;
277 mj_ops_push(out, o);
278 }
279 return true;
280}
281
282bool mj_decode_ops(const void *b, size_t n, mj_ops *out, char **err) {
283 mj_rd r = mj_rd_new(b, n);
284 mj_ops v = {0};
285 if (!mj_rd_ops(&r, &v)) {
286 mj_ops_free(&v);
287 if (err) *err = mj_rd_take_err(&r);
288 mj_rd_free(&r);
289 return false;
290 }
291 *out = v;
292 return true;
293}
294
295mj_parts mj_parts_of(const mj_op *ops, size_t n) {
296 mj_parts p = {0};
297 msb w = {0};
298 for (size_t i = 0; i < n; i++) {
299 mj_put_op(&w, &ops[i]);
300 if (w.n >= MESH_JOBS_PART_BYTES) {
301 p.v = mxrealloc(p.v, (p.n + 1) * sizeof *p.v);
302 p.v[p.n++] = w;
303 w = (msb){0};
304 }
305 }
306 if (w.n || p.n == 0) {
307 p.v = mxrealloc(p.v, (p.n + 1) * sizeof *p.v);
308 p.v[p.n++] = w;
309 } else msb_free(&w);
310 return p;
311}
312
313void mj_parts_free(mj_parts *p) {
314 for (size_t i = 0; i < p->n; i++) msb_free(&p->v[i]);
315 free(p->v);
316 *p = (mj_parts){0};
317}