==> src/blobd.c <==1/* blobd.rs's Blobd: keep, ensure, the events, the frames, news, watch, tick, render, resolve. */
2#include "mesh/blobd/blobd.h"
3#include "mesh/abi/peer_frames.h"
4#include "mesh/blobs/durable.h"
5#include "mesh/blobs/frame.h"
6#include "mesh/omap.h"
7#include "mesh/ref/sha256d.h"
8#include "mesh/strbuf.h"
9#include "pathtrace/scene.h"
10
11#include <stdlib.h>
12#include <string.h>
13
14typedef struct {
15 mj_job_id *v;
16 size_t n;
17} jobset;
18
19static void js_add(jobset *s, mj_job_id j) {
20 size_t at = 0;
21 while (at < s->n && s->v[at] < j) at++;
22 if (at < s->n && s->v[at] == j) return;
23 s->v = mxrealloc(s->v, (s->n + 1) * sizeof *s->v);
24 memmove(&s->v[at + 1], &s->v[at], (s->n - at) * sizeof *s->v);
25 s->v[at] = j;
26 s->n++;
27}
28
29static void js_del(jobset *s, mj_job_id j) {
30 for (size_t i = 0; i < s->n; i++)
31 if (s->v[i] == j) {
32 memmove(&s->v[i], &s->v[i + 1], (s->n - i - 1) * sizeof *s->v);
33 s->n--;
34 return;
35 }
36}
37
38typedef struct {
39 jobset jobs; /* the jobs that still push it */
40 uint8_t *bytes;
41 size_t n;
42} memblob;
43
44typedef struct {
45 mj_job_id job;
46 uint64_t item;
47} jobitem;
48
49typedef struct {
50 mj_job_id job;
51 uint32_t holders, tried;
52 bool asked;
53 uint32_t asked_k;
54 int64_t asked_at;
55 jobset keep; /* the jobs whose reference is kept once it arrives (the leader's parked results) */
56 jobitem *items; /* the assemblies waiting for it */
57 size_t nitems;
58 int64_t after;
59 uint32_t absent; /* members that answered "not here" this round */
60 bool lost; /* a whole round of the members up answered "not here" */
61} fetch;
62
63typedef struct {
64 masm *a;
65 int64_t asked_at;
66 char *sha;
67} asmx;
68
69typedef struct {
70 uint32_t to;
71 mj_job_id job;
72 mj_hash h;
73} pushk;
74
75struct mbd {
76 mb_store *store;
77 uint32_t me;
78 mbd_io io;
79 momap *mem; /* hash -> memblob */
80 pushk *pushq;
81 size_t npushq, cappushq;
82 momap *queued; /* (to, job, hash) */
83 momap *inflight; /* (to, job, hash) -> int64_t sent at */
84 momap *owed; /* (job, hash) -> u32 list: acks owed once a reference being written is durable */
85 momap *forgotten;
86 uint64_t refused;
87 bool full_logged_set;
88 int64_t full_logged;
89 momap *fetches; /* hash -> fetch */
90 momap *jobs; /* id -> mbd_job_info */
91 momap *asm_; /* id -> asmx */
92 uint64_t puts_in, puts_out, gets_served, bad_in;
93};
94
95typedef struct {
96 uint32_t *v;
97 size_t n;
98} u32list;
99
100static void memblob_free(void *v) {
101 memblob *m = v;
102 free(m->jobs.v);
103 free(m->bytes);
104}
105static void fetch_free(void *v) {
106 fetch *f = v;
107 free(f->keep.v);
108 free(f->items);
109}
110static void asm_free(void *v) {
111 asmx *a = v;
112 masm_free(a->a);
113 free(a->sha);
114}
115static void info_free(void *v) {
116 mbd_job_info *j = v;
117 free(j->program);
118 free(j->setup);
119 free(j->state);
120}
121static void u32list_free(void *v) { free(((u32list *)v)->v); }
122
123void mbd_write_free(mbd_write *w) {
124 free(w->dir);
125 free(w->link);
126 free(w->bytes);
127 memset(w, 0, sizeof *w);
128}
129
130void mbd_news_free(mbd_news *n) {
131 for (size_t i = 0; i < n->njobs; i++) info_free(&n->jobs[i]);
132 free(n->jobs);
133 free(n->forgotten);
134 free(n->results);
135 memset(n, 0, sizeof *n);
136}
137
138static void hk(uint64_t k[4], const mj_hash *h) { mj_hash_words(h, k); }
139static void k5(uint64_t k[5], mj_job_id job, const mj_hash *h) {
140 k[0] = job;
141 mj_hash_words(h, k + 1);
142}
143static void k6(uint64_t k[6], uint32_t to, mj_job_id job, const mj_hash *h) {
144 k[0] = to;
145 k[1] = job;
146 mj_hash_words(h, k + 2);
147}
148
149static void say(mbd *b, bool error, char *line) {
150 if (b->io.log) b->io.log(b->io.ctx, error, line);
151 free(line);
152}
153
154static char *word(mj_job_id id) {
155 char w[MJ_JOB_WORD_MAX];
156 mj_job_word(id, w);
157 return mstrdup(w);
158}
159
160static char *hexs(const mj_hash *h) {
161 char *x = mxalloc(65);
162 mj_hex(h, x);
163 return x;
164}
165
166static void stored1(mbd *b, mj_job_id job, const mj_hash *h) {
167 mb_ref r = {job, *h};
168 b->io.stored(b->io.ctx, &r, 1);
169}
170
171mbd *mbd_new(uint32_t me, mb_store *store, mbd_io io) {
172 mbd *b = mxalloc(sizeof *b);
173 b->store = store;
174 b->me = me;
175 b->io = io;
176 b->mem = momap_new(4, sizeof(memblob), memblob_free, NULL);
177 b->queued = momap_new(6, 1, NULL, NULL);
178 b->inflight = momap_new(6, sizeof(int64_t), NULL, NULL);
179 b->owed = momap_new(5, sizeof(u32list), u32list_free, NULL);
180 b->forgotten = momap_new(1, 1, NULL, NULL);
181 b->fetches = momap_new(4, sizeof(fetch), fetch_free, NULL);
182 b->jobs = momap_new(1, sizeof(mbd_job_info), info_free, NULL);
183 b->asm_ = momap_new(1, sizeof(asmx), asm_free, NULL);
184 mj_job_id *js;
185 size_t nj = mb_jobs(store, &js);
186 free(js);
187 say(b, false, mfmt("blobs: %s (%zu blob(s), %llu reference(s) of %zu job(s), %llu of %llu MiB)",
188 mb_persisted(store) ? "kept on disk" : "kept in memory (no --blobs dir)", mb_count(store), (unsigned long long)mb_refs_total(store),
189 nj, (unsigned long long)(mb_bytes(store) >> 20), (unsigned long long)(mb_cap(store) >> 20)));
190 /* The jobs thread learns what is here (the leader commits only what it holds, for that job). */
191 mb_refs all = mb_all(store);
192 if (all.n) b->io.stored(b->io.ctx, all.v, all.n);
193 mb_refs_free(&all);
194 return b;
195}
196
197void mbd_free(mbd *b) {
198 if (!b) return;
199 mb_store_free(b->store);
200 momap_free(b->mem);
201 free(b->pushq);
202 momap_free(b->queued);
203 momap_free(b->inflight);
204 momap_free(b->owed);
205 momap_free(b->forgotten);
206 momap_free(b->fetches);
207 momap_free(b->jobs);
208 momap_free(b->asm_);
209 free(b);
210}
211
212mb_store *mbd_store(mbd *b) { return b->store; }
213
214static bool is_forgotten(const mbd *b, mj_job_id j) { return momap_get1(b->forgotten, j) != NULL; }
215
216static mb_live live(void *ctx, mj_job_id j) { return is_forgotten(ctx, j) ? MB_FORGOTTEN : MB_LISTED; }
217
218static void ack(mbd *b, uint32_t to, mj_job_id job, const mj_hash *h, bool ok) {
219 msb f = {0};
220 mbf_ack(&f, h, ok, job);
221 b->io.send_peer(b->io.ctx, to, (const uint8_t *)f.p, f.n);
222 msb_free(&f);
223}
224
225/* A present or in-memory blob's bytes (a copy), or NULL. */
226static uint8_t *bytes_of(mbd *b, const mj_hash *h, size_t *n) {
227 uint64_t k[4];
228 hk(k, h);
229 memblob *m = momap_get(b->mem, k);
230 if (m) {
231 uint8_t *c = mxalloc(m->n ? m->n : 1);
232 memcpy(c, m->bytes, m->n);
233 *n = m->n;
234 return c;
235 }
236 return mb_get(b->store, h, n);
237}
238
239static bool here(const mbd *b, const mj_hash *h) {
240 uint64_t k[4];
241 hk(k, h);
242 return momap_get(b->mem, k) != NULL || mb_has(b->store, h);
243}
244
245static void written(mbd *b, mj_job_id job, mj_hash hash, mbd_why why, uint32_t from, bool ok, const char *err);
246
247/* Job's reference to hash made (written, or linked to another job's file of the same bytes), or kept in
248 * memory without a store dir. Takes bytes. False with *err (refused: full, forgotten). */
249static bool keep(mbd *b, mj_job_id job, mj_hash hash, uint8_t *bytes, size_t n, mbd_why why, uint32_t from, char **err) {
250 if (is_forgotten(b, job)) {
251 char *w = word(job);
252 *err = mfmt("job %s is forgotten", w);
253 free(w);
254 free(bytes);
255 return false;
256 }
257 mb_reserved r;
258 char *e = NULL;
259 if (!mb_reserve(b->store, job, hash, (uint32_t)n, live, b, &r, &e)) {
260 b->refused++;
261 int64_t now = b->io.now_ms(b->io.ctx);
262 if (!b->full_logged_set || now - b->full_logged > 10000) {
263 b->full_logged_set = true;
264 b->full_logged = now;
265 char *w = word(job);
266 say(b, true, mfmt("blobs: %s: %llu cop(ies) refused so far (job %s the latest); this member is no copy for them, and while it leads no payload result is committed",
267 e, (unsigned long long)b->refused, w));
268 free(w);
269 }
270 *err = e;
271 free(bytes);
272 return false;
273 }
274 if (r.evicted.n) {
275 say(b, false, mfmt("blobs: %zu reference(s) of forgotten jobs evicted for room", r.evicted.n));
276 b->io.gone(b->io.ctx, r.evicted.v, r.evicted.n);
277 }
278 char *d = mb_dir_of(b->store, job);
279 if (d) {
280 mbd_write w = {d, hash, r.link ? r.path : NULL, bytes, n, why, from, job};
281 r.path = NULL;
282 b->io.write(b->io.ctx, w);
283 } else {
284 mb_put_mem(b->store, job, hash, bytes, n);
285 free(bytes);
286 written(b, job, hash, why, from, true, NULL);
287 }
288 mb_reserved_free(&r);
289 return true;
290}
291
292/* Job's copy of hash here: told to jobd if it is, being made if it is, else made from bytes (or the bytes
293 * here for another job). Takes bytes (may be NULL). */
294static bool ensure(mbd *b, mj_job_id job, mj_hash hash, uint8_t *bytes, size_t n, mbd_why why, char **err) {
295 if (mb_has_ref(b->store, job, &hash)) {
296 free(bytes);
297 stored1(b, job, &hash);
298 return true;
299 }
300 if (mb_known_ref(b->store, job, &hash)) {
301 free(bytes);
302 return true;
303 }
304 if (!bytes) bytes = bytes_of(b, &hash, &n);
305 if (!bytes) {
306 *err = mstrdup("its bytes are not here");
307 return false;
308 }
309 return keep(b, job, hash, bytes, n, why, 0, err);
310}
311
312void mbd_local(mbd *b, mj_job_id job, mj_hash hash, uint8_t *bytes, size_t n) {
313 uint64_t k[4];
314 hk(k, &hash);
315 bool fresh;
316 memblob *m = momap_put(b->mem, k, &fresh);
317 if (fresh) {
318 m->bytes = mxalloc(n ? n : 1);
319 memcpy(m->bytes, bytes, n);
320 m->n = n;
321 b->io.mem(b->io.ctx, (int64_t)n);
322 }
323 js_add(&m->jobs, job);
324 char *e = NULL;
325 if (!ensure(b, job, hash, bytes, n, MBD_WHY_LOCAL, &e)) {
326 char *x = hexs(&hash), *w = word(job);
327 say(b, true, mfmt("blobs: own payload %s of job %s: %s", x, w, e));
328 free(x);
329 free(w);
330 free(e);
331 b->io.acked(b->io.ctx, b->me, job, &hash, false);
332 }
333}
334
335void mbd_done(mbd *b, mj_job_id job, mj_hash hash) {
336 uint64_t k[4];
337 hk(k, &hash);
338 memblob *m = momap_get(b->mem, k);
339 if (!m) return;
340 js_del(&m->jobs, job);
341 if (!m->jobs.n) {
342 b->io.mem(b->io.ctx, -(int64_t)m->n);
343 momap_del(b->mem, k);
344 }
345}
346
347void mbd_push(mbd *b, uint32_t to, mj_job_id job, mj_hash hash) {
348 uint64_t k[6];
349 k6(k, to, job, &hash);
350 if (to == b->me || momap_get(b->queued, k) || momap_get(b->inflight, k)) return;
351 momap_put(b->queued, k, NULL);
352 if (b->npushq == b->cappushq) {
353 b->cappushq = b->cappushq ? 2 * b->cappushq : 64;
354 b->pushq = mxrealloc(b->pushq, b->cappushq * sizeof *b->pushq);
355 }
356 b->pushq[b->npushq++] = (pushk){to, job, hash};
357}
358
359static fetch *fetch_of(mbd *b, const mj_hash *h, mj_job_id job, uint32_t holders) {
360 uint64_t k[4];
361 hk(k, h);
362 bool fresh;
363 fetch *f = momap_put(b->fetches, k, &fresh);
364 if (fresh) {
365 f->job = job;
366 f->holders = holders;
367 f->after = b->io.now_ms(b->io.ctx);
368 }
369 return f;
370}
371
372void mbd_fetch(mbd *b, mj_job_id job, mj_hash hash, uint32_t holders) {
373 if (mb_known_ref(b->store, job, &hash) || here(b, &hash)) {
374 char *e = NULL;
375 if (!ensure(b, job, hash, NULL, 0, MBD_WHY_FETCHED, &e)) {
376 char *x = hexs(&hash), *w = word(job);
377 say(b, true, mfmt("blobs: job %s's copy of %s: %s", w, x, e));
378 free(x);
379 free(w);
380 free(e);
381 }
382 return;
383 }
384 fetch *f = fetch_of(b, &hash, job, holders);
385 js_add(&f->keep, job);
386 f->holders |= holders;
387}
388
389static void written(mbd *b, mj_job_id job, mj_hash hash, mbd_why why, uint32_t from, bool ok, const char *err) {
390 mb_written(b->store, job, &hash, ok);
391 uint64_t k[5];
392 k5(k, job, &hash);
393 u32list owed = {0};
394 momap_take(b->owed, k, &owed);
395 if (ok && mb_has_ref(b->store, job, &hash)) {
396 /* jobd hears of the copy before any member does (the leader's commit check follows the ack). */
397 stored1(b, job, &hash);
398 if (why == MBD_WHY_PUT) ack(b, from, job, &hash, true);
399 for (size_t i = 0; i < owed.n; i++) ack(b, owed.v[i], job, &hash, true);
400 } else {
401 /* Failed, or the job was forgotten while it was written (no copy kept). */
402 char *w = word(job), *x = hexs(&hash);
403 char *e = ok ? mfmt("job %s forgotten meanwhile", w) : mstrdup(err ? err : "write failed");
404 say(b, true, mfmt("blobs: job %s's copy of %s: %s", w, x, e));
405 free(w);
406 free(x);
407 free(e);
408 if (why == MBD_WHY_PUT) ack(b, from, job, &hash, false);
409 else if (why == MBD_WHY_LOCAL) b->io.acked(b->io.ctx, b->me, job, &hash, false);
410 for (size_t i = 0; i < owed.n; i++) ack(b, owed.v[i], job, &hash, false);
411 }
412 free(owed.v);
413}
414
415void mbd_written(mbd *b, mj_job_id job, mj_hash hash, mbd_why why, uint32_t from, bool ok, const char *err) { written(b, job, hash, why, from, ok, err); }
416
417/* A blob for an assembly's item: folded if it is still wanted; false if refused (its digest). */
418static bool fold_in(mbd *b, asmx *x, uint64_t item, const uint8_t *bytes, size_t n) {
419 char *e = NULL;
420 if (masm_blob(x->a, item, bytes, n, &e) >= 0) return true;
421 char *w = word(masm_job(x->a));
422 say(b, true, mfmt("render job %s: %s", w, e));
423 free(w);
424 free(e);
425 return false;
426}
427
428/* A fetched blob's bytes (from a member, or here meanwhile): the jobs that keep it get their reference,
429 * the assemblies that wait for it fold it. Takes f's contents and bytes (may be NULL). */
430static void resolve(mbd *b, const mj_hash *hash, fetch *f, uint8_t *bytes, size_t n) {
431 if (!bytes) bytes = bytes_of(b, hash, &n);
432 if (bytes) {
433 for (size_t i = 0; i < f->keep.n; i++) {
434 uint8_t *c = mxalloc(n ? n : 1);
435 memcpy(c, bytes, n);
436 char *e = NULL;
437 if (!ensure(b, f->keep.v[i], *hash, c, n, MBD_WHY_FETCHED, &e)) {
438 char *x = hexs(hash), *w = word(f->keep.v[i]);
439 say(b, true, mfmt("blobs: fetched %s for job %s: %s", x, w, e));
440 free(x);
441 free(w);
442 free(e);
443 }
444 }
445 for (size_t i = 0; i < f->nitems; i++) {
446 asmx *a = momap_get1(b->asm_, f->items[i].job);
447 if (a) fold_in(b, a, f->items[i].item, bytes, n);
448 }
449 free(bytes);
450 }
451 fetch_free(f);
452}
453
454void mbd_frame(mbd *b, uint32_t from, uint8_t ty, const uint8_t *body, size_t n) {
455 mbf_body x;
456 if (n < 32) {
457 say(b, true, mfmt("blobs: a frame of type %u from system %u: short", ty, from));
458 return;
459 }
460 if (!mbf_decode(ty, body, n, &x)) return;
461 switch (ty) {
462 case MESH_F_BLOB_PUT: {
463 mj_hash got;
464 mesh_sha256(x.bytes, x.nbytes, got.b);
465 if (x.nbytes != x.len || !mj_hash_eq(&got, &x.hash)) {
466 b->bad_in++;
467 char *h = hexs(&x.hash);
468 say(b, true, mfmt("blobs: system %u pushed a blob that is not %s (dropped)", from, h));
469 free(h);
470 return;
471 }
472 b->puts_in++;
473 /* This job's copy: here already (acked), being made (acked when durable), else made now. */
474 if (mb_has_ref(b->store, x.job, &x.hash)) {
475 ack(b, from, x.job, &x.hash, true);
476 return;
477 }
478 if (mb_known_ref(b->store, x.job, &x.hash)) {
479 uint64_t k[5];
480 k5(k, x.job, &x.hash);
481 u32list *o = momap_put(b->owed, k, NULL);
482 o->v = mxrealloc(o->v, (o->n + 1) * sizeof *o->v);
483 o->v[o->n++] = from;
484 return;
485 }
486 uint8_t *c = mxalloc(x.nbytes ? x.nbytes : 1);
487 memcpy(c, x.bytes, x.nbytes);
488 char *e = NULL;
489 if (!keep(b, x.job, x.hash, c, x.nbytes, MBD_WHY_PUT, from, &e)) {
490 if (!b->full_logged_set || !strstr(e, "full")) {
491 char *h = hexs(&x.hash), *w = word(x.job);
492 say(b, true, mfmt("blobs: system %u's blob %s of job %s: %s", from, h, w, e));
493 free(h);
494 free(w);
495 }
496 free(e);
497 ack(b, from, x.job, &x.hash, false);
498 }
499 break;
500 }
501 case MESH_F_BLOB_ACK: {
502 uint64_t k[6];
503 k6(k, from, x.job, &x.hash);
504 momap_del(b->inflight, k);
505 b->io.acked(b->io.ctx, from, x.job, &x.hash, x.ok);
506 break;
507 }
508 case MESH_F_BLOB_GET: {
509 size_t bn = 0;
510 uint8_t *found = bytes_of(b, &x.hash, &bn);
511 msb f = {0};
512 mbf_blob(&f, &x.hash, found, bn, found != NULL);
513 if (found) {
514 if (b->io.send_bulk(b->io.ctx, from, (const uint8_t *)f.p, f.n)) b->gets_served++;
515 } else b->io.send_peer(b->io.ctx, from, (const uint8_t *)f.p, f.n);
516 free(found);
517 msb_free(&f);
518 break;
519 }
520 case MESH_F_BLOB: {
521 uint64_t k[4];
522 hk(k, &x.hash);
523 fetch *f = momap_get(b->fetches, k);
524 if (!f) return;
525 if (f->asked && f->asked_k == from) f->asked = false;
526 if (!x.ok) {
527 f->absent |= (uint32_t)1 << from;
528 return;
529 }
530 mj_hash got;
531 mesh_sha256(x.bytes, x.nbytes, got.b);
532 if (!mj_hash_eq(&got, &x.hash)) {
533 b->bad_in++;
534 char *h = hexs(&x.hash);
535 say(b, true, mfmt("blobs: system %u sent a blob that is not %s (dropped; asking another holder)", from, h));
536 free(h);
537 return;
538 }
539 fetch fx;
540 momap_take(b->fetches, k, &fx);
541 uint8_t *c = mxalloc(x.nbytes ? x.nbytes : 1);
542 memcpy(c, x.bytes, x.nbytes);
543 resolve(b, &x.hash, &fx, c, x.nbytes);
544 break;
545 }
546 default: break;
547 }
548}
549
550static mbd_job_info info_clone(const mbd_job_info *j) {
551 mbd_job_info c = *j;
552 c.program = mstrdup(j->program);
553 c.setup = mj_setup_clone(j->setup, j->nsetup);
554 c.state = mstrdup(j->state);
555 return c;
556}
557
558/* Start assembling job id (jobd sends its results from now on). */
559static bool watch(mbd *b, mj_job_id id, char **err) {
560 if (momap_get1(b->asm_, id)) return true;
561 const mbd_job_info *j = momap_get1(b->jobs, id);
562 char *w = word(id);
563 if (!j) {
564 *err = mfmt("no job %s (committed here)", w);
565 free(w);
566 return false;
567 }
568 if (!masm_renders(j->program, j->op)) {
569 *err = mfmt("job %s (%s op %u) is not a pathtrace image", w, j->program, j->op);
570 free(w);
571 return false;
572 }
573 free(w);
574 masm *a = masm_new(id, j->setup, j->nsetup, j->tone, err);
575 if (!a) return false;
576 asmx *x = momap_put1(b->asm_, id, NULL);
577 *x = (asmx){a, b->io.now_ms(b->io.ctx), NULL};
578 b->io.watching(b->io.ctx, id, true);
579 return true;
580}
581
582void mbd_table(mbd *b, mbd_news *n) {
583 for (size_t i = 0; i < n->njobs; i++) {
584 asmx *a = momap_get1(b->asm_, n->jobs[i].id);
585 if (a && masm_tone(a->a) != n->jobs[i].tone) {
586 masm_set_tone(a->a, n->jobs[i].tone);
587 free(a->sha);
588 a->sha = NULL;
589 }
590 }
591 momap_clear(b->jobs);
592 for (size_t i = 0; i < n->njobs; i++) *(mbd_job_info *)momap_put1(b->jobs, n->jobs[i].id, NULL) = info_clone(&n->jobs[i]);
593 for (size_t i = 0; i < n->nforgotten; i++) {
594 mj_job_id id = n->forgotten[i];
595 momap_put1(b->forgotten, id, NULL);
596 momap_del1(b->asm_, id);
597 b->io.watching(b->io.ctx, id, false);
598 uint64_t k[4];
599 for (fetch *f = momap_first(b->fetches, k); f; f = momap_next(b->fetches, k, k)) {
600 js_del(&f->keep, id);
601 size_t kept = 0;
602 for (size_t q = 0; q < f->nitems; q++)
603 if (f->items[q].job != id) f->items[kept++] = f->items[q];
604 f->nitems = kept;
605 if (!f->keep.n && !f->nitems) momap_del(b->fetches, k);
606 }
607 mb_refs gone = mb_drop_job(b->store, id);
608 if (gone.n) {
609 char *w = word(id);
610 say(b, false, mfmt("blobs: job %s forgotten: its %zu reference(s) removed (%zu blob(s), %llu MiB, left for other jobs)", w, gone.n,
611 mb_count(b->store), (unsigned long long)(mb_bytes(b->store) >> 20)));
612 free(w);
613 b->io.gone(b->io.ctx, gone.v, gone.n);
614 }
615 mb_refs_free(&gone);
616 }
617 for (size_t i = 0; i < n->nresults; i++) {
618 asmx *a = momap_get1(b->asm_, n->results[i].job);
619 if (a) masm_commit(a->a, n->results[i].item, n->results[i].it);
620 }
621 /* The leader assembles every active image (its blobs are all here). */
622 if (b->io.leading(b->io.ctx)) {
623 uint64_t id;
624 momap_iter it;
625 momap_iter_init(&it, b->jobs);
626 mj_job_id act[1024];
627 size_t na = 0;
628 for (const mbd_job_info *j = momap_iter_next1(&it, &id); j; j = momap_iter_next1(&it, &id))
629 if (j->active && masm_renders(j->program, j->op) && na < 1024) act[na++] = id;
630 for (size_t i = 0; i < na; i++) {
631 char *e = NULL;
632 if (!watch(b, act[i], &e)) free(e);
633 }
634 }
635 mbd_news_free(n);
636}
637
638static bool asked_by(void *ctx, uint64_t item);
639
640typedef struct {
641 const mbd *b;
642 mj_job_id id;
643 const masm *a;
644} asked_cx;
645
646static bool asked_by(void *ctx, uint64_t item) {
647 const asked_cx *c = ctx;
648 const masm_item *it = masm_wanted(c->a, item);
649 if (!it) return false;
650 uint64_t k[4];
651 hk(k, &it->hash);
652 const fetch *f = momap_get(c->b->fetches, k);
653 if (!f) return false;
654 for (size_t i = 0; i < f->nitems; i++)
655 if (f->items[i].job == c->id && f->items[i].item == item) return true;
656 return false;
657}
658
659static int lowest(uint32_t mask, const uint32_t asked[32]) {
660 int best = -1;
661 for (int k = 0; k < 32; k++)
662 if ((mask >> k & 1) && (best < 0 || asked[k] < asked[best])) best = k;
663 return best;
664}
665
666void mbd_tick(mbd *b) {
667 int64_t now = b->io.now_ms(b->io.ctx);
668 uint32_t alive = b->io.alive(b->io.ctx) | (uint32_t)1 << b->me;
669 /* Pushes: per member at most PUT_WINDOW unacknowledged; a lost member's queue dropped. */
670 uint64_t k6k[6];
671 for (int64_t *t = momap_first(b->inflight, k6k); t; t = momap_next(b->inflight, k6k, k6k))
672 if (now - *t >= (int64_t)MESH_BLOBD_PUT_TIMEOUT_MS) momap_del(b->inflight, k6k);
673 size_t per[32] = {0};
674 for (int64_t *t = momap_first(b->inflight, k6k); t; t = momap_next(b->inflight, k6k, k6k))
675 if (k6k[0] < 32) per[k6k[0]]++;
676 uint32_t blocked = 0;
677 size_t keep = 0;
678 for (size_t i = 0; i < b->npushq; i++) {
679 pushk p = b->pushq[i];
680 uint64_t k[6];
681 k6(k, p.to, p.job, &p.h);
682 if (!(alive >> p.to & 1)) {
683 momap_del(b->queued, k);
684 continue;
685 }
686 if ((blocked >> p.to & 1) || per[p.to] >= MESH_BLOBD_PUT_WINDOW) {
687 blocked |= (uint32_t)1 << p.to;
688 b->pushq[keep++] = p;
689 continue;
690 }
691 size_t n;
692 uint8_t *bytes = bytes_of(b, &p.h, &n);
693 if (!bytes) {
694 momap_del(b->queued, k);
695 char *x = hexs(&p.h);
696 say(b, true, mfmt("blobs: %s to push is gone", x));
697 free(x);
698 continue;
699 }
700 msb f = {0};
701 mbf_put(&f, &p.h, p.job, bytes, (uint32_t)n);
702 free(bytes);
703 if (b->io.send_bulk(b->io.ctx, p.to, (const uint8_t *)f.p, f.n)) {
704 momap_del(b->queued, k);
705 *(int64_t *)momap_put(b->inflight, k, NULL) = now;
706 per[p.to]++;
707 b->puts_out++;
708 } else {
709 blocked |= (uint32_t)1 << p.to;
710 b->pushq[keep++] = p;
711 }
712 msb_free(&f);
713 }
714 b->npushq = keep;
715 /* The assemblies' next passes: read here, or fetched. */
716 size_t reads = 0;
717 uint64_t id = 0;
718 for (asmx *x = momap_first(b->asm_, &id); x; x = momap_next(b->asm_, &id, &id)) {
719 size_t fetching = momap_len(b->fetches);
720 size_t avail = MESH_BLOBD_FETCH_MAX > fetching ? MESH_BLOBD_FETCH_MAX - fetching : 0;
721 size_t budget = avail + MESH_BLOBD_LOCAL_READS > reads ? avail + MESH_BLOBD_LOCAL_READS - reads : 0;
722 if (!budget) break;
723 uint64_t *items = mxalloc(budget * sizeof *items);
724 masm_item *its = mxalloc(budget * sizeof *its);
725 asked_cx cx = {b, id, x->a};
726 size_t nw = masm_wants(x->a, asked_by, &cx, budget, items, its);
727 for (size_t i = 0; i < nw; i++) {
728 if (here(b, &its[i].hash)) {
729 /* Here: read it (the next tick if this one read enough). */
730 if (reads >= MESH_BLOBD_LOCAL_READS) continue;
731 reads++;
732 size_t n;
733 uint8_t *bytes = bytes_of(b, &its[i].hash, &n);
734 if (bytes) {
735 bool ok = fold_in(b, x, items[i], bytes, n);
736 free(bytes);
737 if (ok) continue;
738 }
739 }
740 if (momap_len(b->fetches) >= MESH_BLOBD_FETCH_MAX) continue;
741 fetch *f = fetch_of(b, &its[i].hash, id, its[i].holders);
742 f->holders |= its[i].holders;
743 bool has = false;
744 for (size_t q = 0; q < f->nitems; q++) has |= f->items[q].job == id && f->items[q].item == items[i];
745 if (!has) {
746 f->items = mxrealloc(f->items, (f->nitems + 1) * sizeof *f->items);
747 f->items[f->nitems++] = (jobitem){id, items[i]};
748 }
749 }
750 free(items);
751 free(its);
752 }
753 /* A blob fetched that is here meanwhile (written, pushed here, for any job) is read instead. */
754 uint64_t hk4[4];
755 for (fetch *f = momap_first(b->fetches, hk4); f; f = momap_next(b->fetches, hk4, hk4)) {
756 mj_hash h = mj_hash_of_words(hk4);
757 if (!here(b, &h)) continue;
758 fetch fx;
759 momap_take(b->fetches, hk4, &fx);
760 resolve(b, &h, &fx, NULL, 0);
761 }
762 /* Fetches: one holder at a time, the next on a refusal or a timeout. */
763 uint32_t asked[32] = {0};
764 for (fetch *f = momap_first(b->fetches, hk4); f; f = momap_next(b->fetches, hk4, hk4))
765 if (f->asked && f->asked_k < 32) asked[f->asked_k]++;
766 for (fetch *f = momap_first(b->fetches, hk4); f; f = momap_next(b->fetches, hk4, hk4)) {
767 mj_hash h = mj_hash_of_words(hk4);
768 if (f->asked) {
769 if (now - f->asked_at < (int64_t)MESH_BLOBD_FETCH_TIMEOUT_MS) continue;
770 f->asked = false;
771 }
772 if (now < f->after) continue;
773 uint32_t not_me = ~((uint32_t)1 << b->me);
774 int k = lowest(f->holders & alive & ~f->tried & not_me, asked);
775 if (k < 0) k = lowest(alive & ~f->tried & not_me, asked);
776 if (k < 0) {
777 /* Every member asked: a fetch for the leader's commit ends; an assembly's starts over in a while.
778 * Every member up said "not here": held by no member up -- said loudly. */
779 if (!f->nitems) {
780 momap_del(b->fetches, hk4);
781 continue;
782 }
783 uint32_t others = alive & not_me;
784 if (others && (others & ~f->absent) == 0 && !f->lost) {
785 f->lost = true;
786 char *x = hexs(&h), *w = word(f->items[0].job);
787 say(b, true, mfmt("render: blob %s (job %s item %llu, holders 0x%x) is held by no member up", x, w, (unsigned long long)f->items[0].item,
788 f->holders));
789 free(x);
790 free(w);
791 }
792 f->tried = 0;
793 f->absent = 0;
794 f->after = now + MESH_BLOBD_FETCH_TIMEOUT_MS;
795 continue;
796 }
797 f->tried |= (uint32_t)1 << k;
798 f->asked = true;
799 f->asked_k = (uint32_t)k;
800 f->asked_at = now;
801 asked[k]++;
802 msb fr = {0};
803 mbf_get(&fr, &h, f->job);
804 b->io.send_peer(b->io.ctx, (uint32_t)k, (const uint8_t *)fr.p, fr.n);
805 msb_free(&fr);
806 }
807 /* An assembly nobody asked for in a while stops (not the leader's active ones: news restarts them). */
808 for (asmx *x = momap_first(b->asm_, &id); x; x = momap_next(b->asm_, &id, &id)) {
809 const mbd_job_info *j = momap_get1(b->jobs, id);
810 if (now - x->asked_at > 600000 && !(j && j->active)) {
811 momap_del1(b->asm_, id);
812 b->io.watching(b->io.ctx, id, false);
813 }
814 }
815}
816
817void mbd_report_lost(mbd *b) {
818 mb_refs lost = mb_take_lost(b->store);
819 if (lost.n) b->io.gone(b->io.ctx, lost.v, lost.n);
820 mb_refs_free(&lost);
821}
822
823bool mbd_render(mbd *b, const char *const *w, size_t nw, uint8_t **out, size_t *len, char **err) {
824 const char *job = "latest", *what = "status";
825 if (nw == 1) job = w[0];
826 else if (nw == 2) job = w[0], what = w[1];
827 else if (nw > 2) {
828 *err = mstrdup("usage: render [<job> | latest] [status | ppm | bmp | raw]");
829 return false;
830 }
831 mj_job_id id = 0;
832 if (strcmp(job, "latest") == 0) {
833 bool any = false;
834 uint64_t k;
835 for (const mbd_job_info *j = momap_first(b->jobs, &k); j; j = momap_next(b->jobs, &k, &k))
836 if (masm_renders(j->program, j->op)) {
837 id = k;
838 any = true;
839 }
840 if (!any) {
841 *err = mstrdup("no pathtrace job in the committed table");
842 return false;
843 }
844 } else if (!mj_parse_job(job, &id)) {
845 *err = mfmt("\"%s\": a job id is <term>.<n>, or latest", job);
846 return false;
847 }
848 if (!watch(b, id, err)) return false;
849 const mbd_job_info *info = momap_get1(b->jobs, id);
850 size_t fetching = 0, unavailable = 0;
851 uint64_t hk4[4];
852 for (const fetch *f = momap_first(b->fetches, hk4); f; f = momap_next(b->fetches, hk4, hk4)) {
853 bool mine = false;
854 for (size_t i = 0; i < f->nitems; i++) mine |= f->items[i].job == id;
855 fetching += mine;
856 unavailable += mine && f->lost;
857 }
858 asmx *x = momap_get1(b->asm_, id);
859 x->asked_at = b->io.now_ms(b->io.ctx);
860 masm *a = x->a;
861 if (strcmp(what, "ppm") == 0) {
862 *out = masm_ppm(a, len);
863 return true;
864 }
865 if (strcmp(what, "bmp") == 0) {
866 *out = masm_bmp_of(a, len);
867 return true;
868 }
869 if (strcmp(what, "status") != 0 && strcmp(what, "raw") != 0) {
870 *err = mfmt("render: \"%s\": status, ppm, bmp or raw", what);
871 return false;
872 }
873 uint64_t dacc, dpix;
874 masm_digests(a, &dacc, &dpix);
875 if (masm_complete(a) && !x->sha) x->sha = masm_sha256(a);
876 char *w2 = word(id);
877 msb s = {0};
878 msb_printf(&s, "render job %s program %s state %s system %u width %u height %u tiles %u passes %u spp %u items %llu committed %llu folded %llu tiles_done %u tone %s fetching %zu unavailable %zu bad %zu complete %u\n",
879 w2, info ? info->program : "?", info ? info->state : "?", b->me, masm_width(a), masm_height(a), masm_tiles(a), masm_passes(a), masm_spp(a),
880 (unsigned long long)masm_items(a), (unsigned long long)masm_committed(a), (unsigned long long)masm_folded(a), masm_tiles_done(a),
881 masm_tone(a) == 1 ? "reinhard" : "aces", fetching, unavailable, masm_bad_count(a), masm_complete(a) ? 1u : 0u);
882 free(w2);
883 msb_printf(&s, "digests acc %016llx pixels %016llx sha256 %s\n", (unsigned long long)dacc, (unsigned long long)dpix, x->sha ? x->sha : "-");
884 uint32_t tiles_by[32] = {0};
885 for (uint32_t t = 0; t < masm_tiles(a); t++) {
886 uint32_t np, o;
887 masm_tile(a, t, &np, &o);
888 if (np > 0 && o < 32) tiles_by[o]++;
889 }
890 uint64_t m;
891 for (const uint64_t *c = momap_first(masm_by_member(a), &m); c; c = momap_next(masm_by_member(a), &m, &m))
892 msb_printf(&s, "member %llu items %llu tiles %u\n", (unsigned long long)m, (unsigned long long)*c, m < 32 ? tiles_by[m] : 0);
893 msb_printf(&s, "store blobs %zu bytes %llu puts_in %llu puts_out %llu gets_served %llu bad_in %llu refs %llu cap %llu refused %llu\n", mb_count(b->store),
894 (unsigned long long)mb_bytes(b->store), (unsigned long long)b->puts_in, (unsigned long long)b->puts_out,
895 (unsigned long long)b->gets_served, (unsigned long long)b->bad_in, (unsigned long long)mb_refs_total(b->store),
896 (unsigned long long)mb_cap(b->store), (unsigned long long)b->refused);
897 for (size_t i = 0; i < masm_bad_count(a) && i < 5; i++) msb_printf(&s, "bad %s\n", masm_bad_at(a, i, NULL));
898 if (strcmp(what, "status") == 0) {
899 *len = s.n;
900 *out = (uint8_t *)msb_take(&s);
901 return true;
902 }
903 /* raw (mesh-top's Render tab): the text, a 0 byte, u32 tiles, per tile (passes folded, member of the
904 * last), and the 8-bit pixels (tile-major 0xffBBGGRR). */
905 msb_putc(&s, 0);
906 uint32_t tiles = masm_tiles(a);
907 msb_put(&s, &tiles, 4);
908 for (uint32_t t = 0; t < tiles; t++) {
909 uint32_t np, o;
910 masm_tile(a, t, &np, &o);
911 msb_put(&s, &np, 4);
912 msb_put(&s, &o, 4);
913 }
914 msb_put(&s, masm_rgb(a), (size_t)tiles * PT_TILE_PIX * 4);
915 *len = s.n;
916 *out = (uint8_t *)msb_take(&s);
917 return true;
918}
919
920void mbd_write_batch(mbd_write *ws, size_t n, bool *ok, char **err) {
921 for (size_t i = 0; i < n; i++) {
922 char x[65];
923 mj_hex(&ws[i].hash, x);
924 err[i] = NULL;
925 ok[i] = mb_make_durable(ws[i].dir, x, ws[i].link, ws[i].bytes, ws[i].n, &err[i]);
926 }
927 /* Each directory fsynced once; a failure fails every write that had succeeded. */
928 for (size_t i = 0; i < n; i++) {
929 bool seen = false;
930 for (size_t j = 0; j < i; j++) seen |= strcmp(ws[j].dir, ws[i].dir) == 0;
931 if (seen) continue;
932 char *e = NULL;
933 if (!mb_sync_dir(ws[i].dir, &e)) {
934 for (size_t j = 0; j < n; j++)
935 if (ok[j]) {
936 ok[j] = false;
937 err[j] = mfmt("fsync %s: %s", ws[i].dir, e);
938 }
939 }
940 free(e);
941 }
942}