==> src/blobs.c <==1/* The blob thread (blobd.rs `run`) and its writers (blobd.rs `writer`): @ajax/mesh-blobd over sh->blobs,
2 * the peer links and the jobs thread. */
3#include "internal.h"
4
5#include "mesh/log.h"
6#include "mesh/strbuf.h"
7
8#include <stdlib.h>
9#include <string.h>
10
11typedef struct {
12 mjd *d;
13 mesh_shared *sh;
14} bx;
15
16static bool io_send_peer(void *c, uint32_t to, const uint8_t *f, size_t n) { return mesh_shared_send_peer(((bx *)c)->sh, to, f, n); }
17static bool io_send_bulk(void *c, uint32_t to, const uint8_t *f, size_t n) { return mesh_shared_send_bulk(((bx *)c)->sh, to, f, n); }
18
19static void refs_free(void *p) {
20 mb_refs_free(p);
21 free(p);
22}
23
24static void jobs_send(mesh_shared *sh, mesh_job_event *e) {
25 if (mesh_chan_send(sh->jobs, e) != MESH_CHAN_OK) mesh_job_event_free(e);
26}
27
28static void refs_event(mesh_shared *sh, mesh_job_event_kind k, const mb_ref *refs, size_t n) {
29 if (!n) return;
30 mb_refs *v = mxalloc(sizeof *v);
31 v->v = mxalloc(n * sizeof *v->v);
32 memcpy(v->v, refs, n * sizeof *v->v);
33 v->n = n;
34 mesh_job_event e = mesh_job_event_new(k);
35 e.msg = v;
36 e.msg_free = refs_free;
37 jobs_send(sh, &e);
38}
39
40static void io_stored(void *c, const mb_ref *refs, size_t n) { refs_event(((bx *)c)->sh, MESH_JEV_BLOBS_STORED, refs, n); }
41static void io_gone(void *c, const mb_ref *refs, size_t n) { refs_event(((bx *)c)->sh, MESH_JEV_BLOB_GONE, refs, n); }
42
43static void io_acked(void *c, uint32_t from, mj_job_id job, const mj_hash *h, bool ok) {
44 mb_ref *r = mxalloc(sizeof *r);
45 *r = (mb_ref){job, *h};
46 mesh_job_event e = mesh_job_event_new(MESH_JEV_BLOB_ACKED);
47 e.from = from;
48 e.ok = ok;
49 e.msg = r;
50 e.msg_free = free;
51 jobs_send(((bx *)c)->sh, &e);
52}
53
54static uint32_t io_alive(void *c) {
55 bx *x = c;
56 return mjd_ext_view(x->sh).alive | 1u << x->sh->me;
57}
58
59static bool io_leading(void *c) {
60 bx *x = c;
61 mj_view v = mjd_ext_view(x->sh);
62 return (v.has_leader && v.leader == x->sh->me) || !x->sh->peer;
63}
64
65static void io_watching(void *c, mj_job_id job, bool on) { mjd_ext_set_watch(((bx *)c)->sh, job, on); }
66
67static void io_mem(void *c, int64_t delta) {
68 mesh_shared *sh = ((bx *)c)->sh;
69 if (delta >= 0) atomic_fetch_add_explicit(&sh->blob_mem, (uint64_t)delta, memory_order_relaxed);
70 else atomic_fetch_sub_explicit(&sh->blob_mem, (uint64_t)-delta, memory_order_relaxed);
71}
72
73static void io_write(void *c, mbd_write w) {
74 if (mesh_chan_send(((bx *)c)->d->wq, &w) != MESH_CHAN_OK) mbd_write_free(&w);
75}
76
77static int64_t io_now(void *c) {
78 (void)c;
79 return mesh_mono_ms();
80}
81
82static void io_log(void *c, bool error, const char *line) {
83 (void)c;
84 if (error) mesh_error("%s", line);
85 else mesh_info("%s", line);
86}
87
88static void event(mbd *b, mjd_bev *e) {
89 switch (e->k) {
90 case MJD_BEV_LOCAL:
91 mbd_local(b, e->job, e->hash, e->bytes, e->n);
92 e->bytes = NULL;
93 break;
94 case MJD_BEV_PUSH: mbd_push(b, e->to, e->job, e->hash); break;
95 case MJD_BEV_FETCH: mbd_fetch(b, e->job, e->hash, e->holders); break;
96 case MJD_BEV_DONE: mbd_done(b, e->job, e->hash); break;
97 case MJD_BEV_FRAME: mbd_frame(b, e->from, e->ty, e->bytes, e->n); break;
98 case MJD_BEV_TABLE:
99 mbd_table(b, e->news);
100 free(e->news);
101 e->news = NULL;
102 break;
103 case MJD_BEV_RENDER: {
104 uint8_t *out = NULL;
105 size_t len = 0;
106 char *err = NULL;
107 if (mbd_render(b, (const char *const *)e->words, e->nwords, &out, &len, &err)) mesh_reply_bytes(e->reply, out ? out : (uint8_t *)"", len);
108 else mesh_reply_err(e->reply, err ? err : "");
109 e->reply = NULL;
110 free(out);
111 free(err);
112 break;
113 }
114 case MJD_BEV_WRITTEN: mbd_written(b, e->job, e->hash, e->why, e->from, e->ok, e->err); break;
115 }
116}
117
118void *mjd_blobs_main(void *arg) {
119 mjd *d = arg;
120 mesh_shared *sh = d->sh;
121 bx cx = {d, sh};
122 mb_store *s = d->bstore;
123 d->bstore = NULL;
124 mbd_io io = {&cx, io_send_peer, io_send_bulk, io_stored, io_acked, io_gone, io_alive, io_leading, io_watching, io_mem, io_write, io_now, io_log};
125 /* It logs what the store holds, and the jobs thread learns it (the leader commits only what it holds, for that job). */
126 mbd *b = mbd_new(d->cfg.me, s, io);
127 for (;;) {
128 if (atomic_load(&sh->fatal) || atomic_load(&d->stop)) break;
129 mjd_bev e;
130 int w = mesh_chan_recv(sh->blobs, &e, 5);
131 if (w == MESH_CHAN_CLOSED) break;
132 if (w == MESH_CHAN_OK) {
133 size_t k = 0;
134 do {
135 event(b, &e);
136 mjd_bev_free(&e);
137 } while (++k < 4096 && mesh_chan_try_recv(sh->blobs, &e) == MESH_CHAN_OK);
138 }
139 mbd_tick(b);
140 /* References whose file turned out missing or corrupt: jobd no longer counts them. */
141 mbd_report_lost(b);
142 }
143 mbd_free(b);
144 mesh_chan_drop_sender(d->wq); /* the writers drain their queue and end */
145 return NULL;
146}
147
148/* A writer: a batch of references made durable (their directories fsynced once), then each reported. */
149static pthread_mutex_t writers_mu = PTHREAD_MUTEX_INITIALIZER;
150
151void *mjd_writer_main(void *arg) {
152 mjd *d = arg;
153 enum { BATCH = 64 };
154 mbd_write ws[BATCH];
155 bool ok[BATCH];
156 char *err[BATCH];
157 for (;;) {
158 size_t n = 0;
159 pthread_mutex_lock(&writers_mu);
160 int w = mesh_chan_recv(d->wq, &ws[0], -1);
161 if (w == MESH_CHAN_OK) {
162 n = 1;
163 while (n < BATCH && mesh_chan_try_recv(d->wq, &ws[n]) == MESH_CHAN_OK) n++;
164 }
165 pthread_mutex_unlock(&writers_mu);
166 if (w != MESH_CHAN_OK) return NULL;
167 mbd_write_batch(ws, n, ok, err);
168 for (size_t i = 0; i < n; i++) {
169 mjd_bev e = {.k = MJD_BEV_WRITTEN, .job = ws[i].job, .hash = ws[i].hash, .why = ws[i].why, .from = ws[i].from, .ok = ok[i], .err = err[i]};
170 mjd_blobs_send(d->sh, &e);
171 mbd_write_free(&ws[i]);
172 }
173 }
174}