==> src/frame.c <==1/* jobs.rs's Prog, Held, LeaseMsg, ItemRes and Frame (encode, decode). */
2#include "mesh/jobs/frame.h"
3#include "mesh/abi/peer_frames.h"
4
5#include <stdlib.h>
6#include <string.h>
7
8static uint8_t *dup_bytes(const uint8_t *b, size_t n) {
9 if (!n) return NULL;
10 uint8_t *c = mxalloc(n);
11 memcpy(c, b, n);
12 return c;
13}
14
15void mj_prog_free(mj_prog *p) {
16 free(p->name);
17 p->name = NULL;
18}
19mj_prog mj_prog_clone(const mj_prog *p) { return (mj_prog){mstrdup(p->name), p->exec, p->cap}; }
20bool mj_prog_eq(const mj_prog *a, const mj_prog *b) { return strcmp(a->name, b->name) == 0 && a->exec == b->exec && a->cap == b->cap; }
21bool mj_progs_eq(const mj_prog *a, size_t na, const mj_prog *b, size_t nb) {
22 if (na != nb) return false;
23 for (size_t i = 0; i < na; i++)
24 if (!mj_prog_eq(&a[i], &b[i])) return false;
25 return true;
26}
27mj_prog *mj_progs_clone(const mj_prog *p, size_t n) {
28 if (!n) return NULL;
29 mj_prog *c = mxalloc(n * sizeof *c);
30 for (size_t i = 0; i < n; i++) c[i] = mj_prog_clone(&p[i]);
31 return c;
32}
33void mj_progs_free(mj_prog *p, size_t n) {
34 for (size_t i = 0; i < n; i++) mj_prog_free(&p[i]);
35 free(p);
36}
37
38void mj_held_free(mj_held *h) {
39 free(h->program);
40 mj_ranges_free(&h->items);
41 h->program = NULL;
42}
43mj_held mj_held_clone(const mj_held *h) { return (mj_held){h->lease, h->job, mstrdup(h->program), mj_ranges_clone(&h->items)}; }
44bool mj_held_eq(const mj_held *a, const mj_held *b) {
45 return a->lease == b->lease && a->job == b->job && strcmp(a->program, b->program) == 0 && mj_ranges_eq(&a->items, &b->items);
46}
47
48void mj_lease_msg_free(mj_lease_msg *l) {
49 free(l->program);
50 free(l->setup);
51 mj_ranges_free(&l->items);
52 memset(l, 0, sizeof *l);
53}
54mj_lease_msg mj_lease_msg_clone(const mj_lease_msg *l) {
55 mj_lease_msg c = *l;
56 c.program = mstrdup(l->program);
57 c.setup = mj_setup_clone(l->setup, l->nsetup);
58 c.items = mj_ranges_clone(&l->items);
59 return c;
60}
61bool mj_lease_msg_eq(const mj_lease_msg *a, const mj_lease_msg *b) {
62 return a->lease == b->lease && a->job == b->job && strcmp(a->program, b->program) == 0 && a->op == b->op && a->nsetup == b->nsetup &&
63 (a->nsetup == 0 || memcmp(a->setup, b->setup, a->nsetup * sizeof *a->setup) == 0) && mj_ranges_eq(&a->items, &b->items);
64}
65
66void mj_item_res_free(mj_item_res *r) {
67 free(r->out);
68 r->out = NULL;
69 r->out_len = 0;
70}
71mj_item_res mj_item_res_clone(const mj_item_res *r) {
72 mj_item_res c = *r;
73 c.out = dup_bytes(r->out, r->out_len);
74 return c;
75}
76bool mj_item_res_eq(const mj_item_res *a, const mj_item_res *b) {
77 return a->lease == b->lease && a->job == b->job && a->item == b->item && a->status == b->status && a->result == b->result &&
78 a->out_len == b->out_len && (a->out_len == 0 || memcmp(a->out, b->out, a->out_len) == 0) && a->has_blob == b->has_blob &&
79 (!a->has_blob || mj_blob_ref_eq(&a->blob, &b->blob));
80}
81
82void mj_frame_free(mj_frame *f) {
83 mj_progs_free(f->progs, f->nprogs);
84 for (size_t i = 0; i < f->nheld; i++) mj_held_free(&f->held[i]);
85 free(f->held);
86 mj_ops_free(&f->ops);
87 if (f->ty == MESH_F_LEASE) mj_lease_msg_free(&f->l);
88 for (size_t i = 0; i < f->nitems; i++) mj_item_res_free(&f->items[i]);
89 free(f->items);
90 memset(f, 0, sizeof *f);
91}
92
93mj_frame mj_frame_clone(const mj_frame *f) {
94 mj_frame c = *f;
95 c.progs = mj_progs_clone(f->progs, f->nprogs);
96 c.held = f->nheld ? mxalloc(f->nheld * sizeof *c.held) : NULL;
97 for (size_t i = 0; i < f->nheld; i++) c.held[i] = mj_held_clone(&f->held[i]);
98 c.ops = mj_ops_clone(&f->ops);
99 if (f->ty == MESH_F_LEASE) c.l = mj_lease_msg_clone(&f->l);
100 c.items = f->nitems ? mxalloc(f->nitems * sizeof *c.items) : NULL;
101 for (size_t i = 0; i < f->nitems; i++) c.items[i] = mj_item_res_clone(&f->items[i]);
102 return c;
103}
104
105static bool helds_eq(const mj_held *a, size_t na, const mj_held *b, size_t nb) {
106 if (na != nb) return false;
107 for (size_t i = 0; i < na; i++)
108 if (!mj_held_eq(&a[i], &b[i])) return false;
109 return true;
110}
111
112bool mj_frame_eq(const mj_frame *a, const mj_frame *b) {
113 if (a->ty != b->ty || a->lid != b->lid) return false;
114 switch (a->ty) {
115 case MESH_F_JOB_AVAIL: return mj_progs_eq(a->progs, a->nprogs, b->progs, b->nprogs) && helds_eq(a->held, a->nheld, b->held, b->nheld);
116 case MESH_F_JOB_STATE:
117 return a->kind == b->kind && a->n == b->n && a->commit == b->commit && a->seq == b->seq && mj_ops_eq(&a->ops, &b->ops);
118 case MESH_F_JOB_ACK: return a->n == b->n && a->gap == b->gap;
119 case MESH_F_JOB_PULL: return true;
120 case MESH_F_JOB_HAVE:
121 return a->kind == b->kind && a->seq == b->seq && a->src0 == b->src0 && a->src1 == b->src1 && mj_ops_eq(&a->ops, &b->ops) &&
122 helds_eq(a->held, a->nheld, b->held, b->nheld);
123 case MESH_F_LEASE: return mj_lease_msg_eq(&a->l, &b->l);
124 case MESH_F_LEASE_CUT: return a->lease == b->lease && a->cut == b->cut;
125 case MESH_F_RESULT:
126 if (a->nitems != b->nitems) return false;
127 for (size_t i = 0; i < a->nitems; i++)
128 if (!mj_item_res_eq(&a->items[i], &b->items[i])) return false;
129 return true;
130 }
131 return false;
132}
133
134mj_frame mj_frame_avail(uint64_t lid, mj_prog *progs, size_t nprogs, mj_held *held, size_t nheld) {
135 return (mj_frame){.ty = MESH_F_JOB_AVAIL, .lid = lid, .progs = progs, .nprogs = nprogs, .held = held, .nheld = nheld};
136}
137mj_frame mj_frame_state(uint64_t lid, uint32_t kind, uint64_t n, uint64_t commit, uint32_t seq, mj_ops ops) {
138 return (mj_frame){.ty = MESH_F_JOB_STATE, .lid = lid, .kind = kind, .n = n, .commit = commit, .seq = seq, .ops = ops};
139}
140mj_frame mj_frame_ack(uint64_t lid, uint64_t n, bool gap) { return (mj_frame){.ty = MESH_F_JOB_ACK, .lid = lid, .n = n, .gap = gap}; }
141mj_frame mj_frame_pull(uint64_t lid) { return (mj_frame){.ty = MESH_F_JOB_PULL, .lid = lid}; }
142mj_frame mj_frame_have(uint64_t lid, uint32_t kind, uint32_t seq, uint64_t src0, uint64_t src1, mj_ops ops, mj_held *held, size_t nheld) {
143 return (mj_frame){.ty = MESH_F_JOB_HAVE, .lid = lid, .kind = kind, .seq = seq, .src0 = src0, .src1 = src1, .ops = ops, .held = held, .nheld = nheld};
144}
145mj_frame mj_frame_lease(uint64_t lid, mj_lease_msg l) { return (mj_frame){.ty = MESH_F_LEASE, .lid = lid, .l = l}; }
146mj_frame mj_frame_cut(uint64_t lid, uint64_t lease, uint64_t cut) { return (mj_frame){.ty = MESH_F_LEASE_CUT, .lid = lid, .lease = lease, .cut = cut}; }
147mj_frame mj_frame_result(uint64_t lid, mj_item_res *items, size_t nitems) {
148 return (mj_frame){.ty = MESH_F_RESULT, .lid = lid, .items = items, .nitems = nitems};
149}
150
151void mj_w_held(msb *w, const mj_held *h, size_t n) {
152 mj_w_u32(w, (uint32_t)n);
153 for (size_t i = 0; i < n; i++) {
154 mj_w_u64(w, h[i].lease);
155 mj_w_u64(w, h[i].job);
156 mj_w_str(w, h[i].program);
157 mj_w_ranges(w, &h[i].items);
158 }
159}
160
161bool mj_rd_held(mj_rd *r, mj_held **out, size_t *count) {
162 uint32_t n;
163 if (!mj_rd_u32(r, &n)) return false;
164 mj_held *v = NULL;
165 size_t k = 0;
166 for (uint32_t i = 0; i < n; i++) {
167 mj_held h = {0};
168 if (!mj_rd_u64(r, &h.lease) || !mj_rd_u64(r, &h.job) || !mj_rd_str(r, &h.program) || !mj_rd_ranges(r, &h.items)) {
169 mj_held_free(&h);
170 for (size_t j = 0; j < k; j++) mj_held_free(&v[j]);
171 free(v);
172 return false;
173 }
174 v = mxrealloc(v, (k + 1) * sizeof *v);
175 v[k++] = h;
176 }
177 *out = v;
178 *count = k;
179 return true;
180}
181
182void mj_frame_encode(const mj_frame *f, msb *out) {
183 msb w = {0};
184 mj_w_u64(&w, f->lid);
185 switch (f->ty) {
186 case MESH_F_JOB_AVAIL:
187 mj_w_u32(&w, (uint32_t)f->nprogs);
188 for (size_t i = 0; i < f->nprogs; i++) {
189 mj_w_str(&w, f->progs[i].name);
190 mj_w_u32(&w, f->progs[i].exec);
191 mj_w_u32(&w, f->progs[i].cap);
192 }
193 mj_w_held(&w, f->held, f->nheld);
194 break;
195 case MESH_F_JOB_STATE:
196 mj_w_u32(&w, f->kind);
197 mj_w_u64(&w, f->n);
198 mj_w_u64(&w, f->commit);
199 mj_w_u32(&w, f->seq);
200 mj_encode_ops(f->ops.v, f->ops.n, &w);
201 break;
202 case MESH_F_JOB_ACK:
203 mj_w_u64(&w, f->n);
204 mj_w_u32(&w, f->gap ? 1 : 0);
205 break;
206 case MESH_F_JOB_PULL: break;
207 case MESH_F_JOB_HAVE:
208 mj_w_u32(&w, f->kind);
209 mj_w_u32(&w, f->seq);
210 mj_w_u64(&w, f->src0);
211 mj_w_u64(&w, f->src1);
212 if (f->kind == MESH_JOBS_HAVE_END) mj_w_held(&w, f->held, f->nheld);
213 else mj_encode_ops(f->ops.v, f->ops.n, &w);
214 break;
215 case MESH_F_LEASE:
216 mj_w_u64(&w, f->l.lease);
217 mj_w_u64(&w, f->l.job);
218 mj_w_str(&w, f->l.program);
219 mj_w_u32(&w, f->l.op);
220 mj_w_u32(&w, (uint32_t)f->l.nsetup);
221 for (size_t i = 0; i < f->l.nsetup; i++) {
222 mj_w_u32(&w, f->l.setup[i].op);
223 mj_w_u32(&w, f->l.setup[i].arg);
224 }
225 mj_w_ranges(&w, &f->l.items);
226 break;
227 case MESH_F_LEASE_CUT:
228 mj_w_u64(&w, f->lease);
229 mj_w_u64(&w, f->cut);
230 break;
231 case MESH_F_RESULT:
232 mj_w_u32(&w, (uint32_t)f->nitems);
233 for (size_t i = 0; i < f->nitems; i++) {
234 const mj_item_res *x = &f->items[i];
235 mj_w_u64(&w, x->lease);
236 mj_w_u64(&w, x->job);
237 mj_w_u64(&w, x->item);
238 mj_w_u32(&w, x->status);
239 mj_w_u32(&w, x->result);
240 mj_w_bytes(&w, x->out, x->out_len);
241 mj_w_blob(&w, x->has_blob, &x->blob);
242 }
243 break;
244 }
245 mj_w_u8(out, f->ty);
246 mj_w_u32(out, (uint32_t)w.n);
247 msb_put(out, w.p, w.n);
248 msb_free(&w);
249}
250
251static bool decode_body(uint8_t ty, mj_rd *r, mj_frame *f) {
252 switch (ty) {
253 case MESH_F_JOB_AVAIL: {
254 uint32_t n;
255 if (!mj_rd_u32(r, &n)) return false;
256 for (uint32_t i = 0; i < n; i++) {
257 mj_prog p = {0};
258 if (!mj_rd_str(r, &p.name) || !mj_rd_u32(r, &p.exec) || !mj_rd_u32(r, &p.cap)) {
259 mj_prog_free(&p);
260 return false;
261 }
262 f->progs = mxrealloc(f->progs, (f->nprogs + 1) * sizeof *f->progs);
263 f->progs[f->nprogs++] = p;
264 }
265 return mj_rd_held(r, &f->held, &f->nheld);
266 }
267 case MESH_F_JOB_STATE: return mj_rd_u32(r, &f->kind) && mj_rd_u64(r, &f->n) && mj_rd_u64(r, &f->commit) && mj_rd_u32(r, &f->seq) && mj_rd_ops(r, &f->ops);
268 case MESH_F_JOB_ACK: {
269 uint32_t g;
270 if (!mj_rd_u64(r, &f->n) || !mj_rd_u32(r, &g)) return false;
271 f->gap = g != 0;
272 return true;
273 }
274 case MESH_F_JOB_PULL: return true;
275 case MESH_F_JOB_HAVE:
276 if (!mj_rd_u32(r, &f->kind) || !mj_rd_u32(r, &f->seq) || !mj_rd_u64(r, &f->src0) || !mj_rd_u64(r, &f->src1)) return false;
277 if (f->kind < MESH_JOBS_HAVE_COMMITTED || f->kind > MESH_JOBS_HAVE_END) return mj_rd_fail(r, mfmt("F_JOB_HAVE kind %u?", f->kind));
278 if (f->kind == MESH_JOBS_HAVE_END) return mj_rd_held(r, &f->held, &f->nheld);
279 return mj_rd_ops(r, &f->ops);
280 case MESH_F_LEASE: {
281 uint32_t ns;
282 if (!mj_rd_u64(r, &f->l.lease) || !mj_rd_u64(r, &f->l.job) || !mj_rd_str(r, &f->l.program) || !mj_rd_u32(r, &f->l.op) ||
283 !mj_rd_u32(r, &ns))
284 return false;
285 for (uint32_t i = 0; i < ns; i++) {
286 mj_setup s;
287 if (!mj_rd_u32(r, &s.op) || !mj_rd_u32(r, &s.arg)) return false;
288 f->l.setup = mxrealloc(f->l.setup, (f->l.nsetup + 1) * sizeof *f->l.setup);
289 f->l.setup[f->l.nsetup++] = s;
290 }
291 return mj_rd_ranges(r, &f->l.items);
292 }
293 case MESH_F_LEASE_CUT: return mj_rd_u64(r, &f->lease) && mj_rd_u64(r, &f->cut);
294 case MESH_F_RESULT: {
295 uint32_t n;
296 if (!mj_rd_u32(r, &n)) return false;
297 for (uint32_t i = 0; i < n; i++) {
298 mj_item_res x = {0};
299 if (!mj_rd_u64(r, &x.lease) || !mj_rd_u64(r, &x.job) || !mj_rd_u64(r, &x.item) || !mj_rd_u32(r, &x.status) ||
300 !mj_rd_u32(r, &x.result) || !mj_rd_bytes(r, &x.out, &x.out_len) || !mj_rd_blob(r, &x.has_blob, &x.blob)) {
301 mj_item_res_free(&x);
302 return false;
303 }
304 if (!x.out_len) {
305 free(x.out);
306 x.out = NULL;
307 }
308 f->items = mxrealloc(f->items, (f->nitems + 1) * sizeof *f->items);
309 f->items[f->nitems++] = x;
310 }
311 return true;
312 }
313 default: return mj_rd_fail(r, mfmt("job frame type %u?", ty));
314 }
315}
316
317bool mj_frame_decode(uint8_t ty, const void *body, size_t n, mj_frame *out, char **err) {
318 mj_rd r = mj_rd_new(body, n);
319 mj_frame f = {.ty = ty};
320 bool ok = mj_rd_u64(&r, &f.lid) && decode_body(ty, &r, &f);
321 if (ok && !mj_rd_done(&r)) ok = mj_rd_fail(&r, mfmt("job frame type %u: %zu trailing bytes", ty, n - r.at));
322 if (!ok) {
323 mj_frame_free(&f);
324 if (err) *err = mj_rd_take_err(&r);
325 mj_rd_free(&r);
326 return false;
327 }
328 *out = f;
329 return true;
330}